docs(tests): explain model preview dependencies and loading completion
This commit is contained in:
@@ -75,6 +75,7 @@ test('cancelled active work retains its concurrency slot; its late cleanup canno
|
||||
const first = deferred(), second = deferred(), entered = []; let calls = 0;
|
||||
const loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { const index = ++calls; entered.push(index); await (index === 1 ? first.promise : second.promise); yield chunk(query); } });
|
||||
const a = loader.acquire(scope); await tick(); const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;
|
||||
// a 已经拒绝,但其生成器仍挂在 first;同键 b 必须以新代次排队,不能提前占第二个执行槽。
|
||||
const b = loader.acquire(scope); await tick(); assert.equal(calls, 1); assert.equal(b.getSnapshot().status, 'queued');
|
||||
assert.throws(() => loader.acquire({ ...scope, canvasId: 'third' }), { code: 'load_capacity' });
|
||||
first.resolve(); await tick(); assert.equal(calls, 2); const peer = loader.acquire(scope); assert.equal(peer.getSnapshot().generation, b.getSnapshot().generation);
|
||||
@@ -93,6 +94,7 @@ test('read cache requires explicit source validation and is bounded by byte LRU
|
||||
async *load(query) { calls.set(query.canvasId, (calls.get(query.canvasId) ?? 0) + 1); yield chunk(query); } });
|
||||
// 缓存用例统一领取、等待及释放;只有显式 allow-validated 才允许考虑已缓存结果。
|
||||
const read = async (canvasId, cached = false) => { const handle = loader.acquire({ ...scope, canvasId }, { cache: cached ? 'allow-validated' : 'fresh' }); await handle.ready; handle.release(); };
|
||||
// 每项声明 2 字节、总预算 4:访问 a 后再放 c 应淘汰 b;容量是协议 byteLength 计数,不是进程堆内存测量。
|
||||
await read('a'); await read('b'); await read('a', true); assert.equal(calls.get('a'), 1); await read('c'); await read('b', true); assert.equal(calls.get('b'), 2);
|
||||
accepted = false; await read('b', true); assert.equal(calls.get('b'), 3); accepted = true; time = 201; await read('b', true); assert.equal(calls.get('b'), 4);
|
||||
time = 0; await read('b', true); assert.equal(calls.get('b'), 5); loader.close();
|
||||
|
||||
@@ -48,6 +48,7 @@ test('Host shares concurrent model reads per identity/workspace but never reuses
|
||||
const abort = new AbortController(), a = host.loadModel(account, workspace, root, { signal: abort.signal, navigationGeneration: 1 }), b = host.loadModel(account, workspace, root, { navigationGeneration: 2 });
|
||||
await tick(); const rejected = assert.rejects(a, { code: 'load_aborted' }); abort.abort(); await rejected; assert.equal(calls[0].signal.aborted, false);
|
||||
gate.resolve(); assert.equal((await b).versions.get(root), 1); assert.equal(calls.length, 1); assert.equal(factories, 1);
|
||||
// 两个订阅均已结束,空闲 canonical 结果不能充当下一次权威读取;新调用应看到版本 2。
|
||||
version = 2; assert.equal((await host.loadModel(account, workspace, root)).versions.get(root), 2); assert.equal(calls.length, 2); assert.equal(host.getWorkspaces().size, 0);
|
||||
} finally { gate.resolve(); await host.close(); }
|
||||
});
|
||||
@@ -86,6 +87,7 @@ test('Host previews share one source but cancelled observers never receive a lat
|
||||
yield part(3,{status:'ready',document:(await import('../../src/lib/canvas-core/document.ts')).createDocument(root,canvases,[first,second]),versions:new Map([[root,1]])},true);
|
||||
};}});
|
||||
const abort=new AbortController();const a=host.loadModel(account,workspace,root,{signal:abort.signal,onPreview:value=>aStates.push(value)}),b=host.loadModel(account,workspace,root,{onPreview:value=>bStates.push(value)});
|
||||
// 等首个元素真正进入预览再取消 A;记录 A 的通知数,放行 B 的后缀后它必须保持不变。
|
||||
try{while(!aStates.some(value=>value.receivedElements===1))await tick();assert.equal(reads,1);assert.equal(aStates.at(-1).document.elements.size,1);const rejected=assert.rejects(a,{code:'load_aborted'});abort.abort();await rejected;const count=aStates.length;gate.resolve();const complete=await b;assert.equal(complete.document.elements.size,2);assert.equal(aStates.length,count);assert.equal(bStates.at(-1).document.elements.size,2);assert.equal(host.getWorkspaces().size,0);}finally{gate.resolve();await host.close();}
|
||||
});
|
||||
// onPreview 抛错经 onObserverError 隔离,不阻止 canonical ready;身份退役后不能继续通知第二次读取的预览。
|
||||
|
||||
@@ -26,6 +26,7 @@ test('partial model waits for entry/owner/port cycles and edge endpoints; publis
|
||||
const preview=createModelPreview(registry);const first=preview.update(head());validateDocument(first.document,registry);assert.equal(first.document.canvases.size,1);
|
||||
const second=preview.update(delta([a,edge],2));assert.equal(second.document.elements.size,0);assert.equal(second.waitingElements,2);assert.strictEqual(second.document,first.document);
|
||||
const third=preview.update(delta([b],3));assert.deepEqual([...third.document.elements.keys()],['B']);assert.equal(third.waitingElements,2);
|
||||
// F 是入口、端口又引用 A;只有这一批到来,canvas/节点/边的依赖闭环才一起满足。
|
||||
const fourth=preview.update(delta([frame],4));validateDocument(fourth.document,registry);assert.equal(fourth.document.elements.size,4);assert.equal(fourth.document.canvases.size,2);assert.equal(fourth.waitingElements,0);assert.equal(fourth.document.elements.get('E').ownerCanvasId,root);
|
||||
assert.equal(first.document.elements.size,0);assert.equal(third.document.elements.size,1);assert.equal(fourth.complete,false);preview.close();assert.throws(()=>preview.update(delta([],4)),{code:'model_preview_closed'});
|
||||
});
|
||||
@@ -53,6 +54,7 @@ test('bad topology and version/source switches cannot silently alter a preview',
|
||||
test('owned preview records do not freeze or retain mutable caller payloads; final completion must match',()=>{
|
||||
const preview=createModelPreview(registry),header=structuredClone(head(1)),element=node('one');preview.update(header);const state=preview.update(delta([element],1,1));
|
||||
element.data.text='changed by caller';header.canvases[0].id='changed';assert.equal(state.document.elements.get('one').data.text,'');assert.equal(state.document.canvases.has(root),true);
|
||||
// 最终输入沿用已被调用者改过的 element;与捕获前缀的正文不同,必须拒绝而不是悄悄覆盖预览。
|
||||
const final=createDocument(root,[{id:root,parent:null},{id:child,parent:{canvasId:root,entryElementId:'F'}}],[element]);assert.throws(()=>preview.update({status:'ready',document:final,versions}),{code:'model_preview_changed'});
|
||||
});
|
||||
// 后续块重复已发布 ID 属于非法输入,并使投影关闭;之后即使给合法新项也不能继续同一预览。
|
||||
|
||||
@@ -29,6 +29,7 @@ const scope={accountId:account,workspaceId:workspace,projectId:null,canvasId:roo
|
||||
// 逐字节传入 Unicode/换行,依次输出接收头、元素 delta、最终 ready;正文/墓碑及总 byteLength 完整,记录冻结。
|
||||
test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{
|
||||
const bytes=encode(records());/* 每批只发一个字节,主动拆开多字节 UTF-8 与记录边界。 */ async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
||||
// collect 会读到生成器 EOF:看到 complete 记录还不够,尾部 framing 校验也必须成功。
|
||||
const chunks=await collect(decodeModelStream(split(),root,registry,signal()));
|
||||
assert.deepEqual(chunks.map(chunk=>[chunk.sequence,chunk.complete,chunk.value.status]),[[0,false,'receiving'],[1,false,'receiving'],[2,true,'ready']]);
|
||||
assert.equal(chunks[1].value.elements[0].data.text,'🙂');assert.equal(chunks[2].value.document.elements.get('a').data.title,'中文');assert.equal(chunks[2].value.document.tombstones.get('old'),2);assert.equal(chunks.reduce((sum,chunk)=>sum+chunk.byteLength,0),bytes.length);assert.ok(Object.isFrozen(chunks[1].value.elements[0]));
|
||||
@@ -63,6 +64,7 @@ test('malformed tails retain read deltas without yielding an authoritative model
|
||||
test('terminal record still requires EOF; trailing garbage rejects rather than opening a baseline',async()=>{
|
||||
const gate={},rows=records();let release;gate.promise=new Promise(resolve=>release=resolve);
|
||||
/* complete 之后仍阻塞迭代器 EOF,之后追加垃圾,检查 ready 不会过早成功。 */ async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
||||
// trailing 先交出终态记录再等待,检测 ready 是否在底层迭代结束前过早解决;放行后追加垃圾使它最终拒绝。
|
||||
const loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:trailing()})}),current:()=>true}),handle=loader.acquire(scope);let ready=false;handle.ready.then(()=>ready=true,()=>{});
|
||||
try{while(handle.getSnapshot().chunks.length<2)await new Promise(resolve=>setImmediate(resolve));assert.equal(ready,false);release();await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');}finally{release();handle.release();loader.close();}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user