Pin a byte source's expectedLength as an exact total

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();

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,

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);
+        }
+      }
+    }
+  },
+};

src/tests/streams/readable-byte/main.js

@@ -279,4 +279,6 @@ export {
   byteSourceExpectedLengthRejected,
   byteSourceExpectedLengthCoercion,
   byteSourceExpectedLengthRange,
+  byteSourceExpectedLengthContract,
+  byteSourceExpectedLengthHeldRequest,
 } from 'expected-length';