From fc1db056dd2361311beccaf61ef8982e35529262 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 6a52ce50776..69bae1d1943 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();
+      }
     }
   }