diff --git a/src/lib/measurements/__tests__/adopt-or-mint-hourly-rows.test.ts b/src/lib/measurements/__tests__/adopt-or-mint-hourly-rows.test.ts
new file mode 100644
index 000000000..a0396d9f1
--- /dev/null
+++ b/src/lib/measurements/__tests__/adopt-or-mint-hourly-rows.test.ts
@@ -0,0 +1,331 @@
+/**
+ * `adoptOrMintHourlyRows` resolves a whole folded day in one lookup and one
+ * insert. It replaced a per-slot loop (two lookups and a write per hour, about
+ * 75 statements a day), which was most of the 380 000 statements the dense
+ * retention spent on a 1.25-million-row account (issue #1031).
+ *
+ * Two things are pinned here. The statement shape: one read, at most one
+ * insert, no per-slot lookups. And the semantics: against an in-memory table
+ * that enforces both unique identities, the batch leaves exactly the rows the
+ * per-slot loop leaves and returns the same ids, over randomised days that
+ * include rows already sitting on an anchor, rows already carrying a target
+ * externalId, tombstones, and a row holding one hour's externalId while
+ * parked on another hour's anchor.
+ */
+import { describe, expect, it, vi } from "vitest";
+
+import type { MeasurementType, Prisma } from "@/generated/prisma/client";
+
+import {
+  adoptOrMintHourlyRows,
+  type HourlySlot,
+} from "../dense-intraday-retention";
+
+interface Row {
+  id: string;
+  externalId: string | null;
+  measuredAt: Date;
+  value: number;
+  unit: string;
+  deletedAt: Date | null;
+}
+
+/**
+ * A measurement table for one (user, type, APPLE_HEALTH), with the two unique
+ * identities the real one has: externalId and measuredAt (sleepStage is NULL
+ * for these types). A write that would duplicate either throws, as P2002 does.
+ */
+function fakeTable(seed: Row[]) {
+  const rows = seed.map((row) => ({ ...row }));
+  let minted = 0;
+  const calls = { findMany: 0, findFirst: 0, createMany: 0, create: 0 };
+
+  const assertUnique = () => {
+    const ext = rows
+      .filter((r) => r.externalId !== null)
+      .map((r) => r.externalId);
+    const at = rows.map((r) => r.measuredAt.getTime());
+    if (new Set(ext).size !== ext.length) throw new Error("P2002 externalId");
+    if (new Set(at).size !== at.length) throw new Error("P2002 measuredAt");
+  };
+  const mint = (data: {
+    externalId: string;
+    measuredAt: Date;
+    value: number;
+    unit: string;
+  }) => {
+    minted += 1;
+    const row: Row = {
+      id: `z-minted-${String(minted).padStart(3, "0")}`,
+      externalId: data.externalId,
+      measuredAt: data.measuredAt,
+      value: data.value,
+      unit: data.unit,
+      deletedAt: null,
+    };
+    rows.push(row);
+    assertUnique();
+    return row;
+  };
+  const byId = () => [...rows].sort((a, b) => (a.id < b.id ? -1 : 1));
+
+  const tx = {
+    measurement: {
+      findMany: vi.fn(
+        async (args: {
+          where: {
+            OR: [
+              { externalId: { in: string[] } },
+              { measuredAt: { in: Date[] } },
+            ];
+          };
+        }) => {
+          calls.findMany += 1;
+          const ext = new Set(args.where.OR[0].externalId.in);
+          const at = new Set(
+            args.where.OR[1].measuredAt.in.map((d) => d.getTime()),
+          );
+          return byId()
+            .filter(
+              (r) =>
+                (r.externalId !== null && ext.has(r.externalId)) ||
+                at.has(r.measuredAt.getTime()),
+            )
+            .map((r) => ({
+              id: r.id,
+              externalId: r.externalId,
+              measuredAt: r.measuredAt,
+            }));
+        },
+      ),
+      findFirst: vi.fn(
+        async (args: { where: { externalId?: string; measuredAt?: Date } }) => {
+          calls.findFirst += 1;
+          const hit = byId().find((r) =>
+            args.where.externalId !== undefined
+              ? r.externalId === args.where.externalId
+              : r.measuredAt.getTime() === args.where.measuredAt!.getTime(),
+          );
+          return hit ? { id: hit.id } : null;
+        },
+      ),
+      update: vi.fn(
+        async (args: {
+          where: { id: string };
+          data: Partial<Row> & { measuredAt?: Date };
+        }) => {
+          const row = rows.find((r) => r.id === args.where.id)!;
+          Object.assign(row, args.data);
+          assertUnique();
+          return row;
+        },
+      ),
+      create: vi.fn(
+        async (args: {
+          data: {
+            externalId: string;
+            measuredAt: Date;
+            value: number;
+            unit: string;
+          };
+        }) => {
+          calls.create += 1;
+          return { id: mint(args.data).id };
+        },
+      ),
+      createManyAndReturn: vi.fn(
+        async (args: {
+          data: Array<{
+            externalId: string;
+            measuredAt: Date;
+            value: number;
+            unit: string;
+          }>;
+        }) => {
+          calls.createMany += 1;
+          return args.data.map((d) => {
+            const row = mint(d);
+            return { id: row.id, externalId: row.externalId };
+          });
+        },
+      ),
+    },
+  };
+  return {
+    tx: tx as unknown as Prisma.TransactionClient,
+    rows,
+    calls,
+  };
+}
+
+/**
+ * The per-slot loop the batch replaced, kept verbatim in behaviour as the
+ * reference: externalId first, then the anchor occupant, lowest id on a tie;
+ * re-anchor only onto a free slot; otherwise mint.
+ */
+async function perSlotReference(
+  tx: Prisma.TransactionClient,
+  unit: string,
+  slots: readonly HourlySlot[],
+): Promise<string[]> {
+  const ids: string[] = [];
+  for (const slot of slots) {
+    const eidRow = await tx.measurement.findFirst({
+      where: { externalId: slot.externalId },
+    } as Prisma.MeasurementFindFirstArgs);
+    const slotRow = await tx.measurement.findFirst({
+      where: { measuredAt: slot.anchor },
+    } as Prisma.MeasurementFindFirstArgs);
+    const target = eidRow ?? slotRow;
+    if (target) {
+      const free = slotRow === null || slotRow.id === target.id;
+      await tx.measurement.update({
+        where: { id: target.id },
+        data: {
+          value: slot.value,
+          unit,
+          externalId: slot.externalId,
+          deletedAt: null,
+          ...(free ? { measuredAt: slot.anchor } : {}),
+        },
+      });
+      ids.push(target.id);
+    } else {
+      const created = await tx.measurement.create({
+        data: {
+          value: slot.value,
+          unit,
+          measuredAt: slot.anchor,
+          externalId: slot.externalId,
+        },
+      } as Prisma.MeasurementCreateArgs);
+      ids.push(created.id);
+    }
+  }
+  return ids;
+}
+
+const HOUR = 3_600_000;
+const DAY0 = Date.UTC(2026, 4, 1, 0, 30);
+const anchor = (h: number) => new Date(DAY0 + h * HOUR);
+const ext = (h: number) => `stats:HK:2026-05-01T${String(h).padStart(2, "0")}`;
+
+/** A deterministic pseudo-random generator, so a failure reproduces. */
+function rng(seed: number) {
+  let x = seed;
+  return () => {
+    x = (x * 1103515245 + 12345) % 2 ** 31;
+    return x / 2 ** 31;
+  };
+}
+
+function scenario(seed: number): { seed: Row[]; slots: HourlySlot[] } {
+  const rand = rng(seed);
+  const hours = [...Array(24).keys()].filter(() => rand() < 0.7);
+  const slots = hours.map((h) => ({
+    externalId: ext(h),
+    anchor: anchor(h),
+    value: Math.round(rand() * 100),
+  }));
+  const rows: Row[] = [];
+  const takenAt = new Set<number>();
+  const takenExt = new Set<string>();
+  let n = 0;
+  const add = (externalId: string | null, at: Date) => {
+    if (takenAt.has(at.getTime())) return;
+    if (externalId !== null && takenExt.has(externalId)) return;
+    takenAt.add(at.getTime());
+    if (externalId !== null) takenExt.add(externalId);
+    n += 1;
+    rows.push({
+      id: `r-${String(Math.floor(rand() * 900) + 100)}-${n}`,
+      externalId,
+      measuredAt: at,
+      value: 1,
+      unit: "bpm",
+      deletedAt: rand() < 0.3 ? new Date(DAY0) : null,
+    });
+  };
+  for (const h of hours) {
+    const roll = rand();
+    if (roll < 0.15)
+      add(ext(h), anchor(h)); // a prior fold's own row
+    else if (roll < 0.3)
+      add(`hk-sample-${h}`, anchor(h)); // a sample on the anchor
+    else if (roll < 0.4)
+      add(ext(h), new Date(DAY0 + h * HOUR + 60_000)); // stats row off-anchor
+    else if (roll < 0.5) {
+      // This hour's externalId parked on ANOTHER hour's anchor.
+      const other = hours[Math.floor(rand() * hours.length)]!;
+      add(ext(h), anchor(other));
+    }
+  }
+  return { seed: rows, slots };
+}
+
+function snapshot(rows: Row[]) {
+  return [...rows]
+    .map((r) => ({
+      id: r.id.startsWith("z-minted") ? "minted" : r.id,
+      externalId: r.externalId,
+      at: r.measuredAt.toISOString(),
+      value: r.value,
+      deleted: r.deletedAt !== null,
+    }))
+    .sort((a, b) =>
+      `${a.externalId}|${a.at}` < `${b.externalId}|${b.at}` ? -1 : 1,
+    );
+}
+
+describe("adoptOrMintHourlyRows", () => {
+  it("resolves a fresh day with one lookup and one insert, no per-slot reads", async () => {
+    const { tx, calls, rows } = fakeTable([]);
+    const slots = [...Array(24).keys()].map((h) => ({
+      externalId: ext(h),
+      anchor: anchor(h),
+      value: h,
+    }));
+
+    const ids = await adoptOrMintHourlyRows(tx, {
+      userId: "u",
+      type: "PULSE" as MeasurementType,
+      unit: "bpm",
+      slots,
+    });
+
+    expect(calls).toEqual({
+      findMany: 1,
+      findFirst: 0,
+      createMany: 1,
+      create: 0,
+    });
+    expect(rows).toHaveLength(24);
+    expect(ids).toEqual(
+      slots.map((s) => rows.find((r) => r.externalId === s.externalId)!.id),
+    );
+  });
+
+  it("leaves the same rows and returns the same ids as the per-slot loop", async () => {
+    for (let seed = 1; seed <= 300; seed++) {
+      const { seed: rows, slots } = scenario(seed);
+
+      const batch = fakeTable(rows);
+      const batchIds = await adoptOrMintHourlyRows(batch.tx, {
+        userId: "u",
+        type: "PULSE" as MeasurementType,
+        unit: "bpm",
+        slots,
+      });
+
+      const reference = fakeTable(rows);
+      const referenceIds = await perSlotReference(reference.tx, "bpm", slots);
+
+      const norm = (ids: string[]) =>
+        ids.map((id) => (id.startsWith("z-minted") ? "minted" : id));
+      expect(norm(batchIds), `seed ${seed}`).toEqual(norm(referenceIds));
+      expect(snapshot(batch.rows), `seed ${seed}`).toEqual(
+        snapshot(reference.rows),
+      );
+    }
+  });
+});
diff --git a/src/lib/measurements/__tests__/consolidation-scan-paged.test.ts b/src/lib/measurements/__tests__/consolidation-scan-paged.test.ts
index ea8bf0563..1add7ea76 100644
--- a/src/lib/measurements/__tests__/consolidation-scan-paged.test.ts
+++ b/src/lib/measurements/__tests__/consolidation-scan-paged.test.ts
@@ -1,31 +1,50 @@
 import { describe, expect, it, vi } from "vitest";
 
