fix: File#save iterable fixes (#2293)

* fix: File#save iterable fixes

* fix: if logic

* test: remove unneccesary test

* fix: return on save-stream-bail

* refactor: remove non-reachable case

Author: d-goog

src/file.ts

@@ -463,7 +463,6 @@ export enum FileExceptionMessages {
   UPLOAD_MISMATCH = `The uploaded data did not match the data from the server.
     As a precaution, the file has been deleted.
     To be sure the content is the same, you should try uploading the file again.`,
-  STREAM_NOT_READABLE = 'Stream must be readable.',
 }
 
 /**
@@ -3644,20 +3643,6 @@ class File extends ServiceObject<File, FileMetadata> {
     }
     const returnValue = retry(
       async (bail: (err: Error) => void) => {
-        if (data instanceof Readable) {
-          // Make sure any pending async readable operations are finished before
-          // attempting to check if the stream is readable.
-          await new Promise(resolve => setImmediate(resolve));
-
-          if (!data.readable || data.destroyed) {
-            // Calling pipeline() with a non-readable stream will result in the
-            // callback being called without an error, and no piping taking
-            // place. In that case, file.save() would appear to succeed, but
-            // nothing would be uploaded.
-            return bail(new Error(FileExceptionMessages.STREAM_NOT_READABLE));
-          }
-        }
-
         return new Promise<void>((resolve, reject) => {
           if (maxRetries === 0) {
             this.storage.retryOptions.autoRetry = false;
@@ -3670,13 +3655,13 @@ class File extends ServiceObject<File, FileMetadata> {
 
           const handleError = (err: Error) => {
             if (
-              !this.storage.retryOptions.autoRetry ||
-              !this.storage.retryOptions.retryableErrorFn!(err)
+              this.storage.retryOptions.autoRetry &&
+              this.storage.retryOptions.retryableErrorFn!(err)
             ) {
-              bail(err);
+              return reject(err);
             }
 
-            reject(err);
+            return bail(err);
           };
 
           if (typeof data === 'string' || Buffer.isBuffer(data)) {
@@ -3687,17 +3672,10 @@ class File extends ServiceObject<File, FileMetadata> {
           } else {
             pipeline(data, writable, err => {
               if (err) {
-                // If data is not a valid PipelineSource, then pipeline will
-                // fail without destroying the writable stream. If data is a
-                // PipelineSource that yields invalid chunks (e.g. a stream in
-                // object mode or an iterable that does not yield Buffers or
-                // strings), then pipeline will destroy the writable stream.
-                if (!writable.destroyed) writable.destroy();
-
                 if (typeof data !== 'function') {
                   // Only PipelineSourceFunction can be retried. Async-iterables
                   // and Readable streams can only be consumed once.
-                  bail(err);
+                  return bail(err);
                 }
 
                 handleError(err);

test/file.ts

@@ -4365,31 +4365,6 @@ describe('File', () => {
         }
       });
 
-      it('Destroyed Readable upload should throw', async () => {
-        const options = {resumable: false};
-
-        file.createWriteStream = () => {
-          throw new Error('unreachable');
-        };
-        try {
-          const readable = new Readable({
-            read() {
-              this.push(DATA);
-              this.push(null);
-            },
-          });
-
-          readable.destroy();
-
-          await file.save(readable, options);
-        } catch (e) {
-          assert.strictEqual(
-            (e as Error).message,
-            FileExceptionMessages.STREAM_NOT_READABLE
-          );
-        }
-      });
-
       it('should save a generator with no error', done => {
         const options = {resumable: false};
         file.createWriteStream = () => {
@@ -4439,33 +4414,6 @@ describe('File', () => {
         });
       });
 
-      it('should error on invalid async iterator data', done => {
-        const options = {resumable: false};
-        file.createWriteStream = () => {
-          const writeStream = new PassThrough();
-          let errorCalled = false;
-          writeStream.on('error', () => {
-            errorCalled = true;
-          });
-          writeStream.on('finish', () => {
-            assert.ok(errorCalled);
-          });
-          return writeStream;
-        };
-
-        const generator = async function* () {
-          yield {thisIsNot: 'a buffer or a string'};
-        };
-
-        file.save(generator(), options, (err: Error) => {
-          assert.strictEqual(
-            err.message,
-            'The "chunk" argument must be of type string or an instance of Buffer or Uint8Array. Received an instance of Object'
-          );
-          done();
-        });
-      });
-
       it('buffer upload should retry on first failure', async () => {
         const options = {
           resumable: false,