streams: use Web IDL "a promise resolved with" semantics for callback results

src/js/builtins/StreamInternals.ts

@@ -31,9 +31,40 @@ export function markPromiseAsHandled(promise: Promise<unknown>) {
   $pokePromiseAsHandled(promise);
 }
 
+// Web IDL "a promise resolved with x" — always a fresh promise (the
+// assimilation hop when x is a thenable is observable in the spec's microtask
+// ordering). Promise.$resolve would return x unchanged when x is already a
+// native Promise and skip that hop; the streams ref-impl explicitly avoids
+// Promise.resolve here for the same reason.
+//
+// For native-promise results replicate PromiseResolveThenableJob's timing by
+// queuing a job that chains through the intrinsic Promise.prototype.@then:
+// same two-hop microtask ordering, and the job's try/catch + reject is the
+// spec job's step 3 (abrupt completion of the then call rejects the wrapper,
+// e.g. a throwing @@species on the result). $resolvePromise alone would do
+// Get(result, "then") once Promise.prototype.then has been replaced, reaching
+// the user override for every async callback result.
 export function shieldingPromiseResolve(result) {
-  const promise = Promise.$resolve(result);
-  if (promise.$then === undefined) promise.$then = $Promise.prototype.$then;
+  const promise = $newPromise();
+  if ($isPromise(result)) {
+    $enqueueJob(
+      (p, r) => {
+        try {
+          $Promise.prototype.$then.$call(
+            r,
+            v => $resolvePromise(p, v),
+            e => $rejectPromise(p, e),
+          );
+        } catch (e) {
+          $rejectPromise(p, e);
+        }
+      },
+      promise,
+      result,
+    );
+  } else {
+    $resolvePromise(promise, result);
+  }
   return promise;
 }
 

src/js/builtins/WritableStreamInternals.ts

@@ -587,7 +587,13 @@
   const startAlgorithm = $getByIdDirectPrivate(controller, "startAlgorithm");
   $putByIdDirectPrivate(controller, "startAlgorithm", undefined);
   const stream = $getByIdDirectPrivate(controller, "stream");
