mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
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.
This commit is contained in:
Marc-André Bombeck committed
1 parent
9328a4298d
commit
32c3248b57
7 files changed
+611
-8
No files matched your search
@@ -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<string, unknown>;
|
||||
}>,
|
||||
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([]);
|
||||
});
|
||||
});
|
||||
@@ -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<object>[];
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<LastQueueRun | null> {
|
||||
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<ActiveJob[] | null> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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<string, string | number | boolean>;
|
||||
|
||||
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<string, unknown>,
|
||||
): 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<R>(
|
||||
queue: string,
|
||||
jobs: readonly ObservedJob[],
|
||||
run: () => Promise<R>,
|
||||
clock: JobObserverClock = systemClock,
|
||||
): Promise<R> {
|
||||
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<string, unknown>) => {
|
||||
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<void> {
|
||||
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.",
|
||||
});
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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<T> = (jobs: Job<T>[]) => Promise<JobOutcome>;
|
||||
@@ -80,7 +81,13 @@ export function runJob<J>(
|
||||
handler: (jobs: J[]) => Promise<JobOutcome>,
|
||||
): (jobs: J[]) => Promise<SerializedJobOutcome> {
|
||||
return async (jobs: J[]): Promise<SerializedJobOutcome> => {
|
||||
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);
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user