xds: Fix TSAN data race on ClientCall cancellation in ExternalProcessorClientInterceptor

Prevent concurrent cancellations of the underlying ClientCall in ExternalProcessorClientInterceptor:
- Wrap rawCall with SimpleForwardingClientCall using an AtomicBoolean to ensure the underlying ClientCall.cancel() is executed at most once, even if invoked concurrently across threads or from DelayedListener.
- In DataPlaneClientCall.cancel(), atomically transition extProcStreamState to FAILED, catch exceptions during onError(), and clear extProcClientCallRequestObserver.
- In sendToExtProc(), return early if the ext-proc stream is already completed or the observer is null, and catch unexpected onNext() exceptions to trigger internalOnError() rather than letting exceptions escape into listener callbacks.
- Safely complete and clear extProcClientCallRequestObserver in closeExtProcStream() and halfCloseExtProcStream().
- Route all rawCall.cancel() calls in sendMessage(), handleImmediateResponse(), and DataPlaneListener through cancelDownstream().

Author: kannanjgithub

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

@@ -238,8 +238,18 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
     MethodDescriptor<InputStream, InputStream> rawMethod =
         (MethodDescriptor<InputStream, InputStream>) (MethodDescriptor<?, ?>) method;
     ClientCall<InputStream, InputStream> rawCall =
