fix(storage): handle upload errors before pipeline setup

createWriteStream starts the simple or resumable upload before installing the pipeline that normally owns fileWriteStream errors. If startup fails synchronously, fileWriteStream and the public write stream can be destroyed before pipeline() runs. Node then throws ERR_STREAM_UNABLE_TO_PIPE while attaching invalid streams, and the original upload error can escape uncaught.

Install a temporary error listener before upload startup and forward any early failure through the existing pipeline callback. If startup has already torn down any stream, skip pipeline setup and destroy the remaining internal streams. Keep the temporary listener until the upload stream closes so a queued terminal error is still handled, then remove it to avoid retaining the callback.

Add coverage for a consumer destroying the returned stream during upload startup. The test verifies that the internal upload stream is destroyed and its temporary error listener is removed. Existing synchronous upload-error and pipeline-failure coverage continues to pass.

Author: rockwotj

handwritten/storage/src/file.ts

@@ -2192,12 +2192,36 @@ class File extends ServiceObject<File, FileMetadata> {
     });
 
     writeStream.once('writing', () => {
+      const onPrePipelineError = (error: Error) => {
+        pipelineCallback(error);
+      };
+      fileWriteStream.once('error', onPrePipelineError);
+
       if (options.resumable === false) {
         this.startSimpleUpload_(fileWriteStream, options);
       } else {
         this.startResumableUpload_(fileWriteStream, options);
       }
 
+      if (
+        fileWriteStream.destroyed ||
+        writeStream.destroyed ||
+        emitStream.destroyed
+      ) {
+        // Destroying an upload stream can queue its terminal error before
+        // close, so keep the temporary listener until teardown completes.
+        fileWriteStream.once('close', () => {
+          fileWriteStream.removeListener('error', onPrePipelineError);
+        });
+        if (!fileWriteStream.destroyed) {
+          fileWriteStream.destroy();
+        }
+        emitStream.destroy();
+        return;
+      }
+
+      fileWriteStream.removeListener('error', onPrePipelineError);
+
       // remove temporary noop listener as we now create a pipeline that handles the errors
       emitStream.removeListener('error', noop);
 

handwritten/storage/test/file.ts

@@ -2149,6 +2149,27 @@ describe('File', () => {
       writable.write('data');
     });
 
+    it('should clean up if the returned stream is destroyed during upload startup', done => {
+      const writable = file.createWriteStream();
+      let fileWriteStream: duplexify.Duplexify | undefined;
+
+      file.startResumableUpload_ = (stream: duplexify.Duplexify) => {
+        fileWriteStream = stream;
+        writable.destroy();
+      };
+
+      writable.on('close', () => {
+        setImmediate(() => {
+          assert(fileWriteStream);
+          assert.strictEqual(fileWriteStream.destroyed, true);
+          assert.strictEqual(fileWriteStream.listenerCount('error'), 0);
+          done();
+        });
+      });
+
+      writable.write('data');
+    });
+
     it('should alias contentType to metadata object', done => {
       const contentType = 'text/html';
       const writable = file.createWriteStream({contentType});