stream: skip pipeline callback on sync throw - #65128
Conversation
|
Review requested:
|
Signed-off-by: Lazizbek Ergashev <lazerg2@gmail.com>
369726d to
358beaa
Compare
|
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 Report✅ All modified and coverable lines are covered by tests. 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
🚀 New features to boost your workflow:
|
MILLERMARRU
left a comment
There was a problem hiding this comment.
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 doesBoth 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.
When
pipelineImpl()throws while wiring the streams together, the stages it already wired stay live. As they close they callfinish(),finishCountdrains to zero andfinishImpl()runs withfinalset, so the callback is scheduled witherrorstill 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