diff --git a/src/app/api/fitbit/sync/route.ts b/src/app/api/fitbit/sync/route.ts index 5197ef8e8..2f7641fc7 100644 --- a/src/app/api/fitbit/sync/route.ts +++ b/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 }); }); diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index 2f1bcc721..d6816ef8e 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/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(); +/** + * 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"); } } diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index f2596f334..bb692496e 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -50,12 +50,12 @@ import { syncUserWorkout } from "./sync-workout"; export async function syncUserFitbit( userId: string, opts: { fullSync?: boolean } = {}, -): Promise { +): 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; diff --git a/src/lib/jobs/fitbit-backfill.ts b/src/lib/jobs/fitbit-backfill.ts index 32a5fce36..fc57592fa 100644 --- a/src/lib/jobs/fitbit-backfill.ts +++ b/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 },