fix: retry BulkWriter deletes that fail with RST_STREAM error (#1442)

* fix: retry BulkWriter deletes that fail with RST_STREAM errors * resolve comments * remove .only on test * remove undefined check * lint
googleapis · Mar 9, 2021 · cccf48d · cccf48d
1 parent 6c9319e
commit cccf48d
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 6 deletions.
diff --git a/dev/src/bulk-writer.ts b/dev/src/bulk-writer.ts
@@ -316,10 +316,12 @@ export class BulkWriter {
    * @private
    */
   private _errorFn: (error: BulkWriterError) => boolean = error => {
+    const isRetryableDeleteError =
+      error.operationType === 'delete' &&
+      (error.code as number) === Status.INTERNAL;
     const retryCodes = getRetryCodes('batchWrite');
     return (
-      error.code !== undefined &&
-      retryCodes.includes(error.code) &&
+      (retryCodes.includes(error.code) || isRetryableDeleteError) &&
       error.failedAttempts < MAX_RETRY_ATTEMPTS
     );
   };

diff --git a/dev/test/bulk-writer.ts b/dev/test/bulk-writer.ts
@@ -536,9 +536,9 @@ describe('BulkWriter', () => {
         ]),
         response: mergeResponses([
           successResponse(1),
-          failedResponse(Status.INTERNAL),
-          failedResponse(Status.INTERNAL),
-          failedResponse(Status.INTERNAL),
+          failedResponse(Status.CANCELLED),
+          failedResponse(Status.CANCELLED),
+          failedResponse(Status.CANCELLED),
         ]),
       },
       {
@@ -606,6 +606,35 @@ describe('BulkWriter', () => {
     expect(onWriteErrorCalled).to.be.true;
   });
 
+  it('retries INTERNAL errors for deletes', async () => {
+    const bulkWriter = await instantiateInstance([
+      {
+        request: createRequest([setOp('doc1', 'bar'), deleteOp('doc2')]),
+        response: mergeResponses([
+          failedResponse(Status.INTERNAL),
+          failedResponse(Status.INTERNAL),
+        ]),
+      },
+      {
+        request: createRequest([deleteOp('doc2')]),
+        response: successResponse(2),
+      },
+    ]);
+
+    let errorCaught = false;
+    bulkWriter
+      .set(firestore.doc('collectionId/doc1'), {
+        foo: 'bar',
+      })
+      .catch(() => {
+        errorCaught = true;
+      });
+    const del = bulkWriter.delete(firestore.doc('collectionId/doc2'));
+    await bulkWriter.close();
+    expect((await del).writeTime).to.deep.equal(new Timestamp(2, 0));
+    expect(errorCaught).to.be.true;
+  });
+
   it('surfaces errors thrown by user-provided error callback', async () => {
     const bulkWriter = await instantiateInstance([
       {
@@ -860,7 +889,7 @@ describe('BulkWriter', () => {
     return bulkWriter.close().then(() => verifyOpCount(2));
   });
 
-  it('retries individual writes that fail with ABORTED errors', async () => {
+  it('retries individual writes that fail with ABORTED and UNAVAILABLE errors', async () => {
     setTimeoutHandler(setImmediate);
     // Create mock responses that simulate one successful write followed by
     // failed responses.