Pin shared-memory writes to IdentityTransformStream

src/per_isolate/webstreams/AGENTS.md

@@ -13,7 +13,7 @@ private-brand dispatch, no `instanceof`) apply here — see
 | File          | Role                                                                                         |
 | ------------- | -------------------------------------------------------------------------------------------- |
 | `ring-buffer.ts` | O(1) FIFO with indexed access backing every internal queue below; leaf module             |
-| `view-extent.ts` | A caller-supplied view's byte extent through captured getters; detached/out-of-bounds views (DataViews included) read as empty; `copyExtent` copies an extent; leaf module |
+| `view-extent.ts` | A caller-supplied view's byte extent through captured getters, and any BufferSource's (`bufferSourceExtent`); detached/out-of-bounds inputs (DataViews included) read as empty; `copyExtent` copies an extent; leaf module |
 | `common.ts`   | Helpers shared by the webstreams modules, including `normalizeExpectedLength` (every `expectedLength` and `FixedLengthStream` length converts through it, as C++ converts a uint64_t); leaf module |
 | `queue.ts`    | QUEUED backend: single-queue/multi-cursor, JS sources; fence interfaces; invariant list      |
 | `native.ts`   | NATIVE backend: C++-backed pull conduit; **the C++/JS contract** + invariants                |

src/per_isolate/webstreams/compression.ts

