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

Decode duplicate payloads on demand from indexed snapshots

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 20:28:09 +08:00
parent b5ad3e0cc9
commit 4035956607
11 changed files with 146 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-core/document.ts
+5 -2
View File
@@ -35,7 +35,7 @@ export function commonCanvas(document: Document, left: Id, right: Id): Id {
return result;
}
/** Cooperative validation over an immutable document; synchronous callers drain the same checks. */
export function* validateDocumentSteps(document: Document, registry: Registry): Generator<void, void> {
function* validateDocumentChecks(document: Document, registry: Registry, componentData:boolean): Generator<void, void> {
if (!document.canvases.size) throw fault('missing_root');
for (const [id, incarnation] of document.tombstones) { yield; if (!id || !Number.isSafeInteger(incarnation) || incarnation < 1) throw fault('invalid_tombstone', [id]); }
const entries = new Set<Id>();
@@ -56,7 +56,7 @@ export function* validateDocumentSteps(document: Document, registry: Registry):
const dead = document.tombstones.get(element.id);
if (dead !== undefined && element.incarnation <= dead) throw fault('stale_incarnation', [element.id]);
const definition = registry.components.get(element.componentId);
if (registry.resolve(element) === 'known' && (!definition || definition.structure !== element.structure || !definition.validate(element.data))) throw fault('invalid_component_data', [element.id]);
if (componentData && registry.resolve(element) === 'known' && (!definition || definition.structure !== element.structure || !definition.validate(element.data))) throw fault('invalid_component_data', [element.id]);
if (element.structure === 'node') {
if (!geometryValid(element.geometry)) throw fault('invalid_geometry', [element.id]);
} else {
@@ -67,6 +67,9 @@ export function* validateDocumentSteps(document: Document, registry: Registry):
}
}
}
/** Structural indexes contain no component payload; use only for source planning, never editor publication. */
export function* validateDocumentTopologySteps(document:Document,registry:Registry):Generator<void,void>{yield* validateDocumentChecks(document,registry,false);}
export function* validateDocumentSteps(document:Document,registry:Registry):Generator<void,void>{yield* validateDocumentChecks(document,registry,true);}
export function validateDocument(document: Document, registry: Registry): void {
for (const _ of validateDocumentSteps(document, registry)) { /* Drain identical validation without scheduling. */ }
}
src/lib/canvas-core/duplicate-checkpoint.ts
+3 -3
View File
@@ -1,5 +1,5 @@
import type { CanvasRecord, Change, Document, ElementRecord, Id, Point, Registry, Target } from './types.ts';
import { ancestors, targetElement, validateDocumentSteps } from './document.ts';
import { ancestors, targetElement, validateDocumentSteps, validateDocumentTopologySteps } from './document.ts';
import { duplicateTreeChange, duplicateEdgeChange, duplicateOverlay, type DuplicateOptions, type DuplicatePlan } from './duplicate.ts';
import { clone, equal, fault, freeze, validJson } from './values.ts';
const phases=['selection','canvases','members','roots','seed','dfs','edges','allocate-elements','allocate-canvases','mapping','emit','emit-edges','validate','done'] as const;
@@ -17,7 +17,7 @@ export function validateDuplicateCheckpoint(value:DuplicateCheckpoint):void{
for(const list of [value.children,value.elementIds,value.canvasIds])if(!Array.isArray(list)||list.some(pair=>!ids(pair)||pair.length!==2)||new Set(list.map(pair=>pair[0])).size!==list.length)throw fault('invalid_copy_checkpoint');
if(!Array.isArray(value.members)||value.members.some(pair=>!Array.isArray(pair)||pair.length!==2||typeof pair[0]!=='string'||!pair[0]||!ids(pair[1]))||new Set(value.members.map(pair=>pair[0])).size!==value.members.length||value.emitted>value.order.length||value.emittedEdges>value.edges.length||value.emittedEdges>0&&value.emitted!==value.order.length||value.expanding&&(typeof value.expanding.canvas!=='string'||!value.expanding.canvas||!Number.isSafeInteger(value.expanding.index)||value.expanding.index< -1))throw fault('invalid_copy_checkpoint');
}
export function createDuplicatePlanner(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint}){
export function createDuplicatePlanner(document:Document,registry:Registry,selection:readonly Target[],options:DuplicateOptions,binding:{sourceToken:string;schemaToken:string;checkpoint?:DuplicateCheckpoint;topologyOnly?:boolean}){
const offset=options.worldOffset??{x:32,y:32},idFactory=options.createId,suppliedMapping=options.mapping?freeze(clone(options.mapping)):undefined;
const state:MutableCheckpoint=binding.checkpoint?clone(binding.checkpoint) as MutableCheckpoint:{version:1,projectId:document.projectId,sourceToken:binding.sourceToken,schemaToken:binding.schemaToken,selection:clone(selection) as Target[],offset:clone(offset),...(suppliedMapping?{mapping:suppliedMapping}:{}),phase:'selection',cursor:0,steps:0,selected:[],children:[],members:[],roots:[],copied:[],canvases:[],order:[],work:[],edges:[],elementIds:[],canvasIds:[],emitted:0,emittedEdges:0};
validateDuplicateCheckpoint(state);
@@ -33,7 +33,7 @@ export function createDuplicatePlanner(document:Document,registry:Registry,selec
const limits:Partial<Record<DuplicateCheckpoint['phase'],number>>={selection:state.selection.length,canvases:sourceCanvases.length,members:sourceElements.length,roots:state.selected.length,seed:state.roots.length,edges:sourceElements.length,'allocate-elements':state.copied.length,'allocate-canvases':state.canvases.length};if(limits[state.phase]!==undefined&&state.cursor>limits[state.phase]!)throw fault('invalid_copy_checkpoint');
const supplied=suppliedMapping?[...Object.keys(suppliedMapping.elements).map(id=>({id,kind:'element' as const})),...Object.keys(suppliedMapping.canvases).map(id=>({id,kind:'canvas' as const}))]:[];
const outputElements=new Map<Id,ElementRecord>(),outputCanvases=new Map<Id,CanvasRecord>(),output=duplicateOverlay(document,outputElements,outputCanvases),changes:Change[]=[];
let treeCache=0,edgeCache=0,sourceChecks:Generator<void>|undefined=validateDocumentSteps(document,registry),outputChecks:Generator<void>|undefined,validated=false;
let treeCache=0,edgeCache=0,sourceChecks:Generator<void>|undefined=(binding.topologyOnly?validateDocumentTopologySteps:validateDocumentSteps)(document,registry),outputChecks:Generator<void>|undefined,validated=false;
const phase=(next:DuplicateCheckpoint['phase'],cursor=0)=>{state.phase=next;state.cursor=cursor;};
function add(change:Exclude<Change,{kind:'field'}>){freeze(change);changes.push(change);if(change.kind==='element')outputElements.set(change.id,change.after!);else outputCanvases.set(change.id,change.after!);}
function tree(index:number){const task=state.order[index];add(duplicateTreeChange(document,registry,task.kind,(task.kind==='element'?document.elements:document.canvases).get(task.id)!,elementIds,canvasIds,canvases,state.offset));}
src/lib/canvas-runtime/copy-snapshot-index.ts
+58
View File
@@ -0,0 +1,58 @@
import type {CanvasRecord,Document,ElementRecord,Id,Json,Registry,Data} from '../canvas-core/types.ts';
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));
/** 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]])):{};
const data:Data=Object.keys(presentation).length?{presentation}:{};
const common={id:record.id,incarnation:record.incarnation,ownerCanvasId:record.ownerCanvasId,componentId:record.componentId,dataVersion:record.dataVersion,data};
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;
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');}
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.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();
}
for(const record of document.tombstones){append({kind:'tombstone',record});if(++count%32===0)await task();}
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>{
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;
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;
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;}
if(value.kind==='canvas')canvases.push(value.record as unknown as CanvasRecord);
else if(value.kind==='element-index'){
const record=value.record as unknown as ElementRecord,offset=value.offset as number,byteLength=value.byteLength as number;
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();
}
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();}
// 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){
if(signal.aborted)throw fault('copy_aborted');const locator=locators.get(id);if(!locator)throw fault('element_missing',[id]);
const start=Math.floor(locator.offset/windowBytes)*windowBytes,required=locator.offset+locator.byteLength-start;let window=windows.get(start);
if(!window||window.byteLength<required){const raw=await bytes.slice(payloadOffset+start,Math.min(bytes.size,payloadOffset+start+Math.max(windowBytes,required))).arrayBuffer();if(signal.aborted)throw fault('copy_aborted');const previous=windows.get(start);if(previous)cachedBytes-=previous.byteLength;window=new Uint8Array(raw);windows.delete(start);windows.set(start,window);cachedBytes+=window.byteLength;while(windows.size>4||cachedBytes>1024*1024){const[key,value]=windows.entries().next().value!;windows.delete(key);cachedBytes-=value.byteLength;}}
const raw=window.subarray(locator.offset-start,required);let value:{kind:string;record:ElementRecord};
try{value=JSON.parse(new TextDecoder('utf-8',{fatal:true}).decode(raw));}catch{throw fault('invalid_copy_snapshot');}
if(!value||value.kind!=='element'||!validJson(value.record)||!value.record.data||typeof value.record.data!=='object'||Array.isArray(value.record.data)||!equal(copyTopology(value.record),document.elements.get(id)))throw fault('invalid_copy_snapshot');
const record=value.record,definition=registry.components.get(record.componentId);if(registry.resolve(record)==='known'&&(!definition||definition.structure!==record.structure||!definition.validate(record.data)))throw fault('invalid_component_data',[id]);
return freeze(record);
}});
}
src/lib/canvas-runtime/duplicate-source.ts
+4 -3
View File
@@ -8,10 +8,10 @@ import { duplicateImportLine } from './import-duplicate.ts';
import type { ImportSource } from './imports.ts';
/** A seekable record cursor over the original SHA-bound snapshot, not a second mapped Blob. */
export function createDuplicatePlanSource(document:Document,registry:Registry,state:DuplicateCheckpoint,fingerprint:string):ImportSource{
export function createDuplicatePlanSource(document:Document,registry:Registry,state:DuplicateCheckpoint,fingerprint:string,readElement?:(id:Id,signal:AbortSignal)=>Promise<ElementRecord>):ImportSource{
if(state.phase!=='emit'||state.emitted!==0||state.emittedEdges!==0||state.schemaToken!==copyRegistrySignature(document,registry))throw fault('copy_interpreter_changed');
// Structural/source binding checks; the decoded source has already passed document validation.
createDuplicatePlanner(document,registry,state.selection,{createId:()=>{throw fault('unstable_import_identity');},worldOffset:state.offset,...(state.mapping?{mapping:state.mapping}:{})},{sourceToken:state.sourceToken,schemaToken:state.schemaToken,checkpoint:state});
createDuplicatePlanner(document,registry,state.selection,{createId:()=>{throw fault('unstable_import_identity');},worldOffset:state.offset,...(state.mapping?{mapping:state.mapping}:{})},{sourceToken:state.sourceToken,schemaToken:state.schemaToken,checkpoint:state,topologyOnly:!!readElement});
const elements=new Map(state.elementIds),canvases=new Map(state.canvasIds),copiedCanvases=new Set(state.canvases),keys=new Map<Id,string>(),reverseElements=new Map(state.elementIds.map(([id,mapped])=>[mapped,id])),reverseCanvases=new Map(state.canvasIds.map(([id,mapped])=>[mapped,id]));
for(const[id,mapped]of elements)keys.set(mapped,`element:${id}`);for(const[id,mapped]of canvases)keys.set(mapped,`canvas:${id}`);
// Edge geometry needs mapped endpoint/owner records. Compute these on demand; retain a bounded cache.
@@ -24,7 +24,8 @@ export function createDuplicatePlanSource(document:Document,registry:Registry,st
return Object.freeze({fingerprint,parserVersion:2,async read(cursor:number,signal:AbortSignal){
if(signal.aborted)throw fault('import_aborted');if(!Number.isSafeInteger(cursor)||cursor<0||cursor>length)throw fault('invalid_import_cursor');if(cursor===length)return null;
const task=state.order[cursor],key=task?`${task.kind}:${task.id}`:`element:${state.edges[cursor-state.order.length]}`;
try{const change=cursor<state.order.length?duplicateTreeChange(document,registry,task.kind,(task.kind==='element'?document.elements:document.canvases).get(task.id)!,elements,canvases,copiedCanvases,state.offset):duplicateEdgeChange(document,registry,document.elements.get(state.edges[cursor-state.order.length])!,elements,canvases,projected),line=duplicateImportLine(document,registry,change,keys);
try{const original=task?.kind==='canvas'?document.canvases.get(task.id)!:readElement?await readElement(task?.id??state.edges[cursor-state.order.length],signal):document.elements.get(task?.id??state.edges[cursor-state.order.length])!;
if(signal.aborted)throw fault('import_aborted');const change=cursor<state.order.length?duplicateTreeChange(document,registry,task.kind,original,elements,canvases,copiedCanvases,state.offset):duplicateEdgeChange(document,registry,original as ElementRecord,elements,canvases,projected),line=duplicateImportLine(document,registry,change,keys);
if(new TextEncoder().encode(JSON.stringify(line)).byteLength>256*1024)throw fault('record_too_large');
return{cursor,nextCursor:cursor+1,key:line.key,value:line as unknown as Json};}catch(error){return{cursor,nextCursor:cursor+1,key,error:(error as {code?:string}).code??'copy_record_failed'};}
}});
src/lib/canvas-runtime/duplicate.ts
+1 -1
View File
@@ -62,7 +62,7 @@ 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},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>}={}):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();
src/lib/canvas-runtime/engine.ts
+13 -11
View File
@@ -1,4 +1,5 @@
import { encodeCopySnapshot, decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
import { encodeIndexedCopySnapshot, openIndexedCopySnapshot, type IndexedCopySnapshot } from './copy-snapshot-index.ts';
import { decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
import { createDuplicatePlanSource } from './duplicate-source.ts';
import { planDuplicateAsync, planDuplicateResumable, prepareDuplicateStream } from './duplicate.ts';
import { importSourceDigest } from './import-integrity.ts';
@@ -329,15 +330,15 @@ export async function openProjectEngine(options: EngineOptions) {
const admitted=getImport(id),preparation=admitted.preparation;
if(!preparation||admitted.source?.kind!=='duplicate-snapshot')throw fault('invalid_copy_preparation');
const check=()=>{const job=getImport(id);if(controller.signal.aborted||closing||closed||importHolds.size||job.status!=='preparing'||job.sourceFingerprint!==admitted.sourceFingerprint)throw fault('copy_aborted');return job;};
await persist();check();const document=await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal);
await persist();check();const indexed=admitted.source.snapshotVersion===2?await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal):undefined,document=indexed?.document??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal);
if(copyRegistrySignature(document,options.registry)!==preparation.registrySignature)throw fault('copy_interpreter_changed');
if(preparation.version===3){
const state=await prepareDuplicateStream(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner},{signal:controller.signal,checkpoint:async state=>{
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 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);
const stableIds=Object.fromEntries([...state.elementIds.map(([old,id])=>[`element:${old}`,id]),...state.canvasIds.map(([old,id])=>[`canvas:${old}`,id])]),next={...job,generation:job.generation+1,parserVersion:2,status:'running' as const,stableIds,source:{...job.source!,kind:'duplicate-plan' as const,plan:state,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:state.roots.map(old=>stableIds[`element:${old}`])}}};delete next.preparation;delete next.error;
publishImport(freeze(next),job.generation);return{job:getImport(id),bytes,document};
publishImport(freeze(next),job.generation);return{job:getImport(id),bytes,document,indexed};
}
const allocationIds=[...preparation.allocationIds];let cursor=0;
const plan=preparation.version===2?await planDuplicateResumable(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner},{signal:controller.signal,checkpoint:async state=>{
@@ -355,7 +356,7 @@ export async function openProjectEngine(options: EngineOptions) {
const stableIds=Object.fromEntries([...Object.entries(plan.elementIds).map(([old,id])=>[`element:${old}`,id]),...Object.entries(plan.canvasIds).map(([old,id])=>[`canvas:${old}`,id])]);
const next={...job,generation:job.generation+1,status:'running' as const,stableIds,sourceFingerprint:`local-file:${fileId}:${encoded.size}`,source:{...job.source!,kind:'duplicate-ndjson' as const,fileId,byteLength:encoded.size,sha256,integrityVersion:1 as const,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:plan.roots}}};delete next.preparation;delete next.error;
fileWrites.set(fileId,encoded);try{publishImport(freeze(next),job.generation);}catch(error){fileWrites.delete(fileId);throw error;}
return{job:getImport(id),bytes:encoded,document:undefined};
return{job:getImport(id),bytes:encoded,document:undefined,indexed:undefined};
}
function loadImport(id:Id):Promise<void>{
if(importLoads.has(id))return importLoads.get(id)!;
@@ -376,9 +377,10 @@ export async function openProjectEngine(options: EngineOptions) {
if(importHolds.size){deferredImportLoads.add(id);return;}
if(source.sha256&&source.sha256!==sha256)throw fault('import_source_changed');
if(!source.sha256){publishImport(freeze({...verified,generation:verified.generation+1,source:{...source,sha256}}),verified.generation);job=getImport(id);source=job.source!;}else job=verified;
let copyDocument:Document|undefined;
if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;copyDocument=materialized.document;expectedFingerprint=job.sourceFingerprint;}
const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint});
let copyDocument:Document|undefined,indexed:IndexedCopySnapshot|undefined;
if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;copyDocument=materialized.document;indexed=materialized.indexed;expectedFingerprint=job.sourceFingerprint;}
if(source.kind==='duplicate-plan'&&source.snapshotVersion===2){indexed??=await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal);copyDocument=indexed.document;}
const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint});
if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return;
if(importHolds.size){deferredImportLoads.add(id);return;}
const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:()=>persistenceFailed||pending.size>=32?persist():undefined,source:importSource,getJob:()=>getImport(id),createId,
@@ -413,12 +415,12 @@ export async function openProjectEngine(options: EngineOptions) {
signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)controller.abort();
const task=(async()=>{
if(!target||!targets.length)throw fault('empty_copy_selection');
const bytes=await encodeCopySnapshot(captured.document,controller.signal);assertWritable();if(controller.signal.aborted)throw fault('copy_aborted');
const bytes=await encodeIndexedCopySnapshot(captured.document,controller.signal);assertWritable();if(controller.signal.aborted)throw fault('copy_aborted');
const canvas=captured.document.canvases.get(target);if(!canvas)throw fault('canvas_not_loaded',[target]);
const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;
if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||fileWrites.has(fileId)||[...(current.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict');
const entry=canvas.parent?captured.document.elements.get(canvas.parent.entryElementId)!:undefined;
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'duplicate-snapshot',name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:[]},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'preparing' as const,preparation:{version:3 as const,registrySignature:copyRegistrySignature(captured.document,options.registry),selection:targets,worldOffset:offset,allocationIds:[],steps:0}});
const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'duplicate-snapshot',snapshotVersion:2,name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:[]},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'preparing' as const,preparation:{version:4 as const,registrySignature:copyRegistrySignature(captured.document,options.registry),selection:targets,worldOffset:offset,allocationIds:[],steps:0}});
fileWrites.set(fileId,bytes);const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});});
if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;
})();
src/lib/canvas-runtime/imports.ts
+6 -4
View File
@@ -8,12 +8,12 @@ export type ImportSource = { readonly fingerprint: string; readonly parserVersio
export type ImportUnit = { readonly key: string; readonly dependencies: readonly string[]; readonly changes: readonly Change[]; readonly guard?: Json };
export type ImportLocator = { readonly key: string; readonly cursor: number; readonly nextCursor: number; readonly dependencies: readonly string[]; readonly digest: string; readonly byteLength: number; readonly changeCount: number };
export type ImportFailure = { readonly key: string; readonly cursor: number; readonly code: string; readonly dependencies?: readonly string[] };
export type NativeImportSource = { readonly kind: 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
export type NativeImportSource = { readonly kind: 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
export type ImportJob = {
readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
readonly source?: NativeImportSource;
readonly preparation?: {readonly version:1|2|3;readonly planner?:DuplicateCheckpoint;readonly registrySignature:string;readonly selection:readonly Target[];readonly worldOffset:Point;readonly allocationIds:readonly Id[];readonly steps:number};
readonly preparation?: {readonly version:1|2|3|4;readonly planner?:DuplicateCheckpoint;readonly registrySignature:string;readonly selection:readonly Target[];readonly worldOffset:Point;readonly allocationIds:readonly Id[];readonly steps:number};
readonly generation: number; readonly status: 'preparing' | 'paused' | 'running' | 'completed' | 'partial' | 'cancelled'; readonly cursor: number; readonly eof: boolean;
readonly bindings?: Readonly<Record<Id,ImportBinding>>;
readonly stableIds: Readonly<Record<string, Id>>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string;
@@ -29,14 +29,16 @@ export function validateImportJob(job: ImportJob): void {
if (!Number.isSafeInteger(job.generation) || job.generation<0 || !Number.isSafeInteger(job.cursor) || job.cursor<0 || typeof job.eof!=='boolean' || !['preparing','paused','running','completed','partial','cancelled'].includes(job.status)) throw fault('invalid_import_job');
if(job.source&&(!['native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(job.source.kind)||['duplicate-ndjson','duplicate-plan'].includes(job.source.kind)&&(!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0||!Array.isArray(job.source.copy.roots)||!job.source.copy.roots.length||job.source.copy.roots.some(id=>typeof id!=='string'||!id))||job.source.name!==undefined&&(typeof job.source.name!=='string'||job.source.name.length>255)||!job.source.fileId||!job.source.targetCanvasId||!Number.isSafeInteger(job.source.byteLength)||job.source.byteLength<0||job.source.targetEntry&&(!job.source.targetEntry.id||!Number.isSafeInteger(job.source.targetEntry.incarnation)||job.source.targetEntry.incarnation<1)))throw fault('invalid_import_source');
if(job.source){
if(job.source.snapshotVersion!==undefined&&(job.source.snapshotVersion!==2||!['duplicate-snapshot','duplicate-plan'].includes(job.source.kind)))throw fault('copy_snapshot_interpreter_unavailable');
if(job.source.integrityVersion!==undefined&&job.source.integrityVersion!==1)throw fault('import_source_integrity_interpreter_unavailable');
if(job.source.sha256!==undefined&&(typeof job.source.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(job.source.sha256))))throw fault('invalid_import_source');
if(job.source.integrityVersion===1&&!job.source.sha256&&(job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||(job.preparation?.allocationIds.length??0)>0||(job.preparation?.steps??0)>0))throw fault('import_source_integrity_missing');
}
if(job.source?.kind==='duplicate-snapshot'){
const p=job.preparation;
if(!p||![1,2,3].includes(p.version)||typeof p.registrySignature!=='string'||!p.registrySignature||!Array.isArray(p.selection)||!p.selection.length||p.selection.some(t=>!t||typeof t.id!=='string'||!t.id||!Number.isSafeInteger(t.incarnation)||t.incarnation<1)||!p.worldOffset||![p.worldOffset.x,p.worldOffset.y].every(Number.isFinite)||!Array.isArray(p.allocationIds)||p.allocationIds.some(id=>typeof id!=='string'||!id)||new Set(p.allocationIds).size!==p.allocationIds.length||!Number.isSafeInteger(p.steps)||p.steps<0||!['preparing','paused','cancelled'].includes(job.status)||job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0)throw fault('invalid_copy_preparation');
if(p.planner){validateDuplicateCheckpoint(p.planner);if(![2,3].includes(p.version)||p.planner.sourceToken!==job.source.sha256||p.planner.schemaToken!==p.registrySignature||p.planner.projectId!==job.projectId||!equal(p.planner.selection,p.selection)||!equal(p.planner.offset,p.worldOffset)||p.planner.steps!==p.steps||!equal([...p.planner.elementIds,...p.planner.canvasIds].map(pair=>pair[1]),p.allocationIds))throw fault('invalid_copy_preparation');}
if(!p||![1,2,3,4].includes(p.version)||typeof p.registrySignature!=='string'||!p.registrySignature||!Array.isArray(p.selection)||!p.selection.length||p.selection.some(t=>!t||typeof t.id!=='string'||!t.id||!Number.isSafeInteger(t.incarnation)||t.incarnation<1)||!p.worldOffset||![p.worldOffset.x,p.worldOffset.y].every(Number.isFinite)||!Array.isArray(p.allocationIds)||p.allocationIds.some(id=>typeof id!=='string'||!id)||new Set(p.allocationIds).size!==p.allocationIds.length||!Number.isSafeInteger(p.steps)||p.steps<0||!['preparing','paused','cancelled'].includes(job.status)||job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0)throw fault('invalid_copy_preparation');
if((p.version===4)!==(job.source.snapshotVersion===2))throw fault('invalid_copy_preparation');
if(p.planner){validateDuplicateCheckpoint(p.planner);if(![2,3,4].includes(p.version)||p.planner.sourceToken!==job.source.sha256||p.planner.schemaToken!==p.registrySignature||p.planner.projectId!==job.projectId||!equal(p.planner.selection,p.selection)||!equal(p.planner.offset,p.worldOffset)||p.planner.steps!==p.steps||!equal([...p.planner.elementIds,...p.planner.canvasIds].map(pair=>pair[1]),p.allocationIds))throw fault('invalid_copy_preparation');}
else if(p.version!==1&&(p.steps!==0||p.allocationIds.length))throw fault('invalid_copy_preparation');
}else if(job.status==='preparing'||job.preparation!==undefined)throw fault('invalid_copy_preparation');
const keys=new Set<string>(),ids=Object.values(job.stableIds),stableIdSet=new Set(ids);
tests/headless/browser-storage-cases.mjs
+2 -2
View File
@@ -404,7 +404,7 @@ async function duplicatePlannerRecoveryCase(name,assert){
try{
engine=await openProjectEngine(options);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);
await new Promise((resolve,reject)=>{const timer=setTimeout(()=>{release();reject(new Error('no durable DFS checkpoint'));},30000);const release=engine.editor.subscribe((_,snapshot)=>{const planner=snapshot.runtime.imports.get(id)?.preparation?.planner;if(planner?.phase==='dfs'&&planner.work.length){clearTimeout(timer);release();resolve();}});});
await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),saved=JSON.stringify(before.preparation.planner);assert(before.preparation.version===3&&before.preparation.planner.phase==='dfs','did not pause within DFS');assert(before.preparation.planner.sourceToken===before.source.sha256,'checkpoint lost source binding');assert(before.committedSourceKeys.length===0,'preparation published content prematurely');await engine.close();engine=await openProjectEngine(options);
await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),saved=JSON.stringify(before.preparation.planner);assert(before.preparation.version===4&&before.preparation.planner.phase==='dfs','did not pause within DFS');assert(before.preparation.planner.sourceToken===before.source.sha256,'checkpoint lost source binding');assert(before.committedSourceKeys.length===0,'preparation published content prematurely');await engine.close();engine=await openProjectEngine(options);
assert(JSON.stringify(engine.imports.getJobs().get(id).preparation.planner)===saved,'IndexedDB changed the DFS checkpoint');let resumed=false;const release=engine.editor.subscribe((_,snapshot)=>{const planner=snapshot.runtime.imports.get(id)?.preparation?.planner;if(planner){assert(planner.steps>=before.preparation.steps,'resume regressed the cursor');assert(!['selection','canvases','members','roots','seed'].includes(planner.phase),'resume repeated an earlier preparation phase');resumed=true;}});
await engine.imports.resume(id);await engine.settle();release();const job=engine.imports.getJobs().get(id);assert(resumed&&job.status==='completed'&&job.failures.length===0,'DFS recovery did not complete');assert(job.committedSourceKeys.length===502&&Object.keys(job.stableIds).length===502,'DFS recovery lost a source record');assert(document.elements.size===1002&&document.canvases.size===3,'DFS recovery duplicated or lost content');const copiedChild=job.stableIds[`canvas:${child}`];for(let i=0;i<500;i++)assert(document.elements.get(job.stableIds[`element:member-${i}`])?.ownerCanvasId===copiedChild,'DFS sibling lost its copied owner');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
@@ -610,7 +610,7 @@ export async function run() {
await referenceCycleCase(name,assert);
results.push('native frame reference cycles and copied port mappings commit atomically through protocol 2 and real IndexedDB undo/redo/reopening');
await duplicatePlannerRecoveryCase(name,assert);
results.push('preparation v2 persists a DFS stack/cursor in real IndexedDB and resumes 500 siblings without restarting scans or duplicating identities');
results.push('preparation v4 persists a DFS stack/cursor in real IndexedDB and resumes 500 siblings without restarting scans or duplicating identities');
await duplicateRecoveryCase(name,assert);
results.push('deep copy source/mapping survive IndexedDB pause/reopen/resume, retain C to F-prime and shared assets, preserve edits and do not revive undone content');
await nativeImportRecoveryCase(name,assert);
tests/headless/copy-snapshot-index.test.mjs
+35
View File
@@ -0,0 +1,35 @@
import {test} from 'node:test';
import assert from 'node:assert/strict';
import {createDocument,createRegistry,builtinComponents,planDuplicate,validateDocument} from '../../src/lib/canvas-core/index.ts';
import {encodeIndexedCopySnapshot,openIndexedCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot-index.ts';
import {prepareDuplicateStream} from '../../src/lib/canvas-runtime/duplicate.ts';
import {copyRegistrySignature,decodeCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot.ts';
import {createDuplicatePlanSource} from '../../src/lib/canvas-runtime/duplicate-source.ts';
const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal,node=(id,owner='P',componentId='text')=>({id,incarnation:2,structure:'node',ownerCanvasId:owner,componentId,dataVersion:1,data:{title:id,text:'unparsed-payload-秘密',extra:{retained:true}},geometry:{x:10,y:20,width:200,height:100,rotation:0}});
const document=createDocument('project',[{id:'P',parent:null},{id:'S',parent:{canvasId:'P',entryElementId:'F'}}],[node('C'),{...node('F','P','frame'),data:{...node('F').data,ports:[{id:'port',nodeId:'A'}],presentation:{unitScale:2,contentScale:.75,other:'not-needed-in-index'}},geometry:{...node('F').geometry,rotation:35}},node('A','S','image'),{id:'edge',incarnation:1,structure:'connection',ownerCanvasId:'P',componentId:'connection',dataVersion:1,data:{retained:true},source:{elementId:'C'},target:{elementId:'A'},path:{type:'bezier-chain',knots:[{id:'k',mode:'corner',position:{x:70,y:80},in:{x:1,y:2},out:{x:3,y:4}}]}}],new Map([['deleted',4]]));
const lines=async bytes=>(await bytes.text()).trimEnd().split('\n');
test('indexed snapshots read topology before payload, preserve raw records on demand and reproduce exact mapped output',async()=>{
const bytes=await encodeIndexedCopySnapshot(document,signal()),snapshot=await openIndexedCopySnapshot(bytes,'project',registry,signal());assert.equal(JSON.stringify([...snapshot.document.elements]).includes('unparsed-payload'),false);assert.equal(snapshot.document.elements.get('F').data.ports,undefined);assert.deepEqual(snapshot.document.elements.get('F').data.presentation,{unitScale:2,contentScale:.75});assert.deepEqual([...snapshot.document.tombstones],[...document.tombstones]);assert.throws(()=>validateDocument(snapshot.document,registry));
const loaded=[];let next=0;const state=await prepareDuplicateStream(snapshot.document,registry,[{id:'F',incarnation:2}],{createId:()=>`copy-${++next}`},{sourceToken:'a'.repeat(64),schemaToken:copyRegistrySignature(snapshot.document,registry),topologyOnly:true});assert.equal(loaded.length,0);const source=createDuplicatePlanSource(snapshot.document,registry,state,'indexed',async(id,abort)=>{loaded.push(id);return snapshot.readElement(id,abort);});let reference=0;const expected=planDuplicate(document,registry,[{id:'F',incarnation:2}],{createId:()=>`copy-${++reference}`});
const last=await source.read(expected.changes.length-1,signal());assert.deepEqual(last.value.change,expected.changes.at(-1));assert.deepEqual(loaded,['edge'],'edge geometry decoded unrelated node payload');
for(let i=0;i<expected.changes.length;i++)assert.deepEqual((await source.read(i,signal())).value.change,expected.changes[i]);for(const[id,record]of document.elements)assert.deepEqual(await snapshot.readElement(id,signal()),record);await assert.rejects(decodeCopySnapshot(bytes,'project',registry,signal()),/invalid_copy_snapshot/);
});
test('index decoding defers corrupt or invalid component content until the corresponding record is requested',async()=>{
const rows=await lines(await encodeIndexedCopySnapshot(document,signal())),last=rows.length-1;rows[last]='!'+rows[last].slice(1);const corrupt=new Blob([rows.join('\n')+'\n']),snapshot=await openIndexedCopySnapshot(corrupt,'project',registry,signal());assert.equal((await snapshot.readElement('A',signal())).id,'A');await assert.rejects(snapshot.readElement('edge',signal()),/invalid_copy_snapshot/);
const clean=await lines(await encodeIndexedCopySnapshot(document,signal())),i=clean.findIndex(row=>row.startsWith('{"kind":"element","record":{"id":"A"'));clean[i]=clean[i].replace('"id":"A"','"id":"Z"');const mismatch=await openIndexedCopySnapshot(new Blob([clean.join('\n')+'\n']),'project',registry,signal());await assert.rejects(mismatch.readElement('A',signal()),/invalid_copy_snapshot/);
});
test('indexed source rejects locator gaps, geometry corruption, truncation, foreign versions and aborted reads',async()=>{
const bytes=await encodeIndexedCopySnapshot(document,signal()),rows=await lines(bytes),index=rows.findIndex(row=>JSON.parse(row).kind==='element-index');
for(const change of [row=>({...row,offset:1}),row=>({...row,byteLength:0}),row=>({...row,record:{...row.record,geometry:{...row.record.geometry,width:0}}})]){const bad=[...rows];bad[index]=JSON.stringify(change(JSON.parse(bad[index])));await assert.rejects(openIndexedCopySnapshot(new Blob([bad.join('\n')+'\n']),'project',registry,signal()));}
await assert.rejects(openIndexedCopySnapshot(bytes.slice(0,bytes.size-1),'project',registry,signal()));await assert.rejects(openIndexedCopySnapshot(new Blob([await bytes.text(),'x']),'project',registry,signal()));await assert.rejects(openIndexedCopySnapshot(bytes,'foreign',registry,signal()));const controller=new AbortController();controller.abort();await assert.rejects(encodeIndexedCopySnapshot(document,controller.signal),/copy_aborted/);await assert.rejects(openIndexedCopySnapshot(bytes,'project',registry,controller.signal),/copy_aborted/);const snapshot=await openIndexedCopySnapshot(bytes,'project',registry,signal());await assert.rejects(snapshot.readElement('A',controller.signal),/copy_aborted/);
});
test('bounded raw windows preserve sequential parser batching and evict old bytes without caching decoded content',async()=>{
const source=createDocument('project',[{id:'P',parent:null}],Array.from({length:120},(_,i)=>({...node(`record-${i}`),data:{title:`record-${i}`,text:'x'.repeat(3500)}}))),bytes=await encodeIndexedCopySnapshot(source,signal()),reads=[];
class ObservedBlob extends Blob{slice(...args){reads.push(args);return super.slice(...args);}}
const snapshot=await openIndexedCopySnapshot(new ObservedBlob([bytes]),'project',registry,signal());reads.length=0;
for(let i=0;i<120;i++)assert.equal((await snapshot.readElement(`record-${i}`,signal())).data.title,`record-${i}`);assert.ok(reads.length>=5&&reads.length<120/4,`unexpected per-record Blob reads: ${reads.length}`);const count=reads.length;assert.equal((await snapshot.readElement('record-0',signal())).data.text.length,3500);assert.equal(reads.length,count+1,'first window did not leave the bounded cache');
});
tests/headless/copy-snapshot.test.mjs
+1 -1
View File
@@ -15,6 +15,6 @@ test('snapshot decoding rejects malformed, future, duplicate, invalid and trunca
});
test('preparation checkpoints require a matching interpreter, valid selection, unique allocations and no published progress',()=>{
const base={...createImportJob({id:'job',projectId:'project',sourceFingerprint:'snapshot',parserVersion:1,source:{kind:'duplicate-snapshot',sha256:'a'.repeat(64),fileId:'file',byteLength:10,targetCanvasId:'P',copy:{sourceRevision:0,roots:[]}}}),status:'preparing',preparation:{version:1,registrySignature:'registry-v1',selection:[{id:'A',incarnation:2}],worldOffset:{x:32,y:32},allocationIds:['new-A'],steps:512}};validateImportJob(base);
for(const patch of [{preparation:undefined},{preparation:{...base.preparation,version:4}},{preparation:{...base.preparation,allocationIds:['same','same']}},{preparation:{...base.preparation,selection:[]}},{cursor:1},{committedSourceKeys:['element:A']},{status:'running'}])assert.throws(()=>validateImportJob({...base,...patch}));
for(const patch of [{preparation:undefined},{preparation:{...base.preparation,version:5}},{preparation:{...base.preparation,allocationIds:['same','same']}},{preparation:{...base.preparation,selection:[]}},{cursor:1},{committedSourceKeys:['element:A']},{status:'running'},{source:{...base.source,snapshotVersion:3}},{source:{...base.source,snapshotVersion:2}}])assert.throws(()=>validateImportJob({...base,...patch}));
assert.throws(()=>createImportRunner({getJob:()=>base}),/import_interpreter_unavailable/);
});
tests/headless/imports-engine.test.mjs
+18 -4
View File
@@ -1,3 +1,5 @@
import {encodeCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot.ts';
import {importSourceDigest} from '../../src/lib/canvas-runtime/import-integrity.ts';
import {test} from 'node:test';
import assert from 'node:assert/strict';
import {createDocument,createRegistry,builtinComponents,applyChanges} from '../../src/lib/canvas-core/index.ts';
@@ -130,7 +132,7 @@ test('read-only policy rejects copy/import admission without storing a source or
test('preparing copy resumes saved planning cursors and allocated identities without repeating earlier phases',async(t)=>{
const f=fixture(),save=f.storage.commit;let durableCuts=0;f.storage.commit=async(...args)=>{durableCuts++;return save(...args);};let engine=await f.open();try{
await seedCopy(engine);assert.equal(engine.editor.transaction(tx=>{for(let i=0;i<1000;i++)tx.dispatch({type:'element.create',element:copyNode(`extra-${i}`,'child')});}).status,'applied');await engine.settle();
const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>(snapshot.runtime.imports.get(id).preparation?.allocationIds.length??0)>0);await engine.imports.pause(id);await engine.settle();const checkpoint=engine.imports.getJobs().get(id);assert.equal(checkpoint.status,'paused');assert.equal(checkpoint.source.kind,'duplicate-snapshot');assert.ok(checkpoint.preparation.allocationIds.length>0);assert.ok(checkpoint.preparation.steps>0);assert.equal(checkpoint.preparation.version,3);assert.ok(checkpoint.preparation.planner);assert.equal(checkpoint.preparation.planner.sourceToken,checkpoint.source.sha256);assert.equal(checkpoint.committedSourceKeys.length,0);decodeRecovery(f.image(),registry);
const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>(snapshot.runtime.imports.get(id).preparation?.allocationIds.length??0)>0);await engine.imports.pause(id);await engine.settle();const checkpoint=engine.imports.getJobs().get(id);assert.equal(checkpoint.status,'paused');assert.equal(checkpoint.source.kind,'duplicate-snapshot');assert.ok(checkpoint.preparation.allocationIds.length>0);assert.ok(checkpoint.preparation.steps>0);assert.equal(checkpoint.preparation.version,4);assert.ok(checkpoint.preparation.planner);assert.equal(checkpoint.preparation.planner.sourceToken,checkpoint.source.sha256);assert.equal(checkpoint.committedSourceKeys.length,0);decodeRecovery(f.image(),registry);
engine.editor.dispatch({type:'element.data',target:{id:'F',incarnation:1},patch:{title:'new source title'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');assert.deepEqual(engine.imports.getJobs().get(id).preparation.allocationIds,checkpoint.preparation.allocationIds);assert.deepEqual(engine.imports.getJobs().get(id).preparation.planner,checkpoint.preparation.planner);const observed=[],stop=engine.editor.subscribe((_,snapshot)=>{const state=snapshot.runtime.imports.get(id)?.preparation?.planner;if(state)observed.push({phase:state.phase,steps:state.steps});});await engine.imports.resume(id);await engine.settle();stop();assert.ok(observed.some(state=>state.steps>checkpoint.preparation.steps));assert.ok(observed.every(state=>state.steps>=checkpoint.preparation.steps&&!['selection','canvases','members','roots','seed','dfs','edges'].includes(state.phase)));await engine.settle();
const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);t.diagnostic(JSON.stringify({imported:job.committedSourceKeys.length,durableCuts,requests:f.requests.length}));
}finally{await engine.close();}
@@ -195,8 +197,10 @@ test('legacy unfinished sources remain readable but cannot acquire a new content
});
async function legacyCopySource(f,job,document){const signal=new AbortController().signal,bytes=await encodeCopySnapshot(document,signal);delete job.source.snapshotVersion;job.source.byteLength=bytes.size;job.source.sha256=await importSourceDigest(bytes,signal);job.sourceFingerprint=`local-file:${job.source.fileId}:${bytes.size}`;f.files.set(job.source.fileId,bytes);}
test('legacy preparation version 1 still replays its reserved identities instead of allocating replacements',async()=>{
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=1;job.preparation.steps=1;job.preparation.allocationIds=['legacy-frame-id'];delete job.preparation.planner;engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:F'],'legacy-frame-id');assert.equal(f.errors.length,0);}finally{await engine.close();}
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const sourceDocument=engine.getSnapshot().document,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=1;job.preparation.steps=1;job.preparation.allocationIds=['legacy-frame-id'];delete job.preparation.planner;await legacyCopySource(f,job,sourceDocument);engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:F'],'legacy-frame-id');assert.equal(f.errors.length,0);}finally{await engine.close();}
});
@@ -206,8 +210,8 @@ test('copy converts and publishes a prefix before the output tail, preserves it
}finally{await engine.close();}
});
test('legacy preparation version 2 retains its complete-plan source path after reopening',async()=>{
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=2;engine=await f.open();await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.source.kind,'duplicate-ndjson');assert.equal(done.parserVersion,1);assert.equal(f.errors.length,0);}finally{await engine.close();}
test('legacy preparation versions 2 and 3 retain their source interpreters after reopening',async()=>{
for(const version of [2,3]){const f=fixture();let engine=await f.open();try{await seedCopy(engine);const sourceDocument=engine.getSnapshot().document,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=version;await legacyCopySource(f,job,sourceDocument);engine=await f.open();await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.source.kind,version===2?'duplicate-ndjson':'duplicate-plan');assert.equal(done.parserVersion,version===2?1:2);assert.equal(f.errors.length,0);}finally{await engine.close();}}
});
test('late planned-source decoding respects cancellation and canonical recovery holds',async()=>{
@@ -221,3 +225,13 @@ test('late planned-source decoding respects cancellation and canonical recovery
}finally{gate.resolve();await engine.close();}
}
});
test('indexed copy displays a prefix while a later raw payload read is still blocked',async()=>{
const source=createDocument('project',[{id:'canvas',parent:null},{id:'child',parent:{canvasId:'canvas',entryElementId:'F'}}],[copyNode('F','canvas','frame'),...Array.from({length:100},(_,i)=>({...copyNode(`lazy-${i}`,'child'),data:{title:`lazy-${i}`,text:'x'.repeat(3000)}}))]),f=fixture(source);let engine=await f.open();const entered=deferred(),gate=deferred();try{
const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id);await engine.close();const bytes=f.files.get(before.source.fileId),rows=(await bytes.text()).trimEnd().split('\n');let payload=0;
for(const text of rows){const row=JSON.parse(text);payload+=new TextEncoder().encode(text+'\n').byteLength;if(row.kind==='payload')break;}
class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}
class DelayedPayload extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return start>=payload+64*1024?new DelayedChunk([part]):part;}}
f.files.set(before.source.fileId,new DelayedPayload([bytes]));engine=await f.open();await engine.imports.resume(id);await entered.promise;const cut=engine.getSnapshot(),job=cut.runtime.imports.get(id);assert.equal(job.source.snapshotVersion,2);assert.equal(job.status,'running');assert.ok(job.committedSourceKeys.length>0&&job.committedSourceKeys.length<102);assert.ok(cut.document.elements.size>source.elements.size&&cut.document.elements.size<202,'raw tail read gated the entire copy');gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,202);assert.equal(f.errors.length,0);
}finally{gate.resolve();await engine.close();}
});