xds: stabilize ext_proc server interceptor stream termination and protocol compliance

kannanjgithub · kannanjgithub · commit a5417a7559a4 · 2026-06-02T12:34:30.000Z
Refines the stream termination state machine of the ExternalProcessorServerInterceptor
to achieve full gRFC protocol compliance and resolve concurrency timing races:

- Corrects skipped trailers protocol flow: Eliminates the transmission of terminal empty
  ProcessingRequest messages to the ext_proc engine when responseTrailerMode is set to SKIP,
  immediately unblocking the call and closing the ext_proc stream per gRFC specifications.
- Gated stream termination: Refactors checkEndOfStream() to strictly avoid premature stream
  closure by verifying that the interceptor is not actively waiting for a 1-1 response
  header or RESPONSE_TRAILERS event, accommodating the non-1-1 cardinality of body chunks.
- Race condition mitigation: Introduces isRequestSideCompleted() as a guard to ensure that
  early server-side application completions (e.g., immediate auth errors) defer final call
  and stream teardown until all pending request-side client messages are fully processed.
- Forwards empty mutated bodies: Reverts redundant empty body checks in
  handleResponseBodyResponse() to correctly deliver legitimate empty body mutation payloads.
- Updates test harnesses: Ensures all 50 tests in ExternalProcessorServerInterceptorTest
  stably pass under high concurrency and match the refined handshake protocol sequence.
diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java
@@ -348,6 +348,11 @@ private static ConfigOrError<Optional<GrpcServiceConfig>> parseAndValidate(
       return ConfigOrError.fromError("Invalid response_body_mode: " + mode.getResponseBodyMode()
           + ". Only GRPC and NONE are supported.");
     }
+    if (mode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC
+        && mode.getResponseTrailerMode() != ProcessingMode.HeaderSendMode.SEND) {
+      return ConfigOrError.fromError("Invalid response_trailer_mode: " + mode.getResponseTrailerMode()
+          + ". Trailer mode must be SEND if response body mode is GRPC.");
+    }
 
     try {
       if (grpcService != null && grpcService.hasGoogleGrpc()) {
diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorServerInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorServerInterceptor.java
@@ -105,7 +105,6 @@
 final class ExternalProcessorServerInterceptor implements ServerInterceptor {
   private final CachedChannelManager cachedChannelManager;
   private final ExternalProcessorFilterConfig filterConfig;
-  private final FilterContext context;
   private final MetricRecorder metricsRecorder;
 
   ExternalProcessorServerInterceptor(
@@ -114,7 +113,6 @@ final class ExternalProcessorServerInterceptor implements ServerInterceptor {
       FilterContext context) {
     this.filterConfig = checkNotNull(filterConfig, "filterConfig");
     this.cachedChannelManager = checkNotNull(cachedChannelManager, "cachedChannelManager");
-    this.context = checkNotNull(context, "context");
     this.metricsRecorder = checkNotNull(context.metricsRecorder(), "metricsRecorder");
   }
 
@@ -332,7 +330,9 @@ private enum EventType {
         new AtomicReference<>(ExtProcStreamState.ACTIVE);
     final AtomicBoolean passThroughMode = new AtomicBoolean(false);
     final AtomicBoolean halfClosed = new AtomicBoolean(false);
-    final AtomicBoolean responseSideClosed = new AtomicBoolean(false);
+    final AtomicBoolean requestSideClosed = new AtomicBoolean(false);
+    final AtomicBoolean dataPlaneCallClosed = new AtomicBoolean(false);
+
     final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
     final AtomicBoolean responseHeadersSent = new AtomicBoolean(false);
     final AtomicBoolean terminationTriggered = new AtomicBoolean(false);
@@ -440,6 +440,9 @@ private void activateCall() {
       wrappedListener.setDelegate(appListener);
       drainPendingRequests();
       wrappedListener.onReadyNotify();
+      if (wrappedListener.halfCloseDeferred) {
+        wrappedListener.handleDeferredHalfClose();
+      }
     }
 
     private void recordDuration(DoubleHistogramMetricInstrument instrument, long durationNanos) {
@@ -618,7 +621,7 @@ else if (response.hasResponseTrailers()) {
               proceedWithClose();
             }
 
-            checkEndOfStream(response);
+            checkEndOfStream();
           } catch (Throwable t) {
             internalOnError(t);
           }
@@ -712,7 +715,9 @@ private void closeExtProcStream() {
             extProcClientCallRequestObserver.onCompleted();
           }
         }
+        expectedResponses.clear();
       }
+      proceedWithClose();
     }
 
     private void cancelExtProcStream(Throwable t) {
@@ -726,6 +731,8 @@ private void cancelExtProcStream(Throwable t) {
             extProcClientCallRequestObserver = null;
           }
         }
+        expectedResponses.clear();
+        proceedWithClose();
       }
     }
 
