diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 66e675fd4..d86b28d50 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 84, - "total": 94928, + "total": 94733, "entries": { "app/medications/page-client.tsx": 839, "components/charts/mood-chart.tsx": 1166, @@ -78,7 +78,7 @@ "lib/medications/scheduling/slot-upsert.ts": 912, "lib/notifications/senders/apns.ts": 1100, "lib/oura/sync.ts": 900, - "lib/rollups/measurement-rollups.ts": 1496, + "lib/rollups/measurement-rollups.ts": 1302, "lib/search/search-records.ts": 996, "lib/sharing/grants.ts": 806, "lib/support/report.ts": 903, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index f319e154c..0ff7da49e 100644 --- a/src/__tests__/baselines/version-stamp-comments.json +++ b/src/__tests__/baselines/version-stamp-comments.json @@ -1,7 +1,7 @@ { "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.", "count": 2897, - "total": 9236, + "total": 9230, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2748,7 +2748,7 @@ "lib/rollups/measurement-coverage.ts": 1, "lib/rollups/measurement-read-wmy.ts": 4, "lib/rollups/measurement-read.ts": 10, - "lib/rollups/measurement-rollups.ts": 18, + "lib/rollups/measurement-rollups.ts": 12, "lib/rollups/mood-rollups.ts": 4, "lib/rollups/tiered-context.ts": 6, "lib/score-config/rows.ts": 1, diff --git a/src/app/api/admin/rollups/recompute/route.ts b/src/app/api/admin/rollups/recompute/route.ts index 480b36a97..63f612c49 100644 --- a/src/app/api/admin/rollups/recompute/route.ts +++ b/src/app/api/admin/rollups/recompute/route.ts @@ -15,8 +15,8 @@ * { userId?: string } * * - `userId` — single user mode. Synchronously calls - * `recomputeUserRollups(userId)` and returns the row count when - * the fold completes. The call is awaited inside the request, so + * `foldUserRollupHistory(userId)` (the full fold, a calendar year per + * statement) and returns the row count when the fold completes. The call is awaited inside the request, so * the operator sees the duration in the response. * - `userId` omitted — calls `enqueueBootTimeRollupBackfill()` to * re-run the worker-boot discovery (`{ enqueued, skipped }` shape). @@ -37,8 +37,8 @@ import { import { annotate } from "@/lib/logging/context"; import { enqueueBootTimeRollupBackfill, - recomputeUserRollups, -} from "@/lib/rollups/measurement-rollups"; + foldUserRollupHistory, +} from "@/lib/rollups/full-backfill"; interface RecomputeBody { userId?: string; @@ -68,9 +68,9 @@ export const POST = apiHandler(async (request: NextRequest) => { details: { targetUserId: userId, mode: "single" }, }); - let result: Awaited>; + let result: Awaited>; try { - result = await recomputeUserRollups(userId); + result = await foldUserRollupHistory(userId); } catch (err) { const message = err instanceof Error ? err.message : String(err); return apiError(message, 500); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 3ede654d1..5cbc2d758 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -330,6 +330,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "users_synced", "users_useful", "wedge_skipped", + // The full rollup backfill: calendar-year windows one run folded. + "windows_folded", ]); export const MAX_JOB_FACTS = 32; diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 49fc6d154..fb14c254a 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -51,7 +51,7 @@ import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-int import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker"; import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; -import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/measurement-rollups"; +import { ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS } from "@/lib/rollups/full-backfill"; import { MORNING_DIGEST_REFRESH_EXPIRE_SECONDS } from "@/lib/jobs/morning-digest-refresh-shared"; import { INSIGHT_STATUS_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/insight-status-generate-queue"; import { WORKOUT_INSIGHT_GENERATE_EXPIRE_SECONDS } from "@/lib/jobs/workout-insight-generate-shared"; @@ -261,7 +261,7 @@ export const QUEUE_RUNTIME: Readonly> = { fn: "handleRollupFullBackfill", }, exclusive: "lockedPass", - why: "One account's five-year fold: four aggregates under the statement timeout and four persist transactions of up to 120 s each, about thirteen minutes at worst under the defaults and unbounded when the operator lifts the statement timeout.", + why: "One account's five-year fold, a calendar year at a time with every statement under its own five-minute limit; the run stops at its budget between windows and hands the remaining years to a follow-up.", }, "mood-rollup-recompute": short("One bucket."), "mood-rollup-full-backfill": short( diff --git a/src/lib/jobs/registry/rollup.ts b/src/lib/jobs/registry/rollup.ts index cb885c0dc..b4e68e834 100644 --- a/src/lib/jobs/registry/rollup.ts +++ b/src/lib/jobs/registry/rollup.ts @@ -56,9 +56,11 @@ import { import { ROLLUP_FULL_BACKFILL_CONCURRENCY, ROLLUP_FULL_BACKFILL_QUEUE, + type RollupFullBackfillPayload, +} from "@/lib/rollups/full-backfill"; +import { ROLLUP_RECOMPUTE_CONCURRENCY, ROLLUP_RECOMPUTE_QUEUE, - type RollupFullBackfillPayload, } from "@/lib/rollups/measurement-rollups"; import { MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY, @@ -213,7 +215,7 @@ const discovery: readonly BootDiscovery[] = [ name: "rollup-full-backfill", run: async () => ( - await import("@/lib/rollups/measurement-rollups") + await import("@/lib/rollups/full-backfill") ).enqueueBootTimeRollupBackfill(BOOT_BACKFILL_STAGGER_SECONDS * 0), }, { diff --git a/src/lib/jobs/reminder/rollup-handlers.ts b/src/lib/jobs/reminder/rollup-handlers.ts index 74c765d02..5a0db0c22 100644 --- a/src/lib/jobs/reminder/rollup-handlers.ts +++ b/src/lib/jobs/reminder/rollup-handlers.ts @@ -56,8 +56,13 @@ import { type CumulativePrRederivePayload, } from "@/lib/personal-records/cumulative-pr-rederivation"; import { - recomputeUserRollups, + foldUserRollupHistory, + ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + sendRollupFullBackfillContinuation, type RollupFullBackfillPayload, +} from "@/lib/rollups/full-backfill"; +import { + recomputeUserRollups, type RollupRecomputePayload, } from "@/lib/rollups/measurement-rollups"; import { @@ -91,35 +96,47 @@ export async function handleRollupRecompute( } /** - * The full fold of one uncovered account across the default window and every - * granularity. Each granularity commits on its own, so a run stopped on its - * budget throws and the retry folds the rest. + * The full fold of one account across the five-year window, a calendar year + * at a time and every statement under the full backfill's own limit (#1298). + * A run that reaches its job budget hands the years it did not reach to a + * follow-up and completes; only a run that could not finish even one window + * fails, and its retry starts that window again. */ export async function handleRollupFullBackfill( jobs: Job[], ): Promise { const shouldStop = jobBudget(jobs); let rowsUpsertedTotal = 0; + let windowsTotal = 0; + let continued = 0; for (const job of jobs) { - const { userId } = job.data; - const { rowsUpserted, durationMs, stopped } = await recomputeUserRollups( - userId, - { shouldStop }, - ); - if (stopped) { - throw new Error( - `rollup-full-backfill for user ${userId} stopped on its job budget; the retry folds the rest`, - ); + const { userId, resumeBefore } = job.data; + const done = await foldUserRollupHistory(userId, { + before: resumeBefore ? new Date(resumeBefore) : undefined, + shouldStop, + statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + }); + rowsUpsertedTotal += done.rowsUpserted; + windowsTotal += done.windowsFolded; + if (done.resumeBefore) { + if (done.windowsFolded === 0) { + throw new Error( + `rollup-full-backfill for user ${userId} stopped on its job budget before it folded a window; the retry starts that window again`, + ); + } + await sendRollupFullBackfillContinuation(userId, done.resumeBefore); + continued += 1; } - rowsUpsertedTotal += rowsUpserted; workerLog( "info", - `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`, + `[rollup-full-backfill] user=${userId} rows=${done.rowsUpserted} windows=${done.windowsFolded} duration=${done.durationMs}ms${done.resumeBefore ? ` continuesBefore=${done.resumeBefore.toISOString()}` : ""}`, ); } return jobDone({ users: jobs.length, rows_upserted: rowsUpsertedTotal, + windows_folded: windowsTotal, + continued, }); } diff --git a/src/lib/rollups/__tests__/full-backfill.test.ts b/src/lib/rollups/__tests__/full-backfill.test.ts new file mode 100644 index 000000000..5b87f54c2 --- /dev/null +++ b/src/lib/rollups/__tests__/full-backfill.test.ts @@ -0,0 +1,231 @@ +/** + * The full backfill's windowed fold and its handler (#1298): the windows it + * cuts, the per-statement limit every window runs under, and how a run that + * reaches its job budget hands the remaining years on instead of failing. + * + * The fold itself is mocked here; that the windows write exactly the buckets + * one pass over the whole span writes is pinned against Postgres in + * `tests/integration/rollup-full-backfill.test.ts`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +const mocks = vi.hoisted(() => ({ + recomputeUserRollups: vi.fn(), + bossSend: vi.fn(), + getGlobalBoss: vi.fn(), +})); + +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({ + ...(await importOriginal< + typeof import("@/lib/rollups/measurement-rollups") + >()), + recomputeUserRollups: mocks.recomputeUserRollups, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => mocks.getGlobalBoss(), +})); + +import { handleRollupFullBackfill } from "@/lib/jobs/reminder/rollup-handlers"; +import { + foldUserRollupHistory, + fullBackfillWindows, + ROLLUP_FULL_BACKFILL_QUEUE, + ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + type RollupFullBackfillPayload, +} from "@/lib/rollups/full-backfill"; +import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; + +const NOW = new Date("2026-10-29T10:00:00.000Z"); + +function job( + data: Partial = {}, +): Job { + return { + id: "job-1", + name: ROLLUP_FULL_BACKFILL_QUEUE, + data: { userId: "user-1", enqueuedAt: NOW.toISOString(), ...data }, + expireInSeconds: 3600, + signal: new AbortController().signal, + } as unknown as Job; +} + +beforeEach(() => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + mocks.recomputeUserRollups.mockReset(); + mocks.recomputeUserRollups.mockResolvedValue({ + rowsUpserted: 10, + durationMs: 1, + }); + mocks.bossSend.mockReset(); + mocks.bossSend.mockResolvedValue("job-next"); + mocks.getGlobalBoss.mockReset(); + mocks.getGlobalBoss.mockReturnValue({ send: mocks.bossSend }); +}); + +describe("fullBackfillWindows", () => { + it("cuts the span at UTC New Year, newest first, and covers it exactly", () => { + const from = new Date("2021-10-30T10:00:00.000Z"); + const windows = fullBackfillWindows(from, NOW); + expect( + windows.map((w) => [w.from.toISOString(), w.to.toISOString()]), + ).toEqual([ + ["2026-01-01T00:00:00.000Z", NOW.toISOString()], + ["2025-01-01T00:00:00.000Z", "2026-01-01T00:00:00.000Z"], + ["2024-01-01T00:00:00.000Z", "2025-01-01T00:00:00.000Z"], + ["2023-01-01T00:00:00.000Z", "2024-01-01T00:00:00.000Z"], + ["2022-01-01T00:00:00.000Z", "2023-01-01T00:00:00.000Z"], + [from.toISOString(), "2022-01-01T00:00:00.000Z"], + ]); + for (const w of windows) { + expect(w.to.getTime() - w.from.getTime()).toBeLessThanOrEqual( + 366 * 86_400_000, + ); + } + }); + + it("treats an end on New Year as the end of the year before", () => { + const windows = fullBackfillWindows( + new Date("2024-06-01T00:00:00.000Z"), + new Date("2025-01-01T00:00:00.000Z"), + ); + expect(windows).toEqual([ + { + from: new Date("2024-06-01T00:00:00.000Z"), + to: new Date("2025-01-01T00:00:00.000Z"), + }, + ]); + }); + + it("is empty for an empty span", () => { + expect(fullBackfillWindows(NOW, NOW)).toEqual([]); + }); +}); + +describe("foldUserRollupHistory", () => { + it("folds every window of the fold window under the given statement limit", async () => { + const done = await foldUserRollupHistory("user-1", { + statementTimeoutMs: 1234, + }); + const expected = fullBackfillWindows( + new Date(NOW.getTime() - ROLLUP_FOLD_WINDOW_MS), + NOW, + ); + expect(mocks.recomputeUserRollups).toHaveBeenCalledTimes(expected.length); + expected.forEach((w, i) => { + expect(mocks.recomputeUserRollups).toHaveBeenNthCalledWith( + i + 1, + "user-1", + expect.objectContaining({ + from: w.from, + to: w.to, + statementTimeoutMs: 1234, + }), + ); + }); + expect(done).toMatchObject({ + rowsUpserted: 10 * expected.length, + windowsFolded: expected.length, + }); + expect(done.resumeBefore).toBeUndefined(); + }); + + it("stops before the next window and names where the rest begins", async () => { + const done = await foldUserRollupHistory("user-1", { + shouldStop: () => mocks.recomputeUserRollups.mock.calls.length >= 1, + }); + expect(mocks.recomputeUserRollups).toHaveBeenCalledTimes(1); + expect(done.windowsFolded).toBe(1); + expect(done.resumeBefore).toEqual(new Date("2026-01-01T00:00:00.000Z")); + }); + + it("reports a window the fold stopped inside as still to fold", async () => { + mocks.recomputeUserRollups + .mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 }) + .mockResolvedValueOnce({ rowsUpserted: 2, durationMs: 1, stopped: true }); + const done = await foldUserRollupHistory("user-1", { + shouldStop: () => false, + }); + expect(done.windowsFolded).toBe(1); + expect(done.resumeBefore).toEqual(new Date("2026-01-01T00:00:00.000Z")); + }); + + it("resumes with only the windows that end at or before `before`", async () => { + await foldUserRollupHistory("user-1", { + before: new Date("2025-01-01T00:00:00.000Z"), + }); + const tos = mocks.recomputeUserRollups.mock.calls.map( + (call) => (call[1] as { to: Date }).to, + ); + expect(tos[0]).toEqual(new Date("2025-01-01T00:00:00.000Z")); + for (const to of tos) { + expect(to.getTime()).toBeLessThanOrEqual( + new Date("2025-01-01T00:00:00.000Z").getTime(), + ); + } + }); +}); + +describe("handleRollupFullBackfill", () => { + it("runs every statement of the fold under the full backfill's own limit", async () => { + await handleRollupFullBackfill([job()]); + expect(mocks.recomputeUserRollups).toHaveBeenCalled(); + for (const call of mocks.recomputeUserRollups.mock.calls) { + expect(call[1]).toMatchObject({ + statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + }); + } + expect(mocks.bossSend).not.toHaveBeenCalled(); + }); + + it("hands the years it did not reach to a follow-up and completes", async () => { + mocks.recomputeUserRollups + .mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 }) + .mockResolvedValueOnce({ rowsUpserted: 0, durationMs: 1, stopped: true }); + const outcome = await handleRollupFullBackfill([job()]); + expect(outcome.ok).toBe(true); + expect(mocks.bossSend).toHaveBeenCalledTimes(1); + const [queue, payload, options] = mocks.bossSend.mock.calls[0]; + expect(queue).toBe(ROLLUP_FULL_BACKFILL_QUEUE); + expect(payload).toMatchObject({ + userId: "user-1", + resumeBefore: "2026-01-01T00:00:00.000Z", + }); + // Not the boot key: the running job still holds that one. + expect(options.singletonKey).toBe( + "boot-backfill|user-1|2026-01-01T00:00:00.000Z", + ); + expect(options.startAfter).toBeGreaterThan(0); + }); + + it("resumes a follow-up where the run before it stopped", async () => { + await handleRollupFullBackfill([ + job({ resumeBefore: "2024-01-01T00:00:00.000Z" }), + ]); + expect( + (mocks.recomputeUserRollups.mock.calls[0][1] as { to: Date }).to, + ).toEqual(new Date("2024-01-01T00:00:00.000Z")); + }); + + it("fails a run that could not finish one window, so its retry starts it again", async () => { + mocks.recomputeUserRollups.mockResolvedValueOnce({ + rowsUpserted: 0, + durationMs: 1, + stopped: true, + }); + await expect(handleRollupFullBackfill([job()])).rejects.toThrow( + /before it folded a window/, + ); + expect(mocks.bossSend).not.toHaveBeenCalled(); + }); + + it("fails the run when the follow-up cannot be queued", async () => { + mocks.getGlobalBoss.mockReturnValue(null); + mocks.recomputeUserRollups + .mockResolvedValueOnce({ rowsUpserted: 5, durationMs: 1 }) + .mockResolvedValueOnce({ rowsUpserted: 0, durationMs: 1, stopped: true }); + await expect(handleRollupFullBackfill([job()])).rejects.toThrow(/no queue/); + }); +}); diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 69161ada6..88c499569 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -29,6 +29,7 @@ const mocks = vi.hoisted(() => ({ findMany: vi.fn(), findFirstMeasurement: vi.fn(), txExecuteRawUnsafe: vi.fn(), + txExecuteRaw: vi.fn(), bossSend: vi.fn(), getGlobalBossMock: vi.fn(), })); @@ -87,10 +88,12 @@ void queryRaw; import { ROLLUP_FULL_BACKFILL_QUEUE, + enqueueBootTimeRollupBackfill, +} from "../full-backfill"; +import { ROLLUP_RECOMPUTE_QUEUE, _resetEnsureUserRollupsFreshInFlightForTests, collapseToTypeDayKeys, - enqueueBootTimeRollupBackfill, enqueueRollupRecompute, ensureUserRollupsFresh, recomputeBucketsForMeasurement, @@ -127,6 +130,8 @@ beforeEach(() => { createMany: mocks.createMany, }, $executeRawUnsafe: mocks.txExecuteRawUnsafe, + $executeRaw: mocks.txExecuteRaw, + $queryRaw: mocks.queryRaw, }); } return Promise.all(arg as unknown[]); @@ -996,10 +1001,10 @@ describe("enqueueBootTimeRollupBackfill", () => { const sqlParts = queryRaw.mock.calls[0][0] as TemplateStringsArray; const sqlText = Array.isArray(sqlParts) ? sqlParts.join("?") : ""; - // Inner DISTINCT widens to `(user_id, type, bucket_start)`. - expect(sqlText).toContain("date_trunc('day', m.\"measured_at\")"); - // LEFT JOIN now compares bucket_start on both sides. - expect(sqlText).toContain('r."bucket_start" = mt."bucket_start"'); + // The day walk steps from each day's first reading to the next UTC day. + expect(sqlText).toContain("date_trunc('day', d.at) + interval '1 day'"); + // The coverage check compares the walked day with the DAY bucket. + expect(sqlText).toContain(`r."bucket_start" = date_trunc('day', d.at)`); }); // v1.16.10 — discovery and fold share one trailing window. Unbounded @@ -1017,12 +1022,14 @@ describe("enqueueBootTimeRollupBackfill", () => { const after = Date.now(); const sqlParts = queryRaw.mock.calls[0][0] as TemplateStringsArray; const sqlText = Array.isArray(sqlParts) ? sqlParts.join("?") : ""; - // The inner scan filters on measured_at >= . + // The day walk starts at measured_at >= . expect(sqlText).toContain('m."measured_at" >='); // The bound parameter is now − ROLLUP_FOLD_WINDOW_MS (5 years), // matching the `recomputeUserRollups` default the backfill worker // runs with. - const boundArg = queryRaw.mock.calls[0][1] as Date; + const boundArg = queryRaw.mock.calls[0].find( + (arg: unknown) => arg instanceof Date, + ) as Date; expect(boundArg).toBeInstanceOf(Date); const fiveYearsMs = 5 * 365 * 24 * 60 * 60 * 1000; expect(boundArg.getTime()).toBeGreaterThanOrEqual(before - fiveYearsMs); diff --git a/src/lib/rollups/full-backfill.ts b/src/lib/rollups/full-backfill.ts new file mode 100644 index 000000000..e531a1b56 --- /dev/null +++ b/src/lib/rollups/full-backfill.ts @@ -0,0 +1,337 @@ +/** + * The full fold of one account's rollup tier, and the boot discovery that + * decides which accounts need one. + * + * The write hooks keep the DAY bucket of each touched day current and the + * read path warms the last 90 days, but neither folds the deep WEEK / MONTH / + * YEAR history, and neither refills a column a migration adds to rows that + * already exist (the accumulators of 0190, the reading extremes of 0417). The + * full backfill does both, once per account the discovery finds. + * + * Sized for large accounts (#1298). The fold used to be four aggregates over + * the whole five-year window, each held to the connection's statement limit + * (60 s unless `DATABASE_STATEMENT_TIMEOUT_MS` says otherwise). On an account + * with tens of millions of readings no aggregate finished in time, the job + * failed on every boot, and the columns 0417 added never filled, so every + * reader that needs them stayed on the raw table and timed out as well. The + * fold now works through the window one UTC calendar year at a time, newest + * first, and every statement of the full backfill runs under its own limit + * (`ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`), the shape the pulse day-mean + * rebuild already had. Each window commits on its own, so a run that reaches + * its job budget hands the remaining years to a follow-up job instead of + * starting over on the retry. + */ +import { activeAccountSql } from "@/lib/account-deletion/active-account"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + applyStatementBudget, + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; + +/** + * The boot-time full-fold queue: one job per account the discovery finds, so + * a self-hosted instance converges on the rollup tier without an operator. + */ +export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill"; + +/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ +export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; + +/** + * The full fold's expiry: an hour. A run stops itself at three quarters of it + * (`jobBudget`), between two windows or two granularities, and hands what is + * left to a follow-up, so the expiry bounds one run, not the whole account. + * The margin above the budget covers the statement in flight, which can run + * for `ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`. + */ +export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60; + +/** + * The limit of every statement the full backfill and its discovery run, in + * place of the connection's request limit. Five minutes, as the pulse + * day-mean rebuild uses: one calendar year of one account's readings, + * aggregated per granularity, stays well inside it on the largest accounts + * reported, while a statement that would run longer still ends. + */ +export const ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS = 5 * 60 * 1000; + +/** Seconds a follow-up waits before it picks up the remaining years. */ +export const ROLLUP_FULL_BACKFILL_CONTINUATION_DELAY_SECONDS = 60; + +/** Payload of one full-fold job. */ +export interface RollupFullBackfillPayload { + userId: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; + /** + * ISO instant. A follow-up folds only the windows that end at or before + * it: the run that sent it folded everything newer. + */ + resumeBefore?: string; +} + +/** The retry policy of every full-fold send. */ +const RETRY_OPTIONS = { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, +} as const; + +/** + * `[from, to)` windows covering `[from, to)`, cut at UTC New Year and + * returned newest first. Every DAY, MONTH and YEAR bucket lies inside one + * window, and a WEEK that spans New Year is folded whole by both of its + * windows (the fold snaps each window out to whole buckets), so the windows + * write exactly the buckets one fold over the whole span writes. Pure. + */ +export function fullBackfillWindows( + from: Date, + to: Date, +): Array<{ from: Date; to: Date }> { + const out: Array<{ from: Date; to: Date }> = []; + let end = to.getTime(); + const floor = from.getTime(); + while (end > floor) { + const year = new Date(end).getUTCFullYear(); + const newYear = Date.UTC(year, 0, 1); + const start = Math.max( + floor, + newYear === end ? Date.UTC(year - 1, 0, 1) : newYear, + ); + out.push({ from: new Date(start), to: new Date(end) }); + end = start; + } + return out; +} + +export interface RollupHistoryFold { + rowsUpserted: number; + durationMs: number; + /** Windows folded completely by this call. */ + windowsFolded: number; + /** + * Set when `shouldStop` ended the fold: every window that ends at or + * before this instant is still to fold, and nothing newer is. + */ + resumeBefore?: Date; +} + +/** + * Fold one account's rollup tier over the five-year window, every type and + * granularity, one calendar year at a time, newest first. The same buckets + * `recomputeUserRollups(userId)` writes in one pass; see + * `fullBackfillWindows` for why the windows change nothing about them. + * + * `before` resumes a fold that stopped: only the windows ending at or before + * it run. `shouldStop` is asked before each window and between the + * granularities of a window; a window it interrupts is folded again whole by + * the follow-up, so `resumeBefore` is always a window's end. + */ +export async function foldUserRollupHistory( + userId: string, + opts: { + before?: Date; + shouldStop?: () => boolean; + statementTimeoutMs?: number; + now?: Date; + } = {}, +): Promise { + const startedAt = Date.now(); + const now = opts.now ?? new Date(); + const floor = new Date(now.getTime() - ROLLUP_FOLD_WINDOW_MS); + const ceiling = opts.before && opts.before < now ? opts.before : now; + let rowsUpserted = 0; + let windowsFolded = 0; + const result = (resumeBefore?: Date): RollupHistoryFold => ({ + rowsUpserted, + durationMs: Date.now() - startedAt, + windowsFolded, + ...(resumeBefore ? { resumeBefore } : {}), + }); + for (const window of fullBackfillWindows(floor, ceiling)) { + if (opts.shouldStop?.()) return result(window.to); + const done = await recomputeUserRollups(userId, { + from: window.from, + to: window.to, + statementTimeoutMs: opts.statementTimeoutMs, + shouldStop: opts.shouldStop, + }); + rowsUpserted += done.rowsUpserted; + if (done.stopped) return result(window.to); + windowsFolded += 1; + } + return result(); +} + +/** + * Hand the windows a stopped fold did not reach to a follow-up job. Keyed on + * the account and the resume point: the running job still holds the account's + * boot key, and a second follow-up for the same point is the same work. + * Throws when no queue is attached or the send fails, so the run fails and + * its retry folds the same windows again rather than leaving them. + */ +export async function sendRollupFullBackfillContinuation( + userId: string, + resumeBefore: Date, +): Promise { + const boss = getGlobalBoss(); + if (!boss) { + throw new Error("rollup-full-backfill: no queue to hand the rest to"); + } + const payload: RollupFullBackfillPayload = { + userId, + enqueuedAt: new Date().toISOString(), + resumeBefore: resumeBefore.toISOString(), + }; + const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { + ...RETRY_OPTIONS, + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, + singletonKey: `boot-backfill|${userId}|${payload.resumeBefore}`, + startAfter: ROLLUP_FULL_BACKFILL_CONTINUATION_DELAY_SECONDS, + }); + return jobId !== null; +} + +/** + * Boot-time discovery. Finds the accounts whose rollup tier misses something + * inside the fold window and enqueues one full fold per account. + * + * Two arms: + * + * 1. Missing coverage per day: a (user, type, UTC day) with a live reading + * and no DAY rollup row. Per day rather than per type, because a sync + * or a restore that writes without the write hook leaves gaps in a type + * that already has buckets. + * 2. A DAY rollup row inside the window whose regression accumulators + * (0190) or reading extremes (0417) are NULL: a row folded before the + * column existed. New folds write them, so the arm drains to empty. + * + * Both are bounded to the fold window. The fold writes nothing older, so an + * older gap would flag the account on every boot and never close. + * + * Arm 1 used to read every live reading in the window, which on an account + * with tens of millions of rows is the same scan that kept the fold itself + * from finishing. It now walks the days instead: per account and type, the + * first live reading at or after the start of each next UTC day, read off + * the live `(user_id, type, measured_at)` index, so the cost follows the + * number of days with readings, not the number of readings. The statement + * also runs under the full backfill's own limit. + * + * Idempotent across reboots: an account drops off once it is covered, and + * the per-account `singletonKey` coalesces a send while a fold is queued. + * Best-effort: errors come back in the result so the worker boot never fails. + */ +export async function enqueueBootTimeRollupBackfill( + // Optional boot-storm stagger, seconds. Default 0 keeps immediate semantics + // for any non-boot caller. + startAfterSeconds: number = 0, +): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const users = await prisma.$transaction( + async (tx) => { + await applyStatementBudget( + tx, + ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + ); + return tx.$queryRaw>` + WITH RECURSIVE pairs AS ( + SELECT u."id" AS user_id, t.type + FROM users u + CROSS JOIN unnest(enum_range(NULL::"measurement_type")) AS t(type) + WHERE ${activeAccountSql('u."id"')} + ), + days AS ( + SELECT p.user_id, p.type, f.at + FROM pairs p + CROSS JOIN LATERAL ( + SELECT min(m."measured_at") AS at + FROM measurements m + WHERE m."user_id" = p.user_id + AND m."type" = p.type + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${foldWindowStart} + ) f + WHERE f.at IS NOT NULL + UNION ALL + SELECT d.user_id, d.type, n.at + FROM days d + CROSS JOIN LATERAL ( + SELECT min(m."measured_at") AS at + FROM measurements m + WHERE m."user_id" = d.user_id + AND m."type" = d.type + AND m."deleted_at" IS NULL + AND m."measured_at" >= date_trunc('day', d.at) + interval '1 day' + ) n + WHERE n.at IS NOT NULL + ) + SELECT DISTINCT id FROM ( + SELECT d.user_id AS id + FROM days d + WHERE NOT EXISTS ( + SELECT 1 FROM measurement_rollups r + WHERE r."user_id" = d.user_id + AND r."type" = d.type + AND r."granularity" = 'DAY' + AND r."bucket_start" = date_trunc('day', d.at) + ) + + UNION + + SELECT r."user_id" AS id + FROM measurement_rollups r + WHERE r."granularity" = 'DAY' + AND r."bucket_start" >= ${foldWindowStart} + AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL) + AND ${activeAccountSql('r."user_id"')} + ) discovered + `; + }, + { + maxWait: 10_000, + timeout: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS + 30_000, + }, + ); + + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: RollupFullBackfillPayload = { + userId: id, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { + ...RETRY_OPTIONS, + expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, + // Coalesce: if a backfill for this user is already queued and + // we restart, pg-boss returns null instead of duplicating. + singletonKey: `boot-backfill|${id}`, + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index ad330a2e9..0bda33e66 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -34,9 +34,7 @@ * took. Nothing here reads both tiers; the rollup replaces the legacy read and * falls back to live SQL only on a coverage miss. */ -import { activeAccountSql } from "@/lib/account-deletion/active-account"; import { prisma } from "@/lib/db"; -import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { annotate } from "@/lib/logging/context"; import { @@ -113,39 +111,6 @@ export interface RollupRecomputePayload { enqueuedAt: string; } -/** - * v1.4.35.1 — boot-time full-fold queue. The write-path hooks keep - * the DAY bucket current per measurement, and `ensureUserRollupsFresh` - * warm-on-read covers the trailing 90-day DAY window on cold mount, - * but neither path folds the deep WEEK / MONTH / YEAR history for an - * account that has measurements pre-dating the v1.4.35 deploy. This - * queue exists so the worker boot can enqueue one full - * `recomputeUserRollups(userId)` per uncovered user — no operator - * action needed on self-hosted instances. - */ -export const ROLLUP_FULL_BACKFILL_QUEUE = "rollup-full-backfill"; - -/** Boot-fold worker concurrency. Serial so we don't crowd the pool. */ -export const ROLLUP_FULL_BACKFILL_CONCURRENCY = 1; - -/** - * The full fold's expiry: an hour. One fold is four granularities, each an - * aggregate held to the connection's statement timeout (60 s by default) and - * a persist transaction of up to 120 s, so the worst case under the defaults - * is close to thirteen minutes, too near pg-boss's fifteen-minute default to - * rely on, and an operator who raises `DATABASE_STATEMENT_TIMEOUT_MS` (or - * sets it to 0) lifts that bound altogether. Past the expiry pg-boss retries - * the job, and the retry folded the same account a second time. - */ -export const ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS = 60 * 60; - -/** Payload the boot-time backfill helper enqueues. */ -export interface RollupFullBackfillPayload { - userId: string; - /** Wall-clock kick-off for debugging only. */ - enqueuedAt: string; -} - /** * Trailing fold window for `recomputeUserRollups` defaults AND the * boot-time discovery scan. The two MUST share this bound: the fold @@ -639,7 +604,7 @@ export async function readRollupBuckets( * comprehensive-aggregator + summaries-slice convention. */ /** Give the statements of a transaction their own limit (SET LOCAL). */ -async function applyStatementBudget( +export async function applyStatementBudget( tx: Pick, ms: number, ): Promise { @@ -1316,165 +1281,6 @@ async function ensureUserRollupsFreshImpl( } } -/** - * v1.4.35.1 — boot-time enqueue helper. Finds users with measurements - * but zero rollup coverage and enqueues one full-fold job per account - * onto `ROLLUP_FULL_BACKFILL_QUEUE`. Designed to be called once at - * worker boot so every self-hosted instance auto-converges on the - * persistent rollup tier without operator intervention. - * - * Idempotent across reboots: - * - the discovery query matches accounts that either have zero - * rollup coverage at all OR have at least one measurement type - * whose DAY-rollup partition is empty. Once every type the user - * has logged has at least one DAY bucket, the user drops off the - * list and subsequent boots skip them. The v1.4.35.1 shape (zero- - * rollups only) silently stranded users who logged a brand-new - * type after the initial fold — their `isFullyCovered` probe stays - * `false`, every read fans out to the live SQL aggregator, and - * `/api/analytics` cold-mounts in tens of seconds instead of - * 1.5-3 s. v1.4.38.5 widens discovery to per-type missing coverage. - * - pg-boss `singletonKey` coalesces duplicate sends within the - * queue, so a fast restart while a backfill is queued doesn't - * double up - * - * Best-effort: errors are swallowed and reported through the return - * value so the worker boot never fails because of a backfill miss. - */ -export async function enqueueBootTimeRollupBackfill( - // Optional boot-storm stagger. When > 0 the per-user sends carry a - // `startAfter` delay (seconds) so this full-history fold does not drain in - // parallel with the other boot backfills onto one heavy tenant. Default 0 - // keeps immediate semantics for any non-boot caller. - startAfterSeconds: number = 0, -): Promise<{ - enqueued: number; - skipped: number; - error: string | null; -}> { - const boss = getGlobalBoss(); - if (!boss) { - return { enqueued: 0, skipped: 0, error: null }; - } - - try { - // Users where at least one (type, day) pair in `measurements` - // has no matching DAY rollup row. - // - // v1.4.39.1 — tightened from per-type to per-day. The legacy shape - // matched users where ANY type had zero rollup rows. That stranded - // accounts whose Withings sync / `/api/import` / admin restore - // wrote measurements without firing the rollup write hook: as soon - // as ONE rollup row existed for the type (even from a prior boot - // backfill), the LEFT JOIN found a match and the user dropped off - // the discovery list — leaving the dashboard chart's - // `source=rollup` fast-path under-counting the missing days. - // Switching the join to `(user, type, bucketStart)` surfaces every - // per-day gap. Cost: the inner DISTINCT widens from - // `(user_id, type)` (≈15 rows/user) to - // `(user_id, type, day)` (≈5 k rows/user on a 5-year power user), - // which is still bounded by the `(user_id, type, measured_at)` - // index path. The LEFT JOIN walks `measurement_rollups` against - // its composite primary key. - // - // v1.4.41 — the v1.4.39 W-SUM legacy-NULL `sum_value` UNION arm - // has been retired. The maintainer's tenant converged at v1.4.40, so the - // arm was a permanent no-op seq scan over `measurement_rollups`. - // Per-day missing coverage remains the sole discovery anchor. - // - // v1.16.10 — discovery is bounded to the same trailing window the - // fold writes (`ROLLUP_FOLD_WINDOW_MS`). Unbounded, any account - // with measurements older than the window was re-flagged on every - // boot: the fold can never produce buckets for those days, so the - // per-day gap re-surfaced forever and the full backfill re-ran on - // every restart. Days outside the window are not foldable by - // design and must not anchor discovery. - const foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); - // Two discovery arms, UNION-ed: - // - // 1. Per-day MISSING coverage — a (user, type, day) pair in - // `measurements` with no matching DAY rollup row (v1.4.39.1). - // - // 2. v1.20.0 F6 NULL-accumulator arm — a user whose DAY rollup rows - // predate migration 0190 (regression accumulators are NULL) needs a - // one-time re-fold so the windowed slope / r² / sd readers compose - // from the accumulators instead of falling back to the live REGR_* - // scan. Anchored on `sum_xy IS NULL` (the column the read tier - // treats as the coverage marker). New writes self-populate the - // accumulators, so this arm drains to empty once the boot re-fold - // catches up — mirrors the v1.4.39 sum_value rollout, which the - // v1.4.41 note confirms converged via this same boot queue. Bounded - // to the fold window so days outside it (un-foldable by design) - // never re-flag a user on every boot. The same arm refills the - // reading extremes of migration 0417 (`reading_min_value IS NULL`); - // the extremes reader treats a NULL as uncovered until it lands. - const users = await prisma.$queryRaw>` - SELECT DISTINCT id FROM ( - SELECT DISTINCT mt."user_id" AS id - FROM ( - SELECT DISTINCT - m."user_id", - m."type", - date_trunc('day', m."measured_at") AS bucket_start - FROM measurements m - WHERE m."deleted_at" IS NULL - AND m."measured_at" >= ${foldWindowStart} - ) mt - LEFT JOIN measurement_rollups r - ON r."user_id" = mt."user_id" - AND r."type" = mt."type" - AND r."granularity" = 'DAY' - AND r."bucket_start" = mt."bucket_start" - WHERE r."bucket_start" IS NULL - - UNION - - SELECT DISTINCT r."user_id" AS id - FROM measurement_rollups r - WHERE r."granularity" = 'DAY' - AND r."bucket_start" >= ${foldWindowStart} - AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL) - ) discovered - WHERE ${activeAccountSql('discovered."id"')} - `; - - if (users.length === 0) { - return { enqueued: 0, skipped: 0, error: null }; - } - - let enqueued = 0; - let skipped = 0; - for (const { id } of users) { - const payload: RollupFullBackfillPayload = { - userId: id, - enqueuedAt: new Date().toISOString(), - }; - const jobId = await boss.send(ROLLUP_FULL_BACKFILL_QUEUE, payload, { - expireInSeconds: ROLLUP_FULL_BACKFILL_EXPIRE_SECONDS, - retryLimit: 3, - retryDelay: 60, - retryBackoff: true, - // Coalesce: if a backfill for this user is already queued and - // we restart, pg-boss returns null instead of duplicating. - singletonKey: `boot-backfill|${id}`, - ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), - }); - if (jobId) { - enqueued += 1; - } else { - skipped += 1; - } - } - return { enqueued, skipped, error: null }; - } catch (err) { - return { - enqueued: 0, - skipped: 0, - error: err instanceof Error ? err.message : String(err), - }; - } -} - /** * Collapse a list of `(type, day)` keys to the set of distinct * `(type, dayStart)` pairs. Used by the batch-ingest hooks so a 500- diff --git a/tests/integration/hrv-rmssd-fitbit-google-migration.test.ts b/tests/integration/hrv-rmssd-fitbit-google-migration.test.ts index f31ce755c..cbbadf508 100644 --- a/tests/integration/hrv-rmssd-fitbit-google-migration.test.ts +++ b/tests/integration/hrv-rmssd-fitbit-google-migration.test.ts @@ -12,10 +12,8 @@ import { beforeAll, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import type { Prisma } from "@/generated/prisma/client"; -import { - enqueueBootTimeRollupBackfill, - recomputeUserRollups, -} from "@/lib/rollups/measurement-rollups"; +import { enqueueBootTimeRollupBackfill } from "@/lib/rollups/full-backfill"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index 2e382c9a5..c813fdebf 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -29,6 +29,8 @@ import { getPrismaClient, truncateAllTables } from "./setup"; import { ROLLUP_FULL_BACKFILL_QUEUE, enqueueBootTimeRollupBackfill, +} from "@/lib/rollups/full-backfill"; +import { recomputeBucketsForKeys, recomputeBucketsForMeasurement, recomputeUserRollups, diff --git a/tests/integration/rollup-full-backfill.test.ts b/tests/integration/rollup-full-backfill.test.ts new file mode 100644 index 000000000..01f6fbbd6 --- /dev/null +++ b/tests/integration/rollup-full-backfill.test.ts @@ -0,0 +1,306 @@ +/** + * The full backfill against Postgres (#1298). + * + * - The windowed fold (`foldUserRollupHistory`, a UTC calendar year per + * pass, newest first) writes exactly the rollup rows one pass of + * `recomputeUserRollups` over the whole five-year window writes: every + * granularity, every source, a week across New Year, the YEAR bucket the + * window's oldest edge snaps out to, and a reading older than the window. + * - A fold stopped part way and resumed with `before` ends in the same rows. + * - The boot discovery, rewritten to walk days instead of reading every + * 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 { Prisma, type MeasurementSource } from "@/generated/prisma/client"; +import { activeAccountSql } from "@/lib/account-deletion/active-account"; +import { + enqueueBootTimeRollupBackfill, + foldUserRollupHistory, + ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, +} from "@/lib/rollups/full-backfill"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; + +import { makeUser } from "../kit/factories"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; + +const prisma = getPrismaClient(); +const DAY_MS = 86_400_000; +const HOUR_MS = 3_600_000; + +beforeEach(async () => { + await truncateAllTables(prisma); + vi.mocked(getGlobalBoss).mockReturnValue(null as never); +}); + +function utc(y: number, m: number, d: number, h = 12, min = 0): Date { + return new Date(Date.UTC(y, m - 1, d, h, min)); +} + +/** Readings that can tell a windowed fold from one pass. */ +async function seedHistory(userId: string): Promise { + const now = new Date(); + const year = now.getUTCFullYear(); + const floor = new Date(now.getTime() - ROLLUP_FOLD_WINDOW_MS); + const rows: Prisma.MeasurementCreateManyInput[] = []; + const add = ( + type: "WEIGHT" | "PULSE", + source: MeasurementSource, + at: Date, + value: number, + ) => + rows.push({ + userId, + type, + source, + value, + unit: type === "WEIGHT" ? "kg" : "bpm", + measuredAt: at, + externalId: `${source}:${type}:${at.toISOString()}`, + }); + const days: Date[] = []; + // A week across each of two New Years. + for (let d = 0; d < 9; d++) { + days.push(new Date(utc(year - 3, 12, 27).getTime() + d * DAY_MS)); + days.push(new Date(utc(year - 1, 12, 28).getTime() + d * DAY_MS)); + } + days.push(utc(year - 2, 6, 15), utc(year - 2, 6, 16)); + days.push(new Date(now.getTime() - 3 * DAY_MS)); + // Just inside the window, and inside the YEAR bucket its edge snaps to. + days.push(new Date(floor.getTime() + 2 * DAY_MS)); + days.push(new Date(floor.getTime() - 20 * DAY_MS)); + days.forEach((day, i) => { + const base = Date.UTC( + day.getUTCFullYear(), + day.getUTCMonth(), + day.getUTCDate(), + ); + add("WEIGHT", "MANUAL", new Date(base + 7 * HOUR_MS), 80 + (i % 5) / 3); + if (i % 2 === 0) { + add("WEIGHT", "WITHINGS", new Date(base + 8 * HOUR_MS), 81 + i / 7); + } + for (let k = 0; k < 6; k++) { + add( + "PULSE", + "APPLE_HEALTH", + new Date(base + (k < 4 ? 9 : 15) * HOUR_MS + k * 7 * 60_000), + 55 + ((i * 7 + k * 11) % 40), + ); + } + add("PULSE", "GOOGLE_HEALTH", new Date(base + 23 * HOUR_MS), 60 + i); + }); + await prisma.measurement.createMany({ data: rows }); +} + +async function snapshot(userId: string) { + const rows = await prisma.measurementRollup.findMany({ + where: { userId }, + orderBy: [ + { type: "asc" }, + { granularity: "asc" }, + { bucketStart: "asc" }, + { source: "asc" }, + ], + }); + return rows.map(({ computedAt, ...rest }) => { + void computedAt; + return rest; + }); +} + +function expectSameRows( + actual: Awaited>, + expected: Awaited>, +): void { + expect(actual.length).toBe(expected.length); + expected.forEach((want, i) => { + const got = actual[i] as Record; + for (const [key, value] of Object.entries(want)) { + if (typeof value === "number" && key !== "count") { + // A week across two windows is read in another range, so its sums + // may be added in another order: equal to rounding, not to the bit. + expect(got[key] as number).toBeCloseTo(value, 6); + } else { + expect(got[key]).toEqual(value); + } + } + }); +} + +describe("foldUserRollupHistory", () => { + it("writes the rows one pass over the whole window writes", async () => { + const user = await makeUser(); + await seedHistory(user.id); + + await recomputeUserRollups(user.id); + const onePass = await snapshot(user.id); + expect(onePass.length).toBeGreaterThan(50); + expect(onePass.some((r) => r.granularity === "WEEK")).toBe(true); + + await prisma.measurementRollup.deleteMany({ where: { userId: user.id } }); + const done = await foldUserRollupHistory(user.id, { + statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + }); + expect(done.windowsFolded).toBeGreaterThanOrEqual(5); + expect(done.resumeBefore).toBeUndefined(); + expectSameRows(await snapshot(user.id), onePass); + expect( + (await snapshot(user.id)).every( + (r) => r.readingMinValue !== null && r.readingMaxValue !== null, + ), + ).toBe(true); + }); + + it("ends in the same rows when stopped part way and resumed", async () => { + const user = await makeUser(); + await seedHistory(user.id); + await recomputeUserRollups(user.id); + const onePass = await snapshot(user.id); + await prisma.measurementRollup.deleteMany({ where: { userId: user.id } }); + + // Stop on the seventh question: inside the second window. + let asked = 0; + const first = await foldUserRollupHistory(user.id, { + shouldStop: () => ++asked >= 7, + }); + expect(first.windowsFolded).toBe(1); + expect(first.resumeBefore).toBeDefined(); + + const rest = await foldUserRollupHistory(user.id, { + before: first.resumeBefore, + }); + expect(rest.resumeBefore).toBeUndefined(); + expectSameRows(await snapshot(user.id), onePass); + }); +}); + +/** The discovery statement before the day walk, verbatim. */ +async function referenceDiscovery(): Promise { + const foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const users = await prisma.$queryRaw>` + SELECT DISTINCT id FROM ( + SELECT DISTINCT mt."user_id" AS id + FROM ( + SELECT DISTINCT + m."user_id", + m."type", + date_trunc('day', m."measured_at") AS bucket_start + FROM measurements m + WHERE m."deleted_at" IS NULL + AND m."measured_at" >= ${foldWindowStart} + ) mt + LEFT JOIN measurement_rollups r + ON r."user_id" = mt."user_id" + AND r."type" = mt."type" + AND r."granularity" = 'DAY' + AND r."bucket_start" = mt."bucket_start" + WHERE r."bucket_start" IS NULL + + UNION + + SELECT DISTINCT r."user_id" AS id + FROM measurement_rollups r + WHERE r."granularity" = 'DAY' + AND r."bucket_start" >= ${foldWindowStart} + AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL) + ) discovered + WHERE ${activeAccountSql('discovered."id"')} + `; + return users.map((u) => u.id).sort(); +} + +describe("enqueueBootTimeRollupBackfill", () => { + it("finds the accounts the statement it replaced found", async () => { + const covered = await makeUser(); + const missingDay = await makeUser(); + const nullExtremes = await makeUser(); + const tombstoneOnly = await makeUser(); + const tooOld = await makeUser(); + const pendingDeletion = await makeUser(); + for (const u of [covered, missingDay, nullExtremes, pendingDeletion]) { + await seedHistory(u.id); + await recomputeUserRollups(u.id); + } + // A reading on a day with no DAY bucket. + await prisma.measurement.create({ + data: { + userId: missingDay.id, + type: "BODY_FAT", + value: 20, + unit: "%", + source: "MANUAL", + measuredAt: new Date(Date.now() - 40 * DAY_MS), + }, + }); + await prisma.measurement.create({ + data: { + userId: pendingDeletion.id, + type: "BODY_FAT", + value: 20, + unit: "%", + source: "MANUAL", + measuredAt: new Date(Date.now() - 40 * DAY_MS), + }, + }); + await prisma.user.update({ + where: { id: pendingDeletion.id }, + data: { deletionRequestedAt: new Date() }, + }); + // A row folded before 0417 added the reading extremes. + await prisma.$executeRaw` + UPDATE measurement_rollups SET reading_min_value = NULL + WHERE user_id = ${nullExtremes.id} AND granularity = 'DAY' + AND bucket_start = ( + SELECT max(bucket_start) FROM measurement_rollups + WHERE user_id = ${nullExtremes.id} AND granularity = 'DAY' + ) + `; + // Only a tombstone inside the window: nothing to fold. + await prisma.measurement.create({ + data: { + userId: tombstoneOnly.id, + type: "WEIGHT", + value: 70, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(Date.now() - 10 * DAY_MS), + deletedAt: new Date(), + }, + }); + // Only readings older than the window: the fold writes nothing there. + await prisma.measurement.create({ + data: { + userId: tooOld.id, + type: "WEIGHT", + value: 70, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS - 5 * DAY_MS), + }, + }); + + const send = vi.fn(async () => "job-id"); + vi.mocked(getGlobalBoss).mockReturnValue({ send } as never); + const result = await enqueueBootTimeRollupBackfill(); + expect(result.error).toBeNull(); + const found = ( + send.mock.calls as unknown as Array<[string, { userId: string }]> + ) + .map((call) => call[1].userId) + .sort(); + + expect(found).toEqual([missingDay.id, nullExtremes.id].sort()); + expect(found).toEqual(await referenceDiscovery()); + }); +});