From 4242b454ce460bb0a8856605b71917a918343553 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 25 Sep 2026 01:54:27 +0200 Subject: [PATCH] fix(import): fold rollups over the span the import wrote After an Apple Health import the worker folded the rollup table from the account's first measurement to its last, whatever the import carried, and so also folded history older than the five-year window that the boot backfill, its discovery and every read path keep to. The import now records the oldest and newest instant it wrote (`measuredSpan` on the job result), and the fold covers that span, clamped to the fold window. On the seeded account of 1.25 million rows the full fold took 12 s, the slowest statement 2.7 s against the 60 s statement limit, so this is about the window rather than time. --- src/lib/jobs/apple-health-import-worker.ts | 35 +++--- .../import-apple-health-export.ts | 22 ++++ .../apple-health-import-rollup-window.test.ts | 116 ++++++++++++++++++ 3 files changed, 159 insertions(+), 14 deletions(-) create mode 100644 tests/integration/apple-health-import-rollup-window.test.ts diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index c610a4825..257649e72 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -41,7 +41,10 @@ import { type ImportJobProgress, type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; -import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; import { withBackgroundEvent } from "@/lib/logging/background"; /** @@ -502,19 +505,23 @@ export async function handleAppleHealthImport( // contradict the row the status poll surfaces to the operator. let rollupFailed = false; try { - const span = await prisma.measurement.aggregate({ - where: { userId }, - _min: { measuredAt: true }, - _max: { measuredAt: true }, - }); - if (span._min.measuredAt && span._max.measuredAt) { - // Add a small tail buffer to `to` so the upper bound is - // exclusive-safe under the rollup aggregator's `< to` filter. - const to = new Date(span._max.measuredAt.getTime() + 1); - await recomputeUserRollups(userId, { - from: span._min.measuredAt, - to, - }); + // Folded over what this import wrote, not over the account's whole + // history, and never further back than the fold window every other + // path keeps to: buckets older than `ROLLUP_FOLD_WINDOW_MS` are by + // design never written, so reads there fall back to live SQL. The old + // span ran from the account's first measurement, whatever the import + // carried, and folded any history older than the window as well. + const span = result.measuredSpan; + if (span) { + const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const spanFrom = new Date(span.from); + const from = spanFrom > windowStart ? spanFrom : windowStart; + // `to` is exclusive in the aggregator, so one millisecond past the + // newest row. + const to = new Date(new Date(span.to).getTime() + 1); + if (to > from) { + await recomputeUserRollups(userId, { from, to }); + } } } catch (rollupErr) { // Rollup failure is non-fatal โ€” the next read falls through to diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index 9bc4fe8cf..d175fc5c0 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -125,6 +125,11 @@ export interface ImportJobResult { /** Estimated `(measurement type, local day)` aggregate rows considered. */ rows: number; }; + /** + * Oldest and newest instant this import wrote a measurement at (ISO), or + * null when it wrote none. What the rollup fold after the import covers. + */ + measuredSpan: { from: string; to: string } | null; /** Bounded auxiliary ECG outcomes. Never contains filenames or waveform data. */ ecg: { discovered: number; @@ -396,6 +401,14 @@ export async function streamParseExportXml( let rowsUpserted = 0; const cumulativeEstimatedDays = new Set(); let cumulativeEstimatedRows = 0; + // The instants this import wrote measurements at, oldest and newest. The + // worker folds the rollup table over exactly this span afterwards. + let spanFrom: Date | null = null; + let spanTo: Date | null = null; + const widenSpan = (at: Date) => { + if (spanFrom === null || at < spanFrom) spanFrom = at; + if (spanTo === null || at > spanTo) spanTo = at; + }; // Per R-1 ยง8 the percent stays best-effort and may remain null // until the parser sees the closing `` tag. We don't // currently mutate this in v1.4.34 โ€” the iOS app keeps polling on @@ -438,6 +451,7 @@ export async function streamParseExportXml( const flushSpotBatch = async (): Promise => { if (spotBatch.length === 0) return; const chunk = spotBatch.splice(0, spotBatch.length); + for (const row of chunk) widenSpan(row.measuredAt); const insertedArrivals: Array<{ id: string; type: MeasurementType; @@ -731,6 +745,7 @@ export async function streamParseExportXml( } const externalId = dailyStatsExternalId(mapping.hkIdentifier, dayKey); const measuredAt = canonicalDailyTimestamp(dayKey, userTimezone); + widenSpan(measuredAt); const rowStart = Date.now(); const verdict = await prisma.$transaction((tx) => reconcileExternalMeasurement( @@ -1202,6 +1217,13 @@ export async function streamParseExportXml( days: cumulativeEstimatedDays.size, rows: cumulativeEstimatedRows, }, + measuredSpan: + spanFrom !== null && spanTo !== null + ? { + from: (spanFrom as Date).toISOString(), + to: (spanTo as Date).toISOString(), + } + : null, ecg: { discovered: 0, imported: 0, diff --git a/tests/integration/apple-health-import-rollup-window.test.ts b/tests/integration/apple-health-import-rollup-window.test.ts new file mode 100644 index 000000000..be558605e --- /dev/null +++ b/tests/integration/apple-health-import-rollup-window.test.ts @@ -0,0 +1,116 @@ +/** + * #1031 โ€” the rollup fold after an import covers what the import wrote. + * + * The worker folded the rollup table from the account's first measurement to + * its last, whatever the import carried, and so also folded history older + * than the five-year window every other fold path keeps to. It now folds the + * span of the rows this import wrote, clamped to that window. + */ +import { createHash } from "node:crypto"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { strToU8, zipSync } from "fflate"; +import type { Job } from "pg-boss"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + handleAppleHealthImport, + _setWorkerPrismaForTests, + type AppleHealthImportPayload, +} from "@/lib/jobs/apple-health-import-worker"; +import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const USER = "user-import-rollup-window"; + +function record(at: Date, value: number): string { + const stamp = at.toISOString().replace("T", " ").slice(0, 19) + " +0000"; + return ``; +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + _setWorkerPrismaForTests(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: USER, + username: USER, + email: `${USER}@example.test`, + timezone: "UTC", + }, + }); +}); + +describe("import rollup fold window", () => { + it("folds the imported span, never past the fold window, and leaves older history alone", async () => { + const prisma = getPrismaClient(); + const now = Date.now(); + const day = 86_400_000; + // Already in the account before the import: a reading eight years old. + await prisma.measurement.create({ + data: { + userId: USER, + type: "WEIGHT", + value: 90, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 8 * 365 * day), + }, + }); + // The import: one reading seven years old, a few recent ones. + const xml = ` + + ${record(new Date(now - 7 * 365 * day), 88)} + ${record(new Date(now - 40 * day), 84)} + ${record(new Date(now - 30 * day), 83)} + ${record(new Date(now - 20 * day), 82)} +`; + const dir = mkdtempSync(join(tmpdir(), "hl-import-window-")); + const path = join(dir, "export.zip"); + const bytes = Buffer.from( + zipSync({ "apple_health_export/export.xml": strToU8(xml) }), + ); + writeFileSync(path, bytes); + await prisma.importJob.create({ + data: { + userId: USER, + pgBossJobId: "window-job", + status: "queued", + uploadBytes: bytes.length, + uploadSha256: createHash("sha256").update(bytes).digest("hex"), + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + }, + }); + + const outcome = await handleAppleHealthImport({ + id: "window-job", + data: { + userId: USER, + uploadPath: path, + uploadBytes: bytes.length, + enqueuedAt: new Date().toISOString(), + }, + } as unknown as Job); + expect(outcome.ok).toBe(true); + + const buckets = await prisma.measurementRollup.findMany({ + where: { userId: USER, type: "WEIGHT", granularity: "DAY" }, + orderBy: { bucketStart: "asc" }, + }); + const windowStart = now - ROLLUP_FOLD_WINDOW_MS; + expect(buckets.length).toBeGreaterThanOrEqual(3); + for (const bucket of buckets) { + expect(bucket.bucketStart.getTime()).toBeGreaterThan(windowStart - day); + } + }); +});