Fix errors

Author: tom-andersen

packages/firestore/src/core/firestore_client.ts

@@ -656,6 +656,9 @@ function readDocumentViaSnapshotListener(
 ): Promise<void> {
   const wrappedObserver = new AsyncObserver({
     next: (snap: ViewSnapshot) => {
+      // No further messages should be processed.
+      wrappedObserver.mute();
+
       // Remove query first before passing event to user to avoid
       // user actions affecting the now stale query.
       asyncQueue.enqueueAndForget(() =>
@@ -754,6 +757,9 @@ function executeQueryViaSnapshotListener(
 ): Promise<void> {
   const wrappedObserver = new AsyncObserver<ViewSnapshot>({
     next: snapshot => {
+      // No further messages should be processed.
+      wrappedObserver.mute();
+
       // Remove query first before passing event to user to avoid
       // user actions affecting the now stale query.
       asyncQueue.enqueueAndForget(() =>

packages/firestore/test/integration/api/bundle.test.ts

@@ -202,7 +202,7 @@ apiDescribe('Bundles', persistence => {
       await setDoc(doc(db, 'coll-1/b'), { k: 'b', bar: 0 });
 
       const accumulator = new EventsAccumulator<QuerySnapshot>();
-      onSnapshot(collection(db, 'coll-1'), accumulator.storeEvent);
+      const unsubscribe = onSnapshot(collection(db, 'coll-1'), accumulator.storeEvent);
       await accumulator.awaitEvent();
 
       const progress = await loadBundle(
@@ -216,6 +216,7 @@ apiDescribe('Bundles', persistence => {
       // generated as a result. The case where a bundle has newer doc than
       // cache can only be tested in spec tests.
       await accumulator.assertNoAdditionalEvents();
+      unsubscribe();
 
       let snap = await getDocs((await namedQuery(db, 'limit'))!);
       expect(toDataArray(snap)).to.deep.equal([{ k: 'b', bar: 0 }]);