Stream duplicate record conversion into progressive insertion
This commit is contained in:
@@ -40,6 +40,7 @@ export function createDuplicatePlanner(document:Document,registry:Registry,selec
|
||||
function edge(index:number){add(duplicateEdgeChange(document,registry,document.elements.get(state.edges[index])!,elementIds,canvasIds,output));}
|
||||
function allocate(id:Id,kind:'elements'|'canvases'){const map=suppliedMapping?.[kind],value=map&&Object.hasOwn(map,id)?map[id]:idFactory();if(typeof value!=='string'||!value||used.has(value))throw fault('duplicate_identity',[value]);used.add(value);return value;}
|
||||
return Object.freeze({
|
||||
ready():boolean{return !sourceChecks&&state.phase==='emit'&&state.emitted===0;},
|
||||
checkpoint():DuplicateCheckpoint{return freeze(clone(state));},
|
||||
step():boolean{
|
||||
if(sourceChecks){if(sourceChecks.next().done)sourceChecks=undefined;return false;}
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
import type { Document, ElementRecord, CanvasRecord, Id, Registry, Json } from '../canvas-core/types.ts';
|
||||
import type { DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
|
||||
import { createDuplicatePlanner } from '../canvas-core/duplicate-checkpoint.ts';
|
||||
import { duplicateTreeChange, duplicateEdgeChange } from '../canvas-core/duplicate.ts';
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
import { copyRegistrySignature } from './copy-snapshot.ts';
|
||||
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{
|
||||
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});
|
||||
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.
|
||||
function projection<T extends ElementRecord|CanvasRecord>(base:ReadonlyMap<Id,T>,reverse:ReadonlyMap<Id,Id>,kind:'element'|'canvas'):ReadonlyMap<Id,T>{
|
||||
const cache=new Map<Id,T>();const get=(id:Id):T|undefined=>{const old=reverse.get(id);if(!old)return base.get(id);const found=cache.get(id);if(found)return found;const original=base.get(old);if(!original)throw fault('invalid_copy_stream_checkpoint');const record=duplicateTreeChange(document,registry,kind,original,elements,canvases,copiedCanvases,state.offset).after as T;cache.set(id,record);if(cache.size>32)cache.delete(cache.keys().next().value!);return record;};
|
||||
const entries=function*():Generator<[Id,T],undefined>{yield* base;for(const id of reverse.keys())yield[id,get(id)!];return undefined;};
|
||||
const view:ReadonlyMap<Id,T>={get,has:id=>base.has(id)||reverse.has(id),size:base.size+reverse.size,entries,keys:function*():Generator<Id,undefined>{for(const[id]of entries())yield id;return undefined;},values:function*():Generator<T,undefined>{for(const[,value]of entries())yield value;return undefined;},[Symbol.iterator]:entries,forEach(fn,thisArg){for(const[id,value]of entries())fn.call(thisArg,value,id,view);}};return Object.freeze(view);
|
||||
}
|
||||
const projected:Document={...document,elements:projection(document.elements,reverseElements,'element'),canvases:projection(document.canvases,reverseCanvases,'canvas')},length=state.order.length+state.edges.length;
|
||||
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);
|
||||
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'};}
|
||||
}});
|
||||
}
|
||||
@@ -59,3 +59,13 @@ export async function planDuplicateResumable(document:Document,registry:Registry
|
||||
if(scheduling.checkpoint)await scheduling.checkpoint(planner.checkpoint());check();await yieldTask();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/** 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>{
|
||||
const budget=scheduling.frameBudgetMs??6,limit=scheduling.maxStepsPerSlice??512,now=scheduling.now??(()=>performance.now()),yieldTask=scheduling.yieldTask??(()=>new Promise<void>(resolve=>setTimeout(resolve,0)));
|
||||
if(!Number.isFinite(budget)||budget<=0||!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_copy_budget');
|
||||
const check=()=>{if(scheduling.signal?.aborted)throw fault('copy_aborted');};check();
|
||||
const planner=createDuplicatePlanner(document,registry,selection,options,binding);
|
||||
while(true){const start=now();let count=0;do{check();if(planner.ready())return planner.checkpoint();if(planner.step())throw fault('invalid_copy_stream_checkpoint');count++;}while(count<limit&&now()-start<budget);if(scheduling.checkpoint)await scheduling.checkpoint(planner.checkpoint());check();await yieldTask();}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { encodeCopySnapshot, decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
|
||||
import { planDuplicateAsync, planDuplicateResumable } from './duplicate.ts';
|
||||
import { createDuplicatePlanSource } from './duplicate-source.ts';
|
||||
import { planDuplicateAsync, planDuplicateResumable, prepareDuplicateStream } from './duplicate.ts';
|
||||
import { importSourceDigest } from './import-integrity.ts';
|
||||
import { encodeDuplicateSource, prepareDuplicateImport, assertDuplicateExternal } from './import-duplicate.ts';
|
||||
import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts';
|
||||
@@ -330,6 +331,14 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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);
|
||||
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=>{
|
||||
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};
|
||||
}
|
||||
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=>{
|
||||
const job=check(),previous=job.preparation!;if(previous.planner&&state.steps<=previous.steps)return;
|
||||
@@ -346,16 +355,16 @@ 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};
|
||||
return{job:getImport(id),bytes:encoded,document:undefined};
|
||||
}
|
||||
function loadImport(id:Id):Promise<void>{
|
||||
if(importLoads.has(id))return importLoads.get(id)!;
|
||||
if(importRunners.has(id)){void importRunners.get(id)!.start();return Promise.resolve();}
|
||||
deferredImportLoads.delete(id);
|
||||
let expectedFingerprint=getImport(id).sourceFingerprint;
|
||||
const preparationController=new AbortController();if(getImport(id).source?.kind==='duplicate-snapshot')preparationControllers.set(id,preparationController);
|
||||
const preparationController=new AbortController();if(['duplicate-snapshot','duplicate-plan'].includes(getImport(id).source?.kind??''))preparationControllers.set(id,preparationController);
|
||||
const task=(async()=>{
|
||||
let job=getImport(id),source=job.source;if(!source||!['native-ndjson','duplicate-ndjson','duplicate-snapshot'].includes(source.kind)||job.parserVersion!==1)throw fault('import_interpreter_unavailable');
|
||||
let job=getImport(id),source=job.source;if(!source||!['native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(source.kind)||job.parserVersion!==(source.kind==='duplicate-plan'?2:1))throw fault('import_interpreter_unavailable');
|
||||
let bytes=fileWrites.get(source.fileId)??await options.storage.readFile(leader,source.fileId);
|
||||
if(closing||closed)return;
|
||||
const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status))return;
|
||||
@@ -367,14 +376,18 @@ 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;
|
||||
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;expectedFingerprint=job.sourceFingerprint;}
|
||||
const runner=createImportRunner({optimisticLocal:true,awaitDurabilityBudget:()=>persistenceFailed||pending.size>=32?persist():undefined,source:createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}),getJob:()=>getImport(id),createId,
|
||||
publish:publishImport,prepare:(line,job,reserve)=>job.source?.kind==='duplicate-ndjson'?prepareDuplicateImport(line,job,reserve,options.registry):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry),persist,onError:report,
|
||||
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});
|
||||
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,
|
||||
publish:publishImport,prepare:(line,job,reserve)=>['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')?prepareDuplicateImport(line,job,reserve,options.registry):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry),persist,onError:report,
|
||||
commit(units,next,expected){
|
||||
assertWritable();if(importing)throw fault('reentrant_dispatch');const current=getImport(id);if(current.generation!==expected)throw fault('stale_import_job');
|
||||
try{assertNativeImportTarget(current,editor.getSnapshot().document);for(const unit of units){assertNativeImportDependencies(current,editor.getSnapshot().document,unit.dependencies);if(current.source?.kind==='duplicate-ndjson')assertDuplicateExternal({cursor:0,nextCursor:1,key:unit.key,value:unit.guard},current,editor.getSnapshot().document,options.registry);}}catch(error){return reject(error);}
|
||||
try{assertNativeImportTarget(current,editor.getSnapshot().document);for(const unit of units){assertNativeImportDependencies(current,editor.getSnapshot().document,unit.dependencies);if(['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??''))assertDuplicateExternal({cursor:0,nextCursor:1,key:unit.key,value:unit.guard},current,editor.getSnapshot().document,options.registry);}}catch(error){return reject(error);}
|
||||
importing={job:next,expectedGeneration:expected};
|
||||
try{return editor.transaction(tx=>{for(const unit of units)for(const change of unit.changes){if(change.kind==='element'&&change.before===null&&change.after)tx.dispatch({type:'element.create',element:change.after});else if(change.kind==='canvas'&&change.before===null&&change.after)tx.dispatch({type:'canvas.create',canvas:change.after});else throw fault('invalid_native_import_unit');}},{label:current.source?.kind==='duplicate-ndjson'?'Duplicate selection':'Import canvas content',groupId:`import:${id}`});}finally{importing=undefined;}
|
||||
try{return editor.transaction(tx=>{for(const unit of units)for(const change of unit.changes){if(change.kind==='element'&&change.before===null&&change.after)tx.dispatch({type:'element.create',element:change.after});else if(change.kind==='canvas'&&change.before===null&&change.after)tx.dispatch({type:'canvas.create',canvas:change.after});else throw fault('invalid_native_import_unit');}},{label:['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??'')?'Duplicate selection':'Import canvas content',groupId:`import:${id}`});}finally{importing=undefined;}
|
||||
},
|
||||
});
|
||||
importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold());
|
||||
@@ -405,7 +418,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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:2 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',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}});
|
||||
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;
|
||||
})();
|
||||
|
||||
@@ -17,37 +17,42 @@ export async function encodeDuplicateSource(document: Document, registry: Regist
|
||||
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 dependencies=new Set<string>(),elements=new Map<Id,ExternalElement>(),canvases=new Map<Id,ExternalCanvas>();
|
||||
const canvas=(id:Id)=>{
|
||||
const key=keys.get(id);if(key){dependencies.add(key);return;}
|
||||
let next:Id|undefined=id;
|
||||
while(next&&!canvases.has(next)){
|
||||
const original=document.canvases.get(next);if(!original)throw fault('copy_external_changed',[next]);
|
||||
const entry=original.parent?document.elements.get(original.parent.entryElementId):undefined;
|
||||
canvases.set(next,{id:next,parent:clone(original.parent),...(entry?{entryIncarnation:entry.incarnation}:{})});
|
||||
if(entry)elements.set(entry.id,{id:entry.id,incarnation:entry.incarnation,ownerCanvasId:entry.ownerCanvasId});
|
||||
next=original.parent?.canvasId;
|
||||
}
|
||||
};
|
||||
const element=(id:Id)=>{const key=keys.get(id);if(key){dependencies.add(key);return;}const original=document.elements.get(id);if(!original)throw fault('copy_external_changed',[id]);elements.set(id,{id,incarnation:original.incarnation,ownerCanvasId:original.ownerCanvasId});canvas(original.ownerCanvasId);};
|
||||
if(change.kind==='canvas'){if(!change.after.parent)throw fault('invalid_duplicate_plan');canvas(change.after.parent.canvasId);element(change.after.parent.entryElementId);}
|
||||
else {
|
||||
const record=change.after;canvas(record.ownerCanvasId);
|
||||
if(record.structure==='connection'){element(record.source.elementId);element(record.target.elementId);}
|
||||
const schema=registry.components.get(record.componentId)?.copy;if(!schema)throw fault('copy_schema_unavailable');
|
||||
for(const ref of schema.references)for(const id of copyReferenceIds(record.data,ref)){if(ref.target==='element')element(id);else canvas(id);}
|
||||
}
|
||||
const key=keys.get(change.id);if(!key)throw fault('invalid_duplicate_plan');dependencies.delete(key);
|
||||
const line:CopyLine={key,change,dependencies:[...dependencies],externalElements:[...elements.values()],externalCanvases:[...canvases.values()]};
|
||||
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));
|
||||
}
|
||||
if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'});
|
||||
}
|
||||
|
||||
export function duplicateImportLine(document:Document,registry:Registry,change:Exclude<Change,{kind:'field'}>,keys:ReadonlyMap<Id,string>):CopyLine {
|
||||
if(!change.after||change.before!==null)throw fault('invalid_duplicate_plan');
|
||||
const dependencies=new Set<string>(),elements=new Map<Id,ExternalElement>(),canvases=new Map<Id,ExternalCanvas>();
|
||||
const canvas=(id:Id)=>{
|
||||
const key=keys.get(id);if(key){dependencies.add(key);return;}
|
||||
let next:Id|undefined=id;
|
||||
while(next&&!canvases.has(next)){
|
||||
const original=document.canvases.get(next);if(!original)throw fault('copy_external_changed',[next]);
|
||||
const entry=original.parent?document.elements.get(original.parent.entryElementId):undefined;
|
||||
canvases.set(next,{id:next,parent:clone(original.parent),...(entry?{entryIncarnation:entry.incarnation}:{})});
|
||||
if(entry)elements.set(entry.id,{id:entry.id,incarnation:entry.incarnation,ownerCanvasId:entry.ownerCanvasId});
|
||||
next=original.parent?.canvasId;
|
||||
}
|
||||
};
|
||||
const element=(id:Id)=>{const key=keys.get(id);if(key){dependencies.add(key);return;}const original=document.elements.get(id);if(!original)throw fault('copy_external_changed',[id]);elements.set(id,{id,incarnation:original.incarnation,ownerCanvasId:original.ownerCanvasId});canvas(original.ownerCanvasId);};
|
||||
if(change.kind==='canvas'){if(!change.after.parent)throw fault('invalid_duplicate_plan');canvas(change.after.parent.canvasId);element(change.after.parent.entryElementId);}
|
||||
else {
|
||||
const record=change.after;canvas(record.ownerCanvasId);
|
||||
if(record.structure==='connection'){element(record.source.elementId);element(record.target.elementId);}
|
||||
const schema=registry.components.get(record.componentId)?.copy;if(!schema)throw fault('copy_schema_unavailable');
|
||||
for(const ref of schema.references)for(const id of copyReferenceIds(record.data,ref)){if(ref.target==='element')element(id);else canvas(id);}
|
||||
}
|
||||
const key=keys.get(change.id);if(!key)throw fault('invalid_duplicate_plan');dependencies.delete(key);
|
||||
return {key,change,dependencies:[...dependencies],externalElements:[...elements.values()],externalCanvases:[...canvases.values()]};
|
||||
}
|
||||
|
||||
function copyLine(line:ImportLine,job:ImportJob):CopyLine {
|
||||
const value=line.value as unknown as CopyLine;
|
||||
if(job.source?.kind!=='duplicate-ndjson'||!value||!validJson(value)||typeof value.key!=='string'||!Array.isArray(value.dependencies)||!Array.isArray(value.externalElements)||!Array.isArray(value.externalCanvases)||!value.change||value.change.kind!=='element'&&value.change.kind!=='canvas'||value.change.before!==null||!value.change.after||!Object.hasOwn(job.stableIds,value.key)||job.stableIds[value.key]!==value.change.id||value.change.after.id!==value.change.id||!value.key.startsWith(`${value.change.kind}:`))throw fault('invalid_duplicate_source');
|
||||
if(!['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')||!value||!validJson(value)||typeof value.key!=='string'||!Array.isArray(value.dependencies)||!Array.isArray(value.externalElements)||!Array.isArray(value.externalCanvases)||!value.change||value.change.kind!=='element'&&value.change.kind!=='canvas'||value.change.before!==null||!value.change.after||!Object.hasOwn(job.stableIds,value.key)||job.stableIds[value.key]!==value.change.id||value.change.after.id!==value.change.id||!value.key.startsWith(`${value.change.kind}:`))throw fault('invalid_duplicate_source');
|
||||
if(value.change.kind==='element'&&value.change.after?.incarnation!==1)throw fault('invalid_duplicate_source');
|
||||
return value;
|
||||
}
|
||||
|
||||
@@ -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'; 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 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;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;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;
|
||||
@@ -27,7 +27,7 @@ export function validateImportJob(job: ImportJob): void {
|
||||
if (job.version !== 1 && job.version !== 2 && job.version !== 3) throw fault('import_interpreter_unavailable');
|
||||
createImportJob(job);
|
||||
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'].includes(job.source.kind)||job.source.kind==='duplicate-ndjson'&&(!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&&(!['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.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');
|
||||
@@ -35,13 +35,18 @@ export function validateImportJob(job: ImportJob): void {
|
||||
}
|
||||
if(job.source?.kind==='duplicate-snapshot'){
|
||||
const p=job.preparation;
|
||||
if(!p||![1,2].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(p.version!==2||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===2&&(p.steps!==0||p.allocationIds.length))throw fault('invalid_copy_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');}
|
||||
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);
|
||||
if (stableIdSet.size!==ids.length || ids.some(id=>typeof id!=='string'||!id) || Object.keys(job.stableIds).some(key=>!key)) throw fault('invalid_import_job');
|
||||
if(job.source?.kind==='duplicate-ndjson'&&(!job.source.copy||new Set(job.source.copy.roots).size!==job.source.copy.roots.length||job.source.copy.roots.some(id=>!stableIdSet.has(id))))throw fault('invalid_import_source');
|
||||
if(job.source&&['duplicate-ndjson','duplicate-plan'].includes(job.source.kind)&&(!job.source.copy||new Set(job.source.copy.roots).size!==job.source.copy.roots.length||job.source.copy.roots.some(id=>!stableIdSet.has(id))))throw fault('invalid_import_source');
|
||||
if(job.source?.kind==='duplicate-plan'){
|
||||
const plan=job.source.plan;if(!plan)throw fault('invalid_copy_stream_checkpoint');validateDuplicateCheckpoint(plan);
|
||||
const ids=Object.fromEntries([...plan.elementIds.map(([id,mapped])=>[`element:${id}`,mapped]),...plan.canvasIds.map(([id,mapped])=>[`canvas:${id}`,mapped])]);
|
||||
if(job.parserVersion!==2||plan.projectId!==job.projectId||plan.sourceToken!==job.source.sha256||plan.phase!=='emit'||plan.emitted||plan.emittedEdges||!equal(ids,job.stableIds)||!equal(plan.roots.map(id=>ids[`element:${id}`]),job.source.copy!.roots)||job.cursor>plan.order.length+plan.edges.length)throw fault('invalid_copy_stream_checkpoint');
|
||||
}else if(job.source?.plan!==undefined)throw fault('invalid_copy_stream_checkpoint');
|
||||
for(const[id,binding]of Object.entries(job.bindings??{}))if(!id||!binding||!stableIdSet.has(id)||!['element','canvas'].includes(binding.kind)||binding.kind==='element'&&(!Number.isSafeInteger(binding.incarnation)||binding.incarnation<1)||binding.kind==='canvas'&&binding.entryElementId!==undefined&&typeof binding.entryElementId!=='string')throw fault('invalid_import_job');
|
||||
for(const key of job.committedSourceKeys){if(!key||keys.has(key))throw fault('invalid_import_job');keys.add(key);}
|
||||
if(job.source)for(const key of job.committedSourceKeys){const id=job.stableIds[key],binding=job.bindings?.[id];if(!Object.hasOwn(job.stableIds,key)||!binding||!key.startsWith(`${binding.kind}:`)||binding.kind==='canvas'&&(!binding.entryElementId||job.bindings?.[binding.entryElementId]?.kind!=='element'))throw fault('invalid_import_job');}
|
||||
|
||||
@@ -388,7 +388,7 @@ async function duplicateRecoveryCase(name,assert){
|
||||
await engine.imports.pause(id);await engine.settle();assert(engine.imports.getJobs().get(id).source.kind==='duplicate-snapshot','copy did not expose its preparing source');await engine.close();engine=await openProjectEngine(options);assert(engine.imports.getJobs().get(id).status==='paused'&&engine.imports.getJobs().get(id).preparation.selection.length===2,'preparing copy source/selection did not survive IndexedDB');await engine.imports.resume(id);
|
||||
await new Promise(resolve=>{if(engine.imports.getJobs().get(id).committedSourceKeys.length){resolve();return;}const release=engine.editor.subscribe((_,snapshot)=>{if(snapshot.runtime.imports.get(id).committedSourceKeys.length===1){release();resolve();}});});await engine.imports.pause(id);await engine.settle();
|
||||
const before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert(before.status==='paused','copy pause lost');engine.editor.dispatch({type:'element.data',target:{id:copied,incarnation:1},patch:{title:'edited copy'}});await engine.settle();await engine.close();engine=await openProjectEngine(options);
|
||||
const restored=engine.imports.getJobs().get(id);assert(restored.source.kind==='duplicate-ndjson'&&JSON.stringify(restored.stableIds)===JSON.stringify(before.stableIds),'copy source or mapping changed on reopen');await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id),image=document.elements.get(job.stableIds['element:A']),incoming=document.elements.get(job.stableIds['element:incoming']);
|
||||
const restored=engine.imports.getJobs().get(id);assert(restored.source.kind==='duplicate-plan'&&JSON.stringify(restored.stableIds)===JSON.stringify(before.stableIds),'copy source or mapping changed on reopen');await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id),image=document.elements.get(job.stableIds['element:A']),incoming=document.elements.get(job.stableIds['element:incoming']);
|
||||
assert(job.status==='completed'&&job.failures.length===0,'copy did not complete after reopening');assert(document.elements.get(copied).data.title==='edited copy','copy replay overwrote an edit');assert(image.data.assetId==='shared'&&!Object.hasOwn(image.data,'runId'),'copy reissued image execution');assert(incoming.source.elementId==='C'&&incoming.target.elementId===copied&&incoming.ownerCanvasId===root,'copy lost upper incoming source');assert(document.elements.has('incoming')&&document.canvases.has(job.stableIds[`canvas:${child}`]),'copy changed originals or lost its child canvas');
|
||||
assert(engine.editor.history.undo().status==='applied','copy undo failed');await engine.settle();const count=document.elements.size;await engine.close();engine=await openProjectEngine(options);await engine.settle();assert(document.elements.size===count&&!document.elements.has(job.stableIds['element:A']),'terminal copy revived undone records');
|
||||
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(database);}
|
||||
@@ -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===2&&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===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);
|
||||
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);}
|
||||
|
||||
@@ -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:3}},{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:4}},{preparation:{...base.preparation,allocationIds:['same','same']}},{preparation:{...base.preparation,selection:[]}},{cursor:1},{committedSourceKeys:['element:A']},{status:'running'}])assert.throws(()=>validateImportJob({...base,...patch}));
|
||||
assert.throws(()=>createImportRunner({getJob:()=>base}),/import_interpreter_unavailable/);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createDocument,createRegistry,builtinComponents,planDuplicate} from '../../src/lib/canvas-core/index.ts';
|
||||
import {prepareDuplicateStream} from '../../src/lib/canvas-runtime/duplicate.ts';
|
||||
import {createDuplicatePlanSource} from '../../src/lib/canvas-runtime/duplicate-source.ts';
|
||||
import {copyRegistrySignature} from '../../src/lib/canvas-runtime/copy-snapshot.ts';
|
||||
import {createImportJob,validateImportJob} from '../../src/lib/canvas-runtime/imports.ts';
|
||||
const registry=createRegistry(builtinComponents()),node=(id,owner='P',componentId='text')=>({id,incarnation:1,ownerCanvasId:owner,componentId,dataVersion:1,structure:'node',data:{title:id,text:''},geometry:{x:10,y:20,width:100,height:100,rotation:0}}),signal=new AbortController().signal;
|
||||
const document=createDocument('project',[{id:'P',parent:null},{id:'S',parent:{canvasId:'P',entryElementId:'F'}}],[node('C'),{...node('F','P','frame'),data:{title:'F',text:'',ports:[{id:'p',nodeId:'A'}],presentation:{contentScale:2}},geometry:{...node('F').geometry,rotation:30}},node('A','S'),{id:'incoming',incarnation:1,ownerCanvasId:'P',componentId:'connection',dataVersion:1,structure:'connection',source:{elementId:'C'},target:{elementId:'A'},data:{},path:{type:'bezier-chain',knots:[{id:'k',mode:'corner',position:{x:40,y:60},in:{x:0,y:0},out:{x:1,y:1}}]}}]);
|
||||
async function prepare(source=document){let next=0;return prepareDuplicateStream(source,registry,[{id:'F',incarnation:1}],{createId:()=>`copy-${++next}`},{sourceToken:'a'.repeat(64),schemaToken:copyRegistrySignature(source,registry)},{maxStepsPerSlice:2});}
|
||||
|
||||
test('lazy copy source seeks mapped records with exact synchronous geometry and reference parity',async()=>{
|
||||
const state=await prepare(),source=createDuplicatePlanSource(document,registry,state,'snapshot');let next=0;const plan=planDuplicate(document,registry,[{id:'F',incarnation:1}],{createId:()=>`copy-${++next}`});assert.equal(state.phase,'emit');assert.equal(state.emitted,0);assert.equal(state.emittedEdges,0);
|
||||
for(let i=plan.changes.length-1;i>=0;i--){const line=await source.read(i,signal);assert.deepEqual(line.value.change,plan.changes[i]);assert.equal(line.nextCursor,i+1);assert.deepEqual(await createDuplicatePlanSource(document,registry,JSON.parse(JSON.stringify(state)),'snapshot').read(i,signal),line);}
|
||||
assert.equal(await source.read(plan.changes.length,signal),null);const aborted=new AbortController();aborted.abort();await assert.rejects(source.read(0,aborted.signal),/import_aborted/);await assert.rejects(source.read(-1,signal),/invalid_import_cursor/);
|
||||
});
|
||||
|
||||
test('mapping preparation completes before converting unsupported tail records; source reports per-record failure',async()=>{
|
||||
const sourceDocument=createDocument('project',[{id:'P',parent:null},{id:'S',parent:{canvasId:'P',entryElementId:'F'}}],[node('F','P','frame'),node('good','S'),node('opaque','S','unknown-widget')]),state=await prepare(sourceDocument),source=createDuplicatePlanSource(sourceDocument,registry,state,'snapshot');
|
||||
const lines=[];for(let i=0;;i++){const line=await source.read(i,signal);if(!line)break;lines.push(line);}assert.ok(lines[0].value);assert.ok(lines.some(line=>line.value?.change.after?.data?.title==='good'));assert.equal(lines.at(-1).error,'copy_schema_unavailable');assert.equal(lines.at(-1).key,'element:opaque');assert.throws(()=>planDuplicate(sourceDocument,registry,[{id:'F',incarnation:1}],{createId:()=>crypto.randomUUID()}),/copy_schema_unavailable/);
|
||||
});
|
||||
|
||||
test('persisted record cursor source binds snapshot SHA, mapping and interpreter without a second output Blob',async()=>{
|
||||
const state=await prepare(),stableIds=Object.fromEntries([...state.elementIds.map(([id,mapped])=>[`element:${id}`,mapped]),...state.canvasIds.map(([id,mapped])=>[`canvas:${id}`,mapped])]),job={...createImportJob({id:'job',projectId:'project',sourceFingerprint:'snapshot',parserVersion:2,source:{kind:'duplicate-plan',fileId:'file',byteLength:100,sha256:state.sourceToken,targetCanvasId:'P',plan:state,copy:{sourceRevision:1,roots:state.roots.map(id=>stableIds[`element:${id}`])}}}),status:'running',stableIds};validateImportJob(job);
|
||||
for(const patch of [{parserVersion:1},{cursor:100},{stableIds:{}},{source:{...job.source,sha256:'b'.repeat(64)}},{source:{...job.source,plan:{...state,emitted:1}}}])assert.throws(()=>validateImportJob({...job,...patch}));
|
||||
});
|
||||
@@ -7,12 +7,12 @@ const registry=createRegistry(builtinComponents()),original=createDocument('proj
|
||||
const node=id=>({kind:'element',key:id,record:{id,incarnation:8,structure:'node',ownerCanvasId:'$target',componentId:'text',dataVersion:1,data:{title:id,text:'',runId:'must not run'},geometry:{x:0,y:0,width:100,height:100,rotation:0}}});
|
||||
const file=records=>new Blob([records.map(value=>typeof value==='string'?value:JSON.stringify(value)).join('\n')]);
|
||||
const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};};
|
||||
function fixture(){
|
||||
let image=null,leader,next=0,server=original;const versions=new Map([['canvas',1]]);const files=new Map(),requests=[],receipts=new Map(),errors=[];
|
||||
function fixture(initial=original){
|
||||
let image=null,leader,next=0,server=initial;const versions=new Map([...initial.canvases.keys()].map(id=>[id,1]));const files=new Map(),requests=[],receipts=new Map(),errors=[];
|
||||
const f={files,requests,errors,gate:null,broken:false,
|
||||
storage:{async claim(){assert.equal(leader,undefined);const value={scope,fence:1,active:true,async release(){leader=undefined;value.active=false;}};leader=value;return value;},async recover(){return image&&structuredClone(image);},async commit(owner,batch){assert.equal(owner,leader);if(f.broken)throw Error('quota');for(const part of batch.files??[])files.set(part.id,part.bytes);image=structuredClone(batch.image);return{durableRootRevision:batch.image.rootRevision};},async readFile(owner,id){assert.equal(owner,leader);return files.get(id);},close(){}},
|
||||
port:{compile(intent,requestId,snapshot){return{payload:{requestId,intent},canvasIds:intent.canvasIds,expectedVersions:Object.fromEntries(intent.canvasIds.map(id=>[id,snapshot.runtime.versions.get(id)??0]))};},async submit(request){requests.push(request);if(f.gate)await f.gate.promise;const{intent}=JSON.parse(request.body);if(!receipts.has(request.id)){server=applyChanges(server,intent.changes);for(const canvasId of intent.canvasIds)versions.set(canvasId,(versions.get(canvasId)??0)+1);receipts.set(request.id,{document:server,versions:new Map(versions)});}return{outcome:'applied',canonical:{requestId:request.id}};},async query(request){return receipts.has(request.id)?{outcome:'applied',canonical:{requestId:request.id}}:{outcome:'unresolved'};},canonical(request){return receipts.get(request.id);},async read(){return{document:server,versions:new Map(versions)};}},
|
||||
image:()=>image,server:()=>server,open:(extra={})=>openProjectEngine({scope,registry,document:original,versions:new Map([['canvas',1]]),storage:f.storage,persistence:f.port,createId:()=>`identity-${++next}`,onError:error=>errors.push(error),...extra})};return f;
|
||||
image:()=>image,server:()=>server,open:(extra={})=>openProjectEngine({scope,registry,document:initial,versions:new Map([...initial.canvases.keys()].map(id=>[id,1])),storage:f.storage,persistence:f.port,createId:()=>`identity-${++next}`,onError:error=>errors.push(error),...extra})};return f;
|
||||
}
|
||||
function until(engine,predicate){if(predicate(engine.getSnapshot()))return Promise.resolve();return new Promise((resolve,reject)=>{const timer=setTimeout(()=>{unsubscribe();reject(Error('import condition timed out'));},5000),unsubscribe=engine.editor.subscribe((_,snapshot)=>{if(predicate(snapshot)){clearTimeout(timer);unsubscribe();resolve();}});});}
|
||||
|
||||
@@ -89,7 +89,7 @@ test('deep copy source/mapping recover and preserve upper incoming source, share
|
||||
try{
|
||||
await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1},{id:'A',incarnation:1}]);
|
||||
await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
|
||||
const before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert.equal(before.source.kind,'duplicate-ndjson');assert.equal(before.source.copy.roots.length,1);assert.equal(before.status,'paused');
|
||||
const before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert.equal(before.source.kind,'duplicate-plan');assert.equal(before.source.copy.roots.length,1);assert.equal(before.status,'paused');
|
||||
engine.editor.dispatch({type:'element.data',target:{id:copied,incarnation:1},patch:{title:'edited copy'}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();
|
||||
const job=engine.imports.getJobs().get(id),document=engine.getSnapshot().document,ids=job.stableIds;assert.equal(job.status,'completed');assert.deepEqual(ids,before.stableIds);assert.equal(document.elements.get(copied).data.title,'edited copy');
|
||||
assert.equal(document.canvases.get(ids['canvas:child']).parent.entryElementId,copied);assert.equal(document.elements.get(ids['element:A']).ownerCanvasId,ids['canvas:child']);assert.equal(document.elements.get(ids['element:A']).data.assetId,'shared');assert.equal(document.elements.get(ids['element:A']).data.runId,undefined);assert.equal(document.elements.get(ids['element:A']).data.taskId,undefined);
|
||||
@@ -130,7 +130,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,2);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,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);
|
||||
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();}
|
||||
@@ -198,3 +198,26 @@ test('legacy unfinished sources remain readable but cannot acquire a new content
|
||||
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();}
|
||||
});
|
||||
|
||||
|
||||
test('copy converts and publishes a prefix before the output tail, preserves it on failure and stores only the original source',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(`stream-${i}`,'child')),{...copyNode('opaque','child'),componentId:'unknown-widget'}]),f=fixture(source);let engine=await f.open();
|
||||
try{let prefix;const release=engine.editor.subscribe((_,snapshot)=>{for(const job of snapshot.runtime.imports.values())if(!prefix&&job.committedSourceKeys.length)prefix={cursor:job.cursor,total:job.source.plan.order.length+job.source.plan.edges.length,status:job.status};});const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.settle();release();const job=engine.imports.getJobs().get(id);assert.ok(prefix.cursor<prefix.total,'entire mapped output was converted before the first publication');assert.equal(prefix.status,'running');assert.equal(job.source.kind,'duplicate-plan');assert.equal(job.parserVersion,2);assert.equal(job.source.plan.emitted,0);assert.equal(job.status,'partial');assert.equal(job.committedSourceKeys.length,102);assert.ok(job.failures.some(failure=>failure.key==='element:opaque'&&failure.code==='copy_schema_unavailable'));assert.equal(engine.getSnapshot().document.elements.size,203);assert.equal(f.files.size,1,'copy stored a second complete mapped file');assert.equal(f.server().elements.size,203);assert.equal(f.errors.length,0);await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'partial');assert.equal(engine.getSnapshot().document.elements.size,203);assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,102);
|
||||
}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('late planned-source decoding respects cancellation and canonical recovery holds',async()=>{
|
||||
for(const cancel of [true,false]){
|
||||
const f=fixture();let engine=await f.open();const entered=deferred(),gate=deferred();try{
|
||||
await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),size=engine.getSnapshot().document.elements.size;await engine.close();
|
||||
class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}
|
||||
class DelayedSource extends Blob{slice(...args){return new DelayedChunk([super.slice(...args)]);}}
|
||||
f.files.set(before.source.fileId,new DelayedSource([f.files.get(before.source.fileId)]));engine=await f.open();const resuming=engine.imports.resume(id);await entered.promise;
|
||||
if(cancel){await engine.imports.cancel(id);gate.resolve();await resuming;await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,size);}else{const preparing=engine.prepareRecovery();gate.resolve();await resuming;const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');}assert.equal(f.errors.length,0);
|
||||
}finally{gate.resolve();await engine.close();}
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user