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/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,