Pin the queue policy decisions with tests

Two layers. The integration suite runs a real pg-boss against a real
Postgres and proves the semantics the policies rest on: that a bare
singletonKey under the default policy de-duplicates nothing, that
exclusive suppresses a duplicate while the first job is queued AND while
it is active, that short suppresses only while queued and admits a send
once the job is active, and that a distinct key always enqueues.

It also pins the two facts that make the fix reach a running instance:
createQueue does not change the policy of a queue that already exists,
and updateQueue refuses a policy change, so the reconcile is required.

The unit guard couples each enqueue site to its registrar entry. The key
and the policy live in different files with nothing in the type system
holding them together, so deleting an entry would leave the call site
looking correct while it silently stopped working.

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