diff --git a/CHANGELOG.md b/CHANGELOG.md index 406741f71..bc35f1a3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # Changelog +## [1.43.5] — 2026-10-18 + +### Fixed + +- **Backups of accounts with many documents no longer run out of memory.** + The weekly in-database backup, the manual backup and the restore held the + document vault and ECG traces in memory several times over (bytes, base64, + the serialised section and the preview check), and sealed one oversized + chunk; an account with a large document vault crashed a 524 MB heap. Document + bytes are now read and written in 1.5 MB slices, ECG traces page by page, + every section is written element by element and no chunk exceeds the chunk + size. The restore reads documents without their bytes first and writes each + one on its own. The file format is unchanged (Refs #1239). +- A failed tombstone purge names its phase and condition (e.g. + `statement_timeout`) instead of a bare error. + +### Added + +- The admin support report and System status flag a heap limit below 700 MB + and point to `APP_MEMORY_LIMIT`; `/api/admin/status` carries + `memory: { heapLimitMb, low }`. + ## [1.43.4] — 2026-10-17 ### Fixed diff --git a/docker-compose.yml b/docker-compose.yml index 80d8c9015..e03a72d4e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -492,8 +492,8 @@ services: # workload can grow large enough to starve Postgres on a shared host. # Node sizes its heap from this limit: 1g gives a 524 MB heap, which a # long-used account fills in normal operation, so garbage collection - # stalls requests for seconds and a weekly backup can crash the process. - # 1536m gives 792 MB. On a host with 2 GB or less, lower it with + # stalls requests for seconds. 1536m gives 792 MB. The support report and + # Admin > System status flag a heap below 700 MB. On a host with 2 GB or less, lower it with # APP_MEMORY_LIMIT in .env; on a busy instance, raise it. mem_limit: "${APP_MEMORY_LIMIT:-1536m}" healthcheck: diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 93399bf62..626d0e6b9 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.43.4 + version: 1.43.5 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. 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/docs/self-hosting/getting-started.md b/docs/self-hosting/getting-started.md index 58e7f5856..e9ee7aa1f 100644 --- a/docs/self-hosting/getting-started.md +++ b/docs/self-hosting/getting-started.md @@ -18,7 +18,7 @@ already installed, longer if a TLS-fronted public hostname is in scope heap at about 792 MB). On a 2 GB host that leaves too little for Postgres and the system, so set `APP_MEMORY_LIMIT=1g` in `.env` (a heap of about 524 MB); a long-used account then runs closer to - that heap, and the weekly backup with it. + that heap, and the support report and Admin > System status say so. - **Outbound HTTPS** to `ghcr.io` (image pulls) plus whichever integration endpoints you plan to enable (Withings, OpenAI, etc.). The opt-in environmental-context module additionally reaches diff --git a/messages/de.json b/messages/de.json index 9e8123961..1e7d0ea10 100644 --- a/messages/de.json +++ b/messages/de.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Geodaten", "offlineGeoEnabled": "Offline (GeoLite2)", "offlineGeoFallback": "Online ({host}). GeoLite2-Datenbanken einbinden und GEOLITE2_DIR setzen", + "heapLimitLabel": "Heap-Grenze", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, zu knapp. APP_MEMORY_LIMIT=1536m in .env setzen", "offlineGeoOff": "Aus", "offlineGeoChosen": "Online ({host})", "apiGlobal": "API-Endpunkte und API-Tokens erlauben.", diff --git a/messages/en.json b/messages/en.json index e4ca118bc..5e59955ce 100644 --- a/messages/en.json +++ b/messages/en.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Geo data", "offlineGeoEnabled": "Offline (GeoLite2)", "offlineGeoFallback": "Online ({host}). Mount GeoLite2 databases and set GEOLITE2_DIR", + "heapLimitLabel": "Heap limit", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, low. Set APP_MEMORY_LIMIT=1536m in .env", "offlineGeoOff": "Off", "offlineGeoChosen": "Online ({host})", "apiGlobal": "Allow API endpoints and API tokens.", diff --git a/messages/es.json b/messages/es.json index 8a08be38e..1c799ce67 100644 --- a/messages/es.json +++ b/messages/es.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Datos geográficos", "offlineGeoEnabled": "Sin conexión (GeoLite2)", "offlineGeoFallback": "En línea ({host}): monta las bases GeoLite2 y define GEOLITE2_DIR", + "heapLimitLabel": "Límite de heap", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, escaso: define APP_MEMORY_LIMIT=1536m en .env", "offlineGeoOff": "Desactivado", "offlineGeoChosen": "En línea ({host})", "apiGlobal": "Permitir endpoints de API y tokens de API.", diff --git a/messages/fr.json b/messages/fr.json index 8aff76af4..230f4a60a 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Données géographiques", "offlineGeoEnabled": "Hors ligne (GeoLite2)", "offlineGeoFallback": "En ligne ({host}) : montez les bases GeoLite2 et définissez GEOLITE2_DIR", + "heapLimitLabel": "Limite du tas", + "heapLimitValue": "{mb} Mo", + "heapLimitLow": "{mb} Mo, trop juste : définissez APP_MEMORY_LIMIT=1536m dans .env", "offlineGeoOff": "Désactivé", "offlineGeoChosen": "En ligne ({host})", "apiGlobal": "Autoriser les points de terminaison d'API et les jetons d'API.", diff --git a/messages/it.json b/messages/it.json index 576240f9b..5446a3205 100644 --- a/messages/it.json +++ b/messages/it.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Dati geografici", "offlineGeoEnabled": "Offline (GeoLite2)", "offlineGeoFallback": "Online ({host}): monta i database GeoLite2 e imposta GEOLITE2_DIR", + "heapLimitLabel": "Limite heap", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, poco: imposta APP_MEMORY_LIMIT=1536m in .env", "offlineGeoOff": "Disattivato", "offlineGeoChosen": "Online ({host})", "apiGlobal": "Consentire endpoint API e token API.", diff --git a/messages/ko.json b/messages/ko.json index 51f44d7a8..753810a79 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -7546,6 +7546,9 @@ "offlineGeoLabel": "위치 데이터", "offlineGeoEnabled": "오프라인 (GeoLite2)", "offlineGeoFallback": "온라인 ({host}). GeoLite2 데이터베이스를 마운트하고 GEOLITE2_DIR을 설정하세요", + "heapLimitLabel": "힙 한도", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, 부족합니다. .env에 APP_MEMORY_LIMIT=1536m을 설정하세요", "offlineGeoOff": "꺼짐", "offlineGeoChosen": "온라인 ({host})", "apiGlobal": "API 엔드포인트와 API 토큰을 허용해요.", diff --git a/messages/pl.json b/messages/pl.json index eea453854..3046dc530 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -7571,6 +7571,9 @@ "offlineGeoLabel": "Dane geograficzne", "offlineGeoEnabled": "Offline (GeoLite2)", "offlineGeoFallback": "Online ({host}): podłącz bazy GeoLite2 i ustaw GEOLITE2_DIR", + "heapLimitLabel": "Limit sterty", + "heapLimitValue": "{mb} MB", + "heapLimitLow": "{mb} MB, za mało: ustaw APP_MEMORY_LIMIT=1536m w .env", "offlineGeoOff": "Wyłączone", "offlineGeoChosen": "Online ({host})", "apiGlobal": "Zezwól na punkty końcowe API i tokeny API.", diff --git a/package.json b/package.json index 7fc5051e8..6dab1f0ab 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.43.4", + "version": "1.43.5", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index c867722b4..62fac31c7 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.43.4"; + /* @sw-version-fallback */ "v1.43.5"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/app/api/admin/status/__tests__/route.test.ts b/src/app/api/admin/status/__tests__/route.test.ts index deef1897b..fc4f84c46 100644 --- a/src/app/api/admin/status/__tests__/route.test.ts +++ b/src/app/api/admin/status/__tests__/route.test.ts @@ -44,6 +44,14 @@ vi.mock("@/lib/jobs/worker-status", () => ({ })), })); +const heapMock = vi.hoisted(() => ({ mb: 792 })); +vi.mock("@/lib/support/heap-limit", async () => { + const actual = await vi.importActual< + typeof import("@/lib/support/heap-limit") + >("@/lib/support/heap-limit"); + return { ...actual, heapLimitMb: () => heapMock.mb }; +}); + const geoMock = vi.hoisted(() => ({ lookup: "online" as "offline" | "online" | "off", chosen: false, @@ -187,6 +195,19 @@ describe("GET /api/admin/status (integration health summary)", () => { expect(tokenCallArgs).toEqual({ where: { revoked: false } }); }); + it("flags a heap limit too small for a long-used account", async () => { + heapMock.mb = 524; + const low = (await (await GET()).json()) as { + data: { memory: { heapLimitMb: number; low: boolean } }; + }; + expect(low.data.memory).toEqual({ heapLimitMb: 524, low: true }); + heapMock.mb = 792; + const enough = (await (await GET()).json()) as { + data: { memory: { heapLimitMb: number; low: boolean } }; + }; + expect(enough.data.memory).toEqual({ heapLimitMb: 792, low: false }); + }); + it("reports named failing queues through failingJobs only", async () => { vi.mocked(readFailingQueues).mockResolvedValueOnce([ { diff --git a/src/app/api/admin/status/route.ts b/src/app/api/admin/status/route.ts index 5e7f4b9ab..b016702f3 100644 --- a/src/app/api/admin/status/route.ts +++ b/src/app/api/admin/status/route.ts @@ -9,6 +9,7 @@ import { readFailingQueues, } from "@/lib/jobs/job-failures"; import { readFileSync } from "node:fs"; +import { heapLimitMb, isHeapLimitLow } from "@/lib/support/heap-limit"; import { geoLookupMode, geoProviderChosen, @@ -73,6 +74,9 @@ export const GET = apiHandler(async () => { // reports. `null` = no queue schema to ask (web-only deployment); an empty // array = asked, nothing is failing. const failingJobs = await readFailingQueues(); + // The process's heap ceiling, and whether it is the 1 GB-container size + // that a long-used account can fill (`heap-limit.ts`). + const heapLimit = heapLimitMb(); return apiSuccess({ version: process.env.npm_package_version || "0.1.0", @@ -81,6 +85,7 @@ export const GET = apiHandler(async () => { buildTime: BUILD_TIME, startTime: START_TIME, database: "connected", + memory: { heapLimitMb: heapLimit, low: isHeapLimitLow(heapLimit) }, worker: { running: workerStatus.running, startedAt: workerStatus.startedAt, diff --git a/src/components/admin/_shared.tsx b/src/components/admin/_shared.tsx index 6d4545810..7abbebe4b 100644 --- a/src/components/admin/_shared.tsx +++ b/src/components/admin/_shared.tsx @@ -61,6 +61,8 @@ export interface SystemStatus { buildTime: string; startTime: string; database: string; + /** The process's heap ceiling; `low` when it is below what a long-used account needs. */ + memory: { heapLimitMb: number; low: boolean }; worker: WorkerStatus; /** * Queues whose jobs exhausted their retries recently. `null` means the diff --git a/src/components/admin/system-status-section.tsx b/src/components/admin/system-status-section.tsx index a3cf1f4c4..b2811261e 100644 --- a/src/components/admin/system-status-section.tsx +++ b/src/components/admin/system-status-section.tsx @@ -13,6 +13,7 @@ import { Key, Loader2, Map, + MemoryStick, Server, Users, } from "lucide-react"; @@ -156,6 +157,16 @@ export function SystemStatusSection() { } tone={status.database === "connected" ? "success" : "destructive"} /> + { 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/src/lib/jobs/__tests__/compaction-tombstone-purge-failure.test.ts b/src/lib/jobs/__tests__/compaction-tombstone-purge-failure.test.ts new file mode 100644 index 000000000..70b923704 --- /dev/null +++ b/src/lib/jobs/__tests__/compaction-tombstone-purge-failure.test.ts @@ -0,0 +1,61 @@ +/** + * A purge run that hits a database failure says what it was doing and why. + * + * Before #1239 the run threw the bare Prisma error, whose message carries the + * condition only as a SQLSTATE, and the support report showed seven failed + * runs as "Error". The reason now names the stage and the condition. + */ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: () => null })); +vi.mock("@/lib/jobs/job-run-report", () => ({ reportJobRun: vi.fn() })); + +import { handleCompactionTombstonePurge } from "../compaction-tombstone-purge"; + +const STATEMENT_TIMEOUT = + "\nInvalid `prisma.measurement.findMany()` invocation:\n\n\nDatabase error. Code: `57014`. Message: `canceling statement due to statement timeout`"; + +describe("compaction tombstone purge failures", () => { + it("names the stage and the condition when the scan times out", async () => { + const prisma = { + measurement: { + findMany: vi.fn().mockRejectedValue(new Error(STATEMENT_TIMEOUT)), + }, + }; + const outcome = await handleCompactionTombstonePurge( + [{ id: "job-1", data: {} } as never], + prisma as never, + ); + expect(outcome.ok).toBe(false); + if (!outcome.ok) { + expect(outcome.reason).toBe( + "stopped while scanning tombstones (statement_timeout)", + ); + expect((outcome.cause as Error).message).toBe( + "stopped while scanning tombstones (statement_timeout)", + ); + expect(((outcome.cause as Error).cause as Error).message).toBe( + STATEMENT_TIMEOUT, + ); + } + }); + + it("falls back to the error's class when no condition is named", async () => { + const prisma = { + measurement: { + findMany: vi.fn().mockRejectedValue(new TypeError("boom")), + }, + }; + const outcome = await handleCompactionTombstonePurge( + [{ id: "job-2", data: {} } as never], + prisma as never, + ); + expect(outcome.ok).toBe(false); + if (!outcome.ok) { + expect(outcome.reason).toBe( + "stopped while scanning tombstones (TypeError)", + ); + } + }); +}); diff --git a/src/lib/jobs/compaction-tombstone-purge.ts b/src/lib/jobs/compaction-tombstone-purge.ts index cc3700f9b..415eee0dc 100644 --- a/src/lib/jobs/compaction-tombstone-purge.ts +++ b/src/lib/jobs/compaction-tombstone-purge.ts @@ -55,7 +55,8 @@ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; import { holdAccountFoldLock } from "@/lib/export/restore-lock"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { jobBudget } from "@/lib/jobs/job-budget"; -import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { failureCode } from "@/lib/jobs/failure-condition"; import { reportJobRun, type JobRunCandidate } from "@/lib/jobs/job-run-report"; import { purgeTombstonesByAccount } from "@/lib/jobs/measurement-tombstone-cleanup"; import { PURGE_BATCH_SIZE, PURGE_MAX_BATCHES } from "@/lib/jobs/purge-batch"; @@ -112,6 +113,35 @@ export interface PurgeCompactionTombstonesOptions { shouldStop?: () => boolean; } +/** + * A run that stopped on a database failure, named by what it was doing and + * the condition (`failure-condition.ts`). Before #1239 the run threw the bare + * Prisma error, and the support report read seven failed runs as "Error". + */ +class CompactionPurgeStoppedError extends Error { + constructor( + readonly stage: "scanning tombstones" | "deleting a batch", + cause: unknown, + ) { + super(`stopped while ${stage} (${failureCode(cause)})`, { cause }); + this.name = "CompactionPurgeStoppedError"; + } +} + +/** Run `work`, naming `stage` on anything it throws. */ +async function during( + stage: CompactionPurgeStoppedError["stage"], + work: () => Promise, +): Promise { + try { + return await work(); + } catch (err) { + throw err instanceof CompactionPurgeStoppedError + ? err + : new CompactionPurgeStoppedError(stage, err); + } +} + type ScannedRow = { id: string; userId: string; @@ -245,27 +275,28 @@ export async function purgeCompactionTombstones( ready = ready.filter((row) => !skip.has(row.userId)); while (ready.length < take && !exhausted) { if (options.shouldStop?.()) break; - await classifyNextPage(skip); + await during("scanning tombstones", () => classifyNextPage(skip)); } return ready.splice(0, take); }, - deleteIds: async (tx, ids, userId) => { - // A re-fold reads an account's live samples and its leftovers under the - // fold lock, so it never sees the leftovers half gone; it reads the - // repair marker under the same lock and stays off them once the repair - // has run (`fold-constituents.ts`). - await holdAccountFoldLock(tx, userId); - // Re-checked at delete time: a row a restore wrote back, or one a - // person resurrected since the scan, is not a tombstone any more. - const result = await tx.measurement.deleteMany({ - where: { - id: { in: ids }, - deletedAt: { not: null }, - source: "APPLE_HEALTH", - }, - }); - return result.count; - }, + deleteIds: (tx, ids, userId) => + during("deleting a batch", async () => { + // A re-fold reads an account's live samples and its leftovers under the + // fold lock, so it never sees the leftovers half gone; it reads the + // repair marker under the same lock and stays off them once the repair + // has run (`fold-constituents.ts`). + await holdAccountFoldLock(tx, userId); + // Re-checked at delete time: a row a restore wrote back, or one a + // person resurrected since the scan, is not a tombstone any more. + const result = await tx.measurement.deleteMany({ + where: { + id: { in: ids }, + deletedAt: { not: null }, + source: "APPLE_HEALTH", + }, + }); + return result.count; + }), }); return { @@ -301,7 +332,21 @@ export async function handleCompactionTombstonePurge( ); return withBackgroundEvent("job.compaction_tombstone_purge", async (evt) => { const startedAt = Date.now(); - const result = await purgeCompactionTombstones(prisma, { shouldStop }); + let result: CompactionTombstonePurgeOutcome; + try { + result = await purgeCompactionTombstones(prisma, { shouldStop }); + } catch (err) { + // The reason names the stage and the condition, which is what the + // admin's failing-jobs card and the support report show; the error + // itself rides as the cause, for the log. + return jobFailed( + err instanceof CompactionPurgeStoppedError + ? err.message + : `stopped (${failureCode(err)})`, + err, + { continuation }, + ); + } const candidates: JobRunCandidate[] = [ ...result.deferredUserIds.map((userId): JobRunCandidate => ({ key: accountKey(userId), diff --git a/src/lib/jobs/failure-condition.ts b/src/lib/jobs/failure-condition.ts new file mode 100644 index 000000000..38392fdac --- /dev/null +++ b/src/lib/jobs/failure-condition.ts @@ -0,0 +1,66 @@ +/** + * The database condition behind a failure, as a stable code. + * + * Prisma puts the Postgres SQLSTATE in the message ("Database error. Code: + * `57014`. Message: `canceling statement due to statement timeout`") and not + * in a form the support report's error class recognised, so a statement that + * ran past the 60-second `statement_timeout` read as the bare class "Error" + * (#1239). The code is the part an operator can act on, and it carries no + * account data, so it is what the report and a job's failure reason show. + */ + +/** SQLSTATEs worth a name; any other one is shown as `sqlstate_`. */ +const SQLSTATE_CONDITIONS: Readonly> = { + "57014": "statement_timeout", + "55P03": "lock_timeout", + "40P01": "deadlock", + "40001": "serialization_failure", + "25P03": "idle_in_transaction_timeout", + "53100": "disk_full", + "53200": "out_of_memory", + "53300": "too_many_connections", + "57P01": "admin_shutdown", +}; + +/** Conditions named by the message text alone, in the order they are tried. */ +const TEXT_CONDITIONS: ReadonlyArray = [ + [/statement[ _]timeout/i, "statement_timeout"], + [/lock[ _]timeout/i, "lock_timeout"], + [/deadlock/i, "deadlock"], + [/transaction already closed|expired transaction/i, "transaction_closed"], + // `pg`'s own words when no pooled connection came free in time, and + // Prisma's for the same. + [ + /timeout exceeded when trying to connect|new connection from the connection pool/i, + "pool_timeout", + ], + [ + /connection terminated|connection_lost|server closed the connection/i, + "connection_lost", + ], +]; + +/** The condition a failure message names, or null when it names none. */ +export function failureCondition(text: string): string | null { + const sqlstate = /\bCode: `([0-9A-Z]{5})`/.exec(text)?.[1]; + if (sqlstate) { + return SQLSTATE_CONDITIONS[sqlstate] ?? `sqlstate_${sqlstate}`; + } + for (const [pattern, condition] of TEXT_CONDITIONS) { + if (pattern.test(text)) return condition; + } + return null; +} + +/** + * A thrown value as a short, account-free code for a job's failure reason: + * its database condition, else its class name. + */ +export function failureCode(error: unknown): string { + if (error instanceof Error) { + return failureCondition(error.message) ?? error.name; + } + return typeof error === "string" + ? (failureCondition(error) ?? "error") + : "error"; +} diff --git a/src/lib/support/__tests__/support-report-redaction.test.ts b/src/lib/support/__tests__/support-report-redaction.test.ts index d21046b33..72bc7f1ea 100644 --- a/src/lib/support/__tests__/support-report-redaction.test.ts +++ b/src/lib/support/__tests__/support-report-redaction.test.ts @@ -200,6 +200,25 @@ describe("support report redaction", () => { expect(text).toContain("environment-fetch"); }); + it("says what to raise when the heap limit is low, and keeps it through redaction", () => { + const facts = seededFacts(); + const low = redactSupportText( + renderSupportReport({ + ...facts, + build: { ...facts.build, heapLimitMb: 524 }, + }), + CONTEXT, + ); + expect(low).toContain("- Heap limit: 524 MB"); + expect(low).toContain( + "- Heap limit note: The heap limit of 524 MB is below 700 MB.", + ); + expect(low).toContain("APP_MEMORY_LIMIT=1536m in .env"); + expect(low).toContain("docker compose up -d."); + const enough = redactSupportText(renderSupportReport(facts), CONTEXT); + expect(enough).not.toContain("Heap limit note"); + }); + it("the seeds really are in the unredacted text", () => { const raw = renderSupportReport(seededFacts()); for (const value of FORBIDDEN) expect(raw).toContain(value); @@ -256,6 +275,21 @@ describe("error classes never carry the message", () => { expect(errorClassOf("job timed out: did not finish within 1 h")).toBe( "timeout", ); + // A statement past its timeout, as Prisma words it: the SQLSTATE is the + // class, not the bare "Error" it used to read as. + expect( + errorClassOf( + "\nInvalid `prisma.measurement.findMany()` invocation:\n\n\nDatabase error. Code: `57014`. Message: `canceling statement due to statement timeout`", + ), + ).toBe("statement_timeout"); + expect( + errorClassOf( + "compaction-tombstone-purge: stopped while deleting a batch (lock_timeout)", + ), + ).toBe("lock_timeout"); + expect(errorClassOf("timeout exceeded when trying to connect")).toBe( + "pool_timeout", + ); expect(errorClassOf("anything at all, an email a@b.co")).toBe("Error"); expect(errorClassOf(null)).toBe("unknown"); }); diff --git a/src/lib/support/heap-limit.ts b/src/lib/support/heap-limit.ts new file mode 100644 index 000000000..e11c9656e --- /dev/null +++ b/src/lib/support/heap-limit.ts @@ -0,0 +1,35 @@ +/** + * Whether this process's heap is below what a long-used account needs. + * + * Node sizes its heap from the container's memory limit: the 1 GB limit older + * compose files set gives 524 MB, the current default of 1536m gives 792 MB. + * At 524 MB a long-used account fills the heap in normal operation, garbage + * collection stalls requests, and a large import can run the process out of + * memory, which restarts the whole instance. The support report and the admin + * system status say so in one sentence, because the number alone ("Heap + * limit: 524 MB") reads as a fact rather than as the cause (#1239). + */ +import { getHeapStatistics } from "node:v8"; + +/** Below this, the heap is flagged. Between the 1g (524) and 1536m (792) heaps. */ +export const LOW_HEAP_LIMIT_MB = 700; + +/** V8's heap limit for this process, in megabytes. */ +export function heapLimitMb(): number { + return Math.round(getHeapStatistics().heap_size_limit / 1024 / 1024); +} + +export function isHeapLimitLow(limitMb: number): boolean { + return limitMb < LOW_HEAP_LIMIT_MB; +} + +/** The support report's sentence for a low heap limit. */ +export function lowHeapLimitAdvice(limitMb: number): string { + return ( + `The heap limit of ${limitMb} MB is below ${LOW_HEAP_LIMIT_MB} MB. The ` + + "usual cause is a 1 GB container memory limit. A long-used account can " + + "fill this heap, which slows requests and can restart the app. Set " + + "APP_MEMORY_LIMIT=1536m in .env (about 792 MB of heap) and run " + + "docker compose up -d." + ); +} diff --git a/src/lib/support/report.ts b/src/lib/support/report.ts index 8dae07ce3..15ac11743 100644 --- a/src/lib/support/report.ts +++ b/src/lib/support/report.ts @@ -18,9 +18,14 @@ * * The report is English on purpose: it is read by whoever answers the issue. */ -import { getHeapStatistics } from "node:v8"; import { prisma } from "@/lib/db"; +import { failureCondition } from "@/lib/jobs/failure-condition"; +import { + heapLimitMb, + isHeapLimitLow, + lowHeapLimitAdvice, +} from "@/lib/support/heap-limit"; import { shouldEmitSecureCookie } from "@/lib/auth/secure-cookie"; import { isOidcConfigured, isOidcOnly } from "@/lib/auth/oidc"; import { getLoggingConfig } from "@/lib/logging/config"; @@ -183,6 +188,11 @@ export function errorClassOf(text: string | null | undefined): string { const parts: string[] = []; const named = /^([A-Z][A-Za-z0-9]{1,60}(?:Error|Exception))\b/.exec(trimmed); if (named) parts.push(named[1]); + // The database condition (`statement_timeout`, `lock_timeout`, ...): a + // Prisma error carries it only as a SQLSTATE inside the message, and + // without it a statement timeout read as the bare class "Error" (#1239). + const condition = failureCondition(trimmed); + if (condition) return [...parts, condition].join(" "); const code = /\b(P\d{4})\b/.exec(trimmed) ?? /\b(E[A-Z][A-Z_]{2,30})\b/.exec(trimmed) ?? @@ -605,9 +615,7 @@ export async function collectSupportReportFacts( builtAt: process.env.NEXT_PUBLIC_APP_BUILT_AT?.trim() || null, node: process.version, processType: getProcessType(), - heapLimitMb: Math.round( - getHeapStatistics().heap_size_limit / 1024 / 1024, - ), + heapLimitMb: heapLimitMb(), uptimeSeconds: Math.round(process.uptime()), }, database, @@ -722,6 +730,9 @@ export function renderSupportReport(facts: SupportReportFacts): string { row("Node", facts.build.node); row("Process type", facts.build.processType); row("Heap limit", `${facts.build.heapLimitMb} MB`); + if (isHeapLimitLow(facts.build.heapLimitMb)) { + row("Heap limit note", lowHeapLimitAdvice(facts.build.heapLimitMb)); + } row( "Uptime", `${Math.floor(facts.build.uptimeSeconds / 3600)} h ${Math.floor((facts.build.uptimeSeconds % 3600) / 60)} min`, 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); +});