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)); diff --git a/packages/preview2-shim/src/browser/http.ts b/packages/preview2-shim/src/browser/http.ts index 191f0c865..386b21309 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,128 +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() { + async function waitForReadable(): Promise { + while (!ready()) { + startRead(); + await readPromise; + } + } + + function read(len: bigint): Uint8Array { if (readError) { - throw { tag: "last-operation-failed", val: 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); + } + 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(); + } + 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(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; - } - throw { tag: "would-block" }; - }, - 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; - } - 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" }; - }); - }, + 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(); + }); + } }, }); @@ -885,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; } 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/browser/http-input-stream.ts b/packages/preview2-shim/test/browser/http-input-stream.ts new file mode 100644 index 000000000..eb45b868c --- /dev/null +++ b/packages/preview2-shim/test/browser/http-input-stream.ts @@ -0,0 +1,290 @@ +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)); +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); +} + +async function openBody(response: Response): Promise { + vi.stubGlobal("fetch", async (): Promise => response); + 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 = 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"); + const responseResource = own(result.val.val); + const body = own(responseResource.consume()); + return own(body.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(() => { + 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])( + "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 = own(new types.OutgoingRequest(new types.Fields())); + request.setMethod({ tag: "get" }); + request.setScheme({ tag: "HTTPS" }); + request.setAuthority("example.com"); + request.setPathWithQuery("/"); + const future = own(outgoingHandler.handle(request, undefined)); + const pollable = own(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 = 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); + 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 = own(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 = own(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 = own(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 = own(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 = 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); + 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 = own(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); + }); +}); diff --git a/packages/preview2-shim/test/browser/http.ts b/packages/preview2-shim/test/browser/http.ts index 5354269d7..ebb5decc1 100644 --- a/packages/preview2-shim/test/browser/http.ts +++ b/packages/preview2-shim/test/browser/http.ts @@ -97,6 +97,65 @@ 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; + 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) { + // 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 = own(new types.OutgoingRequest(new types.Fields())); + request.setMethod({ tag: "get" }); + request.setScheme({ tag: "HTTPS" }); + request.setAuthority("example.com"); + request.setPathWithQuery("/"); + + 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 = 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 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); + } + } + }); + test("browser outgoing HTTP waits for the complete request body", async () => { const { _setRequestStreaming, outgoingHandler, types } = await import("../../src/browser/http.js"); 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"; + }, +};