docs(canvas): explain engine ownership durability and file retention
This commit is contained in:
@@ -1,3 +1,10 @@
|
||||
/**
|
||||
* 单项目运行时的装配与协调中心。
|
||||
* editor 拥有统一根快照,文档、history、ledger 和 ACK 基线经同一发布边界更新;
|
||||
* engine 把持久化、网络调度、资源、导入、生成和冲突恢复连接到这个边界。
|
||||
* 这里的 Promise/Map/Set 大多是当前进程的任务、租约或索引,不替代可恢复根中的业务事实。
|
||||
* 打开项目先取得存储领导权并解码恢复镜像;视图挂载/卸载与项目关闭是不同的生命周期。
|
||||
*/
|
||||
import {dataArray,dataRecord} from '../canvas-core/data-containers.ts';
|
||||
import {detachJsonData} from '../canvas-core/json-data.ts';
|
||||
import {createImportReadProof,IMPORT_PROOF_BLOCK_BYTES} from './import-read-proof.ts';
|
||||
@@ -46,7 +53,12 @@ import { stageAppliedRecovery } from './applied-recovery.ts';
|
||||
import { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type StoredAsset, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.ts';
|
||||
import type { StoragePort, StorageScope } from './storage-port.ts';
|
||||
|
||||
/** 摘要执行基础设施的失败码;资源摘要失败时保留具体原因,显式重试可将这些阻塞恢复为 waiting_digest。 */
|
||||
const digestFailures=new Set(['digest_capacity','digest_worker_failed','digest_worker_unavailable','invalid_digest_response','digest_pool_closed']);
|
||||
/** 打开项目的依赖注入参数。scope/document 必须属于同一项目;registry 解释模型和恢复数据。
|
||||
* storage 是本地可靠保存,persistence 是文档网络协议;assets/runs 为可选外部业务服务。
|
||||
* digest/image/display/depot 可共享注入;clock 和 createId 允许宿主或测试控制时间与身份。
|
||||
* loading 表示尚未读完的模型,不能把已有前缀当成完整项目。 */
|
||||
export type EngineOptions = {
|
||||
loading?:LoadingState;
|
||||
scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap<Id, number>;
|
||||
@@ -55,6 +67,9 @@ export type EngineOptions = {
|
||||
onError?: (error: unknown) => void;
|
||||
isCurrentIdentity?: () => boolean;
|
||||
};
|
||||
/** 一次暂停自动协调后的恢复会话,绑定开始时的根 revision 和服务端候选文档。
|
||||
* plan 纯计算用户选择的替代方案,resolve 发布指定方案,apply 尝试默认协调,cancel 释放会话。
|
||||
* conflict 只是诊断信息;不能把“发现冲突”理解为旧文档或请求已经被删掉。 */
|
||||
export type RecoverySession = {
|
||||
readonly baseRootRevision: number;
|
||||
readonly canonical: CanonicalProposal;
|
||||
@@ -68,103 +83,162 @@ export type RecoverySession = {
|
||||
cancel(): Promise<void>;
|
||||
};
|
||||
|
||||
/** One project owner. Detaching a view does not close this engine or abort persistent work. */
|
||||
/** 打开一个项目所有者实例。先独占本地写入,再选择恢复状态或初始服务端状态,
|
||||
* 接线 editor 与异步执行器;此工厂是异步入口,普通编辑提交仍在 editor 内同步发布。 */
|
||||
export async function openProjectEngine(options: EngineOptions) {
|
||||
if (options.scope.projectId !== options.document.projectId) throw fault('project_mismatch');
|
||||
// 在第一次 await 前捕获加载状态和版本,防止调用方随后修改原容器改变这次打开的基线。
|
||||
const openingLoading=options.loading?freeze(clone(options.loading)):undefined,openingVersions=readMap(options.versions);
|
||||
// 存储租约约束同一作用域的写者;后续提交和网络准入均复查领导权及账户身份。
|
||||
const leader = await options.storage.claim(options.scope);
|
||||
try {
|
||||
// 有本地恢复镜像时按 registry 解码整份根,不能只取文档而遗失历史、请求或资源任务。
|
||||
const image = await options.storage.recover(leader), recovered = image ? decodeRecovery(image, options.registry) : undefined;
|
||||
if (recovered && recovered.document.projectId !== options.scope.projectId) throw fault('project_mismatch');
|
||||
// 外部初始模型中的每张画布都必须有有效服务器版本,不能默认补成零。
|
||||
for (const canvas of options.document.canvases.keys()) if (!Number.isSafeInteger(openingVersions.get(canvas)) || openingVersions.get(canvas)! < 1) throw fault('missing_server_version', [canvas]);
|
||||
// 渐进读取的在途任务、准入暂停句柄及完成 revision:完成读取后仍要等对应根可靠保存。
|
||||
let loadingRead:Promise<Result>|undefined,loadingReadController:AbortController|undefined;
|
||||
let acceptingLoading=false,loadingRelease:(()=>void)|undefined,loadingCompletionRevision:number|undefined;
|
||||
// closing 阻止新工作,closed 表示生命周期结束;durableRootRevision 仅由存储回执推进。
|
||||
let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1;
|
||||
// 项目级摘要中止源,与单次任务 signal 合并;关闭时可撤销仍在处理的字节计算。
|
||||
const digestLifecycle=new AbortController();
|
||||
/** 优先使用注入的摘要服务,否则本地分段计算;两条路径都受项目生命周期中止约束。 */
|
||||
const sourceDigest=(bytes:Blob,signal?:AbortSignal)=>{const combined=signal?AbortSignal.any([signal,digestLifecycle.signal]):digestLifecycle.signal;return options.digests?options.digests.digest(bytes,combined):importSourceDigest(bytes,combined);};
|
||||
/** 错误通知只是旁路,监听方抛错不能修改项目状态或打断清理。 */
|
||||
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;
|
||||
/** 返回当前根/已保存根、关闭阶段和存储错误的稳定只读快照。 */
|
||||
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,并在准入时核对存储领导权和认证代次。
|
||||
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);
|
||||
if(depot&&(depot.scope.accountId!==options.scope.accountId||depot.scope.workspaceId!==options.scope.workspaceId))throw fault('asset_depot_scope_mismatch');
|
||||
// 复用资源的异步领用任务与待保存字节,独立于本地新文件上传队列。
|
||||
const adoptionTasks=new Map<AbortController,Promise<Result>>(),adoptedFiles=new Map<Id,{readonly bytes:Blob;readonly rootRevision:number}>();
|
||||
// 仅在同步提交窗口内有效的上下文,供 reducer 原子绑定当前正在领用的资源。
|
||||
let adopting:SharedAsset|undefined;let importAdoptions:readonly SharedAsset[]=[];
|
||||
/** 汇总单资源领用和导入批次领用;不把临时上下文持久化成第二份任务事实。 */
|
||||
const activeAdoptions=()=>[...(adopting?[adopting]:[]),...importAdoptions];
|
||||
// 同一个时钟用于轮询、退避和保留期判定,避免测试或宿主时间源互相矛盾。
|
||||
const runClock: Clock = options.clock ?? { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
|
||||
// displays 保存项目持有的释放函数;共享 displayCache 通过独立 owner 区分各项目使用者。
|
||||
const displays = new Set<() => void>();
|
||||
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 记录已删字节。
|
||||
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};
|
||||
// 每类清理单飞:重复调用共享在途 Promise,结束后才允许下一次扫描。
|
||||
let sourceCollection:Promise<CollectionReport>|undefined;
|
||||
let orphanCollection:Promise<{readonly markedIds:readonly Id[];readonly purgedIds:readonly Id[];readonly changedIds:readonly Id[];readonly complete:boolean;readonly blockers:readonly string[];readonly checkedRootRevision:number;readonly scanned:number;readonly nextAfter?:Id;readonly hashBudgetExhausted:boolean;readonly scanFinished:boolean;readonly durableRootRevision:number}>|undefined;
|
||||
let orphanController:AbortController|undefined;
|
||||
// 恢复时补齐所有已删文件集合,包括孤立文件与导入源,避免再次安排同一物理删除。
|
||||
for(const [id,file]of recovered?.runtime.orphanFiles??[])if(file.retirement?.purgedAt!==undefined)durablyPurged.add(id);
|
||||
for(const [id,entry]of importFiles(recovered?.runtime.imports))if(entry.file.retirement?.purgedAt!==undefined)durablyPurged.add(id);
|
||||
let collection:Promise<CollectionReport>|undefined;
|
||||
// 解码/尺寸补写/摘要等在途索引只用于生命周期等待与去重;恢复事实仍在 runtime 中。
|
||||
const imageTasks=new Map<Id,Promise<void>>();
|
||||
let sizingTask: Promise<void> | undefined;
|
||||
// 待写 Blob 与根快照分开保存,writePrefix 只把当前快照登记的资源/导入源文件纳入原子提交。
|
||||
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
|
||||
// 文件暂存及默认尺寸标记仅覆盖同步编辑事务;事务 reducer 将其写入可恢复资源记录。
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
const stagedDefaultSizes = new Set<Id>();
|
||||
// 将用户提交标为 run.apply 的短暂上下文;不能让一次应用操作泄漏到后续普通编辑。
|
||||
let applying: RunApplyInput | undefined;
|
||||
// 当前批次携带预期 generation,文档改动与导入游标只能在同一次提交中一致推进。
|
||||
let importing: { job: ImportJob; expectedGeneration: number } | undefined;
|
||||
// 证明更新的授权任务,以及读证明、验证器、准备任务等可取消的执行索引。
|
||||
let integrityOwner:Id|undefined;
|
||||
const proofReaders=new Map<Id,{fileId:Id;bytes:Blob;reader:ReturnType<typeof createImportReadProof>}>();
|
||||
const verificationTasks=new Map<Id,{controller:AbortController;promise:Promise<void>}>();
|
||||
const preparationControllers=new Map<Id,AbortController>();
|
||||
const copyPreparations=new Map<AbortController,Promise<Id>>();
|
||||
// runner/载入 Promise/暂停租约均按 job 或 token 索引;它们不是导入作业的权威状态。
|
||||
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;
|
||||
/** 复制准备期间保守保护未清除资源,解码期间保护正在读取的图片;供每次保留扫描派生。 */
|
||||
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'); };
|
||||
/** 合并已知加载资源 URL 与本次暂存/领用资源 URL,供渐进模型的命令校验识别合法引用。 */
|
||||
function assetURLs(){const root=editor.getSnapshot(),urls=loadingAssetURLs(root.runtime);return staged.length||activeAdoptions().length?new Map([...urls,...staged.map(resource=>[resource.id,options.assets!.sourceURL(resource.id)] as const),...activeAdoptions().map(ref=>[ref.asset.id,ref.asset.url] as const)]):urls;}
|
||||
|
||||
// 本地恢复优先;首次打开则从外部模型建立独立 ACK 文档、版本和空任务表。
|
||||
// loading v3+ 额外保存读取基线,后续前缀扩展不把本地编辑误当成服务端原始内容。
|
||||
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ imports:readMap(new Map()), recoveries:readMap(new Map()), runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: openingVersions, ...(openingLoading?{loading:openingLoading,...(openingLoading.version>=3?{loadingRead:Object.freeze({document:createDocument(options.document.projectId,[...options.document.canvases.values()],[...options.document.elements.values()],options.document.tombstones),versions:openingVersions})}:{})}:{}), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
|
||||
// 最初来自可靠镜像/初始状态,之后只在存储回执确认对应根后更新,而非看到内存 seal 就放行。
|
||||
let durableImportSeals=importSeals(initial.imports);
|
||||
validateLoadingRuntime(initial,options.registry);
|
||||
// 已领用资源绑定必须能被当前服务 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,而非先改文档后补队列。
|
||||
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 }; },
|
||||
/** undo/redo 同样受生命周期限制;旧加载协议还需逐项检查历史变更是否触及未就绪区域。 */
|
||||
traversalPolicy(entry, direction) { 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&&(editor.getSnapshot().runtime.loading?.version??0)<4){try{assertLoadingChanges(entry.changes,editor.getSnapshot().document,options.registry,editor.getSnapshot().runtime.loading,editor.getSnapshot().runtime.loadingRead);}catch(error){return{enabled:false,code:(error as Error).message,targets:[]};}}return options.traversalPolicy?.(entry,direction)??{enabled:true}; },
|
||||
onListenerError: report,
|
||||
/** 发布前的所有权与保留规则检查。候选根未通过时整次发布失败,不留下半份资源/加载状态。 */
|
||||
prepareRuntime:(previous,next)=>{
|
||||
if(!leader.active)throw fault('storage_leader_lost');
|
||||
// seal/字节证明只能由当前导入或校验上下文推进,普通 runtime 更新不能伪造完整性证据。
|
||||
assertImportEvidence(previous.runtime.imports,next.runtime.imports,integrityOwner??importing?.job.id);
|
||||
// 已有共享绑定不可删除或改写;新增项必须恰好对应当前提交上下文中的授权领用。
|
||||
if(!equal([...(previous.runtime.assetBindings??[])],[...(next.runtime.assetBindings??[])])){
|
||||
const before=new Map(previous.runtime.assetBindings??[]),after=new Map(next.runtime.assetBindings??[]);
|
||||
const permitted=new Map(activeAdoptions().map(ref=>[ref.asset.id,ref]));if(after.size<before.size||[...before].some(([id,binding])=>!equal(after.get(id),binding))||[...after].some(([id,binding])=>!before.has(id)&&(!permitted.has(id)||!equal(sharedAssetRef(binding),sharedAssetRef(permitted.get(id)!)))))throw fault('asset_bindings_owned');
|
||||
}
|
||||
// 加载状态、读取基线及关联 ACK/版本只能经加载协调或恢复通道推进。
|
||||
if(!acceptingLoading&&!recovery&&(previous.runtime.loading||next.runtime.loading||previous.runtime.loadingRead||next.runtime.loadingRead)&&(!equal(previous.runtime.loading,next.runtime.loading)||previous.runtime.loadingRead!==next.runtime.loadingRead||previous.runtime.acknowledged!==next.runtime.acknowledged||!equal(Object.fromEntries(previous.runtime.versions),Object.fromEntries(next.runtime.versions))))throw fault('loading_state_owned');
|
||||
// 共享资源身份与本地上传、孤立文件、导入源文件不能占用同一个 ID。
|
||||
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);
|
||||
// 先计算尺寸权限,再依文档/历史/账本/显示引用派生资源保留,最后处理源文件和孤立字节。
|
||||
const sizing=prepareDefaultImageSizing(previous,next);
|
||||
const runtime=prepareResourceRetention({...next,runtime:sizing},options.registry,displayPins,collectionReservations,admissionPins(next));
|
||||
// 这些引用事实变化会影响“是否仍无人使用”;保留准备函数据此重新检查退休标记。
|
||||
const changed=previous.document!==next.document||previous.history!==next.history||previous.runtime.ledger!==runtime.ledger||previous.runtime.resourcePins!==runtime.resourcePins;
|
||||
const imports=prepareImportFileRetention(previous.runtime,runtime,()=>resourceRetentionPins({...next,runtime},options.registry,displayPins,admissionPins({...next,runtime})),collectionReservations,changed);
|
||||
return prepareOrphanFiles(previous.runtime,imports,()=>orphanCoverage({...next,runtime:imports},options.registry,displayPins,admissionPins({...next,runtime:imports})),collectionReservations,changed);
|
||||
},
|
||||
/** 只把当前应用输出的用户意图转换为 run.apply,保留普通 editor 的命令生成路径。 */
|
||||
prepareIntent: intent => applying && intent.origin==='user' ? {...intent,operationType:'run.apply',action:applying} : intent,
|
||||
/** 在现有账本最大序号后分配;预留 staged 上传操作的位置,让上传先于引用它的文档操作。 */
|
||||
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 对象。
|
||||
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;
|
||||
// 为每个新资源先建立上传事实并阻塞为 waiting_digest,拿到原始字节摘要后才可编译网络请求。
|
||||
staged.forEach((resource, index) => {
|
||||
if (resources.has(resource.id) || ledger.getSnapshot().operations.has(resource.operationId)) throw fault('resource_identity_conflict', [resource.id]);
|
||||
// 从实际提交结果计算资源使用画布;loading v6 将其绑定到本次来源令牌和 URL。
|
||||
const canvasIds=[...new Set([...next.document.elements.values()].filter(element=>referencedResources([{kind:'element',id:element.id,before:null,after:element}],options.registry).has(resource.id)).map(element=>element.ownerCanvasId))].sort();
|
||||
const admitted=previous.loading?.version===6?freeze({...resource,loadingAdmission:{version:1 as const,sourceToken:previous.loading.sourceToken,sourceURL:options.assets!.sourceURL(resource.id),canvasIds}}):resource;
|
||||
resources.set(resource.id, stagedDefaultSizes.has(resource.id) ? captureDefaultImageSizing(admitted, next, commit, options.registry) : admitted);
|
||||
ledger.add(freeze({ id: resource.operationId, operationType: 'asset.upload', operationSchemaVersion: 1, sequence: firstSequence + index, origin: 'user', canvasIds: [], changes: [], dependencies: [], resourceId: resource.id }), {});
|
||||
ledger.block(resource.operationId, ['waiting_digest']);
|
||||
});
|
||||
// 文档操作在此补齐与旧操作、未完成生成及资源上传的因果依赖。
|
||||
for (const id of commit.operationIds) {
|
||||
const intent = next.operations.get(id)!, footprint = operationFootprint(intent, next.document);
|
||||
// 自动尺寸补写如果属于某个历史单元,把对应目标代次的 pending grant 标为已应用。
|
||||
if (intent.historyEffect) {
|
||||
const effect = intent.historyEffect;
|
||||
for (const [resourceId, resource] of resources) if (resource.sizing?.version === 1) {
|
||||
@@ -173,19 +247,24 @@ 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,以及与本操作足迹冲突的文档/应用操作,都成为先行依赖。
|
||||
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);
|
||||
// 引用尚未上传完成的资源时,文档同步必须等待该资源的上传操作。
|
||||
for (const id of referencedResources(intent.changes, options.registry)) {
|
||||
const resource = resources.get(id);
|
||||
if (resource && !resource.stored) dependencies.add(resource.operationId);
|
||||
}
|
||||
// 合并器需要起始存在性,才能区分本地 create→delete 可消除与已存在对象必须发送删除。
|
||||
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);
|
||||
}
|
||||
// 共享资源领用无需再上传;但须在同根记录使用它的逻辑操作和授权绑定。
|
||||
const assetBindings=new Map(previous.assetBindings??[]);
|
||||
for(const adopting of activeAdoptions()){
|
||||
const id=adopting.asset.id,first=commit.operationIds.map(operation=>next.operations.get(operation)!).find(intent=>referencedResources(intent.changes,options.registry).has(id));
|
||||
// 防止只有资源领用副作用而没有真正引用它的文档操作。
|
||||
if(!first)throw fault('adopted_resource_unused',[id]);
|
||||
const existing=assetBindings.get(id);if(existing&&!equal(sharedAssetRef(existing),sharedAssetRef(adopting)))throw fault('shared_asset_identity_conflict',[id]);
|
||||
if(!existing){
|
||||
@@ -193,55 +272,73 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
assetBindings.set(id,freeze({...adopting,operationId:first.id,...(previous.loading?.version===6?{loadingAdmission:{version:1 as const,sourceToken:previous.loading.sourceToken,sourceURL:adopting.asset.url,canvasIds,operationId:first.id}}:{})}));
|
||||
}
|
||||
}
|
||||
// 导入游标更新与本批元素发布同一边界;异步准备期间作业被替换/取消则拒绝旧 generation。
|
||||
const imports=new Map(previous.imports??[]);
|
||||
if(importing){if(imports.get(importing.job.id)?.generation!==importing.expectedGeneration)throw fault('stale_import_job');imports.set(importing.job.id,importing.job);}
|
||||
return Object.freeze({ ...previous, imports:readMap(imports), ...(assetBindings.size?{assetBindings:readMap(assetBindings)}:{}), resources: readMap(resources), ledger: ledger.getSnapshot() });
|
||||
},
|
||||
});
|
||||
// 待持久化快照按根 revision 递增插入 Map,形成写前缀队列;不会用最新根覆盖未保存旧根。
|
||||
const pending = new Map<number, EngineSnapshot>();
|
||||
// 新项目的初始根也要保存;恢复镜像已经在磁盘上,不重复当成新提交。
|
||||
if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot());
|
||||
// writing 将所有写入串行化;scheduled 合并自动保存微任务,失败标记用于准入恢复和导入背压。
|
||||
let writing: Promise<void> | undefined, scheduled = false, persistenceFailed = false;
|
||||
// Receipt-derived publication obligation survives a failed metadata publication.
|
||||
// 记录由可靠回执确认的 ledger generation;即使派生标记发布失败,也保留这项推进义务。
|
||||
let confirmedLedgerGeneration=editor.getSnapshot().runtime.ledger.durableGeneration;
|
||||
/** 保存不超过 throughRevision 的已捕获前缀。每份根镜像与对应字节写入/删除同事务提交,
|
||||
* 存储回执核对通过后才移除 pending、释放字节和推进可靠游标。已有写入时共享该 Promise。 */
|
||||
function writePrefix(throughRevision:number):Promise<void> {
|
||||
if (writing) return writing;
|
||||
const task = Promise.resolve().then(async () => {
|
||||
/** 把已可靠写入的 generation 发布回 ledger;此更新可能产生新根,不能冒充它也已保存。 */
|
||||
const publishDurability=()=>{if(editor.getSnapshot().runtime.ledger.durableGeneration<confirmedLedgerGeneration)mutate(ledger=>ledger.markDurable(confirmedLedgerGeneration));};
|
||||
try {
|
||||
while (pending.size) {
|
||||
const [revision, snapshot] = pending.entries().next().value!;
|
||||
if(revision>throughRevision)break;
|
||||
// 只附带此快照仍持有的资源/导入源 Blob;退休标记则派生原子字节删除列表。
|
||||
const sourceFiles=importFiles(snapshot.runtime.imports);
|
||||
const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||sourceFiles.has(id)).map(([id, bytes]) => ({ id, bytes }));
|
||||
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 根。
|
||||
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 可靠;后续网络准入依赖这些事实。
|
||||
confirmedLedgerGeneration=Math.max(confirmedLedgerGeneration,snapshot.runtime.ledger.generation);
|
||||
durableRootRevision = revision;durableImportSeals=importSeals(snapshot.runtime.imports); pending.delete(revision);notifyDurability();
|
||||
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
|
||||
for (const file of files) fileWrites.delete(file.id);
|
||||
// 只释放这次 depot cut 确实包含且不晚于本快照的领用字节,保留尚未写入的后来任务。
|
||||
for(const [id,entry]of adoptedFiles)if(entry.rootRevision<=snapshot.rootRevision&&depot?.writes.some(write=>write.ref.asset.id===id&&write.bytes===entry.bytes))adoptedFiles.delete(id);
|
||||
}
|
||||
} catch(error) {
|
||||
// Preserve the known durable prefix even if a later write fails.
|
||||
// 部分前缀成功、后续失败时仍发布成功证据,不能把已可靠写入的操作退回“从未保存”。
|
||||
publishDurability();throw error;
|
||||
}
|
||||
// One derived publication per captured prefix, not one per receipt.
|
||||
publishDurability();
|
||||
persistenceFailed = false;durabilityError=undefined;notifyDurability();
|
||||
});
|
||||
// 写失败更新独立的耐久性状态并保留 pending;成功后才允许恢复加载发送和刷新调度准入。
|
||||
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 已可靠保存。若共享的在途写入目标更早,完成后继续写到所需边界。 */
|
||||
function persistThrough(requiredRevision=editor.getSnapshot().rootRevision):Promise<void> {
|
||||
if(durableRootRevision>=requiredRevision)return Promise.resolve();
|
||||
return writePrefix(requiredRevision).then(()=>durableRootRevision<requiredRevision?persistThrough(requiredRevision):undefined);
|
||||
}
|
||||
/** 排空当前保存义务,包含写回 durableGeneration 派生的新根;不等于网络同步完成。 */
|
||||
async function persist():Promise<void> {
|
||||
do {await writePrefix(editor.getSnapshot().rootRevision);}
|
||||
while(pending.size||writing||confirmedLedgerGeneration>editor.getSnapshot().runtime.ledger.durableGeneration);
|
||||
}
|
||||
/** 本地上传资源的两阶段清理:先持久化无人引用时间,再于保留期后重新扫描并原子删字节。
|
||||
* 引用覆盖不完整就停止,限制每批文件数,返回标记与真正保存成功的删除结果。 */
|
||||
function collectResources(settings:{retentionMs:number;maxFiles?:number}) {
|
||||
assertWritable();
|
||||
if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable');
|
||||
@@ -255,6 +352,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const markedIds:Id[]=[],purgedIds:Id[]=[];
|
||||
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();
|
||||
// 待写或摘要处理中资源不参与清理;必须先证明没有文档、历史、任务或显示租约引用。
|
||||
const fresh=[...root.runtime.resources.values()].filter(resource=>!resource.retirement&&!pins.pins.has(resource.id)&&!fileWrites.has(resource.id)&&!digestTasks.has(resource.id)).slice(0,maxFiles);
|
||||
if(fresh.length){
|
||||
const result=editor.updateRuntime(previous=>{const resources=new Map(previous.resources);for(const resource of fresh){resources.set(resource.id,freeze({...resource,retirement:{version:1 as const,unreferencedAt:now}}));markedIds.push(resource.id);}return Object.freeze({...previous,resources:readMap(resources)});},root.rootRevision);
|
||||
@@ -263,8 +361,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
}
|
||||
root=editor.getSnapshot();pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root));
|
||||
if(!pins.complete)return report();
|
||||
// 标记后的 await 期间可能出现新引用,所以重新取根和 pins,再判定到期资源。
|
||||
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 冲突则撤销保留,不报告已删除。
|
||||
for(const resource of eligible)collectionReservations.add(resource.id);
|
||||
const result=editor.updateRuntime(previous=>{const resources=new Map(previous.resources);for(const resource of eligible)resources.set(resource.id,freeze({...resource,retirement:{...resource.retirement!,purgedAt:now}}));return Object.freeze({...previous,resources:readMap(resources)});},root.rootRevision);
|
||||
if(result.status==='rejected'){for(const resource of eligible)collectionReservations.delete(resource.id);throw fault(result.code,result.targets);}
|
||||
@@ -275,6 +375,8 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
})();
|
||||
collection=task.finally(()=>{collection=undefined;});return collection;
|
||||
}
|
||||
/** 导入源文件的同类两阶段清理;当前源只允许在任务终结且没有载入器时成为候选。
|
||||
* 旧分片/历史源也要经过引用覆盖和保留期,不按任务结束就立即丢弃全部原始字节。 */
|
||||
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;
|
||||
@@ -286,6 +388,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
let root=editor.getSnapshot(),pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root));
|
||||
const markedIds:Id[]=[],purgedIds:Id[]=[];
|
||||
const report=()=>Object.freeze({markedIds:Object.freeze(markedIds),purgedIds:Object.freeze(purgedIds),complete:pins.complete,checkedRootRevision:root.rootRevision,rootRevision:editor.getSnapshot().rootRevision,durableRootRevision});
|
||||
/** 每次读取最新 root/pins 计算候选,保存标记后的重新检查不会复用旧候选集合。 */
|
||||
const candidates=()=>[...importFiles(root.runtime.imports)].filter(([id,entry])=>!pins.pins.has(id)&&!fileWrites.has(id)&&(!entry.current||terminalImport(root.runtime.imports!.get(entry.jobId)!.status)&&!importLoads.has(entry.jobId)));
|
||||
if(!pins.complete)return report();
|
||||
const fresh=candidates().filter(([,entry])=>!entry.file.retirement).slice(0,maxFiles);
|
||||
@@ -307,13 +410,17 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
})();
|
||||
sourceCollection=task.finally(()=>{sourceCollection=undefined;});return sourceCollection;
|
||||
}
|
||||
/** 为备份/查看借出导入原文件。先加引用 pin 再异步读字节,校验长度及已有 SHA,
|
||||
* 成功返回必须释放的句柄;失败自动释放,关闭期间的迟到结果不再交给调用方。 */
|
||||
async function openImportSource(jobId:Id,fileId?:Id) {
|
||||
if(closing||closed)throw fault('engine_closing');const job=getImport(jobId),file=fileId!==undefined?importFiles(new Map([[jobId,job]])).get(fileId)?.file:job.source;if(!file)throw fault('import_source_unavailable');
|
||||
if(collectionReservations.has(file.fileId))throw fault('resource_collection_in_progress',[file.fileId]);
|
||||
if(file.retirement?.purgedAt!==undefined)throw fault('import_source_retired',[file.fileId]);
|
||||
displayPins.set(file.fileId,(displayPins.get(file.fileId)??0)+1);let released=false;
|
||||
/** 幂等归还本次 pin;从 displays 释放表移除,项目关闭不会重复扣除引用计数。 */
|
||||
const release=()=>{if(released)return;released=true;displays.delete(release);const count=displayPins.get(file.fileId)??1;if(count<=1)displayPins.delete(file.fileId);else displayPins.set(file.fileId,count-1);};displays.add(release);
|
||||
try{
|
||||
// pin 已建立后发布一次 runtime 更新,让保留规则撤销已过时的退休标记。
|
||||
if(file.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);}
|
||||
const bytes=fileWrites.get(file.fileId)??await options.storage.readFile(leader,file.fileId);
|
||||
if(closing||closed)throw fault('engine_closing');if(!bytes||bytes.size!==file.byteLength)throw fault('import_source_unavailable',[file.fileId]);
|
||||
@@ -321,6 +428,8 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
if(closing||closed)throw fault('engine_closing');return Object.freeze({bytes,release});
|
||||
}catch(error){release();if(closing||closed)throw fault('engine_closing');throw error;}
|
||||
}
|
||||
/** 校验存储提供的文件分页:绑定根 revision、ID 严格递增、大小合法、尾游标匹配。
|
||||
* 读取期间根变更则拒绝旧清单,不能把两个时点的所有权证据拼成同一份库存。 */
|
||||
async function scanLocalFiles(root:EngineSnapshot,after:Id|undefined,limit:number){
|
||||
if(!options.storage.scanFiles)throw fault('file_inventory_unavailable');
|
||||
const page=await options.storage.scanFiles(leader,{expectedRootRevision:root.rootRevision,after,limit});
|
||||
@@ -330,29 +439,41 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
for(const file of page.files){if(!file.id||typeof file.id!=='string'||previous!==undefined&&file.id<=previous||!Number.isSafeInteger(file.byteLength)||file.byteLength<0||typeof file.mimeType!=='string')throw fault('invalid_file_inventory');previous=file.id;}
|
||||
if(page.nextAfter!==undefined&&(!page.files.length||page.nextAfter!==previous))throw fault('invalid_file_inventory');return page;
|
||||
}
|
||||
/** 先保存根再读取有界库存,给每个文件附上资源/导入源/孤立文件归属与 pin 依据。
|
||||
* 这是诊断读取;unregistered 只表示未登记,不直接证明该文件可安全删除。 */
|
||||
async function inventoryLocalFiles(settings:{after?:Id;limit?:number}={}){
|
||||
assertWritable();const limit=settings.limit??128;if(!Number.isSafeInteger(limit)||limit<1||limit>128||settings.after!==undefined&&(typeof settings.after!=='string'||!settings.after))throw fault('invalid_file_inventory');
|
||||
await persist();assertWritable();const root=editor.getSnapshot(),page=await scanLocalFiles(root,settings.after,limit),sources=importFiles(root.runtime.imports),pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root));
|
||||
return Object.freeze({rootRevision:root.rootRevision,complete:pins.complete,blockers:pins.blockers,nextAfter:page.nextAfter,files:Object.freeze(page.files.map(file=>{const resource=root.runtime.resources.get(file.id),source=sources.get(file.id),orphan=root.runtime.orphanFiles?.get(file.id);return Object.freeze({...file,ownership:resource?'resource' as const:source?'import-source' as const:orphan?'registered-orphan' as const:'unregistered' as const,...(source?{jobId:source.jobId,current:source.current}:{}),retired:(resource?.retirement??source?.file.retirement??orphan?.retirement)?.purgedAt!==undefined,pins:Object.freeze([...(pins.pins.get(file.id)??[])])});}))});
|
||||
}
|
||||
/** 分页盘点并清理没有资源/导入源归属的本地文件。
|
||||
* 仅对本批核对过长度、MIME 和 SHA 的候选字节安排清理;先记无人引用时间,再经保留期删除。
|
||||
* 文件数与散列字节预算分别限流,nextAfter 指向已处理边界,调用方可继续下一页。 */
|
||||
function collectOrphanFiles(settings:{retentionMs:number;maxFiles?:number;after?:Id;pageSize?:number;maxHashBytes?:number;signal?:AbortSignal}){
|
||||
assertWritable();if(options.storage.fileDeletion!=='atomic-v1'||!options.storage.scanFiles)throw fault('file_inventory_unavailable');
|
||||
const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64,pageSize=settings.pageSize??128,maxHashBytes=settings.maxHashBytes??64*1024*1024;
|
||||
if(!Number.isSafeInteger(retentionMs)||retentionMs<0||![maxFiles,pageSize].every(value=>Number.isSafeInteger(value)&&value>=1&&value<=128)||!Number.isSafeInteger(maxHashBytes)||maxHashBytes<1||settings.after!==undefined&&(typeof settings.after!=='string'||!settings.after))throw fault('invalid_file_inventory');
|
||||
if(orphanCollection)return orphanCollection;
|
||||
// 外部取消与项目关闭均可停止库存读取/散列;所有异步边界后仍要检查写入权限。
|
||||
const controller=new AbortController(),signal=settings.signal?AbortSignal.any([controller.signal,settings.signal]):controller.signal;orphanController=controller;
|
||||
/** 取消只停止本轮工作,不把尚未验证的文件登记为可删除。 */
|
||||
const check=()=>{assertWritable();if(signal.aborted)throw fault('file_inventory_aborted');};
|
||||
const task=(async()=>{
|
||||
await persist();check();const now=runClock.now();if(!Number.isSafeInteger(now)||now<0)throw fault('invalid_resource_collection_clock');
|
||||
let root=editor.getSnapshot(),pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root)),scanned=0,nextAfter=settings.after,hashBudgetExhausted=false,scanFinished=false;
|
||||
const markedIds:Id[]=[],purgedIds:Id[]=[],changedIds:Id[]=[],verified=new Map<Id,OrphanFile>();
|
||||
/** 区分引用覆盖完整、扫描到末页与字节预算耗尽;这些状态不能只压成一个 complete。 */
|
||||
const result=()=>Object.freeze({markedIds:Object.freeze(markedIds),purgedIds:Object.freeze(purgedIds),changedIds:Object.freeze(changedIds),complete:pins.complete,blockers:pins.blockers,checkedRootRevision:root.rootRevision,scanned,nextAfter,hashBudgetExhausted,scanFinished,durableRootRevision});
|
||||
if(!pins.complete)return result();
|
||||
// 存储清单只提供候选身份;随后读取实际 Blob 并散列,才形成这次清理使用的内容证据。
|
||||
const page=await abortableFileRead(scanLocalFiles(root,settings.after,pageSize),signal),sources=importFiles(root.runtime.imports);let hashed=0,finished=true;
|
||||
for(const file of page.files){
|
||||
// 已有资源/源文件、已清除文件、仍被 pin 或等待写入的文件都排除,不能靠“未登记”直接删。
|
||||
check();const registered=root.runtime.orphanFiles?.get(file.id),candidate=!root.runtime.resources.has(file.id)&&!sources.has(file.id)&®istered?.retirement?.purgedAt===undefined&&!pins.pins.has(file.id)&&!fileWrites.has(file.id);
|
||||
if(candidate){
|
||||
// 未处理到的候选不推进游标,下一轮可从前一个 ID 继续核对它。
|
||||
if(verified.size>=maxFiles){finished=false;break;}if(file.byteLength>maxHashBytes-hashed){hashBudgetExhausted=true;finished=false;break;}
|
||||
// 每次 await 后检查根身份;文档/任务变化会使原先的所有权扫描失效。
|
||||
const bytes=await abortableFileRead(options.storage.readFile(leader,file.id),signal);check();if(editor.getSnapshot()!==root)throw fault('stale_file_inventory');
|
||||
if(!(bytes instanceof Blob)||bytes.size!==file.byteLength||bytes.type!==file.mimeType)throw fault('file_inventory_changed',[file.id]);
|
||||
const digest=await abortableFileRead(sourceDigest(bytes,signal),signal);check();if(editor.getSnapshot()!==root)throw fault('stale_file_inventory');hashed+=bytes.size;
|
||||
@@ -360,14 +481,17 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
}
|
||||
scanned++;nextAfter=file.id;
|
||||
}
|
||||
// 整页走完才使用存储尾游标;部分处理时保留最后实际检查过的文件 ID。
|
||||
if(finished)nextAfter=page.nextAfter;scanFinished=finished&&page.nextAfter===undefined;
|
||||
pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root));if(!pins.complete)return result();
|
||||
// 同 ID 字节发生变化时重新起算保留期;旧退休证据不能授权删除一份新内容。
|
||||
const updates=new Map<Id,OrphanFile>();
|
||||
for(const [id,file]of verified){if(pins.pins.has(id))continue;const old=root.runtime.orphanFiles?.get(id),changed=!!old&&(old.digest!==file.digest||old.byteLength!==file.byteLength||old.mimeType!==file.mimeType);
|
||||
if(!old?.retirement||changed){updates.set(id,freeze({...file,retirement:{version:1 as const,unreferencedAt:now}}));markedIds.push(id);if(changed)changedIds.push(id);}
|
||||
}
|
||||
if(updates.size){const update=editor.updateRuntime(previous=>{const files=new Map(previous.orphanFiles);for(const [id,file]of updates)files.set(id,file);return Object.freeze({...previous,orphanFiles:readMap(files)});},root.rootRevision);if(update.status==='rejected')throw fault(update.code,update.targets);await persist();check();}
|
||||
root=editor.getSnapshot();pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root));if(!pins.complete)return result();
|
||||
// 保存标记后重新拿当前根,要求仍无引用、仍无归属且内容与本轮证明一致才可删除。
|
||||
const eligible=[...verified].filter(([id,proof])=>{const file=root.runtime.orphanFiles?.get(id);return file?.retirement&&file.retirement.purgedAt===undefined&&now-file.retirement.unreferencedAt>=retentionMs&&file.digest===proof.digest&&file.byteLength===proof.byteLength&&file.mimeType===proof.mimeType&&!pins.pins.has(id)&&!root.runtime.resources.has(id)&&!importFiles(root.runtime.imports).has(id)&&!fileWrites.has(id);});
|
||||
if(eligible.length){
|
||||
for(const [id]of eligible)collectionReservations.add(id);
|
||||
@@ -377,8 +501,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
}
|
||||
return result();
|
||||
})();
|
||||
// 成功、失败或取消都撤销单飞索引;不会删除根中的退休记录或伪造 purged 回执。
|
||||
orphanCollection=task.finally(()=>{orphanCollection=undefined;if(orphanController===controller)orphanController=undefined;});return orphanCollection;
|
||||
}
|
||||
/** 将可变 ledger helper 的一次归约转换为 editor 的同根 runtime 发布;拒绝结果转成领域异常。 */
|
||||
function mutate(reduce: (ledger: ReturnType<typeof createLedger>) => void) {
|
||||
const result = editor.updateRuntime(previous => { const ledger = createLedger(previous.ledger); reduce(ledger); return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); });
|
||||
if (result.status === 'rejected') throw fault(result.code, result.targets);
|
||||
|
||||
Reference in New Issue
Block a user