/** * 跨文档写入、上传和 run 读取共享的并发闸门。 * 等待项按 FIFO 排队,AbortSignal 取消等待;运行中的网络调用仍由 callback 自己响应 signal。 * 闸门只控制容量,不保存 operation/request 身份,也不能代替 ledger 的依赖和结果协调。 */ import { fault } from '../canvas-core/values.ts'; import { createDeque } from './structures.ts'; /** Shared bounded transport capacity for writes, uploads and run reads. */ /** 建立正整数容量的 gate,beforeRun 在实际执行 callback 前同步检查宿主身份等条件。 * 每个实例独立计数,等待 FIFO 无长度上限;这里没有跨实例锁、重试或持久任务记录。 * beforeRun 是同步钩子;异步身份校验应由 callback 自己 await,不能依赖 void 钩子等待。 */ export function createNetworkGate(limit = 2, beforeRun?:()=>void) { if(!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_concurrency'); let active = 0; const waiting = createDeque<{ signal: AbortSignal; grant(): void; cancelled: boolean }>(); /** 有空位直接占用,否则等待 grant;已取消的等待项标记保留,释放时惰性跳过。 */ async function enter(signal: AbortSignal) { signal.throwIfAborted(); // enter 自己推进 active,或等 release 直接转交名额;等待者获准时不再重复 active++。 if (active < limit) { active++; return; } await new Promise((resolve,reject) => { // grant 解除 abort 监听再唤醒;被唤醒后 run 还会再次检查 signal,避免取消竞态执行请求。 const item = { signal, cancelled: false, grant() { signal.removeEventListener('abort',abort);resolve(); } }; /** 取消只拒绝此等待者,不占用/释放活跃名额;队列节点由 release 跳过。 */ const abort = () => { item.cancelled=true;signal.removeEventListener('abort',abort);reject(signal.reason); }; signal.addEventListener('abort',abort,{once:true});waiting.push(item); }); } /** 优先把原名额直接交给下一有效等待者;无人等待时才 active--,避免超发容量。 */ function release() { let item=waiting.pop();while(item?.cancelled)item=waiting.pop(); if(item)item.grant();else active--; } /** enter 成功后即使 signal/beforeRun/callback 抛错,finally 也会归还名额。 * 排队取消会拒绝 enter;执行中取消不自动结束 callback 或提早释放,仍等其 Promise 结束。 * 身份只在准入时检查一次;任务运行中的失效/响应采纳校验由 callback 和上层协调。 */ return Object.freeze({ async run(signal: AbortSignal, callback: () => Promise): Promise { await enter(signal); try { signal.throwIfAborted();beforeRun?.();return await callback(); } finally { release(); } } }); }