diff --git a/packages/loopover-miner/lib/governor-state.js b/packages/loopover-miner/lib/governor-state.js index b35cd477ba..4755b43ccc 100644 --- a/packages/loopover-miner/lib/governor-state.js +++ b/packages/loopover-miner/lib/governor-state.js @@ -208,6 +208,23 @@ export function openGovernorState(dbPath = resolveGovernorStateDbPath()) { }; } + // BEGIN IMMEDIATE takes the write lock BEFORE `fn`'s read, so two processes on the same file (the loop daemon + // saving rate-limit/cap-usage state on every gated write, and an operator's `governor pause`/`resume` CLI + // invocation racing it) cannot interleave a stale read with each other's write and silently clobber the + // scalar-state column-group they don't own -- same fix shape as event-ledger.js's appendEvent (#7221). Shared + // by all three governor_scalar_state save methods below, since they all read-then-write across the same row. + function withTransaction(fn) { + db.exec("BEGIN IMMEDIATE"); + try { + const result = fn(); + db.exec("COMMIT"); + return result; + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } + } + const state = { dbPath: resolvedPath, loadRateLimitState() { @@ -218,32 +235,36 @@ export function openGovernorState(dbPath = resolveGovernorStateDbPath()) { }; }, saveRateLimitState(rateLimitState) { - const row = getScalarStatement.get(); - upsertScalarStatement.run( - JSON.stringify(rateLimitState?.buckets ?? DEFAULT_RATE_LIMIT_BUCKETS), - JSON.stringify(rateLimitState?.backoffAttempts ?? DEFAULT_RATE_LIMIT_BACKOFF), - row ? row.cap_usage_json : JSON.stringify(DEFAULT_CAP_USAGE), - row ? row.paused : 0, - row ? row.pause_reason : null, - row ? row.paused_at : null, - new Date().toISOString(), - ); + withTransaction(() => { + const row = getScalarStatement.get(); + upsertScalarStatement.run( + JSON.stringify(rateLimitState?.buckets ?? DEFAULT_RATE_LIMIT_BUCKETS), + JSON.stringify(rateLimitState?.backoffAttempts ?? DEFAULT_RATE_LIMIT_BACKOFF), + row ? row.cap_usage_json : JSON.stringify(DEFAULT_CAP_USAGE), + row ? row.paused : 0, + row ? row.pause_reason : null, + row ? row.paused_at : null, + new Date().toISOString(), + ); + }); }, loadCapUsage() { const row = getScalarStatement.get(); return parseJsonColumn(row?.cap_usage_json, DEFAULT_CAP_USAGE); }, saveCapUsage(capUsage) { - const row = getScalarStatement.get(); - upsertScalarStatement.run( - row ? row.rate_limit_buckets_json : JSON.stringify(DEFAULT_RATE_LIMIT_BUCKETS), - row ? row.rate_limit_backoff_json : JSON.stringify(DEFAULT_RATE_LIMIT_BACKOFF), - JSON.stringify(capUsage ?? DEFAULT_CAP_USAGE), - row ? row.paused : 0, - row ? row.pause_reason : null, - row ? row.paused_at : null, - new Date().toISOString(), - ); + withTransaction(() => { + const row = getScalarStatement.get(); + upsertScalarStatement.run( + row ? row.rate_limit_buckets_json : JSON.stringify(DEFAULT_RATE_LIMIT_BUCKETS), + row ? row.rate_limit_backoff_json : JSON.stringify(DEFAULT_RATE_LIMIT_BACKOFF), + JSON.stringify(capUsage ?? DEFAULT_CAP_USAGE), + row ? row.paused : 0, + row ? row.pause_reason : null, + row ? row.paused_at : null, + new Date().toISOString(), + ); + }); }, // The governor pause/resume control surface (#4851): a real, persisted, operator/governor-writable flag the // loop checks before each cycle -- distinct from governor-kill-switch.js (a read-only resolver over env/YAML @@ -259,20 +280,22 @@ export function openGovernorState(dbPath = resolveGovernorStateDbPath()) { }; }, savePauseState(pauseState) { - const row = getScalarStatement.get(); const paused = Boolean(pauseState?.paused); const reason = typeof pauseState?.reason === "string" && pauseState.reason.trim() ? pauseState.reason.trim() : null; const pausedAt = paused ? new Date().toISOString() : null; - upsertScalarStatement.run( - row ? row.rate_limit_buckets_json : JSON.stringify(DEFAULT_RATE_LIMIT_BUCKETS), - row ? row.rate_limit_backoff_json : JSON.stringify(DEFAULT_RATE_LIMIT_BACKOFF), - row ? row.cap_usage_json : JSON.stringify(DEFAULT_CAP_USAGE), - paused ? 1 : 0, - reason, - pausedAt, - new Date().toISOString(), - ); + withTransaction(() => { + const row = getScalarStatement.get(); + upsertScalarStatement.run( + row ? row.rate_limit_buckets_json : JSON.stringify(DEFAULT_RATE_LIMIT_BUCKETS), + row ? row.rate_limit_backoff_json : JSON.stringify(DEFAULT_RATE_LIMIT_BACKOFF), + row ? row.cap_usage_json : JSON.stringify(DEFAULT_CAP_USAGE), + paused ? 1 : 0, + reason, + pausedAt, + new Date().toISOString(), + ); + }); return { paused, reason, pausedAt }; }, loadReputationHistory(repoFullName, apiBaseUrl) { diff --git a/test/unit/miner-governor-state.test.ts b/test/unit/miner-governor-state.test.ts index fd6b84a7fd..b4a570f5a2 100644 --- a/test/unit/miner-governor-state.test.ts +++ b/test/unit/miner-governor-state.test.ts @@ -61,6 +61,12 @@ describe("governor-state rate-limit state (#5134)", () => { state.saveRateLimitState({ buckets: { global: {}, perRepo: {} }, backoffAttempts: {} }); expect(state.loadCapUsage()).toEqual({ budgetSpent: 42, turnsTaken: 3, elapsedMs: 5000 }); }); + + it("defaults omitted buckets/backoffAttempts to empty, rather than throwing", () => { + const state = tempState(); + state.saveRateLimitState({} as never); + expect(state.loadRateLimitState()).toEqual({ buckets: { global: {}, perRepo: {} }, backoffAttempts: {} }); + }); }); describe("governor-state cap usage (#5134)", () => { @@ -76,6 +82,12 @@ describe("governor-state cap usage (#5134)", () => { expect(state.loadCapUsage()).toEqual({ budgetSpent: 10, turnsTaken: 1, elapsedMs: 100 }); expect(state.loadRateLimitState().buckets.global.open_pr?.count).toBe(1); }); + + it("defaults an omitted cap usage to zeroed, rather than throwing", () => { + const state = tempState(); + state.saveCapUsage(undefined as never); + expect(state.loadCapUsage()).toEqual({ budgetSpent: 0, turnsTaken: 0, elapsedMs: 0 }); + }); }); describe("governor-state pause/resume control surface (#4851)", () => { @@ -435,3 +447,46 @@ describe("governor-state module-level default singleton (#5134)", () => { expect(loadReputationHistory("acme/widgets")).toEqual({ decided: 0, unfavorable: 0 }); }); }); + +describe("governor-state scalar-state save atomicity (#7221)", () => { + it("REGRESSION: two sibling connections to the same file preserve each other's already-committed column-group across saves", () => { + // Two DatabaseSync-backed handles on ONE file are exactly the two sibling miner PROCESSES the issue + // describes (the loop daemon vs. an operator's `governor pause` CLI invocation) -- SQLite's file locking + // treats them identically, same technique as claim-ledger.test.ts's own #6758 regression test. + const root = mkdtempSync(join(tmpdir(), "loopover-miner-governor-state-shared-")); + roots.push(root); + const dbPath = join(root, "shared-governor-state.sqlite3"); + const processA = openGovernorState(dbPath); + const processB = openGovernorState(dbPath); + states.push(processA, processB); + + processA.saveRateLimitState({ buckets: { global: { open_pr: { count: 5, windowStartMs: 0 } }, perRepo: {} }, backoffAttempts: {} }); + const written = processB.savePauseState({ paused: true, reason: "operator pause" }); + expect(written).toMatchObject({ paused: true, reason: "operator pause" }); + + // B's pause save must not have clobbered A's already-committed rate-limit state back to defaults. + expect(processA.loadRateLimitState().buckets.global.open_pr?.count).toBe(5); + expect(processB.loadRateLimitState().buckets.global.open_pr?.count).toBe(5); + + processA.saveCapUsage({ budgetSpent: 99, turnsTaken: 4, elapsedMs: 2000 }); + + // A's cap-usage save must not have clobbered B's already-committed pause state back to unpaused. + expect(processB.loadPauseState()).toMatchObject({ paused: true, reason: "operator pause" }); + expect(processA.loadCapUsage()).toEqual({ budgetSpent: 99, turnsTaken: 4, elapsedMs: 2000 }); + }); + + it("REGRESSION: rolls the transaction back if the upsert throws, leaving state unchanged and the handle usable", () => { + const state = tempState(); + state.saveCapUsage({ budgetSpent: 10, turnsTaken: 1, elapsedMs: 100 }); + + // A function value JSON.stringifies to `undefined`, which node:sqlite refuses to bind -- this throws INSIDE + // withTransaction's try block (after BEGIN IMMEDIATE), exercising the ROLLBACK path. + expect(() => state.saveRateLimitState({ buckets: (() => {}) as never, backoffAttempts: {} })).toThrow(); + + // Rolled back: cap usage from before the throwing call is untouched, not partially overwritten. + expect(state.loadCapUsage()).toEqual({ budgetSpent: 10, turnsTaken: 1, elapsedMs: 100 }); + // No stranded open transaction: a subsequent valid save on the SAME handle still succeeds. + state.saveRateLimitState({ buckets: { global: {}, perRepo: {} }, backoffAttempts: {} }); + expect(state.loadRateLimitState()).toEqual({ buckets: { global: {}, perRepo: {} }, backoffAttempts: {} }); + }); +});