Inventory scoped files and atomically collect verified legacy orphans

This commit is contained in:
2026-11-08 23:21:27 +08:00
parent 4b594f85eb
commit 3ccb8dbc5c
9 changed files with 232 additions and 23 deletions
+2
View File
@@ -1,3 +1,4 @@
import type {OrphanFile} from './orphan-files.ts';
import type { ImportJob } from './imports.ts';
import type { Document, Id, Json, OperationIntent, Snapshot } from '../canvas-core/types.ts';
import type { LedgerSnapshot, FrozenRequest } from './ledger.ts';
@@ -7,6 +8,7 @@ import type { ResourceRecord } from './resources.ts';
import type { RecoveryAudit } from './conflicts.ts';
export type RuntimeState = {
readonly orphanFiles?:ReadonlyMap<Id,OrphanFile>;
readonly imports?: ReadonlyMap<Id, ImportJob>;
readonly ledger: LedgerSnapshot;
readonly runs: ReadonlyMap<Id, RunRecord>;
+77 -9
View File
@@ -1,3 +1,4 @@
import {orphanCoverage,prepareOrphanFiles,abortableFileRead,type OrphanFile} from './orphan-files.ts';
import {importFiles,terminalImport,markImportFiles,prepareImportFileRetention,type FileRetirement} from './import-file-retention.ts';
import {resourceRetentionPins,prepareResourceRetention} from './resource-retention.ts';
import { encodeIndexedCopySnapshot, openIndexedCopySnapshot, type IndexedCopySnapshot } from './copy-snapshot-index.ts';
@@ -66,6 +67,9 @@ export async function openProjectEngine(options: EngineOptions) {
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};
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;
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
@@ -85,7 +89,9 @@ export async function openProjectEngine(options: EngineOptions) {
onListenerError: report,
prepareRuntime:(previous,next)=>{
const runtime=prepareResourceRetention(next,options.registry,displayPins,collectionReservations,admissionPins(next));
return prepareImportFileRetention(previous.runtime,runtime,()=>resourceRetentionPins({...next,runtime},options.registry,displayPins,admissionPins({...next,runtime})),collectionReservations,previous.document!==next.document||previous.history!==next.history||previous.runtime.ledger!==runtime.ledger||previous.runtime.resourcePins!==runtime.resourcePins);
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);
},
prepareIntent: intent => applying && intent.origin==='user' ? {...intent,operationType:'run.apply',action:applying} : intent,
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; },
@@ -126,7 +132,7 @@ export async function openProjectEngine(options: EngineOptions) {
const [revision, snapshot] = pending.entries().next().value!;
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)];
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 receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files,deleteFiles });
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
durableRootRevision = revision; pending.delete(revision);
@@ -218,6 +224,64 @@ 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;}
}
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});
if(closing||closed)throw fault('engine_closing');if(editor.getSnapshot()!==root)throw fault('stale_file_inventory');
let previous=after;
if(page.rootRevision!==root.rootRevision||!Array.isArray(page.files)||page.files.length>limit)throw fault('invalid_file_inventory');
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;
}
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)??[])])});}))});
}
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>();
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();
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){
check();const registered=root.runtime.orphanFiles?.get(file.id),candidate=!root.runtime.resources.has(file.id)&&!sources.has(file.id)&&registered?.retirement?.purgedAt===undefined&&!pins.pins.has(file.id)&&!fileWrites.has(file.id);
if(candidate){
if(verified.size>=maxFiles){finished=false;break;}if(file.byteLength>maxHashBytes-hashed){hashBudgetExhausted=true;finished=false;break;}
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;
verified.set(file.id,freeze({version:1 as const,id:file.id,byteLength:file.byteLength,mimeType:file.mimeType,digest}));
}
scanned++;nextAfter=file.id;
}
if(finished)nextAfter=page.nextAfter;scanFinished=finished&&page.nextAfter===undefined;
pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root));if(!pins.complete)return result();
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);
const update=editor.updateRuntime(previous=>{const files=new Map(previous.orphanFiles);for(const [id]of eligible){const file=files.get(id)!;files.set(id,freeze({...file,retirement:{...file.retirement!,purgedAt:now}}));}return Object.freeze({...previous,orphanFiles:readMap(files)});},root.rootRevision);
if(update.status==='rejected'){for(const [id]of eligible)collectionReservations.delete(id);throw fault(update.code,update.targets);}
await persist();purgedIds.push(...eligible.map(([id])=>id));
}
return result();
})();
orphanCollection=task.finally(()=>{orphanCollection=undefined;if(orphanController===controller)orphanController=undefined;});return orphanCollection;
}
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);
@@ -500,7 +564,7 @@ export async function openProjectEngine(options: EngineOptions) {
openSource:openImportSource,
startNdjson(bytes:Blob,targetCanvasId:Id):Id{
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source');
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
const entry=canvas.parent?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined;
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const});
fileWrites.set(fileId,bytes);
@@ -510,8 +574,8 @@ export async function openProjectEngine(options: EngineOptions) {
async duplicate(selection:readonly Target[],settings:{readonly worldOffset?:Point;readonly signal?:AbortSignal;readonly viewCanvasId?:Id}={}):Promise<Id>{
assertWritable();if(collectionReservations.size)throw fault('resource_collection_in_progress',[...collectionReservations]);
const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
if([...editor.getSnapshot().runtime.resources.values()].some(resource=>resource.retirement&&resource.retirement.purgedAt===undefined)){
const reset=editor.updateRuntime(previous=>{const resources=new Map(previous.resources);for(const [id,resource]of resources)if(resource.retirement&&resource.retirement.purgedAt===undefined){const {retirement:_,...kept}=resource;resources.set(id,freeze(kept));}return Object.freeze({...previous,resources:readMap(resources)});});if(reset.status==='rejected')throw fault(reset.code,reset.targets);
if([...editor.getSnapshot().runtime.resources.values(),...(editor.getSnapshot().runtime.orphanFiles?.values()??[])].some(file=>file.retirement&&file.retirement.purgedAt===undefined)){
const reset=editor.updateRuntime(previous=>{const resources=new Map(previous.resources);for(const [id,resource]of resources)if(resource.retirement&&resource.retirement.purgedAt===undefined){const {retirement:_,...kept}=resource;resources.set(id,freeze(kept));}const files=new Map(previous.orphanFiles);for(const [id,file]of files)if(file.retirement&&file.retirement.purgedAt===undefined){const {retirement:_,...kept}=file;files.set(id,freeze(kept));}return Object.freeze({...previous,resources:readMap(resources),...(previous.orphanFiles?{orphanFiles:readMap(files)}:{})});});if(reset.status==='rejected')throw fault(reset.code,reset.targets);
}
const captured=editor.getSnapshot(),target=settings.viewCanvasId??(selection[0]?captured.document.elements.get(selection[0].id)?.ownerCanvasId:undefined),signal=settings.signal,controller=new AbortController(),abort=()=>controller.abort();
const targets=freeze(clone(selection)),offset=freeze(clone(settings.worldOffset??{x:32,y:32}));
@@ -522,7 +586,7 @@ export async function openProjectEngine(options: EngineOptions) {
const bytes=await encodeIndexedCopySnapshot(captured.document,controller.signal);assertWritable();if(controller.signal.aborted)throw fault('copy_aborted');
const canvas=captured.document.canvases.get(target);if(!canvas)throw fault('canvas_not_loaded',[target]);
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;
if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
const entry=canvas.parent?captured.document.elements.get(canvas.parent.entryElementId)!:undefined;
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'duplicate-snapshot',snapshotVersion:2,name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:[]},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'preparing' as const,preparation:{version:4 as const,registrySignature:copyRegistrySignature(captured.document,options.registry),selection:targets,worldOffset:offset,allocationIds:[],steps:0}});
fileWrites.set(fileId,bytes);const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
@@ -537,7 +601,7 @@ export async function openProjectEngine(options: EngineOptions) {
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
for(const job of initial.imports?.values()??[])if(['running','preparing'].includes(job.status))void loadImport(job.id);
return Object.freeze({
editor, sessions, imports, collectResources,collectImportSources,
editor, sessions, imports, collectResources,collectImportSources,inventoryLocalFiles,collectOrphanFiles,
getResourcePins:()=>resourceRetentionPins(editor.getSnapshot(),options.registry,displayPins,admissionPins(editor.getSnapshot())),
/** Read/preview under a scheduling hold. No request is rewritten or declared resolved here. */
async prepareRecovery(): Promise<RecoverySession> {
@@ -677,7 +741,7 @@ export async function openProjectEngine(options: EngineOptions) {
if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name) throw fault('invalid_staged_file');
return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size });
});
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.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.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));
try {
const handles = freeze(resources.map(resource => ({ id: resource.id, src: options.assets!.sourceURL(resource.id) })));
@@ -695,6 +759,8 @@ export async function openProjectEngine(options: EngineOptions) {
displays.add(release);
try{
const retained=editor.getSnapshot().runtime.resources.get(id);
const orphan=editor.getSnapshot().runtime.orphanFiles?.get(id);if(orphan?.retirement?.purgedAt!==undefined)throw fault('local_file_retired',[id]);
if(orphan?.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);}
if(retained?.retirement&&retained.retirement.purgedAt===undefined){
const result=editor.updateRuntime(previous=>{const resources=new Map(previous.resources),{retirement:_,...kept}=retained;resources.set(id,freeze(kept));return Object.freeze({...previous,resources:readMap(resources)});});
if(result.status==='rejected')throw fault(result.code,result.targets);
@@ -706,6 +772,8 @@ export async function openProjectEngine(options: EngineOptions) {
}
const bytes = fileWrites.get(id) ?? await options.storage.readFile(leader, id);
if (closing || closed) throw fault('engine_closing');
if(orphan&&(!bytes||bytes.size!==orphan.byteLength||bytes.type!==orphan.mimeType||await sourceDigest(bytes)!==orphan.digest))throw fault('file_inventory_changed',[id]);
if(closing||closed)throw fault('engine_closing');
const resource=editor.getSnapshot().runtime.resources.get(id);
if(!bytes&&resource?.retirement?.purgedAt!==undefined&&!resource.stored)throw fault('resource_bytes_unavailable',[id]);
if (!bytes && !options.assets) throw fault('resource_missing', [id]);
@@ -720,7 +788,7 @@ export async function openProjectEngine(options: EngineOptions) {
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()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
async close() {
if (closed) return;
closing = true;await Promise.allSettled([collection,sourceCollection].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;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();
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()]);
recovery?.controller.abort();recovery?.release();
for (const release of [...displays]) release();
+43
View File
@@ -0,0 +1,43 @@
import type {Id,Registry} from '../canvas-core/types.ts';
import {equal,fault,freeze,readMap} from '../canvas-core/values.ts';
import type {EngineSnapshot,RuntimeState} from './contracts.ts';
import {importFiles,terminalImport,validFileRetirement,type FileRetirement} from './import-file-retention.ts';
import {resourceRetentionPins} from './resource-retention.ts';
export type OrphanFile={readonly version:1;readonly id:Id;readonly byteLength:number;readonly mimeType:string;readonly digest:string;readonly retirement?:FileRetirement};
export function validateOrphanFile(value:unknown,id:Id):asserts value is OrphanFile {
const file=value as OrphanFile;
if(!file||file.version!==1||file.id!==id||!id||!Number.isSafeInteger(file.byteLength)||file.byteLength<0||typeof file.mimeType!=='string'||typeof file.digest!=='string'||!/^[a-f0-9]{64}$/.test(file.digest)||file.retirement!==undefined&&!validFileRetirement(file.retirement))throw fault('invalid_orphan_file');
}
export function orphanCoverage(snapshot:EngineSnapshot,registry:Registry,displays:ReadonlyMap<Id,number>,temporary:ReadonlyMap<Id,readonly Id[]>) {
const pins=resourceRetentionPins(snapshot,registry,displays,temporary),blockers:string[]=[];
if(!pins.complete)blockers.push('opaque_references');
if(snapshot.runtime.runs.size)blockers.push('opaque_run_records');
if(snapshot.runtime.recoveries.size)blockers.push('recovery_audits');
if([...(snapshot.runtime.imports?.values()??[])].some(job=>!terminalImport(job.status)))blockers.push('unfinished_imports');
if(temporary.has('copy-admission'))blockers.push('copy_admission');
if([...snapshot.runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'&&!['document.change','asset.upload'].includes(record.intent.operationType)))blockers.push('opaque_pending_actions');
return Object.freeze({...pins,complete:blockers.length===0,blockers:Object.freeze(blockers)});
}
/** Registered orphan identities survive repinning. Purged identities cannot be forgotten or reassigned. */
export function prepareOrphanFiles(previous:RuntimeState,next:RuntimeState,coverage:()=>{complete:boolean;pins:ReadonlyMap<Id,unknown>},reserved:ReadonlySet<Id>,referencesChanged:boolean):RuntimeState {
if(!previous.orphanFiles?.size&&!next.orphanFiles?.size)return next;
const hasSoft=[...(next.orphanFiles?.values()??[])].some(file=>file.retirement&&file.retirement.purgedAt===undefined);
if(previous.orphanFiles===next.orphanFiles&&previous.resources===next.resources&&previous.imports===next.imports&&!reserved.size&&!referencesChanged&&!hasSoft)return next;
const sources=importFiles(next.imports);
for(const [id,file]of next.orphanFiles??[]){validateOrphanFile(file,id);if(next.resources.has(id)||sources.has(id))throw fault('local_file_identity_conflict',[id]);}
for(const [id,file]of previous.orphanFiles??[]){const after=next.orphanFiles?.get(id);if(!after||file.retirement?.purgedAt!==undefined&&!equal(file,after))throw fault('local_file_retired',[id]);}
const pins=coverage(),clear:Id[]=[];
for(const [id,file]of next.orphanFiles??[]){
if(reserved.has(id)&&(!pins.complete||pins.pins.has(id)))throw fault('resource_collection_in_progress',[id]);
if(file.retirement?.purgedAt!==undefined){if(pins.pins.has(id))throw fault('local_file_retired',[id]);}
else if(file.retirement&&(!pins.complete||pins.pins.has(id)))clear.push(id);
}
if(!clear.length)return next;
const files=new Map(next.orphanFiles);for(const id of clear){const {retirement:_,...file}=files.get(id)!;files.set(id,freeze(file));}return Object.freeze({...next,orphanFiles:readMap(files)});
}
/** Read-only I/O may finish late; cancellation detaches its result without publishing it. */
export function abortableFileRead<T>(task:Promise<T>,signal:AbortSignal):Promise<T>{
return new Promise((resolve,reject)=>{const abort=()=>{signal.removeEventListener('abort',abort);reject(fault('file_inventory_aborted'));};signal.addEventListener('abort',abort,{once:true});if(signal.aborted)abort();task.then(value=>{signal.removeEventListener('abort',abort);if(!signal.aborted)resolve(value);},error=>{signal.removeEventListener('abort',abort);if(!signal.aborted)reject(error);});});
}
+2 -2
View File
@@ -52,8 +52,8 @@ export function encodeRecoveryValue(input:unknown):Json {
}
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
export function sealRecoveryImage(formatVersion:1|2|3|4,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
export function sealRecoveryImage(formatVersion:1|2|3|4|5,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
if(![1,2,3,4,5].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
const image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encodeRecoveryValue(payload)});sealed.add(image);return image;
}
export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed.has(image);}
+9 -5
View File
@@ -1,3 +1,4 @@
import {validateOrphanFile,type OrphanFile} from './orphan-files.ts';
import {importFiles} from './import-file-retention.ts';
import {resourceRetentionPins} from './resource-retention.ts';
import { sealRecoveryImage } from './recovery-encoding.ts';
@@ -22,9 +23,9 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
document: documentValue(snapshot.document),
history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: [...snapshot.history.entries] },
operations: [...snapshot.operations],
runtime: { imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
runtime: { ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:[...snapshot.runtime.orphanFiles]}:{}), imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
};
return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
return sealRecoveryImage(snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
}
type ObjectValue = Record<string, unknown>;
function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; }
@@ -87,7 +88,7 @@ function readHistory(history:ObjectValue,operations:ReadonlyMap<Id,OperationInte
return Object.freeze({undo:freeze(undo),redo:freeze(redo),entries});
}
export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSnapshot & { readonly rootRevision: number; readonly runtime: RuntimeState } {
if (![1,2,3,4].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
integer(image.rootRevision); integer(image.localRevision);
if (image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
// Own the parsed records; mutating a caller's RecoveryImage must not mutate live state.
@@ -227,13 +228,16 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
return freeze({...value as RunRecord,outputEvents:array(raw.outputEvents ?? []) as never});
});
const imports=table<ImportJob>(runtime.imports??[],(value,id)=>{const job=value as ImportJob;validateImportJob(job);if(job.id!==id||job.projectId!==document.projectId)throw fault('corrupt_recovery_image');return freeze(job);});
for(const [id,entry]of importFiles(imports)){if(resources.has(id))throw fault('import_file_identity_conflict',[id]);if((entry.file.retirement||imports.get(entry.jobId)?.source?.previousFiles?.length)&&image.formatVersion!==4)throw fault('unsupported_recovery_version');}
for(const [id,entry]of importFiles(imports)){if(resources.has(id))throw fault('import_file_identity_conflict',[id]);if((entry.file.retirement||imports.get(entry.jobId)?.source?.previousFiles?.length)&&image.formatVersion<4)throw fault('unsupported_recovery_version');}
const orphanFiles=table<OrphanFile>(runtime.orphanFiles??[],(file,id)=>{validateOrphanFile(file,id);return freeze(file);});
if(orphanFiles.size&&image.formatVersion!==5)throw fault('unsupported_recovery_version');for(const id of orphanFiles.keys())if(resources.has(id)||importFiles(imports).has(id))throw fault('local_file_identity_conflict',[id]);
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
const snapshot=Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history:restoredHistory, operations,
runtime: Object.freeze({ imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
runtime: Object.freeze({ orphanFiles, imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
});
if([...resources.values()].some(resource=>resource.retirement?.purgedAt!==undefined&&!resource.stored)){
const pins=resourceRetentionPins(snapshot,registry);for(const [id,resource]of resources)if(resource.retirement?.purgedAt!==undefined&&!resource.stored&&pins.pins.has(id))throw fault('corrupt_recovery_image');
}
if([...orphanFiles.values()].some(file=>file.retirement?.purgedAt!==undefined)){const pins=resourceRetentionPins(snapshot,registry);for(const [id,file]of orphanFiles)if(file.retirement?.purgedAt!==undefined&&pins.pins.has(id))throw fault('corrupt_recovery_image');}
return snapshot;
}
+33 -5
View File
@@ -4,10 +4,13 @@ import type { Json } from '../canvas-core/types.ts';
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise<void> };
export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string};
export type FileInventoryPage={readonly files:readonly StoredFileInfo[];readonly rootRevision:number;readonly nextAfter?:string};
export type StoragePort = {
scanFiles?(leader:StorageLeader,input:{readonly expectedRootRevision:number;readonly after?:string;readonly limit:number}):Promise<FileInventoryPage>;
readonly fileDeletion?: 'atomic-v1';
claim(scope: StorageScope): Promise<StorageLeader>;
recover(leader: StorageLeader): Promise<RecoveryImage | null>;
@@ -33,8 +36,14 @@ const keyFor = (scope: StorageScope) => {
if (![scope.accountId, scope.workspaceId, scope.projectId].every(value => typeof value === 'string' && value.length > 0)) throw fault('invalid_storage_scope');
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]);
};
function purgedFileIds(image:RecoveryImage):Set<string>{
const result=new Set<string>(),payload=image.payload as {runtime?:{resources?:[string,{retirement?:{purgedAt?:number}}][];orphanFiles?:[string,{retirement?:{purgedAt?:number}}][];imports?:[string,{source?:{fileId:string;retirement?:{purgedAt?:number};previousFiles?:{fileId:string;retirement?:{purgedAt?:number}}[]}}][]}};
for(const rows of [payload?.runtime?.resources,payload?.runtime?.orphanFiles])if(Array.isArray(rows))for(const row of rows)if(Array.isArray(row)&&typeof row[0]==='string'&&row[1]?.retirement?.purgedAt!==undefined)result.add(row[0]);
if(Array.isArray(payload?.runtime?.imports))for(const [,job]of payload.runtime.imports){if(!job?.source)continue;for(const file of [job.source,...(job.source.previousFiles??[])])if(file.retirement?.purgedAt!==undefined)result.add(file.fileId);}
return result;
}
function validateImage(image: RecoveryImage) {
if (![1,2,3,4].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (![1,2,3,4,5].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
}
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });
@@ -134,18 +143,20 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
// Detach mutable callers and descriptors before the first await, including queued commits.
const image=isSealedRecoveryImage(batch.image)?batch.image:freeze(clone(batch.image)),files=(batch.files??[]).map(file=>({id:file.id,bytes:file.bytes})),deleteFiles=[...(batch.deleteFiles??[])];if(deleteFiles.length>128||new Set(deleteFiles).size!==deleteFiles.length||deleteFiles.some(id=>typeof id!=='string'||!id||files.some(file=>file.id===id)))throw fault('invalid_file_deletion');if(!isSealedRecoveryImage(image))validateImage(image);
if(deleteFiles.length){
type FileProof={fileId?:string;retirement?:{version:number;unreferencedAt:number;purgedAt?:number}};
const payload=image.payload as {runtime?:{resources?:[string,FileProof][];imports?:[string,{status:string;source?:FileProof&{previousFiles?:FileProof[]}}][]}};
type FileProof={id?:string;version?:number;byteLength?:number;mimeType?:string;digest?:string;fileId?:string;retirement?:{version:number;unreferencedAt:number;purgedAt?:number}};
const payload=image.payload as {runtime?:{orphanFiles?:[string,FileProof][];resources?:[string,FileProof][];imports?:[string,{status:string;source?:FileProof&{previousFiles?:FileProof[]}}][]}};
const proofs=new Map<string,FileProof>(),owned=new Set<string>();
const own=(id:string,file:FileProof,eligible:boolean)=>{if(typeof id!=='string'||!id||owned.has(id))throw fault('invalid_file_deletion');owned.add(id);if(eligible)proofs.set(id,file);};
for(const [id,resource]of payload?.runtime?.resources??[])own(id,resource,true);
if(image.formatVersion===4)for(const [,job]of payload?.runtime?.imports??[]){if(job.source){own(job.source.fileId!,job.source,['completed','partial','cancelled'].includes(job.status));for(const file of job.source.previousFiles??[])own(file.fileId!,file,true);}}
if(image.formatVersion>=4)for(const [,job]of payload?.runtime?.imports??[]){if(job.source){own(job.source.fileId!,job.source,['completed','partial','cancelled'].includes(job.status));for(const file of job.source.previousFiles??[])own(file.fileId!,file,true);}}
if(image.formatVersion===5)for(const [id,file]of payload?.runtime?.orphanFiles??[]){if(file.version!==1||file.id!==id||!Number.isSafeInteger(file.byteLength)||file.byteLength!<0||typeof file.mimeType!=='string'||typeof file.digest!=='string'||!/^[a-f0-9]{64}$/.test(file.digest))throw fault('invalid_file_deletion');own(id,file,true);}
if(image.formatVersion<3||deleteFiles.some(id=>{const marker=proofs.get(id)?.retirement;return !marker||marker.version!==1||!Number.isSafeInteger(marker.unreferencedAt)||marker.unreferencedAt<0||!Number.isSafeInteger(marker.purgedAt)||marker.purgedAt!<marker.unreferencedAt;}))throw fault('invalid_file_deletion');
}
check(leader);
const task=(writeTails.get(leader)??Promise.resolve()).catch(()=>{}).then(async()=>{
let base=lastCuts.get(leader)??await recoverCut(leader);
if(base?.image.rootRevision===image.rootRevision)base=await recoverCut(leader);
if(files.length&&base){const retired=purgedFileIds(base.image);if(files.some(file=>retired.has(file.id)))throw fault('resource_identity_conflict');}
const digest=await imageDigest(image),stored:StoredImage={image,checksum:digest.checksum};let candidate:StoredCut=stored;
if(base&&image.rootRevision===base.image.rootRevision+1){
const patches=diffJournal(base.image.payload,image.payload);
@@ -171,6 +182,23 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
});
writeTails.set(leader,task);return task;
},
scanFiles(leader:StorageLeader,input:{expectedRootRevision:number;after?:string;limit:number}) {
const {expectedRootRevision,after,limit}=input;
if(!Number.isSafeInteger(expectedRootRevision)||expectedRootRevision<0||!Number.isSafeInteger(limit)||limit<1||limit>128||after!==undefined&&(typeof after!=='string'||!after))return Promise.reject(fault('invalid_file_inventory'));
return execute(leader,['files'],'readonly',async(transaction,metadata,key)=>{
if(metadata.throughRevision!==expectedRootRevision)throw fault('stale_file_inventory');
const rows=await new Promise<StoredFileInfo[]>((resolve,reject)=>{
const files:StoredFileInfo[]=[],range=IDBKeyRange.bound(after===undefined?[key]:[key,after],[key,[]],after!==undefined,true),request=transaction.objectStore('files').openCursor(range);
request.onerror=()=>reject(request.error);
request.onsuccess=()=>{const cursor=request.result;if(!cursor){resolve(files);return;}const parts=cursor.key,bytes=cursor.value;
if(!Array.isArray(parts)||parts.length!==2||parts[0]!==key||typeof parts[1]!=='string'||!parts[1]||!(bytes instanceof Blob)){reject(fault('corrupt_file_store'));return;}
files.push(Object.freeze({id:parts[1],byteLength:bytes.size,mimeType:bytes.type}));if(files.length>limit)resolve(files);else cursor.continue();
};
});
const more=rows.length>limit,files=more?rows.slice(0,limit):rows;
return Object.freeze({files:Object.freeze(files),rootRevision:expectedRootRevision,...(more?{nextAfter:files.at(-1)!.id}:{})});
});
},
readFile(leader: StorageLeader, id: string) { return execute(leader, ['files'], 'readonly', (transaction, _metadata, key) => requestValue<Blob | undefined>(transaction.objectStore('files').get([key, id]))); },
close() { closed = true; for (const handle of handles) void handle.release(); void opening.then(database => database.close()).catch(() => {}); },
});