diff --git a/.env.example b/.env.example index b3cca1ea12..cffc8de4a4 100644 --- a/.env.example +++ b/.env.example @@ -308,6 +308,11 @@ REDIS_URL=redis://redis:6379 # REQUIRED for the self-host review # FOREGROUND_LIVENESS_CHECK_INTERVAL_MS=60000 # sweep cadence -- deliberately NOT the 1s poll tick, so a job # # still genuinely rate-limited waits for the next sweep instead # # of busy-looping (1m) +# FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP=25 # ramp-up cap: ceiling on how many foreground jobs one sweep +# # tick releases (oldest first), so a large inherited backlog +# # drains gradually over several ticks instead of every +# # released job re-attempting GitHub at once and immediately +# # re-tripping the same rate-limit bucket it was deferred for # --- Caddy HTTPS terminator (#1203; requires --profile caddy) --- # DOMAIN=gittensory.example.com # fully-qualified domain; Caddy auto-obtains a Let's Encrypt cert diff --git a/src/selfhost/foreground-liveness.ts b/src/selfhost/foreground-liveness.ts index 320a5f43e2..0254fbd28b 100644 --- a/src/selfhost/foreground-liveness.ts +++ b/src/selfhost/foreground-liveness.ts @@ -23,11 +23,18 @@ const DEFAULT_MAX_DEFER_MS = 10 * 60_000; // 10 minutes -- long enough to not fi // (which typically resolves within DEFAULT_GITHUB_RATE_LIMIT_RETRY_MS + jitter, see queue-common.ts), short // enough that live contributor-PR-review work is never parked anywhere near the ~65-minute worst case. const DEFAULT_CHECK_INTERVAL_MS = 60_000; // 1 minute +// Ramp-up cap (#selfhost-queue-liveness): a large inherited backlog (the production incident this module +// exists for had ~190 over-deferred foreground jobs) must not release ALL of it in one sweep tick -- that +// many jobs re-attempting GitHub reads at once can immediately re-trip the same rate-limit bucket they were +// deferred for, undoing the release. Draining a couple dozen per minute clears even a large backlog within +// several minutes while never presenting GitHub with more than a bounded burst. +const DEFAULT_MAX_RELEASE_PER_SWEEP = 25; export interface ForegroundLivenessConfig { enabled: boolean; maxDeferMs: number; checkIntervalMs: number; + maxReleasePerSweep: number; } function foregroundLivenessEnabled(): boolean { @@ -43,6 +50,7 @@ export function resolveForegroundLivenessConfig(): ForegroundLivenessConfig { enabled: foregroundLivenessEnabled(), maxDeferMs: parsePositiveIntEnv("FOREGROUND_LIVENESS_MAX_DEFER_MS", { min: 60_000, fallback: DEFAULT_MAX_DEFER_MS }), checkIntervalMs: parsePositiveIntEnv("FOREGROUND_LIVENESS_CHECK_INTERVAL_MS", { min: 5_000, fallback: DEFAULT_CHECK_INTERVAL_MS }), + maxReleasePerSweep: parsePositiveIntEnv("FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP", { min: 1, fallback: DEFAULT_MAX_RELEASE_PER_SWEEP }), }; } @@ -56,3 +64,22 @@ export function resolveForegroundLivenessConfig(): ForegroundLivenessConfig { export function isForegroundDeferralStale(config: ForegroundLivenessConfig, pendingSinceMs: number, nowMs: number): boolean { return config.enabled && nowMs - pendingSinceMs >= config.maxDeferMs; } + +/** PURE ramp-up selection: given every candidate ELIGIBLE for release this sweep (already filtered by + * isForegroundDeferralStale or a live rate-limit-clear check -- this function does not itself decide + * eligibility), pick at most `maxReleasePerSweep` of them, prioritizing the OLDEST (longest-pending) first. + * When candidates already fit within the cap, every one is released (a small/moderate backlog is never + * artificially throttled) -- the cap only engages for a genuinely large backlog, gradually draining it over + * several sweep ticks instead of releasing hundreds of jobs into one instant. Ties broken by the original + * array order (stable) so behavior is deterministic given the same input. Pure. */ +export function selectForegroundDeferralsToRelease( + candidates: readonly T[], + maxReleasePerSweep: number, +): T[] { + if (candidates.length <= maxReleasePerSweep) return [...candidates]; + return [...candidates] + .map((candidate, index) => ({ candidate, index })) + .sort((a, b) => a.candidate.pendingSinceMs - b.candidate.pendingSinceMs || a.index - b.index) + .slice(0, maxReleasePerSweep) + .map(({ candidate }) => candidate); +} diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 7ca1d15e75..f5eb617e5d 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -143,6 +143,7 @@ import { import { isForegroundDeferralStale, resolveForegroundLivenessConfig, + selectForegroundDeferralsToRelease, type ForegroundLivenessConfig, } from "./foreground-liveness"; import type { JobMessage } from "../types"; @@ -516,16 +517,20 @@ export function createPgQueue( } /** See foreground-liveness.ts for the full rationale. A bounded candidate SELECT (foreground-priority, pending, - * not currently due) then a per-row conditional UPDATE, mirroring reviveEligibleDeadJobs' shape. Each - * candidate is released on EITHER of two independent conditions: it has genuinely been waiting past the - * age-based trickle ceiling (isForegroundDeferralStale, unconditional backstop), OR -- CONDITION-BASED - * recovery (#selfhost-queue-liveness VPS incident) -- re-evaluating rateLimitAdmissionDelayMs against - * CURRENT observations right now says it would be admitted immediately. The age floor alone can leave a job - * pinned to a stale reset timestamp for up to its full original delay (observed up to ~15m) even when a - * fresher, healthier observation arrived moments after it was deferred; the condition check recovers it on - * the NEXT sweep tick instead (bounded by FOREGROUND_LIVENESS_CHECK_INTERVAL_MS, default 60s) whenever the - * underlying rate-limit pressure has actually cleared, regardless of job age. Logs + records a metric ONCE - * per sweep (aggregate count), not per row, so a large release batch cannot spam the log. */ + * not currently due), an eligibility pass, a ramp-up CAP, then a per-row conditional UPDATE only for the + * capped subset -- mirroring reviveEligibleDeadJobs' shape but with the extra ramp-up step. Each candidate is + * ELIGIBLE on EITHER of two independent conditions: it has genuinely been waiting past the age-based trickle + * ceiling (isForegroundDeferralStale, unconditional backstop), OR -- CONDITION-BASED recovery + * (#selfhost-queue-liveness VPS incident) -- re-evaluating rateLimitAdmissionDelayMs against CURRENT + * observations right now says it would be admitted immediately. The age floor alone can leave a job pinned to + * a stale reset timestamp for up to its full original delay (observed up to ~15m) even when a fresher, + * healthier observation arrived moments after it was deferred; the condition check recovers it on the NEXT + * sweep tick instead (bounded by FOREGROUND_LIVENESS_CHECK_INTERVAL_MS, default 60s) whenever the underlying + * rate-limit pressure has actually cleared, regardless of job age. When more jobs are eligible than + * maxReleasePerSweep allows, selectForegroundDeferralsToRelease picks the oldest first -- a large inherited + * backlog drains gradually over several sweep ticks instead of flooding GitHub with every re-attempt at once. + * Logs + records a metric ONCE per sweep (aggregate count), not per row, so a large release batch cannot spam + * the log. */ async function releaseStaleForegroundDeferrals(): Promise { if (!foregroundLivenessConfig.enabled) return 0; const now = Date.now(); @@ -533,19 +538,25 @@ export function createPgQueue( `SELECT id, payload, created_at FROM ${TABLE} WHERE status='pending' AND priority>=$1 AND run_after>$2`, [FOREGROUND_QUEUE_PRIORITY_FLOOR, now], ); + const eligible: Array<{ id: string; pendingSinceMs: number; ageStale: boolean }> = []; + for (const row of res.rows as Array<{ id: string; payload: string; created_at: number | string }>) { + const pendingSinceMs = Number(row.created_at); + const ageStale = isForegroundDeferralStale(foregroundLivenessConfig, pendingSinceMs, now); + if (!ageStale && !(await isRateLimitAdmissionNowClear(row.payload))) continue; + eligible.push({ id: row.id, pendingSinceMs, ageStale }); + } + const toRelease = selectForegroundDeferralsToRelease(eligible, foregroundLivenessConfig.maxReleasePerSweep); let released = 0; let releasedByAge = 0; let releasedByRateLimitClear = 0; - for (const row of res.rows as Array<{ id: string; payload: string; created_at: number | string }>) { - const ageStale = isForegroundDeferralStale(foregroundLivenessConfig, Number(row.created_at), now); - if (!ageStale && !(await isRateLimitAdmissionNowClear(row.payload))) continue; + for (const candidate of toRelease) { const update = await pool.query( `UPDATE ${TABLE} SET run_after=$1 WHERE id=$2 AND status='pending' AND run_after>$1`, - [now, row.id], + [now, candidate.id], ); const rowsChanged = update.rowCount ?? 0; released += rowsChanged; - if (ageStale) releasedByAge += rowsChanged; + if (candidate.ageStale) releasedByAge += rowsChanged; else releasedByRateLimitClear += rowsChanged; } if (released) { diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 190be38f7e..1e3b14ee7f 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -59,6 +59,7 @@ import { import { isForegroundDeferralStale, resolveForegroundLivenessConfig, + selectForegroundDeferralsToRelease, type ForegroundLivenessConfig, } from "./foreground-liveness"; import type { JobMessage } from "../types"; @@ -312,16 +313,20 @@ export function createSqliteQueue( } /** See foreground-liveness.ts for the full rationale. A bounded candidate SELECT (foreground-priority, pending, - * not currently due) then a per-row conditional UPDATE, mirroring reviveEligibleDeadJobs' shape. Each - * candidate is released on EITHER of two independent conditions: it has genuinely been waiting past the - * age-based trickle ceiling (isForegroundDeferralStale, unconditional backstop), OR -- CONDITION-BASED - * recovery (#selfhost-queue-liveness VPS incident) -- re-evaluating rate-limit admission against CURRENT - * observations right now says it would be admitted immediately. The age floor alone can leave a job pinned - * to a stale reset timestamp for up to its full original delay (observed up to ~15m) even when a fresher, - * healthier observation arrived moments after it was deferred; the condition check recovers it on the NEXT - * sweep tick instead (bounded by FOREGROUND_LIVENESS_CHECK_INTERVAL_MS, default 60s) whenever the underlying - * rate-limit pressure has actually cleared, regardless of job age. Logs + records a metric ONCE per sweep - * (aggregate count), not per row, so a large release batch cannot spam the log. */ + * not currently due), an eligibility pass, a ramp-up CAP, then a per-row conditional UPDATE only for the + * capped subset -- mirroring reviveEligibleDeadJobs' shape but with the extra ramp-up step. Each candidate is + * ELIGIBLE on EITHER of two independent conditions: it has genuinely been waiting past the age-based trickle + * ceiling (isForegroundDeferralStale, unconditional backstop), OR -- CONDITION-BASED recovery + * (#selfhost-queue-liveness VPS incident) -- re-evaluating rate-limit admission against CURRENT observations + * right now says it would be admitted immediately. The age floor alone can leave a job pinned to a stale + * reset timestamp for up to its full original delay (observed up to ~15m) even when a fresher, healthier + * observation arrived moments after it was deferred; the condition check recovers it on the NEXT sweep tick + * instead (bounded by FOREGROUND_LIVENESS_CHECK_INTERVAL_MS, default 60s) whenever the underlying rate-limit + * pressure has actually cleared, regardless of job age. When more jobs are eligible than maxReleasePerSweep + * allows, selectForegroundDeferralsToRelease picks the oldest first -- a large inherited backlog drains + * gradually over several sweep ticks instead of flooding GitHub with every re-attempt at once. Logs + + * records a metric ONCE per sweep (aggregate count), not per row, so a large release batch cannot spam the + * log. */ function releaseStaleForegroundDeferrals(): number { if (!foregroundLivenessConfig.enabled) return 0; const now = Date.now(); @@ -329,18 +334,23 @@ export function createSqliteQueue( `SELECT id, payload, created_at FROM ${TABLE} WHERE status='pending' AND priority>=? AND run_after>?`, [FOREGROUND_QUEUE_PRIORITY_FLOOR, now], ); - let released = 0; - let releasedByAge = 0; - let releasedByRateLimitClear = 0; + const eligible: Array<{ id: number; pendingSinceMs: number; ageStale: boolean }> = []; for (const row of rows as Array<{ id: number; payload: string; created_at: number }>) { const ageStale = isForegroundDeferralStale(foregroundLivenessConfig, row.created_at, now); if (!ageStale && !isRateLimitAdmissionNowClear(row.payload)) continue; + eligible.push({ id: row.id, pendingSinceMs: row.created_at, ageStale }); + } + const toRelease = selectForegroundDeferralsToRelease(eligible, foregroundLivenessConfig.maxReleasePerSweep); + let released = 0; + let releasedByAge = 0; + let releasedByRateLimitClear = 0; + for (const candidate of toRelease) { const { changes } = driver.query( `UPDATE ${TABLE} SET run_after=? WHERE id=? AND status='pending' AND run_after>?`, - [now, row.id, now], + [now, candidate.id, now], ); released += changes; - if (ageStale) releasedByAge += changes; + if (candidate.ageStale) releasedByAge += changes; else releasedByRateLimitClear += changes; } if (released) { diff --git a/test/unit/selfhost-foreground-liveness.test.ts b/test/unit/selfhost-foreground-liveness.test.ts index d0f7e18067..782c7029fd 100644 --- a/test/unit/selfhost-foreground-liveness.test.ts +++ b/test/unit/selfhost-foreground-liveness.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { isForegroundDeferralStale, resolveForegroundLivenessConfig, + selectForegroundDeferralsToRelease, type ForegroundLivenessConfig, } from "../../src/selfhost/foreground-liveness"; @@ -10,6 +11,7 @@ describe("resolveForegroundLivenessConfig", () => { "FOREGROUND_LIVENESS_ENABLED", "FOREGROUND_LIVENESS_MAX_DEFER_MS", "FOREGROUND_LIVENESS_CHECK_INTERVAL_MS", + "FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP", ] as const; const saved: Record = {}; @@ -32,6 +34,7 @@ describe("resolveForegroundLivenessConfig", () => { enabled: true, maxDeferMs: 600_000, checkIntervalMs: 60_000, + maxReleasePerSweep: 25, }); }); @@ -43,6 +46,21 @@ describe("resolveForegroundLivenessConfig", () => { expect(config.checkIntervalMs).toBe(10_000); }); + it("reads a custom FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP when set (#selfhost-queue-liveness ramp-up)", () => { + process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP = "5"; + expect(resolveForegroundLivenessConfig().maxReleasePerSweep).toBe(5); + }); + + it("falls back to the default ramp-up cap when the value is non-numeric", () => { + process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP = "not-a-number"; + expect(resolveForegroundLivenessConfig().maxReleasePerSweep).toBe(25); + }); + + it("falls back to the default ramp-up cap when the value is below the min (1)", () => { + process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP = "0"; + expect(resolveForegroundLivenessConfig().maxReleasePerSweep).toBe(25); + }); + it.each(["0", "false", "off", "no"])("treats FOREGROUND_LIVENESS_ENABLED=%s as disabled", (value) => { process.env.FOREGROUND_LIVENESS_ENABLED = value; expect(resolveForegroundLivenessConfig().enabled).toBe(false); @@ -86,7 +104,7 @@ describe("resolveForegroundLivenessConfig", () => { describe("isForegroundDeferralStale", () => { const now = 1_000_000_000; - const config: ForegroundLivenessConfig = { enabled: true, maxDeferMs: 600_000, checkIntervalMs: 60_000 }; + const config: ForegroundLivenessConfig = { enabled: true, maxDeferMs: 600_000, checkIntervalMs: 60_000, maxReleasePerSweep: 25 }; it("is stale once the pending age is at or beyond maxDeferMs", () => { expect(isForegroundDeferralStale(config, now - config.maxDeferMs - 1, now)).toBe(true); @@ -105,3 +123,40 @@ describe("isForegroundDeferralStale", () => { expect(isForegroundDeferralStale(disabled, now - config.maxDeferMs * 100, now)).toBe(false); }); }); + +describe("selectForegroundDeferralsToRelease (#selfhost-queue-liveness ramp-up)", () => { + it("returns every candidate unchanged when the count is at or below the cap", () => { + const candidates = [{ id: "a", pendingSinceMs: 100 }, { id: "b", pendingSinceMs: 50 }]; + expect(selectForegroundDeferralsToRelease(candidates, 2)).toEqual(candidates); + expect(selectForegroundDeferralsToRelease(candidates, 5)).toEqual(candidates); + }); + + it("returns an empty array when given no candidates", () => { + expect(selectForegroundDeferralsToRelease([], 5)).toEqual([]); + }); + + it("picks the OLDEST (smallest pendingSinceMs) candidates first when count exceeds the cap", () => { + const candidates = [ + { id: "newest", pendingSinceMs: 300 }, + { id: "oldest", pendingSinceMs: 100 }, + { id: "middle", pendingSinceMs: 200 }, + ]; + const selected = selectForegroundDeferralsToRelease(candidates, 2); + expect(selected.map((c) => c.id)).toEqual(["oldest", "middle"]); + }); + + it("breaks ties by original array order (stable) when pendingSinceMs is equal", () => { + const candidates = [ + { id: "first", pendingSinceMs: 100 }, + { id: "second", pendingSinceMs: 100 }, + { id: "third", pendingSinceMs: 100 }, + ]; + const selected = selectForegroundDeferralsToRelease(candidates, 2); + expect(selected.map((c) => c.id)).toEqual(["first", "second"]); + }); + + it("a cap of exactly the candidate count releases all of them", () => { + const candidates = [{ id: "a", pendingSinceMs: 1 }, { id: "b", pendingSinceMs: 2 }]; + expect(selectForegroundDeferralsToRelease(candidates, 2).length).toBe(2); + }); +}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 5dad20193f..81efb1a83e 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -1805,6 +1805,44 @@ describe("createPgQueue (durable #977)", () => { expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 3"); }); + // Ramp-up cap (#selfhost-queue-liveness): a large inherited backlog (the production incident had ~190 + // over-deferred rows) must not release ALL of it in one sweep -- that many jobs re-attempting GitHub reads + // at once can immediately re-trip the same rate-limit bucket they were deferred for. With the cap set + // below the eligible count, assert only `cap` rows get their UPDATE issued, and that the OLDEST rows + // (smallest created_at) are the ones chosen. + it("caps releases at FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP, releasing the oldest rows first", async () => { + process.env.FOREGROUND_LIVENESS_MAX_DEFER_MS = "60000"; // 1m floor + process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP = "2"; + const m = makePool(); + const now = Date.now(); + // 4 stale-eligible rows at distinct ages; only the 2 OLDEST should be released. + m.setForegroundLivenessCandidates([ + { id: "oldest", created_at: now - 10 * 60_000 }, + { id: "second-oldest", created_at: now - 8 * 60_000 }, + { id: "newer", created_at: now - 6 * 60_000 }, + { id: "newest", created_at: now - 5 * 60_000 }, + ]); + const q = createPgQueue(m.pool, async () => undefined); + + const released = await q.releaseStaleForegroundDeferrals(); + + expect(released).toBe(2); + for (const id of ["oldest", "second-oldest"]) { + expect(m.fn).toHaveBeenCalledWith( + expect.stringContaining("SET run_after=$1 WHERE id=$2 AND status='pending' AND run_after>$1"), + expect.arrayContaining([id]), + ); + } + for (const id of ["newer", "newest"]) { + expect(m.fn).not.toHaveBeenCalledWith( + expect.stringContaining("SET run_after=$1 WHERE id=$2 AND status='pending' AND run_after>$1"), + expect.arrayContaining([id]), + ); + } + expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 2"); + delete process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP; + }); + // A stale candidate can lose the UPDATE race (another instance/tick already moved it) -- mirrors // reviveDeadLetterJobs' own "AND status='dead'" re-check pattern: only rows whose UPDATE actually matched // (rowCount 1) count toward the release total, never the raw SELECT candidate count. diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index ea3e34043c..affea7985d 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1672,6 +1672,7 @@ describe("createSqliteQueue (durable #980)", () => { afterEach(() => { delete process.env.FOREGROUND_LIVENESS_ENABLED; delete process.env.FOREGROUND_LIVENESS_MAX_DEFER_MS; + delete process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP; }); /** Directly inserts a foreground-priority (>=8 by default) pending row with an explicit created_at/run_after, @@ -1887,6 +1888,36 @@ describe("createSqliteQueue (durable #980)", () => { expect(started.length).toBe(3); }); + // Ramp-up cap (#selfhost-queue-liveness): a large inherited backlog (the production incident had ~190 + // over-deferred rows) must not release ALL of it in one sweep -- that many jobs re-attempting GitHub reads + // at once can immediately re-trip the same rate-limit bucket they were deferred for. With the cap set below + // the eligible count, assert exactly `cap` rows release (not all of them), and that the OLDEST rows + // (smallest created_at) are the ones chosen -- the newest-of-the-batch row must still be pending afterward. + it("caps releases at FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP, releasing the oldest rows first", async () => { + process.env.FOREGROUND_LIVENESS_MAX_DEFER_MS = "60000"; // 1m floor + process.env.FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP = "2"; + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + const now = Date.now(); + const farFuture = now + 60 * 60_000; + // 4 stale-eligible rows at distinct ages; only the 2 OLDEST should be released. + const ages = [10 * 60_000, 8 * 60_000, 6 * 60_000, 5 * 60_000]; // minutes-old, oldest first + for (const ageMs of ages) { + seedForegroundPendingRow(driver, { createdAt: now - ageMs, runAfter: farFuture }); + } + + const released = q.releaseStaleForegroundDeferrals(); + + expect(released).toBe(2); + expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 2"); + const remainingFuture = driver.query( + `SELECT COUNT(*) AS c FROM _selfhost_jobs WHERE status='pending' AND run_after>?`, + [now], + ).rows[0] as { c: number }; + // 2 of the 4 seeded rows remain deferred into the future -- the 2 NEWEST (least stale) ones. + expect(remainingFuture.c).toBe(2); + }); + // Mirrors reviveDeadLetterJobsSafely's own regression test: the foreground-liveness interval had no error // handler of its own, so a thrown driver/metric failure on that tick would surface as an uncaught exception // and could terminate the process -- exactly the failure mode pump()'s own try/catch already guards against