diff --git a/src/lib/jobs/__tests__/queue-policy-tables.test.ts b/src/lib/jobs/__tests__/queue-policy-tables.test.ts new file mode 100644 index 000000000..213362068 --- /dev/null +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -0,0 +1,158 @@ +/** + * Dead-policy guard, in the same spirit as the v1.4.37 dead-queue guards. + * + * A queue's `singletonKey` only de-duplicates if its registrar names a policy + * for it: pg-boss constrains `singleton_key` through partial unique indexes + * scoped by queue policy, and the default `standard` policy has none. So the + * enqueue site and the registrar entry are two halves of one guarantee that + * live in different files, and nothing in the type system couples them — + * deleting the registrar entry leaves the `singletonKey` at the call site + * looking entirely correct while it silently stops doing anything. + * + * These assertions are that coupling. Each queue below is listed with the + * policy its call site depends on; dropping or weakening an entry turns this + * suite red instead of quietly restoring the storm. + * + * The behavioural proof that the policies mean what the names say lives in + * `tests/integration/queue-singleton-policy.integration.test.ts`, which runs a + * real pg-boss against a real Postgres. + */ +import fs from "node:fs"; +import path from "node:path"; + +import { describe, expect, it } from "vitest"; + +const read = (file: string) => + fs.readFileSync(path.resolve(__dirname, "../reminder", file), "utf8"); + +/** + * Assert that `queue` appears in the registrar's `queuePolicies` table with + * `policy`. Anchored to the table so an unrelated mention of the constant + * elsewhere in the file cannot satisfy it. + */ +function expectPolicy(source: string, queue: string, policy: string) { + const table = source.match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table, "registrar has no queuePolicies table").not.toBeNull(); + const entry = new RegExp(`\\[${queue}\\]:\\s*\\{\\s*policy:\\s*"${policy}"`); + expect( + table![1], + `${queue} must carry the "${policy}" policy — without it the singletonKey at its enqueue site is inert`, + ).toMatch(entry); +} + +describe("rollup registrar queue policies", () => { + const source = read("register-rollup.ts"); + + // `short`, not `exclusive`: the handler re-reads the bucket when it starts, + // so only QUEUED duplicates may be collapsed. Promoting either of these to + // `exclusive` would strand a measurement written while a recompute ran. + it.each([ + ["ROLLUP_RECOMPUTE_QUEUE", "short"], + ["MOOD_ROLLUP_RECOMPUTE_QUEUE", "short"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + // Per-user, self-converging backfills: a second concurrent run is pure + // duplicated work and discovery re-offers the job while it is outstanding. + it.each([ + ["ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"], + ["MOOD_ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"], + ["MEDICATION_COMPLIANCE_BACKFILL_QUEUE", "exclusive"], + ["STEP_CONSOLIDATION_QUEUE", "exclusive"], + ["STEP_CONSOLIDATION_REPAIR_QUEUE", "exclusive"], + ["CUMULATIVE_PR_REDERIVE_QUEUE", "exclusive"], + ["MEAN_CONSOLIDATION_QUEUE", "exclusive"], + ["DENSE_INTRADAY_RETENTION_QUEUE", "exclusive"], + ["DENSE_INTRADAY_HOURLY_REBUILD_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); +}); + +describe("status registrar queue policies", () => { + const source = read("register-status.ts"); + + it("MORNING_DIGEST_REFRESH_QUEUE is exclusive", () => { + // Seven sleep-write seams enqueue the same user+localDate key. Without + // this the at-most-once-per-morning contract does not hold and the failure + // path re-triggers forced comprehensive generations against the provider. + expectPolicy(source, "MORNING_DIGEST_REFRESH_QUEUE", "exclusive"); + }); +}); + +describe("integration-sync registrar queue policies", () => { + const source = read("register-integration-sync.ts"); + + it.each([ + ["WHOOP_BACKFILL_QUEUE", "exclusive"], + ["FITBIT_BACKFILL_QUEUE", "exclusive"], + ["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"], + ["STRAVA_BACKFILL_QUEUE", "exclusive"], + ["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"], + ["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"], + ["LAB_BIOMARKER_BACKFILL_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); +}); + +describe("maintenance registrar queue policies", () => { + const source = read("register-maintenance.ts"); + + it.each([ + ["INTAKE_SLOT_DEDUP_QUEUE", "exclusive"], + ["NOTE_ENCRYPTION_BACKFILL_QUEUE", "exclusive"], + ["MED_NOTES_ENCRYPTION_BACKFILL_QUEUE", "exclusive"], + ["DOCUMENT_THUMBNAIL_BACKFILL_QUEUE", "exclusive"], + ["CONTENT_INDEX_BACKFILL_QUEUE", "exclusive"], + ["ENCRYPTION_KEY_ROTATE_QUEUE", "exclusive"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + // Per-document queues have no discovery pass to re-converge, so a re-process + // requested after the running job read the old bytes must be admitted. + it.each([ + ["DOCUMENT_INDEX_QUEUE", "short"], + ["DOCUMENT_THUMBNAIL_QUEUE", "short"], + ["DOCUMENT_SUMMARY_QUEUE", "short"], + ])("%s is %s", (queue, policy) => { + expectPolicy(source, queue, policy); + }); + + it("ENVIRONMENT_FETCH_QUEUE is deliberately left on standard", () => { + // Its explicit-range backfill sends with no key at all, on purpose, so any + // policy would collapse two different requested date ranges onto the shared + // empty key. This asserts the omission is intentional and documented, so a + // future reader does not "complete" the table and silently merge them. + const table = source.match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table![1]).not.toMatch(/\[ENVIRONMENT_FETCH_QUEUE\]/); + expect(source).toMatch(/ENVIRONMENT_FETCH_QUEUE is LEFT ALONE/); + }); +}); + +describe("every policy decision carries a reason", () => { + it.each([ + "register-rollup.ts", + "register-status.ts", + "register-integration-sync.ts", + "register-maintenance.ts", + ])("%s", (file) => { + const table = read(file).match( + /const queuePolicies: QueuePolicyTable = \{([\s\S]*?)\n\};/, + ); + expect(table).not.toBeNull(); + + const policies = table![1].match(/policy:\s*"(?:short|exclusive)"/g) ?? []; + const reasons = table![1].match(/reason:\s*\n?\s*"/g) ?? []; + expect(policies.length).toBeGreaterThan(0); + // A policy without a stated reason is the thing this whole change exists + // to prevent: the next person re-deriving intent from the call sites. + expect(reasons.length).toBe(policies.length); + }); +}); diff --git a/src/lib/jobs/morning-digest-refresh-shared.ts b/src/lib/jobs/morning-digest-refresh-shared.ts index a7486efd2..2540c1630 100644 --- a/src/lib/jobs/morning-digest-refresh-shared.ts +++ b/src/lib/jobs/morning-digest-refresh-shared.ts @@ -37,10 +37,18 @@ export interface MorningDigestRefreshPayload { * pre-checked in the trigger) stops any later sample from re-running it — the * pair is at-most-once per user per local morning. * + * The key alone does NOT buy that: pg-boss only constrains `singleton_key` + * when the queue carries a policy that indexes it, and under the default + * `standard` policy no such index exists, so this key coalesced nothing at all + * until the queue was given `exclusive` in `reminder/register-status.ts`. If + * that entry is ever removed, this debounce silently disappears again — the two + * belong together. + * * No `singletonSeconds`: a time-slot throttle would let a second morning's - * refresh be swallowed if it fell inside the window; the local-date-scoped key - * is already unique per morning, so the plain active-job constraint is exactly - * the debounce we want. No-ops cleanly when no global boss instance is + * refresh be swallowed if it fell inside the window, and its wall-clock + * `floor()` bucketing cannot express a user's local date anyway; the + * local-date-scoped key plus the `exclusive` policy is exactly the debounce we + * want. No-ops cleanly when no global boss instance is * available (a web process without an embedded worker) — the next sleep * landing, and the nightly cron, remain the catch-net. */ diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index d7043366f..b4f4727ca 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -302,6 +302,15 @@ export async function enqueueRollupRecompute(input: { // measurements landing on the same day shouldn't each spawn a // separate worker run. The singleton key uses bucket-start so // two writes inside the same week share one queued job. + // + // The key only does that because the queue carries the `short` policy + // (see `reminder/register-rollup.ts`); under pg-boss's default `standard` + // policy nothing indexes `singleton_key` and this key coalesced nothing, + // which is how a large sync came to fan out thousands of jobs re-deriving + // byte-identical rows. `short` collapses sends only while an identical job + // is still QUEUED — safe here precisely because the handler re-reads the + // bucket's live rows when it starts, so the queued job still sees writes + // that land after it was enqueued. singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`, }); } diff --git a/tests/integration/queue-singleton-policy.integration.test.ts b/tests/integration/queue-singleton-policy.integration.test.ts new file mode 100644 index 000000000..32e8d1117 --- /dev/null +++ b/tests/integration/queue-singleton-policy.integration.test.ts @@ -0,0 +1,281 @@ +/** + * Queue de-duplication semantics, asserted against a real pg-boss on a real + * Postgres. + * + * This suite exists because the bug it pins is invisible to a mocked test. A + * bare `singletonKey` on a `send()` LOOKS like de-duplication at every call + * site, and a unit test with a stubbed boss will happily agree. But pg-boss + * only constrains `singleton_key` through partial unique indexes that are + * scoped BY QUEUE POLICY, and under the default `standard` policy no such index + * exists — so every bare `singletonKey` in the tree de-duplicated nothing. + * Only a real insert against a real index can tell the two apart. + * + * The `standard` case below is deliberately kept as a permanent, executable + * statement of the original defect: if someone drops a policy entry from a + * registrar table, the matching assertion here goes red instead of the + * behaviour silently reverting to "the key does nothing". + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { PgBoss } from "pg-boss"; + +let boss: PgBoss; + +/** Unique per run so a re-run never collides with a queue left behind. */ +const suffix = `${Date.now()}-${Math.floor(Math.random() * 1e6)}`; +const q = (name: string) => `test-${name}-${suffix}`; + +beforeAll(async () => { + boss = new PgBoss({ + connectionString: process.env.DATABASE_URL, + // Keep the maintenance/monitor loops out of the way; this suite only + // exercises createQueue/send/fetch. + schedule: false, + supervise: false, + }); + await boss.start(); +}, 120_000); + +afterAll(async () => { + await boss?.stop({ graceful: false }); +}); + +describe("pg-boss singleton semantics by queue policy", () => { + it("standard: a bare singletonKey de-duplicates NOTHING (the original defect)", async () => { + const name = q("standard"); + await boss.createQueue(name); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + + // Both inserted. This is the finding: under the default policy the key is + // inert, so every "coalesced by singletonKey" comment in the tree was + // describing a guarantee that did not exist. + expect(first).toBeTruthy(); + expect(second).toBeTruthy(); + expect(second).not.toBe(first); + }); + + it("exclusive: suppresses a duplicate while the first is QUEUED", async () => { + const name = q("excl-queued"); + await boss.createQueue(name, { policy: "exclusive" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" }); + + expect(first).toBeTruthy(); + // pg-boss inserts with ON CONFLICT DO NOTHING ... RETURNING, so a + // suppressed send resolves to null rather than throwing. Every boot + // discovery helper in the tree counts exactly this as `skipped`. + expect(second).toBeNull(); + }); + + it("exclusive: still suppresses once the first job is ACTIVE", async () => { + const name = q("excl-active"); + await boss.createQueue(name, { policy: "exclusive" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + expect(first).toBeTruthy(); + + // Move the job out of `created` and into `active`. + const fetched = await boss.fetch(name); + expect(fetched).toHaveLength(1); + + // This is the case that matters for the backfills: a worker restarting + // during a heavy account's multi-hour pass used to append another + // identical full-history job, because `short` semantics would have let it + // through. `exclusive` covers queued OR active. + const second = await boss.send(name, { n: 2 }, { singletonKey: "user-a" }); + expect(second).toBeNull(); + }); + + it("exclusive: a genuinely distinct key still enqueues", async () => { + const name = q("excl-distinct"); + await boss.createQueue(name, { policy: "exclusive" }); + + const a = await boss.send(name, { n: 1 }, { singletonKey: "user-a" }); + const b = await boss.send(name, { n: 2 }, { singletonKey: "user-b" }); + + expect(a).toBeTruthy(); + expect(b).toBeTruthy(); + expect(b).not.toBe(a); + }); + + it("short: suppresses a duplicate while the first is QUEUED", async () => { + const name = q("short-queued"); + await boss.createQueue(name, { policy: "short" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" }); + + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("short: ADMITS a send once the first job is ACTIVE", async () => { + const name = q("short-active"); + await boss.createQueue(name, { policy: "short" }); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "bucket" }); + expect(first).toBeTruthy(); + + const fetched = await boss.fetch(name); + expect(fetched).toHaveLength(1); + + // This is the property that makes `short` the correct choice for the + // rollup recompute and the per-document jobs: the running job already read + // its input, so a write that lands afterwards MUST be able to enqueue its + // own recompute or it would be stranded in a stale bucket. `exclusive` + // would return null here and silently drop that work. + const second = await boss.send(name, { n: 2 }, { singletonKey: "bucket" }); + expect(second).toBeTruthy(); + expect(second).not.toBe(first); + }); + + it("short: a genuinely distinct key still enqueues", async () => { + const name = q("short-distinct"); + await boss.createQueue(name, { policy: "short" }); + + const a = await boss.send(name, { n: 1 }, { singletonKey: "bucket-1" }); + const b = await boss.send(name, { n: 2 }, { singletonKey: "bucket-2" }); + + expect(a).toBeTruthy(); + expect(b).toBeTruthy(); + expect(b).not.toBe(a); + }); +}); + +describe("policy reconcile for queues that already exist", () => { + it("createQueue does NOT change the policy of an existing queue", async () => { + const name = q("recreate"); + await boss.createQueue(name); + + // pg-boss's create_queue() ends in ON CONFLICT DO NOTHING, so this is a + // no-op rather than an upgrade. Without the reconcile in + // `registrar-shared.ts`, shipping the policy tables would therefore change + // nothing at all on an instance whose queues already exist -- which is + // every running instance. + await boss.createQueue(name, { policy: "exclusive" }); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("standard"); + + // And the behaviour confirms it: the key is still inert. + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeTruthy(); + }); + + it("updateQueue refuses a policy change outright", async () => { + const name = q("update"); + await boss.createQueue(name); + + await expect( + // @ts-expect-error -- policy is excluded from UpdateQueueOptions by + // design; this asserts the runtime guard that forces the direct-column + // reconcile the registrar performs. + boss.updateQueue(name, { policy: "exclusive" }), + ).rejects.toThrow(/policy cannot be changed/i); + }); + + it("writing the policy column takes effect on the next send", async () => { + const name = q("reconciled"); + await boss.createQueue(name); + + // Same statement `reconcileQueuePolicies` issues. The partial unique + // indexes for every policy already exist on the shared `job_common` table + // (this deployment does not partition per queue), so claiming a policy + // later is enough -- no schema change, no worker restart. + await boss.getDb().executeSql( + `UPDATE pgboss.queue SET policy = $2, updated_on = now() + WHERE name = $1 AND policy IS DISTINCT FROM $2`, + [name, "exclusive"], + ); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("exclusive"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); +}); + +describe("createAndSchedule applies the registrar's policy table", () => { + it("provisions a NEW queue with the decided policy", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-new"); + + await createAndSchedule(boss, [name], [], { + [name]: { policy: "exclusive", reason: "test" }, + }); + + const queue = await boss.getQueue(name); + expect(queue?.policy).toBe("exclusive"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("reconciles an EXISTING standard queue onto the decided policy", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-existing"); + + // Simulate a running instance: the queue already exists under `standard`, + // which is how every queue on every deployed instance was provisioned. + await boss.createQueue(name); + expect((await boss.getQueue(name))?.policy).toBe("standard"); + + await createAndSchedule(boss, [name], [], { + [name]: { policy: "short", reason: "test" }, + }); + + expect((await boss.getQueue(name))?.policy).toBe("short"); + + const first = await boss.send(name, { n: 1 }, { singletonKey: "same" }); + const second = await boss.send(name, { n: 2 }, { singletonKey: "same" }); + expect(first).toBeTruthy(); + expect(second).toBeNull(); + }); + + it("leaves a queue absent from the table on standard", async () => { + const { createAndSchedule } = + await import("@/lib/jobs/reminder/registrar-shared"); + const name = q("cas-absent"); + + await createAndSchedule(boss, [name], [], {}); + + expect((await boss.getQueue(name))?.policy).toBe("standard"); + }); +}); + +describe("boot-discovery skipped counter", () => { + it("can actually be non-zero once the queue carries a policy", async () => { + const name = q("skipped"); + await boss.createQueue(name, { policy: "exclusive" }); + + // Mirror the shape every boot-discovery helper uses: iterate a candidate + // set, count a null job id as `skipped`. Two of these three users are + // already represented, which under `standard` could never happen -- which + // is why `skipped` was structurally always zero in the boot log lines. + const candidates = ["u1", "u1", "u2", "u2", "u3"]; + let enqueued = 0; + let skipped = 0; + for (const userId of candidates) { + const jobId = await boss.send(name, { userId }, { singletonKey: userId }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + + expect(enqueued).toBe(3); + expect(skipped).toBe(2); + }); +});