retrying transactions with backoff, override asyncQueue

Author: Brian Chen

packages/firestore/src/api/database.ts

@@ -307,9 +307,8 @@ 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.
-  readonly _queue = new AsyncQueue();
+  private _queue = new AsyncQueue();
 
   _dataConverter: UserDataConverter;
 
@@ -456,6 +455,16 @@ 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

@@ -33,7 +33,7 @@ import { Platform } from '../platform/platform';
 import { Datastore } from '../remote/datastore';
 import { RemoteStore } from '../remote/remote_store';
 import { JsonProtoSerializer } from '../remote/serializer';
-import { AsyncQueue } from '../util/async_queue';
+import { AsyncQueue, TimerId } from '../util/async_queue';
 import { Code, FirestoreError } from '../util/error';
 import { debug } from '../util/log';
 import { Deferred } from '../util/promise';
@@ -62,6 +62,7 @@ import { Query } from './query';
 import { Transaction } from './transaction';
 import { OnlineState, OnlineStateSource } from './types';
 import { ViewSnapshot } from './view_snapshot';
+import { ExponentialBackoff } from '../remote/backoff';
 
 const LOG_TAG = 'FirestoreClient';
 
@@ -70,6 +71,17 @@ 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,
@@ -613,6 +625,15 @@ export class FirestoreClient {
     // We have to wait for the async queue to be sure syncEngine is initialized.
     return this.asyncQueue
       .enqueue(async () => {})
-      .then(() => this.syncEngine.runTransaction(updateFunction));
+      .then(() => {
+        const backoff = new ExponentialBackoff(
+          this.asyncQueue,
+          TimerId.RetryTransaction,
+          BACKOFF_INITIAL_DELAY_MS,
+          BACKOFF_FACTOR,
+          BACKOFF_MAX_DELAY_MS
+        );
+        return this.syncEngine.runTransaction(updateFunction, backoff);
+      });
   }
 }

packages/firestore/src/core/sync_engine.ts

@@ -72,6 +72,7 @@ import {
 } from './view';
 import { ViewSnapshot } from './view_snapshot';
 import { isPermanentError } from '../remote/rpc_error';
+import { ExponentialBackoff } from '../remote/backoff';
 
 const LOG_TAG = 'SyncEngine';
 
@@ -352,10 +353,10 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
    * Takes an updateFunction in which a set of reads and writes can be performed
    * atomically. In the updateFunction, the client can read and write values
    * using the supplied transaction object. After the updateFunction, all
-   * changes will be committed. If some other client has changed any of the data
-   * referenced, then the updateFunction will be called again. If the
-   * updateFunction still fails after the given number of retries, then the
-   * transaction will be rejected.
+   * changes will be committed. If a retryable error occurs (ex: some other
+   * client has changed any of the data referenced), then the updateFunction
+   * will be called again after a backoff. If the updateFunction still fails
+   * after the given number of retries, then the transaction will be rejected.
    *
    * The transaction object passed to the updateFunction contains methods for
    * accessing documents and collections. Unlike other datastore access, data
@@ -367,30 +368,41 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
    */
   async runTransaction<T>(
     updateFunction: (transaction: Transaction) => Promise<T>,
+    backoff: ExponentialBackoff,
     retries = 5
   ): Promise<T> {
     assert(retries >= 0, 'Got negative number of retries for transaction.');
-    const transaction = this.remoteStore.createTransaction();
-    const userPromise = updateFunction(transaction);
-    if (
-      isNullOrUndefined(userPromise) ||
-      !userPromise.catch ||
-      !userPromise.then
-    ) {
-      return Promise.reject<T>(
-        Error('Transaction callback must return a Promise')
-      );
-    }
-    try {
-      const result = await userPromise;
-      await transaction.commit();
-      return result;
-    } catch (error) {
-      if (retries > 0 && this.isRetryableTransactionError(error)) {
-        return this.runTransaction(updateFunction, retries - 1);
+    let transactionError;
+    for (let i = 0; i < retries; i++) {
+      const backoffDeferred = new Deferred<void>();
+      backoff.backoffAndRun(async () => {
+        backoffDeferred.resolve();
+      });
+
+      try {
+        await backoffDeferred.promise;
+        const transaction = this.remoteStore.createTransaction();
+        const userPromise = updateFunction(transaction);
+        if (
+          isNullOrUndefined(userPromise) ||
+          !userPromise.catch ||
+          !userPromise.then
+        ) {
+          return Promise.reject<T>(
+            Error('Transaction callback must return a Promise')
+          );
+        }
+        const result = await userPromise;
+        await transaction.commit();
+        return result;
+      } catch (error) {
+        transactionError = error;
+        if (!this.isRetryableTransactionError(error)) {
+          break;
+        }
       }
-      return Promise.reject<T>(error);
     }
+    return Promise.reject<T>(transactionError);
   }
 
   async applyRemoteEvent(remoteEvent: RemoteEvent): Promise<void> {

packages/firestore/src/util/async_queue.ts

@@ -59,7 +59,10 @@ export enum TimerId {
   ClientMetadataRefresh = 'client_metadata_refresh',
 
   /** A timer used to periodically attempt LRU Garbage collection */
-  LruGarbageCollection = 'lru_garbage_collection'
+  LruGarbageCollection = 'lru_garbage_collection',
+
+  /** A timer used to retry transactions. */
+  RetryTransaction = 'retry_transaction'
 }
 
 /**
@@ -263,13 +266,6 @@ export class AsyncQueue {
       `Attempted to schedule an operation with a negative delay of ${delayMs}`
     );
 
-    // While not necessarily harmful, we currently don't expect to have multiple
-    // ops with the same timer id in the queue, so defensively reject them.
-    assert(
-      !this.containsDelayedOperation(timerId),
-      `Attempted to schedule multiple operations with timer id ${timerId}.`
-    );
-
     const delayedOp = DelayedOperation.createAndSchedule<unknown>(
       this,
       timerId,

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

@@ -20,9 +20,31 @@ 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';
 
-const apiDescribe = integrationHelpers.apiDescribe;
+/**
+ * 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;
+  }
+}
 
+const apiDescribe = integrationHelpers.apiDescribe;
 apiDescribe('Database transactions', (persistence: boolean) => {
   type TransactionStage = (
     transaction: firestore.Transaction,
@@ -460,6 +482,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
     let started = 0;
 
     return integrationHelpers.withTestDb(persistence, db => {
+      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
       const doc = db.collection('counters').doc();
       return doc
         .set({
@@ -516,6 +539,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
     let counter = 0;
 
     return integrationHelpers.withTestDb(persistence, db => {
+      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
       const doc = db.collection('counters').doc();
       return doc
         .set({
@@ -654,6 +678,7 @@ apiDescribe('Database transactions', (persistence: boolean) => {
 
   it('handle reading a doc twice with different versions', () => {
     return integrationHelpers.withTestDb(persistence, db => {
+      setAsyncQueue(db, new AsyncQueueWithoutTransactionBackoff());
       const doc = db.collection('counters').doc();
       let counter = 0;
       return doc

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

@@ -31,7 +31,15 @@ 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)._queue;
+  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);
 }
 
 export function getDefaultDatabaseInfo(): DatabaseInfo {