diff --git a/src/lib/jobs/__tests__/queue-policy-tables.test.ts b/src/lib/jobs/__tests__/queue-policy-tables.test.ts new file mode 100644 index 000000000..213362068 --- /dev/null +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -0,0 +1,158 @@ +/** + * Dead-policy guard, in the same spirit as the v1.4.37 dead-queue guards. + * + * A queue's `singletonKey` only de-duplicates if its registrar names a policy + * for it: pg-boss constrains `singleton_key` through partial unique indexes + * scoped by queue policy, and the default `standard` policy has none. So the + * enqueue site and the registrar entry are two halves of one guarantee that + * live in different files, and nothing in the type system couples them — + * deleting the registrar entry leaves the `singletonKey` at the call site + * looking entirely correct while it silently stops doing anything. + * + * These assertions are that coupling. Each queue below is listed with the + * policy its call site depends on; dropping or weakening an entry turns this + * suite red instead of quietly restoring the storm. + * + * The behavioural proof that the policies mean what the names say lives in + * `tests/integration/queue-singleton-policy.integration.test.ts`, which runs a + * real pg-boss against a real Postgres. + */ +import fs from "node:fs"; +import path from "node:path"; + +import { describe, expect, it } from "vitest"; + +const read = (file: string) => + fs.readFileSync(path.resolve(__dirname, "../reminder", file), "utf8"); + +/** + * Assert that `queue` appears in the registrar's `queuePolicies` table with + * `policy`. Anchored to the table so an unrelated mention of the constant + * elsewhere in the file cannot satisfy it. + */ +function expectPolicy(source: string, queue: string, policy: string) { + const table = source.match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table, "registrar has no queuePolicies table").not.toBeNull(); + const entry = new RegExp(`\\[${queue}\\]:\\s*\\{\\s*policy:\\s*"${policy}"`); + expect( + table![1], + `${queue} must carry the "${policy}" policy — without it the singletonKey at its enqueue site is inert`, + ).toMatch(entry); +} + +describe("rollup registrar queue policies", () => { + const source = read("register-rollup.ts"); + + // `short`, not `exclusive`: the handler re-reads the bucket when it starts, + // so only QUEUED duplicates may be collapsed. Promoting either of these to + // `exclusive` would strand a measurement written while a recompute ran. + it.each([ + ["ROLLUP_RECOMPUTE_QUEUE", "short"], + ["MOOD_ROLLUP_RECOMPUTE_QUEUE", "short"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + // Per-user, self-converging backfills: a second concurrent run is pure + // duplicated work and discovery re-offers the job while it is outstanding. + it.each([ + ["ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"], + ["MOOD_ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"], + ["MEDICATION_COMPLIANCE_BACKFILL_QUEUE", "exclusive"], + ["STEP_CONSOLIDATION_QUEUE", "exclusive"], + ["STEP_CONSOLIDATION_REPAIR_QUEUE", "exclusive"], + ["CUMULATIVE_PR_REDERIVE_QUEUE", "exclusive"], + ["MEAN_CONSOLIDATION_QUEUE", "exclusive"], + ["DENSE_INTRADAY_RETENTION_QUEUE", "exclusive"], + ["DENSE_INTRADAY_HOURLY_REBUILD_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); +}); + +describe("status registrar queue policies", () => { + const source = read("register-status.ts"); + + it("MORNING_DIGEST_REFRESH_QUEUE is exclusive", () => { + // Seven sleep-write seams enqueue the same user+localDate key. Without + // this the at-most-once-per-morning contract does not hold and the failure + // path re-triggers forced comprehensive generations against the provider. + expectPolicy(source, "MORNING_DIGEST_REFRESH_QUEUE", "exclusive"); + }); +}); + +describe("integration-sync registrar queue policies", () => { + const source = read("register-integration-sync.ts"); + + it.each([ + ["WHOOP_BACKFILL_QUEUE", "exclusive"], + ["FITBIT_BACKFILL_QUEUE", "exclusive"], + ["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"], + ["STRAVA_BACKFILL_QUEUE", "exclusive"], + ["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"], + ["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"], + ["LAB_BIOMARKER_BACKFILL_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); +}); + +describe("maintenance registrar queue policies", () => { + const source = read("register-maintenance.ts"); + + it.each([ + ["INTAKE_SLOT_DEDUP_QUEUE", "exclusive"], + ["NOTE_ENCRYPTION_BACKFILL_QUEUE", "exclusive"], + ["MED_NOTES_ENCRYPTION_BACKFILL_QUEUE", "exclusive"], + ["DOCUMENT_THUMBNAIL_BACKFILL_QUEUE", "exclusive"], + ["CONTENT_INDEX_BACKFILL_QUEUE", "exclusive"], + ["ENCRYPTION_KEY_ROTATE_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + // Per-document queues have no discovery pass to re-converge, so a re-process + // requested after the running job read the old bytes must be admitted. + it.each([ + ["DOCUMENT_INDEX_QUEUE", "short"], + ["DOCUMENT_THUMBNAIL_QUEUE", "short"], + ["DOCUMENT_SUMMARY_QUEUE", "short"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + it("ENVIRONMENT_FETCH_QUEUE is deliberately left on standard", () => { + // Its explicit-range backfill sends with no key at all, on purpose, so any + // policy would collapse two different requested date ranges onto the shared + // empty key. This asserts the omission is intentional and documented, so a + // future reader does not "complete" the table and silently merge them. + const table = source.match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table![1]).not.toMatch(/\[ENVIRONMENT_FETCH_QUEUE\]/); + expect(source).toMatch(/ENVIRONMENT_FETCH_QUEUE is LEFT ALONE/); + }); +}); + +describe("every policy decision carries a reason", () => { + it.each([ + "register-rollup.ts", + "register-status.ts", + "register-integration-sync.ts", + "register-maintenance.ts", + ])("%s", (file) => { + const table = read(file).match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table).not.toBeNull(); + + const policies = table![1].match(/policy:\s*"(?:short|exclusive)"/g) ?? []; + const reasons = table![1].match(/reason:\s*\n?\s*"/g) ?? []; + expect(policies.length).toBeGreaterThan(0); + // A policy without a stated reason is the thing this whole change exists + // to prevent: the next person re-deriving intent from the call sites. + expect(reasons.length).toBe(policies.length); + }); +}); diff --git a/src/lib/jobs/morning-digest-refresh-shared.ts b/src/lib/jobs/morning-digest-refresh-shared.ts index a7486efd2..2540c1630 100644 --- a/src/lib/jobs/morning-digest-refresh-shared.ts +++ b/src/lib/jobs/morning-digest-refresh-shared.ts @@ -37,10 +37,18 @@ export interface MorningDigestRefreshPayload { * pre-checked in the trigger) stops any later sample from re-running it — the * pair is at-most-once per user per local morning. * + * The key alone does NOT buy that: pg-boss only constrains `singleton_key` + * when the queue carries a policy that indexes it, and under the default + * `standard` policy no such index exists, so this key coalesced nothing at all + * until the queue was given `exclusive` in `reminder/register-status.ts`. If + * that entry is ever removed, this debounce silently disappears again — the two + * belong together. + * * No `singletonSeconds`: a time-slot throttle would let a second morning's - * refresh be swallowed if it fell inside the window; the local-date-scoped key - * is already unique per morning, so the plain active-job constraint is exactly - * the debounce we want. No-ops cleanly when no global boss instance is + * refresh be swallowed if it fell inside the window, and its wall-clock + * `floor()` bucketing cannot express a user's local date anyway; the + * local-date-scoped key plus the `exclusive` policy is exactly the debounce we + * want. No-ops cleanly when no global boss instance is * available (a web process without an embedded worker) — the next sleep * landing, and the nightly cron, remain the catch-net. */ diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index a3bde40a8..1d3256e17 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -64,7 +64,11 @@ import { type LabBiomarkerBackfillPayload, } from "@/lib/jobs/lab-biomarker-backfill"; import { workerLog } from "./shared"; -import { createAndSchedule, type ScheduleEntry } from "./registrar-shared"; +import { + createAndSchedule, + type QueuePolicyTable, + type ScheduleEntry, +} from "./registrar-shared"; import { WithingsSyncPayload, WithingsActivitySyncPayload, @@ -326,6 +330,62 @@ const schedules: ScheduleEntry[] = [ [OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, OIDC_NATIVE_HANDOFF_CLEANUP_CRON], ]; +/** + * De-duplication policy per integration-sync queue. + * + * The recurring poll queues (Withings / WHOOP / Fitbit / Google Health / + * Nightscout / Polar / Oura / Strava sync, and every OAuth-state cleanup) are + * keyless cron ticks — no `singletonKey` on any send — so a policy would only + * constrain the empty key. They are deliberately absent. + * + * The one-shot backfills and repairs below all share the same shape: one job + * per connection, keyed per user (per provider where a provider can differ), + * enqueued by a boot-discovery predicate that only stops offering the job once + * it has COMPLETED. That last property is what made the bug bite — a worker + * restarting during a heavy account's multi-hour full-history pass appended + * another identical job per restart, sustaining the boot storm the stagger + * deferrals exist to damp. `exclusive` covers queued, active, and retry-backoff, + * so the restart case is closed; and because discovery re-offers the job while + * the work is still outstanding, a suppressed send can never lose work. + */ +const queuePolicies: QueuePolicyTable = { + [WHOOP_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection full-history backfill; long-running, and boot discovery re-enqueues until the connection is marked backfilled.", + }, + [FITBIT_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection full-history backfill; boot-discovery driven and self-converging.", + }, + [GOOGLE_HEALTH_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection full-history backfill; boot-discovery driven and self-converging.", + }, + [STRAVA_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection full-history backfill; boot-discovery driven and self-converging.", + }, + [GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.", + }, + [SLEEP_TIMELINE_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-connection one-shot timeline backfill, keyed per provider+user; discovery drops the connection once its rows carry the new shape.", + }, + [LAB_BIOMARKER_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user one-shot catalog link-up; discovery drops the user once no unlinked lab readings remain.", + }, +}; + /** * Register every integration-sync queue: create, schedule, and bind handlers. * Returns the queue names created (for the boot-level aggregate assertion). @@ -333,7 +393,7 @@ const schedules: ScheduleEntry[] = [ export async function registerIntegrationSyncQueues( boss: PgBoss, ): Promise { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( WITHINGS_SYNC_QUEUE, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 62053f2c2..f7950400f 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -115,7 +115,11 @@ import { } from "@/lib/jobs/environment-fetch"; import { recordError } from "@/lib/jobs/worker-status"; import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS } from "./shared"; -import { createAndSchedule, type ScheduleEntry } from "./registrar-shared"; +import { + createAndSchedule, + type QueuePolicyTable, + type ScheduleEntry, +} from "./registrar-shared"; import { DataBackupPayload, OffhostBackupPayload, @@ -425,6 +429,86 @@ const schedules: ScheduleEntry[] = [ [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON], ]; +/** + * De-duplication policy per maintenance queue. + * + * Every retention / cleanup cron here (rate-limit, idempotency, audit-log, + * push-attempt, tombstone, coach-message, MCP-token, document-purge, …) is a + * keyless tick with no `singletonKey` on any send, so a policy would only + * constrain the empty key. Those are deliberately absent from this table. + * + * Two deliberate omissions worth naming, because both look like candidates: + * + * - ENVIRONMENT_FETCH_QUEUE is LEFT ALONE. It is genuinely ambiguous: the + * lookback refresh sends a per-user key, but an explicit-range backfill + * sends with NO options at all, deliberately, so that "it always runs". + * Any policy would collapse every keyless explicit-range backfill onto the + * shared empty key, so two different requested date ranges would silently + * become one. That is exactly the class of silent work-dropping a policy is + * supposed to prevent, so the queue keeps `standard` until the enqueue side + * gives the explicit-range variant a key of its own. + * - APPLE_HEALTH_IMPORT_QUEUE, DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send + * keylessly by design (each import / backup / detection run is a distinct + * unit of work). A policy would coalesce independent runs. + */ +const queuePolicies: QueuePolicyTable = { + // Per-user, boot- and cron-discovery driven, self-converging one-shots. The + // discovery predicate re-offers the job while the work is outstanding, so + // suppressing a duplicate of a queued OR active pass cannot lose work. + [INTAKE_SLOT_DEDUP_QUEUE]: { + policy: "exclusive", + reason: + "Per-user duplicate-slot cleanup; discovery drops the user once no colliding intake pair remains.", + }, + [NOTE_ENCRYPTION_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user note-encryption backfill; discovery drops the user once no plaintext note remains.", + }, + [MED_NOTES_ENCRYPTION_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user medication-note encryption backfill; discovery drops the user once no plaintext note remains.", + }, + [DOCUMENT_THUMBNAIL_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user thumbnail backfill; discovery drops the user once every thumbnailable document has a preview.", + }, + [CONTENT_INDEX_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user index-all run, triggered from the documents surface. Idempotent (it indexes only not-yet-indexed documents), so a second press while one runs is correctly a no-op.", + }, + [ENCRYPTION_KEY_ROTATE_QUEUE]: { + policy: "exclusive", + reason: + "Fixed singleton key, admin-triggered. Two concurrent corpus rotations must never overlap; the route already reports a suppressed send back as alreadyQueued.", + }, + + // Per-document, enqueued on upload. `short`, NOT `exclusive`: each handler + // re-reads the document when it starts, so collapsing sends that arrive while + // an identical job is still queued is safe. `exclusive` would additionally + // suppress a re-process requested after the current job had already read the + // old bytes, leaving a stale summary/thumbnail/index behind with no discovery + // pass to re-offer it — these queues have no cron to re-converge. + [DOCUMENT_INDEX_QUEUE]: { + policy: "short", + reason: + "Per-document, no re-converging discovery pass. Collapse queued duplicates only, so a re-index after a content change is never dropped.", + }, + [DOCUMENT_THUMBNAIL_QUEUE]: { + policy: "short", + reason: + "Per-document, no re-converging discovery pass. Collapse queued duplicates only.", + }, + [DOCUMENT_SUMMARY_QUEUE]: { + policy: "short", + reason: + "Per-document, no re-converging discovery pass. Collapse queued duplicates only.", + }, +}; + /** * Register every maintenance / ops / cleanup queue. Returns the queue names * created (for the boot-level aggregate assertion). @@ -432,7 +516,7 @@ const schedules: ScheduleEntry[] = [ export async function registerMaintenanceQueues( boss: PgBoss, ): Promise { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( DATA_BACKUP_QUEUE, diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index ce03e656c..2c51a2c10 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -102,7 +102,11 @@ import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS, } from "./shared"; -import { createAndSchedule, type ScheduleEntry } from "./registrar-shared"; +import { + createAndSchedule, + type QueuePolicyTable, + type ScheduleEntry, +} from "./registrar-shared"; // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows. // Collapses each user × cumulative-type × calendar-day bucket into one // `stats:…` row so the list view stops painting hundreds of step chunks per @@ -169,6 +173,93 @@ const schedules: ScheduleEntry[] = [ [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON], ]; +/** + * De-duplication policy per rollup queue. Until this table existed every queue + * here ran under pg-boss's `standard` policy, under which no unique index + * constrains `singleton_key` — so the `singletonKey` each enqueue site passes + * was inert and de-duplicated nothing. + * + * The split below is deliberate and the two halves are NOT interchangeable. + */ +const queuePolicies: QueuePolicyTable = { + // `short`, not `exclusive`. A recompute reads the bucket's live rows when it + // STARTS, so collapsing sends that pile up while an identical job is still + // queued is safe — the queued job will observe every write that landed in the + // meantime. This is what kills the storm: a sync posting a few hundred batch + // requests fanned out three jobs per type-day per request, thousands of jobs + // re-deriving byte-identical rows. Under `short` they collapse to one queued + // job per bucket. `exclusive` would be wrong here: it would also suppress a + // send issued after the recompute had already started reading, stranding the + // newer measurement in a stale bucket until the nightly pass. + [ROLLUP_RECOMPUTE_QUEUE]: { + policy: "short", + reason: + "Burst coalescing for the write-path fan-out. The handler re-reads the bucket at run time, so collapsing queued duplicates cannot strand a write; exclusive could.", + }, + [MOOD_ROLLUP_RECOMPUTE_QUEUE]: { + policy: "short", + reason: + "Same shape as the measurement recompute: re-reads the bucket at run time, so only queued duplicates may be collapsed.", + }, + + // The rest are per-user, self-converging boot/discovery backfills. A second + // concurrent run is pure duplicated work, and the discovery predicate + // re-enqueues on the next boot or cron tick while work remains outstanding — + // so `exclusive` (queued OR active OR awaiting retry) can never lose work it + // does not re-offer. `short` would be too weak: these passes run for minutes + // to hours, and the observed failure was precisely a worker restarting + // mid-pass and appending another identical full-history job per restart. + [ROLLUP_FULL_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user full-history backfill; long-running. Boot discovery re-enqueues while coverage is still missing, so suppressing a duplicate of an ACTIVE pass is safe.", + }, + [MOOD_ROLLUP_FULL_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user mood backfill; boot-discovery driven and self-converging on rollup coverage.", + }, + [MEDICATION_COMPLIANCE_BACKFILL_QUEUE]: { + policy: "exclusive", + reason: + "Per-user compliance backfill; boot-discovery driven and self-converging on rollup coverage.", + }, + [STEP_CONSOLIDATION_QUEUE]: { + policy: "exclusive", + reason: + "Per-user legacy step consolidation; discovery drops the user once no live granular rows remain.", + }, + [STEP_CONSOLIDATION_REPAIR_QUEUE]: { + policy: "exclusive", + reason: + "Per-user one-shot repair; discovery drops the user once the affected provider rows are repaired.", + }, + [CUMULATIVE_PR_REDERIVE_QUEUE]: { + policy: "exclusive", + reason: + "Per-user one-shot re-derivation; discovery drops the user once the inflated records are rewritten.", + }, + [MEAN_CONSOLIDATION_QUEUE]: { + policy: "exclusive", + reason: + "Per-user daily-mean consolidation; discovery drops the user once the spot rows are folded.", + }, + [DENSE_INTRADAY_RETENTION_QUEUE]: { + policy: "exclusive", + reason: + "Per-user dense-tier drain; the multi-hour pass on a heavy account is exactly the case where a restart used to append a duplicate full-history job.", + }, + [DENSE_INTRADAY_HOURLY_REBUILD_QUEUE]: { + policy: "exclusive", + reason: + "Per-user one-shot hourly rebuild; discovery drops the day once its retired daily row is gone.", + }, + + // DRAIN_CUMULATIVE_QUEUE is deliberately absent: it is a keyless nightly cron + // tick with no singletonKey on any send, so a policy would only constrain the + // empty key and buy nothing. +}; + /** * Register every rollup / consolidation queue. Returns the queue names created * (for the boot-level aggregate assertion). @@ -176,7 +267,7 @@ const schedules: ScheduleEntry[] = [ export async function registerRollupQueues( boss: PgBoss, ): Promise { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); // v1.5.0 — persistent measurement rollup worker. Folds the // WEEK / MONTH / YEAR buckets that the write-path hooks enqueue; diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index 315ed0eaf..9b49e8e34 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/src/lib/jobs/reminder/register-status.ts @@ -92,6 +92,7 @@ import { getWorkerPrisma, workerLog } from "./shared"; import { createAndSchedule, insightRetryOptions, + type QueuePolicyTable, type ScheduleEntry, } from "./registrar-shared"; import { @@ -261,6 +262,40 @@ const schedules: ScheduleEntry[] = [ [DAILY_BRIEFING_QUEUE, DAILY_BRIEFING_CRON], ]; +/** + * De-duplication policy per status queue. + * + * Most queues here are keyless nightly cron ticks — no `singletonKey` on any + * send — so a policy would only constrain the empty key and is deliberately + * omitted. The LLM-bound warm queues (insight pre-generate, per-metric status + * generate, period narrative, Coach memory refresh) already pass an explicit + * `singletonSeconds` on every send, which populates `singleton_on` and is + * therefore constrained by pg-boss's `job_i4` index REGARDLESS of queue policy. + * Those already de-duplicate today; changing their policy would layer a second, + * differently-shaped constraint over a working one for no gain, so they are + * left alone. + * + * That leaves exactly one queue here whose key is inert today. + */ +const queuePolicies: QueuePolicyTable = { + // The sleep-arrival trigger has seven write seams that can each enqueue for + // the same user and the same local date. The key is + // `morning-refresh::`, and `exclusive` turns that into a + // real at-most-once-per-user-per-local-morning contract across queued, + // active, and retry-backoff states. + // + // `singletonSeconds` is NOT the tool here and the existing comment at the + // enqueue site says so: its window is a floor() over wall-clock, which does + // not line up with a user's local date, so it would both split one morning + // across two buckets and merge two mornings into one. The local date already + // lives in the key; the policy just has to honour it. + [MORNING_DIGEST_REFRESH_QUEUE]: { + policy: "exclusive", + reason: + "Seven sleep-write seams enqueue the same user+localDate key; exclusive makes the at-most-once-per-morning contract real. A wall-clock singletonSeconds window cannot express a local date.", + }, +}; + /** * Register every status / insight / computed-score queue. Returns the queue * names created (for the boot-level aggregate assertion). @@ -268,7 +303,7 @@ const schedules: ScheduleEntry[] = [ export async function registerStatusQueues( boss: PgBoss, ): Promise { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( GENERAL_STATUS_QUEUE, diff --git a/src/lib/jobs/reminder/registrar-shared.ts b/src/lib/jobs/reminder/registrar-shared.ts index ed7f77b62..bb0e11eae 100644 --- a/src/lib/jobs/reminder/registrar-shared.ts +++ b/src/lib/jobs/reminder/registrar-shared.ts @@ -18,6 +18,10 @@ */ import { PgBoss } from "pg-boss"; +import { prisma } from "@/lib/db"; + +import { workerLog } from "./shared"; + /** * Cron schedule tuple: `[queueName, cronExpression, sendOptions?]`. The * optional third element carries per-queue send options (e.g. the LLM-bound @@ -25,6 +29,102 @@ import { PgBoss } from "pg-boss"; */ export type ScheduleEntry = [string, string, Record?]; +/** + * The pg-boss queue policies this codebase names explicitly. `standard` — + * pg-boss's default — is deliberately absent: under `standard` NO partial + * unique index covers `singleton_key` at all, so a bare `singletonKey` on a + * `send()` is inert and de-duplicates nothing. A queue that wants + * de-duplication has to name one of the two policies below. + * + * From the partial unique indexes pg-boss 12.26 creates on the shared + * `job_common` table: + * + * - `short` → UNIQUE (name, COALESCE(singleton_key,'')) WHERE state = 'created' + * Collapses duplicate sends only while a job is still QUEUED. + * Once it goes active, a fresh send is admitted again. + * - `exclusive` → UNIQUE (name, COALESCE(singleton_key,'')) WHERE state <= 'active' + * Collapses duplicate sends while a job is queued OR active + * OR waiting out a retry backoff. + * + * Picking between them is a correctness question, not a taste question: + * + * - `short` when the handler re-reads current state at run time. Suppressing + * a send while an identical job is still QUEUED is then provably safe — + * that job has read nothing yet and will observe the newer write when it + * starts. Suppressing a send after the reader already STARTED would strand + * the newer write, which is why these queues must not be `exclusive`. + * - `exclusive` when a second concurrent run is pure duplicated work AND the + * enqueue side is self-converging — a discovery pass that re-enqueues on + * the next boot or cron tick while the work is still outstanding. + * + * On a conflict pg-boss inserts with `ON CONFLICT DO NOTHING ... RETURNING`, so + * `boss.send()` resolves to `null` instead of throwing. Every enqueue helper in + * this tree already counts a null id as `skipped`. + */ +export type QueuePolicy = "short" | "exclusive"; + +/** + * A per-queue policy decision plus the reason behind it. The reason is not + * decoration: the policy encodes a claim about whether the handler re-reads + * state at run time and whether the enqueue side re-converges. Anyone changing + * a handler needs that claim written down rather than re-derived from the call + * sites. + */ +export type QueuePolicyDecision = { + policy: QueuePolicy; + reason: string; +}; + +/** Queue name → decision. Queues absent from a table keep pg-boss's `standard`. */ +export type QueuePolicyTable = Readonly>; + +/** + * Bring the policy of ALREADY-EXISTING queues in line with the tables below. + * + * Why this is needed at all: pg-boss's `create_queue()` SQL function ends in + * `ON CONFLICT DO NOTHING`, so passing `{ policy }` to `boss.createQueue()` + * only takes effect the first time a queue is provisioned. Every queue on an + * already-running instance was created under the default `standard` policy, and + * `boss.updateQueue()` rejects a policy change outright ("queue policy cannot be + * changed after creation"). Without this reconcile the policy tables would be + * correct on a fresh database and completely inert on every existing + * deployment — a change that tests green and fixes nothing. + * + * Why writing the column directly is sound here: the partial unique indexes for + * every policy are created once on the shared `job_common` table at schema + * setup (this deployment does not use per-queue table partitioning), so the + * index a newly-claimed policy needs already exists. A job row's own `policy` + * value is resolved by joining `pgboss.queue` at insert time, so a changed + * column takes effect on the next `send()` without a worker restart and without + * depending on pg-boss's in-process queue cache. Jobs already in flight keep + * the policy they were inserted under and simply do not participate in the new + * index, which is the correct transitional behaviour. + * + * Scope is deliberately narrow: only queue names this codebase decided on, only + * where the stored policy actually differs, parameter-bound, and never a change + * to a queue absent from the tables. + */ +async function reconcileQueuePolicies( + policies: QueuePolicyTable, +): Promise { + for (const [name, { policy }] of Object.entries(policies)) { + try { + const changed = await prisma.$executeRaw` + UPDATE pgboss.queue + SET policy = ${policy}, updated_on = now() + WHERE name = ${name} AND policy IS DISTINCT FROM ${policy} + `; + if (changed > 0) { + workerLog("info", `[queue-policy] reconciled ${name} to ${policy}`); + } + } catch (err) { + // Never fail worker boot on a reconcile miss: the queue still works, it + // just keeps whatever de-duplication semantics it already had. + workerLog("error", `[queue-policy] failed to reconcile ${name}`, err); + } + } +} + /** * Shared retry policy for the LLM-bound insight queues. A transient failure * (provider hiccup, pool exhaustion) used to fail the nightly tick silently @@ -42,15 +142,25 @@ export const insightRetryOptions = { * Centralised so each registrar provisions before it schedules in the exact * order the monolith did, and the `Europe/Berlin` tz default stays in one * place. + * + * `policies` carries the registrar's per-queue de-duplication decisions. It is + * applied on both legs — as the creation policy for a queue that does not exist + * yet, and through `reconcileQueuePolicies` for one that does. A queue absent + * from the table keeps pg-boss's `standard` policy, which means no + * de-duplication at all; that is a valid choice, but it must be a deliberate + * one, so the tables document the omissions too. */ export async function createAndSchedule( boss: PgBoss, queues: readonly string[], schedules: readonly ScheduleEntry[], + policies: QueuePolicyTable = {}, ): Promise { for (const q of queues) { - await boss.createQueue(q); + const decision = policies[q]; + await boss.createQueue(q, decision ? { policy: decision.policy } : {}); } + await reconcileQueuePolicies(policies); for (const [name, cron, sendOptions] of schedules) { await boss.schedule( name, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index d7043366f..b4f4727ca 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -302,6 +302,15 @@ export async function enqueueRollupRecompute(input: { // measurements landing on the same day shouldn't each spawn a // separate worker run. The singleton key uses bucket-start so // two writes inside the same week share one queued job. + // + // The key only does that because the queue carries the `short` policy + // (see `reminder/register-rollup.ts`); under pg-boss's default `standard` + // policy nothing indexes `singleton_key` and this key coalesced nothing, + // which is how a large sync came to fan out thousands of jobs re-deriving + // byte-identical rows. `short` collapses sends only while an identical job + // is still QUEUED — safe here precisely because the handler re-reads the + // bucket's live rows when it starts, so the queued job still sees writes + // that land after it was enqueued. singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`, }); } diff --git a/tests/integration/queue-singleton-policy.integration.test.ts b/tests/integration/queue-singleton-policy.integration.test.ts new file mode 100644 index 000000000..32e8d1117 --- /dev/null +++ b/tests/integration/queue-singleton-policy.integration.test.ts @@ -0,0 +1,281 @@ +/** + * Queue de-duplication semantics, asserted against a real pg-boss on a real + * Postgres. + * + * This suite exists because the bug it pins is invisible to a mocked test. A + * bare `singletonKey` on a `send()` LOOKS like de-duplication at every call + * site, and a unit test with a stubbed boss will happily agree. But pg-boss + * only constrains `singleton_key` through partial unique indexes that are + * scoped BY QUEUE POLICY, and under the default `standard` policy no such index + * exists — so every bare `singletonKey` in the tree de-duplicated nothing. + * Only a real insert against a real index can tell the two apart. + * + * The `standard` case below is deliberately kept as a permanent, executable + * statement of the original defect: if someone drops a policy entry from a + * registrar table, the matching assertion here goes red instead of the + * behaviour silently reverting to "the key does nothing". + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { PgBoss } from "pg-boss"; + +let boss: PgBoss; + +/** Unique per run so a re-run never collides with a queue left behind. */ +const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`; +const q = (name: string) => `test-${name}-${suffix}`; + +beforeAll(async () => { + boss = new PgBoss({ + connectionString: process.env.DATABASE_URL, + // Keep the maintenance/monitor loops out of the way; this suite only + // exercises createQueue/send/fetch. + schedule: false, + supervise: false, + }); + await boss.start(); +}, 120_000); + +afterAll(async () => { + await boss?.stop({ graceful: false }); +}); + +describe("pg-boss singleton semantics by queue policy", () => { + it("standard: a bare singletonKey de-duplicates NOTHING (the original defect)", async () => { + const name = q("standard"); + await boss.createQueue(name); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + + // Both inserted. This is the finding: under the default policy the key is + // inert, so every "coalesced by singletonKey" comment in the tree was + // describing a guarantee that did not exist. + expect(first).toBeTruthy(); + expect(second).toBeTruthy(); + expect(second).not.toBe(first); + }); + + it("exclusive: suppresses a duplicate while the first is QUEUED", async () => { + const name = q("excl-queued"); + await boss.createQueue(name, { policy: "exclusive" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" }); + + expect(first).toBeTruthy(); + // pg-boss inserts with ON CONFLICT DO NOTHING ... RETURNING, so a + // suppressed send resolves to null rather than throwing. Every boot + // discovery helper in the tree counts exactly this as `skipped`. + expect(second).toBeNull(); + }); + + it("exclusive: still suppresses once the first job is ACTIVE", async () => { + const name = q("excl-active"); + await boss.createQueue(name, { policy: "exclusive" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + expect(first).toBeTruthy(); + + // Move the job out of `created` and into `active`. + const fetched = await boss.fetch(name); + expect(fetched).toHaveLength(1); + + // This is the case that matters for the backfills: a worker restarting + // during a heavy account's multi-hour pass used to append another + // identical full-history job, because `short` semantics would have let it + // through. `exclusive` covers queued OR active. + const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" }); + expect(second).toBeNull(); + }); + + it("exclusive: a genuinely distinct key still enqueues", async () => { + const name = q("excl-distinct"); + await boss.createQueue(name, { policy: "exclusive" }); + + const a = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + const b = await boss.send(name, { n: 2 }, { singletonKey: "user-b" }); + + expect(a).toBeTruthy(); + expect(b).toBeTruthy(); + expect(b).not.toBe(a); + }); + + it("short: suppresses a duplicate while the first is QUEUED", async () => { + const name = q("short-queued"); + await boss.createQueue(name, { policy: "short" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" }); + + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("short: ADMITS a send once the first job is ACTIVE", async () => { + const name = q("short-active"); + await boss.createQueue(name, { policy: "short" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" }); + expect(first).toBeTruthy(); + + const fetched = await boss.fetch(name); + expect(fetched).toHaveLength(1); + + // This is the property that makes `short` the correct choice for the + // rollup recompute and the per-document jobs: the running job already read + // its input, so a write that lands afterwards MUST be able to enqueue its + // own recompute or it would be stranded in a stale bucket. `exclusive` + // would return null here and silently drop that work. + const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" }); + expect(second).toBeTruthy(); + expect(second).not.toBe(first); + }); + + it("short: a genuinely distinct key still enqueues", async () => { + const name = q("short-distinct"); + await boss.createQueue(name, { policy: "short" }); + + const a = await boss.send(name, { n: 1 }, { singletonKey: "bucket-1" }); + const b = await boss.send(name, { n: 2 }, { singletonKey: "bucket-2" }); + + expect(a).toBeTruthy(); + expect(b).toBeTruthy(); + expect(b).not.toBe(a); + }); +}); + +describe("policy reconcile for queues that already exist", () => { + it("createQueue does NOT change the policy of an existing queue", async () => { + const name = q("recreate"); + await boss.createQueue(name); + + // pg-boss's create_queue() ends in ON CONFLICT DO NOTHING, so this is a + // no-op rather than an upgrade. Without the reconcile in + // `registrar-shared.ts`, shipping the policy tables would therefore change + // nothing at all on an instance whose queues already exist -- which is + // every running instance. + await boss.createQueue(name, { policy: "exclusive" }); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("standard"); + + // And the behaviour confirms it: the key is still inert. + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeTruthy(); + }); + + it("updateQueue refuses a policy change outright", async () => { + const name = q("update"); + await boss.createQueue(name); + + await expect( + // @ts-expect-error -- policy is excluded from UpdateQueueOptions by + // design; this asserts the runtime guard that forces the direct-column + // reconcile the registrar performs. + boss.updateQueue(name, { policy: "exclusive" }), + ).rejects.toThrow(/policy cannot be changed/i); + }); + + it("writing the policy column takes effect on the next send", async () => { + const name = q("reconciled"); + await boss.createQueue(name); + + // Same statement `reconcileQueuePolicies` issues. The partial unique + // indexes for every policy already exist on the shared `job_common` table + // (this deployment does not partition per queue), so claiming a policy + // later is enough -- no schema change, no worker restart. + await boss.getDb().executeSql( + `UPDATE pgboss.queue SET policy = $2, updated_on = now() + WHERE name = $1 AND policy IS DISTINCT FROM $2`, + [name, "exclusive"], + ); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("exclusive"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); +}); + +describe("createAndSchedule applies the registrar's policy table", () => { + it("provisions a NEW queue with the decided policy", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-new"); + + await createAndSchedule(boss, [name], [], { + [name]: { policy: "exclusive", reason: "test" }, + }); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("exclusive"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("reconciles an EXISTING standard queue onto the decided policy", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-existing"); + + // Simulate a running instance: the queue already exists under `standard`, + // which is how every queue on every deployed instance was provisioned. + await boss.createQueue(name); + expect((await boss.getQueue(name))?.policy).toBe("standard"); + + await createAndSchedule(boss, [name], [], { + [name]: { policy: "short", reason: "test" }, + }); + + expect((await boss.getQueue(name))?.policy).toBe("short"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("leaves a queue absent from the table on standard", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-absent"); + + await createAndSchedule(boss, [name], [], {}); + + expect((await boss.getQueue(name))?.policy).toBe("standard"); + }); +}); + +describe("boot-discovery skipped counter", () => { + it("can actually be non-zero once the queue carries a policy", async () => { + const name = q("skipped"); + await boss.createQueue(name, { policy: "exclusive" }); + + // Mirror the shape every boot-discovery helper uses: iterate a candidate + // set, count a null job id as `skipped`. Two of these three users are + // already represented, which under `standard` could never happen -- which + // is why `skipped` was structurally always zero in the boot log lines. + const candidates = ["u1", "u1", "u2", "u2", "u3"]; + let enqueued = 0; + let skipped = 0; + for (const userId of candidates) { + const jobId = await boss.send(name, { userId }, { singletonKey: userId }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + + expect(enqueued).toBe(3); + expect(skipped).toBe(2); + }); +});