From ebf0fde601bd67f207387199f0e43e9ec85a3c92 Mon Sep 17 00:00:00 2001
From: Aman Chadha <79802170+ac-mmi@users.noreply.github.com>
Date: Thu, 10 Sep 2026 19:51:23 +0530
Subject: [PATCH] stream: destroy Duplex.from async function on early return

When an AsyncFunction passed to Duplex.from() resolves without
consuming its input, tear down the duplex so pipeline() can finish
and destroy the upstream readable.

Fixes: https://github.com/nodejs/node/issues/55077
Assisted-by: Cursor
Signed-off-by: Aman Chadha <79802170+ac-mmi@users.noreply.github.com>
---
 lib/internal/streams/duplexify.js        |  9 +++++++++
 test/parallel/test-stream-duplex-from.js | 16 ++++++++++++++++
 2 files changed, 25 insertions(+)

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);
+    }),
+  );
+}