diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index 133c7fdf2..d25ac2487 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -228,7 +228,11 @@ export const softSkipStorage = new AsyncLocalStorage(); * the soft-skip tracker. */ export interface RollupDeferTracker { - keys: Array<{ type: MeasurementType; measuredAt: Date }>; + /** + * One entry per touched `(type, UTC day)`, never one per reading, so the + * ledger stays the size of the history in days rather than in samples. + */ + keys: Map; } export const rollupDeferStorage = new AsyncLocalStorage(); @@ -877,14 +881,13 @@ export async function upsertFitbitMeasurements( // path keeps the inline per-day hook here (small touched set, warm next read). if (opts.deferRollup) { const tracker = rollupDeferStorage.getStore(); - // Plain loop, never `push(...touched)`: `touched` carries ONE entry PER - // READING, and a full-history batch on a sample-dense account (a - // multi-year heart-rate series is six figures of rows) blows the call - // stack when spread into a single call — the engine caps argument counts. - // The RangeError aborted the whole metrics collection on exactly the - // accounts with the most data. + // Collapsed to type-days on the way in: `touched` carries ONE entry PER + // READING and the tracker lives for the whole cycle, so keeping the raw + // entries made the ledger grow with the sample count of the history. if (tracker) { - for (const t of touched) tracker.keys.push(t); + for (const k of collapseToTypeDayKeys(touched)) { + tracker.keys.set(`${k.type}|${k.measuredAt.getTime()}`, k); + } } return { imported, touched, inserted: insertedRows }; } diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index 9195f7bcb..449efbff2 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -5,10 +5,7 @@ import { prisma } from "@/lib/db"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; import { annotate, getEvent } from "@/lib/logging/context"; -import { - collapseToTypeDayKeys, - recomputeUserRollups, -} from "@/lib/rollups/measurement-rollups"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; import { syncUserActivity } from "./sync-activity"; import { hardFailStorage, @@ -90,7 +87,7 @@ export async function syncUserFitbit( ]; const tracker: SoftSkipTracker = { count: 0 }; - const deferTracker: RollupDeferTracker = { keys: [] }; + const deferTracker: RollupDeferTracker = { keys: new Map() }; const hardFailTracker: HardFailTracker = { failures: [] }; let total = 0; let anyFailed = false; @@ -120,9 +117,9 @@ export async function syncUserFitbit( // touched days. One pass replaces the thousands of per-(type,day) round-trips a // deep backfill would otherwise pay (each = an aggregate SELECT + rollup upsert // + 3 queue sends). Best-effort: a populator hiccup never fails the backfill. - if (opts.fullSync && deferTracker.keys.length > 0) { + if (opts.fullSync && deferTracker.keys.size > 0) { try { - const days = collapseToTypeDayKeys(deferTracker.keys); + const days = Array.from(deferTracker.keys.values()); const types = Array.from(new Set(days.map((k) => k.type))); const sorted = days .map((k) => k.measuredAt.getTime()) diff --git a/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts new file mode 100644 index 000000000..cea735f92 --- /dev/null +++ b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts @@ -0,0 +1,97 @@ +/** + * #1023 — the deferred-rollup ledger of a backfill cycle holds type-days, not + * readings. + * + * On a full-history backfill every write defers its rollup fold to one pass at + * the end of the cycle, and the cycle-wide ledger used to keep one entry per + * written reading. A per-minute heart-rate history of two and a half years is + * 1.3 M readings over about 900 days, so the ledger alone grew with the stream + * for the whole walk. It now keeps one entry per touched `(type, UTC day)`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { findManyMock, createManyMock } = vi.hoisted(() => ({ + findManyMock: vi.fn(async () => [] as unknown[]), + createManyMock: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { + findMany: findManyMock, + createManyAndReturn: createManyMock, + update: vi.fn(), + }, + }, +})); +vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: vi.fn(async () => false), + recordSyncFailure: vi.fn(async () => {}), + recordSyncSuccess: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: vi.fn(async () => {}), +})); +vi.mock("../credentials", () => ({ + getUserGoogleHealthCredentials: vi.fn(async () => null), +})); + +import { + runWithGoogleHealthSyncCycle, + upsertGoogleHealthMeasurements, +} from "../sync-core"; + +const DAYS = 3; +const PER_DAY = 1440; +const START = Date.parse("2026-09-01T00:00:00.000Z"); + +beforeEach(() => { + findManyMock.mockReset().mockResolvedValue([]); + createManyMock + .mockReset() + .mockImplementation( + async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d, i) => ({ + id: `row-${i}`, + type: d.type, + measuredAt: d.measuredAt, + })), + ); +}); + +describe("deferred rollup ledger on a dense backfill", () => { + it("keeps one key per touched type-day however many readings land", async () => { + const cycle = await runWithGoogleHealthSyncCycle(async () => { + for (let day = 0; day < DAYS; day++) { + const readings = []; + for (let m = 0; m < PER_DAY; m++) { + const at = new Date(START + day * 86_400_000 + m * 60_000); + readings.push({ + type: "PULSE", + value: 60, + unit: "bpm", + measuredAt: at, + externalId: `${at.toISOString()}:hr`, + }); + } + await upsertGoogleHealthMeasurements("dense-user", readings, { + deferRollup: true, + }); + } + return null; + }); + + expect(cycle.deferredRollupKeys).toHaveLength(DAYS); + expect( + cycle.deferredRollupKeys.map((k) => k.measuredAt.toISOString()).sort(), + ).toEqual([ + "2026-09-01T00:00:00.000Z", + "2026-09-02T00:00:00.000Z", + "2026-09-03T00:00:00.000Z", + ]); + }); +}); diff --git a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts new file mode 100644 index 000000000..22ea1ad96 --- /dev/null +++ b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts @@ -0,0 +1,111 @@ +/** + * #1023 — a dense heart-rate collection is written page by page. + * + * Intraday heart rate is one point a minute. `syncUserMetrics` used to collect + * the whole collection, map all of it, and hand every reading to one upsert, + * so a full-history backfill held a million raw points and a million readings + * at once and ran a 1 GB heap out of memory (the reproduction on 1.3 M + * readings dies within seconds of the walk). The contract pinned here: every + * upsert carries at most one page, and a page is written before the next one + * is requested, so nothing but the current page is ever resident. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { safeFetchMock, upsertMock, events } = vi.hoisted(() => ({ + safeFetchMock: vi.fn(), + upsertMock: vi.fn(), + events: [] as string[], +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, safeFetch: safeFetchMock }; +}); + +vi.mock("../sync-core", () => ({ + getValidToken: vi.fn(async () => ({ accessToken: "token" })), + handleCollectionFetchError: vi.fn(async () => 0), + upsertGoogleHealthMeasurements: upsertMock, +})); + +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: vi.fn(), update: vi.fn() } }, +})); + +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "UTC"), +})); + +import { GOOGLE_HEALTH_PAGE_SIZE } from "../client"; +import { syncUserMetrics } from "../sync-metrics"; + +const PAGES = 3; +const NOW = Date.parse("2026-09-27T17:00:00.000Z"); + +function heartRatePage(page: number): unknown { + const dataPoints = []; + for (let i = 0; i < GOOGLE_HEALTH_PAGE_SIZE; i++) { + const g = page * GOOGLE_HEALTH_PAGE_SIZE + i; + dataPoints.push({ + heartRate: { + beatsPerMinute: String(55 + (g % 50)), + sampleTime: { physicalTime: new Date(NOW - g * 60_000).toISOString() }, + }, + }); + } + return { + dataPoints, + ...(page + 1 < PAGES ? { nextPageToken: String(page + 1) } : {}), + }; +} + +beforeEach(() => { + events.length = 0; + upsertMock.mockReset().mockImplementation(async (_u, readings: unknown[]) => { + events.push(`upsert:${readings.length}`); + return { imported: readings.length, touched: [], inserted: [] }; + }); + safeFetchMock.mockReset().mockImplementation(async (url: string) => { + const u = new URL(url); + let body: unknown = {}; + if (u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) { + const page = Number(u.searchParams.get("pageToken") ?? 0); + events.push(`fetch:${page}`); + body = heartRatePage(page); + } + return { + status: 200, + ok: true, + json: async () => body, + } as unknown as Response; + }); +}); + +describe("syncUserMetrics over a dense heart-rate collection", () => { + it("writes each page before the next one is requested", async () => { + const imported = await syncUserMetrics("dense-user", { + deferRollup: true, + }); + + expect(imported).toBe(PAGES * GOOGLE_HEALTH_PAGE_SIZE); + const pulseWrites = upsertMock.mock.calls + .map((c) => c[1] as Array<{ type: string }>) + .filter((readings) => readings.some((r) => r.type === "PULSE")); + expect(pulseWrites).toHaveLength(PAGES); + for (const readings of pulseWrites) { + expect(readings.length).toBeLessThanOrEqual(GOOGLE_HEALTH_PAGE_SIZE); + } + const hr = events.filter( + (e) => + e.startsWith("fetch:") || e === `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + ); + expect(hr).toEqual([ + "fetch:0", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + "fetch:1", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + "fetch:2", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + ]); + }); +}); diff --git a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts index c9cfe6a94..8d48929df 100644 --- a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts @@ -12,6 +12,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const { fetchDataPointsMock, handleErrorMock, mapWeightMock } = vi.hoisted( () => ({ + // Stands in for the page walk: every metric type is streamed through + // `forEachDataPointPage`, which hands its pages to the callback. fetchDataPointsMock: vi.fn(), handleErrorMock: vi.fn(async () => 0), mapWeightMock: vi.fn(() => { @@ -25,6 +27,23 @@ vi.mock("../client", async (importOriginal) => { return { ...actual, fetchDataPoints: fetchDataPointsMock, + forEachDataPointPage: vi.fn( + async ( + dataType: unknown, + token: string, + verb: string, + query: unknown, + onPage: (points: unknown[]) => Promise, + ) => { + const points = (await fetchDataPointsMock( + dataType, + token, + verb, + query, + )) as unknown[]; + if (points.length > 0) await onPage(points); + }, + ), mapWeight: mapWeightMock, }; }); diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index 0a21e75fc..b38e95a1b 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -512,6 +512,20 @@ export const GOOGLE_HEALTH_PAGE_SIZE = 1000; /** Sleep/exercise read cap — matches the Google Health 25-cap for those types. */ export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25; +/** + * Page ceiling for the intraday heart-rate walk. + * + * The default of 1000 pages is a million points, which a watch that records + * a reading a minute passes in under two years. A walk that hits the ceiling + * reports `truncated`, the backfill treats that as incomplete and throws, and + * pg-boss retries it — so an account with a longer history re-ran the whole + * walk on every retry and every boot and never completed. Ten thousand pages + * is nineteen years at that rate: still a bound on a runaway cursor, no longer + * a bound a real history reaches. The walk is paged (`forEachDataPointPage`), + * so the ceiling costs time, not memory. + */ +export const GOOGLE_HEALTH_DENSE_MAX_PAGES = 10_000; + /** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */ interface GoogleHealthDataPointPage { dataPoints?: GoogleHealthDataPoint[]; @@ -715,6 +729,35 @@ export async function fetchDataPoints( verb: string, query: DataPointQuery = {}, ): Promise { + const points: GoogleHealthDataPoint[] = []; + await forEachDataPointPage(dataType, accessToken, verb, query, (page) => { + for (const p of page) points.push(p); + }); + return points; +} + +/** + * The same walk as `fetchDataPoints`, handing each page to `onPage` as it + * arrives instead of collecting the whole collection first. + * + * A dense collection has to be read this way. Intraday heart rate is one + * point a minute, so a full-history backfill of two years is over a million + * points; held as parsed JSON and then again as mapped readings, that is more + * than a 1 GB heap, and the process that runs the backfill also serves the + * app. Page by page, only one page is ever resident. + * + * `onPage` is awaited before the next page is requested, so a slow write + * applies back-pressure to the fetch. The date-filter fallback still works: + * it only fires when the first request is rejected, which is before any page + * reaches `onPage`. + */ +export async function forEachDataPointPage( + dataType: GoogleHealthDataType, + accessToken: string, + verb: string, + query: DataPointQuery, + onPage: (points: GoogleHealthDataPoint[]) => void | Promise, +): Promise { const maxPages = query.maxPages ?? 1000; const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE; @@ -722,8 +765,8 @@ export async function fetchDataPoints( const walk = async ( dateStyle: GoogleHealthDateFilterStyle, - ): Promise => { - const points: GoogleHealthDataPoint[] = []; + ): Promise => { + let fetched = 0; let pageToken: string | null | undefined; let pageCount = 0; @@ -780,13 +823,14 @@ export async function fetchDataPoints( }); } - for (const p of json?.dataPoints ?? []) points.push(p); + const page = json?.dataPoints ?? []; pageToken = json?.nextPageToken ?? null; pageCount += 1; + fetched += page.length; + await onPage(page); } while (pageToken && pageCount < maxPages); - notePagination(pageCount, points.length, Boolean(pageToken)); - return points; + notePagination(pageCount, fetched, Boolean(pageToken)); }; // The fallback only exists for filtered daily-summary reads — every other @@ -795,9 +839,8 @@ export async function fetchDataPoints( dataType.timeField === "date" && query.start !== undefined; let style: GoogleHealthDateFilterStyle = "camel"; - let points: GoogleHealthDataPoint[]; try { - points = await walk("camel"); + await walk("camel"); } catch (err) { const firstRequestRejected = canFallBack && @@ -806,13 +849,12 @@ export async function fetchDataPoints( err.httpStatus === 400; if (!firstRequestRejected) throw err; style = "snake"; - points = await walk("snake"); + await walk("snake"); } if (canFallBack) { annotate({ meta: { "googleHealth.dateFilter.style": style } }); query.onDateFilterStyle?.(style); } - return points; } // ─── Daily roll-up reads (`POST …/dataPoints:dailyRollUp`) ───────────────── diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index bef34d971..3b478e8d4 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -265,7 +265,12 @@ export async function runWithGoogleHealthHardFailLedger( * AsyncLocalStorage so concurrent per-user jobs never cross-pollinate. */ interface RollupDeferTracker { - keys: Array<{ type: MeasurementType; measuredAt: Date }>; + /** + * One entry per touched `(type, UTC day)`, never one per reading. A + * full-history heart-rate backfill touches over a million readings but only + * a few hundred days; a per-reading list grew with the stream. + */ + keys: Map; } const rollupDeferStorage = new AsyncLocalStorage(); @@ -286,7 +291,7 @@ export async function runWithGoogleHealthSyncCycle( ): Promise> { const softSkips: SoftSkipTracker = { count: 0 }; const hardFailures: HardFailTracker = { failures: [] }; - const deferredRollups: RollupDeferTracker = { keys: [] }; + const deferredRollups: RollupDeferTracker = { keys: new Map() }; const result = await softSkipStorage.run(softSkips, () => hardFailStorage.run(hardFailures, () => rollupDeferStorage.run(deferredRollups, fn), @@ -296,7 +301,7 @@ export async function runWithGoogleHealthSyncCycle( result, softSkipCount: softSkips.count, hardFailures: hardFailures.failures, - deferredRollupKeys: deferredRollups.keys, + deferredRollupKeys: Array.from(deferredRollups.keys.values()), }; } @@ -831,14 +836,15 @@ export async function upsertGoogleHealthMeasurements( // of the cycle. The incremental path keeps the inline per-day hook here. if (opts.deferRollup) { const tracker = rollupDeferStorage.getStore(); - // Plain loop, never `push(...touched)`: `touched` carries ONE entry PER - // READING, and a full-history batch on a sample-dense account (a - // multi-year heart-rate series is six figures of rows) blows the call - // stack when spread into a single call — the engine caps argument counts. - // The RangeError aborted the whole metrics collection on exactly the - // accounts with the most data. + // Collapsed to type-days on the way in: `touched` carries ONE entry PER + // READING, and the tracker lives for the whole cycle, so keeping the raw + // entries made the ledger grow with the stream (a million-reading + // heart-rate backfill held a million of them to describe a few hundred + // days). if (tracker) { - for (const t of touched) tracker.keys.push(t); + for (const k of collapseToTypeDayKeys(touched)) { + tracker.keys.set(`${k.type}|${k.measuredAt.getTime()}`, k); + } } noteGoogleHealthWritten(imported); return { imported, touched, inserted: insertedRows }; diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 6c204c7ab..d253ed437 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -27,9 +27,11 @@ */ import { GOOGLE_HEALTH_DATA_TYPES, + GOOGLE_HEALTH_DENSE_MAX_PAGES, type GoogleHealthDataType, type GoogleHealthMappedMeasurement, fetchDataPoints, + forEachDataPointPage, mapBloodGlucose, mapBodyFat, mapCoreBodyTemperature, @@ -61,6 +63,8 @@ interface MetricResource { tz: string, ) => GoogleHealthMappedMeasurement[]; verb: string; + /** Page ceiling for the walk; the client default when omitted. */ + maxPages?: number; } /** The launch metric resources (Measurement-producing). Height handled separately. */ @@ -99,6 +103,7 @@ const METRIC_RESOURCES: MetricResource[] = [ dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, map: mapHeartRate, verb: "fetchHeartRate", + maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES, }, { dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose, @@ -139,48 +144,52 @@ export async function syncUserMetrics( // Each metric data type is fetched + mapped independently so a per-class 403 // soft-skips only that class. + // + // Every type is written page by page as it is read. Intraday heart rate is + // one point a minute, so a full-history backfill is over a million points; + // collecting the whole collection first and mapping it second held the raw + // points and the readings at once and ran a 1 GB heap out of memory on + // exactly the accounts with the most history. A page is at most + // `GOOGLE_HEALTH_PAGE_SIZE` points, and nothing from it outlives its write. + // + // The mapper runs inside the same per-type catch: a single malformed point + // whose `resource.map(point)` throws must not escape the METRIC_RESOURCES + // loop and skip every metric type ordered after it (which also blocked the + // watermark, so the bad point refetched hourly and those types stayed dead). + // A map throw goes through the same ledger as a fetch failure — record it, + // fail the cycle, and move on to the next type. Pages already written before + // the failure stay written: every write is keyed, so the retry that the + // failed verdict causes overwrites them in place. for (const resource of METRIC_RESOURCES) { - let points: Record[]; try { - points = await fetchDataPoints( + await forEachDataPointPage( resource.dataType, tokenInfo.accessToken, resource.verb, - { start }, + { start, maxPages: resource.maxPages }, + async (points) => { + const readings: GoogleHealthMeasurementUpsert[] = []; + for (const point of points) { + for (const m of resource.map(point, tz)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + }); + } + } + imported += ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + }, ); } catch (err) { imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; } - - // The mapper runs INSIDE a per-type catch too: a single malformed point - // whose `resource.map(point)` throws must not escape the METRIC_RESOURCES - // loop and skip every metric type ordered after it (which also blocked the - // watermark, so the bad point refetched hourly and those types stayed dead). - // Route a map throw through the same ledger as a fetch failure — record it, - // fail the cycle, and move on to the next type. - const readings: GoogleHealthMeasurementUpsert[] = []; - try { - for (const point of points) { - for (const m of resource.map(point, tz)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: m.fieldTag, - }); - } - } - } catch (err) { - imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; - } - imported += ( - await upsertGoogleHealthMeasurements(userId, readings, { - deferRollup: opts.deferRollup, - }) - ).imported; } // Height → User.heightCm, only when the user has no height yet. Never mint a diff --git a/tests/integration/google-health-dense-backfill-memory.test.ts b/tests/integration/google-health-dense-backfill-memory.test.ts new file mode 100644 index 000000000..31414108e --- /dev/null +++ b/tests/integration/google-health-dense-backfill-memory.test.ts @@ -0,0 +1,247 @@ +/** + * #1023 — a full-history Google Health backfill over a dense heart-rate + * stream, with its live memory measured rather than assumed. + * + * What this pins. The backfill read every heart-rate point of the account's + * history into one array, mapped all of it into a second array, and wrote the + * lot in one call, so the raw points, the readings and the write's working + * sets were resident together and all of them scaled with the history. A + * watch that records a reading a minute passes a million points in two years: + * reproduced on 1.3 M readings under `--max-old-space-size=1024`, the backfill + * reaches `FATAL ERROR: Reached heap limit` within seconds, and because the + * backfill lane shares the app process, the whole instance restarts. The walk + * now writes each page as it arrives, and the cycle's deferred-rollup ledger + * keeps one key per touched day instead of one per reading. + * + * How it measures. Every heap reading is taken after a forced collection, so + * what is compared is what the code HOLDS, not how much garbage V8 chose to + * leave lying about. And what is compared is GROWTH: the heap once a quarter + * of the pages have been read, against the most it reaches afterwards. The + * first pages carry a fixed cost that has nothing to do with the stream + * (modules loaded on first use, the ORM's compiled queries), and that cost is + * about the size of this fixture, so an absolute figure could not tell a + * streaming walk from one that holds everything. Growth can: a walk that + * holds one page stays flat, a walk that holds what it has read climbs with + * every page. + * + * The two halves are the point: the backfill stays inside the budget, and + * collecting the same collection into one array (what the backfill used to do + * before anything else) does not. Without the second half the budget could + * be any number and the first would still pass. + * + * Google's API is replaced at the `safeFetch` seam by a generator that builds + * each page on request, so the fixture itself never sits in memory. + */ +import v8 from "node:v8"; +import vm from "node:vm"; + +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "gh-dense-backfill-owner"; +const POINTS = 300_000; +const NOW = Date.parse("2026-09-27T17:00:00.000Z"); + +/** + * How much the heap may grow between the first quarter of the walk and its + * end. Measured, not chosen: the paged walk grows by a few MB on this fixture + * (noise around a flat line), and collecting the same pages into one array + * grows by about 80 MB over the same stretch. + */ +const GROWTH_BUDGET_BYTES = 24 * 1024 * 1024; + +const fake = vi.hoisted(() => ({ + total: 0, + now: 0, + heartRatePages: 0, + /** Called as each heart-rate page is requested (the sampler's hook). */ + onPage: null as (() => void) | null, +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + // A plain function, not `vi.fn`: a mock records every call's result, and + // here each result is a whole page's response, so the recorder alone + // would hold the collection this file exists to keep out of memory. + safeFetch: async (url: string) => { + const u = new URL(url); + let body: unknown = {}; + if (u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) { + const pageSize = Number(u.searchParams.get("pageSize") ?? 1000); + const page = Number(u.searchParams.get("pageToken") ?? 0); + const end = Math.min(fake.total, (page + 1) * pageSize); + const dataPoints = []; + for (let g = page * pageSize; g < end; g++) { + dataPoints.push({ + name: `users/me/dataTypes/heart-rate/dataPoints/${9_000_000_000_000 + g}`, + dataSource: { + recordingMethod: "PASSIVE_MEASUREMENT", + device: { manufacturer: "Google", displayName: "Watch" }, + platform: "FITBIT", + }, + heartRate: { + beatsPerMinute: String(55 + (g % 50)), + sampleTime: { + physicalTime: new Date(fake.now - g * 60_000).toISOString(), + }, + }, + }); + } + fake.heartRatePages++; + fake.onPage?.(); + body = { + dataPoints, + ...(end < fake.total ? { nextPageToken: String(page + 1) } : {}), + }; + } + return new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }, + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const forceGc = ((): (() => void) => { + v8.setFlagsFromString("--expose-gc"); + const gc = vm.runInNewContext("gc") as () => void; + v8.setFlagsFromString("--no-expose-gc"); + return gc; +})(); + +function liveHeapBytes(): number { + forceGc(); + forceGc(); + return process.memoryUsage().heapUsed; +} + +const mb = (bytes: number): string => (bytes / 1024 / 1024).toFixed(1); + +function report(line: string): void { + process.stderr.write(`[gh-dense-backfill] ${line}\n`); +} + +/** + * Run `fn` while sampling the collected heap every few pages, from inside the + * page request, so each sample lands between two pages of the work being + * measured: exactly where anything it holds across a page is visible. (A + * timer is not enough; a walk whose pages resolve without touching the event + * loop never lets it fire.) Returns how far the heap rose after the walk + * passed a quarter of its pages. + */ +async function growthPastFirstQuarter( + fn: () => Promise, +): Promise<{ result: T; growth: number; samples: number }> { + const quarter = Math.ceil(POINTS / 1000 / 4); + const startPages = fake.heartRatePages; + let anchor: number | null = null; + let peak = 0; + let samples = 0; + const sample = (): void => { + const live = liveHeapBytes(); + if (anchor === null) { + if (fake.heartRatePages - startPages >= quarter) anchor = peak = live; + return; + } + samples++; + peak = Math.max(peak, live); + }; + fake.onPage = () => { + if ((fake.heartRatePages - startPages) % 5 === 0) sample(); + }; + try { + const result = await fn(); + sample(); + return { + result, + growth: anchor === null ? Number.NaN : peak - anchor, + samples, + }; + } finally { + fake.onPage = null; + } +} + +beforeAll(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: "Europe/Berlin" }, + }); + await prisma.googleHealthConnection.create({ + data: { + userId: USER, + googleUserId: `google-${USER}`, + accessToken: encrypt("access"), + refreshToken: encrypt("refresh"), + tokenExpiresAt: new Date(Date.now() + 86_400_000), + }, + }); + fake.total = POINTS; + fake.now = NOW; +}, 120_000); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("Google Health backfill over a dense heart-rate stream (#1023)", () => { + it("writes the whole history without growing with it", async () => { + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + const { result, growth, samples } = await growthPastFirstQuarter(() => + runGoogleHealthBackfillForUser(USER), + ); + report( + `backfill: imported=${result.imported} growth=${mb(growth)} MB ` + + `over ${samples} samples (budget ${mb(GROWTH_BUDGET_BYTES)} MB)`, + ); + + const prisma = getPrismaClient(); + expect( + await prisma.measurement.count({ + where: { userId: USER, type: "PULSE", source: "GOOGLE_HEALTH" }, + }), + ).toBe(POINTS); + const connection = await prisma.googleHealthConnection.findUniqueOrThrow({ + where: { userId: USER }, + select: { backfillCompletedAt: true }, + }); + expect(connection.backfillCompletedAt).not.toBeNull(); + // The deferred fold reached the rollup tier for the whole span. + const days = await prisma.measurementRollup.count({ + where: { userId: USER, type: "PULSE", granularity: "DAY" }, + }); + expect(days).toBeGreaterThanOrEqual(Math.floor(POINTS / 1440)); + + expect(samples).toBeGreaterThan(5); + expect(growth).toBeLessThan(GROWTH_BUDGET_BYTES); + }, 900_000); + + it("the same pages collected into one array outgrow the budget", async () => { + const { fetchDataPoints, GOOGLE_HEALTH_DATA_TYPES } = + await import("@/lib/google-health/client"); + let held: unknown[] = []; + const { growth, samples } = await growthPastFirstQuarter(async () => { + held = await fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "access", + "fetchHeartRate", + ); + }); + report( + `collected: points=${held.length} growth=${mb(growth)} MB over ${samples} samples`, + ); + expect(held).toHaveLength(POINTS); + expect(growth).toBeGreaterThan(GROWTH_BUDGET_BYTES); + }, 900_000); +});