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

perf(canvas): batch progressive import parsing and publication

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 16:29:39 +08:00
parent bde841330a
commit ff07b52a21
3 changed files with 75 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/imports.ts
+49 -17
View File
@@ -118,6 +118,18 @@ export function createImportRunner(ports:ImportRunnerPorts){
function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:job))throw fault('invalid_import_commit');return result;}
async function pump(signal:AbortSignal){
let frame=now(),didWork=false;
type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown};
let reading:{cursor:number;result:Promise<ReadResult>}|undefined;
const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();let decodedBytes=0;
function cache(key:string,value:{unit:ImportUnit;digest:string;bytes:number}){const previous=decoded.get(key);if(previous)decodedBytes-=previous.bytes;decoded.delete(key);decoded.set(key,value);decodedBytes+=value.bytes;while(decoded.size>maxBatch||decodedBytes>maxBytes){const[key,value]=decoded.entries().next().value!;decoded.delete(key);decodedBytes-=value.bytes;}}
function forget(key:string){const value=decoded.get(key);if(value){decodedBytes-=value.bytes;decoded.delete(key);}}
async function read(cursor:number,flushReady:boolean):Promise<ReadResult|{kind:'flush'}>{
if(reading&&reading.cursor!==cursor)throw fault('invalid_import_cursor');
reading??={cursor,result:Promise.resolve().then(()=>ports.source.read(cursor,signal)).then(line=>({kind:'line' as const,line}),error=>({kind:'error' as const,error}))};
let timer:ReturnType<typeof setTimeout>|undefined;
const result=flushReady?await Promise.race([reading.result,new Promise<{kind:'flush'}>(resolve=>{timer=setTimeout(()=>resolve({kind:'flush'}),0);})]):await reading.result;
if(timer!==undefined)clearTimeout(timer);if(result.kind!=='flush')reading=undefined;return result;
}
if(!ports.optimisticLocal)await ports.persist();
while(!closed&&!signal.aborted&&!holds.size){
let job=current();if(job.status!=='running')return;
@@ -130,6 +142,10 @@ export function createImportRunner(ports:ImportRunnerPorts){
if(chosen.length){
const units:ImportUnit[]=[];let failed:ImportFailure|undefined;
for(const locator of chosen){
const cached=decoded.get(locator.key);
if(cached&&cached.digest===locator.digest){units.push(cached.unit);continue;}
// A source need not support concurrent reads. Finish any speculative read before seeking.
if(reading)await reading.result;
const line=await ports.source.read(locator.cursor,signal);
if(closed||signal.aborted||current().generation!==job.generation)return;
if(!line||line.error||line.skip||line.nextCursor!==locator.nextCursor)throw fault('import_source_changed');
@@ -152,6 +168,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();}
}
}else await ports.persist();
for(const unit of units)forget(unit.key);
continue;
}
if(job.eof){
@@ -159,23 +176,38 @@ export function createImportRunner(ports:ImportRunnerPorts){
publish(job,{status:job.failures.length||unresolved.length?'partial':'completed',failures:[...job.failures,...unresolved]});await ports.persist();return;
}
if(job.pending.length>=maxPending||new TextEncoder().encode(JSON.stringify(job.pending)).byteLength>=maxBytes){publish(job,{status:'paused',error:'dependency_capacity'});await ports.persist();return;}
const line=await ports.source.read(job.cursor,signal);
if(closed||signal.aborted||current().generation!==job.generation)return;
if(!line){publish(job,{eof:true});await ports.persist();continue;}
if(line.cursor!==job.cursor||!Number.isSafeInteger(line.nextCursor)||line.nextCursor<=job.cursor)throw fault('invalid_import_cursor');
if(line.skip){publish(job,{cursor:line.nextCursor});await ports.persist();continue;}
if(line.error){publish(job,{cursor:line.nextCursor,failures:[...job.failures,{key:line.key,cursor:line.cursor,code:line.error}]});await ports.persist();continue;}
let patch:Partial<ImportJob>;
try{
const prepared=prepare(line,job),key=prepared.unit.key;
if(job.committedSourceKeys.includes(key)||job.pending.some(unit=>unit.key===key)||job.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key');
const locator={key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:prepared.unit.dependencies,digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length};
patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:[...job.pending,locator]};
}catch(error){patch={cursor:line.nextCursor,failures:[...job.failures,{key:line.key,cursor:line.cursor,code:(error as {code?:string}).code??'import_decode_failed'}]};}
if(closed||signal.aborted||current().generation!==job.generation)return;
publish(job,patch);
const staged=patch.pending?.at(-1),immediatelyReady=staged?.dependencies.every(key=>job.committedSourceKeys.includes(key));
if(!ports.optimisticLocal||!immediatelyReady)await ports.persist();
// Stage a bounded parser slice, then publish its cursor/IDs/locators once. This is
// unpublished transaction preparation, never another durable task state store.
const base=job;let staged=job,records=0;
while(!staged.eof&&records<maxBatch&&staged.pending.length<maxPending&&new TextEncoder().encode(JSON.stringify(staged.pending)).byteLength<maxBytes){
const ready=staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)));
const result=await read(staged.cursor,ready);
if(closed||signal.aborted||current().generation!==base.generation)return;
if(result.kind==='flush')break;
if(result.kind==='error'){
if(ready){reading={cursor:staged.cursor,result:Promise.resolve(result)};break;}
if(staged!==base){const{generation,...patch}=staged;publish(base,patch);}throw result.error;
}
const line=result.line;
if(!line){staged={...staged,eof:true};break;}
if(line.cursor!==staged.cursor||!Number.isSafeInteger(line.nextCursor)||line.nextCursor<=staged.cursor)throw fault('invalid_import_cursor');
let patch:Partial<ImportJob>;
if(line.skip)patch={cursor:line.nextCursor};
else if(line.error)patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:line.error}]};
else try{
const prepared=prepare(line,staged),key=prepared.unit.key;
if(staged.committedSourceKeys.includes(key)||staged.pending.some(unit=>unit.key===key)||staged.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key');
const locator={key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:prepared.unit.dependencies,digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length};
cache(key,{unit:prepared.unit,digest:locator.digest,bytes:locator.byteLength});
patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:[...staged.pending,locator]};
}catch(error){patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:(error as {code?:string}).code??'import_decode_failed'}]};}
if(closed||signal.aborted||current().generation!==base.generation)return;
staged={...staged,...patch};records++;
// First ready record displays immediately; subsequent slices amortize publication.
if(costPerUnit===undefined&&staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)))||now()-frame>=budget)break;
}
if(staged!==base){const{generation,...patch}=staged;publish(base,patch);if(!ports.optimisticLocal||!staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key))))await ports.persist();}
}
}
function launch(){
tests/headless/imports-engine.test.mjs
+3 -3
View File
@@ -127,12 +127,12 @@ test('read-only policy rejects copy/import admission without storing a source or
});
test('preparing copy checkpoints source and allocated identities, reopens paused and reconstructs the same mapping',async()=>{
const f=fixture();let engine=await f.open();try{
test('preparing copy checkpoints source and allocated identities, reopens paused and reconstructs the same mapping',async(t)=>{
const f=fixture(),save=f.storage.commit;let durableCuts=0;f.storage.commit=async(...args)=>{durableCuts++;return save(...args);};let engine=await f.open();try{
await seedCopy(engine);assert.equal(engine.editor.transaction(tx=>{for(let i=0;i<1000;i++)tx.dispatch({type:'element.create',element:copyNode(`extra-${i}`,'child')});}).status,'applied');await engine.settle();
const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>(snapshot.runtime.imports.get(id).preparation?.allocationIds.length??0)>0);await engine.imports.pause(id);await engine.settle();const checkpoint=engine.imports.getJobs().get(id);assert.equal(checkpoint.status,'paused');assert.equal(checkpoint.source.kind,'duplicate-snapshot');assert.ok(checkpoint.preparation.allocationIds.length>0);assert.ok(checkpoint.preparation.steps>0);assert.equal(checkpoint.committedSourceKeys.length,0);decodeRecovery(f.image(),registry);
engine.editor.dispatch({type:'element.data',target:{id:'F',incarnation:1},patch:{title:'new source title'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');assert.deepEqual(engine.imports.getJobs().get(id).preparation.allocationIds,checkpoint.preparation.allocationIds);await engine.imports.resume(id);await engine.settle();
const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);
const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);t.diagnostic(JSON.stringify({imported:job.committedSourceKeys.length,durableCuts,requests:f.requests.length}));
}finally{await engine.close();}
});
tests/headless/imports.test.mjs
+23
View File
@@ -66,3 +66,26 @@ test('dependency backpressure pauses with locators intact and a larger budget re
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:2}),/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);
});