ext_proc: Fix race in Test/FlowControl_PiggybackedWindowUpdates (#9578)

Fixes : https://github.com/grpc/grpc-go/issues/9400

*Fix*: Capture `defaultWindowSize` once synchronously at stream
initialization in `commonStream` and use `cs.defaultWindowSize`
throughout the stream's lifetime.

*Fix*: Call `stream.CloseSend()` before waiting on both
`upstreamPiggybackedReceived` and `downstreamPiggybackedReceived`, and
use `*grpcsync.Event` for signaling.

RELEASE NOTES: None

internal/xds/httpfilter/extproc/ext_proc.go

diff --git a/internal/xds/httpfilter/extproc/ext_proc.go b/internal/xds/httpfilter/extproc/ext_proc.go
index af7d3a2..6541460 100644
--- a/internal/xds/httpfilter/extproc/ext_proc.go
+++ b/internal/xds/httpfilter/extproc/ext_proc.go
@@ -429,6 +429,7 @@
 		metricsRecorder:        i.metricsRecorder,
 		target:                 i.target,
 		clientHeadersStartTime: timeNow(),
+		defaultWindowSize:      iextproc.DefaultFlowControlWindowSize,
 		// Construct request attributes once during stream initialization to capture
 		// the original, unmutated request metadata and RPC info. For streaming
 		// RPCs, where stream creation and message transmission are separable,
@@ -500,8 +501,8 @@
 		downstreamToSidestreamPositiveUpdateCh: make(chan struct{}, 1),
 		upstreamToSidestreamPositiveUpdateCh:   make(chan struct{}, 1),
 	}
-	cs.downstreamToSidestreamWindow.Store(iextproc.DefaultFlowControlWindowSize)
-	cs.upstreamToSidestreamWindow.Store(iextproc.DefaultFlowControlWindowSize)
+	cs.downstreamToSidestreamWindow.Store(cs.defaultWindowSize)
+	cs.upstreamToSidestreamWindow.Store(cs.defaultWindowSize)
 
 	// When request header processing is "Send", defer creating the dataplane
 	// stream until the external processor responds, because gRPC transmits
@@ -559,6 +560,7 @@
 	metricsRecorder        estats.MetricsRecorder
 	target                 string
 	clientHeadersStartTime time.Time // holds the start time of client headers processing for metrics and relative timestamp calculations
+	defaultWindowSize      int64     // initial flow control window size captured at stream creation
 
 	// onFinish stores OnFinishCallOption callbacks to execute if calling the
 	// delegate (newStream) fails or if the RPC fails before delegating.
@@ -645,10 +647,10 @@
 		}
 		if !cs.config.observabilityMode {
 			req.FlowControlInit = &v3procservicepb.ProcessingRequest_FlowControlInit{
-				InitialWindowDownstreamToSidestream: iextproc.DefaultFlowControlWindowSize,
-				InitialWindowUpstreamToSidestream:   iextproc.DefaultFlowControlWindowSize,
-				InitialWindowSidestreamToUpstream:   iextproc.DefaultFlowControlWindowSize,
-				InitialWindowSidestreamToDownstream: iextproc.DefaultFlowControlWindowSize,
+				InitialWindowDownstreamToSidestream: cs.defaultWindowSize,
+				InitialWindowUpstreamToSidestream:   cs.defaultWindowSize,
+				InitialWindowSidestreamToUpstream:   cs.defaultWindowSize,
+				InitialWindowSidestreamToDownstream: cs.defaultWindowSize,
 			}
 		}
 	}
