235 lines
34 KiB
JavaScript
235 lines
34 KiB
JavaScript
/**
|
|
* 独立导入 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;}
|
|
assert.equal(rows[0].value.id,'中文 😀');assert.equal(rows[1].error,'invalid_json');assert.equal(rows[2].error,'record_too_large');assert.equal(rows[3].value.id,'ok');assert.deepEqual(await source.read(rows[0].cursor,signal),rows[0]);
|
|
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;
|
|
const source={...initial,async read(cursor,signal){if(cursor===firstEnd){entered.resolve();await gate.promise;}return initial.read(cursor,signal);}},f=fixture([],{source,ports:{frameBudgetMs:3}});
|
|
const running=f.runner.start();await entered.promise;assert.equal(f.root.document.elements.size,1);assert.equal(f.root.job.committedSourceKeys[0],'first');assert.ok(f.yields>0);
|
|
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;
|
|
const document=createDocument(cut.document.projectId,cut.document.canvases,cut.document.elements,new Map(cut.document.tombstones)),resumed=fixture([],{source:f.source,root:{document,job:cut.job},ports:{createId(){throw Error('reallocated identity');}}});await resumed.runner.start();assert.equal(resumed.root.job.status,'completed');assert.equal(resumed.root.document.elements.size,1);assert.equal(resumed.root.job.stableIds.one,cut.job.stableIds.one);
|
|
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}});
|
|
const running=f.runner.start();await entered.promise;
|
|
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);
|
|
const r=fixture([],{source:f.source,root:cut,ports:{createId(){throw Error('identity reallocated');}}});await r.runner.start();assert.equal(r.root.job.status,'completed');assert.deepEqual(r.calls,[['A','B']]);assert.deepEqual(r.root.job.stableIds,cut.job.stableIds);
|
|
});
|
|
|
|
|
|
// 新 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;
|
|
f.ports.persist=async()=>{if(broken&&(phase==='content'?f.root.job.eof&&f.root.job.status==='running':f.root.job.status==='completed'))throw fault('quota');return persist();};
|
|
await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'quota');assert.deepEqual(f.root.job.committedSourceKeys,['one','two']);const identities={...f.root.job.stableIds},calls=f.calls.length;broken=false;await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.calls.length,calls);assert.deepEqual(f.root.job.stableIds,identities);
|
|
}
|
|
});
|
|
|
|
// 最后持久化屏障挂起时 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]<sizes[wide]);assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,500);assert.deepEqual(f.calls.flat(),records.map(row=>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();
|
|
const job=f.root.job;assert.equal(job.committedSourceKeys.length,300);let ticks=0,yields=0;
|
|
const image=await sealRecoveryImageAsync(17,1,1,job,{now:()=>ticks++,frameBudgetMs:1,yieldTask:async()=>{yields++;}});
|
|
assert.ok(yields>0);assert.deepEqual(image,sealRecoveryImage(17,1,1,job));
|
|
assert.equal(Object.keys(image.payload.stableIds).length,300);assert.equal(Object.keys(image.payload.bindings).length,300);await f.runner.close();
|
|
});
|
|
|
|
|
|
// 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();
|
|
const pending=f.root.job.pending;assert.equal(pending.length,300);assert.equal(f.root.document.elements.size,0);
|
|
let ticks=0,yields=0;
|
|
const image=await sealRecoveryImageAsync(17,1,1,pending,{now:()=>ticks++,frameBudgetMs:1,yieldTask:async()=>{yields++;}});
|
|
assert.ok(yields>0);assert.deepEqual(image,sealRecoveryImage(17,1,1,pending));
|
|
assert.deepEqual(image.payload[299].dependencies,['absent']);await f.runner.close();
|
|
});
|