From 689796db336c8c5cd66e542571ec94c11330d773 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 10:03:30 +0200 Subject: [PATCH] fix(integrations): log a failed sync as a signal, not an audit row The audit log is the trail of what a person or an admin did to an account. Every failed provider sync wrote a row there, burying that trail, and kept the provider's error message in plain text while the status row seals it. A failed sync now emits an integrations.sync.failed warning with the same counts and no message. --- docs/ops/attempt-number-semantics.md | 30 ++++++---- .../__tests__/admin-alert.test.ts | 11 +++- src/lib/integrations/__tests__/status.test.ts | 55 +++++++++++------- src/lib/integrations/status.ts | 58 +++++++++---------- tests/integration/integration-status.test.ts | 40 +++++++------ .../integration-sync-escalation.test.ts | 21 ++++--- 6 files changed, 125 insertions(+), 90 deletions(-) diff --git a/docs/ops/attempt-number-semantics.md b/docs/ops/attempt-number-semantics.md index b7b9ff5f8..9d6955b7a 100644 --- a/docs/ops/attempt-number-semantics.md +++ b/docs/ops/attempt-number-semantics.md @@ -1,18 +1,24 @@ -# `attemptNumber` audit-row vs alert-signal semantics +# `attempt_number` signal vs alert semantics Two numbers describe a failing integration sync, and they can diverge. -On-call operators reading an audit-log row alongside an admin alert -should expect the alert number to be smaller than (or equal to) the -audit number, never the other way around. +Operators reading an `integrations.sync.failed` log line alongside an admin +alert should expect the alert number to be smaller than (or equal to) the +logged number, never the other way around. + +Since v1.43.7 a failed sync is a `warn` signal in the log store, not an +audit-log row: the audit log records what a person or an admin did to an +account, and a provider answering 503 is neither. The provider's message is +not on the signal; it stays encrypted on the integration's status row and +shows in the integration's settings. ## The two numbers -`integrations.sync.failed` audit rows write -`attemptNumber = transient + reauth_required + persistent` — the **sum** +`integrations.sync.failed` signals carry +`attempt_number = transient + reauth_required + persistent` — the **sum** of all three failure buckets. This is the historical "consecutive failures" running total carried by the legacy `IntegrationStatus.consecutiveFailures` column (dropped in migration -`0077`, v1.4.47). Auditors looking at a single row see one integer +`0077`, v1.4.47). An operator looking at a single line sees one integer that captures how many failures the bucket ledger has accumulated since the last success. @@ -27,14 +33,14 @@ single-bucket threshold even if its sum looks alarming. ## When the two diverge A typed-failure-streak edge case: integration fails 2× transient, then -2× reauth_required, then 1× persistent. Audit row shows -`attemptNumber: 5`. Alert signal is `max(2, 2, 1) = 2`, which is below -the default threshold of 3, so no page fires. From the audit log this +2× reauth_required, then 1× persistent. The last signal shows +`attempt_number: 5`. The alert number is `max(2, 2, 1) = 2`, which is below +the default threshold of 3, so no page fires. From the log this looks like a sustained 5-failure outage; from the alert dispatcher it looks like three small bumps that never crossed a single-bucket ceiling. -Both readings are correct for their purpose. The audit log records what +Both readings are correct for their purpose. The signal records what happened end-to-end; the alert dispatcher only fires when a single failure mode persists, because mixed-bucket streaks usually indicate a transient network blip layered on top of an unrelated reauth prompt — @@ -43,7 +49,7 @@ we want to wake someone up for. ## Runbook implication -If you see an audit row with `attemptNumber >= threshold` but no alert +If you see a signal with `attempt_number >= threshold` but no alert was dispatched, check the per-bucket counts on `IntegrationStatus.consecutiveFailuresByKind` (DB column `consecutive_failures_by_kind`) for that user/integration pair. The diff --git a/src/lib/integrations/__tests__/admin-alert.test.ts b/src/lib/integrations/__tests__/admin-alert.test.ts index 77369ede9..fd38f1806 100644 --- a/src/lib/integrations/__tests__/admin-alert.test.ts +++ b/src/lib/integrations/__tests__/admin-alert.test.ts @@ -31,6 +31,11 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/logging/signal", () => ({ + emitSignal: vi.fn(), + logCaught: vi.fn(), +})); + vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn().mockResolvedValue(undefined), })); @@ -52,6 +57,7 @@ vi.mock("@/lib/crypto", () => ({ import { formatAdminAlertPayload, parkIntegrationAtReauth } from "../status"; import { prisma } from "@/lib/db"; import { auditLog } from "@/lib/auth/audit"; +import { emitSignal } from "@/lib/logging/signal"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; beforeEach(() => { @@ -246,9 +252,8 @@ describe("parkIntegrationAtReauth — silent scope-skip park (v1.4.27 F20)", () }), }), ); - expect(auditLog).not.toHaveBeenCalledWith( - "integrations.sync.failed", - expect.anything(), + expect(emitSignal).not.toHaveBeenCalledWith( + expect.objectContaining({ action: "integrations.sync.failed" }), ); }); diff --git a/src/lib/integrations/__tests__/status.test.ts b/src/lib/integrations/__tests__/status.test.ts index ed75617b0..c970ef923 100644 --- a/src/lib/integrations/__tests__/status.test.ts +++ b/src/lib/integrations/__tests__/status.test.ts @@ -36,6 +36,11 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/logging/signal", () => ({ + emitSignal: vi.fn(), + logCaught: vi.fn(), +})); + vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn().mockResolvedValue(undefined), })); @@ -66,6 +71,7 @@ import { } from "../status"; import { prisma } from "@/lib/db"; import { auditLog } from "@/lib/auth/audit"; +import { emitSignal } from "@/lib/logging/signal"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; beforeEach(() => { @@ -148,17 +154,24 @@ describe("recordSyncFailure — under threshold", () => { }); expect(prisma.integrationStatus.upsert).toHaveBeenCalledOnce(); - expect(auditLog).toHaveBeenCalledWith( - "integrations.sync.failed", + expect(emitSignal).toHaveBeenCalledWith( expect.objectContaining({ - userId: "u1", - details: expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ integration: "withings", - attemptNumber: 1, + attempt_number: 1, state: "error_transient", }), }), ); + // A provider failure is not something a person did to the account: no + // audit row, and the provider's message stays sealed on the status row. + expect(auditLog).not.toHaveBeenCalledWith( + "integrations.sync.failed", + expect.anything(), + ); + const [signal] = vi.mocked(emitSignal).mock.calls[0]!; + expect(JSON.stringify(signal)).not.toContain("upstream"); expect(dispatchNotification).not.toHaveBeenCalled(); }); }); @@ -456,12 +469,12 @@ describe("recordSyncFailure — reauth_required", () => { const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock .calls[0][0]; expect(upsertArgs.update).toMatchObject({ state: "error_reauth" }); - expect(auditLog).toHaveBeenCalledWith( - "integrations.sync.failed", + expect(emitSignal).toHaveBeenCalledWith( expect.objectContaining({ - details: expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ kind: "reauth_required", - errorCode: "100", + error_code: "100", }), }), ); @@ -489,12 +502,12 @@ describe("recordSyncFailure — persistent (v1.4.42 W6)", () => { const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock .calls[0][0]; expect(upsertArgs.update).toMatchObject({ state: "error_transient" }); - expect(auditLog).toHaveBeenCalledWith( - "integrations.sync.failed", + expect(emitSignal).toHaveBeenCalledWith( expect.objectContaining({ - details: expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ kind: "persistent", - errorCode: "293", + error_code: "293", }), }), ); @@ -1089,12 +1102,12 @@ describe("transient-streak escalation", () => { persistent: 1, }); expect(upsertUpdate().persistentFailureStartedAt).toBeInstanceOf(Date); - expect(auditLog).toHaveBeenCalledWith( - "integrations.sync.failed", + expect(emitSignal).toHaveBeenCalledWith( expect.objectContaining({ - details: expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ kind: "persistent", - escalatedFrom: "transient", + escalated_from: "transient", }), }), ); @@ -1116,12 +1129,12 @@ describe("transient-streak escalation", () => { persistent: 0, }); expect(upsertUpdate().persistentFailureStartedAt).toBeNull(); - expect(auditLog).toHaveBeenCalledWith( - "integrations.sync.failed", + expect(emitSignal).toHaveBeenCalledWith( expect.objectContaining({ - details: expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ kind: "transient", - escalatedFrom: null, + escalated_from: null, }), }), ); diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index ec6b0f7cc..5e507d7e5 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -35,6 +35,7 @@ import { prisma } from "@/lib/db"; import { encrypt, decrypt } from "@/lib/crypto"; import { auditLog } from "@/lib/auth/audit"; import { annotate, getEvent } from "@/lib/logging/context"; +import { emitSignal } from "@/lib/logging/signal"; import { dispatchNotification, type DispatchOutcome, @@ -558,7 +559,7 @@ export interface RecordSyncFailureInput { * Record a sync failure. Always: * - increments the per-kind bucket in `consecutiveFailuresByKind` * - persists the encrypted error message - * - writes one `AuditLog` row with `integrations.sync.failed` + * - emits one `integrations.sync.failed` signal (not an audit row) * * If `kind === "reauth_required"` the row is parked at `error_reauth` * so the next scheduled sync skips. If `kind === "persistent"` and @@ -746,39 +747,36 @@ export async function recordSyncFailure( errorCode: errorCode ?? null, }); - // Audit log entry — fire-and-await so an integration test can assert - // it. The auth/audit helper is its own DB write so it's safe to call - // serially without bloating latency in the success path (which never - // calls this). + // One signal per failed sync, not an audit row. The audit log is the + // trail of what a person or an admin did to an account; a provider that + // answered 503 is neither, and one row per failed sync drowned that trail + // (and kept the provider's message there in plain text, while the status + // row seals it). The signal carries the same counts to the log store, and + // the status row keeps the current streak and the sealed message. // - // v1.4.47 — `attemptNumber` is now sourced from the bucket sum - // (the legacy `consecutiveFailures` column was dropped). The sum - // matches the legacy column's value for any row written after - // v1.4.43: a transient burst followed by a single persistent failure - // shows `attemptNumber = transient + persistent`, which is the same - // running total the legacy integer carried. + // `attempt_number` is the sum of the buckets, the running total the + // dropped `consecutiveFailures` column carried; the alert below uses the + // largest single bucket (docs/ops/attempt-number-semantics.md). const bucketTotal = buckets.transient + buckets.reauth_required + buckets.persistent; - await auditLog("integrations.sync.failed", { - userId, - details: { + emitSignal({ + action: "integrations.sync.failed", + level: "warn", + meta: { integration, 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, - // Every leg the row currently holds an error for, not just this one — - // the audit trail is where an operator reconstructs an overlapping - // outage, and one leg per row was never enough to do it. - failingLegs, - failingSinceAt: failingSinceAt.toISOString(), - streakAgeMs, - errorCode: errorCode ?? null, - message, - attemptNumber: bucketTotal, - bucketCount: buckets[effectiveKind], + // Non-null only when the clock overrode the classifier: "the upstream + // said this was permanent" against "a day of retrying concluded it". + escalated_from: escalatedFromTransient ? kind : null, + failing_leg: leg, + // Every leg the row holds an error for, to reconstruct an overlapping + // outage. + failing_legs: failingLegs, + failing_since_at: failingSinceAt.toISOString(), + streak_age_ms: streakAgeMs, + error_code: errorCode ?? null, + attempt_number: bucketTotal, + bucket_count: buckets[effectiveKind], state: newState, }, }); @@ -941,7 +939,7 @@ export async function resumeIntegrationFromPark( * short-circuit. Unlike `recordSyncFailure`, this helper: * * 1. does NOT increment the per-kind failure buckets. - * 2. does NOT write an `integrations.sync.failed` audit row through + * 2. does NOT emit the `integrations.sync.failed` signal of * `recordSyncFailure`. A standalone `integrations.reauth_required` * row is written instead so the operations trail still shows the * park event. diff --git a/tests/integration/integration-status.test.ts b/tests/integration/integration-status.test.ts index 6a862fa1b..39198b593 100644 --- a/tests/integration/integration-status.test.ts +++ b/tests/integration/integration-status.test.ts @@ -3,13 +3,9 @@ * * The unit suite (`src/lib/integrations/__tests__/status.test.ts`) * mocks Prisma. This file lights up the real driver so we lock in the - * upsert + concurrent-write story AND prove that every sync failure - * writes one `AuditLog` row with the documented schema: - * - * action : "integrations.sync.failed" - * userId : - * details : { integration, kind, errorCode, message, - * attemptNumber, state } + * upsert + concurrent-write story, and that a sync failure leaves the audit + * log alone: it emits an `integrations.sync.failed` signal instead, and the + * provider's message stays sealed on the status row. * * The dispatcher is mocked at module level so we don't actually hit * Telegram while testing — the assertion is on the alerting side @@ -34,6 +30,11 @@ vi.mock("@/lib/db-compat", () => ({ // configs and call sendViaTelegram. We're not testing that path — // B3 owns it. Stub returns void so the helper's await chain // resolves cleanly. +vi.mock("@/lib/logging/signal", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, emitSignal: vi.fn(actual.emitSignal) }; +}); + vi.mock("@/lib/notifications/dispatcher", () => ({ dispatchNotification: vi.fn().mockResolvedValue({ dispatched: true, @@ -53,6 +54,7 @@ import { resumeIntegrationFromPark, } from "@/lib/integrations/status"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; +import { emitSignal } from "@/lib/logging/signal"; const TEST_USER_ID = "user-integration-status-e2e"; @@ -66,6 +68,7 @@ beforeEach(async () => { }, }); vi.mocked(dispatchNotification).mockClear(); + vi.mocked(emitSignal).mockClear(); }); describe("IntegrationStatus end-to-end", () => { @@ -103,16 +106,19 @@ describe("IntegrationStatus end-to-end", () => { const audits = await getPrismaClient().auditLog.findMany({ where: { action: "integrations.sync.failed", userId: TEST_USER_ID }, }); - expect(audits).toHaveLength(1); - const detailsParsed = JSON.parse(audits[0].details ?? "{}"); - expect(detailsParsed).toMatchObject({ - integration: "withings", - kind: "transient", - errorCode: "503", - message: "Withings refresh error: 503 - upstream", - attemptNumber: 1, - state: "error_transient", - }); + expect(audits).toHaveLength(0); + expect(emitSignal).toHaveBeenCalledWith( + expect.objectContaining({ + action: "integrations.sync.failed", + meta: expect.objectContaining({ + integration: "withings", + kind: "transient", + error_code: "503", + attempt_number: 1, + state: "error_transient", + }), + }), + ); }); it("getIntegrationStatus returns plaintext error after roundtrip (decryption succeeds)", async () => { diff --git a/tests/integration/integration-sync-escalation.test.ts b/tests/integration/integration-sync-escalation.test.ts index 2fc3f37f8..f39a66614 100644 --- a/tests/integration/integration-sync-escalation.test.ts +++ b/tests/integration/integration-sync-escalation.test.ts @@ -30,6 +30,11 @@ vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), })); +vi.mock("@/lib/logging/signal", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, emitSignal: vi.fn(actual.emitSignal) }; +}); + vi.mock("@/lib/notifications/dispatcher", () => ({ dispatchNotification: vi.fn().mockResolvedValue(undefined), })); @@ -40,6 +45,7 @@ import { recordSyncFailure, recordSyncSuccess, } from "@/lib/integrations/status"; +import { emitSignal } from "@/lib/logging/signal"; const TEST_USER_ID = "user-sync-escalation"; const HOUR_MS = 60 * 60 * 1000; @@ -85,6 +91,7 @@ async function failTransient(integration: string, leg?: string) { } beforeEach(async () => { + vi.mocked(emitSignal).mockClear(); await truncateAllTables(getPrismaClient()); await getPrismaClient().user.create({ data: { @@ -124,15 +131,15 @@ describe("failure escalation", () => { }); 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 ?? "{}"); + const signals = vi + .mocked(emitSignal) + .mock.calls.map(([signal]) => signal) + .filter((signal) => signal.action === "integrations.sync.failed"); + const last = signals[signals.length - 1]!.meta!; expect(last.kind).toBe("persistent"); - // The trail keeps the classifier's own verdict, so an operator can tell + // The signal 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"); + expect(last.escalated_from).toBe("transient"); }); it("parks the escalated streak a further window later and stops attempting", async () => {