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

@@ -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<PerSampleRow[], void, void> {
+): AsyncGenerator<Row[], void, void> {
   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<readonly PerSampleRow[]>,
+export async function* iterateDayBuckets<
+  Row extends { measuredAt: Date; externalId: string | null } = PerSampleRow,
+>(
+  pages: AsyncIterable<readonly Row[]>,
   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);

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<MeasurementCandidate | null> {
-  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<string, { total: number; latest: Date }>();
-  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<CumulativeRow>(
+      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

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<number | undefined> = [];
+    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);
+    }
+  });
+});