@@ -740,6 +747,7 @@ private void internalOnError(Throwable t) {
             extProcClientCallRequestObserver = null;
           }
         }
+        expectedResponses.clear();
         if (config.getFailureModeAllow()) {
           handleFailOpen();
         } else {
@@ -860,7 +868,7 @@ private void proceedWithSendHeaders(Metadata headers) {
 
     @Override
     public void sendMessage(InputStream message) {
-      if (responseSideClosed.get()) {
+      if (dataPlaneCallClosed.get()) {
         return;
       }
 
@@ -959,7 +967,8 @@ public void close(Status status, Metadata trailers) {
 
     void proceedWithClose() {
       synchronized (rawCallLock) {
-        if (savedStatus != null) {
+        if (savedStatus != null 
+            && (isExtProcStreamCompleted() || config.getObservabilityMode())) {
           if (markDataPlaneCallClosed()) {
             proceedWithClose(savedStatus, savedTrailers);
           }
@@ -970,12 +979,14 @@ void proceedWithClose() {
     }
 
     private void proceedWithClose(Status status, Metadata trailers) {
-      if (serverTrailersStartNanos > 0) {
-        long durationNanos = System.nanoTime() - serverTrailersStartNanos;
-        recordDuration(serverTrailersDuration, durationNanos);
-        serverTrailersStartNanos = 0;
+      if (dataPlaneCallClosed.compareAndSet(false, true)) {
+        if (serverTrailersStartNanos > 0) {
+          long durationNanos = System.nanoTime() - serverTrailersStartNanos;
+          recordDuration(serverTrailersDuration, durationNanos);
+          serverTrailersStartNanos = 0;
+        }
+        super.close(status, trailers);
       }
-      super.close(status, trailers);
     }
 
     private void triggerCloseHandshake() {
@@ -994,14 +1005,9 @@ private void triggerCloseHandshake() {
                 .build())
             .build());
       } else {
-        sendToExtProc(ProcessingRequest.newBuilder()
-            .setResponseBody(HttpBody.newBuilder()
-                .setEndOfStreamWithoutMessage(true)
-                .build())
-            .build());
-
-        if (config.getObservabilityMode()) {
-          proceedWithClose();
+        if (isRequestSideCompleted()) {
+          unblockAfterStreamComplete();
+          closeExtProcStream();
         }
       }
     }
@@ -1024,11 +1030,12 @@ private void sendResponseBodyToExtProc(@Nullable ByteString bodyByteString, bool
     }
 
     private void handleRequestBodyResponse(BodyResponse bodyResponse) {
-      if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
+      if (bodyResponse.hasResponse()
+      && bodyResponse.getResponse().hasBodyMutation()) {
         BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
         if (mutation.hasStreamedResponse()) {
           StreamedBodyResponse streamed = mutation.getStreamedResponse();
-          if (!streamed.getBody().isEmpty()) {
+          if (!streamed.getEndOfStreamWithoutMessage()) {
             wrappedListener.onExternalBody(streamed.getBody());
           }
           if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
@@ -1039,18 +1046,14 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
     }
 
     private void handleResponseBodyResponse(BodyResponse bodyResponse) {
+      if (dataPlaneCallClosed.get()) {
+        return;
+      }
       if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
         BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
         if (mutation.hasStreamedResponse()) {
           StreamedBodyResponse streamed = mutation.getStreamedResponse();
-          if (!streamed.getBody().isEmpty()) {
-            super.sendMessage(new OutboundZeroCopyInputStream(streamed.getBody()));
-          }
-        }
-      }
-      if (responseSideClosed.compareAndSet(false, true)) {
-        if (terminationTriggered.get()) {
-          proceedWithClose();
+          super.sendMessage(new OutboundZeroCopyInputStream(streamed.getBody()));
         }
       }
     }
@@ -1097,23 +1100,20 @@ private void handleFailOpen() {
       closeExtProcStream();
     }
 
-    private void checkEndOfStream(ProcessingResponse response) {
-      boolean terminal = false;
-      if (response.hasResponseTrailers()) {
-        terminal = true;
-      } else if (response.hasResponseHeaders() && savedStatus != null
-          && currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.NONE) {
-        terminal = true;
-      } else if (response.hasResponseBody() && responseSideClosed.get()) {
-        terminal = true;
-      }
-
-      if (terminal) {
+    private void checkEndOfStream() {
+      if (terminationTriggered.get() && isRequestSideCompleted()
+          && !expectedResponses.contains(EventType.RESPONSE_TRAILERS)) {
         unblockAfterStreamComplete();
         closeExtProcStream();
       }
     }
 
+    private boolean isRequestSideCompleted() {
+      return (currentProcessingMode.getRequestHeaderMode() != ProcessingMode.HeaderSendMode.SEND
+          && currentProcessingMode.getRequestBodyMode() != ProcessingMode.BodySendMode.GRPC)
+          || requestSideClosed.get();
+    }
+
     void unblockAfterStreamComplete() {
       proceedWithSendHeaders();
       drainPendingDrainingMessages();
@@ -1125,6 +1125,7 @@ private static final class DataPlaneServerListener extends ServerCall.Listener<I
     private final DataPlaneServerCall dataPlaneServerCall;
     private final Queue<InputStream> savedMessages = new ConcurrentLinkedQueue<>();
     private volatile boolean halfCloseReceived;
+    private volatile boolean halfCloseDeferred;
     private volatile ServerCall.Listener<InputStream> delegate;
 
     private DataPlaneServerListener(DataPlaneServerCall dataPlaneServerCall) {
@@ -1164,6 +1165,9 @@ void onReadyNotify() {
     @Override
     public void onMessage(InputStream message) {
       dataPlaneServerCall.delegateExecutor.execute(() -> {
+        if (dataPlaneServerCall.requestSideClosed.get()) {
+          return;
+        }
         ServerCall.Listener<InputStream> del = delegate;
         if (dataPlaneServerCall.passThroughMode.get() && del != null) {
           dataPlaneServerCall.callContext.run(() -> del.onMessage(message));
@@ -1177,10 +1181,10 @@ public void onMessage(InputStream message) {
                 != ProcessingMode.BodySendMode.GRPC
             || dataPlaneServerCall.config.getObservabilityMode()) {
 
-          if (dataPlaneServerCall.dataPlaneCallState.get() == DataPlaneCallState.IDLE) {
-            // We must buffer because the application call hasn't started yet
+          if (del == null) {
+            // We must buffer because the application call hasn't started yet or delegate is not set
             savedMessages.add(message);
-          } else if (del != null) {
+          } else {
             dataPlaneServerCall.callContext.run(() -> del.onMessage(message));
           }
           return;
@@ -1201,6 +1205,9 @@ public void onMessage(InputStream message) {
     @Override
     public void onHalfClose() {
       dataPlaneServerCall.delegateExecutor.execute(() -> {
+        if (dataPlaneServerCall.requestSideClosed.get()) {
+          return;
+        }
         dataPlaneServerCall.clientHalfCloseStartNanos = System.nanoTime();
         dataPlaneServerCall.halfClosed.set(true);
         ServerCall.Listener<InputStream> del = delegate;
@@ -1210,7 +1217,7 @@ public void onHalfClose() {
         }
 
         if (dataPlaneServerCall.dataPlaneCallState.get() == DataPlaneCallState.IDLE) {
-          halfCloseReceived = true;
+          halfCloseDeferred = true;
           return;
         }
 
@@ -1223,7 +1230,22 @@ public void onHalfClose() {
       });
     }
 
+    void handleDeferredHalfClose() {
+      dataPlaneServerCall.delegateExecutor.execute(() -> {
+        if (dataPlaneServerCall.currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE
+            || dataPlaneServerCall.isExtProcStreamCompleted()) {
+          proceedWithHalfClose();
+        } else {
+          sendRequestBodyToExtProc(null, true);
+        }
+      });
+    }
+
     void proceedWithHalfClose() {
+      if (!dataPlaneServerCall.requestSideClosed.compareAndSet(false, true)) {
+        return;
+      }
+      halfCloseReceived = true;
       if (dataPlaneServerCall.clientHalfCloseStartNanos > 0) {
         long durationNanos = System.nanoTime() - dataPlaneServerCall.clientHalfCloseStartNanos;
         dataPlaneServerCall.recordDuration(clientHalfCloseDuration, durationNanos);
@@ -1233,12 +1255,15 @@ void proceedWithHalfClose() {
       if (del != null) {
         dataPlaneServerCall.callContext.run(del::onHalfClose);
       }
+      dataPlaneServerCall.checkEndOfStream();
     }
 
     void onExternalBody(ByteString body) {
       ServerCall.Listener<InputStream> del = delegate;
       if (del != null) {
         dataPlaneServerCall.callContext.run(() -> del.onMessage(new InboundZeroCopyInputStream(body)));
+      } else {
+        savedMessages.add(new InboundZeroCopyInputStream(body));
       }
     }
 
@@ -1252,8 +1277,10 @@ private void sendRequestBodyToExtProc(@Nullable ByteString bodyByteString, boole
       HttpBody.Builder bodyBuilder = HttpBody.newBuilder();
       if (bodyByteString != null) {
         bodyBuilder.setBody(bodyByteString);
+        bodyBuilder.setEndOfStream(endOfStream);
+      } else {
+        bodyBuilder.setEndOfStreamWithoutMessage(true);
       }
-      bodyBuilder.setEndOfStream(endOfStream);
 
       dataPlaneServerCall.sendToExtProc(ProcessingRequest.newBuilder()
           .setRequestBody(bodyBuilder.build())
diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorServerInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorServerInterceptorTest.java
