Align the TS ReadableStream async iterator with WebIDL

src/per_isolate/webstreams/readable.ts

@@ -496,7 +496,8 @@ let resolveGenericReaderPromise: (reader: object) => void;
 let rejectGenericReaderPromise: (reader: object, reason?: unknown) => void;
 interface ReadableStreamIteratorState<R> {
   done: boolean;
-  current?: Promise<IteratorResult<R>> | undefined;
+  // The ongoing promise: a next() result, or a return()'s steps.
+  current?: Promise<IteratorResult<R> | undefined> | undefined;
 }
 
 class ReadableStreamAsyncIteratorReadRequest<R> {
@@ -541,28 +542,21 @@ interface IteratorInternalState {
   reader: ReadableStreamDefaultReaderType;
   preventCancel: boolean;
   state: ReadableStreamIteratorState<unknown>;
-  started: boolean;
 }
 
 const iteratorStateMap = new SafeWeakMap() as WeakMap<
   object,
   IteratorInternalState
 >;
 
-function getIteratorState(iter: object): IteratorInternalState {
-  const s = iteratorStateMap.get(iter);
-  if (s === undefined) {
-    throw new TypeError('Illegal invocation');
-  }
-  return s;
-}
-
 function iteratorNextSteps(s: IteratorInternalState) {
   if (s.state.done) {
     return PromiseResolve(userReadResult(undefined, true));
   }
   if (!isReaderBoundToStream(s.reader)) {
-    throw new TypeError('The reader is not bound to a ReadableStream');
+    return PromiseReject(
+      new TypeError('The reader is not bound to a ReadableStream')
+    );
   }
   const promise = PromiseWithResolvers();
   readableStreamDefaultReaderRead(
@@ -572,34 +566,44 @@ function iteratorNextSteps(s: IteratorInternalState) {
   return promise.promise;
 }
 
-async function iteratorReturnSteps(s: IteratorInternalState, value: unknown) {
+// WebIDL's returnSteps: an already-finished iterator resolves with its own
+// result object; otherwise the stream's return algorithm runs, whose
+// promise settles with undefined. return() builds the caller's result.
+function iteratorReturnSteps(
+  s: IteratorInternalState,
+  value: unknown
+): Promise<IteratorResult<unknown> | undefined> {
   if (s.state.done) {
-    return userReadResult(value, true);
+    return PromiseResolve(userReadResult(value, true));
   }
   s.state.done = true;
 
   if (!isReaderBoundToStream(s.reader)) {
-    throw new TypeError('The reader is not bound to a ReadableStream');
+    return PromiseReject(
+      new TypeError('The reader is not bound to a ReadableStream')
+    );
   }
 
   if (!s.preventCancel) {
     const result = readableStreamReaderGenericCancel(s.reader, value);
     readableStreamReaderGenericRelease(s.reader);
-    await result;
-    return userReadResult(value, true);
+    return result as Promise<undefined>;
   }
 
   readableStreamReaderGenericRelease(s.reader);
-  return userReadResult(value, true);
+  return PromiseResolve(undefined);
 }
 
+// WebIDL: next() and return() called on anything but one of these
+// iterators return a rejected promise rather than throwing, and a call with
+// no ongoing promise runs its steps at once, so the first next() reads (and
+// pulls) synchronously.
 const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf(
   {
     next(this: object) {
-      const s = getIteratorState(this);
-      if (!s.started) {
-        s.state.current = PromiseResolve();
-        s.started = true;
+      const s = iteratorStateMap.get(this);
+      if (s === undefined) {
+        return PromiseReject(new TypeError('Illegal invocation'));
       }
       s.state.current =
         s.state.current !== undefined
@@ -613,8 +617,10 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf(
     },
 
     return(this: object, error: unknown) {
-      const s = getIteratorState(this);
-      s.started = true;
+      const s = iteratorStateMap.get(this);
+      if (s === undefined) {
+        return PromiseReject(new TypeError('Illegal invocation'));
+      }
       s.state.current =
         s.state.current !== undefined
           ? PromisePrototypeThen(
@@ -623,7 +629,11 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf(
               () => iteratorReturnSteps(s, error)
             )
           : iteratorReturnSteps(s, error);
-      return s.state.current;
+      // Unlike next(), the caller gets a separate promise chained on the
+      // ongoing one.
+      return PromisePrototypeThen(s.state.current, () =>
+        userReadResult(error, true)
+      );
     },
 
     [SymbolAsyncIterator](this: object) {
@@ -632,6 +642,13 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf(
   },
   AsyncIteratorPrototype
 );
+ObjectDefineProperty(ReadableStreamAsyncIteratorPrototype, SymbolToStringTag, {
+  __proto__: null,
+  value: 'ReadableStream AsyncIterator',
+  writable: false,
+  enumerable: false,
+  configurable: true,
+});
 // ---- end async iterator prototype ------------------------------------------
 
 class ReadableStreamReaderBase<R> {
@@ -4694,7 +4711,6 @@ class ReadableStream<R> {
       reader,
       preventCancel: !!preventCancel,
       state: { done: false, current: undefined },
-      started: false,
     });
     return iter;
   }

src/tests/streams/readable/AGENTS.md

@@ -27,7 +27,7 @@ behavioral gaps; the reentrancy family is mostly parity at finite hwm.
 | 10 | queue total-size math | integer-ish: saturates (maxSafeInt shape -2,-1,1,0), fractional sizes truncate to 0; DEFECT: reading a fractional-size chunk spins the isolate | full double-precision per spec (all four WPT shapes) | `queue-math.js` (3 tests) |
 | 11 | tee cancel composite | source cancel gets ONLY the pair-completing branch's reason; first branch's cancel promise fulfills immediately | AggregateError of every branch's reason, in the order they cancelled (identity; intentional divergence from spec's array — tee is N consumers on one queue, the last to leave cancels the source); LONE branch cancel promise PENDS until the other cancels (spec) — never await a lone branch cancel | `teeCancelReasonComposite`, `teeCancelReverseOrder` |
 | 12 | from(string) | iterates per code unit ['h','i'] | single chunk ['hi'] (spec: throws — both diverge from spec) | `fromString` |
-| 13 | async-iterator prototype | exposes constructor + next/return | next/return only | `iteratorPrototypeShape` |
+| 13 | async-iterator prototype | exposes constructor + next/return; class string 'ReadableStreamAsyncIterator' (writable) | next/return only; class string 'ReadableStream AsyncIterator' (non-writable; WebIDL) | `iteratorPrototypeShape` |
 | 14 | read() inside size() | in-flight chunk fed DIRECTLY to the reentrant read | chunk queued; the NEXT enqueue bypasses the queue into the reentrant read (spec) — deliveries swapped | `readInsideSize` |
 | 15 | adopted body stream lock after consumption | released | kept locked | `bodyIdentityAndLockCoupling` |
 | 16 | when a read waiting on enqueue()/close() looks up `then` on its result (once per read in both) | after the call has returned | inside the call, as it resolves the read (spec; Node agrees) | `thenGetterTimingForWaitingRead` |
@@ -38,7 +38,9 @@ behavioral gaps; the reentrancy family is mostly parity at finite hwm.
 | 21 | body consumption of a TransformStream readable that delivers more than its declared `expectedLength` | returns everything (consumption ignores the declaration) | rejects with RangeError 'stream delivered more bytes than its declared expectedLength' and cancels the readable with it, which errors the writable (the declaration is the exact-total contract the byte and native sources already enforce) | `transformExpectedLengthOverflow` |
 | 22 | async-iterator next() interleavings where WebIDL clears the ongoing promise | every call strictly serialized: a next() from a continuation registered on an earlier next() still waits for the queued ones, and one queued behind return() reports done; after a next() rejects, later ones reject with the same error | spec: that next() reads ahead of the queued one (n2 'c', n3 'b') and ahead of a queued return() (reads data; the return still cancels); after a next() rejects, the iterator is finished and later ones report done (this last shape is among the WPT async-iterator.any C++ expectedFailures; the read-ahead shapes are derived from the WebIDL algorithm) | `nextFromEarlierContinuationReadsAhead`, `nextFromEarlierContinuationBeatsReturn`, `nextAfterRejectedNextIsDone` |
 | 23 | when a promise returned by start() starts the stream | adopts it: the first pull runs before the first marker chained on it | spec (Node agrees): a new promise is resolved with it, so the pull runs after the second marker, whether it was fulfilled on return or later | `startPromiseSettledInNewPromise` |
-| 24 | then-getter fires on the result of an async-iterator `next()` made while another is still pending (the TypeScript first `next()` is always one) | once | twice: the `next()` settles by adopting the promise of the step it waited for (WebIDL; Node agrees) | `thenGetterPerIteratorNext` |
+| 24 | then-getter fires on the result of an async-iterator `next()` made while another is still pending | once | twice: the `next()` settles by adopting the promise of the step it waited for (WebIDL; Node agrees) | `thenGetterPerIteratorNext` |
+| 25 | async-iterator `next()`/`return()` called with a `this` that is not a stream's iterator | throws TypeError synchronously | returns a promise rejected with TypeError (WebIDL) | `iteratorMethodsRejectForeignThis` |
+| 26 | then-getter fires on the result of an async-iterator `return()` on a finished iterator | once (Node agrees) | twice: the return steps resolve with a result object of their own, and `return()` resolves the caller's promise with another (WebIDL) | `thenGetterPerIteratorReturn` |
 
 Parity worth noting (probed, pinned): pull serialization (never
 re-entered); pull/async-start rejection identity; error-undefined
@@ -51,7 +53,9 @@ hook; tee error propagation identity to both branches, tee pull-per-read
 shape, tee backpressure following the slowest branch (a push source
 stalls both branches on an idle one; the spec's per-branch queues would
 not), tee after partial read; the tee-reentrancy crash regressions;
-from() cancel plumbing identity through return(); async-iterator
+from() cancel plumbing identity through return(); the first
+async-iterator next() pulling synchronously, and a next() from the
+pull() it runs reading right behind it; async-iterator
 protocol interleavings (return/next no-await; the WebIDL
 ongoing-promise shapes are #22); chunks held BY REFERENCE
 (mutation visible, identity) + detach-while-queued observed; and the
@@ -97,15 +101,15 @@ C++ implementation; `draining-reader.js` asserts both sides.
 | `tee.js` | migrated edge cases + error propagation + cancel composite (#11) + pull-per-read + slowest-branch backpressure |
 | `tee-reentrancy.js` | the three C++ push-loop crash regressions (from api/streams/streams-test.js) |
 | `from.js` | 11 migrated + fromString (#12) + return validation messages |
-| `async-iteration.js` | 7 migrated + no-await interleavings + proto shape (#13) + ongoing-promise interleavings (#22) |
+| `async-iteration.js` | 7 migrated + no-await interleavings + proto shape and class string (#13) + foreign `this` (#25) + first next() pulls synchronously, next() from that pull() (parity) + ongoing-promise interleavings (#22) |
 | `reentrancy.js` | enqueue/close/cancel-in-size (parity) + read-in-size (#14; guard the size() or C++ captures every later chunk) |
 | `buffer-lifecycle.js` | chunk by reference, detach observed |
 | `integration-body.js` | readAll family, normalization (incl. detached and out-of-bounds views, DataViews included, SharedArrayBuffer-backed views, resizable-extent pinning), clone, cancel-then-consume, SELF round-trips |
 | `integration-body-failures.js` | consumer-side failures cancel the source with the error (every consumer; the cancel's rejection replaces it; nothing to cancel once the closing batch is in), leave the stream locked and stop pulls; TransformStream `expectedLength` overflow (#21) and exact delivery |
 | `integration-body-memory.js` | body consumption copies bytes out as they arrive: chunk buffers are collectible mid-consumption (WeakRef + gc()), ~2.5 MiB of pseudo-random chunk sizes assembles exactly as bytes and as text, one chunk wider than a 1 MiB collection block assembles exactly, a declared-length byte body is exact |
 | `integration-locked-disturbed.js` | disturbed rejected, locked accepted, body identity + lock coupling (#15) |
 | `gc.js` | pending read + async iteration survive gc(); a controller held with its stream collected (ledger #19); both tee branches collected while the controller is held: enqueue() drops without retaining (WeakRef-checked, backlog included), desiredSize at the high-water mark, close() then enqueue() as ever, and pull() stops (ledger #20); one branch cancelled and the other collected, observed in the gc()'s own job (before finalization callbacks): the controller's queries drop the consumer themselves |
-| `then-interceptors.js` | one then-getter fire per read, on an ordinary result object, whether the read is answered from the queue, waits for enqueue() or close(), or finds the stream closed (a tee branch's too), and when it fires for a waiting read (ledger #16); none for a pipe's reads; per async-iterator next() (ledger #24); a then getter that cancels tee branches while a read result settles (during enqueue() and close(), one branch or two of three): the other branches' pending reads still resolve and their backlog still counts toward desiredSize |
+| `then-interceptors.js` | one then-getter fire per read, on an ordinary result object, whether the read is answered from the queue, waits for enqueue() or close(), or finds the stream closed (a tee branch's too), and when it fires for a waiting read (ledger #16); none for a pipe's reads; per async-iterator next() (ledger #24) and return() (ledger #26); a then getter that cancels tee branches while a read result settles (during enqueue() and close(), one branch or two of three): the other branches' pending reads still resolve and their backlog still counts toward desiredSize |
 | `legacy-constructors.js` | the unflagged cell (see flags table) |
 | `draining-reader.js` | TS only (C++ cell asserts the global's absence): a queued backlog plus the close sentinel swept in ONE batched read; value chunks pass through UNTOUCHED (object identity); pull-driven yields per read with EOF as a separate empty batch; expectedLength undefined; error/cancel propagation; lock exclusivity and release |
 | `data-volumes.js` | value-stream volume axes: 4096-chunk counts, a 1 MiB single string chunk, 8 MiB aggregate (128 × 64 KiB), and 1 MiB through tee on both branches — every chunk index-encoded |

src/tests/streams/readable/async-iteration.js

@@ -244,22 +244,131 @@ export const nextThenReturnNoAwait = {
   },
 };
 
-// DIVERGENCE (the WPT async-iterator.any properties seed): the
-// iterator's prototype exposes next and return on both sides, but C++
-// also exposes a constructor property.
+// DIVERGENCE (ledger #13; the WPT async-iterator.any properties seed):
+// the iterator's prototype exposes next and return on both sides, but C++
+// also exposes a constructor property. The class string is WebIDL's
+// "ReadableStream AsyncIterator" (non-writable) in TS, and
+// "ReadableStreamAsyncIterator" (writable) in C++.
 export const iteratorPrototypeShape = {
   async test() {
     const rs = new ReadableStream();
     const it = rs[Symbol.asyncIterator]();
     deepStrictEqual(Object.getOwnPropertyNames(it), []);
-    const proto = Object.getOwnPropertyNames(Object.getPrototypeOf(it));
+    const proto = Object.getPrototypeOf(it);
     deepStrictEqual(
-      proto.sort(),
+      Object.getOwnPropertyNames(proto).sort(),
       (usingTsImpl
         ? ['next', 'return']
         : ['constructor', 'next', 'return']
       ).sort()
     );
+    const tag = usingTsImpl
+      ? 'ReadableStream AsyncIterator'
+      : 'ReadableStreamAsyncIterator';
+    deepStrictEqual(
+      Object.getOwnPropertyDescriptor(proto, Symbol.toStringTag),
+      {
+        value: tag,
+        writable: !usingTsImpl,
+        enumerable: false,
+        configurable: true,
+      }
+    );
+    strictEqual(Object.prototype.toString.call(it), `[object ${tag}]`);
+    await it.return();
+  },
+};
+
+// DIVERGENCE (ledger #25): next() and return() called on anything but a
+// stream's iterator return a promise rejected with a TypeError (WebIDL);
+// C++ throws it synchronously. A different stream's iterator is a valid
+// receiver: the call acts on that iterator.
+export const iteratorMethodsRejectForeignThis = {
+  async test() {
+    const it = new ReadableStream().values();
+    const other = new ReadableStream({
+      start(c) {
+        c.close();
+      },
+    }).values();
+    const proto = Object.getPrototypeOf(it);
+    for (const method of ['next', 'return']) {
+      for (const receiver of [undefined, {}, proto, new ReadableStream()]) {
+        if (usingTsImpl) {
+          const result = it[method].call(receiver);
+          ok(result instanceof Promise, method);
+          await rejects(result, TypeError);
+        } else {
+          throws(() => it[method].call(receiver), TypeError);
+        }
+      }
+      const result = await it[method].call(other);
+      strictEqual(result.done, true);
+      strictEqual(result.value, undefined);
+    }
+    await it.return();
+  },
+};
+
+// Parity: the first next() reads at once (WebIDL runs its steps with no
+// ongoing promise), so a started stream with an empty queue pulls
+// synchronously, as reader.read() does.
+export const firstNextPullsSynchronously = {
+  async test() {
+    let pulls = 0;
+    const rs = new ReadableStream(
+      {
+        pull(controller) {
+          controller.enqueue(++pulls);
+        },
+      },
+      { highWaterMark: 0 }
+    );
+    await scheduler.wait(0);
+    strictEqual(pulls, 0);
+    const it = rs.values();
+    const next = it.next();
+    strictEqual(pulls, 1);
+    deepStrictEqual(await next, { value: 1, done: false });
+    await it.return();
+  },
+};
+
+// Parity: a pull() run by the first next() that itself calls next() finds
+// no ongoing promise (WebIDL sets it only once the steps return), so its
+// read queues right behind the outer one. The outer next() then becomes
+// the ongoing promise, and a next() after it waits for the outer alone,
+// reading after the reentrant one.
+export const nextFromPullReadsBehindOuter = {
+  async test() {
+    const log = [];
+    let n = 0;
+    let it;
+    let inner;
+    const rs = new ReadableStream(
+      {
+        pull(controller) {
+          inner ??= it.next();
+          controller.enqueue(++n);
+        },
+      },
+      { highWaterMark: 0 }
+    );
+    await scheduler.wait(0);
+    it = rs.values();
+    const outer = it.next();
+    ok(inner instanceof Promise);
+    const third = it.next();
+    const results = [
+      ['outer', outer],
+      ['inner', inner],
+      ['third', third],
+    ];
+    for (const [name, promise] of results) {
+      promise.then(({ value }) => log.push(`${name}=${value}`));
+    }
+    await Promise.all(results.map(([, promise]) => promise));
+    deepStrictEqual(log, ['outer=1', 'inner=2', 'third=3']);
     await it.return();
   },
 };

src/tests/streams/readable/main.js

@@ -133,6 +133,9 @@ export {
   returnThenNextNoAwait,
   nextThenReturnNoAwait,
   iteratorPrototypeShape,
+  iteratorMethodsRejectForeignThis,
+  firstNextPullsSynchronously,
+  nextFromPullReadsBehindOuter,
   nextFromEarlierContinuationReadsAhead,
   nextFromEarlierContinuationBeatsReturn,
   nextAfterRejectedNextIsDone,
@@ -207,6 +210,7 @@ export {
   thenGetterTimingForWaitingRead,
   thenGetterNotConsultedByPipeReads,
   thenGetterPerIteratorNext,
+  thenGetterPerIteratorReturn,
   thenGetterCancelsBranchDuringEnqueue,
   thenGetterCancelsBranchDuringEnqueueBacklog,
   thenGetterCancelsBranchDuringClose,