diff --git a/src/lib/export/backup-domains.ts b/src/lib/export/backup-domains.ts index f013237e5..c16788122 100644 --- a/src/lib/export/backup-domains.ts +++ b/src/lib/export/backup-domains.ts @@ -420,6 +420,7 @@ export const RESTORE_STEPS: readonly RestoreStep[] = [ ctx.payload, ctx.links.writtenReferencedMeasurementIds, ctx.skips, + (fn) => ctx.streamed.forEachEcgTrace(fn), ), ); }, diff --git a/src/lib/export/ecg-backup.ts b/src/lib/export/ecg-backup.ts index d570a3868..c777a7a93 100644 --- a/src/lib/export/ecg-backup.ts +++ b/src/lib/export/ecg-backup.ts @@ -75,6 +75,7 @@ import { recordUnknownKeys, type RestoreSkipLog, } from "@/lib/export/restore-skips"; +import type { EcgTrace } from "@/lib/export/streamed-backup"; export interface EcgBackupOptions { purpose?: "portable-export" | "disaster-recovery"; @@ -278,6 +279,11 @@ export function countEcgBackupSection( return { ecgRecordings: section.ecgRecordings.length }; } +/** Hands over each strip's trace from a second read of the file. */ +export type EcgTraceSource = ( + fn: (trace: EcgTrace, index: number) => Promise, +) => Promise; + /** Counts the ECG restore wiped, for the audit trail. */ export interface EcgRestoreCleared { ecgRecordings: number; @@ -351,13 +357,37 @@ export async function restoreEcgData( payload: EcgRestoreInput, measurementIds: ReadonlySet, skips: RestoreSkipLog, + forEachTrace: EcgTraceSource, ): Promise { const cleared = await tx.ecgRecording.deleteMany({ where: { userId: ownerId }, }); const droppedMeasurementRefs: string[] = []; - for (const recording of payload.ecgRecordings) { + // The traces come from a second read of the file, one strip at a time + // (`forEachEcgTrace`): the first read kept each strip without its trace, + // because a long ECG history held whole is a vault of its own. The strips + // are in file order on both reads. + let traces = 0; + if (payload.ecgRecordings.length > 0) { + await forEachTrace(async (trace, index) => { + traces++; + const listed = payload.ecgRecordings[index]; + if (!listed) { + throw new Error( + `ECG recording ${index} of the backup is missing from the first read`, + ); + } + await writeRecording({ ...listed, ...trace }); + }); + } + if (traces !== payload.ecgRecordings.length) { + throw new Error( + `The backup lists ${payload.ecgRecordings.length} ECG recordings and the second read found ${traces}`, + ); + } + + async function writeRecording(recording: RestoredEcgRecording) { let measurementId = recording.measurementId ?? null; if (measurementId && !measurementIds.has(measurementId)) { droppedMeasurementRefs.push(measurementId); diff --git a/src/lib/export/streamed-backup.ts b/src/lib/export/streamed-backup.ts index 8e5e7dd29..60b7c301a 100644 --- a/src/lib/export/streamed-backup.ts +++ b/src/lib/export/streamed-backup.ts @@ -12,7 +12,9 @@ * 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. + * `forEachDocumentContent` reads them again one document at a time. The ECG + * strips are kept without their traces the same way, and `forEachEcgTrace` + * reads those again one strip at a time. * * The other sections that grow with a record (`BACKUP_STREAMED_SECTIONS`) are * read an element at a time too and kept as they are, so no section of the @@ -60,21 +62,22 @@ export type BackupSource = () => AsyncIterable; const MEASUREMENTS = "measurements"; const DOCUMENTS = "documents"; +const ECG_RECORDINGS = "ecgRecordings"; /** * The sections a backup reader takes an element at a time, never whole: the * ones that grow with a record, in rows (measurements, intake events, mood * entries) or in stored bytes (documents, ECG traces). How the first read * treats each: the measurements are checked and counted, not kept; the - * documents are kept without their stored bytes; the rest are kept element - * by element. + * documents are kept without their stored bytes and the ECG strips without + * their traces; the rest are kept element by element. */ export const BACKUP_STREAMED_SECTIONS = [ MEASUREMENTS, "intakeEvents", "moodEntries", DOCUMENTS, - "ecgRecordings", + ECG_RECORDINGS, ] as const; /** `BACKUP_STREAMED_SECTIONS` as `scanBackupJson` takes it. */ @@ -84,6 +87,7 @@ export const BACKUP_STREAM_KEYS: ReadonlySet = new Set( const MEASUREMENTS_ONLY = new Set([MEASUREMENTS]); const DOCUMENTS_ONLY = new Set([DOCUMENTS]); +const ECG_ONLY = new Set([ECG_RECORDINGS]); /** * What a document's `contentEncrypted` reads as in `raw`, where the stored @@ -94,6 +98,21 @@ const DOCUMENTS_ONLY = new Set([DOCUMENTS]); */ const DOCUMENT_CONTENT_LEFT_OUT = "AA=="; +/** + * What an ECG strip's trace reads as in `raw`: the stored ciphertext as the + * same short stand-in, or an empty sample array on a portable file. The + * restore writes the real trace, from `forEachEcgTrace`. + */ +const ECG_CIPHERTEXT_LEFT_OUT = "AA=="; + +/** One strip's trace, as the second read hands it over. */ +export interface EcgTrace { + /** The stored ciphertext as base64 (a disaster-recovery file). */ + waveformEncrypted?: string; + /** The micro-volt samples (a portable file). */ + waveform?: number[]; +} + /** Base64, as the backup schema accepts it for stored bytes. */ const BASE64 = /^[A-Za-z0-9+/]+={0,2}$/; @@ -138,6 +157,15 @@ export interface StreamedBackup { forEachDocumentContent( fn: (content: string | undefined, index: number) => Promise, ): Promise; + /** + * Read the file again and hand over each ECG strip's trace (the stored + * ciphertext or the sample array the first read left out of `raw`) with + * the strip's position, one strip at a time. Awaits `fn` before reading + * further. + */ + forEachEcgTrace( + fn: (trace: EcgTrace, index: number) => Promise, + ): Promise; } function parseMeasurement( @@ -200,7 +228,11 @@ export async function readStreamedBackup( let section = kept.get(key); if (!section) kept.set(key, (section = [])); section.push( - key === DOCUMENTS ? withoutContent(element, index, keys) : element, + key === DOCUMENTS + ? withoutContent(element, index, keys) + : key === ECG_RECORDINGS + ? withoutTrace(element, index, keys) + : element, ); }, }); @@ -228,6 +260,9 @@ export async function readStreamedBackup( const documentSkipKeys = new Set( Object.keys(document).filter((key) => key !== DOCUMENTS), ); + const ecgSkipKeys = new Set( + Object.keys(document).filter((key) => key !== ECG_RECORDINGS), + ); return { raw: document, @@ -261,9 +296,74 @@ export async function readStreamedBackup( }, }); }, + async forEachEcgTrace(fn) { + await scanBackupJson(source(), { + streamKeys: ECG_ONLY, + skipKeys: ecgSkipKeys, + onElement: async (_key, element, index) => { + const { waveformEncrypted, waveform } = (element ?? {}) as EcgTrace; + await fn( + { + ...(typeof waveformEncrypted === "string" + ? { waveformEncrypted } + : {}), + ...(Array.isArray(waveform) ? { waveform } : {}), + }, + index, + ); + }, + }); + }, }; } +/** + * An ECG strip as `raw` keeps it: its trace replaced by a stand-in, after the + * key the ciphertext was written under is recorded and the trace's form is + * checked, since the whole-file schema only sees the stand-in. A long Withings + * or Apple Watch history is hundreds of strips of tens to hundreds of + * kilobytes each; held whole it is the next vault. + */ +function withoutTrace( + element: unknown, + index: number, + keys: BackupKeyIdCollector, +): unknown { + if (!element || typeof element !== "object") return element; + const { waveformEncrypted, waveform, ...rest } = element as Record< + string, + unknown + >; + if (typeof waveformEncrypted === "string") { + if ( + waveformEncrypted.length === 0 || + waveformEncrypted.length % 4 !== 0 || + !BASE64.test(waveformEncrypted) + ) { + throw new StreamedBackupInvalidError( + `${ECG_RECORDINGS}.${index}.waveformEncrypted`, + "Expected base64-encoded encrypted bytes", + ); + } + keys.add(ECG_RECORDINGS, waveformEncrypted, "waveformEncrypted"); + return { + ...rest, + waveformEncrypted: ECG_CIPHERTEXT_LEFT_OUT, + ...(waveform === undefined ? {} : { waveform }), + }; + } + if (Array.isArray(waveform)) { + if (!waveform.every((sample) => typeof sample === "number")) { + throw new StreamedBackupInvalidError( + `${ECG_RECORDINGS}.${index}.waveform`, + "Expected an array of numbers", + ); + } + return { ...rest, waveform: [] }; + } + return element; +} + /** * A document as `raw` keeps it: its stored bytes replaced by * `DOCUMENT_CONTENT_LEFT_OUT`, after the key they were written under is diff --git a/tests/integration/backup-upload-document-memory.test.ts b/tests/integration/backup-upload-document-memory.test.ts index 413b712a1..c3221d22f 100644 --- a/tests/integration/backup-upload-document-memory.test.ts +++ b/tests/integration/backup-upload-document-memory.test.ts @@ -25,6 +25,12 @@ * `--max-old-space-size=524` it ran out of heap; the streamed route holds a * few megabytes. Run it under that flag for the production ceiling: * `NODE_OPTIONS=--max-old-space-size=524`. + * + * The second file is the same shape with ECG strips in place of documents: a + * long watch history, every strip's trace stored as ciphertext. The shared + * reader kept the strips whole, traces included, after it had stopped doing + * so for the documents; it now keeps them without their traces too, and the + * restore reads the traces again one strip at a time. */ import { randomBytes } from "node:crypto"; import { createReadStream, createWriteStream } from "node:fs"; @@ -50,6 +56,12 @@ const prisma = getPrismaClient(); /** A hundred documents of 2.75 MiB of base64 each: about 275 MB of vault. */ const DOCUMENTS = 100; const DOCUMENT_BYTES = 2 * 1024 * 1024; +/** + * A hundred and fifty strips of 100,000 samples each, about 1.4 MB of stored + * trace per strip once in the file: about 210 MB of ECG history. + */ +const ECG_STRIPS = 150; +const ECG_SAMPLES = 100_000; /** Readings after the vault, so the route is sampled once it has gone past. */ const MEASUREMENTS = 30_000; /** Take a reading every this many chunks of the request body. */ @@ -78,12 +90,17 @@ function liveHeapBytes(): number { const mb = (bytes: number): number => Math.round(bytes / 1024 / 1024); let directory = ""; -let filePath = ""; -let fileBytes = 0; -/** Where the documents section ends in the file, in bytes. */ -let vaultEndsAt = 0; let adminId = ""; +/** A file on disk, and where its heavy section ends in it, in bytes. */ +interface HeavyFile { + path: string; + bytes: number; + heavyEndsAt: number; +} +let documentFile: HeavyFile; +let ecgFile: HeavyFile; + /** Write one piece of the file, waiting when the stream asks for it. */ async function put( out: ReturnType, @@ -93,20 +110,19 @@ async function put( } /** - * The backup file, written a piece at a time so the test never holds it: the - * documents first, each carrying the same real document ciphertext (the key - * check reads its key id), then the readings. + * A backup file, written a piece at a time so the test never holds it: the + * heavy section first (`heavy` writes its elements), then the readings. */ -async function writeBackupFile(ownerId: string): Promise { - const { encryptDocumentContent } = await import("@/lib/documents/store"); +async function writeBackupFile( + path: string, + ownerId: string, + section: string, + count: number, + element: (index: number) => unknown, +): Promise { const { BACKUP_SCHEMA_VERSION } = await import("@/lib/validations/backup-summary"); - const { content, codec } = encryptDocumentContent( - randomBytes(DOCUMENT_BYTES), - ); - const base64 = Buffer.from(content).toString("base64"); - - const out = createWriteStream(filePath); + const out = createWriteStream(path); let written = 0; const write = async (text: string) => { written += Buffer.byteLength(text); @@ -121,25 +137,12 @@ async function writeBackupFile(ownerId: string): Promise { moodEntries: [], }; await write(JSON.stringify(head).slice(0, -1)); - await write(',"documents":['); - for (let i = 0; i < DOCUMENTS; i++) { - const document = { - id: `upload-memory-doc-${i}`, - kind: "DOCTOR_REPORT", - title: `Report ${i}`, - filename: `report-${i}.pdf`, - mimeType: "application/pdf", - byteSize: content.byteLength, - status: "STORED", - contentEncrypted: base64, - contentCodec: codec, - createdAt: "2026-01-01T00:00:00.000Z", - updatedAt: "2026-01-01T00:00:00.000Z", - }; - await write((i === 0 ? "" : ",") + JSON.stringify(document)); + await write(`,"${section}":[`); + for (let i = 0; i < count; i++) { + await write((i === 0 ? "" : ",") + JSON.stringify(element(i))); } await write("]"); - vaultEndsAt = written; + const heavyEndsAt = written; await write(',"measurements":['); for (let i = 0; i < MEASUREMENTS; i++) { const reading = { @@ -156,7 +159,68 @@ async function writeBackupFile(ownerId: string): Promise { await write("]}"); out.end(); await once(out, "finish"); - fileBytes = (await stat(filePath)).size; + return { path, bytes: (await stat(path)).size, heavyEndsAt }; +} + +/** Documents, each carrying the same real document ciphertext. */ +async function writeDocumentFile(ownerId: string): Promise { + const { encryptDocumentContent } = await import("@/lib/documents/store"); + const { content, codec } = encryptDocumentContent( + randomBytes(DOCUMENT_BYTES), + ); + const base64 = Buffer.from(content).toString("base64"); + return writeBackupFile( + join(directory, "documents.json"), + ownerId, + "documents", + DOCUMENTS, + (i) => ({ + id: `upload-memory-doc-${i}`, + kind: "DOCTOR_REPORT", + title: `Report ${i}`, + filename: `report-${i}.pdf`, + mimeType: "application/pdf", + byteSize: content.byteLength, + status: "STORED", + contentEncrypted: base64, + contentCodec: codec, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); +} + +/** ECG strips, each carrying the same real trace ciphertext. */ +async function writeEcgFile(ownerId: string): Promise { + const { encryptWaveformToBytes } = + await import("@/lib/withings/ecg-waveform-codec"); + const samples = Array.from( + { length: ECG_SAMPLES }, + (_, i) => Math.round(Math.sin(i / 25) * 120_000) / 100, + ); + const trace = Buffer.from(encryptWaveformToBytes(samples)).toString("base64"); + return writeBackupFile( + join(directory, "ecg.json"), + ownerId, + "ecgRecordings", + ECG_STRIPS, + (i) => ({ + id: `upload-memory-ecg-${i}`, + source: "WITHINGS", + externalRecordingId: `strip-${i}`, + recordedAt: new Date(Date.UTC(2025, 0, 1) + i * 86_400_000).toISOString(), + waveformEncrypted: trace, + samplingFrequency: 300, + sampleCount: ECG_SAMPLES, + durationSeconds: 30, + lead: "I", + averageHeartRate: 64, + rhythmClassification: null, + measurementId: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); } beforeAll(async () => { @@ -177,8 +241,8 @@ beforeAll(async () => { cookieJar.set("healthlog_session", session.id); directory = await mkdtemp(join(tmpdir(), "backup-upload-memory-")); - filePath = join(directory, "backup.json"); - await writeBackupFile(admin.id); + documentFile = await writeDocumentFile(admin.id); + ecgFile = await writeEcgFile(admin.id); }, 240_000); afterAll(async () => { @@ -186,62 +250,73 @@ afterAll(async () => { await truncateAllTables(prisma); }); -describe("backup upload of a document vault under a memory budget", () => { - it("holds a bounded amount while it checks and stores the file", async () => { - expect(fileBytes).toBeGreaterThan(DOCUMENTS * DOCUMENT_BYTES); +/** + * Send `file` through the real upload route, reading the heap every few + * chunks of the body, and answer what the route held at its peak and once + * the heavy section had gone past. + */ +async function uploadUnderWatch(file: HeavyFile, label: string) { + const { POST } = await import("@/app/api/admin/backups/upload/route"); + // Modules loaded once per process, not memory that grows with the file: + // loaded before the baseline, as a running server has long since done. + await import("@/lib/validations/backup"); + const baseline = liveHeapBytes(); - const { POST } = await import("@/app/api/admin/backups/upload/route"); - // Modules loaded once per process, not memory that grows with the file: - // loaded before the baseline, as a running server has long since done. - await import("@/lib/validations/backup"); - const baseline = liveHeapBytes(); - - let bytesSent = 0; - let chunks = 0; - let peakHeld = 0; - let samplesPastVault = 0; - let peakPastVault = 0; - async function* body(): AsyncGenerator { - for await (const chunk of createReadStream(filePath)) { - const buffer = chunk as Buffer; - bytesSent += buffer.byteLength; - if (chunks++ % SAMPLE_EVERY_CHUNKS === 0) { - const held = liveHeapBytes() - baseline; - peakHeld = Math.max(peakHeld, held); - // The route has had the whole vault once it asks for the chunk - // after the one that closed it. - if (bytesSent - buffer.byteLength > vaultEndsAt) { - samplesPastVault++; - peakPastVault = Math.max(peakPastVault, held); - } + let bytesSent = 0; + let chunks = 0; + let peakHeld = 0; + let samplesPastHeavy = 0; + let peakPastHeavy = 0; + async function* body(): AsyncGenerator { + for await (const chunk of createReadStream(file.path)) { + const buffer = chunk as Buffer; + bytesSent += buffer.byteLength; + if (chunks++ % SAMPLE_EVERY_CHUNKS === 0) { + const held = liveHeapBytes() - baseline; + peakHeld = Math.max(peakHeld, held); + // The route has had the whole heavy section once it asks for the + // chunk after the one that closed it. + if (bytesSent - buffer.byteLength > file.heavyEndsAt) { + samplesPastHeavy++; + peakPastHeavy = Math.max(peakPastHeavy, held); } - yield buffer; } + yield buffer; } + } - const request = new Request("http://localhost/api/admin/backups/upload", { - method: "POST", - headers: { - "content-type": "application/json", - "content-length": String(fileBytes), - }, - body: Readable.toWeb(Readable.from(body())) as ReadableStream, - duplex: "half", - } as RequestInit & { duplex: "half" }); - const res = await POST(request as unknown as Parameters[0]); - peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); + const request = new Request("http://localhost/api/admin/backups/upload", { + method: "POST", + headers: { + "content-type": "application/json", + "content-length": String(file.bytes), + }, + body: Readable.toWeb(Readable.from(body())) as ReadableStream, + duplex: "half", + } as RequestInit & { duplex: "half" }); + const res = await POST(request as unknown as Parameters[0]); + peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline); - process.stderr.write( - `[upload-document-memory] ${mb(fileBytes)} MB file, ${DOCUMENTS} documents: ` + - `peak held ${mb(peakHeld)} MB, past the vault ${mb(peakPastVault)} MB ` + - `(${samplesPastVault} samples), ` + - `${mb(v8.getHeapStatistics().heap_size_limit)} MB heap limit\n`, - ); + process.stderr.write( + `[upload-memory] ${label}, ${mb(file.bytes)} MB file: ` + + `peak held ${mb(peakHeld)} MB, past the heavy section ` + + `${mb(peakPastHeavy)} MB (${samplesPastHeavy} samples), ` + + `${mb(v8.getHeapStatistics().heap_size_limit)} MB heap limit\n`, + ); + + const answer = (await res.json()) as { + data: { id: string; summary: Record } | null; + error: string | null; + }; + return { res, answer, bytesSent, peakHeld, samplesPastHeavy }; +} + +describe("backup upload of a heavy file under a memory budget", () => { + it("holds a bounded amount while it checks and stores a document vault", async () => { + expect(documentFile.bytes).toBeGreaterThan(DOCUMENTS * DOCUMENT_BYTES); + const { res, answer, bytesSent, peakHeld, samplesPastHeavy } = + await uploadUnderWatch(documentFile, `${DOCUMENTS} documents`); - const answer = (await res.json()) as { - data: { id: string; summary: Record } | null; - error: string | null; - }; expect(res.status, JSON.stringify(answer.error)).toBe(201); expect(answer.data!.summary).toMatchObject({ documents: DOCUMENTS, @@ -256,11 +331,29 @@ describe("backup upload of a document vault under a memory budget", () => { // Not vacuous: the whole file went through the route, and the route was // sampled after the vault had gone past it. - expect(bytesSent).toBe(fileBytes); - expect(samplesPastVault).toBeGreaterThan(3); + expect(bytesSent).toBe(documentFile.bytes); + expect(samplesPastHeavy).toBeGreaterThan(3); expect( peakHeld, - `the upload held ${mb(peakHeld)} MB of a ${mb(fileBytes)} MB file`, + `the upload held ${mb(peakHeld)} MB of a ${mb(documentFile.bytes)} MB file`, + ).toBeLessThan(HELD_BUDGET_BYTES); + }, 600_000); + + it("holds a bounded amount while it checks and stores an ECG history", async () => { + expect(ecgFile.bytes).toBeGreaterThan(150 * 1024 * 1024); + const { res, answer, bytesSent, peakHeld, samplesPastHeavy } = + await uploadUnderWatch(ecgFile, `${ECG_STRIPS} ECG strips`); + + expect(res.status, JSON.stringify(answer.error)).toBe(201); + expect(answer.data!.summary).toMatchObject({ + ecgRecordings: ECG_STRIPS, + measurements: MEASUREMENTS, + }); + expect(bytesSent).toBe(ecgFile.bytes); + expect(samplesPastHeavy).toBeGreaterThan(3); + expect( + peakHeld, + `the upload held ${mb(peakHeld)} MB of a ${mb(ecgFile.bytes)} MB file`, ).toBeLessThan(HELD_BUDGET_BYTES); }, 600_000); });