From 62c72782753e6b1035ec33463ff2bba8338fe2d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 8 Jul 2026 23:40:36 +0200 Subject: [PATCH] fix(fitbit): port the non-rethrowing hard-fail ledger from Google Health MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fitbit's handleCollectionFetchError soft-skipped a 403 but RETHREW any other hard failure, aborting every endpoint ordered after the bad one in the same resource and stalling the watermark so the whole integration stayed dead. Record hard failures on a per-cycle AsyncLocalStorage ledger and return 0 instead of rethrowing — the sibling endpoints keep fetching, and the ledger still fails the cycle so success is not stamped and the next tick refetches. Also wrap the per-metric map fan-out in the same per-collection catch so one malformed point can't skip its siblings. --- .../fitbit/__tests__/sync-user-fitbit.test.ts | 50 ++++++++++++ src/lib/fitbit/sync-activity.ts | 29 +++++-- src/lib/fitbit/sync-metrics.ts | 29 +++++-- src/lib/fitbit/sync.ts | 62 ++++++++++---- .../__tests__/sync-metrics-failsoft.test.ts | 81 +++++++++++++++++++ src/lib/google-health/sync-metrics.ts | 29 ++++--- 6 files changed, 240 insertions(+), 40 deletions(-) create mode 100644 src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts diff --git a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts index e138ab964..4d5a6a81e 100644 --- a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts +++ b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts @@ -190,6 +190,56 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => { }); }); +/** A resource sync that HARD-fails its collection (non-403) and, thanks to the + * ported ledger, records + returns 0 instead of rethrowing. */ +async function hardFail500(...args: unknown[]): Promise { + return handleCollectionFetchError( + "test-resource", + String(args[0]), + new FitbitApiError({ + verb: "fetchTest", + classification: "transient", + httpStatus: 500, + reason: "http_500", + }), + ); +} + +describe("syncUserFitbit — hard-fail ledger (F-SYNC-4)", () => { + it("a non-403 hard failure does NOT rethrow and keeps siblings running", async () => { + // Metrics hard-fails (500) but must not abort the cycle: the sibling + // resources still run, and the whole call resolves (no rethrow). + syncUserMetrics.mockImplementation(hardFail500); + syncUserActivity.mockResolvedValue(2); + syncUserSleep.mockResolvedValue(1); + syncUserWorkout.mockResolvedValue(0); + + // Resolves (no rethrow) with the siblings' imports. + const total = await syncUserFitbit("user1"); + expect(total).toBe(3); + + // Every sibling resource still ran despite the metrics hard failure. + expect(syncUserActivity).toHaveBeenCalled(); + expect(syncUserSleep).toHaveBeenCalled(); + expect(syncUserWorkout).toHaveBeenCalled(); + }); + + it("a hard-failed collection fails the cycle: no success stamp, no watermark", async () => { + syncUserMetrics.mockImplementation(hardFail500); + // Others import so the run is not degenerate — only the ledger keeps it honest. + syncUserActivity.mockResolvedValue(4); + + await syncUserFitbit("user1"); + + // The ledger flips anyFailed → success is NOT stamped and the watermark is + // not advanced, so the next tick refetches the broken collection's window. + expect(recordSyncSuccess).not.toHaveBeenCalled(); + expect(update).not.toHaveBeenCalled(); + // The hard failure was recorded on the status ledger. + expect(recordSyncFailure).toHaveBeenCalled(); + }); +}); + describe("syncUserFitbit — single cycle-wide watermark", () => { it("snapshots the incremental start ONCE and threads the SAME watermark to every resource", async () => { syncUserMetrics.mockResolvedValue(1); diff --git a/src/lib/fitbit/sync-activity.ts b/src/lib/fitbit/sync-activity.ts index d9dec20db..4a31e30a1 100644 --- a/src/lib/fitbit/sync-activity.ts +++ b/src/lib/fitbit/sync-activity.ts @@ -109,14 +109,27 @@ export async function syncUserActivity( forbidden = true; break; } - for (const m of resource.map(body)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: externalIdFor(m), - }); + try { + for (const m of resource.map(body)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: externalIdFor(m), + }); + } + } catch (err) { + // A malformed point whose mapper throws routes through the same ledger + // as a fetch failure — record it, short-circuit this resource, and let + // the sibling resources keep syncing. + imported += await handleCollectionFetchError( + resource.verb, + userId, + err, + ); + forbidden = true; + break; } } if (forbidden) continue; diff --git a/src/lib/fitbit/sync-metrics.ts b/src/lib/fitbit/sync-metrics.ts index f05ae2fb8..ec26bce68 100644 --- a/src/lib/fitbit/sync-metrics.ts +++ b/src/lib/fitbit/sync-metrics.ts @@ -103,14 +103,27 @@ export async function syncUserMetrics( forbidden = true; break; } - for (const m of resource.map(body)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: m.fieldTag, - }); + try { + for (const m of resource.map(body)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + }); + } + } catch (err) { + // A malformed point whose mapper throws routes through the same ledger + // as a fetch failure — record it, short-circuit this metric, and let the + // sibling metrics keep syncing. + imported += await handleCollectionFetchError( + resource.verb, + userId, + err, + ); + forbidden = true; + break; } } if (forbidden) continue; diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index ae3c493cf..8411831df 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -227,12 +227,34 @@ interface RollupDeferTracker { } const rollupDeferStorage = new AsyncLocalStorage(); +/** + * Per-cycle ledger of collection fetches that HARD-failed (non-403). A hard + * failure on one endpoint must not abort its siblings — weight 500ing must not + * suppress body-fat / spo2 / hrv / rhr / respiratory in the same resource — so + * `handleCollectionFetchError` records the failure here and RETURNS instead of + * rethrowing (the old pattern that aborted every endpoint ordered after the bad + * one and stalled the watermark). The orchestrator reads the ledger to keep the + * cycle's verdict honest: any entry ⇒ the run counts as failed, so success is + * NOT stamped and the next tick refetches the window. Ported from Google + * Health's `hardFailStorage`. + */ +interface HardFailTracker { + failures: string[]; +} +const hardFailStorage = new AsyncLocalStorage(); + /** * Single-source the per-resource collection-fetch error handling. A 403 on one - * data class soft-skips it (warn + return 0) so sibling resources still sync; - * anything else records a classified sync failure and rethrows. A soft-skip - * increments the ambient tracker so `syncUserFitbit` can refuse to stamp success - * on an all-403 grant-revoke cycle that imported nothing. + * data class soft-skips it (warn + return 0) so sibling resources still sync; a + * soft-skip increments the ambient tracker so `syncUserFitbit` can refuse to + * stamp success on an all-403 grant-revoke cycle that imported nothing. + * + * Anything else records a classified sync failure, notes the collection on the + * ambient hard-fail ledger, and RETURNS 0 — it does NOT rethrow. Every call site + * sits in a per-collection catch, so returning lets the sibling endpoints in the + * same resource keep fetching (one 500 on weight must not kill body-fat / spo2 / + * hrv / rhr / respiratory), while the ledger still fails the cycle so the + * watermark is not stamped past the broken collection. */ export async function handleCollectionFetchError( resource: string, @@ -248,7 +270,10 @@ export async function handleCollectionFetchError( return 0; } await recordFitbitSyncFailure(userId, err); - throw err; + getEvent()?.addWarning(`fitbit ${resource} failed for ${userId}: ${err}`); + const hardTracker = hardFailStorage.getStore(); + if (hardTracker) hardTracker.failures.push(resource); + return 0; } /** @@ -640,22 +665,29 @@ export async function syncUserFitbit( const tracker: SoftSkipTracker = { count: 0 }; const deferTracker: RollupDeferTracker = { keys: [] }; + const hardFailTracker: HardFailTracker = { failures: [] }; let total = 0; let anyFailed = false; await softSkipStorage.run(tracker, async () => { - await rollupDeferStorage.run(deferTracker, async () => { - for (const fn of resources) { - try { - total += await fn(userId, resourceOpts); - } catch (err) { - anyFailed = true; - getEvent()?.addWarning( - `fitbit ${fn.name} failed for ${userId}: ${err}`, - ); + await hardFailStorage.run(hardFailTracker, async () => { + await rollupDeferStorage.run(deferTracker, async () => { + for (const fn of resources) { + try { + total += await fn(userId, resourceOpts); + } catch (err) { + anyFailed = true; + getEvent()?.addWarning( + `fitbit ${fn.name} failed for ${userId}: ${err}`, + ); + } } - } + }); }); }); + // A collection that hard-failed inside a resource (recorded on the ledger by + // `handleCollectionFetchError` without aborting its siblings) still fails the + // cycle: partial error, no watermark stamp. + if (hardFailTracker.failures.length > 0) anyFailed = true; // On a `fullSync` backfill the per-write inline rollup hook was deferred; the // accumulated touched type-days collapse into ONE range-recompute spanning the diff --git a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts new file mode 100644 index 000000000..ccb60412a --- /dev/null +++ b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts @@ -0,0 +1,81 @@ +/** + * F-SYNC-3 — the metric MAPPER runs inside a per-type catch, not just the + * fetch. A single malformed point whose `resource.map(point)` throws used to + * escape the METRIC_RESOURCES loop and skip every metric type ordered after it + * (body-fat, spo2, hrv, rhr, respiratory, glucose, temps), also blocking the + * watermark so the bad point refetched hourly and those types stayed dead. + * + * This pins the contract: a throwing mapper on the first type routes through + * `handleCollectionFetchError` and the loop continues to fetch every sibling. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { fetchDataPointsMock, handleErrorMock, mapWeightMock } = vi.hoisted( + () => ({ + fetchDataPointsMock: vi.fn(), + handleErrorMock: vi.fn(async () => 0), + mapWeightMock: vi.fn(() => { + throw new Error("malformed weight point"); + }), + }), +); + +vi.mock("../client", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + fetchDataPoints: fetchDataPointsMock, + mapWeight: mapWeightMock, + }; +}); + +vi.mock("../sync", () => ({ + getValidToken: vi.fn(async () => ({ + accessToken: "token", + connection: { id: "c1", googleUserId: "g1" }, + })), + handleCollectionFetchError: handleErrorMock, + upsertGoogleHealthMeasurements: vi.fn(async () => ({ + imported: 0, + touched: [], + })), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: vi.fn(), update: vi.fn() } }, +})); + +import { syncUserMetrics } from "../sync-metrics"; + +beforeEach(() => { + handleErrorMock.mockClear(); + mapWeightMock.mockClear(); + // Weight returns one point (whose mapper throws); everything else is empty so + // its real mapper is never exercised. + fetchDataPointsMock.mockReset().mockImplementation( + async (_dt: unknown, _token: string, verb: string) => + verb === "fetchWeight" ? [{ any: "point" }] : [], + ); +}); + +describe("syncUserMetrics — a throwing mapper never skips sibling types", () => { + it("routes the map throw through the ledger and keeps fetching the rest", async () => { + await expect(syncUserMetrics("user-1")).resolves.toBe(0); + + // The weight mapper threw and was routed exactly once. + expect(mapWeightMock).toHaveBeenCalled(); + expect(handleErrorMock).toHaveBeenCalledWith( + "fetchWeight", + "user-1", + expect.any(Error), + ); + + // Every metric type ordered AFTER weight still fetched — the loop was not + // aborted by the throw. + const verbs = fetchDataPointsMock.mock.calls.map((c) => c[2]); + expect(verbs).toContain("fetchBodyFat"); + expect(verbs).toContain("fetchSpo2"); + expect(verbs).toContain("fetchRespiratoryRate"); + expect(verbs).toContain("fetchHeartRate"); + }); +}); diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 408ab7adb..9d366ea38 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -146,17 +146,28 @@ export async function syncUserMetrics( 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[] = []; - for (const point of points) { - for (const m of resource.map(point)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: m.fieldTag, - }); + try { + for (const point of points) { + for (const m of resource.map(point)) { + 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, {