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 229d8668af)

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.

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

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

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 {

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