From f2083eba3629599efa0507dd2b07b76404962274 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 19 Jul 2026 02:17:57 +0200 Subject: [PATCH] Give the status, integration-sync and maintenance queues explicit policies The per-connection backfills and per-user one-shots take 'exclusive': a worker restarting during a heavy account's multi-hour pass used to append another identical full-history job per restart, because the discovery predicate only stops offering a job once it has completed. The morning-digest refresh takes 'exclusive' so its user+localDate key becomes a real at-most-once-per-morning contract; a wall-clock singletonSeconds window cannot express a local date. The per-document queues take 'short' -- they have no discovery pass to re-converge, so only queued duplicates may be collapsed or a re-process after a content change would be lost. Leave the environment-fetch queue on 'standard' and say why: its explicit-range backfill deliberately sends without a key, so any policy would silently merge two different requested ranges into one. --- .../reminder/register-integration-sync.ts | 64 +++++++++++++- src/lib/jobs/reminder/register-maintenance.ts | 88 ++++++++++++++++++- src/lib/jobs/reminder/register-status.ts | 37 +++++++- 3 files changed, 184 insertions(+), 5 deletions(-) diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index a3bde40a8..1d3256e17 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/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 { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( WITHINGS_SYNC_QUEUE, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 62053f2c2..f7950400f 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/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 { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( DATA_BACKUP_QUEUE, diff --git a/src/lib/jobs/reminder/register-status.ts b/src/lib/jobs/reminder/register-status.ts index 315ed0eaf..9b49e8e34 100644 --- a/src/lib/jobs/reminder/register-status.ts +++ b/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::`, 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 { - await createAndSchedule(boss, allQueues, schedules); + await createAndSchedule(boss, allQueues, schedules, queuePolicies); await boss.work( GENERAL_STATUS_QUEUE,