perf(jobs): poll a queue only its cron feeds every 30 seconds

pg-boss polls each idle worker slot every two seconds, so the hundred-odd
bindings asked the database for work about fifty times a second while
most of them wait for one tick a day. `createAndWork` now gives a queue
that a registrar schedules and nothing else sends to a 30-second poll.
Scheduled queues that also take sends keep the default, because with one
job per fetch a slow poll would drain a burst at one job per interval;
the runtime guard holds that list to the send sites in the tree,
including sends through a variable.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 03:35:09 +02:00
1 parent 0770ae3590
commit d9e56482d9
3 files changed
+170 -3

No files matched your search

@@ -41,6 +41,7 @@ import {
type QueueRuntime,
type SourceSite,
} from "@/lib/jobs/queue-runtime";
import { SCHEDULED_AND_SENT_QUEUES } from "@/lib/jobs/reminder/registrar-shared";
const SRC = join(process.cwd(), "src");
const REGISTRARS = [
@@ -369,3 +370,48 @@ describe("every queue declares its runtime", () => {
).toEqual(["planted-long: expiry 900s out of range"]);
});
});
/**
* Queues a file sends to through a variable (`boss.send(queueName, …)`): every
* known queue name the file spells out, as a constant or a map value, counts
* as sent, because the matcher above cannot tell which one the variable holds.
*/
function dynamicallySent(texts: ReadonlyMap<string, string>): Set<string> {
const out = new Set<string>();
for (const [, raw] of texts) {
const text = code(raw);
if (
!/\b(?:boss|getGlobalBoss\(\)\??)\s*\.\s*send\s*(?:<[^(]*?>)?\(\s*[a-z]/.test(
text,
)
) {
continue;
}
for (const m of text.matchAll(/"([a-z0-9-]+)"/g)) {
if (m[1] in QUEUE_RUNTIME) out.add(m[1]);
}
}
return out;
}
describe("a cron-fed queue polls slowly, a sent one does not", () => {
const all = scan(source);
const scheduled = new Set(all.schedules.map((s) => s.queue));
const sent = new Set([
...all.sends.map((s) => s.queue),
...dynamicallySent(source),
]);
it("lists exactly the scheduled queues that also take sends", () => {
const expected = [...scheduled].filter((q) => sent.has(q)).sort();
expect([...SCHEDULED_AND_SENT_QUEUES].sort()).toEqual(expected);
// Most scheduled queues are fed by their cron alone.
expect(scheduled.size - expected.length).toBeGreaterThanOrEqual(40);
});
it("counts a send through a variable as a send to every queue the file names", () => {
const dynamic = dynamicallySent(source);
expect(dynamic.has("withings-activity-sync")).toBe(true);
expect(dynamic.has("whoop-recovery-sync")).toBe(true);
});
});
@@ -1,5 +1,7 @@
/**
* The registrar's boot-time decisions that reach pg-boss on every instance.
* The registrar's two boot-time decisions that reach pg-boss on every
* instance: how often an idle worker polls, and which retired queues are
* removed.
*/
import { describe, expect, it, vi } from "vitest";
@@ -8,7 +10,13 @@ vi.mock("@/lib/db", () => ({
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
import { retireQueues } from "../registrar-shared";
import { jobDone } from "@/lib/jobs/job-outcome";
import {
CRON_FED_POLL_SECONDS,
createAndSchedule,
createAndWork,
retireQueues,
} from "../registrar-shared";
function fakeBoss() {
return {
@@ -21,6 +29,56 @@ function fakeBoss() {
};
}
const handler = async () => jobDone({ jobs: 1 });
function pollOf(boss: ReturnType<typeof fakeBoss>, queue: string) {
const call = boss.work.mock.calls.find((c) => c[0] === queue);
return (call?.[1] as { pollingIntervalSeconds?: number })
.pollingIntervalSeconds;
}
describe("idle polling", () => {
it("polls slowly on a queue only its cron feeds, at pg-boss's default otherwise", async () => {
const boss = fakeBoss();
await createAndSchedule(
boss as never,
["test-cron-only", "pr-detection", "test-send-only"],
[
["test-cron-only", "0 3 * * *"],
// Scheduled, and sent per account on every ingest.
["pr-detection", "*/30 * * * *"],
],
);
for (const queue of ["test-cron-only", "pr-detection", "test-send-only"]) {
await createAndWork(
boss as never,
queue,
{ localConcurrency: 1 },
handler,
);
}
expect(pollOf(boss, "test-cron-only")).toBe(CRON_FED_POLL_SECONDS);
expect(pollOf(boss, "pr-detection")).toBeUndefined();
expect(pollOf(boss, "test-send-only")).toBeUndefined();
});
it("keeps a poll the binding names", async () => {
const boss = fakeBoss();
await createAndSchedule(
boss as never,
["test-named-poll"],
[["test-named-poll", "0 4 * * *"]],
);
await createAndWork(
boss as never,
"test-named-poll",
{ localConcurrency: 1, pollingIntervalSeconds: 5 },
handler,
);
expect(pollOf(boss, "test-named-poll")).toBe(5);
});
});
describe("retireQueues", () => {
it("removes the schedule before the queue, and one failure does not stop the rest", async () => {
const boss = fakeBoss();
+64 -1
View File
@@ -286,6 +286,56 @@ async function reconcileVolumeQueueRetention(
);
}
/**
* The idle poll of a queue only its cron feeds. pg-boss polls every two
* seconds by default, per worker slot, so the hundred-odd bindings asked the
* database for work about fifty times a second while nearly all of them wait
* for one tick a day. A cron job a few seconds late costs nothing.
*/
export const CRON_FED_POLL_SECONDS = 30;
/**
* Scheduled queues that also take sends of their own: a per-account job, a
* webhook, a fan-out from the tick itself, a manual run. They keep pg-boss's
* two-second poll, because with one job per fetch a worker waits a full poll
* interval after every job, and a burst of sends would drain at one job per
* interval. `job-queue-runtime-guard.test.ts` holds this list to the send
* sites in the tree.
*/
export const SCHEDULED_AND_SENT_QUEUES: ReadonlySet<string> = new Set([
"data-backup",
"dawarich-sync",
"drain-per-sample-cumulative",
"environment-aq-history",
"environment-fetch",
"free-text-encryption-backfill",
"geolite2-fetch",
"insight-pregenerate",
"intake-slot-dedup",
"med-notes-encryption-backfill",
"nightscout-sync",
"note-encryption-backfill",
"offhost-backup-purge",
"oura-sync",
"period-narrative-warm",
"polar-sync",
"pr-detection",
"whoop-recovery-sync",
"whoop-sleep-sync",
"whoop-workout-sync",
"withings-activity-sync",
"withings-ecg-sync",
"withings-sleep-sync",
]);
/** Queues a registrar scheduled that nothing else sends to. */
const cronFedQueues = new Set<string>();
/** The poll `createAndWork` gives `queue` when its binding names none. */
export function defaultPollingSeconds(queue: string): number | undefined {
return cronFedQueues.has(queue) ? CRON_FED_POLL_SECONDS : undefined;
}
/**
* Create every queue in `queues`, then schedule every cron in `schedules`.
* Centralised so each registrar provisions before it schedules in the exact
@@ -321,6 +371,7 @@ export async function createAndSchedule(
...(schedules.length > 0 ? [PG_BOSS_SEND_IT_QUEUE] : []),
]);
for (const [name, cron, sendOptions] of schedules) {
if (!SCHEDULED_AND_SENT_QUEUES.has(name)) cronFedQueues.add(name);
await boss.schedule(
name,
cron,
@@ -366,6 +417,10 @@ export async function retireQueues(
*
* Argument order matches `boss.work(queue, options, handler)` so a binding
* reads the same as it did before, with `boss` moved to the front.
*
* A queue only its cron feeds polls at `CRON_FED_POLL_SECONDS` unless the
* binding names a poll; the registrar schedules before it binds, so the
* queue is known by then.
*/
export async function createAndWork<T>(
boss: PgBoss,
@@ -388,5 +443,13 @@ export async function createAndWork<T>(
// `JobWithMetadata<T>` extends `Job<T>`, so the metadata handler accepts
// everything the plain one does; the overloads above are what keeps the
// call sites honest about which shape they asked pg-boss for.
await boss.work<T>(queue, options, runJob(queue, handler as JobHandler<T>));
const pollingIntervalSeconds =
options.pollingIntervalSeconds ?? defaultPollingSeconds(queue);
await boss.work<T>(
queue,
pollingIntervalSeconds === undefined
? options
: { ...options, pollingIntervalSeconds },
runJob(queue, handler as JobHandler<T>),
);
}