fix(measurements): hold the dense fold back while an account's rollups rebuild

The full rollup backfill refills each day's figures from the raw
readings, and the dense fold deletes those readings once it has their
hourly means. On a large account the backfill can run longer than the
ten minutes the fold's boot job waits, and a year it reached after the
fold had taken it would compute its days from hourly means: the day
count, minimum and maximum of every folded day would change.

The consolidation walk now asks a per-account deferUser hook before it
reads an account, and both entries of the dense fold (the per-account
boot job and the nightly drain) pass isRollupFullBackfillPending, which
looks for a queued or running full fold of the account in pg-boss. The
account is folded on a later run, once its rollup tier is rebuilt.

src/lib/jobs/dense-intraday-retention.ts

@@ -30,6 +30,7 @@ import {
   DENSE_INTRADAY_FOLD_SOURCES,
   DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
 } from "@/lib/measurements/dense-intraday-sources";
+import { isRollupFullBackfillPending } from "@/lib/rollups/full-backfill";
 import { DENSE_INTRADAY_RETENTION_QUEUE } from "./dense-intraday-retention-queue";
 
 /**
@@ -105,6 +106,7 @@ export async function runDenseIntradayRetentionForUser(
   const summary = await runDenseIntradayRetention(prisma, {
     userId,
     shouldStop,
+    deferUser: isRollupFullBackfillPending,
     log: () => {
       // Silent inside the queue handler — the worker logs the totals.
     },

src/lib/jobs/reminder/rollup-handlers.ts

@@ -57,6 +57,7 @@ import {
 } from "@/lib/personal-records/cumulative-pr-rederivation";
 import {
   foldUserRollupHistory,
+  isRollupFullBackfillPending,
   ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
   sendRollupFullBackfillContinuation,
   type RollupFullBackfillPayload,
@@ -527,6 +528,7 @@ export function drainCumulativeHandler(
             {
               dryRun: false,
               shouldStop,
+              deferUser: isRollupFullBackfillPending,
               log: (line) => workerLog("info", line),
             },
           );

src/lib/measurements/__tests__/dense-intraday-retention.test.ts

@@ -161,6 +161,16 @@ describe("runDenseIntradayRetention — retention bound", () => {
     }
   });
 
+  it("leaves an account alone while its rollup tier is being rebuilt", async () => {
+    // A full rollup backfill of the account still has to read the raw
+    // samples; folding them first would leave it only hourly means (#1298).
+    const { mock, findManyMeasurement } = buildPrismaMock({});
+    const deferUser = vi.fn(async () => true);
+    await runDenseIntradayRetention(mock, { deferUser, log: () => {} });
+    expect(deferUser).toHaveBeenCalled();
+    expect(findManyMeasurement).not.toHaveBeenCalled();
+  });
+
   it("scans each fold source's dense types, once per source, Apple Health first", async () => {
     const { mock, findManyMeasurement } = buildPrismaMock({});
     await runDenseIntradayRetention(mock, { log: () => {} });

src/lib/measurements/consolidation-base.ts

@@ -76,6 +76,14 @@ export interface ConsolidationOptions {
    * day this one did not reach.
    */
   shouldStop?: () => boolean;
