diff --git a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts b/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts index fa68a9100..84de68b99 100644 --- a/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts +++ b/src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts @@ -21,19 +21,43 @@ import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; const DAY_MS = 86_400_000; +/** + * A client whose interactive transaction hands the callback a client with the + * same delegates, and whose restore-lock probe answers `held`. + */ +function mockClient( + measurement: { findMany: unknown; deleteMany: unknown }, + held: (userId: string) => boolean = () => true, +) { + const lockCalls: string[] = []; + const tx = { + measurement, + $queryRaw: vi.fn(async (_strings: TemplateStringsArray, key: string) => { + const userId = key.replace(/^backup-restore:/, ""); + lockCalls.push(userId); + return [{ held: held(userId) }]; + }), + }; + const prisma = { + measurement, + $transaction: vi.fn(async (fn: (client: typeof tx) => unknown) => fn(tx)), + } as unknown as Parameters[0]; + return { prisma, lockCalls }; +} + 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 findMany = vi.fn().mockResolvedValueOnce([ + { id: "a", userId: "u1" }, + { id: "b", userId: "u1" }, + { id: "c", userId: "u1" }, + ]); const deleteMany = vi.fn().mockResolvedValue({ count: 3 }); - const prisma = { - measurement: { findMany, deleteMany }, - } as unknown as Parameters[0]; + const { prisma } = mockClient({ findMany, deleteMany }); const outcome = await cleanupExpiredMeasurementTombstones(prisma, now); - expect(outcome).toEqual({ deleted: 3, drained: true }); + expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); expect(findMany).toHaveBeenCalledTimes(1); const where = findMany.mock.calls[0][0].where as { @@ -61,12 +85,11 @@ describe("cleanupExpiredMeasurementTombstones", () => { // to stop at the cap rather than loop forever. const fullBatch = Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({ id: `row-${i}`, + userId: "u1", })); 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 { prisma } = mockClient({ findMany, deleteMany }); const outcome = await cleanupExpiredMeasurementTombstones(prisma, now); @@ -81,6 +104,57 @@ describe("cleanupExpiredMeasurementTombstones", () => { }); }); +describe("cleanupExpiredMeasurementTombstones — accounts under restore", () => { + it("deletes one statement per account per batch, each under the account's restore lock", async () => { + const findMany = vi.fn().mockResolvedValueOnce([ + { id: "a1", userId: "a" }, + { id: "b1", userId: "b" }, + { id: "a2", userId: "a" }, + ]); + const deleteMany = vi.fn(async ({ where }) => ({ + count: (where.id.in as string[]).length, + })); + const { prisma, lockCalls } = mockClient({ findMany, deleteMany }); + + const outcome = await cleanupExpiredMeasurementTombstones(prisma); + + expect(outcome).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); + expect(lockCalls).toEqual(["a", "b"]); + expect(deleteMany.mock.calls.map((call) => call[0].where)).toEqual([ + { id: { in: ["a1", "a2"] } }, + { id: { in: ["b1"] } }, + ]); + }); + + it("leaves an account under restore and keeps it out of the rest of the run", async () => { + const full = (userId: string) => + Array.from({ length: PURGE_BATCH_SIZE }, (_, i) => ({ + id: `${userId}-${i}`, + userId, + })); + const findMany = vi + .fn() + .mockResolvedValueOnce(full("restoring")) + .mockResolvedValueOnce([{ id: "other-1", userId: "other" }]); + const deleteMany = vi.fn(async ({ where }) => ({ + count: (where.id.in as string[]).length, + })); + const { prisma } = mockClient( + { findMany, deleteMany }, + (userId) => userId !== "restoring", + ); + + const outcome = await cleanupExpiredMeasurementTombstones(prisma); + + expect(outcome).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 }); + expect(findMany.mock.calls[0][0].where).not.toHaveProperty("userId"); + expect(findMany.mock.calls[1][0].where.userId).toEqual({ + notIn: ["restoring"], + }); + expect(deleteMany).toHaveBeenCalledTimes(1); + }); +}); + describe("reminder-worker — measurement-tombstone-cleanup schedule", () => { // v1.18.1 — the cleanup wiring moved out of the 2143-LOC reminder-worker // boot file into the maintenance registrar. The dead-queue guard follows it. diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index bd55c5c2d..fe6be9808 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -71,6 +71,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "days_skipped", "days_skipped_no_tombstones", "days_stored", + // Tombstone purge: accounts left for the next run because a restore of + // the account was running. + "deferred_accounts", "deleted", "dense_days_consolidated", "dense_rows_soft_deleted", diff --git a/src/lib/jobs/measurement-tombstone-cleanup.ts b/src/lib/jobs/measurement-tombstone-cleanup.ts index 43c2f5db5..306b2bb49 100644 --- a/src/lib/jobs/measurement-tombstone-cleanup.ts +++ b/src/lib/jobs/measurement-tombstone-cleanup.ts @@ -18,9 +18,17 @@ * (`NATIVE_REFRESH_TOKEN_DAYS` + margin) so it moves automatically if the * refresh-token lifetime changes — the two never drift. */ -import type { PrismaClient } from "@/generated/prisma/client"; +import type { Prisma, PrismaClient } from "@/generated/prisma/client"; import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; -import { purgeInBatches, type PurgeOutcome } from "@/lib/jobs/purge-batch"; +import { + AccountRestoreInProgressError, + holdAccountAgainstRestore, +} from "@/lib/export/restore-lock"; +import { + PURGE_BATCH_SIZE, + PURGE_MAX_BATCHES, + type PurgeOutcome, +} from "@/lib/jobs/purge-batch"; const DAY_MS = 86_400_000; @@ -28,6 +36,103 @@ function tombstoneCutoff(now: Date): Date { return new Date(now.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS); } +/** A purge run, plus the accounts it left for the next run. */ +export interface TombstonePurgeOutcome extends PurgeOutcome { + /** + * Accounts whose tombstones were left in place because a restore of the + * account was running. The next night purges them. + */ + deferredAccounts: number; +} + +type TombstoneRow = { id: string; userId: string }; + +interface TombstonePurgeOptions { + prisma: PrismaClient; + /** Up to `take` expired tombstones, none of them owned by `skipUserIds`. */ + findRows: (take: number, skipUserIds: string[]) => Promise; + /** Remove exactly those ids inside `tx`; returns the rows removed. */ + deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise; + batchSize?: number; + maxBatches?: number; +} + +/** + * Walk the expired tombstones in bounded batches, deleting each batch one + * account at a time under the account's restore lock. + * + * A restore clears every row of the account, tombstones included, in one + * long transaction. The purge used to delete a batch spanning accounts in one + * statement, so on an account under restore it waited on the restore's row + * locks, for as long as the restore ran and well past `statement_timeout`, + * or, holding some of the account's rows while the restore held others, + * deadlocked with it. So each account's share of a batch is deleted in a + * transaction that first takes the restore lock in shared mode, the way the + * consolidation passes do (`restore-lock.ts`). An account under restore is + * refused at once, dropped from the rest of the run, and purged the next + * night, when its tombstones are whatever the restore left. + * + * Still one delete statement per account per batch, never one per row. + */ +async function purgeTombstonesByAccount({ + prisma, + findRows, + deleteIds, + batchSize = PURGE_BATCH_SIZE, + maxBatches = PURGE_MAX_BATCHES, +}: TombstonePurgeOptions): Promise { + let deleted = 0; + const deferred = new Set(); + const outcome = (drained: boolean): TombstonePurgeOutcome => ({ + deleted, + drained, + deferredAccounts: deferred.size, + }); + + for (let batch = 0; batch < maxBatches; batch++) { + const rows = await findRows(batchSize, [...deferred]); + if (rows.length === 0) return outcome(true); + + const byAccount = new Map(); + for (const row of rows) { + const ids = byAccount.get(row.userId); + if (ids) ids.push(row.id); + else byAccount.set(row.userId, [row.id]); + } + for (const [userId, ids] of byAccount) { + try { + deleted += await prisma.$transaction( + async (tx) => { + await holdAccountAgainstRestore(tx, userId); + return deleteIds(tx, ids); + }, + // A full batch of one account is one statement over thousands of + // rows; the interactive default of five seconds is too tight for + // it on a slow disk. The connection's statement_timeout bounds it. + { timeout: 60_000 }, + ); + } catch (err) { + if (!(err instanceof AccountRestoreInProgressError)) throw err; + deferred.add(userId); + } + } + + // A short batch means the predicate is exhausted, the deferred accounts + // aside; asking again would cost a round-trip to learn nothing. + if (rows.length < batchSize) return outcome(true); + } + + return outcome(false); +} + +/** The retention predicate, minus the accounts a run has deferred. */ +function expiredTombstones(cutoff: Date, skipUserIds: string[]) { + return { + deletedAt: { not: null, lt: cutoff }, + ...(skipUserIds.length > 0 ? { userId: { notIn: skipUserIds } } : {}), + }; +} + /** * Hard-delete soft-deleted measurement rows whose `deletedAt` is older than * the retention horizon. @@ -44,20 +149,18 @@ function tombstoneCutoff(now: Date): Date { export async function cleanupExpiredMeasurementTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): 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 purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.measurement.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => + (await tx.measurement.deleteMany({ where: { id: { in: ids } } })).count, }); } @@ -65,19 +168,18 @@ export async function cleanupExpiredMeasurementTombstones( export async function cleanupExpiredMoodTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): 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 purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.moodEntry.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => + (await tx.moodEntry.deleteMany({ where: { id: { in: ids } } })).count, }); } @@ -88,20 +190,19 @@ export async function cleanupExpiredMoodTombstones( export async function cleanupExpiredIntakeTombstones( prisma: PrismaClient, now: Date = new Date(), -): Promise { +): Promise { const cutoff = tombstoneCutoff(now); - return purgeInBatches({ - findIds: async (take) => + return purgeTombstonesByAccount({ + prisma, + findRows: (take, skipUserIds) => + prisma.medicationIntakeEvent.findMany({ + where: expiredTombstones(cutoff, skipUserIds), + select: { id: true, userId: true }, + take, + }), + deleteIds: async (tx, ids) => ( - 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({ + await tx.medicationIntakeEvent.deleteMany({ where: { id: { in: ids } }, }) ).count, diff --git a/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts b/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts index 354169b08..20a3f479b 100644 --- a/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts +++ b/src/lib/jobs/reminder/__tests__/cleanup-handlers-visibility.test.ts @@ -62,15 +62,16 @@ import { } from "../cleanup-handlers"; const DRAINED = { deleted: 1, drained: true }; +const TOMBSTONES_DRAINED = { ...DRAINED, deferredAccounts: 0 }; 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); + h.cleanupExpiredMeasurementTombstones.mockResolvedValue(TOMBSTONES_DRAINED); + h.cleanupExpiredMoodTombstones.mockResolvedValue(TOMBSTONES_DRAINED); + h.cleanupExpiredIntakeTombstones.mockResolvedValue(TOMBSTONES_DRAINED); }); describe("audit-log cleanup handler", () => { @@ -115,6 +116,7 @@ describe("tombstone cleanup handler", () => { h.cleanupExpiredMoodTombstones.mockResolvedValue({ deleted: 200_000, drained: false, + deferredAccounts: 0, }); await handleMeasurementTombstoneCleanup([]); expect(h.meta.tombstone_cleanup_drained).toBe(false); @@ -122,6 +124,16 @@ describe("tombstone cleanup handler", () => { expect(h.warnings.join(" ")).toContain("backlog remains"); }); + it("reports the accounts it left because a restore was running", async () => { + h.cleanupExpiredMeasurementTombstones.mockResolvedValue({ + ...TOMBSTONES_DRAINED, + deferredAccounts: 2, + }); + const outcome = await handleMeasurementTombstoneCleanup([]); + expect(h.meta.tombstone_cleanup_deferred_accounts).toBe(2); + expect(outcome).toMatchObject({ did: { deferred_accounts: 2 } }); + }); + it("stays quiet when all three legs drained", async () => { await handleMeasurementTombstoneCleanup([]); expect(h.meta.tombstone_cleanup_drained).toBe(true); diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index 98ad16be9..5795aae78 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -254,6 +254,13 @@ export async function handleMeasurementTombstoneCleanup( // 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); + // Accounts under restore are left alone and purged the next night. + const deferredAccounts = Math.max( + measurements.deferredAccounts, + mood.deferredAccounts, + intakes.deferredAccounts, + ); + evt.addMeta("tombstone_cleanup_deferred_accounts", deferredAccounts); if (!drained) { evt.addWarning( "tombstone-cleanup stopped at the batch cap; a backlog remains for the next run", @@ -264,6 +271,7 @@ export async function handleMeasurementTombstoneCleanup( mood_pruned: mood.deleted, intakes_pruned: intakes.deleted, drained, + deferred_accounts: deferredAccounts, }); } catch (err) { // Rethrow. A retention purge that fails silently every night looks diff --git a/tests/integration/sync-changes.test.ts b/tests/integration/sync-changes.test.ts index 421aa7451..606933304 100644 --- a/tests/integration/sync-changes.test.ts +++ b/tests/integration/sync-changes.test.ts @@ -399,7 +399,7 @@ describe("GET /api/sync/changes (real Postgres)", () => { // pruned; the recent reachable tombstone survives. `drained` says the // batched walk reached the end of the backlog rather than stopping at // its per-run cap. - expect(pruned).toEqual({ deleted: 1, drained: true }); + expect(pruned).toEqual({ deleted: 1, drained: true, deferredAccounts: 0 }); const survivors = await prisma.measurement.findMany({ where: { userId: TEST_USER_ID }, select: { externalId: true }, diff --git a/tests/integration/tombstone-cleanup-restore-lock.test.ts b/tests/integration/tombstone-cleanup-restore-lock.test.ts new file mode 100644 index 000000000..bf49eee7e --- /dev/null +++ b/tests/integration/tombstone-cleanup-restore-lock.test.ts @@ -0,0 +1,152 @@ +/** + * #1031 — the nightly tombstone purge against a running restore. + * + * A restore clears every row of the account, tombstones included, inside one + * long transaction that holds the account's restore lock exclusively. The + * purge used to delete expired tombstones across accounts in one statement + * per batch, without the lock: on an account under restore it waited on the + * restore's row locks for as long as the restore ran, or deadlocked with it. + * + * Here a transaction stands in for the restore: it takes the lock, deletes the + * account's rows, and holds for three seconds before rolling back. The purge + * runs while it holds. What must hold: the purge does not wait, leaves the + * account under restore alone and says so, and still purges every other + * account. + * + * Mutation check: drop the `holdAccountAgainstRestore` call from + * `purgeTombstonesByAccount` and the purge waits out the hold, then deletes + * the restoring account's tombstones once the stand-in rolls back. + */ +import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const RESTORING = "tombstone-restoring"; +const OTHER = "tombstone-other"; +const HOLD_MS = 3_000; +const LONG_AGO = new Date("2020-01-01T00:00:00.000Z"); + +async function seed() { + const prisma = getPrismaClient(); + for (const id of [RESTORING, OTHER]) { + await prisma.user.create({ + data: { id, username: id, role: "USER", timezone: "UTC" }, + }); + for (let i = 0; i < 3; i++) { + await prisma.measurement.create({ + data: { + id: `${id}-tomb-${i}`, + userId: id, + type: "WEIGHT", + value: 70 + i, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(`2020-01-0${i + 1}T08:00:00.000Z`), + deletedAt: LONG_AGO, + }, + }); + } + await prisma.moodEntry.create({ + data: { + id: `${id}-mood-tomb`, + userId: id, + date: "2020-01-01", + mood: "OKAY", + score: 3, + moodLoggedAt: new Date("2020-01-01T08:00:00.000Z"), + deletedAt: LONG_AGO, + }, + }); + } +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("tombstone purge against a running restore (#1031)", () => { + it("leaves the account under restore for the next run without waiting on it", async () => { + const prisma = getPrismaClient(); + await seed(); + const { takeRestoreLock } = await import("@/lib/export/restore-lock"); + const { + cleanupExpiredMeasurementTombstones, + cleanupExpiredMoodTombstones, + } = await import("@/lib/jobs/measurement-tombstone-cleanup"); + + let holding!: () => void; + const isHolding = new Promise((resolve) => (holding = resolve)); + const restore = prisma + .$transaction( + async (tx) => { + await takeRestoreLock(tx, RESTORING); + await tx.measurement.deleteMany({ where: { userId: RESTORING } }); + await tx.moodEntry.deleteMany({ where: { userId: RESTORING } }); + holding(); + await new Promise((resolve) => setTimeout(resolve, HOLD_MS)); + throw new Error("rolled back"); + }, + { timeout: 30_000 }, + ) + .catch(() => undefined); + await isHolding; + + const started = Date.now(); + const measurements = await cleanupExpiredMeasurementTombstones(prisma); + const moods = await cleanupExpiredMoodTombstones(prisma); + const elapsed = Date.now() - started; + await restore; + + expect(elapsed).toBeLessThan(HOLD_MS / 2); + expect(measurements).toEqual({ + deleted: 3, + drained: true, + deferredAccounts: 1, + }); + expect(moods).toEqual({ deleted: 1, drained: true, deferredAccounts: 1 }); + // The stand-in rolled back, so the restoring account still has its + // tombstones; the next run purges them. + expect( + await prisma.measurement.count({ where: { userId: RESTORING } }), + ).toBe(3); + expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe( + 0, + ); + expect(await prisma.moodEntry.count({ where: { userId: OTHER } })).toBe(0); + + const next = await cleanupExpiredMeasurementTombstones(prisma); + expect(next).toEqual({ deleted: 3, drained: true, deferredAccounts: 0 }); + }, 60_000); + + it("a restore that starts during the purge waits only for the account's batch", async () => { + const prisma = getPrismaClient(); + await seed(); + const { takeRestoreLock } = await import("@/lib/export/restore-lock"); + const { cleanupExpiredMeasurementTombstones } = + await import("@/lib/jobs/measurement-tombstone-cleanup"); + + const purge = cleanupExpiredMeasurementTombstones(prisma); + const restore = prisma.$transaction(async (tx) => { + await takeRestoreLock(tx, RESTORING); + return tx.measurement.deleteMany({ where: { userId: RESTORING } }); + }); + const [purged] = await Promise.all([purge, restore]); + + // Whichever went first, both finished and nothing is left. + expect(purged.deferredAccounts).toBeLessThanOrEqual(1); + expect(await prisma.measurement.count({ where: { userId: OTHER } })).toBe( + 0, + ); + expect( + await prisma.measurement.count({ where: { userId: RESTORING } }), + ).toBe(0); + }, 60_000); +});