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); + }); +});