Merge branch 'wave/v125-perf' into release/v1.25.0

prisma/migrations/0210_v125_measurement_useractivity_index/migration.sql

@@ -0,0 +1,30 @@
+-- v1.25 — type-less `(user_id, measured_at)` supporting index on `measurements`.
+--
+-- The bulk feature read (`extractFeatures`, src/lib/insights/features.ts) and
+-- the consolidation scans walk a user's measurements ordered by `measured_at`
+-- with no `type` predicate. The existing composite indexes cover
+-- `(user_id, type, measured_at)`, `(user_id, source, measured_at)`, and
+-- `(user_id, updated_at, id)` — none serves a type-less `(user_id, measured_at)`
+-- range scan as a prefix. Without this index Postgres filters by `user_id` then
+-- sorts the user's entire measurement set in memory on each of those reads,
+-- which on a multi-year HealthKit tenant is hundreds of thousands of rows.
+--
+-- The composite `(user_id, measured_at)` index lets those reads stream straight
+-- off the index in time order, which the newest-first row caps then bound.
+--
+-- Additive + order-safe: a pure `CREATE INDEX IF NOT EXISTS`, no column add, no
+-- backfill, no rewrite of `measurements`.
+--
+-- Lock note: this is a plain, non-`CONCURRENTLY` index build, so it runs inside
+-- Prisma's migration transaction and holds an ACCESS EXCLUSIVE lock on
+-- `measurements` for the build's duration. The index covers every row (no
+-- partial predicate), so on the largest multi-year HealthKit tenants the build
+-- is slow; run the deploy at a low-traffic window on such deployments. A
+-- `CONCURRENTLY` build cannot run inside the migration transaction; a plain
+-- build is acceptable at this table size in the maintenance window.
+--
+-- Reversibility: down migration is
+--   DROP INDEX IF EXISTS "measurements_user_id_measured_at_idx";
+
+CREATE INDEX IF NOT EXISTS "measurements_user_id_measured_at_idx"
+  ON "measurements" ("user_id", "measured_at");

prisma/schema.prisma

