feat(canvas): persist and execute deep duplicate tasks
This commit is contained in:
@@ -307,6 +307,24 @@ async function resourceRecoveryCase(name, assert) {
|
||||
}
|
||||
|
||||
|
||||
async function duplicateRecoveryCase(name,assert){
|
||||
const root=crypto.randomUUID(),child=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),database=`${name}-duplicate`,storage=createBrowserStorage({name:database});
|
||||
const node=(id,owner=root,componentId='text')=>({id,incarnation:1,structure:'node',ownerCanvasId:owner,componentId,dataVersion:1,data:{title:id,text:'',...(componentId==='image'?{assetId:'shared',src:'asset:shared',runId:'do-not-run'}:{})},geometry:{x:0,y:0,width:200,height:100,rotation:0}});
|
||||
const edge={id:'incoming',incarnation:1,structure:'connection',ownerCanvasId:root,componentId:'connection',dataVersion:1,source:{elementId:'C'},target:{elementId:'F'},path:null,data:{title:''}},original=createDocument(root,[{id:root,parent:null},{id:child,parent:{canvasId:root,entryElementId:'F'}}],[node('C'),node('F',root,'frame'),node('A',child,'image'),edge]);
|
||||
let document=original,versions=new Map([[root,1],[child,1]]),engine;
|
||||
const model=participants=>({documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],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=>{const body=JSON.parse(request.body);for(const[id,version]of Object.entries(body.expected_versions))assert(version===(versions.get(id)??0),'copy lost expected version');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(Object.keys(body.expected_versions))}};};
|
||||
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map(versions),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
|
||||
try{
|
||||
engine=await openProjectEngine(options);const id=await engine.imports.duplicate([{id:'F',incarnation:1},{id:'A',incarnation:1}]);
|
||||
await new Promise(resolve=>{if(engine.imports.getJobs().get(id).committedSourceKeys.length){resolve();return;}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 before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert(before.status==='paused','copy pause lost');engine.editor.dispatch({type:'element.data',target:{id:copied,incarnation:1},patch:{title:'edited copy'}});await engine.settle();await engine.close();engine=await openProjectEngine(options);
|
||||
const restored=engine.imports.getJobs().get(id);assert(restored.source.kind==='duplicate-ndjson'&&JSON.stringify(restored.stableIds)===JSON.stringify(before.stableIds),'copy source or mapping changed on reopen');await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id),image=document.elements.get(job.stableIds['element:A']),incoming=document.elements.get(job.stableIds['element:incoming']);
|
||||
assert(job.status==='completed'&&job.failures.length===0,'copy did not complete after reopening');assert(document.elements.get(copied).data.title==='edited copy','copy replay overwrote an edit');assert(image.data.assetId==='shared'&&!Object.hasOwn(image.data,'runId'),'copy reissued image execution');assert(incoming.source.elementId==='C'&&incoming.target.elementId===copied&&incoming.ownerCanvasId===root,'copy lost upper incoming source');assert(document.elements.has('incoming')&&document.canvases.has(job.stableIds[`canvas:${child}`]),'copy changed originals or lost its child canvas');
|
||||
assert(engine.editor.history.undo().status==='applied','copy undo failed');await engine.settle();const count=document.elements.size;await engine.close();engine=await openProjectEngine(options);await engine.settle();assert(document.elements.size===count&&!document.elements.has(job.stableIds['element:A']),'terminal copy revived undone records');
|
||||
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
|
||||
}
|
||||
|
||||
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)]))});
|
||||
@@ -496,6 +514,8 @@ export async function run() {
|
||||
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
|
||||
await cooperativeCopyCase(assert);
|
||||
results.push('cooperative duplicate planning yields to actual browser tasks and cancellation without mutating the source');
|
||||
await duplicateRecoveryCase(name,assert);
|
||||
results.push('deep copy source/mapping survive IndexedDB pause/reopen/resume, retain C to F-prime and shared assets, preserve edits and do not revive undone content');
|
||||
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);
|
||||
|
||||
@@ -8,11 +8,11 @@ const node=id=>({kind:'element',key:id,record:{id,incarnation:8,structure:'node'
|
||||
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=[];
|
||||
let image=null,leader,next=0,server=original;const versions=new Map([['canvas',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;
|
||||
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);for(const canvasId of intent.canvasIds)versions.set(canvasId,(versions.get(canvasId)??0)+1);receipts.set(request.id,{document:server,versions:new Map(versions)});}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(versions)};}},
|
||||
image:()=>image,server:()=>server,open:(extra={})=>openProjectEngine({scope,registry,document:original,versions:new Map([['canvas',1]]),storage:f.storage,persistence:f.port,createId:()=>`identity-${++next}`,onError:error=>errors.push(error),...extra})};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();}});});}
|
||||
|
||||
@@ -76,3 +76,52 @@ test('resuming after a failed source write preserves IDs and wakes the original
|
||||
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();}
|
||||
});
|
||||
|
||||
const copyNode=(id,owner='canvas',componentId='text')=>({id,incarnation:1,structure:'node',ownerCanvasId:owner,componentId,dataVersion:1,data:{title:id,text:'source',...(componentId==='image'?{assetId:'shared',src:'asset:shared',runId:'paid-run',taskId:'task'}:{})},geometry:{x:0,y:0,width:200,height:100,rotation:0}});
|
||||
async function seedCopy(engine){
|
||||
const records=[copyNode('C'),copyNode('F','canvas','frame'),copyNode('A','child','image'),copyNode('B','child')];
|
||||
const edge={id:'incoming',incarnation:1,structure:'connection',ownerCanvasId:'canvas',componentId:'connection',dataVersion:1,source:{elementId:'C'},target:{elementId:'F'},path:null,data:{title:''}};
|
||||
const result=engine.editor.transaction(tx=>{tx.dispatch({type:'element.create',element:records[0]});tx.dispatch({type:'element.create',element:records[1]});tx.dispatch({type:'canvas.create',canvas:{id:'child',parent:{canvasId:'canvas',entryElementId:'F'}}});for(const element of records.slice(2))tx.dispatch({type:'element.create',element});tx.dispatch({type:'element.create',element:edge});},{label:'Seed copy source'});assert.equal(result.status,'applied');await engine.settle();
|
||||
}
|
||||
|
||||
test('deep copy source/mapping recover and preserve upper incoming source, shared assets and user edits',async()=>{
|
||||
const f=fixture();let engine=await f.open();
|
||||
try{
|
||||
await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1},{id:'A',incarnation:1}]);
|
||||
await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
|
||||
const before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert.equal(before.source.kind,'duplicate-ndjson');assert.equal(before.source.copy.roots.length,1);assert.equal(before.status,'paused');
|
||||
engine.editor.dispatch({type:'element.data',target:{id:copied,incarnation:1},patch:{title:'edited copy'}});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),document=engine.getSnapshot().document,ids=job.stableIds;assert.equal(job.status,'completed');assert.deepEqual(ids,before.stableIds);assert.equal(document.elements.get(copied).data.title,'edited copy');
|
||||
assert.equal(document.canvases.get(ids['canvas:child']).parent.entryElementId,copied);assert.equal(document.elements.get(ids['element:A']).ownerCanvasId,ids['canvas:child']);assert.equal(document.elements.get(ids['element:A']).data.assetId,'shared');assert.equal(document.elements.get(ids['element:A']).data.runId,undefined);assert.equal(document.elements.get(ids['element:A']).data.taskId,undefined);
|
||||
const edge=document.elements.get(ids['element:incoming']);assert.equal(edge.source.elementId,'C');assert.equal(edge.target.elementId,copied);assert.equal(edge.ownerCanvasId,'canvas');assert.ok(document.elements.has('incoming'));assert.equal(job.failures.length,0);assert.equal(f.errors.length,0);assert.ok(f.server().elements.has(ids['element:incoming']));decodeRecovery(f.image(),registry);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('late copied edge cannot attach to an external endpoint deleted and restored as a new incarnation',async()=>{
|
||||
const f=fixture(),engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
|
||||
assert.equal(engine.editor.dispatch({type:'element.remove',target:{id:'C',incarnation:1}}).status,'applied');assert.equal(engine.editor.history.undo().status,'applied');assert.equal(engine.getSnapshot().document.elements.get('C').incarnation,2);await engine.settle();await engine.imports.resume(id);await engine.settle();
|
||||
const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(failure=>failure.code==='copy_external_changed'));assert.equal(engine.getSnapshot().document.elements.has(job.stableIds['element:incoming']),false);assert.ok(engine.getSnapshot().document.elements.has(job.stableIds['element:A']));
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('copy cancellation retains the published prefix and does not resume on reopening',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const initial=engine.getSnapshot().document.elements.size,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.cancel(id);await engine.settle();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,initial+1);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('preparation abort and project close do not admit a late copy job',async()=>{
|
||||
const f=fixture(),engine=await f.open();await seedCopy(engine);const controller=new AbortController();controller.abort();await assert.rejects(engine.imports.duplicate([{id:'F',incarnation:1}],{signal:controller.signal}),/copy_aborted/);assert.equal(engine.imports.getJobs().size,0);
|
||||
const pending=engine.imports.duplicate([{id:'F',incarnation:1}]);const rejected=assert.rejects(pending,/copy_aborted|engine_closing/);await engine.close();await rejected;const reopened=await f.open();try{assert.equal(reopened.imports.getJobs().size,0);}finally{await reopened.close();}
|
||||
});
|
||||
|
||||
test('copy publishes independently of backend ACK and associates task visibility with its initiating view',async()=>{
|
||||
const f=fixture(),engine=await f.open(),gate=deferred();try{await seedCopy(engine);f.gate=gate;const before=f.server().elements.size,id=await engine.imports.duplicate([{id:'A',incarnation:1}],{viewCanvasId:'canvas'});await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='completed');const job=engine.imports.getJobs().get(id);assert.equal(job.source.targetCanvasId,'canvas');assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:A']).ownerCanvasId,'child');assert.equal(f.server().elements.size,before);gate.resolve();await engine.settle();assert.equal(f.server().elements.size,before+1);assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
|
||||
});
|
||||
|
||||
test('copy source write failure gates network and resume preserves the planned identities',async()=>{
|
||||
const f=fixture(),engine=await f.open();try{await seedCopy(engine);const sent=f.requests.length;f.broken=true;const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const before=engine.imports.getJobs().get(id);assert.equal(f.requests.length,sent);f.broken=false;await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.deepEqual(engine.imports.getJobs().get(id).stableIds,before.stableIds);}finally{f.broken=false;await engine.close();}
|
||||
});
|
||||
|
||||
|
||||
test('read-only policy rejects copy/import admission without storing a source or a running job',async()=>{
|
||||
const f=fixture(),engine=await f.open({policy:()=>({enabled:false,code:'read_only',targets:[]})});try{await assert.rejects(engine.imports.duplicate([{id:'missing',incarnation:1}]),/read_only/);assert.throws(()=>engine.imports.startNdjson(file([node('one')]),'canvas'),/read_only/);assert.equal(engine.imports.getJobs().size,0);await engine.settle();assert.equal(f.files.size,0);assert.equal(f.requests.length,0);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
@@ -240,6 +240,14 @@ try {
|
||||
for(let i=0;i<200&&documents.get(root).elements.size!==beforeImport+1;i++)await pause(20);assert.equal(documents.get(root).elements.size,beforeImport+1);await pause(200);await call('Page.reload');await until(`document.querySelectorAll('.react-flow__node').length===6&&document.querySelector('[aria-label="画布导入"]')?.textContent.includes('部分加入')`,'import task and content did not recover after reload');
|
||||
await click('撤销');await until(`document.querySelectorAll('.react-flow__node').length===5`,'import undo did not remove its successful content');await pause(200);await call('Page.reload');await until(`document.querySelectorAll('.react-flow__node').length===5`,'import recovered an undone record');
|
||||
console.log('PASS actual NDJSON import displays before backend ACK, reports a failed record, restores task/source state and never revives undone content');
|
||||
await evaluate(`document.querySelector('.react-flow__node[data-id=b]').dispatchEvent(new MouseEvent('click',{bubbles:true}))`);
|
||||
await until(`!document.querySelector('[aria-label="深复制选中内容"]').disabled`,'copy selection button remained disabled');
|
||||
await click('深复制选中内容');await until(`document.querySelectorAll('.react-flow__node').length===6&&document.querySelector('[aria-label="画布导入"]').textContent.includes('复制选中内容')`,'copy did not publish through the durable task pipeline');
|
||||
await until(`document.querySelector('[aria-label="深复制选中内容"]')&&!document.querySelector('[aria-label="正在准备复制"]')`,'copy preparation did not finish');
|
||||
await pause(300);await call('Page.reload');await until(`document.querySelectorAll('.react-flow__node').length===6&&document.querySelector('[aria-label="画布导入"]').textContent.includes('复制选中内容')`,'copy task did not recover after reload');
|
||||
await click('撤销');await until(`document.querySelectorAll('.react-flow__node').length===5`,'copy history did not undo its published content');await pause(200);await call('Page.reload');await until(`document.querySelectorAll('.react-flow__node').length===5`,'copy replay revived an undone record');
|
||||
console.log('PASS actual deep copy toolbar action publishes a durable task, survives reload and does not revive its undone content');
|
||||
|
||||
await click('账户和工作空间');await until(`!!document.querySelector('#workspace-account:popover-open')`,'account menu did not open');await evaluate(`[...document.querySelectorAll('button')].find(button=>button.textContent.trim()==='退出登录').click()`);
|
||||
await until(`location.pathname==='/login'&&!document.body.innerText.includes('This page couldn’t load')`,'logout did not safely detach the view');
|
||||
console.log('PASS logout retires the runtime before dropping its view without a render error');
|
||||
|
||||
Reference in New Issue
Block a user