73 lines
5.7 KiB
TypeScript
73 lines
5.7 KiB
TypeScript
/**
|
||
* 画布响应的流转发与超时分层：响应头截止、整个body时限、单次等待下一块的空闲时限分别管理。
|
||
* 不把大body一次读入内存或缓存，返回保留状态/头的新Response；外部断开与内部超时通过组合signal中止上游。
|
||
*/
|
||
/** Separate the response-header deadline from a large snapshot's transfer.
|
||
* A client disconnect still aborts the upstream; bodies are never cached. */
|
||
/** 单次传输诊断：头耗时、头后首块等待、body/总耗时与字节/块数；记录结果不是协议ACK。 */
|
||
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 };
|
||
/** 默认头15秒、body90秒、单次read30秒；request/report可注入。
|
||
* report由调用方提供且应避免抛错；此处没有把报告异常完全隔离成独立旁路。 */
|
||
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();
|
||
// 外部请求取消与内部超时任一生效都中止同一上游fetch，避免客户端已离开却继续传输。
|
||
const signal = options.signal ? AbortSignal.any([options.signal, abort.signal]) : abort.signal;
|
||
/** 三种时限共用内部取消源；只发送取消信号，实际唤醒等待依赖 request/reader 遵守取消契约。 */
|
||
const timeout = () => abort.abort(new DOMException("Canvas transfer timed out", "TimeoutError"));
|
||
// 初始计时器只约束拿响应头，取得Response后清掉，再给body设置独立总时限。
|
||
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);
|
||
// 无body的响应直接透传，不建立reader，也不产生body完成轨迹。
|
||
if (!upstream.body) return upstream;
|
||
const headersAt = performance.now();
|
||
const reader = upstream.body.getReader();
|
||
let bytes = 0, chunks = 0, firstChunk: number | undefined;
|
||
/** 从性能时钟计算耗时，四舍五入到0.1ms；首块耗时相对headersAt，不相对最初发起时刻。 */
|
||
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;
|
||
/** 标记结束并清除总/空闲计时与组合signal监听，后续迟到read不能再次enqueue或重复结束。
|
||
* 这里只清理调度资源，不 releaseLock；reader 的取消在中断/下游 cancel 路径另行执行。 */
|
||
const cleanup = () => { finished = true; clearTimeout(timer); clearTimeout(idle); signal.removeEventListener("abort", interrupted); };
|
||
let interrupt: (reason: unknown) => void;
|
||
/** 将组合signal的取消原因交给流start中安装的中断处理器。 */
|
||
const interrupted = () => interrupt(signal.reason);
|
||
const body = new ReadableStream<Uint8Array>({
|
||
/** 流开始即启动body总时限，消费者背压暂停不暂停该总时限；中断使下游报错并尽力取消reader。 */
|
||
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();
|
||
},
|
||
/** 按下游需求读下一块，idle只覆盖本次reader.read等待；consumer不pull时不运行idle计时。 */
|
||
async pull(controller) {
|
||
if (finished) return;
|
||
idle = setTimeout(timeout, limits.idleMs);
|
||
try {
|
||
const next = await reader.read(); clearTimeout(idle);
|
||
// read等待期间可能已取消/超时；结果回来后先查finished，不向已关闭controller追加字节。
|
||
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); } }
|
||
},
|
||
/** 下游主动取消记录cancelled，并中止fetch和reader；已结束状态不重复记录轨迹。 */
|
||
async cancel(reason) { if (!finished) { cleanup(); record("cancelled", reason); } abort.abort(reason); await reader.cancel(reason).catch(() => {}); },
|
||
});
|
||
// 透传HTTP状态和头，业务JSON/NDJSON校验由调用方执行，不在代理流中重新编码内容。
|
||
return new Response(body, { status: upstream.status, statusText: upstream.statusText, headers: upstream.headers });
|
||
}
|