78 lines
11 KiB
JavaScript
78 lines
11 KiB
JavaScript
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);
|
|
});
|