Merge branch 'fix/large-account-jobs' into release/v1.39.1

tests/integration/apple-health-import-rollup-window.test.ts

@@ -0,0 +1,116 @@
+/**
+ * #1031 — the rollup fold after an import covers what the import wrote.
+ *
+ * The worker folded the rollup table from the account's first measurement to
+ * its last, whatever the import carried, and so also folded history older
+ * than the five-year window every other fold path keeps to. It now folds the
+ * span of the rows this import wrote, clamped to that window.
+ */
+import { createHash } from "node:crypto";
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { strToU8, zipSync } from "fflate";
+import type { Job } from "pg-boss";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import {
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
+  handleAppleHealthImport,
+  _setWorkerPrismaForTests,
+  type AppleHealthImportPayload,
+} from "@/lib/jobs/apple-health-import-worker";
+import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: vi.fn(() => null),
+}));
+
+const USER = "user-import-rollup-window";
+
+function record(at: Date, value: number): string {
+  const stamp = at.toISOString().replace("T", " ").slice(0, 19) + " +0000";
+  return `<Record type="HKQuantityTypeIdentifierBodyMass" unit="kg" startDate="${stamp}" endDate="${stamp}" value="${value}" sourceName="Scale"/>`;
+}
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  _setWorkerPrismaForTests(getPrismaClient());
+  await getPrismaClient().user.create({
+    data: {
+      id: USER,
+      username: USER,
+      email: `${USER}@example.test`,
+      timezone: "UTC",
+    },
+  });
+});
+
+describe("import rollup fold window", () => {
+  it("folds the imported span, never past the fold window, and leaves older history alone", async () => {
+    const prisma = getPrismaClient();
+    const now = Date.now();
+    const day = 86_400_000;
+    // Already in the account before the import: a reading eight years old.
+    await prisma.measurement.create({
+      data: {
+        userId: USER,
+        type: "WEIGHT",
+        value: 90,
+        unit: "kg",
+        source: "MANUAL",
+        measuredAt: new Date(now - 8 * 365 * day),
+      },
+    });
+    // The import: one reading seven years old, a few recent ones.
+    const xml = `<?xml version="1.0" encoding="UTF-8"?>
+<HealthData locale="en_US">
+  ${record(new Date(now - 7 * 365 * day), 88)}
+  ${record(new Date(now - 40 * day), 84)}
+  ${record(new Date(now - 30 * day), 83)}
+  ${record(new Date(now - 20 * day), 82)}
+</HealthData>`;
+    const dir = mkdtempSync(join(tmpdir(), "hl-import-window-"));
+    const path = join(dir, "export.zip");
+    const bytes = Buffer.from(
+      zipSync({ "apple_health_export/export.xml": strToU8(xml) }),
+    );
+    writeFileSync(path, bytes);
+    await prisma.importJob.create({
+      data: {
+        userId: USER,
+        pgBossJobId: "window-job",
+        status: "queued",
+        uploadBytes: bytes.length,
+        uploadSha256: createHash("sha256").update(bytes).digest("hex"),
+        parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+      },
+    });
+
+    const outcome = await handleAppleHealthImport({
+      id: "window-job",
+      data: {
+        userId: USER,
+        uploadPath: path,
+        uploadBytes: bytes.length,
+        enqueuedAt: new Date().toISOString(),
+      },
+    } as unknown as Job<AppleHealthImportPayload>);
+    expect(outcome.ok).toBe(true);
+
+    const buckets = await prisma.measurementRollup.findMany({
+      where: { userId: USER, type: "WEIGHT", granularity: "DAY" },
+      orderBy: { bucketStart: "asc" },
+    });
+    const windowStart = now - ROLLUP_FOLD_WINDOW_MS;
+    expect(buckets.length).toBeGreaterThanOrEqual(3);
+    for (const bucket of buckets) {
+      expect(bucket.bucketStart.getTime()).toBeGreaterThan(windowStart - day);
+    }
+  });
+});

tests/integration/cumulative-pr-rederive-atomic.test.ts

