Give the rollup queues an explicit pg-boss de-duplication policy

Every queue was created under pg-boss's default 'standard' policy, under
which no partial unique index constrains singleton_key. The singletonKey
each enqueue site passed was therefore inert and de-duplicated nothing.

Add a per-registrar policy table that records the decision and the reason
per queue. The recompute queues take 'short' (collapse only while queued,
because the handler re-reads the bucket at run time); the per-user
backfills take 'exclusive' (queued or active, because a second concurrent
run is pure duplicated work and discovery re-enqueues while work remains).

createQueue only applies a policy on first provision, so also reconcile
the stored policy for queues that already exist -- without that the tables
would be correct on a fresh database and inert on every running instance.

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<readonly string[]> {
-  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;

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<string, unknown>?];
 
+/**
+ * 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<Record<string, QueuePolicyDecision>>;
+
+/**
+ * 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<void> {
+  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<void> {
   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,