Check import backpressure after ready parser publication

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');

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<void>;
+  awaitDurabilityBudget?(stage?:'parsed'|'committed'):void|Promise<void>;
   now?:()=>number;yieldTask?:()=>Promise<void>;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<Id,ImportBinding>=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');}
 
     }
   }

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();
+});