From cc084e85edab7ed0551f536ca200f8888ea370ae Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 5 Sep 2026 23:29:16 +0100 Subject: [PATCH] Record disabled queued notifications as cancelled, not delivered Distinguish policy skips from provider success so suppressed jobs do not create false sent rows or successful audit entries. Reconcile cancelled queue health after releasing alert gates, preserve real attempt history, and cover all three providers for firing/recovery and global/destination disablement. Change-source: pulse-maintainer (cherry picked from commit 229d8668afd82ff1d2892d7c2bea9826f906069d) --- .../v6/internal/subsystems/notifications.md | 16 ++++ internal/notifications/notifications.go | 6 +- internal/notifications/notifications_test.go | 9 +- internal/notifications/queue.go | 28 ++++++- .../queue_disabled_delivery_test.go | 84 +++++++++++++++++++ 5 files changed, 135 insertions(+), 8 deletions(-) create mode 100644 internal/notifications/queue_disabled_delivery_test.go diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 2f5b845c3..674a0aca6 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -554,3 +554,19 @@ preservation of unrelated grouped firing and recovery jobs, retained failed attempts, callback lock release and committed-state visibility, and the per-item retry eligibility matrix. These are component proofs, not installed receiver receipts or exactly-once delivery guarantees. + +### Disabled delivery is cancellation, not a receipt + +At processing time, globally disabled delivery or a disabled/removed destination +returns `ErrNotificationDeliverySkipped`. The queue persists that job as +cancelled with the policy reason and cancelled operational links. It does not +write a provider-attempt audit, a successful receipt, or a delivery failure, and +operator retry does not replay the cancelled job. Existing attempt history is +retained. Queue health reconciliation runs after releasing the database mutex +and alert delivery gates. + +`queue_disabled_delivery_test.go` exercises the real manager/queue boundary for +email, webhook and Apprise, firing and recovery, and global versus destination +disablement. This corrects false successful queue/audit records; it does not +establish maintenance-window expiry, stop an already-started provider request, +or repair historical false-success records. diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 79d494a22..72e32c848 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -3868,7 +3868,7 @@ func (n *NotificationManager) ProcessQueuedNotification(notif *QueuedNotificatio Str("type", baseType). Str("event", string(event)). Msg("skipping queued email notification because email delivery is disabled") - return nil + return ErrNotificationDeliverySkipped } deliveredJob = notificationDeliveryJob{ Type: "email", @@ -3891,7 +3891,7 @@ func (n *NotificationManager) ProcessQueuedNotification(notif *QueuedNotificatio Str("event", string(event)). Str("webhookID", webhookConfig.ID). Msg("skipping queued webhook notification because delivery is disabled") - return nil + return ErrNotificationDeliverySkipped } deliveredJob = notificationDeliveryJob{ Type: "webhook", @@ -3914,7 +3914,7 @@ func (n *NotificationManager) ProcessQueuedNotification(notif *QueuedNotificatio Str("type", baseType). Str("event", string(event)). Msg("skipping queued Apprise notification because delivery is disabled") - return nil + return ErrNotificationDeliverySkipped } deliveredJob = notificationDeliveryJob{ Type: "apprise", diff --git a/internal/notifications/notifications_test.go b/internal/notifications/notifications_test.go index 1038a0e2d..845bb427a 100644 --- a/internal/notifications/notifications_test.go +++ b/internal/notifications/notifications_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "io" "net/http" "os" @@ -3367,8 +3368,8 @@ func TestProcessQueuedNotification_SkipsDisabledEmailDelivery(t *testing.T) { Alerts: []*alerts.Alert{{ID: "alert-1"}}, } - if err := nm.ProcessQueuedNotification(notif); err != nil { - t.Fatalf("expected queued email notification to be skipped without error, got %v", err) + if err := nm.ProcessQueuedNotification(notif); !errors.Is(err, ErrNotificationDeliverySkipped) { + t.Fatalf("expected queued email notification to report a policy skip, got %v", err) } } @@ -3414,8 +3415,8 @@ func TestProcessQueuedNotification_SkipsWhenNotificationsDisabled(t *testing.T) Alerts: []*alerts.Alert{{ID: "alert-1"}}, } - if err := nm.ProcessQueuedNotification(notif); err != nil { - t.Fatalf("expected queued webhook notification to be skipped without error, got %v", err) + if err := nm.ProcessQueuedNotification(notif); !errors.Is(err, ErrNotificationDeliverySkipped) { + t.Fatalf("expected queued webhook notification to report a policy skip, got %v", err) } } diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index b94ec954f..4b7ee37df 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -3,6 +3,7 @@ package notifications import ( "database/sql" "encoding/json" + "errors" "fmt" "net/url" "os" @@ -19,6 +20,10 @@ import ( _ "modernc.org/sqlite" ) +// ErrNotificationDeliverySkipped distinguishes a policy cancellation from a +// successful provider delivery. Queue processors must not report skips as nil. +var ErrNotificationDeliverySkipped = errors.New("notification delivery disabled") + // defaultQueueMaxAttempts is the default number of delivery attempts // before a notification is moved to the dead-letter queue. With the // exponential backoff schedule (1s doubling, capped at 60s) eight attempts @@ -1715,7 +1720,13 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) { return } releaseDeliveryGates := nq.acquireAlertDeliveryGates(alertIdentifiersFromAlerts(notif.Alerts), false) - defer releaseDeliveryGates() + healthChanged := false + defer func() { + releaseDeliveryGates() + if healthChanged { + nq.notifyDeliveryHealthChanged() + } + }() // Atomically claim the pending row. A concurrent resolution may have // cancelled it while it was waiting for its per-alert delivery gate. @@ -1751,6 +1762,21 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) { err = processor(notif) + if errors.Is(err, ErrNotificationDeliverySkipped) { + // No provider attempt occurred. Preserve the cancellation and its reason, + // but do not manufacture a successful (or failed) delivery audit entry. + // Reconcile health only after releasing the per-alert gate. + nq.mu.Lock() + cancelErr := nq.updateNotificationStatusNoLock(notif.ID, QueueStatusCancelled, err.Error(), time.Now()) + nq.mu.Unlock() + if cancelErr != nil { + log.Error().Err(cancelErr).Str("id", notif.ID).Msg("Failed to cancel skipped notification") + } else { + healthChanged = true + } + return + } + success := err == nil errorMsg := "" if err != nil { diff --git a/internal/notifications/queue_disabled_delivery_test.go b/internal/notifications/queue_disabled_delivery_test.go new file mode 100644 index 000000000..ef8c66743 --- /dev/null +++ b/internal/notifications/queue_disabled_delivery_test.go @@ -0,0 +1,84 @@ +package notifications + +import ( + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" +) + +// Policy skips are neither provider receipts nor delivery failures. Exercise +// the real queue processor: checking only its return value hid false success. +func TestQueueDisabledDeliveryIsCancelledNotSent(t *testing.T) { + for _, kind := range []string{"email", "webhook", "apprise"} { + for _, suffix := range []string{"", "_resolved"} { + for _, enabled := range []bool{false, true} { + name := kind + suffix + "/global-disabled" + if enabled { + name = kind + suffix + "/destination-disabled" + } + t.Run(name, func(t *testing.T) { + q, err := NewNotificationQueue(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer q.Stop() + nm := &NotificationManager{enabled: enabled} + if !enabled { + nm.emailConfig.Enabled = true + nm.appriseConfig.Enabled = true + nm.webhooks = []WebhookConfig{{ID: "ops", Enabled: true}} + } + n := &QueuedNotification{ID: "disabled", Type: kind + suffix, Status: QueueStatusPending, + Links: []operationaltrust.NotificationLink{{DestinationID: "ops", OperationalRecordID: "incident", TransitionID: "firing", LifecycleState: operationaltrust.OperationalOpen, CauseKey: "cpu"}}, + Config: []byte(`{"enabled":true,"id":"ops"}`), MaxAttempts: 3, Alerts: []*alerts.Alert{{ID: "incident"}}} + if err := q.Enqueue(n); err != nil { + t.Fatal(err) + } + // Do not start background workers; process one persisted row synchronously. + q.processor = nm.ProcessQueuedNotification + callbacks := 0 + q.SetDeliveryHealthChangedCallback(func() { + callbacks++ + release := q.acquireAlertDeliveryGates([]string{"incident"}, true) + defer release() + if _, err := q.GetQueueStats(); err != nil { + t.Error(err) + } + }) + q.processNotification(n) + var status string + var completed *int64 + if err := q.db.QueryRow(`SELECT status, completed_at FROM notification_queue WHERE id = ?`, n.ID).Scan(&status, &completed); err != nil { + t.Fatal(err) + } + if status != string(QueueStatusCancelled) || completed == nil { + t.Errorf("status=%s completed=%v; want terminal cancellation", status, completed) + } + links, err := q.getNotificationLinks(n.ID) + if err != nil { + t.Fatal(err) + } + if len(links) != 1 || links[0].DeliveryState != operationaltrust.NotificationCancelled { + t.Errorf("links = %+v, want cancelled", links) + } + for _, table := range []string{"notification_audit", "notification_delivery_receipts"} { + var count int + if err := q.db.QueryRow("SELECT count(*) FROM " + table).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("%s has %d rows for a policy skip", table, count) + } + } + if callbacks != 1 { + t.Errorf("health callbacks = %d, want 1", callbacks) + } + if count, err := q.RetryTerminalFailures(); err != nil || count != 0 { + t.Errorf("retry = %d, %v; want no replay", count, err) + } + }) + } + } + } +}