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/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; } diff --git a/src/tests/streams/readable-byte/AGENTS.md b/src/tests/streams/readable-byte/AGENTS.md index ae854fbc619..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); 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 0f75dc0d1c6..1b973a7f99f 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 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 @@ -48,6 +51,49 @@ 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, + // 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( + { + 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, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); + const second = await reader.read({ maxSize: 3 }); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, 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..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; 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 0b1b7c1bb2c..4ab84b1cfdf 100644 --- a/src/tests/streams/readable/draining-reader.js +++ b/src/tests/streams/readable/draining-reader.js @@ -11,10 +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; 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 */ @@ -50,6 +55,77 @@ 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, 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) { + 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, true); + strictEqual(controller.desiredSize, 0); + strictEqual(closed, true); + const second = await reader.read({ maxSize: 1 }); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, 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, 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) { + 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, true); + strictEqual(closed, true); + const second = await reader.read(); + deepStrictEqual(second.chunks, []); + strictEqual(second.done, 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,