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

src/js/builtins/StreamInternals.ts

@@ -31,9 +31,13 @@ export function markPromiseAsHandled(promise: Promise<unknown>) {
   $pokePromiseAsHandled(promise);
 }
 
+// Web IDL "a promise resolved with x": NewPromiseCapability + Resolve(x).
+// Always a fresh promise — Promise.$resolve(x) would return x unchanged when
+// x is already a native Promise and skip the assimilation hop, which is
+// observable in WPT's microtask ordering.
 export function shieldingPromiseResolve(result) {
-  const promise = Promise.$resolve(result);
-  if (promise.$then === undefined) promise.$then = $Promise.prototype.$then;
+  const promise = $newPromise();
+  $resolvePromise(promise, result);
   return promise;
 }
 

src/js/builtins/WritableStreamInternals.ts

@@ -587,7 +587,11 @@ export function writableStreamDefaultControllerStart(controller) {
   const startAlgorithm = $getByIdDirectPrivate(controller, "startAlgorithm");
   $putByIdDirectPrivate(controller, "startAlgorithm", undefined);
   const stream = $getByIdDirectPrivate(controller, "stream");
-  return Promise.$resolve(startAlgorithm.$call()).$then(
+  // SetUpWritableStreamDefaultController step 17: "a promise resolved with
+  // startResult". When startAlgorithm() returns a promise (TransformStream's
+  // does), Promise.$resolve would return it unchanged and skip the spec's
+  // assimilation hop.
+  return $shieldingPromiseResolve(startAlgorithm.$call()).$then(
     () => {
       const state = $getByIdDirectPrivate(stream, "state");
       $assert(state === "writable" || state === "erroring");
@@ -626,7 +630,13 @@ export function setUpWritableStreamDefaultControllerFromUnderlyingSink(
 
   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/bun/util/readablestreamtoarraybuffer.test.ts

@@ -1,7 +1,11 @@
 import { expect, test } from "bun:test";
 
 test("readableStreamToArrayBuffer works", async () => {
-  // the test calls InternalPromise.then. this test ensures that such function is not user-overridable.
+  // Bun.readableStreamToArray returns an InternalPromise, whose own .then is
+  // not Promise.prototype.then; this test pins that the helper's chaining is
+  // unaffected by a user-patched .then. Sync start() so the spec's start-
+  // result wrap (which per Web IDL does call public .then for thenables) is
+  // not in play.
   let _then = Promise.prototype.then;
   let counter = 0;
   // @ts-ignore
@@ -12,7 +16,7 @@ test("readableStreamToArrayBuffer works", async () => {
   try {
     const result = await Bun.readableStreamToArrayBuffer(
       new ReadableStream({
-        async start(controller) {
+        start(controller) {
           controller.enqueue(new TextEncoder().encode("bun is"));
           controller.enqueue(new TextEncoder().encode(" awesome!"));
           controller.close();

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

@@ -260,6 +260,117 @@ 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: an abrupt completion of the then call
+    // rejects the wrapper. Promise.prototype.then runs SpeciesConstructor,
+    // so a throwing @@species on the start()-returned promise must reject
+    // startPromise and error the stream. Subprocess-isolated so a pristine
+    // promiseThenWatchpointSet routes to promiseResolveThenableJobFastSlow
+    // (the JSC code under test). Upstream JSC bug:
+    // https://github.com/oven-sh/WebKit/pull/256
+    it.todo("start() returns a Promise whose @@species throws: the stream errors with the thrown value", async () => {
+      await using proc = Bun.spawn({
+        cmd: [
+          bunExe(),
+          "-e",
+          `const speciesError = new Error("species-boom");
+             const p = Promise.resolve();
+             p.constructor = { get [Symbol.species]() { throw speciesError; } };
+             let result = "pending";
+             new WritableStream({ start: () => p }).getWriter().closed.then(
+               () => result = "fulfilled",
+               e => result = e === speciesError ? "speciesError" : String(e),
+             );
+             for (let i = 0; i < 20; i++) await Promise.resolve();
+             console.log(result);`,
+        ],
+        env: bunEnv,
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({
+        stdout: "speciesError",
+        stderr: expect.any(String),
+        exitCode: 0,
+      });
+    });
+
+    // Resolve(x) does Get(x, "then"); a null-proto Promise has no .then, so
+    // it is treated as a non-thenable and the wrap fulfills with the promise
+    // object itself — [[started]] flips immediately 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", () => {
@@ -1317,3 +1428,41 @@ it("ReadableStream BYOB read pending at cancel() resolves with undefined", async
   expect(value).toBeUndefined();
   await reader.closed;
 });
+
+// Web IDL "a promise resolved with x" is NewPromiseCapability + Resolve(x);
+// Resolve(x) on a thenable does Get(x, "then") and queues a job to call it.
+// So when an underlying-source/sink callback returns a Promise, the wrap
+// calls Promise.prototype.then once — observable, per spec, like Node.
+// Subprocess-isolated: patching Promise.prototype.then permanently
+// invalidates JSC's promiseThenWatchpointSet for the process, which would
+// route every later test through the generic thenable path.
+it("wrapping an async stream callback result observes Promise.prototype.then (Web IDL 'a promise resolved with')", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const _then = Promise.prototype.then;
+       let counter = 0;
+       Promise.prototype.then = function (...args) { counter++; return _then.apply(this, args); };
+       new ReadableStream({ async start() {} });
+       new WritableStream({ async start() {} });
+       // The PromiseResolveThenableJob runs as a microtask; drain enough
+       // rounds for both assimilations to complete. await on a non-thenable
+       // doesn't reach the patched .then.
+       for (let i = 0; i < 20; i++) await 1;
+       Promise.prototype.then = _then;
+       console.log(counter);`,
+    ],
+    env: bunEnv,
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  // Exactly one assimilation per async start() result. A larger count would
+  // indicate a double-wrap regression (the FromUnderlyingSink change exists
+  // to prevent that).
+  expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({
+    stdout: "2",
+    stderr: expect.any(String),
+    exitCode: 0,
+  });
+});