diff --git a/tests/headless/import-batches.test.mjs b/tests/headless/import-batches.test.mjs index 173dc45..bd96bc1 100644 --- a/tests/headless/import-batches.test.mjs +++ b/tests/headless/import-batches.test.mjs @@ -1,11 +1,16 @@ +/** + * 导入发布批量的自适应控制测试。目标耗时保持不变，根据固定发布成本与每项成本增减批量，而不是固定每批若干项。 + */ import {test} from 'node:test'; 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',()=>{ 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)); }); +// 不足一整批的尾项不用于虚增容量；昂贵整批迅速退回一项，后续单项成本降低后再渐进恢复。 test('expensive batches back off, then recover as costs fall; partial tails do not inflate the limit',()=>{ const batches=createImportBatches(6,32);for(let i=0;i<4;i++)batches.record(batches.limit,1);const full=batches.limit;batches.record(1,1);assert.equal(batches.limit,full);batches.record(full,60);assert.equal(batches.limit,1);batches.record(1,12);assert.equal(batches.limit,1);batches.record(1,2);assert.equal(batches.limit,2);batches.record(2,2);assert.equal(batches.limit,3); }); diff --git a/tests/headless/import-document-stream.test.mjs b/tests/headless/import-document-stream.test.mjs index 749e31d..db33107 100644 --- a/tests/headless/import-document-stream.test.mjs +++ b/tests/headless/import-document-stream.test.mjs @@ -1,13 +1,23 @@ +/** + * 文档 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); @@ -16,11 +26,13 @@ test('canonical document opens and emits its first record without reading the ga 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/); }); +// 非法墓碑、垃圾、重复字段、未知尾字段及尾逗号不撤回合法元素，只追加一条可推进游标的错误后结束。 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); @@ -29,22 +41,27 @@ test('bad envelope tails preserve legal records and become one explicit terminal 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); }); diff --git a/tests/headless/import-document.test.mjs b/tests/headless/import-document.test.mjs index fcb2170..d27b990 100644 --- a/tests/headless/import-document.test.mjs +++ b/tests/headless/import-document.test.mjs @@ -1,11 +1,22 @@ +/** + * 完整文档 envelope 的索引导入源测试（parser 4）。头部/拓扑先验证，元素正文按稳定序号读取，原根 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/分隔符正文及墓碑，检查封装而非简单数组。 const signal=()=>new AbortController().signal,source=value=>new Blob([typeof value==='string'?value:JSON.stringify(value)]),document={documentFormatVersion:1,projectId:'source',canvases:[{id:'root',parent:null},{id:'child',parent:{canvasId:'root',entryElementId:'frame'}}],elements:[{id:'frame',data:{title:'中文 [,] "'}},{id:'inside',ownerCanvasId:'child'}],tombstones:{old:2}}; +// 为每次打开绑定固定 fingerprint；重开序号读取应得到同样记录，不依赖上次内存游标。 const open=bytes=>openDocumentImportSource({bytes,fingerprint:'original-file',signal:signal()}); +// 子 canvas 先输出，原根只作 alias；frame/member 以稳定序号读取，重开后同序号结果一致，到末尾返回 null。 test('export envelope retains nested canvas records, emits root alias and seeks payloads by stable ordinal',async()=>{const reader=await open(source(document));assert.equal(reader.parserVersion,4);const a=await reader.read(0,signal()),b=await reader.read(1,signal());assert.equal(a.value.kind,'canvas');assert.equal(a.value.record.id,'child');assert.equal(a.value.sourceRootId,'root');assert.deepEqual(b.value.record,document.elements[0]);const reopened=await open(source(document));assert.deepEqual(await reopened.read(2,signal()),await reader.read(2,signal()));assert.equal(await reader.read(3,signal()),null);}); +// envelope 字段顺序可变，但未来版本、多根、缺失父 canvas、非法墓碑及重复键都不能生成可用源。 test('envelope fields may be reordered, while future versions, duplicate keys and broken topology reject',async()=>{const reordered={elements:document.elements,tombstones:{},canvases:document.canvases,projectId:'source',documentFormatVersion:1};assert.equal((await(await open(source(reordered))).read(1,signal())).value.record.id,'frame');for(const bad of [{...document,documentFormatVersion:2},{...document,canvases:[...document.canvases,{id:'other',parent:null}]},{...document,canvases:[{id:'root',parent:null},{id:'child',parent:{canvasId:'missing',entryElementId:'e'}}]},{...document,tombstones:{x:0}},'{"documentFormatVersion":1,"documentFormatVersion":1}'])await assert.rejects(open(source(bad)));}); +// 超过记录上限时不解码巨型正文并报告 record_too_large，后续记录仍可读；尾部垃圾及预先取消明确拒绝。 test('payload errors are per record and large records are never decoded; malformed envelope tails reject',async()=>{const reader=await open(source({...document,elements:[{id:'huge',text:'x'.repeat(300000)},document.elements[1]]}));assert.equal((await reader.read(1,signal())).error,'record_too_large');assert.equal((await reader.read(2,signal())).value.record.id,'inside');await assert.rejects(open(source(JSON.stringify(document)+'garbage')));const controller=new AbortController();controller.abort();await assert.rejects(openDocumentImportSource({bytes:source(document),fingerprint:'x',signal:controller.signal}),/import_aborted/);}); import {prepareNativeImport} from '../../src/lib/canvas-runtime/import-native.ts'; 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 拒绝。 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);}); diff --git a/tests/headless/import-identities.test.mjs b/tests/headless/import-identities.test.mjs index ec622d9..42556f6 100644 --- a/tests/headless/import-identities.test.mjs +++ b/tests/headless/import-identities.test.mjs @@ -1,13 +1,19 @@ +/** + * 导入稳定 ID 表的不可变捕获与增量分配测试。安全冻结表可缓存反向身份集合，外部可变/getter 表必须防御性捕获。 + */ import {test} from 'node:test'; import assert from 'node:assert/strict'; import {createImportIdentities} from '../../src/lib/canvas-runtime/import-identities.ts'; +// 万项冻结表反复 reserve 已有键不分配新 ID，也不重复 ownKeys 扫描；clear 后下次捕获才重新扫描。 test('preallocated 10000-ID tables keep their identity across records without allocating or rescanning',()=>{ - let reads=0;const raw=Object.freeze(Object.fromEntries(Array.from({length:10000},(_,i)=>['key'+i,'id'+i]))),ids=new Proxy(raw,{ownKeys(target){reads++;return Reflect.ownKeys(target);}}),capture=createImportIdentities(),allocate=()=>{throw Error('unexpected allocation');}; + let reads=0;const raw=Object.freeze(Object.fromEntries(Array.from({length:10000},(_,i)=>['key'+i,'id'+i]))),ids=new Proxy(raw,{/* 只观测枚举次数，不改变表内容，用于识别不必要的万项重扫。 */ ownKeys(target){reads++;return Reflect.ownKeys(target);}}),capture=createImportIdentities(),allocate=()=>{throw Error('unexpected allocation');}; const first=capture(ids,allocate);assert.equal(first.reserve('key0'),'id0');assert.strictEqual(first.finish(),ids);const checked=reads;for(let i=0;i<10000;i++){const row=capture(ids,allocate);assert.equal(row.reserve('key'+i),'id'+i);assert.ok(row.has('id'+i));assert.strictEqual(row.finish(),ids);}assert.equal(reads,checked);capture.clear();capture(ids,allocate);assert.ok(reads>checked); }); +// 首次新增键才分离字典及反向集合，finish 结果不受之后 reserve 影响；兄弟捕获独立，重复目标 ID 必须拒绝。 test('first allocation detaches the dictionary and reverse set, and a late callback cannot change its published result',()=>{ const ids=Object.freeze({old:'old-id'}),capture=createImportIdentities();let next=0;const row=capture(ids,()=>`new-${++next}`);row.reserve('added');const result=row.finish();assert.notStrictEqual(result,ids);row.reserve('late');assert.deepEqual(result,{old:'old-id',added:'new-1'});assert.deepEqual(ids,{old:'old-id'});const sibling=capture(ids,()=> 'new-1');assert.equal(sibling.has('new-1'),false);assert.equal(sibling.reserve('other'),'new-1');assert.throws(()=>capture(ids,()=> 'old-id').reserve('duplicate'),/duplicate_import_identity/); }); +// 可变字典捕获当时值，不受随后外部改写影响；冻结 getter 仍逐次重新读取，不能仅凭 Object.freeze 复用动态值。 test('mutable and frozen getter dictionaries are captured defensively rather than reused',()=>{ const capture=createImportIdentities(),mutable={one:'first'},first=capture(mutable,()=> 'new');mutable.one='second';assert.equal(first.reserve('one'),'first');assert.deepEqual(first.finish(),{one:'first'});assert.equal(capture(mutable,()=> 'new').reserve('one'),'second'); let current='old',reads=0;const dynamic=Object.freeze({get one(){reads++;return current;}}),one=capture(dynamic,()=> 'new');current='changed';assert.equal(one.reserve('one'),'old');assert.equal(reads,1);assert.equal(capture(dynamic,()=> 'new').reserve('one'),'changed');assert.equal(reads,2); diff --git a/tests/headless/import-integrity.test.mjs b/tests/headless/import-integrity.test.mjs index d6aeb17..0be77a4 100644 --- a/tests/headless/import-integrity.test.mjs +++ b/tests/headless/import-integrity.test.mjs @@ -1,11 +1,16 @@ +/** + * 导入源全文件 SHA-256 绑定与任务完整性版本测试。新任务在绑定摘要前不能声明已解析游标或稳定映射，旧格式按明确兼容规则读取。 + */ 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 向量和不同输入区分完整源摘要；预先取消及异步读取返回前取消都拒绝，不接受迟到结果。 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{async arrayBuffer(){await gate;return super.arrayBuffer();}}(['abc']),result=importSourceDigest(bytes,later.signal);later.abort();release();await assert.rejects(result,/import_aborted/); + 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/); }); +// 新 job 为 v3/integrityVersion=1，未绑定摘要前拒绝进度，非法摘要/未来完整性解释器拒绝；显式旧格式及已绑定新格式仍可校验。 test('new source jobs declare integrity version and reject progress without a bound hash; legacy data remains readable',()=>{ const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'file',parserVersion:1,source:{kind:'native-ndjson',fileId:'file',targetCanvasId:'canvas',byteLength:10}});assert.equal(job.version,3);assert.equal(job.source.integrityVersion,1);validateImportJob(job); assert.throws(()=>validateImportJob({...job,cursor:1}),/import_source_integrity_missing/);assert.throws(()=>validateImportJob({...job,stableIds:{'element:A':'new-A'}}),/import_source_integrity_missing/);assert.throws(()=>validateImportJob({...job,source:{...job.source,sha256:'invalid'}}),/invalid_import_source/);assert.throws(()=>validateImportJob({...job,source:{...job.source,integrityVersion:3}}),/import_source_integrity_interpreter_unavailable/); diff --git a/tests/headless/import-json-array.test.mjs b/tests/headless/import-json-array.test.mjs index 343ff69..8904dae 100644 --- a/tests/headless/import-json-array.test.mjs +++ b/tests/headless/import-json-array.test.mjs @@ -1,9 +1,21 @@ +/** + * JSON 数组渐进读取测试。按字节游标扫描 UTF-8 与嵌套结构，可保留已接受前缀并将坏尾部作为显式失败行。 + * 读取器还需限制单条解码大小，在固定字节工作量或实际耗时达到预算时让出。 + */ import {test} from 'node:test';import assert from 'node:assert/strict';import {createJsonArrayImportSource} from '../../src/lib/canvas-runtime/import-json-array.ts'; +// 每次读取提供独立未取消信号，取消用例显式使用自己的 controller。 const signal=()=>new AbortController().signal; +// 把文本包装成 Blob，绑定 fingerprint，并允许各用例覆盖 chunkBytes、单条上限及调度端口。 function source(text,options={}){return createJsonArrayImportSource({bytes:new Blob([text]),fingerprint:'immutable-source',...options});} +// 以 nextCursor 顺序排空且断言游标递增，最多 100 行，避免测试在读取器不前进时无限循环。 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);}); +// 缺失/多余分隔和尾部垃圾保留已接受第一项，随后报告失败；空数组正常，空文件或非法尾逗号明确拒绝。 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');}); +// 长扫描在工作量边界让出，yield 内取消不能返回已完成记录；负字节游标拒绝。 test('long scans yield and honor abort without publishing a completed record',async()=>{const controller=new AbortController();let yields=0;const reader=source(JSON.stringify(['x'.repeat(40000)]),{now:()=>0,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(reader.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);await assert.rejects(reader.read(-1,signal()),/invalid_import_cursor/);}); +// 即使只扫描 2048 字符，耗时预算先到也应让出，取消错误保持 import_aborted。 test('slow array scans yield by elapsed time before the fixed byte ceiling and abort at that boundary',async()=>{const controller=new AbortController();let clock=0,yields=0;const reader=source(JSON.stringify(['x'.repeat(2048)]),{now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(reader.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);}); diff --git a/tests/headless/import-read-proof.test.mjs b/tests/headless/import-read-proof.test.mjs index e7b0d92..e377995 100644 --- a/tests/headless/import-read-proof.test.mjs +++ b/tests/headless/import-read-proof.test.mjs @@ -1,3 +1,7 @@ +/** + * 导入按块读取证明测试。证明只记录成功消费范围涉及的对齐块摘要，不为尚未读到的文件尾部提供真实性保证。 + * 原始块缓存可淘汰，但预期摘要保留；共享读取、取消、关闭与容量限制不能提前发布半份读取证据。 + */ 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'; @@ -6,10 +10,14 @@ import {openStreamingDocumentImportSource} from '../../src/lib/canvas-runtime/im 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)); @@ -17,52 +25,61 @@ test('range reads return exact owned bytes and checkpoint only the aligned block 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 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;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 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;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}} + 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();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 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;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}} + 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;class Count extends Blob{slice(...args){reads++;return super.slice(...args);}} + 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());}}; @@ -70,6 +87,7 @@ test('byte ports cannot return short bytes, wrong source length, or disguise int }); +// 未来块临时 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());}}; diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs index 8236364..1fc0f29 100644 --- a/tests/headless/imports.test.mjs +++ b/tests/headless/imports.test.mjs @@ -1,22 +1,40 @@ +/** + * 独立导入 runner 的解析、依赖分组、发布和持久化状态机测试。每个已提交前缀可见，坏记录及取消不默认回滚全部导入。 + * 依赖循环作为强连通组原子提交；稳定 ID、待处理定位、源身份和 generation 支持恢复并隔离迟到异步准备。 + * fixture 用内存文档与克隆 cut 验证端口调用顺序；引擎/真实存储集成另有测试，不由此模拟持久层事务保证。 + */ import {test} from 'node:test'; import assert from 'node:assert/strict'; import {createDocument,applyChanges,validateDocument,createRegistry,builtinComponents,validJson,fault} from '../../src/lib/canvas-core/index.ts'; import {createImportJob,validateImportJob,createNdjsonImportSource,createImportRunner} from '../../src/lib/canvas-runtime/imports.ts'; +// 内置定义及空单 canvas 文档，供每个 fixture 从相同不可变基线开始。 const registry=createRegistry(builtinComponents()),base=createDocument('project',[{id:'canvas',parent:null}],[]); +// 手动 Promise 屏障用于暂停源读取、persist 或 beforeCommit，定位取消/恢复的边界。 const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};}; +// 有限推进微任务，检查预算尚未放行时确实没有提前发布内容。 const ticks=async()=>{for(let i=0;i<30;i++)await Promise.resolve();}; +// 把对象或故意损坏的原始行拼成 NDJSON Blob，允许同一输入同时含合法和错误记录。 const bytes=records=>new Blob([records.map(record=>typeof record==='string'?record:JSON.stringify(record)).join('\n')]); +// 组装源、当前 document/job 状态根与 runner 端口，options 可覆写端口或指定恢复状态。 function fixture(records,options={}){ const source=options.source??createNdjsonImportSource({bytes:bytes(records),fingerprint:'immutable-source'}),calls=[],errors=[],saved=[],commits=[];let n=0,time=0,root=options.root??{document:base,job:createImportJob({id:'import',projectId:'project',sourceFingerprint:source.fingerprint,parserVersion:1})}; + // root getter/setter 暴露测试当前状态；真实修改通过 publish/commit，少数用例显式替换它来模拟外部编辑或换代。 const f={source,calls,errors,saved,commits,get root(){return root;},set root(value){root=value;},ports:{source,getJob:()=>root.job,createId:()=>`new-${++n}`,now:()=>time, + // 以 job.generation 做条件发布，只替换 job 元数据；文档变化必须经 commit 入口。 publish(job,expected){assert.equal(root.job.generation,expected);root={...root,job};}, + // 最小解析转换：校验输入 ID、reserve 稳定 ID、保留 source-key 依赖，再生成节点创建；bad 标记注入非法宽度。 prepare(line,job,reserve){time+=2;const input=line.value;if(!input||typeof input!=='object'||!input.id)throw fault('invalid_record');const id=reserve(input.id);return{key:input.id,dependencies:input.dependencies??[],changes:[{kind:'element',id,before:null,after:{id,incarnation:1,structure:'node',ownerCanvasId:'canvas',componentId:'text',dataVersion:1,data:{title:input.title??input.id,text:''},geometry:{x:0,y:0,width:input.bad?-1:100,height:100,rotation:0}}}]};}, + // 在私有 document 应用整组变化并校验，成功才把 document/job 一起替换；拒绝时原 root 必须不变。 commit(units,job,expected){time+=2;assert.equal(root.job.generation,expected);calls.push(units.map(unit=>unit.key));const before=root;try{const document=applyChanges(root.document,units.flatMap(unit=>unit.changes));validateDocument(document,registry);root={document,job};commits.push(root);return{status:'applied',revision:commits.length,operationIds:[]};}catch(error){assert.equal(root,before);return{status:'rejected',code:error.code??'invalid',targets:[]};}}, + // 校验 job 可序列化且结构合法，再克隆 job 与文档记录为恢复 cut；saved 用于后续重开对照。 async persist(){validateImportJob(root.job);assert.equal(validJson(root.job),true);saved.push(structuredClone({job:root.job,document:{projectId:root.document.projectId,canvases:[...root.document.canvases.values()],elements:[...root.document.elements.values()],tombstones:[...root.document.tombstones]}}));}, + // 错误上报收集到 errors，yieldTask 计数协作调度；调用者 ports 最后覆盖以注入背压和故障。 onError:error=>errors.push(error),yieldTask:async()=>{f.yields++;},...options.ports},yields:0}; + // 所有端口装好后才创建 runner，避免调用未就绪 fixture；返回端口引用供各用例精确注入故障。 f.runner=createImportRunner(f.ports);return f; } +// UTF-8 按 3 字节块解析并以字节游标重读；坏 JSON/超大行只报各自行错误，后续正常行保留，输入 options 改写不换源。 test('NDJSON parses bounded UTF-8 chunks, seeks by byte cursor and reports malformed/oversized records without losing the next line',async()=>{ const source=createNdjsonImportSource({bytes:bytes([{id:'中文 😀'},'{bad}','x'.repeat(30),{id:'ok'}]),fingerprint:'file-1',maxRecordBytes:24,chunkBytes:3}),signal=new AbortController().signal; let cursor=0;const rows=[];for(;;){const row=await source.read(cursor,signal);if(!row)break;rows.push(row);cursor=row.nextCursor;} @@ -24,6 +42,7 @@ test('NDJSON parses bounded UTF-8 chunks, seeks by byte cursor and reports malfo const input={bytes:bytes([{id:'old'}]),fingerprint:'fixed'},owned=createNdjsonImportSource(input);input.bytes=bytes([{id:'replacement'}]);assert.equal((await owned.read(0,signal)).value.id,'old'); }); +// 第二项读取挂起时第一项已可见且已登记 committed key；取消保留该前缀，不能等 EOF 才显示或把已提交内容回滚。 test('records become visible before EOF, task yields real scheduling points and cancellation preserves the committed prefix',async()=>{ const gate=deferred(),entered=deferred(),raw=bytes([{id:'first'},{id:'second'}]),initial=createNdjsonImportSource({bytes:raw,fingerprint:'immutable-source'});let firstEnd; firstEnd=(await initial.read(0,new AbortController().signal)).nextCursor; @@ -32,19 +51,23 @@ test('records become visible before EOF, task yields real scheduling points and await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.root.document.elements.size,1);assert.deepEqual(f.root.job.committedSourceKeys,['first']);await f.runner.close(); }); +// child 先于 parent 时按源键等待；parent 加入后即使被用户改名，随后 child 加入也不能用旧快照覆盖用户编辑。 test('a child encountered before its parent waits by source key, then joins without overwriting user edits',async()=>{ const f=fixture([{id:'child',dependencies:['parent']},{id:'parent'}]);const real=f.ports.commit;f.ports.commit=(units,next,expected)=>{const result=real(units,next,expected);if(result.status==='applied'&&units.some(unit=>unit.key==='parent')){const id=f.root.job.stableIds.parent,element=f.root.document.elements.get(id);f.root={...f.root,document:applyChanges(f.root.document,[{kind:'field',target:{id,incarnation:1},field:'data.title',before:element.data.title,after:'user edit'}])};}return result;}; await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.root.job.committedSourceKeys,['parent','child']);assert.equal(f.root.document.elements.get(f.root.job.stableIds.parent).data.title,'user edit');assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,0); }); +// 多项批次含非法节点时先整批拒绝，再隔离独立坏项；good/tail 仍提交，失败元素不进入文档。 test('a rejected multi-record batch publishes no prefix, isolates the invalid record and retains independent content',async()=>{ const f=fixture([{id:'good',dependencies:['parent']},{id:'bad',dependencies:['parent'],bad:true},{id:'parent'},{id:'tail'}]);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.calls.some(keys=>keys.includes('good')&&keys.includes('bad')));assert.deepEqual(f.root.job.committedSourceKeys,['parent','good','tail']);assert.equal(f.root.document.elements.has(f.root.job.stableIds.bad),false);assert.equal(f.root.job.failures.find(failure=>failure.key==='bad').code,'invalid_geometry');assert.equal(f.errors.length,0); }); +// 前 100 项后发生解析错误并有缺依赖项，仍保留原内容和后来合法项；以 partial 记录两种失败，不插入悬空依赖。 test('record 101 failure preserves prior content and later valid records; missing dependencies report partial without dangling insertion',async()=>{ const input=Array.from({length:100},(_,index)=>({id:`n-${index}`}));input.push('{bad}',{id:'after'},{id:'edge',dependencies:['missing']});const f=fixture(input);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.equal(f.root.document.elements.size,101);assert.equal(f.root.job.failures.length,2);assert.ok(f.root.job.failures.some(failure=>failure.code==='invalid_json'));assert.ok(f.root.job.failures.some(failure=>failure.code==='unresolved_dependency'));assert.equal(f.root.document.elements.has(f.root.job.stableIds.edge),false);validateImportJob(JSON.parse(JSON.stringify(f.root.job))); }); +// 待处理 locator 已保存后关闭再恢复，必须复用原 ID；已 committed key 对应元素后来被用户删除，也不能重导入复活。 test('a durable pending locator resumes with the same IDs; a committed key is not inserted again after deletion',async()=>{ const gate=deferred(),entered=deferred(),f=fixture([{id:'one',title:'original'}]);const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length&&!f.root.job.committedSourceKeys.length){entered.resolve();await gate.promise;}}; const task=f.runner.start();await entered.promise;const cut=f.saved.at(-1),closing=f.runner.close();gate.resolve();await task;await closing; @@ -52,26 +75,31 @@ test('a durable pending locator resumes with the same IDs; a committed key is no const id=resumed.root.job.stableIds.one,element=resumed.root.document.elements.get(id),deleted=applyChanges(resumed.root.document,[{kind:'element',id,before:element,after:null}]),again=fixture([],{source:f.source,root:{document:deleted,job:{...resumed.root.job,status:'paused',eof:false}},ports:{createId(){throw Error('reallocated identity');}}});await again.runner.start();assert.equal(again.root.document.elements.has(id),false);assert.equal(again.calls.length,0); }); +// 同 fingerprint 字节变更和 fingerprint 不同分别报告 source_changed/mismatch，暂停且不提交待处理项。 test('changed source bytes or source identity cannot claim a durable pending item',async()=>{ const oldSource=createNdjsonImportSource({bytes:bytes([{id:'one',title:'aaa'}]),fingerprint:'immutable-source'}),gate=deferred(),entered=deferred(),f=fixture([],{source:oldSource});const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length){entered.resolve();await gate.promise;}}; const task=f.runner.start();await entered.promise;const root=f.root,closing=f.runner.close();gate.resolve();await task;await closing; for(const [title,fingerprint,code]of[['bbb','immutable-source','import_source_changed'],['aaa','different-source','import_source_mismatch']]){const source=createNdjsonImportSource({bytes:bytes([{id:'one',title}]),fingerprint}),r=fixture([],{source,root});await r.runner.start();assert.equal(r.root.job.status,'paused');assert.equal(r.root.job.error,code);assert.equal(r.root.document.elements.size,0);assert.equal(r.calls.length,0);} }); +// pending 容量满暂停但保留 locator 和映射；提高容量再开可继续同游标/ID，而不是丢弃或重新编号。 test('dependency backpressure pauses with locators intact and a larger budget resumes the same cursor and mapping',async()=>{ const f=fixture([{id:'child',dependencies:['parent']},{id:'parent'}],{ports:{maxPending:1}});await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'dependency_capacity');assert.equal(f.root.job.pending.length,1);const id=f.root.job.stableIds.child; const r=fixture([],{source:f.source,root:f.root,ports:{maxPending:8,createId:()=> 'parent-id'}});await r.runner.start();assert.equal(r.root.job.status,'completed');assert.equal(r.root.job.stableIds.child,id);assert.equal(r.root.document.elements.size,2); }); +// 未来 job 版本、重复目标 ID 和缺乏结束条件的 completed cut 均在校验阶段拒绝。 test('future interpreter versions and malformed job cuts fail before any source read',()=>{ const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'s',parserVersion:1});assert.throws(()=>validateImportJob({...job,version:4}),/import_interpreter_unavailable/);assert.throws(()=>validateImportJob({...job,stableIds:{a:'same',b:'same'}}),/invalid_import_job/);assert.throws(()=>validateImportJob({...job,status:'completed'}),/invalid_import_job/); }); +// 便宜记录首项仍立即发布，随后可合并至最多 16 项；元数据发布次数受控，不能每条都重复发布完整状态。 test('fast independent records stage bounded publications and commits while the first record remains immediate',async()=>{ const f=fixture(Array.from({length:80},(_,index)=>({id:`batch-${index}`})),{ports:{now:()=>0,maxBatch:16,optimisticLocal:true}}),publish=f.ports.publish;let publications=0;f.ports.publish=(...args)=>{publications++;return publish(...args);}; await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,80);assert.deepEqual(f.calls[0],['batch-0']);assert.ok(f.calls.some(group=>group.length>1));assert.ok(f.calls.every(group=>group.length<=16));assert.ok(publications<20,`${publications} metadata publications for 80 records`);assert.equal(f.errors.length,0);validateImportJob(f.root.job); }); +// 第三项预读挂起时前两项已准备内容仍可见，源 read 始终串行；取消保留两项，不要求并发 seek 解锁。 test('a stalled speculative read cannot hide the ready staged prefix or require concurrent source seeks',async()=>{ const gate=deferred(),entered=deferred(),source=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'},{id:'third'}]),fingerprint:'immutable-source'}),signal=new AbortController().signal,first=await source.read(0,signal),second=await source.read(first.nextCursor,signal);let active=0; const f=fixture([],{source:{...source,async read(cursor,signal){assert.equal(active,0,'concurrent source seeks');active++;try{if(cursor===second.nextCursor){entered.resolve();await gate.promise;}return await source.read(cursor,signal);}finally{active--;}}},ports:{now:()=>0,optimisticLocal:true}}); @@ -79,25 +107,30 @@ test('a stalled speculative read cannot hide the ready staged prefix or require for(let i=0;i<100&&f.root.document.elements.size<2;i++)await new Promise(resolve=>setTimeout(resolve,1));assert.equal(f.root.document.elements.size,2);assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.job.eof,false);await f.runner.cancel();gate.resolve();await running;assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.status,'cancelled');assert.equal(f.errors.length,0); }); +// 第二项转换后微任务取消，只保留首项；未发布的第二项游标、稳定 ID 和 pending 不能泄漏到 job。 test('cancelling during a staged decode discards unpublished cursor and identity changes',async()=>{ const f=fixture([{id:'first'},{id:'second'},{id:'third'}],{ports:{now:()=>0,optimisticLocal:true}}),prepare=f.ports.prepare;let cancelled; f.ports.prepare=(line,job,reserve)=>{const unit=prepare(line,job,reserve);if(line.value.id==='second')queueMicrotask(()=>{cancelled=f.runner.cancel();});return unit;}; await f.runner.start();await cancelled;assert.equal(f.root.job.status,'cancelled');assert.deepEqual(f.root.job.committedSourceKeys,['first']);assert.equal(f.root.document.elements.size,1);assert.equal(Object.hasOwn(f.root.job.stableIds,'second'),false);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,0);validateImportJob(f.root.job); }); +// 源 I/O 在已准备项之后失败，先发布 ready 前缀再 paused 并报告原错误，不能把已就绪内容藏在错误后。 test('a source I/O failure is reported after publishing all already prepared ready records',async()=>{ const raw=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'}]),fingerprint:'immutable-source'}),first=await raw.read(0,new AbortController().signal),second=await raw.read(first.nextCursor,new AbortController().signal),f=fixture([],{source:{...raw,async read(cursor,signal){if(cursor===second.nextCursor)throw fault('source_offline');return raw.read(cursor,signal);}},ports:{now:()=>0,optimisticLocal:true}}); await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'source_offline');assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,1); }); +// A/B/C 循环即使 maxBatch=1 仍作为一个完整原子组提交一次，不能用软批量限制拆开依赖环。 test('a reference cycle commits every member in one root even when the adaptive batch budget is one',async()=>{ const f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['C']},{id:'C',dependencies:['A']}],{ports:{maxBatch:1}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.calls,[['A','B','C']]);assert.equal(f.commits.length,1);assert.equal(f.root.document.elements.size,3);validateImportJob(f.root.job); }); +// 依赖环一项非法时所有环成员共同失败，任何尝试都含完整环；无关 tail 仍成功，不能单独隔离并发布环成员。 test('a rejected reference cycle never isolates or publishes one member; independent content survives',async()=>{ const f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['A'],bad:true},{id:'tail'}]);await f.runner.start();assert.equal(f.root.job.status,'partial');assert.deepEqual(f.root.job.committedSourceKeys,['tail']);assert.ok(f.calls.filter(keys=>keys.includes('A')||keys.includes('B')).every(keys=>keys.includes('A')&&keys.includes('B')));assert.deepEqual(f.root.job.failures.map(value=>value.key),['A','B']);assert.equal(f.root.document.elements.size,1); }); +// 整环 locator 保存时文档仍为空；恢复在同一次提交中加入所有成员，不重新分配 ID 或暴露半组。 test('a durable pending cycle resumes without reallocating IDs or exposing a partial group',async()=>{ const gate=deferred(),entered=deferred(),f=fixture([{id:'A',dependencies:['B']},{id:'B',dependencies:['A']}]);const persist=f.ports.persist;f.ports.persist=async()=>{await persist();if(f.root.job.pending.length===2&&!f.root.job.committedSourceKeys.length){entered.resolve();await gate.promise;}}; const running=f.runner.start();await entered.promise;const cut=f.root,closing=f.runner.close();gate.resolve();await running;await closing;assert.equal(cut.document.elements.size,0); @@ -105,16 +138,19 @@ test('a durable pending cycle resumes without reallocating IDs or exposing a par }); +// 新 job 为 v3；显式可读的 v1 旧 job 恢复后在首个 published cut 升级为 v3，不能继续发布旧解释器形态。 test('new jobs carry interpreter version 3 and a resumed legacy job upgrades in its first published cut',async()=>{ const f=fixture([{id:'one'}]);assert.equal(f.root.job.version,3);f.root={...f.root,job:{...f.root.job,version:1}};validateImportJob(f.root.job);const publish=f.ports.publish;f.ports.publish=(job,expected)=>{assert.equal(job.version,3);return publish(job,expected);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.version,3); }); +// 乐观本地已发布两项时，宿主 durability budget 可阻止第三项；到 EOF 后仍先持久化 running 内容，再声明 completed。 test('optimistic local checkpoints use host backpressure without declaring completion ahead of persistence',async()=>{ const gate=deferred(),entered=deferred(),f=fixture([{id:'one'},{id:'two'},{id:'three'}],{ports:{optimisticLocal:true,maxBatch:1}});let blocked=false; f.ports.awaitDurabilityBudget=async()=>{if(!blocked&&f.root.document.elements.size===2){blocked=true;entered.resolve();await gate.promise;}}; const running=f.runner.start();await entered.promise;assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.status,'running');assert.equal(f.root.job.committedSourceKeys.length,2);gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,3);assert.ok(f.saved.some(cut=>cut.job.status==='running'&&cut.job.eof&&cut.document.elements.length===3)); }); +// 最终内容 cut 或 completed 元数据保存失败均 paused；修复后续跑不重复 committed key 或 ID，仅补完持久化终态。 test('both final content and completion-metadata storage failures pause resumably without replaying committed keys',async()=>{ for(const phase of ['content','metadata']){ const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist;let broken=true; @@ -123,46 +159,59 @@ test('both final content and completion-metadata storage failures pause resumabl } }); +// 最后持久化屏障挂起时 pause 立即生效；迟到成功不得发布 completed，再次 start 才合法完成且不重复内容提交。 test('pausing during the final durability barrier prevents a late completion publication',async()=>{ const gate=deferred(),entered=deferred(),f=fixture([{id:'one'}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist; f.ports.persist=async()=>{if(f.root.job.eof){entered.resolve();await gate.promise;}return persist();};const running=f.runner.start();await entered.promise;const pausing=f.runner.pause();assert.equal(f.root.job.status,'paused');gate.resolve();await pausing;await running;assert.equal(f.root.job.status,'paused');await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.calls.length,1); }); +// partial 状态保存失败后恢复，只补存已有 unresolved_dependency 诊断，不能重复追加同一失败或内容。 test('resuming a failed partial-status write does not duplicate unresolved dependency diagnostics',async()=>{ const f=fixture([{id:'good'},{id:'waiting',dependencies:['missing']}],{ports:{optimisticLocal:true,awaitDurabilityBudget(){}}}),persist=f.ports.persist;let broken=true; f.ports.persist=async()=>{if(broken&&f.root.job.status==='partial')throw fault('quota');return persist();};await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.failures.length,1);broken=false;await f.runner.start();assert.equal(f.root.job.status,'partial');assert.equal(f.root.job.failures.length,1);assert.equal(f.root.document.elements.size,1);assert.equal(f.calls.length,1); }); +// parsed locator 已发布但宿主 parsed 阶段预算未放行，内容根仍为空；背压释放后才提交两项并完成。 test('a ready parser publication observes host backpressure before admitting its content root',async()=>{ const gate=deferred(),entered=deferred();let blocked=true;const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,maxBatch:1,awaitDurabilityBudget(stage){if(blocked){assert.equal(stage,'parsed');entered.resolve();return gate.promise;}}}});const running=f.runner.start();await entered.promise;assert.equal(f.root.job.pending.length,1);assert.equal(f.root.job.committedSourceKeys.length,0);assert.equal(f.root.document.elements.size,0);await ticks();assert.equal(f.root.document.elements.size,0,'ready content bypassed host budget');blocked=false;gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,2);assert.equal(f.errors.length,0);await f.runner.close(); }); +// 异步 beforeCommit 挂起时不能提前发布文档或 committed key；期间 cancel 后，迟到 enabled:true 也不得提交。 test('async commit preparation cannot publish while waiting and cancellation fences a late successful result',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'image'}],{ports:{async beforeCommit(units,job,signal){assert.equal(f.root.document.elements.size,0);assert.ok(job.pending.length);entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;assert.equal(f.commits.length,0);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.commits.length,0);assert.equal(f.root.document.elements.size,0);await f.runner.close();}); +// 异步准备拒绝 denied 独立项后记录 asset_forbidden，另一个独立项仍发布，拒绝不冒充系统异常。 test('async preparation denial isolates independent groups without publishing the denied record',async()=>{const prepared=[],f=fixture([{id:'denied'},{id:'allowed'}],{ports:{async beforeCommit(units){prepared.push(units.map(unit=>unit.key));return units.some(unit=>unit.key==='denied')?{enabled:false,code:'asset_forbidden',targets:[]}:{enabled:true};}}});await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.root.job.failures.some(row=>row.key==='denied'&&row.code==='asset_forbidden'));assert.equal(f.root.document.elements.size,1);assert.equal([...f.root.document.elements.values()][0].data.title,'allowed');assert.ok(prepared.length>=2);assert.equal(f.errors.length,0);await f.runner.close();}); +// 异步准备期间 job.generation 外部递增，旧 attempt 返回后不能提交或改进度；新 start 可重新完成。 test('a generation change during async preparation prevents the old attempt from committing or changing progress',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'later'}],{ports:{async beforeCommit(){entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;f.root={...f.root,job:{...f.root.job,generation:f.root.job.generation+1}};const generation=f.root.job.generation;gate.resolve();await running;assert.equal(f.commits.length,0);assert.equal(f.root.job.generation,generation);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,1);await f.runner.close();}); +// NDJSON 扫描短行时也检查耗时预算，在返回已解码记录前 yield；yield 内取消须以 import_aborted 拒绝。 test('NDJSON checks elapsed time before returning a short scanned record and honors cancellation',async()=>{const controller=new AbortController();let clock=0,yields=0;const source=createNdjsonImportSource({bytes:new Blob([JSON.stringify({text:'x'.repeat(2048)})+'\n']),fingerprint:'timed',chunkBytes:256,now:()=>clock+=10,frameBudgetMs:6,yieldTask:async()=>{yields++;controller.abort();}});await assert.rejects(source.read(0,controller.signal),/import_aborted/);assert.equal(yields,1);}); +// 已预分配大量 ID 的恢复 job，在 parsed/committed cut 都保留同一稳定表对象；不得重复分配或为每行复制整张表。 test('a resumed preallocated import retains the exact stable-ID table through parsed and committed checkpoints',async()=>{ const stableIds=Object.freeze(Object.fromEntries([...Array.from({length:1000},(_,i)=>['reserved-'+i,'unused-'+i]),['one','id-one'],['two','id-two']])),job={...createImportJob({id:'import',projectId:'project',sourceFingerprint:'immutable-source',parserVersion:1}),stableIds},f=fixture([{id:'one'},{id:'two'}],{root:{document:base,job},ports:{createId(){throw Error('preallocated ID was replaced');}}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.strictEqual(f.root.job.stableIds,stableIds);for(const cut of f.commits)assert.strictEqual(cut.job.stableIds,stableIds);assert.deepEqual(f.root.job.committedSourceKeys,['one','two']);assert.equal(f.errors.length,0);await f.runner.close(); }); +// createId 是宿主对象方法，runner 必须保留 this receiver，计数和分配结果由原 ports 对象维护。 test('identity allocation keeps the caller port method receiver',async()=>{ const f=fixture([{id:'bound'}],{ports:{allocated:0,createId(){return `bound-${++this.allocated}`;}}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.job.stableIds.bound,'bound-1');assert.equal(f.ports.allocated,1);assert.equal(f.errors.length,0);await f.runner.close(); }); +// 每次提交有固定成本时，首项仍立即显示，后续增长批量摊薄成本；昂贵工作后让出，不增加原耗时预算。 test('fixed-cost commits amortize ready records while preserving the immediate first record and yielding after expensive work',async()=>{ let clock=0;const f=fixture(Array.from({length:80},(_,i)=>({id:`fixed-${i}`})),{ports:{now:()=>clock,frameBudgetMs:6,maxBatch:16,optimisticLocal:true}}),commit=f.ports.commit;f.ports.commit=(units,...rest)=>{clock+=4+units.length*.05;return commit(units,...rest);};await f.runner.start();assert.equal(f.root.job.status,'completed');assert.deepEqual(f.calls[0],['fixed-0']);assert.ok(f.calls.some(group=>group.length>=8));assert.ok(f.calls.length<20);assert.ok(f.yields>0);assert.equal(f.errors.length,0);validateImportJob(f.root.job); }); +// 便宜的 700 条记录可使用最多 128 项上限并超过旧的 32 项批量，全部记录按 FIFO 提交且不丢失。 test('a cheap runner can use the engine-sized ceiling without dropping records or changing FIFO order',async()=>{ const records=Array.from({length:700},(_,index)=>({id:`wide-${index}`})),f=fixture(records,{ports:{maxBatch:128,now:()=>0}});await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,700);assert.ok(f.calls.some(batch=>batch.length>32));assert.ok(f.calls.every(batch=>batch.length<=128));assert.deepEqual(f.calls.flat(),records.map(row=>row.id));assert.equal(f.errors.length,0);await f.runner.close(); }); +// 较大上限不能掩盖单批超时：超过 40 项的模拟提交变贵后下一批必须回退，全部源键仍按序保留。 test('the larger ceiling still backs off when a measured commit exceeds the unchanged time budget',async()=>{ let clock=0;const records=Array.from({length:500},(_,index)=>({id:`cost-${index}`})),f=fixture(records,{ports:{maxBatch:128,now:()=>clock}}),commit=f.ports.commit;f.ports.commit=(units,...args)=>{clock+=units.length>40?12:1;return commit(units,...args);};await f.runner.start();const sizes=f.calls.map(batch=>batch.length),wide=sizes.findIndex(size=>size>40);assert.ok(wide>=0);assert.ok(sizes[wide+1]row.id));assert.equal(f.errors.length,0);await f.runner.close(); }); +// 300 条导入形成的 stableIds/bindings 按预算封存，与同步 image 一致且映射数量完整。 test('growing import tables yield when encoded while keeping captured IDs and bindings stable',async()=>{ const {sealRecoveryImageAsync,sealRecoveryImage}=await import('../../src/lib/canvas-runtime/recovery-encoding.ts'); const f=fixture(Array.from({length:300},(_,i)=>({id:'encode-'+i})));await f.runner.start(); @@ -173,6 +222,7 @@ test('growing import tables yield when encoded while keeping captured IDs and bi }); +// 300 个尚未等到依赖的 locator 也需协作式编码；无正文发布时仍保存完整待处理依赖，不能为了切片丢项。 test('waiting import locators and dependencies encode cooperatively before their endpoints arrive',async()=>{ const {sealRecoveryImageAsync,sealRecoveryImage}=await import('../../src/lib/canvas-runtime/recovery-encoding.ts'); const f=fixture(Array.from({length:300},(_,i)=>({id:'waiting-'+i,dependencies:['absent']})));await f.runner.start();