xds: Implement fail-open queue draining and optimize locks in ext_proc interceptor

- Implement fail-open queue draining for both inbound and outbound directions.
- Defer transport half-close during fail-open until all outbound queues are drained.
- Refactor DataPlaneClientCall.onReady to release streamLock before calling drain/update methods, avoiding deadlocks.
- Consolidate and batch outbound draining in drainPendingDrainingMessages.

Author: kannanjgithub

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

@@ -792,7 +792,8 @@ private void onExtProcStreamReady() {
     void drainPendingRequests() {
       synchronized (streamLock) {
         if (config.getObservabilityMode()
-            || currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
+            || currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC
+            || extProcStreamState.get().isCompleted()) {
           int toRequest = pendingRequests.getAndSet(0);
           if (toRequest > 0) {
             super.request(toRequest);
@@ -853,6 +854,37 @@ private void onReadyNotify() {
       wrappedListener.onReadyNotify();
     }
 
+    void onReady() {
+      boolean isPassThrough;
+      boolean isCompleted;
+      boolean isDraining;
+      
+      synchronized (streamLock) {
+        isPassThrough = passThroughMode.get();
+        ExtProcStreamState state = extProcStreamState.get();
+        isCompleted = state.isCompleted();
+        isDraining = state.isDraining();
+      }
+
+      if (isPassThrough) {
+        onReadyNotify();
+        return;
+      }
+      
+      if (isCompleted) {
+        drainPendingDrainingMessages();
+        return;
+      }
+      
+      // Normal or Draining operation
+      drainPendingUpstreamBodyMessages();
+      if (!isDraining) {
+        trySendAccumulatedWindowUpdates();
+      }
+      drainPendingRequests();
+      onReadyNotify();
+    }
+
     boolean isSidecarReady() {
       ExtProcStreamState state = extProcStreamState.get();
       if (state.isCompleted()) {
@@ -1109,7 +1141,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
             boolean sendImmediately = false;
             synchronized (streamLock) {
               sidestreamToUpstreamWindow -= body.size();
-              if (super.isReady() && pendingUpstreamBodyMessages.isEmpty()) {
+              if (pendingUpstreamBodyMessages.isEmpty() && super.isReady()) {
                 sendImmediately = true;
                 accumulatedWindowUpdateSidestreamToUpstream += body.size();
               } else {
@@ -1202,29 +1234,44 @@ private void drainPendingMutatedResponseBodies() {
       }
     }
 
+    void drainPendingMutatedResponseBodiesDirect(DataPlaneListener listener) {
+      List<ByteString> toDeliver = new ArrayList<>();
+      synchronized (streamLock) {
+        ByteString body;
+        while ((body = pendingMutatedResponseBodies.poll()) != null) {
+          toDeliver.add(body);
+        }
+      }
+      for (ByteString body : toDeliver) {
+        listener.onExternalBody(body);
+      }
+    }
+
     void drainPendingUpstreamBodyMessages() {
-      while (true) {
-        ByteString body = null;
-        boolean triggerHalfClose = false;
-        synchronized (streamLock) {
-          if (super.isReady() && !pendingUpstreamBodyMessages.isEmpty()) {
-            body = pendingUpstreamBodyMessages.poll();
+      List<ByteString> toSend = new ArrayList<>();
+      boolean triggerHalfClose = false;
+      synchronized (streamLock) {
+        if (super.isReady()) {
+          ByteString body;
+          while ((body = pendingUpstreamBodyMessages.poll()) != null) {
+            toSend.add(body);
             accumulatedWindowUpdateSidestreamToUpstream += body.size();
-            if (pendingUpstreamBodyMessages.isEmpty()
-                && pendingUpstreamHalfClose.compareAndSet(true, false)) {
-              triggerHalfClose = true;
-            }
+          }
+          if (pendingUpstreamBodyMessages.isEmpty()
+              && pendingUpstreamHalfClose.compareAndSet(true, false)) {
+            triggerHalfClose = true;
           }
         }
-        if (body == null) {
-          break;
-        }
+      }
+      for (ByteString body : toSend) {
         super.sendMessage(new KnownLengthInputStream(body));
+      }
+      if (!toSend.isEmpty()) {
         trySendAccumulatedWindowUpdates();
-        if (triggerHalfClose) {
-          if (requestSideClosed.compareAndSet(false, true)) {
-            proceedWithHalfClose();
-          }
+      }
+      if (triggerHalfClose) {
+        if (requestSideClosed.compareAndSet(false, true)) {
+          proceedWithHalfClose();
         }
       }
     }
@@ -1258,18 +1305,58 @@ private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListe
     }
 
     private void drainPendingDrainingMessages() {
+      List<ByteString> mutatedToSend = new ArrayList<>();
+      List<ByteString> blockedToSend = new ArrayList<>();
+      List<InputStream> drainingToSend = new ArrayList<>();
+      boolean fullyDrained = false;
+      boolean triggerHalfClose = false;
+
       synchronized (streamLock) {
-        InputStream msg;
-        while ((msg = pendingDrainingMessages.poll()) != null) {
-          super.sendMessage(msg);
+        fullyDrained = pendingUpstreamBodyMessages.isEmpty()
+            && pendingRequestBodyMessages.isEmpty()
+            && pendingDrainingMessages.isEmpty();
+
+        if (!fullyDrained && super.isReady()) {
+          ByteString body;
+          while ((body = pendingUpstreamBodyMessages.poll()) != null) {
+            mutatedToSend.add(body);
+          }
+          while ((body = pendingRequestBodyMessages.poll()) != null) {
+            blockedToSend.add(body);
+          }
+          InputStream msg;
+          while ((msg = pendingDrainingMessages.poll()) != null) {
+            drainingToSend.add(msg);
+          }
+          fullyDrained = pendingUpstreamBodyMessages.isEmpty()
+              && pendingRequestBodyMessages.isEmpty()
+              && pendingDrainingMessages.isEmpty();
         }
-        passThroughMode.set(true);
-        if (pendingHalfClose.get()) {
-          if (requestSideClosed.compareAndSet(false, true)) {
-            proceedWithHalfClose();
+
+        if (fullyDrained) {
+          passThroughMode.set(true);
+          if (pendingHalfClose.get()) {
+            triggerHalfClose = true;
           }
         }
       }
+
+      // Send messages outside the streamLock to prevent potential deadlocks
+      for (ByteString body : mutatedToSend) {
+        super.sendMessage(new KnownLengthInputStream(body));
+      }
+      for (ByteString body : blockedToSend) {
+        super.sendMessage(new KnownLengthInputStream(body));
+      }
+      for (InputStream msg : drainingToSend) {
+        super.sendMessage(msg);
+      }
+
+      if (triggerHalfClose) {
+        if (requestSideClosed.compareAndSet(false, true)) {
+          proceedWithHalfClose();
+        }
+      }
     }
 
     private void handleFailOpen(DataPlaneListener listener) {
@@ -1380,10 +1467,7 @@ void setImmediateResponse(Status status, Metadata trailers) {
 
     @Override
     public void onReady() {
-      dataPlaneClientCall.drainPendingUpstreamBodyMessages();
-      dataPlaneClientCall.trySendAccumulatedWindowUpdates();
-      dataPlaneClientCall.drainPendingRequests();
-      onReadyNotify();
+      dataPlaneClientCall.onReady();
     }
 
     @Override
@@ -1608,7 +1692,11 @@ void onExternalBody(ByteString body) {
 
     void unblockAfterStreamComplete() {
       proceedWithHeaders();
+      // 1. Drain mutated responses first
+      dataPlaneClientCall.drainPendingMutatedResponseBodiesDirect(this);
+      // 2. Drain raw responses
       proceedWithSavedMessages();
+      // 3. Drain outbound requests
       dataPlaneClientCall.drainPendingDrainingMessages();
       proceedWithClose();
     }