From d31c3dc8477da8d3040a63b050f249264bd940bc Mon Sep 17 00:00:00 2001 From: Eduardo Rodrigues <16357187+eduardomourar@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:57:27 +0100 Subject: [PATCH 1/8] chore(p2-shim): reproducing error in test --- packages/preview2-shim/test/browser/http.ts | 49 +++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/packages/preview2-shim/test/browser/http.ts b/packages/preview2-shim/test/browser/http.ts index 5354269d7..8f80c5ff7 100644 --- a/packages/preview2-shim/test/browser/http.ts +++ b/packages/preview2-shim/test/browser/http.ts @@ -97,6 +97,55 @@ suite("Browser HTTP", () => { assert.match(await response.text(), /internal-error.*rejected/); }); + test("browser incoming HTTP body read() returns empty instead of throwing before the first chunk arrives", async () => { + const { outgoingHandler, types } = await import("../../src/browser/http.js"); + const originalFetch = globalThis.fetch; + let releaseChunk: () => void; + const gate = new Promise((resolve) => { + releaseChunk = resolve; + }); + globalThis.fetch = async () => { + const stream = new ReadableStream({ + async start(controller) { + // Delay the first chunk so a non-blocking read() lands + // before any data is buffered - this is the exact + // condition that used to throw an invalid `would-block` + // stream-error and hang JSPI-driven callers forever. + await gate; + controller.enqueue(new TextEncoder().encode("delayed")); + controller.close(); + }, + }); + return new Response(stream, { status: 200 }); + }; + try { + const request = new types.OutgoingRequest(new types.Fields()); + request.setMethod({ tag: "get" }); + request.setScheme({ tag: "HTTPS" }); + request.setAuthority("example.com"); + request.setPathWithQuery("/"); + + const future = outgoingHandler.handle(request, undefined); + await future.subscribe().block(); + const result = future.get(); + if (!result || result.tag !== "ok" || result.val.tag !== "ok") { + throw new Error("expected an ok response result"); + } + const incomingResponse = result.val.val; + const bodyStream = incomingResponse.consume().stream(); + + const firstRead = bodyStream.read(64n); + assert.deepStrictEqual(firstRead, new Uint8Array(0)); + + releaseChunk!(); + await bodyStream.subscribe().block(); + const secondRead = bodyStream.read(64n); + assert.strictEqual(new TextDecoder().decode(secondRead), "delayed"); + } finally { + globalThis.fetch = originalFetch; + } + }); + test("browser outgoing HTTP waits for the complete request body", async () => { const { _setRequestStreaming, outgoingHandler, types } = await import("../../src/browser/http.js"); From ca1faef237de048afda669d2b3a181cd827da699 Mon Sep 17 00:00:00 2001 From: Eduardo Rodrigues <16357187+eduardomourar@users.noreply.github.com> Date: Sat, 5 Sep 2026 16:00:42 +0100 Subject: [PATCH 2/8] fix(p2-shim): throw invalid would-block from browser incoming-body read `wasi:io/streams` stream-error only has `closed` and `last-operation-failed` variants - there's no `would-block`. The browser IncomingBody's non-blocking read() threw `{ tag: "would-block" }` whenever no data was buffered yet, which JSPI-driven callers (e.g. wstd's Reactor loop) can't lower into a valid stream-error, silently swallowing the failure and hanging forever instead of surfacing an error. Return an empty array instead, per spec, and let the existing subscribe()/wait mechanism handle backpressure like it already does for blockingRead(). --- packages/preview2-shim/src/browser/http.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/packages/preview2-shim/src/browser/http.ts b/packages/preview2-shim/src/browser/http.ts index 191f0c865..de149ebec 100644 --- a/packages/preview2-shim/src/browser/http.ts +++ b/packages/preview2-shim/src/browser/http.ts @@ -558,7 +558,16 @@ class IncomingBody implements TypesNamespace.IncomingBody { } return slice; } - throw { tag: "would-block" }; + // No data buffered yet, but the stream isn't closed or errored. + // `would-block` is not a valid `stream-error` variant (only + // `closed` and `last-operation-failed` are) - a non-blocking + // read with nothing available yet must return an empty list, + // not an error. Throwing here caused hosts driving this via + // JSPI to hang forever on the very first read of a body whose + // first chunk hadn't arrived yet (e.g. a slower response, + // while a fast one worked by luck). + startRead(); + return new Uint8Array(0); }, blockingRead(len: bigint): any { checkReadError(); From 65bc23e237cacb7e94070597b2d2dc44a523503a Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 00:43:53 +0000 Subject: [PATCH 3/8] fix(p2-shim): preserve browser HTTP input stream semantics --- packages/preview2-shim/src/browser/http.ts | 208 ++++++++++----------- 1 file changed, 98 insertions(+), 110 deletions(-) diff --git a/packages/preview2-shim/src/browser/http.ts b/packages/preview2-shim/src/browser/http.ts index de149ebec..fa28bd19f 100644 --- a/packages/preview2-shim/src/browser/http.ts +++ b/packages/preview2-shim/src/browser/http.ts @@ -5,6 +5,8 @@ import type { } from "../../types/http.js"; import type { Error as IoError } from "../../types/interfaces/wasi-io-error.js"; import type { Pollable } from "../../types/interfaces/wasi-io-poll.js"; +import type { StreamError } from "../../types/interfaces/wasi-io-streams.js"; +import type { InputStreamHandler } from "./io.js"; import { inputStreamCreate, ioErrorCreate, outputStreamCreate, pollableCreate } from "./io.js"; export { InMemoryHttpClient } from "./in-memory-http.js"; @@ -467,7 +469,7 @@ delete OutgoingRequest._handle; class IncomingBody implements TypesNamespace.IncomingBody { #finished = false; - #stream: any = undefined; + #stream: ReturnType | null = null; stream() { if (!this.#stream) { @@ -496,137 +498,123 @@ class IncomingBody implements TypesNamespace.IncomingBody { let reader: ReadableStreamDefaultReader | null = null; let readPromise: Promise | null = null; let readError: IoError | null = null; + let disposed = false; - function ensureReader() { - if (!reader && fetchResponse.body) { - reader = fetchResponse.body.getReader(); - } + function ready(): boolean { + return done || (buffer !== null && bufferOffset < buffer.byteLength); } - function startRead() { - if (readPromise || done) { + function startRead(): void { + if (readPromise || ready()) { return; } - ensureReader(); - if (!reader) { + if (!fetchResponse.body) { done = true; return; } - readPromise = reader.read().then( - (result) => { - readPromise = null; - if (result.done) { - done = true; - } else { - buffer = result.value; - bufferOffset = 0; + reader ??= fetchResponse.body.getReader(); + const activeReader = reader; + readPromise = (async (): Promise => { + try { + // Empty Fetch chunks do not make a WASI input stream readable. + // Keep a single read in flight and buffer at most one nonempty chunk. + while (!done) { + const result = await activeReader.read(); + if (disposed) { + return; + } + if (result.done) { + done = true; + } else if (result.value.byteLength > 0) { + buffer = result.value; + bufferOffset = 0; + return; + } } - }, - (cause) => { - readPromise = null; + } catch (cause: unknown) { done = true; - readError = ioErrorCreate( - cause instanceof Error ? cause.message : String(cause), - ); - }, - ); + if (!disposed) { + readError = ioErrorCreate( + cause instanceof Error ? cause.message : String(cause), + ); + } + } finally { + readPromise = null; + if (done) { + activeReader.releaseLock(); + } + } + })(); } - function checkReadError() { - if (readError) { - throw { tag: "last-operation-failed", val: readError }; + async function waitForReadable(): Promise { + while (!ready()) { + startRead(); + await readPromise; } } - incomingBody.#stream = inputStreamCreate({ - read(len: bigint) { - checkReadError(); - if (done && (buffer === null || bufferOffset >= buffer.byteLength)) { - throw { tag: "closed" }; - } - if (buffer !== null && bufferOffset < buffer.byteLength) { - const available = buffer.byteLength - bufferOffset; - const toRead = Math.min(Number(len), available); - const slice = buffer.slice(bufferOffset, bufferOffset + toRead); - bufferOffset += toRead; - if (bufferOffset >= buffer.byteLength) { - buffer = null; - bufferOffset = 0; - if (!done) { - startRead(); - } - } - return slice; - } - // No data buffered yet, but the stream isn't closed or errored. - // `would-block` is not a valid `stream-error` variant (only - // `closed` and `last-operation-failed` are) - a non-blocking - // read with nothing available yet must return an empty list, - // not an error. Throwing here caused hosts driving this via - // JSPI to hang forever on the very first read of a body whose - // first chunk hadn't arrived yet (e.g. a slower response, - // while a fast one worked by luck). + function read(len: bigint): Uint8Array { + if (readError) { + const error = readError; + readError = null; + throw { tag: "last-operation-failed", val: error } satisfies StreamError; + } + if (done && (buffer === null || bufferOffset >= buffer.byteLength)) { + throw { tag: "closed" } satisfies StreamError; + } + if (buffer === null || bufferOffset >= buffer.byteLength) { startRead(); return new Uint8Array(0); - }, - blockingRead(len: bigint): any { - checkReadError(); - if (done && (buffer === null || bufferOffset >= buffer.byteLength)) { - throw { tag: "closed" }; - } - if (buffer !== null && bufferOffset < buffer.byteLength) { - const available = buffer.byteLength - bufferOffset; - const toRead = Math.min(Number(len), available); - const slice = buffer.slice(bufferOffset, bufferOffset + toRead); - bufferOffset += toRead; - if (bufferOffset >= buffer.byteLength) { - buffer = null; - bufferOffset = 0; - if (!done) { - startRead(); - } - } - return slice; - } + } + const toRead = Math.min(Number(len), buffer.byteLength - bufferOffset); + const slice = buffer.slice(bufferOffset, bufferOffset + toRead); + bufferOffset += toRead; + if (bufferOffset >= buffer.byteLength) { + buffer = null; + bufferOffset = 0; startRead(); - const waitFor = readPromise || Promise.resolve(); - return waitFor.then(() => { - checkReadError(); - if (done && (buffer === null || bufferOffset >= buffer.byteLength)) { - throw { tag: "closed" }; - } - if (buffer !== null && bufferOffset < buffer.byteLength) { - const available = buffer.byteLength - bufferOffset; - const toRead = Math.min(Number(len), available); - const slice = buffer.slice(bufferOffset, bufferOffset + toRead); - bufferOffset += toRead; - if (bufferOffset >= buffer.byteLength) { - buffer = null; - bufferOffset = 0; - if (!done) { - startRead(); - } - } - return slice; - } - throw { tag: "closed" }; - }); - }, + } + return slice; + } + + function blockingRead(len: bigint): Uint8Array | Promise { + if (len === 0n || ready()) { + return read(len); + } + return waitForReadable().then(() => read(len)); + } + + function blockingSkip(len: bigint): bigint | Promise { + const result = blockingRead(len); + return result instanceof Promise + ? result.then((bytes) => BigInt(bytes.byteLength)) + : BigInt(result.byteLength); + } + + incomingBody.#stream = inputStreamCreate({ + read, + // WIT declares synchronous signatures; JSPI awaits these host results + // for blocking imports. Preserve synchronous results for buffered bodies. + blockingRead: blockingRead as InputStreamHandler["blockingRead"], + blockingSkip: blockingSkip as InputStreamHandler["blockingSkip"], subscribe() { - return pollableCreate({ - ready: () => - readError !== null || - done || - (buffer !== null && bufferOffset < buffer.byteLength), - wait: () => { - startRead(); - return readPromise ?? Promise.resolve(); - }, - }); + return pollableCreate({ ready, wait: waitForReadable }); }, drop() { + disposed = true; done = true; - void reader?.cancel(); + buffer = null; + readError = null; + if (reader) { + const activeReader = reader; + void activeReader + .cancel() + .catch(() => {}) + .finally(() => { + activeReader.releaseLock(); + }); + } }, }); From 18c654341d504fd83897d7484d1dddaefae17fc6 Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 00:43:58 +0000 Subject: [PATCH 4/8] fix(p2-shim): retain HTTP bodies after response future disposal --- packages/preview2-shim/src/browser/http.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/preview2-shim/src/browser/http.ts b/packages/preview2-shim/src/browser/http.ts index fa28bd19f..386b21309 100644 --- a/packages/preview2-shim/src/browser/http.ts +++ b/packages/preview2-shim/src/browser/http.ts @@ -882,6 +882,11 @@ class FutureIncomingResponse implements TypesNamespace.FutureIncomingResponse { } const result = this.#result; this.#result = { tag: "err" }; + // The returned response now owns the body. Dropping this consumed future + // must not abort a Fetch body that the guest is still streaming. + if (result.tag === "ok" && result.val.tag === "ok") { + this.#controller = null; + } return result; } From 8bbe4fe1bfb2f48afde5ca11092ea249603188f8 Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 00:44:06 +0000 Subject: [PATCH 5/8] test(p2-shim): cover browser HTTP stream readiness and lifecycles --- .../test/browser/http-input-stream.ts | 272 ++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 packages/preview2-shim/test/browser/http-input-stream.ts diff --git a/packages/preview2-shim/test/browser/http-input-stream.ts b/packages/preview2-shim/test/browser/http-input-stream.ts new file mode 100644 index 000000000..eab239dbf --- /dev/null +++ b/packages/preview2-shim/test/browser/http-input-stream.ts @@ -0,0 +1,272 @@ +import assert from "node:assert/strict"; + +import { afterEach, suite, test, vi } from "vitest"; + +import { outgoingHandler, types } from "../../src/browser/http.js"; +import type { InputStream } from "../../types/interfaces/wasi-io-streams.js"; + +const encode = (text: string): Uint8Array => new TextEncoder().encode(text); +const turn = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); + +function dispose(resource: object): void { + const method: unknown = Reflect.get(resource, Symbol.dispose || Symbol.for("dispose")); + assert.ok(typeof method === "function"); + method.call(resource); +} + +async function openBody(response: Response): Promise { + vi.stubGlobal("fetch", async (): Promise => response); + const request = new types.OutgoingRequest(new types.Fields()); + request.setMethod({ tag: "get" }); + request.setScheme({ tag: "HTTPS" }); + request.setAuthority("example.com"); + request.setPathWithQuery("/"); + const future = outgoingHandler.handle(request, undefined); + const pollable = future.subscribe(); + await pollable.block(); + dispose(pollable); + const result = future.get(); + assert.ok(result?.tag === "ok" && result.val.tag === "ok"); + return result.val.val.consume().stream(); +} + +async function controlledBody(cancel?: () => void | Promise): Promise<{ + stream: InputStream; + controller: ReadableStreamDefaultController; + source: ReadableStream; +}> { + let controller!: ReadableStreamDefaultController; + const source = new ReadableStream( + { + start: (value) => { + controller = value; + }, + cancel, + }, + { highWaterMark: 0 }, + ); + return { stream: await openBody(new Response(source)), controller, source }; +} + +afterEach(() => vi.unstubAllGlobals()); + +suite("Browser HTTP input stream", () => { + test.each([true, false])( + "only unconsumed response futures abort Fetch on drop (consumed=%s)", + async (consumed) => { + let signal: AbortSignal | null | undefined; + vi.stubGlobal( + "fetch", + async (_input: RequestInfo | URL, init?: RequestInit): Promise => { + signal = init?.signal; + return new Response(null); + }, + ); + const request = new types.OutgoingRequest(new types.Fields()); + request.setMethod({ tag: "get" }); + request.setScheme({ tag: "HTTPS" }); + request.setAuthority("example.com"); + request.setPathWithQuery("/"); + const future = outgoingHandler.handle(request, undefined); + const pollable = future.subscribe(); + await pollable.block(); + dispose(pollable); + assert.ok(signal); + assert.strictEqual(signal.aborted, false); + if (consumed) { + assert.ok(future.get()); + } + dispose(future); + assert.strictEqual(signal.aborted, !consumed); + }, + ); + + test("read and skip return empty across gaps and pollables remain level-triggered", async () => { + const { stream, controller } = await controlledBody(); + const first = stream.subscribe(); + const second = stream.subscribe(); + assert.deepStrictEqual(stream.read(4n), new Uint8Array(0)); + assert.strictEqual(stream.skip(4n), 0n); + assert.strictEqual(first.ready(), false); + controller.enqueue(encode("abcd")); + await Promise.all([first.block(), second.block()]); + assert.strictEqual(first.ready(), true); + assert.strictEqual(second.ready(), true); + assert.deepStrictEqual(stream.read(1n), encode("a")); + assert.strictEqual(first.ready(), true); + assert.strictEqual(stream.skip(2n), 2n); + assert.deepStrictEqual(stream.read(10n), encode("d")); + assert.strictEqual(first.ready(), false); + assert.strictEqual(second.ready(), false); + assert.deepStrictEqual(stream.read(10n), new Uint8Array(0)); + controller.enqueue(encode("ef")); + await first.block(); + assert.deepStrictEqual(stream.read(10n), encode("ef")); + controller.close(); + await second.block(); + assert.throws(() => stream.read(1n), { tag: "closed" }); + assert.throws(() => stream.skip(1n), { tag: "closed" }); + dispose(first); + dispose(second); + dispose(stream); + }); + + test("zero-length operations complete synchronously without consuming data", async () => { + const { stream, controller } = await controlledBody(); + assert.deepStrictEqual(stream.read(0n), new Uint8Array(0)); + assert.deepStrictEqual(stream.blockingRead(0n), new Uint8Array(0)); + assert.strictEqual(stream.skip(0n), 0n); + assert.strictEqual(stream.blockingSkip(0n), 0n); + controller.enqueue(encode("x")); + const pollable = stream.subscribe(); + await pollable.block(); + assert.deepStrictEqual(stream.blockingRead(0n), new Uint8Array(0)); + assert.strictEqual(pollable.ready(), true); + assert.deepStrictEqual(stream.read(1n), encode("x")); + controller.close(); + await pollable.block(); + assert.throws(() => stream.read(0n), { tag: "closed" }); + assert.throws(() => stream.blockingRead(0n), { tag: "closed" }); + dispose(pollable); + dispose(stream); + }); + + test("polling ignores empty Fetch chunks until bytes or EOF arrive", async () => { + const { stream, controller } = await controlledBody(); + const pollable = stream.subscribe(); + let settled = false; + const wait = Promise.resolve(pollable.block()).then(() => { + settled = true; + }); + controller.enqueue(new Uint8Array(0)); + controller.enqueue(new Uint8Array(0)); + await turn(); + assert.strictEqual(settled, false); + assert.strictEqual(pollable.ready(), false); + assert.deepStrictEqual(stream.read(2n), new Uint8Array(0)); + controller.enqueue(encode("ok")); + await wait; + assert.deepStrictEqual(stream.read(2n), encode("ok")); + controller.enqueue(new Uint8Array(0)); + controller.close(); + await pollable.block(); + assert.throws(() => stream.read(1n), { tag: "closed" }); + dispose(pollable); + dispose(stream); + }); + + test("blocking reads and skips wait across empty chunks", async () => { + const { stream, controller } = await controlledBody(); + let settled = false; + const read = Promise.resolve(stream.blockingRead(2n)).then((bytes) => { + settled = true; + return bytes; + }); + controller.enqueue(new Uint8Array(0)); + await turn(); + assert.strictEqual(settled, false); + controller.enqueue(encode("abc")); + assert.deepStrictEqual(await read, encode("ab")); + assert.strictEqual(stream.blockingSkip(1n), 1n); + const skip = stream.blockingSkip(2n); + controller.enqueue(new Uint8Array(0)); + controller.enqueue(encode("def")); + assert.strictEqual(await skip, 2n); + assert.deepStrictEqual(stream.read(10n), encode("f")); + const closed = Promise.resolve(stream.blockingRead(1n)); + controller.close(); + await assert.rejects(closed, { tag: "closed" }); + dispose(stream); + }); + + test.each([null, new Uint8Array(0)])("empty response bodies reach EOF (%s)", async (body) => { + const stream = await openBody(new Response(body)); + const pollable = stream.subscribe(); + await pollable.block(); + assert.strictEqual(pollable.ready(), true); + assert.throws(() => stream.read(1n), { tag: "closed" }); + dispose(pollable); + dispose(stream); + }); + + test("read errors wake pollers, expose an IO error once, and then close", async () => { + const { stream, controller, source } = await controlledBody(); + const pollable = stream.subscribe(); + const wait = pollable.block(); + controller.error(new Error("body failed")); + await wait; + assert.strictEqual(pollable.ready(), true); + assert.throws( + () => stream.read(1n), + (error: unknown): boolean => { + assert.ok( + typeof error === "object" && error !== null && "tag" in error && "val" in error, + ); + assert.strictEqual(error.tag, "last-operation-failed"); + assert.ok(error.val instanceof Error); + assert.strictEqual(error.val.message, "body failed"); + return true; + }, + ); + assert.throws(() => stream.blockingRead(1n), { tag: "closed" }); + assert.strictEqual(source.locked, false); + dispose(pollable); + dispose(stream); + }); + + test("pending blocking reads reject with a valid stream error", async () => { + const { stream, controller } = await controlledBody(); + const read = Promise.resolve(stream.blockingRead(1n)); + controller.error("broken"); + await assert.rejects(read, { tag: "last-operation-failed" }); + assert.throws(() => stream.read(1n), { tag: "closed" }); + dispose(stream); + }); + + test("dropping a pending stream cancels once and handles cancellation rejection", async () => { + const cancel = vi.fn(async (): Promise => { + throw new Error("cancel failed"); + }); + const { stream, source } = await controlledBody(cancel); + const pollable = stream.subscribe(); + const wait = assert.rejects(Promise.resolve(pollable.block()), /disposed/); + const read = assert.rejects(Promise.resolve(stream.blockingRead(1n)), { tag: "closed" }); + dispose(stream); + dispose(stream); + await Promise.all([wait, read]); + await turn(); + assert.strictEqual(cancel.mock.calls.length, 1); + assert.strictEqual(source.locked, false); + dispose(pollable); + }); + + test("prefetch never overwrites partial data or buffers the complete response", async () => { + let pulls = 0; + const stream = await openBody( + new Response( + new ReadableStream( + { + pull(controller): void { + pulls++; + controller.enqueue(encode("abc")); + }, + }, + { highWaterMark: 0 }, + ), + ), + ); + const pollable = stream.subscribe(); + await pollable.block(); + await turn(); + assert.strictEqual(pulls, 1); + assert.deepStrictEqual(stream.read(1n), encode("a")); + await pollable.block(); + await turn(); + assert.strictEqual(pulls, 1); + assert.deepStrictEqual(stream.read(2n), encode("bc")); + await pollable.block(); + assert.strictEqual(pulls, 2); + dispose(pollable); + dispose(stream); + }); +}); From 96d45a0011bff30d6d966bcb965b140199467a87 Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 00:44:19 +0000 Subject: [PATCH 6/8] test(p2-shim): regress delayed browser HTTP reads through JSPI --- .../test/browser/http-input-stream-e2e.ts | 122 ++++++++++++++++++ .../browser/http-input-stream/component.js | 106 +++++++++++++++ 2 files changed, 228 insertions(+) create mode 100644 packages/preview2-shim/test/browser/http-input-stream-e2e.ts create mode 100644 packages/preview2-shim/test/fixtures/browser/http-input-stream/component.js diff --git a/packages/preview2-shim/test/browser/http-input-stream-e2e.ts b/packages/preview2-shim/test/browser/http-input-stream-e2e.ts new file mode 100644 index 000000000..acc4c4e56 --- /dev/null +++ b/packages/preview2-shim/test/browser/http-input-stream-e2e.ts @@ -0,0 +1,122 @@ +import assert from "node:assert/strict"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { createServer } from "node:http"; +import type { ServerResponse } from "node:http"; +import { dirname, join } from "node:path"; + +import { componentize } from "@bytecodealliance/componentize-js"; +import { transpile } from "@bytecodealliance/jco"; +import { test } from "vitest"; + +import { + FIXTURES_WIT_DIR, + getTmpDir, + runBasicHarnessPageTest, + startTestServer, +} from "../common.js"; + +test("browser HTTP input stream crosses JSPI with delayed chunks and errors", async () => { + let body: ServerResponse | undefined; + let broken: ServerResponse | undefined; + const requests: string[] = []; + const api = createServer((request, response): void => { + requests.push(request.url ?? ""); + response.setHeader("access-control-allow-origin", "*"); + if (request.url === "/body" || request.url === "/broken") { + response.writeHead(200, { "content-type": "application/octet-stream" }); + response.flushHeaders(); + if (request.url === "/body") { + body = response; + } else { + broken = response; + } + return; + } + // Explicit guest-controlled gates, not timers: headers must be available + // before body bytes, and /second cannot arrive until the first chunk drains. + switch (request.url) { + case "/first": + body?.write("abcd"); + break; + case "/second": + body?.write("ef"); + break; + case "/finish": + body?.end(); + break; + case "/abort": + broken?.destroy(); + break; + default: + response.writeHead(404).end(); + return; + } + response.writeHead(204).end(); + }); + await new Promise((resolve, reject) => { + api.once("error", reject); + api.listen(0, "127.0.0.1", resolve); + }); + let harness: Awaited> | undefined; + try { + const address = api.address(); + assert.ok(address && typeof address !== "string"); + const outDir = await getTmpDir(); + harness = await startTestServer({ transpiledOutputDir: outDir }); + const source = await readFile( + new URL("../fixtures/browser/http-input-stream/component.js", import.meta.url), + "utf8", + ); + const sourcePath = join(outDir, "source.js"); + await writeFile(sourcePath, source.replace("TEST_AUTHORITY", `127.0.0.1:${address.port}`)); + const { component } = await componentize({ + sourcePath, + witPath: FIXTURES_WIT_DIR, + worldName: "browser-http-fetch", + }); + const { files } = await transpile(component, { + name: "component", + optimize: false, + asyncMode: "jspi", + // read and skip deliberately remain synchronous, as the WASI contract + // requires. Only polling and blocking operations may suspend. + asyncImports: [ + "wasi:io/poll#[method]pollable.block", + "wasi:io/poll#poll", + "wasi:io/streams#[method]input-stream.blocking-read", + "wasi:io/streams#[method]input-stream.blocking-skip", + ], + asyncExports: ["tests:p2-shim/test#run"], + outDir, + }); + for (const [path, bytes] of Object.entries(files)) { + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, bytes); + } + const { statusJSON } = await runBasicHarnessPageTest({ + browser: harness.browser, + url: `${harness.baseURL}/index.html#transpiled:component.js`, + }); + assert.strictEqual(statusJSON.status, "success"); + assert.strictEqual( + statusJSON.msg, + "empty reads, streaming, polling, EOF, and IO errors passed", + ); + assert.deepStrictEqual(requests, [ + "/body", + "/first", + "/second", + "/finish", + "/broken", + "/abort", + ]); + } finally { + api.closeAllConnections(); + await Promise.all([ + harness?.cleanup(), + new Promise((resolve, reject) => + api.close((error) => (error ? reject(error) : resolve())), + ), + ]); + } +}, 120_000); diff --git a/packages/preview2-shim/test/fixtures/browser/http-input-stream/component.js b/packages/preview2-shim/test/fixtures/browser/http-input-stream/component.js new file mode 100644 index 000000000..43b18eb19 --- /dev/null +++ b/packages/preview2-shim/test/fixtures/browser/http-input-stream/component.js @@ -0,0 +1,106 @@ +import { Fields, OutgoingRequest, IncomingBody } from "wasi:http/types@0.2.8"; +import { handle } from "wasi:http/outgoing-handler@0.2.8"; +import { poll } from "wasi:io/poll@0.2.8"; + +const authority = "TEST_AUTHORITY"; +const dispose = (resource) => resource[Symbol.dispose || Symbol.for("dispose")](); + +function check(condition, message) { + if (!condition) { + throw message; + } +} + +function request(path) { + const headers = Fields.fromList([]); + const outgoing = new OutgoingRequest(headers); + outgoing.setMethod({ tag: "get" }); + outgoing.setScheme({ tag: "HTTP" }); + outgoing.setAuthority(authority); + outgoing.setPathWithQuery(path); + const future = handle(outgoing, undefined); + const ready = future.subscribe(); + ready.block(); + dispose(ready); + const result = future.get(); + check(result?.tag === "ok" && result.val.tag === "ok", `request failed: ${path}`); + dispose(future); + // headers and outgoing were transferred into their consuming WASI calls. + return result.val.val; +} + +function control(path) { + const response = request(path); + check(response.status() === 204, `control failed: ${path}`); + dispose(response); +} + +function expectError(operation, tag) { + try { + operation(); + } catch (error) { + const value = error.payload ?? error; + check(value.tag === tag, `expected ${tag}, received ${value.tag}`); + return value; + } + throw `expected ${tag}`; +} + +function finish(response, body, stream, ready) { + dispose(ready); + dispose(stream); + const trailers = IncomingBody.finish(body); + dispose(trailers); + dispose(response); +} + +export const test = { + run() { + const response = request("/body"); + check(response.status() === 200, "missing response headers"); + const body = response.consume(); + const stream = body.stream(); + const ready = stream.subscribe(); + + // The server will not send bytes until this guest requests /first. + // This must cross the non-suspending canonical ABI as ok([]), not throw. + check(stream.read(64n).length === 0, "initial read must be empty"); + check(stream.skip(64n) === 0n, "initial skip must be empty"); + check(stream.blockingRead(0n).length === 0, "zero read must not wait"); + check(stream.blockingSkip(0n) === 0n, "zero skip must not wait"); + check(!ready.ready(), "body ready before first chunk"); + control("/first"); + check(poll([ready])[0] === 0, "first poll did not select body"); + + let first = ""; + while (first.length < 4) { + first += new TextDecoder().decode(stream.blockingRead(BigInt(4 - first.length))); + } + check(first === "abcd", "first chunk corrupted"); + check(stream.read(64n).length === 0, "inter-chunk read must be empty"); + check(!ready.ready(), "drained body still ready"); + + control("/second"); + check(stream.blockingSkip(1n) === 1n, "blocking skip failed"); + check(new TextDecoder().decode(stream.blockingRead(1n)) === "f", "second chunk corrupted"); + control("/finish"); + ready.block(); + expectError(() => stream.read(1n), "closed"); + expectError(() => stream.blockingRead(0n), "closed"); + finish(response, body, stream, ready); + + const brokenResponse = request("/broken"); + const brokenBody = brokenResponse.consume(); + const brokenStream = brokenBody.stream(); + const brokenReady = brokenStream.subscribe(); + check(brokenStream.read(1n).length === 0, "broken body must initially be empty"); + control("/abort"); + brokenReady.block(); + const failure = expectError(() => brokenStream.read(1n), "last-operation-failed"); + check(typeof failure.val.toDebugString() === "string", "missing IO error resource"); + dispose(failure.val); + expectError(() => brokenStream.read(1n), "closed"); + finish(brokenResponse, brokenBody, brokenStream, brokenReady); + return "empty reads, streaming, polling, EOF, and IO errors passed"; + }, +}; From ced83ab6c680dddd41a5ab81e46a67e97b840cd3 Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 06:18:53 +0000 Subject: [PATCH 7/8] test(p2-shim): clean up browser HTTP regressions on failure --- .../test/browser/http-input-stream.ts | 50 +++++++++++++------ packages/preview2-shim/test/browser/http.ts | 32 ++++++++---- 2 files changed, 55 insertions(+), 27 deletions(-) diff --git a/packages/preview2-shim/test/browser/http-input-stream.ts b/packages/preview2-shim/test/browser/http-input-stream.ts index eab239dbf..eb45b868c 100644 --- a/packages/preview2-shim/test/browser/http-input-stream.ts +++ b/packages/preview2-shim/test/browser/http-input-stream.ts @@ -7,8 +7,15 @@ import type { InputStream } from "../../types/interfaces/wasi-io-streams.js"; const encode = (text: string): Uint8Array => new TextEncoder().encode(text); const turn = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); +const resources = new Set(); + +function own(resource: T): T { + resources.add(resource); + return resource; +} function dispose(resource: object): void { + resources.delete(resource); const method: unknown = Reflect.get(resource, Symbol.dispose || Symbol.for("dispose")); assert.ok(typeof method === "function"); method.call(resource); @@ -16,18 +23,20 @@ function dispose(resource: object): void { async function openBody(response: Response): Promise { vi.stubGlobal("fetch", async (): Promise => response); - const request = new types.OutgoingRequest(new types.Fields()); + const request = own(new types.OutgoingRequest(new types.Fields())); request.setMethod({ tag: "get" }); request.setScheme({ tag: "HTTPS" }); request.setAuthority("example.com"); request.setPathWithQuery("/"); - const future = outgoingHandler.handle(request, undefined); - const pollable = future.subscribe(); + const future = own(outgoingHandler.handle(request, undefined)); + const pollable = own(future.subscribe()); await pollable.block(); dispose(pollable); const result = future.get(); assert.ok(result?.tag === "ok" && result.val.tag === "ok"); - return result.val.val.consume().stream(); + const responseResource = own(result.val.val); + const body = own(responseResource.consume()); + return own(body.stream()); } async function controlledBody(cancel?: () => void | Promise): Promise<{ @@ -48,7 +57,16 @@ async function controlledBody(cancel?: () => void | Promise): Promise<{ return { stream: await openBody(new Response(source)), controller, source }; } -afterEach(() => vi.unstubAllGlobals()); +afterEach(() => { + try { + // Also release pending readers and pollables when an assertion fails. + for (const resource of [...resources].reverse()) { + dispose(resource); + } + } finally { + vi.unstubAllGlobals(); + } +}); suite("Browser HTTP input stream", () => { test.each([true, false])( @@ -62,13 +80,13 @@ suite("Browser HTTP input stream", () => { return new Response(null); }, ); - const request = new types.OutgoingRequest(new types.Fields()); + const request = own(new types.OutgoingRequest(new types.Fields())); request.setMethod({ tag: "get" }); request.setScheme({ tag: "HTTPS" }); request.setAuthority("example.com"); request.setPathWithQuery("/"); - const future = outgoingHandler.handle(request, undefined); - const pollable = future.subscribe(); + const future = own(outgoingHandler.handle(request, undefined)); + const pollable = own(future.subscribe()); await pollable.block(); dispose(pollable); assert.ok(signal); @@ -83,8 +101,8 @@ suite("Browser HTTP input stream", () => { test("read and skip return empty across gaps and pollables remain level-triggered", async () => { const { stream, controller } = await controlledBody(); - const first = stream.subscribe(); - const second = stream.subscribe(); + const first = own(stream.subscribe()); + const second = own(stream.subscribe()); assert.deepStrictEqual(stream.read(4n), new Uint8Array(0)); assert.strictEqual(stream.skip(4n), 0n); assert.strictEqual(first.ready(), false); @@ -118,7 +136,7 @@ suite("Browser HTTP input stream", () => { assert.strictEqual(stream.skip(0n), 0n); assert.strictEqual(stream.blockingSkip(0n), 0n); controller.enqueue(encode("x")); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); await pollable.block(); assert.deepStrictEqual(stream.blockingRead(0n), new Uint8Array(0)); assert.strictEqual(pollable.ready(), true); @@ -133,7 +151,7 @@ suite("Browser HTTP input stream", () => { test("polling ignores empty Fetch chunks until bytes or EOF arrive", async () => { const { stream, controller } = await controlledBody(); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); let settled = false; const wait = Promise.resolve(pollable.block()).then(() => { settled = true; @@ -181,7 +199,7 @@ suite("Browser HTTP input stream", () => { test.each([null, new Uint8Array(0)])("empty response bodies reach EOF (%s)", async (body) => { const stream = await openBody(new Response(body)); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); await pollable.block(); assert.strictEqual(pollable.ready(), true); assert.throws(() => stream.read(1n), { tag: "closed" }); @@ -191,7 +209,7 @@ suite("Browser HTTP input stream", () => { test("read errors wake pollers, expose an IO error once, and then close", async () => { const { stream, controller, source } = await controlledBody(); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); const wait = pollable.block(); controller.error(new Error("body failed")); await wait; @@ -228,7 +246,7 @@ suite("Browser HTTP input stream", () => { throw new Error("cancel failed"); }); const { stream, source } = await controlledBody(cancel); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); const wait = assert.rejects(Promise.resolve(pollable.block()), /disposed/); const read = assert.rejects(Promise.resolve(stream.blockingRead(1n)), { tag: "closed" }); dispose(stream); @@ -255,7 +273,7 @@ suite("Browser HTTP input stream", () => { ), ), ); - const pollable = stream.subscribe(); + const pollable = own(stream.subscribe()); await pollable.block(); await turn(); assert.strictEqual(pulls, 1); diff --git a/packages/preview2-shim/test/browser/http.ts b/packages/preview2-shim/test/browser/http.ts index 8f80c5ff7..ebb5decc1 100644 --- a/packages/preview2-shim/test/browser/http.ts +++ b/packages/preview2-shim/test/browser/http.ts @@ -100,10 +100,12 @@ suite("Browser HTTP", () => { test("browser incoming HTTP body read() returns empty instead of throwing before the first chunk arrives", async () => { const { outgoingHandler, types } = await import("../../src/browser/http.js"); const originalFetch = globalThis.fetch; - let releaseChunk: () => void; - const gate = new Promise((resolve) => { - releaseChunk = resolve; - }); + const { promise: gate, resolve: releaseChunk } = Promise.withResolvers(); + const resources: object[] = []; + const own = (resource: T): T => { + resources.push(resource); + return resource; + }; globalThis.fetch = async () => { const stream = new ReadableStream({ async start(controller) { @@ -119,30 +121,38 @@ suite("Browser HTTP", () => { return new Response(stream, { status: 200 }); }; try { - const request = new types.OutgoingRequest(new types.Fields()); + const request = own(new types.OutgoingRequest(new types.Fields())); request.setMethod({ tag: "get" }); request.setScheme({ tag: "HTTPS" }); request.setAuthority("example.com"); request.setPathWithQuery("/"); - const future = outgoingHandler.handle(request, undefined); - await future.subscribe().block(); + const future = own(outgoingHandler.handle(request, undefined)); + await own(future.subscribe()).block(); const result = future.get(); if (!result || result.tag !== "ok" || result.val.tag !== "ok") { throw new Error("expected an ok response result"); } - const incomingResponse = result.val.val; - const bodyStream = incomingResponse.consume().stream(); + const incomingResponse = own(result.val.val); + const body = own(incomingResponse.consume()); + const bodyStream = own(body.stream()); const firstRead = bodyStream.read(64n); assert.deepStrictEqual(firstRead, new Uint8Array(0)); - releaseChunk!(); - await bodyStream.subscribe().block(); + releaseChunk(); + await own(bodyStream.subscribe()).block(); const secondRead = bodyStream.read(64n); assert.strictEqual(new TextDecoder().decode(secondRead), "delayed"); } finally { globalThis.fetch = originalFetch; + // Unblock start() even when an assertion fails, before cancelling its reader. + releaseChunk(); + await gate; + for (const resource of resources.reverse()) { + const dispose = Reflect.get(resource, Symbol.dispose || Symbol.for("dispose")); + dispose.call(resource); + } } }); From d893edd45ce2032f57dfca0fc63c2b215c2f224a Mon Sep 17 00:00:00 2001 From: Victor Adossi Date: Sun, 6 Sep 2026 08:46:49 +0000 Subject: [PATCH 8/8] test(transpile): relax async export call overhead limit in CI --- packages/jco-transpile/test/perf.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/jco-transpile/test/perf.ts b/packages/jco-transpile/test/perf.ts index b4e5f7736..0c4543b6a 100644 --- a/packages/jco-transpile/test/perf.ts +++ b/packages/jco-transpile/test/perf.ts @@ -16,7 +16,7 @@ const ASYNC_G2G_CALL_LIMIT_NS = env.CI ? 70_000_000 : 40_000_000; // test was introduced. These should narrow as call overhead improves. const WASM_MODULE_CALL_OVERHEAD_RATIO_LIMIT = env.CI ? 12 : 10; const SYNC_COMPONENT_CALL_OVERHEAD_RATIO_LIMIT = env.CI ? 1_500 : 1_000; -const ASYNC_COMPONENT_CALL_OVERHEAD_RATIO_LIMIT = env.CI ? 40_000 : 8_000; +const ASYNC_COMPONENT_CALL_OVERHEAD_RATIO_LIMIT = env.CI ? 60_000 : 8_000; const ADDER_COMPONENT_PATH = fileURLToPath(new URL('./fixtures/components/adder.component.wasm', import.meta.url)); const ADDER_MODULE_PATH = fileURLToPath(new URL('./fixtures/runtime/adder.wat', import.meta.url));