Fixed issue where pausing a resumable upload throws an error (#6938)

Author: maneesht

.changeset/eleven-cycles-hang.md

@@ -0,0 +1,5 @@
+---
+"@firebase/storage": patch
+---
+
+Fixed issue where pause throws an error when a request is in flight.

packages/storage/src/implementation/request.ts

@@ -120,9 +120,9 @@ class NetworkRequest<I extends ConnectionType, O> implements Request<O> {
           const hitServer = connection.getErrorCode() === ErrorCode.NO_ERROR;
           const status = connection.getStatus();
           if (
-            (!hitServer ||
-              isRetryStatusCode(status, this.additionalRetryCodes_)) &&
-            this.retry
+            !hitServer ||
+            (isRetryStatusCode(status, this.additionalRetryCodes_) &&
+              this.retry)
           ) {
             const wasCanceled = connection.getErrorCode() === ErrorCode.ABORT;
             backoffCallback(

packages/storage/test/integration/integration.test.ts

@@ -36,6 +36,7 @@ import {
 import { use, expect } from 'chai';
 import chaiAsPromised from 'chai-as-promised';
 import * as types from '../../src/public-types';
+import { Deferred } from '@firebase/util';
 
 use(chaiAsPromised);
 
@@ -187,4 +188,32 @@ describe('FirebaseStorage Exp', () => {
     expect(listResult.items.map(v => v.name)).to.have.members(['a', 'b']);
     expect(listResult.prefixes.map(v => v.name)).to.have.members(['c']);
   });
+
+  it('can pause uploads without an error', async () => {
+    const referenceA = ref(storage, 'public/exp-upload/a');
+    const bytesToUpload = new ArrayBuffer(1024 * 1024);
+    const task = uploadBytesResumable(referenceA, bytesToUpload);
+    const failureDeferred = new Deferred();
+    let hasPaused = false;
+    task.on(
+      'state_changed',
+      snapshot => {
+        if (snapshot.bytesTransferred > 0 && !hasPaused) {
+          task.pause();
+          hasPaused = true;
+        }
+      },
+      () => {
+        failureDeferred.reject('Failed to upload file');
+      }
+    );
+    await Promise.race([
+      failureDeferred.promise,
+      new Promise(resolve => setTimeout(resolve, 4000))
+    ]);
+    task.resume();
+    await task;
+    const bytes = await getBytes(referenceA);
+    expect(bytes).to.deep.eq(bytesToUpload);
+  });
 });

packages/storage/test/unit/connection.ts

@@ -117,7 +117,7 @@ export class TestingConnection implements Connection<string> {
 
   abort(): void {
     this.state = State.START;
-    this.errorCode = ErrorCode.NO_ERROR;
+    this.errorCode = ErrorCode.ABORT;
     this.resolve();
   }
 

packages/storage/test/unit/task.test.ts

@@ -341,7 +341,142 @@ describe('Firebase Storage > Upload Task', () => {
     });
 
     task.pause();
+    return promise;
+  }
+
+  // This is to test to make sure that when you pause in the middle of a request that you do not get an error
+  async function runProgressPauseTest(blob: FbsBlob): Promise<void> {
+    let callbackCount = 0;
+    // Usually the first request is to create the resumable upload and the second is to upload.
+    // Upload requests are not retryable, and this callback is to make sure we pause before the response comes back.
+    function shouldRespondCallback(): boolean {
+      if (callbackCount++ === 1) {
+        task.pause();
+        return false;
+      }
+      return true;
+    }
+    const storageService = storageServiceWithHandler(
+      fakeServerHandler(),
+      shouldRespondCallback
+    );
+    const task = new UploadTask(
+      new Reference(storageService, testLocation),
+      blob
+    );
+
+    // eslint-disable-next-line @typescript-eslint/ban-types
+    let resolve: Function, reject: Function;
+    const promise = new Promise<void>((innerResolve, innerReject) => {
+      resolve = innerResolve;
+      reject = innerReject;
+    });
+
+    // Assert functions throw Errors, which means if we're not in the right stack
+    // the error might not get reported properly. This function wraps existing
+    // assert functions, returning a new function that calls reject with any
+    // caught errors. This should make sure errors are reported properly.
+    function promiseAssertWrapper<T>(func: T): T {
+      function wrapped(..._args: any[]): void {
+        try {
+          (func as any as (...args: any[]) => void).apply(null, _args);
+        } catch (e) {
+          reject(e);
+          pausedStateCompleted.reject(e);
+          // also throw to further unwind the stack
+          throw e;
+        }
+      }
+      return wrapped as any as T;
+    }
+
+    const fixedAssertEquals = promiseAssertWrapper(assert.equal);
+    const fixedAssertFalse = promiseAssertWrapper(assert.isFalse);
+    const fixedAssertTrue = promiseAssertWrapper(assert.isTrue);
+    const fixedAssertFail = promiseAssertWrapper(assert.fail);
+
+    const events: string[] = [];
+    const progress: number[][] = [];
+    // Promise for when we are finally in the pause state
+    const pausedStateCompleted = new Deferred();
+    let complete = 0;
+    // Adds a callback for when the state has changed. The callback resolves the pausedStateCompleted promise
+    // to let our test know when to resume.
+    function addCallbacks(task: UploadTask): void {
+      let lastState: string;
+      task.on(
+        TaskEvent.STATE_CHANGED,
+        snapshot => {
+          fixedAssertEquals(complete, 0);
+
+          const state = snapshot.state;
+          if (lastState !== TaskState.RUNNING && state === TaskState.RUNNING) {
+            events.push('resume');
+          } else if (
+            lastState !== TaskState.PAUSED &&
+            state === TaskState.PAUSED
+          ) {
+            pausedStateCompleted.resolve();
+            events.push('pause');
+          }
+
+          const p = [snapshot.bytesTransferred, snapshot.totalBytes];
+
+          progress.push(p);
+
+          lastState = state;
+        },
+        () => {
+          fixedAssertFail('Failed to upload');
+        },
+        () => {
+          events.push('complete');
+          complete++;
+        }
+      );
+    }
+
+    addCallbacks(task);
 
+    let completeTriggered = false;
+
+    // We should clean this up and just add all callbacks in one function call.
+    // Keeps track of all events that were logged before and asserts on them.
+    task.on(TaskEvent.STATE_CHANGED, undefined, undefined, () => {
+      fixedAssertFalse(completeTriggered);
+      completeTriggered = true;
+
+      fixedAssertEquals(events.length, 4);
+      fixedAssertEquals(events[0], 'resume');
+      fixedAssertEquals(events[1], 'pause');
+      fixedAssertEquals(events[2], 'resume');
+      fixedAssertEquals(events[3], 'complete');
+
+      fixedAssertEquals(complete, 1);
+
+      let increasing = true;
+      let allTotalsTheSame = true;
+      for (let i = 0; i < progress.length - 1; i++) {
+        increasing = increasing && progress[i][0] <= progress[i + 1][0];
+        allTotalsTheSame =
+          allTotalsTheSame && progress[i][1] === progress[i + 1][1];
+      }
+
+      let lastIsAll = false;
+      if (progress.length > 0) {
+        const last = progress[progress.length - 1];
+        lastIsAll = last[0] === last[1];
+      } else {
+        lastIsAll = true;
+      }
+
+      fixedAssertTrue(increasing);
+      fixedAssertTrue(allTotalsTheSame);
+      fixedAssertTrue(lastIsAll);
+      resolve(null);
+    });
+    await pausedStateCompleted.promise;
+    task.resume();
     return promise;
   }
   enum StateType {
@@ -593,6 +728,10 @@ describe('Firebase Storage > Upload Task', () => {
     expect(clock.countTimers()).to.eq(0);
     clock.restore();
   });
+  it('does not error when pausing inflight request', async () => {
+    // Kick off upload
+    await runProgressPauseTest(bigBlob);
+  });
   it('tests if small requests that respond with 500 retry correctly', async () => {
     clock = useFakeTimers();
     // Kick off upload

packages/storage/test/unit/testshared.ts

@@ -205,7 +205,8 @@ export type RequestHandler = (
 ) => Response;
 
 export function storageServiceWithHandler(
-  handler: RequestHandler
+  handler: RequestHandler,
+  shouldResponseCb?: () => boolean
 ): FirebaseStorageImpl {
   function newSend(
     connection: TestingConnection,
@@ -215,11 +216,13 @@ export function storageServiceWithHandler(
     headers?: Headers
   ): void {
     const response = handler(url, method, body, headers);
-    connection.simulateResponse(
-      response.status,
-      response.body,
-      response.headers
-    );
+    if (!shouldResponseCb || shouldResponseCb()) {
+      connection.simulateResponse(
+        response.status,
+        response.body,
+        response.headers
+      );
+    }
   }
 
   injectTestConnection(() => newTestConnection(newSend));