From 2529a801b2affb2964d7dbef30664369ccbbbf06 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Tue, 10 Nov 2026 01:10:05 +0800 Subject: [PATCH] Adapt import commit batches from observed transaction duration --- headless/README.md | 10 ++++++++++ src/lib/canvas-runtime/import-batches.ts | 15 +++++++++++++++ src/lib/canvas-runtime/imports.ts | 9 +++++---- tests/headless/import-batches.test.mjs | 11 +++++++++++ tests/headless/imports.test.mjs | 4 ++++ 5 files changed, 45 insertions(+), 4 deletions(-) create mode 100644 src/lib/canvas-runtime/import-batches.ts create mode 100644 tests/headless/import-batches.test.mjs diff --git a/headless/README.md b/headless/README.md index 5c0f29c..eb697ba 100644 --- a/headless/README.md +++ b/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. diff --git a/src/lib/canvas-runtime/import-batches.ts b/src/lib/canvas-runtime/import-batches.ts new file mode 100644 index 0000000..8808597 --- /dev/null +++ b/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))); + }, + }); +} diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index 24a2de0..382e81c 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/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|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set(); 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{ @@ -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');} diff --git a/tests/headless/import-batches.test.mjs b/tests/headless/import-batches.test.mjs new file mode 100644 index 0000000..173dc45 --- /dev/null +++ b/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); +}); diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index 7cdd037..25cec47 100644 --- a/tests/headless/imports.test.mjs +++ b/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); +});