From aff42fe25ec0faa90eef4a1ba543e3c028502304 Mon Sep 17 00:00:00 2001 From: Warren Lee <5959690+wrn14897@users.noreply.github.com> Date: Fri, 13 Nov 2026 21:14:09 -0800 Subject: [PATCH] fix(api): re-send resolve notifications whose delivery failed A resolve is sent once, when an alert recovers. If that webhook delivery failed, the next evaluation saw the alert as OK and never sent it again, so the incident stayed open in the receiving tool. A failed resolve is now stored in a new PendingNotification collection and sent again before the alert's evaluation on later ticks, with the same Idempotency-Key, backing off from 1 to 30 minutes. It is dropped when the group fires again or its webhook is deleted, held while the alert is silenced, and after 24 hours recorded as a WEBHOOK_ERROR on the resolved window's evaluation history. Delivery failure logs now include the error's cause chain, so "fetch failed" shows the underlying DNS, connect or TLS error. Ref: HDX-6020 --- .changeset/resend-failed-alert-resolves.md | 8 + .../api/src/models/pendingNotification.ts | 57 +++ .../__tests__/checkAlerts.int.test.ts | 5 + .../__tests__/checkAlertsTask.test.ts | 89 +++++ .../pendingNotifications.int.test.ts | 327 ++++++++++++++++++ .../__tests__/pendingNotifications.test.ts | 16 + packages/api/src/tasks/checkAlerts/index.ts | 93 ++++- .../tasks/checkAlerts/pendingNotifications.ts | 229 ++++++++++++ .../tasks/checkAlerts/providers/default.ts | 58 ++++ .../src/tasks/checkAlerts/providers/index.ts | 26 ++ .../api/src/tasks/checkAlerts/template.ts | 6 +- .../tasks/checkAlerts/transports/generic.ts | 4 +- .../api/src/utils/__tests__/errors.test.ts | 56 ++- packages/api/src/utils/errors.ts | 29 ++ packages/api/src/utils/retry.ts | 5 +- 15 files changed, 1000 insertions(+), 8 deletions(-) create mode 100644 .changeset/resend-failed-alert-resolves.md create mode 100644 packages/api/src/models/pendingNotification.ts create mode 100644 packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts create mode 100644 packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts create mode 100644 packages/api/src/tasks/checkAlerts/pendingNotifications.ts diff --git a/.changeset/resend-failed-alert-resolves.md b/.changeset/resend-failed-alert-resolves.md new file mode 100644 index 0000000000..cc40f56d74 --- /dev/null +++ b/.changeset/resend-failed-alert-resolves.md @@ -0,0 +1,8 @@ +--- +'@hyperdx/api': patch +--- + +fix: a resolve notification whose webhook delivery fails is now sent again on +later checks, backing off up to every 30 minutes, for up to 24 hours. A resolve +that is never delivered is recorded on the alert's evaluation history. Webhook +delivery failures now log the underlying cause instead of only "fetch failed". diff --git a/packages/api/src/models/pendingNotification.ts b/packages/api/src/models/pendingNotification.ts new file mode 100644 index 0000000000..c23a4b24df --- /dev/null +++ b/packages/api/src/models/pendingNotification.ts @@ -0,0 +1,57 @@ +import mongoose, { Schema } from 'mongoose'; + +import type { ObjectId } from '.'; + +/** + * An alert notification whose delivery failed and that the check-alerts task + * sends again on later ticks. Only resolves are stored: a firing alert sends + * ALERT again on every evaluation, but a resolve is sent once. + */ +export interface IPendingNotification { + /** Idempotency key of the notification: hash of alert, channel and group. */ + eventId: string; + team: ObjectId; + alert: ObjectId; + /** Group key of a group-by alert; '' for an alert without groups. */ + group: string; + webhook: ObjectId; + /** + * The rendered message as JSON. A string because group-by column names can + * contain dots, which Mongo would read as nested paths in a key. + */ + message: string; + /** Evaluation window whose ERROR history row records the failure. */ + windowStart: Date; + firstFailedAt: Date; + /** Failed deliveries so far, including the original one. */ + attempts: number; + nextAttemptAt: Date; + /** Removes records nothing ever cleans up, e.g. for a deleted alert. */ + expireAt: Date; +} + +const PendingNotificationSchema = new Schema( + { + eventId: { type: String, required: true }, + team: { type: Schema.Types.ObjectId, ref: 'Team', required: true }, + alert: { type: Schema.Types.ObjectId, ref: 'Alert', required: true }, + group: { type: String, default: '' }, + webhook: { type: Schema.Types.ObjectId, ref: 'Webhook', required: true }, + message: { type: String, required: true }, + windowStart: { type: Date, required: true }, + firstFailedAt: { type: Date, required: true }, + attempts: { type: Number, required: true }, + nextAttemptAt: { type: Date, required: true }, + expireAt: { type: Date, required: true }, + }, + { timestamps: true }, +); + +PendingNotificationSchema.index({ eventId: 1 }, { unique: true }); +PendingNotificationSchema.index({ alert: 1 }); +PendingNotificationSchema.index({ expireAt: 1 }, { expireAfterSeconds: 0 }); + +export default mongoose.model( + 'PendingNotification', + PendingNotificationSchema, +); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts index d3d0bbdc3b..1105ca65f3 100644 --- a/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts +++ b/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts @@ -31,6 +31,7 @@ import Alert, { AlertSource, IAlert } from '@/models/alert'; import AlertHistory from '@/models/alertHistory'; import Connection, { IConnection } from '@/models/connection'; import Dashboard, { IDashboard } from '@/models/dashboard'; +import PendingNotification from '@/models/pendingNotification'; import { ISavedSearch, SavedSearch } from '@/models/savedSearch'; import { ISource, LogSource, Source } from '@/models/source'; import { ITeam } from '@/models/team'; @@ -4483,6 +4484,10 @@ describe('checkAlerts', () => { expect(updated!.executionErrors![0].type).toBe( AlertErrorType.WEBHOOK_ERROR, ); + // Stored so a later tick sends the resolve again. + expect( + await PendingNotification.countDocuments({ alert: details.alert.id }), + ).toBe(1); }); it('clears errors after a successful execution', async () => { diff --git a/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts index 5b5d348898..46eb4793da 100644 --- a/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts +++ b/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts @@ -4,10 +4,12 @@ import mongoose from 'mongoose'; import * as config from '@/config'; import { ObjectId } from '@/models'; import { AlertSource, AlertThresholdType, IAlert } from '@/models/alert'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; import * as checkAlerts from '@/tasks/checkAlerts'; import CheckAlertTask from '@/tasks/checkAlerts'; +import * as pendingNotifications from '@/tasks/checkAlerts/pendingNotifications'; import { AlertDetails, AlertProvider, @@ -38,6 +40,11 @@ describe('CheckAlertTask', () => { getWebhooks: jest.fn(), updateAlertState: jest.fn(), recordAlertErrors: jest.fn(), + appendAlertErrorHistory: jest.fn(), + getPendingNotifications: jest.fn().mockResolvedValue([]), + savePendingNotifications: jest.fn(), + updatePendingNotification: jest.fn(), + deletePendingNotification: jest.fn(), asyncDispose: jest.fn(), buildChartLink: jest.fn(), buildChartExplorerLink: jest.fn(), @@ -173,6 +180,88 @@ describe('CheckAlertTask', () => { mockProcessAlert.mockRestore(); }); + const makeTaskForAlert = (alertId: string): AlertTask => ({ + alerts: [ + { + alert: { id: alertId } as IAlert, + taskType: AlertTaskType.SAVED_SEARCH, + previousMap: new Map(), + } as AlertDetails, + ], + conn: { + id: 'conn-123', + _id: new mongoose.Types.ObjectId(), + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + name: '', + team: new mongoose.Types.ObjectId(), + }, + now: new Date(), + }); + + it("sends an alert's pending notifications again before evaluating it", async () => { + const task = new CheckAlertTask({ taskName: TaskName.CHECK_ALERTS }); + const alertId = new mongoose.Types.ObjectId().toString(); + const alertTask = makeTaskForAlert(alertId); + const record: IPendingNotification = { + eventId: 'event-1', + team: alertTask.conn.team, + alert: new mongoose.Types.ObjectId(alertId), + group: '', + webhook: new mongoose.Types.ObjectId(), + message: '{}', + windowStart: alertTask.now, + firstFailedAt: alertTask.now, + attempts: 1, + nextAttemptAt: alertTask.now, + expireAt: alertTask.now, + }; + mockAlertProvider.getAlertTasks.mockResolvedValue([alertTask]); + mockAlertProvider.getWebhooks.mockResolvedValue(new Map()); + mockAlertProvider.getPendingNotifications.mockResolvedValue([record]); + + const calls: string[] = []; + const resendSpy = jest + .spyOn(pendingNotifications, 'resendPendingNotifications') + .mockImplementation(async () => { + calls.push('resend'); + }); + mockProcessAlert.mockImplementation(async () => { + calls.push('process'); + }); + + await task.execute(); + + expect(mockAlertProvider.getPendingNotifications).toHaveBeenCalledWith([ + alertId, + ]); + expect(resendSpy).toHaveBeenCalledWith( + expect.objectContaining({ + alert: alertTask.alerts[0].alert, + pending: [record], + now: alertTask.now, + }), + ); + expect(calls).toEqual(['resend', 'process']); + resendSpy.mockRestore(); + }); + + it('still evaluates alerts when pending notifications cannot be loaded', async () => { + const task = new CheckAlertTask({ taskName: TaskName.CHECK_ALERTS }); + mockAlertProvider.getAlertTasks.mockResolvedValue([ + makeTaskForAlert('alert-123'), + ]); + mockAlertProvider.getWebhooks.mockResolvedValue(new Map()); + mockAlertProvider.getPendingNotifications.mockRejectedValue( + new Error('mongo down'), + ); + + await task.execute(); + + expect(mockProcessAlert).toHaveBeenCalledTimes(1); + }); + it("should ensure that the correct team's webhooks are passed to processAlert", async () => { const args: CheckAlertsTaskArgs = { taskName: TaskName.CHECK_ALERTS }; const task = new CheckAlertTask(args); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts new file mode 100644 index 0000000000..e5cee24d9a --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts @@ -0,0 +1,327 @@ +import { ClickhouseClient } from '@hyperdx/common-utils/dist/clickhouse/node'; +import { AlertErrorType } from '@hyperdx/common-utils/dist/types'; +import mongoose from 'mongoose'; +import ms from 'ms'; + +import { createAlert } from '@/controllers/alerts'; +import { createTeam } from '@/controllers/team'; +import { bulkInsertLogs, getServer } from '@/fixtures'; +import Alert, { + AlertSource, + AlertState, + AlertThresholdType, +} from '@/models/alert'; +import AlertHistory from '@/models/alertHistory'; +import PendingNotification from '@/models/pendingNotification'; +import Webhook, { IWebhook } from '@/models/webhook'; +import { getPreviousAlertHistories, processAlert } from '@/tasks/checkAlerts'; +import { resendPendingNotifications } from '@/tasks/checkAlerts/pendingNotifications'; +import { + AlertProvider, + AlertTaskType, + loadProvider, +} from '@/tasks/checkAlerts/providers'; + +import { + HOOK_BAD, + HOOK_OK, + makeGenericWebhook, + setupAlertFixtures, +} from './multiChannelAlerts.harness'; + +const FIRING_AT = new Date('2023-11-16T22:12:00.000Z'); +const RESOLVING_AT = new Date('2023-11-16T22:17:00.000Z'); +// The 5m window RESOLVING_AT evaluates, where its history rows are written. +const RESOLVED_WINDOW = new Date('2023-11-16T22:15:00.000Z'); +const minutesAfterResolve = (minutes: number) => + new Date(RESOLVING_AT.getTime() + ms(`${minutes}m`)); + +type Request = { url: string; idempotencyKey?: string; body: string }; + +describe('sending failed resolves again', () => { + let alertProvider: AlertProvider; + let server: ReturnType; + let requests: Request[]; + // Endpoints that currently fail every request. + let down: Set; + + beforeAll(async () => { + alertProvider = await loadProvider(); + server = getServer(); + await server.start(); + }); + + beforeEach(() => { + requests = []; + down = new Set(); + jest.mocked(global.fetch).mockImplementation(async (input, init) => { + const url = String(input); + requests.push({ + url, + idempotencyKey: + new Headers(init?.headers).get('Idempotency-Key') ?? undefined, + body: String(init?.body), + }); + if (down.has(url)) { + // 400 so withRetry does not retry within one delivery. + return new Response('unavailable', { status: 400 }); + } + return new Response('ok', { status: 200 }); + }); + }); + + afterEach(async () => { + await server.clearDBs(); + jest.clearAllMocks(); + }); + + afterAll(async () => { + await server.stop(); + }); + + const setup = async ( + webhookUrls: string[] = [HOOK_OK], + ): Promise<{ + alertId: string; + webhooks: IWebhook[]; + evaluate: (now: Date) => Promise; + resend: (now: Date, webhooks?: IWebhook[]) => Promise; + }> => { + const { team, connection, source, savedSearch } = + await setupAlertFixtures(createTeam); + const webhooks = await Promise.all( + webhookUrls.map((url, i) => + makeGenericWebhook(team._id, `Hook ${i}`, url), + ), + ); + const alert = await createAlert( + team._id, + { + source: AlertSource.SAVED_SEARCH, + channels: webhooks.map(w => ({ + type: 'webhook' as const, + webhookId: w._id.toString(), + })), + interval: '5m', + thresholdType: AlertThresholdType.ABOVE, + threshold: 1, + savedSearchId: savedSearch.id, + name: 'Resend Alert', + }, + new mongoose.Types.ObjectId(), + ); + const clickhouseClient = new ClickhouseClient({ + host: connection.host, + username: connection.username, + password: connection.password, + }); + const webhooksById = (list: IWebhook[]) => + new Map(list.map(w => [w._id.toString(), w])); + + const evaluate = async (now: Date) => { + const enhancedAlert: any = await Alert.findById(alert.id).populate([ + 'team', + 'savedSearch', + ]); + await processAlert( + now, + { + alert: enhancedAlert, + source, + taskType: AlertTaskType.SAVED_SEARCH, + savedSearch, + previousMap: await getPreviousAlertHistories([alert.id], now), + }, + clickhouseClient, + connection.id, + alertProvider, + webhooksById(webhooks), + ); + }; + + const resend = async (now: Date, available: IWebhook[] = webhooks) => { + const previousMap = await getPreviousAlertHistories([alert.id], now); + await resendPendingNotifications({ + alert: (await Alert.findById(alert.id))!, + // Non-grouped alerts key their history by the bare alert id. + getPreviousState: () => previousMap.get(alert.id)?.state, + now, + pending: await alertProvider.getPendingNotifications([alert.id]), + provider: alertProvider, + teamWebhooksById: webhooksById(available), + }); + }; + + return { alertId: alert.id, webhooks, evaluate, resend }; + }; + + const seedErrorLogs = (at: Date) => + bulkInsertLogs( + [0, 1].map(() => ({ + ServiceName: 'api', + Timestamp: at, + SeverityText: 'error', + Body: 'Test error message', + })), + ); + + // Fires at FIRING_AT, then resolves at RESOLVING_AT while `failing` is down. + const fireThenFailResolve = async ( + ctx: Awaited>, + failing: string[] = [HOOK_OK], + ) => { + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + await ctx.evaluate(FIRING_AT); + failing.forEach(url => down.add(url)); + await ctx.evaluate(RESOLVING_AT); + requests = []; + }; + + it('stores a resolve whose delivery failed', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + const alert = await Alert.findById(ctx.alertId); + expect(alert!.state).toBe(AlertState.OK); + expect(alert!.executionErrors?.[0]?.type).toBe( + AlertErrorType.WEBHOOK_ERROR, + ); + const records = await PendingNotification.find({ alert: ctx.alertId }); + expect(records).toHaveLength(1); + expect(records[0]).toMatchObject({ + group: '', + attempts: 1, + windowStart: RESOLVED_WINDOW, + firstFailedAt: RESOLVING_AT, + nextAttemptAt: minutesAfterResolve(1), + }); + expect(records[0].webhook.toString()).toBe(ctx.webhooks[0]._id.toString()); + expect(JSON.parse(records[0].message).state).toBe(AlertState.OK); + }); + + it('delivers the resolve on a later tick with the same idempotency key', async () => { + const ctx = await setup(); + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + await ctx.evaluate(FIRING_AT); + down.add(HOOK_OK); + await ctx.evaluate(RESOLVING_AT); + const failedResolve = requests[requests.length - 1]; + + requests = []; + down.clear(); + await ctx.resend(minutesAfterResolve(1)); + + expect(requests).toHaveLength(1); + expect(requests[0].url).toBe(HOOK_OK); + expect(requests[0].idempotencyKey).toBe(failedResolve.idempotencyKey); + expect(requests[0].body).toBe(failedResolve.body); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('backs off after another failure', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + await ctx.resend(minutesAfterResolve(1)); + expect(requests).toHaveLength(1); + const record = await PendingNotification.findOne({ alert: ctx.alertId }); + expect(record).toMatchObject({ + attempts: 2, + nextAttemptAt: minutesAfterResolve(3), + }); + + requests = []; + await ctx.resend(minutesAfterResolve(2)); + expect(requests).toHaveLength(0); + }); + + it('only stores the channels that failed', async () => { + const ctx = await setup([HOOK_OK, HOOK_BAD]); + await fireThenFailResolve(ctx, [HOOK_BAD]); + + const records = await PendingNotification.find({ alert: ctx.alertId }); + expect(records.map(r => r.webhook.toString())).toEqual([ + ctx.webhooks[1]._id.toString(), + ]); + + await ctx.resend(minutesAfterResolve(1)); + expect(requests.map(r => r.url)).toEqual([HOOK_BAD]); + }); + + it('does not store a failed ALERT, which the next evaluation sends again', async () => { + const ctx = await setup(); + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + down.add(HOOK_OK); + await ctx.evaluate(FIRING_AT); + + expect((await Alert.findById(ctx.alertId))!.state).toBe(AlertState.ALERT); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('drops the resolve once the group fires again', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + down.clear(); + await seedErrorLogs(minutesAfterResolve(1)); + await ctx.evaluate(minutesAfterResolve(5)); + requests = []; + + await ctx.resend(minutesAfterResolve(6)); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('records an error on the resolved window and stops after 24 hours', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + await ctx.resend(new Date(RESOLVING_AT.getTime() + ms('24h'))); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + const errorRow = await AlertHistory.findOne({ + alert: ctx.alertId, + createdAt: RESOLVED_WINDOW, + state: AlertState.ERROR, + }); + expect(errorRow!.errors).toHaveLength(2); + expect(errorRow!.errors![1]).toMatchObject({ + type: AlertErrorType.WEBHOOK_ERROR, + message: expect.stringContaining( + 'Resolve notification to webhook "Hook 0" was not delivered within 1 day (1 attempt)', + ), + }); + }); + + it('drops the resolve when its webhook was deleted', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + await Webhook.deleteOne({ _id: ctx.webhooks[0]._id }); + + await ctx.resend(minutesAfterResolve(1), []); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('holds the resolve while the alert is silenced', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + await Alert.updateOne( + { _id: ctx.alertId }, + { + $set: { + silenced: { at: RESOLVING_AT, until: minutesAfterResolve(60) }, + }, + }, + ); + + await ctx.resend(minutesAfterResolve(1)); + + expect(requests).toHaveLength(0); + expect( + await PendingNotification.findOne({ alert: ctx.alertId }), + ).toMatchObject({ attempts: 1 }); + }); +}); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts new file mode 100644 index 0000000000..245ce426c3 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts @@ -0,0 +1,16 @@ +import { getNextAttemptAt } from '@/tasks/checkAlerts/pendingNotifications'; + +describe('getNextAttemptAt', () => { + const now = new Date('2026-11-13T12:00:00.000Z'); + const minutesAfterNow = (attempts: number) => + (getNextAttemptAt(now, attempts).getTime() - now.getTime()) / 60_000; + + it('waits 1 minute after the first failure and doubles after each one', () => { + expect([1, 2, 3, 4, 5].map(minutesAfterNow)).toEqual([1, 2, 4, 8, 16]); + }); + + it('waits at most 30 minutes', () => { + expect(minutesAfterNow(6)).toBe(30); + expect(minutesAfterNow(50)).toBe(30); + }); +}); diff --git a/packages/api/src/tasks/checkAlerts/index.ts b/packages/api/src/tasks/checkAlerts/index.ts index d34af3dd85..0633918b8a 100644 --- a/packages/api/src/tasks/checkAlerts/index.ts +++ b/packages/api/src/tasks/checkAlerts/index.ts @@ -67,6 +67,7 @@ import AlertHistory, { IAlertHistoryAnalytics, } from '@/models/alertHistory'; import { IDashboard } from '@/models/dashboard'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISavedSearch } from '@/models/savedSearch'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; @@ -85,6 +86,10 @@ import { findAlertEvaluator, THRESHOLD_EVALUATOR_NAME, } from '@/tasks/checkAlerts/evaluators'; +import { + resendPendingNotifications, + toPendingNotification, +} from '@/tasks/checkAlerts/pendingNotifications'; import { AlertDetails, AlertProvider, @@ -110,6 +115,7 @@ import { unflattenObject, } from '@/tasks/util'; import { alertConfigHasGroupBy, isPopulatedRef } from '@/utils/alerts'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getActiveTraceId, getCounter, @@ -1286,6 +1292,35 @@ export const processAlert = async ( } : undefined; + // A firing alert sends ALERT again on every evaluation, but a resolve is + // sent once, so a resolve that failed to deliver is stored and sent again + // on later ticks (see resendPendingNotifications). Failures before + // dispatch (deleted webhook, notification cap) carry no job: sending them + // again cannot succeed. + const savePendingResolves = async (failures: NotificationFailure[]) => { + const records = failures.flatMap(failure => { + const record = + failure.job && + toPendingNotification(failure.job, nowInMinsRoundDown, wallClockNow); + return record ? [record] : []; + }); + if (records.length === 0) { + return; + } + try { + await alertProvider.savePendingNotifications(records); + } catch (e) { + logger.error( + { + alertId: alert.id, + eventIds: records.map(r => r.eventId), + error: serializeError(e), + }, + 'Failed to store undelivered alert resolve', + ); + } + }; + // Helper to send a notification, catching and logging any errors. // Resolves true when a dispatch was attempted (even if delivery later // fails — failures surface via executionErrors, mirroring the threshold @@ -1375,12 +1410,15 @@ export const processAlert = async ( alertId: alert.id, group, target: failure.target, - error: serializeError(failure.error), + error: serializeErrorWithCause(failure.error), }, 'Notification target could not be dispatched', ); executionErrors.push(makeNotificationAlertError(failure)); } + if (state === AlertState.OK) { + await savePendingResolves(failures); + } } catch (e) { // Render-level failures (title/link building, template compile) — // nothing was dispatched. @@ -2123,6 +2161,7 @@ export default class CheckAlertTask implements HdxTask { async processAlertTask( alertTask: AlertTask, teamWebhooksById: Map, + pendingByAlert: Map = new Map(), ): Promise { return tasksTracer.startActiveSpan('processAlertTask', async span => { setBusinessContext({ teamId: alertTask.conn.team.toString() }); @@ -2155,6 +2194,20 @@ export default class CheckAlertTask implements HdxTask { 'processAlert', async () => { setBusinessContext({ teamId: conn.team.toString() }); + const pending = pendingByAlert.get(alert.alert.id); + if (pending != null && pending.length > 0) { + await resendPendingNotifications({ + alert: alert.alert, + getPreviousState: group => + alert.previousMap.get( + computeHistoryMapKey(alert.alert.id, group), + )?.state, + now: alertTask.now, + pending, + provider: this.provider, + teamWebhooksById, + }); + } // Set once here, so the alert's own query, the EXPLAIN // probes and the notification's sample rows all get tagged. await withQueryAttribution( @@ -2204,6 +2257,34 @@ export default class CheckAlertTask implements HdxTask { }); } + /** + * Failed resolves to send again this tick, by alert id. A failed lookup only + * delays them to the next tick, so it must not stop evaluation. + */ + private async getPendingNotificationsByAlert( + alertIds: string[], + ): Promise> { + const byAlert = new Map(); + try { + const records = await this.provider.getPendingNotifications(alertIds); + for (const record of records) { + const alertId = record.alert.toString(); + const existing = byAlert.get(alertId); + if (existing) { + existing.push(record); + } else { + byAlert.set(alertId, [record]); + } + } + } catch (e) { + logger.error( + { error: serializeError(e) }, + 'Failed to load pending alert notifications', + ); + } + return byAlert; + } + async execute(): Promise { if (this.args.taskName !== 'check-alerts') { throw new Error( @@ -2246,12 +2327,20 @@ export default class CheckAlertTask implements HdxTask { `Obtained teams and webhooks for all alertTasks`, ); + const pendingByAlert = await this.getPendingNotificationsByAlert( + alertTasks.flatMap(task => task.alerts.map(details => details.alert.id)), + ); + let failedBatchCount = 0; for (const task of alertTasks) { const teamWebhooksById = teamToWebhooks.get(task.conn.team.toString()) ?? new Map(); this.task_queue.add(async () => { - const success = await this.processAlertTask(task, teamWebhooksById); + const success = await this.processAlertTask( + task, + teamWebhooksById, + pendingByAlert, + ); if (!success) { failedBatchCount++; } diff --git a/packages/api/src/tasks/checkAlerts/pendingNotifications.ts b/packages/api/src/tasks/checkAlerts/pendingNotifications.ts new file mode 100644 index 0000000000..6ac370a3f9 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/pendingNotifications.ts @@ -0,0 +1,229 @@ +import { AlertErrorType } from '@hyperdx/common-utils/dist/types'; +import mongoose from 'mongoose'; +import ms from 'ms'; + +import { AlertState, IAlert } from '@/models/alert'; +import { IPendingNotification } from '@/models/pendingNotification'; +import { IWebhook } from '@/models/webhook'; +import { + deliverNotification, + NotificationJob, +} from '@/tasks/checkAlerts/notifications'; +import type { AlertProvider } from '@/tasks/checkAlerts/providers'; +import type { Message } from '@/tasks/checkAlerts/transports'; +import { serializeErrorWithCause } from '@/utils/errors'; +import { getCounter } from '@/utils/instrumentation'; +import logger from '@/utils/logger'; + +/** How long a failed resolve keeps being sent again. */ +export const RESEND_WINDOW_MS = ms('24h'); +const BASE_BACKOFF_MS = ms('1m'); +const MAX_BACKOFF_MS = ms('30m'); +// Longer than the re-send window, so the TTL only removes records the task +// never gets to, such as those of a deleted alert. +const RECORD_TTL_MS = RESEND_WINDOW_MS + ms('7d'); + +export type ResendOutcome = + | 'delivered' + | 'failed' + | 'expired' + | 'superseded' + | 'webhook_missing' + | 'skipped_silenced'; + +const notificationResendsCounter = getCounter( + 'hyperdx.alerts.notification_resends', + { + description: + 'Count of attempts to send a failed alert resolve again, labeled by outcome (delivered, failed, expired, superseded, webhook_missing, skipped_silenced).', + }, +); + +/** 1m after the first failure, doubling up to 30m. */ +export const getNextAttemptAt = (now: Date, attempts: number): Date => + new Date( + now.getTime() + + Math.min( + BASE_BACKOFF_MS * 2 ** Math.max(0, attempts - 1), + MAX_BACKOFF_MS, + ), + ); + +/** + * The record to store for a resolve whose delivery failed, or undefined when + * the job cannot be sent again from a record (it is not a webhook delivery, or + * it has no alert or team). + */ +export const toPendingNotification = ( + job: NotificationJob, + windowStart: Date, + now: Date, +): IPendingNotification | undefined => { + if ( + job.populatedChannel.type !== 'webhook' || + job.alertId == null || + job.teamId == null + ) { + return undefined; + } + return { + eventId: job.eventId, + team: new mongoose.Types.ObjectId(job.teamId), + alert: new mongoose.Types.ObjectId(job.alertId), + group: job.group ?? '', + webhook: new mongoose.Types.ObjectId( + job.populatedChannel.channel._id.toString(), + ), + message: JSON.stringify(job.message), + windowStart, + firstFailedAt: now, + attempts: 1, + nextAttemptAt: getNextAttemptAt(now, 1), + expireAt: new Date(now.getTime() + RECORD_TTL_MS), + }; +}; + +const resendOne = async ({ + alert, + getPreviousState, + now, + provider, + record, + teamWebhooksById, +}: { + alert: IAlert; + getPreviousState: (group: string) => AlertState | undefined; + now: Date; + provider: AlertProvider; + record: IPendingNotification; + teamWebhooksById: Map; +}): Promise => { + // The group fired again after the failed resolve, and that ALERT has gone + // out. A late resolve would close the new incident. + if (getPreviousState(record.group) === AlertState.ALERT) { + await provider.deletePendingNotification(record.eventId); + return 'superseded'; + } + + const webhook = teamWebhooksById.get(record.webhook.toString()); + if (webhook == null) { + await provider.deletePendingNotification(record.eventId); + return 'webhook_missing'; + } + + if (now.getTime() - record.firstFailedAt.getTime() >= RESEND_WINDOW_MS) { + await provider.appendAlertErrorHistory(alert.id, record.windowStart, { + timestamp: now, + type: AlertErrorType.WEBHOOK_ERROR, + message: `Resolve notification to webhook "${webhook.name}" was not delivered within ${ms(RESEND_WINDOW_MS, { long: true })} (${record.attempts} ${record.attempts === 1 ? 'attempt' : 'attempts'}). The incident may still be open in the receiving tool.`, + }); + await provider.deletePendingNotification(record.eventId); + return 'expired'; + } + + if (record.nextAttemptAt.getTime() > now.getTime()) { + return undefined; + } + + if ((alert.silenced?.until?.getTime() ?? 0) > now.getTime()) { + return 'skipped_silenced'; + } + + const message: Message = JSON.parse(record.message); + try { + await deliverNotification({ + eventId: record.eventId, + alertId: alert.id, + teamId: record.team.toString(), + group: record.group, + populatedChannel: { type: 'webhook', channel: webhook }, + message, + }); + } catch (e) { + const attempts = record.attempts + 1; + logger.warn( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: webhook._id.toString(), + attempts, + error: serializeErrorWithCause(e), + }, + 'Failed to send alert resolve again', + ); + await provider.updatePendingNotification(record.eventId, { + attempts, + nextAttemptAt: getNextAttemptAt(now, attempts), + }); + return 'failed'; + } + await provider.deletePendingNotification(record.eventId); + logger.info( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: webhook._id.toString(), + attempts: record.attempts + 1, + }, + 'Sent alert resolve again after an earlier delivery failed', + ); + return 'delivered'; +}; + +/** + * Send the alert's failed resolves again, ahead of its evaluation so a resolve + * still reaches the receiver before any ALERT this evaluation sends. Never + * throws: a failure here must not stop the evaluation. + */ +export const resendPendingNotifications = async ({ + alert, + getPreviousState, + now, + pending, + provider, + teamWebhooksById, +}: { + alert: IAlert; + /** The state of the group's latest history row. */ + getPreviousState: (group: string) => AlertState | undefined; + now: Date; + pending: IPendingNotification[]; + provider: AlertProvider; + teamWebhooksById: Map; +}): Promise => { + for (const record of pending) { + try { + const outcome = await resendOne({ + alert, + getPreviousState, + now, + provider, + record, + teamWebhooksById, + }); + if (outcome != null) { + notificationResendsCounter.add(1, { outcome }); + } + if (outcome === 'expired') { + logger.error( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: record.webhook.toString(), + attempts: record.attempts, + }, + 'Gave up sending alert resolve again', + ); + } + } catch (e) { + logger.error( + { + alertId: alert.id, + eventId: record.eventId, + error: serializeErrorWithCause(e), + }, + 'Failed to process pending alert notification', + ); + } + } +}; diff --git a/packages/api/src/tasks/checkAlerts/providers/default.ts b/packages/api/src/tasks/checkAlerts/providers/default.ts index b409fafe98..ebcc47c3b6 100644 --- a/packages/api/src/tasks/checkAlerts/providers/default.ts +++ b/packages/api/src/tasks/checkAlerts/providers/default.ts @@ -27,6 +27,9 @@ import AlertHistory, { } from '@/models/alertHistory'; import Connection, { IConnection } from '@/models/connection'; import Dashboard from '@/models/dashboard'; +import PendingNotification, { + IPendingNotification, +} from '@/models/pendingNotification'; import { type ISavedSearch, SavedSearch } from '@/models/savedSearch'; import { type ISource, Source } from '@/models/source'; import Webhook, { IWebhook } from '@/models/webhook'; @@ -672,6 +675,61 @@ export default class DefaultAlertProvider implements AlertProvider { ); } + async appendAlertErrorHistory( + alertId: string, + evaluationWindowStart: Date, + error: IAlertError, + ) { + await AlertHistory.updateOne( + { + alert: new mongoose.Types.ObjectId(alertId), + createdAt: evaluationWindowStart, + state: AlertState.ERROR, + }, + { + $push: { errors: error }, + $setOnInsert: { counts: 0, lastValues: [] }, + }, + { upsert: true }, + ); + } + + async getPendingNotifications(alertIds: string[]) { + if (alertIds.length === 0) { + return []; + } + return PendingNotification.find({ + alert: { $in: alertIds.map(id => new mongoose.Types.ObjectId(id)) }, + }).lean(); + } + + async savePendingNotifications(records: IPendingNotification[]) { + if (records.length === 0) { + return; + } + await PendingNotification.bulkWrite( + records.map(record => ({ + replaceOne: { + filter: { eventId: record.eventId }, + replacement: record, + upsert: true, + }, + })), + { ordered: false }, + ); + } + + async updatePendingNotification( + eventId: string, + update: Pick, + ) { + await PendingNotification.updateOne({ eventId }, { $set: update }); + } + + async deletePendingNotification(eventId: string) { + await PendingNotification.deleteOne({ eventId }); + } + async getWebhooks(teamId: string | ObjectId) { const webhooks = await Webhook.find({ team: new mongoose.Types.ObjectId(teamId), diff --git a/packages/api/src/tasks/checkAlerts/providers/index.ts b/packages/api/src/tasks/checkAlerts/providers/index.ts index 96071696ed..151d472e80 100644 --- a/packages/api/src/tasks/checkAlerts/providers/index.ts +++ b/packages/api/src/tasks/checkAlerts/providers/index.ts @@ -7,6 +7,7 @@ import { IAlert, IAlertError } from '@/models/alert'; import { IAlertHistory, IAlertHistoryAnalytics } from '@/models/alertHistory'; import { IConnection } from '@/models/connection'; import { IDashboard } from '@/models/dashboard'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISavedSearch } from '@/models/savedSearch'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; @@ -142,6 +143,31 @@ export interface AlertProvider { analytics?: IAlertHistoryAnalytics, ): Promise; + /** + * Append one error to the ERROR-state history row of the given window, + * keeping the errors already on it. Used for failures noticed after the + * window's evaluation finished, such as a resolve that was never delivered. + */ + appendAlertErrorHistory( + alertId: string, + evaluationWindowStart: Date, + error: IAlertError, + ): Promise; + + /** Notifications waiting to be sent again for the given alerts. */ + getPendingNotifications(alertIds: string[]): Promise; + + /** Store failed notifications, replacing any stored with the same eventId. */ + savePendingNotifications(records: IPendingNotification[]): Promise; + + /** Record another failed attempt for a stored notification. */ + updatePendingNotification( + eventId: string, + update: Pick, + ): Promise; + + deletePendingNotification(eventId: string): Promise; + /** Fetch all webhooks for the given team, returning a map of webhook ID to webhook */ getWebhooks(teamId: string | ObjectId): Promise>; diff --git a/packages/api/src/tasks/checkAlerts/template.ts b/packages/api/src/tasks/checkAlerts/template.ts index 94e96d1dd3..9152df4055 100644 --- a/packages/api/src/tasks/checkAlerts/template.ts +++ b/packages/api/src/tasks/checkAlerts/template.ts @@ -62,6 +62,7 @@ import { createHandlebarsWithHelpers } from '@/tasks/checkAlerts/transports'; import { unflattenObject } from '@/tasks/util'; import { resolveAlertDisplayFields } from '@/utils/alerts'; import { truncateString } from '@/utils/common'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getCounter } from '@/utils/instrumentation'; import logger from '@/utils/logger'; @@ -479,6 +480,8 @@ export type NotificationFailure = { /** The channel type the target belongs to, or 'unknown' when it couldn't be determined (e.g. an unparseable @mention). */ type: AlertChannelType | 'unknown'; error: unknown; + /** Set when `dispatch()` rejected, so the caller can send the job again. */ + job?: NotificationJob; }; // PopulatedAlertChannel only has a `webhook` variant in this repo, but a @@ -995,7 +998,7 @@ ${targetTemplate}`; { alertId: alert.id, webhookId: channelKey(job.populatedChannel), - error: serializeError(e), + error: serializeErrorWithCause(e), }, 'Failed to deliver alert notification', ); @@ -1003,6 +1006,7 @@ ${targetTemplate}`; target: channelLabel(job.populatedChannel), type: job.populatedChannel.type, error: e, + job, }); } finally { timings.push({ diff --git a/packages/api/src/tasks/checkAlerts/transports/generic.ts b/packages/api/src/tasks/checkAlerts/transports/generic.ts index b4b2f04769..f5bb280300 100644 --- a/packages/api/src/tasks/checkAlerts/transports/generic.ts +++ b/packages/api/src/tasks/checkAlerts/transports/generic.ts @@ -20,6 +20,7 @@ import type { WebhookChannel, } from '@/tasks/checkAlerts/transports/types'; import { escapeJsonString } from '@/tasks/util'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getCounter, getHistogram } from '@/utils/instrumentation'; import logger from '@/utils/logger'; import { withRetry } from '@/utils/retry'; @@ -293,7 +294,8 @@ const sendGenericWebhook = async ( } catch (e) { logger.error( { - error: serializeError(e), + error: serializeErrorWithCause(e), + webhookId: webhook._id.toString(), }, 'Failed to send generic webhook message', ); diff --git a/packages/api/src/utils/__tests__/errors.test.ts b/packages/api/src/utils/__tests__/errors.test.ts index 75999d1bd7..fada1dd114 100644 --- a/packages/api/src/utils/__tests__/errors.test.ts +++ b/packages/api/src/utils/__tests__/errors.test.ts @@ -1,4 +1,9 @@ -import { Api500Error, BaseError, isOperationalError } from '@/utils/errors'; +import { + Api500Error, + BaseError, + isOperationalError, + serializeErrorWithCause, +} from '@/utils/errors'; describe('Errors utils', () => { test('BaseError class', () => { @@ -18,4 +23,53 @@ describe('Errors utils', () => { ).toBeTruthy(); expect(isOperationalError(new Api500Error('BANG'))).toBeTruthy(); }); + + describe('serializeErrorWithCause', () => { + it('keeps the cause chain that serializeError drops', () => { + const connectError = Object.assign( + new Error('connect ECONNREFUSED 10.0.0.1:443'), + { code: 'ECONNREFUSED' }, + ); + const error = new TypeError('fetch failed', { cause: connectError }); + + expect(serializeErrorWithCause(error)).toMatchObject({ + name: 'TypeError', + message: 'fetch failed', + cause: { + message: 'connect ECONNREFUSED 10.0.0.1:443', + code: 'ECONNREFUSED', + }, + }); + }); + + it("keeps an AggregateError's errors", () => { + const error = new TypeError('fetch failed', { + cause: new AggregateError( + [new Error('connect ETIMEDOUT a'), new Error('connect ETIMEDOUT b')], + 'all attempts failed', + ), + }); + + expect(serializeErrorWithCause(error)).toMatchObject({ + cause: { + message: 'all attempts failed', + errors: [ + { message: 'connect ETIMEDOUT a' }, + { message: 'connect ETIMEDOUT b' }, + ], + }, + }); + }); + + it('stops at a self-referential cause', () => { + const error = new Error('loop'); + error.cause = error; + + expect(() => serializeErrorWithCause(error)).not.toThrow(); + }); + + it('passes non-errors through serializeError', () => { + expect(serializeErrorWithCause('boom')).toBe('boom'); + }); + }); }); diff --git a/packages/api/src/utils/errors.ts b/packages/api/src/utils/errors.ts index b50c55f99f..0b99319390 100644 --- a/packages/api/src/utils/errors.ts +++ b/packages/api/src/utils/errors.ts @@ -1,3 +1,5 @@ +import { serializeError } from 'serialize-error'; + export enum StatusCode { BAD_REQUEST = 400, CONFLICT = 409, @@ -63,3 +65,30 @@ export const isDuplicateKeyError = (e: unknown): boolean => e != null && typeof e === 'object' && (e as { code?: unknown }).code === 11000; + +// Guards against self-referential cause chains. +const MAX_CAUSE_DEPTH = 5; + +/** + * serializeError, keeping `cause` and AggregateError's `errors`, which it + * drops. undici's fetch rejects with a bare "fetch failed" and puts the + * reason (DNS, connect, TLS) only on `cause`. + */ +export const serializeErrorWithCause = (error: unknown, depth = 0): unknown => { + if (!(error instanceof Error)) { + return serializeError(error); + } + const serialized: Record = { ...serializeError(error) }; + if (depth >= MAX_CAUSE_DEPTH) { + return serialized; + } + if (error.cause != null) { + serialized.cause = serializeErrorWithCause(error.cause, depth + 1); + } + if (error instanceof AggregateError) { + serialized.errors = error.errors.map(e => + serializeErrorWithCause(e, depth + 1), + ); + } + return serialized; +}; diff --git a/packages/api/src/utils/retry.ts b/packages/api/src/utils/retry.ts index fd14bfcdff..1c5c729c84 100644 --- a/packages/api/src/utils/retry.ts +++ b/packages/api/src/utils/retry.ts @@ -1,5 +1,4 @@ -import { serializeError } from 'serialize-error'; - +import { serializeErrorWithCause } from './errors'; import logger from './logger'; interface RetryOptions { @@ -82,7 +81,7 @@ export const withRetry = async ( logger.warn( { - error: serializeError(error), + error: serializeErrorWithCause(error), attempt, maxRetries, delayMs: delay,