fix(storage): destroy local read stream on upload write failure to prevent resource leaks

handwritten/storage/src/bucket.ts

@@ -4501,17 +4501,23 @@ class Bucket extends ServiceObject<Bucket, BucketMetadata> {
             ) {
               newFile.storage.retryOptions.autoRetry = false;
             }
+            const readStream = fs.createReadStream(pathString);
             const writable = newFile.createWriteStream(options);
             if (options.onUploadProgress) {
               writable.on('progress', options.onUploadProgress);
             }
-            fs.createReadStream(pathString)
-              .on('error', bail)
+            readStream
+              .on('error', err => {
+                readStream.destroy();
+                writable.destroy();
+                bail(err);
+              })
               .pipe(writable)
               .on('error', err => {
+                readStream.destroy();
                 if (
                   this.storage.retryOptions.autoRetry &&
-                  this.storage.retryOptions.retryableErrorFn!(err)
+                  this.storage.retryOptions.retryableErrorFn!(err as any)
                 ) {
                   return reject(err);
                 } else {

handwritten/storage/test/bucket.ts

@@ -100,11 +100,18 @@ class FakeNotification {
 }
 
 let fsStatOverride: Function | null;
+let fsCreateReadStreamOverride: Function | null;
 const fakeFs = {
   ...fs,
   stat: (filePath: string, callback: Function) => {
     return (fsStatOverride || fs.stat)(filePath, callback);
   },
+  createReadStream: (filePath: string, options?: any) => {
+    return (fsCreateReadStreamOverride || fs.createReadStream)(
+      filePath,
+      options
+    );
+  },
 };
 
 let pLimitOverride: Function | null;
@@ -234,6 +241,7 @@ describe('Bucket', () => {
 
   beforeEach(() => {
     fsStatOverride = null;
+    fsCreateReadStreamOverride = null;
     pLimitOverride = null;
     bucket = new Bucket(STORAGE, BUCKET_NAME);
   });
@@ -3231,6 +3239,44 @@ describe('Bucket', () => {
       });
     });
 
+    it('should destroy the local read stream if write stream fails', done => {
+      const fakeFile = new FakeFile(bucket, 'file-name');
+      const options = {destination: fakeFile, resumable: false};
+      const originalCreateReadStream = fs.createReadStream;
+      let readStream: fs.ReadStream;
+      fsCreateReadStreamOverride = (path: string, opts: any) => {
+        readStream = originalCreateReadStream(path, opts);
+        return readStream;
+      };
+
+      fakeFile.createWriteStream = (options_: CreateWriteStreamOptions) => {
+        const ws = new stream.Writable({
+          write(chunk, encoding, callback) {
+            callback(new Error('write error'));
+          },
+        });
+        return ws;
+      };
+
+      const textfilepath = path.join(
+        getDirName(),
+        '../../../test/testdata/textfile.txt'
+      );
+
+      bucket.upload(textfilepath, options, (err: Error) => {
+        try {
+          assert.strictEqual(err.message, 'write error');
+          assert.ok(readStream);
+          assert.ok(readStream.destroyed);
+          done();
+        } catch (e) {
+          done(e);
+        } finally {
+          fsCreateReadStreamOverride = null;
+        }
+      });
+    });
+
     it('should allow overriding content type', done => {
       const fakeFile = new FakeFile(bucket, 'file-name');
       const metadata = {contentType: 'made-up-content-type'};