From 8331a7403c2610446236cbdb7d6bdb2104795b2e Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 26 Jul 2026 19:55:48 -0700 Subject: [PATCH] =?UTF-8?q?fix(selfhost):=20durability=20=E2=80=94=20queue?= =?UTF-8?q?=20leases,=20atomic=20migrations,=20and=20two=20repair=20scans?= =?UTF-8?q?=20for=20state=20a=20crash=20used=20to=20lose?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four ways a badly-timed process death left permanent damage. #9007 made this urgent rather than theoretical: busy deploys currently end in SIGKILL, so "killed at the wrong moment" is the normal case, not the rare one. On Postgres the queue is deliberately multi-instance (that is what the FOR UPDATE SKIP LOCKED claim is for), so "every processing row" is not "my crashed predecessor's work" — during any overlapped or rolling deploy it also includes jobs a SIBLING is running right now. Both processes then ran the same PR pass concurrently, producing duplicate gate check-runs and verdict thrash. Recovery is now scoped to rows whose lease has actually expired, the same definition the runtime reaper already uses, so boot and steady-state stop disagreeing about what "abandoned" means, and the re-pend is a compare-and-swap so a sibling finishing first cannot be resurrected. Two related holes in the same file. The lease is now heartbeated by the worker that holds it: without that, the timeout was a ceiling on job DURATION rather than on liveness, so a legitimately long pass — a large PR, a slow AI review, a rate-limited GitHub window — was reclaimed and double-run purely for taking too long, and the activeJobIds guard meant to prevent that only covers one process's in-memory set, which is the wrong scope for a shared queue. And a reclaim now costs an attempt: it never did, so a job that reliably wedges past its lease was requeued forever with attempts frozen, never reaching the dead-letter threshold that exists precisely to stop a poison job from burning the queue. Statements ran individually and the file was recorded applied only after the last one succeeded, so a crash mid-file re-ran the WHOLE file next boot. Harmless for pure DDL and destructive for the several migrations carrying real DML — the table-rebuild INSERT ... SELECT pattern, plus UPDATE/DELETE steps — where a re-run either double-inserts or raises a PK conflict that, not matching the tolerated "already exists" shape, throws and bricks boot outright. Each file now runs as one transaction with its ledger row committed inside it, via a new execTransaction on both self-host adapters. It could not reuse exec: on Postgres each exec call takes whatever connection the pool hands out, so a BEGIN and its COMMIT issued as separate calls can land on different connections and mean nothing. The pre-existing drift tolerance is preserved rather than replaced — a database that already drifted must still heal itself instead of failing to boot, and that only works per-statement. But ONLY drift falls back. Any other failure rethrows with the transaction already rolled back, because retrying a genuinely broken file per-statement would re-apply its valid statements after the rollback and leave exactly the partial state this fix exists to prevent. pr_outcome is what the fleet calibration export INNER JOINs gate_decision against, so a PR missing one does not lose a data point, it vanishes from calibration entirely — neither numerator nor denominator. Both writers are in-process and best-effort: a process that dies between the GitHub mutation and the record call loses the direct write, and on the next pass the PR is already terminal so the planner plans nothing and it never fires; the webhook that would have caught it was delivered while the container was down and GitHub does not redeliver. Nothing scanned for the gap — the repair sweep only visits OPEN PRs. These losses are not a random sample: a superseded close is by definition a wrong close, so they skew toward the gate's mistakes and their absence biased published accuracy upward. Pass 1 of the flag-then-close double-check enqueues a single delayed job to run Pass 2, and that message was the only thing that could finish the sequence. Its documented "next sweep / CI event" backstop is vacuous: #never-endless-reregate makes an already-regated PR permanently sweep-ineligible while the flag itself suppresses merge and approve, and the violation memory is permanent so the flag could never clear either. Pass 1 now also records the deadline, and a watchdog re-enqueues Pass 2 for any flag past it. audit_events is the right store: durable, already queried by type, and repo-agnostic — unlike the label, which is per-repo configurable and would need settings resolution before a cross-repo scan could even recognize it. Both repair scans ride the re-gate sweep's existing fan-out tick rather than each earning a job type and a cron entry for a bounded DB scan, and run before the fan-out so neither can cost the tick its actual work. Also fixes main: test/unit/docs-beta-onboarding.test.ts asserted the doc still says "base-agent", which #9117 intentionally retired when it repositioned the product away from "a deterministic base-agent for the Gittensor ecosystem" toward an agent stack for both sides of the pull request on any GitHub repo. The guard exists to stop LoopOver claiming to BE the official Gittensor frontend, and its other four assertions cover that on their own; pinning the retired wording only made it veto an intended product change. Local gate green end to end (npm run test:ci, exit 0). 100% line and branch coverage on all 210 added src lines. Closes #9023 Closes #9026 Closes #9027 Closes #9031 --- src/queue/job-dispatch.ts | 25 +- src/queue/processors.ts | 6 + src/review/pending-closure-watchdog.ts | 119 +++++++++ src/review/pr-outcome-reconciler.ts | 94 +++++++ src/selfhost/backend-contracts.ts | 7 + src/selfhost/d1-adapter.ts | 18 ++ src/selfhost/migrate.ts | 56 +++- src/selfhost/pg-adapter.ts | 22 ++ src/selfhost/pg-queue.ts | 93 ++++++- test/unit/docs-beta-onboarding.test.ts | 6 +- test/unit/job-dispatch.test.ts | 43 ++++ test/unit/pending-closure-watchdog.test.ts | 144 +++++++++++ test/unit/pr-outcome-reconciler.test.ts | 137 ++++++++++ .../unit/selfhost-migration-atomicity.test.ts | 106 ++++++++ test/unit/selfhost-pg-adapter.test.ts | 55 ++++ test/unit/selfhost-pg-queue.test.ts | 240 +++++++++++++++++- 16 files changed, 1153 insertions(+), 18 deletions(-) create mode 100644 src/review/pending-closure-watchdog.ts create mode 100644 src/review/pr-outcome-reconciler.ts create mode 100644 test/unit/pending-closure-watchdog.test.ts create mode 100644 test/unit/pr-outcome-reconciler.test.ts create mode 100644 test/unit/selfhost-migration-atomicity.test.ts diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 37a3b6eb2f..f3558f5254 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -51,6 +51,8 @@ import { isDecisionAuditEnabled, runDecisionAuditSample } from "../review/decisi import { isRiskControlEnabled, runRiskControlRecalibration } from "../review/risk-control-wire"; 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"; // 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). @@ -273,13 +275,30 @@ export async function processJob(env: Env, message: JobMessage): Promise { } case "agent-regate-sweep": if (!message.repoFullName && message.requestedBy !== "test") { - // #9032: piggyback the approval-queue staleness pass on the sweep's own fan-out tick rather than adding - // a job type and a cron entry for a bounded DB scan. Best-effort and deliberately BEFORE the fan-out: - // a failure here must not cost the tick its re-gate work, which is the sweep's actual job. + // Three 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. + // + // #9032 reminds and expires approval-queue rows nobody has acted on. + // + // #9026 backfills pr_outcome rows lost when a process died between the GitHub mutation and the + // bookkeeping write. Those losses skew toward the gate's MISTAKES, so leaving them out biases published + // accuracy upward -- this is calibration ground truth, not ordinary telemetry. + // + // #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. 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 })); } + const outcomes = await reconcileMissingPrOutcomes(env).catch(() => null); + if (outcomes && outcomes.backfilled > 0) { + console.log(JSON.stringify({ event: "pr_outcomes_reconciled", ...outcomes })); + } + const stranded = await sweepStrandedPendingClosures(env).catch(() => null); + if (stranded && stranded.requeued > 0) { + console.log(JSON.stringify({ event: "pending_closure_verifications_requeued", ...stranded })); + } await fanOutAgentRegateSweepJobs(env, message.requestedBy); return; } diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 3254d6bf67..b0742bafda 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -301,6 +301,7 @@ import { } from "../services/agent-action-executor"; import { activeMergeBlockedSha } from "../services/merge-failure"; import { applyLowConfidenceHoldCap } from "../review/low-confidence-hold-cap"; +import { recordPendingClosureFlag } from "../review/pending-closure-watchdog"; import { loadIssueQualityReportMap } from "../services/issue-quality"; import { generateAndSendReviewRecap } from "../services/review-recap"; import { @@ -3676,6 +3677,11 @@ async function runAgentMaintenancePlanAndExecute( ? env.JOBS.send(verifyJob, { delaySeconds }) : env.JOBS.send(verifyJob) ).catch(() => undefined); + // #9031: the durable half of the same intent. The enqueue above is a single queue message, and until now + // it was the ONLY thing that could finish this sequence -- lose it and the PR sits flagged forever, since + // the flag suppresses merge/approve and #never-endless-reregate makes the PR sweep-ineligible. Recording + // the deadline lets sweepStrandedPendingClosures re-enqueue Pass 2 if it never ran. + await recordPendingClosureFlag(env, { repoFullName, pullNumber: pr.number, installationId }, Date.now() + delaySeconds * 1000); } } diff --git a/src/review/pending-closure-watchdog.ts b/src/review/pending-closure-watchdog.ts new file mode 100644 index 0000000000..eeeef4d177 --- /dev/null +++ b/src/review/pending-closure-watchdog.ts @@ -0,0 +1,119 @@ +import { countRecentAuditEventsForActorAndTarget, getPullRequest, listAuditEventsByType, recordAuditEvent } from "../db/repositories"; +import { errorMessage } from "../utils/json"; + +/** + * #9031 — rescue a PR stranded mid-way through the flag-then-close double-check. + * + * Pass 1 applies the pending-closure label and enqueues a single delayed `recapture-preview` job to run Pass 2. + * Until then that one queue message was the ONLY thing that could finish the sequence. Its documented backstop — + * "the next sweep / CI event" — is largely vacuous: #never-endless-reregate makes an already-regated PR + * permanently sweep-ineligible, and the flag itself suppresses merge and approve via `linkedIssueCloseInFlight`. + * So if the queue lost the job (a crash mid-flight, #9007), the PR sat flagged with pending-closure plus + * changes-requested, going nowhere, waiting on a contributor webhook that may never come — and because the + * violation memory is permanent, `clearLinkedIssueFlag` could never fire either. + * + * The fix the issue asks for is a deadline something re-checks, rather than a sequence that depends on one + * message surviving. Pass 1 now records an audit event carrying that deadline, and this watchdog re-enqueues + * Pass 2 for any flag past it. `audit_events` is the right store for this: durable, already queried by type, + * and repo-agnostic — unlike the label, which is per-repo configurable and would need settings resolution + * before a cross-repo scan could even recognize it. + * + * The re-enqueued job is the same message Pass 1 sends, so a redundant one is harmless: Pass 2 is a re-review + * that re-derives everything from live state. + */ + +export const PENDING_CLOSURE_FLAGGED_EVENT = "agent.linked_issue.pending_closure_flagged"; +export const PENDING_CLOSURE_REQUEUED_EVENT = "agent.linked_issue.pending_closure_requeued"; + +/** Grace beyond the flag's own deadline before the watchdog steps in, so a job merely waiting its turn behind a + * busy queue is never mistaken for a lost one. */ +export const PENDING_CLOSURE_GRACE_MS = 10 * 60 * 1000; + +/** How far back to scan. A flag older than this is past rescuing by re-enqueue — the PR has moved on, or a + * human has dealt with it — and the scan stays cheap by not carrying that history forever. */ +export const PENDING_CLOSURE_LOOKBACK_MS = 7 * 24 * 60 * 60 * 1000; + +/** Minimum spacing between rescues for the same PR, so a PR that is stuck for a reason the re-enqueue cannot + * fix gets retried periodically instead of on every single sweep tick. */ +export const PENDING_CLOSURE_REQUEUE_INTERVAL_MS = 60 * 60 * 1000; + +export type PendingClosureSweepResult = { scanned: number; requeued: number }; + +/** Pass 1's durable record that a flag was applied and when its verification is due. Best-effort, exactly like + * the enqueue it accompanies — but the two failing together is far less likely than the single message being + * lost, which is the whole point of having both. */ +export async function recordPendingClosureFlag( + env: Env, + target: { repoFullName: string; pullNumber: number; installationId: number }, + dueAtMs: number, +): Promise { + await recordAuditEvent(env, { + eventType: PENDING_CLOSURE_FLAGGED_EVENT, + actor: "loopover", + targetKey: `${target.repoFullName}#${target.pullNumber}`, + outcome: "queued", + detail: "pending-closure flag applied; Pass 2 verification scheduled", + metadata: { repoFullName: target.repoFullName, pullNumber: target.pullNumber, installationId: target.installationId, dueAt: new Date(dueAtMs).toISOString() }, + }).catch(() => undefined); +} + +export async function sweepStrandedPendingClosures(env: Env, nowMs: number = Date.now()): Promise { + let flags: Awaited> = []; + try { + flags = await listAuditEventsByType(env, PENDING_CLOSURE_FLAGGED_EVENT, new Date(nowMs - PENDING_CLOSURE_LOOKBACK_MS).toISOString(), 500); + } catch (error) { + console.warn(JSON.stringify({ level: "warn", event: "pending_closure_sweep_scan_failed", message: errorMessage(error).slice(0, 160) })); + return { scanned: 0, requeued: 0 }; + } + + let requeued = 0; + for (const flag of flags) { + const repoFullName = typeof flag.metadata.repoFullName === "string" ? flag.metadata.repoFullName : null; + const pullNumber = typeof flag.metadata.pullNumber === "number" ? flag.metadata.pullNumber : null; + const installationId = typeof flag.metadata.installationId === "number" ? flag.metadata.installationId : null; + if (repoFullName === null || pullNumber === null || installationId === null) continue; + const dueAt = typeof flag.metadata.dueAt === "string" ? Date.parse(flag.metadata.dueAt) : NaN; + // An unreadable deadline falls back to the event's own timestamp: the flag is still real, and refusing to + // rescue it because its metadata is malformed would recreate the exact stranding this exists to end. + const deadline = Number.isFinite(dueAt) ? dueAt : Date.parse(flag.createdAt); + if (!Number.isFinite(deadline) || nowMs < deadline + PENDING_CLOSURE_GRACE_MS) continue; + + // Only an OPEN PR can still be stranded. A closed/merged one already reached a terminal state, whether by + // Pass 2, a maintainer, or the contributor. + const pr = await getPullRequest(env, repoFullName, pullNumber).catch(() => null); + if (!pr || pr.state !== "open") continue; + + const targetKey = `${repoFullName}#${pullNumber}`; + const recent = await countRecentAuditEventsForActorAndTarget( + env, + "loopover", + PENDING_CLOSURE_REQUEUED_EVENT, + targetKey, + new Date(nowMs - PENDING_CLOSURE_REQUEUE_INTERVAL_MS).toISOString(), + ).catch(() => 1); + if (recent > 0) continue; + + const verifyJob = { + type: "recapture-preview" as const, + deliveryId: `linked-issue-verify:${repoFullName}#${pullNumber}`, + repoFullName, + prNumber: pullNumber, + installationId, + attempt: 0, + }; + const sent = await env.JOBS.send(verifyJob) + .then(() => true) + .catch(() => false); + if (!sent) continue; + requeued += 1; + await recordAuditEvent(env, { + eventType: PENDING_CLOSURE_REQUEUED_EVENT, + actor: "loopover", + targetKey, + outcome: "queued", + detail: "pending-closure verification re-enqueued after its deadline passed with the PR still flagged", + metadata: { repoFullName, pullNumber, installationId, deadline: new Date(deadline).toISOString() }, + }).catch(() => undefined); + } + return { scanned: flags.length, requeued }; +} diff --git a/src/review/pr-outcome-reconciler.ts b/src/review/pr-outcome-reconciler.ts new file mode 100644 index 0000000000..cd0d50e7e7 --- /dev/null +++ b/src/review/pr-outcome-reconciler.ts @@ -0,0 +1,94 @@ +import { recordTerminalActionOutcome } from "./outcomes-wire"; +import { errorMessage } from "../utils/json"; + +/** + * #9026 — backfill `pr_outcome` rows that were never written. + * + * `pr_outcome` is the realized ground truth the fleet calibration export inner-joins `gate_decision` against, so + * a PR missing one does not merely lose a data point: it vanishes from calibration entirely, contributing to + * neither numerator nor denominator. It is written two ways, both in-process and both best-effort — + * `recordTerminalActionOutcome` immediately after the bot's own merge/close mutation, and `recordPrOutcome` from + * the inbound `pull_request.closed` webhook. + * + * Neither survives a kill at the wrong moment. A process that dies between `mergePullRequest`/`closePullRequest` + * and the record call loses the direct write, and on the next pass the PR is already terminal so the planner + * plans nothing and the write never fires. The webhook that would have caught it was delivered while the + * container was down, and GitHub does not redeliver. Nothing scanned for the gap: the repair sweep only visits + * OPEN PRs. #8823 narrowed the window; it did not close it. + * + * The rows this loses are not a random sample. A superseded close is by definition a wrong close, so the losses + * skew toward the gate's mistakes and their absence biases published accuracy UPWARD — which is why this matters + * more than ordinary telemetry loss, and why it is worth a reconciler rather than a wider window. + * + * Scoped to PRs that actually carry a `gate_decision`, because those are exactly the ones calibration reads; a + * closed PR the gate never evaluated has no prediction to pair an outcome with. `recordTerminalActionOutcome` is + * already idempotent, so a row that raced in between this scan and the write is a no-op. + */ + +/** How far back to look. Long enough to cover a multi-day outage, bounded so the scan stays cheap on every + * run — anything older has already been exported and is beyond repair for the current calibration window. */ +export const PR_OUTCOME_RECONCILE_LOOKBACK_MS = 14 * 24 * 60 * 60 * 1000; + +/** Cap per run. The scan is ordered oldest-first so a large backlog drains deterministically across runs + * instead of the same newest slice being retried forever. */ +export const PR_OUTCOME_RECONCILE_LIMIT = 200; + +export type PrOutcomeReconcileResult = { scanned: number; backfilled: number }; + +/** + * Find terminal PRs with a gate decision but no recorded outcome, and write the outcome. Returns what it did so + * the caller can log it. Best-effort throughout: this is a repair pass, and a repair pass that can itself break + * the tick it runs on is worse than the gap it closes. + */ +export async function reconcileMissingPrOutcomes(env: Env, nowMs: number = Date.now()): Promise { + const since = new Date(nowMs - PR_OUTCOME_RECONCILE_LOOKBACK_MS).toISOString(); + let rows: Array<{ repoFullName: string; number: number; mergedAt: string | null }> = []; + try { + const result = await env.DB.prepare( + `SELECT pr.repo_full_name AS repoFullName, pr.number AS number, pr.merged_at AS mergedAt + FROM pull_requests AS pr + WHERE pr.state != 'open' + AND pr.updated_at >= ?1 + AND EXISTS ( + SELECT 1 FROM review_audit AS gate + WHERE gate.target_id = pr.repo_full_name || '#' || pr.number + AND gate.event_type = 'gate_decision' + ) + AND NOT EXISTS ( + SELECT 1 FROM review_audit AS outcome + WHERE outcome.target_id = pr.repo_full_name || '#' || pr.number + AND outcome.event_type = 'pr_outcome' + ) + ORDER BY pr.updated_at + LIMIT ?2`, + ) + .bind(since, PR_OUTCOME_RECONCILE_LIMIT) + .all<{ repoFullName: string; number: number; mergedAt: string | null }>(); + rows = result.results ?? []; + } catch (error) { + console.warn(JSON.stringify({ level: "warn", event: "pr_outcome_reconcile_scan_failed", message: errorMessage(error).slice(0, 160) })); + return { scanned: 0, backfilled: 0 }; + } + + let backfilled = 0; + for (const row of rows) { + // `merged_at` is the authoritative signal, not the state string: GitHub reports a merged PR as `closed` + // too, and recording a merge as a plain close would invert the very judgment calibration is scoring. + const decision = row.mergedAt ? "merged" : "closed"; + try { + await recordTerminalActionOutcome(env, row.repoFullName, row.number, decision); + backfilled += 1; + } catch (error) { + console.warn( + JSON.stringify({ + level: "warn", + event: "pr_outcome_reconcile_write_failed", + repo: row.repoFullName, + pr: row.number, + message: errorMessage(error).slice(0, 160), + }), + ); + } + } + return { scanned: rows.length, backfilled }; +} diff --git a/src/selfhost/backend-contracts.ts b/src/selfhost/backend-contracts.ts index 3d65c6dd5e..a108b6d1c7 100644 --- a/src/selfhost/backend-contracts.ts +++ b/src/selfhost/backend-contracts.ts @@ -88,6 +88,13 @@ export interface SelfHostD1Database { prepare(query: string): SelfHostD1PreparedStatement; batch(statements: SelfHostD1PreparedStatement[]): Promise }>>; exec(query: string): Promise<{ count: number; duration: number }>; + /** Run a multi-statement script as ONE all-or-nothing transaction (#9027). `exec` cannot give this: on + * Postgres it takes an arbitrary pooled connection per call, so a BEGIN and its COMMIT issued as separate + * `exec`s can land on different connections and mean nothing. Migrations need it because several of them + * carry non-idempotent DML (the table-rebuild `INSERT ... SELECT` pattern, plus UPDATE/DELETE steps) that a + * crash mid-file would otherwise re-run in full on the next boot — double-inserting, or failing on a PK + * conflict that is not "already exists" and so bricking boot outright. */ + execTransaction(query: string): Promise; /** @deprecated present only because both self-host adapters still implement it for D1 surface completeness; * real D1 no longer uses it either (see d1-adapter.ts / pg-adapter.ts). */ dump(): Promise; diff --git a/src/selfhost/d1-adapter.ts b/src/selfhost/d1-adapter.ts index 394bac1af4..510f0bce4a 100644 --- a/src/selfhost/d1-adapter.ts +++ b/src/selfhost/d1-adapter.ts @@ -92,6 +92,24 @@ export function createD1Adapter(driver: SqliteDriver): D1Database { driver.exec(sql); // runs one or more statements (used for migrations) return { count: (sql.match(/;/g) ?? []).length || 1, duration: 0 }; }, + async execTransaction(sql: string) { + // #9027: SQLite is a single connection here, so BEGIN/COMMIT issued as separate exec calls are already + // the same session — unlike Postgres, where that is exactly the trap. Still explicit rather than relying + // on the driver: node:sqlite's exec() runs statements sequentially with no implicit transaction, so a + // crash partway through a migration file would leave it half-applied without this. + driver.exec("BEGIN"); + try { + driver.exec(sql); + driver.exec("COMMIT"); + } catch (error) { + try { + driver.exec("ROLLBACK"); + } catch { + /* ignore -- the original error is the one worth reporting */ + } + throw error; + } + }, async dump() { return new ArrayBuffer(0); // unused by loopover; present for D1 surface completeness }, diff --git a/src/selfhost/migrate.ts b/src/selfhost/migrate.ts index 7efdb1f7ae..cc5ccc9693 100644 --- a/src/selfhost/migrate.ts +++ b/src/selfhost/migrate.ts @@ -73,16 +73,70 @@ function splitSqlStatements(sql: string): string[] { return statements; } +/** SQL-literal escape for the migration name written into the ledger inside the transactional script. The + * names are our own repo filenames (`NNNN_*.sql`), not user input, but the doubling keeps the script correct + * for any name rather than relying on that. */ +function sqlQuote(value: string): string { + return `'${value.replace(/'/g, "''")}'`; +} + +/** + * Apply pending self-host migrations. + * + * #9027 — each file now runs as ONE transaction, with its ledger row committed inside that same transaction, so + * a file is all-or-nothing. Before this, statements ran individually and the file was recorded only after the + * last one succeeded, so a crash mid-file re-ran the WHOLE file on the next boot. That is fine for pure DDL and + * actively harmful for the several migrations carrying real DML — the table-rebuild `INSERT ... SELECT` + * pattern, plus UPDATE/DELETE steps — where a re-run either double-inserts or raises a PK conflict which, not + * matching the tolerated "already exists" shape, throws and bricks boot. With deploys currently ending in + * SIGKILL under load (#9007), a mid-migration kill is not hypothetical. + * + * The pre-existing drift tolerance is preserved, not replaced. A database that already drifted — a column a + * migration adds is somehow present — must still heal itself rather than fail to boot, and that self-healing + * only works per-statement. So a file that fails atomically is retried through the original tolerant path. + * The two orders matter: atomic first means a genuine mid-file crash is the case that gets protected, and the + * tolerant fallback only runs for a database that was already inconsistent before this boot began. + */ export async function runSelfHostMigrations(db: D1Database, dir: string): Promise { await db.exec("CREATE TABLE IF NOT EXISTS _selfhost_migrations (name TEXT PRIMARY KEY, applied_at TEXT NOT NULL)"); const existing = await db.prepare("SELECT name FROM _selfhost_migrations").all<{ name: string }>(); const applied = new Set(existing.results.map((r) => r.name)); const files = readdirSync(dir).filter((f) => f.endsWith(".sql")).sort(); + // Real Cloudflare D1 has no transactional multi-statement surface; only the two self-host adapters implement + // this. Absent it, behavior is exactly the pre-#9027 path. + const execTransaction = (db as unknown as { execTransaction?: (sql: string) => Promise }).execTransaction; let count = 0; for (const file of files) { if (applied.has(file)) continue; const sql = readFileSync(join(dir, file), "utf8"); - for (const statement of splitSqlStatements(sql)) { + const statements = splitSqlStatements(sql); + const ledger = `INSERT INTO _selfhost_migrations (name, applied_at) VALUES (${sqlQuote(file)}, ${sqlQuote(new Date().toISOString())});`; + if (execTransaction) { + try { + // splitSqlStatements keeps each statement's own trailing `;` but returns a final unterminated tail + // verbatim, so re-terminate before concatenating — otherwise the ledger INSERT fuses onto the last + // statement and the whole file fails to parse. + const script = statements.map((statement) => (statement.endsWith(";") ? statement : `${statement};`)).join("\n"); + await execTransaction.call(db, `${script}\n${ledger}`); + count += 1; + continue; + } catch (error) { + // ONLY drift falls through. Any other failure rethrows here, with the transaction already rolled back -- + // so boot dies loudly against a database in exactly the state it started in. Retrying a genuinely broken + // file through the per-statement path would re-apply its valid statements after the rollback and leave + // precisely the partial state this fix exists to prevent. + if (!/duplicate column|already exists/i.test(errorMessage(error))) throw error; + console.warn( + JSON.stringify({ + level: "warn", + event: "selfhost_migration_transaction_failed", + file, + message: errorMessage(error), + }), + ); + } + } + for (const statement of statements) { try { await db.exec(statement); } catch (error) { diff --git a/src/selfhost/pg-adapter.ts b/src/selfhost/pg-adapter.ts index 1d666fa481..5651453940 100644 --- a/src/selfhost/pg-adapter.ts +++ b/src/selfhost/pg-adapter.ts @@ -90,6 +90,28 @@ export function createPgAdapter(pool: Pool): D1Database { await pool.query(translateDdl(sql)); return { count: (sql.match(/;/g) ?? []).length || 1, duration: 0 }; }, + async execTransaction(sql: string) { + // #9027: a dedicated client, so BEGIN/COMMIT/ROLLBACK are guaranteed to be the SAME session — the thing + // `exec` above cannot promise, since each of its calls takes whatever connection the pool hands out. + // Postgres DDL is transactional, so a migration file wrapped this way is genuinely all-or-nothing. + // Mirrors batch()'s transaction handling above, including releasing the client on every path so a failed + // transaction can never be returned to the pool still open. + const client = await pool.connect(); + try { + await client.query("BEGIN"); + await client.query(translateDdl(sql)); + await client.query("COMMIT"); + } catch (error) { + try { + await client.query("ROLLBACK"); + } catch { + /* ignore -- the original error is the one worth reporting */ + } + throw error; + } finally { + client.release(); + } + }, async dump() { return new ArrayBuffer(0); // unused; present for D1 surface completeness }, diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 7ee5bdd6a0..1f460cd8c1 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -565,24 +565,65 @@ export function createPgQueue( return result.rowCount ?? 0; } + /** + * Boot recovery: re-pend jobs a previous process left mid-flight. + * + * #9023 — this used to re-pend EVERY row in `processing`, unconditionally. On Postgres the queue is + * explicitly multi-instance (that is what the FOR UPDATE SKIP LOCKED claim exists for), so "every processing + * row" is not "my crashed predecessor's work" — during any overlapped or rolling deploy it also includes jobs + * a SIBLING instance is running RIGHT NOW. Both processes then ran the same PR pass concurrently, producing + * duplicate gate check-runs and verdict thrash. + * + * The lease cutoff is the fix: only a row whose lease has actually expired can belong to a dead process, and + * that is exactly the condition reclaimExpiredProcessingJobs already encodes, so boot and steady-state now + * agree on what "abandoned" means instead of using two different definitions. A live sibling heartbeats its + * lease (see heartbeatProcessingLease), so its rows are never eligible however long the job legitimately runs. + */ async function recoverProcessingJobs(): Promise { + // A disabled timeout means leases never expire, so nothing can be proven abandoned -- recovering anything + // would be a guess, and the guess is the bug. The runtime reaper is disabled in that configuration too. + if (processingTimeoutMs <= 0) return 0; + const now = Date.now(); const res = await pool.query( - `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing'`, + `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing' AND run_after<=$1`, + [now - processingTimeoutMs], ); let changed = 0; - const now = Date.now(); const maxJitter = queueRecoveryJitterMs(); for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) { const runAfter = now + deterministicJitterMs(`${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); - await pool.query(`UPDATE ${TABLE} SET status='pending', run_after=$1 WHERE id=$2`, [ - runAfter, - row.id, - ]); - changed += 1; + // AND status='processing' makes this a compare-and-swap: between the SELECT above and this UPDATE a + // sibling may have finished the job (row deleted) or reclaimed it itself. Without the guard this could + // resurrect a completed job as pending. Mirrors reclaimExpiredProcessingJobs' own guard. + const update = await pool.query( + `UPDATE ${TABLE} SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'`, + [runAfter, row.id], + ); + changed += update.rowCount ?? 0; } return changed; } + /** + * Refresh the lease on a job this process is actively running (#9023). + * + * `run_after` doubles as the lease timestamp: the claim sets it to now, and a row is reclaimable once it is + * older than `processingTimeoutMs`. Without a heartbeat that made the timeout a hard ceiling on job DURATION, + * not on liveness — a genuinely long pass (a large PR, a slow AI review, a rate-limited GitHub window) got + * reclaimed and double-run by a sibling purely for taking too long. The `activeJobIds` check that was meant to + * prevent this only covers THIS process's in-memory set, which is precisely the wrong scope for a queue whose + * whole point is that several processes share it. + * + * Best-effort by design: a failed heartbeat just means the lease ages normally, which is the pre-#9023 + * behavior. Losing the row (deleted on completion, or dead-lettered) makes this a harmless no-op via the + * status guard. + */ + async function heartbeatProcessingLease(id: string): Promise { + await pool + .query(`UPDATE ${TABLE} SET run_after=$1 WHERE id=$2 AND status='processing'`, [Date.now(), id]) + .catch(() => undefined); + } + // Dead-letter auto-retry (#audit-rate-headroom): a job dies once `attempts >= maxRetries` (see the // max-retries branch in processOne below). Reviving it here only clears `status`/`run_after`/`last_error` // -- `attempts` is left untouched, so it already satisfies `attempts >= maxRetries` and will die again @@ -1114,6 +1155,13 @@ export function createPgQueue( if (!job) return false; activeJobIds.add(job.id); const claimedAt = Date.now(); + // #9023: hold the lease open for as long as this process is genuinely working on the job. Without this the + // lease timeout is a ceiling on job DURATION rather than on liveness, so a legitimately slow pass gets + // reclaimed and double-run by a sibling. Refreshed at a third of the timeout so two refreshes can be lost + // (a blip, a paused event loop) before the lease is actually at risk. + const heartbeat = setInterval(() => void heartbeatProcessingLease(job.id), Math.max(1000, Math.floor(processingTimeoutMs / 3))); + // Never let the timer hold the process open: shutdown must not wait on a heartbeat tick. + heartbeat.unref?.(); try { let message: JobMessage; try { @@ -1468,6 +1516,7 @@ export function createPgQueue( } return true; } finally { + clearInterval(heartbeat); activeJobIds.delete(job.id); if (job.backgroundSlotReserved) activeBackground = Math.max(0, activeBackground - 1); @@ -1630,17 +1679,39 @@ export function createPgQueue( const now = Date.now(); const cutoff = now - processingTimeoutMs; const res = await pool.query( - `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing' AND run_after<=$1`, + `SELECT id, payload, job_key, attempts FROM ${TABLE} WHERE status='processing' AND run_after<=$1`, [cutoff], ); let changed = 0; const maxJitter = queueRecoveryJitterMs(); - for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) { + for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null; attempts: number | string }>) { if (activeJobIds.has(row.id)) continue; + // #9023: a reclaim now COSTS an attempt. It never did, so a job that reliably wedges past its lease -- + // an infinite loop, a hang against an unresponsive dependency -- was requeued forever: reclaimed, re-run, + // wedged, reclaimed, with `attempts` frozen at whatever the last real failure left it. It could never + // reach the dead-letter threshold that exists precisely to stop a poison job from burning the queue. + // Charging the attempt makes a chronic wedger die on the same budget as a chronic thrower. + const attempts = Number(row.attempts) + 1; + if (attempts >= maxRetries) { + const update = await pool.query( + `UPDATE ${TABLE} SET status='dead', attempts=$1, last_error=$2, dead_at=$3 WHERE id=$4 AND status='processing'`, + [attempts, "processing lease expired repeatedly; dead-lettered", Date.now(), row.id], + ); + if ((update.rowCount ?? 0) > 0) { + await recordQueueMetric("loopover_jobs_dead_total"); + capturePostHogError(new Error("self-host queue job wedged past its lease repeatedly"), { + kind: "job_dead", + reason: "processing_lease_exhausted", + jobId: row.id, + attempts, + }, "processing_lease_exhausted"); + } + continue; + } const runAfter = now + deterministicJitterMs(`${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); const update = await pool.query( - `UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=COALESCE(last_error, $2) WHERE id=$3 AND status='processing'`, - [runAfter, "processing lease expired; requeued", row.id], + `UPDATE ${TABLE} SET status='pending', run_after=$1, attempts=$2, last_error=COALESCE(last_error, $3) WHERE id=$4 AND status='processing'`, + [runAfter, attempts, "processing lease expired; requeued", row.id], ); changed += update.rowCount ?? 0; } diff --git a/test/unit/docs-beta-onboarding.test.ts b/test/unit/docs-beta-onboarding.test.ts index 28c6fe87a8..ff41cb3072 100644 --- a/test/unit/docs-beta-onboarding.test.ts +++ b/test/unit/docs-beta-onboarding.test.ts @@ -51,7 +51,11 @@ describe("docs beta onboarding page", () => { expect(source).toMatch(/official Gittensor product surface/i); expect(source).toMatch(/official Gittensor frontend/i); expect(source).toMatch(/independent of/i); - expect(source).toMatch(/base-agent/i); + // Deliberately NOT asserting "base-agent" any more. #9117 repositioned the product away from "a + // deterministic base-agent for the Gittensor ecosystem" toward an agent stack for both sides of the pull + // request on ANY GitHub repo, which is a narrowing this test had no business vetoing -- it exists to stop + // LoopOver claiming to BE the official Gittensor frontend, and the four assertions here cover that on + // their own. Pinning the old wording only made the guard fire on an intended product change. expect(source).not.toMatch(/the official Gittensor frontend/i); }); }); diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts index 125ae0e9cd..2c786a7cab 100644 --- a/test/unit/job-dispatch.test.ts +++ b/test/unit/job-dispatch.test.ts @@ -144,3 +144,46 @@ describe("agent-regate-sweep also sweeps the stale approval queue (#9032)", () = await expect(processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" })).resolves.toBeUndefined(); }); }); + +// #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)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + async function sweepWith(stubs: { outcomes?: unknown; stranded?: unknown; outcomesThrows?: boolean; strandedThrows?: 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 outcomeSpy = vi.spyOn(reconciler, "reconcileMissingPrOutcomes"); + const strandedSpy = vi.spyOn(watchdog, "sweepStrandedPendingClosures"); + 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); + + 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 } }); + 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 }); + }); + + 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 } }); + expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false); + expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false); + }); + + it("completes the tick even when both scans throw", async () => { + const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true }); + expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false); + expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false); + }); +}); diff --git a/test/unit/pending-closure-watchdog.test.ts b/test/unit/pending-closure-watchdog.test.ts new file mode 100644 index 0000000000..a9303624fd --- /dev/null +++ b/test/unit/pending-closure-watchdog.test.ts @@ -0,0 +1,144 @@ +import { describe, expect, it, vi } from "vitest"; +import { upsertInstallation, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub } from "../../src/db/repositories"; +import { + PENDING_CLOSURE_GRACE_MS, + PENDING_CLOSURE_LOOKBACK_MS, + PENDING_CLOSURE_REQUEUE_INTERVAL_MS, + recordPendingClosureFlag, + sweepStrandedPendingClosures, +} from "../../src/review/pending-closure-watchdog"; +import { createTestEnv } from "../helpers/d1"; + +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, state: string): Promise { + await upsertInstallation(env, { + installation: { id: 88, account: { login: "alice", id: 88, 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" } }, 88); + await upsertPullRequestFromGitHub(env, "alice/repo", { number, title: "PR", state, user: { login: "bob" }, head: { sha: "s1" }, labels: [], body: "b" }); +} + +// #9031: Pass 1 applies the pending-closure label and enqueues ONE delayed job to run Pass 2. That message was +// the only thing that could finish the sequence — and the documented "next sweep / CI event" backstop is +// vacuous, because #never-endless-reregate makes an already-regated PR permanently sweep-ineligible while the +// flag itself suppresses merge and approve. Lose the message and the PR is stranded for good. +describe("sweepStrandedPendingClosures (#9031)", () => { + const PAST_DUE = (now: number) => now - PENDING_CLOSURE_GRACE_MS - 60_000; + + it("re-enqueues Pass 2 for a flag whose deadline passed with the PR still open", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 11, "open"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 11, installationId: 88 }, PAST_DUE(now)); + + expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 1 }); + expect(sent).toEqual([ + { type: "recapture-preview", deliveryId: "linked-issue-verify:alice/repo#11", repoFullName: "alice/repo", prNumber: 11, installationId: 88, attempt: 0 }, + ]); + }); + + it("waits out the grace period, so a job merely queued behind a busy backlog is not mistaken for a lost one", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 12, "open"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 12, installationId: 88 }, now - 1000); + + expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("leaves a PR that already reached a terminal state alone", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 13, "closed"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 13, installationId: 88 }, PAST_DUE(now)); + + expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("does not rescue the same PR on every tick — a PR stuck for another reason is retried periodically", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 14, "open"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 14, installationId: 88 }, PAST_DUE(now)); + + expect((await sweepStrandedPendingClosures(env, now)).requeued).toBe(1); + expect((await sweepStrandedPendingClosures(env, now + 60_000)).requeued).toBe(0); + // Past the interval it tries again, rather than giving up on a PR that is genuinely still stranded. + expect((await sweepStrandedPendingClosures(env, now + PENDING_CLOSURE_REQUEUE_INTERVAL_MS + 1000)).requeued).toBe(1); + expect(sent).toHaveLength(2); + }); + + it("ignores flags older than the lookback window", async () => { + const { env } = envWithQueue(); + await seedPr(env, 15, "open"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 15, installationId: 88 }, PAST_DUE(now)); + + expect(await sweepStrandedPendingClosures(env, now + PENDING_CLOSURE_LOOKBACK_MS + 60_000)).toEqual({ scanned: 0, requeued: 0 }); + }); + + it("falls back to the flag's own timestamp when the recorded deadline is unreadable", async () => { + const { env, sent } = envWithQueue(); + await seedPr(env, 16, "open"); + await env.DB.prepare("INSERT INTO audit_events (id, event_type, actor, target_key, outcome, detail, metadata_json, created_at) VALUES (?,?,?,?,?,?,?,?)") + .bind( + crypto.randomUUID(), + "agent.linked_issue.pending_closure_flagged", + "loopover", + "alice/repo#16", + "queued", + "flagged", + JSON.stringify({ repoFullName: "alice/repo", pullNumber: 16, installationId: 88, dueAt: "not-a-date" }), + new Date(Date.now() - 3 * PENDING_CLOSURE_GRACE_MS).toISOString(), + ) + .run(); + + // Refusing to rescue a real flag because its metadata is malformed would recreate the exact stranding + // this watchdog exists to end. + expect((await sweepStrandedPendingClosures(env)).requeued).toBe(1); + expect(sent).toHaveLength(1); + }); + + it("skips a flag whose metadata cannot identify a PR at all", async () => { + const { env, sent } = envWithQueue(); + await env.DB.prepare("INSERT INTO audit_events (id, event_type, actor, target_key, outcome, detail, metadata_json, created_at) VALUES (?,?,?,?,?,?,?,?)") + .bind(crypto.randomUUID(), "agent.linked_issue.pending_closure_flagged", "loopover", "alice/repo#17", "queued", "flagged", JSON.stringify({ repoFullName: "alice/repo" }), new Date().toISOString()) + .run(); + + expect(await sweepStrandedPendingClosures(env)).toEqual({ scanned: 1, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("does not count a rescue whose enqueue failed", async () => { + const { env } = envWithQueue(); + await seedPr(env, 18, "open"); + const now = Date.now(); + await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 18, installationId: 88 }, PAST_DUE(now)); + (env as unknown as { JOBS: { send: () => Promise } }).JOBS = { send: async () => Promise.reject(new Error("queue down")) }; + + expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 }); + // And nothing was recorded, so the next tick retries rather than believing it already rescued this PR. + expect((await sweepStrandedPendingClosures(env, now + 1000)).scanned).toBe(1); + }); + + it("returns an empty result instead of throwing when the scan fails", async () => { + const { env } = envWithQueue(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const repositories = await import("../../src/db/repositories"); + vi.spyOn(repositories, "listAuditEventsByType").mockRejectedValue(new Error("db unavailable")); + + expect(await sweepStrandedPendingClosures(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(warn.mock.calls.some(([line]) => String(line).includes("pending_closure_sweep_scan_failed"))).toBe(true); + vi.restoreAllMocks(); + }); +}); diff --git a/test/unit/pr-outcome-reconciler.test.ts b/test/unit/pr-outcome-reconciler.test.ts new file mode 100644 index 0000000000..8155c35501 --- /dev/null +++ b/test/unit/pr-outcome-reconciler.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, it, vi } from "vitest"; +import { upsertInstallation, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub } from "../../src/db/repositories"; +import { PR_OUTCOME_RECONCILE_LIMIT, PR_OUTCOME_RECONCILE_LOOKBACK_MS, reconcileMissingPrOutcomes } from "../../src/review/pr-outcome-reconciler"; +import { createTestEnv } from "../helpers/d1"; + +async function seedRepo(env: Env): Promise { + await upsertInstallation(env, { + installation: { id: 55, account: { login: "alice", id: 55, 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" } }, 55); +} + +async function seedPr(env: Env, number: number, opts: { state: string; mergedAt?: string | null }): Promise { + await upsertPullRequestFromGitHub(env, "alice/repo", { + number, + title: `PR ${number}`, + state: opts.state, + user: { login: "bob" }, + head: { sha: `sha${number}` }, + labels: [], + body: "b", + ...(opts.mergedAt ? { merged_at: opts.mergedAt } : {}), + }); +} + +async function addReviewAudit(env: Env, number: number, eventType: string): Promise { + await env.DB.prepare("INSERT INTO review_audit (id, project, target_id, event_type, decision, created_at) VALUES (?, ?, ?, ?, ?, ?)") + .bind(crypto.randomUUID(), "alice/repo", `alice/repo#${number}`, eventType, "closed", new Date().toISOString()) + .run(); +} + +async function outcomeRows(env: Env, number: number): Promise { + const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM review_audit WHERE target_id = ? AND event_type = 'pr_outcome'") + .bind(`alice/repo#${number}`) + .first<{ c: number }>(); + return Number(row?.c ?? 0); +} + +// #9026: pr_outcome is the realized ground truth the fleet calibration export INNER JOINs gate_decision +// against, so a PR missing one vanishes from calibration entirely — neither numerator nor denominator. Both +// writers are in-process and best-effort, and nothing scanned for the gap: the repair sweep only visits OPEN +// PRs. The losses skew toward the gate's mistakes (a superseded close is by definition a wrong close), so their +// absence biased published accuracy UPWARD. +describe("reconcileMissingPrOutcomes (#9026)", () => { + it("backfills a closed PR that has a gate decision but no outcome", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 1, { state: "closed" }); + await addReviewAudit(env, 1, "gate_decision"); + + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 1, backfilled: 1 }); + expect(await outcomeRows(env, 1)).toBe(1); + }); + + it("records a merged PR as merged, not closed — GitHub reports both as state=closed", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 2, { state: "closed", mergedAt: new Date().toISOString() }); + await addReviewAudit(env, 2, "gate_decision"); + + await reconcileMissingPrOutcomes(env); + const row = await env.DB.prepare("SELECT decision FROM review_audit WHERE target_id = ? AND event_type = 'pr_outcome'") + .bind("alice/repo#2") + .first<{ decision: string }>(); + // Recording a merge as a plain close would invert the very judgment calibration is scoring. + expect(row?.decision).toBe("merged"); + }); + + it("leaves open PRs alone — they have no realized outcome yet", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 3, { state: "open" }); + await addReviewAudit(env, 3, "gate_decision"); + + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 }); + }); + + it("ignores a closed PR the gate never evaluated — no prediction means nothing to pair an outcome with", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 4, { state: "closed" }); + + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 }); + expect(await outcomeRows(env, 4)).toBe(0); + }); + + it("skips a PR whose outcome was already recorded, and stays idempotent across runs", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 5, { state: "closed" }); + await addReviewAudit(env, 5, "gate_decision"); + + await reconcileMissingPrOutcomes(env); + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 }); + expect(await outcomeRows(env, 5)).toBe(1); + }); + + it("ignores anything older than the lookback window", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 6, { state: "closed" }); + await addReviewAudit(env, 6, "gate_decision"); + + expect(await reconcileMissingPrOutcomes(env, Date.now() + PR_OUTCOME_RECONCILE_LOOKBACK_MS + 60_000)).toEqual({ scanned: 0, backfilled: 0 }); + }); + + it("returns an empty result instead of throwing when the scan fails — a repair pass must not break its tick", async () => { + const env = createTestEnv(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + vi.spyOn(env.DB, "prepare").mockImplementation(() => { + throw new Error("db unavailable"); + }); + + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 }); + expect(warn.mock.calls.some(([line]) => String(line).includes("pr_outcome_reconcile_scan_failed"))).toBe(true); + vi.restoreAllMocks(); + }); + + it("counts a scanned row it could not write, rather than reporting it as backfilled", async () => { + const env = createTestEnv(); + await seedRepo(env); + await seedPr(env, 7, { state: "closed" }); + await addReviewAudit(env, 7, "gate_decision"); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const outcomes = await import("../../src/review/outcomes-wire"); + vi.spyOn(outcomes, "recordTerminalActionOutcome").mockRejectedValue(new Error("write failed")); + + expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 1, backfilled: 0 }); + expect(warn.mock.calls.some(([line]) => String(line).includes("pr_outcome_reconcile_write_failed"))).toBe(true); + vi.restoreAllMocks(); + }); + + it("bounds each run so a large backlog drains across runs instead of blocking one", () => { + expect(PR_OUTCOME_RECONCILE_LIMIT).toBeGreaterThan(0); + expect(PR_OUTCOME_RECONCILE_LIMIT).toBeLessThanOrEqual(1000); + }); +}); diff --git a/test/unit/selfhost-migration-atomicity.test.ts b/test/unit/selfhost-migration-atomicity.test.ts new file mode 100644 index 0000000000..022b7d35b1 --- /dev/null +++ b/test/unit/selfhost-migration-atomicity.test.ts @@ -0,0 +1,106 @@ +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, it, vi } from "vitest"; +import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; +import { runSelfHostMigrations } from "../../src/selfhost/migrate"; + +function sqliteDb(): { db: ReturnType; raw: DatabaseSync } { + const raw = new DatabaseSync(":memory:"); + return { db: createD1Adapter(nodeSqliteDriver(raw as never)), raw }; +} + +function migrationDir(files: Record): string { + const dir = mkdtempSync(join(tmpdir(), "gtmig-atomic-")); + for (const [name, sql] of Object.entries(files)) writeFileSync(join(dir, name), sql); + return dir; +} + +// #9027: each statement ran individually and the file was recorded applied only after the LAST one succeeded, +// so a crash mid-file re-ran the WHOLE file next boot. Harmless for pure DDL; actively destructive for the +// several migrations carrying real DML (the table-rebuild INSERT ... SELECT pattern, plus UPDATE/DELETE steps), +// where a re-run either double-inserts or raises a PK conflict that — not matching the tolerated "already +// exists" shape — throws and bricks boot. With deploys currently ending in SIGKILL under load (#9007), a +// mid-migration kill is not hypothetical. +describe("self-host migrations are all-or-nothing per file (#9027)", () => { + it("rolls the whole file back when a later statement fails, leaving no partial DDL and no ledger row", async () => { + const { db, raw } = sqliteDb(); + const dir = migrationDir({ + "0001_partial.sql": "CREATE TABLE keep (id INTEGER);\nCREATE TABLE dropme (id INTEGER);\nTHIS IS NOT VALID SQL;", + }); + + await expect(runSelfHostMigrations(db, dir)).rejects.toThrow(); + + const tables = raw.prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as Array<{ name: string }>; + // Neither table survives: before this fix `keep` and `dropme` were both left behind, and the file was NOT + // recorded — so the next boot re-ran it from the top against a schema that had already half-changed. + expect(tables.map((t) => t.name)).not.toContain("dropme"); + expect(tables.map((t) => t.name)).not.toContain("keep"); + }); + + it("does not double-apply DML when a file is retried after a crash", async () => { + const { db, raw } = sqliteDb(); + raw.exec("CREATE TABLE src (id INTEGER PRIMARY KEY); INSERT INTO src (id) VALUES (1), (2);"); + // The table-rebuild shape several real migrations use, with a failing tail standing in for the crash. + const dir = migrationDir({ + "0001_rebuild.sql": "CREATE TABLE dst (id INTEGER PRIMARY KEY);\nINSERT INTO dst (id) SELECT id FROM src;\nBROKEN;", + }); + + await expect(runSelfHostMigrations(db, dir)).rejects.toThrow(); + + const dstExists = (raw.prepare("SELECT COUNT(*) AS c FROM sqlite_master WHERE type='table' AND name='dst'").get() as { c: number }).c; + // The INSERT ... SELECT was rolled back with everything else, so a retry starts from a clean slate rather + // than inserting the same rows a second time. + expect(dstExists).toBe(0); + }); + + it("records the ledger row inside the same transaction, so 'applied' and 'actually applied' cannot diverge", async () => { + const { db, raw } = sqliteDb(); + const dir = migrationDir({ "0001_ok.sql": "CREATE TABLE ok (id INTEGER);" }); + + expect(await runSelfHostMigrations(db, dir)).toBe(1); + const ledger = raw.prepare("SELECT name FROM _selfhost_migrations").all() as Array<{ name: string }>; + expect(ledger.map((row) => row.name)).toEqual(["0001_ok.sql"]); + expect(await runSelfHostMigrations(db, dir)).toBe(0); + }); + + it("still heals a drifted database through the per-statement fallback, and says so", async () => { + const { db, raw } = sqliteDb(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const dir = migrationDir({ + "0001_add_x.sql": "CREATE TABLE t (id INTEGER);\nALTER TABLE t ADD COLUMN x INTEGER;", + // A renumbered-migration collision re-adds the same column: the atomic attempt fails, and the tolerant + // path must still record the file rather than crash-looping the boot. + "0002_readd_x.sql": "ALTER TABLE t ADD COLUMN x INTEGER;", + }); + + expect(await runSelfHostMigrations(db, dir)).toBe(2); + expect(warn.mock.calls.some(([line]) => String(line).includes("selfhost_migration_transaction_failed"))).toBe(true); + const ledger = raw.prepare("SELECT COUNT(*) AS c FROM _selfhost_migrations").get() as { c: number }; + expect(ledger.c).toBe(2); + vi.restoreAllMocks(); + }); + + it("applies a file whose final statement has no trailing semicolon", async () => { + const { db, raw } = sqliteDb(); + // The ledger INSERT is concatenated onto the file's statements, so an unterminated tail would otherwise + // fuse onto it and fail to parse. + const dir = migrationDir({ "0001_tail.sql": "CREATE TABLE tail (id INTEGER)" }); + + expect(await runSelfHostMigrations(db, dir)).toBe(1); + const tables = raw.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='tail'").all(); + expect(tables).toHaveLength(1); + }); + + it("falls back to the pre-existing path when the adapter has no transactional surface (real D1)", async () => { + const { db, raw } = sqliteDb(); + const dir = migrationDir({ "0001_plain.sql": "CREATE TABLE plain (id INTEGER);" }); + // Real Cloudflare D1 exposes no multi-statement transaction; migrations must still apply there. + const withoutTransaction = { ...(db as unknown as Record) }; + delete withoutTransaction.execTransaction; + + expect(await runSelfHostMigrations(withoutTransaction as never, dir)).toBe(1); + expect(raw.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='plain'").all()).toHaveLength(1); + }); +}); diff --git a/test/unit/selfhost-pg-adapter.test.ts b/test/unit/selfhost-pg-adapter.test.ts index ba9ef0a858..c5d3b35978 100644 --- a/test/unit/selfhost-pg-adapter.test.ts +++ b/test/unit/selfhost-pg-adapter.test.ts @@ -110,3 +110,58 @@ describe("createPgAdapter (#977 self-host D1-over-Postgres)", () => { expect(await createPgAdapter(makeMockPool([])).dump()).toBeInstanceOf(ArrayBuffer); }); }); + +// #9027: migrations need all-or-nothing execution per file, and `exec` cannot give it — each `exec` call takes +// whatever connection the pool hands out, so a BEGIN and its COMMIT issued as separate calls can land on +// different connections and mean nothing. This surface exists to make the transaction a single session. +describe("execTransaction runs a script as one transaction (#9027)", () => { + it("wraps the script in BEGIN/COMMIT on a single dedicated client", async () => { + const pool = makeMockPool(); + const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise }; + + await db.execTransaction("CREATE TABLE t (id INTEGER);"); + + expect(pool.queries.map((q) => q.sql)).toEqual(["BEGIN", "CREATE TABLE t (id INTEGER);", "COMMIT"]); + }); + + it("rolls back and rethrows when the script fails", async () => { + const queries: string[] = []; + let released = 0; + const client = { + async query(sql: string) { + queries.push(String(sql)); + if (String(sql).includes("BROKEN")) throw new Error("syntax error"); + return { rows: [], rowCount: 0 }; + }, + release() { + released += 1; + }, + }; + const pool = { async connect() { return client as unknown as PoolClient; } } as unknown as Pool; + const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise }; + + await expect(db.execTransaction("BROKEN;")).rejects.toThrow("syntax error"); + expect(queries).toEqual(["BEGIN", "BROKEN;", "ROLLBACK"]); + // A failed transaction must never be returned to the pool still open. + expect(released).toBe(1); + }); + + it("still releases the client and reports the ORIGINAL error when the rollback itself fails", async () => { + let released = 0; + const client = { + async query(sql: string) { + if (String(sql) === "BEGIN") return { rows: [], rowCount: 0 }; + throw new Error(String(sql) === "ROLLBACK" ? "connection lost" : "syntax error"); + }, + release() { + released += 1; + }, + }; + const pool = { async connect() { return client as unknown as PoolClient; } } as unknown as Pool; + const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise }; + + // The rollback failure must not mask why the migration actually failed. + await expect(db.execTransaction("BROKEN;")).rejects.toThrow("syntax error"); + expect(released).toBe(1); + }); +}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 9d605ab154..137d8360d5 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -281,7 +281,66 @@ describe("createPgQueue (durable #977)", () => { const q = createPgQueue(m.pool, async () => undefined); await q.init(); expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("CREATE TABLE IF NOT EXISTS _selfhost_jobs")); - expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("status='processing'")); + expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("status='processing'"), expect.anything()); + }); + + // #9023: boot recovery used to re-pend EVERY processing row unconditionally. On Postgres the queue is + // deliberately multi-instance, so during any overlapped or rolling deploy "every processing row" includes + // jobs a SIBLING is running right now -- both processes then ran the same PR pass, producing duplicate gate + // check-runs and verdict thrash. Only a row whose lease has actually expired can belong to a dead process. + describe("boot recovery respects the processing lease (#9023)", () => { + async function initWithTimeout(timeoutMs: string | undefined, fn: ReturnType): Promise { + const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + if (timeoutMs === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = timeoutMs; + try { + await createPgQueue({ query: fn } as unknown as Pool, async () => undefined).init(); + } finally { + if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; + } + } + + it("scopes the recovery scan to rows older than the lease, not to every processing row", async () => { + const seen: Array<{ sql: string; params: unknown[] }> = []; + const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { + seen.push({ sql: String(sql), params: params ?? [] }); + return { rows: [], rowCount: 0 }; + }); + await initWithTimeout("60000", fn); + + const scan = seen.find((call) => call.sql.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")); + expect(scan).toBeDefined(); + // The cutoff is "now minus the lease", the same definition the runtime reaper uses -- boot and + // steady-state must not disagree about what counts as abandoned. + expect(Number(scan?.params[0])).toBeLessThanOrEqual(Date.now() - 60000); + }); + + it("re-pends only under a compare-and-swap, so a sibling finishing first cannot be resurrected", async () => { + const seen: string[] = []; + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + const q = String(sql); + seen.push(q); + if (q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) { + return { rows: [{ id: "7", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null }], rowCount: 1 }; + } + return { rows: [], rowCount: 0 }; + }); + await initWithTimeout("60000", fn); + + expect(seen.some((q) => q.includes("SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'"))).toBe(true); + }); + + it("recovers nothing when the lease is disabled — with no lease, nothing can be PROVEN abandoned", async () => { + const seen: string[] = []; + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + seen.push(String(sql)); + return { rows: [], rowCount: 0 }; + }); + await initWithTimeout("0", fn); + + expect(seen.some((q) => q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1"))).toBe(false); + }); }); it("init() handles null rowCount from the recovery query (rowCount ?? 0 nullish arm)", async () => { @@ -2943,18 +3002,195 @@ describe("createPgQueue (durable #977)", () => { expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("connection terminated unexpectedly"))).toBe(true); }); + // #9023: a reclaim never charged an attempt, so a job that reliably wedges past its lease -- an infinite + // loop, a hang against an unresponsive dependency -- was requeued forever with `attempts` frozen at whatever + // the last real failure left it. It could never reach the dead-letter threshold that exists precisely to stop + // a poison job from burning the queue. + describe("reclaimed jobs consume their retry budget (#9023)", () => { + async function drainWithProcessingRow(row: Record): Promise { + const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1"; + const statements: string[] = []; + try { + // Boot recovery now shares the reaper's lease-scoped scan (#9023), so the row is only offered AFTER + // init() -- otherwise this would exercise boot recovery, not the runtime reclaim. + let booted = false; + let served = false; + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + const q = String(sql); + statements.push(q); + if (booted && q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) { + if (served) return { rows: [], rowCount: 0 }; + served = true; + return { rows: [row], rowCount: 1 }; + } + return { rows: [], rowCount: 1 }; + }); + const q = createPgQueue({ query: fn } as unknown as Pool, async () => undefined); + await q.init(); + booted = true; + statements.length = 0; + await q.drain(); + } finally { + if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; + } + return statements; + } + + it("requeues with attempts incremented while the job still has budget", async () => { + const statements = await drainWithProcessingRow({ id: "5", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 0 }); + expect(statements.some((q) => q.includes("SET status='pending', run_after=$1, attempts=$2"))).toBe(true); + }); + + it("dead-letters a job that has wedged past its lease as many times as the retry cap allows", async () => { + const statements = await drainWithProcessingRow({ id: "6", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 4 }); + expect(statements.some((q) => q.includes("SET status='dead', attempts=$1") && q.includes("AND status='processing'"))).toBe(true); + // It must NOT also be requeued -- that would be the forever-loop this fixes, just one attempt later. + expect(statements.some((q) => q.includes("SET status='pending', run_after=$1, attempts=$2"))).toBe(false); + }); + }); + + // #9023: `run_after` doubles as the lease timestamp, so without a heartbeat the lease timeout was a ceiling on + // job DURATION rather than on liveness -- a legitimately long pass (a large PR, a slow AI review, a + // rate-limited GitHub window) got reclaimed and double-run by a sibling purely for taking too long. The + // activeJobIds guard only covers THIS process's in-memory set, which is the wrong scope for a shared queue. + it("heartbeats the lease of a job it is actively running (#9023)", async () => { + const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "3000"; // → a 1s heartbeat interval + vi.useFakeTimers(); + try { + const m = makePool(); + m.enqueueJob("9", { type: "agent-regate-pr" }, 0); + let release: (() => void) | undefined; + const running = new Promise((resolve) => { + release = resolve; + }); + const q = createPgQueue(m.pool, async () => running); + await q.init(); + const drained = q.drain(); + await vi.advanceTimersByTimeAsync(2500); + + const poolCalls = (): unknown[][] => (m.fn as unknown as ReturnType).mock.calls; + const heartbeats = poolCalls().filter((call: unknown[]) => String(call[0]).includes("SET run_after=$1 WHERE id=$2 AND status='processing'")); + expect(heartbeats.length).toBeGreaterThanOrEqual(1); + + release?.(); + await vi.advanceTimersByTimeAsync(10); + await drained; + // The timer must be cleared on completion, or a finished job keeps renewing a lease it no longer holds. + const after = poolCalls().length; + await vi.advanceTimersByTimeAsync(5000); + expect(poolCalls().length).toBe(after); + } finally { + vi.useRealTimers(); + if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; + } + }); + + // The pg driver can return a null rowCount for some UPDATE results; the recovery/reclaim counters must read + // that as "nothing changed" rather than crashing or over-reporting. + describe("#9023 recovery paths tolerate a null rowCount", () => { + async function runWith(fn: ReturnType, afterInit: (q: ReturnType) => Promise): Promise { + const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1"; + try { + const q = createPgQueue({ query: fn } as unknown as Pool, async () => undefined); + await q.init(); + await afterInit(q); + } finally { + if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; + } + } + + it("counts nothing recovered at boot when the compare-and-swap reports a null rowCount", async () => { + const logs: string[] = []; + vi.spyOn(console, "log").mockImplementation((...args: unknown[]) => void logs.push(String(args[0]))); + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + const q = String(sql); + if (q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) { + return { rows: [{ id: "1", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 0 }], rowCount: 1 }; + } + if (q.includes("SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'")) return { rows: [], rowCount: null }; + return { rows: [], rowCount: 1 }; + }); + await runWith(fn, async () => undefined); + + expect(logs.map((line) => JSON.parse(line) as Record).some((log) => log.event === "selfhost_queue_recovered")).toBe(false); + vi.restoreAllMocks(); + }); + + it("does not report a dead-letter it did not actually win", async () => { + let booted = false; + let served = false; + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + const q = String(sql); + if (booted && q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) { + if (served) return { rows: [], rowCount: 0 }; + served = true; + return { rows: [{ id: "2", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 4 }], rowCount: 1 }; + } + // A sibling won the row first, so the guarded UPDATE matched nothing. + if (q.includes("SET status='dead', attempts=$1")) return { rows: [], rowCount: null }; + return { rows: [], rowCount: 1 }; + }); + await runWith(fn, async (q) => { + booted = true; + await q.drain(); + }); + expect(served).toBe(true); + }); + }); + + it("#9023 heartbeat failures are absorbed — a lost renewal just lets the lease age normally", async () => { + const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "3000"; + vi.useFakeTimers(); + try { + const m = makePool(); + m.enqueueJob("10", { type: "agent-regate-pr" }, 0); + let release: (() => void) | undefined; + const running = new Promise((resolve) => { + release = resolve; + }); + const original = m.pool.query.bind(m.pool); + (m.pool as unknown as { query: (sql: string, params?: unknown[]) => Promise }).query = async (sql: string, params?: unknown[]) => { + if (String(sql).includes("SET run_after=$1 WHERE id=$2 AND status='processing'")) throw new Error("connection lost"); + return original(sql, params as never); + }; + const q = createPgQueue(m.pool, async () => running); + await q.init(); + const drained = q.drain(); + await vi.advanceTimersByTimeAsync(2500); + release?.(); + await vi.advanceTimersByTimeAsync(10); + // The job still completes: a heartbeat is an optimization, never a precondition. + await expect(drained).resolves.toBeUndefined(); + } finally { + vi.useRealTimers(); + if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; + } + }); + it("pump absorbs a reclaimExpiredProcessingJobs() pool failure instead of crashing the process (regression for #2498)", async () => { const oldTimeout = process.env.QUEUE_PROCESSING_TIMEOUT_MS; process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1"; try { + // #9023 made boot recovery use the SAME lease-scoped scan as the reaper, so the injection has to start + // AFTER init() -- otherwise this exercises a failing boot, not the pump absorbing a reaper failure. + let failReaper = false; const fn = vi.fn().mockImplementation(async (sql: unknown) => { - if (String(sql).includes("WHERE status='processing' AND run_after<=$1")) throw new Error("connection terminated unexpectedly"); + if (failReaper && String(sql).includes("WHERE status='processing' AND run_after<=$1")) throw new Error("connection terminated unexpectedly"); return { rows: [], rowCount: 0 }; }); const pool = { query: fn } as unknown as Pool; const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); const q = createPgQueue(pool, async () => undefined); await q.init(); + failReaper = true; await expect(q.drain()).resolves.toBeUndefined();