@@ -41,7 +41,6 @@ import type { CommonHelpers } from './common';
 import type { ViewExtentHelpers } from './view-extent';
 
 const {
-  ArrayBufferPrototypeByteLengthGet,
   MathMin,
   ObjectDefineProperties,
   SymbolToStringTag,
@@ -50,7 +49,7 @@ const {
   uncurryThis,
 } = primordials;
 
-const { isArrayBuffer, isArrayBufferView, isSharedArrayBuffer } = utils;
+const { isSharedArrayBuffer } = utils;
 
 // Captured for primordials discipline — ToString coercion per spec.
 const StringCoerce = String;
@@ -67,7 +66,7 @@ const {
 } = require('webstreams/writable');
 const { isActualObject, kEnumerable, toStringTagProperty } =
   require('webstreams/common') as CommonHelpers;
-const { copyExtent, viewByteExtent } =
+const { bufferSourceExtent, copyExtent } =
   require('webstreams/view-extent') as ViewExtentHelpers;
 const { RingBuffer } = require('webstreams/ring-buffer') as {
   RingBuffer: RingBufferConstructor;
@@ -129,22 +128,13 @@ const newCodec = utils.newCompressionCodec as (
 // the spec shape wins over parity with the identity streams, which accept
 // shared views by copying.
 function snapshotChunk(chunk: unknown): Uint8Array {
-  if (isArrayBuffer(chunk)) {
-    return copyExtent(
-      chunk as ArrayBuffer,
-      0,
-      ArrayBufferPrototypeByteLengthGet(chunk) as number
+  const extent = bufferSourceExtent(chunk);
+  if (extent === undefined || isSharedArrayBuffer(extent.buffer)) {
+    throw new TypeError(
+      'The provided value is not of type (ArrayBuffer or ArrayBufferView)'
     );
   }
-  if (isArrayBufferView(chunk)) {
-    const { buffer, byteOffset, byteLength } = viewByteExtent(chunk);
-    if (!isSharedArrayBuffer(buffer)) {
-      return copyExtent(buffer, byteOffset, byteLength);
-    }
-  }
-  throw new TypeError(
-    'The provided value is not of type (ArrayBuffer or ArrayBufferView)'
-  );
+  return copyExtent(extent.buffer, extent.byteOffset, extent.byteLength);
 }
 
 type SnapshotEntry =

src/per_isolate/webstreams/identity.ts

@@ -20,8 +20,9 @@
 //     processChunk copies inside write() for exactly these hazards).
 //   - Zero-length writes are accepted as NO-OPS: the write resolves
 //     immediately without touching the readable queue (no zero-length
-//     chunk enqueued, no backpressure interaction, no pull). A view whose
-//     buffer is detached or out of bounds is zero-length, as in C++.
+//     chunk enqueued, no backpressure interaction, no pull). A detached
+//     ArrayBuffer, and a view whose buffer is detached or out of bounds,
+//     is zero-length, as in C++.
 //
 // See /src/tests/streams/identity/AGENTS.md for the IdentityTransformStream
 // and FixedLengthStream specification.
@@ -118,8 +119,6 @@ const {
   uncurryThis,
 } = primordials;
 
-const { isArrayBufferView, isArrayBuffer, isSharedArrayBuffer } = utils;
-
 const {
   ReadableStream,
   ReadableByteStreamController,
@@ -138,7 +137,7 @@ const {
   normalizeExpectedLength,
   toStringTagProperty,
 } = require('webstreams/common') as CommonHelpers;
-const { copyExtent, viewByteExtent } =
+const { bufferSourceExtent, copyExtent } =
   require('webstreams/view-extent') as ViewExtentHelpers;
 const { RingBuffer } = require('webstreams/ring-buffer') as {
   RingBuffer: RingBufferConstructor;
@@ -187,26 +186,16 @@ function validateAndCopyChunk(chunk: unknown): Uint8Array | undefined {
     if (chunk.length === 0) return undefined;
     return TextEncoderEncode(textEncoderInstance, chunk);
   }
-  if (isArrayBuffer(chunk) || isSharedArrayBuffer(chunk)) {
-    // Read the length through a Uint8Array: its byteLength getter works on
-    // both ArrayBuffer and SharedArrayBuffer (ArrayBuffer.prototype.byteLength
-    // throws on SAB receivers). Constructing it throws for a detached
-    // ArrayBuffer.
-    const byteLength = TypedArrayPrototypeGetByteLength(
-      new Uint8Array(chunk as ArrayBuffer)
-    ) as number;
-    if (byteLength === 0) return undefined;
-    return copyExtent(chunk, 0, byteLength);
-  }
-  if (isArrayBufferView(chunk)) {
-    // A detached or out-of-bounds view is empty (see view-extent.ts).
-    const { buffer, byteOffset, byteLength } = viewByteExtent(chunk);
-    if (byteLength === 0) return undefined;
-    return copyExtent(buffer, byteOffset, byteLength);
+  // A detached or out-of-bounds input is empty (see view-extent.ts).
+  const extent = bufferSourceExtent(chunk);
+  if (extent === undefined) {
+    throw new TypeError(
+      'IdentityTransformStream: chunk must be a BufferSource or string'
+    );
   }
-  throw new TypeError(
-    'IdentityTransformStream: chunk must be a BufferSource or string'
-  );
+  const { buffer, byteOffset, byteLength } = extent;
+  if (byteLength === 0) return undefined;
+  return copyExtent(buffer, byteOffset, byteLength);
 }
 
 let assertIsIdentityTransformStream: (self: IdentityTransformStream) => void;

src/per_isolate/webstreams/view-extent.ts

@@ -1,19 +1,22 @@
 'use strict';
 
-// The byte extent of an ArrayBufferView, read through captured getters, and
-// copies of such extents, for code that consumes caller-supplied views (the
-// internal transform pairs' write paths, the native conduit's enqueue, the
-// digest stream, byte-counting strategies, body collection).
+// The byte extent of an ArrayBufferView or any other BufferSource, read
+// through captured getters, and copies of such extents, for code that
+// consumes caller-supplied bytes (the internal transform pairs' write paths,
+// the native conduit's enqueue, the digest stream, byte-counting strategies,
+// body collection).
 //
 // A view whose buffer has been detached, or that a resizable buffer has
 // shrunk out from under, has no bytes: the typed-array getters report
 // byteOffset and byteLength 0 for it, but the DataView getters throw a
 // TypeError. These helpers give a DataView the typed arrays' answer, so
 // such a view reads as empty whichever kind it is — as it does in the C++
-// implementation, which treats it as zero-length. Leaf module: requires
+// implementation, which treats it as zero-length. A detached ArrayBuffer
+// is empty too: its byteLength getter reports 0. Leaf module: requires
 // nothing.
 
 const {
+  ArrayBufferPrototypeByteLengthGet,
   DataViewPrototypeGetBuffer,
   DataViewPrototypeGetByteLength,
   DataViewPrototypeGetByteOffset,
@@ -25,6 +28,8 @@ const {
   Uint8Array,
 } = primordials;
 
+const { isArrayBuffer, isArrayBufferView, isSharedArrayBuffer } = utils;
+
 interface ViewExtent {
   buffer: ArrayBufferLike;
   byteOffset: number;
@@ -69,6 +74,31 @@ function viewByteExtent(view: ArrayBufferView): ViewExtent {
   return { __proto__: null, buffer, byteOffset, byteLength } as ViewExtent;
 }
 
+// The extent of a BufferSource: a view's (see viewByteExtent), or the whole
+// of an ArrayBuffer or SharedArrayBuffer. Undefined for anything else.
+function bufferSourceExtent(chunk: unknown): ViewExtent | undefined {
+  if (isArrayBufferView(chunk)) return viewByteExtent(chunk);
+  if (isArrayBuffer(chunk)) {
+    return {
+      __proto__: null,
+      buffer: chunk,
+      byteOffset: 0,
+      byteLength: ArrayBufferPrototypeByteLengthGet(chunk),
+    } as ViewExtent;
+  }
+  if (isSharedArrayBuffer(chunk)) {
+    // ArrayBuffer.prototype.byteLength throws on a SharedArrayBuffer, so read
+    // the length through a Uint8Array; shared memory cannot be detached.
+    return {
+      __proto__: null,
+      buffer: chunk,
+      byteOffset: 0,
+      byteLength: TypedArrayPrototypeGetByteLength(new Uint8Array(chunk)),
+    } as ViewExtent;
+  }
+  return undefined;
+}
+
 // Copies byteLength bytes of buffer, from byteOffset, into a new Uint8Array.
 // An empty extent never touches buffer: a detached view's extent is empty,
 // and constructing a view over a detached buffer throws.
@@ -90,7 +120,13 @@ function copyExtent(
 export type ViewExtentHelpers = {
   viewByteLength: typeof viewByteLength;
   viewByteExtent: typeof viewByteExtent;
+  bufferSourceExtent: typeof bufferSourceExtent;
   copyExtent: typeof copyExtent;
 };
 
-module.exports = { viewByteLength, viewByteExtent, copyExtent };
+module.exports = {
+  viewByteLength,
+  viewByteExtent,
+  bufferSourceExtent,
+  copyExtent,
+};

src/tests/streams/identity/AGENTS.md

@@ -93,11 +93,14 @@ readable's queue, before any read occurs.)
 ### Chunk handling
 
 - Accepted chunk types: `ArrayBufferView` (any, honoring offset/length),
-  `ArrayBuffer`, and strings, which are UTF-8 encoded (C++ stops at the
-  first NUL, ledger #25). Anything else fails with `TypeError`.
+  `ArrayBuffer`, `SharedArrayBuffer`, and strings, which are UTF-8 encoded
+  (C++ stops at the first NUL, ledger #25). Shared memory is copied like
+  any other bytes. Anything else fails with `TypeError`.
 - Zero-length chunks (empty view, empty buffer, empty string) are no-ops:
   they resolve without delivering a chunk and without closing the stream.
-  So is an undefined chunk (`write(undefined)`, `write()`).
+  So is an undefined chunk (`write(undefined)`, `write()`). A detached
+  buffer, and a view whose buffer is detached or out of bounds, is
+  zero-length.
 - Writes **copy** their bytes; the delivered chunk never aliases the
   caller's buffer, and the caller's buffer is never detached by a write.
   The copy is taken synchronously inside `write()` in both
@@ -251,7 +254,7 @@ pattern; a change to either side fails its cell.
 | 9 | Writes after abort | `TypeError` "This WritableStream has been closed." | original abort reason | `abortRejectsSubsequentWrites` |
 | 10 | Writes after cancel with close in flight | closed `TypeError` | original cancel reason | `cancelRejectsPendingWriteAndClose` |
 | 11 | FLS enforcement | read-side `TypeError`; the offending write/close succeeds | eager write-side `RangeError`; readable errors too | `fixed-length-errors.js` |
-| 12 | Already-detached `ArrayBuffer` chunk | zero-length no-op | rejects `TypeError` (per-write; the stream survives) | `alreadyDetachedBufferAtWrite` |
+| 12 | (no divergence) Already-detached `ArrayBuffer` chunk: a zero-length no-op, as a detached view is, adding nothing toward a `FixedLengthStream`'s length | same | same | `alreadyDetachedBufferAtWrite` (parity) |
 | 13 | Single tee-branch cancel promise | resolves immediately | WHATWG semantics: shared promise, settles when both branches cancel | `cancelOneBranchKeepsWriterFlowing` |
 | 14 | Write after both tee branches cancel | parks forever (composite cancel not propagated to the writable) | rejects `AggregateError` "All readable stream tee branches were canceled" — also a write in flight when the last branch cancels, even if a branch had read it | `writeAfterBothBranchesCancel`, `inFlightWriteRejectsWhenBranchesLeaveAfterRead` |
 | 15 | Piping between identity streams | not implemented: `pipeTo()` takes both locks then rejects `TypeError` ("Inter-TransformStream ReadableStream.pipeTo() is not implemented."); `pipeThrough()` throws it synchronously | fully functional: delivery, completion, and error propagation in both directions with original reason instances; circular `pipeThrough(its)` currently succeeds and locks both sides — `TODO(streams-ts)`: it should fail | `pipe-integration.js` |
@@ -272,10 +275,10 @@ pattern; a change to either side fails its cell.
 | --- | --- |
 | `api-surface.js` | toStringTag branding; `FixedLengthStream` subclassing; `readable`/`writable` are `ReadableStream`/`WritableStream` instances, stable, enumerable prototype accessors (placement per ledger #5); constructor source text (native code under C++, not under TS); accessor brand checks |
 | `construction.js` | valid lengths (0, 5, −0.0, `MAX_SAFE_INTEGER`, bigints, with strategy); coerced length observable via HWM cap; invalid, coerced and above-2^53−1 lengths, the same in both (ledger #1–3); inheritance (ledger #4); a user-supplied strategy `size` is never invoked (ITS and FLS, with and without explicit HWM); the strategy argument converts as a WebIDL dictionary (undefined/null mean none, other non-objects throw `TypeError`); `highWaterMark` read and converted once by both classes (numeric strings, booleans, null, `valueOf`), its conversion per ledger #24 |
-| `chunk-types.js` | accepted: `Uint8Array`, `ArrayBuffer`, `DataView` subrange, string→UTF-8 (a NUL truncates under C++, ledger #25), subarray offsets; rejected: numbers, plain objects (`TypeError`; per-write — the stream survives, ledger #6); an invalid chunk queued behind valid writes surfaces its error in FIFO order — earlier writes still deliver and later traffic still flows in both implementations |
+| `chunk-types.js` | accepted: `Uint8Array`, `ArrayBuffer`, `DataView` subrange, shared memory (a `SharedArrayBuffer` and views over one, copied at `write()`; an empty one is a no-op), string→UTF-8 (a NUL truncates under C++, ledger #25), subarray offsets; rejected: numbers, plain objects (`TypeError`; per-write — the stream survives, ledger #6); an invalid chunk queued behind valid writes surfaces its error in FIFO order — earlier writes still deliver and later traffic still flows in both implementations |
 | `zero-length-writes.js` | empty view / buffer / string are non-closing no-ops; so is an undefined chunk (`write(undefined)`, `write()`; ITS, FLS, with a byte highWaterMark) |
 | `copy-semantics.js` | delivered chunk never aliases the source; source mutation after delivery is invisible; source is not detached |
-| `buffer-lifecycle.js` | write-time snapshot survives later resize/detach in both implementations; degenerate write-time inputs (already-detached per ledger #12; detached and out-of-bounds typed-array and DataView views are no-ops); shadowing/throwing metadata getters never consulted |
+| `buffer-lifecycle.js` | write-time snapshot survives later resize/detach in both implementations; degenerate write-time inputs are no-ops (an already-detached buffer, ITS and FLS, ledger #12; detached and out-of-bounds typed-array and DataView views); shadowing/throwing metadata getters never consulted |
 | `ordering.js` | 1:1 write/read correspondence in both interleavings; multi-chunk aggregate integrity; clean EOF tails |
 | `byob.js` | BYOB reader support; partial fills across reads with write completion on full consumption (still pending a macrotask after a partial read); reads spanning queued writes, past zero-length and invalid ones, stopping at a close, through FLS (#22); lying destination extents (at call and after enqueue) with sentinel overwrite guards; read-call validation (zero-length view, non-view, missing argument); input buffer detached by read with non-detachable (SAB-backed) destinations rejected; repeated EOF zero-length views with preserved buffers |
 | `backpressure.js` | writes and close queue unboundedly with settlement on consumption (a write read ahead of its turn settles when the writable reaches it, so recovery is asserted after the writes settle); a partial BYOB read keeps the write pending and its bytes counted in `desiredSize`; advisory overfill (negative `desiredSize`); default HWM 1 with divergent accounting (ledger #17); explicit HWM as initial `desiredSize` (negative-zero HWM normalized to +0); byte-level tracking incl. in-flight bytes; string accounting by exact UTF-8 bytes, ITS and FLS (parity); `ready` replacement and recovery |

src/tests/streams/identity/buffer-lifecycle.js

@@ -12,12 +12,11 @@
 // size callback — the one hook the writable machinery runs synchronously
 // inside write() — for the same reasons.
 //
-// Degenerate inputs that are already unusable at write() time (detached
-// buffer, view made out-of-bounds by a shrink) are a separate matter and
-// partly diverge; see the individual tests at the bottom.
+// Inputs already unusable at write() time (a detached buffer, a view made
+// out-of-bounds by a shrink) have no bytes, so both implementations treat
+// them as zero-length no-ops; see the individual tests at the bottom.
 
-import { strictEqual, deepStrictEqual, rejects } from 'node:assert';
-import { usingTsImpl } from 'which-impl';
+import { strictEqual, deepStrictEqual } from 'node:assert';
 
 // Writes chunk, synchronously runs mutate (before any await, exactly as a
 // misbehaving caller would), and asserts the reader receives expectedBytes.
@@ -87,26 +86,17 @@ export const bufferDetachedAfterWrite = {
 
 export const alreadyDetachedBufferAtWrite = {
   async test() {
-    // Writing a buffer that is already detached diverges:
-    // - C++ observes byteLength 0 and treats it as a zero-length no-op.
-    // - TypeScript rejects the write with TypeError (the chunk cannot be
-    //   read at all); like any other invalid chunk the rejection is
-    //   per-write — the stream survives it.
-    const ab = new ArrayBuffer(4);
-    ab.transfer();
-    const { readable, writable } = new IdentityTransformStream();
-    const writer = writable.getWriter();
-    const reader = readable.getReader();
-    if (usingTsImpl) {
-      await rejects(writer.write(ab), TypeError);
-      // The stream is usable after the rejection.
-      const writePromise = writer.write(new Uint8Array([9]));
-      const { value, done } = await reader.read();
-      strictEqual(done, false);
-      strictEqual(value[0], 9);
-      await writePromise;
-      await writer.close();
-    } else {
+    // A buffer that is already detached has byteLength 0, so both
+    // implementations treat it as a zero-length no-op, as they do a
+    // detached view. It adds nothing toward a FixedLengthStream's length.
+    for (const stream of [
+      new IdentityTransformStream(),
+      new FixedLengthStream(0),
+    ]) {
+      const ab = new ArrayBuffer(4);
+      ab.transfer();
+      const writer = stream.writable.getWriter();
+      const reader = stream.readable.getReader();
       await writer.write(ab);
       const closePromise = writer.close();
       strictEqual((await reader.read()).done, true);