Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions src/webfetch/fetcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,9 @@ interface HttpResponse {
}

interface ResponseBodyStream extends AsyncIterable<unknown> {
on?(event: "error", listener: (error: Error) => void): unknown;
destroy(error?: Error): void;
dump(options?: { limit: number; signal?: AbortSignal }): Promise<void>;
dump?(options?: { limit: number; signal?: AbortSignal }): Promise<void>;
}

export async function fetchUrl(options: FetchOptions): Promise<FetchResult> {
Expand Down Expand Up @@ -208,15 +209,29 @@ function getHeader(headers: IncomingHttpHeaders, name: string): string {
return value ?? "";
}

async function discardBody(body: ResponseBodyStream): Promise<void> {
// 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<void> {
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<void> {
let drained = 0;
for await (const chunk of body) {
drained += toUint8Array(chunk).length;
if (drained > MAX_RESPONSE_SIZE_BYTES) return;
}
}

Expand Down
59 changes: 59 additions & 0 deletions test/discard-body.test.ts
Original file line number Diff line number Diff line change
@@ -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<void>;
};

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);
});
});