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