-        (ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
-            next.newCall(method, callOptions);
+        new SimpleForwardingClientCall<InputStream, InputStream>(
+            (ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
+                next.newCall(method, callOptions)) {
+          private final AtomicBoolean cancelled = new AtomicBoolean(false);
+
+          @Override
+          public void cancel(@Nullable String message, @Nullable Throwable cause) {
+            if (cancelled.compareAndSet(false, true)) {
+              super.cancel(message, cause);
+            }
+          }
+        };
 
     // Create a local subclass instance to buffer outbound actions
     DataPlaneDelayedCall<InputStream, InputStream> delayedCall =
@@ -397,13 +407,18 @@ private boolean validateCompressionSupport(BodyResponse bodyResponse) {
               .withDescription("gRPC message compression not supported in ext_proc")
               .asRuntimeException();
           synchronized (streamLock) {
-            if (!extProcStreamState.get().isCompleted()
-                && extProcClientCallRequestObserver != null) {
-              extProcClientCallRequestObserver.onError(ex);
+            if (markExtProcStreamFailed(extProcStreamState)) {
+              if (extProcClientCallRequestObserver != null) {
+                try {
+                  extProcClientCallRequestObserver.onError(ex);
+                } catch (Throwable ignored) {
+                  // Ignore exceptions during cancel/onError propagation
+                }
+                extProcClientCallRequestObserver = null;
+              }
             }
           }
           activateCall();
-          markExtProcStreamFailed(extProcStreamState);
           cancelDownstream("gRPC message compression not supported in ext_proc", ex);
           closeExtProcStream();
           return false;
@@ -600,6 +615,9 @@ public void onError(Throwable t) {
         @Override
         public void onCompleted() {
           if (markExtProcStreamCompleted(extProcStreamState)) {
+            synchronized (streamLock) {
+              extProcClientCallRequestObserver = null;
+            }
             handleFailOpen(wrappedListener);
           }
         }
@@ -628,8 +646,9 @@ public void onCompleted() {
     }
 
     private void sendToExtProc(ProcessingRequest request) {
+      Throwable sendError = null;
       synchronized (streamLock) {
-        if (extProcStreamState.get().isCompleted()) {
+        if (extProcStreamState.get().isCompleted() || extProcClientCallRequestObserver == null) {
           return;
         }
 
@@ -670,7 +689,14 @@ private void sendToExtProc(ProcessingRequest request) {
               .build();
         }
 
-        extProcClientCallRequestObserver.onNext(requestToSend);
+        try {
+          extProcClientCallRequestObserver.onNext(requestToSend);
+        } catch (Throwable t) {
+          sendError = t;
+        }
+      }
+      if (sendError != null) {
+        internalOnError(sendError);
       }
     }
 
@@ -690,7 +716,11 @@ private void closeExtProcStream() {
       synchronized (streamLock) {
         if (markExtProcStreamCompleted(extProcStreamState)) {
           if (extProcClientCallRequestObserver != null) {
-            extProcClientCallRequestObserver.onCompleted();
+            try {
+              extProcClientCallRequestObserver.onCompleted();
+            } catch (Throwable ignored) {
+            }
+            extProcClientCallRequestObserver = null;
           }
         }
       }
@@ -722,7 +752,10 @@ private void internalOnError(Throwable t) {
     private void halfCloseExtProcStream() {
       synchronized (streamLock) {
         if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
-          extProcClientCallRequestObserver.onCompleted();
+          try {
+            extProcClientCallRequestObserver.onCompleted();
+          } catch (Throwable ignored) {
+          }
         }
       }
     }
@@ -809,7 +842,7 @@ public void sendMessage(InputStream message) {
             ByteString copiedBody = ByteString.readFrom(message);
             pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
           } catch (IOException e) {
-            rawCall.cancel("Failed to copy outbound message for buffering", e);
+            cancelDownstream("Failed to copy outbound message for buffering", e);
           }
           return;
         }
@@ -835,7 +868,7 @@ public void sendMessage(InputStream message) {
           super.sendMessage(new KnownLengthInputStream(bodyByteString));
         }
       } catch (IOException e) {
-        rawCall.cancel("Failed to serialize message for External Processor", e);
+        cancelDownstream("Failed to serialize message for External Processor", e);
       }
     }
 
@@ -900,7 +933,7 @@ public void halfClose() {
           .build());
     }
 
-    private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
+    void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
       if (downstreamCancelled.compareAndSet(false, true)) {
         delayedCall.cancel(message, cause);
       }
@@ -909,12 +942,19 @@ private void cancelDownstream(@Nullable String message, @Nullable Throwable caus
     @Override
     public void cancel(@Nullable String message, @Nullable Throwable cause) {
       synchronized (streamLock) {
-        if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
-          extProcClientCallRequestObserver.onError(
-              Status.CANCELLED
-                  .withDescription(message)
-                  .withCause(cause)
-                  .asRuntimeException());
+        if (markExtProcStreamFailed(extProcStreamState)) {
+          if (extProcClientCallRequestObserver != null) {
+            try {
+              extProcClientCallRequestObserver.onError(
+                  Status.CANCELLED
+                      .withDescription(message)
+                      .withCause(cause)
+                      .asRuntimeException());
+            } catch (Throwable ignored) {
+              // Ignore exceptions during cancel/onError propagation
+            }
+            extProcClientCallRequestObserver = null;
+          }
         }
       }
       cancelDownstream(message, cause);
@@ -970,7 +1010,7 @@ private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListe
         // If sent in response to any other event, it will cause the data plane RPC to
         // immediately fail with the specified status as if it were an out-of-band
         // cancellation.
-        rawCall.cancel(status.getDescription(), null);
+        cancelDownstream(status.getDescription(), null);
         listener.unblockAfterStreamComplete();
       }
       closeExtProcStream();
@@ -1153,7 +1193,7 @@ public void onMessage(InputStream message) {
             ByteString copiedBody = ByteString.readFrom(message);
             savedMessages.add(new KnownLengthInputStream(copiedBody));
           } catch (IOException e) {
-            rawCall.cancel("Failed to copy inbound message for buffering", e);
+            dataPlaneClientCall.cancelDownstream("Failed to copy inbound message for buffering", e);
           }
           return;
         }
@@ -1184,7 +1224,7 @@ public void onMessage(InputStream message) {
               () -> delegate().onMessage(bodyByteString.newInput()));
         }
       } catch (IOException e) {
-        rawCall.cancel("Failed to read server response", e);
+        dataPlaneClientCall.cancelDownstream("Failed to read server response", e);
       }
     }