185 lines
27 KiB
JavaScript
185 lines
27 KiB
JavaScript
import {test} from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import {createDocument,applyChanges,validateDocument,createRegistry,builtinComponents,validJson,fault} from '../../src/lib/canvas-core/index.ts';
|
|
import {createImportJob,validateImportJob,createNdjsonImportSource,createImportRunner} from '../../src/lib/canvas-runtime/imports.ts';
|
|
const registry=createRegistry(builtinComponents()),base=createDocument('project',[{id:'canvas',parent:null}],[]);
|
|
const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};};
|
|
const ticks=async()=>{for(let i=0;i<30;i++)await Promise.resolve();};
|
|
const bytes=records=>new Blob([records.map(record=>typeof record==='string'?record:JSON.stringify(record)).join('\n')]);
|
|
function fixture(records,options={}){
|
|
const source=options.source??createNdjsonImportSource({bytes:bytes(records),fingerprint:'immutable-source'}),calls=[],errors=[],saved=[],commits=[];let n=0,time=0,root=options.root??{document:base,job:createImportJob({id:'import',projectId:'project',sourceFingerprint:source.fingerprint,parserVersion:1})};
|
|
const f={source,calls,errors,saved,commits,get root(){return root;},set root(value){root=value;},ports:{source,getJob:()=>root.job,createId:()=>`new-${++n}`,now:()=>time,
|
|
publish(job,expected){assert.equal(root.job.generation,expected);root={...root,job};},
|
|
prepare(line,job,reserve){time+=2;const input=line.value;if(!input||typeof input!=='object'||!input.id)throw fault('invalid_record');const id=reserve(input.id);return{key:input.id,dependencies:input.dependencies??[],changes:[{kind:'element',id,before:null,after:{id,incarnation:1,structure:'node',ownerCanvasId:'canvas',componentId:'text',dataVersion:1,data:{title:input.title??input.id,text:''},geometry:{x:0,y:0,width:input.bad?-1:100,height:100,rotation:0}}}]};},
|
|
commit(units,job,expected){time+=2;assert.equal(root.job.generation,expected);calls.push(units.map(unit=>unit.key));const before=root;try{const document=applyChanges(root.document,units.flatMap(unit=>unit.changes));validateDocument(document,registry);root={document,job};commits.push(root);return{status:'applied',revision:commits.length,operationIds:[]};}catch(error){assert.equal(root,before);return{status:'rejected',code:error.code??'invalid',targets:[]};}},
|
|
async persist(){validateImportJob(root.job);assert.equal(validJson(root.job),true);saved.push(structuredClone({job:root.job,document:{projectId:root.document.projectId,canvases:[...root.document.canvases.values()],elements:[...root.document.elements.values()],tombstones:[...root.document.tombstones]}}));},
|
|
onError:error=>errors.push(error),yieldTask:async()=>{f.yields++;},...options.ports},yields:0};
|
|
f.runner=createImportRunner(f.ports);return f;
|
|
}
|
|
|
|
test('NDJSON parses bounded UTF-8 chunks, seeks by byte cursor and reports malformed/oversized records without losing the next line',async()=>{
|
|
const source=createNdjsonImportSource({bytes:bytes([{id:'中文 😀'},'{bad}','x'.repeat(30),{id:'ok'}]),fingerprint:'file-1',maxRecordBytes:24,chunkBytes:3}),signal=new AbortController().signal;
|
|
let cursor=0;const rows=[];for(;;){const row=await source.read(cursor,signal);if(!row)break;rows.push(row);cursor=row.nextCursor;}
|
|
assert.equal(rows[0].value.id,'中文 😀');assert.equal(rows[1].error,'invalid_json');assert.equal(rows[2].error,'record_too_large');assert.equal(rows[3].value.id,'ok');assert.deepEqual(await source.read(rows[0].cursor,signal),rows[0]);
|
|
const input={bytes:bytes([{id:'old'}]),fingerprint:'fixed'},owned=createNdjsonImportSource(input);input.bytes=bytes([{id:'replacement'}]);assert.equal((await owned.read(0,signal)).value.id,'old');
|
|
});
|
|
|
|
test('records become visible before EOF, task yields real scheduling points and cancellation preserves the committed prefix',async()=>{
|
|
const gate=deferred(),entered=deferred(),raw=bytes([{id:'first'},{id:'second'}]),initial=createNdjsonImportSource({bytes:raw,fingerprint:'immutable-source'});let firstEnd;
|
|
firstEnd=(await initial.read(0,new AbortController().signal)).nextCursor;
|
|
const source={...initial,async read(cursor,signal){if(cursor===firstEnd){entered.resolve();await gate.promise;}return initial.read(cursor,signal);}},f=fixture([],{source,ports:{frameBudgetMs:3}});
|
|
const running=f.runner.start();await entered.promise;assert.equal(f.root.document.elements.size,1);assert.equal(f.root.job.committedSourceKeys[0],'first');assert.ok(f.yields>0);
|
|
await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.root.document.elements.size,1);assert.deepEqual(f.root.job.committedSourceKeys,['first']);await f.runner.close();
|
|
});
|
|
|
|
test('a child encountered before its parent waits by source key, then joins without overwriting user edits',async()=>{
|
|
const f=fixture([{id:'child',dependencies:['parent']},{id:'parent'}]);const real=f.ports.commit;f.ports.commit=(units,next,expected)=>{const result=real(units,next,expected);if(result.status==='applied'&&units.some(unit=>unit.key==='parent')){const id=f.root.job.stableIds.parent,element=f.root.document.elements.get(id);f.root={...f.root,document:applyChanges(f.root.document,[{kind:'field',target:{id,incarnation:1},field:'data.title',before:element.data.title,after:'user edit'}])};}return result;};
|
|
await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.root.job.committedSourceKeys,['parent','child']);assert.equal(f.root.document.elements.get(f.root.job.stableIds.parent).data.title,'user edit');assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,0);
|
|
});
|
|
|
|
test('a rejected multi-record batch publishes no prefix, isolates the invalid record and retains independent content',async()=>{
|
|
const f=fixture([{id:'good',dependencies:['parent']},{id:'bad',dependencies:['parent'],bad:true},{id:'parent'},{id:'tail'}]);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.calls.some(keys=>keys.includes('good')&&keys.includes('bad')));assert.deepEqual(f.root.job.committedSourceKeys,['parent','good','tail']);assert.equal(f.root.document.elements.has(f.root.job.stableIds.bad),false);assert.equal(f.root.job.failures.find(failure=>failure.key==='bad').code,'invalid_geometry');assert.equal(f.errors.length,0);
|
|
});
|
|
|
|
test('record 101 failure preserves prior content and later valid records; missing dependencies report partial without dangling insertion',async()=>{
|
|
const input=Array.from({length:100},(_,index)=>({id:`n-${index}`}));input.push('{bad}',{id:'after'},{id:'edge',dependencies:['missing']});const f=fixture(input);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.equal(f.root.document.elements.size,101);assert.equal(f.root.job.failures.length,2);assert.ok(f.root.job.failures.some(failure=>failure.code==='invalid_json'));assert.ok(f.root.job.failures.some(failure=>failure.code==='unresolved_dependency'));assert.equal(f.root.document.elements.has(f.root.job.stableIds.edge),false);validateImportJob(JSON.parse(JSON.stringify(f.root.job)));
|
|
});
|
|
|
|
test('a durable pending locator resumes with the same IDs; a committed key is not inserted again after deletion',async()=>{
|
|
const gate=deferred(),entered=deferred(),f=fixture([{id:'one',title:'original'}]);const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length&&!f.root.job.committedSourceKeys.length){entered.resolve();await gate.promise;}};
|
|
const task=f.runner.start();await entered.promise;const cut=f.saved.at(-1),closing=f.runner.close();gate.resolve();await task;await closing;
|
|
const document=createDocument(cut.document.projectId,cut.document.canvases,cut.document.elements,new Map(cut.document.tombstones)),resumed=fixture([],{source:f.source,root:{document,job:cut.job},ports:{createId(){throw Error('reallocated identity');}}});await resumed.runner.start();assert.equal(resumed.root.job.status,'completed');assert.equal(resumed.root.document.elements.size,1);assert.equal(resumed.root.job.stableIds.one,cut.job.stableIds.one);
|
|
const id=resumed.root.job.stableIds.one,element=resumed.root.document.elements.get(id),deleted=applyChanges(resumed.root.document,[{kind:'element',id,before:element,after:null}]),again=fixture([],{source:f.source,root:{document:deleted,job:{...resumed.root.job,status:'paused',eof:false}},ports:{createId(){throw Error('reallocated identity');}}});await again.runner.start();assert.equal(again.root.document.elements.has(id),false);assert.equal(again.calls.length,0);
|
|
});
|
|
|
|
test('changed source bytes or source identity cannot claim a durable pending item',async()=>{
|
|
const oldSource=createNdjsonImportSource({bytes:bytes([{id:'one',title:'aaa'}]),fingerprint:'immutable-source'}),gate=deferred(),entered=deferred(),f=fixture([],{source:oldSource});const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length){entered.resolve();await gate.promise;}};
|
|
const task=f.runner.start();await entered.promise;const root=f.root,closing=f.runner.close();gate.resolve();await task;await closing;
|
|
for(const [title,fingerprint,code]of[['bbb','immutable-source','import_source_changed'],['aaa','different-source','import_source_mismatch']]){const source=createNdjsonImportSource({bytes:bytes([{id:'one',title}]),fingerprint}),r=fixture([],{source,root});await r.runner.start();assert.equal(r.root.job.status,'paused');assert.equal(r.root.job.error,code);assert.equal(r.root.document.elements.size,0);assert.equal(r.calls.length,0);}
|
|
});
|
|
|
|
test('dependency backpressure pauses with locators intact and a larger budget resumes the same cursor and mapping',async()=>{
|
|
const f=fixture([{id:'child',dependencies:['parent']},{id:'parent'}],{ports:{maxPending:1}});await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'dependency_capacity');assert.equal(f.root.job.pending.length,1);const id=f.root.job.stableIds.child;
|
|
const r=fixture([],{source:f.source,root:f.root,ports:{maxPending:8,createId:()=> 'parent-id'}});await r.runner.start();assert.equal(r.root.job.status,'completed');assert.equal(r.root.job.stableIds.child,id);assert.equal(r.root.document.elements.size,2);
|
|
});
|
|
|
|
test('future interpreter versions and malformed job cuts fail before any source read',()=>{
|
|
const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'s',parserVersion:1});assert.throws(()=>validateImportJob({...job,version:4}),/import_interpreter_unavailable/);assert.throws(()=>validateImportJob({...job,stableIds:{a:'same',b:'same'}}),/invalid_import_job/);assert.throws(()=>validateImportJob({...job,status:'completed'}),/invalid_import_job/);
|
|
});
|
|
|
|
test('fast independent records stage bounded publications and commits while the first record remains immediate',async()=>{
|
|
const f=fixture(Array.from({length:80},(_,index)=>({id:`batch-${index}`})),{ports:{now:()=>0,maxBatch:16,optimisticLocal:true}}),publish=f.ports.publish;let publications=0;f.ports.publish=(...args)=>{publications++;return publish(...args);};
|
|
await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,80);assert.deepEqual(f.calls[0],['batch-0']);assert.ok(f.calls.some(group=>group.length>1));assert.ok(f.calls.every(group=>group.length<=16));assert.ok(publications<20,`${publications} metadata publications for 80 records`);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
|
|
});
|
|
|
|
test('a stalled speculative read cannot hide the ready staged prefix or require concurrent source seeks',async()=>{
|
|
const gate=deferred(),entered=deferred(),source=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'},{id:'third'}]),fingerprint:'immutable-source'}),signal=new AbortController().signal,first=await source.read(0,signal),second=await source.read(first.nextCursor,signal);let active=0;
|
|
const f=fixture([],{source:{...source,async read(cursor,signal){assert.equal(active,0,'concurrent source seeks');active++;try{if(cursor===second.nextCursor){entered.resolve();await gate.promise;}return await source.read(cursor,signal);}finally{active--;}}},ports:{now:()=>0,optimisticLocal:true}});
|
|
const running=f.runner.start();await entered.promise;
|
|
for(let i=0;i<100&&f.root.document.elements.size<2;i++)await new Promise(resolve=>setTimeout(resolve,1));assert.equal(f.root.document.elements.size,2);assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.job.eof,false);await f.runner.cancel();gate.resolve();await running;assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.status,'cancelled');assert.equal(f.errors.length,0);
|
|
});
|
|
|
|
test('cancelling during a staged decode discards unpublished cursor and identity changes',async()=>{
|
|
const f=fixture([{id:'first'},{id:'second'},{id:'third'}],{ports:{now:()=>0,optimisticLocal:true}}),prepare=f.ports.prepare;let cancelled;
|
|
f.ports.prepare=(line,job,reserve)=>{const unit=prepare(line,job,reserve);if(line.value.id==='second')queueMicrotask(()=>{cancelled=f.runner.cancel();});return unit;};
|
|
await f.runner.start();await cancelled;assert.equal(f.root.job.status,'cancelled');assert.deepEqual(f.root.job.committedSourceKeys,['first']);assert.equal(f.root.document.elements.size,1);assert.equal(Object.hasOwn(f.root.job.stableIds,'second'),false);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
|
|
});
|
|
|
|
test('a source I/O failure is reported after publishing all already prepared ready records',async()=>{
|
|
const raw=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'}]),fingerprint:'immutable-source'}),first=await raw.read(0,new AbortController().signal),second=await raw.read(first.nextCursor,new AbortController().signal),f=fixture([],{source:{...raw,async read(cursor,signal){if(cursor===second.nextCursor)throw fault('source_offline');return raw.read(cursor,signal);}},ports:{now:()=>0,optimisticLocal:true}});
|
|
await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'source_offline');assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,1);
|
|
});
|
|
|
|
test('a reference cycle commits every member in one root even when the adaptive batch budget is one',async()=>{
|
|
const f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['C']},{id:'C',dependencies:['A']}],{ports:{maxBatch:1}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.calls,[['A','B','C']]);assert.equal(f.commits.length,1);assert.equal(f.root.document.elements.size,3);validateImportJob(f.root.job);
|
|
});
|
|
|
|
test('a rejected reference cycle never isolates or publishes one member; independent content survives',async()=>{
|
|
const f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['A'],bad:true},{id:'tail'}]);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.deepEqual(f.root.job.committedSourceKeys,['tail']);assert.ok(f.calls.filter(keys=>keys.includes('A')||keys.includes('B')).every(keys=>keys.includes('A')&&keys.includes('B')));assert.deepEqual(f.root.job.failures.map(value=>value.key),['A','B']);assert.equal(f.root.document.elements.size,1);
|
|
});
|
|
|
|
test('a durable pending cycle resumes without reallocating IDs or exposing a partial group',async()=>{
|
|
const gate=deferred(),entered=deferred(),f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['A']}]);const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length===2&&!f.root.job.committedSourceKeys.length){entered.resolve();await gate.promise;}};
|
|
const running=f.runner.start();await entered.promise;const cut=f.root,closing=f.runner.close();gate.resolve();await running;await closing;assert.equal(cut.document.elements.size,0);
|
|
const r=fixture([],{source:f.source,root:cut,ports:{createId(){throw Error('identity reallocated');}}});await r.runner.start();assert.equal(r.root.job.status,'completed');assert.deepEqual(r.calls,[['A','B']]);assert.deepEqual(r.root.job.stableIds,cut.job.stableIds);
|
|
});
|
|
|
|
|
|
test('new jobs carry interpreter version 3 and a resumed legacy job upgrades in its first published cut',async()=>{
|
|
const f=fixture([{id:'one'}]);assert.equal(f.root.job.version,3);f.root={...f.root,job:{...f.root.job,version:1}};validateImportJob(f.root.job);const publish=f.ports.publish;f.ports.publish=(job,expected)=>{assert.equal(job.version,3);return publish(job,expected);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.version,3);
|
|
});
|
|
|
|
test('optimistic local checkpoints use host backpressure without declaring completion ahead of persistence',async()=>{
|
|
const gate=deferred(),entered=deferred(),f=fixture([{id:'one'},{id:'two'},{id:'three'}],{ports:{optimisticLocal:true,maxBatch:1}});let blocked=false;
|
|
f.ports.awaitDurabilityBudget=async()=>{if(!blocked&&f.root.document.elements.size===2){blocked=true;entered.resolve();await gate.promise;}};
|
|
const running=f.runner.start();await entered.promise;assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.status,'running');assert.equal(f.root.job.committedSourceKeys.length,2);gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,3);assert.ok(f.saved.some(cut=>cut.job.status==='running'&&cut.job.eof&&cut.document.elements.length===3));
|
|
});
|
|
|
|
test('both final content and completion-metadata storage failures pause resumably without replaying committed keys',async()=>{
|
|
for(const phase of ['content','metadata']){
|
|
const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist;let broken=true;
|
|
f.ports.persist=async()=>{if(broken&&(phase==='content'?f.root.job.eof&&f.root.job.status==='running':f.root.job.status==='completed'))throw fault('quota');return persist();};
|
|
await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'quota');assert.deepEqual(f.root.job.committedSourceKeys,['one','two']);const identities={...f.root.job.stableIds},calls=f.calls.length;broken=false;await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.calls.length,calls);assert.deepEqual(f.root.job.stableIds,identities);
|
|
}
|
|
});
|
|
|
|
test('pausing during the final durability barrier prevents a late completion publication',async()=>{
|
|
const gate=deferred(),entered=deferred(),f=fixture([{id:'one'}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist;
|
|
f.ports.persist=async()=>{if(f.root.job.eof){entered.resolve();await gate.promise;}return persist();};const running=f.runner.start();await entered.promise;const pausing=f.runner.pause();assert.equal(f.root.job.status,'paused');gate.resolve();await pausing;await running;assert.equal(f.root.job.status,'paused');await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.calls.length,1);
|
|
});
|
|
|
|
test('resuming a failed partial-status write does not duplicate unresolved dependency diagnostics',async()=>{
|
|
const f=fixture([{id:'good'},{id:'waiting',dependencies:['missing']}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist;let broken=true;
|
|
f.ports.persist=async()=>{if(broken&&f.root.job.status==='partial')throw fault('quota');return persist();};await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.failures.length,1);broken=false;await f.runner.start();assert.equal(f.root.job.status,'partial');assert.equal(f.root.job.failures.length,1);assert.equal(f.root.document.elements.size,1);assert.equal(f.calls.length,1);
|
|
});
|
|
|
|
|
|
test('a ready parser publication observes host backpressure before admitting its content root',async()=>{
|
|
const gate=deferred(),entered=deferred();let blocked=true;const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,maxBatch:1,awaitDurabilityBudget(stage){if(blocked){assert.equal(stage,'parsed');entered.resolve();return gate.promise;}}}});const running=f.runner.start();await entered.promise;assert.equal(f.root.job.pending.length,1);assert.equal(f.root.job.committedSourceKeys.length,0);assert.equal(f.root.document.elements.size,0);await ticks();assert.equal(f.root.document.elements.size,0,'ready content bypassed host budget');blocked=false;gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,2);assert.equal(f.errors.length,0);await f.runner.close();
|
|
});
|
|
|
|
test('async commit preparation cannot publish while waiting and cancellation fences a late successful result',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'image'}],{ports:{async beforeCommit(units,job,signal){assert.equal(f.root.document.elements.size,0);assert.ok(job.pending.length);entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;assert.equal(f.commits.length,0);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.commits.length,0);assert.equal(f.root.document.elements.size,0);await f.runner.close();});
|
|
test('async preparation denial isolates independent groups without publishing the denied record',async()=>{const prepared=[],f=fixture([{id:'denied'},{id:'allowed'}],{ports:{async beforeCommit(units){prepared.push(units.map(unit=>unit.key));return units.some(unit=>unit.key==='denied')?{enabled:false,code:'asset_forbidden',targets:[]}:{enabled:true};}}});await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.root.job.failures.some(row=>row.key==='denied'&&row.code==='asset_forbidden'));assert.equal(f.root.document.elements.size,1);assert.equal([...f.root.document.elements.values()][0].data.title,'allowed');assert.ok(prepared.length>=2);assert.equal(f.errors.length,0);await f.runner.close();});
|
|
test('a generation change during async preparation prevents the old attempt from committing or changing progress',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'later'}],{ports:{async beforeCommit(){entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;f.root={...f.root,job:{...f.root.job,generation:f.root.job.generation+1}};const generation=f.root.job.generation;gate.resolve();await running;assert.equal(f.commits.length,0);assert.equal(f.root.job.generation,generation);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,1);await f.runner.close();});
|
|
|
|
test('NDJSON checks elapsed time before returning a short scanned record and honors cancellation',async()=>{const controller=new AbortController();let clock=0,yields=0;const source=createNdjsonImportSource({bytes:new Blob([JSON.stringify({text:'x'.repeat(2048)})+'\n']),fingerprint:'timed',chunkBytes:256,now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(source.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);});
|
|
|
|
test('a resumed preallocated import retains the exact stable-ID table through parsed and committed checkpoints',async()=>{
|
|
const stableIds=Object.freeze(Object.fromEntries([...Array.from({length:1000},(_,i)=>['reserved-'+i,'unused-'+i]),['one','id-one'],['two','id-two']])),job={...createImportJob({id:'import',projectId:'project',sourceFingerprint:'immutable-source',parserVersion:1}),stableIds},f=fixture([{id:'one'},{id:'two'}],{root:{document:base,job},ports:{createId(){throw Error('preallocated ID was replaced');}}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.strictEqual(f.root.job.stableIds,stableIds);for(const cut of f.commits)assert.strictEqual(cut.job.stableIds,stableIds);assert.deepEqual(f.root.job.committedSourceKeys,['one','two']);assert.equal(f.errors.length,0);await f.runner.close();
|
|
});
|
|
|
|
test('identity allocation keeps the caller port method receiver',async()=>{
|
|
const f=fixture([{id:'bound'}],{ports:{allocated:0,createId(){return `bound-${++this.allocated}`;}}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.stableIds.bound,'bound-1');assert.equal(f.ports.allocated,1);assert.equal(f.errors.length,0);await f.runner.close();
|
|
});
|
|
|
|
test('fixed-cost commits amortize ready records while preserving the immediate first record and yielding after expensive work',async()=>{
|
|
let clock=0;const f=fixture(Array.from({length:80},(_,i)=>({id:`fixed-${i}`})),{ports:{now:()=>clock,frameBudgetMs:6,maxBatch:16,optimisticLocal:true}}),commit=f.ports.commit;f.ports.commit=(units,...rest)=>{clock+=4+units.length*.05;return commit(units,...rest);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.calls[0],['fixed-0']);assert.ok(f.calls.some(group=>group.length>=8));assert.ok(f.calls.length<20);assert.ok(f.yields>0);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
|
|
});
|
|
|
|
test('a cheap runner can use the engine-sized ceiling without dropping records or changing FIFO order',async()=>{
|
|
const records=Array.from({length:700},(_,index)=>({id:`wide-${index}`})),f=fixture(records,{ports:{maxBatch:128,now:()=>0}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,700);assert.ok(f.calls.some(batch=>batch.length>32));assert.ok(f.calls.every(batch=>batch.length<=128));assert.deepEqual(f.calls.flat(),records.map(row=>row.id));assert.equal(f.errors.length,0);await f.runner.close();
|
|
});
|
|
test('the larger ceiling still backs off when a measured commit exceeds the unchanged time budget',async()=>{
|
|
let clock=0;const records=Array.from({length:500},(_,index)=>({id:`cost-${index}`})),f=fixture(records,{ports:{maxBatch:128,now:()=>clock}}),commit=f.ports.commit;f.ports.commit=(units,...args)=>{clock+=units.length>40?12:1;return commit(units,...args);};await f.runner.start();const sizes=f.calls.map(batch=>batch.length),wide=sizes.findIndex(size=>size>40);assert.ok(wide>=0);assert.ok(sizes[wide+1]<sizes[wide]);assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,500);assert.deepEqual(f.calls.flat(),records.map(row=>row.id));assert.equal(f.errors.length,0);await f.runner.close();
|
|
});
|
|
|
|
test('growing import tables yield when encoded while keeping captured IDs and bindings stable',async()=>{
|
|
const {sealRecoveryImageAsync,sealRecoveryImage}=await import('../../src/lib/canvas-runtime/recovery-encoding.ts');
|
|
const f=fixture(Array.from({length:300},(_,i)=>({id:'encode-'+i})));await f.runner.start();
|
|
const job=f.root.job;assert.equal(job.committedSourceKeys.length,300);let ticks=0,yields=0;
|
|
const image=await sealRecoveryImageAsync(17,1,1,job,{now:()=>ticks++,frameBudgetMs:1,yieldTask:async()=>{yields++;}});
|
|
assert.ok(yields>0);assert.deepEqual(image,sealRecoveryImage(17,1,1,job));
|
|
assert.equal(Object.keys(image.payload.stableIds).length,300);assert.equal(Object.keys(image.payload.bindings).length,300);await f.runner.close();
|
|
});
|
|
|
|
|
|
test('waiting import locators and dependencies encode cooperatively before their endpoints arrive',async()=>{
|
|
const {sealRecoveryImageAsync,sealRecoveryImage}=await import('../../src/lib/canvas-runtime/recovery-encoding.ts');
|
|
const f=fixture(Array.from({length:300},(_,i)=>({id:'waiting-'+i,dependencies:['absent']})));await f.runner.start();
|
|
const pending=f.root.job.pending;assert.equal(pending.length,300);assert.equal(f.root.document.elements.size,0);
|
|
let ticks=0,yields=0;
|
|
const image=await sealRecoveryImageAsync(17,1,1,pending,{now:()=>ticks++,frameBudgetMs:1,yieldTask:async()=>{yields++;}});
|
|
assert.ok(yields>0);assert.deepEqual(image,sealRecoveryImage(17,1,1,pending));
|
|
assert.deepEqual(image.payload[299].dependencies,['absent']);await f.runner.close();
|
|
});
|