@@ -0,0 +1,94 @@
+/**
+ * #1031 — the cumulative-record repair must not lose records when the
+ * re-derivation fails.
+ *
+ * The repair deleted every suspect cumulative record, committed, and only
+ * then re-ran the detector. A detector that failed (a timeout on a large
+ * account, a crash) left the records deleted, and because discovery looks
+ * for the suspect rows the deletion removed, the account was never
+ * repaired again: its step, energy and distance records were simply gone.
+ * The delete and the re-derivation now commit together or not at all.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const detect = vi.hoisted(() => ({ fail: false }));
+vi.mock("@/lib/personal-records/pr-detection-worker", async (importActual) => {
+  const actual =
+    await importActual<
+      typeof import("@/lib/personal-records/pr-detection-worker")
+    >();
+  return {
+    ...actual,
+    detectPersonalRecordsForUser: async (
+      ...args: Parameters<typeof actual.detectPersonalRecordsForUser>
+    ) => {
+      if (detect.fail) throw new Error("detector ran out of time");
+      return actual.detectPersonalRecordsForUser(...args);
+    },
+  };
+});
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const { runCumulativePrRederivationForUser, CUMULATIVE_PR_FIX_CUTOFF } =
+  await import("@/lib/personal-records/cumulative-pr-rederivation");
+
+const USER = "user-pr-rederive-atomic";
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  detect.fail = false;
+  await prisma.user.create({
+    data: { id: USER, username: USER, email: `${USER}@example.test` },
+  });
+  const before = new Date(CUMULATIVE_PR_FIX_CUTOFF.getTime() - 86_400_000);
+  await prisma.personalRecord.create({
+    data: {
+      userId: USER,
+      metricType: "ACTIVITY_STEPS",
+      direction: "MAX",
+      value: 99_999,
+      unit: "steps",
+      achievedAt: new Date("2025-03-01T12:00:00Z"),
+      createdAt: before,
+    },
+  });
+  // Enough steps for the detector's warm-up gate, one clear best day.
+  await prisma.measurement.createMany({
+    data: Array.from({ length: 10 }, (_, day) => ({
+      userId: USER,
+      type: "ACTIVITY_STEPS" as const,
+      value: day === 4 ? 12_000 : 5_000,
+      unit: "steps",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: new Date(Date.UTC(2025, 3, 1 + day, 12)),
+      externalId: `s-${day}`,
+    })),
+  });
+});
+
+describe("runCumulativePrRederivationForUser", () => {
+  it("keeps the suspect record when the re-derivation fails", async () => {
+    detect.fail = true;
+    await expect(runCumulativePrRederivationForUser(USER)).rejects.toThrow(
+      "detector ran out of time",
+    );
+    const left = await getPrismaClient().personalRecord.findMany({
+      where: { userId: USER },
+    });
+    expect(left.map((r) => r.value)).toEqual([99_999]);
+  });
+
+  it("replaces the suspect record with the re-derived one when it succeeds", async () => {
+    const summary = await runCumulativePrRederivationForUser(USER);
+    expect(summary.rowsDeleted).toBe(1);
+    const left = await getPrismaClient().personalRecord.findMany({
+      where: { userId: USER },
+    });
+    expect(left.map((r) => r.value)).toEqual([12_000]);
+  });
+});

tests/integration/job-failures.test.ts

@@ -16,6 +16,7 @@ import { PgBoss } from "pg-boss";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
 import {
+  readActiveJobsStartedBefore,
   readFailingQueues,
   readQueueFailureForUser,
 } from "@/lib/jobs/job-failures";
@@ -28,8 +29,16 @@ const OTHER_QUEUE = "reminder-check";
 let boss: PgBoss;
 
 /** Send a job, take it, and fail it terminally so it lands in `state='failed'`. */
