fix(backup): stream every large section of an uploaded backup

The admin upload streamed only the measurements while it checked a file,
so a document-heavy account copy was held whole in memory and an upload of
a few hundred megabytes could take the process down.

The sections a reader takes an element at a time are now named once
(BACKUP_STREAMED_SECTIONS in streamed-backup.ts): measurements, intake
events, mood entries, documents and ECG recordings. The upload reads its
file through readStreamedBackup like the restore and the preview, and the
restore drill and the restore script use the same key set.

A behavioural test pushes a 271 MB document-heavy file through the real
upload route and fails when the heap grows with the file. The stream guard
now asserts that every reader goes through the shared reader or key set
and that no file declares its own.

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

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,

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"]),
         });

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<Uint8Array>;
 
 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<string> = 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<string, unknown[]>();
   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(

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) {