IndexedDB Recovery for Limbo documents (#3039)

Author: schmidt-sebastian

packages/firestore/src/core/sync_engine.ts

@@ -483,14 +483,6 @@ export class SyncEngine implements RemoteSyncer {
     const limboResolution = this.activeLimboResolutionsByTarget.get(targetId);
     const limboKey = limboResolution && limboResolution.key;
     if (limboKey) {
-      // Since this query failed, we won't want to manually unlisten to it.
-      // So go ahead and remove it from bookkeeping.
-      this.activeLimboTargetsByKey = this.activeLimboTargetsByKey.remove(
-        limboKey
-      );
-      this.activeLimboResolutionsByTarget.delete(targetId);
-      this.pumpEnqueuedLimboResolutions();
-
       // TODO(klimt): We really only should do the following on permission
       // denied errors, but we don't have the cause code here.
 
@@ -513,7 +505,19 @@ export class SyncEngine implements RemoteSyncer {
         documentUpdates,
         resolvedLimboDocuments
       );
-      return this.applyRemoteEvent(event);
+
+      await this.applyRemoteEvent(event);
+
+      // Since this query failed, we won't want to manually unlisten to it.
+      // We only remove it from bookkeeping after we successfully applied the
+      // RemoteEvent. If `applyRemoteEvent()` throws, we want to re-listen to
+      // this query when the RemoteStore restarts the Watch stream, which should
+      // re-trigger the target failure.
+      this.activeLimboTargetsByKey = this.activeLimboTargetsByKey.remove(
+        limboKey
+      );
+      this.activeLimboResolutionsByTarget.delete(targetId);
+      this.pumpEnqueuedLimboResolutions();
     } else {
       await this.localStore
         .releaseTarget(targetId, /* keepPersistedTargetData */ false)

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

@@ -20,7 +20,7 @@ import { client, spec } from './spec_builder';
 import { TimerId } from '../../../src/util/async_queue';
 import { Query } from '../../../src/core/query';
 import { Code } from '../../../src/util/error';
-import { doc, path } from '../../util/helpers';
+import { doc, filter, path } from '../../util/helpers';
 import { RpcError } from './spec_rpc_error';
 
 describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
@@ -309,4 +309,101 @@ describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
       );
     }
   );
+
+  specTest(
+    'Recovers when Limbo acknowledgement cannot be persisted',
+    ['durable-persistence'],
+    () => {
+      const fullQuery = Query.atPath(path('collection'));
+      const filteredQuery = Query.atPath(path('collection')).addFilter(
+        filter('included', '==', true)
+      );
+      const doc1a = doc('collection/key1', 1, { included: true });
+      const doc1b = doc('collection/key1', 1500, { included: false });
+      const limboQuery = Query.atPath(doc1a.key.path);
+      return spec()
+        .userListens(fullQuery)
+        .watchAcksFull(fullQuery, 1000, doc1a)
+        .expectEvents(fullQuery, {
+          added: [doc1a]
+        })
+        .userUnlistens(fullQuery)
+        .userListens(filteredQuery)
+        .expectEvents(filteredQuery, {
+          added: [doc1a],
+          fromCache: true
+        })
+        .watchAcksFull(filteredQuery, 2000)
+        .expectLimboDocs(doc1a.key)
+        .failDatabaseTransactions({ 'Get last remote snapshot version': true })
+        .watchAcksFull(limboQuery, 3000, doc1b)
+        .expectActiveTargets()
+        .recoverDatabase()
+        .runTimer(TimerId.AsyncQueueRetry)
+        .expectActiveTargets(
+          {
+            query: filteredQuery,
+            resumeToken: 'resume-token-2000'
+          },
+          { query: limboQuery }
+        )
+        .watchAcksFull(filteredQuery, 4000)
+        .watchAcksFull(limboQuery, 4000, doc1b)
+        .expectLimboDocs()
+        .expectEvents(filteredQuery, {
+          removed: [doc1a]
+        });
+    }
+  );
+
+  specTest(
+    'Recovers when Limbo rejection cannot be persisted',
+    ['durable-persistence'],
+    () => {
+      const fullQuery = Query.atPath(path('collection'));
+      const filteredQuery = Query.atPath(path('collection')).addFilter(
+        filter('included', '==', true)
+      );
+      const doc1 = doc('collection/key1', 1, { included: true });
+      const limboQuery = Query.atPath(doc1.key.path);
+      return spec()
+        .userListens(fullQuery)
+        .watchAcksFull(fullQuery, 1000, doc1)
+        .expectEvents(fullQuery, {
+          added: [doc1]
+        })
+        .userUnlistens(fullQuery)
+        .userListens(filteredQuery)
+        .expectEvents(filteredQuery, {
+          added: [doc1],
+          fromCache: true
+        })
+        .watchAcksFull(filteredQuery, 2000)
+        .expectLimboDocs(doc1.key)
+        .failDatabaseTransactions({
+          'Apply remote event': true,
+          'Get last remote snapshot version': true
+        })
+        .watchRemoves(
+          limboQuery,
+          new RpcError(Code.PERMISSION_DENIED, 'Test error')
+        )
+        .expectActiveTargets()
+        .recoverDatabase()
+        .runTimer(TimerId.AsyncQueueRetry)
+        .expectActiveTargets(
+          { query: filteredQuery, resumeToken: 'resume-token-2000' },
+          { query: limboQuery }
+        )
+        .watchAcksFull(filteredQuery, 3000)
+        .watchRemoves(
+          limboQuery,
+          new RpcError(Code.PERMISSION_DENIED, 'Test error')
+        )
+        .expectLimboDocs()
+        .expectEvents(filteredQuery, {
+          removed: [doc1]
+        });
+    }
+  );
 });

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