-async function failOneJob(queue: string, data: object, error: string) {
-  const jobId = await boss.send(queue, data, { retryLimit: 0 });
+async function failOneJob(
+  queue: string,
+  data: object,
+  error: string,
+  expireInSeconds?: number,
+) {
+  const jobId = await boss.send(queue, data, {
+    retryLimit: 0,
+    ...(expireInSeconds === undefined ? {} : { expireInSeconds }),
+  });
   if (!jobId) throw new Error("queue rejected the job");
   const [job] = await boss.fetch(queue);
   if (!job) throw new Error("nothing to fetch");
@@ -81,6 +90,26 @@ describe("readFailingQueues", () => {
     expect(other?.lastError).toBe("reminder dispatch failed");
   });
 
+  it("says what a bare timeout means, with the job's own limit (#1031)", async () => {
+    await getPrismaClient().$executeRawUnsafe(
+      `DELETE FROM pgboss.job WHERE state = 'failed'`,
+    );
+    // pg-boss writes these words both for a pass too slow for its limit and
+    // for a job whose process died under it; alone they sent an operator
+    // nowhere.
+    await failOneJob(QUEUE, {}, "job timed out", 900);
+    await failOneJob(OTHER_QUEUE, {}, "job timed out", 7200);
+
+    const failing = await readFailingQueues();
+    const fifteen = failing?.find((entry) => entry.queue === QUEUE);
+    expect(fifteen?.lastError).toMatch(
+      /^job timed out: did not finish within its 15 min limit\./,
+    );
+    expect(fifteen?.lastError).toMatch(/restarted/);
+    const twoHours = failing?.find((entry) => entry.queue === OTHER_QUEUE);
+    expect(twoHours?.lastError).toMatch(/within its 2 h limit/);
+  });
+
   it("reports an empty list — not null — when nothing is failing", async () => {
     await getPrismaClient().$executeRawUnsafe(
       `DELETE FROM pgboss.job WHERE state = 'failed'`,
@@ -132,6 +161,34 @@ describe("readQueueFailureForUser", () => {
   });
 });
 
+describe("readActiveJobsStartedBefore (#1031)", () => {
+  it("lists a job still marked active from before the given instant, and nothing after it", async () => {
+    const jobId = await boss.send(
+      OTHER_QUEUE,
+      {},
+      { retryLimit: 0, expireInSeconds: 7200 },
+    );
+    const [job] = await boss.fetch(OTHER_QUEUE);
+    expect(job?.id).toBe(jobId);
+
+    const cutOff = await readActiveJobsStartedBefore(
+      new Date(Date.now() + 60_000),
+    );
+    expect(cutOff).toContainEqual(
+      expect.objectContaining({
+        queue: OTHER_QUEUE,
+        id: jobId,
+        expireSeconds: 7200,
+      }),
+    );
+    await expect(
+      readActiveJobsStartedBefore(new Date(Date.now() - 60 * 60_000)),
+    ).resolves.toEqual([]);
+
+    await boss.complete(OTHER_QUEUE, jobId!);
+  });
+});
+
 describe("no queue schema at all", () => {
   it("answers null rather than throwing, so the status page still renders", async () => {
     const prisma = getPrismaClient();

tests/integration/large-account-jobs.test.ts

@@ -0,0 +1,266 @@
+/**
+ * Issue #1031 — maintenance jobs on a large account.
+ *
+ * A self-hoster with 1.25 million measurement rows (a multi-year Apple Health
+ * export) saw the dense intra-day retention, the nightly drain tick and the
+ * data backup all fail with `job timed out`. Reproduced against a seeded
+ * account of the same size (`scripts/seed-large-account.ts`): the drains held
+ * a type's whole history in memory before folding a day, spent about 90
+ * statements per folded day, and paged with a cursor Postgres could not use as
+ * an index condition; the backup added about 640 MB to the process to store
+ * one 64 MB row. pg-boss declared the jobs dead at their expiry without
+ * stopping them.
+ *
+ * A million rows do not fit a test run, so this file pins the properties the
+ * fix rests on at a scale that does: a pass that runs out of budget stops
+ * cleanly and the next run finishes the job with the same result as one
+ * uninterrupted run, and the backup is stored piecewise with the previous
+ * copy intact if a run fails halfway.
+ */
+import { beforeEach, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention";
+import { consolidateDailyMean } from "@/lib/measurements/consolidate-daily-mean";
+import { unpackBackupBlob } from "@/lib/export/backup-blob";
+import { storeBackupBlob } from "@/lib/export/store-backup-blob";
+
+const TZ = "Europe/Berlin";
+const RUN_A = "user-1031-stopped";
+const RUN_B = "user-1031-straight";
+const DAYS = 12;
+/** Oldest seeded instant: well outside the 90-day raw window. */
+const START = Date.UTC(2026, 0, 1, 0, 7);
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  for (const id of [RUN_A, RUN_B]) {
+    await prisma.user.create({
+      data: { id, username: id, email: `${id}@example.test`, timezone: TZ },
+    });
+  }
+});
+
+/** The same history for both users: heart rate every 20 minutes, HRV hourly. */
+async function seedHistory(userId: string): Promise<void> {
+  const rows = [];
+  for (let minute = 0; minute < DAYS * 24 * 60; minute += 20) {
+    const at = new Date(START + minute * 60_000);
+    rows.push({
+      userId,
+      type: "PULSE" as const,
+      value: 55 + ((minute / 20) % 40),
+      unit: "bpm",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: at,
+      externalId: `hk-hr-${minute}`,
+    });
+    if (minute % 60 === 0) {
+      rows.push({
+        userId,
+        type: "HEART_RATE_VARIABILITY" as const,
+        value: 30 + ((minute / 60) % 25),
+        unit: "ms",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: at,
+        externalId: `hk-hrv-${minute}`,
+      });
+    }
+    if (minute % 40 === 0) {
+      rows.push({
+        userId,
+        type: "WALKING_SPEED" as const,
+        value: 4 + ((minute / 40) % 10) / 10,
+        unit: "km/h",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: at,
+        externalId: `hk-ws-${minute}`,
+      });
+    }
+  }
+  await getPrismaClient().measurement.createMany({ data: rows });
+}
+
+/** What a reader sees: every live row, identity-free. */
+async function liveRows(userId: string) {
+  const rows = await getPrismaClient().measurement.findMany({
+    where: { userId, deletedAt: null },
+    select: {
+      type: true,
+      value: true,
+      unit: true,
+      source: true,
+      measuredAt: true,
+      externalId: true,
+    },
+    orderBy: [{ type: "asc" }, { measuredAt: "asc" }],
+  });
+  return rows.map((row) => ({
+    ...row,
+    measuredAt: row.measuredAt.toISOString(),
+  }));
+}
+
+/** A `shouldStop` that lets `days` day buckets through, then says stop. */
+function afterDays(days: number): () => boolean {
+  let left = days;
+  return () => left-- <= 0;
+}
+
+describe("a pass that runs out of budget (#1031)", () => {
+  it("dense retention: stops cleanly, and the next run ends where one straight run ends", async () => {
+    await seedHistory(RUN_A);
+    await seedHistory(RUN_B);
+    const prisma = getPrismaClient();
+
+    const first = await runDenseIntradayRetention(prisma, {
+      userId: RUN_A,
+      shouldStop: afterDays(5),
+      log: () => {},
+    });
+    expect(first.stoppedEarly).toBe(true);
+    expect(first.totals.daysConsolidated).toBe(5);
+
+    // What the stopped run folded is committed: the next scan no longer sees
+    // it, so the second run folds only what is left.
+    const second = await runDenseIntradayRetention(prisma, {
+      userId: RUN_A,
+      log: () => {},
+    });
+    expect(second.stoppedEarly).toBe(false);
+
+    const straight = await runDenseIntradayRetention(prisma, {
+      userId: RUN_B,
+      log: () => {},
+    });
+    expect(first.totals.daysConsolidated + second.totals.daysConsolidated).toBe(
+      straight.totals.daysConsolidated,
+    );
+    expect(
+      first.totals.perSampleRowsSoftDeleted +
+        second.totals.perSampleRowsSoftDeleted,
+    ).toBe(straight.totals.perSampleRowsSoftDeleted);
+
+    // A third run finds nothing: the pass converged.
+    const third = await runDenseIntradayRetention(prisma, {
+      userId: RUN_A,
+      log: () => {},
+    });
+    expect(third.totals.daysConsolidated).toBe(0);
+
+    const stoppedResult = (await liveRows(RUN_A)).filter(
+      (row) => row.type !== "WALKING_SPEED",
+    );
+    const straightResult = (await liveRows(RUN_B)).filter(
+      (row) => row.type !== "WALKING_SPEED",
+    );
+    expect(stoppedResult.length).toBeGreaterThan(0);
+    expect(stoppedResult).toEqual(straightResult);
+  });
+
+  it("daily-mean consolidation: the same, for the nightly tick's middle pass", async () => {
+    await seedHistory(RUN_A);
+    await seedHistory(RUN_B);
+    const prisma = getPrismaClient();
+
+    const first = await consolidateDailyMean(prisma, {
+      userId: RUN_A,
+      shouldStop: afterDays(4),
+      log: () => {},
+    });
+    expect(first.stoppedEarly).toBe(true);
+    expect(first.totals.daysConsolidated).toBe(4);
+    await consolidateDailyMean(prisma, { userId: RUN_A, log: () => {} });
+    await consolidateDailyMean(prisma, { userId: RUN_B, log: () => {} });
+
+    const stopped = (await liveRows(RUN_A)).filter(
+      (row) => row.type === "WALKING_SPEED",
+    );
+    const straight = (await liveRows(RUN_B)).filter(
+      (row) => row.type === "WALKING_SPEED",
+    );
+    // One daily mean per local day: twelve UTC days touch thirteen Berlin days.
+    expect(stopped).toHaveLength(DAYS + 1);
+    expect(stopped).toEqual(straight);
+  });
+});
+
+describe("storeBackupBlob (#1031)", () => {
+  /** A producer writing `rows` JSON rows in small pieces. */
+  function producer(rows: number, failAfter?: number) {
+    return async (write: (chunk: string) => Promise<void>) => {
+      await write('{"measurements":[');
+      for (let i = 0; i < rows; i++) {
+        if (failAfter !== undefined && i === failAfter) {
+          throw new Error("record read failed halfway");
+        }
+        await write(
+          `${i === 0 ? "" : ","}${JSON.stringify({ id: `m-${i}`, value: i })}`,
+        );
+      }
+      await write("]}");
+    };
+  }
+
+  function expectedJson(rows: number): string {
+    return JSON.stringify({
+      measurements: Array.from({ length: rows }, (_, i) => ({
+        id: `m-${i}`,
+        value: i,
+      })),
+    });
+  }
+
+  it("stores a copy that reads back as exactly the JSON produced", async () => {
+    const prisma = getPrismaClient();
+    const { bytes } = await storeBackupBlob(
+      prisma,
+      { userId: RUN_A, type: "WEEKLY_AUTO" },
+      producer(20_000),
+    );
+
+    const row = await prisma.dataBackup.findUniqueOrThrow({
+      where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } },
+    });
+    expect(row.data.length).toBe(bytes);
+    expect(unpackBackupBlob(row.data)).toBe(expectedJson(20_000));
+  });
+
+  it("keeps the previous copy when a run fails halfway, and the next run still stores", async () => {
+    const prisma = getPrismaClient();
+    await storeBackupBlob(
+      prisma,
+      { userId: RUN_A, type: "WEEKLY_AUTO" },
+      producer(10),
+    );
+    const before = await prisma.dataBackup.findUniqueOrThrow({
+      where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } },
+    });
+
+    await expect(
+      storeBackupBlob(
+        prisma,
+        { userId: RUN_A, type: "WEEKLY_AUTO" },
+        producer(20_000, 15_000),
+      ),
+    ).rejects.toThrow("record read failed halfway");
+
+    const after = await prisma.dataBackup.findUniqueOrThrow({
+      where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } },
+    });
+    expect(after.data).toBe(before.data);
+    expect(after.createdAt.getTime()).toBe(before.createdAt.getTime());
+
+    // The temporary table went with the failed transaction.
+    await storeBackupBlob(
+      prisma,
+      { userId: RUN_A, type: "WEEKLY_AUTO" },
+      producer(30),
+    );
+    const replaced = await prisma.dataBackup.findUniqueOrThrow({
+      where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } },
+    });
+    expect(unpackBackupBlob(replaced.data)).toBe(expectedJson(30));
+  });
+});

