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()); 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 } }]); const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project' }, target = { id: 'a', incarnation: 1 }; 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(); }; function storageFixture() { let image = null, owner = null, fence = 0; const saved = [], fixture = { broken: false, saved, 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); }, 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; } function persistenceFixture(overrides = {}) { let server = original, version = 1; const calls = [], receipts = new Map(); const fixture = { calls, receipts, compile(intent, requestId, snapshot) { return { payload: { requestId, intent }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(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 } }; }, 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; } 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 }; } 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(); }); 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(); }); 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); }); 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]; 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(); }); test('recovery waits for admitted writes, reads a complete base, fences edits and publishes without changing history', async()=>{ const storage=storageFixture(),port=persistenceFixture(),response=deferred(),realSubmit=port.submit,readGate=deferred(); let reads=0;port.submit=async request=>{await response.promise;return realSubmit(request);}; port.read=async()=>{reads++;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 tick();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(); }); 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(); }); 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(); }); 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(); }); 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); }); 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(); }); 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(); } }); 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();} }); 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();} });