From 36e15586f6ce0481fd31821226ede9625d33d9c6 Mon Sep 17 00:00:00 2001 From: e11734937-beep Date: Fri, 17 Jul 2026 14:29:56 +0200 Subject: [PATCH] fix(miner): enforce maxConcurrentClaims atomically across sibling processes The cap was check-then-act: attempt-cli.js read the active-claim count via listActiveClaims, then recorded the claim in a separate claimLedger.claimIssue call with no shared transaction. Two sibling miner processes racing the same repo could both pass the sub-cap count check before either committed, both claim, and exceed maxConcurrentClaims. Add claimIssueWithinCap to the claim ledger: it sweeps orphaned claims, counts the repo active claims, and records the new claim only while under the cap, all inside one BEGIN IMMEDIATE transaction. With node:sqlite shared busy_timeout the loser waits for the winner commit, sees the committed count, and is cleanly rejected (claimed: false) so the caller still logs the cap violation. attempt-cli.js now calls this instead of the split check-then-claim. Closes #6758 --- packages/loopover-miner/lib/attempt-cli.js | 34 ++++--- packages/loopover-miner/lib/claim-ledger.d.ts | 16 ++++ packages/loopover-miner/lib/claim-ledger.js | 51 +++++++++++ test/unit/miner-attempt-cli.test.ts | 14 +-- test/unit/miner-claim-ledger.test.ts | 90 +++++++++++++++++++ 5 files changed, 186 insertions(+), 19 deletions(-) diff --git a/packages/loopover-miner/lib/attempt-cli.js b/packages/loopover-miner/lib/attempt-cli.js index 1e8096d9f6..0b8047c2eb 100644 --- a/packages/loopover-miner/lib/attempt-cli.js +++ b/packages/loopover-miner/lib/attempt-cli.js @@ -458,10 +458,23 @@ export async function runAttempt(args, options = {}) { const reputationHistory = readReputationHistory(parsed.repoFullName); const governor = buildAttemptGovernorContext(env, amsPolicy.spec, repoPaused, convergenceInput, reputationHistory); - // Real maxConcurrentClaims enforcement (#6056): the repo's .loopover-miner.yml cap is parsed and - // validated by resolveMinerGoalSpec above, but must be honored here before recording a new soft-claim. - const activeClaims = claimLedger.listActiveClaims(parsed.repoFullName); - if (activeClaims.length >= minerGoalSpec.spec.maxConcurrentClaims) { + // Real maxConcurrentClaims enforcement (#6758): the repo's .loopover-miner.yml cap is honored ATOMICALLY by + // the ledger's count-and-claim, not by a listActiveClaims pre-check here. The old check-then-act split -- read + // the count in this file, then record the claim in a separate claimLedger call -- let two sibling miner + // processes racing the same repo both pass a stale sub-cap count and both claim, exceeding the cap. + // claimIssueWithinCap fuses the count and the insert into one transaction; the loser gets `claimed: false` + // and is reported below rather than silently dropped. This is also the real soft-claim (#5393): once it + // returns claimed, a sibling process sees it via claimLedger.listActiveClaims while this attempt is in + // flight, it is released in `finally` on every terminal outcome (mirroring the worktree allocation slot's + // acquire-then-always-release), and its claimedAt feeds the post-submission conflict check further down (#4848). + const claimResult = claimLedger.claimIssueWithinCap( + parsed.repoFullName, + parsed.issueNumber, + `attempt:${attemptId}`, + undefined, + minerGoalSpec.spec.maxConcurrentClaims, + ); + if (!claimResult.claimed) { const reason = "max_concurrent_claims_exceeded"; attemptLog.appendAttemptLogEvent({ eventType: "attempt_aborted", @@ -473,7 +486,7 @@ export async function runAttempt(args, options = {}) { repoFullName: parsed.repoFullName, issueNumber: parsed.issueNumber, maxConcurrentClaims: minerGoalSpec.spec.maxConcurrentClaims, - activeClaimCount: activeClaims.length, + activeClaimCount: claimResult.activeClaimCount, }, }); eventLedger.appendEvent({ @@ -485,7 +498,7 @@ export async function runAttempt(args, options = {}) { outcome: "blocked_max_concurrent_claims", reason, maxConcurrentClaims: minerGoalSpec.spec.maxConcurrentClaims, - activeClaimCount: activeClaims.length, + activeClaimCount: claimResult.activeClaimCount, repoFullName: parsed.repoFullName, issueNumber: parsed.issueNumber, minerLogin: parsed.minerLogin, @@ -497,19 +510,14 @@ export async function runAttempt(args, options = {}) { console.log(JSON.stringify(blockedResult, null, 2)); } else { console.error( - `Attempt for ${parsed.repoFullName}#${parsed.issueNumber} is blocked: this repo's maxConcurrentClaims cap (${minerGoalSpec.spec.maxConcurrentClaims}) is already met (${activeClaims.length} active claim(s)).`, + `Attempt for ${parsed.repoFullName}#${parsed.issueNumber} is blocked: this repo's maxConcurrentClaims cap (${minerGoalSpec.spec.maxConcurrentClaims}) is already met (${claimResult.activeClaimCount} active claim(s)).`, ); } options.onResult?.(blockedResult); return 11; } - // Real soft-claim (#5393): recorded once we've committed to a real attempt (past feasibility), so a - // sibling miner process on this machine sees it via claimLedger.listActiveClaims while this - // attempt is in flight. Released in `finally` on every terminal outcome -- mirrors the worktree - // allocation slot's own acquire-then-always-release pattern below. The real claimedAt this returns is - // ALSO this miner's own claim-time for the post-submission conflict check further down (#4848). - const claimRecord = claimLedger.claimIssue(parsed.repoFullName, parsed.issueNumber, `attempt:${attemptId}`); + const claimRecord = claimResult.claim; claimedIssue = true; const runAttemptPipeline = options.runMinerAttempt ?? runMinerAttempt; diff --git a/packages/loopover-miner/lib/claim-ledger.d.ts b/packages/loopover-miner/lib/claim-ledger.d.ts index b148def326..97625f7151 100644 --- a/packages/loopover-miner/lib/claim-ledger.d.ts +++ b/packages/loopover-miner/lib/claim-ledger.d.ts @@ -22,11 +22,27 @@ export type ListClaimsFilter = { status?: ClaimStatus | null; }; +/** Result of an atomic, concurrency-capped claim (#6758). `claimed` discriminates success (a recorded claim) + * from a cap rejection (`claim: null`); both carry the pre-insert active count and the resolved cap so a + * rejected caller can still log the violation. */ +export type ClaimWithinCapResult = + | { claimed: true; claim: ClaimEntry; activeClaimCount: number; maxConcurrentClaims: number } + | { claimed: false; claim: null; activeClaimCount: number; maxConcurrentClaims: number }; + export type ClaimLedger = { dbPath: string; recordClaim(claim: RecordClaimInput): ClaimEntry; /** Claims the issue, expiring any claim orphaned by a dead process first (#6156). */ claimIssue(repoFullName: string, issueNumber: number, note?: string, apiBaseUrl?: string): ClaimEntry; + /** Atomically records the claim only while this repo's active-claim count is under `maxConcurrentClaims`, + * counting and inserting in one transaction so racing sibling processes can't exceed the cap (#6758). */ + claimIssueWithinCap( + repoFullName: string, + issueNumber: number, + note: string | undefined, + apiBaseUrl: string | undefined, + maxConcurrentClaims: number, + ): ClaimWithinCapResult; /** Expire claims orphaned by a crashed/killed process, returning the transitioned rows (#6156). */ reclaimExpiredClaims(maxAgeMs?: number): ClaimEntry[]; releaseClaim(repoFullName: string, issueNumber: number, apiBaseUrl?: string): ClaimEntry | null; diff --git a/packages/loopover-miner/lib/claim-ledger.js b/packages/loopover-miner/lib/claim-ledger.js index 2bf2e8fd1c..a7136ded66 100644 --- a/packages/loopover-miner/lib/claim-ledger.js +++ b/packages/loopover-miner/lib/claim-ledger.js @@ -38,6 +38,16 @@ function normalizeIssueNumber(issueNumber) { return issueNumber; } +// The per-repo concurrent-claim cap the atomic count-and-claim gates on (#6758). Always an already-validated +// positive integer from the caller's MinerGoalSpec, but re-checked here because a bad value must fail loudly +// rather than silently disable the cap (a comparison against `undefined` is always false). +function normalizeMaxConcurrentClaims(maxConcurrentClaims) { + if (!Number.isInteger(maxConcurrentClaims) || maxConcurrentClaims < 1) { + throw new Error("invalid_max_concurrent_claims"); + } + return maxConcurrentClaims; +} + /** Optional forge host, scoping rows so two hosts serving the same owner/repo name never collide (#5563). * Omitted/nullish → the github.com default, so every pre-existing single-forge caller is unaffected. */ function normalizeApiBaseUrl(apiBaseUrl) { @@ -168,6 +178,12 @@ export function openClaimLedger(dbPath = resolveClaimLedgerDbPath()) { const listRepoStatusStatement = db.prepare( "SELECT * FROM miner_claims WHERE repo_full_name = ? AND status = ? ORDER BY id ASC", ); + // Repo-wide active-claim tally for the atomic concurrency cap (#6758). Scoped by repo_full_name only (not + // api_base_url), matching the cross-forge counting that listActiveClaims(repoFullName) -- and the prior + // attempt-cli.js pre-check built on it -- already did, so the cap's MEANING is unchanged; only its atomicity is. + const countActiveRepoStatement = db.prepare( + "SELECT COUNT(*) AS count FROM miner_claims WHERE repo_full_name = ? AND status = 'active'", + ); function normalizeListRepoFilter(repoFullName) { if (repoFullName === undefined || repoFullName === null) return undefined; @@ -235,6 +251,41 @@ export function openClaimLedger(dbPath = resolveClaimLedgerDbPath()) { sweepExpiredClaims(ledger, Date.now(), DEFAULT_MAX_CLAIM_AGE_MS); return ledger.recordClaim({ repoFullName, issueNumber, note, apiBaseUrl }); }, + /** + * Atomic, concurrency-capped claim (#6758). Sweeps orphaned claims, counts this repo's ACTIVE claims, and + * records the new claim ONLY while still strictly under `maxConcurrentClaims` -- all inside ONE `BEGIN + * IMMEDIATE` transaction. The prior enforcement split the count (attempt-cli.js's listActiveClaims) from the + * insert (claimIssue) across two statements with no shared transaction, so two sibling miner processes racing + * the same repo could both read the same sub-cap count and both claim, exceeding the cap. Fusing count + + * insert under an IMMEDIATE write lock -- with node:sqlite's shared `busy_timeout`, so the loser WAITS for the + * winner's commit rather than erroring -- closes that window: the second process sees the committed count and + * is cleanly rejected with `claimed: false` (never silently dropped), so the caller can log the cap violation. + * Returns the pre-insert `activeClaimCount` and the resolved `maxConcurrentClaims` on both paths. + */ + claimIssueWithinCap(repoFullName, issueNumber, note, apiBaseUrl, maxConcurrentClaims) { + const cap = normalizeMaxConcurrentClaims(maxConcurrentClaims); + // Normalize the repo up front: the count query keys on it, and a bad value must throw BEFORE `BEGIN` so it + // can never strand an open transaction. `issueNumber`/`note`/`apiBaseUrl` are validated by recordClaim + // INSIDE the transaction -- a bad value there is rolled back whole via the catch below. + const normalizedRepo = normalizeRepoFullName(repoFullName); + db.exec("BEGIN IMMEDIATE"); + try { + sweepExpiredClaims(ledger, Date.now(), DEFAULT_MAX_CLAIM_AGE_MS); + const activeClaimCount = countActiveRepoStatement.get(normalizedRepo).count; + if (activeClaimCount >= cap) { + // COMMIT, not ROLLBACK: a claim the sweep just expired is a legitimate transition that must persist even + // though THIS claim is rejected -- rolling back would resurrect a dead process's stale claim. + db.exec("COMMIT"); + return { claimed: false, claim: null, activeClaimCount, maxConcurrentClaims: cap }; + } + const claim = ledger.recordClaim({ repoFullName, issueNumber, note, apiBaseUrl }); + db.exec("COMMIT"); + return { claimed: true, claim, activeClaimCount, maxConcurrentClaims: cap }; + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } + }, listActiveClaims(repoFullName) { const filter = { status: "active" }; if (repoFullName !== undefined) filter.repoFullName = repoFullName; diff --git a/test/unit/miner-attempt-cli.test.ts b/test/unit/miner-attempt-cli.test.ts index 31ef886378..12628705fe 100644 --- a/test/unit/miner-attempt-cli.test.ts +++ b/test/unit/miner-attempt-cli.test.ts @@ -1720,7 +1720,7 @@ describe("runAttempt: maxConcurrentClaims enforcement (#6056)", () => { const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); const log = vi.spyOn(console, "log").mockImplementation(() => undefined); claimLedger.claimIssue("acme/widgets", 99, "other-attempt"); - const claimIssueSpy = vi.spyOn(claimLedger, "claimIssue"); + const claimWithinCapSpy = vi.spyOn(claimLedger, "claimIssueWithinCap"); const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice", "--json"], { env: { MINER_CODING_AGENT_PROVIDER: "noop" }, @@ -1735,7 +1735,9 @@ describe("runAttempt: maxConcurrentClaims enforcement (#6056)", () => { }); expect(exitCode).toBe(11); - expect(claimIssueSpy).not.toHaveBeenCalled(); + // The cap is now enforced ATOMICALLY inside claimIssueWithinCap (repo, issue, note, apiBaseUrl, cap), which + // returns claimed: false for the loser -- no separate listActiveClaims pre-check (#6758). + expect(claimWithinCapSpy).toHaveBeenCalledWith("acme/widgets", 7, expect.stringMatching(/^attempt:/), undefined, 1); const payload = JSON.parse(String(log.mock.calls.at(-1)?.[0])); expect(payload).toMatchObject({ outcome: "blocked_max_concurrent_claims", @@ -1792,8 +1794,8 @@ describe("runAttempt: maxConcurrentClaims enforcement (#6056)", () => { it("REGRESSION: proceeds when active claims are below the configured cap", async () => { const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); vi.spyOn(console, "log").mockImplementation(() => undefined); - const claimIssueSpy = vi.spyOn(claimLedger, "claimIssue"); claimLedger.claimIssue("acme/widgets", 99, "other-attempt"); + const claimWithinCapSpy = vi.spyOn(claimLedger, "claimIssueWithinCap"); const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice", "--json"], { env: { MINER_CODING_AGENT_PROVIDER: "noop" }, @@ -1813,13 +1815,13 @@ describe("runAttempt: maxConcurrentClaims enforcement (#6056)", () => { }); expect(exitCode).toBe(7); - expect(claimIssueSpy).toHaveBeenCalledWith("acme/widgets", 7, expect.stringMatching(/^attempt:/)); + expect(claimWithinCapSpy).toHaveBeenCalledWith("acme/widgets", 7, expect.stringMatching(/^attempt:/), undefined, 2); }); it("REGRESSION: proceeds with the default cap when there are zero prior active claims", async () => { const { allocator, claimLedger, eventLedger, attemptLog, governorLedger } = tempLedgers(); vi.spyOn(console, "log").mockImplementation(() => undefined); - const claimIssueSpy = vi.spyOn(claimLedger, "claimIssue"); + const claimWithinCapSpy = vi.spyOn(claimLedger, "claimIssueWithinCap"); const exitCode = await runAttempt(["acme/widgets", "7", "--miner-login", "alice", "--json"], { env: { MINER_CODING_AGENT_PROVIDER: "noop" }, @@ -1834,6 +1836,6 @@ describe("runAttempt: maxConcurrentClaims enforcement (#6056)", () => { }); expect(exitCode).toBe(7); - expect(claimIssueSpy).toHaveBeenCalledWith("acme/widgets", 7, expect.stringMatching(/^attempt:/)); + expect(claimWithinCapSpy).toHaveBeenCalledWith("acme/widgets", 7, expect.stringMatching(/^attempt:/), undefined, 1); }); }); diff --git a/test/unit/miner-claim-ledger.test.ts b/test/unit/miner-claim-ledger.test.ts index 1d019a7b22..3640fbb481 100644 --- a/test/unit/miner-claim-ledger.test.ts +++ b/test/unit/miner-claim-ledger.test.ts @@ -540,6 +540,96 @@ describe("loopover-miner claim ledger (#2314)", () => { }); }); +describe("claimIssueWithinCap: atomic per-repo concurrency cap (#6758)", () => { + it("records the claim when the repo is under the cap, reporting the pre-insert count", () => { + const ledger = tempLedger(); + const result = ledger.claimIssueWithinCap("acme/widgets", 7, "attempt:x", undefined, 2); + expect(result).toMatchObject({ claimed: true, activeClaimCount: 0, maxConcurrentClaims: 2 }); + expect(result.claim).toMatchObject({ + repoFullName: "acme/widgets", + issueNumber: 7, + status: "active", + note: "attempt:x", + }); + expect(ledger.listActiveClaims("acme/widgets")).toHaveLength(1); + }); + + it("rejects a new claim once the repo is at the cap, without recording it", () => { + const ledger = tempLedger(); + ledger.claimIssueWithinCap("acme/widgets", 1, "first", undefined, 1); + const result = ledger.claimIssueWithinCap("acme/widgets", 2, "second", undefined, 1); + expect(result).toEqual({ claimed: false, claim: null, activeClaimCount: 1, maxConcurrentClaims: 1 }); + // The rejected issue was never written; only the winner's claim is active for the repo. + expect(ledger.listActiveClaims("acme/widgets").map((c) => c.issueNumber)).toEqual([1]); + expect(ledger.listClaims({ repoFullName: "acme/widgets" })).toHaveLength(1); + }); + + it("counts the cap PER REPO, so a different repo's active claims never block", () => { + const ledger = tempLedger(); + ledger.claimIssueWithinCap("acme/other", 1, "other", undefined, 1); + expect(ledger.claimIssueWithinCap("acme/widgets", 2, "widgets", undefined, 1).claimed).toBe(true); + }); + + it("REGRESSION: two sibling connections to the same DB racing the cap -- only one wins (#6758)", () => { + // Two DatabaseSync connections to ONE file are exactly the two sibling miner PROCESSES the issue describes: + // SQLite's file locking treats them identically. Before the fix the count and the insert were split across + // attempt-cli.js and claimLedger, so both could pass a stale count. Now each claimIssueWithinCap fuses count + // + insert under BEGIN IMMEDIATE, so the second connection sees the first's committed claim and is rejected. + const root = tempRoot(); + const dbPath = join(root, "shared-claim-ledger.sqlite3"); + const processA = openClaimLedger(dbPath); + const processB = openClaimLedger(dbPath); + ledgers.push(processA, processB); + + const resultA = processA.claimIssueWithinCap("acme/widgets", 1, "A", undefined, 1); + const resultB = processB.claimIssueWithinCap("acme/widgets", 2, "B", undefined, 1); + + expect(resultA.claimed).toBe(true); + expect(resultB).toMatchObject({ claimed: false, claim: null, activeClaimCount: 1, maxConcurrentClaims: 1 }); + // The cap holds ACROSS the two connections: exactly one active claim exists for the repo. + expect(processB.listActiveClaims("acme/widgets").map((c) => c.issueNumber)).toEqual([1]); + }); + + it("sweeps an orphaned claim inside the transaction to free a slot, then claims within the restored cap", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-01T00:00:00Z")); + const ledger = tempLedger(); + ledger.claimIssueWithinCap("acme/widgets", 1, "stale", undefined, 1); + // While the first claim is fresh, a second at cap=1 is rejected. + expect(ledger.claimIssueWithinCap("acme/widgets", 2, "blocked", undefined, 1).claimed).toBe(false); + // Advance well past the 14-day expiry window: the first claim is now orphaned. + vi.setSystemTime(new Date("2026-08-01T00:00:00Z")); + const result = ledger.claimIssueWithinCap("acme/widgets", 2, "after-sweep", undefined, 1); + expect(result.claimed).toBe(true); + // The stale claim was swept to 'expired' by the same transaction; only the new claim is active. + expect(ledger.listActiveClaims("acme/widgets").map((c) => c.issueNumber)).toEqual([2]); + expect( + ledger.listClaims({ repoFullName: "acme/widgets", status: "expired" }).map((c) => c.issueNumber), + ).toEqual([1]); + }); + + it("rejects a non-integer or below-1 maxConcurrentClaims before touching the DB", () => { + const ledger = tempLedger(); + expect(() => ledger.claimIssueWithinCap("acme/widgets", 1, undefined, undefined, 1.5)).toThrow( + "invalid_max_concurrent_claims", + ); + expect(() => ledger.claimIssueWithinCap("acme/widgets", 1, undefined, undefined, 0)).toThrow( + "invalid_max_concurrent_claims", + ); + expect(ledger.listClaims()).toEqual([]); + }); + + it("rolls the transaction back if recording throws (invalid issue), leaving the ledger clean and usable", () => { + const ledger = tempLedger(); + expect(() => ledger.claimIssueWithinCap("acme/widgets", 0, "bad", undefined, 1)).toThrow( + "invalid_issue_number", + ); + // BEGIN IMMEDIATE was rolled back: no partial write, and a subsequent claim still succeeds (no stranded txn). + expect(ledger.listClaims()).toEqual([]); + expect(ledger.claimIssueWithinCap("acme/widgets", 5, "ok", undefined, 1).claimed).toBe(true); + }); +}); + function tempRoot() { const root = mkdtempSync(join(tmpdir(), "loopover-miner-claim-default-")); roots.push(root);