Merge 50d4701 into 0d6f14e

schmidt-sebastian · web-flow · commit 379796b0d746 · 2020-05-28T14:34:02.000-07:00
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
@@ -199,7 +199,11 @@ export class RemoteStore implements TargetMetadataProvider {
 
   private async enableNetworkInternal(): Promise<void> {
     if (this.canUseNetwork()) {
-      this.writeStream.lastStreamToken = await this.localStore.getLastStreamToken();
+      try {
+        this.writeStream.lastStreamToken = await this.localStore.getLastStreamToken();
+      } catch (e) {
+        return this.disableNetworkUntilRecovery(e);
+      }
 
       if (this.shouldStartWatchStream()) {
         this.startWatchStream();
@@ -445,10 +449,14 @@ 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.
    */
-  private async disableNetworkUntilRecovery(e: FirestoreError): Promise<void> {
+  private async disableNetworkUntilRecovery(
+    e: FirestoreError,
+    op?: () => Promise<unknown>
+  ): Promise<void> {
     if (isIndexedDbTransactionError(e)) {
       debugAssert(
         !this.indexedDbFailed,
@@ -460,13 +468,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();
       });
@@ -569,25 +581,30 @@ export class RemoteStore implements TargetMetadataProvider {
    * Starts the write stream if necessary.
    */
   async fillWritePipeline(): Promise<void> {
-    if (this.canAddToWritePipeline()) {
+    while (this.canAddToWritePipeline()) {
       const lastBatchIdRetrieved =
         this.writePipeline.length > 0
           ? this.writePipeline[this.writePipeline.length - 1].batchId
           : BATCHID_UNKNOWN;
-      const batch = await this.localStore.nextMutationBatch(
-        lastBatchIdRetrieved
-      );
 
-      if (batch === null) {
-        if (this.writePipeline.length === 0) {
-          this.writeStream.markIdle();
+      try {
+        const batch = await this.localStore.nextMutationBatch(
+          lastBatchIdRetrieved
+        );
+        if (batch !== null) {
+          this.addToWritePipeline(batch);
+        } else {
+          break;
         }
-      } else {
-        this.addToWritePipeline(batch);
-        await this.fillWritePipeline();
+      } catch (e) {
+        await this.disableNetworkUntilRecovery(e);
       }
     }
 
+    if (this.writePipeline.length === 0) {
+      this.writeStream.markIdle();
+    }
+
     if (this.shouldStartWriteStream()) {
       this.startWriteStream();
     }
@@ -654,10 +671,15 @@ export class RemoteStore implements TargetMetadataProvider {
           this.writeStream.writeMutations(batch.mutations);
         }
       })
-      .catch(ignoreIfPrimaryLeaseLoss);
+      .catch(ignoreIfPrimaryLeaseLoss)
+      .catch(e =>
+        this.disableNetworkUntilRecovery(e, () =>
+          this.localStore.setLastStreamToken(this.writeStream.lastStreamToken)
+        )
+      );
   }
 
-  private onMutationResult(
+  private async onMutationResult(
     commitVersion: SnapshotVersion,
     results: MutationResult[]
   ): Promise<void> {
@@ -674,11 +696,18 @@ export class RemoteStore implements TargetMetadataProvider {
       results,
       this.writeStream.lastStreamToken
     );
-    return this.syncEngine.applySuccessfulWrite(success).then(() => {
-      // It's possible that with the completion of this mutation another
-      // slot has freed up.
-      return this.fillWritePipeline();
-    });
+    try {
+      await this.syncEngine.applySuccessfulWrite(success);
+    } catch (e) {
+      await this.disableNetworkUntilRecovery(e, () =>
+        this.syncEngine.applySuccessfulWrite(success)
+      );
+      return;
+    }
+
+    // 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> {
@@ -727,7 +756,12 @@ export class RemoteStore implements TargetMetadataProvider {
 
       return this.localStore
         .setLastStreamToken(ByteString.EMPTY_BYTE_STRING)
-        .catch(ignoreIfPrimaryLeaseLoss);
+        .catch(ignoreIfPrimaryLeaseLoss)
+        .catch(e =>
+          this.disableNetworkUntilRecovery(e, () =>
+            this.localStore.setLastStreamToken(ByteString.EMPTY_BYTE_STRING)
+          )
+        );
     } else {
       // Some other error, don't reset stream token. Our stream logic will
       // just retry with exponential backoff.
@@ -747,13 +781,18 @@ 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();
-        });
+      try {
+        await this.syncEngine.rejectFailedWrite(batch.batchId, error);
+      } catch (e) {
+        await this.disableNetworkUntilRecovery(e, () =>
+          this.syncEngine.rejectFailedWrite(batch.batchId, error)
+        );
+        return;
+      }
+
+      // 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',
     [],
