From 781a3bb2a8abc7911497639aef3cc9d9353dd9bd Mon Sep 17 00:00:00 2001 From: phamngocquy Date: Sun, 26 Jul 2026 23:26:17 +0800 Subject: [PATCH] fix(miner): governor-chokepoint-persisted's load-evaluate-save cycle isn't atomic Fixes #8856 --- .../lib/governor-chokepoint-persisted.ts | 29 ++++---- packages/loopover-miner/lib/governor-state.ts | 12 ++++ .../chokepoint-child.mjs | 66 +++++++++++++++++ ...iner-governor-chokepoint-persisted.test.ts | 71 ++++++++++++++++++- test/unit/miner-governor-state.test.ts | 15 ++++ 5 files changed, 180 insertions(+), 13 deletions(-) create mode 100644 test/fixtures/miner-concurrent-stores/chokepoint-child.mjs diff --git a/packages/loopover-miner/lib/governor-chokepoint-persisted.ts b/packages/loopover-miner/lib/governor-chokepoint-persisted.ts index 24295d75dd..338c32555f 100644 --- a/packages/loopover-miner/lib/governor-chokepoint-persisted.ts +++ b/packages/loopover-miner/lib/governor-chokepoint-persisted.ts @@ -39,18 +39,23 @@ export function evaluateGovernorChokepointGatePersisted( const ownsGovernorState = options.governorState === undefined; const governorState = options.governorState ?? openGovernorState(); try { - const persistedRateLimit = governorState.loadRateLimitState(); - const persistedCapUsage = governorState.loadCapUsage(); - const resolvedInput: GovernorChokepointInput = { - ...input, - rateLimitBuckets: input.rateLimitBuckets ?? persistedRateLimit.buckets, - rateLimitBackoffAttempts: input.rateLimitBackoffAttempts ?? persistedRateLimit.backoffAttempts, - capUsage: input.capUsage ?? persistedCapUsage, - }; - const gateOptions = options.append === undefined ? {} : { append: options.append }; - const result = evaluateGovernorChokepointGate(resolvedInput, gateOptions); - governorState.saveRateLimitState({ buckets: result.rateLimitBuckets, backoffAttempts: result.rateLimitBackoffAttempts }); - return result; + // Load+evaluate+save must share one BEGIN IMMEDIATE (#8856): a read outside the transaction followed by a + // save inside saveRateLimitState's own transaction let two fleet containers both load the same bucket count, + // each advance by one, and the second save clobber the first. + return governorState.withScalarStateTransaction(() => { + const persistedRateLimit = governorState.loadRateLimitState(); + const persistedCapUsage = governorState.loadCapUsage(); + const resolvedInput: GovernorChokepointInput = { + ...input, + rateLimitBuckets: input.rateLimitBuckets ?? persistedRateLimit.buckets, + rateLimitBackoffAttempts: input.rateLimitBackoffAttempts ?? persistedRateLimit.backoffAttempts, + capUsage: input.capUsage ?? persistedCapUsage, + }; + const gateOptions = options.append === undefined ? {} : { append: options.append }; + const result = evaluateGovernorChokepointGate(resolvedInput, gateOptions); + governorState.saveRateLimitState({ buckets: result.rateLimitBuckets, backoffAttempts: result.rateLimitBackoffAttempts }); + return result; + }); } finally { if (ownsGovernorState) governorState.close(); } diff --git a/packages/loopover-miner/lib/governor-state.ts b/packages/loopover-miner/lib/governor-state.ts index 94f6f65407..3f29bbb184 100644 --- a/packages/loopover-miner/lib/governor-state.ts +++ b/packages/loopover-miner/lib/governor-state.ts @@ -55,6 +55,8 @@ export type GovernorPauseInput = { export type GovernorState = { dbPath: string; + /** Runs `fn` inside one `BEGIN IMMEDIATE` on `governor_scalar_state` (#8856). Re-entrant: nested saves skip a second BEGIN. */ + withScalarStateTransaction(fn: () => T): T; loadRateLimitState(): GovernorRateLimitState; saveRateLimitState(rateLimitState: GovernorRateLimitState): void; loadCapUsage(): GovernorCapUsage; @@ -294,8 +296,13 @@ export function openGovernorState(dbPath: string = resolveGovernorStateDbPath()) // 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. + // Re-entrant so governor-chokepoint-persisted.js can wrap load+evaluate+save in one outer transaction while + // still calling saveRateLimitState internally (#8856). + let transactionDepth = 0; function withTransaction(fn: () => T): T { + if (transactionDepth > 0) return fn(); db.exec("BEGIN IMMEDIATE"); + transactionDepth += 1; try { const result = fn(); db.exec("COMMIT"); @@ -303,11 +310,16 @@ export function openGovernorState(dbPath: string = resolveGovernorStateDbPath()) } catch (error) { db.exec("ROLLBACK"); throw error; + } finally { + transactionDepth -= 1; } } const state: GovernorState = { dbPath: resolvedPath, + withScalarStateTransaction(fn: () => T): T { + return withTransaction(fn); + }, loadRateLimitState(): GovernorRateLimitState { const row = getScalarStatement.get() as ScalarStateRow | undefined; return { diff --git a/test/fixtures/miner-concurrent-stores/chokepoint-child.mjs b/test/fixtures/miner-concurrent-stores/chokepoint-child.mjs new file mode 100644 index 0000000000..f83adb9131 --- /dev/null +++ b/test/fixtures/miner-concurrent-stores/chokepoint-child.mjs @@ -0,0 +1,66 @@ +#!/usr/bin/env node +// Cross-process helper for governor-chokepoint-persisted concurrent-race tests (#8856). +// Opens sibling governor-state + ledger handles, waits for a stdin "go" signal, then runs one persisted +// chokepoint evaluation so multiple Node processes contend on the same rate-limit bucket row. +import { evaluateGovernorChokepointGatePersisted } from "../../../packages/loopover-miner/dist/lib/governor-chokepoint-persisted.js"; +import { initGovernorLedger } from "../../../packages/loopover-miner/dist/lib/governor-ledger.js"; +import { openGovernorState } from "../../../packages/loopover-miner/dist/lib/governor-state.js"; + +const [dbPath, ledgerPath, nowMsStr] = process.argv.slice(2); +if (!dbPath || !ledgerPath || !nowMsStr) { + process.stderr.write("usage: chokepoint-child.mjs \n"); + process.exit(2); +} + +const governorState = openGovernorState(dbPath); +const ledger = initGovernorLedger(ledgerPath); +const policies = { + global: { open_pr: { limit: 100, windowMs: 60_000 } }, + perRepo: { open_pr: { limit: 100, windowMs: 60_000 } }, + backoffBaseMs: 100, +}; +const input = { + actionClass: "open_pr", + repoFullName: "acme/widgets", + nowMs: Number(nowMsStr), + wouldBeAction: { action: "open_pr", title: "Fix bug" }, + killSwitchGlobal: false, + killSwitchRepoPaused: false, + liveModeGlobalOptIn: true, + liveModeRepoOptIn: "live", + capLimits: { budget: 100, turns: 100, elapsedMs: 1_000_000 }, + convergenceInput: { attempts: 0, consecutiveFailures: 0, reenqueues: 0, reachedDone: false }, + rateLimitPolicies: policies, +}; + +let started = false; + +function runChokepoint() { + if (started) return; + started = true; + try { + const result = evaluateGovernorChokepointGatePersisted(input, { + governorState, + append: (event) => ledger.appendGovernorEvent(event), + }); + process.stdout.write( + `${JSON.stringify({ + ok: true, + allowed: result.decision.allowed, + count: result.rateLimitBuckets.global.open_pr?.count ?? 0, + })}\n`, + ); + process.exit(0); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stdout.write(`${JSON.stringify({ ok: false, message })}\n`); + process.exit(1); + } finally { + governorState.close(); + ledger.close(); + } +} + +process.stdin.setEncoding("utf8"); +process.stdin.on("data", () => runChokepoint()); +process.stdout.write("READY\n"); diff --git a/test/unit/miner-governor-chokepoint-persisted.test.ts b/test/unit/miner-governor-chokepoint-persisted.test.ts index 808ca89bd2..481a10cbd1 100644 --- a/test/unit/miner-governor-chokepoint-persisted.test.ts +++ b/test/unit/miner-governor-chokepoint-persisted.test.ts @@ -1,6 +1,8 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { join, dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { spawn } from "node:child_process"; import { afterEach, describe, expect, it, vi } from "vitest"; vi.mock("@loopover/engine", async () => { @@ -11,6 +13,11 @@ import { evaluateGovernorChokepointGatePersisted } from "../../packages/loopover import { closeDefaultGovernorLedger, initGovernorLedger, readGovernorEvents } from "../../packages/loopover-miner/lib/governor-ledger.js"; import { openGovernorState } from "../../packages/loopover-miner/lib/governor-state.js"; +const chokepointChildScript = join( + dirname(fileURLToPath(import.meta.url)), + "../fixtures/miner-concurrent-stores/chokepoint-child.mjs", +); + const roots: string[] = []; const closeables: Array<{ close(): void }> = []; @@ -209,4 +216,66 @@ describe("evaluateGovernorChokepointGatePersisted (#5134)", () => { expect(governorState.loadRateLimitState().backoffAttempts["open_pr:acme/widgets"]).toBe(1); }); + + it("REGRESSION (#8856): two overlapping chokepoint evaluations against the same bucket both advance the count (no lost update)", async () => { + // Two real Node processes on one governor-state file, barrier-started together -- the same fleet-container + // race the issue describes. Without one atomic load+evaluate+save transaction both would read count=0 and + // the second save would clobber the first; with the fix the final persisted count must be 2. + const root = mkdtempSync(join(tmpdir(), "loopover-miner-governor-chokepoint-concurrent-")); + roots.push(root); + const dbPath = join(root, "governor-state.sqlite3"); + const ledgerPath = join(root, "governor-ledger.sqlite3"); + const bootstrap = openGovernorState(dbPath); + bootstrap.close(); + + const children = [ + spawn(process.execPath, [chokepointChildScript, dbPath, ledgerPath, "10000"]), + spawn(process.execPath, [chokepointChildScript, dbPath, ledgerPath, "10100"]), + ]; + await Promise.all( + children.map( + (child) => + new Promise((resolve, reject) => { + let buffer = ""; + child.stdout.on("data", (chunk) => { + buffer += chunk.toString(); + if (buffer.includes("READY\n")) resolve(); + }); + child.once("error", reject); + child.once("exit", (code) => { + if (code !== 0 && code !== null) reject(new Error(`child exited before READY (${code})`)); + }); + }), + ), + ); + for (const child of children) child.stdin.write("go\n"); + const results = await Promise.all( + children.map( + (child) => + new Promise<{ ok: boolean; allowed?: boolean; count?: number; message?: string }>((resolve, reject) => { + let stdout = ""; + child.stdout.on("data", (chunk) => { + stdout += chunk.toString(); + }); + child.once("error", reject); + child.once("exit", () => { + const line = stdout + .split("\n") + .map((entry) => entry.trim()) + .find((entry) => entry.startsWith("{")); + if (!line) { + reject(new Error(`child produced no JSON result: ${stdout}`)); + return; + } + resolve(JSON.parse(line) as { ok: boolean; allowed?: boolean; count?: number; message?: string }); + }); + }), + ), + ); + + expect(results.every((result) => result.ok && result.allowed)).toBe(true); + + const reopened = reopenGovernorState(root); + expect(reopened.loadRateLimitState().buckets.global.open_pr?.count).toBe(2); + }); }); diff --git a/test/unit/miner-governor-state.test.ts b/test/unit/miner-governor-state.test.ts index 76a907ba76..8f5bf720f3 100644 --- a/test/unit/miner-governor-state.test.ts +++ b/test/unit/miner-governor-state.test.ts @@ -502,3 +502,18 @@ describe("governor-state scalar-state save atomicity (#7221)", () => { expect(state.loadRateLimitState()).toEqual({ buckets: { global: {}, perRepo: {} }, backoffAttempts: {} }); }); }); + +describe("governor-state withScalarStateTransaction (#8856)", () => { + it("is re-entrant so nested scalar saves inside one outer transaction commit together", () => { + const state = tempState(); + state.withScalarStateTransaction(() => { + state.saveRateLimitState({ + buckets: { global: { open_pr: { count: 3, windowStartMs: 0 } }, perRepo: {} }, + backoffAttempts: {}, + }); + state.savePauseState({ paused: true, reason: "nested pause" }); + }); + expect(state.loadRateLimitState().buckets.global.open_pr?.count).toBe(3); + expect(state.loadPauseState()).toMatchObject({ paused: true, reason: "nested pause" }); + }); +});