diff --git a/changelog.d/fixes/tls-postheaders-abort-lifetime.md b/changelog.d/fixes/tls-postheaders-abort-lifetime.md new file mode 100644 index 000000000000..c346f4e7d0f8 --- /dev/null +++ b/changelog.d/fixes/tls-postheaders-abort-lifetime.md @@ -0,0 +1 @@ +- **fix(sse):** Cancel TLS response bodies when the caller aborts after headers and release their session leases without disturbing active responses. diff --git a/open-sse/services/browserBackedChat.ts b/open-sse/services/browserBackedChat.ts index fa51a9c2662a..0f73243ac1c9 100644 --- a/open-sse/services/browserBackedChat.ts +++ b/open-sse/services/browserBackedChat.ts @@ -25,6 +25,7 @@ import { type PooledContext, } from "./browserPool.ts"; import tlsClient from "../utils/tlsClient.ts"; +import { discardResponseBody } from "../executors/opencodeResponseBody.ts"; import { sanitizeErrorMessage } from "../utils/error.ts"; import { resolveHttpBackedChatFingerprint } from "./httpBackedChatFingerprint.ts"; import type { @@ -552,6 +553,7 @@ export async function httpBackedChat( if (contentLengthHeader) { const contentLength = parseInt(contentLengthHeader, 10); if (contentLength > MAX_RESPONSE_BYTES) { + discardResponseBody(response); throw new Error("Response too large"); } } diff --git a/open-sse/utils/tlsClient.ts b/open-sse/utils/tlsClient.ts index 7106d3b5aaa9..957d2d673ea1 100644 --- a/open-sse/utils/tlsClient.ts +++ b/open-sse/utils/tlsClient.ts @@ -488,6 +488,24 @@ function sanitizeWreqError(error: unknown, message: string): SafeWreqError { return sanitized; } +// Keep body-phase cancellation compatible with wreq-js's request-phase errors. +function createWreqAbortError(reason: unknown): Error { + if (reason instanceof DOMException) { + return reason.name === "AbortError" + ? reason + : new DOMException(reason.message || "The operation was aborted", "AbortError"); + } + if (reason instanceof Error) { + const error = new Error(reason.message, { cause: reason }); + error.name = "AbortError"; + return error; + } + return new DOMException( + typeof reason === "string" ? reason : "The operation was aborted", + "AbortError" + ); +} + function toNativeResponse( response: WreqResponse, onFinalize: () => void, @@ -498,14 +516,17 @@ function toNativeResponse( let bodyFailureReported = false; let consumerCancelled = false; let consumerCancelReason: unknown; + let removeAbortListener = () => {}; + let cancelBody: (reason: unknown) => Promise = async () => {}; const finalize = () => { if (finalized) return; finalized = true; + removeAbortListener(); onFinalize(); }; const safeBodyError = (error: unknown): unknown => { if (signal?.aborted) { - return signal.reason ?? new DOMException("The operation was aborted", "AbortError"); + return createWreqAbortError(signal.reason); } if (consumerCancelled) { return ( @@ -518,41 +539,74 @@ function toNativeResponse( } return sanitizeWreqError(error, "wreq-js response body failed"); }; - if (response instanceof Response) { + if (response instanceof Response && response.body === null) { finalize(); return response; } - try { const headers = new Headers(); for (const [name, value] of response.headers) headers.append(name, value); let body: ReadableStream | null = null; if (response.body) { const reader = response.body.getReader(); + let stopped = false; + let cancellation: Promise | undefined; + cancelBody = (reason) => { + cancellation ??= (async () => { + try { + return reader.cancel(reason); + } finally { + // Cancel settles pending reads before native cleanup, which may stall. + reader.releaseLock(); + } + })(); + return cancellation; + }; body = new ReadableStream({ + start(controller) { + if (!signal) return; + const onAbort = () => { + if (stopped) return; + stopped = true; + const error = createWreqAbortError(signal.reason); + finalize(); + controller.error(error); + void cancelBody(error).catch(() => {}); + }; + // wreq-js stops observing the signal at headers; keep it for the body. + removeAbortListener = () => signal.removeEventListener("abort", onAbort); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + }, async pull(controller) { try { const chunk = await reader.read(); + if (stopped) return; if (chunk.done) { + stopped = true; + reader.releaseLock(); finalize(); controller.close(); } else { controller.enqueue(chunk.value); } } catch (error) { + if (stopped) return; + stopped = true; + reader.releaseLock(); controller.error(safeBodyError(error)); finalize(); } }, async cancel(reason) { + stopped = true; consumerCancelled = true; consumerCancelReason = reason; + finalize(); try { - await reader.cancel(reason); + await cancelBody(reason); } catch (error) { throw safeBodyError(error); - } finally { - finalize(); } }, }); @@ -576,6 +630,7 @@ function toNativeResponse( return adapted; } catch (error) { finalize(); + void cancelBody(error).catch(() => {}); throw error; } } diff --git a/tests/unit/http-backed-chat-response-lifetime.test.ts b/tests/unit/http-backed-chat-response-lifetime.test.ts new file mode 100644 index 000000000000..9790fb442ae7 --- /dev/null +++ b/tests/unit/http-backed-chat-response-lifetime.test.ts @@ -0,0 +1,328 @@ +import assert from "node:assert/strict"; +import { getEventListeners } from "node:events"; +import test, { type TestContext } from "node:test"; + +import { + httpBackedChat, + tryBackedChat, + __setBrowserBackedChatOverrideForTesting, + __resetBrowserBackedChatOverrideForTesting, +} from "../../open-sse/services/browserBackedChat.ts"; +import type { BrowserBackedChatRequest } from "../../open-sse/services/browserBackedChat/types.ts"; +import tlsClient, { + TlsClient, + type TlsFetchOptions, + type WreqResponse, + type WreqSession, +} from "../../open-sse/utils/tlsClient.ts"; + +const MAX_RESPONSE_BYTES = 10 * 1024 * 1024; +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +function request(signal?: AbortSignal): BrowserBackedChatRequest { + return { + poolKey: "account-a", + chatUrl: "https://synthetic.test/chat", + chatPageUrl: "https://synthetic.test/", + chatUrlMatchDomain: "synthetic.test", + userMessage: "hello", + inputSelector: "textarea", + signal, + }; +} + +function sourceBody( + chunks = ["one", "two", "three", "four", "five", "six", "seven", "eight"], + cancel: () => void | Promise = () => {} +) { + let reads = 0; + let cancels = 0; + let eof = false; + const body = new ReadableStream( + { + pull(controller) { + if (reads === chunks.length) { + eof = true; + controller.close(); + } else { + controller.enqueue(new TextEncoder().encode(chunks[reads++])); + } + }, + cancel() { + cancels += 1; + return cancel(); + }, + }, + { highWaterMark: 0 } + ); + return { body, reads: () => reads, cancels: () => cancels, eof: () => eof }; +} + +function upstream(body: ReadableStream, length: number, native = true): WreqResponse { + const headers = new Headers({ + "content-length": String(length), + "content-type": "application/json", + }); + return native + ? new Response(body, { headers }) + : { status: 200, statusText: "OK", headers, body }; +} + +function fixture(t: TestContext, respond: (url: string) => WreqResponse) { + let creates = 0; + let closes = 0; + const client = new TlsClient(async () => { + creates += 1; + const session: WreqSession = { + async fetch(url) { + return respond(url); + }, + close() { + closes += 1; + }, + getCookies() { + return {}; + }, + }; + return session; + }, 1); + const counts = () => [ + ...(client as unknown as { sessionUseCounts: Map }).sessionUseCounts.values(), + ]; + t.mock.getter(tlsClient, "available", () => true); + t.mock.method(tlsClient, "fetch", (url: string, options?: TlsFetchOptions) => + client.fetch(url, { ...options, proxy: null }) + ); + t.after(() => client.closeAll()); + return { client, counts, creates: () => creates, closes: () => closes }; +} + +function assertOversized(result: Awaited>) { + assert.equal(result.status, 502); + assert.equal(result.contentType, "application/json"); + assert.deepEqual(JSON.parse(result.body.toString()), { + error: { message: "httpBackedChat failed: Response too large", type: "upstream_error" }, + }); +} + +async function promptly(operation: Promise): Promise { + let timer: ReturnType; + try { + return await Promise.race([ + operation, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("caller waited for cancellation")), 500); + }), + ]); + } finally { + clearTimeout(timer!); + } +} + +for (const native of [true, false]) { + for (const withSignal of [true, false]) { + test(`oversized ${native ? "native Response" : "WreqResponse"} releases its lease with ${withSignal ? "a live signal" : "no signal"}`, async (t) => { + const source = sourceBody(); + const abort = new AbortController(); + const { client, counts, creates, closes } = fixture(t, (url) => + url.endsWith("/chat") + ? upstream(source.body, MAX_RESPONSE_BYTES + 1, native) + : new Response("admitted") + ); + + assertOversized(await httpBackedChat(request(withSignal ? abort.signal : undefined))); + await tick(); + assert.equal(source.cancels(), 1); + assert.ok(source.reads() <= 1, "the adapter buffers at most one chunk before cancellation"); + assert.equal(source.eof(), false, "oversized bodies must not be drained"); + assert.deepEqual(counts(), []); + assert.equal(source.body.locked, false); + assert.equal(abort.signal.aborted, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(closes(), 0, "release happens before session teardown"); + + const next = await client.fetch("https://synthetic.test/next", { + proxy: null, + sessionScope: "account-b", + }); + assert.equal(await next.text(), "admitted"); + assert.equal(creates(), 2, "another scope acquires the sole session slot"); + assert.deepEqual(counts(), []); + assert.equal(source.cancels(), 1); + }); + } +} + +for (const ending of ["rejects", "stalls"] as const) { + test(`oversized native cancellation that ${ending} preserves a prompt 502`, async (t) => { + let finishCancellation: (() => void) | undefined; + const source = sourceBody(undefined, () => + ending === "rejects" + ? Promise.reject(new Error("secret cancellation failure")) + : new Promise((resolve) => { + finishCancellation = resolve; + }) + ); + const abort = new AbortController(); + const { client, counts } = fixture(t, (url) => + url.endsWith("/chat") + ? upstream(source.body, MAX_RESPONSE_BYTES + 1) + : new Response("admitted") + ); + try { + assertOversized(await promptly(httpBackedChat(request(abort.signal)))); + await tick(); + assert.equal(source.cancels(), 1); + assert.ok(source.reads() <= 1); + assert.equal(source.eof(), false); + assert.deepEqual(counts(), []); + assert.equal(abort.signal.aborted, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + const next = await client.fetch("https://synthetic.test/next", { + proxy: null, + sessionScope: "account-b", + }); + assert.equal(await next.text(), "admitted"); + assert.equal(client.getCircuitState(null, "account-a").failureCount, 0); + } finally { + finishCancellation?.(); + await tick(); + } + assert.equal(source.cancels(), 1); + assert.equal(source.body.locked, false); + }); +} + +for (const length of [17, MAX_RESPONSE_BYTES]) { + test(`JSON with declared length ${length} is consumed to EOF without cancellation`, async (t) => { + const chunks = ['{"answer":', '"hello"}']; + const source = sourceBody(chunks); + const abort = new AbortController(); + const { counts } = fixture(t, () => upstream(source.body, length)); + + const result = await httpBackedChat(request(abort.signal)); + assert.equal(result.status, 200); + assert.equal(result.contentType, "application/json"); + assert.equal(result.body.toString(), chunks.join("")); + assert.equal(source.reads(), chunks.length); + assert.equal(source.eof(), true); + assert.equal(source.cancels(), 0); + assert.deepEqual(counts(), []); + assert.equal(source.body.locked, false); + assert.equal(abort.signal.aborted, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + }); +} + +test("oversized rejection preserves a slow active body on the same session", async (t) => { + const oversized = sourceBody(); + const active = sourceBody(["first", "second", "third"]); + const abort = new AbortController(); + const { client, counts, creates, closes } = fixture(t, (url) => + url.endsWith("/chat") + ? upstream(oversized.body, MAX_RESPONSE_BYTES + 1) + : upstream(active.body, 16) + ); + const response = await client.fetch("https://synthetic.test/active", { + proxy: null, + sessionScope: "account-a", + signal: abort.signal, + }); + const reader = response.body!.getReader(); + assert.deepEqual(counts(), [1]); + + assertOversized(await httpBackedChat(request(abort.signal))); + await tick(); + assert.equal(oversized.cancels(), 1); + assert.deepEqual(counts(), [1], "only the rejected response releases a lease"); + assert.equal(creates(), 1); + assert.equal(closes(), 0); + assert.equal(active.cancels(), 0); + assert.ok(active.reads() <= 1, "the slow consumer controls its own read demand"); + await assert.rejects( + client.fetch("https://synthetic.test/blocked", { proxy: null, sessionScope: "account-b" }), + { code: "TLS_SESSION_CAPACITY" } + ); + let body = ""; + while (true) { + const chunk = await reader.read(); + if (chunk.done) break; + body += new TextDecoder().decode(chunk.value); + await tick(); + } + reader.releaseLock(); + assert.equal(body, "firstsecondthird"); + assert.equal(active.eof(), true); + assert.equal(active.cancels(), 0); + assert.deepEqual(counts(), []); + assert.equal(abort.signal.aborted, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(oversized.cancels(), 1); +}); + +for (const native of [false, true]) { + for (const combined of [false, true]) { + test(`${combined ? "tryBackedChat" : "httpBackedChat"}: ${native ? "native Response" : "WreqResponse"} post-header timeout follows the AbortError path`, async (t) => { + const reading = Promise.withResolvers(); + const reasons: unknown[] = []; + let controller!: ReadableStreamDefaultController; + const body = new ReadableStream( + { + start(value) { + controller = value; + }, + pull() { + reading.resolve(); + }, + cancel(reason) { + reasons.push(reason); + }, + }, + { highWaterMark: 0 } + ); + const { client, counts } = fixture(t, () => upstream(body, 17, native)); + let browserCalls = 0; + __setBrowserBackedChatOverrideForTesting(async () => { + browserCalls += 1; + throw new Error("unexpected browser fallback after timeout"); + }); + t.after(__resetBrowserBackedChatOverrideForTesting); + const abort = new AbortController(); + const pending = (combined ? tryBackedChat : httpBackedChat)(request(abort.signal)); + const checked = combined + ? promptly(pending) + : assert.rejects(promptly(pending), (error) => { + assert.ok(error instanceof DOMException); + assert.equal(error.name, "AbortError"); + assert.equal(error.message, "deadline expired"); + return true; + }); + void checked.catch(() => {}); + try { + await promptly(reading.promise); + assert.deepEqual(counts(), [1]); + abort.abort(new DOMException("deadline expired", "TimeoutError")); + const result = await checked; + if (combined) { + assert.ok(result); + assert.equal(result.status, 504); + assert.deepEqual(JSON.parse(result.body.toString()), { + error: { message: "tryBackedChat timed out", type: "timeout_error" }, + }); + } + assert.equal(browserCalls, 0); + assert.equal(reasons.length, 1); + assert.ok(reasons[0] instanceof DOMException); + assert.equal(reasons[0].name, "AbortError"); + assert.equal(reasons[0].message, "deadline expired"); + assert.deepEqual(counts(), []); + assert.equal(body.locked, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(client.getCircuitState(null, "account-a").failureCount, 0); + } finally { + controller.error(new DOMException("test body cleanup", "AbortError")); + await pending.catch(() => {}); + } + }); + } +} diff --git a/tests/unit/tls-client-abort-lifetime.test.ts b/tests/unit/tls-client-abort-lifetime.test.ts new file mode 100644 index 000000000000..e61daddc3b62 --- /dev/null +++ b/tests/unit/tls-client-abort-lifetime.test.ts @@ -0,0 +1,449 @@ +import assert from "node:assert/strict"; +import { getEventListeners } from "node:events"; +import test, { type TestContext } from "node:test"; + +import { TlsClient, type WreqResponse, type WreqSession } from "../../open-sse/utils/tlsClient.ts"; + +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +async function promptly(operation: Promise): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + operation, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("body operation did not settle")), 500); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +function source(cancel: (reason: unknown) => void | Promise = () => {}) { + let controller!: ReadableStreamDefaultController; + const reasons: unknown[] = []; + const body = new ReadableStream( + { + start(value) { + controller = value; + }, + cancel(reason) { + reasons.push(reason); + return cancel(reason); + }, + }, + { highWaterMark: 0 } + ); + return { body, controller, reasons }; +} + +function upstream(body: ReadableStream | null, native: boolean): WreqResponse { + const metadata = { + status: 201, + statusText: "Created", + headers: new Headers({ "x-synthetic": "yes", "content-type": "text/plain" }), + url: "https://synthetic.test/redirected", + redirected: true, + }; + if (!native) return { ...metadata, body }; + const response = new Response(body, metadata); + Object.defineProperty(response, "url", { value: metadata.url }); + Object.defineProperty(response, "redirected", { value: metadata.redirected }); + return response; +} + +// Match wreq-js setupAbort's pre-header error contract, including non-Error reasons. +function nativeAbortError(reason: unknown): Error { + if (reason instanceof DOMException) { + return reason.name === "AbortError" + ? reason + : new DOMException(reason.message || "The operation was aborted", "AbortError"); + } + if (reason instanceof Error) { + return Object.assign(new Error(reason.message, { cause: reason }), { name: "AbortError" }); + } + return new DOMException( + typeof reason === "string" ? reason : "The operation was aborted", + "AbortError" + ); +} + +function fixture(t: TestContext, respond: () => WreqResponse) { + let closes = 0; + const session: WreqSession = { + async fetch(url, options) { + const signal = options?.signal as AbortSignal | undefined; + if (signal?.aborted) throw nativeAbortError(signal.reason); + return url.endsWith("/next") ? new Response("admitted") : respond(); + }, + close() { + closes += 1; + }, + getCookies() { + return {}; + }, + }; + const client = new TlsClient(async () => session, 1); + const responses: Response[] = []; + const readers: ReadableStreamDefaultReader[] = []; + const counts = () => [ + ...(client as unknown as { sessionUseCounts: Map }).sessionUseCounts.values(), + ]; + const fetch = async (signal?: AbortSignal, scope = "account-a") => { + const response = await client.fetch("https://synthetic.test/body", { + proxy: null, + signal, + sessionScope: scope, + }); + responses.push(response); + return response; + }; + const read = (response: Response) => { + const reader = response.body!.getReader(); + readers.push(reader); + return reader; + }; + t.after(async () => { + for (const reader of readers) void reader.cancel().catch(() => {}); + for (const response of responses) { + if (!response.body?.locked) void response.body?.cancel().catch(() => {}); + } + await client.closeAll(); + }); + return { client, counts, fetch, read, closes: () => closes }; +} + +for (const native of [false, true]) { + const variant = native ? "native Response" : "WreqResponse"; + + test(`${variant}: a signal aborted before fetching preserves the native rejection`, async (t) => { + const input = source(); + const f = fixture(t, () => upstream(input.body, native)); + const abort = new AbortController(); + const reason = new DOMException("caller stopped before headers", "AbortError"); + abort.abort(reason); + await assert.rejects(f.fetch(abort.signal), (error) => error === reason); + assert.deepEqual(f.counts(), []); + assert.deepEqual(input.reasons, []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + }); + + for (const state of ["idle", "pending", "buffered", "already-aborted"] as const) { + test(`${variant}: caller abort cancels a ${state} body after headers`, async (t) => { + const input = source(); + if (state === "buffered") input.controller.enqueue(Uint8Array.of(7)); + const abort = new AbortController(); + const reason = new DOMException("caller stopped", "AbortError"); + const f = fixture(t, () => { + // Match dispatchRequest's signal cleanup before handing headers to the adapter. + if (state === "already-aborted") abort.abort(reason); + return upstream(input.body, native); + }); + const response = await f.fetch(abort.signal); + assert.equal(response.status, 201); + assert.equal(response.statusText, "Created"); + assert.equal(response.headers.get("x-synthetic"), "yes"); + assert.equal(response.url, "https://synthetic.test/redirected"); + assert.equal(response.redirected, true); + const reader = f.read(response); + const pending = state === "pending" ? reader.read() : undefined; + // Attach rejection handling before abort dispatches synchronously. + const rejected = pending && assert.rejects(promptly(pending), (error) => error === reason); + void rejected?.catch(() => {}); + if (state !== "already-aborted") { + assert.deepEqual(f.counts(), [1], "headers must not make an active body idle"); + assert.equal(getEventListeners(abort.signal, "abort").length, 1); + await tick(); + abort.abort(reason); + } + await tick(); + assert.deepEqual(input.reasons, [reason]); + assert.deepEqual(f.counts(), []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(input.body.locked, false); + if (rejected) await rejected; + else await assert.rejects(promptly(reader.read()), (error) => error === reason); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + assert.equal(f.closes(), 0, "body abort releases the lease without closing the session"); + const next = await f.client.fetch("https://synthetic.test/next", { + proxy: null, + sessionScope: "account-b", + }); + assert.equal(await next.text(), "admitted"); + assert.equal(f.closes(), 1, "another scope can evict the idle session"); + }); + } + + for (const ending of ["eof", "error", "cancel", "cancel-rejects", "no-body"] as const) { + test(`${variant}: ${ending} finalizes once and removes the abort listener`, async (t) => { + const input = source(() => { + if (ending === "cancel-rejects") throw new Error("secret cancellation details"); + }); + const f = fixture(t, () => upstream(ending === "no-body" ? null : input.body, native)); + const abort = new AbortController(); + const response = await f.fetch(abort.signal); + if (ending === "no-body") assert.equal(response.body, null); + else { + assert.deepEqual(f.counts(), [1]); + const reader = f.read(response); + const pending = reader.read(); + if (ending === "error") { + const rejected = assert.rejects(promptly(pending), { + message: "wreq-js response body failed", + code: "ECONNRESET", + }); + input.controller.error( + Object.assign(new Error("secret upstream details"), { code: "ECONNRESET" }) + ); + await rejected; + } else if (ending === "eof") { + input.controller.enqueue(Uint8Array.of(1, 2)); + input.controller.close(); + assert.deepEqual((await promptly(pending)).value, Uint8Array.of(1, 2)); + assert.equal((await promptly(reader.read())).done, true); + } else { + const cancel = reader.cancel("consumer stopped"); + if (ending === "cancel-rejects") { + await assert.rejects(promptly(cancel), (error) => error === "consumer stopped"); + } else await promptly(cancel); + assert.equal((await promptly(pending)).done, true); + } + } + await tick(); + assert.deepEqual(f.counts(), []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(input.body.locked, false); + const expectedCancels = ending.startsWith("cancel") ? ["consumer stopped"] : []; + assert.deepEqual(input.reasons, expectedCancels); + abort.abort(); + await tick(); + assert.deepEqual(input.reasons, expectedCancels); + assert.deepEqual(f.counts(), []); + assert.equal( + f.client.getCircuitState(null, "account-a").failureCount, + ending === "error" ? 1 : 0 + ); + }); + } + + for (const cancellation of ["rejects", "stalls"] as const) { + test(`${variant}: caller abort releases promptly when native cancellation ${cancellation}`, async (t) => { + let finish!: () => void; + const input = source(() => + cancellation === "rejects" + ? Promise.reject(new Error("secret cancellation failure")) + : new Promise((resolve) => { + finish = resolve; + }) + ); + const f = fixture(t, () => upstream(input.body, native)); + const abort = new AbortController(); + const response = await f.fetch(abort.signal); + const reader = f.read(response); + const rejected = assert.rejects(promptly(reader.read()), { name: "AbortError" }); + abort.abort(); + try { + await rejected; + await tick(); + assert.equal(input.reasons.length, 1); + assert.deepEqual(f.counts(), []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + const next = await f.client.fetch("https://synthetic.test/next", { + proxy: null, + sessionScope: "account-b", + }); + assert.equal(await next.text(), "admitted"); + } finally { + finish?.(); + await tick(); + } + assert.equal(input.body.locked, false); + }); + } + + for (const first of [ + "abort", + "cancel", + "eof", + "error", + "eof-pending", + "error-pending", + "chunk-pending", + ] as const) { + test(`${variant}: ${first}/abort race preserves an active sibling lease`, async (t) => { + const inputs = [source(), source()]; + let next = 0; + const f = fixture(t, () => upstream(inputs[next++ % 2].body, native)); + const abort = new AbortController(); + const response = await f.fetch(abort.signal); + const sibling = await f.fetch(); + const siblingReader = f.read(sibling); + assert.deepEqual(f.counts(), [2]); + const reader = f.read(response); + if (first === "abort") { + abort.abort(); + await assert.rejects(promptly(reader.read()), { name: "AbortError" }); + await reader.cancel().catch(() => {}); + } else if (first === "cancel") { + const cancellation = reader.cancel("consumer stopped"); + abort.abort(); + await promptly(cancellation); + } else if (first.endsWith("-pending")) { + const reason = new DOMException("caller won the read race", "AbortError"); + const rejected = assert.rejects(promptly(reader.read()), (error) => error === reason); + if (first === "eof-pending") inputs[0].controller.close(); + else if (first === "error-pending") { + inputs[0].controller.error(new Error("secret upstream details")); + } else inputs[0].controller.enqueue(Uint8Array.of(5)); + abort.abort(reason); + await rejected; + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + } else if (first === "eof") { + const pending = reader.read(); + inputs[0].controller.close(); + assert.equal((await promptly(pending)).done, true); + abort.abort(); + } else { + const rejected = assert.rejects(promptly(reader.read()), { + message: "wreq-js response body failed", + }); + inputs[0].controller.error(new Error("secret upstream details")); + await rejected; + abort.abort(); + } + await tick(); + assert.deepEqual(f.counts(), [1], "only the terminating body's lease is released"); + assert.equal( + inputs[0].reasons.length, + ["abort", "cancel", "chunk-pending"].includes(first) ? 1 : 0 + ); + assert.equal(inputs[1].reasons.length, 0); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + await assert.rejects( + f.client.fetch("https://synthetic.test/blocked", { + proxy: null, + sessionScope: "account-b", + }), + { code: "TLS_SESSION_CAPACITY" } + ); + inputs[1].controller.enqueue(Uint8Array.of(9)); + inputs[1].controller.close(); + assert.deepEqual((await promptly(siblingReader.read())).value, Uint8Array.of(9)); + assert.equal((await promptly(siblingReader.read())).done, true); + assert.deepEqual(f.counts(), []); + assert.equal(inputs[1].body.locked, false); + }); + } +} + +test("bodyless native error responses preserve status zero without a body lease", async (t) => { + const f = fixture(t, () => Response.error()); + const abort = new AbortController(); + const response = await f.fetch(abort.signal); + assert.equal(response.status, 0); + assert.equal(response.statusText, ""); + assert.equal(response.type, "error"); + assert.equal(response.body, null); + assert.deepEqual(f.counts(), []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); +}); + +const abortCases = [ + { + label: "timeout", + reason: () => new DOMException("deadline expired", "TimeoutError"), + dom: true, + message: "deadline expired", + }, + { + label: "empty timeout", + reason: () => new DOMException("", "TimeoutError"), + dom: true, + message: "The operation was aborted", + }, + { + label: "custom Error", + reason: () => new TypeError("caller stopped"), + dom: false, + message: "caller stopped", + }, + { label: "string", reason: () => "caller stopped", dom: true, message: "caller stopped" }, + { label: "empty string", reason: () => "", dom: true, message: "" }, + { label: "default", reason: () => undefined, dom: true, preserve: true }, + { + label: "AbortError", + reason: () => new DOMException("already normalized", "AbortError"), + dom: true, + message: "already normalized", + preserve: true, + }, +]; + +for (const native of [false, true]) { + for (const scenario of abortCases) { + for (const timing of ["before fetch", "at headers", "during read"]) { + test(`${native ? "native Response" : "WreqResponse"}: ${scenario.label} abort ${timing} uses the wreq error shape`, async (t) => { + const input = source(); + const abort = new AbortController(); + const reason = scenario.reason(); + const f = fixture(t, () => { + if (timing === "at headers") abort.abort(reason); + return upstream(input.body, native); + }); + const verify = (error: unknown) => { + assert.ok(error instanceof Error); + assert.equal(error.name, "AbortError"); + assert.equal(error instanceof DOMException, scenario.dom); + if (scenario.message !== undefined) assert.equal(error.message, scenario.message); + if (scenario.preserve) assert.equal(error, abort.signal.reason); + else assert.notEqual(error, reason); + assert.equal(error.cause, scenario.dom ? undefined : reason); + return true; + }; + if (timing === "before fetch") { + abort.abort(reason); + await assert.rejects(f.fetch(abort.signal), verify); + assert.equal(input.reasons.length, 0); + } else { + const response = await f.fetch(abort.signal); + const pending = f.read(response).read(); + const rejected = assert.rejects(promptly(pending), verify); + if (timing === "during read") abort.abort(reason); + await rejected; + await tick(); + assert.equal(input.reasons.length, 1); + verify(input.reasons[0]); + assert.equal(input.body.locked, false); + } + assert.deepEqual(f.counts(), []); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + }); + } + } + + test(`${native ? "native Response" : "WreqResponse"}: timeout during rejected consumer cancellation is normalized`, async (t) => { + const abort = new AbortController(); + const input = source(() => { + abort.abort(new DOMException("deadline expired", "TimeoutError")); + throw new Error("secret cancellation failure"); + }); + const f = fixture(t, () => upstream(input.body, native)); + const response = await f.fetch(abort.signal); + await assert.rejects(response.body!.cancel("consumer stopped"), (error) => { + assert.ok(error instanceof DOMException); + assert.equal(error.name, "AbortError"); + assert.equal(error.message, "deadline expired"); + return true; + }); + assert.deepEqual(input.reasons, ["consumer stopped"]); + assert.deepEqual(f.counts(), []); + assert.equal(input.body.locked, false); + assert.equal(getEventListeners(abort.signal, "abort").length, 0); + assert.equal(f.client.getCircuitState(null, "account-a").failureCount, 0); + }); +} diff --git a/tests/unit/tls-proxy-context.test.ts b/tests/unit/tls-proxy-context.test.ts index 6e1307e5f783..661f0c723567 100644 --- a/tests/unit/tls-proxy-context.test.ts +++ b/tests/unit/tls-proxy-context.test.ts @@ -676,9 +676,10 @@ test("bounded session cache closes the least-recently-used idle session", async }; }, 2); - await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-1:8080" }); - await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-2:8080" }); - await client.fetch("https://upstream.example/v1", { proxy: "http://proxy-3:8080" }); + for (const proxy of ["http://proxy-1:8080", "http://proxy-2:8080", "http://proxy-3:8080"]) { + const response = await client.fetch("https://upstream.example/v1", { proxy }); + assert.equal(await response.text(), "ok"); + } assert.deepEqual(closed, ["http://proxy-1:8080"]); await client.exit();