fix(fitbit): hold the sync verdict when a token is dead or a write fails

A revoked or expired Fitbit refresh token used to fail the cycle's status
ledger and then stamp the same cycle a success: the connection flipped back
to connected, the watermark advanced, and after a reconnect the 24h overlap
could no longer reach the missed days. getValidToken's three null-return
paths (credentials missing, refresh failure, vanished connection) now register
on the per-cycle hard-fail ledger, so the cycle's verdict fails, the park at
error_reauth holds, and lastSyncedAt does not move. The two measurement write
catches do the same, so fetched-but-unwritten rows are re-fetched next tick
instead of aging out of the overlap.

syncUserFitbit now returns { imported, failed }; the manual-sync route, the
poll cohort adapter, and the backfill consume the new shape. The backfill
stamps backfillCompletedAt only after a clean full-history walk: a parked
connection returns without stamping, and a failed verdict throws so pg-boss
retries on the already-configured retry budget.

src/app/api/fitbit/sync/route.ts

@@ -58,6 +58,6 @@ export const POST = apiHandler(async (request: NextRequest) => {
     }
   }
 
-  const imported = await syncUserFitbit(user.id, { fullSync });
+  const { imported } = await syncUserFitbit(user.id, { fullSync });
   return apiSuccess({ imported, fullSync });
 });

src/lib/fitbit/sync-core.ts

@@ -98,6 +98,12 @@ export async function getValidToken(
           message: "Fitbit credentials missing — token refresh skipped",
           errorCode: "credentials_missing",
         });
+        // A dead token means every resource "succeeds" with 0 rows — without a
+        // ledger entry the cycle would read as clean, stamp `markSynced`, and
+        // `recordSyncSuccess` would flip a parked connection back to connected
+        // (an hourly invalid-refresh hammer with an advancing watermark). Fail
+        // the cycle's verdict so the park holds.
+        noteHardFailure(FITBIT_TOKEN_HARD_FAIL);
         return null;
       }
 
@@ -143,7 +149,16 @@ export async function getValidToken(
         },
       );
 
-      if (!persistedAccessToken) return null;
+      if (!persistedAccessToken) {
+        // `persistRotatedToken` returns null ONLY when the connection row
+        // vanished mid-flight (deleted between the read and the CAS re-read) —
+        // the normal lost-race path reuses the peer's rotated token above. A
+        // vanished-row cycle fetches nothing; register on the ledger so the
+        // same verdict rule as the refresh branch holds and the cycle does not
+        // stamp success over a connection that no longer exists.
+        noteHardFailure(FITBIT_TOKEN_HARD_FAIL);
+        return null;
+      }
 
       return {
         accessToken: persistedAccessToken,
@@ -157,6 +172,14 @@ export async function getValidToken(
         `Fitbit token refresh failed for user ${userId}: ${err}`,
       );
       await recordFitbitSyncFailure(userId, err);
+      // Same verdict rule as the credentials-missing branch above: a refresh
+      // failure must fail the cycle, or the all-zeros run stamps success,
+      // un-parks `error_reauth`, and the hourly cohort retries the dead refresh
+      // token forever while `lastSyncedAt` advances past real data. Fitbit's
+      // reconnect CTA is driven by the status-ledger park (`error_reauth`) that
+      // `recordFitbitSyncFailure` sets — there is no `needsReauth` column to
+      // flag (Google's connection-flag write is deliberately not ported).
+      noteHardFailure(FITBIT_TOKEN_HARD_FAIL);
       return null;
     }
   }
@@ -223,6 +246,27 @@ export interface HardFailTracker {
 }
 export const hardFailStorage = new AsyncLocalStorage<HardFailTracker>();
 
