Identified some incorrect handlings of end of stream indicator and fixed them:

1. Client-to-Server EOS & Cardinality Decoupling (handleRequestBodyResponse)
The Issue: Previously, the filter tracked application-initiated half-closes using an internal halfClosed atomic boolean. Once set, it assumed the very next ProcessingResponse from the sidecar was the final one and immediately half-closed the upstream RPC, violating gRFC A93’s 1-to-N / M-to-N streaming body specification.
The Fix:
Stateless Half-Close: Completely eliminated halfClosed state tracking; upstream half-closing is now stateless and driven exclusively by sidecar commands.
Explicit A93 Protocol Boundaries: In handleRequestBodyResponse(), proceedWithHalfClose() is triggered strictly when the sidecar sends a ProcessingResponse explicitly marked with end_of_stream = true (piggybacked on a body chunk) or end_of_stream_without_message = true.
Streaming Queue Decoupling: Refined the expectedResponses queue to enforce 1-to-1 synchronous ordering only for headers and trailers, allowing asynchronous streaming body chunks to be exchanged freely without triggering cardinality mismatch errors.
2. Strict proceedWithClose() Lifecycle on Trailers Response
The Issue: handleResponseBodyResponse() previously triggered premature call closure during body processing. Furthermore, there was ambiguity around whether server-to-client body EOS indicators should trigger call completion.
The Fix:
Removed Body-Triggered Closure: Completely removed proceedWithClose() from server-to-client body processing (handleResponseBodyResponse()), ensuring body EOS indicators do not terminate the RPC.
Trailers-Driven Completion: Enforced that client call completion (proceedWithClose()) relies strictly on the receipt of response trailers from the sidecar (hasResponseTrailers()).
Clean Handshake for Skipped Trailers: If response_trailer_mode is set to SKIP (or default), the filter notifies the sidecar that the server stream is finished via an empty body carrying end_of_stream_without_message = true, and immediately invokes proceedWithClose() without waiting for a response.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java

@@ -851,7 +851,6 @@ private enum EventType {
       final AtomicReference<ExtProcStreamState> extProcStreamState =
           new AtomicReference<>(ExtProcStreamState.ACTIVE);
       final AtomicBoolean passThroughMode = new AtomicBoolean(false);
-      final AtomicBoolean halfClosed = new AtomicBoolean(false);
       final AtomicBoolean requestSideClosed = new AtomicBoolean(false);
       final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
 
@@ -1067,28 +1066,47 @@ public void onNext(ProcessingResponse response) {
               }
 
               EventType expected = expectedResponses.peek();
-              EventType received = null;
               if (response.hasRequestHeaders()) {
-                received = EventType.REQUEST_HEADERS;
-              } else if (response.hasRequestBody()) {
-                received = EventType.REQUEST_BODY;
+                if (expected == null || expected != EventType.REQUEST_HEADERS) {
+                  internalOnError(Status.UNAVAILABLE
+                      .withDescription("Protocol error: received response out of order. Expected: " 
+                          + expected + ", Received: REQUEST_HEADERS")
+                      .asRuntimeException());
+                  return;
+                }
+                expectedResponses.poll();
               } else if (response.hasResponseHeaders()) {
-                received = EventType.RESPONSE_HEADERS;
-              } else if (response.hasResponseBody()) {
-                received = EventType.RESPONSE_BODY;
+                if (expected == null || expected != EventType.RESPONSE_HEADERS) {
+                  internalOnError(Status.UNAVAILABLE
+                      .withDescription("Protocol error: received response out of order. Expected: " 
+                          + expected + ", Received: RESPONSE_HEADERS")
+                      .asRuntimeException());
+                  return;
+                }
+                expectedResponses.poll();
               } else if (response.hasResponseTrailers()) {
-                received = EventType.RESPONSE_TRAILERS;
-              }
-
-              if (received != null) {
-                if (expected == null || expected != received) {
+                if (expected == null || expected != EventType.RESPONSE_TRAILERS) {
                   internalOnError(Status.UNAVAILABLE
                       .withDescription("Protocol error: received response out of order. Expected: " 
-                          + expected + ", Received: " + received)
+                          + expected + ", Received: RESPONSE_TRAILERS")
                       .asRuntimeException());
                   return;
                 }
                 expectedResponses.poll();
+              } else if (response.hasRequestBody()) {
+                if (expected == EventType.REQUEST_HEADERS) {
+                  internalOnError(Status.UNAVAILABLE
+                      .withDescription("Protocol error: received request_body before request_headers response.")
+                      .asRuntimeException());
+                  return;
+                }
+              } else if (response.hasResponseBody()) {
+                if (expected == EventType.REQUEST_HEADERS || expected == EventType.RESPONSE_HEADERS) {
+                  internalOnError(Status.UNAVAILABLE
+                      .withDescription("Protocol error: received response_body before headers response.")
+                      .asRuntimeException());
+                  return;
+                }
               }
 
               if (response.getRequestDrain()) {
@@ -1154,6 +1172,7 @@ else if (response.hasResponseTrailers()) {
                       response.getResponseTrailers().getHeaderMutation()
                   );
                 }
+                wrappedListener.proceedWithClose();
               }
 
               checkEndOfStream(response);
@@ -1219,12 +1238,8 @@ private void sendToExtProc(ProcessingRequest request) {
 
           if (request.hasRequestHeaders()) {
             expectedResponses.add(EventType.REQUEST_HEADERS);
-          } else if (request.hasRequestBody()) {
-            expectedResponses.add(EventType.REQUEST_BODY);
           } else if (request.hasResponseHeaders()) {
             expectedResponses.add(EventType.RESPONSE_HEADERS);
-          } else if (request.hasResponseBody()) {
-            expectedResponses.add(EventType.RESPONSE_BODY);
           } else if (request.hasResponseTrailers()) {
             expectedResponses.add(EventType.RESPONSE_TRAILERS);
           }
@@ -1396,7 +1411,6 @@ private void proceedWithHalfClose() {
       @Override
       public void halfClose() {
         clientHalfCloseStartNanos = System.nanoTime();
-        halfClosed.set(true);
         if (passThroughMode.get() || isExtProcStreamCompleted()) {
           if (requestSideClosed.compareAndSet(false, true)) {
             proceedWithHalfClose();
@@ -1443,13 +1457,11 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
             if (!streamed.getBody().isEmpty()) {
               super.sendMessage(streamed.getBody().newInput());
             }
-          }
-        }
-        // If the application already half-closed, and we just received a response from
-        // the sidecar for the last part of the request body, we can now half-close the data plane.
-        if (halfClosed.get()) {
-          if (requestSideClosed.compareAndSet(false, true)) {
-            proceedWithHalfClose();
+            if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
+              if (requestSideClosed.compareAndSet(false, true)) {
+                proceedWithHalfClose();
+              }
+            }
           }
         }
       }
@@ -1463,9 +1475,6 @@ private void handleResponseBodyResponse(
             if (!streamed.getBody().isEmpty()) {
               listener.onExternalBody(streamed.getBody());
             }
-            if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
-              listener.proceedWithClose();
-            }
           }
         }
       }
@@ -1777,10 +1786,7 @@ private void triggerCloseHandshake() {
                   .build())
               .build());
 
-          if (dataPlaneClientCall.config.getObservabilityMode()) {
-            // In observability mode we don't wait for handshake response
-            proceedWithClose();
-          }
+          proceedWithClose();
         }
       }