docs(tests): explain streaming import cursors and read-proof lifecycle

This commit is contained in:
2026-11-12 17:25:23 +08:00
parent 225170b024
commit 7ed36b1ae0
7 changed files with 15 additions and 2 deletions
+1
View File
@@ -6,6 +6,7 @@ import assert from 'node:assert/strict';
import {createImportBatches} from '../../src/lib/canvas-runtime/import-batches.ts';
// 从单项起步,在固定成本占主导且每项便宜时逐渐增至 32 上限;目标耗时仍为 6,不通过抬高预算换吞吐。
test('fixed publication cost can grow beyond singleton batches without raising the time budget',()=>{
// record 接收合成耗时而非测量真实发布;4 是固定成本,.05 是每项成本,只验证反馈算法的增长轨迹。
const batches=createImportBatches(6,32);assert.equal(batches.observed,false);assert.equal(batches.limit,1);const sizes=[];
for(let i=0;i<20;i++){const n=batches.limit;sizes.push(n);batches.record(n,4+n*.05);}
assert.equal(batches.observed,true);assert.equal(batches.limit,32);assert.ok(sizes.includes(2));assert.ok(sizes.every(n=>n<=32));
@@ -22,6 +22,8 @@ test('canonical document opens and emits its first record without reading the ga
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);
});
@@ -33,6 +35,7 @@ test('reopening and concurrent reads keep stable canvas-first ordinals and origi
});
// 非法墓碑、垃圾、重复字段、未知尾字段及尾逗号不撤回合法元素,只追加一条可推进游标的错误后结束。
// 先读两个合法序号,再断言错误 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);
+3 -1
View File
@@ -1,6 +1,7 @@
/**
* 完整文档 envelope 的索引导入源测试(parser 4)。头部/拓扑先验证,元素正文按稳定序号读取,原根 canvas 映射到目标根。
* 超大单条作为记录失败,不阻止后续合法项;文件尾结构损坏则拒绝打开完整索引源。
* parser 4 先扫描建立索引再开放 read;先出现 canvas 记录是输出顺序,不证明它在扫描尾部前已返回。
*/
import {test} from 'node:test';import assert from 'node:assert/strict';import {openDocumentImportSource} from '../../src/lib/canvas-runtime/import-document.ts';
// 独立信号和 Blob 包装工厂;基线包含两层 canvas、Unicode/分隔符正文及墓碑,检查封装而非简单数组。
@@ -18,5 +19,6 @@ import {createImportJob} from '../../src/lib/canvas-runtime/imports.ts';
import {createRegistry,createDocument} from '../../src/lib/canvas-core/index.ts';
// 原根 alias 同时映射 owner 与组件声明的 canvasRef;不能为原根再 reserve 一个多余的新 canvas ID。
test('document root aliases also remap component-declared canvas references without reserving a duplicate root',()=>{const registry=createRegistry([{id:'ref',dataVersion:1,structure:'node',features:[],validate:()=>true,copy:{references:[{path:['canvasRef'],target:'canvas',external:'reject'}],omit:[]}}]),target=createDocument('p',[{id:'target',parent:null}],[]),job=createImportJob({id:'job',projectId:'p',parserVersion:4,sourceFingerprint:'file',source:{kind:'native-document',fileId:'file',byteLength:100,targetCanvasId:'target'}}),reserved=[],record={id:'a',incarnation:1,ownerCanvasId:'source',componentId:'ref',dataVersion:1,structure:'node',geometry:{x:0,y:0,width:10,height:10,rotation:0},data:{canvasRef:'source'}},unit=prepareNativeImport({cursor:0,nextCursor:1,key:'record:0',value:{kind:'element',key:'a',record,sourceRootId:'source'}},job,key=>{reserved.push(key);return'new-'+key;},target,registry);assert.equal(unit.changes[0].after.data.canvasRef,'target');assert.equal(unit.changes[0].after.ownerCanvasId,'target');assert.ok(!reserved.includes('canvas:source'));});
// 小于固定字节上限的扫描也要按实际耗时让出,yield 内取消后以 import_aborted 拒绝。
// 小于固定字节上限的扫描也要按耗时让出,yield 内取消后以 import_aborted 拒绝。
// now 每次读取人工增加 10,避免依赖机器快慢;只检查预算分支,不测实际主线程帧耗时。
test('document preparation yields by elapsed time even for files below the byte ceiling',async()=>{const controller=new AbortController();let clock=0,yields=0;await assert.rejects(openDocumentImportSource({bytes:source({...document,elements:[{id:'slow',text:'x'.repeat(2048)}]}),fingerprint:'slow',signal:controller.signal,now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}}),/import_aborted/);assert.equal(yields,1);});
@@ -15,6 +15,7 @@ test('source retention marks are immutable metadata with monotonic job generatio
const incomplete=prepareImportFileRetention(next,next,()=>({complete:false,pins:new Map()}),new Set());assert.equal(incomplete.imports.get('job').source.retirement,undefined);
});
// 允许源文件向 previousFiles 归档,但拒绝丢失归属、与资源重名、重复 job 共享身份或回退已 purge 元数据;非法版本/时间也拒绝。
// coverage 返回的完整/无 pin 只是扫描替身结论;这些断言只检查元数据演进,不调用真实文件仓库清理。
test('source ownership, archive direction and completed purge metadata cannot be silently rewritten',()=>{
const archived={...job,source:{...job.source,fileId:'next',previousFiles:[{fileId:'source',byteLength:10}]}};assert.equal(importFiles(new Map([['job',archived]])).size,2);assert.equal(prepareImportFileRetention(runtime,{...runtime,imports:new Map([['job',archived]])},coverage,new Set()).imports.get('job'),archived);
assert.throws(()=>prepareImportFileRetention(runtime,{...runtime,imports:new Map()},coverage,new Set()),/import_file_identity_conflict/);assert.throws(()=>prepareImportFileRetention(runtime,{...runtime,resources:new Map([['source',{}]])},coverage,new Set()),/import_file_identity_conflict/);
+2 -1
View File
@@ -5,7 +5,8 @@ import {test} from 'node:test';
import assert from 'node:assert/strict';
import {importSourceDigest} from '../../src/lib/canvas-runtime/import-integrity.ts';
import {createImportJob,validateImportJob} from '../../src/lib/canvas-runtime/imports.ts';
// 标准 abc 向量和不同输入区分完整源摘要;预先取消及异步读取返回前取消都拒绝,不接受迟到结果。
// 标准 abc 向量和不同输入区分完整源摘要;预先取消及调用后立即取消都必须拒绝。
// 下方整 Blob gate 不拦截当前的 slice 读取,不能用它证明读取被挂起;真正挂起块的取消见 sha256.test.mjs。
test('source SHA256 binds complete bytes and cancellation fences both sides of asynchronous reading',async()=>{
assert.equal(await importSourceDigest(new Blob(['abc'])),'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad');assert.notEqual(await importSourceDigest(new Blob(['abc'])),await importSourceDigest(new Blob(['abd'])));
const controller=new AbortController();controller.abort();await assert.rejects(importSourceDigest(new Blob(['abc']),controller.signal),/import_aborted/);let release;const gate=new Promise(resolve=>{release=resolve;}),later=new AbortController(),bytes=new class extends Blob{/* 此替身只覆写整 Blob 的 arrayBuffer;当前摘要走 slice().arrayBuffer(),故 gate 不拦截分块读取。本例断言调用后取消的拒绝,读块挂起场景另由 sha256.test.mjs 覆盖。 */ async arrayBuffer(){await gate;return super.arrayBuffer();}}(['abc']),result=importSourceDigest(bytes,later.signal);later.abort();release();await assert.rejects(result,/import_aborted/);
@@ -8,10 +8,12 @@ const signal=()=>new AbortController().signal;
// 把文本包装成 Blob,绑定 fingerprint,并允许各用例覆盖 chunkBytes、单条上限及调度端口。
function source(text,options={}){return createJsonArrayImportSource({bytes:new Blob([text]),fingerprint:'immutable-source',...options});}
// 以 nextCursor 顺序排空且断言游标递增,最多 100 行,避免测试在读取器不前进时无限循环。
// 收集返回行而非只收 value:错误行也推进游标,调用方才能报告坏记录并继续或结束。
async function all(reader){let cursor=0;const rows=[];for(let i=0;i<100;i++){const row=await reader.read(cursor,signal());if(!row)return rows;assert.ok(row.nextCursor>cursor);rows.push(row);cursor=row.nextCursor;}throw Error('reader did not terminate');}
// 每字节分块处理中文、转义引号和字符串中的括号/逗号,结果与原生 JSON 相同;换块宽重开仍能从旧字节游标继续。
test('array reader handles UTF8, escaped strings and nested delimiters across one-byte chunks and resumes from the same byte cursor',async()=>{const text=' [ {"title":"图\\\"片,]","nested":[1,{"x":"]"}]},\n {"title":"second"} ] ',reader=source(text,{chunkBytes:1}),rows=await all(reader);assert.deepEqual(rows.map(r=>r.value),JSON.parse(text));assert.ok(rows.every(r=>!r.error));const reopened=source(text,{chunkBytes:3});assert.deepEqual(await reopened.read(rows[1].cursor,signal()),rows[1]);assert.equal(await reopened.read(rows[1].nextCursor,signal()),null);});
// 缺失/多余分隔和尾部垃圾保留已接受第一项,随后报告失败;空数组正常,空文件或非法尾逗号明确拒绝。
// “前缀保留”指读取器已返回的行,不涉及 editor 发布、history 或后端同步;本文件未创建这些组件。
test('structurally malformed tails report failure after the already accepted prefix',async()=>{for(const tail of [',]',',','','] garbage']){const rows=await all(source('[{"id":1},{"id":2}'+tail));assert.deepEqual(rows[0].value,{id:1});assert.ok(rows.at(-1).error,tail);}assert.deepEqual(await all(source('[]')),[]);await assert.rejects(all(source('')),/invalid_json_array/);assert.equal((await all(source('[1,]')))[0].error,'invalid_json_array');});
// 超大字符串只扫描不解码,报告单条错误后仍读到后续对象;非法 UTF-8 保留明确错误。
test('oversized records skip bounded payload and allow a later valid record; bad UTF8 remains explicit',async()=>{const rows=await all(source(JSON.stringify(['x'.repeat(200),{ok:true}]),{maxRecordBytes:32,chunkBytes:8}));assert.equal(rows[0].error,'record_too_large');assert.deepEqual(rows[1].value,{ok:true});const reader=createJsonArrayImportSource({bytes:new Blob([new Uint8Array([91,34,255,34,93])]),fingerprint:'bad'});assert.equal((await all(reader))[0].error,'invalid_utf8');});
@@ -20,6 +20,7 @@ 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});
// 请求跨边界的 12 字节,但证明按对齐整块计算;后两块未被请求,不应进入 checkpoint。
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);
@@ -61,6 +62,7 @@ test('one canceled waiter cannot cancel a shared block or advance its checkpoint
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);
// 第二次调用复用仍挂起的块;close 立即拒绝调用者,底层 I/O 要等 gate 返回后才算真正退出。
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/);
});
@@ -68,6 +70,7 @@ test('canceled multi-block reads and close never publish partial read evidence',
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));
// activeReads 是仍等待结果的调用数,pendingBlocks 才含取消后尚未退出的原生读取;二者不可混作并发计数。
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();
});