Write stream retries

Author: schmidt-sebastian

packages/firestore/src/core/sync_engine.ts

@@ -530,18 +530,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) {
@@ -555,16 +555,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) {

packages/firestore/src/remote/remote_store.ts

@@ -125,14 +125,21 @@ export class RemoteStore implements TargetMetadataProvider {
 
   private onlineStateTracker: OnlineStateTracker;
 
+  /**
+   * A barrier to track unresolved operations that block the restart of the
+   * write stream. This is used to remove writes from the mutation queue if the
+   * initial removal attempt failed.
+   */
+  private writeStreamBarrier = 0;
+
   constructor(
     /**
      * The local store, used to fill the write pipeline with outbound mutations.
      */
     private localStore: LocalStore,
     /** The client-side proxy for interacting with the backend. */
     private datastore: Datastore,
-    asyncQueue: AsyncQueue,
+    private asyncQueue: AsyncQueue,
     onlineStateHandler: (onlineState: OnlineState) => void,
     connectivityMonitor: ConnectivityMonitor
   ) {
@@ -184,9 +191,12 @@ export class RemoteStore implements TargetMetadataProvider {
   }
 
   /** Re-enables the network. Idempotent. */
-  async enableNetwork(): Promise<void> {
+  enableNetwork(): Promise<void> {
     this.networkEnabled = true;
+    return this.enableNetworkInternal();
+  }
 
+  async enableNetworkInternal(): Promise<void> {
     if (this.canUseNetwork()) {
       if (this.shouldStartWatchStream()) {
         this.startWatchStream();
@@ -226,6 +236,29 @@ export class RemoteStore implements TargetMetadataProvider {
     this.cleanUpWatchStreamState();
   }
 
+  /**
+   * Recovery logic for IndexedDB errors that takes the network offline until
+   * `op` succeeds. Retries are scheduled with backoff using `enqueueRetryable()`.
+   */
+  private async disableNetworkUntilRecovery(
+    e: FirestoreError,
+    op: () => Promise<void>
+  ): Promise<void> {
+    if (e.name === 'IndexedDbTransactionError') {
+      // Increment the write stream barrier to prevent out of band stream
+      // restarts.
+      ++this.writeStreamBarrier;
+      await this.disableNetworkInternal();
+      this.asyncQueue.enqueueRetryable(async () => {
+        await op();
+        --this.writeStreamBarrier;
+        await this.enableNetworkInternal();
+      });
+    } else {
+      throw e;
+    }
+  }
+
   async shutdown(): Promise<void> {
     logDebug(LOG_TAG, 'RemoteStore shutting down.');
     this.networkEnabled = false;
@@ -337,7 +370,9 @@ export class RemoteStore implements TargetMetadataProvider {
   }
 
   canUseNetwork(): boolean {
-    return this.isPrimary && this.networkEnabled;
+    return (
+      this.writeStreamBarrier === 0 && this.isPrimary && this.networkEnabled
+    );
   }
 
   private cleanUpWatchStreamState(): void {
@@ -510,27 +545,31 @@ 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
-      );
-
-      if (batch === null) {
-        if (this.writePipeline.length === 0) {
-          this.writeStream.markIdle();
+    try {
+      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) {
+          this.addToWritePipeline(batch);
+        } else {
+          break;
         }
-      } else {
-        this.addToWritePipeline(batch);
-        await this.fillWritePipeline();
       }
-    }
 
-    if (this.shouldStartWriteStream()) {
-      this.startWriteStream();
+      if (this.shouldStartWriteStream()) {
+        this.startWriteStream();
+      } else if (this.writePipeline.length === 0) {
+        this.writeStream.markIdle();
+      }
+    } catch (e) {
+      await this.disableNetworkUntilRecovery(e, () => Promise.resolve());
     }
   }
 
@@ -568,6 +607,7 @@ export class RemoteStore implements TargetMetadataProvider {
   private shouldStartWriteStream(): boolean {
     return (
       this.canUseNetwork() &&
+      this.writeStreamBarrier === 0 &&
       !this.writeStream.isStarted() &&
       this.writePipeline.length > 0
     );
@@ -592,7 +632,7 @@ export class RemoteStore implements TargetMetadataProvider {
     }
   }
 
-  private onMutationResult(
+  private async onMutationResult(
     commitVersion: SnapshotVersion,
     results: MutationResult[]
   ): Promise<void> {
@@ -604,11 +644,16 @@ export class RemoteStore implements TargetMetadataProvider {
     );
     const batch = this.writePipeline.shift()!;
     const success = MutationBatchResult.from(batch, commitVersion, results);
-    return this.syncEngine.applySuccessfulWrite(success).then(() => {
+    try {
+      await this.syncEngine.applySuccessfulWrite(success);
       // It's possible that with the completion of this mutation another
       // slot has freed up.
-      return this.fillWritePipeline();
-    });
+      await this.fillWritePipeline();
+    } catch (e) {
+      await this.disableNetworkUntilRecovery(e, () =>
+        this.syncEngine.applySuccessfulWrite(success)
+      );
+    }
   }
 
   private async onWriteStreamClose(error?: FirestoreError): Promise<void> {
@@ -621,8 +666,8 @@ export class RemoteStore implements TargetMetadataProvider {
       );
     }
 
-    // If the write stream closed after the write handshake completes, a write
-    // operation failed and we fail the pending operation.
+    // An error that occurs after the write handshake completes is an indication
+    // that the write operation itself failed.
     if (error && this.writeStream.handshakeComplete) {
       // This error affects the actual write.
       await this.handleWriteError(error!);
@@ -648,13 +693,16 @@ 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);
+        // It's possible that with the completion of this mutation
+        // another slot has freed up.
+        await this.fillWritePipeline();
+      } catch (e) {
+        await this.disableNetworkUntilRecovery(e, () =>
+          this.syncEngine.rejectFailedWrite(batch.batchId, error)
+        );
+      }
     } else {
       // Transient error, just let the retry logic kick in.
     }

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

@@ -20,6 +20,8 @@ import { client, spec } from './spec_builder';
 import { TimerId } from '../../../src/util/async_queue';
 import { Query } from '../../../src/core/query';
 import { doc, path } from '../../util/helpers';
+import { Code } from '../../../src/util/error';
+import { RpcError } from './spec_rpc_error';
 
 describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
   specTest(
@@ -162,4 +164,112 @@ describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
       .watchAcksFull(query, 2, doc1, doc3)
       .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)
+      .failDatabase()
+      .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')
+      )
+      .failDatabase()
+      .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)
+        .failDatabase()
+        .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 })
+        .failDatabase()
+        .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`.
+        .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] })
+    );
+  });
 });