fix(jobs): let the retention purges finish, and let them fail loudly

The nightly tombstone purge asked for `deleted_at IS NOT NULL AND deleted_at
< cutoff` and the audit-log purge asked for `created_at < cutoff`. Neither
predicate had an index behind it: every deleted_at index in the schema is
partial on IS NULL, the live-rows predicate, and every audit_logs index leads
with user_id or action. Both purges planned a sequential scan and then paid
index maintenance per removed row, in one unbounded statement, on a
connection carrying statement_timeout=60000. Past a certain backlog the
statement aborts, the transaction rolls back, nothing is purged, and the next
night runs the identical statement against the same rows. The mechanism that
gives deletion a horizon could not close on the instances that needed it.

Migration 0277 adds the indexes the predicates ask for; the three tombstone
ones are partial on IS NOT NULL, so on a healthy instance they cover only the
tombstones and shrink as the purge drains. The purges now walk the backlog in
batches of 5 000 by primary key, capped at 200 000 rows a run.

Both handlers used to swallow every error into a wide-event warning and
return, so pg-boss recorded a successful job either way: a purge failing
every night looked exactly like a purge with nothing to do. They now rethrow,
and a run that stops at the cap records that a backlog remains.

prisma/migrations/0277_retention_purge_indexes/migration.sql

@@ -0,0 +1,46 @@
+-- v1.32.40 — indexes for the retention purges.
+--
+-- The nightly tombstone purge asks for `deleted_at IS NOT NULL AND deleted_at
+-- < cutoff`, and the audit-log purge asks for `created_at < cutoff`. Neither
+-- predicate had an index: every `deleted_at` index in the schema is PARTIAL on
+-- `deleted_at IS NULL` (the live-rows predicate, the exact opposite of what a
+-- purge needs), and every `audit_logs` index leads with `user_id` or `action`.
+-- Both purges therefore planned a sequential scan and paid index maintenance
+-- on every removed row. Each connection carries `statement_timeout=60000`
+-- (`src/lib/db.ts`), so past a certain backlog the statement aborts, the
+-- transaction rolls back, nothing is purged, and the next night runs the same
+-- statement against the same rows. The purge that was meant to give deletion a
+-- horizon could not close on the instances that needed it most.
+--
+-- The purges are batched as of the same release, so no single statement is on
+-- the hook for a whole backlog; these indexes are what make each batch's
+-- lookup cheap instead of a fresh sequential scan per batch.
+--
+-- The three tombstone indexes are PARTIAL on `deleted_at IS NOT NULL`, which
+-- is the shape that matters here: on a healthy instance the index covers only
+-- the tombstones, so it stays tiny and shrinks as the purge drains. Prisma
+-- cannot express a partial predicate, so these stay raw-SQL-only — the same
+-- convention as migrations 0087 / 0108 / 0243, noted at the model in
+-- `schema.prisma`. `audit_logs(created_at)` is expressible and is declared on
+-- the model.
+--
+-- Additive-only: every statement is `CREATE INDEX IF NOT EXISTS`, no table
+-- rewrite, no data change. NON-CONCURRENT on purpose — Prisma Migrate wraps a
+-- migration file in a transaction and `CREATE INDEX CONCURRENTLY` cannot run
+-- in one; these build at boot, where docker-entrypoint runs `migrate deploy`
+-- before the app serves traffic, so the build lock lands pre-traffic.
+
+CREATE INDEX IF NOT EXISTS "measurements_tombstone_purge_idx"
+  ON "measurements" ("deleted_at")
+  WHERE "deleted_at" IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS "mood_entries_tombstone_purge_idx"
+  ON "mood_entries" ("deleted_at")
+  WHERE "deleted_at" IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS "medication_intake_events_tombstone_purge_idx"
+  ON "medication_intake_events" ("deleted_at")
+  WHERE "deleted_at" IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS "audit_logs_created_at_idx"
+  ON "audit_logs" ("created_at");

prisma/schema.prisma

