xds: Fix half-close race condition in ext-proc client interceptor

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

@@ -8451,75 +8451,78 @@ public void onCompleted() {
         .directExecutor()
         .build().start());
 
-    ExecutorService extProcChannelExecutor = Executors.newSingleThreadExecutor();
-    try {
-      CachedChannelManager channelManager = new CachedChannelManager(config -> {
-        return grpcCleanup.register(
-            InProcessChannelBuilder.forName(extProcServerName)
-                .executor(extProcChannelExecutor)
-                .build());
-      });
+    CachedChannelManager channelManager = new CachedChannelManager(config -> {
+      return grpcCleanup.register(
+          InProcessChannelBuilder.forName(extProcServerName).directExecutor().build());
+    });
 
-      ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
-          filterConfig, channelManager, scheduler, FAKE_CONTEXT);
-
-      final CountDownLatch dataPlaneLatch = new CountDownLatch(1);
-      final CountDownLatch headersReceivedLatch = 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();
-          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));
+    ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
+        filterConfig, channelManager, scheduler, FAKE_CONTEXT);
 
-      ManagedChannel dataPlaneChannel = grpcCleanup.register(
-          InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
+    final CountDownLatch dataPlaneLatch = new CountDownLatch(1);
+    final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
+    final CountDownLatch resumeAsyncThreadLatch = new CountDownLatch(1);
 
-      final CountDownLatch closedLatch = new CountDownLatch(1);
-      ClientCall.Listener<String> appListener = new ClientCall.Listener<String>() {
-        @Override
-        public void onClose(Status status, Metadata trailers) {
-          closedLatch.countDown();
+    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();
         }
-      };
-      
-      CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor());
-      ClientCall<String, String> proxyCall =
-          interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
-      proxyCall.start(appListener, new Metadata());
+        return next.startCall(call, headers);
+      }
+    };
 
-      // Send message and half-close to trigger unary call reaching server
-      proxyCall.request(1);
-      assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue();
-      
-      // Wait for the onError execution on extProcChannelExecutor to finish completely
-      final CountDownLatch executorSyncLatch = new CountDownLatch(1);
-      extProcChannelExecutor.execute(executorSyncLatch::countDown);
-      assertThat(executorSyncLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    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));
 
-      proxyCall.sendMessage("test");
-      proxyCall.halfClose();
+    ManagedChannel dataPlaneChannel = grpcCleanup.register(
+        InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
 
-      // Verify data plane call reached (failed open)
-      assertThat(dataPlaneLatch.await(5, TimeUnit.SECONDS)).isTrue();
-      
-      proxyCall.cancel("Cleanup", null);
-      channelManager.close();
-    } finally {
-      extProcChannelExecutor.shutdown();
-    }
+    final CountDownLatch closedLatch = new CountDownLatch(1);
+    ClientCall.Listener<String> appListener = new ClientCall.Listener<String>() {
+      @Override
+      public void onClose(Status status, Metadata trailers) {
+        closedLatch.countDown();
+      }
+    };
+    
+    CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor());
+    ClientCall<String, String> proxyCall =
+        interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
+    proxyCall.start(appListener, new Metadata());
+
+    // 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();
+    
+    proxyCall.cancel("Cleanup", null);
+    channelManager.close();
   }
 
   @Test