mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
Merge pull request #920 from MBombeck/fix/backup-heap-budget
fix(backup): bound the blob, not the heap, so the weekly pass runs again
This commit is contained in:
13 files changed
+575
-107
No files matched your search
@@ -1,5 +1,36 @@
|
||||
# Changelog
|
||||
|
||||
## [1.38.7] — 2026-09-03
|
||||
|
||||
The weekly backup actually produces a backup again. v1.38.6 stopped it
|
||||
taking the instance down when it failed, and also stopped it succeeding.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **The weekly backup refused every account because the server had been up a
|
||||
while.** The writer added in v1.38.6 watched the process's live heap and gave
|
||||
up above 80 % of V8's limit. That reading is not the backup's footprint; it
|
||||
is the whole process's, garbage included, and a long-running app sits at
|
||||
400 MB of heap that a collection would take back. On a 1 GB container the
|
||||
budget is 419 MB, so the pass aborted the first chunk it wrote for every
|
||||
account: four out of four in seven seconds, one of them a demo record whose
|
||||
entire stored copy is 1.2 MB, all of them told that their record was too
|
||||
large and that the operator should buy memory. An affected instance can
|
||||
sit for a month and a half with no usable copy. The heap gauge is gone. What is bounded instead
|
||||
is the one copy the pipeline cannot stream away: the stored blob, counted in
|
||||
the bytes it actually produced, against a fifth of this process's heap limit.
|
||||
An account whose backup genuinely does not fit still fails as a job rather
|
||||
than taking the instance down with it, and the message now names what was
|
||||
measured and what it crossed.
|
||||
- **A backup pass that wrote nothing for anybody reported success.** `backed: 0`
|
||||
came back as a completed job, so the failing-queue panel stayed empty and the
|
||||
backups page listed the copies it already had with their perfectly ordinary
|
||||
timestamps. A pass that protected nobody is now a failed run, which is what
|
||||
the backups page reads back when it says the last scheduled run did not
|
||||
finish. A pass where some accounts were written and others were not still
|
||||
passes, with the per-account failures carried as counts, because failing the
|
||||
queue over one record would re-run the whole cohort on every retry.
|
||||
|
||||
## [1.38.6] — 2026-09-03
|
||||
|
||||
Blood glucose can be shown in mmol/L, which it never could before, and the
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
openapi: 3.1.0
|
||||
info:
|
||||
title: HealthLog API
|
||||
version: 1.38.6
|
||||
version: 1.38.7
|
||||
description: >-
|
||||
Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
|
||||
|
||||
|
||||
+25
-11
@@ -150,15 +150,28 @@ Two things changed, and both matter to an operator:
|
||||
released as soon as its JSON exists. On the same fixture doubled to 890 000
|
||||
measurements, the pass completes inside a 296 MB heap limit; before the
|
||||
change, half that record did not fit in 546 MB.
|
||||
- **It stops itself.** The writer watches its own live heap and aborts at 80 %
|
||||
of V8's limit. That backup then fails for that one account, is counted in the
|
||||
run's `users_failed` and `heap_budget_trips` meta, and the pass carries on
|
||||
with everybody else. A memory failure is now a failed job rather than a
|
||||
restart for every user on the host.
|
||||
- **It stops itself, on its own size.** The envelope writer counts the
|
||||
encrypted bytes it has produced for an account and gives up on the account
|
||||
whose stored copy would not fit this process — a fifth of V8's heap limit,
|
||||
which is 105 MB on the 524 MB limit a 1 GB container gets. That backup then
|
||||
fails for that one account, is counted in the run's `users_failed` and
|
||||
`records_oversized` meta, and the pass carries on with everybody else. A
|
||||
memory failure is a failed job rather than a restart for every user on the
|
||||
host.
|
||||
|
||||
If `heap_budget_trips` is non-zero in `job.data_backup`, the answer is more
|
||||
If `records_oversized` is non-zero in `job.data_backup`, the answer is more
|
||||
memory, not a retry: raise the container's limit, or set
|
||||
`NODE_OPTIONS=--max-old-space-size=<MB>` to something under it.
|
||||
`NODE_OPTIONS=--max-old-space-size=<MB>` to something under it. The limit is
|
||||
derived from the heap limit, so it rises with it.
|
||||
|
||||
The check deliberately reads bytes it produced and not the process's heap. A
|
||||
version that read the heap shipped in v1.38.6 and compared the whole process's
|
||||
live usage — garbage included — against 80 % of the limit. A Next.js server
|
||||
that has been up for a week sits at 400 MB of largely collectable heap, so the
|
||||
weekly pass aborted every account on the first chunk, including one whose whole
|
||||
stored copy is 1.2 MB, and reported success while doing it. If you are on
|
||||
v1.38.6, a `data-backup` run that finishes in seconds having backed up nothing
|
||||
is that defect and not your record.
|
||||
|
||||
### The stored column is the remaining ceiling
|
||||
|
||||
@@ -167,10 +180,11 @@ exist as one value before it can be written — that copy cannot be streamed
|
||||
away. It is the only thing left in the job that grows with the record: at a
|
||||
compressed blob of ~42 MB the pass peaks around 236 MB, and the growth from
|
||||
there is roughly twice the blob's size (the base64 answer, plus the driver's
|
||||
copy of it on the way to the wire). A blob past roughly 150 MB — an account
|
||||
several times larger than any seen so far — will hit a 524 MB limit again, and
|
||||
the fix at that point is to stop storing the artifact in one column: chunk it
|
||||
across rows, or keep only the off-host copy. Reading it back has the same
|
||||
copy of it on the way to the wire). That is the ceiling the size check is set
|
||||
against — a fifth of the heap limit, so two copies plus the process's own live
|
||||
set still fit — and an account that reaches it fails as a job with a message
|
||||
naming both numbers. The fix at that point is to stop storing the artifact in
|
||||
one column: chunk it across rows, or keep only the off-host copy. Reading it back has the same
|
||||
shape, and worse: a restore parses the whole document, so the read path needs
|
||||
several times the blob in heap. An operator restoring a very large account
|
||||
should give the container more memory for the duration.
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "healthlog",
|
||||
"version": "1.38.6",
|
||||
"version": "1.38.7",
|
||||
"description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.",
|
||||
"license": "PolyForm-Noncommercial-1.0.0",
|
||||
"homepage": "https://healthlog.dev",
|
||||
|
||||
+1
-1
@@ -36,7 +36,7 @@ try {
|
||||
// v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild.
|
||||
const CACHE_VERSION =
|
||||
(typeof self !== "undefined" && self.__APP_VERSION__) ||
|
||||
/* @sw-version-fallback */ "v1.38.6";
|
||||
/* @sw-version-fallback */ "v1.38.7";
|
||||
const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`;
|
||||
const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`;
|
||||
// v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*`
|
||||
|
||||
@@ -17,10 +17,14 @@
|
||||
process.env.ENCRYPTION_KEY ??=
|
||||
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
|
||||
|
||||
import v8 from "node:v8";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { encrypt } from "@/lib/crypto";
|
||||
import {
|
||||
BackupBlobTooLargeError,
|
||||
defaultBackupBlobLimit,
|
||||
packBackupBlob,
|
||||
packBackupBlobStreaming,
|
||||
unpackBackupBlob,
|
||||
@@ -133,4 +137,79 @@ describe("backup blob envelope", () => {
|
||||
}),
|
||||
).rejects.toThrow("row source failed");
|
||||
});
|
||||
|
||||
it("stops the account whose stored copy would not fit this process", async () => {
|
||||
// The stored blob is the one copy the pipeline has to hold whole, so it is
|
||||
// the one thing here that grows without bound as a record grows. 4 KB is
|
||||
// an absurd limit; the point is that the writer counts what it produced
|
||||
// and stops on that, not on how full the heap happened to be.
|
||||
const err = await packBackupBlobStreaming(
|
||||
async (write) => {
|
||||
await write(sampleJson(5_000));
|
||||
},
|
||||
{ maxBytes: 4_096 },
|
||||
).catch((e: unknown) => e);
|
||||
|
||||
expect(err).toBeInstanceOf(BackupBlobTooLargeError);
|
||||
const failure = err as BackupBlobTooLargeError;
|
||||
expect(failure.limitBytes).toBe(4_096);
|
||||
expect(failure.bytes).toBeGreaterThan(4_096);
|
||||
// The message has to describe what was measured. It says how much
|
||||
// ciphertext this account produced and what it may occupy — not how full
|
||||
// the process's heap was, which is what the check this replaced reported.
|
||||
expect(failure.message).toContain("of encrypted backup for one account");
|
||||
expect(failure.message).toContain("over the 4 KB");
|
||||
// The check this replaced reported "aborted at N MB of heap", which was a
|
||||
// reading of the whole process and said nothing about the account.
|
||||
expect(failure.message).not.toContain("of heap");
|
||||
});
|
||||
|
||||
it("does not stop a record that fits, however dirty the heap is", async () => {
|
||||
// Collectable garbage past 419 MB: 80 % of the 524 MB V8 limit a 1 GB
|
||||
// container gets, which is the exact budget that aborted all four accounts
|
||||
// on the live instance — the smallest of them a 1.2 MB demo record.
|
||||
const CONTAINER_BUDGET = Math.floor(524 * 1024 * 1024 * 0.8);
|
||||
// Capped against this process's own limit, so a runner started with a
|
||||
// smaller heap measures a backup rather than an out-of-memory abort.
|
||||
const target = Math.min(
|
||||
CONTAINER_BUDGET + 8 * 1024 * 1024,
|
||||
Math.floor(v8.getHeapStatistics().heap_size_limit * 0.6),
|
||||
);
|
||||
let garbage: unknown[] = [];
|
||||
for (
|
||||
let round = 0;
|
||||
round < 2_000 && process.memoryUsage().heapUsed < target;
|
||||
round++
|
||||
) {
|
||||
const chunk = new Array(30_000);
|
||||
for (let at = 0; at < 30_000; at++) chunk[at] = { k: round * at };
|
||||
garbage.push(chunk);
|
||||
}
|
||||
const dirty = process.memoryUsage().heapUsed;
|
||||
garbage = [];
|
||||
void garbage;
|
||||
|
||||
const blob = await packStreamed(sampleJson(200));
|
||||
expect(unpackBackupBlob(blob)).toBe(sampleJson(200));
|
||||
// Not a vacuous pass: the heap really was carrying the pile it aimed at
|
||||
// — a 1 GB container's whole budget where the runner has room for it —
|
||||
// when the backup ran.
|
||||
expect(dirty).toBeGreaterThanOrEqual(target);
|
||||
expect(dirty).toBeGreaterThan(64 * 1024 * 1024);
|
||||
});
|
||||
|
||||
it("derives the default limit from the heap limit, not from heap usage", async () => {
|
||||
const limit = defaultBackupBlobLimit();
|
||||
const before = limit;
|
||||
// Allocate and hold: usage moves, the limit must not.
|
||||
const held: unknown[] = [];
|
||||
for (let round = 0; round < 40; round++) {
|
||||
const chunk = new Array(30_000);
|
||||
for (let at = 0; at < 30_000; at++) chunk[at] = { k: round * at };
|
||||
held.push(chunk);
|
||||
}
|
||||
expect(defaultBackupBlobLimit()).toBe(before);
|
||||
expect(held).toHaveLength(40);
|
||||
expect(limit).toBeGreaterThan(16 * 1024 * 1024);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,48 @@
|
||||
/**
|
||||
* The streaming writer must not gate a backup on a heap gauge.
|
||||
*
|
||||
* It did, for one release. Every flush read `process.memoryUsage().heapUsed`
|
||||
* and aborted above 80 % of V8's heap limit — a reading of the whole process,
|
||||
* garbage included. A Next.js server that has been up a week sits at 400 MB of
|
||||
* largely collectable heap, so on a 1 GB container the weekly pass aborted the
|
||||
* first chunk of every account: four out of four, in seven seconds, one of
|
||||
* them a demo record whose entire stored copy is 1.2 MB. The failure told the
|
||||
* operator the record was too large and to buy memory.
|
||||
*
|
||||
* The bound that replaced it counts the bytes the backup itself produced, in
|
||||
* `packBackupBlobStreaming`, where the only copy that grows with a record
|
||||
* actually accumulates. This test is the tripwire for putting the gauge back.
|
||||
*
|
||||
* It is a grep with the limits of one: a heap reading spelled some other way —
|
||||
* a `v8.getHeapStatistics()` destructure, an import alias — would slip it. It
|
||||
* catches the shape that was actually written, which is the shape a revert
|
||||
* would reintroduce.
|
||||
*/
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
const source = readFileSync(
|
||||
join(import.meta.dirname, "..", "full-backup-stream.ts"),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
describe("full-backup-stream", () => {
|
||||
it("reads no process or V8 memory gauge", () => {
|
||||
const code = source
|
||||
.split("\n")
|
||||
.filter((line) => !line.trimStart().startsWith("*"))
|
||||
.join("\n");
|
||||
|
||||
expect(code).not.toMatch(/memoryUsage\s*\(/);
|
||||
expect(code).not.toMatch(/getHeapStatistics\s*\(/);
|
||||
expect(code).not.toMatch(/heap_size_limit/);
|
||||
expect(code).not.toMatch(/from\s+"node:v8"/);
|
||||
});
|
||||
|
||||
it("still explains why, so the next reader does not re-add it", () => {
|
||||
// A tripwire with no reason attached gets deleted by whoever trips it.
|
||||
expect(source).toContain("no heap reading in this writer");
|
||||
});
|
||||
});
|
||||
@@ -36,6 +36,7 @@
|
||||
* usable copy predates any of this is exactly the person who needs it to work.
|
||||
*/
|
||||
import { Buffer } from "node:buffer";
|
||||
import v8 from "node:v8";
|
||||
import { createGzip, gunzipSync, gzipSync } from "node:zlib";
|
||||
|
||||
import {
|
||||
@@ -59,6 +60,64 @@ export function packBackupBlob(json: string): string {
|
||||
return encrypt(`${GZIP_MARKER}${compressed}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* The share of this process's heap limit one stored backup may become.
|
||||
*
|
||||
* The blob is the single copy this pipeline cannot stream away:
|
||||
* `data_backups.data` is one `text` column, so the row has to be one value,
|
||||
* and the driver makes a second copy of it on its way to the wire. Two copies
|
||||
* of the answer plus whatever the process legitimately holds is what has to
|
||||
* fit at once, so a fifth of the heap limit is one account's share.
|
||||
*
|
||||
* It is a share of the LIMIT, not of what is currently used. How large a
|
||||
* single value this process can hold depends on how it was started; it does
|
||||
* not depend on how long it has been running or on how much garbage is
|
||||
* waiting to be collected. The check this replaced read the latter, and so
|
||||
* aborted a 1.2 MB record on a server that had merely been up for a week.
|
||||
*/
|
||||
const BLOB_HEAP_SHARE = 0.2;
|
||||
|
||||
/** Bytes as an operator reads them. Kilobytes below a megabyte. */
|
||||
function size(bytes: number): string {
|
||||
const mb = bytes / 1024 / 1024;
|
||||
if (bytes < 1024 * 1024) return `${Math.round(bytes / 1024)} KB`;
|
||||
return mb < 100 ? `${mb.toFixed(1)} MB` : `${Math.round(mb)} MB`;
|
||||
}
|
||||
|
||||
/** Default cap for one stored blob, in bytes. */
|
||||
export function defaultBackupBlobLimit(): number {
|
||||
return Math.floor(v8.getHeapStatistics().heap_size_limit * BLOB_HEAP_SHARE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when one account's encrypted backup outgrows what this process can
|
||||
* hold as a single value.
|
||||
*
|
||||
* The message states what was counted — the ciphertext written so far — and
|
||||
* the limit it crossed, because that is the pair an operator can act on. It
|
||||
* is about the record, and it is true: raising the heap raises the limit with
|
||||
* it, since the limit is derived from the heap.
|
||||
*/
|
||||
export class BackupBlobTooLargeError extends Error {
|
||||
readonly bytes: number;
|
||||
readonly limitBytes: number;
|
||||
|
||||
constructor(bytes: number, limitBytes: number) {
|
||||
super(
|
||||
`Backup stopped after ${size(bytes)} of encrypted backup for one ` +
|
||||
`account, over the ${size(limitBytes)} a single stored copy may ` +
|
||||
`occupy here (a fifth of this process's ` +
|
||||
`${size(v8.getHeapStatistics().heap_size_limit)} heap limit). This ` +
|
||||
`account's record is genuinely too large to store as one row on this ` +
|
||||
`host; raise the container's memory or NODE_OPTIONS=` +
|
||||
`--max-old-space-size and the limit rises with it.`,
|
||||
);
|
||||
this.name = "BackupBlobTooLargeError";
|
||||
this.bytes = bytes;
|
||||
this.limitBytes = limitBytes;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Produces the backup JSON in pieces. Every piece is written in order.
|
||||
*
|
||||
@@ -69,6 +128,15 @@ export type BackupJsonProducer = (
|
||||
write: (chunk: string) => Promise<void>,
|
||||
) => Promise<unknown>;
|
||||
|
||||
export interface PackBackupBlobOptions {
|
||||
/**
|
||||
* Largest stored blob this call may produce, in bytes. Defaults to
|
||||
* `defaultBackupBlobLimit()`. Tests pass an explicit value; nothing else
|
||||
* should need to.
|
||||
*/
|
||||
maxBytes?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialised backup JSON, produced in pieces → the stored string.
|
||||
*
|
||||
@@ -80,20 +148,33 @@ export type BackupJsonProducer = (
|
||||
*
|
||||
* The one copy that cannot be avoided is the answer itself. `data_backups.data`
|
||||
* is a single `text` column, so the row has to be one value, and one value has
|
||||
* to exist as one string before the driver can bind it.
|
||||
* to exist as one string before the driver can bind it. That copy is therefore
|
||||
* also the only thing here that grows without bound as a record grows, which
|
||||
* is why the size check lives on it: `maxBytes` counts the ciphertext this
|
||||
* call has actually produced and stops on the account whose backup does not
|
||||
* fit, rather than reading a heap gauge every account shares.
|
||||
*/
|
||||
export async function packBackupBlobStreaming(
|
||||
producer: BackupJsonProducer,
|
||||
options: PackBackupBlobOptions = {},
|
||||
): Promise<string> {
|
||||
const limitBytes = options.maxBytes ?? defaultBackupBlobLimit();
|
||||
const encryptor = createStreamEncryptor();
|
||||
const pieces: string[] = [encryptor.header];
|
||||
let heldBytes = encryptor.header.length;
|
||||
const gzip = createGzip();
|
||||
|
||||
let failure: unknown = null;
|
||||
gzip.on("data", (chunk: Buffer) => {
|
||||
try {
|
||||
const piece = encryptor.update(chunk);
|
||||
if (piece !== "") pieces.push(piece);
|
||||
if (piece === "") return;
|
||||
// Base64 is ASCII, so a character counted here is a byte held here.
|
||||
heldBytes += piece.length;
|
||||
if (heldBytes > limitBytes) {
|
||||
throw new BackupBlobTooLargeError(heldBytes, limitBytes);
|
||||
}
|
||||
pieces.push(piece);
|
||||
} catch (err) {
|
||||
failure ??= err;
|
||||
gzip.destroy(err as Error);
|
||||
@@ -104,6 +185,12 @@ export async function packBackupBlobStreaming(
|
||||
gzip.on("end", resolve);
|
||||
gzip.on("error", reject);
|
||||
});
|
||||
// A failure raised inside the `data` handler destroys the stream, which
|
||||
// rejects this promise — and the producer's own `write` rethrows the same
|
||||
// failure first, so nothing ever awaits it. Marking it handled keeps a
|
||||
// failure that IS being reported from also surfacing as an unhandled
|
||||
// rejection; `await finished` below still sees the rejection.
|
||||
finished.catch(() => {});
|
||||
|
||||
const write = async (chunk: string): Promise<void> => {
|
||||
if (failure) throw failure;
|
||||
|
||||
@@ -25,8 +25,6 @@
|
||||
* The integration suite pins it against the materialising builder rather than
|
||||
* trusting that paragraph.
|
||||
*/
|
||||
import v8 from "node:v8";
|
||||
|
||||
import type { PrismaClient } from "@/generated/prisma/client";
|
||||
import {
|
||||
buildFullBackupPayload,
|
||||
@@ -48,42 +46,27 @@ export type BackupJsonSink = (chunk: string) => void | Promise<void>;
|
||||
const FLUSH_BYTES = 256 * 1024;
|
||||
|
||||
/**
|
||||
* Fraction of V8's heap limit the writer refuses to cross.
|
||||
* There is deliberately no heap reading in this writer.
|
||||
*
|
||||
* The point is not to make an oversized account succeed — it is to make it
|
||||
* FAIL AS A JOB. A backup that runs out of heap takes the process with it, and
|
||||
* a pg-boss worker sharing the app process means one account's record can
|
||||
* restart the instance for everybody. Because the work is now incremental
|
||||
* there is a checkpoint every few hundred kilobytes, so the guard actually
|
||||
* gets to run before V8 hits the wall; the old single `JSON.stringify` had no
|
||||
* point at which anything could intervene.
|
||||
* There used to be: every flush compared the process's live heap usage against
|
||||
* 80 % of V8's heap limit and aborted the backup above it. That reading is not
|
||||
* the backup's footprint — it is the whole process's, garbage included, and a
|
||||
* long-lived Next.js server sits at 400 MB of largely COLLECTABLE heap. On a
|
||||
* 1 GB container (a 524 MB V8 limit, so a 419 MB budget) the check therefore
|
||||
* fired on the first flush of every account: a weekly pass over four accounts
|
||||
* aborted all four in seven seconds, and the smallest of them was a demo
|
||||
* record whose entire stored copy is 1.2 MB. It turned "this process has been
|
||||
* up a while" into "your record is too large", and then told the operator to
|
||||
* buy memory.
|
||||
*
|
||||
* What this writer holds is bounded by its own shape and needs no gauge to
|
||||
* prove: the three tables that scale with a record are pulled a page at a
|
||||
* time, every other section is released as soon as its JSON is in the sink,
|
||||
* and at most `FLUSH_BYTES` of text is ever pending. The one copy that DOES
|
||||
* grow with the record is the sink's — the stored blob — so that is where the
|
||||
* bound belongs, and `packBackupBlobStreaming` enforces it there against bytes
|
||||
* it counted itself rather than against a heap the rest of the process shares.
|
||||
*/
|
||||
const HEAP_HIGH_WATER = 0.8;
|
||||
|
||||
/** Thrown when the writer stops itself short of the heap limit. */
|
||||
export class BackupHeapBudgetExceededError extends Error {
|
||||
constructor(usedBytes: number, ceilingBytes: number) {
|
||||
super(
|
||||
`Backup aborted at ${Math.round(usedBytes / 1024 / 1024)} MB of heap ` +
|
||||
`(budget ${Math.round(ceilingBytes / 1024 / 1024)} MB). The record is ` +
|
||||
`too large for this container's memory limit; raise the container's ` +
|
||||
`memory or NODE_OPTIONS=--max-old-space-size.`,
|
||||
);
|
||||
this.name = "BackupHeapBudgetExceededError";
|
||||
}
|
||||
}
|
||||
|
||||
export interface StreamFullBackupOptions extends FullBackupOptions {
|
||||
/**
|
||||
* Heap ceiling in bytes. Defaults to 80 % of V8's limit for this process.
|
||||
* Tests pass an explicit value; nothing else should need to.
|
||||
*/
|
||||
heapCeilingBytes?: number;
|
||||
}
|
||||
|
||||
function defaultHeapCeiling(): number {
|
||||
return Math.floor(v8.getHeapStatistics().heap_size_limit * HEAP_HIGH_WATER);
|
||||
}
|
||||
|
||||
/**
|
||||
* Write the full-backup JSON for `userId` into `sink`, and answer the counts.
|
||||
@@ -95,9 +78,8 @@ export async function streamFullBackupJson(
|
||||
prisma: PrismaClient,
|
||||
userId: string,
|
||||
sink: BackupJsonSink,
|
||||
options: StreamFullBackupOptions = {},
|
||||
options: FullBackupOptions = {},
|
||||
): Promise<FullBackupCounts> {
|
||||
const ceiling = options.heapCeilingBytes ?? defaultHeapCeiling();
|
||||
const { payload, counts } = await buildFullBackupPayload(prisma, userId, {
|
||||
...options,
|
||||
deferBulk: true,
|
||||
@@ -112,8 +94,6 @@ export async function streamFullBackupJson(
|
||||
pending = "";
|
||||
pendingBytes = 0;
|
||||
await sink(chunk);
|
||||
const used = process.memoryUsage().heapUsed;
|
||||
if (used > ceiling) throw new BackupHeapBudgetExceededError(used, ceiling);
|
||||
};
|
||||
|
||||
const write = async (piece: string): Promise<void> => {
|
||||
|
||||
@@ -103,7 +103,6 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"geo_backfill_skipped",
|
||||
"geo_backfill_still_unresolved",
|
||||
"geolite2_fetch_status",
|
||||
"heap_budget_trips",
|
||||
"host_metric_pruned",
|
||||
"hourly_rows_upserted",
|
||||
"imported",
|
||||
@@ -156,6 +155,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
|
||||
"provider",
|
||||
"queued",
|
||||
"rearmed",
|
||||
"records_oversized",
|
||||
"records_read",
|
||||
"refreshed",
|
||||
"refused",
|
||||
|
||||
@@ -21,18 +21,25 @@ vi.mock("@/lib/export/full-backup-payload", () => ({
|
||||
|
||||
// The envelope is exercised end-to-end in
|
||||
// `src/lib/export/__tests__/backup-blob.test.ts`; here it stays transparent so
|
||||
// the stored bytes can be read back as JSON.
|
||||
vi.mock("@/lib/export/backup-blob", () => ({
|
||||
packBackupBlobStreaming: async (
|
||||
produce: (write: (chunk: string) => Promise<void>) => Promise<void>,
|
||||
) => {
|
||||
let out = "";
|
||||
await produce(async (chunk) => {
|
||||
out += chunk;
|
||||
});
|
||||
return mocks.packBlob(out);
|
||||
},
|
||||
}));
|
||||
// the stored bytes can be read back as JSON. The size error is the real class
|
||||
// so the handler's `instanceof` arm is the one that runs.
|
||||
vi.mock("@/lib/export/backup-blob", async () => {
|
||||
const actual = await vi.importActual<
|
||||
typeof import("@/lib/export/backup-blob")
|
||||
>("@/lib/export/backup-blob");
|
||||
return {
|
||||
BackupBlobTooLargeError: actual.BackupBlobTooLargeError,
|
||||
packBackupBlobStreaming: async (
|
||||
produce: (write: (chunk: string) => Promise<void>) => Promise<void>,
|
||||
) => {
|
||||
let out = "";
|
||||
await produce(async (chunk) => {
|
||||
out += chunk;
|
||||
});
|
||||
return mocks.packBlob(out);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/logging/background", () => ({
|
||||
withBackgroundEvent: vi.fn(
|
||||
@@ -50,17 +57,19 @@ vi.mock("../shared", () => ({
|
||||
getWorkerPrisma: mocks.getWorkerPrisma,
|
||||
}));
|
||||
|
||||
import { BackupBlobTooLargeError } from "@/lib/export/backup-blob";
|
||||
|
||||
import { handleDataBackup } from "../backup-handlers";
|
||||
|
||||
const documentCiphertext = Buffer.from([1, 2, 3, 4]).toString("base64");
|
||||
|
||||
function buildPrismaMock() {
|
||||
function buildPrismaMock(
|
||||
users: Array<{ id: string; username: string }> = [
|
||||
{ id: "user-dr", username: "backup-owner" },
|
||||
],
|
||||
) {
|
||||
return {
|
||||
user: {
|
||||
findMany: vi
|
||||
.fn()
|
||||
.mockResolvedValue([{ id: "user-dr", username: "backup-owner" }]),
|
||||
},
|
||||
user: { findMany: vi.fn().mockResolvedValue(users) },
|
||||
dataBackup: { upsert: mocks.upsert },
|
||||
};
|
||||
}
|
||||
@@ -133,3 +142,79 @@ describe("handleDataBackup canonical DR payload", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* What the pass says about itself when it protected nobody.
|
||||
*
|
||||
* The weekly job used to report `ok: true` with `backed: 0` — a completed
|
||||
* pg-boss job, an untouched failing-queue panel, and a backups page listing
|
||||
* copies from six weeks earlier with perfectly ordinary timestamps. Every
|
||||
* surface an operator could look at agreed that a pass which wrote nothing
|
||||
* had gone fine.
|
||||
*/
|
||||
describe("handleDataBackup outcome", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mocks.upsert.mockResolvedValue({});
|
||||
mocks.buildFullBackupPayload.mockResolvedValue({
|
||||
payload: { schemaVersion: "1", userId: "user-dr" },
|
||||
counts: {},
|
||||
});
|
||||
});
|
||||
|
||||
it("fails the run when not one account got a copy", async () => {
|
||||
mocks.getWorkerPrisma.mockReturnValue(buildPrismaMock());
|
||||
mocks.upsert.mockRejectedValue(new Error("write failed"));
|
||||
|
||||
const outcome = await handleDataBackup([]);
|
||||
|
||||
expect(outcome.ok).toBe(false);
|
||||
expect(outcome).toMatchObject({
|
||||
ok: false,
|
||||
reason: "no account could be backed up",
|
||||
did: { backed: 0, total: 1, users_failed: 1, records_oversized: 0 },
|
||||
});
|
||||
});
|
||||
|
||||
it("counts an oversized record as the reason it could not", async () => {
|
||||
mocks.getWorkerPrisma.mockReturnValue(buildPrismaMock());
|
||||
mocks.upsert.mockRejectedValue(new BackupBlobTooLargeError(9_000, 4_096));
|
||||
|
||||
const outcome = await handleDataBackup([]);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
ok: false,
|
||||
did: { backed: 0, users_failed: 1, records_oversized: 1 },
|
||||
});
|
||||
});
|
||||
|
||||
it("still passes when one account failed and another was written", async () => {
|
||||
// The fan-out rule: a pass is judged on the pass. Failing the queue over
|
||||
// one account's record would re-run the whole cohort on every retry, and
|
||||
// that account's own copy ages on the backups page either way.
|
||||
mocks.getWorkerPrisma.mockReturnValue(
|
||||
buildPrismaMock([
|
||||
{ id: "user-a", username: "a" },
|
||||
{ id: "user-b", username: "b" },
|
||||
]),
|
||||
);
|
||||
mocks.upsert
|
||||
.mockRejectedValueOnce(new Error("write failed"))
|
||||
.mockResolvedValueOnce({});
|
||||
|
||||
const outcome = await handleDataBackup([]);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
ok: true,
|
||||
did: { backed: 1, total: 2, users_failed: 1 },
|
||||
});
|
||||
});
|
||||
|
||||
it("passes on an instance with no accounts at all", async () => {
|
||||
mocks.getWorkerPrisma.mockReturnValue(buildPrismaMock([]));
|
||||
|
||||
const outcome = await handleDataBackup([]);
|
||||
|
||||
expect(outcome).toMatchObject({ ok: true, did: { backed: 0, total: 0 } });
|
||||
});
|
||||
});
|
||||
@@ -6,11 +6,11 @@
|
||||
*/
|
||||
import { type Job } from "pg-boss";
|
||||
import { recordError } from "@/lib/jobs/worker-status";
|
||||
import { packBackupBlobStreaming } from "@/lib/export/backup-blob";
|
||||
import {
|
||||
BackupHeapBudgetExceededError,
|
||||
streamFullBackupJson,
|
||||
} from "@/lib/export/full-backup-stream";
|
||||
BackupBlobTooLargeError,
|
||||
packBackupBlobStreaming,
|
||||
} from "@/lib/export/backup-blob";
|
||||
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
|
||||
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import {
|
||||
@@ -82,8 +82,12 @@ export async function handleDataBackup(
|
||||
|
||||
let backed = 0;
|
||||
let usersFailed = 0;
|
||||
let heapBudgetTrips = 0;
|
||||
let oversized = 0;
|
||||
let largestBlobBytes = 0;
|
||||
// Kept for the failure report: without it the run that failed for
|
||||
// everybody names no reason at all, which is the shape this pass spent
|
||||
// six weeks in.
|
||||
let lastError: unknown;
|
||||
for (const user of users) {
|
||||
try {
|
||||
// Streamed, compressed, then encrypted (the record contains
|
||||
@@ -123,13 +127,14 @@ export async function handleDataBackup(
|
||||
// it rides out as a count so the weekly run is not retried for the
|
||||
// whole cohort.
|
||||
//
|
||||
// A record too large for this container's heap lands here too, and
|
||||
// that is the whole point of the budget the writer enforces: before
|
||||
// it existed the same condition was an uncatchable V8 abort that
|
||||
// restarted the instance, so one account's size was a denial of
|
||||
// service on every other account on the host.
|
||||
// A record whose stored copy does not fit this process lands here
|
||||
// too, and that is the point of the size limit the envelope writer
|
||||
// enforces: before it existed the same condition was an uncatchable
|
||||
// V8 abort that restarted the instance, so one account's size was a
|
||||
// denial of service on every other account on the host.
|
||||
usersFailed++;
|
||||
if (err instanceof BackupHeapBudgetExceededError) heapBudgetTrips++;
|
||||
lastError = err;
|
||||
if (err instanceof BackupBlobTooLargeError) oversized++;
|
||||
evt.addWarning(`Failed for user ${user.id}: ${err}`);
|
||||
}
|
||||
}
|
||||
@@ -137,19 +142,34 @@ export async function handleDataBackup(
|
||||
// The stored size is the one number that says whether this pass is
|
||||
// heading back towards the wall it hit before: it tracks the record.
|
||||
evt.addMeta("data_backup_largest_blob_bytes", largestBlobBytes);
|
||||
// How many accounts the writer stopped rather than let the process die.
|
||||
// Non-zero means this host needs more memory, not a retry.
|
||||
evt.addMeta("data_backup_heap_budget_trips", heapBudgetTrips);
|
||||
// How many accounts the envelope writer stopped rather than let the
|
||||
// process die. Non-zero means this host needs more memory, not a retry.
|
||||
evt.addMeta("data_backup_records_oversized", oversized);
|
||||
evt.setBackground({
|
||||
task_name: "job.data_backup",
|
||||
result: { backed, total: users.length },
|
||||
result: { backed, total: users.length, failed: usersFailed },
|
||||
});
|
||||
return jobDone({
|
||||
|
||||
const did = {
|
||||
backed,
|
||||
total: users.length,
|
||||
users_failed: usersFailed,
|
||||
heap_budget_trips: heapBudgetTrips,
|
||||
});
|
||||
records_oversized: oversized,
|
||||
};
|
||||
|
||||
// A pass that wrote nothing for anybody protected nobody, and saying
|
||||
// `ok: true` about it is how an instance goes a month and a half without
|
||||
// a usable copy while every surface reads healthy. Per-account failures
|
||||
// still ride out as counts when SOME account got a copy — that is the
|
||||
// fan-out rule, and that account's own row ages on the backups page —
|
||||
// but zero out of a non-empty cohort is the pass failing rather than a
|
||||
// leg of it, so pg-boss records a failed run and the backups page reads
|
||||
// it back through `readLastQueueRun`.
|
||||
if (users.length > 0 && backed === 0) {
|
||||
return jobFailed("no account could be backed up", lastError, did);
|
||||
}
|
||||
|
||||
return jobDone(did);
|
||||
} catch (err) {
|
||||
evt.setError(err);
|
||||
recordError();
|
||||
|
||||
@@ -33,20 +33,43 @@ import vm from "node:vm";
|
||||
import { beforeAll, afterAll, describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
BackupBlobTooLargeError,
|
||||
packBackupBlobStreaming,
|
||||
unpackBackupBlob,
|
||||
} from "@/lib/export/backup-blob";
|
||||
import {
|
||||
BackupHeapBudgetExceededError,
|
||||
streamFullBackupJson,
|
||||
} from "@/lib/export/full-backup-stream";
|
||||
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
|
||||
import { buildFullBackupPayload } from "@/lib/export/full-backup-payload";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const OWNER_ID = "backup-streaming-owner";
|
||||
/** A second account with almost nothing in it — the demo-record case. */
|
||||
const TINY_OWNER_ID = "backup-streaming-tiny";
|
||||
const MEASUREMENT_ROWS = 120_000;
|
||||
const MOOD_ROWS = 8_000;
|
||||
|
||||
/**
|
||||
* 80 % of the 524 MB V8 limit a 1 GB container gets.
|
||||
*
|
||||
* The number the weekly pass aborted every account against on the live
|
||||
* instance, including one whose entire stored backup is 1.2 MB.
|
||||
*/
|
||||
const CONTAINER_BUDGET_BYTES = Math.floor(524 * 1024 * 1024 * 0.8);
|
||||
|
||||
/**
|
||||
* How much garbage to leave lying about before a backup runs.
|
||||
*
|
||||
* A container's whole budget where the runner has room for it, and a majority
|
||||
* share of the heap where it does not — this file is also run under
|
||||
* `--max-old-space-size=450`, and a target above that heap's own limit would
|
||||
* measure an out-of-memory abort rather than a backup.
|
||||
*/
|
||||
function dirtyTargetBytes(): number {
|
||||
return Math.min(
|
||||
CONTAINER_BUDGET_BYTES + 8 * 1024 * 1024,
|
||||
Math.floor(v8.getHeapStatistics().heap_size_limit * 0.6),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* What the streaming writer may hold on top of the process's own baseline.
|
||||
*
|
||||
@@ -73,6 +96,20 @@ const forceGc = ((): (() => void) => {
|
||||
return gc;
|
||||
})();
|
||||
|
||||
/**
|
||||
* The one number this file exists to produce, on stderr where a gate run keeps
|
||||
* it. Peaks are the evidence, and evidence that only prints on failure is not
|
||||
* evidence.
|
||||
*/
|
||||
function reportPeak(label: string, heldBytes: number, liveBytes: number): void {
|
||||
const mb = (bytes: number): number => Math.round(bytes / 1024 / 1024);
|
||||
process.stderr.write(
|
||||
`[backup-memory] ${label}: peak ${mb(heldBytes)} MB held by the backup, ` +
|
||||
`${mb(liveBytes)} MB live in the process, ` +
|
||||
`${mb(v8.getHeapStatistics().heap_size_limit)} MB heap limit\n`,
|
||||
);
|
||||
}
|
||||
|
||||
/** Heap held after a forced collection. Garbage is not a measurement. */
|
||||
function liveHeapBytes(): number {
|
||||
// Twice: the first pass frees the objects, the second collects what the
|
||||
@@ -82,6 +119,42 @@ function liveHeapBytes(): number {
|
||||
return process.memoryUsage().heapUsed;
|
||||
}
|
||||
|
||||
/** Leave the heap carrying `targetBytes` of garbage nothing holds any more. */
|
||||
function dirtyHeap(targetBytes: number): number {
|
||||
let garbage: unknown[] = [];
|
||||
for (
|
||||
let round = 0;
|
||||
round < 2_000 && process.memoryUsage().heapUsed < targetBytes;
|
||||
round++
|
||||
) {
|
||||
const chunk = new Array(30_000);
|
||||
for (let at = 0; at < 30_000; at++) chunk[at] = { k: round * at };
|
||||
garbage.push(chunk);
|
||||
}
|
||||
const reached = process.memoryUsage().heapUsed;
|
||||
// Dropped. Every byte counted above is collectable from here on, which is
|
||||
// exactly what a long-lived Next.js server's heap looks like.
|
||||
garbage = [];
|
||||
void garbage;
|
||||
return reached;
|
||||
}
|
||||
|
||||
async function seedTinyRecord(): Promise<void> {
|
||||
await prisma.user.create({
|
||||
data: { id: TINY_OWNER_ID, username: "backup-streaming-tiny" },
|
||||
});
|
||||
await prisma.measurement.create({
|
||||
data: {
|
||||
userId: TINY_OWNER_ID,
|
||||
type: "WEIGHT",
|
||||
value: 80,
|
||||
unit: "kg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date("2026-01-01T00:00:00Z"),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function seedLargeRecord(): Promise<void> {
|
||||
await prisma.user.create({
|
||||
data: { id: OWNER_ID, username: "backup-streaming-owner" },
|
||||
@@ -144,6 +217,7 @@ describe("weekly backup under a memory budget", () => {
|
||||
).toBe("function");
|
||||
await truncateAllTables(prisma);
|
||||
await seedLargeRecord();
|
||||
await seedTinyRecord();
|
||||
}, 240_000);
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -151,6 +225,10 @@ describe("weekly backup under a memory budget", () => {
|
||||
});
|
||||
|
||||
it("writes a restorable blob while holding a bounded amount of the record", async () => {
|
||||
// Dirtied first, and only then measured: the baseline is taken after a
|
||||
// forced collection, so what is compared below is what the writer HOLDS,
|
||||
// not what the process happens to have lying about.
|
||||
dirtyHeap(dirtyTargetBytes());
|
||||
const baseline = liveHeapBytes();
|
||||
let peakHeld = 0;
|
||||
let chunks = 0;
|
||||
@@ -176,6 +254,7 @@ describe("weekly backup under a memory budget", () => {
|
||||
expect(counts.moodEntries).toBe(MOOD_ROWS);
|
||||
});
|
||||
peakHeld = Math.max(peakHeld, liveHeapBytes() - baseline);
|
||||
reportPeak("large record", peakHeld, liveHeapBytes());
|
||||
|
||||
expect(
|
||||
peakHeld,
|
||||
@@ -215,17 +294,62 @@ describe("weekly backup under a memory budget", () => {
|
||||
).toBeGreaterThan(STREAM_BUDGET_BYTES * 2);
|
||||
}, 300_000);
|
||||
|
||||
it("fails as a job rather than as a process when the record does not fit", async () => {
|
||||
// The budget the writer enforces on itself is the reason an oversized
|
||||
// account is now a failed backup for that account instead of a restart
|
||||
// for every account on the host.
|
||||
it("backs up a one-row record on a heap already full of garbage", async () => {
|
||||
// The second way this pass has failed. The writer used to compare the
|
||||
// process's whole heap usage against a fraction of the heap limit, so on a
|
||||
// server that had been up a week it aborted the FIRST chunk of every
|
||||
// account — 422, 447, 448 and 441 MB read against a 419 MB budget, for
|
||||
// records from 445 000 measurements down to 1.2 MB. None of what it read
|
||||
// was the backup's.
|
||||
const target = dirtyTargetBytes();
|
||||
const dirty = dirtyHeap(target);
|
||||
const baseline = process.memoryUsage().heapUsed;
|
||||
let peak = baseline;
|
||||
|
||||
let jsonBytes = 0;
|
||||
const blob = await packBackupBlobStreaming(async (write) => {
|
||||
const counts = await streamFullBackupJson(
|
||||
prisma,
|
||||
TINY_OWNER_ID,
|
||||
async (chunk) => {
|
||||
jsonBytes += chunk.length;
|
||||
peak = Math.max(peak, process.memoryUsage().heapUsed);
|
||||
await write(chunk);
|
||||
},
|
||||
{ purpose: "disaster-recovery" },
|
||||
);
|
||||
expect(counts.measurements).toBe(1);
|
||||
});
|
||||
|
||||
const restored = JSON.parse(unpackBackupBlob(blob)) as {
|
||||
measurements: unknown[];
|
||||
};
|
||||
expect(restored.measurements).toHaveLength(1);
|
||||
// The record really is tiny and the heap really was carrying the pile it
|
||||
// aimed at — without both halves this passes vacuously.
|
||||
expect(jsonBytes).toBeLessThan(64 * 1024);
|
||||
expect(dirty).toBeGreaterThanOrEqual(target);
|
||||
expect(dirty).toBeGreaterThan(64 * 1024 * 1024);
|
||||
// The backup itself moved the heap by a rounding error next to the pile of
|
||||
// garbage it ran on top of.
|
||||
expect(peak - baseline).toBeLessThan(32 * 1024 * 1024);
|
||||
reportPeak("tiny record", peak - baseline, liveHeapBytes());
|
||||
}, 180_000);
|
||||
|
||||
it("fails as a job rather than as a process when the stored copy does not fit", async () => {
|
||||
// The bound that remains is on the one copy the pipeline cannot stream
|
||||
// away: the blob itself, counted in bytes it produced. An account whose
|
||||
// backup outgrows what this process can hold as a single value is now a
|
||||
// failed backup for that account instead of a restart for every account
|
||||
// on the host.
|
||||
await expect(
|
||||
packBackupBlobStreaming((write) =>
|
||||
streamFullBackupJson(prisma, OWNER_ID, write, {
|
||||
purpose: "disaster-recovery",
|
||||
heapCeilingBytes: 1,
|
||||
}),
|
||||
packBackupBlobStreaming(
|
||||
(write) =>
|
||||
streamFullBackupJson(prisma, OWNER_ID, write, {
|
||||
purpose: "disaster-recovery",
|
||||
}),
|
||||
{ maxBytes: 64 * 1024 },
|
||||
),
|
||||
).rejects.toBeInstanceOf(BackupHeapBudgetExceededError);
|
||||
).rejects.toBeInstanceOf(BackupBlobTooLargeError);
|
||||
}, 120_000);
|
||||
});
|
||||
Reference in new issue
Block a user