-import { bucketRowsByDay, scanSourceRowsPaged } from "../consolidation-base";
+import {
+  bucketRowsByDay,
+  iterateDayBuckets,
+  iterateSourcePages,
+  runConsolidation,
+  type ConsolidationParams,
+} from "../consolidation-base";
 import type { PerSampleRow } from "../consolidation-tz";
-import type { Prisma, PrismaClient } from "@/generated/prisma/client";
+import type {
+  MeasurementType,
+  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.
+// The consolidation drains walk a user's whole history of a type. Issue #1031:
+// on an account with 1.25 million rows the walk gathered every page into one
+// array before folding a single day (≈230 MB of heap for the heart-rate type
+// alone) and paged with a cursor Postgres could not use as an index condition,
+// so every page re-read the rows before it. These tests pin the three
+// properties the fix rests on: pages are handed on one at a time and a day is
+// folded before the scan reads past it, the cursor carries an index-usable
+// bound, and a pass stops cleanly when asked to.
 
-/** Build a deterministic per-sample row at a 1-minute-per-index offset. */
-function row(index: number): PerSampleRow {
+/** A deterministic per-sample row at `index` hours past 2026-01-01T00:00Z. */
+function row(index: number, minuteOffset = 0): 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)),
+    measuredAt: new Date(Date.UTC(2026, 0, 1, index, minuteOffset, 0)),
     externalId: `hk-uuid-${index}`,
   };
 }
 