@@ -1353,6 +1353,12 @@ model Measurement {
   // (Postgres semantics) so manual entries don't collide here.
   @@unique([userId, type, source, externalId])
   @@index([userId, type, measuredAt])
+  // v1.25 — type-less `(userId, measuredAt)` range scan. The bulk feature read
+  // (`extractFeatures`) and consolidation scans walk a user's measurements by
+  // time with no `type` predicate; `(userId, type, measuredAt)` can't serve
+  // that as a prefix, so without this index Postgres scans the user's whole set
+  // and sorts in memory. See migration 0210.
+  @@index([userId, measuredAt])
   // v1.15.13 — leads with `(userId, source)` so the management list's
   // source filter + recency sort is an index range scan instead of a
   // `userId` scan + in-memory sort. See migration 0136.

src/app/api/admin/audit-log/route.ts

@@ -3,6 +3,7 @@ import { apiHandler, requireAdmin } from "@/lib/api-handler";
 import { apiSuccess, returnAllZodIssues } from "@/lib/api-response";
 import { annotate } from "@/lib/logging/context";
 import { redactSecrets } from "@/lib/logging/redact";
+import { getAuditLogRetentionDays } from "@/lib/jobs/audit-log-cleanup";
 import { NextRequest } from "next/server";
 import { z } from "zod/v4";
 
@@ -39,7 +40,9 @@ export const dynamic = "force-dynamic";
  *   - actor    — substring match against `userId` OR `user.username`
  *   - action   — exact match (e.g. `auth.login.failed`)
  *   - target   — substring match against the JSON-encoded `details` field
- *   - since    — ISO-8601 lower bound on `createdAt`
+ *   - since    — ISO-8601 lower bound on `createdAt`. Absent, the scan
+ *                floors at the audit-log retention horizon so an
+ *                open-ended sweep can't seqscan the table unbounded.
  *   - until    — ISO-8601 upper bound on `createdAt`
  *   - filter   — legacy shortcut: `auth` restricts to `auth.*` actions
  *
@@ -117,15 +120,25 @@ export const GET = apiHandler(async (request: NextRequest) => {
     ands.push({ details: { contains: parsed.target } });
   }
 
-  if (parsed.since || parsed.until) {
-    const range: { gte?: Date; lte?: Date } = {};
-    if (parsed.since) range.gte = new Date(parsed.since);
-    if (parsed.until) range.lte = new Date(parsed.until);
-    ands.push({ createdAt: range });
-  }
+  // Bound the lower edge of the scan. `actor`/`target` are substring
+  // matches (seqscans) and the exact `count` below runs a second pass over
+  // the same predicate, so an open-ended history sweep on this admin-only,
+  // low-frequency surface would scan the table twice with no floor. The
+  // cleanup job already prunes rows past the retention horizon, so aligning
+  // the default floor to that same horizon bounds both passes without ever
+  // hiding a row the deployment still keeps. An explicit `since` overrides.
+  const lowerBound = parsed.since
+    ? new Date(parsed.since)
+    : new Date(Date.now() - getAuditLogRetentionDays() * 86_400_000);
+  const createdAt: { gte: Date; lte?: Date } = { gte: lowerBound };
+  if (parsed.until) createdAt.lte = new Date(parsed.until);
 
-  const where: Record<string, unknown> | undefined =
-    ands.length === 0 ? undefined : ands.length === 1 ? ands[0] : { AND: ands };
+  // Keep `createdAt` a top-level sibling of the optional filter clause so a
+  // single substring/exact filter stays flat (no `AND` wrapper); the
+  // collapse below only folds the independent filter `ands` entries.
+  const filterWhere: Record<string, unknown> =
+    ands.length === 0 ? {} : ands.length === 1 ? ands[0] : { AND: ands };
+  const where: Record<string, unknown> = { ...filterWhere, createdAt };
 
   const [entries, total] = await Promise.all([
     prisma.auditLog.findMany({

src/lib/__tests__/require-auth-bearer.test.ts

@@ -105,9 +105,11 @@ describe("requireAuth — Bearer token path", () => {
       where: { id: "token-1" },
       data: { lastUsedAt: expect.any(Date) },
     });
-    expect(auditLog).toHaveBeenCalledWith(
+    // v1.25 — the success path no longer writes a per-request audit row; the
+    // wide event carries `auth_method: "bearer"` + `user_id` instead.
+    expect(auditLog).not.toHaveBeenCalledWith(
       "auth.bearer.success",
-      expect.objectContaining({ userId: "user-1" }),
+      expect.anything(),
     );
   });
 
@@ -219,9 +221,10 @@ describe("requireAuth — Bearer token path", () => {
     const ctx = await requireAuth();
     expect(ctx.user.id).toBe("user-1");
     expect(ctx.session.id).toBe("token-5a");
-    expect(auditLog).toHaveBeenCalledWith(
+    // v1.25 — no per-request success audit row on the accept path.
+    expect(auditLog).not.toHaveBeenCalledWith(
       "auth.bearer.success",
-      expect.objectContaining({ userId: "user-1" }),
+      expect.anything(),
     );
   });
 

src/lib/ai/coach/persistence.ts

@@ -253,11 +253,21 @@ export async function recordProactiveNudge(
   });
 }
 
+// Very long threads previously decrypted every message on each open; the
+// newest messages up to this cap cover the rendered window without the
+// unbounded per-open AES-decrypt cost. The response shape is unchanged —
+// the messages array still arrives oldest->newest (see the reverse below).
+const CONVERSATION_MESSAGE_DETAIL_CAP = 200;
+
 /**
- * Fetch one conversation + every message, decrypting each body on
+ * Fetch one conversation + its messages, decrypting each body on
  * read. Returns null when the conversation does not exist OR when the
  * supplied `userId` does not own it — callers should map both cases to
  * a 404 to avoid an existence-leak side channel.
+ *
+ * Only the newest `CONVERSATION_MESSAGE_DETAIL_CAP` messages are loaded
+ * and decrypted; the result stays ascending so the response envelope is
+ * byte-for-byte the same shape callers already consume.
  */
 export async function fetchConversationWithMessages(
   userId: string,
@@ -267,13 +277,18 @@ export async function fetchConversationWithMessages(
     where: { id: conversationId, userId },
     include: {
       messages: {
-        orderBy: { createdAt: "asc" },
+        // Fetch the newest N first, then restore ascending order in code
+        // so the unbounded per-open decrypt cost is capped without
+        // changing the oldest->newest contract the client renders.
+        orderBy: { createdAt: "desc" },
+        take: CONVERSATION_MESSAGE_DETAIL_CAP,
       },
     },
   });
   if (!row) return null;
 
-  const messages: CoachMessageDTO[] = row.messages.map((m) => ({
+  const orderedMessages = [...row.messages].reverse();
+  const messages: CoachMessageDTO[] = orderedMessages.map((m) => ({
     id: m.id,
     role: m.role as CoachMessageRole,
     content: decryptFromBytes(m.encryptedContent),

src/lib/api-handler.ts

@@ -406,11 +406,11 @@ async function authenticateBearer(
     }
   }
 
-  auditLog("auth.bearer.success", {
-    userId: user.id,
-    details: { tokenId },
-  }).catch(() => {});
-
+  // v1.25 — no per-request success audit row. The polling iOS client drove a
+  // constant INSERT + pool checkout on every authenticated Bearer request; the
+  // wide event below already records `auth_method: "bearer"` + `user_id`, so the
+  // success path stays fully observable without the write churn. The failure
+  // path keeps its audit row.
   const evt = getEvent();
   if (evt) {
     evt.setAuth({

src/lib/insights/features.ts

@@ -752,6 +752,16 @@ const WORKOUT_WINDOW_DAYS = 90;
 
 const MS_PER_DAY = 24 * 60 * 60 * 1000;
 
+/**
+ * v1.25 — newest-first row cap on the bulk measurement read in
+ * `extractFeatures`. Mirrors the Coach snapshot's `SNAPSHOT_MEASUREMENT_ROW_CAP`
+ * (6000): the prompt / insights aggregates only ever fold this read into a
+ * bounded set of summaries, so a year of dense PULSE / glucose rows is wasted
+ * I/O on the shared pool for a heavy-data tenant. Full-history extremes come
+ * from `readAllTimeExtremes`, not this read, so the cap stays correct.
+ */
+const FEATURE_MEASUREMENT_ROW_CAP = 6000;
+
 /**
  * Strip control chars + collapse whitespace, then bound the length, before a
  * user-supplied label can reach the briefing prompt. Mirrors the labs /
@@ -1010,24 +1020,39 @@ export async function extractFeatures(
     typeof sinceDays === "number" && sinceDays > 0
       ? new Date(now - sinceDays * 24 * 60 * 60 * 1000)
       : null;
-  const measurements = await prisma.measurement.findMany({
-    where: sinceCutoff
-      ? { userId, measuredAt: { gte: sinceCutoff }, deletedAt: null }
-      : { userId, deletedAt: null },
-    orderBy: { measuredAt: "asc" },
-    // Project only the columns every downstream consumer reads (`byType`,
-    // `summarize`, BP pairing, and `reconstructSleepNights`'s `SleepStageRow`).
-    // The PULSE / glucose windows are 200k-row-class; pulling every column
-    // (notes, externalId, …) is pure wasted I/O on the shared Prisma pool.
-    select: {
-      type: true,
-      value: true,
-      measuredAt: true,
-      sleepStage: true,
-      source: true,
-      deviceType: true,
-    },
-  });
+  // v1.25 — newest-first cap on the bulk feature read, mirroring the Coach
+  // snapshot (`SNAPSHOT_MEASUREMENT_ROW_CAP`). Callers pass windows of 365-400
+  // days, and PULSE / glucose are 200k-row-class types, so an uncapped read
+  // pulled hundreds of thousands of rows per Coach turn and nightly briefing
+  // only to fold them into a handful of summaries. Read newest-first, cap, then
+  // reverse so every downstream consumer (`byType`, `summarize`, BP pairing,
+  // sleep reconstruction, oldest/newest span) still sees ascending order. The
+  // cap is safe even unbounded: the genuine all-time extremes are sourced
+  // separately via `readAllTimeExtremes` on the bounded branch, so capping only
+  // sheds the deepest rows of the bulk read and never relabels an extreme.
+  const measurements = await prisma.measurement
+    .findMany({
+      where: sinceCutoff
+        ? { userId, measuredAt: { gte: sinceCutoff }, deletedAt: null }
+        : { userId, deletedAt: null },
+      orderBy: { measuredAt: "desc" },
+      take: FEATURE_MEASUREMENT_ROW_CAP,
+      // Project only the columns every downstream consumer reads (`byType`,
+      // `summarize`, BP pairing, and `reconstructSleepNights`'s `SleepStageRow`).
+      // The PULSE / glucose windows are 200k-row-class; pulling every column
+      // (notes, externalId, …) is pure wasted I/O on the shared Prisma pool.
+      select: {
+        type: true,
+        value: true,
+        measuredAt: true,
+        sleepStage: true,
+        source: true,
+        deviceType: true,
+      },
+    })
+    // Restore ascending order so order-sensitive consumers (oldest/newest span,
+    // BP pairing, sleep reconstruction) see the same shape as before the cap.
+    .then((rows) => rows.reverse());
 
   // v1.18.11 P1 — when the bulk read is bounded to a recent window, the
   // windowed `summarize()` no longer covers the full history, so the `allTime*`

src/lib/jobs/__tests__/mean-consolidation-queue.test.ts

@@ -41,6 +41,8 @@ describe("reminder-worker — mean-consolidation wiring", () => {
   });
 
   it("fires the boot-discovery enqueue helper", () => {
-    expect(source).toMatch(/await enqueueBootTimeMeanConsolidation\(\)/);
+    // The boot wiring passes a stagger offset so the heavy backfills don't all
+    // drain in parallel onto one tenant; allow an optional argument here.
+    expect(source).toMatch(/await enqueueBootTimeMeanConsolidation\([^)]*\)/);
   });
 });

src/lib/jobs/backup/__tests__/measurement-page.test.ts

@@ -0,0 +1,238 @@
+/**
+ * Round-trip + paging contract for the backup measurement reader.
+ *
+ * Runs without a database: the keyset pager takes a `fetchPage` callback,
+ * so an in-memory fixture stands in for Prisma. This pins the two things
+ * the heap-bounding refactor must not break:
+ *   1. the keyset scan visits every row exactly once, in id order,
+ *      across page boundaries and the exact-multiple edge case;
+ *   2. a measurement survives the full serialize → encrypt-blob shape →
+ *      `parseBackupPayload` → restore-mapping round-trip unchanged,
+ *      including a soft-deleted (tombstoned) row, which both writers keep.
+ */
+import { describe, it, expect } from "vitest";
+import {
+  collectPagedMeasurements,
+  toWeeklyBackupMeasurement,
+  sortWeeklyMeasurementsDesc,
+  type WeeklyMeasurementRow,
+} from "@/lib/jobs/backup/measurement-page";
+import { parseBackupPayload } from "@/lib/validations/backup";
+
+/** Build an id-ascending fixture; `id` is zero-padded so string order == insert order. */
+function makeRows(n: number): WeeklyMeasurementRow[] {
+  return Array.from({ length: n }, (_, i) => ({
+    id: `m${String(i).padStart(4, "0")}`,
+    type: "WEIGHT",
+    value: 70 + i,
+    unit: "kg",
+    source: "MANUAL",
+    measuredAt: new Date(Date.UTC(2026, 0, 1 + i, 8, 0, 0)),
+    notes: i % 2 === 0 ? `note ${i}` : null,
+    // notesEncrypted left null so readNote returns the plaintext fallback
+    // and the pure test needs no encryption key.
+    notesEncrypted: null,
+  }));
+}
+
+/** In-memory keyset fetcher over an id-sorted fixture. */
+function fakeFetcher(rows: WeeklyMeasurementRow[]) {
+  const sorted = [...rows].sort((a, b) =>
+    a.id < b.id ? -1 : a.id > b.id ? 1 : 0,
+  );
+  return (afterId: string | null, take: number) => {
+    const start = afterId ? sorted.findIndex((r) => r.id === afterId) + 1 : 0;
+    return Promise.resolve(sorted.slice(start, start + take));
+  };
+}
+
+describe("collectPagedMeasurements", () => {
+  it("visits every row exactly once, in id order, across pages", async () => {
+    const rows = makeRows(25);
+    const out = await collectPagedMeasurements({
+      fetchPage: fakeFetcher(rows),
+      project: (r) => r.id,
+      pageSize: 10,
+    });
+    expect(out).toEqual(rows.map((r) => r.id));
+    expect(new Set(out).size).toBe(out.length);
+  });
+
+  it("terminates on the exact-multiple boundary without dropping rows", async () => {
+    const rows = makeRows(20);
+    const out = await collectPagedMeasurements({
+      fetchPage: fakeFetcher(rows),
+      project: (r) => r.id,
+      pageSize: 10,
+    });
+    expect(out).toHaveLength(20);
+    expect(out[out.length - 1]).toBe("m0019");
+  });
+
+  it("handles an empty set", async () => {
+    const out = await collectPagedMeasurements({
+      fetchPage: fakeFetcher([]),
+      project: (r) => r.id,
+      pageSize: 10,
+    });
+    expect(out).toEqual([]);
+  });
+});
+
+describe("weekly backup measurement round-trip", () => {
+  it("serialize → parseBackupPayload → restore-map equals originals", async () => {
+    const rows: WeeklyMeasurementRow[] = [
+      {
+        id: "m0001",
+        type: "WEIGHT",
+        value: 82.4,
+        unit: "kg",
+        source: "MANUAL",
+        measuredAt: new Date("2026-02-10T07:30:00.000Z"),
+        notes: "morning",
+        notesEncrypted: null,
+      },
+      {
+        id: "m0002",
+        type: "PULSE",
+        value: 58,
+        unit: "bpm",
+        source: "WITHINGS",
+        measuredAt: new Date("2026-02-11T06:00:00.000Z"),
+        notes: null,
+        notesEncrypted: null,
+      },
+      {
+        // A tombstoned row: the weekly writer applies no deletedAt filter,
+        // so a soft-deleted measurement is still backed up. The narrow
+        // weekly projection does not carry deletedAt (the restore does not
+        // recreate it), but the row must still appear in the payload.
+        id: "m0003",
+        type: "BLOOD_GLUCOSE",
+        value: 95,
+        unit: "mg/dL",
+        source: "IMPORT",
+        measuredAt: new Date("2026-02-12T12:00:00.000Z"),
+        notes: "fasting",
+        notesEncrypted: null,
+      },
+    ];
+
+    const measurements = sortWeeklyMeasurementsDesc(
+      await collectPagedMeasurements({
+        fetchPage: fakeFetcher(rows),
+        project: toWeeklyBackupMeasurement,
+        pageSize: 2,
+      }),
+    );
+
+    // Same envelope the writer produces, then parse it back exactly as the
+    // admin restore route does.
+    const backupJson = JSON.stringify({
+      schemaVersion: "1",
+      exportedAt: new Date("2026-02-13T00:00:00.000Z").toISOString(),
+      userId: "user-1",
+      measurements,
+      medications: [],
+      intakeEvents: [],
+      moodEntries: [],
+    });
+    const payload = parseBackupPayload(backupJson);
+
+    expect(payload.measurements).toHaveLength(rows.length);
+
+    // measuredAt-desc ordering preserved (newest first).
+    expect(payload.measurements.map((m) => m.measuredAt)).toEqual([
+      "2026-02-12T12:00:00.000Z",
+      "2026-02-11T06:00:00.000Z",
+      "2026-02-10T07:30:00.000Z",
+    ]);
+
+    // Every original row survives the round-trip by value (the tombstoned
+    // m0003 included).
+    for (const original of rows) {
+      const restored = payload.measurements.find(
+        (m) => m.measuredAt === original.measuredAt.toISOString(),
+      );
+      expect(restored, `row ${original.id} missing from payload`).toBeDefined();
+      expect(restored!.type).toBe(original.type);
+      expect(restored!.value).toBe(original.value);
+      expect(restored!.unit).toBe(original.unit);
+      expect(restored!.source).toBe(original.source);
+      expect(restored!.notes ?? null).toBe(original.notes ?? null);
+    }
+  });
+});
+
+describe("offhost backup measurement round-trip", () => {
+  it("full-row JSON dump → parse preserves all columns incl. soft-deleted", async () => {
+    // The off-host DR dump keeps the full row shape (identity projection).
+    // Mirror the writer's select output for a live row and a tombstoned row.
+    const liveRow = {
+      id: "m0001",
+      userId: "user-1",
+      type: "WEIGHT",
+      value: 80.1,
+      valueMin: null,
+      valueMax: null,
+      unit: "kg",
+      source: "MANUAL",
+      measuredAt: new Date("2026-03-01T07:00:00.000Z"),
+      notes: null,
+      notesEncrypted: null,
+      externalId: null,
+      externalSourceVersion: null,
+      glucoseContext: null,
+      sleepStage: null,
+      rhythmClassification: null,
+      deviceType: "scale",
+      syncVersion: 1,
+      deletedAt: null,
+      createdAt: new Date("2026-03-01T07:00:01.000Z"),
+      updatedAt: new Date("2026-03-01T07:00:01.000Z"),
+    };
+    const tombstonedRow = {
+      ...liveRow,
+      id: "m0002",
+      value: 81.2,
+      measuredAt: new Date("2026-03-02T07:00:00.000Z"),
+      deletedAt: new Date("2026-03-03T09:00:00.000Z"),
+    };
+
+    const measurements = await collectPagedMeasurements({
+      fetchPage: fakeFetcher([
+        liveRow,
+        tombstonedRow,
+      ] as unknown as WeeklyMeasurementRow[]),
+      project: (row) => row,
+      pageSize: 1,
+    });
+
+    const payload = JSON.parse(
+      JSON.stringify({
+        exportedAt: new Date("2026-03-04T00:00:00.000Z").toISOString(),
+        userId: "user-1",
+        measurements,
+        medications: [],
+        intakeEvents: [],
+        moodEntries: [],
+        cycleProfile: null,
+        cycles: [],
+        cycleDayLogs: [],
+      }),
+    ) as { measurements: Array<Record<string, unknown>> };
+
+    expect(payload.measurements).toHaveLength(2);
+
+    // The soft-deleted row survives with its tombstone instant intact —
+    // the DR snapshot must include it.
+    const restoredTombstone = payload.measurements.find(
+      (m) => m.id === "m0002",
+    )!;
+    expect(restoredTombstone.deletedAt).toBe("2026-03-03T09:00:00.000Z");
+
+    // Every scalar column the writer selected is present on the dump.
+    const expectedKeys = Object.keys(liveRow).sort();
+    expect(Object.keys(payload.measurements[0]).sort()).toEqual(expectedKeys);
+  });
+});

src/lib/jobs/backup/measurement-page.ts

@@ -0,0 +1,126 @@
+/**
+ * Keyset-paged measurement reader shared by the two backup writers.
+ *
+ * Both writers — the weekly on-host `DataBackup` snapshot
+ * (`reminder/backup-handlers.ts`) and the nightly off-host DR dump
+ * (`offhost-backup.ts`) — used to load a user's ENTIRE measurement set
+ * as full ORM rows into one array and then run a single
+ * `JSON.stringify` over the whole payload. On a heavy multi-year tenant
+ * that holds every measurement row plus the giant serialized string in
+ * memory at the same instant, which is the dominant heap spike of either
+ * job and scales linearly with history depth.
+ *
+ * This reader walks the set in id-ascending keyset pages: at most one
+ * page of rows is resident at a time, and each page is projected to its
+ * compact backup shape before the next page is fetched. Transient peak
+ * heap is bounded by the page size, not by the size of the user's
+ * history. The accumulated output is the compact projection the payload
+ * needs anyway — not the wide ORM rows.
+ *
+ * Keyset (cursor on `id`) rather than `skip`/`take` offset paging so the
+ * scan stays O(rows) instead of O(rows^2) as the offset grows, and so a
+ * concurrent insert can't shift rows across a page boundary.
+ */
+import { readNote } from "@/lib/crypto/note-cipher";
+
+/**
+ * Page size for the backup keyset scan. Large enough that the per-page
+ * round-trip overhead stays negligible, small enough that a single page
+ * of rows is a bounded slice of any tenant's history.
+ */
+export const MEASUREMENT_BACKUP_PAGE_SIZE = 2000;
+
+/**
+ * Walk a keyset-paged result set, projecting each row to its compact
+ * backup shape as it arrives. `fetchPage` returns the next page of rows
+ * after `afterId` (exclusive, id-ascending), bounded to `take` rows;
+ * returning fewer than `take` rows signals the final page. The generic
+ * stays free of any Prisma dependency so the paging contract is unit
+ * testable without a database.
+ */
+export async function collectPagedMeasurements<
+  Row extends { id: string },
+  Out,
+>(args: {
+  fetchPage: (afterId: string | null, take: number) => Promise<Row[]>;
+  project: (row: Row) => Out;
+  pageSize?: number;
+}): Promise<Out[]> {
+  const { fetchPage, project } = args;
+  const pageSize = args.pageSize ?? MEASUREMENT_BACKUP_PAGE_SIZE;
+  const out: Out[] = [];
+  let afterId: string | null = null;
+  for (;;) {
+    const page = await fetchPage(afterId, pageSize);
+    if (page.length === 0) break;
+    for (const row of page) out.push(project(row));
+    if (page.length < pageSize) break;
+    afterId = page[page.length - 1].id;
+  }
+  return out;
+}
+
+/**
+ * Narrow row shape the weekly snapshot reads — exactly the columns the
+ * admin restore path recreates a measurement from
+ * (`api/admin/backups/[id]/restore`): `type`, `value`, `unit`, `source`,
+ * `measuredAt`, plus the note pair so the decrypted note rides into the
+ * (whole-blob-encrypted) payload. `id` rides along only as the keyset
+ * cursor.
+ */
+export interface WeeklyMeasurementRow {
+  id: string;
+  type: string;
+  value: number;
+  unit: string;
+  source: string;
+  measuredAt: Date;
+  notes: string | null;
+  notesEncrypted: Uint8Array | null;
+}
+
+/** Compact measurement shape the weekly `DataBackup` JSON carries. */
+export interface WeeklyBackupMeasurement {
+  type: string;
+  value: number;
+  unit: string;
+  measuredAt: string;
+  source: string;
+  notes: string | null;
+}
+
+/**
+ * Project a narrow measurement row into the weekly backup shape. The
+ * note is decrypted here so the restore re-encrypts on re-insert — the
+ * same contract the previous inline writer used. Shared with the writer
+ * so the round-trip test exercises the real projection.
+ */
+export function toWeeklyBackupMeasurement(
+  row: WeeklyMeasurementRow,
+): WeeklyBackupMeasurement {
+  return {
+    type: row.type,
+    value: row.value,
+    unit: row.unit,
+    measuredAt: row.measuredAt.toISOString(),
+    source: row.source,
+    notes: readNote(row.notesEncrypted, row.notes),
+  };
+}
+
+/**
+ * Order the collected weekly measurements newest-first to match the
+ * pre-paging `orderBy: { measuredAt: "desc" }` output. The keyset scan
+ * reads id-ascending for a stable cursor; this restores the historical
+ * array order. Operates on the compact projections, so it does not
+ * reintroduce the wide-row heap spike. ISO-8601 strings sort
+ * lexicographically in chronological order, so a string compare is the
+ * chronological compare.
+ */
+export function sortWeeklyMeasurementsDesc(
+  measurements: WeeklyBackupMeasurement[],
+): WeeklyBackupMeasurement[] {
+  return measurements.sort((a, b) =>
+    a.measuredAt < b.measuredAt ? 1 : a.measuredAt > b.measuredAt ? -1 : 0,
+  );
+}

src/lib/jobs/dense-intraday-retention.ts

@@ -199,7 +199,8 @@ export async function enqueueBootTimeDenseIntradayRetention(): Promise<{
         retryDelay: 60,
         retryBackoff: true,
         // Defer past the boot storm so the drain never contends with
-        // startup + foreground for the shared connection pool.
+        // startup + foreground for the shared connection pool. This drain is
+        // the furthest-out stage among the boot backfills by design.
         startAfter: DENSE_INTRADAY_RETENTION_BOOT_DELAY_SECONDS,
         singletonKey: `dense-intraday-retention|${userId}`,
       });

src/lib/jobs/mean-consolidation.ts

@@ -82,7 +82,13 @@ export async function runMeanConsolidationForUser(
  * Best-effort: errors are returned through the result value so the
  * worker boot never fails because of a consolidation miss.
  */
-export async function enqueueBootTimeMeanConsolidation(): Promise<{
+export async function enqueueBootTimeMeanConsolidation(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this consolidation does not drain in
+  // parallel with the other boot backfills onto one heavy tenant. Default 0
+  // keeps immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -127,6 +133,7 @@ export async function enqueueBootTimeMeanConsolidation(): Promise<{
         retryDelay: 60,
         retryBackoff: true,
         singletonKey: `mean-consolidation|${userId}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
       });
       if (jobId) {
         enqueued += 1;

src/lib/jobs/note-encryption-backfill.ts

@@ -175,7 +175,13 @@ export async function runNoteEncryptionBackfillForUser(
  * set. pg-boss `singletonKey` coalesces duplicate sends. Best-effort: errors
  * come back through the result value so worker boot never fails on a miss.
  */
-export async function enqueueBootTimeNoteEncryptionBackfill(): Promise<{
+export async function enqueueBootTimeNoteEncryptionBackfill(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this migration does not drain in parallel
+  // with the other boot backfills onto one heavy tenant. Default 0 keeps
+  // immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -219,6 +225,7 @@ export async function enqueueBootTimeNoteEncryptionBackfill(): Promise<{
         retryDelay: 60,
         retryBackoff: true,
         singletonKey: `note-encryption-backfill|${userId}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
       });
       if (jobId) {
         enqueued += 1;

src/lib/jobs/offhost-backup.ts

@@ -20,6 +20,10 @@
 import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto";
 import type { PrismaClient } from "@/generated/prisma/client";
 import { getEvent } from "@/lib/logging/context";
+import {
+  collectPagedMeasurements,
+  MEASUREMENT_BACKUP_PAGE_SIZE,
+} from "@/lib/jobs/backup/measurement-page";
 
 const ALGORITHM = "aes-256-gcm";
 const IV_LENGTH = 12;
@@ -226,7 +230,51 @@ export async function runOffhostBackup(
         // includes soft-deleted rows because this is the DR snapshot,
         // not a user-facing export — see
         // `/api/export/full-backup/route.ts` for the symmetric exclusion.
-        prisma.measurement.findMany({ where: { userId: user.id } }),
+        //
+        // Keyset-paged narrow read instead of one full-set findMany: a
+        // heavy multi-year tenant otherwise loads every measurement ORM
+        // row into a single array alongside the giant JSON.stringify
+        // below, the worker's dominant heap spike. The DR dump keeps the
+        // FULL row shape, so the select lists every scalar column (no
+        // relations) — identical bytes to the prior default findMany —
+        // while the page reader bounds peak heap to the page size. No
+        // `deletedAt` filter: the tombstoned rows stay in the snapshot.
+        collectPagedMeasurements({
+          fetchPage: (afterId, take) =>
+            prisma.measurement.findMany({
+              where: {
+                userId: user.id,
+                ...(afterId ? { id: { gt: afterId } } : {}),
+              },
+              select: {
+                id: true,
+                userId: true,
+                type: true,
+                value: true,
+                valueMin: true,
+                valueMax: true,
+                unit: true,
+                source: true,
+                measuredAt: true,
+                notes: true,
+                notesEncrypted: true,
+                externalId: true,
+                externalSourceVersion: true,
+                glucoseContext: true,
+                sleepStage: true,
+                rhythmClassification: true,
+                deviceType: true,
+                syncVersion: true,
+                deletedAt: true,
+                createdAt: true,
+                updatedAt: true,
+              },
+              orderBy: { id: "asc" },
+              take,
+            }),
+          project: (row) => row,
+          pageSize: MEASUREMENT_BACKUP_PAGE_SIZE,
+        }),
         prisma.medication.findMany({
           where: { userId: user.id },
           include: { schedules: true },

src/lib/jobs/reminder/backup-handlers.ts

@@ -7,10 +7,15 @@
 import { type Job } from "pg-boss";
 import { recordError } from "@/lib/jobs/worker-status";
 import { encrypt } from "@/lib/crypto";
-import { readNote } from "@/lib/crypto/note-cipher";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { buildCycleBackupSection } from "@/lib/cycle/backup";
 import { runOffhostBackup } from "@/lib/jobs/offhost-backup";
+import {
+  collectPagedMeasurements,
+  MEASUREMENT_BACKUP_PAGE_SIZE,
+  sortWeeklyMeasurementsDesc,
+  toWeeklyBackupMeasurement,
+} from "@/lib/jobs/backup/measurement-page";
 import { getWorkerPrisma } from "./shared";
 
 export interface DataBackupPayload {
@@ -61,10 +66,41 @@ export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
         try {
           const [measurements, medications, intakeEvents, moodEntries, cycle] =
             await Promise.all([
-              prisma.measurement.findMany({
-                where: { userId: user.id },
-                orderBy: { measuredAt: "desc" },
-              }),
+              // Keyset-paged narrow read instead of one full-set
+              // findMany + map: a heavy multi-year tenant otherwise
+              // loads every measurement ORM row into a single array
+              // alongside the giant JSON.stringify below, the worker's
+              // dominant heap spike. The page reader projects each page
+              // to the compact backup shape before the next page loads,
+              // bounding peak heap to the page size. Row scope is
+              // unchanged — no `deletedAt` filter, so tombstoned rows
+              // still round-trip exactly as before. The order is
+              // restored to measuredAt-desc to match the prior output.
+              collectPagedMeasurements({
+                fetchPage: (afterId, take) =>
+                  prisma.measurement.findMany({
+                    where: {
+                      userId: user.id,
+                      ...(afterId ? { id: { gt: afterId } } : {}),
+                    },
+                    // Exactly the columns the admin restore recreates a
+                    // row from, plus `id` for the keyset cursor.
+                    select: {
+                      id: true,
+                      type: true,
+                      value: true,
+                      unit: true,
+                      source: true,
+                      measuredAt: true,
+                      notes: true,
+                      notesEncrypted: true,
+                    },
+                    orderBy: { id: "asc" },
+                    take,
+                  }),
+                project: toWeeklyBackupMeasurement,
+                pageSize: MEASUREMENT_BACKUP_PAGE_SIZE,
+              }).then(sortWeeklyMeasurementsDesc),
               prisma.medication.findMany({
                 where: { userId: user.id },
                 include: { schedules: true },
@@ -89,16 +125,10 @@ export async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
             schemaVersion: "1",
             exportedAt: new Date().toISOString(),
             userId: user.id,
-            measurements: measurements.map((m) => ({
-              type: m.type,
-              value: m.value,
-              unit: m.unit,
-              measuredAt: m.measuredAt.toISOString(),
-              source: m.source,
-              // v1.23 — decrypt into the (whole-blob-encrypted) backup payload
-              // so an admin restore re-encrypts on re-insert.
-              notes: readNote(m.notesEncrypted, m.notes),
-            })),
+            // Already projected + ordered by the paged reader above; the
+            // decrypted note rode into each row so an admin restore
+            // re-encrypts on re-insert (v1.23 contract).
+            measurements,
             medications: medications.map((m) => ({
               name: m.name,
               dose: m.dose,

src/lib/jobs/reminder/register-maintenance.ts

@@ -69,7 +69,7 @@ import {
   type EncryptionKeyRotatePayload,
 } from "@/lib/jobs/encryption-key-rotate";
 import { recordError } from "@/lib/jobs/worker-status";
-import { workerLog } from "./shared";
+import { workerLog, BOOT_BACKFILL_STAGGER_SECONDS } from "./shared";
 import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
 import {
   DataBackupPayload,
@@ -566,8 +566,15 @@ export async function enqueueMaintenanceBootDiscovery(): Promise<void> {
   // migration job per account. Idempotent across reboots: a migrated row nulls
   // its plaintext column and drops off the discovery predicate.
   try {
+    // Stage 5 — last of the staggered boot backfills. The heavy full-history
+    // loads each `localConcurrency: 1` used to drain in parallel onto one heavy
+    // tenant at boot; threading an increasing `startAfter` delay spreads them
+    // across a window. The daily cron tick (above) passes no offset and stays
+    // immediate.
     const { enqueued, skipped, error } =
-      await enqueueBootTimeNoteEncryptionBackfill();
+      await enqueueBootTimeNoteEncryptionBackfill(
+        BOOT_BACKFILL_STAGGER_SECONDS * 5,
+      );
     if (error) {
       workerLog(
         "error",

src/lib/jobs/reminder/register-rollup.ts

@@ -76,7 +76,11 @@ import {
   type DenseIntradayRetentionPayload,
 } from "@/lib/jobs/dense-intraday-retention";
 import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention";
-import { getWorkerPrisma, workerLog } from "./shared";
+import {
+  getWorkerPrisma,
+  workerLog,
+  BOOT_BACKFILL_STAGGER_SECONDS,
+} from "./shared";
 import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
 // v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows.
 // Collapses each user × cumulative-type × calendar-day bucket into one
@@ -445,6 +449,15 @@ export async function registerRollupQueues(
  * Fire-and-forget boot discovery for the self-converging rollup / consolidation
  * backfills. Each pass is idempotent across reboots and never fails worker boot
  * on a miss (errors come back through the helper's result value).
+ *
+ * Each backfill is `localConcurrency: 1` on its own, but every one of them used
+ * to drain from the first pg-boss poll at boot — on a heavy tenant that meant
+ * several full-history loads contending for the connection pool at once (a boot
+ * storm / crash-loop risk). Each type now gets an increasing
+ * `BOOT_BACKFILL_STAGGER_SECONDS` multiple as a `startAfter` delay so their
+ * loads spread across a window instead of all landing on the same poll. Dense
+ * intra-day retention keeps its own larger boot defer (the established P2028
+ * pool-exhaustion fix) and stays the furthest-out stage.
  */
 export async function enqueueRollupBootDiscovery(): Promise<void> {
   // v1.4.35.1 — measurement-rollup backfill. Finds every user with
@@ -452,7 +465,10 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // The discovery query only matches accounts with zero rollup rows, so once
   // a fold completes the user drops off the list.
   try {
-    const { enqueued, skipped, error } = await enqueueBootTimeRollupBackfill();
+    // Stage 0 — first off the line; no stagger delay.
+    const { enqueued, skipped, error } = await enqueueBootTimeRollupBackfill(
+      BOOT_BACKFILL_STAGGER_SECONDS * 0,
+    );
     if (error) {
       workerLog(
         "error",
@@ -483,8 +499,10 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // account. Consolidated legacy rows are soft-deleted, so the
   // `deleted_at IS NULL` discovery predicate drops them.
   try {
-    const { enqueued, skipped, error } =
-      await enqueueBootTimeStepConsolidation();
+    // Stage 1.
+    const { enqueued, skipped, error } = await enqueueBootTimeStepConsolidation(
+      BOOT_BACKFILL_STAGGER_SECONDS * 1,
+    );
     if (error) {
       workerLog(
         "error",
@@ -508,8 +526,10 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // per-sample high-frequency mean-type rows and enqueues one job per account.
   // Consolidated rows are soft-deleted, so the discovery predicate drops them.
   try {
-    const { enqueued, skipped, error } =
-      await enqueueBootTimeMeanConsolidation();
+    // Stage 2.
+    const { enqueued, skipped, error } = await enqueueBootTimeMeanConsolidation(
+      BOOT_BACKFILL_STAGGER_SECONDS * 2,
+    );
     if (error) {
       workerLog(
         "error",
@@ -534,6 +554,9 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // and enqueues one job per account. Folded rows are soft-deleted, so the
   // discovery predicate drops them.
   try {
+    // Furthest-out stage — the helper self-defers past the boot window via its
+    // own larger constant (the P2028 pool-exhaustion fix), so no stagger
+    // argument is threaded here.
     const { enqueued, skipped, error } =
       await enqueueBootTimeDenseIntradayRetention();
     if (error) {
@@ -560,8 +583,11 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // rollup coverage. Idempotent across reboots and singleton-keyed inside
   // pg-boss so a fast restart while a backfill is queued doesn't double up.
   try {
+    // Stage 3.
     const { enqueued, skipped, error } =
-      await enqueueBootTimeMoodRollupBackfill();
+      await enqueueBootTimeMoodRollupBackfill(
+        BOOT_BACKFILL_STAGGER_SECONDS * 3,
+      );
     if (error) {
       workerLog(
         "error",
@@ -585,8 +611,11 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
   // pattern: one job per user with intake events but no rollup coverage.
   // Idempotent across reboots and singleton-keyed inside pg-boss.
   try {
+    // Stage 4.
     const { enqueued, skipped, error } =
-      await enqueueBootTimeMedicationComplianceBackfill();
+      await enqueueBootTimeMedicationComplianceBackfill(
+        BOOT_BACKFILL_STAGGER_SECONDS * 4,
+      );
     if (error) {
       workerLog(
         "error",

src/lib/jobs/reminder/shared.ts

@@ -19,6 +19,17 @@ export function parseTimeToMinutes(value: string): number {
 
 export const DATABASE_URL = process.env.DATABASE_URL!;
 
+// Boot-backfill stagger step (seconds). The self-converging full-history
+// backfills are each `localConcurrency: 1` in isolation, but all of them used
+// to drain from the very first pg-boss poll at worker boot — on a heavy tenant
+// that meant ~half a dozen full-history loads contending for the same
+// connection pool at once (a boot storm / crash-loop risk). The boot-discovery
+// wrappers hand each backfill type an increasing multiple of this step as a
+// `startAfter` delay so their loads start at staggered times instead of all on
+// the first poll. Non-boot callers (cron) pass no offset and keep their
+// immediate semantics.
+export const BOOT_BACKFILL_STAGGER_SECONDS = 30;
+
 // Reuse a single PrismaClient across all job handlers to avoid connection pool exhaustion
 
 let workerPrisma: PrismaClient | null = null;

src/lib/jobs/step-consolidation.ts

@@ -76,7 +76,13 @@ export async function runStepConsolidationForUser(
  * Best-effort: errors are returned through the result value so the
  * worker boot never fails because of a consolidation miss.
  */
-export async function enqueueBootTimeStepConsolidation(): Promise<{
+export async function enqueueBootTimeStepConsolidation(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this consolidation does not drain in
+  // parallel with the other boot backfills onto one heavy tenant. Default 0
+  // keeps immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -119,6 +125,7 @@ export async function enqueueBootTimeStepConsolidation(): Promise<{
         retryDelay: 60,
         retryBackoff: true,
         singletonKey: `step-consolidation|${id}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
       });
       if (jobId) {
         enqueued += 1;

src/lib/measurements/__tests__/consolidation-scan-paged.test.ts

@@ -0,0 +1,195 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { bucketRowsByDay, scanSourceRowsPaged } from "../consolidation-base";
+import type { PerSampleRow } from "../consolidation-tz";
+import type { Prisma, PrismaClient } from "@/generated/prisma/client";
+
+// `scanSourceRowsPaged` keyset-paginates the per-(user, type) source scan so a
+// multi-year tenant no longer materialises its whole history in one Prisma
+// result set. These tests pin the two properties that matter: the accumulated
+// array equals a single unbounded scan (same rows, same order, no gaps /
+// duplicates), and the cursor advances on `(measuredAt, id)` after every full
+// page.
+
+/** Build a deterministic per-sample row at a 1-minute-per-index offset. */
+function row(index: number): PerSampleRow {
+  return {
+    id: `m-${String(index).padStart(4, "0")}`,
+    type: "ACTIVITY_STEPS",
+    value: index,
+    measuredAt: new Date(Date.UTC(2026, 0, 1, 0, index, 0)),
+    externalId: `hk-uuid-${index}`,
+  };
+}
+
+/**
+ * A findMany mock that serves `allRows` in keyset order, honouring `take`
+ * (page size) and the `(measuredAt, id)` cursor clause the helper AND-combines
+ * onto the base where. This mirrors how Postgres would resolve the keyset.
+ */
+function buildPagingFindMany(allRows: PerSampleRow[]) {
+  const ordered = [...allRows].sort((a, b) => {
+    const byTime = a.measuredAt.getTime() - b.measuredAt.getTime();
+    return byTime !== 0 ? byTime : a.id < b.id ? -1 : a.id > b.id ? 1 : 0;
+  });
+
+  return vi.fn(
+    async (args: {
+      where: Prisma.MeasurementWhereInput;
+      take?: number;
+      orderBy?: unknown;
+    }) => {
+      const take = args.take ?? ordered.length;
+
+      // Extract the keyset cursor the helper threads through `where.AND[1].OR`.
+      let afterTime: number | null = null;
+      let afterId: string | null = null;
+      const and = (args.where as { AND?: unknown[] }).AND;
+      if (Array.isArray(and)) {
+        const or = (and[1] as { OR?: Array<Record<string, unknown>> }).OR;
+        const gt = or?.[0]?.measuredAt as { gt?: Date } | undefined;
+        const tie = or?.[1] as
+          | { measuredAt?: Date; id?: { gt?: string } }
+          | undefined;
+        afterTime = gt?.gt ? gt.gt.getTime() : null;
+        afterId = tie?.id?.gt ?? null;
+      }
+
+      const remaining =
+        afterTime === null
+          ? ordered
+          : ordered.filter((r) => {
+              const t = r.measuredAt.getTime();
+              return t > afterTime! || (t === afterTime! && r.id > afterId!);
+            });
+
+      return remaining.slice(0, take);
+    },
+  );
+}
+
+function mockClient(findMany: ReturnType<typeof vi.fn>): PrismaClient {
+  return {
+    measurement: { findMany },
+  } as unknown as PrismaClient;
+}
+
+const BASE_WHERE: Prisma.MeasurementWhereInput = {
+  userId: "user-1",
+  type: "ACTIVITY_STEPS",
+};
+const SELECT: Prisma.MeasurementSelect = {
+  id: true,
+  type: true,
+  value: true,
+  measuredAt: true,
+  externalId: true,
+};
+
+describe("scanSourceRowsPaged", () => {
+  it("accumulates every row in keyset order across multiple pages with no gaps or duplicates", async () => {
+    const all = Array.from({ length: 23 }, (_, i) => row(i));
+    const findMany = buildPagingFindMany(all);
+
+    const result = await scanSourceRowsPaged(
+      mockClient(findMany),
+      BASE_WHERE,
+      SELECT,
+      10,
+    );
+
+    // 23 rows at page size 10 → pages of 10, 10, 3 (the short page ends it).
+    expect(result.map((r) => r.id)).toEqual(all.map((r) => r.id));
+    expect(new Set(result.map((r) => r.id)).size).toBe(all.length);
+    expect(findMany).toHaveBeenCalledTimes(3);
+  });
+
+  it("issues exactly one extra probing page when the row count is an exact multiple of the page size", async () => {
+    const all = Array.from({ length: 20 }, (_, i) => row(i));
+    const findMany = buildPagingFindMany(all);
+
+    const result = await scanSourceRowsPaged(
+      mockClient(findMany),
+      BASE_WHERE,
+      SELECT,
+      10,
+    );
+
+    // 20 rows at page size 10 → full pages 10, 10, then an empty probe page.
+    expect(result).toHaveLength(20);
+    expect(findMany).toHaveBeenCalledTimes(3);
+  });
+
+  it("advances the cursor on (measuredAt, id) after each full page", async () => {
+    const all = Array.from({ length: 12 }, (_, i) => row(i));
+    const findMany = buildPagingFindMany(all);
+
+    await scanSourceRowsPaged(mockClient(findMany), BASE_WHERE, SELECT, 5);
+
+    // First page: no cursor (base where only).
+    const firstWhere = findMany.mock.calls[0]?.[0]?.where;
+    expect((firstWhere as { AND?: unknown }).AND).toBeUndefined();
+    expect(firstWhere).toEqual(BASE_WHERE);
+
+    // Second page: cursor resumes after the 5th row (id m-0004).
+    const secondWhere = findMany.mock.calls[1]?.[0]?.where as {
+      AND: unknown[];
+    };
+    expect(Array.isArray(secondWhere.AND)).toBe(true);
+    expect(secondWhere.AND[0]).toEqual(BASE_WHERE);
+    const cursorClause = secondWhere.AND[1] as {
+      OR: Array<{ measuredAt?: { gt?: Date }; id?: { gt?: string } }>;
+    };
+    expect(cursorClause.OR[0].measuredAt?.gt).toEqual(row(4).measuredAt);
+    expect(cursorClause.OR[1].id?.gt).toBe(row(4).id);
+
+    // Every page is requested in stable keyset order.
+    for (const call of findMany.mock.calls) {
+      expect(call[0]?.orderBy).toEqual([{ measuredAt: "asc" }, { id: "asc" }]);
+    }
+  });
+
+  it("returns the empty array when the first page is empty", async () => {
+    const findMany = buildPagingFindMany([]);
+    const result = await scanSourceRowsPaged(
+      mockClient(findMany),
+      BASE_WHERE,
+      SELECT,
+      10,
+    );
+    expect(result).toEqual([]);
+    expect(findMany).toHaveBeenCalledTimes(1);
+  });
+
+  it("produces output that buckets identically to a single unpaged scan", async () => {
+    // Rows spanning three calendar days; the paged accumulation must bucket
+    // byte-identically to the same rows fed in one array.
+    const all = [
+      { ...row(1), measuredAt: new Date("2026-05-16T08:00:00.000Z") },
+      { ...row(2), measuredAt: new Date("2026-05-16T14:00:00.000Z") },
+      { ...row(3), measuredAt: new Date("2026-05-17T06:00:00.000Z") },
+      { ...row(4), measuredAt: new Date("2026-05-17T20:00:00.000Z") },
+      { ...row(5), measuredAt: new Date("2026-05-18T09:00:00.000Z") },
+    ];
+    const findMany = buildPagingFindMany(all);
+
+    const paged = await scanSourceRowsPaged(
+      mockClient(findMany),
+      BASE_WHERE,
+      SELECT,
+      2,
+    );
+
+    const baselineByDay = bucketRowsByDay(all, "Europe/Berlin", "stats:");
+    const pagedByDay = bucketRowsByDay(paged, "Europe/Berlin", "stats:");
+
+    expect([...pagedByDay.keys()].sort()).toEqual(
+      [...baselineByDay.keys()].sort(),
+    );
+    for (const [dateKey, rows] of baselineByDay) {
+      expect(pagedByDay.get(dateKey)?.map((r) => r.id)).toEqual(
+        rows.map((r) => r.id),
+      );
+    }
+  });
+});

src/lib/measurements/consolidation-base.ts

@@ -267,6 +267,74 @@ const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
   externalId: true,
 };
 
+/**
+ * Page size for the keyset-paginated source-row scan. Heavy multi-year
+ * tenants previously materialised the entire per-(user, type) history in a
+ * single unbounded `findMany` result set — the dominant Prisma intermediate
+ * cost and the root of the consolidation boot-storm on large accounts.
+ * Scanning in bounded keyset pages caps the per-query result set while still
+ * accumulating the same full row set.
+ */
+const CONSOLIDATION_SCAN_PAGE_SIZE = 5000;
+
+/**
+ * Keyset-paginate the per-(user, type) source-row scan on `(measuredAt, id)`
+ * ascending, accumulating every page into one array. Output is identical to a
+ * single unbounded `findMany` — same `where`, same `select`, same ascending
+ * order — but the Prisma client never holds more than one page of rows at a
+ * time, so a multi-year tenant no longer materialises its whole history in one
+ * result set at worker boot.
+ *
+ * `(measuredAt, id)` is a stable keyset: `measuredAt` is the primary order and
+ * `id` breaks ties, so each page resumes exactly after the last row with no
+ * gaps or duplicates. The cursor clause is AND-combined with the caller's
+ * `baseWhere` so the scope / soft-delete / grace filters always still apply.
+ */
+export async function scanSourceRowsPaged(
+  prismaClient: PrismaClient,
+  baseWhere: Prisma.MeasurementWhereInput,
+  scanSelect: Prisma.MeasurementSelect,
+  pageSize: number = CONSOLIDATION_SCAN_PAGE_SIZE,
+): Promise<PerSampleRow[]> {
+  const accumulated: PerSampleRow[] = [];
+  let cursor: { measuredAt: Date; id: string } | null = null;
+
+  for (;;) {
+    const where: Prisma.MeasurementWhereInput = cursor
+      ? {
+          AND: [
+            baseWhere,
+            {
+              OR: [
+                { measuredAt: { gt: cursor.measuredAt } },
+                { measuredAt: cursor.measuredAt, id: { gt: cursor.id } },
+              ],
+            },
+          ],
+        }
+      : baseWhere;
+
+    const page = (await prismaClient.measurement.findMany({
+      where,
+      select: scanSelect,
+      orderBy: [{ measuredAt: "asc" }, { id: "asc" }],
+      take: pageSize,
+    })) as PerSampleRow[];
+
+    if (page.length === 0) break;
+
+    for (const row of page) accumulated.push(row);
+
+    // A short page is the last page — no further rows can satisfy the keyset.
+    if (page.length < pageSize) break;
+
+    const last = page[page.length - 1]!;
+    cursor = { measuredAt: last.measuredAt, id: last.id };
+  }
+
+  return accumulated;
+}
+
 /**
  * Drive one consolidation pass. Walks `users → types → days`, scans live
  * source rows inside the grace window, buckets them, reduces each day,
@@ -298,16 +366,20 @@ export async function runConsolidation<TType extends MeasurementType>(
       const hkIdentifier = params.hkIdentifierForType(type);
       if (!hkIdentifier) continue;
 
-      const sourceRows = (await prismaClient.measurement.findMany({
-        where: params.buildScanWhere({
+      // Keyset-paginate the scan instead of loading the whole per-(user, type)
+      // history in one result set; the accumulated array is byte-identical to a
+      // single unbounded `findMany`, but a multi-year tenant no longer spikes
+      // worker memory at boot. See `scanSourceRowsPaged`.
+      const sourceRows = await scanSourceRowsPaged(
+        prismaClient,
+        params.buildScanWhere({
           userId: user.id,
           type,
           cutoffAt,
           statsPrefix: params.statsPrefix,
         }),
-        select: scanSelect,
-        orderBy: { measuredAt: "asc" },
-      })) as PerSampleRow[];
+        scanSelect,
+      );
 
       if (sourceRows.length === 0) continue;
 

src/lib/rollups/measurement-rollups.ts

@@ -841,7 +841,13 @@ async function ensureUserRollupsFreshImpl(
  * Best-effort: errors are swallowed and reported through the return
  * value so the worker boot never fails because of a backfill miss.
  */
-export async function enqueueBootTimeRollupBackfill(): Promise<{
+export async function enqueueBootTimeRollupBackfill(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this full-history fold does not drain in
+  // parallel with the other boot backfills onto one heavy tenant. Default 0
+  // keeps immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -947,6 +953,7 @@ export async function enqueueBootTimeRollupBackfill(): Promise<{
         // Coalesce: if a backfill for this user is already queued and
         // we restart, pg-boss returns null instead of duplicating.
         singletonKey: `boot-backfill|${id}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
       });
       if (jobId) {
         enqueued += 1;

src/lib/rollups/medication-compliance-rollups.ts

@@ -541,7 +541,13 @@ export async function enqueueUserMedicationComplianceBackfill(
  * row inside the trailing window the discovery query drops them.
  * pg-boss `singletonKey` coalesces duplicate sends.
  */
-export async function enqueueBootTimeMedicationComplianceBackfill(): Promise<{
+export async function enqueueBootTimeMedicationComplianceBackfill(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this full-history fold does not drain in
+  // parallel with the other boot backfills onto one heavy tenant. Default 0
+  // keeps immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -587,6 +593,7 @@ export async function enqueueBootTimeMedicationComplianceBackfill(): Promise<{
           retryDelay: 60,
           retryBackoff: true,
           singletonKey: `medication-compliance-boot-backfill|${id}`,
+          ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
         },
       );
       if (jobId) {

src/lib/rollups/mood-rollups.ts

@@ -451,7 +451,13 @@ async function ensureUserMoodRollupsFreshImpl(
  * Best-effort: errors are swallowed and reported through the return
  * value so the worker boot never fails because of a backfill miss.
  */
-export async function enqueueBootTimeMoodRollupBackfill(): Promise<{
+export async function enqueueBootTimeMoodRollupBackfill(
+  // Optional boot-storm stagger. When > 0 the per-user sends carry a
+  // `startAfter` delay (seconds) so this full-history fold does not drain in
+  // parallel with the other boot backfills onto one heavy tenant. Default 0
+  // keeps immediate semantics for any non-boot caller.
+  startAfterSeconds: number = 0,
+): Promise<{
   enqueued: number;
   skipped: number;
   error: string | null;
@@ -500,6 +506,7 @@ export async function enqueueBootTimeMoodRollupBackfill(): Promise<{
         // Coalesce: if a backfill for this user is already queued and
         // we restart, pg-boss returns null instead of duplicating.
         singletonKey: `mood-boot-backfill|${id}`,
+        ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
       });
       if (jobId) {
         enqueued += 1;