From c3b7cf93d6755adc5098a33b77fcf9df406a2caa Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 01:00:57 +0800 Subject: [PATCH] docs(canvas): explain engine recovery public APIs and shutdown --- src/lib/canvas-runtime/engine.ts | 81 +++++++++++++++++++++++++++++++- 1 file changed, 79 insertions(+), 2 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 52635ab..3a8d155 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -1142,65 +1142,91 @@ export async function openProjectEngine(options: EngineOptions) { scheduler.start(); prepareDigests();prepareImages(); runTracker.notify([...initial.runs.keys()]); // 终态作业若仍欠 seal 也装配验证；running/preparing 则恢复尚未完成的内容处理。 for(const job of initial.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status))void loadImport(job.id); + // 对外只暴露冻结的控制面；根快照由 editor 读取，异步执行器和持久化内部表不直接交给界面修改。 return Object.freeze({ editor, sessions, imports, + /** 渐进加载控制面：accept 合入同一流的前缀，complete 提交完整结果，resume 重新读取权威模型。 */ loading:Object.freeze({getState:()=>editor.getSnapshot().runtime.loading, + /** 同一时间共享一个恢复读取。v3+ 等旧请求确认后先 hold 新发送，再读服务端， + * 避免读取期间又发出写入造成模型包含范围不明；finally 释放本次读取的暂停租约。 */ resume():Promise{ if(loadingRead)return loadingRead; const task=(async()=>{assertWritable(false);if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};if(!options.persistence.read)throw fault('canonical_reader_unavailable');const controller=new AbortController();loadingReadController=controller;let releaseRead:(()=>void)|undefined; try{if((editor.getSnapshot().runtime.loading?.version??0)>=3)await new Promise((resolve,reject)=>{ + // 监听请求是否已全部 resolved；满足条件时同步取得 hold，再解除监听，关闭条件检查与准入之间的缝隙。 let unsubscribe=()=>{};const finish=(error?:unknown)=>{unsubscribe();controller.signal.removeEventListener('abort',abort);error===undefined?resolve():reject(error);},abort=()=>finish(controller.signal.reason),check=()=>{if(![...editor.getSnapshot().runtime.ledger.requests.values()].some(request=>request.phase!=='resolved')){releaseRead=scheduler.hold();finish();}}; unsubscribe=editor.subscribe(check);controller.signal.addEventListener('abort',abort,{once:true});if(controller.signal.aborted)abort();else check(); });const read=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertWritable(false);controller.signal.throwIfAborted();if(!editor.getSnapshot().runtime.loading)return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:[]};const result=completeLoading(read.document,read.versions,{fresh:true});if(result.status==='applied')await persist();return result;}finally{releaseRead?.();if(loadingReadController===controller)loadingReadController=undefined;} })();loadingRead=task.finally(()=>{if(loadingRead===wrapped)loadingRead=undefined;});const wrapped=loadingRead;return wrapped; }, + /** 校验来源前缀只能合法扩展，再合并期间已确认的修改并重放 pending； + * 原始读取 nextRead 与 working 分开保存，本地编辑不冒充服务端新增内容。 */ accept(document:Document,versions:ReadonlyMap,state:LoadingState):Result{ try{assertWritable(false);const root=editor.getSnapshot(),previous=root.runtime;if(!previous.loading)throw fault('loading_not_active');const source=previous.loadingRead;validatePrefixExtension(source?.document??previous.acknowledged,source?.versions??previous.versions,previous.loading,document,versions,state,options.registry);const nextRead=Object.freeze({document:createDocument(document.projectId,[...document.canvases.values()],[...document.elements.values()],document.tombstones),versions:readMap(versions)}),proposal=mergeLoadingRead(previous,nextRead,options.registry),acknowledged=proposal.document,working=replayPending(acknowledged,previous.ledger,options.registry);acceptingLoading=true;return editor.reconcile(working,Object.freeze({...previous,loading:freeze(clone(state)),...(state.version>=3?{loadingRead:nextRead}:{}),versions:readMap(proposal.versions),acknowledged}),root.rootRevision);}catch(error){return reject(error);}finally{acceptingLoading=false;} }, + /** 成功退出 loading 后等待该根可靠保存再返回；校验拒绝不发布半份完整模型。 */ async complete(document:Document,versions:ReadonlyMap,settings:{fresh?:boolean;sourceToken?:string}={}):Promise{const result=completeLoading(document,versions,settings);if(result.status==='applied')await persist();return result;}, }), + // 清理与库存诊断复用上面定义的受控流程；getResourcePins 每次由当前根和瞬态租约派生。 collectResources,collectImportSources,inventoryLocalFiles,collectOrphanFiles, getResourcePins:()=>resourceRetentionPins(editor.getSnapshot(),options.registry,displayPins,admissionPins(editor.getSnapshot())), - /** Read/preview under a scheduling hold. No request is rewritten or declared resolved here. */ + /** 暂停调度/生成轮询/导入，排空在途协调并保存，再读取服务端模型生成审阅会话。 + * prepare 本身不替用户丢弃编辑或改写原请求；发现冲突只返回诊断，发布发生在 apply/resolve。 */ async prepareRecovery(): Promise { assertWritable(false); if(!options.persistence.read)throw fault('canonical_reader_unavailable'); + // 在任何 await 前取得三类暂停，防止等待一条执行链时另一条继续准入新任务。 const releaseScheduler=scheduler.hold(),releaseRuns=runTracker.hold(),releaseImports=holdImports(),controller=new AbortController(); + /** 会话 release 只释放自己持有的暂停且幂等；准备阶段 cancel 还中止网络读取。 */ const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();releaseImports();queueMicrotask(prepareImages);},async cancel(){controller.abort();active.release();}}; recovery=active; + // 先取消交互预览，防止用户仍在进行的手势在恢复规划期间提交旧几何。 sessions.cancel('recovery'); + /** 跨 await 后检查会话身份，已关闭或被撤销的会话不能继续发布结果。 */ const assertCurrent=()=>{if(closing||closed)throw fault('engine_closing');if(recovery!==active||controller.signal.aborted)throw fault('recovery_session_closed');}; try { - // Both holds are acquired before awaiting either lane; already admitted callbacks finish normally. + // 所有 hold 已在上面取得；这里只等已经准入的回调正常结束，不中途丢弃它们的协调结果。 await Promise.all([...importLoads.values()]);assertCurrent();await Promise.all([scheduler.settle(),runTracker.settle(),...[...importRunners.values()].map(runner=>runner.settle())]);assertCurrent(); while(digestTasks.size){await Promise.all([...digestTasks.values()]);assertCurrent();} await persist();assertCurrent(); + // 从已知 applied 回执推导可采用的候选证据；先保存现有根，不将未验证的候选提前发布。 const staged=stageAppliedRecovery(editor.getSnapshot(),{persistence:options.persistence,runs:options.runs,assets:options.assets,now:runClock.now()}); const root=staged.snapshot,previous=root.runtime; + // 读取后的根 revision 必须仍一致；后续计划不能建立在与当前状态不同的旧底稿上。 const loaded=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertCurrent(); if(editor.getSnapshot().rootRevision!==root.rootRevision)throw fault('stale_plan'); if(previous.loading)validateLoadingCompletion(previous.acknowledged,previous.versions,previous.loading,loaded.document,loaded.versions,options.registry,true);else validateCanonicalRead(previous.acknowledged,previous.versions,loaded.document,loaded.versions,options.registry); const acknowledged=createDocument(loaded.document.projectId,[...loaded.document.canvases.values()],[...loaded.document.elements.values()],loaded.document.tombstones); + // 权威模型还必须包含此前已知生效的回执证据，不能借恢复把已确认写入抹掉。 const canonical=Object.freeze({document:acknowledged,versions:readMap(loaded.versions)});staged.assertCanonical(canonical); let working:Document|undefined,conflict:RecoverySession['conflict']; + // 条件重放若失败，只记录冲突；原 working/history/requests 保留供用户选择。 try{working=replayPending(acknowledged,previous.ledger,options.registry);}catch(error){const rejected=reject(error);conflict=freeze({code:rejected.code,targets:rejected.targets});} let committed:ConflictPlan|'refresh'|undefined,committedIds:readonly Id[]=[]; + // 只接受本会话实际生成的 plan 对象，拒绝调用方拼接或跨会话重用的方案。 const plans=new WeakSet(); + /** 已发布后先过保存屏障，再恢复调度/轮询并释放会话。 + * 保存失败继续持有暂停；再次调用可重试屏障，不重复发布相同恢复操作。 */ const finish=async()=>{if(previous.loading&&!editor.getSnapshot().runtime.loading)loadingCompletionRevision=editor.getSnapshot().rootRevision;await persist();assertCurrent();await scheduler.resume();assertCurrent();scheduler.notify(committedIds);runTracker.notify(staged.runIds);active.release();return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:committedIds};}; + // 返回会话后，未发布可直接退出；已经发布则必须完成保存，cancel 不等于回滚已发布恢复。 active.cancel=async()=>{if(recovery!==active)return;if(committed){await finish();return;}active.release();}; return Object.freeze({baseRootRevision:root.rootRevision,canonical,acknowledgedRequestIds:staged.requestIds, pendingOperationIds:Object.freeze([...previous.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)), ...(conflict?{conflict}:{}), + /** 在固定根上纯计算明确取舍的恢复方案，分配新操作身份并保存可审计备份。 */ plan(choices:ReadonlyMap){ assertCurrent();if(committed)throw fault('recovery_already_committed');if(editor.getSnapshot().rootRevision!==root.rootRevision)throw fault('stale_plan'); const plan=planConflictRecovery(root,canonical,choices,options.registry,createId,options.historyLimit??100);plans.add(plan);return plan; }, + /** 采用本会话方案：重查策略/生成输出适用性，发布文档、历史、账本、审计及资源 pin。 + * 同一方案已发布时只重试 finish；不能接着换另一方案覆盖第一次发布。 */ async resolve(plan:ConflictPlan):Promise{ assertCurrent();if(!plans.has(plan))throw fault('invalid_recovery_plan');if(committed&&committed!==plan)throw fault('recovery_already_committed'); if(!committed){ const permission=options.policy?.({type:'changes.apply',changes:[]},plan.document);if(permission&&!permission.enabled)return{status:'rejected',code:permission.code,targets:permission.targets}; for(const id of plan.operationIds){const intent=plan.operations.get(id)!;const allowed=options.policy?.({type:'changes.apply',changes:intent.changes},plan.document);if(allowed&&!allowed.enabled)return{status:'rejected',code:allowed.code,targets:allowed.targets}; if(intent.operationType==='run.apply'){const run=previous.runs.get(intent.action.runId);if(!run||run.blocked||run.status!=='conflict'||!runContainsTarget(run,intent.action.target))return{status:'rejected',code:'output_not_applicable',targets:[intent.action.target.id]};}} + // 恢复证据和原资源保持显式引用；用户以后仍可追查取舍，不因清理丢失备份字节。 const recoveries=new Map(previous.recoveries),resourcePins=new Map(previous.resourcePins),conflicts=new Map(previous.conflicts);recoveries.set(plan.id,plan.audit);resourcePins.set(`recovery:${plan.id}`,plan.resourceIds);conflicts.delete('refresh'); for(const decision of plan.audit.decisions){const requestId=previous.ledger.operations.get(decision.operationId)?.requestId;if(requestId)conflicts.delete(requestId);} const result=editor.reconcile(plan.document,Object.freeze({...previous,ledger:plan.ledger,loading:undefined,loadingRead:undefined,acknowledged,versions:canonical.versions,recoveries:readMap(recoveries),resourcePins:readMap(resourcePins),conflicts:readMap(conflicts)}),root.rootRevision,plan); @@ -1208,6 +1234,7 @@ export async function openProjectEngine(options: EngineOptions) { } return finish(); }, + /** 无需用户替代决策时采用条件重放结果；存在诊断冲突则拒绝，不自动扩大“保留本地”的语义。 */ async apply():Promise{ assertCurrent(); if(committed&&committed!=='refresh')throw fault('recovery_already_committed'); @@ -1221,15 +1248,18 @@ export async function openProjectEngine(options: EngineOptions) { // Persistence failure retains the hold; apply/cancel can retry this barrier without republishing. return finish(); }, + // 使用更新后的 cancel 语义，已提交会话仍需先完成可靠保存。 cancel:()=>active.cancel(), }); }catch(error){active.release();throw error;} }, + /** 读取当前项目画布的模型目录，经过共享网络门；不创建生成任务。 */ models(boardId: Id, signal: AbortSignal) { if(closing||closed||!options.runs)return Promise.reject(fault('run_interpreter_unavailable')); if(!editor.getSnapshot().document.canvases.has(boardId))return Promise.reject(fault('canvas_missing')); return network.run(signal,()=>options.runs!.catalog(boardId,signal)); }, + /** 校验任务和业务权限，拒绝重复未决取消动作，再将取消意图入账；返回不代表服务端已取消。 */ cancelRun(runId:Id):Result { try{ assertWritable();if(!options.runs)throw fault('run_interpreter_unavailable'); @@ -1239,6 +1269,8 @@ export async function openProjectEngine(options: EngineOptions) { return queueRunAction('run.cancel',{runId,boardId:run.boardId}); }catch(error){return reject(error);} }, + /** 用户明确采用冲突任务的一份输出。校验目标代次/任务成员/输出身份并防重复， + * 通过普通数据编辑取得 history；若内容本已相同，仍登记空 changes 的 run.apply 控制动作。 */ applyRunOutput(runId:Id,target:RunApplyInput['target'],outputId:Id):Result { try{ assertWritable();if(!options.runs)throw fault('run_interpreter_unavailable');if(applying)throw fault('reentrant_dispatch'); @@ -1246,15 +1278,19 @@ export async function openProjectEngine(options: EngineOptions) { if(!run||run.blocked||!runContainsTarget(run,target)||!element||element.incarnation!==target.incarnation)throw fault('target_changed'); if((run.appliedOutputIds??[]).includes(outputId))throw fault('output_already_applied'); if([...root.runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'&&(record.intent.operationType==='run.apply'||record.intent.operationType==='run.cancel')&&record.intent.action.runId===runId))throw fault('run_action_pending'); + // 只能从当前任务保存的输出事件选择内容，不接受任意调用方提交的一段伪造结果。 const event=run.outputEvents.find(event=>event.node_id===target.id&&event.output&&typeof event.output==='object'&&!Array.isArray(event.output)&&event.output.output_id===outputId),output=event?.output as {[key:string]:Json}|undefined; if(!output||run.status!=='conflict')throw fault('output_not_applicable'); const patch:Data|undefined=output.kind==='text'&&typeof output.text==='string'?{text:output.text}:output.kind==='image'&&typeof output.src==='string'&&typeof output.assetId==='string'?{src:output.src,assetId:output.assetId}:undefined;if(!patch)throw fault('invalid_run_output'); + // applying 只在本次同步 dispatch 内让 prepareIntent 将数据修改标成 run.apply，finally 总会清理。 const action:RunApplyInput={runId,boardId:run.boardId,targetCanvasId:element.ownerCanvasId,target,outputId};applying=action; const result=editor.dispatch({type:'element.data',target,patch},{label:'Apply generated result'}); if(result.status==='rejected'||result.operationIds.length)return result; return queueRunAction('run.apply',action); }catch(error){return reject(error);}finally{applying=undefined;} }, + /** 登记生成启动意图，真正受理/执行由服务完成。只允许当前已知的支持类型节点， + * 启动依赖所有旧未决操作，保证服务端生成所见输入先包含当前编辑；不在此直接请求模型。 */ startRun(input: RunInput): Result { try { assertWritable(); @@ -1264,16 +1300,20 @@ export async function openProjectEngine(options: EngineOptions) { if(!element||element.incarnation!==input.target.incarnation||element.ownerCanvasId!==input.boardId||element.structure!=='node'||options.registry.resolve(element)!=='known'||!['text','plain-text','script','image','plain-image'].includes(element.componentId))throw fault('target_changed'); const allowed=editor.can({type:'element.data',target:input.target,patch:{ai:{enabled:true,modelId:input.modelId,prompt:input.prompt,parameters:input.parameters,propagate:input.propagate}}}); if(!allowed.enabled)throw fault(allowed.code,allowed.targets); + // 未决启动、正在执行、还有尾页或待协调页的任务都占用目标，避免同一输入重复生成。 if([...root.runtime.ledger.operations.values()].some(record=>record.intent.operationType==='run.start'&&record.phase!=='resolved'&&record.intent.run.target.id===input.target.id))throw fault('node_busy'); if([...root.runtime.runs.values()].some(run=>runContainsTarget(run,input.target)&&(run.hasMore||run.pendingPage||['queued','running'].includes(run.status))))throw fault('node_busy'); const id=createId(),dependencies=[...root.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id); let sequence=1;for(const record of root.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence+1); + // 将输入复制为稳定 JSON 并登记可恢复逻辑事实；此处不修改元素内容或新增一条文档编辑历史。 const result=editor.updateRuntime(previous=>{const ledger=createLedger(previous.ledger);ledger.add(freeze({id,operationType:'run.start',operationSchemaVersion:1,sequence,origin:'user',canvasIds:[...root.document.canvases.keys()].sort(),changes:[],dependencies,run:JSON.parse(JSON.stringify(input))}),{});return Object.freeze({...previous,ledger:ledger.getSnapshot()});},root.rootRevision); if(result.status==='rejected')return result; queueMicrotask(()=>{if(!closing&&!closed)scheduler.notify([id]);}); return{status:'applied',revision:root.revision,operationIds:[id]}; }catch(error){return reject(error);} }, + /** 用新的权威模型刷新 ACK，再条件重放本地未决修改；未确认请求存在时拒绝模糊刷新。 + * 与 prepareRecovery 不同，此入口只尝试安全重放，冲突时保留文档并登记诊断。 */ refresh(document: Document, versions: ReadonlyMap): Result { if(editor.getSnapshot().runtime.loading)return completeLoading(document,versions,{fresh:true}); const root = editor.getSnapshot(), previous = root.runtime; @@ -1281,6 +1321,7 @@ export async function openProjectEngine(options: EngineOptions) { assertWritable(); if ([...previous.ledger.requests.values()].some(request => request.phase !== 'resolved')) throw fault('refresh_waiting_for_receipts'); validateCanonicalRead(previous.acknowledged,previous.versions,document,versions,options.registry); + // 内容、版本与墓碑都完全一致时直接返回，不凭空增加一次根发布。 if (equal([...previous.versions], [...versions]) && equal([...previous.acknowledged.canvases], [...document.canvases]) && equal([...previous.acknowledged.elements], [...document.elements]) && equal([...previous.acknowledged.tombstones], [...document.tombstones])) return { status: 'applied', revision: root.revision, operationIds: [] }; const acknowledged = createDocument(document.projectId, [...document.canvases.values()], [...document.elements.values()], document.tombstones); const working = replayPending(acknowledged, previous.ledger, options.registry); @@ -1292,50 +1333,65 @@ export async function openProjectEngine(options: EngineOptions) { return result; } }, + /** 查询图片暂存能力；渐进加载仅 v6 已完整画布开放，真正事务还需经过 editor 策略校验。 */ canStageFiles(canvasId:Id){try{assertWritable(false);if(!options.assets)throw fault('asset_interpreter_unavailable');const snapshot=editor.getSnapshot();if(!snapshot.document.canvases.has(canvasId))throw fault('canvas_not_loaded',[canvasId]);if(snapshot.runtime.loading&&(snapshot.runtime.loading.version!==6||!loadingQueryCoverage(snapshot).completeCanvasIds.includes(canvasId)))throw fault('loading_scope_incomplete',[canvasId]);const permission=options.policy?.({type:'changes.apply',changes:[]},snapshot.document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);return{enabled:true as const};}catch(error){const result=reject(error);return{enabled:false as const,code:result.code,targets:result.targets};}}, + /** 同步图片加入入口：校验字节/身份，暂存 Blob，再由 callback 在同一事务创建资源引用。 + * reducer 同根建立上传意图和资源记录，后续异步摘要/上传；事务失败清理未被根采用的字节。 */ stageFiles(files: readonly StagedFile[], callback: (resources: readonly ResourceHandle[], transaction: Transaction) => unknown) { assertWritable(false);if(editor.getSnapshot().runtime.loading&&editor.getSnapshot().runtime.loading?.version!==6)throw fault('model_loading'); if (staged.length||adopting) throw fault('reentrant_dispatch'); if (!options.assets) throw fault('asset_interpreter_unavailable'); + // 每份文件分配独立 resource ID 和 upload operation ID；图片解码能力存在时同时登记 pending 状态。 const resources = files.map(file => { if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name || file.defaultSize !== undefined && typeof file.defaultSize !== 'boolean') throw fault('invalid_staged_file'); return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size,...(options.images?{image:{version:1 as const,status:'pending' as const}}:{}) }); }); if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || editor.getSnapshot().runtime.assetBindings?.has(resource.id) || editor.getSnapshot().runtime.orphanFiles?.has(resource.id) || fileWrites.has(resource.id) || importFiles(editor.getSnapshot().runtime.imports).has(resource.id))) throw fault('resource_identity_conflict'); + // callback 必须在同步事务窗口使用这些 handles；staged 仅作 reducer 上下文，不另建全局文档。 staged = resources; resources.forEach((resource, index) => { fileWrites.set(resource.id, files[index].bytes);if(files[index].defaultSize)stagedDefaultSizes.add(resource.id); }); try { const handles = freeze(resources.map(resource => ({ id: resource.id, src: options.assets!.sourceURL(resource.id) }))); + // 事务的逻辑操作与资源上传依赖一起发布；空事务不保留这次无引用的待写字节。 const result = editor.transaction(tx => callback(handles, tx), { label: 'Add images' }); if (result.status === 'rejected' || !result.operationIds.length) for (const resource of resources) fileWrites.delete(resource.id); return result; } finally { staged = []; stagedDefaultSizes.clear();for (const resource of resources) if (!editor.getSnapshot().runtime.resources.has(resource.id)) fileWrites.delete(resource.id); } }, + /** 复用同工作区已存在的图片：先走服务授权/取字节，再用同步事务建立本项目绑定。 + * 这条路径不重复创建上传任务；采用失败则还原原来的 adoptedFiles 暂存值。 */ async adoptResource(id:Id,callback:(resource:ResourceHandle&{readonly asset:StoredAsset},transaction:Transaction)=>unknown,settings:{signal?:AbortSignal}={}):Promise{ assertWritable(false);if(!depot||!options.assets?.authority||!options.storage.assetDepot)throw fault('asset_depot_unavailable');if(editor.getSnapshot().runtime.loading&&editor.getSnapshot().runtime.loading?.version!==6)throw fault('model_loading'); if(staged.length||adopting)throw fault('reentrant_dispatch'); const controller=new AbortController(),signal=AbortSignal.any([controller.signal,...(settings.signal?[settings.signal]:[])]); const task=(async()=>{ + // 即使仓库缓存命中也要通过本次授权流程；等待期间的关闭、身份变化和取消均需重新检查。 const authorized=await depot.authorize(leader,{scope:options.assets!.authority!.scope,resolve:(id,abort)=>network.run(abort,()=>options.assets!.authority!.resolve(id,abort)),read:(id,abort)=>network.run(abort,()=>options.assets!.authority!.read(id,abort))},id,signal); assertWritable(false);signal.throwIfAborted();if(authorized.ref.asset.url!==options.assets!.sourceURL(id))throw fault('asset_authorization_mismatch'); if(staged.length||adopting)throw fault('reentrant_dispatch');const runtime=editor.getSnapshot().runtime; if(runtime.resources.has(id)||runtime.orphanFiles?.has(id)||importFiles(runtime.imports).has(id)||fileWrites.has(id)||collectionReservations.has(id))throw fault('resource_identity_conflict',[id]); + // MAX_SAFE_INTEGER 先标记尚未绑定提交 revision，防止较早的持久化快照提前释放本次字节。 const previous=adoptedFiles.get(id);adopting=authorized.ref;adoptedFiles.set(id,{bytes:authorized.bytes,rootRevision:Number.MAX_SAFE_INTEGER}); let committed=false; try{const result=editor.transaction(tx=>callback(freeze({id,src:authorized.ref.asset.url,asset:authorized.ref.asset}),tx),{label:'Add existing image'});committed=result.status==='applied'&&result.operationIds.length>0;if(committed)adoptedFiles.set(id,{bytes:authorized.bytes,rootRevision:editor.getSnapshot().rootRevision});return result;} finally{adopting=undefined;if(!committed){if(previous)adoptedFiles.set(id,previous);else adoptedFiles.delete(id);}} })(); + // 项目级任务表允许 settle/close 等待或中止这次领用；完成后删除执行索引，不删除已发布绑定。 adoptionTasks.set(controller,task);void task.finally(()=>adoptionTasks.delete(controller)).catch(()=>{});return task; }, + /** 借出图片显示句柄：先 pin 防清理，再取得共享显示 lease，调用方用完必须 release。 + * 外部 signal、项目关闭和失败路径也会释放；Blob URL 由缓存的最后一个使用者撤销。 */ async displayResource(id: Id, settings: { signal?: AbortSignal } = {}) { assertDisplayIdentity(); if(collectionReservations.has(id))throw fault('resource_collection_in_progress',[id]); displayPins.set(id,(displayPins.get(id)??0)+1); let released=false, lease: DisplayLease | undefined; + /** 幂等释放本次 pin、缓存 lease 和 abort 监听；与同资源的其它显示句柄分别计数。 */ const release=()=>{if(released)return;released=true;settings.signal?.removeEventListener('abort',release);displays.delete(release);const count=displayPins.get(id)??1;if(count<=1)displayPins.delete(id);else displayPins.set(id,count-1);lease?.release();}; displays.add(release); settings.signal?.addEventListener('abort',release,{once:true}); try{ if(settings.signal?.aborted)throw fault('display_aborted'); + // 在 pin 已建立后撤销尚未物理删除的退休标记；已删且无远端副本的文件不能假装仍可本地显示。 const retained=editor.getSnapshot().runtime.resources.get(id); const orphan=editor.getSnapshot().runtime.orphanFiles?.get(id);if(orphan?.retirement?.purgedAt!==undefined)throw fault('local_file_retired',[id]); if(orphan?.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);} @@ -1350,7 +1406,10 @@ export async function openProjectEngine(options: EngineOptions) { } if(released)throw fault('display_aborted'); const binding=editor.getSnapshot().runtime.assetBindings?.get(id); + // 缓存键带授权/摘要/操作身份或退休代次，同 ID 换来源不会误用旧 URL。 const key=JSON.stringify(binding?['shared-asset',id,binding.authority,binding.digest]:retained?['asset',id,retained.operationId,retained.retirement?.purgedAt??null]:orphan?['orphan',id,orphan.digest]:source?['source',id,source.sha256??null]:['unmanaged',id]); + /** 加载回调先找待保存/本地字节，再尝试共享仓库，最后使用服务端显示 URL。 + * 孤立文件必须再次核对内容证据；所有 await 后复查项目身份，迟到 URL 不交给旧用户。 */ lease=displayCache.acquire(displayOwner,key,async signal=>{ let bytes=adoptedFiles.get(id)?.bytes??fileWrites.get(id)??await options.storage.readFile(leader,id); if(!bytes&&binding&&depot)bytes=await depot.read(leader,binding,signal); @@ -1360,14 +1419,18 @@ export async function openProjectEngine(options: EngineOptions) { const resource=editor.getSnapshot().runtime.resources.get(id); if(!bytes&&resource?.retirement?.purgedAt!==undefined&&!resource.stored)throw fault('resource_bytes_unavailable',[id]); if(!bytes&&!options.assets)throw fault('resource_missing',[id]); + // 仅本地 Blob URL 带 revoke 释放器；服务端 URL 不是本次创建的对象，不调用 revoke。 const url=bytes?URL.createObjectURL(bytes):options.assets!.displayURL(id); return Object.freeze({url,...(bytes?{release:()=>URL.revokeObjectURL(url)}:{})}); }); + // 等待缓存的共享读取后仍检查本句柄是否已被取消，失败由 catch 归还本次租约。 const url=await lease.ready; assertDisplayIdentity();if(released||settings.signal?.aborted)throw fault('display_aborted'); return Object.freeze({url,release}); }catch(error){release();assertDisplayIdentity();if(settings.signal?.aborted)throw fault('display_aborted');throw error;} }, + /** 用户主动按原始图片比例调整大小，形成独立编辑事务；与创建时默认尺寸补写区别在历史语义。 + * 要求资源确属目标、代次未变且无手势，事务后核对实际几何，策略约束不满足则整次拒绝。 */ fitImage(target:Target,resourceId:Id):Result{ try{assertWritable();if(sessions.getSnapshot().gesture)throw fault('gesture_in_progress');const root=editor.getSnapshot(),element=root.document.elements.get(target.id),resource=root.runtime.resources.get(resourceId),size=resource?.image?.version===1&&resource.image.status==='ready'?resource.image:resource?.stored; if(!element||element.incarnation!==target.incarnation||element.structure!=='node'||options.registry.resolve(element)!=='known'||!options.registry.components.get(element.componentId)?.resourceFields?.includes('assetId')||element.data.assetId!==resourceId)throw fault('target_changed');if(!size||!validImageDimensions(size))throw fault('image_dimensions_unavailable'); @@ -1375,28 +1438,42 @@ export async function openProjectEngine(options: EngineOptions) { return editor.transaction(tx=>{tx.dispatch({type:'element.geometry',target,patch:{width,height}});const after=tx.getDocument().elements.get(target.id);if(after?.structure!=='node'||Math.abs(after.geometry.width-width)>1e-8||Math.abs(after.geometry.height-height)>1e-8)throw fault('image_fit_constraint');},{label:'Fit image ratio'}); }catch(error){return reject(error);} }, + // 暴露同一根快照与本地保存屏障；persist 不代表网络请求或生成任务全部结束。 getSnapshot: editor.getSnapshot, persist, + /** 退出当前恢复会话，已发布方案仍按会话语义完成保存后再释放暂停。 */ cancelRecovery:async()=>{await recovery?.cancel();}, getDurability, + /** 订阅本地耐久性变化，返回取消订阅；已关闭时不再登记监听。 */ subscribeDurability(listener:()=>void){if(closed)return()=>{};durabilityListeners.add(listener);return()=>{durabilityListeners.delete(listener);};}, + /** 显式重试先解决恢复会话和保存屏障，再重置可重试摘要/解码/尺寸授权及来源验证。 + * 最后恢复调度与生成追尾；复用原逻辑事实和冻结请求，不把失败重试变成重复创建。 */ retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();retryDefaultSizes();const runtime=editor.getSnapshot().runtime,ids=[...runtime.resources.values()].filter(resource=>!resource.digest&&runtime.ledger.operations.get(resource.operationId)?.blockers.some(code=>digestFailures.has(code))).map(resource=>resource.operationId);if(ids.length){mutate(ledger=>{for(const id of ids)ledger.block(id,['waiting_digest']);});prepareDigests();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)){await importLoads.get(job.id);await resetImportReader(job.id);await loadImport(job.id);}await scheduler.resume();await runTracker.retry(); }, + /** 排空当前执行链及其派生保存，根变化时再次检查直到稳定。 + * 不等待未来轮询时刻或服务端生成终结；暂停任务和未到期重试也不等于已经同步成功。 */ async settle() { let revision:number; do { revision=editor.getSnapshot().rootRevision; await Promise.resolve();await Promise.allSettled([...adoptionTasks.values()]);await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle()));while(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));while(digestTasks.size)await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; + // 协调回调可能新增根/保存义务；两次调度-保存屏障覆盖当前执行链产生的后续工作。 await scheduler.settle();await runTracker.settle();await persist();await scheduler.settle();await persist(); } while(revision!==editor.getSnapshot().rootRevision||pending.size||writing||adoptionTasks.size||copyPreparations.size||importLoads.size||verificationTasks.size||digestTasks.size||imageTasks.size||sizingTask); }, + /** 关闭项目所有者：先阻止新准入，中止并排空执行器，再归还显示资源，最后可靠保存与释放领导权。 + * 保存失败时保留 leader/根与字节材料，允许再次 close 重试；视图 detach 不应调用本方法。 */ async close() { if (closed) return; + // 关闭标志先发布，再逐类取消/等待；共享 display/depot 只有本项目自己创建时才负责关闭整个实例。 closing = true;notifyDurability();loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]);for(const controller of adoptionTasks.keys())controller.abort();await Promise.allSettled([...adoptionTasks.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop(); + // runner close 保留已提交前缀；等所有载入/校验/摘要/解码结束后才能释放读取器和底层存储。 await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]);while(verificationTasks.size)await Promise.allSettled([...verificationTasks.values()].map(task=>task.promise));for(const {reader}of proofReaders.values())reader.close();proofReaders.clear();while(digestTasks.size)await Promise.allSettled([...digestTasks.values()]);while(imageTasks.size)await Promise.allSettled([...imageTasks.values()]);while(sizingTask)await sizingTask; recovery?.controller.abort();recovery?.release(); + // 释放本项目所有显示句柄和 cache owner；其它项目仍可继续使用注入的共享缓存。 for (const release of [...displays]) release();displayCache.releaseOwner(displayOwner);if(!options.displays)displayCache.close(); // Failure keeps the leader/recovery material available for an explicit retry of close. await persist(); unsubscribe(); editor.dispose();if(!options.assetDepot)depot?.close();adoptedFiles.clear(); await leader.release(); closed = true;notifyDurability();durabilityListeners.clear(); }, }); + // 工厂打开失败时释放已取得的领导权；成功返回后的关闭责任交给实例 close。 } catch (error) { await leader.release(); throw error; } }