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, {