From cf8840d5e4aa9cd0945a42da4ec4851fe3effcfc Mon Sep 17 00:00:00 2001 From: Thiyagu K Date: Fri, 9 Oct 2026 06:58:12 +0000 Subject: [PATCH 1/3] fix: handle stream pipeline errors and premature destruction by cleaning up rawResponseStream in file.createReadStream --- handwritten/storage/src/file.ts | 38 +++++++++--- handwritten/storage/test/file.ts | 101 +++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 7 deletions(-) diff --git a/handwritten/storage/src/file.ts b/handwritten/storage/src/file.ts index 977e3df4a3ca..afd7b152d1e1 100644 --- a/handwritten/storage/src/file.ts +++ b/handwritten/storage/src/file.ts @@ -1682,8 +1682,9 @@ class File extends ServiceObject { 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,35 @@ class File extends ServiceObject { 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(); + } + }; + + if (throughStream.destroyed) { + cleanupSourceStream(); + return; + } + + try { + pipeline( + responseStream, + ...(transformStreams as [Transform]), + throughStream, + onComplete + ); + } catch (err) { + if ((err as RequestError)?.code === 'ERR_STREAM_UNABLE_TO_PIPE') { + shouldIgnoreComplete = true; + cleanupSourceStream(); + return; + } + throw err; + } }; // Authenticate the request, then pipe the remote API request to the stream diff --git a/handwritten/storage/test/file.ts b/handwritten/storage/test/file.ts index 938e8e317bba..9d037c48337b 100644 --- a/handwritten/storage/test/file.ts +++ b/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,93 @@ describe('File', () => { }) .resume(); }); + + it('should destroy rawResponseStream and return early if throughStream is already destroyed', 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.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); + done(); + }; + + file.requestStream = () => { + setImmediate(() => { + requestStream.emit('response', rawResponseStream); + }); + return requestStream; + }; + + readStream.resume(); + }); + + it('should catch ERR_STREAM_UNABLE_TO_PIPE and destroy rawResponseStream', done => { + const rawResponseStream = new PassThrough(); + Object.assign(rawResponseStream, { + toJSON() { + return {headers: {}}; + }, + }); + const destroySpy = sinon.spy(rawResponseStream, 'destroy'); + const requestStream = new PassThrough(); + + const readStream = file.createReadStream({validation: false}); + + pipelineOverride = () => { + const err = new Error( + 'Cannot pipe to a closed or destroyed stream' + ) as Error & {code?: string}; + err.code = 'ERR_STREAM_UNABLE_TO_PIPE'; + 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); + done(); + }; + + file.requestStream = () => { + setImmediate(() => { + requestStream.emit('response', rawResponseStream); + }); + return requestStream; + }; + + readStream.on('error', done); + readStream.resume(); + }); }); }); From 8b3b4e259be7369bd8244b45483e1eecf866bacd Mon Sep 17 00:00:00 2001 From: Thiyagu K Date: Fri, 9 Oct 2026 07:19:47 +0000 Subject: [PATCH 2/3] fix: update type assertion for ERR_STREAM_UNABLE_TO_PIPE check in file stream handling --- handwritten/storage/src/file.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/handwritten/storage/src/file.ts b/handwritten/storage/src/file.ts index afd7b152d1e1..cc5d1777eeff 100644 --- a/handwritten/storage/src/file.ts +++ b/handwritten/storage/src/file.ts @@ -1803,7 +1803,7 @@ class File extends ServiceObject { onComplete ); } catch (err) { - if ((err as RequestError)?.code === 'ERR_STREAM_UNABLE_TO_PIPE') { + if ((err as {code?: string})?.code === 'ERR_STREAM_UNABLE_TO_PIPE') { shouldIgnoreComplete = true; cleanupSourceStream(); return; From 8e6c57e7572342a3b251079a01bd79790c426269 Mon Sep 17 00:00:00 2001 From: Thiyagu K Date: Tue, 20 Oct 2026 09:07:23 +0000 Subject: [PATCH 3/3] fix: destroy transform streams and handle closed or writable ended states during read stream cleanup --- handwritten/storage/src/file.ts | 9 +- handwritten/storage/test/file.ts | 168 ++++++++++++++++++++++++++++++- 2 files changed, 171 insertions(+), 6 deletions(-) diff --git a/handwritten/storage/src/file.ts b/handwritten/storage/src/file.ts index cc5d1777eeff..916dbf11cf00 100644 --- a/handwritten/storage/src/file.ts +++ b/handwritten/storage/src/file.ts @@ -1788,9 +1788,16 @@ class File extends ServiceObject { } else if (typeof responseStream?.resume === 'function') { responseStream.resume(); } + for (const stream of transformStreams) { + stream.destroy(); + } }; - if (throughStream.destroyed) { + if ( + throughStream.destroyed || + throughStream.closed || + throughStream.writableEnded + ) { cleanupSourceStream(); return; } diff --git a/handwritten/storage/test/file.ts b/handwritten/storage/test/file.ts index 9d037c48337b..a7e2846ce69a 100644 --- a/handwritten/storage/test/file.ts +++ b/handwritten/storage/test/file.ts @@ -1557,14 +1557,21 @@ describe('File', () => { .resume(); }); - it('should destroy rawResponseStream and return early if throughStream is already destroyed', done => { + 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: {}}; + 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(); @@ -1585,6 +1592,8 @@ describe('File', () => { 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(); }; @@ -1598,7 +1607,7 @@ describe('File', () => { readStream.resume(); }); - it('should catch ERR_STREAM_UNABLE_TO_PIPE and destroy rawResponseStream', done => { + it('should destroy rawResponseStream and return early if throughStream is closed', done => { const rawResponseStream = new PassThrough(); Object.assign(rawResponseStream, { toJSON() { @@ -1606,15 +1615,116 @@ describe('File', () => { }, }); const destroySpy = sinon.spy(rawResponseStream, 'destroy'); + const pipelineSpy = sinon.spy(); + pipelineOverride = pipelineSpy; const requestStream = new PassThrough(); const readStream = file.createReadStream({validation: false}); - pipelineOverride = () => { + 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; }; @@ -1630,7 +1740,12 @@ describe('File', () => { }); assert.strictEqual(destroySpy.calledOnce, true); assert.strictEqual(rawResponseStream.destroyed, true); - done(); + assert.strictEqual(gunzipDestroySpy.calledOnce, true); + assert.strictEqual(gunzipStream.destroyed, true); + setImmediate(() => { + assert.strictEqual(readStream.destroyed, false); + done(); + }); }; file.requestStream = () => { @@ -1643,6 +1758,49 @@ describe('File', () => { 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(); + }); }); });