diff --git a/src/per_isolate/webstreams/AGENTS.md b/src/per_isolate/webstreams/AGENTS.md index cafadd15847..0d5adec0c92 100644 --- a/src/per_isolate/webstreams/AGENTS.md +++ b/src/per_isolate/webstreams/AGENTS.md @@ -30,9 +30,10 @@ private-brand dispatch, no `instanceof`) apply here — see lists (`queue.ts` and `native.ts` headers). - The native source contract (marker symbol, standard pull/cancel hooks, byobRequest discrimination, once-per-pull delivery, per-pull abort - signal for cancellation, under-delivery = fused - `{done: true, value: partial}` EOF, tee hook, `expectedLength` - exact-total byte contract) is specified in the `native.ts` header. + signal for cancellation, under-delivery = EOF signal delivering the + partial as `{done: false, value: partial}` with the next read + observing EOF, tee hook, `expectedLength` exact-total byte contract) + is specified in the `native.ts` header. The C++ implementation (`ReadableStreamNativeSource` in `src/workerd/api/js-readable-stream.{h,c++}`) MUST conform to it; JS mocks in tests exercise the conduit independently. Key addition: diff --git a/src/per_isolate/webstreams/compression.ts b/src/per_isolate/webstreams/compression.ts index e9868adbf82..1e37eb1453b 100644 --- a/src/per_isolate/webstreams/compression.ts +++ b/src/per_isolate/webstreams/compression.ts @@ -108,17 +108,17 @@ function isActualObject(value: unknown): boolean { } // True for BufferSource chunks the codec accepts: ArrayBuffers and views, -// excluding anything SharedArrayBuffer-backed (per Web IDL, [AllowShared] is -// not granted here; WPT pins the rejection). Captured getters are used for -// the view's buffer — prototype accessors are user-patchable. +// INCLUDING anything SharedArrayBuffer-backed — the shared bytes are +// copied out by snapshotChunk, matching the identity streams and the C++ +// implementation (DECIDED 2026-08-28; the strict [AllowShared]-less +// BufferSource reading was considered and overridden for internal +// consistency, which is why the WPT bad-chunks files are disabled). function isValidChunk(chunk: unknown): boolean { - if (isArrayBuffer(chunk)) return true; - if (isSharedArrayBuffer(chunk)) return false; - if (!isArrayBufferView(chunk)) return false; - const buffer = isDataView(chunk) - ? DataViewPrototypeGetBuffer(chunk) - : TypedArrayPrototypeGetBuffer(chunk); - return !isSharedArrayBuffer(buffer); + return ( + isArrayBuffer(chunk) || + isSharedArrayBuffer(chunk) || + isArrayBufferView(chunk) + ); } // Validates a chunk and copies its CURRENT bytes. Runs synchronously @@ -141,6 +141,15 @@ function snapshotChunk(chunk: unknown): Uint8Array { buffer = chunk as ArrayBuffer; byteOffset = 0; byteLength = ArrayBufferPrototypeByteLengthGet(chunk) as number; + } else if (isSharedArrayBuffer(chunk)) { + // A raw SharedArrayBuffer: measure through a fresh whole-buffer view + // (SharedArrayBuffer.prototype.byteLength is not among the primordial + // captures; the view's length is read through the captured getter). + buffer = chunk as unknown as ArrayBuffer; + byteOffset = 0; + byteLength = TypedArrayPrototypeGetByteLength( + new Uint8Array(chunk as unknown as ArrayBuffer) + ) as number; } else if (isDataView(chunk)) { buffer = DataViewPrototypeGetBuffer(chunk) as ArrayBuffer; byteOffset = DataViewPrototypeGetByteOffset(chunk) as number; @@ -251,12 +260,12 @@ function createCodecPair( ); } if (!entry.ok) { - // An invalid chunk errors BOTH sides, matching the legacy - // implementation (any write failure errored the whole pair) — - // without this the readable side would hang on its pending - // pull. - failBoth(entry.error); - throw entry.error; + // An invalid chunk rejects ITS OWN write only — the non-fatal + // write-rejection channel keeps the pair usable and later + // queued writes still deliver (the per-write invalid-chunk + // contract shared with the identity streams and the C++ + // implementation). Codec failures below remain FATAL. + throw writableInternals.nonFatalWriteRejection(entry.error); } // EAGER: the codec consumes the snapshot; a codec error throws // HERE, rejecting the write — the spec's transform-time error diff --git a/src/per_isolate/webstreams/identity.ts b/src/per_isolate/webstreams/identity.ts index a91b1bf2528..83669602588 100644 --- a/src/per_isolate/webstreams/identity.ts +++ b/src/per_isolate/webstreams/identity.ts @@ -410,13 +410,13 @@ class IdentityTransformStream { ); } const entry = ArrayPrototypeShift(snapshots) as SnapshotEntry; - // A recorded validation error surfaces here, at its FIFO turn: this - // write rejects and the stream errors, but everything written before - // it has already been delivered. Queued writes are discarded by the - // erroring writable without sink steps; their snapshots go with them. + // A recorded validation error surfaces here, at its FIFO turn, as a + // NON-FATAL write rejection: this write's promise rejects while the + // stream stays usable and queued writes behind it still deliver — + // the per-write invalid-chunk contract shared with the C++ internal + // controllers. if (!entry.ok) { - snapshots.length = 0; - throw entry.error; + throw writableInternals.nonFatalWriteRejection(entry.error); } const copied = entry.copied; if (copied === undefined) return; // zero-length no-op diff --git a/src/per_isolate/webstreams/native.ts b/src/per_isolate/webstreams/native.ts index 590f9fc961b..48c10391c6c 100644 --- a/src/per_isolate/webstreams/native.ts +++ b/src/per_isolate/webstreams/native.ts @@ -49,17 +49,16 @@ // accumulation toward `atLeast` happens inside the native source; // the conduit never re-pulls an unsatisfied read. Responding with // fewer bytes than `atLeast` IMPLICITLY SIGNALS CLOSURE: the partial -// fill commits fused as { done: true, value: partialView }, the -// stream closes, and every subsequent read returns EOF. This result -// is deliberately INDISTINGUISHABLE from the standard's one fused -// corner — a pending min read committed in the closed state via -// close() + respond(0) (RespondInClosedState → -// CommitPullIntoDescriptor, which fulfills the read-into request -// with the partial view and done: true). The signaling MECHANISM -// differs (a queued JS source is pulled again and must close() -// explicitly; for a native source the under-delivery itself is the -// signal), but consumers observe the same result shapes on both -// backends. +// fill commits as { done: false, value: partialView }, the stream +// closes, and every subsequent read returns EOF (BYOB reads get a +// zero-length view over their transferred buffer). This is the +// decided C++-parity readAtLeast tail shape — the below-minimum +// tail is never fused with the done flag — and it matches the +// queued backend's deferred end-of-data commit, so consumers +// observe the same tail on both backends. (Only an explicit +// same-turn respond(0)-after-close on the QUEUED backend produces +// the spec's fused { done: true, value: partial } fold; the native +// conduit has no respond(0) — close() is the EOF verb.) // - Close is otherwise a fused close-commit: controller.close() may // follow a respond() in the same pull turn — deliberately divergent // from the queued backend's drain-then-sentinel model. @@ -979,16 +978,17 @@ class NativePullConduit implements ByteStreamConsumerType { } // MIN-READ CONTRACT: the respond is the source's COMPLETE answer for // this read. Delivering fewer bytes than the requested minimum - // (atLeast) signals end-of-stream: the partial fill commits FUSED - // with done: true, the stream closes, and every subsequent read - // returns EOF. The conduit never re-pulls an unsatisfied read — any + // (atLeast) signals end-of-stream: the partial fill commits + // { done: false, value: partialView }, the stream closes, and every + // subsequent read returns EOF — the decided C++-parity readAtLeast + // tail shape. The conduit never re-pulls an unsatisfied read — any // accumulation toward the minimum happens inside the native source // (tryRead-style). State transitions before promise resolutions. // // EXPECTED-LENGTH: under-delivery is a close signal, so the // exact-total contract applies — landing short of expectedLength is // underflow and errors the stream instead of closing it. (Landing - // exactly on it is a legitimate fused close.) + // exactly on it is a legitimate close-with-delivery.) if ( this.#expectedLength !== undefined && this.#bytesDelivered < this.#expectedLength @@ -1005,7 +1005,7 @@ class NativePullConduit implements ByteStreamConsumerType { return; } this.#status = 'closed'; - desc.resolve({ done: true, value }); + desc.resolve({ done: false, value }); this.#settleRemainingAsEof(); this.#hooks.closeStream(); } @@ -1075,11 +1075,11 @@ class NativePullConduit implements ByteStreamConsumerType { this.#byobRequestCache = null; // DEFENSIVE, currently unreachable: under the min-read contract every - // respond commits (satisfied → done: false; under-delivered → fused - // EOF), so a partially-filled descriptor cannot persist to close(). - // Retained because the contract describes close() fusing with a - // partial fill, should a future revision reintroduce partial - // responds. + // respond commits (satisfied → done: false; under-delivered → the + // done: false tail followed by EOF), so a partially-filled descriptor + // cannot persist to close(). Retained should a future revision + // reintroduce partial responds; the shape mirrors the under-delivery + // tail (partial done: false, then #settleRemainingAsEof). const head = this.#requests[0]; if ( head !== undefined && @@ -1101,7 +1101,7 @@ class NativePullConduit implements ByteStreamConsumerType { } ArrayPrototypeSplice(this.#requests, 0, 1); desc.resolve({ - done: true, + done: false, value: new desc.viewCtor( desc.buffer, desc.byteOffset, diff --git a/src/per_isolate/webstreams/queue.ts b/src/per_isolate/webstreams/queue.ts index c9a0ded1107..4e9cc4e30ca 100644 --- a/src/per_isolate/webstreams/queue.ts +++ b/src/per_isolate/webstreams/queue.ts @@ -15,9 +15,16 @@ // backend in native.ts has a DIFFERENT set — do not port logic across the // fence without checking both): // - The CLOSE_SENTINEL is always the LAST slot; it is the ONLY close- -// propagation mechanism (drain-then-close per cursor; BYOB descriptors -// stay pending at the sentinel per the spec footgun — the native -// backend deliberately differs with its fused close-commit). +// propagation mechanism (drain-then-close per cursor). BYOB +// descriptors facing the sentinel settle via the DEFERRED +// end-of-data commit (one microtask): a same-turn respond(0) first +// claims the spec's RespondInClosedState fold (done: true with the +// partial bytes); descriptors nobody responds to then commit with +// the decided C++-parity tail shape — element-aligned partial fills +// resolve { done: false, value: partial } and unfilled descriptors +// resolve { done: true, value: empty view }. (The spec as written +// leaves unresponded descriptors pending forever — a footgun we +// deliberately do not reproduce.) // - read() takes the fast path ONLY when no reads are pending (per-reader // FIFO; entries and pending reads coexist under batched notification). // - Byte entries are {buffer, byteOffset, byteLength} triples; queue @@ -51,6 +58,7 @@ const { FinalizationRegistryPrototypeUnregister, MathMin, ObjectCreate, + PromisePrototypeThen, PromiseResolve, PromiseReject, PromiseWithResolvers, @@ -832,6 +840,10 @@ class ByteStreamCursor // autoAllocateChunkSize creates synthetic descriptors for default reads. #pendingPullIntos: PullIntoDescriptor[] = []; + // One-shot latch for the deferred end-of-data commit (see + // #scheduleEndOfDataCommit). + #endOfDataCommitScheduled: boolean = false; + // Callback invoked when the cursor detects a fractional-element fill at // the close sentinel — the stream must be errored with a TypeError. Set // by the controller (the cursor layer cannot error the stream directly). @@ -914,13 +926,14 @@ class ByteStreamCursor // descriptor construction. Same FIFO invariant as read(): the fast path // is taken only when nothing is already pending. // - // PRECONDITION: the stream is readable. Unlike read(), the cursor does - // not resolve BYOB reads at the sentinel — a read submitted after close - // must be resolved by the READER layer with { done: true, value: - // zero-length view over the transferred buffer } before ever reaching - // here, and errored streams reject at the reader layer. A descriptor - // pushed here while the queue is already closed would pend until - // respond(0)/cancel/release, per the byte-cursor close matrix. + // PRECONDITION: the owning stream is readable. A read submitted after + // the STREAM state flips to closed is resolved by the READER layer with + // { done: true, value: zero-length view over the transferred buffer } + // before ever reaching here, and errored streams reject at the reader + // layer. The stream can still be readable while the QUEUE is already + // closed (buffered data not yet drained — e.g. a tee branch that has + // not read); a below-minimum fill that lands at the sentinel then + // settles via the deferred end-of-data commit. readBYOB( desc: PullIntoDescriptor ): Promise> { @@ -929,21 +942,23 @@ class ByteStreamCursor if (desc.bytesFilled >= desc.minimumFill) { return PromiseResolve(createReadResult(this.#convert(desc), false)); } - // After filling, if the cursor is at the close sentinel and the - // descriptor has a fractional element fill, the remaining bytes can - // never complete an element — error the stream immediately. - if ( - desc.bytesFilled > 0 && - desc.bytesFilled % desc.elementSize !== 0 && - this.queue.getEntry(this.position) === CLOSE_SENTINEL - ) { - const e = new TypeError( - 'Insufficient bytes to fill elements in the given view' - ); - if (this.#errorStreamCallback !== undefined) { - this.#errorStreamCallback(e); + if (this.queue.getEntry(this.position) === CLOSE_SENTINEL) { + // The cursor faces the sentinel: no more data can ever arrive + // for this descriptor. + if (desc.bytesFilled > 0 && desc.bytesFilled % desc.elementSize !== 0) { + // Fractional element fill — the remaining bytes can never + // complete an element; error the stream immediately. + const e = new TypeError( + 'Insufficient bytes to fill elements in the given view' + ); + if (this.#errorStreamCallback !== undefined) { + this.#errorStreamCallback(e); + } + return PromiseReject(e); } - return PromiseReject(e); + // Element-aligned (possibly empty) fill: settle via the deferred + // end-of-data commit. + this.#scheduleEndOfDataCommit(); } } ArrayPrototypePush(this.#pendingPullIntos, desc); @@ -975,16 +990,15 @@ class ByteStreamCursor // follow default-read close semantics: ReadableStreamClose drains // read requests with done, so they resolve { done: true } now. this.#resolveDefaultPullIntosAsDone(); - // TRUE BYOB descriptors in single-cursor mode are NOT committed - // here — the source can call respond(0)-while-closed, which - // reaches commitPullIntosOnClose() via the controller. But in - // multi-cursor mode (tee branches), byobRequest is null and - // respond(0) is unreachable, so we must commit them now. This - // is the equivalent of the spec's per-branch controller running - // ReadableByteStreamControllerRespondInClosedState. - if (this.queue.singleCursor === undefined) { - this.commitPullIntosOnClose(); - } + // TRUE BYOB descriptors settle via the deferred end-of-data + // commit: the source may still call respond(0) in this same turn + // (RespondInClosedState reaches commitPullIntosOnClose() via the + // controller and claims the spec's fold shape first); whatever is + // left when the microtask runs commits with the decided C++- + // parity tail shape. In multi-cursor mode (tee branches), + // byobRequest is null and respond(0) is unreachable, so the + // deferred commit is what settles every branch descriptor. + this.#scheduleEndOfDataCommit(); break; } const head = this.#pendingPullIntos[0] as PullIntoDescriptor; @@ -1031,10 +1045,62 @@ class ByteStreamCursor return false; } + // The deferred end-of-data commit. Scheduled (once per turn) whenever + // BYOB descriptors face the close sentinel — from notify() when close() + // lands with reads parked, and from readBYOB() when a read submitted + // against a closed-but-undrained queue fills below its minimum. The + // one-microtask deferral is load-bearing: a source that calls + // respond(0)/respondWithNewView(empty) in the same turn as close() + // commits first via commitPullIntosOnClose() and claims the spec's + // RespondInClosedState fold shape ({ done: true, value: partial }, the + // WPT read-min pinned behavior); only descriptors nobody responds to + // fall through to the decided C++-parity tail shape here. + #scheduleEndOfDataCommit(): void { + if (this.#endOfDataCommitScheduled) return; + this.#endOfDataCommitScheduled = true; + PromisePrototypeThen(PromiseResolve(), () => { + this.#endOfDataCommitScheduled = false; + this.#commitPullIntosAtEndOfData(); + }); + } + + // Settle every still-pending descriptor with the decided tail shape: + // element-aligned partial fills resolve { done: false, value: partial } + // (a subsequent read observes the closed stream and resolves done with + // an empty view — the C++ readAtLeast tail contract); unfilled + // descriptors resolve { done: true, value: empty view }, handing the + // transferred buffer back. Bails when the cursor no longer faces the + // sentinel: error() dropped the entries (getEntry returns undefined), + // and cancel()/respond(0) paths already emptied the descriptor list. + #commitPullIntosAtEndOfData(): void { + if (this.queue.getEntry(this.position) !== CLOSE_SENTINEL) return; + const pending = this.#pendingPullIntos; + this.#pendingPullIntos = []; + for (let i = 0; i < pending.length; i++) { + const desc = pending[i] as PullIntoDescriptor; + // 'none': released reader (or an auto-allocate read already + // resolved by #resolveDefaultPullIntosAsDone) — nothing to settle. + if (desc.readerType === 'none') continue; + if (desc.readerType === 'default') { + // Defensive: auto-allocate default reads normally resolve in + // #resolveDefaultPullIntosAsDone before this runs. Default-read + // close semantics: done with value undefined, never a view. + desc.resolve(createReadResult(undefined, true)); + } else if (desc.bytesFilled > 0) { + // assert: desc.bytesFilled % desc.elementSize === 0 (fractional + // fills errored the stream before any commit could be scheduled) + desc.resolve(createReadResult(this.#convert(desc), false)); + } else { + desc.resolve(createReadResult(this.#convert(desc), true)); + } + } + } + // Resolve synthetic default-read descriptors (autoAllocateChunkSize) as - // done at end-of-stream; keep true BYOB descriptors pending per the close - // matrix. In practice the list is homogeneous (single reader type at a - // time), so the filtering is defensive. + // done at end-of-stream; true BYOB descriptors are left for the deferred + // end-of-data commit (or a same-turn respond(0)). In practice the list + // is homogeneous (single reader type at a time), so the filtering is + // defensive. // // The descriptor is KEPT in the list (not shifted) so that a subsequent // respond(0) finds it and doesn't throw. commitPullIntosOnClose skips @@ -1128,9 +1194,12 @@ class ByteStreamCursor // Commit all pending pull-into descriptors at end-of-stream: resolve with // the filled-so-far view (possibly zero-length — the buffer is handed - // back) and done: true in a SINGLE result. Called by the controller from - // the respond(0)-while-closed path. A fractional-element fill never - // reaches this point: controller.close() throws for it. + // back) and done: true in a SINGLE result — the spec's + // RespondInClosedState fold shape. Called by the controller from the + // respond(0)-while-closed path ONLY; descriptors the source never + // responds to settle through #commitPullIntosAtEndOfData instead, with + // the split tail shape. A fractional-element fill never reaches this + // point: controller.close() throws for it. commitPullIntosOnClose(): void { const pending = this.#pendingPullIntos; this.#pendingPullIntos = []; diff --git a/src/per_isolate/webstreams/readable.ts b/src/per_isolate/webstreams/readable.ts index e7e1e8f7334..94b26408fb8 100644 --- a/src/per_isolate/webstreams/readable.ts +++ b/src/per_isolate/webstreams/readable.ts @@ -358,14 +358,20 @@ let extractNativeSource: (this: ReadableStream) => object; // The non-standard expectedLength pass-through for the DrainingReader // (and the C++ bridge). Chained like the other controller helpers: -// default → undefined; queued byte → cached construction value; native → -// cached construction value (joined in ReadableStream's static block). +// default → the value installed by the TransformStream expectedLength +// extension (undefined otherwise); queued byte → cached construction +// value; native → cached construction value (joined in ReadableStream's +// static block). let getControllerExpectedLength: ( controller: | ReadableStreamDefaultControllerType | ReadableByteStreamControllerType | NativeReadableStreamControllerType ) => bigint | undefined; +let setDefaultControllerExpectedLength: ( + controller: ReadableStreamDefaultController, + length: bigint | undefined +) => void; let setReadableStreamPendingClosure: (stream: ReadableStream) => void; let isReadableStreamPendingClosure: (stream: ReadableStream) => boolean; @@ -781,6 +787,38 @@ function defaultReaderReadInternal( ); } +// Submit a default-style read through the BYOB machinery via a synthetic +// auto-allocate pull-into descriptor (spec ReadableByteStreamController +// PullSteps step 3, [[autoAllocateChunkSize]] present): the source's pull +// then observes a byobRequest over the auto-allocated buffer. Shared by +// the default reader's read path and the draining reader's +// empty-fallback wait-read (the body/pipe pump). +function readViaAutoAllocateDescriptor( + consumer: ByteStreamConsumerType, + autoAllocateChunkSize: number, + reader: object +): Promise> { + const withResolvers = PromiseWithResolvers() as PromiseWithResolversType< + ReadableStreamReadResult + >; + const descriptor: PullIntoDescriptor = { + buffer: new ArrayBuffer(autoAllocateChunkSize), + bufferByteLength: autoAllocateChunkSize, + byteOffset: 0, + byteLength: autoAllocateChunkSize, + bytesFilled: 0, + minimumFill: 1, + elementSize: 1, + viewCtor: Uint8Array, + readerType: 'default', + promise: withResolvers.promise, + resolve: withResolvers.resolve, + reject: withResolvers.reject, + reader, + }; + return consumer.readBYOB(descriptor); +} + // Async continuation of defaultReaderReadInternal for cases where // the data is not synchronously available. async function defaultReaderReadInternalAsync( @@ -800,26 +838,11 @@ async function defaultReaderReadInternalAsync( const autoAllocateChunkSize = getByteControllerAutoAllocateChunkSize( controller as ReadableByteStreamController ); - const withResolvers = PromiseWithResolvers() as PromiseWithResolversType< - ReadableStreamReadResult - >; - const descriptor: PullIntoDescriptor = { - buffer: new ArrayBuffer(autoAllocateChunkSize as number), - bufferByteLength: autoAllocateChunkSize as number, - byteOffset: 0, - byteLength: autoAllocateChunkSize as number, - bytesFilled: 0, - minimumFill: 1, - elementSize: 1, - viewCtor: Uint8Array, - readerType: 'default', - promise: withResolvers.promise, - resolve: withResolvers.resolve, - reject: withResolvers.reject, - reader, - }; - const byteConsumer = consumer as unknown as ByteStreamConsumerType; - promise = byteConsumer.readBYOB(descriptor); + promise = readViaAutoAllocateDescriptor( + consumer as unknown as ByteStreamConsumerType, + autoAllocateChunkSize as number, + reader + ); } else { promise = consumer.read(reader); } @@ -1201,6 +1224,12 @@ class ReadableStreamDefaultController< > implements ReadableStreamDefaultControllerType { #stream: ReadableStream; #queue: StreamQueueType; + // The non-standard expectedLength pass-through (undefined = unknown). + // Default controllers never read it from an underlying source — it is + // set ONLY through the internal setter, by the TransformStream + // constructor's workerd `expectedLength` extension, so the C++ bridge + // can derive Content-Length for bodies built from such transforms. + #expectedLength: bigint | undefined = undefined; // Algorithms are cleared (closures dropped) on close-complete, error, and // cancel, per spec ClearAlgorithms. #sizeAlgorithm: ((chunk: R) => number) | undefined; @@ -1235,10 +1264,17 @@ class ReadableStreamDefaultController< // The byte controller wires its own branch in the byte pass. }; - // expectedLength is byte-stream-only: the default controller never - // reads it from the source (silently ignored if declared) and always - // reports undefined. - getControllerExpectedLength = () => undefined; + // Default controllers never read expectedLength from an underlying + // source (silently ignored if declared); it is populated only via the + // internal setter (the TransformStream `expectedLength` extension). + // The byte and native arms of this chain report their own cached + // values. + getControllerExpectedLength = (controller) => + (controller as ReadableStreamDefaultController).#expectedLength; + + setDefaultControllerExpectedLength = (controller, length) => { + controller.#expectedLength = length; + }; controllerCancelSteps = (controller, reason) => { if (#queue in controller) { @@ -2305,7 +2341,28 @@ async function drainingReaderReadInternal( if (result.chunks.length === 0 && !result.done) { // Nothing buffered — wait for one chunk through the normal pending-read // machinery (FIFO with everything else), then sweep the rest. - const promise = consumer.read(reader); + // + // QUEUED-BYTE-SPECIFIC (sanctioned, mirrors defaultReaderReadInternal): + // with autoAllocateChunkSize set, the wait-read goes through the BYOB + // 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); + } if (controller !== undefined) controllerPullIfNeeded(controller); const single = await promise; if (single.done) { @@ -2485,6 +2542,13 @@ function pipeToInternal( undefined ) as Promise; + // Set when a write rejects NON-FATALLY (dest still writable — the + // internal transforms' invalid-chunk contract) while a clean shutdown + // is already waiting for write acknowledgment; runAction upgrades the + // clean close into a pipe failure with this reason. See + // onWriteRejectedNonFatally. + let pendingNonFatalWriteFailure: { reason: unknown } | undefined; + const shutdownWithAction = ( action: (() => Promise) | undefined, error?: { reason: unknown } @@ -2493,6 +2557,25 @@ function pipeToInternal( shuttingDown = true; const runAction = (): void => { + // A non-fatal tail-write rejection recorded during the + // acknowledgment wait upgrades a CLEAN shutdown into a failure + // (the C++ pipe, which awaits each write, can never reach its + // close step past a failed write). Shutdowns that already carry + // an error keep it — first cause wins. + if (error === undefined && pendingNonFatalWriteFailure !== undefined) { + const reason = pendingNonFatalWriteFailure.reason; + const failureActions = nonFatalWriteFailureActions(reason); + if (failureActions.length === 0) { + finalize({ reason }); + return; + } + PromisePrototypeThen( + combineShutdownActions(failureActions), + () => finalize({ reason }), + (actionError: unknown) => finalize({ reason: actionError }) + ); + return; + } if (action === undefined) { finalize(error); return; @@ -2550,6 +2633,78 @@ function pipeToInternal( ); }; + // Runs `actions` in parallel and settles when all have settled, + // rejecting with the first failure (spec: shutdown actions run in + // parallel). Shared by the abort-signal algorithm and the non-fatal + // write-rejection path. + const combineShutdownActions = ( + actions: (() => Promise)[] + ): Promise => { + let remaining = actions.length; + let failed: { reason: unknown } | undefined; + const all = PromiseWithResolvers() as PromiseWithResolversType; + for (let i = 0; i < actions.length; i++) { + const action = actions[i] as () => Promise; + PromisePrototypeThen( + action(), + () => { + if (--remaining === 0) { + if (failed !== undefined) { + all.reject(failed.reason); + } else { + all.resolve(); + } + } + }, + (e: unknown) => { + failed ??= { reason: e }; + if (--remaining === 0) all.reject(failed.reason); + } + ); + } + return all.promise; + }; + + // The failure actions for a non-fatal write rejection: abort the + // destination (unless preventAbort) and cancel the source (unless + // preventCancel), both with the write's reason — the C++ pipe outcome + // for a rejected write. + const nonFatalWriteFailureActions = ( + e: unknown + ): (() => Promise)[] => { + const actions: (() => Promise)[] = []; + if (!preventAbort) { + ArrayPrototypePush(actions, () => + writableInternals.getState(destination) === 'writable' + ? writableInternals.writableStreamAbort(destination, e) + : (PromiseResolve(undefined) as Promise) + ); + } + if (!preventCancel) { + ArrayPrototypePush(actions, () => + getReadableStreamGetState(source) === 'readable' + ? readableStreamCancel(source, e) + : (PromiseResolve(undefined) as Promise) + ); + } + return actions; + }; + + // Workerd extension: the internal transforms (identity, compression) + // reject an invalid chunk's write NON-FATALLY, leaving the destination + // writable — a state unreachable under pure WHATWG semantics, where any + // sink rejection errors the destination (and the closed-promise + // observer handles it). Without this, the failed chunk would be + // silently dropped and the pipe would complete. Match the C++ pipe + // outcome instead: treat the rejection as a pipe failure. + const onWriteRejectedNonFatally = (e: unknown): void => { + const actions = nonFatalWriteFailureActions(e); + shutdownWithAction( + actions.length === 0 ? undefined : () => combineShutdownActions(actions), + { reason: e } + ); + }; + if (signal !== undefined) { abortAlgorithm = () => { const abortReason = AbortSignalReasonGet(signal); @@ -2573,34 +2728,7 @@ function pipeToInternal( shutdownWithAction( actions.length === 0 ? undefined - : () => { - // Settle when all actions settle; reject with the first - // failure (spec: shutdown actions run in parallel). - let remaining = actions.length; - let failed: { reason: unknown } | undefined; - const all = - PromiseWithResolvers() as PromiseWithResolversType; - for (let i = 0; i < actions.length; i++) { - const action = actions[i] as () => Promise; - PromisePrototypeThen( - action(), - () => { - if (--remaining === 0) { - if (failed !== undefined) { - all.reject(failed.reason); - } else { - all.resolve(); - } - } - }, - (e: unknown) => { - failed ??= { reason: e }; - if (--remaining === 0) all.reject(failed.reason); - } - ); - } - return all.promise; - }, + : () => combineShutdownActions(actions), { reason: abortReason } ); }; @@ -2757,6 +2885,22 @@ function pipeToInternal( chunks[i] as R ); markPromiseHandled(writePromise); + // A rejection that leaves the destination WRITABLE is the + // internal transforms' non-fatal invalid-chunk rejection — fail + // the pipe with it (see onWriteRejectedNonFatally). Fatal + // rejections error the destination and are handled by the + // closed-promise observer instead. When a clean shutdown is + // already waiting for acknowledgment, record the failure for + // runAction's upgrade path. + PromisePrototypeThen(writePromise, undefined, (e: unknown) => { + if (writableInternals.getState(destination) !== 'writable') return; + if (shuttingDown) { + pendingNonFatalWriteFailure ??= { reason: e }; + return; + } + onWriteRejectedNonFatally(e); + poke(); + }); // Track the last write so shutdownWithAction can wait for it. // The writable serializes writes, so when this settles all // preceding writes have already settled. @@ -4260,12 +4404,22 @@ module.exports = { // possible future public exposure. ReadableStreamDrainingReader, // Internal operations consumed by the TransformStream cancel/flush - // coordination (finishPromise guard). Unreachable from user code. + // coordination (finishPromise guard) and the workerd expectedLength + // extension. Unreachable from user code. internalsForTransform: ObjectFreeze({ getState: (stream: ReadableStream) => getReadableStreamGetState(stream), getStoredError: (stream: ReadableStream) => getReadableStreamStoredError(stream), + normalizeExpectedLength, + setControllerExpectedLength: ( + controller: object, + length: bigint | undefined + ) => + setDefaultControllerExpectedLength( + controller as ReadableStreamDefaultController, + length + ), }), // Part of the internal implementation. Do not re-export to user code diff --git a/src/per_isolate/webstreams/transform.ts b/src/per_isolate/webstreams/transform.ts index 36684e42cfd..98dcdad6c10 100644 --- a/src/per_isolate/webstreams/transform.ts +++ b/src/per_isolate/webstreams/transform.ts @@ -312,6 +312,13 @@ class TransformStream { if (cancelFn !== undefined && typeof cancelFn !== 'function') { throw new TypeError('transformer.cancel must be a function'); } + // Non-standard workerd extension: the TOTAL bytes the readable side + // will produce (undefined = unknown). Advertised through the + // readable's controller so the C++ bridge derives a Content-Length + // for bodies built from this transform; not enforced here. + const expectedLength = readableInternals.normalizeExpectedLength( + (transformer as { expectedLength?: unknown }).expectedLength + ); const flushFn = transformer.flush; if (flushFn !== undefined && typeof flushFn !== 'function') { throw new TypeError('transformer.flush must be a function'); @@ -411,6 +418,12 @@ class TransformStream { }, { highWaterMark: readableHWM, size: readableStrategy.size } ); + if (expectedLength !== undefined) { + readableInternals.setControllerExpectedLength( + this.#readableController as object, + expectedLength + ); + } } else { // ---- STANDARD PATH (transformer has algorithms) ---- @@ -675,6 +688,12 @@ class TransformStream { }, { highWaterMark: readableHWM, size: readableStrategy.size } ); + if (expectedLength !== undefined) { + readableInternals.setControllerExpectedLength( + this.#readableController as object, + expectedLength + ); + } // --- Start the transformer --- const startResult: unknown = diff --git a/src/per_isolate/webstreams/writable.ts b/src/per_isolate/webstreams/writable.ts index 3733a7cb1af..33c3c517652 100644 --- a/src/per_isolate/webstreams/writable.ts +++ b/src/per_isolate/webstreams/writable.ts @@ -73,6 +73,35 @@ function isActualObject(value: unknown): value is object { return value != null && typeof value === 'object'; } +// A sink write() rejection wrapped in this marker rejects the WRITE +// REQUEST but leaves the stream writable — the workerd-internal contract +// for the identity and compression transforms (matching the C++ internal +// controllers), where an invalid chunk is a per-write error, not a stream +// error. Under pure WHATWG semantics every sink rejection errors the +// stream, so this channel is unreachable for user-provided sinks: the +// wrapper is minted only via internalsForPipe.nonFatalWriteRejection, +// which user code cannot reach. Detection is by private brand. +let isNonFatalWriteRejection: ( + value: unknown +) => value is NonFatalWriteRejection; + +class NonFatalWriteRejection { + #error: unknown; + + static { + isNonFatalWriteRejection = (value): value is NonFatalWriteRejection => + isActualObject(value) && #error in value; + } + + constructor(error: unknown) { + this.#error = error; + } + + get error(): unknown { + return this.#error; + } +} + function assertPrivateSymbol(symbol: symbol): void { if (symbol !== kPrivateSymbol) { throw new TypeError('Illegal constructor'); @@ -123,6 +152,10 @@ let writableStreamFinishErroringIfNeeded: ( let writableStreamMarkFirstWriteRequestInFlight: ( stream: WritableStream ) => void; +let writableStreamFinishInFlightWriteWithNonFatalError: ( + stream: WritableStream, + error: unknown +) => void; let writableStreamFinishInFlightWrite: (stream: WritableStream) => void; let writableStreamFinishInFlightWriteWithError: ( stream: WritableStream, @@ -585,6 +618,17 @@ class WritableStream { writableStreamDealWithRejection(stream, error); }; + // The non-fatal variant (see NonFatalWriteRejection): rejects the + // in-flight write's request WITHOUT the deal-with-rejection state + // transition — the stream stays writable and the queue keeps going. + writableStreamFinishInFlightWriteWithNonFatalError = (stream, error) => { + // assert: in-flight write request is set (caller guarantee) + const request = stream + .#inFlightWriteRequest as PromiseWithResolversType; + stream.#inFlightWriteRequest = undefined; + request.reject(error); + }; + writableStreamMarkCloseRequestInFlight = (stream) => { // assert: no in-flight close; closeRequest set stream.#inFlightCloseRequest = stream.#closeRequest; @@ -1108,6 +1152,28 @@ class WritableStreamDefaultController< this.#advanceQueueIfNeeded(); }, (e: unknown) => { + // Workerd-internal non-fatal rejection (identity/compression + // invalid chunks): reject THIS write's request, keep the stream + // writable, and continue with the queue — the same bookkeeping + // as the fulfillment path, with the request rejected instead. + if (isNonFatalWriteRejection(e)) { + writableStreamFinishInFlightWriteWithNonFatalError(stream, e.error); + const state = getWritableStreamState(stream); + const entry = ArrayPrototypeShift(this.#queue) as QueuedWrite; + this.#queueTotalSize -= entry.size; + if (this.#queueTotalSize < 0) this.#queueTotalSize = 0; + if ( + !writableStreamCloseQueuedOrInFlight(stream) && + state === 'writable' + ) { + writableStreamUpdateBackpressure( + stream, + controllerGetDesiredSize(this) <= 0 + ); + } + this.#advanceQueueIfNeeded(); + return; + } if (getWritableStreamState(stream) === 'writable') { this.#clearAlgorithms(); } @@ -1730,6 +1796,12 @@ module.exports = { willAcceptWrite: (stream: WritableStream): boolean => getWritableStreamState(stream) === 'writable' && !writableStreamCloseQueuedOrInFlight(stream), + // Wraps a sink write() rejection so it rejects only ITS write request, + // leaving the stream writable (the identity/compression invalid-chunk + // contract — see NonFatalWriteRejection). The sink throws (or rejects + // with) the wrapper; the write request rejects with `error`. + nonFatalWriteRejection: (error: unknown): NonFatalWriteRejection => + new NonFatalWriteRejection(error), getWriterReadyPromise: (writer: WritableStreamDefaultWriter) => getWriterReadyPromiseInternal(writer), getWriterClosedPromise: (writer: WritableStreamDefaultWriter) => diff --git a/src/tests/streams/compression/AGENTS.md b/src/tests/streams/compression/AGENTS.md index 78f4cbf2f1c..1757d50f132 100644 --- a/src/tests/streams/compression/AGENTS.md +++ b/src/tests/streams/compression/AGENTS.md @@ -61,11 +61,12 @@ interface DecompressionStream { already-detached chunk is a zero-byte no-op; shadowing metadata getters are never consulted. - **Chunks:** ArrayBuffer, any view (offsets honored), empty and detached - inputs are accepted by both. Strings (#1) and SharedArrayBuffers incl. - SAB-backed views (#2) are accepted and encoded/copied by C++ but rejected - by TypeScript per spec. Everything else rejects with TypeError (#3 - message) — after which the C++ stream SURVIVES (later writes flow, clean - close) while TypeScript errors both sides (#4). + inputs are accepted by both, and so are SharedArrayBuffers incl. + SAB-backed views — copied out on both sides (#2, DECIDED 2026-08-28). + Strings are UTF-8 encoded by C++ but rejected by TypeScript (#1). + Everything else rejects with TypeError (#3 message) — a PER-WRITE + rejection on both sides: the stream survives, later writes flow, and + close is clean (#4). - **Corrupt input** (DecompressionStream): the WRITE rejects (TypeError "Decompression failed.") and both sides error, in both implementations; the failure propagates through downstream pipes to consumers. @@ -131,10 +132,10 @@ pedantic branches shifting anything the suite pins. | # | Area | C++ | TypeScript | Pinned in | | --- | --- | --- | --- | --- | -| 1 | String chunks | accepted, UTF-8 encoded (the WPT compression-bad-chunks expected failure) | rejected TypeError | `stringChunkDiverges` | -| 2 | SharedArrayBuffer / SAB-backed view chunks | accepted (copied out) | rejected TypeError | `sharedArrayBufferChunkDiverges` | -| 3 | Invalid-chunk TypeError message | "This TransformStream is being used as a byte stream, but received an object of non-ArrayBuffer/ArrayBufferView type on its writable side." | "The provided value is not of type (ArrayBuffer or ArrayBufferView)" | `invalidChunkAftermathDiverges` | -| 4 | Invalid-chunk aftermath | stream survives | both sides error | `invalidChunkAftermathDiverges` | +| 1 | String chunks | accepted, UTF-8 encoded | rejected TypeError (per-write; the stream survives) | `stringChunkDiverges` | +| 2 | SharedArrayBuffer / SAB-backed view chunks | accepted (copied out) | same (DECIDED 2026-08-28; the reason the WPT bad-chunks files are disabled for both impls) | `sharedArrayBufferChunkAccepted` | +| 3 | Invalid-chunk TypeError message | "This TransformStream is being used as a byte stream, but received an object of non-ArrayBuffer/ArrayBufferView type on its writable side." | "The provided value is not of type (ArrayBuffer or ArrayBufferView)" | `invalidChunkRejectsWriteOnly` | +| 4 | Invalid-chunk aftermath | stream survives | same — non-fatal write rejection (the DECIDED per-write contract) | `invalidChunkRejectsWriteOnly` | | 5 | `TransformStream` inheritance + accessor placement | subclass; readable/writable inherited | standalone; own accessors | `transformStreamInheritance` | | 6 | `constructor.length` | 0 | 1 | `constructorSurface` | | 7 | Missing/undefined format | jsg type-boundary TypeError ("not of type 'string'") | ToString-coerced into format validation | `nonStringFormatThrows` | @@ -159,7 +160,7 @@ pedantic branches shifting anything the suite pins. | `empty-stream.js` | close-with-no-writes emits a valid empty member; decompressing it yields EOF | | `corrupt-input.js` | write-time rejection with "Decompression failed."; both-sides error; iteration rejection; bad magic bytes | | `strict-checks.js` | trailing-data write rejection; close-with-no-data rejection; truncated-member close rejection | -| `chunk-types.js` | BufferSource acceptance incl. offsets; string (#1), SAB (#2), invalid-chunk message+aftermath (#3, #4) | +| `chunk-types.js` | BufferSource acceptance incl. offsets and SAB-backed inputs by copy (#2); string (#1); invalid-chunk message+per-write aftermath (#3, #4) | | `buffer-lifecycle.js` | snapshot-at-write: post-write mutation/detach/shrink invisible; already-detached no-op; lying metadata getters never consulted | | `byob.js` | BYOB reader fills a 2-byte destination with the gzip magic | | `backpressure.js` | eager write settlement without reads; desiredSize accounting (#8) | diff --git a/src/tests/streams/compression/chunk-types.js b/src/tests/streams/compression/chunk-types.js index bb496cb2e3c..1118280ceea 100644 --- a/src/tests/streams/compression/chunk-types.js +++ b/src/tests/streams/compression/chunk-types.js @@ -51,12 +51,23 @@ export const stringChunkDiverges = { const cs = new CompressionStream('gzip'); const writer = cs.writable.getWriter(); if (usingTsImpl) { + // Rejected — but per-write only (the DECIDED invalid-chunk + // contract): the stream survives and later traffic flows. await rejects(writer.write('hi'), (err) => { strictEqual(err.constructor, TypeError); strictEqual(err.message, tsBadChunkMsg); return true; }); - await rejects(writer.closed, TypeError); + await writer.write(enc.encode('hi')); + await writer.close(); + const chunks = []; + for await (const chunk of cs.readable) { + chunks.push(chunk); + } + strictEqual( + dec.decode(await pump(new DecompressionStream('gzip'), chunks)), + 'hi' + ); } else { // Accepted and UTF-8 encoded, like the identity streams. await writer.write('hi'); @@ -73,36 +84,36 @@ export const stringChunkDiverges = { }, }; -export const sharedArrayBufferChunkDiverges = { +// SAB-backed chunks are accepted by copying the shared bytes (parity — +// DECIDED 2026-08-28, matching the identity streams and C++; the strict +// [AllowShared]-less BufferSource reading was considered and overridden, +// which is why the WPT bad-chunks files are disabled for both impls). +export const sharedArrayBufferChunkAccepted = { async test() { const sab = new SharedArrayBuffer(1); new Uint8Array(sab)[0] = 0x41; for (const chunk of [sab, new Uint8Array(sab)]) { const cs = new CompressionStream('gzip'); const writer = cs.writable.getWriter(); - if (usingTsImpl) { - await rejects(writer.write(chunk), (err) => { - strictEqual(err.constructor, TypeError); - strictEqual(err.message, tsBadChunkMsg); - return true; - }); - } else { - await writer.write(chunk); - await writer.close(); - const chunks = []; - for await (const c of cs.readable) { - chunks.push(c); - } - strictEqual( - dec.decode(await pump(new DecompressionStream('gzip'), chunks)), - 'A' - ); + await writer.write(chunk); + await writer.close(); + const chunks = []; + for await (const c of cs.readable) { + chunks.push(c); } + strictEqual( + dec.decode(await pump(new DecompressionStream('gzip'), chunks)), + 'A' + ); + // The shared bytes were copied, never consumed in place. + strictEqual(new Uint8Array(sab)[0], 0x41); } }, }; -export const invalidChunkAftermathDiverges = { +// An invalid chunk rejects ITS OWN write only (message per impl); the +// stream survives on both sides (parity — the DECIDED contract). +export const invalidChunkRejectsWriteOnly = { async test() { const cs = new CompressionStream('gzip'); const writer = cs.writable.getWriter(); @@ -112,23 +123,16 @@ export const invalidChunkAftermathDiverges = { strictEqual(err.message, expectedMsg); return true; }); - if (usingTsImpl) { - // Both sides errored: later writes and the closed promise reject. - await rejects(writer.write(enc.encode('x')), TypeError); - await rejects(writer.closed, TypeError); - await rejects(cs.readable.getReader().read(), TypeError); - } else { - // The stream survives: later traffic flows and close is clean. - await writer.write(enc.encode('ok')); - await writer.close(); - const chunks = []; - for await (const chunk of cs.readable) { - chunks.push(chunk); - } - strictEqual( - dec.decode(await pump(new DecompressionStream('gzip'), chunks)), - 'ok' - ); + // The stream survives: later traffic flows and close is clean. + await writer.write(enc.encode('ok')); + await writer.close(); + const chunks = []; + for await (const chunk of cs.readable) { + chunks.push(chunk); } + strictEqual( + dec.decode(await pump(new DecompressionStream('gzip'), chunks)), + 'ok' + ); }, }; diff --git a/src/tests/streams/compression/main.js b/src/tests/streams/compression/main.js index b0f50c9ff7c..62a41621579 100644 --- a/src/tests/streams/compression/main.js +++ b/src/tests/streams/compression/main.js @@ -95,8 +95,8 @@ export { export { bufferSourceChunksAccepted, stringChunkDiverges, - sharedArrayBufferChunkDiverges, - invalidChunkAftermathDiverges, + sharedArrayBufferChunkAccepted, + invalidChunkRejectsWriteOnly, } from 'chunk-types'; export { diff --git a/src/tests/streams/identity/AGENTS.md b/src/tests/streams/identity/AGENTS.md index bf65a889a12..979ac2dcc3b 100644 --- a/src/tests/streams/identity/AGENTS.md +++ b/src/tests/streams/identity/AGENTS.md @@ -214,13 +214,13 @@ pattern; a change to either side fails its cell. | 3 | FLS length range | rejects > 2^53−1 (`TypeError`) | accepts full uint64 | `fixedLengthLengthsAboveMaxSafeInteger` | | 4 | `TransformStream` inheritance | `its instanceof TransformStream` is true | false (deliberate) | `identityBrandChecks` | | 5 | Accessor placement | inherited from `TransformStream.prototype` | own on `IdentityTransformStream.prototype` | `propertyPlacement` | -| 6 | Invalid chunk aftermath | stream unaffected, remains usable | stream errors; `closed` rejects, later writes reject | `rejectsNumberChunk` | +| 6 | Invalid chunk aftermath | stream unaffected, remains usable | same — the sink signals the rejection through the non-fatal write-rejection channel; the write rejects, the stream survives (the DECIDED per-write contract) | `rejectsNumberChunk` | | 7 | String `desiredSize` accounting | exact UTF-8 byte count | `length × 3` upper-bound estimate | `stringWriteDesiredSizeAccounting` | | 8 | Abort/cancel reason identity | re-created `Error`, same message (crosses kj); exception: `writer.closed` under abort gets the original instance | original instance everywhere | `abort-propagation.js`, `cancel-propagation.js` | | 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`, errors the stream | `alreadyDetachedBufferAtWrite` | +| 12 | Already-detached `ArrayBuffer` chunk | zero-length no-op | rejects `TypeError` (per-write; the stream survives) | `alreadyDetachedBufferAtWrite` | | 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" | `writeAfterBothBranchesCancel` | | 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` | @@ -235,7 +235,7 @@ 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 lengths throw (types per ledger #1–3); inheritance (ledger #4); a user-supplied strategy `size` is never invoked (ITS and FLS, with and without explicit HWM) | -| `chunk-types.js` | accepted: `Uint8Array`, `ArrayBuffer`, `DataView` subrange, string→UTF-8, subarray offsets; rejected: numbers, plain objects (`TypeError`; aftermath per ledger #6); an invalid chunk queued behind valid writes surfaces its error in FIFO order — the earlier writes still deliver in both implementations | +| `chunk-types.js` | accepted: `Uint8Array`, `ArrayBuffer`, `DataView` subrange, string→UTF-8, 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 | | `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, out-of-bounds views); shadowing/throwing metadata getters never consulted | diff --git a/src/tests/streams/identity/buffer-lifecycle.js b/src/tests/streams/identity/buffer-lifecycle.js index f0e27cb88ab..cee7a268e86 100644 --- a/src/tests/streams/identity/buffer-lifecycle.js +++ b/src/tests/streams/identity/buffer-lifecycle.js @@ -90,7 +90,8 @@ export const alreadyDetachedBufferAtWrite = { // 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), erroring the stream like any other invalid chunk. + // 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(); @@ -98,6 +99,13 @@ export const alreadyDetachedBufferAtWrite = { 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 { await writer.write(ab); const closePromise = writer.close(); diff --git a/src/tests/streams/identity/chunk-types.js b/src/tests/streams/identity/chunk-types.js index b69300c6092..12c23c9943c 100644 --- a/src/tests/streams/identity/chunk-types.js +++ b/src/tests/streams/identity/chunk-types.js @@ -16,7 +16,6 @@ // promise rejects and subsequent writes reject. import { strictEqual, deepStrictEqual, rejects } from 'node:assert'; -import { usingTsImpl } from 'which-impl'; export const acceptsUint8Array = { async test() { @@ -94,25 +93,23 @@ export const respectsViewOffsets = { }, }; +// An invalid chunk rejects ITS OWN write only; the stream stays usable +// (parity — the DECIDED contract for the internal transforms, matching +// the C++ internal controllers; under TypeScript the sink signals the +// rejection through the non-fatal write-rejection channel). export const rejectsNumberChunk = { async test() { const { readable, writable } = new IdentityTransformStream(); const writer = writable.getWriter(); await rejects(writer.write(42), TypeError); - if (usingTsImpl) { - // The invalid chunk errored the stream. - await rejects(writer.closed, TypeError); - await rejects(writer.write(new Uint8Array([1])), TypeError); - } else { - // The stream is unaffected: a subsequent valid write still flows. - const reader = readable.getReader(); - const writePromise = writer.write(new Uint8Array([1])); - const { value, done } = await reader.read(); - strictEqual(done, false); - strictEqual(value[0], 1); - await writePromise; - await writer.close(); - } + // The stream is unaffected: a subsequent valid write still flows. + const reader = readable.getReader(); + const writePromise = writer.write(new Uint8Array([1])); + const { value, done } = await reader.read(); + strictEqual(done, false); + strictEqual(value[0], 1); + await writePromise; + await writer.close(); }, }; @@ -120,9 +117,8 @@ export const invalidChunkAfterQueuedValidWrites = { async test() { // An invalid chunk queued BEHIND valid writes must not cost them their // delivery: validation errors surface in FIFO order, after everything - // written earlier has been consumed. Only the aftermath diverges - // (ledger #6): TypeScript errors the stream once the bad chunk's turn - // arrives; C++ leaves the stream usable. + // written earlier has been consumed, and the stream survives the + // rejection (parity — the DECIDED per-write-rejection contract). const { readable, writable } = new IdentityTransformStream(); const writer = writable.getWriter(); const reader = readable.getReader(); @@ -136,15 +132,11 @@ export const invalidChunkAfterQueuedValidWrites = { const [r2] = await Promise.all([reader.read(), w2]); strictEqual(dec.decode(r2.value), 'b'); await rejects(wBad, TypeError); - if (usingTsImpl) { - await rejects(writer.closed, TypeError); - } else { - // The stream survives; later traffic still flows. - const w4 = writer.write(enc.encode('c')); - const [r4] = await Promise.all([reader.read(), w4]); - strictEqual(dec.decode(r4.value), 'c'); - await writer.close(); - } + // The stream survives; later traffic still flows. + const w4 = writer.write(enc.encode('c')); + const [r4] = await Promise.all([reader.read(), w4]); + strictEqual(dec.decode(r4.value), 'c'); + await writer.close(); }, }; diff --git a/src/tests/streams/piping/AGENTS.md b/src/tests/streams/piping/AGENTS.md index 3875616806d..cf7aa59293f 100644 --- a/src/tests/streams/piping/AGENTS.md +++ b/src/tests/streams/piping/AGENTS.md @@ -18,8 +18,8 @@ a deliberate defect pin, not a hole). | # | Area | C++ | TypeScript | Pinned in | | --- | --- | --- | --- | --- | -| 1 | non-byte chunks piped into native identity | pipe rejects 'This WritableStream only supports writing byte types.' | STRINGS are UTF-8 ENCODED and pass through; a NUMBER chunk stalls the pipe silently — the next read pends forever (bounded defect pin) | `pipeThroughJsToInternal` | -| 2 | writable lock after a completed pipeThrough | stays locked; getWriter throws | getWriter() SUCCEEDS; `.locked` is transient/racy (observed both values) — only getWriter pinned | `pipeThroughJsToInternalCloses` | +| 1 | non-byte chunks piped into native identity | strings are UTF-8 encoded and pass through on both; the NUMBER chunk fails the pipe — C++ rejects 'This WritableStream only supports writing byte types.'; TypeScript surfaces the identity validation TypeError through the non-fatal write-rejection pipe path (dest aborted, downstream reads reject) | `pipeThroughJsToInternal` | +| 2 | writable lock after a completed pipeThrough | stays locked; getWriter throws | spec finalize: both locks release when the pipe settles — one macrotask after the output's done, `.locked` is deterministically false and getWriter() succeeds (the release cascade is not synchronized with the output's done delivery, so loop-exit-instant reads are unspecified; `.locked` and getWriter share one predicate and never disagree at an instant) | `pipeThroughJsToInternalCloses` | | 3 | write-after-close rejection message | 'This WritableStream has been closed.' | 'Cannot write to a stream that is closing or closed' (`CLOSED_WRITE_MSG` helper) | `pipeToInternalToJsSimple`, `pipeToInternalToJsClose` | | 4 | ws.close() queued BEFORE pipeTo | pipe locks both ends, waits, cancels source with 'This destination writable stream is closed.', then RESOLVES (see the TODO(conform) in the test) | pipe REJECTS IMMEDIATELY 'Destination closed before the pipe completed', cancels source with the same error (preventCancel suppresses), locks never observed held | `pipeToJsToJsCloseQueuedDestination`(+`PreventCancel`) | | 5 | pipeTo brand check on a broken `this` | THROWS synchronously (before the capture_async_api_throws wrapper; the WPT general.any seed); a real stream with a bad destination REJECTS | both reject (spec) | `brandChecks` | @@ -27,8 +27,8 @@ a deliberate defect pin, not a hole). | 7 | source queue after a preventCancel'd failing pipe | the not-yet-written chunk remains readable | read-ahead already consumed it; a fresh read PENDS (bounded) | `destWriteThrowsMidPipePreventCancel` | | 8 | dest controller error()s while the pipe waits on a read | HALF-PROPAGATES: cancels the source with the error but FULFILLS the pipe promise | rejects the pipe and cancels the source with the error (spec) | `destControllerErrorsMidPipe` | | 9 | FixedLengthStream length violations via pipe | overflow: pipe NEVER SETTLES (bounded); underflow: never settles | overflow: rejects RangeError; underflow: never settles (parity of nonconformance) | `fixedLengthStreamPipeOverflow`/`Underflow` | -| 10 | already-closed source → already-closed dest | rejects TypeError (spec; the WPT multiple-propagation seed) | FULFILLS as a trivially complete pipe | `closedSourceToClosedDest` | -| 11 | SharedArrayBuffer-backed views into CompressionStream | copies the shared bytes; round-trips | write path REJECTS TypeError 'The provided value is not of type (ArrayBuffer or ArrayBufferView)' — while its identity stream ACCEPTS the same views | `sabViewThroughCompressionRoundTrip` | +| 10 | already-closed source → already-closed dest | rejects TypeError (a C++ deviation: the spec's ordered shutdown conditions give closing-forward priority) | FULFILLS (spec; WPT multiple-propagation 'closed readable to closed writable' pins the fulfillment) | `closedSourceToClosedDest` | +| 11 | SharedArrayBuffer-backed views into CompressionStream | copies the shared bytes; round-trips | same (DECIDED 2026-08-28) | `sabViewThroughCompressionRoundTrip` | Parity worth noting (probed, pinned): the whole error-propagation- forward core matrix (starts-errored rejection/hook IDENTITY on both diff --git a/src/tests/streams/piping/interop.js b/src/tests/streams/piping/interop.js index d57e00d8b8b..2a3a4d928ba 100644 --- a/src/tests/streams/piping/interop.js +++ b/src/tests/streams/piping/interop.js @@ -139,10 +139,14 @@ export const fixedLengthStreamPipeUnderflow = { }; // An already-closed source piped into an already-closed destination -// (the WPT multiple-propagation seed). DIVERGENCE: C++ rejects -// TypeError (per spec, the destination's closed state must reject the -// pipe); TypeScript treats it as a trivially complete pipe and -// FULFILLS. +// (the WPT multiple-propagation seed). DIVERGENCE — C++ deviates from +// the spec: the spec's shutdown conditions apply IN ORDER, so +// closing-forward (source closed → close dest, trivially resolved on an +// already-closed dest) wins over closing-backward (dest closed → +// TypeError), and the pipe FULFILLS — WPT multiple-propagation 'Piping +// from a closed readable stream to a closed writable stream' pins the +// fulfillment, and TypeScript conforms. C++ applies the dest-closed +// TypeError instead. export const closedSourceToClosedDest = { async test() { const rs = new ReadableStream({ diff --git a/src/tests/streams/piping/pipe-matrix.js b/src/tests/streams/piping/pipe-matrix.js index abee17fbf00..9ff8f317f3c 100644 --- a/src/tests/streams/piping/pipe-matrix.js +++ b/src/tests/streams/piping/pipe-matrix.js @@ -35,21 +35,21 @@ export const pipeThroughJsToInternal = { const output = []; if (usingTsImpl) { - // DIVERGENCE: TypeScript ENCODES string chunks through the native - // identity writable (UTF-8), and a NUMBER chunk stalls the pipe - // silently — the next read pends forever (bounded observation; - // defect shape). C++ rejects the pipe at the first non-byte chunk. + // Both implementations UTF-8-encode the string chunk through the + // identity writable and fail the pipe at the NUMBER chunk (whose + // write rejects non-fatally; the pipe aborts the destination with + // the validation error, so downstream reads reject). Message per + // implementation. const reader = readable.getReader(); for (let i = 0; i < 3; i++) { output.push(dec.decode((await reader.read()).value)); } deepStrictEqual(output, ['hello', 'there', 'hello']); - const outcome = await Promise.race([ - reader.read().then(() => 'settled'), - scheduler.wait(100).then(() => 'pending'), - ]); - strictEqual(outcome, 'pending'); - await reader.cancel('cleanup'); + await rejects(reader.read(), { + name: 'TypeError', + message: + 'IdentityTransformStream: chunk must be a BufferSource or string', + }); } else { async function consumeStream() { for await (const chunk of readable) { @@ -283,11 +283,17 @@ export const pipeThroughJsToInternalCloses = { } // DIVERGENCE: after the pipe completes, C++ keeps the writable - // locked (getWriter throws). Under TypeScript getWriter() SUCCEEDS; - // the .locked getter's value at this point is transient (observed - // both true and false across runs), so only the deterministic - // getWriter behavior is pinned. + // locked forever (getWriter throws). TypeScript follows the spec's + // pipe finalize: both locks release when the pipe settles. The + // release cascades through microtasks that are not synchronized + // with the OUTPUT's done delivery (pipeThrough discards the pipe + // promise), so .locked must not be read at the loop-exit instant; + // one macrotask later the state is deterministically settled — + // unlocked, getWriter succeeds. (.locked and getWriter() share one + // predicate and can never disagree at a single instant.) if (usingTsImpl) { + await scheduler.wait(0); + strictEqual(transform.writable.locked, false); transform.writable.getWriter(); } else { strictEqual(transform.writable.locked, true); diff --git a/src/tests/streams/piping/special-buffers.js b/src/tests/streams/piping/special-buffers.js index c4088f9813f..ee9beb925c5 100644 --- a/src/tests/streams/piping/special-buffers.js +++ b/src/tests/streams/piping/special-buffers.js @@ -8,8 +8,7 @@ // pipe-write-special-buffer-test.js, strengthened from length checks // to content verification). -import { strictEqual, ok, rejects } from 'node:assert'; -import { usingTsImpl } from 'which-impl'; +import { strictEqual, ok } from 'node:assert'; function filledSabView(length, byte) { const sab = new SharedArrayBuffer(length); @@ -56,33 +55,21 @@ function assertAllBytes(bytes, expectedLength, expectedByte) { } } -// A SAB-backed view piped through CompressionStream. DIVERGENCE: C++ -// copies the shared bytes and round-trips them; the TypeScript -// CompressionStream write path REJECTS SharedArrayBuffer-backed views -// ('The provided value is not of type (ArrayBuffer or -// ArrayBufferView)') even though its identity stream accepts them (see -// the next test). The shared buffer is untouched either way. +// A SAB-backed view piped through CompressionStream: the shared bytes +// are copied and round-trip on both implementations (the DECIDED +// SAB-acceptance contract, matching the identity streams — see the +// compression suite's sharedArrayBufferChunkAccepted). The shared +// buffer is untouched either way. export const sabViewThroughCompressionRoundTrip = { async test() { const view = filledSabView(100, 0x41); const compressed = streamOf(view).pipeThrough( new CompressionStream('gzip') ); - if (usingTsImpl) { - await rejects( - drainToBytes(compressed.pipeThrough(new DecompressionStream('gzip'))), - { - name: 'TypeError', - message: - 'The provided value is not of type (ArrayBuffer or ArrayBufferView)', - } - ); - } else { - const restored = await drainToBytes( - compressed.pipeThrough(new DecompressionStream('gzip')) - ); - assertAllBytes(restored, 100, 0x41); - } + const restored = await drainToBytes( + compressed.pipeThrough(new DecompressionStream('gzip')) + ); + assertAllBytes(restored, 100, 0x41); // The shared buffer itself must be untouched (it cannot be // detached). assertAllBytes(view, 100, 0x41); diff --git a/src/tests/streams/r2-patterns/AGENTS.md b/src/tests/streams/r2-patterns/AGENTS.md index db6c5c65932..379994495be 100644 --- a/src/tests/streams/r2-patterns/AGENTS.md +++ b/src/tests/streams/r2-patterns/AGENTS.md @@ -8,17 +8,17 @@ normative artifact.** ## Divergence ledger (C++ vs TypeScript) -The TS side of every entry is a bounded defect pin (the readable-byte -suite's ledger #5/#6 and #12 families surfacing in R2's exact usage) — -this suite is effectively the R2 to-do list for the TypeScript streams -migration. +The readAtLeast tail rows all follow the DECIDED contract (matching +C++): a close below the minimum folds the available bytes into a +done=false result and a follow-up read resolves done. The one remaining +divergence is the tee byobRequest model (row 2, accepted). | # | Pattern | C++ | TypeScript | Pinned in | | --- | --- | --- | --- | --- | -| 1 | readAtLeast(min) when close arrives below min | folds the available bytes into a done=false result, then a done read | PENDS FOREVER (bounded) | `byobReadAtLeastAutomatic`, `closedByobTeeOnStart` (both tee branches) | -| 2 | tee over a source that touches `c.byobRequest` unconditionally | tee pulls carry a byobRequest; branch minimums aggregate across responds | tee-driven pulls present a NULL byobRequest: the source throws TypeError into the stream and every branch read rejects with it | `byobReadAtLeastTee`, `byobReadAtLeastTeeComplex1/2/3` | -| 3 | identity-stream readAtLeast tail | trailing below-min remainder delivered done=false, then 0-length done | view-exact fills, then the below-min tail PENDS FOREVER at close | `identityTransformStreamReadAtLeast` | -| 4 | chained identity→byte-stream readAtLeast consumption (the R2 body pump shape) | full 5000-byte body in 102-byte minimums | 49 full reads (4998 bytes), then the 2-byte tail PENDS | `partiallyFilledByobAtLeast` | +| 1 | readAtLeast(min) when close arrives below min | folds the available bytes into a done=false result, then a done read | same (deferred end-of-data commit; readable-byte ledger #12/#19) | `byobReadAtLeastAutomatic`, `closedByobTeeOnStart` (both tee branches) | +| 2 | tee pulls and `c.byobRequest` | tee pulls carry a byobRequest (atLeast/freshness asserted when present) | tee-driven pulls present a NULL byobRequest — ACCEPTED DIVERGENCE (decided 2026-08-28): sources must be null-tolerant (check `c.byobRequest`, enqueue() as the fallback); with that supported pattern the tee shapes are PARITY, tail included | `byobReadAtLeastTee`, `byobReadAtLeastTeeComplex1/2/3` (all four demonstrate the supported dual-path source) | +| 3 | identity-stream readAtLeast tail | trailing below-min remainder delivered done=false, then 0-length done — the DECIDED contract | same | `identityTransformStreamReadAtLeast`, tee complex variants (r3 done flag) | +| 4 | chained identity→byte-stream readAtLeast consumption (the R2 body pump shape) | full 5000-byte body in 102-byte minimums | same | `partiallyFilledByobAtLeast` | Parity worth noting: manual `byobRequest.atLeast` handling on a DIRECT byob reader (`byobReadAtLeastManual` — byobRequest is synthesized for diff --git a/src/tests/streams/r2-patterns/r2-consumption.js b/src/tests/streams/r2-patterns/r2-consumption.js index 516e06ab0db..d243985d936 100644 --- a/src/tests/streams/r2-patterns/r2-consumption.js +++ b/src/tests/streams/r2-patterns/r2-consumption.js @@ -9,10 +9,9 @@ // Request body. import { strictEqual, ok } from 'node:assert'; -import { usingTsImpl } from 'which-impl'; -// Bounded observation of a promise's outcome; a pinned 'pending' is a -// deliberate defect pin. +// Bounded observation of a promise's outcome, so a regression back to a +// hang fails the assertion instead of wedging the test. const outcomeOf = (p, ms = 250) => Promise.race([ p.then( @@ -22,35 +21,16 @@ const outcomeOf = (p, ms = 250) => scheduler.wait(ms).then(() => ({ state: 'pending' })), ]); -// The TS-side pins in this file follow the readable-byte suite's -// ledger: readAtLeast PENDS FOREVER when the stream closes below the -// minimum (ledger #12 family), and tee-driven pulls present a NULL -// byobRequest (ledger #5/#6), so sources that touch c.byobRequest -// unconditionally throw TypeError into the stream. +// Close-below-min follows the DECIDED readAtLeast tail contract on both +// implementations: the available bytes are folded into a done=false +// result, and a follow-up read resolves done. Tee-driven pulls under +// TypeScript present a NULL byobRequest (readable-byte ledger #5/#6), +// so sources that touch c.byobRequest unconditionally throw TypeError +// into the stream — the dual-path source below is the supported shape. // Test BYOB readAtLeast with automatic atLeast handling export const byobReadAtLeastAutomatic = { async test() { - if (usingTsImpl) { - // Close arrives below the 100-byte minimum: the readAtLeast - // PENDS FOREVER (C++ folds the 10 available bytes into a - // done=false result). - const enc = new TextEncoder(); - const chunks = ['hello', 'there']; - const rs = new ReadableStream({ - type: 'bytes', - pull(c) { - c.enqueue(enc.encode(chunks.shift())); - if (chunks.length === 0) c.close(); - }, - }); - const reader = rs.getReader({ mode: 'byob' }); - const outcome = await outcomeOf( - reader.readAtLeast(100, new Uint8Array(100)) - ); - strictEqual(outcome.state, 'pending'); - return; - } const enc = new TextEncoder(); const dec = new TextDecoder(); const chunks = ['hello', 'there']; @@ -66,9 +46,15 @@ export const byobReadAtLeastAutomatic = { const reader = rs.getReader({ mode: 'byob' }); + // Close arrives below the 100-byte minimum: the available 10 bytes + // fold into a done=false result (the DECIDED tail contract, parity). const res = await reader.readAtLeast(100, new Uint8Array(100)); + strictEqual(res.done, false); strictEqual(dec.decode(res.value), 'hellothere'); + + const tail = await reader.readAtLeast(4, new Uint8Array(4)); + strictEqual(tail.done, true); }, }; @@ -181,28 +167,6 @@ export const fixedLengthStreamReadAtLeast = { // still works correctly when one branch is consumed via waitUntil export const closedByobTeeOnStart = { async test(ctrl, env, ctx) { - if (usingTsImpl) { - // Both branches close below the 10-byte minimum: each branch's - // readAtLeast PENDS FOREVER. - const enc = new TextEncoder(); - const rs = new ReadableStream({ - type: 'bytes', - start(c) { - c.enqueue(enc.encode('hello')); - c.close(); - }, - }); - const [b1, b2] = rs.tee(); - const o1 = await outcomeOf( - b1.getReader({ mode: 'byob' }).readAtLeast(10, new Uint8Array(10)) - ); - const o2 = await outcomeOf( - b2.getReader({ mode: 'byob' }).readAtLeast(10, new Uint8Array(10)) - ); - strictEqual(o1.state, 'pending'); - strictEqual(o2.state, 'pending'); - return; - } const enc = new TextEncoder(); const dec = new TextDecoder(); @@ -243,34 +207,6 @@ export const closedByobTeeOnStart = { // Test IdentityTransformStream properly handles readAtLeast export const identityTransformStreamReadAtLeast = { async test() { - if (usingTsImpl) { - // The first 100-byte minimum is satisfiable, but the trailing - // 1-byte remainder plus close never fulfills a 100-byte minimum: - // the consumption chain PENDS FOREVER at the tail. - const { readable, writable } = new IdentityTransformStream(); - const reader = readable.getReader({ mode: 'byob' }); - const writer = writable.getWriter(); - void writer.write(new Uint8Array(100)); - void writer.write(new Uint8Array(1)); - void writer.write(new Uint8Array(100)); - void writer.close(); - const first = await outcomeOf( - reader.readAtLeast(100, new Uint8Array(100)) - ); - strictEqual(first.state, 'fulfilled'); - strictEqual(first.value.value.byteLength, 100); - const second = await outcomeOf( - reader.readAtLeast(100, new Uint8Array(100)) - ); - strictEqual(second.state, 'fulfilled'); - strictEqual(second.value.value.byteLength, 100); // view filled exactly - // One byte remains, below the minimum, with the close behind it. - const tail = await outcomeOf( - reader.readAtLeast(100, new Uint8Array(100)) - ); - strictEqual(tail.state, 'pending'); - return; - } const { readable, writable } = new IdentityTransformStream(); const reader = readable.getReader({ mode: 'byob' }); @@ -299,30 +235,6 @@ export const identityTransformStreamReadAtLeast = { // Test BYOB readAtLeast partially filled export const partiallyFilledByobAtLeast = { async test() { - if (usingTsImpl) { - // 5000 bytes consumed in 102-byte minimums leaves a final - // below-minimum remainder; the last readAtLeast PENDS FOREVER - // when the close arrives. - const { readable, writable } = new IdentityTransformStream(); - const enc = new TextEncoder(); - const writer = writable.getWriter(); - void writer.write(enc.encode('hello'.repeat(1000))); - void writer.close(); - const reader = readable.getReader({ mode: 'byob' }); - let received = 0; - let ab = new ArrayBuffer(102); - for (;;) { - const outcome = await outcomeOf( - reader.readAtLeast(102, new Uint8Array(ab)) - ); - if (outcome.state === 'pending') break; - strictEqual(outcome.state, 'fulfilled'); - received += outcome.value.value.byteLength; - ab = outcome.value.value.buffer; - } - strictEqual(received, 4998); // 49 full reads; the 2-byte tail hangs - return; - } const { readable, writable } = new IdentityTransformStream(); const reader = readable.getReader({ mode: 'byob' }); const rs = new ReadableStream({ @@ -367,68 +279,46 @@ export const partiallyFilledByobAtLeast = { }, }; -// A byte source in the shape the tee tests use: touches c.byobRequest -// unconditionally (fine under C++ tee pulls, TypeError under TS). -function teeSourceTouchingByobRequest() { - const enc = new TextEncoder(); - const chunks = ['hello', 'there']; +// The SUPPORTED tee source pattern (decided 2026-08-28): tee-driven +// pulls under the TypeScript shared-queue model present a NULL +// byobRequest, so sources must be null-tolerant — fill the request when +// present (the C++ path), enqueue() otherwise. Sources that touch +// c.byobRequest unconditionally are INCORRECT under tee. +function dualPathTeeSource(chunks, onByobRequest) { + const pending = [...chunks]; return new ReadableStream({ type: 'bytes', pull(c) { - if (chunks.length === 0) { + if (pending.length === 0) { c.close(); - c.byobRequest.respond(0); + c.byobRequest?.respond(0); + return; + } + const chunk = pending.shift(); + if (c.byobRequest) { + onByobRequest?.(c.byobRequest); + c.byobRequest.view.set(chunk); + c.byobRequest.respond(chunk.length); } else { - enc.encodeInto(chunks.shift(), c.byobRequest.view); - c.byobRequest.respond(5); + c.enqueue(chunk.slice()); } }, }); } -// Test BYOB readAtLeast with tee +// Test BYOB readAtLeast with tee, using the supported null-tolerant +// source pattern. C++ tee pulls carry a byobRequest (atLeast asserted); +// TypeScript tee pulls enqueue. export const byobReadAtLeastTee = { async test() { - if (usingTsImpl) { - // Tee-driven pulls present a NULL byobRequest: the source's - // unconditional c.byobRequest access throws into the stream and - // every branch read rejects with that TypeError. - const rs = teeSourceTouchingByobRequest(); - const [branch1, branchB] = rs.tee(); - const [branch2, branch3] = branchB.tee(); - for (const [branch, min, size] of [ - [branch1, 100, 100], - [branch2, 5, 100], - [branch3, 3, 3], - ]) { - const outcome = await outcomeOf( - branch - .getReader({ mode: 'byob' }) - .readAtLeast(min, new Uint8Array(size)) - ); - strictEqual(outcome.state, 'rejected'); - strictEqual(outcome.reason.name, 'TypeError'); - ok(/byobRequest|null/.test(outcome.reason.message)); - } - return; - } - const enc = new TextEncoder(); const dec = new TextDecoder(); - const chunks = ['hello', 'there']; const expectedAtLeasts = [100, 95]; - const rs = new ReadableStream({ - type: 'bytes', - pull(c) { - if (chunks.length === 0) { - c.close(); - c.byobRequest.respond(0); - } else { - strictEqual(c.byobRequest.atLeast, expectedAtLeasts.shift()); - enc.encodeInto(chunks.shift(), c.byobRequest.view); - c.byobRequest.respond(5); - } - }, - }); + const rs = dualPathTeeSource( + ['hello', 'there'].map((t) => new TextEncoder().encode(t)), + (req) => { + strictEqual(req.atLeast, expectedAtLeasts.shift()); + } + ); const [branch1, branchB] = rs.tee(); const [branch2, branch3] = branchB.tee(); @@ -437,72 +327,30 @@ export const byobReadAtLeastTee = { const reader2 = branch2.getReader({ mode: 'byob' }); const reader3 = branch3.getReader({ mode: 'byob' }); - const p1 = reader.readAtLeast(100, new Uint8Array(100)); - const p2 = reader2.readAtLeast(5, new Uint8Array(100)); - const p3 = reader3.readAtLeast(3, new Uint8Array(3)); - - const res = await Promise.all([p1, p2, p3]); - - strictEqual(dec.decode(res[0].value), 'hellothere'); - strictEqual(dec.decode(res[1].value), 'hello'); - strictEqual(dec.decode(res[2].value), 'hel'); - - const res2 = await reader2.readAtLeast(5, new Uint8Array(100)); - strictEqual(dec.decode(res2.value), 'there'); - - const res3 = await reader3.readAtLeast(4, new Uint8Array(4)); - strictEqual(dec.decode(res3.value), 'loth'); - - const res4 = await reader.readAtLeast(100, new Uint8Array(100)); - strictEqual(res4.done, true); - - const res5 = await reader2.readAtLeast(5, new Uint8Array(100)); - strictEqual(res5.done, true); - - const res6 = await reader3.readAtLeast(4, new Uint8Array(4)); - strictEqual(dec.decode(res6.value), 'ere'); - - const res7 = await reader2.readAtLeast(5, new Uint8Array(100)); - strictEqual(res7.done, true); + // PARITY with the null-tolerant source: all three minimums deliver + // the same bytes on both implementations (the min-100 read collects + // the full 10 available bytes at close — the below-min tail-shape + // done flag is pinned in the complex variants). + const p1 = outcomeOf(reader.readAtLeast(100, new Uint8Array(100))); + const p2 = outcomeOf(reader2.readAtLeast(5, new Uint8Array(100))); + const p3 = outcomeOf(reader3.readAtLeast(3, new Uint8Array(3))); + const [o1, o2, o3] = await Promise.all([p1, p2, p3]); + strictEqual(o1.state, 'fulfilled'); + strictEqual(dec.decode(o1.value.value), 'hellothere'); + strictEqual(o2.state, 'fulfilled'); + strictEqual(dec.decode(o2.value.value), 'hello'); + strictEqual(o3.state, 'fulfilled'); + strictEqual(dec.decode(o3.value.value), 'hel'); }, }; -// Test BYOB readAtLeast with tee complex variant 1 +// Complex variant 1: staggered reads across three branches with a +// null-tolerant source ('helloth' + 'ere'). export const byobReadAtLeastTeeComplex1 = { async test() { - if (usingTsImpl) { - // Tee-driven pulls present a NULL byobRequest: the source's - // unconditional c.byobRequest access throws into the stream and - // every branch read rejects with that TypeError. - const rs = teeSourceTouchingByobRequest(); - const [b1, b2] = rs.tee(); - const outcome = await outcomeOf( - b1.getReader({ mode: 'byob' }).readAtLeast(5, new Uint8Array(100)) - ); - strictEqual(outcome.state, 'rejected'); - strictEqual(outcome.reason.name, 'TypeError'); - void b2; - return; - } - const enc = new TextEncoder(); const dec = new TextDecoder(); - const chunks = ['helloth', 'ere']; - let previousByobRequest; - const rs = new ReadableStream({ - type: 'bytes', - pull(c) { - const req = c.byobRequest; - if (chunks.length === 0) { - c.close(); - req.respond(0); - } else { - ok(!(req === previousByobRequest)); - const chunk = chunks.shift(); - enc.encodeInto(chunk, req.view); - req.respond(chunk.length); - } - }, - }); + const enc = new TextEncoder(); + const rs = dualPathTeeSource([enc.encode('helloth'), enc.encode('ere')]); const [branch1, branchB] = rs.tee(); const [branch2, branch3] = branchB.tee(); @@ -511,54 +359,41 @@ export const byobReadAtLeastTeeComplex1 = { const reader2 = branch2.getReader({ mode: 'byob' }); const reader3 = branch3.getReader({ mode: 'byob' }); - const res1 = await reader1.readAtLeast(5, new Uint8Array(10)); - strictEqual(dec.decode(res1.value), 'helloth'); - const res2 = await reader2.readAtLeast(10, new Uint8Array(10)); - strictEqual(dec.decode(res2.value), 'hellothere'); - - const res3 = await reader1.readAtLeast(5, new Uint8Array(10)); - strictEqual(dec.decode(res3.value), 'ere'); - - const res4 = await reader3.readAtLeast(2, new Uint8Array(12)); - strictEqual(dec.decode(res4.value), 'hellothere'); + const r1 = await outcomeOf(reader1.readAtLeast(5, new Uint8Array(10))); + const r2 = await outcomeOf(reader2.readAtLeast(10, new Uint8Array(10))); + const r3 = await outcomeOf(reader1.readAtLeast(5, new Uint8Array(10))); + const r4 = await outcomeOf(reader3.readAtLeast(2, new Uint8Array(12))); + strictEqual(r1.state, 'fulfilled'); + strictEqual(dec.decode(r1.value.value), 'helloth'); + strictEqual(r1.value.done, false); + strictEqual(r2.state, 'fulfilled'); + strictEqual(dec.decode(r2.value.value), 'hellothere'); + strictEqual(r2.value.done, false); + // The below-min tail at close is delivered done=false (a separate + // zero-length done read follows) — the DECIDED contract, parity. + strictEqual(r3.state, 'fulfilled'); + strictEqual(dec.decode(r3.value.value), 'ere'); + strictEqual(r3.value.done, false); + strictEqual(r4.state, 'fulfilled'); + strictEqual(dec.decode(r4.value.value), 'hellothere'); + strictEqual(r4.value.done, false); }, }; -// Test BYOB readAtLeast with tee complex variant 2 +// Complex variant 2: as variant 1, with byobRequest freshness asserted +// per C++ pull. export const byobReadAtLeastTeeComplex2 = { async test() { - if (usingTsImpl) { - // Tee-driven pulls present a NULL byobRequest: the source's - // unconditional c.byobRequest access throws into the stream and - // every branch read rejects with that TypeError. - const rs = teeSourceTouchingByobRequest(); - const [b1, b2] = rs.tee(); - const outcome = await outcomeOf( - b1.getReader({ mode: 'byob' }).readAtLeast(5, new Uint8Array(100)) - ); - strictEqual(outcome.state, 'rejected'); - strictEqual(outcome.reason.name, 'TypeError'); - void b2; - return; - } - const enc = new TextEncoder(); const dec = new TextDecoder(); - const chunks = ['helloth', 'ere']; + const enc = new TextEncoder(); let previousByobRequest; - const rs = new ReadableStream({ - type: 'bytes', - pull(c) { - if (chunks.length === 0) { - c.close(); - c.byobRequest.respond(0); - } else { - ok(!(c.byobRequest === previousByobRequest)); - const chunk = chunks.shift(); - enc.encodeInto(chunk, c.byobRequest.view); - c.byobRequest.respond(chunk.length); - } - }, - }); + const rs = dualPathTeeSource( + [enc.encode('helloth'), enc.encode('ere')], + (req) => { + ok(req !== previousByobRequest); + previousByobRequest = req; + } + ); const [branch1, branchB] = rs.tee(); const [branch2, branch3] = branchB.tee(); @@ -567,85 +402,60 @@ export const byobReadAtLeastTeeComplex2 = { const reader2 = branch2.getReader({ mode: 'byob' }); const reader3 = branch3.getReader({ mode: 'byob' }); - const res1 = await reader1.readAtLeast(5, new Uint8Array(10)); - strictEqual(dec.decode(res1.value), 'helloth'); - const res2 = await reader2.readAtLeast(10, new Uint8Array(10)); - strictEqual(dec.decode(res2.value), 'hellothere'); - - const res3 = await reader1.readAtLeast(5, new Uint8Array(10)); - strictEqual(dec.decode(res3.value), 'ere'); - - const res4 = await reader3.readAtLeast(2, new Uint8Array(12)); - strictEqual(dec.decode(res4.value), 'hellothere'); + const r1 = await outcomeOf(reader1.readAtLeast(5, new Uint8Array(10))); + const r2 = await outcomeOf(reader2.readAtLeast(10, new Uint8Array(10))); + const r3 = await outcomeOf(reader1.readAtLeast(5, new Uint8Array(10))); + const r4 = await outcomeOf(reader3.readAtLeast(2, new Uint8Array(12))); + strictEqual(r1.state, 'fulfilled'); + strictEqual(dec.decode(r1.value.value), 'helloth'); + strictEqual(r1.value.done, false); + strictEqual(r2.state, 'fulfilled'); + strictEqual(dec.decode(r2.value.value), 'hellothere'); + strictEqual(r2.value.done, false); + // The below-min tail at close is delivered done=false (a separate + // zero-length done read follows) — the DECIDED contract, parity. + strictEqual(r3.state, 'fulfilled'); + strictEqual(dec.decode(r3.value.value), 'ere'); + strictEqual(r3.value.done, false); + strictEqual(r4.state, 'fulfilled'); + strictEqual(dec.decode(r4.value.value), 'hellothere'); + strictEqual(r4.value.done, false); }, }; -// Test BYOB readAtLeast with tee complex variant 3 (typed arrays) +// Complex variant 3: mixed view types (Uint16/Uint8/Uint32) across two +// branches over five small chunks. export const byobReadAtLeastTeeComplex3 = { async test() { - if (usingTsImpl) { - // Tee-driven pulls present a NULL byobRequest: the source's - // unconditional c.byobRequest access throws into the stream and - // every branch read rejects with that TypeError. - const rs = teeSourceTouchingByobRequest(); - const [b1, b2] = rs.tee(); - const outcome = await outcomeOf( - b1.getReader({ mode: 'byob' }).readAtLeast(5, new Uint8Array(100)) - ); - strictEqual(outcome.state, 'rejected'); - strictEqual(outcome.reason.name, 'TypeError'); - void b2; - return; - } - const chunks = [ + const rs = dualPathTeeSource([ new Uint8Array([0x01]), new Uint8Array([0x02]), new Uint8Array([0x03]), new Uint8Array([0x04]), new Uint8Array([0x05, 0x06]), - ]; - - const rs = new ReadableStream({ - type: 'bytes', - pull(c) { - if (chunks.length === 0) { - c.close(); - c.byobRequest.respond(0); - } else { - const view = c.byobRequest.view; - const chunk = chunks.shift(); - for (let n = 0; n < chunk.length; n++) { - view[n] = chunk[n]; - } - c.byobRequest.respond(chunk.length); - } - }, - }); + ]); const [branch1, branch2] = rs.tee(); - const reader1 = branch1.getReader({ mode: 'byob' }); const reader2 = branch2.getReader({ mode: 'byob' }); - const [res1, res2, res3, res4] = await Promise.all([ - reader1.readAtLeast(2, new Uint16Array(2)), - reader1.readAtLeast(2, new Uint8Array(2)), - reader2.readAtLeast(2, new Uint8Array(2)), - reader2.readAtLeast(1, new Uint32Array(1)), + const [o1, o2, o3, o4] = await Promise.all([ + outcomeOf(reader1.readAtLeast(2, new Uint16Array(2))), + outcomeOf(reader1.readAtLeast(2, new Uint8Array(2))), + outcomeOf(reader2.readAtLeast(2, new Uint8Array(2))), + outcomeOf(reader2.readAtLeast(1, new Uint32Array(1))), ]); - - strictEqual(res1.value instanceof Uint16Array, true); - strictEqual(res2.value instanceof Uint8Array, true); - strictEqual(res1.value[0], 0x0201); - strictEqual(res1.value[1], 0x0403); - strictEqual(res2.value[0], 0x05); - strictEqual(res2.value[1], 0x06); - - strictEqual(res3.value instanceof Uint8Array, true); - strictEqual(res4.value instanceof Uint32Array, true); - strictEqual(res3.value[0], 0x1); - strictEqual(res3.value[1], 0x2); - strictEqual(res4.value[0], 0x06050403); + // PARITY: mixed view types deliver identical assemblies. + for (const [o, Ctor, expected] of [ + [o1, Uint16Array, [513, 1027]], + [o2, Uint8Array, [5, 6]], + [o3, Uint8Array, [1, 2]], + [o4, Uint32Array, [100992003]], + ]) { + strictEqual(o.state, 'fulfilled'); + strictEqual(o.value.value instanceof Ctor, true); + strictEqual(Array.from(o.value.value).join(','), expected.join(',')); + } }, }; diff --git a/src/tests/streams/readable-byte/AGENTS.md b/src/tests/streams/readable-byte/AGENTS.md index 6ccd9e708ac..661825f9b86 100644 --- a/src/tests/streams/readable-byte/AGENTS.md +++ b/src/tests/streams/readable-byte/AGENTS.md @@ -20,20 +20,20 @@ behavior-parity (messages aside). | 3 | pull counts (hwm 1, enqueue-in-pull) | 1,1,3 (readable ledger #4 mirror) | 1,2,3 (spec) | `pullCountShape` | | 4 | sync start() throw | captured; stream errored (readable #6 mirror) | escapes constructor (spec) | `syncStartThrow`, `jsSourceError` | | 5 | byobRequest on DEFAULT read, no autoAllocate | auto-allocates anyway: view(4096), or view(16384) under the UPDATED_AUTO_ALLOCATE_CHUNK_SIZE autogate (@all-autogates) | null (spec) — the subject of the streams_no_default_auto_allocate_chunk_size flag cell | `byobRequestOnDefaultRead` | -| 6 | Body-pump reads | pump fills byobRequest (BYOB reads) | pump pulls with byobRequest NULL even WITH autoAllocateChunkSize (direct default reads DO synthesize it) — respond-driven sources need dual paths | `bodyPumpByobRequestPresence` | +| 6 | Body-pump reads | pump fills byobRequest (BYOB reads) | WITH autoAllocateChunkSize: same — the draining conduit's wait-read synthesizes the auto-allocate descriptor, so pump pulls carry a byobRequest (parity, pinned). WITHOUT it: pump pulls present byobRequest null (ledger #5's spec side) while C++ auto-allocates anyway — sources without autoAllocateChunkSize stay dual-path | `bodyPumpByobRequestPresence` | | 7 | close() with partially-filled read(view) | close succeeds; read resolves EMPTY view done=FALSE; closed fulfills | TypeError 'Insufficient bytes to fill elements in the given view' from close(), read, and closed (spec) | `closeWithPartiallyFilledView` | | 8 | enqueue of detached/zero-length chunk | TypeError 'Cannot enqueue a zero-length ArrayBuffer.' | TypeError 'chunk must have a non-zero byteLength' | `enqueueDetachedBuffer`, `enqueueChunkMultipleTimesBytes` | | 9 | released pending read's rejection | 'This ReadableStream reader has been released.' | 'This reader has been released' | `relockRespondRoutesToSecondReader` | -| 10 | respond(N) overflowing the current read's smaller view | RangeError 'Too many bytes [N]...'; second read stays pending | accepted; commits to the released descriptor; second read fulfills its view UNTOUCHED (zeros) | `relockRespondOverflowSecondView` | +| 10 | respond(N) exceeding the second reader's smaller view (released 4-byte descriptor at head) | RangeError 'Too many bytes [N]...' validated against the SECOND read's view (a C++ deviation); second read stays pending | spec: the bounds check is against the HEAD descriptor — the respond is accepted, the released descriptor's bytes are enqueued, and the second read is served from the queue (2 of 3 bytes delivered, the third queued) | `relockRespondOverflowSecondView` | | 11 | read min validation | min=0 TypeError; min>view TypeError | min=0 TypeError (other msg); min>view RANGEError | `readMinValidation` | -| 12 | close() below min with partial bytes (WPT read-min disable root) | read fulfills partial, done=false | read PENDS FOREVER while closed fulfills — BOTH nonconforming (spec: TypeError) | `closeBelowMin` | -| 13 | readAtLeast/min at native end-of-stream | below-min tail delivered done=false, then an extra read resolves done + empty view | done=true folded into the final below-min bytes | `readAtLeastByobReader` | +| 12 | close() below min with partial bytes | read fulfills the partial bytes done=false; a subsequent read resolves done + empty view (the DECIDED tail contract; the spec's TypeError shape is implemented by neither side) | same — the parked read settles via the deferred end-of-data commit, one microtask after close() | `closeBelowMin` | +| 13 | readAtLeast/min at native end-of-stream | below-min tail delivered done=false, then an extra read resolves done + empty view | same (the conduit's under-delivery commit; decided contract) | `readAtLeastByobReader` | | 14 | tee cancel composite | pair-completing branch's reason only (readable #11 mirror) | AggregateError[r1, r2]; lone-branch cancel PENDS — never await it | `teeCancelComposite` | | 15 | respondWithNewView with a different element size | adopts the NEW view's element size (6 bytes at once) | keeps the ORIGINAL read view's element size (4-byte multiple), queues the remainder | `readableStreamByteRespondWithNewViewUsesNewElementSize` | | 16 | invalidated byobRequest message | 'This ReadableStreamBYOBRequest has been invalidated.' | 'This BYOB request has been invalidated' | `readableStreamByteRespond` | | 17 | default-read delivery of a multi-chunk queue | COALESCES all queued chunks into one read | chunk-by-chunk (spec) | `byteDesiredSizeAccounting` | | 18 | buffer-hazard messages (read detached view, respond after view detach, respondWithNewView foreign buffer, WASM Memory) | own texts | own texts (behavior parity everywhere) | `buffer-lifecycle.js` | -| 19 | close() with a pending UNFILLED BYOB read | read resolves done with an empty view | read PENDS FOREVER while close() succeeds (bounded; the #12 defect family without any min) — drain loops must close WITH the last enqueue, never against a parked empty read | `closeWithPendingUnfilledByobRead` | +| 19 | close() with a pending UNFILLED BYOB read | read resolves done with an empty view | same — via the deferred end-of-data commit (a same-turn respond(0) claims the spec's committed shape first) | `closeWithPendingUnfilledByobRead` | Parity worth noting (probed, pinned): byte hwm defaults to 0 with NO automatic pull; pull-throw and error-then-throw identity; enqueue @@ -113,8 +113,8 @@ named suite test pins directly, differing only in incidental asserts. | `js-compat.js` | ledger #17; 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); the conduit drives pull with byobRequest null (ledger #5/#6 shape); expectedLength undefined; 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 (see ledger #19) | +| `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; 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) | 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/controller.js b/src/tests/streams/readable-byte/controller.js index 12aa5c6919e..a162f4aeb37 100644 --- a/src/tests/streams/readable-byte/controller.js +++ b/src/tests/streams/readable-byte/controller.js @@ -149,10 +149,11 @@ export const readDetachesCallerBuffer = { }, }; -// close() while an UNFILLED BYOB read is pending. DIVERGENCE: C++ -// resolves the read done with an empty view; the TypeScript read PENDS -// FOREVER while close() itself succeeds (bounded observation — the -// close-below-min defect family, without any min involved). +// close() while an UNFILLED BYOB read is pending: the read resolves done +// with an empty view over the transferred buffer (parity; under +// TypeScript the parked read settles via the deferred end-of-data +// commit, one microtask after close() — a same-turn respond(0) would +// win with the spec's committed shape instead). export const closeWithPendingUnfilledByobRead = { async test() { let controller; @@ -172,13 +173,9 @@ export const closeWithPendingUnfilledByobRead = { ), scheduler.wait(250).then(() => ({ state: 'pending' })), ]); - if (usingTsImpl) { - strictEqual(outcome.state, 'pending'); - } else { - strictEqual(outcome.state, 'fulfilled'); - strictEqual(outcome.r.done, true); - strictEqual(outcome.r.value.byteLength, 0); - } + strictEqual(outcome.state, 'fulfilled'); + strictEqual(outcome.r.done, true); + strictEqual(outcome.r.value.byteLength, 0); await reader.closed; }, }; diff --git a/src/tests/streams/readable-byte/data-volumes.js b/src/tests/streams/readable-byte/data-volumes.js index f5810b94cb5..e7f2b356349 100644 --- a/src/tests/streams/readable-byte/data-volumes.js +++ b/src/tests/streams/readable-byte/data-volumes.js @@ -31,12 +31,11 @@ function assertPatternedBytes(bytes, total) { } // A byte ReadableStream producing `total` patterned bytes in -// `chunkLength`-sized enqueues. -// Closes in the SAME pull as the final enqueue: a close() with a -// pending unfilled BYOB read pends forever under the TypeScript -// implementation (pinned in controller.js -// closeWithPendingUnfilledByobRead), so a drain loop must never park a -// fresh read against an empty, about-to-close source. +// `chunkLength`-sized enqueues, closing in the SAME pull as the final +// enqueue. (A read parked against an empty, about-to-close source also +// works — close() settles pending BYOB reads done with an empty view, +// pinned in controller.js closeWithPendingUnfilledByobRead — but +// closing with the last enqueue keeps the volume loops single-shape.) function patternedByteSource(total, chunkLength) { let offset = 0; return new ReadableStream({ diff --git a/src/tests/streams/readable-byte/draining-reader.js b/src/tests/streams/readable-byte/draining-reader.js index eaf387dabae..1c211ac1618 100644 --- a/src/tests/streams/readable-byte/draining-reader.js +++ b/src/tests/streams/readable-byte/draining-reader.js @@ -11,10 +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); the conduit drives pull like a -// default reader — under the TS implementation byobRequest is null even -// with autoAllocateChunkSize set (the suite's ledger #5/#6 shape); byte -// streams never declare an expectedLength. +// INTACT (no coalescing, no re-slicing); 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); +// byte streams never declare an expectedLength. /* global ReadableStreamDrainingReader */ @@ -53,9 +53,11 @@ export const drainingReaderDrivesBytePull = { strictEqual(typeof ReadableStreamDrainingReader, 'undefined'); return; } - // The conduit's demand drives pull() like a default reader's: the - // TS implementation presents byobRequest null even with - // autoAllocateChunkSize set, so the source must enqueue. + // The conduit's demand drives pull() like a default reader's: with + // autoAllocateChunkSize set, each wait-read synthesizes the + // auto-allocate descriptor, so every pull observes a byobRequest. + // The source may still enqueue() — the enqueued chunk fulfills the + // read directly and the auto-allocated buffer is discarded. const byobRequests = []; let pulls = 0; const rs = new ReadableStream({ @@ -76,7 +78,7 @@ export const drainingReaderDrivesBytePull = { if (done) break; } deepStrictEqual(seen, [1, 2]); - deepStrictEqual(byobRequests, ['null', 'null', 'null']); + deepStrictEqual(byobRequests, ['present', 'present', 'present']); }, }; diff --git a/src/tests/streams/readable-byte/read-min.js b/src/tests/streams/readable-byte/read-min.js index a1bb73972ef..6a3ab0f15da 100644 --- a/src/tests/streams/readable-byte/read-min.js +++ b/src/tests/streams/readable-byte/read-min.js @@ -4,8 +4,11 @@ // The read-minimum machinery: the standard read(view, {min}) option and // the workerd readAtLeast(min, view) extension (implemented by BOTH -// sides). The WPT read-min.any suite is disabled for hangs; the -// close-below-min divergence pinned here is the underlying reason. +// sides). Close-below-min follows the DECIDED tail contract (matching +// C++, diverging from the spec's TypeError): the below-min bytes are +// delivered done=false and a subsequent read resolves done with an +// empty view. The WPT read-min.any suite stays disabled on the C++ +// side for its own hangs; the TypeScript side runs it. import { strictEqual, ok, throws, rejects } from 'node:assert'; import { usingTsImpl } from 'which-impl'; @@ -102,11 +105,12 @@ export const readMinValidation = { }, }; -// DIVERGENCE (the WPT read-min disable root): close() while a min-read -// holds SOME bytes (2 of 3). C++ fulfills the read with the partial -// bytes and done=false; TypeScript leaves the read PENDING FOREVER -// while reader.closed fulfills (bounded observation). The spec calls -// for a TypeError — neither side conforms. +// close() while a min-read holds SOME bytes (2 of 3): the read fulfills +// with the partial bytes, done=false, and a subsequent read resolves +// done with an empty view — the DECIDED readAtLeast tail contract, +// parity on both implementations. (The spec's TypeError-on-close shape +// is implemented by neither side; TypeScript settles the parked read +// via a deferred end-of-data commit one microtask after close().) export const closeBelowMin = { async test() { const { rs, controller } = byteStream(); @@ -117,19 +121,15 @@ export const closeBelowMin = { await scheduler.wait(5); controller().close(); strictEqual(await reader.closed, undefined); - if (usingTsImpl) { - const outcome = await Promise.race([ - read.then(() => 'settled'), - scheduler.wait(100).then(() => 'pending'), - ]); - strictEqual(outcome, 'pending'); - } else { - const { value, done } = await read; - strictEqual(done, false); - strictEqual(value.byteLength, 2); - strictEqual(value[0], 1); - strictEqual(value[1], 2); - } + const { value, done } = await read; + strictEqual(done, false); + strictEqual(value.byteLength, 2); + strictEqual(value[0], 1); + strictEqual(value[1], 2); + const tail = await reader.read(new Uint8Array(4)); + strictEqual(tail.done, true); + ok(tail.value instanceof Uint8Array); + strictEqual(tail.value.byteLength, 0); }, }; @@ -209,17 +209,14 @@ export const readAtLeastByobReader = { result = await reader.readAtLeast(4, new Uint8Array(20)); value = new TextDecoder().decode(result.value); strictEqual(value, 'az'); - // DIVERGENCE in the end-of-stream shape: TypeScript reports done - // together with the final below-min bytes; C++ returns them - // done=false and requires one more read, which resolves done with - // an empty view (the pinned internal_stream_byob_return_view + // End-of-stream tail shape (the DECIDED contract, parity): the + // below-min tail is delivered done=false; one more read resolves + // done with an empty view (the pinned internal_stream_byob_return_view // behavior). - strictEqual(result.done, usingTsImpl); - if (!usingTsImpl) { - result = await reader.readAtLeast(4, new Uint8Array(20)); - strictEqual(result.done, true); - ok(result.value instanceof Uint8Array); - strictEqual(result.value.byteLength, 0); - } + strictEqual(result.done, false); + result = await reader.readAtLeast(4, new Uint8Array(20)); + strictEqual(result.done, true); + ok(result.value instanceof Uint8Array); + strictEqual(result.value.byteLength, 0); }, }; diff --git a/src/tests/streams/readable-byte/release-relock.js b/src/tests/streams/readable-byte/release-relock.js index f9f96343756..25d7c9cb7dd 100644 --- a/src/tests/streams/readable-byte/release-relock.js +++ b/src/tests/streams/readable-byte/release-relock.js @@ -147,12 +147,16 @@ export const relockAutoAllocateEnqueue = { }, }; -// DIVERGENCE: respond(3) when the released 4-byte descriptor heads the -// queue but the second reader's read wants only 2 bytes. C++ throws -// RangeError from respond() and the second read stays pending -// (bounded); TypeScript accepts the respond, committing the bytes to -// the released descriptor, and fulfills the second read with its -// 2-byte view UNTOUCHED (zeros). +// DIVERGENCE — C++ deviates from the spec: respond(3) when the released +// 4-byte descriptor heads the queue but the second reader's read wants +// only 2 bytes. The spec bounds-checks respond() against the HEAD +// descriptor (the released 4-byte one — 3 fits), fills it, enqueues the +// filled bytes ('none' reader type), and services the second read from +// the queue: 2 of the 3 responded bytes delivered done=false, the third +// queued for the next read. TypeScript implements exactly that (data +// flow asserted below). C++ instead validates against the SECOND read's +// 2-byte view and throws RangeError, leaving the second read pending +// (bounded). export const relockRespondOverflowSecondView = { async test() { const { rs, controller } = byteStream(); @@ -165,12 +169,20 @@ export const relockRespondOverflowSecondView = { const read2 = r2.read(new Uint8Array(2)); const req = controller().byobRequest; if (usingTsImpl) { + req.view[0] = 7; + req.view[1] = 8; + req.view[2] = 9; req.respond(3); const { value, done } = await read2; strictEqual(done, false); strictEqual(value.byteLength, 2); - strictEqual(value[0], 0); - strictEqual(value[1], 0); + strictEqual(value[0], 7); + strictEqual(value[1], 8); + // The remainder byte stays queued for the next read. + const read3 = await r2.read(new Uint8Array(2)); + strictEqual(read3.done, false); + strictEqual(read3.value.byteLength, 1); + strictEqual(read3.value[0], 9); } else { let caught; try { diff --git a/src/tests/streams/readable-byte/respond.js b/src/tests/streams/readable-byte/respond.js index 62e0d4b6b71..bfcbe72776c 100644 --- a/src/tests/streams/readable-byte/respond.js +++ b/src/tests/streams/readable-byte/respond.js @@ -16,19 +16,17 @@ export const responseBodyMethodsJsByob = { const enc = new TextEncoder(); const dec = new TextDecoder(); + // With autoAllocateChunkSize set, the body pump's pulls carry a + // byobRequest on both implementations (bodyPumpByobRequestPresence + // pins the presence), so a respond()-only source works as a body. { const rs = new ReadableStream({ type: 'bytes', autoAllocateChunkSize: 4096, async pull(c) { - if (c.byobRequest) { - enc.encodeInto('hello', c.byobRequest.view); - c.byobRequest.respond(5); - c.close(); - } else { - c.enqueue(enc.encode('hello')); - c.close(); - } + enc.encodeInto('hello', c.byobRequest.view); + c.byobRequest.respond(5); + c.close(); }, }); @@ -42,14 +40,9 @@ export const responseBodyMethodsJsByob = { type: 'bytes', autoAllocateChunkSize: 4096, async pull(c) { - if (c.byobRequest) { - enc.encodeInto('hello', c.byobRequest.view); - c.byobRequest.respond(5); - c.close(); - } else { - c.enqueue(enc.encode('hello')); - c.close(); - } + enc.encodeInto('hello', c.byobRequest.view); + c.byobRequest.respond(5); + c.close(); }, }); @@ -116,7 +109,9 @@ export const jsSourceAsyncPull = { }, }; -// Test BYOB ReadableStream as Response body +// Test BYOB ReadableStream as Response body. The body pump's pull +// carries a byobRequest under both implementations when +// autoAllocateChunkSize is set (see bodyPumpByobRequestPresence). export const jsByteSource = { async test() { const enc = new TextEncoder(); @@ -125,16 +120,9 @@ export const jsByteSource = { autoAllocateChunkSize: 4096, pull(c) { const request = c.byobRequest; - if (request != null) { - enc.encodeInto('hello', request.view); - request.respond(5); - c.close(); - } else { - // The TypeScript Body pump reads without a BYOB request even - // with autoAllocateChunkSize (see bodyPumpByobRequestPresence). - c.enqueue(enc.encode('hello')); - c.close(); - } + enc.encodeInto('hello', request.view); + request.respond(5); + c.close(); }, }); @@ -143,7 +131,9 @@ export const jsByteSource = { }, }; -// Test BYOB ReadableStream with multiple chunks +// Test BYOB ReadableStream with multiple chunks. No autoAllocateChunkSize +// here, so the pump's byobRequest presence differs per implementation +// (ledger #5/#6): the source stays dual-path. export const jsByteSourceMultipleChunks = { async test() { const enc = new TextEncoder(); @@ -873,7 +863,11 @@ export const bodyPumpByobRequestPresence = { }, }); strictEqual(await new Response(rs).text(), 'x'); - strictEqual(seen, usingTsImpl ? 'null' : 'view(4096)'); + // With autoAllocateChunkSize set, the body pump's pull carries a + // byobRequest over the auto-allocated buffer on both implementations + // (the TypeScript draining conduit synthesizes the descriptor for its + // wait-read). Without it, direct-read behavior is ledger #5's. + strictEqual(seen, 'view(4096)'); }, }; diff --git a/src/tests/streams/transform/AGENTS.md b/src/tests/streams/transform/AGENTS.md index 85ef25bd871..2a2d7c29017 100644 --- a/src/tests/streams/transform/AGENTS.md +++ b/src/tests/streams/transform/AGENTS.md @@ -81,7 +81,7 @@ C++ implementation; `draining-reader.js` asserts both sides. | `reentrancy.js` | size()-error UAF regressions (#2) + sequential/identity shapes; all 9 WPT reentrant-in-size() ops: 6 parity at finite hwm, #10-#12 divergences | | `buffer-lifecycle.js` | chunk identity; detach-while-queued observed by reader (parity) | | `then-interceptors.js` | ledger #7 | -| `roundtrip.js` | JS transform → ITS pipe does not hang (regression) | +| `roundtrip.js` | JS transform → ITS pipe does not hang (regression); the workerd `TransformStream({ expectedLength })` extension surfaces a concrete Content-Length on fetch/Request bodies (parity; issue #5113 regression — the TS transform advertises the value through its readable's controller) | | `gc.js` | write→read handoff survives gc() with the stream dropped (--expose-gc) | | `legacy-identity-fallback.js` / `legacy-backpressure.js` | see Compatibility flags | | `draining-reader.js` | TS only (C++ cell asserts the global's absence): writes flow through the transformer into conduit reads; a readable-side backlog plus close sentinel swept in one batch; flush() output rides the final batch; expectedLength undefined; transformer errors propagate | diff --git a/src/tests/streams/transform/roundtrip.js b/src/tests/streams/transform/roundtrip.js index b7efed71b06..f742d259b42 100644 --- a/src/tests/streams/transform/roundtrip.js +++ b/src/tests/streams/transform/roundtrip.js @@ -7,7 +7,6 @@ // from transform-streams-test.js. import { strictEqual } from 'node:assert'; -import { usingTsImpl } from 'which-impl'; export const transformRoundtrip = { async test(ctrl, env, ctx) { @@ -75,11 +74,10 @@ export const transformRoundtrip = { }; // The workerd TransformStream({ expectedLength }) extension: posting -// the readable as a fetch body. DIVERGENCE: C++ surfaces the declared -// length as a concrete Content-Length on the subrequest; the -// TypeScript implementation does not consult expectedLength — the -// subrequest goes out chunked (Content-Length null). The body arrives -// intact either way. +// the readable as a fetch body surfaces the declared length as a +// concrete Content-Length on the subrequest (parity — the TypeScript +// transform advertises it through its readable's controller and the +// C++ bridge derives the header from the draining reader). export const transformExpectedLengthFetchBody = { async test(ctrl, env) { const enc = new TextEncoder(); @@ -93,10 +91,7 @@ export const transformExpectedLengthFetchBody = { method: 'POST', body: readable, }); - strictEqual( - resp.headers.get('observed-content-length'), - usingTsImpl ? 'null' : '10' - ); + strictEqual(resp.headers.get('observed-content-length'), '10'); strictEqual(await resp.text(), 'hellohello'); }, }; @@ -119,10 +114,7 @@ export const transformExpectedLengthRequestBody = { body: readable, }) ); - strictEqual( - resp.headers.get('observed-content-length'), - usingTsImpl ? 'null' : '10' - ); + strictEqual(resp.headers.get('observed-content-length'), '10'); strictEqual(await resp.text(), 'hellohello'); }, }; diff --git a/src/tests/streams/writable/AGENTS.md b/src/tests/streams/writable/AGENTS.md index 886b20a85fb..ed5c0919a95 100644 --- a/src/tests/streams/writable/AGENTS.md +++ b/src/tests/streams/writable/AGENTS.md @@ -46,6 +46,7 @@ the abort reason). | 10 | signal.reason for reasonless abort() | undefined (pedantic_wpt: AbortError DOMException) | AbortError DOMException (spec) | `abortSignalReason` | | 11 | desiredSize while erroring | queue accounting value (pedantic_wpt: null) | null (spec) | `desiredSizeWhileErroring` | | 12 | non-callable size / released-writer messages | jsg dictionary / "This WritableStream writer has been released." | TS validator / "This writer has been released" | `nonCallableSizeThrows`, `releaseLockInsideSize` | +| 13 | releaseLock() with writes still queued | cancels them: queued writes reject with the released-writer error and their chunks are dropped (a C++ deviation — the source of its WPT piping/flow-control release-then-pipe expectedFailures) | spec: release rejects only ready/closed; queued writes stay in [[writeRequests]] and drain on the sink's schedule (a new writer can relock and they still complete — WPT flow-control pins that; behind a never-settling in-flight write they wait forever on backpressure) | `cancelWriteOnReleaseLock` | Parity worth noting (probed, pinned): the whole in-flight abort matrix — abort-before-start reason identity on ready/closed, errored-state reason @@ -86,7 +87,7 @@ promises (they resolve with undefined). | `api-surface.js` | writable globals exist; controller not constructable; bare ctor works (full IDL shape is WPT's) | | `construction.js` | ledger #1–#5, #12; fractional and ToNumber-coerced hwm accepted | | `sink-algorithms.js` | which sink hooks run with what arguments/controller; sync+async hook errors surface on writer promises (#5, #6); size() consulted per write; hook getters read once; second-write rejection fan-out; hooks silent after start throw | -| `write-semantics.js` | chunk identity (subarrays, any JS value via Object.is); multiple pending writes; settlement ordering incl. under abort (#7) | +| `write-semantics.js` | chunk identity (subarrays, any JS value via Object.is); multiple pending writes; settlement ordering incl. under abort (#7); queued-write fate at releaseLock (#13) | | `buffer-lifecycle.js` | chunks never copied/validated: already-detached AB accepted (byteLength 0); post-write() mutation, detach, resizable grow, and shrink-out-of-bounds all observed per the startedness model (#7); size() runs inside write() so queue totals are immune to later detach | | `close-semantics.js` | close-throw promise fan-out vs abort (#7, #8); double close rejects TypeError | | `abort-semantics.js` | migrated abort lifecycle: reason propagation, signal event, persistent errored state, in-flight sequencing, terminal-state interactions (#7) | diff --git a/src/tests/streams/writable/write-semantics.js b/src/tests/streams/writable/write-semantics.js index b21664d4c0f..e5314b81d25 100644 --- a/src/tests/streams/writable/write-semantics.js +++ b/src/tests/streams/writable/write-semantics.js @@ -152,11 +152,18 @@ export const writableStreamPromisesResolvedInOrder = { }; // releaseLock() with a write QUEUED behind an in-flight one. -// DIVERGENCE: C++ rejects the queued write with the released-writer -// error; under TypeScript the queued write stays PENDING FOREVER -// (bounded observation) — the release orphans it. The release itself +// DIVERGENCE — C++ deviates from the spec: it cancels queued writes at +// release (rejecting them with the released-writer error), which is why +// its WPT config carries the piping/flow-control release-then-pipe +// expectedFailures. TypeScript follows the spec: release rejects only +// ready/closed; queued writes stay in [[writeRequests]] and drain on +// the sink's schedule (a new writer — e.g. a pipe — can relock and the +// writes still complete; WPT flow-control pins that). Here the sink's +// in-flight write never settles, so the queued write legitimately waits +// forever on backpressure (bounded observation). The release itself // succeeds and the stream is re-lockable in both (migrated from -// streams-test.js). +// streams-test.js; reclassification decided 2026-08-29 — the earlier +// reading of this pin as a TypeScript orphan defect was wrong). export const cancelWriteOnReleaseLock = { async test() { const ws = new WritableStream({ diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 15943885eb3..c54b0f404fb 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -1352,7 +1352,8 @@ jsg::Promise ReadableStreamNativeSource::pullByob(jsg::Lock& js, // source performs its own internal accumulation toward minBytes (KJ tryRead semantics), // so this single read is the source's complete answer for the read: delivering fewer // than the minimum in total implicitly signals EOF (the conduit commits the partial fill - // fused as {done: true, value: partialView} and closes the stream). + // as {done: false, value: partialView} and closes the stream; the next read observes + // EOF — the C++-parity readAtLeast tail shape). size_t stashed = stash.size(); size_t minBytes = atLeast - stashed; ensureScratch(kj::max(kScratchSize, minBytes)); @@ -1413,10 +1414,10 @@ jsg::Promise ReadableStreamNativeSource::pullByob(jsg::Lock& js, webstreams::invokeMethod( js, byobRequest.getHandle(js), "respond"_kj, js.num(static_cast(total))); if (eof) { - // Fused close-commit: deliver the partial bytes, then explicitly signal EOF in the - // same pull turn. (The under-delivered respond() above already implies closure to - // the conduit, which tolerates this close as a no-op; the explicit close keeps the - // EOF signal unambiguous rather than relying on that inference.) + // Deliver the partial bytes, then explicitly signal EOF in the same pull turn. + // (The under-delivered respond() above already implies closure to the conduit, + // which tolerates this close as a no-op; the explicit close keeps the EOF signal + // unambiguous rather than relying on that inference.) webstreams::invokeMethod(js, controller.getHandle(js), "close"_kj); } }).catch_(js, [self = JSG_THIS](jsg::Lock& js, jsg::Value exception) mutable { diff --git a/src/workerd/api/tests/ts-webstreams-test.js b/src/workerd/api/tests/ts-webstreams-test.js index 9d2ed84d908..7e77f170bd1 100644 --- a/src/workerd/api/tests/ts-webstreams-test.js +++ b/src/workerd/api/tests/ts-webstreams-test.js @@ -106,10 +106,14 @@ export const nativeBackedMinReadUnderDelivery = { const stream = new Blob(['hello world']).stream(); const reader = stream.getReader({ mode: 'byob' }); // A minimum larger than the source's total: KJ tryRead semantics make the short read - // the EOF signal, and the partial fill commits fused as {done: true, value: partial}. + // the EOF signal. The partial fill is delivered done=false and the stream closes; the + // next read observes EOF with an empty view (the C++-parity readAtLeast tail shape). const { done, value } = await reader.read(new Uint8Array(64), { min: 20 }); - strictEqual(done, true); + strictEqual(done, false); strictEqual(new TextDecoder().decode(value), 'hello world'); + const eof = await reader.read(new Uint8Array(16)); + strictEqual(eof.done, true); + strictEqual(eof.value.byteLength, 0); }, }; diff --git a/src/wpt/compression-test-ts.ts b/src/wpt/compression-test-ts.ts index 903737f7261..31bb6abfa6e 100644 --- a/src/wpt/compression-test-ts.ts +++ b/src/wpt/compression-test-ts.ts @@ -10,8 +10,15 @@ import { type TestRunnerConfig } from 'harness/harness'; // (compression-test.ts): the pair is behavior-matching by design. export default { 'compression-bad-chunks.any.js': { - comment: 'brotli compression is not supported', - expectedFailures: [/brotli/], + comment: + 'INTENTIONAL SPEC DIVERGENCE (decided 2026-08-28, matching the C++ ' + + 'implementation and the identity streams): SharedArrayBuffer-backed ' + + 'chunks are accepted by copying, and invalid chunks reject only ' + + "their own write, leaving the stream usable — so the file's " + + '"read should reject" assertions hang (the read legitimately stays ' + + 'pending on a usable stream) and the whole file must be disabled, ' + + 'mirroring the C++ configuration.', + disabledTests: true, }, 'compression-constructor-error.any.js': {}, 'compression-including-empty-chunk.any.js': { @@ -44,8 +51,12 @@ export default { }, 'compression-with-detach.any.js': {}, 'decompression-bad-chunks.any.js': { - comment: 'brotli compression is not supported', - expectedFailures: [/brotli/], + comment: + 'INTENTIONAL SPEC DIVERGENCE: same invalid-chunk contract as ' + + 'compression-bad-chunks.any.js above (per-write rejection on a ' + + 'usable stream hangs the "read should reject" assertions; SAB ' + + 'chunks are accepted by copying).', + disabledTests: true, }, 'decompression-buffersource.any.js': { comment: 'brotli compression is not supported',