diff --git a/src/queue/ai-review-orchestration.ts b/src/queue/ai-review-orchestration.ts index 0adcb1d3a6..1b8f5adbd7 100644 --- a/src/queue/ai-review-orchestration.ts +++ b/src/queue/ai-review-orchestration.ts @@ -21,6 +21,7 @@ import { } from "./transient-locks"; import { buildPullRequestAdvisory } from "../rules/advisory"; import { recordAuditEvent, getDecryptedRepositoryAiKey, getRepository, listCheckSummaries, listPullRequestFiles } from "../db/repositories"; +import { registerHeldLock, unregisterHeldLock } from "./held-lock-registry"; import { recordRoutingShadow } from "../services/reviewer-routing"; import { createInstallationToken } from "../github/app"; import type { AgentActionMode } from "../settings/agent-execution"; @@ -120,12 +121,15 @@ export async function claimAiReviewLock( mode: string, options?: { steal?: boolean }, ): Promise { - return claimTransientLock( - env, - aiReviewLockKey(repoFullName, prNumber, headSha, mode), - AI_REVIEW_LOCK_TTL_SECONDS, - options, - ); + const key = aiReviewLockKey(repoFullName, prNumber, headSha, mode); + const claim = await claimTransientLock(env, key, AI_REVIEW_LOCK_TTL_SECONDS, options); + // #8998: register a REAL claim (a non-null ownerToken means this call, not a fail-open passthrough or a + // no-op adapter, actually owns the key) so a shutdown signal can release it immediately instead of only via + // graceful drain or the full 1800s TTL. A fail-open claim (ownerToken null) has nothing to release. + if (claim.acquired && claim.ownerToken !== null) { + registerHeldLock(key, () => releaseTransientLockIfOwner(env, key, claim.ownerToken)); + } + return claim; } /** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */ @@ -137,7 +141,12 @@ export async function releaseAiReviewLock( mode: string, ownerToken: string | null, ): Promise { - await releaseTransientLockIfOwner(env, aiReviewLockKey(repoFullName, prNumber, headSha, mode), ownerToken); + const key = aiReviewLockKey(repoFullName, prNumber, headSha, mode); + await releaseTransientLockIfOwner(env, key, ownerToken); + // #8998: this process no longer holds it -- a later shutdown must not attempt to release a key it already + // gave up (harmless either way, since the compare-and-delete release is idempotent, but there is no reason + // to carry a stale entry). + unregisterHeldLock(key); } /** diff --git a/src/queue/held-lock-registry.ts b/src/queue/held-lock-registry.ts new file mode 100644 index 0000000000..0a38064acd --- /dev/null +++ b/src/queue/held-lock-registry.ts @@ -0,0 +1,58 @@ +/** + * #8998 — release the transient locks THIS process currently holds the instant a shutdown signal arrives, + * rather than only via graceful drain or TTL expiry. + * + * `queue.stop()` (#9007) already lets an in-flight job finish naturally before the process exits, and a job's + * own `finally` block releases whatever lock it holds as part of that. That is the right behavior when the + * orchestrator's SIGKILL grace period is long enough for the drain to complete. It is not always long enough: + * an AI-review LLM call can legitimately run for tens of seconds to minutes, while a common container-platform + * grace period is 10-30s. When the hard kill lands before the drain finishes, the lock this process claimed — + * `ai-review-lock` most consequentially, at a 1800s TTL — outlives the process that claimed it by up to that + * TTL, starving every subsequent pass (scheduled or an explicit maintainer re-run) with the "already in + * progress" placeholder for real time nobody is actually reviewing anything. + * + * This is the proactive half: a tiny process-local registry of "locks I currently hold and how to release + * them", so the shutdown handler can best-effort release every one of them immediately on SIGTERM/SIGINT — + * before, and independent of, whether the graceful drain itself has time to finish. Complementary to, not a + * replacement for, the boot-time orphaned-lock flush (#9021): that catches whatever this process didn't get a + * chance to release (a truly hard `kill -9`, which delivers no signal at all); this catches everything else. + */ + +type HeldLockRelease = () => Promise; + +const heldLocks = new Map(); + +/** Record that this process now holds `key`, with `release` as how to give it up. Call the moment a claim + * actually succeeds with real ownership (a fail-open "acquired but nothing to release" claim has nothing + * worth registering — see the call site's own guard). */ +export function registerHeldLock(key: string, release: HeldLockRelease): void { + heldLocks.set(key, release); +} + +/** Record that this process no longer holds `key` (its own release already ran). A shutdown racing the same + * release is harmless either way: `releaseTransientLockIfOwner`'s compare-and-delete makes a second release + * attempt against an already-released key a safe no-op. */ +export function unregisterHeldLock(key: string): void { + heldLocks.delete(key); +} + +/** Best-effort release every lock currently on record, clearing the registry as it goes. Never throws: a + * release failure here just means that one lock rides out its own TTL, exactly the pre-#8998 behavior for + * every lock, rather than blocking the rest of shutdown. Returns the count attempted, for one shutdown log + * line — not the count that definitely succeeded, since a released key is removed from the registry either + * way and there is no further signal to distinguish the two once shutdown is already underway. */ +export async function releaseAllHeldLocksAtShutdown(): Promise { + const entries = [...heldLocks.entries()]; + heldLocks.clear(); + let attempted = 0; + for (const [, release] of entries) { + attempted += 1; + await release().catch(() => undefined); + } + return attempted; +} + +/** Test-only introspection — never used by production code. */ +export function heldLockCountForTest(): number { + return heldLocks.size; +} diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index f3558f5254..706325e6cc 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -53,6 +53,7 @@ import { runRetentionPrune } from "./retention"; import { sweepStaleApprovalQueue } from "../services/agent-approval-queue"; import { reconcileMissingPrOutcomes } from "../review/pr-outcome-reconciler"; import { sweepStrandedPendingClosures } from "../review/pending-closure-watchdog"; +import { reconcileSurfaceWithoutDisposition } from "../review/surface-disposition-reconciler"; // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for // mapWithConcurrency, ALSO used by other still-in-processors.ts code), so they stay put and are exported // there purely for this one-directional import-back (processors.ts itself never calls processJob). @@ -275,7 +276,7 @@ export async function processJob(env: Env, message: JobMessage): Promise { } case "agent-regate-sweep": if (!message.repoFullName && message.requestedBy !== "test") { - // Three bounded repair scans ride the sweep's own fan-out tick rather than each earning a job type and + // Four bounded repair scans ride the sweep's own fan-out tick rather than each earning a job type and // a cron entry. All are best-effort and deliberately BEFORE the fan-out: none may cost the tick its // re-gate work, which is the sweep's actual job. // @@ -287,6 +288,11 @@ export async function processJob(env: Env, message: JobMessage): Promise { // // #9031 re-enqueues a pending-closure Pass 2 whose single delayed job was lost, which otherwise strands // the PR permanently: flagged, un-mergeable, un-approvable, and sweep-ineligible. + // + // #8997 re-enqueues a regate for a PR whose published surface has no matching disposition marker for + // that exact head -- the "decisive panel, PR still open" shape a restart killing the pass between + // publish and disposition leaves behind, which the ordinary stale-surface repair check cannot see + // because the surface itself is not stale. const staleness = await sweepStaleApprovalQueue(env).catch(() => null); if (staleness && (staleness.reminded > 0 || staleness.expired > 0)) { console.log(JSON.stringify({ event: "approval_queue_staleness_swept", ...staleness })); @@ -299,6 +305,10 @@ export async function processJob(env: Env, message: JobMessage): Promise { if (stranded && stranded.requeued > 0) { console.log(JSON.stringify({ event: "pending_closure_verifications_requeued", ...stranded })); } + const orphanedDispositions = await reconcileSurfaceWithoutDisposition(env).catch(() => null); + if (orphanedDispositions && orphanedDispositions.requeued > 0) { + console.log(JSON.stringify({ event: "surface_without_disposition_reconciled", ...orphanedDispositions })); + } await fanOutAgentRegateSweepJobs(env, message.requestedBy); return; } diff --git a/src/queue/processors.ts b/src/queue/processors.ts index dc84d64d94..e00c00b030 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -682,6 +682,7 @@ import type { } from "../types"; import { sha256Hex } from "../utils/crypto"; import { errorMessage, nowIso, repoParts } from "../utils/json"; +import { DISPOSITION_CONSIDERED_EVENT_TYPE } from "../review/surface-disposition-reconciler"; import { maybeSuggestMilestoneMatchForPr } from "../integrations/project-tracker-adapter"; const OFFICIAL_MINER_DETECTION_TTL_MS = 5 * 60 * 1000; @@ -1150,6 +1151,24 @@ async function isRegateRepairExhausted(env: Env, repoFullName: string, pr: Pick< return true; } +/** Record that a real disposition attempt began for this exact head (#8997) -- the signal + * reconcileSurfaceWithoutDisposition (surface-disposition-reconciler.ts) reads to tell "a restart killed the + * pass right after publish, before disposition ever started" apart from "disposition genuinely ran (or is + * still running) for this head". Written the moment maintenance CLAIMS its per-PR actuation lock -- i.e. the + * moment a real attempt begins, before anything about WHAT it decides. Best-effort: a failed write here must + * never block the maintenance pass it is only bookkeeping for -- worst case, a missed write costs one extra + * (safe, actuation-lock-deduplicated) repair dispatch on a later sweep tick. */ +async function recordDispositionConsidered(env: Env, repoFullName: string, prNumber: number, headSha: string): Promise { + await recordAuditEvent(env, { + eventType: DISPOSITION_CONSIDERED_EVENT_TYPE, + actor: "loopover", + targetKey: `${repoFullName}#${prNumber}#${headSha}`, + outcome: "completed", + detail: "maintenance actuation lock claimed; disposition attempt begins for this head", + metadata: { repoFullName, prNumber, headSha }, + }).catch(() => undefined); +} + export async function surfaceRepairPriorityPullNumbers( env: Env, repoFullName: string, @@ -2493,6 +2512,11 @@ async function maybeRunAgentMaintenance( }).catch(() => undefined); throw new PrActuationLockContendedError(repoFullName, pr.number, "agent-maintenance"); } + // #8997: the lock is now genuinely held -- a real disposition attempt for this exact head begins here. See + // recordDispositionConsidered's own doc comment for why this specific point is what the boot-time repair + // sweep checks for. Absent headSha (a sparse/never-synced PR row) has nothing to key the marker on and is + // skipped -- the pre-existing lastPublishedSurfaceSha check already covers a PR with no resolvable head. + if (pr.headSha) await recordDispositionConsidered(env, repoFullName, pr.number, pr.headSha); try { await runAgentMaintenancePlanAndExecute(env, { installationId, @@ -10084,9 +10108,26 @@ async function maybePublishPrPublicSurface( // #4889: per-repo admin mode swaps the global-allowlist grant for the live per-repo permission. (await isPerTenantAdmin(env, installationId, repoFullName, author)) || isProtectedAutomationAuthor(author, env)); + // #8999: exempt the freeze when the label's own provenance -- as far as this pass can tell -- is the + // #9009 lock-contention marker and NOTHING else. Without this, the freeze deadlocked: a lost AI-review lock + // race applies this exact label (aiReviewLockContendedResult's hold), the freeze then reads that SAME label + // on the very next pass and refuses a FRESH AI review (only replay), so the pass that could produce the + // real verdict needed to auto-clear the label (#9009, below in runAgentMaintenancePlanAndExecute) never + // runs -- the label outlives the transient contention that caused it, requiring a human to unstick a PR + // that was never substantively held. Reading the SAME marker key #9009 already writes/reads keeps this a + // single source of truth for "why is this label here right now" rather than a second, drifting one. + // + // This does not reopen the gaming surface the freeze exists to close (see #regate-churn doc above): that + // surface is a CONTRIBUTOR iterating pushes to game the bot. A lock-contention hold is not a contributor + // action -- it is an infra artifact of two ORB passes racing -- so exempting it changes nothing about a + // contributor's ability to buy a fresh verdict by pushing. If some OTHER reason also justifies the hold, + // that reason re-applies the label on this very pass's own disposition regardless of this exemption. + const manualReviewLockContentionMarkerKey = `manual-review-lock-contention:${repoFullName.toLowerCase()}#${pr.number}`; + const manualReviewLabelIsPurelyLockContention = manualReviewLabel !== null && (await getTransientKey(env, manualReviewLockContentionMarkerKey)) !== null; const isFrozenForManualReview = webhook.forceAiReview !== true && !authorIsExemptFromFreeze && + !manualReviewLabelIsPurelyLockContention && manualReviewLabel !== null && pr.labels.some((label) => label.toLowerCase() === manualReviewLabel.toLowerCase()); let reviewManifestForAutoReview: FocusManifest | null = null; diff --git a/src/review/surface-disposition-reconciler.ts b/src/review/surface-disposition-reconciler.ts new file mode 100644 index 0000000000..f9fd3c35bb --- /dev/null +++ b/src/review/surface-disposition-reconciler.ts @@ -0,0 +1,101 @@ +import { recordAuditEvent } from "../db/repositories"; +import { errorMessage } from "../utils/json"; + +/** + * #8997 — rescue a PR left wearing a decisive panel with no matching disposition. + * + * A deploy restart (SIGTERM, no drain) can kill a pass at the worst possible cut point: after the public-surface + * publish commits (panel comment, gate check-run, CI aggregate all reflect the current head) but before + * `maybeRunAgentMaintenance` ever claims its per-PR actuation lock for that head. The confirmed live shape + * (#8965, 2026-07-26): a red-CI panel published at 14:55:24Z from the dying container, and the matching close + * only executed at 15:07:53Z — ~12 minutes later, purely because an UNRELATED later sweep tick happened to + * re-run the whole pass from scratch. The regular outage-repair priority check + * (`surfaceRepairPriorityPullNumbers`, processors.ts) cannot see this: it flags a STALE surface + * (`lastPublishedSurfaceSha !== headSha`), and this incident's surface is not stale — it published successfully + * right before the kill. From that check's point of view the PR already looks healthy. + * + * This scan is the missing half: it looks for the disposition side directly. `maybeRunAgentMaintenance` records + * a durable marker (`DISPOSITION_CONSIDERED_EVENT_TYPE`, processors.ts) the moment it claims the actuation lock + * for a head — i.e. the moment a real disposition attempt begins. A PR whose surface is current for its head but + * carries no such marker never got that far, for whatever reason (a restart being the confirmed one; a thrown + * error before the claim is another). Re-enqueuing a regate is cheap and safe either way: `agent-regate-pr` + * re-derives everything from live state, and a PR that in fact already has a real disposition in flight is + * simply denied by the SAME actuation lock this scan is checking for, exactly as any other contending pass would + * be (#9013's per-PR mutex). + * + * Deliberately independent of surfaceRepairPriorityPullNumbers rather than folded into it: that function's + * return value directly drives the regular sweep's staleness-ordered fan-out and is asserted against by a large + * existing test surface (dispatch shape, ordering, backlog-restriction semantics). Riding the SAME sweep tick as + * its own bounded scan — mirroring reconcileMissingPrOutcomes/sweepStrandedPendingClosures (#9026/#9031) — closes + * the identical gap without touching that already-tested surface at all. + */ + +export const DISPOSITION_CONSIDERED_EVENT_TYPE = "agent.maintenance.disposition_considered"; + +/** How far back to look. Bounded so the scan stays cheap on every tick; a surface-without-disposition PR older + * than this has almost certainly been superseded by a later commit or resolved by a human already. */ +export const SURFACE_DISPOSITION_RECONCILE_LOOKBACK_MS = 24 * 60 * 60 * 1000; + +/** Cap per run, mirroring the other bounded repair scans on this same sweep tick. */ +export const SURFACE_DISPOSITION_RECONCILE_LIMIT = 200; + +export type SurfaceDispositionReconcileResult = { scanned: number; requeued: number }; + +/** + * Find open PRs with a current published surface but no disposition marker for that exact head, and re-enqueue + * an `agent-regate-pr` job for each. Best-effort throughout: a repair pass that can itself break the tick it + * rides on is worse than the gap it exists to close. + */ +export async function reconcileSurfaceWithoutDisposition(env: Env, nowMs: number = Date.now()): Promise { + const since = new Date(nowMs - SURFACE_DISPOSITION_RECONCILE_LOOKBACK_MS).toISOString(); + let rows: Array<{ repoFullName: string; number: number; installationId: number; headSha: string }> = []; + try { + const result = await env.DB.prepare( + `SELECT pr.repo_full_name AS repoFullName, pr.number AS number, repo.installation_id AS installationId, pr.head_sha AS headSha + FROM pull_requests AS pr + JOIN repositories AS repo ON repo.full_name = pr.repo_full_name + WHERE pr.state = 'open' + AND pr.head_sha IS NOT NULL + AND pr.last_published_surface_sha = pr.head_sha + AND pr.updated_at >= ?1 + AND repo.installation_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM audit_events AS disposition + WHERE disposition.target_key = pr.repo_full_name || '#' || pr.number || '#' || pr.head_sha + AND disposition.event_type = ?2 + ) + ORDER BY pr.updated_at + LIMIT ?3`, + ) + .bind(since, DISPOSITION_CONSIDERED_EVENT_TYPE, SURFACE_DISPOSITION_RECONCILE_LIMIT) + .all<{ repoFullName: string; number: number; installationId: number; headSha: string }>(); + rows = result.results ?? []; + } catch (error) { + console.warn(JSON.stringify({ level: "warn", event: "surface_disposition_reconcile_scan_failed", message: errorMessage(error).slice(0, 160) })); + return { scanned: 0, requeued: 0 }; + } + + let requeued = 0; + for (const row of rows) { + const sent = await env.JOBS.send({ + type: "agent-regate-pr", + deliveryId: `surface-without-disposition:${row.repoFullName}#${row.number}#${row.headSha}`, + repoFullName: row.repoFullName, + prNumber: row.number, + installationId: row.installationId, + }) + .then(() => true) + .catch(() => false); + if (!sent) continue; + requeued += 1; + await recordAuditEvent(env, { + eventType: "agent.sweep.surface_without_disposition", + actor: "loopover", + targetKey: `${row.repoFullName}#${row.number}`, + outcome: "queued", + detail: "published surface has no matching disposition for this head; re-gating to close the gap", + metadata: { repoFullName: row.repoFullName, pullNumber: row.number, headSha: row.headSha }, + }).catch(() => undefined); + } + return { scanned: rows.length, requeued }; +} diff --git a/src/server.ts b/src/server.ts index 4fc4227141..c2e594c726 100644 --- a/src/server.ts +++ b/src/server.ts @@ -15,6 +15,7 @@ import packageJson from "../package.json"; import worker from "./index"; import { githubRestRateLimitRemainingSamples } from "./github/client"; import { processJob } from "./queue/processors"; +import { releaseAllHeldLocksAtShutdown } from "./queue/held-lock-registry"; import { createOpenAiCompatibleAi, createSelfHostAi, @@ -1323,6 +1324,13 @@ async function main(): Promise { if (shuttingDown) return; shuttingDown = true; console.log(JSON.stringify({ event: "selfhost_shutdown", signal })); + // #8998: release every transient lock THIS process currently holds FIRST, before anything that might take + // real time (the queue drain below, telemetry flush). A container orchestrator's SIGKILL grace period is + // often shorter than an in-flight AI-review call legitimately runs, so waiting for the graceful drain to + // release a lock via its own finally block is not always fast enough -- doing it proactively here means the + // lock is gone the instant SIGTERM/SIGINT arrives, regardless of whether the rest of shutdown gets to finish. + const releasedLocks = await releaseAllHeldLocksAtShutdown(); + if (releasedLocks > 0) console.log(JSON.stringify({ event: "selfhost_shutdown_locks_released", count: releasedLocks })); clearInterval(cron); server.close(); await backend.shutdown(); diff --git a/test/unit/held-lock-registry.test.ts b/test/unit/held-lock-registry.test.ts new file mode 100644 index 0000000000..660441d8e8 --- /dev/null +++ b/test/unit/held-lock-registry.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it, vi } from "vitest"; +import { heldLockCountForTest, registerHeldLock, releaseAllHeldLocksAtShutdown, unregisterHeldLock } from "../../src/queue/held-lock-registry"; +import { claimAiReviewLock, releaseAiReviewLock } from "../../src/queue/ai-review-orchestration"; +import { createTestEnv } from "../helpers/d1"; + +// #8998: queue.stop() (#9007) already lets an in-flight job finish naturally, releasing its own lock via its +// own finally block — but only if the orchestrator's SIGKILL grace period is long enough for that drain to +// finish, and an AI-review LLM call can legitimately outrun a common 10-30s grace period. This registry is the +// proactive half: release every lock this process holds THE INSTANT a shutdown signal arrives, rather than +// only via graceful drain or the lock's own (up to 1800s) TTL. +describe("held-lock-registry (#8998)", () => { + it("releases every registered lock and reports how many it attempted", async () => { + const released: string[] = []; + registerHeldLock("lock:a", async () => void released.push("a")); + registerHeldLock("lock:b", async () => void released.push("b")); + + expect(await releaseAllHeldLocksAtShutdown()).toBe(2); + expect(released.sort()).toEqual(["a", "b"]); + }); + + it("clears the registry as it goes, so a second shutdown call has nothing left to release", async () => { + registerHeldLock("lock:c", async () => undefined); + await releaseAllHeldLocksAtShutdown(); + + expect(await releaseAllHeldLocksAtShutdown()).toBe(0); + }); + + it("unregistering removes a lock before shutdown ever needs to touch it — the normal, non-crash path", async () => { + const release = vi.fn(async () => undefined); + registerHeldLock("lock:d", release); + unregisterHeldLock("lock:d"); + + expect(await releaseAllHeldLocksAtShutdown()).toBe(0); + expect(release).not.toHaveBeenCalled(); + }); + + it("keeps releasing the rest when one release throws — one bad lock must not strand the others", async () => { + const released: string[] = []; + registerHeldLock("lock:e", async () => { + throw new Error("cache unreachable"); + }); + registerHeldLock("lock:f", async () => void released.push("f")); + + expect(await releaseAllHeldLocksAtShutdown()).toBe(2); + expect(released).toEqual(["f"]); + }); + + it("re-registering the same key replaces the prior release rather than accumulating both", async () => { + const first = vi.fn(async () => undefined); + const second = vi.fn(async () => undefined); + registerHeldLock("lock:g", first); + registerHeldLock("lock:g", second); + + expect(heldLockCountForTest()).toBe(1); + await releaseAllHeldLocksAtShutdown(); + expect(first).not.toHaveBeenCalled(); + expect(second).toHaveBeenCalledTimes(1); + }); + + it("does nothing and reports zero when the registry is empty", async () => { + expect(await releaseAllHeldLocksAtShutdown()).toBe(0); + }); +}); + +// The actual integration this registry exists for: claimAiReviewLock/releaseAiReviewLock are the canonical +// claim/release points for the SPECIFIC lock class #8998 is about (a 1800s TTL is the whole reason a proactive +// release matters far more here than for a short-TTL lock). +describe("claimAiReviewLock / releaseAiReviewLock register with the held-lock registry (#8998)", () => { + it("registers a real claim, and the shutdown release actually frees the underlying lock", async () => { + const env = createTestEnv(); + const before = heldLockCountForTest(); + const claim = await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block"); + expect(claim.acquired).toBe(true); + expect(heldLockCountForTest()).toBe(before + 1); + + // A second claim on the SAME key must be denied while this process holds it... + expect((await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).acquired).toBe(false); + + // ...until the shutdown-time release runs, which is the entire point: it frees the lock even though + // nothing ever explicitly called releaseAiReviewLock itself (standing in for "the process died"). + expect(await releaseAllHeldLocksAtShutdown()).toBe(before + 1); + expect((await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).acquired).toBe(true); + }); + + it("unregisters on the normal release path, so shutdown finds nothing left for an already-completed pass", async () => { + const env = createTestEnv(); + const before = heldLockCountForTest(); + const claim = await claimAiReviewLock(env, "acme/widgets", 4, "sha4", "block"); + await releaseAiReviewLock(env, "acme/widgets", 4, "sha4", "block", claim.ownerToken); + + expect(heldLockCountForTest()).toBe(before); + }); + + it("does not register a fail-open claim — there is nothing real to release", async () => { + // #5027: createTestEnv() always carries a working SELFHOST_TRANSIENT_CACHE, so it has to be stripped + // explicitly here to reach claimTransientLock's own fail-open path (no atomic primitive bound at all), + // which returns ownerToken: null (transient-locks.ts) — that must not be registered as a held lock. + const env = createTestEnv(); + delete (env as { SELFHOST_TRANSIENT_CACHE?: unknown }).SELFHOST_TRANSIENT_CACHE; + const before = heldLockCountForTest(); + const claim = await claimAiReviewLock(env, "acme/widgets", 5, "sha5", "block"); + + expect(claim.ownerToken).toBeNull(); + expect(heldLockCountForTest()).toBe(before); + }); +}); diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts index 2c786a7cab..8a6e95e59f 100644 --- a/test/unit/job-dispatch.test.ts +++ b/test/unit/job-dispatch.test.ts @@ -145,45 +145,59 @@ describe("agent-regate-sweep also sweeps the stale approval queue (#9032)", () = }); }); -// #9026 / #9031: two bounded repair scans ride the sweep's own fan-out tick rather than each earning a job type -// and a cron entry. Both must be best-effort — neither may cost the tick its re-gate work, which is the sweep's -// actual job — and both must be quiet when there is nothing to repair. -describe("agent-regate-sweep runs the durability repair scans (#9026, #9031)", () => { +// #9026 / #9031 / #8997: three bounded repair scans ride the sweep's own fan-out tick rather than each earning +// a job type and a cron entry. All must be best-effort — none may cost the tick its re-gate work, which is the +// sweep's actual job — and all must be quiet when there is nothing to repair. +describe("agent-regate-sweep runs the durability repair scans (#9026, #9031, #8997)", () => { afterEach(() => { vi.restoreAllMocks(); }); - async function sweepWith(stubs: { outcomes?: unknown; stranded?: unknown; outcomesThrows?: boolean; strandedThrows?: boolean }): Promise[]> { + async function sweepWith(stubs: { + outcomes?: unknown; + stranded?: unknown; + orphaned?: unknown; + outcomesThrows?: boolean; + strandedThrows?: boolean; + orphanedThrows?: boolean; + }): Promise[]> { const logs: string[] = []; vi.spyOn(console, "log").mockImplementation((...args: unknown[]) => void logs.push(String(args[0]))); const reconciler = await import("../../src/review/pr-outcome-reconciler"); const watchdog = await import("../../src/review/pending-closure-watchdog"); + const surfaceReconciler = await import("../../src/review/surface-disposition-reconciler"); const outcomeSpy = vi.spyOn(reconciler, "reconcileMissingPrOutcomes"); const strandedSpy = vi.spyOn(watchdog, "sweepStrandedPendingClosures"); + const orphanedSpy = vi.spyOn(surfaceReconciler, "reconcileSurfaceWithoutDisposition"); if (stubs.outcomesThrows) outcomeSpy.mockRejectedValue(new Error("db down")); else outcomeSpy.mockResolvedValue((stubs.outcomes ?? { scanned: 0, backfilled: 0 }) as never); if (stubs.strandedThrows) strandedSpy.mockRejectedValue(new Error("db down")); else strandedSpy.mockResolvedValue((stubs.stranded ?? { scanned: 0, requeued: 0 }) as never); + if (stubs.orphanedThrows) orphanedSpy.mockRejectedValue(new Error("db down")); + else orphanedSpy.mockResolvedValue((stubs.orphaned ?? { scanned: 0, requeued: 0 }) as never); await processJob(createTestEnv(), { type: "agent-regate-sweep", requestedBy: "schedule" }); return logs.map((line) => JSON.parse(line) as Record); } it("reports what each scan repaired", async () => { - const logs = await sweepWith({ outcomes: { scanned: 5, backfilled: 3 }, stranded: { scanned: 2, requeued: 1 } }); + const logs = await sweepWith({ outcomes: { scanned: 5, backfilled: 3 }, stranded: { scanned: 2, requeued: 1 }, orphaned: { scanned: 3, requeued: 2 } }); expect(logs.find((log) => log.event === "pr_outcomes_reconciled")).toMatchObject({ scanned: 5, backfilled: 3 }); expect(logs.find((log) => log.event === "pending_closure_verifications_requeued")).toMatchObject({ scanned: 2, requeued: 1 }); + expect(logs.find((log) => log.event === "surface_without_disposition_reconciled")).toMatchObject({ scanned: 3, requeued: 2 }); }); it("stays quiet when a scan looked but repaired nothing", async () => { - const logs = await sweepWith({ outcomes: { scanned: 9, backfilled: 0 }, stranded: { scanned: 4, requeued: 0 } }); + const logs = await sweepWith({ outcomes: { scanned: 9, backfilled: 0 }, stranded: { scanned: 4, requeued: 0 }, orphaned: { scanned: 6, requeued: 0 } }); expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false); expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false); + expect(logs.some((log) => log.event === "surface_without_disposition_reconciled")).toBe(false); }); - it("completes the tick even when both scans throw", async () => { - const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true }); + it("completes the tick even when all three scans throw", async () => { + const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true, orphanedThrows: true }); expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false); expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false); + expect(logs.some((log) => log.event === "surface_without_disposition_reconciled")).toBe(false); }); }); diff --git a/test/unit/surface-disposition-reconciler.test.ts b/test/unit/surface-disposition-reconciler.test.ts new file mode 100644 index 0000000000..0683e20cd7 --- /dev/null +++ b/test/unit/surface-disposition-reconciler.test.ts @@ -0,0 +1,184 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { upsertInstallation, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub, upsertRepositorySettings } from "../../src/db/repositories"; +import { processJob } from "../../src/queue/processors"; +import { + DISPOSITION_CONSIDERED_EVENT_TYPE, + reconcileSurfaceWithoutDisposition, + SURFACE_DISPOSITION_RECONCILE_LIMIT, + SURFACE_DISPOSITION_RECONCILE_LOOKBACK_MS, +} from "../../src/review/surface-disposition-reconciler"; +import { createTestEnv } from "../helpers/d1"; +import { generatePrivateKeyPem } from "../helpers/github-app-key"; + +function envWithQueue(): { env: Env; sent: unknown[] } { + const sent: unknown[] = []; + const env = createTestEnv(); + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { + send: async (msg: unknown) => void sent.push(msg), + }; + return { env, sent }; +} + +async function seedPr(env: Env, number: number, headSha: string, opts: { lastPublishedSurfaceSha?: string | null; installationId?: number } = {}): Promise { + const installationId = opts.installationId ?? 77; + await upsertInstallation(env, { + installation: { id: installationId, account: { login: "alice", id: installationId, type: "User" }, repository_selection: "selected", permissions: { metadata: "read" }, events: ["pull_request"] }, + }); + await upsertRepositoryFromGitHub(env, { name: "repo", full_name: "alice/repo", private: false, owner: { login: "alice" } }, installationId); + await upsertPullRequestFromGitHub(env, "alice/repo", { number, title: `PR ${number}`, state: "open", user: { login: "bob" }, head: { sha: headSha }, labels: [], body: "b" }); + if (opts.lastPublishedSurfaceSha !== undefined) { + await env.DB.prepare("UPDATE pull_requests SET last_published_surface_sha = ? WHERE repo_full_name = ? AND number = ?") + .bind(opts.lastPublishedSurfaceSha, "alice/repo", number) + .run(); + } +} + +async function recordDispositionMarker(env: Env, number: number, headSha: string): Promise { + await env.DB.prepare("INSERT INTO audit_events (id, event_type, actor, target_key, outcome, detail, metadata_json, created_at) VALUES (?,?,?,?,?,?,?,?)") + .bind( + crypto.randomUUID(), + DISPOSITION_CONSIDERED_EVENT_TYPE, + "loopover", + `alice/repo#${number}#${headSha}`, + "completed", + "maintenance actuation lock claimed; disposition attempt begins for this head", + "{}", + new Date().toISOString(), + ) + .run(); +} + +// #8997: a deploy restart can kill the pass between the public-surface publish and the disposition +// plan/execute for the SAME head. The panel/check-run/CI aggregate are all current for that head (publish +// completed before the kill), so the ordinary stale-surface repair check sees nothing wrong — the confirmed +// "decisive panel, PR still open" shape (#8965: panel published 14:55:24Z, matching close only landed +// 15:07:53Z, ~12 minutes later, purely because an unrelated later sweep happened to re-run the whole pass). +// The write side of the same fix: maybeRunAgentMaintenance (processors.ts) records the marker the moment it +// claims the per-PR actuation lock for a head — i.e. the moment a real disposition attempt begins — so a real, +// end-to-end pass leaves the exact row the scan above reads. +describe("maybeRunAgentMaintenance records the disposition marker (#8997)", () => { + afterEach(() => vi.unstubAllGlobals()); + + it("writes the marker once a real maintenance pass claims the actuation lock for this head", async () => { + const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); + await upsertInstallation(env, { + action: "created", + installation: { id: 9500, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] }, + }); + await upsertRepositoryFromGitHub(env, { name: "marker-repo", full_name: "owner/marker-repo", private: false, owner: { login: "owner" } }, 9500); + await upsertRepositorySettings(env, { repoFullName: "owner/marker-repo", autonomy: { review_state_label: "auto" } }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (/\/pulls\/40(?:\?|$)/.test(url)) return Response.json({ number: 40, title: "Real pass", state: "open", user: { login: "contributor" }, head: { sha: "m1" }, labels: [] }); + if (url.includes("/commits/m1/check-runs")) return Response.json({ total_count: 0, check_runs: [] }); + if (url.includes("/commits/m1/status")) return Response.json({ statuses: [] }); + if (url.includes("/check-runs")) return Response.json({ id: 901 }, { status: 201 }); + return new Response("not found", { status: 404 }); + }); + + await processJob(env, { + type: "github-webhook", + deliveryId: "marker-1", + eventName: "pull_request", + payload: { + action: "opened", + installation: { id: 9500 }, + repository: { name: "marker-repo", full_name: "owner/marker-repo", private: false, owner: { login: "owner" } }, + pull_request: { number: 40, title: "Real pass", state: "open", user: { login: "contributor" }, head: { sha: "m1" }, labels: [], body: "x" }, + }, + }); + + const row = await env.DB.prepare("select target_key as targetKey from audit_events where event_type = ?").bind(DISPOSITION_CONSIDERED_EVENT_TYPE).first<{ targetKey: string }>(); + expect(row?.targetKey).toBe("owner/marker-repo#40#m1"); + }); +}); + +describe("reconcileSurfaceWithoutDisposition (#8997)", () => { + it("re-enqueues a regate for a PR whose current-head surface has no disposition marker", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 1, "sha1", { lastPublishedSurfaceSha: "sha1" }); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 1, requeued: 1 }); + expect(sent).toEqual([{ type: "agent-regate-pr", deliveryId: "surface-without-disposition:alice/repo#1#sha1", repoFullName: "alice/repo", prNumber: 1, installationId: 77 }]); + }); + + it("does NOT re-enqueue once a disposition marker is on record for that exact head", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 2, "sha1", { lastPublishedSurfaceSha: "sha1" }); + await recordDispositionMarker(env, 2, "sha1"); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("does NOT flag a PR whose surface is stale — that is the pre-existing outage-repair check's job", async () => { + const { env, sent } = envWithQueue(); + // Surface still describes an OLDER head; publish itself never completed for the current one. + await seedPr(env, 3, "sha-new", { lastPublishedSurfaceSha: "sha-old" }); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("a marker is scoped to its own head SHA — a stale marker from a prior commit does not cover a new push", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 4, "sha-new", { lastPublishedSurfaceSha: "sha-new" }); + // The marker was recorded for the PR's PREVIOUS head; a fresh push moved the head with no new marker yet. + await recordDispositionMarker(env, 4, "sha-old"); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 1, requeued: 1 }); + expect(sent).toHaveLength(1); + }); + + it("leaves a closed/merged PR alone — there is no live disposition left to reconcile", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 5, "sha1", { lastPublishedSurfaceSha: "sha1" }); + await env.DB.prepare("UPDATE pull_requests SET state = 'closed' WHERE repo_full_name = ? AND number = ?").bind("alice/repo", 5).run(); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("ignores anything older than the lookback window", async () => { + const { env } = envWithQueue(); + await seedPr(env, 6, "sha1", { lastPublishedSurfaceSha: "sha1" }); + + expect(await reconcileSurfaceWithoutDisposition(env, Date.now() + SURFACE_DISPOSITION_RECONCILE_LOOKBACK_MS + 60_000)).toEqual({ scanned: 0, requeued: 0 }); + }); + + it("does not count a rescue whose enqueue failed", async () => { + const { env } = envWithQueue(); + await seedPr(env, 7, "sha1", { lastPublishedSurfaceSha: "sha1" }); + (env as unknown as { JOBS: { send: () => Promise } }).JOBS = { send: async () => Promise.reject(new Error("queue down")) }; + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 1, requeued: 0 }); + }); + + it("returns an empty result instead of throwing when the scan fails", async () => { + const { env } = envWithQueue(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + vi.spyOn(env.DB, "prepare").mockImplementation(() => { + throw new Error("db unavailable"); + }); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(warn.mock.calls.some(([line]) => String(line).includes("surface_disposition_reconcile_scan_failed"))).toBe(true); + vi.restoreAllMocks(); + }); + + it("bounds each run so a large backlog drains across runs instead of blocking one", () => { + expect(SURFACE_DISPOSITION_RECONCILE_LIMIT).toBeGreaterThan(0); + expect(SURFACE_DISPOSITION_RECONCILE_LIMIT).toBeLessThanOrEqual(1000); + }); + + it("skips a PR belonging to an uninstalled/unregistered repository — nothing to enqueue against", async () => { + const { env, sent } = envWithQueue(); + await upsertRepositoryFromGitHub(env, { name: "orphan-repo", full_name: "alice/orphan-repo", private: false, owner: { login: "alice" } }); + await upsertPullRequestFromGitHub(env, "alice/orphan-repo", { number: 8, title: "PR 8", state: "open", user: { login: "bob" }, head: { sha: "sha1" }, labels: [], body: "b" }); + await env.DB.prepare("UPDATE pull_requests SET last_published_surface_sha = ? WHERE repo_full_name = ? AND number = ?").bind("sha1", "alice/orphan-repo", 8).run(); + + expect(await reconcileSurfaceWithoutDisposition(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); +});