diff --git a/src/platform/compat/http/pinned-fetch.test.ts b/src/platform/compat/http/pinned-fetch.test.ts index 88b15ccdfe..f51b5ccb05 100644 --- a/src/platform/compat/http/pinned-fetch.test.ts +++ b/src/platform/compat/http/pinned-fetch.test.ts @@ -6,6 +6,9 @@ import { createPinnedFetchResponse, DEFAULT_OUTBOUND_USER_AGENT, fetchWithPinnedAddresses, + isReplayableRequestBody, + isRetriableConnectFailure, + planPinnedConnectAttempts, } from "./pinned-fetch.ts"; describe("fetchWithPinnedAddresses", () => { @@ -200,6 +203,42 @@ describe("fetchWithPinnedAddresses", () => { } }); + it("falls through to the next validated address when the first refuses", async () => { + if (!isNode) return; + + const { createServer } = await import("node:http"); + const server = createServer((_request, response) => { + response.writeHead(200, { "content-type": "text/plain" }); + response.end("reached"); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + + try { + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Node test server did not expose a TCP address"); + } + // 127.0.0.2 is loopback with nothing listening, so it refuses fast. Only + // the second validated address serves. autoSelectFamily races IPv4 + // against IPv6 and does nothing for two addresses of the same family, so + // the transport has to walk the list itself. + const response = await fetchWithPinnedAddresses( + new URL(`http://localhost:${address.port}/resource`), + ["127.0.0.2", "127.0.0.1"], + { method: "GET" }, + ); + assertEquals(response.status, 200); + assertEquals(await response.text(), "reached"); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + }); + } + }); + it("returns a null body for HEAD through the native Node transport", async () => { if (!isNode) return; @@ -240,3 +279,65 @@ describe("fetchWithPinnedAddresses", () => { } }); }); + +describe("pinned connect attempts", () => { + it("keeps a single validated address as one attempt", () => { + assertEquals(planPinnedConnectAttempts(["203.0.113.7"]), [["203.0.113.7"]]); + }); + + it("dials one address per attempt", () => { + // Every attempt dials one address; the runtime is never asked to choose. + const plan = planPinnedConnectAttempts(["2606:4700::1", "104.26.14.209"]); + assertEquals(plan[0], ["2606:4700::1"]); + assertEquals(plan.length, 2); + assertEquals(plan[1], ["104.26.14.209"]); + }); + + it("tries the other address family before a sibling of the failed one", () => { + const plan = planPinnedConnectAttempts([ + "2606:4700::1", + "2606:4700::2", + "104.26.14.209", + ]); + // A host with no IPv6 route fails on both AAAA records, so the A record has + // to come before the second AAAA. + assertEquals(plan[1], ["104.26.14.209"]); + assertEquals(plan[2], ["2606:4700::2"]); + }); + + it("retries only connect-level failures", () => { + assertEquals(isRetriableConnectFailure({ code: "ECONNREFUSED" }), true); + assertEquals(isRetriableConnectFailure({ code: "ENETUNREACH" }), true); + assertEquals(isRetriableConnectFailure({ code: "ECONNRESET" }), false); + assertEquals(isRetriableConnectFailure(new Error("boom")), false); + assertEquals(isRetriableConnectFailure(null), false); + }); + + it("retries ETIMEDOUT only when it came from connect", () => { + // A socket timeout after the request was written carries the same code, and + // replaying it could deliver a non-idempotent request twice. + assertEquals( + isRetriableConnectFailure({ code: "ETIMEDOUT", syscall: "connect" }), + true, + ); + assertEquals( + isRetriableConnectFailure({ code: "ETIMEDOUT", syscall: "read" }), + false, + ); + assertEquals(isRetriableConnectFailure({ code: "ETIMEDOUT" }), false); + }); + + it("replays only bodies that re-read identically", () => { + assertEquals(isReplayableRequestBody(null), true); + assertEquals(isReplayableRequestBody("{}"), true); + assertEquals(isReplayableRequestBody(new Uint8Array([1, 2])), true); + assertEquals(isReplayableRequestBody(new URLSearchParams("a=1")), true); + // Immutable, and writeRequestBody takes a fresh body.stream() per attempt. + assertEquals(isReplayableRequestBody(new Blob(["x"])), true); + // Already drained by the attempt that failed, so a retry would send nothing. + assertEquals( + isReplayableRequestBody(new Blob(["x"]).stream() as unknown as BodyInit), + false, + ); + }); +}); diff --git a/src/platform/compat/http/pinned-fetch.ts b/src/platform/compat/http/pinned-fetch.ts index a5c18befe2..3e9b06dd5f 100644 --- a/src/platform/compat/http/pinned-fetch.ts +++ b/src/platform/compat/http/pinned-fetch.ts @@ -88,33 +88,68 @@ function addressFamily(address: string): 4 | 6 { return address.includes(":") ? 6 : 4; } -function createPinnedLookup(addresses: readonly string[]): RequestOptions["lookup"] { - let nextIndex = 0; - return ((_hostname: string, options: unknown, callback: (...args: unknown[]) => void) => { - const requestedFamily = typeof options === "number" - ? options - : typeof options === "object" && options !== null && "family" in options - ? Number((options as { family?: unknown }).family ?? 0) - : 0; - const candidates = addresses.filter((address) => - requestedFamily === 0 || addressFamily(address) === requestedFamily - ); - if (candidates.length === 0) { - callback(new Error("No validated address matches the requested address family")); - return; - } - const wantsAll = typeof options === "object" && options !== null && - (options as { all?: unknown }).all === true; - if (wantsAll) { - callback( - null, - candidates.map((address) => ({ address, family: addressFamily(address) })), - ); - return; - } - const address = candidates[nextIndex++ % candidates.length]!; - callback(null, address, addressFamily(address)); - }) as RequestOptions["lookup"]; +/** Connect-level failures that mean "this address is unusable", not "this request is bad". */ +const RETRIABLE_CONNECT_CODES = new Set([ + "ECONNREFUSED", + "EHOSTUNREACH", + "ENETUNREACH", + "EADDRNOTAVAIL", + "ETIMEDOUT", +]); + +/** + * Order the validated addresses into connection attempts. + * + * Each attempt dials exactly one validated address, so the walk happens here + * rather than depending on the runtime: Node honours `autoSelectFamily` and a + * custom `lookup`, Bun honours neither. A different family is tried before a + * sibling of the one that just failed, because a host with no IPv6 route fails + * on every AAAA record its DNS carries. + * + * Every address is already validated by the egress policy, so trying them in + * turn narrows nothing: the set is identical, only the order of use changes. + */ +export function planPinnedConnectAttempts( + addresses: readonly string[], +): readonly (readonly string[])[] { + if (addresses.length <= 1) return addresses.map((address) => [address]); + const first = addresses[0]!; + const otherFamily = addresses.filter((address) => + addressFamily(address) !== addressFamily(first) + ); + const sameFamily = addresses.slice(1).filter((address) => + addressFamily(address) === addressFamily(first) + ); + return [[first], ...[...otherFamily, ...sameFamily].map((address) => [address])]; +} + +/** True when the request may be issued again against a different address. */ +export function isRetriableConnectFailure(error: unknown): boolean { + if (typeof error !== "object" || error === null) return false; + const code = (error as { code?: unknown }).code; + if (typeof code !== "string" || !RETRIABLE_CONNECT_CODES.has(code)) return false; + // ETIMEDOUT is the one code here that is not exclusively a connect failure: + // it also surfaces when a socket times out after the request was written, and + // replaying then could deliver a non-idempotent request twice. Only the + // connect syscall is known to have reached no server. + if (code === "ETIMEDOUT") { + return (error as { syscall?: unknown }).syscall === "connect"; + } + return true; +} + +/** + * A body may only be replayed when re-reading it yields the same bytes. A + * ReadableStream does not qualify: the failed attempt already drained it, so a + * retry would send nothing. + */ +export function isReplayableRequestBody(body: BodyInit | null): boolean { + // A Blob counts: it is immutable and `writeRequestBody` calls `body.stream()` + // per attempt, so each attempt gets a fresh stream over identical bytes. A + // ReadableStream does not, because the attempt that failed already drained it. + return body === null || typeof body === "string" || + body instanceof URLSearchParams || body instanceof ArrayBuffer || + ArrayBuffer.isView(body) || body instanceof Blob; } function copyResponseHeaders(message: IncomingMessage): Headers { @@ -225,82 +260,118 @@ export async function fetchWithPinnedAddresses( const transport = url.protocol === "https:" ? await import("node:https") : await import("node:http"); - const requestOptions: RequestOptions & { autoSelectFamily?: boolean } = { - protocol: url.protocol, - hostname: url.hostname, - port: url.port || undefined, - path: `${url.pathname}${url.search}`, - method, - headers: requestHeaders, - lookup: createPinnedLookup(addresses), - // Let Node/Bun race the complete validated address set instead of binding - // availability to whichever A/AAAA record happened to be returned first. - autoSelectFamily: true, - ...(url.protocol === "https:" ? { servername: url.hostname } : {}), - }; + const attempts = planPinnedConnectAttempts(addresses); + const bodyIsReplayable = isReplayableRequestBody(body); + let lastConnectError: unknown; - return await new Promise((resolve, reject) => { - let settled = false; - let responseMessage: IncomingMessage | undefined; - const cleanupAbortListener = () => init.signal?.removeEventListener("abort", abort); - const rejectBeforeResponse = (error: unknown) => { - cleanupAbortListener(); - reject(error); + for (let attemptIndex = 0; attemptIndex < attempts.length; attemptIndex++) { + const requestOptions: RequestOptions & { autoSelectFamily?: boolean } = { + protocol: url.protocol, + // Connect straight to the validated address. Overriding DNS through a + // custom `lookup` is the documented way to pin and Node honours it, but + // Bun's node:https ignores the address it returns and fails with + // ECONNREFUSED even for a reachable one, so the pin was inert there. + // Dialling the address directly needs no runtime cooperation; identity + // travels in the Host header and the TLS SNI name instead. + hostname: attempts[attemptIndex]![0]!, + port: url.port || (url.protocol === "https:" ? 443 : 80), + path: `${url.pathname}${url.search}`, + method, + headers: { ...requestHeaders, host: url.host }, + + ...(url.protocol === "https:" ? { servername: url.hostname } : {}), }; - const request = transport.request(requestOptions, async (message) => { - responseMessage = message; - try { - const responseHeaders = copyResponseHeaders(message); - const status = message.statusCode ?? 500; - if (method === "HEAD" || NULL_BODY_STATUSES.has(status)) { - message.once("end", cleanupAbortListener); - message.once("close", cleanupAbortListener); - message.once("error", cleanupAbortListener); - // Drain any protocol-invalid payload without exposing it through the - // Fetch response. Response rejects stream bodies for these statuses. - message.resume(); - settled = true; - resolve(createPinnedFetchResponse( - status, - message.statusMessage ?? "", - responseHeaders, - null, - method, - )); + + let pendingRequest: ClientRequest | undefined; + try { + return await new Promise((resolve, reject) => { + let settled = false; + let responseMessage: IncomingMessage | undefined; + const cleanupAbortListener = () => init.signal?.removeEventListener("abort", abort); + const rejectBeforeResponse = (error: unknown) => { + cleanupAbortListener(); + reject(error); + }; + const request = transport.request(requestOptions, async (message) => { + responseMessage = message; + try { + const responseHeaders = copyResponseHeaders(message); + const status = message.statusCode ?? 500; + if (method === "HEAD" || NULL_BODY_STATUSES.has(status)) { + message.once("end", cleanupAbortListener); + message.once("close", cleanupAbortListener); + message.once("error", cleanupAbortListener); + // Drain any protocol-invalid payload without exposing it through the + // Fetch response. Response rejects stream bodies for these statuses. + message.resume(); + settled = true; + resolve(createPinnedFetchResponse( + status, + message.statusMessage ?? "", + responseHeaders, + null, + method, + )); + return; + } + const decoded = await decodeResponseBody(message, responseHeaders); + decoded.once("end", cleanupAbortListener); + decoded.once("close", cleanupAbortListener); + decoded.once("error", cleanupAbortListener); + const { Readable } = await import("node:stream"); + const webBody = Readable.toWeb(decoded) as globalThis.ReadableStream; + settled = true; + resolve(createPinnedFetchResponse( + status, + message.statusMessage ?? "", + responseHeaders, + webBody, + method, + )); + } catch (error) { + rejectBeforeResponse(error); + } + }); + + const abort = () => { + const reason = init.signal?.reason ?? + new DOMException("The operation was aborted", "AbortError"); + responseMessage?.destroy(isErrorAcrossRealms(reason) ? reason : undefined); + request.destroy(isErrorAcrossRealms(reason) ? reason : undefined); + if (!settled) rejectBeforeResponse(reason); + }; + init.signal?.addEventListener("abort", abort, { once: true }); + if (init.signal?.aborted) { + abort(); return; } - const decoded = await decodeResponseBody(message, responseHeaders); - decoded.once("end", cleanupAbortListener); - decoded.once("close", cleanupAbortListener); - decoded.once("error", cleanupAbortListener); - const { Readable } = await import("node:stream"); - const webBody = Readable.toWeb(decoded) as globalThis.ReadableStream; - settled = true; - resolve(createPinnedFetchResponse( - status, - message.statusMessage ?? "", - responseHeaders, - webBody, - method, - )); - } catch (error) { - rejectBeforeResponse(error); + request.once("error", rejectBeforeResponse); + // Bun reports connect failures through + // `process.nextTick(() => self.emit("error", err))`, so the emit can + // land after this promise has settled and after the `once` listener + // above has been consumed. With no listener left, Node stream + // semantics turn it into an uncaught exception and the process exits, + // which is how one refused address took down the dev server instead of + // failing a single request. This sink absorbs the late emit; the first + // error still rejects through `rejectBeforeResponse`. + request.on("error", () => {}); + pendingRequest = request; + void writeRequestBody(request, body).catch((error) => request.destroy(error)); + }); + } catch (error) { + // Release the socket of the attempt being abandoned. The sink above stays + // attached, so a teardown error from this destroy has somewhere to land. + pendingRequest?.destroy(); + lastConnectError = error; + const hasAnotherAddress = attemptIndex < attempts.length - 1; + if ( + !hasAnotherAddress || !bodyIsReplayable || init.signal?.aborted || + !isRetriableConnectFailure(error) + ) { + throw error; } - }); - - const abort = () => { - const reason = init.signal?.reason ?? - new DOMException("The operation was aborted", "AbortError"); - responseMessage?.destroy(isErrorAcrossRealms(reason) ? reason : undefined); - request.destroy(isErrorAcrossRealms(reason) ? reason : undefined); - if (!settled) rejectBeforeResponse(reason); - }; - init.signal?.addEventListener("abort", abort, { once: true }); - if (init.signal?.aborted) { - abort(); - return; } - request.once("error", rejectBeforeResponse); - void writeRequestBody(request, body).catch((error) => request.destroy(error)); - }); + } + + throw lastConnectError; }