From 77bfa2f4e1ca2b9f425af7cae7b25826afbc3fb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 25 Sep 2026 01:44:39 +0200 Subject: [PATCH] perf(records): find the best cumulative day a day at a time The personal-record pass, every half hour for every account, read every live row of each cumulative type into memory to find the best day. With 784 000 active-energy rows on one account (the per-sample rows pile up when the nightly drain cannot finish, #1031) it ran a 524 MB heap out of memory; the old code needed 1.1 GB of heap for that account. It now reads the type in bounded keyset pages and handles one local day at a time. The canonical-source pick it applies is per day, so the day totals and the winner are the same; a tie still keeps the earliest day. Measured on the same account: 189 MB peak heap, the same record. The warm-up gate counts only up to its threshold instead of counting a whole history. `iterateSourcePages` and `iterateDayBuckets` in the consolidation base take the row type as a parameter, and the day grouping can keep every row. --- src/lib/measurements/consolidation-base.ts | 33 ++++-- .../personal-records/pr-detection-worker.ts | 108 ++++++++++------- .../pr-detection-cumulative-paged.test.ts | 111 ++++++++++++++++++ 3 files changed, 197 insertions(+), 55 deletions(-) create mode 100644 tests/integration/pr-detection-cumulative-paged.test.ts diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index 01abebbbe..382f2335e 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/src/lib/measurements/consolidation-base.ts @@ -315,12 +315,14 @@ const CONSOLIDATION_SCAN_PAGE_SIZE = 5000; * The caller's `baseWhere` is AND-combined, so the scope / soft-delete / * grace filters always still apply. */ -export async function* iterateSourcePages( +export async function* iterateSourcePages< + Row extends { id: string; measuredAt: Date } = PerSampleRow, +>( prismaClient: PrismaClient, baseWhere: Prisma.MeasurementWhereInput, scanSelect: Prisma.MeasurementSelect, pageSize: number = CONSOLIDATION_SCAN_PAGE_SIZE, -): AsyncGenerator { +): AsyncGenerator { let cursor: { measuredAt: Date; id: string } | null = null; for (;;) { @@ -344,7 +346,7 @@ export async function* iterateSourcePages( select: scanSelect, orderBy: [{ measuredAt: "asc" }, { id: "asc" }], take: pageSize, - })) as PerSampleRow[]; + })) as unknown as Row[]; if (page.length === 0) return; @@ -364,22 +366,29 @@ export async function* iterateSourcePages( * * The rows arrive in `measuredAt` order and a local calendar day is one * contiguous span of instants, so a day is complete the moment a row from a - * later day shows up. Rows already in the daily-stats shape are skipped, the - * same as `bucketRowsByDay`. `onRow` sees every scanned row, stats rows - * included, for the scan-count log line. + * later day shows up. Rows whose externalId starts with `statsPrefix` (the + * daily-stats shape) are skipped, the same as `bucketRowsByDay`; pass `null` + * to keep every row. `onRow` sees every scanned row, skipped ones included, + * for the scan-count log line. */ -export async function* iterateDayBuckets( - pages: AsyncIterable, +export async function* iterateDayBuckets< + Row extends { measuredAt: Date; externalId: string | null } = PerSampleRow, +>( + pages: AsyncIterable, tz: string, - statsPrefix: string, + statsPrefix: string | null, onRow?: () => void, -): AsyncGenerator<[dateKey: string, rows: PerSampleRow[]], void, void> { +): AsyncGenerator<[dateKey: string, rows: Row[]], void, void> { let currentKey: string | null = null; - let current: PerSampleRow[] = []; + let current: Row[] = []; for await (const page of pages) { for (const row of page) { onRow?.(); - if (row.externalId !== null && row.externalId.startsWith(statsPrefix)) { + if ( + statsPrefix !== null && + row.externalId !== null && + row.externalId.startsWith(statsPrefix) + ) { continue; } const key = dayKeyForUserTz(row.measuredAt, tz); diff --git a/src/lib/personal-records/pr-detection-worker.ts b/src/lib/personal-records/pr-detection-worker.ts index 107fcc704..77739c6a5 100644 --- a/src/lib/personal-records/pr-detection-worker.ts +++ b/src/lib/personal-records/pr-detection-worker.ts @@ -40,6 +40,10 @@ import { prisma as defaultPrisma } from "@/lib/db"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz"; +import { + iterateDayBuckets, + iterateSourcePages, +} from "@/lib/measurements/consolidation-base"; import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; import { measurementTypeEnum } from "@/lib/validations/measurement"; @@ -273,8 +277,12 @@ export async function detectPersonalRecordsForUser( // Warm-up gate. Cheap count first so we don't pull a 30k-row // scan on a brand-new account. + // Bounded: the gate only asks whether there are enough, so it stops + // counting at the threshold instead of counting a whole history (766 000 + // heart-rate rows on a large account, every half hour). const sampleCount = await prisma.measurement.count({ where: { userId, type: metricType, deletedAt: null }, + take: PR_DETECTION_WARMUP_THRESHOLD, }); if (sampleCount < PR_DETECTION_WARMUP_THRESHOLD) continue; @@ -542,18 +550,28 @@ async function findBestMeasurement( * cumulative-day surface. * * Order of operations: - * 1. Read every row for the type (already scoped by the caller's - * warm-up gate to a metric with at least - * `PR_DETECTION_WARMUP_THRESHOLD` samples). - * 2. Collapse to the canonical source per local day. - * 3. Bucket-and-sum the canonical rows per local day; pick the - * winning day by `direction`. + * 1. Read the type's rows one local day at a time, in time order + * (already scoped by the caller's warm-up gate to a metric with at + * least `PR_DETECTION_WARMUP_THRESHOLD` samples). + * 2. Collapse each day to its canonical source. + * 3. Sum the day's canonical rows; keep the winning day by `direction`. * 4. Re-read the winning day's latest slice so the PR row carries a * real row's unit/source/externalId (the day-sum overrides * `value` below). * The unique-index contract is unaffected — a re-run picks the same * day and the same latest-slice timestamp, so `achievedAt` is stable. */ +/** One row of the cumulative-day scan. */ +interface CumulativeRow { + id: string; + value: number; + unit: string; + measuredAt: Date; + source: MeasurementSource; + externalId: string | null; + deviceType: string | null; +} + async function findBestCumulativeDay( prisma: PrismaClient, userId: string, @@ -562,46 +580,50 @@ async function findBestCumulativeDay( tz: string, priorityJson: unknown, ): Promise { - const rows = await prisma.measurement.findMany({ - where: { userId, type, deletedAt: null }, - orderBy: { measuredAt: "asc" }, - select: { - id: true, - value: true, - unit: true, - measuredAt: true, - source: true, - externalId: true, - deviceType: true, - }, - }); - if (rows.length === 0) return null; - + // Read a day at a time, in time order, never the whole history: the + // canonical-source pick is per day, so running it on each day alone gives + // the same rows it gave on the whole list. Loading every row at once was + // unbounded; with 784 000 rows of one cumulative type the pass ran a + // 524 MB heap out of memory (#1031). const metricKey = metricKeyForType(type); - const canonicalRows = metricKey - ? pickCanonicalSourceRows( - rows.map((r) => ({ ...r, type })), - metricKey, - priorityJson, - (d) => dayKeyForUserTz(d, tz), - ).canonicalRows - : rows; - - const byDay = new Map(); - for (const row of canonicalRows) { - const key = dayKeyForUserTz(row.measuredAt, tz); - const slot = byDay.get(key); - if (!slot) { - byDay.set(key, { total: row.value, latest: row.measuredAt }); - } else { - slot.total += row.value; - if (row.measuredAt > slot.latest) slot.latest = row.measuredAt; - } - } - let winnerTotal: number | null = null; let winnerLatest: Date | null = null; - for (const { total, latest } of byDay.values()) { + const days = iterateDayBuckets( + iterateSourcePages( + prisma, + { userId, type, deletedAt: null }, + { + id: true, + value: true, + unit: true, + measuredAt: true, + source: true, + externalId: true, + deviceType: true, + }, + ), + tz, + null, + ); + for await (const [, dayRows] of days) { + const canonicalRows = metricKey + ? pickCanonicalSourceRows( + dayRows.map((r) => ({ ...r, type })), + metricKey, + priorityJson, + (d) => dayKeyForUserTz(d, tz), + ).canonicalRows + : dayRows; + if (canonicalRows.length === 0) continue; + + let total = 0; + let latest = canonicalRows[0]!.measuredAt; + for (const row of canonicalRows) { + total += row.value; + if (row.measuredAt > latest) latest = row.measuredAt; + } + // Days arrive in order, and only a strictly better day replaces the + // winner, so a tie keeps the earliest day, as before. const better = winnerTotal === null || (direction === PersonalRecordDirection.MAX diff --git a/tests/integration/pr-detection-cumulative-paged.test.ts b/tests/integration/pr-detection-cumulative-paged.test.ts new file mode 100644 index 000000000..9bfc38159 --- /dev/null +++ b/tests/integration/pr-detection-cumulative-paged.test.ts @@ -0,0 +1,111 @@ +/** + * #1031 — the best cumulative day, found a day at a time. + * + * `findBestCumulativeDay` read every row of a cumulative type into memory, + * every half hour, for every account. On an account holding 784 000 active + * energy rows it ran a 524 MB heap out of memory. It now reads a local day at + * a time in time order, which gives the same answer because the canonical- + * source pick it applies is per day. + * + * Pinned here at a size that spans several read pages: two sources report + * the same days, the source ladder must decide each day on its own, the + * best day sits in the middle of the history, and no read of the history is + * unbounded. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const { detectPersonalRecordsForUser } = + await import("@/lib/personal-records/pr-detection-worker"); + +const USER = "user-pr-cumulative-paged"; +const DAYS = 40; +const PER_DAY = 300; // 12 000 Apple rows plus a Withings total per day + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: USER, + username: USER, + email: `${USER}@example.test`, + timezone: "UTC", + }, + }); +}); + +describe("best cumulative day across many pages", () => { + it("sums each day's canonical source and finds the best day without an unbounded read", async () => { + const prisma = getPrismaClient(); + const start = Date.UTC(2025, 0, 1, 6); + const rows = []; + for (let day = 0; day < DAYS; day++) { + // Day 23 is the best Apple day; the Withings total on day 5 is larger + // still but loses to Apple on that day by the default ladder, so it + // must not win: if the days were pooled across sources it would. + const perSample = day === 23 ? 40 : 20; + for (let i = 0; i < PER_DAY; i++) { + rows.push({ + userId: USER, + type: "ACTIVITY_STEPS" as const, + value: perSample, + unit: "steps", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(start + day * 86_400_000 + i * 60_000), + externalId: `hk-${day}-${i}`, + }); + } + rows.push({ + userId: USER, + type: "ACTIVITY_STEPS" as const, + value: day === 5 ? 50_000 : 1_000, + unit: "steps", + source: "WITHINGS" as const, + measuredAt: new Date(start + day * 86_400_000 + 12 * 3_600_000), + externalId: `w-${day}`, + }); + } + await prisma.measurement.createMany({ data: rows }); + + // Every read of the steps history, through a client that records it. + const reads: Array = []; + const recording = prisma.$extends({ + query: { + measurement: { + findMany({ args, query }) { + if (JSON.stringify(args.where ?? {}).includes("ACTIVITY_STEPS")) { + reads.push(args.take); + } + return query(args); + }, + }, + }, + }); + + await detectPersonalRecordsForUser(USER, { + silent: true, + prisma: recording as unknown as typeof prisma, + }); + + const record = await prisma.personalRecord.findFirstOrThrow({ + where: { userId: USER, metricType: "ACTIVITY_STEPS" }, + }); + expect(record.value).toBe(40 * PER_DAY); + expect(record.achievedAt.toISOString()).toBe( + new Date(start + 23 * 86_400_000 + (PER_DAY - 1) * 60_000).toISOString(), + ); + + // Several bounded pages, never one read of the whole history. + expect(reads.length).toBeGreaterThan(1); + for (const take of reads) { + expect(take).toBeDefined(); + expect(take!).toBeLessThanOrEqual(5_000); + } + }); +});