Merge pull request #1052 from MBombeck/fix/1031-restore-memory

Bound restore memory and order it against the consolidation passes

src/__tests__/restore-lock-consolidation-guard.test.ts

@@ -0,0 +1,75 @@
+/**
+ * Every day transaction of the consolidation family holds the account
+ * against a restore before it touches a reading (`restore-lock.ts`).
+ *
+ * The passes built on `runConsolidation` fold an account a day at a time, and
+ * each day's transaction takes row locks on the account's readings. A restore
+ * deletes all of them in one statement. Without a common first lock the two
+ * take the same rows in different orders and deadlock (#1031, pinned end to
+ * end by `tests/integration/restore-consolidation-deadlock.test.ts` for the
+ * step consolidation). This guard extends the rule to every pass on the same
+ * base, present and future: each file that calls `runConsolidation` must open
+ * every transaction it starts with `holdAccountAgainstRestore(tx, …)`.
+ *
+ * The matcher is whitespace-tolerant and the file and transaction counts are
+ * asserted to be non-zero, so a pattern that matches nothing fails instead of
+ * passing.
+ */
+import { readdirSync, readFileSync, statSync } from "node:fs";
+import { join, relative } from "node:path";
+
+import { describe, expect, it } from "vitest";
+
+const ROOT = join(__dirname, "..", "lib");
+
+function sourceFiles(dir: string): string[] {
+  const out: string[] = [];
+  for (const name of readdirSync(dir)) {
+    const path = join(dir, name);
+    if (statSync(path).isDirectory()) {
+      if (name === "__tests__") continue;
+      out.push(...sourceFiles(path));
+    } else if (/\.ts$/.test(name) && !/\.test\.ts$/.test(name)) {
+      out.push(path);
+    }
+  }
+  return out;
+}
+
+/** The text of a file with its comments blanked, so prose cannot match. */
+function code(path: string): string {
+  return readFileSync(path, "utf8")
+    .replace(/\/\*[\s\S]*?\*\//g, " ")
+    .replace(/(^|[^:])\/\/.*$/gm, "$1");
+}
+
+const passes = sourceFiles(ROOT).filter((path) =>
+  /\bawait\s+runConsolidation\b/.test(code(path)),
+);
+
+describe("consolidation passes hold the account against a restore", () => {
+  it("finds the passes", () => {
+    // Step, mean, per-sample cumulative, dense intraday retention.
+    expect(passes.length).toBeGreaterThanOrEqual(4);
+  });
+
+  it.each(passes.map((path) => [relative(ROOT, path), path]))(
+    "%s opens every transaction with holdAccountAgainstRestore",
+    (_label, path) => {
+      const text = code(path);
+      const opened = [
+        ...text.matchAll(
+          /\$transaction\(\s*async\s*\(\s*(\w+)\s*\)\s*=>\s*\{/g,
+        ),
+      ];
+      expect(opened.length).toBeGreaterThan(0);
+      for (const match of opened) {
+        const tx = match[1];
+        const after = text.slice(match.index! + match[0].length).trimStart();
+        expect(after.startsWith(`await holdAccountAgainstRestore(${tx},`)).toBe(
+          true,
+        );
+      }
+    },
+  );
+});

src/lib/export/measurement-bulk-insert.ts

@@ -1,5 +1,5 @@
 /**
- * Insert many measurements in one statement, as column arrays.
+ * Insert many measurements a hundred rows to a statement, as column arrays.
  *
  * Why not `createMany`. Prisma compiles and caches a query plan per query
  * shape, and a `createMany` of N rows is its own shape: measured, one
@@ -7,8 +7,8 @@
  * heap behind, a 2 000-row one about 290 MB, and the smaller last batch of a
  * restore adds its own plan on top. In a 1 GB container that alone took a
  * restore of 1.25 million measurements past the heap limit (#1031). One
- * `INSERT … SELECT FROM unnest(…)` with an array per column is a single small
- * statement whatever the batch size, and about four times faster.
+ * `INSERT … SELECT FROM unnest(…)` with an array per column is one small
+ * statement shape whatever the number of rows, and about four times faster.
  *
  * The Apple Health export import writes its spot rows through the same
  * statement (`insertNewMeasurementRows`), for the same reason. There the cost
@@ -18,6 +18,19 @@
  * heap and a run of back-to-back flushes reached well over 100 MB. Through
  * this statement the same import holds a few megabytes.
  *
+ * Why no more than {@link MEASUREMENT_INSERT_ROWS_PER_STATEMENT} rows per
+ * statement, whatever the caller hands over. Prisma's debug logger keeps the
+ * arguments of its last hundred calls in memory whether or not logging is
+ * switched on, and both the client and the Postgres adapter log every
+ * statement together with its parameters. So the parameters of about the last
+ * hundred statements stay reachable after they ran, as a fresh copy of every
+ * value. At 1 000 rows a statement that held about 60 MB for the whole of a
+ * restore's measurement write (#1031, measured on 1.8 million Apple Health
+ * readings), next to an app whose heap limit in the default 1 GB container is
+ * 524 MB. At 100 rows it is about 7 MB. The extra statements cost time: on a
+ * fast host the same write took 94 s instead of 76 s, a price paid once per
+ * restore, where the memory was held for all of it.
+ *
  * `COLUMNS` maps every field of `MeasurementInsertRow` to its column and
  * type. It is typed as a complete record over the row's keys, so a field
  * added to the row without a column here does not compile, and the restore
@@ -137,17 +150,42 @@ function cell(value: unknown, kind: Kind): unknown {
   return value;
 }
 
-/** Insert `rows` in one statement. Resolves to the number inserted. */
+/**
+ * Rows one statement carries at most. See the module comment: the parameters
+ * of roughly the last hundred statements stay in memory, so this, not the
+ * caller's batch, is what bounds that memory.
+ */
+export const MEASUREMENT_INSERT_ROWS_PER_STATEMENT = 100;
+
+/** The rows as parameter arrays, one per column, for one statement. */
+function columnArrays(rows: readonly MeasurementInsertRow[]): unknown[][] {
+  return FIELDS.map((field) => {
+    const kind = COLUMNS[field][1];
+    return rows.map((row) => cell(row[field], kind));
+  });
+}
+
+/**
+ * Insert `rows`, {@link MEASUREMENT_INSERT_ROWS_PER_STATEMENT} to a statement.
+ * Resolves to the number inserted.
+ */
 export async function insertMeasurementRows(
   tx: Prisma.TransactionClient,
   rows: readonly MeasurementInsertRow[],
 ): Promise<number> {
-  if (rows.length === 0) return 0;
-  const arrays = FIELDS.map((field) => {
-    const kind = COLUMNS[field][1];
-    return rows.map((row) => cell(row[field], kind));
-  });
-  return tx.$executeRawUnsafe(MEASUREMENT_BULK_INSERT_SQL, ...arrays);
+  let inserted = 0;
+  for (
+    let at = 0;
+    at < rows.length;
+    at += MEASUREMENT_INSERT_ROWS_PER_STATEMENT
+  ) {
+    const slice = rows.slice(at, at + MEASUREMENT_INSERT_ROWS_PER_STATEMENT);
+    inserted += await tx.$executeRawUnsafe(
+      MEASUREMENT_BULK_INSERT_SQL,
+      ...columnArrays(slice),
+    );
+  }
+  return inserted;
 }
 
 /**
@@ -203,7 +241,7 @@ export type NewMeasurementRow = Omit<
   >;
 
 /**
- * Insert brand-new rows in one statement, skipping duplicates, and return the
+ * Insert brand-new rows, skipping duplicates, and return the
  * rows that landed. The replacement for
  * `measurement.createManyAndReturn({ data, skipDuplicates: true })`.
  *
@@ -244,14 +282,21 @@ export async function insertNewMeasurementRows(
     createdAt: now,
     updatedAt: now,
   }));
-  const arrays = FIELDS.map((field) => {
-    const kind = COLUMNS[field][1];
-    return full.map((row) => cell(row[field], kind));
-  });
-  return db.$queryRawUnsafe<InsertedMeasurementRow[]>(
-    MEASUREMENT_BULK_INSERT_SKIP_DUPLICATES_SQL,
-    ...arrays,
-  );
+  const inserted: InsertedMeasurementRow[] = [];
+  for (
+    let at = 0;
+    at < full.length;
+    at += MEASUREMENT_INSERT_ROWS_PER_STATEMENT
+  ) {
+    const slice = full.slice(at, at + MEASUREMENT_INSERT_ROWS_PER_STATEMENT);
+    inserted.push(
+      ...(await db.$queryRawUnsafe<InsertedMeasurementRow[]>(
+        MEASUREMENT_BULK_INSERT_SKIP_DUPLICATES_SQL,
+        ...columnArrays(slice),
+      )),
+    );
+  }
+  return inserted;
 }
 
 // ── Ids ──────────────────────────────────────────────────────────────────

src/lib/export/restore-backup.ts

@@ -38,6 +38,7 @@ import {
   storedBackupRefusal,
   type StoredBackupRef,
 } from "@/lib/export/stored-backup";
+import { takeRestoreLock } from "@/lib/export/restore-lock";
 import {
   insertMeasurementRows,
   type MeasurementInsertRow,
@@ -683,12 +684,6 @@ export async function restoreBackup(
     report("clearing");
     outcome = await prisma.$transaction(
       async (tx) => {
-        // One restore of an account at a time, whatever let a second one
-        // start: the job table admits one queued or running job per account,
-        // and this lock holds even if a worker that was thought gone is
-        // still writing. A second transaction waits here for the first to
-        // commit or roll back, then replaces the account again from its own
-        // file, which is what it was asked to do.
         // Every connection carries the request limits from `src/lib/db.ts`
         // (60 s per statement, 60 s idle inside a transaction). A restore is
         // not a request: deleting two million readings is one statement, and
@@ -697,7 +692,18 @@ export async function restoreBackup(
         // has its own limit; SET LOCAL gives each statement the same one, and
         // ends with the transaction.
         await tx.$executeRaw`SELECT set_config('statement_timeout', ${String(transactionTimeoutMs)}, true), set_config('idle_in_transaction_session_timeout', ${String(transactionTimeoutMs)}, true)`;
-        await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${`backup-restore:${ownerId}`}))`;
+        // One restore of an account at a time, whatever let a second one
+        // start: the job table admits one queued or running job per account,
+        // and this lock holds even if a worker that was thought gone is
+        // still writing. A second transaction waits here for the first to
+        // commit or roll back, then replaces the account again from its own
+        // file, which is what it was asked to do.
+        //
+        // The same lock orders the restore against the background passes that
+        // rewrite readings a day at a time (`restore-lock.ts`): they hold it
+        // shared for each day, so the delete below never meets a day half
+        // folded, which is where the two used to deadlock.
+        await takeRestoreLock(tx, ownerId);
         // Every reference this restore writes stays inside the account: the
         // references that leave it are counted now and again after the last
         // write, and any that grew rolls the whole transaction back

src/lib/export/restore-lock.ts

@@ -0,0 +1,80 @@
+/**
+ * The account lock a restore holds, and the shared side of it for the passes
+ * that rewrite an account's readings in the background.
+ *
+ * A restore deletes every reading of the account in one statement, inside
+ * one long transaction. The consolidation passes (step and mean
+ * consolidation, the per-sample cumulative drain, the dense intraday
+ * retention) fold a day at a time: each day's transaction updates or mints
+ * the day's total and then soft-deletes or deletes the day's per-sample
+ * rows. Both take row locks on the same readings, each in its own order, and
+ * when the restore reached a day's samples before its total while a pass was
+ * folding that day, each held what the other needed next. Postgres broke the
+ * cycle by cancelling one of them (`40P01`, #1031), and on a freshly booted
+ * instance, where the step consolidation runs over every account with legacy
+ * rows, that was the restore.
+ *
+ * So the two are ordered by one advisory lock before either touches a row.
+ * The restore takes it exclusively as the first act of its transaction
+ * ({@link takeRestoreLock}); every day transaction of those passes takes it in
+ * shared mode as its first act ({@link holdAccountAgainstRestore}). Passes
+ * share the lock with each other, so nothing about how they run together
+ * changes. A restore that starts while a day is being folded waits for that
+ * one day to commit, milliseconds. A day that would start while a restore is
+ * running does not wait for it, which could take many minutes: it is refused
+ * with {@link AccountRestoreInProgressError}, and the pass leaves the account
+ * for its next run, which reads the restored rows rather than the ones it had
+ * scanned before the restore replaced them.
+ */
+import type { Prisma } from "@/generated/prisma/client";
+
+type LockClient = Pick<Prisma.TransactionClient, "$queryRaw">;
+
+/** The lock's key. One per account; the restore has always used this one. */
+function restoreLockKey(userId: string): string {
+  return `backup-restore:${userId}`;
+}
+
+/**
+ * Take the account's restore lock exclusively, for the rest of the calling
+ * transaction. Waits for any day transaction of a background pass to commit,
+ * and for another restore of the same account to finish.
+ */
+export async function takeRestoreLock(
+  tx: LockClient,
+  userId: string,
+): Promise<void> {
+  // `pg_advisory_xact_lock` returns void, which the client cannot read as a
+  // column; selecting from it yields a plain row.
+  await tx.$queryRaw`
+    SELECT 1 AS locked
+    FROM pg_advisory_xact_lock(hashtext(${restoreLockKey(userId)}))
+  `;
+}
+
+/** A restore of the account is running; the pass should leave it for now. */
+export class AccountRestoreInProgressError extends Error {
+  readonly userId: string;
+  constructor(userId: string) {
+    super("A restore of this account is running");
+    this.name = "AccountRestoreInProgressError";
+    this.userId = userId;
+  }
+}
+
+/**
+ * Hold the account against a restore for the rest of the calling
+ * transaction: the shared side of {@link takeRestoreLock}. Call it before the
+ * transaction's first read or write of the account's readings. Throws
+ * {@link AccountRestoreInProgressError} at once when a restore holds the
+ * lock, rather than waiting for it.
+ */
+export async function holdAccountAgainstRestore(
+  tx: LockClient,
+  userId: string,
+): Promise<void> {
+  const [row] = await tx.$queryRaw<Array<{ held: boolean }>>`
+    SELECT pg_try_advisory_xact_lock_shared(hashtext(${restoreLockKey(userId)})) AS held
+  `;
+  if (!row?.held) throw new AccountRestoreInProgressError(userId);
+}

src/lib/measurements/__tests__/consolidate-daily-mean.test.ts

@@ -16,6 +16,14 @@ import { canonicalDailyTimestamp } from "../consolidation-tz";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 function row(
   id: string,
   value: number,

src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts

@@ -35,6 +35,14 @@ import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 function pulseRow(id: string, value: number, iso: string): PerSampleRow {
   return {
     id,

src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts

@@ -34,6 +34,14 @@ import {
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 /** A day well outside the retention window, so the fold definitely runs. */
 const FOLD_DAY = new Date(
   Date.now() - (DENSE_INTRADAY_RETENTION_DAYS + 10) * 24 * 60 * 60 * 1000,

src/lib/measurements/__tests__/dense-intraday-retention.test.ts

@@ -27,6 +27,14 @@ import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 function row(
   id: string,
   value: number,

src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts

@@ -31,6 +31,14 @@ import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
 import { candidateLookup, createManyVia } from "./hourly-mint-mock";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 function spo2Row(id: string, value: number, iso: string): PerSampleRow {
   return {
     id,

src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts

@@ -22,6 +22,14 @@ import {
 } from "@/lib/tz/local-day";
 import type { PrismaClient } from "@/generated/prisma/client";
 
+// The restore lock is a Postgres advisory lock; the mocked client has no
+// database behind it. The lock itself is pinned against a real one in
+// `tests/integration/restore-consolidation-deadlock.test.ts`.
+vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
+  holdAccountAgainstRestore: vi.fn(async () => {}),
+}));
+
 describe("dayKeyForUserTz", () => {
   it("anchors the calendar day to the user's IANA zone", () => {
     // 23:45 NZST on 2026-05-16 is still 11:45 UTC on the same day, but

src/lib/measurements/consolidate-daily-mean.ts

@@ -45,6 +45,7 @@
  * production standalone image strips `tsx`. Modelled on the
  * `step-consolidation` boot-time converging-backfill pattern.
  */
+import { holdAccountAgainstRestore } from "@/lib/export/restore-lock";
 import type {
   MeasurementType,
   Prisma,
@@ -273,6 +274,8 @@ export async function consolidateDailyMean(
       const unit = dayRows[0]?.unit ?? "unknown";
       let removed = 0;
       await pc.$transaction(async (tx) => {
+        // First, before any reading is touched: see `restore-lock.ts`.
+        await holdAccountAgainstRestore(tx, userId);
         // The upsert below arbiters on (userId, type, source, externalId),
         // but the Measurement model carries a SECOND full unique —
         // (userId, type, measuredAt, source, sleepStage), NULLS NOT

src/lib/measurements/consolidate-legacy-steps.ts

@@ -44,6 +44,7 @@
  * production standalone image strips `tsx`. Modelled on the
  * `rollup-full-backfill` boot-time converging-backfill pattern.
  */
+import { holdAccountAgainstRestore } from "@/lib/export/restore-lock";
 import type {
   MeasurementSource,
   PrismaClient,
@@ -318,6 +319,8 @@ export async function consolidateLegacySteps(
       try {
         let removed = 0;
         await pc.$transaction(async (tx) => {
+          // First, before any reading is touched: see `restore-lock.ts`.
+          await holdAccountAgainstRestore(tx, userId);
           if (shouldMint) {
             // Mint the canonical daily-total row. The unique index
             // (userId, type, source, externalId) makes the upsert

src/lib/measurements/consolidation-base.ts

@@ -26,6 +26,8 @@
  * is no value-level cycle between the two — a cycle would otherwise trip
  * a module-init TDZ in the production bundle.
  */
+import { annotate } from "@/lib/logging/context";
+import { AccountRestoreInProgressError } from "@/lib/export/restore-lock";
 import type {
   MeasurementType,
   PrismaClient,
@@ -425,6 +427,8 @@ export async function runConsolidation<TType extends MeasurementType>(
   dryRun: boolean;
   daysFailed: number;
   stoppedEarly: boolean;
+  /** Accounts left for the next run because a restore of them was running. */
+  usersDeferredForRestore: number;
 }> {
   const { prismaClient, options } = params;
   const dryRun = options.dryRun ?? false;
@@ -434,6 +438,7 @@ export async function runConsolidation<TType extends MeasurementType>(
   let daysFailed = 0;
   let daysWalked = 0;
   let stoppedEarly = false;
+  let usersDeferredForRestore = 0;
 
   const users = await loadConsolidationUsers(prismaClient, options.userId);
 
@@ -533,6 +538,20 @@ export async function runConsolidation<TType extends MeasurementType>(
             outcome,
           });
         } catch (err) {
+          // A restore of this account started while the pass was walking
+          // it. The rows scanned so far are about to be replaced, so the
+          // account is left for the next run, which reads the restored ones
+          // (`restore-lock.ts`). Not a failed day, and no reason to stop the
+          // other accounts.
+          if (err instanceof AccountRestoreInProgressError) {
+            usersDeferredForRestore += 1;
+            annotate({
+              meta: {
+                consolidation_deferred_for_restore: usersDeferredForRestore,
+              },
+            });
+            continue walk;
+          }
           // No boundary supplied → preserve the historical abort-the-run
           // behaviour for the drains that classify errors in `writeDay`.
           if (!params.onBucketError) throw err;
@@ -561,5 +580,11 @@ export async function runConsolidation<TType extends MeasurementType>(
     params.onUserComplete?.({ userId: user.id, tz, dryRun });
   }
 
-  return { usersScanned: users.length, dryRun, daysFailed, stoppedEarly };
+  return {
+    usersScanned: users.length,
+    dryRun,
+    daysFailed,
+    stoppedEarly,
+    usersDeferredForRestore,
+  };
 }

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

@@ -47,6 +47,7 @@
  * The one-shot history rebuild that converts ALREADY-folded daily rows to
  * the hourly grain lives in `dense-intraday-hourly-rebuild.ts`.
  */
+import { holdAccountAgainstRestore } from "@/lib/export/restore-lock";
 import type {
   MeasurementType,
   Prisma,
@@ -617,6 +618,8 @@ export async function runDenseIntradayRetention(
         let removed = 0;
         let retiredDaily = false;
         await pc.$transaction(async (tx) => {
+          // First, before any reading is touched: see `restore-lock.ts`.
+          await holdAccountAgainstRestore(tx, userId);
           const canonicalRowIds = await adoptOrMintHourlyRows(tx, {
             userId,
             type,

src/lib/measurements/drain-per-sample-cumulative.ts

@@ -34,6 +34,7 @@
  * touching the DB so the operator can inspect what would change before
  * committing.
  */
+import { holdAccountAgainstRestore } from "@/lib/export/restore-lock";
 import { Prisma } from "@/generated/prisma/client";
 import type {
   MeasurementAggregationProvenance,
@@ -291,6 +292,8 @@ export async function drainPerSampleCumulative(
       const foldOnce = async (): Promise<number> => {
         let removed = 0;
         await pc.$transaction(async (tx) => {
+          // First, before any reading is touched: see `restore-lock.ts`.
+          await holdAccountAgainstRestore(tx, userId);
           // Single-flight the per-(user, type, day) fold. Two overlapping
           // drains — a rolling-deploy nightly overlap, or an operator's manual
           // `POST /api/admin/drain-per-sample-cumulative` racing the cron —

tests/integration/backup-restore-memory.test.ts

@@ -0,0 +1,179 @@
+/**
+ * #1031 — what a restore holds while it writes the measurements, measured.
+ *
+ * A self-hoster restored 1.8 million readings in the default 1 GB container
+ * (V8 heap limit 524 MB). Near the end of the measurement write the server
+ * restarted, and the job started over from the beginning. Reproduced against
+ * the same restore of 1.8 million Apple Health readings, the restore itself
+ * held a steady 60 MB more than the process did before it started, from the
+ * hundredth statement to the last: not the rows it was writing, but the
+ * parameters of the hundred statements before them.
+ *
+ * Prisma's debug logger keeps the arguments of its last hundred calls whether
+ * or not logging is on, and the client and the Postgres adapter log each
+ * statement with its parameters. A restore that wrote 1 000 readings per
+ * statement therefore kept about the last 100 000 readings alive as copies.
+ * Bounded, and the same at 150 000 readings as at 1.8 million, but 60 MB out
+ * of a heap that a long-running server already fills well past half. The
+ * insert now carries at most a hundred rows per statement, and the history
+ * holds about a tenth of that.
+ *
+ * How it measures. Every reading is taken after a forced collection, so what
+ * is compared is what the restore HOLDS, not what V8 left lying about. The
+ * anchor is taken when the measurement write starts, after the checks and the
+ * clearing, so the modules and compiled queries they load are not counted.
+ * The fixture is sized so the old statement size fills the whole history
+ * well before the end: 150 statements of 1 000 readings.
+ *
+ * Mutation check: set `MEASUREMENT_INSERT_ROWS_PER_STATEMENT` to 1 000 and
+ * the growth test goes red at about 64 MB against the 20 MB budget.
+ */
+import v8 from "node:v8";
+import vm from "node:vm";
+
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const OWNER = "restore-memory-owner";
+const ROWS = 150_000;
+
+/**
+ * How much the heap may grow during the measurement write. Measured, not
+ * chosen: about 8 MB with a hundred rows per statement, about 64 MB with a
+ * thousand, on this fixture.
+ */
+const GROWTH_BUDGET_BYTES = 20 * 1024 * 1024;
+
+const forceGc = ((): (() => void) => {
+  v8.setFlagsFromString("--expose-gc");
+  const gc = vm.runInNewContext("gc") as () => void;
+  v8.setFlagsFromString("--no-expose-gc");
+  return gc;
+})();
+
+function liveHeapBytes(): number {
+  forceGc();
+  forceGc();
+  return process.memoryUsage().heapUsed;
+}
+
+const mb = (bytes: number): string => (bytes / 1024 / 1024).toFixed(1);
+
+let backupId = "";
+
+beforeAll(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: { id: OWNER, username: OWNER, role: "ADMIN" },
+  });
+  const { encryptNote } = await import("@/lib/crypto/note-cipher");
+  const note = Buffer.from(encryptNote("a note kept with a reading")!);
+  // Apple Health's shape: an external id and a source version on every row,
+  // ciphertext on one in twenty-five, a note on one in forty, and a recently
+  // deleted row on one in two hundred. One statement, so the fixture costs
+  // seconds rather than dominating the file.
+  await prisma.$executeRawUnsafe(
+    `INSERT INTO measurements (
+       id, user_id, type, value, unit, source, measured_at, notes,
+       notes_encrypted, external_id, external_source_version, device_type,
+       created_at, updated_at, sync_version, deleted_at)
+     SELECT
+       'c' || lpad(to_hex(g), 24, '0'),
+       $1,
+       (ARRAY['PULSE', 'ACTIVITY_STEPS', 'ACTIVE_ENERGY_BURNED',
+              'WALKING_RUNNING_DISTANCE'])[1 + (g % 4)]::measurement_type,
+       60 + (g % 40),
+       'count',
+       'APPLE_HEALTH'::measurement_source,
+       timestamp '2019-01-01 00:00:00' + (g * interval '47 seconds'),
+       CASE WHEN g % 40 = 0 THEN 'a note recorded with reading ' || g END,
+       CASE WHEN g % 25 = 0 THEN $2::bytea END,
+       '5B3E1C2A-' || lpad(to_hex(g), 8, '0') || '-4A1F-9C3D-8E7F6A5B4C3D',
+       '17.4',
+       'Apple Watch',
+       timestamp '2019-01-01 00:00:00' + (g * interval '47 seconds'),
+       timestamp '2019-01-01 00:00:00' + (g * interval '47 seconds'),
+       1,
+       CASE WHEN g % 200 = 0 THEN now() - interval '1 day' END
+     FROM generate_series(1, ${ROWS}) AS g`,
+    OWNER,
+    note,
+  );
+  const { storeBackupBlob } = await import("@/lib/export/store-backup-blob");
+  const { streamFullBackupJson } =
+    await import("@/lib/export/full-backup-stream");
+  const stored = await storeBackupBlob(
+    prisma,
+    { userId: OWNER, type: "WEEKLY_AUTO" },
+    (write) =>
+      streamFullBackupJson(prisma, OWNER, write, {
+        purpose: "disaster-recovery",
+      }),
+  );
+  backupId = stored.id;
+}, 240_000);
+
+afterAll(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+describe("backup restore under a memory budget (#1031)", () => {
+  it("holds a bounded amount while it writes the measurements", async () => {
+    const prisma = getPrismaClient();
+    const { restoreBackup } = await import("@/lib/export/restore-backup");
+    const { STORED_BACKUP_SELECT } = await import("@/lib/export/stored-backup");
+    const backup = await prisma.dataBackup.findUniqueOrThrow({
+      where: { id: backupId },
+      select: STORED_BACKUP_SELECT,
+    });
+
+    let anchor: number | null = null;
+    let peak = 0;
+    let lastWritten = 0;
+    let samples = 0;
+    const outcome = await restoreBackup({
+      backup,
+      actorUserId: OWNER,
+      ipAddress: null,
+      restoreInstanceSettings: false,
+      // Called from inside the write, between two batches, which is where
+      // anything the restore holds across batches is visible.
+      progress: (phase, progress) => {
+        if (phase !== "measurements") return;
+        lastWritten = progress.measurementsWritten;
+        if (anchor === null) {
+          anchor = peak = liveHeapBytes();
+          return;
+        }
+        samples++;
+        peak = Math.max(peak, liveHeapBytes());
+      },
+    });
+
+    expect(outcome.ok).toBe(true);
+    expect(await prisma.measurement.count({ where: { userId: OWNER } })).toBe(
+      ROWS,
+    );
+    // Not vacuous: the write was sampled, and sampled past the point where
+    // the old statement size had filled the history.
+    expect(anchor).not.toBeNull();
+    expect(samples).toBeGreaterThan(5);
+    expect(lastWritten).toBeGreaterThanOrEqual(120_000);
+
+    const growth = peak - (anchor ?? 0);
+    process.stderr.write(
+      `[restore-memory] ${ROWS} readings: the heap grew by ${mb(growth)} MB ` +
+        `during the measurement write (${samples} samples)\n`,
+    );
+    expect(
+      growth,
+      `the restore held ${mb(growth)} MB more while writing the measurements`,
+    ).toBeLessThan(GROWTH_BUDGET_BYTES);
+  }, 300_000);
+});

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

@@ -11,6 +11,7 @@ import { getPrismaClient, truncateAllTables } from "./setup";
 import {
   insertMeasurementRows,
   insertNewMeasurementRows,
+  MEASUREMENT_INSERT_ROWS_PER_STATEMENT,
   newMeasurementId,
   type MeasurementInsertRow,
   type NewMeasurementRow,
@@ -123,7 +124,8 @@ describe("insertMeasurementRows", () => {
     expect(Date.now() - sparseRow.createdAt.getTime()).toBeLessThan(60_000);
   });
 
-  it("inserts a whole batch in one statement and reports the count", async () => {
+  it("splits a batch across statements and reports the whole count", async () => {
+    // 2 500 rows is 25 statements of MEASUREMENT_INSERT_ROWS_PER_STATEMENT.
     const prisma = getPrismaClient();
     const batch = Array.from({ length: 2_500 }, (_, i) =>
       sparse(
@@ -272,6 +274,24 @@ describe("insertNewMeasurementRows", () => {
     expect(viaRaw.returned).toHaveLength(1);
   });
 
+  it("skips a duplicate that falls in a later statement than its twin", async () => {
+    // More rows than one statement carries, with the twin of row 20 at row
+    // 150: the two land in different statements, and the second has to be
+    // skipped exactly as Prisma skips it within one.
+    const rows = Array.from({ length: 250 }, (_, i) =>
+      spot(
+        i === 150 ? "hr-20" : `hr-${i}`,
+        new Date(Date.UTC(2026, 2, 5) + i * 60_000).toISOString(),
+        { value: 50 + (i % 40) },
+      ),
+    );
+    expect(rows.length).toBeGreaterThan(MEASUREMENT_INSERT_ROWS_PER_STATEMENT);
+    const { viaPrisma, viaRaw } = await both([], rows);
+    expect(viaRaw.returned).toEqual(viaPrisma.returned);
+    expect(viaRaw.table).toEqual(viaPrisma.table);
+    expect(viaRaw.returned).toHaveLength(249);
+  });
+
   it("mints ids in the shape the cuid() default produces", async () => {
     const prisma = getPrismaClient();
     await prisma.measurement.deleteMany({ where: { userId: USER } });

tests/integration/restore-consolidation-deadlock.test.ts

@@ -0,0 +1,277 @@
+/**
+ * #1031 — a restore and a consolidation pass over the same account must not
+ * deadlock.
+ *
+ * Seen on a freshly booted instance: the restore's clearing step failed with
+ * `40P01` while the boot-time step consolidation was folding the same
+ * account. Each takes row locks on the account's readings in its own order.
+ * A consolidation day first updates (or mints) the day's `stats:` total, then
+ * soft-deletes the day's per-sample rows; the restore deletes every reading
+ * of the account in one statement, in the order the table gives them. When
+ * the restore reaches the day's samples before the total, each side holds
+ * what the other one needs next.
+ *
+ * This file builds exactly that order, deterministically. The day's samples
+ * are written before its (soft-deleted) total, so the restore's delete meets
+ * them first, and a row trigger holds the consolidation for three seconds
+ * right after it has locked the total. The restore is started while it waits.
+ *
+ * What must hold: both finish, neither is chosen as a deadlock victim, and
+ * the account ends up as the backup says. Every writer of the consolidation
+ * family now takes the restore's account lock in shared mode at the start of
+ * its day transaction, and the restore takes it exclusively before its first
+ * delete, so the two are ordered by that lock before either touches a row. A
+ * consolidation that finds a restore running leaves the account for its next
+ * run instead of waiting on it.
+ *
+ * Mutation check: remove the `holdAccountAgainstRestore` call from the step
+ * consolidation's day transaction and the first test fails with the
+ * consolidation rejected on `40P01`.
+ */
+import {
+  afterAll,
+  afterEach,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import { acceptingBoss } from "./restore-job-driver";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const OWNER = "restore-deadlock-owner";
+const DAY = "2026-01-10";
+const TOTAL_ID = "deadlock-total";
+const TOTAL_EXTERNAL_ID = `stats:HKQuantityTypeIdentifierStepCount:${DAY}`;
+
+async function dropTrigger() {
+  const prisma = getPrismaClient();
+  await prisma.$executeRawUnsafe(
+    `DROP TRIGGER IF EXISTS restore_deadlock_hold ON measurements`,
+  );
+  await prisma.$executeRawUnsafe(
+    `DROP FUNCTION IF EXISTS restore_deadlock_hold()`,
+  );
+}
+
+/** Hold the transaction that updates the day's total for three seconds. */
+async function holdTheTotalsUpdate() {
+  const prisma = getPrismaClient();
+  await prisma.$executeRawUnsafe(
+    `CREATE FUNCTION restore_deadlock_hold() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(3); RETURN NEW; END $$`,
+  );
+  await prisma.$executeRawUnsafe(
+    `CREATE TRIGGER restore_deadlock_hold BEFORE UPDATE ON measurements FOR EACH ROW WHEN (OLD.id = '${TOTAL_ID}') EXECUTE FUNCTION restore_deadlock_hold()`,
+  );
+}
+
+async function seedAccountWithBackup(): Promise<string> {
+  const prisma = getPrismaClient();
+  await prisma.user.create({
+    data: { id: OWNER, username: OWNER, role: "ADMIN", timezone: "UTC" },
+  });
+  // The day's per-sample rows first, then its total, so a delete of the
+  // account meets the samples first. All samples before noon, so an index
+  // walk in time order meets them first too.
+  for (const [i, hour] of [6, 8, 10].entries()) {
+    await prisma.measurement.create({
+      data: {
+        id: `deadlock-sample-${i}`,
+        userId: OWNER,
+        type: "ACTIVITY_STEPS",
+        value: 1000 + i,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date(
+          `${DAY}T${String(hour).padStart(2, "0")}:00:00.000Z`,
+        ),
+        externalId: `hk-sample-${i}`,
+      },
+    });
+  }
+  // A total a previous pass minted and something since removed: the next
+  // pass mints it again through the upsert's update branch, which locks it.
+  await prisma.measurement.create({
+    data: {
+      id: TOTAL_ID,
+      userId: OWNER,
+      type: "ACTIVITY_STEPS",
+      value: 1,
+      unit: "steps",
+      source: "MANUAL",
+      measuredAt: new Date(`${DAY}T12:00:00.000Z`),
+      externalId: TOTAL_EXTERNAL_ID,
+      deletedAt: new Date(),
+    },
+  });
+  const { storeBackupBlob } = await import("@/lib/export/store-backup-blob");
+  const { streamFullBackupJson } =
+    await import("@/lib/export/full-backup-stream");
+  const { id } = await storeBackupBlob(
+    prisma,
+    { userId: OWNER, type: "WEEKLY_AUTO" },
+    (write) =>
+      streamFullBackupJson(prisma, OWNER, write, {
+        purpose: "disaster-recovery",
+      }),
+  );
+  return id;
+}
+
+async function startRestore(backupId: string) {
+  const prisma = getPrismaClient();
+  const jobs = await import("@/lib/jobs/backup-restore");
+  const { storedBackupIdentity, STORED_BACKUP_SELECT } =
+    await import("@/lib/export/stored-backup");
+  const backup = await prisma.dataBackup.findUniqueOrThrow({
+    where: { id: backupId },
+    select: STORED_BACKUP_SELECT,
+  });
+  const admitted = await jobs.admitBackupRestore({
+    userId: OWNER,
+    actorUserId: OWNER,
+    backupId,
+    backupDigest: jobs.backupDigest(storedBackupIdentity(backup)),
+    restoreInstanceSettings: false,
+    boss: acceptingBoss,
+  });
+  if (!admitted.admitted) throw new Error("restore not admitted");
+  return {
+    jobId: admitted.jobId,
+    done: jobs.runBackupRestoreJob(admitted.jobId),
+  };
+}
+
+/** Wait until a backend sits in the trigger's sleep. */
+async function untilHeldInTrigger() {
+  const prisma = getPrismaClient();
+  for (let i = 0; i < 100; i++) {
+    const rows = await prisma.$queryRaw<Array<{ n: bigint }>>`
+      SELECT count(*) AS n FROM pg_stat_activity WHERE wait_event = 'PgSleep'
+    `;
+    if (Number(rows[0]?.n ?? 0) > 0) return;
+    await new Promise((resolve) => setTimeout(resolve, 50));
+  }
+  throw new Error("the consolidation never reached the day's total");
+}
+
+beforeEach(async () => {
+  await dropTrigger();
+  await truncateAllTables(getPrismaClient());
+});
+
+afterEach(async () => {
+  await dropTrigger();
+});
+
+afterAll(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+describe("restore against a running consolidation (#1031)", () => {
+  it("does not deadlock when the restore starts while a day is being folded", async () => {
+    const prisma = getPrismaClient();
+    const backupId = await seedAccountWithBackup();
+    await holdTheTotalsUpdate();
+
+    const { runStepConsolidationForUser } =
+      await import("@/lib/jobs/step-consolidation");
+    const consolidation = runStepConsolidationForUser(OWNER).then(
+      (value) => ({ ok: true as const, value }),
+      (error: unknown) => ({ ok: false as const, error }),
+    );
+    await untilHeldInTrigger();
+    const restore = await startRestore(backupId);
+
+    const [folded] = await Promise.all([consolidation, restore.done]);
+    const { readBackupRestoreJob } = await import("@/lib/jobs/backup-restore");
+    const job = await readBackupRestoreJob(restore.jobId);
+
+    expect(
+      folded.ok,
+      `the consolidation failed: ${folded.ok ? "" : String((folded.error as Error)?.message ?? folded.error)}`,
+    ).toBe(true);
+    expect(job?.status, JSON.stringify(job?.failure)).toBe("succeeded");
+    // The day was folded before the restore went ahead, and the restore then
+    // put the account back exactly as the backup holds it.
+    if (folded.ok) expect(folded.value.daysConsolidated).toBe(1);
+    const rows = await prisma.measurement.findMany({
+      where: { userId: OWNER },
+      orderBy: { id: "asc" },
+      select: { id: true, deletedAt: true },
+    });
+    expect(rows.map((row) => [row.id, row.deletedAt === null])).toEqual([
+      ["deadlock-sample-0", true],
+      ["deadlock-sample-1", true],
+      ["deadlock-sample-2", true],
+      [TOTAL_ID, false],
+    ]);
+  }, 60_000);
+
+  it("leaves an account alone while its restore holds the lock", async () => {
+    const prisma = getPrismaClient();
+    await seedAccountWithBackup();
+    const { holdAccountAgainstRestore, takeRestoreLock } =
+      await import("@/lib/export/restore-lock");
+
+    // A restore's transaction, standing in for the real one: the exclusive
+    // account lock is all a consolidation can see of it.
+    let release!: () => void;
+    const released = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    let locked!: () => void;
+    const isLocked = new Promise<void>((resolve) => {
+      locked = resolve;
+    });
+    const restoreTx = prisma.$transaction(
+      async (tx) => {
+        await takeRestoreLock(tx, OWNER);
+        locked();
+        await released;
+      },
+      { timeout: 30_000 },
+    );
+    await isLocked;
+
+    try {
+      // Refused at once, rather than waiting on the restore.
+      const started = Date.now();
+      await expect(
+        prisma.$transaction((tx) => holdAccountAgainstRestore(tx, OWNER)),
+      ).rejects.toMatchObject({ name: "AccountRestoreInProgressError" });
+      expect(Date.now() - started).toBeLessThan(2_000);
+
+      // The pass leaves the account's rows as they are and says so.
+      const { consolidateLegacySteps } =
+        await import("@/lib/measurements/consolidate-legacy-steps");
+      const summary = await consolidateLegacySteps(prisma, {
+        userId: OWNER,
+        log: () => {},
+      });
+      expect(summary.totals.daysConsolidated).toBe(0);
+      expect(
+        await prisma.measurement.count({
+          where: { userId: OWNER, deletedAt: null },
+        }),
+      ).toBe(3);
+    } finally {
+      release();
+      await restoreTx;
+    }
+
+    // Two passes of the family can still share the account.
+    await prisma.$transaction(async (tx) => {
+      await holdAccountAgainstRestore(tx, OWNER);
+      await prisma.$transaction((other) =>
+        holdAccountAgainstRestore(other, OWNER),
+      );
+    });
+  }, 60_000);
+});