+/**
+ * Ledger label for a dead-token failure (`getValidToken` returning null on a
+ * missing-credentials, failed-refresh, or vanished-connection path). Distinct
+ * from a fetch/write label so a future one-shot leaf job could tell "the token
+ * is dead" (park, don't retry) apart from "a fetch/write hard-failed" (retry).
+ * Ported from Google Health's `GOOGLE_HEALTH_TOKEN_HARD_FAIL`.
+ */
+export const FITBIT_TOKEN_HARD_FAIL = "token";
+
+/**
+ * Record a hard failure on the ambient per-cycle ledger, if one is in scope.
+ * No-op outside a ledger scope (e.g. a direct call from a REPL script). Used by
+ * `getValidToken`'s dead-token paths and the measurement write catches —
+ * anything that must fail the cycle's verdict without aborting sibling work.
+ * Ported from Google Health's `noteHardFailure`.
+ */
+export function noteHardFailure(label: string): void {
+  const tracker = hardFailStorage.getStore();
+  if (tracker) tracker.failures.push(label);
+}
+
 /**
  * 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; a
@@ -251,8 +295,7 @@ export async function handleCollectionFetchError(
   }
   await recordFitbitSyncFailure(userId, err);
   getEvent()?.addWarning(`fitbit ${resource} failed for ${userId}: ${err}`);
-  const hardTracker = hardFailStorage.getStore();
-  if (hardTracker) hardTracker.failures.push(resource);
+  noteHardFailure(resource);
   return 0;
 }
 
@@ -570,8 +613,7 @@ export async function upsertFitbitMeasurements(
       // ambient ledger so the watermark holds and the next tick retries the
       // rescue rather than losing the rows for good.
       getEvent()?.addWarning(`Fitbit: natural-key rescue probe failed: ${err}`);
-      const hardTracker = hardFailStorage.getStore();
-      if (hardTracker) hardTracker.failures.push("measurements:rescue");
+      noteHardFailure("measurements:rescue");
     }
   }
 
@@ -612,6 +654,11 @@ export async function upsertFitbitMeasurements(
       }
     } catch (err) {
       getEvent()?.addWarning(`Fitbit: failed to create measurements: ${err}`);
+      // Fetched-but-unwritten rows are lost for good once they age out of the
+      // 24 h overlap — fail the cycle's verdict so the watermark holds and the
+      // next tick re-fetches. No rethrow: sibling chunks and resources must
+      // keep writing.
+      noteHardFailure("measurements:create");
     }
   }
 
@@ -643,6 +690,10 @@ export async function upsertFitbitMeasurements(
       imported++;
     } catch (err) {
       getEvent()?.addWarning(`Fitbit: failed to update measurement: ${err}`);
+      // Same rule as the create catch: a swallowed update loses the re-fetched
+      // value once it ages out of the overlap — hold the watermark so the next
+      // tick retries. No rethrow, so the remaining rows still write.
+      noteHardFailure("measurements:update");
     }
   }
 

src/lib/fitbit/sync.ts

@@ -50,12 +50,12 @@ import { syncUserWorkout } from "./sync-workout";
 export async function syncUserFitbit(
   userId: string,
   opts: { fullSync?: boolean } = {},
-): Promise<number> {
+): Promise<{ imported: number; failed: boolean }> {
   if (await isReauthRequired(userId, "fitbit")) {
     getEvent()?.addWarning(
       `fitbit sync skipped for ${userId}: parked at error_reauth`,
     );
-    return 0;
+    return { imported: 0, failed: true };
   }
 
   // Snapshot the incremental window ONCE for the whole cycle. Every resource
@@ -65,7 +65,7 @@ export async function syncUserFitbit(
     where: { userId },
     select: { lastSyncedAt: true },
   });
-  if (!connection) return 0;
+  if (!connection) return { imported: 0, failed: true };
   const end = new Date();
   const start = incrementalStart(connection.lastSyncedAt, {
     fullSync: opts.fullSync,
@@ -151,8 +151,9 @@ export async function syncUserFitbit(
   // partial cycle (some rows imported, or at least one resource that did not
   // soft-skip) stamps success as normal.
   const allSoftSkipped = tracker.count >= resources.length && total === 0;
+  const failed = anyFailed || allSoftSkipped;
 
-  if (!anyFailed && !allSoftSkipped) {
+  if (!failed) {
     // Stamp the watermark ONCE for the whole cycle, only on a non-degenerate
     // run. Every resource already saw the snapshot `start`, so stamping now()
     // here can't shrink a later resource's window.
@@ -161,9 +162,9 @@ export async function syncUserFitbit(
   }
 
   annotate({
-    action: { name: "fitbit.sync", details: { imported: total } },
+    action: { name: "fitbit.sync", details: { imported: total, failed } },
   });
-  return total;
+  return { imported: total, failed };
 }
 
 /**
@@ -201,7 +202,9 @@ export async function runFitbitPollCohort(
     onUserSynced?: (userId: string, imported: number) => void;
   } = {},
 ): Promise<{ usersSynced: number; measurementsImported: number }> {
-  const sync = opts.sync ?? ((userId: string) => syncUserFitbit(userId));
+  const sync =
+    opts.sync ??
+    ((userId: string) => syncUserFitbit(userId).then((r) => r.imported));
   const limit = pLimit(opts.concurrency ?? FITBIT_POLL_CONCURRENCY);
 
   let usersSynced = 0;

src/lib/jobs/fitbit-backfill.ts

@@ -14,6 +14,7 @@ import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
+import { isReauthRequired } from "@/lib/integrations/status";
 import { syncUserFitbit } from "@/lib/fitbit/sync";
 
 export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill";
@@ -32,14 +33,45 @@ export interface FitbitBackfillPayload {
 
 /**
  * Per-user backfill handler. Runs a full-history sync for one account and stamps
- * `backfillCompletedAt` so the discovery query drops it. Idempotent: the
- * per-resource upserts are key-stable, so a re-run (e.g. a reboot mid-walk)
- * overwrites rather than duplicating. Mirrors `runWhoopBackfillForUser`.
+ * `backfillCompletedAt` ONLY on a clean run, so the discovery query drops it.
+ * Idempotent: the per-resource upserts are key-stable, so a re-run (e.g. a
+ * reboot mid-walk) overwrites rather than duplicating. Mirrors
+ * `runWhoopBackfillForUser` (whose identical unconditional-stamp bug is fixed
+ * separately — this is the template for that port).
+ *
+ * Verdict-gated: a partial hard failure (one collection 400ing, a write
+ * failing, a dead token) THROWS so pg-boss retries the job — stamping the
+ * marker over a partial walk would freeze the history gap in forever and the
+ * configured `retryLimit` would be dead code. A connection parked at
+ * `error_reauth` is NOT clean either, but throwing against a dead grant would
+ * just burn the retry budget on the same no-op: return WITHOUT stamping — the
+ * boot discovery re-enqueues the account on the next boot, and the stamp lands
+ * once the user reconnects and a clean walk completes.
  */
 export async function runFitbitBackfillForUser(
   userId: string,
 ): Promise<{ imported: number }> {
-  const imported = await syncUserFitbit(userId, { fullSync: true });
+  if (await isReauthRequired(userId, "fitbit")) {
+    annotate({
+      action: {
+        name: "fitbit.backfill.skipped_reauth",
+        details: { imported: 0 },
+      },
+    });
+    return { imported: 0 };
+  }
+
+  const { imported, failed } = await syncUserFitbit(userId, { fullSync: true });
+
+  if (failed) {
+    // Surface through the pg-boss retry path (retryLimit 3, backoff, set by
+    // `integrationBackfillSourceOptions`). If the failure parked the connection
+    // at error_reauth meanwhile, the next attempt takes the reauth return above
+    // instead of failing again.
+    throw new Error(
+      `fitbit backfill incomplete for user ${userId} — marker not stamped`,
+    );
+  }
 
   await prisma.fitbitConnection.update({
     where: { userId },