diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 4152040..ca83913 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -387,7 +387,7 @@ export async function openProjectEngine(options: EngineOptions) { const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}); if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return; if(importHolds.size){deferredImportLoads.add(id);return;} - const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:()=>persistenceFailed||pending.size>=32?persist():undefined,source:importSource,getJob:()=>getImport(id),createId, + const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:stage=>pending.size>=32||stage!=='parsed'&&persistenceFailed?persist():undefined,source:importSource,getJob:()=>getImport(id),createId, publish:publishImport,prepare:(line,job,reserve)=>['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')?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 886d05d..8c1d5dd 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -102,7 +102,7 @@ export type ImportRunnerPorts={ /** 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; + awaitDurabilityBudget?(stage?:'parsed'|'committed'):void|Promise; now?:()=>number;yieldTask?:()=>Promise;frameBudgetMs?:number;maxPending?:number;maxPendingBytes?:number;maxBatch?:number; }; @@ -128,7 +128,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 checkpoint=async(stage:'parsed'|'committed'='committed')=>{if(ports.optimisticLocal&&ports.awaitDurabilityBudget)await ports.awaitDurabilityBudget(stage);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??{}); @@ -229,7 +229,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 checkpoint();} + if(staged!==base){const{generation,...patch}=staged;publish(base,patch);const ready=staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)));if(ports.awaitDurabilityBudget||!ports.optimisticLocal||!ready)await checkpoint(ready?'parsed':'committed');} } } diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index 7978b36..0557d1d 100644 --- a/tests/headless/imports.test.mjs +++ b/tests/headless/imports.test.mjs @@ -110,8 +110,8 @@ test('new jobs carry interpreter version 3 and a resumed legacy job upgrades in }); 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 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)); }); @@ -132,3 +132,8 @@ test('resuming a failed partial-status write does not duplicate unresolved depen 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(); +});