Handle IndexedDB errors in unsubscribe callback

packages/firestore/src/core/event_manager.ts

@@ -102,21 +102,19 @@ export class EventManager implements SyncEngineListener {
 
   async unlisten(listener: QueryListener): Promise<void> {
     const query = listener.query;
-    let lastListen = false;
 
     const queryInfo = this.queries.get(query);
     if (queryInfo) {
       const i = queryInfo.listeners.indexOf(listener);
       if (i >= 0) {
-        queryInfo.listeners.splice(i, 1);
-        lastListen = queryInfo.listeners.length === 0;
+        if (queryInfo.listeners.length > 1) {
+          queryInfo.listeners.splice(i, 1);
+        } else {
+          await this.syncEngine.unlisten(query);
+          this.queries.delete(query);
+        }
       }
     }
-
-    if (lastListen) {
-      this.queries.delete(query);
-      return this.syncEngine.unlisten(query);
-    }
   }
 
   onWatchChange(viewSnaps: ViewSnapshot[]): void {

packages/firestore/src/core/firestore_client.ts

@@ -404,7 +404,7 @@ export class FirestoreClient {
     if (this.clientTerminated) {
       return;
     }
-    this.asyncQueue.enqueueAndForget(() => {
+    this.asyncQueue.enqueueRetryable(() => {
       return this.eventMgr.unlisten(listener);
     });
   }

packages/firestore/test/unit/specs/recovery_spec.test.ts

@@ -746,4 +746,58 @@ describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
         });
     }
   );
+
+  specTest('Does not raise events when unlisten fails', [], () => {
+    const query = Query.atPath(path('collection'));
+    const doc1 = doc('collection/key1', 1, { foo: 'a' });
+    const doc2 = doc('collection/key2', 2, { foo: 'b' });
+    return (
+      spec()
+        .userListens(query)
+        .watchAcksFull(query, 1000, doc1)
+        .expectEvents(query, {
+          added: [doc1]
+        })
+        .failDatabaseTransactions('Release target')
+        .userUnlistens(query)
+        // Target remains active in RemoteStore, but events are suppressed
+        .expectActiveTargets({ query })
+        .watchSends({ affects: [query] }, doc2)
+        .recoverDatabase()
+        .runTimer(TimerId.AsyncQueueRetry)
+        .expectActiveTargets()
+    );
+  });
+
+  specTest('Can re-listen to query when unlisten fails', [], () => {
+    const query = Query.atPath(path('collection'));
+    const doc1 = doc('collection/key1', 1, { foo: 'a' });
+    const doc2 = doc('collection/key2', 2, { foo: 'b' });
+    return (
+      spec()
+        .userListens(query)
+        .watchAcksFull(query, 1000, doc1)
+        .expectEvents(query, {
+          added: [doc1]
+        })
+        .failDatabaseTransactions('Release target')
+        .userUnlistens(query)
+        // Target remains active since 'Release target' failed
+        .expectActiveTargets({ query })
+        .userListens(query)
+        .expectEvents(query, {
+          added: [doc1]
+        })
+        .recoverDatabase()
+        .runTimer(TimerId.AsyncQueueRetry)
+        // Target remains active as it has now been listened to twice and
+        // unsubscribed from only once
+        .watchSends({ affects: [query] }, doc2)
+        .watchSnapshots(2000)
+        .expectEvents(query, {
+          added: [doc2]
+        })
+        .userUnlistens(query)
+    );
+  });
 });

packages/firestore/test/unit/specs/spec_builder.ts

@@ -293,7 +293,11 @@ export class SpecBuilder {
     const targetId = this.queryMapping.get(target)!;
     this.removeQueryFromActiveTargets(query, targetId);
 
-    if (this.config.useGarbageCollection && !this.activeTargets[targetId]) {
+    if (
+      this.config.useGarbageCollection &&
+      !this.injectFailures &&
+      !this.activeTargets[targetId]
+    ) {
       this.queryMapping.delete(target);
       this.queryIdGenerator.purge(target);
     }

packages/firestore/test/unit/specs/spec_test_runner.ts

@@ -399,8 +399,29 @@ abstract class TestRunner {
     const query = parseQuery(querySpec);
     const eventEmitter = this.queryListeners.get(query);
     debugAssert(!!eventEmitter, 'There must be a query to unlisten too!');
-    this.queryListeners.delete(query);
-    await this.queue.enqueue(() => this.eventManager.unlisten(eventEmitter!));
+
+    const deferred = new Deferred<void>();
+    await this.queue.enqueueRetryable(async () => {
+      try {
+        await this.eventManager.unlisten(eventEmitter!);
+        const currentEventEmitter = this.queryListeners.get(query);
+        // Before removing the listener, verify that the query hasn't been
+        // relistened to in between retry events. We only remove the listener
+        // if the current event emitter still matches the event emitter at the
+        // time of the unlisten.
+        if (currentEventEmitter === eventEmitter) {
+          this.queryListeners.delete(query);
+        }
+      } catch (e) {
+        expect(this.persistence.injectFailures).contains('Release target');
+        throw e;
+      } finally {
+        // Resolve the deferred Promise even if the operation failed. This allows
+        // the spec tests to manually retry the failed user change.
+        deferred.resolve();
+      }
+    });
+    return deferred.promise;
   }
 
   private doSet(setSpec: SpecUserSet): Promise<void> {