fix(firestore): guard against gRPC call was cancelled exceptions

firebase-firestore/CHANGELOG.md

@@ -1,5 +1,7 @@
 # Unreleased
 
+- [changed] Guard gRPC `call was cancelled exceptions` (#8608). Fixes #8601.
+
 # 26.6.0
 
 - [feature] Implemented support for retrieving documents up to 16MB over gRPC (#8363)

firebase-firestore/gradle.properties

@@ -1,2 +1,2 @@
-version=26.6.1
+version=26.6.2
 latestReleasedVersion=26.6.0

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java

@@ -409,7 +409,21 @@ protected void writeRequest(ReqT message) {
         System.identityHashCode(this),
         message);
     cancelIdleCheck();
-    call.sendMessage(message);
+    if (call != null) {
+      try {
+        call.sendMessage(message);
+      } catch (IllegalStateException e) {
+        if (e.getMessage() != null && e.getMessage().contains("call was cancelled")) {
+          Logger.debug(
+              getClass().getSimpleName(),
+              "(%x) Stream writeRequest failed because call was cancelled: [%s]",
+              System.identityHashCode(this),
+              e);
+        } else {
+          throw e;
+        }
+      }
+    }
   }
 
   /** Called by the idle timer when the stream should close due to inactivity. */

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java

@@ -38,6 +38,7 @@
 import io.grpc.android.AndroidChannelBuilder;
 import io.grpc.okhttp.OkHttpChannelBuilder;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 /** Manages the gRPC channel and encapsulates all SSL and gRPC initialization. */
@@ -260,13 +261,35 @@ private void initChannelTask() {
             () -> {
               ManagedChannel channel = initChannel(context, databaseInfo);
               asyncQueue.enqueueAndForget(() -> onConnectivityStateChange(channel));
+              // Ensure all callbacks and internal delayed call drains are issued on the worker
+              // queue. Intercept 'call was cancelled' IllegalStateException from gRPC's internal
+              // DelayedClientCall.drainPendingCalls() to prevent crashing the AsyncQueue.
+              // See: https://github.com/firebase/firebase-android-sdk/issues/8601.
+              Executor guardedGrpcExecutor =
+                  command ->
+                      asyncQueue
+                          .getExecutor()
+                          .execute(
+                              () -> {
+                                try {
+                                  command.run();
+                                } catch (IllegalStateException e) {
+                                  String message = e.getMessage();
+                                  if (message != null && message.contains("call was cancelled")) {
+                                    Logger.debug(
+                                        LOG_TAG,
+                                        "Suppressed gRPC 'call was cancelled' exception: %s",
+                                        e);
+                                  } else {
+                                    throw e;
+                                  }
+                                }
+                              });
+
               FirestoreGrpc.FirestoreStub firestoreStub =
                   FirestoreGrpc.newStub(channel)
                       .withCallCredentials(firestoreHeaders)
-                      // Ensure all callbacks are issued on the worker queue. If this call is
-                      // removed, all calls need to be audited to make sure they are executed on the
-                      // right thread.
-                      .withExecutor(asyncQueue.getExecutor());
+                      .withExecutor(guardedGrpcExecutor);
               callOptions = firestoreStub.getCallOptions();
               Logger.debug(LOG_TAG, "Channel successfully reset.");
               return channel;

firebase-firestore/src/test/java/com/google/firebase/firestore/remote/AbstractStreamTest.java

@@ -0,0 +1,112 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.firestore.remote;
+
+import static org.junit.Assert.assertThrows;
+
+import com.google.firebase.firestore.remote.Stream.StreamCallback;
+import com.google.firebase.firestore.util.AsyncQueue;
+import com.google.firebase.firestore.util.AsyncQueue.TimerId;
+import io.grpc.ClientCall;
+import io.grpc.MethodDescriptor;
+import java.lang.reflect.Field;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.robolectric.RobolectricTestRunner;
+
+@RunWith(RobolectricTestRunner.class)
+public class AbstractStreamTest {
+
+  private static class TestStream extends AbstractStream<String, String, StreamCallback> {
+    TestStream(FirestoreChannel channel, AsyncQueue workerQueue, StreamCallback listener) {
+      super(
+          channel,
+          Mockito.mock(MethodDescriptor.class),
+          workerQueue,
+          TimerId.LISTEN_STREAM_CONNECTION_BACKOFF,
+          TimerId.LISTEN_STREAM_IDLE,
+          TimerId.HEALTH_CHECK_TIMEOUT,
+          listener);
+    }
+
+    @Override
+    public void onFirst(String change) {}
+
+    @Override
+    public void onNext(String change) {}
+
+    public void write(String message) {
+      writeRequest(message);
+    }
+  }
+
+  @Test
+  public void writeRequest_whenCallAlreadyCancelled_doesNotThrow() throws Exception {
+    AsyncQueue asyncQueue = new AsyncQueue();
+    FirestoreChannel channel = Mockito.mock(FirestoreChannel.class);
+    StreamCallback callback = Mockito.mock(StreamCallback.class);
+
+    TestStream stream = new TestStream(channel, asyncQueue, callback);
+
+    @SuppressWarnings("unchecked")
+    ClientCall<String, String> mockCall = Mockito.mock(ClientCall.class);
+
+    // Simulate gRPC ClientCallImpl.sendMessageInternal() throwing on cancelled call
+    Mockito.doThrow(new IllegalStateException("call was cancelled"))
+        .when(mockCall)
+        .sendMessage(Mockito.any());
+
+    Field callField = AbstractStream.class.getDeclaredField("call");
+    callField.setAccessible(true);
+    callField.set(stream, mockCall);
+
+    Field stateField = AbstractStream.class.getDeclaredField("state");
+    stateField.setAccessible(true);
+    stateField.set(stream, AbstractStream.State.Open);
+
+    // writeRequest() catches the IllegalStateException, logs it, and does not throw.
+    asyncQueue.runSync(() -> stream.write("test-message"));
+  }
+
+  @Test
+  public void writeRequest_whenOtherIllegalStateException_rethrows() throws Exception {
+    AsyncQueue asyncQueue = new AsyncQueue();
+    FirestoreChannel channel = Mockito.mock(FirestoreChannel.class);
+    StreamCallback callback = Mockito.mock(StreamCallback.class);
+
+    TestStream stream = new TestStream(channel, asyncQueue, callback);
+
+    @SuppressWarnings("unchecked")
+    ClientCall<String, String> mockCall = Mockito.mock(ClientCall.class);
+
+    // Simulate unexpected IllegalStateException (for example, invalid internal state)
+    Mockito.doThrow(new IllegalStateException("unrelated stream state error"))
+        .when(mockCall)
+        .sendMessage(Mockito.any());
+
+    Field callField = AbstractStream.class.getDeclaredField("call");
+    callField.setAccessible(true);
+    callField.set(stream, mockCall);
+
+    Field stateField = AbstractStream.class.getDeclaredField("state");
+    stateField.setAccessible(true);
+    stateField.set(stream, AbstractStream.State.Open);
+
+    // Non-cancellation IllegalStateException must be re-thrown by writeRequest().
+    assertThrows(
+        RuntimeException.class, () -> asyncQueue.runSync(() -> stream.write("test-message")));
+  }
+}

firebase-firestore/src/test/java/com/google/firebase/firestore/remote/GrpcCallProviderTest.java

@@ -15,6 +15,8 @@
 package com.google.firebase.firestore.remote;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 
 import android.content.Context;
 import androidx.test.core.app.ApplicationProvider;
@@ -23,9 +25,13 @@
 import com.google.firebase.firestore.util.AsyncQueue;
 import com.google.firebase.firestore.util.Supplier;
 import io.grpc.CallCredentials;
+import io.grpc.CallOptions;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.okhttp.OkHttpChannelBuilder;
 import java.lang.reflect.Field;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -94,4 +100,119 @@ public void configuresChannelFlowControlWindow() throws Exception {
 
     assertEquals(512 * 1024, flowControlWindow);
   }
+
+  @Test
+  public void grpcExecutor_whenCallCancelledDuringDrain_doesNotPanicAsyncQueue() throws Exception {
+    Context context = ApplicationProvider.getApplicationContext();
+    AsyncQueue asyncQueue = new AsyncQueue();
+    DatabaseId databaseId = DatabaseId.forProject("project");
+    DatabaseInfo databaseInfo = new DatabaseInfo(databaseId, "key", "host", true, 512 * 1024);
+
+    GrpcCallProvider grpcCallProvider =
+        new GrpcCallProvider(
+            asyncQueue, context, databaseInfo, Mockito.mock(CallCredentials.class));
+
+    // Await initialization of the channel task on a background thread
+    Field taskField = GrpcCallProvider.class.getDeclaredField("channelTask");
+    taskField.setAccessible(true);
+    com.google.android.gms.tasks.Task<?> task =
+        (com.google.android.gms.tasks.Task<?>) taskField.get(grpcCallProvider);
+
+    Thread thread =
+        new Thread(
+            () -> {
+              try {
+                com.google.android.gms.tasks.Tasks.await(task);
+              } catch (Exception e) {
+                // ignore
+              }
+            });
+    thread.start();
+    thread.join();
+
+    Field optionsField = GrpcCallProvider.class.getDeclaredField("callOptions");
+    optionsField.setAccessible(true);
+    CallOptions callOptions = (CallOptions) optionsField.get(grpcCallProvider);
+    Executor grpcExecutor = callOptions.getExecutor();
+
+    CountDownLatch drainExecuted = new CountDownLatch(1);
+    // Simulate gRPC DelayedClientCall throwing during drain
+    Runnable delayedDrainRunnable =
+        () -> {
+          try {
+            throw new IllegalStateException("call was cancelled");
+          } finally {
+            drainExecuted.countDown();
+          }
+        };
+
+    grpcExecutor.execute(delayedDrainRunnable);
+    assertTrue(drainExecuted.await(5, TimeUnit.SECONDS));
+
+    // Synchronously flush the AsyncQueue to ensure afterExecute() has completed.
+    asyncQueue.runSync(() -> {});
+
+    // GuardedGrpcExecutor catches the "call was cancelled" IllegalStateException, logs it,
+    // and prevents it from escaping into afterExecute() and tripping AsyncQueue.panic().
+    org.robolectric.shadows.ShadowLooper.idleMainLooper();
+  }
+
+  @Test
+  public void grpcExecutor_whenOtherExceptionDuringDrain_panicsAsyncQueue() throws Exception {
+    Context context = ApplicationProvider.getApplicationContext();
+    AsyncQueue asyncQueue = new AsyncQueue();
+    DatabaseId databaseId = DatabaseId.forProject("project");
+    DatabaseInfo databaseInfo = new DatabaseInfo(databaseId, "key", "host", true, 512 * 1024);
+
+    GrpcCallProvider grpcCallProvider =
+        new GrpcCallProvider(
+            asyncQueue, context, databaseInfo, Mockito.mock(CallCredentials.class));
+
+    Field taskField = GrpcCallProvider.class.getDeclaredField("channelTask");
+    taskField.setAccessible(true);
+    com.google.android.gms.tasks.Task<?> task =
+        (com.google.android.gms.tasks.Task<?>) taskField.get(grpcCallProvider);
+
+    Thread thread =
+        new Thread(
+            () -> {
+              try {
+                com.google.android.gms.tasks.Tasks.await(task);
+              } catch (Exception ignored) {
+              }
+            });
+    thread.start();
+    thread.join();
+
+    Field optionsField = GrpcCallProvider.class.getDeclaredField("callOptions");
+    optionsField.setAccessible(true);
+    CallOptions callOptions = (CallOptions) optionsField.get(grpcCallProvider);
+    Executor grpcExecutor = callOptions.getExecutor();
+
+    CountDownLatch drainExecuted = new CountDownLatch(1);
+    Runnable failingRunnable =
+        () -> {
+          try {
+            throw new IllegalStateException("unrelated internal state error");
+          } finally {
+            drainExecuted.countDown();
+          }
+        };
+
+    grpcExecutor.execute(failingRunnable);
+    assertTrue(drainExecuted.await(5, TimeUnit.SECONDS));
+
+    org.robolectric.shadows.ShadowLooper shadowLooper =
+        org.robolectric.Shadows.shadowOf(android.os.Looper.getMainLooper());
+    // Await the panic runnable posted to the Main Looper
+    long deadline = System.currentTimeMillis() + 5000;
+    while (shadowLooper.isIdle() && System.currentTimeMillis() < deadline) {
+      Thread.sleep(10);
+    }
+
+    // Non-cancellation exceptions must rethrow, reaching afterExecute() and tripping
+    // AsyncQueue.panic().
+    assertThrows(
+        RuntimeException.class, () -> org.robolectric.shadows.ShadowLooper.idleMainLooper());
+  }
 }