fix(jobs): let the nightly tombstone purge step around a running restore

A restore clears every row of the account, tombstones included, inside one
long transaction under the account's exclusive restore lock. The nightly
purge deleted expired tombstones across accounts, one statement per batch,
without that lock: on an account under restore it waited on the restore's
row locks for as long as the restore ran, past its statement timeout, or
deadlocked with it once each side held rows the other needed.

Each batch is now deleted one account at a time, in a transaction that
first takes the restore lock in shared mode, the same order the
consolidation passes keep. An account under restore is refused at once,
kept out of the rest of the run, and purged the next night. Still one
delete per account per batch, never one per row. The job reports how many
accounts it deferred.

The integration test holds a stand-in restore on one account while the
purge runs: the purge returns without waiting, leaves that account alone
and purges the other. With the lock call removed it waits out the hold.

Refs #1031

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<typeof cleanupExpiredMeasurementTombstones>[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<typeof cleanupExpiredMeasurementTombstones>[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<typeof cleanupExpiredMeasurementTombstones>[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.

src/lib/jobs/job-outcome.ts

@@ -71,6 +71,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = 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",

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<TombstoneRow[]>;
+  /** Remove exactly those ids inside `tx`; returns the rows removed. */
+  deleteIds: (tx: Prisma.TransactionClient, ids: string[]) => Promise<number>;
+  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<TombstonePurgeOutcome> {
+  let deleted = 0;
+  const deferred = new Set<string>();
+  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<string, string[]>();
+    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<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   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<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   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<PurgeOutcome> {
+): Promise<TombstonePurgeOutcome> {
   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,

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);

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