From 7e08dd996bf1cfda2f3d7d0d791fe7e1f15e6bc8 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 15:00:22 +0800 Subject: [PATCH] docs(runtime): explain engine image and import preparation lifecycle --- src/lib/canvas-runtime/engine.ts | 72 ++++++++++++++++++++++---------- 1 file changed, 50 insertions(+), 22 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 3018169..7949c03 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -823,7 +823,9 @@ export async function openProjectEngine(options: EngineOptions) { /** 统一摘要入口，继承 sourceDigest 的注入选择与项目生命周期中止。 */ async function digest(bytes:Blob,signal?:AbortSignal){return sourceDigest(bytes,signal);} /** 最多并发两个原文件摘要。只发现 waiting_digest 的上传事实，成功原子写摘要并解除阻塞。 - * 失败保留可识别原因并退出自动发现条件，等待显式 retry，而非无限计算同一损坏文件。 */ + * 失败保留可识别原因并退出自动发现条件，等待显式 retry，而非无限计算同一损坏文件。 + * digestTasks 只是本项目两项并发的索引；共享 digest pool 可能再排队，摘要可先于本地 Blob 落盘完成。 + * hash 与解除 waiting_digest 同根发布，通知 scheduler 后仍须过保存门禁才能上传。 */ function prepareDigests() { if (closing || closed) return; for (const resource of editor.getSnapshot().runtime.resources.values()) { @@ -846,20 +848,24 @@ export async function openProjectEngine(options: EngineOptions) { return Object.freeze({ ...previous, resources: readMap(resources), ledger: ledger.getSnapshot() }); }); if (result.status === 'rejected') throw fault(result.code, result.targets); + // 这里只通知已有上传操作变得可调度，不直接发送字节或生成新的 requestId。 scheduler.notify([resource.operationId]); }).catch(error => { if (!closing && !closed) { const code=(error as {code?:string}).code;mutate(ledger => ledger.block(resource.operationId, [code&&digestFailures.has(code)?code:'resource_bytes_unavailable'])); report(error); } }).finally(() => { digestTasks.delete(resource.id); prepareDigests(); }); digestTasks.set(resource.id, task); } } /** 并发两项图片尺寸准备，与上传摘要分别调度。优先解码本地原件，无字节时可用已存服务端尺寸。 - * image 状态先写成 pending，完成时只更新仍指向同一 image 记录的资源，失败不会覆盖后来重试。 */ + * image 状态先发布为 pending，完成时核 operationId 与原 image 对象身份，失败也不覆盖后来重试的记录。 + * 有 options.images 才启动本函数的解码链；其缺省时 stored 尺寸由 prepareDefaultSizes 直接采用。 + * imageTasks 在 finally 清理并继续发现下一项，任务完成不代表默认几何或上传已经完成。 */ function prepareImages(){ queueDefaultSizes(); if(closing||closed||recovery||!options.images)return; for(const resource of editor.getSnapshot().runtime.resources.values()){ if(imageTasks.size>=2)break; if(imageTasks.has(resource.id)||resource.image&&(resource.image.version!==1||resource.image.status!=='pending')||resource.retirement?.purgedAt!==undefined)continue; - // 先发布可恢复 pending，再由发布通知进入实际解码，避免隐形在途任务没有状态依据。 + // 缺 image 时本轮只发布 pending 并返回，由根订阅再次发现任务；不在同一发布栈里启动解码。 + // “可恢复”表示根中已有任务声明，可靠性仍需后续 persist，不要求解码开始前单独等磁盘回执。 if(!resource.image){editor.updateRuntime(previous=>{const current=previous.resources.get(resource.id);if(!current||current.image)return previous;const resources=new Map(previous.resources);resources.set(resource.id,freeze({...current,image:{version:1 as const,status:'pending' as const}}));return Object.freeze({...previous,resources:readMap(resources)});});return;} const task=(async()=>{ const bytes=fileWrites.get(resource.id)??await options.storage.readFile(leader,resource.id); @@ -873,7 +879,8 @@ export async function openProjectEngine(options: EngineOptions) { })().catch(error=>{if(closing||closed||recovery)return;const code=(error as{code?:string}).code??'image_decode_failed';editor.updateRuntime(previous=>{const current=previous.resources.get(resource.id);if(!current||current.operationId!==resource.operationId||current.image!==resource.image)return previous;const resources=new Map(previous.resources);resources.set(resource.id,freeze({...current,image:{version:1 as const,status:'failed' as const,code}}));return Object.freeze({...previous,resources:readMap(resources)});});}).finally(()=>{imageTasks.delete(resource.id);prepareImages();});imageTasks.set(resource.id,task); } } - /** 将多个尺寸补写通知合并为一次宏任务；不是网络重试时刻，也不忙等尺寸解码。 */ + /** 将多个尺寸补写通知合并为一个0ms timer；只要有 pending grant 就可排队，不保证尺寸现在已可用。 + * 定时器执行前先清 sizingTask，允许本轮发布触发后续 grant 的新检查；Promise 供关闭流程等待。 */ function queueDefaultSizes() { if (sizingTask || closing || closed || recovery || ![...editor.getSnapshot().runtime.resources.values()].some(resource => resource.sizing?.version === 1 && resource.sizing.grants.some(grant => grant.status === 'pending'))) return; // A scheduling yield, not a logical retry deadline: avoid a self-extending @@ -881,7 +888,9 @@ export async function openProjectEngine(options: EngineOptions) { sizingTask = new Promise(resolve => { setTimeout(() => { sizingTask = undefined; try { prepareDefaultSizes(); } catch (error) { report(error); } finally { resolve(); } }, 0); }); } /** 一次最多处理一项创建尺寸授权。仅在无手势/无恢复时，把原始比例适配到当前宽度和单位约束。 - * resolveCreationSize 验证目标代次与创建历史，补写同一历史单元；不能改写任意已编辑对象。 */ + * resolveCreationSize 验证目标代次与创建历史，补写同一历史单元；不能改写任意已编辑对象。 + * 尺寸已知且目标为 node 才尝试：保留当前宽度/图片比例，按单位尺度约束双边，再转为接口单位。 + * 成功的 historyEffect 在 reduceRuntime 中结清 grant；拒绝或无净变化则在下面显式结清，避免反复空转。 */ function prepareDefaultSizes() { if (closing || closed || recovery || sessions.getSnapshot().gesture) return; const root = editor.getSnapshot(); @@ -911,11 +920,13 @@ export async function openProjectEngine(options: EngineOptions) { }); if (result.status === 'rejected') report(fault(result.code, result.targets)); } - return; // One publication per pump; its notification schedules further grants. + // 每轮只尝试一项可处理 grant；可能还有结清状态的派生发布，后续项由根/手势/任务通知再次触发。 + return; } } } - /** 显式重试把 blocked grant 恢复为 pending，不改已应用项；重新排一次尺寸检查。 */ + /** 显式重试把 blocked grant 恢复为 pending 并清错误；不复活 revoked，不修改 applied 项。 + * 只重排尺寸检查，不重新上传/解码文件；入口调用者负责生命周期准入。 */ function retryDefaultSizes() { editor.updateRuntime(previous => { const resources = new Map(previous.resources); let changed = false; @@ -928,7 +939,8 @@ export async function openProjectEngine(options: EngineOptions) { } /** 将没有文档 changes 的取消/应用动作登记为逻辑操作。调用方先做业务校验， * apply 等待所有旧未决操作；cancel 不添加这些依赖，以便及时控制服务端任务。 - * 只更新账本，返回的 revision 仍是当前文档 revision，发送由后续调度完成。 */ + * 只更新 ledger/runtime，产生新 rootRevision 而文档 revision 保持原值；返回 applied 表示本地接纳。 + * id/sequence 只分配一次，依赖冻结在意图里；发送/查询/服务回执由后续 scheduler 使用相同身份处理。 */ function queueRunAction(operationType:'run.cancel'|'run.apply',action:RunCancelInput|RunApplyInput):Result { const root=editor.getSnapshot();let sequence=1;for(const record of root.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence+1); const id=createId(),intent={id,operationType,action,operationSchemaVersion:1,sequence,origin:'user',canvasIds:operationType==='run.apply'?[(action as RunApplyInput).targetCanvasId]:[],changes:[],dependencies:operationType==='run.cancel'?[]:[...root.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)} as OperationIntent; @@ -940,13 +952,16 @@ export async function openProjectEngine(options: EngineOptions) { /** 从唯一根读取当前作业，不以 runner 捕获的旧对象代替可恢复事实。 */ function getImport(id:Id){const job=editor.getSnapshot().runtime.imports?.get(id);if(!job)throw fault('import_missing',[id]);return job;} /** 仅发布作业元数据；generation 乐观校验防止旧异步结果覆盖暂停/取消/替换后的作业。 - * integrityOwner 只在本次同步更新内允许该 job 推进校验信息，finally 撤销权限。 */ + * integrityOwner 只在本次同步更新内允许该 job 推进校验信息，finally 撤销权限。 + * 此内部入口再查宿主 changes.apply 策略，原样采用 job（旧版仅升外层 version）；不是任意外来 job 的完整捕获器。 + * 不在此 persist 或调整传入 generation，调用方须先构造正确候选并等待所需可靠屏障。 */ function publishImport(job:ImportJob,expectedGeneration:number){ assertImportWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); integrityOwner=job.id;try{const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job.version===3?job:freeze({...job,version:3 as const}));return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);}finally{integrityOwner=undefined;} } /** 为新建检查点声明各层记录容器的所有权，便于增量恢复编码复用。 - * dataRecord/dataArray 不代表所有子数据也归这里所有，不跳过后续结构与内容校验。 */ + * dataRecord/dataArray 不代表所有子数据也归这里所有，不跳过后续结构与内容校验。 + * 仅重建 job/source/preparation 外壳并标记 allocationIds 容器，嵌套只读值可复用，未进行 JSON 深复制。 */ function importCheckpoint(job:ImportJob):ImportJob { return dataRecord(freeze({...job, ...(job.source?{source:dataRecord(freeze({...job.source}))}:{}), @@ -954,7 +969,8 @@ export async function openProjectEngine(options: EngineOptions) { })); } /** 将当前分块读取证明附入检查点；SHA/校验错误以根中最新值为准， - * 防止前台解析的旧 job 覆盖后台全文件验证刚发布的结果。 */ + * 防止前台解析的旧 job 覆盖后台全文件验证刚发布的结果。 + * context.reader 提供当前读证明；源 fileId 必须对应该 context，本函数不读字节或自行生成 SHA。 */ function checkpointImportSource(job:ImportJob):ImportJob{ if(job.source?.integrityVersion!==2)return job; const context=proofReaders.get(job.id),current=editor.getSnapshot().runtime.imports?.get(job.id)?.source; @@ -963,19 +979,23 @@ export async function openProjectEngine(options: EngineOptions) { return importCheckpoint({...job,source:{...source,readProof:context.reader.getProof(),...(current?.sha256?{sha256:current.sha256}:{}),...(current?.integrityError?{integrityError:current.integrityError}:{})}}); } /** 文件校验失败增加 generation，保留已取消/部分完成状态，其余转为暂停。 - * 错误写入源证据，既有已提交内容保留；短暂 hold 触发 runner 的暂停边界，不持有永久暂停租约。 */ + * 错误写入 source.integrityError 与 job.error，阻止封存/发送，既有已提交内容不回滚。 + * 短暂 hold 请求中止 runner 旧 pump 后立即释放；job 已 paused（或保留终态），不会因此自动续跑。 */ function failImportVerification(job:ImportJob,error:unknown){ const code=(error as {code?:string}).code??'import_source_verification_failed'; publishImport(importCheckpoint({...job,generation:job.generation+1,status:['cancelled','partial'].includes(job.status)?job.status:'paused',source:{...job.source!,integrityError:code},error:code}),job.generation); const release=importRunners.get(job.id)?.hold();release?.();report(error); } /** 每 job 单飞的后台完整性校验：先核对已有分块证明，再计算整文件 SHA。 - * 中止/关闭/全局 hold 时不把取消当损坏；真正失败记录证据，成功保存 seal 后才允许相应同步。 */ + * 中止/关闭/全局 hold 时不把取消当损坏；真正失败记录证据，成功保存 seal 后才允许相应同步。 + * 工作基于固定 Blob/reader，采纳时查来源身份而非要求 generation 不变，允许解析进度同期正常前进。 + * 成功封存沿用当前 job.generation，只替换完整性证据；publish 的授权窗口与 durableImportSeals 保存门仍生效。 */ function verifyImportSource(id:Id,context:{fileId:Id;bytes:Blob;reader:ReturnType}):Promise{ // 已中止的旧实例要先结束，再确认上下文仍有效才重建；不能并行验证两个同名来源。 const active=verificationTasks.get(id);if(active)return active.controller.signal.aborted?active.promise.then(()=>{if(!closing&&!closed&&!importHolds.size&&proofReaders.get(id)===context&&sourceNeedsSeal(getImport(id)))return verifyImportSource(id,context);}):active.promise; const admitted=getImport(id),controller=new AbortController(),signal=AbortSignal.any([controller.signal,digestLifecycle.signal]); - /** 每次跨 await 后按 fingerprint、fileId、reader 身份复查，不把旧文件结果写给新作业。 */ + /** 在采纳验证结果/记录错误时复查 fingerprint、fileId、reader 身份及 hold/关闭/取消信号。 + * 不比较解析 generation；本函数只取当前事实，不自动发布或锁住整个 job。 */ const current=()=>{signal.throwIfAborted();if(closing||closed||importHolds.size)throw fault('import_aborted');const job=getImport(id);if(job.sourceFingerprint!==admitted.sourceFingerprint||job.source?.fileId!==context.fileId||proofReaders.get(id)!==context)throw fault('import_identity_changed');return job;}; const promise=(async()=>{ let sha256:string; @@ -987,7 +1007,8 @@ export async function openProjectEngine(options: EngineOptions) { })().catch(error=>{if(!signal.aborted&&!closing&&!closed)report(error);}).finally(()=>{if(verificationTasks.get(id)?.controller===controller)verificationTasks.delete(id);}); verificationTasks.set(id,{controller,promise});return promise; } - /** 重建读取器前先中止并等验证结束、关闭 runner，最后释放分块证明缓存；持久作业不删除。 */ + /** 重建读取器前先请求中止并等待验证结束，再 close runner，最后释放 reader 缓存。 + * 删除的是 verification/runner/proof 执行索引，不删持久 job、源 Blob、已加入内容或已存 readProof。 */ async function resetImportReader(id:Id){ const active=verificationTasks.get(id);active?.controller.abort();if(active)await active.promise; const runner=importRunners.get(id);if(runner){await runner.close();importRunners.delete(id);} @@ -998,28 +1019,33 @@ export async function openProjectEngine(options: EngineOptions) { // 暂停期间被中断、尚未建立 runner 的任务要记下，释放 hold 后才能重新装配。 const deferredImportLoads=new Set(); /** 可嵌套全局导入暂停：中止准备/验证，给已装配 runner 各加一个 hold。 - * 释放只撤销本 token 的租约，再按根中状态启动需要恢复的任务；不把业务 status 改成 cancelled。 */ + * 释放只撤销本 token 的租约，再按根中状态启动需要恢复的任务；不把业务 status 改成 cancelled。 + * hold 本身同步返回，不等待这些任务退出；runner 的 hold 可嵌套，最后一个租约释放后才允许续跑。 */ function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);const token=Symbol(),releases=new Mapvoid>();importHolds.set(token,releases);for(const controller of preparationControllers.values())controller.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status)&&!importRunners.has(job.id))void loadImport(job.id);};} /** 从固定复制快照恢复规划。规划版本决定生成可流式读取的计划，或旧版 NDJSON 文件。 - * 期间不断保存 ID 分配/步骤检查点，保证恢复后新元素身份稳定；此阶段还未向画布加入元素。 */ + * 按检查点保存 ID 分配/步骤，续跑重用已保存的映射；尚未被 checkpoint 持久化的工作仍可重算。 + * 本阶段只发布 preparing/计划/源文件事实，不创建画布元素；实际内容随后由普通导入 runner 逐批加入。 */ async function materializeCopy(id:Id,bytes:Blob,controller:AbortController){ const admitted=getImport(id),preparation=admitted.preparation; if(!preparation||admitted.source?.kind!=='duplicate-snapshot')throw fault('invalid_copy_preparation'); - /** 复制规划只接受原 fingerprint 且仍 preparing 的作业；关闭、hold、取消均使旧结果失效。 */ + /** 要求同 fingerprint 且仍 preparing，另查 signal/关闭/hold；允许本规划器自己的 generation 前进。 + * 这不是逐字段验证全部 source 的入口，源文件稳定身份仍由装配与持久任务合同约束。 */ const check=()=>{const job=getImport(id);if(controller.signal.aborted||closing||closed||importHolds.size||job.status!=='preparing'||job.sourceFingerprint!==admitted.sourceFingerprint)throw fault('copy_aborted');return job;}; // 先保存快照作业再规划；v2 索引快照可按需读取，旧快照走完整解码。 await persist();check();const indexed=admitted.source.snapshotVersion===2?await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal):undefined,document=indexed?.document??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal); // 渐进模型还要证明复制快照对应合法来源范围；类型/复制声明变化则拒绝续用旧规划器。 validateLoadingCopySource(editor.getSnapshot(),admitted,document); if(copyRegistrySignature(document,options.registry)!==preparation.registrySignature)throw fault('copy_interpreter_changed'); - // 新规划器持久化拓扑与稳定 ID，不先展开所有复制数据为另一份 NDJSON。 + // v3/v4 准备协议产出 lazy emit 计划；v4 可配 format2 索引快照，避免再展开一份完整复制 NDJSON。 + // checkpoint 回调发布向前的步骤并 persist；最终返回的计划仍需下方另行发布，不能只依赖最后一次回调。 if(preparation.version===3||preparation.version===4){ const state=await prepareDuplicateStream(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner,topologyOnly:!!indexed},{signal:controller.signal,checkpointIntervalMs:50,maxStepsPerCheckpoint:8192,checkpoint:async state=>{ // 只采纳向前推进的 planner 检查点，保存完成后再继续规划，避免恢复丢失已分配身份。 const job=check(),previous=job.preparation!;if(previous.planner&&state.steps<=previous.steps)return; publishImport(importCheckpoint({...job,generation:job.generation+1,preparation:{...previous,steps:state.steps,allocationIds:[...state.elementIds,...state.canvasIds].map(pair=>pair[1]),planner:state}}),job.generation);await persist();check(); }}),job=check();assertNativeImportTarget(job,editor.getSnapshot().document); - // 将旧 ID → 新 ID 和计划一起发布为 running；移除 preparation 后由普通导入 runner 逐批落图。 + // 以 element:/canvas: 命名空间把旧 ID→新 ID 放进 stableIds，与 emit 计划一起转 running，清 preparation。 + // 原快照字节继续作读取源，parserVersion 改2表示按计划 ordinal 读取；此发布本身尚不等于已经落盘。 const stableIds=dataRecord(Object.fromEntries([...state.elementIds.map(([old,id])=>[`element:${old}`,id]),...state.canvasIds.map(([old,id])=>[`canvas:${old}`,id])])),next={...job,generation:job.generation+1,parserVersion:2,status:'running' as const,stableIds,source:{...job.source!,kind:'duplicate-plan' as const,plan:state,copy:dataRecord({sourceRevision:job.source!.copy!.sourceRevision,roots:dataArray(state.roots.map(old=>stableIds[`element:${old}`]))})}};delete next.preparation;delete next.error; publishImport(importCheckpoint(next),job.generation);return{job:getImport(id),bytes,document,indexed}; } @@ -1033,14 +1059,16 @@ export async function openProjectEngine(options: EngineOptions) { if(steps<=previous.steps&&allocationIds.length===previous.allocationIds.length)return; publishImport(importCheckpoint({...job,generation:job.generation+1,preparation:{...previous,steps:Math.max(previous.steps,steps),allocationIds:[...allocationIds]}}),job.generation);await persist();check(); }}); - // 旧解释器若消耗的 ID 数量变了，原分配顺序已不可信，不能静默生成另一份复制结果。 + // v1 续跑按 allocationIds 消费顺序重放分配；消耗数量不同就拒绝，防旧 checkpoint 被新解释器静默重解释。 + // 新 createId 分配不做外部回滚，持久化检查点才定义后续重开必须复用的身份集合。 check();if(preparation.version===1&&cursor!==allocationIds.length)throw fault('copy_interpreter_changed'); // 旧版计划编码成固定 NDJSON 并算 SHA；异步完成后再次校验目标入口及当前业务策略。 const encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),sha256=await sourceDigest(encoded,controller.signal),job=check();assertImportWritable();assertNativeImportTarget(job,editor.getSnapshot().document); const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); const fileId=createId(),runtime=editor.getSnapshot().runtime;if(!fileId||fileWrites.has(fileId)||runtime.resources.has(fileId)||[...(runtime.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict'); const stableIds=Object.fromEntries([...Object.entries(plan.elementIds).map(([old,id])=>[`element:${old}`,id]),...Object.entries(plan.canvasIds).map(([old,id])=>[`canvas:${old}`,id])]); - // 新源替换旧快照时把原文件保留在 previousFiles，后续清理仍需保留期与引用证明。 + // 旧路径把复制结果写成新的固定 NDJSON 源，指纹绑定新 fileId/长度；不是原地改写旧快照 Blob。 + // 原文件进入 previousFiles，引用/保留期证明仍约束后续清理，不能在转换成功时直接删除。 const next={...job,generation:job.generation+1,status:'running' as const,stableIds,sourceFingerprint:`local-file:${fileId}:${encoded.size}`,source:{...job.source!,previousFiles:[...(job.source!.previousFiles??[]),{fileId:job.source!.fileId,byteLength:job.source!.byteLength,...(job.source!.sha256?{sha256:job.source!.sha256}:{})}],kind:'duplicate-ndjson' as const,fileId,byteLength:encoded.size,sha256,integrityVersion:1 as const,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:plan.roots}}};delete next.preparation;delete next.error; // 先把字节放入待保存集再发布引用它的根，发布失败则撤回本次字节暂存。 fileWrites.set(fileId,encoded);try{publishImport(importCheckpoint(next),job.generation);}catch(error){fileWrites.delete(fileId);throw error;}