From e00ac017f1a4096de25de7f42c38c43b13b90af6 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 18:18:23 +0800 Subject: [PATCH] Bound optimistic imports with persistence backpressure --- src/lib/canvas-runtime/engine.ts | 7 ++++--- src/lib/canvas-runtime/imports.ts | 15 ++++++++++----- tests/headless/browser-storage-cases.mjs | 13 +++++++++++++ tests/headless/imports-engine.test.mjs | 9 +++++++++ tests/headless/imports.test.mjs | 24 ++++++++++++++++++++++++ 5 files changed, 60 insertions(+), 8 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 7705aa9..8c607a2 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -99,7 +99,7 @@ export async function openProjectEngine(options: EngineOptions) { }); const pending = new Map(); if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot()); - let writing: Promise | undefined, scheduled = false; + let writing: Promise | undefined, scheduled = false, persistenceFailed = false; function persist(): Promise { 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) => 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'); diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index 411f546..8f60d3a 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -86,6 +86,8 @@ export type ImportRunnerPorts={ persist():Promise;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; now?:()=>number;yieldTask?:()=>Promise;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=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; diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index b28e12f..4883d6f 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -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); diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs index a749b1f..e24a790 100644 --- a/tests/headless/imports-engine.test.mjs +++ b/tests/headless/imports-engine.test.mjs @@ -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();} +}); diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index cb2cda3..c6aa163 100644 --- a/tests/headless/imports.test.mjs +++ b/tests/headless/imports.test.mjs @@ -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); +});