diff --git a/src/js/builtins/StreamInternals.ts b/src/js/builtins/StreamInternals.ts index daf9b96569b9..2e2c102c26cd 100644 --- a/src/js/builtins/StreamInternals.ts +++ b/src/js/builtins/StreamInternals.ts @@ -31,9 +31,13 @@ export function markPromiseAsHandled(promise: Promise) { $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; } diff --git a/src/js/builtins/WritableStreamInternals.ts b/src/js/builtins/WritableStreamInternals.ts index 9fe4583c6407..8faaa527071c 100644 --- a/src/js/builtins/WritableStreamInternals.ts +++ b/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"]; diff --git a/test/js/bun/util/readablestreamtoarraybuffer.test.ts b/test/js/bun/util/readablestreamtoarraybuffer.test.ts index 7b15222bc3c3..f0f8cb7f972f 100644 --- a/test/js/bun/util/readablestreamtoarraybuffer.test.ts +++ b/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(); diff --git a/test/js/web/streams/streams.test.js b/test/js/web/streams/streams.test.js index 537be8d970d5..9310cc44ee73 100644 --- a/test/js/web/streams/streams.test.js +++ b/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, + }); +});