new attempt of asyncqueue

Author: Brian Chen

packages/firestore/src/api/database.ts

@@ -307,8 +307,9 @@ export class Firestore implements firestore.FirebaseFirestore, FirebaseService {
   // are already set to synchronize on the async queue.
   private _firestoreClient: FirestoreClient | undefined;
 
+  // Public for use in tests.
   // TODO(mikelehen): Use modularized initialization instead.
-  private _queue = new AsyncQueue();
+  readonly _queue = new AsyncQueue();
 
   _dataConverter: UserDataConverter;
 
@@ -455,16 +456,6 @@ export class Firestore implements firestore.FirebaseFirestore, FirebaseService {
     return this._firestoreClient as FirestoreClient;
   }
 
-  // Public for use in tests.
-  getQueue(): AsyncQueue {
-    return this._queue;
-  }
-
-  // Public for use in tests.
-  setQueue(newQueue: AsyncQueue): void {
-    this._queue = newQueue;
-  }
-
   private makeDatabaseInfo(): DatabaseInfo {
     return new DatabaseInfo(
       this._config.databaseId,

packages/firestore/src/core/firestore_client.ts

@@ -71,17 +71,6 @@ const DOM_EXCEPTION_INVALID_STATE = 11;
 const DOM_EXCEPTION_ABORTED = 20;
 const DOM_EXCEPTION_QUOTA_EXCEEDED = 22;
 
-/**
- * Initial backoff time in milliseconds after an error.
- * Set to 1s according to https://cloud.google.com/apis/design/errors.
- */
-const BACKOFF_INITIAL_DELAY_MS = 1000;
-
-/** Maximum backoff time in milliseconds */
-const BACKOFF_MAX_DELAY_MS = 60 * 1000;
-
-const BACKOFF_FACTOR = 1.5;
-
 export class IndexedDbPersistenceSettings {
   constructor(
     readonly cacheSizeBytes: number,
@@ -628,10 +617,7 @@ export class FirestoreClient {
       .then(() => {
         const backoff = new ExponentialBackoff(
           this.asyncQueue,
-          TimerId.RetryTransaction,
-          BACKOFF_INITIAL_DELAY_MS,
-          BACKOFF_FACTOR,
-          BACKOFF_MAX_DELAY_MS
+          TimerId.RetryTransaction
         );
         return this.syncEngine.runTransaction(updateFunction, backoff);
       });

packages/firestore/src/remote/backoff.ts

@@ -20,6 +20,17 @@ import * as log from '../util/log';
 import { CancelablePromise } from '../util/promise';
 const LOG_TAG = 'ExponentialBackoff';
 
+/**
+ * Initial backoff time in milliseconds after an error.
+ * Set to 1s according to https://cloud.google.com/apis/design/errors.
+ */
+const DEFAULT_BACKOFF_INITIAL_DELAY_MS = 1000;
+
+const DEFAULT_BACKOFF_FACTOR = 1.5;
+
+/** Maximum backoff time in milliseconds */
+const DEFAULT_BACKOFF_MAX_DELAY_MS = 60 * 1000;
+
 /**
  * A helper for running delayed tasks following an exponential backoff curve
  * between attempts.
@@ -49,18 +60,18 @@ export class ExponentialBackoff {
      * Note that jitter will still be applied, so the actual delay could be as
      * little as 0.5*initialDelayMs.
      */
-    private readonly initialDelayMs: number,
+    private readonly initialDelayMs: number = DEFAULT_BACKOFF_INITIAL_DELAY_MS,
     /**
      * The multiplier to use to determine the extended base delay after each
      * attempt.
      */
-    private readonly backoffFactor: number,
+    private readonly backoffFactor: number = DEFAULT_BACKOFF_FACTOR,
     /**
      * The maximum base delay after which no further backoff is performed.
      * Note that jitter will still be applied, so the actual delay could be as
      * much as 1.5*maxDelayMs.
      */
-    private readonly maxDelayMs: number
+    private readonly maxDelayMs: number = DEFAULT_BACKOFF_MAX_DELAY_MS
   ) {
     this.reset();
   }

packages/firestore/src/remote/persistent_stream.ts

@@ -115,17 +115,6 @@ export interface PersistentStreamListener {
   onClose: (err?: FirestoreError) => Promise<void>;
 }
 
-/**
- * Initial backoff time in milliseconds after an error.
- * Set to 1s according to https://cloud.google.com/apis/design/errors.
- */
-const BACKOFF_INITIAL_DELAY_MS = 1000;
-
-/** Maximum backoff time in milliseconds */
-const BACKOFF_MAX_DELAY_MS = 60 * 1000;
-
-const BACKOFF_FACTOR = 1.5;
-
 /** The time a stream stays open after it is marked idle. */
 const IDLE_TIMEOUT_MS = 60 * 1000;
 
@@ -187,13 +176,7 @@ export abstract class PersistentStream<
     private credentialsProvider: CredentialsProvider,
     protected listener: ListenerType
   ) {
-    this.backoff = new ExponentialBackoff(
-      queue,
-      connectionTimerId,
-      BACKOFF_INITIAL_DELAY_MS,
-      BACKOFF_FACTOR,
-      BACKOFF_MAX_DELAY_MS
-    );
+    this.backoff = new ExponentialBackoff(queue, connectionTimerId);
   }
 
   /**

packages/firestore/src/util/async_queue.ts

@@ -202,6 +202,9 @@ export class AsyncQueue {
   // assertion sanity-checks.
   private operationInProgress = false;
 
+  // List of TimerIds to fast-forward delays for.
+  private timerIdsToSkip: TimerId[] = [];
+
   /**
    * Adds a new operation to the queue without waiting for it to complete (i.e.
    * we ignore the Promise result).
@@ -275,6 +278,15 @@ export class AsyncQueue {
     );
     this.delayedOperations.push(delayedOp);
 
+    // Fast-forward delays for timerIds that have been overriden.
+    for (const timerIdToSkip of this.timerIdsToSkip) {
+      if (this.containsDelayedOperation(timerIdToSkip)) {
+        this.runDelayedOperationsEarly(timerIdToSkip)
+          .then(() => {})
+          .catch(() => 'obligatory catch');
+      }
+    }
+
     return delayedOp;
   }
 
@@ -352,6 +364,13 @@ export class AsyncQueue {
     });
   }
 
+  /**
+   * For Tests: Skip all delays for a timer id.
+   */
+  skipDelaysForTimerId(timerId: TimerId): void {
+    this.timerIdsToSkip.push(timerId);
+  }
+
   /** Called once a DelayedOperation is run or canceled. */
   private removeDelayedOperation(op: DelayedOperation<unknown>): void {
     // NOTE: indexOf / slice are O(n), but delayedOperations is expected to be small.

packages/firestore/test/integration/api/transactions.test.ts

@@ -20,29 +20,8 @@ import { expect } from 'chai';
 import { Deferred } from '../../util/promise';
 import firebase from '../util/firebase_export';
 import * as integrationHelpers from '../util/helpers';
-import { setAsyncQueue } from '../util/internal_helpers';
-import { AsyncQueue, TimerId } from '../../../src/util/async_queue';
-import { CancelablePromise } from '../../../src/util/promise';
-
-/**
- * Version of the AsyncQueue that overrides `enqueueAfterDelay` to fast-forward
- * the backoffs used in retrying transactions.
- */
-class AsyncQueueWithoutTransactionBackoff extends AsyncQueue {
-  enqueueAfterDelay<T extends unknown>(
-    timerId: TimerId,
-    delayMs: number,
-    op: () => Promise<T>
-  ): CancelablePromise<T> {
-    const result = super.enqueueAfterDelay(timerId, delayMs, op);
-    if (this.containsDelayedOperation(TimerId.RetryTransaction)) {
-      this.runDelayedOperationsEarly(TimerId.RetryTransaction)
-        .then(() => {})
-        .catch(() => 'obligatory catch');
-    }
-    return result;
-  }
-}
+import { asyncQueue } from '../util/internal_helpers';
+import { TimerId } from '../../../src/util/async_queue';
 
 const apiDescribe = integrationHelpers.apiDescribe;
 apiDescribe('Database transactions', (persistence: boolean) => {
@@ -482,7 +461,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
     let started = 0;
 
     return integrationHelpers.withTestDb(persistence, db => {
-      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
+      asyncQueue(db).skipDelaysForTimerId(TimerId.RetryTransaction);
       const doc = db.collection('counters').doc();
       return doc
         .set({
@@ -539,7 +518,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
     let counter = 0;
 
     return integrationHelpers.withTestDb(persistence, db => {
-      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
+      asyncQueue(db).skipDelaysForTimerId(TimerId.RetryTransaction);
       const doc = db.collection('counters').doc();
       return doc
         .set({
@@ -678,7 +657,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
 
   it('handle reading a doc twice with different versions', () => {
     return integrationHelpers.withTestDb(persistence, db => {
-      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
+      asyncQueue(db).skipDelaysForTimerId(TimerId.RetryTransaction);
       const doc = db.collection('counters').doc();
       let counter = 0;
       return doc

packages/firestore/test/integration/util/internal_helpers.ts

@@ -31,15 +31,7 @@ import { DEFAULT_PROJECT_ID, DEFAULT_SETTINGS } from './helpers';
 
 /** Helper to retrieve the AsyncQueue for a give FirebaseFirestore instance. */
 export function asyncQueue(db: firestore.FirebaseFirestore): AsyncQueue {
-  return (db as Firestore).getQueue();
-}
-
-/** Helper to override the AsyncQueue for a give FirebaseFirestore instance. */
-export function setAsyncQueue(
-  db: firestore.FirebaseFirestore,
-  newQueue: AsyncQueue
-): void {
-  (db as Firestore).setQueue(newQueue);
+  return (db as Firestore)._queue;
 }
 
 export function getDefaultDatabaseInfo(): DatabaseInfo {