+  /**
+   * Asked once per account before it is walked; `true` leaves the account
+   * for a later run, and nothing of it is read or written. The dense fold
+   * passes `isRollupFullBackfillPending`: while an account's rollup tier is
+   * rebuilt from its raw readings, folding those readings would leave the
+   * rebuild only hourly means to take the old days' figures from.
+   */
+  deferUser?: (userId: string) => Promise<boolean>;
 }
 
 /**
@@ -458,6 +466,10 @@ export async function runConsolidation<TType extends MeasurementType>(
   const users = await loadConsolidationUsers(prismaClient, options.userId);
 
   walk: for (const user of users) {
+    if (await options.deferUser?.(user.id)) {
+      annotate({ meta: { consolidation_deferred_user: true } });
+      continue;
+    }
     // The zone the server publishes for the account (`/api/auth/me`), which
     // is the zone the phone cuts its `stats:` days and hour buckets in and
     // the one the hour-bucket writer stores them under. A stored value the

src/lib/measurements/dense-intraday-retention.ts

@@ -66,7 +66,11 @@ import {
   dailyStatsExternalId,
   hkIdentifierForType,
 } from "./apple-health-mapping";
-import { runConsolidation, type DayWriteOutcome } from "./consolidation-base";
+import {
+  runConsolidation,
+  type ConsolidationOptions,
+  type DayWriteOutcome,
+} from "./consolidation-base";
 import {
   canonicalHourlyTimestamp,
   hourOfDayForUserTz,
@@ -438,13 +442,8 @@ export interface DenseIntradayRetentionSummary {
   };
 }
 
-export interface DenseIntradayRetentionOptions {
-  /** Limit the pass to a single user. Default = every user. */
-  userId?: string;
-  /** Preview-only mode — no DB writes. */
-  dryRun?: boolean;
-  /** Logger sink — defaults to `console.log`. */
-  log?: (line: string) => void;
+/** The shared drain options (`ConsolidationOptions`) and the retention window. */
+export interface DenseIntradayRetentionOptions extends ConsolidationOptions {
   /**
    * Retention window in days. Per-sample rows OLDER than `now() -
    * retentionDays` are folded; rows inside the window stay raw. Defaults to
@@ -453,11 +452,6 @@ export interface DenseIntradayRetentionOptions {
    * full-collapse, never by the scheduled pass.
    */
   retentionDays?: number;
-  /**
-   * Ends the pass cleanly before the next day when it returns `true`; the
-   * summary then reports `stoppedEarly`. See `ConsolidationOptions`.
-   */
-  shouldStop?: () => boolean;
 }
 
 /**

src/lib/rollups/full-backfill.ts

@@ -194,6 +194,31 @@ export async function sendRollupFullBackfillContinuation(
   return jobId !== null;
 }
 
+/**
+ * Whether a full fold of the account is queued or running. The dense fold
+ * leaves such an account alone (`ConsolidationOptions.deferUser`): it deletes
+ * the raw samples a fold still has to read, and a year the full backfill
+ * reached after the dense fold would take its day figures from hourly means.
+ * False when the queue schema cannot be read, as a web-only deployment has
+ * no full backfill either.
+ */
+export async function isRollupFullBackfillPending(
+  userId: string,
+): Promise<boolean> {
+  try {
+    const rows = await prisma.$queryRaw<Array<{ n: number }>>`
+      SELECT count(*)::int AS n
+      FROM pgboss.job
+      WHERE name = ${ROLLUP_FULL_BACKFILL_QUEUE}
+        AND state IN ('created', 'retry', 'active')
+        AND data->>'userId' = ${userId}
+    `;
+    return (rows[0]?.n ?? 0) > 0;
+  } catch {
+    return false;
+  }
+}
+
 /**
  * Boot-time discovery. Finds the accounts whose rollup tier misses something
  * inside the fold window and enqueues one full fold per account.

tests/integration/rollup-full-backfill.test.ts

@@ -11,13 +11,24 @@
  *     reading in the window, finds the same accounts as the statement it
  *     replaced, kept verbatim below as the reference.
  */
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { PgBoss } from "pg-boss";
+import {
+  afterAll,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from "vitest";
 
 import { Prisma, type MeasurementSource } from "@/generated/prisma/client";
 import { activeAccountSql } from "@/lib/account-deletion/active-account";
 import {
   enqueueBootTimeRollupBackfill,
   foldUserRollupHistory,
+  isRollupFullBackfillPending,
+  ROLLUP_FULL_BACKFILL_QUEUE,
   ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS,
 } from "@/lib/rollups/full-backfill";
 import {
@@ -304,3 +315,33 @@ describe("enqueueBootTimeRollupBackfill", () => {
     expect(found).toEqual(await referenceDiscovery());
   });
 });
+
+describe("isRollupFullBackfillPending", () => {
+  let boss: PgBoss;
+
+  beforeAll(async () => {
+    boss = new PgBoss({
+      connectionString: process.env.DATABASE_URL,
+      supervise: false,
+      schedule: false,
+    });
+    await boss.start();
+    await boss.createQueue(ROLLUP_FULL_BACKFILL_QUEUE);
+  }, 60_000);
+
+  afterAll(async () => {
+    await boss?.stop({ graceful: false, close: true });
+    await prisma.$executeRawUnsafe(`DROP SCHEMA IF EXISTS pgboss CASCADE`);
+  });
+
+  it("is true while the account's full fold is queued or running, and only for that account", async () => {
+    expect(await isRollupFullBackfillPending("user-a")).toBe(false);
+    await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, { userId: "user-a" });
+    expect(await isRollupFullBackfillPending("user-a")).toBe(true);
+    expect(await isRollupFullBackfillPending("user-b")).toBe(false);
+    const [job] = await boss.fetch(ROLLUP_FULL_BACKFILL_QUEUE);
+    expect(await isRollupFullBackfillPending("user-a")).toBe(true);
+    await boss.complete(ROLLUP_FULL_BACKFILL_QUEUE, job.id);
+    expect(await isRollupFullBackfillPending("user-a")).toBe(false);
+  });
+});