diff --git a/CHANGELOG.md b/CHANGELOG.md index 24ea46653..a4e13ca4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -127,6 +127,25 @@ the element sat on the parser's list of things to skip. The parser reads it now and writes it the moment it goes past, so a run that later fails still says which export it was working from. +- **The weekly backup no longer takes the instance down with it.** The + compression that went out last release was real, and it was measured in the + wrong place. On a development machine with a multi-gigabyte heap the pass + finished; the container it actually runs in is capped at 1 GB, which gives + Node a 524 MB heap, and there the same pass died with + `Reached heap limit — JavaScript heap out of memory` about fifteen seconds + in. Because the job shares the application process, that was not one failed + backup: it was a restart, and every signed-in session on the host went with + it. The document is now written incrementally. The three tables that grow + without bound — readings, doses, mood entries — are read a page at a time and + serialised straight into the compressor and the cipher, every other section + is released as soon as its own JSON exists, and nothing between the database + rows and the stored value is ever held whole. On a seeded account of 445 000 + readings the pass used to exhaust a 546 MB heap; the same account doubled to + 890 000 now finishes inside 296 MB. The writer also watches its own memory + and stops at 80 percent of the limit, so an account too large for its host is + a backup that failed for that one account and is counted in the run's meta, + not a restart for everybody. Backups written under either older shape still + restore, which is checked both ways rather than asserted. ### Internal diff --git a/docs/ops/backup-restore.md b/docs/ops/backup-restore.md index b465f2b5d..b2d060a80 100644 --- a/docs/ops/backup-restore.md +++ b/docs/ops/backup-restore.md @@ -122,6 +122,59 @@ The full per-model reasoning lives in `src/lib/export/backup-plan.ts`, where every excluded model carries a written verdict and a structural test refuses to let a new model land without one. +## The weekly in-database backup (`data-backup`) + +Separate from the off-host job above, and easy to confuse with it. A second +pg-boss job writes one `WEEKLY_AUTO` row per user into `data_backups.data` — +the same JSON document, gzipped and then encrypted under `ENCRYPTION_KEY` / +`ENCRYPTION_KEYS`, staying inside the instance. It is what +`/api/admin/backups//restore` reads. + +### Container memory + +This is the part that bites. The job runs inside the app process, so V8's heap +limit is the app's heap limit, and a container capped at 1 GB gives Node a +524 MB old-space limit by default. A long-lived record is bigger than it looks: +several hundred thousand measurements serialise to a JSON document of a few +hundred megabytes, and the writer used to need the object graph and that +document resident at the same time. On a seeded account of 445 000 measurements +under a 546 MB limit that is `FATAL ERROR: Reached heap limit` about thirty +seconds in — and since the job shares the process, the whole instance restarted +and every signed-in session on it went with it. + +Two things changed, and both matter to an operator: + +- **The writer streams.** The three tables that grow without bound — + measurements, intake events, mood entries — are read a page at a time and + serialised straight into gzip and the cipher, and every other section is + released as soon as its JSON exists. On the same fixture doubled to 890 000 + measurements, the pass completes inside a 296 MB heap limit; before the + change, half that record did not fit in 546 MB. +- **It stops itself.** The writer watches its own live heap and aborts at 80 % + of V8's limit. That backup then fails for that one account, is counted in the + run's `users_failed` and `heap_budget_trips` meta, and the pass carries on + with everybody else. A memory failure is now a failed job rather than a + restart for every user on the host. + +If `heap_budget_trips` is non-zero in `job.data_backup`, the answer is more +memory, not a retry: raise the container's limit, or set +`NODE_OPTIONS=--max-old-space-size=` to something under it. + +### The stored column is the remaining ceiling + +`data_backups.data` is a single `text` column, so the finished artifact has to +exist as one value before it can be written — that copy cannot be streamed +away. It is the only thing left in the job that grows with the record: at a +compressed blob of ~42 MB the pass peaks around 236 MB, and the growth from +there is roughly twice the blob's size (the base64 answer, plus the driver's +copy of it on the way to the wire). A blob past roughly 150 MB — an account +several times larger than any seen so far — will hit a 524 MB limit again, and +the fix at that point is to stop storing the artifact in one column: chunk it +across rows, or keep only the off-host copy. Reading it back has the same +shape, and worse: a restore parses the whole document, so the read path needs +several times the blob in heap. An operator restoring a very large account +should give the container more memory for the duration. + ## Monthly restore drill (automatic) Since v1.16.4 a pg-boss job (`data-restore-drill`, cron `11 4 1 * *` — diff --git a/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts b/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts index d824dd30b..5d52faa1a 100644 --- a/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts +++ b/src/app/api/admin/backups/[id]/restore/__tests__/owner-mismatch.test.ts @@ -36,7 +36,12 @@ vi.mock("@/lib/db", () => ({ toJson: (value: T) => value, })); -vi.mock("@/lib/crypto", () => ({ decrypt: vi.fn(() => "{}") })); +vi.mock("@/lib/crypto", () => ({ + decrypt: vi.fn(() => "{}"), + // The envelope reader asks which stored shape it was handed first. + isStreamCiphertext: () => false, + decryptStream: vi.fn(), +})); vi.mock("@/lib/crypto/note-cipher", () => ({ encryptNote: vi.fn() })); vi.mock("@/lib/validations/backup", () => ({ parseBackupPayload: vi.fn(), diff --git a/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts b/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts index d06486115..51441643c 100644 --- a/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts +++ b/src/app/api/admin/backups/[id]/summary/__tests__/route.test.ts @@ -30,8 +30,12 @@ vi.mock("@/lib/db", () => ({ })); const decryptMock = vi.fn(); +// The envelope reader asks which of the stored shapes it was handed before it +// decrypts anything, so a stub of this module has to answer that too. vi.mock("@/lib/crypto", () => ({ decrypt: (...a: unknown[]) => decryptMock(...a), + isStreamCiphertext: () => false, + decryptStream: vi.fn(), })); const parseMock = vi.fn(); diff --git a/src/lib/crypto.ts b/src/lib/crypto.ts index 242dffa37..5d8506387 100644 --- a/src/lib/crypto.ts +++ b/src/lib/crypto.ts @@ -396,3 +396,139 @@ export function extractKeyId(encoded: string): string | null { if (!/^[A-Za-z0-9_-]{1,32}$/.test(id)) return null; return id; } + +// ─── Streaming codec ("stream1") ───────────────────────────────────────────── +// +// AES-256-GCM over a plaintext that is produced INCREMENTALLY and consumed as +// one base64 string. Written for the weekly backup blob, whose plaintext is +// hundreds of megabytes on a long-lived record: the string codec above has to +// hold the whole plaintext, the whole ciphertext AND the whole base64 at once, +// which is three full copies of a record that already does not fit. +// +// Layout: "~hlgcm1." + keyId + "." + base64(iv | ciphertext | authTag) +// +// Two deliberate differences from the string codec, and no others: +// +// - The tag rides at the END rather than in front of the ciphertext. GCM +// only produces it once the last block is in, so a leading tag is what +// makes the string codec un-streamable. Nothing about the authentication +// changes: the tag still covers every ciphertext byte and `decryptStream` +// still verifies it BEFORE returning a single plaintext byte. +// - The `~` prefix. Key ids match [A-Za-z0-9_-]{1,32} and a legacy payload is +// bare base64, so no value the string codec can ever write starts with a +// tilde. The two formats are disjoint by construction rather than by +// sniffing, which is why `decrypt()` below is left untouched. +// +// The write side never materialises the ciphertext: base64 is emitted in +// 3-byte-aligned pieces as the cipher produces them, with the sub-group +// remainder carried across chunks. The read side is deliberately NOT +// incremental — a caller that reads a backup parses it as one document +// anyway, and releasing unverified plaintext to save a copy would trade the +// authentication for memory. + +/** Marker prefix of a streamed ciphertext. Disjoint from every other format. */ +const STREAM_CODEC_PREFIX = "~hlgcm1."; + +/** Incremental AES-256-GCM writer. Emits base64 pieces; concatenate in order. */ +export interface StreamEncryptor { + /** The header, including base64(iv). Emit before any `update()` output. */ + readonly header: string; + /** Encrypt one plaintext chunk into a base64 piece (may be empty). */ + update(chunk: Buffer): string; + /** Flush the cipher and the auth tag. No further calls after this. */ + final(): string; +} + +/** + * Open a streaming encryptor under the ACTIVE key. + * + * The 12-byte IV is exactly four base64 groups, so it encodes standalone and + * the ciphertext continues on a group boundary — which is what lets the rest + * be emitted piecewise without ever holding the whole thing. + */ +export function createStreamEncryptor(): StreamEncryptor { + const { id, key } = getActiveKey(); + const iv = randomBytes(IV_LENGTH); + const cipher = createCipheriv(ALGORITHM, key, iv); + // Bytes that did not fill a 3-byte base64 group in the previous chunk. + let carry = Buffer.alloc(0); + let done = false; + + const emit = (bytes: Buffer): string => { + const buf = carry.byteLength > 0 ? Buffer.concat([carry, bytes]) : bytes; + const aligned = buf.byteLength - (buf.byteLength % 3); + carry = Buffer.from(buf.subarray(aligned)); + return aligned === 0 ? "" : buf.subarray(0, aligned).toString("base64"); + }; + + return { + header: `${STREAM_CODEC_PREFIX}${id}.${iv.toString("base64")}`, + update(chunk: Buffer): string { + if (done) throw new Error("Stream encryptor already finalised"); + return emit(cipher.update(chunk)); + }, + final(): string { + if (done) throw new Error("Stream encryptor already finalised"); + done = true; + // The tag is plaintext-independent trailing data, so it simply joins the + // byte stream; the last group is padded exactly once, here. + const tail = Buffer.concat([carry, cipher.final(), cipher.getAuthTag()]); + carry = Buffer.alloc(0); + return tail.toString("base64"); + }, + }; +} + +/** True when `stored` was written by `createStreamEncryptor`. */ +export function isStreamCiphertext(stored: string): boolean { + return stored.startsWith(STREAM_CODEC_PREFIX); +} + +/** + * Decrypt a streamed ciphertext back to its raw plaintext bytes. + * + * Fail-closed on every arm — a malformed header, an unconfigured key id, a + * truncated payload or a tag that does not verify all throw, and no plaintext + * is returned until `final()` has authenticated the whole ciphertext. + */ +export function decryptStream(stored: string): Buffer { + if (!isStreamCiphertext(stored)) { + throw new Error("Not a streamed ciphertext"); + } + const rest = stored.slice(STREAM_CODEC_PREFIX.length); + const dot = rest.indexOf("."); + if (dot <= 0 || dot >= rest.length - 1) { + throw new Error("Streamed ciphertext has a malformed header"); + } + const keyId = rest.slice(0, dot); + if (!/^[A-Za-z0-9_-]{1,32}$/.test(keyId)) { + throw new Error("Streamed ciphertext carries a malformed key id"); + } + const key = getKeyById(keyId); + if (!key) { + throw new Error( + `Encryption key id '${keyId}' is not configured. Add it to ` + + `ENCRYPTION_KEYS before decrypting rows written under that key.`, + ); + } + const packed = Buffer.from(rest.slice(dot + 1), "base64"); + if (packed.byteLength < IV_LENGTH + AUTH_TAG_LENGTH) { + throw new Error("Streamed ciphertext is truncated"); + } + const iv = packed.subarray(0, IV_LENGTH); + const tag = packed.subarray(packed.byteLength - AUTH_TAG_LENGTH); + const ct = packed.subarray(IV_LENGTH, packed.byteLength - AUTH_TAG_LENGTH); + const dec = createDecipheriv(ALGORITHM, key, iv); + dec.setAuthTag(tag); + return Buffer.concat([dec.update(ct), dec.final()]); +} + +/** The key id a streamed ciphertext was written under, or null when unparsable. */ +export function extractStreamKeyId(stored: string): string | null { + if (!isStreamCiphertext(stored)) return null; + const rest = stored.slice(STREAM_CODEC_PREFIX.length); + const dot = rest.indexOf("."); + if (dot <= 0) return null; + const keyId = rest.slice(0, dot); + return /^[A-Za-z0-9_-]{1,32}$/.test(keyId) ? keyId : null; +} diff --git a/src/lib/export/__tests__/backup-blob.test.ts b/src/lib/export/__tests__/backup-blob.test.ts index 995ed07a6..a07b30ba4 100644 --- a/src/lib/export/__tests__/backup-blob.test.ts +++ b/src/lib/export/__tests__/backup-blob.test.ts @@ -20,7 +20,20 @@ process.env.ENCRYPTION_KEY ??= import { describe, expect, it } from "vitest"; import { encrypt } from "@/lib/crypto"; -import { packBackupBlob, unpackBackupBlob } from "../backup-blob"; +import { + packBackupBlob, + packBackupBlobStreaming, + unpackBackupBlob, +} from "../backup-blob"; + +/** Pack a whole string through the streaming writer, in small pieces. */ +async function packStreamed(json: string, pieceSize = 4_096): Promise { + return packBackupBlobStreaming(async (write) => { + for (let at = 0; at < json.length; at += pieceSize) { + await write(json.slice(at, at + pieceSize)); + } + }); +} /** A record-shaped string: many rows, few distinct keys. */ function sampleJson(rows: number): string { @@ -62,4 +75,62 @@ describe("backup blob envelope", () => { const mangled = `${packed.slice(0, -8)}AAAAAAAA`; expect(() => unpackBackupBlob(mangled)).toThrow(); }); + + it("reads back a streamed blob byte for byte", async () => { + const json = sampleJson(2_000); + expect(unpackBackupBlob(await packStreamed(json))).toBe(json); + }); + + it("is indifferent to where the producer's pieces fall", async () => { + // Base64 encodes three bytes at a time, so a writer that carries its + // sub-group remainder across chunks wrongly still round-trips at some + // piece sizes and corrupts at others. Cover every residue class. + // Small on purpose: a one-byte piece size means one write per character, + // and the point of the case is the residue class, not the volume. + const json = sampleJson(20); + for (const pieceSize of [1, 2, 3, 7, 64, 1_000, 1_001, json.length]) { + expect(unpackBackupBlob(await packStreamed(json, pieceSize))).toBe(json); + } + }); + + it("compresses the streamed form as hard as the buffered one", async () => { + const json = sampleJson(5_000); + expect((await packStreamed(json)).length).toBeLessThan(json.length / 4); + }); + + it("keeps the three stored shapes readable side by side", async () => { + // An operator's newest usable copy may predate either change, and the + // restore has to work for exactly that person. Plain, compressed and + // streamed all decode through the one entry point. + const json = sampleJson(50); + expect(unpackBackupBlob(encrypt(json))).toBe(json); + expect(unpackBackupBlob(packBackupBlob(json))).toBe(json); + expect(unpackBackupBlob(await packStreamed(json))).toBe(json); + }); + + it("cannot be confused for a legacy blob", async () => { + // The marker starts with a tilde, which is in neither the key-id charset + // nor the base64 alphabet, so no value the older writers can produce + // begins with it and the two families never have to be sniffed apart. + const streamed = await packStreamed(sampleJson(10)); + expect(streamed.startsWith("~")).toBe(true); + expect(packBackupBlob(sampleJson(10)).startsWith("~")).toBe(false); + expect(encrypt("x").startsWith("~")).toBe(false); + }); + + it("refuses a streamed blob whose authentication tag was altered", async () => { + const streamed = await packStreamed(sampleJson(20)); + // The tag is the last 16 bytes of the payload, so the tail of the base64. + const mangled = `${streamed.slice(0, -6)}${streamed.slice(-6) === "AAAAAA" ? "BBBBBB" : "AAAAAA"}`; + expect(() => unpackBackupBlob(mangled)).toThrow(); + }); + + it("propagates a producer failure instead of storing a truncated blob", async () => { + await expect( + packBackupBlobStreaming(async (write) => { + await write('{"measurements":['); + throw new Error("row source failed"); + }), + ).rejects.toThrow("row source failed"); + }); }); diff --git a/src/lib/export/backup-blob.ts b/src/lib/export/backup-blob.ts index 92d26d2b2..31c4a09e7 100644 --- a/src/lib/export/backup-blob.ts +++ b/src/lib/export/backup-blob.ts @@ -13,22 +13,38 @@ * * Compressing first is what makes the whole tail of that pipeline cheap. A * health record's JSON is extremely repetitive — the same twenty keys per row, - * timestamps sharing a prefix — so gzip takes that 242 MB to 14 MB, and every - * copy after it shrinks with it. The same run then completes in about eight - * seconds with a peak heap of 434 MB. It is also the reason the stored row - * stops being a liability of its own: the backup lives INSIDE the database it - * would be needed to restore, so its size is not a cosmetic concern. + * timestamps sharing a prefix — so gzip takes that 242 MB to a few tens of + * megabytes, and every copy after it shrinks with it. It is also the reason + * the stored row stops being a liability of its own: the backup lives INSIDE + * the database it would be needed to restore, so its size is not a cosmetic + * concern. * - * Both directions. Rows written before this envelope existed are plain - * `encrypt(json)` and have to keep restoring — an operator whose newest usable - * copy predates the fix is exactly the person who needs it to work. The marker - * sits inside the ciphertext rather than in front of it so the stored column - * keeps the one shape `decrypt()` already understands, key ids and all. + * Compressing first was not enough on its own. Even with a small stored blob, + * `packBackupBlob` still needs the whole JSON as one argument, and building + * that string is what exhausts the heap — measured under + * `--max-old-space-size=450` on the seeded account: `FATAL ERROR: Reached heap + * limit`. So the writer the weekly job actually uses is + * `packBackupBlobStreaming`, which never sees a complete copy of the JSON, the + * gzip output or the ciphertext: rows go in a page at a time, gzip and the + * cipher consume them as they arrive, and only the base64 answer accumulates, + * because the destination is a single `text` column and one value is what the + * column takes. + * + * Every direction reads. Three shapes exist in the wild and all three restore: + * the original `encrypt(json)`, the compressed `encrypt("HLZ1:" + gz)`, and + * the streamed `~hlgcm1.…` form written from now on. An operator whose newest + * usable copy predates any of this is exactly the person who needs it to work. */ import { Buffer } from "node:buffer"; -import { gunzipSync, gzipSync } from "node:zlib"; +import { createGzip, gunzipSync, gzipSync } from "node:zlib"; -import { decrypt, encrypt } from "@/lib/crypto"; +import { + createStreamEncryptor, + decrypt, + decryptStream, + encrypt, + isStreamCiphertext, +} from "@/lib/crypto"; /** * Prefix of the DECRYPTED plaintext when the body is gzipped-then-base64'd. @@ -43,14 +59,98 @@ export function packBackupBlob(json: string): string { return encrypt(`${GZIP_MARKER}${compressed}`); } +/** + * Produces the backup JSON in pieces. Every piece is written in order. + * + * Whatever it resolves to is ignored — the writer's own return value (the row + * counts) is the caller's business, not the envelope's. + */ +export type BackupJsonProducer = ( + write: (chunk: string) => Promise, +) => Promise; + +/** + * Serialised backup JSON, produced in pieces → the stored string. + * + * The pipeline is JSON piece → gzip → AES-256-GCM → base64, with nothing + * buffered end to end but the base64 answer. `producer` decides how big its + * pieces are; the gzip stream applies backpressure through the promise this + * hands back, so a fast producer cannot outrun the compressor and pile up + * chunks in the stream's internal queue. + * + * The one copy that cannot be avoided is the answer itself. `data_backups.data` + * is a single `text` column, so the row has to be one value, and one value has + * to exist as one string before the driver can bind it. + */ +export async function packBackupBlobStreaming( + producer: BackupJsonProducer, +): Promise { + const encryptor = createStreamEncryptor(); + const pieces: string[] = [encryptor.header]; + const gzip = createGzip(); + + let failure: unknown = null; + gzip.on("data", (chunk: Buffer) => { + try { + const piece = encryptor.update(chunk); + if (piece !== "") pieces.push(piece); + } catch (err) { + failure ??= err; + gzip.destroy(err as Error); + } + }); + + const finished = new Promise((resolve, reject) => { + gzip.on("end", resolve); + gzip.on("error", reject); + }); + + const write = async (chunk: string): Promise => { + if (failure) throw failure; + if (gzip.write(chunk, "utf8")) return; + await new Promise((resolve, reject) => { + const onDrain = () => { + gzip.off("error", onError); + resolve(); + }; + const onError = (err: Error) => { + gzip.off("drain", onDrain); + reject(err); + }; + gzip.once("drain", onDrain); + gzip.once("error", onError); + }); + }; + + try { + await producer(write); + } catch (err) { + gzip.destroy(); + throw err; + } + gzip.end(); + await finished; + if (failure) throw failure; + + pieces.push(encryptor.final()); + return pieces.join(""); +} + /** * A stored `DataBackup.data` string → the backup JSON. * - * Fails closed on every arm: a bad key, a mangled ciphertext or a truncated - * gzip member throws rather than returning a partial document, because every - * caller goes on to parse the result as a whole account. + * Fails closed on every arm: a bad key, a mangled ciphertext, a tag that does + * not verify or a truncated gzip member throws rather than returning a partial + * document, because every caller goes on to parse the result as a whole + * account. */ export function unpackBackupBlob(stored: string): string { + // Streamed form. Always gzipped — the streaming writer has no other mode — + // and the tag is verified over the whole ciphertext before a byte of this + // is unpacked. + if (isStreamCiphertext(stored)) { + return gunzipSync(decryptStream(stored)).toString("utf8"); + } const plaintext = decrypt(stored); if (!plaintext.startsWith(GZIP_MARKER)) return plaintext; return gunzipSync( diff --git a/src/lib/export/full-backup-payload.ts b/src/lib/export/full-backup-payload.ts index fbe8947e0..6f1328b98 100644 --- a/src/lib/export/full-backup-payload.ts +++ b/src/lib/export/full-backup-payload.ts @@ -9,6 +9,16 @@ * * The decrypted notes are surfaced here (the backup is the human-readable * artefact); an admin restore re-encrypts them on re-insert. + * + * Two writers read this file, not one. `buildFullBackupPayload` materialises + * the whole payload, which is what the two export routes and every test want. + * `streamFullBackupJson` (`full-backup-stream.ts`) asks for the same payload + * with `deferBulk`, which leaves the three unbounded tables as markers and + * pulls their rows through itself — the weekly job takes that path because + * materialising a large record is what used to exhaust the container's heap. + * Both go through the same per-row serialisers below, so a column that reaches + * one reaches the other; the integration suite pins the two outputs byte for + * byte rather than trusting that sentence. */ import { Buffer } from "node:buffer"; import type { Prisma, PrismaClient } from "@/generated/prisma/client"; @@ -146,6 +156,14 @@ export interface FullBackupResult { export interface FullBackupOptions { purpose?: "portable-export" | "disaster-recovery"; exportedAt?: Date; + /** + * Declare the three unbounded tables instead of reading them. + * + * `measurements`, `intakeEvents` and `moodEntries` come back as + * `DeferredRows` markers and their counts as 0. Only `streamFullBackupJson` + * sets this; every other caller wants the arrays and gets them. + */ + deferBulk?: boolean; } /** * Every restorable `Measurement` scalar. `userId` is deliberately absent — the @@ -266,6 +284,35 @@ type BackupMoodContext = Prisma.MoodContextGetPayload<{ select: typeof MOOD_CONTEXT_BACKUP_SELECT; }>; +type BackupMeasurement = Prisma.MeasurementGetPayload<{ + select: typeof MEASUREMENT_BACKUP_SELECT; +}>; + +type BackupIntakeEvent = Prisma.MedicationIntakeEventGetPayload<{ + include: { medication: { select: { name: true } } }; +}>; + +/** + * The mood read, select and joins together. Named because two readers issue + * it — the materialising one and the paged one — and a select that lived in + * only one of them is exactly the drift the completeness guard cannot see. + */ +const MOOD_ENTRY_QUERY_SELECT = { + ...MOOD_ENTRY_BACKUP_SELECT, + // Every link, not only the rated ones. This read used to filter + // `rating: { not: null }`, and a BINARY link carries a NULL rating by + // definition — so the present/absent tags, the kind most people pick, + // were absent from the file and could not come back from it. + tagLinks: { select: MOOD_ENTRY_TAG_LINK_SELECT }, + // The day context, when the entry has one. Sparse by design — most + // entries carry none, and the join answers null on those. + context: { select: MOOD_CONTEXT_BACKUP_SELECT }, +} as const satisfies Prisma.MoodEntrySelect; + +type BackupMoodEntry = Prisma.MoodEntryGetPayload<{ + select: typeof MOOD_ENTRY_QUERY_SELECT; +}>; + /** * One context row, in whichever of the two wire shapes the caller asked for. * @@ -316,24 +363,339 @@ function serialiseMoodContext( }; } +/** + * One measurement, in whichever of the two wire shapes the caller asked for. + * + * Named rather than inlined so the paged reader below and the materialising + * reader beside it cannot drift: both call this, so a column that reaches one + * shape reaches the other by construction. `measurement-backup-completeness. + * test.ts` proves a selected column reaches a payload by matching + * `: measurement.`, which is why the parameter keeps that spelling. + */ +function serialiseMeasurement( + measurement: BackupMeasurement, + disasterRecovery: boolean, +) { + return disasterRecovery + ? { + id: measurement.id, + type: measurement.type, + value: measurement.value, + valueMin: measurement.valueMin, + valueMax: measurement.valueMax, + unit: measurement.unit, + measuredAt: measurement.measuredAt.toISOString(), + source: measurement.source, + notes: measurement.notes, + notesEncrypted: measurement.notesEncrypted + ? Buffer.from(measurement.notesEncrypted).toString("base64") + : null, + externalId: measurement.externalId, + externalSourceVersion: measurement.externalSourceVersion, + aggregationProvenance: measurement.aggregationProvenance, + glucoseContext: measurement.glucoseContext, + sleepStage: measurement.sleepStage, + rhythmClassification: measurement.rhythmClassification, + deviceType: measurement.deviceType, + syncVersion: measurement.syncVersion, + deletedAt: measurement.deletedAt?.toISOString() ?? null, + createdAt: measurement.createdAt.toISOString(), + updatedAt: measurement.updatedAt.toISOString(), + } + : { + id: measurement.id, + type: measurement.type, + value: measurement.value, + unit: measurement.unit, + measuredAt: measurement.measuredAt.toISOString(), + source: measurement.source, + notes: readNote(measurement.notesEncrypted, measurement.notes), + deletedAt: measurement.deletedAt?.toISOString() ?? null, + }; +} + +/** One intake event, as it rides the wire. */ +function serialiseIntakeEvent(e: BackupIntakeEvent, disasterRecovery: boolean) { + return { + ...(disasterRecovery + ? { + id: e.id, + medicationId: e.medicationId, + autoMissed: e.autoMissed, + attributionSource: e.attributionSource, + idempotencyKey: e.idempotencyKey, + createdAt: e.createdAt.toISOString(), + injectionSite: e.injectionSite, + doseTaken: e.doseTaken, + inventoryConsumption: e.inventoryConsumption, + externalId: e.externalId, + updatedAt: e.updatedAt.toISOString(), + syncVersion: e.syncVersion, + deletedAt: e.deletedAt?.toISOString() ?? null, + } + : {}), + medication: e.medication.name, + scheduledFor: e.scheduledFor.toISOString(), + takenAt: e.takenAt?.toISOString() ?? null, + skipped: e.skipped, + source: e.source, + }; +} + +/** + * One mood entry, with its links partitioned and its day context folded in. + * + * The parameter is spelled out rather than the `e` every other mapper here + * uses, and deliberately: `mood-backup-completeness.test.ts` proves a + * selected column reaches a payload by matching `: moodEntry.`. A matcher + * on `: e.` would also match the intake-event mapper above it and could + * therefore never fail. + */ +function serialiseMoodEntry( + moodEntry: BackupMoodEntry, + disasterRecovery: boolean, +) { + // The two link arms partition the entry's links exhaustively: a link is + // a rated factor when its tag is RATED and it actually carries a score, + // and a structured tag otherwise. Total on purpose — a malformed link + // (a RATED tag whose rating went missing) rides as a tag key and comes + // back NAMED as unresolvable on restore, rather than falling between the + // two arms and disappearing the way every BINARY link used to. + const isRatedFactor = (link: (typeof moodEntry.tagLinks)[number]) => + link.moodTag.kind === "RATED" && link.rating !== null; + + const factors = moodEntry.tagLinks.filter(isRatedFactor).map((tagLink) => ({ + key: tagLink.moodTag.key, + rating: tagLink.rating as number, + })); + // BINARY link keys. Their own array rather than a nullable rating inside + // `factors`: the restore resolves `factors` against `kind: "RATED"` and + // reports what it cannot resolve, so a rating-less entry in that array + // would collide with both. A separate key leaves every backup file + // written before this parsing and restoring exactly as it did. + const structuredTags = moodEntry.tagLinks + .filter((link) => !isRatedFactor(link)) + .map((tagLink) => tagLink.moodTag.key); + + return { + ...(disasterRecovery + ? { + note: moodEntry.note, + noteEncrypted: moodEntry.noteEncrypted + ? Buffer.from(moodEntry.noteEncrypted).toString("base64") + : null, + syncedAt: moodEntry.syncedAt.toISOString(), + syncVersion: moodEntry.syncVersion, + createdAt: moodEntry.createdAt.toISOString(), + updatedAt: moodEntry.updatedAt.toISOString(), + } + : { note: readNote(moodEntry.noteEncrypted, moodEntry.note) }), + id: moodEntry.id, + date: moodEntry.date, + mood: moodEntry.mood, + score: moodEntry.score, + // The five level-A values, under the keys the write path takes. Both + // arms carry them: they are the entry's own answers, not a storage + // detail, and a portable export that dropped them would hand back a + // file describing the day less well than the row it came from. + a1: moodEntry.moodA1, + a2: moodEntry.stressA2, + a3: moodEntry.energyA3, + a4: moodEntry.connectionA4, + a5: moodEntry.stabilityA5, + tags: moodEntry.tags, + source: moodEntry.source, + loggedAt: moodEntry.moodLoggedAt.toISOString(), + externalId: moodEntry.externalId, + // The zone the `date` string is anchored to. Without it a restored row + // falls back to the legacy Europe/Berlin reading and an account that + // logged elsewhere gets its day boundaries quietly moved. + tz: moodEntry.tz, + deletedAt: moodEntry.deletedAt?.toISOString() ?? null, + factors, + structuredTags, + // The day context, or null. Null rather than an empty object, because + // "no context" and "a context whose every field is empty" are + // different facts and the restore has to keep them apart. + context: moodEntry.context + ? serialiseMoodContext(moodEntry.context, disasterRecovery) + : null, + }; +} + +/** + * How many rows one page of a bulk table carries. + * + * Smaller than the measurement pager's 10 000 on purpose: an intake event + * drags its medication's name along and a mood entry drags its links and its + * day context, so a page of either costs several times what a page of + * measurements costs. The number is a memory budget, not a throughput knob. + */ +const BULK_PAGE_SIZE = 5_000; + +/** + * Every backup measurement, one serialised row at a time. + * + * Keyset-paged through `iterateMeasurementPages`, so the caller decides + * whether to keep the rows (the materialising payload does) or hand each + * straight to a writer and forget it (the streaming writer does). Nothing here + * holds more than one page. + */ +export async function* iterateBackupMeasurements( + prisma: PrismaClient, + userId: string, + disasterRecovery: boolean, +): AsyncGenerator, void, void> { + const pages = iterateMeasurementPages( + prisma, + disasterRecovery ? { userId } : { userId, deletedAt: null }, + MEASUREMENT_BACKUP_SELECT, + ); + for await (const page of pages) { + for (const measurement of page) { + yield serialiseMeasurement(measurement, disasterRecovery); + } + } +} + +/** + * Every backup intake event, one serialised row at a time. + * + * The sort is TOTAL — `scheduledFor desc` then `id desc` — where the + * materialising read used to order by the timestamp alone. A keyset cursor + * needs a total order to page correctly, and a partial one also let two + * exports of the same account emit two different files whenever two doses + * shared a scheduled minute, which every multi-dose day has. + */ +export async function* iterateBackupIntakeEvents( + prisma: PrismaClient, + userId: string, + disasterRecovery: boolean, +): AsyncGenerator, void, void> { + let cursorId: string | null = null; + for (;;) { + const page: BackupIntakeEvent[] = + await prisma.medicationIntakeEvent.findMany({ + where: disasterRecovery ? { userId } : { userId, deletedAt: null }, + include: { medication: { select: { name: true } } }, + orderBy: [{ scheduledFor: "desc" }, { id: "desc" }], + take: BULK_PAGE_SIZE, + ...(cursorId !== null ? { cursor: { id: cursorId }, skip: 1 } : {}), + }); + if (page.length === 0) return; + for (const e of page) yield serialiseIntakeEvent(e, disasterRecovery); + if (page.length < BULK_PAGE_SIZE) return; + cursorId = page[page.length - 1]!.id; + } +} + +/** + * Every backup mood entry, one serialised row at a time. + * + * Same total-order argument as the intake events above: `moodLoggedAt desc` + * alone is not a cursor, and two entries logged in the same millisecond are + * exactly what a bulk import produces. + */ +export async function* iterateBackupMoodEntries( + prisma: PrismaClient, + userId: string, + disasterRecovery: boolean, +): AsyncGenerator, void, void> { + let cursorId: string | null = null; + for (;;) { + const page: BackupMoodEntry[] = await prisma.moodEntry.findMany({ + where: disasterRecovery ? { userId } : { userId, deletedAt: null }, + orderBy: [{ moodLoggedAt: "desc" }, { id: "desc" }], + take: BULK_PAGE_SIZE, + ...(cursorId !== null ? { cursor: { id: cursorId }, skip: 1 } : {}), + select: MOOD_ENTRY_QUERY_SELECT, + }); + if (page.length === 0) return; + for (const moodEntry of page) { + yield serialiseMoodEntry(moodEntry, disasterRecovery); + } + if (page.length < BULK_PAGE_SIZE) return; + cursorId = page[page.length - 1]!.id; + } +} + +/** Drain an async row source into an array. The materialising arm's only use. */ +async function collect(rows: AsyncIterable): Promise { + const out: T[] = []; + for await (const row of rows) out.push(row); + return out; +} + +/** + * A bulk table the payload declares but has NOT read. + * + * `buildFullBackupPayload({ deferBulk: true })` puts one of these where the + * row array would sit. The streaming writer recognises it, opens the JSON + * array itself and pulls rows through page by page, so a table with half a + * million rows never exists as an array at all. Every other caller leaves + * `deferBulk` unset and gets the arrays, unchanged. + * + * The marker key is a symbol so it cannot collide with a payload field and so + * the deferred payload can never be handed to `JSON.stringify` by accident and + * come back looking merely empty — `streamFullBackupJson` is the only reader. + */ +const DEFERRED_ROWS = Symbol("healthlog.backup.deferredRows"); + +export interface DeferredRows { + readonly [DEFERRED_ROWS]: true; + /** Open the row source. Called exactly once, by the streaming writer. */ + rows(): AsyncIterable>; +} + +function deferredRows( + rows: () => AsyncIterable>, +): DeferredRows { + return { [DEFERRED_ROWS]: true, rows }; +} + +/** True when a payload value is a deferred bulk table rather than an array. */ +export function isDeferredRows(value: unknown): value is DeferredRows { + return ( + typeof value === "object" && + value !== null && + (value as Record)[DEFERRED_ROWS] === true + ); +} + /** * The same payload, already serialised, with the object graph released. * - * Its own frame on purpose, and both backup jobs go through it. The payload - * graph is the largest thing either job allocates — on an account with a few - * hundred thousand measurements it is several hundred megabytes of small - * objects — and neither job wants anything but the string. Returning from here - * makes the graph unreachable before the compress-and-encrypt step allocates - * anything of its own, which is part of why the weekly pass now fits in - * memory. Inlining these two statements back into a caller quietly undoes it. + * Built through the streaming writer rather than by stringifying a finished + * payload, so the object graph and the string never coexist: the three + * unbounded tables go past a page at a time and every other section is + * released as soon as its JSON exists. What is left is the answer itself, + * which a caller that wants a string necessarily holds. + * + * A caller that does NOT need the string — the weekly job, which only wants + * the compressed ciphertext — should go through `packBackupBlobStreaming` and + * `streamFullBackupJson` directly and never let the whole document exist at + * all. This function is for the callers that genuinely hand a JSON body on. */ export async function buildFullBackupJson( prisma: PrismaClient, userId: string, options: FullBackupOptions = {}, ): Promise { - const { payload } = await buildFullBackupPayload(prisma, userId, options); - return JSON.stringify(payload); + // Imported here rather than at the top of the file: the writer imports this + // module for `buildFullBackupPayload`, and a static edge in both directions + // is a cycle. The call is once per backup, so the cost is nil. + const { streamFullBackupJson } = + await import("@/lib/export/full-backup-stream"); + const pieces: string[] = []; + await streamFullBackupJson( + prisma, + userId, + (chunk) => { + pieces.push(chunk); + }, + options, + ); + return pieces.join(""); } /** @@ -347,6 +709,7 @@ export async function buildFullBackupPayload( options: FullBackupOptions = {}, ): Promise { const disasterRecovery = options.purpose === "disaster-recovery"; + const deferBulk = options.deferBulk === true; const [ appSettings, measurements, @@ -376,60 +739,9 @@ export async function buildFullBackupPayload( disasterRecovery ? prisma.appSettings.findUnique({ where: { id: "singleton" } }) : Promise.resolve(null), - (async () => { - const rows: Array> = []; - const pages = iterateMeasurementPages( - prisma, - disasterRecovery ? { userId } : { userId, deletedAt: null }, - MEASUREMENT_BACKUP_SELECT, - ); - for await (const page of pages) { - for (const measurement of page) { - rows.push( - disasterRecovery - ? { - id: measurement.id, - type: measurement.type, - value: measurement.value, - valueMin: measurement.valueMin, - valueMax: measurement.valueMax, - unit: measurement.unit, - measuredAt: measurement.measuredAt.toISOString(), - source: measurement.source, - notes: measurement.notes, - notesEncrypted: measurement.notesEncrypted - ? Buffer.from(measurement.notesEncrypted).toString("base64") - : null, - externalId: measurement.externalId, - externalSourceVersion: measurement.externalSourceVersion, - aggregationProvenance: measurement.aggregationProvenance, - glucoseContext: measurement.glucoseContext, - sleepStage: measurement.sleepStage, - rhythmClassification: measurement.rhythmClassification, - deviceType: measurement.deviceType, - syncVersion: measurement.syncVersion, - deletedAt: measurement.deletedAt?.toISOString() ?? null, - createdAt: measurement.createdAt.toISOString(), - updatedAt: measurement.updatedAt.toISOString(), - } - : { - id: measurement.id, - type: measurement.type, - value: measurement.value, - unit: measurement.unit, - measuredAt: measurement.measuredAt.toISOString(), - source: measurement.source, - notes: readNote( - measurement.notesEncrypted, - measurement.notes, - ), - deletedAt: measurement.deletedAt?.toISOString() ?? null, - }, - ); - } - } - return rows; - })(), + deferBulk + ? null + : collect(iterateBackupMeasurements(prisma, userId, disasterRecovery)), prisma.medication.findMany({ where: { userId }, // Side effects ride INSIDE their medication, exactly as the schedules @@ -469,26 +781,12 @@ export async function buildFullBackupPayload( }, }, }), - prisma.medicationIntakeEvent.findMany({ - where: disasterRecovery ? { userId } : { userId, deletedAt: null }, - include: { medication: { select: { name: true } } }, - orderBy: { scheduledFor: "desc" }, - }), - prisma.moodEntry.findMany({ - where: disasterRecovery ? { userId } : { userId, deletedAt: null }, - orderBy: { moodLoggedAt: "desc" }, - select: { - ...MOOD_ENTRY_BACKUP_SELECT, - // Every link, not only the rated ones. This read used to filter - // `rating: { not: null }`, and a BINARY link carries a NULL rating by - // definition — so the present/absent tags, the kind most people pick, - // were absent from the file and could not come back from it. - tagLinks: { select: MOOD_ENTRY_TAG_LINK_SELECT }, - // The day context, when the entry has one. Sparse by design — most - // entries carry none, and the join answers null on those. - context: { select: MOOD_CONTEXT_BACKUP_SELECT }, - }, - }), + deferBulk + ? null + : collect(iterateBackupIntakeEvents(prisma, userId, disasterRecovery)), + deferBulk + ? null + : collect(iterateBackupMoodEntries(prisma, userId, disasterRecovery)), // The account's OWN mood tags. The seeded catalogue is reference data every // instance has; a tag the user made lives only here. Without it the restore // cannot resolve the entry's factor keys — and unlike the cycle path, which @@ -646,6 +944,27 @@ export async function buildFullBackupPayload( const environmentSection: EnvironmentBackupSection = environment; const ecgSection: EcgBackupSection = ecg; + // The three unbounded tables, either read into arrays or left as markers + // the streaming writer will pull through. One place decides, so the payload + // literal below reads the same in both modes. + const bulk = { + measurements: deferBulk + ? deferredRows(() => + iterateBackupMeasurements(prisma, userId, disasterRecovery), + ) + : measurements!, + intakeEvents: deferBulk + ? deferredRows(() => + iterateBackupIntakeEvents(prisma, userId, disasterRecovery), + ) + : intakeEvents!, + moodEntries: deferBulk + ? deferredRows(() => + iterateBackupMoodEntries(prisma, userId, disasterRecovery), + ) + : moodEntries!, + }; + // Where each archived era sits in its OWN drug's ordered list. Built once // here rather than per row, and scoped per medication because the supersede // pointer never crosses a drug — every route that writes it scopes the @@ -674,7 +993,7 @@ export async function buildFullBackupPayload( documentQuotaBytes: appSettings.documentQuotaBytes.toString(), } : null, - measurements, + measurements: bulk.measurements, medications: medications.map((m) => ({ ...(disasterRecovery ? { @@ -903,105 +1222,8 @@ export async function buildFullBackupPayload( primary: t.primary, })), })), - intakeEvents: intakeEvents.map((e) => ({ - ...(disasterRecovery - ? { - id: e.id, - medicationId: e.medicationId, - autoMissed: e.autoMissed, - attributionSource: e.attributionSource, - idempotencyKey: e.idempotencyKey, - createdAt: e.createdAt.toISOString(), - injectionSite: e.injectionSite, - doseTaken: e.doseTaken, - inventoryConsumption: e.inventoryConsumption, - externalId: e.externalId, - updatedAt: e.updatedAt.toISOString(), - syncVersion: e.syncVersion, - deletedAt: e.deletedAt?.toISOString() ?? null, - } - : {}), - medication: e.medication.name, - scheduledFor: e.scheduledFor.toISOString(), - takenAt: e.takenAt?.toISOString() ?? null, - skipped: e.skipped, - source: e.source, - })), - // The parameter is spelled out rather than the `e` every other mapper here - // uses, and deliberately: `mood-backup-completeness.test.ts` proves a - // selected column reaches a payload by matching `: moodEntry.`. A matcher - // on `: e.` would also match the intake-event mapper above it and could - // therefore never fail. - moodEntries: moodEntries.map((moodEntry) => { - // The two link arms partition the entry's links exhaustively: a link is - // a rated factor when its tag is RATED and it actually carries a score, - // and a structured tag otherwise. Total on purpose — a malformed link - // (a RATED tag whose rating went missing) rides as a tag key and comes - // back NAMED as unresolvable on restore, rather than falling between the - // two arms and disappearing the way every BINARY link used to. - const isRatedFactor = (link: (typeof moodEntry.tagLinks)[number]) => - link.moodTag.kind === "RATED" && link.rating !== null; - - const factors = moodEntry.tagLinks - .filter(isRatedFactor) - .map((tagLink) => ({ - key: tagLink.moodTag.key, - rating: tagLink.rating as number, - })); - // BINARY link keys. Their own array rather than a nullable rating inside - // `factors`: the restore resolves `factors` against `kind: "RATED"` and - // reports what it cannot resolve, so a rating-less entry in that array - // would collide with both. A separate key leaves every backup file - // written before this parsing and restoring exactly as it did. - const structuredTags = moodEntry.tagLinks - .filter((link) => !isRatedFactor(link)) - .map((tagLink) => tagLink.moodTag.key); - - return { - ...(disasterRecovery - ? { - note: moodEntry.note, - noteEncrypted: moodEntry.noteEncrypted - ? Buffer.from(moodEntry.noteEncrypted).toString("base64") - : null, - syncedAt: moodEntry.syncedAt.toISOString(), - syncVersion: moodEntry.syncVersion, - createdAt: moodEntry.createdAt.toISOString(), - updatedAt: moodEntry.updatedAt.toISOString(), - } - : { note: readNote(moodEntry.noteEncrypted, moodEntry.note) }), - id: moodEntry.id, - date: moodEntry.date, - mood: moodEntry.mood, - score: moodEntry.score, - // The five level-A values, under the keys the write path takes. Both - // arms carry them: they are the entry's own answers, not a storage - // detail, and a portable export that dropped them would hand back a - // file describing the day less well than the row it came from. - a1: moodEntry.moodA1, - a2: moodEntry.stressA2, - a3: moodEntry.energyA3, - a4: moodEntry.connectionA4, - a5: moodEntry.stabilityA5, - tags: moodEntry.tags, - source: moodEntry.source, - loggedAt: moodEntry.moodLoggedAt.toISOString(), - externalId: moodEntry.externalId, - // The zone the `date` string is anchored to. Without it a restored row - // falls back to the legacy Europe/Berlin reading and an account that - // logged elsewhere gets its day boundaries quietly moved. - tz: moodEntry.tz, - deletedAt: moodEntry.deletedAt?.toISOString() ?? null, - factors, - structuredTags, - // The day context, or null. Null rather than an empty object, because - // "no context" and "a context whose every field is empty" are - // different facts and the restore has to keep them apart. - context: moodEntry.context - ? serialiseMoodContext(moodEntry.context, disasterRecovery) - : null, - }; - }), + intakeEvents: bulk.intakeEvents, + moodEntries: bulk.moodEntries, customMoodTagCategories: customMoodTagCategories.map((category) => ({ id: category.id, key: category.key, @@ -1079,9 +1301,9 @@ export async function buildFullBackupPayload( return { payload, counts: { - measurements: measurements.length, + measurements: measurements?.length ?? 0, medications: medications.length, - intakeEvents: intakeEvents.length, + intakeEvents: intakeEvents?.length ?? 0, medicationSideEffects: medications.reduce( (total, m) => total + m.sideEffects.length, 0, @@ -1111,7 +1333,7 @@ export async function buildFullBackupPayload( (total, m) => total + m.scheduleRevisions.length, 0, ), - moodEntries: moodEntries.length, + moodEntries: moodEntries?.length ?? 0, customMoodTagCategories: customMoodTagCategories.length, hiddenMoodTags: hiddenMoodTags.length, cycles: cycle.cycles.length, diff --git a/src/lib/export/full-backup-stream.ts b/src/lib/export/full-backup-stream.ts new file mode 100644 index 000000000..bf4cb2b03 --- /dev/null +++ b/src/lib/export/full-backup-stream.ts @@ -0,0 +1,160 @@ +/** + * The full-backup payload, written out incrementally instead of built. + * + * Why it exists. `buildFullBackupJson` has to hold two things at once that + * both scale with the record: the payload object graph, and the JSON string + * `JSON.stringify` makes of it. On a seeded account of 445 000 measurements, + * 30 000 mood entries and 60 000 intake events, that pair alone exhausts a + * 546 MB heap and the process dies with `Reached heap limit` — measured, not + * inferred, and reproducible with `--max-old-space-size=450`. The production + * container is capped at 1 GB, which puts V8's limit at 524 MB, so the weekly + * pass took the whole app down with it: every other user of that instance lost + * their session because one account's backup did not fit. + * + * What changed. The three tables that scale with a long-lived record — + * measurements, intake events, mood entries — are declared rather than read + * (`deferBulk`), and this writer pulls them through page by page, serialising + * one row at a time straight into the sink. Every other section is stringified + * and then RELEASED from the payload before the next one is touched, so the + * peak is the largest single section rather than the sum of all of them. + * + * The output is byte-identical to `JSON.stringify(payload)`. That is not a + * hope: `Object.entries` walks a plain object in insertion order, which is the + * order `JSON.stringify` uses, and `JSON.stringify([a, b])` is exactly + * `"[" + JSON.stringify(a) + "," + JSON.stringify(b) + "]"` for plain rows. + * The integration suite pins it against the materialising builder rather than + * trusting that paragraph. + */ +import v8 from "node:v8"; + +import type { PrismaClient } from "@/generated/prisma/client"; +import { + buildFullBackupPayload, + isDeferredRows, + type FullBackupCounts, + type FullBackupOptions, +} from "@/lib/export/full-backup-payload"; + +/** Where the writer puts its pieces. Awaited, so a sink can apply backpressure. */ +export type BackupJsonSink = (chunk: string) => void | Promise; + +/** + * How much serialised JSON to gather before handing it to the sink. + * + * Row-at-a-time writes would push half a million tiny strings through gzip and + * spend more time in framing than in compression; a batch this size keeps the + * resident buffer trivial and the write count in the low thousands. + */ +const FLUSH_BYTES = 256 * 1024; + +/** + * Fraction of V8's heap limit the writer refuses to cross. + * + * The point is not to make an oversized account succeed — it is to make it + * FAIL AS A JOB. A backup that runs out of heap takes the process with it, and + * a pg-boss worker sharing the app process means one account's record can + * restart the instance for everybody. Because the work is now incremental + * there is a checkpoint every few hundred kilobytes, so the guard actually + * gets to run before V8 hits the wall; the old single `JSON.stringify` had no + * point at which anything could intervene. + */ +const HEAP_HIGH_WATER = 0.8; + +/** Thrown when the writer stops itself short of the heap limit. */ +export class BackupHeapBudgetExceededError extends Error { + constructor(usedBytes: number, ceilingBytes: number) { + super( + `Backup aborted at ${Math.round(usedBytes / 1024 / 1024)} MB of heap ` + + `(budget ${Math.round(ceilingBytes / 1024 / 1024)} MB). The record is ` + + `too large for this container's memory limit; raise the container's ` + + `memory or NODE_OPTIONS=--max-old-space-size.`, + ); + this.name = "BackupHeapBudgetExceededError"; + } +} + +export interface StreamFullBackupOptions extends FullBackupOptions { + /** + * Heap ceiling in bytes. Defaults to 80 % of V8's limit for this process. + * Tests pass an explicit value; nothing else should need to. + */ + heapCeilingBytes?: number; +} + +function defaultHeapCeiling(): number { + return Math.floor(v8.getHeapStatistics().heap_size_limit * HEAP_HIGH_WATER); +} + +/** + * Write the full-backup JSON for `userId` into `sink`, and answer the counts. + * + * The counts for the three deferred tables are filled in as their rows go + * past, so a caller gets the same numbers the materialising builder reports. + */ +export async function streamFullBackupJson( + prisma: PrismaClient, + userId: string, + sink: BackupJsonSink, + options: StreamFullBackupOptions = {}, +): Promise { + const ceiling = options.heapCeilingBytes ?? defaultHeapCeiling(); + const { payload, counts } = await buildFullBackupPayload(prisma, userId, { + ...options, + deferBulk: true, + }); + + let pending = ""; + let pendingBytes = 0; + + const flush = async (force: boolean): Promise => { + if (pending === "" || (!force && pendingBytes < FLUSH_BYTES)) return; + const chunk = pending; + pending = ""; + pendingBytes = 0; + await sink(chunk); + const used = process.memoryUsage().heapUsed; + if (used > ceiling) throw new BackupHeapBudgetExceededError(used, ceiling); + }; + + const write = async (piece: string): Promise => { + pending += piece; + pendingBytes += piece.length; + await flush(false); + }; + + const bulkCounts: Record = {}; + + await write("{"); + let first = true; + // The walk is destructive: each section is released from the payload as soon + // as its JSON is in the sink, which is what keeps the peak at one section + // rather than at all of them. Nothing else reads this object. + for (const key of Object.keys(payload)) { + const value = payload[key]; + // `JSON.stringify` omits an undefined-valued key; so does this. + if (value === undefined) continue; + if (!first) await write(","); + first = false; + await write(`${JSON.stringify(key)}:`); + + if (isDeferredRows(value)) { + await write("["); + let rows = 0; + for await (const row of value.rows()) { + await write( + rows === 0 ? JSON.stringify(row) : `,${JSON.stringify(row)}`, + ); + rows++; + } + await write("]"); + bulkCounts[key] = rows; + } else { + await write(JSON.stringify(value)); + } + delete payload[key]; + } + await write("}"); + await flush(true); + + return { ...counts, ...bulkCounts } as FullBackupCounts; +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 7b54b3985..673378a85 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -103,6 +103,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "geo_backfill_skipped", "geo_backfill_still_unresolved", "geolite2_fetch_status", + "heap_budget_trips", "host_metric_pruned", "hourly_rows_upserted", "imported", diff --git a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts index bfb009c8f..918571742 100644 --- a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts +++ b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts @@ -9,23 +9,29 @@ const mocks = vi.hoisted(() => ({ upsert: vi.fn(), })); -// The handler takes the already-serialised form; the stand-in serialises -// whatever the payload mock was told to return, so this file keeps stubbing -// and asserting the PAYLOAD, which is what it is about. +// Only the payload builder is stubbed. The REAL streaming writer runs on top +// of it, so this file keeps stubbing and asserting the PAYLOAD — which is what +// it is about — while the framing that turns it into stored bytes is the +// framing the job actually uses. `isDeferredRows` rides along because the +// writer asks it about every section; nothing here defers. vi.mock("@/lib/export/full-backup-payload", () => ({ buildFullBackupPayload: mocks.buildFullBackupPayload, - buildFullBackupJson: async (...args: unknown[]) => - JSON.stringify( - ((await mocks.buildFullBackupPayload(...args)) as { payload: unknown }) - .payload, - ), + isDeferredRows: () => false, })); // The envelope is exercised end-to-end in // `src/lib/export/__tests__/backup-blob.test.ts`; here it stays transparent so // the stored bytes can be read back as JSON. vi.mock("@/lib/export/backup-blob", () => ({ - packBackupBlob: mocks.packBlob, + packBackupBlobStreaming: async ( + produce: (write: (chunk: string) => Promise) => Promise, + ) => { + let out = ""; + await produce(async (chunk) => { + out += chunk; + }); + return mocks.packBlob(out); + }, })); vi.mock("@/lib/logging/background", () => ({ @@ -95,7 +101,12 @@ describe("handleDataBackup canonical DR payload", () => { expect(mocks.buildFullBackupPayload).toHaveBeenCalledWith( prisma, "user-dr", - { purpose: "disaster-recovery" }, + // `deferBulk` is the writer's own ask: it declares the unbounded tables + // rather than reading them, and pulls their rows through itself. + expect.objectContaining({ + purpose: "disaster-recovery", + deferBulk: true, + }), ); expect(mocks.upsert).toHaveBeenCalledOnce(); const encrypted = mocks.upsert.mock.calls[0]![0].create.data as string; diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index e45da75a8..1676fe626 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -6,8 +6,11 @@ */ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; -import { buildFullBackupJson } from "@/lib/export/full-backup-payload"; -import { packBackupBlob } from "@/lib/export/backup-blob"; +import { packBackupBlobStreaming } from "@/lib/export/backup-blob"; +import { + BackupHeapBudgetExceededError, + streamFullBackupJson, +} from "@/lib/export/full-backup-stream"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; import { @@ -79,14 +82,19 @@ export async function handleDataBackup( let backed = 0; let usersFailed = 0; + let heapBudgetTrips = 0; let largestBlobBytes = 0; for (const user of users) { try { - // Compressed, then encrypted (the record contains sensitive health - // information). Both legs live in `packBackupBlob`, which is also - // what the restore path reads back through. - const encryptedBackup = packBackupBlob( - await buildFullBackupJson(prisma, user.id, { + // Streamed, compressed, then encrypted (the record contains + // sensitive health information). Nothing between the database rows + // and this string exists as a whole: the payload goes into gzip a + // page at a time and the cipher consumes gzip's output as it comes, + // because a materialised copy of a large record is what used to take + // the process down. Both legs live in `packBackupBlobStreaming`, + // which is also what the restore path reads back through. + const encryptedBackup = await packBackupBlobStreaming((write) => + streamFullBackupJson(prisma, user.id, write, { purpose: "disaster-recovery", }), ); @@ -114,7 +122,14 @@ export async function handleDataBackup( // One user's payload failing is that user's problem, not the pass's: // it rides out as a count so the weekly run is not retried for the // whole cohort. + // + // A record too large for this container's heap lands here too, and + // that is the whole point of the budget the writer enforces: before + // it existed the same condition was an uncatchable V8 abort that + // restarted the instance, so one account's size was a denial of + // service on every other account on the host. usersFailed++; + if (err instanceof BackupHeapBudgetExceededError) heapBudgetTrips++; evt.addWarning(`Failed for user ${user.id}: ${err}`); } } @@ -122,6 +137,9 @@ export async function handleDataBackup( // The stored size is the one number that says whether this pass is // heading back towards the wall it hit before: it tracks the record. evt.addMeta("data_backup_largest_blob_bytes", largestBlobBytes); + // How many accounts the writer stopped rather than let the process die. + // Non-zero means this host needs more memory, not a retry. + evt.addMeta("data_backup_heap_budget_trips", heapBudgetTrips); evt.setBackground({ task_name: "job.data_backup", result: { backed, total: users.length }, @@ -130,6 +148,7 @@ export async function handleDataBackup( backed, total: users.length, users_failed: usersFailed, + heap_budget_trips: heapBudgetTrips, }); } catch (err) { evt.setError(err); diff --git a/tests/integration/backup-round-trip.test.ts b/tests/integration/backup-round-trip.test.ts index e35d7cd4c..7ecb17c1a 100644 --- a/tests/integration/backup-round-trip.test.ts +++ b/tests/integration/backup-round-trip.test.ts @@ -66,7 +66,8 @@ import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; import { UNREADABLE_EXPORT_MARKER } from "@/lib/export/unreadable-marker"; import { decryptNoteFromBytes } from "@/lib/labs/store"; import { decryptContextFromBytes } from "@/lib/labs/biomarker-store"; -import { packBackupBlob } from "@/lib/export/backup-blob"; +import { packBackupBlobStreaming } from "@/lib/export/backup-blob"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; import { TWO_ENDED_MODELS, type TwoEndedModel } from "@/lib/export/backup-plan"; import { POST } from "@/app/api/admin/backups/[id]/restore/route"; @@ -1252,10 +1253,36 @@ describe("every model the plan claims two-ended survives a real restore", () => "compare nothing against nothing and pass", ).toEqual([]); + // Pinned so the two writers below describe the same instant and can be + // compared byte for byte; `exportedAt` otherwise defaults to `new Date()`. + const exportedAt = new Date("2026-08-01T00:00:00.000Z"); const { payload } = await buildFullBackupPayload(prisma, OWNER_ID, { purpose: "disaster-recovery", + exportedAt, }); + // The writer the weekly job actually uses never builds this payload: it + // pulls the unbounded tables through page by page and serialises each row + // straight into the cipher, because materialising them is what used to + // exhaust the heap. Two writers is two chances to drift, so the file it + // produces is pinned against the one above — every per-table assertion + // below then covers BOTH, since the row that gets restored comes from the + // streamed bytes. + let streamedJson = ""; + await streamFullBackupJson( + prisma, + OWNER_ID, + (chunk) => { + streamedJson += chunk; + }, + { purpose: "disaster-recovery", exportedAt }, + ); + expect( + streamedJson, + "the streaming writer and the materialising builder disagree; the " + + "weekly backup no longer contains what this test proves restorable", + ).toBe(JSON.stringify(payload)); + // The tombstone has to be IN the file. A disaster-recovery payload that // carries only live rows is smaller and cheaper and wrong: the restore // wipes the account first, so a deletion that is not in the file is a @@ -1292,12 +1319,15 @@ describe("every model the plan claims two-ended survives a real restore", () => data: { userId: OWNER_ID, type: "TWO_ENDED_ROUND_TRIP", - // The envelope the weekly worker actually writes: gzip, then - // encrypt. The three cases further down deliberately keep the plain - // `encrypt(json)` form, which is what every row written before this - // envelope existed looks like — an operator's newest usable copy may - // well be one of those, so both arms have to reach the restore route. - data: packBackupBlob(JSON.stringify(payload)), + // The envelope the weekly worker actually writes: stream the JSON + // through gzip and the cipher without ever holding a whole copy. The + // three cases further down deliberately keep the plain `encrypt(json)` + // form, which is what every row written before any of this existed + // looks like — an operator's newest usable copy may well be one of + // those, so every arm has to reach the restore route. + data: await packBackupBlobStreaming(async (write) => { + await write(streamedJson); + }), }, }); const response = await POST( diff --git a/tests/integration/backup-streaming-memory.test.ts b/tests/integration/backup-streaming-memory.test.ts new file mode 100644 index 000000000..ed4b3b0f2 --- /dev/null +++ b/tests/integration/backup-streaming-memory.test.ts @@ -0,0 +1,231 @@ +/** + * The weekly backup, run against a record big enough to have killed the + * process, with its live memory measured rather than assumed. + * + * What this pins. The `data-backup` job used to build the whole payload as an + * object graph and then `JSON.stringify` it, so both the graph and the string + * were resident at once and both scaled with the record. On a seeded account + * of 445 000 measurements under `--max-old-space-size=450` that is + * `FATAL ERROR: Reached heap limit` about thirty seconds in — and because the + * job shares the app process, one account's size restarted the instance for + * everybody on it. + * + * Why it measures the way it does. `process.memoryUsage().heapUsed` on its own + * is not a measurement: V8 lets garbage float in proportion to the heap limit, + * and a test fork's limit is several times a container's, so the same code + * "peaks" at wildly different numbers depending on who is running it. Every + * reading below is taken after a forced collection, so what is compared is + * what each writer HOLDS. + * + * The two halves are the whole point. One says the streaming writer stays + * inside a budget; the other says the materialising builder does not fit that + * budget on the same fixture in the same process. Without the second, the + * budget could be any number at all and the first would still pass — green + * because nothing was measured rather than because something was proved. + * + * The fixture is seeded in one `INSERT … generate_series` rather than through + * Prisma. 120 000 rows through the client would dominate the runtime and prove + * nothing about the writer. + */ +import v8 from "node:v8"; +import vm from "node:vm"; + +import { beforeAll, afterAll, describe, expect, it } from "vitest"; + +import { + packBackupBlobStreaming, + unpackBackupBlob, +} from "@/lib/export/backup-blob"; +import { + BackupHeapBudgetExceededError, + streamFullBackupJson, +} from "@/lib/export/full-backup-stream"; +import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const OWNER_ID = "backup-streaming-owner"; +const MEASUREMENT_ROWS = 120_000; +const MOOD_ROWS = 8_000; + +/** + * What the streaming writer may hold on top of the process's own baseline. + * + * Measured, not chosen: on this fixture the streaming writer holds 16 MB — a + * page of rows and the base64 answer so far — and materialising the same + * fixture holds 168 MB, the object graph and the document at once. The budget + * sits with a wide margin either side of it, which is what makes it a test + * rather than a coin toss. + */ +const STREAM_BUDGET_BYTES = 48 * 1024 * 1024; + +const prisma = getPrismaClient(); + +/** + * A collection this process can ask for, without the runner having to be + * started with `--expose-gc`. The flag is turned on, the binding is pulled out + * of a throwaway context, and the flag is turned back off — so nothing about + * how the rest of the suite runs changes. + */ +const forceGc = ((): (() => void) => { + v8.setFlagsFromString("--expose-gc"); + const gc = vm.runInNewContext("gc") as () => void; + v8.setFlagsFromString("--no-expose-gc"); + return gc; +})(); + +/** Heap held after a forced collection. Garbage is not a measurement. */ +function liveHeapBytes(): number { + // Twice: the first pass frees the objects, the second collects what the + // first pass's own bookkeeping released. + forceGc(); + forceGc(); + return process.memoryUsage().heapUsed; +} + +async function seedLargeRecord(): Promise { + await prisma.user.create({ + data: { id: OWNER_ID, username: "backup-streaming-owner" }, + }); + // One statement, one round trip. Every row carries a note on one row in + // forty and ciphertext on one in twenty-five so the base64 arm of the + // serialiser is exercised at scale, and one in two hundred is a tombstone + // because the delta sync reads exactly those and a backup that drops them + // resurrects deleted readings on the next device sync. + await prisma.$executeRawUnsafe( + `INSERT INTO measurements ( + id, user_id, type, value, unit, source, measured_at, notes, + notes_encrypted, external_id, created_at, updated_at, sync_version, + deleted_at) + SELECT + 'sm' || lpad(g::text, 10, '0'), + $1, + 'PULSE'::measurement_type, + 60 + (g % 40), + 'bpm', + 'APPLE_HEALTH'::measurement_source, + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + CASE WHEN g % 40 = 0 THEN 'a note recorded with reading ' || g END, + CASE WHEN g % 25 = 0 + THEN decode(md5(g::text) || md5((g + 1)::text), 'hex') END, + 'stream-' || g, + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'), + 1, + CASE WHEN g % 200 = 0 + THEN timestamp '2026-01-01 00:00:00' + (g * interval '1 second') END + FROM generate_series(1, ${MEASUREMENT_ROWS}) AS g`, + OWNER_ID, + ); + await prisma.$executeRawUnsafe( + `INSERT INTO mood_entries ( + id, user_id, date, mood, score, source, mood_logged_at, synced_at, + created_at, updated_at, tz, note, sync_version, deleted_at) + SELECT + 'sd' || lpad(g::text, 10, '0'), + $1, + to_char(timestamp '2019-01-01' + (g * interval '1 hour'), 'YYYY-MM-DD'), + 'okay', 3, 'MOODLOG', + timestamp '2019-01-01' + (g * interval '1 hour'), + now(), now(), now(), 'Europe/Berlin', + CASE WHEN g % 3 = 0 THEN 'a journal line about day ' || g END, + 1, + CASE WHEN g % 150 = 0 THEN now() END + FROM generate_series(1, ${MOOD_ROWS}) AS g`, + OWNER_ID, + ); +} + +describe("weekly backup under a memory budget", () => { + beforeAll(async () => { + expect( + typeof forceGc, + "without a real collection every reading in this file measures " + + "uncollected garbage and nothing here can fail", + ).toBe("function"); + await truncateAllTables(prisma); + await seedLargeRecord(); + }, 240_000); + + afterAll(async () => { + await truncateAllTables(prisma); + }); + + it("writes a restorable blob while holding a bounded amount of the record", async () => { + const baseline = liveHeapBytes(); + let peakHeld = 0; + let chunks = 0; + + const blob = await packBackupBlobStreaming(async (write) => { + const counts = await streamFullBackupJson( + prisma, + OWNER_ID, + async (chunk) => { + await write(chunk); + // Sampled rather than continuous: a forced collection per chunk + // would dominate the runtime, and one in forty still lands inside + // every phase of the walk. + if (chunks++ % 40 === 0) { + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + } + }, + // The purpose the weekly job asks for: tombstones and ciphertext ride + // verbatim, which is the arm that has to fit in memory. + { purpose: "disaster-recovery" }, + ); + expect(counts.measurements).toBe(MEASUREMENT_ROWS); + expect(counts.moodEntries).toBe(MOOD_ROWS); + }); + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + + expect( + peakHeld, + `the streaming writer held ${Math.round(peakHeld / 1024 / 1024)} MB of ` + + "a record it is supposed to pass through a page at a time", + ).toBeLessThan(STREAM_BUDGET_BYTES); + + // A blob that writes but does not read is worse than none. + const restored = JSON.parse(unpackBackupBlob(blob)) as { + measurements: Array<{ deletedAt: string | null }>; + moodEntries: unknown[]; + }; + expect(restored.measurements).toHaveLength(MEASUREMENT_ROWS); + expect(restored.moodEntries).toHaveLength(MOOD_ROWS); + // Tombstones ride along, or the next device sync resurrects them. + expect( + restored.measurements.filter((row) => row.deletedAt !== null).length, + ).toBe(Math.floor(MEASUREMENT_ROWS / 200)); + }, 300_000); + + it("would not fit that budget if the record were materialised", async () => { + const baseline = liveHeapBytes(); + const { payload } = await buildFullBackupPayload(prisma, OWNER_ID, { + purpose: "disaster-recovery", + }); + const json = JSON.stringify(payload); + // Both are deliberately still reachable at the reading: holding the + // graph and the document at the same time is exactly the shape that + // exhausted the container. + const held = liveHeapBytes() - baseline; + expect(json.length).toBeGreaterThan(0); + expect(payload.measurements).toHaveLength(MEASUREMENT_ROWS); + expect( + held, + "materialising this fixture no longer costs what the budget in this " + + "file assumes; re-measure the budget rather than widening it", + ).toBeGreaterThan(STREAM_BUDGET_BYTES * 2); + }, 300_000); + + it("fails as a job rather than as a process when the record does not fit", async () => { + // The budget the writer enforces on itself is the reason an oversized + // account is now a failed backup for that account instead of a restart + // for every account on the host. + await expect( + packBackupBlobStreaming((write) => + streamFullBackupJson(prisma, OWNER_ID, write, { + purpose: "disaster-recovery", + heapCeilingBytes: 1, + }), + ), + ).rejects.toBeInstanceOf(BackupHeapBudgetExceededError); + }, 120_000); +});