@@ -1176,7 +1178,7 @@
 		// standalone processing request with window update and clear the pending
 		// update.
 		if bodySize := int64(len(streamedResp.GetBody())); bodySize > 0 {
-			if cs.pendingRespWindowUpdate.Add(bodySize) >= iextproc.DefaultFlowControlWindowSize/2 {
+			if cs.pendingRespWindowUpdate.Add(bodySize) >= cs.defaultWindowSize/2 {
 				if err := cs.sendClientWindowUpdate(); err != nil {
 					return err
 				}
@@ -1456,7 +1458,7 @@
 			// standalone processing request with window update and clear the pending
 			// update.
 			if bodySize := int64(len(streamedResp.GetBody())); bodySize > 0 {
-				if cs.pendingReqWindowUpdate.Add(bodySize) >= iextproc.DefaultFlowControlWindowSize/2 {
+				if cs.pendingReqWindowUpdate.Add(bodySize) >= cs.defaultWindowSize/2 {
 					if err := cs.sendClientWindowUpdate(); err != nil {
 						return
 					}

internal/xds/httpfilter/extproc/ext_proc_ext_test.go

diff --git a/internal/xds/httpfilter/extproc/ext_proc_ext_test.go b/internal/xds/httpfilter/extproc/ext_proc_ext_test.go
index 6a738c1..2186785 100644
--- a/internal/xds/httpfilter/extproc/ext_proc_ext_test.go
+++ b/internal/xds/httpfilter/extproc/ext_proc_ext_test.go
@@ -6641,8 +6641,8 @@
 	const defaultWindowSize = 100
 	overrideDefaultFlowControlWindow(t, defaultWindowSize)
 
-	downstreamPiggybackedReceived := make(chan struct{})
-	upstreamPiggybackedReceived := make(chan struct{})
+	downstreamPiggybackedReceived := grpcsync.NewEvent()
+	upstreamPiggybackedReceived := grpcsync.NewEvent()
 
 	mutatedResp := &testpb.StreamingOutputCallResponse{Payload: &testpb.Payload{Body: make([]byte, 15)}}
 	mutatedRespBytes, err := proto.Marshal(mutatedResp)
@@ -6671,10 +6671,10 @@
 			// Check for piggybacked window updates on ProcessingRequests.
 			if update := req.GetClientWindowUpdate(); update != nil {
 				if update.GetWindowIncrementSidestreamToDownstream() >= wantRespInc {
-					close(downstreamPiggybackedReceived)
+					downstreamPiggybackedReceived.Fire()
 				}
 				if update.GetWindowIncrementSidestreamToUpstream() >= wantReqInc {
-					close(upstreamPiggybackedReceived)
+					upstreamPiggybackedReceived.Fire()
 				}
 			}
 
@@ -6751,36 +6751,36 @@
 		t.Fatalf("FullDuplexCall() failed: %v", err)
 	}
 
-	// Send request 1 (mutated by extproc to 20k, forwarding to backend triggers
-	// upstream window update piggybacked on ResponseBody).
+	// Send request 1 (mutated by extproc to < defaultWindowSize/2, so its
+	// sidestream-to-upstream window increment stays pending for piggybacking).
 	if err := stream.Send(&testpb.StreamingOutputCallRequest{
 		Payload: &testpb.Payload{Body: []byte("c1")},
 	}); err != nil {
 		t.Fatalf("stream.Send() failed: %v", err)
 	}
 
-	// Read response 1 (mutated by extproc to 15k, reading triggers downstream
-	// window update piggybacked on CloseSend).
+	// Read response 1 (mutated by extproc to < defaultWindowSize/2, so its
+	// sidestream-to-downstream window increment stays pending for piggybacking).
 	if _, err := stream.Recv(); err != nil {
 		t.Fatalf("stream.Recv() failed: %v", err)
 	}
 
-	// Verify that upstream piggybacked window update is received.
-	select {
-	case <-upstreamPiggybackedReceived:
-	case <-time.After(defaultTestTimeout):
-		t.Fatalf("Timed out waiting for upstream piggybacked window update on processor stream")
-	}
-
-	// CloseSend sends EndOfStreamWithoutMessage to proc server, piggybacking the
-	// downstream window update.
+	// CloseSend sends EndOfStreamWithoutMessage to proc server, piggybacking
+	// pending window updates (including upstream if ResponseBody was dispatched
+	// before requestForwardingToDataplaneLoop updated pendingReqWindowUpdate).
 	if err := stream.CloseSend(); err != nil {
 		t.Fatalf("stream.CloseSend() failed: %v", err)
 	}
 
-	// Verify that downstream piggybacked window update is received.
+	// Verify that both upstream and downstream piggybacked window updates are
+	// received.
 	select {
-	case <-downstreamPiggybackedReceived:
+	case <-upstreamPiggybackedReceived.Done():
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timed out waiting for upstream piggybacked window update on processor stream")
+	}
+	select {
+	case <-downstreamPiggybackedReceived.Done():
 	case <-time.After(defaultTestTimeout):
 		t.Fatalf("Timed out waiting for downstream piggybacked window update on processor stream")
 	}