Separate cooperative copy planning from durable checkpoint cadence

headless/README.md

@@ -986,3 +986,16 @@ A blocked schema upgrade retires its pending open. If the old connection closes
 later, that abandoned request aborts in `onupgradeneeded` rather than silently
 upgrading after rejection. A fresh adapter can retry once the blocking client has
 closed; closing an adapter before its upgrade event also aborts the pending open.
+
+Stream-copy planning separates cooperative task yields from durable preparation
+checkpoints. ProjectEngine requests checkpoints on phase changes, the first
+nonempty ID allocation, approximately 50ms of elapsed work or 8192 planning
+steps at slice boundaries. It also persists the final ready mapping before
+entering execution. The planner exposes only a tiny immutable progress read to
+make this decision; full checkpoint cloning happens only when saving/returning.
+Persistence wait time is excluded from the next interval. Each small planning
+slice still yields and checks cancellation. Already-published Roots retain their
+normal ordered durability; unsaved work is private preparation with no created
+elements or requests. The low-level helper defaults to its prior per-slice
+checkpoint policy unless a cadence is requested. Checkpoint/source formats and
+resume IDs stay unchanged; a crash/pause can redo only unsaved private planning.

src/lib/canvas-core/duplicate-checkpoint.ts

@@ -41,6 +41,7 @@ export function createDuplicatePlanner(document:Document,registry:Registry,selec
   function allocate(id:Id,kind:'elements'|'canvases'){const map=suppliedMapping?.[kind],value=map&&Object.hasOwn(map,id)?map[id]:idFactory();if(typeof value!=='string'||!value||used.has(value))throw fault('duplicate_identity',[value]);used.add(value);return value;}
   return Object.freeze({
     ready():boolean{return !sourceChecks&&state.phase==='emit'&&state.emitted===0;},
+    progress(){return Object.freeze({phase:state.phase,steps:state.steps,allocated:state.elementIds.length+state.canvasIds.length});},
     checkpoint():DuplicateCheckpoint{return freeze(clone(state));},
     step():boolean{
       if(sourceChecks){if(sourceChecks.next().done)sourceChecks=undefined;return false;}

src/lib/canvas-runtime/duplicate.ts

@@ -62,10 +62,25 @@ export async function planDuplicateResumable(document:Document,registry:Registry
 
 
 /** Stop before record conversion: the stable order/mappings are sufficient for seekable lazy output. */
-export async function prepareDuplicateStream(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint;topologyOnly?:boolean},scheduling:Omit<DuplicateScheduling,'checkpoint'>&{checkpoint?:(state:DuplicateCheckpoint)=>Promise<void>}={}):Promise<DuplicateCheckpoint>{
+export async function prepareDuplicateStream(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint;topologyOnly?:boolean},scheduling:Omit<DuplicateScheduling,'checkpoint'>&{checkpoint?:(state:DuplicateCheckpoint)=>Promise<void>;checkpointIntervalMs?:number;maxStepsPerCheckpoint?:number}={}):Promise<DuplicateCheckpoint>{
   const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise<void>(resolve=>setTimeout(resolve,0)));
   if(!Number.isFinite(budget)||budget<=0||!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_copy_budget');
   const check=()=>{if(scheduling.signal?.aborted)throw fault('copy_aborted');};check();
+  const interval=scheduling.checkpointIntervalMs??0,maxCheckpointSteps=scheduling.maxStepsPerCheckpoint??8192;
+  if(!Number.isFinite(interval)||interval<0||!Number.isSafeInteger(maxCheckpointSteps)||maxCheckpointSteps<1)throw fault('invalid_copy_budget');
   const planner=createDuplicatePlanner(document,registry,selection,options,binding);
-  while(true){const start=now();let count=0;do{check();if(planner.ready())return planner.checkpoint();if(planner.step())throw fault('invalid_copy_stream_checkpoint');count++;}while(count<limit&&now()-start<budget);if(scheduling.checkpoint)await scheduling.checkpoint(planner.checkpoint());check();await yieldTask();}
+  let savedPhase:DuplicateCheckpoint['phase']|undefined,savedSteps=binding.checkpoint?.steps??0,savedAllocated=(binding.checkpoint?.elementIds.length??0)+(binding.checkpoint?.canvasIds.length??0),savedAt=now();
+  while(true){
+    const start=now();let count=0;
+    do{check();if(planner.ready()){const state=planner.checkpoint();if(interval&&scheduling.checkpoint){await scheduling.checkpoint(state);check();}return state;}if(planner.step())throw fault('invalid_copy_stream_checkpoint');count++;}while(count<limit&&now()-start<budget);
+    if(scheduling.checkpoint){
+      const progress=planner.progress(),at=now(),elapsed=at-savedAt;
+      if(!interval||progress.phase!==savedPhase||!savedAllocated&&progress.allocated>0||!Number.isFinite(elapsed)||elapsed<0||elapsed>=interval||progress.steps-savedSteps>=maxCheckpointSteps){
+        await scheduling.checkpoint(planner.checkpoint());
+        // Persistence latency is not more private planning work.
+        savedPhase=progress.phase;savedSteps=progress.steps;savedAllocated=progress.allocated;savedAt=now();
+      }
+    }
+    check();await yieldTask();
+  }
 }

src/lib/canvas-runtime/engine.ts

@@ -710,7 +710,7 @@ export async function openProjectEngine(options: EngineOptions) {
       validateLoadingCopySource(editor.getSnapshot(),admitted,document);
       if(copyRegistrySignature(document,options.registry)!==preparation.registrySignature)throw fault('copy_interpreter_changed');
       if(preparation.version===3||preparation.version===4){
-        const state=await prepareDuplicateStream(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner,topologyOnly:!!indexed},{signal:controller.signal,checkpoint:async state=>{
+        const state=await prepareDuplicateStream(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner,topologyOnly:!!indexed},{signal:controller.signal,checkpointIntervalMs:50,maxStepsPerCheckpoint:8192,checkpoint:async state=>{
           const job=check(),previous=job.preparation!;if(previous.planner&&state.steps<=previous.steps)return;
           publishImport(freeze({...job,generation:job.generation+1,preparation:{...previous,steps:state.steps,allocationIds:[...state.elementIds,...state.canvasIds].map(pair=>pair[1]),planner:state}}),job.generation);await persist();check();
         }}),job=check();assertNativeImportTarget(job,editor.getSnapshot().document);

tests/headless/duplicate-scheduling.test.mjs

@@ -1,7 +1,7 @@
 import {test} from 'node:test';
 import assert from 'node:assert/strict';
 import {createDocument,createRegistry,builtinComponents,planDuplicate,planDuplicateSteps,applyChanges,validateDocument,validateDocumentSteps,validJson} from '../../src/lib/canvas-core/index.ts';
-import {planDuplicateAsync} from '../../src/lib/canvas-runtime/duplicate.ts';
+import {planDuplicateAsync,prepareDuplicateStream} from '../../src/lib/canvas-runtime/duplicate.ts';
 const registry=createRegistry(builtinComponents());
 const node=(id,owner='P',componentId='text')=>({id,incarnation:1,ownerCanvasId:owner,componentId,dataVersion:1,structure:'node',data:{title:id,text:'copy'},geometry:{x:0,y:0,width:100,height:100,rotation:0}});
 const edge={id:'edge',incarnation:1,ownerCanvasId:'P',componentId:'connection',dataVersion:1,structure:'connection',source:{elementId:'C'},target:{elementId:'F'},path:null,data:{title:''}};
@@ -55,3 +55,18 @@ test('mapping output preserves special source IDs as own JSON fields without pro
  const plan=await planDuplicateAsync(source,registry,[{id:'__proto__',incarnation:1},{id:'constructor',incarnation:1}],options());
  assert.equal(Object.getPrototypeOf(plan.elementIds),Object.prototype);assert.equal(Object.hasOwn(plan.elementIds,'__proto__'),true);assert.equal(validJson(plan),true);assert.deepEqual(JSON.parse(JSON.stringify(plan.elementIds)),{['__proto__']:'new-1',constructor:'new-2'});
 });
+
+
+test('stream checkpoint cadence reduces full snapshots while preserving every task yield, phase and final mapping',async()=>{
+ const binding={sourceToken:'captured',schemaToken:'schema'},run=async interval=>{let yields=0;const checkpoints=[],state=await prepareDuplicateStream(document,registry,selection,options(),binding,{now:()=>0,maxStepsPerSlice:64,checkpointIntervalMs:interval,maxStepsPerCheckpoint:2048,yieldTask:async()=>{yields++;},checkpoint:async state=>{checkpoints.push(state);}});return{state,yields,checkpoints};},every=await run(0),paced=await run(50);
+ assert.deepEqual(paced.state,every.state);assert.equal(paced.yields,every.yields);assert.ok(paced.checkpoints.length<every.checkpoints.length/4);const phases=new Set(paced.checkpoints.map(state=>state.phase));for(const state of every.checkpoints)assert.ok(phases.has(state.phase));assert.equal(paced.checkpoints.at(-1).phase,'emit');assert.ok(paced.checkpoints.every(Object.isFrozen));for(let i=1;i<paced.checkpoints.length;i++)assert.ok(paced.checkpoints[i].steps-paced.checkpoints[i-1].steps<=2048+64);assert.equal(document.elements.size,2003);
+});
+test('paced preparation resumes allocated IDs exactly and checkpoint I/O time does not force the next snapshot',async()=>{
+ const binding={sourceToken:'captured',schemaToken:'schema'},controller=new AbortController(),allocation=options();let saved;
+ await assert.rejects(prepareDuplicateStream(document,registry,selection,allocation,binding,{signal:controller.signal,now:()=>0,maxStepsPerSlice:64,checkpointIntervalMs:50,maxStepsPerCheckpoint:512,yieldTask:async()=>{},checkpoint:async state=>{if(state.phase==='allocate-elements'&&state.elementIds.length){saved=state;controller.abort();}}}),/copy_aborted/);assert.ok(saved.elementIds.length>0);
+ const resumed=await prepareDuplicateStream(document,registry,selection,allocation,{...binding,checkpoint:saved},{now:()=>0,maxStepsPerSlice:64,yieldTask:async()=>{}}),complete=await prepareDuplicateStream(document,registry,selection,options(),binding,{now:()=>0,maxStepsPerSlice:64,yieldTask:async()=>{}});assert.deepEqual(resumed,complete);assert.deepEqual(resumed.elementIds.slice(0,saved.elementIds.length),saved.elementIds);
+ let time=0,yields=0;const saves=[];await prepareDuplicateStream(document,registry,selection,options(),binding,{now:()=>time,maxStepsPerSlice:64,checkpointIntervalMs:10,yieldTask:async()=>{time++;yields++;},checkpoint:async state=>{const start=time;time+=1000;saves.push({phase:state.phase,start,end:time});}});assert.ok(saves.length<yields/3);for(let i=1;i<saves.length;i++)if(saves[i].phase===saves[i-1].phase)assert.ok(saves[i].start-saves[i-1].end>=10);
+});
+test('invalid stream checkpoint cadence rejects before allocation or publication',async()=>{
+ for(const setting of [{checkpointIntervalMs:-1},{checkpointIntervalMs:NaN},{checkpointIntervalMs:Infinity},{maxStepsPerCheckpoint:0},{maxStepsPerCheckpoint:1.5}]){let allocated=0,published=0;await assert.rejects(prepareDuplicateStream(document,registry,selection,{createId:()=>{allocated++;return'new';}},{sourceToken:'captured',schemaToken:'schema'},{...setting,checkpoint:async()=>{published++;}}),/invalid_copy_budget/);assert.equal(allocated,0);assert.equal(published,0);}
+});