diff --git a/clients/web/src/test/core/auth/requestTimeout.test.ts b/clients/web/src/test/core/auth/requestTimeout.test.ts index c7cb609f..5f5b0465 100644 --- a/clients/web/src/test/core/auth/requestTimeout.test.ts +++ b/clients/web/src/test/core/auth/requestTimeout.test.ts @@ -298,6 +298,78 @@ describe("withOAuthRequestTimeout", () => { expect((await assertion) as Error).toBeInstanceOf(OAuthRequestTimeoutError); }); + it("releases both tee branches when the deadline wins", async () => { + vi.useFakeTimers(); + // Losing the race rejects the caller; on its own it does not stop the read. + // A drain still running would keep pulling bytes *and* keep buffering them + // for the unread original branch — unbounded memory on exactly the + // signal-ignoring body the race exists to contain. Observed through the + // stream's own `cancel`, which is what tearing the branches down calls. + let cancelled = false; + const stalled = new ReadableStream({ + start(ctrl) { + ctrl.enqueue(new TextEncoder().encode('{"issuer":')); + }, + // Never resolves: headers arrived, the body never finishes. + pull() { + return new Promise(() => {}); + }, + cancel() { + cancelled = true; + }, + }); + const wrapped = withOAuthRequestTimeout( + vi + .fn() + .mockResolvedValue( + new Response(stalled, { headers: { "content-type": "text/plain" } }), + ), + 1000, + ); + + const assertion = wrapped(URL_UNDER_TEST).catch((err: unknown) => err); + await vi.advanceTimersByTimeAsync(1000); + + expect((await assertion) as Error).toBeInstanceOf(OAuthRequestTimeoutError); + // The cancels are deliberately not awaited — cancelling one branch of a tee + // can wait on the other — so let them settle before observing. + for (let i = 0; i < 20; i++) await Promise.resolve(); + expect(cancelled).toBe(true); + }); + + it("still reports the timeout when tearing the branches down fails", async () => { + vi.useFakeTimers(); + // Cancelling a stream that is already errored, locked elsewhere, or whose + // source rejects is not a failure here — the cleanup runs on a path that is + // already rejecting with something the caller cares about far more. + const hostile = new ReadableStream({ + start(ctrl) { + ctrl.enqueue(new TextEncoder().encode("x")); + }, + pull() { + return new Promise(() => {}); + }, + cancel() { + throw new Error("cancel blew up"); + }, + }); + const wrapped = withOAuthRequestTimeout( + vi + .fn() + .mockResolvedValue( + new Response(hostile, { headers: { "content-type": "text/plain" } }), + ), + 1000, + ); + + const assertion = wrapped(URL_UNDER_TEST).catch((err: unknown) => err); + await vi.advanceTimersByTimeAsync(1000); + + expect((await assertion) as Error).toBeInstanceOf(OAuthRequestTimeoutError); + // And no unhandled rejection escapes from the cleanup. + for (let i = 0; i < 20; i++) await Promise.resolve(); + }); + it("hands back the original response, not a copy", async () => { const original = new Response('{"issuer":"https://as.example.com"}', { status: 201, diff --git a/core/auth/requestTimeout.ts b/core/auth/requestTimeout.ts index ba06e53f..7740b58c 100644 --- a/core/auth/requestTimeout.ts +++ b/core/auth/requestTimeout.ts @@ -319,6 +319,44 @@ function requestHeadersOf( return new Headers(); } +/** + * Pull a reader to completion, discarding what it yields. An absent reader — a + * null-body response — is already complete. + * + * Read through an explicit reader rather than `clone().arrayBuffer()` so the + * losing side of the race has something it can cancel: `arrayBuffer()` locks + * the stream to an internal reader nothing else can reach. + */ +async function drainToEnd( + reader: ReadableStreamDefaultReader | undefined, +): Promise { + if (!reader) return; + for (;;) { + const { done } = await reader.read(); + if (done) return; + } +} + +/** + * Tear down both branches of the tee. + * + * Deliberately **not awaited**, and the `void`s say which of the documented + * cases this is: the callee owns its failures (each carries its own `catch`), + * and the caller genuinely cannot await — cancelling one branch of a tee can + * itself wait on the other, which would hold the timeout open for exactly as + * long as the stall it is meant to end. Best-effort on each besides: a stream + * already cancelled, errored or locked elsewhere is not a failure here, and + * this runs on a path that is already rejecting with something the caller cares + * about more. + */ +function cancelBoth( + reader: ReadableStreamDefaultReader | undefined, + response: Response | undefined, +): void { + void reader?.cancel().catch(() => {}); + void response?.body?.cancel().catch(() => {}); +} + /** The request URL, whatever form `fetch`'s first argument took. */ function requestUrlOf(input: RequestInfo | URL): string { if (typeof input === "string") return input; @@ -466,11 +504,16 @@ export function withOAuthRequestTimeout( const nextInit: RequestInit = { ...init, signal }; REQUEST_DEADLINES.set(nextInit, budget); + // Held so the losing path can tear both tee branches down — see below. + let settled: Response | undefined; + let drainReader: ReadableStreamDefaultReader | undefined; + try { const response = await Promise.race([ fetchFn(input, nextInit), abandoned, ]); + settled = response; // `fetch` resolves once the response *headers* arrive, so stopping here // would leave the body unbounded: an authorization server can send // headers, or half a JSON document, and then stall, and the caller's @@ -496,9 +539,19 @@ export function withOAuthRequestTimeout( // `content-length` wrong, so those had to be stripped — a rebuilt // response was observably not the original in at least four ways. Not // rebuilding removes the whole class. - await Promise.race([response.clone().arrayBuffer(), abandoned]); + const copy = response.clone(); + drainReader = copy.body?.getReader(); + await Promise.race([drainToEnd(drainReader), abandoned]); return response; } catch (err) { + // ⚠️ Losing the race rejects the caller; on its own it does not stop the + // read. `clone()` tees the body, so a drain still running would keep + // pulling bytes *and* keep buffering them for the unread original branch + // — unbounded memory on exactly the signal-ignoring body the race exists + // to contain (Copilot). Both branches are cancelled here, which is safe + // precisely because this path throws: the caller never receives the + // response, so nothing is left to read it. + cancelBoth(drainReader, settled); // `controller.abort()` above can reject the underlying fetch *before* // `reject` runs, in which case the race settles with undici's // `AbortError` instead of ours. Which of the two wins is an ordering