Repository navigation
streams: settle a native sink's owner when a direct stream closes, not when pull() returns - #42957
Merged
Jarred-Sumner merged 3 commits intoSep 16, 2026
Merged
Conversation
…nd(), not when pull() returns readDirectStream() gave the sink's owner a promise derived from pull() when pull() was async, so the owner heard the end of the stream only when pull() settled. A pull() that kept running after its own end() or close() kept the owner waiting: Bun.serve kept the request counted (pendingRequests climbs, a graceful stop() never resolves, over HTTP/1.1 in particular), fetch() never finished the upload, and HTMLRewriter never finished the body. An async pull() now gets the same close promise a sync pull() already got. It settles when the controller closes, whoever closes it. The pull() reactions only close a controller that pull() left open, and settle the promise themselves so the owner still hears a failed end() and the rejection reason of pull(). The close promise is rejected as handled. The owner can close the sink before it subscribes (server.stop(true) in the microtask drain that follows the attach), and that rejection was reported as unhandled.
Contributor
|
Warning Review limit reached
On-demand reviews are free for the next 4 days. After that, they cost $0.25 per reviewed file. Or wait 7 minutes for your next included review. View limit detailsLimit details: You’ve used all 10 included reviews currently available. Review configuration: ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Essentials Run ID: 📒 Files selected for processing (9)
Comment |
Collaborator
Author
|
Status Reproduced on bun 1.4.3 and on a debug build of main a8e4e90 with this file: const server = Bun.serve({ port: 0, fetch: () => new Response(new ReadableStream({ type: "direct", async pull(c) { c.write("streamed"); await Promise.resolve(); c.end(); await new Promise(() => {}); } })) });
for (let i = 0; i < 3; i++) console.log(await (await fetch(server.url)).text(), server.pendingRequests);
await server.stop(); console.log("stopped", server.pendingRequests);
Fail-before proof on the debug ASAN build: PR: #42957 |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Nothing blocking. The comments below are optional suggestions. There is no need to push a fix for them before merging.
Beyond the inline finding, I also checked whether the close promise can be settled twice now that both directStreamOnClose and the two pull() reactions settle it: sinkControllerOnClose clears m_closePromise before end()/close() re-enter, so takeClosePromise in a later reaction sees null, and a reaction that closes the controller itself takes the promise first so the nested sinkControllerOnClose finds nothing. I also checked that the removed m_failReason checks in the reactions are covered: close(error) detaches synchronously, so by the time a reaction runs the owner already heard the error via the close path or via readDirectStream's rejected return.
Extended reasoning...
The one surviving finding (bytes()/arrayBuffer() consumers still waiting for pull() to return) is pre-existing and the description scopes it out to a follow-up. The other candidate concerns examined this run — a rejected async pull() after the sink pointer was cleared, swallowed close(error) when no owner subscribes, non-serve owners tearing down on backpressure, server.stop(true) invoking the error handler, and a failed implicit end() looking clean — were each ruled out by reading the ownership and settle ordering in src/jsc/bindings/webcore/streams/BunStreamSource.cpp. The change still reworks the lifecycle contract between direct streams and every native sink owner (Bun.serve, fetch upload, HTMLRewriter), so a human should weigh the behavioural shift (graceful stop no longer waiting for post-end() work) rather than this being an approve.
One verified lower-impact observation (a convention, logging or cleanup point) was not posted.
Windows takes the first write of a response whole, whatever its size, so one tryEnd() never parks there and the precondition of these two cases cannot hold.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Collaborator
Author
|
Updated 12:46 PM PT - Sep 16th, 2026
✅ @robobun, your commit bf3579800348808232c64a6add23a713f987b330 passed in 🧪 To try this PR locally: bunx bun-pr 42957That installs a local version of the PR into your bun-42957 --bun |
Jarred-Sumner
deleted the
robobun/0a7045f8/serve-h1-direct-stream-end-leak
branch
September 16, 2026 23:36
Jarred-Sumner
pushed a commit
that referenced
this pull request
Sep 17, 2026
…se(), not when pull() returns (#42950) ### Problem - On a `type: "direct"` ReadableStream, nine whole-body readers wait for `pull()` to settle, not for `controller.end()`/`close()`. A `pull()` that never returns after `end()` hangs them. `text()`, `blob()` and reader loops settle at `end()`. - A caught `close(error)` rejection exits the process with code 1. It happens when an async `pull()` awaits cleanup after `close(error)`, and when a sync `pull()` throws after it. New on main since #41894. - The cause is `consumeDirectStreamToArrayBuffer` (`src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp:1114`). `end()`/`close()` settle the capability promise. For an async `pull()` the caller got a second promise that adopted it only after `pull()` settled. ### Fix - Return the capability for an async `pull()`, as a sync `pull()` does. The pull reactions settle it when `pull()` settles first. A clean `end()`/`close()` then also closes and unlocks the stream. - A sync `pull()` that throws after `close(error)` still rejects with the throw. The rejected capability that nobody holds is now marked handled. - The native sink path (`readDirectStream`) had the same wait. #42957 fixed it, and this PR is rebased on it. - Verified: `test/js/web/streams/streams.test.js` (16 new tests, main at 1e31339 fails 14). Also `test/js/web/streams/`, `body*.test.ts`, `serve-direct-readable-stream.test.ts`, `html-rewriter.test.js`. ### Background - A direct stream has no queue. `pull()` gets a controller with `write()`, `end()` and `close()`. Per `docs/runtime/streams.mdx:76`, every destination finishes at `close()`/`end()`, and `close(error)` rejects the consumer. - The nine readers call `pull()` once with a one-shot controller (`JSOneShotDirectSink`) over an `ArrayBufferSink`. The capability promise carries the result. - Bun reports a rejected promise as unhandled if it has no handler when microtasks drain. No user reported this bug. <details><summary>Notes</summary> Demand: there is no issue and no user report. The bug was found by reading the code during work on the same wait in the native sink path. It needs a user-written direct stream whose async `pull()` keeps running after its own `end()`/`close()`, read through one of the nine readers. The failure is silent (a promise that never settles, or a false exit code 1), and it contradicts the documented contract. The nine readers: `bytes()` and `arrayBuffer()` on `Response` and `Request`, `ReadableStream.prototype.bytes()`, `Bun.readableStreamToBytes`, `Bun.readableStreamToArrayBuffer`, and `arrayBuffer`/`buffer` from `node:stream/consumers`. All nine hang on main and settle with this change. Repro for the hang: ```ts const make = () => new ReadableStream({ type: "direct", async pull(c) { c.write("hello"); await new Promise(r => setImmediate(r)); c.end(); await new Promise(() => {}); } } as any); await new Response(make()).text(); // settles await new Response(make()).bytes(); // never settles before this change ``` Repro for the exit code. On main it prints `caught: query failed`, then `error: query failed`, and exits with code 1. On 1.4.2 `close(error)` was a clean close, so nothing rejected. ```ts const body = new ReadableStream({ type: "direct", async pull(c) { try { c.write("partial"); await new Promise(r => setImmediate(r)); throw new Error("query failed"); } catch (e) { c.close(e); } finally { await new Promise(r => setImmediate(r)); } } } as any); try { await new Response(body).arrayBuffer(); } catch (e) { console.log("caught:", e.message); } ``` Before and after, `Bun.readableStreamToBytes(s)`: ``` async: end(), then never returns pending -> resolves async: close(error), then never returns pending + unhandled "boom" -> rejects "boom", nothing stray async: close(error), then returns rejects + unhandled "boom" -> rejects "boom", nothing stray async: close(), await, throw resolves, stream locked -> resolves, stream closed and unlocked async: end() then return resolves, closed, unlocked -> same sync: close(error), then throw rejects + unhandled "boom" -> rejects with the throw, nothing stray other sync shapes unchanged ``` Each added mechanism has a test that fails without it (checked by removing it and running the tests): - `m_insidePullCall` exists for one case: a sync `pull()` that calls `end()` and then throws. The caller rejects with the throw and the stream ends up errored, as before. Without the check `end()` closes the stream first. Test: `a sync pull() that throws after end() rejects with the throw and errors the stream`. - The fall-through in the async arm (`end()` before the first await) and the close in `oneShotDirectClose` (`end()` after an await). Test: `leaves the stream closed and unlocked once it settles`. The rejected reaction now fails the consumer whenever the result is still pending, not only when the sink is open. The one case where that differs: a source `close()` hook that throws out of `end()`, and `pull()` does not catch it. That hung before. It now rejects with the hook's error. The case where `pull()` catches the hook's throw still hangs, as on main. #41721 and #41472 each fix it. The contract matrix: #42957 added the three "never returns" shapes to `shapes` and skipped them for the three ArrayBuffer consumers (`waitsForPullToReturn`, `pullNeverReturns`). This PR removes both, so every consumer runs every shape. Those nine cells are new here and time out on main. The other seven new tests are in `direct stream edge cases`. The two sync `end()`-then-throw tests pass on main by design: they pin `m_insidePullCall`. An earlier revision of this PR had its own copies of these shapes, and the ASAN lane reported a leaked `FileSink` for the `Bun.write` and `Bun.spawn` stdin cells, because a native sink lived until `pull()` settled. #42957 removed that cause. The direct stream tests pass here with the ASAN lane's leak check on. A larger alternative is to drop the one-shot controller and read these bodies through `JSDirectStreamController`, as `text()` does, so that every consumer shares one close path. This PR does not do that. The one-shot path writes straight into the `ArrayBufferSink` without a reader, and this change is about 40 lines. The new contract shapes apply to either design. That fold is a separate decision for a maintainer. Overlap with open PRs in the same function: - #41721 adds `markPromiseAsHandled(vm, capability)` in the async arm, to hide the stray unhandled rejection. With this change the caller holds the capability, so that line is no longer needed there, and it would hide a rejection that the caller ignores. The test `close(error) is reported as unhandled only when the caller ignores the rejection, and then once` pins this. This PR also has the same line as #41721 in the sync throw path, after review feedback here. The rest of #41721 (the throwing `close()` hook, `FileSink`) is independent of this change and still needed. - #41472 reorders `oneShotDirectClose` (settle first, then the `close()` hook). It is independent of this change. Both edit the tail of `oneShotDirectClose`, so the second one to land needs a small rebase. Self-reviewed: 5 concerns raised, 5 addressed. No user report and the reach are stated. The larger alternative is named. The second site (`readDirectStream`) is fixed on main by #42957, and this PR removes the skip that #42957 left for these consumers. </details> <!-- robobun:evidence:begin --> --- **no test proof** · iteration 2 · platform-specific test(s) that do not run on this machine, deferring to CI, which covers all platforms: test/js/web/streams/streams.test.js <!-- robobun:evidence:end -->
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.
Problem
pull()callsend()from a later microtask and then never returns keeps its request forever.server.pendingRequestsclimbs and a gracefulserver.stop()never resolves. Bun.serve: end the request when an HTTP/2 or HTTP/3 stream closes after a streamed response #42844 fixed HTTP/2 and HTTP/3 only.readDirectStream(src/jsc/bindings/webcore/streams/BunStreamSource.cpp:758). For an asyncpull(), the sink's owner got a promise that settles whenpull()settles. A syncpull()got one that settles when the controller closes.fetch()upload andHTMLRewriternever finished.bytes()andarrayBuffer()wait in a different function (streams: settle bytes()/arrayBuffer() on a direct stream at end()/close(), not when pull() returns #42950).Fix
pull()now gets the same close promise. It settles atend(),close(),close(error), or a peer abort. Thepull()reactions only close a controller thatpull()left open.server.stop(true)in the microtask drain after the attach rejects it before the request subscribes. Main exits with code 1 there for a syncpull().end()andclose()detach the controller first. Nothingpull()does later reaches the sink.test/js/bun/http/serve-direct-readable-stream.test.ts(19 cases fail on main),test/js/workerd/html-rewriter.test.js(2). Other suites: Notes. Self-reviewed: 6 concerns raised, 6 addressed.Background
type: "direct") callspull(controller)once. The controller writes straight into a native sink, such as theBun.serveresponse sink.Bun.serve, theRequestContext) gets a promise fromassignToStream. When it settles, the owner tears the sink down and releases itself.end()andclose()detach the controller. ThendirectStreamOnClosesettles the close promise.Notes
pull()that outlives its ownend(). Repro, one file:streamed 1,streamed 2,streamed 3, thenstop()never resolves. This branch:streamed 0three times, thenstopped 0.fetch()upload hangs whenend()comes in the first tick ofpull()or later.HTMLRewriter.transform()hangs whenend()comes after an await.text(),blob(), reader loops,Bun.writeandBun.spawn({ stdin })already settle atend().consumeDirectStreamToArrayBuffer(BunStreamConsumers.cpp:1114) also returns a promise derived from an asyncpull(). SoResponse.bytes(),Response.arrayBuffer(),Bun.readableStreamToBytesandBun.readableStreamToArrayBufferstill wait forpull()to return. streams: settle bytes()/arrayBuffer() on a direct stream at end()/close(), not when pull() returns #42950 fixes that function. The contract matrix names those consumers inwaitsForPullToReturnand skips only the threepullNeverReturnsshapes for them. The PR that lands second removes that set.RequestContext, which the notes of Bun.serve: end the request when an HTTP/2 or HTTP/3 stream closes after a streamed response #42844 proposed. That fixes onlyBun.serve. The sink would call its owner from insideend_from_js(&mut self), and the owner frees the sink when its last ref goes, so the release needs a deferral. The close promise already is that deferred notification, and a syncpull()uses it today. Theon_abortarm from Bun.serve: end the request when an HTTP/2 or HTTP/3 stream closes after a streamed response #42844 stays: it covers a response that the sink ends from a native callback (onWritable), where HTTP/2 and HTTP/3 free the stream before the reaction runs.pull(). A cleanend()orclose(): the owner resolves at the call, not whenpull()settles.close(error): the owner rejects with the error at the call. A peer abort whilepull()is pending: the owner's promise rejects with the abort reason then, as it already did for a syncpull(). Before, it resolved whenpull()settled.Bun.servehas reclaimed its claim by then, andfetch()checks its aborted flag. Apull()that closes the controller in its first tick:readDirectStreamreturnsundefined, or a rejected and handled promise forclose(error), exactly as for a syncpull().pull()that resolves withoutclose()/end()ends the controller and then resolves the owner.pull()that rejects while the controller is open closes it with the reason and rejects the owner with the raw reason. Anend()that throws in that implicit close rejects the owner. A rejection after a clean close is dropped.stop()no longer waits for work thatpull()does afterend(). The response is complete at that point, and HTTP/2 and HTTP/3 already release the request there since Bun.serve: end the request when an HTTP/2 or HTTP/3 stream closes after a streamed response #42844.pendingWhileParked: 1whilepull()was parked afterclose(). They now pin 0. "stop(true) after the response completed still releases the parked request" opened the gate before it awaited the stop. It is now "... does not wait for the parked pull()" and samples the stop with the gate shut.live_resp()andended_responsestay. The window they guard (the sink ended the response, the context is still alive) is now one microtask checkpoint, not "untilpull()settles".readDirectStreamfor the aborted-request case and reverted it over the abort-then-reject path. Bun.serve: tear down an aborted request's context at abort instead of waiting for GC #39743 solved that case inon_abort.serve-stream-reject-flush-leak.test.ts, the test that change broke, passes here.end()in the same tick. That makes a singletryEnd()of 32 MB to a client that does not read, soend()parks on the drain. The request then waits for the drain only. These two cases are skipped on Windows. It takes the first write of a response whole, whatever its size (probed up to 250 MB on main), so onetryEnd()never parks there.serve-direct-readable-stream(169),html-rewriter(186),html-rewriter-leak,streams.test.js(596),serve-http2(93),serve-http3(72),serve-http2-lifecycle,serve-protocols,serve-stream-body-error,serve-error-handler-stream,serve-stream-reject-flush-leak,serve-pending-promise-abort-leak,serve-async-stream-client-abort,serve-response-gc-backpressure-abort,serve-response-stream-sink-leak,serve-body-leak,serve-reused-response,async-iterator-stream,bun-server,body-stream(9086),body.test,fetch-abort-stream-body,body-stream-excess,fetch-gzip,fetch-http3-client,fetch-http3-adversarial,spawn-stdin-readable-stream,spawn.test,node-stream,web-stream-state,s3.test,bun-install-streaming-extract.serve.test,fetch.test,fetch.stream.test,bun-write.test,streams-leakandAsyncLocalStorage.testhave failures in this container. The same tests fail on a build of main, so they are not from this change.no test proof · iteration 1 · platform-specific test(s) that do not run on this machine, deferring to CI, which covers all platforms: test/js/workerd/html-rewriter.test.js, test/js/web/streams/streams.test.js, test/js/bun/http/serve-direct-readable-stream.test.ts