diff --git a/src/lib/canvas-runtime/applied-recovery.ts b/src/lib/canvas-runtime/applied-recovery.ts index 342f5a2..ca68f7c 100644 --- a/src/lib/canvas-runtime/applied-recovery.ts +++ b/src/lib/canvas-runtime/applied-recovery.ts @@ -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(); 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) encoder.encode(JSON.stringify(value)).byteLength; +/** wire 中保留的 logical ID 与对应条件 changes;不是新的本地历史单元。 */ export type WireOperation = { readonly id: Id; readonly changes: readonly Change[] }; +/** 对比规范化表的键值效果,忽略表迭代顺序但包含实际 value;墓碑也用同规则检查。 */ const sameMap = (a: ReadonlyMap, b: ReadonlyMap) => 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(); const canvases = new Map(acknowledged.canvases.has(acknowledged.projectId) ? [[acknowledged.projectId, acknowledged.canvases.get(acknowledged.projectId)!]] : []); const tombstones = new Map(), seen = new Set(); @@ -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); diff --git a/src/lib/canvas-runtime/change-rebase.ts b/src/lib/canvas-runtime/change-rebase.ts index 35fd865..9071b89 100644 --- a/src/lib/canvas-runtime/change-rebase.ts +++ b/src/lib/canvas-runtime/change-rebase.ts @@ -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; readonly after:ReadonlyMap; readonly transforms:ReadonlyMap; readonly beforeTransforms:ReadonlyMap }; /** 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 { const elements=new Map(document.elements),canvases=new Map(document.canvases),contexts=new Map(); const cursor={...document,elements,canvases}; for(const intent of [...intents].reverse()){ + // 收集自身、连线端点与 canvas 入口,捕获足够的身份和坐标上下文,不复制全项目到每条意图。 const ids=new Set(),owners=new Set(); 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();for(const id of ids){const record=elements.get(id);if(record){after.set(id,record);owners.add(record.ownerCanvasId);}} const transforms=new Map();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,originalBefore:Extract,originalAfter:Extract,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,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`${change.kind}:${(change as Extract).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(),ready:number[]=[]; batch.forEach((change,index)=>{ const dependencies=new Set(); @@ -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(),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)[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'&¤t.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)}; } diff --git a/src/lib/canvas-runtime/elision.ts b/src/lib/canvas-runtime/elision.ts index 91c1448..b42436f 100644 --- a/src/lib/canvas-runtime/elision.ts +++ b/src/lib/canvas-runtime/elision.ts @@ -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>; readonly tombstones: Readonly> }; /** 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>): 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(), canvases = new Set(); + // 用起始+本段删除水位阻止创建/恢复复用已死亡 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 = new Set()): ElisionProof | undefined { const selected: OperationIntent[] = [], seen = new Set(), canvases = new Set(), bases: Record = 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 = {}; 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(); @@ -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): void { const bases: Record = 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): 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(); 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(), 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): vo } /** Local high-water marks belong to working state, never to the server ACK document. */ +/** 只给 working 候选补本地墓碑最大值,不修改源文档,也不向服务器 ACK 事实表灌入本地证明。 */ export function withLocalLifecycles(document: Document, ledger: Pick): 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)); diff --git a/src/lib/canvas-runtime/ledger.ts b/src/lib/canvas-runtime/ledger.ts index 3640f0a..f6d989b 100644 --- a/src/lib/canvas-runtime/ledger.ts +++ b/src/lib/canvas-runtime/ledger.ts @@ -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> }; +/** generation 随账本事实改变递增,durableGeneration 仅确认已实际存储的前缀。 */ 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;不能用最后整节点覆盖替代字段链。 */ export function compressChanges(changes: readonly Change[], options: { preserveLifecycle?: boolean; preserveChecks?: boolean } = {}): readonly Change[] { + // 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; 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(); + /** 依赖不能回到待添加意图或已在当前路径上的节点,避免 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>) { 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>; 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(), 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(); 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'; }, }); } diff --git a/src/lib/canvas-runtime/reconcile.ts b/src/lib/canvas-runtime/reconcile.ts index dfb7feb..6168ddb 100644 --- a/src/lib/canvas-runtime/reconcile.ts +++ b/src/lib/canvas-runtime/reconcile.ts @@ -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, next: Document, versions: ReadonlyMap, 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(); for(const [id,version]of previousVersions){ if((versions.get(id)??-1)(); 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; diff --git a/src/lib/canvas-runtime/scheduler.ts b/src/lib/canvas-runtime/scheduler.ts index e0fcce6..0a1eab7 100644 --- a/src/lib/canvas-runtime/scheduler.ts +++ b/src/lib/canvas-runtime/scheduler.ts @@ -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> }; +/** 可注入时钟和 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) }; +/** mutate 回调内使用的临时可变账本接口,不直接暴露其内部表。 */ type Ledger = ReturnType; +/** engine 提供准入、纯编译、原子账本更新、持久屏障、外部端口和完成协调。 + * complete 必须连回执/基线/working/ledger 一起处理,不允许仅把请求先标成功。 */ export type SchedulerOptions = { admit?: (intent:OperationIntent)=>boolean; + /** 每次准入/唤醒读取当前权威快照,不信任入队时捕获的旧状态。 */ getLedger(): LedgerSnapshot; + /** 在 engine 的受控发布路径内修改账本,失败不能留下半份持久事实。 */ mutate(reduce: (ledger: Ledger) => void): void; + /** 等待当前 Root 可靠存储;失败保持请求身份与占用,不退回生成新请求。 */ persist(): Promise; 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(); + // FIFO 保存待处理逻辑 ID,queued 防重复;claims 是 canvas→request,多个参与者一次认领。 const ready = createDeque(), queued = new Set(), claims = new Map(); + // waiters 按被占 canvas 等待,dependents 按逻辑依赖等待,ordered 保留全局序列以挑连续批。 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。 const tokens = new Map(), active = new Map; kind: FrozenRequest['kind'] }>(); + // 多个恢复/关闭流程可独立 hold;只有最后一个 token 释放才继续准入,已启动工作仍正常收尾。 const holds = new Set(); const ordered: { id: Id; sequence: number }[] = []; let scheduledPump: Promise | undefined; + // 上传 lane 单并发等容量限制不能让到期请求反复立即入堆,先 deferred,active 完成再唤醒。 const laneWaiters = new Map, Set>(), deferredRetries = new Map(); 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): 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(); 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(); }, }); }