From bc900680c06890ca18234f2571d7e286a4207b79 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 01:09:47 -0700 Subject: [PATCH 1/4] fix(selfhost): add bounded age escape for the backlog-vs-fresh lane priority gate The lane-scoped foreground claim requires beating the best due unclassified priority, but githubWebhookPriority returns 10 (equal to fresh's own priority, above backlog's 9) for nearly every webhook other than a fresh PR open/reopen/ synchronize/ready-for-review event. On any repo with CI at least one such row is essentially always due, so neither lane's priority can ever satisfy the strict `>` gate and both lane claims permanently fall back to plain priority ordering -- the exact starvation the fairness mechanism exists to prevent. Add shouldEscapeLanePriorityGate: once the oldest due row in the lane being claimed has waited at least DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS (10 minutes, mirroring maintenance-admission's trickle_max_defer_age), the gate is bypassed so an old lane row can win on its own merits. Closes #9153 --- src/selfhost/queue-fairness.ts | 28 ++++++++++++++++++++++ test/unit/selfhost-queue-fairness.test.ts | 29 +++++++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/src/selfhost/queue-fairness.ts b/src/selfhost/queue-fairness.ts index 6a5d2022e9..0e21b16fc7 100644 --- a/src/selfhost/queue-fairness.ts +++ b/src/selfhost/queue-fairness.ts @@ -46,6 +46,34 @@ export function foregroundLaneForJob(type: string, payload: string): ForegroundL } } +// #9153: claimNextForegroundLane requires the lane-scoped claim to beat the best DUE unclassified foreground +// priority (`candidate.priority > maxDueUnclassifiedForegroundPriority()`), so that manual/repair work the +// classifier deliberately leaves lane `null` doesn't get shut out by a perpetually non-empty classified lane +// (see the module header above). But `githubWebhookPriority` returns 10 -- equal to fresh's own priority, and +// ABOVE backlog's 9 -- for every webhook OTHER than a fresh pull_request open/reopen/synchronize/ready-for- +// review event (check_suite.completed, check_run.completed, issue_comment, review events, ...), so on any repo +// with CI at least one such row is essentially always due. Neither lane's priority can then ever satisfy a +// strict `>` against 10, and BOTH lane claims permanently fall back to the plain unscoped claim -- exactly the +// starvation this fairness mechanism exists to prevent. This bounded age escape mirrors maintenance-admission's +// trickle_max_defer_age: once the OLDEST due row in the lane being claimed THIS cycle has waited at least +// `maxStarveAgeMs`, the unclassified-priority gate is bypassed (the caller falls back to the same `>=` floor +// the unscoped claim itself uses, scoped to the lane), so an old lane row can win on its own merits instead of +// being blocked forever by an unrelated row's priority. +export const DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS = 10 * 60_000; // 10 minutes + +/** + * Whether the lane-priority gate (`candidate.priority > maxDueUnclassifiedForegroundPriority()`) should be + * bypassed this claim, given the age of the OLDEST due row in the lane currently being claimed. + * `oldestDueLaneAgeMs` is `null` when that lane currently has no due candidate at all (nothing to escape for -- + * the lane-scoped claim will find no rows regardless of predicate). Pure. + */ +export function shouldEscapeLanePriorityGate( + oldestDueLaneAgeMs: number | null, + maxStarveAgeMs: number = DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS, +): boolean { + return oldestDueLaneAgeMs !== null && oldestDueLaneAgeMs >= maxStarveAgeMs; +} + export type ForegroundLaneRatio = { backlogPer: number; freshPer: number }; // 3 backlog claims for every 1 fresh claim (suggested by the operator report, not a fixed law): heavily favors diff --git a/test/unit/selfhost-queue-fairness.test.ts b/test/unit/selfhost-queue-fairness.test.ts index 6a2195fcc8..9ac7fb43f4 100644 --- a/test/unit/selfhost-queue-fairness.test.ts +++ b/test/unit/selfhost-queue-fairness.test.ts @@ -1,10 +1,12 @@ import { describe, expect, it } from "vitest"; import { + DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS, DEFAULT_FOREGROUND_LANE_RATIO, backlogRepoCandidatesFromJobKeys, foregroundLaneForJob, nextForegroundLane, pickBacklogRepo, + shouldEscapeLanePriorityGate, } from "../../src/selfhost/queue-fairness"; function webhookPayload(eventName: string, action?: string): string { @@ -170,6 +172,33 @@ describe("pickBacklogRepo (#selfhost-backlog-convergence)", () => { }); }); +describe("shouldEscapeLanePriorityGate (#9153)", () => { + it("does not escape when the lane has no due candidate at all", () => { + expect(shouldEscapeLanePriorityGate(null)).toBe(false); + }); + + it("does not escape when the oldest due lane row is younger than the starve threshold", () => { + expect(shouldEscapeLanePriorityGate(DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS - 1)).toBe(false); + }); + + it("escapes once the oldest due lane row has waited exactly the starve threshold", () => { + expect(shouldEscapeLanePriorityGate(DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS)).toBe(true); + }); + + it("escapes when the oldest due lane row has waited well past the starve threshold", () => { + expect(shouldEscapeLanePriorityGate(DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS + 60_000)).toBe(true); + }); + + it("honors a custom maxStarveAgeMs instead of the default", () => { + expect(shouldEscapeLanePriorityGate(5_000, 10_000)).toBe(false); + expect(shouldEscapeLanePriorityGate(10_000, 10_000)).toBe(true); + }); + + it("the exported default matches the documented 10-minute bound", () => { + expect(DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS).toBe(10 * 60_000); + }); +}); + describe("backlogRepoCandidatesFromJobKeys (#selfhost-backlog-convergence)", () => { it("extracts the repo and oldest pending age from backlog-lane job keys", () => { const candidates = backlogRepoCandidatesFromJobKeys( From 56fb416917eb505e81f5b7cb0493704adafc7bbc Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 01:11:29 -0700 Subject: [PATCH 2/4] fix(selfhost): apply the lane-priority age escape in both queue backends Wires shouldEscapeLanePriorityGate into claimNextForegroundLane for both the Postgres and sqlite queue backends: each computes the oldest due row's age for the lane being claimed (fresh via a dedicated query, backlog by reusing the row set already fetched for repo selection) and falls back to the plain `>=` floor once that age crosses the bounded threshold, instead of staying permanently gated behind an unclassified priority-10 webhook. Includes regression tests for both backends covering the escape arm (an aged backlog row wins preferentially) and the control arm (a fresh backlog row stays gated this cycle), plus a fix to a pre-existing pg-queue test whose placeholder created_at (an epoch-adjacent 1000ms) incidentally tripped the new age escape. Closes #9153 --- src/selfhost/pg-queue.ts | 61 ++++++++-- src/selfhost/sqlite-queue.ts | 52 ++++++++- test/unit/selfhost-pg-queue.test.ts | 143 +++++++++++++++++++++++- test/unit/selfhost-sqlite-queue.test.ts | 55 +++++++++ 4 files changed, 294 insertions(+), 17 deletions(-) diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 67fd208a80..64c45613c0 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -153,6 +153,7 @@ import { foregroundLaneForJob, nextForegroundLane, pickBacklogRepo, + shouldEscapeLanePriorityGate, type BacklogRepoCount, type ForegroundLane, } from "./queue-fairness"; @@ -1124,12 +1125,15 @@ export function createPgQueue( * this cycle," never "no foreground work at all." One slot per fairness window is deliberately left unscoped, * and lane-scoped claims must beat the best unclassified foreground priority, so manual/repair work the * classifier intentionally leaves as lane `null` keeps its plain priority ordering instead of sitting behind a - * perpetually non-empty classified lane. The fairness singleton's claim_sequence always advances (best-effort, - * hit or miss) so the ratio cycle keeps progressing even through empty cycles. Sequence - * allocation is a single atomic UPDATE ... RETURNING (not a separate SELECT-then-UPDATE): this backend is - * the multi-instance one (multiple app instances can share one Postgres, see the file header), so two - * concurrent callers reading the same pre-increment value would both compute the SAME lane and defeat the - * bounded-ratio guarantee -- the row's own lock serializes concurrent allocations instead. */ + * perpetually non-empty classified lane -- UNLESS that gate has starved the lane past + * shouldEscapeLanePriorityGate's bounded age (#9153: an unclassified priority-10 webhook is essentially always + * due on a repo with CI, which would otherwise make the `>` gate permanently unsatisfiable and this whole + * mechanism inert; see queue-fairness.ts's module comment on that constant). The fairness singleton's + * claim_sequence always advances (best-effort, hit or miss) so the ratio cycle keeps progressing even through + * empty cycles. Sequence allocation is a single atomic UPDATE ... RETURNING (not a separate SELECT-then- + * UPDATE): this backend is the multi-instance one (multiple app instances can share one Postgres, see the file + * header), so two concurrent callers reading the same pre-increment value would both compute the SAME lane + * and defeat the bounded-ratio guarantee -- the row's own lock serializes concurrent allocations instead. */ async function claimNextForegroundLane(now: number): Promise { const fairnessRes = await pool.query( `UPDATE ${FAIRNESS_TABLE} SET claim_sequence=claim_sequence+1 WHERE id='singleton' RETURNING claim_sequence, last_backlog_repo`, @@ -1140,10 +1144,12 @@ export function createPgQueue( const lane: ForegroundLane = nextForegroundLane(sequence); if (sequence % (fairnessWindow + 1) === fairnessWindow) return null; const unclassifiedPriority = await maxDueUnclassifiedForegroundPriority(now); - const lanePriorityPredicate = - unclassifiedPriority === null ? "candidate.priority >= $2" : "candidate.priority > $2"; - const lanePriorityFloor = unclassifiedPriority ?? FOREGROUND_QUEUE_PRIORITY_FLOOR; if (lane === "fresh") { + const oldestDueLaneAgeMs = await oldestDueForegroundLaneAgeMs(now, "fresh"); + const { predicate: lanePriorityPredicate, floor: lanePriorityFloor } = lanePriorityGate( + unclassifiedPriority, + oldestDueLaneAgeMs, + ); const freshRow = await claimNextWhere(now, lanePriorityPredicate, { sql: "candidate.foreground_lane='fresh'", params: [] }, lanePriorityFloor); if (freshRow) incr("loopover_jobs_claimed_by_lane_total", { lane: "fresh" }); return freshRow; @@ -1152,8 +1158,17 @@ export function createPgQueue( `SELECT job_key, created_at FROM ${TABLE} WHERE status='pending' AND run_after<=$1 AND foreground_lane='backlog'`, [now], ); + const backlogRows = backlogRes.rows as Array<{ job_key: string | null; created_at: number | string }>; + // Oldest due backlog row's age, computed straight from the rows just fetched above -- no extra query needed + // (unlike the fresh lane, which has no equivalent pre-fetched row set to reuse). + const oldestDueLaneAgeMs = + backlogRows.length === 0 ? null : Math.max(...backlogRows.map((row) => now - Number(row.created_at))); + const { predicate: lanePriorityPredicate, floor: lanePriorityFloor } = lanePriorityGate( + unclassifiedPriority, + oldestDueLaneAgeMs, + ); const candidates = backlogRepoCandidatesFromJobKeys( - (backlogRes.rows as Array<{ job_key: string | null; created_at: number | string }>).map((row) => ({ + backlogRows.map((row) => ({ jobKey: row.job_key, createdAtMs: Number(row.created_at), })), @@ -1172,6 +1187,20 @@ export function createPgQueue( return row; } + /** The lane-scoped claim's priority predicate + floor for this cycle: strictly beat `unclassifiedPriority` + * (when there is a due unclassified row) unless the age escape (#9153) fires, in which case fall back to the + * same `>=` floor the unscoped claim itself uses -- letting the lane's own priority ordering (still scoped to + * this lane via claimNextWhere's `extra` predicate) decide instead of being blocked outright. */ + function lanePriorityGate( + unclassifiedPriority: number | null, + oldestDueLaneAgeMs: number | null, + ): { predicate: string; floor: number } { + if (unclassifiedPriority === null || shouldEscapeLanePriorityGate(oldestDueLaneAgeMs)) { + return { predicate: "candidate.priority >= $2", floor: FOREGROUND_QUEUE_PRIORITY_FLOOR }; + } + return { predicate: "candidate.priority > $2", floor: unclassifiedPriority }; + } + async function maxDueUnclassifiedForegroundPriority(now: number): Promise { const res = await pool.query( `SELECT MAX(priority) AS priority FROM ${TABLE} WHERE status='pending' AND run_after<=$1 AND priority>=$2 AND foreground_lane IS NULL`, @@ -1181,6 +1210,18 @@ export function createPgQueue( return row?.priority === null || row?.priority === undefined ? null : Number(row.priority); } + /** Age (ms) of the oldest DUE row currently in the given classified lane -- null when the lane has no due + * candidate at all right now. Only needed for the fresh lane; the backlog lane derives this directly from + * the row set claimNextForegroundLane already fetches for repo selection, without a separate query. */ + async function oldestDueForegroundLaneAgeMs(now: number, lane: "fresh" | "backlog"): Promise { + const res = await pool.query( + `SELECT MIN(created_at) AS oldest FROM ${TABLE} WHERE status='pending' AND run_after<=$1 AND foreground_lane=$2`, + [now, lane], + ); + const row = res.rows[0] as { oldest: number | string | null } | undefined; + return row?.oldest != null ? now - Number(row.oldest) : null; + } + async function claimNextWhere( now: number, priorityPredicate: string, diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index bc203e66b6..485aa86b62 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -66,6 +66,7 @@ import { foregroundLaneForJob, nextForegroundLane, pickBacklogRepo, + shouldEscapeLanePriorityGate, type BacklogRepoCount, type ForegroundLane, } from "./queue-fairness"; @@ -697,8 +698,11 @@ export function createSqliteQueue( * cycle," never "no foreground work at all." One slot per fairness window is deliberately left unscoped, and * lane-scoped claims must beat the best unclassified foreground priority, so manual/repair work the classifier * intentionally leaves as lane `null` keeps its plain priority ordering instead of sitting behind a perpetually - * non-empty classified lane. The fairness singleton's claim_sequence always advances (best-effort, hit or - * miss) so the ratio cycle keeps progressing even through empty cycles. */ + * non-empty classified lane -- UNLESS that gate has starved the lane past shouldEscapeLanePriorityGate's + * bounded age (#9153: an unclassified priority-10 webhook is essentially always due on a repo with CI, which + * would otherwise make the `>` gate permanently unsatisfiable and this whole mechanism inert; see + * queue-fairness.ts's module comment on that constant). The fairness singleton's claim_sequence always + * advances (best-effort, hit or miss) so the ratio cycle keeps progressing even through empty cycles. */ function claimNextForegroundLane(now: number): JobRow | null { const fairness = driver.query( `SELECT claim_sequence, last_backlog_repo FROM ${FAIRNESS_TABLE} WHERE id='singleton'`, @@ -710,10 +714,12 @@ export function createSqliteQueue( driver.query(`UPDATE ${FAIRNESS_TABLE} SET claim_sequence=claim_sequence+1 WHERE id='singleton'`, []); if (sequence % (fairnessWindow + 1) === fairnessWindow) return null; const unclassifiedPriority = maxDueUnclassifiedForegroundPriority(now); - const lanePriorityPredicate = - unclassifiedPriority === null ? "candidate.priority>=?" : "candidate.priority>?"; - const lanePriorityFloor = unclassifiedPriority ?? FOREGROUND_QUEUE_PRIORITY_FLOOR; if (lane === "fresh") { + const oldestDueLaneAgeMs = oldestDueForegroundLaneAgeMs(now, "fresh"); + const { predicate: lanePriorityPredicate, floor: lanePriorityFloor } = lanePriorityGate( + unclassifiedPriority, + oldestDueLaneAgeMs, + ); const freshRow = claimNextWhere(now, lanePriorityPredicate, { sql: "candidate.foreground_lane='fresh'", params: [] }, lanePriorityFloor); if (freshRow) incr("loopover_jobs_claimed_by_lane_total", { lane: "fresh" }); return freshRow; @@ -722,8 +728,17 @@ export function createSqliteQueue( `SELECT job_key, created_at FROM ${TABLE} WHERE status='pending' AND run_after<=? AND foreground_lane='backlog'`, [now], ); + const backlogRowsTyped = backlogRows as Array<{ job_key: string | null; created_at: number }>; + // Oldest due backlog row's age, computed straight from the rows just fetched above -- no extra query needed + // (unlike the fresh lane, which has no equivalent pre-fetched row set to reuse). + const oldestDueLaneAgeMs = + backlogRowsTyped.length === 0 ? null : Math.max(...backlogRowsTyped.map((row) => now - Number(row.created_at))); + const { predicate: lanePriorityPredicate, floor: lanePriorityFloor } = lanePriorityGate( + unclassifiedPriority, + oldestDueLaneAgeMs, + ); const candidates = backlogRepoCandidatesFromJobKeys( - (backlogRows as Array<{ job_key: string | null; created_at: number }>).map((row) => ({ + backlogRowsTyped.map((row) => ({ jobKey: row.job_key, createdAtMs: Number(row.created_at), })), @@ -742,6 +757,20 @@ export function createSqliteQueue( return row; } + /** The lane-scoped claim's priority predicate + floor for this cycle: strictly beat `unclassifiedPriority` + * (when there is a due unclassified row) unless the age escape (#9153) fires, in which case fall back to the + * same `>=` floor the unscoped claim itself uses -- letting the lane's own priority ordering (still scoped to + * this lane via claimNextWhere's `extra` predicate) decide instead of being blocked outright. */ + function lanePriorityGate( + unclassifiedPriority: number | null, + oldestDueLaneAgeMs: number | null, + ): { predicate: string; floor: number } { + if (unclassifiedPriority === null || shouldEscapeLanePriorityGate(oldestDueLaneAgeMs)) { + return { predicate: "candidate.priority>=?", floor: FOREGROUND_QUEUE_PRIORITY_FLOOR }; + } + return { predicate: "candidate.priority>?", floor: unclassifiedPriority }; + } + function maxDueUnclassifiedForegroundPriority(now: number): number | null { const row = driver.query( `SELECT MAX(priority) AS priority FROM ${TABLE} WHERE status='pending' AND run_after<=? AND priority>=? AND foreground_lane IS NULL`, @@ -750,6 +779,17 @@ export function createSqliteQueue( return row?.priority === null || row?.priority === undefined ? null : Number(row.priority); } + /** Age (ms) of the oldest DUE row currently in the given classified lane -- null when the lane has no due + * candidate at all right now. Only needed for the fresh lane; the backlog lane derives this directly from + * the row set claimNextForegroundLane already fetches for repo selection, without a separate query. */ + function oldestDueForegroundLaneAgeMs(now: number, lane: "fresh" | "backlog"): number | null { + const row = driver.query( + `SELECT MIN(created_at) AS oldest FROM ${TABLE} WHERE status='pending' AND run_after<=? AND foreground_lane=?`, + [now, lane], + ).rows[0] as { oldest: number | null } | undefined; + return row?.oldest != null ? now - Number(row.oldest) : null; + } + /** Top-N repos by backlog-convergence pending DEPTH, for the observability dashboard's per-repo backlog panel * (#selfhost-lane-observability) -- a snapshot read, distinct from claimNextForegroundLane's own backlog * query (which is scoped to run_after<=now and only reads job_key+created_at for the round-robin picker). diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index e373a72dac..c24a666ae6 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { Pool, QueryResult } from "pg"; import { createPgQueue, setPgRetryPoolQueryDelayMsForTest } from "../../src/selfhost/pg-queue"; import { queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; +import { DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS } from "../../src/selfhost/queue-fairness"; import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; import { RetryableJobError } from "../../src/queue/retryable"; import { hostLoadAvg1PerCore } from "../../src/selfhost/host-pressure"; @@ -75,6 +76,16 @@ interface MockPool { * on a specific row (rowCount 0) while another succeeds (rowCount 1). */ setReviveUpdateRowCounts(rowCounts: number[]): void; setRateLimitRows(rows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }>): void; + /** #9153: the best DUE, UNCLASSIFIED foreground priority claimNextForegroundLane gates the lane-scoped claim + * against (maxDueUnclassifiedForegroundPriority's own query) — null (the default) means no due unclassified + * row exists, matching every pre-existing test that doesn't care about this gate. */ + setMaxDueUnclassifiedForegroundPriority(value: number | null): void; + /** #9153: the raw (job_key, created_at) rows claimNextForegroundLane's backlog branch reads to pick a repo + * AND (post-fix) to compute the oldest due backlog row's age for the starvation escape. Empty by default. */ + setBacklogLaneRows(rows: Array<{ job_key: string | null; created_at: number }>): void; + /** #9153: the created_at of the oldest DUE fresh-lane row, backing oldestDueForegroundLaneAgeMs("fresh"). + * Null (the default) means no due fresh row exists. */ + setOldestDueFreshLaneCreatedAt(value: number | null): void; /** Configures the four maintenance-admission pressure aggregate queries (live + maintenance + backlog- * convergence + fresh-intake lane). Defaults to zero pending / null oldest in all lanes (pressure clear) * until set. `runnableCnt`/`oldestRunnable` back the #selfhost-queue-liveness runnable-now split (see @@ -130,6 +141,9 @@ function makePool(): MockPool { let pressureMaintenance: { cnt: number; oldest: number | null } = { cnt: 0, oldest: null }; let pressureBacklogConvergence: { cnt: number } = { cnt: 0 }; let pressureFreshIntake: { cnt: number } = { cnt: 0 }; + let unclassifiedForegroundPriority: number | null = null; + let backlogLaneRows: Array<{ job_key: string | null; created_at: number }> = []; + let oldestDueFreshLaneCreatedAt: number | null = null; let foregroundLivenessOldestCandidates: Array<{ id: string; created_at: number; payload?: string; deferred_by?: string | null }> = []; let foregroundLivenessNewestCandidates: Array<{ id: string; created_at: number; payload?: string; deferred_by?: string | null }> = []; const foregroundLivenessUpdateRowCounts: number[] = []; @@ -192,6 +206,21 @@ function makePool(): MockPool { if (q.includes("AS cnt") && q.includes("foreground_lane='fresh'")) { return { rows: [{ cnt: String(pressureFreshIntake.cnt) }], rowCount: 1 }; } + // #9153: maxDueUnclassifiedForegroundPriority's own query (no "AS cnt" — distinguishes it from the + // pressure-signal aggregates above). + if (q.includes("SELECT MAX(priority) AS priority FROM")) { + return { rows: [{ priority: unclassifiedForegroundPriority }], rowCount: 1 }; + } + // #9153: claimNextForegroundLane's backlog-repo-candidate row read (job_key + created_at, unaggregated — + // distinguishes it from the "AS cnt"-bearing backlog-convergence pressure count above). + if (q.includes("SELECT job_key, created_at FROM") && q.includes("foreground_lane='backlog'")) { + return { rows: backlogLaneRows, rowCount: backlogLaneRows.length }; + } + // #9153: oldestDueForegroundLaneAgeMs's own query (parameterized foreground_lane=$2, only ever called with + // "fresh" in practice — the backlog lane derives its age from the row read above instead). + if (q.includes("SELECT MIN(created_at) AS oldest FROM") && q.includes("foreground_lane=$2")) { + return { rows: [{ oldest: oldestDueFreshLaneCreatedAt }], rowCount: 1 }; + } if (q.includes("FROM github_rate_limit_observations")) { const admissionKey = typeof params?.[0] === "string" ? params[0] : null; const newest = (rows: typeof rateLimitRows) => @@ -249,6 +278,15 @@ function makePool(): MockPool { reviveUpdateRowCounts.length = 0; reviveUpdateRowCounts.push(...rowCounts); }, + setMaxDueUnclassifiedForegroundPriority(value) { + unclassifiedForegroundPriority = value; + }, + setBacklogLaneRows(rows) { + backlogLaneRows = rows; + }, + setOldestDueFreshLaneCreatedAt(value) { + oldestDueFreshLaneCreatedAt = value; + }, setPressureSignals(signals) { if (signals.live) pressureLive = signals.live; if (signals.maintenance) pressureMaintenance = signals.maintenance; @@ -1316,6 +1354,11 @@ describe("createPgQueue (durable #977)", () => { prNumber: 1, installationId: 1, }; + // #9153: created_at must be RECENT (well under DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS), not the + // epoch-adjacent placeholder used elsewhere in this file -- an ancient created_at here would trip the + // bounded age escape and switch the predicate away from the `> 99` this test asserts on, which is not + // the scenario under test (that's covered by the dedicated #9153 regression tests above). + const recentCreatedAt = Date.now() - 60_000; const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { const q = String(sql); if (q.includes("UPDATE _selfhost_queue_fairness SET claim_sequence=claim_sequence+1") && q.includes("RETURNING claim_sequence, last_backlog_repo")) { @@ -1325,7 +1368,7 @@ describe("createPgQueue (durable #977)", () => { return { rows: [{ priority: 99 }], rowCount: 1 }; } if (q.includes("SELECT job_key, created_at") && q.includes("foreground_lane='backlog'")) { - return { rows: [{ job_key: "agent-regate-pr:owner/repo#2", created_at: 1000 }], rowCount: 1 }; + return { rows: [{ job_key: "agent-regate-pr:owner/repo#2", created_at: recentCreatedAt }], rowCount: 1 }; } if (q.includes("UPDATE _selfhost_jobs SET status='processing'")) { claimSql.push(q); @@ -1357,6 +1400,104 @@ describe("createPgQueue (durable #977)", () => { expect(await renderMetrics()).not.toContain("loopover_jobs_claimed_by_lane_total"); }); + it("REGRESSION (#9153): a continuous stream of unclassified priority-10 webhooks cannot starve an aged backlog row past the bounded escape", async () => { + const claimSql: string[] = []; + let claimed = false; + const now = Date.now(); + const staleAgeMs = DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS + 60_000; + const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { + const q = String(sql); + if (q.includes("UPDATE _selfhost_queue_fairness SET claim_sequence=claim_sequence+1") && q.includes("RETURNING claim_sequence, last_backlog_repo")) { + return { rows: [{ claim_sequence: 0, last_backlog_repo: null }], rowCount: 1 }; // sequence 0 -> backlog lane + } + // An unclassified priority-10 row is ALWAYS due -- e.g. a check_suite.completed webhook (#9153's + // trigger): githubWebhookPriority returns 10 for every webhook OTHER than a fresh pull_request + // open/reopen/synchronize/ready_for_review event. + if (q.includes("SELECT MAX(priority) AS priority") && q.includes("foreground_lane IS NULL")) { + return { rows: [{ priority: 10 }], rowCount: 1 }; + } + if (q.includes("SELECT job_key, created_at") && q.includes("foreground_lane='backlog'")) { + return { rows: [{ job_key: "agent-regate-pr:owner/repo#1", created_at: now - staleAgeMs }], rowCount: 1 }; + } + if (q.includes("UPDATE _selfhost_jobs SET status='processing'")) { + claimSql.push(q); + if (!claimed && q.includes("foreground_lane='backlog'")) { + claimed = true; + // Escaped: predicate falls back to `>=` against the plain floor, NOT `>` against the due + // unclassified priority (10) that a priority-9 backlog row could never satisfy. + expect(q).toContain("candidate.priority >= $2"); + expect((params as unknown[])[1]).not.toBe(10); + return { + rows: [{ id: "backlog-1", payload: JSON.stringify(backlogJob("owner/repo", 1)), attempts: 0, job_key: "agent-regate-pr:owner/repo#1", priority: 9, created_at: now - staleAgeMs }], + rowCount: 1, + }; + } + return { rows: [], rowCount: 0 }; + } + return { rows: [], rowCount: 0 }; + }); + const seen: string[] = []; + const q = createPgQueue({ query: fn } as unknown as Pool, async (m) => void seen.push(typeOf(m))); + + await q.init(); + await q.drain(); + + expect(seen).toEqual(["agent-regate-pr"]); + expect(claimSql[0]).toContain("foreground_lane='backlog'"); + expect(await renderMetrics()).toContain('loopover_jobs_claimed_by_lane_total{lane="backlog"} 1'); + }); + + it("REGRESSION (#9153, control arm): a FRESH backlog row (under the starve threshold) stays gated behind a due unclassified priority-10 row this cycle", async () => { + const claimSql: string[] = []; + let unscopedClaimed = false; + const now = Date.now(); + const freshAgeMs = DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS - 60_000; + const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { + const q = String(sql); + if (q.includes("UPDATE _selfhost_queue_fairness SET claim_sequence=claim_sequence+1") && q.includes("RETURNING claim_sequence, last_backlog_repo")) { + return { rows: [{ claim_sequence: 0, last_backlog_repo: null }], rowCount: 1 }; + } + if (q.includes("SELECT MAX(priority) AS priority") && q.includes("foreground_lane IS NULL")) { + return { rows: [{ priority: 10 }], rowCount: 1 }; + } + if (q.includes("SELECT job_key, created_at") && q.includes("foreground_lane='backlog'")) { + return { rows: [{ job_key: "agent-regate-pr:owner/repo#1", created_at: now - freshAgeMs }], rowCount: 1 }; + } + if (q.includes("UPDATE _selfhost_jobs SET status='processing'")) { + claimSql.push(q); + if (q.includes("foreground_lane='backlog'")) { + // NOT escaped: predicate stays `>` against the due unclassified priority (10), which a priority-9 + // backlog row can never satisfy -- the exact unsatisfiable gate #9153 reports. + expect(q).toContain("candidate.priority > $2"); + expect((params as unknown[])[1]).toBe(10); + return { rows: [], rowCount: 0 }; + } + if (!unscopedClaimed && !q.includes("foreground_lane")) { + unscopedClaimed = true; + return { + rows: [{ id: "backlog-1", payload: JSON.stringify(backlogJob("owner/repo", 1)), attempts: 0, job_key: "agent-regate-pr:owner/repo#1", priority: 9, created_at: now - freshAgeMs }], + rowCount: 1, + }; + } + return { rows: [], rowCount: 0 }; + } + return { rows: [], rowCount: 0 }; + }); + const seen: string[] = []; + const q = createPgQueue({ query: fn } as unknown as Pool, async (m) => void seen.push(typeOf(m))); + + await q.init(); + await q.drain(); + + // The job still eventually runs (via the unscoped fallback claim) -- #9153 is a FAIRNESS bug, not job + // loss -- but it was never PREFERRED by the lane-scoped claim, exactly the inert-mechanism symptom #9153 + // reports. + expect(seen).toEqual(["agent-regate-pr"]); + expect(claimSql[0]).toContain("foreground_lane='backlog'"); + expect(claimSql[1]).not.toContain("foreground_lane"); + expect(await renderMetrics()).not.toContain("loopover_jobs_claimed_by_lane_total"); + }); + it("records the fresh-intake lane-claim counter on a successful fresh-lane claim (#selfhost-lane-observability)", async () => { let claimed = false; // one-shot: the row is only claimable until the first successful claim const fn = vi.fn().mockImplementation(async (sql: unknown) => { diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index cd9ad0fa54..96057d25a4 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; import { createSqliteQueue } from "../../src/selfhost/sqlite-queue"; import { jobCoalesceKey, queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; +import { DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS } from "../../src/selfhost/queue-fairness"; import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; import { RetryableJobError } from "../../src/queue/retryable"; import { hostLoadAvg1PerCore } from "../../src/selfhost/host-pressure"; @@ -1518,6 +1519,60 @@ describe("createSqliteQueue (durable #980)", () => { expect(seen).toEqual(["manual-regate:owner/repo#1:operator", "backlog-convergence:owner/repo#2"]); }); + it("REGRESSION (#9153): a continuous stream of unclassified priority-10 webhooks cannot starve an aged backlog row past the bounded escape", async () => { + const driver = makeDriver(); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push((m as unknown as { deliveryId?: string; type: string }).deliveryId ?? typeOf(m)), { concurrency: 1 }); + // A backlog-convergence row old enough to have crossed the bounded starve-age escape. Inserted directly + // (not via binding.send) and BEFORE the webhook below, so both rows already exist in the table before + // drain() ever claims anything -- send()'s own kickOne() would otherwise race a claim against this insert. + const staleCreatedAt = Date.now() - (DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS + 60_000); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog')", + [JSON.stringify(backlogJob("owner/repo", 1)), staleCreatedAt, "agent-regate-pr:owner/repo#1"], + ); + // An unclassified priority-10 webhook, always due -- e.g. check_suite.completed, #9153's trigger scenario + // (githubWebhookPriority returns 10 for every webhook OTHER than a fresh pull_request open/reopen/ + // synchronize/ready_for_review event). Also inserted directly, for the same race-free reason as above. + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane) VALUES (?, 'pending', 0, 0, ?, 10, NULL, NULL)", + [JSON.stringify(ciWebhook("ci-1")), Date.now()], + ); + await q.drain(); + // The stale backlog row is claimed FIRST (preferentially, via the lane-scoped claim) despite the + // perpetually-due unclassified webhook -- before the fix, the lane-scoped claim's `priority > 10` gate + // could never be satisfied by a priority-9 backlog row, so this preference was permanently inert. + expect(seen).toEqual(["backlog-convergence:owner/repo#1", "ci-1"]); + expect(await renderMetrics()).toContain('loopover_jobs_claimed_by_lane_total{lane="backlog"} 1'); + }); + + it("REGRESSION (#9153, control arm): a FRESH backlog row (under the starve threshold) is not preferentially claimed over a due unclassified priority-10 webhook", async () => { + const driver = makeDriver(); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push((m as unknown as { deliveryId?: string; type: string }).deliveryId ?? typeOf(m)), { concurrency: 1 }); + // Both rows inserted directly (not via binding.send) BEFORE drain() ever runs -- see the escape-arm test + // above for why send()'s own kickOne() would otherwise race a claim against these inserts. + const freshCreatedAt = Date.now() - (DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS - 60_000); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog')", + [JSON.stringify(backlogJob("owner/repo", 1)), freshCreatedAt, "agent-regate-pr:owner/repo#1"], + ); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane) VALUES (?, 'pending', 0, 0, ?, 10, NULL, NULL)", + [JSON.stringify(ciWebhook("ci-1")), Date.now()], + ); + await q.drain(); + // The unclassified priority-10 webhook wins the very first claim (the lane-scoped claim's `priority > 10` + // gate is still unsatisfiable, so it falls through to the plain unscoped priority-DESC ordering) -- the + // backlog row still runs eventually, just never PREFERRED, exactly the inert-mechanism symptom #9153 + // reports. Once the webhook is consumed there's no due unclassified row left AT ALL, so the SECOND cycle's + // lane-scoped claim legitimately succeeds on its own merits (unclassifiedPriority is genuinely null by + // then, not merely escaped) -- this is real end-to-end driver state, unlike a mocked pool that could hold + // the unclassified signal artificially constant across both cycles. + expect(seen).toEqual(["ci-1", "backlog-convergence:owner/repo#1"]); + expect(await renderMetrics()).toContain('loopover_jobs_claimed_by_lane_total{lane="backlog"} 1'); + }); + it("falls through to the plain unscoped foreground claim when the preferred lane has nothing pending", async () => { const driver = makeDriver(); const seen: string[] = []; From ec9d40d1a22585f41f2314d3232944e3f392a23e Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 01:11:39 -0700 Subject: [PATCH 3/4] fix(selfhost): filter repair-exhausted PRs before the backlog-convergence cap selectBacklogConvergenceCandidates sorted oldest-open-first, sliced to `max`, and only then did the caller filter out repair-exhausted PRs -- so five old, permanently-unpublishable PRs could occupy every slot forever, shadowing the 6th+ PR needing convergence behind them indefinitely. Split the pure ordering out into sortedBacklogConvergenceCandidates (unsliced) and have sweepRepoBacklogConvergence walk that full order, skipping exhausted candidates as it goes and stopping once `max` actionable ones are found -- selectBacklogConvergenceCandidates itself becomes a thin slice-after-sort wrapper over the new function for callers that don't need the exhaustion filter. Also logs + records an audit event when every examined candidate is repair-exhausted, so a permanently wedged head of the backlog is visible instead of the sweep silently returning early every cycle. Closes #9154 --- src/queue/processors.ts | 47 ++++++- src/selfhost/backlog-convergence.ts | 42 ++++-- test/unit/queue-2.test.ts | 131 ++++++++++++++++++ .../unit/selfhost-backlog-convergence.test.ts | 36 +++++ 4 files changed, 237 insertions(+), 19 deletions(-) diff --git a/src/queue/processors.ts b/src/queue/processors.ts index a53b1b8c74..996dc8e36e 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -258,7 +258,7 @@ import { isRegateSweepDraining, selectRegateCandidates, } from "../settings/agent-sweep"; -import { selectBacklogConvergenceCandidates } from "../selfhost/backlog-convergence"; +import { BACKLOG_CONVERGENCE_SWEEP_MAX_PRS, sortedBacklogConvergenceCandidates } from "../selfhost/backlog-convergence"; import { LOW_REST_RATE_LIMIT_REMAINING, MAINTENANCE_RESERVED_HEADROOM, @@ -1851,16 +1851,53 @@ export async function sweepRepoBacklogConvergence( const sweepInstallationId = repo?.installationId ?? null; if (sweepInstallationId == null) return; const openPullRequests = await listOpenPullRequests(env, repoFullName); - const allCandidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests }); + // #9154: the exhaustion filter must run BEFORE the `max` cap, not after -- capping first (the previous + // shape: selectBacklogConvergenceCandidates already sliced to 5, THEN filtered) let five permanently + // repair-exhausted old PRs occupy every slot forever, shadowing every other PR needing convergence behind + // them. Walk the FULL oldest-open-first order and skip exhausted candidates as we go, stopping as soon as + // `max` actionable ones are found -- bounded to however far into the backlog we actually need to look, not + // the whole list, in the common case where most of the head is actionable. + // // #orb-retry-storm (backlog-convergence half): needsSurfaceConvergence re-fires on the exact same // lastPublishedSurfaceSha-mismatch signal as the main sweep's outage-repair priority path, but this sweeper // had no memory of prior attempts at all -- a PR whose gate-check finalize kept failing silently got a fresh // full re-review dispatched every ~30 minutes indefinitely. Share the same per-SHA attempt budget as the main // sweep (isRegateRepairExhausted) rather than adding an independent cap, since both sweeps competing for the // same stuck PR would otherwise double the wasted spend the cap exists to prevent. - const exhaustedFlags = await Promise.all(allCandidates.map((pr) => isRegateRepairExhausted(env, repoFullName, pr))); - const candidates = allCandidates.filter((_pr, index) => !exhaustedFlags[index]); - if (candidates.length === 0) return; + const orderedCandidates = sortedBacklogConvergenceCandidates(openPullRequests); + const candidates: PullRequestRecord[] = []; + let examinedCount = 0; + for (const pr of orderedCandidates) { + if (candidates.length >= BACKLOG_CONVERGENCE_SWEEP_MAX_PRS) break; + examinedCount += 1; + if (await isRegateRepairExhausted(env, repoFullName, pr)) continue; + candidates.push(pr); + } + if (candidates.length === 0) { + // Every candidate examined was repair-exhausted -- the sweep is returning early with nothing to show for + // it. Previously silent (the caller just saw an empty array and bailed); log + audit so a permanently + // wedged head of the backlog is visible instead of masquerading as "nothing needs convergence". + if (orderedCandidates.length > 0) { + console.warn( + JSON.stringify({ + level: "warn", + event: "backlog_convergence_sweep_all_exhausted", + repository: repoFullName, + examined: examinedCount, + totalCandidates: orderedCandidates.length, + }), + ); + await recordAuditEvent(env, { + eventType: "agent.sweep.backlog_convergence", + actor: "loopover", + targetKey: repoFullName, + outcome: "denied", + detail: `backlog-convergence sweep found ${orderedCandidates.length} open PR(s) needing convergence, all repair-exhausted`, + metadata: { repoFullName, examined: examinedCount, totalCandidates: orderedCandidates.length }, + }); + } + return; + } // Stamp the backlog-convergence draining marker for EVERY candidate NOW, at dispatch — not in the downstream // per-PR job (#4502, mirrors #audit-sweep-dispatch-stamp). This makes getLatestBacklogConvergenceRegatedAt // reflect this sweep immediately, so fanOutBacklogConvergenceSweepJobs's in-flight guard skips re-arming this diff --git a/src/selfhost/backlog-convergence.ts b/src/selfhost/backlog-convergence.ts index b975bfe4e3..75eda24509 100644 --- a/src/selfhost/backlog-convergence.ts +++ b/src/selfhost/backlog-convergence.ts @@ -32,25 +32,39 @@ export function needsSurfaceConvergence(pr: Pick same ordered batch. + * Every open PR a repo's backlog-convergence sweep considers, in the order it should serve them: drop drafts + * and anything whose surface is already published at the current head, then order oldest-open-first (oldest + * `createdAt` first, falling back to the epoch so a PR with no known creation time still sorts + * deterministically rather than being silently dropped) — this is the explicit "oldest open PRs first" + * fairness ordering the backlog-drain lane depends on (see queue-fairness.ts, PR2). Ties broken by PR number. + * Deliberately UNSLICED (#9154): selectBacklogConvergenceCandidates below applies the sweep's `max` cap + * directly to this order, which shadows every PR behind the first `max` whenever any of THOSE are + * permanently repair-exhausted (a caller that also needs to skip exhausted candidates -- see + * sweepRepoBacklogConvergence in processors.ts -- must walk this full order and exclude exhausted PRs BEFORE + * capping, not after). Pure + deterministic: same inputs -> same ordered list. */ -export function selectBacklogConvergenceCandidates(input: { - pulls: PullRequestRecord[]; - max?: number; -}): PullRequestRecord[] { - const max = input.max ?? BACKLOG_CONVERGENCE_SWEEP_MAX_PRS; +export function sortedBacklogConvergenceCandidates(pulls: PullRequestRecord[]): PullRequestRecord[] { const ageKey = (pr: PullRequestRecord): number => { const created = pr.createdAt ? Date.parse(pr.createdAt) : Number.NaN; return Number.isFinite(created) ? created : 0; }; - return input.pulls + return pulls .filter((pr) => pr.state === "open" && !pr.isDraft) .filter((pr) => needsSurfaceConvergence(pr)) - .sort((a, b) => ageKey(a) - ageKey(b) || a.number - b.number) - .slice(0, Math.max(0, max)); + .sort((a, b) => ageKey(a) - ageKey(b) || a.number - b.number); +} + +/** + * Select the open PRs a single repo's backlog-convergence sweep should re-enqueue: the first `max` PRs (by + * default BACKLOG_CONVERGENCE_SWEEP_MAX_PRS) from sortedBacklogConvergenceCandidates' full oldest-open-first + * order. This convenience wrapper caps WITHOUT regard to repair-exhaustion — fine for a caller that doesn't + * need that filter, but see sortedBacklogConvergenceCandidates' doc comment for why sweepRepoBacklogConvergence + * itself calls that function directly instead. Pure + deterministic. + */ +export function selectBacklogConvergenceCandidates(input: { + pulls: PullRequestRecord[]; + max?: number; +}): PullRequestRecord[] { + const max = input.max ?? BACKLOG_CONVERGENCE_SWEEP_MAX_PRS; + return sortedBacklogConvergenceCandidates(input.pulls).slice(0, Math.max(0, max)); } diff --git a/test/unit/queue-2.test.ts b/test/unit/queue-2.test.ts index 766baa9340..05d026a56c 100644 --- a/test/unit/queue-2.test.ts +++ b/test/unit/queue-2.test.ts @@ -2409,6 +2409,137 @@ describe("queue processors", () => { } }, 60_000); + it("REGRESSION (#9154): five permanently repair-exhausted old PRs cannot occupy every cap slot and shadow an actionable PR behind them", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertInstallation(env, { action: "created", installation: { id: 9415, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo6", full_name: "owner/agent-repo6", private: false, owner: { login: "owner" } }, 9415); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo6", autonomy: { merge: "auto" } }); + await upsertRepoFocusManifest(env, "owner/agent-repo6", { settings: { checkRunMode: "off", commentMode: "off", publicSurface: "off", reviewCheckMode: "required" } }); + + // Five OLD PRs -- exactly BACKLOG_CONVERGENCE_SWEEP_MAX_PRS worth -- each already exhausted its shared + // repair-attempt budget for its current head SHA (#orb-retry-storm's isRegateRepairExhausted). Before the + // #9154 fix, selectBacklogConvergenceCandidates sliced to these 5 FIRST (oldest-open-first), and the + // exhaustion filter only ran afterward -- so these five permanently wedged PRs occupied every slot the cap + // allows, forever, and the sweep bailed with an empty candidate list every cycle. + const wedgedCreated = [ + "2026-05-01T00:00:00.000Z", + "2026-05-02T00:00:00.000Z", + "2026-05-03T00:00:00.000Z", + "2026-05-04T00:00:00.000Z", + "2026-05-05T00:00:00.000Z", + ]; + for (let i = 0; i < 5; i += 1) { + const number = i + 1; + const headSha = `wedged-${number}`; + await upsertPullRequestFromGitHub(env, "owner/agent-repo6", { + number, + title: `Wedged ${number}`, + state: "open", + user: { login: "c" }, + head: { sha: headSha }, + labels: [], + body: "", + created_at: wedgedCreated[i]!, + updated_at: wedgedCreated[i]!, + }); + const targetKey = `owner/agent-repo6#${number}#${headSha}`; + for (let attempt = 0; attempt < 5; attempt += 1) { + await repositoriesModule.recordAuditEvent(env, { + eventType: "agent.sweep.regate.repair_attempt", + actor: "loopover", + targetKey, + outcome: "queued", + detail: "prior attempt", + metadata: {}, + }); + } + } + // One NEWER, genuinely actionable PR (never attempted before) — sorts LAST (newest createdAt), directly + // behind the five wedged PRs above: exactly the position the pre-fix bug shadowed forever. + await upsertPullRequestFromGitHub(env, "owner/agent-repo6", { + number: 6, + title: "Actionable", + state: "open", + user: { login: "c" }, + head: { sha: "fresh-6" }, + labels: [], + body: "", + created_at: "2026-05-10T00:00:00.000Z", + updated_at: "2026-05-10T00:00:00.000Z", + }); + vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + const warned = vi.spyOn(console, "warn").mockImplementation(() => undefined); + + try { + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo6" }); + + const fanned = sent.filter((job): job is Extract => job.type === "agent-regate-pr"); + // The actionable PR (#6) IS dispatched — it must not be shadowed by the five wedged PRs ahead of it. + expect(fanned.some((job) => job.deliveryId === "backlog-convergence:owner/agent-repo6#6")).toBe(true); + // None of the five wedged (exhausted) PRs are dispatched. + for (let number = 1; number <= 5; number += 1) { + expect(fanned.some((job) => job.deliveryId === `backlog-convergence:owner/agent-repo6#${number}`)).toBe(false); + } + // A sweep that DOES find an actionable candidate must not log the all-exhausted signal. + expect(warned.mock.calls.some(([line]) => typeof line === "string" && line.includes("backlog_convergence_sweep_all_exhausted"))).toBe(false); + } finally { + errors.mockRestore(); + warned.mockRestore(); + } + }, 60_000); + + it("REGRESSION (#9154): logs + audits when EVERY backlog-convergence candidate is repair-exhausted, instead of silently returning nothing", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertInstallation(env, { action: "created", installation: { id: 9416, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo7", full_name: "owner/agent-repo7", private: false, owner: { login: "owner" } }, 9416); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo7", autonomy: { merge: "auto" } }); + await upsertRepoFocusManifest(env, "owner/agent-repo7", { settings: { checkRunMode: "off", commentMode: "off", publicSurface: "off", reviewCheckMode: "required" } }); + const headSha = "wedged-only"; + await upsertPullRequestFromGitHub(env, "owner/agent-repo7", { + number: 1, + title: "Only wedged PR", + state: "open", + user: { login: "c" }, + head: { sha: headSha }, + labels: [], + body: "", + }); + const targetKey = `owner/agent-repo7#1#${headSha}`; + for (let attempt = 0; attempt < 5; attempt += 1) { + await repositoriesModule.recordAuditEvent(env, { + eventType: "agent.sweep.regate.repair_attempt", + actor: "loopover", + targetKey, + outcome: "queued", + detail: "prior attempt", + metadata: {}, + }); + } + vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + const warned = vi.spyOn(console, "warn").mockImplementation(() => undefined); + + try { + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo7" }); + + const fanned = sent.filter((job) => job.type === "agent-regate-pr"); + expect(fanned).toHaveLength(0); + expect(warned.mock.calls.some(([line]) => typeof line === "string" && line.includes("backlog_convergence_sweep_all_exhausted"))).toBe(true); + const skipped = await env.DB.prepare( + "select count(*) as n from audit_events where event_type = ? and target_key = ? and outcome = 'denied'", + ) + .bind("agent.sweep.backlog_convergence", "owner/agent-repo7") + .first<{ n: number }>(); + expect(skipped?.n).toBe(1); + } finally { + errors.mockRestore(); + warned.mockRestore(); + } + }, 60_000); + it("a fresh backlog-convergence candidate under the cap is dispatched with repairHeadSha set, and executing it charges the SAME shared attempt budget the main sweep's repair path reads", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); diff --git a/test/unit/selfhost-backlog-convergence.test.ts b/test/unit/selfhost-backlog-convergence.test.ts index 8648108dd8..2e33cccbc8 100644 --- a/test/unit/selfhost-backlog-convergence.test.ts +++ b/test/unit/selfhost-backlog-convergence.test.ts @@ -3,6 +3,7 @@ import { BACKLOG_CONVERGENCE_SWEEP_MAX_PRS, needsSurfaceConvergence, selectBacklogConvergenceCandidates, + sortedBacklogConvergenceCandidates, } from "../../src/selfhost/backlog-convergence"; import type { PullRequestRecord } from "../../src/types"; @@ -120,3 +121,38 @@ describe("selectBacklogConvergenceCandidates (#selfhost-backlog-convergence)", ( expect(selectBacklogConvergenceCandidates({ pulls })).toEqual([]); }); }); + +// #9154: selectBacklogConvergenceCandidates caps to `max` BEFORE any repair-exhaustion filter can run (it has +// no notion of exhaustion at all). sortedBacklogConvergenceCandidates is the unsliced order a caller that ALSO +// needs to skip exhausted candidates must walk instead, so the cap is applied only to ACTIONABLE candidates. +describe("sortedBacklogConvergenceCandidates (#9154)", () => { + it("returns every eligible PR in the same oldest-open-first order, without capping", () => { + const total = BACKLOG_CONVERGENCE_SWEEP_MAX_PRS + 3; + // PR number i+1 was created minutesAgo(i) — larger i is further in the past, i.e. OLDER. So oldest-first + // order is numbers descending: [total, total-1, ..., 1]. + const pulls = Array.from({ length: total }, (_, i) => + pr({ number: i + 1, headSha: "a", lastPublishedSurfaceSha: null, createdAt: minutesAgo(i) }), + ); + const sorted = sortedBacklogConvergenceCandidates(pulls); + expect(sorted).toHaveLength(total); + expect(sorted.map((p) => p.number)).toEqual(Array.from({ length: total }, (_, i) => total - i)); + }); + + it("applies the same drop rules as selectBacklogConvergenceCandidates (closed/draft/already-published)", () => { + const pulls = [ + pr({ number: 1, state: "closed", headSha: "a", lastPublishedSurfaceSha: null }), + pr({ number: 2, isDraft: true, headSha: "a", lastPublishedSurfaceSha: null }), + pr({ number: 3, headSha: "a", lastPublishedSurfaceSha: "a" }), + pr({ number: 4, headSha: "a", lastPublishedSurfaceSha: null }), + ]; + expect(sortedBacklogConvergenceCandidates(pulls).map((p) => p.number)).toEqual([4]); + }); + + it("selectBacklogConvergenceCandidates({max}) is exactly sortedBacklogConvergenceCandidates sliced to max", () => { + const pulls = Array.from({ length: 8 }, (_, i) => + pr({ number: i + 1, headSha: "a", lastPublishedSurfaceSha: null, createdAt: minutesAgo(i) }), + ); + const sorted = sortedBacklogConvergenceCandidates(pulls); + expect(selectBacklogConvergenceCandidates({ pulls, max: 3 })).toEqual(sorted.slice(0, 3)); + }); +}); From 45536f7ee35ede87b5da17f1bdf2b3c984d39c21 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 01:11:51 -0700 Subject: [PATCH 4/4] fix(selfhost): exclude in-flight jobs from live_job_age_high and cache pressure signals Two compounding defects in maintenance admission: - oldestLiveRunnableAgeMs was computed over rows that are either 'processing' or due-and-pending, so a normal in-flight job (a routine multi-minute AI review) alone pushed the age past maxLiveJobAgeMs for its whole duration, tripping live_job_age_high with no drain escape and collapsing the entire maintenance lane -- including the watchdog/alerter that would report an actual overload -- to the 4-hour trickle backstop. Narrow oldest_runnable's own FILTER to 'pending AND due' rows only, excluding 'processing' entirely. - Every claimed maintenance job recomputes maintenancePressureSignals' four aggregate scans, and a denial returns `true` from processOne(), so the pump loop immediately claims the next due maintenance row and repeats -- a burst of N due maintenance rows means 4N sequential scans in one tight loop, a positive feedback loop (more denials -> more scans -> higher load -> more denials). Add a short-TTL (default 1.5s, configurable via MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS) memoized wrapper shared by the admission check and the pressureSignals() observability method, in both the Postgres and sqlite backends. Also add a partial index on (is_maintenance, status) for the maintenance-lane aggregate, which previously had no supporting index. Closes #9155 --- .../src/lib/selfhost-env-reference.ts | 5 + src/selfhost/maintenance-admission.ts | 26 ++++- src/selfhost/pg-queue.ts | 48 +++++++- src/selfhost/sqlite-queue.ts | 46 ++++++-- .../selfhost-maintenance-admission.test.ts | 18 +++ test/unit/selfhost-pg-queue.test.ts | 42 +++++++ test/unit/selfhost-sqlite-queue.test.ts | 107 ++++++++++++++++++ 7 files changed, 275 insertions(+), 17 deletions(-) diff --git a/apps/loopover-ui/src/lib/selfhost-env-reference.ts b/apps/loopover-ui/src/lib/selfhost-env-reference.ts index cae60d54d3..1b60375297 100644 --- a/apps/loopover-ui/src/lib/selfhost-env-reference.ts +++ b/apps/loopover-ui/src/lib/selfhost-env-reference.ts @@ -337,6 +337,10 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "MAINTENANCE_ADMISSION_MAX_PENDING", firstReference: "src/selfhost/maintenance-admission.ts", }, + { + name: "MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS", + firstReference: "src/selfhost/maintenance-admission.ts", + }, { name: "MIGRATIONS_DIR", firstReference: "src/server.ts", @@ -729,6 +733,7 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `MAINTENANCE_ADMISSION_MAX_LIVE_AGE_MS` | `src/selfhost/maintenance-admission.ts` |", "| `MAINTENANCE_ADMISSION_MAX_LIVE_PENDING` | `src/selfhost/maintenance-admission.ts` |", "| `MAINTENANCE_ADMISSION_MAX_PENDING` | `src/selfhost/maintenance-admission.ts` |", + "| `MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS` | `src/selfhost/maintenance-admission.ts` |", "| `MIGRATIONS_DIR` | `src/server.ts` |", "| `OBSERVABILITY_SMOKE_POLL_MS` | `scripts/smoke-observability-traces.ts` |", "| `OBSERVABILITY_SMOKE_TIMEOUT_MS` | `scripts/smoke-observability-traces.ts` |", diff --git a/src/selfhost/maintenance-admission.ts b/src/selfhost/maintenance-admission.ts index df13a47fa3..2950c6313d 100644 --- a/src/selfhost/maintenance-admission.ts +++ b/src/selfhost/maintenance-admission.ts @@ -96,10 +96,13 @@ export interface MaintenancePressureSignals { * keeps finding stale work every sweep). This is the field evaluateMaintenanceAdmission's live_pending_high * check actually gates on. */ liveRunnableNowCount: number; - /** Age in ms of the oldest genuinely-active (processing, or pending AND due) foreground job -- null when - * none qualifies right now. Distinct from oldestLivePendingAgeMs, which is dominated by a job intentionally - * scheduled far in the future and says nothing about how long already-active work has sat unclaimed/running. - * This is the field evaluateMaintenanceAdmission's live_job_age_high check actually gates on. */ + /** Age in ms of the oldest DUE-AND-UNCLAIMED (status='pending', run_after<=now) foreground job -- null when + * none qualifies right now. Deliberately EXCLUDES 'processing' rows (#9155): a job actively being worked (a + * normal in-flight AI review routinely takes minutes) must not, by merely running, trip live_job_age_high + * and collapse the entire maintenance lane -- including the watchdog/alerter that would report an actual + * overload -- to the 4-hour trickle backstop. Distinct from oldestLivePendingAgeMs, which is dominated by a + * job intentionally scheduled far in the future and says nothing about how long work has sat unclaimed. This + * is the field evaluateMaintenanceAdmission's live_job_age_high check actually gates on. */ oldestLiveRunnableAgeMs: number | null; maintenancePendingCount: number; oldestMaintenancePendingAgeMs: number | null; @@ -129,6 +132,14 @@ export interface MaintenanceAdmissionConfig { deferMs: number; maxDeferAgeMs: number; maintenanceDrainAgeMs: number; + /** #9155: how long a computed MaintenancePressureSignals snapshot may be reused across successive claim + * attempts before it must be recomputed. A denied maintenance job returns `true` from processOne(), so the + * pump's drain loop immediately claims the next due maintenance row and re-evaluates admission -- without + * this, a burst of N due maintenance rows means 4N sequential aggregate scans in one tight loop (more + * denials -> more scans -> higher DB/host load -> higher hostLoadAvg1PerCore -> more denials, a positive + * feedback loop). A short TTL, not "once per drain pass": pressure is still re-measured often enough to + * react to a genuinely changing queue instead of latching a stale reading for a whole burst. */ + pressureSignalsCacheTtlMs: number; } const DEFAULT_MAX_LIVE_PENDING_COUNT = 5; @@ -143,6 +154,9 @@ const DEFAULT_MAX_BACKLOG_CONVERGENCE_PENDING_COUNT = 10; const DEFAULT_DEFER_MS = 3 * 60_000; const DEFAULT_MAX_DEFER_AGE_MS = 4 * 60 * 60_000; const DEFAULT_MAINTENANCE_DRAIN_AGE_MS = 10 * 60_000; +// #9155: within the suggested 1-2s memoization window -- long enough to collapse a burst of denials sharing +// one drain pass, short enough that a real pressure change is still visible within a couple of poll ticks. +const DEFAULT_PRESSURE_SIGNALS_CACHE_TTL_MS = 1_500; function maintenanceAdmissionEnabled(): boolean { const raw = (process.env.MAINTENANCE_ADMISSION_ENABLED ?? "").trim().toLowerCase(); @@ -195,6 +209,10 @@ export function resolveMaintenanceAdmissionConfig(): MaintenanceAdmissionConfig // Never longer than the trickle backstop itself -- a misconfigured drain age above maxDeferAgeMs would be a // no-op (the trickle would always win first), so clamp it down rather than let it silently do nothing. maintenanceDrainAgeMs: Math.min(requestedDrainAgeMs, maxDeferAgeMs), + pressureSignalsCacheTtlMs: parsePositiveIntEnv("MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS", { + min: 0, + fallback: DEFAULT_PRESSURE_SIGNALS_CACHE_TTL_MS, + }), }; } diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 64c45613c0..c6902cf509 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -207,6 +207,11 @@ CREATE INDEX IF NOT EXISTS ${TABLE}_claim ON ${TABLE}(status, priority, claim_so CREATE INDEX IF NOT EXISTS ${TABLE}_pending_job_key ON ${TABLE}(job_key, status); CREATE INDEX IF NOT EXISTS ${TABLE}_lane_claim ON ${TABLE}(status, foreground_lane, run_after); CREATE INDEX IF NOT EXISTS ${TABLE}_dead ON ${TABLE}(status, dead_at, id); +-- #9155: maintenancePressureSignals' maintenance-lane COUNT/MIN scan (WHERE is_maintenance=1 AND status IN +-- (...)) had no supporting index -- every other aggregate in that function reuses the claim index's leading +-- status column, but this one filters on is_maintenance first with no index at all. Partial (is_maintenance=1 +-- only) since the maintenance lane is a small minority of rows and non-maintenance jobs never need it. +CREATE INDEX IF NOT EXISTS ${TABLE}_maintenance_status ON ${TABLE}(is_maintenance, status) WHERE is_maintenance=1; CREATE TABLE IF NOT EXISTS ${STATS_TABLE} ( name TEXT PRIMARY KEY, value BIGINT NOT NULL DEFAULT 0 @@ -480,15 +485,18 @@ export function createPgQueue( * count, see maintenance-admission.ts). The runnable-now split is the #selfhost-queue-liveness diagnostic: * distinguishes "queue large but intentionally deferred" from "queue stuck, nothing runnable" without * manual SQL. Host load is an independent, optional signal. */ - async function maintenancePressureSignals(now: number): Promise { - // runnable_cnt/oldest_runnable count a row as genuinely active RIGHT NOW when it's either already - // 'processing' (real, in-flight resource use) or 'pending' AND due (run_after<=now) -- NOT merely present - // in the outer pending/processing set, which also includes work deliberately deferred to the future (see - // maintenance-admission.ts's MaintenancePressureSignals doc comments). + async function computeMaintenancePressureSignals(now: number): Promise { + // runnable_cnt counts a row as genuinely active RIGHT NOW when it's either already 'processing' (real, + // in-flight resource use) or 'pending' AND due (run_after<=now) -- NOT merely present in the outer + // pending/processing set, which also includes work deliberately deferred to the future. oldest_runnable is + // narrower still (#9155): it excludes 'processing' rows entirely, so a normal in-flight job (routinely + // minutes long) never by itself ages past maxLiveJobAgeMs and trips live_job_age_high -- only a due row + // that's still WAITING to be claimed counts toward that signal (see maintenance-admission.ts's + // MaintenancePressureSignals doc comments). const liveRes = await pool.query( `SELECT COUNT(*) AS cnt, MIN(created_at) AS oldest, COUNT(*) FILTER (WHERE status='processing' OR run_after<=$2) AS runnable_cnt, - MIN(created_at) FILTER (WHERE status='processing' OR run_after<=$2) AS oldest_runnable + MIN(created_at) FILTER (WHERE status='pending' AND run_after<=$2) AS oldest_runnable FROM ${TABLE} WHERE status IN ('pending','processing') AND priority>=$1`, [FOREGROUND_QUEUE_PRIORITY_FLOOR, now], ); @@ -523,6 +531,34 @@ export function createPgQueue( }; } + // #9155: memoization state for maintenancePressureSignals below -- see maintenanceAdmissionConfig's + // pressureSignalsCacheTtlMs doc comment for the feedback-loop this closes. `pendingPressureSignals` dedupes + // overlapping in-flight computations across this instance's concurrent pump() loops (up to `concurrency` + // of them), not just successive calls from the SAME loop. + let cachedPressureSignals: { value: MaintenancePressureSignals; computedAtMs: number } | null = null; + let pendingPressureSignals: Promise | null = null; + + /** Cached wrapper around computeMaintenancePressureSignals -- the actual four-aggregate-query read. Every + * caller (maintenance-admission gating in processOne, and the binding.pressureSignals() observability + * method) goes through this, so both benefit from -- and share -- the same short-TTL cache. */ + async function maintenancePressureSignals(now: number): Promise { + if ( + cachedPressureSignals && + now - cachedPressureSignals.computedAtMs < maintenanceAdmissionConfig.pressureSignalsCacheTtlMs + ) { + return cachedPressureSignals.value; + } + if (pendingPressureSignals) return pendingPressureSignals; + pendingPressureSignals = computeMaintenancePressureSignals(now); + try { + const value = await pendingPressureSignals; + cachedPressureSignals = { value, computedAtMs: now }; + return value; + } finally { + pendingPressureSignals = null; + } + } + /** Top-N repos by backlog-convergence pending DEPTH, for the observability dashboard's per-repo backlog panel * (#selfhost-lane-observability) -- a snapshot read, distinct from claimNextForegroundLane's own backlog * query (which is scoped to run_after<=now and only reads job_key+created_at for the round-robin picker). diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 485aa86b62..3f18af59da 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -119,6 +119,11 @@ const JOB_KEY_INDEX_DDL = ` CREATE INDEX IF NOT EXISTS ${TABLE}_pending_job_key ON ${TABLE}(job_key, status);`; const LANE_INDEX_DDL = ` CREATE INDEX IF NOT EXISTS ${TABLE}_lane_claim ON ${TABLE}(status, foreground_lane, run_after);`; +// #9155: mirrors pg-queue.ts's ${TABLE}_maintenance_status index -- maintenancePressureSignals' maintenance-lane +// COUNT/MIN scan (WHERE is_maintenance=1 AND status IN (...)) had no supporting index; partial (is_maintenance=1 +// only) since the maintenance lane is a small minority of rows and non-maintenance jobs never need it. +const MAINTENANCE_INDEX_DDL = ` +CREATE INDEX IF NOT EXISTS ${TABLE}_maintenance_status ON ${TABLE}(is_maintenance, status) WHERE is_maintenance=1;`; interface JobRow { id: number; @@ -245,6 +250,7 @@ export function createSqliteQueue( driver.exec(CLAIM_INDEX_DDL); driver.exec(JOB_KEY_INDEX_DDL); driver.exec(LANE_INDEX_DDL); + driver.exec(MAINTENANCE_INDEX_DDL); driver.exec(DEAD_LETTER_INDEX_DDL); driver.exec(FAIRNESS_DDL); driver.exec(`INSERT OR IGNORE INTO ${FAIRNESS_TABLE} (id, claim_sequence) VALUES ('singleton', 0)`); @@ -292,6 +298,29 @@ export function createSqliteQueue( const foregroundLivenessConfig: ForegroundLivenessConfig = resolveForegroundLivenessConfig(); const installationConcurrencyConfig = resolveInstallationConcurrencyConfig(); const installationConcurrencyTracker = new InstallationConcurrencyTracker(); + // #9155: memoization state for maintenancePressureSignals below -- same feedback-loop rationale as + // pg-queue.ts's cache (see maintenanceAdmissionConfig's pressureSignalsCacheTtlMs doc comment): a denied + // maintenance job returns `true` from processOne(), so the pump's drain loop immediately claims the next due + // maintenance row and re-evaluates admission, recomputing all four aggregate scans per denial without this. + // Unlike pg-queue.ts's version, no separate in-flight-promise dedup is needed: node:sqlite's queries are + // synchronous, so a cache check followed by a (synchronous) recompute can never interleave with another call + // the way concurrent async Postgres round-trips could. + let cachedPressureSignals: { value: MaintenancePressureSignals; computedAtMs: number } | null = null; + + /** Cached wrapper around maintenancePressureSignals -- the actual four-aggregate-query read. Every caller + * (maintenance-admission gating in processOne, and the binding.pressureSignals() observability method) goes + * through this, so both benefit from -- and share -- the same short-TTL cache. */ + function getMaintenancePressureSignals(now: number): MaintenancePressureSignals { + if ( + cachedPressureSignals && + now - cachedPressureSignals.computedAtMs < maintenanceAdmissionConfig.pressureSignalsCacheTtlMs + ) { + return cachedPressureSignals.value; + } + const value = maintenancePressureSignals(driver, now); + cachedPressureSignals = { value, computedAtMs: now }; + return value; + } // Recover jobs a crashed previous run left mid-flight → make them claimable again. const recovered = recoverProcessingJobs(driver); if (recovered) { @@ -1019,7 +1048,7 @@ export function createSqliteQueue( } if (!isForegroundJobPriority(job.priority) && isMaintenanceJobType(message.type)) { const decision = evaluateMaintenanceAdmission( - maintenancePressureSignals(driver, Date.now()), + getMaintenancePressureSignals(Date.now()), maintenanceAdmissionConfig, job.created_at, Date.now(), @@ -1414,7 +1443,7 @@ export function createSqliteQueue( reviveDeadLetterJobs, releaseStaleForegroundDeferrals, async pressureSignals() { - return maintenancePressureSignals(driver, Date.now()); + return getMaintenancePressureSignals(Date.now()); }, topBacklogRepos, listDeadLetterJobs, @@ -1514,14 +1543,17 @@ function backfillJobForegroundLanes(driver: SqliteDriver): number { * distinguishes "queue large but intentionally deferred" from "queue stuck, nothing runnable" without manual * SQL. Host load is an independent, optional signal (see host-pressure.ts). */ function maintenancePressureSignals(driver: SqliteDriver, now: number): MaintenancePressureSignals { - // runnable_cnt/oldest_runnable count a row as genuinely active RIGHT NOW when it's either already - // 'processing' (real, in-flight resource use) or 'pending' AND due (run_after<=now) -- NOT merely present - // in the outer pending/processing set, which also includes work deliberately deferred to the future (see - // maintenance-admission.ts's MaintenancePressureSignals doc comments). + // runnable_cnt counts a row as genuinely active RIGHT NOW when it's either already 'processing' (real, + // in-flight resource use) or 'pending' AND due (run_after<=now) -- NOT merely present in the outer + // pending/processing set, which also includes work deliberately deferred to the future. oldest_runnable is + // narrower still (#9155): it excludes 'processing' rows entirely, so a normal in-flight job (routinely + // minutes long) never by itself ages past maxLiveJobAgeMs and trips live_job_age_high -- only a due row + // that's still WAITING to be claimed counts toward that signal (see maintenance-admission.ts's + // MaintenancePressureSignals doc comments). const live = driver.query( `SELECT COUNT(*) as cnt, MIN(created_at) as oldest, SUM(CASE WHEN status='processing' OR run_after<=? THEN 1 ELSE 0 END) as runnable_cnt, - MIN(CASE WHEN status='processing' OR run_after<=? THEN created_at ELSE NULL END) as oldest_runnable + MIN(CASE WHEN status='pending' AND run_after<=? THEN created_at ELSE NULL END) as oldest_runnable FROM ${TABLE} WHERE status IN ('pending','processing') AND priority>=?`, [now, now, FOREGROUND_QUEUE_PRIORITY_FLOOR], ).rows[0] as { cnt: number; oldest: number | null; runnable_cnt: number | null; oldest_runnable: number | null }; diff --git a/test/unit/selfhost-maintenance-admission.test.ts b/test/unit/selfhost-maintenance-admission.test.ts index 6c3eb8514c..c67cd4b5eb 100644 --- a/test/unit/selfhost-maintenance-admission.test.ts +++ b/test/unit/selfhost-maintenance-admission.test.ts @@ -33,6 +33,7 @@ const CONFIG: MaintenanceAdmissionConfig = { deferMs: 180_000, maxDeferAgeMs: 4 * 60 * 60_000, maintenanceDrainAgeMs: 600_000, + pressureSignalsCacheTtlMs: 1_500, }; describe("isMaintenanceJobType", () => { @@ -386,6 +387,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { "MAINTENANCE_ADMISSION_DEFER_MS", "MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", "MAINTENANCE_ADMISSION_DRAIN_AGE_MS", + "MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS", ] as const; const saved: Record = {}; @@ -414,6 +416,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { deferMs: 180_000, maxDeferAgeMs: 4 * 60 * 60_000, maintenanceDrainAgeMs: 600_000, + pressureSignalsCacheTtlMs: 1_500, }); }); @@ -426,6 +429,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { process.env.MAINTENANCE_ADMISSION_DEFER_MS = "5000"; process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS = "3600000"; process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; + process.env.MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS = "2000"; const config = resolveMaintenanceAdmissionConfig(); expect(config.maxLivePendingCount).toBe(10); expect(config.maxLiveJobAgeMs).toBe(60_000); @@ -435,6 +439,20 @@ describe("resolveMaintenanceAdmissionConfig", () => { expect(config.deferMs).toBe(5_000); expect(config.maxDeferAgeMs).toBe(3_600_000); expect(config.maintenanceDrainAgeMs).toBe(60_000); + expect(config.pressureSignalsCacheTtlMs).toBe(2_000); + }); + + // #9155: the memoization TTL is independently configurable (not clamped against maxDeferAgeMs the way + // maintenanceDrainAgeMs is -- there is no equivalent "would otherwise never fire" failure mode for a cache + // TTL, so a misconfigured value just means more or fewer redundant scans, never a stuck admission decision). + it("falls back to the default pressure-signals cache TTL on an invalid value", () => { + process.env.MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS = "not-a-number"; + expect(resolveMaintenanceAdmissionConfig().pressureSignalsCacheTtlMs).toBe(1_500); + }); + + it("allows a pressure-signals cache TTL of zero (memoization effectively disabled)", () => { + process.env.MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS = "0"; + expect(resolveMaintenanceAdmissionConfig().pressureSignalsCacheTtlMs).toBe(0); }); it("clamps a drain age above the max defer age down to the max defer age", () => { diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index c24a666ae6..6eaa69bf8d 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -4077,6 +4077,48 @@ describe("createPgQueue (durable #977)", () => { expect(signals.liveRunnableNowCount).toBe(0); expect(signals.oldestLiveRunnableAgeMs).toBeNull(); }); + + // REGRESSION (#9155): oldest_runnable's own FILTER clause must exclude 'processing' rows -- a normal + // in-flight job (routinely minutes long) must never, by merely running, age into live_job_age_high. The + // mock pool doesn't apply real SQL semantics, so this asserts the query TEXT itself carries the fix rather + // than relying on a numeric outcome (see selfhost-sqlite-queue.test.ts for an end-to-end behavioral version + // against a real driver). + it("computes oldest_runnable from a filter that excludes 'processing' rows", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.pressureSignals(); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("MIN(created_at) FILTER (WHERE status='pending' AND run_after<=$2) AS oldest_runnable"), + expect.anything(), + ); + }); + + // REGRESSION (#9155): a denied maintenance job returns `true` from processOne(), so the pump's drain loop + // immediately claims the next due maintenance row and re-evaluates admission -- without memoization this + // recomputes the four pressure aggregate queries on EVERY denied job (4N scans for N due maintenance rows + // in one burst). Two maintenance jobs claimed in the same drain() pass, both denied by the SAME pressure + // reading, must share ONE computation. + it("memoizes maintenancePressureSignals across denied jobs claimed in quick succession, instead of recomputing per denial", async () => { + const m = makePool(); + m.setPressureSignals({ backlogConvergence: { cnt: 11 } }); // over the default threshold of 10 + m.enqueueResult({ rows: [], rowCount: 0 }); // empty foreground claim (job 1) + m.enqueueResult({ rows: [{ ...maintenanceRow, id: "m1" }], rowCount: 1 }); // background claim (job 1) + m.enqueueResult({ rows: [], rowCount: 0 }); // empty foreground claim (job 2) + m.enqueueResult({ rows: [{ ...maintenanceRow, id: "m2" }], rowCount: 1 }); // background claim (job 2) + const started: string[] = []; + const q = createPgQueue(m.pool, async (j) => void started.push(typeOf(j))); + await q.drain(); + + expect(started).toEqual([]); // both denied + // "AS cnt, MIN(created_at) AS oldest" matches BOTH the live and maintenance aggregate reads that make up + // ONE computeMaintenancePressureSignals computation (the mock's own pressure branch distinguishes them by + // "is_maintenance=1") -- 2 calls means ONE shared computation across both denied jobs, not 4 (one + // computation's worth per denial). + const pressureCalls = (m.fn as unknown as { mock: { calls: unknown[][] } }).mock.calls.filter( + ([sql]) => typeof sql === "string" && sql.includes("AS cnt, MIN(created_at) AS oldest"), + ); + expect(pressureCalls).toHaveLength(2); // one computation's worth, not once per denied job + }); }); describe("installation-concurrency admission (#selfhost-installation-concurrency)", () => { diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 96057d25a4..e8e960a0ba 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -3666,6 +3666,7 @@ describe("createSqliteQueue (durable #980)", () => { "MAINTENANCE_ADMISSION_DEFER_MS", "MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", "MAINTENANCE_ADMISSION_DRAIN_AGE_MS", + "MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS", ] as const; const saved: Record = {}; @@ -4048,6 +4049,112 @@ describe("createSqliteQueue (durable #980)", () => { expect(signals.oldestLiveRunnableAgeMs).toBeNull(); }); + it("REGRESSION (#9155): a normal in-flight PROCESSING job does not by itself age into oldestLiveRunnableAgeMs — only due-and-unclaimed pending rows do", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + const now = Date.now(); + // A job that has been PROCESSING for a long time -- a normal, still-running AI review (routinely takes + // minutes), not a stuck one. Before the fix this alone would dominate oldestLiveRunnableAgeMs and could + // trip live_job_age_high on its own, collapsing the whole maintenance lane. + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'processing', 0, ?, ?, 10, NULL, 0)`, + [JSON.stringify(msg("github-webhook")), now, now - 600_000], // "processing" for 10 minutes + ); + // A genuinely due-but-unclaimed row, much younger. + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, ?, ?, 10, NULL, 0)`, + [JSON.stringify(msg("github-webhook")), now - 1_000, now - 5_000], + ); + const signals = await q.pressureSignals(); + // Both rows still count toward runnable-now (processing OR due) -- that check is unchanged. + expect(signals.liveRunnableNowCount).toBe(2); + // But the AGE signal reflects only the due-and-unclaimed row (~5s), not the 10-minute-old processing row. + expect(signals.oldestLiveRunnableAgeMs).toBeGreaterThanOrEqual(5_000); + expect(signals.oldestLiveRunnableAgeMs).toBeLessThan(600_000); + }); + + it("REGRESSION (#9155): oldestLiveRunnableAgeMs is null when every runnable row is PROCESSING (none due-and-unclaimed)", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + const now = Date.now(); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'processing', 0, ?, ?, 10, NULL, 0)`, + [JSON.stringify(msg("github-webhook")), now, now - 600_000], + ); + const signals = await q.pressureSignals(); + expect(signals.liveRunnableNowCount).toBe(1); // still counted as runnable (in-flight resource use) + expect(signals.oldestLiveRunnableAgeMs).toBeNull(); // nothing WAITING, so no age signal to gate on + }); + + // REGRESSION (#9155, part 2): a denied maintenance job returns `true` from processOne(), so the pump's + // drain loop immediately claims the next due maintenance row and re-evaluates admission -- without + // memoization this recomputes the pressure aggregate queries on EVERY denied job (four query round-trips + // per denial, in one tight loop for a burst of N due maintenance rows). Two maintenance jobs denied by the + // SAME live-pending-high pressure reading in the same drain() pass must share ONE computation. + it("memoizes maintenancePressureSignals across denied jobs claimed in quick succession, instead of recomputing per denial", async () => { + const driver = makeDriver(); + const started: string[] = []; + // The queue is created FIRST (so its startup DDL creates _selfhost_jobs), and the two due maintenance + // rows are inserted DIRECTLY afterward (not via binding.send) — send()'s own kickOne() fires a background + // pump immediately, which would otherwise race a claim against these inserts and consume both jobs before + // the spy below is even installed (see the #9153 regression tests above in this file for the same race). + const q = createSqliteQueue(driver, async (m) => void started.push(typeOf(m))); + const now = Date.now(); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify(msg("build-contributor-evidence")), now], + ); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify(msg("rollup-product-usage")), now], + ); + seedLiveRows(driver, 6); // over the default live-pending threshold of 5 -- every maintenance claim denied + const querySpy = vi.spyOn(driver, "query"); + await q.drain(); + + expect(started).toEqual([]); // both denied + // The live-pressure aggregate ("AS cnt, MIN(created_at) as oldest ... SUM(CASE") is the query text unique + // to ONE computeMaintenancePressureSignals computation's live-lane read. One call means one shared + // computation across both denied jobs, not two (one computation's worth per denial). + const liveAggregateCalls = querySpy.mock.calls.filter(([sql]) => + typeof sql === "string" && sql.includes("SUM(CASE WHEN status='processing' OR run_after<=? THEN 1 ELSE 0 END) as runnable_cnt"), + ); + expect(liveAggregateCalls).toHaveLength(1); + }); + + it("recomputes maintenancePressureSignals once the cache TTL has elapsed, not indefinitely", async () => { + process.env.MAINTENANCE_ADMISSION_PRESSURE_CACHE_TTL_MS = "0"; // effectively disables memoization + const driver = makeDriver(); + const started: string[] = []; + const q = createSqliteQueue(driver, async (m) => void started.push(typeOf(m))); + const now = Date.now(); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify(msg("build-contributor-evidence")), now], + ); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify(msg("rollup-product-usage")), now], + ); + seedLiveRows(driver, 6); + const querySpy = vi.spyOn(driver, "query"); + await q.drain(); + + expect(started).toEqual([]); + const liveAggregateCalls = querySpy.mock.calls.filter(([sql]) => + typeof sql === "string" && sql.includes("SUM(CASE WHEN status='processing' OR run_after<=? THEN 1 ELSE 0 END) as runnable_cnt"), + ); + // TTL of zero: every claim attempt recomputes -- one call per denied job, not one shared computation. + expect(liveAggregateCalls).toHaveLength(2); + }); + it("backfills the is_maintenance flag on startup for jobs enqueued by an older version", async () => { const driver = nodeSqliteDriver(new DatabaseSync(":memory:") as never); driver.exec(`