@@ -206,6 +206,9 @@ export class MockConnection implements Connection {
   /** A Deferred that is resolved once watch opens. */
   watchOpen = new Deferred<void>();
 
+  /** Whether the Watch stream is open. */
+  isWatchOpen = false;
+
   invokeRPC<Req>(rpcName: string, request: Req): never {
     throw new Error('Not implemented!');
   }
@@ -263,6 +266,7 @@ export class MockConnection implements Connection {
     this.watchOpen = new Deferred<void>();
     this.watchStream!.callOnClose(err);
     this.watchStream = null;
+    this.isWatchOpen = false;
   }
 
   openStream<Req, Resp>(
@@ -351,6 +355,7 @@ export class MockConnection implements Connection {
       this.queue.enqueueAndForget(async () => {
         if (this.watchStream === watchStream) {
           watchStream.callOnOpen();
+          this.isWatchOpen = true;
           this.watchOpen.resolve();
         }
       });

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

@@ -847,18 +847,22 @@ abstract class TestRunner {
 
   private validateActiveLimboDocs(): void {
     let actualLimboDocs = this.syncEngine.activeLimboDocumentResolutions();
-    // Validate that each active limbo doc has an expected active target
-    actualLimboDocs.forEach((key, targetId) => {
-      const targetIds = new Array(this.expectedActiveTargets.keys()).map(
-        n => '' + n
-      );
-      expect(this.expectedActiveTargets.has(targetId)).to.equal(
-        true,
-        `Found limbo doc ${key.toString()}, but its target ID ${targetId} ` +
-          `was not in the set of expected active target IDs ` +
-          `(${targetIds.join(', ')})`
-      );
-    });
+
+    if (this.connection.isWatchOpen) {
+      // Validate that each active limbo doc has an expected active target
+      actualLimboDocs.forEach((key, targetId) => {
+        const targetIds = new Array(this.expectedActiveTargets.keys()).map(
+          n => '' + n
+        );
+        expect(this.expectedActiveTargets.has(targetId)).to.equal(
+          true,
+          `Found limbo doc ${key.toString()}, but its target ID ${targetId} ` +
+            `was not in the set of expected active target IDs ` +
+            `(${targetIds.join(', ')})`
+        );
+      });
+    }
+
     for (const expectedLimboDoc of this.expectedActiveLimboDocs) {
       expect(actualLimboDocs.get(expectedLimboDoc)).to.not.equal(
         null,