From 17d9c15dbcd993f96e7adacd429e48f6161b5d95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 10:34:01 +0100 Subject: [PATCH 1/9] fix(measurements): keep a folded hour out of a blood-pressure reading The reading delete found the cuff's pulse by owner, source and identical instant. The dense fold anchors an hour of heart rate at the local HH:30 under the samples' source, so a reading taken at half past took that hour with it. Aggregate (stats:) rows are never part of a reading now, for the delete and for the restore. --- .../measurements/blood-pressure-reading.ts | 25 ++++++++ .../blood-pressure-reading-delete.test.ts | 59 ++++++++++++++++++- 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/src/lib/measurements/blood-pressure-reading.ts b/src/lib/measurements/blood-pressure-reading.ts index 589a7fb66..717cf508c 100644 --- a/src/lib/measurements/blood-pressure-reading.ts +++ b/src/lib/measurements/blood-pressure-reading.ts @@ -14,6 +14,14 @@ * Deleting a systolic or a diastolic takes the whole reading with it. * Deleting a pulse takes only the pulse: a pulse is also a value of its own, * and removing one is no reason to remove a blood pressure. + * + * An aggregate row is never part of a reading. The dense fold stores an + * hour of heart rate as one `stats:` row at the local HH:30 of that hour + * (`canonicalHourlyTimestamp`), under the same source as the samples it + * folded, and the client's hourly heart-rate bucket is a `stats:` row as + * well. A cuff reading taken at half past, from the same source, shares its + * instant with that row; it is still an hour of somebody's heart rate, not + * the cuff's pulse, and a delete must not take it. */ import type { MeasurementType, Prisma } from "@/generated/prisma/client"; @@ -30,6 +38,9 @@ const BP_READING_TYPES: MeasurementType[] = [ "PULSE", ]; +/** The externalId prefix every aggregate (folded or bucketed) row carries. */ +const AGGREGATE_EXTERNAL_ID_PREFIX = "stats:"; + export interface ReadingAnchor { id: string; type: MeasurementType | string; @@ -70,6 +81,20 @@ export async function bloodPressureSiblings( userId, id: { notIn: rows.map((r) => r.id) }, type: { in: BP_READING_TYPES }, + // Spelled with the NULL arm: a bare `NOT startsWith` is NULL, not + // true, for the NULL externalId of a manual row and would drop it. + AND: [ + { + OR: [ + { externalId: null }, + { + NOT: { + externalId: { startsWith: AGGREGATE_EXTERNAL_ID_PREFIX }, + }, + }, + ], + }, + ], OR: anchors.map((a) => ({ source: a.source as Prisma.MeasurementWhereInput["source"], measuredAt: a.measuredAt, diff --git a/tests/integration/blood-pressure-reading-delete.test.ts b/tests/integration/blood-pressure-reading-delete.test.ts index d7f4028d9..2e930187d 100644 --- a/tests/integration/blood-pressure-reading-delete.test.ts +++ b/tests/integration/blood-pressure-reading-delete.test.ts @@ -10,11 +10,16 @@ * its own syncVersion (so the sync feed carries each tombstone), and the * undo, which names only the row the person tapped, brings the whole * reading back. A reading of another source at the same instant, a reading - * a minute later, and a pulse deleted on its own are left alone. + * a minute later, and a pulse deleted on its own are left alone. So is the + * folded hour of heart rate the dense fold anchors at the local HH:30: a + * reading taken at half past shares its instant and its source, and is + * still not the same thing. */ import { NextRequest } from "next/server"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import { canonicalHourlyTimestamp } from "@/lib/measurements/consolidation-tz"; + import { cookieJar, headerJar } from "./mock-next-headers"; import { getPrismaClient, truncateAllTables } from "./setup"; @@ -195,4 +200,56 @@ describe("deleting a blood-pressure reading", () => { // The pulse deleted at another moment stays deleted. expect(await deletedIds()).toEqual(["next-pulse"]); }); + + it("BP at :30 beside a folded hour: leaves the hour of heart rate alone", async () => { + const prisma = getPrismaClient(); + // 09:30 local (Europe/Berlin, the default zone) on a day the fold took: + // the anchor of that hour's mean. + const halfPast = canonicalHourlyTimestamp("2026-10-20", 9, "Europe/Berlin"); + await prisma.measurement.createMany({ + data: [ + { + id: "ah-sys", + userId: USER, + type: "BLOOD_PRESSURE_SYS", + value: 134, + unit: "mmHg", + source: "APPLE_HEALTH", + measuredAt: halfPast, + externalId: "uuid-ah-sys", + }, + { + id: "ah-dia", + userId: USER, + type: "BLOOD_PRESSURE_DIA", + value: 86, + unit: "mmHg", + source: "APPLE_HEALTH", + measuredAt: halfPast, + externalId: "uuid-ah-dia", + }, + { + id: "folded-hour", + userId: USER, + type: "PULSE", + value: 71, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: halfPast, + externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-10-20T09", + }, + ], + }); + + await deleteOne("ah-sys"); + expect(await deletedIds()).toEqual(["ah-dia", "ah-sys"]); + expect((await rows())["folded-hour"]!.syncVersion).toBe(1); + + const { restored } = await post("restore", ["ah-sys"]); + expect(restored).toBe(2); + + const { deleted } = await post("bulk-delete", ["ah-dia"]); + expect(deleted).toBe(2); + expect(await deletedIds()).toEqual(["ah-dia", "ah-sys"]); + }); }); From ec2ec400033ec5a4d9fee31c14bf2c02424b8c71 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 10:52:54 +0100 Subject: [PATCH 2/9] fix(measurements): report folded provider heart rate to the change feed The dense fold deleted Google Health and Oura samples outright, on the reasoning that no client pulls them. The change feed serves every source, so a client mirroring it kept those samples next to the hourly means with nothing to tell it they were gone. The nightly increment of a server-pulled source is tombstoned again: the feed reports each sample and the tombstone cleanup purges them after the retention. A backlog day, such as an account's first fold, is still deleted outright, and the account's delta cursors expire instead, since tombstoning millions of rows would undo the point of the fold. The compaction-tombstone class leaves these sources out, so the feed, the backlog purge and the restore treat their tombstones like any deletion. Apple Health and Health Connect are unchanged. --- .../baselines/source-file-lines.json | 4 +- .../baselines/source-function-lines.json | 8 +- .../baselines/version-stamp-comments.json | 4 +- .../measurement-write-tail-guard.test.ts | 2 + src/lib/jobs/dense-intraday-retention.ts | 3 + src/lib/measurements/dense-fold-removal.ts | 89 +++++++++++ .../measurements/dense-intraday-retention.ts | 42 ++++-- .../measurements/dense-intraday-sources.ts | 15 ++ src/lib/measurements/folded-window.ts | 6 + tests/integration/provider-dense-fold.test.ts | 139 ++++++++++++++++++ 10 files changed, 289 insertions(+), 23 deletions(-) create mode 100644 src/lib/measurements/dense-fold-removal.ts diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 31b681f6f..4f04a28da 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": 94686, + "total": 94692, "entries": { "app/medications/page-client.tsx": 839, "components/charts/mood-chart.tsx": 1166, @@ -71,7 +71,7 @@ "lib/managed-profiles/handover.ts": 986, "lib/mcp/record-search.ts": 1467, "lib/mcp/rich-reads.ts": 1467, - "lib/measurements/dense-intraday-retention.ts": 1010, + "lib/measurements/dense-intraday-retention.ts": 1016, "lib/measurements/import-apple-health-export.ts": 1722, "lib/medications/scheduling/cadence.ts": 976, "lib/medications/scheduling/recurrence.ts": 1100, diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index e43f8a5c3..05a8b99d1 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 346, - "total": 112215, + "total": 112219, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 286, @@ -319,9 +319,9 @@ "lib/labs/biomarker-merge.ts :: runMerge": 254, "lib/measurements/consolidate-daily-mean.ts :: consolidateDailyMean": 278, "lib/measurements/consolidate-legacy-steps.ts :: consolidateLegacySteps": 263, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 537, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 447, - "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 373, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention": 539, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource": 448, + "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay": 374, "lib/measurements/dense-intraday-retention.ts :: runDenseIntradayRetention > walkSource > writeDay > foldOnce": 212, "lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative": 344, "lib/measurements/drain-per-sample-cumulative.ts :: drainPerSampleCumulative > writeDay": 205, diff --git a/src/__tests__/baselines/version-stamp-comments.json b/src/__tests__/baselines/version-stamp-comments.json index c1d8207df..2e5febe0e 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": 2896, - "total": 9229, + "total": 9228, "entries": { "__tests__/account-payload-consumer-guard.test.ts": 2, "__tests__/acting-account-boundary-guard.test.ts": 7, @@ -2458,7 +2458,7 @@ "lib/measurements/cumulative-day-sum.ts": 6, "lib/measurements/daily-series-read.ts": 2, "lib/measurements/dense-intraday-hourly-rebuild.ts": 1, - "lib/measurements/dense-intraday-retention.ts": 4, + "lib/measurements/dense-intraday-retention.ts": 3, "lib/measurements/display-transform.ts": 6, "lib/measurements/drain-per-sample-cumulative.ts": 8, "lib/measurements/import-apple-health-export.ts": 5, diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 66e358b74..3d3d11f24 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -110,6 +110,8 @@ const WRITE_SITES: Record = { "value-preserving hourly fold; refolds the touched spans", "lib/measurements/dense-intraday-retention.ts": "value-preserving retention fold; refolds the touched days", + "lib/measurements/dense-fold-removal.ts": + "removes the samples the retention fold has just taken into its hourly means, inside the fold's transaction; the fold refolds the touched days", "lib/jobs/sleep-timeline-backfill.ts": "one-shot re-stamp of stored sleep rows onto their timeline; refolds the touched days", "lib/jobs/step-consolidation-repair.ts": diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index 6b679fbfe..16891f3a5 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -123,6 +123,9 @@ export async function runDenseIntradayRetentionForUser( // days for proxy users, preserving the resting signal post-fold. derived_resting_rows_upserted: summary.totals.derivedRestingRowsUpserted, + // Backlog days of a server-pulled source that expired the account's + // delta cursors rather than leave millions of tombstones. + sync_resets_stamped: summary.totals.syncResetsStamped, stopped_early: summary.stoppedEarly, }, }, diff --git a/src/lib/measurements/dense-fold-removal.ts b/src/lib/measurements/dense-fold-removal.ts new file mode 100644 index 000000000..93eae7254 --- /dev/null +++ b/src/lib/measurements/dense-fold-removal.ts @@ -0,0 +1,89 @@ +/** + * How the dense fold (`dense-intraday-retention.ts`) removes the raw samples + * of a day it folded, and how the change feed learns of it. + * + * Apple Health and Health Connect samples are deleted outright (v1.42), like + * the cumulative drain's rows, where they used to be tombstoned for the + * 75-day retention. They come from the person's own health store and no + * client reads them from `/api/sync/changes`, so the tombstone told nobody + * anything; it only held the re-upload of a folded sample off, and the + * `folded_window` ingest guard (`folded-window.ts`) does that now from the + * live hourly row. The tombstones made up most of the table and its indexes. + * + * A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`: Google Health, Oura) + * is different. The feed served its samples, and a client mirroring the feed + * keeps them until the feed says they are gone. So: + * - the nightly increment, a day at most {@link FOLD_FEED_TOMBSTONE_DAYS} + * past the retention boundary, is tombstoned. The feed reports every + * sample, and the 75-day tombstone cleanup purges them later. The + * classifier of compaction tombstones leaves these sources out, so the + * feed, the backlog purge and the restore treat them like any deletion. + * - an older day is backlog: an account's first fold, an old range a + * backfill brought in, nights the fold did not run. Tombstoning that + * would put millions of rows back into the table (#1297), so its samples + * are deleted outright and the account's delta cursors expire instead + * (`stampSyncReset`). A client that pulled them re-initialises rather + * than keep them beside the hourly means. + * The tombstones share one deletion instant and keep their `syncVersion`, so + * the re-fold reads them as fold leftovers (`fold-constituents.ts`), exactly + * like the Apple Health tombstones of the releases before v1.42. + */ +import type { MeasurementSource, Prisma } from "@/generated/prisma/client"; +import { stampSyncReset } from "@/lib/sync/reset"; + +import { SERVER_PULLED_FOLD_SOURCES } from "./dense-intraday-sources"; +import { absorbIntoFold } from "./fold-constituents"; + +/** Days past the retention boundary that still count as the nightly fold. */ +const FOLD_FEED_TOMBSTONE_DAYS = 7; + +const DAY_MS = 86_400_000; + +export type FoldRemoval = "delete" | "tombstone" | "delete-and-reset"; + +/** How a fold removes the samples of one local day. Pure. */ +export function foldRemovalFor( + source: MeasurementSource, + dayStart: Date, + retentionDays: number, + now: Date = new Date(), +): FoldRemoval { + if (!SERVER_PULLED_FOLD_SOURCES.has(source)) return "delete"; + const nightly = + now.getTime() - (retentionDays + FOLD_FEED_TOMBSTONE_DAYS) * DAY_MS; + return dayStart.getTime() >= nightly ? "tombstone" : "delete-and-reset"; +} + +/** + * Remove the folded samples `ids`, never one of `keep` (the hourly rows the + * fold just adopted), inside the fold's transaction. Returns how many went + * and whether the account's delta cursors were expired. The caller stamps + * the reset once more after the commit, so a cursor issued while the + * transaction ran expires too; the stamp in here survives a crash after it. + */ +export async function removeFoldedSamples( + tx: Prisma.TransactionClient, + input: { + userId: string; + removal: FoldRemoval; + ids: readonly string[]; + keep: readonly string[]; + }, +): Promise<{ removed: number; resetFeed: boolean }> { + if (input.removal === "tombstone") { + const removed = await absorbIntoFold( + tx, + input.ids.filter((id) => !input.keep.includes(id)), + ); + return { removed, resetFeed: false }; + } + const del = await tx.measurement.deleteMany({ + where: { + id: { in: [...input.ids], notIn: [...input.keep] }, + deletedAt: null, + }, + }); + const resetFeed = input.removal === "delete-and-reset" && del.count > 0; + if (resetFeed) await stampSyncReset(tx, input.userId); + return { removed: del.count, resetFeed }; +} diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index e75cd7546..4e4234201 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -86,6 +86,8 @@ import { DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, type DenseIntradayFoldSource, } from "./dense-intraday-sources"; +import { foldRemovalFor, removeFoldedSamples } from "./dense-fold-removal"; +import { stampSyncReset } from "@/lib/sync/reset"; import { absorbIntoFold, foldConstituentHorizon, @@ -433,6 +435,12 @@ export interface DenseIntradayRetentionSummary { * resting signal the raw-PULSE proxy can no longer compute after fold. */ derivedRestingRowsUpserted: number; + /** + * Folded days of a server-pulled source whose samples were deleted + * outright, each of which expired the account's delta cursors + * (`dense-fold-removal.ts`). + */ + syncResetsStamped: number; /** * Hours that already had a live hourly row whose samples could not be * shown to be complete (`fold-constituents.ts`): the stored mean and the @@ -458,7 +466,9 @@ export interface DenseIntradayRetentionOptions extends ConsolidationOptions { * Run the dense intra-day retention drain. Folds per-sample dense-tier * rows older than the retention window into hourly-mean `stats:` rows * (user-local hours, anchored at local HH:30) and deletes the raw rows - * (soft-deleted before v1.42); rows inside the window stay raw. Idempotent — + * (soft-deleted before v1.42; a server-pulled source's nightly increment is + * still tombstoned, see `dense-fold-removal.ts`); rows inside the window + * stay raw. Idempotent — * re-invocation after a successful pass converges to zero work because the * folded rows are gone from the live scan, and the minted hourly * rows are excluded by the `NOT startsWith('stats:')` predicate. @@ -487,6 +497,7 @@ export async function runDenseIntradayRetention( hourlyRowsUpserted: 0, dailyRowsRetired: 0, derivedRestingRowsUpserted: 0, + syncResetsStamped: 0, hoursLeftAsStored: 0, }, }; @@ -647,8 +658,10 @@ export async function runDenseIntradayRetention( hoursLeftAsStored: number; /** Every raw sample of the day, when that is provably all of it. */ wholeDay: FoldConstituent[] | null; + resetFeed: boolean; }> => { let removed = 0; + let resetFeed = false; let retiredDaily = false; let slotsWritten = 0; let hoursLeftAsStored = 0; @@ -806,24 +819,20 @@ export async function runDenseIntradayRetention( // anchor is the row just adopted as that hour's mean, so removing // it would erase the fold. // - // v1.42 — deleted outright, like the cumulative drain's rows, - // where they used to be tombstoned for the 75-day retention. No - // client ever pulled these rows, so the tombstone told nobody - // anything; it only held the re-upload of a folded sample off, - // and the `folded_window` ingest guard (`folded-window.ts`) does - // that now from the live hourly row. The tombstones made up most - // of the table and its indexes. - const del = await tx.measurement.deleteMany({ - where: { - id: { in: deleteIds, notIn: canonicalRowIds }, - deletedAt: null, - }, + // How, and whether the change feed hears of it, depends on the + // source and the day's age: `dense-fold-removal.ts`. + const dropped = await removeFoldedSamples(tx, { + userId, + removal: foldRemovalFor(source, dayStart, retentionDays), + ids: deleteIds, + keep: canonicalRowIds, }); + resetFeed = dropped.resetFeed; const absorbed = await absorbIntoFold( tx, absorbIds.filter((id) => !canonicalRowIds.includes(id)), ); - removed = del.count + absorbed; + removed = dropped.removed + absorbed; // The day's raw samples, for the derived resting figure: the rows // in hand for a day folded here for the first time; for a day @@ -844,12 +853,14 @@ export async function runDenseIntradayRetention( ]; } }, FOLD_TRANSACTION_OPTIONS); + if (resetFeed) await stampSyncReset(pc, userId); return { removed, retiredDaily, slotsWritten, hoursLeftAsStored, wholeDay, + resetFeed, }; }; @@ -868,6 +879,7 @@ export async function runDenseIntradayRetention( summary.totals.hourlyRowsUpserted += foldResult.slotsWritten; summary.totals.hoursLeftAsStored += foldResult.hoursLeftAsStored; if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; + if (foldResult.resetFeed) summary.totals.syncResetsStamped += 1; // iOS#34 — preserve the resting signal for PROXY users. For a user // with zero native `RESTING_HEART_RATE` rows, the read-path resolver @@ -997,7 +1009,7 @@ export async function runDenseIntradayRetention( } log( - `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`, + `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted} syncResetsStamped=${summary.totals.syncResetsStamped} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/src/lib/measurements/dense-intraday-sources.ts b/src/lib/measurements/dense-intraday-sources.ts index 3c1891e54..08c01a69c 100644 --- a/src/lib/measurements/dense-intraday-sources.ts +++ b/src/lib/measurements/dense-intraday-sources.ts @@ -84,3 +84,18 @@ export type DenseIntradayFoldSource = export const DENSE_INTRADAY_FOLD_SOURCES = Object.keys( DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, ) as readonly DenseIntradayFoldSource[]; + +/** + * The fold sources whose rows the server pulled from a provider itself. A + * client can hold one of these rows only because `/api/sync/changes` served + * it, so when the fold removes them the feed has to say so: the nightly fold + * leaves tombstones for them, and a fold of an older backlog expires the + * account's delta cursors instead (`dense-intraday-retention.ts`). + * + * Apple Health and Health Connect samples come from the person's own health + * store, uploaded by the device or imported from its export, and the fold has + * deleted them outright since v1.42. No client reads the feed for them: the + * iOS app reads its own HealthKit store and has no feed consumer. + */ +export const SERVER_PULLED_FOLD_SOURCES: ReadonlySet = + new Set(["GOOGLE_HEALTH", "OURA"]); diff --git a/src/lib/measurements/folded-window.ts b/src/lib/measurements/folded-window.ts index eb84dfb2c..9990632c2 100644 --- a/src/lib/measurements/folded-window.ts +++ b/src/lib/measurements/folded-window.ts @@ -79,6 +79,7 @@ import { import { DENSE_INTRADAY_FOLD_SOURCES, DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE, + SERVER_PULLED_FOLD_SOURCES, DENSE_INTRADAY_RETENTION_TYPES, } from "@/lib/measurements/dense-intraday-sources"; @@ -475,12 +476,17 @@ export interface TombstoneCandidateRow extends FoldedWindowCandidateRow { * enough: an older release folded a day in two runs, and a person who * deleted a sample of such a half-folded day deleted something the fold had * not taken. That deletion is class B. + * + * A server-pulled source (`SERVER_PULLED_FOLD_SOURCES`) has no class A. The + * fold tombstones its samples so the change feed reports them to a client + * that pulled them, which is exactly what this class is kept out of. */ export function compactionTombstoneCoverId( row: TombstoneCandidateRow, tz: string, ): string | null { if (row.deletedAt === null) return null; + if (SERVER_PULLED_FOLD_SOURCES.has(row.source)) return null; if (!isFoldedWindowCandidate(row, row.deletedAt)) return null; if (!isWindowFoldedAt(row, row.deletedAt, tz)) return null; return coveringStatsExternalId(row, tz); diff --git a/tests/integration/provider-dense-fold.test.ts b/tests/integration/provider-dense-fold.test.ts index d36c21a52..d25d0a9f6 100644 --- a/tests/integration/provider-dense-fold.test.ts +++ b/tests/integration/provider-dense-fold.test.ts @@ -11,11 +11,16 @@ * hour whose anchor a sample fell on (Google Health's natural-key rescue * used to re-key that row back into the sample). * - Samples inside the window still write as before. + * - The change feed served these samples, so their removal reaches it: the + * nightly increment leaves tombstones the feed reports, and a backlog + * day, deleted outright, expires the account's delta cursors. An Apple + * Health fold expires nothing. */ import { beforeEach, describe, expect, it, vi } from "vitest"; import { upsertGoogleHealthMeasurements } from "@/lib/google-health/sync-core"; import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; +import { pullSyncChanges } from "@/lib/sync/changes/pull"; import { TouchedDays, writeOuraHeartRatePage, @@ -182,4 +187,138 @@ describe("dense fold of provider heart rate (#1297)", () => { recent.length, ); }); + + describe("and the change feed", () => { + /** Drain the feed from the start; the last page's cursor. */ + async function caughtUpCursor(userId: string): Promise { + let token: string | undefined; + for (;;) { + const page = await pullSyncChanges({ + userId, + token, + limit: 500, + serverNow: new Date(), + }); + expect(page.cursorExpired).toBe(false); + token = page.cursor!; + if (!page.hasMore) return token; + } + } + + async function pullFrom(userId: string, token: string) { + return pullSyncChanges({ + userId, + token, + limit: 500, + serverNow: new Date(), + }); + } + + async function syncResetAt(userId: string) { + const user = await prisma.user.findUniqueOrThrow({ + where: { id: userId }, + select: { syncResetAt: true }, + }); + return user.syncResetAt; + } + + it("reports the nightly increment as tombstones and keeps the cursor", async () => { + const user = await makeUser({ timezone: "UTC" }); + // Three days past the window: the day the nightly fold reaches next. + const day = samples(dayStart(93)).map((s) => ({ + bpm: s.bpm, + timestamp: s.at.toISOString(), + source: "awake", + })); + await writeOuraHeartRatePage(user.id, day, new TouchedDays(), { + emitArrivals: false, + }); + const raw = await prisma.measurement.findMany({ + where: { userId: user.id, source: "OURA" }, + select: { id: true }, + }); + expect(raw).toHaveLength(day.length); + const cursor = await caughtUpCursor(user.id); + + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.syncResetsStamped).toBe(0); + expect(await syncResetAt(user.id)).toBeNull(); + expect( + (await liveRows(user.id, "OURA")).every((r) => + r.externalId?.startsWith("stats:"), + ), + ).toBe(true); + + const page = await pullFrom(user.id, cursor); + expect(page.cursorExpired).toBe(false); + const { tombstones } = page.changes.measurements; + // The derived resting figure rides along under COMPUTED. + const upserts = page.changes.measurements.upserts.filter( + (u) => u.source === "OURA", + ); + // The samples on 10:30 and 11:30 became the two hourly means; every + // other sample arrives as a tombstone, so the client's mirror ends up + // holding exactly the hourly means. + expect(upserts).toHaveLength(2); + expect(upserts.every((u) => u.externalId?.startsWith("stats:"))).toBe( + true, + ); + expect(tombstones).toHaveLength(day.length - 2); + expect( + [...tombstones.map((t) => t.id), ...upserts.map((u) => u.id)].sort(), + ).toEqual(raw.map((r) => r.id).sort()); + }); + + it("expires the cursor when a backlog day is deleted outright", async () => { + const user = await makeUser({ timezone: "UTC" }); + await upsertGoogleHealthMeasurements( + user.id, + samples(dayStart(120)).map(gh), + ); + const cursor = await caughtUpCursor(user.id); + + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.syncResetsStamped).toBe(1); + expect(await syncResetAt(user.id)).not.toBeNull(); + // No tombstone is left behind for the backlog. + expect( + await prisma.measurement.count({ + where: { userId: user.id, deletedAt: { not: null } }, + }), + ).toBe(0); + expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(true); + // A client that re-initialises catches up from a fresh cursor. + await caughtUpCursor(user.id); + }); + + it("expires nothing for an Apple Health fold", async () => { + const user = await makeUser({ timezone: "UTC" }); + await prisma.measurement.createMany({ + data: samples(dayStart(120)).map((s) => ({ + userId: user.id, + type: "PULSE" as const, + value: s.bpm, + unit: "bpm", + source: "APPLE_HEALTH" as const, + measuredAt: s.at, + externalId: `uuid-${s.at.getTime()}`, + })), + }); + const cursor = await caughtUpCursor(user.id); + const summary = await runDenseIntradayRetention(prisma, { + userId: user.id, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.syncResetsStamped).toBe(0); + expect(await syncResetAt(user.id)).toBeNull(); + expect((await pullFrom(user.id, cursor)).cursorExpired).toBe(false); + }); + }); }); From a6f4c386fedeab9bff9928628905f00232a2af4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 10:58:04 +0100 Subject: [PATCH 3/9] fix(jobs): judge a system job's failure streak per job, per account The alert added every attempt of a failed job to the streak, so one job with two retries read as three failures. It also judged a queue by its newest job across all accounts: on a queue that runs one job per account, one account failing every hour beside accounts that succeed never reached the streak, or flipped between failing and recovered. The streak now counts jobs, and is kept per account (the job's userId, or the whole queue for jobs without one). A queue is failing while any account has failed three jobs in a row, and recovered only when a failing account has itself run successfully since. A reminder needs a failure newer than the notice before it, so a one-off job that failed and does not run again is announced once. --- docs/self-hosting/scaling.md | 6 +- .../jobs/__tests__/system-job-alert.test.ts | 144 ++++++++++---- src/lib/jobs/system-job-alert.ts | 185 ++++++++++++------ tests/integration/system-job-alert.test.ts | 99 ++++++++-- 4 files changed, 323 insertions(+), 111 deletions(-) diff --git a/docs/self-hosting/scaling.md b/docs/self-hosting/scaling.md index 6d703d9ab..6ecac0731 100644 --- a/docs/self-hosting/scaling.md +++ b/docs/self-hosting/scaling.md @@ -268,8 +268,10 @@ turns the limit off). `canceling statement due to statement timeout`. - The admin status page lists `rollup-full-backfill` under failing background jobs, and every admin gets a push once a system job has - failed three times in a row, a reminder at most once a day while it keeps - failing, and a notice when it runs again. + failed three jobs in a row (retries of one job count once; a job that runs + per account counts per account), a reminder at most once a day while it + keeps failing, and a notice when the failing job runs again. A one-off job + that failed and does not run again is announced once. ### What the app does about it diff --git a/src/lib/jobs/__tests__/system-job-alert.test.ts b/src/lib/jobs/__tests__/system-job-alert.test.ts index 9719308d4..224fb0209 100644 --- a/src/lib/jobs/__tests__/system-job-alert.test.ts +++ b/src/lib/jobs/__tests__/system-job-alert.test.ts @@ -13,68 +13,140 @@ import { } from "@/lib/jobs/system-job-alert"; const at = new Date("2026-10-29T10:00:00.000Z"); +const HOUR_MS = 3_600_000; -function run(overrides: Partial = {}): TerminalRun { +/** The `ago`-th newest job (0 = newest), an hour apart. */ +function run(ago = 0, overrides: Partial = {}): TerminalRun { return { + key: "", state: "failed", - retryCount: 0, allFailed: false, - completedOn: at, + completedOn: new Date(at.getTime() - ago * HOUR_MS), error: "boom", expireSeconds: null, ...overrides, }; } +/** Jobs newest first, one hour apart, from a list of outcomes. */ +function history(key: string, outcomes: Array<"fail" | "ok">, from = 0) { + return outcomes.map((o, i) => + run(from + i, { + key, + state: o === "fail" ? "failed" : "completed", + error: o === "fail" ? `boom ${key}` : null, + }), + ); +} + describe("judgeQueueRuns", () => { it("is inactive with no run left in the ledger", () => { expect(judgeQueueRuns([]).state).toBe("inactive"); }); - it("counts every attempt of a run, so one job that used up its retries is a streak", () => { - const verdict = judgeQueueRuns([run({ retryCount: 3 })]); - expect(verdict).toMatchObject({ state: "failing", failures: 4 }); - expect(verdict.lastError).toBe("boom"); - }); - - it("calls the streak at the threshold, not before", () => { - const below = Array.from({ length: SYSTEM_JOB_FAILURE_STREAK - 1 }, () => - run(), - ); - expect(judgeQueueRuns(below).state).toBe("pending"); - expect(judgeQueueRuns([...below, run()]).state).toBe("failing"); - }); - - it("ends the streak at the newest success", () => { - const verdict = judgeQueueRuns([ - run({ error: "newest" }), - run({ state: "completed", error: null }), - run(), - run(), - run(), - ]); - expect(verdict).toMatchObject({ - state: "ok", - failures: 1, - lastError: "newest", + it("counts jobs, not attempts: one job that used up its retries is one failure", () => { + expect(judgeQueueRuns([run(0)])).toMatchObject({ + state: "pending", + failures: 0, }); }); - it("counts a completed run that says every candidate failed", () => { - const allFailed = run({ state: "completed", allFailed: true }); - expect(judgeQueueRuns([allFailed, allFailed, allFailed]).state).toBe( - "failing", + it("calls the streak at the threshold, not before", () => { + const below = Array.from( + { length: SYSTEM_JOB_FAILURE_STREAK - 1 }, + (_, i) => run(i), ); + expect(judgeQueueRuns(below).state).toBe("pending"); + const verdict = judgeQueueRuns([...below, run(below.length)]); + expect(verdict).toMatchObject({ + state: "failing", + failures: SYSTEM_JOB_FAILURE_STREAK, + lastError: "boom", + lastFailureAt: at, + }); + }); + + it("does not report a recovery while the newest job failed", () => { + expect( + judgeQueueRuns(history("", ["fail", "ok", "fail", "fail", "fail"])), + ).toMatchObject({ state: "pending", failures: 0 }); + expect( + judgeQueueRuns(history("", ["ok", "fail", "fail", "fail"])).state, + ).toBe("ok"); + }); + + it("has nothing to report for a queue that never failed", () => { + expect(judgeQueueRuns(history("", ["ok", "ok"])).state).toBe("inactive"); + }); + + it("counts a completed job that says every candidate failed", () => { + const allFailed = (ago: number) => + run(ago, { state: "completed", allFailed: true }); + expect( + judgeQueueRuns([allFailed(0), allFailed(1), allFailed(2)]).state, + ).toBe("failing"); }); it("explains a bare timeout the way the admin card does", () => { - const verdict = judgeQueueRuns([ - run({ error: "job timed out", expireSeconds: 3600, retryCount: 2 }), - ]); + const verdict = judgeQueueRuns( + [0, 1, 2].map((ago) => + run(ago, { error: "job timed out", expireSeconds: 3600 }), + ), + ); expect(verdict.lastError).toMatch( /^job timed out: did not finish within its 1 h limit/, ); }); + + describe("per account", () => { + it("is failing while one account fails every hour and the others succeed", () => { + // Interleaved: the queue's newest job is always another account's + // success, which used to read as a recovery every other hour. + const verdict = judgeQueueRuns([ + ...history("a", ["fail", "fail", "fail"], 1), + ...history("b", ["ok", "ok", "ok", "ok"], 0), + ...history("c", ["ok", "ok", "ok"], 0), + ]); + expect(verdict).toMatchObject({ + state: "failing", + failures: 3, + lastError: "boom a", + }); + }); + + it("does not build a streak from three accounts failing once each", () => { + const verdict = judgeQueueRuns([ + ...history("a", ["fail"], 0), + ...history("b", ["fail"], 1), + ...history("c", ["fail"], 2), + ]); + expect(verdict.state).toBe("pending"); + }); + + it("recovers only when the failing account itself succeeds", () => { + const others = history("b", ["ok", "ok"], 0); + expect( + judgeQueueRuns([ + ...history("a", ["fail", "fail", "fail"], 0), + ...others, + ]).state, + ).toBe("failing"); + expect( + judgeQueueRuns([ + ...history("a", ["ok", "fail", "fail", "fail"], 0), + ...others, + ]).state, + ).toBe("ok"); + }); + + it("reports the account that failed most recently", () => { + const verdict = judgeQueueRuns([ + ...history("old", ["fail", "fail", "fail", "fail"], 5), + ...history("new", ["fail", "fail", "fail"], 0), + ]); + expect(verdict).toMatchObject({ failures: 3, lastError: "boom new" }); + }); + }); }); describe("watchedQueues", () => { diff --git a/src/lib/jobs/system-job-alert.ts b/src/lib/jobs/system-job-alert.ts index d6174eec7..ff6187462 100644 --- a/src/lib/jobs/system-job-alert.ts +++ b/src/lib/jobs/system-job-alert.ts @@ -1,6 +1,6 @@ /** * Push to every admin when a background job of the instance keeps failing: - * once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` runs in a row, a + * once when a queue has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, a * reminder at most once a day while it keeps failing, and once when it runs * successfully again. Nothing else. * @@ -12,11 +12,28 @@ * account-facing status of its own (the `system` audience in * `subsystem-surface.ts`). * - * The streak is read from pg-boss's own ledger, newest terminal run first: a - * run that failed (or completed and said every candidate failed, - * `did.all_failed`) adds its attempts, retries included, and the first run - * that succeeded ends the streak. A queue with no terminal run left in the - * ledger is not judged at all. + * The streak is read from pg-boss's own ledger, newest terminal job first, + * and it counts jobs, not attempts: a job that used up its retries is one + * failure. A job that failed, or completed and said every candidate failed + * (`did.all_failed`), adds one; the first job that succeeded ends it. + * + * Many of these queues run one job per account (a rollup recompute, a + * backfill, a score), so the streak is kept per job key: the account the job + * carries (`data.userId`), or the whole queue for a job that carries none. + * Not the `singletonKey`: it often names a bucket, a window or a single + * workout, which makes nearly every job its own key, and a queue failing on + * every job would then never build a streak. A queue is failing while any + * key has failed `SYSTEM_JOB_FAILURE_STREAK` jobs in a row, and it has + * recovered only when no key is failing and a key that had failed ran + * successfully since; another account's success says nothing about the one + * that fails. One account failing every hour while the rest succeed is + * therefore one problem notice and one recovery, not a notice each hour. + * + * A reminder needs a failure newer than the notice before it. A queue that + * keeps failing is reminded once a day; a one-shot job (a boot backfill) that + * failed and does not run again is announced once and then left to the admin + * page, instead of a reminder every day until pg-boss deletes its row. A + * queue with nothing left in the ledger to judge is forgotten silently. * * The memory is one row per announced queue (`SystemJobAlertNotice`), claimed * with a compare-and-set before the notice is sent, exactly as the readiness @@ -41,21 +58,24 @@ import { type StoredNotice, } from "@/lib/recovery/readiness-notify"; -/** Failed runs in a row, retries included, before the admins are told. */ +/** Failed jobs in a row, of one job key, before the admins are told. */ export const SYSTEM_JOB_FAILURE_STREAK = 3; -/** Newest terminal runs read per queue. */ -const LOOKBACK_ROWS = 50; +/** Newest terminal jobs read per queue, across every key. */ +const LOOKBACK_ROWS = 5000; + +/** Newest terminal jobs judged per job key. */ +const LOOKBACK_ROWS_PER_KEY = 20; /** Where the notice takes the admin: the page with the failing-jobs card. */ const ADMIN_STATUS_PATH = "/admin/system-status"; -/** One terminal run as the ledger holds it, newest first. */ +/** One terminal job as the ledger holds it. */ export interface TerminalRun { + /** The account the job ran for (`data.userId`), or "" for the queue. */ + key: string; state: "completed" | "failed"; - /** Retries pg-boss ran before this run settled. */ - retryCount: number; - /** The run completed but reported that every candidate failed. */ + /** The job completed but reported that every candidate failed. */ allFailed: boolean; completedOn: Date; error: string | null; @@ -65,39 +85,66 @@ export interface TerminalRun { /** What the ledger says about one queue. */ export interface QueueVerdict { /** - * `failing`: the streak reached the threshold. `ok`: the newest run, or a - * run after the last failures, succeeded. `pending`: fewer failures than - * the threshold and no success to judge by. `inactive`: no run left. + * `failing`: some key reached the streak. `ok`: no key is failing, and a + * key that had failed has since run successfully. `pending`: a failure + * short of the streak, and no recovery to report. `inactive`: nothing to + * report, no job left or none that failed. */ state: "failing" | "ok" | "pending" | "inactive"; - /** Failed attempts in a row, newest first. */ + /** Failed jobs in a row of the key that failed most recently. */ failures: number; - /** The newest failure's message, redacted and truncated. */ + /** That key's newest failure, redacted and truncated. */ lastError: string; + /** When the newest failure of a failing key settled, if any key is. */ + lastFailureAt: Date | null; } -/** Pure: judge one queue's newest terminal runs. */ +function failed(run: TerminalRun): boolean { + return run.state === "failed" || run.allFailed; +} + +/** Pure: judge one queue's terminal jobs, in any order. */ export function judgeQueueRuns(runs: readonly TerminalRun[]): QueueVerdict { - if (runs.length === 0) - return { state: "inactive", failures: 0, lastError: "" }; - let failures = 0; - let lastError: string | null = null; - let succeeded = false; + const byKey = new Map(); for (const run of runs) { - if (run.state === "completed" && !run.allFailed) { - succeeded = true; - break; - } - failures += run.retryCount + 1; - lastError ??= presentError(run.error, run.expireSeconds); + const list = byKey.get(run.key) ?? []; + list.push(run); + byKey.set(run.key, list); } - const state = - failures >= SYSTEM_JOB_FAILURE_STREAK - ? "failing" - : succeeded - ? "ok" - : "pending"; - return { state, failures, lastError: lastError ?? "" }; + let failing: { streak: number; newest: TerminalRun } | null = null; + let recovered = false; + let anyFailure = false; + for (const list of byKey.values()) { + list.sort((a, b) => b.completedOn.getTime() - a.completedOn.getTime()); + let streak = 0; + while (streak < list.length && failed(list[streak])) streak += 1; + if (list.some(failed)) anyFailure = true; + if (streak === 0) { + // Newest job succeeded: a recovery, when this key had failed before. + if (list.some(failed)) recovered = true; + continue; + } + if ( + streak >= SYSTEM_JOB_FAILURE_STREAK && + (failing === null || + list[0].completedOn.getTime() > failing.newest.completedOn.getTime()) + ) { + failing = { streak, newest: list[0] }; + } + } + if (failing) { + return { + state: "failing", + failures: failing.streak, + lastError: presentError( + failing.newest.error, + failing.newest.expireSeconds, + ), + lastFailureAt: failing.newest.completedOn, + }; + } + const state = recovered ? "ok" : anyFailure ? "pending" : "inactive"; + return { state, failures: 0, lastError: "", lastFailureAt: null }; } /** The queues this alert watches: every one whose failure has no other surface. */ @@ -108,9 +155,11 @@ export function watchedQueues(): string[] { } /** - * The newest terminal runs of each queue, newest first, one statement over - * pg-boss's `(name, completed_on)` index. `null` when the queue schema is not - * there (a web-only deployment): no queue to ask is not a healthy queue. + * The newest terminal jobs of each queue, keyed by account, one statement + * over pg-boss's `(name, completed_on)` index: the newest `LOOKBACK_ROWS` of + * the queue, and of those the newest `LOOKBACK_ROWS_PER_KEY` of each key. + * `null` when the queue schema is not there (a web-only deployment): no + * queue to ask is not a healthy queue. */ export async function readTerminalRuns( queues: readonly string[], @@ -119,40 +168,48 @@ export async function readTerminalRuns( const rows = await prisma.$queryRaw< Array<{ name: string; + job_key: string; state: string; - retry_count: number; all_failed: string | null; completed_on: Date; error: string | null; expire_seconds: number | null; }> >` - SELECT q.name, j.state, j.retry_count, j.all_failed, j.completed_on, - j.error, j.expire_seconds + SELECT q.name, k.job_key, k.state, k.all_failed, k.completed_on, + k.error, k.expire_seconds FROM unnest(${queues as string[]}::text[]) AS q(name) CROSS JOIN LATERAL ( - SELECT state, retry_count, completed_on, expire_seconds, - output->'did'->>'all_failed' AS all_failed, - COALESCE( - output->>'message', - output->'value'->>'message', - output::text - ) AS error - FROM pgboss.job - WHERE name = q.name - AND completed_on IS NOT NULL - AND state IN ('completed', 'failed') - ORDER BY completed_on DESC - LIMIT ${LOOKBACK_ROWS} - ) j - ORDER BY q.name, j.completed_on DESC + SELECT j.*, + ROW_NUMBER() OVER ( + PARTITION BY j.job_key ORDER BY j.completed_on DESC + ) AS key_rank + FROM ( + SELECT state, completed_on, expire_seconds, + COALESCE(data->>'userId', '') AS job_key, + output->'did'->>'all_failed' AS all_failed, + COALESCE( + output->>'message', + output->'value'->>'message', + output::text + ) AS error + FROM pgboss.job + WHERE name = q.name + AND completed_on IS NOT NULL + AND state IN ('completed', 'failed') + ORDER BY completed_on DESC + LIMIT ${LOOKBACK_ROWS} + ) j + ) k + WHERE k.key_rank <= ${LOOKBACK_ROWS_PER_KEY} + ORDER BY q.name, k.completed_on DESC `; const byQueue = new Map(); for (const row of rows) { const runs = byQueue.get(row.name) ?? []; runs.push({ + key: row.job_key, state: row.state === "failed" ? "failed" : "completed", - retryCount: Number(row.retry_count ?? 0), allFailed: row.all_failed === "true", completedOn: row.completed_on, error: row.error, @@ -222,6 +279,16 @@ export async function notifySystemJobFailures( // With no admin yet there is nobody to tell, so nothing is claimed: the // first admin hears about a queue that is still failing. if (!decision.next || admins.length === 0) continue; + // Nothing failed since the last notice: a job that does not run again is + // not news a second time. + if ( + decision.send === "reminder" && + stored?.notifiedAt && + verdict.lastFailureAt && + verdict.lastFailureAt.getTime() <= stored.notifiedAt.getTime() + ) { + continue; + } // Claim: only the evaluation that moves the row from what it read sends. const claimed = row diff --git a/tests/integration/system-job-alert.test.ts b/tests/integration/system-job-alert.test.ts index bb267bf76..dd9518143 100644 --- a/tests/integration/system-job-alert.test.ts +++ b/tests/integration/system-job-alert.test.ts @@ -4,7 +4,10 @@ * A background job of the instance that keeps failing pushes every admin * once, a reminder at most once a day while it keeps failing, and once when it * runs successfully again. Fewer failures than the streak, and queues that - * have an account-facing surface of their own, push nothing. + * have an account-facing surface of their own, push nothing. The streak + * counts jobs, not retries, and is kept per account: one account failing + * every hour beside accounts that succeed is one problem and one recovery, + * and a job that does not run again is announced once, not every day. * * pg-boss provisions its own schema here and the jobs are failed and completed * through its own API, as in `job-failures.test.ts`, so the reader is proven @@ -54,19 +57,32 @@ const HOUR_MS = 3_600_000; let boss: PgBoss; -async function runOnce(queue: string, outcome: "fail" | "complete") { - const id = await boss.send(queue, { userId: "u" }, { retryLimit: 0 }); +async function runOnce( + queue: string, + outcome: "fail" | "complete", + userId = "u", + retryLimit = 0, +) { + const id = await boss.send( + queue, + { userId }, + { retryLimit, retryDelay: 0, retryBackoff: false }, + ); if (!id) throw new Error("queue rejected the job"); - const [job] = await boss.fetch(queue); - if (!job) throw new Error("nothing to fetch"); - if (outcome === "fail") { - await boss.fail( - queue, - job.id, - new Error("canceling statement due to statement timeout"), - ); - } else { - await boss.complete(queue, job.id, { ok: true }); + // Every attempt, retries included, until pg-boss settles the job. + for (let attempt = 0; attempt <= retryLimit; attempt++) { + const [job] = await boss.fetch(queue); + if (!job) throw new Error("nothing to fetch"); + if (outcome === "fail") { + await boss.fail( + queue, + job.id, + new Error(`canceling statement due to statement timeout (${userId})`), + ); + } else { + await boss.complete(queue, job.id, { ok: true }); + return; + } } } @@ -127,7 +143,8 @@ describe("notifySystemJobFailures", () => { expect(again.sent).toEqual([]); expect(dispatched.calls).toEqual([]); - // A day later: one reminder. + // It fails once more; a day after the notice: one reminder. + await runOnce(SYSTEM_QUEUE, "fail"); const reminder = await notifySystemJobFailures( new Date(now.getTime() + 25 * HOUR_MS), ); @@ -175,4 +192,58 @@ describe("notifySystemJobFailures", () => { expect(outcome.failing).toBe(0); expect(outcome.sent).toEqual([]); }); + + it("counts a job that used up its retries as one failure", async () => { + await makeUser({ role: "ADMIN" }); + await runOnce(SYSTEM_QUEUE, "fail", "u", 2); + const outcome = await notifySystemJobFailures(); + expect(outcome.failing).toBe(0); + expect(outcome.sent).toEqual([]); + }); + + it("tells once about one account failing every hour beside accounts that succeed", async () => { + await makeUser({ role: "ADMIN" }); + const start = Date.now(); + const sent: string[] = []; + // Ten hours: account `a` fails every hour, the others succeed after it, + // so the queue's newest job is always a success. + for (let hour = 0; hour < 10; hour++) { + await runOnce(SYSTEM_QUEUE, "fail", "a"); + await runOnce(SYSTEM_QUEUE, "complete", "b"); + await runOnce(SYSTEM_QUEUE, "complete", "c"); + const outcome = await notifySystemJobFailures( + new Date(start + hour * HOUR_MS), + ); + sent.push(...outcome.sent.map((n) => n.kind)); + } + expect(sent).toEqual(["problem"]); + expect(String(dispatched.calls[0].params?.error)).toContain("(a)"); + + // `a` succeeds: one recovery. + await runOnce(SYSTEM_QUEUE, "complete", "a"); + const recovered = await notifySystemJobFailures( + new Date(start + 10 * HOUR_MS), + ); + expect(recovered.sent).toEqual([ + { queue: SYSTEM_QUEUE, kind: "recovered" }, + ]); + }); + + it("does not remind about a job that has not run again since the notice", async () => { + await makeUser({ role: "ADMIN" }); + for (let i = 0; i < SYSTEM_JOB_FAILURE_STREAK; i++) { + await runOnce(SYSTEM_QUEUE, "fail"); + } + const now = new Date(Date.now() + 1_000); + expect((await notifySystemJobFailures(now)).sent).toEqual([ + { queue: SYSTEM_QUEUE, kind: "problem" }, + ]); + for (const days of [1, 2, 5]) { + const later = await notifySystemJobFailures( + new Date(now.getTime() + days * 24 * HOUR_MS), + ); + expect(later.failing).toBe(1); + expect(later.sent).toEqual([]); + } + }); }); From 6df0fad3b7780ad8a9249fedcad2a93c8c5e4bed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:00:01 +0100 Subject: [PATCH 4/9] fix(labs): let the lab date repair skip a damaged fact and read the app's zone One staged fact that would not decrypt threw out of the whole repair, so the job failed, retried, and was enqueued again on every boot. It is now skipped and counted as undecryptable. The candidate query read an account zone Postgres does not know as UTC, while the repair resolves it to the instance default and then Europe/Berlin. A value confirmed near midnight stayed a candidate the repair always left alone, so the set never converged. The query now falls back the same way. --- src/lib/jobs/lab-document-date-repair.ts | 48 ++++++++++++--- .../lab-document-date-repair.test.ts | 60 ++++++++++++++++++- 2 files changed, 98 insertions(+), 10 deletions(-) diff --git a/src/lib/jobs/lab-document-date-repair.ts b/src/lib/jobs/lab-document-date-repair.ts index 57edb24e5..59f6c9542 100644 --- a/src/lib/jobs/lab-document-date-repair.ts +++ b/src/lib/jobs/lab-document-date-repair.ts @@ -44,7 +44,7 @@ import { annotate } from "@/lib/logging/context"; import { withBackgroundEvent } from "@/lib/logging/background"; import { dateOnlyAtNoonUtc } from "@/lib/tz/date-only"; import { userDayKey } from "@/lib/tz/format"; -import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { DEFAULT_TIMEZONE, resolveUserTimezone } from "@/lib/tz/resolver"; import type { ObservationFactData } from "@/lib/validations/inbound-documents"; import { @@ -76,11 +76,25 @@ interface CandidateRow { function candidateQuery(limit: number | null): Prisma.Sql { // The day comparison runs here as well as in the repair, so a value that // really was taken on its confirmation day drops out of the boot check and - // the pass converges to an empty set. A zone Postgres does not know is read - // as UTC rather than failing the query; the repair re-checks every row in - // the zone the app resolves. + // the pass converges to an empty set. Both must read the same zone: a + // stored zone Postgres does not know falls back the way + // `resolveUserTimezone` falls back, to the instance default and then to + // `DEFAULT_TIMEZONE`. Reading it as UTC instead kept a value confirmed + // near midnight in the candidate set that the repair, in the app's zone, + // left alone, so every boot enqueued the job again. The repair still + // re-checks every row in the zone the app resolves. return Prisma.sql` - WITH "known_zones" AS (SELECT "name" FROM pg_timezone_names) + WITH "known_zones" AS (SELECT "name" FROM pg_timezone_names), + "instance_zone" AS ( + SELECT COALESCE( + ( + SELECT s."default_user_timezone" FROM "app_settings" s + WHERE s."id" = 'singleton' + AND s."default_user_timezone" IN (SELECT "name" FROM "known_zones") + ), + ${DEFAULT_TIMEZONE} + ) AS "name" + ) SELECT r."id", r."user_id" AS "userId", r."taken_at" AS "takenAt", @@ -106,7 +120,8 @@ function candidateQuery(limit: number | null): Prisma.Sql { <> to_char( (r."taken_at" AT TIME ZONE 'UTC') AT TIME ZONE ( CASE WHEN u."timezone" IN (SELECT "name" FROM "known_zones") - THEN u."timezone" ELSE 'UTC' END + THEN u."timezone" + ELSE (SELECT "name" FROM "instance_zone") END ), 'YYYY-MM-DD' ) @@ -128,6 +143,13 @@ export interface LabDocumentDateRepairResult { updated: number; /** Rows left alone: an own date, or already on the document's day. */ unchanged: number; + /** + * Rows whose staged fact would not decrypt (a key retired too early, a + * damaged row). Skipped: without the fact there is no telling whether the + * value carries a date of its own, and one such row must not stop the + * repair of every other account. + */ + undecryptable: number; } /** The repair itself, across every account. */ @@ -138,9 +160,16 @@ export async function repairUndatedDocumentLabValues( const touchedUsers = new Set(); let updated = 0; let unchanged = 0; + let undecryptable = 0; for (const row of rows) { - const data = decryptFactData(row.dataEncrypted) as ObservationFactData; + let data: ObservationFactData; + try { + data = decryptFactData(row.dataEncrypted) as ObservationFactData; + } catch { + undecryptable += 1; + continue; + } // A value with a date of its own is never touched, whatever its row says. if (statedDay(data.effectiveDate ?? null) !== null) { unchanged += 1; @@ -175,7 +204,7 @@ export async function repairUndatedDocumentLabValues( } for (const userId of touchedUsers) invalidateUserHealthScore(userId); - return { scanned: rows.length, updated, unchanged }; + return { scanned: rows.length, updated, unchanged, undecryptable }; } /** The queue handler: one run over every account. */ @@ -193,7 +222,7 @@ export async function handleLabDocumentDateRepair( } workerLog( "info", - `[lab-document-date-repair] scanned=${result.scanned} updated=${result.updated} unchanged=${result.unchanged}`, + `[lab-document-date-repair] scanned=${result.scanned} updated=${result.updated} unchanged=${result.unchanged} undecryptable=${result.undecryptable}`, ); annotate({ action: { name: "labs.documentDate.repair" }, @@ -203,6 +232,7 @@ export async function handleLabDocumentDateRepair( candidates_scanned: result.scanned, updated: result.updated, unchanged: result.unchanged, + undecryptable: result.undecryptable, }); }); } diff --git a/tests/integration/lab-document-date-repair.test.ts b/tests/integration/lab-document-date-repair.test.ts index 82c6aa086..97ba1611c 100644 --- a/tests/integration/lab-document-date-repair.test.ts +++ b/tests/integration/lab-document-date-repair.test.ts @@ -8,12 +8,15 @@ * at noon UTC, and leaves alone a value with a date of its own, a value that * really was taken on its confirmation day in the owner's zone, one the * person re-dated since, and one whose document has no date. A second run - * changes nothing. + * changes nothing. A fact that does not decrypt is skipped and counted + * rather than stopping the run, and an account zone Postgres does not know + * is read the way the app reads it, so the candidate set still converges. */ import { beforeEach, describe, expect, it } from "vitest"; import { encryptFactData, encryptFactProvenance } from "@/lib/documents/store"; import { repairUndatedDocumentLabValues } from "@/lib/jobs/lab-document-date-repair"; +import { invalidateUserTimezone } from "@/lib/tz/resolver"; import { getPrismaClient, truncateAllTables } from "./setup"; const USER = "user-lab-document-date-repair"; @@ -22,6 +25,7 @@ const noon = (day: string) => new Date(`${day}T12:00:00.000Z`); beforeEach(async () => { const prisma = getPrismaClient(); await truncateAllTables(prisma); + invalidateUserTimezone(USER); await prisma.user.create({ data: { id: USER, @@ -229,4 +233,58 @@ describe("lab document date repair", () => { expect(result.updated).toBe(0); expect((await takenAt())["linked-twice"]).toBe(confirmed.toISOString()); }); + + it("skips and counts a fact that does not decrypt, and repairs the rest", async () => { + const confirmed = new Date("2026-10-20T08:00:00.000Z"); + for (const id of ["damaged", "misdated"]) { + await confirmedValue({ + id, + reportDate: "2021-03-01", + documentDate: null, + effectiveDate: null, + createdAt: confirmed, + takenAt: confirmed, + }); + } + const prisma = getPrismaClient(); + await prisma.extractedFact.updateMany({ + where: { committedRecordId: "damaged" }, + data: { dataEncrypted: Buffer.from("not a ciphertext at all") }, + }); + + const result = await repairUndatedDocumentLabValues(prisma); + + expect(result).toMatchObject({ updated: 1, undecryptable: 1 }); + expect(await takenAt()).toEqual({ + damaged: confirmed.toISOString(), + misdated: "2021-03-01T12:00:00.000Z", + }); + }); + + it("reads a zone Postgres does not know the way the app does, so the candidates converge", async () => { + const prisma = getPrismaClient(); + await prisma.user.update({ + where: { id: USER }, + data: { timezone: "Mars/Olympus_Mons" }, + }); + // 00:30 on the report day in the instance default (Europe/Berlin), the + // evening before in UTC: on its document's day for the app. + const justAfterMidnight = new Date("2026-10-19T22:30:00.000Z"); + await confirmedValue({ + id: "taken-on-confirm-day", + reportDate: "2026-10-20", + documentDate: null, + effectiveDate: null, + createdAt: justAfterMidnight, + takenAt: justAfterMidnight, + }); + + const result = await repairUndatedDocumentLabValues(prisma); + + // Not even a candidate: the boot check finds nothing to enqueue. + expect(result).toMatchObject({ scanned: 0, updated: 0 }); + expect((await takenAt())["taken-on-confirm-day"]).toBe( + justAfterMidnight.toISOString(), + ); + }); }); From fc117d7cd9e03a7080b291edf975bff62d4e8cef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:01:25 +0100 Subject: [PATCH 5/9] fix(auth): date a provider re-proof by its auth_time The re-proof accepted an auth_time up to five minutes old and then stamped the session at callback time, so the proof held for close to ten minutes after the person signed in at the provider. It now stamps the authentication instant, clamped to now for a provider clock that runs ahead, and the gates' five-minute window runs from there. --- src/lib/auth/__tests__/oidc-reproof.test.ts | 32 ++++++++++++++++++++- src/lib/auth/oidc-reproof.ts | 11 ++++++- 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/src/lib/auth/__tests__/oidc-reproof.test.ts b/src/lib/auth/__tests__/oidc-reproof.test.ts index dd396a041..660cdb747 100644 --- a/src/lib/auth/__tests__/oidc-reproof.test.ts +++ b/src/lib/auth/__tests__/oidc-reproof.test.ts @@ -47,6 +47,7 @@ import { reproofReturnUrl, } from "@/lib/auth/oidc-reproof"; import { readStoredOidcState } from "@/lib/auth/oidc-callback/state"; +import { RECENT_PROOF_MAX_AGE_MS } from "@/lib/auth/existing-factor-proof"; const ISSUER = "https://idp.example.test"; const OIDC_ENV = { @@ -195,9 +196,10 @@ describe("recordOidcReproof", () => { identity: { sub: "sub-1", authTime: Math.floor(now / 1000) - 5 }, }); expect(outcome).toEqual({ ok: true }); + // Stamped at the authentication, not at the callback. expect(db.updateSession).toHaveBeenCalledWith({ where: { id: "s1" }, - data: { reproofAt: new Date(now) }, + data: { reproofAt: new Date((Math.floor(now / 1000) - 5) * 1000) }, }); expect(refundRateLimit).toHaveBeenCalledWith("auth:step-up:u1"); expect(auditLog).toHaveBeenCalledWith( @@ -206,6 +208,34 @@ describe("recordOidcReproof", () => { ); }); + it("runs the window from the authentication, so a token near its end buys no extra minutes", async () => { + db.findUser.mockResolvedValue(accountRow()); + const authTime = Math.floor(now / 1000) - 4 * 60; + await recordOidcReproof({ ...args, identity: { sub: "sub-1", authTime } }); + const [call] = db.updateSession.mock.calls[0] as [ + { data: { reproofAt: Date } }, + ]; + const stamped = call.data.reproofAt; + expect(stamped.getTime()).toBe(authTime * 1000); + // Two minutes after the callback the proof is six minutes old: past the + // window every recent-proof gate applies. + expect(now + 2 * 60_000 - stamped.getTime()).toBeGreaterThan( + RECENT_PROOF_MAX_AGE_MS, + ); + }); + + it("never stamps a proof in the future for a provider clock a little ahead", async () => { + db.findUser.mockResolvedValue(accountRow()); + await recordOidcReproof({ + ...args, + identity: { sub: "sub-1", authTime: Math.floor(now / 1000) + 30 }, + }); + expect(db.updateSession).toHaveBeenCalledWith({ + where: { id: "s1" }, + data: { reproofAt: new Date(now) }, + }); + }); + it("refuses a stale auth_time without stamping", async () => { db.findUser.mockResolvedValue(accountRow()); const outcome = await recordOidcReproof({ diff --git a/src/lib/auth/oidc-reproof.ts b/src/lib/auth/oidc-reproof.ts index a84e03948..c87b82eb5 100644 --- a/src/lib/auth/oidc-reproof.ts +++ b/src/lib/auth/oidc-reproof.ts @@ -140,9 +140,18 @@ export async function recordOidcReproof(args: { return { ok: false, reason: refusal }; } + // The proof is as old as the authentication, not the callback: the gate's + // window then runs from the moment the person signed in at the provider, + // so a token near the end of its window buys no extra minutes. A provider + // clock a little ahead is clamped to now, never stamped into the future. + // (A token without `auth_time` was refused above.) + const authenticatedAtMs = + identity.authTime === null + ? nowMs + : Math.min(identity.authTime * 1000, nowMs); await prisma.session.update({ where: { id: sessionId }, - data: { reproofAt: new Date(nowMs) }, + data: { reproofAt: new Date(authenticatedAtMs) }, }); await refundReproof(userId); await auditLog("auth.reproof.succeeded", { From 6fb50b01b8c6f9531ff001dc1465c4af18672f80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:03:09 +0100 Subject: [PATCH 6/9] fix(auth): offer the provider re-proof to a browser only The MFA status read listed oidc among the proof methods for a Bearer caller too, although the provider round trip stamps a browser session and nothing else, and the API contract says a token is never offered it. The status read now leaves it out on the Bearer path. --- src/app/api/auth/me/mfa/route.ts | 11 +++++++--- tests/integration/oidc-reproof.test.ts | 29 ++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/src/app/api/auth/me/mfa/route.ts b/src/app/api/auth/me/mfa/route.ts index 703fbf091..c4663f7c0 100644 --- a/src/app/api/auth/me/mfa/route.ts +++ b/src/app/api/auth/me/mfa/route.ts @@ -27,7 +27,7 @@ import { recentProofMethods } from "@/lib/auth/second-factor"; export const dynamic = "force-dynamic"; export const GET = apiHandler(async () => { - const { user } = await requireAuth(); + const { user, authMethod } = await requireAuth(); const [recoveryCodesRemaining, webauthn, proofMethods] = await Promise.all([ countRemainingRecoveryCodes(user.id), @@ -50,7 +50,12 @@ export const GET = apiHandler(async () => { passkeyNudgeDismissed: Boolean(user.passkeyUpgradeNudgeDismissed), // The ways this account can confirm it is you, the list the recent-proof // refusal names. The passkey card offers exactly these before adding one, - // so an account without a password is not handed a password field. - proofMethods, + // so an account without a password is not handed a password field. The + // provider round trip stamps a browser session and nothing else, so a + // token is never offered it. + proofMethods: + authMethod === "cookie" + ? proofMethods + : proofMethods.filter((method) => method !== "oidc"), }); }); diff --git a/tests/integration/oidc-reproof.test.ts b/tests/integration/oidc-reproof.test.ts index 7d8beb9e7..cc5f728dc 100644 --- a/tests/integration/oidc-reproof.test.ts +++ b/tests/integration/oidc-reproof.test.ts @@ -59,6 +59,7 @@ vi.mock("@/lib/auth/oidc", async (importOriginal) => { }); const oidc = await import("@/lib/auth/oidc"); +const { hashToken } = await import("@/lib/auth/hmac"); const USER_ID = "user-oidc-reproof"; const SUB = "idp-sub-reproof"; @@ -313,4 +314,32 @@ describe("confirming at the identity provider", () => { expect(back.searchParams.get("reproof")).toBe("failed"); expect(oidc.discoverOidcMetadata).not.toHaveBeenCalled(); }); + + it("is listed on the status read for the browser, never for a token", async () => { + const { GET } = await import("@/app/api/auth/me/mfa/route"); + const status = async () => { + const res = await (GET as unknown as RouteFn)( + new NextRequest("http://localhost:3000/api/auth/me/mfa"), + ); + expect(res.status).toBe(200); + return ((await res.json()) as { data: { proofMethods: string[] } }).data + .proofMethods; + }; + + await startSession(HOUR); + expect(await status()).toEqual(["oidc"]); + + cookieJar.clear(); + const raw = `hlk_mfastatus${"0".repeat(64 - "mfastatus".length)}`; + await getPrismaClient().apiToken.create({ + data: { + userId: USER_ID, + name: "device", + tokenHash: hashToken(raw), + permissions: ["*"], + }, + }); + headerJar.set("authorization", `Bearer ${raw}`); + expect(await status()).toEqual([]); + }); }); From ec0f67f8505d568959688c4c8b78a93d11debfe8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:04:21 +0100 Subject: [PATCH 7/9] fix(admin): give the single-account rollup recompute the boot fold's budget The admin recompute for one account called the full fold without a statement budget, so on the large accounts it is meant for each yearly statement ran under the database's ordinary limit and timed out, as the boot job had before it got its own. It now passes the same per-statement budget. --- src/app/api/admin/rollups/recompute/route.ts | 13 +- .../admin-rollups-recompute.test.ts | 129 ++++++++++++++++++ 2 files changed, 139 insertions(+), 3 deletions(-) create mode 100644 tests/integration/admin-rollups-recompute.test.ts diff --git a/src/app/api/admin/rollups/recompute/route.ts b/src/app/api/admin/rollups/recompute/route.ts index 63f612c49..e947f635c 100644 --- a/src/app/api/admin/rollups/recompute/route.ts +++ b/src/app/api/admin/rollups/recompute/route.ts @@ -16,8 +16,9 @@ * * - `userId` — single user mode. Synchronously calls * `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. + * statement, each statement under the boot job's budget) 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). * @@ -38,6 +39,7 @@ import { annotate } from "@/lib/logging/context"; import { enqueueBootTimeRollupBackfill, foldUserRollupHistory, + ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, } from "@/lib/rollups/full-backfill"; interface RecomputeBody { @@ -70,7 +72,12 @@ export const POST = apiHandler(async (request: NextRequest) => { let result: Awaited>; try { - result = await foldUserRollupHistory(userId); + // The per-statement budget the boot job folds with: this mode is for + // the accounts the boot job could not finish, and under the database's + // ordinary limit their yearly statements time out here too. + result = await foldUserRollupHistory(userId, { + statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + }); } catch (err) { const message = err instanceof Error ? err.message : String(err); return apiError(message, 500); diff --git a/tests/integration/admin-rollups-recompute.test.ts b/tests/integration/admin-rollups-recompute.test.ts new file mode 100644 index 000000000..0ae49e279 --- /dev/null +++ b/tests/integration/admin-rollups-recompute.test.ts @@ -0,0 +1,129 @@ +/** + * `POST /api/admin/rollups/recompute` for one account, against Postgres. + * + * The single-account mode exists for the accounts the boot backfill could not + * finish, which are the large ones. It folds the account in the request, and + * it must give every statement the same budget the boot job gives it + * (`ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS`): under the database's ordinary + * statement limit the yearly fold of such an account times out, which is the + * very failure the operator is trying to get past. The fold itself runs for + * real; the wrapper only records what the route handed it. + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS } from "@/lib/rollups/full-backfill"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const folds = vi.hoisted(() => ({ + calls: [] as Array<{ userId: string; statementTimeoutMs?: number }>, +})); + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/rollups/full-backfill", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + foldUserRollupHistory: async ( + ...args: Parameters + ) => { + folds.calls.push({ + userId: args[0], + statementTimeoutMs: args[1]?.statementTimeoutMs, + }); + return actual.foldUserRollupHistory(...args); + }, + }; +}); + +const ADMIN = "user-admin-recompute"; +const TARGET = "user-recompute-target"; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + cookieJar.clear(); + folds.calls.length = 0; + await prisma.user.createMany({ + data: [ + { + id: ADMIN, + username: "admin-recompute", + email: "admin-recompute@example.test", + role: "ADMIN", + }, + { + id: TARGET, + username: "recompute-target", + email: "recompute-target@example.test", + }, + ], + }); + const session = await prisma.session.create({ + data: { userId: ADMIN, expiresAt: new Date(Date.now() + 3_600_000) }, + }); + cookieJar.set("healthlog_session", session.id); + await prisma.measurement.create({ + data: { + userId: TARGET, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(Date.now() - 30 * 86_400_000), + }, + }); +}); + +describe("POST /api/admin/rollups/recompute for one account", () => { + it("folds it with the boot job's statement budget", async () => { + const { POST } = await import("@/app/api/admin/rollups/recompute/route"); + const res = await POST( + new NextRequest("http://localhost/api/admin/rollups/recompute", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ userId: TARGET }), + }), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { mode: string; rowsUpserted: number }; + }; + expect(body.data.mode).toBe("single"); + expect(body.data.rowsUpserted).toBeGreaterThan(0); + + expect(folds.calls).toEqual([ + { + userId: TARGET, + statementTimeoutMs: ROLLUP_FULL_BACKFILL_STATEMENT_BUDGET_MS, + }, + ]); + }); +}); From 6b820fa3c5ebc0c1b6444bffa26889a258bb8208 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:05:43 +0100 Subject: [PATCH 8/9] fix(rollups): defer the dense fold when the backfill check cannot read the queue The check whether an account's full rollup backfill is queued or running answered false on any error, so a failed read let the dense fold delete raw samples the backfill still had to read. A missing queue table still answers false, since a web-only deployment has no backfill; any other failure now counts as pending and is annotated. A backfill that failed for good is still not pending, as the comment now says. --- src/lib/rollups/full-backfill.ts | 27 ++++++++++++++++--- .../integration/rollup-full-backfill.test.ts | 21 +++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/src/lib/rollups/full-backfill.ts b/src/lib/rollups/full-backfill.ts index 1d42b7a33..9894f3402 100644 --- a/src/lib/rollups/full-backfill.ts +++ b/src/lib/rollups/full-backfill.ts @@ -24,6 +24,7 @@ import { activeAccountSql } from "@/lib/account-deletion/active-account"; import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { annotate } from "@/lib/logging/context"; import { applyStatementBudget, recomputeUserRollups, @@ -199,13 +200,24 @@ export async function sendRollupFullBackfillContinuation( * 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. + * + * False when there is no queue table (a web-only deployment has no full + * backfill either). Any other failure to read the queue counts as pending: + * the dense fold's deletion cannot be undone, and deferring an account costs + * one night. The failure is annotated on the caller's event. + * + * A full fold that failed for good (retries spent) is not pending, and the + * dense fold goes ahead on that account; the system job alert reports the + * failed backfill, and the next boot's discovery enqueues it again. */ export async function isRollupFullBackfillPending( userId: string, ): Promise { try { + const [queue] = await prisma.$queryRaw>` + SELECT to_regclass('pgboss.job') IS NOT NULL AS present + `; + if (!queue?.present) return false; const rows = await prisma.$queryRaw>` SELECT count(*)::int AS n FROM pgboss.job @@ -214,8 +226,15 @@ export async function isRollupFullBackfillPending( AND data->>'userId' = ${userId} `; return (rows[0]?.n ?? 0) > 0; - } catch { - return false; + } catch (err) { + annotate({ + meta: { + rollup_backfill_pending_check_failed: true, + rollup_backfill_pending_check_error: + err instanceof Error ? err.message.slice(0, 200) : String(err), + }, + }); + return true; } } diff --git a/tests/integration/rollup-full-backfill.test.ts b/tests/integration/rollup-full-backfill.test.ts index cdf0742fa..a41182abe 100644 --- a/tests/integration/rollup-full-backfill.test.ts +++ b/tests/integration/rollup-full-backfill.test.ts @@ -344,4 +344,25 @@ describe("isRollupFullBackfillPending", () => { await boss.complete(ROLLUP_FULL_BACKFILL_QUEUE, job.id); expect(await isRollupFullBackfillPending("user-a")).toBe(false); }); + + it("counts a queue it cannot read as pending, so the dense fold waits", async () => { + // The table is there but the read fails. + await prisma.$executeRawUnsafe( + `ALTER TABLE pgboss.job RENAME COLUMN data TO data_unreadable`, + ); + try { + expect(await isRollupFullBackfillPending("user-a")).toBe(true); + } finally { + await prisma.$executeRawUnsafe( + `ALTER TABLE pgboss.job RENAME COLUMN data_unreadable TO data`, + ); + } + }); +}); + +describe("isRollupFullBackfillPending without a queue", () => { + it("is false where there is no queue table at all", async () => { + await prisma.$executeRawUnsafe(`DROP SCHEMA IF EXISTS pgboss CASCADE`); + expect(await isRollupFullBackfillPending("user-a")).toBe(false); + }); }); From 34f8a856b44afd4d02fbae283498e3ee3696b8da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 29 Oct 2026 11:16:18 +0100 Subject: [PATCH 9/9] fix(labs): register the lab repair's undecryptable count as a job fact The count rides out in the job's outcome facts, whose vocabulary is an allowlist, and is kept as a failure tally so the catch that feeds it is not a silent one. --- src/lib/jobs/job-outcome.ts | 2 ++ src/lib/jobs/lab-document-date-repair.ts | 12 +++++++++--- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 40bc1d872..942b5fde6 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -317,6 +317,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "total", "trusted_devices_deleted", "unchanged", + // The lab date repair: staged facts that would not decrypt, skipped. + "undecryptable", "updated", "users", "users_clean_zero", diff --git a/src/lib/jobs/lab-document-date-repair.ts b/src/lib/jobs/lab-document-date-repair.ts index 59f6c9542..4813ab39e 100644 --- a/src/lib/jobs/lab-document-date-repair.ts +++ b/src/lib/jobs/lab-document-date-repair.ts @@ -160,14 +160,15 @@ export async function repairUndatedDocumentLabValues( const touchedUsers = new Set(); let updated = 0; let unchanged = 0; - let undecryptable = 0; + // Reported as `undecryptable` in the outcome facts. + let decryptFailed = 0; for (const row of rows) { let data: ObservationFactData; try { data = decryptFactData(row.dataEncrypted) as ObservationFactData; } catch { - undecryptable += 1; + decryptFailed += 1; continue; } // A value with a date of its own is never touched, whatever its row says. @@ -204,7 +205,12 @@ export async function repairUndatedDocumentLabValues( } for (const userId of touchedUsers) invalidateUserHealthScore(userId); - return { scanned: rows.length, updated, unchanged, undecryptable }; + return { + scanned: rows.length, + updated, + unchanged, + undecryptable: decryptFailed, + }; } /** The queue handler: one run over every account. */