79 lines
13 KiB
JavaScript
79 lines
13 KiB
JavaScript
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();}
|
|
});
|