fix(records): repair cumulative records in one transaction

The repair for the inflated cumulative records deleted the suspect rows,
committed, and then re-ran the detector. When the detector failed, which
a timeout on a large account or a restart can cause, the records stayed
deleted, and since discovery looks for the suspect rows the account was
never repaired again (#1031).

The delete and the re-derivation now run in one transaction, with the
detector reading through it, so they commit together or not at all. The
transaction gets five minutes; the detector took 19.8 s on the largest
seeded account.

src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts

@@ -8,12 +8,20 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
 
 const deleteMany = vi.fn();
 const findMany = vi.fn();
+// The delete and the re-derivation share one transaction; the stub hands the
+// callback a client that routes to the same spies.
+const txClient = {
+  personalRecord: {
+    deleteMany: (a: unknown) => deleteMany(a),
+  },
+};
 vi.mock("@/lib/db", () => ({
   prisma: {
     personalRecord: {
       deleteMany: (a: unknown) => deleteMany(a),
       findMany: (a: unknown) => findMany(a),
     },
+    $transaction: (fn: (tx: unknown) => Promise<unknown>) => fn(txClient),
   },
 }));
 
@@ -68,8 +76,10 @@ describe("runCumulativePrRederivationForUser", () => {
     });
     // Re-derivation always runs silent — this is data hygiene, not a new
     // achievement, so no push should fire for it.
+    // It reads through the transaction, so it sees the deletion.
     expect(detectPersonalRecordsForUser).toHaveBeenCalledWith("u1", {
       silent: true,
+      prisma: txClient,
     });
     expect(summary).toEqual({ rowsDeleted: 1, rowsReinserted: 1 });
   });

src/lib/personal-records/cumulative-pr-rederivation.ts

