Skip to content

stream: skip pipeline callback on sync throw - #65128

Open
lazerg wants to merge 1 commit into
nodejs:mainfrom
lazerg:fix/issue-65127-pipeline-sync-throw-callback
Open

stream: skip pipeline callback on sync throw#65128
lazerg wants to merge 1 commit into
nodejs:mainfrom
lazerg:fix/issue-65127-pipeline-sync-throw-callback

Conversation

@lazerg

@lazerg lazerg commented Aug 8, 2026

Copy link
Copy Markdown
Contributor

When pipelineImpl() throws while wiring the streams together, the stages it already wired stay live. As they close they call finish(), finishCount drains to zero and finishImpl() runs with final set, so the callback is scheduled with error still undefined. The caller ends up with the same failure reported twice, once as the exception and once as a successful completion.

The callback is now only scheduled if the wiring loop actually finished. An error that arrives on an already-wired stream after the throw is no longer forwarded to the callback either, which is intentional and matches what already happens when the throw comes before anything was wired. The streams themselves are still left untouched, since ownership is not taken until pipeline() succeeds.

Fixes: #65127

@nodejs-github-bot

Copy link
Copy Markdown
Collaborator

Review requested:

  • @nodejs/streams

@nodejs-github-bot nodejs-github-bot added needs-ci PRs that need a full CI run. stream Issues and PRs related to the stream subsystem. labels Aug 8, 2026
Signed-off-by: Lazizbek Ergashev <lazerg2@gmail.com>
@lazerg
lazerg force-pushed the fix/issue-65127-pipeline-sync-throw-callback branch from 369726d to 358beaa Compare August 8, 2026 06:21
@lazerg lazerg changed the title stream: don't call pipeline callback after a sync throw stream: skip pipeline callback on sync throw Aug 8, 2026
@lazerg

lazerg commented Aug 8, 2026

Copy link
Copy Markdown
Contributor Author

The x86_64-darwin shared-libraries job failed on a test-tick-processor-arguments timeout. That test is unrelated to this change, which only touches stream code.

@codecov

codecov Bot commented Aug 8, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 90.30%. Comparing base (e2d7b34) to head (358beaa).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files
@@            Coverage Diff             @@
##             main   #65128      +/-   ##
==========================================
- Coverage   90.31%   90.30%   -0.01%     
==========================================
  Files         759      759              
  Lines      248290   248297       +7     
  Branches    46859    46854       -5     
==========================================
- Hits       224241   224230      -11     
- Misses      15472    15516      +44     
+ Partials     8577     8551      -26     
Files with missing lines Coverage Δ
lib/internal/streams/pipeline.js 95.19% <100.00%> (+0.07%) ⬆️

... and 32 files with indirect coverage changes

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@jasnell
jasnell requested review from mcollina and ronag August 8, 2026 15:29

@MILLERMARRU MILLERMARRU left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The wired flag correctly stops the erroneous success callback, traced it against finishImpl and it matches the repro in #65127 exactly: wired only flips to true after the whole loop completes without throwing, so a stray finish() firing later from an already-wired stage can't reach process.nextTick(callback, ...) anymore.

One gap though: this only suppresses the callback, it doesn't do anything about the already-wired stages themselves. finishImpl's existing while (destroys.length) { destroys.shift()(error) } still runs when that stray finish() eventually fires, so destruction is still happening, just deferred to whenever the wired-but-orphaned stage naturally reaches its own completion. For a Readable that's already at EOF by the time the throw happens (like the added test), that's basically immediate and you'd never notice. For a stage that's still actively flowing when the later stage throws (a socket, a long-lived generator, anything that hasn't hit EOF yet), nothing here prompts it to stop, since finish() only fires on natural completion, not because pipeline() decided to abort.

Checked this isn't just theoretical, isolated the two approaches (this PR's wired flag vs #65165's proactive destroy in the catch block) against an infinite Readable piped into a Transform, then threw synchronously the way pipeline() does when a later stage is invalid:

const readable = makeInfiniteReadable(() => destroyed.push('readable'))
readable.pipe(transform)
try {
  throw new Error('bad stage')
} catch (err) {
  // this PR's approach: nothing here
}
// 200ms later: destroyed === [] for this approach,
// vs destroyed === ['readable', 'transform'] when the catch block
// proactively calls destroy() the way #65165 does

Both readable and transform are still alive 200ms after the throw with just the wired guard, since nothing ever tells them to stop.

Separately, #65165 is open against the same issue with a different approach (try/catch around the whole wiring loop, explicit destroys/ac.abort() in the catch block), worth reconciling with that one before this merges so you don't end up with two PRs fixing the same bug from different angles.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

needs-ci PRs that need a full CI run. stream Issues and PRs related to the stream subsystem.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

3 participants