ByteStream::drain: empty buffer before signal_drained (JS-reader path) - #36879
ByteStream::drain: empty buffer before signal_drained (JS-reader path)#36879robobun wants to merge 8 commits into
Conversation
drain() was calling signal_drained() (producer.ready()) before replacing the buffer with an empty Vec. A producer whose on_ready inspects the ByteStream's buffer length to decide whether output backpressure has cleared (RewriterPipe does this in output_backpressured()) would see the old buffer still present and bail without waking its own upstream. In the TextEncoderStream -> HTMLRewriter -> CompressionStream chain this deadlocked: the rewriter's write() returned Backpressure once one input chunk's output exceeded the 256-byte high-water mark, parking the encodeIntoSink write on m_nativeSinkReadyPromise; the only thing that resolves that promise is the sink's onReady, which RewriterPipe::resume -> drain_pending_input -> src.ready() drives, and resume() bailed because drain() had signalled while the buffer was still full. Nothing re-signals once the buffer is actually emptied, so the encoder stayed parked, the TransformStream's readable never closed, and the readStreamIntoSink pump's readMany never settled. on_pull() already empties first and signals second; make drain() match.
Walkthrough
ChangesHTMLRewriter backpressure
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
|
Updated 7:08 AM PT - Aug 4th, 2026
❌ @robobun, your commit bcb3861 has 2 failures in
🧪 To try this PR locally: bunx bun-pr 36879That installs a local version of the PR into your bun-36879 --bun |
![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 `@test/js/workerd/html-rewriter.test.js`:
- Around line 1760-1764: Remove the await Bun.sleep(0) calls from both
multi-chunk source pull() implementations and enqueue the first chunk directly
when pull() is invoked, retaining the existing counter-based close behavior for
subsequent pulls.
🪄 Autofix (Beta)
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: 085d1908-048d-4cef-a12f-aa1cb439b88a
📒 Files selected for processing (2)
src/runtime/webcore/ByteStream.rstest/js/workerd/html-rewriter.test.js
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
… native-sink wiring paths The drain() reorder in the previous commit fixed the JS-reader path but opened a re-entrancy window for the three callers that install self.sink before calling drain() (wire_native_sink, S3 upload, RequestContext): a producer whose on_ready feeds synchronously (RewriterPipe) would emit newer bytes to the sink before the caller wrote the older drained bytes. Factor resume()'s write-then-signal body into ByteStream::flush_to_sink() and have those callers use it instead of drain() + manual sink.write. drain() is now only the JS-reader entry point (native-source adapter / blob hand-off); a debug_assert enforces it is not called with a sink installed. The rewriter-output-to-native-sink state was a pre-existing deadlock on main (resume bailed on output_backpressured and nothing re-signalled); this makes it complete in order instead.
robobun
changed the title
materializeNativeSource can reach drain() with a sink already wired: getReader() on a natively-locked stream materialises before the lock check throws. Guard the signal_drained call instead so that path neither panics nor lets a synchronous producer emit ahead of the drained bytes.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/runtime/webcore/ByteStream.rs (1)
186-193: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winSlice buffered bytes from
self.offsetin bothdrain()andflush_to_sink()before returning or sending to sink.
self.offsetmarks the start of unconsumed data inself.buffer. Line 549 showson_pullalways reads fromb[self.offset.get()..]and updatesoffseton consumption (line 555). Line 481 showsappend()setsself.offsetwhen storing pull leftovers.
drain()at line 693 takes the entire buffer withVec::move_from_list(self.buffer.replace(...))without slicing fromself.offset. If a reader partially consumed the buffer throughon_pull(leavingoffset > 0with remaining data), a laterdrain()call returns the already-delivered prefixbuffer[..offset]along with the remaining bytes. This causes duplicate bytes to be returned to consumers.
flush_to_sink()at lines 186–193 takes the entire buffer and resetsoffsetto0afterward (line 188) without slicingbuffer[self.offset..]first. The sink receives all bytes including the already-delivered prefix, causing duplicate bytes.Both functions must slice from
self.offsetbefore using the buffer contents and resetoffsetto0after extracting bytes. For example:buffer[self.offset..].to_vec()or equivalent, matching the consumption logic inon_pullat lines 549–555.🤖 Prompt for 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. In `@src/runtime/webcore/ByteStream.rs` around lines 186 - 193, Update both flush_to_sink() and drain() to extract only the unconsumed portion of self.buffer starting at self.offset before returning or sending bytes. Reset self.offset to 0 after extraction, preserving the existing completion and sink-handling behavior while preventing already-consumed prefixes from being emitted again.
🤖 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.
Outside diff comments:
In `@src/runtime/webcore/ByteStream.rs`:
- Around line 186-193: Update both flush_to_sink() and drain() to extract only
the unconsumed portion of self.buffer starting at self.offset before returning
or sending bytes. Reset self.offset to 0 after extraction, preserving the
existing completion and sink-handling behavior while preventing already-consumed
prefixes from being emitted again.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: a30536f2-379b-4746-9bb0-0c8277661726
📒 Files selected for processing (1)
src/runtime/webcore/ByteStream.rs
…or the sync producer behind a sink flush_to_sink() changed the EndedInline contract: it called sink.end() on every terminal path, and every wire_native_sink caller then ran its own end again (FetchTasklet::write_end_request double-deref, NetworkSink pump-ref double-release). The state those changes targeted (RewriterPipe output buffer > hwm then wired to a native sink) is a pre-existing deadlock on main; leave it as-is rather than trading it for a refcount over-release. drain() now empties before signalling (the JS-reader fix) and skips the signal only for the one producer whose on_ready feeds synchronously (RewriterPipe) when a sink is installed, so async producers (FetchTasklet, FileReader) are still resumed exactly as on main.
robobun
changed the title
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/runtime/webcore/ByteStream.rs (1)
675-694: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winReset
self.offsetto 0 before moving the entire buffer indrain().
drain()takes the entireself.bufferwithout accounting forself.offset, which tracks how much of the buffer a prioron_pull()call has already delivered to the JS pump. If the JS pump has partially consumed the buffer viaon_pull()(advancingoffsetbut leaving unconsumed bytes), thenwire_native_sink()callsdrain()whileoffsetis nonzero. The returnedVec<u8>then includes the already-delivered prefix, duplicating bytes for the native sink consumer. Meanwhile,offsetremains stale whilebufferbecomes empty. The nextappend()call with an empty buffer compounds the stale offset value instead of resetting it (line 471). A subsequenton_pull()computesb.len() - self.offset.get()(line 536) against a fresh, small buffer, causing ausizeunderflow.Match the pattern in
resume()(line 178), which resetsoffsetbefore taking the entire buffer. Resetoffsetto 0 and, ifoffsetis nonzero, skip it when moving the drained bytes.Proposed fix
pub(crate) fn drain(&self) -> Vec<u8> { if self.buffer.get().is_empty() { return Vec::<u8>::default(); } - let drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new())); + let offset = self.offset.replace(0); + let mut drained = Vec::<u8>::move_from_list(self.buffer.replace(Vec::new())); + if offset > 0 && offset < drained.len() { + drained.drain(..offset); + } let sync_behind_sink = self.sink.get().is_some()🤖 Prompt for 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. In `@src/runtime/webcore/ByteStream.rs` around lines 675 - 694, Update ByteStream::drain to capture and reset self.offset before replacing the entire buffer, then remove the already-delivered prefix when the offset is nonzero so the returned bytes contain only unconsumed data. Follow the existing reset-and-skip pattern in resume, ensuring subsequent append and on_pull calls see an empty-buffer offset of zero.
🤖 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.
Outside diff comments:
In `@src/runtime/webcore/ByteStream.rs`:
- Around line 675-694: Update ByteStream::drain to capture and reset self.offset
before replacing the entire buffer, then remove the already-delivered prefix
when the offset is nonzero so the returned bytes contain only unconsumed data.
Follow the existing reset-and-skip pattern in resume, ensuring subsequent append
and on_pull calls see an empty-buffer offset of zero.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: d4c42db8-896f-495f-8ea0-f629abb460cc
📒 Files selected for processing (2)
src/runtime/webcore/ByteStream.rstest/js/workerd/html-rewriter.test.js
💤 Files with no reviewable changes (1)
- test/js/workerd/html-rewriter.test.js
There was a problem hiding this comment.
No bugs found in the final revision. The two re-entrancy issues I flagged earlier (native-sink caller reordering, then the flush_to_sink double-release/double-end in S3 and FetchTasklet) were both addressed by reverting to the minimal fix in 29e33eb — the three sink-first callers are back on main's code path, and drain()'s reorder is guarded to skip signal_drained() only for SourceHandle::HTMLRewriter behind a sink, which preserves main's (pre-existing, separately-filed) deadlock there rather than trading it for byte reordering.
Deferring for a human sign-off on the layering: hardcoding a matches!(producer, SourceHandle::HTMLRewriter(_)) check inside ByteStream::drain() is a producer-specific special case in a generic primitive. It's the pragmatic choice given the more principled refactor broke two refcount contracts, but a maintainer should confirm this is the shape they want to land vs. e.g. a SourceHandle::ready_is_sync() predicate or fixing the native-sink wiring callers.
What was reviewed:
- Traced every
SourceHandle::ready()arm — onlyHTMLRewritergates onout.buffer.len(), so the reorder is behavior-neutral for the others (they either only schedule, or would have re-entered identically on main sinceon_datadoesn't consultbuffer). - Confirmed the guard leaves the HTMLRewriter+sink path exactly as on main (
resume()bailed on the full buffer before; now it's not called — same net effect). - Tests: 8 new cases cover arrayBuffer/HTTP/all 5 compression codecs/multi-chunk; the
Bun.sleep(0)is documented fixture shaping, not a condition wait.
Extended reasoning...
Overview
Two files changed. ByteStream::drain() (src/runtime/webcore/ByteStream.rs) now moves self.buffer out before calling signal_drained(), so a producer's on_ready that inspects buffer.len() sees the drained state. The signal is skipped when self.sink.is_some() and the producer is SourceHandle::HTMLRewriter, to avoid opening a re-entrancy window on the native-sink wiring paths. Eight regression tests are added to test/js/workerd/html-rewriter.test.js covering the TextEncoderStream → HTMLRewriter → CompressionStream deadlock across arrayBuffer(), Bun.serve, all five compression formats, and multi-chunk input.
Security risks
None. This is internal backpressure/ordering plumbing between a native ReadableStream source and its producer's ready signal; no untrusted input parsing, auth, or crypto is touched.
Level of scrutiny
High. ByteStream::drain() is called from every native-sink wiring path (wire_native_sink, S3 upload, RequestContext) and from the JS-reader adapter, and its ordering wrt signal_drained() is exactly the invariant this PR changes. The two earlier iterations of this PR each introduced real bugs (byte reordering on the sink-first paths; then a UAF in S3 and a double FetchTasklet::deref when the flush_to_sink refactor ran sink.end() before the caller's own EndedInline handling). The final revision reverts to a 13-line targeted change, which I've now verified: the reorder only changes observable behavior for producers whose on_ready reads the downstream buffer.len() — RewriterPipe::output_backpressured() is the sole such producer, and it is guarded on the sink-installed path so the native-sink callers keep main's exact semantics.
Other factors
- The hardcoded
SourceHandle::HTMLRewritermatch in a genericByteStreammethod is a layering compromise. REVIEW.md's "fix bugs at the layer that owns the violated invariant" would point at either the native-sink wiring callers (write drained bytes before installing the sink) or atSourceHandle(aready_is_sync()predicate). The PR author tried the caller-side fix and it broke two refcount contracts, so this is the conservative fallback — but that trade-off is a maintainer call. - The PR explicitly leaves a pre-existing deadlock in place (HTMLRewriter output → native sink) and says it's filed separately; the guard is what preserves it. A maintainer should confirm that's acceptable scoping.
- The comment-cop bot flagged the 5-line inline comment; robobun kept it as recording why the guard is scoped to one variant. I agree it's load-bearing (a future reader would otherwise widen or remove the condition), but it's another thing a human should glance at.
- Test coverage is good: the fixtures assert exact decompressed output (not just "doesn't hang"), the
Bun.sleep(0)fixture-shaping is commented with the mechanism, andit.eachcovers the codec matrix.
|
The diff is green locally (129 pass in
Ready for review. |
There was a problem hiding this comment.
The revert in 29e33eb resolves both refcount issues I raised (S3 double-release, FetchTasklet double-end) — wire_native_sink/S3/RequestContext are back to main and the EndedInline contract is unchanged. The remaining drain() change is minimal and I verified only FetchResponseBody/ServerRequestBody/HTMLRewriter are ever set as a ByteStream producer; the first two only schedule async work from on_ready, so the reorder is a no-op for them and the guard covers the one sync producer.
Leaving this for a human to sign off on the matches!(SourceHandle::HTMLRewriter(_)) special-case in shared ByteStream code and the explicitly-punted native-sink→rewriter deadlock.
Extended reasoning...
Overview
Two files: src/runtime/webcore/ByteStream.rs (~13 net lines in drain()) and 8 new tests in test/js/workerd/html-rewriter.test.js. The Rust change reorders drain() to move bytes out of self.buffer before signal_drained() (matching on_pull()), and skips the signal when the producer is SourceHandle::HTMLRewriter and a native sink is already installed. This fixes a deadlock in TextEncoderStream → HTMLRewriter → CompressionStream chains where the rewriter's output ByteStream is drained by a JS reader.
Review history
This is the third revision I have looked at. My first pass flagged a re-entrancy window the reorder opened for the three sink-first drain() callers; the author factored out flush_to_sink() and rewired those callers. My second pass found that refactor introduced double-releases in the S3 upload path and FetchTasklet::write_end_request because flush_to_sink() called sink.end() where the callers' EndedInline contract expected them to own end handling. The author reverted the refactor entirely in 29e33eb and scoped the fix to drain() alone with a producer-variant guard. That revert cleanly resolves both issues — the three callers are byte-identical to main.
Correctness of the current diff
I grepped for every site that sets a ByteStream's producer (the NewSource.producer field, distinct from a sink's source field). Only FetchResponseBody (FetchTasklet.rs:1770), HTMLRewriter (html_rewriter.rs:971), and ServerRequestBody (server/mod.rs:836, server_body.rs:3323) are ever assigned. FetchTasklet::on_stream_drained transitions receive mode + schedule_receive_resume() (async); RequestContext::on_request_body_stream_drained calls resp.resume() on the uWS socket (async). Neither reads the downstream buffer or feeds synchronously, so moving the buffer swap before the signal is a no-op for them. HTMLRewriter is the one sync feeder and is guarded. The SourceHandle::ByteStream/FileReader arms in ready() are unreachable from a ByteStream's signal_drained() because nothing sets those variants as a producer.
For the sink-installed HTMLRewriter case, the guard preserves main's behaviour exactly (deadlock — resume() bailed on buffer.len() > hwm before, now it is never called), which the PR punts to a separate issue rather than trading for silent byte reordering.
Security risks
None. No untrusted-input parsing, allocation sizing, or auth/crypto surface. The change is control-flow ordering inside a single-threaded stream pump.
Level of scrutiny / why defer
The fix itself is small, correct as far as I can trace, and well-covered (8 tests, all deadlock on main; broader stream suites reported green). Two things a maintainer should weigh:
ByteStream::drain()now hardcodes knowledge of one specific producer variant's sync-feed behaviour. REVIEW.md flags variant-specific special-casing in shared infrastructure as a layering smell; the author already tried the caller-side fix and it broke refcount invariants, so this is a defensible tradeoff — but it is a tradeoff.- The rewriter-output→native-sink deadlock is explicitly left as-is on main and filed separately. That is a reasonable scope boundary but should be acknowledged by a human.
The comment-cop bot flagged the inline comment repeatedly; the author's defence (it documents why the guard is scoped to exactly one variant, and a debug_assert! of the same invariant crashed CI build 88797) is reasonable and I would not block on it.
…wiring writes the buffered bytes (#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 #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 #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 ```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.
…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.
|
Second, independent route into the same lost wake-up, so this is not specific to the TextEncoderStream or CompressionStream wiring: a plain JS const doc = Buffer.from(
"<!DOCTYPE html><html><body>" + '<div id="x" a=1>hello<!--cm--><i>t</i></div>'.repeat(8) + "tail text</body></html>",
);
let i = 0, ended = false;
const input = new Response(
new ReadableStream({
async pull(c) {
if (i >= doc.length) return c.close();
await Bun.sleep(0);
c.enqueue(doc.subarray(i, i + 24));
i += 24;
},
}),
);
const out = new HTMLRewriter()
.on("div", { element: e => e.setAttribute("q", "1") })
.onDocument({ end: d => { ended = true; d.append("<!--end-->", { html: true }); } })
.transform(input);
await out.body.pipeTo(new WritableStream({ async write() { await Bun.sleep(3); } }));
// main (1.4.0-canary da3851e57; ByteStream.rs is unchanged through 9a543cc18): never resolves.
// The sink receives 3 chunks (300 bytes), the input's pull() stops being called, end() never runs.Same sequence as described above: the pump's Verified on current main (9a543cc) with the Two notes for a rebase, since main moved under this PR:
Branch with that form of the change plus two deterministic regression tests for the |
1 participant
What
ByteStream::drain()now moves the bytes out ofself.bufferbeforesignal_drained()(which dispatchesproducer.ready()), matchingon_pull(). The signal is skipped only for the one producer whoseon_readyfeeds synchronously (SourceHandle::HTMLRewriter) when a sink is already installed on the ByteStream.Why
RewriterPipe(from #36733) is the only producer that synchronously emits fromon_ready:resume()→drain_pending_input()mayfeed()orend_rewrite()inline. Itsoutput_backpressured()gate isout.sink_paused || out.buffer.len() > high_water_mark, so it only produces when the output buffer is already empty.In the chain
the rewriter's output ByteStream has no
SinkHandle(it is consumed by the specpipeToinsidepipeThrough), so lol-html's output chunks accumulate inbufferuntil the native-source adapter'sdrain()takes them. Once one input chunk produces more than the 256-byte high-water mark:RewriterPipe::writereturnsWritable::Backpressure.encodeAndEnqueueparks the TextEncoderStream's transform-algorithm promise onm_nativeSinkReadyPromise; the body→encoderpipeTowaits on writer.ready.handle.drain().signal_drained()fired while the buffer was still full;RewriterPipe::resume()sawout.buffer.len() > hwmand returned early.drain()then emptied the buffer, but nothing re-signals, soresume()never reachedsrc.ready(), the sink's onReady never fired, andm_nativeSinkReadyPromisenever resolved.Deadlock. A small input happened to work because the first output byte lands in a pending pull and the remaining buffered output stayed under 256 bytes, so
writenever returnedBackpressure.The sink-installed guard
The reorder alone opens a re-entrancy window for the three callers that set
self.sinkbefore callingdrain()(wire_native_sink, the S3 upload fast-path,RequestContext): with the buffer now empty whensignal_drained()runs,RewriterPipe::resume()proceeds and may emit newer bytes to the installed sink (or end it) before the caller writes the older drained bytes. Onmainthat same state is already a deadlock (resume()bailed, never re-signalled); the guard keeps it that way rather than trading a hang for silent data loss. All other producers'on_readyonly schedule async work (FetchTasklet::on_stream_drained,FileReader::pull_into_sink), so for them the reorder just moves the scheduling point across an atomic swap. That pre-existing native-sink hang is filed separately.Repro
Verification
bun bd test test/js/workerd/html-rewriter.test.js→ 129 pass, 0 fail (8 new). All 8 time out on main.text-encoder-stream.test.ts,compression.test.ts,body-stream.test.ts(9086 tests),spawn-stdin-readable-stream.test.ts,fetch.stream.test.ts, andhtml-rewriter-leak.test.tsgreen.cargo clippy -p bun_runtimeclean.