71 lines
8.3 KiB
JavaScript
71 lines
8.3 KiB
JavaScript
/**
|
|
* 文档 parser 5 的渐进源测试：标准头部后可先输出前项，后续尾部仍可挂起、报错或尚未读取。
|
|
* 记录序号稳定，已就绪项不应被未来扫描阻塞；错误尾部只追加一条终态失败，不撤回合法前缀。
|
|
*/
|
|
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;
|
|
// 最小标准 envelope 头部，后续用例在其后追加 elements/tombstones 或打乱字段顺序。
|
|
const header={documentFormatVersion:1,projectId:'source',canvases:[{id:'root',parent:null}]};
|
|
// 统一把对象/字符串转成 Blob，也接受可观测 Blob 子类；固定 fingerprint 并允许覆盖调度/字节端口。
|
|
const open=(bytes,options={})=>openStreamingDocumentImportSource({bytes:bytes instanceof Blob?bytes:new Blob([typeof bytes==='string'?bytes:JSON.stringify(bytes)]),fingerprint:'immutable',signal:signal(),...options});
|
|
// 手动 gate 和 entered 信号精确控制尾部 slice 返回，不依赖随意延时猜测并发时序。
|
|
const deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};};
|
|
|
|
// 首项在尾部 gate 未打开时可读；未来项扫描挂起也不阻塞重读已就绪首项，取消后同序号可重新扫描成功。
|
|
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;
|
|
// 记录切片，前 32 KiB 立即返回，之后提供可暂停 arrayBuffer；用于证明打开及首项读取不访问挂起尾部。
|
|
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);
|
|
// read(1) 挂在未来切片；同时重读缓存序号 0，并用 500ms 上限捕获串行锁误阻塞。
|
|
// finally 无论重读是否成功都取消挂起调用并释放 gate，随后检查原调用拒绝且同序号可重新扫描。
|
|
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);
|
|
});
|
|
|
|
// 子 canvas 优先、元素随后，并发/重复读取和重开都保持相同序号与根 alias；跳过未定义范围的游标拒绝。
|
|
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/);
|
|
});
|
|
|
|
// 非法墓碑、垃圾、重复字段、未知尾字段及尾逗号不撤回合法元素，只追加一条可推进游标的错误后结束。
|
|
// 先读两个合法序号，再断言错误 nextCursor=3、下一次为 null，区分可恢复 I/O 拒绝和终态格式错误。
|
|
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');
|
|
});
|
|
|
|
// 非标准字段顺序沿用索引兼容路径，但仍以 parser 5 暴露；未来格式、循环分区及缺失父 canvas 在打开时拒绝。
|
|
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/);
|
|
});
|
|
|
|
// 巨型正文只扫描并报告 record_too_large，后续合法记录仍读出，到末尾正常结束。
|
|
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);
|
|
});
|
|
|
|
// 1000 个小记录复用 16 KiB 扫描块，切片数量按文件块数增长而非记录数增长。
|
|
test('sequential records reuse the source scan chunk rather than reading a new 16 KiB window per row',async()=>{
|
|
// 只统计完整 16 KiB 扫描窗口请求，检查连续记录共用扫描块，而非把正文小范围读取也计为窗口。
|
|
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`);
|
|
});
|
|
|
|
// 首次 await 前捕获 bytes/fingerprint；调用者随后替换 options 字段不能改变正在打开的源及其记录。
|
|
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);
|
|
});
|