tests/integration/large-backup-round-trip.test.ts

@@ -0,0 +1,203 @@
+/**
+ * #1031 — an account too large to export or restore.
+ *
+ * For an account of 1.25 million measurements the full export took the app
+ * down, and its weekly backup could not be restored, downloaded or summarised
+ * at all: the backup's JSON (662 MB) is longer than any string V8 can hold,
+ * and every reader turned the file into one string first. Measured on a
+ * seeded account of that size in a 1 GB container, the streamed paths now
+ * export in 25 s, upload in under 20 s and restore in 94 s, and the restored
+ * rows match the file exactly.
+ *
+ * A million rows do not fit a test run. This file runs the same paths at a
+ * size that spans many pages and batches, and checks the rows come back
+ * exactly: the export streamed into the response, uploaded as gzip through
+ * the raw-body path, and restored through the streamed two-pass reader; and a
+ * weekly backup restored the same way, every column intact, with a personal
+ * record still pointing at the measurement it was found in.
+ */
+import { gzipSync } from "node:zlib";
+
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+process.env.ENCRYPTION_KEY ??=
+  "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+
+import { cookieJar, headerJar } from "./mock-next-headers";
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
+import { storeBackupBlob } from "@/lib/export/store-backup-blob";
+import { encryptNote } from "@/lib/crypto/note-cipher";
+
+vi.mock("next/headers", async () => {
+  const { cookieJar, headerJar } = await import("./mock-next-headers");
+  return {
+    headers: vi.fn(async () => ({
+      get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
+    })),
+    cookies: vi.fn(async () => ({
+      get: (name: string) => {
+        const value = cookieJar.get(name);
+        return value ? { name, value } : undefined;
+      },
+      set: (name: string, value: string) => {
+        cookieJar.set(name, value);
+      },
+      delete: (name: string) => {
+        cookieJar.delete(name);
+      },
+    })),
+  };
+});
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const COUNT = 7_300; // more than one read page and several write batches
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  cookieJar.clear();
+  headerJar.clear();
+});
+
+async function seedAccount() {
+  const prisma = getPrismaClient();
+  const user = await prisma.user.create({
+    data: {
+      username: "large-account",
+      email: "large-account@example.test",
+      role: "ADMIN",
+    },
+  });
+  const session = await prisma.session.create({
+    data: { userId: user.id, expiresAt: new Date(Date.now() + 600_000) },
+  });
+  cookieJar.set("healthlog_session", session.id);
+
+  const start = Date.UTC(2024, 0, 1);
+  await prisma.measurement.createMany({
+    data: Array.from({ length: COUNT }, (_, i) => ({
+      id: `m-${String(i).padStart(5, "0")}`,
+      userId: user.id,
+      type: i % 3 === 0 ? ("PULSE" as const) : ("SLEEP_DURATION" as const),
+      value: 50 + (i % 40) + 0.25,
+      valueMax: i % 3 === 0 ? 120 : null,
+      unit: i % 3 === 0 ? "bpm" : "minutes",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: new Date(start + i * 300_000),
+      externalId: `hk-${i}`,
+      sleepStage: i % 3 === 0 ? null : ("CORE" as const),
+      deviceType: "watch",
+      syncVersion: 1 + (i % 4),
+      notesEncrypted: i === 17 ? encryptNote("after the long run") : null,
+      deletedAt: i % 97 === 0 ? new Date(start) : null,
+    })),
+  });
+  await prisma.personalRecord.create({
+    data: {
+      userId: user.id,
+      metricType: "PULSE",
+      direction: "MAX",
+      value: 89.25,
+      unit: "bpm",
+      achievedAt: new Date(start + 3 * 300_000),
+      sourceMeasurementId: "m-00003",
+    },
+  });
+  return user;
+}
+
+async function measurementsOf(userId: string) {
+  return getPrismaClient().measurement.findMany({
+    where: { userId },
+    orderBy: { id: "asc" },
+    omit: { createdAt: true, updatedAt: true },
+  });
+}
+
+async function restore(id: string) {
+  const { POST } = await import("@/app/api/admin/backups/[id]/restore/route");
+  const res = await POST(
+    new Request(`http://localhost/api/admin/backups/${id}/restore`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ confirm: "RESTORE" }),
+    }) as never,
+    { params: Promise.resolve({ id }) },
+  );
+  return res;
+}
+
+describe("large backups round-trip through the streamed paths (#1031)", () => {
+  it("export → gzip upload → restore gives back every live measurement", async () => {
+    const user = await seedAccount();
+    const live = (await measurementsOf(user.id)).filter((m) => !m.deletedAt);
+
+    const { GET } = await import("@/app/api/export/full-backup/route");
+    const exported = await GET(
+      new Request("http://localhost/api/export/full-backup") as never,
+    );
+    expect(exported.status).toBe(200);
+    const file = Buffer.from(await exported.arrayBuffer());
+    expect(JSON.parse(file.toString("utf8")).measurements).toHaveLength(
+      live.length,
+    );
+
+    const { POST: upload } =
+      await import("@/app/api/admin/backups/upload/route");
+    const uploaded = await upload(
+      new Request("http://localhost/api/admin/backups/upload", {
+        method: "POST",
+        headers: { "content-type": "application/gzip" },
+        body: gzipSync(file),
+      }) as never,
+    );
+    expect(uploaded.status).toBe(201);
+    const { data } = (await uploaded.json()) as {
+      data: { id: string; summary: { measurements: number } };
+    };
+    expect(data.summary.measurements).toBe(live.length);
+
+    expect((await restore(data.id)).status).toBe(200);
+    const restored = await measurementsOf(user.id);
+    // A portable file carries the fields a person reads, not the sync
+    // bookkeeping, so those are what must come back.
+    const readable = (rows: typeof restored) =>
+      rows.map((m) => ({
+        id: m.id,
+        type: m.type,
+        value: m.value,
+        unit: m.unit,
+        measuredAt: m.measuredAt.toISOString(),
+        source: m.source,
+      }));
+    expect(readable(restored)).toEqual(readable(live));
+  });
+
+  it("a weekly backup restores every column, and a record's pointer survives", async () => {
+    const user = await seedAccount();
+    const before = await measurementsOf(user.id);
+    const prisma = getPrismaClient();
+
+    const { id } = await storeBackupBlob(
+      prisma,
+      { userId: user.id, type: "WEEKLY_AUTO" },
+      (write) =>
+        streamFullBackupJson(prisma, user.id, write, {
+          purpose: "disaster-recovery",
+        }),
+    );
+
+    // Change the account so the restore has something to undo.
+    await prisma.measurement.deleteMany({ where: { userId: user.id } });
+
+    const res = await restore(id);
+    expect(res.status).toBe(200);
+    expect(await measurementsOf(user.id)).toEqual(before);
+    const record = await prisma.personalRecord.findFirstOrThrow({
+      where: { userId: user.id },
+    });
+    expect(record.sourceMeasurementId).toBe("m-00003");
+  });
+});

