From 4476075c189be1de0c958117e761e31a73744ecb Mon Sep 17 00:00:00 2001 From: dlarocque Date: Wed, 16 Sep 2026 13:17:07 -0400 Subject: [PATCH 1/2] fix(firestore): guard against gRPC `call was cancelled` exceptions --- .../firestore/remote/AbstractStream.java | 16 ++- .../firestore/remote/GrpcCallProvider.java | 31 ++++- .../firestore/remote/AbstractStreamTest.java | 112 ++++++++++++++++ .../remote/GrpcCallProviderTest.java | 121 ++++++++++++++++++ 4 files changed, 275 insertions(+), 5 deletions(-) create mode 100644 firebase-firestore/src/test/java/com/google/firebase/firestore/remote/AbstractStreamTest.java diff --git a/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java b/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java index 963cb4ca532..09c12a8ab66 100644 --- a/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java +++ b/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. */ diff --git a/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java b/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java index f2eabe4a251..ebf6cf9f822 100644 --- a/firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java +++ b/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; diff --git a/firebase-firestore/src/test/java/com/google/firebase/firestore/remote/AbstractStreamTest.java b/firebase-firestore/src/test/java/com/google/firebase/firestore/remote/AbstractStreamTest.java new file mode 100644 index 00000000000..b99ad566e83 --- /dev/null +++ b/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 { + 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 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 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"))); + } +} diff --git a/firebase-firestore/src/test/java/com/google/firebase/firestore/remote/GrpcCallProviderTest.java b/firebase-firestore/src/test/java/com/google/firebase/firestore/remote/GrpcCallProviderTest.java index 90804864862..e18d695a0f0 100644 --- a/firebase-firestore/src/test/java/com/google/firebase/firestore/remote/GrpcCallProviderTest.java +++ b/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()); + } } From c563cdc73af7347636ffb9a058aca9aeb387fdfe Mon Sep 17 00:00:00 2001 From: dlarocque Date: Wed, 16 Sep 2026 15:25:18 -0400 Subject: [PATCH 2/2] changelog and version bump --- firebase-firestore/CHANGELOG.md | 2 ++ firebase-firestore/gradle.properties | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/firebase-firestore/CHANGELOG.md b/firebase-firestore/CHANGELOG.md index 7d5358af449..36505be4bf4 100644 --- a/firebase-firestore/CHANGELOG.md +++ b/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) diff --git a/firebase-firestore/gradle.properties b/firebase-firestore/gradle.properties index a65ca8d3183..07fd2f1e314 100644 --- a/firebase-firestore/gradle.properties +++ b/firebase-firestore/gradle.properties @@ -1,2 +1,2 @@ -version=26.6.1 +version=26.6.2 latestReleasedVersion=26.6.0