Bind import recovery checkpoints to source content digests
This commit is contained in:
@@ -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_copy_snapshot: "复制快照不可用", invalid_copy_preparation: "复制准备记录不可用", copy_interpreter_changed: "复制规则已变化，需要重新复制", 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_source_integrity_unavailable: "旧任务缺少源内容校验，无法安全继续；已加入内容保留", import_source_integrity_missing: "源校验记录缺失，任务已暂停", import_source_integrity_interpreter_unavailable: "源校验版本暂不支持", 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]);
|
||||
@@ -25,7 +25,7 @@ export function HeadlessImports({ engine, canvasId }: { engine: ProjectEngine; c
|
||||
{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 === "preparing" ? "正在准备复制" : 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">{["running","preparing"].includes(job.status) && job.source?.integrityVersion === 1 && !job.source.sha256 ? "正在校验源文件" : 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}>
|
||||
{["running","preparing"].includes(job.status) && <button onClick={() => control(engine.imports.pause(job.id))}>暂停导入</button>}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { encodeCopySnapshot, decodeCopySnapshot, copyRegistrySignature } from './copy-snapshot.ts';
|
||||
import { planDuplicateAsync } 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';
|
||||
import { assertNativeImportDependencies, assertNativeImportTarget, prepareNativeImport } from './import-native.ts';
|
||||
@@ -319,9 +320,10 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
function getImport(id:Id){const job=editor.getSnapshot().runtime.imports?.get(id);if(!job)throw fault('import_missing',[id]);return job;}
|
||||
function publishImport(job:ImportJob,expectedGeneration:number){
|
||||
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.version===2?job:freeze({...job,version:2 as const}));return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.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.version===3?job:freeze({...job,version:3 as const}));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 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);};}
|
||||
const deferredImportLoads=new Set<Id>();
|
||||
function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);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');
|
||||
@@ -335,17 +337,18 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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 encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),sha256=await importSourceDigest(encoded,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;
|
||||
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};
|
||||
}
|
||||
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 task=(async()=>{
|
||||
@@ -354,6 +357,13 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
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.integrityVersion!==1)throw fault('import_source_integrity_unavailable');
|
||||
const sha256=await importSourceDigest(bytes,preparationController.signal);
|
||||
if(closing||closed)return;
|
||||
const verified=getImport(id);if(verified.generation!==latest.generation||verified.sourceFingerprint!==job.sourceFingerprint||verified.source?.fileId!==source.fileId||['completed','partial','cancelled','paused'].includes(verified.status))return;
|
||||
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,
|
||||
@@ -366,7 +376,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(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);});
|
||||
})().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);if(!['running','preparing'].includes(getImport(id).status))deferredImportLoads.delete(id);if(!closing&&!closed&&!importHolds.size&&deferredImportLoads.delete(id)&&!importRunners.has(id)&&['running','preparing'].includes(getImport(id).status))queueMicrotask(()=>{if(!closing&&!closed)void loadImport(id);});});
|
||||
importLoads.set(id,task);return task;
|
||||
}
|
||||
const imports=Object.freeze({
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
/** Bind durable preparation/parse checkpoints to the complete immutable byte source. */
|
||||
export async function importSourceDigest(bytes:Blob,signal?:AbortSignal):Promise<string>{
|
||||
if(signal?.aborted)throw fault('import_aborted');
|
||||
const buffer=await bytes.arrayBuffer();if(signal?.aborted)throw fault('import_aborted');
|
||||
const digest=await crypto.subtle.digest('SHA-256',buffer);if(signal?.aborted)throw fault('import_aborted');
|
||||
return [...new Uint8Array(digest)].map(byte=>byte.toString(16).padStart(2,'0')).join('');
|
||||
}
|
||||
@@ -7,10 +7,10 @@ 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 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 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; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
|
||||
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;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;
|
||||
@@ -19,14 +19,19 @@ export type ImportJob = {
|
||||
};
|
||||
export function createImportJob(input: Pick<ImportJob, 'id' | 'projectId' | 'sourceFingerprint' | 'parserVersion' | 'source'>): ImportJob {
|
||||
if (![input.id,input.projectId,input.sourceFingerprint].every(value=>typeof value==='string'&&!!value) || !Number.isSafeInteger(input.parserVersion) || input.parserVersion<1) throw fault('invalid_import_job');
|
||||
return freeze({ id: input.id, projectId: input.projectId, sourceFingerprint: input.sourceFingerprint, parserVersion: input.parserVersion, ...(input.source?{source:clone(input.source)}:{}), version: 2, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
|
||||
return freeze({ id: input.id, projectId: input.projectId, sourceFingerprint: input.sourceFingerprint, parserVersion: input.parserVersion, ...(input.source?{source:{...clone(input.source),integrityVersion:input.source.integrityVersion??1}}:{}), version: 3, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
|
||||
}
|
||||
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 && job.version !== 2) throw fault('import_interpreter_unavailable');
|
||||
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){
|
||||
if(job.source.integrityVersion!==undefined&&job.source.integrityVersion!==1)throw fault('import_source_integrity_interpreter_unavailable');
|
||||
if(job.source.sha256!==undefined&&(typeof job.source.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(job.source.sha256))))throw fault('invalid_import_source');
|
||||
if(job.source.integrityVersion===1&&!job.source.sha256&&(job.cursor!==0||job.eof||job.pending.length||job.committedSourceKeys.length||Object.keys(job.stableIds).length||(job.preparation?.allocationIds.length??0)>0||(job.preparation?.steps??0)>0))throw fault('import_source_integrity_missing');
|
||||
}
|
||||
if(job.source?.kind==='duplicate-snapshot'){
|
||||
const p=job.preparation;
|
||||
if(!p||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');
|
||||
@@ -96,12 +101,13 @@ 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(initial.source&&(!initial.source.sha256||initial.source.integrityVersion!==1))throw fault('import_source_integrity_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>();
|
||||
const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}};
|
||||
function current(){const job=ports.getJob();if(job.id!==identity.id||job.projectId!==identity.projectId||job.sourceFingerprint!==identity.sourceFingerprint||job.parserVersion!==identity.parserVersion)throw fault('import_identity_changed');return job;}
|
||||
function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:2 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(value);}
|
||||
function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(value);}
|
||||
function publish(job:ImportJob,patch:Partial<ImportJob>){const value=next(job,patch);ports.publish(value,job.generation);if(!equal(current(),value))throw fault('invalid_import_publish');return value;}
|
||||
function prepare(line:ImportLine,job:ImportJob){
|
||||
const ids:Record<string,Id>=Object.assign(Object.create(null),job.stableIds),used=new Set(Object.values(ids));
|
||||
|
||||
@@ -319,6 +319,18 @@ async function resourceRecoveryCase(name, assert) {
|
||||
}
|
||||
|
||||
|
||||
async function sourceIntegrityCase(name,assert){
|
||||
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),registry=createRegistry(builtinComponents()),databaseName=`${name}-source-integrity`,storage=createBrowserStorage({name:databaseName}),scope={accountId:crypto.randomUUID(),workspaceId:workspace,projectId:root},original=createDocument(root,[{id:root,parent:null}],[]);let document=original,versions=new Map([[root,1]]),engine,calls=0,raw;
|
||||
const transport=async request=>{calls++;const body=JSON.parse(request.body);for(const[id,version]of Object.entries(body.expected_versions))assert(version===(versions.get(id)??0),'integrity lost a version fence');for(const operation of body.operations)document=applyChanges(document,operation.changes);for(const id of Object.keys(body.expected_versions))versions.set(id,(versions.get(id)??0)+1);return{status:200,body:{protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(row=>row.id),model:{documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:Object.fromEntries(document.tombstones),versions:Object.fromEntries(versions)}}};};
|
||||
const node=id=>({kind:'element',key:id,record:{id,incarnation:1,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:'$target',data:{title:id,text:''},geometry:{x:0,y:0,width:200,height:100,rotation:0}}}),bytes=new Blob([[node('first'),node('last')].map(row=>JSON.stringify(row)).join('\n')]),options={scope,registry,document:original,versions:new Map(versions),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport})};
|
||||
try{
|
||||
engine=await openProjectEngine(options);const id=engine.imports.startNdjson(bytes,root);await new Promise(resolve=>{const stop=engine.editor.subscribe((_,snapshot)=>{if(snapshot.runtime.imports.get(id).committedSourceKeys.length===1){stop();resolve();}});});await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),first=before.stableIds['element:first'];assert(/^[a-f0-9]{64}$/.test(before.source.sha256),'source hash was not bound before publication');await engine.close();const sent=calls;
|
||||
raw=await new Promise((resolve,reject)=>{const request=indexedDB.open(databaseName);request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);});const key=[JSON.stringify([scope.accountId,scope.workspaceId,scope.projectId]),before.source.fileId];
|
||||
const write=async value=>{const tx=raw.transaction('files','readwrite'),done=new Promise((resolve,reject)=>{tx.oncomplete=resolve;tx.onabort=()=>reject(tx.error);});tx.objectStore('files').put(value,key);await done;};const text=await bytes.text(),changed=text.replaceAll('last','evil');assert(new Blob([changed]).size===bytes.size,'corruption fixture changed its byte length');await write(new Blob([changed]));engine=await openProjectEngine(options);await engine.imports.resume(id);await engine.settle();assert(engine.imports.getJobs().get(id).error==='import_source_changed','same-length source tampering was accepted');assert(engine.getSnapshot().document.elements.size===1&&engine.getSnapshot().document.elements.has(first)&&calls===sent,'source tampering changed the preserved prefix or sent new work');
|
||||
await write(bytes);await engine.imports.resume(id);await engine.settle();assert(engine.imports.getJobs().get(id).status==='completed'&&document.elements.size===2&&engine.imports.getJobs().get(id).stableIds['element:first']===first,'restoring exact bytes did not resume the original task');
|
||||
}finally{raw?.close();await engine?.close();storage.close();indexedDB.deleteDatabase(databaseName);}
|
||||
}
|
||||
|
||||
async function deltaJournalCase(name,assert){
|
||||
const databaseName=`${name}-delta-journal`,scope={accountId:'delta-account',workspaceId:'workspace',projectId:'project'},key=JSON.stringify(Object.values(scope)),rows=Array.from({length:100},(_,i)=>({id:String(i),text:'x'.repeat(1000)})),image=revision=>({formatVersion:1,rootRevision:revision,localRevision:revision,payload:{rows:rows.map((row,i)=>i===0?{...row,text:`version-${revision}`}:row),ledger:{frozenBody:'exact original request bytes'}}});
|
||||
const done=tx=>new Promise((resolve,reject)=>{tx.oncomplete=resolve;tx.onabort=()=>reject(tx.error);}),read=request=>new Promise((resolve,reject)=>{request.onsuccess=()=>resolve(request.result);request.onerror=()=>reject(request.error);});
|
||||
@@ -573,6 +585,8 @@ export async function run() {
|
||||
results.push('owned immutable recovery images survive IndexedDB checksums; mutable images and file descriptors are captured before asynchronous work');
|
||||
await cooperativeCopyCase(assert);
|
||||
results.push('cooperative duplicate planning yields to actual browser tasks and cancellation without mutating the source');
|
||||
await sourceIntegrityCase(name,assert);
|
||||
results.push('source SHA256 detects same-length Blob tampering before recovery adds or sends a tail; exact bytes resume with the original prefix and IDs');
|
||||
await deltaJournalCase(name,assert);
|
||||
results.push('delta journals migrate legacy checkpoints, retain atomic files, verify hashes/base chains, reject old readers and compact through serialized immutable commits');
|
||||
await importBackpressureCase(name,assert);
|
||||
|
||||
@@ -14,7 +14,7 @@ test('snapshot decoding rejects malformed, future, duplicate, invalid and trunca
|
||||
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);
|
||||
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: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/);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {importSourceDigest} from '../../src/lib/canvas-runtime/import-integrity.ts';
|
||||
import {createImportJob,validateImportJob} from '../../src/lib/canvas-runtime/imports.ts';
|
||||
test('source SHA256 binds complete bytes and cancellation fences both sides of asynchronous reading',async()=>{
|
||||
assert.equal(await importSourceDigest(new Blob(['abc'])),'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad');assert.notEqual(await importSourceDigest(new Blob(['abc'])),await importSourceDigest(new Blob(['abd'])));
|
||||
const controller=new AbortController();controller.abort();await assert.rejects(importSourceDigest(new Blob(['abc']),controller.signal),/import_aborted/);let release;const gate=new Promise(resolve=>{release=resolve;}),later=new AbortController(),bytes=new class extends Blob{async arrayBuffer(){await gate;return super.arrayBuffer();}}(['abc']),result=importSourceDigest(bytes,later.signal);later.abort();release();await assert.rejects(result,/import_aborted/);
|
||||
});
|
||||
test('new source jobs declare integrity version and reject progress without a bound hash; legacy data remains readable',()=>{
|
||||
const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'file',parserVersion:1,source:{kind:'native-ndjson',fileId:'file',targetCanvasId:'canvas',byteLength:10}});assert.equal(job.version,3);assert.equal(job.source.integrityVersion,1);validateImportJob(job);
|
||||
assert.throws(()=>validateImportJob({...job,cursor:1}),/import_source_integrity_missing/);assert.throws(()=>validateImportJob({...job,stableIds:{'element:A':'new-A'}}),/import_source_integrity_missing/);assert.throws(()=>validateImportJob({...job,source:{...job.source,sha256:'invalid'}}),/invalid_import_source/);assert.throws(()=>validateImportJob({...job,source:{...job.source,integrityVersion:2}}),/import_source_integrity_interpreter_unavailable/);
|
||||
const source={...job.source};delete source.integrityVersion;validateImportJob({...job,version:2,source,cursor:1});validateImportJob({...job,source:{...job.source,sha256:'a'.repeat(64)},cursor:1});
|
||||
});
|
||||
@@ -174,3 +174,22 @@ test('blocked source persistence admits several local batches then applies bound
|
||||
gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,500);assert.equal(f.errors.length,0);assert.equal(f.files.size,1);decodeRecovery(f.image(),registry);
|
||||
}finally{gate.resolve();await engine.close();}
|
||||
});
|
||||
|
||||
test('same-length changes to an admitted copy snapshot are rejected before preparation allocates or publishes content',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 until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),raw=await f.files.get(job.source.fileId).text(),changed=raw.replace('source','tamper');assert.equal(changed.length,raw.length);await engine.close();const sent=f.requests.length;f.files.set(job.source.fileId,new Blob([changed]));engine=await f.open();await engine.imports.resume(id);await engine.settle();const blocked=engine.imports.getJobs().get(id);assert.equal(blocked.status,'paused');assert.equal(blocked.error,'import_source_changed');assert.equal(blocked.committedSourceKeys.length,0);assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal(f.requests.length,sent);assert.equal(blocked.source.sha256,job.source.sha256);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('a changed unparsed tail cannot pass native import recovery or alter its preserved prefix',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),raw=await f.files.get(job.source.fileId).text(),changed=raw.replaceAll('three','other');assert.equal(changed.length,raw.length);const first=job.stableIds['element:one'];await engine.close();const sent=f.requests.length;f.files.set(job.source.fileId,new Blob([changed]));engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).error,'import_source_changed');assert.equal(engine.getSnapshot().document.elements.size,1);assert.ok(engine.getSnapshot().document.elements.has(first));assert.equal(f.requests.length,sent);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
test('late source digest cannot revive a cancelled job or publish into a held recovery root',async()=>{
|
||||
for(const cancel of [true,false]){
|
||||
const f=fixture(),engine=await f.open(),entered=deferred(),gate=deferred(),bytes=new class extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}([JSON.stringify(node('one'))]);
|
||||
try{const id=engine.imports.startNdjson(bytes,'canvas');await entered.promise;if(cancel){await engine.imports.cancel(id);gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.imports.getJobs().get(id).source.sha256,undefined);}else{const preparing=engine.prepareRecovery();gate.resolve();const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,1);}assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
|
||||
}
|
||||
});
|
||||
|
||||
test('legacy unfinished sources remain readable but cannot acquire a new content identity silently',async()=>{
|
||||
const f=fixture();let engine=await f.open();try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();await engine.close();const image=f.image(),job=image.payload.runtime.imports.find(([key])=>key===id)[1];job.version=2;delete job.source.sha256;delete job.source.integrityVersion;const sent=f.requests.length;engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).error,'import_source_integrity_unavailable');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.requests.length,sent);decodeRecovery(f.image(),registry);}finally{await engine.close();}
|
||||
});
|
||||
|
||||
@@ -64,7 +64,7 @@ test('dependency backpressure pauses with locators intact and a larger budget re
|
||||
});
|
||||
|
||||
test('future interpreter versions and malformed job cuts fail before any source read',()=>{
|
||||
const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'s',parserVersion:1});assert.throws(()=>validateImportJob({...job,version:3}),/import_interpreter_unavailable/);assert.throws(()=>validateImportJob({...job,stableIds:{a:'same',b:'same'}}),/invalid_import_job/);assert.throws(()=>validateImportJob({...job,status:'completed'}),/invalid_import_job/);
|
||||
const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'s',parserVersion:1});assert.throws(()=>validateImportJob({...job,version:4}),/import_interpreter_unavailable/);assert.throws(()=>validateImportJob({...job,stableIds:{a:'same',b:'same'}}),/invalid_import_job/);assert.throws(()=>validateImportJob({...job,status:'completed'}),/invalid_import_job/);
|
||||
});
|
||||
|
||||
test('fast independent records stage bounded publications and commits while the first record remains immediate',async()=>{
|
||||
@@ -105,8 +105,8 @@ test('a durable pending cycle resumes without reallocating IDs or exposing a par
|
||||
});
|
||||
|
||||
|
||||
test('new jobs carry interpreter version 2 and a resumed legacy job upgrades in its first published cut',async()=>{
|
||||
const f=fixture([{id:'one'}]);assert.equal(f.root.job.version,2);f.root={...f.root,job:{...f.root.job,version:1}};validateImportJob(f.root.job);const publish=f.ports.publish;f.ports.publish=(job,expected)=>{assert.equal(job.version,2);return publish(job,expected);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.version,2);
|
||||
test('new jobs carry interpreter version 3 and a resumed legacy job upgrades in its first published cut',async()=>{
|
||||
const f=fixture([{id:'one'}]);assert.equal(f.root.job.version,3);f.root={...f.root,job:{...f.root.job,version:1}};validateImportJob(f.root.job);const publish=f.ports.publish;f.ports.publish=(job,expected)=>{assert.equal(job.version,3);return publish(job,expected);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.version,3);
|
||||
});
|
||||
|
||||
test('optimistic local checkpoints use host backpressure without declaring completion ahead of persistence',async()=>{
|
||||
|
||||
Reference in New Issue
Block a user