From fc13b9de5ce8697f759090cf7bbbe7f142187021 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 13:21:18 +0200 Subject: [PATCH] fix(restore): skip expired tombstones and give the chart rebuild its own limit A restore now leaves out measurement tombstones deleted longer ago than the tombstone retention. The nightly purge removes exactly those rows, the delta feed answers cursorExpired for every cursor that could still be served one, and the consolidation repairs look only inside the same horizon. Younger tombstones keep their readers (a lagging paired client, the undo of a recent delete, the repairs, the duplicate guard on re-uploaded samples) and are written as before. The backup file is unchanged. On a seeded account of 1.89 million readings whose 1.68 million tombstones were past retention, the restore on a slowed database took 5.4 min instead of 15. The rollup fold after the commit rebuilt five years of buckets under the 60 s request statement limit and could be dropped on a slow disk, leaving the charts empty until the next boot. `recomputeUserRollups` takes a `statementTimeoutMs` that scopes each aggregate and persist transaction with SET LOCAL, and the restore passes its after-commit allowance. Fixture tombstones in two round-trip tests were dated in July 2026 and had aged past the retention; they are now relative to the run. --- src/lib/export/restore-backup.ts | 66 ++++++++++++++-- src/lib/rollups/measurement-rollups.ts | 75 ++++++++++++++----- .../admin-backups-canonical-roundtrip.test.ts | 12 ++- .../backup-restore-statement-timeout.test.ts | 66 ++++++++++++++-- .../large-backup-round-trip.test.ts | 5 +- 5 files changed, 189 insertions(+), 35 deletions(-) diff --git a/src/lib/export/restore-backup.ts b/src/lib/export/restore-backup.ts index e2b317ce1..2cf75bdf4 100644 --- a/src/lib/export/restore-backup.ts +++ b/src/lib/export/restore-backup.ts @@ -102,6 +102,7 @@ import { restoreEnvironmentData } from "@/lib/export/environment-backup"; import { restoreEcgData } from "@/lib/export/ecg-backup"; import { restoredMedicationCreatedAt } from "@/lib/export/medication-created-at"; import { invalidateUserData } from "@/lib/cache/invalidate"; +import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; import { classifyRestoreFailure, RESTORE_FAILURE_CAUSE_MESSAGES, @@ -657,6 +658,7 @@ export async function restoreBackup( } let outcome: { + expiredTombstonesSkipped: number; cleared: RestoreResponse["cleared"]; skipped: RestoreSkipSummary; }; @@ -917,10 +919,42 @@ export async function restoreBackup( } const writtenReferencedIds = new Set(); const measurementBatchSize = 1_000; + // Deleted readings the file carries are restored as deleted rows + // (tombstones), with one exception: a tombstone deleted longer ago + // than `TOMBSTONE_RETENTION_DAYS` is not written at all. + // + // Such a row is exactly what the nightly tombstone purge + // (`measurement-tombstone-cleanup.ts`) hard-deletes, so writing it + // costs the restore a row in thirteen indexes for a row that is gone + // by the next morning. Nothing reads it in between: the delta feed + // answers `cursorExpired` for every cursor older than the same + // horizon, so no client that would still be served this tombstone + // exists; the step-consolidation repair and the hourly rebuild only + // look inside the horizon; and the undo of a delete is offered for + // what the purge has not removed yet, which this row would not + // survive either. + // + // Younger tombstones are written. They still have readers: a paired + // client that has not yet drained that deletion from the delta feed, + // the undo of a recent delete, the consolidation repairs that rebuild + // from folded samples, and the unique `(user, type, source, + // externalId)` key that stops a device from uploading a folded + // sample again. The backup file itself is unchanged and keeps every + // row. + const tombstoneHorizonMs = + Date.now() - TOMBSTONE_RETENTION_DAYS * 86_400_000; + const isExpiredTombstone = (measurement: BackupMeasurement) => + measurement.deletedAt != null && + new Date(measurement.deletedAt).getTime() < tombstoneHorizonMs; + let expiredTombstonesSkipped = 0; report("measurements"); await streamed.forEachMeasurementBatch( measurementBatchSize, - async (batch) => { + async (fileBatch) => { + const batch = fileBatch.filter( + (measurement) => !isExpiredTombstone(measurement), + ); + expiredTombstonesSkipped += fileBatch.length - batch.length; const stableRows = batch.flatMap( (measurement): MeasurementInsertRow[] => measurement.id @@ -936,7 +970,8 @@ export async function restoreBackup( // One statement per batch, not `createMany`: see // `insertMeasurementRows` for what the latter cost here. await insertMeasurementRows(tx, stableRows); - progress.measurementsWritten += batch.length; + // Counted through the file, so the bar ends at the file's total. + progress.measurementsWritten += fileBatch.length; report("measurements"); for (const row of stableRows) { if (referencedMeasurementIds.has(row.id)) { @@ -2379,7 +2414,11 @@ export async function restoreBackup( ); progress.sectionsDone = RESTORE_SECTION_STEPS.length; report("sections", true); - return { cleared, skipped: summarizeRestoreSkips(skips) }; + return { + cleared, + skipped: summarizeRestoreSkips(skips), + expiredTombstonesSkipped, + }; }, { maxWait: 10_000, @@ -2440,7 +2479,12 @@ export async function restoreBackup( ); } - const { cleared, skipped } = outcome; + const { cleared, skipped, expiredTombstonesSkipped } = outcome; + const measurementsRestored = + streamed.measurementCount - expiredTombstonesSkipped; + annotate({ + meta: { restore_expired_tombstones_skipped: expiredTombstonesSkipped }, + }); report("rebuilding", true); await input.onCommitted?.(); @@ -2464,9 +2508,17 @@ export async function restoreBackup( // outside the transaction (the 5-year fold would otherwise hold a // long write lock) and is best-effort so a populator hiccup never // undoes the restore. The boot-time backfill is the safety net. - if (streamed.measurementCount > 0) { + // + // The fold covers five years of the account in a handful of statements, + // right after the restore wrote every reading. On a slow disk one of them + // ran past the per-request statement limit, the fold was dropped, and the + // account's charts stayed empty until the next boot (#1031). It gets the + // time the deadline check above keeps free for it instead. + if (measurementsRestored > 0) { try { - await recomputeUserRollups(ownerId); + await recomputeUserRollups(ownerId, { + statementTimeoutMs: AFTER_TRANSACTION_ALLOWANCE_MS, + }); } catch (err) { annotate({ meta: { @@ -2526,7 +2578,7 @@ export async function restoreBackup( const summary = { ...summarizeBackup(payload), - measurements: streamed.measurementCount, + measurements: measurementsRestored, }; await auditLog("admin.backups.restore", { diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 6afdac3d4..867863642 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -187,6 +187,17 @@ export async function recomputeUserRollups( granularities?: RollupGranularity[]; from?: Date; to?: Date; + /** + * Run every statement of the fold under this limit instead of the + * per-connection request limit (`src/lib/db.ts`, 60 s by default). + * For callers that fold a whole account outside a request: the restore + * rebuilds five years of buckets right after writing up to millions of + * readings, and on a slow disk one granularity's aggregate can run past + * the request limit, which left the account's charts empty until the + * next boot backfill (#1031). Scoped with SET LOCAL to a transaction per + * statement group, so it ends with it. + */ + statementTimeoutMs?: number; } = {}, ): Promise<{ rowsUpserted: number; durationMs: number }> { const startedAt = Date.now(); @@ -218,22 +229,34 @@ export async function recomputeUserRollups( to > alignedFrom ? bucketSpan(new Date(to.getTime() - 1), granularity).to : bucketSpan(from, granularity).to; - const rows = await runRollupAggregate({ + const aggregateInput = { userId, types: opts.types, granularity, from: alignedFrom, to: alignedTo, - }); + }; + const budget = opts.statementTimeoutMs; + const rows = budget + ? await prisma.$transaction( + async (tx) => { + await applyStatementBudget(tx, budget); + return runRollupAggregate(aggregateInput, tx); + }, + { maxWait: 10_000, timeout: budget + 30_000 }, + ) + : await runRollupAggregate(aggregateInput); // No `continue` on an empty aggregate: an empty result is a STATEMENT — // this window holds no readings any more — and the prune below is what // acts on it. Skipping the write here is what let a bucket outlive its // source rows. - rowsUpserted += await persistRollupRows(userId, granularity, rows, { - from: alignedFrom, - to: alignedTo, - types: opts.types, - }); + rowsUpserted += await persistRollupRows( + userId, + granularity, + rows, + { from: alignedFrom, to: alignedTo, types: opts.types }, + opts.statementTimeoutMs, + ); } return { rowsUpserted, durationMs: Date.now() - startedAt }; @@ -412,13 +435,25 @@ export async function readRollupBuckets( * (epoch / 86400) so the slope is in "units per day", matching the * comprehensive-aggregator + summaries-slice convention. */ -async function runRollupAggregate(input: { - userId: string; - types: MeasurementType[] | undefined; - granularity: RollupGranularity; - from: Date; - to: Date; -}): Promise { +/** Give the statements of a transaction their own limit (SET LOCAL). */ +async function applyStatementBudget( + tx: Pick, + ms: number, +): Promise { + const value = String(Math.max(1, Math.ceil(ms))); + await tx.$executeRaw`SELECT set_config('statement_timeout', ${value}, true), set_config('idle_in_transaction_session_timeout', ${value}, true)`; +} + +async function runRollupAggregate( + input: { + userId: string; + types: MeasurementType[] | undefined; + granularity: RollupGranularity; + from: Date; + to: Date; + }, + db: Pick = prisma, +): Promise { if (input.types && input.types.length === 0) return []; // Postgres rejects `date_trunc($1::text, m."measured_at")` in a @@ -491,7 +526,7 @@ async function runRollupAggregate(input: { AND m."deleted_at" IS NULL GROUP BY m."type", m."source", ${dateTrunc} `; - return prisma.$queryRawUnsafe( + return db.$queryRawUnsafe( sql, input.userId, input.from, @@ -534,7 +569,7 @@ async function runRollupAggregate(input: { AND m."deleted_at" IS NULL GROUP BY m."type", m."source", ${dateTrunc} `; - return prisma.$queryRawUnsafe( + return db.$queryRawUnsafe( sql, input.userId, input.from, @@ -598,6 +633,7 @@ async function persistRollupRows( granularity: RollupGranularity, rows: RollupRow[], prune: PruneScope | null, + statementTimeoutMs?: number, ): Promise { // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected // (type, bucket) partitions across ALL sources before writing, so a source @@ -696,6 +732,8 @@ async function persistRollupRows( let touched = 0; await prisma.$transaction( async (tx) => { + if (statementTimeoutMs) + await applyStatementBudget(tx, statementTimeoutMs); await tx.measurementRollup.deleteMany({ where: deleteWhere }); for (let i = 0; i < ordered.length; i += CHUNK) { const slice = ordered.slice(i, i + CHUNK); @@ -703,7 +741,10 @@ async function persistRollupRows( touched += await tx.$executeRawUnsafe(sql, ...params); } }, - { maxWait: 10_000, timeout: 120_000 }, + { + maxWait: 10_000, + timeout: Math.max(120_000, (statementTimeoutMs ?? 0) + 30_000), + }, ); return touched; } diff --git a/tests/integration/admin-backups-canonical-roundtrip.test.ts b/tests/integration/admin-backups-canonical-roundtrip.test.ts index f26ff943c..ee3015901 100644 --- a/tests/integration/admin-backups-canonical-roundtrip.test.ts +++ b/tests/integration/admin-backups-canonical-roundtrip.test.ts @@ -182,7 +182,11 @@ describe("canonical disaster-recovery backup round-trip", () => { source: "MANUAL", }, }); - const tombstoneDeletedAt = new Date("2026-07-01T12:00:00.000Z"); + // Recent, so the restore writes it back: a tombstone older than the + // tombstone retention is left out, as the nightly purge would drop it. + const tombstoneDeletedAt = new Date( + Math.floor(Date.now() / 1000) * 1000 - 86_400_000, + ); const deletedMeasurement = await prisma.measurement.create({ data: { id: "measurement-deleted-dr", @@ -919,7 +923,11 @@ describe("canonical disaster-recovery backup round-trip", () => { source: "IMPORT" as const, externalId: `external-${index}`, syncVersion: 3, - deletedAt: index % 10 === 0 ? "2026-07-11T00:00:00.000Z" : null, + // Recent: a tombstone past the retention is not written back. + deletedAt: + index % 10 === 0 + ? new Date(Date.now() - 86_400_000).toISOString() + : null, })); const payload = backupPayloadSchema.parse({ schemaVersion: "2", diff --git a/tests/integration/backup-restore-statement-timeout.test.ts b/tests/integration/backup-restore-statement-timeout.test.ts index e58adf1f7..e5ac4eb3b 100644 --- a/tests/integration/backup-restore-statement-timeout.test.ts +++ b/tests/integration/backup-restore-statement-timeout.test.ts @@ -15,12 +15,16 @@ * because nothing was slow. The restore then has to finish and give back * exactly the rows the backup holds. * - * The last test makes the database refuse the clearing step with a disk-full - * error and checks the job says so, and that nothing changed. + * The third test makes the database refuse the clearing step with a + * disk-full error and checks the job says so, and that nothing changed. The + * last makes every write of chart buckets outlast the session limit and + * checks the rebuild after the commit still lands every granularity. * - * Mutation check: remove the `set_config('statement_timeout', …)` call at the - * top of the restore transaction and the restore test goes red with - * `cause: "timeout"`. + * Mutation checks: remove the `set_config('statement_timeout', …)` call at + * the top of the restore transaction and the restore test goes red with + * `cause: "timeout"`; drop `statementTimeoutMs` from the restore's + * `recomputeUserRollups` call and the rebuild test finds no buckets; write + * every tombstone again and the restore test finds the expired ones back. */ import { afterAll, @@ -146,8 +150,14 @@ async function seedAccountWithBackup() { source: "APPLE_HEALTH" as const, measuredAt: new Date(start + i * 60_000), externalId: `hk-${i}`, - // Most of them deleted, as after the nightly consolidation. - deletedAt: i % 8 === 0 ? null : new Date(start), + // Most of them deleted, as after the nightly consolidation: half of + // those yesterday, half long past the tombstone retention. + deletedAt: + i % 8 === 0 + ? null + : i % 2 === 0 + ? new Date(start) + : new Date(Date.now() - 86_400_000), })), }); const { id } = await storeBackupBlob( @@ -218,7 +228,14 @@ describe("restore under a short statement timeout (#1031)", () => { expect(job?.failure ?? null).toBeNull(); expect(res.status).toBe(200); expect(job?.status).toBe("succeeded"); - expect(await measurementsOf(user.id)).toEqual(before); + // Every row comes back but the tombstones past retention, which the + // nightly purge would remove anyway. + const kept = before.filter( + (m) => m.deletedAt === null || m.deletedAt.getTime() > Date.UTC(2025, 0), + ); + expect(kept.some((m) => m.deletedAt !== null)).toBe(true); + expect(kept.length).toBeLessThan(before.length); + expect(await measurementsOf(user.id)).toEqual(kept); }); it("a refused clearing step names its cause and changes nothing", async () => { @@ -249,4 +266,37 @@ describe("restore under a short statement timeout (#1031)", () => { code: "53100", }); }); + + it("the chart rebuild after the commit outlasts the session limit too", async () => { + const { user, backupId } = await seedAccountWithBackup(); + const prisma = housekeeping(); + // Every write of rollup buckets takes longer than the session limit. + await prisma.$executeRawUnsafe( + `CREATE FUNCTION restore_test_slow_rollups() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_sleep(6); RETURN NULL; END $$`, + ); + await prisma.$executeRawUnsafe( + `CREATE TRIGGER restore_test_slow_rollups BEFORE INSERT ON measurement_rollups FOR EACH STATEMENT EXECUTE FUNCTION restore_test_slow_rollups()`, + ); + try { + expect((await restore(backupId)).status).toBe(200); + const buckets = await prisma.measurementRollup.groupBy({ + by: ["granularity"], + where: { userId: user.id }, + _count: true, + }); + expect(buckets.map((b) => b.granularity).sort()).toEqual([ + "DAY", + "MONTH", + "WEEK", + "YEAR", + ]); + } finally { + await prisma.$executeRawUnsafe( + `DROP TRIGGER IF EXISTS restore_test_slow_rollups ON measurement_rollups`, + ); + await prisma.$executeRawUnsafe( + `DROP FUNCTION IF EXISTS restore_test_slow_rollups()`, + ); + } + }, 120_000); }); diff --git a/tests/integration/large-backup-round-trip.test.ts b/tests/integration/large-backup-round-trip.test.ts index 69e77689e..13276e15c 100644 --- a/tests/integration/large-backup-round-trip.test.ts +++ b/tests/integration/large-backup-round-trip.test.ts @@ -76,6 +76,7 @@ async function seedAccount() { cookieJar.set("healthlog_session", session.id); const start = Date.UTC(2024, 0, 1); + const yesterday = new Date(Date.now() - 86_400_000); await prisma.measurement.createMany({ data: Array.from({ length: COUNT }, (_, i) => ({ id: `m-${String(i).padStart(5, "0")}`, @@ -91,7 +92,9 @@ async function seedAccount() { deviceType: "watch", syncVersion: 1 + (i % 4), notesEncrypted: i === 17 ? encryptNote("after the long run") : null, - deletedAt: i % 97 === 0 ? new Date(start) : null, + // Deleted yesterday, inside the tombstone retention, so a weekly + // backup's restore writes it back. + deletedAt: i % 97 === 0 ? yesterday : null, })), }); await prisma.personalRecord.create({