tests/integration/measurement-bulk-insert.test.ts

@@ -0,0 +1,139 @@
+/**
+ * `insertMeasurementRows` replaced `createMany` in the restore because
+ * Prisma's cached plan for a 1 000-row `createMany` kept about 145 MB alive
+ * (#1031). It writes raw SQL, so this pins that a row written through it
+ * reads back exactly as the same row written through Prisma, every column
+ * set, and every column null where it may be.
+ */
+import { beforeEach, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import {
+  insertMeasurementRows,
+  type MeasurementInsertRow,
+} from "@/lib/export/measurement-bulk-insert";
+
+const USER = "user-bulk-insert";
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: { id: USER, username: USER, email: `${USER}@example.test` },
+  });
+});
+
+function full(id: string, at: string): MeasurementInsertRow {
+  return {
+    id,
+    userId: USER,
+    type: "SLEEP_DURATION",
+    value: 47.25,
+    valueMin: 1.5,
+    valueMax: 99.75,
+    unit: "minutes",
+    source: "APPLE_HEALTH",
+    measuredAt: new Date(at),
+    notes: "legacy note",
+    notesEncrypted: new Uint8Array([0, 1, 2, 254, 255]),
+    externalId: `ext-${id}`,
+    externalSourceVersion: "17.4",
+    aggregationProvenance: "HEALTHKIT_STATISTICS",
+    glucoseContext: null,
+    sleepStage: "REM",
+    rhythmClassification: null,
+    deviceType: "watch",
+    syncVersion: 7,
+    deletedAt: new Date("2026-02-03T04:05:06.789Z"),
+    createdAt: new Date("2025-01-02T03:04:05.678Z"),
+    updatedAt: new Date("2025-06-07T08:09:10.111Z"),
+  };
+}
+
+function sparse(id: string, at: string): MeasurementInsertRow {
+  return {
+    id,
+    userId: USER,
+    type: "BLOOD_GLUCOSE",
+    value: 5.4,
+    valueMin: null,
+    valueMax: null,
+    unit: "mmol/L",
+    source: "MANUAL",
+    measuredAt: new Date(at),
+    notes: null,
+    notesEncrypted: null,
+    externalId: null,
+    externalSourceVersion: null,
+    aggregationProvenance: null,
+    glucoseContext: "FASTING",
+    sleepStage: null,
+    rhythmClassification: null,
+    deviceType: null,
+    syncVersion: 1,
+    deletedAt: null,
+  };
+}
+
+describe("insertMeasurementRows", () => {
+  it("writes every column exactly as Prisma create does", async () => {
+    const prisma = getPrismaClient();
+    const rows = [
+      [
+        full("a-bulk", "2026-01-01T10:00:00.123Z"),
+        full("a-prisma", "2026-01-02T10:00:00.123Z"),
+      ],
+      [
+        sparse("b-bulk", "2026-01-03T10:00:00.000Z"),
+        sparse("b-prisma", "2026-01-04T10:00:00.000Z"),
+      ],
+    ] as const;
+
+    for (const [viaBulk, viaPrisma] of rows) {
+      await prisma.$transaction((tx) => insertMeasurementRows(tx, [viaBulk]));
+      await prisma.measurement.create({ data: viaPrisma as never });
+    }
+
+    for (const [viaBulk, viaPrisma] of rows) {
+      const a = await prisma.measurement.findUniqueOrThrow({
+        where: { id: viaBulk.id },
+      });
+      const b = await prisma.measurement.findUniqueOrThrow({
+        where: { id: viaPrisma.id },
+      });
+      const strip = (row: typeof a) => ({
+        ...row,
+        id: undefined,
+        externalId: row.externalId?.replace(/-(bulk|prisma)$/, ""),
+        measuredAt: undefined,
+        // Filled with the current time when the row carries none.
+        ...(viaBulk.createdAt
+          ? {}
+          : { createdAt: undefined, updatedAt: undefined }),
+      });
+      expect(strip(a)).toEqual(strip(b));
+      expect(a.measuredAt.toISOString()).toBe(viaBulk.measuredAt.toISOString());
+    }
+    const sparseRow = await prisma.measurement.findUniqueOrThrow({
+      where: { id: "b-bulk" },
+    });
+    expect(Date.now() - sparseRow.createdAt.getTime()).toBeLessThan(60_000);
+  });
+
+  it("inserts a whole batch in one statement and reports the count", async () => {
+    const prisma = getPrismaClient();
+    const batch = Array.from({ length: 2_500 }, (_, i) =>
+      sparse(
+        `m-${i}`,
+        new Date(Date.UTC(2026, 0, 1) + i * 60_000).toISOString(),
+      ),
+    );
+    const inserted = await prisma.$transaction((tx) =>
+      insertMeasurementRows(tx, batch),
+    );
+    expect(inserted).toBe(2_500);
+    expect(await prisma.measurement.count({ where: { userId: USER } })).toBe(
+      2_500,
+    );
+  });
+});

