ByteStream: re-signal a backpressured RewriterPipe after native-sink wiring writes the buffered bytes - #36899
Conversation
…wiring write A RewriterPipe's output ByteStream can hold more than the pipe's high-water mark with input_ended=true (reached by materialising .body, then resolving an async handler before any consumer attaches). Wiring a native sink afterwards called drain(), which signalled the producer before emptying the buffer; RewriterPipe::resume() saw the full buffer, bailed on output_backpressured(), and nothing re-signalled once the caller wrote the drained bytes to the sink. end_rewrite() never ran and the downstream consumer hung forever. Take the buffered bytes without signalling, write them to the sink, and only then signal the producer: the sink is installed and the older bytes are already delivered, so a synchronous producer that emits or ends inline does so through on_data (the normal post-install path) instead of racing the caller's write. The had_last branch is unchanged (the producer was already done). The body-mixin collector (to_buffered_value) had the same ordering bug with buffer_action: signal the producer after the action is installed, and have output_backpressured() treat an active buffer_action as 'no backpressure' since the collector grows the buffer deliberately.
WalkthroughChangesStream backpressure and completion
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
…y ended the sink
wire_native_sink's post-write signal can now drive a RewriterPipe with
input_ended all the way through on_data(Done) -> FileSink::end() inline
before the Wired arm runs; end_from_stream clears source on that path,
so gate the keepalive on source still being installed. Covers
Bun.spawn({stdin: parkedRewriterBody}).
A streaming input (direct ReadableStream parked on flush(true)) with a handler that emits the >16 KiB chunk itself parks the rewriter with input_ended=false; signal_drained() must then wake the upstream via drain_pending_input()'s src.ready()/pending.run() tail. Covers the 2nd HTMLRewriter, Bun.serve, and .text() consumers.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Thanks for addressing the FileSink Wired keepalive nit in 399d936 — the self.source gate plus the Bun.spawn({stdin}) test cover it. No further issues from me, but I'm deferring to a human given this reorders the drain-signal across four independent native-sink state machines (RequestContext, FileSink, NetworkSink/S3, RewriterPipe) plus the body-mixin collector, and each now admits synchronous producer re-entry through on_data during wiring.
What was reviewed:
take_buffer+ post-writesignal_drained()ordering inwire_native_sink, RequestContextSource::Bytes, and the S3 fast-path — buffered bytes reach the sink before the producer resumes;had_laststill returnsEndedInline.to_buffered_value: promise is captured beforesignal_drained(), so an inlineaction.fulfill()resolves the same promise the caller receives.output_backpressured()treating an activebuffer_actionas no-backpressure — checked thatsink_pausedstill gates first.- RequestContext:
signal_drained()runs afterrun_corked_with_typereturns, so synchronous producer output goes through the normal uncorkedon_datapath.
Extended reasoning...
Overview
The PR fixes a hang where a RewriterPipe producer parked on output_backpressured() (buffer > 16 KiB) is never re-signalled after a native sink drains that buffer. The fix splits ByteStream::drain() into take_buffer() (empty without signalling) + a separate signal_drained() call placed after the buffered bytes are written to the sink, so the producer observes an empty buffer and resumes. The same reordering is applied in wire_native_sink (ReadableStream.rs), the Bun.serve Source::Bytes arm (RequestContext.rs), the S3 upload fast-path (s3/client.rs), and to_buffered_value (ByteStream.rs). RewriterPipe::output_backpressured() also learns to treat an active buffer_action as no-backpressure. FileSink's Wired arm now guards its keepalive on self.source still being set (my previous review's suggestion, addressed in 399d936).
Security risks
None identified. This is internal streaming-backpressure ordering; no user-controlled input parsing, auth, or crypto is touched.
Level of scrutiny
High. The change is conceptually one pattern (signal-after-write instead of signal-before-empty), but it is applied across four independent native-sink state machines plus the body-mixin collector, and each now admits synchronous producer re-entry (signal_drained → RewriterPipe::resume → on_data(Done) → sink.end()) during the wiring step. I already traced one such path (FileSink) in the previous review and the author addressed it; the RequestContext and S3 paths look correct on inspection (had_last still short-circuits to EndedInline; end_from_stream clears source so the double-end concern doesn't apply), but this is exactly the kind of ordering-sensitive change across production-critical paths (Bun.serve response bodies, spawn stdin, S3 uploads) that benefits from a maintainer's eyes.
Other factors
Test coverage is strong: 11 new tests in html-rewriter.test.js exercise every affected consumer (chained HTMLRewriter, Bun.serve body/Response, Bun.spawn stdin, .text()/.arrayBuffer()) across both input_ended=true and input_ended=false producer states, and the PR description confirms all 11 time out on main. The comment-cop bot flags were addressed in b68f306 (comments trimmed to one/two lines). All prior review threads are resolved. The bug hunting system found nothing this run.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/runtime/webcore/ByteStream.rs`:
- Around line 226-235: Add a short warning doc comment to the `drain()` method
documenting that it signals the producer before clearing the buffer and must not
be used in sink-wiring paths; those paths should use `take_buffer()` followed by
`signal_drained()`. Leave the existing `drain()` behavior and current caller
unchanged.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 92f11119-4b84-43f2-9553-c16b19f63b00
📒 Files selected for processing (7)
src/runtime/api/html_rewriter.rssrc/runtime/server/RequestContext.rssrc/runtime/webcore/ByteStream.rssrc/runtime/webcore/FileSink.rssrc/runtime/webcore/ReadableStream.rssrc/runtime/webcore/s3/client.rstest/js/workerd/html-rewriter.test.js
There was a problem hiding this comment.
Deferring to human review — this reorders producer signalling across four native-sink wiring sites (wire_native_sink, RequestContext Source::Bytes, S3 upload, body-mixin) and lets signal_drained() synchronously drive end_from_stream during wiring, which is a re-entrancy/ordering change to core streaming that warrants maintainer eyes.
What was reviewed: the FileSink Wired keepalive concern from the earlier pass is addressed (399d936 gates on self.source still set); to_buffered_value captures the promise before the inline-fulfil signal so the .unwrap() is safe; RequestContext's unconditional signal_drained() is reached with sink_paused unset (buffered bytes go via on_first_write, not sink.write), so the missing gate is a no-op there; test matrix covers both input_ended states across all four consumers.
Extended reasoning...
Overview
The PR splits ByteStream::drain() into take_buffer() (empty without signalling) + signal_drained() (now pub(crate)), and reorders every native-sink wiring site to signal the producer after the buffered bytes are in the sink instead of before. It also makes RewriterPipe::output_backpressured() treat an active buffer_action as no-backpressure, and reorders to_buffered_value to install the action before signalling. FileSink's Wired arm now skips its keepalive when end_from_stream already cleared source inline. Eleven new tests cover chained rewriter, Bun.serve (body and Response), Bun.spawn({stdin}), body-mixin, and the input_ended=false streaming-input branch.
Security risks
None identified. No user-controlled input parsing changes; the reordering affects internal producer/consumer signalling only.
Level of scrutiny
High. This changes the point at which a backpressure-gated producer is woken across four independent sink paths, and — because RewriterPipe is a synchronous producer — the new signal can now drive on_data(Done) → sink.end() inside the wiring call, before control returns to the caller's Wired/post-install arm. Each caller's post-Wired invariants had to be re-checked (the earlier FileSink note was exactly this). RequestContext.rs is production HTTP-serving hot path; ByteStream::on_data re-entrancy is memory-safety-adjacent per REVIEW.md's "anything that can run user JS can synchronously free your state".
Other factors
My earlier inline concern (keepalive taken against a closed writer) was addressed in 399d936 by gating on self.source; the comment-cop nits were trimmed in b68f306. I traced the RequestContext arm: sink_paused cannot be true at the new signal site (the buffered bytes go via on_first_write/on_writable_bytes, not SinkHandle::write), so its unconditional signal matches the gated ones elsewhere. to_buffered_value reads .value() before signalling, so an inline fulfil that clears buffer_action cannot panic the .unwrap(). The change looks correct to me, but the number of wiring sites and the new inline-end behaviour are exactly the kind of thing a maintainer who owns the SinkHandle/SourceHandle protocol should sign off on.
…wiring writes the buffered bytes (oven-sh#36899) ## What When a `RewriterPipe`'s output `ByteStream` has buffered more than the pipe's high-water mark and that stream is then wired to a native sink (`wire_native_sink`, the `Bun.serve` `Source::Bytes` arm, the S3 upload fast-path, `Bun.spawn` stdin) or consumed by the body-mixin collector (`.arrayBuffer()`/`.text()` on a `new Response(body)`), the consumer now completes instead of hanging forever. ## Why `RewriterPipe` (from oven-sh#36733) is the one `SourceHandle` producer whose `on_ready` feeds synchronously, and its `output_backpressured()` gate is `out.sink_paused || out.buffer.len() > high_water_mark`. Once a handler suspends and the output `.body` is materialised, the resumed rewrite emits into the `ByteStream` buffer; with more than 16 KiB of trailing output it parks on `output_backpressured()` without calling `end_rewrite()`. Wiring a native sink then called `ByteStream::drain()`, which dispatched `signal_drained()` **before** emptying the buffer. `RewriterPipe::resume()` saw the full buffer and bailed; `drain()` emptied it, the caller wrote those bytes to the sink, and nothing ever re-signalled the pipe. `end_rewrite()` never ran, `has_received_last_chunk` stayed `false`, and the sink's `end()` was never reached. The body-mixin path (`to_buffered_value`) had the same stale check: it signalled before installing the `BufferAction`, and each `on_data` while the action is active *appends* to the buffer, so `output_backpressured()` was permanently true. ### Fix - `ByteStream::take_buffer()` empties the buffer without signalling; `signal_drained()` is now `pub(crate)`. - `wire_native_sink` / the S3 fast-path / the `RequestContext` `Source::Bytes` arm use `take_buffer()`, write the older bytes to the sink, and **then** signal. Any synchronous producer output now goes through `on_data` with the sink installed, which is the normal post-install path: ordering is correct and `sink.end()` is driven by `on_data(Done)`, so there is no double-end. The `had_last` branch still returns `EndedInline` (the producer was already done and never re-enters). - `FileSink`'s `Wired` arm skips the keepalive when `source` was already cleared by an inline `end_from_stream`, so the ref is never taken against a closed writer. - `to_buffered_value` installs the `BufferAction` before signalling; `RewriterPipe::output_backpressured()` treats an active `buffer_action` as "no backpressure" (the collector grows `buffer` deliberately until Done). This is the "native-sink hang" that oven-sh#36879 explicitly left in place; that PR fixes the JS-reader `drain()` ordering and guards the sink-installed case to keep the hang rather than trade it for a race. This PR removes the hang on the wiring side without changing oven-sh#36879's guard. ## Repro ```js const prefix = "<p>" + Buffer.alloc(100, "A").toString() + "</p>"; const suffix = "<q>" + Buffer.alloc(30_000, "B").toString() + "</q>"; const html = prefix + "<x></x>" + suffix; const gate = Promise.withResolvers(); const out = new HTMLRewriter() .on("x", { element: () => gate.promise }) .transform(new Response(html)); const body = out.body; gate.resolve(); await 0; await new HTMLRewriter().transform(new Response(body)).text(); // never resolves on main ``` ## Verification `bun bd test test/js/workerd/html-rewriter.test.js` passes 132/132 (11 new); all 11 time out on `main`. The 11 cover both `input_ended=true` (materialized string body) and `input_ended=false` (streaming input parked on `flush(true)`), across a chained HTMLRewriter, `Bun.serve` (body and Response), `Bun.spawn({stdin})`, and the body-mixin path. `body-stream.test.ts` (9086), `streams.test.js`, `compression.test.ts`, `text-encoder-stream.test.ts`, `spawn-stdin-readable-stream.test.ts`, and the other `html-rewriter-*.test.ts` files are green. `cargo clippy -p bun_runtime` clean.
2 participants
What
When a
RewriterPipe's outputByteStreamhas buffered more than the pipe's high-water mark and that stream is then wired to a native sink (wire_native_sink, theBun.serveSource::Bytesarm, the S3 upload fast-path,Bun.spawnstdin) or consumed by the body-mixin collector (.arrayBuffer()/.text()on anew Response(body)), the consumer now completes instead of hanging forever.Why
RewriterPipe(from #36733) is the oneSourceHandleproducer whoseon_readyfeeds synchronously, and itsoutput_backpressured()gate isout.sink_paused || out.buffer.len() > high_water_mark. Once a handler suspends and the output.bodyis materialised, the resumed rewrite emits into theByteStreambuffer; with more than 16 KiB of trailing output it parks onoutput_backpressured()without callingend_rewrite().Wiring a native sink then called
ByteStream::drain(), which dispatchedsignal_drained()before emptying the buffer.RewriterPipe::resume()saw the full buffer and bailed;drain()emptied it, the caller wrote those bytes to the sink, and nothing ever re-signalled the pipe.end_rewrite()never ran,has_received_last_chunkstayedfalse, and the sink'send()was never reached. The body-mixin path (to_buffered_value) had the same stale check: it signalled before installing theBufferAction, and eachon_datawhile the action is active appends to the buffer, sooutput_backpressured()was permanently true.Fix
ByteStream::take_buffer()empties the buffer without signalling;signal_drained()is nowpub(crate).wire_native_sink/ the S3 fast-path / theRequestContextSource::Bytesarm usetake_buffer(), write the older bytes to the sink, and then signal. Any synchronous producer output now goes throughon_datawith the sink installed, which is the normal post-install path: ordering is correct andsink.end()is driven byon_data(Done), so there is no double-end. Thehad_lastbranch still returnsEndedInline(the producer was already done and never re-enters).FileSink'sWiredarm skips the keepalive whensourcewas already cleared by an inlineend_from_stream, so the ref is never taken against a closed writer.to_buffered_valueinstalls theBufferActionbefore signalling;RewriterPipe::output_backpressured()treats an activebuffer_actionas "no backpressure" (the collector growsbufferdeliberately until Done).This is the "native-sink hang" that #36879 explicitly left in place; that PR fixes the JS-reader
drain()ordering and guards the sink-installed case to keep the hang rather than trade it for a race. This PR removes the hang on the wiring side without changing #36879's guard.Repro
Verification
bun bd test test/js/workerd/html-rewriter.test.jspasses 132/132 (11 new); all 11 time out onmain. The 11 cover bothinput_ended=true(materialized string body) andinput_ended=false(streaming input parked onflush(true)), across a chained HTMLRewriter,Bun.serve(body and Response),Bun.spawn({stdin}), and the body-mixin path.body-stream.test.ts(9086),streams.test.js,compression.test.ts,text-encoder-stream.test.ts,spawn-stdin-readable-stream.test.ts, and the otherhtml-rewriter-*.test.tsfiles are green.cargo clippy -p bun_runtimeclean.