fetch/S3: replace ResumableSink with proper JSSinks - #36087
Merged
Jarred-Sumner merged 164 commits intoJul 31, 2026
Conversation
…uffer-sink direct controller A node:stream Readable passed as a fetch request body is consumed via its Symbol.asyncIterator, wrapped as a type:"direct" ReadableStream, then materialized into a JSDirectStreamController backed by an ArrayBufferSink before being drained into the ResumableSink (HTTP socket). driveAsyncIterator loops on iterator.next() results that are already fulfilled, writing each value to the controller and continuing synchronously. A Readable whose _read() pushes synchronously (the common case) makes every next() fulfill synchronously, so the loop never yields: controller.write on an ArrayBufferSink-backed direct controller returns bytes-written (>= 0) and never signals backpressure, so the wrote < 0 suspend path cannot fire. The process spins at 100% CPU growing the sink's Vec unbounded and starving the event loop, so an external destroy() never runs and timers never fire. The fix yields the pull promise after each successful write when the controller is an ArrayBuffer-kind JSDirectStreamController. That controller's pull-fulfilled reaction then flushes the buffered bytes to the waiting read request (the ResumableSink pump), whose own backpressure (HTTP socket buffer vs highWaterMark) naturally bounds the chain and lets the event loop run. Text/Array sinks and native JSSinks keep their existing drive-to-completion semantics; the one-shot arrayBuffer() sink is not a JSDirectStreamController.
Collaborator
Author
|
Updated 8:05 AM PT - Jul 31st, 2026
@Jarred-Sumner, your commit 6b34885 is building: |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughChangesFetch request bodies now use ChangesStream sink migration
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}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
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/web/fetch/fetch.test.ts`:
- Around line 2532-2540: Update the hang regression test around proc.exited to
await proc.exited directly, removing the local sleep-based Promise.race and
timeout kill path. Remove the per-test 15000 timeout so the test runner enforces
the failure bound, while preserving the stdout and stderr assertions.
🪄 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: f4a7a3c8-4803-4dfc-b472-59505f3cd590
📒 Files selected for processing (3)
src/jsc/bindings/webcore/streams/BunAsyncIterableSource.cppsrc/jsc/bindings/webcore/streams/JSAsyncIteratorSourceOperation.htest/js/web/fetch/fetch.test.ts
Replaces the yield-per-write workaround with the real fix: drop the ArrayBufferSink intermediary on the fetch/S3 request-body path. assignStreamIntoResumableSink now recognises a DirectPending stream and, like readDirectStream does for Bun.serve responses, calls the underlying source's pull() with the ResumableSink itself as the controller instead of first materialising into a JSDirectStreamController backed by an ArrayBufferSink and draining that through a default reader. ResumableSink grows the controller surface pull() needs: write(chunk) already returns true/false for backpressure. flush(true) now returns a pending promise while paused; drain() fulfils it and cancel() rejects it, so the pump's existing flush(true) suspend path works unchanged. close() and error(e) alias end()/end(e) so a user type:'direct' pull keeps the same controller API it had via the JSDirectStreamController. driveAsyncIterator treats write() === false as backpressure alongside the wrote < 0 JSSink protocol. A node:stream Readable whose _read() pushes synchronously produces an async iterator whose next() fulfills synchronously. On the old path the pump wrote into the ArrayBufferSink (whose write never signals backpressure) in a tight loop, growing the sink unbounded and starving the event loop. On the new path the first 16 KiB write fills the FetchTasklet buffer, write() returns false, the pump suspends on the drain promise, and the event loop runs; destroy() and timers fire, the iterator throws, the pull promise rejects, and fetch rejects.
robobun
changed the title
Contributor
There was a problem hiding this comment.
Additional findings (outside current diff — PR may have been updated during review):
-
🔴
src/jsc/bindings/webcore/streams/BunAsyncIterableSource.cpp:310-313— This suspends the pump even whencontroller.write()wrote zero bytes (an emptyUint8Array/Buffer/""), which deadlocks the upload:onFlush()with an empty sink restoresm_pendingReadwithout settingm_pullAgain, soonDirectPullFulfilledclearsm_pullInFlightand never re-pulls, and the reader's pending read is never fulfilled. Before this change a zero-byte write fell through toContinueLoopand the pump immediately callednext()again, so this is a regression — gate the yield on bytes actually buffered, e.g.wrote.isNumber() && wrote.asNumber() > 0.Extended reasoning...
What the bug is
The new yield-per-write block fulfills the pull promise and returns
NextStep::Suspendedafter every successfulcontroller.write(), relying on theJSDirectStreamController'sonDirectPullFulfilled→onFlush()reaction to deliver the buffered bytes to the waiting read request and re-armm_pullAgainso the loop re-pulls. ButonFlush()only delivers — and only setsm_pullAgain— whenbyteLengthOf(flushed) > 0. If the yielded value writes 0 bytes into the ArrayBufferSink, the sink is empty at flush time,onFlushrestoresm_pendingReadand returns without touchingm_pullAgain, and nothing ever re-enterspull(). The consumer's read promise is orphaned and thefetch()request-body upload hangs forever.Step-by-step proof
Take:
await fetch(url, { method: "POST", duplex: "half", body: (async function*() { yield new Uint8Array(0); // pump stalls here yield new Uint8Array([1,2,3]); // never sent })(), });
- The ResumableSink pump calls
reader.read()→JSDirectStreamController::onPull.m_pullInFlightis false, socallDirectPullinvokesboundAsyncIterableSourcePull, which setsm_yieldPerWrite = true, creates a freshpullPromise, and entersdriveAsyncIterator. iter.next()fulfills synchronously with{done:false, value: new Uint8Array(0)}. The value is notisUndefinedOrNull(), soasyncIterHandleNextResultcallscontroller.write(value).ArrayBufferSink.writereturns 0 (bytes written).wrote = 0is not< 0and not a promise, so control reaches the new block at lines 310-313:op->m_yieldPerWrite && !op->m_iteratorDone→yieldPullPromisefulfillspullPromiseand clearsm_running, returnsNextStep::Suspended.callDirectPullregistersonDirectPullFulfilledon the (now-fulfilled)pullPromiseand leavesm_pullInFlight = true.onPullthen createsm_pendingRead(line 586-588) and returns it to the ResumableSink pump.- The
onDirectPullFulfilledmicrotask fires and callscontroller->onFlush()(line 792). InonFlush(line 714),m_pendingReadis set → it is cleared,flushDirectSinkreturns a 0-length value,byteLengthOf(flushed) == 0→ the entire delivery block at lines 718-744 is skipped,m_pendingReadis restored at line 745, andonFlushreturns without settingm_pullAgain. - Back in
onDirectPullFulfilled:m_pullInFlight = false(line 793),takeDirectPullAgain()returns false (line 795 —m_pullAgainwas never set; line 566's else-branch only fires when a read arrives while a pull was already in flight, which is not the case for the first read). Thewhile (pullAgain && …)loop at line 798 never enters.
Final state:
m_pendingReadstill holds the ResumableSink's unfulfilled read promise,m_pullInFlight = false,m_pullAgain = false,op->m_running = false,op->m_pullPromisecleared. The ResumableSink awaits that read forever; nothing will callpull()again until the pump reads again, and the pump won't read again until its current read fulfills. Deadlock.The same happens on the async-fulfilled path (
onAsyncIterableSourceNextFulfilled), since it routes through the sameasyncIterHandleNextResultblock.Why existing safeguards don't catch it
- The end-of-tick flush deferred task also lands in
onFlush(), which is the same no-op-restore with 0 buffered bytes. directControllerHasWaitingConsumerwould report a waiter, but the re-pull loop is gated onpullAgainfirst, which is false.m_pullAgainis only set at lines 566/727/740/754, none of which are reachable on this path.
Why this is a regression
Before this PR, a 0-byte write fell through past the (absent) yield block to
return NextStep::ContinueLoop, anddriveAsyncIteratorimmediately callednext()again. Now it suspends with nothing to deliver. Empty chunks are realistic input: anasync function*body that yieldsnew Uint8Array(0), a nodeTransformwhose_transformproduces empty output for some chunks, or user code yieldingBuffer.alloc(0).Impact
Any
fetch()request whose async-iterable body yields an empty chunk hangs the upload permanently — the request never completes, the promise never settles, and the body iterator is never resumed. Per REVIEW.md ("Every error/abort/timeout path actively completes the operation… an unsettled promise pins objects and hangs callers forever"), this is exactly the class of defect that blocks merges.Fix
Only yield when bytes were actually buffered, so a zero-byte write falls through to
ContinueLoopas before:if (op->m_yieldPerWrite && !op->m_iteratorDone && wrote && wrote.isNumber() && wrote.asNumber() > 0) { yieldPullPromise(globalObject, op); return NextStep::Suspended; }
ArrayBufferSink.writereturns bytes-written, so> 0is the correct "something was buffered for onFlush to deliver" predicate. This preserves the PR's fix for the spinning case (non-empty writes still yield) while restoring the pre-PR behavior for empty chunks (loop continues and callsnext()again immediately). - The ResumableSink pump calls
Collaborator
Author
|
Re the empty-chunk deadlock finding in #36087 (review): that code path ( |
end and close both route to jsClose (clean close regardless of argument) so a user type:"direct" pull calling controller.end(value) completes the upload as it did via the JSDirectStreamController; only error(e) aborts. Internal error-end callers (resumableEnd(hasError), onResumableSinkDirectPullRejected, the direct-path abrupt-completion branch) switch to the error name. jsEnd and jsClose settle any pending flush(true) promise before detaching so a pull that holds one across close()/error() does not await it forever. The !pull.isCallable() early-return releases its ThrowScope before the tail invokeMethod, matching readDirectStream's equivalent branch.
js_end is now exclusively the error target, so drop the leftover arity branch that made a zero-arg call clean-close. Rename to js_error to match. Also drop the per-test timeout on the hang regression and tighten the race so it fits under the runner default on debug builds.
Contributor
There was a problem hiding this comment.
All feedback from the last two rounds is addressed in ae14d0e (zero-arg controller.error() now aborts; per-test timeout dropped); the two declined nits (destructured-method receiver, underlyingSource.close hook) are reasonable to leave as-is. No new issues found this pass. This still reworks the fetch/S3 request-body streaming path for every async-iterator/type:"direct" body and adds a GC-visited flushPromise slot to ResumableSink, so it's worth a human look before merge.
What was reviewed:
resumableSetupDirectcontrol flow, exception-scope discipline on every exit, and the abrupt-completion →sink.error()path inassignStreamIntoResumableSink.ResumableSinkflush/close/errorsurface vs the pre-PRJSDirectStreamControllercontract;flushPromisesettled on all four terminal transitions (drain/cancel/js_close/js_error) beforedetach_js().drain()now flipsstatus = Startedoutside theondrainguard — required because the direct path installs noondrainhandler.asyncIterHandleNextResult's newwrote.isFalse()backpressure check and the hang-regression test's self-bounding race.
Extended reasoning...
Overview
The PR replaces the ArrayBufferSink-backed materialisation of type:"direct" request bodies with a direct hand-off of the ResumableSink to the underlying source's pull(). It touches: BunStreamSource.cpp (new resumableSetupDirect, new DirectPending fast path in assignStreamIntoResumableSink, two new bound-handler bodies, resumableEnd now dispatches to error vs end), BunAsyncIterableSource.cpp (wrote.isFalse() backpressure check), JSStreamsRuntime.h (two new handler-list entries), ResumableSink.classes.ts / generated.rs (adds flush/close/error prototype methods and a flushPromise cached-value slot), ResumableSink.rs (implements js_flush/js_close/js_error, take_flush_promise, and settles the flush promise on drain/cancel), and one hang-regression test in fetch.test.ts.
Security risks
None identified. No auth/crypto/permission surface. The new path only changes which controller object a type:"direct" pull() receives when the stream is a fetch/S3 request body; the sink's write() already validated input via StringOrBuffer::from_js.
Level of scrutiny
High. This is production hot-path code (every fetch() with a node Readable or async-iterator body, plus S3 multipart uploads via ResumableS3UploadSink) with GC-visited state additions and new terminal-transition ordering. Two prior review rounds each surfaced real behavioural regressions (the end↔error mapping inversion, the unsettled-flushPromise-on-close hang, the missing scope.release(), the zero-arg error() clean-close), all of which were fixed. That history plus the cross-language surface (C++ ThrowScope discipline, Rust JsRef/cached-slot lifecycle, codegen'd class surface) argues for a human pass.
Other factors
- All prior findings are addressed and marked resolved; the two intentionally-declined items are undocumented-behaviour parity gaps for hand-written
type:'direct'fetch bodies, matching the siblingreadDirectStream/JSSink path. - The
drain()change (movingstatus = Startedoutside theif let Some(ondrain)branch) is a semantic broadening on the pre-existing reader-pump path too, but thereondrainis always set (resumableSetupinstallsboundResumableSinkDrain), so it's a no-op for existing callers. - The bug-hunting system found nothing on this revision. The regression test self-bounds via
Promise.race+kill(9)and assertsexited === 0with a specific rejection-message match, so it fails on both the original spin and a hypothetical resolves-instead-of-rejects regression. - PR description lists
async-iterator-stream.test.ts(91),body-stream.test.ts(9086),body.test.ts(348), andfetch-abort-stream-body.test.tsas passing.
This was referenced Aug 1, 2026
This was referenced Aug 1, 2026
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 2, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 2, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 2, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 2, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 2, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 3, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
added a commit
that referenced
this pull request
Aug 3, 2026
Rebasing over #36087 (ResumableSink -> JSSink/SourceHandle) needed real integration in two places: - RequestContext: the new sink callbacks (`write_chunk`, `end_chunk`, `on_writable_byte_stream`) and the typed drain callback (`on_request_body_stream_drained`) arrived in the `this: *mut Self` + fn-long `&mut *this` style; they are converted to the `&self` model, and the old `on_pipe` / pipe registration is gone with the pipe they served. - S3: main's ref-per-request protocol in the single-upload path (the callback `adopt`s the ref `process_buffered` took, and the retry path takes one for its re-dispatched request) is kept, expressed over the `Cell`-based MultiPartUpload; `fail`/`done` gain main's `is_queue_empty()` re-entry guards. The new `SinkHandle` / `SourceHandle` payloads that are written through (`FetchRequestBody`, `S3Upload`, `ShellWritable`, `FetchResponseBody`, `S3DownloadBody`, and `FetchRequestBodySink.task`) are declared `BackRef<T, Mut>` — the provenance marker surfacing, as designed, that they had been built from read-only-provenance pointers; their heap / `&raw mut` constructions now use `from_raw_mut`. `FileSink`'s handle is read-only and stays Shared.
Jarred-Sumner
pushed a commit
that referenced
this pull request
Aug 4, 2026
…andlers instead of wait_for_promise (#36733) ### What `HTMLRewriter.transform()` now streams: input body chunks flow through `lol-html` into an output `ByteStream` as they arrive, with backpressure propagated end-to-end via the `SinkHandle`/`SourceHandle` pattern from #36087. Async content handlers no longer nest the event loop. ``` input body ──► SinkHandle::HTMLRewriter ──► lol_html::HtmlRewriter ──► output ByteStream ▲ │ └─────────── SourceHandle::HTMLRewriter (producer) ◄────────────────────┘ ``` ### Why `BufferOutputSink` fully buffered the source body via `ValueBufferer` before a single `rewriter.write()` + `end()`, and `handler_callback` spun `vm.wait_for_promise()` six native frames deep inside lol-html for any handler that returned a Promise. That meant handlers fired only at source-end (TTFB = full download), `.body` / `Bun.serve` saw an empty stream (#6068, #19305), JS-backed `ReadableStream` inputs were rejected outright (#11758, #14216), and the nested loop was a known hazard class (deadlocks, ready-poll clobbering, pending-exception leaks). lol-html could not be suspended before because its tokens are stack locals borrowing a stack-local lexeme; returning from `write()` destroys them, and an async handler must be able to mutate the element after its `await`. ### lol-html fork (`oven-sh/lol-html`, branch `bun`) A handler returns `Err(SuspensionRequest)` to suspend. The in-flight unit is deep-copied onto the heap, `write()`/`end()` return the non-poisoning `RewritingError::Suspended`, and `HtmlRewriter::resume()` continues from a `StateMachineBookmark`. The pending captured-text flush is hoisted from `Dispatcher::handle_tag` into the lexer actions so every suspension point has a uniform shape. `Arena::shift` advances a start offset instead of memmoving the tail (a suspended rewrite re-feeds its unconsumed tail on every resume). 23 in-crate tests added; the new `.github/workflows/lolhtml.yml` runs the fork's own `cargo test` at the pinned commit, gated on `scripts/build/deps/lolhtml.ts`. The fork is a `github-archive` source (no patch file), so rebasing onto a new upstream tag is a `git rebase --onto` in the fork plus a commit bump here. ### Bun side (`RewriterPipe` replaces `BufferOutputSink`) - **Ownership**: a generated `HTMLRewriterTransform` JS cell (not user-visible) owns the pipe; its GC finalizer frees it, and nothing pins anything. Liveness is plain GC edges: the output Response's `transform` WriteBarrier slot and the `.then()` context of a suspended handler's (or the JS pump's) promise reach the cell, the cell's five slots root the Response, the input/output streams, the pending flush promise, and a captured handler error, and a wired native source's `owner` WriteBarrier slot (new on the generated NewSource cells) points back at the cell, so I/O that roots the source (a FetchTasklet, a FileReader's read refs, a reader on the output stream) roots the rewrite for exactly the window the raw `SinkHandle`/producer backrefs are wired. A handler promise collected without settling lets `finalize` defer to an event-loop task (`abandon_suspension`) that rejects the body before freeing; the pipe holds one native `+1` on the Response (released in `Drop`) so that task can still reach the body. No intrusive refcount, no `Strong` fields on the pipe; `finish()` frees the boxed lol-html state machine eagerly, and `fail()`/output-cancel close the upstream producer instead of draining it to EOF. - **Output**: the returned `Response`'s body is `Locked(PendingValue { task: pipe, on_start_streaming, on_readable_stream_available, producer: SourceHandle::HTMLRewriter })`, the `FetchTasklet::to_body_value` shape. The `ByteStream` is created lazily when a consumer reads `.body` or a body-mixin method; until then, rewriter output buffers in a `Vec<u8>` handed over as `DrainResult::Owned`. - **Input**: mirrors `FetchTasklet::start_request_stream`. Native `ByteStream`/`FileReader` sources wire `byte_stream.sink = SinkHandle::HTMLRewriter` + `lock_native` + `drain()`; other stream kinds go through `JSSink::<RewriterPipe>::assign_to_stream` (new `HTMLRewriterSink` codegen entry). Materialized bodies (`InternalBlob`/bytes `Blob`/`WTFStringImpl`) feed synchronously. - **Backpressure**: `RewriterPipe::write` feeds one chunk through `rewriter.write()` (output chunks push via `ByteStream::on_data`). If the output is paused or a handler suspended, `write` returns `Writable::Backpressure`; `ByteStream::resume` → `SourceHandle::HTMLRewriter::on_ready` → `pipe.resume()` drains `pending_input` then `input_source.ready()`. - **Async handlers**: `handler_callback` returns `HandlerOutcome::{Continue, Stop, Suspend}`. On a pending Promise it runs one microtask checkpoint (`process.nextTick` then promise jobs, never the loop); a genuinely pending Promise suspends. The JS wrapper is retargeted at the heap-parked token so post-`await` mutations land where they should. The `.then()` context is the Transform cell itself (the reactions recover the pipe via `from_js`), so a handler promise collected without settling abandons the parked rewrite instead of leaking it. - **Error handling**: `handler_error` on the pipe replaces the stack `Cell` + `unhandled_pending_rejection_to_capture` override. A handler error on a streaming input now rejects the body with the real error instead of `The rewriter has been stopped.`. - `AttributeIterator` holds a backref to the `Element` plus an index instead of a boxed `slice::Iter`, so `for (const [k, v] of el.attributes) { await ... }` keeps working across a suspension. ### Deletions `ValueBufferer` (~430 lines of `Body.rs`) and its host-fn exports; `SinkHandle::ValueBufferer` + `SinkWriteFn`; `JSSink<ArrayBufferSink>::detach_self`; `crate::Error::{StreamAlreadyUsed, InvalidStream, UnsupportedStreamType}`; `NativePromiseContext::Tag::BodyValueBufferer` (ordinal 4 reused for `HTMLRewriterSuspension`); the two `Bun__BodyValueBufferer__*` `PromiseFunctions` (slots reused for `Bun__HTMLRewriter__onHandler{Resolve,Reject}`). ### Behavior changes 1. **Error channel is decided by the overload, not by timing.** `transform(string)` / `transform(ArrayBuffer)` throw from `transform()`. Every `Response` input rejects its output body instead. Five existing tests that pinned the old timing-dependent split are updated. Input-body errors (already-failed or aborted body) still throw synchronously from `transform()`. 2. A handler whose Promise needs the event loop to turn makes `transform(string)` / `transform(ArrayBuffer)` throw a `TypeError` (`pass a Response and await its body`) instead of spinning. A Promise that settles within a microtask checkpoint still works. 3. A rejection a handler neither awaits nor returns reaches `unhandledRejection` instead of being captured and thrown from `transform()`. 4. `transform()` types corrected: `Bun.BufferSource` returns `ArrayBuffer` (it always has at runtime); `Blob` removed from the overload (it threw at runtime). 5. `Bun.serve` with an HTMLRewriter-produced response body defers status/headers until the first body byte or clean end, so a handler that fails before emitting any bytes is routed to the server'''s `error()` hook instead of committing `200 OK` then force-closing the connection. Headers never preceded the first byte on this path before either (the old implementation buffered the whole rewrite). All other native-ByteStream bodies (proxied `fetch()`, S3, spawn stdout) keep sending status/headers immediately, and JS `ReadableStream` bodies (`do_render_stream`) are unchanged. `docs/runtime/html-rewriter.mdx` and `packages/bun-types/html-rewriter.d.ts` cover 1-4. ### Verification - `test/js/workerd/html-rewriter.test.js`: 107 pass, 0 fail under ASAN debug and under `BUN_JSC_validateExceptionChecks=1`. New coverage: async element/text/comment/doctype/onEndTag/document-end handlers, `Bun.gc(true)` while an element is heap-parked across an `await`, re-suspension by a second handler on the same element, nested `transform()` inside a suspended handler, strict document-order across 8 awaiting handlers, `Bun.serve` with a live client, client abort mid-suspension, the consumer matrix for a pending output body, JS-backed and `type:'direct'` `ReadableStream` inputs. - `test/js/workerd/html-rewriter-leak.test.ts`: protected-object + `Response` count regression over 120 suspending rewrites; a never-settling handler rejects the body instead of leaking. - Fail-before: `works with payload of type direct` → `ERR_STREAM_CANNOT_PIPE` on the released bun. - `bun-types` green, `cargo clippy -p bun_runtime` clean. - `vendor/lolhtml/.ref` re-fetch of the fork verified through the real ninja edge. Fixes #11758 Fixes #14216 Fixes #6068 Fixes #19305 Closes #33243 (same lol-html fork, different input layer), Closes #35324 (ResumableSink, deleted in #36087), Closes #32988 (per-chunk ValueBufferer callback). <!-- robobun:evidence:begin --> --- **no test proof** · iteration 20 · Platform-specific test(s) that do not run on this machine. Deferring to CI, which covers all platforms: test/js/bun/io/bun-write.test.js test/js/node/process/process.test.js test/js/workerd/html-rewriter-leak.test.ts <!-- robobun:evidence:end -->
robobun
added a commit
that referenced
this pull request
Aug 4, 2026
…le backpressured Follow-up to #36087, carrying over the abort-path fix from the now-closed #35547 that the SinkHandle rewrite did not pick up. on_abort handles this.sink (the JS-stream sink path) but not this.byte_stream (the native SinkHandle::ServerResponse path). When a client aborts while the ByteStream sink is paused for backpressure, the upstream FetchTasklet stays in Paused with no wake path and no cancellation, and the ref taken when the sink was installed is never released (end_chunk only runs via sink.end(), which requires a drain that never comes on a closed socket). on_abort now calls cancel_from_sink() on the held ByteStream (detaches the sink and closes the producer, aborting the upstream fetch), deinits the Strong, and drops the sink-install ref. end_chunk clears this.byte_stream so the field tracks whether that ref is still held, keeping the two release sites mutually exclusive.
springmin
pushed a commit
to springmin/bun
that referenced
this pull request
Aug 4, 2026
…andlers instead of wait_for_promise (oven-sh#36733) `HTMLRewriter.transform()` now streams: input body chunks flow through `lol-html` into an output `ByteStream` as they arrive, with backpressure propagated end-to-end via the `SinkHandle`/`SourceHandle` pattern from ``` input body ──► SinkHandle::HTMLRewriter ──► lol_html::HtmlRewriter ──► output ByteStream ▲ │ └─────────── SourceHandle::HTMLRewriter (producer) ◄────────────────────┘ ``` `BufferOutputSink` fully buffered the source body via `ValueBufferer` before a single `rewriter.write()` + `end()`, and `handler_callback` spun `vm.wait_for_promise()` six native frames deep inside lol-html for any handler that returned a Promise. That meant handlers fired only at source-end (TTFB = full download), `.body` / `Bun.serve` saw an empty stream (oven-sh#6068, oven-sh#19305), JS-backed `ReadableStream` inputs were rejected outright (oven-sh#11758, oven-sh#14216), and the nested loop was a known hazard class (deadlocks, ready-poll clobbering, pending-exception leaks). lol-html could not be suspended before because its tokens are stack locals borrowing a stack-local lexeme; returning from `write()` destroys them, and an async handler must be able to mutate the element after its `await`. A handler returns `Err(SuspensionRequest)` to suspend. The in-flight unit is deep-copied onto the heap, `write()`/`end()` return the non-poisoning `RewritingError::Suspended`, and `HtmlRewriter::resume()` continues from a `StateMachineBookmark`. The pending captured-text flush is hoisted from `Dispatcher::handle_tag` into the lexer actions so every suspension point has a uniform shape. `Arena::shift` advances a start offset instead of memmoving the tail (a suspended rewrite re-feeds its unconsumed tail on every resume). 23 in-crate tests added; the new `.github/workflows/lolhtml.yml` runs the fork's own `cargo test` at the pinned commit, gated on `scripts/build/deps/lolhtml.ts`. The fork is a `github-archive` source (no patch file), so rebasing onto a new upstream tag is a `git rebase --onto` in the fork plus a commit bump here. - **Ownership**: a generated `HTMLRewriterTransform` JS cell (not user-visible) owns the pipe; its GC finalizer frees it, and nothing pins anything. Liveness is plain GC edges: the output Response's `transform` WriteBarrier slot and the `.then()` context of a suspended handler's (or the JS pump's) promise reach the cell, the cell's five slots root the Response, the input/output streams, the pending flush promise, and a captured handler error, and a wired native source's `owner` WriteBarrier slot (new on the generated NewSource cells) points back at the cell, so I/O that roots the source (a FetchTasklet, a FileReader's read refs, a reader on the output stream) roots the rewrite for exactly the window the raw `SinkHandle`/producer backrefs are wired. A handler promise collected without settling lets `finalize` defer to an event-loop task (`abandon_suspension`) that rejects the body before freeing; the pipe holds one native `+1` on the Response (released in `Drop`) so that task can still reach the body. No intrusive refcount, no `Strong` fields on the pipe; `finish()` frees the boxed lol-html state machine eagerly, and `fail()`/output-cancel close the upstream producer instead of draining it to EOF. - **Output**: the returned `Response`'s body is `Locked(PendingValue { task: pipe, on_start_streaming, on_readable_stream_available, producer: SourceHandle::HTMLRewriter })`, the `FetchTasklet::to_body_value` shape. The `ByteStream` is created lazily when a consumer reads `.body` or a body-mixin method; until then, rewriter output buffers in a `Vec<u8>` handed over as `DrainResult::Owned`. - **Input**: mirrors `FetchTasklet::start_request_stream`. Native `ByteStream`/`FileReader` sources wire `byte_stream.sink = SinkHandle::HTMLRewriter` + `lock_native` + `drain()`; other stream kinds go through `JSSink::<RewriterPipe>::assign_to_stream` (new `HTMLRewriterSink` codegen entry). Materialized bodies (`InternalBlob`/bytes `Blob`/`WTFStringImpl`) feed synchronously. - **Backpressure**: `RewriterPipe::write` feeds one chunk through `rewriter.write()` (output chunks push via `ByteStream::on_data`). If the output is paused or a handler suspended, `write` returns `Writable::Backpressure`; `ByteStream::resume` → `SourceHandle::HTMLRewriter::on_ready` → `pipe.resume()` drains `pending_input` then `input_source.ready()`. - **Async handlers**: `handler_callback` returns `HandlerOutcome::{Continue, Stop, Suspend}`. On a pending Promise it runs one microtask checkpoint (`process.nextTick` then promise jobs, never the loop); a genuinely pending Promise suspends. The JS wrapper is retargeted at the heap-parked token so post-`await` mutations land where they should. The `.then()` context is the Transform cell itself (the reactions recover the pipe via `from_js`), so a handler promise collected without settling abandons the parked rewrite instead of leaking it. - **Error handling**: `handler_error` on the pipe replaces the stack `Cell` + `unhandled_pending_rejection_to_capture` override. A handler error on a streaming input now rejects the body with the real error instead of `The rewriter has been stopped.`. - `AttributeIterator` holds a backref to the `Element` plus an index instead of a boxed `slice::Iter`, so `for (const [k, v] of el.attributes) { await ... }` keeps working across a suspension. `ValueBufferer` (~430 lines of `Body.rs`) and its host-fn exports; `SinkHandle::ValueBufferer` + `SinkWriteFn`; `JSSink<ArrayBufferSink>::detach_self`; `crate::Error::{StreamAlreadyUsed, InvalidStream, UnsupportedStreamType}`; `NativePromiseContext::Tag::BodyValueBufferer` (ordinal 4 reused for `HTMLRewriterSuspension`); the two `Bun__BodyValueBufferer__*` `PromiseFunctions` (slots reused for `Bun__HTMLRewriter__onHandler{Resolve,Reject}`). 1. **Error channel is decided by the overload, not by timing.** `transform(string)` / `transform(ArrayBuffer)` throw from `transform()`. Every `Response` input rejects its output body instead. Five existing tests that pinned the old timing-dependent split are updated. Input-body errors (already-failed or aborted body) still throw synchronously from `transform()`. 2. A handler whose Promise needs the event loop to turn makes `transform(string)` / `transform(ArrayBuffer)` throw a `TypeError` (`pass a Response and await its body`) instead of spinning. A Promise that settles within a microtask checkpoint still works. 3. A rejection a handler neither awaits nor returns reaches `unhandledRejection` instead of being captured and thrown from `transform()`. 4. `transform()` types corrected: `Bun.BufferSource` returns `ArrayBuffer` (it always has at runtime); `Blob` removed from the overload (it threw at runtime). 5. `Bun.serve` with an HTMLRewriter-produced response body defers status/headers until the first body byte or clean end, so a handler that fails before emitting any bytes is routed to the server'''s `error()` hook instead of committing `200 OK` then force-closing the connection. Headers never preceded the first byte on this path before either (the old implementation buffered the whole rewrite). All other native-ByteStream bodies (proxied `fetch()`, S3, spawn stdout) keep sending status/headers immediately, and JS `ReadableStream` bodies (`do_render_stream`) are unchanged. `docs/runtime/html-rewriter.mdx` and `packages/bun-types/html-rewriter.d.ts` cover 1-4. - `test/js/workerd/html-rewriter.test.js`: 107 pass, 0 fail under ASAN debug and under `BUN_JSC_validateExceptionChecks=1`. New coverage: async element/text/comment/doctype/onEndTag/document-end handlers, `Bun.gc(true)` while an element is heap-parked across an `await`, re-suspension by a second handler on the same element, nested `transform()` inside a suspended handler, strict document-order across 8 awaiting handlers, `Bun.serve` with a live client, client abort mid-suspension, the consumer matrix for a pending output body, JS-backed and `type:'direct'` `ReadableStream` inputs. - `test/js/workerd/html-rewriter-leak.test.ts`: protected-object + `Response` count regression over 120 suspending rewrites; a never-settling handler rejects the body instead of leaking. - Fail-before: `works with payload of type direct` → `ERR_STREAM_CANNOT_PIPE` on the released bun. - `bun-types` green, `cargo clippy -p bun_runtime` clean. - `vendor/lolhtml/.ref` re-fetch of the fork verified through the real ninja edge. Fixes oven-sh#11758 Fixes oven-sh#14216 Fixes oven-sh#6068 Fixes oven-sh#19305 Closes oven-sh#33243 (same lol-html fork, different input layer), Closes oven-sh#35324 (ResumableSink, deleted in oven-sh#36087), Closes oven-sh#32988 (per-chunk ValueBufferer callback). <!-- robobun:evidence:begin --> --- **no test proof** · iteration 20 · Platform-specific test(s) that do not run on this machine. Deferring to CI, which covers all platforms: test/js/bun/io/bun-write.test.js test/js/node/process/process.test.js test/js/workerd/html-rewriter-leak.test.ts <!-- robobun:evidence:end -->
This was referenced Aug 4, 2026
This was referenced Aug 8, 2026
This was referenced Aug 13, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Using a
node:streamReadableas afetch()request body wedges the process at 100% CPU when the Readable's_read()pushes synchronously: the event loop never runs again,destroy()never fires, and RSS climbs until OOM.Reproduction
Node/undici rejects after ~120ms. Bun spins forever; gdb on the main thread shows
ArrayBufferSink.writere-entered from JS indefinitely.Cause
ResumableSink(the sink fetch/S3 used for request-body streaming) was a parallel implementation of what theJsSinkType/generate-jssink.ts/assignToStreammachinery already does for every other sink. Its pump (assignStreamIntoResumableSink) materialised a DirectPending stream into anArrayBufferSink-backedJSDirectStreamControllerwhosewritereturns bytes-written (>= 0), sodriveAsyncIterator's only backpressure check (wrote < 0) never fires and a synchronous-yielding iterator spins forever.Fix
Delete
ResumableSinkand fold its two consumers into the JSSink family:FetchRequestBodySinkis a newJsSinkTypethat wraps theFetchTaskletback-ref.write_bytescarries the existing chunked-framing +ThreadSafeStreamBufferwrite, returningWritable::Backpressureonce the buffer reacheshighWaterMark;flush_from_js(true)returns a pending promise the HTTP-thread drain resolves, andSignal::ready()re-enters the pump.upload_stream()switches to the existingNetworkSink+assign_to_stream.NetworkSink::write_*now propagateMultiPartUpload's backpressure asWritable::Backpressure, andon_writablefiressignal.ready()on drain.Both callers go through
assign_to_stream, so a DirectPending body reachesreadDirectStream, which hands the sink controller straight topull().driveAsyncIterator's existingwrote < 0check fires, the pump suspends onflush(true)'s drain promise, and the event loop runs.Removed:
ResumableSink.rs,ResumableSink.classes.ts,JSResumableSinkPumpOperation,assignStreamIntoResumableSinkand itsresumable*pump,ReadRequestKind::ResumableSinkPump, and the associated handler/structure registrations. Net -460 lines.Verification
test/js/bun/http/async-iterator-stream.test.ts(91),test/js/web/fetch/body-stream.test.ts(9086),test/js/web/fetch/body.test.ts+fetch-abort-stream-body.test.ts(451), and thefetch.test.tsduplex block pass.Benchmark
Release build, 50 concurrent operations x 10 MB each (500 MB total per scenario), median of 3 runs, linux x64. Each scenario runs in a fresh subprocess; RSS delta is peak RSS minus baseline sampled every 50 ms; CPU is
process.cpuUsage()user+sys for the worker process (the echo server runs in a separate subprocess and is not counted).Builds: this PR
fd8b82482, main1.3.14+0d9b296af.fetch()response body intoBun.spawnstdin surfaces unhandledEPIPEfromreadStreamIntoSinkand the operation never completes at 50x concurrency. This PR handles it cleanly.SinkHandlepath keeps more data in flight (eachreq.body -> fetch bodyholds its chunked buffer until the echo response arrives), whereas main'sResumableSinkpath applied tighter backpressure earlier. Throughput is within ~15% either way and memory stays bounded below the 500 MB payload total per iteration, but this is a known trade-off worth noting.no test proof · iteration 17 · Platform-specific test(s) that do not run on this machine. Deferring to CI, which covers all platforms: test/js/bun/http/serve.test.ts test/js/bun/spawn/spawn-stdin-readable-stream.test.ts test/js/web/fetch/fetch.test.ts