diff --git a/src/lib/canvas-core/editor.ts b/src/lib/canvas-core/editor.ts index 6422616..91bac81 100644 --- a/src/lib/canvas-core/editor.ts +++ b/src/lib/canvas-core/editor.ts @@ -21,6 +21,8 @@ export type EditorOptions = { 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, next: Snapshot) => Runtime; nextOperationSequence?: (runtime: Runtime, fallback: number) => number; }; export type Transaction = { dispatch(command: Command): Availability; getDocument(): Document }; @@ -119,6 +121,12 @@ export function createEditor(options: EditorOptions): Snapshot { + 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(options: EditorOptions(options: EditorOptions(options: EditorOptions!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: [] })); diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 715f91a..919f875 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -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) }; const displays = new Set<() => void>(); + const displayPins=new Map(),collectionReservations=new Set(),durablyPurged=new Set([...(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(), digestTasks = new Map>(); let staged: readonly ResourceRecord[] = []; let applying: RunApplyInput | undefined; @@ -69,12 +72,14 @@ export async function openProjectEngine(options: EngineOptions) { const copyPreparations=new Map>(); const importRunners=new Map>(),importLoads=new Map>(),importHolds=new Mapvoid>>(); let recovery: { controller: AbortController; release(): void; cancel():Promise } | 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(); 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) => 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{ - 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 { 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(); diff --git a/src/lib/canvas-runtime/recovery-encoding.ts b/src/lib/canvas-runtime/recovery-encoding.ts index bf41cd9..cb70ba4 100644 --- a/src/lib/canvas-runtime/recovery-encoding.ts +++ b/src/lib/canvas-runtime/recovery-encoding.ts @@ -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);} diff --git a/src/lib/canvas-runtime/recovery.ts b/src/lib/canvas-runtime/recovery.ts index d4c40f2..3d1363a 100644 --- a/src/lib/canvas-runtime/recovery.ts +++ b/src/lib/canvas-runtime/recovery.ts @@ -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; 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 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(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; } diff --git a/src/lib/canvas-runtime/resource-retention.ts b/src/lib/canvas-runtime/resource-retention.ts new file mode 100644 index 0000000..a92e865 --- /dev/null +++ b/src/lib/canvas-runtime/resource-retention.ts @@ -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=new Map(),temporary:ReadonlyMap=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>(); + 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,reserved:ReadonlySet,temporary:ReadonlyMap=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).set(id,freeze(kept)); + } + } + return resources===next.runtime.resources?next.runtime:Object.freeze({...next.runtime,resources:readMap(resources)}); +} diff --git a/src/lib/canvas-runtime/resources.ts b/src/lib/canvas-runtime/resources.ts index 420c8db..d3eb91f 100644 --- a/src/lib/canvas-runtime/resources.ts +++ b/src/lib/canvas-runtime/resources.ts @@ -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(); + 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}`); diff --git a/src/lib/canvas-runtime/storage.ts b/src/lib/canvas-runtime/storage.ts index 2f57586..b696718 100644 --- a/src/lib/canvas-runtime/storage.ts +++ b/src/lib/canvas-runtime/storage.ts @@ -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 }; export type StoragePort = { + readonly fileDeletion?: 'atomic-v1'; claim(scope: StorageScope): Promise; recover(leader: StorageLeader): Promise; 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 = (request: IDBRequest) => new Promise((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(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=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]); diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index 2197b02..782fd69 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -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); diff --git a/tests/headless/publication-guard.test.mjs b/tests/headless/publication-guard.test.mjs new file mode 100644 index 0000000..32d3096 --- /dev/null +++ b/tests/headless/publication-guard.test.mjs @@ -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); +}); diff --git a/tests/headless/recovery-encoding.test.mjs b/tests/headless/recovery-encoding.test.mjs index 584abb7..0c28b28 100644 --- a/tests/headless/recovery-encoding.test.mjs +++ b/tests/headless/recovery-encoding.test.mjs @@ -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/); }); diff --git a/tests/headless/resources.test.mjs b/tests/headless/resources.test.mjs index cd70bca..f8ee55a 100644 --- a/tests/headless/resources.test.mjs +++ b/tests/headless/resources.test.mjs @@ -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();} +});