fix(polar): isolate each vitals collection so one dead read no longer blanks the rest

Restructure the Polar vitals sync from a single Promise.all over the five
collections into a Promise.allSettled over a per-collection spec, each closure
fetching and mapping so a throwing mapper is contained exactly like a failing
fetch. The healthy collections import; a failing one is recorded once as a
partial failure that names it, without stamping success and without rethrowing,
so a single unavailable data type — SpO2 on a watch without the sensor, a
higher-tier cardio-load read an account lacks — stops blanking recharge, sleep,
and activity on every tick.

A 401 is now a per-collection failure classified reauth_required, not a
whole-leg throw; the write-path escape still surfaces unmarked for the poll
boundary to record once. Mirrors the shipped Oura isolation, minus the
token-refresh leg Polar has no path for.

src/lib/jobs/reminder/__tests__/polar-sync-legs.test.ts

@@ -30,6 +30,18 @@ describe("syncUserPolarLegs — composite independence on the shared polar ledge
     expect(syncUserPolarWorkouts).toHaveBeenCalledWith("u1");
   });
 
+  it("sums both counts without throwing when the vitals leg returns a partial-failure count", async () => {
+    // Post-R7 the vitals leg records a partial failure INTERNALLY and RETURNS
+    // its healthy-collections import count rather than throwing. The wrapper
+    // must sum it with the workout leg and complete cleanly — a partial vitals
+    // failure never starves the user's success accounting at the cohort level.
+    syncUserPolar.mockResolvedValue(4);
+    syncUserPolarWorkouts.mockResolvedValue(2);
+    await expect(syncUserPolarLegs("u1")).resolves.toBe(6);
+    expect(syncUserPolar).toHaveBeenCalledWith("u1");
+    expect(syncUserPolarWorkouts).toHaveBeenCalledWith("u1");
+  });
+
   it("still attempts the workout leg when the vitals leg fails (vice-versa isolation)", async () => {
     syncUserPolar.mockRejectedValue(new Error("vitals down"));
     syncUserPolarWorkouts.mockResolvedValue(3);

src/lib/jobs/reminder/polar-sync.ts

@@ -27,6 +27,12 @@ export type PolarSyncPayload = PollCohortPayload;
  * always attempted, so a workouts-leg failure can never mask a vitals success
  * and vice versa — then the FIRST error is rethrown so the cohort handler warns
  * and skips this user's success accounting without losing the error.
+ *
+ * The vitals leg (`syncUserPolar`) now settles its five collections
+ * independently: a fetch/map failure on one collection is recorded as a partial
+ * failure and the leg RETURNS (no rethrow, no success stamp), importing the
+ * healthy collections. So the vitals leg reaching this wrapper's catch means a
+ * write-path/unexpected error, not a collection fetch failure.
  */
 export async function syncUserPolarLegs(userId: string): Promise<number> {
   let total = 0;
@@ -63,8 +69,10 @@ export const handlePolarSync = makePollCohortHandler({
     return users.map((u) => u.id);
   },
   syncUser: syncUserPolarLegs,
-  // Catches an UNMARKED escape — today the `upsertPolarMeasurements` /
-  // `upsertPolarWorkouts` write-path throws (the fetch paths record + mark
-  // their own failures, and the legs wrapper rethrows the marked error).
+  // Catches an UNMARKED escape — the `upsertPolarMeasurements` /
+  // `upsertPolarWorkouts` write-path throw. The vitals leg records its own
+  // per-collection partial failures inline and returns without rethrowing; the
+  // workouts leg records + marks its own fetch failures. Only an unmarked
+  // write-path/unexpected escape reaches this boundary recorder.
   recordFailure: recordPolarSyncFailure,
 });

src/lib/polar/__tests__/sync.test.ts

@@ -331,8 +331,9 @@ describe("syncUserPolar", () => {
     );
   });
 
-  it("records a reauth_required failure on a 401 and rethrows", async () => {
+  it("records a reauth_required partial failure on a 401 and does NOT throw (per-collection isolation)", async () => {
     getConnMock.mockResolvedValue(CONN);
+    // The recharge collection 401s; the four siblings return data.
     fetchRechargesMock.mockRejectedValue(
       new PolarApiError({
         verb: "fetchNightlyRecharges",
@@ -341,11 +342,17 @@ describe("syncUserPolar", () => {
         reason: "http_401",
       }),
     );
-    const err = await syncUserPolar("u1").catch((e) => e);
-    expect(err).toBeInstanceOf(PolarApiError);
-    // Recorded at the source AND marked (the legs wrapper rethrows the marked
-    // error), so the poll-cohort boundary does not double-record it.
-    expect(isSyncFailureRecorded(err)).toBe(true);
+    fetchCardioLoadsMock.mockResolvedValue([
+      { date: "2026-06-10", cardio_load: 42 },
+    ]);
+
+    const imported = await syncUserPolar("u1");
+
+    // No rethrow now — a fetch failure is isolated per-collection.
+    expect(imported).toBe(1);
+    expect(createdRows().map((row) => row.type)).toContain("CARDIO_LOAD");
+    // Exactly one partial failure recorded, classified off the first error,
+    // naming the failed collection.
     expect(recordFailureMock).toHaveBeenCalledTimes(1);
     expect(recordFailureMock).toHaveBeenCalledWith(
       expect.objectContaining({
@@ -353,11 +360,206 @@ describe("syncUserPolar", () => {
         integration: "polar",
         kind: "reauth_required",
         errorCode: "401",
+        message: expect.stringContaining("recharge"),
       }),
     );
     expect(recordSuccessMock).not.toHaveBeenCalled();
   });
 
+  it("isolates one dead collection — imports the healthy four and records one named failure", async () => {
+    getConnMock.mockResolvedValue(CONN);
+    // SpO2 403s (no sensor / consent), the other four return one row each.
+    fetchSpo2Mock.mockRejectedValue(
+      new PolarApiError({
+        verb: "fetchSpo2",
+        classification: "reauth_required",
+        httpStatus: 403,
+        reason: "http_403",
+      }),
+    );
+    fetchRechargesMock.mockResolvedValue([
+      { date: "2026-06-10", nightly_recharge_status: 6 },
+    ]);
+    fetchSleepsMock.mockResolvedValue([
+      {
+        date: "2026-06-10",
+        sleep_start_time: "2026-06-09T23:00:00+02:00",
+        sleep_end_time: "2026-06-10T07:00:00+02:00",
+        light_sleep: 3600,
+      },
+    ]);
+    fetchActivitiesMock.mockResolvedValue([
+      { start_time: "2026-06-10T00:00:00", steps: 8000 },
+    ]);
+    fetchCardioLoadsMock.mockResolvedValue([
+      { date: "2026-06-10", cardio_load: 55 },
+    ]);
+
+    const imported = await syncUserPolar("u1");
+
+    // The healthy four still wrote their rows — the source is not blanked.
+    const written = createdRows().map((row) => row.type);
+    expect(written).toContain("RECOVERY_SCORE");
+    expect(written).toContain("SLEEP_DURATION");
+    expect(written).toContain("ACTIVITY_STEPS");
+    expect(written).toContain("CARDIO_LOAD");
+    expect(written).not.toContain("OXYGEN_SATURATION");
+    expect(imported).toBeGreaterThan(0);
+    // Exactly one failure, classified off the 403, naming spo2. No success.
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
+    expect(recordFailureMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        integration: "polar",
+        kind: "reauth_required",
+        errorCode: "403",
+        message: expect.stringContaining("spo2"),
+      }),
+    );
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+  });
+
+  it("isolates a throwing mapper — one malformed record does not blank siblings", async () => {
+    getConnMock.mockResolvedValue(CONN);
+    // A malformed recharge record whose mapper throws on first field access
+    // must not reject the whole batch (the old Promise.all failure mode).
+    fetchRechargesMock.mockResolvedValue([
+      new Proxy(
+        {},
+        {
+          get() {
+            throw new Error("malformed recharge point");
+          },
+        },
+      ),
+    ]);
+    fetchCardioLoadsMock.mockResolvedValue([
+      { date: "2026-06-10", cardio_load: 77 },
+    ]);
+
+    const imported = await syncUserPolar("u1");
+
+    expect(imported).toBe(1);
+    expect(createdRows().map((row) => row.type)).toContain("CARDIO_LOAD");
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
+    expect(recordFailureMock).toHaveBeenCalledWith(
+      expect.objectContaining({ message: expect.stringContaining("recharge") }),
+    );
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+  });
+
+  it("sleep-collection failure skips the sweep, siblings still import", async () => {
+    getConnMock.mockResolvedValue(CONN);
+    fetchSleepsMock.mockRejectedValue(
+      new PolarApiError({
+        verb: "fetchSleeps",
+        classification: "transient",
+        httpStatus: 500,
+        reason: "http_500",
+      }),
+    );
+    fetchActivitiesMock.mockResolvedValue([
+      { start_time: "2026-06-10T00:00:00", steps: 8000 },
+    ]);
+
+    const imported = await syncUserPolar("u1");
+
+    // The sweep never ran — a failed sleep collection pushes no sweep entries.
+    expect(updateManyMock).not.toHaveBeenCalled();
+    // The activity sibling still imported.
+    expect(createdRows().map((row) => row.type)).toContain("ACTIVITY_STEPS");
+    expect(imported).toBeGreaterThan(0);
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+  });
+
+  it("all five healthy → success stamped once, no failure recorded", async () => {
+    getConnMock.mockResolvedValue(CONN);
+    fetchRechargesMock.mockResolvedValue([
+      { date: "2026-06-10", nightly_recharge_status: 6 },
+    ]);
+
+    await syncUserPolar("u1");
+
+    expect(recordFailureMock).not.toHaveBeenCalled();
+    expect(recordSuccessMock).toHaveBeenCalledTimes(1);
+    expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar");
+  });
+
+  it.each([
+    { name: "recharge", type: "RECOVERY_SCORE" },
+    { name: "sleep", type: "SLEEP_DURATION" },
+    { name: "activity", type: "ACTIVITY_STEPS" },
+    { name: "cardioload", type: "CARDIO_LOAD" },
+    { name: "spo2", type: "OXYGEN_SATURATION" },
+  ])(
+    "Oura-parity: a 4xx on the $name collection names it in the single failure and imports every sibling",
+    async ({ name }) => {
+      getConnMock.mockResolvedValue(CONN);
+      // Every collection returns one healthy row...
+      fetchRechargesMock.mockResolvedValue([
+        { date: "2026-06-10", nightly_recharge_status: 6 },
+      ]);
+      fetchSleepsMock.mockResolvedValue([
+        {
+          date: "2026-06-10",
+          sleep_start_time: "2026-06-09T23:00:00+02:00",
+          sleep_end_time: "2026-06-10T07:00:00+02:00",
+          light_sleep: 3600,
+        },
+      ]);
+      fetchActivitiesMock.mockResolvedValue([
+        { start_time: "2026-06-10T00:00:00", steps: 8000 },
+      ]);
+      fetchCardioLoadsMock.mockResolvedValue([
+        { date: "2026-06-10", cardio_load: 55 },
+      ]);
+      fetchSpo2Mock.mockResolvedValue([
+        { test_time: 1_781_059_600, blood_oxygen_percent: 96 },
+      ]);
+      // ...except the one under test, which 4xxs.
+      const mockByName: Record<string, ReturnType<typeof vi.fn>> = {
+        recharge: fetchRechargesMock,
+        sleep: fetchSleepsMock,
+        activity: fetchActivitiesMock,
+        cardioload: fetchCardioLoadsMock,
+        spo2: fetchSpo2Mock,
+      };
+      mockByName[name]!.mockRejectedValue(
+        new PolarApiError({
+          verb: name,
+          classification: "reauth_required",
+          httpStatus: 403,
+          reason: "http_403",
+        }),
+      );
+
+      await syncUserPolar("u1");
+
+      // Exactly one failure, naming this collection; no success stamp.
+      expect(recordFailureMock).toHaveBeenCalledTimes(1);
+      expect(recordFailureMock).toHaveBeenCalledWith(
+        expect.objectContaining({
+          integration: "polar",
+          message: expect.stringContaining(name),
+        }),
+      );
+      expect(recordSuccessMock).not.toHaveBeenCalled();
+      // Every sibling collection still imported at least one row.
+      const written = createdRows().map((row) => row.type);
+      const siblingTypes: Record<string, string> = {
+        recharge: "RECOVERY_SCORE",
+        sleep: "SLEEP_DURATION",
+        activity: "ACTIVITY_STEPS",
+        cardioload: "CARDIO_LOAD",
+        spo2: "OXYGEN_SATURATION",
+      };
+      for (const [collName, collType] of Object.entries(siblingTypes)) {
+        if (collName === name) continue;
+        expect(written).toContain(collType);
+      }
+    },
+  );
+
   it("leaves a write-path failure UNMARKED so the cohort boundary records it once", async () => {
     // The `upsertPolarMeasurements` transaction throw escapes `syncUserPolar`
     // uncaught: no source-site record, no mark — the poll-cohort boundary is

src/lib/polar/sync.ts

@@ -12,6 +12,22 @@
  * as a 401 → `reauth_required` on the shared integration ledger (`polar` key),
  * exactly like Nightscout's token-rejected path.
  *
+ * Isolation: the five vitals collections are settled independently
+ * (`Promise.allSettled`), so a single dead collection — e.g. a 403 on SpO2 or
+ * cardio-load on an account without the sensor or the higher tier — imports the
+ * healthy four and records ONE partial failure rather than blanking every
+ * collection. Mirrors Oura's shipped per-collection isolation, minus the
+ * token-refresh leg Polar structurally lacks (a 401 is just a per-collection
+ * failure the classifier maps to `reauth_required`).
+ *
+ * Window contract: every hourly tick re-reads the full server-fixed window for
+ * each collection — 28 days for recharge / sleep / activity / cardio-load /
+ * SpO2 (there is no narrower request and no pagination on any of them). A
+ * late-synced device, a re-scored night, or a HealthLog outage shorter than the
+ * window all self-heal on the next tick with no cursor or watermark. History
+ * older than the window is an upstream cap, not a HealthLog gap. See the
+ * client's per-fetcher docstrings for the verified per-endpoint windows.
+ *
  * Idempotency: each row's `externalId` is `<date>:<fieldTag>`. The shared
  * reconciler protects both external and natural identity. Polar re-scores a
  * night for a short window after the fact, so an exact re-post overwrites in
@@ -25,7 +41,6 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { getEvent } from "@/lib/logging/context";
 import {
-  markSyncFailureRecorded,
   recordSyncFailure,
   recordSyncSuccess,
   toFailureKind,
@@ -124,61 +139,115 @@ function toUpsert(
   }));
 }
 
+/** One Polar vitals collection: its ledger name (which doubles as the
+ * externalId resource prefix) and a self-contained fetch+map closure that
+ * resolves to the collection's mapped upserts. Each closure owns its own client
+ * call and mapper so a throw — a failing fetch OR a throwing mapper — is
+ * contained to this one collection under `Promise.allSettled`. */
+interface PolarCollectionSpec {
+  name: string;
+  collect: (token: string) => Promise<PolarMeasurementUpsert[]>;
+}
+
+/** A single collection that failed to fetch/map, carried so the caller can
+ * classify + record ONE partial failure without blanking the collections that
+ * did succeed. */
+export interface PolarCollectionFailure {
+  name: string;
+  err: unknown;
+}
+
 /**
  * Sync one user's Polar data. Returns the count of measurement rows written.
  * A user with no Polar connection is a clean no-op (returns 0, touches no
  * status row).
+ *
+ * The five vitals collections are settled independently: the healthy ones
+ * import, a failing one is recorded as a single partial failure (no success
+ * stamp), and nothing is rethrown for a fetch/map failure. The only throw that
+ * escapes now is a write-path/unexpected error out of `upsertPolarMeasurements`
+ * — the poll-cohort boundary records that unmarked escape once.
  */
 export async function syncUserPolar(userId: string): Promise<number> {
   const conn = await getPolarConnection(userId);
   if (!conn) return 0;
 
-  const readings: PolarMeasurementUpsert[] = [];
+  // Filled by the sleep collection's closure as it maps records; carried so the
+  // caller can run the record-scoped sweep before the upsert. A collection that
+  // throws pushes nothing (the fetch is its first await), so a failed sleep
+  // collection leaves the sweep list empty and no rows are tombstoned.
   const sleepSweeps: SleepSegmentSweep[] = [];
-  try {
-    const [recharges, sleeps, activities, cardioLoads, spo2Tests] =
-      await Promise.all([
-        fetchNightlyRecharges(conn.accessToken),
-        fetchSleeps(conn.accessToken),
-        fetchActivities(conn.accessToken),
-        fetchCardioLoads(conn.accessToken),
-        fetchSpo2(conn.accessToken),
-      ]);
-    for (const r of recharges) {
-      readings.push(...toUpsert(mapNightlyRecharge(r), "recharge"));
+
+  const collections: PolarCollectionSpec[] = [
+    {
+      name: "recharge",
+      collect: async (t) =>
+        (await fetchNightlyRecharges(t)).flatMap((r) =>
+          toUpsert(mapNightlyRecharge(r), "recharge"),
+        ),
+    },
+    {
+      name: "sleep",
+      collect: async (t) => {
+        const records = await fetchSleeps(t);
+        const rowsOut: PolarMeasurementUpsert[] = [];
+        for (const s of records) {
+          const rows = toUpsert(mapSleep(s), "sleep");
+          rowsOut.push(...rows);
+          // Night-scoped sweep entry: the reconstructed segments of this date
+          // all key under `sleep:<date>:seg:` (mapper-supplied). Any live row
+          // under that prefix this fetch did NOT re-produce is a re-score
+          // orphan or a legacy `:seg:<tag>:<i>` indexed row — tombstoned
+          // before the upsert. The prefix deliberately stays on `:seg:` — the
+          // IN_BED envelope keys on its measuredAt's UTC date, which can drift
+          // a calendar day from `s.date`, so a broader `sleep:<date>:` bound
+          // could cross nights.
+          sleepSweeps.push({
+            prefix: `sleep:${s.date}:seg:`,
+            keepIds: rows
+              .filter((r) => r.type === "SLEEP_DURATION")
+              .map((r) => r.externalId),
+          });
+        }
+        return rowsOut;
+      },
+    },
+    {
+      name: "activity",
+      collect: async (t) =>
+        (await fetchActivities(t)).flatMap((a) =>
+          toUpsert(mapActivity(a), "activity"),
+        ),
+    },
+    {
+      name: "cardioload",
+      collect: async (t) =>
+        (await fetchCardioLoads(t)).flatMap((c) =>
+          toUpsert(mapCardioLoad(c), "cardioload"),
+        ),
+    },
+    {
+      name: "spo2",
+      collect: async (t) =>
+        (await fetchSpo2(t)).flatMap((s) => toUpsert(mapSpo2(s), "spo2")),
+    },
+  ];
+
+  // Isolate each collection: a rejection is captured per-collection, never
+  // propagated, so one dead collection no longer blanks the healthy siblings.
+  const settled = await Promise.allSettled(
+    collections.map((spec) => spec.collect(conn.accessToken)),
+  );
+  const readings: PolarMeasurementUpsert[] = [];
+  const failures: PolarCollectionFailure[] = [];
+  settled.forEach((res, i) => {
+    const spec = collections[i]!;
+    if (res.status === "fulfilled") {
+      readings.push(...res.value);
+    } else {
+      failures.push({ name: spec.name, err: res.reason });
     }
-    for (const s of sleeps) {
-      const rows = toUpsert(mapSleep(s), "sleep");
-      readings.push(...rows);
-      // Night-scoped sweep entry: the reconstructed segments of this date all
-      // key under `sleep:<date>:seg:` (mapper-supplied). Any live row under
-      // that prefix this fetch did NOT re-produce is a re-score orphan or a
-      // legacy `:seg:<tag>:<i>` indexed row — tombstoned before the upsert.
-      // The prefix deliberately stays on `:seg:` — the IN_BED envelope keys
-      // on its measuredAt's UTC date, which can drift a calendar day from
-      // `s.date`, so a broader `sleep:<date>:` bound could cross nights.
-      sleepSweeps.push({
-        prefix: `sleep:${s.date}:seg:`,
-        keepIds: rows
-          .filter((r) => r.type === "SLEEP_DURATION")
-          .map((r) => r.externalId),
-      });
-    }
-    for (const a of activities) {
-      readings.push(...toUpsert(mapActivity(a), "activity"));
-    }
-    for (const c of cardioLoads) {
-      readings.push(...toUpsert(mapCardioLoad(c), "cardioload"));
-    }
-    for (const t of spo2Tests) {
-      readings.push(...toUpsert(mapSpo2(t), "spo2"));
-    }
-  } catch (err) {
-    // Recorded here, then marked so the poll-cohort boundary does not
-    // double-record it when the legs wrapper rethrows it there.
-    await recordPolarSyncFailure(userId, err);
-    throw markSyncFailureRecorded(err);
-  }
+  });
 
   // Clear whatever an earlier scoring left under the re-fetched nights before
   // the fresh set upserts (mirrors Google Health's replace-by-window order).
@@ -195,11 +264,42 @@ export async function syncUserPolar(userId: string): Promise<number> {
   });
 
   // S4 — trigger the debounced morning refresh on a last-night segment landing
-  // (mirrors the Withings / WHOOP / Apple seams).
+  // (mirrors the Withings / WHOOP / Apple seams). Fires whether or not a
+  // sibling collection failed, since the sleep readings were still imported.
   void maybeEnqueueMorningRefresh(userId, insertedSleepMeasuredAts).catch(
     () => {},
   );
 
+  if (failures.length > 0) {
+    // Partial failure: import what the healthy collections returned, but keep
+    // the cycle honest — record ONE failure and do NOT stamp success, so the
+    // freshness surface reflects that some collections are behind and the next
+    // tick refetches them rather than showing green. Mirrors Oura's
+    // partial-failure ledger block; the record is inline (not
+    // `recordPolarSyncFailure`) because it needs the `partial sync failure
+    // (<names>)` message prefix — the R6 recorder's boundary contract stays
+    // frozen. No rethrow: a rethrow would suppress the cohort's users_synced /
+    // reminder-satisfy accounting for a user whose data DID land.
+    const firstErr = failures[0]!.err;
+    const names = failures.map((f) => f.name).join(", ");
+    getEvent()?.addWarning(
+      `polar: ${failures.length} collection(s) failed for ${userId}: ${names}`,
+    );
+    await recordSyncFailure({
+      userId,
+      integration: "polar",
+      kind: classifyPolarFailure(firstErr),
+      message: `partial sync failure (${names}): ${
+        firstErr instanceof Error ? firstErr.message : String(firstErr)
+      }`,
+      errorCode:
+        firstErr instanceof PolarApiError && firstErr.httpStatus != null
+          ? String(firstErr.httpStatus)
+          : undefined,
+    });
+    return imported;
+  }
+
   await recordSyncSuccess(userId, "polar");
   return imported;
 }