fix(backup): read ECG traces a strip at a time

The shared backup reader kept documents without their stored bytes but
kept every ECG strip whole, trace included. A long watch history is
hundreds of strips of tens to hundreds of kilobytes each, so an upload,
preview or restore of such a copy held the whole history in memory.

The first read now keeps each strip without its trace (a short stand-in,
after the key it was written under is recorded and its form checked),
and forEachEcgTrace reads the traces again one strip at a time. The ECG
restore writes each strip from that second read, as the document restore
already did, and refuses a file whose two reads disagree on the count.

The upload memory test gains an ECG-heavy file: 150 strips, about 200 MB.
The old reader held 197 MB of it and failed the 48 MB bound; the new one
holds about 1 MB, also under a 524 MB heap.
This commit is contained in:
Marc-André Bombeck committed 2026-10-25 14:27:56 +01:00
1 parent 0c5f29303b
commit 459b5baa9d
4 files changed
+317 -93

No files matched your search

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