fix(storage): handle early read stream destruction

File#createReadStream emits the response event before attaching the raw response pipeline. A consumer can destroy the returned stream from that event, leaving pipeline() to receive a destroyed destination and throw ERR_STREAM_UNABLE_TO_PIPE synchronously.

Check the public stream after the response arrives and discard the raw response body when the consumer has already closed it. Reuse the existing request-agent cleanup so the abandoned response does not retain its underlying socket resources.

Add a regression test using the real stream event ordering. The test destroys the returned stream from its response handler and verifies that the delayed raw response and request agent are cleaned up.

This adapts the core approach from PR #7604 by Michael Latman, which was closed before merge.

Co-authored-by: Michael Latman <mlatman@gmail.com>

Author: rockwotj

handwritten/storage/src/file.ts

@@ -1574,6 +1574,17 @@ class File extends ServiceObject<File, FileMetadata> {
 
     const shouldRunValidation = !rangeRequest && (crc32c || md5);
 
+    const cleanupRequest = () => {
+      if (request?.agent) {
+        request.agent.destroy();
+      }
+    };
+
+    const cleanupRawResponse = (rawResponseStream: Readable) => {
+      rawResponseStream.destroy();
+      cleanupRequest();
+    };
+
     if (rangeRequest) {
       if (
         typeof options.validation === 'string' ||
@@ -1590,9 +1601,7 @@ class File extends ServiceObject<File, FileMetadata> {
       if (err) {
         // 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) {
-          request.agent.destroy();
-        }
+        cleanupRequest();
         throughStream.destroy(err);
       }
     };
@@ -1622,6 +1631,11 @@ class File extends ServiceObject<File, FileMetadata> {
       }
 
       request = (rawResponseStream as r.Response).request;
+      if (throughStream.destroyed) {
+        cleanupRawResponse(rawResponseStream as Readable);
+        return;
+      }
+
       const headers = (rawResponseStream as ResponseBody).toJSON().headers;
       const isCompressed = headers['content-encoding'] === 'gzip';
       const hashes: {crc32c?: string; md5?: string} = {};

handwritten/storage/test/file.ts

@@ -1165,6 +1165,49 @@ describe('File', () => {
         file.createReadStream().resume();
       });
 
+      it('should clean up if the returned stream is destroyed before the response is piped', done => {
+        const rawResponseStream = new PassThrough();
+        const agentDestroy = sinon.spy();
+        Object.assign(rawResponseStream, {
+          request: {
+            agent: {
+              destroy: agentDestroy,
+            },
+          },
+          toJSON() {
+            return {headers: {}};
+          },
+        });
+
+        handleRespOverride = (
+          _err: Error,
+          _res: {},
+          _body: {},
+          callback: Function
+        ) => {
+          callback(null, null, rawResponseStream);
+        };
+
+        file.requestStream = () => {
+          const requestStream = new PassThrough();
+          setImmediate(() => {
+            requestStream.emit('response', rawResponseStream);
+          });
+          return requestStream;
+        };
+
+        const readStream = file.createReadStream({validation: false});
+        readStream.once('response', () => {
+          readStream.destroy();
+        });
+        readStream.once('close', () => {
+          assert.strictEqual(rawResponseStream.destroyed, true);
+          assert.strictEqual(agentDestroy.calledOnce, true);
+          done();
+        });
+        readStream.resume();
+      });
+
       describe('errors', () => {
         const ERROR = new Error('Error.');