diff --git a/packages/loopover-miner/lib/governor-state.ts b/packages/loopover-miner/lib/governor-state.ts index 94f6f65407..867197cd9f 100644 --- a/packages/loopover-miner/lib/governor-state.ts +++ b/packages/loopover-miner/lib/governor-state.ts @@ -63,6 +63,11 @@ export type GovernorState = { savePauseState(pauseState: GovernorPauseInput): GovernorPauseState; loadReputationHistory(repoFullName: string, apiBaseUrl?: string): RepoOutcomeHistory; saveReputationHistory(repoFullName: string, history: RepoOutcomeHistory, apiBaseUrl?: string): RepoOutcomeHistory; + incrementReputationHistory( + repoFullName: string, + delta: { decided?: number; unfavorable?: number }, + apiBaseUrl?: string, + ): RepoOutcomeHistory; recordOwnSubmission(record: OwnSubmissionRecord): OwnSubmissionRecord; listRecentOwnSubmissions(filter?: ListRecentOwnSubmissionsFilter): OwnSubmissionRecord[]; /** Delete every repo-scoped row for one repo across both governor tables (#7091); returns total rows removed. */ @@ -394,6 +399,24 @@ export function openGovernorState(dbPath: string = resolveGovernorStateDbPath()) upsertReputationStatement.run(normalizedForge, normalizedRepo, decided, unfavorable, new Date().toISOString()); return { decided, unfavorable }; }, + incrementReputationHistory( + repoFullName: string, + delta: { decided?: number; unfavorable?: number }, + apiBaseUrl?: string, + ): RepoOutcomeHistory { + const normalizedForge = normalizeApiBaseUrl(apiBaseUrl); + const normalizedRepo = normalizeRepoFullName(repoFullName); + const decidedDelta = Number.isInteger(delta?.decided) ? Number(delta.decided) : 0; + const unfavorableDelta = Number.isInteger(delta?.unfavorable) ? Number(delta.unfavorable) : 0; + return withTransaction(() => { + const row = getReputationStatement.get(normalizedForge, normalizedRepo) as ReputationHistoryRow | undefined; + const prior = row ? { decided: row.decided, unfavorable: row.unfavorable } : { ...DEFAULT_REPUTATION_HISTORY }; + const decided = prior.decided + decidedDelta; + const unfavorable = prior.unfavorable + unfavorableDelta; + upsertReputationStatement.run(normalizedForge, normalizedRepo, decided, unfavorable, new Date().toISOString()); + return { decided, unfavorable }; + }); + }, recordOwnSubmission(record: OwnSubmissionRecord): OwnSubmissionRecord { const normalized = normalizeRepoFullName(record?.repoFullName); if (typeof record?.fingerprint !== "string" || !record.fingerprint.trim()) { @@ -471,6 +494,14 @@ export function saveReputationHistory(repoFullName: string, history: RepoOutcome return getDefaultGovernorState().saveReputationHistory(repoFullName, history, apiBaseUrl); } +export function incrementReputationHistory( + repoFullName: string, + delta: { decided?: number; unfavorable?: number }, + apiBaseUrl?: string, +): RepoOutcomeHistory { + return getDefaultGovernorState().incrementReputationHistory(repoFullName, delta, apiBaseUrl); +} + export function recordOwnSubmission(record: OwnSubmissionRecord): OwnSubmissionRecord { return getDefaultGovernorState().recordOwnSubmission(record); } diff --git a/packages/loopover-miner/lib/loop-cli.ts b/packages/loopover-miner/lib/loop-cli.ts index 13080976b5..a388ceba9f 100644 --- a/packages/loopover-miner/lib/loop-cli.ts +++ b/packages/loopover-miner/lib/loop-cli.ts @@ -583,13 +583,9 @@ export async function runLoop(args: string[], options: RunLoopOptions = {}): Pro // `unfavorable` only on a closed-without-merge (rejection-state-machine.js's isRejectedPr, matching // #5655's own-rejection classification). Forge-scoped by claimed.apiBaseUrl (#5563), like every other // governor-state write here. - const priorReputation = governorState.loadReputationHistory(claimed.repoFullName, claimed.apiBaseUrl); - governorState.saveReputationHistory( + governorState.incrementReputationHistory( claimed.repoFullName, - { - decided: priorReputation.decided + 1, - unfavorable: priorReputation.unfavorable + (isRejectedPr(prDisposition) ? 1 : 0), - }, + { decided: 1, unfavorable: isRejectedPr(prDisposition) ? 1 : 0 }, claimed.apiBaseUrl, ); reentryOutcome = classifyPrDisposition(prDisposition) as "merged" | "disengaged" | "other"; diff --git a/test/fixtures/miner-concurrent-stores/increment-reputation-child.mjs b/test/fixtures/miner-concurrent-stores/increment-reputation-child.mjs new file mode 100644 index 0000000000..07b75586c7 --- /dev/null +++ b/test/fixtures/miner-concurrent-stores/increment-reputation-child.mjs @@ -0,0 +1,39 @@ +#!/usr/bin/env node +// Cross-process helper for governor reputation-history concurrent-race tests (#8855). +// Opens the shared governor-state db, waits for a stdin "go" signal, then calls incrementReputationHistory() +// so multiple Node processes contend on the same governor_reputation_history row via the same dbPath. +import { openGovernorState } from "../../../packages/loopover-miner/dist/lib/governor-state.js"; + +const [dbPath, repoFullName, decidedDeltaStr, unfavorableDeltaStr] = process.argv.slice(2); +if (!dbPath || !repoFullName || !decidedDeltaStr || !unfavorableDeltaStr) { + process.stderr.write( + "usage: increment-reputation-child.mjs \n", + ); + process.exit(2); +} + +const state = openGovernorState(dbPath); +let started = false; + +function runIncrement() { + if (started) return; + started = true; + try { + const history = state.incrementReputationHistory(repoFullName, { + decided: Number(decidedDeltaStr), + unfavorable: Number(unfavorableDeltaStr), + }); + process.stdout.write(`${JSON.stringify({ ok: true, history })}\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 { + state.close(); + } +} + +process.stdin.setEncoding("utf8"); +process.stdin.on("data", () => runIncrement()); +process.stdout.write("READY\n"); diff --git a/test/unit/miner-concurrent-store-races.test.ts b/test/unit/miner-concurrent-store-races.test.ts index 643bb25c6e..b3322d3240 100644 --- a/test/unit/miner-concurrent-store-races.test.ts +++ b/test/unit/miner-concurrent-store-races.test.ts @@ -5,10 +5,11 @@ import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import { afterEach, describe, expect, it } from "vitest"; import { openClaimLedger } from "../../packages/loopover-miner/lib/claim-ledger.js"; +import { openGovernorState } from "../../packages/loopover-miner/lib/governor-state.js"; import { initPortfolioQueueStore } from "../../packages/loopover-miner/lib/portfolio-queue.js"; -// Real cross-process concurrency coverage for the claim-ledger and portfolio-queue stores (#4867). Only the -// worktree-allocator had a dedicated multi-process collision test before this; claim-ledger/portfolio-queue +// Real cross-process concurrency coverage for the claim-ledger, portfolio-queue, and governor reputation-history +// stores (#4867, #8855). Only the worktree-allocator had a dedicated multi-process collision test before this; // atomicity was previously only exercised per-function (single process). This spawns two real Node child // processes racing the same on-disk SQLite file and asserts no double-claim/double-dequeue or corrupted state // results — the store's own atomic UPSERT/UPDATE...RETURNING statements are what's under test, not the @@ -22,6 +23,10 @@ const dequeueChildScript = join( dirname(fileURLToPath(import.meta.url)), "../fixtures/miner-concurrent-stores/dequeue-child.mjs", ); +const incrementReputationChildScript = join( + dirname(fileURLToPath(import.meta.url)), + "../fixtures/miner-concurrent-stores/increment-reputation-child.mjs", +); const roots: string[] = []; @@ -212,3 +217,49 @@ describe("portfolio-queue cross-process races (#4867)", () => { expect(exitCode).toBe(2); }); }); + +type IncrementReputationChildResult = { + ok: boolean; + history?: { decided: number; unfavorable: number }; + message?: string; +}; + +describe("governor reputation-history cross-process races (#8855)", () => { + it("two processes racing incrementReputationHistory() on the same repo both apply their deltas", async () => { + const { dbPath } = tempRoot(); + const bootstrap = openGovernorState(dbPath); + bootstrap.saveReputationHistory("acme/widgets", { decided: 0, unfavorable: 0 }); + bootstrap.close(); + + const children = [ + spawnChild(incrementReputationChildScript, [dbPath, "acme/widgets", "1", "0"]), + spawnChild(incrementReputationChildScript, [dbPath, "acme/widgets", "1", "1"]), + ]; + const results = await runBarriered(children); + + expect(results.every((result) => result.ok)).toBe(true); + const histories = results.map((result) => result.history); + expect(histories).toHaveLength(2); + expect(histories).toContainEqual({ decided: 2, unfavorable: 1 }); + expect( + histories.some((history) => history?.decided === 1 && history.unfavorable === 0) || + histories.some((history) => history?.decided === 1 && history.unfavorable === 1), + ).toBe(true); + + const state = openGovernorState(dbPath); + try { + expect(state.loadReputationHistory("acme/widgets")).toEqual({ decided: 2, unfavorable: 1 }); + } finally { + state.close(); + } + }); + + it("rejects the increment-reputation-child helper when required args are missing", async () => { + const child = spawn(process.execPath, [incrementReputationChildScript], { stdio: ["ignore", "pipe", "pipe"] }); + const exitCode = await new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", resolve); + }); + expect(exitCode).toBe(2); + }); +}); diff --git a/test/unit/miner-governor-state.test.ts b/test/unit/miner-governor-state.test.ts index 76a907ba76..0096bae121 100644 --- a/test/unit/miner-governor-state.test.ts +++ b/test/unit/miner-governor-state.test.ts @@ -5,6 +5,7 @@ import { DatabaseSync } from "node:sqlite"; import { afterEach, describe, expect, it, vi } from "vitest"; import { closeDefaultGovernorState, + incrementReputationHistory, loadPauseState, loadReputationHistory, openGovernorState, @@ -386,6 +387,85 @@ describe("governor-state reputation history (#5134)", () => { }); }); +describe("governor-state reputation-history increment atomicity (#8855)", () => { + it("REGRESSION: overlapping load/save increments from two sibling connections lose a count without a transaction", () => { + const root = mkdtempSync(join(tmpdir(), "loopover-miner-governor-state-reputation-race-")); + roots.push(root); + const dbPath = join(root, "shared-governor-state.sqlite3"); + const processA = openGovernorState(dbPath); + const processB = openGovernorState(dbPath); + states.push(processA, processB); + + processA.saveReputationHistory("acme/widgets", { decided: 0, unfavorable: 0 }); + + // Classic read-modify-write overlap: both loads see decided=0 before either save commits. + const priorA = processA.loadReputationHistory("acme/widgets"); + const priorB = processB.loadReputationHistory("acme/widgets"); + processA.saveReputationHistory("acme/widgets", { decided: priorA.decided + 1, unfavorable: 0 }); + processB.saveReputationHistory("acme/widgets", { decided: priorB.decided + 1, unfavorable: 0 }); + expect(processA.loadReputationHistory("acme/widgets")).toEqual({ decided: 1, unfavorable: 0 }); + }); + + it("REGRESSION: incrementReputationHistory serializes sibling overlapping increments so none are lost", () => { + const root = mkdtempSync(join(tmpdir(), "loopover-miner-governor-state-reputation-increment-")); + roots.push(root); + const dbPath = join(root, "shared-governor-state.sqlite3"); + const processA = openGovernorState(dbPath); + const processB = openGovernorState(dbPath); + states.push(processA, processB); + + processA.saveReputationHistory("acme/widgets", { decided: 0, unfavorable: 0 }); + processA.incrementReputationHistory("acme/widgets", { decided: 1, unfavorable: 0 }); + processB.incrementReputationHistory("acme/widgets", { decided: 1, unfavorable: 1 }); + expect(processA.loadReputationHistory("acme/widgets")).toEqual({ decided: 2, unfavorable: 1 }); + expect(processB.loadReputationHistory("acme/widgets")).toEqual({ decided: 2, unfavorable: 1 }); + }); + + it("increments from defaults when no prior row exists", () => { + const state = tempState(); + expect(state.incrementReputationHistory("acme/widgets", { decided: 1, unfavorable: 1 })).toEqual({ + decided: 1, + unfavorable: 1, + }); + }); + + it("defaults non-integer delta fields to zero", () => { + const state = tempState(); + state.saveReputationHistory("acme/widgets", { decided: 3, unfavorable: 1 }); + expect(state.incrementReputationHistory("acme/widgets", { decided: 1 })).toEqual({ decided: 4, unfavorable: 1 }); + expect(state.incrementReputationHistory("acme/widgets", {})).toEqual({ decided: 4, unfavorable: 1 }); + }); + + it("rolls the transaction back if the upsert throws, leaving the handle usable for a later increment", () => { + const state = tempState(); + state.saveReputationHistory("acme/widgets", { decided: 2, unfavorable: 1 }); + + const raw = new DatabaseSync(state.dbPath); + raw.exec("DROP TABLE governor_reputation_history"); + raw.close(); + + expect(() => state.incrementReputationHistory("acme/widgets", { decided: 1, unfavorable: 0 })).toThrow( + /no such table/i, + ); + + // afterEach closes the handle cleanly, proving the failed increment left no dangling write transaction. + }); + + it("incrementReputationHistory module-level wrapper round-trips through the default singleton", () => { + const root = mkdtempSync(join(tmpdir(), "loopover-miner-governor-state-singleton-increment-")); + roots.push(root); + vi.stubEnv("LOOPOVER_MINER_GOVERNOR_STATE_DB", join(root, "governor-state.sqlite3")); + saveReputationHistory("acme/widgets", { decided: 1, unfavorable: 0 }, "https://ghe.example.com/api/v3"); + const written = incrementReputationHistory( + "acme/widgets", + { decided: 2, unfavorable: 1 }, + "https://ghe.example.com/api/v3", + ); + expect(written).toEqual({ decided: 3, unfavorable: 1 }); + expect(loadReputationHistory("acme/widgets", "https://ghe.example.com/api/v3")).toEqual(written); + }); +}); + describe("governor-state own-submission history (#5134)", () => { it("records and lists submissions newest-first", () => { const state = tempState();