Merge branch 'fix/restore-large-account-clearing' into release/v1.39.4

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<string>();
         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", {

src/lib/import/__tests__/unzip-export-xml.test.ts

@@ -187,12 +187,20 @@ describe("extractExportXml", () => {
     const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-"));
     const zipPath = join(tmp, "export.zip");
     writeFileSync(zipPath, zip);
-    const staged = () =>
-      readdirSync(tmpdir()).filter((n) => n.startsWith("healthlog-import-"));
-    const before = new Set(staged());
-
-    await expect(extractExportXml(zipPath)).rejects.toThrow(/zip bomb/);
-    expect(staged().filter((n) => !before.has(n))).toEqual([]);
+    // Staged into a directory of this test's own. The shared system tmpdir
+    // also holds whatever other test files extracting in parallel have
+    // staged at that moment, and a before/after diff over it failed on
+    // their files, not this one's.
+    const staging = mkdtempSync(join(tmpdir(), "healthlog-unzip-staging-"));
+    const previousTmpdir = process.env.TMPDIR;
+    process.env.TMPDIR = staging;
+    try {
+      await expect(extractExportXml(zipPath)).rejects.toThrow(/zip bomb/);
+    } finally {
+      if (previousTmpdir === undefined) delete process.env.TMPDIR;
+      else process.env.TMPDIR = previousTmpdir;
+    }
+    expect(readdirSync(staging)).toEqual([]);
     // Inflating past the 64 MiB floor takes a moment on a loaded runner.
   }, 60_000);
 

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<RollupRow[]> {
+/** Give the statements of a transaction their own limit (SET LOCAL). */
+async function applyStatementBudget(
+  tx: Pick<typeof prisma, "$executeRaw">,
+  ms: number,
+): Promise<void> {
+  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<typeof prisma, "$queryRawUnsafe"> = prisma,
+): Promise<RollupRow[]> {
   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<RollupRow[]>(
+    return db.$queryRawUnsafe<RollupRow[]>(
       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<RollupRow[]>(
+  return db.$queryRawUnsafe<RollupRow[]>(
     sql,
     input.userId,
     input.from,
@@ -598,6 +633,7 @@ async function persistRollupRows(
   granularity: RollupGranularity,
   rows: RollupRow[],
   prune: PruneScope | null,
+  statementTimeoutMs?: number,
 ): Promise<number> {
   // 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;
 }