457 lines
53 KiB
JavaScript
457 lines
53 KiB
JavaScript
/**
|
|
* ProjectEngine 的组合状态测试：文档、history、操作账本、ACK 基线由统一发布边界协调。
|
|
* 使用内存存储和持久化端口，验证保存顺序、未知结果、恢复会话、本地抵消与回执压缩；不等同于真实磁盘或服务端集成。
|
|
* Root 是不可变业务切点，durability 另行记录已确认保存到哪里；可见修改不等于安全离开。
|
|
*/
|
|
import { test } from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { createDocument, applyChanges, builtinComponents, createRegistry } from '../../src/lib/canvas-core/index.ts';
|
|
import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts';
|
|
import { encodeRecovery, decodeRecovery } from '../../src/lib/canvas-runtime/recovery.ts';
|
|
|
|
// 内置类型参与全部模型校验，避免测试通过空注册表绕开数据约束。
|
|
const registry = createRegistry(builtinComponents());
|
|
// 固定单 canvas、单文本元素的初始模型；后续场景只改变所需字段或生命周期。
|
|
const original = createDocument('project', [{ id: 'canvas', parent: null }], [{ id: 'a', incarnation: 1, structure: 'node', ownerCanvasId: 'canvas', componentId: 'text', dataVersion: 1, data: { title: 'a', text: '' }, geometry: { x: 0, y: 0, width: 100, height: 100, rotation: 0 } }]);
|
|
// 存储恢复作用域包含账户、workspace 和 project；目标引用还带 incarnation。
|
|
const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project' }, target = { id: 'a', incarnation: 1 };
|
|
// 用手动完成的 Promise 精确控制网络、编码或写盘何时越过屏障。
|
|
const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; };
|
|
// 推进有限轮微任务；适用于保留在途请求而又要等调度器进入挂起点的测试。
|
|
const tick = async () => { for (let i = 0; i < 30; i++) await Promise.resolve(); };
|
|
// 内存 storage 持有独占 owner 和递增 fence；每次 commit 保存克隆镜像，并要求修订连续或重试同一修订。
|
|
function storageFixture() {
|
|
let image = null, owner = null, fence = 0;
|
|
const saved = [], fixture = {
|
|
broken: false, saved,
|
|
// 同一时刻只允许一个 owner；释放后再申请会获得新 fence，供重开测试辨别新存储所有者。
|
|
async claim(scope) { assert.equal(owner, null); const leader = { scope, fence: ++fence, active: true, async release() { owner = null; leader.active = false; } }; owner = leader; return leader; },
|
|
// 恢复返回镜像副本，调用方无法修改上一次已保存的对象。
|
|
async recover() { return image && structuredClone(image); },
|
|
// 先检查故障和当前 owner，再保存对应 Root；返回的 durableRootRevision 是引擎推进确认边界的证据。
|
|
async commit(leader, batch) { if (fixture.broken) throw new Error('quota'); assert.equal(owner, leader); assert.equal(leader.active, true); if (image) assert.ok(batch.image.rootRevision === image.rootRevision + 1 || batch.image.rootRevision === image.rootRevision); image = structuredClone(batch.image); saved.push(image); return { durableRootRevision: image.rootRevision }; },
|
|
async readFile() {}, close() {},
|
|
image: () => structuredClone(image), owner: () => owner,
|
|
};
|
|
return fixture;
|
|
}
|
|
// 服务替身按请求 ID 去重，只在首次 submit 应用修改；query 查询既有结果，overrides 可替换任一协议边界。
|
|
// 这里只模拟成功快照与去重，不校验真实服务端的版本前置条件或同 ID 异体请求；不能用此替身证明完整服务端协议正确。
|
|
function persistenceFixture(overrides = {}) {
|
|
let server = original, version = 1;
|
|
const calls = [], receipts = new Map();
|
|
const fixture = {
|
|
calls, receipts,
|
|
// 把逻辑意图与当前参与版本冻结进 payload；测试编译器不执行真实 Foundry wire 格式压缩。
|
|
compile(intent, requestId, snapshot) { return { payload: { requestId, intent }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(id)])) }; },
|
|
// 首次写入产生服务端快照并递增版本；重复同 ID 只返回该回执，不重复修改。
|
|
async submit(request) {
|
|
calls.push(['submit', request]);
|
|
if (!receipts.has(request.id)) { const { intent } = JSON.parse(request.body); server = applyChanges(server, intent.changes); version++; receipts.set(request.id, { document: server, versions: new Map([['canvas', version]]) }); }
|
|
return { outcome: 'applied', canonical: { receipt: request.id } };
|
|
},
|
|
// 记录查询并区分已知成功与 unresolved；没有回执不提供零写入保证。
|
|
async query(request) { calls.push(['query', request]); return receipts.has(request.id) ? { outcome: 'applied', canonical: { receipt: request.id } } : { outcome: 'unresolved' }; },
|
|
// 由已保存回执取规范快照；可在个别测试中替换为冲突数据，检查重放失败边界。
|
|
canonical(request) { return receipts.get(request.id); }, ...overrides,
|
|
};
|
|
return fixture;
|
|
}
|
|
// 统一装配 engine 并收集异步错误；extra 可覆盖时钟、ID、文档、registry 等默认装配选项。
|
|
async function open(storage, persistence, extra = {}) {
|
|
let next = 0; const errors = [];
|
|
const engine = await openProjectEngine({ scope, registry, document: original, versions: new Map([['canvas', 1]]), storage, persistence, createId: () => `identity-${++next}`, onError: error => errors.push(error), ...extra });
|
|
return { engine, errors };
|
|
}
|
|
|
|
// 本地命令一次发布文档、历史和账本；首个请求在途时继续移动，ACK 后基于新版本重放移动，最终两次修改都持久化。
|
|
test('local document/history/ledger publish together; ACK replays edits made while a request is in flight', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), gate = deferred(), realSubmit = port.submit;
|
|
port.submit = async request => { if (!port.calls.length) await gate.promise; return realSubmit(request); };
|
|
const { engine } = await open(storage, port); await engine.persist();
|
|
const first = engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'local' } });
|
|
assert.equal(engine.getSnapshot().history.undo.length, 1);
|
|
assert.ok(engine.getSnapshot().runtime.ledger.operations.has(first.operationIds[0]));
|
|
await tick();
|
|
engine.editor.dispatch({ type: 'element.geometry', target, patch: { x: 200 } });
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200);
|
|
gate.resolve(); await engine.settle();
|
|
assert.equal(port.calls.length, 2);
|
|
assert.equal(port.calls[1][1].expectedVersions.canvas, 2);
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').data.title, 'local');
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200);
|
|
assert.equal(engine.getSnapshot().history.undo.length, 2);
|
|
assert.ok([...engine.getSnapshot().runtime.ledger.operations.values()].every(operation => operation.phase === 'resolved'));
|
|
assert.equal(engine.getDurability().durableRootRevision, engine.getSnapshot().rootRevision);
|
|
assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration, engine.getSnapshot().runtime.ledger.generation);
|
|
const payload = storage.image().payload;
|
|
assert.equal(payload.history.undo.length, 2); assert.equal(payload.runtime.ledger.requests.length, 2);
|
|
await engine.close();
|
|
});
|
|
|
|
// detach 只关闭视图，后台请求继续；引擎关闭再打开后恢复历史并查询原未知请求，ID 和冻结字节保持不变。
|
|
test('closing a view keeps jobs alive; closing/reopening the engine restores history and queries the original unknown request', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), realSubmit = port.submit;
|
|
port.submit = async request => { await realSubmit(request); return { outcome: 'unresolved' }; };
|
|
const { engine } = await open(storage, port);
|
|
const detach = engine.sessions.attach('view');
|
|
engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'once' } }); detach();
|
|
await engine.settle();
|
|
assert.equal(port.calls.length, 1); const request = port.calls[0][1];
|
|
await engine.close();
|
|
const futureClock = { now: () => Date.now() + 100_000, setTimer: (callback, delay) => setTimeout(callback, delay), clearTimer: clearTimeout };
|
|
const recovered = await open(storage, port, { clock: futureClock });
|
|
await new Promise(resolve => setTimeout(resolve, 1)); await recovered.engine.settle();
|
|
assert.equal(port.calls[1][0], 'query'); assert.equal(port.calls[1][1].id, request.id); assert.equal(port.calls[1][1].body, request.body);
|
|
assert.equal(recovered.engine.getSnapshot().history.undo.length, 1);
|
|
assert.equal(recovered.engine.getSnapshot().document.elements.get('a').data.title, 'once');
|
|
assert.equal(recovered.engine.getSnapshot().operations.size, 1);
|
|
await recovered.engine.close();
|
|
});
|
|
|
|
// 创建实际已执行但 ACK 丢失，随后删除仍依赖原创建；恢复先查询再发送独立删除，最终服务端只留墓碑而无幽灵元素。
|
|
test('create applied with a lost ACK followed by delete recovers the original request and leaves no server element',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),submit=port.submit;let lost=true,now=1000,next=0;
|
|
port.submit=async request=>{const status=await submit(request);if(lost){lost=false;return{outcome:'unresolved'};}return status;};
|
|
const options={clock:{now:()=>now,setTimer:()=>0,clearTimer(){}},createId:()=>`lifetime-${++next}`};let{engine}=await open(storage,port,options);
|
|
try{
|
|
const element={...original.elements.get('a'),id:'temporary'},created=engine.editor.dispatch({type:'element.create',element});await engine.settle();
|
|
const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.equal(request.outcome,'unresolved');assert.equal(port.receipts.get(request.id).document.elements.has(element.id),true);
|
|
const removed=engine.editor.dispatch({type:'element.remove',target:{id:element.id,incarnation:1}});await engine.settle();assert.equal(port.calls.length,1);assert.equal(engine.getSnapshot().runtime.ledger.operations.get(removed.operationIds[0]).phase,'pending');assert.ok(engine.getSnapshot().runtime.ledger.operations.get(removed.operationIds[0]).intent.dependencies.includes(created.operationIds[0]));
|
|
await engine.close();now=10000;({engine}=await open(storage,port,options));await engine.settle();
|
|
assert.deepEqual(port.calls.map(([kind])=>kind),['submit','query','submit']);assert.equal(port.calls[1][1].id,request.id);assert.equal(port.calls[1][1].body,request.body);assert.notEqual(port.calls[2][1].id,request.id);
|
|
const final=port.receipts.get(port.calls[2][1].id).document;assert.equal(final.elements.has(element.id),false);assert.equal(final.tombstones.get(element.id),1);assert.equal(engine.getSnapshot().document.elements.has(element.id),false);assert.equal(engine.getSnapshot().history.undo.length,2);
|
|
for(const id of[...created.operationIds,...removed.operationIds])assert.equal(engine.getSnapshot().runtime.ledger.operations.get(id).phase,'resolved');
|
|
}finally{await engine.close();}
|
|
});
|
|
|
|
// quota 失败保留可见编辑但阻止网络；失败 close 保留存储 owner 并封闭编辑入口，存储恢复后重试 close 才释放。
|
|
test('failed local durability preserves editing, prevents sends, and failed close retains the recovery leader', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), { engine, errors } = await open(storage, port);
|
|
await engine.persist(); storage.broken = true;
|
|
engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'unsaved' } }); await tick();
|
|
await assert.rejects(() => engine.persist(), /quota/); await tick();
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').data.title, 'unsaved'); assert.equal(port.calls.length, 0); assert.ok(errors.length);
|
|
await assert.rejects(() => engine.close(), /quota/); assert.ok(storage.owner());
|
|
assert.equal(engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'too-late' } }).code, 'engine_closing');
|
|
storage.broken = false; await engine.close(); assert.equal(storage.owner(), null);
|
|
});
|
|
|
|
// 已知成功回执与后续几何重放冲突时保留本地内容、原始回执和 waiting 占用，不发送下一请求。
|
|
test('canonical replay conflict preserves working content and the known receipt without releasing the writer', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), gate = deferred(), realSubmit = port.submit;
|
|
port.submit = async request => { await gate.promise; return realSubmit(request); };
|
|
const realCanonical = port.canonical;
|
|
port.canonical = request => { const proposal = realCanonical(request); return { ...proposal, document: applyChanges(proposal.document, [{ kind: 'field', target, field: 'geometry', before: original.elements.get('a').geometry, after: { ...original.elements.get('a').geometry, x: 500 } }]) }; };
|
|
const { engine } = await open(storage, port);
|
|
engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'first' } }); await tick();
|
|
engine.editor.dispatch({ type: 'element.geometry', target, patch: { x: 200 } });
|
|
gate.resolve(); await engine.settle();
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x, 200);
|
|
assert.equal(engine.getSnapshot().runtime.conflicts.size, 1); assert.equal(port.calls.length, 1);
|
|
const request = [...engine.getSnapshot().runtime.ledger.requests.values()][0];
|
|
// outcome 说明服务端已经执行；phase 说明本地协调尚未完成。重放冲突不能把已执行改成未执行，也不能释放 writer。
|
|
assert.equal(request.outcome, 'applied'); assert.equal(request.phase, 'waiting');
|
|
assert.ok(engine.getSnapshot().runtime.receipts.has(request.id));
|
|
await engine.close();
|
|
});
|
|
|
|
// 编码解码恢复只读集合及历史；缺失历史引用和自依赖均被拒绝，另一个镜像副本的修改不改变已恢复内容。
|
|
test('recovery codec restores immutable maps and rejects broken history/dependency references', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), { engine } = await open(storage, port);
|
|
engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'encoded' } }); await engine.settle();
|
|
const image = encodeRecovery(engine.getSnapshot()), restored = decodeRecovery(image, registry);
|
|
assert.equal(typeof restored.document.elements.set, 'undefined'); assert.equal(restored.history.undo.length, 1);
|
|
const corrupted = structuredClone(image); corrupted.payload.history.undo[0] = 'missing';
|
|
assert.throws(() => decodeRecovery(corrupted, registry), /corrupt_recovery_image/);
|
|
const cycle = structuredClone(image), record = cycle.payload.runtime.ledger.operations[0][1]; record.intent.dependencies = [record.intent.id];
|
|
assert.throws(() => decodeRecovery(cycle, registry), /corrupt_recovery_image/);
|
|
const changed = structuredClone(image); changed.payload.document.elements[0].data.title = 'mutated';
|
|
assert.equal(restored.document.elements.get('a').data.title, 'encoded');
|
|
await engine.close();
|
|
});
|
|
|
|
// 无未决写入时可接纳更新规范基线并保留历史；旧版本刷新被拒绝，不回退新几何。
|
|
test('a fresh canonical read replaces a clean recovered base, keeps history and refuses older versions', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), first = await open(storage, port);
|
|
first.engine.editor.dispatch({ type: 'element.data', target, patch: { title: 'local' } }); await first.engine.settle(); await first.engine.close();
|
|
const reopened = await open(storage, port), current = reopened.engine.getSnapshot();
|
|
const remote = applyChanges(current.runtime.acknowledged, [{ kind: 'field', target, field: 'geometry', before: current.document.elements.get('a').geometry, after: { ...current.document.elements.get('a').geometry, x: 400 } }]);
|
|
assert.equal(reopened.engine.refresh(remote, new Map([['canvas', 3]])).status, 'applied');
|
|
assert.equal(reopened.engine.getSnapshot().document.elements.get('a').geometry.x, 400);
|
|
assert.equal(reopened.engine.getSnapshot().history.undo.length, 1);
|
|
assert.equal(reopened.engine.refresh(original, new Map([['canvas', 1]])).code, 'stale_canonical_version');
|
|
assert.equal(reopened.engine.getSnapshot().document.elements.get('a').geometry.x, 400);
|
|
await reopened.engine.close();
|
|
});
|
|
|
|
// 恢复先等已入场请求完成再读取完整基线；准备期间封闭编辑、生成和刷新，预览不发布，apply 才发布并落盘且不改原历史引用。
|
|
test('recovery waits for admitted writes, reads a complete base, fences edits and publishes without changing history', {timeout:5000}, async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),response=deferred(),realSubmit=port.submit,readGate=deferred(),readEntered=deferred();
|
|
let reads=0;port.submit=async request=>{await response.promise;return realSubmit(request);};
|
|
port.read=async()=>{reads++;readEntered.resolve();await readGate.promise;return{document:applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'local',beforePresent:true,afterPresent:true},{kind:'field',target,field:'geometry',before:original.elements.get('a').geometry,after:{...original.elements.get('a').geometry,x:300}}]),versions:new Map([['canvas',3]])};};
|
|
const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await tick();
|
|
const preparing=engine.prepareRecovery();await tick();assert.equal(reads,0);
|
|
assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'blocked'}}).code,'recovery_in_progress');
|
|
assert.equal(engine.startRun({}).code,'recovery_in_progress');assert.equal(engine.refresh(original,new Map([['canvas',1]])).code,'recovery_in_progress');
|
|
await assert.rejects(engine.prepareRecovery(),/recovery_in_progress/);
|
|
response.resolve();await readEntered.promise;assert.equal(reads,1);const history=engine.getSnapshot().history;
|
|
readGate.resolve();const session=await preparing;assert.equal(session.conflict,undefined);assert.equal(session.canonical.document.elements.get('a').geometry.x,300);
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,0);assert.equal(engine.editor.history.undo().code,'recovery_in_progress');
|
|
assert.equal((await session.apply()).status,'applied');assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,300);assert.equal(engine.getSnapshot().history,history);
|
|
assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);await assert.rejects(session.apply(),/recovery_session_closed/);
|
|
await engine.close();
|
|
});
|
|
|
|
// 未知回执不能靠全量读跳过确认；prepareRecovery 失败后保留原请求与本地状态，并解除临时编辑屏障。
|
|
test('unknown receipts prevent a recovery read and preserve the frozen request identity',async()=>{
|
|
const storage=storageFixture();let reads=0;const port=persistenceFixture({async submit(){return{outcome:'unresolved'};},async read(){reads++;return{document:original,versions:new Map([['canvas',1]])};}});
|
|
const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle();
|
|
const before=engine.getSnapshot(),request=[...before.runtime.ledger.requests.values()][0];
|
|
await assert.rejects(engine.prepareRecovery(),/recovery_waiting_for_receipts/);assert.equal(reads,0);
|
|
const after=engine.getSnapshot();assert.equal(after.document,before.document);assert.equal(after.history,before.history);assert.equal(after.runtime.ledger.requests.get(request.id).body,request.body);
|
|
assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'allowed'}}).enabled,true);await engine.close();
|
|
});
|
|
|
|
// 冲突预览不覆盖本地内容或历史，直接 apply 被拒绝；重复 cancel 安全，取消后恢复普通编辑。
|
|
test('recovery conflict previews preserve local data/history and cancel releases the edit fence',async()=>{
|
|
const remote=applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'remote',beforePresent:true,afterPresent:true}]);
|
|
const storage=storageFixture(),port=persistenceFixture({async submit(){return{outcome:'not-applied',error:'version_conflict'};},async read(){return{document:remote,versions:new Map([['canvas',2]])};}});
|
|
const{engine}=await open(storage,port);engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle();
|
|
const before=engine.getSnapshot(),session=await engine.prepareRecovery();assert.ok(session.conflict);assert.equal(session.pendingOperationIds.length,1);
|
|
assert.equal((await session.apply()).status,'rejected');assert.equal(engine.getSnapshot(),before);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'local');
|
|
await session.cancel();await session.cancel();assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'more'}}).enabled,true);await engine.close();
|
|
});
|
|
|
|
// 恢复 Root 已发布但保存失败时保留编辑屏障；重试保存同一 Root，不再创建一次恢复修订。
|
|
test('failed recovery durability retains its fence and retries the same published root',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture({async read(){return{document:original,versions:new Map([['canvas',1]])};}}),{engine}=await open(storage,port);
|
|
const session=await engine.prepareRecovery();storage.broken=true;await assert.rejects(session.apply(),/quota/);const revision=engine.getSnapshot().rootRevision;
|
|
assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'blocked'}}).code,'recovery_in_progress');await assert.rejects(session.cancel(),/quota/);
|
|
storage.broken=false;assert.equal((await session.apply()).status,'applied');assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getDurability().durableRootRevision,revision);
|
|
await engine.close();
|
|
});
|
|
|
|
// 关闭引擎 abort 恢复读取并释放 owner；端口仍返回迟到结果也必须拒绝，不能创建恢复会话。
|
|
test('closing the engine aborts a recovery read and rejects a late canonical response',async()=>{
|
|
const storage=storageFixture(),response=deferred();let signal;
|
|
const port=persistenceFixture({async read(input){signal=input;return response.promise;}}),{engine}=await open(storage,port);
|
|
const preparing=engine.prepareRecovery();await tick();assert.ok(signal);await engine.close();assert.equal(signal.aborted,true);
|
|
response.resolve({document:original,versions:new Map([['canvas',1]])});await assert.rejects(preparing,/engine_closing/);assert.equal(storage.owner(),null);
|
|
});
|
|
|
|
// 恢复读和普通 refresh 都拒绝同版本内容改写；另外直接检查墓碑回退与生命周期变化，不凭版本增加就接受旧生命期。
|
|
test('complete recovery reads reject same-version rewrites and lifecycle regression before creating a session',async()=>{
|
|
const remote=applyChanges(original,[{kind:'field',target,field:'data.title',before:'a',after:'remote',beforePresent:true,afterPresent:true}]);
|
|
const storage=storageFixture(),port=persistenceFixture({async read(){return{document:remote,versions:new Map([['canvas',1]])};}}),{engine}=await open(storage,port);
|
|
await assert.rejects(engine.prepareRecovery(),/canonical_version_mismatch/);
|
|
assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'a');
|
|
assert.equal(engine.editor.can({type:'element.data',target,patch:{title:'allowed'}}).enabled,true);
|
|
assert.equal(engine.refresh(remote,new Map([['canvas',1]])).code,'canonical_version_mismatch');
|
|
const {validateCanonicalRead}=await import('../../src/lib/canvas-runtime/reconcile.ts');
|
|
const withTombstone=createDocument(original.projectId,[...original.canvases.values()],[...original.elements.values()],new Map([['deleted',2]]));
|
|
assert.throws(()=>validateCanonicalRead(withTombstone,new Map([['canvas',1]]),original,new Map([['canvas',2]]),registry),/stale_canonical_lifecycle/);
|
|
assert.throws(()=>validateCanonicalRead(original,new Map([['canvas',1]]),withTombstone,new Map([['canvas',1]]),registry),/canonical_version_mismatch/);
|
|
await engine.close();
|
|
});
|
|
|
|
// 同一事务先创建再删除，basePresence 必须保留最初 absent；本地留下墓碑但无需发送，也不能伪造 ACK 墓碑。
|
|
test('one transaction captures the first presence baseline even when its created element is later deleted', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(), { engine } = await open(storage, port);
|
|
try {
|
|
const node = { ...original.elements.get('a'), id: 'temporary' };
|
|
const result = engine.editor.transaction(tx => {
|
|
tx.dispatch({ type: 'element.create', element: node });
|
|
tx.dispatch({ type: 'element.remove', target: { id: node.id, incarnation: 1 } });
|
|
});
|
|
assert.equal(result.status, 'applied'); assert.equal(result.operationIds.length, 1);
|
|
const operation = engine.getSnapshot().runtime.ledger.operations.get(result.operationIds[0]);
|
|
assert.equal(operation.basePresence.temporary, 'absent');
|
|
assert.equal(engine.getSnapshot().document.tombstones.get(node.id), 1);
|
|
await engine.settle(); assert.equal(port.calls.length, 0);
|
|
assert.equal(engine.getSnapshot().runtime.ledger.operations.get(result.operationIds[0]).resolution, 'elided');
|
|
assert.equal(engine.getSnapshot().runtime.acknowledged.tombstones.has(node.id), false);
|
|
} finally { await engine.close(); }
|
|
});
|
|
|
|
|
|
// 完全未发送的创建删除在本地抵消，仍保留两个历史单元和墓碑；无关 ACK 与重开不丢它，撤销恢复为新 incarnation。
|
|
test('unsent create/delete retires locally, survives unrelated ACK and reopening, and undo restores a fresh incarnation', async () => {
|
|
const storage = storageFixture(), port = persistenceFixture(); let next=0, engine;
|
|
const reopen=async()=>{if(engine)await engine.close();engine=(await open(storage,port,{createId:()=>`elision-${++next}`})).engine;};
|
|
try {
|
|
await reopen();const node={...original.elements.get('a'),id:'temporary'};
|
|
const created=engine.editor.dispatch({type:'element.create',element:node});
|
|
const removed=engine.editor.dispatch({type:'element.remove',target:{id:node.id,incarnation:1}});
|
|
await engine.settle();assert.equal(port.calls.length,0);assert.equal(engine.getSnapshot().history.undo.length,2);
|
|
for(const id of [...created.operationIds,...removed.operationIds])assert.equal(engine.getSnapshot().runtime.ledger.operations.get(id).resolution,'elided');
|
|
engine.editor.dispatch({type:'element.data',target,patch:{title:'unrelated'}});await engine.settle();assert.equal(port.calls.length,1);
|
|
assert.equal(engine.getSnapshot().document.tombstones.get(node.id),1);assert.equal(engine.getSnapshot().runtime.acknowledged.tombstones.has(node.id),false);
|
|
await reopen();assert.equal(engine.getSnapshot().document.tombstones.get(node.id),1);
|
|
assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();
|
|
assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();
|
|
assert.equal(engine.getSnapshot().document.elements.get(node.id).incarnation,2);assert.equal(port.calls.length,3);
|
|
assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.has(node.id),false);assert.equal(engine.getSnapshot().document.tombstones.get(node.id),2);
|
|
await reopen();assert.equal(engine.getSnapshot().document.tombstones.get(node.id),2);
|
|
} finally { await engine?.close(); }
|
|
});
|
|
|
|
|
|
// 本地抵消证据、工作墓碑和历史内容必须相互匹配；伪造或缺失会拒绝，未来解释版本单独报告不可解释。
|
|
test('recovery rejects forged or unsupported local completion proofs and missing working lifecycle marks', async () => {
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port);
|
|
try {
|
|
const n={...original.elements.get('a'),id:'temporary'};
|
|
engine.editor.dispatch({type:'element.create',element:n});engine.editor.dispatch({type:'element.remove',target:{id:n.id,incarnation:1}});await engine.settle();
|
|
const image=storage.image(),anchorIndex=image.payload.runtime.ledger.operations.findIndex(([,record])=>record.elision);
|
|
for(const corrupt of [
|
|
image=>{image.payload.runtime.ledger.operations[anchorIndex][1].elision.tombstones.temporary=2;},
|
|
image=>{image.payload.document.tombstones=[];},
|
|
image=>{image.payload.operations[0][1].changes[0].after.data.title='different historical content';},
|
|
]){const bad=structuredClone(image);corrupt(bad);assert.throws(()=>decodeRecovery(bad,registry),/corrupt_recovery_image/);}
|
|
const future=structuredClone(image);future.payload.runtime.ledger.operations[anchorIndex][1].elision.version=2;assert.throws(()=>decodeRecovery(future,registry),/elision_interpreter_unavailable/);
|
|
assert.equal(decodeRecovery(image,registry).document.tombstones.get(n.id),1);
|
|
}finally{await engine.close();}
|
|
});
|
|
|
|
|
|
// 保存失败的本地抵消只在内存成立，不虚构远程回执；重试保存并重开后仍使用原抵消证据且无网络请求。
|
|
test('failed durability keeps a local elision in memory without inventing a network success, then persists the same proof', async () => {
|
|
const storage=storageFixture(),port=persistenceFixture();let engine=(await open(storage,port)).engine;
|
|
try{
|
|
await engine.persist();storage.broken=true;const n={...original.elements.get('a'),id:'temporary'};
|
|
engine.editor.dispatch({type:'element.create',element:n});engine.editor.dispatch({type:'element.remove',target:{id:n.id,incarnation:1}});await tick();
|
|
await assert.rejects(engine.persist(),/quota/);assert.equal(port.calls.length,0);
|
|
const anchor=[...engine.getSnapshot().runtime.ledger.operations.values()].find(record=>record.elision),proof=JSON.stringify(anchor.elision);assert.equal(anchor.resolution,'elided');assert.equal(engine.getSnapshot().runtime.receipts.size,0);
|
|
storage.broken=false;await engine.persist();await engine.close();engine=(await open(storage,port)).engine;await engine.settle();
|
|
assert.equal(port.calls.length,0);assert.equal(JSON.stringify(engine.getSnapshot().runtime.ledger.operations.get(anchor.intent.id).elision),proof);
|
|
}finally{storage.broken=false;await engine.close();}
|
|
});
|
|
|
|
|
|
// 反复未发送的 undo/redo 仍可本地抵消并保留历次墓碑；首次真正发布恢复时使用第四代，跳过已取消代次。
|
|
test('repeated unsent undo/redo pairs remain local and the first published restore skips all canceled incarnations',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture();let next=0,engine=(await open(storage,port,{createId:()=>`repeat-${++next}`})).engine;
|
|
try{
|
|
const n={...original.elements.get('a'),id:'temporary'};engine.editor.dispatch({type:'element.create',element:n});engine.editor.dispatch({type:'element.remove',target:{id:n.id,incarnation:1}});await engine.settle();
|
|
for(let incarnation=2;incarnation<=3;incarnation++){
|
|
assert.equal(engine.editor.history.undo().status,'applied');assert.equal(engine.getSnapshot().document.elements.get(n.id).incarnation,incarnation);assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();assert.equal(port.calls.length,0);assert.equal(engine.getSnapshot().document.tombstones.get(n.id),incarnation);
|
|
await engine.close();engine=(await open(storage,port,{createId:()=>`repeat-${++next}`})).engine;
|
|
}
|
|
assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(port.calls.length,1);assert.equal(engine.getSnapshot().document.elements.get(n.id).incarnation,4);assert.equal(engine.getSnapshot().runtime.acknowledged.elements.get(n.id).incarnation,4);
|
|
}finally{await engine.close();}
|
|
});
|
|
|
|
|
|
// 即使端口支持批量编译，也先抵消后面的闭合创建删除；仅发送前面真实编辑，批量编译器若收到闭合生命期就使测试失败。
|
|
test('batch-capable persistence sends an earlier real edit separately and never freezes the later closed lifetime',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture({compileBatch(){throw Error('closed lifetime entered the wire batch');}}),{engine,errors}=await open(storage,port);
|
|
try{
|
|
engine.editor.dispatch({type:'element.data',target,patch:{title:'prefix'}});const n={...original.elements.get('a'),id:'temporary'},created=engine.editor.dispatch({type:'element.create',element:n}),removed=engine.editor.dispatch({type:'element.remove',target:{id:n.id,incarnation:1}});
|
|
await engine.settle();assert.equal(errors.length,0);assert.equal(port.calls.length,1);assert.equal(engine.getSnapshot().runtime.acknowledged.elements.get('a').data.title,'prefix');
|
|
for(const id of [...created.operationIds,...removed.operationIds])assert.equal(engine.getSnapshot().runtime.ledger.operations.get(id).resolution,'elided');assert.equal(engine.getSnapshot().runtime.acknowledged.tombstones.has(n.id),false);
|
|
}finally{await engine.close();}
|
|
});
|
|
|
|
// 不主动触发重试的假时钟；定时器仅返回 token，使持久化竞争用例集中观察显式保存和在途网络。
|
|
const idleClock=()=>({now:()=>0,setTimer:()=>Symbol('idle'),clearTimer(){}});
|
|
// 八次编辑各保留业务 Root，保存只按确认前缀发布 durableGeneration 回声；派发前镜像必须已含 dispatch-marked 请求和原体。
|
|
test('a pending write train preserves every business root and publishes one durable-generation echo per captured prefix',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let sends=0;port.submit=async request=>{sends++;const stored=storage.image();assert.ok(stored.payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;let held=false,markers=[],through=start.runtime.ledger.durableGeneration;const unsubscribe=engine.editor.subscribe((_,root)=>{if(root.runtime.ledger.durableGeneration>through){markers.push(root.runtime.ledger.durableGeneration);through=root.runtime.ledger.durableGeneration;}});
|
|
storage.commit=async(...args)=>{if(!held){held=true;entered.resolve();await gate.promise;}return commit(...args);};
|
|
for(let i=0;i<8;i++)assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'train-'+i}}).status,'applied');
|
|
|
|
// 保存队列挂起时记录调用方目标 generation；调度器可能再发布冻结请求，因此还要核对后续捕获前缀的确认回声。
|
|
const requested=engine.getSnapshot().runtime.ledger.generation,saving=engine.persist();await entered.promise;await tick();const expected=engine.getSnapshot().runtime.ledger.generation,queued=engine.getSnapshot().rootRevision-start.rootRevision;assert.equal(sends,0);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,start.runtime.ledger.durableGeneration);gate.resolve();await saving;unsubscribe();
|
|
// durableGeneration 的确认也要发布为新 Root 并保存，因此镜像数量包含业务切点和确认切点；后者不会新增用户编辑。
|
|
assert.deepEqual(markers,[...new Set([requested,expected])]);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,expected);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);const rows=storage.saved.slice(saved);assert.equal(rows.length,queued+markers.length);assert.deepEqual(rows.map(image=>image.rootRevision),Array.from({length:queued+markers.length},(_,i)=>start.rootRevision+i+1));assert.deepEqual(rows.slice(0,8).map(image=>image.payload.document.elements[0].data.title),Array.from({length:8},(_,i)=>'train-'+i));assert.equal(decodeRecovery(storage.image(),registry).runtime.ledger.durableGeneration,expected);await tick();assert.equal(sends,1);
|
|
}finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
// 连续保存第四次失败时只确认前三个切点，内存仍显示最后编辑；重试从完整未确认后缀继续，修订不得跳号且不能提前发送。
|
|
test('a later quota failure records only confirmed generation and retries the full continuous suffix',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let sends=0;port.submit=async request=>{sends++;const stored=storage.image();assert.ok(stored.payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;let writes=0,broken=true;storage.commit=async(...args)=>{writes++;if(writes===1){entered.resolve();await gate.promise;}if(broken&&writes===4)throw Error('quota-later');return commit(...args);};
|
|
for(let i=0;i<8;i++)engine.editor.dispatch({type:'element.data',target,patch:{title:'partial-'+i}});const saving=engine.persist();await entered.promise;await tick();const expected=engine.getSnapshot().runtime.ledger.generation,queued=engine.getSnapshot().rootRevision-start.rootRevision;assert.equal(sends,0);gate.resolve();await assert.rejects(saving,/quota-later/);const confirmed=storage.saved.at(-1);assert.equal(storage.saved.length-saved,3);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,confirmed.payload.runtime.ledger.generation);assert.ok(engine.getSnapshot().runtime.ledger.durableGeneration<expected);assert.equal(engine.getDurability().durableRootRevision,confirmed.rootRevision);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'partial-7');assert.equal(sends,0);
|
|
|
|
// 失败前缀已被确认，重试只补连续后缀和确认 Root；不能因内存已显示最后编辑就跳过缺失修订。
|
|
broken=false;await engine.persist();assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,expected);const rows=storage.saved.slice(saved);assert.deepEqual(rows.map(image=>image.rootRevision),Array.from({length:rows.length},(_,i)=>start.rootRevision+i+1));assert.equal(decodeRecovery(storage.image(),registry).document.elements.get('a').data.title,'partial-7');assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);
|
|
}finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
// commit 返回超前修订是无效证据，即使内存存储已经写入也不能推进确认；重试必须提交与原切点完全相同的镜像。
|
|
test('an invalid receipt cannot advance confirmed generation and the original cut remains retryable',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;let bad=true,held=false,sends=0;port.submit=async()=>{sends++;await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();const start=engine.getSnapshot(),saved=storage.saved.length;storage.commit=async(...args)=>{if(!held){held=true;entered.resolve();await gate.promise;}const receipt=await commit(...args);return bad?{durableRootRevision:receipt.durableRootRevision+1}:receipt;};engine.editor.dispatch({type:'element.data',target,patch:{title:'receipt-test'}});const saving=engine.persist();await entered.promise;await tick();gate.resolve();await assert.rejects(saving,/invalid_durable_receipt/);assert.equal(engine.getDurability().durableRootRevision,start.rootRevision);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,start.runtime.ledger.durableGeneration);assert.equal(sends,0);
|
|
// 底层实际写入与上层认可证据分开：坏回执未获认可，下一次必须重试同一镜像，而不是假设可以跳到下个 Root。
|
|
const first=storage.saved[saved];bad=false;await engine.persist();const retry=storage.saved[saved+1];assert.deepEqual(retry,first);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,engine.getSnapshot().runtime.ledger.generation);
|
|
}finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
// 保存成功后因 owner 失效而无法发布确认时保留已验证回执；恢复 owner 后只补确认 Root，监听器重入 persist 也不能并发写盘或重写旧切点。
|
|
// 手动翻转 leader.active 是内存替身的故障注入，用于隔离确认发布重试；不表示真实存储可以复活已失效的租约。
|
|
test('a failed durability publication retains its verified receipt through retry without rewriting saved business cuts',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),network=deferred(),commit=storage.commit;let lose=true;port.submit=async()=>{await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();const leader=storage.owner(),saved=storage.saved.length;storage.commit=async(...args)=>{const receipt=await commit(...args);if(lose&&args[1].image.rootRevision===engine.getSnapshot().rootRevision){lose=false;leader.active=false;}return receipt;};engine.editor.dispatch({type:'element.data',target,patch:{title:'publication-retry'}});await assert.rejects(engine.persist(),/storage_leader_lost/);const confirmed=storage.saved.at(-1);assert.equal(engine.getDurability().durableRootRevision,confirmed.rootRevision);assert.ok(confirmed.payload.runtime.ledger.generation>engine.getSnapshot().runtime.ledger.durableGeneration);const count=storage.saved.length;leader.active=true;let activeWrites=0,maxWrites=0;const nested=[];storage.commit=async(...args)=>{activeWrites++;maxWrites=Math.max(maxWrites,activeWrites);try{await tick();return await commit(...args);}finally{activeWrites--;}};const unsubscribe=engine.editor.subscribe((_,root)=>{if(root.runtime.ledger.durableGeneration===confirmed.payload.runtime.ledger.generation)nested.push(engine.persist());});await engine.persist();await Promise.all(nested);unsubscribe();assert.equal(maxWrites,1);assert.equal(storage.saved.length,count+1);assert.equal(storage.saved.at(-1).rootRevision,confirmed.rootRevision+1);assert.equal(engine.getSnapshot().runtime.ledger.durableGeneration,confirmed.payload.runtime.ledger.generation);assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);assert.ok(storage.saved.length>saved);
|
|
}finally{const leader=storage.owner();if(leader)leader.active=true;network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
// 空写入泵刚排空后再产生编辑，新的 persist 调用仍必须等自己的目标 Root，不能复用旧的已完成等待。
|
|
test('a caller arriving after an empty writer drain still waits for its own newly published Root', {timeout:5000},async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),gate=deferred(),entered=deferred(),network=deferred(),commit=storage.commit;port.submit=async()=>{await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();storage.commit=async(...args)=>{entered.resolve();await gate.promise;return commit(...args);};const empty=engine.persist();let edited,required;await new Promise(resolve=>queueMicrotask(()=>{engine.editor.dispatch({type:'element.data',target,patch:{title:'after-empty-writer'}});required=engine.getSnapshot().rootRevision;edited=engine.persist();resolve();}));assert.equal(await Promise.race([entered.promise.then(()=>'write'),edited.then(()=>'returned')]),'write');assert.ok(engine.getDurability().durableRootRevision<required);gate.resolve();await Promise.all([empty,edited]);assert.ok(engine.getDurability().durableRootRevision>=required);assert.equal(storage.image().payload.document.elements[0].data.title,'after-empty-writer');
|
|
}finally{gate.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
// 已保存的 dispatch 前缀可以发送，无需等待更晚的本地 Root；但显式 persist 仍等待它捕获的目标修订全部保存。
|
|
test('a durable dispatch prefix can send while a later local Root is still waiting for storage', {timeout:10000},async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture(),{engine}=await open(storage,port,{clock:idleClock()}),body=deferred(),bodyEntered=deferred(),later=deferred(),laterEntered=deferred(),sent=deferred(),network=deferred(),commit=storage.commit;let bodyCut,after=Infinity,flushed=false;port.submit=async request=>{assert.ok(storage.image().payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));sent.resolve();await network.promise;return{outcome:'unresolved'};};try{
|
|
await engine.persist();storage.commit=async(...args)=>{const image=args[1].image;if(bodyCut===undefined&&image.payload.runtime.ledger.requests.some(([,row])=>row.phase==='dispatch-marked')){bodyCut=image.rootRevision;bodyEntered.resolve();await body.promise;}if(image.rootRevision>after){laterEntered.resolve();await later.promise;}return commit(...args);};engine.editor.dispatch({type:'element.data',target,patch:{title:'first'}});const saving=engine.persist().then(()=>{flushed=true;});await bodyEntered.promise;after=engine.getSnapshot().rootRevision;engine.editor.dispatch({type:'element.geometry',target,patch:{x:42}});body.resolve();await laterEntered.promise;await new Promise((resolve,reject)=>{const timer=setTimeout(()=>reject(Error('durable request waited for a later Root')),2000);sent.promise.then(()=>{clearTimeout(timer);resolve();},reject);});assert.equal(flushed,false);assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,42);later.resolve();await saving;assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);
|
|
}finally{body.resolve();later.resolve();network.resolve();storage.commit=commit;await engine.close();}
|
|
});
|
|
|
|
// 可选成功回执摘要必须与 ACK、history、ledger 一起发布；每个可恢复切点保持一致，重开不重传且冻结 body 不变。
|
|
test('an opted-in resolved receipt summary publishes with ACK/history/ledger and reopens without retransmission',async()=>{
|
|
const storage=storageFixture(),port=persistenceFixture({compactResolvedReceipt(request,receipt,applied){assert.equal(request.phase,'resolved');assert.equal(request.outcome,'applied');assert.equal(receipt.receipt,request.id);assert.equal(applied.document.elements.get('a').data.title,'summary');return{localReceiptVersion:1,id:request.id,versions:[...applied.versions]};}});let{engine}=await open(storage,port);
|
|
try{engine.editor.dispatch({type:'element.data',target,patch:{title:'summary'}});await engine.settle();const snapshot=engine.getSnapshot(),request=[...snapshot.runtime.ledger.requests.values()][0],body=request.body;assert.equal(request.phase,'resolved');assert.deepEqual(snapshot.runtime.receipts.get(request.id),{localReceiptVersion:1,id:request.id,versions:[['canvas',2]]});assert.equal(snapshot.history.undo.length,1);for(const image of storage.saved){const restored=decodeRecovery(image,registry),saved=restored.runtime.ledger.requests.get(request.id);if(restored.runtime.receipts.get(request.id)?.localReceiptVersion===1){assert.equal(saved.phase,'resolved');assert.equal(restored.document.elements.get('a').data.title,'summary');assert.equal(restored.history.undo.length,1);}if(saved)assert.equal(saved.body,body);}await engine.close();({engine}=await open(storage,port));await engine.settle();assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).body,body);assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'summary');assert.equal(port.calls.filter(([kind])=>kind==='submit').length,1);
|
|
}finally{await engine.close();}
|
|
});
|
|
// 摘要准备后最终模型校验失败，必须保留原始成功回执和 waiting 占用；校验恢复后查询修复，不第二次 submit。
|
|
// compactResolvedReceipt 首次调用时打开 blocked，使自定义 text 校验器拒绝 local；故障发生在本地发布，而非请求执行。
|
|
test('a failed final publication keeps the raw receipt and waiting claim instead of the prepared summary',async()=>{
|
|
let blocked=false,compactions=0;const custom=createRegistry(builtinComponents().map(definition=>definition.id==='text'?{...definition,validate:data=>typeof data.title==='string'&&typeof data.text==='string'&&(!blocked||data.title!=='local')}:definition)),storage=storageFixture(),port=persistenceFixture({compactResolvedReceipt(request){compactions++;if(compactions===1)blocked=true;return{summary:request.id};}});let{engine}=await open(storage,port,{registry:custom});
|
|
try{engine.editor.dispatch({type:'element.data',target,patch:{title:'local'}});await engine.settle();const snapshot=engine.getSnapshot(),request=[...snapshot.runtime.ledger.requests.values()][0];assert.equal(compactions,1);assert.equal(request.outcome,'applied');assert.equal(request.phase,'waiting');assert.deepEqual(snapshot.runtime.receipts.get(request.id),{receipt:request.id});assert.equal(snapshot.document.elements.get('a').data.title,'local');assert.equal(snapshot.history.undo.length,1);blocked=false;await engine.close();({engine}=await open(storage,port,{registry:custom}));await engine.retry();await engine.settle();assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'resolved');assert.deepEqual(engine.getSnapshot().runtime.receipts.get(request.id),{summary:request.id});assert.equal(port.calls.filter(([kind])=>kind==='submit').length,1);assert.ok(port.calls.some(([kind])=>kind==='query'));
|
|
}finally{blocked=false;await engine.close();}
|
|
});
|
|
// 未提供、抛错或返回空值的压缩器不影响合法 ACK；保留原始回执，只有失败压缩器通过错误回调报告。
|
|
test('receipt compaction is optional and invalid codecs cannot block a valid ACK or discard its raw payload',async()=>{
|
|
for(const compactResolvedReceipt of [undefined,()=>{throw Error('compact failed');},()=>undefined]){const storage=storageFixture(),port=persistenceFixture(compactResolvedReceipt?{compactResolvedReceipt}:{}),{engine,errors}=await open(storage,port);try{engine.editor.dispatch({type:'element.data',target,patch:{title:'keep raw'}});await engine.settle();const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.equal(request.phase,'resolved');assert.deepEqual(engine.getSnapshot().runtime.receipts.get(request.id),{receipt:request.id});assert.equal(engine.getSnapshot().runtime.conflicts.size,0);assert.equal(errors.length,compactResolvedReceipt?1:0);}finally{await engine.close();}}
|
|
});
|
|
|
|
// 大模型编码在 scheduler.yield 挂起期间继续编辑，首个镜像仍对应捕获的旧 Root；随后镜像按修订连续保存，冻结请求未落盘前不发送。
|
|
test('encoding yields preserve the captured Root, ordered durable prefix and send barrier', {timeout:15000},async()=>{
|
|
const previous=Object.getOwnPropertyDescriptor(globalThis,'scheduler'),entered=deferred(),resume=deferred();let hold=false,paused=false,yields=0;
|
|
Object.defineProperty(globalThis,'scheduler',{configurable:true,value:{async yield(){yields++;if(hold&&!paused){paused=true;entered.resolve();await resume.promise;}}}});
|
|
let engine;
|
|
try{
|
|
|
|
// 超过编码让出阈值的大模型；每个记录合法且 ID 独立，避免用编码失败替代调度竞争。
|
|
const rows=Array.from({length:17000},(_,i)=>({...original.elements.get('a'),id:i===0?'a':`encoding-${i}`}));
|
|
const document=createDocument('project',[{id:'canvas',parent:null}],rows),storage=storageFixture(),port=persistenceFixture({async submit(request){port.calls.push(['submit',request]);const stored=storage.image();assert.ok(stored.payload.runtime.ledger.requests.some(([id,row])=>id===request.id&&row.body===request.body&&row.phase==='dispatch-marked'));return{outcome:'unresolved'};}});
|
|
({engine}=await open(storage,port,{document,clock:idleClock()}));await engine.persist();
|
|
const start=engine.getSnapshot().rootRevision,saved=storage.saved.length;hold=true;
|
|
assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'captured-first'}}).status,'applied');
|
|
|
|
// 首个 persist 捕获旧 Root 后暂停编码，再允许第二次编辑产生新 Root；两次保存必须各自对应其捕获内容。
|
|
const captured=engine.getSnapshot(),first=engine.persist();await entered.promise;
|
|
assert.equal(storage.saved.length,saved);assert.equal(port.calls.length,0);
|
|
assert.equal(engine.editor.dispatch({type:'element.data',target,patch:{title:'edited-during-yield'}}).status,'applied');
|
|
const second=engine.persist();assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'edited-during-yield');
|
|
resume.resolve();await Promise.all([first,second]);
|
|
|
|
// 比较完整首镜像与同步 encodeRecovery(captured)，验证让出主线程后没有混入后来编辑；同时检查修订序列无空洞。
|
|
const cuts=storage.saved.slice(saved);assert.ok(yields>0);
|
|
assert.deepEqual(cuts.map(cut=>cut.rootRevision),Array.from({length:cuts.length},(_,i)=>start+i+1));
|
|
assert.deepEqual(cuts[0],encodeRecovery(captured));
|
|
assert.equal(cuts[0].payload.document.elements.find(row=>row.id==='a').data.title,'captured-first');
|
|
assert.equal(storage.image().payload.document.elements.find(row=>row.id==='a').data.title,'edited-during-yield');
|
|
assert.equal(engine.getDurability().durableRootRevision,engine.getSnapshot().rootRevision);
|
|
}finally{resume.resolve();if(engine)await engine.close();if(previous)Object.defineProperty(globalThis,'scheduler',previous);else delete globalThis.scheduler;}
|
|
});
|