From 07cf9bba3e54166db000c0a571f28fb2e114ab72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 09:28:31 +0100 Subject: [PATCH] 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 | 2 + src/lib/jobs/reminder/rollup-handlers.ts | 2 + .../dense-intraday-retention.test.ts | 10 +++++ src/lib/measurements/consolidation-base.ts | 12 ++++++ .../measurements/dense-intraday-retention.ts | 20 +++------ src/lib/rollups/full-backfill.ts | 25 +++++++++++ .../integration/rollup-full-backfill.test.ts | 43 ++++++++++++++++++- 7 files changed, 100 insertions(+), 14 deletions(-) diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index 2ced5d3f5..6b679fbfe 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/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. }, diff --git a/src/lib/jobs/reminder/rollup-handlers.ts b/src/lib/jobs/reminder/rollup-handlers.ts index 5a0db0c22..d9bde2063 100644 --- a/src/lib/jobs/reminder/rollup-handlers.ts +++ b/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), }, ); diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index 6c7c88ba3..8d44ef417 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/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: () => {} }); diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index c74e79e52..5cab6ae06 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/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; } /** @@ -458,6 +466,10 @@ export async function runConsolidation( 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 diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 89abd6362..e75cd7546 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/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; } /** diff --git a/src/lib/rollups/full-backfill.ts b/src/lib/rollups/full-backfill.ts index e531a1b56..1d42b7a33 100644 --- a/src/lib/rollups/full-backfill.ts +++ b/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 { + try { + const rows = await prisma.$queryRaw>` + 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. diff --git a/tests/integration/rollup-full-backfill.test.ts b/tests/integration/rollup-full-backfill.test.ts index 01f6fbbd6..cdf0742fa 100644 --- a/tests/integration/rollup-full-backfill.test.ts +++ b/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); + }); +});