Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Schedule copy snapshot preparation by time and byte budgets

Browse Source
This commit is contained in:
Codex Bot
2026-11-10 18:11:25 +08:00
parent afccb8c95c
commit 2452fd6ae9
6 changed files with 91 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
headless/README.md
+11
View File
@@ -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.
src/lib/canvas-runtime/copy-budget.ts
+13
View File
@@ -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<void>|undefined{
if(signal.aborted)throw fault('copy_aborted');
try{return budget.step(units)?.catch(fail);}catch(error){return fail(error);}
}});
}
src/lib/canvas-runtime/copy-snapshot-index.ts
+15 -13
View File
@@ -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<void>(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<Blob>{
const index:BlobPart[]=[],payload:BlobPart[]=[];let indexBytes=0,payloadBytes=0,count=0;
export async function encodeIndexedCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise<Blob>{
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<ElementRecord>};
/** 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<IndexedCopySnapshot>{
export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise<IndexedCopySnapshot>{
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<Id,number>(),locators=new Map<Id,{offset:number;byteLength:number}>();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<Id,number>(),locators=new Map<Id,{offset:number;byteLength:number}>();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<string,Json>;
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<number,Uint8Array>(),windowBytes=64*1024;let cachedBytes=0;
return Object.freeze({document,async readElement(id:Id,signal:AbortSignal){
src/lib/canvas-runtime/copy-snapshot.ts
+13 -11
View File
@@ -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<void>(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<Blob>{
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<Blob>{
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<Document>{
const source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map<string,number>();let cursor=0,header=false,count=0;
export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise<Document>{
const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map<string,number>();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;
}
src/lib/canvas-runtime/import-duplicate.ts
+6 -3
View File
@@ -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<Change,{kind:'field'}>; 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<Blob> {
export async function encodeDuplicateSource(document: Document, registry: Registry, plan: DuplicatePlan, signal: AbortSignal, scheduling:ImportScheduling={}): Promise<Blob> {
const keys = new Map<Id,string>(),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<void>(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'});
}
tests/headless/copy-budget.test.mjs
+33
View File
@@ -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');
});