Retry WebStorage operations

Author: schmidt-sebastian

packages/firestore/src/local/shared_client_state.ts

@@ -755,7 +755,7 @@ export class WebStorageSharedClientState implements SharedClientState {
         return;
       }
 
-      this.queue.enqueueAndForget(async () => {
+      this.queue.enqueueRetryable(async () => {
         if (!this.started) {
           this.earlyEvents.push(event);
           return;

packages/firestore/src/platform_node/grpc_connection.ts

@@ -1,4 +1,3 @@
-
 /**
  * @license
  * Copyright 2017 Google LLC

packages/firestore/src/remote/backoff.ts

@@ -119,24 +119,28 @@ export class ExponentialBackoff {
       desiredDelayWithJitterMs - delaySoFarMs
     );
 
-    if (this.currentBaseMs > 0) {
+    if (remainingDelayMs > 0) {
       logDebug(
         LOG_TAG,
         `Backing off for ${remainingDelayMs} ms ` +
           `(base delay: ${this.currentBaseMs} ms, ` +
           `delay with jitter: ${desiredDelayWithJitterMs} ms, ` +
           `last attempt: ${delaySoFarMs} ms ago)`
       );
-    }
-
-    this.timerPromise = this.queue.enqueueAfterDelay(
-      this.timerId,
-      remainingDelayMs,
-      () => {
+      this.timerPromise = this.queue.enqueueAfterDelay(
+        this.timerId,
+        remainingDelayMs,
+        () => {
+          this.lastAttemptTime = Date.now();
+          return op();
+        }
+      );
+    } else {
+      this.queue.enqueueAndForget(() => {
         this.lastAttemptTime = Date.now();
         return op();
-      }
-    );
+      });
+    }
 
     // Apply backoff factor to determine next delay and ensure it is within
     // bounds.

packages/firestore/src/util/async_queue.ts

@@ -17,8 +17,12 @@
 
 import { assert, fail } from './assert';
 import { Code, FirestoreError } from './error';
-import { logError } from './log';
+import { logDebug, logError } from './log';
 import { CancelablePromise, Deferred } from './promise';
+import { ExponentialBackoff } from '../remote/backoff';
+import { PlatformSupport } from '../platform/platform';
+
+const LOG_TAG = 'AsyncQueue';
 
 // Accept any return type from setTimeout().
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -66,7 +70,13 @@ export const enum TimerId {
    * A timer used to retry transactions. Since there can be multiple concurrent
    * transactions, multiple of these may be in the queue at a given time.
    */
-  RetryTransaction = 'retry_transaction'
+  RetryTransaction = 'retry_transaction',
+
+  /**
+   * A timer used to retry operations scheduled via retryable AsyncQueue
+   * operations.
+   */
+  AsyncQueueRetry = 'async_queue_retry'
 }
 
 /**
@@ -213,6 +223,29 @@ export class AsyncQueue {
   // List of TimerIds to fast-forward delays for.
   private timerIdsToSkip: TimerId[] = [];
 
+  // Backoff timer used for retryable operations
+  private backoff = new ExponentialBackoff(this, TimerId.AsyncQueueRetry);
+
+  // If set, points to the first retryable operation. The first retryable
+  // operation is the only operation that is retried with backoff. If this
+  // operation suceeds, all other retryable operations are run right away.
+  private firstRetryableOperation?: Promise<void>;
+
+  // Visibility handler that triggers an immediate retry of all retryable
+  // operations. Meant to speed up recovery when we regain file system access
+  // after page comes into foreground.
+  private visibilityHandler = (): void => {
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    this.runDelayedOperationsEarly(TimerId.AsyncQueueRetry);
+  };
+
+  constructor() {
+    const window = PlatformSupport.getPlatform().window;
+    if (window) {
+      window.addEventListener('visibilitychange', this.visibilityHandler);
+    }
+  }
+
   // Is this AsyncQueue being shut down? If true, this instance will not enqueue
   // any new operations, Promises from enqueue requests will not resolve.
   get isShuttingDown(): boolean {
@@ -262,6 +295,10 @@ export class AsyncQueue {
     this.verifyNotFailed();
     if (!this._isShuttingDown) {
       this._isShuttingDown = true;
+      const window = PlatformSupport.getPlatform().window;
+      if (window) {
+        window.removeEventListener('visibilitychange', this.visibilityHandler);
+      }
       await this.enqueueEvenAfterShutdown(op);
     }
   }
@@ -279,6 +316,41 @@ export class AsyncQueue {
     return this.enqueueInternal(op);
   }
 
+  enqueueRetryable(op: () => Promise<void>): void {
+    this.verifyNotFailed();
+
+    if (this._isShuttingDown) {
+      return;
+    }
+
+    if (this.firstRetryableOperation) {
+      // If there is already a retryable operation, enqueue the current
+      // operation right after. We want to successfully run the first retryable
+      // operation before running all others.
+      // eslint-disable-next-line @typescript-eslint/no-floating-promises
+      this.firstRetryableOperation.then(() => {
+        this.enqueueRetryable(op);
+      });
+    } else {
+      const deferred = new Deferred<void>();
+      this.firstRetryableOperation = deferred.promise;
+
+      const retryingOp = async (): Promise<void> => {
+        try {
+          await op();
+          deferred.resolve();
+          this.backoff.reset();
+          this.firstRetryableOperation = undefined;
+        } catch (e) {
+          logDebug(LOG_TAG, 'Retryable operation failed: ' + e.message);
+          this.backoff.backoffAndRun(retryingOp);
+        }
+      };
+
+      this.backoff.backoffAndRun(retryingOp);
+    }
+  }
+
   private enqueueInternal<T extends unknown>(op: () => Promise<T>): Promise<T> {
     const newTail = this.tail.then(() => {
       this.operationInProgress = true;
@@ -399,12 +471,6 @@ export class AsyncQueue {
   runDelayedOperationsEarly(lastTimerId: TimerId): Promise<void> {
     // Note that draining may generate more delayed ops, so we do that first.
     return this.drain().then(() => {
-      assert(
-        lastTimerId === TimerId.All ||
-          this.containsDelayedOperation(lastTimerId),
-        `Attempted to drain to missing operation ${lastTimerId}`
-      );
-
       // Run ops in the same order they'd run if they ran naturally.
       this.delayedOperations.sort((a, b) => a.targetTimeMs - b.targetTimeMs);
 

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

@@ -0,0 +1,71 @@
+/**
+ * @license
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describeSpec, specTest } from './describe_spec';
+import { client } from './spec_builder';
+import { TimerId } from '../../../src/util/async_queue';
+import { Query } from '../../../src/core/query';
+import { path } from '../../util/helpers';
+
+describeSpec(
+  'Persistence Recovery',
+  ['durable-persistence', 'no-ios', 'no-android'],
+  () => {
+    specTest(
+      'Write is acknowledged by primary client (with recovery)',
+      ['multi-client'],
+      () => {
+        return client(0)
+          .expectPrimaryState(true)
+          .client(1)
+          .expectPrimaryState(false)
+          .userSets('collection/a', { v: 1 })
+          .failDatabase()
+          .client(0)
+          .writeAcks('collection/a', 1, { expectUserCallback: false })
+          .client(1)
+          .recoverDatabase()
+          .runTimer(TimerId.AsyncQueueRetry)
+          .expectUserCallbacks({
+            acknowledged: ['collection/a']
+          });
+      }
+    );
+
+    specTest(
+      'Query raises events in secondary client  (with recovery)',
+      ['multi-client'],
+      () => {
+        const query = Query.atPath(path('collection'));
+
+        return client(0)
+          .expectPrimaryState(true)
+          .client(1)
+          .expectPrimaryState(false)
+          .userListens(query)
+          .failDatabase()
+          .client(0)
+          .expectListen(query)
+          .watchAcksFull(query, 1000)
+          .client(1)
+          .recoverDatabase()
+          .runTimer(TimerId.AsyncQueueRetry)
+          .expectEvents(query, {});
+      }
+    );
+  }
+);

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

@@ -84,36 +84,4 @@ describeSpec('Remote store:', [], () => {
         .expectEvents(query, { added: [doc1] })
     );
   });
-
-  // TODO(b/72313632): This test is web-only because the Android / iOS spec
-  // tests exclude backoff entirely.
-  specTest(
-    'Handles user changes while offline (b/74749605).',
-    ['no-android', 'no-ios'],
-    () => {
-      const query = Query.atPath(path('collection'));
-      return (
-        spec()
-          .userListens(query)
-
-          // close the stream (this should trigger retry with backoff; but don't
-          // run it in an attempt to reproduce b/74749605).
-          .watchStreamCloses(Code.UNAVAILABLE, { runBackoffTimer: false })
-          .expectEvents(query, { fromCache: true })
-
-          // Because we didn't let the backoff timer run and restart the watch
-          // stream, there will be no active targets.
-          .expectActiveTargets()
-
-          // Change user (will shut down existing streams and start new ones).
-          .changeUser('abc')
-          // Our query should be sent to the new stream.
-          .expectActiveTargets({ query, resumeToken: '' })
-
-          // Close the (newly-created) stream as if it too failed (should trigger
-          // retry with backoff, potentially reproducing the crash in b/74749605).
-          .watchStreamCloses(Code.UNAVAILABLE)
-      );
-    }
-  );
 });

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

@@ -418,6 +418,22 @@ export class SpecBuilder {
     return this;
   }
 
+  failDatabase(): this {
+    this.nextStep();
+    this.currentStep = {
+      failDatabase: true
+    };
+    return this;
+  }
+
+  recoverDatabase(): this {
+    this.nextStep();
+    this.currentStep = {
+      failDatabase: false
+    };
+    return this;
+  }
+
   expectIsShutdown(): this {
     this.assertStep('Active target expectation requires previous step');
     const currentStep = this.currentStep!;
@@ -716,19 +732,14 @@ export class SpecBuilder {
     return this;
   }
 
-  watchStreamCloses(error: Code, opts?: { runBackoffTimer: boolean }): this {
-    if (!opts) {
-      opts = { runBackoffTimer: true };
-    }
-
+  watchStreamCloses(error: Code): this {
     this.nextStep();
     this.currentStep = {
       watchStreamClose: {
         error: {
           code: mapRpcCodeFromCode(error),
           message: 'Simulated Backend Error'
-        },
-        runBackoffTimer: opts.runBackoffTimer
+        }
       }
     };
     return this;