From 56b83e2009b68fab12cca0f438fbc785c429fcae Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 14:50:08 +0800 Subject: [PATCH] docs(runtime): explain engine assembly and durable prefix writes --- src/lib/canvas-runtime/engine.ts | 80 +++++++++++++++++++++++--------- 1 file changed, 57 insertions(+), 23 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 3a8d155..6e8b992 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -4,6 +4,8 @@ * engine 把持久化、网络调度、资源、导入、生成和冲突恢复连接到这个边界。 * 这里的 Promise/Map/Set 大多是当前进程的任务、租约或索引，不替代可恢复根中的业务事实。 * 打开项目先取得存储领导权并解码恢复镜像；视图挂载/卸载与项目关闭是不同的生命周期。 + * 阅读顺序可按“打开/初始根→editor hooks→持久队列→资源清理→请求协调→任务执行→公开能力/关闭”。 + * 临时上下文只在同步提交期间授予对应 reducer 权限，异步准备不得跨 await 长期占用这些上下文。 */ import {dataArray,dataRecord} from '../canvas-core/data-containers.ts'; import {detachJsonData} from '../canvas-core/json-data.ts'; @@ -58,7 +60,9 @@ const digestFailures=new Set(['digest_capacity','digest_worker_failed','digest_w /** 打开项目的依赖注入参数。scope/document 必须属于同一项目；registry 解释模型和恢复数据。 * storage 是本地可靠保存，persistence 是文档网络协议；assets/runs 为可选外部业务服务。 * digest/image/display/depot 可共享注入；clock 和 createId 允许宿主或测试控制时间与身份。 - * loading 表示尚未读完的模型，不能把已有前缀当成完整项目。 */ + * loading 表示尚未读完的模型，不能把已有前缀当成完整项目。 + * options/端口大多借用而非深复制，宿主须保持配置与 scope 身份稳定；动态认证撤销走 isCurrentIdentity。 + * 注入共享池的拥有者负责其全局关闭，本项目只释放自己持有的任务、显示租约和存储 leader。 */ export type EngineOptions = { loading?:LoadingState; scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap; @@ -84,7 +88,9 @@ export type RecoverySession = { }; /** 打开一个项目所有者实例。先独占本地写入，再选择恢复状态或初始服务端状态， - * 接线 editor 与异步执行器；此工厂是异步入口，普通编辑提交仍在 editor 内同步发布。 */ + * 接线 editor 与异步执行器；此工厂是异步入口，普通编辑提交仍在 editor 内同步发布。 + * scope/document 先核项目 ID，loading/versions 在首个 await 前捕获；document/registry/其它 options 仍借用。 + * 本地 recovered Root 优先于外部初始模型，但传入模型的每张 canvas 仍须带有效 openingVersions。 */ export async function openProjectEngine(options: EngineOptions) { if (options.scope.projectId !== options.document.projectId) throw fault('project_mismatch'); // 在第一次 await 前捕获加载状态和版本，防止调用方随后修改原容器改变这次打开的基线。 @@ -100,7 +106,8 @@ export async function openProjectEngine(options: EngineOptions) { // 渐进读取的在途任务、准入暂停句柄及完成 revision：完成读取后仍要等对应根可靠保存。 let loadingRead:Promise|undefined,loadingReadController:AbortController|undefined; let acceptingLoading=false,loadingRelease:(()=>void)|undefined,loadingCompletionRevision:number|undefined; - // closing 阻止新工作，closed 表示生命周期结束；durableRootRevision 仅由存储回执推进。 + // closing 阻止新工作，closed 表示生命周期结束；durableRootRevision 从恢复镜像起步，之后只认存储回执。 + // 新项目设为-1，保证 revision=0 的初始根也必须持久化，不能误当成已保存。 let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1; // 项目级摘要中止源，与单次任务 signal 合并；关闭时可撤销仍在处理的字节计算。 const digestLifecycle=new AbortController(); @@ -110,13 +117,15 @@ export async function openProjectEngine(options: EngineOptions) { const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } }; // 持久化订阅与文档订阅分开；缓存同值对象，便于界面按快照身份避免无意义更新。 const durabilityListeners=new Set<()=>void>();let durabilityError:string|undefined,durabilityCache:Readonly<{rootRevision:number;durableRootRevision:number;closing:boolean;closed:boolean;error?:string}>|undefined; - /** 返回当前根/已保存根、关闭阶段和存储错误的稳定只读快照。 */ + /** 返回当前根/已保存根、关闭阶段和存储错误的稳定只读快照；同字段值复用对象供订阅比较。 + * 这是本地持久化诊断，不是网络 ACK/服务同步进度；editor 初始化后才允许调用此闭包。 */ function getDurability(){const rootRevision=editor.getSnapshot().rootRevision;if(!durabilityCache||durabilityCache.rootRevision!==rootRevision||durabilityCache.durableRootRevision!==durableRootRevision||durabilityCache.closing!==closing||durabilityCache.closed!==closed||durabilityCache.error!==durabilityError)durabilityCache=Object.freeze({rootRevision,durableRootRevision,closing,closed,...(durabilityError?{error:durabilityError}:{})});return durabilityCache;} /** 拷贝监听集合后逐个通知，订阅过程中增删监听不影响本轮，也不传播监听器异常。 */ function notifyDurability(){for(const listener of [...durabilityListeners])try{listener();}catch(error){report(error);}} // 所有新任务/请求需要独立身份；宿主可注入生成器，默认使用浏览器 UUID。 const createId = options.createId ?? (() => crypto.randomUUID()); - // 网络门限制并发为 2，并在准入时核对存储领导权和认证代次。 + // 网络门限制同时在途为2，在准入时查 leader.active 与当前认证；这里不比较远端数据库 fence。 + // 本地可靠屏障/任务依赖由 scheduler 另管，network 不自己判断请求是否已经安全落盘。 const network = createNetworkGate(2,()=>{if(!leader.active)throw fault('storage_leader_lost');if(options.isCurrentIdentity&&!options.isCurrentIdentity())throw fault('identity_changed');}); // 有共享存储端口时可装配资源仓库；跨项目共享仅限同账户、同工作区。 const depot=options.assetDepot??(options.storage.assetDepot?createAssetDepot({scope:options.scope,port:options.storage.assetDepot,current:()=>!closing&&!closed&&(!options.isCurrentIdentity||options.isCurrentIdentity()),digest:sourceDigest}):undefined); @@ -134,7 +143,8 @@ export async function openProjectEngine(options: EngineOptions) { const displayOwner = Object.freeze({}), displayCache = options.displays ?? createDisplayCache({ onError: options.onError }); /** 即使缓存已有显示结果，也先检查项目/身份仍有效，不能跨账户复用旧视图授权。 */ const assertDisplayIdentity=()=>{if(closing||closed)throw fault('engine_closing');if(options.isCurrentIdentity&&!options.isCurrentIdentity())throw fault('identity_changed');}; - // 显示租约计数阻止清理；reservation 阻止领用即将删除的文件；durablyPurged 记录已删字节。 + // displayPins 计数显示/文件借用租约；reservation 在规划 purge 后阻止新领用，直到可靠回执释放。 + // durablyPurged 是当前 leader 已可靠记录的 purge 身份集合，用来避免重复安排物理删除，并非 Blob 缓存。 const displayPins=new Map(),collectionReservations=new Set(),durablyPurged=new Set([...(recovered?.runtime.resources.values()??[])].filter(resource=>resource.retirement?.purgedAt!==undefined).map(resource=>resource.id)); /** 清理结果同时报告检查根、当前根和可靠保存根，区分“规划删除”和“字节已删除”。 */ type CollectionReport={readonly markedIds:readonly Id[];readonly purgedIds:readonly Id[];readonly complete:boolean;readonly checkedRootRevision:number;readonly rootRevision:number;readonly durableRootRevision:number}; @@ -149,9 +159,11 @@ export async function openProjectEngine(options: EngineOptions) { // 解码/尺寸补写/摘要等在途索引只用于生命周期等待与去重；恢复事实仍在 runtime 中。 const imageTasks=new Map>(); let sizingTask: Promise | undefined; - // 待写 Blob 与根快照分开保存，writePrefix 只把当前快照登记的资源/导入源文件纳入原子提交。 + // fileWrites 暂存未可靠落盘的 Blob；digestTasks 只去重并等待摘要工作，二者都不是持久任务表。 + // writePrefix 按某一捕获根的资源/源归属筛选文件，不能把未来编辑的字节提前配给旧根。 const fileWrites = new Map(), digestTasks = new Map>(); - // 文件暂存及默认尺寸标记仅覆盖同步编辑事务；事务 reducer 将其写入可恢复资源记录。 + // staged 与 stagedDefaultSizes 仅是同步编辑事务的授权上下文；reduceRuntime 将其转成根中的资源事实。 + // 分配/上传/解码不能在 reducer 内启动，失败候选也不能留下第二份已接纳任务。 let staged: readonly ResourceRecord[] = []; const stagedDefaultSizes = new Set(); // 将用户提交标为 run.apply 的短暂上下文；不能让一次应用操作泄漏到后续普通编辑。 @@ -168,7 +180,8 @@ export async function openProjectEngine(options: EngineOptions) { const importRunners=new Map>(),importLoads=new Map>(),importHolds=new Mapvoid>>(); // 恢复会话存在期间阻止一般写入；会话负责持有/释放调度暂停及中止控制。 let recovery: { controller: AbortController; release(): void; cancel():Promise } | undefined; - /** 复制准备期间保守保护未清除资源，解码期间保护正在读取的图片；供每次保留扫描派生。 */ + /** 每次派生临时保留清单：任一复制准备仍活跃就保守 pin 全部未 purge 的本地资源，解码按任务 ID pin。 + * Map 是本次扫描的输入，不发布第二份 resourcePins；任务结束后下次扫描自然不再包含该临时 owner。 */ const admissionPins=(snapshot:EngineSnapshot)=>{const pins=new Map();if(copyPreparations.size)pins.set('copy-admission',[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt===undefined).map(resource=>resource.id));if(imageTasks.size)pins.set('image-decode',[...imageTasks.keys()]);return pins;}; /** 通用写入护栏。complete=false 仅放宽“模型必须完整”，不绕过身份、租约或恢复会话限制。 */ const assertWritable = (complete=true) => { if(!leader.active)throw fault('storage_leader_lost');if(options.isCurrentIdentity&&!options.isCurrentIdentity())throw fault('identity_changed');if(complete&&editor.getSnapshot().runtime.loading)throw fault('model_loading');if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); }; @@ -184,7 +197,8 @@ export async function openProjectEngine(options: EngineOptions) { // 已领用资源绑定必须能被当前服务 authority 和 URL 解释；换服务不能静默复用旧授权。 for(const [id,binding]of initial.assetBindings??[])if(!depot||!options.assets?.authority||binding.authority!==options.assets.authority.scope||binding.asset.url!==options.assets.sourceURL(id))throw fault('asset_binding_interpreter_unavailable',[id]); if(initial.loading?.version===6&&options.assets)for(const resource of initial.resources.values())if(resource.loadingAdmission?.sourceURL!==options.assets.sourceURL(resource.id))throw fault('asset_source_changed',[resource.id]); - // editor 是唯一状态根的发布者，下面的 hooks 在发布前整合 runtime，而非先改文档后补队列。 + // editor 是唯一状态根的发布者，hooks 在候选发布前整合/检查 runtime，而非先改文档后补队列。 + // prepareRuntime 是全部发布的保留/所有权边界，reduceRuntime 只处理带 commit 的编辑；二者不做 I/O。 const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,queryBudget:options.queryBudget,queryCoverage:loadingQueryCoverage, /** 命令入口先检查基础生命周期，再检查渐进加载允许范围，最后应用宿主业务策略。 */ policy(command, document) { if(!leader.active)return{enabled:false,code:'storage_leader_lost',targets:[]};if(options.isCurrentIdentity&&!options.isCurrentIdentity())return{enabled:false,code:'identity_changed',targets:[]};if (closing || closed || recovery) return { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] }; if(editor.getSnapshot().runtime.loading){const allowed=loadingCommand(command,document,options.registry,editor.getSnapshot().runtime.loading,editor.getSnapshot().runtime.loadingRead?.document??editor.getSnapshot().runtime.acknowledged,editor.getSnapshot().runtime.loadingRead,assetURLs());if(!allowed.enabled)return allowed;}return options.policy?.(command, document) ?? { enabled: true }; }, @@ -207,6 +221,8 @@ export async function openProjectEngine(options: EngineOptions) { for(const id of next.runtime.assetBindings?.keys()??[])if(next.runtime.resources.has(id)||next.runtime.orphanFiles?.has(id)||importFiles(next.runtime.imports).has(id))throw fault('resource_identity_conflict',[id]); validateLoadingRuntime(next.runtime,options.registry); // 先计算尺寸权限，再依文档/历史/账本/显示引用派生资源保留，最后处理源文件和孤立字节。 + // 后一步使用前一步返回的候选：先撤销失效尺寸权限，再算资源 pin，再算导入源和孤儿退休状态。 + // previous/next 只是候选快照，此处抛错交给 editor 拒绝整次发布，不在这里自行修改旧根。 const sizing=prepareDefaultImageSizing(previous,next); const runtime=prepareResourceRetention({...next,runtime:sizing},options.registry,displayPins,collectionReservations,admissionPins(next)); // 这些引用事实变化会影响“是否仍无人使用”；保留准备函数据此重新检查退休标记。 @@ -216,12 +232,13 @@ export async function openProjectEngine(options: EngineOptions) { }, /** 只把当前应用输出的用户意图转换为 run.apply，保留普通 editor 的命令生成路径。 */ prepareIntent: intent => applying && intent.origin==='user' ? {...intent,operationType:'run.apply',action:applying} : intent, - /** 在现有账本最大序号后分配；预留 staged 上传操作的位置，让上传先于引用它的文档操作。 */ + /** 取 fallback 与账本最大 sequence+1 的较大值，再为 staged 上传留连续序号。 + * reduceRuntime 从首个文档操作序号倒推上传起点，保证资源任务排在依赖它的文档意图之前。 */ nextOperationSequence(runtime, fallback) { let next = fallback; for (const record of runtime.ledger.operations.values()) next = Math.max(next, record.intent.sequence + 1); return next + staged.length; }, /** 把一次 editor 提交转换为账本、资源及导入作业的同根更新；不在这里发送请求。 */ reduceRuntime(previous, next, commit) { if(previous.loading)assertLoadingChanges(commit.changes,previous.loading.version>=4?editor.getSnapshot().document:previous.acknowledged,options.registry,previous.loading,previous.loadingRead,assetURLs()); - // 无逻辑操作的提交不创建上传依赖；保留原 runtime 对象。 + // 没有逻辑 operationIds 就保持原 runtime；资源暂存本身不能脱离文档提交变成上传任务。 if (!commit.operationIds.length) return previous; const ledger = createLedger(previous.ledger), resources = new Map(previous.resources); const firstSequence = next.operations.get(commit.operationIds[0])!.sequence - staged.length; @@ -247,7 +264,8 @@ export async function openProjectEngine(options: EngineOptions) { if (grants.some((grant, index) => grant !== resource.sizing!.grants[index])) resources.set(resourceId, freeze({ ...resource, sizing: { version: 1 as const, grants } })); } } - // 所有未决 run.start，以及与本操作足迹冲突的文档/应用操作，都成为先行依赖。 + // 未决 run.start 全部作为先行依赖；document.change/run.apply 只在对象足迹冲突时加入。 + // Set 合并原依赖、先前任务和下方资源上传依赖，保留因果关系而不复制第二份调度状态。 const dependencies = new Set(intent.dependencies); for (const record of ledger.getSnapshot().operations.values()) if (record.phase !== 'resolved' && (record.intent.operationType === 'run.start' || ['document.change','run.apply'].includes(record.intent.operationType) && conflictsWith(operationFootprint(record.intent, next.document), footprint))) dependencies.add(record.intent.id); // 引用尚未上传完成的资源时，文档同步必须等待该资源的上传操作。 @@ -255,7 +273,8 @@ export async function openProjectEngine(options: EngineOptions) { const resource = resources.get(id); if (resource && !resource.stored) dependencies.add(resource.operationId); } - // 合并器需要起始存在性，才能区分本地 create→delete 可消除与已存在对象必须发送删除。 + // 同一对象首个变化决定 basePresence：element.before 空为 absent，field 默认 exists，后续不覆盖。 + // 它不是“最后节点对象”；给合并/抵消保留起始事实，是否已发出仍由 ledger/request 证据约束。 const basePresence: Record = Object.create(null); for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] ??= change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists'; ledger.add(freeze({ ...intent, dependencies: [...dependencies] }), basePresence); @@ -278,7 +297,8 @@ export async function openProjectEngine(options: EngineOptions) { return Object.freeze({ ...previous, imports:readMap(imports), ...(assetBindings.size?{assetBindings:readMap(assetBindings)}:{}), resources: readMap(resources), ledger: ledger.getSnapshot() }); }, }); - // 待持久化快照按根 revision 递增插入 Map，形成写前缀队列；不会用最新根覆盖未保存旧根。 + // pending 是 revision→不可变完整根的插入有序 Map，订阅按发布顺序追加；不是只保留最新根的防抖槽。 + // writePrefix 总取首项，成功后才移除；每项共享不可变子树，但积压仍会占内存，导入背压在后面限制增长。 const pending = new Map(); // 新项目的初始根也要保存；恢复镜像已经在磁盘上，不重复当成新提交。 if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot()); @@ -288,11 +308,14 @@ export async function openProjectEngine(options: EngineOptions) { // 记录由可靠回执确认的 ledger generation；即使派生标记发布失败，也保留这项推进义务。 let confirmedLedgerGeneration=editor.getSnapshot().runtime.ledger.durableGeneration; /** 保存不超过 throughRevision 的已捕获前缀。每份根镜像与对应字节写入/删除同事务提交， - * 存储回执核对通过后才移除 pending、释放字节和推进可靠游标。已有写入时共享该 Promise。 */ + * 存储回执核对通过后才移除 pending、释放字节和推进可靠游标。已有写入时共享该 Promise。 + * 后来的调用不会扩大当前 throughRevision；persistThrough 在旧写入完成后继续补写所需边界。 + * pending Map 的插入序由根订阅维护，不在此排序/补洞；存储 adapter 再校验连续 revision。 */ function writePrefix(throughRevision:number):Promise { if (writing) return writing; const task = Promise.resolve().then(async () => { - /** 把已可靠写入的 generation 发布回 ledger；此更新可能产生新根，不能冒充它也已保存。 */ + /** 一次前缀结束后把已确认 generation 反映到 ledger；mutate 可能产生新的待保存根。 + * confirmedLedgerGeneration 在回执后保留推进义务，即使本次派生发布失败，下次也可补办。 */ const publishDurability=()=>{if(editor.getSnapshot().runtime.ledger.durableGenerationledger.markDurable(confirmedLedgerGeneration));}; try { while (pending.size) { @@ -304,7 +327,8 @@ export async function openProjectEngine(options: EngineOptions) { const deleteFiles=[...[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id),...[...sourceFiles].filter(([id,entry])=>entry.file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id),...[...(snapshot.runtime.orphanFiles??[])].filter(([id,file])=>file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id)]; // 共享仓库写入和项目 pin 清单也由同一快照准备，避免项目镜像与资源引用各自提交。 const depot=options.assets?.authority&&options.storage.assetDepot?await prepareDepotCut({snapshot,registry:options.registry,storage:options.storage,leader,authority:options.assets.authority.scope,files:fileWrites,adoptedFiles,displays:displayPins,temporary:admissionPins(snapshot)}):undefined; - // 编码可能让出线程；snapshot 保持不可变，不能改用等待后出现的新 working 根。 + // depot 准备/编码都可能 await，始终使用本轮 snapshot；不能等待后改取 live 根再配旧 revision/files。 + // storage.commit 必须把 image/本批 Blob/删除/depot 同事务处理；这里只在匹配回执后承认可靠。 const receipt = await options.storage.commit(leader, { image: await encodeRecoveryAsync(snapshot), files,deleteFiles,...(depot?{depot}:{}) }); if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt'); // 回执之后才承认 generation/源 seal 可靠；后续网络准入依赖这些事实。 @@ -319,7 +343,7 @@ export async function openProjectEngine(options: EngineOptions) { // 部分前缀成功、后续失败时仍发布成功证据，不能把已可靠写入的操作退回“从未保存”。 publishDurability();throw error; } - // One derived publication per captured prefix, not one per receipt. + // 一次前缀尽量只发布一次派生 durableGeneration，避免每个回执都再制造一个待保存根。 publishDurability(); persistenceFailed = false;durabilityError=undefined;notifyDurability(); }); @@ -327,18 +351,23 @@ export async function openProjectEngine(options: EngineOptions) { writing = task.catch(error => { persistenceFailed = true;const failure=error&&typeof error==='object'?error as {code?:unknown;message?:unknown}:{};durabilityError=typeof failure.code==='string'&&failure.code?failure.code:typeof failure.message==='string'&&failure.message?failure.message:'persistence_failed';notifyDurability(); throw error; }).finally(() => { writing = undefined;if(!persistenceFailed){if(loadingCompletionRevision!==undefined&&durableRootRevision>=loadingCompletionRevision&&!editor.getSnapshot().runtime.loading){const release=loadingRelease;loadingRelease=undefined;loadingCompletionRevision=undefined;release?.();}if(!closing&&!closed)scheduler.refreshAdmission();} }); return writing; } - /** 仅保证指定 revision 已可靠保存。若共享的在途写入目标更早，完成后继续写到所需边界。 */ + /** 捕获 requiredRevision（缺省为调用时根 revision），只保证这一边界已可靠保存。 + * 共用在途写入可能只到更早根，因此 .then 再查水位并补写；不是保证所有未来编辑都已保存。 + * 写失败向调用者传播，不在这里无限自动重试；必须由后续显式/自动保存重新触发。 */ function persistThrough(requiredRevision=editor.getSnapshot().rootRevision):Promise { if(durableRootRevision>=requiredRevision)return Promise.resolve(); return writePrefix(requiredRevision).then(()=>durableRootRevision { do {await writePrefix(editor.getSnapshot().rootRevision);} while(pending.size||writing||confirmedLedgerGeneration>editor.getSnapshot().runtime.ledger.durableGeneration); } /** 本地上传资源的两阶段清理：先持久化无人引用时间，再于保留期后重新扫描并原子删字节。 - * 引用覆盖不完整就停止，限制每批文件数，返回标记与真正保存成功的删除结果。 */ + * 引用覆盖不完整就停止，限制每阶段候选数；先保存现有根，再标记，再复取根/pin 判定删除。 + * markedIds/purgedIds 只随成功返回报告；中途失败抛错，已保存标记保留供下一轮继续。 + * maxFiles 分别限制标记/删除，不是全表扫描或总工作量上限；各资源仍需遍历后筛选。 */ function collectResources(settings:{retentionMs:number;maxFiles?:number}) { assertWritable(); if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable'); @@ -347,9 +376,11 @@ export async function openProjectEngine(options: EngineOptions) { if(collection)return collection; const task=(async()=>{ await persist();assertWritable(); + // 一轮使用一个 now；保存后重查引用，但不把等待时间追加进本轮保留期判断。 const now=runClock.now();if(!Number.isSafeInteger(now)||now<0)throw fault('invalid_resource_collection_clock'); let root=editor.getSnapshot(),pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root)); const markedIds:Id[]=[],purgedIds:Id[]=[]; + // complete 只表示引用扫描可解释完整，不表示仓库中所有资源都已 purge。 const report=()=>Object.freeze({markedIds:Object.freeze(markedIds),purgedIds:Object.freeze(purgedIds),complete:pins.complete,checkedRootRevision:root.rootRevision,rootRevision:editor.getSnapshot().rootRevision,durableRootRevision}); if(!pins.complete)return report(); // 待写或摘要处理中资源不参与清理；必须先证明没有文档、历史、任务或显示租约引用。 @@ -362,6 +393,7 @@ export async function openProjectEngine(options: EngineOptions) { root=editor.getSnapshot();pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root)); if(!pins.complete)return report(); // 标记后的 await 期间可能出现新引用，所以重新取根和 pins，再判定到期资源。 + // retentionMs=0 时本轮新标记也可能直接进入 purge，但前面的标记根已先可靠保存。 const eligible=[...root.runtime.resources.values()].filter(resource=>resource.retirement&&resource.retirement.purgedAt===undefined&&now-resource.retirement.unreferencedAt>=retentionMs&&!pins.pins.has(resource.id)&&!fileWrites.has(resource.id)&&!digestTasks.has(resource.id)).slice(0,maxFiles); if(eligible.length){ // 先保留删除席位阻止新领用；若根 revision 冲突则撤销保留，不报告已删除。 @@ -376,7 +408,9 @@ export async function openProjectEngine(options: EngineOptions) { collection=task.finally(()=>{collection=undefined;});return collection; } /** 导入源文件的同类两阶段清理；当前源只允许在任务终结且没有载入器时成为候选。 - * 旧分片/历史源也要经过引用覆盖和保留期，不按任务结束就立即丢弃全部原始字节。 */ + * 旧分片/历史源也要经过引用覆盖和保留期，不按任务结束就立即丢弃全部原始字节。 + * single-flight 只覆盖本类 sourceCollection；运行中重复调用共用首次 settings，不另开一轮不同参数扫描。 + * 与资源回收一样 maxFiles 分阶段限候选，一轮 now 固定，可靠回执后才报告 purgedIds。 */ function collectImportSources(settings:{retentionMs:number;maxFiles?:number}):Promise { assertWritable();if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable'); const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64;