From 7071fa91af41577b27da2687dd375242f27399cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 25 Sep 2026 01:46:09 +0200 Subject: [PATCH] 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. --- .../cumulative-pr-rederivation.test.ts | 10 ++ .../cumulative-pr-rederivation.ts | 53 +++++++---- .../cumulative-pr-rederive-atomic.test.ts | 94 +++++++++++++++++++ 3 files changed, 141 insertions(+), 16 deletions(-) create mode 100644 tests/integration/cumulative-pr-rederive-atomic.test.ts diff --git a/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts b/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts index e4172f434..bcda29b20 100644 --- a/src/lib/personal-records/__tests__/cumulative-pr-rederivation.test.ts +++ b/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) => 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 }); }); diff --git a/src/lib/personal-records/cumulative-pr-rederivation.ts b/src/lib/personal-records/cumulative-pr-rederivation.ts index a6202ac24..b93f15755 100644 --- a/src/lib/personal-records/cumulative-pr-rederivation.ts +++ b/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 { - 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: { diff --git a/tests/integration/cumulative-pr-rederive-atomic.test.ts b/tests/integration/cumulative-pr-rederive-atomic.test.ts new file mode 100644 index 000000000..77c8509a2 --- /dev/null +++ b/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 + ) => { + 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]); + }); +});