docs(tests): explain model streams previews and scoped read caches
This commit is contained in:
@@ -1,24 +1,35 @@
|
||||
/**
|
||||
* 资源显示租约缓存测试。按 owner 和资源键共享进行中的读取/URL,最后租约释放即清理,不保留空闲 URL。
|
||||
* 每个等待者可独立取消;旧读取迟到结果应清理自身,不能替换同键的新代次或破坏其他 owner。
|
||||
*/
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createDisplayCache } from '../../src/lib/canvas-runtime/display-cache.ts';
|
||||
// 手动 gate 可返回成功显示源或抛错,让取消和迟到清理时序可控。
|
||||
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
|
||||
// 推进事件循环等待 load/清理执行,不做真实磁盘或浏览器图片加载。
|
||||
const tick = () => new Promise(resolve => setImmediate(resolve));
|
||||
|
||||
// 两个租约共享一次 load 和同一 URL,release 幂等;最后释放清理后再 acquire 必须重新 load,关闭清理新源。
|
||||
test('concurrent display leases share one source until the last idempotent release; there is no idle URL retention', async () => {
|
||||
const cache = createDisplayCache(), owner = {}; let loads = 0, disposed = 0;
|
||||
// 可观测显示源工厂:每次真实读取产生新 URL,源 release 计数代表底层资源释放。
|
||||
const load = async () => ({ url: `blob:${++loads}`, release: () => disposed++ });
|
||||
const a = cache.acquire(owner, 'asset', load), b = cache.acquire(owner, 'asset', load);
|
||||
assert.deepEqual(await Promise.all([a.ready, b.ready]), ['blob:1', 'blob:1']); assert.equal(loads, 1);
|
||||
a.release(); a.release(); assert.equal(disposed, 0); b.release(); assert.equal(disposed, 1);
|
||||
const c = cache.acquire(owner, 'asset', load); assert.equal(await c.ready, 'blob:2'); c.release(); cache.close(); assert.equal(disposed, 2);
|
||||
});
|
||||
// 取消一个等待租约只拒绝它自己的 ready,共享 AbortSignal 仍有效;最后一个 reader 释放才 abort 并清理。
|
||||
test('cancelling one waiting lease does not abort a shared load or another reader', async () => {
|
||||
const cache = createDisplayCache(), owner = {}, gate = deferred(); let signal, disposed = 0;
|
||||
// 捕获共享读取的 AbortSignal 并挂起,区分释放单个租约与终止整个底层读取。
|
||||
const load = async value => { signal = value; return gate.promise; };
|
||||
const a = cache.acquire(owner, 'asset', load), b = cache.acquire(owner, 'asset', load); await tick();
|
||||
const cancelled = assert.rejects(a.ready, { code: 'display_released' }); a.release(); await cancelled; assert.equal(signal.aborted, false);
|
||||
gate.resolve({ url: 'blob:shared', release: () => disposed++ }); assert.equal(await b.ready, 'blob:shared'); b.release(); assert.equal(signal.aborted, true); assert.equal(disposed, 1);
|
||||
});
|
||||
// 最后旧租约释放即 abort,随后同键可新建;旧源迟到只释放自身,不能替换/逐出新缓存条目。
|
||||
test('last release aborts a load; a late source cannot replace or evict a newly acquired entry', async () => {
|
||||
const cache = createDisplayCache(), owner = {}, gate = deferred(); let signal, oldDisposed = 0, newDisposed = 0;
|
||||
const old = cache.acquire(owner, 'asset', async value => { signal = value; return gate.promise; }); await tick();
|
||||
@@ -27,12 +38,14 @@ test('last release aborts a load; a late source cannot replace or evict a newly
|
||||
gate.resolve({ url: 'blob:old', release: () => oldDisposed++ }); await tick(); assert.equal(oldDisposed, 1); assert.equal(newDisposed, 0);
|
||||
const peer = cache.acquire(owner, 'asset', async () => { throw Error('unexpected duplicate read'); }); assert.equal(await peer.ready, 'blob:new'); peer.release(); next.release(); assert.equal(newDisposed, 1);
|
||||
});
|
||||
// 读取失败保留原错误码并移除失败条目,新 acquire 可重新读取成功,不永久缓存错误。
|
||||
test('read failure retains its cause and removes the failed cache entry for a fresh retry', async () => {
|
||||
const cache = createDisplayCache(), owner = {};
|
||||
const a = cache.acquire(owner, 'asset', async () => { throw Object.assign(Error('disk_failed'), { code: 'disk_failed' }); });
|
||||
await assert.rejects(a.ready, { code: 'disk_failed' });
|
||||
const b = cache.acquire(owner, 'asset', async () => ({ url: '/remote' })); assert.equal(await b.ready, '/remote'); b.release(); cache.close();
|
||||
});
|
||||
// 同资源 ID 不同 owner 各有源;releaseOwner 只关闭自己的作用域并禁止再次领取,不影响另一个 owner 的共享租约。
|
||||
test('project owner scopes isolate identical IDs and retired owners cannot reacquire', async () => {
|
||||
const cache = createDisplayCache(), a = {}, b = {}; let disposed = 0;
|
||||
const first = cache.acquire(a, 'same-id', async () => ({ url: 'blob:a', release: () => disposed++ }));
|
||||
@@ -41,6 +54,7 @@ test('project owner scopes isolate identical IDs and retired owners cannot reacq
|
||||
assert.throws(() => cache.acquire(a, 'same-id', async () => ({ url: '/forbidden' })), { code: 'display_scope_closed' });
|
||||
const peer = cache.acquire(b, 'same-id', async () => { throw Error('unexpected'); }); assert.equal(await peer.ready, 'blob:b'); peer.release(); second.release(); assert.equal(disposed, 2);
|
||||
});
|
||||
// close 立即拒绝挂起 ready,迟到源仍需释放;关闭后新 acquire 拒绝,多次 close 安全。
|
||||
test('cache close rejects waiting readers immediately and disposes late data without reopening a scope', async () => {
|
||||
const cache = createDisplayCache(), gate = deferred(); let disposed = 0;
|
||||
const lease = cache.acquire({}, 'asset', () => gate.promise); await tick();
|
||||
@@ -48,6 +62,7 @@ test('cache close rejects waiting readers immediately and disposes late data wit
|
||||
gate.resolve({ url: 'blob:late', release: () => disposed++ }); await tick(); assert.equal(disposed, 1);
|
||||
assert.throws(() => cache.acquire({}, 'asset', async () => ({ url: '/late' })), { code: 'display_scope_closed' }); cache.close();
|
||||
});
|
||||
// 空 URL 等非法源需清理已分配资源;某个 release 抛错通过 onError 报告,不能阻止其他 owner 的清理。
|
||||
test('invalid sources clean their allocation and cleanup errors do not interrupt other owners', async () => {
|
||||
const errors = [], cache = createDisplayCache({ onError: error => errors.push(error) }); let disposed = 0;
|
||||
const invalid = cache.acquire({}, 'invalid', async () => ({ url: '', release: () => disposed++ })); await assert.rejects(invalid.ready, { code: 'invalid_display_source' }); assert.equal(disposed, 1);
|
||||
|
||||
@@ -1,11 +1,21 @@
|
||||
/**
|
||||
* 作用域加载协调器测试:多个订阅者可共享同一渐进源,但各自保留导航元数据、取消状态和快照生命周期。
|
||||
* 块需绑定 query/权限/project/snapshot/version/sequence,完整标记后仍须正常 EOF;错误尾部只能保留只读前缀。
|
||||
* 活动并发、排队、单响应字节/块数及校验后缓存分别受限,旧任务迟到清理不能删除新代次。
|
||||
*/
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createScopedLoader, loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
|
||||
// 完整查询作用域:账户、workspace、project/canvas、格式、过滤指纹、来源及权限/导航代次;导航代次不阻止共享源。
|
||||
const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project', canvasId: 'canvas', viewKey: 'canvas', format: 'v1', queryFingerprint: 'all', source: 'fixture', authEpoch: 1, navigationGeneration: 1 };
|
||||
// 可手动 resolve/reject 的 Promise,用来暂停源迭代,检查取消与新代次的关系。
|
||||
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
|
||||
// 让出至 setImmediate,使源执行及清理有机会推进,不模拟实际网络延迟。
|
||||
const tick = () => new Promise(resolve => setImmediate(resolve));
|
||||
// 构造绑定查询键的合法加载块,extra 专供逐字段篡改;覆盖信息按序累积,value 使用冻结样本。
|
||||
const chunk = (query, sequence = 0, complete = true, extra = {}) => ({ queryKey: loadQueryKey(query), projectId: 'project', snapshotToken: 'snapshot', serverVersions: { canvas: 3 }, permissionEpoch: query.authEpoch, sequence, coverage: [`part-${sequence}`], complete, byteLength: 2, value: Object.freeze({ part: sequence }), ...extra });
|
||||
|
||||
// 两个导航订阅共用一次渐进读取;取消 A 不取消 B,A 保留取消时前缀且不再更新,B 收齐覆盖并完成。
|
||||
test('subscribers share a progressive read while retaining navigation scopes; cancelling one preserves the other', async () => {
|
||||
const gate = deferred(), updates = []; let calls = 0, signal;
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query, abort) { calls++; signal = abort; yield chunk(query, 0, false); await gate.promise; yield chunk(query, 1); } });
|
||||
@@ -19,41 +29,48 @@ test('subscribers share a progressive read while retaining navigation scopes; ca
|
||||
assert.strictEqual(a.getSnapshot(),stopped);assert.strictEqual(b.getSnapshot(),result);
|
||||
assert.deepEqual(updates, ['queued', 'loading', 'initial-ready']); b.release(); loader.close();
|
||||
});
|
||||
// 尾部失败保留已接受首块为 failed-with-partial;重新 acquire 必须启动新源,不复用失败 inflight。
|
||||
test('tail failure keeps the accepted prefix and a fresh retry replaces the failed inflight entry', async () => {
|
||||
let calls = 0;
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query, 0, false); if (calls === 1) throw Object.assign(Error('tail_failed'), { code: 'tail_failed' }); yield chunk(query, 1); } });
|
||||
const a = loader.acquire(scope); await assert.rejects(a.ready, { code: 'tail_failed' }); assert.equal(a.getSnapshot().status, 'failed-with-partial'); assert.equal(a.getSnapshot().chunks[0].value.part, 0);
|
||||
const b = loader.acquire(scope); assert.equal((await b.ready).chunks.length, 2); assert.equal(calls, 2); a.release(); b.release(); loader.close();
|
||||
});
|
||||
// 源先给 complete 再抛尾部校验错误,ready 必须拒绝且观察者从未收到 complete 状态。
|
||||
test('a complete marker is not authoritative before iterator EOF; a checksum/trailing failure retains partial status', async () => {
|
||||
const statuses = [], loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query); throw Error('trailing_checksum'); } });
|
||||
const handle = loader.acquire(scope, { onUpdate: value => statuses.push(value.status) }); await assert.rejects(handle.ready, /trailing_checksum/);
|
||||
assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(statuses.includes('complete'), false); handle.release(); loader.close();
|
||||
});
|
||||
// 尾块换 snapshotToken、版本向量或 project 时拒绝,前缀 chunks/coverage 保留原值,不混入另一快照。
|
||||
test('snapshot token, version vector and project changes reject the tail without mixing it into accepted coverage', async () => {
|
||||
for (const extra of [{ snapshotToken: 'other' }, { serverVersions: { canvas: 4 } }, { projectId: 'other' }]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1, true, extra); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(handle.getSnapshot().chunks.length, 1); assert.deepEqual(handle.getSnapshot().coverage, ['part-0']); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
// 错误 queryKey、权限代次、序号、重复覆盖项、负字节数或非法版本在该块暴露前拒绝,结果为空 failed。
|
||||
test('query, permission, sequence and malformed coverage headers reject before exposing a chunk', async () => {
|
||||
for (const extra of [{ queryKey: 'foreign' }, { permissionEpoch: 2 }, { sequence: 3 }, { coverage: ['x', 'x'] }, { byteLength: -1 }, { serverVersions: { canvas: 0 } }]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, true, extra); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'invalid_load_chunk' }); assert.equal(handle.getSnapshot().chunks.length, 0); assert.equal(handle.getSnapshot().status, 'failed'); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
// EOF 前没有 complete,或 complete 之后又多一块,都不能成为完整结果,已有内容只作为失败前缀保留。
|
||||
test('an incomplete EOF or extra chunk after a complete marker cannot become a complete result', async () => {
|
||||
for (const extra of [false, true]) {
|
||||
const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, extra); if (extra) yield chunk(query, 1); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
// 响应字节数或块数上限阻止接受第二块;保留首块计数,不能为满足预算静默丢块还宣称成功。
|
||||
test('response byte and chunk budgets bound retained read copies while preserving earlier chunks', async () => {
|
||||
for (const limits of [{ maxResponseBytes: 3 }, { maxChunks: 1 }]) {
|
||||
const loader = createScopedLoader({ ...limits, current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1); } });
|
||||
const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'load_response_budget' }); assert.equal(handle.getSnapshot().byteLength, 2); assert.equal(handle.getSnapshot().chunks.length, 1); handle.release(); loader.close();
|
||||
}
|
||||
});
|
||||
// 活动任务取消后仍占计算/读取槽位;旧迭代真正退出才启动新代次,其清理不能从共享表移除新任务。
|
||||
test('cancelled active work retains its concurrency slot; its late cleanup cannot delete the newer generation', async () => {
|
||||
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); } });
|
||||
@@ -63,30 +80,36 @@ test('cancelled active work retains its concurrency slot; its late cleanup canno
|
||||
first.resolve(); await tick(); assert.equal(calls, 2); const peer = loader.acquire(scope); assert.equal(peer.getSnapshot().generation, b.getSnapshot().generation);
|
||||
second.resolve(); await Promise.all([b.ready, peer.ready]); assert.deepEqual(entered, [1, 2]); b.release(); peer.release(); loader.close();
|
||||
});
|
||||
// 等待中的订阅取消可立即释放排队名额,放弃的 source 不执行;后续查询按空出的队列位置进入。
|
||||
test('queued cancellation frees admission capacity and never executes the abandoned source', async () => {
|
||||
const gate = deferred(), called = [], loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { called.push(query.canvasId); await gate.promise; yield chunk(query); } });
|
||||
const a = loader.acquire(scope), b = loader.acquire({ ...scope, canvasId: 'cancelled' }); const cancelled = assert.rejects(b.ready, { code: 'load_aborted' }); b.release(); await cancelled;
|
||||
const c = loader.acquire({ ...scope, canvasId: 'next' }); gate.resolve(); await Promise.all([a.ready, c.ready]); assert.deepEqual(called, ['canvas', 'next']); a.release(); c.release(); loader.close();
|
||||
});
|
||||
// 已完成读取缓存受字节 LRU、显式 source validation 和权限时间限制;不认可、过期及墙上时钟回退均重新读取。
|
||||
test('read cache requires explicit source validation and is bounded by byte LRU and permission age', async () => {
|
||||
let time = 100, accepted = true; const calls = new Map();
|
||||
const loader = createScopedLoader({ maxCacheEntries: 10, maxCacheBytes: 4, cacheMaxAgeMs: 100, now: () => time, acceptCached: () => accepted, current: () => true,
|
||||
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(); };
|
||||
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();
|
||||
});
|
||||
// 相同 canvas 的不同过滤指纹或权限 epoch 不能共享请求,三个查询各自加载。
|
||||
test('different query fingerprints and permission epochs do not share requests', async () => {
|
||||
let calls = 0; const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query); } });
|
||||
const handles = [scope, { ...scope, queryFingerprint: 'filtered' }, { ...scope, authEpoch: 2 }].map(query => loader.acquire(query));
|
||||
await Promise.all(handles.map(handle => handle.ready)); assert.equal(calls, 3); handles.forEach(handle => handle.release()); loader.close();
|
||||
});
|
||||
// 捕获 scope 只接管声明字段并冻结副本;外部额外对象仍可变,输入 canvas 后改不影响已领取查询。
|
||||
test('scope capture owns only declared metadata without freezing unrelated caller state', async () => {
|
||||
const extra={filter:'mutable'},input={...scope,extra};const loader=createScopedLoader({current:()=>true,async *load(query){yield chunk(query);}});
|
||||
const handle=loader.acquire(input);assert.equal(Object.isFrozen(extra),false);input.canvasId='later';extra.filter='changed';
|
||||
const result=await handle.ready;assert.equal(result.scope.canvasId,'canvas');assert.equal(Object.hasOwn(result.scope,'extra'),false);assert.equal(Object.isFrozen(result.scope),true);handle.release();loader.close();
|
||||
});
|
||||
// 身份失效拒绝迟到块;close 即使源不理 abort 也立即拒绝等待者,迟到完成不能重开 loader。
|
||||
test('identity changes discard late chunks, and close promptly rejects waiting readers even if the source ignores abort', async () => {
|
||||
const gate = deferred(); let current = true;
|
||||
const loader = createScopedLoader({ current: () => current, async *load(query) { await gate.promise; yield chunk(query); } });
|
||||
@@ -94,6 +117,7 @@ test('identity changes discard late chunks, and close promptly rejects waiting r
|
||||
const pending = deferred(), closed = createScopedLoader({ current: () => true, async *load(query) { await pending.promise; yield chunk(query); } });
|
||||
const b = closed.acquire(scope); const stopped = assert.rejects(b.ready, { code: 'loader_closed' }); closed.close(); await stopped; pending.resolve(); await tick(); assert.equal(b.getSnapshot().status, 'cancelled'); assert.throws(() => closed.acquire(scope), { code: 'loader_closed' });
|
||||
});
|
||||
// 观察者回调重入 acquire 不应重通知旧订阅者;新观察者抛错通过专用错误口收集,不使源读取失败。
|
||||
test('observer reentrancy does not re-notify old subscribers or allow observer exceptions to fail a read', async () => {
|
||||
const errors = []; let peer, updates = 0;
|
||||
const loader = createScopedLoader({ current: () => true, onObserverError: error => errors.push(error), async *load(query) { yield chunk(query); } });
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
/**
|
||||
* Foundry 模型 loader、RuntimeHost 共享读取及画布 entry 注入测试。只读加载不应提前打开 workspace/project 存储。
|
||||
* 并发请求可以共享进行中的读取,但空闲 canonical 快照不自动成为下次打开的权威基线;身份退役隔离迟到预览。
|
||||
*/
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createRegistry, builtinComponents } from '../../src/lib/canvas-core/index.ts';
|
||||
@@ -5,13 +9,20 @@ import { createRuntimeHost } from '../../src/lib/canvas-runtime/host.ts';
|
||||
import { loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
|
||||
import { createFoundryModelLoader } from '../../src/lib/canvas-service/model-loader.ts';
|
||||
import { loadCanvasEntry } from '../../src/lib/canvas-entry.ts';
|
||||
// 确定性 UUID 工厂,满足 service 的输入校验并区分账户、workspace 与根 canvas。
|
||||
const id = n => `10000000-0000-4000-8000-${String(n).padStart(12, '0')}`;
|
||||
// 当前身份和 registry;scope 中 projectId 尚未知,由 canonical 模型解析得到真实项目根。
|
||||
const account = id(1), workspace = id(2), root = id(3), registry = createRegistry(builtinComponents());
|
||||
// 完整模型查询绑定权限 epoch、导航代次、格式和来源,防止不同读取用途误共享。
|
||||
const scope = { accountId: account, workspaceId: workspace, projectId: null, canvasId: root, authEpoch: 1, navigationGeneration: 7, format: 'canonical-model-v1', source: 'foundry', viewKey: 'canonical-project', queryFingerprint: 'full-project' };
|
||||
// 手动 gate 控制首次响应及尾部预览返回,精确安排取消/身份退役。
|
||||
const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; };
|
||||
// 推进一次事件循环,让共享 loader 和观察者执行至可检查的状态。
|
||||
const tick = () => new Promise(resolve => setImmediate(resolve));
|
||||
// 生成版本可变的完整 canonical 响应,标题可用于检测是否误用了旧账户结果。
|
||||
const reply = (version = 1, title = account) => ({ status: 200, body: { model: { documentFormatVersion: 1, projectId: root, complete: true, canvases: [{ id: root, parent: null }], elements: [{ id: 'node', incarnation: 1, ownerCanvasId: root, componentId: 'text', dataVersion: 1, structure: 'node', data: { title, text: '' }, geometry: { x: 0, y: 0, width: 100, height: 80, rotation: 0 } }], tombstones: {}, versions: { [root]: version }, removedCanvasIds: [] } } });
|
||||
|
||||
// JSON adapter 只发一个已验证完整块,绑定 query/project、canvas 覆盖和版本,GET 沿用原 AbortSignal。
|
||||
test('Foundry adapter emits one validated complete chunk with bound query, project, coverage and version metadata', async () => {
|
||||
const abort = new AbortController(), calls = [];
|
||||
const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async input => { calls.push(input); return reply(3); } });
|
||||
@@ -20,12 +31,14 @@ test('Foundry adapter emits one validated complete chunk with bound query, proje
|
||||
assert.deepEqual(chunks[0].coverage, [`canvas:${root}`]); assert.equal(chunks[0].serverVersions[root], 3); assert.ok(chunks[0].byteLength > 0); assert.equal(chunks[0].value.document.elements.get('node').data.title, account);
|
||||
assert.strictEqual(calls[0].signal, abort.signal); assert.equal(calls[0].method, 'GET');
|
||||
});
|
||||
// 局部模型、查询 canvas 不在模型及不匹配迁移头均不能冒充初始就绪,必须在 yield 任何块前拒绝。
|
||||
test('partial canonical replies, foreign requested canvases and unbound migration headers cannot masquerade as initial readiness', async () => {
|
||||
for (const [response, expected] of [[{ ...reply(), body: { model: { ...reply().body.model, complete: false } } }, 'incomplete_initial_model'], [reply(), 'project_mismatch'], [{ status: 409, body: { error: 'canonical_migration_required', document_format_version: 1, root_id: root, source_versions: { [root]: 1 } } }, 'project_mismatch']]) {
|
||||
const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async () => response }), query = expected === 'incomplete_initial_model' ? scope : { ...scope, canvasId: id(99) };
|
||||
await assert.rejects(async () => { for await (const _ of port(query, new AbortController().signal)) assert.fail('invalid response yielded a chunk'); }, { code: expected });
|
||||
}
|
||||
});
|
||||
// 同身份/workspace 并发读取共享端口,取消一方不伤另一方;完成释放后下次重新读取新版本,始终不打开 workspace runtime。
|
||||
test('Host shares concurrent model reads per identity/workspace but never reuses an idle authoritative snapshot', async () => {
|
||||
const gate = deferred(), calls = []; let factories = 0, version = 1;
|
||||
const host = createRuntimeHost({ registry, workspace() { assert.fail('readonly load opened a project/storage runtime'); }, modelLoader(accountId, workspaceId, _epoch, current) {
|
||||
@@ -38,6 +51,7 @@ test('Host shares concurrent model reads per identity/workspace but never reuses
|
||||
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(); }
|
||||
});
|
||||
// 身份退役关闭旧 loader 并 abort 原信号,新账户用新读取;旧响应迟到不能覆盖当前 Host 身份或模型。
|
||||
test('Host retirement aborts the old read and a late response cannot pollute the new account reader', async () => {
|
||||
const gate = deferred(), calls = [];
|
||||
const host = createRuntimeHost({ registry, workspace() { assert.fail('unexpected project open'); }, modelLoader(accountId, workspaceId) {
|
||||
@@ -49,6 +63,7 @@ test('Host retirement aborts the old read and a late response cannot pollute the
|
||||
gate.resolve(); await tick(); assert.equal(host.identity().accountId, id(8)); assert.equal(calls[0].input.signal.aborted, true); assert.equal(calls.length, 2);
|
||||
} finally { gate.resolve(); await host.close(); }
|
||||
});
|
||||
// entry 解析 slug 后传入真实 account/workspace/canvas 与导航上下文,走注入 readModel;canonical 失败不能偷退回旧 canvas 接口。
|
||||
test('entry injection uses resolved IDs and navigation context while preserving canonical-first legacy policy', async () => {
|
||||
const board = { id: root, workspace_id: workspace, slug: 'board', kind: 'board', title: 'Board', version: 1 }, inputs = [], reads = [];
|
||||
const options = { workspaceRef: 'workspace-slug', boardRef: 'board', registry, navigationGeneration: 9, request: async path => { reads.push(path); return path === 'profile' ? { id: account } : board; },
|
||||
@@ -57,10 +72,12 @@ test('entry injection uses resolved IDs and navigation context while preserving
|
||||
await assert.rejects(() => loadCanvasEntry({ ...options, readModel: async () => { throw Object.assign(Error('forbidden'), { code: 'forbidden' }); } }), { code: 'forbidden' });
|
||||
});
|
||||
|
||||
// 两个预览观察者共享一个渐进源,取消一个后它不再收到后缀;另一方完成,Host 未创建任何存储 workspace。
|
||||
test('Host previews share one source but cancelled observers never receive a late prefix or open storage',async()=>{
|
||||
const gate=deferred(),aStates=[],bStates=[];let reads=0;
|
||||
const first=reply().body.model.elements[0],second={...first,id:'second'},canvases=[{id:root,parent:null}],serverVersions={[root]:1};
|
||||
const host=createRuntimeHost({registry,workspace(){assert.fail('preview opened storage');},modelLoader(){return async function*(scope){
|
||||
// 每个原始读取共用 token/版本;part 将状态包装成连续序号块,gate 只阻塞后缀。
|
||||
reads++;const part=(sequence,value,complete=false)=>({queryKey:loadQueryKey(scope),projectId:root,snapshotToken:'one-read',serverVersions,permissionEpoch:scope.authEpoch,sequence,coverage:[],complete,byteLength:1,value});
|
||||
yield part(0,{status:'receiving',projectId:root,canvases,elements:[],tombstones:new Map(),versions:new Map([[root,1]]),receivedElements:0,totalElements:2});
|
||||
yield part(1,{status:'receiving',projectId:root,canvases:[],elements:[first],tombstones:new Map(),versions:new Map([[root,1]]),receivedElements:1,totalElements:2});
|
||||
@@ -71,9 +88,11 @@ test('Host previews share one source but cancelled observers never receive a lat
|
||||
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)});
|
||||
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;身份退役后不能继续通知第二次读取的预览。
|
||||
test('preview callback failures do not stop canonical loading and identity retirement fences future previews',async()=>{
|
||||
const gate=deferred(),secondGate=deferred(),errors=[],states=[];let reading=0;
|
||||
const host=createRuntimeHost({registry,workspace(){assert.fail('preview opened a runtime');},onObserverError:error=>errors.push(error),modelLoader(){return async function*(scope){
|
||||
// 两次查询使用不同 gate,分别检查回调异常隔离和身份退役;每次源仍有独立接收/ready 阶段。
|
||||
const pass=++reading;const part=(sequence,value,complete=false)=>({queryKey:loadQueryKey(scope),projectId:root,snapshotToken:'read',serverVersions:{[root]:1},permissionEpoch:scope.authEpoch,sequence,coverage:[],complete,byteLength:1,value});
|
||||
yield part(0,{status:'receiving',projectId:root,canvases:[{id:root,parent:null}],elements:[],tombstones:new Map(),versions:new Map([[root,1]]),receivedElements:0,totalElements:0});await(pass===1?gate.promise:secondGate.promise);
|
||||
yield part(1,{status:'ready',document:(await import('../../src/lib/canvas-core/document.ts')).createDocument(root,[{id:root,parent:null}]),versions:new Map([[root,1]])},true);
|
||||
|
||||
@@ -1,15 +1,27 @@
|
||||
/**
|
||||
* 渐进 canonical 模型的只读预览投影测试。接收进度与合法可发布前缀不同:入口/归属/端口/端点依赖未闭合时先等待。
|
||||
* 每次发布需是合法文档,失败保持上次快照并关闭预览;最终完整模型还需与已显示内容一致,不能直接变成可编辑半模型。
|
||||
*/
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createRegistry,builtinComponents,createDocument,validateDocument} from '../../src/lib/canvas-core/index.ts';
|
||||
import {createModelPreview} from '../../src/lib/canvas-runtime/model-preview.ts';
|
||||
// 内置类型定义及 root/child 身份;frame 端口与 child 成员形成依赖闭环。
|
||||
const registry=createRegistry(builtinComponents()),root='root',child='child';
|
||||
// 文本节点工厂,可替换 owner,用于构造已就绪节点、等待成员及非法记录。
|
||||
const node=(id,ownerCanvasId=root)=>({id,incarnation:1,ownerCanvasId,componentId:'text',dataVersion:1,structure:'node',data:{title:id,text:''},geometry:{x:0,y:0,width:100,height:80,rotation:0}});
|
||||
// F 在 root,端口引用 child 中 A;child 又以 F 为入口,必须闭合后一起显示。
|
||||
const frame={...node('F'),componentId:'frame',data:{title:'F',text:'',ports:[{id:'p',nodeId:'A',direction:'output',handle:'right'}]}};
|
||||
// A 与 B 作为跨分区连线端点;E 归共同 root,不能在任一端点缺失时显示成完整边。
|
||||
const a=node('A',child),b=node('B'),edge={id:'E',incarnation:1,ownerCanvasId:root,componentId:'connection',dataVersion:1,structure:'connection',data:{},source:{elementId:'A'},target:{elementId:'B'},path:null};
|
||||
// 源 canvas 树和版本表;预览开始时虽已知 child 关系,入口尚未来到仍不能发布 child。
|
||||
const canvases=[{id:root,parent:null},{id:child,parent:{canvasId:root,entryElementId:'F'}}],versions=new Map([[root,1],[child,1]]);
|
||||
// 空接收头声明总元素数,complete 仍由最终 ready 状态决定,不由 received==total 推定。
|
||||
const head=(total=4)=>({status:'receiving',projectId:root,canvases,elements:[],tombstones:new Map(),versions,receivedElements:0,totalElements:total});
|
||||
// 后续 delta 不重复 canvas 头,只带新元素和接收计数,测试增量累积而非整文档替换。
|
||||
const delta=(elements,received,total=4)=>({...head(total),canvases:[],elements,receivedElements:received});
|
||||
|
||||
// 先收 A/E 仍等待,独立 B 可先显示;F 到来才闭合 child/端口/连线并发布合法前缀,旧文档不变,收到全部元素仍不等于 complete。
|
||||
test('partial model waits for entry/owner/port cycles and edge endpoints; publishes one closed legal prefix',()=>{
|
||||
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);
|
||||
@@ -17,32 +29,38 @@ test('partial model waits for entry/owner/port cycles and edge endpoints; publis
|
||||
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'});
|
||||
});
|
||||
// 未知组件保持不透明只读候选,不猜测私有引用;已知 frame 声明端口缺目标时必须继续等待。
|
||||
test('declared external reference waits; unsupported component data remains an opaque read-only candidate',()=>{
|
||||
const preview=createModelPreview(registry);preview.update(head(2));const future={...node('future'),componentId:'plugin.future',dataVersion:99,data:{opaque:{ref:'absent',assetId:'secret'}}};
|
||||
const shown=preview.update(delta([future],1,2));assert.deepEqual(shown.document.elements.get('future'),future);assert.equal(registry.resolve(shown.document.elements.get('future')),'unknown-component');
|
||||
const delayed={...node('delayed'),componentId:'frame',data:{title:'delayed',text:'',ports:[{nodeId:'absent'}]}};
|
||||
const next=preview.update(delta([delayed],2,2));assert.equal(next.document.elements.size,1);assert.equal(next.waitingElements,1);
|
||||
});
|
||||
// 同批 good+非法几何/身份/owner 或重复 ID 时整体拒绝,不能先发布 good,也不能替换上一合法快照。
|
||||
test('invalid batches do not replace the last legal prefix or publish a partial batch',()=>{
|
||||
for(const bad of [{...node('bad'),geometry:{...node('bad').geometry,width:-1}},{...node('bad'),incarnation:0},{...node('bad'),ownerCanvasId:'foreign'}]){
|
||||
const preview=createModelPreview(registry),first=preview.update(head(2));assert.throws(()=>preview.update(delta([node('good'),bad],2,2)));assert.strictEqual(preview.getSnapshot(),first);assert.equal(first.document.elements.size,0);
|
||||
}
|
||||
const preview=createModelPreview(registry),first=preview.update(head(2));assert.throws(()=>preview.update(delta([node('same'),node('same')],2,2)));assert.strictEqual(preview.getSnapshot(),first);
|
||||
});
|
||||
// 循环/多根 canvas 拓扑和跨 project 的后续输入拒绝,不能混合成同一个预览。
|
||||
test('bad topology and version/source switches cannot silently alter a preview',()=>{
|
||||
for(const topology of [[{id:root,parent:{canvasId:child,entryElementId:'F'}},{id:child,parent:{canvasId:root,entryElementId:'G'}}],[{id:root,parent:null},{id:child,parent:null}]])assert.throws(()=>createModelPreview(registry).update({...head(),canvases:topology}));
|
||||
const preview=createModelPreview(registry);preview.update(head());assert.throws(()=>preview.update({...delta([b],1),projectId:'other'}));
|
||||
});
|
||||
|
||||
// 预览接管元素/头部数据副本,调用者后改不影响已显示内容;最终模型若与前缀不符,报 model_preview_changed。
|
||||
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);
|
||||
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 属于非法输入,并使投影关闭;之后即使给合法新项也不能继续同一预览。
|
||||
test('published identities cannot be repeated in later chunks and invalid input closes the projection',()=>{
|
||||
const preview=createModelPreview(registry);preview.update({...head(2),canvases:[{id:root,parent:null}]});preview.update(delta([node('one')],1,2));assert.throws(()=>preview.update(delta([node('one')],2,2)));assert.throws(()=>preview.update(delta([node('two')],2,2)),{code:'model_preview_closed'});
|
||||
});
|
||||
|
||||
// canvasElementCounts 统计收到记录,不把依赖等待者当已发布;证书先防御复制,单 canvas 数量不符在发布前拒绝。
|
||||
test('scope certificates count received records without treating dependency-blocked members as published',()=>{
|
||||
const preview=createModelPreview(registry);const counts={[root]:3,[child]:1};preview.update({...head(),canvasElementCounts:counts});counts[child]=0;
|
||||
const waiting=preview.update(delta([a,edge],2));assert.equal(waiting.document.canvases.has(child),false);assert.equal(waiting.canvasElementCounts[child],1);assert.equal(waiting.waitingElements,2);
|
||||
@@ -50,6 +68,7 @@ test('scope certificates count received records without treating dependency-bloc
|
||||
const invalid=createModelPreview(registry),first=invalid.update({...head(1),canvasElementCounts:{[root]:0,[child]:1}});assert.throws(()=>invalid.update(delta([b],1,1)),{code:'invalid_model_coverage'});assert.strictEqual(invalid.getSnapshot(),first);
|
||||
});
|
||||
|
||||
// elementOwners 清单包含尚未显示的 ID,按捕获归属校验后续记录;外部改清单无效,未声明 ID 的块不能发布。
|
||||
test('custom loader identity manifests are owned and checked before a receiving batch can publish',()=>{
|
||||
const preview=createModelPreview(registry),elementOwners={F:root,E:root,B:root,A:child};const first=preview.update({...head(),canvasElementCounts:{[root]:3,[child]:1},elementOwners});elementOwners.A=root;assert.equal(first.elementOwners.A,child);assert.deepEqual(first.sourceCanvases,head().canvases);assert.equal(first.document.canvases.has(child),false);assert.ok(Object.isFrozen(first.sourceCanvases));const before=preview.getSnapshot();assert.throws(()=>preview.update(delta([node('not-in-capture')],1)),{code:'invalid_model_identity'});assert.strictEqual(preview.getSnapshot(),before);
|
||||
});
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
/**
|
||||
* Foundry 模型流 framing、完整校验及 transport/loader 协调测试。流可以先交付只读 delta,完整文档必须等终态记录与正常 EOF。
|
||||
* 快照/序号/覆盖/身份清单绑定所有后续记录,错误尾部保留可读前缀但不能打开为权威基线。
|
||||
*/
|
||||
import {test} from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {createRegistry,builtinComponents} from '../../src/lib/canvas-core/index.ts';
|
||||
@@ -5,28 +9,39 @@ import {decodeModelStream,modelStreamType} from '../../src/lib/canvas-service/mo
|
||||
import {createFoundryTransport} from '../../src/lib/canvas-service/foundry.ts';
|
||||
import {createFoundryModelLoader} from '../../src/lib/canvas-service/model-loader.ts';
|
||||
import {createScopedLoader} from '../../src/lib/canvas-runtime/loader.ts';
|
||||
// 固定 UUID 作用域,满足 transport 与模型解析的身份约束。
|
||||
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003';
|
||||
// 内置定义和独立信号工厂,避免取消跨用例复用。
|
||||
const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal;
|
||||
// 含中文及 emoji 的合法节点样本,验证逐字节传输时 UTF-8 边界处理。
|
||||
const node={id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:'中文',text:'🙂'},geometry:{x:0,y:0,width:100,height:80,rotation:0}};
|
||||
// 每次新建 header/elements/complete 记录,快照 token 相同、序号连续且声明一个元素;便于逐字段篡改。
|
||||
const records=()=>[{streamVersion:1,snapshotToken:'snapshot-a',sequence:0,kind:'header',documentFormatVersion:1,projectId:root,canvases:[{id:root,parent:null}],versions:{[root]:3},tombstones:{old:2},removedCanvasIds:[],elementCount:1},{streamVersion:1,snapshotToken:'snapshot-a',sequence:1,kind:'elements',elements:[structuredClone(node)]},{streamVersion:1,snapshotToken:'snapshot-a',sequence:2,kind:'complete',elementCount:1}];
|
||||
// 编码为带末尾换行的 UTF-8 NDJSON;byteLength 断言以真实传输字节为准。
|
||||
const encode=rows=>new TextEncoder().encode(rows.map(row=>JSON.stringify(row)).join('\n')+'\n');
|
||||
// 单批字节异步源,不模拟网络切片,逐字节场景使用另一个 split generator。
|
||||
async function* source(rows){yield encode(rows);}
|
||||
// 排空异步迭代器到数组,让尾部或 EOF 校验错误也传播到测试断言。
|
||||
const collect=async iterable=>{const result=[];for await(const item of iterable)result.push(item);return result;};
|
||||
// canonical 完整项目查询作用域,交给通用 scoped loader 检查权限和查询绑定。
|
||||
const scope={accountId:account,workspaceId:workspace,projectId:null,canvasId:root,authEpoch:1,navigationGeneration:1,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry'};
|
||||
|
||||
// 逐字节传入 Unicode/换行,依次输出接收头、元素 delta、最终 ready;正文/墓碑及总 byteLength 完整,记录冻结。
|
||||
test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{
|
||||
const bytes=encode(records());async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
||||
const bytes=encode(records());/* 每批只发一个字节,主动拆开多字节 UTF-8 与记录边界。 */ async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
||||
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]));
|
||||
});
|
||||
// 快照切换、序号缺口、计数不符、重复 ID、错误 owner/几何/版本/拓扑、缺少结束及缺末换行都不能完成,超大行按流预算拒绝。
|
||||
test('snapshot switches, sequence gaps, duplicate IDs, wrong counts, late records and invalid topology cannot finish',async()=>{
|
||||
const cases=[rows=>rows[1].snapshotToken='other',rows=>rows[1].sequence=2,rows=>rows[0].elementCount=2,rows=>rows[2].elementCount=2,rows=>rows.push({...rows[2],sequence:3}),rows=>rows[1].elements.push(rows[1].elements[0]),rows=>rows[1].elements[0].ownerCanvasId=workspace,rows=>rows[1].elements[0].geometry.width=-1,rows=>rows[0].versions={},rows=>rows[0].canvases[0].parent={canvasId:root,entryElementId:'a'}];
|
||||
for(const mutate of cases){const rows=records();mutate(rows);await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));}
|
||||
for(const rows of [[],records().slice(0,1),records().slice(0,2)])await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));
|
||||
async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'});
|
||||
async function* huge(){yield new Uint8Array(8*1024*1024+1).fill(65);}await assert.rejects(()=>collect(decodeModelStream(huge(),root,registry,signal())),{code:'model_stream_budget'});
|
||||
/* 删除文件末换行,验证终态记录也不能绕过 framing 完整性。 */ async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'});
|
||||
/* 单行超过 8 MiB,没有换行,必须在无限积累前触发预算。 */ async function* huge(){yield new Uint8Array(8*1024*1024+1).fill(65);}await assert.rejects(()=>collect(decodeModelStream(huge(),root,registry,signal())),{code:'model_stream_budget'});
|
||||
});
|
||||
// transport 以 Accept 协商流并交出 ReadableStream;尾部 gate 前 loader 已看到首个元素 delta,放行后才 complete。
|
||||
test('Foundry transport negotiates streaming without buffering; coordinator retains first delta before gated tail',async()=>{
|
||||
let release,opened;const gate=new Promise(resolve=>release=resolve),started=new Promise(resolve=>opened=resolve);let requested;
|
||||
const rows=records(),stream=new ReadableStream({async start(controller){controller.enqueue(encode(rows.slice(0,2)));opened();await gate;controller.enqueue(encode(rows.slice(2)));controller.close();}});
|
||||
@@ -38,29 +53,34 @@ test('Foundry transport negotiates streaming without buffering; coordinator reta
|
||||
release();const done=await handle.ready;assert.equal(done.status,'complete');assert.equal(done.chunks.at(-1).value.document.elements.size,1);
|
||||
}finally{release();handle.release();loader.close();}
|
||||
});
|
||||
// 坏 JSON 尾部使 loader failed-with-partial,保留两个 receiving 块,不能输出权威模型。
|
||||
test('malformed tails retain read deltas without yielding an authoritative model',async()=>{
|
||||
async function* broken(){yield encode(records().slice(0,2));yield new TextEncoder().encode('{broken}\n');}
|
||||
/* 合法头/元素后追加坏 JSON,保留可读前缀并拒绝最终模型。 */ async function* broken(){yield encode(records().slice(0,2));yield new TextEncoder().encode('{broken}\n');}
|
||||
const states=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:broken()})}),current:()=>true});const handle=loader.acquire(scope,{onUpdate:value=>states.push(value)});
|
||||
try{await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');assert.equal(handle.getSnapshot().chunks.length,2);assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');}finally{handle.release();loader.close();}
|
||||
});
|
||||
// complete 记录已到但流未结束时 ready 仍未解决;随后垃圾尾行使读取失败,不能先打开基线再反悔。
|
||||
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);
|
||||
async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
||||
/* complete 之后仍阻塞迭代器 EOF,之后追加垃圾,检查 ready 不会过早成功。 */ async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
||||
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();}
|
||||
});
|
||||
// abort 取消阻塞的 body reader;身份失效也拒绝下一次读取并取消底层流,分别核对 cancel 次数。
|
||||
test('stream transport fences changed identity and abort cancels a blocked body reader',async()=>{
|
||||
let current=true,cancelled=0;const abort=new AbortController();const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>current,fetch:async()=>new Response(new ReadableStream({cancel(){cancelled++;}}),{headers:{'Content-Type':modelStreamType}})});
|
||||
const reply=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1',signal:abort.signal});const iterator=reply.stream[Symbol.asyncIterator]();const pending=iterator.next();await Promise.resolve();abort.abort();await assert.rejects(pending);assert.equal(cancelled,1);
|
||||
const other=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1'});current=false;await assert.rejects(()=>other.stream[Symbol.asyncIterator]().next(),{code:'identity_changed'});assert.equal(cancelled,2);
|
||||
});
|
||||
|
||||
// 可选 canvasElementCounts 保持旧流兼容;非法清单或后续元素超过单 canvas 声明在发布前拒绝。
|
||||
test('optional snapshot scope counts preserve legacy streams and reject invalid or exceeded coverage before publication',async()=>{
|
||||
const rows=records();rows[0].canvasElementCounts={[root]:1};const seen=await collect(decodeModelStream(source(rows),root,registry,signal()));assert.deepEqual(seen[0].value.canvasElementCounts,{[root]:1});assert.equal(Object.isFrozen(seen[0].value.canvasElementCounts),true);
|
||||
for(const manifest of [null,[],{}, {[root]:-1},{[root]:.5},{[root]:2},{[root]:1,[workspace]:0}]){const broken=records();broken[0].canvasElementCounts=manifest;await assert.rejects(()=>collect(decodeModelStream(source(broken),root,registry,signal())),{code:'invalid_model_coverage'});}
|
||||
const broken=records();broken[0].canvases.push({id:workspace,parent:{canvasId:root,entryElementId:'frame'}});broken[0].versions[workspace]=1;broken[0].canvasElementCounts={[root]:0,[workspace]:1};const iterator=decodeModelStream(source(broken),root,registry,signal())[Symbol.asyncIterator]();assert.equal((await iterator.next()).value.value.receivedElements,0);await assert.rejects(iterator.next(),{code:'invalid_model_coverage'});
|
||||
});
|
||||
|
||||
// elementOwners 声明尚未加载的 ID 与 owner;非法清单或后续未捕获 ID 在发布前拒绝,清单自身冻结。
|
||||
test('identity manifest covers unloaded IDs and binds each received record to its captured owner',async()=>{
|
||||
const rows=records();rows[0].canvasElementCounts={[root]:1};rows[0].elementOwners={a:root};const chunks=await collect(decodeModelStream(source(rows),root,registry,signal()));assert.deepEqual(chunks[0].value.elementOwners,{a:root});assert.ok(Object.isFrozen(chunks[0].value.elementOwners));
|
||||
for(const manifest of [null,[],{}, {other:workspace},{a:root,other:root}]){const broken=structuredClone(rows);broken[0].elementOwners=manifest;await assert.rejects(()=>collect(decodeModelStream(source(broken),root,registry,signal())),{code:'invalid_model_identity'});}
|
||||
|
||||
Reference in New Issue
Block a user