// ── Giving back what a run took (EVENTS_PLAN.md Phase 8) ─────────────────── // // The phase's shipped claim: **cleanup is generated from the ledger and runs on // every terminal path.** An operator cannot be relied on to write the undo, and // an aborted run never reaches the phase they wrote it in — so there is no // cleanup phase in a spec, no `on_teardown` on an action, and one function that // reads rows. // // The properties around it are §L's, and two of them are ones this codebase has // already paid for once: // // • **a revert that never succeeds stays visible rather than cycling** — rule 2, // and `MAX_REVERT_ATTEMPTS` is what stops the automatic retry. Only a human // clears the counter, which is Engagement Phase 14's rule stated a third time // • **reverting something that does not exist is a SUCCESS** — §L, and what a // Rust wipe needs // • **drift is not an error.** The module did exactly what it was asked and // found somebody else's value in place. A restore that wrote anyway would // silently revert an operator's manual fix // • **a resource the module no longer has becomes `orphaned`, never `reverted`** — // reverting it would be core recording that it put something back when what // happened is that the thing vanished // • **"I do not know" is never read as "it is gone".** Every unanswerable // reconcile leaves the ledger alone // // The sweep is driven against a stubbed db layer, exactly as `eventRunner.test.js` // drives the runner: what a stub cannot prove is the SQL, and the unique key that // makes two events unable to lease one target runs against a real MariaDB in // `eventRunnerSql.test.js`. process.env.DB_HOST = '127.0.0.1' process.env.DB_PORT = '59999' const { test, beforeEach, afterEach, after } = require('node:test') const assert = require('node:assert/strict') const registries = require('../src/modules/registries') const cleanup = require('../src/events/cleanup') const resourcesDb = require('../src/model/events/eventRunResources.db') const runsDb = require('../src/model/events/eventRuns.db') const stepsDb = require('../src/model/events/eventRunSteps.db') const logDb = require('../src/model/events/eventRunLog.db') const db = require('../src/utils/db') after(() => db.close()) const UNRESOLVED = ['pending', 'confirmed', 'reverting', 'orphaned', 'drifted'] let store const originals = { resourcesDb: { ...resourcesDb }, runsDb: { ...runsDb }, stepsDb: { ...stepsDb }, logDb: { ...logDb }, } beforeEach(() => { registries._reset() store = { rows: new Map(), steps: new Map(), log: [], next: 1, cleanupStatus: 'pending' } resourcesDb.unresolvedForRun = async (runId, { maxAttempts = null } = {}) => [...store.rows.values()] .filter((r) => r.run_id === runId && UNRESOLVED.includes(r.status)) .filter((r) => maxAttempts === null || r.revert_attempts < maxAttempts) .map((r) => ({ ...r })) resourcesDb.unresolvedCount = async (runId) => [...store.rows.values()].filter((r) => r.run_id === runId && UNRESOLVED.includes(r.status)).length resourcesDb.claimRevert = async (id) => { const r = store.rows.get(id) if (!r || !['pending', 'confirmed', 'orphaned', 'drifted'].includes(r.status)) return false r.status = 'reverting' return true } resourcesDb.markReverted = async (id) => { const r = store.rows.get(id) if (r) Object.assign(r, { status: 'reverted', last_error: null }) } resourcesDb.failRevert = async (id, error, restoreTo = 'confirmed') => { const r = store.rows.get(id) if (r) Object.assign(r, { status: restoreTo, revert_attempts: r.revert_attempts + 1, last_error: String(error) }) } resourcesDb.resetAttempts = async (runId) => { let n = 0 for (const r of store.rows.values()) { if (r.run_id === runId && UNRESOLVED.includes(r.status)) { r.revert_attempts = 0 n += 1 } } return n } resourcesDb.markOrphaned = async (id, detail = null) => { const r = store.rows.get(id) if (r && ['pending', 'confirmed', 'reverting'].includes(r.status)) { Object.assign(r, { status: 'orphaned', last_error: detail }) } } resourcesDb.liveForModule = async (owner) => [...store.rows.values()].filter((r) => r.owner_module === owner && ['pending', 'confirmed'].includes(r.status)).map((r) => ({ ...r })) resourcesDb.modulesWithLiveRows = async () => [ ...new Set([...store.rows.values()].filter((r) => ['pending', 'confirmed'].includes(r.status)).map((r) => r.owner_module)), ] resourcesDb.runsNeedingCleanup = async () => store.candidates || [] runsDb.setCleanupStatus = async (id, to, from = null) => { if (from && !from.includes(store.cleanupStatus)) return false store.cleanupStatus = to return true } stepsDb.getById = async (id) => store.steps.get(id) || null logDb.write = async (entry) => { store.log.push(entry) } }) afterEach(() => { Object.assign(resourcesDb, originals.resourcesDb) Object.assign(runsDb, originals.runsDb) Object.assign(stepsDb, originals.stepsDb) Object.assign(logDb, originals.logDb) registries._reset() }) const RUN = { id: 7, status: 'completed', cleanup_status: 'pending' } function addStep(id, actionId, key = 'k'.repeat(40)) { store.steps.set(id, { id, run_id: RUN.id, action_id: actionId, idempotency_key: key, phase: 'p', seq: 0 }) } function addResource(over = {}) { const id = store.next++ const row = { id, run_id: RUN.id, step_id: 1, owner_module: 'demo', kind: 'creature', ref: `0x${id}`, payload: null, lease_until: null, status: 'confirmed', revert_attempts: 0, last_error: null, member_key: null, ...over, } store.rows.set(id, row) return row } function registerAction(over = {}) { const api = registries.stage('demo') api.registerEventActions([ { id: 'demo.spawn', label: 'Spawn', risk: 'change', reversible: 'ledger', budgetMs: 500, perform: async () => ({ ok: true }), revert: async () => ({ ok: true }), ...over, }, ]) registries.apply(api.staged) } function registerLease(over = {}) { const api = registries.stage('demo') api.registerEventLeases([ { id: 'demo.rate', label: 'Gather rate', type: 'float', min: 0.5, max: 5, maxDurationMs: 3_600_000, read: async () => ({ ok: true, value: 1 }), apply: async () => ({ ok: true }), restore: async () => ({ ok: true }), ...over, }, ]) registries.apply(api.staged) } const kinds = () => store.log.map((e) => e.kind) // ── The shipped claim ────────────────────────────────────────────────────── test('a run\'s ledger is given back in one call per step, and the run goes complete', async () => { // `revert` takes a LIST because that is what makes twelve creatures one round // trip rather than twelve. The grouping key is the STEP, because the resource // row records the module and the opaque names while the step records the verb. let seen = null registerAction({ revert: async (arg) => { seen = arg; return { ok: true } } }) addStep(1, 'demo.spawn', 'key-one') addResource({ ref: '0xA' }) addResource({ ref: '0xB' }) const summary = await cleanup.cleanupRun(RUN) assert.deepEqual(summary, { attempted: 2, reverted: 2, drifted: 0, failed: 0, remaining: 0 }) assert.equal(seen.runId, RUN.id) assert.equal(seen.idempotencyKey, 'key-one') assert.deepEqual(seen.resources.map((r) => r.ref), ['0xA', '0xB']) assert.equal(store.cleanupStatus, 'complete') }) test('two steps are two calls, and one failing does not take the other down', async () => { const calls = [] registerAction({ revert: async ({ idempotencyKey, resources }) => { calls.push(idempotencyKey) return idempotencyKey === 'bad' ? { ok: false, error: 'the shard did not answer' } : { ok: true, resources } }, }) addStep(1, 'demo.spawn', 'good') addStep(2, 'demo.spawn', 'bad') addResource({ step_id: 1 }) addResource({ step_id: 2 }) const summary = await cleanup.cleanupRun(RUN) assert.equal(calls.length, 2) assert.equal(summary.reverted, 1) assert.equal(summary.failed, 1) assert.equal(summary.remaining, 1) // **Still `pending`, because the failed row has retries left.** `incomplete` // means "finished with, and not finished" — it is what takes a run out of the // sweep's own scan, so writing it after the FIRST failure made // `MAX_REVERT_ATTEMPTS` quietly mean one attempt. Found by watching // `revert_attempts` sit at 1 through half a minute of live ticks. assert.equal(store.cleanupStatus, 'pending') }) test('reverting something that does not exist is a success', async () => { // §L, and the Rust wipe: "gone, and that is fine". The module never has to // distinguish "I deleted it" from "it was not there" — which is also what makes // a placeholder for an object that may never have existed safe to write. registerAction({ revert: async () => ({ ok: true, detail: 'resource no longer exists' }) }) addStep(1, 'demo.spawn') addResource() const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.reverted, 1) assert.equal(store.cleanupStatus, 'complete') }) test('a module may name the ones that did not come back', async () => { // Partial cleanup is the ordinary case — eleven of twelve creatures deleted — // and it is why the ledger is a row per object rather than a row per step. registerAction({ revert: async () => ({ ok: true, failed: ['0x2'] }) }) addStep(1, 'demo.spawn') addResource({ ref: '0x1' }) addResource({ ref: '0x2' }) const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.reverted, 1) assert.equal(summary.failed, 1) assert.match([...store.rows.values()].find((r) => r.ref === '0x2').last_error, /could not give "0x2" back/) }) // ── Rule 2: loud and sticky ──────────────────────────────────────────────── test('a revert that never works stops retrying and stays visible', async () => { let calls = 0 registerAction({ revert: async () => { calls += 1; return { ok: false, error: 'nope' } } }) addStep(1, 'demo.spawn') addResource() for (let i = 0; i < 6; i++) await cleanup.cleanupRun(RUN) // Bounded at MAX_REVERT_ATTEMPTS, exactly like a step's attempts. A fourth ask // of a shard that has answered the same way three times is not new information, // and an unbounded counter is a row nothing can ever sweep. assert.equal(calls, cleanup.MAX_REVERT_ATTEMPTS) const row = [...store.rows.values()][0] assert.equal(row.revert_attempts, cleanup.MAX_REVERT_ATTEMPTS) assert.equal(row.status, 'confirmed') assert.equal(row.last_error, 'nope') // And ONLY now, with nothing left to try, does the run stop being the sweep's // business. §L: it does not stay `running` — an event whose world changes are // still up is a real state, and pretending the event is in progress hides it. assert.equal(store.cleanupStatus, 'incomplete') }) test('the run goes back to pending each time it still has an attempt left', async () => { // The other half of the same rule, watched one pass at a time rather than at // the end. Each of the first two sweeps leaves the run in the scan; the third // takes it out. A test that only looked at the end state would pass against the // defect this replaced. registerAction({ revert: async () => ({ ok: false, error: 'nope' }) }) addStep(1, 'demo.spawn') addResource() const seen = [] for (let i = 0; i < 3; i++) { await cleanup.cleanupRun(RUN) seen.push(store.cleanupStatus) } assert.deepEqual(seen, ['pending', 'pending', 'incomplete']) }) test('only a human clears the attempt counter', async () => { // Engagement Phase 14's defect, stated a third time: a SWEEP that returned every // stale row to its start state made the attempt ceiling unreachable, so the row // cycled for ever and was never eligible for any retention sweep. The automatic // leg must never do this; the cleanup route may, because a person asked. let calls = 0 registerAction({ revert: async () => { calls += 1; return { ok: false, error: 'nope' } } }) addStep(1, 'demo.spawn') addResource() for (let i = 0; i < 5; i++) await cleanup.cleanupRun(RUN) assert.equal(calls, cleanup.MAX_REVERT_ATTEMPTS) await cleanup.cleanupRun(RUN, { resetAttempts: true, actor: 9 }) assert.equal(calls, cleanup.MAX_REVERT_ATTEMPTS + 1) assert.ok(store.log.some((e) => e.kind === 'cleanup.retry' && e.detail.by === 9)) }) test('a module that throws from revert is a transient failure, not a crashed sweep', async () => { registerAction({ revert: async () => { throw new Error('socket hung up') } }) addStep(1, 'demo.spawn') addResource() const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.failed, 1) assert.match([...store.rows.values()][0].last_error, /socket hung up/) }) test('no shape a revert failure can take reads as success', () => { // `dispatch.classify`'s rule, applied to the other direction of the contract. // The expensive mistake here is the mirror of the one there: recording that a // world change was UNDONE when it was not. for (const raw of [null, undefined, 'ok', [], {}, { ok: 'yes' }, { ok: 1 }]) { assert.notEqual(cleanup.classifyRevert(raw, 'demo.spawn').outcome, 'done', JSON.stringify(raw)) } assert.equal(cleanup.classifyRevert({ __timedOut: true, error: 'slow' }, 'x').outcome, 'retry') assert.equal(cleanup.classifyRevert({ ok: false, retry: false, error: 'never' }, 'x').outcome, 'terminal') assert.equal(cleanup.classifyRevert({ ok: true }, 'x').outcome, 'done') }) test('an action whose module is gone leaves its rows unresolved with the reason', async () => { // Not a retry — nothing will change until an operator reinstalls it — and not // an orphan either, because core has no idea whether the thing is still there. addStep(1, 'demo.spawn') addResource() const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.failed, 1) assert.match([...store.rows.values()][0].last_error, /no module registers "demo.spawn"/) // Retried like any other failure rather than given up on at once, and that is // the right uniformity here: "nothing registers this" stops being true the // moment an operator reinstalls the module, and three registry lookups cost // nothing. So it is `pending` until the attempts are spent. assert.equal(store.cleanupStatus, 'pending') }) // ── Leases ───────────────────────────────────────────────────────────────── test('a lease is restored through the LEASE registry, not through any action', async () => { // The split §F draws: core owns the duration and the conflict check, the module // owns reading and writing. It is why `core.lease` needs no `revert()` of its // own, and why an `override` row routes here rather than to its step's action. let seen = null registerLease({ restore: async (baseline, opts) => { seen = { baseline, opts }; return { ok: true } } }) addResource({ kind: 'override', ref: 'demo.rate', step_id: null, payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.reverted, 1) assert.equal(seen.baseline, 1) // The drift check's input. `restore` MUST verify current === expected before // writing, and a lease whose restore wrote blindly would silently revert an // operator's manual fix. assert.equal(seen.opts.expected, 3) }) test('drift is not an error: the world is left alone and the row says so', async () => { registerLease({ restore: async () => ({ ok: false, drifted: true, current: 4.5 }) }) addResource({ kind: 'override', ref: 'demo.rate', step_id: null, payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.drifted, 1) assert.equal(summary.failed, 0) const row = [...store.rows.values()][0] assert.equal(row.status, 'drifted') assert.match(row.last_error, /now 4\.5 rather than what this run applied/) // Still surfaced. §L: "surfaced beside the unreverted ones" — the run does not // get to call itself clean because somebody else took the value. It is `pending` // rather than `incomplete` for one more reason worth keeping: drift is retried // like any other failure, because a GM who puts the value back between two ticks // should have the lease close cleanly. assert.equal(store.cleanupStatus, 'pending') }) test('a lease whose module is uninstalled is unresolved, never assumed restored', async () => { addResource({ kind: 'override', ref: 'demo.rate', step_id: null, payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.cleanupRun(RUN) assert.equal(summary.failed, 1) assert.match([...store.rows.values()][0].last_error, /no module registers the lease "demo.rate"/) }) // ── The sweep ────────────────────────────────────────────────────────────── test('the sweep only touches TERMINAL runs', async () => { // A run still in flight has a ledger that is still growing, and reverting a // resource the next step is about to use would be core undoing an event while // it is happening. registerAction() addStep(1, 'demo.spawn') addResource() store.candidates = [{ id: RUN.id, status: 'running', cleanup_status: 'pending' }] assert.equal(await cleanup.sweep(), 0) assert.equal([...store.rows.values()][0].status, 'confirmed') store.candidates = [{ id: RUN.id, status: 'cancelled', cleanup_status: 'pending' }] assert.equal(await cleanup.sweep(), 1) assert.equal([...store.rows.values()][0].status, 'reverted') }) // ── Reconcile ────────────────────────────────────────────────────────────── test('a resource the module no longer has becomes orphaned, never reverted', async () => { // §L, and the distinction matters to the operator reading the console // afterwards: `reverted` says core put something back, `orphaned` says the // thing vanished while nobody was looking. Recording the second as the first // would be core claiming credit for a shard restart. registerAction({ reconcile: async () => ({ ok: true, inForce: ['0x1'] }) }) addStep(1, 'demo.spawn') addResource({ ref: '0x1' }) addResource({ ref: '0x2' }) const summary = await cleanup.reconcileModule('demo') assert.deepEqual(summary, { asked: 2, inForce: 1, orphaned: 1, unanswered: 0 }) assert.equal([...store.rows.values()].find((r) => r.ref === '0x2').status, 'orphaned') assert.equal([...store.rows.values()].find((r) => r.ref === '0x1').status, 'confirmed') assert.ok(store.log.some((e) => e.kind === 'resource.orphaned')) }) test('"I do not know" is never read as "it is gone"', async () => { // Every unanswerable shape leaves the ledger exactly as it was. A reconcile // that read silence as absence would orphan a whole shard's worth of live // spawns the first time a sidecar was slow. for (const answer of [null, undefined, { ok: false }, { ok: true }, { ok: true, inForce: 'all' }, 'yes']) { store.rows.clear() store.log.length = 0 registries._reset() registerAction({ reconcile: async () => answer }) addStep(1, 'demo.spawn') addResource({ ref: '0x1' }) const summary = await cleanup.reconcileModule('demo') assert.equal(summary.orphaned, 0, JSON.stringify(answer)) assert.equal(summary.unanswered, 1, JSON.stringify(answer)) assert.equal([...store.rows.values()][0].status, 'confirmed') } }) test('a module with no reconcile is not broken; core keeps believing its ledger', async () => { // Optional where `revert` is required. A module that cannot answer leaves core // exactly where it was before this phase, which is a capability its deployment // does without rather than a boot it fails. registerAction() addStep(1, 'demo.spawn') addResource() const summary = await cleanup.reconcileModule('demo') assert.deepEqual(summary, { asked: 0, inForce: 0, orphaned: 0, unanswered: 1 }) assert.equal([...store.rows.values()][0].status, 'confirmed') }) test('a reconcile that throws orphans nothing', async () => { registerAction({ reconcile: async () => { throw new Error('sidecar gone') } }) addStep(1, 'demo.spawn') addResource() const summary = await cleanup.reconcileModule('demo') assert.equal(summary.unanswered, 1) assert.equal([...store.rows.values()][0].status, 'confirmed') }) test('placeholders are not asked about, because there is nothing to ask yet', async () => { // A `@step` row names no object — it says "a dispatch was in flight and may // have made something". Asking a module whether it is in force is a question // with no answer, and reading a shrug as absence would resolve the one row whose // survival is the safety property. registerAction({ reconcile: async () => ({ ok: true, inForce: [] }) }) addStep(1, 'demo.spawn') addResource({ kind: resourcesDb.STEP_KIND, ref: 'a'.repeat(40), status: 'pending' }) const summary = await cleanup.reconcileModule('demo') assert.deepEqual(summary, { asked: 0, inForce: 0, orphaned: 0, unanswered: 0 }) assert.equal([...store.rows.values()][0].status, 'pending') }) // ── Reconciling a LEASE (Phase 11b) ──────────────────────────────────────── /// A lease row's action is `core.lease`, so these need core's own registrations. function registerCoreAnd(leaseOver = {}) { registries.registerCore() registerLease(leaseOver) } test('a lease row had no reconcile path at all until core.lease grew one', async () => { // The hole this phase closed, asserted from the outside. `reconcileModule` // resolves a resource to the action of the step that made it, and for a lease // that action is CORE's — a path no module can register anything on. So every // `override` row came back `unanswered` for the life of the run, and a lease the // shard had quietly dropped stayed in the ledger as live until teardown went // looking for a baseline nobody was holding. registerCoreAnd({ inForce: async () => ({ ok: true, held: false }) }) addStep(1, 'core.lease') addResource({ kind: 'override', ref: 'demo.rate', payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.reconcileModule('demo') assert.deepEqual(summary, { asked: 1, inForce: 0, orphaned: 1, unanswered: 0 }) assert.equal([...store.rows.values()][0].status, 'orphaned') }) test('a lease the shard still has a record of stays put', async () => { registerCoreAnd({ inForce: async () => ({ ok: true, held: true }) }) addStep(1, 'core.lease') addResource({ kind: 'override', ref: 'demo.rate', payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.reconcileModule('demo') assert.deepEqual(summary, { asked: 1, inForce: 1, orphaned: 0, unanswered: 0 }) assert.equal([...store.rows.values()][0].status, 'confirmed') }) test('a lease that cannot say is left alone, and DRIFT is not what this asks about', async () => { // Two properties in one walk. Every unanswerable shape leaves the row exactly as // it was, which is core's posture everywhere else — and a lease with no // `inForce()` at all is one of those shapes rather than a boot failure. // // The second is the reason `inForce` exists instead of a comparison against // `read()`: a value that differs from what the event applied is drift, and drift // is teardown's verdict to deliver through `restore` so the row lands `drifted` // with the current value beside it. A reconcile that inferred absence from a // changed value would orphan the row first and throw that away — telling the // operator the lease vanished rather than that somebody moved it. const shrugs = [ undefined, // no inForce() declared at all async () => null, async () => ({ ok: false, held: false }), // could not ask; not an answer async () => ({ ok: true }), // answered without saying async () => { throw new Error('sidecar gone') }, // The drift shape. `read()` would report 4.5 against an applied 3, and this // must NOT be read as "the lease is gone". async () => ({ ok: true, held: true, current: 4.5 }), ] for (const inForce of shrugs) { store.rows.clear() store.log.length = 0 registries._reset() registerCoreAnd(inForce === undefined ? {} : { inForce }) addStep(1, 'core.lease') addResource({ kind: 'override', ref: 'demo.rate', payload: { baseline: 1, applied: 3 } }) const summary = await cleanup.reconcileModule('demo') assert.equal(summary.orphaned, 0, String(inForce)) assert.equal([...store.rows.values()][0].status, 'confirmed', String(inForce)) } }) test('reconcileAll asks every module that owns a live row', async () => { registerAction({ reconcile: async () => ({ ok: true, inForce: [] }) }) addStep(1, 'demo.spawn') addStep(2, 'other.thing') addResource({ owner_module: 'demo' }) addResource({ owner_module: 'other', step_id: 2 }) const out = await cleanup.reconcileAll() assert.deepEqual(Object.keys(out).sort(), ['demo', 'other']) assert.equal(out.demo.orphaned, 1) // The other module registers nothing, so its row is left alone rather than // orphaned by a module that is not there to be asked. assert.equal(out.other.unanswered, 1) })