@@ -1763,6 +1763,13 @@ model Measurement {
   // (v1.28.46 H4). It makes the note-encryption boot-discovery `SELECT DISTINCT
   // user_id` an index-only scan that converges to empty. Same partial-predicate
   // reason it is omitted here.
+  // `measurements_tombstone_purge_idx` is a partial index over `(deleted_at)
+  // WHERE deleted_at IS NOT NULL` — see migration 0277 (v1.32.40). Every
+  // other deleted_at index here is partial on IS NULL, which is the live-rows
+  // predicate and the opposite of what the nightly tombstone purge asks for;
+  // without this one the purge planned a sequential scan and could not finish
+  // inside statement_timeout. Raw SQL only (partial predicate not expressible
+  // in Prisma).
   @@map("measurements")
 }
 
@@ -2854,6 +2861,13 @@ model MedicationIntakeEvent {
   // scanned the whole intake history. These two let both index-seek instead.
   @@index([userId, scheduledFor])
   @@index([userId, medicationId, takenAt])
+  // `medication_intake_events_tombstone_purge_idx` is a partial index over `(deleted_at)
+  // WHERE deleted_at IS NOT NULL` — see migration 0277 (v1.32.40). Every
+  // other deleted_at index here is partial on IS NULL, which is the live-rows
+  // predicate and the opposite of what the nightly tombstone purge asks for;
+  // without this one the purge planned a sequential scan and could not finish
+  // inside statement_timeout. Raw SQL only (partial predicate not expressible
+  // in Prisma).
   @@map("medication_intake_events")
 }
 
@@ -3852,6 +3866,13 @@ model MoodEntry {
   // WHERE note IS NOT NULL AND note_encrypted IS NULL` — see migration 0243
   // (v1.28.46 H4), serving the note-encryption boot-discovery scan. Raw SQL
   // only (partial predicate not expressible in Prisma).
+  // `mood_entries_tombstone_purge_idx` is a partial index over `(deleted_at)
+  // WHERE deleted_at IS NOT NULL` — see migration 0277 (v1.32.40). Every
+  // other deleted_at index here is partial on IS NULL, which is the live-rows
+  // predicate and the opposite of what the nightly tombstone purge asks for;
+  // without this one the purge planned a sequential scan and could not finish
+  // inside statement_timeout. Raw SQL only (partial predicate not expressible
+  // in Prisma).
   @@map("mood_entries")
 }
 
@@ -4277,6 +4298,11 @@ model AuditLog {
   // this three-column index serves both the findFirst and the take:5 as
   // an index range scan over the highest-churn table. See migration 0123.
   @@index([userId, action, createdAt(sort: Desc)])
+  // v1.32.40 — the retention purge filters on `createdAt` alone. Every index
+  // above leads with `userId` or `action`, so the purge planned a sequential
+  // scan and could not finish inside statement_timeout on an instance with
+  // enough history. See migration 0277.
+  @@index([createdAt])
   @@map("audit_logs")
 }
 

src/lib/jobs/__tests__/audit-log-cleanup.test.ts

@@ -1,14 +1,28 @@
-import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import {
+  describe,
+  it,
+  expect,
+  vi,
+  beforeEach,
+  afterEach,
+  type Mock,
+} from "vitest";
 import type { PrismaClient } from "@/generated/prisma/client";
 import {
   cleanupOldAuditLogs,
   getAuditLogRetentionDays,
   DEFAULT_AUDIT_LOG_RETENTION_DAYS,
 } from "../audit-log-cleanup";
