Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(canvas): explain ledger scheduling and recovery invariants

Browse Source
This commit is contained in:
Codex Bot
2026-11-12 00:00:36 +08:00
parent 391cd48be8
commit 4c1b35e0cf
7 changed files with 207 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/applied-recovery.ts
+16
View File
@@ -1,3 +1,8 @@
/**
* 外部执行已证明 applied、但本地协调未完成时的恢复准备。
* 只在所有待确认请求都有 applied 证据时暂存资源/run/ledger 结果；旧回执的模型不直接发布。
* 调用者还要取完整 fresh read、验证 assertCanonical、重放 pending 并可靠发布同一 Root。
*/
import type { Id } from '../canvas-core/types.ts';
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
import type { AppliedModelProof, CanonicalProposal, EngineSnapshot, PersistencePort } from './contracts.ts';
@@ -6,26 +11,33 @@ import { runTerminal, type RunPort } from './runs.ts';
import type { AssetPort } from './resources.ts';
/** Stage acknowledgements only. The caller must validate a complete fresh read and durably publish it. */
/** 在私有表上解释已保存回执，返回 staged 快照与 fresh read 断言；缺证据/解释器即整体拒绝。 */
export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence:PersistencePort;runs?:RunPort;assets?:AssetPort;now:number}) {
const waiting=[...snapshot.runtime.ledger.requests.values()].filter(request=>request.phase!=='resolved');
// 只要还有 accepted/unresolved，就不能把整组当成确定已执行恢复，也不能新建替代请求重发。
const unknown=waiting.filter(request=>request.outcome!=='applied');
if(unknown.length)throw fault('recovery_waiting_for_receipts',unknown.map(request=>request.id));
/** 同一请求可能包含多个意图，以最早逻辑 sequence 排序，保证 run 受理先于其控制动作解释。 */
const sequence=(request:FrozenRequest)=>request.operationIds.reduce((result,id)=>Math.min(result,snapshot.runtime.ledger.operations.get(id)?.intent.sequence??Infinity),Infinity);
waiting.sort((a,b)=>sequence(a)-sequence(b));
// 新建候选账本和资源/run 表，后面的任何校验失败不会污染原 snapshot。
const ledger=createLedger(snapshot.runtime.ledger),runs=new Map(snapshot.runtime.runs),resources=new Map(snapshot.runtime.resources),conflicts=new Map(snapshot.runtime.conflicts);
const proofs:AppliedModelProof[]=[],runIds=new Set<Id>();
for(const request of waiting){
const receipt=snapshot.runtime.receipts.get(request.id);if(receipt===undefined)throw fault('recovery_receipt_missing',[request.id]);
let proof:AppliedModelProof|undefined;
// 资源回执核对逻辑资源 ID 与原字节长度；资源上传不声明 canvas 文档版本推进。
if(request.kind==='asset.upload'){
const intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent,resource=intent?.operationType==='asset.upload'?resources.get(intent.resourceId):undefined;
if(!ports.assets||!resource)throw fault('asset_interpreter_unavailable',[request.id]);
const stored=freeze(ports.assets.accept(request,{outcome:'applied',canonical:receipt},resource));
if(stored.id!==resource.id||stored.bytes!==resource.byteLength)throw fault('asset_receipt_mismatch');
resources.set(resource.id,freeze({...resource,stored}));
// run 由专用解释器验证受理/动作身份，不能把文档回执当成任务已执行证明。
}else if(['run.start','run.cancel','run.apply'].includes(request.kind??'')){
if(!ports.runs?.verifyApplied)throw fault('run_receipt_interpreter_unavailable',[request.id]);
const verified=ports.runs.verifyApplied(request,receipt);proof=verified.model;
// 受理必须绑定原 intent/request/board/target；已有 run 也必须具有相同身份和计划。
if(request.kind==='run.start'){
const run=verified.admission,intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent;
if(!run||verified.action||intent?.operationType!=='run.start'||run.operationId!==intent.id||run.requestId!==request.id||run.boardId!==intent.run.boardId||!equal(run.target,intent.run.target))throw fault('run_identity_mismatch');
@@ -38,19 +50,23 @@ export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence:
if(request.kind==='run.cancel'){
if(!action.runStatus||action.outputId||runTerminal(run.status)&&runTerminal(action.runStatus)&&run.status!==action.runStatus)throw fault('run_status_conflict');
}else if(!action.outputId||action.runStatus)throw fault('run_identity_mismatch');
// 控制 epoch 递增阻止旧 poll 盖回新状态；标 hasMore/nextPollAt 让任务按原 ID 继续追尾。
runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,...(action.runStatus?{status:action.runStatus}:{}),...(action.outputId?{appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId])]}:{}),hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:ports.now}));runIds.add(run.id);
}
}else{
if(!ports.persistence.verifyApplied)throw fault('document_receipt_interpreter_unavailable',[request.id]);
proof=ports.persistence.verifyApplied(request,receipt);
}
// 各请求有不同写版本范围：start 只 owner，cancel/upload 无文档写，普通/apply 要完整参与者。
const writes=request.kind==='run.start'?[request.ownerCanvasId!]:request.kind==='run.cancel'||request.kind==='asset.upload'?[]:request.canvasIds;
if(writes.length!==(proof?.versions.size??0)||writes.some(id=>!proof?.versions.has(id)))throw fault('receipt_coverage_mismatch',[request.id]);
if(proof){for(const [canvas,version]of proof.versions)if(!Number.isSafeInteger(version)||version<=request.expectedVersions[canvas])throw fault('receipt_version_mismatch',[canvas]);proofs.push(proof);}
// 仅在候选账本标已协调 applied；真正采用仍需外层 fresh read 校验和持久化。
ledger.result(request.id,'applied');conflicts.delete(request.id);
}
const staged:EngineSnapshot=Object.freeze({...snapshot,runtime:Object.freeze({...snapshot.runtime,ledger:ledger.getSnapshot(),runs:readMap(runs),resources:readMap(resources),conflicts:readMap(conflicts)})});
return Object.freeze({snapshot:staged,requestIds:Object.freeze(waiting.map(request=>request.id)),runIds:Object.freeze([...runIds]),
/** fresh read 版本不得落后于任何执行证明，并逐个执行协议专用内容断言。 */
assertCanonical(canonical:CanonicalProposal){for(const proof of proofs){for(const [canvas,version]of proof.versions)if((canonical.versions.get(canvas)??-1)<version)throw fault('recovery_read_behind_receipt',[canvas]);proof.assertCanonical(canonical);}},
});
}
src/lib/canvas-runtime/batch-projection.ts
+17
View File
@@ -1,19 +1,30 @@
/**
* 把已选中的未冻结逻辑段投影成更小的 wire operations；不修改 ledger 意图或用户 history。
* 协议 2 要求每个 logical ID 有非空 changes，所以前缀 ID 用显式条件 guard，末 ID 携带组合效果。
* 只有基线可证明、完整条件链合法且字节确实减少才采用，否则保留原始逐项请求表示。
*/
import type { Change, Document, ElementRecord, Id, OperationIntent } from '../canvas-core/types.ts';
import { applyChanges } from '../canvas-core/document.ts';
import { equal, fault, freeze } from '../canvas-core/values.ts';
import { compressChanges, type LedgerSnapshot } from './ledger.ts';
/** 使用 UTF-8 实际字节衡量 wire 大小，不能把 JS 字符串长度当 HTTP payload 大小。 */
const encoder = new TextEncoder();
/** 仅计算候选 JSON 表示的字节成本，不在这里发送或冻结请求。 */
const wireBytes = (value: unknown) => encoder.encode(JSON.stringify(value)).byteLength;
/** wire 中保留的 logical ID 与对应条件 changes；不是新的本地历史单元。 */
export type WireOperation = { readonly id: Id; readonly changes: readonly Change[] };
/** 对比规范化表的键值效果，忽略表迭代顺序但包含实际 value；墓碑也用同规则检查。 */
const sameMap = <T>(a: ReadonlyMap<Id, T>, b: ReadonlyMap<Id, T>) => a.size === b.size && [...a].every(([id, value]) => b.has(id) && equal(value, b.get(id)));
/** Pure projection for an already-selected mutable segment. Original ledger/history records never change. */
/** 尝试无损压缩多意图的 wire 投影；不足两项或任一不安全前提直接返回 ordered。 */
export function projectBatch(intents: readonly OperationIntent[], acknowledged: Document, ledger: LedgerSnapshot, participants: readonly Id[]): readonly WireOperation[] {
const scope = new Set(participants);
const ordered = intents.map(({ id, changes }) => ({ id, changes }));
if (intents.length < 2) return ordered;
// 只收集这段触及的基线记录组成 sparse 文档，减少复制/比较完整项目的成本。
const elements = new Map<Id, ElementRecord>();
const canvases = new Map(acknowledged.canvases.has(acknowledged.projectId) ? [[acknowledged.projectId, acknowledged.canvases.get(acknowledged.projectId)!]] : []);
const tombstones = new Map<Id, number>(), seen = new Set<Id>();
@@ -26,6 +37,7 @@ export function projectBatch(intents: readonly OperationIntent[], acknowledged:
if (change.kind === 'canvas') { const canvas = acknowledged.canvases.get(id); if (canvas) canvases.set(id, canvas); continue; }
if (seen.has(id)) continue;
seen.add(id);
// 首次触及必须有明确 absent/exists 且与 acknowledged 一致；unknown 不能推测为不存在。
const element = acknowledged.elements.get(id), presence = record.basePresence[id];
if (!Object.hasOwn(record.basePresence, id) || (presence !== 'absent' && presence !== 'exists') || (presence === 'exists') !== !!element) return ordered;
if (element) elements.set(id, element);
@@ -35,11 +47,14 @@ export function projectBatch(intents: readonly OperationIntent[], acknowledged:
const sparse: Document = { projectId: acknowledged.projectId, elements, canvases, tombstones };
const raw = intents.flatMap(intent => intent.changes);
// Check the original conditional chain before any transformation can remove a check.
// 先执行原始条件链，防止压缩把原本应失败的 before 检查吞掉。
const original = applyChanges(sparse, raw);
// wire 投影保留生命周期和 no-op 检查，完整零效果段由独立 elision 证明处理。
const compacted = compressChanges(raw, { preserveLifecycle: true, preserveChecks: true });
if (!compacted.length) return ordered;
// Protocol 2 requires a nonempty change list per logical ID. Prefix IDs get an explicit
// conditional no-op; the final ID carries the composed effect. All IDs remain deduplicated.
// 从参与 owner 的少量元素 title/text 或根 canvas 选 no-op guard，不扩大原 claim 范围。
const guards: Change[] = [];
let candidates = 0;
for (const element of elements.values()) if (scope.has(element.ownerCanvasId)) {
@@ -52,11 +67,13 @@ export function projectBatch(intents: readonly OperationIntent[], acknowledged:
const root = canvases.get(acknowledged.projectId);
// A child canvas no-op would touch its parent. Never silently widen the selected claims.
if (root && root.parent === null && scope.has(root.id)) guards.push({ kind: 'canvas', id: root.id, before: root, after: root });
// 选择最小字节 guard；没有安全 guard、超 changes 限制或不省字节就不压缩。
let guard: Change | undefined, guardBytes = Infinity;
for (const candidate of guards) { const bytes = wireBytes(candidate); if (bytes < guardBytes) { guard = candidate; guardBytes = bytes; } }
if (!guard) return ordered;
const projected = intents.map((intent, index) => ({ id: intent.id, changes: index === intents.length - 1 ? compacted : [guard!] }));
if (projected.reduce((count, operation) => count + operation.changes.length, 0) > 10_000 || wireBytes(projected) >= wireBytes(ordered)) return ordered;
// 最后执行投影并对比 elements/canvases/tombstones 全效果，不只比较可见节点内容。
const result = applyChanges(sparse, projected.flatMap(operation => operation.changes));
if (!sameMap(original.elements, result.elements) || !sameMap(original.canvases, result.canvases) || !sameMap(original.tombstones, result.tombstones)) throw fault('batch_projection_mismatch');
return freeze(projected);
src/lib/canvas-runtime/change-rebase.ts
+38
View File
@@ -1,3 +1,8 @@
/**
* 用户审阅冲突后按新基线重新规划原意图的纯工具，不是正常 ACK 路径的自动覆盖合并。
* 保留用户实际修改的字段/世界几何意图，未修改的当前字段继续沿用 fresh 数据；未知类型或
* 生命周期/坐标上下文无法确认时拒绝。生成的是新 changes，原冻结 body/key 不在此改写。
*/
import type { Change, Document, ElementRecord, FieldChange, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
import { applyChanges, commonCanvas, validateDocument } from '../canvas-core/document.ts';
import { canvasTransform, convertGeometry, convertPath, type Transform } from '../canvas-core/coordinates.ts';
@@ -5,12 +10,15 @@ import { closeNumber, closeScale, presentationRecord, unitScale, withUnitScale }
import { removalChanges, repairConnectionOwners } from '../canvas-core/structure.ts';
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
/** 原操作相关对象的前后记录及 owner 变换；用于区分用户意图与仅由坐标转换引起的差异。 */
export type IntentContext = { readonly before:ReadonlyMap<Id,ElementRecord>; readonly after:ReadonlyMap<Id,ElementRecord>; readonly transforms:ReadonlyMap<Id,Transform>; readonly beforeTransforms:ReadonlyMap<Id,Transform> };
/** Reconstruct only immutable intention metadata. This mutable cursor is never published as a document. */
/** 从当前 working 反向遍历意图，重建每项的局部前后上下文；可变 cursor 永不作为业务文档发布。 */
export function intentContexts(document:Document,intents:readonly OperationIntent[]):ReadonlyMap<Id,IntentContext> {
const elements=new Map(document.elements),canvases=new Map(document.canvases),contexts=new Map<Id,IntentContext>();
const cursor={...document,elements,canvases};
for(const intent of [...intents].reverse()){
// 收集自身、连线端点与 canvas 入口，捕获足够的身份和坐标上下文，不复制全项目到每条意图。
const ids=new Set<Id>(),owners=new Set<Id>();
for(const change of intent.changes){
if(change.kind==='field'){ids.add(change.target.id);if(change.field==='source'||change.field==='target')for(const endpoint of [change.before,change.after])if(endpoint&&typeof endpoint==='object'&&!Array.isArray(endpoint)&&typeof endpoint.elementId==='string')ids.add(endpoint.elementId);}
@@ -19,6 +27,7 @@ export function intentContexts(document:Document,intents:readonly OperationInten
}
const after=new Map<Id,ElementRecord>();for(const id of ids){const record=elements.get(id);if(record){after.set(id,record);owners.add(record.ownerCanvasId);}}
const transforms=new Map<Id,Transform>();for(const owner of owners)if(canvases.has(owner))transforms.set(owner,canvasTransform(cursor,owner));
// 私有 cursor 按原 before 逆向退回，不执行用户可见的 restore，也不因此分配新 incarnation。
for(const change of [...intent.changes].reverse()){
if(change.kind==='element'){if(change.before)elements.set(change.id,change.before);else elements.delete(change.id);}
else if(change.kind==='canvas'){if(change.before)canvases.set(change.id,change.before);else canvases.delete(change.id);}
@@ -33,13 +42,16 @@ export function intentContexts(document:Document,intents:readonly OperationInten
return readMap(contexts);
}
/** 比较路径是否仅存在坐标换算误差：点坐标用容差，其它字段与 knot 顺序仍严格比较。 */
function sameConvertedPath(left:Json,right:Json):boolean {
if(equal(left,right))return true;
if(!left||!right||typeof left!=='object'||typeof right!=='object'||Array.isArray(left)||Array.isArray(right)||!Array.isArray(left.knots)||!Array.isArray(right.knots))return false;
const {knots:a,startControl:as,endControl:ae,...ar}=left,{knots:b,startControl:bs,endControl:be,...br}=right;
/** 控制点容差比较同时保留扩展字段检查，不能把两条实际不同的曲线误认为坐标等效。 */
const point=(a:Json|undefined,b:Json|undefined)=>{if(a===undefined||b===undefined)return a===b;if(!a||!b||typeof a!=='object'||typeof b!=='object'||Array.isArray(a)||Array.isArray(b))return false;const{x:ax,y:ay,...restA}=a,{x:bx,y:by,...restB}=b;return typeof ax==='number'&&typeof ay==='number'&&typeof bx==='number'&&typeof by==='number'&&closeNumber(ax,bx)&&closeNumber(ay,by)&&equal(restA,restB);};
return equal(ar,br)&&point(as,bs)&&point(ae,be)&&a.length===b.length&&a.every((value,index)=>{const other=b[index];if(!value||!other||typeof value!=='object'||typeof other!=='object'||Array.isArray(value)||Array.isArray(other))return false;const{position:p,in:i,out:o,...rest}=value,{position:q,in:j,out:k,...remaining}=other;return equal(rest,remaining)&&point(p,q)&&point(i,j)&&point(o,k);});
}
/** 只把原意图实际改过的 presentation 键应用到 current；unitScale 另走视觉单位换算。 */
function presentationDelta(current:Json|undefined,before:Json|undefined,after:Json|undefined):Json|undefined {
if(equal(before,after))return current;
const start=presentationRecord(before),end=presentationRecord(after),keys=new Set([...Object.keys(start),...Object.keys(end)]);keys.delete('unitScale');
@@ -47,11 +59,14 @@ function presentationDelta(current:Json|undefined,before:Json|undefined,after:Js
const result=new Map(Object.entries(presentationRecord(current)));for(const key of changed){if(Object.hasOwn(end,key))result.set(key,end[key]);else result.delete(key);}
return result.size?Object.fromEntries(result):undefined;
}
/** 重挂节点时先将原前/后几何与当前几何转换到世界空间，保留未修改维度的当前值。
* 用户明确改变的中心/尺寸/旋转取其意图值，再转到 fresh destination；不是盲加一份旧 local patch。 */
function movedNode(current:Extract<ElementRecord,{structure:'node'}>,originalBefore:Extract<ElementRecord,{structure:'node'}>,originalAfter:Extract<ElementRecord,{structure:'node'}>,beforeBasis:Transform,afterBasis:Transform,source:Transform,destination:Transform,intent:OperationIntent){
const world:Transform={x:0,y:0,scale:1,rotation:0},start=convertGeometry(originalBefore.geometry,beforeBasis,world),end=convertGeometry(originalAfter.geometry,afterBasis,world),fresh=convertGeometry(current.geometry,source,world);
const center=(g:typeof start)=>({x:g.x+g.width/2,y:g.y+g.height/2}),a=center(start),b=center(end),c=center(fresh),width=closeNumber(start.width,end.width)?fresh.width:end.width,height=closeNumber(start.height,end.height)?fresh.height:end.height;
const x=closeNumber(a.x,b.x)?c.x:b.x,y=closeNumber(a.y,b.y)?c.y:b.y,rotation=Math.abs(((start.rotation-end.rotation+540)%360)-180)<1e-9?fresh.rotation:end.rotation;
const geometry=convertGeometry({x:x-width/2,y:y-height/2,width,height,rotation},world,destination);
// 只有原意图显式改变视觉 unitScale 才采用目标视觉尺度；纯换 owner 保留当前外观。
const unitEdited=intent.changes.some(change=>change.kind==='field'&&change.target.id===current.id&&change.field==='data.presentation'&&!closeScale(unitScale(change.before),unitScale(change.after)));
const oldStyle=unitScale(originalBefore.data.presentation)*beforeBasis.scale,newStyle=unitScale(originalAfter.data.presentation)*afterBasis.scale;
const desiredStyle=unitEdited&&!closeScale(oldStyle,newStyle)?newStyle:unitScale(current.data.presentation)*source.scale;
@@ -60,14 +75,17 @@ function movedNode(current:Extract<ElementRecord,{structure:'node'}>,originalBef
}
/** Restore batches may have inverse storage order (canvas before entry). Only reorder adjacent creates, never move across an edit/delete. */
/** 只在连续创建段内按 parent/entry/owner/端点依赖拓扑排序，绝不跨编辑/删除重排。 */
function orderedCreates(changes:readonly Change[]):readonly Change[]{
const result:Change[]=[];
/** before=null 且 after 存在才是创建，整记录替换不能冒充依赖排序候选。 */
const creating=(change:Change)=>change.kind!=='field'&&!change.before&&!!change.after;
for(let offset=0;offset<changes.length;){
if(!creating(changes[offset])){result.push(changes[offset++]);continue;}
let end=offset;while(end<changes.length&&creating(changes[end]))end++;
const batch=changes.slice(offset,end),keys=batch.map(change=>`${change.kind}:${(change as Extract<Change,{kind:'element'|'canvas'}>).id}`),positions=new Map(keys.map((key,index)=>[key,index]));
if(positions.size!==batch.length)throw fault('recovery_creation_dependency_conflict');
// Kahn 排序：counts 记段内依赖数，waiting 存反向边，ready 按原遇到顺序入列。
const counts=batch.map(()=>0),waiting=new Map<number,number[]>(),ready:number[]=[];
batch.forEach((change,index)=>{
const dependencies=new Set<string>();
@@ -76,6 +94,7 @@ function orderedCreates(changes:readonly Change[]):readonly Change[]{
for(const key of dependencies){const dependency=positions.get(key);if(dependency===undefined)continue;counts[index]++;const list=waiting.get(dependency)??[];list.push(index);waiting.set(dependency,list);}
if(!counts[index])ready.push(index);
});
// 每输出一个创建就降低依赖项计数；输出不足整段说明依赖环/冲突，整体拒绝。
let emitted=0;for(let next=0;next<ready.length;next++){const index=ready[next];result.push(batch[index]);emitted++;for(const dependent of waiting.get(index)??[])if(--counts[dependent]===0)ready.push(dependent);}
if(emitted!==batch.length)throw fault('recovery_creation_dependency_conflict');offset=end;
}
@@ -83,27 +102,37 @@ function orderedCreates(changes:readonly Change[]):readonly Change[]{
}
/** Re-plan against current data. Whole records are used only for explicit creation/removal. */
/** 在 fresh 私有文档上重新生成条件 changes，原 intent/context 保持不变；最终完整校验才返回。 */
export function rebaseIntent(document:Document,intent:OperationIntent,context:IntentContext,registry:Registry):{document:Document;changes:readonly Change[]} {
// moved 标记需整体世界空间重规划的节点；deferredPaths 等端点/树最终确定后再算曲线。
const before=document,changes:Change[]=[],moved=new Set<Id>(),deferredPaths:FieldChange[]=[];
for(const change of intent.changes)if(change.kind==='field'&&change.field==='ownerCanvasId'&&context.after.get(change.target.id)?.structure==='node')moved.add(change.target.id);
/** 每条新变化立即作用于私有 candidate，后面的条件读取已重规划后的实际 before。 */
function apply(change:Change){document=applyChanges(document,[change]);changes.push(freeze(change));}
/** 同 ID 还必须同 incarnation/结构/组件/版本且 registry 可解释，防止旧意图套到新对象上。 */
function target(change:FieldChange){const current=document.elements.get(change.target.id),prior=context.before.get(change.target.id)??context.after.get(change.target.id);if(!current||current.incarnation!==change.target.incarnation)throw fault('stale_incarnation',[change.target.id]);if(prior&&(current.componentId!==prior.componentId||current.dataVersion!==prior.dataVersion||current.structure!==prior.structure))throw fault('recovery_target_type_changed',[current.id]);if(registry.resolve(current)!=='known')throw fault('unsupported_component',[current.id]);return current;}
/** 端点必须仍为原上下文中的同代次节点；不能把恢复出来的新节点视为旧外部引用。 */
function endpoint(id:Id){const current=document.elements.get(id),expected=context.after.get(id);if(!current||current.structure!=='node'||!expected||current.incarnation!==expected.incarnation)throw fault('stale_endpoint',[id]);}
// 按受限依赖顺序处理创建；删除/字段意图的原语义顺序仍保留。
for(const original of orderedCreates(intent.changes)){
if(original.kind==='element'){
const current=document.elements.get(original.id);
if(original.before===null&&original.after){
// 仅 undo/redo 显式恢复允许 dead+1；其它碰撞不能静默改 ID 或复用墓碑。
const dead=document.tombstones.get(original.id),restoring=['undo','redo'].includes(intent.origin)&&dead!==undefined&&original.after.incarnation===dead+1;
if(current||dead!==undefined&&!restoring)throw fault('recovery_identity_collision',[original.id]);
if(registry.resolve(original.after)!=='known')throw fault('unsupported_component',[original.id]);
let after=original.after;
// 新建节点按原意图世界位置转到 fresh owner，表现单位随其比例换算。
if(after.structure==='node'){const basis=context.transforms.get(after.ownerCanvasId);if(!basis)throw fault('recovery_coordinate_context_missing',[after.id]);const destination=canvasTransform(document,after.ownerCanvasId),presentation=withUnitScale(after.data.presentation,unitScale(after.data.presentation)*basis.scale/destination.scale),data={...after.data};if(presentation===undefined)delete data.presentation;else data.presentation=presentation;after={...after,data,geometry:convertGeometry(after.geometry,basis,destination)};}
// 新建边校验新端点代次，重新解算共同 owner，再转换原路径坐标。
if(after.structure==='connection'){
endpoint(after.source.elementId);endpoint(after.target.elementId);const owner=commonCanvas(document,document.elements.get(after.source.elementId)!.ownerCanvasId,document.elements.get(after.target.elementId)!.ownerCanvasId);
const basis=context.transforms.get(after.ownerCanvasId);if(!basis)throw fault('recovery_coordinate_context_missing',[after.id]);
after={...after,ownerCanvasId:owner,path:convertPath(after.path,basis,canvasTransform(document,owner))};
}
apply({...original,after});
// 已经不存在的删除可满足；仍存在时按 fresh 拥有树规划移除，不能直接覆盖旧整记录。
}else if(original.before&&original.after===null){
if(!current)continue;
if(current.incarnation!==original.before.incarnation)throw fault('stale_incarnation',[original.id]);
@@ -112,6 +141,7 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
}else throw fault('recovery_record_replacement_not_supported',[original.id]);
continue;
}
// 节点重挂会统一维护其入口子 canvas；其它树变化校验入口，删除非空 canvas 明确拒绝。
if(original.kind==='canvas'){
if(original.before?.parent&&original.after?.parent&&original.before.parent.entryElementId===original.after.parent.entryElementId&&moved.has(original.after.parent.entryElementId))continue;
const current=document.canvases.get(original.id);
@@ -122,6 +152,7 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
continue;
}
let current=target(original);const prior=context.before.get(current.id),afterContext=context.after.get(current.id);
// 边 owner 留到最终端点/树确定后重算；节点重挂一次规划 owner+geometry+presentation+入口 parent。
if(original.field==='ownerCanvasId'){
if(current.structure==='connection')continue; // Recomputed from the final endpoints and tree below.
if(!registry.supports(current,'geometry.move')||typeof original.after!=='string'||!document.canvases.has(original.after))throw fault('unsupported_capability',[current.id]);
@@ -135,7 +166,9 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
current=document.elements.get(current.id)!;if(current.structure==='node'&&!equal(current.geometry,converted.geometry))apply({kind:'field',target:original.target,field:'geometry',before:current.geometry,after:converted.geometry});
continue;
}
// 已由重挂分支统一转换的几何/样式不再套旧字段变化，避免重复缩放或位移。
if((original.field==='geometry'||original.field==='data.presentation')&&moved.has(current.id))continue;
// 合并用户改过的样式键；unitScale 按原后空间与 fresh owner 换算，保留未改样式。
if(original.field==='data.presentation'){
const changedUnits=!closeScale(unitScale(original.before),unitScale(original.after));
let desired=presentationDelta(current.data.presentation,original.before,original.after);
@@ -144,11 +177,13 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
continue;
}
// 换端点仍要求 connection.edit 与当前节点身份，相关曲线路径随后再处理。
if(original.field==='source'||original.field==='target'){
if(current.structure!=='connection'||!registry.supports(current,'connection.edit'))throw fault('unsupported_capability',[current.id]);
const next=original.after as unknown as {elementId:Id};endpoint(next.elementId);
if(!equal(current[original.field],original.after))apply({...original,before:current[original.field]});continue;
}
// 与 owner/端点同时改的路径延后；孤立几何/路径若 owner 已变则拒绝盲重放。
if(original.field==='path'){
const related=intent.changes.some(change=>change.kind==='field'&&change.target.id===current.id&&['ownerCanvasId','source','target'].includes(change.field));
if(related){if(!registry.supports(current,'connection.edit'))throw fault('unsupported_capability',[current.id]);deferredPaths.push(original);continue;}
@@ -157,6 +192,7 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
if(original.field.startsWith('data.')&&!registry.supports(current,'content.edit')||original.field==='path'&&!registry.supports(current,'connection.edit'))throw fault('unsupported_capability',[current.id]);
const key=original.field.startsWith('data.')?original.field.slice(5):undefined,actual=key===undefined?(current as unknown as Record<string,Json>)[original.field]:current.data[key]??null,present=key===undefined?undefined:Object.hasOwn(current.data,key);
let desired=original.after;
// 普通几何修改仅覆盖 before→after 实际变化的轴，其余轴沿用 fresh；逐能力复核准入。
if(original.field==='geometry'&&current.structure==='node'){
const start=original.before as unknown as typeof current.geometry,end=original.after as unknown as typeof current.geometry;
desired={...current.geometry,...Object.fromEntries((['x','y','width','height','rotation'] as const).filter(key=>start[key]!==end[key]).map(key=>[key,end[key]]))};
@@ -164,6 +200,7 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
}
if(!equal(actual,desired)||(present??true)!==(original.afterPresent??true)){const{beforePresent:_previous,...record}=original;apply({...record,before:actual,...(present===undefined?{}:{beforePresent:present}),after:desired});}
}
// 纯坐标转换造成的路径差异不当作用户改曲线；显式路径意图转到最终 owner 空间再发布候选。
for(const original of deferredPaths){
const edge=target(original),intended=context.after.get(edge.id);if(edge.structure!=='connection'||intended?.structure!=='connection')throw fault('invalid_endpoint',[edge.id]);
const previous=context.before.get(edge.id),oldBasis=previous?context.beforeTransforms.get(previous.ownerCanvasId):undefined,basis=context.transforms.get(intended.ownerCanvasId);if(!basis)throw fault('recovery_coordinate_context_missing',[edge.id]);
@@ -173,6 +210,7 @@ export function rebaseIntent(document:Document,intent:OperationIntent,context:In
if(edge.ownerCanvasId!==owner)apply({kind:'field',target:original.target,field:'ownerCanvasId',before:edge.ownerCanvasId,after:owner});
if(!equal(edge.path,path))apply({...original,before:edge.path,after:path});
}
// 所有端点/树变化结束后统一修复边 owner/path，再验证最终文档，避免暴露半份图。
for(const change of repairConnectionOwners(before,document))apply(change);
validateDocument(document,registry);return{document,changes:freeze(changes)};
}
src/lib/canvas-runtime/elision.ts
+22
View File
@@ -1,14 +1,23 @@
/**
* 未发出的创建→编辑→删除闭合生命周期可以不发 wire 请求，但不能抹掉本地身份事实。
* 抵消要求 absent 基线、连续未冻结意图、闭合依赖和完整零最终记录效果；墓碑保留在 working。
* 已发/可能已应用或存在基线上的修改不进入该路径，原 history/意图继续保留以支持撤销与恢复。
*/
import type { Document, Id, OperationIntent } from '../canvas-core/types.ts';
import { applyChanges } from '../canvas-core/document.ts';
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
import type { LedgerSnapshot } from './ledger.ts';
/** 抵消证据：连续逻辑 ID、起始墓碑与最终生命周期水位；version 管理本地证明解释器。 */
export type ElisionProof = { readonly version: 1; readonly operationIds: readonly Id[]; readonly baseTombstones: Readonly<Record<Id, number>>; readonly tombstones: Readonly<Record<Id, number>> };
/** Reconstruct a closed lifetime from an empty fragment. Existing objects cannot enter this proof. */
/** 在空记录片段上重建至多 64 意图/10000 changes，验证每个对象首触及都是创建且最终全删。
* 墓碑变化是允许保留的本地效果，不等于服务器执行过这些创建/删除。 */
export function proveElision(intents: readonly OperationIntent[], baseTombstones: Readonly<Record<Id, number>>): ElisionProof {
if (!intents.length || intents.length > 64 || new Set(intents.map(intent => intent.id)).size !== intents.length) throw fault('invalid_elision');
const created = new Set<Id>(), canvases = new Set<Id>();
// 用起始+本段删除水位阻止创建/恢复复用已死亡 incarnation。
const highWater = new Map(Object.entries(baseTombstones));
let count = 0;
for (const [id, incarnation] of Object.entries(baseTombstones)) if (!id || !Number.isSafeInteger(incarnation) || incarnation < 1) throw fault('invalid_elision');
@@ -17,6 +26,7 @@ export function proveElision(intents: readonly OperationIntent[], baseTombstones
for (const change of intent.changes) {
if (++count > 10_000) throw fault('invalid_elision');
const id = change.kind === 'field' ? change.target.id : change.id, seen = change.kind === 'canvas' ? canvases : created;
// 字段修改不能作为首次接触：它意味着已有外部对象，不能纳入纯本地闭合证明。
if (!seen.has(id)) {
if (change.kind === 'field' || change.before !== null || !change.after) throw fault('elision_existing_target', [id]);
seen.add(id);
@@ -29,15 +39,19 @@ export function proveElision(intents: readonly OperationIntent[], baseTombstones
}
if (Object.keys(baseTombstones).some(id => !created.has(id))) throw fault('invalid_elision');
const fragment: Document = { projectId: '', elements: new Map(), canvases: new Map(), tombstones: new Map(Object.entries(baseTombstones)) };
// 条件执行仍使用真实 applyChanges；before 不匹配不能因最终想删除就被忽略。
const final = applyChanges(fragment, intents.flatMap(intent => intent.changes));
if (final.elements.size || final.canvases.size) throw fault('elision_has_effect');
return freeze({ version: 1, operationIds: intents.map(intent => intent.id), baseTombstones: { ...baseTombstones }, tombstones: Object.fromEntries(final.tombstones) });
}
/** Longest closed prefix before an existing-object write or another unsafe precondition. */
/** 寻找最长安全闭合前缀；遇 frozen/blocker/外部未解依赖/已有对象首写即停止扩展。
* preceding 只允许前面已选独立真实段作为依赖屏障，不把它当作已经 applied 的回执。 */
export function findElision(intents: readonly OperationIntent[], acknowledged: Document, ledger: LedgerSnapshot, preceding: ReadonlySet<Id> = new Set()): ElisionProof | undefined {
const selected: OperationIntent[] = [], seen = new Set<Id>(), canvases = new Set<Id>(), bases: Record<Id, number> = Object.create(null);
if (intents.length > 64 || !intents.some(intent => intent.changes.some(change => change.kind !== 'field' && change.after === null))) return undefined;
// 每次片段归空时保存最佳前缀；后续又创建未删对象，不影响之前已找到的闭合边界。
let bestLength = 0, bestBases: Record<Id, number> = {};
let fragment: Document = { projectId: acknowledged.projectId, elements: new Map(), canvases: new Map(), tombstones: new Map() };
for (const intent of intents) {
@@ -54,6 +68,7 @@ export function findElision(intents: readonly OperationIntent[], acknowledged: D
}
if (!safe) break;
selected.push(intent);
// 新发现对象的 ACK 墓碑合入私有片段后再回放，验证本次生命周期严格高于已知水位。
const tombstones = new Map(fragment.tombstones);
for (const [id, incarnation] of Object.entries(bases)) tombstones.set(id, Math.max(tombstones.get(id) ?? 0, incarnation));
fragment = applyChanges({ ...fragment, tombstones }, intent.changes);
@@ -63,6 +78,7 @@ export function findElision(intents: readonly OperationIntent[], acknowledged: D
}
/** Keep a later closed segment mutable while earlier real edits obtain their own receipt. */
/** 若后面存在可抵消段，限制前面的真实请求前缀，避免一次冻结把可抵消段也锁死。 */
export function elisionBatchLimit(intents: readonly OperationIntent[], acknowledged: Document, ledger: LedgerSnapshot): number {
if (!intents.some(intent => intent.changes.some(change => change.kind !== 'field' && change.after === null)) || findElision(intents, acknowledged, ledger)) return intents.length;
const preceding = new Set<Id>();
@@ -74,6 +90,7 @@ export function elisionBatchLimit(intents: readonly OperationIntent[], acknowled
}
/** A later locally canceled lifetime cannot reuse an incarnation retired by an earlier one. */
/** 用更早本地抵消的墓碑水位重新证明本段，拒绝跨两段复用已死代次。 */
export function assertElisionLineage(proof: ElisionProof, intents: readonly OperationIntent[], previous: ReadonlyMap<Id, number>): void {
const bases: Record<Id, number> = Object.assign(Object.create(null), proof.baseTombstones);
for (const id of Object.keys(proof.tombstones)) if ((previous.get(id) ?? 0) > (bases[id] ?? 0)) bases[id] = previous.get(id)!;
@@ -81,17 +98,21 @@ export function assertElisionLineage(proof: ElisionProof, intents: readonly Oper
}
/** Validate persisted local facts independently of the current (possibly newer) server baseline. */
/** 恢复时按逻辑顺序复验持久抵消组：成员连续、依赖闭合、anchor 唯一、证明与水位一致。
* 不依赖当前可能更新过的 ACK 基线来重新猜历史 absent 事实。 */
export function validateElisions(ledger: Pick<LedgerSnapshot, 'operations'>): void {
const ordered = [...ledger.operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence), positions = new Map(ordered.map((record, index) => [record.intent.id, index]));
const previous = new Map<Id, number>();
for (const record of ordered) {
if (record.resolution !== 'elided') { if (record.elision !== undefined || record.elisionId !== undefined) throw fault('invalid_elision'); continue; }
// 只有组首成员保存整份 proof，其余成员通过 elisionId 引用它，避免重复矛盾证明。
const anchor = record.elisionId ? ledger.operations.get(record.elisionId) : undefined, proof = anchor?.elision;
if (proof && Number.isSafeInteger(proof.version) && proof.version > 1) throw fault('elision_interpreter_unavailable');
if (!proof || proof.version !== 1 || proof.operationIds[0] !== record.elisionId || !proof.operationIds.includes(record.intent.id) || record.phase !== 'resolved' || record.requestId || record.blockers.length) throw fault('invalid_elision');
if (record !== anchor && record.elision !== undefined) throw fault('invalid_elision');
if (record === anchor) {
const seen = new Set<Id>(), members = new Set(proof.operationIds), offset = positions.get(record.intent.id)!;
// 每个成员仍必须 resolved/elided、无 request/blocker，不能伪造为已发送后又本地抵消。
const intents = proof.operationIds.map((id, index) => {
const member = ledger.operations.get(id);
if (!member || member.phase !== 'resolved' || member.resolution !== 'elided' || member.elisionId !== record.intent.id || member.requestId || member.blockers.length) throw fault('invalid_elision');
@@ -107,6 +128,7 @@ export function validateElisions(ledger: Pick<LedgerSnapshot, 'operations'>): vo
}
/** Local high-water marks belong to working state, never to the server ACK document. */
/** 只给 working 候选补本地墓碑最大值，不修改源文档，也不向服务器 ACK 事实表灌入本地证明。 */
export function withLocalLifecycles(document: Document, ledger: Pick<LedgerSnapshot, 'operations'>): Document {
const tombstones = new Map(document.tombstones);
for (const record of ledger.operations.values()) if (record.elision) for (const [id, incarnation] of Object.entries(record.elision.tombstones)) tombstones.set(id, Math.max(tombstones.get(id) ?? 0, incarnation));
src/lib/canvas-runtime/ledger.ts
+44
View File
@@ -1,11 +1,21 @@
/**
* 同步操作账本：保存逻辑意图、冻结请求、阶段、结果与恢复证据，是 scheduler 的事实源。
* 本模块只生成内存快照，不执行 HTTP/落盘；engine 将其纳入单 Root，再由持久化屏障允许发送。
* 逻辑 operation 与 wire request 分开：多个意图可共享请求，历史和资源保留不因请求合并而消失。
*/
import { assertElisionLineage, proveElision, type ElisionProof } from './elision.ts';
import { applyChanges } from '../canvas-core/document.ts';
import type { Change, Id, Json, OperationIntent } from '../canvas-core/types.ts';
import { clone, equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
/** 操作所依据的元素起始状态；unknown 不能当作 absent 来抵消 create/delete。 */
export type Presence = 'absent' | 'exists' | 'unknown';
/** 请求生命周期：冻结→保守发送标记→等待协调→已解决；发送标记必须可靠保存后才调用外部端口。 */
export type RequestPhase = 'frozen' | 'dispatch-marked' | 'waiting' | 'resolved';
/** accepted 仅表示受理；applied 是执行证明，仍需本地协调；not-applied 要求协议的确定零写入终态。 */
export type Outcome = 'unresolved' | 'accepted' | 'applied' | 'not-applied';
/** 冻结后的原 body/key、协议、逻辑 ID、参与 canvas 与预期版本，重试不得重新生成它们。
* phase/attempt/retry 是可推进的控制元数据；generation 记录冻结时的账本代次。 */
export type FrozenRequest = {
readonly kind?: OperationIntent['operationType'];
readonly ownerCanvasId?: Id; readonly runId?: Id;
@@ -14,19 +24,26 @@ export type FrozenRequest = {
readonly phase: RequestPhase; readonly outcome: Outcome; readonly error?: string; readonly attempt: number;
readonly retryAt?: number; readonly retryAction?: 'submit' | 'query';
};
/** 意图及其本地解析状态；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<Record<Id, Presence>> };
/** generation 随账本事实改变递增，durableGeneration 仅确认已实际存储的前缀。 */
export type LedgerSnapshot = { readonly generation: number; readonly durableGeneration: number; readonly operations: ReadonlyMap<Id, OperationRecord>; readonly requests: ReadonlyMap<Id, FrozenRequest> };
/** A field-safe compressor for a wholly unsent segment. Frozen records never enter here. */
/** 仅压缩完全未冻结的变化段：create 后字段修改并入新对象，同字段连续变化合并。
* preserveLifecycle 保留创建/删除，preserveChecks 保留条件 no-op；不能用最后整节点覆盖替代字段链。 */
export function compressChanges(changes: readonly Change[], options: { preserveLifecycle?: boolean; preserveChecks?: boolean } = {}): readonly Change[] {
// positions 按元素/canvas ID 保存结果索引栈，只尝试合并同对象最近变化，保留原条件顺序。
const result: (Change | undefined)[] = [], positions = new Map<string, number[]>();
for (const change of changes) {
const id = change.kind === 'field' ? change.target.id : change.id;
const key = `${change.kind === 'canvas' ? 'canvas' : 'element'}:${id}`, stack = positions.get(key) ?? [];
const previousIndex = stack.at(-1), previous = previousIndex === undefined ? undefined : result[previousIndex];
/** 标空而不移动全数组，弹掉该对象上一索引；最后统一过滤空槽。 */
const removePrevious = () => { result[previousIndex!] = undefined; stack.pop(); };
if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'field') {
// Apply the same conditional field semantics as the document; never discard a failed before check.
// 用与正常文档相同的条件字段执行器合并创建内容，before 不匹配仍应抛错，不能被压缩吞掉。
const document = { projectId: '', elements: new Map([[id, previous.after]]), canvases: new Map(), tombstones: new Map() };
const after = applyChanges(document, [change]).elements.get(id)!;
result[previousIndex!] = freeze({ ...previous, after });
@@ -40,16 +57,22 @@ export function compressChanges(changes: readonly Change[], options: { preserveL
return freeze(result.filter((change): change is Change => change !== undefined));
}
/** 从已有快照复制可变表，所有有效修改后重新捕获只读快照；旧 Snapshot 的成员不会改变。 */
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);
let cached: LedgerSnapshot;
/** 捕获当前 generation 和表成员，供外部一致读取；不在此处通知 UI 或执行持久化。 */
const refresh = () => { cached = Object.freeze({ generation, durableGeneration, operations: readMap(operations), requests: readMap(requests) }); };
refresh();
/** 账本事实变化推进内存 generation，durable 水位保持不动。 */
function changed() { generation++; refresh(); }
/** 按原 request ID 定位，缺失抛领域错误，不自动创建一条替代请求。 */
function requireRequest(id: Id) { const request = requests.get(id); if (!request) throw fault('request_missing', [id]); return request; }
/** 验证新增意图依赖存在且无环；visiting 仅表示当前 DFS 路径，退出节点后移除。 */
function assertAcyclic(intent: OperationIntent) {
const visiting = new Set<Id>();
/** 依赖不能回到待添加意图或已在当前路径上的节点，避免 scheduler 永久等待闭环。 */
const visit = (id: Id): void => {
if (id === intent.id || visiting.has(id)) throw fault('dependency_cycle', [id]);
const operation = operations.get(id);
@@ -59,7 +82,9 @@ export function createLedger(initial?: LedgerSnapshot) {
for (const id of intent.dependencies) visit(id);
}
return Object.freeze({
/** 读取最新捕获的只读事实，不暴露内部可变表。 */
getSnapshot: () => cached,
/** 校验身份/序列/基线与依赖，再添加 pending；同 ID 同意图幂等，改意图则拒绝。 */
add(intent: OperationIntent, basePresence: Readonly<Record<Id, Presence>>) {
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);
@@ -67,11 +92,15 @@ export function createLedger(initial?: LedgerSnapshot) {
assertAcyclic(intent);
operations.set(intent.id, freeze({ intent: clone(intent), phase: 'pending', blockers: [], basePresence: { ...basePresence } })); changed();
},
/** 初步可执行性：pending、无 blocker、依赖均 resolved；结果的业务语义由更上层协调。 */
eligible(id: Id) {
const operation = operations.get(id);
return !!operation && operation.phase === 'pending' && !operation.blockers.length && operation.intent.dependencies.every(dependency => operations.get(dependency)?.phase === 'resolved');
},
/** 给可变意图设置/清除 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。 */
freeze(input: { id: Id; kind?: FrozenRequest['kind']; ownerCanvasId?: Id; runId?: Id; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly<Record<Id, number>>; 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');
@@ -81,16 +110,20 @@ export function createLedger(initial?: LedgerSnapshot) {
if ((input.kind ?? 'document.change') !== operation?.intent.operationType) throw fault('request_kind_mismatch',[id]);
if (!operation || operation.phase !== 'pending' || operation.blockers.length || operation.intent.dependencies.some(dependency => operations.get(dependency)?.phase !== 'resolved' && !input.operationIds.includes(dependency))) throw fault('operation_not_ready', [id]);
}
// canvasIds 排序稳定 claim 顺序；请求和所有 operation 的 requestId/phase 同次更新。
const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', ...(input.ownerCanvasId ? { ownerCanvasId: input.ownerCanvasId } : {}), ...(input.runId ? { runId: input.runId } : {}), protocolVersion: input.protocolVersion ?? 1, generation: generation + 1, body: JSON.stringify(input.payload), operationIds: [...input.operationIds], canvasIds: [...input.canvasIds].sort(), expectedVersions: { ...input.expectedVersions }, phase: 'frozen', outcome: 'unresolved', attempt: 0 });
requests.set(request.id, request);
for (const id of input.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'frozen', requestId: request.id }));
changed(); return request;
},
/** 保守标记请求可能即将发出并增加 attempt；自身不发送，持久化后才允许外部 I/O。 */
markDispatch(id: Id) {
const request = requireRequest(id);
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。
* 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]);
@@ -100,6 +133,8 @@ export function createLedger(initial?: LedgerSnapshot) {
if (outcome === 'not-applied') for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'blocked', blockers: [error || 'not_applied'] }));
changed();
},
/** 仅凭完整闭合生命周期证明抵消连续未发意图；复验依赖、absent 基线、序列与墓碑。
* 标记 resolved/elided 并保留意图/本地生命周期，history 和资源 pin 不在这里删除。 */
elide(proof: ElisionProof) {
const ids = new Set(proof.operationIds), seen = new Set<Id>(), intents = proof.operationIds.map(id => {
const record = operations.get(id);
@@ -108,6 +143,7 @@ export function createLedger(initial?: LedgerSnapshot) {
return record.intent;
});
if (!equal(proof, proveElision(intents, proof.baseTombstones))) throw fault('invalid_elision');
// 之前已抵消的生命周期也占用 incarnation 高水位，不能因服务端从未见过就复用旧代次。
const previous = new Map<Id, number>();
for (const record of operations.values()) if (record.elision) for (const [id, incarnation] of Object.entries(record.elision.tombstones)) previous.set(id, Math.max(previous.get(id) ?? 0, incarnation));
assertElisionLineage(proof, intents, previous);
@@ -116,6 +152,7 @@ export function createLedger(initial?: LedgerSnapshot) {
for (const id of proof.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'resolved', resolution: 'elided', elisionId: proof.operationIds[0], ...(id === proof.operationIds[0] ? { elision: clone(proof) } : {}) }));
changed();
},
/** 对已确定 not-applied 的 run.start/cancel 解除控制依赖，不把它标作 applied。 */
resolveNoEffect(id: Id) {
const request=requireRequest(id);
if(request.outcome!=='not-applied'||!['run.start','run.cancel'].includes(request.kind??''))throw fault('invalid_no_effect_resolution');
@@ -123,6 +160,8 @@ export function createLedger(initial?: LedgerSnapshot) {
changed();
},
/** Local retirement is not an applied receipt. Validate the complete closure before mutating. */
/** 用户恢复决策的本地退休：先验证整组与依赖闭包，再标记 superseded/discarded/satisfied。
* 上传和执行结果未知的请求不能在这里被静默丢弃；replacement ID 必须是全新的。 */
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');
@@ -131,17 +170,22 @@ export function createLedger(initial?: LedgerSnapshot) {
if(!record||record.phase==='resolved'||record.intent.operationType==='asset.upload'||record.requestId&&(!request||request.phase!=='resolved'||request.outcome!=='not-applied'))throw fault('recovery_operation_not_safe',[decision.operationId]);
if(!['superseded','discarded','satisfied'].includes(decision.resolution)||(decision.resolution==='superseded')!==!!decision.replacementId||decision.replacementId&&(operations.has(decision.replacementId)||ids.has(decision.replacementId)))throw fault('invalid_recovery_decisions');
}
// 外部未解决操作若仍依赖这组意图，必须一并审阅，不能留下指向被退休操作的悬空工作。
for(const record of operations.values())if(record.phase!=='resolved'&&!ids.has(record.intent.id)&&record.intent.dependencies.some(id=>ids.has(id)))throw fault('recovery_dependency_closure',[record.intent.id]);
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 的派生索引。 */
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 的恢复依据。 */
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 均按可能已执行处理；这是保守风险依据，不是成功证明。 */
possiblyApplied(id: Id) { const request = requireRequest(id); return request.phase !== 'frozen' && request.outcome !== 'not-applied'; },
});
}
src/lib/canvas-runtime/reconcile.ts
+15
View File
@@ -1,3 +1,8 @@
/**
* 已确认基线和待同步意图的纯协调工具。先验证 canonical 文档/版本不可倒退，
* 再按原逻辑次序条件回放未解决的文档意图，形成 working 候选；不把旧 ACK 整份覆盖当前编辑。
* 这里只计算/校验，最终 document/runtime/ledger 的原子发布由 engine/editor 负责。
*/
import { withLocalLifecycles } from './elision.ts';
import { validateCreationSizeEffect } from '../canvas-core/creation-size-plan.ts';
import type { Document, Id, OperationIntent, Registry } from '../canvas-core/types.ts';
@@ -6,11 +11,14 @@ import { equal, fault } from '../canvas-core/values.ts';
import type { LedgerSnapshot } from './ledger.ts';
/** A complete read cannot rewind known scopes or change an unchanged-version canvas. */
/** 校验完整 fresh read：同项目、合法文档、活 canvas 版本齐全、已知版本和墓碑均不可倒退。
* 某 canvas 版本没变时，其记录/直属元素必须相同；全向量没变时墓碑集合也不能偷偷改变。 */
export function validateCanonicalRead(previous: Document, previousVersions: ReadonlyMap<Id,number>, next: Document, versions: ReadonlyMap<Id,number>, registry: Registry) {
if(next.projectId!==previous.projectId)throw fault('project_mismatch');
validateDocument(next,registry);
for(const [id,version]of versions)if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_canonical_version',[id]);
for(const canvas of next.canvases.keys())if(!versions.has(canvas))throw fault('missing_server_version',[canvas]);
// 记录服务声明“版本未变”的 owner，随后双向检查元素，既防遗漏删除也防夹带新增。
const unchanged=new Set<Id>();
for(const [id,version]of previousVersions){
if((versions.get(id)??-1)<version)throw fault('stale_canonical_version',[id]);
@@ -26,9 +34,12 @@ export function validateCanonicalRead(previous: Document, previousVersions: Read
}
/** The footprint follows ownership; geometry does not decide which canvas owns an object. */
/** 计算跨 canvas 操作 footprint：有已冻结意图范围时沿用，否则收集新旧 owner/父级/端点归属。
* 上传和 run.cancel 不占文档参与者；空间位置或画框相交不决定 owner。 */
export function operationCanvases(intent: OperationIntent, document: Document, after:Document=document): readonly Id[] {
if(intent.operationType==='asset.upload'||intent.operationType==='run.cancel')return[];
if (intent.canvasIds?.length) return intent.canvasIds;
// 字段变化不能只查 after；重挂/改端点时旧 owner 也参与版本条件和占用。
const ids = new Set<Id>();
for (const change of intent.changes) {
if (change.kind === 'canvas') {
@@ -52,7 +63,10 @@ export function operationCanvases(intent: OperationIntent, document: Document, a
}
/** Replay once, by logical sequence. A receipt can already contain a desired field value. */
/** 将本地生命周期水位叠到 ACK 副本，再按 sequence 回放 document.change/run.apply。
* 只有已满足的 after 可安全跳过，条件冲突仍拒绝；不会自动改写原意图或冻结请求。 */
export function replayPending(base: Document, ledger: LedgerSnapshot, registry: Registry): Document {
/** 已 resolved 项不重放；未冻结、冻结或 blocked 的未解决文档意图仍按原事实参与重建。 */
function* pending(){
for(const record of [...ledger.operations.values()].sort((a,b)=>a.intent.sequence-b.intent.sequence)){
if(record.phase==='resolved'||!['document.change','run.apply'].includes(record.intent.operationType))continue;
@@ -60,6 +74,7 @@ export function replayPending(base: Document, ledger: LedgerSnapshot, registry:
yield {changes:record.intent.changes,canvasIds:record.intent.canvasIds};
}
}
// 本地抵消可能从未进入服务器，水位只加到 working 候选，不能冒充服务器已确认墓碑。
const document=replayChanges(withLocalLifecycles(base,ledger),pending());
validateDocument(document,registry);
return document;
src/lib/canvas-runtime/scheduler.ts
+55
View File
@@ -1,17 +1,32 @@
/**
* 从 ledger 派生执行顺序的调度器：ready FIFO+queued 去重、依赖索引、canvas claims 与重试 min-heap。
* 所有权威变化通过 options.mutate 写回 Root；队列本身既不是历史，也不保存另一套任务状态。
* 发送前先持久化精确 body/key 和 dispatch 标记，结果协调及持久化成功后才释放占用。
*/
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 = {
admit?: (intent:OperationIntent)=>boolean;
/** 每次准入/唤醒读取当前权威快照，不信任入队时捕获的旧状态。 */
getLedger(): LedgerSnapshot;
/** 在 engine 的受控发布路径内修改账本，失败不能留下半份持久事实。 */
mutate(reduce: (ledger: Ledger) => void): void;
/** 等待当前 Root 可靠存储；失败保持请求身份与占用，不退回生成新请求。 */
persist(): Promise<void>;
createId(): Id;
participants(intent: OperationIntent): readonly Id[];
@@ -31,24 +46,37 @@ export type SchedulerOptions = {
};
/** 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 只跟踪当前异步句柄/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;
/** 总并发上限之外，上传独占一个 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);
@@ -56,6 +84,7 @@ export function createScheduler(options: SchedulerOptions) {
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;
@@ -71,6 +100,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，遇任务屏障或占用即停。
* 仅合并有相交参与者的 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;
@@ -90,10 +121,12 @@ export function createScheduler(options: SchedulerOptions) {
}
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);
@@ -101,6 +134,7 @@ export function createScheduler(options: SchedulerOptions) {
}
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);
@@ -109,20 +143,24 @@ 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 避免重复写事实。 */
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 句柄后在微任务中执行：dispatch 标记→持久化→submit/query→原子协调→持久化。 */
function launch(id: Id, action: 'submit' | 'query') {
if (active.has(id) || stopped || holds.size) return;
const controller = new AbortController();
@@ -132,6 +170,7 @@ export function createScheduler(options: SchedulerOptions) {
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)!;
@@ -139,18 +178,21 @@ export function createScheduler(options: SchedulerOptions) {
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 一直查，不换请求身份。
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);
@@ -161,10 +203,12 @@ export function createScheduler(options: SchedulerOptions) {
});
active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind });
}
/** 可重入保护下消费到期请求和最多 64 个 ready 项；等待者移到索引，不在 FIFO 无界反复轮询。 */
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;
@@ -187,6 +231,7 @@ export function createScheduler(options: SchedulerOptions) {
}
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); }
@@ -196,6 +241,7 @@ export function createScheduler(options: SchedulerOptions) {
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');
@@ -204,6 +250,7 @@ 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；失败阻塞原意图而非偷偷重新编另一请求。
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) }));
@@ -220,6 +267,7 @@ export function createScheduler(options: SchedulerOptions) {
}
}
return Object.freeze({
/** 首次启动（已运行时幂等）：先恢复全部未解决 claims，按原 retry/body/key 继续；解释器缺失仍保留占用。 */
start() {
if (closed) throw fault('scheduler_closed');
if (!stopped) return;
@@ -234,9 +282,12 @@ export function createScheduler(options: SchedulerOptions) {
}
pump();
},
/** 门禁事实变化后显式重试准入，未满足者继续等待，不制造定时器忙轮询。 */
refreshAdmission(){for(const id of [...admissionWaiters])enqueue(id);if(ready.size)requestPump();},
/** 新逻辑操作通知只推进派生索引，重复 notify 不重复注册依赖或入队。 */
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);
@@ -249,6 +300,7 @@ export function createScheduler(options: SchedulerOptions) {
};
},
/** 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.
@@ -262,7 +314,9 @@ export function createScheduler(options: SchedulerOptions) {
for (const id of indexed) enqueue(id);
pump();
},
/** 返回当前占用表的只读副本，调用者不能手改 claim 绕过未知请求。 */
getClaims: () => readMap(claims),
/** 等待当前微任务/pump/active 收尾，不主动等待未来 retry timer，因此不是“全部已同步”保证。 */
async settle() {
do {
if (scheduledPump) await scheduledPump;
@@ -270,6 +324,7 @@ 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。 */
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(); },
});
}