fix: stop the drain and release both tee branches on timeout (#2319 round 24)
Losing the race rejected the caller but did not stop the read. Because clone() tees the body, a drain still running kept pulling bytes AND kept buffering them for the unread original branch — unbounded memory on exactly the signal-ignoring body the race exists to contain. The clone is now drained through an explicit reader rather than clone().arrayBuffer(), precisely so the losing side has something it can cancel: arrayBuffer() locks the stream to an internal reader nothing else can reach. On any throw both branches are torn down, which is safe because that path never hands the response to a caller. The cancels are deliberately not awaited: cancelling one branch of a tee can itself wait on the other, which would hold the timeout open for as long as the stall it is meant to end. Each carries its own catch, so a stream already cancelled, errored or locked elsewhere does not mask the timeout the caller actually needs — covered by a test whose stream throws from cancel(). Verified by mutation: removing the teardown makes the new test fail. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YFnM6RqB4nXqNysgUYmjmq Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
1e501fa5cb
commit
78f82475c4
2 changed files with 126 additions and 1 deletions
|
|
@ -298,6 +298,78 @@ describe("withOAuthRequestTimeout", () => {
|
||||||
expect((await assertion) as Error).toBeInstanceOf(OAuthRequestTimeoutError);
|
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<Uint8Array>({
|
||||||
|
start(ctrl) {
|
||||||
|
ctrl.enqueue(new TextEncoder().encode('{"issuer":'));
|
||||||
|
},
|
||||||
|
// Never resolves: headers arrived, the body never finishes.
|
||||||
|
pull() {
|
||||||
|
return new Promise<void>(() => {});
|
||||||
|
},
|
||||||
|
cancel() {
|
||||||
|
cancelled = true;
|
||||||
|
},
|
||||||
|
});
|
||||||
|
const wrapped = withOAuthRequestTimeout(
|
||||||
|
vi
|
||||||
|
.fn<typeof fetch>()
|
||||||
|
.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<Uint8Array>({
|
||||||
|
start(ctrl) {
|
||||||
|
ctrl.enqueue(new TextEncoder().encode("x"));
|
||||||
|
},
|
||||||
|
pull() {
|
||||||
|
return new Promise<void>(() => {});
|
||||||
|
},
|
||||||
|
cancel() {
|
||||||
|
throw new Error("cancel blew up");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
const wrapped = withOAuthRequestTimeout(
|
||||||
|
vi
|
||||||
|
.fn<typeof fetch>()
|
||||||
|
.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 () => {
|
it("hands back the original response, not a copy", async () => {
|
||||||
const original = new Response('{"issuer":"https://as.example.com"}', {
|
const original = new Response('{"issuer":"https://as.example.com"}', {
|
||||||
status: 201,
|
status: 201,
|
||||||
|
|
|
||||||
|
|
@ -319,6 +319,44 @@ function requestHeadersOf(
|
||||||
return new Headers();
|
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<Uint8Array> | undefined,
|
||||||
|
): Promise<void> {
|
||||||
|
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<Uint8Array> | undefined,
|
||||||
|
response: Response | undefined,
|
||||||
|
): void {
|
||||||
|
void reader?.cancel().catch(() => {});
|
||||||
|
void response?.body?.cancel().catch(() => {});
|
||||||
|
}
|
||||||
|
|
||||||
/** The request URL, whatever form `fetch`'s first argument took. */
|
/** The request URL, whatever form `fetch`'s first argument took. */
|
||||||
function requestUrlOf(input: RequestInfo | URL): string {
|
function requestUrlOf(input: RequestInfo | URL): string {
|
||||||
if (typeof input === "string") return input;
|
if (typeof input === "string") return input;
|
||||||
|
|
@ -466,11 +504,16 @@ export function withOAuthRequestTimeout(
|
||||||
const nextInit: RequestInit = { ...init, signal };
|
const nextInit: RequestInit = { ...init, signal };
|
||||||
REQUEST_DEADLINES.set(nextInit, budget);
|
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<Uint8Array> | undefined;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const response = await Promise.race([
|
const response = await Promise.race([
|
||||||
fetchFn(input, nextInit),
|
fetchFn(input, nextInit),
|
||||||
abandoned,
|
abandoned,
|
||||||
]);
|
]);
|
||||||
|
settled = response;
|
||||||
// `fetch` resolves once the response *headers* arrive, so stopping here
|
// `fetch` resolves once the response *headers* arrive, so stopping here
|
||||||
// would leave the body unbounded: an authorization server can send
|
// would leave the body unbounded: an authorization server can send
|
||||||
// headers, or half a JSON document, and then stall, and the caller's
|
// 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
|
// `content-length` wrong, so those had to be stripped — a rebuilt
|
||||||
// response was observably not the original in at least four ways. Not
|
// response was observably not the original in at least four ways. Not
|
||||||
// rebuilding removes the whole class.
|
// 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;
|
return response;
|
||||||
} catch (err) {
|
} 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*
|
// `controller.abort()` above can reject the underlying fetch *before*
|
||||||
// `reject` runs, in which case the race settles with undici's
|
// `reject` runs, in which case the race settles with undici's
|
||||||
// `AbortError` instead of ours. Which of the two wins is an ordering
|
// `AbortError` instead of ours. Which of the two wins is an ordering
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue