From a72af533ab5fbfbffb4111e50a6cc6012ebd84c9 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 24 Jun 2023 09:58:03 +0200
Subject: [PATCH] stream: fix premature pipeline end

Fixes: https://github.com/nodejs/node/issues/48406
PR-URL: https://github.com/nodejs/node/pull/48435
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Marco Ippolito <marcoippolito54@gmail.com>
Reviewed-By: Debadree Chatterjee <debadree333@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
---
 lib/internal/streams/pipeline.js      |  4 ++--
 test/parallel/test-stream-pipeline.js | 28 +++++++++++++++++++++++++++
 2 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index 062bdc192d1310..fb2cd90a2678ea 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -38,7 +38,7 @@ const {
   isTransformStream,
   isWebStream,
   isReadableStream,
-  isReadableEnded,
+  isReadableFinished,
 } = require('internal/streams/utils');
 const { AbortController } = require('internal/abort_controller');
 
@@ -424,7 +424,7 @@ function pipe(src, dst, finish, { end }) {
       dst.end();
     }
 
-    if (isReadableEnded(src)) { // End the destination if the source has already ended.
+    if (isReadableFinished(src)) { // End the destination if the source has already ended.
       process.nextTick(endFn);
     } else {
       src.once('end', endFn);
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index e9f6a2fdf711d3..915a035264c7a7 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -1634,3 +1634,31 @@ const tsp = require('timers/promises');
     assert.strictEqual(writable.closed, false);
   }));
 }
+
+{
+  const r = new Readable();
+  for (let i = 0; i < 4000; i++) {
+    r.push('asdfdagljanfgkaljdfn');
+  }
+  r.push(null);
+
+  let ended = false;
+  r.on('end', () => {
+    ended = true;
+  });
+
+  const w = new Writable({
+    write(chunk, enc, cb) {
+      cb(null);
+    },
+    final: common.mustCall((cb) => {
+      assert.strictEqual(ended, true);
+      cb(null);
+    })
+  });
+
+  pipeline(r, w, common.mustCall((err) => {
+    assert.strictEqual(err, undefined);
+  }));
+
+}