-  return Promise.$resolve(startAlgorithm.$call()).$then(
+  // Web IDL "a promise resolved with x" is always a fresh promise — when
+  // startAlgorithm() returns one (as TransformStream's does), the assimilation
+  // hop is observable in the spec's microtask ordering. Promise.$resolve would
+  // return the input promise unchanged and skip that hop.
+  // $shieldingPromiseResolve takes the hop via .$then so a monkey-patched
+  // Promise.prototype.then is not reached during assimilation.
+  return $shieldingPromiseResolve(startAlgorithm.$call()).$then(
     () => {
       const state = $getByIdDirectPrivate(stream, "state");
       $assert(state === "writable" || state === "erroring");
@@ -626,7 +632,13 @@
 
   if ("start" in underlyingSinkDict) {
     const startMethod = underlyingSinkDict["start"];
-    startAlgorithm = () => $promiseInvokeOrNoopMethodNoCatch(underlyingSink, startMethod, [controller]);
+    // UnderlyingSinkStartCallback's IDL return type is `any`, so Web IDL
+    // "invoke" performs no promise conversion here. The single "a promise
+    // resolved with startResult" wrap happens in
+    // writableStreamDefaultControllerStart (SetUpWritableStreamDefaultController
+    // step 17). A synchronous throw propagates out of the WritableStream
+    // constructor, matching spec.
+    startAlgorithm = () => startMethod.$call(underlyingSink, controller);
   }
   if ("write" in underlyingSinkDict) {
     const writeMethod = underlyingSinkDict["write"];

test/js/web/streams/streams.test.js

@@ -260,6 +260,101 @@ describe("WritableStream", () => {
     await rs.pipeTo(ws);
     expect(received).toBe("hello world");
   });
+
+  // SetUpWritableStreamDefaultController step 17: "Let startPromise be a
+  // promise resolved with startResult." Web IDL "a promise resolved with x" is
+  // always a fresh promise; when x is a thenable the PromiseResolveThenableJob
+  // hop is observable in microtask ordering. Promise.resolve(x) would return x
+  // unchanged when x is already a native Promise and skip that hop.
+  describe('[[started]] timing (Web IDL "a promise resolved with")', () => {
+    async function observe(sink) {
+      const order = [];
+      const { promise: done, resolve } = Promise.withResolvers();
+      const ws = new WritableStream({
+        ...sink,
+        write() {
+          order.push("write");
+          resolve();
+        },
+      });
+      ws.getWriter().write("x");
+      queueMicrotask(() => {
+        order.push("mt1");
+        queueMicrotask(() => order.push("mt2"));
+      });
+      await done;
+      return order;
+    }
+
+    it("start() returns a fulfilled Promise: write after the assimilation hop", async () => {
+      expect(await observe({ start: () => Promise.resolve() })).toEqual(["mt1", "mt2", "write"]);
+    });
+
+    it("start() returns undefined: write before the first queued microtask (single wrap, no double-wrap regression)", async () => {
+      expect(await observe({ start() {} })).toEqual(["write", "mt1", "mt2"]);
+    });
+
+    it("no start(): write before the first queued microtask", async () => {
+      expect(await observe({})).toEqual(["write", "mt1", "mt2"]);
+    });
+
+    it("TransformStream writable startAlgorithm returns startPromise: transform after the assimilation hop", async () => {
+      const order = [];
+      const { promise: done, resolve } = Promise.withResolvers();
+      const ts = new TransformStream({
+        transform(chunk, controller) {
+          order.push("transform");
+          controller.enqueue(chunk);
+          resolve();
+        },
+      });
+      ts.readable.getReader().read();
+      ts.writable.getWriter().write("x");
+      queueMicrotask(() => {
+        order.push("mt1");
+        queueMicrotask(() => order.push("mt2"));
+      });
+      await done;
+      expect(order).toEqual(["mt1", "mt2", "transform"]);
+    });
+
+    // PromiseResolveThenableJob step 3: an abrupt completion of the then call
+    // rejects the wrapper. Intrinsic Promise.prototype.then runs
+    // SpeciesConstructor(this), so a throwing @@species on the start()-
+    // returned promise must reject startPromise and error the stream, not
+    // leave startPromise pending forever.
+    it("start() returns a Promise whose @@species throws: the stream errors with the thrown value", async () => {
+      const speciesError = new Error("species-boom");
+      const p = Promise.resolve();
+      p.constructor = { get [Symbol.species]() { throw speciesError; } };
+
+      let closedResult = "pending";
+      const writer = new WritableStream({ start: () => p }).getWriter();
+      writer.closed.then(
+        () => { closedResult = "fulfilled"; },
+        e => { closedResult = e; },
+      );
+      // [[started]] settles within a bounded number of microtask rounds; poll
+      // rather than awaiting writer.closed so a never-settling wrapper fails
+      // fast instead of hanging the test.
+      for (let i = 0; i < 20; i++) await Promise.resolve();
+      expect(closedResult).toBe(speciesError);
+    });
+
+    // The assimilation job looks up the intrinsic @then on Promise.prototype
+    // directly, not via the result's prototype chain, so a native promise with
+    // a detached prototype still lets [[started]] flip and writes proceed.
+    it("start() returns a Promise with a null prototype: [[started]] flips and write proceeds", async () => {
+      const p = Promise.resolve();
+      Object.setPrototypeOf(p, null);
+
+      let written = false;
+      const writer = new WritableStream({ start: () => p, write() { written = true; } }).getWriter();
+      writer.write("x").catch(() => {});
+      for (let i = 0; i < 20; i++) await Promise.resolve();
+      expect(written).toBe(true);
+    });
+  });
 });
 
 describe("ReadableStream.prototype.tee", () => {