stream: fix pipeline callback not called on ended stream

Fixes: nodejs#46595
PR-URL: nodejs#46600
Reviewed-By: Robert Nagy <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Colin Ihrig <[email protected]>

lib/internal/streams/pipeline.js

@@ -38,6 +38,7 @@ const {
   isTransformStream,
   isWebStream,
   isReadableStream,
+  isReadableEnded,
 } = require('internal/streams/utils');
 const { AbortController } = require('internal/abort_controller');
 
@@ -417,10 +418,17 @@ function pipe(src, dst, finish, { end }) {
     // Compat. Before node v10.12.0 stdio used to throw an error so
     // pipe() did/does not end() stdio destinations.
     // Now they allow it but "secretly" don't close the underlying fd.
-    src.once('end', () => {
+
+    function endFn() {
       ended = true;
       dst.end();
-    });
+    }
+
+    if (isReadableEnded(src)) { // End the destination if the source has already ended.
+      process.nextTick(endFn);
+    } else {
+      src.once('end', endFn);
+    }
   } else {
     finish();
   }

test/parallel/test-stream-pipeline.js

@@ -1591,3 +1591,28 @@ const tsp = require('timers/promises');
     assert.strictEqual(writable.endCount, 1);
   }));
 }
+
+{
+  const readable = new Readable({
+    read() {}
+  });
+  readable.on('end', common.mustCall(() => {
+    pipeline(readable, new PassThrough(), common.mustSucceed());
+  }));
+  readable.push(null);
+  readable.read();
+}
+
+{
+  const dup = new Duplex({
+    read() {},
+    write(chunk, enc, cb) {
+      cb();
+    }
+  });
+  dup.on('end', common.mustCall(() => {
+    pipeline(dup, new PassThrough(), common.mustSucceed());
+  }));
+  dup.push(null);
+  dup.read();
+}

test/parallel/test-webstreams-pipeline.js

@@ -2,7 +2,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { Readable, Writable, Transform, pipeline } = require('stream');
+const { Readable, Writable, Transform, pipeline, PassThrough } = require('stream');
 const { pipeline: pipelinePromise } = require('stream/promises');
 const { ReadableStream, WritableStream, TransformStream } = require('stream/web');
 const http = require('http');
@@ -410,3 +410,13 @@ const http = require('http');
   }
   c.close();
 }
+
+{
+  const rs = new ReadableStream({
+    start(controller) {
+      controller.close();
+    }
+  });
+
+  pipeline(rs, new PassThrough(), common.mustSucceed());
+}