xds: Fix half-close race condition in ext-proc client interceptor (#12888)

Resolve a protocol violation bug in ExternalProcessorClientInterceptor
and fix a related data race in its fail-open unit tests:

1. Interceptor Bug Fix (Incorrect Frame Order):
When the external processor stream fails (extProcStreamState is
FAILED/completed)
but the interceptor has not yet transitioned to pass-through mode, a
call to
sendMessage() is buffered, but a call to halfClose() immediately
executed
super.halfClose() because it only checked stream completion. This sent
the
halfClose frame to the backend server before the buffered message was
drained,
violating the gRPC protocol and causing the backend to drop the message.
   
Fixed by modifying halfClose() to only execute immediately if
passThroughMode
is true. If passThroughMode is false and the stream is completed/failed,
halfClose is buffered (pendingHalfClose = true) and deferred until the
draining
   task drains the buffered messages and triggers it.

2. Test Fix (Deterministic Race Simulation):
Redesigned
givenFailureModeAllowTrue_whenExtProcStreamFails_thenCallFailsOpen
in ExternalProcessorClientInterceptorTest to deterministically simulate
the
interleaving that caused the bug. Used a custom ServerInterceptor on the
backend server to suspend the async onError thread inside activateLine()
(midway through fail-open processing) while the client main thread calls
sendMessage() and halfClose(). This forces the race condition and
verifies
   that the message is no longer dropped.

Author: kannanjgithub

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

@@ -841,7 +841,7 @@ private void proceedWithHalfClose() {
     @Override
     public void halfClose() {
       clientHalfCloseStartNanos = System.nanoTime();
-      if (passThroughMode.get() || extProcStreamState.get().isCompleted()) {
+      if (passThroughMode.get()) {
         if (requestSideClosed.compareAndSet(false, true)) {
           proceedWithHalfClose();
         }
@@ -850,6 +850,10 @@ public void halfClose() {
 
       pendingHalfClose.set(true);
 
+      if (extProcStreamState.get().isCompleted()) {
+        return;
+      }
+
       if (extProcStreamState.get().isDraining()) {
         return;
       }
@@ -867,8 +871,6 @@ public void halfClose() {
               .setEndOfStreamWithoutMessage(true)
               .build())
           .build());
-      
-      // Defer super.halfClose() until ext-proc response signals end_of_stream.
     }
 
     @Override

xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java

@@ -8460,22 +8460,43 @@ public void onCompleted() {
         filterConfig, channelManager, scheduler, FAKE_CONTEXT);
 
     final CountDownLatch dataPlaneLatch = new CountDownLatch(1);
-    dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
-        .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
-            (request, responseObserver) -> {
-              responseObserver.onNext("Hello " + request);
-              responseObserver.onCompleted();
-              dataPlaneLatch.countDown();
-            }))
-        .build());
+    final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
+    final CountDownLatch resumeAsyncThreadLatch = new CountDownLatch(1);
+
+    ServerInterceptor dataPlaneInterceptor = new ServerInterceptor() {
+      @Override
+      public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+          ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+        headersReceivedLatch.countDown();
+        try {
+          resumeAsyncThreadLatch.await();
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        }
+        return next.startCall(call, headers);
+      }
+    };
+
+    dataPlaneServiceRegistry.addService(ServerInterceptors.intercept(
+        ServerServiceDefinition.builder("test.TestService")
+            .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
+                (request, responseObserver) -> {
+                  responseObserver.onNext("Hello " + request);
+                  responseObserver.onCompleted();
+                  dataPlaneLatch.countDown();
+                }))
+            .build(),
+        dataPlaneInterceptor));
 
     ManagedChannel dataPlaneChannel = grpcCleanup.register(
         InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
 
+    final AtomicReference<Status> statusRef = new AtomicReference<>();
     final CountDownLatch closedLatch = new CountDownLatch(1);
     ClientCall.Listener<String> appListener = new ClientCall.Listener<String>() {
       @Override
       public void onClose(Status status, Metadata trailers) {
+        statusRef.set(status);
         closedLatch.countDown();
       }
     };
@@ -8485,15 +8506,27 @@ public void onClose(Status status, Metadata trailers) {
         interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
     proxyCall.start(appListener, new Metadata());
 
-    // Send message and half-close to trigger unary call reaching server
+    // Trigger unary call. request(1) starts it.
     proxyCall.request(1);
+
+    // Wait for the async sidecar thread to enter activateCall() and block inside interceptCall
+    assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // Now, while the async thread is blocked (and passThroughMode is still false),
+    // send a message and half-close.
     proxyCall.sendMessage("test");
     proxyCall.halfClose();
 
+    // Unblock the async thread
+    resumeAsyncThreadLatch.countDown();
+
     // Verify data plane call reached (failed open)
     assertThat(dataPlaneLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // Verify client call completes successfully
+    assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(statusRef.get().isOk()).isTrue();
 
-    proxyCall.cancel("Cleanup", null);
     channelManager.close();
   }