From 64ed8fe57a58dd427089d46e6b2bf3a953451ec5 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Tue, 10 Nov 2026 19:28:10 +0800 Subject: [PATCH] Separate cooperative copy planning from durable checkpoint cadence --- headless/README.md | 13 +++++++++++++ src/lib/canvas-core/duplicate-checkpoint.ts | 1 + src/lib/canvas-runtime/duplicate.ts | 19 +++++++++++++++++-- src/lib/canvas-runtime/engine.ts | 2 +- tests/headless/duplicate-scheduling.test.mjs | 17 ++++++++++++++++- 5 files changed, 48 insertions(+), 4 deletions(-) diff --git a/headless/README.md b/headless/README.md index e070eb6..95a19c1 100644 --- a/headless/README.md +++ b/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. diff --git a/src/lib/canvas-core/duplicate-checkpoint.ts b/src/lib/canvas-core/duplicate-checkpoint.ts index 2acfb43..b593e50 100644 --- a/src/lib/canvas-core/duplicate-checkpoint.ts +++ b/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;} diff --git a/src/lib/canvas-runtime/duplicate.ts b/src/lib/canvas-runtime/duplicate.ts index b664e57..7f14596 100644 --- a/src/lib/canvas-runtime/duplicate.ts +++ b/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&{checkpoint?:(state:DuplicateCheckpoint)=>Promise}={}):Promise{ +export async function prepareDuplicateStream(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint;topologyOnly?:boolean},scheduling:Omit&{checkpoint?:(state:DuplicateCheckpoint)=>Promise;checkpointIntervalMs?:number;maxStepsPerCheckpoint?:number}={}):Promise{ const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise(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(count0||!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(); + } } diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 55863d3..b3f5260 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/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); diff --git a/tests/headless/duplicate-scheduling.test.mjs b/tests/headless/duplicate-scheduling.test.mjs index 246dc07..e78687b 100644 --- a/tests/headless/duplicate-scheduling.test.mjs +++ b/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.lengthstate.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{ + 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=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);} +});