fix: Retry EPIPE Connection Errors + Attempt Retries in Resumable Upload Connection Errors (#2040)

* feat: Add `epipe` as retryable error

* fix: capture and retry potential connection errors

* test: Add tests and remove logs

* test: set `retryOptions` by copy rather than reference

* fix: grammar

Author: d-goog

src/resumable-upload.ts

@@ -742,9 +742,18 @@ export class Upload extends Writable {
         responseReceived = true;
         this.responseHandler(resp);
       }
-    } catch (err) {
-      const e = err as Error;
-      this.destroy(e);
+    } catch (e) {
+      const err = e as ApiError;
+
+      if (this.retryOptions.retryableErrorFn!(err)) {
+        this.attemptDelayedRetry({
+          status: NaN,
+          data: err,
+        });
+        return;
+      }
+
+      this.destroy(err);
     }
   }
 
@@ -833,7 +842,16 @@ export class Upload extends Writable {
       }
       this.offset = 0;
     } catch (e) {
-      const err = e as GaxiosError;
+      const err = e as ApiError;
+
+      if (this.retryOptions.retryableErrorFn!(err)) {
+        this.attemptDelayedRetry({
+          status: NaN,
+          data: err,
+        });
+        return;
+      }
+
       this.destroy(err);
     }
   }
@@ -922,7 +940,7 @@ export class Upload extends Writable {
   /**
    * @param resp GaxiosResponse object from previous attempt
    */
-  private attemptDelayedRetry(resp: GaxiosResponse) {
+  private attemptDelayedRetry(resp: Pick<GaxiosResponse, 'data' | 'status'>) {
     if (this.numRetries < this.retryOptions.maxRetries!) {
       if (
         resp.status === NOT_FOUND_STATUS_CODE &&

src/storage.ts

@@ -261,11 +261,12 @@ const IDEMPOTENCY_STRATEGY_DEFAULT = IdempotencyStrategy.RetryConditional;
  * @return {boolean} True if the API request should be retried, false otherwise.
  */
 export const RETRYABLE_ERR_FN_DEFAULT = function (err?: ApiError) {
-  const isConnectionProblem = (reason: string | undefined) => {
+  const isConnectionProblem = (reason: string) => {
     return (
-      (reason && reason.includes('eai_again')) || //DNS lookup error
+      reason.includes('eai_again') || // DNS lookup error
       reason === 'econnreset' ||
-      reason === 'unexpected connection closure'
+      reason === 'unexpected connection closure' ||
+      reason === 'epipe'
     );
   };
 
@@ -284,7 +285,7 @@ export const RETRYABLE_ERR_FN_DEFAULT = function (err?: ApiError) {
     if (err.errors) {
       for (const e of err.errors) {
         const reason = e?.reason?.toString().toLowerCase();
-        if (isConnectionProblem(reason)) {
+        if (reason && isConnectionProblem(reason)) {
           return true;
         }
       }

test/index.ts

@@ -311,6 +311,20 @@ describe('Storage', () => {
       assert.strictEqual(calledWith.retryOptions.retryableErrorFn(error), true);
     });
 
+    it('should retry a broken pipe error', () => {
+      const storage = new Storage({
+        projectId: PROJECT_ID,
+      });
+      const calledWith = storage.calledWith_[0];
+      const error = new ApiError('Broken pipe');
+      error.errors = [
+        {
+          reason: 'EPIPE',
+        },
+      ];
+      assert.strictEqual(calledWith.retryOptions.retryableErrorFn(error), true);
+    });
+
     it('should not retry a 999 error', () => {
       const storage = new Storage({
         projectId: PROJECT_ID,

test/resumable-upload.ts

@@ -112,7 +112,7 @@ describe('resumable-upload', () => {
       userProject: USER_PROJECT,
       authConfig: {keyFile},
       apiEndpoint: API_ENDPOINT,
-      retryOptions: RETRY_OPTIONS,
+      retryOptions: {...RETRY_OPTIONS},
     });
   });
 
@@ -1028,6 +1028,22 @@ describe('resumable-upload', () => {
       up.startUploading();
     });
 
+    it('should retry retryable errors if the request failed', done => {
+      const error = new Error('Error.');
+
+      // mock as retryable
+      up.retryOptions.retryableErrorFn = () => true;
+
+      up.on('error', done);
+      up.attemptDelayedRetry = () => done();
+
+      up.makeRequestStream = async () => {
+        throw error;
+      };
+
+      up.startUploading();
+    });
+
     describe('request preparation', () => {
       // a convenient handle for getting the request options
       let reqOpts: GaxiosOptions;
@@ -1384,6 +1400,22 @@ describe('resumable-upload', () => {
       await up.getAndSetOffset();
       assert.strictEqual(up.offset, 0);
     });
+
+    it('should retry retryable errors if the request failed', done => {
+      const error = new Error('Error.');
+
+      // mock as retryable
+      up.retryOptions.retryableErrorFn = () => true;
+
+      up.on('error', done);
+      up.attemptDelayedRetry = () => done();
+
+      up.makeRequest = async () => {
+        throw error;
+      };
+
+      up.getAndSetOffset();
+    });
   });
 
   describe('#makeRequest', () => {