From 2452fd6ae9ade9f67784159c68b4193039e9bd54 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Tue, 10 Nov 2026 18:11:25 +0800 Subject: [PATCH] Schedule copy snapshot preparation by time and byte budgets --- headless/README.md | 11 +++++++ src/lib/canvas-runtime/copy-budget.ts | 13 ++++++++ src/lib/canvas-runtime/copy-snapshot-index.ts | 28 ++++++++-------- src/lib/canvas-runtime/copy-snapshot.ts | 24 +++++++------- src/lib/canvas-runtime/import-duplicate.ts | 9 +++-- tests/headless/copy-budget.test.mjs | 33 +++++++++++++++++++ 6 files changed, 91 insertions(+), 27 deletions(-) create mode 100644 src/lib/canvas-runtime/copy-budget.ts create mode 100644 tests/headless/copy-budget.test.mjs diff --git a/headless/README.md b/headless/README.md index 3c1daa0..bc179fb 100644 --- a/headless/README.md +++ b/headless/README.md @@ -936,3 +936,14 @@ HEADLESS_PERFORMANCE_COUNT=10000 HEADLESS_PERFORMANCE_REPORT=/private/tmp/headle Both modes retain the same full-copy, upper-canvas edge, synchronization and far-card viewport assertions. Production here means an optimized frontend served against the fixture HTTP service; it does not claim production Foundry validation. + +Copy snapshot encoding/index scanning and legacy prepared-source encoding share +the import time-slice scheduler through an internal copy-budget adapter. They no +longer sleep every 32 records. Record bytes contribute weighted work so larger +records trigger budget checks sooner; the existing coarse-clock work ceiling is +retained. Topology/full validation iterators use the same slice budget. This is +a soft scheduling budget: JSON operations, document construction and a single +validation step remain indivisible. Copy cancellation still reports +`copy_aborted`; host clock/yield errors remain observable. Snapshot formats, +byte locators, size limits, source digests, lazy payload reads and admission/seal +ordering are unchanged. No partially encoded snapshot is admitted or returned. diff --git a/src/lib/canvas-runtime/copy-budget.ts b/src/lib/canvas-runtime/copy-budget.ts new file mode 100644 index 0000000..ee66a52 --- /dev/null +++ b/src/lib/canvas-runtime/copy-budget.ts @@ -0,0 +1,13 @@ +import {fault} from '../canvas-core/values.ts'; +import {createImportBudget,type ImportScheduling} from './import-budget.ts'; +/** Share import scheduling while retaining the copy cancellation/error contract. */ +export function createCopyBudget(signal:AbortSignal,options:ImportScheduling={}){ + if(signal.aborted)throw fault('copy_aborted'); + if(options.frameBudgetMs!==undefined&&(!Number.isFinite(options.frameBudgetMs)||options.frameBudgetMs<=0))throw fault('invalid_copy_budget'); + const budget=createImportBudget(signal,options); + const fail=(error:unknown):never=>{if(signal.aborted)throw fault('copy_aborted');throw error;}; + return Object.freeze({step(units=1):Promise|undefined{ + if(signal.aborted)throw fault('copy_aborted'); + try{return budget.step(units)?.catch(fail);}catch(error){return fail(error);} + }}); +} diff --git a/src/lib/canvas-runtime/copy-snapshot-index.ts b/src/lib/canvas-runtime/copy-snapshot-index.ts index e852c4e..e334cd3 100644 --- a/src/lib/canvas-runtime/copy-snapshot-index.ts +++ b/src/lib/canvas-runtime/copy-snapshot-index.ts @@ -2,7 +2,8 @@ import type {CanvasRecord,Document,ElementRecord,Id,Json,Registry,Data} from '.. import {createDocument,validateDocumentTopologySteps} from '../canvas-core/document.ts'; import {equal,fault,freeze,validJson} from '../canvas-core/values.ts'; import {createNdjsonImportSource} from './imports.ts'; -const task=()=>new Promise(resolve=>setTimeout(resolve,0)); +import {createCopyBudget} from './copy-budget.ts'; +import type {ImportScheduling} from './import-budget.ts'; /** Only ownership, endpoint and geometry fields needed to decide scope/coordinates. No component content. */ export function copyTopology(record:ElementRecord):ElementRecord{ const raw=record.data.presentation,presentation=raw&&typeof raw==='object'&&!Array.isArray(raw)?Object.fromEntries(['contentScale','unitScale'].filter(key=>Object.hasOwn(raw,key)).map(key=>[key,raw[key]])):{}; @@ -11,25 +12,25 @@ export function copyTopology(record:ElementRecord):ElementRecord{ return record.structure==='node'?{...common,structure:'node',geometry:record.geometry}:{...common,structure:'connection',source:record.source,target:record.target,path:null}; } /** Format 2 places a compact topology/byte-locator index before the uninterpreted payload region. */ -export async function encodeIndexedCopySnapshot(document:Document,signal:AbortSignal):Promise{ - const index:BlobPart[]=[],payload:BlobPart[]=[];let indexBytes=0,payloadBytes=0,count=0; +export async function encodeIndexedCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ + const index:BlobPart[]=[],payload:BlobPart[]=[];const budget=createCopyBudget(signal,scheduling);let indexBytes=0,payloadBytes=0; function encode(value:unknown){if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');return{text,size};} - function append(value:unknown){const part=encode(value);index.push(part.text);indexBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');} + function append(value:unknown){const part=encode(value);index.push(part.text);indexBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');return part.size;} append({kind:'header',version:2,projectId:document.projectId}); - for(const record of document.canvases.values()){append({kind:'canvas',record});if(++count%32===0)await task();} + for(const record of document.canvases.values()){const size=append({kind:'canvas',record});await budget.step(Math.max(1,Math.ceil(size/1024)));} for(const record of document.elements.values()){ - const part=encode({kind:'element',record});append({kind:'element-index',record:copyTopology(record),offset:payloadBytes,byteLength:part.size});payload.push(part.text);payloadBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');if(++count%32===0)await task(); + const part=encode({kind:'element',record}),indexSize=append({kind:'element-index',record:copyTopology(record),offset:payloadBytes,byteLength:part.size});payload.push(part.text);payloadBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');await budget.step(Math.max(1,Math.ceil((part.size+indexSize)/1024))); } - for(const record of document.tombstones){append({kind:'tombstone',record});if(++count%32===0)await task();} + for(const record of document.tombstones){const size=append({kind:'tombstone',record});await budget.step(Math.max(1,Math.ceil(size/1024)));} append({kind:'payload',byteLength:payloadBytes});if(signal.aborted)throw fault('copy_aborted');return new Blob([...index,...payload],{type:'application/x-ndjson'}); } export type IndexedCopySnapshot={readonly document:Document;readElement(id:Id,signal:AbortSignal):Promise}; /** Scan only the small index. Component JSON/schema checks occur just before each record enters the runner. */ -export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:Registry,signal:AbortSignal):Promise{ +export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source'); - const source=createNdjsonImportSource({bytes,fingerprint:'indexed-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map(),locators=new Map();let cursor=0,header=false,payloadOffset=-1,expectedOffset=0,count=0; + const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'indexed-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map(),locators=new Map();let cursor=0,header=false,payloadOffset=-1,expectedOffset=0; while(true){ - if(signal.aborted)throw fault('copy_aborted');const line=await source.read(cursor,signal);if(!line||line.error||line.skip||!line.value)throw fault('invalid_copy_snapshot');cursor=line.nextCursor; + if(signal.aborted)throw fault('copy_aborted');const start=cursor,line=await source.read(cursor,signal);if(!line||line.error||line.skip||!line.value)throw fault('invalid_copy_snapshot');cursor=line.nextCursor; const value=line.value as Record; if(!header){if(value.kind!=='header'||value.version!==2||value.projectId!==projectId)throw fault('invalid_copy_snapshot');header=true;continue;} if(value.kind==='payload'){if(value.byteLength!==expectedOffset||cursor+expectedOffset!==bytes.size)throw fault('invalid_copy_snapshot');payloadOffset=cursor;break;} @@ -39,10 +40,11 @@ export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:R if(!record||!Number.isSafeInteger(offset)||offset!==expectedOffset||!Number.isSafeInteger(byteLength)||byteLength<1||byteLength>256*1024||locators.has(record.id)||!equal(copyTopology(record),record))throw fault('invalid_copy_snapshot'); locators.set(record.id,{offset,byteLength});elements.push(record);expectedOffset+=byteLength;if(expectedOffset>bytes.size)throw fault('invalid_copy_snapshot'); }else if(value.kind==='tombstone'){const row=value.record as [Id,number];if(!Array.isArray(row)||row.length!==2||tombstones.has(row[0]))throw fault('invalid_copy_snapshot');tombstones.set(row[0],row[1]);}else throw fault('invalid_copy_snapshot'); - if(++count%32===0)await task(); + await budget.step(Math.max(1,Math.ceil((cursor-start)/1024))); } - const document=createDocument(projectId,canvases,elements,tombstones);let checks=0; - for(const _ of validateDocumentTopologySteps(document,registry)){if(signal.aborted)throw fault('copy_aborted');if(++checks%512===0)await task();} + const document=createDocument(projectId,canvases,elements,tombstones); + for(const _ of validateDocumentTopologySteps(document,registry))await budget.step(); + if(signal.aborted)throw fault('copy_aborted'); // Cache raw byte windows, never decoded component objects. Preserve parser batching across Blob task boundaries. const windows=new Map(),windowBytes=64*1024;let cachedBytes=0; return Object.freeze({document,async readElement(id:Id,signal:AbortSignal){ diff --git a/src/lib/canvas-runtime/copy-snapshot.ts b/src/lib/canvas-runtime/copy-snapshot.ts index d879efb..8d03725 100644 --- a/src/lib/canvas-runtime/copy-snapshot.ts +++ b/src/lib/canvas-runtime/copy-snapshot.ts @@ -3,27 +3,29 @@ import { createDocument, validateDocumentSteps } from '../canvas-core/document.t import { fault } from '../canvas-core/values.ts'; import { createNdjsonImportSource } from './imports.ts'; -const task=()=>new Promise(resolve=>setTimeout(resolve,0)); +import {createCopyBudget} from './copy-budget.ts'; +import type {ImportScheduling} from './import-budget.ts'; /** Capture immutable source bytes before a preparation job is admitted. No mutable live getters. */ -export async function encodeCopySnapshot(document:Document,signal:AbortSignal):Promise{ - const parts:BlobPart[]=[];let bytes=0,count=0; - const append=(value:unknown)=>{if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');bytes+=size;if(bytes>64*1024*1024)throw fault('invalid_import_source');parts.push(text);}; +export async function encodeCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ + const budget=createCopyBudget(signal,scheduling),parts:BlobPart[]=[];let bytes=0; + const append=(value:unknown)=>{if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');bytes+=size;if(bytes>64*1024*1024)throw fault('invalid_import_source');parts.push(text);return size;}; append({kind:'header',version:1,projectId:document.projectId}); - for(const[kind,records]of [['canvas',document.canvases.values()],['element',document.elements.values()],['tombstone',document.tombstones.entries()]] as const)for(const record of records){append({kind,record});if(++count%32===0)await task();} + for(const[kind,records]of [['canvas',document.canvases.values()],['element',document.elements.values()],['tombstone',document.tombstones.entries()]] as const)for(const record of records){const size=append({kind,record});await budget.step(Math.max(1,Math.ceil(size/1024)));} if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'}); } -export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Registry,signal:AbortSignal):Promise{ - const source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map();let cursor=0,header=false,count=0; +export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise{ + const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map();let cursor=0,header=false; while(true){ - if(signal.aborted)throw fault('copy_aborted');const line=await source.read(cursor,signal);if(!line)break;cursor=line.nextCursor; + if(signal.aborted)throw fault('copy_aborted');const start=cursor,line=await source.read(cursor,signal);if(!line)break;cursor=line.nextCursor; const value=line.value as unknown as {kind:string;version?:number;projectId?:string;record:never};if(line.error||line.skip||!value)throw fault('invalid_copy_snapshot'); if(!header){if(value.kind!=='header'||value.version!==1||value.projectId!==projectId)throw fault('invalid_copy_snapshot');header=true;continue;} if(value.kind==='canvas')canvases.push(value.record);else if(value.kind==='element')elements.push(value.record);else if(value.kind==='tombstone'){const record=value.record as [string,number];if(!Array.isArray(record)||record.length!==2||tombstones.has(record[0]))throw fault('invalid_copy_snapshot');tombstones.set(record[0],record[1]);}else throw fault('invalid_copy_snapshot'); - if(++count%32===0)await task(); + await budget.step(Math.max(1,Math.ceil((cursor-start)/1024))); } if(!header||signal.aborted)throw fault(signal.aborted?'copy_aborted':'invalid_copy_snapshot'); - const document=createDocument(projectId,canvases,elements,tombstones);let checks=0; - for(const _ of validateDocumentSteps(document,registry)){if(signal.aborted)throw fault('copy_aborted');if(++checks%512===0)await task();} + const document=createDocument(projectId,canvases,elements,tombstones); + for(const _ of validateDocumentSteps(document,registry))await budget.step(); + if(signal.aborted)throw fault('copy_aborted'); return document; } diff --git a/src/lib/canvas-runtime/import-duplicate.ts b/src/lib/canvas-runtime/import-duplicate.ts index 060c6b7..426685b 100644 --- a/src/lib/canvas-runtime/import-duplicate.ts +++ b/src/lib/canvas-runtime/import-duplicate.ts @@ -4,22 +4,25 @@ import type { DuplicatePlan } from '../canvas-core/duplicate.ts'; import { clone, equal, fault, validJson } from '../canvas-core/values.ts'; import type { ImportJob, ImportLine, ImportUnit } from './imports.ts'; +import {createCopyBudget} from './copy-budget.ts'; +import type {ImportScheduling} from './import-budget.ts'; + type ExternalElement = {id: Id; incarnation: number; ownerCanvasId: Id}; type ExternalCanvas = {id: Id; parent: CanvasRecord['parent']; entryIncarnation?: number}; type CopyLine = {key: string; change: Exclude; dependencies: string[]; externalElements: ExternalElement[]; externalCanvases: ExternalCanvas[]}; /** Internal source: already mapped records, with explicit guards for every retained external reference. */ -export async function encodeDuplicateSource(document: Document, registry: Registry, plan: DuplicatePlan, signal: AbortSignal): Promise { +export async function encodeDuplicateSource(document: Document, registry: Registry, plan: DuplicatePlan, signal: AbortSignal, scheduling:ImportScheduling={}): Promise { const keys = new Map(),parts: BlobPart[] = []; for (const [id,mapped] of Object.entries(plan.elementIds)) keys.set(mapped,`element:${id}`); for (const [id,mapped] of Object.entries(plan.canvasIds)) keys.set(mapped,`canvas:${id}`); - let bytes=0,count=0; + const budget=createCopyBudget(signal,scheduling);let bytes=0; for (const change of plan.changes) { if(signal.aborted)throw fault('copy_aborted'); if(change.kind==='field'||!change.after||change.before!==null)throw fault('invalid_duplicate_plan'); const line=duplicateImportLine(document,registry,change,keys); const text=JSON.stringify(line)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');bytes+=size;if(bytes>64*1024*1024)throw fault('invalid_import_source');parts.push(text); - if(++count%32===0)await new Promise(resolve=>setTimeout(resolve,0)); + await budget.step(Math.max(1,Math.ceil(size/1024))); } if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'}); } diff --git a/tests/headless/copy-budget.test.mjs b/tests/headless/copy-budget.test.mjs new file mode 100644 index 0000000..a8e2e68 --- /dev/null +++ b/tests/headless/copy-budget.test.mjs @@ -0,0 +1,33 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createDocument,createRegistry,builtinComponents,planDuplicate} from '../../src/lib/canvas-core/index.ts'; +import {encodeCopySnapshot,decodeCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot.ts'; +import {encodeIndexedCopySnapshot,openIndexedCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot-index.ts'; +import {encodeDuplicateSource} from '../../src/lib/canvas-runtime/import-duplicate.ts'; +import {createCopyBudget} from '../../src/lib/canvas-runtime/copy-budget.ts'; +const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal; +const node=i=>({id:`n-${i}`,incarnation:1,ownerCanvasId:'canvas',componentId:'text',dataVersion:1,structure:'node',data:{title:`卡片 ${i}`,text:'emoji 👀\n"quoted"'},geometry:{x:i,y:0,width:100,height:100,rotation:0}}); +const document=createDocument('project',[{id:'canvas',parent:null}],Array.from({length:512},(_,i)=>node(i)),new Map([['retired',2]])); +function scheduling(slow=false){let yields=0,tick=0;return{options:{now:()=>slow?(++tick)*10:0,yieldTask:async()=>{yields++;}},get yields(){return yields;}};} + +test('copy snapshot bytes and decoded records are invariant across cheap and expensive scheduling slices',async()=>{ + for(const [encode,decode,indexed] of [[encodeCopySnapshot,decodeCopySnapshot,false],[encodeIndexedCopySnapshot,openIndexedCopySnapshot,true]]){ + const cheap=scheduling(),costly=scheduling(true),a=await encode(document,signal(),cheap.options),b=await encode(document,signal(),costly.options);assert.equal(await a.text(),await b.text());assert.equal(cheap.yields,0,'cheap work slept at a fixed record count');assert.ok(costly.yields>0);const before=costly.yields,snapshot=await decode(b,'project',registry,signal(),costly.options),restored=indexed?snapshot.document:snapshot;assert.equal(restored.elements.size,512);assert.deepEqual([...restored.tombstones],[['retired',2]]);assert.ok(costly.yields>before);if(indexed)assert.deepEqual(await snapshot.readElement('n-511',signal()),document.elements.get('n-511'));else assert.deepEqual([...restored.elements],[...document.elements]); + } +}); +test('abort at a scheduled yield prevents copy encoders and readers from returning a partial snapshot',async()=>{ + const old=await encodeCopySnapshot(document,signal(),scheduling().options),indexed=await encodeIndexedCopySnapshot(document,signal(),scheduling().options); + for(const run of [(s,o)=>encodeCopySnapshot(document,s,o),(s,o)=>encodeIndexedCopySnapshot(document,s,o),(s,o)=>decodeCopySnapshot(old,'project',registry,s,o),(s,o)=>openIndexedCopySnapshot(indexed,'project',registry,s,o)]){const controller=new AbortController();let time=0,yields=0;await assert.rejects(run(controller.signal,{now:()=>time+=10,yieldTask:async()=>{yields++;controller.abort();}}),error=>error.code==='copy_aborted');assert.equal(yields,1);} + assert.equal(document.elements.size,512);assert.equal(document.elements.get('n-0').data.title,'卡片 0'); +}); +test('large records consume work budget and coarse clocks still yield without changing copy cancellation errors',async()=>{ + const controller=new AbortController();let yields=0;const budget=createCopyBudget(controller.signal,{now:()=>0,yieldTask:async()=>{yields++;}});for(let i=0;i<64;i++)await budget.step(256);assert.ok(yields>0,'coarse clock allowed an unbounded byte workload');controller.abort();assert.throws(()=>budget.step(),error=>error.code==='copy_aborted'); + const aborted=new AbortController();aborted.abort();assert.throws(()=>createCopyBudget(aborted.signal),error=>error.code==='copy_aborted'); +}); +test('copy budget rejects invalid scheduling and preserves host scheduling failures',async()=>{ + for(const frameBudgetMs of [0,-1,NaN,Infinity])await assert.rejects(encodeIndexedCopySnapshot(document,signal(),{frameBudgetMs}),error=>error.code==='invalid_copy_budget'); + const failure=Error('host unavailable');let time=0;await assert.rejects(encodeCopySnapshot(document,signal(),{now:()=>time+=10,yieldTask:async()=>{throw failure;}}),error=>error===failure); +}); +test('legacy prepared duplicate sources preserve exact bytes and abort before returning on a budget boundary',async()=>{ + let id=0;const plan=planDuplicate(document,registry,[...document.elements.values()].map(({id,incarnation})=>({id,incarnation})),{createId:()=>`copy-${++id}`}),cheap=scheduling(),costly=scheduling(true),a=await encodeDuplicateSource(document,registry,plan,signal(),cheap.options),b=await encodeDuplicateSource(document,registry,plan,signal(),costly.options);assert.equal(await a.text(),await b.text());assert.equal(cheap.yields,0);assert.ok(costly.yields>0);const controller=new AbortController();let time=0;await assert.rejects(encodeDuplicateSource(document,registry,plan,controller.signal,{now:()=>time+=10,yieldTask:async()=>controller.abort()}),error=>error.code==='copy_aborted'); +});