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

Bound optimistic imports with persistence backpressure

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 18:18:23 +08:00
parent 60fad39847
commit e00ac017f1
5 changed files with 60 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+4 -3
View File
@@ -99,7 +99,7 @@ export async function openProjectEngine(options: EngineOptions) {
});
const pending = new Map<number, EngineSnapshot>();
if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot());
let writing: Promise<void> | undefined, scheduled = false;
let writing: Promise<void> | undefined, scheduled = false, persistenceFailed = false;
function persist(): Promise<void> {
if (writing) return writing;
const task = (async () => {
@@ -112,8 +112,9 @@ export async function openProjectEngine(options: EngineOptions) {
for (const file of files) fileWrites.delete(file.id);
if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation));
}
persistenceFailed = false;
})();
writing = task.finally(() => { writing = undefined; });
writing = task.catch(error => { persistenceFailed = true; throw error; }).finally(() => { writing = undefined; });
return writing;
}
function mutate(reduce: (ledger: ReturnType<typeof createLedger>) => void) {
@@ -354,7 +355,7 @@ export async function openProjectEngine(options: EngineOptions) {
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');
if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;expectedFingerprint=job.sourceFingerprint;}
const runner=createImportRunner({optimisticLocal:true,source:createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}),getJob:()=>getImport(id),createId,
const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:()=>persistenceFailed||pending.size>=32?persist():undefined,source:createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}),getJob:()=>getImport(id),createId,
publish:publishImport,prepare:(line,job,reserve)=>job.source?.kind==='duplicate-ndjson'?prepareDuplicateImport(line,job,reserve,options.registry):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');
src/lib/canvas-runtime/imports.ts
+10 -5
View File
@@ -86,6 +86,8 @@ export type ImportRunnerPorts={
persist():Promise<void>;onError(error:unknown):void;
/** Host must durably gate every network write on the same document/job cut. */
optimisticLocal?:boolean;
/** Host backpressure for speculative local cuts. Network remains durably gated by persist(). */
awaitDurabilityBudget?():void|Promise<void>;
now?:()=>number;yieldTask?:()=>Promise<void>;frameBudgetMs?:number;maxPending?:number;maxPendingBytes?:number;maxBatch?:number;
};
@@ -110,6 +112,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
if(new TextEncoder().encode(JSON.stringify(unit)).byteLength>maxBytes)throw fault('import_unit_capacity');
return{unit:freeze(clone(unit)),ids:{...ids}};
}
const checkpoint=async()=>{if(ports.optimisticLocal&&ports.awaitDurabilityBudget)await ports.awaitDurabilityBudget();else await ports.persist();};
const digest=async(unit:ImportUnit)=>[...new Uint8Array(await crypto.subtle.digest('SHA-256',new TextEncoder().encode(JSON.stringify(unit))))].map(value=>value.toString(16).padStart(2,'0')).join('');
function completed(job:ImportJob,units:readonly ImportUnit[],pending:readonly ImportLocator[]){
const bindings:Record<Id,ImportBinding>=Object.assign(Object.create(null),job.bindings??{});
@@ -166,15 +169,17 @@ export function createImportRunner(ports:ImportRunnerPorts){
proposed=completed(job,members,job.pending.filter(value=>!internal.has(value.key)));
result=commit(members,job,proposed);
if(result.status==='rejected'){const code=result.code;publish(job,{pending:job.pending.filter(value=>!internal.has(value.key)),failures:[...job.failures,...group.map(row=>({key:row.key,cursor:row.cursor,code}))]});}
await ports.persist();if(closed||signal.aborted||current().status!=='running')return;
await checkpoint();if(closed||signal.aborted||current().status!=='running')return;
if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();}
}
}else await ports.persist();
}else await checkpoint();
for(const unit of units)forget(unit.key);
continue;
}
if(job.eof){
const unresolved=job.pending.map(unit=>({key:unit.key,cursor:unit.cursor,code:'unresolved_dependency',dependencies:unit.dependencies.filter(key=>!committed.has(key))}));
// Completion must never overtake the final document/source durability barrier.
await ports.persist();if(closed||signal.aborted||current().generation!==job.generation)return;
const unresolved=job.pending.filter(unit=>!job.failures.some(failure=>failure.key===unit.key&&failure.code==='unresolved_dependency')).map(unit=>({key:unit.key,cursor:unit.cursor,code:'unresolved_dependency',dependencies:unit.dependencies.filter(key=>!committed.has(key))}));
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;}
@@ -208,7 +213,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
// 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();}
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 checkpoint();}
}
}
@@ -217,7 +222,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
restart=false;controller=new AbortController();
active=pump(controller.signal).catch(async error=>{
if(closed||controller?.signal.aborted)return;
try{const job=current();if(job.status==='running'){publish(job,{status:'paused',error:(error as {code?:string}).code??'import_failed'});try{await ports.persist();}catch{/* Storage may still be unavailable. */}}}catch{/* A replaced identity cannot receive an old error. */}
try{const job=current();if(['running','completed','partial'].includes(job.status)){publish(job,{status:'paused',error:(error as {code?:string}).code??'import_failed'});try{await ports.persist();}catch{/* Storage may still be unavailable. */}}}catch{/* A replaced identity cannot receive an old error. */}
report(error);
}).finally(()=>{active=undefined;controller=undefined;if(restart&&!closed&&!holds.size){try{if(current().status==='running')void launch();}catch{/* Replaced scope stays untouched. */}}});
return active;
tests/headless/browser-storage-cases.mjs
+13
View File
@@ -319,6 +319,17 @@ async function resourceRecoveryCase(name, assert) {
}
async function importBackpressureCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),registry=createRegistry(builtinComponents()),database=`${name}-import-pressure`,storage=createBrowserStorage({name:database}),original=createDocument(root,[{id:root,parent:null}],[]);let document=original,versions=new Map([[root,1]]),engine,calls=0,release;const gate=new Promise(resolve=>{release=resolve;});
const transport=async request=>{calls++;const body=JSON.parse(request.body);for(const[id,version]of Object.entries(body.expected_versions))assert(version===(versions.get(id)??0),'pressure lost a 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);const participants=Object.keys(body.expected_versions);return{status:200,body:{protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(row=>row.id),model:{documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:Object.fromEntries(document.tombstones),versions:Object.fromEntries(participants.map(id=>[id,versions.get(id)]))}}};};
const options={scope:{accountId:crypto.randomUUID(),workspaceId:workspace,projectId:root},registry,document:original,versions:new Map(versions),storage:{...storage,async commit(owner,batch){if(batch.files?.length)await gate;return storage.commit(owner,batch);}},persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
try{
engine=await openProjectEngine(options);await engine.persist();const id=engine.imports.startNdjson(new Blob([Array.from({length:500},(_,index)=>JSON.stringify({kind:'element',key:`n-${index}`,record:{id:`n-${index}`,incarnation:1,componentId:'text',dataVersion:1,structure:'node',ownerCanvasId:'$target',data:{title:`n-${index}`,text:''},geometry:{x:0,y:0,width:200,height:100,rotation:0}}})).join('\n')]),root);
await new Promise((resolve,reject)=>{const stop=engine.editor.subscribe((_,snapshot)=>{if(snapshot.rootRevision-engine.getDurability().durableRootRevision>=32){clearTimeout(timer);stop();resolve();}}),timer=setTimeout(()=>{stop();reject(Error('browser import did not reach its durability budget'));},5000);});
const count=engine.imports.getJobs().get(id).committedSourceKeys.length;assert(count>1&&count<500&&calls===0,'source durability did not bound speculative progress or fence the network');await new Promise(resolve=>setTimeout(resolve,20));assert(engine.imports.getJobs().get(id).committedSourceKeys.length===count,'backpressure did not stop the producer');release();await engine.settle();assert(document.elements.size===500&&engine.imports.getJobs().get(id).status==='completed','pressure release did not complete all records');await engine.close();engine=await openProjectEngine(options);assert(engine.getSnapshot().document.elements.size===500&&engine.imports.getJobs().get(id).committedSourceKeys.length===500,'IndexedDB lost the completed prefix');
}finally{release();await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
}
async function referenceCycleCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),registry=createRegistry(builtinComponents()),database=`${name}-reference-cycle`,storage=createBrowserStorage({name:database}),original=createDocument(root,[{id:root,parent:null}],[]);
let document=original,versions=new Map([[root,1]]),engine;
@@ -542,6 +553,8 @@ export async function run() {
results.push('owned immutable recovery images survive IndexedDB checksums; mutable images and file descriptors are captured before asynchronous work');
await cooperativeCopyCase(assert);
results.push('cooperative duplicate planning yields to actual browser tasks and cancellation without mutating the source');
await importBackpressureCase(name,assert);
results.push('blocked source durability permits a bounded local prefix, fences all network writes and recovers all records after real IndexedDB resumes');
await referenceCycleCase(name,assert);
results.push('native frame reference cycles and copied port mappings commit atomically through protocol 2 and real IndexedDB undo/redo/reopening');
await duplicateRecoveryCase(name,assert);
tests/headless/imports-engine.test.mjs
+9
View File
@@ -165,3 +165,12 @@ test('native frame ports and deep copy close ownership/reference cycles atomical
await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(copied).status,'completed');assert.equal(engine.getSnapshot().document.elements.get(F2).data.ports[0].nodeId,A2);assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.has(F2),false);assert.ok(engine.getSnapshot().document.elements.has(F));assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get(F2).data.ports[0].nodeId,A2);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);
}finally{await engine.close();}
});
test('blocked source persistence admits several local batches then applies bounded backpressure and gates every request',async()=>{
const f=fixture(),engine=await f.open(),gate=deferred(),entered=deferred(),commit=f.storage.commit;
try{
await engine.persist();f.storage.commit=async(owner,batch)=>{if(batch.files?.length){entered.resolve();await gate.promise;}return commit(owner,batch);};
const id=engine.imports.startNdjson(file(Array.from({length:500},(_,index)=>node(`pressure-${index}`))),'canvas');await entered.promise;await until(engine,snapshot=>snapshot.rootRevision-engine.getDurability().durableRootRevision>=32);const cut=engine.getSnapshot(),count=cut.runtime.imports.get(id).committedSourceKeys.length;assert.ok(count>1&&count<500,`unexpected speculative prefix ${count}`);assert.equal(cut.runtime.imports.get(id).status,'running');assert.equal(f.requests.length,0);assert.equal(f.files.size,0);await new Promise(resolve=>setTimeout(resolve,20));assert.equal(engine.imports.getJobs().get(id).committedSourceKeys.length,count,'import ignored persistence backpressure');
gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,500);assert.equal(f.errors.length,0);assert.equal(f.files.size,1);decodeRecovery(f.image(),registry);
}finally{gate.resolve();await engine.close();}
});
tests/headless/imports.test.mjs
+24
View File
@@ -108,3 +108,27 @@ test('a durable pending cycle resumes without reallocating IDs or exposing a par
test('new jobs carry interpreter version 2 and a resumed legacy job upgrades in its first published cut',async()=>{
const f=fixture([{id:'one'}]);assert.equal(f.root.job.version,2);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,2);return publish(job,expected);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.version,2);
});
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 checkpoints=0;
f.ports.awaitDurabilityBudget=async()=>{if(++checkpoints===2){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);
});