Merge pull request #546 from MBombeck/release/v1.30.22

release/v1.30.22 — background work stops piling up on itself

CHANGELOG.md

@@ -2,6 +2,16 @@
 
 ## [Unreleased]
 
+## [1.30.22] — 2026-07-19
+
+Background work stops piling up on itself.
+
+- **Duplicate background jobs are now actually suppressed.** Jobs were queued with a de-duplication key, but the queues were configured in a way that made that key have no effect — so the key looked right in the code and did nothing at run time. A large sync could fan out thousands of identical recompute jobs where tens were intended; a restart during a long history import could append another full import per restart; and the once-per-morning refresh had no debounce, so several paths could each trigger their own run. Each queue now carries an explicit policy chosen for what that queue does, recorded with its reason.
+- Existing installations are migrated on the next start — the setting could not be changed after a queue was created, so setting it at creation alone would have fixed nothing on any running instance. The first start after this release logs one line per migrated queue, which is how you can confirm it took effect.
+- One queue is deliberately left as it was: the explicit-range environment backfill sends without a key on purpose, and any de-duplication there would merge two different requested date ranges into one.
+
+No breaking changes.
+
 ## [1.30.21] — 2026-07-19
 
 Dose reminders land on the right hour across a clock change.

docs/api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: HealthLog API
-  version: 1.30.21
+  version: 1.30.22
   description: >-
     Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "healthlog",
-  "version": "1.30.21",
+  "version": "1.30.22",
   "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.",
   "license": "PolyForm-Noncommercial-1.0.0",
   "homepage": "https://healthlog.dev",

public/sw.js

@@ -36,7 +36,7 @@ try {
 // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild.
 const CACHE_VERSION =
   (typeof self !== "undefined" && self.__APP_VERSION__) ||
-  /* @sw-version-fallback */ "v1.30.21";
+  /* @sw-version-fallback */ "v1.30.22";
 const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`;
 const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`;
 // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*`

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);
+  });
+});

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.
  */

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<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<WithingsSyncPayload>(
     WITHINGS_SYNC_QUEUE,

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<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<DataBackupPayload>(
     DATA_BACKUP_QUEUE,

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/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:<user>:<localDate>`, 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<readonly string[]> {
-  await createAndSchedule(boss, allQueues, schedules);
+  await createAndSchedule(boss, allQueues, schedules, queuePolicies);
 
   await boss.work<GeneralStatusPayload>(
     GENERAL_STATUS_QUEUE,

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,

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()}`,
   });
 }

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);
+  });
+});