From 6a6caf98bb18a881a0b1a6de4e4663df1d191959 Mon Sep 17 00:00:00 2001
From: Benjamin Peterson <benjamin@engflow.com>
Date: Mon, 22 Sep 2025 09:07:59 -0700
Subject: [PATCH] stub: ensure BlockingClientCall tasks run after cancellation

After canceling a `BlockingClientCall`, the caller may not interact with it further. That means the call executor, a `ThreadSafeThreadlessExecutor`, will not execute any more tasks. There may still be tasks submitted to the call executor, though, until the underlying call completes. Some of these tasks (e.g., server messages available) may leak native resources unless executed. So, we convert the executor to a "direct" executor during cancellation in order to ensure all call tasks run.

Fixes https://github.com/grpc/grpc-java/issues/12355.
---
 .../java/io/grpc/stub/BlockingClientCall.java |  7 +++++
 .../main/java/io/grpc/stub/ClientCalls.java   | 28 +++++++++++++++++--
 2 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
index 27bd42e53bb..e0390a7475b 100644
--- a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
+++ b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
@@ -249,6 +249,13 @@ void sendSingleRequest(ReqT request) {
   public void cancel(String message, Throwable cause) {
     writeClosed = true;
     call.cancel(message, cause);
+    // After canceling a BlockingClientCall, the caller may not interact with it further. That means
+    // the call executor, a ThreadSafeThreadlessExecutor, will not execute any more tasks. There may
+    // still be tasks submitted to the call executor, though, until the underlying call completes.
+    // Some of these tasks (e.g., server messages available) may leak native resources unless
+    // executed. So, we convert the executor to a "direct" executor in order to ensure all call
+    // tasks run.
+    executor.becomeDirect();
   }
 
   /**
diff --git a/stub/src/main/java/io/grpc/stub/ClientCalls.java b/stub/src/main/java/io/grpc/stub/ClientCalls.java
index ff2804a0a1f..6432523a9bb 100644
--- a/stub/src/main/java/io/grpc/stub/ClientCalls.java
+++ b/stub/src/main/java/io/grpc/stub/ClientCalls.java
@@ -871,6 +871,7 @@ static final class ThreadSafeThreadlessExecutor extends ConcurrentLinkedQueue<Ru
 
     private final Lock waiterLock = new ReentrantLock();
     private final Condition waiterCondition = waiterLock.newCondition();
+    private boolean direct;
 
     // Non private to avoid synthetic class
     ThreadSafeThreadlessExecutor() {}
@@ -950,6 +951,20 @@ void drain() {
       }
     }
 
+    /**
+     * Turn the executor into a "direct" executor. Further calls to {@link #execute(Runnable)} will
+     * executable the runnable immediately in the calling thread.
+     */
+    void becomeDirect() {
+      waiterLock.lock();
+      try {
+        direct = true;
+      } finally {
+        waiterLock.unlock();
+      }
+      drain();
+    }
+
     private void signalAll() {
       waiterLock.lock();
       try {
@@ -975,13 +990,22 @@ private static void throwIfInterrupted() throws InterruptedException {
 
     @Override
     public void execute(Runnable runnable) {
+      boolean added = false;
       waiterLock.lock();
       try {
-        add(runnable);
-        waiterCondition.signalAll(); // If anything is waiting let it wake up and process this task
+        if (!direct) {
+          added = true;
+          add(runnable);
+          // If anything is waiting, let it wake up and process this task.
+          waiterCondition.signalAll();
+        }
       } finally {
         waiterLock.unlock();
       }
+      if (!added) {
+        runQuietly(runnable);
+        signalAll();
+      }
     }
   }