import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createLedger } from '../../src/lib/canvas-runtime/ledger.ts'; import { createScheduler } from '../../src/lib/canvas-runtime/scheduler.ts'; const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; }; const tick = async () => { for (let i = 0; i < 20; i++) await Promise.resolve(); }; function setup(overrides = {}, initial) { let ledger = createLedger(initial), time = 1000, next = 0, timerId = 0; const timers = new Map(), calls = [], persisted = [], errors = []; const scheduler = createScheduler({ getLedger: () => ledger.getSnapshot(), mutate(reduce) { const draft = createLedger(ledger.getSnapshot()); reduce(draft); ledger = draft; }, async persist() { persisted.push(ledger.getSnapshot()); }, createId: () => `request-${++next}`, participants: intent => intent.changes.map(change => change.id), compile: (intent, requestId) => ({ canvasIds: intent.changes.map(change => change.id), expectedVersions: Object.fromEntries(intent.changes.map(change => [change.id, 1])), payload: { requestId, operationId: intent.id } }), async submit(request) { calls.push(['submit', request]); return { outcome: 'applied' }; }, async query(request) { calls.push(['query', request]); return { outcome: 'applied' }; }, complete(request, status) { const draft = createLedger(ledger.getSnapshot()); draft.result(request.id, status.outcome, status.error); ledger = draft; }, clock: { now: () => time, setTimer(callback, delay) { const id = ++timerId; timers.set(id, { callback, at: time + delay }); return id; }, clearTimer: id => timers.delete(id) }, retryDelay: () => 100, onError: error => errors.push(error), ...overrides, }); return { scheduler, calls, persisted, errors, ledger: () => ledger, add(id, canvases = ['canvas'], dependencies = []) { ledger.add({ id, operationType: 'document.change', operationSchemaVersion: 1, sequence: ledger.getSnapshot().operations.size + 1, origin: 'user', changes: canvases.map(id => ({ kind: 'canvas', id, before: null, after: { id, parent: null } })), dependencies }, {}); scheduler.notify([id]); }, async advance(ms, settle = true) { time += ms; for (const [id, timer] of [...timers]) if (timer.at <= time) { timers.delete(id); timer.callback(); } await tick(); if (settle) await scheduler.settle(); }, timerCount: () => timers.size, }; } test('freeze and dispatch evidence must be durable before submit, then dependencies wake', async () => { const barrier = deferred(), sent = [], state = setup({ persist: () => barrier.promise, submit: async request => { sent.push(request); return { outcome: 'applied' }; } }); state.add('create'); state.add('delete', ['canvas'], ['create']); state.scheduler.start(); await tick(); assert.equal(sent.length, 0); const frozen = [...state.ledger().getSnapshot().requests.values()][0]; assert.equal(frozen.phase, 'dispatch-marked'); assert.equal(state.scheduler.getClaims().get('canvas'), frozen.id); barrier.resolve(); await state.scheduler.settle(); assert.equal(sent.length, 2); assert.equal(state.ledger().getSnapshot().operations.get('delete').phase, 'resolved'); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); }); test('unknown request holds its canvas; another canvas proceeds; accepted stays on query path', async () => { const calls = []; let queries = 0; const state = setup({ async submit(request) { calls.push(['submit', request]); return request.operationIds[0] === 'create' ? { outcome: 'unresolved' } : { outcome: 'applied' }; }, async query(request) { calls.push(['query', request]); return { outcome: ++queries === 1 ? 'accepted' : 'applied' }; }, }); state.add('create'); state.add('same-canvas'); state.add('independent', ['other']); state.scheduler.start(); await state.scheduler.settle(); assert.deepEqual(calls.map(([_, request]) => request.operationIds[0]), ['create', 'independent']); assert.equal(state.scheduler.getClaims().size, 1); assert.equal(state.timerCount(), 1); await state.advance(100); assert.equal(calls.at(-1)[0], 'query'); assert.equal(state.ledger().getSnapshot().operations.get('create').phase, 'frozen'); await state.advance(100); assert.deepEqual(calls.map(([action, request]) => [action, request.operationIds[0]]), [['submit','create'], ['submit','independent'], ['query','create'], ['query','create'], ['submit','same-canvas']]); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); }); test('receipt missing after recovery triggers exact retry, never a new identity/body', async () => { const original = setup({ submit: async () => ({ outcome: 'unresolved' }) }); original.add('create'); original.scheduler.start(); await original.scheduler.settle(); const saved = original.ledger().getSnapshot(), before = saved.requests.get('request-1'); original.scheduler.stop(); const calls = []; const restored = setup({ query: async request => { calls.push(['query', request]); return { outcome: 'unresolved', error: 'receipt_unknown' }; }, submit: async request => { calls.push(['submit', request]); return { outcome: 'applied' }; } }, saved); restored.scheduler.start(); await restored.advance(100); await restored.advance(100); assert.deepEqual(calls.map(([action]) => action), ['query', 'submit']); for (const [, request] of calls) { assert.equal(request.id, before.id); assert.equal(request.body, before.body); } assert.equal(restored.ledger().getSnapshot().requests.size, 1); restored.scheduler.stop(); }); test('storage failure never sends and keeps claims; explicit resume retries persistence first', async () => { let broken = true; const sent = []; const state = setup({ persist: async () => { if (broken) throw new Error('quota'); }, submit: async request => { sent.push(request); return { outcome: 'applied' }; } }); state.add('create'); state.scheduler.start(); await state.scheduler.settle(); assert.equal(sent.length, 0); assert.equal(state.errors[0].message, 'quota'); assert.equal(state.scheduler.getClaims().size, 1); broken = false; await state.scheduler.resume(); await state.scheduler.settle(); // Dispatch evidence means query first, even if the failed barrier preceded the network call. assert.equal(state.calls[0][0], 'query'); assert.equal(sent.length, 0); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); }); test('multi-canvas claims are atomic and stop fences late results', async () => { const response = deferred(), state = setup({ submit: () => response.promise }); state.add('cross', ['a', 'b']); state.add('single', ['b']); state.scheduler.start(); await tick(); assert.equal(new Set(state.scheduler.getClaims().values()).size, 1); assert.equal(state.scheduler.getClaims().size, 2); state.scheduler.stop(); response.resolve({ outcome: 'applied' }); await state.scheduler.settle(); assert.equal(state.ledger().getSnapshot().requests.get('request-1').outcome, 'unresolved'); assert.throws(() => state.scheduler.start(), /scheduler_closed/); }); test('missing frozen-request interpreter preserves bytes and claims while unrelated work proceeds', async () => { const initial = setup({ submit: async () => ({ outcome: 'unresolved' }) }); initial.add('create'); initial.scheduler.start(); await initial.scheduler.settle(); initial.scheduler.stop(); const saved = initial.ledger().getSnapshot(), requests = new Map(saved.requests), old = requests.get('request-1'); requests.set(old.id, { ...old, protocolVersion: 99 }); const restored = setup({ createId: () => 'other-request' }, { ...saved, requests }); restored.add('other', ['other']); restored.scheduler.start(); await restored.scheduler.settle(); assert.equal(restored.calls.length, 1); assert.equal(restored.calls[0][1].id, 'other-request'); assert.equal(restored.ledger().getSnapshot().requests.get(old.id).body, old.body); assert.equal(restored.ledger().getSnapshot().requests.get(old.id).error, 'request_interpreter_unavailable'); assert.equal(restored.scheduler.getClaims().get('canvas'), old.id); restored.scheduler.stop(); }); test('permission/protocol rejection keeps unknown claims without an automatic retry loop', async()=>{ const calls=[],state=setup({submit:async request=>{calls.push(request);return{outcome:'unresolved',blocked:true,error:'identity_changed'}}}); state.add('create');state.scheduler.start();await state.scheduler.settle(); assert.equal(state.ledger().getSnapshot().requests.get('request-1').outcome,'unresolved');assert.equal(state.scheduler.getClaims().size,1);assert.equal(state.timerCount(),0); await state.advance(10000);assert.equal(calls.length,1);state.scheduler.stop(); }); const mixedCompile = (intent, requestId) => ({ kind: intent.operationType, protocolVersion: 2, canvasIds: intent.operationType === 'asset.upload' ? [] : intent.changes.map(change => change.id), expectedVersions: Object.fromEntries(intent.changes.map(change => [change.id, 1])), payload: { requestId, operationId: intent.id } }); function addAsset(state, id) { state.ledger().add({ id, operationType: 'asset.upload', operationSchemaVersion: 1, sequence: state.ledger().getSnapshot().operations.size + 1, origin: 'user', changes: [], canvasIds: [], dependencies: [], resourceId: `resource-${id}` }, {}); state.scheduler.notify([id]); } test('an upload burst yields without filling both network slots or starving independent edits', async () => { const gate = deferred(), calls = []; const state = setup({ compile: mixedCompile, submit: async request => { calls.push(request); if (request.operationIds[0] === 'asset-0') await gate.promise; return { outcome: 'applied' }; } }); for (let index = 0; index < 80; index++) addAsset(state, `asset-${index}`); state.add('independent', ['canvas']); state.scheduler.start(); await tick(); await state.advance(0, false); assert.deepEqual(calls.map(request => request.operationIds[0]), ['asset-0', 'independent']); assert.equal(state.timerCount(), 0); gate.resolve(); await state.scheduler.settle(); assert.equal(calls.length, 81); assert.equal(state.errors.length, 0); state.scheduler.stop(); }); test('a capacity-blocked upload retry does not spin or hide a later document retry deadline', async () => { const gate = deferred(), calls = []; const state = setup({ compile: mixedCompile, submit: async request => { const id = request.operationIds[0]; calls.push(['submit', id]); if (id === 'asset-2') { await gate.promise; return { outcome: 'applied' }; } return { outcome: 'unresolved', retryAfterMs: id === 'doc' ? 200 : 100 }; }, query: async request => { calls.push(['query', request.operationIds[0]]); return { outcome: 'applied' }; }, }); addAsset(state, 'asset-1'); addAsset(state, 'asset-2'); state.add('doc'); state.scheduler.start(); await tick(); await state.advance(100, false); assert.equal(state.timerCount(), 1); await state.advance(100, false); assert.ok(calls.some(([action, id]) => action === 'query' && id === 'doc')); assert.equal(calls.some(([action, id]) => action === 'query' && id === 'asset-1'), false); assert.equal(state.timerCount(), 0); gate.resolve(); await state.scheduler.settle(); assert.ok(calls.some(([action, id]) => action === 'query' && id === 'asset-1')); state.scheduler.stop(); }); test('nested holds finish admitted receipts but suppress dependent dispatch and due retries until the final release', async () => { const response = deferred(), calls = []; const state = setup({ concurrency: 1, submit: async request => { calls.push(['submit', request]); return request.operationIds[0] === 'first' ? response.promise : { outcome: 'applied' }; }, query: async request => { calls.push(['query', request]); return { outcome: 'applied' }; } }); state.add('first'); state.add('dependent', ['canvas'], ['first']); state.scheduler.start(); await tick(); const releaseA = state.scheduler.hold(), releaseB = state.scheduler.hold(); state.add('later', ['other']); response.resolve({ outcome: 'unresolved' }); await state.scheduler.settle(); const frozen = state.ledger().getSnapshot().requests.get('request-1'); assert.equal(frozen.outcome, 'unresolved'); assert.equal(state.scheduler.getClaims().get('canvas'), frozen.id); await state.advance(1000); assert.equal(calls.length, 1); assert.equal(state.timerCount(), 0); releaseA(); releaseA(); await tick(); assert.equal(calls.length, 1); releaseB(); await state.scheduler.settle(); assert.deepEqual(calls.map(([action, request]) => [action, request.operationIds[0]]), [['submit', 'first'], ['query', 'first'], ['submit', 'later'], ['submit', 'dependent']]); assert.equal(calls[1][1].id, frozen.id); assert.equal(calls[1][1].body, frozen.body); assert.equal(state.scheduler.getClaims().size, 0); state.scheduler.stop(); }); test('holds acquired before start retain notified work; stop is terminal even when a hold is later released', async () => { const state = setup(), release = state.scheduler.hold(); state.add('first'); state.scheduler.start(); await tick(); assert.equal(state.calls.length, 0); release(); await state.scheduler.settle(); assert.equal(state.calls.length, 1); const finalRelease = state.scheduler.hold(); state.add('never'); state.scheduler.stop(); finalRelease(); await tick(); assert.equal(state.calls.length, 1); assert.throws(() => state.scheduler.hold(), /scheduler_closed/); });