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

docs(runtime): explain engine image and import preparation lifecycle

Browse Source
This commit is contained in:
Codex Bot
2026-11-12 15:00:22 +08:00
parent b3a634ce6b
commit 7e08dd996b
1 changed files with 50 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+50 -22
View File
@@ -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<typeof createImportReadProof>}):Promise<void>{
// 已中止的旧实例要先结束，再确认上下文仍有效才重建；不能并行验证两个同名来源。
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<Id>();
/** 可嵌套全局导入暂停：中止准备/验证，给已装配 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 Map<Id,()=>void>();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;}