-
-
Notifications
You must be signed in to change notification settings - Fork 163
fix(stream): push() no longer marks a Readable disturbed; readableDidRead is dataEmitted #11228
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| **node:stream: `push()` no longer marks a Readable as disturbed (#11212).** `stream.isDisturbed(r)` and `r.readableDidRead` became `true` as soon as data was pushed, and also on a bare `resume()` / `pipe()`, on an async-iterator attach, and on a `read()` that returned `null`. In Node they flip only when a chunk reaches a consumer. undici 8.9.0 pushes the response into its `BodyReadable` before `body.text()` runs, so every `body.text()` / `.json()` rejected with `TypeError: unusable`. | ||
|
|
||
| The stream now keeps one flag, which is Node's `_readableState.dataEmitted`: `readableDidRead` reads it, `_readableState.dataEmitted` reads and writes it, and it is set only by a `'data'` emission or a `read()` that returns data. `stream.isDisturbed()` now also counts `readableAborted`, as Node does. | ||
|
|
||
| With this, undici 8.9.0 `request()` + `body.text()` / `.json()` matches Node for GET, POST and 404. | ||
|
|
||
| Files: `crates/perry-runtime/src/node_stream_readwrite.rs`, `node_stream.rs`, `node_stream/async_iterator.rs`, `node_stream/readable_from_promises.rs`, `node_stream_constructors/{introspection,pipeline}.rs`, `node_stream_state_view.rs`, `node_stream_state_tests.rs`. The existing empty-`read()` assertion was corrected to match Node. Gap test: `test-files/test_gap_stream_push_not_disturbed.ts`. It fails on main and is byte-identical to Node 26.5.1. |
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
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
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
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
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
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
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| // #11212: `push()` is the producer side and must not disturb a Readable. | ||
| // Node derives `readableDidRead` from `_readableState.dataEmitted` (set only | ||
| // when a chunk is handed to a consumer — a 'data' emission, which `read()` | ||
| // performs too) and `stream.isDisturbed(r)` from `readableDidRead || | ||
| // readableAborted`. undici's `body.text()` rejects with "unusable" when a | ||
| // body it has only pushed into reports disturbed. | ||
| import { Readable, Writable, PassThrough, isDisturbed } from "node:stream"; | ||
|
|
||
| function show(label: string, r: any): void { | ||
| console.log(label, "isDisturbed", isDisturbed(r), "readableDidRead", r.readableDidRead); | ||
| } | ||
| const tick = () => new Promise<void>((res) => setImmediate(res)); | ||
|
|
||
| async function main(): Promise<void> { | ||
| // push / unshift only. | ||
| const a = new Readable({ read() {} }); | ||
| show("fresh", a); | ||
| a.push("hello"); | ||
| a.push(Buffer.from("world")); | ||
| show("after push", a); | ||
| a.unshift("x"); | ||
| show("after unshift", a); | ||
| await tick(); | ||
| show("after a tick", a); | ||
|
|
||
| // read(): null does not count, data does. | ||
| const b = new Readable({ read() {} }); | ||
| console.log("read on empty:", b.read()); | ||
| show("after empty read", b); | ||
| b.push("abc"); | ||
| console.log("read:", String(b.read())); | ||
| show("after read", b); | ||
|
|
||
| // 'data' listener: disturbed once a chunk is emitted, not at attach time. | ||
| const c = new Readable({ read() {} }); | ||
| c.on("data", (chunk: any) => console.log("data:", String(chunk))); | ||
| show("data listener attached", c); | ||
| await tick(); | ||
| show("flowing, nothing pushed", c); | ||
| c.push("one"); | ||
| await tick(); | ||
| show("after data emitted", c); | ||
|
|
||
| // resume() alone. | ||
| const d = new Readable({ read() {} }); | ||
| d.resume(); | ||
| await tick(); | ||
| show("resumed, nothing pushed", d); | ||
| d.push("two"); | ||
| await tick(); | ||
| show("resumed, after push", d); | ||
|
|
||
| // pipe. | ||
| const e = new Readable({ read() {} }); | ||
| const sinkChunks: string[] = []; | ||
| const sink = new Writable({ | ||
| write(chunk: any, _enc: any, cb: any) { | ||
| sinkChunks.push(String(chunk)); | ||
| cb(); | ||
| }, | ||
| }); | ||
| e.pipe(sink); | ||
| await tick(); | ||
| show("piped, nothing pushed", e); | ||
| e.push("three"); | ||
| await tick(); | ||
| show("piped, after push", e); | ||
| console.log("sink got:", sinkChunks.join(",")); | ||
|
|
||
| // async iterator. | ||
| const f = new Readable({ read() {} }); | ||
| f.push("four"); | ||
| f.push(null); | ||
| show("before iteration", f); | ||
| const got: string[] = []; | ||
| for await (const chunk of f) got.push(String(chunk)); | ||
| console.log("iterated:", got.join(",")); | ||
| show("after iteration", f); | ||
|
|
||
| // destroy before end: aborted counts for isDisturbed, not readableDidRead. | ||
| const g = new Readable({ read() {} }); | ||
| g.push("five"); | ||
| g.destroy(); | ||
| await tick(); | ||
| show("destroyed with unread data", g); | ||
|
|
||
| // destroy after a clean end. | ||
| const h = new Readable({ read() {} }); | ||
| h.push(null); | ||
| h.resume(); | ||
| await new Promise<void>((res) => h.on("close", () => res())); | ||
| show("ended and closed", h); | ||
|
|
||
| // Web conversion only wraps the stream; it disturbs once the web side reads. | ||
| const w = new Readable({ read() {} }); | ||
| w.push("six"); | ||
| w.push(null); | ||
| const web = Readable.toWeb(w); | ||
| show("after toWeb", w); | ||
| const reader = (web as any).getReader(); | ||
| const first = await reader.read(); | ||
| console.log("web read:", first.done, String(Buffer.from(first.value))); | ||
| show("after web read", w); | ||
|
|
||
| // A PassThrough written to but never read. | ||
| const p = new PassThrough(); | ||
| p.write("seven"); | ||
| await tick(); | ||
| show("passthrough written, unread", p); | ||
| } | ||
| main(); |
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🔎 Supported by static analysis
🏁 Script executed:
Repository: PerryTS/perry
Length of output: 5019
🏁 Script executed:
Repository: PerryTS/perry
Length of output: 41674
🌐 Web query:
Node.js v26.5.1 readable.js readableAborted endEmitted push null destroy before end💡 Result:
🏁 Script executed:
Repository: PerryTS/perry
Length of output: 42277
Use emitted
end, not pushed EOF, for readable abortion.When
push(null)marks EOF, the stream schedulesendbut has not emitted it. If the unread stream is destroyed before that microtask runs,readable_aborted_value()sees the ended flag and returnsfalse. With no delivered data,isDisturbed()also returnsfalse. Node reports this state as aborted.Use only the emitted-end flag and add a regression test for this sequence.
Suggested fix
🤖 Prompt for AI Agents