@@ -41,7 +41,10 @@ import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
 import { detectPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker";
-import type { MeasurementType } from "@/generated/prisma/client";
+import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+
+/** Budget for one account's delete-and-re-derive transaction. */
+const REDERIVE_TRANSACTION_TIMEOUT_MS = 5 * 60 * 1000;
 
 export const CUMULATIVE_PR_REDERIVE_QUEUE = "cumulative-pr-rederive";
 
@@ -90,22 +93,40 @@ export interface CumulativePrRederiveSummary {
 export async function runCumulativePrRederivationForUser(
   userId: string,
 ): Promise<CumulativePrRederiveSummary> {
-  const { count: rowsDeleted } = await prisma.personalRecord.deleteMany({
-    where: {
-      userId,
-      metricSlot: null,
-      metricType: { in: CUMULATIVE_TYPES_ARRAY },
-      createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+  // One transaction: the suspect rows go and the re-derived ones arrive
+  // together, or neither happens. Deleting first and re-deriving after the
+  // commit lost the records whenever the detector failed (a timeout on a
+  // large account, a restart): discovery looks for the suspect rows, so once
+  // they were gone the account was never repaired again (#1031). The detector
+  // reads through the transaction, so it sees the suspect rows as deleted and
+  // compares against what is left. It only reads and inserts, and sends no
+  // notification when `silent`.
+  const { rowsDeleted, rowsReinserted } = await prisma.$transaction(
+    async (tx) => {
+      const { count } = await tx.personalRecord.deleteMany({
+        where: {
+          userId,
+          metricSlot: null,
+          metricType: { in: CUMULATIVE_TYPES_ARRAY },
+          createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF },
+        },
+      });
+      if (count === 0) return { rowsDeleted: 0, rowsReinserted: 0 };
+      const result = await detectPersonalRecordsForUser(userId, {
+        silent: true,
+        prisma: tx as unknown as PrismaClient,
+      });
+      return {
+        rowsDeleted: count,
+        rowsReinserted: result.inserted + result.ties,
+      };
     },
-  });
-
-  let rowsReinserted = 0;
-  if (rowsDeleted > 0) {
-    const result = await detectPersonalRecordsForUser(userId, {
-      silent: true,
-    });
-    rowsReinserted = result.inserted + result.ties;
-  }
+    // The detector walks every trackable type of the account; on a large
+    // record that is tens of seconds (19.8 s measured for 1.25 million
+    // measurements plus 784 000 energy rows), so the default five seconds
+    // would roll back every repair that matters.
+    { timeout: REDERIVE_TRANSACTION_TIMEOUT_MS, maxWait: 10_000 },
+  );
 
   annotate({
     action: {

tests/integration/cumulative-pr-rederive-atomic.test.ts

@@ -0,0 +1,94 @@
+/**
+ * #1031 — the cumulative-record repair must not lose records when the
+ * re-derivation fails.
+ *
+ * The repair deleted every suspect cumulative record, committed, and only
+ * then re-ran the detector. A detector that failed (a timeout on a large
+ * account, a crash) left the records deleted, and because discovery looks
+ * for the suspect rows the deletion removed, the account was never
+ * repaired again: its step, energy and distance records were simply gone.
+ * The delete and the re-derivation now commit together or not at all.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const detect = vi.hoisted(() => ({ fail: false }));
+vi.mock("@/lib/personal-records/pr-detection-worker", async (importActual) => {
+  const actual =
+    await importActual<
+      typeof import("@/lib/personal-records/pr-detection-worker")
+    >();
+  return {
+    ...actual,
+    detectPersonalRecordsForUser: async (
+      ...args: Parameters<typeof actual.detectPersonalRecordsForUser>
+    ) => {
+      if (detect.fail) throw new Error("detector ran out of time");
+      return actual.detectPersonalRecordsForUser(...args);
+    },
+  };
+});
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const { runCumulativePrRederivationForUser, CUMULATIVE_PR_FIX_CUTOFF } =
+  await import("@/lib/personal-records/cumulative-pr-rederivation");
+
+const USER = "user-pr-rederive-atomic";
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  detect.fail = false;
+  await prisma.user.create({
+    data: { id: USER, username: USER, email: `${USER}@example.test` },
+  });
+  const before = new Date(CUMULATIVE_PR_FIX_CUTOFF.getTime() - 86_400_000);
+  await prisma.personalRecord.create({
+    data: {
+      userId: USER,
+      metricType: "ACTIVITY_STEPS",
+      direction: "MAX",
+      value: 99_999,
+      unit: "steps",
+      achievedAt: new Date("2025-03-01T12:00:00Z"),
+      createdAt: before,
+    },
+  });
+  // Enough steps for the detector's warm-up gate, one clear best day.
+  await prisma.measurement.createMany({
+    data: Array.from({ length: 10 }, (_, day) => ({
+      userId: USER,
+      type: "ACTIVITY_STEPS" as const,
+      value: day === 4 ? 12_000 : 5_000,
+      unit: "steps",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: new Date(Date.UTC(2025, 3, 1 + day, 12)),
+      externalId: `s-${day}`,
+    })),
+  });
+});
+
+describe("runCumulativePrRederivationForUser", () => {
+  it("keeps the suspect record when the re-derivation fails", async () => {
+    detect.fail = true;
+    await expect(runCumulativePrRederivationForUser(USER)).rejects.toThrow(
+      "detector ran out of time",
+    );
+    const left = await getPrismaClient().personalRecord.findMany({
+      where: { userId: USER },
+    });
+    expect(left.map((r) => r.value)).toEqual([99_999]);
+  });
+
+  it("replaces the suspect record with the re-derived one when it succeeds", async () => {
+    const summary = await runCumulativePrRederivationForUser(USER);
+    expect(summary.rowsDeleted).toBe(1);
+    const left = await getPrismaClient().personalRecord.findMany({
+      where: { userId: USER },
+    });
+    expect(left.map((r) => r.value)).toEqual([12_000]);
+  });
+});