Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adopt authorized shared assets with durable project bindings and recovery

Browse Source
This commit is contained in:
Codex Bot
2026-11-09 16:11:25 +08:00
parent a294c6ca41
commit a626e6c300
17 changed files with 214 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+56 -15
View File
@@ -1,3 +1,4 @@
import {createAssetDepot,sharedAssetRef,type AssetDepot,type SharedAsset} from './asset-depot.ts';
import {prepareDepotCut} from './depot-cut.ts';
import {loadingAssetURLs} from './loading-assets.ts';
import {loadingImportAdmission,validateLoadingCopySource} from './loading-import.ts';
@@ -34,14 +35,14 @@ import { decodeRecovery, encodeRecovery } from './recovery.ts';
import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts';
import { planConflictRecovery, type ConflictPlan, type RecoveryChoice } from './conflicts.ts';
import { stageAppliedRecovery } from './applied-recovery.ts';
import { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.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';
const digestFailures=new Set(['digest_capacity','digest_worker_failed','digest_worker_unavailable','invalid_digest_response','digest_pool_closed']);
export type EngineOptions = {
loading?:LoadingState;
scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap<Id, number>;
storage: StoragePort; displays?: DisplayCache; images?:Pick<ImageDecoderPort,'decode'>; digests?:Pick<DigestPort,'digest'>; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id;
storage: StoragePort; assetDepot?:AssetDepot; displays?: DisplayCache; images?:Pick<ImageDecoderPort,'decode'>; digests?:Pick<DigestPort,'digest'>; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id;
historyLimit?: number; queryBudget?:EditorOptions['queryBudget']; policy?: EditorOptions['policy']; traversalPolicy?: EditorOptions['traversalPolicy']; clock?: Clock;
onError?: (error: unknown) => void;
isCurrentIdentity?: () => boolean;
@@ -76,6 +77,10 @@ export async function openProjectEngine(options: EngineOptions) {
const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } };
const createId = options.createId ?? (() => crypto.randomUUID());
const network = createNetworkGate(2);
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,Blob>();
let adopting:SharedAsset|undefined;
const runClock: Clock = options.clock ?? { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
const displays = new Set<() => void>();
const displayOwner = Object.freeze({}), displayCache = options.displays ?? createDisplayCache({ onError: options.onError });
@@ -100,18 +105,25 @@ 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;
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;};
const assertWritable = (complete=true) => { 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'); };
function assetURLs(){const root=editor.getSnapshot(),urls=loadingAssetURLs(root.runtime);return staged.length?new Map([...urls,...staged.map(resource=>[resource.id,options.assets!.sourceURL(resource.id)] as const)]):urls;}
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'); };
function assetURLs(){const root=editor.getSnapshot(),urls=loadingAssetURLs(root.runtime);return staged.length||adopting?new Map([...urls,...staged.map(resource=>[resource.id,options.assets!.sourceURL(resource.id)] as const),...(adopting?[[adopting.asset.id,adopting.asset.url] as const]:[])]):urls;}
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()) });
validateLoadingRuntime(initial,options.registry);
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]);
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(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 }; },
traversalPolicy(entry, direction) { 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}; },
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 }; },
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');
if(!equal([...(previous.runtime.assetBindings??[])],[...(next.runtime.assetBindings??[])])){
const before=new Map(previous.runtime.assetBindings??[]),after=new Map(next.runtime.assetBindings??[]);
if(!adopting||after.size!==before.size+1||before.has(adopting.asset.id)||!after.has(adopting.asset.id)||!equal(sharedAssetRef(after.get(adopting.asset.id)!),sharedAssetRef(adopting))||[...before].some(([id,binding])=>!equal(after.get(id),binding)))throw fault('asset_bindings_owned');
}
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');
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));
@@ -154,9 +166,19 @@ export async function openProjectEngine(options: EngineOptions) {
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??[]);
if(adopting){
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){
const canvasIds=[...new Set([...next.document.elements.values()].filter(element=>referencedResources([{kind:'element',id:element.id,before:null,after:element}],options.registry).has(id)).map(element=>element.ownerCanvasId))].sort();
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}}:{})}));
}
}
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), resources: readMap(resources), ledger: ledger.getSnapshot() });
return Object.freeze({ ...previous, imports:readMap(imports), ...(assetBindings.size?{assetBindings:readMap(assetBindings)}:{}), resources: readMap(resources), ledger: ledger.getSnapshot() });
},
});
const pending = new Map<number, EngineSnapshot>();
@@ -170,12 +192,13 @@ export async function openProjectEngine(options: EngineOptions) {
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)];
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,displays:displayPins,temporary:admissionPins(snapshot)}):undefined;
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;
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles,...(depot?{depot}:{}) });
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
durableRootRevision = revision; pending.delete(revision);
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
for (const file of files) fileWrites.delete(file.id);
for(const id of snapshot.runtime.assetBindings?.keys()??[])adoptedFiles.delete(id);
if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation));
}
persistenceFailed = false;
@@ -875,13 +898,13 @@ export async function openProjectEngine(options: EngineOptions) {
canStageFiles(canvasId:Id){try{assertWritable(false);if(!options.assets)throw fault('asset_interpreter_unavailable');const snapshot=editor.getSnapshot();if(!snapshot.document.canvases.has(canvasId))throw fault('canvas_not_loaded',[canvasId]);if(snapshot.runtime.loading&&(snapshot.runtime.loading.version!==6||!loadingQueryCoverage(snapshot).completeCanvasIds.includes(canvasId)))throw fault('loading_scope_incomplete',[canvasId]);const permission=options.policy?.({type:'changes.apply',changes:[]},snapshot.document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);return{enabled:true as const};}catch(error){const result=reject(error);return{enabled:false as const,code:result.code,targets:result.targets};}},
stageFiles(files: readonly StagedFile[], callback: (resources: readonly ResourceHandle[], transaction: Transaction) => unknown) {
assertWritable(false);if(editor.getSnapshot().runtime.loading&&editor.getSnapshot().runtime.loading?.version!==6)throw fault('model_loading');
if (staged.length) throw fault('reentrant_dispatch');
if (staged.length||adopting) throw fault('reentrant_dispatch');
if (!options.assets) throw fault('asset_interpreter_unavailable');
const resources = files.map(file => {
if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name || file.defaultSize !== undefined && typeof file.defaultSize !== 'boolean') throw fault('invalid_staged_file');
return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size,...(options.images?{image:{version:1 as const,status:'pending' as const}}:{}) });
});
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || editor.getSnapshot().runtime.orphanFiles?.has(resource.id) || fileWrites.has(resource.id) || importFiles(editor.getSnapshot().runtime.imports).has(resource.id))) throw fault('resource_identity_conflict');
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || editor.getSnapshot().runtime.assetBindings?.has(resource.id) || editor.getSnapshot().runtime.orphanFiles?.has(resource.id) || fileWrites.has(resource.id) || importFiles(editor.getSnapshot().runtime.imports).has(resource.id))) throw fault('resource_identity_conflict');
staged = resources; resources.forEach((resource, index) => { fileWrites.set(resource.id, files[index].bytes);if(files[index].defaultSize)stagedDefaultSizes.add(resource.id); });
try {
const handles = freeze(resources.map(resource => ({ id: resource.id, src: options.assets!.sourceURL(resource.id) })));
@@ -890,6 +913,22 @@ export async function openProjectEngine(options: EngineOptions) {
return result;
} finally { staged = []; stagedDefaultSizes.clear();for (const resource of resources) if (!editor.getSnapshot().runtime.resources.has(resource.id)) fileWrites.delete(resource.id); }
},
async adoptResource(id:Id,callback:(resource:ResourceHandle&{readonly asset:StoredAsset},transaction:Transaction)=>unknown,settings:{signal?:AbortSignal}={}):Promise<Result>{
assertWritable(false);if(!depot||!options.assets?.authority||!options.storage.assetDepot)throw fault('asset_depot_unavailable');if(editor.getSnapshot().runtime.loading&&editor.getSnapshot().runtime.loading?.version!==6)throw fault('model_loading');
if(staged.length||adopting)throw fault('reentrant_dispatch');
const controller=new AbortController(),signal=AbortSignal.any([controller.signal,...(settings.signal?[settings.signal]:[])]);
const task=(async()=>{
const authorized=await depot.authorize(leader,{scope:options.assets!.authority!.scope,resolve:(id,abort)=>network.run(abort,()=>options.assets!.authority!.resolve(id,abort)),read:(id,abort)=>network.run(abort,()=>options.assets!.authority!.read(id,abort))},id,signal);
assertWritable(false);signal.throwIfAborted();if(authorized.ref.asset.url!==options.assets!.sourceURL(id))throw fault('asset_authorization_mismatch');
if(staged.length||adopting)throw fault('reentrant_dispatch');const runtime=editor.getSnapshot().runtime;
if(runtime.resources.has(id)||runtime.orphanFiles?.has(id)||importFiles(runtime.imports).has(id)||fileWrites.has(id)||collectionReservations.has(id))throw fault('resource_identity_conflict',[id]);
const previous=adoptedFiles.get(id);adopting=authorized.ref;adoptedFiles.set(id,authorized.bytes);
let committed=false;
try{const result=editor.transaction(tx=>callback(freeze({id,src:authorized.ref.asset.url,asset:authorized.ref.asset}),tx),{label:'Add existing image'});committed=result.status==='applied'&&result.operationIds.length>0;return result;}
finally{adopting=undefined;if(!committed){if(previous)adoptedFiles.set(id,previous);else adoptedFiles.delete(id);}}
})();
adoptionTasks.set(controller,task);void task.finally(()=>adoptionTasks.delete(controller)).catch(()=>{});return task;
},
async displayResource(id: Id, settings: { signal?: AbortSignal } = {}) {
assertDisplayIdentity();
if(collectionReservations.has(id))throw fault('resource_collection_in_progress',[id]);
@@ -913,9 +952,11 @@ export async function openProjectEngine(options: EngineOptions) {
if(source?.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);}
}
if(released)throw fault('display_aborted');
const key=JSON.stringify(retained?['asset',id,retained.operationId,retained.retirement?.purgedAt??null]:orphan?['orphan',id,orphan.digest]:source?['source',id,source.sha256??null]:['unmanaged',id]);
const binding=editor.getSnapshot().runtime.assetBindings?.get(id);
const key=JSON.stringify(binding?['shared-asset',id,binding.authority,binding.digest]:retained?['asset',id,retained.operationId,retained.retirement?.purgedAt??null]:orphan?['orphan',id,orphan.digest]:source?['source',id,source.sha256??null]:['unmanaged',id]);
lease=displayCache.acquire(displayOwner,key,async signal=>{
const bytes=fileWrites.get(id)??await options.storage.readFile(leader,id);
let bytes=adoptedFiles.get(id)??fileWrites.get(id)??await options.storage.readFile(leader,id);
if(!bytes&&binding&&depot)bytes=await depot.read(leader,binding,signal);
assertDisplayIdentity();if(signal.aborted)throw fault('display_aborted');
if(orphan&&(!bytes||bytes.size!==orphan.byteLength||bytes.type!==orphan.mimeType||await sourceDigest(bytes,signal)!==orphan.digest))throw fault('file_inventory_changed',[id]);
assertDisplayIdentity();if(signal.aborted)throw fault('display_aborted');
@@ -941,15 +982,15 @@ export async function openProjectEngine(options: EngineOptions) {
cancelRecovery:async()=>{await recovery?.cancel();},
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
retry: async () => { if(recovery)await recovery.cancel();assertWritable(false);await persist();retryDefaultSizes();const runtime=editor.getSnapshot().runtime,ids=[...runtime.resources.values()].filter(resource=>!resource.digest&&runtime.ledger.operations.get(resource.operationId)?.blockers.some(code=>digestFailures.has(code))).map(resource=>resource.operationId);if(ids.length){mutate(ledger=>{for(const id of ids)ledger.block(id,['waiting_digest']);});prepareDigests();}if(options.images){editor.updateRuntime(previous=>{const resources=new Map(previous.resources);let changed=false;for(const [id,resource]of resources)if(resource.image?.version===1&&resource.image.status==='failed'){resources.set(id,freeze({...resource,image:{version:1 as const,status:'pending' as const}}));changed=true;}return changed?Object.freeze({...previous,resources:readMap(resources)}):previous;});prepareImages();}await scheduler.resume();await runTracker.retry(); },
async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist();await scheduler.settle();await persist(); },
async settle() { await Promise.resolve();await Promise.allSettled([...adoptionTasks.values()]);await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]);while(imageTasks.size)await Promise.all([...imageTasks.values()]);while(sizingTask)await sizingTask; await scheduler.settle(); await runTracker.settle(); await persist();await scheduler.settle();await persist(); },
async close() {
if (closed) return;
closing = true;loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
closing = true;loadingReadController?.abort();await loadingRead?.catch(()=>{});unsubscribeSizing();orphanController?.abort();await Promise.allSettled([collection,sourceCollection,orphanCollection].filter(Boolean));digestLifecycle.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]);for(const controller of adoptionTasks.keys())controller.abort();await Promise.allSettled([...adoptionTasks.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]);while(digestTasks.size)await Promise.allSettled([...digestTasks.values()]);while(imageTasks.size)await Promise.allSettled([...imageTasks.values()]);while(sizingTask)await sizingTask;
recovery?.controller.abort();recovery?.release();
for (const release of [...displays]) release();displayCache.releaseOwner(displayOwner);if(!options.displays)displayCache.close();
// Failure keeps the leader/recovery material available for an explicit retry of close.
await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true;
await persist(); unsubscribe(); editor.dispose();if(!options.assetDepot)depot?.close();adoptedFiles.clear(); await leader.release(); closed = true;
},
});
} catch (error) { await leader.release(); throw error; }