fix(backup): store the weekly and manual backup piece by piece

The backup streamed its JSON through gzip and the cipher, but kept the
stored copy as one string and handed that to Postgres in one parameter.
On a seeded account of 1.25 million measurements the copy is 64 MB of
base64, and the pieces, the joined string and the driver's copies of it
grew the process from 309 MB to 946 MB resident. A bare process needed a
heap limit above 400 MB to finish it (the new path finishes under 220 MB).
With the dense retention running beside it in a 1 GB container the
process died of heap exhaustion, and the backup job sat "active" until its
two-hour limit marked it "job timed out" (#1031).

`packBackupBlobInto` hands the stored copy on in pieces of up to 4 MB as
it is produced, and `storeBackupBlob` writes them into a temporary table
on its own transaction and assembles the row with one `string_agg`. The
process never holds the whole copy. The account's previous copy stays in
place until the new one is complete: a failure rolls back to it.

Measured on the seeded account in a 1 GB container, web and worker in one
process: the backup completes in about 30 s with the dense retention
running beside it, where v1.39.0 crashed within 30 s.

The streamed JSON also reports its progress to the job every 50 000 rows.

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);
+  });
+});

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<string> {
+  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<void>;
+
+/**
+ * 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<void> {
   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<void> => {
+    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<void> => {
     if (failure) throw failure;
+    await flush(false);
     if (gzip.write(chunk, "utf8")) return;
     await new Promise<void>((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);
 }
 
 /**

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<void>;
  */
 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;

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<number> {
+  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 },
+  );
+}

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<void>) => Promise<void>,
-    ) => {
-      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<void>) => Promise<void>,
+  ) => {
+    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([]);
 

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:

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<void> {
+  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<void>) => {
+      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));
+  });
+});