Handle multiple algorithm clearance/re-entrancy issues

src/per_isolate/webstreams/readable.ts

@@ -455,6 +455,9 @@ let setDefaultControllerExpectedLength: <R>(
   controller: ReadableStreamDefaultController<R>,
   length: bigint | undefined
 ) => void;
+let defaultControllerCanCloseOrEnqueue: <R>(
+  controller: ReadableStreamDefaultController<R>
+) => boolean;
 
 let setReadableStreamPendingClosure: <R>(stream: ReadableStream<R>) => void;
 let isReadableStreamPendingClosure: <R>(stream: ReadableStream<R>) => boolean;
@@ -1514,6 +1517,9 @@ class ReadableStreamDefaultController<
       controller.#expectedLength = length;
     };
 
+    defaultControllerCanCloseOrEnqueue = (controller) =>
+      controller.#canCloseOrEnqueue();
+
     controllerCancelSteps = (controller, reason) => {
       if (#queue in controller) {
         return (controller as ReadableStreamDefaultController).#cancelSteps(
@@ -5334,6 +5340,12 @@ module.exports = {
     getStoredError: <R>(stream: ReadableStream<R>) =>
       getReadableStreamStoredError(stream),
     normalizeExpectedLength,
+    // ReadableStreamDefaultControllerCanCloseOrEnqueue, for the
+    // TransformStream controller's enqueue pre-check.
+    canCloseOrEnqueue: <R>(controller: object) =>
+      defaultControllerCanCloseOrEnqueue(
+        controller as ReadableStreamDefaultController<R>
+      ),
     // The identity streams' delivery (see identity.ts): a batched enqueue
     // with one notification, and the queue's consumption notification.
     enqueueBytesBatch: (controller: object, chunks: ArrayBufferView[]) =>

src/per_isolate/webstreams/transform.ts

@@ -285,13 +285,12 @@ class TransformStream<I = unknown, O = unknown> {
         throw new TypeError('TransformStream is not fully initialized');
       }
       // Spec TransformStreamDefaultControllerEnqueue step 4:
-      // if ReadableStreamDefaultControllerCanCloseOrEnqueue is false,
-      // throw a TypeError.  This pre-check must happen BEFORE we attempt
-      // the enqueue, so that enqueue-after-error throws TypeError (not
-      // the storedError).  The readable controller's own enqueue() does
-      // the same check internally; the pre-check here ensures the
-      // try/catch below only catches size()-originated errors.
-      if (readableInternals.getState(stream.#readable) !== 'readable') {
+      // if ReadableStreamDefaultControllerCanCloseOrEnqueue is false
+      // (close requested, or the readable no longer readable), throw a
+      // TypeError.  The readable controller's own enqueue() makes the
+      // same check, but its TypeError would land in the catch below,
+      // which treats every error as size()'s and errors the writable.
+      if (!readableInternals.canCloseOrEnqueue(readableController)) {
         throw new TypeError(
           'Cannot enqueue a chunk into a stream that is closed or has been errored'
         );
@@ -304,8 +303,10 @@ class TransformStream<I = unknown, O = unknown> {
         // Spec step 5.2: throw stream.[[readable]].[[storedError]], not
         // the caught exception.  When size() calls controller.error(e1)
         // then throws e2, the storedError is e1 — the first error wins.
-        const storedError = readableInternals.getStoredError(stream.#readable);
-        throw storedError !== undefined ? storedError : e;
+        // When size() closed the readable first (readable.cancel(),
+        // terminate()), the readable has no stored error and this throws
+        // undefined, as the spec and Node do.
+        throw readableInternals.getStoredError(stream.#readable);
       }
       // Mirror the readable side's backpressure state.
       const desiredSize = readableControllerDesiredSizeGet(readableController);

src/per_isolate/webstreams/writable.ts

@@ -1019,12 +1019,17 @@ class WritableStreamDefaultController<
       controller.#errorIfNeeded(reason);
     };
 
+    // [[AbortSteps]]: the algorithms are cleared after the abort algorithm
+    // returns, so a write made from inside sink.abort() still consults
+    // size() (WritableStreamDefaultWriterWrite step 4) before it rejects.
     controllerAbortSteps = (controller, reason) => {
       const abortAlgorithm = controller.#abortAlgorithm;
+      const result =
+        abortAlgorithm === undefined
+          ? (PromiseResolve() as Promise<void>)
+          : abortAlgorithm(reason);
       controller.#clearAlgorithms();
-      return abortAlgorithm === undefined
-        ? (PromiseResolve() as Promise<void>)
-        : abortAlgorithm(reason);
+      return result;
     };
 
     controllerClose = (controller) => {
@@ -1267,12 +1272,15 @@ class WritableStreamDefaultController<
     // Dequeue the close marker; the queue must then be empty.
     this.#queue.shift();
     this.#queueTotalSize = 0;
+    // As in [[AbortSteps]], the algorithms are cleared after the close
+    // algorithm returns: a write made from inside sink.close() consults
+    // size() before it rejects.
     const closeAlgorithm = this.#closeAlgorithm;
-    this.#clearAlgorithms();
     const promise =
       closeAlgorithm === undefined
         ? (PromiseResolve() as Promise<void>)
         : closeAlgorithm();
+    this.#clearAlgorithms();
     PromisePrototypeThen(
       promise,
       () => {

src/tests/streams/transform/AGENTS.md

@@ -36,6 +36,7 @@ basics) are parity under direct observation and are pinned as such.
 | 17 | when transformer.start() starts the pair (a write queued during start) | transform() runs after the first marker, for a synchronous start() and a returned promise alike | spec (Node agrees): after the third marker chained on a promise fulfilled before construction (synchronous start()), after the fourth chained on a returned promise fulfilled in the same turn — the start promise is resolved with start()'s result and each inner controller settles a new promise with it | `startSettlementTiming` |
 | 18 | transformer object after the stream closes (stream and controller still held) | retained | collectable (spec ClearAlgorithms; Node agrees); after abort, cancel, terminate() and error() both collect | `transformerCollectedAfterFinish` |
 | 19 | write released by a read, reaching the sink after a same-turn readable.cancel() cleared the algorithms (writable still writable) | rejects with 'The readable side of this TransformStream is no longer readable.' (TypeError); with a cancel hook, transform() still runs after cancel() | spec undefined: transform() is not called and the write rejects with the writable's stored error (the cancel reason, or the cancel hook's rejection) once the cancel settles; Node fulfills the write and drops the chunk | `writeReachesSinkAfterCancelClearedAlgorithms` |
+| 20 | controller.enqueue() whose readable-side size() throws | enqueue() returns without throwing; with no re-entry in size() the write never settles; a readable.cancel() in size() errors the writable with the cancel reason | throws the readable's stored error (spec; Node agrees): e1 after error(e1), undefined after error(undefined), and undefined after readable.cancel() or terminate() (the closed readable has no stored error); the writable is errored with size()'s error unless an earlier error got there | `enqueueThrowsReadableStoredError` |
 
 Parity worth noting (probed, pinned): cancel hook runs (not flush) with
 the reason for both readable.cancel() and writable.abort(), exactly once
@@ -80,12 +81,12 @@ C++ implementation; `draining-reader.js` asserts both sides.
 | `api-surface.js` | transform globals; controller not constructable; bare ctor is standard pass-through, not ITS |
 | `construction.js` | ledger #5, #6, #9 |
 | `transformer-algorithms.js` | ledger #17; start/transform/flush ordering, async hooks, chunk-type freedom; hook shape + prototype-chain (parity) |
-| `error-propagation.js` | sync/async start/transform/flush error fan-out across writes/close/readable (#1); controller.error() rejects reads; error() no-op after hook throw (parity, identity) |
+| `error-propagation.js` | sync/async start/transform/flush error fan-out across writes/close/readable (#1); controller.error() rejects reads; error() no-op after hook throw (parity, identity); enqueue() once flush() has settled throws TypeError and leaves the writable's close untouched (parity) |
 | `backpressure.js` | writable desiredSize through the transform; dual strategies; default readable hwm 0; latch + racy release (#8); a parked write released by a read transforms even when a same-turn enqueue re-asserts backpressure (parity at hwm 0; #14 at hwm 1) or rejects when the turn also errors the controller (#15) |
 | `cancel-matrix.js` | cancel-hook reason/identity/once (parity) + fan-out, sync (parity) and async (#3) |
 | `cancel-outcomes.js` | readable.cancel() outcome with a same-turn abort, terminate(), size() error or close, hooks and zero-hook path (#16); a write reaching the sink after a same-turn cancel cleared the algorithms (#19) |
 | `terminate.js` | terminate fates (parity) + late error (#4) + terminate-after-cancel (#13) |
-| `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 |
+| `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; what enqueue() throws when size() throws after re-entering (#20) |
 | `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); 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) |

src/tests/streams/transform/error-propagation.js

@@ -294,3 +294,65 @@ export const errorNoopAfterTransformThrow = {
     strictEqual(closedErr, thrown);
   },
 };
+
+// controller.enqueue() once the readable's close is requested but its
+// queue still holds a chunk: flush() has settled, so the transform has
+// closed the readable, and the writable's close is still in flight for a
+// few microtasks. The enqueue throws a TypeError and leaves the writable
+// alone (spec TransformStreamDefaultControllerEnqueue step 4, Node
+// agrees): its desiredSize is unchanged and ready does not reject. The
+// enqueue is retried at successive microtasks to cover that window
+// whatever its length (parity; C++ keeps it open longer).
+export const enqueueAfterCloseRequestedLeavesWritable = {
+  async test() {
+    let threwInWindow = false;
+    for (let n = 0; n <= 4; n++) {
+      let thrown;
+      let before;
+      let after;
+      let writer;
+      const ts = new TransformStream(
+        {
+          flush(controller) {
+            // Stays queued: no reader yet, readable hwm 0.
+            controller.enqueue('a');
+            let p = Promise.resolve();
+            for (let i = 0; i < n; i++) p = p.then(() => {});
+            p.then(() => {
+              before = writer.desiredSize;
+              try {
+                controller.enqueue('b');
+                thrown = null;
+              } catch (e) {
+                thrown = e;
+              }
+              after = writer.desiredSize;
+            });
+          },
+        },
+        undefined,
+        { highWaterMark: 0 }
+      );
+      writer = ts.writable.getWriter();
+      const close = writer.close();
+      await scheduler.wait(0);
+      const data = await consume(ts.readable);
+      strictEqual(await close, undefined);
+      strictEqual(await writer.closed, undefined);
+      strictEqual(await writer.ready, undefined);
+      if (n === 0) {
+        // Before flush() settles: the enqueue is accepted.
+        strictEqual(thrown, null);
+        strictEqual(data, 'ab');
+        continue;
+      }
+      ok(thrown instanceof TypeError, `tick ${n}`);
+      strictEqual(data, 'a', `tick ${n}`);
+      strictEqual(after, before, `tick ${n}`);
+      if (before !== 0) threwInWindow = true;
+    }
+    // At least one retry landed while the writable's close was still in
+    // flight (desiredSize not yet 0).
+    ok(threwInWindow);
+  },
+};

src/tests/streams/transform/main.js

@@ -51,6 +51,7 @@ export {
   asyncErrorDuringFlush,
   errorInTransformFlush,
   errorNoopAfterTransformThrow,
+  enqueueAfterCloseRequestedLeavesWritable,
 } from 'error-propagation';
 
 export {
@@ -76,6 +77,7 @@ export {
   readInsideSize,
   writerCloseInsideSize,
   writableAbortInsideSize,
+  enqueueThrowsReadableStoredError,
 } from 'reentrancy';
 
 export { transformRoundtrip } from 'roundtrip';

src/tests/streams/transform/reentrancy.js

@@ -477,3 +477,73 @@ export const writableAbortInsideSize = {
     strictEqual(await abortPromise, undefined);
   },
 };
+
+// controller.enqueue() whose readable-side size() throws e2 after
+// re-entering the stream throws the readable's stored error, not e2
+// (spec TransformStreamDefaultControllerEnqueue step 5.2; Node agrees):
+// error(e1) gives e1, error(undefined) gives undefined, and a
+// readable.cancel() or terminate(), which close the readable, give
+// undefined (it has no stored error). The writable is errored with e2
+// unless an earlier error got there first. DIVERGENCE (#20): C++
+// swallows the throw, so enqueue() returns; with no re-entry it leaves
+// the write pending forever, so that shape is TS only.
+export const enqueueThrowsReadableStoredError = {
+  async test() {
+    const e1 = new Error('e1');
+    const e2 = new Error('e2');
+    const shapes = [
+      ['error(undefined)', (c) => c.error(undefined), undefined, undefined],
+      ['error(e1)', (c) => c.error(e1), e1, e1],
+      ['cancel', (c, ts) => ts.readable.cancel(new Error('r')), undefined, e2],
+      ['terminate', (c) => c.terminate(), undefined, TypeError],
+    ];
+    if (usingTsImpl) shapes.push(['no re-entry', () => {}, e2, e2]);
+    for (const [name, inSize, expectedThrow, expectedClosed] of shapes) {
+      let outcome = 'not called';
+      let controller;
+      const ts = new TransformStream(
+        {
+          start(c) {
+            controller = c;
+          },
+          transform(chunk, c) {
+            try {
+              c.enqueue(chunk);
+              outcome = 'returned';
+            } catch (e) {
+              outcome = { thrown: e };
+            }
+          },
+        },
+        undefined,
+        {
+          highWaterMark: 1,
+          size() {
+            inSize(controller, ts);
+            throw e2;
+          },
+        }
+      );
+      const writer = ts.writable.getWriter();
+      strictEqual(await writer.write('x'), undefined, name);
+      if (usingTsImpl) {
+        ok(typeof outcome === 'object', name);
+        strictEqual(outcome.thrown, expectedThrow, name);
+      } else {
+        strictEqual(outcome, 'returned', name);
+      }
+      const closedError = await writer.closed.then(
+        () => 'fulfilled',
+        (e) => e
+      );
+      if (expectedClosed === TypeError) {
+        ok(closedError instanceof TypeError, name);
+      } else if (!usingTsImpl && name === 'cancel') {
+        // The cancel reason reaches C++'s writable first.
+        strictEqual(closedError.message, 'r', name);
+      } else {
+        strictEqual(closedError, expectedClosed, name);
+      }
+    }
+  },
+};

src/tests/streams/writable/AGENTS.md

@@ -48,6 +48,7 @@ the abort reason).
 | 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` |
 | 14 | when a promise returned by start() starts the stream | adopts it: a write queued during start reaches the sink before the first marker chained on it | spec (Node agrees): a new promise is resolved with it, so the write reaches the sink after the second marker, whether it was fulfilled on return or later | `startPromiseSettledInNewPromise` |
+| 15 | a write from inside sink.close() whose size() throws | rejects with size()'s error | TypeError (closing): the close in flight is checked before the erroring state (spec; Node agrees); the close completes either way | `sizeConsultedForWriteInsideSinkClose` |
 
 Parity worth noting (probed, pinned): the whole in-flight abort matrix —
 abort-before-start reason identity on ready/closed, errored-state reason
@@ -68,7 +69,9 @@ ArrayBuffer is an acceptable chunk (byteLength 0); a coercing
 `{valueOf}` highWaterMark is accepted via ToNumber; the user size() runs
 with an undefined receiver and one argument; reentrant `writer.write()`
 from size() enqueues the inner chunk first; size() is never consulted
-for doomed writes (the TS sinks' `willAcceptWrite` invariant); a patched
+for doomed writes (the TS sinks' `willAcceptWrite` invariant), but is
+for a write made from inside sink.close() or sink.abort(), which then
+rejects (closing / the abort reason); a patched
 `Object.prototype.then` getter never fires while settling writer
 promises (they resolve with undefined).
 
@@ -96,7 +99,7 @@ promises (they resolve with undefined).
 | `abort-semantics.js` | migrated abort lifecycle: reason propagation, signal event, persistent errored state, in-flight sequencing, outstanding-write startedness, terminal-state interactions (#7) |
 | `abort-matrix.js` | probed parity matrix (see above), including controller-error races and sink-hook suppression + signal reason (#10) + concurrent-abort identity (#6b) |
 | `backpressure.js` | desiredSize accounting/recovery; ready replaced at capacity; WPT floating-point scenarios and invalid sizes, incl. after close() (#9); erroring desiredSize (#11) |
-| `reentrancy.js` | size()-reentrant write ordering; releaseLock inside size (#12; flag-gated, cf. legacy-writer); controller.error inside write hook; doomed-write size skip; size receiver/arity |
+| `reentrancy.js` | size()-reentrant write ordering; releaseLock inside size (#12; flag-gated, cf. legacy-writer); controller.error inside write hook; doomed-write size skip; size receiver/arity; size() consulted for a write from inside sink.close() (#15) and sink.abort() |
 | `then-interceptors.js` | then-getter never fires on writer promise settlement |
 | `gc.js` | pending write survives gc() with all user refs dropped (--expose-gc) |
 | `legacy-ctor-gate.js` | fully-unflagged: ctor Error + absent controller global |

src/tests/streams/writable/main.js

@@ -107,6 +107,8 @@ export {
   controllerErrorInsideWriteHook,
   sizeNotCalledForDoomedWrite,
   sizeReceiverAndArity,
+  sizeConsultedForWriteInsideSinkClose,
+  sizeConsultedForWriteInsideSinkAbort,
 } from 'reentrancy';
 
 export { thenGetterDoesNotFireOnWriterPromises } from 'then-interceptors';