+interface CursorWhere {
+  AND?: [
+    Prisma.MeasurementWhereInput,
+    { measuredAt?: { gte?: Date } },
+    { OR?: Array<{ measuredAt?: Date | { gt?: Date }; id?: { gt?: string } }> },
+  ];
+}
+
 /**
- * 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.
+ * A findMany mock that serves `allRows` in keyset order, honouring `take` and
+ * the `(measuredAt, id)` cursor the pager AND-combines onto the base where.
  */
 function buildPagingFindMany(allRows: PerSampleRow[]) {
   const ordered = [...allRows].sort((a, b) => {
@@ -34,43 +53,29 @@ function buildPagingFindMany(allRows: PerSampleRow[]) {
   });
 
   return vi.fn(
-    async (args: {
-      where: Prisma.MeasurementWhereInput;
-      take?: number;
-      orderBy?: unknown;
-    }) => {
+    async (args: { where: CursorWhere; 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);
+      const and = args.where.AND;
+      if (!and) return ordered.slice(0, take);
+      const gte = and[1].measuredAt?.gte?.getTime();
+      const or = and[2].OR ?? [];
+      const gt = (
+        or[0]?.measuredAt as { gt?: Date } | undefined
+      )?.gt?.getTime();
+      const tieId = or[1]?.id?.gt;
+      return ordered
+        .filter((r) => {
+          const t = r.measuredAt.getTime();
+          if (gte !== undefined && t < gte) return false;
+          return t > gt! || (t === gt! && r.id > tieId!);
+        })
+        .slice(0, take);
     },
   );
 }
 
 function mockClient(findMany: ReturnType<typeof vi.fn>): PrismaClient {
-  return {
-    measurement: { findMany },
-  } as unknown as PrismaClient;
+  return { measurement: { findMany } } as unknown as PrismaClient;
 }
 
 const BASE_WHERE: Prisma.MeasurementWhereInput = {
@@ -85,21 +90,25 @@ const SELECT: Prisma.MeasurementSelect = {
   externalId: true,
 };
 
-describe("scanSourceRowsPaged", () => {
-  it("accumulates every row in keyset order across multiple pages with no gaps or duplicates", async () => {
+async function collectPages(
+  pages: AsyncIterable<PerSampleRow[]>,
+): Promise<PerSampleRow[][]> {
+  const out: PerSampleRow[][] = [];
+  for await (const page of pages) out.push(page);
+  return out;
+}
+
+describe("iterateSourcePages", () => {
+  it("yields every row once, in keyset order, one bounded page at a time", async () => {
     const all = Array.from({ length: 23 }, (_, i) => row(i));
     const findMany = buildPagingFindMany(all);
 
-    const result = await scanSourceRowsPaged(
-      mockClient(findMany),
-      BASE_WHERE,
-      SELECT,
-      10,
+    const pages = await collectPages(
+      iterateSourcePages(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(pages.map((p) => p.length)).toEqual([10, 10, 3]);
+    expect(pages.flat().map((r) => r.id)).toEqual(all.map((r) => r.id));
     expect(findMany).toHaveBeenCalledTimes(3);
   });
 
@@ -107,88 +116,191 @@ describe("scanSourceRowsPaged", () => {
     const all = Array.from({ length: 20 }, (_, i) => row(i));
     const findMany = buildPagingFindMany(all);
 
-    const result = await scanSourceRowsPaged(
-      mockClient(findMany),
-      BASE_WHERE,
-      SELECT,
-      10,
+    const pages = await collectPages(
+      iterateSourcePages(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(pages.flat()).toHaveLength(20);
     expect(findMany).toHaveBeenCalledTimes(3);
   });
 
-  it("advances the cursor on (measuredAt, id) after each full page", async () => {
+  it("yields nothing when the first page is empty", async () => {
+    const findMany = buildPagingFindMany([]);
+    const pages = await collectPages(
+      iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 10),
+    );
+    expect(pages).toEqual([]);
+    expect(findMany).toHaveBeenCalledTimes(1);
+  });
+
+  it("bounds every later page with an index-usable measuredAt >= cursor conjunct", async () => {
+    // `measuredAt > c OR (measuredAt = c AND id > i)` alone is not an index
+    // condition: Postgres walked the index from the user's first row on
+    // every page. The redundant `>=` bound is what the planner can use.
     const all = Array.from({ length: 12 }, (_, i) => row(i));
     const findMany = buildPagingFindMany(all);
 
-    await scanSourceRowsPaged(mockClient(findMany), BASE_WHERE, SELECT, 5);
+    await collectPages(
+      iterateSourcePages(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.
+    expect(findMany.mock.calls[0]?.[0]?.where).toEqual(BASE_WHERE);
+    const second = findMany.mock.calls[1]?.[0]?.where as CursorWhere;
+    expect(second.AND?.[0]).toEqual(BASE_WHERE);
+    expect(second.AND?.[1]).toEqual({
+      measuredAt: { gte: row(4).measuredAt },
+    });
+    expect(second.AND?.[2]).toEqual({
+      OR: [
+        { measuredAt: { gt: row(4).measuredAt } },
+        { measuredAt: row(4).measuredAt, id: { gt: row(4).id } },
+      ],
+    });
     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") },
-    ];
+  it("resumes past rows that share the boundary instant without gaps or duplicates", async () => {
+    // Five rows on the same instant straddle the page boundary.
+    const all = [0, 1, 2, 3, 4].map((i) => ({
+      ...row(i),
+      measuredAt: new Date("2026-05-16T08:00:00.000Z"),
+    }));
     const findMany = buildPagingFindMany(all);
-
-    const paged = await scanSourceRowsPaged(
-      mockClient(findMany),
-      BASE_WHERE,
-      SELECT,
-      2,
+    const pages = await collectPages(
+      iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
     );
+    expect(pages.flat().map((r) => r.id)).toEqual(all.map((r) => r.id));
+  });
+});
 
-    const baselineByDay = bucketRowsByDay(all, "Europe/Berlin", "stats:");
-    const pagedByDay = bucketRowsByDay(paged, "Europe/Berlin", "stats:");
+describe("iterateDayBuckets", () => {
+  const days = [
+    { ...row(1), measuredAt: new Date("2026-05-16T08:00:00.000Z") },
+    { ...row(2), measuredAt: new Date("2026-05-16T14:00:00.000Z") },
+    {
+      ...row(3),
+      externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16",
+      measuredAt: new Date("2026-05-16T15:00:00.000Z"),
+    },
+    { ...row(4), measuredAt: new Date("2026-05-17T06:00:00.000Z") },
+    { ...row(5), measuredAt: new Date("2026-05-17T20:00:00.000Z") },
+    { ...row(6), measuredAt: new Date("2026-05-18T09:00:00.000Z") },
+  ];
 
-    expect([...pagedByDay.keys()].sort()).toEqual(
-      [...baselineByDay.keys()].sort(),
-    );
-    for (const [dateKey, rows] of baselineByDay) {
-      expect(pagedByDay.get(dateKey)?.map((r) => r.id)).toEqual(
+  it("groups exactly as the whole-array bucketing does, stats rows skipped", async () => {
+    const findMany = buildPagingFindMany(days);
+    let scanned = 0;
+    const streamed = new Map<string, string[]>();
+    for await (const [key, rows] of iterateDayBuckets(
+      iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
+      "Europe/Berlin",
+      "stats:",
+      () => {
+        scanned += 1;
+      },
+    )) {
+      streamed.set(
+        key,
         rows.map((r) => r.id),
       );
     }
+
+    const baseline = bucketRowsByDay(days, "Europe/Berlin", "stats:");
+    expect([...streamed.keys()]).toEqual([...baseline.keys()]);
+    for (const [key, rows] of baseline) {
+      expect(streamed.get(key)).toEqual(rows.map((r) => r.id));
+    }
+    expect(scanned).toBe(days.length);
+  });
+
+  it("hands a day on before the scan reads the page after the one that closed it", async () => {
+    const findMany = buildPagingFindMany(days);
+    const pagesReadWhenYielded: number[] = [];
+    for await (const _day of iterateDayBuckets(
+      iterateSourcePages(mockClient(findMany), BASE_WHERE, SELECT, 2),
+      "Europe/Berlin",
+      "stats:",
+    )) {
+      pagesReadWhenYielded.push(findMany.mock.calls.length);
+    }
+    // Pages: [r1 r2] [r3 r4] [r5 r6] [] — 05-16 closes on page 2 (r4),
+    // 05-17 on page 3 (r6), 05-18 at the end of the scan.
+    expect(pagesReadWhenYielded).toEqual([2, 3, 4]);
+  });
+});
+
+describe("runConsolidation — streamed walk", () => {
+  /** One row per hour for `dayCount` UTC days. */
+  function hourlyRows(dayCount: number): PerSampleRow[] {
+    return Array.from({ length: dayCount * 24 }, (_, i) => row(i, 30));
+  }
+
+  function params(
+    findMany: ReturnType<typeof vi.fn>,
+    writeDay: ConsolidationParams<MeasurementType>["writeDay"],
+    shouldStop?: () => boolean,
+  ): ConsolidationParams<MeasurementType> {
+    const client = {
+      user: {
+        findMany: vi
+          .fn()
+          .mockResolvedValue([{ id: "user-1", timezone: "UTC" }]),
+      },
+      measurement: { findMany },
+    } as unknown as PrismaClient;
+    return {
+      prismaClient: client,
+      options: { shouldStop },
+      types: ["ACTIVITY_STEPS"] as MeasurementType[],
+      hkIdentifierForType: () => "HKQuantityTypeIdentifierStepCount",
+      dailyStatsExternalId: (hk: string, day: string) => `stats:${hk}:${day}`,
+      statsPrefix: "stats:",
+      reduce: (rows: readonly PerSampleRow[]) => rows.length,
+      buildScanWhere: () => BASE_WHERE,
+      writeDay,
+      recordBucket: () => {},
+    };
+  }
+
+  it("folds the first day before the scan has read the whole history", async () => {
+    // 500 days × 24 rows = 12 000 rows → default pages of 5000, 5000, 2000.
+    const all = hourlyRows(500);
+    const findMany = buildPagingFindMany(all);
+    const pagesReadAtFirstWrite: number[] = [];
+    const writeDay = vi.fn(async () => {
+      if (pagesReadAtFirstWrite.length === 0) {
+        pagesReadAtFirstWrite.push(findMany.mock.calls.length);
+      }
+      return { kind: "written" as const, sourceRowsRemoved: 0 };
+    });
+
+    const result = await runConsolidation(params(findMany, writeDay));
+
+    expect(writeDay).toHaveBeenCalledTimes(500);
+    expect(findMany).toHaveBeenCalledTimes(3);
+    // The old walk read all three pages before the first write.
+    expect(pagesReadAtFirstWrite).toEqual([1]);
+    expect(result.stoppedEarly).toBe(false);
+  });
+
+  it("stops before the next day once shouldStop answers true, and reads no further", async () => {
+    const all = hourlyRows(500);
+    const findMany = buildPagingFindMany(all);
+    const writeDay = vi.fn(async () => ({
+      kind: "written" as const,
+      sourceRowsRemoved: 0,
+    }));
+    let budget = 3;
+    const shouldStop = () => budget-- <= 0;
+
+    const result = await runConsolidation(
+      params(findMany, writeDay, shouldStop),
+    );
+
+    expect(writeDay).toHaveBeenCalledTimes(3);
+    expect(result.stoppedEarly).toBe(true);
+    expect(findMany).toHaveBeenCalledTimes(1);
   });
 });
diff --git a/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts b/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts
index de394991c..c69715be9 100644
--- a/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts
+++ b/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts
@@ -30,6 +30,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
 
 import { runDenseIntradayHourlyRebuild } from "../dense-intraday-hourly-rebuild";
 import type { PrismaClient } from "@/generated/prisma/client";
+import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
 const HRV_DAILY_ID =
   "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01";
@@ -79,6 +80,8 @@ function buildPrismaMock(opts: {
       create: txCreate,
       update: txUpdate,
       findFirst: txFindFirst,
+      findMany: candidateLookup(),
+      createManyAndReturn: createManyVia(txCreate),
     },
   };
 
@@ -444,6 +447,10 @@ describe("runDenseIntradayHourlyRebuild — rebuild flow", () => {
             create: vi.fn().mockResolvedValue({ id: "minted" }),
             update: vi.fn().mockResolvedValue({}),
             findFirst: vi.fn().mockResolvedValue(null),
+            findMany: candidateLookup(),
+            createManyAndReturn: createManyVia(
+              vi.fn().mockResolvedValue({ id: "minted" }),
+            ),
           },
         });
       },
diff --git a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts
index 6d7973204..76f03037d 100644
--- a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts
+++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts
@@ -33,6 +33,7 @@ import {
 } from "../dense-intraday-retention";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
 function pulseRow(id: string, value: number, iso: string): PerSampleRow {
   return {
@@ -68,6 +69,8 @@ function buildPrismaMock(opts: {
       update: txUpdate,
       findFirst: txFindFirst,
       updateMany: txUpdateMany,
+      findMany: candidateLookup(),
+      createManyAndReturn: createManyVia(txCreate),
     },
   };
 
@@ -325,6 +328,10 @@ describe("PULSE fold — per-day failure boundary", () => {
             update: vi.fn(),
             findFirst: vi.fn().mockResolvedValue(null),
             updateMany: vi.fn().mockResolvedValue({ count: 5 }),
+            findMany: candidateLookup(),
+            createManyAndReturn: createManyVia(
+              vi.fn().mockResolvedValue({ id: "minted" }),
+            ),
           },
         });
       },
diff --git a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts
index 92d88f07b..fb0579290 100644
--- a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts
+++ b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts
@@ -32,6 +32,7 @@ import {
   DENSE_INTRADAY_RETENTION_DAYS,
 } from "../dense-intraday-retention";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
 /** A day well outside the retention window, so the fold definitely runs. */
 const FOLD_DAY = new Date(
@@ -76,7 +77,15 @@ function buildPrismaMock(existingResting: Array<{ source: string }>) {
   );
 
   const tx = {
-    measurement: { create, update, findFirst: txFindFirst, updateMany, upsert },
+    measurement: {
+      create,
+      update,
+      findFirst: txFindFirst,
+      updateMany,
+      upsert,
+      findMany: candidateLookup(),
+      createManyAndReturn: createManyVia(create),
+    },
   };
 
   return {
diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts
index 9f10409db..4ea0b1250 100644
--- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts
+++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts
@@ -25,6 +25,7 @@ import {
 } from "../dense-intraday-retention";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
 function row(
   id: string,
@@ -56,7 +57,16 @@ function buildPrismaMock(
     async (args: { where: { type: string } }) =>
       rowsByType[args.where.type] ?? [],
   );
-  const tx = { measurement: { create, update, findFirst, updateMany } };
+  const tx = {
+    measurement: {
+      create,
+      update,
+      findFirst,
+      updateMany,
+      findMany: candidateLookup(existingCanonicalId),
+      createManyAndReturn: createManyVia(create),
+    },
+  };
   return {
     mock: {
       user: {
diff --git a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts
index cf19ba322..a61d7a1d8 100644
--- a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts
+++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts
@@ -29,6 +29,7 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
 import { runDenseIntradayRetention } from "../dense-intraday-retention";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
 function spo2Row(id: string, value: number, iso: string): PerSampleRow {
   return {
@@ -60,6 +61,8 @@ function buildPrismaMock(opts: { spo2Rows: PerSampleRow[] }) {
       update: txUpdate,
       findFirst: txFindFirst,
       updateMany: txUpdateMany,
+      findMany: candidateLookup(),
+      createManyAndReturn: createManyVia(txCreate),
     },
   };
 
@@ -254,6 +257,10 @@ describe("SpO2 fold — per-day failure boundary", () => {
             update: vi.fn(),
             findFirst: vi.fn().mockResolvedValue(null),
             updateMany: vi.fn().mockResolvedValue({ count: 2 }),
+            findMany: candidateLookup(),
+            createManyAndReturn: createManyVia(
+              vi.fn().mockResolvedValue({ id: "minted" }),
+            ),
           },
         });
       },
diff --git a/src/lib/measurements/__tests__/hourly-mint-mock.ts b/src/lib/measurements/__tests__/hourly-mint-mock.ts
new file mode 100644
index 000000000..a5131cee9
--- /dev/null
+++ b/src/lib/measurements/__tests__/hourly-mint-mock.ts
@@ -0,0 +1,49 @@
+/**
+ * Transaction-client stubs for `adoptOrMintHourlyRows`, shared by the mocked
+ * dense-tier fold tests.
+ *
+ * The hourly mint reads the day's candidate rows with one `findMany` and
+ * writes its mints with one `createManyAndReturn`. The tests were written
+ * against the per-slot loop and assert on individual `create` calls, so the
+ * batch insert here is routed through the test's own `create` spy, one call
+ * per minted row: the assertions keep reading what was minted, in slot order.
+ */
+import { vi } from "vitest";
+
+type CreateSpy = (args: {
+  data: { externalId: string };
+}) => Promise<{ id: string }>;
+
+/** `createManyAndReturn` that forwards every row to `create`. */
+export function createManyVia(create: CreateSpy) {
+  return vi.fn(async (args: { data: Array<{ externalId: string }> }) => {
+    const out: Array<{ id: string; externalId: string }> = [];
+    for (const data of args.data) {
+      const row = await create({ data });
+      out.push({ id: row.id, externalId: data.externalId });
+    }
+    return out;
+  });
+}
+
+/**
+ * The day's candidate lookup. With `existingId`, one row already occupies the
+ * first slot's anchor instant (the adopt-in-place case); otherwise nothing is
+ * in the way.
+ */
+export function candidateLookup(existingId: string | null = null) {
+  return vi.fn(
+    async (args: {
+      where: { OR: [unknown, { measuredAt: { in: Date[] } }] };
+    }) =>
+      existingId === null
+        ? []
+        : [
+            {
+              id: existingId,
+              externalId: null,
+              measuredAt: args.where.OR[1].measuredAt.in[0]!,
+            },
+          ],
+  );
+}
diff --git a/src/lib/measurements/consolidate-daily-mean.ts b/src/lib/measurements/consolidate-daily-mean.ts
index 097fed61a..dd8c376e0 100644
--- a/src/lib/measurements/consolidate-daily-mean.ts
+++ b/src/lib/measurements/consolidate-daily-mean.ts
@@ -93,6 +93,12 @@ export interface MeanConsolidationBucket {
 
 export interface MeanConsolidationSummary {
   dryRun: boolean;
+  /**
+   * `shouldStop` ended the walk before every day was reached. Nothing is
+   * lost: folded days committed one by one, and the next run starts at the
+   * first day this one did not reach.
+   */
+  stoppedEarly: boolean;
   buckets: MeanConsolidationBucket[];
   totals: {
     usersScanned: number;
@@ -123,6 +129,11 @@ export interface MeanConsolidationOptions {
    * `undefined` to drain everything.
    */
   cutoffHours?: number;
+  /**
+   * Ends the pass cleanly before the next day when it returns `true`; the
+   * summary then reports `stoppedEarly`. See `ConsolidationOptions`.
+   */
+  shouldStop?: () => boolean;
 }
 
 /** Arithmetic mean of a per-day bucket. Exposed for unit testing. */
@@ -205,6 +216,7 @@ export async function consolidateDailyMean(
 
   const summary: MeanConsolidationSummary = {
     dryRun: options.dryRun ?? false,
+    stoppedEarly: false,
     buckets: [],
     totals: {
       usersScanned: 0,
@@ -215,7 +227,7 @@ export async function consolidateDailyMean(
     },
   };
 
-  const { usersScanned, daysFailed } = await runConsolidation<MeasurementType>({
+  const walk = await runConsolidation<MeasurementType>({
     prismaClient,
     options,
     types: HIGH_FREQUENCY_MEAN_TYPES,
@@ -420,8 +432,9 @@ export async function consolidateDailyMean(
     },
   });
 
-  summary.totals.usersScanned = usersScanned;
-  summary.totals.daysFailed = daysFailed;
+  summary.totals.usersScanned = walk.usersScanned;
+  summary.stoppedEarly = walk.stoppedEarly;
+  summary.totals.daysFailed = walk.daysFailed;
 
   log(
     `[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`,
diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts
index a94dce1b2..01abebbbe 100644
--- a/src/lib/measurements/consolidation-base.ts
+++ b/src/lib/measurements/consolidation-base.ts
@@ -32,6 +32,8 @@ import type {
   Prisma,
 } from "@/generated/prisma/client";
 
+import { reportJobProgress } from "@/lib/jobs/job-observer";
+
 import {
   CONSOLIDATION_GRACE_CUTOFF_HOURS,
   canonicalDailyTimestamp,
@@ -57,6 +59,18 @@ export interface ConsolidationOptions {
   dryRun?: boolean;
   /** Logger sink — defaults to `console.log`. */
   log?: (line: string) => void;
+  /**
+   * Asked before every day bucket. Returning `true` ends the pass cleanly
+   * after the day in flight has committed, and the result reports
+   * `stoppedEarly`. The queue handlers wire it to the job's time budget and
+   * to pg-boss's abort signal, so a pass that cannot finish inside its expiry
+   * stops on its own instead of being declared dead while it keeps writing.
+   *
+   * Stopping loses nothing: every day commits on its own, and a folded day's
+   * source rows leave the scan predicate, so the next run starts at the first
+   * day this one did not reach.
+   */
+  shouldStop?: () => boolean;
 }
 
 /**
@@ -229,7 +243,7 @@ export interface ConsolidationParams<TType extends MeasurementType> {
     dryRun: boolean;
   }) => void;
   /**
-   * Fired once per (user, type) immediately after the rows are bucketed,
+   * Fired once per (user, type) after the type's rows have been walked,
    * with the raw scanned-row count and per-day bucket count. Only invoked
    * for types that yielded at least one source row. Lets a single-type
    * drain (legacy steps) reproduce its scan-time log line. Optional.
@@ -275,35 +289,38 @@ const DEFAULT_SCAN_SELECT: Prisma.MeasurementSelect = {
 };
 
 /**
- * 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.
+ * Page size for the keyset-paginated source-row scan.
  */
 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.
+ * ascending, one page at a time.
  *
- * `(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.
+ * Two properties matter, and both used to be missing.
+ *
+ * The pages are yielded, not accumulated. The previous helper gathered every
+ * page into one array before a single day was folded, so the pass held a
+ * type's whole history at once: on an account with 766 000 heart-rate samples
+ * that is about 230 MB of live heap for the array alone, most of a 524 MB
+ * limit, and nothing was written until the whole scan had finished.
+ *
+ * The cursor is index-usable. `measuredAt > c OR (measuredAt = c AND id > i)`
+ * cannot be an index condition, so Postgres walked the index from the user's
+ * first row on every page and filtered its way to the cursor, which makes the
+ * whole scan quadratic (measured: 47 ms and 248 000 rows filtered for one
+ * page halfway through). The redundant `measuredAt >= c` conjunct is the
+ * index condition; the OR only breaks ties at the boundary instant.
+ *
+ * The caller's `baseWhere` is AND-combined, so the scope / soft-delete /
+ * grace filters always still apply.
  */
-export async function scanSourceRowsPaged(
+export async function* iterateSourcePages(
   prismaClient: PrismaClient,
   baseWhere: Prisma.MeasurementWhereInput,
   scanSelect: Prisma.MeasurementSelect,
   pageSize: number = CONSOLIDATION_SCAN_PAGE_SIZE,
-): Promise<PerSampleRow[]> {
-  const accumulated: PerSampleRow[] = [];
+): AsyncGenerator<PerSampleRow[], void, void> {
   let cursor: { measuredAt: Date; id: string } | null = null;
 
   for (;;) {
@@ -311,6 +328,7 @@ export async function scanSourceRowsPaged(
       ? {
           AND: [
             baseWhere,
+            { measuredAt: { gte: cursor.measuredAt } },
             {
               OR: [
                 { measuredAt: { gt: cursor.measuredAt } },
@@ -328,18 +346,54 @@ export async function scanSourceRowsPaged(
       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;
+    if (page.length === 0) return;
 
     const last = page[page.length - 1]!;
+    const isLastPage = page.length < pageSize;
+    yield page;
+
+    // A short page is the last page — no further rows can satisfy the keyset.
+    if (isLastPage) return;
     cursor = { measuredAt: last.measuredAt, id: last.id };
   }
+}
 
-  return accumulated;
+/**
+ * Group a stream of ascending pages into complete per-day buckets, in the
+ * user's timezone, without holding more than one day at a time.
+ *
+ * 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.
+ */
+export async function* iterateDayBuckets(
+  pages: AsyncIterable<readonly PerSampleRow[]>,
+  tz: string,
+  statsPrefix: string,
+  onRow?: () => void,
+): AsyncGenerator<[dateKey: string, rows: PerSampleRow[]], void, void> {
+  let currentKey: string | null = null;
+  let current: PerSampleRow[] = [];
+  for await (const page of pages) {
+    for (const row of page) {
+      onRow?.();
+      if (row.externalId !== null && row.externalId.startsWith(statsPrefix)) {
+        continue;
+      }
+      const key = dayKeyForUserTz(row.measuredAt, tz);
+      if (key !== currentKey) {
+        if (currentKey !== null && current.length > 0) {
+          yield [currentKey, current];
+        }
+        currentKey = key;
+        current = [];
+      }
+      current.push(row);
+    }
+  }
+  if (currentKey !== null && current.length > 0) yield [currentKey, current];
 }
 
 /**
@@ -348,7 +402,8 @@ export async function scanSourceRowsPaged(
  * and delegates the per-day mint + delete to `writeDay`. Returns the
  * number of users scanned (so the caller can seed its summary totals)
  * plus the number of day buckets absorbed by `onBucketError` (0 when the
- * boundary is not supplied).
+ * boundary is not supplied), and whether `options.shouldStop` ended the
+ * walk before every day was reached.
  *
  * Idempotency, the grace-window cutoff, and the "skip already-collapsed
  * rows" predicate are all owned here; the divergent reducer / delete /
@@ -356,16 +411,24 @@ export async function scanSourceRowsPaged(
  */
 export async function runConsolidation<TType extends MeasurementType>(
   params: ConsolidationParams<TType>,
-): Promise<{ usersScanned: number; dryRun: boolean; daysFailed: number }> {
+): Promise<{
+  usersScanned: number;
+  dryRun: boolean;
+  daysFailed: number;
+  stoppedEarly: boolean;
+}> {
   const { prismaClient, options } = params;
   const dryRun = options.dryRun ?? false;
   const cutoffAt = resolveCutoffInstant(options.cutoffHours);
   const scanSelect = params.scanSelect ?? DEFAULT_SCAN_SELECT;
+  const shouldStop = options.shouldStop ?? (() => false);
   let daysFailed = 0;
+  let daysWalked = 0;
+  let stoppedEarly = false;
 
   const users = await loadConsolidationUsers(prismaClient, options.userId);
 
-  for (const user of users) {
+  walk: for (const user of users) {
     const tz = resolveUserTimezone(user.timezone);
     params.onUserStart?.({ userId: user.id, tz, dryRun });
 
@@ -373,36 +436,45 @@ export async function runConsolidation<TType extends MeasurementType>(
       const hkIdentifier = params.hkIdentifierForType(type);
       if (!hkIdentifier) continue;
 
-      // 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,
-        }),
-        scanSelect,
+      // Streamed: each day is folded as soon as its last row has been read,
+      // so the pass holds one page and one day, never the type's history.
+      // See `iterateSourcePages` and `iterateDayBuckets`.
+      let rowCount = 0;
+      let dayCount = 0;
+      const days = iterateDayBuckets(
+        iterateSourcePages(
+          prismaClient,
+          params.buildScanWhere({
+            userId: user.id,
+            type,
+            cutoffAt,
+            statsPrefix: params.statsPrefix,
+          }),
+          scanSelect,
+        ),
+        tz,
+        params.statsPrefix,
+        () => {
+          rowCount += 1;
+        },
       );
 
-      if (sourceRows.length === 0) continue;
-
-      const byDay = bucketRowsByDay(sourceRows, tz, params.statsPrefix);
-
-      params.onScan?.({
-        userId: user.id,
-        type,
-        tz,
-        rowCount: sourceRows.length,
-        dayCount: byDay.size,
-        dryRun,
-      });
-
-      for (const [dateKey, dayRows] of byDay) {
-        if (dayRows.length === 0) continue;
+      for await (const [dateKey, dayRows] of days) {
+        if (shouldStop()) {
+          stoppedEarly = true;
+          // Leaving the loop closes both generators, so no further page is
+          // requested.
+          break walk;
+        }
+        dayCount += 1;
+        daysWalked += 1;
+        // How far the pass has got, for the job's progress and expiry lines.
+        reportJobProgress({
+          consolidation_user: user.id,
+          consolidation_type: type,
+          consolidation_day: dateKey,
+          consolidation_days_walked: daysWalked,
+        });
 
         try {
           const reducedValue = params.reduce(dayRows);
@@ -464,10 +536,21 @@ export async function runConsolidation<TType extends MeasurementType>(
           });
         }
       }
+
+      if (rowCount > 0) {
+        params.onScan?.({
+          userId: user.id,
+          type,
+          tz,
+          rowCount,
+          dayCount,
+          dryRun,
+        });
+      }
     }
 
     params.onUserComplete?.({ userId: user.id, tz, dryRun });
   }
 
-  return { usersScanned: users.length, dryRun, daysFailed };
+  return { usersScanned: users.length, dryRun, daysFailed, stoppedEarly };
 }
diff --git a/src/lib/measurements/dense-intraday-hourly-rebuild.ts b/src/lib/measurements/dense-intraday-hourly-rebuild.ts
index 455e4b613..15e116724 100644
--- a/src/lib/measurements/dense-intraday-hourly-rebuild.ts
+++ b/src/lib/measurements/dense-intraday-hourly-rebuild.ts
@@ -19,7 +19,7 @@
  *       day's local window (the fold's soft-deleted inputs).
  *
  * Per qualifying day: compute per-local-hour means from the tombstoned raw
- * rows (same `meanBucketValue` reducer + `adoptOrMintHourlyRow`
+ * rows (same `meanBucketValue` reducer + `adoptOrMintHourlyRows`
  * adopt-in-place slot logic as the live fold, so a re-run converges), then
  * tombstone the daily row IN THE SAME TRANSACTION as the hourly mint — at
  * no instant are the daily row and its hourly rows both live, so an
@@ -73,7 +73,7 @@ import { meanBucketValue } from "./consolidate-daily-mean";
 import {
   DENSE_INTRADAY_RETENTION_DAYS,
   DENSE_INTRADAY_RETENTION_TYPES,
-  adoptOrMintHourlyRow,
+  adoptOrMintHourlyRows,
   bucketRowsByLocalHour,
   hourlyStatsExternalId,
 } from "./dense-intraday-retention";
@@ -318,11 +318,11 @@ async function rebuildDay(
   const rebuildOnce = async (): Promise<{ dailyRetired: boolean }> => {
     let dailyRetired = false;
     await prismaClient.$transaction(async (tx) => {
-      const canonicalRowIds: string[] = [];
-      for (const [hour, hourRows] of byHour) {
-        const rowId = await adoptOrMintHourlyRow(tx, {
-          userId: input.userId,
-          type: input.type,
+      const canonicalRowIds = await adoptOrMintHourlyRows(tx, {
+        userId: input.userId,
+        type: input.type,
+        unit,
+        slots: byHour.map(([hour, hourRows]) => ({
           externalId: hourlyStatsExternalId(
             input.hkIdentifier,
             input.dateKey,
@@ -330,10 +330,8 @@ async function rebuildDay(
           ),
           anchor: canonicalHourlyTimestamp(input.dateKey, hour, input.tz),
           value: meanBucketValue(hourRows),
-          unit,
-        });
-        canonicalRowIds.push(rowId);
-      }
+        })),
+      });
 
       // Retire the daily row in the SAME transaction — at no instant are
       // the daily row and the hourly rows both live (an AVG-over-live-rows
@@ -359,7 +357,7 @@ async function rebuildDay(
     if (!isUniqueConstraintViolation(err)) throw err;
     // A concurrent writer (the nightly fold racing this rebuild) won a slot
     // mid-transaction. Retry once: the deterministic lookup inside
-    // `adoptOrMintHourlyRow` now resolves the winning row and adopts it.
+    // `adoptOrMintHourlyRows` now resolves the winning row and adopts it.
     outcome = await rebuildOnce();
   }
 
diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts
index 515f8ea19..5c6772c3c 100644
--- a/src/lib/measurements/dense-intraday-retention.ts
+++ b/src/lib/measurements/dense-intraday-retention.ts
@@ -191,10 +191,20 @@ export function bucketRowsByLocalHour(
   return byHour;
 }
 
+/** One hourly slot of a folded day, as `adoptOrMintHourlyRows` receives it. */
+export interface HourlySlot {
+  /** Target hourly `stats:` externalId (index-A identity). */
+  externalId: string;
+  /** Local HH:30 anchor instant (index-B identity). */
+  anchor: Date;
+  value: number;
+}
+
 /**
- * Adopt-or-mint one hourly `stats:` row inside the caller's transaction.
+ * Adopt-or-mint a day's hourly `stats:` rows inside the caller's transaction.
+ * Returns the canonical row id for every slot, in slot order.
  *
- * TWO unique indexes can collide on the mint:
+ * TWO unique indexes can collide on a mint:
  *   A) `(userId, type, source, externalId)` — a row already carrying the
  *      target hourly `stats:` externalId (a prior fold / rebuild pass).
  *   B) `(userId, type, measuredAt, source, sleepStage)` (NULLS NOT
@@ -202,98 +212,149 @@ export function bucketRowsByLocalHour(
  *      anchor instant, which may carry a DIFFERENT externalId (a
  *      per-sample row that happened at HH:30, or a sibling stats row).
  *
- * Determinism (VECTOR 1): resolve by index A FIRST — the row already
- * holding the target externalId IS the canonical hourly row by
- * construction. Only when no such row exists fall back to the index-B row
- * physically occupying the anchor. A stable `orderBy: id` pins the pick.
- * Adopting the wrong sibling and stamping the target externalId onto it
- * would otherwise collide with the real `stats:` row on index A.
+ * Determinism: per slot, resolve by index A FIRST — the row already holding
+ * the target externalId IS the canonical hourly row by construction. Only
+ * when no such row exists fall back to the index-B row physically occupying
+ * the anchor. The lowest id wins a tie. Adopting the wrong sibling and
+ * stamping the target externalId onto it would otherwise collide with the
+ * real `stats:` row on index A.
  *
- * Concurrency (VECTOR 2): the resolve→create is a check-then-act not
- * serialised by the unique index under READ COMMITTED. If a concurrent
- * writer wins the slot between the lookup and the INSERT, the `create`
- * throws P2002 and Postgres aborts the whole transaction. Callers
- * therefore catch P2002 OUTSIDE their transaction (a caught error inside
- * would leave the tx in the aborted `25P02` state) and retry the fold
- * once: on the retry the deterministic lookup finds the now-existing row
- * and ADOPTS it, so the create path never fires twice. Built
- * field-by-field (no spread) per the no-mass-assignment convention.
+ * Why a batch. This used to run per slot — two lookups and a write for each
+ * of up to 24 hours, about 75 statements for one day — and on a multi-year
+ * heart-rate history that loop was most of the pass's 380 000 statements.
+ * Now one lookup reads every row that could matter for the whole day, the
+ * slots are resolved against that snapshot in order, and the mints go out as
+ * one insert. The snapshot is kept up to date as each slot adopts or mints,
+ * so a later slot sees exactly what the one-by-one loop would have read back
+ * from the database: an adopted row that moves to its anchor frees the
+ * instant it left and occupies the one it moved to, and a minted row occupies
+ * its anchor and externalId. Updates run in slot order, as before, and no
+ * mint can collide with them because a slot mints only when nothing in the
+ * snapshot holds its externalId or its anchor.
+ *
+ * Concurrency: the resolve→write is a check-then-act not serialised by the
+ * unique index under READ COMMITTED. If a concurrent writer wins a slot
+ * between the lookup and the INSERT, the insert throws P2002 and Postgres
+ * aborts the whole transaction. Callers therefore catch P2002 OUTSIDE their
+ * transaction (a caught error inside would leave the tx in the aborted
+ * `25P02` state) and retry the fold once: on the retry the lookup finds the
+ * now-existing row and ADOPTS it. Built field-by-field (no spread) per the
+ * no-mass-assignment convention.
  */
-export async function adoptOrMintHourlyRow(
+export async function adoptOrMintHourlyRows(
   tx: Prisma.TransactionClient,
   input: {
     userId: string;
     type: MeasurementType;
-    /** Target hourly `stats:` externalId (index-A identity). */
-    externalId: string;
-    /** Local HH:30 anchor instant (index-B identity). */
-    anchor: Date;
-    value: number;
     unit: string;
+    slots: readonly HourlySlot[];
   },
-): Promise<string> {
-  const eidRow = await tx.measurement.findFirst({
+): Promise<string[]> {
+  if (input.slots.length === 0) return [];
+
+  // Every row that either index could put in the way, lowest id first — the
+  // order the per-slot `findFirst … orderBy id` used to resolve ties by.
+  // `sleepStage` is NULL for these continuous types, matched via the
+  // NULLS-NOT-DISTINCT index.
+  const existing = await tx.measurement.findMany({
     where: {
       userId: input.userId,
       type: input.type,
       source: "APPLE_HEALTH",
-      externalId: input.externalId,
+      OR: [
+        { externalId: { in: input.slots.map((slot) => slot.externalId) } },
+        {
+          measuredAt: { in: input.slots.map((slot) => slot.anchor) },
+          sleepStage: null,
+        },
+      ],
     },
-    select: { id: true },
-    orderBy: { id: "asc" },
-  });
-  // Index-B occupant (`sleepStage` is NULL for these continuous types,
-  // matched via the NULLS-NOT-DISTINCT index). At most one row can exist.
-  const slotRow = await tx.measurement.findFirst({
-    where: {
-      userId: input.userId,
-      type: input.type,
-      source: "APPLE_HEALTH",
-      measuredAt: input.anchor,
-      sleepStage: null,
-    },
-    select: { id: true },
+    select: { id: true, externalId: true, measuredAt: true },
     orderBy: { id: "asc" },
   });
 
-  const adoptTarget = eidRow ?? slotRow;
-  if (adoptTarget) {
-    // Pin the adopted row to the anchor, but only when the anchor slot is
-    // free or already this row — a DIFFERENT row occupying the slot would
-    // otherwise collide on index B. In that rare case the row keeps its
-    // existing instant; the `stats:` externalId is the identity,
-    // measuredAt is secondary.
-    const slotIsFreeForTarget =
-      slotRow === null || slotRow.id === adoptTarget.id;
-    // Adopt: refresh value, stamp the hourly externalId, optionally
-    // re-anchor, and un-tombstone. This coexists with any pre-existing row
-    // on the slot instead of colliding with it.
-    await tx.measurement.update({
-      where: { id: adoptTarget.id },
-      data: {
-        value: input.value,
-        unit: input.unit,
-        externalId: input.externalId,
-        deletedAt: null,
-        ...(slotIsFreeForTarget ? { measuredAt: input.anchor } : {}),
-      },
-    });
-    return adoptTarget.id;
+  // The snapshot the slots resolve against. `id: null` marks a row this
+  // call is about to mint.
+  const state: Array<{
+    id: string | null;
+    externalId: string | null;
+    at: number;
+  }> = existing.map((row) => ({
+    id: row.id,
+    externalId: row.externalId,
+    at: row.measuredAt.getTime(),
+  }));
+
+  const resolved: Array<string | null> = [];
+  const mints: HourlySlot[] = [];
+
+  for (const slot of input.slots) {
+    const anchorAt = slot.anchor.getTime();
+    const eidRow = state.find((row) => row.externalId === slot.externalId);
+    const slotRow = state.find((row) => row.at === anchorAt);
+    const adoptTarget = eidRow ?? slotRow;
+
+    if (adoptTarget && adoptTarget.id !== null) {
+      // Pin the adopted row to the anchor, but only when the anchor slot is
+      // free or already this row — a DIFFERENT row occupying the slot would
+      // otherwise collide on index B. In that rare case the row keeps its
+      // existing instant; the `stats:` externalId is the identity,
+      // measuredAt is secondary.
+      const slotIsFreeForTarget =
+        slotRow === undefined || slotRow === adoptTarget;
+      // Adopt: refresh value, stamp the hourly externalId, optionally
+      // re-anchor, and un-tombstone. This coexists with any pre-existing row
+      // on the slot instead of colliding with it.
+      await tx.measurement.update({
+        where: { id: adoptTarget.id },
+        data: {
+          value: slot.value,
+          unit: input.unit,
+          externalId: slot.externalId,
+          deletedAt: null,
+          ...(slotIsFreeForTarget ? { measuredAt: slot.anchor } : {}),
+        },
+      });
+      adoptTarget.externalId = slot.externalId;
+      if (slotIsFreeForTarget) adoptTarget.at = anchorAt;
+      resolved.push(adoptTarget.id);
+    } else {
+      // Nothing holds this slot's identity or instant — including rows this
+      // call already decided to mint, which carry a distinct anchor and
+      // externalId by construction (one slot per local hour).
+      state.push({ id: null, externalId: slot.externalId, at: anchorAt });
+      mints.push(slot);
+      resolved.push(null);
+    }
   }
 
-  const created = await tx.measurement.create({
-    data: {
-      userId: input.userId,
-      type: input.type,
-      value: input.value,
-      unit: input.unit,
-      source: "APPLE_HEALTH",
-      measuredAt: input.anchor,
-      externalId: input.externalId,
-    },
-    select: { id: true },
-  });
-  return created.id;
+  if (mints.length > 0) {
+    const created = await tx.measurement.createManyAndReturn({
+      data: mints.map((slot) => ({
+        userId: input.userId,
+        type: input.type,
+        value: slot.value,
+        unit: input.unit,
+        source: "APPLE_HEALTH" as const,
+        measuredAt: slot.anchor,
+        externalId: slot.externalId,
+      })),
+      select: { id: true, externalId: true },
+    });
+    const idByExternalId = new Map(
+      created.map((row) => [row.externalId, row.id] as const),
+    );
+    input.slots.forEach((slot, index) => {
+      if (resolved[index] !== null) return;
+      const id = idByExternalId.get(slot.externalId);
+      if (id === undefined) {
+        throw new Error(`hourly mint returned no row for ${slot.externalId}`);
+      }
+      resolved[index] = id;
+    });
+  }
+
+  return resolved as string[];
 }
 
 /**
@@ -347,6 +408,12 @@ export function deriveDailyRestingFromPulse(
 
 export interface DenseIntradayRetentionSummary {
   dryRun: boolean;
+  /**
+   * `shouldStop` ended the walk before every day was reached. Nothing is
+   * lost: folded days committed one by one, and the next run starts at the
+   * first day this one did not reach.
+   */
+  stoppedEarly: boolean;
   totals: {
     usersScanned: number;
     daysConsolidated: number;
@@ -385,6 +452,11 @@ export interface DenseIntradayRetentionOptions {
    * full-collapse, never by the scheduled pass.
    */
   retentionDays?: number;
+  /**
+   * Ends the pass cleanly before the next day when it returns `true`; the
+   * summary then reports `stoppedEarly`. See `ConsolidationOptions`.
+   */
+  shouldStop?: () => boolean;
 }
 
 /**
@@ -412,6 +484,7 @@ export async function runDenseIntradayRetention(
 
   const summary: DenseIntradayRetentionSummary = {
     dryRun: options.dryRun ?? false,
+    stoppedEarly: false,
     totals: {
       usersScanned: 0,
       daysConsolidated: 0,
@@ -462,7 +535,7 @@ export async function runDenseIntradayRetention(
   // for the dry-run hourly fan-out estimate (writeDay never runs on dry-run).
   let currentTz = "Europe/Berlin";
 
-  const { usersScanned } = await runConsolidation<MeasurementType>({
+  const walk = await runConsolidation<MeasurementType>({
     prismaClient,
     options: { ...options, cutoffHours },
     types: DENSE_INTRADAY_RETENTION_TYPES,
@@ -543,18 +616,16 @@ export async function runDenseIntradayRetention(
         let removed = 0;
         let retiredDaily = false;
         await pc.$transaction(async (tx) => {
-          const canonicalRowIds: string[] = [];
-          for (const [hour, hourRows] of byHour) {
-            const rowId = await adoptOrMintHourlyRow(tx, {
-              userId,
-              type,
+          const canonicalRowIds = await adoptOrMintHourlyRows(tx, {
+            userId,
+            type,
+            unit,
+            slots: byHour.map(([hour, hourRows]) => ({
               externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour),
               anchor: canonicalHourlyTimestamp(dateKey, hour, tz),
               value: meanBucketValue(hourRows),
-              unit,
-            });
-            canonicalRowIds.push(rowId);
-          }
+            })),
+          });
 
           // Retire a live pre-hourly DAILY `stats:` row for this day (the
           // late-sync path: raw samples arriving for a day folded to the
@@ -712,7 +783,8 @@ export async function runDenseIntradayRetention(
     },
   });
 
-  summary.totals.usersScanned = usersScanned;
+  summary.totals.usersScanned = walk.usersScanned;
+  summary.stoppedEarly = walk.stoppedEarly;
 
   log(
     `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`,
diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts
index aa0f9c91b..e0cf58135 100644
--- a/src/lib/measurements/drain-per-sample-cumulative.ts
+++ b/src/lib/measurements/drain-per-sample-cumulative.ts
@@ -124,6 +124,12 @@ export interface DrainBucket {
 export interface DrainSummary {
   /** Did the run actually write to the DB (false for `dryRun`). */
   dryRun: boolean;
+  /**
+   * `shouldStop` ended the walk before every day was reached. Nothing is
+   * lost: folded days committed one by one, and the next run starts at the
+   * first day this one did not reach.
+   */
+  stoppedEarly: boolean;
   /** Per-user-day buckets the drain rewrote (or would rewrite). */
   buckets: DrainBucket[];
   /** Aggregate counts across the run. */
@@ -167,6 +173,11 @@ export interface DrainOptions {
    * the operator points at) for explicit one-shot use.
    */
   cutoffHours?: number;
+  /**
+   * Ends the pass cleanly before the next day when it returns `true`; the
+   * summary then reports `stoppedEarly`. See `ConsolidationOptions`.
+   */
+  shouldStop?: () => boolean;
 }
 
 export interface BucketedRows {
@@ -210,6 +221,7 @@ export async function drainPerSampleCumulative(
 
   const summary: DrainSummary = {
     dryRun: options.dryRun ?? false,
+    stoppedEarly: false,
     buckets: [],
     totals: {
       usersScanned: 0,
@@ -229,7 +241,7 @@ export async function drainPerSampleCumulative(
   let beforePerSampleDeleted = 0;
   let beforeDailyUpserted = 0;
 
-  const { usersScanned } = await runConsolidation<MeasurementType>({
+  const walk = await runConsolidation<MeasurementType>({
     prismaClient,
     options,
     types: CUMULATIVE_HK_TYPES,
@@ -521,7 +533,8 @@ export async function drainPerSampleCumulative(
     },
   });
 
-  summary.totals.usersScanned = usersScanned;
+  summary.totals.usersScanned = walk.usersScanned;
+  summary.stoppedEarly = walk.stoppedEarly;
 
   // v1.34.0 — bound the profile sidecar. One indexed range delete per run,
   // across every account, on the same nightly tick that writes the profiles.