diff --git a/src/per_isolate/webstreams/readable.ts b/src/per_isolate/webstreams/readable.ts index dd631f6e9c1..c98069074e2 100644 --- a/src/per_isolate/webstreams/readable.ts +++ b/src/per_isolate/webstreams/readable.ts @@ -2445,22 +2445,7 @@ class ReadableByteStreamController implements ReadableByteStreamControllerType { if (!head.settledAtEndOfData) { head.buffer = ArrayBufferPrototypeTransferToFixedLength(head.buffer); } - this.#invalidateByobRequest(); - if (cursor === undefined) { - this.#respondToReleasedHead(bytesWritten, state); - } else if (state === 'closed') { - // respond(0)-while-closed: commit all pending descriptors with - // { done: true, value: filled-so-far view }. - cursor.commitPullIntosOnClose(); - } else { - // EXPECTED-LENGTH CONTRACT: respond() bytes count toward the total - // (the second ingress path alongside enqueue). - this.#accountDelivery(bytesWritten); - cursor.respondBYOB(bytesWritten); - // The respond may have drained the queue with close requested. - this.#core.maybeCloseStream(); - this.#core.callPullIfNeeded(); - } + this.#respondInternal(cursor, bytesWritten, state); } // Exact-total accounting (see #expectedLength). On overflow (more bytes @@ -2534,15 +2519,31 @@ class ReadableByteStreamController implements ReadableByteStreamControllerType { if (!head.settledAtEndOfData) { head.buffer = ArrayBufferPrototypeTransferToFixedLength(info.buffer); } + this.#respondInternal(cursor, info.byteLength, state); + } + + // Spec ReadableByteStreamControllerRespondInternal: respond() and + // respondWithNewView() once they have validated the response and + // transferred the head pull-into's buffer. `cursor` is the sole cursor + // whose head was answered, or undefined for #releasedHead. + #respondInternal( + cursor: ByteStreamCursorType | undefined, + bytesWritten: number, + state: ReadableStreamState + ): void { this.#invalidateByobRequest(); if (cursor === undefined) { - this.#respondToReleasedHead(info.byteLength, state); + this.#respondToReleasedHead(bytesWritten, state); } else if (state === 'closed') { + // A closed-state response (respond(0), or a zero-length view): + // commit all pending descriptors with { done: true, value: + // filled-so-far view }. cursor.commitPullIntosOnClose(); } else { - // EXPECTED-LENGTH CONTRACT: counts toward the total like respond(). - this.#accountDelivery(info.byteLength); - cursor.respondBYOB(info.byteLength); + // EXPECTED-LENGTH CONTRACT: responded bytes count toward the total + // (the second ingress path alongside enqueue). + this.#accountDelivery(bytesWritten); + cursor.respondBYOB(bytesWritten); // The respond may have drained the queue with close requested. this.#core.maybeCloseStream(); this.#core.callPullIfNeeded(); diff --git a/src/tests/streams/readable-byte/AGENTS.md b/src/tests/streams/readable-byte/AGENTS.md index cacdb6c955a..c2ffa99ef36 100644 --- a/src/tests/streams/readable-byte/AGENTS.md +++ b/src/tests/streams/readable-byte/AGENTS.md @@ -56,6 +56,7 @@ behavior-parity (messages aside). | 39 | ctor argument conversion order, every member of a byte source (the readable suite's #2 for a default source) | the source first, every member in its own order (type, autoAllocateChunkSize, start, pull, cancel, expectedLength), then the strategy: hwm and its ToNumber, size | the strategy first (spec); then the source's members in lexicographic order (spec: autoAllocateChunkSize, cancel, expectedLength, pull, start, type) | `argumentConversionOrder` | | 40 | which error wins when two are possible (the readable suite's #32; in both, spec: every source member converts before the strategy and a zero autoAllocateChunkSize are checked, so a non-callable pull, TypeError with different messages, beats an invalid highWaterMark, a size function and a zero autoAllocateChunkSize; a zero autoAllocateChunkSize is checked after an invalid highWaterMark, whose error wins, TypeError in C++ and RangeError in TypeScript) | a non-callable pull also beats a negative autoAllocateChunkSize, which the int conversion accepts and a later range check rejects | a negative autoAllocateChunkSize (TypeError) beats a non-callable pull (spec: it converts first) | `conversionErrorPrecedence` | | 41 | a bare ArrayBuffer passed to enqueue() or respondWithNewView(), whose WebIDL parameter is an ArrayBufferView (any other non-view there, and every non-view at read(view) and readAtLeast(), is a TypeError in both, messages aside) | taken as a view over the whole buffer (the parameter is a BufferSource): enqueue() detaches it and delivers its bytes; respondWithNewView() accepts one of the request's buffer length, detaches it and delivers its bytes | TypeError 'chunk must be an ArrayBufferView' or 'view must be an ArrayBufferView' (spec); the buffer stays attached, nothing is enqueued, and the pending read is left for a later response | `nonViewArgumentsRejected` | +| 42 | a byte source's declared `expectedLength` as an exact total of the bytes it delivers through enqueue(), respond() and respondWithNewView(), a response to a byobRequest held across tee() (#25) included (delivering exactly the total, then close(), is parity) | not enforced: a delivery past the total reaches the reader, and close() short of it closes the stream | by design: the delivery past the total throws RangeError 'byte source delivered more bytes than its declared expectedLength', and close() short of it RangeError 'byte source closed before producing its declared expectedLength'; either errors the stream with that error | `byteSourceExpectedLengthContract`, `byteSourceExpectedLengthHeldRequest` | Parity worth noting (probed, pinned): byte hwm defaults to 0 with NO automatic pull; pull-throw and error-then-throw identity; enqueue @@ -145,7 +146,7 @@ named suite test pins directly, differing only in incidental asserts. | `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 | +| `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; ledger #42, the declared length as an exact total through enqueue(), respond() and respondWithNewView(), and through a byobRequest held across tee() | Consumed sources (deleted): streams-js-test.js (value halves were already covered by the readable suite), streams-tee-edge-cases-test.js, diff --git a/src/tests/streams/readable-byte/expected-length.js b/src/tests/streams/readable-byte/expected-length.js index e08c0085ef7..018886749ae 100644 --- a/src/tests/streams/readable-byte/expected-length.js +++ b/src/tests/streams/readable-byte/expected-length.js @@ -14,8 +14,16 @@ // 0. Every failure is a TypeError. C++ lets the number 2 ** 64 through its // range check to a cast whose result is undefined, so that one value is // asserted for TypeScript only. +// +// DIVERGENCE (ledger #42), by design: TS also holds the source to the +// declared length as an exact total of what it delivers through enqueue(), +// respond() and respondWithNewView(), a response to a byobRequest held +// across tee() included. The delivery that goes past it throws RangeError, +// and so does a close() short of it; either errors the stream with that +// error. C++ enforces neither: every delivery reaches the reader, and +// close() closes the stream. -import { ok, strictEqual, throws } from 'node:assert'; +import { ok, rejects, strictEqual, throws } from 'node:assert'; import { usingTsImpl } from 'which-impl'; function byteSource(expectedLength, byteCount = 0) { @@ -109,3 +117,153 @@ export const byteSourceExpectedLengthRange = { if (usingTsImpl) throws(() => byteSource(2 ** 64), TypeError); }, }; + +// A byte source declaring expectedLength, with a read(view) of 8 bytes +// pending, so that its controller holds a byobRequest. +async function pendingByobRead(expectedLength) { + let controller; + const rs = new ReadableStream({ + type: 'bytes', + expectedLength, + start(c) { + controller = c; + }, + }); + const reader = rs.getReader({ mode: 'byob' }); + const read = reader.read(new Uint8Array(8)); + await scheduler.wait(5); + return { rs, controller, reader, read }; +} + +const OVERFLOW = { + name: 'RangeError', + message: 'byte source delivered more bytes than its declared expectedLength', +}; + +const UNDERFLOW = { + name: 'RangeError', + message: 'byte source closed before producing its declared expectedLength', +}; + +// Five bytes, one past an expectedLength of 4, through each way of +// answering the byobRequest. +const respondFive = { + 'respond()': (c) => c.byobRequest.respond(5), + 'respondWithNewView()': (c) => { + const { buffer, byteOffset } = c.byobRequest.view; + c.byobRequest.respondWithNewView(new Uint8Array(buffer, byteOffset, 5)); + }, +}; + +// DIVERGENCE (ledger #42) +export const byteSourceExpectedLengthContract = { + async test() { + const deliverFive = { + ...respondFive, + 'enqueue()': (c) => c.enqueue(new Uint8Array(5)), + }; + for (const [name, deliver] of Object.entries(deliverFive)) { + const { controller, reader, read } = await pendingByobRead(4); + if (usingTsImpl) { + throws(() => deliver(controller), OVERFLOW, name); + await rejects(read, OVERFLOW); + await rejects(reader.closed, OVERFLOW); + } else { + deliver(controller); + strictEqual((await read).value.byteLength, 5, name); + await reader.cancel('cleanup'); + } + } + + // Exactly the declared total, then close(): parity. + { + const { controller, reader, read } = await pendingByobRead(4); + controller.byobRequest.respond(4); + controller.close(); + strictEqual((await read).value.byteLength, 4); + await reader.closed; + } + + // close() short of the declared total. + { + const { controller, reader, read } = await pendingByobRead(4); + controller.byobRequest.respond(2); + if (usingTsImpl) { + throws(() => controller.close(), UNDERFLOW); + strictEqual((await read).value.byteLength, 2); + await rejects(reader.closed, UNDERFLOW); + } else { + controller.close(); + strictEqual((await read).value.byteLength, 2); + await reader.closed; + } + } + }, +}; + +// pendingByobRead's source with its reader released, the source holding +// the byobRequest over the released read, then teed: the request survives +// the fork, and a response to it reaches both branches (ledger #25). +async function teeWithHeldByobRequest(expectedLength) { + const { rs, controller, reader, read } = + await pendingByobRead(expectedLength); + const request = controller.byobRequest; + reader.releaseLock(); + await rejects(read, TypeError); + const branches = rs.tee(); + strictEqual(controller.byobRequest, request); + return { controller, branches }; +} + +// DIVERGENCE (ledger #42) +export const byteSourceExpectedLengthHeldRequest = { + async test() { + for (const [name, deliver] of Object.entries(respondFive)) { + const { controller, branches } = await teeWithHeldByobRequest(4); + if (usingTsImpl) { + throws(() => deliver(controller), OVERFLOW, name); + for (const branch of branches) { + await rejects(branch.getReader().read(), OVERFLOW); + } + } else { + deliver(controller); + for (const branch of branches) { + const { value } = await branch.getReader().read(); + strictEqual(value.byteLength, 5, name); + } + controller.error(new Error('cleanup')); + } + } + + // Exactly the declared total, then close(): parity. + { + const { controller, branches } = await teeWithHeldByobRequest(4); + controller.byobRequest.respond(4); + controller.close(); + for (const branch of branches) { + const reader = branch.getReader(); + strictEqual((await reader.read()).value.byteLength, 4); + strictEqual((await reader.read()).done, true); + } + } + + // close() short of the declared total. + { + const { controller, branches } = await teeWithHeldByobRequest(4); + controller.byobRequest.respond(2); + if (usingTsImpl) { + throws(() => controller.close(), UNDERFLOW); + for (const branch of branches) { + await rejects(branch.getReader().read(), UNDERFLOW); + } + } else { + controller.close(); + for (const branch of branches) { + const reader = branch.getReader(); + strictEqual((await reader.read()).value.byteLength, 2); + strictEqual((await reader.read()).done, true); + } + } + } + }, +}; diff --git a/src/tests/streams/readable-byte/main.js b/src/tests/streams/readable-byte/main.js index 02981942bef..c6aaf50c032 100644 --- a/src/tests/streams/readable-byte/main.js +++ b/src/tests/streams/readable-byte/main.js @@ -279,4 +279,6 @@ export { byteSourceExpectedLengthRejected, byteSourceExpectedLengthCoercion, byteSourceExpectedLengthRange, + byteSourceExpectedLengthContract, + byteSourceExpectedLengthHeldRequest, } from 'expected-length';