From e8e3c69b068eec06bf080c4d6a86eea2c419ea31 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BB=98=E5=85=B8?= <fudianchn@gmail.com>
Date: Fri, 28 Aug 2026 11:32:53 +0800
Subject: [PATCH] stub: drain pending reads on cancel to release their buffers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

BlockingClientCall.cancel() stopped all reads and writes without
draining the call executor, orphaning read-delivery tasks already
queued by the transport.  Those tasks hold the message buffers, which
are only released when the task runs and observes the cancelled
stream, so cancelling a call with an undelivered read leaked the
transport's ByteBufs.

Drain the executor in cancel() so the queued tasks run and their
buffers are released.

Fixes #12355

Signed-off-by: 付典 <fudianchn@gmail.com>
---
 .../java/io/grpc/stub/BlockingClientCall.java   | 12 +++++++++++-
 .../io/grpc/stub/BlockingClientCallTest.java    | 17 +++++++++++++++++
 2 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
index 6a52ce50776..a68380dc2e5 100644
--- a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
+++ b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java
@@ -241,7 +241,8 @@ void sendSingleRequest(ReqT request) {
 
   /**
    * Cancel stream and stop any further writes.  Note that some reads that are in flight may still
-   * happen after the cancel.
+   * happen after the cancel.  Reads that have been received but not yet delivered are processed
+   * when cancelling, releasing their resources.
    *
    * @param message if not {@code null}, will appear as the description of the CANCELLED status
    * @param cause if not {@code null}, will appear as the cause of the CANCELLED status
@@ -249,6 +250,10 @@ void sendSingleRequest(ReqT request) {
   public void cancel(String message, Throwable cause) {
     writeClosed = true;
     call.cancel(message, cause);
+    // Reads queued by the transport hold buffers that are only released when their delivery task
+    // runs and observes the cancelled stream.  Nothing else will drain the executor once the user
+    // abandons the call, so the queued tasks must be processed here.
+    executor.drain();
   }
 
   /**
@@ -278,6 +283,11 @@ Status getClosedStatus() {
     return (state == null) ? null : state.status;
   }
 
+  @VisibleForTesting
+  ThreadSafeThreadlessExecutor getExecutor() {
+    return executor;
+  }
+
   /**
    * Check for whether some action is ready.
    *
diff --git a/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java b/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java
index e3a4f90e2c2..abb1e1661d7 100644
--- a/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java
+++ b/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java
@@ -221,6 +221,23 @@ public void testCancel() throws Exception {
 
   }
 
+  @Test
+  public void testCancelDrainsPendingReads() throws Exception {
+    biDiStream = ClientCalls.blockingBidiStreamingCall(channel, BIDI_STREAMING_METHOD,
+        CallOptions.DEFAULT);
+
+    // The server delivering a message queues a read-delivery task in the call executor, which
+    // only runs when the user reads/writes or the call is cancelled
+    testMethod.sendValueToClient(60);
+    assertThat(biDiStream.getExecutor()).isNotEmpty();
+
+    biDiStream.cancel("done reading", null);
+
+    // The queued read task must be processed by cancel() itself; when orphaned it holds the
+    // message's transport buffers, which leak (see #12355)
+    assertThat(biDiStream.getExecutor()).isEmpty();
+  }
+
   @Test
   public void testIsActivityReady() throws Exception {
     biDiStream = ClientCalls.blockingBidiStreamingCall(channel,  BIDI_STREAMING_METHOD,