fix(sync): surface poll-cohort sync failures that previously vanished

Introduce a recorded-marker primitive on the integration ledger so a provider
that records-then-rethrows a sync failure is not recorded a second time at the
poll-cohort boundary, while an unrecorded escape (an Oura or Polar write-path
throw, or any future path) lands on the ledger exactly once, correctly
classified and redacted by the source that owns it. Without the marker a
boundary-blind recorder would double-count the failure streak and could persist
a Nightscout token-bearing URL into the stored error.

Give the poll-cohort factory a required per-provider failure recorder and wire
one for Oura, Polar, Nightscout, and Strava.

Nightscout now counts only freshly inserted readings instead of insert plus
re-confirm (the inflated count fired the reminder trigger every tick), records a
transient failure and skips the success stamp when some rows fail to write, and
emits the measurement-arrival spine for its inserts. Correct the docstring that
promised a far-history backfill that does not exist: the hourly window imports
about 48 hours and nothing deeper.

Strava now enqueues its history backfill at connect time, mirroring the WHOOP
and Fitbit callbacks, so deep history imports within the hour instead of waiting
for the next worker restart.

src/app/api/strava/callback/__tests__/route.test.ts

@@ -0,0 +1,109 @@
+/**
+ * S-1 — the Strava OAuth callback enqueues the self-converging history backfill
+ * at connect time (mirroring the WHOOP / Fitbit callbacks) so deep history
+ * lands within the hour instead of waiting for the next worker reboot. A missing
+ * boss instance is a warning, never a failure — boot discovery is the net.
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest } from "next/server";
+
+const {
+  bossSend,
+  getGlobalBossMock,
+  exchangeCodeMock,
+  getCredsMock,
+  userFindUnique,
+  userUpdate,
+  markReconnectedMock,
+} = vi.hoisted(() => ({
+  bossSend: vi.fn(async () => "job-id"),
+  getGlobalBossMock: vi.fn(),
+  exchangeCodeMock: vi.fn(),
+  getCredsMock: vi.fn(),
+  userFindUnique: vi.fn(),
+  userUpdate: vi.fn(async () => ({})),
+  markReconnectedMock: vi.fn(async () => {}),
+}));
+
+vi.mock("@/lib/api-handler", () => ({
+  apiHandler: <T extends (...args: unknown[]) => unknown>(fn: T) => fn,
+}));
+vi.mock("@/lib/db", () => ({
+  prisma: { user: { findUnique: userFindUnique, update: userUpdate } },
+}));
+vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn(async () => null) }));
+vi.mock("@/lib/logging/context", () => ({
+  annotate: vi.fn(),
+  getEvent: () => ({
+    setAuth: vi.fn(),
+    setError: vi.fn(),
+    addWarning: vi.fn(),
+  }),
+}));
+vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn(async () => {}) }));
+vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc(${s})` }));
+vi.mock("@/lib/strava/client", () => ({ exchangeCode: exchangeCodeMock }));
+vi.mock("@/lib/strava/credentials", () => ({
+  getStravaClientCredentials: getCredsMock,
+}));
+vi.mock("@/lib/oauth/signed-state", () => ({
+  oauthStateCookieName: () => "strava_oauth_state",
+  stateMatchesCookie: () => true,
+  verifySignedState: () => ({ userId: "u1" }),
+}));
+vi.mock("@/lib/integrations/status", () => ({
+  markReconnected: markReconnectedMock,
+}));
+vi.mock("@/lib/jobs/strava-backfill", () => ({
+  STRAVA_BACKFILL_QUEUE: "strava-backfill",
+}));
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: getGlobalBossMock,
+}));
+
+import { GET } from "../route";
+
+process.env.NEXT_PUBLIC_APP_URL = "https://app.example";
+
+function callback(): Promise<{ headers: { get(k: string): string | null } }> {
+  const req = new NextRequest(
+    "http://localhost/api/strava/callback?code=abc&state=signed-state",
+    { headers: { cookie: "strava_oauth_state=signed-state" } },
+  );
+  return (GET as unknown as (r: NextRequest) => Promise<never>)(req);
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  getCredsMock.mockResolvedValue({ clientId: "c", clientSecret: "s" });
+  exchangeCodeMock.mockResolvedValue({
+    access_token: "at",
+    refresh_token: "rt",
+    athlete: { id: 42 },
+  });
+  userFindUnique.mockResolvedValue({ stravaAthleteId: null });
+  getGlobalBossMock.mockReturnValue({ send: bossSend });
+});
+
+describe("GET /api/strava/callback — S-1 backfill enqueue", () => {
+  it("enqueues one STRAVA_BACKFILL_QUEUE job for the connecting user", async () => {
+    const res = await callback();
+
+    expect(res.headers.get("location")).toContain("strava=connected");
+    expect(markReconnectedMock).toHaveBeenCalledWith("u1", "strava");
+    expect(bossSend).toHaveBeenCalledTimes(1);
+    expect(bossSend).toHaveBeenCalledWith(
+      "strava-backfill",
+      expect.objectContaining({ userId: "u1" }),
+    );
+  });
+
+  it("still connects when the boss instance is unavailable (warning, not failure)", async () => {
+    getGlobalBossMock.mockReturnValue(null);
+
+    const res = await callback();
+
+    expect(res.headers.get("location")).toContain("strava=connected");
+    expect(bossSend).not.toHaveBeenCalled();
+  });
+});

src/app/api/strava/callback/route.ts

@@ -12,6 +12,8 @@ import {
   verifySignedState,
 } from "@/lib/oauth/signed-state";
 import { markReconnected } from "@/lib/integrations/status";
+import { STRAVA_BACKFILL_QUEUE } from "@/lib/jobs/strava-backfill";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { NextRequest, NextResponse } from "next/server";
 
 /**
@@ -107,6 +109,24 @@ export const GET = apiHandler(async (request: NextRequest) => {
     await auditLog("strava.connect", { userId });
     await markReconnected(userId, "strava");
 
+    // Enqueue the self-converging history backfill at connect time so deep
+    // history lands within the hour instead of waiting for the next worker
+    // reboot. Best-effort: the boot-time discovery query
+    // (`stravaBackfillCompletedAt IS NULL`) is the safety net, so a missing
+    // boss instance here never strands the connection. Mirrors the WHOOP /
+    // Fitbit callback enqueue.
+    const boss = getGlobalBoss();
+    if (boss) {
+      await boss
+        .send(STRAVA_BACKFILL_QUEUE, {
+          userId,
+          enqueuedAt: new Date().toISOString(),
+        })
+        .catch((err) =>
+          getEvent()?.addWarning(`strava-backfill enqueue failed: ${err}`),
+        );
+    }
+
     const response = NextResponse.redirect(
       new URL(
         "/settings/integrations?strava=connected",

src/lib/integrations/__tests__/status.test.ts

@@ -47,6 +47,8 @@ import {
   isReauthRequired,
   getPersistentFailureThreshold,
   resumeIntegrationFromPark,
+  markSyncFailureRecorded,
+  isSyncFailureRecorded,
 } from "../status";
 import { prisma } from "@/lib/db";
 import { auditLog } from "@/lib/auth/audit";
@@ -856,3 +858,36 @@ describe("resumeIntegrationFromPark", () => {
     expect(auditLog).not.toHaveBeenCalled();
   });
 });
+
+describe("sync-failure-recorded marker (§5)", () => {
+  it("round-trips: a marked error reads back as recorded", () => {
+    const err = new Error("boom");
+    expect(isSyncFailureRecorded(err)).toBe(false);
+    const returned = markSyncFailureRecorded(err);
+    // Returns the SAME object so `throw markSyncFailureRecorded(err)` works.
+    expect(returned).toBe(err);
+    expect(isSyncFailureRecorded(err)).toBe(true);
+  });
+
+  it("the marker is non-enumerable (never serialised into logs / JSON)", () => {
+    const err = markSyncFailureRecorded(new Error("boom"));
+    expect(Object.keys(err)).not.toContain("healthlog.syncFailureRecorded");
+    expect(JSON.stringify({ ...err })).not.toContain("syncFailureRecorded");
+  });
+
+  it("is a no-op on non-objects and never reports them as recorded", () => {
+    expect(markSyncFailureRecorded("boom")).toBe("boom");
+    expect(markSyncFailureRecorded(null)).toBe(null);
+    expect(markSyncFailureRecorded(undefined)).toBe(undefined);
+    expect(isSyncFailureRecorded("boom")).toBe(false);
+    expect(isSyncFailureRecorded(null)).toBe(false);
+    expect(isSyncFailureRecorded(undefined)).toBe(false);
+  });
+
+  it("tolerates a frozen error without throwing (falls back to unmarked)", () => {
+    const frozen = Object.freeze(new Error("frozen"));
+    expect(() => markSyncFailureRecorded(frozen)).not.toThrow();
+    // Frozen can't carry the marker, so the boundary records it once — safe.
+    expect(isSyncFailureRecorded(frozen)).toBe(false);
+  });
+});

src/lib/integrations/status.ts

@@ -98,6 +98,51 @@ export function toFailureKind(
   return "transient";
 }
 
+/**
+ * Non-enumerable marker stamped on an error a provider sync ALREADY recorded on
+ * the integration ledger before rethrowing. The poll-cohort boundary
+ * (`makePollCohortHandler`) reads it so a record-then-rethrow failure is not
+ * recorded a SECOND time when it surfaces in the cohort catch — a double-record
+ * would inflate the per-kind failure buckets (firing the 3-strike admin alert a
+ * tick early) and churn `lastError`. A provider-blind recorder at the boundary
+ * would also persist an UNREDACTED message (e.g. Nightscout's token-bearing
+ * URL), so the marker keeps classification + redaction provider-owned: only an
+ * UNMARKED escape reaches the boundary recorder.
+ *
+ * No-op on non-objects; tolerant of a frozen error (the boundary then records
+ * it once, which is the correct fallback — never a double-record).
+ */
+const SYNC_FAILURE_RECORDED: unique symbol = Symbol(
+  "healthlog.syncFailureRecorded",
+);
+
+/** Stamp `err` as already-recorded and return it (for `throw markSync…(err)`). */
+export function markSyncFailureRecorded<T>(err: T): T {
+  if (err !== null && typeof err === "object") {
+    try {
+      Object.defineProperty(err, SYNC_FAILURE_RECORDED, {
+        value: true,
+        enumerable: false,
+        configurable: true,
+        writable: true,
+      });
+    } catch {
+      // A frozen / sealed error can't carry the marker; the boundary records
+      // it once — the safe fallback, never a double-record.
+    }
+  }
+  return err;
+}
+
+/** True when `err` was already recorded on the ledger by its source site. */
+export function isSyncFailureRecorded(err: unknown): boolean {
+  return (
+    err !== null &&
+    typeof err === "object" &&
+    (err as Record<symbol, unknown>)[SYNC_FAILURE_RECORDED] === true
+  );
+}
+
 /**
  * Recognised IntegrationStatus states.
  *

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

@@ -5,7 +5,12 @@ const { syncUserPolar, syncUserPolarWorkouts } = vi.hoisted(() => ({
   syncUserPolarWorkouts: vi.fn(),
 }));
 
-vi.mock("@/lib/polar/sync", () => ({ syncUserPolar }));
+vi.mock("@/lib/polar/sync", () => ({
+  syncUserPolar,
+  // The cohort config wires `recordPolarSyncFailure` as its recorder; the legs
+  // handler imports the module, so the mock must expose it too.
+  recordPolarSyncFailure: vi.fn(async () => {}),
+}));
 vi.mock("@/lib/polar/sync-workouts", () => ({ syncUserPolarWorkouts }));
 
 import { syncUserPolarLegs } from "../polar-sync";

src/lib/jobs/reminder/__tests__/poll-cohort.test.ts

@@ -0,0 +1,115 @@
+/**
+ * §5 — the poll-cohort factory boundary. A provider that records-then-rethrows
+ * marks its error so the boundary does NOT record it a second time (which would
+ * inflate the per-kind buckets and, for Nightscout, risk re-persisting a
+ * token-bearing URL). An UNMARKED escape — today's Oura / Polar write-path throw
+ * — is recorded exactly once through the per-provider `recordFailure` callback.
+ * A recorder rejection never breaks the cohort loop.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { Job } from "pg-boss";
+
+const warnings: string[] = [];
+
+vi.mock("@/lib/jobs/worker-status", () => ({ recordError: vi.fn() }));
+vi.mock("@/lib/jobs/reminder-satisfy", () => ({
+  enqueueReminderSatisfy: vi.fn(async () => undefined),
+}));
+vi.mock("@/lib/logging/fire-and-forget", () => ({ fireAndForget: vi.fn() }));
+vi.mock("@/lib/logging/background", () => ({
+  withBackgroundEvent: async (
+    _name: string,
+    fn: (evt: {
+      setBackground: () => void;
+      setError: () => void;
+      addWarning: (w: string) => void;
+    }) => Promise<void>,
+  ) =>
+    fn({
+      setBackground: vi.fn(),
+      setError: vi.fn(),
+      addWarning: (w: string) => warnings.push(w),
+    }),
+}));
+vi.mock("../shared", () => ({
+  getWorkerPrisma: () => ({}),
+}));
+
+import { makePollCohortHandler } from "../poll-cohort";
+import { markSyncFailureRecorded } from "@/lib/integrations/status";
+
+function job(userId?: string): Job<{ userId?: string }> {
+  return { data: userId ? { userId } : {} } as Job<{ userId?: string }>;
+}
+
+beforeEach(() => {
+  warnings.length = 0;
+  vi.clearAllMocks();
+});
+
+describe("makePollCohortHandler — marker-aware recordFailure boundary", () => {
+  it("records an UNMARKED escape once, then warns, and continues the cohort", async () => {
+    const recordFailure = vi.fn(async () => {});
+    const syncUser = vi.fn(async (userId: string) => {
+      if (userId === "boom") throw new Error("unmarked write failure");
+      return 1;
+    });
+    const handler = makePollCohortHandler({
+      taskName: "job.test_sync",
+      findCohort: async () => ["a", "boom", "b"],
+      syncUser,
+      recordFailure,
+    });
+
+    await handler([job()]);
+
+    // The failing user is recorded once and warned; the other two still sync.
+    expect(syncUser).toHaveBeenCalledTimes(3);
+    expect(recordFailure).toHaveBeenCalledTimes(1);
+    expect(recordFailure).toHaveBeenCalledWith(
+      "boom",
+      expect.objectContaining({ message: "unmarked write failure" }),
+    );
+    expect(warnings.some((w) => w.includes("boom"))).toBe(true);
+  });
+
+  it("does NOT record a MARKED error (the source already recorded it)", async () => {
+    const recordFailure = vi.fn(async () => {});
+    const syncUser = vi.fn(async () => {
+      throw markSyncFailureRecorded(new Error("already recorded at source"));
+    });
+    const handler = makePollCohortHandler({
+      taskName: "job.test_sync",
+      findCohort: async () => ["a"],
+      syncUser,
+      recordFailure,
+    });
+
+    await handler([job()]);
+
+    expect(recordFailure).not.toHaveBeenCalled();
+    // Still warned so the cohort event carries the failure.
+    expect(warnings.some((w) => w.includes("already recorded"))).toBe(true);
+  });
+
+  it("a recorder rejection never breaks the cohort loop", async () => {
+    const recordFailure = vi.fn(async () => {
+      throw new Error("recorder down");
+    });
+    const syncUser = vi.fn(async (userId: string) => {
+      if (userId === "boom") throw new Error("unmarked");
+      return 1;
+    });
+    const handler = makePollCohortHandler({
+      taskName: "job.test_sync",
+      findCohort: async () => ["boom", "next"],
+      syncUser,
+      recordFailure,
+    });
+
+    await expect(handler([job()])).resolves.toBeUndefined();
+    // The next user still synced despite the recorder throwing.
+    expect(syncUser).toHaveBeenCalledTimes(2);
+    expect(syncUser).toHaveBeenLastCalledWith("next");
+  });
+});

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

@@ -9,7 +9,10 @@
  * `./poll-cohort`; reminder-worker.ts owns the queue name, cron schedule, and
  * boss.work registration.
  */