+import { PURGE_BATCH_SIZE, PURGE_MAX_BATCHES } from "../purge-batch";
 
 function makePrismaMock(deletedCount: number) {
   return {
     auditLog: {
+      findMany: vi
+        .fn()
+        .mockResolvedValue(
+          Array.from({ length: deletedCount }, (_, i) => ({ id: `row-${i}` })),
+        ),
       deleteMany: vi.fn().mockResolvedValue({ count: deletedCount }),
     },
   } as unknown as PrismaClient;
@@ -56,14 +70,23 @@ describe("cleanupOldAuditLogs", () => {
   it("deletes audit log rows older than 365 days by default", async () => {
     const prisma = makePrismaMock(42);
     const now = new Date("2026-05-04T00:00:00Z");
-    const deleted = await cleanupOldAuditLogs(prisma, now);
+    const outcome = await cleanupOldAuditLogs(prisma, now);
 
-    expect(deleted).toBe(42);
-    expect(prisma.auditLog.deleteMany).toHaveBeenCalledTimes(1);
+    expect(outcome).toEqual({ deleted: 42, drained: true });
+    // The predicate is applied on the bounded lookup; the delete itself is by
+    // primary key so no statement is on the hook for the whole backlog.
+    expect(prisma.auditLog.findMany).toHaveBeenCalledTimes(1);
     const cutoff = new Date(now.getTime() - 365 * 86_400_000);
-    expect(prisma.auditLog.deleteMany).toHaveBeenCalledWith({
+    expect(prisma.auditLog.findMany).toHaveBeenCalledWith({
       where: { createdAt: { lt: cutoff } },
+      select: { id: true },
+      take: PURGE_BATCH_SIZE,
     });
+    expect(prisma.auditLog.deleteMany).toHaveBeenCalledTimes(1);
+    expect(
+      (prisma.auditLog.deleteMany as unknown as Mock).mock.calls[0][0].where.id
+        .in,
+    ).toHaveLength(42);
   });
 
   it("uses the configured retention", async () => {
@@ -73,8 +96,22 @@ describe("cleanupOldAuditLogs", () => {
     await cleanupOldAuditLogs(prisma, now);
 
     const cutoff = new Date(now.getTime() - 90 * 86_400_000);
-    expect(prisma.auditLog.deleteMany).toHaveBeenCalledWith({
+    expect(prisma.auditLog.findMany).toHaveBeenCalledWith({
       where: { createdAt: { lt: cutoff } },
+      select: { id: true },
+      take: PURGE_BATCH_SIZE,
     });
   });
+
+  it("stops at the batch cap and says the backlog is not drained", async () => {
+    const prisma = makePrismaMock(PURGE_BATCH_SIZE);
+    const outcome = await cleanupOldAuditLogs(
+      prisma,
+      new Date("2026-05-04T00:00:00Z"),
+    );
+
+    expect(prisma.auditLog.findMany).toHaveBeenCalledTimes(PURGE_MAX_BATCHES);
+    expect(outcome.drained).toBe(false);
+    expect(outcome.deleted).toBe(PURGE_BATCH_SIZE * PURGE_MAX_BATCHES);
+  });
 });

src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts

@@ -16,6 +16,7 @@ import { join } from "node:path";
 import { describe, expect, it, vi } from "vitest";
 
 import { cleanupExpiredMeasurementTombstones } from "../measurement-tombstone-cleanup";
+import { PURGE_BATCH_SIZE, PURGE_MAX_BATCHES } from "../purge-batch";
 import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
 
 const DAY_MS = 86_400_000;
@@ -23,16 +24,19 @@ const DAY_MS = 86_400_000;
 describe("cleanupExpiredMeasurementTombstones", () => {
   it("prunes only tombstones older than the retention horizon", async () => {
     const now = new Date("2026-06-01T00:00:00.000Z");
+    const findMany = vi
+      .fn()
+      .mockResolvedValueOnce([{ id: "a" }, { id: "b" }, { id: "c" }]);
     const deleteMany = vi.fn().mockResolvedValue({ count: 3 });
     const prisma = {
-      measurement: { deleteMany },
+      measurement: { findMany, deleteMany },
     } as unknown as Parameters<typeof cleanupExpiredMeasurementTombstones>[0];
 
-    const pruned = await cleanupExpiredMeasurementTombstones(prisma, now);
-    expect(pruned).toBe(3);
+    const outcome = await cleanupExpiredMeasurementTombstones(prisma, now);
+    expect(outcome).toEqual({ deleted: 3, drained: true });
 
-    expect(deleteMany).toHaveBeenCalledTimes(1);
-    const where = deleteMany.mock.calls[0][0].where as {
+    expect(findMany).toHaveBeenCalledTimes(1);
+    const where = findMany.mock.calls[0][0].where as {
       deletedAt: { not: null; lt: Date };
     };
     // Only soft-deleted rows are eligible …
@@ -42,6 +46,38 @@ describe("cleanupExpiredMeasurementTombstones", () => {
       now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS,
     );
     expect(where.deletedAt.lt.getTime()).toBe(expectedCutoff.getTime());
+
+    // The delete is by primary key, so no single statement is on the hook for
+    // the whole backlog.
+    expect(deleteMany).toHaveBeenCalledTimes(1);
+    expect(deleteMany.mock.calls[0][0].where).toEqual({
+      id: { in: ["a", "b", "c"] },
+    });
+  });
+
+  it("keeps walking a backlog that exceeds one batch, and reports when it stops short", async () => {
+    const now = new Date("2026-06-01T00:00:00.000Z");
+    // Always a full batch back — the predicate never exhausts, so the run has
+    // to stop at the cap rather than loop forever.
+    const fullBatch = Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({
+      id: `row-${i}`,
+    }));
+    const findMany = vi.fn().mockResolvedValue(fullBatch);
+    const deleteMany = vi.fn().mockResolvedValue({ count: PURGE_BATCH_SIZE });
+    const prisma = {
+      measurement: { findMany, deleteMany },
+    } as unknown as Parameters<typeof cleanupExpiredMeasurementTombstones>[0];
+
+    const outcome = await cleanupExpiredMeasurementTombstones(prisma, now);
+
+    expect(findMany).toHaveBeenCalledTimes(PURGE_MAX_BATCHES);
+    expect(outcome.deleted).toBe(PURGE_BATCH_SIZE * PURGE_MAX_BATCHES);
+    expect(outcome.drained).toBe(false);
+    // Every lookup is bounded — an unbounded one is the shape that could not
+    // finish inside statement_timeout.
+    for (const call of findMany.mock.calls) {
+      expect(call[0].take).toBe(PURGE_BATCH_SIZE);
+    }
   });
 });
 

src/lib/jobs/audit-log-cleanup.ts

@@ -11,6 +11,7 @@
  * `deleteMany`; runs daily via pg-boss.
  */
 import type { PrismaClient } from "@/generated/prisma/client";
+import { purgeInBatches, type PurgeOutcome } from "@/lib/jobs/purge-batch";
 
 export const DEFAULT_AUDIT_LOG_RETENTION_DAYS = 365;
 
@@ -27,14 +28,34 @@ export function getAuditLogRetentionDays(): number {
   return parsed;
 }
 
+/**
+ * Delete audit rows past the retention horizon, in bounded batches.
+ *
+ * Until v1.32.40 this was one unbounded `deleteMany` on `created_at < cutoff`.
+ * Every index on the table leads with `user_id` or `action`, so the predicate
+ * had none and the statement planned a sequential scan plus index maintenance
+ * on every removed row. Past the 60-second `statement_timeout` that aborts and
+ * rolls back, and the next night repeats it against the same rows — a
+ * retention window that can never close on an instance that has accumulated
+ * enough of them. Migration 0277 adds the `created_at` index; the batching
+ * keeps any single statement bounded regardless.
+ */
 export async function cleanupOldAuditLogs(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<number> {
+): Promise<PurgeOutcome> {
   const days = getAuditLogRetentionDays();
   const cutoff = new Date(now.getTime() - days * 86_400_000);
-  const { count } = await prisma.auditLog.deleteMany({
-    where: { createdAt: { lt: cutoff } },
+  return purgeInBatches({
+    findIds: async (take) =>
+      (
+        await prisma.auditLog.findMany({
+          where: { createdAt: { lt: cutoff } },
+          select: { id: true },
+          take,
+        })
+      ).map((row) => row.id),
+    deleteIds: async (ids) =>
+      (await prisma.auditLog.deleteMany({ where: { id: { in: ids } } })).count,
   });
-  return count;
 }

src/lib/jobs/measurement-tombstone-cleanup.ts

@@ -20,50 +20,90 @@
  */
 import type { PrismaClient } from "@/generated/prisma/client";
 import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
+import { purgeInBatches, type PurgeOutcome } from "@/lib/jobs/purge-batch";
 
 const DAY_MS = 86_400_000;
 
+function tombstoneCutoff(now: Date): Date {
+  return new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS);
+}
+
 /**
- * Hard-delete soft-deleted measurement rows whose `deletedAt` is older
- * than the retention horizon. Returns the number of rows pruned.
+ * Hard-delete soft-deleted measurement rows whose `deletedAt` is older than
+ * the retention horizon.
+ *
+ * Batched. Until v1.32.40 this was one unbounded `deleteMany` over a predicate
+ * that no index supported, which meant a backlog large enough to exceed the
+ * 60-second `statement_timeout` could never be drained: the statement aborted,
+ * the transaction rolled back, and the next night ran the identical statement
+ * against the same rows. `measurements` is the densest table in the schema and
+ * a bulk source delete can tombstone a six-figure row count in one action, so
+ * this is the one most likely to have been stuck. Migration 0277 adds the
+ * partial index the predicate needs.
  */
 export async function cleanupExpiredMeasurementTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<number> {
-  const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS);
-  const { count } = await prisma.measurement.deleteMany({
-    where: { deletedAt: { not: null, lt: cutoff } },
+): Promise<PurgeOutcome> {
+  const cutoff = tombstoneCutoff(now);
+  return purgeInBatches({
+    findIds: async (take) =>
+      (
+        await prisma.measurement.findMany({
+          where: { deletedAt: { not: null, lt: cutoff } },
+          select: { id: true },
+          take,
+        })
+      ).map((row) => row.id),
+    deleteIds: async (ids) =>
+      (await prisma.measurement.deleteMany({ where: { id: { in: ids } } }))
+        .count,
   });
-  return count;
 }
 
-/**
- * Hard-delete soft-deleted mood-entry rows past the retention horizon.
- * Returns the number of rows pruned.
- */
+/** Hard-delete soft-deleted mood-entry rows past the retention horizon. */
 export async function cleanupExpiredMoodTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<number> {
-  const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS);
-  const { count } = await prisma.moodEntry.deleteMany({
-    where: { deletedAt: { not: null, lt: cutoff } },
+): Promise<PurgeOutcome> {
+  const cutoff = tombstoneCutoff(now);
+  return purgeInBatches({
+    findIds: async (take) =>
+      (
+        await prisma.moodEntry.findMany({
+          where: { deletedAt: { not: null, lt: cutoff } },
+          select: { id: true },
+          take,
+        })
+      ).map((row) => row.id),
+    deleteIds: async (ids) =>
+      (await prisma.moodEntry.deleteMany({ where: { id: { in: ids } } })).count,
   });
-  return count;
 }
 
 /**
  * Hard-delete soft-deleted medication-intake-event rows past the retention
- * horizon. Returns the number of rows pruned.
+ * horizon.
  */
 export async function cleanupExpiredIntakeTombstones(
   prisma: PrismaClient,
   now: Date = new Date(),
-): Promise<number> {
-  const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS);
-  const { count } = await prisma.medicationIntakeEvent.deleteMany({
-    where: { deletedAt: { not: null, lt: cutoff } },
+): Promise<PurgeOutcome> {
+  const cutoff = tombstoneCutoff(now);
+  return purgeInBatches({
+    findIds: async (take) =>
+      (
+        await prisma.medicationIntakeEvent.findMany({
+          where: { deletedAt: { not: null, lt: cutoff } },
+          select: { id: true },
+          take,
+        })
+      ).map((row) => row.id),
+    deleteIds: async (ids) =>
+      (
+        await prisma.medicationIntakeEvent.deleteMany({
+          where: { id: { in: ids } },
+        })
+      ).count,
   });
-  return count;
 }

src/lib/jobs/purge-batch.ts

@@ -0,0 +1,81 @@
+/**
+ * Batched retention purges.
+ *
+ * The retention jobs used to issue one unbounded `deleteMany` over a predicate
+ * no index supported. Every `deleted_at` index in the schema is PARTIAL on
+ * `IS NULL` — the live-rows predicate, the exact opposite of what a purge asks
+ * for — so the purge planned a sequential scan and then paid index maintenance
+ * on every row it removed. Below the 60-second `statement_timeout` the
+ * connection carries (`src/lib/db.ts`) that is invisible; above it the
+ * statement aborts, the transaction rolls back, nothing is purged, and the job
+ * runs the identical statement again the next night. A backlog that grows past
+ * the timeout can therefore never be drained by the mechanism meant to drain
+ * it, and the job that was supposed to make deletion real never completes.
+ *
+ * The fix has three parts and this file is one of them:
+ *   - the predicates get indexes that match them (migration 0277);
+ *   - the delete runs in bounded batches, so no single statement is on the
+ *     hook for the whole backlog;
+ *   - a run that stops at the cap says so, so "still draining" and "nothing
+ *     left to do" stop looking identical from the outside.
+ */
+
+/** Rows removed per statement. */
+export const PURGE_BATCH_SIZE = 5_000;
+
+/**
+ * Batches per run. 40 × 5 000 = 200 000 rows a night, which drains a large
+ * backlog over a handful of nights instead of asking one statement to survive
+ * it. The cap exists so a nightly job cannot hold connections and bloat WAL
+ * for an unbounded stretch on a instance that has accumulated years of rows.
+ */
+export const PURGE_MAX_BATCHES = 40;
+
+export interface PurgeOutcome {
+  /** Rows removed this run. */
+  deleted: number;
+  /**
+   * `false` when the run stopped at {@link PURGE_MAX_BATCHES} with rows still
+   * matching. The next run continues; an operator watching this stay `false`
+   * for many nights is watching a backlog that needs a wider window.
+   */
+  drained: boolean;
+}
+
+export interface PurgeBatchOptions {
+  /** Ids of up to `take` rows matching the retention predicate. */
+  findIds: (take: number) => Promise<string[]>;
+  /** Remove exactly those ids; returns the row count actually removed. */
+  deleteIds: (ids: string[]) => Promise<number>;
+  batchSize?: number;
+  maxBatches?: number;
+}
+
+/**
+ * Delete rows matching a retention predicate in bounded batches.
+ *
+ * Selecting ids first and deleting by primary key keeps each statement small
+ * and predictable regardless of how selective the predicate is, and keeps the
+ * scan on the index that supports the predicate rather than on the delete.
+ */
+export async function purgeInBatches({
+  findIds,
+  deleteIds,
+  batchSize = PURGE_BATCH_SIZE,
+  maxBatches = PURGE_MAX_BATCHES,
+}: PurgeBatchOptions): Promise<PurgeOutcome> {
+  let deleted = 0;
+
+  for (let batch = 0; batch < maxBatches; batch++) {
+    const ids = await findIds(batchSize);
+    if (ids.length === 0) return { deleted, drained: true };
+
+    deleted += await deleteIds(ids);
+
+    // A short final batch means the predicate is exhausted; asking again would
+    // cost a round-trip to learn nothing.
+    if (ids.length < batchSize) return { deleted, drained: true };
+  }
+
+  return { deleted, drained: false };
+}

src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts

@@ -0,0 +1,130 @@
+/**
+ * v1.32.40 — the retention purges have to be able to fail out loud.
+ *
+ * Both handlers used to catch every error into a wide-event warning and return
+ * normally. pg-boss then recorded a successful job, so a purge that aborted on
+ * `statement_timeout` every night for a year looked exactly like a purge with
+ * nothing to do. That is the failure this whole change exists to remove, so
+ * the propagation is pinned rather than left to convention.
+ *
+ * The second half pins the other half of the same problem: a run that stopped
+ * at the batch cap still has a backlog, and without the flag that is also
+ * indistinguishable from a clean run.
+ *
+ * Mutation check: drop the `throw err` from either handler and the matching
+ * "surfaces the failure" test goes red; drop the `drained` meta and the
+ * matching backlog test goes red.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const h = vi.hoisted(() => ({
+  cleanupOldAuditLogs: vi.fn(),
+  cleanupExpiredMeasurementTombstones: vi.fn(),
+  cleanupExpiredMoodTombstones: vi.fn(),
+  cleanupExpiredIntakeTombstones: vi.fn(),
+  meta: {} as Record<string, unknown>,
+  warnings: [] as string[],
+}));
+
+vi.mock("@/lib/jobs/audit-log-cleanup", () => ({
+  cleanupOldAuditLogs: h.cleanupOldAuditLogs,
+}));
+vi.mock("@/lib/jobs/measurement-tombstone-cleanup", () => ({
+  cleanupExpiredMeasurementTombstones: h.cleanupExpiredMeasurementTombstones,
+  cleanupExpiredMoodTombstones: h.cleanupExpiredMoodTombstones,
+  cleanupExpiredIntakeTombstones: h.cleanupExpiredIntakeTombstones,
+}));
+vi.mock("../shared", async (importOriginal) => ({
+  ...(await importOriginal<Record<string, unknown>>()),
+  getWorkerPrisma: () => ({}),
+}));
+vi.mock("@/lib/logging/background", () => ({
+  withBackgroundEvent: async (
+    _name: string,
+    fn: (evt: {
+      addMeta: (k: string, v: unknown) => void;
+      addWarning: (w: string) => void;
+    }) => Promise<void>,
+  ) =>
+    fn({
+      addMeta: (k, v) => {
+        h.meta[k] = v;
+      },
+      addWarning: (w) => {
+        h.warnings.push(w);
+      },
+    }),
+}));
+
+import {
+  handleAuditLogCleanup,
+  handleMeasurementTombstoneCleanup,
+} from "../cleanup-handlers";
+
+const DRAINED = { deleted: 1, drained: true };
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  for (const key of Object.keys(h.meta)) delete h.meta[key];
+  h.warnings.length = 0;
+  h.cleanupOldAuditLogs.mockResolvedValue(DRAINED);
+  h.cleanupExpiredMeasurementTombstones.mockResolvedValue(DRAINED);
+  h.cleanupExpiredMoodTombstones.mockResolvedValue(DRAINED);
+  h.cleanupExpiredIntakeTombstones.mockResolvedValue(DRAINED);
+});
+
+describe("audit-log cleanup handler", () => {
+  it("surfaces the failure instead of recording a successful job", async () => {
+    h.cleanupOldAuditLogs.mockRejectedValue(new Error("statement timeout"));
+    await expect(handleAuditLogCleanup([])).rejects.toThrow(
+      "statement timeout",
+    );
+    expect(h.warnings.join(" ")).toContain("audit-log-cleanup failed");
+  });
+
+  it("reports a backlog left behind by the batch cap", async () => {
+    h.cleanupOldAuditLogs.mockResolvedValue({
+      deleted: 200_000,
+      drained: false,
+    });
+    await handleAuditLogCleanup([]);
+    expect(h.meta.audit_log_cleanup_deleted).toBe(200_000);
+    expect(h.meta.audit_log_cleanup_drained).toBe(false);
+    expect(h.warnings.join(" ")).toContain("backlog remains");
+  });
+
+  it("stays quiet when the backlog is drained", async () => {
+    await handleAuditLogCleanup([]);
+    expect(h.meta.audit_log_cleanup_drained).toBe(true);
+    expect(h.warnings).toEqual([]);
+  });
+});
+
+describe("tombstone cleanup handler", () => {
+  it("surfaces the failure instead of recording a successful job", async () => {
+    h.cleanupExpiredMeasurementTombstones.mockRejectedValue(
+      new Error("statement timeout"),
+    );
+    await expect(handleMeasurementTombstoneCleanup([])).rejects.toThrow(
+      "statement timeout",
+    );
+    expect(h.warnings.join(" ")).toContain("tombstone-cleanup failed");
+  });
+
+  it("reports a backlog when any of the three legs stopped short", async () => {
+    h.cleanupExpiredMoodTombstones.mockResolvedValue({
+      deleted: 200_000,
+      drained: false,
+    });
+    await handleMeasurementTombstoneCleanup([]);
+    expect(h.meta.tombstone_cleanup_drained).toBe(false);
+    expect(h.meta.mood_tombstone_cleanup_pruned).toBe(200_000);
+    expect(h.warnings.join(" ")).toContain("backlog remains");
+  });
+
+  it("stays quiet when all three legs drained", async () => {
+    await handleMeasurementTombstoneCleanup([]);
+    expect(h.meta.tombstone_cleanup_drained).toBe(true);
+    expect(h.warnings).toEqual([]);
+  });
+});

src/lib/jobs/reminder/cleanup-handlers.ts

@@ -184,16 +184,36 @@ export async function handleMeasurementTombstoneCleanup(
       try {
         // v1.7.0 sync — prune tombstones across all three sync domains on
         // the same retention horizon.
-        const [measurements, mood, intakes] = await Promise.all([
-          cleanupExpiredMeasurementTombstones(p),
-          cleanupExpiredMoodTombstones(p),
-          cleanupExpiredIntakeTombstones(p),
-        ]);
-        evt.addMeta("measurement_tombstone_cleanup_pruned", measurements);
-        evt.addMeta("mood_tombstone_cleanup_pruned", mood);
-        evt.addMeta("intake_tombstone_cleanup_pruned", intakes);
+        //
+        // v1.32.40 — sequential rather than parallel. Each leg now walks its
+        // backlog in batches, and three concurrent walks against the three
+        // densest tables competed for the worker pool for no gain: the job has
+        // the whole night.
+        const measurements = await cleanupExpiredMeasurementTombstones(p);
+        const mood = await cleanupExpiredMoodTombstones(p);
+        const intakes = await cleanupExpiredIntakeTombstones(p);
+        evt.addMeta(
+          "measurement_tombstone_cleanup_pruned",
+          measurements.deleted,
+        );
+        evt.addMeta("mood_tombstone_cleanup_pruned", mood.deleted);
+        evt.addMeta("intake_tombstone_cleanup_pruned", intakes.deleted);
+        // A run that stopped at the batch cap has a backlog left. Without this
+        // flag a partial run and a complete one are indistinguishable from
+        // outside, which is how a purge that never finished stayed invisible.
+        const drained = measurements.drained && mood.drained && intakes.drained;
+        evt.addMeta("tombstone_cleanup_drained", drained);
+        if (!drained) {
+          evt.addWarning(
+            "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run",
+          );
+        }
       } catch (err) {
+        // Rethrow. A retention purge that fails silently every night looks
+        // exactly like one that has nothing to do; pg-boss recording a failed
+        // job is the only signal an operator can act on.
         evt.addWarning(`tombstone-cleanup failed: ${err}`);
+        throw err;
       }
     },
   );
@@ -268,10 +288,19 @@ export async function handleAuditLogCleanup(
   await withBackgroundEvent("job.audit_log_cleanup", async (evt) => {
     const p = getWorkerPrisma();
     try {
-      const deleted = await cleanupOldAuditLogs(p);
-      evt.addMeta("audit_log_cleanup_deleted", deleted);
+      const outcome = await cleanupOldAuditLogs(p);
+      evt.addMeta("audit_log_cleanup_deleted", outcome.deleted);
+      evt.addMeta("audit_log_cleanup_drained", outcome.drained);
+      if (!outcome.drained) {
+        evt.addWarning(
+          "audit-log-cleanup stopped at the batch cap; a backlog remains for the next run",
+        );
+      }
     } catch (err) {
+      // Rethrow — see the tombstone handler. Storage-limitation retention that
+      // fails quietly is retention that is not happening.
       evt.addWarning(`audit-log-cleanup failed: ${err}`);
+      throw err;
     }
   });
 }