From 31331b5451360ce90f7fccf069ae11c12de61f90 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 13 May 2026 12:59:59 +0100 Subject: [PATCH] Fix grouped notification cancellation --- internal/notifications/queue.go | 84 +++++++++++++++++++++++----- internal/notifications/queue_test.go | 44 ++++++++++++--- 2 files changed, 106 insertions(+), 22 deletions(-) diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index adea012df..f8e76a60f 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -1032,7 +1032,8 @@ func calculateBackoff(attempt int) time.Duration { return backoff } -// CancelByAlertIdentifiers marks all queued notifications containing any of the given alert identifiers as cancelled. +// CancelByAlertIdentifiers suppresses queued firing notifications for resolved +// alerts while preserving unrelated alerts in the same grouped queue row. func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) error { if len(alertIdentifiers) == 0 { return nil @@ -1041,7 +1042,6 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) nq.mu.Lock() defer nq.mu.Unlock() - // Query pending/sending notifications query := ` SELECT id, type, alerts FROM notification_queue @@ -1053,11 +1053,28 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) return fmt.Errorf("failed to query notifications for cancellation: %w", err) } - var toCancelIDs []string alertIdentifierSet := make(map[string]struct{}) for _, id := range alertIdentifiers { - alertIdentifierSet[id] = struct{}{} + id = strings.TrimSpace(id) + if id != "" { + alertIdentifierSet[id] = struct{}{} + } } + if len(alertIdentifierSet) == 0 { + if err := rows.Close(); err != nil { + return fmt.Errorf("failed to close cancellation query: %w", err) + } + return nil + } + + type queuedAlertCancellation struct { + notificationID string + remaining []*alerts.Alert + } + + var toCancelIDs []string + var toRewrite []queuedAlertCancellation + suppressedAlertCount := 0 for rows.Next() { var notifID string @@ -1075,8 +1092,8 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) continue } - var alerts []*alerts.Alert - if err := json.Unmarshal(alertsJSON, &alerts); err != nil { + var queuedAlerts []*alerts.Alert + if err := json.Unmarshal(alertsJSON, &queuedAlerts); err != nil { log.Error(). Err(err). Str("component", "notification_queue"). @@ -1086,13 +1103,32 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) continue } - // Check if any alert in this notification matches - for _, alert := range alerts { - if _, exists := alertIdentifierSet[alert.ID]; exists { - toCancelIDs = append(toCancelIDs, notifID) - break + remainingAlerts := make([]*alerts.Alert, 0, len(queuedAlerts)) + matchedAlertCount := 0 + for _, alert := range queuedAlerts { + if alert == nil { + remainingAlerts = append(remainingAlerts, alert) + continue } + if _, exists := alertIdentifierSet[alert.ID]; exists { + matchedAlertCount++ + continue + } + remainingAlerts = append(remainingAlerts, alert) } + if matchedAlertCount == 0 { + continue + } + + suppressedAlertCount += matchedAlertCount + if len(remainingAlerts) == 0 { + toCancelIDs = append(toCancelIDs, notifID) + continue + } + toRewrite = append(toRewrite, queuedAlertCancellation{ + notificationID: notifID, + remaining: remainingAlerts, + }) } rowsErr := rows.Err() @@ -1101,7 +1137,23 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) return fmt.Errorf("error iterating notifications for cancellation: %w", rowsErr) } - // Cancel the matched notifications (using direct SQL since we already hold the lock) + if len(toRewrite) > 0 { + updateAlertsQuery := ` + UPDATE notification_queue + SET alerts = ? + WHERE id = ? + ` + for _, rewrite := range toRewrite { + alertsJSON, err := json.Marshal(rewrite.remaining) + if err != nil { + return fmt.Errorf("failed to marshal remaining alerts for %s: %w", rewrite.notificationID, err) + } + if _, err := nq.db.Exec(updateAlertsQuery, string(alertsJSON), rewrite.notificationID); err != nil { + return fmt.Errorf("failed to rewrite queued notification %s after alert resolution: %w", rewrite.notificationID, err) + } + } + } + if len(toCancelIDs) > 0 { now := time.Now().Unix() updateQuery := ` @@ -1119,13 +1171,17 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) Msg("Failed to mark notification as cancelled") } } + } + if suppressedAlertCount > 0 { log.Info(). Str("component", "notification_queue"). Str("action", "cancel_alert_identifiers"). - Int("count", len(toCancelIDs)). + Int("suppressedAlertCount", suppressedAlertCount). + Int("cancelledRows", len(toCancelIDs)). + Int("rewrittenRows", len(toRewrite)). Strs("alertIdentifiers", alertIdentifiers). - Msg("cancelled queued notifications for resolved alert identifiers") + Msg("suppressed resolved alerts in queued notifications") } return nil diff --git a/internal/notifications/queue_test.go b/internal/notifications/queue_test.go index 89c110f93..ac3d105da 100644 --- a/internal/notifications/queue_test.go +++ b/internal/notifications/queue_test.go @@ -2,6 +2,7 @@ package notifications import ( "database/sql" + "encoding/json" "fmt" "os" "path/filepath" @@ -631,33 +632,60 @@ func TestCancelByAlertIdentifiers_MultipleAlertsPartialMatch(t *testing.T) { Type: "webhook", Status: QueueStatusPending, MaxAttempts: 3, - Config: []byte(`{}`), + Config: []byte(`{"url":"https://hooks.example.test"}`), NextRetryAt: &futureRetry, Alerts: []*alerts.Alert{ {ID: "alert-1"}, {ID: "alert-2"}, + {ID: "alert-3"}, }, } if err := nq.Enqueue(notif); err != nil { t.Fatalf("Failed to enqueue: %v", err) } - // Cancel with only one matching alert ID - should still cancel the notification - err = nq.CancelByAlertIdentifiers([]string{"alert-1"}) + err = nq.CancelByAlertIdentifiers([]string{"alert-1", "alert-3"}) if err != nil { t.Errorf("CancelByAlertIdentifiers returned error: %v", err) } - // Verify the notification is cancelled (any matching alert should cancel) stats, err := nq.GetQueueStats() if err != nil { t.Fatalf("GetQueueStats failed: %v", err) } - if stats["pending"] != 0 { - t.Errorf("Expected 0 pending notifications after partial match cancel, got %d", stats["pending"]) + if stats["pending"] != 1 { + t.Errorf("Expected 1 pending notification after partial match rewrite, got %d", stats["pending"]) } - if stats["cancelled"] != 1 { - t.Errorf("Expected 1 cancelled notification, got %d (stats: %v)", stats["cancelled"], stats) + if stats["cancelled"] != 0 { + t.Errorf("Expected 0 cancelled notifications after partial match rewrite, got %d (stats: %v)", stats["cancelled"], stats) + } + + var notifType string + var alertsJSON string + var configJSON string + var nextRetryAt sql.NullInt64 + if err := nq.db.QueryRow( + `SELECT type, alerts, config, next_retry_at FROM notification_queue WHERE id = ?`, + notif.ID, + ).Scan(¬ifType, &alertsJSON, &configJSON, &nextRetryAt); err != nil { + t.Fatalf("Failed to query rewritten notification row: %v", err) + } + if notifType != notif.Type { + t.Fatalf("type = %q, want %q", notifType, notif.Type) + } + if configJSON != string(notif.Config) { + t.Fatalf("config = %q, want %q", configJSON, string(notif.Config)) + } + if !nextRetryAt.Valid || nextRetryAt.Int64 != futureRetry.Unix() { + t.Fatalf("next_retry_at = %+v, want %d", nextRetryAt, futureRetry.Unix()) + } + + var remaining []*alerts.Alert + if err := json.Unmarshal([]byte(alertsJSON), &remaining); err != nil { + t.Fatalf("failed to unmarshal rewritten alerts: %v", err) + } + if len(remaining) != 1 || remaining[0].ID != "alert-2" { + t.Fatalf("remaining alerts = %#v, want only alert-2", remaining) } }