diff --git a/src/lib/integrations/__tests__/watchdog-classify.test.ts b/src/lib/integrations/__tests__/watchdog-classify.test.ts index 9e9be7b9a..9de4196db 100644 --- a/src/lib/integrations/__tests__/watchdog-classify.test.ts +++ b/src/lib/integrations/__tests__/watchdog-classify.test.ts @@ -244,6 +244,43 @@ describe("currentIncident", () => { ).toEqual({ reason: "unexpected_silence", since: ago(30) }); }); + it("never opens an incident for stale or stalled data on its own, however long it lasts", () => { + // An Apple Health account whose phone app stopped running reads as + // `no_new_data` for good; nothing an operator can do about it. + for (const reason of ["no_new_data", "sync_stalled"] as const) { + expect( + currentIncident({ + health: health("degraded", reason, ago(24 * 90)), + silentSince: null, + ledgerAlerted: false, + now: NOW, + }), + ).toBeNull(); + } + }); + + it("opens an incident for provider errors that have lasted more than a day", () => { + expect( + currentIncident({ + health: health("degraded", "provider_errors", ago(25)), + silentSince: null, + ledgerAlerted: false, + now: NOW, + }), + ).toEqual({ reason: "provider_errors", since: ago(25) }); + }); + + it("still opens a silence incident beside stale data", () => { + expect( + currentIncident({ + health: health("degraded", "no_new_data", ago(50)), + silentSince: ago(30), + ledgerAlerted: false, + now: NOW, + }), + ).toEqual({ reason: "unexpected_silence", since: ago(30) }); + }); + it("has no incident for a limited connection", () => { expect( currentIncident({ diff --git a/src/lib/integrations/watchdog-classify.ts b/src/lib/integrations/watchdog-classify.ts index 0e0ab77f8..73f27aa2f 100644 --- a/src/lib/integrations/watchdog-classify.ts +++ b/src/lib/integrations/watchdog-classify.ts @@ -13,10 +13,14 @@ * is watched for errors only. * * 2. Is there an incident the operator should hear about? - * Unexpected silence; or a health state of `action_required` / - * `degraded` that has lasted longer than `INCIDENT_AFTER_MS`; or a - * failure streak the sync ledger already paged about (adopted, so the - * operator still gets the one recovery note when it ends). + * Unexpected silence; or an `action_required` state, or a `degraded` + * one because the provider keeps erroring (`provider_errors`), that has + * lasted longer than `INCIDENT_AFTER_MS`; or a failure streak the sync + * ledger already paged about (adopted, so the operator still gets the + * one recovery note when it ends). `no_new_data` and `sync_stalled` on + * their own never page: they mostly mean the owner stopped using the + * source (an Apple Health account whose phone app has not run for a week + * reads as stale for good), which is nothing an operator can fix. * * One incident per connection at a time. It is announced once and closed once; * a change of reason while it is open (silence turning into a reauth) is the @@ -162,7 +166,13 @@ export function currentIncident(input: { if (health === null) return null; const troubled = health.state === "action_required" || health.state === "degraded"; - if (troubled) { + // What pages on its own: the owner has to act, or the provider errors. A + // streak the ledger already paged about is followed whatever it reads as + // now, so its end is still reported. + const pages = + health.state === "action_required" || + (health.state === "degraded" && health.reason === "provider_errors"); + if (troubled && (pages || input.ledgerAlerted)) { // A provider-side reauth flag (Google Health) can stand without a ledger // streak to date it; the last success is then when it last worked. const sinceIso = health.since ?? health.lastSuccessAt; diff --git a/src/lib/integrations/watchdog.ts b/src/lib/integrations/watchdog.ts index 5fbe4f211..649050406 100644 --- a/src/lib/integrations/watchdog.ts +++ b/src/lib/integrations/watchdog.ts @@ -50,6 +50,13 @@ export interface WatchdogSummary { incidentsOpened: number; incidentsClosed: number; notified: number; + /** + * The pass found no watch row at all: the first one after the watchdog was + * deployed. Incidents standing at that moment are recorded as already + * announced instead of paged, so a deploy does not page once for every + * connection that has been broken or quiet for months. + */ + firstRun: boolean; } const DAY_MS = 24 * 60 * 60 * 1000; @@ -219,6 +226,7 @@ export async function runIntegrationWatchdog(opts: { incidentsOpened: 0, incidentsClosed: 0, notified: 0, + firstRun: (await prisma.integrationWatch.count()) === 0, }; const userIds = await listWatchedUserIds(); for (const userId of userIds) { @@ -357,7 +365,13 @@ async function watchConnection(input: { summary.incidentsOpened += 1; incidentReason = transition.reason; incidentOpenedAt = now; - await announce(transition.reason, transition.since); + if (summary.firstRun) { + // Standing since before the watchdog first looked: recorded as + // announced, not paged (see `WatchdogSummary.firstRun`). + alertedAt = now; + } else { + await announce(transition.reason, transition.since); + } break; case "adopt": // The ledger already paged about this streak; record it as announced so @@ -405,8 +419,17 @@ async function watchConnection(input: { break; } - // Nothing to remember and nothing remembered: no row. - if (row === null && silentSince === null && incidentReason === null) return; + // Nothing to remember and nothing remembered: no row. Except on the first + // pass, which leaves a row for every connection it looked at, so the pass + // after it is not taken for a first one too. + if ( + row === null && + silentSince === null && + incidentReason === null && + !(summary.firstRun && fact.connected) + ) { + return; + } await prisma.integrationWatch.upsert({ where: { userId_integration: { userId, integration: key } }, diff --git a/src/lib/jobs/integration-watchdog.ts b/src/lib/jobs/integration-watchdog.ts index ce6373037..f4ce883e8 100644 --- a/src/lib/jobs/integration-watchdog.ts +++ b/src/lib/jobs/integration-watchdog.ts @@ -57,6 +57,7 @@ function handlerFor(mode: WatchdogMode) { incidents_opened: s.incidentsOpened, incidents_closed: s.incidentsClosed, notified: s.notified, + first_run: s.firstRun ? 1 : 0, }); } catch (err) { evt.addWarning(`integration watchdog failed: ${err}`); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index f62c00e65..b09cb08e8 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -113,6 +113,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "fetches", "finalised", "finalized", + // v1.42.4 — integration watchdog: the pass found no watch row (first run). + "first_run", "forced", "generated", "geo_backfill_carrier_resolved", diff --git a/tests/integration/integration-watchdog.test.ts b/tests/integration/integration-watchdog.test.ts index 44de37b40..0ef329cc8 100644 --- a/tests/integration/integration-watchdog.test.ts +++ b/tests/integration/integration-watchdog.test.ts @@ -92,6 +92,11 @@ beforeEach(async () => { }, }); vi.mocked(dispatchNotification).mockClear(); + // The watchdog has run on this instance before: its first pass after a + // deploy records what stands without paging (pinned on its own below). + await prisma().integrationWatch.create({ + data: { userId: ADMIN, integration: "withings" }, + }); }); describe("integration watchdog", () => { @@ -176,7 +181,9 @@ describe("integration watchdog", () => { const summary = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); expect(summary.silencesFlagged).toBe(0); expect(dispatchNotification).not.toHaveBeenCalled(); - expect(await prisma().integrationWatch.count()).toBe(0); + expect( + await prisma().integrationWatch.count({ where: { userId: USER } }), + ).toBe(0); }); it("never alerts on silence for an on-demand source such as a scale", async () => { @@ -245,6 +252,115 @@ describe("integration watchdog", () => { }); }); +describe("what pages and what does not", () => { + it("never pages for an Apple Health account whose phone app stopped delivering weeks ago", async () => { + // Daily data until forty days ago, then nothing: the push cadence reads + // stale (`no_new_data`) for good. The owner stopped using the app; there + // is nothing for an operator to do. + await prisma().user.update({ + where: { id: USER }, + data: { + ouraAccessTokenEncrypted: null, + healthKitLastSyncedAt: new Date(NOW.getTime() - 40 * DAY), + }, + }); + await prisma().measurement.createMany({ + data: Array.from({ length: 30 }, (_, i) => ({ + userId: USER, + type: "ACTIVITY_STEPS" as const, + value: 5000, + unit: "count", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(NOW.getTime() - (40 + i) * DAY), + })), + }); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + try { + const health = await loadConnectionHealth(USER, ["apple-health"]); + expect(health["apple-health"]?.state).toBe("degraded"); + expect(health["apple-health"]?.reason).toBe("no_new_data"); + // A week of hourly and daily passes. + for (let day = 0; day < 7; day++) { + const at = new Date(NOW.getTime() + day * DAY); + const hourly = await runIntegrationWatchdog({ + mode: "hourly", + now: at, + }); + const daily = await runIntegrationWatchdog({ mode: "daily", now: at }); + expect(hourly.incidentsOpened + daily.incidentsOpened).toBe(0); + } + expect(dispatchNotification).not.toHaveBeenCalled(); + expect( + await prisma().integrationWatch.count({ + where: { userId: USER, incidentReason: { not: null } }, + }), + ).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("records what stands on its first pass after a deploy without paging, and pages normally after", async () => { + await prisma().integrationWatch.deleteMany(); + // A wearable that has been silent since before the watchdog existed. + await seedDailyReadings( + USER, + "OURA", + Array.from({ length: 13 }, (_, i) => i + 3), + 54, + ); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + try { + await recordSyncSuccess(USER, "oura"); + const first = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); + expect(first.firstRun).toBe(true); + expect(first.incidentsOpened).toBe(1); + expect(dispatchNotification).not.toHaveBeenCalled(); + const watch = await prisma().integrationWatch.findUniqueOrThrow({ + where: { userId_integration: { userId: USER, integration: "oura" } }, + }); + expect(watch.incidentReason).toBe("unexpected_silence"); + // Marked as announced, so no later pass pages it either. + expect(watch.alertedAt).not.toBeNull(); + + const later = new Date(NOW.getTime() + HOUR); + const second = await runIntegrationWatchdog({ + mode: "hourly", + now: later, + }); + expect(second.firstRun).toBe(false); + expect(dispatchNotification).not.toHaveBeenCalled(); + + // A new incident after the first pass pages as usual. + for (let i = 0; i < 3; i++) { + await recordSyncFailure({ + userId: USER, + integration: "whoop", + kind: "transient", + message: "WHOOP error: 503", + errorCode: "503", + }); + } + expect(dispatchNotification).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it("leaves a row behind on a first pass that found nothing wrong, so the next pass is not a first one", async () => { + await prisma().integrationWatch.deleteMany(); + await seedDailyReadings(USER, "OURA", [1, 2, 3], 2); + await recordSyncSuccess(USER, "oura"); + const first = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); + expect(first.firstRun).toBe(true); + expect(await prisma().integrationWatch.count()).toBe(1); + const second = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); + expect(second.firstRun).toBe(false); + }); +}); + describe("recent-call ring", () => { it("keeps the newest ten attempts, newest first, without any message text", async () => { for (let i = 0; i < 7; i++) {