From f11ea36b48775c47df3d4d009ab671768f3d5b3a Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Mon, 9 Nov 2026 22:47:49 +0800 Subject: [PATCH] Add proof-backed parser byte reads and preserve retryable source failures --- headless/README.md | 27 +++++++ src/lib/canvas-runtime/import-byte-reader.ts | 17 ++++ .../canvas-runtime/import-document-stream.ts | 13 ++-- src/lib/canvas-runtime/import-document.ts | 9 ++- src/lib/canvas-runtime/import-json-array.ts | 5 +- src/lib/canvas-runtime/import-read-proof.ts | 71 +++++++++++++++++ src/lib/canvas-runtime/imports.ts | 5 +- tests/headless/import-read-proof.test.mjs | 77 +++++++++++++++++++ tests/headless/imports-engine.test.mjs | 10 +++ 9 files changed, 220 insertions(+), 14 deletions(-) create mode 100644 src/lib/canvas-runtime/import-byte-reader.ts create mode 100644 src/lib/canvas-runtime/import-read-proof.ts create mode 100644 tests/headless/import-read-proof.test.mjs diff --git a/headless/README.md b/headless/README.md index d99e015..93f82d8 100644 --- a/headless/README.md +++ b/headless/README.md @@ -617,3 +617,30 @@ excluded. Target component/version changes cancel with `target_changed`; removal of the view canvas publishes fallback navigation and `canvas_changed` together. Late updates/commit cannot use the canceled preview. Failed admission does not replace the accepted baseline, and no chain, preview or gesture enters recovery. + +Internal parser byte ports and read evidence (`import-byte-reader.ts`, +`import-read-proof.ts`) provide the next integration boundary for parsing before +whole-file hashing. NDJSON, JSON array, streaming document and its indexed +fallback all accept the optional `readBytes` port; the default Blob path keeps +its original slice request bounds and verifies the returned byte count. + +The proof reader binds a file ID and byte length to SHA-256 hashes of aligned +16 KiB blocks, recording evidence only after an entire range is delivered. +A restored manifest can verify its recorded blocks before parsing resumes. +Unrecorded blocks are explicitly outside this proof: the manifest is neither a +whole-file digest nor authority to dispatch mutations. The current engine still +uses integrity version 1 and waits for its complete SHA; activating the new path +requires atomic job/proof checkpoints plus a durable verification/send barrier. +No recovery or source interpreter version changed in this unit. + +The reader owns a hash Map (at most 4096 blocks for the 64 MiB source limit), a +bounded LRU byte Map (four blocks by default), and up to four shared in-flight +block Promises/four range reads. Each range is capped at 4 MiB. Aborted waiters +release their own read but do not free occupied native reads early or cancel +another waiter. Close fences late results and clears byte buffers; checkpoints +contain no byte buffers. These bounds do not assert native Blob memory release. + +Byte I/O/capacity/integrity failures throw instead of becoming a terminal malformed +record. In particular a document reader preserves its ordinal for retry; the +engine pauses the import, retains its editable prefix, and resumes after the I/O +recovers. Ordinary malformed JSON/UTF-8 still uses the existing record diagnostics. diff --git a/src/lib/canvas-runtime/import-byte-reader.ts b/src/lib/canvas-runtime/import-byte-reader.ts new file mode 100644 index 0000000..4102fc6 --- /dev/null +++ b/src/lib/canvas-runtime/import-byte-reader.ts @@ -0,0 +1,17 @@ +import {fault} from '../canvas-core/values.ts'; + +const failures=new Set(['import_source_changed','invalid_import_byte_range','import_read_capacity','import_source_closed','import_source_read_failed','invalid_import_read_proof','import_read_proof_interpreter_unavailable']); +export const isImportByteFailure=(error:unknown)=>failures.has((error as {code?:string}|undefined)?.code??''); + +/** Optional immutable-source port. Implementations return exactly the requested bytes. */ +export type ImportByteReader={readonly byteLength:number;read(start:number,end:number,signal:AbortSignal):Promise}; +export async function readImportRange(bytes:Blob,reader:ImportByteReader|undefined,start:number,end:number,signal:AbortSignal):Promise>{ + if(signal.aborted)throw fault('import_aborted'); + if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||endbytes.size||reader&&reader.byteLength!==bytes.size)throw fault('invalid_import_byte_range'); + const limit=Math.min(end,bytes.size);let value:Uint8Array; + try{value=reader?await reader.read(start,limit,signal):new Uint8Array(await bytes.slice(start,end).arrayBuffer());} + catch(error){if(signal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;throw fault('import_source_read_failed');} + if(signal.aborted)throw fault('import_aborted'); + if(!(value instanceof Uint8Array)||value.byteLength!==limit-start)throw fault('import_source_changed'); + return reader?new Uint8Array(value):value as Uint8Array; +} diff --git a/src/lib/canvas-runtime/import-document-stream.ts b/src/lib/canvas-runtime/import-document-stream.ts index 6ca62d6..81a3ef1 100644 --- a/src/lib/canvas-runtime/import-document-stream.ts +++ b/src/lib/canvas-runtime/import-document-stream.ts @@ -1,3 +1,4 @@ +import {readImportRange,isImportByteFailure,type ImportByteReader} from './import-byte-reader.ts'; import type {ImportLine,ImportSource} from './imports.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; import {openDocumentImportSource} from './import-document.ts'; @@ -6,7 +7,7 @@ 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}; +type Options=ImportScheduling&{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;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; @@ -17,7 +18,7 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta 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();} + if(cursor=cache.start+cache.bytes.length){const loaded=await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal);cache.start=cursor;cache.bytes=loaded;check();} return cache.bytes[cursor-cache.start]; } function step(){cursor++;return budget.step();} @@ -40,8 +41,8 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta } 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');} + const data=await readImportRange(bytes,options.readBytes,range.start,range.end,signal);let text:string; + try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}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}; @@ -124,7 +125,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise=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';} + try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;failure=(error as {code?:string}).code??'invalid_document_import';phase='done';} } const count=children.length+ranges.length; if(position>=count){ @@ -137,7 +138,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise{ +export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;signal:AbortSignal;maxRecordBytes?:number}&ImportScheduling):Promise{ 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'); let cursor=0,cache={start:-1,bytes:new Uint8Array()};const budget=createImportBudget(signal,options); const space=(b:number)=>[9,10,13,32].includes(b); - async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.start+cache.bytes.length){cache={start:cursor,bytes:new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];} + async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.start+cache.bytes.length){cache={start:cursor,bytes:await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];} function step(){cursor++;return budget.step();} async function spaces(){while(space(await peek()))await step();} async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();} @@ -19,7 +20,7 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s const ended=!quoted&&stack.length===0&&(b===34||b===93||b===125);await step();if(ended)break; }if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor}; } - async function decode(range:Range,limit=max){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{throw fault('invalid_utf8');}try{const result=JSON.parse(text);if(!validJson(result))throw Error();return result;}catch{throw fault('invalid_json');}} + async function decode(range:Range,limit=max,readSignal=signal){if(range.end-range.start>limit)throw fault('record_too_large');const data=await readImportRange(bytes,options.readBytes,range.start,range.end,readSignal);let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}catch{throw fault('invalid_utf8');}try{const result=JSON.parse(text);if(!validJson(result))throw Error();return result;}catch{throw fault('invalid_json');}} const fields=new Map(),elements:Range[]=[]; await take(123);await spaces();if(await peek()===125)throw fault('invalid_document_import'); while(true){const key=await decode(await value(),1024);if(typeof key!=='string'||fields.has(key))throw fault('invalid_document_import');await take(58); @@ -38,6 +39,6 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s return Object.freeze({fingerprint:options.fingerprint,parserVersion:4,async read(position:number,readSignal:AbortSignal):Promise{ if(readSignal.aborted)throw fault('import_aborted');if(!Number.isSafeInteger(position)||position<0||position>count)throw fault('invalid_import_cursor');if(position===count)return null; const base={cursor:position,nextCursor:position+1,key:`record:${position}`};let record:any,kind:string; - try{if(positionmax)throw fault('invalid_import_source'); let cache={offset:-1,value:new Uint8Array()};const whitespace=(b:number)=>[9,10,13,32].includes(b); @@ -12,7 +13,7 @@ export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:stri if(start>0&&start===bytes.size){if(signal.aborted)throw fault('import_aborted');return null;} if(bytes.size===0)throw fault('invalid_json_array'); let cursor=start;const budget=createImportBudget(signal,options); - async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.offset+cache.value.length){cache={offset:cursor,value:new Uint8Array(await bytes.slice(cursor,cursor+chunk).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];} + async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor=cache.offset+cache.value.length){cache={offset:cursor,value:await readImportRange(bytes,options.readBytes,cursor,cursor+chunk,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];} function advance(){cursor++;return budget.step();} async function spaces(){while(whitespace(await peek()))await advance();} const failure=(error:string):ImportLine=>({cursor:start,nextCursor:bytes.size,key:`byte:${start}`,error}); diff --git a/src/lib/canvas-runtime/import-read-proof.ts b/src/lib/canvas-runtime/import-read-proof.ts new file mode 100644 index 0000000..dcf0d1c --- /dev/null +++ b/src/lib/canvas-runtime/import-read-proof.ts @@ -0,0 +1,71 @@ +import {fault,freeze,validJson} from '../canvas-core/values.ts'; +import {createSha256} from './sha256.ts'; +import {createImportBudget,type ImportScheduling} from './import-budget.ts'; +import type {ImportByteReader} from './import-byte-reader.ts'; + +export const IMPORT_PROOF_BLOCK_BYTES=16384; +const MAX_SOURCE_BYTES=64*1024*1024; +export type ImportReadProof={readonly version:1;readonly fileId:string;readonly byteLength:number;readonly blockBytes:16384;readonly blocks:readonly {readonly index:number;readonly sha256:string}[]}; +/** This binds only read blocks. It is not a whole-source digest or permission to send mutations. */ +export function validateImportReadProof(value:unknown,fileId:string,byteLength:number):ImportReadProof{ + if(typeof fileId!=='string'||!fileId||!Number.isSafeInteger(byteLength)||byteLength<0||byteLength>MAX_SOURCE_BYTES)throw fault('invalid_import_read_proof'); + const proof=value as ImportReadProof; + if(!proof||!validJson(proof)||proof.version!==1)throw fault('import_read_proof_interpreter_unavailable'); + if(proof.fileId!==fileId||proof.byteLength!==byteLength||proof.blockBytes!==IMPORT_PROOF_BLOCK_BYTES||!Array.isArray(proof.blocks)||Object.keys(proof).some(key=>!['version','fileId','byteLength','blockBytes','blocks'].includes(key)))throw fault('invalid_import_read_proof'); + let last=-1;const count=Math.ceil(byteLength/IMPORT_PROOF_BLOCK_BYTES); + if(proof.blocks.length>count)throw fault('invalid_import_read_proof'); + for(const block of proof.blocks){if(!block||!Number.isSafeInteger(block.index)||block.index<=last||block.index>=count||typeof block.sha256!=='string'||!/^[a-f0-9]{64}$/.test(block.sha256)||Object.keys(block).some(key=>!['index','sha256'].includes(key)))throw fault('invalid_import_read_proof');last=block.index;} + return freeze({version:1 as const,fileId,byteLength,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:proof.blocks.map(block=>({...block}))}); +} + +/** Bounded, seekable verified reads for parsers; checkpoints contain hashes, never byte buffers. + * Restored evidence must be checked with verify() before a runner resumes business changes. */ +export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:ImportReadProof;maxResidentBlocks?:number;maxReadBytes?:number}&ImportScheduling){ + const options=Object.freeze({...input}); + const {bytes,fileId}=options,residentLimit=options.maxResidentBlocks??4,readLimit=options.maxReadBytes??4*1024*1024; + if(!(bytes instanceof Blob)||!fileId||typeof fileId!=='string'||bytes.size>MAX_SOURCE_BYTES||!Number.isSafeInteger(residentLimit)||residentLimit<1||residentLimit>64||!Number.isSafeInteger(readLimit)||readLimit<1||readLimit>4*1024*1024)throw fault('invalid_import_read_source'); + const initial:ImportReadProof=options.proof?validateImportReadProof(options.proof,fileId,bytes.size):freeze({version:1 as const,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[]}); + const hashes=new Map(initial.blocks.map(block=>[block.index,block.sha256])); + type Block={index:number;sha256:string;data:Uint8Array}; + const resident=new Map(),pending=new Map>(),lifecycle=new AbortController(); + let closed=false,activeReads=0,proof:ImportReadProof=initial; + const check=(signal:AbortSignal)=>{if(closed)throw fault('import_source_closed');if(signal.aborted)throw fault('import_aborted');}; + function wait(task:Promise,signal:AbortSignal):Promise{ + check(signal);return new Promise((resolve,reject)=>{const abort=()=>{try{check(signal);}catch(error){reject(error);}};signal.addEventListener('abort',abort,{once:true});task.then(value=>{signal.removeEventListener('abort',abort);try{check(signal);resolve(value);}catch(error){reject(error);}},error=>{signal.removeEventListener('abort',abort);reject(error);});}); + } + function block(index:number):Promise{ + const cached=resident.get(index);if(cached){resident.delete(index);resident.set(index,cached);return Promise.resolve(cached);} + const active=pending.get(index);if(active)return active; + if(pending.size>=4)throw fault('import_read_capacity'); + const task=(async()=>{ + const start=index*IMPORT_PROOF_BLOCK_BYTES,end=Math.min(start+IMPORT_PROOF_BLOCK_BYTES,bytes.size),data=new Uint8Array(await bytes.slice(start,end).arrayBuffer()); + if(closed)throw fault('import_source_closed');if(data.byteLength!==end-start)throw fault('import_source_changed'); + const hash=createSha256();hash.update(data);const sha256=hash.hex(),expected=hashes.get(index); + if(expected!==undefined&&expected!==sha256)throw fault('import_source_changed'); + const value={index,sha256,data};resident.set(index,value);while(resident.size>residentLimit)resident.delete(resident.keys().next().value!);return value; + })(); + pending.set(index,task);void task.finally(()=>{if(pending.get(index)===task)pending.delete(index);}).catch(()=>{});return task; + } + const port:ImportByteReader=Object.freeze({byteLength:bytes.size,async read(start:number,end:number,signal:AbortSignal){ + const combined=AbortSignal.any([signal,lifecycle.signal]);check(combined); + if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||endbytes.size||end-start>readLimit)throw fault('invalid_import_byte_range'); + if(start===end)return new Uint8Array(); + if(activeReads>=4||pending.size>=4&&!resident.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES))&&!pending.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES)))throw fault('import_read_capacity'); + activeReads++;try{ + const output=new Uint8Array(end-start),used:Pick[]=[],budget=createImportBudget(combined,options); + for(let index=Math.floor(start/IMPORT_PROOF_BLOCK_BYTES);index*IMPORT_PROOF_BLOCK_BYTESa-b).map(([index,sha256])=>({index,sha256}))}); + return output; + }finally{activeReads--;} + }}); + return Object.freeze({port,getProof:()=>proof, + async verify(signal:AbortSignal){const combined=AbortSignal.any([signal,lifecycle.signal]),budget=createImportBudget(combined,options);check(combined);for(const index of [...hashes.keys()]){await wait(block(index),combined);await budget.step(128);}check(combined);}, + close(){if(closed)return;closed=true;resident.clear();lifecycle.abort();}, + stats:()=>Object.freeze({residentBlocks:resident.size,activeReads,pendingBlocks:pending.size,proofBlocks:hashes.size,closed}), + }); +} diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts index 6ad07e1..0ce531f 100644 --- a/src/lib/canvas-runtime/imports.ts +++ b/src/lib/canvas-runtime/imports.ts @@ -1,3 +1,4 @@ +import {readImportRange,type ImportByteReader} from './import-byte-reader.ts'; import {createImportBudget,type ImportScheduling} from './import-budget.ts'; import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts'; import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts'; @@ -67,7 +68,7 @@ export function validateImportJob(job: ImportJob): void { const yieldTask=()=>new Promise(resolve=>setTimeout(resolve,0)); /** Seekable bounded NDJSON parser. Fingerprint must identify the host's immutable, durable byte source. */ -export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource { +export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource { const max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16*1024; if(!(options.bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source'); const sourceBytes=options.bytes; @@ -79,7 +80,7 @@ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string; const parts:Uint8Array[]=[],start=cursor,budget=createImportBudget(signal,options);let total=0,oversized=false; while(cursor=cache.offset+cache.bytes.length)cache={offset:cursor,bytes:new Uint8Array(await sourceBytes.slice(cursor,cursor+chunk).arrayBuffer())}; + if(cursor=cache.offset+cache.bytes.length)cache={offset:cursor,bytes:await readImportRange(sourceBytes,options.readBytes,cursor,cursor+chunk,signal)}; if(signal.aborted)throw fault('import_aborted'); const begin=cursor-cache.offset,newline=cache.bytes.indexOf(10,begin),end=newline<0?cache.bytes.length:newline,part=cache.bytes.subarray(begin,end);total+=part.length; if(total>max){oversized=true;parts.length=0;}else if(!oversized)parts.push(part); diff --git a/tests/headless/import-read-proof.test.mjs b/tests/headless/import-read-proof.test.mjs new file mode 100644 index 0000000..e7b0d92 --- /dev/null +++ b/tests/headless/import-read-proof.test.mjs @@ -0,0 +1,77 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createImportReadProof,validateImportReadProof,IMPORT_PROOF_BLOCK_BYTES as B} from '../../src/lib/canvas-runtime/import-read-proof.ts'; +import {readImportRange} from '../../src/lib/canvas-runtime/import-byte-reader.ts'; +import {openStreamingDocumentImportSource} from '../../src/lib/canvas-runtime/import-document-stream.ts'; +import {createNdjsonImportSource} from '../../src/lib/canvas-runtime/imports.ts'; +import {createJsonArrayImportSource} from '../../src/lib/canvas-runtime/import-json-array.ts'; +import {fault} from '../../src/lib/canvas-core/values.ts'; +const signal=()=>new AbortController().signal,deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};}; +const sha=async bytes=>Buffer.from(await crypto.subtle.digest('SHA-256',bytes)).toString('hex'); +const tick=()=>new Promise(resolve=>setTimeout(resolve,0)); + +test('range reads return exact owned bytes and checkpoint only the aligned blocks they consumed',async()=>{ + const data=Uint8Array.from({length:B*3+19},(_,i)=>i%251),bytes=new Blob([data]),reader=createImportReadProof({bytes,fileId:'f',maxResidentBlocks:1}); + const value=await reader.port.read(B-5,B+7,signal());assert.deepEqual(value,data.slice(B-5,B+7));value.fill(0);assert.deepEqual(await reader.port.read(B-5,B+7,signal()),data.slice(B-5,B+7)); + const proof=reader.getProof();assert.deepEqual(proof.blocks,[{index:0,sha256:await sha(data.slice(0,B))},{index:1,sha256:await sha(data.slice(B,B*2))}]);assert.strictEqual(reader.getProof(),proof);assert.ok(Object.isFrozen(proof)&&Object.isFrozen(proof.blocks));assert.equal(reader.stats().residentBlocks,1); + assert.deepEqual(validateImportReadProof(JSON.parse(JSON.stringify(proof)),'f',bytes.size),proof);reader.close();assert.equal(reader.stats().residentBlocks,0); +}); + +test('parser 5 emits its first proof-backed record while the unread tail is unavailable',async()=>{ + const reads=[];class Gated extends Blob{slice(start=0,end=this.size){reads.push([start,end]);if(start>=B)throw Error('tail unavailable');return super.slice(start,end);}} + const bytes=new Gated([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'first'},{id:'later',text:'x'.repeat(B*3)}]})]),reader=createImportReadProof({bytes,fileId:'f'}); + const source=await openStreamingDocumentImportSource({bytes,readBytes:reader.port,fingerprint:'f',signal:signal()});assert.equal((await source.read(0,signal())).value.record.id,'first');assert.deepEqual(reader.getProof().blocks.map(block=>block.index),[0]);assert.equal(reads.some(([start])=>start>=B),false);reader.close(); +}); + +test('NDJSON, array and out-of-order document fallback all use the byte reader',async()=>{ + const cases=[['{"id":"a"}\n',o=>createNdjsonImportSource(o)],['[{"id":"a"}]',o=>createJsonArrayImportSource(o)],[JSON.stringify({elements:[{id:'a'}],documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}]}),o=>openStreamingDocumentImportSource({...o,signal:signal()})]]; + for(const[text,open]of cases){const bytes=new Blob([text]),reader=createImportReadProof({bytes,fileId:'f'}),source=await open({bytes,readBytes:reader.port,fingerprint:'f'}),row=await source.read(0,signal());assert.equal((row.value.record??row.value).id,'a');assert.equal(reader.getProof().blocks.length,1);reader.close();} +}); + +test('restored proof rejects changed consumed bytes and does not claim to authenticate unread blocks',async()=>{ + const data=new Uint8Array(B*3).fill(7),first=createImportReadProof({bytes:new Blob([data]),fileId:'f'});await first.port.read(0,1,signal());await first.port.read(B*2,B*2+1,signal());const proof=first.getProof();first.close(); + const changed=data.slice();changed[B*2]=8;const bad=createImportReadProof({bytes:new Blob([changed]),fileId:'f',proof});await assert.rejects(bad.verify(signal()),/import_source_changed/);assert.strictEqual(bad.getProof().blocks.length,2);bad.close(); + changed[B*2]=7;changed[B]=8;const unseen=createImportReadProof({bytes:new Blob([changed]),fileId:'f',proof});await unseen.verify(signal());assert.deepEqual(unseen.getProof(),proof);assert.equal((await unseen.port.read(B,B+1,signal()))[0],8);assert.equal(unseen.getProof().blocks.length,3);unseen.close(); +}); + +test('eviction does not discard the expected hash; changed rereads fail without poisoning a retry',async()=>{ + const data=new Uint8Array(B*2).fill(4);let altered=false;class Mutable extends Blob{slice(start,end){const part=data.slice(start,end);if(altered&&start===0)part[0]=9;return new Blob([part]);}} + const reader=createImportReadProof({bytes:new Mutable([data]),fileId:'f',maxResidentBlocks:1});await reader.port.read(0,1,signal());await reader.port.read(B,B+1,signal());const proof=reader.getProof();altered=true;await assert.rejects(reader.port.read(0,1,signal()),/import_source_changed/);assert.strictEqual(reader.getProof(),proof);altered=false;assert.equal((await reader.port.read(0,1,signal()))[0],4);reader.close(); +}); + +test('one canceled waiter cannot cancel a shared block or advance its checkpoint',async()=>{ + const gate=deferred();let reads=0;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}} + const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B)]),fileId:'f'}),controller=new AbortController(),a=reader.port.read(0,1,controller.signal),b=reader.port.read(1,2,signal());controller.abort();await assert.rejects(a,/import_aborted/);assert.equal(reader.getProof().blocks.length,0);assert.equal(reader.stats().pendingBlocks,1);gate.resolve();await b;assert.equal(reads,1);assert.equal(reader.getProof().blocks.length,1);reader.close(); +}); + +test('canceled multi-block reads and close never publish partial read evidence',async()=>{ + const gate=deferred(),entered=deferred();class Slow extends Blob{slice(start,end){const part=super.slice(start,end);if(start===0)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}} + const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B*2)]),fileId:'f'}),controller=new AbortController(),read=reader.port.read(0,B+1,controller.signal);await entered.promise;controller.abort();await assert.rejects(read,/import_aborted/);assert.equal(reader.getProof().blocks.length,0); + const late=reader.port.read(B,B+1,signal());reader.close();reader.close();await assert.rejects(late,/import_source_closed/);assert.equal(reader.stats().pendingBlocks,1);gate.resolve();await tick();assert.equal(reader.stats().pendingBlocks,0);assert.equal(reader.stats().residentBlocks,0);assert.equal(reader.getProof().blocks.length,0);await assert.rejects(reader.port.read(0,0,signal()),/import_source_closed/); +}); + +test('bounded admission retains occupied native reads until settlement instead of opening more slots on abort',async()=>{ + const gate=deferred();let reads=0;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}} + const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B*5)]),fileId:'f',maxResidentBlocks:2}),controllers=Array.from({length:4},()=>new AbortController()),pending=controllers.map((controller,i)=>reader.port.read(B*i,B*i+1,controller.signal)); + await assert.rejects(reader.port.read(B*4,B*4+1,signal()),/import_read_capacity/);assert.equal((await reader.port.read(B*4,B*4,signal())).byteLength,0);controllers.forEach(controller=>controller.abort());await Promise.all(pending.map(task=>assert.rejects(task,/import_aborted/)));assert.equal(reader.stats().activeReads,0);assert.equal(reader.stats().pendingBlocks,4);await assert.rejects(reader.port.read(B*4,B*4+1,signal()),/import_read_capacity/);assert.equal(reads,4);gate.resolve();await tick();assert.ok(reader.stats().residentBlocks<=2);assert.equal(reader.getProof().blocks.length,0);reader.close(); +}); + +test('zero-length reads do no I/O and invalid proof, identity and ranges fail closed',async()=>{ + let reads=0;class Count extends Blob{slice(...args){reads++;return super.slice(...args);}} + const bytes=new Count(['abc']),reader=createImportReadProof({bytes,fileId:'f'});await reader.port.read(1,1,signal());await reader.port.read(3,3,signal());assert.equal(reads,0);assert.equal(reader.getProof().blocks.length,0);const proof=reader.getProof(); + for(const bad of [{...proof,version:2},{...proof,fileId:'other'},{...proof,byteLength:4},{...proof,blockBytes:1},{...proof,extra:true},{...proof,blocks:[{index:1,sha256:'a'.repeat(64)}]},{...proof,blocks:[{index:0,sha256:'A'.repeat(64)}]},{...proof,blocks:[{index:0,sha256:'a'.repeat(64)},{index:0,sha256:'a'.repeat(64)}]}])assert.throws(()=>createImportReadProof({bytes,fileId:'f',proof:bad}),/import_read_proof/); + for(const[start,end]of [[-1,1],[2,1],[0,4],[.5,1]])await assert.rejects(reader.port.read(start,end,signal()),/invalid_import_byte_range/);reader.close(); +}); + +test('byte ports cannot return short bytes, wrong source length, or disguise integrity errors as UTF-8 failures',async()=>{ + const bytes=new Blob(['abcdef']);await assert.rejects(readImportRange(bytes,{byteLength:6,async read(){return new Uint8Array(1);}},0,2,signal()),/import_source_changed/);await assert.rejects(readImportRange(bytes,{byteLength:7,async read(){throw Error('must not run');}},0,2,signal()),/invalid_import_byte_range/); + const document=new Blob([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'a'}]})]);let rejectDecode=false;const port={byteLength:document.size,async read(start,end){if(rejectDecode&&end-start<100)throw fault('import_source_changed');return new Uint8Array(await document.slice(start,end).arrayBuffer());}}; + const source=await openStreamingDocumentImportSource({bytes:document,readBytes:port,fingerprint:'f',signal:signal()});rejectDecode=true;await assert.rejects(source.read(0,signal()),/import_source_changed/);rejectDecode=false;assert.equal((await source.read(0,signal())).value.record.id,'a'); +}); + + +test('transient byte I/O failure does not advance a streaming ordinal into a terminal malformed-tail error',async()=>{ + const bytes=new Blob([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'a'},{id:'b',text:'x'.repeat(B*2)}]})]);let fail=true; + const readBytes={byteLength:bytes.size,async read(start,end){if(fail&&start>=B)throw Error('offline');return new Uint8Array(await bytes.slice(start,end).arrayBuffer());}}; + const source=await openStreamingDocumentImportSource({bytes,readBytes,fingerprint:'f',signal:signal()});assert.equal((await source.read(0,signal())).value.record.id,'a');await assert.rejects(source.read(1,signal()),/import_source_read_failed/);fail=false;assert.equal((await source.read(1,signal())).value.record.id,'b');assert.equal(await source.read(2,signal()),null); +}); diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs index 5323959..4602652 100644 --- a/tests/headless/imports-engine.test.mjs +++ b/tests/headless/imports-engine.test.mjs @@ -324,3 +324,13 @@ test('legacy format 15 parser 4 document jobs keep their original interpreter an 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();} }); + +test('temporary document byte I/O pauses without consuming a bad-tail ordinal and resumes with edited prefix intact',async()=>{ + const f=fixture(),engine=await f.open();let fail=true; + class Flaky extends Blob{slice(start=0,end=this.size,...rest){if(fail&&end-start===16384&&start>=16384)throw Error('temporary source I/O');return super.slice(start,end,...rest);}} + try{ + const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('io-first').record,ownerCanvasId:'source'},{...node('io-tail').record,ownerCanvasId:'source',data:{title:'tail',text:'x'.repeat(100000)}}]},id=engine.imports.startDocument(new Flaky([JSON.stringify(input)]),'canvas'); + await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const job=engine.imports.getJobs().get(id),first=job.stableIds['element:io-first'];assert.equal(job.error,'import_source_read_failed');assert.equal(job.eof,false);assert.equal(job.cursor,1);assert.equal(job.failures.length,0);assert.equal(job.committedSourceKeys.length,1); + assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edited prefix'}}).status,'applied');fail=false;await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.failures.length,0);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'edited prefix');assert.equal(f.server().elements.get(first).data.title,'edited prefix');assert.ok(f.errors.some(error=>error.code==='import_source_read_failed'));decodeRecovery(f.image(),registry); + }finally{fail=false;await engine.close();} +});