diff --git a/src/lib/export/__tests__/backup-blob.test.ts b/src/lib/export/__tests__/backup-blob.test.ts index cf3225a34..4f632bebb 100644 --- a/src/lib/export/__tests__/backup-blob.test.ts +++ b/src/lib/export/__tests__/backup-blob.test.ts @@ -17,6 +17,7 @@ process.env.ENCRYPTION_KEY ??= "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; +import { randomBytes } from "node:crypto"; import v8 from "node:v8"; import { describe, expect, it } from "vitest"; @@ -26,6 +27,7 @@ import { BackupBlobTooLargeError, defaultBackupBlobLimit, packBackupBlob, + packBackupBlobInto, packBackupBlobStreaming, unpackBackupBlob, } from "../backup-blob"; @@ -213,3 +215,66 @@ describe("backup blob envelope", () => { expect(limit).toBeGreaterThan(16 * 1024 * 1024); }); }); + +/** + * Issue #1031: the weekly and manual backup kept the stored copy as one string + * before writing it, and at 1.25 million measurements that string and the + * driver's copies of it added about 640 MB to the process. `packBackupBlobInto` + * hands the stored copy on in bounded pieces instead, as it is produced. + */ +describe("piecewise envelope", () => { + /** JSON that gzip cannot shrink much, so the stored copy is several MB. */ + function incompressibleJson(bytes: number): string { + return JSON.stringify({ + noise: randomBytes(Math.ceil((bytes * 3) / 4)).toString("base64"), + }); + } + + it("delivers pieces that concatenate to a blob reading back the exact JSON", async () => { + const json = sampleJson(3_000); + const pieces: string[] = []; + await packBackupBlobInto( + (piece) => { + pieces.push(piece); + }, + async (write) => { + for (let at = 0; at < json.length; at += 4_096) { + await write(json.slice(at, at + 4_096)); + } + }, + ); + expect(unpackBackupBlob(pieces.join(""))).toBe(json); + }); + + it("hands pieces on while the producer is still writing, each one bounded", async () => { + const json = incompressibleJson(12 * 1024 * 1024); + const pieces: number[] = []; + let producing = false; + let piecesWhileProducing = 0; + const chunks: string[] = []; + + await packBackupBlobInto( + (piece) => { + pieces.push(piece.length); + chunks.push(piece); + if (producing) piecesWhileProducing += 1; + }, + async (write) => { + producing = true; + for (let at = 0; at < json.length; at += 64 * 1024) { + await write(json.slice(at, at + 64 * 1024)); + } + producing = false; + }, + { maxBytes: Number.MAX_SAFE_INTEGER }, + ); + + expect(pieces.length).toBeGreaterThan(2); + expect(piecesWhileProducing).toBeGreaterThan(1); + // The flush threshold is 4 MB; one gzip chunk's worth of slack on top. + for (const length of pieces) { + expect(length).toBeLessThan(4 * 1024 * 1024 + 256 * 1024); + } + expect(unpackBackupBlob(chunks.join(""))).toBe(json); + }); +}); diff --git a/src/lib/export/backup-blob.ts b/src/lib/export/backup-blob.ts index a05760650..f305a67b4 100644 --- a/src/lib/export/backup-blob.ts +++ b/src/lib/export/backup-blob.ts @@ -141,26 +141,59 @@ export interface PackBackupBlobOptions { * Serialised backup JSON, produced in pieces → the stored string. * * The pipeline is JSON piece → gzip → AES-256-GCM → base64, with nothing - * buffered end to end but the base64 answer. `producer` decides how big its - * pieces are; the gzip stream applies backpressure through the promise this - * hands back, so a fast producer cannot outrun the compressor and pile up - * chunks in the stream's internal queue. - * - * 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. 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. + * buffered end to end but the base64 answer. Kept for callers that need the + * answer as one value; the weekly and manual backup no longer do, they hand + * the pieces to Postgres as they come (`packBackupBlobInto` below, via + * `storeBackupBlob`). */ export async function packBackupBlobStreaming( producer: BackupJsonProducer, options: PackBackupBlobOptions = {}, ): Promise { + const pieces: string[] = []; + await packBackupBlobInto( + (piece) => { + pieces.push(piece); + }, + producer, + options, + ); + return pieces.join(""); +} + +/** Receives the stored string a piece at a time, in order. Awaited. */ +export type BackupBlobSink = (piece: string) => void | Promise; + +/** + * How much base64 to gather before handing it to the sink. Large enough that + * a sink writing to the database makes a few dozen round trips for a large + * record rather than thousands, small enough to be irrelevant to the heap. + */ +const SINK_FLUSH_BYTES = 4 * 1024 * 1024; + +/** + * Serialised backup JSON, produced in pieces → the stored string, delivered + * in pieces. + * + * The concatenation of every piece `sink` receives is exactly the string + * `packBackupBlobStreaming` returns. `producer` decides how big its JSON + * pieces are; the gzip stream applies backpressure through the promise this + * hands back, so a fast producer cannot outrun the compressor, and a slow + * sink holds the producer back the same way, so neither side piles up. + * + * `maxBytes` still counts the whole answer. Nothing here holds it any more, + * but every reader of a stored copy (restore, download, summary) opens it as + * one value, so the limit is what keeps a copy restorable on this host. + */ +export async function packBackupBlobInto( + sink: BackupBlobSink, + producer: BackupJsonProducer, + options: PackBackupBlobOptions = {}, +): Promise { const limitBytes = options.maxBytes ?? defaultBackupBlobLimit(); const encryptor = createStreamEncryptor(); - const pieces: string[] = [encryptor.header]; + let pending: string[] = [encryptor.header]; + let pendingBytes = encryptor.header.length; let heldBytes = encryptor.header.length; const gzip = createGzip(); @@ -169,12 +202,13 @@ export async function packBackupBlobStreaming( try { const piece = encryptor.update(chunk); if (piece === "") return; - // Base64 is ASCII, so a character counted here is a byte held here. + // Base64 is ASCII, so a character counted here is a byte written here. heldBytes += piece.length; if (heldBytes > limitBytes) { throw new BackupBlobTooLargeError(heldBytes, limitBytes); } - pieces.push(piece); + pending.push(piece); + pendingBytes += piece.length; } catch (err) { failure ??= err; gzip.destroy(err as Error); @@ -192,8 +226,19 @@ export async function packBackupBlobStreaming( // rejection; `await finished` below still sees the rejection. finished.catch(() => {}); + const flush = async (force: boolean): Promise => { + if (pendingBytes === 0 || (!force && pendingBytes < SINK_FLUSH_BYTES)) { + return; + } + const piece = pending.join(""); + pending = []; + pendingBytes = 0; + await sink(piece); + }; + const write = async (chunk: string): Promise => { if (failure) throw failure; + await flush(false); if (gzip.write(chunk, "utf8")) return; await new Promise((resolve, reject) => { const onDrain = () => { @@ -219,8 +264,10 @@ export async function packBackupBlobStreaming( await finished; if (failure) throw failure; - pieces.push(encryptor.final()); - return pieces.join(""); + const tail = encryptor.final(); + pending.push(tail); + pendingBytes += tail.length; + await flush(true); } /** diff --git a/src/lib/export/full-backup-stream.ts b/src/lib/export/full-backup-stream.ts index c5b364aa0..1f5b1cdcf 100644 --- a/src/lib/export/full-backup-stream.ts +++ b/src/lib/export/full-backup-stream.ts @@ -26,6 +26,7 @@ * trusting that paragraph. */ import type { PrismaClient } from "@/generated/prisma/client"; +import { reportJobProgress } from "@/lib/jobs/job-observer"; import { buildFullBackupPayload, isDeferredRows, @@ -45,6 +46,9 @@ export type BackupJsonSink = (chunk: string) => void | Promise; */ const FLUSH_BYTES = 256 * 1024; +/** How often a bulk section reports how far it has got to a running job. */ +const PROGRESS_EVERY_ROWS = 50_000; + /** * There is deliberately no heap reading in this writer. * @@ -125,6 +129,9 @@ export async function streamFullBackupJson( rows === 0 ? JSON.stringify(row) : `,${JSON.stringify(row)}`, ); rows++; + if (rows % PROGRESS_EVERY_ROWS === 0) { + reportJobProgress({ backup_section: key, backup_section_rows: rows }); + } } await write("]"); bulkCounts[key] = rows; diff --git a/src/lib/export/store-backup-blob.ts b/src/lib/export/store-backup-blob.ts new file mode 100644 index 000000000..cf7989269 --- /dev/null +++ b/src/lib/export/store-backup-blob.ts @@ -0,0 +1,113 @@ +/** + * Write one account's stored backup (`data_backups.data`) without ever + * holding the stored copy in this process. + * + * Why. `packBackupBlobStreaming` streams the JSON through gzip and the cipher, + * but its answer is one string, because the column takes one value. Measured + * on a seeded account of 1.25 million measurements: the answer is 64 MB of + * base64, and between the pieces, the joined string and the copies the + * database driver makes to bind it, the backup grew the process by about + * 640 MB of resident memory (309 MB → 946 MB). The default compose stack runs + * the web server and the worker in one container capped at 1 GB, so on a + * record of that size the weekly or manual backup is the thing that takes the + * container down; a killed worker leaves its job `active` until pg-boss + * expires it two hours later as `job timed out`. + * + * What instead. The pieces go to Postgres as they come, a few megabytes at a + * time, into a temporary table on the transaction's own connection, and one + * statement assembles them into the row with `string_agg`. The process never + * holds more than one piece; Postgres builds the value it was going to store + * anyway. + * + * All of it is one transaction, so the account's previous copy stays in place + * until the new one is complete: a failure halfway through rolls back to it, + * and the temporary table goes with the transaction either way. + */ +import type { PrismaClient } from "@/generated/prisma/client"; + +import { + packBackupBlobInto, + type BackupJsonProducer, + type PackBackupBlobOptions, +} from "@/lib/export/backup-blob"; + +/** + * How long the storing transaction may stay open. The job around it expires + * after two hours (`DATA_BACKUP_SEND_OPTIONS`); the transaction has to give + * up before that so the failure is the transaction's own, with its own + * message, rather than an expired job. + */ +const STORE_TRANSACTION_TIMEOUT_MS = 90 * 60 * 1000; + +/** + * How long the transaction may sit idle between two pieces. The session + * default (`idle_in_transaction_session_timeout`, 60 s, see `src/lib/db.ts`) + * is right for a request and wrong here: the producer reads the whole record + * on other connections while this one waits for the next piece, and the + * sections before the first measurement can take longer than a minute on a + * large account. Scoped with SET LOCAL, so it ends with the transaction. + */ +const STORE_IDLE_TIMEOUT = "10min"; + +export interface StoreBackupBlobInput { + userId: string; + /** `DataBackup.type`: `WEEKLY_AUTO` for the scheduled and manual pass. */ + type: string; +} + +/** + * Pack `producer`'s JSON into the stored envelope and upsert it as the + * account's `(userId, type)` backup. Resolves to the stored size in bytes. + */ +export async function storeBackupBlob( + prisma: PrismaClient, + input: StoreBackupBlobInput, + producer: BackupJsonProducer, + options: PackBackupBlobOptions = {}, +): Promise { + return prisma.$transaction( + async (tx) => { + await tx.$executeRawUnsafe( + `SET LOCAL idle_in_transaction_session_timeout = '${STORE_IDLE_TIMEOUT}'`, + ); + await tx.$executeRaw` + CREATE TEMP TABLE backup_blob_parts ( + seq integer PRIMARY KEY, + piece text NOT NULL + ) ON COMMIT DROP + `; + + let seq = 0; + let bytes = 0; + await packBackupBlobInto( + async (piece) => { + await tx.$executeRaw` + INSERT INTO backup_blob_parts (seq, piece) VALUES (${seq}, ${piece}) + `; + seq += 1; + bytes += piece.length; + }, + producer, + options, + ); + + // The row first, through Prisma, so a new account's backup gets its id + // the same way every other row does; the data it briefly carries is + // never visible outside this transaction. + await tx.dataBackup.upsert({ + where: { userId_type: { userId: input.userId, type: input.type } }, + update: { createdAt: new Date() }, + create: { userId: input.userId, type: input.type, data: "" }, + }); + await tx.$executeRaw` + UPDATE data_backups + SET data = ( + SELECT string_agg(piece, '' ORDER BY seq) FROM backup_blob_parts + ) + WHERE user_id = ${input.userId} AND type = ${input.type} + `; + return bytes; + }, + { timeout: STORE_TRANSACTION_TIMEOUT_MS, maxWait: 60_000 }, + ); +} diff --git a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts index 7692ca02d..fe2a94b55 100644 --- a/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts +++ b/src/lib/jobs/reminder/__tests__/backup-handlers.test.ts @@ -4,9 +4,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ buildFullBackupPayload: vi.fn(), - packBlob: vi.fn((value: string) => value), getWorkerPrisma: vi.fn(), - upsert: vi.fn(), + store: vi.fn(), })); // Only the payload builder is stubbed. The REAL streaming writer runs on top @@ -19,27 +18,25 @@ vi.mock("@/lib/export/full-backup-payload", () => ({ isDeferredRows: () => false, })); -// 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. 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) => Promise, - ) => { - let out = ""; - await produce(async (chunk) => { - out += chunk; - }); - return mocks.packBlob(out); - }, - }; -}); +// The envelope and the piecewise store are exercised end-to-end in +// `src/lib/export/__tests__/backup-blob.test.ts` and the integration suite; +// here the store stays transparent so the stored bytes can be read back as +// JSON. `mocks.store` receives what would have been stored and decides +// whether the write succeeds. +vi.mock("@/lib/export/store-backup-blob", () => ({ + storeBackupBlob: async ( + _prisma: unknown, + input: { userId: string; type: string }, + produce: (write: (chunk: string) => Promise) => Promise, + ) => { + let out = ""; + await produce(async (chunk) => { + out += chunk; + }); + await mocks.store(input, out); + return out.length; + }, +})); vi.mock("@/lib/logging/background", () => ({ withBackgroundEvent: vi.fn( @@ -70,7 +67,6 @@ function buildPrismaMock( ) { return { user: { findMany: vi.fn().mockResolvedValue(users) }, - dataBackup: { upsert: mocks.upsert }, }; } @@ -78,7 +74,7 @@ describe("handleDataBackup canonical DR payload", () => { beforeEach(() => { vi.clearAllMocks(); mocks.getWorkerPrisma.mockReturnValue(buildPrismaMock()); - mocks.upsert.mockResolvedValue({}); + mocks.store.mockResolvedValue(undefined); mocks.buildFullBackupPayload.mockResolvedValue({ payload: { schemaVersion: "1", @@ -117,8 +113,12 @@ describe("handleDataBackup canonical DR payload", () => { deferBulk: true, }), ); - expect(mocks.upsert).toHaveBeenCalledOnce(); - const encrypted = mocks.upsert.mock.calls[0]![0].create.data as string; + expect(mocks.store).toHaveBeenCalledOnce(); + expect(mocks.store.mock.calls[0]![0]).toEqual({ + userId: "user-dr", + type: "WEEKLY_AUTO", + }); + const encrypted = mocks.store.mock.calls[0]![1] as string; const payload = JSON.parse(encrypted) as { moodEntries: Array<{ id: string; @@ -155,7 +155,7 @@ describe("handleDataBackup canonical DR payload", () => { describe("handleDataBackup outcome", () => { beforeEach(() => { vi.clearAllMocks(); - mocks.upsert.mockResolvedValue({}); + mocks.store.mockResolvedValue(undefined); mocks.buildFullBackupPayload.mockResolvedValue({ payload: { schemaVersion: "1", userId: "user-dr" }, counts: {}, @@ -164,7 +164,7 @@ describe("handleDataBackup outcome", () => { it("fails the run when not one account got a copy", async () => { mocks.getWorkerPrisma.mockReturnValue(buildPrismaMock()); - mocks.upsert.mockRejectedValue(new Error("write failed")); + mocks.store.mockRejectedValue(new Error("write failed")); const outcome = await handleDataBackup([]); @@ -178,7 +178,7 @@ describe("handleDataBackup outcome", () => { 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)); + mocks.store.mockRejectedValue(new BackupBlobTooLargeError(9_000, 4_096)); const outcome = await handleDataBackup([]); @@ -198,9 +198,9 @@ describe("handleDataBackup outcome", () => { { id: "user-b", username: "b" }, ]), ); - mocks.upsert + mocks.store .mockRejectedValueOnce(new Error("write failed")) - .mockResolvedValueOnce({}); + .mockResolvedValueOnce(undefined); const outcome = await handleDataBackup([]); diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 48c6ae397..13c258894 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -6,10 +6,8 @@ */ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; -import { - BackupBlobTooLargeError, - packBackupBlobStreaming, -} from "@/lib/export/backup-blob"; +import { BackupBlobTooLargeError } from "@/lib/export/backup-blob"; +import { storeBackupBlob } from "@/lib/export/store-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"; @@ -18,6 +16,7 @@ import { runOffhostBackup, } from "@/lib/jobs/offhost-backup"; import { getWorkerPrisma } from "./shared"; +import { reportJobProgress } from "@/lib/jobs/job-observer"; export interface DataBackupPayload { triggeredAt: string; @@ -118,38 +117,30 @@ export async function handleDataBackup( // six weeks in. let lastError: unknown; for (const user of users) { + reportJobProgress({ + backup_user: user.id, + backup_users_done: backed + usersFailed, + backup_users_total: users.length, + }); try { // Streamed, compressed, then encrypted (the record contains - // sensitive health information). Nothing between the database rows - // and this string exists as a whole: the payload goes into gzip a - // page at a time and the cipher consumes gzip's output as it comes, - // because a materialised copy of a large record is what used to take - // the process down. Both legs live in `packBackupBlobStreaming`, - // which is also what the restore path reads back through. - const encryptedBackup = await packBackupBlobStreaming((write) => - streamFullBackupJson(prisma, user.id, write, { - purpose: "disaster-recovery", - }), + // sensitive health information), and stored a piece at a time. + // Nothing between the database rows and the stored row exists as a + // whole in this process: the payload goes into gzip a page at a + // time, the cipher consumes gzip's output as it comes, and the + // ciphertext goes back to Postgres in pieces that one statement + // assembles into the row (`storeBackupBlob`). The stored copy used + // to be built here as one string first, which on a large record + // added several hundred megabytes to a container capped at 1 GB. + const storedBytes = await storeBackupBlob( + prisma, + { userId: user.id, type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, user.id, write, { + purpose: "disaster-recovery", + }), ); - largestBlobBytes = Math.max( - largestBlobBytes, - Buffer.byteLength(encryptedBackup, "utf8"), - ); - - await prisma.dataBackup.upsert({ - where: { - userId_type: { userId: user.id, type: "WEEKLY_AUTO" }, - }, - update: { - data: encryptedBackup, - createdAt: new Date(), - }, - create: { - userId: user.id, - type: "WEEKLY_AUTO", - data: encryptedBackup, - }, - }); + largestBlobBytes = Math.max(largestBlobBytes, storedBytes); backed++; } catch (err) { // One user's payload failing is that user's problem, not the pass's: diff --git a/tests/integration/large-account-jobs.test.ts b/tests/integration/large-account-jobs.test.ts new file mode 100644 index 000000000..9568c0235 --- /dev/null +++ b/tests/integration/large-account-jobs.test.ts @@ -0,0 +1,266 @@ +/** + * Issue #1031 — maintenance jobs on a large account. + * + * A self-hoster with 1.25 million measurement rows (a multi-year Apple Health + * export) saw the dense intra-day retention, the nightly drain tick and the + * data backup all fail with `job timed out`. Reproduced against a seeded + * account of the same size (`scripts/seed-large-account.ts`): the drains held + * a type's whole history in memory before folding a day, spent about 90 + * statements per folded day, and paged with a cursor Postgres could not use as + * an index condition; the backup added about 640 MB to the process to store + * one 64 MB row. pg-boss declared the jobs dead at their expiry without + * stopping them. + * + * A million rows do not fit a test run, so this file pins the properties the + * fix rests on at a scale that does: a pass that runs out of budget stops + * cleanly and the next run finishes the job with the same result as one + * uninterrupted run, and the backup is stored piecewise with the previous + * copy intact if a run fails halfway. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; +import { consolidateDailyMean } from "@/lib/measurements/consolidate-daily-mean"; +import { unpackBackupBlob } from "@/lib/export/backup-blob"; +import { storeBackupBlob } from "@/lib/export/store-backup-blob"; + +const TZ = "Europe/Berlin"; +const RUN_A = "user-1031-stopped"; +const RUN_B = "user-1031-straight"; +const DAYS = 12; +/** Oldest seeded instant: well outside the 90-day raw window. */ +const START = Date.UTC(2026, 0, 1, 0, 7); + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + for (const id of [RUN_A, RUN_B]) { + await prisma.user.create({ + data: { id, username: id, email: `${id}@example.test`, timezone: TZ }, + }); + } +}); + +/** The same history for both users: heart rate every 20 minutes, HRV hourly. */ +async function seedHistory(userId: string): Promise { + const rows = []; + for (let minute = 0; minute < DAYS * 24 * 60; minute += 20) { + const at = new Date(START + minute * 60_000); + rows.push({ + userId, + type: "PULSE" as const, + value: 55 + ((minute / 20) % 40), + unit: "bpm", + source: "APPLE_HEALTH" as const, + measuredAt: at, + externalId: `hk-hr-${minute}`, + }); + if (minute % 60 === 0) { + rows.push({ + userId, + type: "HEART_RATE_VARIABILITY" as const, + value: 30 + ((minute / 60) % 25), + unit: "ms", + source: "APPLE_HEALTH" as const, + measuredAt: at, + externalId: `hk-hrv-${minute}`, + }); + } + if (minute % 40 === 0) { + rows.push({ + userId, + type: "WALKING_SPEED" as const, + value: 4 + ((minute / 40) % 10) / 10, + unit: "km/h", + source: "APPLE_HEALTH" as const, + measuredAt: at, + externalId: `hk-ws-${minute}`, + }); + } + } + await getPrismaClient().measurement.createMany({ data: rows }); +} + +/** What a reader sees: every live row, identity-free. */ +async function liveRows(userId: string) { + const rows = await getPrismaClient().measurement.findMany({ + where: { userId, deletedAt: null }, + select: { + type: true, + value: true, + unit: true, + source: true, + measuredAt: true, + externalId: true, + }, + orderBy: [{ type: "asc" }, { measuredAt: "asc" }], + }); + return rows.map((row) => ({ + ...row, + measuredAt: row.measuredAt.toISOString(), + })); +} + +/** A `shouldStop` that lets `days` day buckets through, then says stop. */ +function afterDays(days: number): () => boolean { + let left = days; + return () => left-- <= 0; +} + +describe("a pass that runs out of budget (#1031)", () => { + it("dense retention: stops cleanly, and the next run ends where one straight run ends", async () => { + await seedHistory(RUN_A); + await seedHistory(RUN_B); + const prisma = getPrismaClient(); + + const first = await runDenseIntradayRetention(prisma, { + userId: RUN_A, + shouldStop: afterDays(5), + log: () => {}, + }); + expect(first.stoppedEarly).toBe(true); + expect(first.totals.daysConsolidated).toBe(5); + + // What the stopped run folded is committed: the next scan no longer sees + // it, so the second run folds only what is left. + const second = await runDenseIntradayRetention(prisma, { + userId: RUN_A, + log: () => {}, + }); + expect(second.stoppedEarly).toBe(false); + + const straight = await runDenseIntradayRetention(prisma, { + userId: RUN_B, + log: () => {}, + }); + expect(first.totals.daysConsolidated + second.totals.daysConsolidated).toBe( + straight.totals.daysConsolidated, + ); + expect( + first.totals.perSampleRowsSoftDeleted + + second.totals.perSampleRowsSoftDeleted, + ).toBe(straight.totals.perSampleRowsSoftDeleted); + + // A third run finds nothing: the pass converged. + const third = await runDenseIntradayRetention(prisma, { + userId: RUN_A, + log: () => {}, + }); + expect(third.totals.daysConsolidated).toBe(0); + + const stoppedResult = (await liveRows(RUN_A)).filter( + (row) => row.type !== "WALKING_SPEED", + ); + const straightResult = (await liveRows(RUN_B)).filter( + (row) => row.type !== "WALKING_SPEED", + ); + expect(stoppedResult.length).toBeGreaterThan(0); + expect(stoppedResult).toEqual(straightResult); + }); + + it("daily-mean consolidation: the same, for the nightly tick's middle pass", async () => { + await seedHistory(RUN_A); + await seedHistory(RUN_B); + const prisma = getPrismaClient(); + + const first = await consolidateDailyMean(prisma, { + userId: RUN_A, + shouldStop: afterDays(4), + log: () => {}, + }); + expect(first.stoppedEarly).toBe(true); + expect(first.totals.daysConsolidated).toBe(4); + await consolidateDailyMean(prisma, { userId: RUN_A, log: () => {} }); + await consolidateDailyMean(prisma, { userId: RUN_B, log: () => {} }); + + const stopped = (await liveRows(RUN_A)).filter( + (row) => row.type === "WALKING_SPEED", + ); + const straight = (await liveRows(RUN_B)).filter( + (row) => row.type === "WALKING_SPEED", + ); + // One daily mean per local day: twelve UTC days touch thirteen Berlin days. + expect(stopped).toHaveLength(DAYS + 1); + expect(stopped).toEqual(straight); + }); +}); + +describe("storeBackupBlob (#1031)", () => { + /** A producer writing `rows` JSON rows in small pieces. */ + function producer(rows: number, failAfter?: number) { + return async (write: (chunk: string) => Promise) => { + await write('{"measurements":['); + for (let i = 0; i < rows; i++) { + if (failAfter !== undefined && i === failAfter) { + throw new Error("record read failed halfway"); + } + await write( + `${i === 0 ? "" : ","}${JSON.stringify({ id: `m-${i}`, value: i })}`, + ); + } + await write("]}"); + }; + } + + function expectedJson(rows: number): string { + return JSON.stringify({ + measurements: Array.from({ length: rows }, (_, i) => ({ + id: `m-${i}`, + value: i, + })), + }); + } + + it("stores a copy that reads back as exactly the JSON produced", async () => { + const prisma = getPrismaClient(); + const bytes = await storeBackupBlob( + prisma, + { userId: RUN_A, type: "WEEKLY_AUTO" }, + producer(20_000), + ); + + const row = await prisma.dataBackup.findUniqueOrThrow({ + where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } }, + }); + expect(row.data.length).toBe(bytes); + expect(unpackBackupBlob(row.data)).toBe(expectedJson(20_000)); + }); + + it("keeps the previous copy when a run fails halfway, and the next run still stores", async () => { + const prisma = getPrismaClient(); + await storeBackupBlob( + prisma, + { userId: RUN_A, type: "WEEKLY_AUTO" }, + producer(10), + ); + const before = await prisma.dataBackup.findUniqueOrThrow({ + where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } }, + }); + + await expect( + storeBackupBlob( + prisma, + { userId: RUN_A, type: "WEEKLY_AUTO" }, + producer(20_000, 15_000), + ), + ).rejects.toThrow("record read failed halfway"); + + const after = await prisma.dataBackup.findUniqueOrThrow({ + where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } }, + }); + expect(after.data).toBe(before.data); + expect(after.createdAt.getTime()).toBe(before.createdAt.getTime()); + + // The temporary table went with the failed transaction. + await storeBackupBlob( + prisma, + { userId: RUN_A, type: "WEEKLY_AUTO" }, + producer(30), + ); + const replaced = await prisma.dataBackup.findUniqueOrThrow({ + where: { userId_type: { userId: RUN_A, type: "WEEKLY_AUTO" } }, + }); + expect(unpackBackupBlob(replaced.data)).toBe(expectedJson(30)); + }); +});