diff --git a/src/components/canvas/headless-imports.tsx b/src/components/canvas/headless-imports.tsx index ad031cf..5f25086 100644 --- a/src/components/canvas/headless-imports.tsx +++ b/src/components/canvas/headless-imports.tsx @@ -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)[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)[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 &&
setOpen(event.currentTarget.open)}>导入与复制任务 · {jobs.some(job => job.status === "preparing") && "正在准备复制 · "}{jobs.reduce((count, job) => count + job.committedSourceKeys.length, 0)} 项已加入 {jobs.map((job, index) =>
{job.source?.name || `导入任务 ${index + 1}`} -

{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} 项等待`}

+

{["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} 项等待`}

{job.error &&

{reason(job.error)}

}
{["running","preparing"].includes(job.status) && } diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 8c607a2..2b43fce 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -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 Mapvoid>();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(); + function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);const token=Symbol(),releases=new Mapvoid>();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{ 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({ diff --git a/src/lib/canvas-runtime/import-integrity.ts b/src/lib/canvas-runtime/import-integrity.ts new file mode 100644 index 0000000..1a39f1b --- /dev/null +++ b/src/lib/canvas-runtime/import-integrity.ts @@ -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{ + 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(''); +} diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index 8f60d3a..8aa91b8 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -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 { 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|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set(); 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{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{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){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=Object.assign(Object.create(null),job.stableIds),used=new Set(Object.values(ids)); diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index caa2765..1f32c0f 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -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); diff --git a/tests/headless/copy-snapshot.test.mjs b/tests/headless/copy-snapshot.test.mjs index ee0a73e..a509d57 100644 --- a/tests/headless/copy-snapshot.test.mjs +++ b/tests/headless/copy-snapshot.test.mjs @@ -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/); }); diff --git a/tests/headless/import-integrity.test.mjs b/tests/headless/import-integrity.test.mjs new file mode 100644 index 0000000..3e70d3a --- /dev/null +++ b/tests/headless/import-integrity.test.mjs @@ -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}); +}); diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs index e24a790..32b2734 100644 --- a/tests/headless/imports-engine.test.mjs +++ b/tests/headless/imports-engine.test.mjs @@ -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();} +}); diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index c6aa163..7978b36 100644 --- a/tests/headless/imports.test.mjs +++ b/tests/headless/imports.test.mjs @@ -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()=>{