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/__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/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/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/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", { 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/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/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 57edb24e5..4813ab39e 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,17 @@ export async function repairUndatedDocumentLabValues( const touchedUsers = new Set(); let updated = 0; let unchanged = 0; + // Reported as `undecryptable` in the outcome facts. + let decryptFailed = 0; for (const row of rows) { - const data = decryptFactData(row.dataEncrypted) as ObservationFactData; + let data: ObservationFactData; + try { + data = decryptFactData(row.dataEncrypted) as ObservationFactData; + } catch { + decryptFailed += 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 +205,12 @@ export async function repairUndatedDocumentLabValues( } for (const userId of touchedUsers) invalidateUserHealthScore(userId); - return { scanned: rows.length, updated, unchanged }; + return { + scanned: rows.length, + updated, + unchanged, + undecryptable: decryptFailed, + }; } /** The queue handler: one run over every account. */ @@ -193,7 +228,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 +238,7 @@ export async function handleLabDocumentDateRepair( candidates_scanned: result.scanned, updated: result.updated, unchanged: result.unchanged, + undecryptable: result.undecryptable, }); }); } 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/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/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/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/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, + }, + ]); + }); +}); 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"]); + }); }); 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(), + ); + }); }); 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([]); + }); }); 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); + }); + }); }); 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); + }); }); 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([]); + } + }); });