From be6a092c4017a42fdbb1321d4906b324b3f04bb7 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Mon, 9 Nov 2026 20:01:32 +0800 Subject: [PATCH] Stream exported document records with versioned recovery --- headless/README.md | 24 ++- src/lib/canvas-runtime/engine.ts | 7 +- .../canvas-runtime/import-document-stream.ts | 147 ++++++++++++++++++ src/lib/canvas-runtime/imports.ts | 2 +- src/lib/canvas-runtime/recovery-encoding.ts | 4 +- src/lib/canvas-runtime/recovery.ts | 6 +- src/lib/canvas-runtime/storage-port.ts | 2 +- src/lib/canvas-runtime/storage.ts | 2 +- tests/headless/asset-adoption.test.mjs | 2 +- .../headless/import-document-stream.test.mjs | 50 ++++++ tests/headless/imports-engine.test.mjs | 29 +++- tests/headless/recovery-encoding.test.mjs | 2 +- tests/headless/renderer.integration.mjs | 7 + 13 files changed, 262 insertions(+), 22 deletions(-) create mode 100644 src/lib/canvas-runtime/import-document-stream.ts create mode 100644 tests/headless/import-document-stream.test.mjs diff --git a/headless/README.md b/headless/README.md index d6a68d1..cb48b9f 100644 --- a/headless/README.md +++ b/headless/README.md @@ -341,7 +341,7 @@ array. Array record errors are distinct from document schema/reference failures. JSON array jobs preserve the original Blob and integrity digest, parserVersion 3, byte cursor, stable IDs, pending dependencies and failures using the same ImportJob and commit boundary as NDJSON. Such roots use recovery envelope 14; readers accept -1–14 and reject 15. Downgrading an array-bearing image to 13 rejects. This changes +1–16 and reject 17. Downgrading an array-bearing image to 13 rejects. This changes neither IndexedDB schema v4 nor the server wire protocol. Native array items have the same `{kind,key,record}` schema as NDJSON; `$target` denotes the chosen canvas. @@ -349,12 +349,19 @@ the same `{kind,key,record}` schema as NDJSON; `$target` denotes the chosen canv format v1. The JSON picker distinguishes object envelopes from record arrays. Source canvas ownership and references map to fresh IDs; its root maps to the explicit target, and tombstones are not copied into the target. The original -file/digest and parserVersion 4 persist; the ordinal cursor resumes against the -same indexed bytes. Document jobs use recovery v15 (readers accept 1–15, reject -16), without changing DB v4 or the wire protocol. Resource-bearing records use fresh authorization when the host provides a depot -and authority; missing capabilities or denied resources produce explicit partial -results. Complete service acceptance remains pending. Preparation -still scans the envelope before progressive payload decoding and insertion. +file/digest and parserVersion persist. New jobs use parser 5: after the version, +project and canvas headers, records are scanned and decoded on demand rather than +indexing the full elements array first. The ordinal cursor resumes against the +same bytes. Legal prefixes survive malformed array/footer tails with an explicit +partial result. Headers following the elements array require indexed preparation +to establish ownership first. Legacy parser 4 jobs keep their indexed interpreter. +New document jobs use recovery v16; legacy document jobs keep v15 (readers accept +1–16, reject 17), without changing DB v4 or the wire protocol. Resource-bearing +records use fresh authorization when the host provides a depot and authority; +missing capabilities or denied resources produce explicit partial results. +Complete service acceptance remains pending. Source SHA verification still reads +the original file before the runner starts; this removes the second full range +scan, not the remaining integrity startup barrier. ImportRunner hosts may provide async `beforeCommit(units, job, signal)` for read-only preparation such as resource permission checks. It returns Availability; a denial @@ -481,4 +488,5 @@ Native import scanners share `ImportScheduling` (`now`, `yieldTask`, threshold also yields when the clock does not advance. Cancellation is checked again after each real task yield. Byte/ordinal cursors, parser versions and source fingerprints are unchanged. This is a soft budget: a bounded JSON decode is -indivisible, and document-envelope range indexing still precedes record display. +indivisible. Legacy and out-of-order document headers retain indexed preparation; +canonical parser 5 headers allow progressive payload scanning. diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index c30cb65..18b4b1c 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -19,6 +19,7 @@ import type { DigestPort } from './digest-pool.ts'; import { importSourceDigest } from './import-integrity.ts'; import { encodeDuplicateSource, prepareDuplicateImport, assertDuplicateExternal } from './import-duplicate.ts'; import {openDocumentImportSource} from './import-document.ts'; +import {openStreamingDocumentImportSource} from './import-document-stream.ts'; import {createJsonArrayImportSource} from './import-json-array.ts'; import { createImportJob, createImportRunner, createNdjsonImportSource, type ImportJob } from './imports.ts'; import { assertNativeImportDependencies, assertNativeImportTarget, prepareNativeImport } from './import-native.ts'; @@ -674,7 +675,7 @@ export async function openProjectEngine(options: EngineOptions) { let expectedFingerprint=getImport(id).sourceFingerprint; const preparationController=new AbortController();if(['native-document','duplicate-snapshot','duplicate-plan'].includes(getImport(id).source?.kind??''))preparationControllers.set(id,preparationController); const task=(async()=>{ - let job=getImport(id),source=job.source;if(!source||!['native-document','native-json-array','native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(source.kind)||job.parserVersion!==(source.kind==='native-document'?4:source.kind==='native-json-array'?3:source.kind==='duplicate-plan'?2:1))throw fault('import_interpreter_unavailable'); + let job=getImport(id),source=job.source;if(!source||!['native-document','native-json-array','native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(source.kind)||(source.kind==='native-document'?![4,5].includes(job.parserVersion):job.parserVersion!==(source.kind==='native-json-array'?3:source.kind==='duplicate-plan'?2:1)))throw fault('import_interpreter_unavailable'); let bytes=fileWrites.get(source.fileId)??await options.storage.readFile(leader,source.fileId); if(closing||closed)return; const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status))return; @@ -690,7 +691,7 @@ export async function openProjectEngine(options: EngineOptions) { if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;copyDocument=materialized.document;indexed=materialized.indexed;expectedFingerprint=job.sourceFingerprint;} if(source.kind==='duplicate-plan'&&source.snapshotVersion===2){indexed??=await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal);copyDocument=indexed.document;} if(source.kind==='duplicate-plan'&©Document)validateLoadingCopySource(editor.getSnapshot(),job,copyDocument); - const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await openDocumentImportSource({bytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}); + const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await (job.parserVersion===5?openStreamingDocumentImportSource:openDocumentImportSource)({bytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,fingerprint:job.sourceFingerprint}); if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return; if(importHolds.size){deferredImportLoads.add(id);return;} const authorizedImportAssets=new Map(),assetPreparationSignals=new WeakSet(); @@ -723,7 +724,7 @@ export async function openProjectEngine(options: EngineOptions) { assertImportWritable();const loadingAdmission=loadingImportAdmission(editor.getSnapshot(),targetCanvasId);const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source'); const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict'); const entry=canvas.parent?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined; - const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:format==='document'?4:format==='json-array'?3:1,source:{...(loadingAdmission?{loadingAdmission}:{}),kind:format==='document'?'native-document':format==='json-array'?'native-json-array':'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const}); + const job=freeze({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:format==='document'?5:format==='json-array'?3:1,source:{...(loadingAdmission?{loadingAdmission}:{}),kind:format==='document'?'native-document':format==='json-array'?'native-json-array':'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const}); 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; diff --git a/src/lib/canvas-runtime/import-document-stream.ts b/src/lib/canvas-runtime/import-document-stream.ts new file mode 100644 index 0000000..6ca62d6 --- /dev/null +++ b/src/lib/canvas-runtime/import-document-stream.ts @@ -0,0 +1,147 @@ +import type {ImportLine,ImportSource} from './imports.ts'; +import {createImportBudget,type ImportScheduling} from './import-budget.ts'; +import {openDocumentImportSource} from './import-document.ts'; +import {fault,validJson} from '../canvas-core/values.ts'; + +type Range={start:number;end:number}; +type ScanCache={start:number;bytes:Uint8Array}; +type Canvas={id:string;parent:null|{canvasId:string;entryElementId:string}}; +type Options=ImportScheduling&{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number}; +const allowed=new Set(['documentFormatVersion','projectId','canvases','elements','tombstones']); +const whitespace=(byte:number)=>byte===9||byte===10||byte===13||byte===32; + +/** Each scan owns its cursor. An aborted scan cannot advance the resumable source. */ +function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={start:-1,bytes:new Uint8Array()}){ + const {bytes}=options,budget=createImportBudget(signal,options); + let cursor=start; + const check=()=>{if(signal.aborted)throw fault('import_aborted');}; + async function peek(){ + check();if(cursor>=bytes.size)return -1; + if(cursor=cache.start+cache.bytes.length){const loaded=new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer());cache.start=cursor;cache.bytes=loaded;check();} + return cache.bytes[cursor-cache.start]; + } + function step(){cursor++;return budget.step();} + async function spaces(){while(whitespace(await peek()))await step();} + async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();} + async function value():Promise{ + await spaces();const start=cursor,stack:number[]=[];let quoted=false,escaped=false; + while(true){ + const byte=await peek(); + if(byte<0){if(quoted||stack.length)throw fault('invalid_document_import');break;} + if(!quoted&&!stack.length&&(byte===44||byte===93||byte===125||whitespace(byte)))break; + if(quoted){if(escaped)escaped=false;else if(byte===92)escaped=true;else if(byte===34)quoted=false;} + else if(byte===34)quoted=true; + else if(byte===91||byte===123){stack.push(byte);if(stack.length>256)throw fault('json_nesting_limit');} + else if(byte===93||byte===125){if(stack.pop()!==(byte===93?91:123))throw fault('invalid_document_import');} + const ended=!quoted&&!stack.length&&(byte===34||byte===93||byte===125); + await step();if(ended)break; + } + if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor}; + } + async function decode(range:Range,limit=options.maxRecordBytes??256*1024):Promise{ + check();if(range.end-range.start>limit)throw fault('record_too_large'); + let text:string; + try{text=new TextDecoder('utf-8',{fatal:true}).decode(await bytes.slice(range.start,range.end).arrayBuffer());}catch(error){check();throw fault('invalid_utf8');} + check();try{const value=JSON.parse(text);if(!validJson(value))throw Error();return value;}catch{throw fault('invalid_json');} + } + return{peek,step,spaces,take,value,decode,position:()=>cursor}; +} + +function tombstones(value:unknown){ + if(!value||typeof value!=='object'||Array.isArray(value)||Object.values(value).some(n=>!Number.isSafeInteger(n)||(n as number)<1))throw fault('invalid_document_import'); +} + +/** Parser 5: canonical headers open before scanning the elements tail. + * Out-of-order headers require the indexed interpreter to establish ownership first. + * Ordinals remain stable; ranges are retained only as records are requested. */ +export async function openStreamingDocumentImportSource(input:Options):Promise{ + const options:Options=Object.freeze({...input}); + const {bytes,signal}=options,max=options.maxRecordBytes??256*1024; + if(!(bytes instanceof Blob)||bytes.size>64*1024*1024||!options.fingerprint||!Number.isSafeInteger(max)||max<1)throw fault('invalid_import_source'); + // Only range extension scans bytes; ready-record decoding uses independent slices. + // Keep one source-owned chunk across attempts instead of re-reading 16 KiB per row. + const cache:ScanCache={start:-1,bytes:new Uint8Array()},scanAt=(signal:AbortSignal,start=0)=>scanner(options,signal,start,cache); + const header=scanAt(signal),fields=new Map(); + await header.take(123); + while(true){ + const key=await header.decode(await header.value(),1024); + if(typeof key!=='string'||!allowed.has(key)||fields.has(key))throw fault('invalid_document_import'); + await header.take(58); + if(key==='elements'){ + if(!fields.has('documentFormatVersion')||!fields.has('projectId')||!fields.has('canvases')){ + const indexed=await openDocumentImportSource(options); + return Object.freeze({...indexed,parserVersion:5}); + } + await header.take(91);fields.set(key,true);break; + } + fields.set(key,await header.decode(await header.value(),4*1024*1024)); + await header.take(44); + } + if(fields.get('documentFormatVersion')!==1||typeof fields.get('projectId')!=='string'||!fields.get('projectId')||!Array.isArray(fields.get('canvases')))throw fault('invalid_document_import'); + if(fields.has('tombstones'))tombstones(fields.get('tombstones')); + const canvases=fields.get('canvases') as Canvas[],byId=new Map(); + const budget=createImportBudget(signal,options); + for(const canvas of canvases){ + if(!canvas||typeof canvas.id!=='string'||!canvas.id||byId.has(canvas.id)||canvas.parent!==null&&(!canvas.parent||typeof canvas.parent.canvasId!=='string'||typeof canvas.parent.entryElementId!=='string'))throw fault('invalid_document_import'); + byId.set(canvas.id,canvas);await budget.step(); + } + const roots=canvases.filter(canvas=>canvas.parent===null);if(roots.length!==1)throw fault('invalid_document_import'); + const rootId=roots[0].id,rooted=new Set([rootId]); + for(const canvas of canvases){ + const path=new Set();let current=canvas; + while(!rooted.has(current.id)){ + if(path.has(current.id)||!current.parent)throw fault('invalid_document_import'); + path.add(current.id);const parent=byId.get(current.parent.canvasId);if(!parent)throw fault('invalid_document_import');current=parent;await budget.step(); + } + for(const id of path){rooted.add(id);await budget.step();} + } + const children=canvases.filter(canvas=>canvas.id!==rootId),ranges:Range[]=[]; + let cursor=header.position(),phase:'elements'|'tail'|'done'='elements',first=true,failure:string|undefined,serial=Promise.resolve(); + async function scanNext(readSignal:AbortSignal){ + const scan=scanAt(readSignal,cursor); + if(phase==='elements'){ + await scan.spaces(); + if(await scan.peek()===93){if(!first)throw fault('invalid_document_import');await scan.step();cursor=scan.position();phase='tail';} + else{ + const range=await scan.value();await scan.spaces();const delimiter=await scan.peek(); + if(delimiter!==44&&delimiter!==93)throw fault('invalid_document_import'); + await scan.step();cursor=scan.position();phase=delimiter===93?'tail':'elements';first=false;ranges.push(range);return; + } + } + const seen=new Set(fields.keys());await scan.spaces(); + while(await scan.peek()!==125){ + await scan.take(44);const key=await scan.decode(await scan.value(),1024); + if(typeof key!=='string'||!allowed.has(key)||seen.has(key))throw fault('invalid_document_import'); + seen.add(key);await scan.take(58);const value=await scan.decode(await scan.value(),4*1024*1024); + if(key==='tombstones')tombstones(value); + await scan.spaces(); + } + await scan.take(125);await scan.spaces();if(scan.position()!==bytes.size)throw fault('invalid_document_import'); + cursor=scan.position();phase='done'; + } + async function read(position:number,readSignal:AbortSignal):Promise{ + if(readSignal.aborted)throw fault('import_aborted'); + if(!Number.isSafeInteger(position)||position<0)throw fault('invalid_import_cursor'); + const elementIndex=position-children.length; + while(elementIndex>=ranges.length&&phase!=='done'){ + try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');failure=(error as {code?:string}).code??'invalid_document_import';phase='done';} + } + const count=children.length+ranges.length; + if(position>=count){ + if(failure&&position===count)return{cursor:position,nextCursor:position+1,key:`record:${position}`,error:failure}; + if(position===count+(failure?1:0))return null; + throw fault('invalid_import_cursor'); + } + const base={cursor:position,nextCursor:position+1,key:`record:${position}`}; + try{ + const kind=positionread(position,readSignal));serial=result.then(()=>{},()=>{});return result; + }}); +} diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index e272547..6ad07e1 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -31,7 +31,7 @@ export function validateImportJob(job: ImportJob): void { 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-document','native-json-array','native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(job.source.kind)||['duplicate-ndjson','duplicate-plan'].includes(job.source.kind)&&(!job.source.copy||!Number.isSafeInteger(job.source.copy.sourceRevision)||job.source.copy.sourceRevision<0||!Array.isArray(job.source.copy.roots)||!job.source.copy.roots.length||job.source.copy.roots.some(id=>typeof id!=='string'||!id))||job.source.name!==undefined&&(typeof job.source.name!=='string'||job.source.name.length>255)||!job.source.fileId||!job.source.targetCanvasId||!Number.isSafeInteger(job.source.byteLength)||job.source.byteLength<0||job.source.targetEntry&&(!job.source.targetEntry.id||!Number.isSafeInteger(job.source.targetEntry.incarnation)||job.source.targetEntry.incarnation<1)))throw fault('invalid_import_source'); - if(job.source?.kind==='native-document'&&job.parserVersion!==4)throw fault('import_interpreter_unavailable'); + if(job.source?.kind==='native-document'&&![4,5].includes(job.parserVersion))throw fault('import_interpreter_unavailable'); if(job.source?.kind==='native-json-array'&&job.parserVersion!==3)throw fault('import_interpreter_unavailable'); if(job.source?.loadingAdmission){const proof=job.source.loadingAdmission;if(proof.version!==1||typeof proof.sourceToken!=='string'||!proof.sourceToken||proof.sourceToken.length>256||!Number.isSafeInteger(proof.operationSequence)||proof.operationSequence<0||!Array.isArray(proof.requiredCanvasIds)||!proof.requiredCanvasIds.length||proof.requiredCanvasIds.some(id=>typeof id!=='string'||!id)||new Set(proof.requiredCanvasIds).size!==proof.requiredCanvasIds.length)throw fault('loading_import_interpreter_unavailable');} if(job.source){ diff --git a/src/lib/canvas-runtime/recovery-encoding.ts b/src/lib/canvas-runtime/recovery-encoding.ts index 34cbb49..dd25756 100644 --- a/src/lib/canvas-runtime/recovery-encoding.ts +++ b/src/lib/canvas-runtime/recovery-encoding.ts @@ -52,8 +52,8 @@ export function encodeRecoveryValue(input:unknown):Json { } /** Only codec-owned, validated immutable images qualify for the storage copy fast path. */ -export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9|10|11|12|13|14|15,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage { - if(![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image'); +export function sealRecoveryImage(formatVersion:1|2|3|4|5|6|7|8|9|10|11|12|13|14|15|16,rootRevision:number,localRevision:number,payload:unknown):RecoveryImage { + if(![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16].includes(formatVersion)||!Number.isSafeInteger(rootRevision)||rootRevision<0||!Number.isSafeInteger(localRevision)||localRevision<0||localRevision>rootRevision)throw fault('corrupt_recovery_image'); const image=Object.freeze({formatVersion,rootRevision,localRevision,payload:encodeRecoveryValue(payload)});sealed.add(image);return image; } export function isSealedRecoveryImage(image:RecoveryImage):boolean{return sealed.has(image);} diff --git a/src/lib/canvas-runtime/recovery.ts b/src/lib/canvas-runtime/recovery.ts index 9651f8f..03ec2fd 100644 --- a/src/lib/canvas-runtime/recovery.ts +++ b/src/lib/canvas-runtime/recovery.ts @@ -30,7 +30,7 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage { operations: [...snapshot.operations], runtime: { ...(snapshot.runtime.assetBindings?.size?{assetBindings:[...snapshot.runtime.assetBindings]}:{}), ...(snapshot.runtime.loadingRead?{loadingRead:{document:documentValue(snapshot.runtime.loadingRead.document),versions:[...snapshot.runtime.loadingRead.versions]}}:{}), ...(snapshot.runtime.loading?{loading:snapshot.runtime.loading}:{}), ...(snapshot.runtime.orphanFiles?.size?{orphanFiles:[...snapshot.runtime.orphanFiles]}:{}), imports:[...(snapshot.runtime.imports??[])], recoveries:[...(snapshot.runtime.recoveries??[])], runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } }, }; - return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload); + return sealRecoveryImage([...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5)?16:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-document')?15:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.kind==='native-json-array')?14:snapshot.runtime.assetBindings?.size?13:snapshot.runtime.loading?(snapshot.runtime.loading.version===6?12:snapshot.runtime.loading.version===5?11:snapshot.runtime.loading.version===4?10:snapshot.runtime.loading.version===3?9:snapshot.runtime.loading.version===2?8:7):[...snapshot.operations.values()].some(intent=>intent.historyEffect)||[...snapshot.runtime.resources.values()].some(resource=>resource.sizing)?6:snapshot.runtime.orphanFiles?.size?5:[...(snapshot.runtime.imports?.values()??[])].some(job=>job.source?.retirement||job.source?.previousFiles?.length)?4:[...snapshot.runtime.resources.values()].some(resource=>resource.retirement)?3:snapshot.runtime.imports?.size ? 2 : 1,snapshot.rootRevision,snapshot.revision,payload); } type ObjectValue = Record; function object(value: unknown): ObjectValue { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('corrupt_recovery_image'); return value as ObjectValue; } @@ -94,7 +94,7 @@ function readHistory(history:ObjectValue,operations:ReadonlyMap image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image'); // Own the parsed records; mutating a caller's RecoveryImage must not mutate live state. @@ -250,7 +250,7 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn }); const imports=table(runtime.imports??[],(value,id)=>{const job=value as ImportJob;validateImportJob(job);if(job.id!==id||job.projectId!==document.projectId)throw fault('corrupt_recovery_image');return freeze(job);}); const hasJsonArrays=[...imports.values()].some(job=>job.source?.kind==='native-json-array');if(hasJsonArrays&&image.formatVersion<14||image.formatVersion===14&&!hasJsonArrays)throw fault('unsupported_recovery_version'); - const hasDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document');if(hasDocuments&&image.formatVersion!==15||image.formatVersion===15&&!hasDocuments)throw fault('unsupported_recovery_version'); + const hasDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'),hasStreamingDocuments=[...imports.values()].some(job=>job.source?.kind==='native-document'&&job.parserVersion===5);if(hasDocuments&&image.formatVersion!==(hasStreamingDocuments?16:15)||image.formatVersion>=15&&!hasDocuments)throw fault('unsupported_recovery_version'); for(const [id,entry]of importFiles(imports)){if(resources.has(id)||assetBindings.has(id))throw fault('import_file_identity_conflict',[id]);if((entry.file.retirement||imports.get(entry.jobId)?.source?.previousFiles?.length)&&image.formatVersion<4)throw fault('unsupported_recovery_version');} const orphanFiles=table(runtime.orphanFiles??[],(file,id)=>{validateOrphanFile(file,id);return freeze(file);}); if(orphanFiles.size&&image.formatVersion<5)throw fault('unsupported_recovery_version');for(const id of orphanFiles.keys())if(resources.has(id)||assetBindings.has(id)||importFiles(imports).has(id))throw fault('local_file_identity_conflict',[id]); diff --git a/src/lib/canvas-runtime/storage-port.ts b/src/lib/canvas-runtime/storage-port.ts index d3cfa61..00f6916 100644 --- a/src/lib/canvas-runtime/storage-port.ts +++ b/src/lib/canvas-runtime/storage-port.ts @@ -2,7 +2,7 @@ import type { Json } from '../canvas-core/types.ts'; /** Portable durable-store contract. Browser database types belong to the adapter. */ export type StorageScope = { readonly accountId: string; readonly workspaceId: string; readonly projectId: string }; -export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json }; +export type RecoveryImage = { readonly formatVersion: 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16; readonly rootRevision: number; readonly localRevision: number; readonly payload: Json }; export type DurableBatch = { readonly image: RecoveryImage; readonly deleteFiles?:readonly string[]; readonly depot?:import('./asset-depot.ts').DepotCut; readonly files?: readonly { readonly id: string; readonly bytes: Blob }[] }; export type StorageLeader = { readonly scope: StorageScope; readonly fence: number; readonly active: boolean; release(): Promise }; export type StoredFileInfo={readonly id:string;readonly byteLength:number;readonly mimeType:string}; diff --git a/src/lib/canvas-runtime/storage.ts b/src/lib/canvas-runtime/storage.ts index 8cfdc1a..b8a162a 100644 --- a/src/lib/canvas-runtime/storage.ts +++ b/src/lib/canvas-runtime/storage.ts @@ -31,7 +31,7 @@ function purgedFileIds(image:RecoveryImage):Set{ return result; } function validateImage(image: RecoveryImage) { - if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15].includes(image.formatVersion)) throw fault('unsupported_recovery_version'); + if (![1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16].includes(image.formatVersion)) throw fault('unsupported_recovery_version'); if (!Number.isSafeInteger(image.rootRevision) || image.rootRevision < 0 || !Number.isSafeInteger(image.localRevision) || image.localRevision < 0 || image.localRevision > image.rootRevision || !validJson(image.payload)) throw fault('corrupt_recovery_image'); } const requestValue = (request: IDBRequest) => new Promise((resolve, reject) => { request.onsuccess = () => resolve(request.result); request.onerror = () => reject(request.error); }); diff --git a/tests/headless/asset-adoption.test.mjs b/tests/headless/asset-adoption.test.mjs index cd5d9c0..478911a 100644 --- a/tests/headless/asset-adoption.test.mjs +++ b/tests/headless/asset-adoption.test.mjs @@ -68,7 +68,7 @@ test('an older Root cannot release newly adopted bytes while its unpinned manife }finally{release?.();f.storage.commit=commit;await engine.close();} }); const exportedImages=(ids=['shared'])=>new Blob([JSON.stringify({documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:ids.map((id,index)=>node({id,src:'https://untrusted.test/ignored'},'imported-'+index,'source')),tombstones:{}})]); -test('document image import authorizes original bytes and atomically publishes binding, content and progress without upload, then reopens',async()=>{const f=fixture();let engine=await f.open();try{const observed=[],stop=engine.editor.subscribe((_,root)=>{if(root.document.elements.size){const job=[...root.runtime.imports.values()][0];observed.push(root);assert.ok(root.runtime.assetBindings.get('shared'));assert.equal(job.committedSourceKeys.length,root.document.elements.size);}}),id=engine.imports.startDocument(exportedImages(),'canvas');await engine.settle();stop();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.ok(observed.length);const image=[...engine.getSnapshot().document.elements.values()][0];assert.equal(image.data.src,ref.asset.url);assert.equal(image.data.assetId,'shared');assert.equal(engine.getSnapshot().runtime.resources.size,0);assert.equal(f.calls.filter(x=>x==='authorize').length,1);assert.equal(f.roots.get('p').formatVersion,15);decodeRecovery(f.roots.get('p'),registry);await engine.close();engine=await f.open();const handle=await engine.displayResource('shared');assert.equal(await(await fetch(handle.url)).text(),'original bytes');handle.release();assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.errors.length,0);}finally{await engine.close();}}); +test('document image import authorizes original bytes and atomically publishes binding, content and progress without upload, then reopens',async()=>{const f=fixture();let engine=await f.open();try{const observed=[],stop=engine.editor.subscribe((_,root)=>{if(root.document.elements.size){const job=[...root.runtime.imports.values()][0];observed.push(root);assert.ok(root.runtime.assetBindings.get('shared'));assert.equal(job.committedSourceKeys.length,root.document.elements.size);}}),id=engine.imports.startDocument(exportedImages(),'canvas');await engine.settle();stop();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.ok(observed.length);const image=[...engine.getSnapshot().document.elements.values()][0];assert.equal(image.data.src,ref.asset.url);assert.equal(image.data.assetId,'shared');assert.equal(engine.getSnapshot().runtime.resources.size,0);assert.equal(f.calls.filter(x=>x==='authorize').length,1);assert.equal(f.roots.get('p').formatVersion,16);decodeRecovery(f.roots.get('p'),registry);await engine.close();engine=await f.open();const handle=await engine.displayResource('shared');assert.equal(await(await fetch(handle.url)).text(),'original bytes');handle.release();assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.errors.length,0);}finally{await engine.close();}}); test('document image authorization cannot publish after cancel and denied imports retain explicit failure',async()=>{const f=fixture(),engine=await f.open();let release;try{f.gate(new Promise(resolve=>release=resolve));const id=engine.imports.startDocument(exportedImages(),'canvas');for(let i=0;i<100&&!f.calls.includes('authorize');i++)await new Promise(resolve=>setTimeout(resolve,1));assert.ok(f.calls.includes('authorize'));assert.equal(engine.getSnapshot().document.elements.size,0);await engine.imports.cancel(id);release();f.gate();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.getSnapshot().runtime.assetBindings?.size??0,0);f.deny();const denied=engine.imports.startDocument(exportedImages(),'canvas');await engine.settle();assert.equal(engine.imports.getJobs().get(denied).status,'partial');assert.equal(engine.getSnapshot().runtime.assetBindings?.size??0,0);}finally{release?.();await engine.close();}}); test('one document batch can adopt multiple immutable resources with one consistent Root',async()=>{const f=fixture(),engine=await f.open();try{const id=engine.imports.startDocument(exportedImages(['shared','second']),'canvas');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().runtime.assetBindings.size,2);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(f.errors.length,0);decodeRecovery(f.roots.get('p'),registry);}finally{await engine.close();}}); test('rejected document image transaction never leaves an adopted binding or durable byte write',async()=>{const f=fixture(),engine=await f.open();try{const input=JSON.parse(await exportedImages().text());input.elements[0].geometry.width=-1;const id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'partial');assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.getSnapshot().runtime.assetBindings?.size??0,0);assert.equal(f.cache.size,0);assert.ok(!f.batches.some(batch=>batch.depot?.writes.length));}finally{await engine.close();}}); diff --git a/tests/headless/import-document-stream.test.mjs b/tests/headless/import-document-stream.test.mjs new file mode 100644 index 0000000..749e31d --- /dev/null +++ b/tests/headless/import-document-stream.test.mjs @@ -0,0 +1,50 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {openStreamingDocumentImportSource} from '../../src/lib/canvas-runtime/import-document-stream.ts'; +const signal=()=>new AbortController().signal; +const header={documentFormatVersion:1,projectId:'source',canvases:[{id:'root',parent:null}]}; +const open=(bytes,options={})=>openStreamingDocumentImportSource({bytes:bytes instanceof Blob?bytes:new Blob([typeof bytes==='string'?bytes:JSON.stringify(bytes)]),fingerprint:'immutable',signal:signal(),...options}); +const deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};}; + +test('canonical document opens and emits its first record without reading the gated payload tail; aborted scanning can retry the same ordinal',async()=>{ + const gate=deferred(),entered=deferred(),reads=[];let block=true; + class Source extends Blob{slice(start=0,end=this.size,type){reads.push([start,end]);const part=super.slice(start,end,type);if(start<32768||!block)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}} + const bytes=new Source([JSON.stringify({...header,elements:[{id:'first'},{id:'second',text:'x'.repeat(100000)}],tombstones:{}})]),reader=await open(bytes); + assert.equal(reader.parserVersion,5);assert.equal(reads.some(([start])=>start>=32768),false); + assert.equal((await reader.read(0,signal())).value.record.id,'first');assert.equal(reads.some(([start])=>start>=32768),false); + const controller=new AbortController(),pending=reader.read(1,controller.signal);await entered.promise;let timer;try{const repeated=await Promise.race([reader.read(0,signal()),new Promise((_,reject)=>timer=setTimeout(()=>reject(Error('a blocked future scan prevented re-reading a ready record')),500))]);assert.equal(repeated.value.record.id,'first');}finally{clearTimeout(timer);controller.abort();block=false;gate.resolve();}await assert.rejects(pending,/import_aborted/); + assert.equal((await reader.read(1,signal())).value.record.id,'second');assert.equal(await reader.read(2,signal()),null); +}); + +test('reopening and concurrent reads keep stable canvas-first ordinals and original root aliases',async()=>{ + const input={...header,canvases:[...header.canvases,{id:'nested',parent:{canvasId:'root',entryElementId:'frame'}}],elements:[{id:'frame'},{id:'member',ownerCanvasId:'nested',data:{text:'图,] \\"'}}],tombstones:{old:2}},reader=await open(input); + const rows=await Promise.all([0,1,2,1].map(position=>reader.read(position,signal())));assert.deepEqual(rows.map(row=>row.value.record.id),['nested','frame','member','frame']);assert.equal(rows[2].value.sourceRootId,'root');const reopened=await open(input);assert.deepEqual(await reopened.read(2,signal()),rows[2]);assert.equal(await reader.read(3,signal()),null);await assert.rejects(reader.read(5,signal()),/invalid_import_cursor/); +}); + +test('bad envelope tails preserve legal records and become one explicit terminal failure',async()=>{ + for(const suffix of [',"tombstones":{"bad":0}}','} trailing',',"elements":[]}',',"unknown":true}',',}']){ + const reader=await open(JSON.stringify({...header,elements:[{id:'first'},{id:'last'}]}).slice(0,-1)+suffix); + assert.equal((await reader.read(0,signal())).value.record.id,'first');assert.equal((await reader.read(1,signal())).value.record.id,'last');const failure=await reader.read(2,signal());assert.ok(failure.error,suffix);assert.equal(failure.nextCursor,3);assert.equal(await reader.read(3,signal()),null); + } + const trailing=await open(JSON.stringify(header).slice(0,-1)+',"elements":[{"id":"first"},]}');assert.equal((await trailing.read(0,signal())).value.record.id,'first');assert.equal((await trailing.read(1,signal())).error,'invalid_document_import'); +}); + +test('out-of-order headers retain indexed compatibility; invalid headers and cycles are rejected before publishing',async()=>{ + const reader=await open({elements:[{id:'first'}],...header});assert.equal(reader.parserVersion,5);assert.equal((await reader.read(0,signal())).value.record.id,'first'); + for(const input of [{...header,documentFormatVersion:2,elements:[]},{...header,canvases:[...header.canvases,{id:'cycle',parent:{canvasId:'cycle',entryElementId:'self'}}],elements:[]},{...header,canvases:[...header.canvases,{id:'lost',parent:{canvasId:'missing',entryElementId:'x'}}],elements:[]}])await assert.rejects(open(input),/invalid_document_import/); +}); + +test('oversized payload is scanned without decoding and later records still import',async()=>{ + const reader=await open({...header,elements:[{id:'huge',text:'x'.repeat(300000)},{id:'next'}]});assert.equal((await reader.read(0,signal())).error,'record_too_large');assert.equal((await reader.read(1,signal())).value.record.id,'next');assert.equal(await reader.read(2,signal()),null); +}); + +test('sequential records reuse the source scan chunk rather than reading a new 16 KiB window per row',async()=>{ + let chunks=0;class Counted extends Blob{slice(start=0,end=this.size,...rest){if(end-start===16384)chunks++;return super.slice(start,end,...rest);}} + const input={...header,elements:Array.from({length:1000},(_,index)=>({id:`element-${index}`,text:'small'}))},bytes=new Counted([JSON.stringify(input)]),reader=await open(bytes); + for(let index=0;index<1000;index++)assert.equal((await reader.read(index,signal())).value.record.id,`element-${index}`); + assert.equal(await reader.read(1000,signal()),null);assert.ok(chunks<=Math.ceil(bytes.size/16384)+1,`${chunks} scan reads for ${bytes.size} bytes`); +}); + +test('source identity and bytes are captured before the first await, independent of caller option mutation',async()=>{ + const options={bytes:new Blob([JSON.stringify({...header,elements:[{id:'original'}]})]),fingerprint:'original-file',signal:signal()},opening=openStreamingDocumentImportSource(options);options.bytes=new Blob([JSON.stringify({...header,elements:[{id:'replaced'}]})]);options.fingerprint='replacement-file';const reader=await opening;assert.equal(reader.fingerprint,'original-file');assert.equal((await reader.read(0,signal())).value.record.id,'original');assert.equal(await reader.read(1,signal()),null); +}); diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs index 8f41855..5323959 100644 --- a/tests/headless/imports-engine.test.mjs +++ b/tests/headless/imports-engine.test.mjs @@ -294,6 +294,33 @@ test('source read handles verify bound bytes and never publish a result after pr test('JSON array import persists original bytes and byte cursor, resumes after reopen, and preserves edits without duplicating IDs',async()=>{const f=fixture();let engine=await f.open();try{const bytes=new Blob([JSON.stringify([node('json-one'),node('json-two'),node('json-three')])]),id=engine.imports.startJsonArray(bytes,'canvas');await until(engine,s=>s.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:json-one'];assert.equal(job.parserVersion,3);assert.equal(job.source.kind,'native-json-array');assert.ok(job.cursor>0);assert.equal(f.image().formatVersion,14);decodeRecovery(f.image(),registry);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:13},registry),/unsupported_recovery_version/);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:16},registry),/unsupported_recovery_version/);engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'preserved edit'}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'preserved edit');assert.equal(f.server().elements.size,3);assert.equal(f.errors.length,0);}finally{await engine.close();}}); test('JSON array malformed tail keeps its visible prefix and an explicit failure while backend waits',async()=>{const f=fixture(),engine=await f.open(),gate=deferred();f.gate=gate;try{const id=engine.imports.startJsonArray(new Blob(['['+JSON.stringify(node('json-first'))+','+JSON.stringify(node('json-bad'))+',]']),'canvas');await until(engine,s=>s.runtime.imports.get(id).status==='partial');const job=engine.imports.getJobs().get(id);assert.equal(job.committedSourceKeys.length,1);assert.equal(job.failures.at(-1).code,'invalid_json_array');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,0);gate.resolve();await engine.settle();assert.equal(f.server().elements.size,1);decodeRecovery(f.image(),registry);}finally{gate.resolve();await engine.close();}}); -test('exported document imports into a chosen root with fresh IDs, child ownership and upper connections then reopens',async()=>{const f=fixture();let engine=await f.open();try{const frame={...node('frame').record,componentId:'frame',ownerCanvasId:'source-root'},child={...node('child').record,ownerCanvasId:'source-child'},outside={...node('outside').record,ownerCanvasId:'source-root'},edge={id:'incoming',incarnation:1,ownerCanvasId:'source-root',componentId:'connection',dataVersion:1,structure:'connection',source:{elementId:'outside'},target:{elementId:'frame'},data:{},path:null},input={documentFormatVersion:1,projectId:'source-root',canvases:[{id:'source-root',parent:null},{id:'source-child',parent:{canvasId:'source-root',entryElementId:'frame'}}],elements:[frame,child,outside,edge],tombstones:{old:2}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed',JSON.stringify(job.failures));const doc=engine.getSnapshot().document,frameId=job.stableIds['element:frame'],childId=job.stableIds['canvas:source-child'];assert.equal(doc.canvases.size,2);assert.equal(doc.canvases.get(childId).parent.entryElementId,frameId);assert.equal(doc.elements.get(job.stableIds['element:child']).ownerCanvasId,childId);assert.equal(doc.elements.get(job.stableIds['element:incoming']).ownerCanvasId,'canvas');assert.equal(doc.elements.get(job.stableIds['element:incoming']).target.elementId,frameId);assert.equal(doc.tombstones.has('old'),false);assert.equal(f.image().formatVersion,15);decodeRecovery(f.image(),registry);await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,4);assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.errors.length,0);}finally{await engine.close();}}); +test('exported document imports into a chosen root with fresh IDs, child ownership and upper connections then reopens',async()=>{const f=fixture();let engine=await f.open();try{const frame={...node('frame').record,componentId:'frame',ownerCanvasId:'source-root'},child={...node('child').record,ownerCanvasId:'source-child'},outside={...node('outside').record,ownerCanvasId:'source-root'},edge={id:'incoming',incarnation:1,ownerCanvasId:'source-root',componentId:'connection',dataVersion:1,structure:'connection',source:{elementId:'outside'},target:{elementId:'frame'},data:{},path:null},input={documentFormatVersion:1,projectId:'source-root',canvases:[{id:'source-root',parent:null},{id:'source-child',parent:{canvasId:'source-root',entryElementId:'frame'}}],elements:[frame,child,outside,edge],tombstones:{old:2}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed',JSON.stringify(job.failures));const doc=engine.getSnapshot().document,frameId=job.stableIds['element:frame'],childId=job.stableIds['canvas:source-child'];assert.equal(doc.canvases.size,2);assert.equal(doc.canvases.get(childId).parent.entryElementId,frameId);assert.equal(doc.elements.get(job.stableIds['element:child']).ownerCanvasId,childId);assert.equal(doc.elements.get(job.stableIds['element:incoming']).ownerCanvasId,'canvas');assert.equal(doc.elements.get(job.stableIds['element:incoming']).target.elementId,frameId);assert.equal(doc.tombstones.has('old'),false);assert.equal(f.image().formatVersion,16);decodeRecovery(f.image(),registry);await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,4);assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.errors.length,0);}finally{await engine.close();}}); test('document resource references report an authorization failure while independent text imports',async()=>{const f=fixture(),engine=await f.open();try{const image={...node('image').record,componentId:'image',ownerCanvasId:'source',data:{title:'image',assetId:'untrusted',src:'https://assets.test/untrusted'}},text={...node('text').record,ownerCanvasId:'source'},input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[image,text],tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(row=>row.code==='import_resource_authorization_required'));assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);}finally{await engine.close();}}); test('paused exported document resumes ordinal progress after reopen without overwriting an edited prefix',async()=>{const f=fixture();let engine=await f.open();try{const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:['d-one','d-two','d-three'].map(id=>({...node(id).record,ownerCanvasId:'source'})),tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await until(engine,s=>s.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:d-one'];assert.equal(job.cursor,1);engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'local edit'}});await engine.settle();await engine.close();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).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'local edit');assert.equal(f.errors.length,0);}finally{await engine.close();}}); + +test('streaming exported document publishes an editable prefix before the parser tail and resumes parser 5 after reopen',async()=>{ + const f=fixture(),gate=deferred(),entered=deferred();let block=true,engine=await f.open(); + class Source extends Blob{slice(start=0,end=this.size,...rest){const part=super.slice(start,end,...rest);if(end-start!==16384||start<32768||!block)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}} + try{ + const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('stream-first').record,ownerCanvasId:'source'},{...node('stream-tail').record,ownerCanvasId:'source',data:{title:'tail',text:'x'.repeat(100000)}}],tombstones:{}},id=engine.imports.startDocument(new Source([JSON.stringify(input)]),'canvas'); + await entered.promise;await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.includes('element:stream-first')); + const job=engine.imports.getJobs().get(id),first=job.stableIds['element:stream-first'];assert.equal(job.parserVersion,5);assert.equal(engine.getSnapshot().document.elements.size,1); + assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edited while tail blocked'}}).status,'applied'); + const paused=engine.imports.pause(id);block=false;gate.resolve();await paused;await engine.settle();assert.equal(f.image().formatVersion,16);decodeRecovery(f.image(),registry);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:15},registry),/unsupported_recovery_version/);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:17},registry),/unsupported_recovery_version/); + await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'edited while tail blocked');assert.equal(f.server().elements.size,2);assert.equal(f.errors.length,0); + }finally{block=false;gate.resolve();await engine.close();} +}); + +test('bad document footer produces durable partial success rather than discarding imported content',async()=>{ + const f=fixture();let engine=await f.open();try{ + const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('legal').record,ownerCanvasId:'source'}],tombstones:{broken:0}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(row=>row.code==='invalid_document_import'));assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'partial');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.errors.length,0); + }finally{await engine.close();} +}); + +test('legacy format 15 parser 4 document jobs keep their original interpreter and stable IDs when resumed',async()=>{ + const f=fixture();let engine=await f.open();try{ + const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:['legacy-one','legacy-two'].map(id=>({...node(id).record,ownerCanvasId:'source'})),tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:legacy-one'];await engine.close(); + const legacy=structuredClone(f.image());legacy.formatVersion=15;legacy.payload.runtime.imports.find(([key])=>key===id)[1].parserVersion=4;decodeRecovery(legacy,registry);f.storage.recover=async()=>structuredClone(legacy); + engine=await f.open();assert.equal(engine.imports.getJobs().get(id).parserVersion,4);await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:legacy-one'],first);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(f.image().formatVersion,15);assert.equal(f.errors.length,0); + }finally{await engine.close();} +}); diff --git a/tests/headless/recovery-encoding.test.mjs b/tests/headless/recovery-encoding.test.mjs index 284fb93..18de13c 100644 --- a/tests/headless/recovery-encoding.test.mjs +++ b/tests/headless/recovery-encoding.test.mjs @@ -24,5 +24,5 @@ test('cached subtree height cannot bypass the depth bound and omitted object fie const cycle={};cycle.self=cycle;assert.throws(()=>encodeRecoveryValue(cycle),/invalid_recovery_payload/);assert.throws(()=>encodeRecoveryValue({big:1n})); }); test('only factory-owned images are recognized; immutable-looking impostors retain defensive-copy handling',()=>{ - const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.equal(isSealedRecoveryImage(sealRecoveryImage(3,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(4,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(5,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(6,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(7,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(8,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(9,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(10,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(11,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(12,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(13,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(14,1,1,{})),true);assert.throws(()=>sealRecoveryImage(16,1,1,{}),/corrupt_recovery_image/); + const mutable={title:'before'},image=sealRecoveryImage(2,3,2,{mutable});assert.equal(isSealedRecoveryImage(image),true);assert.equal(isSealedRecoveryImage(structuredClone(image)),false);assert.equal(isSealedRecoveryImage(Object.freeze({...image})),false);mutable.title='after';assert.equal(image.payload.mutable.title,'before');assert.ok(Object.isFrozen(image.payload.mutable));assert.throws(()=>sealRecoveryImage(2,1,2,{}),/corrupt_recovery_image/);assert.equal(isSealedRecoveryImage(sealRecoveryImage(3,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(4,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(5,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(6,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(7,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(8,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(9,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(10,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(11,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(12,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(13,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(14,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(15,1,1,{})),true);assert.equal(isSealedRecoveryImage(sealRecoveryImage(16,1,1,{})),true);assert.throws(()=>sealRecoveryImage(17,1,1,{}),/corrupt_recovery_image/); }); diff --git a/tests/headless/renderer.integration.mjs b/tests/headless/renderer.integration.mjs index 1b58b65..a80a74b 100644 --- a/tests/headless/renderer.integration.mjs +++ b/tests/headless/renderer.integration.mjs @@ -440,6 +440,7 @@ try { await evaluate(`(()=>{const transfer=new DataTransfer();transfer.items.add(new File([${JSON.stringify(exported)}],'round-trip.json',{type:'application/json'}));const input=document.querySelector('input[data-canvas-import]');input.files=transfer.files;input.dispatchEvent(new Event('change',{bubbles:true}));})()`); await until(`document.querySelector('[aria-label="画布导入"]')?.textContent.includes('round-trip.json')&&[...document.querySelectorAll('[data-headless-canvas] textarea')].filter(input=>input.value==='JSON array UI text').length===2`,'exported document did not reimport'); for(let i=0;i<200&&documents.get(other).elements.size!==beforeDocumentImport+roundTripElementCount;i++)await pause(20);assert.equal(documents.get(other).elements.size,beforeDocumentImport+roundTripElementCount);await call('Page.reload');await until(`document.querySelector('[aria-label="画布导入"]')?.textContent.includes('round-trip.json')&&[...document.querySelectorAll('[data-headless-canvas] textarea')].filter(input=>input.value==='JSON array UI text').length===2`,'document round trip task/content did not recover'); + assert.equal(await evaluate(`(()=>{const job=[...window.__fixtureActualEngine.imports.getJobs().values()].find(job=>job.source?.name==='round-trip.json');return job?.parserVersion;})()`),5,'document picker did not recover the streaming interpreter'); console.log('PASS actual toolbar JSON export reimports through file picker with fresh elements and recovers after reload'); const importedAssetId=[...assets.keys()][0],uploadsBeforeDocumentImage=uploads.length,imageImportDocument={documentFormatVersion:1,projectId:'image-source',canvases:[{id:'image-source',parent:null}],elements:[{id:'image-from-file',incarnation:1,ownerCanvasId:'image-source',componentId:'image',dataVersion:1,structure:'node',geometry:{x:400,y:100,width:200,height:100,rotation:0},data:{title:'Document imported image',text:'',assetId:importedAssetId,src:'https://untrusted.test/stale'}}],tombstones:{}}; const beforeImageImport=documents.get(other).elements.size; @@ -447,6 +448,12 @@ try { for(let i=0;i<200&&![...documents.get(other).elements.values()].some(e=>e.data.title==='Document imported image');i++)await pause(20);assert.equal(documents.get(other).elements.size,beforeImageImport+1);const importedImage=[...documents.get(other).elements.values()].find(e=>e.data.title==='Document imported image');assert.equal(importedImage.data.assetId,importedAssetId);assert.notEqual(importedImage.data.src,'https://untrusted.test/stale');assert.equal(uploads.length,uploadsBeforeDocumentImage); await call('Page.reload');await until(`document.querySelector('[aria-label="画布导入"]')?.textContent.includes('image-document.json')&&!!document.querySelector('.react-flow__node[data-id="${importedImage.id}"] img')`,'authorized document image did not recover after reload');assert.equal(uploads.length,uploadsBeforeDocumentImage); console.log('PASS document image import reauthorizes existing resource, ignores stale URL and recovers without reuploading'); + await evaluate(`(()=>{const engine=window.__fixtureActualEngine,view=[...engine.sessions.getSnapshot().views.values()][0];let release;const gate=new Promise(resolve=>release=resolve);window.__releaseDocumentTail=()=>{window.__documentTailReleased=true;release();};class GatedDocument extends Blob{slice(start=0,end=this.size,...rest){const part=super.slice(start,end,...rest);if(end-start!==16384||start<32768||window.__documentTailReleased)return part;return{async arrayBuffer(){window.__documentTailWaiting=true;await gate;return part.arrayBuffer();}};}}const element=(id,text)=>({id,incarnation:1,structure:'node',ownerCanvasId:'stream-source',componentId:'text',dataVersion:1,data:{title:id,text},geometry:{x:0,y:0,width:260,height:200,rotation:0}}),input={documentFormatVersion:1,projectId:'stream-source',canvases:[{id:'stream-source',parent:null}],elements:[element('dom-prefix','Document progressive prefix'),element('dom-tail','x'.repeat(100000))],tombstones:{}};window.__streamDocumentJob=engine.imports.startDocument(new GatedDocument([JSON.stringify(input)]),view.canvasId);})()`); + await until(`(()=>{const job=window.__fixtureActualEngine.imports.getJobs().get(window.__streamDocumentJob),id=job.stableIds['element:dom-prefix'];window.__streamDocumentFirst=id;return window.__documentTailWaiting&&!window.__documentTailReleased&&!!document.querySelector('.react-flow__node[data-id="'+id+'"] textarea');})()`,'document prefix did not display while parser tail was blocked'); + await evaluate(`(()=>{const node=document.querySelector('.react-flow__node[data-id="'+window.__streamDocumentFirst+'"]');node.dispatchEvent(new MouseEvent('click',{bubbles:true}));node.querySelector('[aria-label^="单击选中"]').dispatchEvent(new MouseEvent('dblclick',{bubbles:true}));})()`);await until(`document.activeElement===document.querySelector('.react-flow__node[data-id="'+window.__streamDocumentFirst+'"] textarea')&&!document.activeElement.readOnly`,'progressive document prefix did not accept editing');await call('Input.insertText',{text:' edited before tail'});assert.equal(await evaluate(`window.__fixtureActualEngine.getSnapshot().document.elements.get(window.__streamDocumentFirst).data.text.endsWith('edited before tail')`),true);assert.equal(await evaluate('window.__documentTailReleased===true'),false);await evaluate('window.__releaseDocumentTail()'); + await until(`window.__fixtureActualEngine.imports.getJobs().get(window.__streamDocumentJob).status==='completed'`,'streaming document did not complete after releasing tail');const streamFirstId=await evaluate('window.__streamDocumentFirst');for(let i=0;i<200&&!documents.get(other).elements.get(streamFirstId)?.data.text.endsWith('edited before tail');i++)await pause(20);assert.ok(documents.get(other).elements.get(streamFirstId).data.text.endsWith('edited before tail'));assert.equal(await evaluate(`window.__fixtureActualEngine.imports.getJobs().get(window.__streamDocumentJob).committedSourceKeys.length`),2); + console.log('PASS actual document prefix displays and accepts native text edits while its parser tail is blocked, then completion preserves those edits'); +