Retire terminal import sources with durable ownership and recovery v4
This commit is contained in:
@@ -26,6 +26,7 @@ export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; c
|
||||
{jobs.map((job, index) => <section className={styles.importJob} key={job.id} aria-label={`导入任务 ${index + 1}`}>
|
||||
<strong>{job.source?.name || `导入任务 ${index + 1}`}</strong>
|
||||
<p role="status">{["running","preparing"].includes(job.status) && job.source?.integrityVersion === 1 && !job.source.sha256 ? "正在校验源文件" : job.status === "preparing" ? "正在准备复制" : job.status === "running" ? "正在加入" : job.status === "paused" ? "已暂停" : job.status === "cancelled" ? "已停止" : job.status === "partial" ? "部分加入" : "已加入画布"} · {job.committedSourceKeys.length} 项{job.pending.length > 0 && `，${job.pending.length} 项等待`}</p>
|
||||
{job.source?.retirement?.purgedAt !== undefined && <small>本地源文件已清理；画布内容和撤销记录保留。</small>}
|
||||
{job.error && <p>{reason(job.error)}</p>}
|
||||
<div className={styles.importActions}>
|
||||
{["running","preparing"].includes(job.status) && <button onClick={() => control(engine.imports.pause(job.id))}>暂停导入</button>}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import {importFiles,terminalImport,markImportFiles,prepareImportFileRetention,type FileRetirement} from './import-file-retention.ts';
|
||||
import {resourceRetentionPins,prepareResourceRetention} from './resource-retention.ts';
|
||||
import { encodeIndexedCopySnapshot, openIndexedCopySnapshot, type IndexedCopySnapshot } from './copy-snapshot-index.ts';
|
||||
import { decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
|
||||
@@ -63,7 +64,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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;
|
||||
type CollectionReport={readonly markedIds:readonly Id[];readonly purgedIds:readonly Id[];readonly complete:boolean;readonly checkedRootRevision:number;readonly rootRevision:number;readonly durableRootRevision:number};
|
||||
let sourceCollection:Promise<CollectionReport>|undefined;
|
||||
for(const [id,entry]of importFiles(recovered?.runtime.imports))if(entry.file.retirement?.purgedAt!==undefined)durablyPurged.add(id);
|
||||
let collection:Promise<CollectionReport>|undefined;
|
||||
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
let applying: RunApplyInput | undefined;
|
||||
@@ -79,7 +83,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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)),
|
||||
prepareRuntime:(previous,next)=>{
|
||||
const runtime=prepareResourceRetention(next,options.registry,displayPins,collectionReservations,admissionPins(next));
|
||||
return prepareImportFileRetention(previous.runtime,runtime,()=>resourceRetentionPins({...next,runtime},options.registry,displayPins,admissionPins({...next,runtime})),collectionReservations,previous.document!==next.document||previous.history!==next.history||previous.runtime.ledger!==runtime.ledger||previous.runtime.resourcePins!==runtime.resourcePins);
|
||||
},
|
||||
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) {
|
||||
@@ -117,8 +124,9 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const task = (async () => {
|
||||
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 deleteFiles=[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id);
|
||||
const sourceFiles=importFiles(snapshot.runtime.imports);
|
||||
const files = [...fileWrites].filter(([id]) => snapshot.runtime.resources.has(id)||sourceFiles.has(id)).map(([id, bytes]) => ({ id, bytes }));
|
||||
const deleteFiles=[...[...snapshot.runtime.resources.values()].filter(resource=>resource.retirement?.purgedAt!==undefined&&!durablyPurged.has(resource.id)).map(resource=>resource.id),...[...sourceFiles].filter(([id,entry])=>entry.file.retirement?.purgedAt!==undefined&&!durablyPurged.has(id)).map(([id])=>id)];
|
||||
const 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);
|
||||
@@ -164,6 +172,52 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
})();
|
||||
collection=task.finally(()=>{collection=undefined;});return collection;
|
||||
}
|
||||
function collectImportSources(settings:{retentionMs:number;maxFiles?:number}):Promise<CollectionReport> {
|
||||
assertWritable();if(options.storage.fileDeletion!=='atomic-v1')throw fault('resource_collection_unavailable');
|
||||
const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64;
|
||||
if(!Number.isSafeInteger(retentionMs)||retentionMs<0||!Number.isSafeInteger(maxFiles)||maxFiles<1||maxFiles>128)throw fault('invalid_resource_collection');
|
||||
if(sourceCollection)return sourceCollection;
|
||||
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});
|
||||
const candidates=()=>[...importFiles(root.runtime.imports)].filter(([id,entry])=>!pins.pins.has(id)&&!fileWrites.has(id)&&(!entry.current||terminalImport(root.runtime.imports!.get(entry.jobId)!.status)&&!importLoads.has(entry.jobId)));
|
||||
if(!pins.complete)return report();
|
||||
const fresh=candidates().filter(([,entry])=>!entry.file.retirement).slice(0,maxFiles);
|
||||
if(fresh.length){
|
||||
const markers=new Map(fresh.map(([id])=>[id,{version:1 as const,unreferencedAt:now}]));
|
||||
const result=editor.updateRuntime(previous=>markImportFiles(previous,markers),root.rootRevision);if(result.status==='rejected')throw fault(result.code,result.targets);
|
||||
markedIds.push(...markers.keys());await persist();assertWritable();
|
||||
}
|
||||
root=editor.getSnapshot();pins=resourceRetentionPins(root,options.registry,displayPins,admissionPins(root));if(!pins.complete)return report();
|
||||
const eligible=candidates().filter(([,entry])=>entry.file.retirement&&entry.file.retirement.purgedAt===undefined&&now-entry.file.retirement.unreferencedAt>=retentionMs).slice(0,maxFiles);
|
||||
if(eligible.length){
|
||||
const markers=new Map<Id,FileRetirement>(eligible.map(([id,entry])=>[id,{...entry.file.retirement!,purgedAt:now}]));
|
||||
for(const id of markers.keys())collectionReservations.add(id);
|
||||
const result=editor.updateRuntime(previous=>markImportFiles(previous,markers),root.rootRevision);
|
||||
if(result.status==='rejected'){for(const id of markers.keys())collectionReservations.delete(id);throw fault(result.code,result.targets);}
|
||||
await persist();purgedIds.push(...markers.keys());
|
||||
}
|
||||
return report();
|
||||
})();
|
||||
sourceCollection=task.finally(()=>{sourceCollection=undefined;});return sourceCollection;
|
||||
}
|
||||
async function openImportSource(jobId:Id,fileId?:Id) {
|
||||
if(closing||closed)throw fault('engine_closing');const job=getImport(jobId),file=fileId!==undefined?importFiles(new Map([[jobId,job]])).get(fileId)?.file:job.source;if(!file)throw fault('import_source_unavailable');
|
||||
if(collectionReservations.has(file.fileId))throw fault('resource_collection_in_progress',[file.fileId]);
|
||||
if(file.retirement?.purgedAt!==undefined)throw fault('import_source_retired',[file.fileId]);
|
||||
displayPins.set(file.fileId,(displayPins.get(file.fileId)??0)+1);let released=false;
|
||||
const release=()=>{if(released)return;released=true;displays.delete(release);const count=displayPins.get(file.fileId)??1;if(count<=1)displayPins.delete(file.fileId);else displayPins.set(file.fileId,count-1);};displays.add(release);
|
||||
try{
|
||||
if(file.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);}
|
||||
const bytes=fileWrites.get(file.fileId)??await options.storage.readFile(leader,file.fileId);
|
||||
if(closing||closed)throw fault('engine_closing');if(!bytes||bytes.size!==file.byteLength)throw fault('import_source_unavailable',[file.fileId]);
|
||||
if(file.sha256&&await sourceDigest(bytes)!==file.sha256)throw fault('import_source_changed',[file.fileId]);
|
||||
if(closing||closed)throw fault('engine_closing');return Object.freeze({bytes,release});
|
||||
}catch(error){release();if(closing||closed)throw fault('engine_closing');throw error;}
|
||||
}
|
||||
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);
|
||||
@@ -398,7 +452,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
|
||||
const fileId=createId(),runtime=editor.getSnapshot().runtime;if(!fileId||fileWrites.has(fileId)||runtime.resources.has(fileId)||[...(runtime.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict');
|
||||
const stableIds=Object.fromEntries([...Object.entries(plan.elementIds).map(([old,id])=>[`element:${old}`,id]),...Object.entries(plan.canvasIds).map(([old,id])=>[`canvas:${old}`,id])]);
|
||||
const next={...job,generation:job.generation+1,status:'running' as const,stableIds,sourceFingerprint:`local-file:${fileId}:${encoded.size}`,source:{...job.source!,kind:'duplicate-ndjson' as const,fileId,byteLength:encoded.size,sha256,integrityVersion:1 as const,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:plan.roots}}};delete next.preparation;delete next.error;
|
||||
const next={...job,generation:job.generation+1,status:'running' as const,stableIds,sourceFingerprint:`local-file:${fileId}:${encoded.size}`,source:{...job.source!,previousFiles:[...(job.source!.previousFiles??[]),{fileId:job.source!.fileId,byteLength:job.source!.byteLength,...(job.source!.sha256?{sha256:job.source!.sha256}:{})}],kind:'duplicate-ndjson' as const,fileId,byteLength:encoded.size,sha256,integrityVersion:1 as const,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:plan.roots}}};delete next.preparation;delete next.error;
|
||||
fileWrites.set(fileId,encoded);try{publishImport(freeze(next),job.generation);}catch(error){fileWrites.delete(fileId);throw error;}
|
||||
return{job:getImport(id),bytes:encoded,document:undefined,indexed:undefined};
|
||||
}
|
||||
@@ -443,9 +497,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
}
|
||||
const imports=Object.freeze({
|
||||
getJobs:()=>readMap(editor.getSnapshot().runtime.imports??new Map()),
|
||||
openSource:openImportSource,
|
||||
startNdjson(bytes:Blob,targetCanvasId:Id):Id{
|
||||
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source');
|
||||
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||[...(current.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict');
|
||||
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
|
||||
const entry=canvas.parent?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined;
|
||||
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const});
|
||||
fileWrites.set(fileId,bytes);
|
||||
@@ -467,7 +522,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
const bytes=await encodeIndexedCopySnapshot(captured.document,controller.signal);assertWritable();if(controller.signal.aborted)throw fault('copy_aborted');
|
||||
const canvas=captured.document.canvases.get(target);if(!canvas)throw fault('canvas_not_loaded',[target]);
|
||||
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;
|
||||
if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||[...(current.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict');
|
||||
if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict');
|
||||
const entry=canvas.parent?captured.document.elements.get(canvas.parent.entryElementId)!:undefined;
|
||||
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'duplicate-snapshot',snapshotVersion:2,name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:[]},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'preparing' as const,preparation:{version:4 as const,registrySignature:copyRegistrySignature(captured.document,options.registry),selection:targets,worldOffset:offset,allocationIds:[],steps:0}});
|
||||
fileWrites.set(fileId,bytes);const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
|
||||
@@ -482,7 +537,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
|
||||
for(const job of initial.imports?.values()??[])if(['running','preparing'].includes(job.status))void loadImport(job.id);
|
||||
return Object.freeze({
|
||||
editor, sessions, imports, collectResources,
|
||||
editor, sessions, imports, collectResources,collectImportSources,
|
||||
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> {
|
||||
@@ -622,7 +677,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
if (!(file.bytes instanceof Blob) || !file.bytes.size || file.bytes.size > maxResourceBytes || !file.bytes.type.startsWith('image/') || typeof file.name !== 'string' || !file.name) throw fault('invalid_staged_file');
|
||||
return freeze({ id: createId(), operationId: createId(), name: file.name, mimeType: file.bytes.type, byteLength: file.bytes.size });
|
||||
});
|
||||
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || fileWrites.has(resource.id) || [...(editor.getSnapshot().runtime.imports?.values()??[])].some(job=>job.source?.fileId===resource.id))) throw fault('resource_identity_conflict');
|
||||
if (new Set(resources.flatMap(resource => [resource.id, resource.operationId])).size !== resources.length * 2 || resources.some(resource => editor.getSnapshot().runtime.resources.has(resource.id) || fileWrites.has(resource.id) || importFiles(editor.getSnapshot().runtime.imports).has(resource.id))) throw fault('resource_identity_conflict');
|
||||
staged = resources; resources.forEach((resource, index) => fileWrites.set(resource.id, files[index].bytes));
|
||||
try {
|
||||
const handles = freeze(resources.map(resource => ({ id: resource.id, src: options.assets!.sourceURL(resource.id) })));
|
||||
@@ -644,6 +699,11 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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);
|
||||
}
|
||||
if(!retained){
|
||||
const source=importFiles(editor.getSnapshot().runtime.imports).get(id)?.file;
|
||||
if(source?.retirement?.purgedAt!==undefined)throw fault('import_source_retired',[id]);
|
||||
if(source?.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));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);
|
||||
@@ -660,7 +720,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
|
||||
async close() {
|
||||
if (closed) return;
|
||||
closing = true;await 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();
|
||||
closing = true;await Promise.allSettled([collection,sourceCollection].filter(Boolean));digestLifecycle.abort();for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
|
||||
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();
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
import type {Id} from '../canvas-core/types.ts';
|
||||
import {equal,fault,freeze,readMap} from '../canvas-core/values.ts';
|
||||
import type {ImportJob,NativeImportSource} from './imports.ts';
|
||||
import type {RuntimeState} from './contracts.ts';
|
||||
|
||||
export type FileRetirement={readonly version:1;readonly unreferencedAt:number;readonly purgedAt?:number};
|
||||
export type PreviousImportFile={readonly fileId:Id;readonly byteLength:number;readonly sha256?:string;readonly retirement?:FileRetirement};
|
||||
export type ImportFile={readonly jobId:Id;readonly current:boolean;readonly file:PreviousImportFile};
|
||||
export const terminalImport=(status:ImportJob['status'])=>['completed','partial','cancelled'].includes(status);
|
||||
export function validFileRetirement(value:unknown):value is FileRetirement {
|
||||
const marker=value as FileRetirement;
|
||||
return !!marker&&marker.version===1&&Number.isSafeInteger(marker.unreferencedAt)&&marker.unreferencedAt>=0&&(marker.purgedAt===undefined||Number.isSafeInteger(marker.purgedAt)&&marker.purgedAt>=marker.unreferencedAt);
|
||||
}
|
||||
export function importFiles(jobs:ReadonlyMap<Id,ImportJob>|undefined):ReadonlyMap<Id,ImportFile> {
|
||||
const files=new Map<Id,ImportFile>();
|
||||
for(const [jobId,job]of jobs??[]){
|
||||
if(jobId!==job.id)throw fault('import_file_identity_conflict',[jobId]);
|
||||
if(!job.source)continue;
|
||||
const add=(file:PreviousImportFile,current:boolean)=>{if(files.has(file.fileId))throw fault('import_file_identity_conflict',[file.fileId]);files.set(file.fileId,{jobId:job.id,current,file});};
|
||||
add(job.source,true);for(const file of job.source.previousFiles??[])add(file,false);
|
||||
}
|
||||
return files;
|
||||
}
|
||||
export function validateImportFileRetention(source:NativeImportSource,status:ImportJob['status']) {
|
||||
if(source.retirement!==undefined&&(!validFileRetirement(source.retirement)||!terminalImport(status)))throw fault('invalid_import_file_retirement');
|
||||
if(source.previousFiles!==undefined&&(!Array.isArray(source.previousFiles)||source.previousFiles.some(file=>!file||typeof file.fileId!=='string'||!file.fileId||!Number.isSafeInteger(file.byteLength)||file.byteLength<0||file.sha256!==undefined&&!/^[a-f0-9]{64}$/.test(file.sha256)||file.retirement!==undefined&&!validFileRetirement(file.retirement))))throw fault('invalid_import_file_retirement');
|
||||
const ids=[source.fileId,...(source.previousFiles??[]).map(file=>file.fileId)];if(new Set(ids).size!==ids.length)throw fault('import_file_identity_conflict');
|
||||
}
|
||||
function mark<T extends PreviousImportFile>(file:T,retirement:FileRetirement|undefined):T {
|
||||
if(equal(file.retirement,retirement))return file;
|
||||
const {retirement:_,...rest}=file;return freeze(retirement?{...rest,retirement}:rest) as T;
|
||||
}
|
||||
export function markImportFiles(runtime:RuntimeState,markers:ReadonlyMap<Id,FileRetirement|undefined>):RuntimeState {
|
||||
let imports=runtime.imports;
|
||||
for(const [id,job]of runtime.imports??[]){
|
||||
if(!job.source)continue;
|
||||
let source=markers.has(job.source.fileId)?mark(job.source,markers.get(job.source.fileId)):job.source;
|
||||
const previous=source.previousFiles?.map(file=>markers.has(file.fileId)?mark(file,markers.get(file.fileId)):file);
|
||||
if(previous&&previous.some((file,index)=>file!==source.previousFiles![index]))source=freeze({...source,previousFiles:previous});
|
||||
if(source!==job.source){if(!Number.isSafeInteger(job.generation+1))throw fault('invalid_import_job');if(imports===runtime.imports)imports=new Map(imports);(imports as Map<Id,ImportJob>).set(id,freeze({...job,source,generation:job.generation+1}));}
|
||||
}
|
||||
return imports===runtime.imports?runtime:Object.freeze({...runtime,imports:readMap(imports!)});
|
||||
}
|
||||
/** A source file cannot disappear from its owning job, become an asset or regain execution after purge. */
|
||||
export function prepareImportFileRetention(previous:RuntimeState,next:RuntimeState,pins:()=>{complete:boolean;pins:ReadonlyMap<Id,{has(owner:string):boolean}>},reserved:ReadonlySet<Id>,referencesChanged=true):RuntimeState {
|
||||
const hasSoft=[...(next.imports?.values()??[])].some(job=>job.source&&[job.source,...(job.source.previousFiles??[])].some(file=>file.retirement&&file.retirement.purgedAt===undefined));
|
||||
if(previous.imports===next.imports&&previous.resources===next.resources&&previous.resourcePins===next.resourcePins&&!reserved.size&&!hasSoft&&!referencesChanged)return next;
|
||||
const before=importFiles(previous.imports),after=importFiles(next.imports);
|
||||
for(const job of next.imports?.values()??[])if(job.source)validateImportFileRetention(job.source,job.status);
|
||||
for(const id of after.keys())if(next.resources.has(id))throw fault('import_file_identity_conflict',[id]);
|
||||
for(const [id,old]of before){
|
||||
const entry=after.get(id);
|
||||
if(!entry||entry.jobId!==old.jobId||entry.file.byteLength!==old.file.byteLength||old.file.sha256!==undefined&&entry.file.sha256!==old.file.sha256||!old.current&&entry.current)throw fault('import_file_identity_conflict',[id]);
|
||||
if(old.file.retirement?.purgedAt!==undefined&&!equal(old.file.retirement,entry.file.retirement))throw fault('import_source_retired',[id]);
|
||||
}
|
||||
if(!hasSoft&&!reserved.size&&![...after.values()].some(entry=>entry.file.retirement?.purgedAt!==undefined))return next;
|
||||
const coverage=pins(),clear=new Map<Id,undefined>();
|
||||
for(const [id,entry]of after){
|
||||
if(entry.file.retirement?.purgedAt!==undefined){if(coverage.pins.has(id))throw fault('import_source_retired',[id]);continue;}
|
||||
if(!entry.file.retirement)continue;
|
||||
if(!coverage.complete||coverage.pins.has(id))clear.set(id,undefined);
|
||||
}
|
||||
return markImportFiles(next,clear);
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts';
|
||||
import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
|
||||
import type { Change, Id, Json, Result, Target, Point } from '../canvas-core/types.ts';
|
||||
import { selectImportGroups } from './import-groups.ts';
|
||||
@@ -8,7 +9,7 @@ export type ImportSource = { readonly fingerprint: string; readonly parserVersio
|
||||
export type ImportUnit = { readonly key: string; readonly dependencies: readonly string[]; readonly changes: readonly Change[]; readonly guard?: Json };
|
||||
export type ImportLocator = { readonly key: string; readonly cursor: number; readonly nextCursor: number; readonly dependencies: readonly string[]; readonly digest: string; readonly byteLength: number; readonly changeCount: number };
|
||||
export type ImportFailure = { readonly key: string; readonly cursor: number; readonly code: string; readonly dependencies?: readonly string[] };
|
||||
export type NativeImportSource = { readonly kind: 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
|
||||
export type NativeImportSource = { readonly kind: 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly retirement?:FileRetirement;readonly previousFiles?:readonly PreviousImportFile[];readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
|
||||
export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
|
||||
export type ImportJob = {
|
||||
readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
|
||||
@@ -29,6 +30,7 @@ export function validateImportJob(job: ImportJob): void {
|
||||
if (!Number.isSafeInteger(job.generation) || job.generation<0 || !Number.isSafeInteger(job.cursor) || job.cursor<0 || typeof job.eof!=='boolean' || !['preparing','paused','running','completed','partial','cancelled'].includes(job.status)) throw fault('invalid_import_job');
|
||||
if(job.source&&(!['native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(job.source.kind)||['duplicate-ndjson','duplicate-plan'].includes(job.source.kind)&&(!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0||!Array.isArray(job.source.copy.roots)||!job.source.copy.roots.length||job.source.copy.roots.some(id=>typeof id!=='string'||!id))||job.source.name!==undefined&&(typeof job.source.name!=='string'||job.source.name.length>255)||!job.source.fileId||!job.source.targetCanvasId||!Number.isSafeInteger(job.source.byteLength)||job.source.byteLength<0||job.source.targetEntry&&(!job.source.targetEntry.id||!Number.isSafeInteger(job.source.targetEntry.incarnation)||job.source.targetEntry.incarnation<1)))throw fault('invalid_import_source');
|
||||
if(job.source){
|
||||
validateImportFileRetention(job.source,job.status);
|
||||
if(job.source.snapshotVersion!==undefined&&(job.source.snapshotVersion!==2||!['duplicate-snapshot','duplicate-plan'].includes(job.source.kind)))throw fault('copy_snapshot_interpreter_unavailable');
|
||||
if(job.source.integrityVersion!==undefined&&job.source.integrityVersion!==1)throw fault('import_source_integrity_interpreter_unavailable');
|
||||
if(job.source.sha256!==undefined&&(typeof job.source.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(job.source.sha256))))throw fault('invalid_import_source');
|
||||
|
||||
@@ -52,8 +52,8 @@ export function encodeRecoveryValue(input:unknown):Json {
|
||||
}
|
||||
|
||||
/** Only codec-owned, validated immutable images qualify for the storage copy fast path. */
|
||||
export function sealRecoveryImage(formatVersion:1|2|3,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');
|
||||
export function sealRecoveryImage(formatVersion:1|2|3|4,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage {
|
||||
if(![1,2,3,4].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image');
|
||||
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 {importFiles} from './import-file-retention.ts';
|
||||
import {resourceRetentionPins} from './resource-retention.ts';
|
||||
import { sealRecoveryImage } from './recovery-encoding.ts';
|
||||
import { validateImportJob, type ImportJob } from './imports.ts';
|
||||
@@ -23,7 +24,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.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
|
||||
return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload);
|
||||
}
|
||||
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; }
|
||||
@@ -86,7 +87,7 @@ function readHistory(history:ObjectValue,operations:ReadonlyMap<Id,OperationInte
|
||||
return Object.freeze({undo:freeze(undo),redo:freeze(redo),entries});
|
||||
}
|
||||
export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSnapshot & { readonly rootRevision: number; readonly runtime: RuntimeState } {
|
||||
if (![1,2,3].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||||
if (![1,2,3,4].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.
|
||||
@@ -163,7 +164,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.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');
|
||||
@@ -226,6 +227,7 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
return freeze({...value as RunRecord,outputEvents:array(raw.outputEvents ?? []) as never});
|
||||
});
|
||||
const imports=table<ImportJob>(runtime.imports??[],(value,id)=>{const job=value as ImportJob;validateImportJob(job);if(job.id!==id||job.projectId!==document.projectId)throw fault('corrupt_recovery_image');return freeze(job);});
|
||||
for(const [id,entry]of importFiles(imports)){if(resources.has(id))throw fault('import_file_identity_conflict',[id]);if((entry.file.retirement||imports.get(entry.jobId)?.source?.previousFiles?.length)&&image.formatVersion!==4)throw fault('unsupported_recovery_version');}
|
||||
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
|
||||
const snapshot=Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history:restoredHistory, operations,
|
||||
runtime: Object.freeze({ imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
|
||||
|
||||
@@ -14,7 +14,7 @@ export function resourceRetentionPins(snapshot:EngineSnapshot,registry:Registry,
|
||||
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 job of runtime.imports?.values()??[])if(!['completed','partial','cancelled'].includes(job.status)){if(job.source)add(job.source.fileId,`import-source:${job.id}`);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');
|
||||
|
||||
@@ -4,7 +4,7 @@ import type { Json } from '../canvas-core/types.ts';
|
||||
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
|
||||
|
||||
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
|
||||
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
|
||||
export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4; 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 = {
|
||||
@@ -34,7 +34,7 @@ const keyFor = (scope: StorageScope) => {
|
||||
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]);
|
||||
};
|
||||
function validateImage(image: RecoveryImage) {
|
||||
if (![1,2,3].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||||
if (![1,2,3,4].includes(image.formatVersion)) throw fault('unsupported_recovery_version');
|
||||
if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image');
|
||||
}
|
||||
const requestValue = <T>(request: IDBRequest<T>) => new Promise<T>((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); });
|
||||
@@ -134,9 +134,13 @@ export function createBrowserStorage(options: { indexedDB?: IDBFactory; locks?:
|
||||
// Detach mutable callers and descriptors before the first await, including queued commits.
|
||||
const image=isSealedRecoveryImage(batch.image)?batch.image:freeze(clone(batch.image)),files=(batch.files??[]).map(file=>({id:file.id,bytes:file.bytes})),deleteFiles=[...(batch.deleteFiles??[])];if(deleteFiles.length>128||new Set(deleteFiles).size!==deleteFiles.length||deleteFiles.some(id=>typeof id!=='string'||!id||files.some(file=>file.id===id)))throw fault('invalid_file_deletion');if(!isSealedRecoveryImage(image))validateImage(image);
|
||||
if(deleteFiles.length){
|
||||
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');
|
||||
type FileProof={fileId?:string;retirement?:{version:number;unreferencedAt:number;purgedAt?:number}};
|
||||
const payload=image.payload as {runtime?:{resources?:[string,FileProof][];imports?:[string,{status:string;source?:FileProof&{previousFiles?:FileProof[]}}][]}};
|
||||
const proofs=new Map<string,FileProof>(),owned=new Set<string>();
|
||||
const own=(id:string,file:FileProof,eligible:boolean)=>{if(typeof id!=='string'||!id||owned.has(id))throw fault('invalid_file_deletion');owned.add(id);if(eligible)proofs.set(id,file);};
|
||||
for(const [id,resource]of payload?.runtime?.resources??[])own(id,resource,true);
|
||||
if(image.formatVersion===4)for(const [,job]of payload?.runtime?.imports??[]){if(job.source){own(job.source.fileId!,job.source,['completed','partial','cancelled'].includes(job.status));for(const file of job.source.previousFiles??[])own(file.fileId!,file,true);}}
|
||||
if(image.formatVersion<3||deleteFiles.some(id=>{const marker=proofs.get(id)?.retirement;return !marker||marker.version!==1||!Number.isSafeInteger(marker.unreferencedAt)||marker.unreferencedAt<0||!Number.isSafeInteger(marker.purgedAt)||marker.purgedAt!<marker.unreferencedAt;}))throw fault('invalid_file_deletion');
|
||||
}
|
||||
check(leader);
|
||||
const task=(writeTails.get(leader)??Promise.resolve()).catch(()=>{}).then(async()=>{
|
||||
|
||||
Reference in New Issue
Block a user