mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-13 22:32:04 +00:00
Merge upstream v1.38.7
Only CHANGELOG.md conflicted: v1.38.7 released while this branch was open, so the new release heading landed where the branch's [Unreleased] section sits. The HRV entries stay under [Unreleased] above 1.38.7; everything from 1.38.7 down is upstream's file verbatim. docs/api/openapi.yaml auto-merged. Upstream's only change there was the `info.version` bump to 1.38.7, which the merge took while keeping this branch's DashboardLayoutReadResult, and package.json now reads 1.38.7 to match — so the merged file should equal a fresh generate. It is a generated file, though, so confirm with `pnpm openapi:generate` rather than trusting the three-way merge.
This commit is contained in:
13 files changed
+575
-107
No files matched your search
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user