docs(runtime): explain engine assembly and durable prefix writes
This commit is contained in:
@@ -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<Id, number>;
|
||||
@@ -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<Result>|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<Id,number>(),collectionReservations=new Set<Id>(),durablyPurged=new Set<Id>([...(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<Id,Promise<void>>();
|
||||
let sizingTask: Promise<void> | undefined;
|
||||
// 待写 Blob 与根快照分开保存，writePrefix 只把当前快照登记的资源/导入源文件纳入原子提交。
|
||||
// fileWrites 暂存未可靠落盘的 Blob；digestTasks 只去重并等待摘要工作，二者都不是持久任务表。
|
||||
// writePrefix 按某一捕获根的资源/源归属筛选文件，不能把未来编辑的字节提前配给旧根。
|
||||
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
|
||||
// 文件暂存及默认尺寸标记仅覆盖同步编辑事务；事务 reducer 将其写入可恢复资源记录。
|
||||
// staged 与 stagedDefaultSizes 仅是同步编辑事务的授权上下文；reduceRuntime 将其转成根中的资源事实。
|
||||
// 分配/上传/解码不能在 reducer 内启动，失败候选也不能留下第二份已接纳任务。
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
const stagedDefaultSizes = new Set<Id>();
|
||||
// 将用户提交标为 run.apply 的短暂上下文；不能让一次应用操作泄漏到后续普通编辑。
|
||||
@@ -168,7 +180,8 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const importRunners=new Map<Id,ReturnType<typeof createImportRunner>>(),importLoads=new Map<Id,Promise<void>>(),importHolds=new Map<symbol,Map<Id,()=>void>>();
|
||||
// 恢复会话存在期间阻止一般写入；会话负责持有/释放调度暂停及中止控制。
|
||||
let recovery: { controller: AbortController; release(): void; cancel():Promise<void> } | undefined;
|
||||
/** 复制准备期间保守保护未清除资源，解码期间保护正在读取的图片；供每次保留扫描派生。 */
|
||||
/** 每次派生临时保留清单：任一复制准备仍活跃就保守 pin 全部未 purge 的本地资源，解码按任务 ID pin。
|
||||
* Map 是本次扫描的输入，不发布第二份 resourcePins；任务结束后下次扫描自然不再包含该临时 owner。 */
|
||||
const admissionPins=(snapshot:EngineSnapshot)=>{const pins=new Map<Id,readonly Id[]>();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<Id, 'absent' | 'exists'> = 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<number, EngineSnapshot>();
|
||||
// 新项目的初始根也要保存；恢复镜像已经在磁盘上，不重复当成新提交。
|
||||
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<void> {
|
||||
if (writing) return writing;
|
||||
const task = Promise.resolve().then(async () => {
|
||||
/** 把已可靠写入的 generation 发布回 ledger；此更新可能产生新根，不能冒充它也已保存。 */
|
||||
/** 一次前缀结束后把已确认 generation 反映到 ledger；mutate 可能产生新的待保存根。
|
||||
* confirmedLedgerGeneration 在回执后保留推进义务，即使本次派生发布失败，下次也可补办。 */
|
||||
const publishDurability=()=>{if(editor.getSnapshot().runtime.ledger.durableGeneration<confirmedLedgerGeneration)mutate(ledger=>ledger.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<void> {
|
||||
if(durableRootRevision>=requiredRevision)return Promise.resolve();
|
||||
return writePrefix(requiredRevision).then(()=>durableRootRevision<requiredRevision?persistThrough(requiredRevision):undefined);
|
||||
}
|
||||
/** 排空当前保存义务，包含写回 durableGeneration 派生的新根；不等于网络同步完成。 */
|
||||
/** 循环排空 pending 和 durableGeneration 派生义务，不等于等待网络 ACK/生成/导入任务全部完成。
|
||||
* 与 persistThrough 的固定边界不同，循环期间新增本地根也纳入等待；调用方须先阻止需关闭的生产者。 */
|
||||
async function persist():Promise<void> {
|
||||
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<CollectionReport> {
|
||||
assertWritable();if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable');
|
||||
const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64;
|
||||
|
||||
Reference in New Issue
Block a user