Take WriteStream offline when IndexedDB is unavailable (#2995)

schmidt-sebastian · web-flow · commit 5c41eab666fe · 2020-06-02T14:59:05.000-07:00
diff --git a/packages/firestore/CHANGELOG.md b/packages/firestore/CHANGELOG.md
@@ -1,11 +1,21 @@
 # Unreleased
+- [changed] All known failure cases for Indexed-related crashes have now been
+  addressed. Instead of crashing the client, IndexedDB failures will result
+  in rejected operations (e.g. rejected Writes or errored Query listeners). 
+  If these rejections surface in your app, you can retry these operations 
+  when IndexedDB access is restored. 
+  IndexedDB failures that occur due to background work are automatically 
+  retried. 
+  
+  If you continue to see Indexed-related crashes, we appreciate feedback 
+  (https://github.com/firebase/firebase-js-sdk/issues/2755).
+
+# Released
 - [fixed] Fixed an issue that could cause Firestore to temporarily go 
   offline when a Window visibility event occurred.
 - [feature] Added support for calling  `FirebaseFiresore.settings` with 
   `{ ignoreUndefinedProperties: true }`. When set, Firestore ignores 
   undefined properties inside objects rather than rejecting the API call.
-
-# Released
 - [fixed] Fixed a regression introduced in v7.14.2 that incorrectly applied
   a `FieldValue.increment` in combination with `set({...}, {merge: true})`.
 - [fixed] Firestore now rejects `onSnapshot()` listeners if they cannot be
diff --git a/packages/firestore/src/core/sync_engine.ts b/packages/firestore/src/core/sync_engine.ts
@@ -525,18 +525,18 @@ export class SyncEngine implements RemoteSyncer {
 
     const batchId = mutationBatchResult.batch.batchId;
 
-    // The local store may or may not be able to apply the write result and
-    // raise events immediately (depending on whether the watcher is caught
-    // up), so we raise user callbacks first so that they consistently happen
-    // before listen events.
-    this.processUserCallback(batchId, /*error=*/ null);
-
-    this.triggerPendingWritesCallbacks(batchId);
-
     try {
       const changes = await this.localStore.acknowledgeBatch(
         mutationBatchResult
       );
+
+      // The local store may or may not be able to apply the write result and
+      // raise events immediately (depending on whether the watcher is caught
+      // up), so we raise user callbacks first so that they consistently happen
+      // before listen events.
+      this.processUserCallback(batchId, /*error=*/ null);
+      this.triggerPendingWritesCallbacks(batchId);
+
       this.sharedClientState.updateMutationState(batchId, 'acknowledged');
       await this.emitNewSnapsAndNotifyLocalStore(changes);
     } catch (error) {
@@ -550,16 +550,16 @@ export class SyncEngine implements RemoteSyncer {
   ): Promise<void> {
     this.assertSubscribed('rejectFailedWrite()');
 
-    // The local store may or may not be able to apply the write result and
-    // raise events immediately (depending on whether the watcher is caught up),
-    // so we raise user callbacks first so that they consistently happen before
-    // listen events.
-    this.processUserCallback(batchId, error);
-
-    this.triggerPendingWritesCallbacks(batchId);
-
     try {
       const changes = await this.localStore.rejectBatch(batchId);
+
+      // The local store may or may not be able to apply the write result and
+      // raise events immediately (depending on whether the watcher is caught up),
+      // so we raise user callbacks first so that they consistently happen before
+      // listen events.
+      this.processUserCallback(batchId, error);
+      this.triggerPendingWritesCallbacks(batchId);
+
       this.sharedClientState.updateMutationState(batchId, 'rejected', error);
       await this.emitNewSnapsAndNotifyLocalStore(changes);
     } catch (error) {
diff --git a/packages/firestore/src/remote/remote_store.ts b/packages/firestore/src/remote/remote_store.ts
@@ -443,10 +443,17 @@ export class RemoteStore implements TargetMetadataProvider {
 
   /**
    * Recovery logic for IndexedDB errors that takes the network offline until
-   * IndexedDb probing succeeds. Retries are scheduled with backoff using
-   * `enqueueRetryable()`.
+   * `op` succeeds. Retries are scheduled with backoff using
+   * `enqueueRetryable()`. If `op()` is not provided, IndexedDB access is
+   * validated via a generic operation.
+   *
+   * The returned Promise is resolved once the network is disabled and before
+   * any retry attempt.
    */
-  private async disableNetworkUntilRecovery(e: FirestoreError): Promise<void> {
+  private async disableNetworkUntilRecovery(
+    e: FirestoreError,
+    op?: () => Promise<unknown>
+  ): Promise<void> {
     if (isIndexedDbTransactionError(e)) {
       debugAssert(
         !this.indexedDbFailed,
@@ -458,13 +465,17 @@ export class RemoteStore implements TargetMetadataProvider {
       await this.disableNetworkInternal();
       this.onlineStateTracker.set(OnlineState.Offline);
 
+      if (!op) {
+        // Use a simple read operation to determine if IndexedDB recovered.
+        // Ideally, we would expose a health check directly on SimpleDb, but
+        // RemoteStore only has access to persistence through LocalStore.
+        op = () => this.localStore.getLastRemoteSnapshotVersion();
+      }
+
       // Probe IndexedDB periodically and re-enable network
       this.asyncQueue.enqueueRetryable(async () => {
         logDebug(LOG_TAG, 'Retrying IndexedDB access');
-        // Issue a simple read operation to determine if IndexedDB recovered.
-        // Ideally, we would expose a health check directly on SimpleDb, but
-        // RemoteStore only has access to persistence through LocalStore.
-        await this.localStore.getLastRemoteSnapshotVersion();
+        await op!();
         this.indexedDbFailed = false;
         await this.enableNetworkInternal();
       });
@@ -473,6 +484,14 @@ export class RemoteStore implements TargetMetadataProvider {
     }
   }
 
+  /**
+   * Executes `op`. If `op` fails, takes the network offline until `op`
+   * succeeds. Returns after the first attempt.
+   */
+  private executeWithRecovery(op: () => Promise<void>): Promise<void> {
+    return op().catch(e => this.disableNetworkUntilRecovery(e, op));
+  }
+
   /**
    * Takes a batch of changes from the Datastore, repackages them as a
    * RemoteEvent, and passes that on to the listener, which is typically the
@@ -567,22 +586,28 @@ export class RemoteStore implements TargetMetadataProvider {
    * Starts the write stream if necessary.
    */
   async fillWritePipeline(): Promise<void> {
-    if (this.canAddToWritePipeline()) {
-      const lastBatchIdRetrieved =
-        this.writePipeline.length > 0
-          ? this.writePipeline[this.writePipeline.length - 1].batchId
-          : BATCHID_UNKNOWN;
-      const batch = await this.localStore.nextMutationBatch(
-        lastBatchIdRetrieved
-      );
+    let lastBatchIdRetrieved =
+      this.writePipeline.length > 0
+        ? this.writePipeline[this.writePipeline.length - 1].batchId
+        : BATCHID_UNKNOWN;
 
-      if (batch === null) {
-        if (this.writePipeline.length === 0) {
-          this.writeStream.markIdle();
+    while (this.canAddToWritePipeline()) {
+      try {
+        const batch = await this.localStore.nextMutationBatch(
+          lastBatchIdRetrieved
+        );
+
+        if (batch === null) {
+          if (this.writePipeline.length === 0) {
+            this.writeStream.markIdle();
+          }
+          break;
+        } else {
+          lastBatchIdRetrieved = batch.batchId;
+          this.addToWritePipeline(batch);
         }
-      } else {
-        this.addToWritePipeline(batch);
-        await this.fillWritePipeline();
+      } catch (e) {
+        await this.disableNetworkUntilRecovery(e);
       }
     }
 
@@ -649,7 +674,7 @@ export class RemoteStore implements TargetMetadataProvider {
     }
   }
 
-  private onMutationResult(
+  private async onMutationResult(
     commitVersion: SnapshotVersion,
     results: MutationResult[]
   ): Promise<void> {
@@ -661,11 +686,14 @@ export class RemoteStore implements TargetMetadataProvider {
     );
     const batch = this.writePipeline.shift()!;
     const success = MutationBatchResult.from(batch, commitVersion, results);
-    return this.syncEngine.applySuccessfulWrite(success).then(() => {
-      // It's possible that with the completion of this mutation another
-      // slot has freed up.
-      return this.fillWritePipeline();
-    });
+
+    await this.executeWithRecovery(() =>
+      this.syncEngine.applySuccessfulWrite(success)
+    );
+
+    // It's possible that with the completion of this mutation another
+    // slot has freed up.
+    await this.fillWritePipeline();
   }
 
   private async onWriteStreamClose(error?: FirestoreError): Promise<void> {
@@ -705,13 +733,13 @@ export class RemoteStore implements TargetMetadataProvider {
       // restart.
       this.writeStream.inhibitBackoff();
 
-      return this.syncEngine
-        .rejectFailedWrite(batch.batchId, error)
-        .then(() => {
-          // It's possible that with the completion of this mutation
-          // another slot has freed up.
-          return this.fillWritePipeline();
-        });
+      await this.executeWithRecovery(() =>
+        this.syncEngine.rejectFailedWrite(batch.batchId, error)
+      );
+
+      // It's possible that with the completion of this mutation
+      // another slot has freed up.
+      await this.fillWritePipeline();
     } else {
       // Transient error, just let the retry logic kick in.
     }
diff --git a/packages/firestore/test/unit/specs/recovery_spec.test.ts b/packages/firestore/test/unit/specs/recovery_spec.test.ts
@@ -336,6 +336,115 @@ describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
       .expectEvents(query, { metadata: [doc1, doc3] });
   });
 
+  specTest('Recovers when write acknowledgment cannot be persisted', [], () => {
+    return spec()
+      .userSets('collection/a', { v: 1 })
+      .userSets('collection/b', { v: 2 })
+      .userSets('collection/c', { v: 3 })
+      .writeAcks('collection/a', 1)
+      .failDatabaseTransactions('Acknowledge batch')
+      .writeAcks('collection/b', 2, { expectUserCallback: false })
+      .recoverDatabase()
+      .runTimer(TimerId.AsyncQueueRetry)
+      .expectUserCallbacks({ acknowledged: ['collection/b'] })
+      .writeAcks('collection/c', 1);
+  });
+
+  specTest('Recovers when write rejection cannot be persisted', [], () => {
+    return spec()
+      .userPatches('collection/a', { v: 1 })
+      .userPatches('collection/a', { v: 2 })
+      .userPatches('collection/c', { v: 3 })
+      .failWrite(
+        'collection/a',
+        new RpcError(Code.FAILED_PRECONDITION, 'Simulated test error')
+      )
+      .failDatabaseTransactions('Reject batch')
+      .failWrite(
+        'collection/b',
+        new RpcError(Code.FAILED_PRECONDITION, 'Simulated test error'),
+        { expectUserCallback: false }
+      )
+      .recoverDatabase()
+      .runTimer(TimerId.AsyncQueueRetry)
+      .expectUserCallbacks({ rejected: ['collection/a'] })
+      .failWrite(
+        'collection/c',
+        new RpcError(Code.FAILED_PRECONDITION, 'Simulated test error')
+      );
+  });
+
+  specTest(
+    'Recovers when write acknowledgment cannot be persisted (with restart)',
+    ['durable-persistence'],
+    () => {
+      // This test verifies the current behavior of the client, which is not
+      // ideal. Instead of resending the write to 'collection/b' (whose
+      // rejection failed with an IndexedDB failure), the client should drop the
+      // write.
+      return spec()
+        .userSets('collection/a', { v: 1 })
+        .userSets('collection/b', { v: 2 })
+        .userSets('collection/c', { v: 3 })
+        .writeAcks('collection/a', 1)
+        .failDatabaseTransactions('Acknowledge batch')
+        .writeAcks('collection/b', 2, {
+          expectUserCallback: false,
+          keepInQueue: true
+        })
+        .restart()
+        .expectNumOutstandingWrites(2)
+        .writeAcks('collection/b', 2, { expectUserCallback: false })
+        .writeAcks('collection/c', 3, { expectUserCallback: false });
+    }
+  );
+
+  specTest('Writes are pending until acknowledgement is persisted', [], () => {
+    const query = Query.atPath(path('collection'));
+    const doc1Local = doc(
+      'collection/a',
+      0,
+      { v: 1 },
+      { hasLocalMutations: true }
+    );
+    const doc1 = doc('collection/a', 1001, { v: 1 });
+    const doc2Local = doc(
+      'collection/b',
+      0,
+      { v: 2 },
+      { hasLocalMutations: true }
+    );
+    const doc2 = doc('collection/b', 1002, { v: 2 });
+    return (
+      spec()
+        .userListens(query)
+        .watchAcksFull(query, 1000)
+        .expectEvents(query, {})
+        .userSets('collection/a', { v: 1 })
+        .expectEvents(query, { added: [doc1Local], hasPendingWrites: true })
+        .userSets('collection/b', { v: 2 })
+        .expectEvents(query, { added: [doc2Local], hasPendingWrites: true })
+        .failDatabaseTransactions('Acknowledge batch')
+        .writeAcks('collection/a', 1, { expectUserCallback: false })
+        // The write ack cannot be persisted and the client goes offline, which
+        // clears all active targets, but doesn't raise a new snapshot since
+        // the document is still marked `hasPendingWrites`.
+        .expectEvents(query, { fromCache: true, hasPendingWrites: true })
+        .expectActiveTargets()
+        .recoverDatabase()
+        .runTimer(TimerId.AsyncQueueRetry)
+        // Client is back online
+        .expectActiveTargets({ query, resumeToken: 'resume-token-1000' })
+        .expectUserCallbacks({ acknowledged: ['collection/a'] })
+        .watchAcksFull(query, 1001, doc1)
+        .expectEvents(query, { metadata: [doc1], hasPendingWrites: true })
+        .writeAcks('collection/b', 2)
+        .watchSends({ affects: [query] }, doc2)
+        .watchSnapshots(1002)
+        .expectEvents(query, { metadata: [doc2] })
+    );
+  });
+
   specTest(
     'Surfaces local documents if notifyLocalViewChanges fails',
     [],
