96 lines
14 KiB
JavaScript
96 lines
14 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';
|
|
// 独立读取信号和可控 Promise，分别控制取消者、其他等待者及底层原生读取返回。
|
|
const signal=()=>new AbortController().signal,deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};};
|
|
// 使用 WebCrypto 独立计算预期块摘要，对照证明中记录的 SHA-256。
|
|
const sha=async bytes=>Buffer.from(await crypto.subtle.digest('SHA-256',bytes)).toString('hex');
|
|
// 让出一次真实 timer，使不可取消的底层读在 gate 释放后有机会完成清理。
|
|
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);
|
|
});
|
|
|
|
// 尾部块读取不可用时，parser 5 仍返回首项；证明仅含第一块，不需要先认证整份文件。
|
|
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();
|
|
});
|
|
|
|
// NDJSON、数组和乱序文档兼容路径均通过注入的 readBytes 端口读取，并形成实际消费证明。
|
|
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();}
|
|
});
|
|
|
|
// 恢复证明只重验已消费的第 0/2 块，变化即拒绝；未消费第 1 块变化不误报已验证，首次读取时才纳入新证明。
|
|
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();
|
|
});
|
|
|
|
// 原始字节淘汰不丢预期 hash；已消费块后来篡改时拒绝且不污染旧证明，恢复原字节后可重试。
|
|
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;/* 人为违反 Blob 字节稳定性；淘汰缓存后重读首块会见到改变，用于校验保存的预期摘要。 */ 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();
|
|
});
|
|
|
|
// 两个 waiter 共享同一原生块读取；取消一个不取消另一个，也不提前推进 proof，剩余 waiter 成功后才提交块证据。
|
|
test('one canceled waiter cannot cancel a shared block or advance its checkpoint',async()=>{
|
|
const gate=deferred();let reads=0;/* 把底层块读取停在 gate，模拟调用者已取消但不可取消的原生 I/O 尚未返回。 */ 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();
|
|
});
|
|
|
|
// 跨块范围取消后不发布已读首块的部分证据；close 幂等拒绝新/挂起调用，底层读真正退出后才清空 pending/resident。
|
|
test('canceled multi-block reads and close never publish partial read evidence',async()=>{
|
|
const gate=deferred(),entered=deferred();/* 把底层块读取停在 gate，模拟调用者已取消但不可取消的原生 I/O 尚未返回。 */ 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/);
|
|
});
|
|
|
|
// 四个不可取消的原生读取已占容量，即使调用者全 abort 也不能新开第五个；底层退出后才释放，取消读取不进证明。
|
|
test('bounded admission retains occupied native reads until settlement instead of opening more slots on abort',async()=>{
|
|
const gate=deferred();let reads=0;/* 把底层块读取停在 gate，模拟调用者已取消但不可取消的原生 I/O 尚未返回。 */ 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;/* 对 slice 计数，证明零长度读取不触发底层 I/O。 */ 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();
|
|
});
|
|
|
|
// 字节端口返回过短结果或源长度不符时拒绝；完整性错误不能被包装为 UTF-8 格式错误，修复后同记录可重试。
|
|
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');
|
|
});
|
|
|
|
|
|
// 未来块临时 I/O 失败暂停该次读取，不把序号推进成终态坏尾部；恢复网络后重读同序号获得 b，再正常 EOF。
|
|
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);
|
|
});
|