fix(google-health): write a dense collection page by page during a backfill

A full-history backfill read every heart-rate point of an account into one
array, mapped it into a second, and wrote the lot in one call; the cycle's
deferred-rollup ledger then kept one entry per written reading. On a watch
that records a reading a minute that is over a million points, and the
backfill ran a 1 GB heap out of memory within seconds, restarting the
instance it shares with the app (#1023).

The metric walk now hands each page to its write before requesting the
next, and both the Google Health and Fitbit ledgers collapse to one key per
touched type-day as they record. The heart-rate walk's page ceiling rises
from 1000 to 10000: at one point a minute the old ceiling is under two
years, a longer history always ended truncated, and the backfill retried
the whole walk on every attempt and every boot without completing.
This commit is contained in:
Marc-André Bombeck committed 2026-09-27 20:42:54 +02:00
1 parent c2a1522634
commit e34741c077
9 files changed
+598 -67

No files matched your search

+11 -8
View File
@@ -228,7 +228,11 @@ export const softSkipStorage = new AsyncLocalStorage<SoftSkipTracker>();
* 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<string, { type: MeasurementType; measuredAt: Date }>;
}
export const rollupDeferStorage = new AsyncLocalStorage<RollupDeferTracker>();
@@ -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 };
}
+4 -7
View File
@@ -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())
@@ -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",
]);
});
});
@@ -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<typeof import("@/lib/safe-fetch")>();
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}`,
]);
});
});
@@ -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<void>,
) => {
const points = (await fetchDataPointsMock(
dataType,
token,
verb,
query,
)) as unknown[];
if (points.length > 0) await onPage(points);
},
),
mapWeight: mapWeightMock,
};
});
+51 -9
View File
@@ -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<GoogleHealthDataPoint[]> {
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<void>,
): Promise<void> {
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<GoogleHealthDataPoint[]> => {
const points: GoogleHealthDataPoint[] = [];
): Promise<void> => {
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`) ─────────────────
+16 -10
View File
@@ -265,7 +265,12 @@ export async function runWithGoogleHealthHardFailLedger<T>(
* 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<string, { type: MeasurementType; measuredAt: Date }>;
}
const rollupDeferStorage = new AsyncLocalStorage<RollupDeferTracker>();
@@ -286,7 +291,7 @@ export async function runWithGoogleHealthSyncCycle<T>(
): Promise<GoogleHealthSyncCycleResult<T>> {
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<T>(
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 };
+42 -33
View File
@@ -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<string, unknown>[];
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