Don't delay retryable operations (#3270)

Author: schmidt-sebastian

packages/firestore/src/util/async_queue.ts

@@ -207,9 +207,9 @@ export class AsyncQueue {
   // The last promise in the queue.
   private tail: Promise<unknown> = Promise.resolve();
 
-  // The last retryable operation. Retryable operation are run in order and
+  // A list of retryable operations. Retryable operations are run in order and
   // retried with backoff.
-  private retryableTail: Promise<void> = Promise.resolve();
+  private retryableOps: Array<() => Promise<void>> = [];
 
   // Is this AsyncQueue being shut down? Once it is set to true, it will not
   // be changed again.
@@ -323,32 +323,44 @@ export class AsyncQueue {
    * operations were retried successfully.
    */
   enqueueRetryable(op: () => Promise<void>): void {
-    this.verifyNotFailed();
+    this.retryableOps.push(op);
+    this.enqueueAndForget(() => this.retryNextOp());
+  }
 
-    if (this._isShuttingDown) {
+  /**
+   * Runs the next operation from the retryable queue. If the operation fails,
+   * reschedules with backoff.
+   */
+  private async retryNextOp(): Promise<void> {
+    if (this.retryableOps.length === 0) {
       return;
     }
 
-    this.retryableTail = this.retryableTail.then(() => {
-      const deferred = new Deferred<void>();
-      const retryingOp = async (): Promise<void> => {
-        try {
-          await op();
-          deferred.resolve();
-          this.backoff.reset();
-        } catch (e) {
-          if (isIndexedDbTransactionError(e)) {
-            logDebug(LOG_TAG, 'Operation failed with retryable error: ' + e);
-            this.backoff.backoffAndRun(retryingOp);
-          } else {
-            deferred.resolve();
-            throw e; // Failure will be handled by AsyncQueue
-          }
-        }
-      };
-      this.enqueueAndForget(retryingOp);
-      return deferred.promise;
-    });
+    try {
+      await this.retryableOps[0]();
+      this.retryableOps.shift();
+      this.backoff.reset();
+    } catch (e) {
+      if (isIndexedDbTransactionError(e)) {
+        logDebug(LOG_TAG, 'Operation failed with retryable error: ' + e);
+      } else {
+        throw e; // Failure will be handled by AsyncQueue
+      }
+    }
+
+    if (this.retryableOps.length > 0) {
+      // If there are additional operations, we re-schedule `retryNextOp()`.
+      // This is necessary to run retryable operations that failed during
+      // their initial attempt since we don't know whether they are already
+      // enqueued. If, for example, `op1`, `op2`, `op3` are enqueued and `op1`
+      // needs to  be re-run, we will run `op1`, `op1`, `op2` using the
+      // already enqueued calls to `retryNextOp()`. `op3()` will then run in the
+      // call scheduled here.
+      // Since `backoffAndRun()` cancels an existing backoff and schedules a
+      // new backoff on every call, there is only ever a single additional
+      // operation in the queue.
+      this.backoff.backoffAndRun(() => this.retryNextOp());
+    }
   }
 
   private enqueueInternal<T extends unknown>(op: () => Promise<T>): Promise<T> {

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

@@ -787,6 +787,7 @@ describeSpec('Persistence Recovery', ['no-ios', 'no-android'], () => {
           .expectActiveTargets({ query })
           // We are now user 2
           .expectEvents(query, { removed: [doc1], fromCache: true })
+          .runTimer(TimerId.AsyncQueueRetry)
           // We are now user 1
           .expectEvents(query, {
             added: [doc1],

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

@@ -277,7 +277,7 @@ abstract class TestRunner {
   }
 
   async shutdown(): Promise<void> {
-    await this.queue.enqueue(async () => {
+    await this.queue.enqueueAndInitiateShutdown(async () => {
       if (this.started) {
         await this.doShutdown();
       }

packages/firestore/test/unit/util/async_queue.test.ts

@@ -297,8 +297,7 @@ describe('AsyncQueue', () => {
 
     queue.enqueueRetryable(async () => {
       doStep(1);
-      if (completedSteps.length > 1) {
-      } else {
+      if (completedSteps.length === 1) {
         throw new IndexedDbTransactionError(
           new Error('Simulated retryable error')
         );
@@ -313,6 +312,65 @@ describe('AsyncQueue', () => {
     expect(completedSteps).to.deep.equal([1, 1, 2]);
   });
 
+  it('Does not delay retryable operations that succeed', async () => {
+    const queue = new AsyncQueue();
+    const completedSteps: number[] = [];
+    const doStep = (n: number): void => {
+      completedSteps.push(n);
+    };
+
+    queue.enqueueRetryable(async () => {
+      doStep(1);
+    });
+    queue.enqueueAndForget(async () => {
+      doStep(2);
+    });
+    await queue.enqueue(async () => {
+      doStep(3);
+    });
+
+    expect(completedSteps).to.deep.equal([1, 2, 3]);
+  });
+
+  it('Catches up when retryable operation fails', async () => {
+    const queue = new AsyncQueue();
+    const completedSteps: number[] = [];
+    const doStep = (n: number): void => {
+      completedSteps.push(n);
+    };
+
+    const blockingPromise = new Deferred<void>();
+
+    queue.enqueueRetryable(async () => {
+      doStep(1);
+      if (completedSteps.length === 1) {
+        throw new IndexedDbTransactionError(
+          new Error('Simulated retryable error')
+        );
+      }
+    });
+    queue.enqueueAndForget(async () => {
+      doStep(2);
+    });
+    queue.enqueueRetryable(async () => {
+      doStep(3);
+      blockingPromise.resolve();
+    });
+    await blockingPromise.promise;
+
+    // Once all existing retryable operations succeeded, they are scheduled
+    // in the order they are enqueued.
+    queue.enqueueAndForget(async () => {
+      doStep(4);
+    });
+    await queue.enqueue(async () => {
+      doStep(5);
+    });
+
+    await blockingPromise.promise;
+    expect(completedSteps).to.deep.equal([1, 2, 1, 3, 4, 5]);
+  });
+
   it('Can drain (non-delayed) operations', async () => {
     const queue = new AsyncQueue();
     const completedSteps: number[] = [];