Collect unreferenced local resource bytes atomically with recovery cuts
This commit is contained in:
@@ -21,6 +21,8 @@ export type EditorOptions<Runtime = undefined> = {
|
||||
reduceRuntime?: (previous: Runtime, next: CoreSnapshot, commit: Commit) => Runtime;
|
||||
/** Engine may attach transport semantics without changing identity, order or document changes. */
|
||||
prepareIntent?: (intent: OperationIntent, runtime: Runtime) => OperationIntent;
|
||||
/** Pure coordinator hook for all publication paths; may replace runtime only or reject. */
|
||||
prepareRuntime?: (previous: Snapshot<Runtime>, next: Snapshot<Runtime>) => Runtime;
|
||||
nextOperationSequence?: (runtime: Runtime, fallback: number) => number;
|
||||
};
|
||||
export type Transaction = { dispatch(command: Command): Availability; getDocument(): Document };
|
||||
@@ -119,6 +121,12 @@ export function createEditor<Runtime = undefined>(options: EditorOptions<Runtime
|
||||
if (disposed) throw fault('editor_disposed');
|
||||
if (entered) throw fault('reentrant_dispatch');
|
||||
}
|
||||
function prepareState(next: Snapshot<Runtime>): Snapshot<Runtime> {
|
||||
if(!options.prepareRuntime)return next;
|
||||
const runtime=options.prepareRuntime(state,next);
|
||||
if(isThenable(runtime))throw fault('async_transaction');
|
||||
return runtime===next.runtime?next:Object.freeze({...next,runtime});
|
||||
}
|
||||
function emit(commit: Commit) {
|
||||
for (const listener of [...listeners]) try { listener(commit, state); } catch (error) { try { options.onListenerError?.(error); } catch { /* Observers cannot roll back a committed state. */ } }
|
||||
}
|
||||
@@ -181,7 +189,7 @@ export function createEditor<Runtime = undefined>(options: EditorOptions<Runtime
|
||||
const commit = freeze({ revision: next.revision, origin, changes, operationIds });
|
||||
const runtime = options.reduceRuntime ? options.reduceRuntime(state.runtime, next, commit) : state.runtime;
|
||||
if (isThenable(runtime)) throw fault('async_transaction');
|
||||
const nextState=Object.freeze({ ...next, rootRevision: state.rootRevision + 1, runtime }),nextQueries=advanceDocumentQueries(queries,document,options.registry,next.revision,changes);
|
||||
const nextState=prepareState(Object.freeze({ ...next, rootRevision: state.rootRevision + 1, runtime })),nextQueries=advanceDocumentQueries(queries,document,options.registry,next.revision,changes);
|
||||
state=nextState;queries=nextQueries;queriesByDocument.set(document,queries);
|
||||
if (outbound) sequence = nextSequence;
|
||||
emit(commit);
|
||||
@@ -270,7 +278,7 @@ export function createEditor<Runtime = undefined>(options: EditorOptions<Runtime
|
||||
const runtime = reduce(state.runtime);
|
||||
if (isThenable(runtime)) throw fault('async_transaction');
|
||||
if (runtime !== state.runtime) {
|
||||
state = Object.freeze({ ...state, rootRevision: state.rootRevision + 1, runtime });
|
||||
state = prepareState(Object.freeze({ ...state, rootRevision: state.rootRevision + 1, runtime }));
|
||||
emit(freeze({ revision: state.revision, origin: 'ack', changes: [], operationIds: [] }));
|
||||
}
|
||||
return Object.freeze({ status: 'applied', revision: state.revision, operationIds: Object.freeze([]) });
|
||||
@@ -295,7 +303,7 @@ export function createEditor<Runtime = undefined>(options: EditorOptions<Runtime
|
||||
for(const [id,entry]of history.entries)if(entry.id!==id||entry.operationIds.some(operationId=>!operations.has(operationId)))throw fault('invalid_history');
|
||||
}
|
||||
const changed = !documentsEqual(next, state.document);
|
||||
const nextState=Object.freeze({ ...state, ...(replacement?{history:replacement.history,operations:replacement.operations}:{}), document: changed ? next : state.document, runtime, revision: state.revision + Number(changed), rootRevision: state.rootRevision + 1 }),nextQueries=changed?advanceDocumentQueries(queries,next,options.registry,state.revision+1):queries;
|
||||
const nextState=prepareState(Object.freeze({ ...state, ...(replacement?{history:replacement.history,operations:replacement.operations}:{}), document: changed ? next : state.document, runtime, revision: state.revision + Number(changed), rootRevision: state.rootRevision + 1 })),nextQueries=changed?advanceDocumentQueries(queries,next,options.registry,state.revision+1):queries;
|
||||
state=nextState;queries=nextQueries;if(changed)queriesByDocument.set(next,queries);
|
||||
sequence=nextSequence;
|
||||
emit(freeze({ revision: state.revision, origin: 'ack', changes: [], operationIds: [] }));
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {resourceRetentionPins,prepareResourceRetention} from './resource-retention.ts';
|
||||
import { encodeIndexedCopySnapshot, openIndexedCopySnapshot, type IndexedCopySnapshot } from './copy-snapshot-index.ts';
|
||||
import { decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
|
||||
import { createDuplicatePlanSource } from './duplicate-source.ts';
|
||||
@@ -61,6 +62,8 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const network = createNetworkGate(2);
|
||||
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 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));
|
||||
let collection:Promise<{readonly markedIds:readonly Id[];readonly purgedIds:readonly Id[];readonly complete:boolean;readonly checkedRootRevision:number;readonly rootRevision:number;readonly durableRootRevision:number}>|undefined;
|
||||
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
let applying: RunApplyInput | undefined;
|
||||
@@ -69,12 +72,14 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const copyPreparations=new Map<AbortController,Promise<Id>>();
|
||||
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)=>copyPreparations.size?new Map([['copy-admission',[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt===undefined).map(resource=>resource.id)]]):new Map<Id,readonly Id[]>();
|
||||
const assertWritable = () => { if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); };
|
||||
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: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
|
||||
const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,queryBudget:options.queryBudget,
|
||||
policy(command, document) { if (closing || closed || recovery) return { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; },
|
||||
traversalPolicy(entry, direction) { return closing || closed || recovery ? { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; },
|
||||
onListenerError: report,
|
||||
prepareRuntime:(_previous,next)=>prepareResourceRetention(next,options.registry,displayPins,collectionReservations,admissionPins(next)),
|
||||
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; },
|
||||
reduceRuntime(previous, next, commit) {
|
||||
@@ -113,9 +118,11 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
while (pending.size) {
|
||||
const [revision, snapshot] = pending.entries().next().value!;
|
||||
const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.fileId===id)).map(([id, bytes]) => ({ id, bytes }));
|
||||
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot), files });
|
||||
const deleteFiles=[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.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);
|
||||
for(const id of deleteFiles){durablyPurged.add(id);collectionReservations.delete(id);}
|
||||
for (const file of files) fileWrites.delete(file.id);
|
||||
if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation));
|
||||
}
|
||||
@@ -124,6 +131,39 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
writing = task.catch(error => { persistenceFailed = true; throw error; }).finally(() => { writing = undefined; });
|
||||
return writing;
|
||||
}
|
||||
function collectResources(settings:{retentionMs:number;maxFiles?:number}) {
|
||||
assertWritable();
|
||||
if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable');
|
||||
const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64;
|
||||
if(!Number.isSafeInteger(retentionMs)||retentionMs<0||!Number.isSafeInteger(maxFiles)||maxFiles<1||maxFiles>128)throw fault('invalid_resource_collection');
|
||||
if(collection)return collection;
|
||||
const task=(async()=>{
|
||||
await persist();assertWritable();
|
||||
const now=runClock.now();if(!Number.isSafeInteger(now)||now<0)throw fault('invalid_resource_collection_clock');
|
||||
let root=editor.getSnapshot(),pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root));
|
||||
const markedIds:Id[]=[],purgedIds:Id[]=[];
|
||||
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);
|
||||
if(result.status==='rejected')throw fault(result.code,result.targets);
|
||||
await persist();assertWritable();
|
||||
}
|
||||
root=editor.getSnapshot();pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root));
|
||||
if(!pins.complete)return report();
|
||||
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){
|
||||
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);}
|
||||
// Persistence derives byte deletions from this immutable root, even if a listener already started it.
|
||||
await persist();for(const resource of eligible)purgedIds.push(resource.id);
|
||||
}
|
||||
return report();
|
||||
})();
|
||||
collection=task.finally(()=>{collection=undefined;});return collection;
|
||||
}
|
||||
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);
|
||||
@@ -413,7 +453,12 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;
|
||||
},
|
||||
async duplicate(selection:readonly Target[],settings:{readonly worldOffset?:Point;readonly signal?:AbortSignal;readonly viewCanvasId?:Id}={}):Promise<Id>{
|
||||
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.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();
|
||||
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);
|
||||
}
|
||||
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}));
|
||||
for(const selected of targets){const element=captured.document.elements.get(selected.id);if(!element||element.incarnation!==selected.incarnation)throw fault('stale_incarnation',[selected.id]);}if(![offset.x,offset.y].every(Number.isFinite))throw fault('invalid_copy_offset');
|
||||
signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)controller.abort();
|
||||
@@ -437,7 +482,8 @@ 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,
|
||||
editor, sessions, imports, collectResources,
|
||||
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> {
|
||||
assertWritable();
|
||||
@@ -587,13 +633,25 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
},
|
||||
async displayResource(id: Id) {
|
||||
if (closing || closed) throw fault('engine_closing');
|
||||
const bytes = fileWrites.get(id) ?? await options.storage.readFile(leader, id);
|
||||
if (closing || closed) throw fault('engine_closing');
|
||||
if (!bytes && !options.assets) throw fault('resource_missing', [id]);
|
||||
const url = bytes ? URL.createObjectURL(bytes) : options.assets!.displayURL(id);
|
||||
let released = false;
|
||||
const release = () => { if (!released) { released = true; displays.delete(release); if (bytes) URL.revokeObjectURL(url); } };
|
||||
displays.add(release); return Object.freeze({ url, release });
|
||||
if(collectionReservations.has(id))throw fault('resource_collection_in_progress',[id]);
|
||||
displayPins.set(id,(displayPins.get(id)??0)+1);
|
||||
let released=false,url:string|undefined,local=false;
|
||||
const release=()=>{if(released)return;released=true;displays.delete(release);const count=displayPins.get(id)??1;if(count<=1)displayPins.delete(id);else displayPins.set(id,count-1);if(local&&url)URL.revokeObjectURL(url);};
|
||||
displays.add(release);
|
||||
try{
|
||||
const retained=editor.getSnapshot().runtime.resources.get(id);
|
||||
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);
|
||||
}
|
||||
const bytes = fileWrites.get(id) ?? await options.storage.readFile(leader, 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]);
|
||||
local=!!bytes;url=bytes?URL.createObjectURL(bytes):options.assets!.displayURL(id);
|
||||
return Object.freeze({url,release});
|
||||
}catch(error){release();throw error;}
|
||||
},
|
||||
getSnapshot: editor.getSnapshot, persist,
|
||||
cancelRecovery:async()=>{await recovery?.cancel();},
|
||||
@@ -602,7 +660,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;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;await collection?.catch(report);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();
|
||||
|
||||
@@ -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,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
|
||||
if(![1,2].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,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
|
||||
if(![1,2,3].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);}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {resourceRetentionPins} from './resource-retention.ts';
|
||||
import { sealRecoveryImage } from './recovery-encoding.ts';
|
||||
import { validateImportJob, type ImportJob } from './imports.ts';
|
||||
import { validateElisions, withLocalLifecycles } from './elision.ts';
|
||||
@@ -22,7 +23,7 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
|
||||
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] } },
|
||||
};
|
||||
return sealRecoveryImage(snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
|
||||
return sealRecoveryImage([...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; }
|
||||
@@ -85,7 +86,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].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||||
if (![1,2,3].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.
|
||||
@@ -162,6 +163,7 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
const raw = object(value);
|
||||
if (string(raw.id) !== id || string(raw.operationId) === id) throw fault('corrupt_recovery_image');
|
||||
string(raw.name); string(raw.mimeType); integer(raw.byteLength, 1);
|
||||
if(raw.retirement!==undefined){if(image.formatVersion!==3)throw fault('unsupported_recovery_version');const marker=object(raw.retirement);if(marker.version!==1)throw fault('resource_retirement_interpreter_unavailable');integer(marker.unreferencedAt);if(marker.purgedAt!==undefined&&integer(marker.purgedAt)<(marker.unreferencedAt as number))throw fault('corrupt_recovery_image');}
|
||||
if (raw.digest !== undefined && !/^[a-f0-9]{64}$/.test(string(raw.digest))) throw fault('corrupt_recovery_image');
|
||||
const operation = records.get(raw.operationId as string)?.intent;
|
||||
if (operation?.operationType !== 'asset.upload' || operation.resourceId !== id) throw fault('corrupt_recovery_image');
|
||||
@@ -225,7 +227,11 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
});
|
||||
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);});
|
||||
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
|
||||
return Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history:restoredHistory, operations,
|
||||
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) }),
|
||||
});
|
||||
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');
|
||||
}
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import type {Id,Registry} from '../canvas-core/types.ts';
|
||||
import {fault,readMap,readSet,freeze} from '../canvas-core/values.ts';
|
||||
import type {EngineSnapshot,RuntimeState} from './contracts.ts';
|
||||
import {resourcePins} from './resources.ts';
|
||||
|
||||
/** Complete local-byte protection for one project scope; no DOM/viewport filtering. */
|
||||
export function resourceRetentionPins(snapshot:EngineSnapshot,registry:Registry,displays:ReadonlyMap<Id,number>=new Map(),temporary:ReadonlyMap<Id,readonly Id[]>=new Map()) {
|
||||
const {runtime}=snapshot;
|
||||
const current=resourcePins(snapshot,runtime.ledger,registry,runtime.resourcePins);
|
||||
const base=resourcePins({...snapshot,document:runtime.acknowledged},runtime.ledger,registry,new Map());
|
||||
const pins=new Map<Id,Set<string>>();
|
||||
const add=(id:Id,owner:string)=>{let owners=pins.get(id);if(!owners){owners=new Set();pins.set(id,owners);}owners.add(owner);};
|
||||
for(const [id,owners]of current.pins)for(const owner of owners)add(id,owner);
|
||||
for(const [id,owners]of base.pins)for(const owner of owners)if(owner.startsWith('element:'))add(id,`acknowledged:${owner}`);
|
||||
// Sources and server run envelopes are not all interpreted resource schemas yet.
|
||||
// Pin the whole known local depot rather than claiming opaque task data contains no references.
|
||||
for(const job of runtime.imports?.values()??[])if(!['completed','partial','cancelled'].includes(job.status))for(const [id,resource]of runtime.resources)if(resource.retirement?.purgedAt===undefined)add(id,`import:${job.id}`);
|
||||
for(const run of runtime.runs.values())for(const [id,resource]of runtime.resources)if(resource.retirement?.purgedAt===undefined)add(id,`run:${run.id}`);
|
||||
for(const [owner,ids]of temporary)for(const id of ids)add(id,owner);
|
||||
for(const [id,count]of displays)if(count>0)add(id,'display');
|
||||
return Object.freeze({complete:current.complete&&base.complete,pins:readMap(new Map([...pins].map(([id,owners])=>[id,readSet(owners)])))});
|
||||
}
|
||||
|
||||
/** Runs inside the editor's single publication boundary, including reconcile and metadata commits. */
|
||||
export function prepareResourceRetention(next:EngineSnapshot,registry:Registry,displays:ReadonlyMap<Id,number>,reserved:ReadonlySet<Id>,temporary:ReadonlyMap<Id,readonly Id[]>=new Map()):RuntimeState {
|
||||
if(!reserved.size&&![...next.runtime.resources.values()].some(resource=>resource.retirement&&(resource.retirement.purgedAt===undefined||!resource.stored)))return next.runtime;
|
||||
const pins=resourceRetentionPins(next,registry,displays,temporary);
|
||||
if(reserved.size&&(!pins.complete||[...reserved].some(id=>pins.pins.has(id))))throw fault('resource_collection_in_progress',[...reserved]);
|
||||
let resources=next.runtime.resources;
|
||||
for(const [id,resource]of next.runtime.resources){
|
||||
if(!resource.retirement)continue;
|
||||
if(resource.retirement.purgedAt!==undefined){
|
||||
if(!resource.stored&&pins.pins.has(id))throw fault('resource_bytes_unavailable',[id]);
|
||||
}else if(!pins.complete||pins.pins.has(id)){
|
||||
if(resources===next.runtime.resources)resources=new Map(resources);
|
||||
const {retirement:_,...kept}=resource;(resources as Map<Id,typeof resource>).set(id,freeze(kept));
|
||||
}
|
||||
}
|
||||
return resources===next.runtime.resources?next.runtime:Object.freeze({...next.runtime,resources:readMap(resources)});
|
||||
}
|
||||
@@ -6,7 +6,7 @@ import type { OperationStatus } from './scheduler.ts';
|
||||
import { readMap, readSet } from '../canvas-core/values.ts';
|
||||
|
||||
export type StoredAsset = { readonly id: Id; readonly url: string; readonly width: number; readonly height: number; readonly bytes: number; readonly mime_type: string };
|
||||
export type ResourceRecord = { readonly id: Id; readonly operationId: Id; readonly name: string; readonly mimeType: string; readonly byteLength: number; readonly digest?: string; readonly stored?: StoredAsset };
|
||||
export type ResourceRecord = { readonly id: Id; readonly operationId: Id; readonly name: string; readonly mimeType: string; readonly byteLength: number; readonly digest?: string; readonly retirement?: {readonly version:1;readonly unreferencedAt:number;readonly purgedAt?:number}; readonly stored?: StoredAsset };
|
||||
export type DisplayHandle = { readonly url: string; release(): void };
|
||||
export type AssetPort = {
|
||||
sourceURL(resourceId: Id): string;
|
||||
@@ -77,10 +77,24 @@ export function resourcePins(snapshot: CoreSnapshot, ledger: LedgerSnapshot, reg
|
||||
if (!record) return;
|
||||
const component = registry.components.get(record.componentId);
|
||||
if (registry.resolve(record) !== 'known' || component?.resourceFields === undefined) { complete = false; return; }
|
||||
for (const field of component.resourceFields) { const ref = record.data[field]; if (typeof ref === 'string') pin(ref, owner); }
|
||||
for (const field of component.resourceFields) { const ref = record.data[field]; if (typeof ref === 'string') pin(ref, owner);else if(ref!==undefined&&ref!==null)complete=false; }
|
||||
};
|
||||
const schemas=new Map<Id,ElementRecord[]>();
|
||||
const remember=(record:ElementRecord|null)=>{if(record){const list=schemas.get(record.id)??[];if(!list.some(item=>item.componentId===record.componentId&&item.dataVersion===record.dataVersion))list.push(record);schemas.set(record.id,list);}};
|
||||
for(const record of snapshot.document.elements.values())remember(record);
|
||||
const rememberChanges=(changes:readonly Change[])=>{for(const change of changes)if(change.kind==='element'){remember(change.before);remember(change.after);}};
|
||||
for(const entry of snapshot.history.entries.values())rememberChanges(entry.changes);
|
||||
for(const record of ledger.operations.values()){rememberChanges(record.intent.changes);if(record.intent.historyTransition?.version===1)rememberChanges(historyTransitionChanges(record.intent.historyTransition));}
|
||||
const changes = (values: readonly Change[], owner: string) => {
|
||||
for (const change of values) if (change.kind === 'element') { element(change.before, owner); element(change.after, owner); }
|
||||
for(const change of values){
|
||||
if(change.kind==='element'){element(change.before,owner);element(change.after,owner);}
|
||||
else if(change.kind==='field'&&change.field.startsWith('data.')){
|
||||
const candidates=schemas.get(change.target.id);if(!candidates?.length)complete=false;
|
||||
for(const record of candidates??[]){const component=registry.components.get(record.componentId);if(registry.resolve(record)!=='known'||component?.resourceFields===undefined){complete=false;continue;}
|
||||
if(component.resourceFields.includes(change.field.slice(5)))for(const value of [change.before,change.after])if(typeof value==='string')pin(value,owner);else if(value!==undefined&&value!==null)complete=false;
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const id of referencedResources(values, registry)) pin(id, owner);
|
||||
};
|
||||
for (const record of snapshot.document.elements.values()) element(record, `element:${record.id}:${record.incarnation}`);
|
||||
|
||||
@@ -4,10 +4,11 @@ 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; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
|
||||
export type DurableBatch = { readonly image: RecoveryImage; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] };
|
||||
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3; 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 StoragePort = {
|
||||
readonly fileDeletion?: 'atomic-v1';
|
||||
claim(scope: StorageScope): Promise<StorageLeader>;
|
||||
recover(leader: StorageLeader): Promise<RecoveryImage | null>;
|
||||
commit(leader: StorageLeader, batch: DurableBatch): Promise<{ readonly durableRootRevision: number }>;
|
||||
@@ -33,7 +34,7 @@ const keyFor = (scope: StorageScope) => {
|
||||
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]);
|
||||
};
|
||||
function validateImage(image: RecoveryImage) {
|
||||
if (![1,2].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||||
if (![1,2,3].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); });
|
||||
@@ -102,6 +103,7 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
check(leader);if(previous&&(!lastCuts.has(leader)||lastCuts.get(leader)!.image.rootRevision<=previous.image.rootRevision))lastCuts.set(leader,previous);return previous;
|
||||
}
|
||||
return Object.freeze({
|
||||
fileDeletion:'atomic-v1' as const,
|
||||
async claim(scope: StorageScope) {
|
||||
const key = keyFor(scope);
|
||||
if (closed) throw fault('storage_closed');
|
||||
@@ -130,7 +132,13 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
async recover(leader: StorageLeader) {const cut=await recoverCut(leader);return cut?clone(cut.image):null;},
|
||||
async commit(leader: StorageLeader, batch: DurableBatch) {
|
||||
// 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}));if(!isSealedRecoveryImage(image))validateImage(image);check(leader);
|
||||
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){
|
||||
const payload=image.payload as {runtime?:{resources?:[string,{retirement?:{version:number;unreferencedAt:number;purgedAt?:number}}][]}};
|
||||
const rows=payload?.runtime?.resources;
|
||||
if(image.formatVersion!==3||!Array.isArray(rows)||deleteFiles.some(id=>!rows.some(([key,resource])=>key===id&&resource.retirement?.version===1&&Number.isSafeInteger(resource.retirement.unreferencedAt)&&resource.retirement.unreferencedAt>=0&&Number.isSafeInteger(resource.retirement.purgedAt)&&resource.retirement.purgedAt!>=resource.retirement.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);
|
||||
@@ -144,11 +152,12 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
const saved=previous===undefined?undefined:await requestValue<StoredCut>(transaction.objectStore(metadata.checkpointRevision===previous?'checkpoints':'journal').get([key,previous]));
|
||||
if(previous!==undefined&&image.rootRevision!==previous+1){
|
||||
if(image.rootRevision!==previous)throw fault('storage_revision_gap');
|
||||
if(!saved||saved.checksum!==stored.checksum||!base||!equal(base.image,image)||files.length)throw fault('storage_revision_conflict');return{durableRootRevision:previous};
|
||||
if(!saved||saved.checksum!==stored.checksum||!base||!equal(base.image,image)||files.length)throw fault('storage_revision_conflict');for(const id of deleteFiles)if(await requestValue(transaction.objectStore('files').getKey([key,id])))throw fault('storage_revision_conflict');return{durableRootRevision:previous};
|
||||
}
|
||||
if(previous!==undefined&&(!saved||image.localRevision<cutHeader(saved).localRevision))throw fault('storage_revision_conflict');
|
||||
if(!('image' in candidate)&&(candidate.baseRevision!==previous||candidate.baseChecksum!==saved?.checksum))throw fault('recovery_delta_base_mismatch');
|
||||
for(const file of files){if(!file.id||!(file.bytes instanceof Blob))throw fault('invalid_staged_file');if(await requestValue(transaction.objectStore('files').getKey([key,file.id])))throw fault('resource_identity_conflict',[file.id]);transaction.objectStore('files').put(file.bytes,[key,file.id]);}
|
||||
for(const id of deleteFiles)transaction.objectStore('files').delete([key,id]);
|
||||
if(previous===undefined||image.rootRevision-(metadata.checkpointRevision??0)>=checkpointEvery){
|
||||
transaction.objectStore('checkpoints').put(stored,[key,image.rootRevision]);if(metadata.checkpointRevision!==undefined)transaction.objectStore('checkpoints').delete([key,metadata.checkpointRevision]);transaction.objectStore('journal').delete(IDBKeyRange.bound([key,0],[key,image.rootRevision]));metadata.checkpointRevision=image.rootRevision;
|
||||
}else transaction.objectStore('journal').put(candidate,[key,image.rootRevision]);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {decodeRecovery} from '../../src/lib/canvas-runtime/recovery.ts';
|
||||
import {createBrowserDigests} from '../../src/lib/canvas-runtime/browser-digests.ts';
|
||||
import {sealRecoveryImage} from '../../src/lib/canvas-runtime/recovery-encoding.ts';
|
||||
import {planDuplicateAsync} from '../../src/lib/canvas-runtime/duplicate.ts';
|
||||
@@ -284,6 +285,23 @@ async function runRecoveryCase(name,assert){
|
||||
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(`${name}-runs`);}
|
||||
}
|
||||
|
||||
async function atomicResourceDeletionCase(name,assert){
|
||||
const database=`${name}-file-collection`,storage=createBrowserStorage({name:database,checkpointEvery:2}),scope={accountId:'account',workspaceId:'workspace',projectId:'project'};let leader,other;
|
||||
const marker={version:1,unreferencedAt:100,purgedAt:200},image={formatVersion:3,rootRevision:1,localRevision:0,payload:{runtime:{resources:[['asset',{retirement:marker}]]}}};
|
||||
try{
|
||||
other=await storage.claim({...scope,projectId:'other-project'});await storage.commit(other,{image:{formatVersion:1,rootRevision:0,localRevision:0,payload:{}},files:[{id:'asset',bytes:new Blob(['other project'])}]});
|
||||
leader=await storage.claim(scope);await storage.commit(leader,{image:{formatVersion:1,rootRevision:0,localRevision:0,payload:{}},files:[{id:'asset',bytes:new Blob(['delete me'])},{id:'keep',bytes:new Blob(['keep me'])}]});
|
||||
const nativeDelete=IDBObjectStore.prototype.delete;let queued=false;
|
||||
IDBObjectStore.prototype.delete=function(key){const request=nativeDelete.call(this,key);if(this.name==='files'&&Array.isArray(key)&&key[1]==='asset'){queued=true;throw Error('fault_after_delete');}return request;};
|
||||
let failure;try{await storage.commit(leader,{image,deleteFiles:['asset']});}catch(error){failure=error.message;}finally{IDBObjectStore.prototype.delete=nativeDelete;}
|
||||
assert(queued&&failure==='fault_after_delete','fault did not occur after a real IDB delete request');assert(await(await storage.readFile(leader,'asset')).text()==='delete me','aborted GC transaction lost bytes');assert((await storage.recover(leader)).rootRevision===0,'aborted GC transaction advanced recovery');
|
||||
const ids=['asset'],pending=storage.commit(leader,{image,deleteFiles:ids});ids[0]='keep';await pending;
|
||||
assert(await storage.readFile(leader,'asset')===undefined&&await(await storage.readFile(leader,'keep')).text()==='keep me','mutable deletion descriptor escaped capture');assert((await storage.recover(leader)).formatVersion===3,'GC cut did not persist');await storage.commit(leader,{image,deleteFiles:['asset']});
|
||||
let code;try{await storage.commit(leader,{image:{...image,rootRevision:2},deleteFiles:['keep']});}catch(error){code=error.code;}assert(code==='invalid_file_deletion','file deletion without a matching purge marker was accepted');
|
||||
await leader.release();leader=await storage.claim(scope);assert(await storage.readFile(leader,'asset')===undefined&&(await storage.recover(leader)).rootRevision===1,'GC did not survive owner reopening');assert(await(await storage.readFile(other,'asset')).text()==='other project','local GC crossed its project scope');
|
||||
}finally{await other?.release();await leader?.release();storage.close();indexedDB.deleteDatabase(database);}
|
||||
}
|
||||
|
||||
async function resourceRecoveryCase(name, assert) {
|
||||
const scope = { accountId: crypto.randomUUID(), workspaceId: crypto.randomUUID(), projectId: crypto.randomUUID() };
|
||||
const registry = createRegistry(builtinComponents()), document = createDocument(scope.projectId, [{ id: scope.projectId, parent: null }], []);
|
||||
@@ -304,7 +322,7 @@ async function resourceRecoveryCase(name, assert) {
|
||||
async query() { return { outcome: 'unresolved' }; },
|
||||
canonical: () => ({ document: server, versions: new Map([[scope.projectId, writes.length + 1]]) }),
|
||||
};
|
||||
const options = { scope, registry, document, versions: new Map([[scope.projectId, 1]]), storage, persistence, assets };
|
||||
const options = { scope, registry, document, versions: new Map([[scope.projectId, 1]]), storage, persistence, assets,historyLimit:1 };
|
||||
let engine, resource;
|
||||
try {
|
||||
engine = await openProjectEngine(options);
|
||||
@@ -326,6 +344,13 @@ async function resourceRecoveryCase(name, assert) {
|
||||
assert(queries[0].path.endsWith(uploads[0].requestId), 'recovery queried a replacement request identity');
|
||||
assert(engine.getSnapshot().runtime.resources.get(resource.id).stored.id === resource.id, 'confirmed asset changed resource identity');
|
||||
assert(engine.getSnapshot().history.undo.length === 1 && engine.getSnapshot().document.elements.size === 1, 'resource ACK recreated the node/history');
|
||||
assert((await engine.collectResources({retentionMs:0})).purgedIds.length===0,'live image was collected');
|
||||
engine.editor.dispatch({type:'element.remove',target:{id:'picture',incarnation:1}});await engine.settle();assert((await engine.collectResources({retentionMs:0})).purgedIds.length===0,'undo-owned image was collected');
|
||||
engine.editor.dispatch({type:'element.create',element:{id:'keep-text',incarnation:1,componentId:'text',dataVersion:1,structure:'node',ownerCanvasId:scope.projectId,data:{title:'keep',text:''},geometry:{x:0,y:0,width:100,height:100,rotation:0}}});await engine.settle();
|
||||
const collected=await engine.collectResources({retentionMs:0});assert(collected.purgedIds.length===1&&collected.purgedIds[0]===resource.id,'unreferenced local image was not collected');await engine.close();
|
||||
const inspected=await storage.claim(scope),cut=await storage.recover(inspected);assert(cut.formatVersion===3&&await storage.readFile(inspected,resource.id)===undefined,'recovered GC marker and byte deletion disagree');assert(decodeRecovery(cut,registry).runtime.resources.get(resource.id).retirement.purgedAt!==undefined,'GC marker did not decode');await inspected.release();
|
||||
engine=await openProjectEngine(options);const remote=await engine.displayResource(resource.id);assert(remote.url===assets.displayURL(resource.id),'purged confirmed image did not fall back to its existing remote asset');remote.release();assert(uploads.length===1,'reopening collected bytes repeated upload');
|
||||
|
||||
} finally { await engine?.close(); storage.close(); indexedDB.deleteDatabase(`${name}-resources`); }
|
||||
}
|
||||
|
||||
@@ -567,7 +592,7 @@ export async function run() {
|
||||
await storage.commit(leader, { image: image(1) });
|
||||
await rejects(() => storage.commit(leader, { image: { ...image(1), payload: null } }), 'storage_revision_conflict');
|
||||
await rejects(() => storage.commit(leader, { image: image(3) }), 'storage_revision_gap');
|
||||
await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 3 } }), 'unsupported_recovery_version');
|
||||
await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 4 } }), 'unsupported_recovery_version');
|
||||
results.push('same-cut retry is idempotent; conflicting, skipped and future-format cuts reject');
|
||||
|
||||
// The first put succeeds before a later duplicate resource aborts the whole transaction.
|
||||
@@ -608,6 +633,8 @@ export async function run() {
|
||||
results.push('journal gap is reported as corruption and never converted to an empty document');
|
||||
await engineRecoveryCase(name, assert);
|
||||
results.push('project engine reopens real IndexedDB with one history unit and the original request; ACK resolves without duplicate edit');
|
||||
await atomicResourceDeletionCase(name,assert);
|
||||
results.push('atomic local file collection rolls back injected failure after IDB delete, captures descriptors, rejects missing markers and survives reopening');
|
||||
await resourceRecoveryCase(name, assert);
|
||||
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
|
||||
await digestWorkerCase(assert);
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createDocument,createRegistry,builtinComponents,createEditor} from '../../src/lib/canvas-core/index.ts';
|
||||
const registry=createRegistry(builtinComponents()),record={id:'A',incarnation:1,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:'P',data:{title:'A',text:''},geometry:{x:0,y:0,width:100,height:100,rotation:0}},document=createDocument('project',[{id:'P',parent:null}],[record]);
|
||||
test('coordinator prepares runtime atomically for commands, history, metadata and canonical replacement',()=>{
|
||||
let nextId=0,reject=false,calls=0;const editor=createEditor({document,registry,runtime:{prepared:0},createId:()=>`op${++nextId}`,prepareRuntime:(previous,next)=>{calls++;assert.equal(previous,editor.getSnapshot());assert.ok(Object.isFrozen(next));if(reject)throw Error('reserved resource');return Object.freeze({...next.runtime,prepared:previous.runtime.prepared+1});}});
|
||||
const snapshots=[];editor.subscribe((_,snapshot)=>{assert.equal(snapshot.runtime.prepared,snapshots.length+1);assert.equal(editor.getQueries().document,snapshot.document);snapshots.push(snapshot);});
|
||||
assert.equal(editor.dispatch({type:'element.data',target:{id:'A',incarnation:1},patch:{text:'new'}}).status,'applied');assert.equal(editor.history.undo().status,'applied');assert.equal(editor.history.redo().status,'applied');assert.equal(editor.updateRuntime(previous=>({...previous,extra:true})).status,'applied');const source=editor.getSnapshot();assert.equal(editor.reconcile(document,source.runtime,source.rootRevision).status,'applied');assert.equal(calls,5);
|
||||
reject=true;const root=editor.getSnapshot(),query=editor.getQueries();for(const operation of [()=>editor.dispatch({type:'element.data',target:{id:'A',incarnation:1},patch:{text:'blocked'}}),()=>editor.history.undo(),()=>editor.updateRuntime(()=>({prepared:100})),()=>editor.reconcile(source.document,source.runtime,root.rootRevision)]){assert.equal(operation().status,'rejected');assert.equal(editor.getSnapshot(),root);assert.equal(editor.getQueries(),query);}assert.equal(snapshots.length,5);
|
||||
});
|
||||
test('an async coordinator cannot publish a partially prepared root',()=>{
|
||||
const editor=createEditor({document,registry,runtime:0,prepareRuntime:()=>Promise.resolve(1)}),root=editor.getSnapshot();assert.equal(editor.dispatch({type:'element.data',target:{id:'A',incarnation:1},patch:{text:'blocked'}}).code,'async_transaction');assert.equal(editor.updateRuntime(()=>1).code,'async_transaction');assert.equal(editor.reconcile(document,1,root.rootRevision).code,'async_transaction');assert.equal(editor.getSnapshot(),root);
|
||||
});
|
||||
@@ -24,5 +24,5 @@ test('cached subtree height cannot bypass the depth bound and omitted object fie
|
||||
const cycle={};cycle.self=cycle;assert.throws(()=>encodeRecoveryValue(cycle),/invalid_recovery_payload/);assert.throws(()=>encodeRecoveryValue({big:1n}));
|
||||
});
|
||||
test('only factory-owned images are recognized; immutable-looking impostors retain defensive-copy handling',()=>{
|
||||
const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.throws(()=>sealRecoveryImage(3,1,1,{}),/corrupt_recovery_image/);
|
||||
const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.equal(isSealedRecoveryImage(sealRecoveryImage(3,1,1,{})),true);assert.throws(()=>sealRecoveryImage(4,1,1,{}),/corrupt_recovery_image/);
|
||||
});
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {resourceRetentionPins} from '../../src/lib/canvas-runtime/resource-retention.ts';
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createDocument, applyChanges, builtinComponents, createRegistry } from '../../src/lib/canvas-core/index.ts';
|
||||
@@ -16,9 +17,10 @@ function fixture() {
|
||||
let image = null, owner, counter = 0, server = original, version = 1;
|
||||
const files = new Map(), batches = [], uploads = [], writes = [], receipts = new Map(), assetReceipts = new Map();
|
||||
const storage = {
|
||||
fileDeletion: 'atomic-v1',
|
||||
async claim(scope) { assert.equal(owner, undefined); const handle = { scope, fence: 1, active: true, async release() { owner = undefined; handle.active = false; } }; owner = handle; return handle; },
|
||||
async recover() { return structuredClone(image); },
|
||||
async commit(_leader, batch) { for (const file of batch.files ?? []) { assert.equal(files.has(file.id), false); files.set(file.id, file.bytes); } image = structuredClone(batch.image); batches.push({ ...batch, image }); return { durableRootRevision: image.rootRevision }; },
|
||||
async commit(_leader, batch) { for (const file of batch.files ?? []) { assert.equal(files.has(file.id), false); files.set(file.id, file.bytes); } for(const id of batch.deleteFiles??[])files.delete(id); image = structuredClone(batch.image); batches.push({ ...batch, image }); return { durableRootRevision: image.rootRevision }; },
|
||||
async readFile(_leader, id) { return files.get(id); }, close() {},
|
||||
};
|
||||
const assets = {
|
||||
@@ -170,3 +172,80 @@ test('digest worker failure remains distinct from missing bytes and explicit ret
|
||||
const {importSourceDigest}=await import('../../src/lib/canvas-runtime/import-integrity.ts'),f=fixture(),errors=[];let calls=0;const engine=await f.open({digests:{async digest(bytes,signal){if(++calls===1)throw Object.assign(Error('worker failed'),{code:'digest_worker_failed'});return importSourceDigest(bytes,signal);}},onError:error=>errors.push(error)});
|
||||
try{const ref=addImage(engine);await engine.settle();const resource=engine.getSnapshot().runtime.resources.get(ref.id);assert.deepEqual(engine.getSnapshot().runtime.ledger.operations.get(resource.operationId).blockers,['digest_worker_failed']);assert.equal(f.files.size,1);assert.equal(f.uploads.length,0);await new Promise(resolve=>setTimeout(resolve,10));assert.equal(calls,1,'failed worker retried in a busy loop');await engine.retry();await engine.settle();assert.equal(f.uploads.length,1);assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).stored.id,ref.id);assert.equal(engine.getSnapshot().history.undo.length,1);assert.equal(errors.length,1);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
const resourceClock=()=>{let now=1000;return{advance:ms=>{now+=ms;},now:()=>now,setTimer:(fn,ms)=>setTimeout(fn,ms),clearTimer:id=>clearTimeout(id)};};
|
||||
async function unreferenceImage(engine){assert.equal(engine.editor.dispatch({type:'element.remove',target:{id:'picture',incarnation:1}}).status,'applied');assert.equal(engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'evict old image history'}}).status,'applied');await engine.settle();}
|
||||
|
||||
test('local collection persists a soft marker, waits retention and atomically deletes bytes with a v3 recovery tombstone',async()=>{
|
||||
const f=fixture(),clock=resourceClock();let engine=await f.open({historyLimit:1,clock});const ref=addImage(engine);try{
|
||||
await engine.settle();await unreferenceImage(engine);assert.equal(engine.getResourcePins().pins.has(ref.id),false);
|
||||
const content=engine.getSnapshot().document,history=engine.getSnapshot().history;
|
||||
const soft=await engine.collectResources({retentionMs:100});assert.deepEqual(soft.markedIds,[ref.id]);assert.deepEqual(soft.purgedIds,[]);assert.ok(f.files.has(ref.id));assert.equal(f.image().formatVersion,3);decodeRecovery(f.image(),registry);
|
||||
clock.advance(99);assert.deepEqual((await engine.collectResources({retentionMs:100})).purgedIds,[]);clock.advance(1);
|
||||
const collected=await engine.collectResources({retentionMs:100});assert.deepEqual(collected.purgedIds,[ref.id]);assert.equal(f.files.has(ref.id),false);assert.equal(engine.getSnapshot().document,content);assert.equal(engine.getSnapshot().history,history);
|
||||
const deletion=f.batches.find(batch=>batch.deleteFiles?.includes(ref.id));assert.equal(deletion.image.payload.runtime.resources.find(([id])=>id===ref.id)[1].retirement.purgedAt,1100);assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).stored.id,ref.id);
|
||||
const cut=decodeRecovery(f.image(),registry);assert.equal(cut.runtime.resources.get(ref.id).retirement.purgedAt,1100);
|
||||
for(const version of [1,2]){const bad=structuredClone(f.image());bad.formatVersion=version;assert.throws(()=>decodeRecovery(bad,registry),/unsupported_recovery_version/);}
|
||||
await engine.close();engine=await f.open({historyLimit:1,clock});const handle=await engine.displayResource(ref.id);assert.equal(handle.url,f.assets.displayURL(ref.id));handle.release();assert.equal(f.uploads.length,1);assert.deepEqual((await engine.collectResources({retentionMs:0})).purgedIds,[]);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('document/history/ACK/explicit pins and unknown coverage prevent collection, and repinning resets the soft retention clock',async()=>{
|
||||
const f=fixture(),clock=resourceClock(),engine=await f.open({historyLimit:1,clock}),ref=addImage(engine);try{
|
||||
await engine.settle();assert.deepEqual((await engine.collectResources({retentionMs:0})).purgedIds,[]);
|
||||
engine.editor.dispatch({type:'element.remove',target:{id:'picture',incarnation:1}});await engine.settle();assert.ok([...engine.getResourcePins().pins.get(ref.id)].some(owner=>owner.startsWith('history:')));assert.deepEqual((await engine.collectResources({retentionMs:0})).purgedIds,[]);
|
||||
engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'clear history'}});await engine.settle();await engine.collectResources({retentionMs:100});clock.advance(200);
|
||||
assert.equal(engine.editor.dispatch({type:'element.create',element:node('again',{assetId:ref.id})}).status,'applied');assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).retirement,undefined);await engine.settle();
|
||||
engine.editor.dispatch({type:'element.remove',target:{id:'again',incarnation:1}});engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'clear again'}});await engine.settle();const next=await engine.collectResources({retentionMs:100});assert.deepEqual(next.markedIds,[ref.id]);assert.deepEqual(next.purgedIds,[]);assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).retirement.unreferencedAt,1200);
|
||||
engine.editor.updateRuntime(previous=>({...previous,resourcePins:new Map([['export-job',[ref.id]]])}));assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).retirement,undefined);assert.deepEqual((await engine.collectResources({retentionMs:0})).purgedIds,[]);
|
||||
engine.editor.updateRuntime(previous=>({...previous,resourcePins:new Map()}));const root=engine.getSnapshot(),unknown=createDocument('project',[...root.document.canvases.values()],[...root.document.elements.values(),{...node('unknown'),componentId:'future'}]);assert.equal(engine.editor.reconcile(unknown,root.runtime,root.rootRevision).status,'applied');const incomplete=await engine.collectResources({retentionMs:0});assert.equal(incomplete.complete,false);assert.deepEqual(incomplete.purgedIds,[]);assert.ok(f.files.has(ref.id));
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('a display pins before its asynchronous read and resets retirement; releasing it starts a fresh retention window',async()=>{
|
||||
const f=fixture(),clock=resourceClock(),engine=await f.open({historyLimit:1,clock}),ref=addImage(engine);const read=f.storage.readFile,gate=deferred(),entered=deferred();let blocked=false;try{
|
||||
await engine.settle();await unreferenceImage(engine);await engine.collectResources({retentionMs:100});clock.advance(200);
|
||||
f.storage.readFile=async(...args)=>{if(blocked){entered.resolve();await gate.promise;}return read(...args);};blocked=true;const display=engine.displayResource(ref.id);await entered.promise;
|
||||
assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).retirement,undefined);assert.ok(engine.getResourcePins().pins.get(ref.id).has('display'));assert.deepEqual((await engine.collectResources({retentionMs:0})).purgedIds,[]);gate.resolve();const handle=await display;assert.match(handle.url,/^blob:/);handle.release();handle.release();assert.equal(engine.getResourcePins().pins.has(ref.id),false);assert.deepEqual((await engine.collectResources({retentionMs:100})).purgedIds,[]);assert.equal(engine.getSnapshot().runtime.resources.get(ref.id).retirement.unreferencedAt,1200);
|
||||
}finally{gate.resolve();await engine.close();}
|
||||
});
|
||||
|
||||
test('only recollection of the reserved resource is fenced during deletion; unrelated editing stays available',async()=>{
|
||||
const f=fixture(),engine=await f.open({historyLimit:1}),ref=addImage(engine),gate=deferred(),entered=deferred(),commit=f.storage.commit;try{
|
||||
await engine.settle();await unreferenceImage(engine);f.storage.commit=async(...args)=>{if(args[1].deleteFiles?.length){entered.resolve();await gate.promise;}return commit(...args);};
|
||||
const collecting=engine.collectResources({retentionMs:0});await entered.promise;
|
||||
assert.equal(engine.editor.dispatch({type:'element.create',element:node('late',{assetId:ref.id})}).code,'resource_collection_in_progress');await assert.rejects(engine.displayResource(ref.id),/resource_collection_in_progress/);
|
||||
assert.equal(engine.editor.dispatch({type:'element.data',target:{id:'a',incarnation:1},patch:{title:'can still edit'}}).status,'applied');assert.ok(f.files.has(ref.id));gate.resolve();await collecting;assert.equal(f.files.has(ref.id),false);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'can still edit');
|
||||
}finally{gate.resolve();await engine.close();}
|
||||
});
|
||||
|
||||
test('failed deletion keeps bytes, pending recovery cut and reservation until an explicit persistence retry succeeds',async()=>{
|
||||
const f=fixture(),engine=await f.open({historyLimit:1}),ref=addImage(engine),commit=f.storage.commit;let fail=false;try{
|
||||
await engine.settle();await unreferenceImage(engine);f.storage.commit=async(...args)=>{if(fail&&args[1].deleteFiles?.length)throw Error('disk unavailable');return commit(...args);};fail=true;
|
||||
await assert.rejects(engine.collectResources({retentionMs:0}),/disk unavailable/);assert.ok(f.files.has(ref.id));assert.equal(f.image().payload.runtime.resources.find(([id])=>id===ref.id)[1].retirement.purgedAt,undefined);await assert.rejects(engine.displayResource(ref.id),/resource_collection_in_progress/);
|
||||
fail=false;await engine.persist();assert.equal(f.files.has(ref.id),false);assert.ok(engine.getSnapshot().runtime.resources.get(ref.id).retirement.purgedAt!==undefined);const display=await engine.displayResource(ref.id);assert.equal(display.url,f.assets.displayURL(ref.id));display.release();
|
||||
}finally{fail=false;await engine.close();}
|
||||
});
|
||||
|
||||
test('a synchronous publication listener cannot persist a purge marker without its file deletion',async()=>{
|
||||
const f=fixture(),engine=await f.open({historyLimit:1}),ref=addImage(engine);let unsubscribe;try{
|
||||
await engine.settle();await unreferenceImage(engine);unsubscribe=engine.editor.subscribe(()=>{void engine.persist().catch(()=>{});});await engine.collectResources({retentionMs:0});assert.equal(f.files.has(ref.id),false);
|
||||
const purged=f.batches.filter(batch=>batch.image.payload.runtime.resources.some(([id,resource])=>id===ref.id&&resource.retirement?.purgedAt!==undefined));assert.ok(purged.length);assert.ok(purged[0].deleteFiles.includes(ref.id),'first durable purge cut omitted physical deletion');
|
||||
}finally{unsubscribe?.();await engine.close();}
|
||||
});
|
||||
|
||||
|
||||
test('retention covers ACK-only content, opaque job envelopes and historical field schemas independently of visibility',async()=>{
|
||||
const f=fixture(),engine=await f.open();try{
|
||||
const root=engine.getSnapshot(),empty=createDocument('project',[...root.document.canvases.values()],[]),resource={id:'kept',operationId:'upload',name:'x',mimeType:'image/png',byteLength:1},base={...root,document:empty,runtime:{...root.runtime,resources:new Map([['kept',resource]]),acknowledged:createDocument('project',[...empty.canvases.values()],[node('remote',{assetId:'kept'})])}};
|
||||
assert.ok([...resourceRetentionPins(base,registry).pins.get('kept')].some(owner=>owner.startsWith('acknowledged:')));
|
||||
const plain={...base,runtime:{...base.runtime,acknowledged:empty}};
|
||||
for(const status of ['preparing','running','paused']){const snapshot={...plain,runtime:{...plain.runtime,imports:new Map([['job',{id:'job',status}]])}};assert.ok(resourceRetentionPins(snapshot,registry).pins.get('kept').has('import:job'));}
|
||||
for(const status of ['completed','partial','cancelled'])assert.equal(resourceRetentionPins({...plain,runtime:{...plain.runtime,imports:new Map([['job',{id:'job',status}]])}},registry).pins.has('kept'),false);
|
||||
assert.ok(resourceRetentionPins({...plain,runtime:{...plain.runtime,runs:new Map([['run',{id:'run',status:'succeeded',outputs:[{opaque:'kept'}]}]])}},registry).pins.get('kept').has('run:run'));
|
||||
assert.ok(resourceRetentionPins(plain,registry,new Map(),new Map([['copy-admission',['kept']]])).pins.get('kept').has('copy-admission'));
|
||||
const unknown={...plain,runtime:{...plain.runtime,acknowledged:createDocument('project',[...empty.canvases.values()],[{...node('future'),componentId:'future'}])}};assert.equal(resourceRetentionPins(unknown,registry).complete,false);
|
||||
const field={kind:'field',target:{id:'deleted',incarnation:1},field:'data.assetId',before:'kept',after:'next'},history={undo:['h'],redo:[],entries:new Map([['h',{id:'h',changes:[field]}]])};assert.equal(resourceRetentionPins({...plain,history},registry).complete,false);
|
||||
const malformed={...plain,document:createDocument('project',[...empty.canvases.values()],[node('bad',{assetId:{opaque:'kept'}})])};assert.equal(resourceRetentionPins(malformed,registry).complete,false);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user