From ac047854ba6725d2e715abfb9df99f971a740fea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 18 Oct 2026 13:10:19 +0200 Subject: [PATCH] fix(backup): stream documents and ECG traces through backup and restore The weekly copy read every stored document and ECG trace whole, then held their base64, the section JSON and the preview copies beside them. A vault of 190 MB peaked at 930 MB of heap and took a 524 MB process down on every run. Documents now leave the database one at a time in 1.5 MB slices, the traces a page at a time, every other section is written an element at a time, and a sealed piece never exceeds its chunk size. The restore keeps document bytes out of its first read and writes each document from a second one. Same file format, byte for byte. Refs #1239 --- docs/ops/backup-restore.md | 15 + src/lib/export/__tests__/backup-blob.test.ts | 12 + .../__tests__/full-backup-payload.test.ts | 4 +- src/lib/export/backup-blob.ts | 12 +- src/lib/export/ecg-backup.ts | 151 +++++--- src/lib/export/full-backup-payload.ts | 42 ++- src/lib/export/full-backup-stream.ts | 66 +++- src/lib/export/records-backup.ts | 341 ++++++++++++------ src/lib/export/restore-backup.ts | 120 +++--- src/lib/export/streamed-backup.ts | 80 +++- src/lib/export/streamed-field.ts | 53 +++ .../backup-document-streaming.test.ts | 279 ++++++++++++++ 12 files changed, 946 insertions(+), 229 deletions(-) create mode 100644 src/lib/export/streamed-field.ts create mode 100644 tests/integration/backup-document-streaming.test.ts diff --git a/docs/ops/backup-restore.md b/docs/ops/backup-restore.md index 83fd03ec3..09524b2c8 100644 --- a/docs/ops/backup-restore.md +++ b/docs/ops/backup-restore.md @@ -230,6 +230,21 @@ rather than by the size of the record going through it: measured on the same died and the new one finished holding tens of megabytes, writing a 9.1 MB object that restores to the identical record. +Up to 1.43.4 that held for the tables that are long in rows, not for the +ones that are heavy in bytes. The stored documents and the ECG traces were +still read whole, and a vault of 190 MB cost about five times that in heap: +the bytes, their base64, the section's JSON and the preview's copies of each +row. A single account with that many documents took a 524 MB process down on +every weekly run, while 1.25 million measurements beside them fitted easily. +Both writers now read the traces a page at a time and the documents one at a +time, each document's bytes in slices of 1.5 MB. Measured on that account +under `--max-old-space-size=524`, the peak went from 930 MB (a crash) to +209 MB. The restore had the same problem in the other direction: it kept +every document of the file in memory and peaked at 1.25 GB on the same copy. +It now leaves the stored bytes out of its first read and writes the +documents one at a time from a second one, and the copy restores every +document byte for byte inside the 524 MB heap. + One ceiling remains, and it is structural rather than a memory bound: a multipart upload carries 10 000 parts, so 80 GB is the largest object one account can produce. Past it the account's backup fails with a clear refusal, diff --git a/src/lib/export/__tests__/backup-blob.test.ts b/src/lib/export/__tests__/backup-blob.test.ts index bde829289..75a81bd64 100644 --- a/src/lib/export/__tests__/backup-blob.test.ts +++ b/src/lib/export/__tests__/backup-blob.test.ts @@ -196,6 +196,18 @@ describe("the copy in sealed pieces", () => { expect(readChunked(sealed)).toBe(json); }); + it("keeps every piece to its size when one write carries megabytes", async () => { + // A single write of a whole section used to become a single piece of the + // section's size, copied twice more on its way into the database (#1239). + const json = incompressibleJson(5 * BACKUP_CHUNK_BYTES); + const sealed = await packChunked(json, json.length); + expect(sealed.length).toBeGreaterThan(2); + for (const piece of sealed) { + expect(piece.byteLength).toBeLessThan(BACKUP_CHUNK_BYTES + 1024); + } + expect(readChunked(sealed)).toBe(json); + }); + it("always ends with a piece marked as the last, even an empty one", async () => { const sealed = await packChunked("{}"); expect(sealed).toHaveLength(1); diff --git a/src/lib/export/__tests__/full-backup-payload.test.ts b/src/lib/export/__tests__/full-backup-payload.test.ts index 71d627745..0f5f486d5 100644 --- a/src/lib/export/__tests__/full-backup-payload.test.ts +++ b/src/lib/export/__tests__/full-backup-payload.test.ts @@ -657,7 +657,9 @@ describe("buildFullBackupPayload disaster-recovery mode", () => { expect(mocks.buildRecordsBackupSection).toHaveBeenCalledWith( prisma, "user-1", - { purpose: "disaster-recovery" }, + // The materialising payload reads the documents itself; only the + // streaming writer defers them. + { purpose: "disaster-recovery", deferDocuments: false }, ); expect(payload).toMatchObject({ schemaVersion: "2", diff --git a/src/lib/export/backup-blob.ts b/src/lib/export/backup-blob.ts index 34c5480df..0a06a33a5 100644 --- a/src/lib/export/backup-blob.ts +++ b/src/lib/export/backup-blob.ts @@ -175,10 +175,16 @@ function chunkSealer( let seq = 0; let stored = 0; + // A piece carries at most `chunkBytes`. Before #1239 it carried everything + // pending, and one large write (a section's whole JSON) became one piece of + // hundreds of megabytes, copied twice more on its way into the database. const seal = async (last: boolean): Promise => { - const payload = pending.length === 1 ? pending[0]! : Buffer.concat(pending); - pending = []; - pendingBytes = 0; + const all = pending.length === 1 ? pending[0]! : Buffer.concat(pending); + const payload = + all.byteLength > chunkBytes ? all.subarray(0, chunkBytes) : all; + const rest = all.subarray(payload.byteLength); + pending = rest.byteLength > 0 ? [Buffer.from(rest)] : []; + pendingBytes = rest.byteLength; const sealed = sealBackupChunk(streamId, seq, last, payload); stored += sealed.byteLength; if (stored > limitBytes) { diff --git a/src/lib/export/ecg-backup.ts b/src/lib/export/ecg-backup.ts index c21aa2682..d570a3868 100644 --- a/src/lib/export/ecg-backup.ts +++ b/src/lib/export/ecg-backup.ts @@ -139,70 +139,135 @@ const ECG_RECORDING_BACKUP_SELECT = { updatedAt: true, } satisfies Prisma.EcgRecordingSelect; +type EcgRecordingBackupRow = Prisma.EcgRecordingGetPayload<{ + select: typeof ECG_RECORDING_BACKUP_SELECT; +}>; + +/** + * The order both readers below walk the strips in. Total, with the id after + * the instant, because the paged reader's cursor needs a total order and two + * strips recorded in the same second must not swap places between two + * exports of the same account. + */ +const ECG_RECORDING_ORDER = [ + { recordedAt: "asc" }, + { id: "asc" }, +] as const satisfies Prisma.EcgRecordingOrderByWithRelationInput[]; + +/** + * One strip as it rides the wire, or null for a portable export of a strip + * that will not decrypt (see the file header). Shared by both readers, so the + * paged one and the materialising one cannot drift. + */ +function serialiseEcgRecording( + row: EcgRecordingBackupRow, + disasterRecovery: boolean, +): EcgRecordingBackupEntry | null { + const common = { + source: row.source, + externalRecordingId: row.externalRecordingId, + recordedAt: row.recordedAt.toISOString(), + samplingFrequency: row.samplingFrequency, + sampleCount: row.sampleCount, + durationSeconds: row.durationSeconds, + lead: row.lead, + averageHeartRate: row.averageHeartRate, + rhythmClassification: row.rhythmClassification, + measurementId: row.measurementId, + createdAt: row.createdAt.toISOString(), + updatedAt: row.updatedAt.toISOString(), + }; + + if (disasterRecovery) { + return { + id: row.id, + waveformEncrypted: Buffer.from(row.waveformEncrypted).toString("base64"), + ...common, + }; + } + + // See the file header: a strip that will not decrypt is left out rather + // than written as a recording with no trace behind it. + try { + return { + waveform: decryptWaveformFromBytes(row.waveformEncrypted), + ...common, + }; + } catch (err) { + getEvent()?.addWarning( + `ECG waveform decrypt failed, recording omitted from portable export: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + return null; + } +} + +/** + * How many strips one page of the paged reader carries. A strip's trace is + * tens of kilobytes of ciphertext, so a page is a few megabytes. + */ +const ECG_PAGE_SIZE = 50; + +/** + * Every strip of the account, one serialised row at a time, read a page at a + * time. For the streaming backup writer, which must not hold every trace of a + * long-lived account at once (#1239). + */ +export async function* iterateBackupEcgRecordings( + prisma: Pick, + userId: string, + options: EcgBackupOptions = {}, +): AsyncGenerator { + const disasterRecovery = options.purpose === "disaster-recovery"; + let cursorId: string | null = null; + for (;;) { + const page: EcgRecordingBackupRow[] = await prisma.ecgRecording.findMany({ + where: { userId }, + orderBy: ECG_RECORDING_ORDER, + take: ECG_PAGE_SIZE, + ...(cursorId !== null ? { cursor: { id: cursorId }, skip: 1 } : {}), + select: ECG_RECORDING_BACKUP_SELECT, + }); + if (page.length === 0) return; + for (const row of page) { + const entry = serialiseEcgRecording(row, disasterRecovery); + if (entry) yield entry; + } + if (page.length < ECG_PAGE_SIZE) return; + cursorId = page[page.length - 1]!.id; + } +} + /** * Build the ECG slice of a user's full backup. * * Takes the delegate it uses rather than a whole client, matching the other * section builders. The table carries no tombstone column, so both purposes * read the same rows; they differ only in how the waveform travels. + * + * With `defer`, nothing is read and the slice comes back empty: the streaming + * writer pulls the strips through `iterateBackupEcgRecordings` itself. */ export async function buildEcgBackupSection( prisma: Pick, userId: string, - options: EcgBackupOptions = {}, + options: EcgBackupOptions & { defer?: boolean } = {}, ): Promise { + if (options.defer) return { ecgRecordings: [] }; const disasterRecovery = options.purpose === "disaster-recovery"; const rows = await prisma.ecgRecording.findMany({ where: { userId }, - orderBy: { recordedAt: "asc" }, + orderBy: ECG_RECORDING_ORDER, select: ECG_RECORDING_BACKUP_SELECT, }); const ecgRecordings: EcgRecordingBackupEntry[] = []; for (const row of rows) { - const common = { - source: row.source, - externalRecordingId: row.externalRecordingId, - recordedAt: row.recordedAt.toISOString(), - samplingFrequency: row.samplingFrequency, - sampleCount: row.sampleCount, - durationSeconds: row.durationSeconds, - lead: row.lead, - averageHeartRate: row.averageHeartRate, - rhythmClassification: row.rhythmClassification, - measurementId: row.measurementId, - createdAt: row.createdAt.toISOString(), - updatedAt: row.updatedAt.toISOString(), - }; - - if (disasterRecovery) { - ecgRecordings.push({ - id: row.id, - waveformEncrypted: Buffer.from(row.waveformEncrypted).toString( - "base64", - ), - ...common, - }); - continue; - } - - // See the file header: a strip that will not decrypt is left out rather - // than written as a recording with no trace behind it. - let waveform: number[]; - try { - waveform = decryptWaveformFromBytes(row.waveformEncrypted); - } catch (err) { - getEvent()?.addWarning( - `ECG waveform decrypt failed, recording omitted from portable export: ${ - err instanceof Error ? err.message : String(err) - }`, - ); - continue; - } - ecgRecordings.push({ waveform, ...common }); + const entry = serialiseEcgRecording(row, disasterRecovery); + if (entry) ecgRecordings.push(entry); } - return { ecgRecordings }; } diff --git a/src/lib/export/full-backup-payload.ts b/src/lib/export/full-backup-payload.ts index 2879e98f9..7702e7f96 100644 --- a/src/lib/export/full-backup-payload.ts +++ b/src/lib/export/full-backup-payload.ts @@ -13,8 +13,8 @@ * 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 + * with `deferBulk`, which leaves the unbounded tables (and the two whose rows + * carry stored bytes) 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 @@ -35,6 +35,7 @@ import { import { buildRecordsBackupSection, countRecordsBackupSection, + iterateBackupDocuments, type RecordsBackupCounts, type RecordsBackupSection, } from "@/lib/export/records-backup"; @@ -112,6 +113,7 @@ import { import { buildEcgBackupSection, countEcgBackupSection, + iterateBackupEcgRecordings, type EcgBackupCounts, type EcgBackupSection, } from "@/lib/export/ecg-backup"; @@ -196,9 +198,10 @@ export interface FullBackupOptions { purpose?: "portable-export" | "disaster-recovery"; exportedAt?: Date; /** - * Declare the three unbounded tables instead of reading them. + * Declare the large tables instead of reading them. * - * `measurements`, `intakeEvents` and `moodEntries` come back as + * `measurements`, `intakeEvents` and `moodEntries` (unbounded in rows) and + * `documents` and `ecgRecordings` (large in bytes) come back as * `DeferredRows` markers and their counts as 0. Only `streamFullBackupJson` * sets this; every other caller wants the arrays and gets them. */ @@ -686,13 +689,15 @@ 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>; + /** + * Open the row source. Called exactly once, by the streaming writer. A row + * is a plain object, or a `StreamedField` (`streamed-field.ts`) for a row + * with one field too large to hold as a string. + */ + rows(): AsyncIterable; } -function deferredRows( - rows: () => AsyncIterable>, -): DeferredRows { +function deferredRows(rows: () => AsyncIterable): DeferredRows { return { [DEFERRED_ROWS]: true, rows }; } @@ -836,6 +841,7 @@ export async function buildFullBackupPayload( }), buildRecordsBackupSection(prisma, userId, { purpose: disasterRecovery ? "disaster-recovery" : "portable-export", + deferDocuments: deferBulk, }), // Durable self-context, user-defined metrics, and persisted pattern // decisions. These account-owned rows cannot be reconstructed from an @@ -925,6 +931,7 @@ export async function buildFullBackupPayload( // arrangement as the sections above and for the same reason. buildEcgBackupSection(prisma, userId, { purpose: disasterRecovery ? "disaster-recovery" : "portable-export", + defer: deferBulk, }), // The person's own symptoms with their occurrences nested inside. Both // ends live in `src/lib/export/symptoms-backup.ts`. @@ -1019,6 +1026,22 @@ export async function buildFullBackupPayload( ) : moodEntries!, }; + // The two sections whose rows carry stored bytes: the documents (up to a + // hundred megabytes each) and the ECG traces. Deferred, they keep the + // position their section gives them in the payload (a key that already + // exists keeps its place when it is assigned again) and are read a document + // or a page of strips at a time (#1239). + const purpose = disasterRecovery ? "disaster-recovery" : "portable-export"; + const deferredBytes = deferBulk + ? { + documents: deferredRows(() => + iterateBackupDocuments(prisma, userId, { purpose }), + ), + ecgRecordings: deferredRows(() => + iterateBackupEcgRecordings(prisma, userId, { purpose }), + ), + } + : {}; // The clinical category lives in its own table rather than on the // medication row and is read through the one helper that normalises it. @@ -1421,6 +1444,7 @@ export async function buildFullBackupPayload( ...symptomsSection, ...lifeEventsSection, ...relationLinksSection, + ...deferredBytes, nutrientDays: nutrientDays.map((n) => ({ day: n.day, nutrient: n.nutrient, diff --git a/src/lib/export/full-backup-stream.ts b/src/lib/export/full-backup-stream.ts index 6db5689f7..4176e2846 100644 --- a/src/lib/export/full-backup-stream.ts +++ b/src/lib/export/full-backup-stream.ts @@ -14,9 +14,14 @@ * 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. + * one row at a time straight into the sink. The two sections whose rows carry + * stored bytes are deferred the same way (#1239): the ECG traces a page at a + * time, and the documents one at a time with each document's bytes arriving + * in slices (`StreamedField`), because a vault of 190 MB read whole cost about + * five times that in heap and took a 524 MB process down on its own. Every + * other section is written an element at a time and then RELEASED from the + * payload before the next one is touched, so the largest string this writer + * makes is one row's. * * 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 @@ -25,6 +30,8 @@ * The integration suite pins it against the materialising builder rather than * trusting that paragraph. */ +import { randomUUID } from "node:crypto"; + import type { PrismaClient } from "@/generated/prisma/client"; import { reportJobProgress } from "@/lib/jobs/job-observer"; import { @@ -33,6 +40,34 @@ import { type FullBackupCounts, type FullBackupOptions, } from "@/lib/export/full-backup-payload"; +import { + isStreamedField, + type StreamedField, +} from "@/lib/export/streamed-field"; + +/** + * Write a row whose one large field arrives in pieces, exactly as + * `JSON.stringify` writes the whole row: the row is serialised with a marker + * in the field's place, and the marker is replaced by the quoted pieces on the + * way out. The marker is random per call, so no value in the row can equal it. + */ +async function writeStreamedRow( + streamed: StreamedField, + write: (piece: string) => Promise, +): Promise { + const marker = `\u0000streamed:${randomUUID()}`; + const json = JSON.stringify({ ...streamed.row, [streamed.field]: marker }); + const quoted = JSON.stringify(marker); + const at = json.indexOf(quoted); + if (at < 0) { + throw new Error(`Streamed field ${streamed.field} is not in its row`); + } + await write(json.slice(0, at)); + await write('"'); + for await (const piece of streamed.pieces()) await write(piece); + await write('"'); + await write(json.slice(at + quoted.length)); +} /** Where the writer puts its pieces. Awaited, so a sink can apply backpressure. */ export type BackupJsonSink = (chunk: string) => void | Promise; @@ -139,10 +174,14 @@ export async function streamFullBackupJson( await write("["); let rows = 0; for await (const row of value.rows()) { - observe?.(key, row); - await write( - rows === 0 ? JSON.stringify(row) : `,${JSON.stringify(row)}`, - ); + if (rows > 0) await write(","); + if (isStreamedField(row)) { + observe?.(key, { ...row.row, [row.field]: row.observed }); + await writeStreamedRow(row, write); + } else { + observe?.(key, row); + await write(JSON.stringify(row)); + } rows++; if (rows % PROGRESS_EVERY_ROWS === 0) { reportJobProgress({ backup_section: key, backup_section_rows: rows }); @@ -150,6 +189,19 @@ export async function streamFullBackupJson( } await write("]"); bulkCounts[key] = rows; + } else if (Array.isArray(value)) { + // An element at a time, so the largest string this writer makes is one + // row's, never a section's: `JSON.stringify` of a whole section was a + // second copy of it on top of the rows (#1239). The same bytes either + // way; an element `JSON.stringify` cannot write becomes `null`, as it + // does inside an array. + observe?.(key, value); + await write("["); + for (let at = 0; at < value.length; at++) { + if (at > 0) await write(","); + await write(JSON.stringify(value[at]) ?? "null"); + } + await write("]"); } else { observe?.(key, value); await write(JSON.stringify(value)); diff --git a/src/lib/export/records-backup.ts b/src/lib/export/records-backup.ts index 239d3e5a3..eb191d378 100644 --- a/src/lib/export/records-backup.ts +++ b/src/lib/export/records-backup.ts @@ -33,6 +33,7 @@ import { type FamilyHistoryEntryDTO, } from "@/lib/records/dto"; import { dateOnlyKey } from "@/lib/tz/date-only"; +import { streamedField, type StreamedField } from "./streamed-field"; /** Disclosed in the payload manifest AND mirrored in the export UI copy. */ export const DOCUMENTS_MANIFEST_NOTE = @@ -226,21 +227,11 @@ export interface DocumentBackupEntry { export interface RecordsBackupOptions { purpose?: "portable-export" | "disaster-recovery"; -} - -interface DisasterRecoveryDocumentRow { - contentEncrypted: Uint8Array; - contentSha256: string | null; - contentCodec: string; - providerType: string | null; - errorReason: string | null; - summaryEncrypted: Uint8Array | null; - summaryGeneratedAt: Date | null; - summaryState: string; - lastIndexAttemptAt: Date | null; - lastIndexOutcome: string | null; - aiReadDeferred: boolean; - updatedAt: Date; + /** + * Leave the documents unread (`documents: []`): the streaming writer pulls + * them through `iterateBackupDocuments` itself, one at a time. + */ + deferDocuments?: boolean; } type CanonicalIllnessDayLog = IllnessDayLogDTO & { @@ -293,6 +284,208 @@ export interface RecordsBackupCounts { documents: number; } +/** The document columns both purposes carry. */ +const DOCUMENT_PORTABLE_SELECT = { + id: true, + kind: true, + title: true, + filename: true, + mimeType: true, + byteSize: true, + status: true, + reportDate: true, + documentDate: true, + summaryEncrypted: true, + sourceSystem: true, + sourceId: true, + sourceInstance: true, + createdAt: true, +} as const satisfies Prisma.InboundDocumentSelect; + +/** + * A disaster-recovery document without its stored bytes. The paged reader + * reads this for every document at once and the bytes one slice at a time. + */ +const DOCUMENT_DR_METADATA_SELECT = { + ...DOCUMENT_PORTABLE_SELECT, + contentSha256: true, + contentCodec: true, + providerType: true, + errorReason: true, + summaryGeneratedAt: true, + summaryState: true, + lastIndexAttemptAt: true, + lastIndexOutcome: true, + aiReadDeferred: true, + updatedAt: true, +} as const satisfies Prisma.InboundDocumentSelect; + +const DOCUMENT_DR_SELECT = { + ...DOCUMENT_DR_METADATA_SELECT, + contentEncrypted: true, +} as const satisfies Prisma.InboundDocumentSelect; + +/** Total, so two exports of the same account list the documents alike. */ +const DOCUMENT_ORDER = [ + { createdAt: "desc" }, + { id: "desc" }, +] as const satisfies Prisma.InboundDocumentOrderByWithRelationInput[]; + +type PortableDocumentRow = Prisma.InboundDocumentGetPayload<{ + select: typeof DOCUMENT_PORTABLE_SELECT; +}>; +type DisasterRecoveryDocumentRow = Prisma.InboundDocumentGetPayload<{ + select: typeof DOCUMENT_DR_METADATA_SELECT; +}>; +type DocumentRow = + | PortableDocumentRow + | Prisma.InboundDocumentGetPayload<{ select: typeof DOCUMENT_DR_SELECT }>; + +/** + * One document as it rides the wire. `contentEncrypted` is the base64 of the + * stored bytes for a disaster-recovery entry and null for a portable one, + * which carries the readable summary instead. Shared by the materialising + * reader and the paged one, so the two cannot drift. + */ +function serialiseDocument( + d: PortableDocumentRow | DisasterRecoveryDocumentRow, + contentEncrypted: string | null, +): DocumentBackupEntry { + const metadata = { + id: d.id, + kind: d.kind, + title: d.title, + filename: d.filename, + mimeType: d.mimeType, + byteSize: d.byteSize, + status: d.status, + reportDate: d.reportDate ? dateOnlyKey(d.reportDate) : null, + documentDate: d.documentDate ? dateOnlyKey(d.documentDate) : null, + sourceSystem: d.sourceSystem, + sourceId: d.sourceId, + sourceInstance: d.sourceInstance, + createdAt: d.createdAt.toISOString(), + }; + + if (contentEncrypted !== null && "contentCodec" in d) { + const dr: DisasterRecoveryDocumentRow = d; + return { + ...metadata, + reportDate: dr.reportDate?.toISOString() ?? null, + documentDate: dr.documentDate?.toISOString() ?? null, + updatedAt: dr.updatedAt.toISOString(), + contentEncrypted, + contentSha256: dr.contentSha256, + contentCodec: dr.contentCodec, + providerType: dr.providerType, + errorReason: dr.errorReason, + summaryEncrypted: dr.summaryEncrypted + ? Buffer.from(dr.summaryEncrypted).toString("base64") + : null, + summaryGeneratedAt: dr.summaryGeneratedAt?.toISOString() ?? null, + summaryState: dr.summaryState, + lastIndexAttemptAt: dr.lastIndexAttemptAt?.toISOString() ?? null, + lastIndexOutcome: dr.lastIndexOutcome, + aiReadDeferred: dr.aiReadDeferred, + }; + } + + return { + ...metadata, + summary: decryptSummarySoft(d.summaryEncrypted), + }; +} + +/** + * How much of a document's stored bytes one read takes. A multiple of three, + * so the base64 of every slice joins into the base64 of the whole. + */ +const DOCUMENT_SLICE_BYTES = 1536 * 1024; + +/** + * How much of a document's base64 the streaming writer shows its observer + * (the restore preview). The key id sits in the first characters; the cap is + * above the longest value the preview keeps as a key sample, so a shortened + * value is never kept as one. A multiple of four, so it still reads as base64. + */ +const DOCUMENT_OBSERVED_CHARS = 128 * 1024; + +/** + * Every document of the account, one at a time, for the streaming backup + * writer (#1239). + * + * A document is up to a hundred megabytes, and the materialising read holds + * every document's bytes, their base64 and the section's JSON at once: about + * five times the size of the vault in heap, which a 524 MB process does not + * have for a vault of 100 MB. Here the metadata of all documents is read + * once, and each document's bytes are read a slice at a time and handed on as + * base64 pieces (`streamedField`), so what is held is one slice. + * + * A document that disappears between the metadata read and its first slice + * is left out, as a later read would not have seen it. One that disappears + * halfway fails the copy: half a document is not a document. + */ +export async function* iterateBackupDocuments( + prisma: Pick, + userId: string, + options: RecordsBackupOptions = {}, +): AsyncGenerator { + if (options.purpose !== "disaster-recovery") { + const rows = await prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null }, + orderBy: DOCUMENT_ORDER, + select: DOCUMENT_PORTABLE_SELECT, + }); + for (const row of rows) yield serialiseDocument(row, null); + return; + } + const rows = await prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null }, + orderBy: DOCUMENT_ORDER, + select: DOCUMENT_DR_METADATA_SELECT, + }); + const slice = async (id: string, from: number): Promise => { + const [row] = await prisma.$queryRaw>` + SELECT substring(content_encrypted FROM ${from + 1}::int + FOR ${DOCUMENT_SLICE_BYTES}::int) AS part + FROM inbound_documents + WHERE id = ${id} AND user_id = ${userId} + `; + if (!row) return null; + const part = row.part ?? new Uint8Array(0); + // A view, not a copy: the slice is read once and dropped. + return Buffer.from(part.buffer, part.byteOffset, part.byteLength); + }; + for (const row of rows) { + const first = await slice(row.id, 0); + if (first === null) continue; + const head = first.toString("base64"); + const pieces = async function* (): AsyncGenerator { + yield head; + let last = first.byteLength; + for (let at = last; last === DOCUMENT_SLICE_BYTES; at += last) { + const next = await slice(row.id, at); + if (next === null) { + throw new Error( + `Document ${row.id} was removed while it was being backed up; ` + + "the next run copies the account again.", + ); + } + last = next.byteLength; + if (last > 0) yield next.toString("base64"); + } + }; + yield streamedField( + serialiseDocument(row, ""), + "contentEncrypted", + head.length > DOCUMENT_OBSERVED_CHARS + ? head.slice(0, DOCUMENT_OBSERVED_CHARS) + : head, + pieces, + ); + } +} + /** Decrypt a document's optional AI summary fail-soft (null on any error). */ function decryptSummarySoft(buf: Uint8Array | null): string | null { if (!buf || buf.byteLength === 0) return null; @@ -329,58 +522,19 @@ export async function buildRecordsBackupSection( options: RecordsBackupOptions = {}, ): Promise { const disasterRecovery = options.purpose === "disaster-recovery"; - const documentQuery = disasterRecovery - ? prisma.inboundDocument.findMany({ - where: { userId, deletedAt: null }, - orderBy: { createdAt: "desc" }, - select: { - id: true, - kind: true, - title: true, - filename: true, - mimeType: true, - byteSize: true, - contentEncrypted: true, - contentSha256: true, - contentCodec: true, - status: true, - providerType: true, - reportDate: true, - documentDate: true, - errorReason: true, - summaryEncrypted: true, - summaryGeneratedAt: true, - summaryState: true, - lastIndexAttemptAt: true, - lastIndexOutcome: true, - sourceSystem: true, - sourceId: true, - sourceInstance: true, - aiReadDeferred: true, - createdAt: true, - updatedAt: true, - }, - }) - : prisma.inboundDocument.findMany({ - where: { userId, deletedAt: null }, - orderBy: { createdAt: "desc" }, - select: { - id: true, - kind: true, - title: true, - filename: true, - mimeType: true, - byteSize: true, - status: true, - reportDate: true, - documentDate: true, - summaryEncrypted: true, - sourceSystem: true, - sourceId: true, - sourceInstance: true, - createdAt: true, - }, - }); + const documentQuery = options.deferDocuments + ? Promise.resolve([] as DocumentRow[]) + : disasterRecovery + ? prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null }, + orderBy: DOCUMENT_ORDER, + select: DOCUMENT_DR_SELECT, + }) + : prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null }, + orderBy: DOCUMENT_ORDER, + select: DOCUMENT_PORTABLE_SELECT, + }); const [ labResultRows, biomarkerRows, @@ -630,51 +784,14 @@ export async function buildRecordsBackupSection( externalId: w.externalId, })); - const documents: DocumentBackupEntry[] = documentRows.map((d) => { - const metadata = { - id: d.id, - kind: d.kind, - title: d.title, - filename: d.filename, - mimeType: d.mimeType, - byteSize: d.byteSize, - status: d.status, - reportDate: d.reportDate ? dateOnlyKey(d.reportDate) : null, - documentDate: d.documentDate ? dateOnlyKey(d.documentDate) : null, - sourceSystem: d.sourceSystem, - sourceId: d.sourceId, - sourceInstance: d.sourceInstance, - createdAt: d.createdAt.toISOString(), - }; - - if (disasterRecovery && "contentEncrypted" in d) { - const dr = d as typeof d & DisasterRecoveryDocumentRow; - return { - ...metadata, - reportDate: dr.reportDate?.toISOString() ?? null, - documentDate: dr.documentDate?.toISOString() ?? null, - updatedAt: dr.updatedAt.toISOString(), - contentEncrypted: Buffer.from(dr.contentEncrypted).toString("base64"), - contentSha256: dr.contentSha256, - contentCodec: dr.contentCodec, - providerType: dr.providerType, - errorReason: dr.errorReason, - summaryEncrypted: dr.summaryEncrypted - ? Buffer.from(dr.summaryEncrypted).toString("base64") - : null, - summaryGeneratedAt: dr.summaryGeneratedAt?.toISOString() ?? null, - summaryState: dr.summaryState, - lastIndexAttemptAt: dr.lastIndexAttemptAt?.toISOString() ?? null, - lastIndexOutcome: dr.lastIndexOutcome, - aiReadDeferred: dr.aiReadDeferred, - }; - } - - return { - ...metadata, - summary: decryptSummarySoft(d.summaryEncrypted), - }; - }); + const documents: DocumentBackupEntry[] = documentRows.map((d) => + serialiseDocument( + d, + disasterRecovery && "contentEncrypted" in d + ? Buffer.from(d.contentEncrypted).toString("base64") + : null, + ), + ); return { labResults, diff --git a/src/lib/export/restore-backup.ts b/src/lib/export/restore-backup.ts index c860002cf..7a2c43ce0 100644 --- a/src/lib/export/restore-backup.ts +++ b/src/lib/export/restore-backup.ts @@ -2561,59 +2561,77 @@ export async function restoreBackup( } if (payload.documents.length > 0) { - await tx.inboundDocument.createMany({ - data: payload.documents.map((document) => ({ - id: document.id, - userId: ownerId, - kind: document.kind as never, - title: document.title ?? null, - filename: document.filename ?? null, - mimeType: document.mimeType, - byteSize: document.byteSize, - contentEncrypted: decodeEncryptedBytes( - document.contentEncrypted!, - ), - contentSha256: document.contentSha256 ?? null, - contentCodec: document.contentCodec!, - status: document.status as never, - providerType: document.providerType ?? null, - reportDate: document.reportDate - ? new Date(document.reportDate) - : null, - documentDate: document.documentDate - ? new Date(document.documentDate) - : null, - errorReason: document.errorReason ?? null, - summaryEncrypted: document.summaryEncrypted - ? decodeEncryptedBytes(document.summaryEncrypted) - : null, - summaryGeneratedAt: document.summaryGeneratedAt - ? new Date(document.summaryGeneratedAt) - : null, - summaryState: document.summaryState as never, - lastIndexAttemptAt: document.lastIndexAttemptAt - ? new Date(document.lastIndexAttemptAt) - : null, - lastIndexOutcome: document.lastIndexOutcome ?? null, - // An older file has no marker: those documents predate the - // hold and restore as ordinary ones. - aiReadDeferred: document.aiReadDeferred ?? false, - sourceSystem: document.sourceSystem ?? null, - sourceId: document.sourceSystem - ? (document.sourceId ?? null) - : null, - // A file written before v1.39.3 has no instance: the key then - // matches any instance of its system, as it did when stored. - sourceInstance: - document.sourceSystem && document.sourceId - ? document.sourceInstance - ? normaliseSourceInstance(document.sourceInstance) - : null + // One document at a time, its stored bytes read from the file + // again (`forEachDocumentContent`): the first read left them out, + // because a vault held whole does not fit a 524 MB heap (#1239). + // The rows are in file order on both reads. + let written = 0; + await streamed.forEachDocumentContent(async (content, index) => { + const document = payload.documents[index]; + if (!document || content === undefined) { + throw new Error( + `Document ${index} of the backup has no stored bytes on the second read`, + ); + } + await tx.inboundDocument.create({ + data: { + id: document.id, + userId: ownerId, + kind: document.kind as never, + title: document.title ?? null, + filename: document.filename ?? null, + mimeType: document.mimeType, + byteSize: document.byteSize, + contentEncrypted: decodeEncryptedBytes(content), + contentSha256: document.contentSha256 ?? null, + contentCodec: document.contentCodec!, + status: document.status as never, + providerType: document.providerType ?? null, + reportDate: document.reportDate + ? new Date(document.reportDate) : null, - createdAt: new Date(document.createdAt!), - updatedAt: new Date(document.updatedAt!), - })), + documentDate: document.documentDate + ? new Date(document.documentDate) + : null, + errorReason: document.errorReason ?? null, + summaryEncrypted: document.summaryEncrypted + ? decodeEncryptedBytes(document.summaryEncrypted) + : null, + summaryGeneratedAt: document.summaryGeneratedAt + ? new Date(document.summaryGeneratedAt) + : null, + summaryState: document.summaryState as never, + lastIndexAttemptAt: document.lastIndexAttemptAt + ? new Date(document.lastIndexAttemptAt) + : null, + lastIndexOutcome: document.lastIndexOutcome ?? null, + // An older file has no marker: those documents predate the + // hold and restore as ordinary ones. + aiReadDeferred: document.aiReadDeferred ?? false, + sourceSystem: document.sourceSystem ?? null, + sourceId: document.sourceSystem + ? (document.sourceId ?? null) + : null, + // A file written before v1.39.3 has no instance: the key then + // matches any instance of its system, as it did when stored. + sourceInstance: + document.sourceSystem && document.sourceId + ? document.sourceInstance + ? normaliseSourceInstance(document.sourceInstance) + : null + : null, + createdAt: new Date(document.createdAt!), + updatedAt: new Date(document.updatedAt!), + }, + select: { id: true }, + }); + written += 1; }); + if (written !== payload.documents.length) { + throw new Error( + `The backup's second read found ${written} of ${payload.documents.length} documents`, + ); + } } // What the vault was filed against, and what was read out of it. diff --git a/src/lib/export/streamed-backup.ts b/src/lib/export/streamed-backup.ts index bdcd6dccf..7bf436291 100644 --- a/src/lib/export/streamed-backup.ts +++ b/src/lib/export/streamed-backup.ts @@ -8,6 +8,12 @@ * alone would not fit a 1 GB container beside the running app (#1031). Every * other section of the same file is a few megabytes. * + * The documents are the other one: a stored document is up to a hundred + * megabytes, and a vault of 190 MB held whole cost more heap than a 524 MB + * process has (#1239). The first read keeps each document without its stored + * bytes (`DOCUMENT_CONTENT_LEFT_OUT` in their place), and + * `forEachDocumentContent` reads them again one document at a time. + * * So the file is read twice. The first read keeps everything except the * measurements, and checks each measurement against the same element schema * the whole-file schema uses, counting them. Nothing is written on the @@ -44,7 +50,22 @@ async function loadMeasurementSchema(): Promise { export type BackupSource = () => AsyncIterable; const MEASUREMENTS = "measurements"; -const STREAMED = new Set([MEASUREMENTS]); +const DOCUMENTS = "documents"; +const STREAMED = new Set([MEASUREMENTS, DOCUMENTS]); +const MEASUREMENTS_ONLY = new Set([MEASUREMENTS]); +const DOCUMENTS_ONLY = new Set([DOCUMENTS]); + +/** + * What a document's `contentEncrypted` reads as in `raw`, where the stored + * bytes are left out (#1239): valid base64, so the whole-file schema still + * sees a document that carries content, and too short to be ciphertext, so + * nothing could mistake it for some. The restore writes the real value, from + * `forEachDocumentContent`. + */ +const DOCUMENT_CONTENT_LEFT_OUT = "AA=="; + +/** Base64, as the backup schema accepts it for stored bytes. */ +const BASE64 = /^[A-Za-z0-9+/]+={0,2}$/; /** A measurement the element schema refused, with where it is in the file. */ export class StreamedBackupInvalidError extends Error { @@ -78,6 +99,15 @@ export interface StreamedBackup { size: number, fn: (batch: BackupMeasurement[]) => Promise, ): Promise; + /** + * Read the file again and hand over each document's `contentEncrypted` + * (the base64 the first read left out of `raw`, or undefined where the + * file carries none) with the document's position, one document at a time. + * Awaits `fn` before reading further. + */ + forEachDocumentContent( + fn: (content: string | undefined, index: number) => Promise, + ): Promise; } function parseMeasurement( @@ -119,6 +149,7 @@ export async function readStreamedBackup( let firstWithoutId: number | null = null; let bytes = 0; const keys = new BackupKeyIdCollector(); + const documents: unknown[] = []; async function* counted() { for await (const chunk of source()) { bytes += chunk.byteLength; @@ -127,7 +158,11 @@ export async function readStreamedBackup( } const { document, streamedCounts } = await scanBackupJson(counted(), { streamKeys: STREAMED, - onElement: (_key, element, index) => { + onElement: (key, element, index) => { + if (key === DOCUMENTS) { + documents.push(withoutContent(element, index, keys)); + return; + } const row = parseMeasurement(schema, element, index); if (!row.id && firstWithoutId === null) firstWithoutId = index; keys.visit(element, MEASUREMENTS); @@ -147,11 +182,15 @@ export async function readStreamedBackup( ); } + if (DOCUMENTS in document) document[DOCUMENTS] = documents; keys.visit(document); const skipKeys = new Set( Object.keys(document).filter((key) => key !== MEASUREMENTS), ); + const documentSkipKeys = new Set( + Object.keys(document).filter((key) => key !== DOCUMENTS), + ); return { raw: document, @@ -161,7 +200,7 @@ export async function readStreamedBackup( async forEachMeasurementBatch(size, fn) { let batch: BackupMeasurement[] = []; await scanBackupJson(source(), { - streamKeys: STREAMED, + streamKeys: MEASUREMENTS_ONLY, skipKeys, onElement: async (_key, element, index) => { batch.push(parseMeasurement(schema, element, index)); @@ -174,5 +213,40 @@ export async function readStreamedBackup( }); if (batch.length > 0) await fn(batch); }, + async forEachDocumentContent(fn) { + await scanBackupJson(source(), { + streamKeys: DOCUMENTS_ONLY, + skipKeys: documentSkipKeys, + onElement: async (_key, element, index) => { + const content = (element as { contentEncrypted?: unknown }) + ?.contentEncrypted; + await fn(typeof content === "string" ? content : undefined, index); + }, + }); + }, }; } + +/** + * A document as `raw` keeps it: its stored bytes replaced by + * `DOCUMENT_CONTENT_LEFT_OUT`, after the key they were written under is + * recorded and their form is checked, since the whole-file schema only sees + * the stand-in. + */ +function withoutContent( + element: unknown, + index: number, + keys: BackupKeyIdCollector, +): unknown { + if (!element || typeof element !== "object") return element; + const { contentEncrypted, ...rest } = element as Record; + if (typeof contentEncrypted !== "string") return element; + if (contentEncrypted.length % 4 !== 0 || !BASE64.test(contentEncrypted)) { + throw new StreamedBackupInvalidError( + `${DOCUMENTS}.${index}.contentEncrypted`, + "Expected base64-encoded encrypted bytes", + ); + } + keys.add(DOCUMENTS, contentEncrypted, "contentEncrypted"); + return { ...rest, contentEncrypted: DOCUMENT_CONTENT_LEFT_OUT }; +} diff --git a/src/lib/export/streamed-field.ts b/src/lib/export/streamed-field.ts new file mode 100644 index 000000000..a2ed200a0 --- /dev/null +++ b/src/lib/export/streamed-field.ts @@ -0,0 +1,53 @@ +/** + * A backup row with one field too large to hold as a string. + * + * A stored document is up to a hundred megabytes, and its base64 in the backup + * is a third larger. Built as a string, then serialised with the row, then + * checked by the restore preview, one such field is several copies of itself + * in a heap of 524 MB (#1239). A row of this kind carries the field as pieces + * instead: the streaming writer (`full-backup-stream.ts`) writes the rest of + * the row as `JSON.stringify` would, and the field's pieces in its place, so + * the copy reads exactly as if the field had been a string. + * + * The pieces must not need JSON escaping (base64 does not), because they are + * written between the quotes as they come. + */ +const STREAMED_FIELD = Symbol("healthlog.backup.streamedField"); + +export interface StreamedField { + readonly [STREAMED_FIELD]: true; + /** The row, with the streamed field present (its value is ignored). */ + readonly row: Record; + /** Which member of `row` is written from `pieces`. */ + readonly field: string; + /** + * What an observer of the writer is shown in the field's place: the start + * of the value, or the whole value when it is short. + */ + readonly observed: string; + /** The field's value, in order. Opened once, by the writer. */ + readonly pieces: () => AsyncIterable; +} + +export function streamedField( + row: object, + field: string, + observed: string, + pieces: () => AsyncIterable, +): StreamedField { + return { + [STREAMED_FIELD]: true, + row: row as Record, + field, + observed, + pieces, + }; +} + +export function isStreamedField(value: unknown): value is StreamedField { + return ( + typeof value === "object" && + value !== null && + (value as Record)[STREAMED_FIELD] === true + ); +} diff --git a/tests/integration/backup-document-streaming.test.ts b/tests/integration/backup-document-streaming.test.ts new file mode 100644 index 000000000..63c087d9d --- /dev/null +++ b/tests/integration/backup-document-streaming.test.ts @@ -0,0 +1,279 @@ +/** + * The weekly backup of an account whose weight is in its stored bytes: the + * document vault and the ECG traces (#1239). + * + * What this pins. The streaming writer pulled the three tables that are large + * in ROWS through a page at a time, and read every other section whole. The + * documents are large in BYTES instead (up to a hundred megabytes each), and + * whole they cost about five times the vault in heap: the stored bytes, their + * base64, the section's JSON, and the preview's copies of each row. A single + * account with 190 MB of documents took a 524 MB process down every time its + * weekly copy ran, while the measurements beside them (1.25 million rows) + * fitted easily. Here the documents are read one at a time and each one's + * bytes a slice at a time, and the traces a page at a time. + * + * Four things are proved on one fixture: the streamed copy is byte for byte + * what the materialising builder writes, including documents that span + * several slices and one that ends exactly on a slice boundary; the preview + * taken from the writer agrees with a full read; the writer holds a bounded + * amount while the materialised form does not fit that bound; and the copy + * restores every document and strip exactly. + */ +import { createHash, randomBytes } from "node:crypto"; +import v8 from "node:v8"; +import vm from "node:vm"; + +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { readStoredBackup } from "./stored-backup-read"; +import { encryptDocumentContent } from "@/lib/documents/store"; +import { encryptWaveformToBytes } from "@/lib/withings/ecg-waveform-codec"; +import { BackupKeyIdCollector } from "@/lib/export/backup-key-ids"; +import { createBackupPreviewCollector } from "@/lib/export/backup-preview"; +import { buildFullBackupPayload } from "@/lib/export/full-backup-payload"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; +import { restoreBackup } from "@/lib/export/restore-backup"; +import { storeBackupBlob } from "@/lib/export/store-backup-blob"; +import { STORED_BACKUP_SELECT } from "@/lib/export/stored-backup"; +import { parseBackupPayload } from "@/lib/validations/backup"; +import { summarizeBackup } from "@/lib/validations/backup-summary"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const prisma = getPrismaClient(); +const OWNER_ID = "backup-document-owner"; + +/** The slice the document reader takes (`DOCUMENT_SLICE_BYTES`). */ +const SLICE = 1536 * 1024; +/** Ten documents of four megabytes: three slices each, the last one short. */ +const LARGE_DOCUMENTS = 10; +const LARGE_BYTES = 4 * 1024 * 1024; +/** More strips than one page of the trace reader (50). */ +const ECG_STRIPS = 120; + +/** + * What the writer may hold on top of the process's baseline. Measured on this + * fixture: the streaming writer holds 4 MB (one slice, its base64, a page of + * strips), and materialising it holds 61 MB. The bound sits with a wide + * margin either side. + */ +const STREAM_BUDGET_BYTES = 16 * 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; +} + +function reportPeak(label: string, heldBytes: number): void { + process.stderr.write( + `[backup-document-memory] ${label}: ${Math.round(heldBytes / 1024 / 1024)} MB held, ` + + `${Math.round(v8.getHeapStatistics().heap_size_limit / 1024 / 1024)} MB heap limit\n`, + ); +} + +/** The ciphertext overhead of the active document codec, in bytes. */ +const CODEC_OVERHEAD = + encryptDocumentContent(Buffer.alloc(1)).content.byteLength - 1; + +async function addDocument(index: number, cipherBytes: number): Promise { + const { content, codec } = encryptDocumentContent( + randomBytes(Math.max(1, cipherBytes - CODEC_OVERHEAD)), + ); + await prisma.inboundDocument.create({ + data: { + id: `doc-${String(index).padStart(3, "0")}`, + userId: OWNER_ID, + filename: `scan-${index}.pdf`, + mimeType: "application/pdf", + byteSize: content.byteLength, + contentEncrypted: content, + contentCodec: codec, + contentSha256: createHash("sha256").update(content).digest("hex"), + title: `Scan ${index}`, + // Two documents share an instant, so the order needs its tiebreak. + createdAt: new Date(Date.UTC(2026, 0, 1 + Math.min(index, 5))), + }, + }); +} + +async function seed(): Promise { + await prisma.user.create({ + data: { id: OWNER_ID, username: "backup-document-owner", role: "ADMIN" }, + }); + for (let i = 0; i < LARGE_DOCUMENTS; i++) await addDocument(i, LARGE_BYTES); + // Exactly two slices: the reader has to stop on an empty third read. + await addDocument(LARGE_DOCUMENTS, 2 * SLICE); + // One short enough that the preview sees the whole value. + await addDocument(LARGE_DOCUMENTS + 1, 200); + const samples = Array.from({ length: 2_000 }, (_, i) => + Math.round(Math.sin(i / 9) * 800), + ); + await prisma.ecgRecording.createMany({ + data: Array.from({ length: ECG_STRIPS }, (_, i) => ({ + id: `ecg-${String(i).padStart(4, "0")}`, + userId: OWNER_ID, + source: "APPLE_HEALTH" as const, + externalRecordingId: `strip-${i}`, + // Pairs share an instant (at two rates, the natural key), so the paged + // order needs its tiebreak. + recordedAt: new Date(Date.UTC(2025, 0, 1 + Math.floor(i / 2))), + waveformEncrypted: encryptWaveformToBytes(samples), + samplingFrequency: i % 2 === 0 ? 512 : 256, + sampleCount: samples.length, + })), + }); +} + +async function documentsOf(userId: string) { + const rows = await prisma.inboundDocument.findMany({ + where: { userId }, + orderBy: { id: "asc" }, + select: { id: true, byteSize: true, contentEncrypted: true, title: true }, + }); + return rows.map(({ contentEncrypted, ...row }) => ({ + ...row, + sha: createHash("sha256").update(contentEncrypted).digest("hex"), + })); +} + +async function stripsOf(userId: string) { + const rows = await prisma.ecgRecording.findMany({ + where: { userId }, + orderBy: { id: "asc" }, + select: { id: true, recordedAt: true, waveformEncrypted: true }, + }); + return rows.map(({ waveformEncrypted, ...row }) => ({ + ...row, + waveform: Buffer.from(waveformEncrypted).toString("base64"), + })); +} + +describe("weekly backup of an account heavy in stored bytes", () => { + beforeAll(async () => { + expect(typeof forceGc).toBe("function"); + await truncateAllTables(prisma); + await seed(); + }, 240_000); + + afterAll(async () => { + await truncateAllTables(prisma); + }); + + it("writes what the materialising builder writes, and previews it alike", async () => { + const exportedAt = new Date("2026-08-01T00:00:00.000Z"); + let streamed = ""; + const preview = await createBackupPreviewCollector(); + const counts = await streamFullBackupJson( + prisma, + OWNER_ID, + (chunk) => { + streamed += chunk; + }, + { purpose: "disaster-recovery", exportedAt, observe: preview.observe }, + ); + expect(counts.documents).toBe(LARGE_DOCUMENTS + 2); + expect(counts.ecgRecordings).toBe(ECG_STRIPS); + + const { payload } = await buildFullBackupPayload(prisma, OWNER_ID, { + purpose: "disaster-recovery", + exportedAt, + }); + expect( + streamed === JSON.stringify(payload), + "the streaming writer and the materialising builder disagree", + ).toBe(true); + + const fromWriter = preview.finish(counts); + expect(fromWriter).not.toBeNull(); + expect(fromWriter!.summary).toEqual( + summarizeBackup(parseBackupPayload(streamed)), + ); + const readBack = new BackupKeyIdCollector(); + readBack.visit(JSON.parse(streamed)); + expect(fromWriter!.keys.toStored()).toEqual(readBack.toStored()); + }, 300_000); + + it("holds a bounded amount of the vault while storing it", async () => { + const baseline = liveHeapBytes(); + let peakHeld = 0; + let pieces = 0; + await storeBackupBlob( + prisma, + { userId: OWNER_ID, type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson( + prisma, + OWNER_ID, + async (chunk) => { + await write(chunk); + if (pieces++ % 8 === 0) { + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + } + }, + { purpose: "disaster-recovery" }, + ), + ); + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + reportPeak("streamed", peakHeld); + expect(peakHeld).toBeLessThan(STREAM_BUDGET_BYTES); + + // The other half: the same fixture, materialised, does not fit the bound, + // so the bound above measures something. + const before = liveHeapBytes(); + const { payload } = await buildFullBackupPayload(prisma, OWNER_ID, { + purpose: "disaster-recovery", + }); + const json = JSON.stringify(payload); + const held = liveHeapBytes() - before; + reportPeak("materialised", held); + expect(json.length).toBeGreaterThan(LARGE_DOCUMENTS * LARGE_BYTES); + expect(held).toBeGreaterThan(STREAM_BUDGET_BYTES * 2); + }, 300_000); + + it("restores every document and strip exactly from the stored copy", async () => { + const documents = await documentsOf(OWNER_ID); + const strips = await stripsOf(OWNER_ID); + const { id } = await storeBackupBlob( + prisma, + { userId: OWNER_ID, type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, OWNER_ID, write, { + purpose: "disaster-recovery", + }), + ); + const copy = JSON.parse(await readStoredBackup(prisma, id)) as { + documents: Array<{ id: string; contentEncrypted: string }>; + }; + expect(copy.documents).toHaveLength(LARGE_DOCUMENTS + 2); + + await prisma.inboundDocument.deleteMany({ where: { userId: OWNER_ID } }); + await prisma.ecgRecording.deleteMany({ where: { userId: OWNER_ID } }); + const backup = await prisma.dataBackup.findUniqueOrThrow({ + where: { id }, + select: STORED_BACKUP_SELECT, + }); + const outcome = await restoreBackup({ + backup, + actorUserId: OWNER_ID, + ipAddress: null, + restoreInstanceSettings: false, + }); + expect(outcome.ok, JSON.stringify(outcome)).toBe(true); + expect(await documentsOf(OWNER_ID)).toEqual(documents); + expect(await stripsOf(OWNER_ID)).toEqual(strips); + }, 300_000); +});