fix(jobs): take the heavy lane from a holder that outlived its expiry

A hold ended only when its handler settled. pg-boss fails a job at its
expiry and fires its abort signal but does not stop the handler, so a
backup that hung held the lane for the life of the process: every heavy
job, a restore an operator started among them, went back to the queue
every minute from then on, and the only trace was one warn line after
two hours.

A hold now carries the longest expiry of the jobs it runs. A job that
finds the lane held past that expiry and one deferral of grace reports
job.heavy_lane.holder_expired at error and takes the lane; the old
handler, should it ever settle, releases only a hold that is still its
own.
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 21:43:55 +02:00
1 parent a840e30fac
commit 6a87810dbf
2 files changed
+142 -5

No files matched your search

+91 -1
View File
@@ -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<void>((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<void>((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[] = [];
+51 -4
View File
@@ -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<string>;
/** 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<unknown>[], 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<J extends Job<unknown>>(
}
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<J extends Job<unknown>>(
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;
}
};
}