Repository navigation
Bun.serve direct streams: flush on controller.close(), end the response once, call cancel() only on a client abort - #41894
Conversation
|
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:
WalkthroughChangesDirect stream lifecycle
Suggested reviewers: Priority: ⚪ Not assessed Severity of issue fixed: Low Merge Risk: 🟡 Moderate · up to A stream error during Bun.write or related piping can trigger unsafe FileSink lifetime handling and potentially crash the process. Add the local reference guard before merging. 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
|
Status: this PR now also carries #37696 and #41893, which reworked the same Reproduction, with the released binary: // bun repro.mjs
import net from "node:net";
const ev = [];
using server = Bun.serve({ port: 0, fetch(req) {
const path = new URL(req.url).pathname;
if (path === "/plain") return new Response("ok");
if (path === "/close") return new Response(new ReadableStream({ type: "direct",
pull(c) { c.write("hello"); c.close(); }, cancel(r) { ev.push("close cancel(" + r + ")"); } }));
return new Response(new ReadableStream({ type: "direct",
async pull(c) { c.write("hello"); await c.flush(); await new Promise(r => setImmediate(r)); c.write("-world"); c.close(); } }));
} });
ev.push("close body=" + await (await fetch(`http://127.0.0.1:${server.port}/close`)).text());
// One keep-alive connection: the chunked /late response, then /plain.
const raw = await new Promise(resolve => {
let data = "", second = false;
const sock = net.connect(server.port, "127.0.0.1", () => sock.write("GET /late HTTP/1.1\r\nHost: x\r\n\r\n"));
sock.on("data", d => { data += d; if (!second && data.includes("\r\n0\r\n\r\n")) { second = true; sock.write("GET /plain HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"); } });
sock.on("close", () => resolve(data));
});
ev.push("late wire=" + JSON.stringify(raw.slice(raw.indexOf("\r\n\r\n") + 4, raw.indexOf("HTTP/1.1 200", 10))));
await Bun.sleep(50);
console.log(ev);Before: A client that disconnects mid-body still calls Test: #37696 and #41893 are now closed in favor of this PR. Their repros pass on a debug build of this branch (2123e40) and still fail on main at 85e9ddc. This PR also fixes #17175 ( |
|
Updated 9:22 PM PT - Sep 8th, 2026
✅ @Jarred-Sumner, your commit c70a379c8f599737900500cf53e65bb836fc54ce passed in 🧪 To try this PR locally: bunx bun-pr 41894That installs a local version of the PR into your bun-41894 --bun |
There was a problem hiding this comment.
I reviewed this PR and didn't find any bugs. Because it changes the user-observable cancel() contract for direct-stream response bodies, rewires HTTPServerWritable::end() onto the flushing path, and modifies two existing test expectations in a close path with UAF history and three other open PRs touching it, a human look is still worthwhile.
What was reviewed:
- The
sinkClosedthird argument threaded throughdetach()(false) vsJSSinkController__onClose(true) and consumed inreadDirectStreamCloseImpl— the two call sites cover the source-initiated vs sink-initiated split as described. HTTPServerWritable::end()now delegating toend_from_jsviaself.global_this.expect(...): traced the only remainingend()caller tojs_closewith an empty reason (Sink.rs:667), which runs afterstart()has setglobal_this; the previous defaultclose_with_errorcaller is now overridden tofail().SourceHandle::abort()mirrorsclose()'s exception guard and JSController dispatch; non-JSController variants fall through toclose(None)unchanged.- The two edited existing assertions (dropped
"cancel()"event, added microtask yield incloseThenPark) follow directly from the new semantics rather than weakening coverage.
Extended reasoning...
Overview
This PR changes when a type: "direct" ReadableStream's cancel() hook fires when used as a Bun.serve response body, and makes controller.close() flush buffered bytes. It touches four files: the JSSink codegen script (adds a third sinkClosed boolean to the onClose callback protocol), BunStreamSource.cpp (gates the underlyingSource.cancel() call on sinkClosed), streams.rs (rewires HTTPServerWritable::end() to end_from_js(), adds fail() for the errored path, adds SourceHandle::abort() to pass ConnectionClosed as the cancel reason, overrides close_with_error), and the test file (adds ~280 lines of new coverage plus edits two existing assertions).
Security risks
No direct security surface. The change is in HTTP response-body streaming lifecycle — no auth, crypto, or input parsing. The area does have a documented UAF history (the test file's own comments describe an ASAN-gated regression test for end() after a settled pull()), and this PR reroutes end() onto a path that calls finalize(), res.end(), and source.close() where the old end() (with buffered bytes) only set flags. The fail() path preserves the old no-flush behavior for errored sources. The abort() addition follows the existing pattern of copying source out before calling into JS to avoid holding a borrow across a potential free.
Level of scrutiny
High. This is a user-visible behavior change to a Web-adjacent API (cancel() semantics on direct streams), it modifies two existing test expectations (one drops an asserted event, one adds a yield to keep a timing window open), and the PR description itself names three other open PRs (#41721, #41757, #34854) touching the same close path that will need coordination. The end() → end_from_js() delegation introduces an .expect() on self.global_this; I traced the remaining call site (Sink.rs js_close line 667, reached only after start() populates it) and the default close_with_error is now overridden, so it looks sound — but a maintainer who knows every entry into HTTPServerWritable::end should confirm.
Other factors
Test coverage is thorough: a six-shape completion matrix asserting cancel() is not called, a direct/default client-disconnect test asserting the AbortError reason matches request.signal.reason, and a two-round keep-alive raw-socket test with a strict chunked/content-length parser to catch stray terminators or lost bytes. The tests await conditions rather than sleeping, use port: 0, and Buffer.alloc(n, fill) per repo conventions. The bug hunt exited on dry_streak with no findings and no ruled-out candidates. Given the behavioral change, the modified existing assertions, and the cross-PR coordination the author flags, a maintainer sign-off is appropriate before merge.
|
On the
The message is the one the existing |
|
CI on build 112496: no failure is new for this change.
|
|
I worked the backpressure variant of the
|
…ody consumers end the stream when an async pull() resolves
…sink owner's promise; each falsy close() argument runs as the first close(); fix a stale bun-write comment
1b065df to
11fe484
Compare
… it for the auto-flusher; the promise a sync close(error) returns is marked handled
|
One edge of the folded const rs = new ReadableStream({ type: "direct", pull(c) { c.write("x"); throw "not an Error"; } });
await Bun.write("/tmp/out.txt", new Response(rs));
// canary: rejects with "not an Error". main: never settles.A sync The rest checks out on main: the GC repro from #41895, #41897 and #41917 is clean under ASAN, the ten new |
… it ended it (#42115) ### Problem - A `Bun.serve` `type: "direct"` stream whose synchronous `pull()` ends the response in chunked mode and then throws in the same call writes the chunked last-chunk twice. The wire carries `a␍␊xxxxxxxxxx␍␊0␍␊␍␊0␍␊␍␊` and the 5 surplus bytes sit in front of the next keep-alive response. A strict client parses `0` as that response's status line (the node keep-alive agent reports `HPE_INVALID_CONSTANT`). - Three orderings hit it: `flush()` then `close()`, `flush()` then `end()`, and `close()` with more than the sink's high water mark buffered. The Content-Length path is unaffected. - The cause is the order of two arms in `do_render_stream` (`src/runtime/server/RequestContext.rs:2170`). The thrown error was read first, so the already-responded arm never ran. ### Fix - Check `resp.has_responded()` before the thrown-error arm. One arm now handles every stream that already ended the response, whether `pull()` returned or threw. - Correct because the body is complete on the wire before the throw. `handle_reject()` does nothing once the response has responded, which left the request looking unanswered, and the handler tail (`should_render_missing()` in `src/runtime/server/mod.rs:1015`) then ended it a second time. - A `pull()` that throws with an incomplete body is unchanged: the error is reported and the connection is reset. - The throw is swallowed after a clean end, which is what a `pull()` promise that rejects after `close()` already does (`jsWebStreamsHandler_onReadDirectStreamPullRejected`). Both orderings now agree. - Verified: `test/js/bun/http/serve-direct-readable-stream.test.ts` (3 new cases, all 3 fail without the diff). Also `test/js/web/streams/streams.test.js`, `serve-error-handler-stream`, `serve-stream-body-error`, `async-iterator-stream`, `serve-reused-response`, `serve-http2`, `serve-http2-lifecycle`, and the two stream-sink leak suites. ### Background - A `type: "direct"` stream writes through a native sink (`HTTPServerWritable`). `controller.close()` and `controller.end()` both reach `end_from_js()`, which flushes the tail, calls uWS `end()`, and records `ended_response`. - `do_render_stream` attaches the stream to that sink. A synchronous `pull()` runs to completion inside the attach call, so the response can be finished before the attach returns. - A synchronous throw in `pull()` comes back from `assignToStream` as an `Exception` cell, not as a rejected promise (`GlobalObject::assignToStream`). That is the arm that ran too early. - uWS `hasResponded()` is false only while `HTTP_RESPONSE_PENDING` is set. `markDone()` clears it when the terminating chunk or the last Content-Length byte goes out, so it is true exactly when nothing more may be written. <details><summary>Notes</summary> Wire bytes on `8f33aaa9` (debug build, shape `write(10) flush() close() throw`): ``` HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\na\r\nxxxxxxxxxx\r\n0\r\n\r\n0\r\n\r\nHTTP/1.1 200 OK...Content-Length: 2\r\n\r\nok ``` With the diff the first `0\r\n\r\n` is the only one and the second response starts right after it. The second end came from the request-handler tail, not from the stream code: `should_render_missing()` tests `resp != null`, not aborted, not marked complete, not marked pending. The sink ends the response through uWS directly, so none of those flags change, and `render_missing()` called `ctx.end("")`. In chunked mode `internalEnd` then writes `0\r\n\r\n` again. Order of events for the failing shape, from the debug log: ``` doRenderStream -> start(256) -> writeLatin1(10) -> flushFromJS -> endFromJS -> finalize responded returned an error <- the arm that ran first destroy() shouldRenderMissing has response not aborted not marked complete not marked pending end <- second 0\r\n\r\n ``` Shapes checked by hand against the debug build, before and after: | `pull()` | before | after | | --- | --- | --- | | `write, flush, close, throw` | two last-chunks | one | | `write(1000), close, throw` | two last-chunks | one | | `write, flush, end, throw` | two last-chunks | one | | `write, close, throw` (Content-Length) | one end, no bytes written twice | same | | `write, flush, close, reject` | one | one | | `throw` with nothing written | error reported, connection reset | same | | `write, flush, throw` | error reported, connection reset | same | `end()` plus a throw doubled on 1.4.2 too, so this is not a regression from #41894. There, `close()` plus a throw reset the connection instead, because `close()` dropped the buffered tail (bug 1 of #41894). #41894 routed `close()` onto the `end()` path, which already had this. The hoisted arm also deinits the response body's stream reference and calls `stream.done()`, which the thrown-error arm did not. `done()` is a no-op for a JS-backed direct stream (it only cancels native Blob, File and Bytes sources). Two failures in `test/js/bun/http/serve.test.ts` are environmental and fail the same way with the released binary: `should use correct error when using a root range port(#7187)` (the test runs as root) and `only serves /bun:info to loopback clients in development mode`. </details>
|
Follow-up for one path this change left open: a |
…ils or its output is cancelled (#42130) ### Problem - Since #41894 a `type: "direct"` input to `HTMLRewriter.transform()` no longer gets `cancel()` when the output reader cancels, a handler throws, or the client behind a transformed `Bun.serve` response disconnects. The source keeps producing, and its next `write()` throws instead of returning `0`. 1.4.3 called `cancel(undefined)`. - `RewriterPipe::detach_input_source` (`src/runtime/api/html_rewriter.rs`) only detached a JS controller. Since #41894 `detach()` does not call `cancel()` (it is also the source's own close). Only `JSSinkController__onClose`, the destination-went-away signal, does. ### Fix - On the failed and cancelled paths, `detach_input_source` closes a JS controller with a reason before the detach: the handler's error from `fail()`, an `AbortError` from `cancel_from_output()`. That is `JSSinkController__onClose`, so `cancel(reason)` runs once and later writes return `0`. - `cancel_from_output()` marks the pipe done before the detach, as `fail()` does, because closing the controller re-enters `end_from_js`. - `SourceHandle::cancel(reason)` is the arbitrary-reason form of `abort()`. - Verified: `test/js/workerd/html-rewriter.test.js` (4 new cases fail on 1.4.3 and on main), plus the suites in Notes. ### Background - `HTMLRewriter.transform(response)` pipes the input body into a `RewriterPipe` sink. A JS stream input goes through a JSSink controller, which a direct stream's `pull()` writes into. - The controller has two close signals: `detach()` (the source closed, or the owner released a finished sink) and `JSSinkController__onClose(reason)` (the destination went away). Only the second runs the source's `cancel(reason)` and makes later `write()`/`flush()` return `0`. - A default `ReadableStream` input is pumped by `readStreamIntoSink`, whose close handler cancels the stream on either signal, before with `undefined`, now with the reason. <details><summary>Notes</summary> The review of #41894 flagged this path after the merge. Checked on main at e3b6d63 with a debug build: for `reader.cancel()` on the output, a throwing handler, and a client disconnect on a `Bun.serve` response that serves `rewriter.transform(directStreamResponse)`, the direct source's `cancel()` never ran and the `pull()` that resumed afterwards threw the "already been closed" message from its next `write()`. With this change each case records one `cancel(reason)` and the next `write()` returns `0`, the same contract #41894 gave the HTTP sink for a peer that went away. The reason for a cancelled output is `CommonAbortReason::UserAbort` ("The operation was aborted."). The JS reason passed to `reader.cancel(reason)` does not travel through the output `ByteStream`'s native cancel, and the output already hands the same `AbortError` to a native sink wired to it (`ByteStream::on_cancel`). A client disconnect reaches the pipe through that same `ByteStream` cancel, so it also reads as `UserAbort` rather than `ConnectionClosed`. Re-entrancy: closing the controller runs `directStreamOnClose` / `pumpOnClose`, which call the controller's `end()`, which calls `RewriterPipe::end_from_js` → `end_from_stream(None)`. Both callers set `done` first, so that pass is a no-op apart from an idempotent `release_input_roots(None)`, and the caller's `PipePin` keeps the pipe alive across the controller's `release_pump_ref`. The same `end()` re-entry already happened through `detach()` before this change; only the `cancel(reason)` call and `m_destinationClosed` are new. `fail()` now materializes the error's JS value before the output receives it; `ValueError::to_js` caches it, so the output and the source's `cancel()` see the same object. Suites run locally on the debug ASAN build: `test/js/workerd/html-rewriter.test.js` (184/184, also with `BUN_JSC_validateExceptionChecks=1` for the direct-stream cases), `html-rewriter-leak.test.ts`, `html-rewriter-end-error.test.ts`, `test/js/web/streams/streams.test.js`, `test/js/bun/http/serve-direct-readable-stream.test.ts`, `test/js/bun/stream/direct-readable-stream.test.tsx`, `test/js/bun/http/serve-stream-body-error.test.ts`. </details>
…irect stream's pull() (#42146) ### Problem - A `type: "direct"` `ReadableStream` read from JavaScript delivers bytes written after `end()`. `pull(c) { c.write("A"); c.end(); c.write("LATE") }` reads as `"ALATE"` and the late `write()` returns `4`. In the same position `error(e)` fails the stream, and `close(error)` turns the clean close into a failure, which 1.4.3 did not. - The cause is the deferred close in `JSDirectStreamController::onClose` (`src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp:856`). An `end()`/`close()` made while `pull()` is on the stack is recorded in `m_deferClose` and runs when `pull()` returns. `m_closed` stays unset until then, and the five controller methods (`:1128` on) checked only `m_closed`. ### Fix - Add `sourceEnded()` (`m_closed || m_deferClose == 1`) and gate `write`, `end`, `close`, `flush`, `error` on it. `write()` returns `0`, the rest return `undefined`. - Correct because the same calls already do this once `pull()` has yielded, and the native sinks (`Bun.serve`, `Bun.write`, `Bun.spawn` stdin) already deliver only what was written before `end()`. The deferral is unchanged. - Verified: `test/js/web/streams/streams.test.js` (two new contract shapes across 14 consumers plus two reader-path tests, 20 of the 30 fail on main). Other suites in Notes. ### Background - When JavaScript consumes a direct stream (`getReader()`, `for await`, `.text()`), its controller is `JSDirectStreamController`. It buffers `write()`s and hands them to reads as `Uint8Array`s. - `onPull` runs the user's `pull()`, then registers the waiting read. A `close()` inside `pull()` is deferred past that registration so the final chunk has a read to land in. - Native sinks (`HTTPResponseSink`, `FileSink`) are a different controller: `end()` completes at once and a later `write()` throws, on purpose (#41894). <details><summary>Notes</summary> Shapes checked by hand on a debug build of `e3b6d63dd7`, JS reader and `.text()` consumers, before and after: | `pull()` (synchronous) | before | after | | --- | --- | --- | | `write("A"), end(), write("LATE")` | `"ALATE"`, write returns 4 | `"A"`, write returns 0 | | `write("A"), close(), write("LATE")` | `"ALATE"` | `"A"` | | `write("A"), end(), flush()` | `"A"` | `"A"` | | `write("A"), end(), error(x)` | rejects with `x` | `"A"` | | `write("A"), end(), close(err)` | rejects with `err` (1.4.3: `"A"`) | `"A"` | | async: `write("A"), end(), write("LATE"), await` | `"ALATE"` | `"A"` | | async: `write("A"), await, end(), write("LATE")` | `"A"`, write returns 0 | same | The same shapes through `Bun.write(file)`, `Bun.spawn({ stdin })` and `Bun.serve` already produced `"hello world"` before the change (late `write()`/`flush()` throw there, `end()`/`close(err)` after close are no-ops), so the new contract cells in `serve-direct-readable-stream.test.ts` pass both ways. `.bytes()`, `readableStreamToBytes` and `readableStreamToArrayBuffer` also passed before: they consume through the native `ArrayBufferSink`, not `JSDirectStreamController`. The `close(err)`-after-`end()` case changed with #41894: before it, `close(error)` on the reader path was a clean close, so the overwritten deferred reason made no difference. After it, the second call replaced the deferred `undefined` reason with the error and the stream failed. **Intentionally not changed (same input family, other readers of the deferred state)** `m_deferClose` / `m_deferFlush` are `int8_t` tri-states: `-1` while the `pull()` body is on the stack, `1` once a close / flush was requested inside it. Requesting one overwrites the `-1`, so the readers that ask "is the `pull()` body still running?" answer wrong after a `close()` or `flush()` inside that body. This PR only changes the five controller methods. The rest stays as is, each verified on the debug build: - A re-entrant `reader.read()` issued inside `pull()` after `close()` passes the re-entrancy guard (`onPull`, `m_deferClose == -1`) and takes the final chunk; the read that triggered the `pull()` resolves `done`. With a close that is not deferred the triggering read gets the chunk. Fixing it changes what a refused re-entrant read resolves with, which is a pump change with its own tests. - `reader.cancel()` inside `pull()` after `close()` runs the source's `cancel()` hook and not its `close()` hook (`cancelSteps` does not look at the deferred close). The chunk is dropped either way, because the cancel runs before the triggering read is registered. - `flush()` then `write()` inside a synchronous `pull()` arms an end-of-tick flush job (`armEndOfTickFlush` tests `m_deferFlush == -1`), and `~StagedBytesScope` syncs the reported capacity mid-pull. Both are redundant there (`onPull` drains as soon as `pull()` returns), not wrong. - A synchronous `pull()` that calls `end()` and then throws still errors the stream on the reader path (the throw is handled before the deferred close is replayed). `Bun.serve` completes the response in that case (#42115). Splitting the tri-states into `inPullBody` / `closeRequested` / `flushRequested` bits would make those readers right by construction; `sourceEnded()` becomes `m_closed || closeRequested` under that split. Left out to keep this diff to the reported behavior. Suites run locally on the debug ASAN build: `test/js/web/streams/streams.test.js` (549 pass), `test/js/bun/http/serve-direct-readable-stream.test.ts` (148 pass), `test/js/web/fetch/body-stream.test.ts`, `test/js/web/streams/readable-stream-terminal-barrier-release.test.ts`, `test/js/bun/spawn/spawn-stdin-readable-stream.test.ts`, `test/js/bun/io/bun-write.test.js`, `test/js/node/stream/node-stream.test.js`. </details> <!-- robobun:evidence:begin --> --- **no test proof** · iteration 0 · 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 -->
…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 -->
…w into it (#44862) ### Problem - When the consumer of `new Response(asyncIterable)` leaves, the iterator can stay open for good. `Bun.serve`, the client hangs up: `return() calls: 0; listeners left on the emitter: 5`. 1.4.2 returns all five. - The cancel hook picks `reason ? iterator.throw?.(reason) : iterator.return?.()` (`src/jsc/bindings/webcore/streams/BunAsyncIterableSource.cpp:462`). The error tail calls `iterator.throw?.(error)` (`:184`). Since #41456, #41757, #41894 and #42130 (none released) consumers send a reason. - An iterator with no `throw()` gets no call. A generator that catches the throw never runs `finally`. ### Fix - Every cancel calls `return()`, the rule of #13687 and `docs/guides/http/sse.mdx:19`. - The error tail follows `for await`: a failed write calls `return()`, a failed iterator gets no call. A write that fails at once takes this path too. - Correct because a body cannot catch `return()`. node and `ReadableStream.from()` never call `throw()`. - Verified: 120 new cells in 5 test files, 115 fail on main. Self-reviewed: 14 concerns raised, 13 addressed. ### Background - The body is a direct `ReadableStream` pumped from `iterator.next()`. The cancel hook runs when the consumer leaves, the error tail when the pump fails. - Considered `return()` only where `throw()` is missing: a generator that catches stays open. A reason-less close at `streams.rs:2015` misses `FileSink::on_close`. ### Downsides - Reverses #41757 (unreleased): `cancel(reason)` no longer throws into a generator. Released change: `catch` no longer sees a failed write, and an iterator with only `throw()` gets no call. - Not closed: a generator that never reaches a `yield` again. - Cost: a refused chunk takes one more reaction (14 to 15 cells). Per chunk: none. Release text: -1,280 B. <details><summary>Notes</summary> **Who sends a reason (all end in the same cancel hook)** | Producer | Since | Reason | | --- | --- | --- | | `HTTPServerWritable::abort` (`streams.rs:2015`): client hang-up, idle timeout, `server.stop(true)` | #41894 | connection-closed `AbortError` | | `FileSink::on_close`: a failed write on `Bun.spawn` stdin or `Bun.write` to a pipe or file | #41456 | the write error | | HTMLRewriter input cancel | #42130 | the handler error or an `AbortError` | | `reader.cancel(reason)`, `pipeTo` error or abort, `tee()`/`clone()`, `Readable.fromWeb().destroy(err)`, `fetch()` abort | #41757 | the caller's reason | The producers are not changed. Direct and default streams still get the reason in `cancel(reason)`. **Rulings this follows** - #13687 (in every release since 1.1.27) wrote the abort arm: "undefined === Abort. We don't want to throw here because it will almost inevitably become an uncatchable exception." - `docs/guides/http/sse.mdx:19`: "If the client disconnects, the generator's `finally` block runs so you can clean up." - #41894 lists "Async generator and `Response` bodies" under "Not changed". - #33020 and #33167 set the rule for pipes: a sink that closes under an async iterable returns the generator. #36087 made the pump wait on a FileSink write, which put the error tail in front of that close. **What it reverses** - #41757 stated "`cancel(reason)` throws the reason into the generator (as `Bun.serve` already does on abort)". At that commit `Bun.serve` called `cancel(undefined)`, which returned the generator. Its test `cancel(reason) on an async generator body throws the reason into the generator` now expects `finally` and no `catch`. - #41894's test `async generator body: client abort between yields throws the connection-closed error into the generator once, then finally` now expects `["finally"]`, which is what 1.4.2 gives. - The error tail has called `iterator.throw?.(error)` since 1.0.27. No existing test observed it. - Not reversed: what `return()` itself throws still replaces the body's error (#39913). A rejected `return()` promise does not. **Results** (debug ASAN build unless a build is named) | Check | main bd599f5 | this PR | 1.4.2 | | --- | --- | --- | --- | | The 8 ledger scripts at exit 0 | 0 of 8 | 8 of 8 | 7 of 8 | | Twelve ways a JS consumer leaves, iterator not returned | 7 of 12 | 0 of 12 | 12 of 12 | | A body that ends its own controller, then yields: shapes left open (of 4) | 2 | 0 | 0 | | `Bun.write` to `/dev/full`, 64 KiB chunks | pulls for ever | rejects `ENOSPC` after 1 pull, body closed | pulls for ever | node v26.3.0 returns the iterator in 12 of 12 ways and never calls `throw()`. New tests on release builds of main and of this PR: | File | main | this PR | | --- | --- | --- | | `test/js/web/fetch/body-async-iterator.test.ts` | 10 pass, 40 fail | 50 pass | | `test/js/bun/http/async-iterator-stream.test.ts` | 97 pass, 42 fail | 139 pass | | `test/js/workerd/html-rewriter.test.js` (new rows) | 10 fail | 10 pass | | `test/js/bun/spawn/spawn-stdin-readable-stream.test.ts` (new rows) | 10 fail | 10 pass | | `test/js/bun/io/bun-write.test.js` (new rows) | 14 fail | 14 pass | | `test/js/bun/http/serve-direct-readable-stream.test.ts` (rewritten pin) | 1 fail | 1 pass | The 5 new cells that pass on main are the `reader.cancel()` controls with no reason. Each cell runs five body shapes from `test/js/web/streams/async-iterable-body-shapes.ts`: an iterator with `next()` and `return()`, one with `throw()` too, a generator with try/catch/finally, a generator whose loop catches, and a node `Readable` with `autoDestroy: false`. **The two commits** - Commit 1 changes only the cancel hook. Built alone, the 8 ledger scripts exit 0. It leaves the end-then-yield route open for 2 of 4 shapes, and `/dev/full` with 64 KiB chunks still pulls for ever. - Commit 2 changes the error tail and adds the rejected-write branch. Both routes above need it. **Still open, for a maintainer** - A generator acts on `return()` at a `yield`. One that waits in an `await` is returned when that `await` settles (the same on 1.4.2). The docs paragraph says so and shows `req.signal`. - A generator that writes through the yielded controller in a loop and never yields again is not stopped. 1.4.2 stopped it because `write()` threw once the client left. Since #41894 `write()` returns 0. The question is which of the two a write through the yielded controller must do after the consumer left. - A `return()` that throws synchronously while `Bun.serve` aborts the body is an uncaught exception, as on 1.4.2. - Open PR #44181 adds three tests that expect a refused chunk's error in a generator's `catch` (`thrown: outOfMemory`). They need a `finally` flag so that neither PR turns main red. - Not run: Windows and macOS. The stdin rows are skipped on Windows because `fs.closeSync(0)` does nothing there. HTTPS, HTTP/2 and HTTP/3 share the abort path by reading, and only HTTP/1.1 was run. **Measurements** (release builds of main bd599f5 and of this PR) - Release text (`size`): 88,771,341 to 88,770,061 B (-1,280 B). Reaction handlers per global: 98 to 98. - Cells per chunk (`heapStats` slope, GC off, 2,000 to 6,000 chunks, 3 runs): 3.000 to 3.000 on `Bun.serve` and `.bytes()`. - Per client abort of a generator body (N=200): Promise 10 to 9, reactions 6 to 5. - `body.cancel(err)`: an iterator with `throw()` Promise 4 to 3. An iterator with `return()` only Promise 2 to 3, because `return()` is now called. - Error tail per failing body under `.text()` (N=1000): a generator that throws 13 to 9 cells and 4 to 3 microtask turns. A refused chunk with a generator 14 to 15 cells and 6 to 7 turns. A refused chunk with `{next, return}` 9 to 13 cells, because `return()` is now called. - End of a body whose `return()` gives a plain object and not a promise: 10 to 12 cells. The result is now awaited, so that a thenable or a lazy `Bun.sql` promise runs. - Static (`llvm-objdump`): the four iteration-protocol exception checks in the pump go from 3 instructions to 2. A `write()` that returns a promise gets one more status compare. - Not measured: dynamic instruction counts. `perf` hardware counters fail in the build VM and valgrind could not be installed. **Suites run** on the debug ASAN build: the six files above, `streams.test.js`, `body.test.ts`, `body-stream.test.ts`, `fetch-args.test.ts`, `serve-stream-body-error.test.ts`, and the two main files with `BUN_JSC_validateExceptionChecks=1`. In a CPU-capped container the subprocess tests of `serve-direct-readable-stream`, `spawn-stdin-readable-stream`, `bun-write`, `streams` and `serve-stream-body-error` time out at the default concurrency. All pass with `--max-concurrency 4 --timeout 60000`. Two older chunk-count tests in `async-iterator-stream.test.ts` (`async generator function`, `[Symbol.asyncIterator]`) fail in 2 of 6 debug runs with the new tests filtered out. **Self-review, the one concern not taken:** deleting the `lastResort` overload of `atStreamsBoundary`. The error tail still uses it, because what `return()` throws still replaces the error. </details> <!-- robobun:evidence:begin --> --- **no test proof** · iteration 0 · 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/bun/spawn/spawn-stdin-readable-stream.test.ts, test/js/bun/io/bun-write.test.js, test/js/bun/http/serve-direct-readable-stream.test.ts <!-- robobun:evidence:end -->
…led rejection (#41721) ### What does this PR do? Fixes a regression in Bun 1.4.3. When a direct stream's sync `pull()` calls `controller.close(error)` and then throws, `Bun.write()` rejects and the caller catches it, but the process also reports an unhandled rejection. ```js const s = new ReadableStream({ type: "direct", pull(c) { const e = new Error("source failed"); c.write("partial"); c.close(e); throw e; } }); try { await Bun.write("/tmp/out.txt", new Response(s)); } catch (e) { console.log("caught: " + e.message); } ``` | | output | exit code | | --- | --- | --- | | Bun 1.4.2 | `caught: source failed` | 0 | | Bun 1.4.3, main | `caught: source failed`, then an unhandled `error: source failed` | 1 | | this PR | `caught: source failed` | 0 | **Cause.** `FileSink::pipe_stream` creates a done-promise before it starts the pump. Since #41894, `close(error)` rejects that promise (`close_with_error` → `end` → `on_close` → `settle_stream_done`). When `pull()` then throws, `pipe_stream` returns the throw as `Err`, and `Blob.rs` makes a new rejected promise from it for the caller. Nobody ever receives the done-promise, so its rejection is reported as unhandled. **Fix.** On that `Err` path, `pipe_stream` marks the done-promise handled: `done_promise.set_handled()` in the `inspect_err` closure in `src/runtime/webcore/FileSink.rs`. That line is the whole fix. The caller still gets the throw as the rejection of `Bun.write()`. **Behavior after the fix** - After `close(A)` and `throw B`, `Bun.write()` rejects with `B`, the same as on main, on 1.4.2, and in `text()`, `bytes()` and `blob()`. Nothing reports `A`. - A caller that ignores the rejection gets one report (`B`). Main gives two (`A` and `B`). - A `pull()` that does not throw is not affected: the mark runs only on the `Err` path. - No other consumer shows this stray rejection. S3 uploads return the same promise they settle. `text()`, `bytes()`, `blob()`, `Bun.spawn({ stdin })` and `HTMLRewriter` were probed with the shapes below, and the `Bun.serve` and `fetch` body cells of the contract matrix pass. ### How did you verify your code works? Tests: - `test/js/web/streams/streams.test.js` - `close(error) from a sync pull() that closes and then throws: the caller's try/catch is the only report` has a new `Bun.write` row. - `Bun.write: close(error) from a sync pull() that then throws is reported once when the caller ignores the rejection` is new. It pins that the failure is not swallowed. - `test/js/web/streams/direct-stream-contract.ts` has a new shape, `sync pull: close(error), then throws it` (the repro above). It runs against every consumer in `streams.test.js` and in `test/js/bun/http/serve-direct-readable-stream.test.ts`. On a debug build with `src/` equal to main (491aaad), those tests give 2 failures and 1 `Unhandled error between tests`. With the fix: | file | result | | --- | --- | | `streams.test.js` | 727 pass, 2 skip, 0 fail | | `serve-direct-readable-stream.test.ts` | 173 pass, 0 fail | I also ran 15 `pull()` shapes under `Bun.write`, one process each, once with the rejection caught and once with it ignored, on debug builds of main and of this branch. They cover `close(x)` then a throw of the same error, another error, a string and `undefined`; without a write; after a 4 MB write; `close()` or `end()` then a throw; a throw alone; and the async forms. - 14 of the 30 rows differ, all of them `close(x)` followed by a sync throw. Caught: 1 stray report becomes 0. Ignored: 2 reports become 1. - The other 16 rows are identical, including the result and the bytes on disk. The repro is clean under `BUN_JSC_validateExceptionChecks=1`. --- _Generated by [Claude Code](https://claude.ai/code)_ --------- Co-authored-by: Claude <noreply@anthropic.com>
What does this PR do?
Fixes four bugs in
type: "direct"ReadableStreams and gives every whole-body consumer the same rule for when such a stream ends.Fixes #17175
Closes #41917
Bugs fixed
1.
controller.close()sentContent-Length: 0.A
pull()that wrote and calledclose()in the same tick answered200with an empty body.HTTPServerWritable::end()left the buffered tail to the auto-flusher, butdo_render_streamfrees the sink as soon as a syncpull()returns.Fix:
end()(whatcontroller.close()reaches) now takes theend_from_js()path. It flushes and ends the response, or parkspending_flushunder backpressure.close(error)goes tofail(), which drops what is buffered and finishes the sink without ending the response; the owner then closes it as incomplete.2. The chunked last-chunk was written twice.
When
finalize()sent a parked tail,res.end(tail)wrote0\r\n\r\nwithout recordingended_response, soend_stream()wrote it again. The next keep-alive response then fails to parse (HPE_INVALID_CONSTANT).Fix:
ended_responseis set in one place,mark_response_ended(), at the point where uWSend()/try_end()completes. The other paths assert it.3.
cancel(undefined)ran after every completed response (#17175).The sink's
onClosecallback fires both when the source closes itself and when the peer goes away, and the handler could not tell them apart.Fix:
onClosecarriessinkClosed, true only fromJSSinkController__onClose.directStreamOnClosecalls the source'scancel(reason)only then.HTTPServerWritable::abortpassesCommonAbortReason::ConnectionClosedas the reason.4. An
async pull()that resolved withoutclose()ended differently per consumer.Bun.servefinished the response.Bun.write(file, new Response(stream))andBun.spawn({ stdin })closed the fd and dropped the unflushed tail..text()/.bytes()/Bun.readableStreamTo*()never settled.Bun.writealso freed theFileSinkwith its controller still attached, a use-after-free at the next GC.Fix: one rule, below.
FileSinkends with the flushingend(), andBun.writepipes JS streams throughFileSink::pipe_stream, which detaches the controller before the sink is freed.The rule
A consumer that takes the whole body (
Bun.serve,fetchrequest bodies,Bun.write,Bun.spawnstdin,.text()/.bytes()/Bun.readableStreamTo*()) callspull()once.close()/end().controller.close(error)fails the stream. Falsy arguments (undefined,0,"") are a clean close.end(x)is always a clean close.readDirectStream(native sinks) and the one-shot consumers (.text()etc., viam_closeOnPullSettled) follow it.The JS reader path (
getReader(),for await,pipeTo(),tee()) is unchanged:pull()is a demand signal there, called again for a later read once the previous call has settled (#33782). A reader has a demand signal, a whole-body consumer does not.Behavior changes
async pull()that resolves withoutclose()ends the stream on every whole-body consumerBun.servealready did..text()etc. hung;Bun.writetruncated or crashedcontroller.close()flushes buffered bytes on the HTTP sinkpull()+close()sentContent-Length: 0controller.close(error)fails the stream everywhere: readers reject,Bun.serveaborts the response with no chunked terminator,fetch()with such a request body rejectscancel(reason)runs only when the consumer goes away. On a client disconnect the reason is the connection-closedAbortErrorclose()/end(), withundefinedcontroller.write()/flush()return0after the client, file, or child process went away"This HTTPResponseSink has already been closed…"pull()that rejects after the stream already closed cleanly, or after the peer left, is ignored.bytes()andfetch()request bodies surfaced it as a late rejectionNot changed
Bun.serveerror reporting for a body that fails after the status line is committed.error()stays bypassed (Bunch of streams fixes #4251), the failure prints to stderr, and the production async path stays quiet.controller.write()/flush()on a native sink still throw the "already been closed… onceasync pull()returns" message after the source closed the stream itself. That message is what catches apull()that forgot toawait.pull()again per read (never while a previous call is pending).write()after close returns0there, as before.Bun.ArrayBufferSinkandBun.file().writer()still throw afterend().Responsebodies. New tests pin the existing behavior.Refactors in this PR (no behavior change intended)
JSReadableSinkControllerBaseholds the source cell and the close promise itself.detach(),JSSinkController__onClose, andJSSinkController__onReadycallBun::WebStreams::sinkControllerOnClose/OnReadydirectly, andend()dispatches onSinkID. TheJSDirectSinkCloseStatecell and threeJSBoundFunctionhandlers are deleted. Every consumer of those handlers was C++.Blob/File/Bytes) notify their C++ adapter throughBun__NativeStreamSourceAdapter__onCloseinstead of a bound function stored through a JS setter.handle.onDrainhad no caller and is deleted, along withclose_handler/close_ctx.Blob.rsFileStreamWrapperis deleted.FileSink::pipe_streamcovers that path.How did you verify your code works?
test/js/web/streams/direct-stream-contract.tsis a source-shapes × consumers matrix. Each cell asserts thatpull()ran once, the bytes delivered, the error, and thatcancel()ran only on a consumer abort.pull()returns or resolves without closing are markedoneShotOnlyand skip the reader consumers, which would callpull()again.streams.test.js:.text()/.bytes()/.blob(),readableStreamTo*,getReader(),for await,pipeTo,pipeThrough,tee,Bun.write,Bun.spawnstdin.serve-direct-readable-stream.test.ts:Bun.serveHTTP and HTTPS bodies,fetchrequest bodies.serve-direct-readable-stream.test.tsalso covers theclose()/end()matrix over HTTP, HTTPS, and HTTP/3, the last-chunk orderings,close()under transport backpressure, keep-alive afterclose(), and client-abort races.serve-direct-readable-stream.test.tsfail: 6 syncclose()cases over HTTP and HTTPS, 2 over HTTP/3, 7 last-chunk orderings, 4 backpressure cases, 8cancel()cases, the keep-aliveclose()test, and 2stop(true) after end()expectations.bun-write.test.jsandspawn-stdin-readable-stream.test.tscover theFileSinktruncation and the GC-after-free case.streams.test.js,bun-write.test.js,spawn-stdin-readable-stream.test.ts. TheBun.servesuites run in CI.Notes
Folded PRs
close()with a buffered tail sends it at once. Same change as bug 1. Its HTTP / HTTPS matrix and two HTTP/3 cases are in the test file. Closed in favor of this PR.ended_responsewhere the uWS end happens. Same as bug 2. Its seven orderings are in the test file, plus aclose(error)ordering that pins only the framing. Closed in favor of this PR.Two existing expectations change
server.stop(true) from inside pull() after end()no longer records acancel()event.closeThenParkfixture yields once beforeclose(), because a synchronousclose()now completes the request inside the dispatch, as a synchronousend()already did.Follow-on from that
When a synchronous
close()inside anasync pull()completes the response beforepull()first yields,do_render_streamtakes itshas_responded()path and releases the request whilepull()keeps running.end()already behaved this way. Apull()that throws after that point is an unhandled rejection. Holding the request untilpull()settles would keep the context alive for as long aspull()runs, so it is left out.Implementation notes
fail()never sends the buffered tail. The previousend()body parked it withrequested_endset, and the auto-flusher (whichdo_render_stream'sdrain_microtasks()runs) then sent it withres.end(), completing a failed body. A syncpull(c){ c.write(x); c.close(err) }hit that.pull()that callsclose(error)makesreadDirectStreamreturn a promise rejected with the error (marked handled: every caller is a native owner that reads its state), so owners take the same reject path as the async shape.JsSinkType::endis reached only fromjs_closewith an empty reason, so itserrargument was alwaysNonehere.global_thisis set whendo_render_streamconstructs the sink, beforeassign_to_streamruns.pending_flushfrom a syncclose()under backpressure is picked up bydo_render_streamandhandle_resolve_stream, the same way a syncend()already worked over HTTP/3.no test proof · iteration 3 · 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, test/js/bun/spawn/spawn-stdin-readable-stream.test.ts, test/js/bun/io/bun-write.test.js