diff --git a/scripts/restore-backup.ts b/scripts/restore-backup.ts index 15cc8e80a..6f077577b 100644 --- a/scripts/restore-backup.ts +++ b/scripts/restore-backup.ts @@ -28,6 +28,7 @@ import { pipeline } from "node:stream/promises"; import { createGzip } from "node:zlib"; import { scanBackupJson } from "@/lib/export/backup-json-scan"; +import { BACKUP_STREAM_KEYS } from "@/lib/export/streamed-backup"; import { loadOffhostConfig, getS3Client, @@ -115,7 +116,7 @@ async function main() { // Checked first, in one streamed pass that keeps only the small sections, // so nothing is written for an object that is not a backup. const { document, streamedCounts } = await scanBackupJson(source(), { - streamKeys: new Set(["measurements", "intakeEvents", "moodEntries"]), + streamKeys: BACKUP_STREAM_KEYS, }); const body = validateBackupShape({ ...document, diff --git a/src/__tests__/backup-readers-stream-guard.test.ts b/src/__tests__/backup-readers-stream-guard.test.ts index 8292f49d6..01ef00baa 100644 --- a/src/__tests__/backup-readers-stream-guard.test.ts +++ b/src/__tests__/backup-readers-stream-guard.test.ts @@ -1,5 +1,6 @@ /** - * Every reader of a stored or uploaded backup reads it as a stream (#1031). + * Every reader of a stored or uploaded backup reads it as a stream (#1031), + * through the one shared reader (#1239). * * A backup's JSON grows with the record, and past 536 870 888 characters it * cannot exist as one JavaScript string at all: the disaster-recovery JSON of @@ -7,19 +8,50 @@ * below used to turn the file into one string first, and so none of them * could open that account's backup. This pins that none of them goes back to * it: no whole-document decrypt-to-string, no whole-body text read. + * + * Streaming is not enough on its own. The scanner keeps every section it is + * not told to stream as one string and one parse, so a reader is only as + * light as the section list it passes. The upload route once passed its own + * list with the measurements alone in it, and held a document vault of a few + * hundred megabytes whole while the restore beside it streamed the same + * documents. So every reader goes through `readStreamedBackup`, or, where it + * only counts and checks keys, scans with `BACKUP_STREAM_KEYS`, and no reader + * names a section list of its own. The sweep at the end finds any file that + * calls the scanner or names a stream list, so a new reader cannot sit + * outside this list unnoticed. + * + * Every matcher is whitespace-tolerant and every check also asserts that it + * matched something: a pattern that matches nothing proves nothing. */ -import { readFileSync } from "node:fs"; +import { readFileSync, readdirSync, statSync } from "node:fs"; +import { join } from "node:path"; import { describe, expect, it } from "vitest"; -const READERS = [ +import { + BACKUP_STREAMED_SECTIONS, + BACKUP_STREAM_KEYS, +} from "@/lib/export/streamed-backup"; + +/** Where the stream list is defined, and the scanner it is handed to. */ +const SHARED_READER = "src/lib/export/streamed-backup.ts"; +const SCANNER = "src/lib/export/backup-json-scan.ts"; + +/** Readers that read the file through `readStreamedBackup`. */ +const SHARED_READER_CALLERS = [ "src/lib/export/restore-preflight.ts", "src/app/api/admin/backups/[id]/summary/route.ts", "src/app/api/admin/backups/[id]/download/route.ts", "src/app/api/admin/backups/upload/route.ts", +] as const; + +/** Readers that only count and check keys: the scanner, with the shared list. */ +const SHARED_KEY_SCANNERS = [ "src/lib/jobs/restore-drill.ts", "scripts/restore-backup.ts", ] as const; +const READERS = [...SHARED_READER_CALLERS, ...SHARED_KEY_SCANNERS]; + /** Calls that hand back the whole document as one string or parse it whole. */ const WHOLE_DOCUMENT = [ /\bunpackBackupBlob\s*\(/, @@ -30,6 +62,35 @@ const WHOLE_DOCUMENT = [ /\bawait\s+request\.text\s*\(\s*\)[\s\S]{0,200}parseBackupPayload/, ]; +const SHARED_READER_CALL = /\breadStreamedBackup\s*\(/g; +const SCANNER_CALL = /\bscanBackupJson\s*\(/g; +/** Any stream list handed to the scanner, whatever its value. */ +const STREAM_KEYS_OPTION = /\bstreamKeys\s*:/g; +/** The one stream list a reader may hand it. */ +const SHARED_STREAM_KEYS_OPTION = /\bstreamKeys\s*:\s*BACKUP_STREAM_KEYS\b/g; +const SHARED_IMPORT = + /import\s*\{[^}]*\bBACKUP_STREAM_KEYS\b[^}]*\}\s*from\s*["']@\/lib\/export\/streamed-backup["']/; +const SHARED_READER_IMPORT = + /import\s*\{[^}]*\breadStreamedBackup\b[^}]*\}\s*from\s*["']@\/lib\/export\/streamed-backup["']/; + +function count(source: string, pattern: RegExp): number { + return [...source.matchAll(pattern)].length; +} + +function walk(dir: string, out: string[] = []): string[] { + for (const name of readdirSync(dir)) { + const path = join(dir, name); + if (statSync(path).isDirectory()) { + if (name === "node_modules" || name === "generated") continue; + if (name === "__tests__") continue; + walk(path, out); + } else if (/\.(ts|tsx|mts)$/.test(name) && !/\.test\.tsx?$/.test(name)) { + out.push(path); + } + } + return out; +} + describe("backup readers stream the file", () => { it.each(READERS)("%s never holds the whole document", (path) => { const source = readFileSync(path, "utf8"); @@ -38,8 +99,63 @@ describe("backup readers stream the file", () => { } }); - it.each(READERS)("%s reads through the streaming reader", (path) => { - const source = readFileSync(path, "utf8"); - expect(source).toMatch(/\b(readStreamedBackup|scanBackupJson)\b/); + it.each(SHARED_READER_CALLERS)( + "%s reads through the shared reader and scans nothing itself", + (path) => { + const source = readFileSync(path, "utf8"); + expect(source).toMatch(SHARED_READER_IMPORT); + expect(count(source, SHARED_READER_CALL)).toBeGreaterThan(0); + expect(count(source, SCANNER_CALL), `${path} calls the scanner`).toBe(0); + expect( + count(source, STREAM_KEYS_OPTION), + `${path} names a stream list of its own`, + ).toBe(0); + }, + ); + + it.each(SHARED_KEY_SCANNERS)( + "%s scans with the shared stream list and no list of its own", + (path) => { + const source = readFileSync(path, "utf8"); + const calls = count(source, SCANNER_CALL); + expect(calls).toBeGreaterThan(0); + expect(source).toMatch(SHARED_IMPORT); + // Every scan names a list, and every list named is the shared one. + expect(count(source, SHARED_STREAM_KEYS_OPTION)).toBe(calls); + expect(count(source, STREAM_KEYS_OPTION)).toBe(calls); + }, + ); + + it("the shared reader's first read uses the shared stream list", () => { + const source = readFileSync(SHARED_READER, "utf8"); + expect(count(source, SHARED_STREAM_KEYS_OPTION)).toBeGreaterThan(0); + // The sections that grow with a record, in rows or in stored bytes. + expect([...BACKUP_STREAM_KEYS].sort()).toEqual( + [...BACKUP_STREAMED_SECTIONS].sort(), + ); + for (const section of [ + "measurements", + "intakeEvents", + "moodEntries", + "documents", + "ecgRecordings", + ]) { + expect(BACKUP_STREAM_KEYS.has(section), section).toBe(true); + } + }); + + it("no file outside the list scans a backup or names a stream list", () => { + const found = [...walk("src"), ...walk("scripts")] + .filter((path) => path !== SHARED_READER && path !== SCANNER) + .filter((path) => { + const source = readFileSync(path, "utf8"); + return ( + count(source, SCANNER_CALL) > 0 || + count(source, STREAM_KEYS_OPTION) > 0 + ); + }) + .sort(); + expect(found.length).toBeGreaterThan(0); + expect(found).toEqual([...SHARED_KEY_SCANNERS].sort()); }); }); diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index 3f0155aef..d3aec717a 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,10 +1,10 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 348, - "total": 112838, + "total": 112810, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, - "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, + "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 286, "app/api/admin/notifications/reminder-check/route.ts :: apiHandler(arg)": 208, "app/api/admin/settings/route.ts :: apiHandler(arg) #2": 211, "app/api/auth/login/route.ts :: apiHandler(arg)": 262, diff --git a/src/app/api/admin/backups/upload/route.ts b/src/app/api/admin/backups/upload/route.ts index b0d14a5ea..c37642213 100644 --- a/src/app/api/admin/backups/upload/route.ts +++ b/src/app/api/admin/backups/upload/route.ts @@ -42,7 +42,7 @@ import { } from "@/lib/api-handler"; import { apiError, apiSuccess, getClientIp } from "@/lib/api-response"; import { auditLog } from "@/lib/auth/audit"; -import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; +import { BackupJsonError } from "@/lib/export/backup-json-scan"; import { BackupUploadDecodeError, decodeBackupUpload, @@ -58,12 +58,15 @@ import { BackupBlobTooLargeError, } from "@/lib/export/backup-blob"; import { storeBackupBlob } from "@/lib/export/store-backup-blob"; -import { StreamedBackupInvalidError } from "@/lib/export/streamed-backup"; +import { + readStreamedBackup, + StreamedBackupInvalidError, + type StreamedBackup, +} from "@/lib/export/streamed-backup"; import { annotate } from "@/lib/logging/context"; import { checkRateLimit } from "@/lib/rate-limit"; import type { BackupPayload } from "@/lib/validations/backup"; import { - BACKUP_SCHEMA_VERSION, isCompatibleSchemaVersion, summarizeBackup, type BackupSummary, @@ -219,9 +222,9 @@ export const POST = apiHandler(async (request: NextRequest) => { let measurementCount = 0; let owner: { id: string; username: string } | null = null; - // Filled while the file is read; the preview is the summary this route + // Filled once the file is read; the preview is the summary this route // answers with and the key uses it already holds (`backup-preview.ts`). - const keys = new BackupKeyIdCollector(); + let keys: BackupKeyIdCollector | undefined; let created: { id: string }; try { created = await storeBackupBlob( @@ -231,7 +234,7 @@ export const POST = apiHandler(async (request: NextRequest) => { type: uploadType, ownerAfterRead: () => owner!.id, preview: () => - payload + payload && keys ? { summary: { ...summarizeBackup(payload), @@ -250,29 +253,15 @@ export const POST = apiHandler(async (request: NextRequest) => { yield chunk; } } - // Loaded on first use, not at import (see `backup-summary.ts`). - const { backupMeasurementSchema, backupPayloadSchema } = - await import("@/lib/validations/backup"); - let firstWithoutId: number | null = null; - let scanned; + // Read through the one streamed reader every backup reader shares + // (`streamed-backup.ts`): the bulk sections an element at a time and + // the documents without their stored bytes, so a document vault of + // hundreds of megabytes is never in memory at once (#1239). It checks + // each measurement and the canonical v2 id rule as it goes, and + // gathers the key ids the file's inner ciphertext needs. + let streamed: StreamedBackup; try { - scanned = await scanBackupJson(storedAsRead(), { - streamKeys: new Set(["measurements"]), - onElement: (_key, element, index) => { - const parsed = backupMeasurementSchema.safeParse(element); - if (!parsed.success) { - const issue = parsed.error.issues[0]; - throw new StreamedBackupInvalidError( - ["measurements", index, ...(issue?.path ?? [])].join("."), - issue?.message ?? "invalid measurement", - ); - } - if (!parsed.data.id && firstWithoutId === null) { - firstWithoutId = index; - } - keys.visit(element, "measurements"); - }, - }); + streamed = await readStreamedBackup(storedAsRead); } catch (err) { if (err instanceof BackupJsonError) { throw new UploadRefused(422, "Uploaded file is not valid JSON", { @@ -291,10 +280,13 @@ export const POST = apiHandler(async (request: NextRequest) => { } throw err; } - measurementCount = scanned.streamedCounts.measurements ?? 0; + measurementCount = streamed.measurementCount; + // Loaded on first use, not at import (see `backup-summary.ts`). + const { backupPayloadSchema } = + await import("@/lib/validations/backup"); try { - payload = backupPayloadSchema.parse(scanned.document); + payload = backupPayloadSchema.parse(streamed.raw); } catch (err) { const issues = err instanceof ZodError @@ -310,23 +302,6 @@ export const POST = apiHandler(async (request: NextRequest) => { { issues }, ); } - if ( - payload.schemaVersion === BACKUP_SCHEMA_VERSION && - firstWithoutId !== null - ) { - const issues = [ - { - path: `measurements.${firstWithoutId}.id`, - message: "Canonical v2 measurements require a stable id", - }, - ]; - throw new UploadRefused( - 422, - "Backup payload failed schema validation", - { reason: "schema_invalid", issues }, - { issues }, - ); - } if (!isCompatibleSchemaVersion(payload.schemaVersion)) { throw new UploadRefused( @@ -345,7 +320,7 @@ export const POST = apiHandler(async (request: NextRequest) => { // nobody can open. Refused here, where the operator is looking, with // the key named. The instance settings are left to the restore, which // checks them only when they are asked for back. - keys.visit(scanned.document); + keys = streamed.keys; const keyVerdict = assessBackupKeys(keys, { ignoreSections: new Set(["appSettings"]), }); diff --git a/src/lib/export/streamed-backup.ts b/src/lib/export/streamed-backup.ts index 7bf436291..8e5e7dd29 100644 --- a/src/lib/export/streamed-backup.ts +++ b/src/lib/export/streamed-backup.ts @@ -14,6 +14,15 @@ * bytes (`DOCUMENT_CONTENT_LEFT_OUT` in their place), and * `forEachDocumentContent` reads them again one document 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 + * file is ever one string of its own. That string, and the parse beside it, + * is what a reader that takes a section whole pays twice over. + * + * Every reader of a backup file goes through `readStreamedBackup`, or, where + * it only counts and checks keys, through `BACKUP_STREAM_KEYS`: the sections + * read element by element are named here and nowhere else. + * * 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 @@ -51,7 +60,28 @@ export type BackupSource = () => AsyncIterable; const MEASUREMENTS = "measurements"; const DOCUMENTS = "documents"; -const STREAMED = new Set([MEASUREMENTS, DOCUMENTS]); + +/** + * 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. + */ +export const BACKUP_STREAMED_SECTIONS = [ + MEASUREMENTS, + "intakeEvents", + "moodEntries", + DOCUMENTS, + "ecgRecordings", +] as const; + +/** `BACKUP_STREAMED_SECTIONS` as `scanBackupJson` takes it. */ +export const BACKUP_STREAM_KEYS: ReadonlySet = new Set( + BACKUP_STREAMED_SECTIONS, +); + const MEASUREMENTS_ONLY = new Set([MEASUREMENTS]); const DOCUMENTS_ONLY = new Set([DOCUMENTS]); @@ -149,7 +179,8 @@ export async function readStreamedBackup( let firstWithoutId: number | null = null; let bytes = 0; const keys = new BackupKeyIdCollector(); - const documents: unknown[] = []; + // Every streamed section but the measurements, as `raw` keeps it. + const kept = new Map(); async function* counted() { for await (const chunk of source()) { bytes += chunk.byteLength; @@ -157,16 +188,20 @@ export async function readStreamedBackup( } } const { document, streamedCounts } = await scanBackupJson(counted(), { - streamKeys: STREAMED, + streamKeys: BACKUP_STREAM_KEYS, onElement: (key, element, index) => { - if (key === DOCUMENTS) { - documents.push(withoutContent(element, index, keys)); + if (key === MEASUREMENTS) { + const row = parseMeasurement(schema, element, index); + if (!row.id && firstWithoutId === null) firstWithoutId = index; + keys.visit(element, MEASUREMENTS); + options.onMeasurementChecked?.(index + 1, row); return; } - const row = parseMeasurement(schema, element, index); - if (!row.id && firstWithoutId === null) firstWithoutId = index; - keys.visit(element, MEASUREMENTS); - options.onMeasurementChecked?.(index + 1, row); + let section = kept.get(key); + if (!section) kept.set(key, (section = [])); + section.push( + key === DOCUMENTS ? withoutContent(element, index, keys) : element, + ); }, }); @@ -182,7 +217,9 @@ export async function readStreamedBackup( ); } - if (DOCUMENTS in document) document[DOCUMENTS] = documents; + // A section the file carries holds what was kept of it (an empty one kept + // nothing); one it does not carry stays absent. + for (const [key, elements] of kept) document[key] = elements; keys.visit(document); const skipKeys = new Set( diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index a2d38c476..3e1ed2aec 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -31,6 +31,7 @@ */ import type { Job } from "pg-boss"; import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; +import { BACKUP_STREAM_KEYS } from "@/lib/export/streamed-backup"; import { assessBackupKeys, BackupKeyIdCollector, @@ -133,7 +134,7 @@ async function drillObject( let scanned; try { scanned = await scanBackupJson(counted(), { - streamKeys: new Set(["measurements", "intakeEvents", "moodEntries"]), + streamKeys: BACKUP_STREAM_KEYS, onElement: (key, element) => keys.visit(element, key), }); } catch (err) { diff --git a/tests/integration/backup-upload-document-memory.test.ts b/tests/integration/backup-upload-document-memory.test.ts new file mode 100644 index 000000000..413b712a1 --- /dev/null +++ b/tests/integration/backup-upload-document-memory.test.ts @@ -0,0 +1,266 @@ +/** + * An uploaded backup with a large document vault, checked under a memory + * budget (#1239). + * + * What this pins. The upload route streamed the measurements while it checked + * a file and read every other section whole: the scanner gathered the + * `documents` member as one string and parsed it in one go. A copy of an + * account whose documents carry a few hundred megabytes of base64 then cost + * the text of the section plus its parse, more than the 524 MB heap of a 1 GB + * container, and the process was killed in the middle of the upload. The + * weekly copy and the restore had stopped doing that already; the upload had + * its own section list. It now reads through the shared streamed reader, + * which takes the documents one at a time and keeps them without their bytes. + * + * How it measures. The file is written to disk a document at a time and sent + * to the real route handler as a streamed request body, as an admin with a + * cookie session. Readings are taken from inside that body, every few + * chunks, after a forced collection, so what is compared is what the route + * HOLDS while it reads. The documents come first in the file and a few + * megabytes of readings after them, so there are readings taken once the + * whole vault has gone past: a route that kept the vault holds it there. + * + * Measured on this fixture: the route that read the documents whole held + * about 270 MB once they had gone past (the parsed base64), and run under + * `--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`. + */ +import { randomBytes } from "node:crypto"; +import { createReadStream, createWriteStream } from "node:fs"; +import { mkdtemp, rm, stat } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Readable } from "node:stream"; +import { once } from "node:events"; +import v8 from "node:v8"; +import vm from "node:vm"; + +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +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; +/** 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. */ +const SAMPLE_EVERY_CHUNKS = 8; + +/** + * What the route may hold on top of the process's baseline while it reads + * the file. Measured: a few megabytes when the documents are streamed, about + * 270 MB when the section is read whole. The bound sits far from both. + */ +const HELD_BUDGET_BYTES = 48 * 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; +} + +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 = ""; + +/** Write one piece of the file, waiting when the stream asks for it. */ +async function put( + out: ReturnType, + text: string, +): Promise { + if (!out.write(text)) await once(out, "drain"); +} + +/** + * 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. + */ +async function writeBackupFile(ownerId: string): Promise { + const { encryptDocumentContent } = await import("@/lib/documents/store"); + 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); + let written = 0; + const write = async (text: string) => { + written += Buffer.byteLength(text); + await put(out, text); + }; + const head = { + schemaVersion: BACKUP_SCHEMA_VERSION, + exportedAt: "2026-10-01T10:00:00.000Z", + userId: ownerId, + medications: [], + intakeEvents: [], + 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("]"); + vaultEndsAt = written; + await write(',"measurements":['); + for (let i = 0; i < MEASUREMENTS; i++) { + const reading = { + id: `upload-memory-reading-${i}`, + type: "WEIGHT", + value: 80 + (i % 10) / 10, + unit: "kg", + measuredAt: new Date(Date.UTC(2020, 0, 1) + i * 3_600_000).toISOString(), + source: "MANUAL", + notes: null, + }; + await write((i === 0 ? "" : ",") + JSON.stringify(reading)); + } + await write("]}"); + out.end(); + await once(out, "finish"); + fileBytes = (await stat(filePath)).size; +} + +beforeAll(async () => { + await truncateAllTables(prisma); + cookieJar.clear(); + headerJar.clear(); + const admin = await prisma.user.create({ + data: { + username: "upload-memory-admin", + email: "upload-memory-admin@example.test", + role: "ADMIN", + }, + }); + adminId = admin.id; + const session = await prisma.session.create({ + data: { userId: admin.id, expiresAt: new Date(Date.now() + 600_000) }, + }); + cookieJar.set("healthlog_session", session.id); + + directory = await mkdtemp(join(tmpdir(), "backup-upload-memory-")); + filePath = join(directory, "backup.json"); + await writeBackupFile(admin.id); +}, 240_000); + +afterAll(async () => { + if (directory) await rm(directory, { recursive: true, force: true }); + 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); + + 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); + } + } + 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); + + 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`, + ); + + 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, + measurements: MEASUREMENTS, + }); + const row = await prisma.dataBackup.findUniqueOrThrow({ + where: { id: answer.data!.id }, + select: { userId: true, type: true }, + }); + expect(row.userId).toBe(adminId); + expect(row.type).toMatch(/^MANUAL_UPLOAD_\d+$/); + + // 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( + peakHeld, + `the upload held ${mb(peakHeld)} MB of a ${mb(fileBytes)} MB file`, + ).toBeLessThan(HELD_BUDGET_BYTES); + }, 600_000); +});