From 32c3248b57513dd82437e63ce2225b77bc088679 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 25 Sep 2026 00:31:30 +0200 Subject: [PATCH] feat(jobs): log long-running, expired and cut-off jobs by queue A job that ran into its pg-boss expiry left no trace in the app log. Most handlers annotate a wide event nobody emits, pg-boss records an expiry only in its own table, and a worker that dies of heap exhaustion writes one FATAL line that names no queue. On a large account three queues failed with "job timed out" for days and the log said nothing about them (#1031). Every handler now runs inside `observeJob` (through `runJob`), which writes `job.progress` every two minutes while a handler runs (elapsed time, heap, RSS and the last progress the pass reported), `job.expired` when the job passes its expiry while the handler is still running, and `job.finished` when a long or expired run settles. At worker boot, `worker.boot.jobs_cut_off` names every job a previous process was still running when it stopped. The admin card's "job timed out" now says what the limit was and that the cause is either a run too slow for it or a restart while it ran. --- src/lib/jobs/__tests__/job-observer.test.ts | 200 +++++++++++++++++++ src/lib/jobs/__tests__/run-job.test.ts | 22 ++ src/lib/jobs/job-failures.ts | 111 +++++++++- src/lib/jobs/job-observer.ts | 211 ++++++++++++++++++++ src/lib/jobs/reminder-worker.ts | 5 + src/lib/jobs/run-job.ts | 9 +- tests/integration/job-failures.test.ts | 61 +++++- 7 files changed, 611 insertions(+), 8 deletions(-) create mode 100644 src/lib/jobs/__tests__/job-observer.test.ts create mode 100644 src/lib/jobs/job-observer.ts diff --git a/src/lib/jobs/__tests__/job-observer.test.ts b/src/lib/jobs/__tests__/job-observer.test.ts new file mode 100644 index 000000000..9e7c20920 --- /dev/null +++ b/src/lib/jobs/__tests__/job-observer.test.ts @@ -0,0 +1,200 @@ +/** + * Issue #1031: three queues failed with `job timed out` night after night and + * the operator found no line about any of them in the app log. These tests + * pin the lines `observeJob` now writes — progress while a job runs, a warning + * at its expiry, and a closing line for a long or expired run — and the boot + * line naming jobs a previous process was running when it stopped. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + emitted: [] as Array<{ + level: string; + background?: { task_name?: string }; + meta?: Record; + }>, + active: vi.fn(), +})); + +vi.mock("@/lib/logging/transports", () => ({ + emitIfSampled: (event: (typeof mocks.emitted)[number]) => { + mocks.emitted.push(event); + }, +})); +vi.mock("@/lib/jobs/job-failures", () => ({ + readActiveJobsStartedBefore: mocks.active, +})); + +import { + JOB_LONG_RUN_MS, + JOB_PROGRESS_INTERVAL_MS, + observeJob, + reportJobProgress, + reportJobsCutOffAtBoot, + type JobObserverClock, +} from "@/lib/jobs/job-observer"; + +/** A manual clock: `advance` fires due timers in order. */ +function manualClock() { + let now = 0; + let seq = 0; + const timers = new Map< + number, + { at: number; every: number | null; fn: () => void } + >(); + const clock: JobObserverClock = { + now: () => now, + setTimeout: (fn, ms) => { + seq += 1; + timers.set(seq, { at: now + ms, every: null, fn }); + return seq; + }, + setInterval: (fn, ms) => { + seq += 1; + timers.set(seq, { at: now + ms, every: ms, fn }); + return seq; + }, + clearTimeout: (handle) => timers.delete(handle as number), + clearInterval: (handle) => timers.delete(handle as number), + }; + const advance = (ms: number) => { + const target = now + ms; + for (;;) { + const due = [...timers.entries()] + .filter(([, t]) => t.at <= target) + .sort((a, b) => a[1].at - b[1].at)[0]; + if (!due) break; + const [id, timer] = due; + now = timer.at; + if (timer.every === null) timers.delete(id); + else timer.at += timer.every; + timer.fn(); + } + now = target; + }; + return { clock, advance }; +} + +function lines(task: string) { + return mocks.emitted.filter((e) => e.background?.task_name === task); +} + +beforeEach(() => { + mocks.emitted.length = 0; + mocks.active.mockReset(); +}); + +describe("observeJob", () => { + it("writes progress while a long job runs, a warning at its expiry, and a closing line", async () => { + const { clock, advance } = manualClock(); + let release!: (value: { ok: true; did: { days: number } }) => void; + + const running = observeJob( + "dense-intraday-retention", + [{ id: "job-1", expireInSeconds: 900 }], + () => + new Promise<{ ok: true; did: { days: number } }>((resolve) => { + reportJobProgress({ consolidation_day: "2024-03-01" }); + release = resolve; + }), + clock, + ); + + advance(JOB_PROGRESS_INTERVAL_MS); + expect(lines("job.progress")).toHaveLength(1); + expect(lines("job.progress")[0]!.meta).toMatchObject({ + queue: "dense-intraday-retention", + job_ids: "job-1", + expire_s: 900, + consolidation_day: "2024-03-01", + }); + expect(lines("job.progress")[0]!.meta?.heap_used_mb).toEqual( + expect.any(Number), + ); + + advance(900 * 1000 - JOB_PROGRESS_INTERVAL_MS); + const expired = lines("job.expired"); + expect(expired).toHaveLength(1); + expect(expired[0]!.level).toBe("warn"); + expect(expired[0]!.meta).toMatchObject({ + queue: "dense-intraday-retention", + elapsed_s: 900, + consolidation_day: "2024-03-01", + }); + + release({ ok: true, did: { days: 12 } }); + await running; + const finished = lines("job.finished"); + expect(finished).toHaveLength(1); + expect(finished[0]!.level).toBe("warn"); + expect(finished[0]!.meta).toMatchObject({ + queue: "dense-intraday-retention", + finished_after_expiry: true, + outcome: "completed", + did: { days: 12 }, + }); + }); + + it("stays silent for a short job", async () => { + const { clock } = manualClock(); + await observeJob( + "host-metric-sample", + [{ id: "j", expireInSeconds: 900 }], + async () => ({ ok: true }), + clock, + ); + expect(mocks.emitted).toEqual([]); + }); + + it("closes a long job that threw with the error, and rethrows it", async () => { + const { clock, advance } = manualClock(); + const running = observeJob( + "data-backup", + [{ id: "b", expireInSeconds: 7200 }], + async () => { + advance(JOB_LONG_RUN_MS); + throw new Error("write failed"); + }, + clock, + ); + await expect(running).rejects.toThrow("write failed"); + expect(lines("job.finished")[0]!.meta).toMatchObject({ + queue: "data-backup", + outcome: "threw", + error: "write failed", + }); + expect(lines("job.expired")).toEqual([]); + }); +}); + +describe("reportJobsCutOffAtBoot", () => { + it("names each job a previous process was running", async () => { + mocks.active.mockResolvedValue([ + { + queue: "data-backup", + id: "a", + startedAt: "2026-09-24T09:36:00.000Z", + expireSeconds: 7200, + }, + ]); + await reportJobsCutOffAtBoot(new Date("2026-09-24T10:00:00.000Z")); + const line = lines("worker.boot.jobs_cut_off"); + expect(line).toHaveLength(1); + expect(line[0]!.level).toBe("warn"); + expect(line[0]!.meta).toMatchObject({ + jobs_cut_off: 1, + queues: "data-backup", + }); + expect(String(line[0]!.meta?.jobs)).toContain( + "data-backup@2026-09-24T09:36:00.000Z", + ); + }); + + it("stays silent when nothing was cut off, or there is no queue schema", async () => { + mocks.active.mockResolvedValueOnce([]); + await reportJobsCutOffAtBoot(); + mocks.active.mockResolvedValueOnce(null); + await reportJobsCutOffAtBoot(); + expect(mocks.emitted).toEqual([]); + }); +}); diff --git a/src/lib/jobs/__tests__/run-job.test.ts b/src/lib/jobs/__tests__/run-job.test.ts index 4558ccea9..9b32e591f 100644 --- a/src/lib/jobs/__tests__/run-job.test.ts +++ b/src/lib/jobs/__tests__/run-job.test.ts @@ -149,3 +149,25 @@ describe("runJob", () => { expect(handler).toHaveBeenCalledWith(jobs); }); }); + +describe("runJob observation (#1031)", () => { + it("runs every handler inside the job observer, so a long or expired job leaves a log line", async () => { + const observer = await import("../job-observer"); + const observe = vi.spyOn(observer, "observeJob"); + const jobs = [ + { id: "j-1", expireInSeconds: 900 }, + ] as unknown as Job[]; + const wrapped = runJob("dense-intraday-retention", async () => + jobDone({ days_consolidated: 1 }), + ); + + await wrapped(jobs); + + expect(observe).toHaveBeenCalledWith( + "dense-intraday-retention", + jobs, + expect.any(Function), + ); + observe.mockRestore(); + }); +}); diff --git a/src/lib/jobs/job-failures.ts b/src/lib/jobs/job-failures.ts index d846c88be..6406e320c 100644 --- a/src/lib/jobs/job-failures.ts +++ b/src/lib/jobs/job-failures.ts @@ -66,6 +66,35 @@ interface FailingQueueRow { failures: number; last_failed_at: Date | null; last_error: string | null; + last_expire_seconds: number | null; +} + +/** + * The two texts pg-boss writes when a job ran out of time: the supervisor's, + * for a job still `active` past its expiry, and the worker's own, for a + * handler that did not settle inside it. + */ +const TIMEOUT_MESSAGE = /^(job timed out|handler execution exceeded \d+s)$/; + +/** + * Say what a bare timeout means. "job timed out" alone sent an operator + * nowhere: pg-boss writes the same words for a pass that was too slow for its + * limit and for a job whose process died under it (a restart, or a container + * killed for memory), in which case the job sits `active` until the limit runs + * out. The limit and the two causes are what the operator can check. + */ +function explainTimeout(message: string, expireSeconds: number | null): string { + const limit = + expireSeconds !== null && expireSeconds > 0 + ? expireSeconds >= 3600 && expireSeconds % 3600 === 0 + ? `${expireSeconds / 3600} h` + : `${Math.round(expireSeconds / 60)} min` + : null; + return ( + `${message}: did not finish within its ${limit ?? "time"} limit. ` + + `Either the run was too slow for the limit, or the app restarted while it ` + + `ran; a restart shows in the container's restart count and log.` + ); } /** @@ -73,10 +102,15 @@ interface FailingQueueRow { * serialises to `{ message, stack, … }`; a thrown non-Error can be any JSON at * all, so the raw text is the fallback rather than an invented placeholder. */ -function presentError(raw: string | null): string { +function presentError( + raw: string | null, + expireSeconds: number | null = null, +): string { if (raw === null) return ""; const trimmed = raw.trim(); if (trimmed.length === 0) return ""; + if (TIMEOUT_MESSAGE.test(trimmed)) + return explainTimeout(trimmed, expireSeconds); const redacted = redactSecrets(trimmed); return redacted.length > MAX_ERROR_CHARS ? `${redacted.slice(0, MAX_ERROR_CHARS)}…` @@ -105,7 +139,10 @@ export async function readFailingQueues( ) ORDER BY completed_on DESC NULLS LAST ) - )[1] AS last_error + )[1] AS last_error, + ( + array_agg(expire_seconds ORDER BY completed_on DESC NULLS LAST) + )[1] AS last_expire_seconds FROM pgboss.job WHERE state = 'failed' AND completed_on > now() - make_interval(hours => ${withinHours}::int) @@ -119,7 +156,7 @@ export async function readFailingQueues( queue: row.name, failures: row.failures, lastFailedAt: row.last_failed_at!.toISOString(), - lastError: presentError(row.last_error), + lastError: presentError(row.last_error, row.last_expire_seconds), })); } catch { // No `pgboss` schema (web-only deployment) or no permission to read it. @@ -160,11 +197,17 @@ export async function readLastQueueRun( ): Promise { try { const rows = await prisma.$queryRaw< - Array<{ state: string; completed_on: Date | null; output: string | null }> + Array<{ + state: string; + completed_on: Date | null; + output: string | null; + expire_seconds: number | null; + }> >` SELECT state, completed_on, + expire_seconds, -- pg-boss wraps a thrown value: an expired job's row reads -- {"value":{"message":"job timed out"}}, so a plain ->>'message' -- misses it and the operator gets raw JSON instead of the sentence. @@ -189,7 +232,10 @@ export async function readLastQueueRun( return { at: row.completed_on.toISOString(), state, - error: state === "completed" ? null : presentError(row.output), + error: + state === "completed" + ? null + : presentError(row.output, row.expire_seconds), }; } catch { // No `pgboss` schema (web-only deployment) or no permission to read it. @@ -238,3 +284,58 @@ export async function readQueueFailureForUser( return null; } } + +/** A job pg-boss still has as running, from `readActiveJobsStartedBefore`. */ +export interface ActiveJob { + queue: string; + id: string; + /** ISO instant the job was picked up. */ + startedAt: string; + /** The job's expiry, in seconds, as pg-boss holds it. */ + expireSeconds: number; +} + +/** + * Jobs still marked active that were picked up before `before` — at worker + * boot, before this process has taken any job, that is every job a previous + * process was running when it stopped. + * + * Why it matters (#1031): a worker killed for memory writes nothing, and + * pg-boss only marks its job `job timed out` once the expiry has passed, + * which for the backup is two hours later. Reading these rows at boot turns + * a silent restart into a line naming what was cut off. With more than one + * worker process on the same database, a job another live worker is running + * would show up here too, which the boot line says. + * + * `null` when the queue schema is absent, like the readers above. + */ +export async function readActiveJobsStartedBefore( + before: Date, +): Promise { + try { + const rows = await prisma.$queryRaw< + Array<{ + name: string; + id: string; + started_on: Date | null; + expire_seconds: number; + }> + >` + SELECT name, id::text AS id, started_on, expire_seconds + FROM pgboss.job + WHERE state = 'active' AND started_on < ${before} + ORDER BY started_on + LIMIT ${MAX_QUEUES} + `; + return rows + .filter((row) => row.started_on !== null) + .map((row) => ({ + queue: row.name, + id: row.id, + startedAt: row.started_on!.toISOString(), + expireSeconds: row.expire_seconds, + })); + } catch { + return null; + } +} diff --git a/src/lib/jobs/job-observer.ts b/src/lib/jobs/job-observer.ts new file mode 100644 index 000000000..18c4166d6 --- /dev/null +++ b/src/lib/jobs/job-observer.ts @@ -0,0 +1,211 @@ +/** + * Log lines for long-running and cut-off jobs. + * + * Issue #1031: on a large account three queues failed night after night with + * `job timed out`, and the operator found not one line about any of them in + * the app log. Nothing wrote one. Most handlers annotate a wide event that + * nobody emits, pg-boss records an expiry only in its own table, and a worker + * killed for memory cannot say anything at all. The admin card showed the + * failure; the log showed nothing about what the job was doing, how far it + * got, or why it stopped. + * + * `observeJob` wraps every handler (through `runJob`) and writes three kinds + * of line, each a background wide event naming the queue: + * + * - `job.progress`, every two minutes while a handler runs, with how long it + * has been running, the process's memory, and whatever the handler last + * reported through `reportJobProgress`. A job that runs for a minute never + * writes one; a job that is killed leaves its last one behind. + * - `job.expired`, at the job's expiry if the handler has not settled: the + * moment pg-boss declares it timed out. pg-boss does not stop the handler, + * so this line says it is still running and how far it has got. + * - `job.finished`, when a handler that ran long or past its expiry + * settles, with its outcome and the facts it returned. + * + * `reportOrphanedJobs` covers the case no running handler can: at worker boot, + * jobs still marked active were started by a process that is gone. + */ +import { AsyncLocalStorage } from "node:async_hooks"; + +import { WideEventBuilder } from "@/lib/logging/event-builder"; +import { emitIfSampled } from "@/lib/logging/transports"; +import { readActiveJobsStartedBefore } from "@/lib/jobs/job-failures"; + +/** How often a running job writes `job.progress`. */ +export const JOB_PROGRESS_INTERVAL_MS = 2 * 60 * 1000; + +/** A handler that ran at least this long writes `job.finished`. */ +export const JOB_LONG_RUN_MS = 60 * 1000; + +type ProgressFacts = Record; + +const progressStore = new AsyncLocalStorage<{ facts: ProgressFacts }>(); + +/** + * Record how far the running job has got. Cheap: it only replaces the facts + * the next `job.progress` / `job.expired` line carries, so a pass can call it + * once per unit of work. A no-op outside an observed job. + */ +export function reportJobProgress(facts: ProgressFacts): void { + const cell = progressStore.getStore(); + if (cell) cell.facts = { ...cell.facts, ...facts }; +} + +/** What `observeJob` reads from a pg-boss job. */ +interface ObservedJob { + id?: string; + expireInSeconds?: number; +} + +/** Injectable clock and timers, for tests. */ +export interface JobObserverClock { + now: () => number; + setTimeout: (fn: () => void, ms: number) => unknown; + setInterval: (fn: () => void, ms: number) => unknown; + clearTimeout: (handle: unknown) => void; + clearInterval: (handle: unknown) => void; +} + +const systemClock: JobObserverClock = { + now: () => Date.now(), + setTimeout: (fn, ms) => setTimeout(fn, ms).unref(), + setInterval: (fn, ms) => setInterval(fn, ms).unref(), + clearTimeout: (handle) => clearTimeout(handle as NodeJS.Timeout), + clearInterval: (handle) => clearInterval(handle as NodeJS.Timeout), +}; + +function megabytes(bytes: number): number { + return Math.round(bytes / 1024 / 1024); +} + +function emitJobLine( + taskName: string, + level: "info" | "warn", + meta: Record, +): void { + try { + const event = new WideEventBuilder("background"); + event.setBackground({ task_name: taskName }); + for (const [key, value] of Object.entries(meta)) event.addMeta(key, value); + if (level === "warn") event.elevateLevel("warn"); + event.finish(); + emitIfSampled(event.toJSON()); + } catch { + // A log line must never break the job it describes. + } +} + +/** + * Run `run` for `jobs` of `queue`, writing the progress, expiry and finish + * lines described above. Resolves or rejects exactly as `run` does. + */ +export async function observeJob( + queue: string, + jobs: readonly ObservedJob[], + run: () => Promise, + clock: JobObserverClock = systemClock, +): Promise { + const startedAt = clock.now(); + const cell: { facts: ProgressFacts } = { facts: {} }; + const jobIds = jobs + .map((job) => job.id) + .filter(Boolean) + .join(","); + const expireSeconds = Math.min( + ...jobs.map((job) => + typeof job.expireInSeconds === "number" && job.expireInSeconds > 0 + ? job.expireInSeconds + : Number.POSITIVE_INFINITY, + ), + ); + let expired = false; + + const snapshot = () => { + const memory = process.memoryUsage(); + return { + queue, + job_ids: jobIds, + elapsed_s: Math.round((clock.now() - startedAt) / 1000), + ...(Number.isFinite(expireSeconds) ? { expire_s: expireSeconds } : {}), + heap_used_mb: megabytes(memory.heapUsed), + rss_mb: megabytes(memory.rss), + ...cell.facts, + }; + }; + + const heartbeat = clock.setInterval(() => { + emitJobLine("job.progress", "info", snapshot()); + }, JOB_PROGRESS_INTERVAL_MS); + const expiry = Number.isFinite(expireSeconds) + ? clock.setTimeout(() => { + expired = true; + emitJobLine("job.expired", "warn", { + ...snapshot(), + reason: + "pg-boss has marked this job timed out; the handler is still running", + }); + }, expireSeconds * 1000) + : null; + + const finish = (outcome: Record) => { + clock.clearInterval(heartbeat); + if (expiry !== null) clock.clearTimeout(expiry); + const elapsed = clock.now() - startedAt; + if (!expired && elapsed < JOB_LONG_RUN_MS) return; + emitJobLine("job.finished", expired ? "warn" : "info", { + ...snapshot(), + ...(expired ? { finished_after_expiry: true } : {}), + ...outcome, + }); + }; + + try { + const result = await progressStore.run(cell, run); + const did = + result !== null && typeof result === "object" && "did" in result + ? (result as { did?: unknown }).did + : undefined; + const ok = + result !== null && typeof result === "object" && "ok" in result + ? (result as { ok?: unknown }).ok !== false + : true; + finish({ + outcome: ok ? "completed" : "failed", + ...(did && typeof did === "object" ? { did } : {}), + }); + return result; + } catch (err) { + finish({ + outcome: "threw", + error: err instanceof Error ? err.message : String(err), + }); + throw err; + } +} + +/** + * At worker boot, write one line naming every job a previous process was + * running when it stopped. Must run before this process takes any job. + * + * A worker killed for memory cannot log its own death, and pg-boss marks the + * job `job timed out` only once the expiry has passed. This line is where a + * restart in the middle of a job becomes visible: the queue, when the job + * started, and its limit. Silent when nothing was cut off. + */ +export async function reportJobsCutOffAtBoot( + bootedAt: Date = new Date(), +): Promise { + const active = await readActiveJobsStartedBefore(bootedAt); + if (active === null || active.length === 0) return; + emitJobLine("worker.boot.jobs_cut_off", "warn", { + jobs_cut_off: active.length, + queues: [...new Set(active.map((job) => job.queue))].join(","), + jobs: active + .map( + (job) => `${job.queue}@${job.startedAt}(limit ${job.expireSeconds}s)`, + ) + .join(" "), + reason: + "these jobs were still running when the previous worker process stopped (a restart, a deploy, or the container running out of memory); pg-boss retries or fails them once their limit passes. If another worker process shares this database, its running jobs are listed too.", + }); +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index f8a20a384..e3e732d2c 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -16,6 +16,7 @@ */ import { markWorkerStarted, recordError } from "@/lib/jobs/worker-status"; import { createWorkerBoss, setGlobalBoss } from "@/lib/jobs/boss-instance"; +import { reportJobsCutOffAtBoot } from "@/lib/jobs/job-observer"; import { assertSubsystemEnabled } from "@/lib/process-type"; import { DATABASE_URL, workerLog } from "./reminder/shared"; import { @@ -55,7 +56,11 @@ export async function startReminderWorker() { recordError(); }); + // Read before any `boss.work` binding below can take a job: whatever is + // still active now was started by a process that is gone. + const bootedAt = new Date(); await boss.start(); + await reportJobsCutOffAtBoot(bootedAt); setGlobalBoss(boss); markWorkerStarted(); diff --git a/src/lib/jobs/run-job.ts b/src/lib/jobs/run-job.ts index 980bc5a1d..9a673bc8c 100644 --- a/src/lib/jobs/run-job.ts +++ b/src/lib/jobs/run-job.ts @@ -28,6 +28,7 @@ import { type SerializedJobOutcome, } from "./job-outcome"; import { reportWorkerError } from "./report-worker-error"; +import { observeJob } from "./job-observer"; /** A converted handler: it says what it did instead of just finishing. */ export type JobHandler = (jobs: Job[]) => Promise; @@ -80,7 +81,13 @@ export function runJob( handler: (jobs: J[]) => Promise, ): (jobs: J[]) => Promise { return async (jobs: J[]): Promise => { - const outcome = await handler(jobs); + // Observed, so a long run, an expiry and a cut-off leave a line in the + // log naming the queue and how far it got (see `job-observer.ts`). + const outcome = await observeJob( + queue, + jobs as ReadonlyArray<{ id?: string; expireInSeconds?: number }>, + () => handler(jobs), + ); if (outcome.ok) return serializeJobOutcome(outcome); const failure = new JobFailure(queue, outcome.reason, outcome.cause); diff --git a/tests/integration/job-failures.test.ts b/tests/integration/job-failures.test.ts index 18a6f6812..a4c108986 100644 --- a/tests/integration/job-failures.test.ts +++ b/tests/integration/job-failures.test.ts @@ -16,6 +16,7 @@ import { PgBoss } from "pg-boss"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { + readActiveJobsStartedBefore, readFailingQueues, readQueueFailureForUser, } from "@/lib/jobs/job-failures"; @@ -28,8 +29,16 @@ const OTHER_QUEUE = "reminder-check"; let boss: PgBoss; /** Send a job, take it, and fail it terminally so it lands in `state='failed'`. */ -async function failOneJob(queue: string, data: object, error: string) { - const jobId = await boss.send(queue, data, { retryLimit: 0 }); +async function failOneJob( + queue: string, + data: object, + error: string, + expireInSeconds?: number, +) { + const jobId = await boss.send(queue, data, { + retryLimit: 0, + ...(expireInSeconds === undefined ? {} : { expireInSeconds }), + }); if (!jobId) throw new Error("queue rejected the job"); const [job] = await boss.fetch(queue); if (!job) throw new Error("nothing to fetch"); @@ -81,6 +90,26 @@ describe("readFailingQueues", () => { expect(other?.lastError).toBe("reminder dispatch failed"); }); + it("says what a bare timeout means, with the job's own limit (#1031)", async () => { + await getPrismaClient().$executeRawUnsafe( + `DELETE FROM pgboss.job WHERE state = 'failed'`, + ); + // pg-boss writes these words both for a pass too slow for its limit and + // for a job whose process died under it; alone they sent an operator + // nowhere. + await failOneJob(QUEUE, {}, "job timed out", 900); + await failOneJob(OTHER_QUEUE, {}, "job timed out", 7200); + + const failing = await readFailingQueues(); + const fifteen = failing?.find((entry) => entry.queue === QUEUE); + expect(fifteen?.lastError).toMatch( + /^job timed out: did not finish within its 15 min limit\./, + ); + expect(fifteen?.lastError).toMatch(/restarted/); + const twoHours = failing?.find((entry) => entry.queue === OTHER_QUEUE); + expect(twoHours?.lastError).toMatch(/within its 2 h limit/); + }); + it("reports an empty list — not null — when nothing is failing", async () => { await getPrismaClient().$executeRawUnsafe( `DELETE FROM pgboss.job WHERE state = 'failed'`, @@ -132,6 +161,34 @@ describe("readQueueFailureForUser", () => { }); }); +describe("readActiveJobsStartedBefore (#1031)", () => { + it("lists a job still marked active from before the given instant, and nothing after it", async () => { + const jobId = await boss.send( + OTHER_QUEUE, + {}, + { retryLimit: 0, expireInSeconds: 7200 }, + ); + const [job] = await boss.fetch(OTHER_QUEUE); + expect(job?.id).toBe(jobId); + + const cutOff = await readActiveJobsStartedBefore( + new Date(Date.now() + 60_000), + ); + expect(cutOff).toContainEqual( + expect.objectContaining({ + queue: OTHER_QUEUE, + id: jobId, + expireSeconds: 7200, + }), + ); + await expect( + readActiveJobsStartedBefore(new Date(Date.now() - 60 * 60_000)), + ).resolves.toEqual([]); + + await boss.complete(OTHER_QUEUE, jobId!); + }); +}); + describe("no queue schema at all", () => { it("answers null rather than throwing, so the status page still renders", async () => { const prisma = getPrismaClient();