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({