From 56a9806bc053ac4e5d18b50bcd7c009bd9fbcad6 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 12:44:51 +0800 Subject: [PATCH] docs(runtime): clarify ledger authority and scheduler durability boundaries --- src/lib/canvas-runtime/ledger.ts | 38 +++++++++++++------- src/lib/canvas-runtime/scheduler.ts | 53 ++++++++++++++++++++-------- src/lib/canvas-runtime/structures.ts | 6 ++-- 3 files changed, 68 insertions(+), 29 deletions(-) diff --git a/src/lib/canvas-runtime/ledger.ts b/src/lib/canvas-runtime/ledger.ts index f6d989b..9f4f349 100644 --- a/src/lib/canvas-runtime/ledger.ts +++ b/src/lib/canvas-runtime/ledger.ts @@ -26,14 +26,17 @@ export type FrozenRequest = { }; /** 意图及其本地解析状态；blocked 不等于丢弃，elided/retired 不等于收到服务器 applied 回执。 */ export type OperationRecord = { readonly intent: OperationIntent; readonly phase: 'pending' | 'frozen' | 'resolved' | 'blocked'; readonly requestId?: Id; readonly blockers: readonly string[]; readonly resolution?: 'applied' | 'not-applied' | 'superseded' | 'discarded' | 'satisfied' | 'elided'; readonly elisionId?: Id; readonly elision?: ElisionProof; readonly recoveryId?: Id; readonly replacementId?: Id; readonly basePresence: Readonly> }; -/** generation 随账本事实改变递增，durableGeneration 仅确认已实际存储的前缀。 */ +/** generation 随账本事实改变递增；durableGeneration 是宿主报告的已存前缀。 + * markDurable 只检查水位范围，本模块不访问存储或证明 fsync/事务真的完成。 */ export type LedgerSnapshot = { readonly generation: number; readonly durableGeneration: number; readonly operations: ReadonlyMap; readonly requests: ReadonlyMap }; /** A field-safe compressor for a wholly unsent segment. Frozen records never enter here. */ -/** 仅压缩完全未冻结的变化段：create 后字段修改并入新对象，同字段连续变化合并。 - * preserveLifecycle 保留创建/删除，preserveChecks 保留条件 no-op；不能用最后整节点覆盖替代字段链。 */ +/** 用于调用方已确认完全未冻结的变化段：合并 create 后字段修改及同字段链。 + * 本函数不读 ledger、basePresence 或 possiblyApplied，不能单独证明可抵消已发送/未知结果的生命周期。 + * preserveLifecycle 只阻止 create→delete 抵消，preserveChecks 只保留合并后字段 no-op；不是停用所有压缩。 */ export function compressChanges(changes: readonly Change[], options: { preserveLifecycle?: boolean; preserveChecks?: boolean } = {}): readonly Change[] { - // positions 按元素/canvas ID 保存结果索引栈，只尝试合并同对象最近变化，保留原条件顺序。 + // positions 按元素/canvas ID 保存索引栈，只尝试同对象最近一项；其它对象的变化可以夹在两项之间。 + // 若同对象中间出现不同字段，则不跨过它继续向前找同字段，因此不是任意字段的全段最后值覆盖。 const result: (Change | undefined)[] = [], positions = new Map(); for (const change of changes) { const id = change.kind === 'field' ? change.target.id : change.id; @@ -57,7 +60,8 @@ export function compressChanges(changes: readonly Change[], options: { preserveL return freeze(result.filter((change): change is Change => change !== undefined)); } -/** 从已有快照复制可变表，所有有效修改后重新捕获只读快照；旧 Snapshot 的成员不会改变。 */ +/** 从已有快照复制 Map 成员，修改后发布新的只读表视图；不在此验证完整恢复 DTO。 + * 初始记录沿用引用，宿主须先解码捕获为不可变记录；构造器不是任意外部对象的深隔离入口。 */ export function createLedger(initial?: LedgerSnapshot) { let generation = initial?.generation ?? 0, durableGeneration = initial?.durableGeneration ?? 0; const operations = new Map(initial?.operations), requests = new Map(initial?.requests); @@ -84,7 +88,8 @@ export function createLedger(initial?: LedgerSnapshot) { return Object.freeze({ /** 读取最新捕获的只读事实，不暴露内部可变表。 */ getSnapshot: () => cached, - /** 校验身份/序列/基线与依赖，再添加 pending；同 ID 同意图幂等，改意图则拒绝。 */ + /** 检查单条身份、正安全整数序列、基线枚举和依赖，再添加 pending。 + * 不在此证明全账本 sequence 唯一/递增；同 ID 同意图直接返回，也不会更新第一次的 basePresence。 */ add(intent: OperationIntent, basePresence: Readonly>) { if (!intent.id || !Number.isSafeInteger(intent.sequence) || intent.sequence < 1 || new Set(intent.dependencies).size !== intent.dependencies.length || Object.values(basePresence).some(value => !['absent', 'exists', 'unknown'].includes(value))) throw fault('invalid_operation'); const existing = operations.get(intent.id); @@ -99,8 +104,9 @@ export function createLedger(initial?: LedgerSnapshot) { }, /** 给可变意图设置/清除 blocker；有原因变 blocked，无原因回 pending，不改冻结请求。 */ block(id: Id, reasons: readonly string[]) { const operation = operations.get(id); if (!operation || !['pending', 'blocked'].includes(operation.phase)) return; operations.set(id, freeze({ ...operation, phase: reasons.length ? 'blocked' : 'pending', blockers: [...reasons] })); changed(); }, - /** 原子绑定多个 ready 意图到同一请求；参与者版本精确覆盖、类型一致、依赖已解决或在同批。 - * 只在这里序列化 payload 一次，后续 submit/query 都使用保存的 body 和请求 ID。 */ + /** 完成输入/ready 检查后，同次更新请求和所有操作的绑定：参与版本覆盖、类型一致，依赖已解决或在同批。 + * payload 在这里 JSON 序列化一次，后续沿用 body/ID；wire 语义、意图到参与者的推导由编译/调度层核对。 + * 这次内存更新本身不等于一次持久事务，也不自动取得 scheduler claims。 */ freeze(input: { id: Id; kind?: FrozenRequest['kind']; ownerCanvasId?: Id; runId?: Id; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly>; payload: Json }) { if (!input.id || requests.has(input.id) || !Number.isSafeInteger(input.protocolVersion ?? 1) || (input.protocolVersion ?? 1) < 1 || !input.operationIds.length || (!input.canvasIds.length && input.kind!=='asset.upload' && input.kind!=='run.cancel') || !validJson(input.payload) || new Set(input.operationIds).size !== input.operationIds.length || new Set(input.canvasIds).size !== input.canvasIds.length || Object.keys(input.expectedVersions).length !== input.canvasIds.length || input.canvasIds.some(id => !id || !Number.isSafeInteger(input.expectedVersions[id]) || input.expectedVersions[id] < 0)) throw fault('invalid_frozen_request'); if (input.kind === 'run.start' && (!input.ownerCanvasId || !input.canvasIds.includes(input.ownerCanvasId))) throw fault('invalid_run_request'); @@ -122,11 +128,13 @@ export function createLedger(initial?: LedgerSnapshot) { if (request.phase === 'resolved') throw fault('request_resolved', [id]); const next: FrozenRequest = freeze({ ...request, phase: 'dispatch-marked', attempt: request.attempt + 1, retryAt: undefined, retryAction: undefined }); requests.set(id, next); changed(); return next; }, - /** 记录执行结果；applied 且 reconciled 才解除逻辑操作，受理/未协调成功仍 waiting。 + /** 接受上层已解释的 Outcome；本函数不验证协议身份、canonical 模型或 not-applied 的零写入证据。 + * applied 且 reconciled 才解除逻辑操作，受理/未协调成功仍 waiting；reconciled 默认 true，调用方需按协调状态传入。 * not-applied 结束旧请求但阻塞原意图，不能自动换 key 当作用户已重新授权执行。 */ result(id: Id, outcome: Outcome, error?: string, reconciled = true) { const request = requireRequest(id); if ((request.phase === 'resolved' || request.outcome === 'applied') && request.outcome !== outcome) throw fault('receipt_conflict', [id]); + // 这里只拒绝与既有 applied/resolved outcome 不同的回执；相同 outcome 的阶段准入仍由调用链维持。 const phase = (outcome === 'applied' && reconciled) || outcome === 'not-applied' ? 'resolved' : 'waiting'; requests.set(id, freeze({ ...request, phase, outcome, error, retryAt: undefined, retryAction: undefined })); if (outcome === 'applied' && reconciled) for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'resolved', resolution: 'applied' })); @@ -161,7 +169,8 @@ export function createLedger(initial?: LedgerSnapshot) { }, /** Local retirement is not an applied receipt. Validate the complete closure before mutating. */ /** 用户恢复决策的本地退休：先验证整组与依赖闭包，再标记 superseded/discarded/satisfied。 - * 上传和执行结果未知的请求不能在这里被静默丢弃；replacement ID 必须是全新的。 */ + * 上传和执行结果未知的请求不能在这里被静默丢弃；replacement ID 不能撞已有/本批操作。 + * 只标记原操作退休，不创建 replacement，不改工作文档/history，须由 engine 在同一 Root 组织其它修改。 */ retire(recoveryId:Id,decisions:readonly {operationId:Id;resolution:'superseded'|'discarded'|'satisfied';replacementId?:Id}[]) { const ids=new Set(decisions.map(decision=>decision.operationId)); if(!recoveryId||ids.size!==decisions.length||new Set(decisions.flatMap(decision=>decision.replacementId?[decision.replacementId]:[])).size!==decisions.filter(decision=>decision.replacementId).length)throw fault('invalid_recovery_decisions'); @@ -175,17 +184,20 @@ export function createLedger(initial?: LedgerSnapshot) { for(const decision of decisions)operations.set(decision.operationId,freeze({...operations.get(decision.operationId)!,phase:'resolved',resolution:decision.resolution,recoveryId,...(decision.replacementId?{replacementId:decision.replacementId}:{})})); if(decisions.length)changed(); }, - /** 保存原请求的 dueAt/action，重试计划也是可恢复事实；实际 heap 是 scheduler 的派生索引。 */ + /** 保存原请求的 dueAt/action；这里只要求 dueAt 有限，不要求大于当前时间。 + * 重试计划是可恢复事实，实际 heap/timer 是 scheduler 的派生索引；这里不启动 timer。 */ scheduleRetry(id: Id, dueAt: number, action: 'submit' | 'query') { const request = requireRequest(id); if (request.phase === 'resolved' || !Number.isFinite(dueAt)) throw fault('invalid_retry', [id]); requests.set(id, freeze({ ...request, retryAt: dueAt, retryAction: action })); changed(); }, - /** 为未解决请求记录解释器/权限等阻碍，并清重试计划；保留 body/key 和 claims 的恢复依据。 */ + /** 为未解决请求记录解释器/权限等阻碍，发生写入时清 retry 元数据，保留 body/key。 + * 已 resolved 或 error 相同直接 no-op；相同 error 的早返回不会再清已有 retry 元数据。 */ blockRequest(id: Id, error: string) { const request = requireRequest(id); if (request.phase === 'resolved' || request.error === error) return; requests.set(id, freeze({ ...request, error, retryAt: undefined, retryAction: undefined })); changed(); }, /** 仅允许 durable 水位单调前进到当前 generation 内；不新增业务 generation。 */ markDurable(through: number) { if (!Number.isSafeInteger(through) || through < durableGeneration || through > generation) throw fault('invalid_durable_generation'); durableGeneration = through; refresh(); }, - /** 发送标记后且未确定 not-applied 均按可能已执行处理；这是保守风险依据，不是成功证明。 */ + /** 精确条件为 phase!=frozen 且 outcome!=not-applied，按可能已执行处理；applied/resolved 也可为 true。 + * 这是基于账本标记的保守风险读取，不查询网络，不证明成功，也不取代实际发送前持久化屏障。 */ possiblyApplied(id: Id) { const request = requireRequest(id); return request.phase !== 'frozen' && request.outcome !== 'not-applied'; }, }); } diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index 0a1eab7..ddabe9b 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -1,7 +1,8 @@ /** * 从 ledger 派生执行顺序的调度器：ready FIFO+queued 去重、依赖索引、canvas claims 与重试 min-heap。 * 所有权威变化通过 options.mutate 写回 Root；队列本身既不是历史，也不保存另一套任务状态。 - * 发送前先持久化精确 body/key 和 dispatch 标记，结果协调及持久化成功后才释放占用。 + * 发送前等待宿主持久化精确 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'; @@ -21,27 +22,40 @@ type Ledger = ReturnType; /** engine 提供准入、纯编译、原子账本更新、持久屏障、外部端口和完成协调。 * complete 必须连回执/基线/working/ledger 一起处理，不允许仅把请求先标成功。 */ export type SchedulerOptions = { + /** false 进入准入等待集合，由 refreshAdmission 唤醒；不在此把逻辑操作改为 blocked。 */ admit?: (intent:OperationIntent)=>boolean; /** 每次准入/唤醒读取当前权威快照，不信任入队时捕获的旧状态。 */ getLedger(): LedgerSnapshot; /** 在 engine 的受控发布路径内修改账本，失败不能留下半份持久事实。 */ mutate(reduce: (ledger: Ledger) => void): void; - /** 等待当前 Root 可靠存储；失败保持请求身份与占用，不退回生成新请求。 */ + /** 等待当前 Root 可靠存储；失败保持请求身份与占用，不退回生成新请求。 + * scheduler 信任此 Promise 的持久承诺，本身不核对 durableGeneration/底层事务。 */ persist(): Promise; + /** 为新冻结请求分配 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; + /** 按原请求身份查结果，端口解释回执；端口抛错时 scheduler 统一记为 unresolved/transport_unknown。 */ query(request: FrozenRequest, signal: AbortSignal): Promise; + /** 恢复时确认冻结格式是否有解释器；省略钩子只默认支持 protocolVersion=1。 */ supportsRequest?: (request: FrozenRequest) => boolean; - /** Atomically reconcile the receipt/base/working document and the ledger outcome. */ + /** 同步协调回执/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; }; @@ -58,7 +72,8 @@ export function createScheduler(options: SchedulerOptions) { const waiters = new Map>(), dependents = new Map>(), indexed = new Set(); // 最早 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 只跟踪当前异步句柄/AbortController；任务完成事实仍必须回到 ledger。 + // active 跟踪整条异步流水线（包含发送前/后的持久化等待），不是仅统计正在进行的 HTTP 请求。 + // 句柄/AbortController 属于调度器，任务完成事实仍必须回到 ledger。 const tokens = new Map(), active = new Map; kind: FrozenRequest['kind'] }>(); // 多个恢复/关闭流程可独立 hold；只有最后一个 token 释放才继续准入，已启动工作仍正常收尾。 const holds = new Set(); @@ -73,7 +88,7 @@ export function createScheduler(options: SchedulerOptions) { 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; - /** 总并发上限之外，上传独占一个 lane；其它类别仍可使用剩余网络执行位。 */ + /** 总 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；不是通知一次就盲发一次。 */ @@ -100,7 +115,8 @@ export function createScheduler(options: SchedulerOptions) { enqueue(id); } /** Only a contiguous mutable document prefix may share a request. Task barriers stay visible. */ - /** 只选连续未冻结 document.change 前缀，最多 64 意图/10000 changes，遇任务屏障或占用即停。 + /** 只选连续未冻结 document.change 前缀，最多 64 意图；追加候选时不让累计 changes 超过 10000。 + * 首意图本身超过 10000 时仍可单独返回，不把它拆成字段请求；遇任务屏障/占用即停。 * 仅合并有相交参与者的 lane；独立 canvas 保留各自失败/重试边界。 */ function batch(first: OperationIntent, canvases: Set): readonly OperationIntent[] { const selected = [first], ids = new Set([first.id]); @@ -143,7 +159,8 @@ export function createScheduler(options: SchedulerOptions) { } for (const id of request.operationIds) for (const dependent of dependents.get(id) ?? []) enqueue(dependent); } - /** 可选写回可恢复 retry 元数据，再用新 token 入 heap；重建索引时 persist=false 避免重复写事实。 */ + /** 可选 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); @@ -160,7 +177,8 @@ export function createScheduler(options: SchedulerOptions) { 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 句柄后在微任务中执行：dispatch 标记→持久化→submit/query→原子协调→持久化。 */ + /** 安装 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(); @@ -186,6 +204,7 @@ export function createScheduler(options: SchedulerOptions) { 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(); @@ -203,7 +222,8 @@ export function createScheduler(options: SchedulerOptions) { }); active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind }); } - /** 可重入保护下消费到期请求和最多 64 个 ready 项；等待者移到索引，不在 FIFO 无界反复轮询。 */ + /** 防同步重入地处理到期请求，再从本轮 ready 前缀最多取 64 项；retry 循环没有同样的 64 项上限。 + * 等待者移到索引，ready 尚有余量且有容量时安排下一轮 timer，不代表总队列大小有上限。 */ function pump() { if (stopped || holds.size || pumping) return; pumping = true; @@ -250,7 +270,8 @@ export function createScheduler(options: SchedulerOptions) { } const intents = options.compileBatch ? candidates : [record.intent]; const participants = [...new Set(intents.flatMap(intent => options.participants(intent)))].sort(); - // 编译参与者必须与规划一致，随后 freeze/own/launch；失败阻塞原意图而非偷偷重新编另一请求。 + // 编译参与者必须与实际 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) })); @@ -267,7 +288,8 @@ export function createScheduler(options: SchedulerOptions) { } } return Object.freeze({ - /** 首次启动（已运行时幂等）：先恢复全部未解决 claims，按原 retry/body/key 继续；解释器缺失仍保留占用。 */ + /** 首次启动（已运行时幂等）：先恢复全部未解决 claims，之后才开始端口工作；claim 冲突会在仍 stopped 时抛错。 + * 解释器缺失保留占用并写 blocker；此步骤不做完整恢复 DTO 校验，也不从单个错误推断请求未执行。 */ start() { if (closed) throw fault('scheduler_closed'); if (!stopped) return; @@ -284,7 +306,8 @@ export function createScheduler(options: SchedulerOptions) { }, /** 门禁事实变化后显式重试准入，未满足者继续等待，不制造定时器忙轮询。 */ refreshAdmission(){for(const id of [...admissionWaiters])enqueue(id);if(ready.size)requestPump();}, - /** 新逻辑操作通知只推进派生索引，重复 notify 不重复注册依赖或入队。 */ + /** 新逻辑操作通知推进派生索引；正在 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；返回幂等释放函数，已准入异步工作不被撤销或换身份。 */ @@ -316,7 +339,8 @@ export function createScheduler(options: SchedulerOptions) { }, /** 返回当前占用表的只读副本，调用者不能手改 claim 绕过未知请求。 */ getClaims: () => readMap(claims), - /** 等待当前微任务/pump/active 收尾，不主动等待未来 retry timer，因此不是“全部已同步”保证。 */ + /** 等待当前微任务/pump/active 收尾，不主动等待未来 retry timer，因此不是“全部已同步”保证。 + * 端口若长期不完成（包括忽略 abort），active Promise 仍可让 settle 一直等待。 */ async settle() { do { if (scheduledPump) await scheduledPump; @@ -324,7 +348,8 @@ export function createScheduler(options: SchedulerOptions) { 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。 */ + /** 永久关闭本 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(); }, }); } diff --git a/src/lib/canvas-runtime/structures.ts b/src/lib/canvas-runtime/structures.ts index a2705c9..9c84067 100644 --- a/src/lib/canvas-runtime/structures.ts +++ b/src/lib/canvas-runtime/structures.ts @@ -1,6 +1,7 @@ /** * runtime 使用的基础调度容器，不包含任务状态事实：FIFO 存可执行顺序，min-heap 存优先项。 * 两者都是内存结构，ledger/checkpoint 才负责可恢复事实；容器本身不去重、不校验依赖或执行网络。 + * T 值按引用保存/返回，不复制冻结任务对象，也不以 size/clear 修改外部任务事实。 */ /** FIFO with amortized O(1) removal, without Array.shift on each item. */ /** 当前接口是先进先出队列：push 入尾、pop 出头，并非 history 的后进先出 stack。 @@ -24,12 +25,13 @@ export function createDeque() { }, /** 同时丢弃底层数组和游标，释放旧任务引用。 */ clear() { items = []; head = 0; }, - /** 返回有效区间的数组副本，外部改副本不会更改队列成员。 */ + /** 返回有效区间的浅数组副本，改副本成员位置不改队列；对象成员仍共享引用。 */ values() { return items.slice(head); }, }; } /** 数组实现二叉最小堆，compare<0 表示左项优先；peek O(1)，push/pop O(log n)。 - * 相等优先级不保证插入稳定性，上层若需要 FIFO tie-break 必须把序列写进 comparator。 */ + * 相等优先级不保证插入稳定性，上层若需要 FIFO tie-break 必须把序列写进 comparator。 + * compare 应稳定、同步且不抛错；容器不校验 comparator，比较中途异常也不提供事务回滚。 */ export function createMinHeap(compare: (left: T, right: T) => number) { // 父下标 (i-1)>>1，左右孩子 2i+1/2i+2；堆序只要求父不大于孩子，不是全数组排序。 const items: T[] = [];