fix(storage): handle early stream destruction in createReadStream

handwritten/storage/src/file.ts

@@ -1682,8 +1682,9 @@ class File extends ServiceObject<File, FileMetadata> {
       md5 = false;
     }
 
+    let shouldIgnoreComplete = false;
     const onComplete = (err: Error | null) => {
-      if (err) {
+      if (err && !shouldIgnoreComplete) {
         // There is an issue with node-fetch 2.x that if the stream errors the underlying socket connection is not closed.
         // This causes a memory leak, so cleanup the sockets manually here by destroying the agent.
         if (request?.agent) {
@@ -1780,12 +1781,42 @@ class File extends ServiceObject<File, FileMetadata> {
         transformStreams.push(zlib.createGunzip());
       }
 
-      pipeline(
-        rawResponseStream as Readable,
-        ...(transformStreams as [Transform]),
-        throughStream,
-        onComplete
-      );
+      const responseStream = rawResponseStream as Readable;
+      const cleanupSourceStream = () => {
+        if (typeof responseStream?.destroy === 'function') {
+          responseStream.destroy();
+        } else if (typeof responseStream?.resume === 'function') {
+          responseStream.resume();
+        }
+        for (const stream of transformStreams) {
+          stream.destroy();
+        }
+      };
+
+      if (
+        throughStream.destroyed ||
+        throughStream.closed ||
+        throughStream.writableEnded
+      ) {
+        cleanupSourceStream();
+        return;
+      }
+
+      try {
+        pipeline(
+          responseStream,
+          ...(transformStreams as [Transform]),
+          throughStream,
+          onComplete
+        );
+      } catch (err) {
+        if ((err as {code?: string})?.code === 'ERR_STREAM_UNABLE_TO_PIPE') {
+          shouldIgnoreComplete = true;
+          cleanupSourceStream();
+          return;
+        }
+        throw err;
+      }
     };
 
     // Authenticate the request, then pipe the remote API request to the stream

handwritten/storage/test/file.ts

@@ -137,6 +137,18 @@ const fakeZlib = {
   },
 };
 
+let pipelineOverride: Function | null;
+const fakeStream = {
+  Readable,
+  PassThrough,
+  Stream,
+  Duplex,
+  Transform,
+  pipeline(...args: Array<{}>) {
+    return (pipelineOverride || (pipeline as Function))(...args);
+  },
+};
+
 // eslint-disable-next-line @typescript-eslint/no-var-requires
 const osCached = require('os');
 const fakeOs = {...osCached};
@@ -223,6 +235,7 @@ describe('File', () => {
       '../src/resumable-upload': fakeResumableUpload,
       os: fakeOs,
       './signer': fakeSigner,
+      stream: fakeStream,
       zlib: fakeZlib,
     }).File;
   });
@@ -270,6 +283,7 @@ describe('File', () => {
     createGunzipOverride = null;
     handleRespOverride = null;
     makeWritableStreamOverride = null;
+    pipelineOverride = null;
     resumableUploadOverride = null;
   });
 
@@ -1542,6 +1556,251 @@ describe('File', () => {
             })
             .resume();
         });
