xds: synchronize ext_proc message draining during stream termination

I identified the following synchronization issues by analyzing the potential race conditions between the thread handling external processor responses (or stream completion) and the application thread calling sendMessage().

Specifically, here is how the two major race conditions occurred during state transitions:

1. The isExtProcStreamCompleted() Check Bypass Race
In the previous implementation, sendMessage(InputStream message) started with:

java
if (passThroughMode.get() || isExtProcStreamCompleted()) {
  super.sendMessage(message);
  return;
}
When the external processor stream terminated (e.g., via onCompleted()), the event listener immediately called markExtProcStreamCompleted(). This updated the extProcStreamState atomic reference to COMPLETED before handleFailOpen() ran drainPendingDrainingMessages(). If an application thread called sendMessage() in that exact window, isExtProcStreamCompleted() returned true. The application thread bypassed pendingDrainingMessages and sent the message directly over the wire (super.sendMessage). Shortly after, handleFailOpen() flushed the queue, delivering previously buffered draining messages to the wire after the newer message, resulting in out-of-order (FIFO violation) message delivery.

2. The Queue Flushed / Preemption Race
If an application thread entered sendMessage() while the stream was in the DRAINING state, it evaluated:

java
if (isExtProcStreamDraining()) {
  pendingDrainingMessages.add(message);
  return;
}
If the application thread was preempted right before executing pendingDrainingMessages.add(message), the external processor stream could complete in the background on another thread. E.g., handleFailOpen() would run, invoke drainPendingDrainingMessages() (which found the queue empty), set passThroughMode = true, and finish. When the application thread resumed, it added its message to pendingDrainingMessages. Because drainPendingDrainingMessages() had already executed, that message sat in the queue indefinitely and was dropped.

The Solution
By wrapping the state checks and queue additions in sendMessage() with synchronized (streamLock), and similarly synchronizing drainPendingDrainingMessages() (where passThroughMode = true is updated atomically with the queue flush), we eliminate both race windows and ensure 100% robust, FIFO-ordered delivery across all stream transitions.

Author: kannanjgithub

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

@@ -1344,14 +1344,21 @@ public void sendMessage(InputStream message) {
           return;
         }
 
-        if (passThroughMode.get() || isExtProcStreamCompleted()) {
+        if (passThroughMode.get()) {
           super.sendMessage(message);
           return;
         }
 
-        if (isExtProcStreamDraining()) {
-          pendingDrainingMessages.add(message);
-          return;
+        synchronized (streamLock) {
+          if (passThroughMode.get()) {
+            super.sendMessage(message);
+            return;
+          }
+
+          if (isExtProcStreamDraining() || isExtProcStreamCompleted()) {
+            pendingDrainingMessages.add(message);
+            return;
+          }
         }
 
         if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) {
@@ -1495,9 +1502,12 @@ private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListe
       }
 
       private void drainPendingDrainingMessages() {
-        InputStream msg;
-        while ((msg = pendingDrainingMessages.poll()) != null) {
-          super.sendMessage(msg);
+        synchronized (streamLock) {
+          passThroughMode.set(true);
+          InputStream msg;
+          while ((msg = pendingDrainingMessages.poll()) != null) {
+            super.sendMessage(msg);
+          }
         }
       }
 
@@ -1722,7 +1732,7 @@ void onExternalBody(ByteString body) {
 
       void unblockAfterStreamComplete() {
         proceedWithHeaders();
-        dataPlaneClientCall.passThroughMode.set(true);
+        dataPlaneClientCall.drainPendingDrainingMessages();
         proceedWithClose();
       }