diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts index 8ebea3ea1..ea90f59c6 100644 --- a/src/lib/jobs/__tests__/heavy-lane.test.ts +++ b/src/lib/jobs/__tests__/heavy-lane.test.ts @@ -15,7 +15,12 @@ vi.mock("@/lib/observability/container-memory", () => ({ })); import { emitSignal } from "@/lib/logging/signal"; -import { heavyLane, LANE_WAIT_WARN_MS, laneWanted } from "../heavy-lane"; +import { + heavyLane, + LANE_HOLD_GRACE_SECONDS, + LANE_WAIT_WARN_MS, + laneWanted, +} from "../heavy-lane"; import { jobDone } from "../job-outcome"; type TestJob = Job<{ force?: boolean }>; @@ -271,6 +276,91 @@ describe("a long wait for the lane", () => { }); }); +describe("a holder that never finishes", () => { + it("loses the lane once it has outlived its own expiry, and the overrun is reported", async () => { + vi.mocked(emitSignal).mockClear(); + const start = Date.parse("2026-10-20T02:00:00Z"); + let clock = start; + const now = () => clock; + const { back, defer } = deferrals(); + // A backup whose handler never settles: pg-boss failed it at its expiry, + // but the handler is still awaiting. + let unstick!: () => void; + const stuck = new Promise((r) => (unstick = r)); + const backup = heavyLane( + "data-backup", + async () => { + await stuck; + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + now, + )([{ ...job("nightly"), expireInSeconds: 7200 } as TestJob]); + await tick(); + + const restore = vi.fn(async () => jobDone({ jobs: 1 })); + const tryRestore = () => + heavyLane( + "backup-restore", + restore, + undefined, + defer, + now, + )([{ ...job("restore"), expireInSeconds: 21600 } as TestJob]); + + // Inside the backup's expiry and grace, the restore waits its turn. + clock = start + (7200 + LANE_HOLD_GRACE_SECONDS) * 1000; + await tryRestore(); + expect(restore).not.toHaveBeenCalled(); + expect(back.map((b) => b.queue)).toEqual(["backup-restore"]); + expect(emitSignal).not.toHaveBeenCalled(); + + // Past it, the lane is the restore's, and the overrun is an error. + clock += 60_000; + await tryRestore(); + expect(restore).toHaveBeenCalledTimes(1); + expect(emitSignal).toHaveBeenCalledWith({ + action: "job.heavy_lane.holder_expired", + level: "error", + meta: { + holder: "data-backup", + queue: "backup-restore", + overran_minutes: 1, + }, + }); + + // The stuck handler settling later does not free a lane it no longer + // holds: a job that arrives while the next holder runs still waits. + let release!: () => void; + const released = new Promise((r) => (release = r)); + const next = heavyLane( + "rollup-full-backfill", + async () => { + await released; + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + now, + )([{ ...job("fold"), expireInSeconds: 3600 } as TestJob]); + await tick(); + unstick(); + await backup; + const waiting = vi.fn(async () => jobDone({ jobs: 1 })); + await heavyLane( + "data-backup", + waiting, + undefined, + defer, + now, + )([{ ...job("weekly"), expireInSeconds: 7200 } as TestJob]); + expect(waiting).not.toHaveBeenCalled(); + release(); + await next; + }); +}); + describe("the lane under container memory pressure", () => { it("sends a heavy job back while the container is near its limit", async () => { const log: string[] = []; diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts index 4d3bd2c7d..93ae7f85a 100644 --- a/src/lib/jobs/heavy-lane.ts +++ b/src/lib/jobs/heavy-lane.ts @@ -34,6 +34,17 @@ * (`job.heavy_lane.waiting_long`), once per holder, so a lane that is held * for hours shows in the alerts instead of only in `info` lines. * + * A hold ends when its handler settles, and a handler that never settles + * would hold the lane for the life of the process: pg-boss fails a job at its + * expiry and fires its abort signal, but does not stop the handler. Every + * heavy job would go back to the queue every minute from then on, a restore + * an operator started among them, with nothing but one `warn` line to show for + * it. So a hold carries the expiry of the jobs it runs. A job that finds the + * lane held past that expiry and a grace of one deferral takes the lane, and + * the overrun is reported at `error` (`job.heavy_lane.holder_expired`): the + * holder has already been failed by pg-boss, so the lane is the only thing it + * still holds. + * * The container memory guard (`observability/container-memory.ts`) closes * the lane for a while when the cgroup's working set reaches 85 % of its * limit: a heavy job that arrives then goes back to the queue the same way, @@ -102,6 +113,9 @@ export const LANE_YIELD_DELAY_SECONDS = 2 * LANE_DEFER_SECONDS; /** How long a job may stand in line before the wait is reported at `warn`. */ export const LANE_WAIT_WARN_MS = 2 * 60 * 60 * 1000; +/** How long past its jobs' expiry a holder keeps the lane before it is taken. */ +export const LANE_HOLD_GRACE_SECONDS = LANE_DEFER_SECONDS; + /** What one trip back to the queue did. */ export interface LaneDeferral { /** Rows that went back. */ @@ -151,6 +165,18 @@ interface LaneHold { wantedBy: string | null; /** Queues whose long wait has been reported during this hold. */ warned: Set; + /** When the hold has outlived its jobs' expiry and the grace after it. */ + expiresAt: number; +} + +/** + * When a hold of `jobs` entered at `at` has outlived them. A job without an + * expiry (only a test stand-in has none) never forfeits the lane. + */ +function holdExpiresAt(jobs: readonly Job[], at: number): number { + const expiry = Math.max(...jobs.map((job) => job.expireInSeconds)); + if (!Number.isFinite(expiry)) return Number.POSITIVE_INFINITY; + return at + (expiry + LANE_HOLD_GRACE_SECONDS) * 1000; } let hold: LaneHold | null = null; @@ -209,7 +235,21 @@ export function heavyLane>( } return async (jobs) => { if (!jobs.some(inLane)) return handler(jobs); - const current = hold; + let current = hold; + const at = now(); + if (current && at > current.expiresAt) { + emitSignal({ + action: "job.heavy_lane.holder_expired", + level: "error", + meta: { + holder: current.queue, + queue, + overran_minutes: Math.floor((at - current.expiresAt) / 60_000), + }, + }); + hold = null; + current = null; + } if (!current && memoryPressureActive()) { const deferral = await defer(queue, jobs); annotate({ @@ -235,16 +275,23 @@ export function heavyLane>( deferred: deferral.deferred, }, }); - reportLongWait(current, queue, deferral, now()); + reportLongWait(current, queue, deferral, at); return jobDone({ lane_deferred: deferral.deferred }); } - const entered: LaneHold = { queue, wantedBy: null, warned: new Set() }; + const entered: LaneHold = { + queue, + wantedBy: null, + warned: new Set(), + expiresAt: holdExpiresAt(jobs, at), + }; hold = entered; annotate({ action: { name: "job.heavy_lane.entered" }, meta: { queue } }); try { return await holdScope.run(entered, () => handler(jobs)); } finally { - hold = null; + // Only its own hold: one taken over past its expiry belongs to the job + // that took it. + if (hold === entered) hold = null; } }; }