stream: fix pipeline callback not called on ended stream

Fixes: nodejs#46595

lib/internal/streams/pipeline.js

@@ -38,6 +38,7 @@ const {
  isTransformStream,
  isWebStream,
  isReadableStream,
+ isReadableFinished,
 } = 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();
- });
+ }
+
+ src.once('end', endFn);
+
+ if (isReadableFinished(src)) { // End the destination if the source has already ended.
+ process.nextTick(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());
+}