diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js index 0c6701fa2711..76d12fdbc449 100644 --- a/lib/internal/streams/duplexify.js +++ b/lib/internal/streams/duplexify.js @@ -103,6 +103,8 @@ module.exports = function duplexify(body, name) { const then = value?.then; if (typeof then === 'function') { let d; + // Tracks whether writable final() has started. + let finalized = false; const promise = FunctionPrototypeCall( then, @@ -111,6 +113,12 @@ module.exports = function duplexify(body, name) { if (val != null) { throw new ERR_INVALID_RETURN_VALUE('nully', 'body', val); } + // The async function returned without (fully) consuming the input. + // Destroy the duplex so that pipeline propagates destruction + // upstream. See https://github.com/nodejs/node/issues/55077. + if (!finalized) { + destroyer(d); + } }, (err) => { destroyer(d, err); @@ -123,6 +131,7 @@ module.exports = function duplexify(body, name) { readable: false, write, final(cb) { + finalized = true; final(async () => { try { await promise; diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js index e12599fed17c..a553a90f96e3 100644 --- a/test/parallel/test-stream-duplex-from.js +++ b/test/parallel/test-stream-duplex-from.js @@ -418,3 +418,19 @@ function makeATestWritableStream(writeFunc) { })); r.destroy(expectedErr); } + +// Regression for https://github.com/nodejs/node/issues/55077: +// An AsyncFunction passed to Duplex.from() that returns without consuming its +// input must still allow pipeline() to complete and destroy the upstream. +{ + const r = Readable.from(['foo', 'bar', 'baz']); + pipeline( + r, + Duplex.from(async function() { + // Intentionally do not consume the async iterable input. + }), + common.mustCall(() => { + assert.strictEqual(r.destroyed, true); + }), + ); +}