diff --git a/src/env.d.ts b/src/env.d.ts index 7053788fe7..9169af8c3b 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -28,6 +28,12 @@ declare global { get(key: string): Promise; set(key: string, value: string, ttlSeconds: number): Promise; del?(key: string): Promise; + /** Atomic "set only if absent": returns true when this call newly claimed the key, false when it was + * already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent + * callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the + * check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still + * type-checks; callers fall back to the non-atomic get/set pair when absent (#2129). */ + claim?(key: string, value: string, ttlSeconds: number): Promise; }; /** TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) is a separate, * more-involved sub-task — it needs the ported DO class + its own migration tag, not just a binding here. diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 83beac8d8e..09c5b961a8 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1694,6 +1694,48 @@ async function maybeRunAgentMaintenance( if (pr.isDraft) return; if (!gate) return; + // Per-PR mutual exclusion (#2129): a webhook re-review and a sweep-driven agent-regate-pr job use different + // coalesce-key shapes (jobCoalesceKey never matches one against the other) and QUEUE_CONCURRENCY explicitly + // overlaps I/O-bound jobs, so two passes for the SAME PR can both reach this point concurrently, each with its + // own independently-timed live CI/mergeable/reviewDecision read. If those reads disagree, both could plan and + // execute DIFFERENT actions for the same PR. Claim a short-TTL advisory lock before the plan-and-execute + // critical section (extracted below so the try/finally doesn't force-reindent that whole block); a pass that + // loses the race defers cleanly — the next webhook/sweep tick is the backstop. Lightweight stand-in for the + // per-PR SubmissionLock Durable Object noted as a longer-term TODO in env.d.ts. + if (!(await claimAgentMaintenanceLock(env, repoFullName, pr.number))) return; + try { + await runAgentMaintenancePlanAndExecute(env, { + installationId, + repoFullName, + repo: args.repo, + pr, + settings, + otherOpenPullRequests, + gate, + liveFacts: args.liveFacts, + }); + } finally { + await releaseAgentMaintenanceLock(env, repoFullName, pr.number); + } +} + +/** The plan-and-execute critical section of {@link maybeRunAgentMaintenance}, extracted so the caller's + * per-PR lock (#2129) wraps it in a try/finally without reindenting this whole block. */ +async function runAgentMaintenancePlanAndExecute( + env: Env, + args: { + installationId: number; + repoFullName: string; + repo: Awaited>; + pr: PullRequestRecord; + settings: RepositorySettings; + otherOpenPullRequests: PullRequestRecord[]; + gate: ReturnType; + liveFacts: LiveGithubFacts; + }, +): Promise { + const { installationId, repoFullName, pr, settings, otherOpenPullRequests, gate } = args; + // Convergence safety: feed the planner the PR's changed paths + the repo's hard-guardrail globs so guarded // paths force manual review, and flag owner-authored PRs so they are never auto-closed (standing rule). // FIX B: resolve files via the shared resolver so an EMPTY stored list (the maintenance ran before the @@ -2307,6 +2349,62 @@ async function ciHeadShaResolutionCoalesced( ); } +// Per-PR advisory lock around maybeRunAgentMaintenance's plan-and-execute critical section (#2129). The TTL is a +// crash-safety backstop only — the normal path releases explicitly in a finally block within a few seconds — so +// it is sized well above any realistic pass duration (matches CI_COALESCE_WINDOW_SECONDS, an already-vetted +// value for a comparable-scale operation in this file), not to bound throughput. +const AGENT_MAINTENANCE_LOCK_TTL_SECONDS = 60; + +function agentMaintenanceLockKey(repoFullName: string, prNumber: number): string { + return `agent-maintenance-lock:${repoFullName.toLowerCase()}#${prNumber}`; +} + +/** + * Claim the per-PR advisory lock. Returns false when another pass already holds it (caller must skip this pass + * — the next webhook/sweep tick is the backstop). A missing cache or cache hiccup fails OPEN (returns true — + * the lock is a defense-in-depth serializer, not the primary safety gate, and must never itself block actuation). + */ +export async function claimAgentMaintenanceLock( + env: Env, + repoFullName: string, + prNumber: number, +): Promise { + const key = agentMaintenanceLockKey(repoFullName, prNumber); + // Atomic claim (#2129): a get-then-set pair has a window between the read and the write where two concurrent + // passes for the SAME PR can both observe an absent key and both claim it, defeating the serializer entirely. + // env.SELFHOST_TRANSIENT_CACHE.claim performs the check-and-set as one operation (Redis SET NX server-side), + // closing that window. Falls back to the non-atomic get/set pair only for a cache adapter that hasn't + // implemented claim yet — strictly no worse than this function's prior behavior. + if (env.SELFHOST_TRANSIENT_CACHE?.claim) { + try { + return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS); + } catch { + return true; // fail open — see the doc comment above. + } + } + // getTransientKey/putTransientKey already fail open internally (a missing cache or a thrown read/write error + // both resolve rather than throw), so this never needs its own try/catch — a cache fault surfaces here as + // "no lock held", which correctly falls through to claiming it. + if (await getTransientKey(env, key)) return false; // another pass is already in-flight for this PR + await putTransientKey(env, key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS); + return true; +} + +/** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */ +export async function releaseAgentMaintenanceLock( + env: Env, + repoFullName: string, + prNumber: number, +): Promise { + try { + await env.SELFHOST_TRANSIENT_CACHE?.del?.( + agentMaintenanceLockKey(repoFullName, prNumber), + ); + } catch { + // best-effort; the TTL is the backstop if release fails + } +} + /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`; * `check_run` also nests it under `check_suite.head_sha`). Returns "" when absent. The payload type doesn't model * these events, so we narrow off `Record` the same way the `pull_requests[]` read does. */ diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index d5d6b13d37..b38297ad8d 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -16,6 +16,13 @@ export function createRedisCache(redis: Redis) { async del(key: string): Promise { await redis.del(key); }, + // Redis performs the existence check and the write as a single atomic command server-side (SET ... NX), so + // two concurrent callers racing on the same key can never both receive "OK" -- unlike a get-then-set pair, + // which has a window between the read and the write where both callers can observe an absent key. + async claim(key: string, value: string, ttlSeconds: number): Promise { + const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); + return result === "OK"; + }, }; } diff --git a/test/helpers/d1.ts b/test/helpers/d1.ts index 324244c8c4..984840531b 100644 --- a/test/helpers/d1.ts +++ b/test/helpers/d1.ts @@ -89,6 +89,15 @@ export function createTestEnv(overrides: Partial = {}): Env { async del(key: string) { transientCache.delete(key); }, + // Mirrors createRedisCache's atomic claim (#2129): the check-and-set below has no `await` between the + // `has` read and the `set` write, so it completes synchronously within one microtask — a concurrent + // caller can never observe the key as absent partway through another caller's claim, matching Redis's + // SET NX server-side atomicity. + async claim(key: string, value: string) { + if (transientCache.has(key)) return false; + transientCache.set(key, value); + return true; + }, }, // Per-repo review allowlist: default to the test repos so flag-ON wiring tests activate the // gated review features. Override to "" to assert the dormant (no-repo) default. diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index e4afc1194a..0dc379929c 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -44,7 +44,7 @@ import { upsertRepositoryFromGitHub, putCachedAiReview, } from "../../src/db/repositories"; -import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, contributorEvidenceBatchSize, processJob } from "../../src/queue/processors"; +import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock } from "../../src/queue/processors"; import { aiReviewCacheInputFingerprint } from "../../src/review/ai-review-cache-input"; import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; import { normalizeRegistryPayload } from "../../src/registry/normalize"; @@ -2922,6 +2922,132 @@ describe("queue processors", () => { expect(denied?.n).toBe(1); }); + it("claimAgentMaintenanceLock claims when free, denies when held (per-PR, not per-repo), and release frees it again (#2129)", async () => { + const env = createTestEnv({}); + // First claim for this PR succeeds — no prior pass in-flight. + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + // A second, concurrent pass for the SAME PR (regardless of what triggered it — webhook or sweep) is denied + // while the first is still in-flight — exactly the race #2129 describes, since job-coalesce keys never + // match across trigger shapes but this lock is keyed purely on repo+PR. + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false); + // A DIFFERENT PR in the same repo is unaffected — the lock is per-PR, not a repo-wide serializer. + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 8)).toBe(true); + // Release (the finally block's job) frees the PR — a subsequent pass can claim it again. + await releaseAgentMaintenanceLock(env, "owner/agent-repo", 7); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + }); + + it("claimAgentMaintenanceLock fails OPEN on a broken transient cache — never itself blocks actuation (#2129)", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => { throw new Error("cache read error"); }, + set: async () => { throw new Error("cache write error"); }, + del: async () => { throw new Error("cache delete error"); }, + }, + }); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + await expect(releaseAgentMaintenanceLock(env, "owner/agent-repo", 7)).resolves.toBeUndefined(); + }); + + it("claimAgentMaintenanceLock fails OPEN when the atomic claim primitive itself throws (#2368)", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async () => { throw new Error("redis unavailable"); }, + }, + }); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + }); + + it("REGRESSION (#2368): claimAgentMaintenanceLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => { + // #2368: a get-then-set pair has a window between the read and the write where two concurrent callers can + // both observe an absent key and both claim it — exactly what the per-PR lock exists to prevent. This test + // races two claims for the same PR via Promise.all (both kick off before either resolves) against the + // default test cache's claim(), which mirrors createRedisCache's atomic SET NX: the check-and-set happens + // with no `await` boundary in between, so it is impossible for both callers to see "unclaimed". + const env = createTestEnv({}); + const [first, second] = await Promise.all([ + claimAgentMaintenanceLock(env, "owner/agent-repo", 7), + claimAgentMaintenanceLock(env, "owner/agent-repo", 7), + ]); + expect([first, second].filter(Boolean)).toHaveLength(1); + }); + + it("REGRESSION (#2368): claimAgentMaintenanceLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => { + const calls: string[] = []; + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => { calls.push("get"); return null; }, + set: async () => { calls.push("set"); }, + claim: async () => { calls.push("claim"); return true; }, + }, + }); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available + }); + + it("claimAgentMaintenanceLock falls back to the get/set pair and still denies a held key when the cache has no claim() (#2368)", async () => { + // A cache adapter that hasn't implemented the atomic claim() primitive yet must still deny a second claim — + // just via the older, non-atomic get-then-set pair (documented as strictly no worse than this function's + // prior behavior), not by skipping the check entirely. + const values = new Map(); + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async (key: string) => values.get(key) ?? null, + set: async (key: string, value: string) => { values.set(key, value); }, + }, + }); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); + expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false); + }); + + it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => { + const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + let mergeCalls = 0; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (url.includes("/pulls/7/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]); + if (url.includes("/pulls/7/merge")) { + mergeCalls += 1; + return new Response(null, { status: 204 }); + } + if (url.includes("/pulls/7/reviews") && init?.method === "POST") return Response.json({ id: 1 }); + if (url.includes("/pulls/7/reviews")) return Response.json([]); + // Only the bare PR resource (no sub-path) — the more specific checks above already claimed + // /pulls/7/files, /pulls/7/merge, and /pulls/7/reviews. + if (/\/pulls\/7(\?|$)/.test(url)) return Response.json({ number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", mergeable_state: "clean" }); + if (url.includes("/commits/a7/check-runs")) return Response.json({ total_count: 0, check_runs: [] }); + if (url.includes("/commits/a7/status")) return Response.json({ state: "success", statuses: [] }); + if (url.includes("/commits/a7/check-suites")) return Response.json({ check_suites: [] }); + if (url.includes("/issues/1")) return Response.json({ number: 1, title: "Issue", state: "open", labels: [], user: { login: "reporter" } }); + if (url.includes("/branches/")) return Response.json({ protected: false, protection: { required_status_checks: { contexts: [] } } }); + if (url.includes(".gittensory.yml")) return new Response("Not Found", { status: 404 }); + if (url.endsWith("/check-runs") && init?.method === "POST") return Response.json({ id: 1 }); + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.endsWith("/graphql")) return Response.json({ data: {} }); + return Response.json({}); + }); + vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); + + // Simulate a webhook pass already in-flight for this exact PR — a github-webhook:pr-refresh job's coalesce + // key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the new + // per-PR advisory lock is what makes a second, independently-triggered pass defer instead of racing it. + await env.SELFHOST_TRANSIENT_CACHE?.set("agent-maintenance-lock:owner/agent-repo#7", "1", 60); + + await processJob(env, { type: "agent-regate-pr", deliveryId: "race-sweep", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 }); + + // The held lock made this pass skip its plan-and-execute critical section entirely — no mutation attempted. + expect(mergeCalls).toBe(0); + const actionAudits = await env.DB.prepare("select count(*) as n from audit_events where event_type like 'agent.action.%'").first<{ n: number }>(); + expect(actionAudits?.n).toBe(0); + }); + it("the sweep stamps the marker INLINE when the repo has no installation (audit-only, still converges) (#audit-sweep-fanout)", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index d350afb5cf..b4bf3d5afd 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -2,7 +2,9 @@ import type { Redis } from "ioredis"; import { describe, expect, it } from "vitest"; import { checkAndMarkDelivery, createRedisCache } from "../../src/selfhost/redis-cache"; -/** Minimal in-memory stand-in for the ioredis methods the cache uses. */ +/** Minimal in-memory stand-in for the ioredis methods the cache uses. Emulates real Redis SET NX + * semantics (refuse + return null when NX is requested and the key already exists) so a test + * using this fake actually exercises the atomicity claim() depends on, not just a plain overwrite. */ function fakeRedis(): Redis & { _store: Map } { const _store = new Map(); return { @@ -10,7 +12,8 @@ function fakeRedis(): Redis & { _store: Map } { async get(k: string) { return _store.get(k) ?? null; }, - async set(k: string, v: string, _ex: "EX", _ttl: number) { + async set(k: string, v: string, _ex: "EX", _ttl: number, nx?: "NX") { + if (nx === "NX" && _store.has(k)) return null; _store.set(k, v); return "OK"; }, @@ -40,6 +43,26 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { await cache.del("k"); expect(await cache.get("k")).toBeNull(); }); + + it("claim atomically sets an absent key and returns true (#2129)", async () => { + const cache = createRedisCache(fakeRedis()); + expect(await cache.claim("lock", "1", 60)).toBe(true); + expect(await cache.get("lock")).toBe("1"); + }); + + it("claim refuses and returns false when the key is already held, without overwriting it (#2129)", async () => { + const r = fakeRedis(); + const cache = createRedisCache(r); + await cache.set("lock", "holder-A", 60); + expect(await cache.claim("lock", "holder-B", 60)).toBe(false); + expect(await cache.get("lock")).toBe("holder-A"); // the second claimant never overwrote the first + }); + + it("claim propagates a Redis error to the caller (claimAgentMaintenanceLock is responsible for failing open)", async () => { + const brokenRedis = { async set() { throw new Error("connection refused"); } } as unknown as Redis; + const cache = createRedisCache(brokenRedis); + await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused"); + }); }); describe("checkAndMarkDelivery (#1216 webhook idempotency)", () => {