diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 7949c03..61af4c4 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -1075,7 +1075,9 @@ export async function openProjectEngine(options: EngineOptions) { return{job:getImport(id),bytes:encoded,document:undefined,indexed:undefined}; } /** 按 job 装配或复用导入执行器。验证来源、选择协议解析器、准备依赖与资源,再逐批提交。 - * importLoads 是装配任务的单飞表,importRunners 是执行器表,两者都从持久 job 派生。 */ + * importLoads 是装配任务的单飞表,importRunners 是执行器表,两者都从持久 job 派生。 + * 返回 Promise 等装配/校验前置步骤,不等 runner 消费完整源;runner.start 与后台 seal 可继续执行。 + * 需要时重开证明读取器,按持久 source.kind/parserVersion 选实现;不把扩展名或当前 UI 选择当协议。 */ function loadImport(id:Id):Promise{ // 重复调用共享正在装配的任务;已有 runner 时只补 seal/继续运行,不重复解析和提交。 if(importLoads.has(id))return importLoads.get(id)!; @@ -1093,14 +1095,16 @@ export async function openProjectEngine(options: EngineOptions) { const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status)&&(source.integrityVersion!==2||!sourceNeedsSeal(latest)))return; if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed'); let readBytes:ReturnType['port']|undefined; - // 分块协议可在按需验证读取的同时做全文件校验;已有 SHA 则先确认整文件匹配再继续。 + // integrity-v2 先重验已保存 proof,再把 reader.port 注入解析器,后续每次读仍受块证明约束。 + // 无 SHA 时后台 seal 与解析并行;已有 SHA 时先等 verifying,校验失败会把 job 暂停而不继续加入。 if(source.integrityVersion===2){ let context=proofReaders.get(id);if(!context||context.fileId!==source.fileId){context={fileId:source.fileId,bytes,reader:createImportReadProof({bytes,fileId:source.fileId,proof:source.readProof})};proofReaders.set(id,context);} // 恢复先重验已保存分块证明;后续解析读口同样经过 reader,不能直接绕过它读取任意字节。 await context.reader.verify(preparationController.signal);const checking=verifyImportSource(id,context);if(source.sha256)await checking; job=getImport(id);source=job.source!;if(preparationController.signal.aborted||['completed','partial','cancelled','paused'].includes(job.status)||importHolds.size)return;readBytes=context.reader.port; }else{ - // 旧完整性协议在开始解析前计算全文件摘要,不用新的分块语义猜测旧格式。 + // integrity-v1 先全文件 SHA,再装配解析器;缺 SHA 的旧空任务可发布一份新证据并推进 generation。 + // 这条路径没有 v2 的按块渐进读证明,不能仅因文件相同就交换解释语义。 if(source.integrityVersion!==1)throw fault('import_source_integrity_unavailable'); const sha256=await sourceDigest(bytes,preparationController.signal); if(closing||closed)return; @@ -1119,15 +1123,20 @@ export async function openProjectEngine(options: EngineOptions) { const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await (job.parserVersion===5?openStreamingDocumentImportSource:openDocumentImportSource)({bytes,readBytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint}); if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return; if(importHolds.size){deferredImportLoads.add(id);return;} - // beforeCommit 与同步 commit 之间的临时资源授权;不是第二份持久资源表。 + // authorizedImportAssets 只在一次 beforeCommit→commit 之间借用,下一次准备/中止会清空。 + // WeakSet 只避免同一 signal 重复装 abort 监听;授权/Blob 缓存不是已发布 assetBindings 或 durable 写入。 const authorizedImportAssets=new Map(),assetPreparationSignals=new WeakSet(); // 乐观本地模式允许边解析边落图;普通批次最多128项,积压32根或存储失败时施加保存背压。 + // parsed 阶段只有积压>=32根才等 persist;committed 阶段还会因 persistenceFailed 等保存恢复。 + // 32是触发背压的当前队列观察值,不是跨所有生产者的严格内存容量;网络仍独立受可靠 seal 门禁限制。 // SCC 等不可分依赖单元另有限制;128 并不承诺每128项一定让出或一定耗时很短。 const runner=createImportRunner({maxBatch:128,optimisticLocal:true,checkpointSource:source.integrityVersion===2?checkpointImportSource:undefined,awaitDurabilityBudget:stage=>pending.size>=32||stage!=='parsed'&&persistenceFailed?persist():undefined,source:importSource,getJob:()=>getImport(id),createId, // prepare 只生成候选 unit/稳定 ID,实际文档修改由后面的同步事务执行。 publish:publishImport,prepare:(line,job,reserve)=>['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')?prepareDuplicateImport(line,job,reserve,options.registry):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry,depot&&options.assets?.authority?options.assets.sourceURL:undefined),persist,onError:report, - /** 原生文档格式引用的共享资源先逐项授权、校验 URL、读取字节;批次累计不超过32MiB。 - * 异步阶段不改业务根,失败清空准备材料;同步 commit 仍须重新核对 job/目标/依赖。 */ + /** 仅 native-document 对引用资源逐项重新授权、校验 URL 并备好原 Blob;其它源由各自规则处理。 + * 已有本地资源也先走授权,commit 再核对是否同一身份;不能用热缓存绕过当前远端权限。 + * 32MiB 按每次 authorize 返回后的 Blob 长度累加,超限拒绝,不是下载过程中的总内存硬限。 + * 异步阶段不发布业务根,失败清材料;runner 检查 generation 后,同步 commit 再查目标/依赖。 */ async beforeCommit(units,job,signal){ // 其它源格式不走该外部资源领用路径;其引用规则由各自 unit 准备/守卫处理。 authorizedImportAssets.clear();if(job.source?.kind!=='native-document')return{enabled:true}; @@ -1137,14 +1146,16 @@ export async function openProjectEngine(options: EngineOptions) { try{for(const id of ids){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);signal.throwIfAborted();assertImportWritable();if(authorized.ref.asset.url!==options.assets.sourceURL(id))throw fault('asset_authorization_mismatch');total+=authorized.bytes.size;if(total>32*1024*1024)throw fault('import_asset_capacity');authorizedImportAssets.set(id,authorized);}return{enabled:true};}catch(error){authorizedImportAssets.clear();return{enabled:false,code:(error as {code?:string}).code??'import_asset_authorization_failed',targets:[]};} }, /** 将一批元素/canvas、授权资源绑定及 next job 检查点放入一次 editor 发布。 - * 失败回滚的只是本批临时领用材料;此前成功加入并显示的批次保持不变。 */ + * 事务拒绝不发布本批文档/job,finally 撤回临时领用字节与授权上下文;此前成功批次保持不变。 + * 允许的 unit 只有 before=null 的 element/canvas 创建;任意字段修改或删除都拒绝。 */ commit(units,next,expected){ // 消费一次 beforeCommit 材料,防止后续批次或重入误用上次授权。 const preparedAssets=new Map(authorizedImportAssets);authorizedImportAssets.clear(); assertImportWritable();if(importing)throw fault('reentrant_dispatch');const current=getImport(id);if(current.generation!==expected)throw fault('stale_import_job'); // 等待资源授权期间目标或外部连线端点可能变化,提交前重新检查入口代次和依赖。 try{assertNativeImportTarget(current,editor.getSnapshot().document);for(const unit of units){assertNativeImportDependencies(current,editor.getSnapshot().document,unit.dependencies);if(['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??''))assertDuplicateExternal({cursor:0,nextCursor:1,key:unit.key,value:unit.guard},current,editor.getSnapshot().document,options.registry);}}catch(error){return reject(error);} - // 临时登记待保存的领用字节,同时保存旧值,保证事务拒绝后可准确恢复原暂存集合。 + // 已有上传资源须匹配 stored/digest;新领用先占 adoptedFiles 并记旧值,防和孤儿/源/待删文件撞 ID。 + // 临时 rootRevision=MAX_SAFE_INTEGER 让旧根保存不提前采纳;事务成功后再绑定真实发布 revision。 const previousFiles=new Map();let accepted=false; try{for(const [assetId,authorized]of preparedAssets){const runtime=editor.getSnapshot().runtime,own=runtime.resources.get(assetId);if(own){if(!own.stored||own.digest!==authorized.ref.digest||!equal(own.stored,authorized.ref.asset))throw fault('shared_asset_identity_conflict',[assetId]);continue;}if(runtime.orphanFiles?.has(assetId)||importFiles(runtime.imports).has(assetId)||fileWrites.has(assetId)||collectionReservations.has(assetId))throw fault('resource_identity_conflict',[assetId]);previousFiles.set(assetId,adoptedFiles.get(assetId));adoptedFiles.set(assetId,{bytes:authorized.bytes,rootRevision:Number.MAX_SAFE_INTEGER});}importAdoptions=[...preparedAssets].filter(([assetId])=>!editor.getSnapshot().runtime.resources.has(assetId)).map(([,value])=>value.ref);}catch(error){for(const [assetId,previous]of previousFiles)if(previous)adoptedFiles.set(assetId,previous);else adoptedFiles.delete(assetId);authorizedImportAssets.clear();return reject(error);} // reducer 在此同步窗口读 importing/importAdoptions,使 job 游标和新增元素同根更新。 @@ -1153,16 +1164,21 @@ export async function openProjectEngine(options: EngineOptions) { try{const result=editor.transaction(tx=>{for(const unit of units)for(const change of unit.changes){if(change.kind==='element'&&change.before===null&&change.after)tx.dispatch({type:'element.create',element:change.after});else if(change.kind==='canvas'&&change.before===null&&change.after)tx.dispatch({type:'canvas.create',canvas:change.after});else throw fault('invalid_native_import_unit');}},{label:['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??'')?'Duplicate selection':'Import canvas content',groupId:`import:${id}`});accepted=result.status==='applied';if(accepted)for(const assetId of previousFiles.keys())adoptedFiles.set(assetId,{bytes:preparedAssets.get(assetId)!.bytes,rootRevision:editor.getSnapshot().rootRevision});return result;}finally{importing=undefined;importAdoptions=[];if(!accepted)for(const [assetId,previous]of previousFiles)if(previous)adoptedFiles.set(assetId,previous);else adoptedFiles.delete(assetId);authorizedImportAssets.clear();} }, }); - // 装配完成后继承仍有效的全局 holds;业务 status=running 才启动加入画布的工作。 + // 注册 runner 后继承所有仍有效的全局 hold,避免装配期间新增 hold 被漏掉。 + // 只有根中 status=running 才启动消费;void start 不延长 loadImport Promise 到导入结束。 importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold()); if(getImport(id).status==='running')void runner.start(); // 中止不改业务终态;真实读取损坏留下完整性错误,其它装配失败转 paused。 - // finally 清除执行索引,仅在仍有延迟恢复义务且 job 可运行时重新装配,避免覆盖后来换源的状态。 + // 失败收尾按最终 expectedFingerprint 判断归属:复制物化合法换源后也只能修改对应来源。 + // v2 字节错误留 integrityError,其它非终态装配错误转 paused;失败不能清空已加入内容或重置稳定映射。 + // finally 清执行索引;仅有 deferred 恢复义务、无全局 hold、且仍 running/preparing 时安排重装,不无限重试错误。 })().catch(error=>{if(preparationController.signal.aborted)return;if(!closing&&!closed){try{const job=getImport(id);if(job.source?.integrityVersion===2&&!importRunners.has(id)&&!verificationTasks.has(id)){proofReaders.get(id)?.reader.close();proofReaders.delete(id);}if(job.sourceFingerprint!==expectedFingerprint)return;if(job.source?.integrityVersion===2&&(isImportByteFailure(error)||(error as {code?:string}).code==='import_source_unavailable')){failImportVerification(job,error);void persist().catch(report);return;}if(['completed','partial','cancelled','paused'].includes(job.status))return;publishImport(importCheckpoint({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{preparationControllers.delete(id);importLoads.delete(id);if(!['running','preparing'].includes(getImport(id).status))deferredImportLoads.delete(id);if(!closing&&!closed&&!importHolds.size&&deferredImportLoads.delete(id)&&!importRunners.has(id)&&['running','preparing'].includes(getImport(id).status))queueMicrotask(()=>{if(!closing&&!closed)void loadImport(id);});}); importLoads.set(id,task);return task; } /** 接收本地 Blob 并同步登记一个可恢复作业(源文件上限64MiB),随后后台读取解析。 - * 文件 ID、源证明和目标入口代次从一开始就写进根,页面可立即观察 running,不等待全文件解析。 */ + * 文件 ID、空读证明和目标入口代次从一开始写进根,页面可立即观察 running,不等全文件解析。 + * fileWrites 先暂存源 Blob,登记根被拒绝则撤回;接受仅表示本地任务存在,磁盘保存与 SHA 封存随后完成。 + * targetEntry 固定子分区入口代次,parserVersion 按显式 format 固定5/3/1,创建时不猜内容格式。 */ function startNativeImport(bytes:Blob,targetCanvasId:Id,format:'ndjson'|'json-array'|'document'):Id{ if(!['ndjson','json-array','document'].includes(format))throw fault('import_interpreter_unavailable'); assertImportWritable();const loadingAdmission=loadingImportAdmission(editor.getSnapshot(),targetCanvasId);const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source'); @@ -1190,7 +1206,9 @@ export async function openProjectEngine(options: EngineOptions) { startDocument:(bytes:Blob,targetCanvasId:Id)=>startNativeImport(bytes,targetCanvasId,'document'), startJsonArray:(bytes:Blob,targetCanvasId:Id)=>startNativeImport(bytes,targetCanvasId,'json-array'), /** 捕获当前不可变文档和独立选择/偏移,异步编码索引快照,再登记 preparing 复制任务。 - * 后续原画布编辑不会改变复制来源;完成编码不代表已创建全部副本。 */ + * 先编码 captured.document 的完整索引快照,再用 selection 规划复制闭包;不是只序列化选中记录。 + * Promise 返回的是已登记 preparing 的 job ID,不等待副本创建;快照编码后目标入口仍要在后续提交重查。 + * 外部 signal 只覆盖本次编码/登记准备句柄,返回后由 imports.pause/cancel 控制作业生命周期。 */ async duplicate(selection:readonly Target[],settings:{readonly worldOffset?:Point;readonly signal?:AbortSignal;readonly viewCanvasId?:Id}={}):Promise{ assertImportWritable();if(collectionReservations.size)throw fault('resource_collection_in_progress',[...collectionReservations]); const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); @@ -1221,16 +1239,21 @@ export async function openProjectEngine(options: EngineOptions) { // 从编码开始到作业登记结束持有准备任务/pin;所有结束路径清除任务及外部 abort 监听。 copyPreparations.set(controller,task);void task.finally(()=>{copyPreparations.delete(controller);signal?.removeEventListener('abort',abort);}).catch(()=>{});return task; }, - /** 有 runner 就停在其批次边界;未装配时直接推进 job generation 并保存 paused。 - * 分块来源的验证仍可能继续补齐已提交前缀的证据,不能把暂停当作删除来源。 */ + /** 有 runner 则调用其 pause:发布 paused/保存并请求 abort,不额外等待 active 完全退出。 + * 未装配则推进 generation;v1 请求中止准备,v2 可继续补齐来源证明,但不继续加入内容。 + * 暂停不删除源文件、已加入内容或稳定映射;需要排空时仍须使用相关 settle/关闭流程。 */ async pause(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(['running','preparing'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();publishImport(importCheckpoint({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}}, - /** 终态不重启;paused 先等旧装配/runner 结束,必要时重建证明读取器,再保存并恢复调度。 - * 复制快照回 preparing,其它源回 running,避免跳过尚未完成的复制规划。 */ + /** completed/partial/cancelled 不重启;已 running/preparing 仅确保 loadImport 装配。 + * paused 先等旧装配/runner 收尾,v2 重建 reader,再保存现有根并恢复 scheduler,随后重新查状态。 + * 复制快照回 preparing,其它源回 running;清 job.error 但不擅自清 source.integrityError,重验成功才解封。 */ async resume(id:Id){assertImportWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(['running','preparing'].includes(job.status)){await loadImport(id);return;}await importLoads.get(id);await importRunners.get(id)?.settle();if(getImport(id).source?.integrityVersion===2)await resetImportReader(id);await persist();await scheduler.resume();assertImportWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:current.source?.kind==='duplicate-snapshot'?'preparing' as const:'running' as const};delete next.error;publishImport(importCheckpoint(next),current.generation);await loadImport(id);}, - /** 取消后续加入,保留已有内容;尚欠 seal 时仍可继续校验,证据完整才中止额外验证。 */ + /** 取消后续加入并保存取消事实,保留成功内容;无 runner 的 preparing 任务也可在此取消。 + * v2 若已加入内容但欠 seal,仍保留验证任务;无成功源项的取消任务不欠 seal,可请求中止额外验证。 + * 不等同于撤销已同步操作、删除源文件或调用后台生成取消接口。 */ async cancel(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','preparing','paused'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(importCheckpoint(next),job.generation);await persist();}}if(!sourceNeedsSeal(getImport(id)))verificationTasks.get(id)?.controller.abort();}, }); - // 全部协调回调已接线后启动调度,恢复既有生成轮询和资源任务。 + // 协调回调与根订阅都装配完才启动执行器;恢复的 run IDs 显式首次 notify,不能只等新发布发现。 + // 工厂将控制面交回宿主时后台任务可已在运行,并非等所有网络/导入完成才返回。 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);