stream: prevent pipeline() callback after sync throw

[pacocartones]

Fixes: #65127

Summary

When pipeline() throws synchronously while wiring streams together — for
example, when a middle stage returns an invalid value (ERR_INVALID_RETURN_VALUE)
— the stages already wired have already incremented finishCount. When those
stages complete afterwards, their finish callbacks invoke the completion
callback with no error, double-reporting the failure that the caller
already received as a synchronous exception.

This change tracks synchronous throws while wiring, destroys the stages wired
so far, and skips the completion callback in that case.

Repro (issue case A)

const { pipeline, Readable, Transform } = require('node:stream');
const r = Readable.from(['a']);
const t = new Transform({ transform(c, e, cb) { cb(null, c); } });
try {
  pipeline(r, t, () => 42, (err) => console.log('callback:', err ?? 'NO ERROR'));
} catch (err) {
  console.log('threw:', err.code);
}

Before: threw: ERR_INVALID_RETURN_VALUE and callback: NO ERROR (double report).
After: threw: ERR_INVALID_RETURN_VALUE only; the callback is never invoked.

Verification

• Regression test added to test/parallel/test-stream-pipeline.js:
  • fails on the unfixed code (common.mustNotCall() gets invoked with success),
  • passes with the fix.
• Behavior of normal success, stream-error propagation, generator sources and
  abort paths verified unchanged against the pristine implementation.

Checklist

• tests and/or benchmarks are included
• commit message follows commit guidelines

[nodejs-github-bot]

Review requested:

• @nodejs/streams

[pacocartones]

Thanks for the careful trace on both PRs — you're right that these need reconciling, and having looked into it properly, I think the answer is that this one should be closed in favour of #65128.

Here's what I found after digging into the history, because it changes the picture:

There's a third PR against this code path, #65064 by @shani-singh1 (opened 2026-08-06, before both of these). Its original version did essentially what this PR does: wrap the wiring loop in try/catch, drain destroys, dispose the abort listener, ac.abort(), rethrow. @ronag requested changes on it with:

I would say that ownership is not taken until pipeline succeeds...

and CI then confirmed it concretely. @shani-singh1's follow-up:

That line is assert.strictEqual(s.destroyed, false) inside the ERR_INVALID_RETURN_VALUE block, so "ownership is not taken until pipeline succeeds" is already asserted behaviour and my patch was contradicting it.

So the proactive-destroy behaviour you liked here is the part that's actually wrong. There are four blocks in test/parallel/test-stream-pipeline.js (~L871-916) that assert it explicitly, e.g. L890:

const s = new PassThrough();
assert.throws(() => {
  pipeline(s, function(source) {
  }, s, () => {});
}, (err) => {
  assert.strictEqual(err.code, 'ERR_INVALID_RETURN_VALUE');
  assert.strictEqual(s.destroyed, false);   // L890
  return true;
});

At the point that throws, s is already in destroys, and my catch block runs destroys.shift()(err) before rethrowing. destroy() sets .destroyed synchronously, so s.destroyed is true by the time the validator runs. I checked on v24.14.1 that main gives false there today, so this PR would turn three of those four assertions red. CI hasn't caught it because every workflow run on this PR is still action_required — nothing has actually run.

Two more things worth putting on the record:

• The abort-listener half of my catch block (disposable?.[SymbolDispose]()) is legitimate, but it's not mine to add: that's exactly what stream: clean up when pipeline throws synchronously #65064 was narrowed down to, and it's already approved.
• Where the two differ on callback suppression, lazerg's is the better shape. My syncThrow guard returns at the top of finishImpl, which also skips lastStreamCleanup.forEach(...) and leaves the onError listeners pipeline() attached in place. The wired gate sits directly on process.nextTick(callback, error, value), so that cleanup still runs.

So the split I'd propose is: #65128 fixes #65127 (the double report), #65064 fixes #65063 (the abort-listener leak), and this PR gets closed. The only thing to watch is that they'll conflict textually at the end of the wiring loop — wired = true will need to move after the try/catch, depending on merge order.

@nodejs/streams — could someone confirm that split? And #65128 is still carrying needs-ci with no Jenkins run; a CI start there would be more useful than one here.

@lazerg sorry for the duplicate work — I opened this without spotting #65064 and the ownership discussion on it, which would have saved us both some time.

Closing this one.