fix: retry on abort and limit retry count to 10

dev/src/backoff.ts

@@ -135,6 +135,13 @@ export class ExponentialBackoff {
    */
   private readonly jitterFactor: number;
 
+  /**
+   * The number of retries that has been attempted.
+   *
+   * @private
+   */
+  private retryCount = 0;
+
   /**
    * The backoff delay of the current attempt.
    *
@@ -162,7 +169,7 @@ export class ExponentialBackoff {
   }
 
   /**
-   * Resets the backoff delay.
+   * Resets the backoff delay and retry count.
    *
    * The very next backoffAndWait() will have no delay. If it is called again
    * (i.e. due to an error), initialDelayMs (plus jitter) will be used, and
@@ -171,6 +178,7 @@ export class ExponentialBackoff {
    * @private
    */
   reset(): void {
+    this.retryCount = 0;
     this.currentBaseMs = 0;
   }
 
@@ -209,12 +217,16 @@ export class ExponentialBackoff {
     this.currentBaseMs *= this.backoffFactor;
     this.currentBaseMs = Math.max(this.currentBaseMs, this.initialDelayMs);
     this.currentBaseMs = Math.min(this.currentBaseMs, this.maxDelayMs);
-
+    this.retryCount += 1;
     return new Promise(resolve => {
       delayExecution(resolve, delayWithJitterMs);
     });
   }
 
+  getRetryCount(): number {
+    return this.retryCount;
+  }
+
   /**
    * Returns a randomized "jitter" delay based on the current base and jitter
    * factor.

dev/src/watch.ts

@@ -40,6 +40,12 @@ import api = google.firestore.v1;
  */
 const WATCH_TARGET_ID = 0x1;
 
+/*!
+ * The default maximum number of retries that will be attempted by backoff
+ * before stopping all retry attempts.
+ */
+export const DEFAULT_MAX_RETRY_ATTEMPTS = 10;
+
 /*!
  * The change type for document change events.
  */
@@ -271,6 +277,7 @@ abstract class Watch {
     }
 
     switch (error.code) {
+      case GRPC_STATUS_CODE.ABORTED:
       case GRPC_STATUS_CODE.CANCELLED:
       case GRPC_STATUS_CODE.UNKNOWN:
       case GRPC_STATUS_CODE.DEADLINE_EXCEEDED:
@@ -426,6 +433,15 @@ abstract class Watch {
      * Initializes a new stream to the backend with backoff.
      */
     const initStream = () => {
+      if (this._backoff.getRetryCount() > DEFAULT_MAX_RETRY_ATTEMPTS) {
+        closeStream(
+          new Error(
+            'Exceeded maximum number of retries before any ' +
+              'response was received.'
+          )
+        );
+        return;
+      }
       this._backoff.backoffAndWait().then(async () => {
         if (!isActive) {
           logger(

dev/test/backoff.ts

@@ -144,4 +144,19 @@ describe('ExponentialBackoff', () => {
     await backoff.backoffAndWait().then(nop);
     assertDelayBetween(36, 44);
   });
+
+  it('tracks the number of retry attempts', async () => {
+    const backoff = new ExponentialBackoff({
+      initialDelayMs: 10,
+      backoffFactor: 2,
+      jitterFactor: 0.1,
+    });
+    expect(backoff.getRetryCount()).to.equal(0);
+    await backoff.backoffAndWait().then(nop);
+    expect(backoff.getRetryCount()).to.equal(1);
+    await backoff.backoffAndWait().then(nop);
+    expect(backoff.getRetryCount()).to.equal(2);
+    backoff.reset();
+    expect(backoff.getRetryCount()).to.equal(0);
+  });
 });

dev/test/watch.ts

@@ -44,7 +44,12 @@ import {DocumentSnapshotBuilder} from '../src/document';
 import {DocumentChangeType} from '../src/document-change';
 import {Serializer} from '../src/serializer';
 import {GrpcError} from '../src/types';
-import {createInstance, InvalidApiUsage, verifyInstance} from './util/helpers';
+import {DEFAULT_MAX_RETRY_ATTEMPTS} from './../src/watch';
+import {
+  createInstance,
+  InvalidApiUsage,
+  verifyInstance,
+} from './util/helpers';
 
 import api = google.firestore.v1;
 
@@ -819,6 +824,36 @@ describe('Query watch', () => {
     });
   });
 
+  it('stops attempts after maximum retry attempts', () => {
+    const err = new GrpcError('GRPC Error');
+    err.code = Number(10 /* ABORTED */);
+    return watchHelper.runFailedTest(
+      collQueryJSON(),
+      async () => {
+        let chain = Promise.resolve();
+        // Retry for the maximum of retry attempts.
+        for (let i = 0; i < DEFAULT_MAX_RETRY_ATTEMPTS; i++) {
+          chain = chain.then(async () => {
+            streamHelper.destroyStream(err);
+            await streamHelper.awaitReopen();
+          });
+        }
+        // The next retry should fail with an error.
+        chain
+          .then(() => {
+            streamHelper.destroyStream(err);
+          })
+          .then(() => {
+            return streamHelper.await('error');
+          })
+          .then(() => {
+            return streamHelper.await('close');
+          });
+      },
+      'Exceeded maximum number of retries before any response was received.'
+    );
+  });
+
   it("doesn't re-open inactive stream", () => {
     // This test uses the normal timeout handler since it relies on the actual
     // backoff window during the the stream recovery. We then use this window to
@@ -854,7 +889,7 @@ describe('Query watch', () => {
       /* PermissionDenied */ 7: false,
       /* ResourceExhausted */ 8: true,
       /* FailedPrecondition */ 9: false,
-      /* Aborted */ 10: false,
+      /* Aborted */ 10: true,
       /* OutOfRange */ 11: false,
       /* Unimplemented */ 12: false,
       /* Internal */ 13: true,