356 lines
26 KiB
TypeScript
356 lines
26 KiB
TypeScript
/**
|
||
* 从 ledger 派生执行顺序的调度器：ready FIFO+queued 去重、依赖索引、canvas claims 与重试 min-heap。
|
||
* 所有权威变化通过 options.mutate 写回 Root；队列本身既不是历史，也不保存另一套任务状态。
|
||
* 发送前等待宿主持久化精确 body/key 和 dispatch 标记，结果协调及持久化成功后才释放占用。
|
||
* claims/holds 只约束本 scheduler；跨标签、进程或服务端的 writer/身份隔离由 engine 与 storage 端口负责。
|
||
*/
|
||
import type { Id, Json, OperationIntent } from '../canvas-core/types.ts';
|
||
import { fault, readMap } from '../canvas-core/values.ts';
|
||
import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } from './ledger.ts';
|
||
import { createDeque, createMinHeap } from './structures.ts';
|
||
|
||
/** 端口解释后的执行状态；blocked 表示需显式恢复，canonical 是尚待协调校验的回执数据。 */
|
||
export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json; readonly blocked?: boolean };
|
||
/** 当前可变意图编译出的 wire 计划，交 ledger.freeze 捕获后才成为不可更改的 FrozenRequest。 */
|
||
export type CompiledRequest = { readonly kind?: FrozenRequest['kind']; readonly ownerCanvasId?: Id; readonly runId?: Id; readonly protocolVersion?: number; readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>> };
|
||
/** 可注入时钟和 timer 端口，便于确定性测试重试/时钟变化；默认 now 使用 Date.now。 */
|
||
export type Clock = { now(): number; setTimer(callback: () => void, delay: number): unknown; clearTimer(handle: unknown): void };
|
||
/** 默认平台计时适配，定时器只负责唤醒，pump 仍检查当前账本和到期时间。 */
|
||
const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
|
||
/** mutate 回调内使用的临时可变账本接口，不直接暴露其内部表。 */
|
||
type Ledger = ReturnType<typeof createLedger>;
|
||
/** engine 提供准入、纯编译、原子账本更新、持久屏障、外部端口和完成协调。
|
||
* complete 必须连回执/基线/working/ledger 一起处理，不允许仅把请求先标成功。 */
|
||
export type SchedulerOptions = {
|
||
/** false 进入准入等待集合，由 refreshAdmission 唤醒；不在此把逻辑操作改为 blocked。 */
|
||
admit?: (intent:OperationIntent)=>boolean;
|
||
/** 每次准入/唤醒读取当前权威快照，不信任入队时捕获的旧状态。 */
|
||
getLedger(): LedgerSnapshot;
|
||
/** 在 engine 的受控发布路径内修改账本，失败不能留下半份持久事实。 */
|
||
mutate(reduce: (ledger: Ledger) => void): void;
|
||
/** 等待当前 Root 可靠存储；失败保持请求身份与占用，不退回生成新请求。
|
||
* scheduler 信任此 Promise 的持久承诺，本身不核对 durableGeneration/底层事务。 */
|
||
persist(): Promise<void>;
|
||
/** 为新冻结请求分配 ID；恢复/重试沿用原 ID，不再次调用此工厂。 */
|
||
createId(): Id;
|
||
/** 从当前意图推导需要 claim 的 canvas；编译结果必须覆盖同一集合。 */
|
||
participants(intent: OperationIntent): readonly Id[];
|
||
/** 同步编译一个意图；不在这里发送，payload/版本随后由 ledger.freeze 捕获。 */
|
||
compile(intent: OperationIntent, requestId: Id): CompiledRequest;
|
||
/** 将选中的连续候选缩为非空前缀，返回值必须是 1..intents.length 的整数。 */
|
||
batchLimit?: (intents: readonly OperationIntent[]) => number;
|
||
/** 可选抵消协调钩子：先实际标记 ledger，再返回已抵消的候选前缀 ID；调度器不自行构造证明。 */
|
||
elide?: (intents: readonly OperationIntent[]) => readonly Id[];
|
||
/** 多意图编译器仅在最终候选数 >1 时使用，单项仍走 compile。 */
|
||
compileBatch?: (intents: readonly OperationIntent[], requestId: Id) => CompiledRequest;
|
||
/** 使用保存的冻结请求发写端口；端口负责协议/权限与回执解释，不得重编 payload。 */
|
||
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||
/** 按原请求身份查结果，端口解释回执；端口抛错时 scheduler 统一记为 unresolved/transport_unknown。 */
|
||
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||
/** 恢复时确认冻结格式是否有解释器；省略钩子只默认支持 protocolVersion=1。 */
|
||
supportsRequest?: (request: FrozenRequest) => boolean;
|
||
/** 同步协调回执/base/working 与 ledger 的宿主边界；不得返回异步准备结果。 */
|
||
complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void;
|
||
/** 报告存储/协调/编译错误，scheduler 隔离此回调抛错，不能用它改写请求结果。 */
|
||
onError(error: unknown): void;
|
||
clock?: Clock;
|
||
concurrency?: number;
|
||
/** 按 submit attempt 提供有限退避毫秒；正常值取至少 1ms，端口 retryAfterMs 可覆盖此次值。 */
|
||
retryDelay?: (attempt: number) => number;
|
||
};
|
||
|
||
/** Queue/heap/claims are indexes only. Every authoritative transition lives in the ledger. */
|
||
/** 创建暂停态 scheduler；start 先恢复完整 claim 集，再允许执行任何新工作。 */
|
||
export function createScheduler(options: SchedulerOptions) {
|
||
const clock = options.clock ?? browserClock, concurrency = options.concurrency ?? 2;
|
||
if (!Number.isSafeInteger(concurrency) || concurrency < 1) throw fault('invalid_concurrency');
|
||
// 准入门禁等待者不进入忙轮询；加载/授权等条件变化时由 refreshAdmission 显式唤醒。
|
||
const admissionWaiters=new Set<Id>();
|
||
// FIFO 保存待处理逻辑 ID，queued 防重复；claims 是 canvas→request，多个参与者一次认领。
|
||
const ready = createDeque<Id>(), queued = new Set<Id>(), claims = new Map<Id, Id>();
|
||
// waiters 按被占 canvas 等待，dependents 按逻辑依赖等待，ordered 保留全局序列以挑连续批。
|
||
const waiters = new Map<Id, Set<Id>>(), dependents = new Map<Id, Set<Id>>(), indexed = new Set<Id>();
|
||
// 最早 dueAt 优先；同到期按入堆 sequence 稳定排序，token 让旧 timer/旧重试条目惰性失效。
|
||
const retry = createMinHeap<{ id: Id; dueAt: number; sequence: number; token: number }>((a, b) => a.dueAt - b.dueAt || a.sequence - b.sequence);
|
||
// active 跟踪整条异步流水线（包含发送前/后的持久化等待），不是仅统计正在进行的 HTTP 请求。
|
||
// 句柄/AbortController 属于调度器，任务完成事实仍必须回到 ledger。
|
||
const tokens = new Map<Id, number>(), active = new Map<Id, { controller: AbortController; promise: Promise<void>; kind: FrozenRequest['kind'] }>();
|
||
// 多个恢复/关闭流程可独立 hold；只有最后一个 token 释放才继续准入，已启动工作仍正常收尾。
|
||
const holds = new Set<symbol>();
|
||
const ordered: { id: Id; sequence: number }[] = [];
|
||
let scheduledPump: Promise<void> | undefined;
|
||
// 上传 lane 单并发等容量限制不能让到期请求反复立即入堆，先 deferred，active 完成再唤醒。
|
||
const laneWaiters = new Map<NonNullable<FrozenRequest['kind']>, Set<Id>>(), deferredRetries = new Map<Id, { id: Id; dueAt: number; sequence: number; token: number }>();
|
||
let timer: unknown, pumpTimer: unknown, enqueueSequence = 0, stopped = true, closed = false, pumping = false;
|
||
/** 默认指数退避封顶 30s，最少 1ms；端口可提供 retryAfterMs 覆盖此次等待。 */
|
||
const delay = (attempt: number) => Math.max(1, options.retryDelay?.(attempt) ?? Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)));
|
||
/** 隔离错误报告回调，日志/宿主提示失败不能改变已知回执状态。 */
|
||
const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot change a receipt. */ } };
|
||
/** 旧冻结协议必须有解释器才重试；缺钩子时仅默认支持协议 1。 */
|
||
const supported = (request: FrozenRequest) => options.supportsRequest?.(request) ?? request.protocolVersion === 1;
|
||
/** 总 active 上限之外，asset.upload 至多占一个 lane；其它类别可用剩余流水线执行位。 */
|
||
const canRun = (kind: FrozenRequest['kind']) => active.size < concurrency && (kind !== 'asset.upload' || ![...active.values()].some(task => task.kind === 'asset.upload'));
|
||
|
||
/** 先读 ledger 检查 pending/依赖/门禁，再用 queued 去重入 FIFO；不是通知一次就盲发一次。 */
|
||
function enqueue(id: Id) {
|
||
if (queued.has(id)) return;
|
||
const record = options.getLedger().operations.get(id);
|
||
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) {admissionWaiters.delete(id);return;}
|
||
if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);return;}
|
||
admissionWaiters.delete(id);queued.add(id); ready.push(id);
|
||
}
|
||
/** 新逻辑 ID 仅索引一次：二分插入 sequence 表、登记反向依赖，再尝试入队。 */
|
||
function index(id: Id) {
|
||
const record = options.getLedger().operations.get(id);
|
||
if (!record) return;
|
||
if (!indexed.has(id)) {
|
||
indexed.add(id);
|
||
let lo = 0, hi = ordered.length;
|
||
while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence < record.intent.sequence) lo = mid + 1; else hi = mid; }
|
||
ordered.splice(lo, 0, { id, sequence: record.intent.sequence });
|
||
for (const dependency of record.intent.dependencies) {
|
||
const set = dependents.get(dependency) ?? new Set<Id>(); set.add(id); dependents.set(dependency, set);
|
||
}
|
||
}
|
||
enqueue(id);
|
||
}
|
||
/** Only a contiguous mutable document prefix may share a request. Task barriers stay visible. */
|
||
/** 只选连续未冻结 document.change 前缀，最多 64 意图；追加候选时不让累计 changes 超过 10000。
|
||
* 首意图本身超过 10000 时仍可单独返回，不把它拆成字段请求；遇任务屏障/占用即停。
|
||
* 仅合并有相交参与者的 lane；独立 canvas 保留各自失败/重试边界。 */
|
||
function batch(first: OperationIntent, canvases: Set<Id>): readonly OperationIntent[] {
|
||
const selected = [first], ids = new Set([first.id]);
|
||
if ((!options.compileBatch && !options.elide) || first.operationType !== 'document.change' || first.operationSchemaVersion !== 1) return selected;
|
||
let lo = 0, hi = ordered.length, changeCount = first.changes.length;
|
||
while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence <= first.sequence) lo = mid + 1; else hi = mid; }
|
||
const ledger = options.getLedger();
|
||
for (let i = lo; i < ordered.length && selected.length < 64; i++) {
|
||
const record = ledger.operations.get(ordered[i].id);
|
||
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.operationType !== 'document.change' || record.intent.operationSchemaVersion !== 1) break;
|
||
if (record.intent.dependencies.some(id => !ids.has(id) && ledger.operations.get(id)?.phase !== 'resolved')) break;
|
||
if(options.admit&&!options.admit(record.intent))break;
|
||
const scope = options.participants(record.intent);
|
||
// Independent canvas lanes keep independent failure and retry boundaries.
|
||
if (!scope.length || !scope.some(id => canvases.has(id)) || scope.some(id => claims.has(id)) || changeCount + record.intent.changes.length > 10_000) break;
|
||
selected.push(record.intent); ids.add(record.intent.id); changeCount += record.intent.changes.length;
|
||
for (const id of scope) canvases.add(id);
|
||
}
|
||
return selected;
|
||
}
|
||
/** 支持合批时延到微任务，让同轮连续编辑有机会聚合；其它情况直接 pump。 */
|
||
function requestPump() {
|
||
if (!options.compileBatch && !options.elide) { pump(); return; }
|
||
if (!scheduledPump) scheduledPump = Promise.resolve().then(() => { scheduledPump = undefined; pump(); });
|
||
}
|
||
/** 先检查全部 canvas 无冲突，再统一设置 claims，避免持有一部分后等待另一部分。 */
|
||
function own(request: FrozenRequest) {
|
||
for (const canvas of request.canvasIds) {
|
||
const prior = claims.get(canvas);
|
||
if (prior && prior !== request.id) throw fault('recovery_claim_conflict', [canvas]);
|
||
}
|
||
for (const canvas of request.canvasIds) claims.set(canvas, request.id);
|
||
}
|
||
/** 释放仍属于原 request 的 claims，并唤醒对应 canvas 等待者及逻辑依赖者。 */
|
||
function release(request: FrozenRequest) {
|
||
for (const canvas of request.canvasIds) if (claims.get(canvas) === request.id) {
|
||
claims.delete(canvas);
|
||
const waiting = waiters.get(canvas); waiters.delete(canvas);
|
||
for (const id of waiting ?? []) enqueue(id);
|
||
}
|
||
for (const id of request.operationIds) for (const dependent of dependents.get(id) ?? []) enqueue(dependent);
|
||
}
|
||
/** 可选 mutate 写回 retry 元数据，再用新 token 入 heap；persist=false 仅表示不重复写账本。
|
||
* 此处没有 await 存储屏障，真正可靠落盘仍由 launch/resume 的 options.persist 完成。 */
|
||
function schedule(id: Id, dueAt: number, action: 'submit' | 'query', persist = true) {
|
||
if (persist) options.mutate(ledger => ledger.scheduleRetry(id, dueAt, action));
|
||
const token = (tokens.get(id) ?? 0) + 1; tokens.set(id, token);
|
||
retry.push({ id, dueAt, sequence: ++enqueueSequence, token }); armTimer();
|
||
}
|
||
/** 忽略已被新计划替代或已解决请求的旧 heap 条目。 */
|
||
function validRetry(item: { id: Id; token: number }) {
|
||
return tokens.get(item.id) === item.token && options.getLedger().requests.get(item.id)?.phase !== 'resolved';
|
||
}
|
||
/** 仅为最早有效重试设一个 timer；暂停、hold 或容量满时不安排无效唤醒。 */
|
||
function armTimer() {
|
||
if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; }
|
||
while (retry.peek() && !validRetry(retry.peek()!)) retry.pop();
|
||
if (stopped || holds.size || active.size >= concurrency || !retry.peek()) return;
|
||
timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now()));
|
||
}
|
||
/** 安装 active 句柄后在微任务中执行：submit 先标 dispatch；query 保留现有发送风险，再共同等待持久化。
|
||
* 调端口→同步 complete→再次持久化；hold 不撤回已启动流水线，stop 则让晚到结果不再采纳。 */
|
||
function launch(id: Id, action: 'submit' | 'query') {
|
||
if (active.has(id) || stopped || holds.size) return;
|
||
const controller = new AbortController();
|
||
// Start in a microtask so the handle exists before any synchronous adapter error.
|
||
const promise = Promise.resolve().then(async () => {
|
||
try {
|
||
if (stopped) return;
|
||
if (action === 'submit') options.mutate(ledger => ledger.markDispatch(id));
|
||
// This barrier includes the exact body/key and the conservative dispatch marker.
|
||
// 请求身份与保守发送标记在这里可靠存储；失败就不接触外部端口，便于重开查询原请求。
|
||
await options.persist();
|
||
if (stopped) return;
|
||
const request = options.getLedger().requests.get(id)!;
|
||
let status: OperationStatus;
|
||
try { status = await (action === 'submit' ? options.submit(request, controller.signal) : options.query(request, controller.signal)); }
|
||
catch { status = { outcome: 'unresolved', error: 'transport_unknown' }; }
|
||
if (stopped) return; // A stale identity/runtime cannot accept a late callback.
|
||
// complete 可接受执行证据但阻塞后续协调；不能把网络 2xx 本身等同于本地基线已完成。
|
||
const completion = options.complete(request, status);
|
||
const current = options.getLedger().requests.get(id)!;
|
||
if (current.phase === 'resolved') { tokens.delete(id); await options.persist(); if (!stopped) release(current); }
|
||
else if (completion?.blocked || status.blocked) { await options.persist(); }
|
||
else {
|
||
// A missing receipt is unknown, so the only write retry reuses this frozen request.
|
||
// 首次 submit 未知先 query；query 仍未知才原样重试 submit；accepted 一直查，不换请求身份。
|
||
// accepted 的查询/submit 都继续 query；其它 submit 结果先 query，非 accepted 的 query 结果才允许同身份 submit。
|
||
const nextAction = status.outcome === 'accepted' || action === 'submit' ? 'query' : 'submit';
|
||
schedule(id, clock.now() + Math.max(1, status.retryAfterMs ?? delay(current.attempt)), nextAction);
|
||
await options.persist();
|
||
}
|
||
} catch (error) { report(error); } // Storage/reconciliation failure keeps its claims held.
|
||
}).finally(() => {
|
||
// 真实句柄结束后归还容量，重入 deferred retries/可用 lane，再 pump 独立工作。
|
||
active.delete(id);
|
||
if (!stopped) {
|
||
for (const item of deferredRetries.values()) retry.push(item);
|
||
deferredRetries.clear();
|
||
for (const [kind, waiting] of laneWaiters) if (canRun(kind)) { for (const operationId of waiting) enqueue(operationId); laneWaiters.delete(kind); }
|
||
pump();
|
||
}
|
||
});
|
||
active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind });
|
||
}
|
||
/** 防同步重入地处理到期请求，再从本轮 ready 前缀最多取 64 项；retry 循环没有同样的 64 项上限。
|
||
* 等待者移到索引，ready 尚有余量且有容量时安排下一轮 timer，不代表总队列大小有上限。 */
|
||
function pump() {
|
||
if (stopped || holds.size || pumping) return;
|
||
pumping = true;
|
||
try {
|
||
// 先处理到期恢复请求；上传 lane 被占时暂存，避免它阻塞独立文档请求。
|
||
while (active.size < concurrency && retry.peek() && retry.peek()!.dueAt <= clock.now()) {
|
||
const item = retry.pop()!;
|
||
if (!validRetry(item)) continue;
|
||
const request = options.getLedger().requests.get(item.id);
|
||
if (request && !active.has(item.id)) {
|
||
if (!canRun(request.kind)) { deferredRetries.set(item.id, item); continue; }
|
||
launch(item.id, request.retryAction ?? 'query');
|
||
}
|
||
}
|
||
let budget = Math.min(ready.size, 64);
|
||
while (active.size < concurrency && budget-- > 0) {
|
||
const id = ready.pop(); if (!id) break;
|
||
queued.delete(id);
|
||
const record = options.getLedger().operations.get(id);
|
||
if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
|
||
try {
|
||
if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);continue;}
|
||
if (!canRun(record.intent.operationType)) {
|
||
const set = laneWaiters.get(record.intent.operationType) ?? new Set<Id>(); set.add(id); laneWaiters.set(record.intent.operationType, set); continue;
|
||
}
|
||
const canvases = [...new Set(options.participants(record.intent))].sort();
|
||
if (!canvases.length && record.intent.operationType !== 'asset.upload' && record.intent.operationType !== 'run.cancel') throw fault('missing_participants');
|
||
// 同时获取全部参与者，无“先锁 A 再等 B”；占用冲突时只登记等待，不持有局部 claim。
|
||
const occupied = canvases.filter(canvas => claims.has(canvas));
|
||
if (occupied.length) {
|
||
for (const canvas of occupied) { const set = waiters.get(canvas) ?? new Set<Id>(); set.add(id); waiters.set(canvas, set); }
|
||
continue;
|
||
}
|
||
const scope = new Set(canvases), selected = batch(record.intent, scope);
|
||
const limit = options.batchLimit?.(selected) ?? selected.length;
|
||
if (!Number.isSafeInteger(limit) || limit < 1 || limit > selected.length) throw fault('invalid_batch_limit');
|
||
const candidates = selected.slice(0, limit);
|
||
// 可证明无外部效果的闭合前缀可本地抵消；必须核对实际 ledger 已标 elided，不能只信返回 ID。
|
||
const elided = options.elide?.(candidates) ?? [];
|
||
if (elided.length) {
|
||
if (elided.some((id, index) => id !== candidates[index]?.id || options.getLedger().operations.get(id)?.resolution !== 'elided')) throw fault('invalid_elision');
|
||
for (const id of elided) for (const dependent of dependents.get(id) ?? []) enqueue(dependent);
|
||
continue;
|
||
}
|
||
const intents = options.compileBatch ? candidates : [record.intent];
|
||
const participants = [...new Set(intents.flatMap(intent => options.participants(intent)))].sort();
|
||
// 编译参与者必须与实际 intents 推导的集合一致，随后 freeze/own/launch；没有 compileBatch 时只编译首意图。
|
||
// 异常走原逻辑 ID 的 block；若已 frozen，ledger.block 不会把它退回 pending，冻结身份仍保留。
|
||
const requestId = options.createId(), compiled = intents.length > 1 ? options.compileBatch!(intents, requestId) : options.compile(record.intent, requestId);
|
||
if (participants.length !== compiled.canvasIds.length || participants.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch');
|
||
options.mutate(ledger => ledger.freeze({ ...compiled, id: requestId, operationIds: intents.map(intent => intent.id) }));
|
||
own(options.getLedger().requests.get(requestId)!); launch(requestId, 'submit');
|
||
} catch (error) { options.mutate(ledger => ledger.block(id, [(error as Error).message || 'compile_failed'])); report(error); }
|
||
}
|
||
} catch(error) {
|
||
// A retired identity can reject both the attempt and its blocked-state write.
|
||
// Report the failure without rejecting an unobserved scheduled pump promise.
|
||
report(error);
|
||
} finally {
|
||
pumping = false; armTimer();
|
||
if (!stopped && !holds.size && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0);
|
||
}
|
||
}
|
||
return Object.freeze({
|
||
/** 首次启动（已运行时幂等）：先恢复全部未解决 claims，之后才开始端口工作；claim 冲突会在仍 stopped 时抛错。
|
||
* 解释器缺失保留占用并写 blocker；此步骤不做完整恢复 DTO 校验，也不从单个错误推断请求未执行。 */
|
||
start() {
|
||
if (closed) throw fault('scheduler_closed');
|
||
if (!stopped) return;
|
||
// Validate the entire recovered claim set before sending anything.
|
||
claims.clear();
|
||
for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') own(request);
|
||
stopped = false;
|
||
for (const operation of [...options.getLedger().operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) index(operation.intent.id);
|
||
for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') {
|
||
if (!supported(request)) { options.mutate(ledger => ledger.blockRequest(request.id, 'request_interpreter_unavailable')); continue; }
|
||
schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? (request.phase === 'frozen' ? 'submit' : 'query'), false);
|
||
}
|
||
pump();
|
||
},
|
||
/** 门禁事实变化后显式重试准入，未满足者继续等待，不制造定时器忙轮询。 */
|
||
refreshAdmission(){for(const id of [...admissionWaiters])enqueue(id);if(ready.size)requestPump();},
|
||
/** 新逻辑操作通知推进派生索引；正在 queued 时不重复入队，indexed 让依赖只注册一次。
|
||
* 已出队的 pending 操作后续 notify 可再次尝试准入；通知本身不新增 ledger 意图。 */
|
||
notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); requestPump(); },
|
||
/** Acquire synchronously, then settle(): already admitted work completes with its original identity. */
|
||
/** 同步阻止新准入并清唤醒 timer；返回幂等释放函数，已准入异步工作不被撤销或换身份。 */
|
||
hold() {
|
||
if (closed) throw fault('scheduler_closed');
|
||
const token = Symbol(); holds.add(token);
|
||
if (timer !== undefined) clock.clearTimer(timer);
|
||
if (pumpTimer !== undefined) clock.clearTimer(pumpTimer);
|
||
timer = pumpTimer = undefined;
|
||
return () => {
|
||
if (!holds.delete(token) || holds.size || stopped) return;
|
||
pump();
|
||
};
|
||
},
|
||
/** Explicit retry after storage/auth/permission recovery; preserves frozen bodies. */
|
||
/** 存储/授权恢复后的显式重试：先补持久化，再释放已解决 claims 或继续查原请求。 */
|
||
async resume() {
|
||
if (stopped) return;
|
||
// A receipt may be in memory after its durable write failed. Keep claims until retry succeeds.
|
||
await options.persist();
|
||
if (stopped) return;
|
||
for (const request of options.getLedger().requests.values()) {
|
||
if (active.has(request.id)) continue;
|
||
if (request.phase === 'resolved') release(request);
|
||
else if (supported(request)) schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? 'query', false);
|
||
}
|
||
for (const id of indexed) enqueue(id);
|
||
pump();
|
||
},
|
||
/** 返回当前占用表的只读副本，调用者不能手改 claim 绕过未知请求。 */
|
||
getClaims: () => readMap(claims),
|
||
/** 等待当前微任务/pump/active 收尾，不主动等待未来 retry timer，因此不是“全部已同步”保证。
|
||
* 端口若长期不完成（包括忽略 abort），active Promise 仍可让 settle 一直等待。 */
|
||
async settle() {
|
||
do {
|
||
if (scheduledPump) await scheduledPump;
|
||
if (active.size) await Promise.all([...active.values()].map(item => item.promise));
|
||
if (!stopped && !holds.size && pumpTimer !== undefined) { clock.clearTimer(pumpTimer); pumpTimer = undefined; pump(); }
|
||
} while (scheduledPump || active.size || (!stopped && !holds.size && pumpTimer !== undefined));
|
||
},
|
||
/** 永久关闭本 scheduler、取消 timer 并请求 abort active；晚到响应因 stopped 不再采纳，不能再 start。
|
||
* abort 不证明外部未执行，也不清 ledger/claims/派生容器或替 engine 关闭存储；活动句柄在 finally 收尾。 */
|
||
stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); if (pumpTimer !== undefined) clock.clearTimer(pumpTimer); timer = pumpTimer = undefined; for (const item of active.values()) item.controller.abort(); },
|
||
});
|
||
}
|