From 5e5e387fab74d100545806a9757947bb2e6ba952 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 20:55:14 +0200 Subject: [PATCH 1/9] chore(release): v1.40.1 --- CHANGELOG.md | 40 ++++++++++++++++++++++++++++++++++++++++ docs/api/openapi.yaml | 2 +- package.json | 2 +- public/sw.js | 2 +- 4 files changed, 43 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9048aa07d..152d56968 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,45 @@ # Changelog +## [1.40.1] — 2026-10-04 + +Restore opens older encrypted values again; faster pages on a long-used +server; a day of pulse is the mean of its hours; Apple Health imports leave +out HealthLog's own samples; more memory for the app by default. + +### Fixed + +- **Restore key probe.** A legacy unprefixed ciphertext was read as a base64 + string and refused a restore the key opens; legacy values open as strings, + probes come only from sections that count, binary probes use their AAD. + The drill checks every account of the newest date. +- **Daily pulse (#1107, migration 0372).** A day is the mean of its local + hours' means and a window the mean of its days, in every reader; stored + pulse rollups are recomputed. Count, min and max stay over every reading. +- **Apple Health import (#1109).** Records carrying HealthLog's origin + marker or a HealthLog id are left out and counted; cycle samples too. +- **Blood pressure series.** One source per day for both values. +- **Backups.** Heartbeat on both backup jobs; per-account attempts recorded + (migration 0373), interrupted accounts go last; the preview is built from + the writer's counts. +- **Document previews.** A permanent render failure is recorded (migration 0374) instead of re-queued at every start. +- **Umami.** Events from a tracker built with a custom endpoint are proxied + to the configured Umami host. +- **Display.** Rest Mode shows its start date; the favicon is transparent; + the sidebar no longer flashes scrollbars while it expands. + +### Changed + +- **Performance.** One Prisma client per process instead of one per bundle; + the rollup coverage probe uses a loose index scan and concurrent readers + share one probe; timezone offsets are memoised per hour; chart text is + measured outside ``; Insights mounts lower sections near the + viewport; a nav prefetch is not re-sent on mount; MCP single-metric reads + build only the requested metric; the backup schema loads on first use. +- **Lab list (#1057).** The range bar and name use a wide card. +- **Compose.** `APP_MEMORY_LIMIT` defaults to `1536m` (792 MB heap). +- Dependencies: next 16.3.8, eslint-config-next 16.3.8, zod-openapi 6.0.2, + testcontainers 12.2.0, p-limit 7.3.3. + ## [1.40.0] — 2026-10-04 Back up the encryption key and refuse a key that cannot open the data; diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index bb71a2e8e..54c7d32f5 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.40.0 + version: 1.40.1 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index 3655fa286..32ef4824a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.40.0", + "version": "1.40.1", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 32351a83e..0c7bb5af9 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.40.0"; + /* @sw-version-fallback */ "v1.40.1"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` From 4be7109a2b92c0029779281f3f3a7ffd184df3ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:13:28 +0200 Subject: [PATCH 2/9] fix(medications): refetch an invalidated prefetch on mount A hover prefetch excused the mount refetch even after a dose logged elsewhere had invalidated the medication reads, so the page showed the compliance from before the dose. An invalidated entry now always refetches. --- .../__tests__/prefetch-medications.test.ts | 22 +++++++++++++++++++ src/lib/queries/prefetch-medications.ts | 9 +++++++- 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/src/lib/queries/__tests__/prefetch-medications.test.ts b/src/lib/queries/__tests__/prefetch-medications.test.ts index 22ee0f750..dc004b368 100644 --- a/src/lib/queries/__tests__/prefetch-medications.test.ts +++ b/src/lib/queries/__tests__/prefetch-medications.test.ts @@ -125,6 +125,28 @@ describe("refetchMedicationsOnMount", () => { unsubscribe(); }); + it("refetches a prefetched read that was invalidated before the mount", async () => { + // Hover the nav link, then log a dose elsewhere before the page mounts: + // the dose invalidates the medication reads, and the prefetched answer + // no longer holds the compliance the user just changed. + const { client, calls } = setup(); + prefetchMedicationsList(client); + await vi.waitFor(() => expect(calls("/api/medications")).toBe(1)); + await vi.waitFor(() => + expect(client.getQueryState(queryKeys.medications())?.status).toBe( + "success", + ), + ); + await client.invalidateQueries({ queryKey: queryKeys.medications() }); + expect(client.getQueryState(queryKeys.medications())?.isInvalidated).toBe( + true, + ); + + const { refetched, unsubscribe } = await mountList(client); + expect(refetched).toBe(true); + unsubscribe(); + }); + it("still refetches on mount when the cache holds an older read", async () => { const { client } = setup(); const queryFn = vi.fn(async () => []); diff --git a/src/lib/queries/prefetch-medications.ts b/src/lib/queries/prefetch-medications.ts index 6058fc137..d78e63c52 100644 --- a/src/lib/queries/prefetch-medications.ts +++ b/src/lib/queries/prefetch-medications.ts @@ -93,11 +93,18 @@ export function prefetchMedicationsList( * the identical request a second time, a few hundred milliseconds after the * first answer landed, on every visit to the page. A return to the page with * an older cache entry, or a direct load, still refetches on mount. + * + * An invalidated entry always refetches: a dose logged elsewhere in the app + * between the prefetch and the mount invalidates these reads, and the + * prefetched answer is then the stale one. */ export function refetchMedicationsOnMount( read: MedicationsRead, -): (query: { state: { dataUpdatedAt: number } }) => boolean | "always" { +): (query: { + state: { dataUpdatedAt: number; isInvalidated: boolean }; +}) => boolean | "always" { return (query) => { + if (query.state.isInvalidated) return "always"; const startedAt = prefetchStartedAt.get(read); const answeredThisVisit = startedAt !== undefined && From 4261139dfae6f7318ab5c1c8b5629afbefd6dabf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:19:32 +0200 Subject: [PATCH 3/9] fix(backup): run the restore drill as a locked long pass The drill now reads every account's newest off-host object but still ran on the default fifteen-minute expiry, without a lock and with two retries, so a large instance could see it killed, run beside itself, and page every failed account three times. It now carries a four-hour expiry, stops between accounts on its budget and reports the objects it did not reach, holds a pass lock at the binding, and is not retried. --- src/lib/jobs/__tests__/restore-drill.test.ts | 32 +++++++++++++ src/lib/jobs/job-outcome.ts | 1 + src/lib/jobs/queue-runtime.ts | 10 +++- src/lib/jobs/reminder/register-maintenance.ts | 13 +++-- src/lib/jobs/restore-drill.ts | 47 +++++++++++++++++-- 5 files changed, 94 insertions(+), 9 deletions(-) diff --git a/src/lib/jobs/__tests__/restore-drill.test.ts b/src/lib/jobs/__tests__/restore-drill.test.ts index d223c5a45..81eb035ef 100644 --- a/src/lib/jobs/__tests__/restore-drill.test.ts +++ b/src/lib/jobs/__tests__/restore-drill.test.ts @@ -7,7 +7,9 @@ import { runRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "../restore-drill"; +import { QUEUE_RUNTIME } from "../queue-runtime"; vi.mock("../report-worker-error", () => ({ reportWorkerError: vi.fn().mockResolvedValue(undefined), @@ -86,6 +88,36 @@ describe("runRestoreDrill", () => { expect(RESTORE_DRILL_CRON).toBe("11 4 1 * *"); }); + it("runs as a locked long pass that pages each account once per run", () => { + // Every account's object is read, so the pg-boss default of fifteen + // minutes is no ceiling for it; no retry, so a failed account is paged + // once per run instead of three times. + expect(RESTORE_DRILL_SEND_OPTIONS.expireInSeconds).toBeGreaterThan(15 * 60); + expect(RESTORE_DRILL_SEND_OPTIONS.retryLimit).toBe(0); + expect(QUEUE_RUNTIME[RESTORE_DRILL_QUEUE]).toMatchObject({ + runtime: "long", + exclusive: "lockedPass", + }); + }); + + it("stops between accounts on its budget and counts the ones it did not reach", async () => { + const s3 = makeS3Mock({ + "2026-06-01/user-aaa.json.enc": backupObject({ userId: "user-aaa" }), + "2026-06-01/user-mmm.json.enc": backupObject({ userId: "user-mmm" }), + "2026-06-01/user-zzz.json.enc": backupObject({ userId: "user-zzz" }), + }); + let asked = 0; + const report = await runRestoreDrill( + s3, + new Date("2026-06-02T04:11:00Z"), + () => ++asked > 1, + ); + expect(report.accounts.map((a) => a.objectKey)).toEqual([ + "2026-06-01/user-aaa.json.enc", + ]); + expect(report.unchecked).toBe(2); + }); + it("fetches, decrypts, and parses the newest backup object", async () => { const s3 = makeS3Mock({ "2026-05-30/user-old.json.enc": backupObject(), diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 2ef5e5cc0..33e994a32 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -200,6 +200,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "restore_drill_medications", "restore_drill_mood_entries", "restore_drill_plaintext_bytes", + "restore_drill_unchecked", "restore_drill_stale", "restore_fail_cause", "restore_failed", diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 18177c274..3dd254b8f 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -36,6 +36,7 @@ import { DOCUMENT_AI_RUN_EXPIRE_SECONDS } from "@/lib/jobs/document-ai-run"; import { DATA_BACKUP_SEND_OPTIONS } from "@/lib/jobs/data-backup-policy"; import { OFFHOST_BACKUP_EXPIRE_SECONDS } from "@/lib/jobs/offhost-backup"; import { BACKUP_RESTORE_EXPIRE_SECONDS } from "@/lib/jobs/backup-restore"; +import { RESTORE_DRILL_EXPIRE_SECONDS } from "@/lib/jobs/restore-drill"; import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import-worker"; import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import"; @@ -267,7 +268,14 @@ export const QUEUE_RUNTIME: Readonly> = { exclusive: "lockedPass", why: "Uploads every account's record off-host.", }, - "data-restore-drill": short("Fetches and stream-parses one off-host object."), + "data-restore-drill": { + runtime: "long", + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + expiryVia: ["RESTORE_DRILL_SEND_OPTIONS"], + stop: { file: "lib/jobs/restore-drill.ts", fn: "handleRestoreDrill" }, + exclusive: "lockedPass", + why: "Fetches and stream-parses every account's newest off-host object.", + }, "offhost-backup-purge": short("At most 200 delete requests per run."), "backup-restore": { runtime: "long", diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 6fd906114..d774495d4 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -83,6 +83,7 @@ import { handleRestoreDrill, RESTORE_DRILL_CRON, RESTORE_DRILL_QUEUE, + RESTORE_DRILL_SEND_OPTIONS, } from "@/lib/jobs/restore-drill"; import { BACKUP_RESTORE_QUEUE, @@ -551,8 +552,10 @@ const allQueues = [ // a bulk DELETE or UPDATE that timed out will time out again on an immediate // retry, so pg-boss's default two would run the same doomed statement three // times a night. The queues without it are the ones whose failure is -// transient — the TLS probe, the off-host upload, the restore drill, the -// per-user environment fetch — and those still want the default retries. +// transient — the TLS probe, the off-host upload, the per-user environment +// fetch — and those still want the default retries. The restore drill sends +// `retryLimit: 0` in its own options for the other reason: it pages each +// failed account itself, and a retry would page the same accounts again. const schedules: ScheduleEntry[] = [ // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload) // that fans out one per-user environment fetch per opted-in account. @@ -571,7 +574,7 @@ const schedules: ScheduleEntry[] = [ cronIsTheRetry, ], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON, OFFHOST_BACKUP_SEND_OPTIONS], - [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON], + [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON, RESTORE_DRILL_SEND_OPTIONS], // Nightly backstop for the purge the deletion kicks straight away: a // request the bucket refused, or one written while the queue was down. [OFFHOST_PURGE_QUEUE, OFFHOST_PURGE_CRON, cronIsTheRetry], @@ -832,11 +835,13 @@ export async function registerMaintenanceQueues( { localConcurrency: 1, includeMetadata: true }, lockedPass(OFFHOST_BACKUP_QUEUE, () => WHOLE_PASS, handleOffhostBackup), ); + // Locked: a drill that outlives a restart or a manual run never reads the + // bucket beside another one. await createAndWork( boss, RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, - handleRestoreDrill, + lockedPass(RESTORE_DRILL_QUEUE, () => WHOLE_PASS, handleRestoreDrill), ); await createAndWork( boss, diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index caec85d00..1a6854bc1 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -21,6 +21,13 @@ * Schedule: 04:11 on the 1st of each month (`11 4 1 * *`), after the * 02:30 nightly upload and the 03:xx cleanup window, on a minute slot * no other cron uses. + * + * The drill reads every account's object, so it runs as a long pass: an + * explicit expiry (`RESTORE_DRILL_EXPIRE_SECONDS`), a stop between accounts + * on the job's budget, a lock so two drills never overlap (`lockedPass` at + * the binding), and no retry. A retry would re-read the same objects and + * page the same failed accounts a second and third time; each account is + * reported once per run, and the next drill is the retry. */ import type { Job } from "pg-boss"; import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan"; @@ -38,12 +45,28 @@ import { type S3Like, } from "@/lib/jobs/offhost-backup"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; +import { jobBudget } from "@/lib/jobs/job-budget"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; export const RESTORE_DRILL_QUEUE = "data-restore-drill"; export const RESTORE_DRILL_CRON = "11 4 1 * *"; +/** + * Four hours, the off-host upload's own window: the drill downloads and + * stream-parses every object that upload wrote in one night. + */ +export const RESTORE_DRILL_EXPIRE_SECONDS = 4 * 60 * 60; + +/** + * What the monthly schedule sends with: the expiry above and no retry + * (`retryLimit: 0`, the maintenance schedules' `cronIsTheRetry`). + */ +export const RESTORE_DRILL_SEND_OPTIONS = { + expireInSeconds: RESTORE_DRILL_EXPIRE_SECONDS, + retryLimit: 0, +} as const; + /** * A drill run that finds the newest backup older than this is reported * as a failure even when decrypt + parse succeed: the nightly uploader @@ -80,6 +103,8 @@ export interface RestoreDrillReport { accounts: RestoreDrillAccount[]; /** The accounts whose object did not pass. */ failed: RestoreDrillAccount[]; + /** Objects of the newest date the run stopped before, on its budget. */ + unchecked: number; } function countArray(value: unknown): number { @@ -178,6 +203,7 @@ async function drillObject( export async function runRestoreDrill( s3Override?: S3Like, now: Date = new Date(), + shouldStop: () => boolean = () => false, ): Promise { const cfg = loadOffhostConfig(); if (!cfg) { @@ -208,6 +234,8 @@ export async function runRestoreDrill( const accounts: RestoreDrillAccount[] = []; for (const objectKey of newest) { + // Between accounts, never inside one: the job's budget is spent. + if (shouldStop()) break; const account: RestoreDrillAccount = { objectKey, ok: true, @@ -239,6 +267,7 @@ export async function runRestoreDrill( stale: ageDays > MAX_BACKUP_AGE_DAYS, accounts, failed: accounts.filter((a) => !a.ok), + unchecked: newest.length - accounts.length, }; } @@ -252,10 +281,10 @@ export async function runRestoreDrill( export async function handleRestoreDrill( jobs: Job[], ): Promise { - void jobs; + const shouldStop = jobBudget(jobs); return withBackgroundEvent("job.restore_drill", async (evt) => { try { - const report = await runRestoreDrill(); + const report = await runRestoreDrill(undefined, new Date(), shouldStop); const total = (pick: (a: RestoreDrillAccount) => number) => report.accounts.reduce((sum, a) => sum + pick(a), 0); const meta = { @@ -270,6 +299,7 @@ export async function handleRestoreDrill( restore_drill_intake_events: total((a) => a.recordCounts.intakeEvents), restore_drill_mood_entries: total((a) => a.recordCounts.moodEntries), restore_drill_stale: report.stale, + restore_drill_unchecked: report.unchecked, }; for (const [key, value] of Object.entries(meta)) evt.addMeta(key, value); evt.addMeta( @@ -301,9 +331,18 @@ export async function handleRestoreDrill( { dateKey: report.dateKey, ageDays: report.ageDays }, ); } + if (report.unchecked > 0) { + await reportWorkerError( + RESTORE_DRILL_QUEUE, + new Error( + `Restore drill stopped on its time budget with ${report.unchecked} of ${report.accounts.length + report.unchecked} backup objects unchecked.`, + ), + { dateKey: report.dateKey, unchecked: report.unchecked }, + ); + } if (report.failed.length > 0) { - // Already paged above, per account. A retry reads the same objects - // and cannot change the verdict before the next upload. + // Already paged above, per account. The queue does not retry: a + // retry reads the same objects and would page the same accounts again. return jobFailed( `restore drill: ${report.failed.length} of ${report.accounts.length} accounts failed`, new Error(report.failed.map((a) => a.objectKey).join(", ")), From da56177290c198a02cbd86c0fe01895adacccdb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:21:32 +0200 Subject: [PATCH 4/9] fix(backup): take an interrupted account last for one run only An account whose last backup attempt never came back was kept behind every other account for as long as its attempt stayed open. A pass that regularly stops on its time budget never reached it, so it never got a copy again. It is now held back only until a later run has started on another account; if its record kills the process again, its attempt is the newest once more and it goes last again. A crash loop still leaves the other accounts their copies, and the account itself is attempted at least every second run. --- .../__tests__/backup-pass-attempts.test.ts | 118 ++++++++++++++++++ src/lib/jobs/backup-pass-attempts.ts | 56 ++++++++- src/lib/jobs/offhost-backup.ts | 9 +- src/lib/jobs/reminder/backup-handlers.ts | 8 +- 4 files changed, 183 insertions(+), 8 deletions(-) create mode 100644 src/lib/jobs/__tests__/backup-pass-attempts.test.ts diff --git a/src/lib/jobs/__tests__/backup-pass-attempts.test.ts b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts new file mode 100644 index 000000000..722fec5ff --- /dev/null +++ b/src/lib/jobs/__tests__/backup-pass-attempts.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, it } from "vitest"; + +import { + heldBackAccounts, + orderInterruptedLast, +} from "../backup-pass-attempts"; + +type Row = { userId: string; startedAt: Date; finishedAt: Date | null }; + +/** + * A weekly-style pass over `ids`: oldest copy first, held-back accounts last, + * at most `budget` accounts per run, `kills` the accounts whose record kills + * the process. Returns the accounts copied in each run. + */ +function simulate(opts: { + ids: string[]; + budget: number; + kills: Set; + runs: number; + interruptedFirst?: string; +}): string[][] { + let clock = 0; + const attempts = new Map(); + const copiedAt = new Map(); + if (opts.interruptedFirst) { + attempts.set(opts.interruptedFirst, { + userId: opts.interruptedFirst, + startedAt: new Date(++clock), + finishedAt: null, + }); + } + const copied: string[][] = []; + for (let run = 0; run < opts.runs; run += 1) { + const order = orderInterruptedLast( + [...opts.ids] + .map((id) => ({ id })) + .sort( + (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1), + ), + heldBackAccounts([...attempts.values()]), + ); + const thisRun: string[] = []; + for (const { id } of order.slice(0, opts.budget)) { + const row = attempts.get(id) ?? { + userId: id, + startedAt: new Date(0), + finishedAt: null, + }; + row.startedAt = new Date(++clock); + attempts.set(id, row); + if (opts.kills.has(id)) break; // the process dies; the run ends here + row.finishedAt = new Date(++clock); + copiedAt.set(id, clock); + thisRun.push(id); + } + copied.push(thisRun); + } + return copied; +} + +describe("heldBackAccounts", () => { + const at = (ms: number) => new Date(ms); + + it("holds back an interrupted attempt that is the newest of the pass", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: at(10) }, + { userId: "a", startedAt: at(20), finishedAt: at(21) }, + ]); + expect([...held.keys()]).toEqual(["x"]); + }); + + it("releases it once a later run has started on another account", () => { + const held = heldBackAccounts([ + { userId: "x", startedAt: at(30), finishedAt: null }, + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); + + it("ignores finished attempts", () => { + const held = heldBackAccounts([ + { userId: "a", startedAt: at(40), finishedAt: at(41) }, + ]); + expect(held.size).toBe(0); + }); +}); + +describe("the order across runs", () => { + const ids = ["x", "a", "b", "c", "d"]; + + it("keeps a record that kills the process from costing the others their copies", () => { + // `x` has the oldest copy and kills every run that reaches it. + const runs = simulate({ ids, budget: 99, kills: new Set(["x"]), runs: 6 }); + // The first run meets it unwarned; from then on it goes last every run, + // because each crash makes its attempt the newest again. + expect(runs[0]).toEqual([]); + for (const run of runs.slice(1)) { + expect([...run].sort()).toEqual(["a", "b", "c", "d"]); + } + }); + + it("still copies an interrupted account when every run stops on its budget", () => { + // `x` was interrupted once (a restart, not its record). Every run stops + // after three accounts, so kept last for good it would never be reached. + const runs = simulate({ + ids, + budget: 3, + kills: new Set(), + runs: 4, + interruptedFirst: "x", + }); + const firstCopy = runs.findIndex((run) => run.includes("x")); + expect(firstCopy).toBeGreaterThanOrEqual(0); + expect(firstCopy).toBeLessThanOrEqual(1); + // And every account gets a copy within the four runs. + expect(new Set(runs.flat())).toEqual(new Set(ids)); + }); +}); diff --git a/src/lib/jobs/backup-pass-attempts.ts b/src/lib/jobs/backup-pass-attempts.ts index 05c0d919b..389252334 100644 --- a/src/lib/jobs/backup-pass-attempts.ts +++ b/src/lib/jobs/backup-pass-attempts.ts @@ -16,6 +16,16 @@ * takes that account last, after everybody else has their copy, and the admin * backups page names it. Every write and read here is best effort: the marker * is bookkeeping about a backup, and must never be the reason one fails. + * + * Last only once. A pass that regularly stops on its time budget never reaches + * the end of its list, so an account kept last for as long as its attempt + * stays open would never be attempted again, and would never get a copy. An + * interrupted account is therefore held back only until a later run has + * started on another account (`accountsToTakeLast`): from then on it takes its + * ordinary place again. If it kills the process again, its new attempt is the + * newest one and it goes last once more. So a crash loop costs the other + * accounts at most every second run, and the account itself is attempted at + * least every second run. */ import type { PrismaClient } from "@/generated/prisma/client"; @@ -83,7 +93,51 @@ export async function readInterruptedBackupAttempts( } /** - * `accounts` in their order, with every interrupted one moved behind the + * The interrupted accounts a pass should take last: those whose interrupted + * attempt is still the newest attempt of the pass, i.e. no later run has + * started on another account since. Rows are one per account. Pure, so the + * rule can be tested apart from the database. + */ +export function heldBackAccounts( + rows: ReadonlyArray<{ + userId: string; + startedAt: Date; + finishedAt: Date | null; + }>, +): Map { + const out = new Map(); + for (const row of rows) { + const open = row.finishedAt === null || row.finishedAt < row.startedAt; + if (!open) continue; + const passedOver = rows.some( + (other) => other.userId !== row.userId && other.startedAt > row.startedAt, + ); + if (!passedOver) out.set(row.userId, row.startedAt); + } + return out; +} + +/** + * What a pass orders by: the interrupted accounts it takes last + * (`heldBackAccounts`). Empty when the table cannot be read. + */ +export async function readAccountsToTakeLast( + prisma: Db, + pass: BackupPass, +): Promise> { + try { + const rows = await prisma.backupPassAttempt.findMany({ + where: { pass }, + select: { userId: true, startedAt: true, finishedAt: true }, + }); + return heldBackAccounts(rows); + } catch { + return new Map(); + } +} + +/** + * `accounts` in their order, with every held-back one moved behind the * rest; among those, the one interrupted longest ago first. */ export function orderInterruptedLast( diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index ad9cf04dd..05a9f23b5 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -39,7 +39,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { BACKUP_HEARTBEAT_SECONDS } from "@/lib/jobs/data-backup-policy"; import { annotate, getEvent } from "@/lib/logging/context"; @@ -930,14 +930,15 @@ export async function runOffhostBackup( // By id, with an account whose last walk never came back behind the rest: // a record that killed the process once would otherwise lead every retry - // and keep every account after it out of the bucket - // (`backup-pass-attempts.ts`). + // and keep every account after it out of the bucket. Behind them for one + // run only, so a pass that stops on its budget still reaches it the run + // after (`backup-pass-attempts.ts`). const users = orderInterruptedLast( await prisma.user.findMany({ select: { id: true }, orderBy: { id: "asc" }, }), - await readInterruptedBackupAttempts(prisma, OFFHOST_BACKUP_QUEUE), + await readAccountsToTakeLast(prisma, OFFHOST_BACKUP_QUEUE), ); const doneThisRun = new Set(); if (runStartedAt) { diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index d4bb4b1a4..62e1adeda 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -27,7 +27,7 @@ import { markBackupAttemptFinished, markBackupAttemptStarted, orderInterruptedLast, - readInterruptedBackupAttempts, + readAccountsToTakeLast, } from "@/lib/jobs/backup-pass-attempts"; import { DATA_BACKUP_QUEUE } from "@/lib/jobs/data-backup-policy"; @@ -154,7 +154,7 @@ export async function handleDataBackup( where: { type: "WEEKLY_AUTO" }, select: { userId: true, createdAt: true }, }), - readInterruptedBackupAttempts(prisma, DATA_BACKUP_QUEUE), + readAccountsToTakeLast(prisma, DATA_BACKUP_QUEUE), ]); const copiedAt = new Map( copies.map((copy) => [copy.userId, copy.createdAt.getTime()]), @@ -163,7 +163,9 @@ export async function handleDataBackup( // attempt never came back goes behind all of them: it is the one with // the oldest copy, so it would lead every retry, and a record that // killed the process once would keep every account after it from a - // copy (`backup-pass-attempts.ts`). + // copy. Only until a later run has started on another account, so a + // pass that stops on its budget still reaches it the run after + // (`backup-pass-attempts.ts`). const users = orderInterruptedLast( [...accounts].sort( (a, b) => (copiedAt.get(a.id) ?? -1) - (copiedAt.get(b.id) ?? -1), From 2bbcce73e4fd901364af30d5a5f4f3c0ce0a6e97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:25:28 +0200 Subject: [PATCH 5/9] docs: name the memory cap on small hosts and the shared connection pool Getting started names the APP_MEMORY_LIMIT default and how to lower it on a 2 GB host; the scaling guide says that web and worker share one pool in the bundled single-container setup. The rollup comment names migration 0372, which carries the hourly-mean statistic. --- docs/self-hosting/getting-started.md | 7 ++++++- docs/self-hosting/scaling.md | 6 +++++- src/lib/rollups/measurement-rollups.ts | 2 +- 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/docs/self-hosting/getting-started.md b/docs/self-hosting/getting-started.md index 88b88f0eb..c55033afd 100644 --- a/docs/self-hosting/getting-started.md +++ b/docs/self-hosting/getting-started.md @@ -13,7 +13,12 @@ already installed, longer if a TLS-fronted public hostname is in scope unchanged. - **2 GB RAM, 10 GB free disk.** The Postgres data volume grows with measurement history; the Apple Health import worker can briefly - hold a 1.5 GB upload in `/tmp` while it parses. + hold a 1.5 GB upload in `/tmp` while it parses. The app container is + capped at `APP_MEMORY_LIMIT`, `1536m` by default (Node then sizes its + heap at about 792 MB). On a 2 GB host that leaves too little for + Postgres and the system, so set `APP_MEMORY_LIMIT=1g` in `.env` + (a heap of about 524 MB); a long-used account then runs closer to + that heap, and the weekly backup with it. - **Outbound HTTPS** to `ghcr.io` (image pulls) plus whichever integration endpoints you plan to enable (Withings, OpenAI, etc.). The opt-in environmental-context module additionally reaches diff --git a/docs/self-hosting/scaling.md b/docs/self-hosting/scaling.md index 8bde98d1b..6776d44a3 100644 --- a/docs/self-hosting/scaling.md +++ b/docs/self-hosting/scaling.md @@ -163,7 +163,11 @@ the knob as `DB_CONNECTION_LIMIT`, baked into `DATABASE_URL` as `connection_limit`; `DATABASE_POOL_MAX` remains a working explicit override. Resolution order: `DB_CONNECTION_LIMIT`, then `DATABASE_POOL_MAX`, then the `connection_limit` URL parameter, then -the default of 20. +the default of 20. In the bundled single-container setup +(`HEALTHLOG_PROCESS_TYPE=all`) the web requests and the background +worker run in one process and share that one Prisma pool, so a budget +sized for the web traffic alone leaves the worker's jobs queueing for +the same connections. Plan total Postgres slots as **container_count × budget**. A stock Postgres 16 container ships with `max_connections = 100`, so diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index df8e52bb2..8f6643007 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -597,7 +597,7 @@ async function runRollupAggregate( * every reading, so the band and the windowed slope are untouched. * * Hours and days are UTC, matching the UTC day this tier buckets by; migration - * 0373 applies the same statistic to rows stored before it. + * 0372 applies the same statistic to rows stored before it. */ async function applyHourlyMeanDay( db: Pick, From 3681b23d30fbe591cd11ccd0b0ee189dcd48210d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 21:36:20 +0200 Subject: [PATCH 6/9] test(e2e): mount every overview section before asserting none paints The AI-optional overview and account journeys asserted zero error cards, Coach hand-offs and briefings without scrolling the deferred sections into view, so the sections below the first three never mounted and the zero counts held whatever they would paint. Both journeys now reveal every section and wait for the reads those sections fire to land; waiting on networkidle again resolves at once, because the page reached that state before. --- e2e/ai-optional-account.spec.ts | 17 ++++++++++----- e2e/ai-optional-overview.spec.ts | 13 +++++++++++ e2e/utils/deferred-sections.ts | 37 ++++++++++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 5 deletions(-) diff --git a/e2e/ai-optional-account.spec.ts b/e2e/ai-optional-account.spec.ts index 3e7f138fe..6a27a80cd 100644 --- a/e2e/ai-optional-account.spec.ts +++ b/e2e/ai-optional-account.spec.ts @@ -25,6 +25,7 @@ import { seedStoredBriefing, } from "./setup/ai-optional-fixture"; import { expect, test } from "./setup/test"; +import { revealDeferredSectionsSettled } from "./utils/deferred-sections"; test.use({ storageState: AI_OPTIONAL_STORAGE_STATE_PATH }); test.describe.configure({ mode: "serial" }); @@ -36,12 +37,18 @@ test.afterAll(async () => { await resetAiOptionalAccount(); }); +/** + * The overview with every section mounted. Sections past the first three + * render as an empty sentinel until they near the viewport, so a count of + * zero taken before they mount would pass whatever they paint. + */ async function openOverview(page: Page): Promise { await page.goto("/insights"); await expect(page.locator('[data-slot="insights-hero-strip"]')).toBeVisible({ timeout: 20_000, }); await page.waitForLoadState("networkidle"); + await revealDeferredSectionsSettled(page); } test("J5: withdrawing AI consent deletes the notes written under it", async ({ @@ -80,12 +87,12 @@ test("J4: the Coach memory stays readable and deletable with the Coach hidden", const title = `Stored conversation ${Date.now()}`; await seedHiddenCoachWithConversation(title); - // Hide Coach takes the Coach page and its launcher away... - await page.goto("/insights"); - await expect(page.locator('[data-slot="insights-hero-strip"]')).toBeVisible({ - timeout: 20_000, - }); + // Hide Coach takes the Coach page, its launcher and every per-card + // hand-off away... + await openOverview(page); await expect(page.locator('[data-slot="coach-fab"]')).toHaveCount(0); + await expect(page.locator('[data-slot="ask-coach-action"]')).toHaveCount(0); + await expect(page.locator('[data-slot="ask-coach-icon"]')).toHaveCount(0); // ...and Settings, AI lists what the Coach stored. await page.goto("/settings/ai"); diff --git a/e2e/ai-optional-overview.spec.ts b/e2e/ai-optional-overview.spec.ts index dac4533bd..ba0158dd0 100644 --- a/e2e/ai-optional-overview.spec.ts +++ b/e2e/ai-optional-overview.spec.ts @@ -21,9 +21,19 @@ import type { Page, Request, Response } from "@playwright/test"; import { STORAGE_STATE_PATH } from "./setup/global-setup"; import { aiBlockUnavailable, serveAiBlock } from "./setup/ai-capabilities"; import { expect, test } from "./setup/test"; +import { revealDeferredSectionsSettled } from "./utils/deferred-sections"; test.use({ storageState: STORAGE_STATE_PATH }); +/** + * Mount every overview section, then let their reads land. Sections past the + * first three render as an empty sentinel until they near the viewport, so a + * count of zero taken before this would pass whatever those sections paint. + */ +async function mountWholeOverview(page: Page): Promise { + await revealDeferredSectionsSettled(page); +} + /** Routes that call a model or serve model-written text. */ const AI_ROUTE = new RegExp( [ @@ -89,6 +99,7 @@ test.describe("J1: AI switched off by the operator", () => { ); // Let every section's reads land before judging what painted. await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); // A read refused for want of AI paints one of these on the overview: // the section boundary, the error card, or the dense error row. @@ -137,6 +148,7 @@ test.describe("J3: no AI provider", () => { timeout: 20_000, }); await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); await expect(page.locator('[data-slot="coach-fab"]')).toHaveCount(0); await expect(page.locator('[data-slot="daily-briefing"]')).toHaveCount(0); // The per-card Coach hand-offs paint only while the Coach is available. @@ -163,6 +175,7 @@ test.describe("J3: no AI provider", () => { { timeout: 20_000 }, ); await page.waitForLoadState("networkidle"); + await mountWholeOverview(page); await expect(page.locator('[data-slot="ai-setup-hint"]')).toHaveCount(0); expect(network.aiRequests).toEqual([]); diff --git a/e2e/utils/deferred-sections.ts b/e2e/utils/deferred-sections.ts index 0ca4748ea..247ab9d80 100644 --- a/e2e/utils/deferred-sections.ts +++ b/e2e/utils/deferred-sections.ts @@ -36,3 +36,40 @@ export async function revealDeferredSections(page: Page): Promise { .toBe(0); await page.evaluate(() => document.querySelector("main")?.scrollTo(0, 0)); } + +/** + * `revealDeferredSections`, then wait until the reads the newly mounted + * sections fired have landed: no request in flight for 750 ms. A + * `waitForLoadState("networkidle")` cannot do this: the page reached that + * state once already, so it resolves at once, and a count of zero taken then + * passes before a deferred section has painted anything. + */ +export async function revealDeferredSectionsSettled(page: Page): Promise { + let inFlight = 0; + const started = () => { + inFlight += 1; + }; + const ended = () => { + inFlight -= 1; + }; + page.on("request", started); + page.on("requestfinished", ended); + page.on("requestfailed", ended); + try { + await revealDeferredSections(page); + let quietSince = Date.now(); + await expect + .poll( + () => { + if (inFlight > 0) quietSince = Date.now(); + return Date.now() - quietSince >= 750; + }, + { timeout: 20_000, intervals: [100] }, + ) + .toBe(true); + } finally { + page.off("request", started); + page.off("requestfinished", ended); + page.off("requestfailed", ended); + } +} From 090e3d9a7ee159cb74665f06b7a8bf0102d85b41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 22:14:14 +0200 Subject: [PATCH 7/9] perf(measurements): weigh pulse days with window functions over one scan The day weight joined its source to an hour grouping of itself. With stale or missing planner statistics, as after a new account imports a year of pulse below the autoanalyze threshold, the planner ran that join as a nested loop and the all-time pulse aggregate ran into the 60 s statement timeout. The weight is now two window passes over one scan: the readings in a row's hour, and one marker per hour summed over the day for the hour count. The weights are unchanged. A new integration test seeds over 100k pulse rows with autovacuum off and no ANALYZE, bounds the all-time reader and a zoned per-source mean well under the timeout, and compares every row's weight against the earlier self-join after ANALYZE. The ANALYZE the metric-series snapshot test needed for the same trap is gone. --- src/lib/measurements/day-mean.ts | 46 ++-- .../day-weight-unanalysed-scale.test.ts | 232 ++++++++++++++++++ .../mcp-metric-series-source-snapshot.test.ts | 11 +- 3 files changed, 258 insertions(+), 31 deletions(-) create mode 100644 tests/integration/day-weight-unanalysed-scale.test.ts diff --git a/src/lib/measurements/day-mean.ts b/src/lib/measurements/day-mean.ts index 38b2e2970..9d90908bc 100644 --- a/src/lib/measurements/day-mean.ts +++ b/src/lib/measurements/day-mean.ts @@ -245,34 +245,32 @@ function weightedRowsSql( if (!/^[a-z_][a-z0-9_]*$/.test(source)) { throw new Error(`source must be a CTE name: ${source}`); } - const hourOf = (alias: string) => - `date_trunc('hour', ${frameTs(frame, `${alias}."measured_at"`)})`; - const dayOf = (alias: string) => - `date_trunc('day', ${frameTs(frame, `${alias}."measured_at"`)})`; - const src = bySource ? `, t."source"` : ""; - const part = bySource ? `, x."source"` : ""; - const join = bySource ? ` AND hw."source" = s."source"` : ""; + // One scan of the CTE and window functions only. The earlier shape joined + // the CTE to an hour grouping of itself; with stale or missing statistics + // (a fresh import below the autoanalyze threshold) the planner ran that + // join as a nested loop and a year of pulse ran into the statement timeout. + // + // `dm_hour_n` counts the readings in a row's hour and `dm_hour_first` marks + // one row per hour, so the sum of `dm_hour_first` over a day is the day's + // hour count. Both columns ride along on the row; every caller aggregates. + const partition = (alias: string, unit: "hour" | "day") => { + const keys = bySource + ? `${alias}."type", ${alias}."source"` + : `${alias}."type"`; + const ts = frameTs(frame, `${alias}."measured_at"`); + return `PARTITION BY ${keys}, date_trunc('${unit}', ${ts})`; + }; return `( SELECT s.*, (CASE WHEN ${isHourlyMeanTypeSql('s."type"')} - THEN 1.0 / (hw.n * hw.hours) + THEN 1.0 / (s.dm_hour_n * SUM(s.dm_hour_first) OVER (${partition("s", "day")})) ELSE 1.0 END)::double precision AS day_weight - FROM ${source} s - LEFT JOIN ( - SELECT x."type"${part}, x.local_hour, x.n, - COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours - FROM ( - SELECT t."type"${src}, - ${hourOf("t")} AS local_hour, - ${dayOf("t")} AS local_day, - COUNT(*) AS n - FROM ${source} t - WHERE ${isHourlyMeanTypeSql('t."type"')} - GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} - ) x - ) hw - ON hw."type" = s."type"${join} - AND hw.local_hour = ${hourOf("s")} + FROM ( + SELECT t.*, + COUNT(*) OVER (${partition("t", "hour")}) AS dm_hour_n, + (ROW_NUMBER() OVER (${partition("t", "hour")}) = 1)::int AS dm_hour_first + FROM ${source} t + ) s )`; } diff --git a/tests/integration/day-weight-unanalysed-scale.test.ts b/tests/integration/day-weight-unanalysed-scale.test.ts new file mode 100644 index 000000000..977f32575 --- /dev/null +++ b/tests/integration/day-weight-unanalysed-scale.test.ts @@ -0,0 +1,232 @@ +/** + * The shared pulse day weight (`dayWeightedRows`) must not depend on planner + * statistics. A new account importing a year of pulse lands below the + * autoanalyze threshold, so the first reads run against a table the planner + * believes is empty. The earlier shape joined the CTE to an hour grouping of + * itself; on such a table the planner chose a nested loop and the all-time + * pulse aggregate ran into the 60 s production statement timeout. + * + * This file seeds a year of dense pulse (well over 100k rows) with autovacuum + * off for the table and no ANALYZE, then: + * + * - times the real all-time reader (`readAllTimeExtremes`) and a bare + * day-weighted mean, each well under the timeout; + * - after ANALYZE, compares the weights per day, per frame and per source + * against the earlier self-join SQL kept below as the oracle. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import type { Prisma } from "@/generated/prisma/client"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { + SESSION_DAY_FRAME, + dayWeightedRows, + zoneDayFrame, + type DayFrame, +} from "@/lib/measurements/day-mean"; + +const prisma = getPrismaClient(); +const USER = "day-weight-scale"; +const DAY = 86_400_000; +const MINUTE = 60_000; +const DAYS = 365; +/** Generous against a 60 s production timeout, far above what one scan takes. */ +const BOUND_MS = 15_000; + +const start = + Date.UTC( + new Date().getUTCFullYear(), + new Date().getUTCMonth(), + new Date().getUTCDate(), + ) - + DAYS * DAY; + +/** + * About 330 readings a day, unevenly spread so hours carry different counts, + * plus a second source on every fifth day so the per-source weights differ. + */ +function seedRows(): Prisma.MeasurementCreateManyInput[] { + const rows: Prisma.MeasurementCreateManyInput[] = []; + let i = 0; + for (let d = 0; d < DAYS; d += 1) { + for (let m = 0; m < 24 * 60; m += 4) { + if ((m + d) % 13 === 0) continue; + const hour = Math.floor(m / 60); + // Night hours sparse, day hours dense. + if (hour < 6 && m % 12 !== 0) continue; + rows.push({ + id: `${USER}-a-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "APPLE_HEALTH", + value: 55 + ((i * 7 + d) % 50), + measuredAt: new Date(start + d * DAY + m * MINUTE + (i % 3) * 1000), + }); + i += 1; + } + if (d % 5 === 0) { + for (const h of [7, 7, 13, 22]) { + rows.push({ + id: `${USER}-w-${i}`, + userId: USER, + type: "PULSE", + unit: "bpm", + source: "WITHINGS", + value: 60 + (i % 20), + measuredAt: new Date( + start + d * DAY + h * 3_600_000 + (i % 50) * MINUTE, + ), + }); + i += 1; + } + } + } + return rows; +} + +/** The earlier self-join, kept verbatim as the oracle for the weights. */ +function oracleWeightedRows( + source: string, + frame: DayFrame, + bySource: boolean, +): string { + const ts = (column: string) => + frame.kind === "session" + ? column + : `((${column} AT TIME ZONE 'UTC') AT TIME ZONE ${frame.tzSql})`; + const hourOf = (alias: string) => + `date_trunc('hour', ${ts(`${alias}."measured_at"`)})`; + const dayOf = (alias: string) => + `date_trunc('day', ${ts(`${alias}."measured_at"`)})`; + const src = bySource ? `, t."source"` : ""; + const part = bySource ? `, x."source"` : ""; + const join = bySource ? ` AND hw."source" = s."source"` : ""; + return `( + SELECT s.*, + (CASE WHEN (s."type")::text IN ('PULSE') + THEN 1.0 / (hw.n * hw.hours) + ELSE 1.0 END)::double precision AS day_weight + FROM ${source} s + LEFT JOIN ( + SELECT x."type"${part}, x.local_hour, x.n, + COUNT(*) OVER (PARTITION BY x."type"${part}, x.local_day) AS hours + FROM ( + SELECT t."type"${src}, + ${hourOf("t")} AS local_hour, + ${dayOf("t")} AS local_day, + COUNT(*) AS n + FROM ${source} t + WHERE (t."type")::text IN ('PULSE') + GROUP BY ${bySource ? "1, 2, 3, 4" : "1, 2, 3"} + ) x + ) hw + ON hw."type" = s."type"${join} + AND hw.local_hour = ${hourOf("s")} + )`; +} + +const SRC_CTE = `WITH src AS ( + SELECT m."id", m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = $1 AND m."deleted_at" IS NULL AND m."type" = 'PULSE' +)`; + +async function timed(fn: () => Promise): Promise<[T, number]> { + const t0 = performance.now(); + const out = await fn(); + return [out, performance.now() - t0]; +} + +let rowCount = 0; + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements SET (autovacuum_enabled = false)", + ); + await prisma.user.create({ data: { id: USER, username: USER } }); + const rows = seedRows(); + rowCount = rows.length; + for (let k = 0; k < rows.length; k += 20_000) { + await prisma.measurement.createMany({ data: rows.slice(k, k + 20_000) }); + } +}, 180_000); + +afterAll(async () => { + await prisma.$executeRawUnsafe( + "ALTER TABLE measurements RESET (autovacuum_enabled)", + ); +}); + +describe("day weight on an unanalysed table", () => { + it("seeds a year of dense pulse", () => { + expect(rowCount).toBeGreaterThan(100_000); + }); + + it("answers the all-time pulse aggregate well under the timeout", async () => { + const [out, ms] = await timed(() => readAllTimeExtremes(USER, ["PULSE"])); + console.info(`[day-weight] readAllTimeExtremes ${ms.toFixed(0)} ms`); + expect(out.get("PULSE")?.mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); + + it("answers a bare day-weighted mean well under the timeout", async () => { + const [rows, ms] = await timed(() => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT (SUM(w."value" * w.day_weight) / SUM(w.day_weight))::double precision AS mean + FROM ${dayWeightedRows("src", zoneDayFrame("$2"), { bySource: true })} w`, + USER, + "Europe/Berlin", + ), + ); + console.info(`[day-weight] zone + bySource mean ${ms.toFixed(0)} ms`); + expect(rows[0].mean).toBeGreaterThan(55); + expect(ms).toBeLessThan(BOUND_MS); + }); +}); + +describe("day weight against the earlier self-join", () => { + beforeAll(async () => { + // Statistics now, so the oracle plans the way it did in production. + await prisma.$executeRawUnsafe("ANALYZE measurements"); + }, 120_000); + + const cases: Array<[string, DayFrame, boolean, unknown[]]> = [ + ["session frame", SESSION_DAY_FRAME, false, []], + ["session frame by source", SESSION_DAY_FRAME, true, []], + ["Berlin frame", zoneDayFrame("$2"), false, ["Europe/Berlin"]], + ["Kolkata frame by source", zoneDayFrame("$2"), true, ["Asia/Kolkata"]], + ]; + + it.each(cases)( + "%s gives every row the same weight", + async (_label, frame, bySource, extra) => { + const query = (sub: string) => + prisma.$queryRawUnsafe>( + `${SRC_CTE} + SELECT w."id" AS id, w.day_weight AS w + FROM ${sub} w + ORDER BY w."id"`, + USER, + ...extra, + ); + const fresh = await query(dayWeightedRows("src", frame, { bySource })); + const oracle = await query(oracleWeightedRows("src", frame, bySource)); + expect(fresh.length).toBe(rowCount); + expect(fresh.length).toBe(oracle.length); + let mismatches = 0; + for (let k = 0; k < fresh.length; k += 1) { + if ( + fresh[k].id !== oracle[k].id || + Math.abs(fresh[k].w - oracle[k].w) > 1e-12 + ) { + mismatches += 1; + } + } + expect(mismatches).toBe(0); + }, + ); +}); diff --git a/tests/integration/mcp-metric-series-source-snapshot.test.ts b/tests/integration/mcp-metric-series-source-snapshot.test.ts index 3759189dc..ab95bc21a 100644 --- a/tests/integration/mcp-metric-series-source-snapshot.test.ts +++ b/tests/integration/mcp-metric-series-source-snapshot.test.ts @@ -218,13 +218,10 @@ beforeAll(async () => { // The mood feature warms its rollup tier fire-and-forget on first read; // settle it now so both builds read the same tier. await ensureUserMoodRollupsFresh(userId); - // Planner statistics for the rows just written. Without them the - // all-time pulse aggregate (`readAllTimeExtremes`, day-weighted for pulse) - // plans against an empty table and runs for ~14 s per call until - // autovacuum's analyze happens to land, which on a loaded runner took the - // first case past its 60 s budget. A production table always has - // statistics, so this measures what production runs. - await prisma.$executeRawUnsafe("ANALYZE measurements, measurement_rollups"); + // No ANALYZE here, on purpose: a fresh import below the autoanalyze + // threshold is what production runs too, and the all-time pulse aggregate + // must not depend on planner statistics + // (`day-weight-unanalysed-scale.test.ts`). }, 120_000); describe("get_metric_series built from one source's rows", () => { From 48ed149ef0229a6d742c226f157853d89e0b6e25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 22:15:48 +0200 Subject: [PATCH 8/9] fix(import): leave out a marked Apple Health sample only when this account holds it A sample carrying the HealthLog origin marker was always dropped from an export import. After a move to a new instance without a backup, the export is how the mirrored values come back, and every one of them was silently left out. A marked sample now counts as written by HealthLog only when its HKExternalUUID names a row of this account of the same kind, deleted rows included; a marked cycle sample only when the account holds a day-log on its day. Every other marked sample is imported, and still meets the manual-value match. A correlation's own HKExternalUUID now covers records that carry none. --- src/lib/apple-health/own-origin.ts | 10 +- .../import-apple-health-export.ts | 243 +++++++++++++----- .../export-own-origin.synthetic.xml | 5 +- .../apple-health-import-own-samples.test.ts | 193 ++++++++++++-- 4 files changed, 367 insertions(+), 84 deletions(-) diff --git a/src/lib/apple-health/own-origin.ts b/src/lib/apple-health/own-origin.ts index 34f6f2f8c..1a60a05aa 100644 --- a/src/lib/apple-health/own-origin.ts +++ b/src/lib/apple-health/own-origin.ts @@ -4,9 +4,13 @@ * from the app, and any Withings, import or manual row the server-to-Health * mirror writes. Builds from before the marker existed wrote the id alone. * - * Such a sample is already in the database under its own source. When the same - * sample comes back through an Apple Health export it is a second copy of that - * reading, so the export import leaves it out. + * Such a sample is usually already in the database under its own source. When + * it comes back through an Apple Health export it is a second copy of that + * reading, so the export import leaves it out, but only once its + * `HKExternalUUID` names a row of the importing account (a cycle sample: once + * that account holds a day-log on its day). The marker alone says HealthLog + * wrote the sample, not that this instance still holds it: after a move to a + * new instance without a backup, the export is how those values come back. * * This is not the predicate the app's own sync uses. The app decides "our own * echo" from `HKExternalUUID` plus the sample's authoring source being the app diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index dc280d285..f75e2a53e 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -122,11 +122,14 @@ export interface ImportJobResult { clinical: { skipped: number }; /** * Records left out because HealthLog wrote them into Apple Health itself and - * already holds them under their own source, each counted once however often - * the export repeats it: `byMarker` carry the app's origin marker, - * `byExternalId` carry a HealthLog row id as `HKExternalUUID` (samples from - * before the marker existed), `matchedManual` match a manual entry (same type - * and value, within 2 s) without either. + * this account already holds them under their own source, each counted once + * however often the export repeats it: `byMarker` carry the app's origin + * marker and a HealthLog row id of this account (a cycle sample: a day-log of + * this account on its day), `byExternalId` carry such a row id without the + * marker (samples from before the marker existed), `matchedManual` match a + * manual entry (same type and value, within 2 s). A marked sample this + * account does not hold (a move to a new instance without a backup) is + * imported like any other. */ writtenByHealthLog: { byMarker: number; @@ -186,8 +189,11 @@ interface PreparedMeasurement { * at the close tag, once its `` children have been read. */ interface PendingRecord { - /** Queue the record; `healthLogId` is the `HKExternalUUID` it carried. */ - commit: (healthLogId: string | null) => void; + /** + * Queue the record; `healthLogId` is the `HKExternalUUID` it carried and + * `marked` whether it (or its correlation) carried the origin marker. + */ + commit: (healthLogId: string | null, marked: boolean) => void; ownOrigin: boolean; healthLogId: string | null; /** The record's sample identity, computed only when it is left out. */ @@ -369,7 +375,11 @@ export interface StreamParseInput { /** Prisma client to flush rows through. */ prisma: Pick< PrismaClient, - "measurement" | "workout" | "$transaction" | "$queryRawUnsafe" + | "measurement" + | "workout" + | "cycleDayLog" + | "$transaction" + | "$queryRawUnsafe" >; /** * Live progress hook. Called every `PROGRESS_TICK_RECORDS` records @@ -449,6 +459,15 @@ export async function streamParseExportXml( sampleKey: string; ownOrigin: boolean; } | null = null; + // Marked cycle samples, held to the end of the parse: one counts as written + // by HealthLog only when this account holds a day-log on its day. + const markedCycle: Array<{ + hkType: string; + dayKey: string; + rawValue: string | undefined; + protectionUsed?: boolean; + sampleKey: string; + }> = []; // Cumulative fold: type -> local day -> hashed source identity -> subtotal. const cumulativeBucket = new Map< @@ -471,13 +490,25 @@ export async function streamParseExportXml( countedOwn.add(sampleKey); writtenByHealthLog[reason] += 1; }; - // The `HKExternalUUID` a queued spot row carried, checked at flush. + // The `HKExternalUUID` a queued spot row carried, checked at flush, and the + // queued rows that carried the origin marker. const healthLogIdOf = new Map(); + const markedRows = new Set(); + // Marked cumulative samples that named a HealthLog row id, held to the end + // of the parse and added to their day only when the id is not this + // account's. Small: one per mirrored total. + const markedCumulative: Array<{ + type: MeasurementType; + healthLogId: string; + sampleKey: string; + add: () => void; + }> = []; // A `` (a blood pressure reading) wraps its records, and the // marker may sit on the correlation itself, before or after them. Its records // are held until it closes, then all kept or all left out together. let currentCorrelation: { ownOrigin: boolean; + healthLogId: string | null; held: PendingRecord[]; } | null = null; // Spot-row batch awaiting flush. @@ -536,12 +567,29 @@ export async function streamParseExportXml( }; // ── Flush helpers ────────────────────────────────────────── - // A sample the app wrote before it stamped its origin marker still carries - // the HealthLog row id as `HKExternalUUID`. A hit on one of this account's - // rows (deleted ones included: the sample is ours either way) of the same - // kind is an exact match. Blood pressure is one row per half, while the app - // stamps both samples of a reading with one id (the systolic row's, the - // diastolic row's in older builds), so either half matches either id. + // A sample the app wrote carries the HealthLog row id as `HKExternalUUID`, + // with the origin marker or (before the marker existed) without it. A hit on + // one of this account's rows (deleted ones included: the sample is ours + // either way) of the same kind is an exact match. The marker alone proves + // nothing about THIS account: after a move to a new instance without a + // backup, re-importing the export must bring the mirrored values back. + // Blood pressure is one row per half, while the app stamps both samples of a + // reading with one id (the systolic row's, the diastolic row's in older + // builds), so either half matches either id. + const ownRowKinds = async ( + ids: ReadonlySet, + ): Promise> => { + const out = new Map(); + const list = [...ids]; + for (let i = 0; i < list.length; i += 1000) { + const known = await prisma.measurement.findMany({ + where: { userId, id: { in: list.slice(i, i + 1000) } }, + select: { id: true, type: true }, + }); + for (const row of known) out.set(row.id, kindOf(row.type)); + } + return out; + }; const withoutHealthLogIds = async ( rows: PreparedMeasurement[], ): Promise => { @@ -551,17 +599,16 @@ export async function streamParseExportXml( if (id) ids.add(id); } if (ids.size === 0) return rows; - const known = await prisma.measurement.findMany({ - where: { userId, id: { in: [...ids] } }, - select: { id: true, type: true }, - }); - if (known.length === 0) return rows; - const typeOf = new Map(known.map((row) => [row.id, row.type])); + const kindOfId = await ownRowKinds(ids); + if (kindOfId.size === 0) return rows; return rows.filter((row) => { const id = healthLogIdOf.get(row); - const rowType = id ? typeOf.get(id) : undefined; - if (!rowType || kindOf(rowType) !== kindOf(row.type)) return true; - countOwn("byExternalId", row.externalId); + const rowKind = id ? kindOfId.get(id) : undefined; + if (!rowKind || rowKind !== kindOf(row.type)) return true; + countOwn( + markedRows.has(row) ? "byMarker" : "byExternalId", + row.externalId, + ); return false; }); }; @@ -663,7 +710,10 @@ export async function streamParseExportXml( const chunk = await withoutManualMirrors( await withoutHealthLogIds(incoming), ); - for (const row of incoming) healthLogIdOf.delete(row); + for (const row of incoming) { + healthLogIdOf.delete(row); + markedRows.delete(row); + } for (const row of chunk) widenSpan(row.measuredAt); const insertedArrivals: Array<{ id: string; @@ -993,6 +1043,61 @@ export async function streamParseExportXml( } }; + const consumeCycle = (rec: { + hkType: string; + dayKey: string; + rawValue: string | undefined; + protectionUsed?: boolean; + }): void => { + const consumed = cycleAccumulator.consume( + rec.hkType, + rec.dayKey, + rec.rawValue, + rec.protectionUsed, + ); + if (!consumed) { + // Recognised identifier but unrecognised value — count it under + // `unknown` with the reason tag so operators can spot it. + unknown[`${rec.hkType}::cycle_unmapped`] = + (unknown[`${rec.hkType}::cycle_unmapped`] ?? 0) + 1; + } + }; + + // Marked cumulative samples: left out when their id is a row of this + // account of the same kind, added to their day otherwise. + const settleMarkedCumulative = async (): Promise => { + if (markedCumulative.length === 0) return; + const kindOfId = await ownRowKinds( + new Set(markedCumulative.map((m) => m.healthLogId)), + ); + for (const m of markedCumulative.splice(0, markedCumulative.length)) { + if (kindOfId.get(m.healthLogId) === kindOf(m.type)) { + countOwn("byMarker", m.sampleKey); + } else { + m.add(); + } + } + }; + + // Marked cycle samples: left out when this account holds a day-log on + // their day (deleted ones included), folded in otherwise. + const settleMarkedCycle = async (): Promise => { + if (markedCycle.length === 0) return; + const days = [...new Set(markedCycle.map((rec) => rec.dayKey))]; + const held = new Set(); + for (let i = 0; i < days.length; i += 1000) { + const logs = await prisma.cycleDayLog.findMany({ + where: { userId, date: { in: days.slice(i, i + 1000) } }, + select: { date: true }, + }); + for (const log of logs) held.add(log.date); + } + for (const rec of markedCycle.splice(0, markedCycle.length)) { + if (held.has(rec.dayKey)) countOwn("byMarker", rec.sampleKey); + else consumeCycle(rec); + } + }; + // ── SAX parser configuration ──────────────────────────────── const parser = sax.parser(true, { trim: true }); let pendingError: Error | null = null; @@ -1121,27 +1226,44 @@ export async function streamParseExportXml( attrs.startDate, attrs.endDate, ), - commit: (healthLogId) => { + commit: (healthLogId, marked) => { if (CUMULATIVE_HK_TYPES.has(mapped.type)) { const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone); const sourceHash = hashCumulativeSourceIdentity( attrs.sourceName, attrs.device, ); - let byDay = cumulativeBucket.get(mapped.type); - if (!byDay) { - byDay = new Map(); - cumulativeBucket.set(mapped.type, byDay); + const add = () => { + let byDay = cumulativeBucket.get(mapped.type); + if (!byDay) { + byDay = new Map(); + cumulativeBucket.set(mapped.type, byDay); + } + let bySource = byDay.get(dayKey); + if (!bySource) { + bySource = new Map(); + byDay.set(dayKey, bySource); + } + bySource.set( + sourceHash, + (bySource.get(sourceHash) ?? 0) + mapped.value, + ); + }; + if (marked && healthLogId) { + markedCumulative.push({ + type: mapped.type, + healthLogId, + sampleKey: hashSampleKey( + hkType, + attrs.value ?? "", + attrs.startDate, + attrs.endDate, + ), + add, + }); + } else { + add(); } - let bySource = byDay.get(dayKey); - if (!bySource) { - bySource = new Map(); - byDay.set(dayKey, bySource); - } - bySource.set( - sourceHash, - (bySource.get(sourceHash) ?? 0) + mapped.value, - ); } else { // Spot row: derive a stable externalId, queue for flush. const externalId = hashSampleKey( @@ -1163,6 +1285,7 @@ export async function streamParseExportXml( }; spotBatch.push(row); if (healthLogId) healthLogIdOf.set(row, healthLogId); + if (marked) markedRows.add(row); } }, }; @@ -1244,7 +1367,7 @@ export async function streamParseExportXml( } if (name === "Correlation") { - currentCorrelation = { ownOrigin: false, held: [] }; + currentCorrelation = { ownOrigin: false, healthLogId: null, held: [] }; return; } @@ -1255,8 +1378,12 @@ export async function streamParseExportXml( else if (currentCycleRecord) currentCycleRecord.ownOrigin = true; else if (currentCorrelation) currentCorrelation.ownOrigin = true; } - if (pendingRecord && attrs.key === HK_EXTERNAL_UUID_METADATA_KEY) { - pendingRecord.healthLogId = attrs.value || null; + if (attrs.key === HK_EXTERNAL_UUID_METADATA_KEY) { + // On a record, or on the correlation that wraps it. + if (pendingRecord) pendingRecord.healthLogId = attrs.value || null; + else if (currentCorrelation && !currentCycleRecord) { + currentCorrelation.healthLogId = attrs.value || null; + } } // Attach the SexualActivity protection flag to the open cycle record. // Apple writes `HKMetadataKeySexualActivityProtectionUsed` with a @@ -1306,18 +1433,16 @@ export async function streamParseExportXml( const rec = pendingRecord; pendingRecord = null; if (currentCorrelation) currentCorrelation.held.push(rec); - else if (rec.ownOrigin) countOwn("byMarker", rec.sampleKey()); - else rec.commit(rec.healthLogId); + else rec.commit(rec.healthLogId, rec.ownOrigin); } if (tagName === "Correlation" && currentCorrelation) { const correlation = currentCorrelation; currentCorrelation = null; for (const rec of correlation.held) { - if (correlation.ownOrigin || rec.ownOrigin) { - countOwn("byMarker", rec.sampleKey()); - } else { - rec.commit(rec.healthLogId); - } + rec.commit( + rec.healthLogId ?? correlation.healthLogId, + correlation.ownOrigin || rec.ownOrigin, + ); } } if (tagName === "Workout" && currentWorkout) { @@ -1328,22 +1453,12 @@ export async function streamParseExportXml( const rec = currentCycleRecord; currentCycleRecord = null; if (rec.ownOrigin) { - // A day HealthLog mirrored into Apple Health from its own day-log. - countOwn("byMarker", rec.sampleKey); + // A day HealthLog mirrored into Apple Health from a day-log; decided + // at the end of the parse against this account's day-logs. + markedCycle.push(rec); return; } - const consumed = cycleAccumulator.consume( - rec.hkType, - rec.dayKey, - rec.rawValue, - rec.protectionUsed, - ); - if (!consumed) { - // Recognised identifier but unrecognised value — count it under - // `unknown` with the reason tag so operators can spot it. - unknown[`${rec.hkType}::cycle_unmapped`] = - (unknown[`${rec.hkType}::cycle_unmapped`] ?? 0) + 1; - } + consumeCycle(rec); } }; @@ -1421,7 +1536,9 @@ export async function streamParseExportXml( // threshold during the parse. await flushSpotBatch(); await flushWorkoutBatch(); + await settleMarkedCumulative(); await flushCumulativeBuckets(); + await settleMarkedCycle(); // v1.15.0 — fold the accumulated reproductive samples into CYCLE // day-logs. Gated on cycle-tracking being enabled for the account so a diff --git a/tests/fixtures/apple-health/export-own-origin.synthetic.xml b/tests/fixtures/apple-health/export-own-origin.synthetic.xml index aff5f87ac..fa9491018 100644 --- a/tests/fixtures/apple-health/export-own-origin.synthetic.xml +++ b/tests/fixtures/apple-health/export-own-origin.synthetic.xml @@ -9,9 +9,10 @@ verbatim; that has to be checked against a real export (see #1109). Records, in order: - 1. weight 80.1 carries the HealthLog origin marker -> left out + 1. weight 80.1 origin marker and a HealthLog row id -> left out + when the importing account holds that row 2. weight 80.2 user-entered, no marker -> imported - 3. weight 80.3 marker among other metadata entries -> left out + 3. weight 80.3 marker among other metadata entries, no id -> imported 4. weight 80.4 no metadata at all -> imported 5. weight 80.5 origin key with another value -> imported --> diff --git a/tests/integration/apple-health-import-own-samples.test.ts b/tests/integration/apple-health-import-own-samples.test.ts index a24b663de..e252a6fea 100644 --- a/tests/integration/apple-health-import-own-samples.test.ts +++ b/tests/integration/apple-health-import-own-samples.test.ts @@ -9,7 +9,12 @@ * twice and the rollups count it twice. * * Asserted against a real Postgres: - * - a record carrying the marker is left out, and counted; + * - a record carrying the marker and a row id of this account is left out, + * and counted; the marker alone proves nothing about this account (a move + * to a new instance without a backup), so a marked record whose id is + * missing, unknown or another account's is imported, a deleted row of this + * account still counts, and a marked cycle day counts only when the + * account holds a day-log on it; * - a record without it lands as before; * - the marker is read from a child `` that follows the * record's open tag, so the record is committed at its close tag; @@ -56,13 +61,46 @@ ${records} const weight = (value: number, end: string, children = "") => ` ${children}`; -const OWN = ``; +const MARKER = ``; +const ext = (id: string) => + ``; +/** The row the app mirrored: a weight of this account, its id on the sample. */ +const OWN_WEIGHT_ID = "own-weight-row"; +const OWN_SYS_ID = "own-sys-row"; +const OWN = `${MARKER}${ext(OWN_WEIGHT_ID)}`; +const OWN_BP = `${MARKER}${ext(OWN_SYS_ID)}`; const OTHER = ``; +/** + * A row of this account the samples point at. Withings and ten minutes off the + * samples, so neither the series under test nor the manual-value match sees it. + */ +async function seedOwnRow( + userId: string, + id: string, + type: "WEIGHT" | "BLOOD_PRESSURE_SYS" | "ACTIVITY_STEPS", + deletedAt: Date | null = null, +) { + await prisma.measurement.create({ + data: { + id, + userId, + type, + value: type === "ACTIVITY_STEPS" ? 1000 : 70, + unit: + type === "WEIGHT" ? "kg" : type === "ACTIVITY_STEPS" ? "steps" : "mmHg", + source: "WITHINGS", + measuredAt: new Date("2026-05-14T06:10:00.000Z"), + deletedAt, + }, + }); +} + async function run(records: string) { const user = await prisma.user.create({ data: { username: "own-samples", email: "own@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_WEIGHT_ID, "WEIGHT"); const result = await streamParseExportXml({ xmlPath: writeXml(records), userId: user.id, @@ -139,6 +177,7 @@ describe("blood pressure correlations and self-closing records", () => { const user = await prisma.user.create({ data: { username: "own-bp", email: "bp@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_SYS_ID, "BLOOD_PRESSURE_SYS"); const result = await streamParseExportXml({ xmlPath: writeXml(records), userId: user.id, @@ -148,6 +187,7 @@ describe("blood pressure correlations and self-closing records", () => { const rows = await prisma.measurement.findMany({ where: { userId: user.id, + source: "APPLE_HEALTH", type: { in: ["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] }, }, orderBy: [{ type: "asc" }], @@ -173,7 +213,7 @@ describe("blood pressure correlations and self-closing records", () => { [bpRecord("Systolic", 121, AT), bpRecord("Diastolic", 79, AT)].join( "\n", ), - `\n ${OWN}`, + `\n ${OWN_BP}`, ), ); expect(values).toEqual([]); @@ -186,7 +226,7 @@ describe("blood pressure correlations and self-closing records", () => { bpRecord("Diastolic", 79, AT), ].join("\n"); const { values } = await runBp( - ` \n${records}\n ${OWN}\n `, + ` \n${records}\n ${OWN_BP}\n `, ); expect(values).toEqual([]); }); @@ -195,7 +235,7 @@ describe("blood pressure correlations and self-closing records", () => { const { values } = await runBp( correlation( [ - bpRecord("Systolic", 121, AT, `\n ${OWN}\n `), + bpRecord("Systolic", 121, AT, `\n ${OWN_BP}\n `), bpRecord("Diastolic", 79, AT), ].join("\n"), ), @@ -210,7 +250,7 @@ describe("blood pressure correlations and self-closing records", () => { [bpRecord("Systolic", 121, AT), bpRecord("Diastolic", 79, AT)].join( "\n", ), - `\n ${OWN}`, + `\n ${OWN_BP}`, ), bpRecord("Systolic", 130, "2026-05-14 09:00:00 +0200"), ].join("\n"), @@ -230,10 +270,11 @@ describe("the synthetic export fixture", () => { // `export-own-origin.synthetic.xml` is invented, in the shape Apple's DTD // documents. It shows the parser reads a file laid out like an export; it // cannot show that Apple writes a custom metadata key verbatim. - it("leaves out the two marked records and imports the other three", async () => { + it("leaves out the marked record this account holds and imports the other four", async () => { const user = await prisma.user.create({ data: { username: "own-fixture", email: "fx@example.test", role: "USER" }, }); + await seedOwnRow(user.id, "00000000-0000-0000-0000-000000000001", "WEIGHT"); const result = await streamParseExportXml({ xmlPath: join( process.cwd(), @@ -247,9 +288,9 @@ describe("the synthetic export fixture", () => { where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, orderBy: { measuredAt: "asc" }, }); - expect(rows.map((r) => r.value)).toEqual([80.2, 80.4, 80.5]); + expect(rows.map((r) => r.value)).toEqual([80.2, 80.3, 80.4, 80.5]); expect(result.writtenByHealthLog).toEqual({ - byMarker: 2, + byMarker: 1, byExternalId: 0, matchedManual: 0, }); @@ -421,8 +462,6 @@ describe("the manual-mirror lookup stays off the hot path", () => { describe("a sample whose HKExternalUUID is a HealthLog row id", () => { const AT = "2026-05-14 08:00:00 +0200"; - const ext = (id: string) => - ``; async function runWith( seed: (userId: string) => Promise>, @@ -524,11 +563,12 @@ describe("a sample whose HKExternalUUID is a HealthLog row id", () => { describe("each sample left out is counted once", () => { it("does not count a marked record twice when it appears top level and inside its correlation", async () => { const AT = "2026-05-14 08:00:00 +0200"; - const sys = bpRecord("Systolic", 121, AT, `\n ${OWN}\n `); - const dia = bpRecord("Diastolic", 79, AT, `\n ${OWN}\n `); + const sys = bpRecord("Systolic", 121, AT, `\n ${OWN_BP}\n `); + const dia = bpRecord("Diastolic", 79, AT, `\n ${OWN_BP}\n `); const user = await prisma.user.create({ data: { username: "own-once", email: "once@example.test", role: "USER" }, }); + await seedOwnRow(user.id, OWN_SYS_ID, "BLOOD_PRESSURE_SYS"); const result = await streamParseExportXml({ xmlPath: writeXml( [sys, dia, correlation([sys, dia].join("\n"))].join("\n"), @@ -542,7 +582,7 @@ describe("each sample left out is counted once", () => { }); describe("a cycle sample HealthLog wrote itself", () => { - it("is left out of the day-log import", async () => { + it("is left out on a day this account holds, imported on a day it does not", async () => { const user = await prisma.user.create({ data: { username: "own-cycle", @@ -552,11 +592,20 @@ describe("a cycle sample HealthLog wrote itself", () => { timezone: "Europe/Berlin", }, }); + // The day the app mirrored from: a day-log of this account. + await prisma.cycleDayLog.create({ + data: { userId: user.id, date: "2026-03-02", flow: "MEDIUM" }, + }); const flow = (day: string, children = "") => ` ${children}`; const result = await streamParseExportXml({ xmlPath: writeXml( - [flow("2026-03-02", OWN), flow("2026-03-03")].join("\n"), + [ + flow("2026-03-02", MARKER), + flow("2026-03-03"), + // Marked, but no day-log here: a move to a new instance. + flow("2026-03-04", MARKER), + ].join("\n"), ), userId: user.id, userTimezone: "Europe/Berlin", @@ -566,7 +615,119 @@ describe("a cycle sample HealthLog wrote itself", () => { where: { userId: user.id }, orderBy: { date: "asc" }, }); - expect(days.map((d) => d.date)).toEqual(["2026-03-03"]); + expect(days.map((d) => d.date)).toEqual([ + "2026-03-02", + "2026-03-03", + "2026-03-04", + ]); + // The held day stays the account's own: the marked sample did not + // re-import it as an Apple Health day. + expect(days[0].source).toBe("MANUAL"); + expect(result.writtenByHealthLog.byMarker).toBe(1); + }); +}); + +describe("a marked record this account does not hold", () => { + // A move to a new instance without a backup: the export still carries the + // marker on every value the app mirrored, but none of the rows exist here. + it("is imported when the id is missing, unknown or another account's", async () => { + const other = await prisma.user.create({ + data: { username: "own-move-2", email: "mv2@example.test", role: "USER" }, + }); + await seedOwnRow(other.id, "foreign-weight-row", "WEIGHT"); + const user = await prisma.user.create({ + data: { username: "own-move", email: "mv@example.test", role: "USER" }, + }); + const result = await streamParseExportXml({ + xmlPath: writeXml( + [ + weight(80.1, "2026-05-14 08:00:00 +0200", MARKER), + weight( + 80.2, + "2026-05-14 09:00:00 +0200", + `${MARKER}${ext("00000000-0000-4000-8000-000000000009")}`, + ), + weight( + 80.3, + "2026-05-14 10:00:00 +0200", + `${MARKER}${ext("foreign-weight-row")}`, + ), + ].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, + orderBy: { measuredAt: "asc" }, + }); + expect(rows.map((r) => r.value)).toEqual([80.1, 80.2, 80.3]); + expect(result.writtenByHealthLog).toEqual({ + byMarker: 0, + byExternalId: 0, + matchedManual: 0, + }); + }); + + it("is still left out when its id names a deleted row of this account", async () => { + const user = await prisma.user.create({ + data: { + username: "own-deleted", + email: "del@example.test", + role: "USER", + }, + }); + await seedOwnRow(user.id, OWN_WEIGHT_ID, "WEIGHT", new Date()); + const result = await streamParseExportXml({ + xmlPath: writeXml(weight(80.1, "2026-05-14 08:00:00 +0200", OWN)), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { userId: user.id, type: "WEIGHT", source: "APPLE_HEALTH" }, + }); + expect(rows).toHaveLength(0); + expect(result.writtenByHealthLog.byMarker).toBe(1); + }); + + it("adds a marked cumulative sample to its day unless its id is this account's", async () => { + const user = await prisma.user.create({ + data: { username: "own-steps", email: "st@example.test", role: "USER" }, + }); + await seedOwnRow(user.id, "own-steps-row", "ACTIVITY_STEPS"); + const steps = (value: number, at: string, children: string) => + ` ${children}`; + const result = await streamParseExportXml({ + xmlPath: writeXml( + [ + steps( + 700, + "2026-05-14 08:00:00 +0200", + `${MARKER}${ext("own-steps-row")}`, + ), + steps( + 300, + "2026-05-14 09:00:00 +0200", + `${MARKER}${ext("not-a-row")}`, + ), + steps(200, "2026-05-14 10:00:00 +0200", ""), + ].join("\n"), + ), + userId: user.id, + userTimezone: "Europe/Berlin", + prisma, + }); + const rows = await prisma.measurement.findMany({ + where: { + userId: user.id, + type: "ACTIVITY_STEPS", + source: "APPLE_HEALTH", + }, + }); + // 300 + 200: the 700 the app mirrored from this account's row stays out. + expect(rows.map((r) => r.value)).toEqual([500]); expect(result.writtenByHealthLog.byMarker).toBe(1); }); }); From 4769502a4b70b19dffd9fbf8fdbab07ffcd421d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 22:17:27 +0200 Subject: [PATCH 9/9] fix(backup): check the weekly copy's preview against the schema section by section Taking the weekly preview from the writer dropped the schema check the old scanner ran at the end, so a copy that violates the schema was described as restorable. The collector now checks each section in its JSON form as the writer shows it, an array section and a bulk table one element at a time, plus the schema's one whole-document rule (a canonical copy names every measurement by id). A copy that fails gets no stored preview, and the preview route reads the copy itself, as it did before. Nothing is kept beyond one element. Measured on the preview memory test under --max-old-space-size=524: held after the producer 3 MB with and without the check, raw peak heapUsed 293 MB with it (335 MB without). The test now loads the backup schema before its baseline: a module loaded once per process, not memory that grows with the record. --- .../export/__tests__/backup-preview.test.ts | 66 ++++++++++++++++-- src/lib/export/backup-preview.ts | 68 +++++++++++++++++-- src/lib/jobs/reminder/backup-handlers.ts | 2 +- .../integration/backup-preview-memory.test.ts | 7 +- tests/integration/backup-round-trip.test.ts | 4 +- 5 files changed, 133 insertions(+), 14 deletions(-) diff --git a/src/lib/export/__tests__/backup-preview.test.ts b/src/lib/export/__tests__/backup-preview.test.ts index 8016c6c4c..b82ac0b6f 100644 --- a/src/lib/export/__tests__/backup-preview.test.ts +++ b/src/lib/export/__tests__/backup-preview.test.ts @@ -101,8 +101,8 @@ describe("createBackupPreviewCollector", () => { const expectedKeys = new BackupKeyIdCollector(); expectedKeys.visit(FILE); - it("says what a full read of the copy says", () => { - const collector = createBackupPreviewCollector(); + it("says what a full read of the copy says", async () => { + const collector = await createBackupPreviewCollector(); observeAsTheWriterDoes(FILE, collector.observe); const result = collector.finish(countsOf(FILE)); expect(result).not.toBeNull(); @@ -112,8 +112,8 @@ describe("createBackupPreviewCollector", () => { expect(result!.keys.keyIds()).toEqual(["v3"]); }); - it("answers null for a copy that does not name its version, owner and date", () => { - const collector = createBackupPreviewCollector(); + it("answers null for a copy that does not name its version, owner and date", async () => { + const collector = await createBackupPreviewCollector(); const { userId, ...rest } = FILE; void userId; observeAsTheWriterDoes(rest, collector.observe); @@ -121,6 +121,64 @@ describe("createBackupPreviewCollector", () => { }); }); +describe("createBackupPreviewCollector against the schema", () => { + // A copy the schema refuses must not be described as restorable: a null + // preview sends the preview route to read the copy, which says why not. + async function previewOf(file: Record) { + const collector = await createBackupPreviewCollector(); + observeAsTheWriterDoes(file, collector.observe); + return collector.finish(countsOf(FILE)); + } + + it("answers null when a bulk row violates the schema", async () => { + const measurements = FILE.measurements.map((m, i) => + i === 1 ? { ...m, measuredAt: "not a date" } : m, + ); + expect(await previewOf({ ...FILE, measurements })).toBeNull(); + }); + + it("answers null when a whole section violates the schema", async () => { + const labResults = [{ ...FILE.labResults[0], takenAt: 42 }]; + expect(await previewOf({ ...FILE, labResults })).toBeNull(); + }); + + it("answers null when the header does not parse", async () => { + expect( + await previewOf({ ...FILE, exportedAt: "yesterday-ish" }), + ).toBeNull(); + }); + + it("answers null for a canonical copy with a measurement without an id", async () => { + const measurements = FILE.measurements.map((m, i) => + i === 0 ? { ...m, id: undefined } : m, + ); + expect(await previewOf({ ...FILE, measurements })).toBeNull(); + }); + + it("agrees with the whole-document parse on every case above", () => { + // The oracle: each refused file is refused by the full schema too. + const bad = [ + { + ...FILE, + measurements: FILE.measurements.map((m, i) => + i === 1 ? { ...m, measuredAt: "not a date" } : m, + ), + }, + { ...FILE, labResults: [{ ...FILE.labResults[0], takenAt: 42 }] }, + { ...FILE, exportedAt: "yesterday-ish" }, + { + ...FILE, + measurements: FILE.measurements.map((m, i) => + i === 0 ? { ...m, id: undefined } : m, + ), + }, + ]; + for (const file of bad) { + expect(backupPayloadSchema.safeParse(file).success).toBe(false); + } + }); +}); + describe("storedPreviewFor", () => { const chunked = { data: null, chunkCount: 4, chunkStreamId: "s1" }; const preview = () => diff --git a/src/lib/export/backup-preview.ts b/src/lib/export/backup-preview.ts index 91eebab21..446998d36 100644 --- a/src/lib/export/backup-preview.ts +++ b/src/lib/export/backup-preview.ts @@ -35,7 +35,10 @@ import { } from "@/lib/export/backup-key-ids"; import type { FullBackupCounts } from "@/lib/export/full-backup-payload"; import type { StoredBackupRef } from "@/lib/export/stored-backup"; -import type { BackupSummary } from "@/lib/validations/backup-summary"; +import { + BACKUP_SCHEMA_VERSION, + type BackupSummary, +} from "@/lib/validations/backup-summary"; export interface BackupPreview { version: 1; @@ -207,7 +210,9 @@ export interface BackupPreviewCollector { observe(member: string, value: unknown): void; /** * The counts and key uses, once the writer has finished; null when the - * copy did not name its schema version, owner and date. + * copy did not name its schema version, owner and date, or when a section + * did not pass the backup schema. A null preview sends the preview route + * to read the copy itself, which then says why it cannot be restored. */ finish(counts: FullBackupCounts): { summary: BackupSummary; @@ -215,19 +220,64 @@ export interface BackupPreviewCollector { } | null; } +type SchemaModule = typeof import("@/lib/validations/backup"); + +/** + * A check of one member of the payload against `backupPayloadSchema`, as the + * writer shows it: a whole section, or one row of a bulk table. An array + * section is checked an element at a time, so the check holds one element's + * copy at most, never a section's. Each value is checked in its JSON form, + * exactly as it lands in the copy. A member the schema does not name passes, + * as it does under the schema's passthrough. + */ +function sectionChecker( + mod: SchemaModule, +): (member: string, value: unknown) => boolean { + const shape = mod.backupPayloadSchema.shape as Record; + const asWritten = (value: unknown): unknown => + value === undefined ? undefined : JSON.parse(JSON.stringify(value)); + const unwrap = (schema: z.ZodType): z.ZodType => + schema instanceof z.ZodDefault + ? unwrap(schema.unwrap() as z.ZodType) + : schema; + return (member, value) => { + const declared = shape[member]; + if (!declared) return true; + const section = unwrap(declared); + if (section instanceof z.ZodArray) { + const element = section.element as z.ZodType; + const rows = Array.isArray(value) ? value : [value]; + return rows.every((row) => element.safeParse(asWritten(row)).success); + } + return declared.safeParse(asWritten(value)).success; + }; +} + /** * Work out a weekly copy's preview from what its writer already has in hand: - * each section as it is written, for the key uses, and the writer's own - * counts. Nothing is parsed and nothing is kept but the key samples, so the - * preview costs the pass no memory that grows with the record. + * each section as it is written, for the key uses and the schema check, and + * the writer's own counts. Nothing is parsed back and nothing is kept but the + * key samples, so the preview costs the pass no memory that grows with the + * record. The check loads the backup schema (`backup-summary.ts` says why it + * is loaded on first use), a fixed cost of the module, not of the record. */ -export function createBackupPreviewCollector(): BackupPreviewCollector { +export async function createBackupPreviewCollector(): Promise { + const check = sectionChecker(await import("@/lib/validations/backup")); const keys = new BackupKeyIdCollector(); const header: Partial< Record<"schemaVersion" | "userId" | "exportedAt", string> > = {}; + let valid = true; + let measurementWithoutId = false; return { observe(member, value) { + if (valid && !check(member, value)) valid = false; + if (member === "measurements") { + const rows: unknown[] = Array.isArray(value) ? value : [value]; + if (rows.some((row) => !(row as { id?: unknown } | null)?.id)) { + measurementWithoutId = true; + } + } if ( (member === "schemaVersion" || member === "userId" || @@ -242,6 +292,12 @@ export function createBackupPreviewCollector(): BackupPreviewCollector { finish(counts) { const { schemaVersion, userId, exportedAt } = header; if (!schemaVersion || !userId || !exportedAt) return null; + if (!valid) return null; + // The schema's one whole-document rule: a canonical copy names every + // measurement by a stable id. + if (schemaVersion === BACKUP_SCHEMA_VERSION && measurementWithoutId) { + return null; + } const summary = { schemaVersion, userId, exportedAt } as BackupSummary; for (const field of SUMMARY_COUNTS) summary[field] = counts[field]; return { summary, keys }; diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 62e1adeda..c8bc8fa83 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -207,7 +207,7 @@ export async function handleDataBackup( // so its size does not depend on this process's memory (#1031). // The restore preview is taken from the writer's own counts and // the sections as they are written, never by parsing the copy. - const preview = createBackupPreviewCollector(); + const preview = await createBackupPreviewCollector(); let counts: FullBackupCounts | undefined; const { bytes: storedBytes } = await storeBackupBlob( prisma, diff --git a/tests/integration/backup-preview-memory.test.ts b/tests/integration/backup-preview-memory.test.ts index fbe8eb6f6..569ec6e7b 100644 --- a/tests/integration/backup-preview-memory.test.ts +++ b/tests/integration/backup-preview-memory.test.ts @@ -118,6 +118,11 @@ describe("weekly copy preview under a memory budget", () => { }); it("holds nothing of the record once the producer is done", async () => { + // The backup schema the preview checks each section against is a module + // loaded once per process (about 12 MB, `backup-summary.ts`), not memory + // that grows with the record: load it before the baseline, as a running + // server has long since done. + await import("@/lib/validations/backup"); const baseline = liveHeapBytes(); let peakHeld = 0; let heldAtEnd = 0; @@ -127,7 +132,7 @@ describe("weekly copy preview under a memory budget", () => { rawPeak = Math.max(rawPeak, process.memoryUsage().heapUsed); }, 5); // Wired exactly as the weekly job wires it (`backup-handlers.ts`). - const preview = createBackupPreviewCollector(); + const preview = await createBackupPreviewCollector(); let counts: FullBackupCounts | undefined; try { await storeBackupBlob( diff --git a/tests/integration/backup-round-trip.test.ts b/tests/integration/backup-round-trip.test.ts index 045447121..28e9b83bc 100644 --- a/tests/integration/backup-round-trip.test.ts +++ b/tests/integration/backup-round-trip.test.ts @@ -1495,7 +1495,7 @@ describe("every model the plan claims two-ended survives a real restore", () => // below then covers BOTH, since the row that gets restored comes from the // streamed bytes. let streamedJson = ""; - const previewCollector = createBackupPreviewCollector(); + const previewCollector = await createBackupPreviewCollector(); const streamedCounts = await streamFullBackupJson( prisma, OWNER_ID, @@ -3779,7 +3779,7 @@ describe("every column of every two-ended model survives a real restore", () => const exportedAt = new Date("2026-08-01T00:00:00.000Z"); let json = ""; - const previewCollector = createBackupPreviewCollector(); + const previewCollector = await createBackupPreviewCollector(); const counts = await streamFullBackupJson( prisma, OWNER_ID,