From f9bcf83bad989024cfe2016caf7c064ffafd0567 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 14 Oct 2026 23:14:36 +0000 Subject: [PATCH 1/3] Pin a draining read that stops just before the close A ReadableStreamDrainingReader read that takes the last queued chunk but stops before the close sentinel reports done: false and leaves the stream readable, its desiredSize at the high-water mark and its closed promise pending, until the next read reaches the close. A default read closes the stream as it takes the last chunk. Pin it ahead of reporting the close with the chunks. The drain stops there for its maxSize, which only the internal-testing global can pass, or where the chunk sizes add up to Infinity, the one stop a read without a maxSize can meet: the C++ bridge's pump and body consumption read that way. Pin both on a value stream, and the maxSize case on a byte stream, whose controller has its own close check. The reader exists only under the TypeScript implementation; the C++ cells assert its absence, as the rest of these files do. --- src/tests/streams/readable-byte/AGENTS.md | 2 +- .../streams/readable-byte/draining-reader.js | 50 ++++++++++- src/tests/streams/readable-byte/main.js | 1 + src/tests/streams/readable/AGENTS.md | 2 +- src/tests/streams/readable/draining-reader.js | 87 ++++++++++++++++++- src/tests/streams/readable/main.js | 2 + 6 files changed, 137 insertions(+), 7 deletions(-) diff --git a/src/tests/streams/readable-byte/AGENTS.md b/src/tests/streams/readable-byte/AGENTS.md index ae854fbc619..fac39308f75 100644 --- a/src/tests/streams/readable-byte/AGENTS.md +++ b/src/tests/streams/readable-byte/AGENTS.md @@ -136,7 +136,7 @@ named suite test pins directly, differing only in incidental asserts. | `js-compat.js` | ledger #17, #22; byte halves of the mixed streams-js-test tests (closed promise, cancel reads, locked ops, globals) | | `flag-no-auto-allocate.js` | the flag cell (migrated streams-no-auto-allocate-test) | | `legacy-constructors.js` / `legacy-nodetach.js` | the flags table's legacy windows | -| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued byte backlog plus the close sentinel swept in one batched read with chunks INTACT (no coalescing); with autoAllocateChunkSize the conduit's wait-read synthesizes the descriptor so pull carries a byobRequest (ledger #6); expectedLength undefined when undeclared, a declared one passed through as the exact bigint (above 2^53 and up to the uint64 maximum); error/cancel propagation | +| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued byte backlog plus the close sentinel swept in one batched read with chunks INTACT (no coalescing); a read whose maxSize runs out before the close reports done: false, and the stream stays readable until the next read; with autoAllocateChunkSize the conduit's wait-read synthesizes the descriptor so pull carries a byobRequest (ledger #6); expectedLength undefined when undeclared, a declared one passed through as the exact bigint (above 2^53 and up to the uint64 maximum); error/cancel propagation | | `data-volumes.js` | byte-transfer volumes 64 B / 64 KiB / 1 MiB / 8 MiB via default and BYOB readers (incl. mismatched view/enqueue granularity), continuous prime-modulus pattern verified byte-exact; the source closes WITH its last enqueue (single-shape loops; parked-read close settlement is ledger #19's) | | `pollution.js` | prototype pollution neither implementation observes: a patched byte controller error() still errors the stream; readAtLeast() ignores a patched read(); tee copies and respond() remainders never consult the ArrayBuffer or %TypedArray% species; a native body ignores Object.prototype `type`/`autoAllocateChunkSize`; one Object.prototype.then getter fire per read, on an ordinary result object (default reads, autoAllocateChunkSize, BYOB and readAtLeast, at close, native bodies); ledger #34 | | `expected-length.js` | ledger #35, with the converted length observed as a fetch body's Content-Length via SELF; integral lengths up to the uint64 maximum accepted | diff --git a/src/tests/streams/readable-byte/draining-reader.js b/src/tests/streams/readable-byte/draining-reader.js index 0f75dc0d1c6..81fc880aa6f 100644 --- a/src/tests/streams/readable-byte/draining-reader.js +++ b/src/tests/streams/readable-byte/draining-reader.js @@ -11,7 +11,10 @@ // // Byte-stream facts pinned here: a queued byte backlog plus the close // sentinel is swept in one batched read with each enqueued chunk kept -// INTACT (no coalescing, no re-slicing); with autoAllocateChunkSize set +// INTACT (no coalescing, no re-slicing); a read whose maxSize runs out +// at the last bytes before the close reports done: false and leaves the +// stream readable until the next read reaches the close (the readable +// suite pins the same for value streams); with autoAllocateChunkSize set // the conduit's wait-read synthesizes the auto-allocate descriptor, so // pull observes a byobRequest (without it, pull sees null — ledger #5); // a byte stream that declares no expectedLength reads undefined, and a @@ -48,6 +51,51 @@ export const drainingReaderSweepsByteBacklog = { }, }; +export const drainingReaderByteMaxSizeEndsAtClose = { + async test() { + if (!usingTsImpl) { + strictEqual(typeof ReadableStreamDrainingReader, 'undefined'); + return; + } + // Three bytes, then close(): a read with maxSize 3 takes them and + // stops before the close sentinel. The stream stays readable, + // desiredSize at the high-water mark and closed pending, until the + // next read reaches the close. (A high-water mark of 8 tells + // readable from closed: at the byte default of 0, both read 0.) + let controller; + const rs = new ReadableStream( + { + type: 'bytes', + start(c) { + controller = c; + c.enqueue(new Uint8Array([1, 2, 3])); + c.close(); + }, + }, + { highWaterMark: 8 } + ); + const reader = new ReadableStreamDrainingReader(rs); + let closed = false; + reader.closed.then(() => { + closed = true; + }); + const first = await reader.read({ maxSize: 3 }); + deepStrictEqual( + first.chunks.map((chunk) => Array.from(chunk)), + [[1, 2, 3]] + ); + strictEqual(first.done, false); + strictEqual(controller.desiredSize, 8); + await scheduler.wait(0); + strictEqual(closed, false); + const second = await reader.read({ maxSize: 3 }); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); + }, +}; + export const drainingReaderByteExpectedLengthPassThrough = { async test() { if (!usingTsImpl) { diff --git a/src/tests/streams/readable-byte/main.js b/src/tests/streams/readable-byte/main.js index 16d07592e49..6daf90dc51e 100644 --- a/src/tests/streams/readable-byte/main.js +++ b/src/tests/streams/readable-byte/main.js @@ -236,6 +236,7 @@ export { export { drainingReaderSweepsByteBacklog, + drainingReaderByteMaxSizeEndsAtClose, drainingReaderByteExpectedLengthPassThrough, drainingReaderDrivesBytePull, drainingReaderByteErrorPropagation, diff --git a/src/tests/streams/readable/AGENTS.md b/src/tests/streams/readable/AGENTS.md index 4eb93ae50aa..9290185d690 100644 --- a/src/tests/streams/readable/AGENTS.md +++ b/src/tests/streams/readable/AGENTS.md @@ -123,7 +123,7 @@ C++ implementation; `draining-reader.js` asserts both sides. | `gc.js` | pending read + async iteration survive gc(); a controller held with its stream collected (ledger #19); both tee branches collected while the controller is held: enqueue() drops without retaining (WeakRef-checked, backlog included), desiredSize at the high-water mark, close() then enqueue() as ever, and pull() stops (ledger #20); one branch cancelled and the other collected, observed in the gc()'s own job (before finalization callbacks): the controller's queries drop the consumer themselves | | `then-interceptors.js` | one then-getter fire per read, on an ordinary result object, whether the read is answered from the queue, waits for enqueue() or close(), or finds the stream closed (a tee branch's too), and when it fires for a waiting read (ledger #16); none for a pipe's reads; per async-iterator next() (ledger #24) and return() (ledger #26); a then getter that cancels tee branches while a read result settles (during enqueue() and close(), one branch or two of three): the other branches' pending reads still resolve and their backlog still counts toward desiredSize | | `legacy-constructors.js` | the unflagged cell (see flags table) | -| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued backlog plus the close sentinel swept in ONE batched read; value chunks pass through UNTOUCHED (object identity); pull-driven yields per read with EOF as a separate empty batch; expectedLength undefined; error/cancel propagation; lock exclusivity and release | +| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued backlog plus the close sentinel swept in ONE batched read; a read that stops before the close sentinel, for its maxSize or because the chunk sizes add up to Infinity, reports done: false, and the stream stays readable until the next read; value chunks pass through UNTOUCHED (object identity); pull-driven yields per read with EOF as a separate empty batch; expectedLength undefined; error/cancel propagation; lock exclusivity and release | | `data-volumes.js` | value-stream volume axes: 4096-chunk counts, a 1 MiB single string chunk, 8 MiB aggregate (128 × 64 KiB), and 1 MiB through tee on both branches — every chunk index-encoded | | `pollution.js` | prototype pollution neither implementation observes: a patched array iterator or replaced Number global leaves bodies intact; a patched controller error() still errors the stream; omitted dictionaries read nothing from Object.prototype, nor does from() for the source and strategy it builds; @@asyncIterator shape (TS: the values() function object, per WebIDL; C++: a separate function) and for-await ignoring a patched values() | diff --git a/src/tests/streams/readable/draining-reader.js b/src/tests/streams/readable/draining-reader.js index 0b1b7c1bb2c..2a146be12db 100644 --- a/src/tests/streams/readable/draining-reader.js +++ b/src/tests/streams/readable/draining-reader.js @@ -11,10 +11,16 @@ // // Value-stream facts pinned here: a JS stream's real queue (unlike the // identity rendezvous) makes batching directly observable — a queued -// backlog plus the close sentinel is swept in ONE read; value chunks -// pass through the conduit UNTOUCHED (object identity included); a JS -// stream never declares an expectedLength; cancel and error propagate -// through the conduit like a default reader's. +// backlog plus the close sentinel is swept in ONE read; a read that +// takes the last chunk but stops before the close sentinel, for its +// maxSize or because the chunk sizes add up to Infinity (the one stop +// a read without a maxSize, such as the C++ bridge's, can meet), +// reports done: false and leaves the stream readable until the next +// read reaches the close, where a default read closes the stream as it +// takes the last chunk; value chunks pass through the conduit +// UNTOUCHED (object identity included); a JS stream never declares an +// expectedLength; cancel and error propagate through the conduit like +// a default reader's. /* global ReadableStreamDrainingReader */ @@ -50,6 +56,79 @@ export const drainingReaderSweepsQueuedBacklog = { }, }; +export const drainingReaderMaxSizeEndsAtClose = { + async test() { + if (!usingTsImpl) { + strictEqual(typeof ReadableStreamDrainingReader, 'undefined'); + return; + } + // One chunk of the default strategy's size 1, then close(): a read + // with maxSize 1 takes the chunk and stops before the close sentinel. + // The stream stays readable, desiredSize at the high-water mark and + // closed pending, until the next read reaches the close. + let controller; + const rs = new ReadableStream({ + start(c) { + controller = c; + c.enqueue('a'); + c.close(); + }, + }); + const reader = new ReadableStreamDrainingReader(rs); + let closed = false; + reader.closed.then(() => { + closed = true; + }); + const first = await reader.read({ maxSize: 1 }); + deepStrictEqual(first.chunks, ['a']); + strictEqual(first.done, false); + strictEqual(controller.desiredSize, 1); + await scheduler.wait(0); + strictEqual(closed, false); + const second = await reader.read({ maxSize: 1 }); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); + }, +}; + +export const drainingReaderSizeOverflowEndsAtClose = { + async test() { + if (!usingTsImpl) { + strictEqual(typeof ReadableStreamDrainingReader, 'undefined'); + return; + } + // A read without a maxSize, as the C++ bridge and body consumption + // read, stops where the chunk sizes add up to Infinity: here at the + // second of two chunks of size Number.MAX_VALUE, before the close. + const rs = new ReadableStream( + { + start(c) { + c.enqueue('a'); + c.enqueue('b'); + c.close(); + }, + }, + { size: () => Number.MAX_VALUE } + ); + const reader = new ReadableStreamDrainingReader(rs); + let closed = false; + reader.closed.then(() => { + closed = true; + }); + const first = await reader.read(); + deepStrictEqual(first.chunks, ['a', 'b']); + strictEqual(first.done, false); + await scheduler.wait(0); + strictEqual(closed, false); + const second = await reader.read(); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, true); + strictEqual(closed, true); + }, +}; + export const drainingReaderPullDriven = { async test() { if (!usingTsImpl) { diff --git a/src/tests/streams/readable/main.js b/src/tests/streams/readable/main.js index d13ec867281..3a987edb842 100644 --- a/src/tests/streams/readable/main.js +++ b/src/tests/streams/readable/main.js @@ -240,6 +240,8 @@ export { export { drainingReaderSweepsQueuedBacklog, + drainingReaderMaxSizeEndsAtClose, + drainingReaderSizeOverflowEndsAtClose, drainingReaderPullDriven, drainingReaderErrorPropagation, drainingReaderCancelReachesSource, From 823647d76755d6454d1795255cd367a6e482f505 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 14 Oct 2026 23:23:15 +0000 Subject: [PATCH 2/3] Report the close with a draining read's last chunks A ReadableStreamDrainingReader read whose size limit ran out at the last chunk before the close sentinel stopped short of the sentinel and reported done: false. The stream stayed readable, its closed promise pending, until a second read found the close. A controller.error() in between errored it, and the second read rejected, although every chunk had been read. A default read closes the stream as it takes the last chunk (spec PullSteps), and the error is then a no-op. The sentinel has no size, so the limit has no bearing on it. QueueCursor.drain now reports done whenever it ends at the sentinel, as the byte cursor's drain does through it, and the read closes the stream in the same call, as a default read does. The limit runs out at the read's maxSize, which only the internal-testing global passes, or, for a read without one, where the chunk sizes add up to Infinity. The C++ bridge's pump and body consumption read without one, so in production only streams whose chunk sizes add up to Infinity change: they finish in one read instead of two. Flip drainingReaderMaxSizeEndsAtClose, drainingReaderSizeOverflowEndsAtClose and drainingReaderByteMaxSizeEndsAtClose. typescript_implemented_streams is experimental, so this needs no compatibility flag. --- src/per_isolate/webstreams/queue.ts | 25 +++++----- src/tests/streams/readable-byte/AGENTS.md | 2 +- .../streams/readable-byte/draining-reader.js | 26 +++++----- src/tests/streams/readable/AGENTS.md | 2 +- src/tests/streams/readable/draining-reader.js | 47 +++++++++---------- 5 files changed, 47 insertions(+), 55 deletions(-) diff --git a/src/per_isolate/webstreams/queue.ts b/src/per_isolate/webstreams/queue.ts index 0ae3bb49db6..484013e3646 100644 --- a/src/per_isolate/webstreams/queue.ts +++ b/src/per_isolate/webstreams/queue.ts @@ -278,7 +278,8 @@ export interface StreamConsumer { // preserving spec PullSteps timing for the drain-then-close check. tryReadSync(reader: object): ReadableStreamReadResult | undefined; // Bulk read for the draining reader / pipeTo. Synchronous: collects what - // is buffered (the draining reader handles the empty-then-wait case). + // is buffered (the draining reader handles the empty-then-wait case), + // and is done once the consumer has reached the end of the stream. drain(maxSize?: number): { chunks: V[]; done: boolean }; // Reject pending reads submitted by a specific reader (lock release). cancelReadsForReader(reader: object, reason: unknown): void; @@ -992,29 +993,25 @@ class QueueCursor implements StreamConsumer { // (in strategy size units — bytes for byte streams). Returns // synchronously; the draining reader wraps in a promise at its API // boundary. Always takes at least one available entry so callers make - // progress even when maxSize is smaller than the next entry. + // progress even when maxSize is smaller than the next entry. Done + // whenever the cursor ends at the close sentinel, the limit + // notwithstanding: the sentinel has no size. drain(maxSize: number = Infinity): { chunks: V[]; done: boolean } { const chunks: V[] = []; let total = 0; - let done = false; - while (total < maxSize) { - const slot = this.#queue.getEntry(this.#position); - if (slot === undefined) break; - if (slot === CLOSE_SENTINEL) { - done = true; - break; - } - const entry = slot as QueueEntry; - ArrayPrototypePush(chunks, this.readEntryValue(entry)); - total += entry.size; + let slot = this.#queue.getEntry(this.#position); + while (isQueueEntry(slot) && total < maxSize) { + ArrayPrototypePush(chunks, this.readEntryValue(slot)); + total += slot.size; this.advancePastEntry(); + slot = this.#queue.getEntry(this.#position); } if (chunks.length > 0) { this.#queue.onCursorAdvanced(); } const result = ObjectCreate(null) as { chunks: V[]; done: boolean }; result.chunks = chunks; - result.done = done; + result.done = slot === CLOSE_SENTINEL; return result; } } diff --git a/src/tests/streams/readable-byte/AGENTS.md b/src/tests/streams/readable-byte/AGENTS.md index fac39308f75..f28d8141b75 100644 --- a/src/tests/streams/readable-byte/AGENTS.md +++ b/src/tests/streams/readable-byte/AGENTS.md @@ -136,7 +136,7 @@ named suite test pins directly, differing only in incidental asserts. | `js-compat.js` | ledger #17, #22; byte halves of the mixed streams-js-test tests (closed promise, cancel reads, locked ops, globals) | | `flag-no-auto-allocate.js` | the flag cell (migrated streams-no-auto-allocate-test) | | `legacy-constructors.js` / `legacy-nodetach.js` | the flags table's legacy windows | -| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued byte backlog plus the close sentinel swept in one batched read with chunks INTACT (no coalescing); a read whose maxSize runs out before the close reports done: false, and the stream stays readable until the next read; with autoAllocateChunkSize the conduit's wait-read synthesizes the descriptor so pull carries a byobRequest (ledger #6); expectedLength undefined when undeclared, a declared one passed through as the exact bigint (above 2^53 and up to the uint64 maximum); error/cancel propagation | +| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued byte backlog plus the close sentinel swept in one batched read with chunks INTACT (no coalescing); a read whose maxSize runs out at the last bytes before the close reaches the close all the same, reporting done and closing the stream in that call; with autoAllocateChunkSize the conduit's wait-read synthesizes the descriptor so pull carries a byobRequest (ledger #6); expectedLength undefined when undeclared, a declared one passed through as the exact bigint (above 2^53 and up to the uint64 maximum); error/cancel propagation | | `data-volumes.js` | byte-transfer volumes 64 B / 64 KiB / 1 MiB / 8 MiB via default and BYOB readers (incl. mismatched view/enqueue granularity), continuous prime-modulus pattern verified byte-exact; the source closes WITH its last enqueue (single-shape loops; parked-read close settlement is ledger #19's) | | `pollution.js` | prototype pollution neither implementation observes: a patched byte controller error() still errors the stream; readAtLeast() ignores a patched read(); tee copies and respond() remainders never consult the ArrayBuffer or %TypedArray% species; a native body ignores Object.prototype `type`/`autoAllocateChunkSize`; one Object.prototype.then getter fire per read, on an ordinary result object (default reads, autoAllocateChunkSize, BYOB and readAtLeast, at close, native bodies); ledger #34 | | `expected-length.js` | ledger #35, with the converted length observed as a fetch body's Content-Length via SELF; integral lengths up to the uint64 maximum accepted | diff --git a/src/tests/streams/readable-byte/draining-reader.js b/src/tests/streams/readable-byte/draining-reader.js index 81fc880aa6f..1b973a7f99f 100644 --- a/src/tests/streams/readable-byte/draining-reader.js +++ b/src/tests/streams/readable-byte/draining-reader.js @@ -12,9 +12,9 @@ // Byte-stream facts pinned here: a queued byte backlog plus the close // sentinel is swept in one batched read with each enqueued chunk kept // INTACT (no coalescing, no re-slicing); a read whose maxSize runs out -// at the last bytes before the close reports done: false and leaves the -// stream readable until the next read reaches the close (the readable -// suite pins the same for value streams); with autoAllocateChunkSize set +// at the last bytes before the close reaches the close all the same: it +// reports done and closes the stream in that call (the readable suite +// pins the same for value streams); with autoAllocateChunkSize set // the conduit's wait-read synthesizes the auto-allocate descriptor, so // pull observes a byobRequest (without it, pull sees null — ledger #5); // a byte stream that declares no expectedLength reads undefined, and a @@ -57,11 +57,12 @@ export const drainingReaderByteMaxSizeEndsAtClose = { strictEqual(typeof ReadableStreamDrainingReader, 'undefined'); return; } - // Three bytes, then close(): a read with maxSize 3 takes them and - // stops before the close sentinel. The stream stays readable, - // desiredSize at the high-water mark and closed pending, until the - // next read reaches the close. (A high-water mark of 8 tells - // readable from closed: at the byte default of 0, both read 0.) + // Three bytes, then close(): a read with maxSize 3 takes them, + // using up its maxSize, and reaches the close sentinel all the same. + // It reports done, and the stream closes in that call: closed + // resolves before the read does, and desiredSize reads 0. (A + // high-water mark of 8 tells readable from closed: at the byte + // default of 0, both read 0.) let controller; const rs = new ReadableStream( { @@ -84,15 +85,12 @@ export const drainingReaderByteMaxSizeEndsAtClose = { first.chunks.map((chunk) => Array.from(chunk)), [[1, 2, 3]] ); - strictEqual(first.done, false); - strictEqual(controller.desiredSize, 8); - await scheduler.wait(0); - strictEqual(closed, false); + strictEqual(first.done, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); const second = await reader.read({ maxSize: 3 }); deepStrictEqual(second.chunks, []); strictEqual(second.done, true); - strictEqual(controller.desiredSize, 0); - strictEqual(closed, true); }, }; diff --git a/src/tests/streams/readable/AGENTS.md b/src/tests/streams/readable/AGENTS.md index 9290185d690..b089deb55cd 100644 --- a/src/tests/streams/readable/AGENTS.md +++ b/src/tests/streams/readable/AGENTS.md @@ -123,7 +123,7 @@ C++ implementation; `draining-reader.js` asserts both sides. | `gc.js` | pending read + async iteration survive gc(); a controller held with its stream collected (ledger #19); both tee branches collected while the controller is held: enqueue() drops without retaining (WeakRef-checked, backlog included), desiredSize at the high-water mark, close() then enqueue() as ever, and pull() stops (ledger #20); one branch cancelled and the other collected, observed in the gc()'s own job (before finalization callbacks): the controller's queries drop the consumer themselves | | `then-interceptors.js` | one then-getter fire per read, on an ordinary result object, whether the read is answered from the queue, waits for enqueue() or close(), or finds the stream closed (a tee branch's too), and when it fires for a waiting read (ledger #16); none for a pipe's reads; per async-iterator next() (ledger #24) and return() (ledger #26); a then getter that cancels tee branches while a read result settles (during enqueue() and close(), one branch or two of three): the other branches' pending reads still resolve and their backlog still counts toward desiredSize | | `legacy-constructors.js` | the unflagged cell (see flags table) | -| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued backlog plus the close sentinel swept in ONE batched read; a read that stops before the close sentinel, for its maxSize or because the chunk sizes add up to Infinity, reports done: false, and the stream stays readable until the next read; value chunks pass through UNTOUCHED (object identity); pull-driven yields per read with EOF as a separate empty batch; expectedLength undefined; error/cancel propagation; lock exclusivity and release | +| `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued backlog plus the close sentinel swept in ONE batched read; a read whose size limit runs out at the last chunk (its maxSize, or chunk sizes adding up to Infinity) reaches the close all the same, reporting done and closing the stream in that call; value chunks pass through UNTOUCHED (object identity); pull-driven yields per read with EOF as a separate empty batch; expectedLength undefined; error/cancel propagation; lock exclusivity and release | | `data-volumes.js` | value-stream volume axes: 4096-chunk counts, a 1 MiB single string chunk, 8 MiB aggregate (128 × 64 KiB), and 1 MiB through tee on both branches — every chunk index-encoded | | `pollution.js` | prototype pollution neither implementation observes: a patched array iterator or replaced Number global leaves bodies intact; a patched controller error() still errors the stream; omitted dictionaries read nothing from Object.prototype, nor does from() for the source and strategy it builds; @@asyncIterator shape (TS: the values() function object, per WebIDL; C++: a separate function) and for-await ignoring a patched values() | diff --git a/src/tests/streams/readable/draining-reader.js b/src/tests/streams/readable/draining-reader.js index 2a146be12db..4ab84b1cfdf 100644 --- a/src/tests/streams/readable/draining-reader.js +++ b/src/tests/streams/readable/draining-reader.js @@ -11,16 +11,15 @@ // // Value-stream facts pinned here: a JS stream's real queue (unlike the // identity rendezvous) makes batching directly observable — a queued -// backlog plus the close sentinel is swept in ONE read; a read that -// takes the last chunk but stops before the close sentinel, for its -// maxSize or because the chunk sizes add up to Infinity (the one stop -// a read without a maxSize, such as the C++ bridge's, can meet), -// reports done: false and leaves the stream readable until the next -// read reaches the close, where a default read closes the stream as it -// takes the last chunk; value chunks pass through the conduit -// UNTOUCHED (object identity included); a JS stream never declares an -// expectedLength; cancel and error propagate through the conduit like -// a default reader's. +// backlog plus the close sentinel is swept in ONE read; a read whose +// size limit runs out at the last chunk, its maxSize or the chunk sizes +// adding up to Infinity (the one limit a read without a maxSize, such +// as the C++ bridge's, can meet), reaches the close all the same: it +// reports done and closes the stream in that call, as a default read +// closes it as it takes the last chunk; value chunks pass through the +// conduit UNTOUCHED (object identity included); a JS stream never +// declares an expectedLength; cancel and error propagate through the +// conduit like a default reader's. /* global ReadableStreamDrainingReader */ @@ -63,9 +62,10 @@ export const drainingReaderMaxSizeEndsAtClose = { return; } // One chunk of the default strategy's size 1, then close(): a read - // with maxSize 1 takes the chunk and stops before the close sentinel. - // The stream stays readable, desiredSize at the high-water mark and - // closed pending, until the next read reaches the close. + // with maxSize 1 takes the chunk, using up its maxSize, and reaches + // the close sentinel all the same. It reports done, and the stream + // closes in that call: closed resolves before the read does, and + // desiredSize reads 0. let controller; const rs = new ReadableStream({ start(c) { @@ -81,15 +81,12 @@ export const drainingReaderMaxSizeEndsAtClose = { }); const first = await reader.read({ maxSize: 1 }); deepStrictEqual(first.chunks, ['a']); - strictEqual(first.done, false); - strictEqual(controller.desiredSize, 1); - await scheduler.wait(0); - strictEqual(closed, false); + strictEqual(first.done, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); const second = await reader.read({ maxSize: 1 }); deepStrictEqual(second.chunks, []); strictEqual(second.done, true); - strictEqual(controller.desiredSize, 0); - strictEqual(closed, true); }, }; @@ -100,8 +97,10 @@ export const drainingReaderSizeOverflowEndsAtClose = { return; } // A read without a maxSize, as the C++ bridge and body consumption - // read, stops where the chunk sizes add up to Infinity: here at the - // second of two chunks of size Number.MAX_VALUE, before the close. + // read, has its limit used up once the chunk sizes add up to + // Infinity: here by the second of two chunks of size + // Number.MAX_VALUE, the last before the close. It reaches the close + // all the same: it reports done, and the stream closes in that call. const rs = new ReadableStream( { start(c) { @@ -119,13 +118,11 @@ export const drainingReaderSizeOverflowEndsAtClose = { }); const first = await reader.read(); deepStrictEqual(first.chunks, ['a', 'b']); - strictEqual(first.done, false); - await scheduler.wait(0); - strictEqual(closed, false); + strictEqual(first.done, true); + strictEqual(closed, true); const second = await reader.read(); deepStrictEqual(second.chunks, []); strictEqual(second.done, true); - strictEqual(closed, true); }, }; From bc0ab46f110497b93d667d66645d87431a2f8b66 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 14 Oct 2026 23:32:14 +0000 Subject: [PATCH 3/3] Share the read completion steps with the draining read A default read that waits takes its completion steps, the controller's drain-then-close check and then closing the stream if the read is done, through readCompletionSteps. Answered from the queue (readBufferedSync, which the pipe's synchronous read shares), it repeated them inline, and the draining read repeated them too, making the check only when done. Both now pull as before and then call readCompletionSteps, which takes the done flag in place of the result. The default read and the draining read's wait-read each looked up autoAllocateChunkSize behind their own queued-byte brand check and cast. getByteControllerAutoAllocateChunkSize now takes any controller and makes the check itself, answering undefined for all but the queued byte controller. No behavior change. The draining read now makes the check on every completion. Since the previous commit it reports done whenever its cursor ends at the close sentinel, so after any other completion the check finds nothing to close: a stream whose cursor is at the sentinel closed when the cursor got there, or at close(). The steps keep their order, and the brand check is the same private-field test. --- src/per_isolate/webstreams/readable.ts | 73 +++++++++++--------------- 1 file changed, 32 insertions(+), 41 deletions(-) diff --git a/src/per_isolate/webstreams/readable.ts b/src/per_isolate/webstreams/readable.ts index 245e3ddb1d6..d0d6ad3b891 100644 --- a/src/per_isolate/webstreams/readable.ts +++ b/src/per_isolate/webstreams/readable.ts @@ -296,8 +296,8 @@ let setReadableStreamInteropErrorHook: ( // BYOB reads: the queued byte controller, or ANY native controller — // native sources are byte-capable by definition (the marker is // sufficient). Distinct from isByteStreamController, which remains the -// QUEUED-byte brand check used by the queued-only paths (autoAllocate -// synthesis, tee's cursor fork). +// QUEUED-byte brand check used by the queued-only paths (the cursor forks +// of tee and detach). function isByteCapableController(value: unknown): boolean { return isByteStreamController(value) || isNativeController(value); } @@ -746,13 +746,8 @@ function readBufferedSync( const result = consumer.tryReadSync(reader) as ReadableStreamReadResult | undefined; if (result === undefined) return undefined; - if (controller !== undefined) { - controllerPullIfNeeded(controller); - controllerMaybeCloseStream(controller); - } - if (result.done) { - readableStreamClose(stream); - } + if (controller !== undefined) controllerPullIfNeeded(controller); + readCompletionSteps(stream, controller, result.done); return result; } @@ -825,14 +820,12 @@ function defaultReaderReadSteps( ReadableStreamReadResult >; const resolve = (result: ReadableStreamReadResult): void => { - readCompletionSteps(stream, controller, result); + readCompletionSteps(stream, controller, result.done); pending.resolve(present(result as ReadableStreamReadResult)); }; const autoAllocateChunkSize = - controller !== undefined && isByteStreamController(controller) - ? getByteControllerAutoAllocateChunkSize( - controller as ReadableByteStreamController - ) + controller !== undefined + ? getByteControllerAutoAllocateChunkSize(controller) : undefined; if (autoAllocateChunkSize !== undefined) { // QUEUED-BYTE-SPECIFIC (sanctioned exception to backend-blindness): @@ -925,19 +918,20 @@ function readViaAutoAllocateDescriptor( return consumer.readBYOB(descriptor); } -// What a default read does once it has its result, before handing it over. +// What a read does once it has its result, before handing it over. +// Shared by the default and draining reads; each does its own pull. function readCompletionSteps( stream: ReadableStream, controller: AnyReadableController | undefined, - result: ReadableStreamReadResult + done: boolean ): void { // Drain-then-close (spec HandleQueueDrain): if this read consumed the // last data with close requested, the controller's primary stream // transitions now — checked on every read completion, not just done - // results, because the read that drains the final chunk resolves with - // done: false. + // results, because a default read that drains the final chunk resolves + // with done: false. if (controller !== undefined) controllerMaybeCloseStream(controller); - if (result.done) { + if (done) { // This stream's cursor reached the sentinel. Close THIS stream (tee // branches close independently of the controller's primary stream). readableStreamClose(stream); @@ -1883,8 +1877,10 @@ let byteControllerRespondWithNewView: ( controller: ReadableByteStreamController, view: ArrayBufferView ) => void; +// The queued byte controller's autoAllocateChunkSize; undefined for any +// other controller. let getByteControllerAutoAllocateChunkSize: ( - controller: ReadableByteStreamController + controller: AnyReadableController ) => number | undefined; // tee() and detach are about to replace the cursor `from`; `wasSole` when // it was the queue's only cursor. @@ -2113,7 +2109,9 @@ class ReadableByteStreamController implements ReadableByteStreamControllerType { }; getByteControllerAutoAllocateChunkSize = (controller) => { - return controller.#autoAllocateChunkSize; + return #queue in controller + ? controller.#autoAllocateChunkSize + : undefined; }; byteControllerOnFork = (controller, from, wasSole) => { @@ -2830,22 +2828,18 @@ async function drainingReaderReadInternal( // machinery so the source's pull observes a byobRequest over the // auto-allocated buffer — the body and pipe pumps drive // respond()-oriented sources exactly like C++'s BYOB pump. - let promise: Promise> | undefined; - if (controller !== undefined && isByteStreamController(controller)) { - const autoAllocateChunkSize = getByteControllerAutoAllocateChunkSize( - controller as ReadableByteStreamController - ); - if (autoAllocateChunkSize !== undefined) { - promise = readViaAutoAllocateDescriptor( - consumer as unknown as ByteStreamConsumerType, - autoAllocateChunkSize, - reader - ); - } - } - if (promise === undefined) { - promise = consumer.read(reader); - } + const autoAllocateChunkSize = + controller !== undefined + ? getByteControllerAutoAllocateChunkSize(controller) + : undefined; + const promise: Promise> = + autoAllocateChunkSize !== undefined + ? readViaAutoAllocateDescriptor( + consumer as unknown as ByteStreamConsumerType, + autoAllocateChunkSize, + reader + ) + : consumer.read(reader); if (controller !== undefined) controllerPullIfNeeded(controller); const single = await promise; if (single.done) { @@ -2862,10 +2856,7 @@ async function drainingReaderReadInternal( // Pump: draining freed queue space, so the source may pull again // immediately (synchronous sources refill before we return). if (controller !== undefined) controllerPullIfNeeded(controller); - if (result.done) { - if (controller !== undefined) controllerMaybeCloseStream(controller); - readableStreamClose(stream); - } + readCompletionSteps(stream, controller, result.done); return result; }