+
+        it('should destroy rawResponseStream and transformStreams and return early if throughStream is already destroyed', done => {
+          const rawResponseStream = new PassThrough();
+          Object.assign(rawResponseStream, {
+            toJSON() {
+              return {
+                headers: {
+                  'content-encoding': 'gzip',
+                },
+              };
+            },
+          });
+          const gunzipStream = new PassThrough();
+          createGunzipOverride = () => gunzipStream;
+          const destroySpy = sinon.spy(rawResponseStream, 'destroy');
+          const gunzipDestroySpy = sinon.spy(gunzipStream, 'destroy');
+          const pipelineSpy = sinon.spy();
+          pipelineOverride = pipelineSpy;
+          const requestStream = new PassThrough();
+
+          const readStream = file.createReadStream({validation: false});
+
+          handleRespOverride = (
+            err: Error,
+            res: {},
+            body: {},
+            callback: Function
+          ) => {
+            readStream.destroy();
+            assert.strictEqual(readStream.destroyed, true);
+            assert.doesNotThrow(() => {
+              callback(null, null, rawResponseStream);
+            });
+            assert.strictEqual(pipelineSpy.called, false);
+            assert.strictEqual(destroySpy.calledOnce, true);
+            assert.strictEqual(rawResponseStream.destroyed, true);
+            assert.strictEqual(gunzipDestroySpy.calledOnce, true);
+            assert.strictEqual(gunzipStream.destroyed, true);
+            done();
+          };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          readStream.resume();
+        });
+
+        it('should destroy rawResponseStream and return early if throughStream is closed', done => {
+          const rawResponseStream = new PassThrough();
+          Object.assign(rawResponseStream, {
+            toJSON() {
+              return {headers: {}};
+            },
+          });
+          const destroySpy = sinon.spy(rawResponseStream, 'destroy');
+          const pipelineSpy = sinon.spy();
+          pipelineOverride = pipelineSpy;
+          const requestStream = new PassThrough();
+
+          const readStream = file.createReadStream({validation: false});
+
+          handleRespOverride = (
+            err: Error,
+            res: {},
+            body: {},
+            callback: Function
+          ) => {
+            Object.defineProperty(readStream, 'closed', {value: true});
+            assert.strictEqual(readStream.destroyed, false);
+            assert.strictEqual(readStream.closed, true);
+            assert.doesNotThrow(() => {
+              callback(null, null, rawResponseStream);
+            });
+            assert.strictEqual(pipelineSpy.called, false);
+            assert.strictEqual(destroySpy.calledOnce, true);
+            assert.strictEqual(rawResponseStream.destroyed, true);
+            done();
+          };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          readStream.on('error', done);
+          readStream.resume();
+        });
+
+        it('should destroy rawResponseStream and return early if throughStream is already writableEnded', done => {
+          const rawResponseStream = new PassThrough();
+          Object.assign(rawResponseStream, {
+            toJSON() {
+              return {headers: {}};
+            },
+          });
+          const destroySpy = sinon.spy(rawResponseStream, 'destroy');
+          const pipelineSpy = sinon.spy();
+          pipelineOverride = pipelineSpy;
+          const requestStream = new PassThrough();
+
+          const readStream = file.createReadStream({validation: false});
+
+          handleRespOverride = (
+            err: Error,
+            res: {},
+            body: {},
+            callback: Function
+          ) => {
+            (readStream as unknown as PassThrough).end();
+            assert.strictEqual(
+              (readStream as unknown as PassThrough).writableEnded,
+              true
+            );
+            assert.doesNotThrow(() => {
+              callback(null, null, rawResponseStream);
+            });
+            assert.strictEqual(pipelineSpy.called, false);
+            assert.strictEqual(destroySpy.calledOnce, true);
+            assert.strictEqual(rawResponseStream.destroyed, true);
+            done();
+          };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          readStream.on('error', done);
+          readStream.resume();
+        });
+
+        it('should catch ERR_STREAM_UNABLE_TO_PIPE, ignore onComplete, and destroy rawResponseStream and transformStreams', done => {
+          const rawResponseStream = new PassThrough();
+          Object.assign(rawResponseStream, {
+            toJSON() {
+              return {
+                headers: {
+                  'content-encoding': 'gzip',
+                },
+              };
+            },
+          });
+          const gunzipStream = new PassThrough();
+          createGunzipOverride = () => gunzipStream;
+          const destroySpy = sinon.spy(rawResponseStream, 'destroy');
+          const gunzipDestroySpy = sinon.spy(gunzipStream, 'destroy');
+          const requestStream = new PassThrough();
+
+          const readStream = file.createReadStream({validation: false});
+
+          pipelineOverride = (...args: unknown[]) => {
+            const onComplete = args[args.length - 1] as (
+              err: Error | null
+            ) => void;
+            const err = new Error(
+              'Cannot pipe to a closed or destroyed stream'
+            ) as Error & {code?: string};
+            err.code = 'ERR_STREAM_UNABLE_TO_PIPE';
+            process.nextTick(() => {
+              onComplete(err);
+            });
+            throw err;
+          };
+
+          handleRespOverride = (
+            err: Error,
+            res: {},
+            body: {},
+            callback: Function
+          ) => {
+            assert.strictEqual(readStream.destroyed, false);
+            assert.doesNotThrow(() => {
+              callback(null, null, rawResponseStream);
+            });
+            assert.strictEqual(destroySpy.calledOnce, true);
+            assert.strictEqual(rawResponseStream.destroyed, true);
+            assert.strictEqual(gunzipDestroySpy.calledOnce, true);
+            assert.strictEqual(gunzipStream.destroyed, true);
+            setImmediate(() => {
+              assert.strictEqual(readStream.destroyed, false);
+              done();
+            });
+          };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          readStream.on('error', done);
+          readStream.resume();
+        });
+
+        it('should rethrow unexpected errors from pipeline', done => {
+          const rawResponseStream = new PassThrough();
+          Object.assign(rawResponseStream, {
+            toJSON() {
+              return {headers: {}};
+            },
+          });
+          const requestStream = new PassThrough();
+          const readStream = file.createReadStream({validation: false});
+
+          const unexpectedErr = new Error(
+            'Unexpected pipeline error'
+          ) as Error & {
+            code?: string;
+          };
+          unexpectedErr.code = 'ERR_INVALID_ARG_TYPE';
+
+          pipelineOverride = () => {
+            throw unexpectedErr;
+          };
+
+          handleRespOverride = (
+            err: Error,
+            res: {},
+            body: {},
+            callback: Function
+          ) => {
+            assert.throws(() => {
+              callback(null, null, rawResponseStream);
+            }, unexpectedErr);
+            done();
+          };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          readStream.resume();
+        });
       });
     });