tests/integration/offhost-backup-ledger.test.ts

@@ -228,4 +228,50 @@ describe("off-host backup ledger (real Postgres)", () => {
     await prisma.user.delete({ where: { id: "account-one" } });
     expect(await prisma.offhostBackupState.count()).toBe(0);
   });
+
+  it("a retry of the same run resumes after the accounts it already uploaded (#1031)", async () => {
+    const prisma = getPrismaClient();
+    for (const id of ["acct-a", "acct-b", "acct-c"]) await seedUser(id);
+    const bucket = makeBucket();
+    const puts: string[] = [];
+    const putStream = bucket.putStream;
+    bucket.putStream = async (key, body) => {
+      puts.push(key);
+      return putStream(key, body);
+    };
+    // The job was created a moment ago; its retries share that instant.
+    const runStartedAt = new Date(Date.now() - 60_000);
+    const day = runStartedAt.toISOString().slice(0, 10);
+
+    // First attempt: out of budget after one account.
+    let asked = 0;
+    const first = await runOffhostBackup(prisma, bucket, new Date(), {
+      runStartedAt,
+      shouldStop: () => asked++ >= 1,
+    });
+    expect(first.stoppedEarly).toBe(true);
+    expect(first.uploaded).toBe(1);
+
+    // The retry: same run, so the account already done is skipped.
+    const retry = await runOffhostBackup(prisma, bucket, new Date(), {
+      runStartedAt,
+    });
+    expect(retry.stoppedEarly).toBe(false);
+    expect(retry.alreadyUploaded).toBe(1);
+    expect(retry.uploaded).toBe(2);
+
+    // Each account uploaded exactly once, all under the run's own day.
+    expect(puts.sort()).toEqual([
+      `${day}/user-acct-a.json.enc`,
+      `${day}/user-acct-b.json.enc`,
+      `${day}/user-acct-c.json.enc`,
+    ]);
+
+    // A new run (a later job) uploads everybody again.
+    const next = await runOffhostBackup(prisma, bucket, new Date(), {
+      runStartedAt: new Date(),
+    });
+    expect(next.alreadyUploaded).toBe(0);
+    expect(next.uploaded).toBe(3);
+  });
 });