-import { syncUserNightscout } from "@/lib/nightscout/sync";
+import {
+  recordNightscoutSyncFailure,
+  syncUserNightscout,
+} from "@/lib/nightscout/sync";
 import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort";
 
 export type NightscoutSyncPayload = PollCohortPayload;
@@ -25,4 +28,8 @@ export const handleNightscoutSync = makePollCohortHandler({
     return users.map((u) => u.id);
   },
   syncUser: syncUserNightscout,
+  // The fetch path already records + marks its own failure (token-redacted), so
+  // this fires only for a future UNMARKED escape — and redacts the token before
+  // the message reaches the ledger.
+  recordFailure: recordNightscoutSyncFailure,
 });

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

@@ -7,7 +7,7 @@
  * lives in `./poll-cohort`; the queue name, cron schedule, and boss.work
  * registration live in reminder-worker.ts.
  */
-import { syncUserOura } from "@/lib/oura/sync";
+import { recordOuraSyncFailure, syncUserOura } from "@/lib/oura/sync";
 import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort";
 
 export type OuraSyncPayload = PollCohortPayload;
@@ -22,4 +22,7 @@ export const handleOuraSync = makePollCohortHandler({
     return users.map((u) => u.id);
   },
   syncUser: syncUserOura,
+  // Catches an UNMARKED escape — today the `upsertOuraMeasurements` write-path
+  // throw (the fetch/refresh path already records + marks its own failures).
+  recordFailure: recordOuraSyncFailure,
 });

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

