Abort snapshot listeners on terminate

firebase-firestore/CHANGELOG.md

@@ -1,5 +1,5 @@
 # Unreleased
-
+* [changed] Fail snapshot listeners when Firestore terminates. [#6136](//github.com/firebase/firebase-android-sdk/pull/6136)
 
 # 25.1.0
 * [feature] Add support for the VectorValue type. [#6154](//github.com/firebase/firebase-android-sdk/pull/6154)

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/FirestoreTest.java

@@ -1263,6 +1263,8 @@ public void testCanStopListeningAfterTerminate() {
 
     waitFor(instance.terminate());
 
+    assertEquals(eventAccumulator.awaitError().getCode(), Code.ABORTED);
+
     // This should proceed without error.
     registration.remove();
     // Multiple calls should proceed as an effectively no-op.

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/ListenerRegistrationTest.java

@@ -165,12 +165,13 @@ private void activityScopedListenerStopsListeningWhenActivityStops(Activity acti
     DocumentReference documentReference = collectionReference.document();
 
     Semaphore events = new Semaphore(0);
-    collectionReference.addSnapshotListener(
-        activity,
-        (value, error) -> {
-          assertNull(error);
-          events.release();
-        });
+    ListenerRegistration listener =
+        collectionReference.addSnapshotListener(
+            activity,
+            (value, error) -> {
+              assertNull(error);
+              events.release();
+            });
 
     // Initial events
     waitFor(events, 1);
@@ -188,6 +189,8 @@ private void activityScopedListenerStopsListeningWhenActivityStops(Activity acti
     // No listeners, therefore, there should be no events.
     waitFor(documentReference.set(map("foo", "new-bar")));
     assertEquals(0, events.availablePermits());
+
+    listener.remove();
   }
 
   /** @param activity Must be a TestActivity or a TestFragmentActivity */

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/QueryTest.java

@@ -147,7 +147,7 @@ public void testListenUnlistenRelistenSequenceOfMirrorQueries() {
 
     // Unlisten then re-listen limit query.
     limitRegistration.remove();
-    limit.addSnapshotListener(limitAccumulator.listener());
+    limitRegistration = limit.addSnapshotListener(limitAccumulator.listener());
 
     // Verify `limit` query still works.
     data = querySnapshotToValues(limitAccumulator.await());
@@ -165,13 +165,16 @@ public void testListenUnlistenRelistenSequenceOfMirrorQueries() {
     // Unlisten to limitToLast, update a doc, then relisten to limitToLast
     limitToLastRegistration.remove();
     waitFor(collection.document("a").update(map("k", "a", "sort", -2)));
-    limitToLast.addSnapshotListener(limitToLastAccumulator.listener());
+    limitToLastRegistration = limitToLast.addSnapshotListener(limitToLastAccumulator.listener());
 
     // Verify both query get expected result.
     data = querySnapshotToValues(limitAccumulator.await());
     assertEquals(asList(map("k", "a", "sort", -2L), map("k", "e", "sort", -1L)), data);
     data = querySnapshotToValues(limitToLastAccumulator.await());
     assertEquals(asList(map("k", "e", "sort", -1L), map("k", "a", "sort", -2L)), data);
+
+    limitRegistration.remove();
+    limitToLastRegistration.remove();
   }
 
   @Test
@@ -507,19 +510,22 @@ public void watchSurvivesNetworkDisconnect() {
 
     Semaphore receivedDocument = new Semaphore(0);
 
-    collectionReference.addSnapshotListener(
-        MetadataChanges.INCLUDE,
-        (snapshot, error) -> {
-          if (!snapshot.isEmpty() && !snapshot.getMetadata().isFromCache()) {
-            receivedDocument.release();
-          }
-        });
+    ListenerRegistration listener =
+        collectionReference.addSnapshotListener(
+            MetadataChanges.INCLUDE,
+            (snapshot, error) -> {
+              if (!snapshot.isEmpty() && !snapshot.getMetadata().isFromCache()) {
+                receivedDocument.release();
+              }
+            });
 
     waitFor(firestore.disableNetwork());
     collectionReference.add(map("foo", FieldValue.serverTimestamp()));
     waitFor(firestore.enableNetwork());
 
     waitFor(receivedDocument);
+
+    listener.remove();
   }
 
   @Test

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/SourceTest.java

@@ -289,15 +289,17 @@ public void getDocumentWhileOfflineWithDifferentGetOptions() {
     // Create an initial listener for this query (to attempt to disrupt the gets below) and wait for
     // the listener to deliver its initial snapshot before continuing.
     TaskCompletionSource<Void> source = new TaskCompletionSource<>();
-    docRef.addSnapshotListener(
-        (docSnap, error) -> {
-          if (error != null) {
-            source.setException(error);
-          } else {
-            source.setResult(null);
-          }
-        });
+    ListenerRegistration listener =
+        docRef.addSnapshotListener(
+            (docSnap, error) -> {
+              if (error != null) {
+                source.setException(error);
+              } else {
+                source.setResult(null);
+              }
+            });
     waitFor(source.getTask());
+    listener.remove();
 
     Task<DocumentSnapshot> docTask = docRef.get(Source.CACHE);
     waitFor(docTask);
@@ -339,15 +341,17 @@ public void getCollectionWhileOfflineWithDifferentGetOptions() {
     // Create an initial listener for this query (to attempt to disrupt the gets below) and wait for
     // the listener to deliver its initial snapshot before continuing.
     TaskCompletionSource<Void> source = new TaskCompletionSource<>();
-    colRef.addSnapshotListener(
-        (qrySnap, error) -> {
-          if (error != null) {
-            source.setException(error);
-          } else {
-            source.setResult(null);
-          }
-        });
+    ListenerRegistration listener =
+        colRef.addSnapshotListener(
+            (qrySnap, error) -> {
+              if (error != null) {
+                source.setException(error);
+              } else {
+                source.setResult(null);
+              }
+            });
     waitFor(source.getTask());
+    listener.remove();
 
     Task<QuerySnapshot> qrySnapTask = colRef.get(Source.CACHE);
     waitFor(qrySnapTask);

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/testutil/EventAccumulator.java

@@ -19,6 +19,7 @@
 
 import com.google.firebase.firestore.DocumentSnapshot;
 import com.google.firebase.firestore.EventListener;
+import com.google.firebase.firestore.FirebaseFirestoreException;
 import com.google.firebase.firestore.QuerySnapshot;
 import com.google.firebase.firestore.util.Logger;
 import java.util.ArrayList;
@@ -30,28 +31,45 @@
 public class EventAccumulator<T> {
   private static final int MAX_EVENTS = 10;
 
-  private final BlockingQueue<T> events;
+  private final BlockingQueue<Object> events;
   private boolean rejectAdditionalEvents;
 
   public EventAccumulator() {
-    events = new ArrayBlockingQueue<T>(MAX_EVENTS);
+    events = new ArrayBlockingQueue<>(MAX_EVENTS);
   }
 
   public EventListener<T> listener() {
     return (value, error) -> {
-      hardAssert(error == null, "Unexpected error: %s", error);
       hardAssert(
           !rejectAdditionalEvents, "Received event after `assertNoAdditionalEvents()` was called");
-      Logger.debug("EventAccumulator", "Received new event: " + value);
-      events.offer(value);
+      if (error == null) {
+        Logger.debug("EventAccumulator", "Received new event: " + value);
+        events.add(value);
+      } else {
+        Logger.debug("EventAccumulator", "Received error: " + error);
+        events.add(error);
+      }
     };
   }
 
+  public FirebaseFirestoreException awaitError() {
+    try {
+      return (FirebaseFirestoreException) events.take();
+    } catch (Exception e) {
+      Logger.debug("EventAccumulator", e.toString());
+      throw fail("Failed to receive error");
+    }
+  }
+
   public List<T> await(int numEvents) {
     try {
       List<T> result = new ArrayList<>(numEvents);
       for (int i = 0; i < numEvents; ++i) {
-        result.add(events.take());
+        Object event = events.take();
+        if (event instanceof FirebaseFirestoreException) {
+          fail("Unexpected error: %s", event);
+        }
+        result.add((T) event);
       }
       return result;
     } catch (InterruptedException e) {

firebase-firestore/src/main/java/com/google/firebase/firestore/FirebaseFirestore.java

@@ -581,8 +581,8 @@ public Task<Void> runBatch(@NonNull WriteBatch.Function batchFunction) {
    * <p>To restart after termination, simply create a new instance of {@code FirebaseFirestore} with
    * {@link #getInstance()} or {@link #getInstance(FirebaseApp)}.
    *
-   * <p>{@code terminate()} does not cancel any pending writes and any tasks that are awaiting a
-   * response from the server will not be resolved. The next time you start this instance, it will
+   * <p>{@code terminate()} does not cancel any pending writes but any write tasks that are awaiting
+   * a response from the server will not be resolved. The next time you start this instance, it will
    * resume attempting to send these writes to the server.
    *
    * <p>Note: Under normal circumstances, calling {@code terminate()} is not required. This method

firebase-firestore/src/main/java/com/google/firebase/firestore/core/EventManager.java

@@ -17,6 +17,7 @@
 import static com.google.firebase.firestore.util.Assert.hardAssert;
 
 import com.google.firebase.firestore.EventListener;
+import com.google.firebase.firestore.FirebaseFirestoreException;
 import com.google.firebase.firestore.ListenSource;
 import com.google.firebase.firestore.core.SyncEngine.SyncEngineCallback;
 import com.google.firebase.firestore.util.Util;
@@ -265,4 +266,14 @@ public void handleOnlineStateChange(OnlineState onlineState) {
       raiseSnapshotsInSyncEvent();
     }
   }
+
+  public void abortAllTargets() {
+    FirebaseFirestoreException error = Util.exceptionFromStatus(Status.ABORTED);
+    for (QueryListenersInfo info : queries.values()) {
+      for (QueryListener listener : info.listeners) {
+        listener.onError(error);
+      }
+    }
+    queries.clear();
+  }
 }

firebase-firestore/src/main/java/com/google/firebase/firestore/core/FirestoreClient.java

@@ -148,6 +148,7 @@ public Task<Void> enableNetwork() {
   public Task<Void> terminate() {
     authProvider.removeChangeListener();
     appCheckProvider.removeChangeListener();
+    asyncQueue.enqueueAndForgetEvenAfterShutdown(() -> eventManager.abortAllTargets());
     return asyncQueue.enqueueAndInitiateShutdown(
         () -> {
           remoteStore.shutdown();