tests/integration/pr-detection-cumulative-paged.test.ts

@@ -0,0 +1,111 @@
+/**
+ * #1031 — the best cumulative day, found a day at a time.
+ *
+ * `findBestCumulativeDay` read every row of a cumulative type into memory,
+ * every half hour, for every account. On an account holding 784 000 active
+ * energy rows it ran a 524 MB heap out of memory. It now reads a local day at
+ * a time in time order, which gives the same answer because the canonical-
+ * source pick it applies is per day.
+ *
+ * Pinned here at a size that spans several read pages: two sources report
+ * the same days, the source ladder must decide each day on its own, the
+ * best day sits in the middle of the history, and no read of the history is
+ * unbounded.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const { detectPersonalRecordsForUser } =
+  await import("@/lib/personal-records/pr-detection-worker");
+
+const USER = "user-pr-cumulative-paged";
+const DAYS = 40;
+const PER_DAY = 300; // 12 000 Apple rows plus a Withings total per day
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: {
+      id: USER,
+      username: USER,
+      email: `${USER}@example.test`,
+      timezone: "UTC",
+    },
+  });
+});
+
+describe("best cumulative day across many pages", () => {
+  it("sums each day's canonical source and finds the best day without an unbounded read", async () => {
+    const prisma = getPrismaClient();
+    const start = Date.UTC(2025, 0, 1, 6);
+    const rows = [];
+    for (let day = 0; day < DAYS; day++) {
+      // Day 23 is the best Apple day; the Withings total on day 5 is larger
+      // still but loses to Apple on that day by the default ladder, so it
+      // must not win: if the days were pooled across sources it would.
+      const perSample = day === 23 ? 40 : 20;
+      for (let i = 0; i < PER_DAY; i++) {
+        rows.push({
+          userId: USER,
+          type: "ACTIVITY_STEPS" as const,
+          value: perSample,
+          unit: "steps",
+          source: "APPLE_HEALTH" as const,
+          measuredAt: new Date(start + day * 86_400_000 + i * 60_000),
+          externalId: `hk-${day}-${i}`,
+        });
+      }
+      rows.push({
+        userId: USER,
+        type: "ACTIVITY_STEPS" as const,
+        value: day === 5 ? 50_000 : 1_000,
+        unit: "steps",
+        source: "WITHINGS" as const,
+        measuredAt: new Date(start + day * 86_400_000 + 12 * 3_600_000),
+        externalId: `w-${day}`,
+      });
+    }
+    await prisma.measurement.createMany({ data: rows });
+
+    // Every read of the steps history, through a client that records it.
+    const reads: Array<number | undefined> = [];
+    const recording = prisma.$extends({
+      query: {
+        measurement: {
+          findMany({ args, query }) {
+            if (JSON.stringify(args.where ?? {}).includes("ACTIVITY_STEPS")) {
+              reads.push(args.take);
+            }
+            return query(args);
+          },
+        },
+      },
+    });
+
+    await detectPersonalRecordsForUser(USER, {
+      silent: true,
+      prisma: recording as unknown as typeof prisma,
+    });
+
+    const record = await prisma.personalRecord.findFirstOrThrow({
+      where: { userId: USER, metricType: "ACTIVITY_STEPS" },
+    });
+    expect(record.value).toBe(40 * PER_DAY);
+    expect(record.achievedAt.toISOString()).toBe(
+      new Date(start + 23 * 86_400_000 + (PER_DAY - 1) * 60_000).toISOString(),
+    );
+
+    // Several bounded pages, never one read of the whole history.
+    expect(reads.length).toBeGreaterThan(1);
+    for (const take of reads) {
+      expect(take).toBeDefined();
+      expect(take!).toBeLessThanOrEqual(5_000);
+    }
+  });
+});