53 lines
3.6 KiB
TypeScript
53 lines
3.6 KiB
TypeScript
/** Separate the response-header deadline from a large snapshot's transfer.
|
|
* A client disconnect still aborts the upstream; bodies are never cached. */
|
|
export type CanvasTransferTrace = { started_at: string; finished_at: string; headers_ms: number; first_chunk_ms?: number; transfer_ms: number; total_ms: number; bytes: number; chunks: number; outcome: "complete" | "cancelled" | "error"; error?: string };
|
|
export async function fetchCanvasResponse(url: string, options: RequestInit, limits = { headersMs: 15000, bodyMs: 90000, idleMs: 30000 }, request = fetch, report?: (trace: CanvasTransferTrace) => void) {
|
|
const started = performance.now();
|
|
const startedAt = new Date().toISOString();
|
|
const abort = new AbortController();
|
|
const signal = options.signal ? AbortSignal.any([options.signal, abort.signal]) : abort.signal;
|
|
const timeout = () => abort.abort(new DOMException("Canvas transfer timed out", "TimeoutError"));
|
|
let timer = setTimeout(timeout, limits.headersMs);
|
|
let upstream: Response;
|
|
try { upstream = await request(url, { ...options, signal }); }
|
|
catch (error) {
|
|
clearTimeout(timer);
|
|
const elapsed = Math.round((performance.now() - started) * 10) / 10;
|
|
report?.({ started_at: startedAt, finished_at: new Date().toISOString(), headers_ms: elapsed, transfer_ms: 0, total_ms: elapsed, bytes: 0, chunks: 0, outcome: "error", ...(error instanceof Error ? { error: error.name } : {}) });
|
|
throw error;
|
|
}
|
|
clearTimeout(timer);
|
|
if (!upstream.body) return upstream;
|
|
const headersAt = performance.now();
|
|
const reader = upstream.body.getReader();
|
|
let bytes = 0, chunks = 0, firstChunk: number | undefined;
|
|
const record = (outcome: CanvasTransferTrace["outcome"], error?: unknown) => {
|
|
const now = performance.now(), rounded = (value: number) => Math.round(value * 10) / 10;
|
|
report?.({ started_at: startedAt, finished_at: new Date().toISOString(), headers_ms: rounded(headersAt - started), ...(firstChunk === undefined ? {} : { first_chunk_ms: rounded(firstChunk - headersAt) }), transfer_ms: rounded(now - headersAt), total_ms: rounded(now - started), bytes, chunks, outcome, ...(error instanceof Error ? { error: error.name } : {}) });
|
|
};
|
|
let idle: ReturnType<typeof setTimeout> | undefined;
|
|
let finished = false;
|
|
const cleanup = () => { finished = true; clearTimeout(timer); clearTimeout(idle); signal.removeEventListener("abort", interrupted); };
|
|
let interrupt: (reason: unknown) => void;
|
|
const interrupted = () => interrupt(signal.reason);
|
|
const body = new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
interrupt = (reason) => { if (finished) return; cleanup(); record("error", reason); controller.error(reason); void reader.cancel(reason).catch(() => {}); };
|
|
signal.addEventListener("abort", interrupted, { once: true });
|
|
timer = setTimeout(timeout, limits.bodyMs);
|
|
if (signal.aborted) interrupted();
|
|
},
|
|
async pull(controller) {
|
|
if (finished) return;
|
|
idle = setTimeout(timeout, limits.idleMs);
|
|
try {
|
|
const next = await reader.read(); clearTimeout(idle);
|
|
if (finished) return;
|
|
if (next.done) { cleanup(); record("complete"); controller.close(); } else { firstChunk ??= performance.now(); bytes += next.value.byteLength; chunks++; controller.enqueue(next.value); }
|
|
} catch (error) { if (!finished) { cleanup(); record("error", error); controller.error(error); } }
|
|
},
|
|
async cancel(reason) { if (!finished) { cleanup(); record("cancelled", reason); } abort.abort(reason); await reader.cancel(reason).catch(() => {}); },
|
|
});
|
|
return new Response(body, { status: upstream.status, statusText: upstream.statusText, headers: upstream.headers });
|
|
}
|