[HDX-6020] fix(api): re-send resolve notifications whose delivery failed

.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".

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<IPendingNotification>(
+  {
+    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<IPendingNotification>(
+  'PendingNotification',
+  PendingNotificationSchema,
+);

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 () => {

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);