@@ -13,7 +13,7 @@
  * reminder-worker.ts — no new queue: the workout leg rides the existing hourly
  * `polar-sync` tick (one extra request per connected user per hour).
  */
-import { syncUserPolar } from "@/lib/polar/sync";
+import { recordPolarSyncFailure, syncUserPolar } from "@/lib/polar/sync";
 import { syncUserPolarWorkouts } from "@/lib/polar/sync-workouts";
 import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort";
 
@@ -63,4 +63,8 @@ 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).
+  recordFailure: recordPolarSyncFailure,
 });

src/lib/jobs/reminder/poll-cohort.ts

@@ -23,6 +23,7 @@ import { fireAndForget } from "@/lib/logging/fire-and-forget";
 import { recordError } from "@/lib/jobs/worker-status";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
+import { isSyncFailureRecorded } from "@/lib/integrations/status";
 import { getWorkerPrisma } from "./shared";
 
 /** Every poll-cohort handler accepts an optional single-user payload. */
@@ -40,6 +41,16 @@ export interface PollCohortConfig {
   findCohort: (prisma: ReturnType<typeof getWorkerPrisma>) => Promise<string[]>;
   /** Sync one user; returns the count of measurements imported. */
   syncUser: (userId: string) => Promise<number>;
+  /**
+   * Record a per-user sync failure that escaped `syncUser` UNRECORDED. Called
+   * only for an error the source site did not already mark via
+   * `markSyncFailureRecorded` — so a provider that records-then-rethrows never
+   * double-records, and the recorder keeps classification + secret redaction
+   * provider-owned (a boundary-blind recorder would leak Nightscout's
+   * token-bearing URL into `lastError`). A rejection here never breaks the
+   * cohort loop.
+   */
+  recordFailure: (userId: string, err: unknown) => Promise<void>;
 }
 
 /**
@@ -52,6 +63,7 @@ export function makePollCohortHandler({
   taskName,
   findCohort,
   syncUser,
+  recordFailure,
 }: PollCohortConfig) {
   return async function handlePollCohort(jobs: Job<PollCohortPayload>[]) {
     await withBackgroundEvent(taskName, async (evt) => {
@@ -82,6 +94,18 @@ export function makePollCohortHandler({
               });
             }
           } catch (err) {
+            // A record-then-rethrow provider already stamped the ledger and
+            // marked the error; recording again would double-count the streak
+            // and (for Nightscout) risk re-persisting an unredacted message.
+            // Only an UNMARKED escape — today's Oura / Polar write-path throw,
+            // or any future provider path — is recorded here.
+            if (!isSyncFailureRecorded(err)) {
+              await recordFailure(userId, err).catch((recErr) => {
+                evt.addWarning(
+                  `${taskName} recordFailure failed for user ${userId}: ${recErr}`,
+                );
+              });
+            }
             evt.addWarning(`${taskName} failed for user ${userId}: ${err}`);
           }
         }

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

@@ -11,7 +11,7 @@
  * The queue name, cron schedule, and boss.work registration live in
  * `register-integration-sync.ts`.
  */
