fix(miner): governor-chokepoint-persisted's load-evaluate-save cycle isn't atomic

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();
   }

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<T>(fn: () => T): T;
   loadRateLimitState(): GovernorRateLimitState;
   saveRateLimitState(rateLimitState: GovernorRateLimitState): void;
   loadCapUsage(): GovernorCapUsage;
@@ -294,20 +296,30 @@ 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<T>(fn: () => T): T {
+    if (transactionDepth > 0) return fn();
     db.exec("BEGIN IMMEDIATE");
+    transactionDepth += 1;
     try {
       const result = fn();
       db.exec("COMMIT");
       return result;
     } catch (error) {
       db.exec("ROLLBACK");
       throw error;
+    } finally {
+      transactionDepth -= 1;
     }
   }
 
   const state: GovernorState = {
     dbPath: resolvedPath,
+    withScalarStateTransaction<T>(fn: () => T): T {
+      return withTransaction(fn);
+    },
     loadRateLimitState(): GovernorRateLimitState {
       const row = getScalarStatement.get() as ScalarStateRow | undefined;
       return {

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 <dbPath> <ledgerPath> <nowMs>\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");

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<void>((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);
+  });
 });

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" });
+  });
+});