Adapt import commit batches from observed transaction duration

headless/README.md

@@ -776,3 +776,13 @@ still exceeds 240 seconds. Removing ID-table copies and metadata-only document
 scans is directly verified; a general throughput improvement at 10000 nodes is
 not established. Per-record commit, ACK and remaining table traversal costs remain
 part of the larger implementation goal.
+
+Import commit sizing uses runner-local feedback rather than dividing total
+transaction time by unit count and treating it as a purely marginal cost.
+The first ready record remains immediate. Successful batches within the soft
+frame budget grow by a bounded step; an over-budget batch reduces the limit.
+A short tail or dependency group cannot increase it. `maxBatch`, byte/change
+limits, atomic reference cycles, parser slices and durability backpressure keep
+their existing roles. Rejected attempts do not train the success-cost estimate.
+The limit is a scheduling hint, not another persistent queue or an assurance
+that an indivisible transaction or subsequent renderer work fits one frame.

src/lib/canvas-runtime/import-batches.ts

@@ -0,0 +1,15 @@
+/** Runner-owned feedback, not persisted job state. Probe bounded batches to
+ * distinguish fixed publication cost from the marginal cost of another unit. */
+export function createImportBatches(budget:number,max:number){
+ let limit=1,observed=false;
+ return Object.freeze({
+  get limit(){return limit;},
+  get observed(){return observed;},
+  record(units:number,elapsed:number){
+   observed=true;
+   if(elapsed>budget){limit=Math.max(1,Math.min(limit-1,Math.floor(units*budget/elapsed)));return;}
+   // A short tail/dependency group does not prove that a larger batch is cheap.
+   if(units>=limit)limit=Math.min(max,limit+Math.max(1,Math.floor(limit/2)));
+  },
+ });
+}

src/lib/canvas-runtime/imports.ts

@@ -1,3 +1,4 @@
+import {createImportBatches} from './import-batches.ts';
 import {createImportIdentities} from './import-identities.ts';
 import {validateImportReadProof,type ImportReadProof} from './import-read-proof.ts';
 import {readImportRange,type ImportByteReader} from './import-byte-reader.ts';
@@ -132,7 +133,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
   if(initial.source&&(initial.source.integrityVersion===2?!ports.checkpointSource:!initial.source.sha256||initial.source.integrityVersion!==1))throw fault('import_source_integrity_unavailable');
   if(!Number.isFinite(budget)||budget<=0||![maxPending,maxBytes,maxBatch].every(value=>Number.isSafeInteger(value)&&value>0))throw fault('invalid_import_budget');
   const identities=createImportIdentities(),allocateIdentity=()=>ports.createId();
-  let costPerUnit:number|undefined;
+  const batches=createImportBatches(budget,maxBatch);
   let active:Promise<void>|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set<symbol>();
   const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}};
   function current(){const job=ports.getJob();if(job.id!==identity.id||job.projectId!==identity.projectId||job.sourceFingerprint!==identity.sourceFingerprint||job.parserVersion!==identity.parserVersion)throw fault('import_identity_changed');return job;}
@@ -162,7 +163,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
     for(const unit of units)for(const change of unit.changes)if(change.kind==='element'&&change.before===null&&change.after)bindings[change.id]={kind:'element',incarnation:change.after.incarnation};else if(change.kind==='canvas'&&change.before===null&&change.after)bindings[change.id]={kind:'canvas',...(change.after.parent?{entryElementId:change.after.parent.entryElementId}:{})};
     return next(job,{pending,committedSourceKeys:[...job.committedSourceKeys,...units.map(unit=>unit.key)],bindings:{...bindings}});
   }
-  function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const base=current();if(base.generation!==job.generation)throw fault('stale_import_job');value=freeze(checkpointEvidence(value));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:base))throw fault('invalid_import_commit');return result;}
+  function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const base=current();if(base.generation!==job.generation)throw fault('stale_import_job');value=freeze(checkpointEvidence(value));const start=now(),result=ports.commit(units,value,job.generation),elapsed=Math.max(0,now()-start);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:base))throw fault('invalid_import_commit');if(result.status==='applied')batches.record(units.length,elapsed);return result;}
   async function pump(signal:AbortSignal){
     let frame=now(),didWork=false;
     async function attempt(units:readonly ImportUnit[],job:ImportJob,proposed:ImportJob):Promise<Result|undefined>{
@@ -191,7 +192,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
       if(didWork&&now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();didWork=false;continue;}
       didWork=true;
       const committed=new Set(job.committedSourceKeys);
-      const batchLimit=Math.max(1,Math.min(maxBatch,costPerUnit===undefined?1:Math.floor(budget/(costPerUnit*1.5))));
+      const batchLimit=batches.limit;
       const groups=selectImportGroups(job.pending,committed,{units:batchLimit,bytes:maxBytes,changes:10000}),chosen=groups.flat();
       if(chosen.length){
         const units:ImportUnit[]=[];
@@ -261,7 +262,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
         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(!batches.observed&&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);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/import-batches.test.mjs

@@ -0,0 +1,11 @@
+import {test} from 'node:test';
+import assert from 'node:assert/strict';
+import {createImportBatches} from '../../src/lib/canvas-runtime/import-batches.ts';
+test('fixed publication cost can grow beyond singleton batches without raising the time budget',()=>{
+ const batches=createImportBatches(6,32);assert.equal(batches.observed,false);assert.equal(batches.limit,1);const sizes=[];
+ for(let i=0;i<20;i++){const n=batches.limit;sizes.push(n);batches.record(n,4+n*.05);}
+ assert.equal(batches.observed,true);assert.equal(batches.limit,32);assert.ok(sizes.includes(2));assert.ok(sizes.every(n=>n<=32));
+});
+test('expensive batches back off, then recover as costs fall; partial tails do not inflate the limit',()=>{
+ const batches=createImportBatches(6,32);for(let i=0;i<4;i++)batches.record(batches.limit,1);const full=batches.limit;batches.record(1,1);assert.equal(batches.limit,full);batches.record(full,60);assert.equal(batches.limit,1);batches.record(1,12);assert.equal(batches.limit,1);batches.record(1,2);assert.equal(batches.limit,2);batches.record(2,2);assert.equal(batches.limit,3);
+});

tests/headless/imports.test.mjs

@@ -151,3 +151,7 @@ test('a resumed preallocated import retains the exact stable-ID table through pa
 test('identity allocation keeps the caller port method receiver',async()=>{
  const f=fixture([{id:'bound'}],{ports:{allocated:0,createId(){return `bound-${++this.allocated}`;}}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.stableIds.bound,'bound-1');assert.equal(f.ports.allocated,1);assert.equal(f.errors.length,0);await f.runner.close();
 });
+
+test('fixed-cost commits amortize ready records while preserving the immediate first record and yielding after expensive work',async()=>{
+ let clock=0;const f=fixture(Array.from({length:80},(_,i)=>({id:`fixed-${i}`})),{ports:{now:()=>clock,frameBudgetMs:6,maxBatch:16,optimisticLocal:true}}),commit=f.ports.commit;f.ports.commit=(units,...rest)=>{clock+=4+units.length*.05;return commit(units,...rest);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.calls[0],['fixed-0']);assert.ok(f.calls.some(group=>group.length>=8));assert.ok(f.calls.length<20);assert.ok(f.yields>0);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
+});