diff --git a/messages/de.json b/messages/de.json index ee218f830..861e0a940 100644 --- a/messages/de.json +++ b/messages/de.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Verbunden", "errorReconnect": "Fehler — neu verbinden", - "warningServerError": "Verbunden, Sync-Problem", + "warningServerError": "Sync fehlgeschlagen", "parkedReconnect": "Pausiert — manuell wieder verbinden", "notConnected": "Nicht verbunden", "justNow": "gerade eben", diff --git a/messages/en.json b/messages/en.json index 4c498d987..7edc56190 100644 --- a/messages/en.json +++ b/messages/en.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Connected", "errorReconnect": "Error — reconnect", - "warningServerError": "Connected, sync issue", + "warningServerError": "Sync failing", "parkedReconnect": "Paused — reconnect manually", "notConnected": "Not connected", "justNow": "just now", diff --git a/messages/es.json b/messages/es.json index 8c609dfcd..a4c955d17 100644 --- a/messages/es.json +++ b/messages/es.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Conectado", "errorReconnect": "Error — reconectar", - "warningServerError": "Conectado, problema de sincronización", + "warningServerError": "Sincronización con errores", "parkedReconnect": "Pausado — reconectar manualmente", "notConnected": "No conectado", "justNow": "ahora mismo", diff --git a/messages/fr.json b/messages/fr.json index 678b46765..cd3cc3217 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Connecté", "errorReconnect": "Erreur — reconnecter", - "warningServerError": "Connecté, problème de synchronisation", + "warningServerError": "Échec de synchronisation", "parkedReconnect": "En pause — reconnecter manuellement", "notConnected": "Non connecté", "justNow": "à l’instant", diff --git a/messages/it.json b/messages/it.json index 5a791cdc9..858cab7f8 100644 --- a/messages/it.json +++ b/messages/it.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Connesso", "errorReconnect": "Errore — riconnetti", - "warningServerError": "Connesso, problema di sincronizzazione", + "warningServerError": "Sincronizzazione in errore", "parkedReconnect": "In pausa — riconnettere manualmente", "notConnected": "Non connesso", "justNow": "proprio ora", diff --git a/messages/pl.json b/messages/pl.json index 92b747092..4a8f61594 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -5155,7 +5155,7 @@ "integrationPill": { "connected": "Połączono", "errorReconnect": "Błąd — połącz ponownie", - "warningServerError": "Połączono, problem z synchronizacją", + "warningServerError": "Synchronizacja nie działa", "parkedReconnect": "Wstrzymane — połącz ręcznie", "notConnected": "Nie połączono", "justNow": "przed chwilą", diff --git a/prisma/migrations/0275_integration_failure_streak_and_leg/migration.sql b/prisma/migrations/0275_integration_failure_streak_and_leg/migration.sql new file mode 100644 index 000000000..d473ff3a2 --- /dev/null +++ b/prisma/migrations/0275_integration_failure_streak_and_leg/migration.sql @@ -0,0 +1,30 @@ +-- Integration sync ledger: date the failure streak, and name the leg that owns it. +-- +-- `failing_since_at` is the start of the current unbroken failure streak, +-- whatever kind the classifier assigned. The per-kind buckets counted attempts +-- but nothing dated them, so a streak the classifier had called `transient` +-- could run for a thousand attempts without ever becoming anything: the park +-- test read `kind = 'persistent'` and the persistent anchor was only ever +-- stamped by a persistent failure. +-- +-- `failing_leg` records which sync leg the error came from. The table is keyed +-- (user_id, integration) while Withings runs four legs on four crons and WHOOP +-- runs four resources on four more, so a sibling leg's success was clearing an +-- error it had not caused — several times an hour, which is why the strike +-- ladder never climbed. +ALTER TABLE "integration_statuses" + ADD COLUMN "failing_since_at" TIMESTAMP(3), + ADD COLUMN "failing_leg" TEXT; + +-- Backfill for rows that are already failing. Without this a connection that +-- has been broken for weeks would restart its clock at the next attempt and +-- need another two days before it escalated. `last_success_at` is the honest +-- anchor where it exists (the streak began at most one cron tick after it); +-- rows that never succeeded fall back to the last attempt, then to creation. +-- `failing_leg` is left NULL: which leg broke is not recoverable from the row, +-- and NULL means "unattributed", under which any success still clears — the +-- pre-existing behaviour, applied only to the backfilled rows. +UPDATE "integration_statuses" +SET "failing_since_at" = COALESCE("last_success_at", "last_attempt_at", "created_at") +WHERE "state" IN ('error_transient', 'error_reauth', 'parked') + AND "failing_since_at" IS NULL; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index b625607b2..b54bdac41 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4137,6 +4137,16 @@ model IntegrationStatus { // sole source of truth for the alert ladder. consecutiveFailuresByKind Json? @map("consecutive_failures_by_kind") // v1.4.43 W14 — { transient, reauth_required, persistent } persistentFailureStartedAt DateTime? @map("persistent_failure_started_at") // v1.4.43 W14 — start of the current persistent streak (drives >24h park) + // Start of the current unbroken failure streak, whatever the kind. The + // per-kind buckets count attempts; this dates them, which is what tells a + // blip apart from a dead pipe. Cleared by a success that is entitled to + // clear it, by a reconnect, a disconnect, and a resume. + failingSinceAt DateTime? @map("failing_since_at") + // Which sync leg owns the error currently on the row. `null` = the provider + // has a single leg, or the failure predates leg attribution. A success from + // a DIFFERENT leg may not clear an error this one did not cause — see + // `recordSyncSuccess`. + failingLeg String? @map("failing_leg") alertedAt DateTime? @map("alerted_at") // last time admin Telegram alert fired (idempotency guard) createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") diff --git a/src/app/api/integrations/status/route.ts b/src/app/api/integrations/status/route.ts index e74ad777f..9611e5c57 100644 --- a/src/app/api/integrations/status/route.ts +++ b/src/app/api/integrations/status/route.ts @@ -168,6 +168,7 @@ export const GET = apiHandler(async () => { state: string; lastSuccessAt: string | null; lastAttemptAt: string | null; + failingSince: string | null; }, opts: { connected?: boolean; @@ -181,6 +182,7 @@ export const GET = apiHandler(async () => { state: snapshot.state, lastSuccessAt: snapshot.lastSuccessAt, lastAttemptAt: snapshot.lastAttemptAt, + failingSinceAt: snapshot.failingSince, legacyLastSyncedAt: opts.legacyLastSyncedAt ?? null, cadence: INTEGRATION_CADENCE[integration], now, @@ -381,6 +383,12 @@ interface IntegrationViewModel { lastSuccessAt: string | null; lastAttemptAt: string | null; lastError: string | null; + /** + * Start of the current unbroken failure streak, or null when the provider is + * not failing. `syncHealth.since` is derived from it; the raw instant rides + * along so a client can render the age without re-deriving the verdict. + */ + failingSince: string | null; /** * The server-resolved liveness verdict. Present on EVERY entry, whether or * not a card renders it — the unit of coverage is the envelope entry, not diff --git a/src/components/settings/integration-status-pill.tsx b/src/components/settings/integration-status-pill.tsx index 35d8ef14d..0850617cc 100644 --- a/src/components/settings/integration-status-pill.tsx +++ b/src/components/settings/integration-status-pill.tsx @@ -110,11 +110,20 @@ const WARNING_STATES = new Set([ "parked", ]); -/** The states whose label carries an inline relative timestamp. */ +/** + * The states whose label carries an inline relative timestamp. + * + * `warning` is in the set because leaving it out was the whole lie: the pill + * said "sync issue" identically on hour one and on day seventeen, and the age + * — which the mapper computes and passes in — was dropped on the floor. How + * long it has been broken is the only part of that sentence a person can act + * on. + */ const RELATIVE_STATES = new Set([ "connected", "stale", "stalled", + "warning", ]); export function IntegrationStatusPill({ diff --git a/src/components/settings/integrations/shared.tsx b/src/components/settings/integrations/shared.tsx index 24583bec8..a7fb3a2d6 100644 --- a/src/components/settings/integrations/shared.tsx +++ b/src/components/settings/integrations/shared.tsx @@ -161,14 +161,19 @@ export function pillStateFor( /** * The timestamp the pill shows inline. `connected` reads the last success (a * connection row's own `lastSyncedAt` wins when it is newer, which is what - * repaints a fresh "just now" straight after a manual sync); the two aging - * states read the instant that triggered the verdict. + * repaints a fresh "just now" straight after a manual sync); the aging and + * failing states read the instant that triggered the verdict. + * + * `failing` must NOT fall through to the `legacyLastSyncedAt` arm below. That + * field is a connection row's own clock, kept fresh by the very sync that is + * erroring, so a provider that had delivered nothing for a fortnight painted a + * "just now" beside its warning. */ export function pillTimestampFor( status: IntegrationStatusViewModel | undefined, ): string | null { const verdict = status?.syncHealth?.verdict; - if (verdict === "stale" || verdict === "stalled") { + if (verdict === "stale" || verdict === "stalled" || verdict === "failing") { return status?.syncHealth?.since ?? null; } return status?.legacyLastSyncedAt ?? status?.lastSuccessAt ?? null; diff --git a/src/lib/integrations/__tests__/status.test.ts b/src/lib/integrations/__tests__/status.test.ts index 4eea74911..e8a95ac1b 100644 --- a/src/lib/integrations/__tests__/status.test.ts +++ b/src/lib/integrations/__tests__/status.test.ts @@ -108,6 +108,8 @@ describe("recordSyncSuccess", () => { persistent: 0, }, persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }); }); @@ -493,6 +495,8 @@ describe("markReauthRequired / markDisconnected / markReconnected", () => { persistent: 0, }, persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }); }); @@ -516,6 +520,8 @@ describe("markReauthRequired / markDisconnected / markReconnected", () => { persistent: 0, }, persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }); }); @@ -833,6 +839,8 @@ describe("resumeIntegrationFromPark", () => { persistent: 0, }, persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }); diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index b844cb4ed..8edc70e9f 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -34,7 +34,7 @@ import { prisma } from "@/lib/db"; import { encrypt, decrypt } from "@/lib/crypto"; import { auditLog } from "@/lib/auth/audit"; -import { getEvent } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; import type { IntegrationClassification } from "@/lib/integrations/http-status-classifier"; @@ -194,6 +194,41 @@ const ALERT_REPEAT_WINDOW_MS = 24 * 60 * 60 * 1000; */ const PARK_PERSISTENT_FAILURE_AFTER_MS = 24 * 60 * 60 * 1000; +/** + * How long an unbroken failure streak may keep claiming to be `transient` + * before the claim is treated as refuted and the failure is recorded as + * `persistent` instead. + * + * `transient` is a hypothesis, not an observation: the classifier defaults a + * timeout, a 429, a 5xx and any unrecognised response to it precisely because + * the wire cannot tell "the upstream is rebooting" from "this will never work + * again". That default is right for one attempt and wrong for a thousand — a + * streak that has been retried hourly for a full day without a single success + * has falsified the hypothesis by experiment, and continuing to call it + * transient is the ledger asserting a certainty the wire never gave it. + * + * 24 h is the same window the persistent park already uses. An upstream + * maintenance night, a rate-limit cool-off, a redeploy and a DNS wobble all + * fit comfortably inside it, so a genuine blip recovers and resets the streak + * long before this trips — nothing that heals on its own takes a day. The + * escalation is additionally gated on `getPersistentFailureThreshold()` + * consecutive failures, so a provider whose cron ticks rarely cannot escalate + * on its second-ever attempt just because wall-clock time passed. + * + * Escalating feeds the EXISTING ladder rather than a second one: the + * `persistent` bucket starts counting, `persistentFailureStartedAt` anchors, + * and `PARK_PERSISTENT_FAILURE_AFTER_MS` later the row parks and stops + * retrying — 48 h of unbroken failure end to end before anything is disabled. + */ +const TRANSIENT_ESCALATION_AFTER_MS = 24 * 60 * 60 * 1000; + +/** Row states that mean "an error is currently held on this row". */ +const ERROR_STATES: ReadonlySet = new Set([ + "error_transient", + "error_reauth", + "parked", +]); + export interface IntegrationStatusSnapshot { integration: IntegrationKey; state: IntegrationState; @@ -205,6 +240,13 @@ export interface IntegrationStatusSnapshot { * of truth now that the legacy `consecutiveFailures` column is * gone. */ consecutiveFailuresByKind: ConsecutiveFailuresByKind | null; + /** + * Start of the current unbroken failure streak, whatever the kind. The + * buckets count attempts; this dates them, which is what tells a blip apart + * from a pipe that has been dead for weeks. `null` whenever the row is not + * currently failing. + */ + failingSince: string | null; } /** @@ -271,6 +313,7 @@ export async function getIntegrationStatus( lastAttemptAt: null, lastError: null, consecutiveFailuresByKind: null, + failingSince: null, }; } return { @@ -280,6 +323,7 @@ export async function getIntegrationStatus( lastAttemptAt: row.lastAttemptAt?.toISOString() ?? null, lastError: row.lastError ? safeDecryptError(row.lastError) : null, consecutiveFailuresByKind: readBucketColumn(row.consecutiveFailuresByKind), + failingSince: row.failingSinceAt?.toISOString() ?? null, }; } @@ -305,17 +349,79 @@ export async function isReauthRequired( return row?.state === "error_reauth" || row?.state === "parked"; } +export interface RecordSyncSuccessOptions { + /** + * The sync leg that succeeded, for a provider whose legs run on separate + * schedules (Withings measures / activity / sleep / ECG, the four WHOOP + * resources, the two Polar legs). Omit it for a single-leg provider or for a + * full pass that covered every leg — an omitted leg clears the row whole, + * which is the pre-existing behaviour and the correct one for those callers. + */ + leg?: string; +} + /** * Record a successful sync. Resets the failure counter, clears any * prior error message, and (importantly) flips state back to * `connected` even from `error_reauth` — the Withings flow re-enters * after the OAuth callback writes a new refresh token. + * + * A success only clears what it is entitled to clear. `IntegrationStatus` is + * keyed `(userId, integration)`, but Withings runs four legs on four crons and + * WHOOP runs four resources on four more — so an unconditional clear meant the + * ECG leg succeeding at :41 wiped the error, the buckets, the streak anchor and + * the state that the sleep leg had recorded at :15. The failing leg's strike + * ladder was reset several times an hour, so it could never reach the alert + * threshold or the park window, and the card painted a green "connected · 12 + * minutes ago" over a data class that had stopped arriving. When the row holds + * an error owned by a DIFFERENT named leg, this success records its attempt and + * leaves the error standing. + * + * The partial path deliberately does NOT advance `lastSuccessAt`. That field is + * what the freshness arms and the card's "connected · X ago" read, and moving + * it forward while another leg is dead is exactly the sentence this change + * exists to stop the ledger from saying. `lastAttemptAt` does advance, so the + * row still reads as "being tried" rather than ageing into `stalled`. */ export async function recordSyncSuccess( userId: string, integration: IntegrationKey, + options?: RecordSyncSuccessOptions, ): Promise { const now = new Date(); + const leg = options?.leg ?? null; + + if (leg !== null) { + const existing = await prisma.integrationStatus.findUnique({ + where: { userId_integration: { userId, integration } }, + select: { state: true, failingLeg: true }, + }); + const heldByAnotherLeg = + existing != null && + existing.failingLeg != null && + existing.failingLeg !== leg && + ERROR_STATES.has(existing.state); + + if (heldByAnotherLeg) { + await prisma.integrationStatus.update({ + where: { userId_integration: { userId, integration } }, + data: { lastAttemptAt: now }, + }); + // Meta only — the enclosing wide event owns its own action name (the + // cron task), and overwriting it here would rename the job. + annotate({ + meta: { + integration_partial_success: { + integration, + succeeded_leg: leg, + failing_leg: existing.failingLeg, + }, + }, + }); + return; + } + } + // v1.4.43 W14 — a success resets ALL per-kind buckets back to zero // and clears the persistent-streak start timestamp. // v1.4.47 W1 — the legacy `consecutiveFailures` column was dropped @@ -337,6 +443,8 @@ export async function recordSyncSuccess( lastError: null, consecutiveFailuresByKind: zeroBuckets(), persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }, }); @@ -349,6 +457,12 @@ export interface RecordSyncFailureInput { message: string; /** Optional structured error code (e.g. "invalid_grant", "401"). */ errorCode?: string; + /** + * The sync leg that failed, for a provider whose legs run on separate + * schedules. Recorded on the row so a sibling leg's success cannot clear an + * error it did not cause. Omit it for a single-leg provider. + */ + leg?: string; } /** @@ -375,23 +489,35 @@ export interface RecordSyncFailureInput { * v1.4.47 W1 — the legacy single-column `consecutiveFailures` integer * was dropped (migration 0077); the bucket is now the sole counter * and the back-fill branch is gone. + * + * A `transient` failure whose streak has run unbroken for longer than + * `TRANSIENT_ESCALATION_AFTER_MS` is recorded as `persistent` instead. Before + * that, a `transient` streak had no exit at all: the park test read + * `kind === "persistent"`, `persistentFailureStartedAt` was only stamped on a + * persistent failure, and so a streak the classifier had bucketed as transient + * could run for a thousand attempts with the persistent counter at zero, the + * state frozen at `error_transient`, and no ladder it could ever climb. See + * the constant for why time, rather than count, is the honest signal. */ export async function recordSyncFailure( input: RecordSyncFailureInput, ): Promise { const { userId, integration, kind, message, errorCode } = input; + const leg = input.leg ?? null; const now = new Date(); const encryptedError = safeEncryptError(message); // Read the current row first so we can: // (a) compute the new per-kind bucket value - // (b) decide whether the persistent streak has exceeded the + // (b) decide whether the streak has outlived the `transient` claim + // (c) decide whether the persistent streak has exceeded the // park threshold const existing = await prisma.integrationStatus.findUnique({ where: { userId_integration: { userId, integration } }, select: { consecutiveFailuresByKind: true, persistentFailureStartedAt: true, + failingSinceAt: true, alertedAt: true, }, }); @@ -402,6 +528,30 @@ export async function recordSyncFailure( // or for the first-ever write of this (user, integration) pair const startingBuckets = readBucketColumn(existing?.consecutiveFailuresByKind); + // Streak anchor: the first failure of the current unbroken run. Every path + // that resolves a failure (success entitled to clear, reconnect, disconnect, + // resume) nulls it, so a non-null value always dates a live streak. + const failingSinceAt = existing?.failingSinceAt ?? now; + const streakAgeMs = now.getTime() - failingSinceAt.getTime(); + + // The `transient` hypothesis, tested against the clock. Both conditions are + // required: the wall-clock window rules out a blip, and the consecutive-count + // gate keeps a rarely-polled provider from escalating on its second attempt + // just because a day of real time went by. `+ 1` because this failure has not + // been written into the bucket yet. + const transientStreakAfterThis = (startingBuckets?.transient ?? 0) + 1; + const escalatedFromTransient = + kind === "transient" && + streakAgeMs >= TRANSIENT_ESCALATION_AFTER_MS && + transientStreakAfterThis >= getPersistentFailureThreshold(); + + // Everything downstream — bucket, state mapping, park test, audit detail, + // alert copy — reads the effective kind, so the escalation feeds the ladder + // that already exists instead of growing a parallel one. + const effectiveKind: FailureKind = escalatedFromTransient + ? "persistent" + : kind; + // Snapshot the persistent-bucket count BEFORE we increment so the // streak-anchor decision below can see the pre-increment value. // Otherwise we'd inspect the freshly-incremented bucket and never @@ -419,14 +569,21 @@ export async function recordSyncFailure( // Increment only the bucket matching this failure's kind. // The other two buckets stay at their current value so a persistent // streak isn't reset by an intervening transient hiccup. - buckets[kind] = (buckets[kind] ?? 0) + 1; + buckets[effectiveKind] = (buckets[effectiveKind] ?? 0) + 1; // Track the persistent-streak start so the >24h park check has a // wall-clock anchor. Only stamped on the FIRST persistent failure of // a streak; cleared on success or when the persistent bucket goes // back to zero (which today only happens via success, but the logic // is symmetric for future "transient drained the streak" rules). - const isPersistent = kind === "persistent"; + // + // An escalated transient anchors here too, at the moment of escalation + // rather than at the streak start. That is deliberate: it dates when the + // failure started being treated as unrecoverable, which is what the park + // window measures, and it puts a full second day between "this is no longer + // transient" and "stop retrying" — 48 h of unbroken failure before an + // integration disables itself over a verdict the wire never actually gave. + const isPersistent = effectiveKind === "persistent"; let persistentFailureStartedAt: Date | null = existing?.persistentFailureStartedAt ?? null; if (isPersistent && persistentStreakBefore === 0) { @@ -455,7 +612,7 @@ export async function recordSyncFailure( // grep for contract-bug bursts) const newState: IntegrationState = shouldPark ? "parked" - : kind === "reauth_required" + : effectiveKind === "reauth_required" ? "error_reauth" : "error_transient"; @@ -469,6 +626,8 @@ export async function recordSyncFailure( lastError: encryptedError, consecutiveFailuresByKind: buckets, persistentFailureStartedAt: isPersistent ? now : null, + failingSinceAt, + failingLeg: leg, }, update: { state: newState, @@ -476,6 +635,8 @@ export async function recordSyncFailure( lastError: encryptedError, consecutiveFailuresByKind: buckets, persistentFailureStartedAt, + failingSinceAt, + failingLeg: leg, }, }); @@ -496,11 +657,18 @@ export async function recordSyncFailure( userId, details: { integration, - kind, + kind: effectiveKind, + // Non-null only when the clock overrode the classifier, so an operator + // reading the trail can tell "the upstream said this was permanent" from + // "we concluded it was after a day of retrying". + escalatedFrom: escalatedFromTransient ? kind : null, + failingLeg: leg, + failingSinceAt: failingSinceAt.toISOString(), + streakAgeMs, errorCode: errorCode ?? null, message, attemptNumber: bucketTotal, - bucketCount: buckets[kind], + bucketCount: buckets[effectiveKind], state: newState, }, }); @@ -528,7 +696,7 @@ export async function recordSyncFailure( await maybeAlertAdmins({ userId, integration, - kind, + kind: effectiveKind, message, errorCode, consecutiveFailures: alertSignal, @@ -560,6 +728,10 @@ export async function recordSyncFailure( persistentFailureStartedAt: existing.persistentFailureStartedAt.toISOString(), persistentStreakAgeMs, + // The whole streak, which for an escalated transient is a full day + // longer than the persistent window above. + failingSinceAt: failingSinceAt.toISOString(), + streakAgeMs, errorCode: errorCode ?? null, message, }, @@ -603,6 +775,8 @@ export async function resumeIntegrationFromPark( lastError: null, consecutiveFailuresByKind: zeroBuckets(), persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }, }); @@ -646,8 +820,11 @@ export async function parkIntegrationAtReauth(opts: { integration: IntegrationKey; message: string; errorCode: string; + /** The leg whose scope gap forced the park, for multi-leg providers. */ + leg?: string; }): Promise { const { userId, integration, message, errorCode } = opts; + const leg = opts.leg ?? null; const now = new Date(); const encryptedError = safeEncryptError(message); @@ -656,7 +833,7 @@ export async function parkIntegrationAtReauth(opts: { // audit log when the state or error changes. const existing = await prisma.integrationStatus.findUnique({ where: { userId_integration: { userId, integration } }, - select: { state: true, lastError: true }, + select: { state: true, lastError: true, failingSinceAt: true }, }); const isFreshPark = existing?.state !== "error_reauth" || existing.lastError !== encryptedError; @@ -673,6 +850,8 @@ export async function parkIntegrationAtReauth(opts: { // zero so a later genuine transient burst still has the full // 3-strike runway before paging. consecutiveFailuresByKind: zeroBuckets(), + failingSinceAt: now, + failingLeg: leg, }, update: { state: "error_reauth", @@ -681,6 +860,8 @@ export async function parkIntegrationAtReauth(opts: { // Deliberately omit `consecutiveFailuresByKind` — the existing // bucket values are preserved exactly. This is the whole point // of the helper. + failingSinceAt: existing?.failingSinceAt ?? now, + failingLeg: leg, }, }); @@ -703,6 +884,12 @@ export async function markReauthRequired( integration: IntegrationKey, message: string, ): Promise { + const now = new Date(); + const existing = await prisma.integrationStatus.findUnique({ + where: { userId_integration: { userId, integration } }, + select: { failingSinceAt: true }, + }); + // v1.4.47 W1 — the legacy `consecutiveFailures` column was dropped // (migration 0077). Out-of-band reauth detection (proactive token // refresh) seeds the `reauth_required` bucket at 1 on a first-ever @@ -720,11 +907,15 @@ export async function markReauthRequired( reauth_required: 1, persistent: 0, }, - lastAttemptAt: new Date(), + lastAttemptAt: now, + failingSinceAt: now, }, update: { state: "error_reauth", lastError: safeEncryptError(message), + // Never re-stamp a streak already running — the anchor dates the + // failure, not the detection. + failingSinceAt: existing?.failingSinceAt ?? now, }, }); @@ -756,6 +947,8 @@ export async function markDisconnected( lastError: null, consecutiveFailuresByKind: zeroBuckets(), persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }, }); @@ -783,6 +976,8 @@ export async function markReconnected( lastError: null, consecutiveFailuresByKind: zeroBuckets(), persistentFailureStartedAt: null, + failingSinceAt: null, + failingLeg: null, alertedAt: null, }, }); diff --git a/src/lib/integrations/sync-verdict.ts b/src/lib/integrations/sync-verdict.ts index f4ab30bce..d0313902b 100644 --- a/src/lib/integrations/sync-verdict.ts +++ b/src/lib/integrations/sync-verdict.ts @@ -132,6 +132,14 @@ export interface SyncVerdictInput { lastAttemptAt?: string | Date | null; /** A connection row's own `lastSyncedAt`, for rows predating the ledger. */ legacyLastSyncedAt?: string | Date | null; + /** + * Start of the current unbroken failure streak, from the ledger. This is what + * the error verdicts date themselves by. `lastAttemptAt` is the wrong anchor + * — it is refreshed every hour by the retry itself, so a connection dead for + * a fortnight reported "failing, just now". `lastSuccessAt` is nearly right + * but a multi-leg provider can advance it while one leg stays broken. + */ + failingSinceAt?: string | Date | null; /** Newest observed data instant, when the caller knows it (Apple Health). */ lastDataAt?: string | Date | null; cadence: SyncCadence; @@ -183,6 +191,7 @@ export function resolveSyncVerdict(input: SyncVerdictInput): SyncHealth { const lastAttempt = toMillis(input.lastAttemptAt); const legacy = toMillis(input.legacyLastSyncedAt); const lastData = toMillis(input.lastDataAt); + const failingSince = toMillis(input.failingSinceAt); // 1. No live connection, or one the user explicitly ended. if (input.connected === false) @@ -196,10 +205,13 @@ export function resolveSyncVerdict(input: SyncVerdictInput): SyncHealth { // 2. States the user has to act on, in order of specificity. if (input.state === "parked") { - return { verdict: "parked", since: iso(lastAttempt) }; + return { verdict: "parked", since: iso(failingSince ?? lastAttempt) }; } if (input.state === "error_reauth") { - return { verdict: "reauth_required", since: iso(lastAttempt) }; + return { + verdict: "reauth_required", + since: iso(failingSince ?? lastAttempt), + }; } // 3. Nothing has ever run and nothing has ever arrived — the honest reading @@ -218,9 +230,12 @@ export function resolveSyncVerdict(input: SyncVerdictInput): SyncHealth { } } - // 5. Trying and erroring. + // 5. Trying and erroring. `since` dates the streak, not the last retry: this + // is the field the card renders, and "failing since a minute ago" on a + // connection that last delivered a fortnight back is the sentence this + // whole path exists to stop. if (input.state === "error_transient") { - return { verdict: "failing", since: iso(lastSuccess) }; + return { verdict: "failing", since: iso(failingSince ?? lastSuccess) }; } // 6. Running, but nothing new is arriving. The primary arm for push sources. diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index b8df364f5..10f07df56 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -24,6 +24,7 @@ import { import { recordWhoopSyncFailure, syncWhoopResourceWithStatus, + type WhoopSyncLeg, } from "@/lib/whoop/sync-core"; import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; @@ -68,6 +69,7 @@ type ByIdSync = (userId: string, resourceId: string) => Promise; */ export async function runWhoopResourceSync( taskName: string, + leg: WhoopSyncLeg, jobs: Job[], syncFn: CollectionSync, byIdFn?: ByIdSync, @@ -101,7 +103,7 @@ export async function runWhoopResourceSync( // Stamp `IntegrationStatus.lastSuccessAt` on a genuine per-resource // import — a webhook-driven refresh advanced the cursor but used to // leave the sync-status ledger frozen until the next full poll. - const imported = await syncWhoopResourceWithStatus(userId, () => + const imported = await syncWhoopResourceWithStatus(userId, leg, () => resourceId && byIdFn ? byIdFn(userId, resourceId) : syncFn(userId), ); measurementsImported += imported; @@ -119,7 +121,7 @@ export async function runWhoopResourceSync( // marked in `handleCollectionFetchError`, so record only the UNMARKED // escapes here. if (!isSyncFailureRecorded(err)) { - await recordWhoopSyncFailure(userId, err); + await recordWhoopSyncFailure(userId, err, leg); } evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } @@ -144,6 +146,7 @@ export async function runWhoopResourceSync( export function handleWhoopRecoverySync(jobs: Job[]) { return runWhoopResourceSync( "job.whoop_recovery_sync", + "recovery", jobs, syncUserRecovery, syncWhoopRecoveryById, @@ -153,6 +156,7 @@ export function handleWhoopRecoverySync(jobs: Job[]) { export function handleWhoopSleepSync(jobs: Job[]) { return runWhoopResourceSync( "job.whoop_sleep_sync", + "sleep", jobs, syncWhoopSleep, syncWhoopSleepById, @@ -162,6 +166,7 @@ export function handleWhoopSleepSync(jobs: Job[]) { export function handleWhoopWorkoutSync(jobs: Job[]) { return runWhoopResourceSync( "job.whoop_workout_sync", + "workout", jobs, syncUserWorkout, syncWhoopWorkoutById, @@ -170,5 +175,10 @@ export function handleWhoopWorkoutSync(jobs: Job[]) { export function handleWhoopCycleSync(jobs: Job[]) { // Cycle has no webhook (poll-only), so no fetch-by-id path. - return runWhoopResourceSync("job.whoop_cycle_sync", jobs, syncUserCycle); + return runWhoopResourceSync( + "job.whoop_cycle_sync", + "cycle", + jobs, + syncUserCycle, + ); } diff --git a/src/lib/mcp/__tests__/tools.test.ts b/src/lib/mcp/__tests__/tools.test.ts index e5c7526c5..ddf9c85f8 100644 --- a/src/lib/mcp/__tests__/tools.test.ts +++ b/src/lib/mcp/__tests__/tools.test.ts @@ -572,6 +572,7 @@ describe("get_integration_status", () => { lastAttemptAt: "2026-06-27T00:00:00.000Z", lastError: "token revoked", consecutiveFailuresByKind: null, + failingSince: null, }); const result = (await tool("get_integration_status").run(CTX, {})) as { @@ -611,6 +612,7 @@ describe("get_integration_status", () => { lastAttemptAt, lastError: null, consecutiveFailuresByKind: null, + failingSince: null, }); return tool("get_integration_status") diff --git a/src/lib/mcp/tools.ts b/src/lib/mcp/tools.ts index 46be574db..13abed92e 100644 --- a/src/lib/mcp/tools.ts +++ b/src/lib/mcp/tools.ts @@ -1518,6 +1518,7 @@ export const MCP_TOOLS: McpToolDefinition[] = [ state: s.state, lastSuccessAt: s.lastSuccessAt, lastAttemptAt: s.lastAttemptAt, + failingSinceAt: s.failingSince, cadence: INTEGRATION_CADENCE[s.integration], }); return { diff --git a/src/lib/polar/__tests__/sync.test.ts b/src/lib/polar/__tests__/sync.test.ts index 56e33a343..f5d12ff9c 100644 --- a/src/lib/polar/__tests__/sync.test.ts +++ b/src/lib/polar/__tests__/sync.test.ts @@ -196,7 +196,9 @@ describe("syncUserPolar", () => { externalId: "recharge:2026-06-10:recovery", value: 100, }); - expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar"); + expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar", { + leg: "vitals", + }); }); it("maps a cardio-load row with source POLAR + stable externalId", async () => { @@ -482,7 +484,9 @@ describe("syncUserPolar", () => { expect(recordFailureMock).not.toHaveBeenCalled(); expect(recordSuccessMock).toHaveBeenCalledTimes(1); - expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar"); + expect(recordSuccessMock).toHaveBeenCalledWith("u1", "polar", { + leg: "vitals", + }); }); it.each([ diff --git a/src/lib/polar/sync-workouts.ts b/src/lib/polar/sync-workouts.ts index 0f7e942c0..7084ae511 100644 --- a/src/lib/polar/sync-workouts.ts +++ b/src/lib/polar/sync-workouts.ts @@ -40,6 +40,8 @@ import { fetchExercises, mapExercise, type PolarWorkoutRow } from "./client"; import { getPolarConnection } from "./credentials"; import { PolarApiError, classifyPolarError } from "./response-classifier"; +const POLAR_LEG_WORKOUTS = "workouts"; + async function recordPolarWorkoutFailure( userId: string, err: unknown, @@ -47,6 +49,9 @@ async function recordPolarWorkoutFailure( await recordSyncFailure({ userId, integration: "polar", + // The workouts leg rides the same hourly tick as vitals but owns its own + // errors: the vitals success stamp must not clear a failure from here. + leg: POLAR_LEG_WORKOUTS, // Same classification the vitals leg uses (`classifyPolarFailure` in // `./sync`); resolved directly off the shared classifier so this leg does // not have to import the heavier vitals-sync module. diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index 048445a4f..e6f62a920 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -91,6 +91,8 @@ export function classifyPolarFailure(err: unknown): FailureKind { * record a future escape with the SAME provider-owned classification the inline * catch already applies. */ +export const POLAR_LEG_VITALS = "vitals"; + export async function recordPolarSyncFailure( userId: string, err: unknown, @@ -98,6 +100,7 @@ export async function recordPolarSyncFailure( await recordSyncFailure({ userId, integration: "polar", + leg: POLAR_LEG_VITALS, kind: classifyPolarFailure(err), message: err instanceof Error ? err.message : String(err), errorCode: @@ -288,6 +291,7 @@ export async function syncUserPolar(userId: string): Promise { await recordSyncFailure({ userId, integration: "polar", + leg: POLAR_LEG_VITALS, kind: classifyPolarFailure(firstErr), message: `partial sync failure (${names}): ${ firstErr instanceof Error ? firstErr.message : String(firstErr) @@ -300,7 +304,7 @@ export async function syncUserPolar(userId: string): Promise { return imported; } - await recordSyncSuccess(userId, "polar"); + await recordSyncSuccess(userId, "polar", { leg: POLAR_LEG_VITALS }); return imported; } diff --git a/src/lib/whoop/__tests__/sync-dead-token.test.ts b/src/lib/whoop/__tests__/sync-dead-token.test.ts index 3065f74a2..cea0a9b4e 100644 --- a/src/lib/whoop/__tests__/sync-dead-token.test.ts +++ b/src/lib/whoop/__tests__/sync-dead-token.test.ts @@ -259,19 +259,29 @@ describe("syncWhoopResourceWithStatus — dominant per-resource cron path", () = // The hourly-cron arm Fitbit has no twin for: a dead token noted by // `getValidToken` inside `run` must block the per-resource success stamp, // or the release is inert on the path that runs most often. - const imported = await syncWhoopResourceWithStatus("user-1", async () => { - noteHardFailure(WHOOP_TOKEN_HARD_FAIL); - return 0; - }); + const imported = await syncWhoopResourceWithStatus( + "user-1", + "recovery", + async () => { + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return 0; + }, + ); expect(imported).toBe(0); expect(recordSyncSuccessMock).not.toHaveBeenCalled(); }); it("stamps success on a clean per-resource run (no hard failure)", async () => { - const imported = await syncWhoopResourceWithStatus("user-1", async () => 3); + const imported = await syncWhoopResourceWithStatus( + "user-1", + "recovery", + async () => 3, + ); expect(imported).toBe(3); - expect(recordSyncSuccessMock).toHaveBeenCalledWith("user-1", "whoop"); + expect(recordSyncSuccessMock).toHaveBeenCalledWith("user-1", "whoop", { + leg: "recovery", + }); }); }); diff --git a/src/lib/whoop/__tests__/sync-user-whoop.test.ts b/src/lib/whoop/__tests__/sync-user-whoop.test.ts index 2c5fcfbdc..00110e072 100644 --- a/src/lib/whoop/__tests__/sync-user-whoop.test.ts +++ b/src/lib/whoop/__tests__/sync-user-whoop.test.ts @@ -177,24 +177,38 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { describe("syncWhoopResourceWithStatus — per-resource lastSuccessAt stamp", () => { it("stamps success when the resource imported rows", async () => { - const imported = await syncWhoopResourceWithStatus("user1", async () => 5); + const imported = await syncWhoopResourceWithStatus( + "user1", + "recovery", + async () => 5, + ); expect(imported).toBe(5); - expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop", { + leg: "recovery", + }); }); it("stamps success on a clean fetch that imported nothing new", async () => { // No 403 soft-skip → a genuine "nothing changed" webhook/cron tick still // proves the connection is alive, so `lastSuccessAt` advances. - const imported = await syncWhoopResourceWithStatus("user1", async () => 0); + const imported = await syncWhoopResourceWithStatus( + "user1", + "recovery", + async () => 0, + ); expect(imported).toBe(0); - expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop", { + leg: "recovery", + }); }); it("does NOT stamp success when the resource soft-skipped a 403 and imported nothing", async () => { - const imported = await syncWhoopResourceWithStatus("user1", () => - softSkip403("user1"), + const imported = await syncWhoopResourceWithStatus( + "user1", + "recovery", + () => softSkip403("user1"), ); expect(imported).toBe(0); @@ -204,7 +218,7 @@ describe("syncWhoopResourceWithStatus — per-resource lastSuccessAt stamp", () it("propagates a hard failure without stamping success", async () => { await expect( - syncWhoopResourceWithStatus("user1", async () => { + syncWhoopResourceWithStatus("user1", "recovery", async () => { throw new Error("boom"); }), ).rejects.toThrow("boom"); diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index dc54db379..2e2e41212 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -591,8 +591,11 @@ export async function markResourceSynced( * a genuine import clears the error state; a 403 soft-skip leaves the * "looks-healthy" window closed. */ +export type WhoopSyncLeg = "recovery" | "sleep" | "workout" | "cycle"; + export async function syncWhoopResourceWithStatus( userId: string, + leg: WhoopSyncLeg, run: () => Promise, ): Promise { // Run under BOTH trackers: soft-skip (a 403 scope gate) and hard-fail (a dead @@ -607,7 +610,10 @@ export async function syncWhoopResourceWithStatus( ); const softSkippedOnly = softSkipCount > 0 && imported === 0; if (!softSkippedOnly && hardFailTracker.failures.length === 0) { - await recordSyncSuccess(userId, "whoop"); + // Leg-scoped: WHOOP runs four resource crons against one ledger row, so a + // healthy recovery tick at :05 must not clear the error the workout tick + // recorded at :35 the hour before. + await recordSyncSuccess(userId, "whoop", { leg }); } return imported; } @@ -619,6 +625,7 @@ export async function syncWhoopResourceWithStatus( export async function recordWhoopSyncFailure( userId: string, err: unknown, + leg?: WhoopSyncLeg, ): Promise { const message = err instanceof Error ? err.message : String(err); await recordSyncFailure({ @@ -626,6 +633,7 @@ export async function recordWhoopSyncFailure( integration: "whoop", kind: classificationToFailureKind(classifyWhoopError(err)), message, + leg, errorCode: err instanceof WhoopApiError ? err.httpStatus?.toString() : undefined, }); diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index 0f9c2a80d..8c1ba17fe 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -370,7 +370,9 @@ describe("syncUserActivity — field mapping + idempotency", () => { } as never); await syncUserActivity("user-1"); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "activity", + }); }); it("W-2 — records a transient failure and refuses the success stamp when the batch write throws", async () => { diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index 92317bd21..5baca7a92 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -216,7 +216,9 @@ describe("syncUserEcg", () => { rhythmClassification: "IRREGULAR", externalId: "withings:ecg:user-1:111", }); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "ecg", + }); }); it("never narrows the broad catch-up to an ordinary webhook window", async () => { @@ -384,6 +386,7 @@ describe("syncUserEcg", () => { expect(recordWithingsSyncFailure).toHaveBeenCalledWith( "user-1", expect.objectContaining({ message: "source write failed" }), + "ecg", ); }); @@ -405,7 +408,11 @@ describe("syncUserEcg", () => { "IRREGULAR_RHYTHM_NOTIFICATION", new Date("2024-05-06T00:00:00.000Z"), ); - expect(recordWithingsSyncFailure).toHaveBeenCalledWith("user-1", poison); + expect(recordWithingsSyncFailure).toHaveBeenCalledWith( + "user-1", + poison, + "ecg", + ); expect(recordSyncSuccess).not.toHaveBeenCalled(); }); @@ -435,6 +442,8 @@ describe("syncUserEcg", () => { expect(imported).toBe(1); expect(prisma.measurement.upsert).toHaveBeenCalledTimes(1); expect(prisma.ecgRecording.upsert).not.toHaveBeenCalled(); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "ecg", + }); }); }); diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts index c81a9d4a6..c597f80b4 100644 --- a/src/lib/withings/__tests__/sync-measure-watermark.test.ts +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -201,7 +201,9 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () // still healthy: stamp the watermark + success, record no failure. expect(imported).toBe(0); expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "measures", + }); expect(recordSyncFailure).not.toHaveBeenCalled(); }); @@ -218,7 +220,9 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () expect(imported).toBe(1); expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "measures", + }); expect(recordSyncFailure).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index 98fcff22d..bf414bfe0 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -470,7 +470,9 @@ describe("syncUserSleep — segment writes + idempotency", () => { it("calls recordSyncSuccess after a clean round-trip", async () => { installFetchMock([]); await syncUserSleep("user-1"); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "sleep", + }); }); it("W-3 — records a transient failure and rethrows when a sleep write throws", async () => { @@ -580,7 +582,9 @@ describe("syncUserSleep — segment writes + idempotency", () => { // The one stage segment is still written; the summary failure is swallowed. expect(imported).toBe(1); expect(reconcileMock).toHaveBeenCalledTimes(1); - expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings", { + leg: "sleep", + }); }); it("tolerates a ScanWatch night with no REM segment (no all-zeros row)", async () => { diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index 05431f7ef..869fb1d18 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -50,7 +50,11 @@ import { WithingsApiError, classifyWithingsResponse, } from "./response-classifier"; -import { getValidToken, recordWithingsSyncFailure } from "./sync"; +import { + WITHINGS_LEG_ACTIVITY, + getValidToken, + recordWithingsSyncFailure, +} from "./sync"; import { isReauthRequired, parkIntegrationAtReauth, @@ -253,6 +257,7 @@ export async function syncUserActivity( message: "Withings connection is missing the user.activity scope. Reconnect Withings in Settings to enable activity sync.", errorCode: "scope_missing", + leg: WITHINGS_LEG_ACTIVITY, }); return 0; } @@ -303,10 +308,11 @@ export async function syncUserActivity( kind: "reauth_required", message: err instanceof Error ? err.message : String(err), errorCode: "403", + leg: WITHINGS_LEG_ACTIVITY, }); throw err; } - await recordWithingsSyncFailure(userId, err); + await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_ACTIVITY); throw err; } @@ -468,6 +474,7 @@ export async function syncUserActivity( message: `activity batch write failed: ${ err instanceof Error ? err.message : String(err) }`, + leg: WITHINGS_LEG_ACTIVITY, }); } } @@ -500,7 +507,9 @@ export async function syncUserActivity( // W-2 — only a clean write stamps success. A swallowed batch-write left the // ledger honest via the transient failure recorded above. if (!writeFailed) { - await recordSyncSuccess(userId, "withings"); + await recordSyncSuccess(userId, "withings", { + leg: WITHINGS_LEG_ACTIVITY, + }); } return imported; } diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 04c99bb1a..7f226913f 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -64,7 +64,11 @@ import { WithingsApiError, classifyWithingsResponse, } from "./response-classifier"; -import { getValidToken, recordWithingsSyncFailure } from "./sync"; +import { + WITHINGS_LEG_ECG, + getValidToken, + recordWithingsSyncFailure, +} from "./sync"; import { isReauthRequired, recordSyncFailure, @@ -408,10 +412,11 @@ export async function syncUserEcg( kind: "reauth_required", message: err instanceof Error ? err.message : String(err), errorCode: "403", + leg: WITHINGS_LEG_ECG, }); throw err; } - await recordWithingsSyncFailure(userId, err); + await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_ECG); throw err; } @@ -526,10 +531,14 @@ export async function syncUserEcg( } if (sourceWriteError) { - await recordWithingsSyncFailure(userId, sourceWriteError.cause); + await recordWithingsSyncFailure( + userId, + sourceWriteError.cause, + WITHINGS_LEG_ECG, + ); throw sourceWriteError.cause; } - await recordSyncSuccess(userId, "withings"); + await recordSyncSuccess(userId, "withings", { leg: WITHINGS_LEG_ECG }); return imported; } diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 37ba27262..840b5d8d3 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -64,7 +64,11 @@ import { WithingsApiError, classifyWithingsResponse, } from "./response-classifier"; -import { getValidToken, recordWithingsSyncFailure } from "./sync"; +import { + WITHINGS_LEG_SLEEP, + getValidToken, + recordWithingsSyncFailure, +} from "./sync"; import { isReauthRequired, parkIntegrationAtReauth, @@ -355,6 +359,7 @@ export async function syncUserSleep( message: "Withings connection is missing the user.activity scope. Reconnect Withings in Settings to enable sleep sync.", errorCode: "scope_missing", + leg: WITHINGS_LEG_SLEEP, }); return 0; } @@ -394,10 +399,11 @@ export async function syncUserSleep( kind: "reauth_required", message: err instanceof Error ? err.message : String(err), errorCode: "403", + leg: WITHINGS_LEG_SLEEP, }); throw err; } - await recordWithingsSyncFailure(userId, err); + await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_SLEEP); throw err; } @@ -429,6 +435,7 @@ export async function syncUserSleep( message: `sleep write failed: ${ err instanceof Error ? err.message : String(err) }`, + leg: WITHINGS_LEG_SLEEP, }); throw err; }; @@ -581,6 +588,6 @@ export async function syncUserSleep( () => {}, ); - await recordSyncSuccess(userId, "withings"); + await recordSyncSuccess(userId, "withings", { leg: WITHINGS_LEG_SLEEP }); return imported; } diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 5594b39f9..3da585814 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -156,7 +156,7 @@ export async function getValidToken( }, PROVIDER_REFRESH_TRANSACTION_OPTIONS); } catch (err) { getEvent()?.addWarning(`Token refresh failed for user ${userId}`); - await recordWithingsSyncFailure(userId, err); + await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES); if (options.throwOnRefreshFailure) throw err; return null; } @@ -226,7 +226,7 @@ export async function syncUserMeasurements( try { measures = await fetchMeasurements(tokenInfo.accessToken, startDate); } catch (err) { - await recordWithingsSyncFailure(userId, err); + await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES); throw err; } @@ -388,6 +388,7 @@ export async function syncUserMeasurements( await recordWithingsSyncFailure( userId, firstRowError ?? new Error("Withings measure row write failed"), + WITHINGS_LEG_MEASURES, ); return imported; } @@ -402,7 +403,9 @@ export async function syncUserMeasurements( // round-trip is the signal "the connection is healthy". A user who // doesn't weigh themselves for a week shouldn't see "error" on // their settings page. - await recordSyncSuccess(userId, "withings"); + await recordSyncSuccess(userId, "withings", { + leg: WITHINGS_LEG_MEASURES, + }); return imported; } @@ -423,9 +426,23 @@ export async function syncUserMeasurements( * `Error` instances (e.g. a pg-boss job retry that lost the prototype * during the JSON round-trip). */ +/** + * The four Withings sync legs. Each runs on its own cron at its own minute and + * writes the SAME `(user, withings)` ledger row, so a failure has to say which + * leg it came from — otherwise the next sibling to succeed clears an error it + * did not cause, which is what kept a broken leg invisible. + */ +export type WithingsSyncLeg = "measures" | "activity" | "sleep" | "ecg"; + +export const WITHINGS_LEG_MEASURES: WithingsSyncLeg = "measures"; +export const WITHINGS_LEG_ACTIVITY: WithingsSyncLeg = "activity"; +export const WITHINGS_LEG_SLEEP: WithingsSyncLeg = "sleep"; +export const WITHINGS_LEG_ECG: WithingsSyncLeg = "ecg"; + export async function recordWithingsSyncFailure( userId: string, err: unknown, + leg: WithingsSyncLeg, ): Promise { const message = err instanceof Error ? err.message : String(err); await recordSyncFailure({ @@ -433,6 +450,7 @@ export async function recordWithingsSyncFailure( integration: "withings", kind: classificationToFailureKind(classifyError(err)), message, + leg, errorCode: err instanceof WithingsApiError ? err.withingsStatus?.toString() diff --git a/tests/integration/integration-sync-escalation.test.ts b/tests/integration/integration-sync-escalation.test.ts new file mode 100644 index 000000000..fee900389 --- /dev/null +++ b/tests/integration/integration-sync-escalation.test.ts @@ -0,0 +1,358 @@ +/** + * Failure escalation and leg-scoped clearing on the integration ledger (real + * Postgres). + * + * Two defects are pinned here, and both are about state that only exists across + * several writes — a mocked Prisma cannot show either of them. + * + * 1. A `transient` streak had no exit. The park test read + * `kind === "persistent"` and the persistent anchor was only ever stamped + * by a persistent failure, so a streak the classifier had bucketed as + * transient could run indefinitely with the persistent counter at zero and + * the state frozen at `error_transient`. + * 2. `IntegrationStatus` is keyed `(userId, integration)` while Withings runs + * four legs on four crons and WHOOP four resources on four more. An + * unconditional success cleared the error, the buckets, the streak anchor + * and the state that a DIFFERENT leg had recorded, several times an hour, + * so the failing leg's strike ladder could never climb. + * + * The dispatcher is mocked at module level so the alert path resolves without + * dialling Telegram; the assertions are on ledger state, not the network. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/notifications/dispatcher", () => ({ + dispatchNotification: vi.fn().mockResolvedValue(undefined), +})); + +import { + getIntegrationStatus, + isReauthRequired, + recordSyncFailure, + recordSyncSuccess, +} from "@/lib/integrations/status"; + +const TEST_USER_ID = "user-sync-escalation"; +const HOUR_MS = 60 * 60 * 1000; + +function row(integration: string) { + return getPrismaClient().integrationStatus.findUnique({ + where: { userId_integration: { userId: TEST_USER_ID, integration } }, + }); +} + +/** + * Age the streak anchor backwards so a multi-day window can be exercised + * without a multi-day test. Only the anchor moves — every counter, the state + * and the error stay exactly as the production writes left them. + */ +async function ageStreakBy(integration: string, ms: number) { + const current = await row(integration); + const anchor = current!.failingSinceAt!; + await getPrismaClient().integrationStatus.update({ + where: { userId_integration: { userId: TEST_USER_ID, integration } }, + data: { failingSinceAt: new Date(anchor.getTime() - ms) }, + }); +} + +async function agePersistentAnchorBy(integration: string, ms: number) { + const current = await row(integration); + const anchor = current!.persistentFailureStartedAt!; + await getPrismaClient().integrationStatus.update({ + where: { userId_integration: { userId: TEST_USER_ID, integration } }, + data: { persistentFailureStartedAt: new Date(anchor.getTime() - ms) }, + }); +} + +async function failTransient(integration: string, leg?: string) { + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: integration as "withings", + kind: "transient", + message: `${integration} sync error: 503 - upstream unavailable`, + errorCode: "503", + leg, + }); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "sync-escalation", + email: "sync-escalation@example.test", + }, + }); +}); + +describe("failure escalation", () => { + it("escalates an unbroken transient streak to persistent once it outlives the transient window", async () => { + // Three hourly failures — a real burst, still inside the window. + await failTransient("withings"); + await failTransient("withings"); + await failTransient("withings"); + + let current = await row("withings"); + expect(current!.state).toBe("error_transient"); + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 3, + persistent: 0, + }); + expect(current!.persistentFailureStartedAt).toBeNull(); + expect(current!.failingSinceAt).toBeInstanceOf(Date); + + // The upstream stays broken for a full day. Nothing else changes. + await ageStreakBy("withings", 25 * HOUR_MS); + await failTransient("withings"); + + current = await row("withings"); + // The transient bucket is frozen where the window closed; the failure now + // counts against the persistent bucket the park ladder reads. + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 3, + persistent: 1, + }); + expect(current!.persistentFailureStartedAt).toBeInstanceOf(Date); + + const audits = await getPrismaClient().auditLog.findMany({ + where: { action: "integrations.sync.failed", userId: TEST_USER_ID }, + orderBy: { createdAt: "asc" }, + }); + const last = JSON.parse(audits[audits.length - 1]!.details ?? "{}"); + expect(last.kind).toBe("persistent"); + // The trail keeps the classifier's own verdict, so an operator can tell + // "the upstream said this was permanent" from "we concluded it was". + expect(last.escalatedFrom).toBe("transient"); + }); + + it("parks the escalated streak a further window later and stops attempting", async () => { + await failTransient("withings"); + await failTransient("withings"); + await failTransient("withings"); + await ageStreakBy("withings", 25 * HOUR_MS); + await failTransient("withings"); + + // Escalated, still retrying — parking on the same tick would turn one long + // upstream outage into a manual reconnect for every account. + expect((await row("withings"))!.state).toBe("error_transient"); + expect(await isReauthRequired(TEST_USER_ID, "withings")).toBe(false); + + // A second day of the same. + await agePersistentAnchorBy("withings", 25 * HOUR_MS); + await failTransient("withings"); + + const current = await row("withings"); + expect(current!.state).toBe("parked"); + expect(await isReauthRequired(TEST_USER_ID, "withings")).toBe(true); + + const parked = await getPrismaClient().auditLog.findMany({ + where: { action: "integrations.parked", userId: TEST_USER_ID }, + }); + expect(parked).toHaveLength(1); + }); + + it("does not escalate a short outage that recovers", async () => { + // Well past the wall-clock window, but the streak keeps being broken by a + // success — which is exactly what a flapping upstream looks like. + for (let cycle = 0; cycle < 4; cycle++) { + await failTransient("withings"); + await failTransient("withings"); + await ageStreakBy("withings", 48 * HOUR_MS); + await recordSyncSuccess(TEST_USER_ID, "withings"); + } + + const current = await row("withings"); + expect(current!.state).toBe("connected"); + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 0, + reauth_required: 0, + persistent: 0, + }); + expect(current!.persistentFailureStartedAt).toBeNull(); + expect(current!.failingSinceAt).toBeNull(); + expect(current!.lastError).toBeNull(); + }); + + it("does not escalate on wall-clock alone below the consecutive-failure threshold", async () => { + // A rarely-polled provider: one failure, then a day passes, then a second. + // Time is necessary but not sufficient — otherwise a provider that ticks + // once a day would escalate on its second-ever attempt. + await failTransient("nightscout"); + await ageStreakBy("nightscout", 25 * HOUR_MS); + await failTransient("nightscout"); + + const current = await row("nightscout"); + expect(current!.state).toBe("error_transient"); + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 2, + persistent: 0, + }); + expect(current!.persistentFailureStartedAt).toBeNull(); + }); + + it("parks a rejected credential at error_reauth on the first failure instead of retrying it", async () => { + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: "whoop", + kind: "reauth_required", + message: "WHOOP refresh error: 400 - invalid_grant", + errorCode: "invalid_grant", + leg: "recovery", + }); + + const current = await row("whoop"); + expect(current!.state).toBe("error_reauth"); + expect(current!.failingSinceAt).toBeInstanceOf(Date); + // The sync entry-points short-circuit — a revoked grant is not retried a + // thousand times. + expect(await isReauthRequired(TEST_USER_ID, "whoop")).toBe(true); + }); + + it("dates the streak from its first failure, not from the last retry", async () => { + await failTransient("withings"); + const firstAttempt = (await row("withings"))!.failingSinceAt!; + + await failTransient("withings"); + await failTransient("withings"); + + const current = await row("withings"); + expect(current!.failingSinceAt!.getTime()).toBe(firstAttempt.getTime()); + // `lastAttemptAt` keeps moving — that is the field that made a fortnight of + // silence read as "failing, just now". + expect(current!.lastAttemptAt!.getTime()).toBeGreaterThanOrEqual( + firstAttempt.getTime(), + ); + + const snapshot = await getIntegrationStatus(TEST_USER_ID, "withings"); + expect(snapshot.failingSince).toBe(firstAttempt.toISOString()); + }); +}); + +describe("leg-scoped success", () => { + it("does not let a sibling leg's success clear another leg's error", async () => { + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: "withings", + kind: "transient", + message: "sleep write failed: reconciliation conflict", + leg: "sleep", + }); + const afterFailure = await row("withings"); + const streakStart = afterFailure!.failingSinceAt!; + + // The ECG leg ticks 26 minutes later and is perfectly healthy. + await recordSyncSuccess(TEST_USER_ID, "withings", { leg: "ecg" }); + + const current = await row("withings"); + expect(current!.state).toBe("error_transient"); + expect(current!.lastError).toBe(afterFailure!.lastError); + expect(current!.consecutiveFailuresByKind).toMatchObject({ transient: 1 }); + expect(current!.failingLeg).toBe("sleep"); + expect(current!.failingSinceAt!.getTime()).toBe(streakStart.getTime()); + // Not advanced: it is what the card's "connected · X ago" reads, and the + // sleep pipe has delivered nothing. + expect(current!.lastSuccessAt).toBeNull(); + // The attempt did happen, so the row must not age into `stalled`. + expect(current!.lastAttemptAt).toBeInstanceOf(Date); + }); + + it("lets the owning leg's own success clear the row whole", async () => { + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: "withings", + kind: "transient", + message: "sleep write failed: reconciliation conflict", + leg: "sleep", + }); + await recordSyncSuccess(TEST_USER_ID, "withings", { leg: "ecg" }); + await recordSyncSuccess(TEST_USER_ID, "withings", { leg: "sleep" }); + + const current = await row("withings"); + expect(current!.state).toBe("connected"); + expect(current!.lastError).toBeNull(); + expect(current!.failingLeg).toBeNull(); + expect(current!.failingSinceAt).toBeNull(); + expect(current!.lastSuccessAt).toBeInstanceOf(Date); + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 0, + reauth_required: 0, + persistent: 0, + }); + }); + + it("keeps a leg's reauth park standing against a sibling leg's success", async () => { + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: "withings", + kind: "reauth_required", + message: "Withings sleep error: 403", + errorCode: "403", + leg: "sleep", + }); + expect(await isReauthRequired(TEST_USER_ID, "withings")).toBe(true); + + await recordSyncSuccess(TEST_USER_ID, "withings", { leg: "ecg" }); + + // Un-parking here is the worst shape of the bug: the sleep leg's grant is + // gone, and the row went back to claiming a healthy connection. + expect(await isReauthRequired(TEST_USER_ID, "withings")).toBe(true); + expect((await row("withings"))!.state).toBe("error_reauth"); + }); + + it("lets a sibling success be counted while the strike ladder still climbs", async () => { + // The regression, end to end: the failing leg fails hourly, a healthy leg + // succeeds in between, and the streak must still reach the escalation. + for (let hour = 0; hour < 3; hour++) { + await failTransient("withings", "sleep"); + await recordSyncSuccess(TEST_USER_ID, "withings", { leg: "ecg" }); + } + await ageStreakBy("withings", 25 * HOUR_MS); + await failTransient("withings", "sleep"); + + const current = await row("withings"); + expect(current!.consecutiveFailuresByKind).toMatchObject({ + transient: 3, + persistent: 1, + }); + expect(current!.persistentFailureStartedAt).toBeInstanceOf(Date); + }); + + it("clears the row whole when a full pass names no leg", async () => { + // A run that covered every resource is entitled to clear everything, and a + // single-leg provider that never names a leg keeps its old behaviour. + await recordSyncFailure({ + userId: TEST_USER_ID, + integration: "whoop", + kind: "transient", + message: "WHOOP workout error: 503", + leg: "workout", + }); + await recordSyncSuccess(TEST_USER_ID, "whoop"); + + const current = await row("whoop"); + expect(current!.state).toBe("connected"); + expect(current!.lastError).toBeNull(); + expect(current!.failingLeg).toBeNull(); + expect(current!.failingSinceAt).toBeNull(); + }); + + it("lets any success clear an unattributed failure", async () => { + await failTransient("oura"); + await recordSyncSuccess(TEST_USER_ID, "oura", { leg: "vitals" }); + + const current = await row("oura"); + expect(current!.state).toBe("connected"); + expect(current!.lastError).toBeNull(); + expect(current!.failingSinceAt).toBeNull(); + }); +});