`stream.pipeline()` can both throw synchronously and invoke the callback reporting success

Beginne im Verdrahtungs-Loop von stream pipelineImpl() und konzentriere dich auf finishCount und den im Bericht beschriebenen synchronen-throw-Pfad. Reproduziere die Fälle A und B aus dem Issue und überprüfe anschließend, dass ein synchroner throw den Callback später nicht mit Erfolg aufruft und dass der Callback nicht zweimal aufgerufen wird.

Version

v24.11.1, and main (aed4eaf89dd)

Platform

Reproduced on Windows 11 x64; the code path is platform independent.

Subsystem

stream

What steps will reproduce the bug?

When pipeline() throws synchronously while wiring the streams together, it can also invoke the callback, and the callback reports success. The caller therefore receives the same operation reported twice, once as an exception and once as a successful completion.

import { pipeline, Readable, Transform, PassThrough } from 'node:stream';

function run(label, streams) {
  return new Promise((resolve) => {
    let threw = null;
    const calls = [];
    try {
      pipeline(...streams, (err) => calls.push(err ? err.code : 'NO ERROR'));
    } catch (err) {
      threw = err.code;
    }
    setTimeout(() => {
      console.log(`${label}`);
      console.log(`   threw            : ${threw}`);
      console.log(`   callback invoked : ${calls.length} time(s) ${calls.length ? JSON.stringify(calls) : ''}`);
      resolve();
    }, 150);
  });
}

// A node stream in the middle wires pipe(), which increments finishCount
// before the later stage throws.
await run('A. readable -> transform -> bad fn', [
  Readable.from(['a']),
  new Transform({ transform(c, e, cb) { cb(null, c); } }),
  () => 42,
]);

// Control: the throw happens before anything increments finishCount.
await run('B. readable -> bad fn -> passthrough  (control)', [
  Readable.from(['a']),
  () => 42,
  new PassThrough(),
]);

How often does it reproduce? Is there a required condition?

Every time, for case A. The condition is that finishCount has already been incremented by an earlier stage before a later stage throws, which happens whenever a node stream is wired before the throwing stage.

What is the expected behavior? Why is that the expected behavior?

A single failure should be reported once. If pipeline() throws synchronously then the callback should not also run, and it certainly should not run with no error, since the pipeline did not succeed.

Case B shows the intended behaviour: the callback is not invoked when pipeline() throws.

What do you see instead?

node v24.11.1

A. readable -> transform -> bad fn
   threw            : ERR_INVALID_RETURN_VALUE
   callback invoked : 1 time(s) ["NO ERROR"]

B. readable -> bad fn -> passthrough  (control)
   threw            : ERR_INVALID_RETURN_VALUE
   callback invoked : 0 time(s)

In case A the caller gets an ERR_INVALID_RETURN_VALUE exception and, shortly afterwards, a callback saying the pipeline finished with no error.

Additional information

The wiring loop in pipelineImpl() increments finishCount as it wires each stage, for example finishCount += 2 before pipe() in the node stream branch. If a later stage throws, those already-wired stages remain live. When they subsequently close they call finish(), which decrements finishCount; on reaching zero finishImpl() runs with final true and schedules process.nextTick(callback, error, value).

Because the streams were never destroyed by pipeline() on this path, error is still undefined at that point, so the callback reports success.

In case B the throw happens before any stage incremented finishCount, so nothing later reaches zero and the callback is correctly never scheduled. That difference is what makes the behaviour inconsistent rather than simply undesirable.

This was noticed while working on #65064, which touches the same synchronous-throw path but deliberately does not change this behaviour.