/** * 跨文档写入、上传和 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 前检查宿主身份等条件。 */ 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(); 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 也会归还名额。 */ return Object.freeze({ async run(signal: AbortSignal, callback: () => Promise): Promise { await enter(signal); try { signal.throwIfAborted();beforeRun?.();return await callback(); } finally { release(); } } }); }