mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
perf(backup): take the weekly copy's preview from the writer, not a second parse
Since v1.39.6 the preview scanner parsed the JSON on its way into the pieces, kept every section but the three bulk tables, and validated the whole document at the end: a second copy of the record beside the writer's own, inside the weekly job. On an account with a document vault that is most of the record. The writer now shows each section and each bulk row to an observer just before writing it. The preview collects key uses from that and takes its counts from the writer's own FullBackupCounts; nothing is parsed or kept. An upload hands over the summary and key uses it has already validated. The round trip pins the writer's counts and key uses to a full read of the copy on an account with a row in every model, and a memory test holds what the pass keeps once the producer is done. Measured on 70,000 measurements, 3,000 mood entries and 40 one-megabyte documents under --max-old-space-size=524: held after the producer 56 MB before, 3 MB after; peak heapUsed 398 MB before, 334 MB after.
This commit is contained in:
Marc-André Bombeck committed
1 parent
5345c8be1e
commit
d55e329c48
10 files changed
+426
-144
No files matched your search
@@ -221,6 +221,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
|
||||
// answers with and the key uses it already holds (`backup-preview.ts`).
|
||||
const keys = new BackupKeyIdCollector();
|
||||
let created: { id: string };
|
||||
try {
|
||||
created = await storeBackupBlob(
|
||||
@@ -229,6 +232,16 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
userId: admin.id,
|
||||
type: uploadType,
|
||||
ownerAfterRead: () => owner!.id,
|
||||
preview: () =>
|
||||
payload
|
||||
? {
|
||||
summary: {
|
||||
...summarizeBackup(payload),
|
||||
measurements: measurementCount,
|
||||
},
|
||||
keys,
|
||||
}
|
||||
: null,
|
||||
},
|
||||
async (write) => {
|
||||
// One pass: every chunk is stored as it is checked, and a check that
|
||||
@@ -240,7 +253,6 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
}
|
||||
}
|
||||
let firstWithoutId: number | null = null;
|
||||
const keys = new BackupKeyIdCollector();
|
||||
let scanned;
|
||||
try {
|
||||
scanned = await scanBackupJson(storedAsRead(), {
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
/**
|
||||
* The preview worked out while a copy is written has to say what the preview
|
||||
* route used to derive by reading the stored copy: the same counts, the same
|
||||
* key uses. Held here against that derivation over the same file, fed in
|
||||
* pieces that split values, strings and multi-byte characters.
|
||||
* key uses. Held here against that derivation over the same file; the
|
||||
* integration round trip holds the writer's real counts to it on an account
|
||||
* with a row in every model.
|
||||
*/
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
|
||||
@@ -14,10 +15,11 @@ import {
|
||||
} from "@/lib/export/backup-key-ids";
|
||||
import {
|
||||
buildBackupPreview,
|
||||
createBackupPreviewScanner,
|
||||
createBackupPreviewCollector,
|
||||
storedCopyIdentity,
|
||||
storedPreviewFor,
|
||||
} from "@/lib/export/backup-preview";
|
||||
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
|
||||
import { backupPayloadSchema, summarizeBackup } from "@/lib/validations/backup";
|
||||
|
||||
const ts = "2026-07-19T07:00:00.000Z";
|
||||
@@ -69,52 +71,52 @@ const FILE = {
|
||||
],
|
||||
};
|
||||
|
||||
async function scan(pieces: Array<string | Uint8Array>) {
|
||||
const scanner = createBackupPreviewScanner();
|
||||
for (const piece of pieces) await scanner.feed(piece);
|
||||
return scanner.finish();
|
||||
}
|
||||
|
||||
function split(text: string, size: number): Uint8Array[] {
|
||||
const bytes = Buffer.from(text, "utf8");
|
||||
const out: Uint8Array[] = [];
|
||||
for (let at = 0; at < bytes.length; at += size) {
|
||||
out.push(bytes.subarray(at, at + size));
|
||||
/** What the weekly writer shows the collector, member by member. */
|
||||
const BULK = new Set(["measurements", "intakeEvents", "moodEntries"]);
|
||||
function observeAsTheWriterDoes(
|
||||
file: Record<string, unknown>,
|
||||
observe: (member: string, value: unknown) => void,
|
||||
) {
|
||||
for (const [member, value] of Object.entries(file)) {
|
||||
if (BULK.has(member))
|
||||
for (const row of value as unknown[]) observe(member, row);
|
||||
else observe(member, value);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
describe("createBackupPreviewScanner", () => {
|
||||
const json = JSON.stringify(FILE);
|
||||
// What the preview route derived from a stored copy.
|
||||
/** The writer's counts for FILE: every count the summary reads, and more. */
|
||||
function countsOf(file: typeof FILE): FullBackupCounts {
|
||||
const summary = summarizeBackup(backupPayloadSchema.parse(file));
|
||||
const { schemaVersion, userId, exportedAt, ...counts } = summary;
|
||||
void schemaVersion;
|
||||
void userId;
|
||||
void exportedAt;
|
||||
return { ...counts, medicationPauseEras: 0 } as unknown as FullBackupCounts;
|
||||
}
|
||||
|
||||
describe("createBackupPreviewCollector", () => {
|
||||
// What the preview route derives from a stored copy.
|
||||
const expected = summarizeBackup(backupPayloadSchema.parse(FILE));
|
||||
const expectedKeys = new BackupKeyIdCollector();
|
||||
expectedKeys.visit(FILE);
|
||||
|
||||
it.each([1, 7, 64, 100_000])(
|
||||
"counts what the full read counts, in pieces of %i bytes",
|
||||
async (size) => {
|
||||
const result = await scan(split(json, size));
|
||||
expect(result).not.toBeNull();
|
||||
expect(result!.summary).toEqual(expected);
|
||||
expect(result!.summary.measurements).toBe(3);
|
||||
expect(result!.summary.intakeEvents).toBe(2);
|
||||
expect(result!.summary.moodEntries).toBe(1);
|
||||
expect(result!.summary.illnessDayLogs).toBe(2);
|
||||
expect(result!.keys.toStored()).toEqual(expectedKeys.toStored());
|
||||
expect(result!.keys.keyIds()).toEqual(["v3"]);
|
||||
},
|
||||
);
|
||||
|
||||
it("takes text pieces the way the weekly writer hands them over", async () => {
|
||||
const result = await scan([json.slice(0, 50), json.slice(50)]);
|
||||
it("says what a full read of the copy says", () => {
|
||||
const collector = createBackupPreviewCollector();
|
||||
observeAsTheWriterDoes(FILE, collector.observe);
|
||||
const result = collector.finish(countsOf(FILE));
|
||||
expect(result).not.toBeNull();
|
||||
expect(result!.summary).toEqual(expected);
|
||||
expect(result!.summary.illnessDayLogs).toBe(2);
|
||||
expect(result!.keys.toStored()).toEqual(expectedKeys.toStored());
|
||||
expect(result!.keys.keyIds()).toEqual(["v3"]);
|
||||
});
|
||||
|
||||
it("answers null for a file that is not a backup, and never holds the writer", async () => {
|
||||
expect(await scan(["{ not json", "more", "and more"])).toBeNull();
|
||||
expect(await scan([JSON.stringify({ schemaVersion: 2 })])).toBeNull();
|
||||
expect(await scan([json.slice(0, 40)])).toBeNull();
|
||||
it("answers null for a copy that does not name its version, owner and date", () => {
|
||||
const collector = createBackupPreviewCollector();
|
||||
const { userId, ...rest } = FILE;
|
||||
void userId;
|
||||
observeAsTheWriterDoes(rest, collector.observe);
|
||||
expect(collector.finish(countsOf(FILE))).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -7,7 +7,12 @@
|
||||
* For an account of 1.8 million readings on slow hardware that took two
|
||||
* minutes, the dialog gave up long before, and the operator was told the
|
||||
* contents could not be read (#1031). The copy passes through this process
|
||||
* once anyway, on its way into the pieces, so the counts are taken there.
|
||||
* once anyway, on its way into the pieces, so the counts are taken there:
|
||||
* the weekly copy's from its writer's own counts and the sections it is
|
||||
* about to write, an upload's from the file the upload route has already
|
||||
* validated. Neither parses the copy a second time; the v1.39.6 scanner that
|
||||
* did kept every section but the bulk tables parsed beside the writer's own,
|
||||
* a second copy of the record inside the weekly job.
|
||||
*
|
||||
* What is kept. The `summarizeBackup` counts of the copy, its schema version,
|
||||
* and the inner key ids with the sections they sit in and the shortest
|
||||
@@ -28,13 +33,9 @@ import {
|
||||
BackupKeyIdCollector,
|
||||
type StoredKeyUse,
|
||||
} from "@/lib/export/backup-key-ids";
|
||||
import { scanBackupJson } from "@/lib/export/backup-json-scan";
|
||||
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
|
||||
import type { StoredBackupRef } from "@/lib/export/stored-backup";
|
||||
import {
|
||||
backupPayloadSchema,
|
||||
summarizeBackup,
|
||||
type BackupSummary,
|
||||
} from "@/lib/validations/backup";
|
||||
import type { BackupSummary } from "@/lib/validations/backup";
|
||||
|
||||
export interface BackupPreview {
|
||||
version: 1;
|
||||
@@ -138,98 +139,112 @@ export function storedPreviewFor(
|
||||
}
|
||||
|
||||
/**
|
||||
* The sections that grow with a record. Their elements are counted and their
|
||||
* key ids read as they pass, and never kept, so the scan holds no more of the
|
||||
* copy than the writer itself does (`full-backup-stream.ts`).
|
||||
* The summary fields that are counts, each also counted by the writer
|
||||
* (`FullBackupCounts`). The type below fails to compile when the summary
|
||||
* gains a count the writer does not report.
|
||||
*/
|
||||
const STREAMED = new Set(["measurements", "intakeEvents", "moodEntries"]);
|
||||
const SUMMARY_COUNTS = [
|
||||
"measurements",
|
||||
"medications",
|
||||
"intakeEvents",
|
||||
"medicationSideEffects",
|
||||
"medicationEfficacyTargets",
|
||||
"medicationScheduleRevisions",
|
||||
"moodEntries",
|
||||
"cycles",
|
||||
"cycleDayLogs",
|
||||
"labResults",
|
||||
"nutrientDays",
|
||||
"biomarkers",
|
||||
"illnessEpisodes",
|
||||
"illnessDayLogs",
|
||||
"allergies",
|
||||
"familyHistory",
|
||||
"workouts",
|
||||
"documents",
|
||||
"documentConditionLinks",
|
||||
"extractedFacts",
|
||||
"healthProfile",
|
||||
"healthProfileFactRevisions",
|
||||
"customMetrics",
|
||||
"customMetricEntries",
|
||||
"correlationPatterns",
|
||||
"intradayProfiles",
|
||||
"healthScoreRecords",
|
||||
"onboardingRecords",
|
||||
"practitioners",
|
||||
"encounters",
|
||||
"encounterLinks",
|
||||
"vaccinations",
|
||||
"vaccinationLinks",
|
||||
"measurementReminders",
|
||||
"measurementReminderEvents",
|
||||
"coachConversations",
|
||||
"coachMessages",
|
||||
"coachFacts",
|
||||
"coachPlans",
|
||||
"coachReminders",
|
||||
"mentalHealthAssessments",
|
||||
"consentReceipts",
|
||||
"personalRecords",
|
||||
"userAchievements",
|
||||
"environmentContexts",
|
||||
"environmentTravelLocations",
|
||||
"ecgRecordings",
|
||||
] as const satisfies readonly (keyof BackupSummary & keyof FullBackupCounts)[];
|
||||
|
||||
export interface BackupPreviewScanner {
|
||||
type UncountedSummaryField = Exclude<
|
||||
keyof BackupSummary,
|
||||
(typeof SUMMARY_COUNTS)[number] | "schemaVersion" | "userId" | "exportedAt"
|
||||
>;
|
||||
const everySummaryCountIsCounted: [UncountedSummaryField] extends [never]
|
||||
? true
|
||||
: UncountedSummaryField = true;
|
||||
void everySummaryCountIsCounted;
|
||||
|
||||
export interface BackupPreviewCollector {
|
||||
/** Pass as `observe` to `streamFullBackupJson`. */
|
||||
observe(member: string, value: unknown): void;
|
||||
/**
|
||||
* Hand over the next piece of the JSON. Resolves once the scan has taken
|
||||
* it, so a writer never runs more than a piece ahead of the scan.
|
||||
* The counts and key uses, once the writer has finished; null when the
|
||||
* copy did not name its schema version, owner and date.
|
||||
*/
|
||||
feed(piece: string | Uint8Array): Promise<void>;
|
||||
/**
|
||||
* The counts and key uses, once every piece has been fed; null when the
|
||||
* JSON was not a backup the schema accepts. Never rejects: a copy is
|
||||
* stored whether or not its preview could be worked out.
|
||||
*/
|
||||
finish(): Promise<{
|
||||
finish(counts: FullBackupCounts): {
|
||||
summary: BackupSummary;
|
||||
keys: BackupKeyIdCollector;
|
||||
} | null>;
|
||||
} | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Work out a preview from a copy's JSON as it is written. The counts are the
|
||||
* ones the preview route derives from the stored copy (`summarizeBackup` over
|
||||
* the parsed file), the three bulk sections counted rather than parsed.
|
||||
* Work out a weekly copy's preview from what its writer already has in hand:
|
||||
* each section as it is written, for the key uses, and the writer's own
|
||||
* counts. Nothing is parsed and nothing is kept but the key samples, so the
|
||||
* preview costs the pass no memory that grows with the record.
|
||||
*/
|
||||
export function createBackupPreviewScanner(): BackupPreviewScanner {
|
||||
const queue: Array<{ chunk: Uint8Array; taken: () => void }> = [];
|
||||
let ended = false;
|
||||
let stopped = false;
|
||||
let wake: (() => void) | null = null;
|
||||
const signal = () => {
|
||||
const w = wake;
|
||||
wake = null;
|
||||
w?.();
|
||||
};
|
||||
|
||||
async function* pieces(): AsyncGenerator<Uint8Array> {
|
||||
for (;;) {
|
||||
const next = queue.shift();
|
||||
if (next) {
|
||||
next.taken();
|
||||
yield next.chunk;
|
||||
continue;
|
||||
}
|
||||
if (ended) return;
|
||||
await new Promise<void>((resolve) => {
|
||||
wake = resolve;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function createBackupPreviewCollector(): BackupPreviewCollector {
|
||||
const keys = new BackupKeyIdCollector();
|
||||
const result = scanBackupJson(pieces(), {
|
||||
streamKeys: STREAMED,
|
||||
onElement: (key, element) => keys.visit(element, key),
|
||||
})
|
||||
.then(({ document, streamedCounts }) => {
|
||||
keys.visit(document);
|
||||
const parsed = backupPayloadSchema.safeParse(document);
|
||||
if (!parsed.success) return null;
|
||||
const summary: BackupSummary = {
|
||||
...summarizeBackup(parsed.data),
|
||||
measurements: streamedCounts.measurements ?? 0,
|
||||
intakeEvents: streamedCounts.intakeEvents ?? 0,
|
||||
moodEntries: streamedCounts.moodEntries ?? 0,
|
||||
};
|
||||
return { summary, keys };
|
||||
})
|
||||
.catch(() => null)
|
||||
.finally(() => {
|
||||
// A scan that stopped early takes nothing more: release the writer.
|
||||
stopped = true;
|
||||
for (const waiting of queue.splice(0)) waiting.taken();
|
||||
});
|
||||
|
||||
const header: Partial<
|
||||
Record<"schemaVersion" | "userId" | "exportedAt", string>
|
||||
> = {};
|
||||
return {
|
||||
feed(piece) {
|
||||
if (stopped) return Promise.resolve();
|
||||
const chunk =
|
||||
typeof piece === "string" ? Buffer.from(piece, "utf8") : piece;
|
||||
return new Promise<void>((taken) => {
|
||||
queue.push({ chunk, taken });
|
||||
signal();
|
||||
});
|
||||
observe(member, value) {
|
||||
if (
|
||||
(member === "schemaVersion" ||
|
||||
member === "userId" ||
|
||||
member === "exportedAt") &&
|
||||
typeof value === "string"
|
||||
) {
|
||||
header[member] = value;
|
||||
return;
|
||||
}
|
||||
keys.visit(value, member);
|
||||
},
|
||||
finish() {
|
||||
ended = true;
|
||||
signal();
|
||||
return result;
|
||||
finish(counts) {
|
||||
const { schemaVersion, userId, exportedAt } = header;
|
||||
if (!schemaVersion || !userId || !exportedAt) return null;
|
||||
const summary = { schemaVersion, userId, exportedAt } as BackupSummary;
|
||||
for (const field of SUMMARY_COUNTS) summary[field] = counts[field];
|
||||
return { summary, keys };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -79,14 +79,27 @@ const PROGRESS_EVERY_ROWS = 50_000;
|
||||
* The counts for the three deferred tables are filled in as their rows go
|
||||
* past, so a caller gets the same numbers the materialising builder reports.
|
||||
*/
|
||||
export interface StreamFullBackupOptions extends FullBackupOptions {
|
||||
/**
|
||||
* Shown each top-level member's value just before it is written, and each
|
||||
* row of the three bulk tables as it passes, under the member's name. For
|
||||
* a reader that needs to see the copy as it is written (the stored copy's
|
||||
* preview, `backup-preview.ts`) without parsing the JSON a second time.
|
||||
* It must not keep what it is shown: the writer releases each section as
|
||||
* soon as it is written, and that is what bounds its memory.
|
||||
*/
|
||||
observe?: (member: string, value: unknown) => void;
|
||||
}
|
||||
|
||||
export async function streamFullBackupJson(
|
||||
prisma: PrismaClient,
|
||||
userId: string,
|
||||
sink: BackupJsonSink,
|
||||
options: FullBackupOptions = {},
|
||||
options: StreamFullBackupOptions = {},
|
||||
): Promise<FullBackupCounts> {
|
||||
const { observe, ...payloadOptions } = options;
|
||||
const { payload, counts } = await buildFullBackupPayload(prisma, userId, {
|
||||
...options,
|
||||
...payloadOptions,
|
||||
deferBulk: true,
|
||||
});
|
||||
|
||||
@@ -126,6 +139,7 @@ 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)}`,
|
||||
);
|
||||
@@ -137,6 +151,7 @@ export async function streamFullBackupJson(
|
||||
await write("]");
|
||||
bulkCounts[key] = rows;
|
||||
} else {
|
||||
observe?.(key, value);
|
||||
await write(JSON.stringify(value));
|
||||
}
|
||||
delete payload[key];
|
||||
|
||||
@@ -35,9 +35,9 @@ import {
|
||||
type PackBackupOptions,
|
||||
} from "@/lib/export/backup-blob";
|
||||
import { createBackupKeyIdTextScanner } from "@/lib/export/backup-key-ids";
|
||||
import type { BackupKeyIdCollector } from "@/lib/export/backup-key-ids";
|
||||
import {
|
||||
buildBackupPreview,
|
||||
createBackupPreviewScanner,
|
||||
SINGLE_VALUE_IDENTITY_HEAD,
|
||||
singleValueIdentity,
|
||||
storedCopyIdentity,
|
||||
@@ -48,6 +48,7 @@ import {
|
||||
newChunkStreamId,
|
||||
} from "@/lib/export/backup-chunks";
|
||||
import { BackupFormsConflictError } from "@/lib/export/stored-backup";
|
||||
import type { BackupSummary } from "@/lib/validations/backup";
|
||||
|
||||
/**
|
||||
* How long the storing transaction may stay open. The job around it expires
|
||||
@@ -79,6 +80,16 @@ export interface StoreBackupBlobInput {
|
||||
* the row under the wrong owner.
|
||||
*/
|
||||
ownerAfterRead?: () => string;
|
||||
/**
|
||||
* What the restore preview shows for the copy, asked once the producer has
|
||||
* finished (`backup-preview.ts`): the producer knows the counts and the
|
||||
* key uses without reading the copy back. Null, or absent, leaves the row
|
||||
* without a preview, and the preview route reads the copy once instead.
|
||||
*/
|
||||
preview?: () => {
|
||||
summary: BackupSummary;
|
||||
keys: BackupKeyIdCollector;
|
||||
} | null;
|
||||
}
|
||||
|
||||
type Tx = Parameters<Parameters<PrismaClient["$transaction"]>[0]>[0];
|
||||
@@ -126,9 +137,6 @@ export async function storeBackupBlob(
|
||||
// what is inside them, so this is what tells an operator how long a
|
||||
// retired key is still needed (`GET /api/admin/encryption/status`).
|
||||
const keyScanner = createBackupKeyIdTextScanner();
|
||||
// And what the restore preview shows, from the same pass, so the
|
||||
// preview never has to read the copy back (`backup-preview.ts`).
|
||||
const previewScanner = createBackupPreviewScanner();
|
||||
const { chunks, bytes } = await packBackupChunks(
|
||||
async (sealed, seq) => {
|
||||
await tx.dataBackupChunk.create({
|
||||
@@ -140,12 +148,11 @@ export async function storeBackupBlob(
|
||||
(write) =>
|
||||
producer(async (chunk) => {
|
||||
keyScanner.feed(chunk);
|
||||
await previewScanner.feed(chunk);
|
||||
await write(chunk);
|
||||
}),
|
||||
options,
|
||||
);
|
||||
const scanned = await previewScanner.finish();
|
||||
const scanned = input.preview?.() ?? null;
|
||||
|
||||
// Again at the end: a restore may have been queued while this ran. The
|
||||
// window left is the few milliseconds to the commit, and a restore
|
||||
|
||||
@@ -121,6 +121,7 @@ describe("handleDataBackup canonical DR payload", () => {
|
||||
expect(mocks.store.mock.calls[0]![0]).toEqual({
|
||||
userId: "user-dr",
|
||||
type: "WEEKLY_AUTO",
|
||||
preview: expect.any(Function),
|
||||
});
|
||||
const encrypted = mocks.store.mock.calls[0]![1] as string;
|
||||
const payload = JSON.parse(encrypted) as {
|
||||
|
||||
@@ -12,6 +12,8 @@ import {
|
||||
BackupBusyError,
|
||||
} from "@/lib/export/backup-blob";
|
||||
import { storeBackupBlob } from "@/lib/export/store-backup-blob";
|
||||
import { createBackupPreviewCollector } from "@/lib/export/backup-preview";
|
||||
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
|
||||
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
|
||||
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
@@ -182,13 +184,23 @@ export async function handleDataBackup(
|
||||
// time, and gzip's output is sealed and stored in pieces of about a
|
||||
// megabyte (`storeBackupBlob`). The copy is kept as those pieces,
|
||||
// so its size does not depend on this process's memory (#1031).
|
||||
// The restore preview is taken from the writer's own counts and
|
||||
// the sections as they are written, never by parsing the copy.
|
||||
const preview = createBackupPreviewCollector();
|
||||
let counts: FullBackupCounts | undefined;
|
||||
const { bytes: storedBytes } = await storeBackupBlob(
|
||||
prisma,
|
||||
{ userId: user.id, type: "WEEKLY_AUTO" },
|
||||
(write) =>
|
||||
streamFullBackupJson(prisma, user.id, write, {
|
||||
{
|
||||
userId: user.id,
|
||||
type: "WEEKLY_AUTO",
|
||||
preview: () => (counts ? preview.finish(counts) : null),
|
||||
},
|
||||
async (write) => {
|
||||
counts = await streamFullBackupJson(prisma, user.id, write, {
|
||||
purpose: "disaster-recovery",
|
||||
}),
|
||||
observe: preview.observe,
|
||||
});
|
||||
},
|
||||
);
|
||||
largestBlobBytes = Math.max(largestBlobBytes, storedBytes);
|
||||
backed++;
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
/**
|
||||
* The preview a weekly copy carries is worked out from the counts the writer
|
||||
* already produces, not by parsing the copy a second time.
|
||||
*
|
||||
* What this pins. From v1.39.6 the preview scanner parsed the JSON on its way
|
||||
* into the pieces, kept every section that is not one of the three bulk tables
|
||||
* (documents with their ciphertext, coach turns, profiles), and validated the
|
||||
* whole document at the end. On an account with a document vault that is a
|
||||
* second full copy of the record beside the writer's own, inside the weekly
|
||||
* job that shares the app's heap.
|
||||
*
|
||||
* How it measures. Like `backup-streaming-memory.test.ts`: every reading is
|
||||
* taken after a forced collection, so what is compared is what the pass
|
||||
* HOLDS. The reading at the end of the producer is the one that matters: the
|
||||
* writer has released its sections by then, so anything still held belongs
|
||||
* to the preview. Run under `--max-old-space-size=524` (the V8 limit of a
|
||||
* 1 GB container) for the production ceiling.
|
||||
*/
|
||||
import { randomBytes } from "node:crypto";
|
||||
import v8 from "node:v8";
|
||||
import vm from "node:vm";
|
||||
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
|
||||
import { createBackupPreviewCollector } from "@/lib/export/backup-preview";
|
||||
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
|
||||
import { storeBackupBlob } from "@/lib/export/store-backup-blob";
|
||||
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const OWNER_ID = "backup-preview-memory-owner";
|
||||
const MEASUREMENT_ROWS = 70_000;
|
||||
const MOOD_ROWS = 3_000;
|
||||
const DOCUMENTS = 40;
|
||||
const DOCUMENT_BYTES = 1024 * 1024;
|
||||
|
||||
/**
|
||||
* What the pass may still hold once the producer is done. Measured: the
|
||||
* scanner that kept the sections held about twice the document vault at
|
||||
* that point (the parsed sections and the base64 text inside them); the
|
||||
* pass without it holds a few megabytes.
|
||||
*/
|
||||
const HELD_AT_END_BUDGET_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
const prisma = getPrismaClient();
|
||||
|
||||
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);
|
||||
|
||||
async function seed(): Promise<void> {
|
||||
await prisma.user.create({
|
||||
data: { id: OWNER_ID, username: "backup-preview-memory" },
|
||||
});
|
||||
await prisma.$executeRawUnsafe(
|
||||
`INSERT INTO measurements (
|
||||
id, user_id, type, value, unit, source, measured_at, notes,
|
||||
notes_encrypted, external_id, created_at, updated_at, sync_version)
|
||||
SELECT
|
||||
'pm' || lpad(g::text, 10, '0'), $1, 'PULSE'::measurement_type,
|
||||
60 + (g % 40), 'bpm', 'APPLE_HEALTH'::measurement_source,
|
||||
timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'),
|
||||
CASE WHEN g % 40 = 0 THEN 'a note recorded with reading ' || g END,
|
||||
CASE WHEN g % 25 = 0
|
||||
THEN decode(md5(g::text) || md5((g + 1)::text), 'hex') END,
|
||||
'preview-' || g,
|
||||
timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'),
|
||||
timestamp '2019-01-01 00:00:00' + (g * interval '30 seconds'),
|
||||
1
|
||||
FROM generate_series(1, ${MEASUREMENT_ROWS}) AS g`,
|
||||
OWNER_ID,
|
||||
);
|
||||
await prisma.$executeRawUnsafe(
|
||||
`INSERT INTO mood_entries (
|
||||
id, user_id, date, mood, score, source, mood_logged_at, synced_at,
|
||||
created_at, updated_at, tz, note, sync_version)
|
||||
SELECT
|
||||
'pd' || lpad(g::text, 10, '0'), $1,
|
||||
to_char(timestamp '2019-01-01' + (g * interval '1 day'), 'YYYY-MM-DD'),
|
||||
'okay', 3, 'MOODLOG', timestamp '2019-01-01' + (g * interval '1 day'),
|
||||
now(), now(), now(), 'Europe/Berlin', 'a journal line about day ' || g, 1
|
||||
FROM generate_series(1, ${MOOD_ROWS}) AS g`,
|
||||
OWNER_ID,
|
||||
);
|
||||
for (let i = 0; i < DOCUMENTS; i++) {
|
||||
await prisma.inboundDocument.create({
|
||||
data: {
|
||||
userId: OWNER_ID,
|
||||
mimeType: "application/pdf",
|
||||
byteSize: DOCUMENT_BYTES,
|
||||
contentEncrypted: new Uint8Array(randomBytes(DOCUMENT_BYTES)),
|
||||
contentCodec: "binary2",
|
||||
title: `document ${i}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
describe("weekly copy preview under a memory budget", () => {
|
||||
beforeAll(async () => {
|
||||
await truncateAllTables(prisma);
|
||||
await seed();
|
||||
}, 240_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await truncateAllTables(prisma);
|
||||
});
|
||||
|
||||
it("holds nothing of the record once the producer is done", async () => {
|
||||
const baseline = liveHeapBytes();
|
||||
let peakHeld = 0;
|
||||
let heldAtEnd = 0;
|
||||
let rawPeak = 0;
|
||||
let chunks = 0;
|
||||
const poll = setInterval(() => {
|
||||
rawPeak = Math.max(rawPeak, process.memoryUsage().heapUsed);
|
||||
}, 5);
|
||||
// Wired exactly as the weekly job wires it (`backup-handlers.ts`).
|
||||
const preview = createBackupPreviewCollector();
|
||||
let counts: FullBackupCounts | undefined;
|
||||
try {
|
||||
await storeBackupBlob(
|
||||
prisma,
|
||||
{
|
||||
userId: OWNER_ID,
|
||||
type: "WEEKLY_AUTO",
|
||||
preview: () => (counts ? preview.finish(counts) : null),
|
||||
},
|
||||
async (write) => {
|
||||
counts = await streamFullBackupJson(
|
||||
prisma,
|
||||
OWNER_ID,
|
||||
async (chunk) => {
|
||||
await write(chunk);
|
||||
if (chunks++ % 40 === 0) {
|
||||
peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline);
|
||||
}
|
||||
},
|
||||
{ purpose: "disaster-recovery", observe: preview.observe },
|
||||
);
|
||||
heldAtEnd = liveHeapBytes() - baseline;
|
||||
peakHeld = Math.max(peakHeld, heldAtEnd);
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
clearInterval(poll);
|
||||
}
|
||||
rawPeak = Math.max(rawPeak, process.memoryUsage().heapUsed);
|
||||
process.stderr.write(
|
||||
`[preview-memory] held at end of producer ${mb(heldAtEnd)} MB, ` +
|
||||
`peak held ${mb(peakHeld)} MB, raw peak heapUsed ${mb(rawPeak)} MB ` +
|
||||
`(baseline ${mb(baseline)} MB), ` +
|
||||
`${mb(v8.getHeapStatistics().heap_size_limit)} MB heap limit\n`,
|
||||
);
|
||||
|
||||
const row = await prisma.dataBackup.findFirstOrThrow({
|
||||
where: { userId: OWNER_ID, type: "WEEKLY_AUTO" },
|
||||
select: { preview: true },
|
||||
});
|
||||
expect(row.preview).toMatchObject({
|
||||
summary: {
|
||||
measurements: MEASUREMENT_ROWS,
|
||||
moodEntries: MOOD_ROWS,
|
||||
documents: DOCUMENTS,
|
||||
},
|
||||
});
|
||||
expect(
|
||||
heldAtEnd,
|
||||
`the pass still held ${mb(heldAtEnd)} MB after the producer finished`,
|
||||
).toBeLessThan(HELD_AT_END_BUDGET_BYTES);
|
||||
}, 600_000);
|
||||
});
|
||||
@@ -9,8 +9,9 @@
|
||||
* and its preview kept; a key dropped after the copy was written is still
|
||||
* named, because the verdict is taken again at every read.
|
||||
*
|
||||
* Mutations that must turn this red: drop the preview scanner from
|
||||
* `storeBackupBlob` (no preview on the row, the route opens the copy), or
|
||||
* Mutations that must turn this red: drop the `preview` the weekly job or
|
||||
* the upload route hands `storeBackupBlob` (no preview on the row, the route
|
||||
* opens the copy), or
|
||||
* answer from the stored verdict instead of taking it again (the dropped key
|
||||
* is not named).
|
||||
*/
|
||||
|
||||
@@ -78,6 +78,8 @@ import {
|
||||
import { legacyStreamedBlobFrom } from "@/__tests__/helpers/legacy-backup-blob";
|
||||
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
|
||||
import { BackupKeyIdCollector } from "@/lib/export/backup-key-ids";
|
||||
import { createBackupPreviewCollector } from "@/lib/export/backup-preview";
|
||||
import { parseBackupPayload, summarizeBackup } from "@/lib/validations/backup";
|
||||
import {
|
||||
TWO_ENDED_MODELS,
|
||||
USER_COLUMN_BACKUP_CLASS,
|
||||
@@ -1492,13 +1494,18 @@ describe("every model the plan claims two-ended survives a real restore", () =>
|
||||
// below then covers BOTH, since the row that gets restored comes from the
|
||||
// streamed bytes.
|
||||
let streamedJson = "";
|
||||
await streamFullBackupJson(
|
||||
const previewCollector = createBackupPreviewCollector();
|
||||
const streamedCounts = await streamFullBackupJson(
|
||||
prisma,
|
||||
OWNER_ID,
|
||||
(chunk) => {
|
||||
streamedJson += chunk;
|
||||
},
|
||||
{ purpose: "disaster-recovery", exportedAt },
|
||||
{
|
||||
purpose: "disaster-recovery",
|
||||
exportedAt,
|
||||
observe: previewCollector.observe,
|
||||
},
|
||||
);
|
||||
expect(
|
||||
streamedJson,
|
||||
@@ -1506,6 +1513,20 @@ describe("every model the plan claims two-ended survives a real restore", () =>
|
||||
"weekly backup no longer contains what this test proves restorable",
|
||||
).toBe(JSON.stringify(payload));
|
||||
|
||||
// The weekly copy's restore preview is taken from the writer's counts and
|
||||
// the sections it was shown, never by reading the copy. On an account
|
||||
// with a row in every model, it has to say what a full read of the copy
|
||||
// says, count for count and key use for key use.
|
||||
const fromWriter = previewCollector.finish(streamedCounts);
|
||||
expect(fromWriter).not.toBeNull();
|
||||
expect(
|
||||
fromWriter!.summary,
|
||||
"the preview taken from the writer's counts disagrees with a full read",
|
||||
).toEqual(summarizeBackup(parseBackupPayload(streamedJson)));
|
||||
const readBack = new BackupKeyIdCollector();
|
||||
readBack.visit(JSON.parse(streamedJson));
|
||||
expect(fromWriter!.keys.toStored()).toEqual(readBack.toStored());
|
||||
|
||||
// The tombstone has to be IN the file. A disaster-recovery payload that
|
||||
// carries only live rows is smaller and cheaper and wrong: the restore
|
||||
// wipes the account first, so a deletion that is not in the file is a
|
||||
@@ -3757,13 +3778,26 @@ describe("every column of every two-ended model survives a real restore", () =>
|
||||
|
||||
const exportedAt = new Date("2026-08-01T00:00:00.000Z");
|
||||
let json = "";
|
||||
await streamFullBackupJson(
|
||||
const previewCollector = createBackupPreviewCollector();
|
||||
const counts = await streamFullBackupJson(
|
||||
prisma,
|
||||
OWNER_ID,
|
||||
(chunk) => {
|
||||
json += chunk;
|
||||
},
|
||||
{ purpose: "disaster-recovery", exportedAt },
|
||||
{
|
||||
purpose: "disaster-recovery",
|
||||
exportedAt,
|
||||
observe: previewCollector.observe,
|
||||
},
|
||||
);
|
||||
// Every sealed column is filled here: the key uses the preview takes from
|
||||
// the sections as they are written are the ones a read of the file finds.
|
||||
const keysReadBack = new BackupKeyIdCollector();
|
||||
keysReadBack.visit(JSON.parse(json));
|
||||
expect(keysReadBack.keyIds().length).toBeGreaterThan(0);
|
||||
expect(previewCollector.finish(counts)!.keys.toStored()).toEqual(
|
||||
keysReadBack.toStored(),
|
||||
);
|
||||
|
||||
// ── No ciphertext in a portable file ──────────────────────────────────
|
||||
|
||||
Reference in new issue
Block a user