diff --git a/prisma/migrations/0277_retention_purge_indexes/migration.sql b/prisma/migrations/0277_retention_purge_indexes/migration.sql new file mode 100644 index 000000000..acff53d8d --- /dev/null +++ b/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"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 90ca68d2b..b625607b2 100644 --- a/prisma/schema.prisma +++ b/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") } diff --git a/src/lib/jobs/__tests__/audit-log-cleanup.test.ts b/src/lib/jobs/__tests__/audit-log-cleanup.test.ts index c7b431543..25e2eb1a3 100644 --- a/src/lib/jobs/__tests__/audit-log-cleanup.test.ts +++ b/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); + }); }); diff --git a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts b/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts index 5733f252c..ed96efe98 100644 --- a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts +++ b/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[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[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); + } }); }); diff --git a/src/lib/jobs/audit-log-cleanup.ts b/src/lib/jobs/audit-log-cleanup.ts index 5fb51179c..ddb890b24 100644 --- a/src/lib/jobs/audit-log-cleanup.ts +++ b/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 { +): Promise { 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; } diff --git a/src/lib/jobs/measurement-tombstone-cleanup.ts b/src/lib/jobs/measurement-tombstone-cleanup.ts index 681b3e2f6..d627059d9 100644 --- a/src/lib/jobs/measurement-tombstone-cleanup.ts +++ b/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 { - const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS); - const { count } = await prisma.measurement.deleteMany({ - where: { deletedAt: { not: null, lt: cutoff } }, +): Promise { + 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 { - const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS); - const { count } = await prisma.moodEntry.deleteMany({ - where: { deletedAt: { not: null, lt: cutoff } }, +): Promise { + 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 { - const cutoff = new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS); - const { count } = await prisma.medicationIntakeEvent.deleteMany({ - where: { deletedAt: { not: null, lt: cutoff } }, +): Promise { + 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; } diff --git a/src/lib/jobs/purge-batch.ts b/src/lib/jobs/purge-batch.ts new file mode 100644 index 000000000..8db2b5fac --- /dev/null +++ b/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; + /** Remove exactly those ids; returns the row count actually removed. */ + deleteIds: (ids: string[]) => Promise; + 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 { + 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 }; +} diff --git a/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts b/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts new file mode 100644 index 000000000..766971dab --- /dev/null +++ b/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, + 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>()), + getWorkerPrisma: () => ({}), +})); +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: async ( + _name: string, + fn: (evt: { + addMeta: (k: string, v: unknown) => void; + addWarning: (w: string) => void; + }) => Promise, + ) => + 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([]); + }); +}); diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index 4513975c1..a3b1e6f43 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/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; } }); }