fix(restore): carry at most a hundred readings per insert statement

Prisma's debug logger keeps the arguments of its last hundred calls in
memory whether or not logging is on, and the client and the Postgres
adapter both log every statement with its parameters. A restore that
wrote 1 000 readings per statement therefore kept a copy of about the
last 100 000 readings alive for the whole measurement write: about
60 MB on 1.8 million Apple Health readings, next to a V8 heap limit of
524 MB in the default 1 GB container.

The bulk insert now splits whatever it is handed into statements of at
most 100 rows, which holds about 7 MB. The Apple Health export import
goes through the same helper and gets the same bound. The write costs a
little time: 94 s instead of 76 s for 1.8 million readings on a fast
host.

Refs #1031

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 ──────────────────────────────────────────────────────────────────

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 } });