Apply backpressure when piping a ReadableStream into a subprocess - #33399
Apply backpressure when piping a ReadableStream into a subprocess#33399robobun wants to merge 5 commits into
Conversation
|
Updated 2:05 AM PT - Jul 7th, 2026
❌ @robobun, your commit 2d34e88 has some failures in 🧪 To try this PR locally: bunx bun-pr 33399That installs a local version of the PR into your bun-33399 --bun |
|
Found 3 issues this PR may fix:
🤖 Generated with Claude Code |
WalkthroughThis PR adds backed-up-state tracking to streaming writers, uses it to map FileSink writes to ChangesStream backpressure signaling and pump fixes
Possibly related PRs
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
|
I checked all three against a build of this branch before adding any #18315 — no. The repro still fails on this branch: let cancelled = false;
const s = new ReadableStream({ type: "direct", cancel() { cancelled = true; } });
await s.cancel();
console.log(cancelled); // still falseDifferent path. That issue is #12194 — no, and deliberately so. Still reproduces:
#20815 — partially, but not enough to close it. const producer = Bun.spawn({ cmd: [...], stdout: "pipe" });
const rs = producer.stdout; // never read
// rssDelta=496MB after ~1sSo bounding the sink slows the growth but the reader keeps draining the pipe into memory regardless of |
|
Fair catch on the description check — restructured it under the template's |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
I didn't find any bugs, but this changes behavior in the shared streams pump (cancel semantics, rejected-write handling) that also backs HTTP/fetch/S3 sinks, and the Pending → Backpressure override in write_result has subtle Windows-side interactions — worth a human look before landing.
Extended reasoning...
Overview
Five files: a new buffered_size() accessor on both POSIX and Windows streaming writers (src/io/PipeWriter.rs); FileSink now reports Writable::Backpressure once >256 KiB is buffered and a ReadableStream is attached (src/runtime/webcore/FileSink.rs); the C++ pump now aborts on an already-rejected write promise and cancels the source via the ReadableStreamCancel abstract op instead of the public locked-check path (BunStreamSource.cpp); a doc-comment tweak on Writable::Backpressure (streams.rs); and test updates including un-todo'ing the long-standing cancellation test and two new sync-pull backpressure tests.
Security risks
None identified. The change reduces a DoS surface (unbounded buffering when a subprocess stops reading stdin). No new untrusted-input parsing, auth, or crypto paths.
Level of scrutiny
Moderate-to-high. The FileSink backpressure gate is scoped to the readable_stream.has() case (spawn stdin only), which limits blast radius nicely. But the two C++ pump changes are not scoped to FileSink: cancelStreamIgnoringResult and the rejected-promise abort in rsisWriteChunk apply to every sink routed through readStreamIntoSink / assignStreamIntoResumableSink (HTTP/HTTPS/H3 response, Network/S3). The old cancel path was a no-op by construction, so making it actually run cancel() is a real behavior change for those consumers too.
Other factors
- Design choices a human should confirm: the 256 KiB HWM constant; gating backpressure on
readable_stream.has()(soBun.file().writer()and directproc.stdin.write()still buffer unboundedly by design); and whether un-todo'ing the #20582 test is now safe on all platforms. write_resultsubtlety: whento_resultreturnsPending(every write on Windows), it has already bumpedpending.consumed, taken the keep-alive ref, and setpending.result— thenwrite_resultdiscards that and returnsBackpressureinstead. The follow-upflush(true)→flush_from_jspath re-entersto_resultand creates the promise there, and themust_be_kept_alive_until_eofguard prevents a double-ref, so it looks correct, but the interaction is non-obvious and worth a maintainer's eye.- Existing-test edits: two helpers now kill the child on chunk 2 instead of 4 because backpressure stops pulls earlier — the assertions are unchanged, but a reviewer should confirm the tests still exercise the same failure mode they were written for.
- The PR description reports broad regression testing (streams/, filesink, fetch.stream, serve*, spawn/) and
rust:check-allon all targets, and CI is building.
Good instinct, and it paid off: there was a real bug next door, though not in The flush promise that let has_pending_data = (*this).writer.get().has_pending_data(); // snapshot
...
FileSink::run_pending(this); // re-enters JS, buffers more
if (*this).done.get() && status == WriteStatus::Drained {
(*this).writer.with_mut(|w| w.end()); // POSIX end() closes the fd now
}POSIX The fix re-reads Latent before this PR: nothing ever awaited that promise, so the pump never resumed inside On your other points:
New regression test for the truncation: chunks larger than the HWM park every other write, so an odd number of them reliably leaves one buffered as the stream closes. 12/12 against the unguarded intermediate, 25/25 clean with the fix. |
There was a problem hiding this comment.
I didn't find any bugs, but this changes core stream-pump and FileSink write/drain semantics (including the re-entrant on_write → run_pending → re-read has_pending_data path and the pump's cancel/reject handling), so it's worth a human look before merging.
Extended reasoning...
Overview
This PR spans five files: a trivial buffered_size() accessor added to both the POSIX and Windows streaming writers in src/io/PipeWriter.rs; a new write_result() path in src/runtime/webcore/FileSink.rs that converts non-terminal write outcomes into Writable::Backpressure once >256 KiB is buffered (gated on readable_stream.has()), plus a subtle change in on_write to re-read has_pending_data after run_pending() drains microtasks; a doc-comment refinement in src/runtime/webcore/streams.rs; and in src/jsc/bindings/webcore/streams/BunStreamSource.cpp, replacing publicStreamCancelIgnoringResult (which checked isReadableStreamLocked and rejected) with cancelStreamIgnoringResult (which calls readableStreamCancel unconditionally) at both rsisAbrupt and resumableHandleAbrupt, and making rsisWriteChunk throw when sink.write() returns an already-rejected promise. Tests are added/updated in test/js/bun/spawn/spawn-stdin-readable-stream.test.ts, including un-todo'ing a cancellation test and adding three new tests for synchronous-pull backpressure, byte-count integrity behind backpressure, and source cancellation on early child exit.
Security risks
None identified. This is I/O flow-control and error-path handling; no auth, crypto, path handling, or untrusted-input parsing is touched. The fix itself mitigates a resource-exhaustion / event-loop-starvation vector (unbounded buffering when a subprocess consumes stdin slowly or exits early).
Level of scrutiny
High. While each individual hunk is small, the changes touch the intersection of the C++ readStreamIntoSink/resumableSink pumps, the Rust FileSink write-completion state machine, and re-entrant JS execution during run_pending. In particular: (1) the on_write change re-reads has_pending_data after resolving the flush promise, because the resumed pump can synchronously buffer more bytes and end() — closing the fd on a stale "drained" snapshot would truncate the child's stdin (covered by the new "receives every byte" test); (2) dropping the locked-stream check on the pump's abort path changes cancellation semantics for two pump implementations; (3) the Backpressure sentinel is deliberately scoped via readable_stream.has() so the public FileSink.write() number-or-Promise contract is unchanged, but that gating condition is worth a maintainer's confirmation. These are exactly the kind of lifecycle/re-entrancy edges where a subtle mistake causes hangs, truncation, or leaks on one platform.
Other factors
The PR description is exceptionally thorough — root cause, mechanism, gating rationale, and a manual regression sweep across streams/fetch/serve/spawn suites are all documented — and the author verified rust:check-all on all 10 targets for the cfg-gated Windows accessor. The bug-hunting system found nothing. Two existing test fixtures had their kill trigger moved from the 4th to the 2nd chunk to accommodate the new backpressure behavior; the assertions themselves are unchanged and the rationale is commented. Still, this is a behavioral change to a hot, cross-language, cross-platform I/O path and not a mechanical fix, so I'm deferring rather than approving.
CI status: the diff is green, the red lanes are unrelatedThe What is currently red on #68726, and why none of it is this PR:
Across the three builds on this branch the red lane has been different every time ( I also chased the one mechanism by which the Verification on the diff itself:
I've used my one re-trigger already ( |
|
Thanks — I chased both down against a build of the branch. Neither is reachable today, but (2) pointed at a real gap: the gate's safety rests on an invariant nothing pinned. Fixed in (2) the sentinel cannot reach
|
|
Update on the CI triage above, since the red lane moved again. #68743 ( That test cannot reach this diff:
Everything this PR changes is behind Running tally of red lanes across the five builds on this branch:
A different test every time, never the same one twice, and I've used my one re-trigger ( |
Jarred-Sumner
left a comment
Jarred-Sumner
left a comment
There was a problem hiding this comment.
Hardcoding a number to signal backpressure is incorrect. Backpressure should be based on something from the operating system - EAGAIN or similar.
|
You're right, and the signal was already sitting there — I just didn't use it. Reworked in
sys::Result::Err(err) => {
if err.is_retry() {
return WriteResult::Pending(offset);
}and that reaches the parent as if !self.readable_stream.with_mut(|s| s.has()) || !self.writer.get().is_backed_up() {
return result;
}
Worth noting Two things fell out of it, both good: The buffer is now bounded by one chunk instead of 256 KiB. The pump parks on the first refused write. A child that never reads stdin now takes 5 pulls and ~12 MB RSS, where the old threshold took 9. It costs nothing when the consumer keeps up. Same build, 64 MB through Identical within noise — a fast reader drains the pipe, One honest consequence. The Also had to retune the two child-dies-mid-write fixtures. They killed on the second chunk; the pump now parks on the first refused write, so a second chunk is never produced. They kill on the first now, from a macrotask, which lands after the microtask that issued the write. |
|
CI on The single red check is Three annotations, all marked flaky and all passed on retry:
To save a re-read, the state of this PR:
I've used my one re-trigger ( |
FileSink accepted every chunk the readStreamIntoSink pump handed it, buffering whatever the destination would not take. With a synchronous pull() the pump's read/write cycle never leaves the microtask queue, so the event loop stopped servicing timers and IO while the buffer grew without bound. FileSink now reports Writable::Backpressure once more than 256 KiB is sitting unflushed in the writer, which the pump already handles by awaiting flush(true). Only a sink being fed by a stream pump reports it: the negative sentinel is private to that pump, so Bun.file().writer() and proc.stdin keep write()'s number-or-Promise contract. Two related defects in the pump: - A rejected write promise (EPIPE once the child closes its stdin) was markAsHandled'd and the pump read on into a dead sink. It now aborts. - Both pumps cancelled the source through ReadableStream.prototype.cancel while holding the stream's reader, so the lock check rejected every call and the source's cancel() never ran. Use the ReadableStreamCancel abstract op, as pipeTo does.
Backpressure parks the pump on `sink.flush(true)`. Resolving that promise happens inside `FileSink::on_write` via `run_pending`, which drains microtasks: the pump resumes there, writes the stream's last chunks and calls `end()`, all before `on_write` continues. `on_write` then saw `done && Drained` from the snapshot it took before the re-entry and called `writer.end()`, which on POSIX closes the fd immediately, discarding the bytes that had just been buffered. The child's stdin was truncated (1 MiB sent, 911168 bytes delivered). Re-read `has_pending_data` after `run_pending` instead of trusting the snapshot. Chunks larger than the high-water mark make every other write park, so a stream of an odd number of them reliably lands a chunk in the buffer as the stream closes; the new test pins that.
The negative `write()` return is private to `readStreamIntoSink`, which is safe only because a sink fed by a ReadableStream has no JS handle: spawn caches `proc.stdin` as the stream itself rather than exposing the FileSink. Nothing pinned that, so a change to the `stdin` getter would silently leak the sentinel into user code. Add a test for it, and say so at `write_result` instead of implying `proc.stdin` is a FileSink that happens to have no stream attached.
The 256 KiB high-water mark was invented, not measured. The kernel already says when the destination is full: `write(2)` returns EAGAIN, which `try_write` surfaces as `WriteResult::Pending` and reports to the parent as `WriteStatus::Pending`. Track that and gate backpressure on it. `has_pending_data()` is not the same signal: it's also true while writes below CHUNK_SIZE coalesce in a buffer the kernel hasn't been shown yet. On Windows `uv_write` is always async, so a write in flight means nothing; the equivalent is `process_send` finding one already outstanding and leaving the bytes in `outgoing`. The writer now holds at most the one chunk the destination refused, rather than 256 KiB. Costs nothing when the consumer keeps up: 64 MB through `cat` measures 461 MB/s with backpressure and 444 MB/s without, i.e. the same. A fast reader drains the pipe, the write completes, and the pump never parks. Both child-dies-mid-write fixtures killed on the second chunk. The pump now parks on the first refused write, so no second chunk is ever produced: kill on the first, from a macrotask, which lands after the microtask that issued the write.
93ac734 to
2d34e88
Compare
|
Rebased onto main at The conflicts were with #33538, which independently added
Post-rebase: 30 pass / 0 fail on CI on the rebased sha (#69612, final): 280 test jobs passed; 0 test failures. The 6 red lanes are all macOS infrastructure: 2 × My one re-trigger is spent and re-rolling won't fix a macOS agent pool, so I'm not pushing another empty commit. Needs a maintainer to re-run the macOS lanes or merge over them. |
|
Closing this since #36087 (fetch/S3: replace ResumableSink with proper JSSinks) merged and covers the same ground. Thank you @robobun for the PR — if there's a piece of this that #36087 didn't pick up, please say so and we'll take another look. (This comment was written by Claude, on behalf of the Bun team.) |
|
Thanks for the heads-up — I checked against current main ( Covered by #36087The headline bug is fixed. A synchronous Not covered: child exits → EPIPE → pump never stopsThe original report's exact shape — a synchronous const chunk = Buffer.alloc(64 * 1024, "x");
let pulls = 0;
const rs = new ReadableStream({
pull(c) { if (++pulls > 500) { console.log("never bounded"); process.exit(1); } c.enqueue(chunk); },
cancel: () => console.log("cancelled"),
});
await Bun.spawn({ cmd: ["head", "-c", "10"], stdin: rs, stdout: "ignore" }).exited;On main: It's specific to how the child closes its end. Verified across five shapes on main:
So orderly close works; a direct-exec child that dies holding the pipe (→ EPIPE on the next write) does not. Main's shouldSuspend = wrotePromise->status() == JSPromise::Status::Pending;which means a Rejected write promise neither suspends nor aborts — the pump keeps reading into a dead sink. That's what this PR's My tests used a bun child, so they pass on main and aren't a guard for this. The Happy to open a small focused PR with just that piece: the rejected-write abort in |
2 participants
What does this PR do?
Fixes a process-level DoS when a
ReadableStreamis piped into a subprocess'stdin: the stream is drained with no backpressure, starving the event loop and
growing memory without bound.
Repro
On
mainthis never printstimer firedand never resolvesproc.exited: RSSclimbs past 4 GB in a few seconds while
pull()is called ~500k times/s. Achild that reads nothing at all (
sleep 10) buffers 3.1 GB in 3 s with zerotimer ticks. Node paces the same source through
child.stdinand stays flat.A synchronous
pull()is the natural shape for a generator-backed source, andnothing in the docs asks for an async one, so this is reachable from any child
that consumes slowly or exits early.
Cause
readStreamIntoSinkonly stops reading whensink.write()returns a negativenumber.
FileSinknever returns one: it buffers whatever the destination willnot take and reports the bytes as written. Since the read is already fulfilled
from the stream's queue, the whole read/write cycle stays inside the microtask
queue and the event loop never runs.
Two more defects fall out of the same pump:
FileSink.write()surfaces a failed write (EPIPE, once the child closes itsstdin) as an already-rejected Promise.
rsisWriteChunkcalledmarkPromiseAsHandledon it and kept reading into a dead sink forever.publicStreamCancelIgnoringResult, i.e.ReadableStream.prototype.cancelsemantics. A pump always holds the stream'sreader, so
isReadableStreamLockedwas always true and every call built arejected TypeError that was immediately swallowed. The source's
cancel()could never run.
Fix
FileSinkreportsWritable::Backpressure(the existing-(len + 1)sentinel the HTTP sink already uses) once the destination refuses bytes,
and the pump's existing
await sink.flush(true)path takes it from there.The signal comes from the OS, not a threshold:
write(2)returnsEAGAIN,which
try_writesurfaces asWriteResult::Pendingand reports asWriteStatus::Pending. (has_pending_data()is a different thing — it isalso true while sub-
CHUNK_SIZEwrites coalesce in a buffer the kernel hasnot been shown. On Windows
uv_writeis always async, so the equivalent isprocess_sendfinding one already in flight and leaving the bytes inoutgoing.) The writer ends up holding at most the one chunk the destinationrefused. Gated on a
ReadableStreamactually being pumped into the sink: thesentinel is private to that pump, so
Bun.file().writer()andproc.stdin.write()keepwrite()'s number-or-Promise contract.ReadableStreamCancelabstract op, theway
pipeTodoes, so the lock check no longer eats the cancel.Making the pump actually park on
flush(true)then exposed a latent re-entrancybug in
FileSink::on_write. The flush promise is resolved from insideon_writeby
run_pending, which drains microtasks: the pump resumes right there, writesthe stream's last chunks and calls
end(), all beforeon_writecontinues.on_writethen acted on thedone && Drainedsnapshot it took before there-entry and called
writer.end(), which on POSIX closes the fd immediately anddiscards whatever had just been buffered, truncating the child's stdin (1 MiB
sent, 911168 bytes delivered). It now re-reads
has_pending_dataafterrun_pendingrather than trusting the snapshot.With the fix the repro exits cleanly after 3 pulls, RSS stays flat, and
cancel()runs with the EPIPE reason.Related
Partially addresses #20815 (
stdin: otherProc.stdout): that shape does gothrough this pump, but it still grows without bound, because the
subprocess-stdout source has its own missing backpressure. A
proc.stdoutnobody reads buffers ~496 MB in about a second on this branch. Leaving that
issue open for the separate fix.
How did you verify your code works?
New and updated tests in
test/js/bun/spawn/spawn-stdin-readable-stream.test.ts:pull(). One asserts timers keep firing whilea child that never reads stdin is fed; one asserts the source is cancelled
when the child exits early. Both bail out of
pull()past a bound so theyfail fast rather than exhausting memory. On
mainboth fail withpull() was never bounded by backpressure; they pass on this branch.high-water mark park every other write, so an odd number of them reliably
leaves one in the buffer as the stream closes. It passes on
main(nothingparks there, so the re-entrancy window never opens) and caught the bug 12/12
against the intermediate version of this PR that lacked the
on_writeguard.proc.stdinisthe stream itself, not a
FileSink, whenstdinis a ReadableStream (spawncaches it), so the sink that reports backpressure has no JS handle. Nothing
enforced that before, and a change to the
stdingetter would have leaked-(len + 1)into user code.todosReadableStream cancellation when process exits early(todo sinceEnable ReadableStream as stdin for Bun.spawn #20582) and makes it await the cancel instead of sleeping 100 ms.
expectNoUnhandledRejectionWhenChildDies/expectParentExitsAfterChildDieskilled the child on the 4th chunk; backpressure now stops the pull before
that, so they kill on the 2nd (the first write is already in flight by then).
Assertions unchanged.
Regression sweep, each suite compared against a build of the same tree with
src/reverted, so this container's slow-test timeouts don't read asregressions:
test/js/web/streams/,test/js/bun/util/filesink.test.ts,test/js/web/fetch/fetch.stream.test.ts,fetch-leak,body,fetch-response-finalizer-sweep,test/js/bun/http/serve*and the rest oftest/js/bun/spawn/show no new failures.Backpressure costs nothing when the consumer keeps up: 64 MB through
cat,same build, measures 461 MB/s with it and 444 MB/s without — identical within
noise, because a fast reader drains the pipe, the write completes, and the pump
never parks.
bun run rust:check-allpasses on all 10 targets (the Windows writer'sis_backed_up()iscfg-gated).Rebase note
Rebased onto main after #33538 landed, which independently added the
buffered_len()accessor and gaveto_resultanaccepted: u64argument.Resolved by keeping main's accepted-byte computation (the post-transcode buffer
delta, which is more accurate than the input length this PR originally used) and
routing it through
write_result.is_backed_up()now sits alongside main'sbuffered_len()rather than replacing it.