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

feat(canvas): persist native import jobs with atomic editor progress

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 15:40:10 +08:00
parent 160c16afb7
commit 72d6105e7c
6 changed files with 171 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/contracts.ts
+2
View File
@@ -1,3 +1,4 @@
import type { ImportJob } from './imports.ts';
import type { Document, Id, Json, OperationIntent, Snapshot } from '../canvas-core/types.ts';
import type { LedgerSnapshot, FrozenRequest } from './ledger.ts';
import type { CompiledRequest, OperationStatus } from './scheduler.ts';
@@ -6,6 +7,7 @@ import type { ResourceRecord } from './resources.ts';
import type { RecoveryAudit } from './conflicts.ts';
export type RuntimeState = {
readonly imports?: ReadonlyMap<Id, ImportJob>;
readonly ledger: LedgerSnapshot;
readonly runs: ReadonlyMap<Id, RunRecord>;
readonly acknowledged: Document;
src/lib/canvas-runtime/engine.ts
+62 -9
View File
@@ -1,3 +1,5 @@
import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts';
import { assertNativeImportDependencies, assertNativeImportTarget, prepareNativeImport } from './import-native.ts';
import { elisionBatchLimit, findElision } from './elision.ts';
import type { Document, Id, Registry, Result, RunInput, RunApplyInput, RunCancelInput, Json, OperationIntent, Data } from '../canvas-core/types.ts';
import { createEditor, type EditorOptions, type Transaction } from '../canvas-core/editor.ts';
@@ -52,9 +54,11 @@ export async function openProjectEngine(options: EngineOptions) {
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
let staged: readonly ResourceRecord[] = [];
let applying: RunApplyInput | undefined;
let importing: { job: ImportJob; expectedGeneration: number } | undefined;
const importRunners=new Map<Id,ReturnType<typeof createImportRunner>>(),importLoads=new Map<Id,Promise<void>>(),importHolds=new Map<symbol,Map<Id,()=>void>>();
let recovery: { controller: AbortController; release(): void; cancel():Promise<void> } | undefined;
const assertWritable = () => { if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); };
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ 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 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,
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 }; },
@@ -83,7 +87,9 @@ export async function openProjectEngine(options: EngineOptions) {
for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] ??= change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists';
ledger.add(freeze({ ...intent, dependencies: [...dependencies] }), basePresence);
}
return Object.freeze({ ...previous, resources: readMap(resources), ledger: ledger.getSnapshot() });
const imports=new Map(previous.imports??[]);
if(importing){if(imports.get(importing.job.id)?.generation!==importing.expectedGeneration)throw fault('stale_import_job');imports.set(importing.job.id,importing.job);}
return Object.freeze({ ...previous, imports:readMap(imports), resources: readMap(resources), ledger: ledger.getSnapshot() });
},
});
const pending = new Map<number, EngineSnapshot>();
@@ -94,7 +100,7 @@ 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)).map(([id, bytes]) => ({ id, bytes }));
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 });
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
durableRootRevision = revision; pending.delete(revision);
@@ -304,21 +310,67 @@ export async function openProjectEngine(options: EngineOptions) {
const result=editor.updateRuntime(previous=>{const ledger=createLedger(previous.ledger);ledger.add(freeze(intent),{});return Object.freeze({...previous,ledger:ledger.getSnapshot()});},root.rootRevision);
if(result.status==='rejected')return result;queueMicrotask(()=>{if(!closing&&!closed)scheduler.notify([id]);});return{status:'applied',revision:root.revision,operationIds:[id]};
}
function getImport(id:Id){const job=editor.getSnapshot().runtime.imports?.get(id);if(!job)throw fault('import_missing',[id]);return job;}
function publishImport(job:ImportJob,expectedGeneration:number){
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job);return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);
}
function holdImports(){const token=Symbol(),releases=new Map<Id,()=>void>();importHolds.set(token,releases);for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(job.status==='running'&&!importRunners.has(job.id))void loadImport(job.id);};}
function loadImport(id:Id):Promise<void>{
if(importLoads.has(id))return importLoads.get(id)!;
if(importRunners.has(id)){void importRunners.get(id)!.start();return Promise.resolve();}
const expectedFingerprint=getImport(id).sourceFingerprint;
const task=(async()=>{
const job=getImport(id),source=job.source;if(!source||source.kind!=='native-ndjson'||job.parserVersion!==1)throw fault('import_interpreter_unavailable');
const bytes=fileWrites.get(source.fileId)??await options.storage.readFile(leader,source.fileId);
if(closing||closed)return;
const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status))return;
if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed');
const runner=createImportRunner({optimisticLocal:true,source:createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}),getJob:()=>getImport(id),createId,
publish:publishImport,prepare:(line,job,reserve)=>prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry),persist,onError:report,
commit(units,next,expected){
assertWritable();if(importing)throw fault('reentrant_dispatch');const current=getImport(id);if(current.generation!==expected)throw fault('stale_import_job');
try{assertNativeImportTarget(current,editor.getSnapshot().document);for(const unit of units)assertNativeImportDependencies(current,editor.getSnapshot().document,unit.dependencies);}catch(error){return reject(error);}
importing={job:next,expectedGeneration:expected};
try{return editor.transaction(tx=>{for(const unit of units)for(const change of unit.changes){if(change.kind==='element'&&change.before===null&&change.after)tx.dispatch({type:'element.create',element:change.after});else if(change.kind==='canvas'&&change.before===null&&change.after)tx.dispatch({type:'canvas.create',canvas:change.after});else throw fault('invalid_native_import_unit');}},{label:'Import canvas content',groupId:`import:${id}`});}finally{importing=undefined;}
},
});
importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold());
if(getImport(id).status==='running')void runner.start();
})().catch(error=>{if(!closing&&!closed){try{const job=getImport(id);if(job.sourceFingerprint!==expectedFingerprint||['completed','partial','cancelled'].includes(job.status))return;publishImport(freeze({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{importLoads.delete(id);});
importLoads.set(id,task);return task;
}
const imports=Object.freeze({
getJobs:()=>readMap(editor.getSnapshot().runtime.imports??new Map()),
startNdjson(bytes:Blob,targetCanvasId:Id):Id{
assertWritable();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 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);
const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;
},
async pause(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(job.status==='running'){publishImport(freeze({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}},
async resume(id:Id){assertWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(job.status==='running'){await loadImport(id);return;}await importRunners.get(id)?.settle();await persist();await scheduler.resume();assertWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:'running' as const};delete next.error;publishImport(freeze(next),current.generation);await loadImport(id);},
async cancel(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','paused'].includes(job.status)){const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(freeze(next),job.generation);await persist();}}},
});
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
for(const job of initial.imports?.values()??[])if(job.status==='running')void loadImport(job.id);
return Object.freeze({
editor, sessions,
editor, sessions, imports,
/** Read/preview under a scheduling hold. No request is rewritten or declared resolved here. */
async prepareRecovery(): Promise<RecoverySession> {
assertWritable();
if(!options.persistence.read)throw fault('canonical_reader_unavailable');
const releaseScheduler=scheduler.hold(),releaseRuns=runTracker.hold(),controller=new AbortController();
const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();},async cancel(){controller.abort();active.release();}};
const releaseScheduler=scheduler.hold(),releaseRuns=runTracker.hold(),releaseImports=holdImports(),controller=new AbortController();
const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();releaseImports();},async cancel(){controller.abort();active.release();}};
recovery=active;
sessions.cancel('recovery');
const assertCurrent=()=>{if(closing||closed)throw fault('engine_closing');if(recovery!==active||controller.signal.aborted)throw fault('recovery_session_closed');};
try {
// Both holds are acquired before awaiting either lane; already admitted callbacks finish normally.
await Promise.all([scheduler.settle(),runTracker.settle()]);assertCurrent();
await Promise.all([...importLoads.values()]);assertCurrent();await Promise.all([scheduler.settle(),runTracker.settle(),...[...importRunners.values()].map(runner=>runner.settle())]);assertCurrent();
while(digestTasks.size){await Promise.all([...digestTasks.values()]);assertCurrent();}
await persist();assertCurrent();
const staged=stageAppliedRecovery(editor.getSnapshot(),{persistence:options.persistence,runs:options.runs,assets:options.assets,now:runClock.now()});
@@ -445,7 +497,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))) 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) || [...(editor.getSnapshot().runtime.imports?.values()??[])].some(job=>job.source?.fileId===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) })));
@@ -468,10 +520,11 @@ export async function openProjectEngine(options: EngineOptions) {
cancelRecovery:async()=>{await recovery?.cancel();},
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
retry: async () => { if(recovery)await recovery.cancel();assertWritable();await scheduler.resume();await runTracker.retry(); },
async settle() { await Promise.resolve(); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
async settle() { await Promise.resolve();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; sessions.dispose(); scheduler.stop(); runTracker.stop();
await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]);
recovery?.controller.abort();recovery?.release();
for (const release of [...displays]) release();
// Failure keeps the leader/recovery material available for an explicit retry of close.
src/lib/canvas-runtime/recovery.ts
+7 -4
View File
@@ -1,3 +1,4 @@
import { validateImportJob, type ImportJob } from './imports.ts';
import { validateElisions, withLocalLifecycles } from './elision.ts';
import type { Change, CoreSnapshot, Document, HistoryEntry, HistoryState, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
import { validateHistoryReplay, replayHistoryEvidence, sameHistory } from './history-replay.ts';
@@ -18,10 +19,10 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
document: documentValue(snapshot.document),
history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: [...snapshot.history.entries] },
operations: [...snapshot.operations],
runtime: { recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
runtime: { 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] } },
})) as Json;
if (!validJson(payload)) throw fault('invalid_recovery_payload');
return freeze({ formatVersion: 1, rootRevision: snapshot.rootRevision, localRevision: snapshot.revision, payload });
return freeze({ formatVersion: snapshot.runtime.imports?.size ? 2 : 1, rootRevision: snapshot.rootRevision, localRevision: 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; }
@@ -84,13 +85,14 @@ 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 (image.formatVersion !== 1) throw fault('unsupported_recovery_version');
if (![1,2].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.
const raw = object(JSON.parse(JSON.stringify(image.payload))), document = readDocument(raw.document, registry), history = object(raw.history), runtime = object(raw.runtime), ledger = object(runtime.ledger);
const operations = table(raw.operations, intent),restoredHistory=readHistory(history,operations);
for(const operation of operations.values())if(operation.historyTransition?.version===1&&(operation.historyTransition.rootRevision<1||operation.historyTransition.rootRevision>image.rootRevision))throw fault('corrupt_recovery_image');
if(image.formatVersion===1&&array(runtime.imports??[]).length)throw fault('unsupported_recovery_version');
const records = table<OperationRecord>(ledger.operations, (value, id) => {
const raw = object(value); intent(raw.intent, id); strings(raw.blockers); object(raw.basePresence);
if (((object(raw.intent).operationType === 'document.change'||object(raw.intent).operationType==='run.apply'&&array(object(raw.intent).changes).length>0) && !operations.has(id)) || !['pending', 'frozen', 'resolved', 'blocked'].includes(string(raw.phase))) throw fault('corrupt_recovery_image');
@@ -221,8 +223,9 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
for(const value of array(raw.outputEvents ?? [])){const event=object(value),step=string(event.step_id);if(event.run_id!==id||event.type!=='step.output'||outputIds.has(step)||integer(event.seq,1)>Math.max(integer(raw.cursor),raw.pendingPage===undefined?0:integer(object(raw.pendingPage).next)))throw fault('corrupt_recovery_image');object(event.output);outputIds.add(step);}
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);});
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
return Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history:restoredHistory, operations,
runtime: Object.freeze({ recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
runtime: Object.freeze({ imports, recoveries, runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts, conflicts: table(runtime.conflicts, string) }),
});
}
src/lib/canvas-runtime/storage.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import type { Json } from '../canvas-core/types.ts';
import { clone, equal, fault, validJson } from '../canvas-core/values.ts';
export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string };
export type RecoveryImage = { readonly formatVersion: 1; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json };
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 StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise<void> };
export type StoragePort = {
@@ -23,7 +23,7 @@ const keyFor = (scope: StorageScope) => {
return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId]);
};
function validateImage(image: RecoveryImage) {
if (image.formatVersion !== 1) throw fault('unsupported_recovery_version');
if (![1,2].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); });
tests/headless/browser-storage-cases.mjs
+20 -1
View File
@@ -294,6 +294,23 @@ async function resourceRecoveryCase(name, assert) {
}
async function nativeImportRecoveryCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),original=createDocument(root,[{id:root,parent:null}],[]),database=`${name}-native-import`,storage=createBrowserStorage({name:database}),calls=[];let document=original,versions=new Map([[root,1]]),engine;
const model=(complete,participants=[...versions.keys()])=>({documentFormatVersion:1,projectId:root,complete,removedCanvasIds:participants.filter(id=>!document.canvases.has(id)),canvases:[...document.canvases.values()].filter(canvas=>participants.includes(canvas.id)),elements:[...document.elements.values()].filter(element=>participants.includes(element.ownerCanvasId)),tombstones:Object.fromEntries(document.tombstones),versions:Object.fromEntries(participants.map(id=>[id,versions.get(id)]))});
const transport=async request=>{calls.push(request);const body=JSON.parse(request.body);for(const[id,version]of Object.entries(body.expected_versions))assert(version===(versions.get(id)??0),'import lost the version fence');for(const operation of body.operations)document=applyChanges(document,operation.changes);for(const id of Object.keys(body.expected_versions))versions.set(id,(versions.get(id)??0)+1);return{status:200,body:{protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:model(false,Object.keys(body.expected_versions))}};};
const node=id=>({kind:'element',key:id,record:{id,incarnation:5,structure:'node',ownerCanvasId:'$target',componentId:'text',dataVersion:1,data:{title:id,text:'',taskId:'not copied'},geometry:{x:0,y:0,width:200,height:100,rotation:0}}}),edge={kind:'element',key:'edge',record:{id:'edge',incarnation:1,structure:'connection',componentId:'connection',dataVersion:1,ownerCanvasId:'$target',source:{elementId:'first'},target:{elementId:'second'},path:null,data:{}}},text=[{...node('first'),record:{...node('first').record,componentId:'frame'}},edge,{...node('second'),record:{...node('second').record,ownerCanvasId:'child'}},{kind:'canvas',key:'child',record:{id:'child',parent:{canvasId:'$target',entryElementId:'first'}}}].map(value=>JSON.stringify(value)).join('\n')+'\n{bad}';
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
try{
engine=await openProjectEngine(options);const id=engine.imports.startNdjson(new Blob([text]),root);
await new Promise(resolve=>{const release=engine.editor.subscribe((_,snapshot)=>{if(snapshot.runtime.imports.get(id).committedSourceKeys.length===1){release();resolve();}});});await engine.imports.pause(id);await engine.settle();
const first=engine.imports.getJobs().get(id).stableIds['element:first'],fileId=engine.imports.getJobs().get(id).source.fileId;engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'user edit'}});await engine.settle();await engine.close();engine=await openProjectEngine(options);
assert(engine.imports.getJobs().get(id).status==='paused','paused import did not survive IndexedDB');await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);
assert(job.status==='partial'&&job.committedSourceKeys.length===4&&job.failures.length===1,'partial native import lost progress or failure report');assert(job.stableIds['element:first']===first&&document.elements.get(first).data.title==='user edit','import reallocated an identity or overwrote user content');assert(document.canvases.size===2&&document.elements.size===3&&engine.getSnapshot().runtime.runs.size===0,'native import duplicated an object or launched a run');assert(!Object.hasOwn(document.elements.get(first).data,'taskId'),'native import retained execution metadata');
const leaderRead=await (async()=>{await engine.close();const leader=await storage.claim(options.scope);try{return await storage.readFile(leader,fileId);}finally{await leader.release();}})();assert(await leaderRead.text()===text,'original source bytes were not durably stored');engine=await openProjectEngine(options);
assert(engine.editor.history.undo().status==='applied','import batch undo failed');await engine.settle();assert(document.elements.size===1&&document.elements.has(first),'undo did not preserve the interrupted user edit');await engine.close();engine=await openProjectEngine(options);await engine.settle();assert(document.elements.size===1&&engine.getSnapshot().document.elements.size===1,'terminal job revived records removed by undo');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
}
async function elisionRecoveryCase(name,assert){
for(const structural of [false,true]){
const root=crypto.randomUUID(),child=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),node=(id,ownerCanvasId=root)=>({id,incarnation:1,ownerCanvasId,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:''},geometry:{x:0,y:0,width:200,height:100,rotation:0}}),original=createDocument(root,[{id:root,parent:null}],[node('a')]),database=`${name}-elision-${structural}`,storage=createBrowserStorage({name:database}),calls=[];let document=original,versions=new Map([[root,1]]),engine;
@@ -421,7 +438,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: 2 } }), 'unsupported_recovery_version');
await rejects(() => storage.commit(leader, { image: { ...image(2), formatVersion: 3 } }), '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.
@@ -464,6 +481,8 @@ export async function run() {
results.push('project engine reopens real IndexedDB with one history unit and the original request; ACK resolves without duplicate edit');
await resourceRecoveryCase(name, assert);
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
await nativeImportRecoveryCase(name,assert);
results.push('native import source/cursor/IDs survive real IndexedDB pause/reopen/resume, preserve user edits, report partial failures and do not revive undone records');
await elisionRecoveryCase(name,assert);
results.push('unsent node/partition create-delete lifetimes make zero requests, retain local proofs through IndexedDB/ACK/fresh reads, and undo restores new incarnations');
await batchRecoveryCase(name,assert);
tests/headless/imports-engine.test.mjs
+78
View File
@@ -0,0 +1,78 @@
import {test} from 'node:test';
import assert from 'node:assert/strict';
import {createDocument,createRegistry,builtinComponents,applyChanges} from '../../src/lib/canvas-core/index.ts';
import {openProjectEngine} from '../../src/lib/canvas-runtime/engine.ts';
import {decodeRecovery} from '../../src/lib/canvas-runtime/recovery.ts';
const registry=createRegistry(builtinComponents()),original=createDocument('project',[{id:'canvas',parent:null}],[]),scope={accountId:'account',workspaceId:'workspace',projectId:'project'};
const node=id=>({kind:'element',key:id,record:{id,incarnation:8,structure:'node',ownerCanvasId:'$target',componentId:'text',dataVersion:1,data:{title:id,text:'',runId:'must not run'},geometry:{x:0,y:0,width:100,height:100,rotation:0}}});
const file=records=>new Blob([records.map(value=>typeof value==='string'?value:JSON.stringify(value)).join('\n')]);
const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};};
function fixture(){
let image=null,leader,next=0,server=original,version=1;const files=new Map(),requests=[],receipts=new Map(),errors=[];
const f={files,requests,errors,gate:null,broken:false,
storage:{async claim(){assert.equal(leader,undefined);const value={scope,fence:1,active:true,async release(){leader=undefined;value.active=false;}};leader=value;return value;},async recover(){return image&&structuredClone(image);},async commit(owner,batch){assert.equal(owner,leader);if(f.broken)throw Error('quota');for(const part of batch.files??[])files.set(part.id,part.bytes);image=structuredClone(batch.image);return{durableRootRevision:batch.image.rootRevision};},async readFile(owner,id){assert.equal(owner,leader);return files.get(id);},close(){}},
port:{compile(intent,requestId,snapshot){return{payload:{requestId,intent},canvasIds:intent.canvasIds,expectedVersions:Object.fromEntries(intent.canvasIds.map(id=>[id,snapshot.runtime.versions.get(id)??0]))};},async submit(request){requests.push(request);if(f.gate)await f.gate.promise;const{intent}=JSON.parse(request.body);if(!receipts.has(request.id)){server=applyChanges(server,intent.changes);version++;receipts.set(request.id,{document:server,versions:new Map([['canvas',version]])});}return{outcome:'applied',canonical:{requestId:request.id}};},async query(request){return receipts.has(request.id)?{outcome:'applied',canonical:{requestId:request.id}}:{outcome:'unresolved'};},canonical(request){return receipts.get(request.id);},async read(){return{document:server,versions:new Map([['canvas',version]])};}},
image:()=>image,server:()=>server,open:()=>openProjectEngine({scope,registry,document:original,versions:new Map([['canvas',1]]),storage:f.storage,persistence:f.port,createId:()=>`identity-${++next}`,onError:error=>errors.push(error)})};return f;
}
function until(engine,predicate){if(predicate(engine.getSnapshot()))return Promise.resolve();return new Promise((resolve,reject)=>{const timer=setTimeout(()=>{unsubscribe();reject(Error('import condition timed out'));},5000),unsubscribe=engine.editor.subscribe((_,snapshot)=>{if(predicate(snapshot)){clearTimeout(timer);unsubscribe();resolve();}});});}
test('native import adds visible records while its backend writer is still awaiting ACK',async()=>{
const f=fixture(),engine=await f.open(),gate=deferred();f.gate=gate;
try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.server().elements.size,0);assert.equal(engine.getSnapshot().history.undo.length,1);assert.equal(f.files.size,1);for(const element of engine.getSnapshot().document.elements.values()){assert.equal(element.incarnation,1);assert.equal(element.data.runId,undefined);}gate.resolve();await engine.settle();assert.equal(f.server().elements.size,3);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);}finally{gate.resolve();await engine.close();}
});
test('pause/reopen/resume reuses the durable source and IDs without overwriting a user edit',async()=>{
const f=fixture();let engine=await f.open();
try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:one'];assert.equal(job.status,'paused');engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'user edit'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'user edit');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);}finally{await engine.close();}
});
test('cancel keeps the already added prefix and survives reopening without further insertion',async()=>{
const f=fixture();let engine=await f.open();
try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.cancel(id);await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,1);await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);}finally{await engine.close();}
});
test('deleting an imported endpoint while paused prevents a late edge from reviving it',async()=>{
const f=fixture();let engine=await f.open();
const edge={kind:'element',key:'edge',record:{id:'edge',incarnation:1,structure:'connection',componentId:'connection',dataVersion:1,ownerCanvasId:'$target',source:{elementId:'one'},target:{elementId:'two'},path:null,data:{}}};
try{const id=engine.imports.startNdjson(file([node('one'),edge,node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:one'];engine.editor.dispatch({type:'element.remove',target:{id:first,incarnation:1}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.equal(engine.getSnapshot().document.elements.has(first),false);assert.equal(engine.getSnapshot().document.elements.size,1);assert.ok(job.failures.some(failure=>failure.key==='element:edge'));assert.equal(f.server().elements.has(first),false);}finally{await engine.close();}
});
test('canonical recovery holds active imports and releases them after accepting the fresh base',async()=>{
const f=fixture(),engine=await f.open();
try{const id=engine.imports.startNdjson(file(Array.from({length:20},(_,index)=>node(`n-${index}`))),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length>=1);const session=await engine.prepareRecovery(),cut=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,cut);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,20);assert.equal(f.errors.length,0);}finally{await engine.close();}
});
test('closing an active import preserves a running checkpoint and reopening resumes without duplicate records',async()=>{
const f=fixture();let engine=await f.open();
try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);const first=engine.imports.getJobs().get(id).stableIds['element:one'];await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.errors.length,0);}finally{await engine.close();}
});
test('a late connection cannot claim an endpoint restored into a new incarnation',async()=>{
const f=fixture(),engine=await f.open(),edge={kind:'element',key:'edge',record:{id:'edge',incarnation:1,structure:'connection',componentId:'connection',dataVersion:1,ownerCanvasId:'$target',source:{elementId:'one'},target:{elementId:'two'},path:null,data:{}}};
try{const id=engine.imports.startNdjson(file([node('one'),edge,node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:one'];engine.editor.dispatch({type:'element.remove',target:{id:first,incarnation:1}});await engine.settle();assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get(first).incarnation,2);await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.equal(engine.getSnapshot().document.elements.get(first).incarnation,2);assert.equal(engine.getSnapshot().document.elements.has(job.stableIds['element:edge']),false);assert.ok(job.failures.some(failure=>failure.code==='import_dependency_changed'));}finally{await engine.close();}
});
test('a late unavailable-source result cannot change a job cancelled during reopening',async()=>{
const f=fixture();let engine=await f.open();const gate=deferred(),entered=deferred();
try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.close();f.storage.readFile=async()=>{entered.resolve();return gate.promise;};engine=await f.open();await entered.promise;await engine.imports.cancel(id);gate.resolve(undefined);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(f.errors.length,0);}finally{gate.resolve(undefined);await engine.close();}
});
test('import-bearing recovery cuts use format 2 and cannot be silently downgraded to a legacy image',async()=>{
const f=fixture(),engine=await f.open();
try{await engine.persist();assert.equal(f.image().formatVersion,1);const id=engine.imports.startNdjson(file([node('one')]),'canvas');await engine.settle();const image=f.image();assert.equal(image.formatVersion,2);assert.equal(decodeRecovery(image,registry).runtime.imports.get(id).status,'completed');assert.throws(()=>decodeRecovery({...image,formatVersion:1},registry),/unsupported_recovery_version/);}finally{await engine.close();}
});
test('the first valid import record displays while the source write is pending, but no backend request crosses that barrier',async()=>{
const f=fixture(),engine=await f.open(),gate=deferred(),entered=deferred(),commit=f.storage.commit;f.storage.commit=async(owner,batch)=>{if(batch.files?.length){entered.resolve();await gate.promise;}return commit(owner,batch);};
try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await entered.promise;await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length>=1);assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.requests.length,0);assert.equal(f.files.size,0);gate.resolve();await engine.settle();assert.equal(f.files.size,1);assert.equal(f.server().elements.size,2);assert.equal(engine.imports.getJobs().get(id).status,'completed');}finally{gate.resolve();await engine.close();}
});
test('resuming after a failed source write preserves IDs and wakes the original durable writer',async()=>{
const f=fixture(),engine=await f.open();
try{await engine.persist();f.broken=true;const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const first=engine.imports.getJobs().get(id).stableIds['element:one'];assert.equal(f.requests.length,0);assert.equal(engine.getSnapshot().document.elements.size,1);f.broken=false;await engine.imports.resume(id);await engine.settle();await until(engine,()=>f.server().elements.size===2);assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(f.files.size,1);assert.equal(new Set(f.requests.map(request=>request.id)).size,f.requests.length);}finally{f.broken=false;await engine.close();}
});