diff --git a/src/webfetch/fetcher.ts b/src/webfetch/fetcher.ts index 8f878b5..effa74b 100644 --- a/src/webfetch/fetcher.ts +++ b/src/webfetch/fetcher.ts @@ -46,8 +46,9 @@ interface HttpResponse { } interface ResponseBodyStream extends AsyncIterable { + on?(event: "error", listener: (error: Error) => void): unknown; destroy(error?: Error): void; - dump(options?: { limit: number; signal?: AbortSignal }): Promise; + dump?(options?: { limit: number; signal?: AbortSignal }): Promise; } export async function fetchUrl(options: FetchOptions): Promise { @@ -208,15 +209,29 @@ function getHeader(headers: IncomingHttpHeaders, name: string): string { return value ?? ""; } -async function discardBody(body: ResponseBodyStream): Promise { +// Some runtimes resolve "undici" to builds whose response body lacks dump(). +// Discarding must stay best-effort: re-emitting an error on an unlistened +// stream crashes the host process. +export async function discardBody(body: ResponseBodyStream): Promise { try { - await body.dump({ limit: 1024 }); - } catch (error) { - if (error instanceof Error) { - body.destroy(error); + if (typeof body.dump === "function") { + await body.dump({ limit: 1024 }); return; } - throw error; + await drainBody(body); + } catch { + // Discard failures are not actionable for the caller. + } finally { + body.on?.("error", () => {}); + body.destroy(); + } +} + +async function drainBody(body: ResponseBodyStream): Promise { + let drained = 0; + for await (const chunk of body) { + drained += toUint8Array(chunk).length; + if (drained > MAX_RESPONSE_SIZE_BYTES) return; } } diff --git a/test/discard-body.test.ts b/test/discard-body.test.ts new file mode 100644 index 0000000..e59b7f9 --- /dev/null +++ b/test/discard-body.test.ts @@ -0,0 +1,59 @@ +import { Readable, type ReadableOptions } from "node:stream"; +import { describe, expect, it } from "vitest"; + +import { discardBody } from "../src/webfetch/fetcher.js"; + +type MinimalBody = { + destroy(error?: Error): void; + dump?(options?: { limit: number }): Promise; +}; + +function streamWithoutDump(chunks: string[], options?: ReadableOptions): Readable & MinimalBody { + const stream = new Readable(options ?? {}); + for (const chunk of chunks) { + stream.push(chunk); + } + stream.push(null); + return stream as Readable & MinimalBody; +} + +describe("discardBody", () => { + it("#given body without dump #when discarding #then resolves and drains the stream", async () => { + // given + const body = streamWithoutDump(["hello ", "world"]); + + // when + await expect(discardBody(body)).resolves.toBeUndefined(); + + // then + expect(body.readableEnded).toBe(true); + }); + + it("#given body without dump that errors mid-stream #when discarding #then resolves without surfacing the error", async () => { + // given + const body = streamWithoutDump(["partial"]); + const failure = new Error("stream exploded"); + queueMicrotask(() => body.destroy(failure)); + + // when + await expect(discardBody(body)).resolves.toBeUndefined(); + + // then + expect(body.destroyed).toBe(true); + }); + + it("#given dump-capable body #when discarding #then uses dump instead of manual drain", async () => { + // given + let dumped = false; + const body = streamWithoutDump(["ignored"]); + body.dump = async () => { + dumped = true; + }; + + // when + await discardBody(body); + + // then + expect(dumped).toBe(true); + }); +});