feat(canvas): recover admitted copy preparation from immutable snapshots
This commit is contained in:
@@ -39,6 +39,7 @@ function CanvasView({ engine, board, viewId }: { engine: ProjectEngine; board: B
|
||||
const liveView = sessions.views.get(viewId), lastView = useRef(liveView);
|
||||
if (liveView) lastView.current = liveView;
|
||||
const view = liveView ?? lastView.current!, flow = useReactFlow<ProjectionNode>();
|
||||
const preparingCopy = [...(root.runtime.imports?.values() ?? [])].some(job => job.status === 'preparing' && job.source?.targetCanvasId === view.canvasId);
|
||||
const [copying, setCopying] = useState(false);
|
||||
const [showAI, setShowAI] = useState(false), [showRecovery, setShowRecovery] = useState(false);
|
||||
const [error, setError] = useState(""), [sources, setSources] = useState(new Map<string, string>());
|
||||
@@ -195,7 +196,7 @@ function CanvasView({ engine, board, viewId }: { engine: ProjectEngine; board: B
|
||||
</EdgeEditing.Provider></NodeEditing.Provider></ImagePreviews.Provider>
|
||||
<div className={styles.topRight}><div className={`${toolbar.surface} ${styles.toolbar}`} role="toolbar" aria-label="图谱视图控制">
|
||||
<IconButton icon={Undo2} label="撤销" disabled={!root.history.undo.length} onClick={() => report(engine.editor.history.undo())} /><IconButton icon={Redo2} label="重做" disabled={!root.history.redo.length} onClick={() => report(engine.editor.history.redo())} />
|
||||
<IconButton icon={Copy} label={copying ? "正在准备复制" : "深复制选中内容"} disabled={copying || !view.selection.size} onClick={() => {
|
||||
<IconButton icon={Copy} label={copying || preparingCopy ? "正在准备复制" : "深复制选中内容"} disabled={copying || preparingCopy || !view.selection.size} onClick={() => {
|
||||
const captured = current.current, selection = [...captured.view.selection].map(id => captured.root.document.elements.get(id)).filter((item): item is ElementRecord => !!item).map(({id, incarnation}) => ({id, incarnation}));
|
||||
setCopying(true); setError("");
|
||||
void engine.imports.duplicate(selection, {viewCanvasId: captured.view.canvasId}).catch(error => { if (current.current.engine === engine && current.current.view.id === viewId) setError(errorText(error)); }).finally(() => { if (current.current.engine === engine && current.current.view.id === viewId) setCopying(false); });
|
||||
|
||||
@@ -7,7 +7,7 @@ import { IconButton } from "@/components/ui/icon-button";
|
||||
import { errorText } from "@/lib/api";
|
||||
import styles from "./canvas.module.css";
|
||||
|
||||
const reason = (code: string) => ({ invalid_json: "记录格式错误", invalid_json_value: "记录包含不支持的数据", invalid_utf8: "文件编码不是 UTF-8", record_too_large: "单条记录过大", unresolved_dependency: "引用的记录未加入", invalid_native_import_record: "不是画布记录", import_component_unavailable: "元素类型暂不支持", invalid_geometry: "元素尺寸不合法", invalid_endpoint: "连线端点已不存在", duplicate_import_source_key: "记录标识重复", import_source_unavailable: "导入源文件不可用", import_source_changed: "导入源或解析规则已变化", import_target_missing: "目标分区已不存在", import_target_changed: "目标分区已改变", import_dependency_changed: "引用对象已删除或恢复为新版本", dependency_capacity: "等待引用过多，请调整文件顺序", copy_external_changed: "原图引用已删除、移动或恢复为新版本", invalid_duplicate_source: "复制源数据不可用", invalid_import_source: "文件过大或格式不支持" } as Record<string,string>)[code] ?? "这条记录暂时无法加入";
|
||||
const reason = (code: string) => ({ invalid_json: "记录格式错误", invalid_json_value: "记录包含不支持的数据", invalid_utf8: "文件编码不是 UTF-8", record_too_large: "单条记录过大", unresolved_dependency: "引用的记录未加入", invalid_native_import_record: "不是画布记录", import_component_unavailable: "元素类型暂不支持", invalid_geometry: "元素尺寸不合法", invalid_endpoint: "连线端点已不存在", duplicate_import_source_key: "记录标识重复", import_source_unavailable: "导入源文件不可用", import_source_changed: "导入源或解析规则已变化", import_target_missing: "目标分区已不存在", import_target_changed: "目标分区已改变", import_dependency_changed: "引用对象已删除或恢复为新版本", dependency_capacity: "等待引用过多，请调整文件顺序", copy_external_changed: "原图引用已删除、移动或恢复为新版本", invalid_duplicate_source: "复制源数据不可用", invalid_copy_snapshot: "复制快照不可用", invalid_copy_preparation: "复制准备记录不可用", copy_interpreter_changed: "复制规则已变化，需要重新复制", invalid_import_source: "文件过大或格式不支持" } as Record<string,string>)[code] ?? "这条记录暂时无法加入";
|
||||
|
||||
export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; canvasId: string }) {
|
||||
const subscribe = useCallback((notify: () => void) => engine.editor.subscribe(notify), [engine]);
|
||||
@@ -22,15 +22,15 @@ export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; c
|
||||
const file = event.target.files?.[0]; event.target.value = ""; if (!file) return;
|
||||
try { engine.imports.startNdjson(file, canvasId); setOpen(true); setError(""); } catch (error) { setError(errorText(error)); }
|
||||
}} />
|
||||
{jobs.length > 0 && <details open={open} onToggle={event => setOpen(event.currentTarget.open)}><summary>导入与复制任务 · {jobs.reduce((count, job) => count + job.committedSourceKeys.length, 0)} 项已加入</summary>
|
||||
{jobs.length > 0 && <details open={open} onToggle={event => setOpen(event.currentTarget.open)}><summary>导入与复制任务 · {jobs.some(job => job.status === "preparing") && "正在准备复制 · "}{jobs.reduce((count, job) => count + job.committedSourceKeys.length, 0)} 项已加入</summary>
|
||||
{jobs.map((job, index) => <section className={styles.importJob} key={job.id} aria-label={`导入任务 ${index + 1}`}>
|
||||
<strong>{job.source?.name || `导入任务 ${index + 1}`}</strong>
|
||||
<p role="status">{job.status === "running" ? "正在加入" : job.status === "paused" ? "已暂停" : job.status === "cancelled" ? "已停止" : job.status === "partial" ? "部分加入" : "已加入画布"} · {job.committedSourceKeys.length} 项{job.pending.length > 0 && `，${job.pending.length} 项等待`}</p>
|
||||
<p role="status">{job.status === "preparing" ? "正在准备复制" : job.status === "running" ? "正在加入" : job.status === "paused" ? "已暂停" : job.status === "cancelled" ? "已停止" : job.status === "partial" ? "部分加入" : "已加入画布"} · {job.committedSourceKeys.length} 项{job.pending.length > 0 && `，${job.pending.length} 项等待`}</p>
|
||||
{job.error && <p>{reason(job.error)}</p>}
|
||||
<div className={styles.importActions}>
|
||||
{job.status === "running" && <button onClick={() => control(engine.imports.pause(job.id))}>暂停导入</button>}
|
||||
{["running","preparing"].includes(job.status) && <button onClick={() => control(engine.imports.pause(job.id))}>暂停导入</button>}
|
||||
{job.status === "paused" && <button onClick={() => control(engine.imports.resume(job.id))}>继续导入</button>}
|
||||
{["running","paused"].includes(job.status) && <button onClick={() => control(engine.imports.cancel(job.id))}>停止导入</button>}
|
||||
{["running","preparing","paused"].includes(job.status) && <button onClick={() => control(engine.imports.cancel(job.id))}>停止导入</button>}
|
||||
</div>
|
||||
{job.status === "cancelled" && <small>已加入的内容保留并继续同步。</small>}
|
||||
{job.failures.length > 0 && <details><summary>{job.failures.length} 条未加入记录</summary><ul>{job.failures.slice(0, 10).map((failure, item) => <li key={`${failure.key}:${item}`}>{failure.key.startsWith("byte:") ? `文件位置 ${failure.cursor}` : failure.key}：{reason(failure.code)}</li>)}</ul></details>}
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import type { CanvasRecord, Document, ElementRecord, Registry } from '../canvas-core/types.ts';
|
||||
import { createDocument, validateDocumentSteps } from '../canvas-core/document.ts';
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
import { createNdjsonImportSource } from './imports.ts';
|
||||
|
||||
const task=()=>new Promise<void>(resolve=>setTimeout(resolve,0));
|
||||
/** Capture immutable source bytes before a preparation job is admitted. No mutable live getters. */
|
||||
export async function encodeCopySnapshot(document:Document,signal:AbortSignal):Promise<Blob>{
|
||||
const parts:BlobPart[]=[];let bytes=0,count=0;
|
||||
const append=(value:unknown)=>{if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');bytes+=size;if(bytes>64*1024*1024)throw fault('invalid_import_source');parts.push(text);};
|
||||
append({kind:'header',version:1,projectId:document.projectId});
|
||||
for(const[kind,records]of [['canvas',document.canvases.values()],['element',document.elements.values()],['tombstone',document.tombstones.entries()]] as const)for(const record of records){append({kind,record});if(++count%32===0)await task();}
|
||||
if(signal.aborted)throw fault('copy_aborted');return new Blob(parts,{type:'application/x-ndjson'});
|
||||
}
|
||||
export async function decodeCopySnapshot(bytes:Blob,projectId:string,registry:Registry,signal:AbortSignal):Promise<Document>{
|
||||
const source=createNdjsonImportSource({bytes,fingerprint:'captured-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map<string,number>();let cursor=0,header=false,count=0;
|
||||
while(true){
|
||||
if(signal.aborted)throw fault('copy_aborted');const line=await source.read(cursor,signal);if(!line)break;cursor=line.nextCursor;
|
||||
const value=line.value as unknown as {kind:string;version?:number;projectId?:string;record:never};if(line.error||line.skip||!value)throw fault('invalid_copy_snapshot');
|
||||
if(!header){if(value.kind!=='header'||value.version!==1||value.projectId!==projectId)throw fault('invalid_copy_snapshot');header=true;continue;}
|
||||
if(value.kind==='canvas')canvases.push(value.record);else if(value.kind==='element')elements.push(value.record);else if(value.kind==='tombstone'){const record=value.record as [string,number];if(!Array.isArray(record)||record.length!==2||tombstones.has(record[0]))throw fault('invalid_copy_snapshot');tombstones.set(record[0],record[1]);}else throw fault('invalid_copy_snapshot');
|
||||
if(++count%32===0)await task();
|
||||
}
|
||||
if(!header||signal.aborted)throw fault(signal.aborted?'copy_aborted':'invalid_copy_snapshot');
|
||||
const document=createDocument(projectId,canvases,elements,tombstones);let checks=0;
|
||||
for(const _ of validateDocumentSteps(document,registry)){if(signal.aborted)throw fault('copy_aborted');if(++checks%512===0)await task();}
|
||||
return document;
|
||||
}
|
||||
|
||||
/** Copy declarations are a versioned input; never silently reinterpret a persisted preparation. */
|
||||
export function copyRegistrySignature(document:Document,registry:Registry):string{
|
||||
const ids=new Set([...document.elements.values()].map(element=>element.componentId));
|
||||
return JSON.stringify([...ids].sort().map(id=>{const definition=registry.components.get(id);return definition?{id,dataVersion:definition.dataVersion,structure:definition.structure,copy:definition.copy??null}: {id,missing:true};}));
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import { clone, fault, freeze } from '../canvas-core/values.ts';
|
||||
|
||||
export type DuplicateScheduling = {
|
||||
readonly signal?: AbortSignal;
|
||||
readonly checkpoint?: (steps:number) => Promise<void>;
|
||||
readonly frameBudgetMs?: number;
|
||||
readonly maxStepsPerSlice?: number;
|
||||
readonly now?: () => number;
|
||||
@@ -24,6 +25,7 @@ export async function planDuplicateAsync(document: Document, registry: Registry,
|
||||
// Capture caller-owned small inputs before the first await. Document/registry are immutable core snapshots.
|
||||
const targets = freeze(clone(selection)), settings: DuplicateOptions = Object.freeze({ createId: options.createId, ...(options.worldOffset ? {worldOffset: freeze(clone(options.worldOffset))} : {}), ...(options.mapping ? {mapping: freeze(clone(options.mapping))} : {}) });
|
||||
const steps = planDuplicateSteps(document, registry, targets, settings);
|
||||
let completedSteps=0;
|
||||
try {
|
||||
while (true) {
|
||||
check();
|
||||
@@ -33,9 +35,10 @@ export async function planDuplicateAsync(document: Document, registry: Registry,
|
||||
const step = steps.next();
|
||||
check();
|
||||
if (step.done) return step.value;
|
||||
count++;
|
||||
count++;completedSteps++;
|
||||
} while (count < limit && now() - start < budget);
|
||||
await yieldTask();
|
||||
if(scheduling.checkpoint)await scheduling.checkpoint(completedSteps);
|
||||
check();await yieldTask();
|
||||
}
|
||||
} finally {
|
||||
// Release snapshot references after cancellation or host scheduling failures.
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { encodeCopySnapshot, decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
|
||||
import { planDuplicateAsync } from './duplicate.ts';
|
||||
import { encodeDuplicateSource, prepareDuplicateImport, assertDuplicateExternal } from './import-duplicate.ts';
|
||||
import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts';
|
||||
@@ -7,7 +8,7 @@ import type { Document, Id, Registry, Result, RunInput, RunApplyInput, RunCancel
|
||||
import { createEditor, type EditorOptions, type Transaction } from '../canvas-core/editor.ts';
|
||||
import { createDocument } from '../canvas-core/document.ts';
|
||||
import { createSessions } from '../canvas-core/gestures.ts';
|
||||
import { equal, fault, freeze, readMap, reject } from '../canvas-core/values.ts';
|
||||
import { clone, equal, fault, freeze, readMap, reject } from '../canvas-core/values.ts';
|
||||
import type { CanonicalProposal, EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts';
|
||||
import { createRunTracker, validateRunInput, runContainsTarget, type RunPort, type RunRecord, type RunPage } from './runs.ts';
|
||||
import { createNetworkGate } from './network.ts';
|
||||
@@ -57,6 +58,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
let applying: RunApplyInput | undefined;
|
||||
let importing: { job: ImportJob; expectedGeneration: number } | undefined;
|
||||
const preparationControllers=new Map<Id,AbortController>();
|
||||
const copyPreparations=new Map<AbortController,Promise<Id>>();
|
||||
const importRunners=new Map<Id,ReturnType<typeof createImportRunner>>(),importLoads=new Map<Id,Promise<void>>(),importHolds=new Map<symbol,Map<Id,()=>void>>();
|
||||
let recovery: { controller: AbortController; release(): void; cancel():Promise<void> } | undefined;
|
||||
@@ -318,17 +320,40 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
|
||||
const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job);return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);
|
||||
}
|
||||
function holdImports(){const token=Symbol(),releases=new Map<Id,()=>void>();importHolds.set(token,releases);for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(job.status==='running'&&!importRunners.has(job.id))void loadImport(job.id);};}
|
||||
function holdImports(){const token=Symbol(),releases=new Map<Id,()=>void>();importHolds.set(token,releases);for(const controller of preparationControllers.values())controller.abort();for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(['running','preparing'].includes(job.status)&&!importRunners.has(job.id))void loadImport(job.id);};}
|
||||
async function materializeCopy(id:Id,bytes:Blob,controller:AbortController){
|
||||
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);
|
||||
if(copyRegistrySignature(document,options.registry)!==preparation.registrySignature)throw fault('copy_interpreter_changed');
|
||||
const allocationIds=[...preparation.allocationIds];let cursor=0;
|
||||
const plan=await planDuplicateAsync(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId:()=>{const index=cursor++;if(index<allocationIds.length)return allocationIds[index];const id=createId();allocationIds.push(id);return id;}},{signal:controller.signal,checkpoint:async steps=>{
|
||||
const job=check(),previous=job.preparation!;
|
||||
if(steps<=previous.steps&&allocationIds.length===previous.allocationIds.length)return;
|
||||
publishImport(freeze({...job,generation:job.generation+1,preparation:{...previous,steps:Math.max(previous.steps,steps),allocationIds:[...allocationIds]}}),job.generation);await persist();check();
|
||||
}});
|
||||
check();if(cursor!==allocationIds.length)throw fault('copy_interpreter_changed');
|
||||
const encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),job=check();assertWritable();assertNativeImportTarget(job,editor.getSnapshot().document);
|
||||
const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
|
||||
const fileId=createId(),runtime=editor.getSnapshot().runtime;if(!fileId||fileWrites.has(fileId)||runtime.resources.has(fileId)||[...(runtime.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict');
|
||||
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,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};
|
||||
}
|
||||
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();}
|
||||
const expectedFingerprint=getImport(id).sourceFingerprint;
|
||||
let expectedFingerprint=getImport(id).sourceFingerprint;
|
||||
const preparationController=new AbortController();if(getImport(id).source?.kind==='duplicate-snapshot')preparationControllers.set(id,preparationController);
|
||||
const task=(async()=>{
|
||||
const job=getImport(id),source=job.source;if(!source||!['native-ndjson','duplicate-ndjson'].includes(source.kind)||job.parserVersion!==1)throw fault('import_interpreter_unavailable');
|
||||
const bytes=fileWrites.get(source.fileId)??await options.storage.readFile(leader,source.fileId);
|
||||
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 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;
|
||||
if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed');
|
||||
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,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):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry),persist,onError:report,
|
||||
commit(units,next,expected){
|
||||
@@ -340,7 +365,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
});
|
||||
importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold());
|
||||
if(getImport(id).status==='running')void runner.start();
|
||||
})().catch(error=>{if(!closing&&!closed){try{const job=getImport(id);if(job.sourceFingerprint!==expectedFingerprint||['completed','partial','cancelled'].includes(job.status))return;publishImport(freeze({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{importLoads.delete(id);});
|
||||
})().catch(error=>{if(preparationController.signal.aborted)return;if(!closing&&!closed){try{const job=getImport(id);if(job.sourceFingerprint!==expectedFingerprint||['completed','partial','cancelled','paused'].includes(job.status))return;publishImport(freeze({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{preparationControllers.delete(id);importLoads.delete(id);});
|
||||
importLoads.set(id,task);return task;
|
||||
}
|
||||
const imports=Object.freeze({
|
||||
@@ -356,30 +381,28 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
},
|
||||
async duplicate(selection:readonly Target[],settings:{readonly worldOffset?:Point;readonly signal?:AbortSignal;readonly viewCanvasId?:Id}={}):Promise<Id>{
|
||||
assertWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const captured=editor.getSnapshot(),target=settings.viewCanvasId??(selection[0]?captured.document.elements.get(selection[0].id)?.ownerCanvasId:undefined),signal=settings.signal,controller=new AbortController(),abort=()=>controller.abort();
|
||||
const targets=freeze(clone(selection)),offset=freeze(clone(settings.worldOffset??{x:32,y:32}));
|
||||
for(const selected of targets){const element=captured.document.elements.get(selected.id);if(!element||element.incarnation!==selected.incarnation)throw fault('stale_incarnation',[selected.id]);}if(![offset.x,offset.y].every(Number.isFinite))throw fault('invalid_copy_offset');
|
||||
signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)controller.abort();
|
||||
const task=(async()=>{
|
||||
const plan=await planDuplicateAsync(captured.document,options.registry,selection,{createId,worldOffset:settings.worldOffset},{signal:controller.signal});
|
||||
const bytes=await encodeDuplicateSource(captured.document,options.registry,plan,controller.signal);
|
||||
assertWritable();if(controller.signal.aborted)throw fault('copy_aborted');const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);
|
||||
// Task belongs to the initiating view; copied records retain their mapped document owners.
|
||||
if(!target)throw fault('empty_copy_selection');
|
||||
const canvas=editor.getSnapshot().document.canvases.get(target);if(!canvas)throw fault('copy_external_changed',[target]);
|
||||
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 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?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined;
|
||||
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 job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{kind:'duplicate-ndjson',name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:plan.roots},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),stableIds,status:'running' as const});
|
||||
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:1 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;
|
||||
})();
|
||||
copyPreparations.set(controller,task);void task.finally(()=>{copyPreparations.delete(controller);signal?.removeEventListener('abort',abort);}).catch(()=>{});return task;
|
||||
},
|
||||
async pause(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(job.status==='running'){publishImport(freeze({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}},
|
||||
async resume(id:Id){assertWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(job.status==='running'){await loadImport(id);return;}await importRunners.get(id)?.settle();await persist();await scheduler.resume();assertWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:'running' as const};delete next.error;publishImport(freeze(next),current.generation);await loadImport(id);},
|
||||
async cancel(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','paused'].includes(job.status)){const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(freeze(next),job.generation);await persist();}}},
|
||||
async pause(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(['running','preparing'].includes(job.status)){preparationControllers.get(id)?.abort();publishImport(freeze({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}},
|
||||
async resume(id:Id){assertWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(['running','preparing'].includes(job.status)){await loadImport(id);return;}await importLoads.get(id);await importRunners.get(id)?.settle();await persist();await scheduler.resume();assertWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:current.source?.kind==='duplicate-snapshot'?'preparing' as const:'running' as const};delete next.error;publishImport(freeze(next),current.generation);await loadImport(id);},
|
||||
async cancel(id:Id){assertWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','preparing','paused'].includes(job.status)){preparationControllers.get(id)?.abort();const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(freeze(next),job.generation);await persist();}}},
|
||||
});
|
||||
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
|
||||
for(const job of initial.imports?.values()??[])if(job.status==='running')void loadImport(job.id);
|
||||
for(const job of initial.imports?.values()??[])if(['running','preparing'].includes(job.status))void loadImport(job.id);
|
||||
return Object.freeze({
|
||||
editor, sessions, imports,
|
||||
/** Read/preview under a scheduling hold. No request is rewritten or declared resolved here. */
|
||||
@@ -546,7 +569,7 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
async settle() { await Promise.resolve();await Promise.allSettled([...copyPreparations.values()]);while(importLoads.size)await Promise.all([...importLoads.values()]);await Promise.all([...importRunners.values()].map(runner=>runner.settle())); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
|
||||
async close() {
|
||||
if (closed) return;
|
||||
closing = true;for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
|
||||
closing = true;for(const controller of preparationControllers.values())controller.abort();for(const controller of copyPreparations.keys())controller.abort();await Promise.allSettled([...copyPreparations.values()]); sessions.dispose(); scheduler.stop(); runTracker.stop();
|
||||
await Promise.all([...importRunners.values()].map(runner=>runner.close()));while(importLoads.size)await Promise.all([...importLoads.values()]);
|
||||
recovery?.controller.abort();recovery?.release();
|
||||
for (const release of [...displays]) release();
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Change, Id, Json, Result } from '../canvas-core/types.ts';
|
||||
import type { Change, Id, Json, Result, Target, Point } from '../canvas-core/types.ts';
|
||||
import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
|
||||
|
||||
export type ImportLine = { readonly cursor: number; readonly nextCursor: number; readonly key: string; readonly value?: Json; readonly error?: string; readonly skip?: boolean };
|
||||
@@ -6,12 +6,13 @@ 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'; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; 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'; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; 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; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
|
||||
readonly source?: NativeImportSource;
|
||||
readonly generation: number; readonly status: 'paused' | 'running' | 'completed' | 'partial' | 'cancelled'; readonly cursor: number; readonly eof: boolean;
|
||||
readonly preparation?: {readonly version:1;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;
|
||||
};
|
||||
@@ -23,8 +24,12 @@ export function validateImportJob(job: ImportJob): void {
|
||||
if(!job||typeof job!=='object'||!validJson(job)||!job.stableIds||typeof job.stableIds!=='object'||Array.isArray(job.stableIds)||![job.pending,job.committedSourceKeys,job.failures].every(Array.isArray)||job.error!==undefined&&typeof job.error!=='string')throw fault('invalid_import_job');
|
||||
if (job.version !== 1) 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' || !['paused','running','completed','partial','cancelled'].includes(job.status)) throw fault('invalid_import_job');
|
||||
if(job.source&&(!['native-ndjson','duplicate-ndjson'].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 (!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?.kind==='duplicate-snapshot'){
|
||||
const p=job.preparation;
|
||||
if(!p||p.version!==1||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');
|
||||
}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');
|
||||
@@ -87,6 +92,7 @@ export type ImportRunnerPorts={
|
||||
export function createImportRunner(ports:ImportRunnerPorts){
|
||||
const initial=ports.getJob(),identity={id:initial.id,projectId:initial.projectId,sourceFingerprint:initial.sourceFingerprint,parserVersion:initial.parserVersion},budget=ports.frameBudgetMs??6,maxPending=ports.maxPending??4096,maxBytes=ports.maxPendingBytes??1024*1024,maxBatch=ports.maxBatch??32,now=ports.now??(()=>performance.now());
|
||||
validateImportJob(initial);
|
||||
if(initial.source?.kind==='duplicate-snapshot')throw fault('import_interpreter_unavailable');
|
||||
if(!Number.isFinite(budget)||budget<=0||![maxPending,maxBytes,maxBatch].every(value=>Number.isSafeInteger(value)&&value>0))throw fault('invalid_import_budget');
|
||||
let costPerUnit:number|undefined;
|
||||
let active:Promise<void>|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set<symbol>();
|
||||
|
||||
@@ -317,6 +317,7 @@ async function duplicateRecoveryCase(name,assert){
|
||||
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map(versions),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
|
||||
try{
|
||||
engine=await openProjectEngine(options);const id=await engine.imports.duplicate([{id:'F',incarnation:1},{id:'A',incarnation:1}]);
|
||||
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']);
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createDocument,createRegistry,builtinComponents} from '../../src/lib/canvas-core/index.ts';
|
||||
import {encodeCopySnapshot,decodeCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot.ts';
|
||||
import {createImportJob,validateImportJob,createImportRunner} from '../../src/lib/canvas-runtime/imports.ts';
|
||||
const registry=createRegistry(builtinComponents()),node={id:'A',incarnation:2,structure:'node',ownerCanvasId:'P',componentId:'text',dataVersion:1,data:{title:'原始',text:'内容',unknown:{preserved:true}},geometry:{x:0,y:0,width:100,height:100,rotation:0}},document=createDocument('project',[{id:'P',parent:null}],[node],new Map([['deleted',3]]));
|
||||
const signal=()=>new AbortController().signal;
|
||||
test('copy snapshot roundtrip preserves the complete source, identity lifecycles and unknown data',async()=>{
|
||||
const bytes=await encodeCopySnapshot(document,signal()),decoded=await decodeCopySnapshot(bytes,'project',registry,signal());assert.deepEqual([...decoded.elements],[...document.elements]);assert.deepEqual([...decoded.canvases],[...document.canvases]);assert.deepEqual([...decoded.tombstones],[...document.tombstones]);assert.ok(Object.isFrozen(decoded.elements.get('A').data));await assert.rejects(decodeCopySnapshot(bytes,'different-project',registry,signal()),/invalid_copy_snapshot/);
|
||||
});
|
||||
test('snapshot decoding rejects malformed, future, duplicate, invalid and truncated sources without inventing a document',async()=>{
|
||||
for(const text of ['','{}','{bad}',JSON.stringify({kind:'header',version:2,projectId:'project'}),JSON.stringify({kind:'header',version:1,projectId:'project'})+'\n'+JSON.stringify({kind:'other',record:{}})])await assert.rejects(decodeCopySnapshot(new Blob([text]),'project',registry,signal()));
|
||||
const original=await(await encodeCopySnapshot(document,signal())).text(),row=original.trim().split('\n').at(-1);await assert.rejects(decodeCopySnapshot(new Blob([original+row]),'project',registry,signal()),/invalid_copy_snapshot/);
|
||||
const controller=new AbortController();controller.abort();await assert.rejects(encodeCopySnapshot(document,controller.signal),/copy_aborted/);await assert.rejects(decodeCopySnapshot(new Blob([original]),'project',registry,controller.signal),/copy_aborted/);
|
||||
});
|
||||
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',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:2}},{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/);
|
||||
});
|
||||
@@ -118,10 +118,39 @@ test('copy publishes independently of backend ACK and associates task visibility
|
||||
});
|
||||
|
||||
test('copy source write failure gates network and resume preserves the planned identities',async()=>{
|
||||
const f=fixture(),engine=await f.open();try{await seedCopy(engine);const sent=f.requests.length;f.broken=true;const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const before=engine.imports.getJobs().get(id);assert.equal(f.requests.length,sent);f.broken=false;await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.deepEqual(engine.imports.getJobs().get(id).stableIds,before.stableIds);}finally{f.broken=false;await engine.close();}
|
||||
const f=fixture(),engine=await f.open();try{await seedCopy(engine);const sent=f.requests.length;f.broken=true;const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const before=engine.imports.getJobs().get(id);assert.equal(f.requests.length,sent);f.broken=false;await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(before.source.kind,'duplicate-snapshot');assert.equal(before.preparation.allocationIds.length,0);assert.equal(Object.keys(engine.imports.getJobs().get(id).stableIds).length,5);}finally{f.broken=false;await engine.close();}
|
||||
});
|
||||
|
||||
|
||||
test('read-only policy rejects copy/import admission without storing a source or a running job',async()=>{
|
||||
const f=fixture(),engine=await f.open({policy:()=>({enabled:false,code:'read_only',targets:[]})});try{await assert.rejects(engine.imports.duplicate([{id:'missing',incarnation:1}]),/read_only/);assert.throws(()=>engine.imports.startNdjson(file([node('one')]),'canvas'),/read_only/);assert.equal(engine.imports.getJobs().size,0);await engine.settle();assert.equal(f.files.size,0);assert.equal(f.requests.length,0);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
|
||||
test('preparing copy checkpoints source and allocated identities, reopens paused and reconstructs the same mapping',async()=>{
|
||||
const f=fixture();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.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);await engine.imports.resume(id);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);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('closing an admitted preparing copy preserves its task and snapshot for automatic restart',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);assert.equal(engine.imports.getJobs().get(id).status,'preparing');await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).committedSourceKeys.length,5);assert.equal(f.errors.length,0);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('cancel during preparation is durable and never publishes a copy on reopening',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const size=engine.getSnapshot().document.elements.size,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.imports.cancel(id);await engine.settle();await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal(f.errors.length,0);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
|
||||
test('a reopened preparation rejects changed registry copy declarations instead of silently reinterpreting the source',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.imports.pause(id);await engine.close();
|
||||
const changed=createRegistry(builtinComponents().map(component=>component.id==='image'?{...component,copy:{...component.copy,omit:[...component.copy.omit,['customExecution']]}}:component));engine=await f.open({registry:changed});await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'paused');assert.equal(job.error,'copy_interpreter_changed');assert.equal(job.committedSourceKeys.length,0);
|
||||
}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('canonical recovery holds a preparing copy and restarts it only after the coordinated root is adopted',async()=>{
|
||||
const f=fixture(),engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);const recovery=await engine.prepareRecovery(),cut=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,cut);assert.equal(engine.imports.getJobs().get(id).status,'preparing');assert.equal((await recovery.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.errors.length,0);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user