-import { syncUserStrava } from "@/lib/strava/sync";
+import { recordStravaFailure, syncUserStrava } from "@/lib/strava/sync";
 import { makePollCohortHandler, type PollCohortPayload } from "./poll-cohort";
 
 export type StravaSyncPayload = PollCohortPayload;
@@ -26,4 +26,8 @@ export const handleStravaSync = makePollCohortHandler({
     return users.map((u) => u.id);
   },
   syncUser: (userId) => syncUserStrava(userId),
+  // Strava records + marks every failure at its source today, so this is inert
+  // now — it's the net for a future unmarked escape (matching the cohort's
+  // contract that every provider supplies a recorder).
+  recordFailure: recordStravaFailure,
 });

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

@@ -4,6 +4,8 @@ const {
   fetchSgvEntriesMock,
   getCredsMock,
   upsertMock,
+  findManyMock,
+  emitArrivalsMock,
   recordSuccessMock,
   recordFailureMock,
   recomputeMock,
@@ -12,6 +14,8 @@ const {
   fetchSgvEntriesMock: vi.fn(),
   getCredsMock: vi.fn(),
   upsertMock: vi.fn(),
+  findManyMock: vi.fn(),
+  emitArrivalsMock: vi.fn(),
   recordSuccessMock: vi.fn(),
   recordFailureMock: vi.fn(),
   recomputeMock: vi.fn(),
@@ -23,7 +27,11 @@ vi.mock("../credentials", () => ({
 }));
 
 vi.mock("@/lib/db", () => ({
-  prisma: { measurement: { upsert: upsertMock } },
+  prisma: { measurement: { upsert: upsertMock, findMany: findManyMock } },
+}));
+
+vi.mock("@/lib/arrivals/measurement-emit", () => ({
+  emitInsertedMeasurementArrivals: emitArrivalsMock,
 }));
 
 vi.mock("@/lib/integrations/status", async (importOriginal) => ({
@@ -63,12 +71,17 @@ beforeEach(() => {
   fetchSgvEntriesMock.mockReset();
   getCredsMock.mockReset();
   upsertMock.mockReset();
+  findManyMock.mockReset();
+  emitArrivalsMock.mockReset();
   recordSuccessMock.mockReset();
   recordFailureMock.mockReset();
   recomputeMock.mockReset();
   invalidateMock.mockReset();
   getCredsMock.mockResolvedValue(CREDS);
-  upsertMock.mockResolvedValue({});
+  upsertMock.mockResolvedValue({ id: "row" });
+  // Default: no rows pre-exist → every entry is a fresh insert.
+  findManyMock.mockResolvedValue([]);
+  emitArrivalsMock.mockResolvedValue(undefined);
   invalidateMock.mockResolvedValue(undefined);
 });
 
@@ -128,6 +141,51 @@ describe("syncUserNightscout", () => {
     expect(recordSuccessMock).toHaveBeenCalledWith("u1", "nightscout");
   });
 
+  it("N-1 — counts only fresh inserts and emits ONLY the newly-inserted rows", async () => {
+    fetchSgvEntriesMock.mockResolvedValue([ENTRY_A, ENTRY_B]);
+    // ENTRY_A already exists (a re-confirm); ENTRY_B is genuinely new.
+    findManyMock.mockResolvedValue([{ externalId: "ns:abc" }]);
+
+    const n = await syncUserNightscout("u1");
+
+    // The inflated insert+reconfirm count used to fire the reminder-satisfy
+    // trigger every tick; the honest count is the single fresh insert.
+    expect(n).toBe(1);
+    // Both rows are still written (idempotent), but only the fresh one emits.
+    expect(upsertMock).toHaveBeenCalledTimes(2);
+    expect(emitArrivalsMock).toHaveBeenCalledTimes(1);
+    const [uid, rows, source] = emitArrivalsMock.mock.calls[0]!;
+    expect(uid).toBe("u1");
+    expect(source).toBe("nightscout");
+    expect(rows).toHaveLength(1);
+    expect(rows[0]).toMatchObject({
+      type: "BLOOD_GLUCOSE",
+      measuredAt: new Date(1718000300000),
+    });
+    expect(recordSuccessMock).toHaveBeenCalledWith("u1", "nightscout");
+  });
+
+  it("N-2 — a partial row failure records a transient failure and SKIPS success", async () => {
+    fetchSgvEntriesMock.mockResolvedValue([ENTRY_A, ENTRY_B]);
+    findManyMock.mockResolvedValue([]);
+    upsertMock
+      .mockResolvedValueOnce({ id: "r1" })
+      .mockRejectedValueOnce(new Error("bad row"));
+
+    const n = await syncUserNightscout("u1");
+
+    // The one good row still landed; the tick is honest about the failed row.
+    expect(n).toBe(1);
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+    expect(recordFailureMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        integration: "nightscout",
+        kind: "transient",
+        message: "1 of 2 SGV rows failed to write",
+      }),
+    );
+  });
+
   it("records a failure and rethrows when the instance is unreachable", async () => {
     fetchSgvEntriesMock.mockRejectedValue(
       new NightscoutApiError("Nightscout responded 401", 401),

src/lib/nightscout/sync.ts

@@ -23,6 +23,7 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { getEvent } from "@/lib/logging/context";
 import {
+  markSyncFailureRecorded,
   recordSyncFailure,
   recordSyncSuccess,
   toFailureKind,
@@ -33,6 +34,10 @@ import {
   collapseToTypeDayKeys,
   recomputeBucketsForMeasurement,
 } from "@/lib/rollups/measurement-rollups";
+import {
+  emitInsertedMeasurementArrivals,
+  type InsertedMeasurementArrivalRow,
+} from "@/lib/arrivals/measurement-emit";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
   fetchSgvEntries,
@@ -46,7 +51,12 @@ import { getUserNightscoutCredentials } from "./credentials";
  * How many recent SGV entries to pull per sync tick. A CGM emits one reading
  * every ~5 min (≈288/day); 576 covers ~48 h of catch-up so a missed hourly
  * tick (or a brief outage) still backfills on the next run without paging the
- * full instance. The far-history backfill walks larger windows separately.
+ * full instance.
+ *
+ * This ~48 h window is the ONLY history Nightscout imports: there is no
+ * far-history backfill queue, so connecting Nightscout brings in about two
+ * days of readings and every tick thereafter walks the same recent set. A
+ * deeper walk would need its own windowed-pagination queue; it is not built.
  */
 export const NIGHTSCOUT_SYNC_COUNT = 576;
 
@@ -69,6 +79,32 @@ export function redactNightscoutSecret(message: string): string {
   return message.replace(/\b(token|api-secret)=[^\s&"']+/gi, "$1=REDACTED");
 }
 
+/**
+ * Record a Nightscout sync failure on the shared `nightscout` ledger with the
+ * correct classification + HTTP code + the token STRIPPED from the message.
+ * Extracted so the poll-cohort boundary can record a future escape with the
+ * same provider-owned classification AND redaction the inline fetch catch
+ * applies — a boundary-blind recorder would persist the token-bearing URL a
+ * `SafeFetchError` embeds.
+ */
+export async function recordNightscoutSyncFailure(
+  userId: string,
+  err: unknown,
+): Promise<void> {
+  await recordSyncFailure({
+    userId,
+    integration: "nightscout",
+    kind: classifyNightscoutFailure(err),
+    message: redactNightscoutSecret(
+      err instanceof Error ? err.message : String(err),
+    ),
+    errorCode:
+      err instanceof NightscoutApiError && err.status != null
+        ? String(err.status)
+        : undefined,
+  });
+}
+
 /** Map a Nightscout HTTP status / network error onto the shared ledger kind. */
 export function classifyNightscoutFailure(err: unknown): FailureKind {
   let classification: IntegrationClassification = "transient";
@@ -86,9 +122,12 @@ export function classifyNightscoutFailure(err: unknown): FailureKind {
 }
 
 /**
- * Sync one user's recent SGV entries. Returns the count of rows written
- * (newly inserted or re-confirmed). A user with no configured instance is a
- * clean no-op (returns 0, touches no status row).
+ * Sync one user's recent SGV entries. Returns the count of rows FRESHLY
+ * INSERTED this tick (re-confirmed rows the upsert merely touched are not
+ * counted — the old insert+reconfirm total inflated the cohort's
+ * `measurements_imported` and fired the reminder-satisfy trigger every tick
+ * regardless of whether new glucose actually arrived). A user with no
+ * configured instance is a clean no-op (returns 0, touches no status row).
  */
 export async function syncUserNightscout(
   userId: string,
@@ -106,25 +145,34 @@ export async function syncUserNightscout(
       allowPrivateHost: creds.allowPrivateHost,
     });
   } catch (err) {
+    // Recorded here (token-redacted), then marked so the poll-cohort boundary
+    // does not double-record it — and never re-persists the token-bearing URL
+    // a boundary-blind recorder would.
+    await recordNightscoutSyncFailure(userId, err);
+    throw markSyncFailureRecorded(err);
+  }
+
+  const { inserted, failedRows } = await upsertNightscoutEntries(
+    userId,
+    entries,
+  );
+
+  if (failedRows > 0) {
+    // Partial write failure: keep the tick honest. `transient` because the
+    // ~48 h re-fetch window self-heals a transient DB hiccup, while a
+    // persistent malformed-row failure now accrues visible strikes instead of
+    // being invisibly reset by a success stamp. Do NOT stamp success.
     await recordSyncFailure({
       userId,
       integration: "nightscout",
-      kind: classifyNightscoutFailure(err),
-      message: redactNightscoutSecret(
-        err instanceof Error ? err.message : String(err),
-      ),
-      errorCode:
-        err instanceof NightscoutApiError && err.status != null
-          ? String(err.status)
-          : undefined,
+      kind: "transient",
+      message: `${failedRows} of ${entries.length} SGV rows failed to write`,
     });
-    throw err;
+    return inserted;
   }
 
-  const imported = await upsertNightscoutEntries(userId, entries);
-
   await recordSyncSuccess(userId, "nightscout");
-  return imported;
+  return inserted;
 }
 
 /**
@@ -132,38 +180,71 @@ export async function syncUserNightscout(
  * rollup tier + invalidate status-insight caches once at the end (mirrors the
  * Withings / WHOOP sync tail). First-write-wins: the `update` branch is empty
  * so a re-sync of the same reading is a no-op rather than a rewrite. Best-effort
- * on the per-row write + the rollup fold — one bad row never aborts the pass.
+ * on the per-row write + the rollup fold — one bad row never aborts the pass,
+ * and the count of failed rows is returned so the caller can surface a partial
+ * write failure instead of stamping success over it.
+ *
+ * Returns `inserted` (rows whose key was absent before this pass — the honest
+ * "new readings" count that also drives the arrival emit) and `failedRows`.
  */
 export async function upsertNightscoutEntries(
   userId: string,
   entries: ParsedSgvEntry[],
-): Promise<number> {
-  if (entries.length === 0) return 0;
+): Promise<{ inserted: number; failedRows: number }> {
+  if (entries.length === 0) return { inserted: 0, failedRows: 0 };
 
   const type = "BLOOD_GLUCOSE" as MeasurementType;
-  let imported = 0;
+  const mapped = entries.map((entry) => mapSgvEntryToMeasurement(entry));
+
+  // Probe which keys already exist so a re-confirmed reading (the upsert's
+  // no-op update branch) is not miscounted as a fresh import and the arrival
+  // spine emits ONLY the genuinely-new rows. Cohort passes are sequential per
+  // user, so no concurrent insert races this probe. A probe failure degrades
+  // to "treat every row as fresh" — the rollup fold + count self-correct on
+  // the next tick.
+  const existingKeys = new Set<string>();
+  try {
+    const existing = await prisma.measurement.findMany({
+      where: {
+        userId,
+        type,
+        source: "NIGHTSCOUT",
+        externalId: { in: mapped.map((m) => m.externalId) },
+      },
+      select: { externalId: true },
+    });
+    for (const row of existing) {
+      if (row.externalId) existingKeys.add(row.externalId);
+    }
+  } catch (err) {
+    getEvent()?.addWarning(
+      `nightscout: insert-probe failed for ${userId}: ${err}`,
+    );
+  }
+
+  let failedRows = 0;
+  const insertedRows: InsertedMeasurementArrivalRow[] = [];
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
 
-  for (const entry of entries) {
-    const mapped = mapSgvEntryToMeasurement(entry);
+  for (const m of mapped) {
     try {
-      await prisma.measurement.upsert({
+      const row = await prisma.measurement.upsert({
         where: {
           userId_type_source_externalId: {
             userId,
             type,
             source: "NIGHTSCOUT",
-            externalId: mapped.externalId,
+            externalId: m.externalId,
           },
         },
         create: {
           userId,
           type,
           source: "NIGHTSCOUT",
-          value: mapped.value,
-          unit: mapped.unit,
-          measuredAt: mapped.measuredAt,
-          externalId: mapped.externalId,
+          value: m.value,
+          unit: m.unit,
+          measuredAt: m.measuredAt,
+          externalId: m.externalId,
         },
         // First-write-wins: a CGM sample is immutable, so a re-sync of the
         // same reading must not rewrite the stored value or measuredAt.
@@ -172,16 +253,32 @@ export async function upsertNightscoutEntries(
         // truth for its own rows, so a re-synced reading brings the row
         // back (mirrors Google / Fitbit).
         update: { deletedAt: null },
+        select: { id: true },
       });
-      touched.push({ type, measuredAt: mapped.measuredAt });
-      imported++;
+      touched.push({ type, measuredAt: m.measuredAt });
+      if (!existingKeys.has(m.externalId)) {
+        insertedRows.push({ type, measuredAt: m.measuredAt, id: row.id });
+      }
     } catch (err) {
       getEvent()?.addWarning(
         `nightscout: failed to upsert measurement: ${err}`,
       );
+      failedRows++;
     }
   }
 
+  // Spine parity with every other measurement writer: emit at most one arrival
+  // per supported kind for the rows freshly inserted. BLOOD_GLUCOSE is not yet
+  // an arrival kind (`ARRIVAL_MEASUREMENT_KIND`), so this is behaviourally
+  // inert for glucose today — it removes Nightscout as the spine's odd one out
+  // and lights up automatically once a `glucose` arrival kind + its consuming
+  // surface land in a later release.
+  void emitInsertedMeasurementArrivals(
+    userId,
+    insertedRows,
+    "nightscout",
+  ).catch(() => {});
+
   try {
     const keys = collapseToTypeDayKeys(touched);
     for (const k of keys) {
@@ -201,5 +298,5 @@ export async function upsertNightscoutEntries(
     );
   }
 
-  return imported;
+  return { inserted: insertedRows.length, failedRows };
 }

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

@@ -114,6 +114,7 @@ vi.mock("../client", async (importOriginal) => {
 
 import { syncUserOura, upsertOuraMeasurements } from "../sync";
 import { OuraApiError } from "../response-classifier";
+import { isSyncFailureRecorded } from "@/lib/integrations/status";
 
 const CONN = {
   accessToken: "acc",
@@ -284,13 +285,39 @@ describe("syncUserOura", () => {
         upstreamError: "invalid_grant",
       }),
     );
-    await expect(syncUserOura("u1")).rejects.toBeInstanceOf(OuraApiError);
+    const err = await syncUserOura("u1").catch((e) => e);
+    expect(err).toBeInstanceOf(OuraApiError);
+    // Recorded at the source AND marked, so the poll-cohort boundary does not
+    // record it a second time.
+    expect(isSyncFailureRecorded(err)).toBe(true);
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
     expect(recordFailureMock).toHaveBeenCalledWith(
       expect.objectContaining({ integration: "oura", kind: "reauth_required" }),
     );
     expect(recordSuccessMock).not.toHaveBeenCalled();
   });
 
+  it("leaves a write-path failure UNMARKED so the cohort boundary records it once", async () => {
+    // The `upsertOuraMeasurements` transaction throw escapes `syncUserOura`
+    // uncaught: no source-site record, no mark — the poll-cohort boundary is
+    // what records it (exactly once).
+    getConnMock.mockResolvedValue(CONN);
+    fetchReadinessMock.mockResolvedValue([
+      { id: "r1", day: "2026-06-10", score: 71 },
+    ]);
+    reconcileMock.mockResolvedValue({
+      status: "failed",
+      reason: "db_error",
+      error: new Error("write boom"),
+    });
+
+    const err = await syncUserOura("u1").catch((e) => e);
+    expect(err).toBeInstanceOf(Error);
+    expect(isSyncFailureRecorded(err)).toBe(false);
+    expect(recordFailureMock).not.toHaveBeenCalled();
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+  });
+
   it("keeps a nap and the main sleep distinct via record-scoped externalIds (B2)", async () => {
     getConnMock.mockResolvedValue(CONN);
     // Two sleep records on the same day: a main sleep + a nap.

src/lib/oura/sync.ts

@@ -37,6 +37,7 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { getEvent } from "@/lib/logging/context";
 import {
+  markSyncFailureRecorded,
   recordSyncFailure,
   recordSyncSuccess,
   toFailureKind,
@@ -97,6 +98,28 @@ export function classifyOuraFailure(err: unknown): FailureKind {
   return toFailureKind(classifyOuraError(err));
 }
 
+/**
+ * Record a whole-connection Oura sync failure on the shared `oura` ledger with
+ * the correct classification + HTTP code. Extracted so the poll-cohort boundary
+ * can record a future escape with the SAME provider-owned classification the
+ * inline catch already applies.
+ */
+export async function recordOuraSyncFailure(
+  userId: string,
+  err: unknown,
+): Promise<void> {
+  await recordSyncFailure({
+    userId,
+    integration: "oura",
+    kind: classifyOuraFailure(err),
+    message: err instanceof Error ? err.message : String(err),
+    errorCode:
+      err instanceof OuraApiError && err.httpStatus != null
+        ? String(err.httpStatus)
+        : undefined,
+  });
+}
+
 function ymd(d: Date): string {
   return d.toISOString().slice(0, 10);
 }
@@ -355,17 +378,10 @@ export async function syncUserOura(
   } catch (err) {
     // Only a whole-connection failure (a failed token refresh → reauth) reaches
     // here now; per-collection failures are captured on `result.failures`.
-    await recordSyncFailure({
-      userId,
-      integration: "oura",
-      kind: classifyOuraFailure(err),
-      message: err instanceof Error ? err.message : String(err),
-      errorCode:
-        err instanceof OuraApiError && err.httpStatus != null
-          ? String(err.httpStatus)
-          : undefined,
-    });
-    throw err;
+    // Recorded here, then marked so the poll-cohort boundary does not
+    // double-record it when it surfaces there.
+    await recordOuraSyncFailure(userId, err);
+    throw markSyncFailureRecorded(err);
   }
 
   // Clear whatever an earlier scoring left under the re-fetched sleep records

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

@@ -92,6 +92,7 @@ vi.mock("../client", async (importOriginal) => {
 });
 
 import { syncUserPolar, upsertPolarMeasurements } from "../sync";
+import { isSyncFailureRecorded } from "@/lib/integrations/status";
 import { PolarApiError } from "../response-classifier";
 
 const CONN = { accessToken: "tok", polarUserId: "42" };
@@ -340,7 +341,12 @@ describe("syncUserPolar", () => {
         reason: "http_401",
       }),
     );
-    await expect(syncUserPolar("u1")).rejects.toBeInstanceOf(PolarApiError);
+    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);
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
     expect(recordFailureMock).toHaveBeenCalledWith(
       expect.objectContaining({
         userId: "u1",
@@ -351,6 +357,27 @@ describe("syncUserPolar", () => {
     );
     expect(recordSuccessMock).not.toHaveBeenCalled();
   });
+
+  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
+    // what records it (exactly once).
+    getConnMock.mockResolvedValue(CONN);
+    fetchRechargesMock.mockResolvedValue([
+      { date: "2026-06-10", nightly_recharge_status: 6 },
+    ]);
+    reconcileMock.mockReset().mockResolvedValue({
+      status: "failed",
+      reason: "db_error",
+      error: new Error("write boom"),
+    });
+
+    const err = await syncUserPolar("u1").catch((e) => e);
+    expect(err).toBeInstanceOf(Error);
+    expect(isSyncFailureRecorded(err)).toBe(false);
+    expect(recordFailureMock).not.toHaveBeenCalled();
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+  });
 });
 
 describe("upsertPolarMeasurements — exact insertion results", () => {

src/lib/polar/sync-workouts.ts

@@ -31,7 +31,11 @@
 import { prisma } from "@/lib/db";
 import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
-import { recordSyncFailure, toFailureKind } from "@/lib/integrations/status";
+import {
+  markSyncFailureRecorded,
+  recordSyncFailure,
+  toFailureKind,
+} from "@/lib/integrations/status";
 import { fetchExercises, mapExercise, type PolarWorkoutRow } from "./client";
 import { getPolarConnection } from "./credentials";
 import { PolarApiError, classifyPolarError } from "./response-classifier";
@@ -82,7 +86,7 @@ export async function syncUserPolarWorkouts(userId: string): Promise<number> {
     }
   } catch (err) {
     await recordPolarWorkoutFailure(userId, err);
-    throw err;
+    throw markSyncFailureRecorded(err);
   }
 
   let imported: number;
@@ -90,7 +94,7 @@ export async function syncUserPolarWorkouts(userId: string): Promise<number> {
     imported = await upsertPolarWorkouts(userId, rows);
   } catch (err) {
     await recordPolarWorkoutFailure(userId, err);
-    throw err;
+    throw markSyncFailureRecorded(err);
   }
 
   return imported;

src/lib/polar/sync.ts

@@ -25,6 +25,7 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { getEvent } from "@/lib/logging/context";
 import {
+  markSyncFailureRecorded,
   recordSyncFailure,
   recordSyncSuccess,
   toFailureKind,
@@ -69,6 +70,28 @@ export function classifyPolarFailure(err: unknown): FailureKind {
   return toFailureKind(classifyPolarError(err));
 }
 
+/**
+ * Record a Polar vitals-leg sync failure on the shared `polar` ledger with the
+ * correct classification + HTTP code. Extracted so the poll-cohort boundary can
+ * record a future escape with the SAME provider-owned classification the inline
+ * catch already applies.
+ */
+export async function recordPolarSyncFailure(
+  userId: string,
+  err: unknown,
+): Promise<void> {
+  await recordSyncFailure({
+    userId,
+    integration: "polar",
+    kind: classifyPolarFailure(err),
+    message: err instanceof Error ? err.message : String(err),
+    errorCode:
+      err instanceof PolarApiError && err.httpStatus != null
+        ? String(err.httpStatus)
+        : undefined,
+  });
+}
+
 /** One mapped reading with its `externalId` resolved. */
 export interface PolarMeasurementUpsert {
   type: string;
@@ -151,17 +174,10 @@ export async function syncUserPolar(userId: string): Promise<number> {
       readings.push(...toUpsert(mapSpo2(t), "spo2"));
     }
   } catch (err) {
-    await recordSyncFailure({
-      userId,
-      integration: "polar",
-      kind: classifyPolarFailure(err),
-      message: err instanceof Error ? err.message : String(err),
-      errorCode:
-        err instanceof PolarApiError && err.httpStatus != null
-          ? String(err.httpStatus)
-          : undefined,
-    });
-    throw 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

src/lib/strava/__tests__/sync-cursor-durability.test.ts

@@ -39,6 +39,7 @@ vi.mock("@/lib/integrations/status", () => ({
   recordSyncFailure: mocks.recordSyncFailure,
   recordSyncSuccess: mocks.recordSyncSuccess,
   toFailureKind: vi.fn(() => "transient"),
+  markSyncFailureRecorded: <T>(err: T) => err,
 }));
 vi.mock("../credentials", () => ({
   getStravaConnection: mocks.getStravaConnection,

src/lib/strava/sync.ts

@@ -27,6 +27,7 @@ import { prisma } from "@/lib/db";
 import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit";
 import { annotate, getEvent } from "@/lib/logging/context";
 import {
+  markSyncFailureRecorded,
   recordSyncFailure,
   recordSyncSuccess,
   toFailureKind,
@@ -68,7 +69,7 @@ export function classifyStravaFailure(err: unknown): FailureKind {
   return toFailureKind(classifyStravaError(err));
 }
 
-async function recordStravaFailure(
+export async function recordStravaFailure(
   userId: string,
   err: unknown,
 ): Promise<void> {
@@ -215,7 +216,7 @@ export async function syncUserStrava(
     }
   } catch (err) {
     await recordStravaFailure(userId, err);
-    throw err;
+    throw markSyncFailureRecorded(err);
   }
 
   let imported: number;
@@ -239,7 +240,7 @@ export async function syncUserStrava(
     }
   } catch (err) {
     await recordStravaFailure(userId, err);
-    throw err;
+    throw markSyncFailureRecorded(err);
   }
 
   await recordSyncSuccess(userId, "strava");