From d9e56482d97bb1b6c748fc0ca4ad136a45973849 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 03:35:09 +0200 Subject: [PATCH] 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. --- src/__tests__/job-queue-runtime-guard.test.ts | 46 +++++++++++++ .../__tests__/registrar-shared.test.ts | 62 +++++++++++++++++- src/lib/jobs/reminder/registrar-shared.ts | 65 ++++++++++++++++++- 3 files changed, 170 insertions(+), 3 deletions(-) diff --git a/src/__tests__/job-queue-runtime-guard.test.ts b/src/__tests__/job-queue-runtime-guard.test.ts index 5110a5c93..7f5adbf72 100644 --- a/src/__tests__/job-queue-runtime-guard.test.ts +++ b/src/__tests__/job-queue-runtime-guard.test.ts @@ -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): Set { + const out = new Set(); + 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); + }); +}); diff --git a/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts b/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts index 00d3998ad..fb7e471c5 100644 --- a/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts +++ b/src/lib/jobs/reminder/__tests__/registrar-shared.test.ts @@ -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, 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(); diff --git a/src/lib/jobs/reminder/registrar-shared.ts b/src/lib/jobs/reminder/registrar-shared.ts index b76474815..b180aadc3 100644 --- a/src/lib/jobs/reminder/registrar-shared.ts +++ b/src/lib/jobs/reminder/registrar-shared.ts @@ -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 = 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(); + +/** 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( boss: PgBoss, @@ -388,5 +443,13 @@ export async function createAndWork( // `JobWithMetadata` extends `Job`, 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(queue, options, runJob(queue, handler as JobHandler)); + const pollingIntervalSeconds = + options.pollingIntervalSeconds ?? defaultPollingSeconds(queue); + await boss.work( + queue, + pollingIntervalSeconds === undefined + ? options + : { ...options, pollingIntervalSeconds }, + runJob(queue, handler as JobHandler), + ); }