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) + } + }) + } + } + } +}