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.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 10:03:30 +02:00
1 parent 537a73d3b2
commit 689796db33
6 files changed
+125 -90

No files matched your search

+18 -12
View File
@@ -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
@@ -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" }),
);
});
+34 -21
View File
@@ -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,
}),
}),
);
+28 -30
View File
@@ -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.
+23 -17
View File
@@ -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 : <subject>
* 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<typeof import("@/lib/logging/signal")>();
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 () => {
@@ -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<typeof import("@/lib/logging/signal")>();
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 () => {