Repository navigation
fix(stream): expose web stream state to node:stream - #42416
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Advanced Run ID: 📒 Files selected for processing (2)
Included review availability: Your plan provides up to 10 included reviews per hour; 9 remain after this review. WalkthroughThe change adds three symbol-keyed ChangesReadableStream state introspection
Suggested reviewers: Priority: ➖ Normal Merge Risk: ⚪ Minimal · up to The new stream-state accessors match the tested lifecycle and locked-stream behavior, with no remaining merge-blocking risk identified. 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
|
Thank you for this fix. It overlaps with #34624, an earlier attempt at the same bug. That PR added a private intrinsic and special cases inside the four I built this branch and compared it with Node v26.3.0.
The property descriptors also match Node ( One gap:
|
|
The |
…ng it (#42516) ### Problem - `const body = res.body; await res.arrayBuffer()` (or `bytes()`, `blob()`, `json()`, `formData()`) leaves `body` `readable` when a string, `Blob`, typed array or file backs it. `isReadable(body)` returns `true`, `Bun.inspect(body)` prints `state: 'readable'`, and `await finished(body)` never settles. Node v26.3.0 and `text()` close the stream. - The cause is `ReadableStream::to_any_blob` (`src/runtime/webcore/ReadableStream.rs:195`). It takes the payload out of the native source and marks the stream disturbed and locked, but never changes `m_state`. Since #42116 (not released) the stream stays locked, so nothing can drain it either. - `.body` read after the consume returns a stand-in stream that is `readable` and not disturbed. ### Fix - `to_any_blob` calls the new `ReadableStream__closeConsumedAsBody`: the `markConsumedAsBody` flags, then a close. - Correct because `to_any_blob` first rejects a disturbed or locked stream. No reader or controller exists, so the close only sets the state and settles the closed promise. - `ReadableStream__used` makes the stand-in for a consumed body closed and disturbed. The stand-in for a body still being read is unchanged. - Verified: `test/js/web/fetch/body.test.ts` (99 new cases, 50 fail on main), plus the suites in Notes. Self-reviewed: 3 changes asked, 3 made. ### Background - Such a body gets a lazy native `ReadableStream`. It gets a controller only when something reads it. - Until then a consumer can take the whole payload back as a `Blob`. The body methods, `clone()`, `Bun.serve`, `fetch()` uploads, `Bun.write` and `Bun.spawn` stdin do so when the payload is in memory or is a file. - #42416 made `isReadable()`, `isErrored()` and `isDisturbed()` read the stream's own state. `finished()` waits on a closed promise that each terminal transition settles. <details><summary>Notes</summary> **Origin.** A differential run against node v26.3.0 after #42416 found this. No user reported it. On 1.4.2 the stream was also left `readable`, but it was unlocked, so `body.getReader().read()` returned `done` and closed it. #42116 keeps the stream locked (as undici and Chromium do), which removed that way out. This PR should land before the next release. **Repro.** ```js import { isReadable, isDisturbed } from "node:stream"; import { finished } from "node:stream/promises"; const res = new Response("hello world"); const body = res.body; await res.arrayBuffer(); console.log(isReadable(body), isDisturbed(body), body.locked, Bun.inspect(body)); await finished(body); // main: never settles ``` | | main b993710 | this PR | node v26.3.0 | | --- | --- | --- | --- | | `.body` then `arrayBuffer()` / `bytes()` / `blob()` / `json()` / `formData()` | readable, `finished()` pending | closed | closed | | `.body` then `text()` | closed | closed | closed | | `new Response(blob.stream()).bytes()`, `Bun.file().stream()`, `new Response(Bun.file()).body` | readable | closed | closed | | `fetch()` response, `.body` taken, whole body arrived, then `arrayBuffer()` | readable | closed | closed | | `Bun.serve` sends it, `fetch()` uploads it, `Bun.write` writes it, `Bun.spawn` stdin (payload in memory or a file) | readable | closed | n/a | | `await res.text(); res.body` (stand-in) | readable, not disturbed | closed, disturbed | closed, disturbed | | old `.body` after `clone()` | readable forever | closed at `clone()` | readable until a branch is drained, then closed | | JS-source stream | closed | closed | closed | The error messages of `getReader()`, `cancel()`, `tee()`, `pipeTo()`, `text()`, `Bun.readableStreamToText()` and `new Response(body)` on the consumed stream are the same as before. `finished()` registered before the consume settles too. **Not in this PR.** - A stream that a native sink attaches to while the payload still arrives (`ReadableStream__lockNative`: a `fetch()` upload, `Bun.write`, `Bun.spawn` stdin or `Bun.serve` response of a body that still streams). It does not go through `to_any_blob`. #42477 fixes it. This PR should land first. The conflict in `WebStreamsExports.cpp` is keep-both, and both use `readableStreamCloseIfPossible` for the close. - `Readable.fromWeb()` on a native stream (`m_transferred`): the web stream stays `readable` after the node stream ends. It is the same on 1.4.2. The fix touches the code that #42281 changes, so it follows that PR. - The `node-fetch` and `undici` shims wrap `res.body` in a node `Readable`. Since #42116, `res.body.resume()` or `destroy()` after a body method emits `'error'` (the stream is locked). 1.4.2 emits `'end'` / `'close'`. This PR does not change that. It is tracked separately. - `.body` read while `text()` is still pending returns a stand-in that stays `readable` and locked. Nothing tells it how the read ends, so it must not report closed. A new test pins this. - `getReader().releaseLock()` marks a native stream disturbed: #33461. `WritableStream` state: #42457. **The stand-in for a consumed body** is also what `.body` returns once a native consumer took a body that still streams (`Bun.write(path, res)` sets the body to used at once). It reads closed from then on. Before, `finished()` on it never settled. **Self-review.** It asked for three changes. The stand-in for a body still being read keeps its old state (`ReadableStream::in_use`). The close goes through `readableStreamCloseIfPossible`, not a hand-written state write. The commit message and this body name the excluded paths and say that the native consumers take this path only when the payload is in memory or is a file. **Suites run** with the debug build and `BUN_JSC_validateExceptionChecks=1`: `test/js/web/fetch/body.test.ts` (734 pass), `body-stream.test.ts` (9086 pass), `body-clone.test.ts`, `body-mixin-errors.test.ts`, `blob.test.ts`, `fetch.stream.test.ts`, `test/js/web/streams/streams.test.js` (549 pass), `readable-stream-blob-consumed.test.ts`, `test/js/node/stream/web-stream-state.test.ts`, `test/js/third_party/wpt-streams/wpt-streams.test.ts` (1175 pass), `test/js/bun/spawn/spawn-stdin-readable-stream.test.ts`, `test/js/bun/io/bun-write.test.js`, `test/js/bun/util/inspect.test.js`. </details> <!-- robobun:evidence:begin --> --- **[human-review]** gate passed · iteration 1 · 5 files touched <details><summary>fails on main (without fix)</summary> ```console ASAN without fix: 50 failed, 4 skipped $ BUN_DEBUG_QUIET_LOGS=1 bun scripts/build.ts --profile=debug --quiet test "--reporter=junit" "--reporter-outfile=/tmp/pr_gate.xml" test/js/web/fetch/body.test.ts bun test v1.4.3 (6a92015) test/js/web/fetch/body.test.ts: (pass) Request > constructor > undefined [6.84ms] (pass) Request > constructor > null [3.58ms] (pass) Request > constructor > string > "" [6.59ms] (pass) Request > constructor > string > "Hello world" [1.39ms] (pass) Request > constructor > string > "🫠" [1.30ms] (pass) Request > constructor > string > "⁉️ " [1.21ms] (pass) Request > constructor > ArrayBuffer > empty buffer [11.00ms] (pass) Request > constructor > ArrayBuffer > small buffer [5.11ms] (pass) Request > constructor > ArrayBuffer > large buffer [4.31ms] (pass) Request > constructor > SharedArrayBuffer > empty buffer [1.90ms] (pass) Request > constructor > SharedArrayBuffer > small buffer [1.58ms] (pass) Request > constructor > SharedArrayBuffer > large buffer [8.91ms] (pass) Request > constructor > Buffer > empty buffer [1.49ms] (pass) Request > constructor > Buffer > small buffer [12.96ms] (pass) Request > constructor > Buffer > large buffer [3.37ms] (pass) Request > constructor ... (truncated) release without fix: 4 skipped bun test v1.4.3-canary.1 (2c1543d) test/js/web/fetch/body.test.ts: (pass) Request > constructor > undefined [0.08ms] (pass) Request > constructor > null [0.05ms] (pass) Request > constructor > string > "" [0.12ms] (pass) Request > constructor > string > "Hello world" [0.01ms] (pass) Request > constructor > string > "🫠" [0.02ms] (pass) Request > constructor > string > "⁉️ " (pass) Request > constructor > ArrayBuffer > empty buffer [0.14ms] (pass) Request > constructor > ArrayBuffer > small buffer [1.40ms] (pass) Request > constructor > ArrayBuffer > large buffer [1.62ms] (pass) Request > constructor > SharedArrayBuffer > empty buffer [0.03ms] (pass) Request > constructor > SharedArrayBuffer > small buffer [0.02ms] (pass) Request > constructor > SharedArrayBuffer > large buffer [2.61ms] (pass) Request > constructor > Buffer > empty buffer [0.61ms] (pass) Request > constructor > Buffer > small buffer [0.02ms] (pass) Request > constructor > Buffer > large buffer [0.47ms] (pass) Request > constructor > Uint8Array > empty buffer [0.02ms] (pass) Request > constructor > Uint8Array > small buffer [0.03ms] (pass) Request > constructor > Uint8Array > large buffer [1.6 ... (truncated) ``` </details> <details><summary>passes on PR (with fix)</summary> ```console ASAN with fix: 4 skipped $ BUN_DEBUG_QUIET_LOGS=1 bun scripts/build.ts --profile=debug --quiet test "--reporter=junit" "--reporter-outfile=/tmp/pr_gate.xml" test/js/web/fetch/body.test.ts bun test v1.4.3 (6a92015) test/js/web/fetch/body.test.ts: (pass) Request > constructor > undefined [7.98ms] (pass) Request > constructor > null [3.44ms] (pass) Request > constructor > string > "" [6.99ms] (pass) Request > constructor > string > "Hello world" [1.36ms] (pass) Request > constructor > string > "🫠" [1.41ms] (pass) Request > constructor > string > "⁉️ " [1.13ms] (pass) Request > constructor > ArrayBuffer > empty buffer [10.53ms] (pass) Request > constructor > ArrayBuffer > small buffer [5.26ms] (pass) Request > constructor > ArrayBuffer > large buffer [4.52ms] (pass) Request > constructor > SharedArrayBuffer > empty buffer [1.81ms] (pass) Request > constructor > SharedArrayBuffer > small buffer [1.74ms] (pass) Request > constructor > SharedArrayBuffer > large buffer [10.36ms] (pass) Request > constructor > Buffer > empty buffer [10.14ms] (pass) Request > constructor > Buffer > small buffer [2.11ms] (pass) Request > constructor > Buffer > large buffer [3.02ms] (pass) Request > constructo ... (truncated) release with fix: 4 skipped $ bun scripts/build.ts --profile=release [configured] bun-profile → bun (stripped) in 615ms (unchanged) ninja: Entering directory `/workspace/bun/build/release' [1/40] cxx obj/src/jsc/bindings/webcore/streams/CrossRealmTransform.cpp.o [2/40] cxx obj/src/jsc/bindings/webcore/streams/BunAsyncIterableSource.cpp.o [3/40] cxx obj/src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp.o [4/40] cxx obj/src/jsc/bindings/webcore/streams/BunStreamSource.cpp.o [5/40] cxx obj/src/jsc/bindings/webcore/streams/JSByteLengthQueuingStrategy.cpp.o [6/40] cxx obj/src/jsc/bindings/webcore/streams/JSCompressionStream.cpp.o [7/40] cxx obj/src/jsc/bindings/webcore/streams/JSCompressionStreamShared.cpp.o [8/40] cxx obj/src/jsc/bindings/webcore/streams/JSCountQueuingStrategy.cpp.o [9/40] cxx obj/src/jsc/bindings/webcore/streams/JSDecompressionStream.cpp.o [10/40] cxx obj/src/jsc/bindings/webcore/streams/JSDirectStreamController.cpp.o [11/40] cxx obj/src/jsc/bindings/webcore/streams/JSReadableByteStreamController.cpp.o [12/40] cxx obj/src/jsc/bindings/webcore/streams/JSReadableStreamAsyncIterator.cpp.o [13/40] cxx obj/src/jsc/bindings/webcore/streams/JSReadableStreamBYOBReader.cpp.o [14/4 ... (truncated) ``` </details> <details><summary>diff hotspot</summary> ``` .../bindings/webcore/streams/WebStreamsExports.cpp | 22 +++- .../bindings/webcore/streams/WebStreamsInternals.h | 3 +- src/runtime/webcore/Body.rs | 2 +- src/runtime/webcore/ReadableStream.rs | 15 ++- test/js/web/fetch/body.test.ts | 136 ++++++++++++++++++++- 5 files changed, 170 insertions(+), 8 deletions(-) ``` </details> **gate history** · 2 passed · 0 rejected · iteration 1 <details><summary>evidence per changed file</summary> ``` file reads edits tests src/jsc/bindings/webcore/streams/WebStreamsExports.cpp 4 6 19 src/jsc/bindings/webcore/streams/WebStreamsInternals.h 3 3 19 src/runtime/webcore/Body.rs 5 1 19 src/runtime/webcore/ReadableStream.rs 5 3 19 test/js/web/fetch/body.test.ts 4 7 19 ``` </details> <!-- robobun:evidence:end -->
What does this PR do?
Bun's WHATWG
ReadableStreamprototype did not expose Node's registered stream-state symbols. As a result,node:stream.isReadable(new ReadableStream())returnednullinstead of tracking the stream's live state.This breaks installed Undici on a forced HTTP socket close. Undici gates response-body termination on
node:stream.isReadable(), so under Bun an original response and its clone could remain pending forever instead of rejecting with the transport error.This change exposes
nodejs.stream.readable,nodejs.stream.errored, andnodejs.stream.disturbedas read-only prototype accessors backed by Bun's existing stream state. It matches Node's behavior for readable, locked, disturbed, closed, and errored streams.AI assistance: This change was developed and reviewed with Codex.
How did you verify your code works?
response.text()and a cloned response reject promptly withTypeError: terminatedand the underlyingUND_ERR_SOCKET.