diff --git a/src/lib/export/measurement-bulk-insert.ts b/src/lib/export/measurement-bulk-insert.ts index 4d0e7c21f..e056817a0 100644 --- a/src/lib/export/measurement-bulk-insert.ts +++ b/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 { - 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( - 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( + MEASUREMENT_BULK_INSERT_SKIP_DUPLICATES_SQL, + ...columnArrays(slice), + )), + ); + } + return inserted; } // ── Ids ────────────────────────────────────────────────────────────────── diff --git a/tests/integration/backup-restore-memory.test.ts b/tests/integration/backup-restore-memory.test.ts new file mode 100644 index 000000000..a3d00e570 --- /dev/null +++ b/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); +}); diff --git a/tests/integration/measurement-bulk-insert.test.ts b/tests/integration/measurement-bulk-insert.test.ts index 81379639e..2d3116fb5 100644 --- a/tests/integration/measurement-bulk-insert.test.ts +++ b/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 } });