Prevent notification retries from replaying resolved alerts

Resolution only cancelled pending/sending rows, so retained failures could resurrect healthy incidents after operator retry and restart. Suppress terminal firing entries too, preserve unrelated grouped alerts and recovery jobs, and reconcile queue health after releasing locks. Reject stale per-item retries of cancelled or delivered rows. Regression tests reproduce both bypasses and retain failed-attempt history; focused race tests pass.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/notifications.md

@@ -571,11 +571,43 @@ every destination the decision webhook delivery already made for HTTP 4xx in
 `isRetryableWebhookError`.
 
 Dead-lettering early must not lose the notification: `RetryTerminalFailures`
-remains the operator's recovery path, returning retained terminal failures to
-the queue with a fresh budget once the credentials or configuration are fixed.
+remains the operator's recovery path, returning eligible retained terminal
+failures to the queue with a fresh budget once the credentials or configuration
+are fixed. Resolution removes obsolete firing entries from that eligibility;
+retry must not resurrect an incident which has already recovered.
 A dead-letter row records `failureClass` and a `deadLetterReason` of
 `failure_class_not_retryable` or `max_retries_exhausted` so the two are
 distinguishable in local logs.
 
 `internal/notifications/failure_class_test.go` pins the retryable split and
 that a deterministic failure dead-letters on its first attempt.
+
+
+### Resolution remains final across terminal retries and restart
+
+Resolution cancellation covers pending, sending, failed, and dead-lettered
+firing rows. A wholly obsolete row becomes cancelled; a grouped row retains
+only unrelated firing alerts and their operational links. Recovery jobs are
+not cancelled by this operation. Only removed pending entries contribute to
+the pending-suppression return count; terminal or interrupted sends must not
+be counted as proof that firing was never delivered.
+
+The cancellation verdict persists across queue reopen and bulk operator
+retry. Per-item retry also rejects cancelled and already-sent rows atomically,
+so a stale retry request cannot bypass resolution or duplicate a completed
+delivery. Pending, sending, failed, and dead-lettered rows remain eligible for
+the existing retry scheduler.
+
+Cancelling a row retains its failed-attempt audit history and announces the
+changed queue-health verdict only after releasing both the database mutex and
+per-alert delivery gates. Clearing obsolete retained failures does not prove
+that a destination has been repaired. Nor does this operation retrospectively
+identify obsolete rows whose resolution happened before this behaviour was
+installed; historical backlogs still require incident reconciliation.
+
+`internal/notifications/queue_resolution_retry_test.go` proves failed and
+dead-lettered cancellation across durable reopen and actual queue processing,
+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.

internal/notifications/queue.go

@@ -1205,7 +1205,8 @@ func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotificati
 	return &notif, nil
 }
 
-// ScheduleRetry schedules a notification for retry with exponential backoff
+// ScheduleRetry schedules a notification for retry with exponential backoff.
+// Cancelled and delivered rows are final, even for stale operator requests.
 func (nq *NotificationQueue) ScheduleRetry(id string, attempt int) error {
 	backoff := calculateBackoff(attempt)
 	nextRetry := time.Now().Add(backoff)
@@ -1231,10 +1232,10 @@ func (nq *NotificationQueue) ScheduleRetry(id string, attempt int) error {
 		UPDATE notification_queue
 		SET status = 'pending', next_retry_at = ?, last_attempt = ?,
 		    operational_links = ?, completed_at = NULL, last_error = NULL
-		WHERE id = ?
+		WHERE id = ? AND status IN ('pending', 'sending', 'failed', 'dlq')
 	`
 
-	_, err = nq.db.Exec(
+	result, err := nq.db.Exec(
 		query,
 		nextRetry.Unix(),
 		time.Now().Unix(),
@@ -1245,7 +1246,14 @@ func (nq *NotificationQueue) ScheduleRetry(id string, attempt int) error {
 		nq.mu.Unlock()
 		return fmt.Errorf("failed to schedule retry: %w", err)
 	}
+	affected, err := result.RowsAffected()
 	nq.mu.Unlock()
+	if err != nil {
+		return fmt.Errorf("read scheduled retry result: %w", err)
+	}
+	if affected == 0 {
+		return fmt.Errorf("notification %s is no longer eligible for retry", id)
+	}
 
 	log.Debug().
 		Str("id", id).
@@ -2074,22 +2082,29 @@ func calculateBackoff(attempt int) time.Duration {
 // returns the number of matched firing-alert entries removed from rows that
 // were still waiting for delivery ('pending'). Entries in rows already
 // mid-send ('sending') are cancelled best-effort but not counted, because
-// their delivery may still complete.
+// their delivery may still complete. Failed/dead-lettered firing entries are
+// also suppressed so operator retry cannot resurrect a resolved incident;
+// these do not contribute to the pending-only return count.
 func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) (int, error) {
 	alertIdentifiers = normalizeAlertIdentifiers(alertIdentifiers)
 	if len(alertIdentifiers) == 0 {
 		return 0, nil
 	}
 	releaseDeliveryGates := nq.acquireAlertDeliveryGates(alertIdentifiers, true)
-	defer releaseDeliveryGates()
-
+	healthChanged := false
 	nq.mu.Lock()
-	defer nq.mu.Unlock()
+	defer func() {
+		nq.mu.Unlock()
+		releaseDeliveryGates()
+		if healthChanged {
+			nq.notifyDeliveryHealthChanged()
+		}
+	}()
 
 	query := `
 		SELECT id, type, status, alerts, operational_links
 		FROM notification_queue
-		WHERE status IN ('pending', 'sending')
+		WHERE status IN ('pending', 'sending', 'failed', 'dlq')
 	`
 
 	rows, err := nq.db.Query(query)
@@ -2248,7 +2263,9 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
 					Str("action", "cancel_mark_notification").
 					Str("notifID", notifID).
 					Msg("Failed to mark notification as cancelled")
+				return 0, fmt.Errorf("cancel resolved notification %s: %w", notifID, err)
 			}
+			healthChanged = true
 		}
 	}
 

internal/notifications/queue_resolution_retry_test.go

@@ -0,0 +1,163 @@
+package notifications
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+)
+
+// A repaired destination must not replay obsolete firing alerts. Exercise the
+// operator retry after reopening the database, not just the cancellation query.
+func TestResolvedTerminalFiringIsNotReplayedAfterRestart(t *testing.T) {
+	for _, terminal := range []NotificationQueueStatus{QueueStatusFailed, QueueStatusDLQ} {
+		t.Run(string(terminal), func(t *testing.T) {
+			dir := t.TempDir()
+			q, err := NewNotificationQueue(dir)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { _ = q.Stop() }()
+			for _, n := range []*QueuedNotification{
+				{ID: "obsolete", Type: "webhook", Alerts: []*alerts.Alert{{ID: "healthy"}}},
+				{ID: "group", Type: "webhook", Alerts: []*alerts.Alert{{ID: "healthy"}, {ID: "still-firing"}}},
+				{ID: "recovery", Type: "webhook_resolved", Alerts: []*alerts.Alert{{ID: "healthy"}}},
+			} {
+				n.Config = []byte("{}")
+				n.Status = QueueStatusPending
+				n.MaxAttempts = 3
+				if err := q.Enqueue(n); err != nil {
+					t.Fatal(err)
+				}
+				if err := q.UpdateStatus(n.ID, terminal, "destination unavailable"); err != nil {
+					t.Fatal(err)
+				}
+				if err := q.RecordAudit(n, false, "destination unavailable"); err != nil {
+					t.Fatal(err)
+				}
+			}
+			healthCallbacks := 0
+			q.SetDeliveryHealthChangedCallback(func() {
+				healthCallbacks++
+				// Callback must run after both the DB mutex and alert gate
+				// are released, and see the committed cancellation.
+				release := q.acquireAlertDeliveryGates([]string{"healthy"}, true)
+				defer release()
+				stats, err := q.GetQueueStats()
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				if stats[string(terminal)] != 2 {
+					t.Errorf("remaining terminal rows = %d, want 2", stats[string(terminal)])
+				}
+			})
+			count, err := q.CancelByAlertIdentifiers([]string{"healthy"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			q.SetDeliveryHealthChangedCallback(nil)
+			if healthCallbacks != 1 {
+				t.Errorf("health callbacks = %d, want 1", healthCallbacks)
+			}
+			if count != 0 {
+				t.Fatalf("pending suppression count = %d, want 0 for terminal rows", count)
+			}
+			if err := q.Stop(); err != nil {
+				t.Fatal(err)
+			}
+			q, err = NewNotificationQueue(dir)
+			if err != nil {
+				t.Fatal(err)
+			}
+			retried, err := q.RetryTerminalFailures()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if retried != 2 {
+				t.Errorf("retried %d rows, want only surviving group and recovery", retried)
+			}
+			delivered := map[string][]string{}
+			var mu sync.Mutex
+			q.SetProcessor(func(n *QueuedNotification) error {
+				mu.Lock()
+				defer mu.Unlock()
+				for _, a := range n.Alerts {
+					delivered[n.ID] = append(delivered[n.ID], a.ID)
+				}
+				return nil
+			})
+			q.processBatch()
+			deadline := time.Now().Add(3 * time.Second)
+			for {
+				var pending int
+				if err := q.db.QueryRow("SELECT count(*) FROM notification_queue WHERE status IN ('pending', 'sending')").Scan(&pending); err != nil {
+					t.Fatal(err)
+				}
+				if pending == 0 {
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatal("queue did not drain")
+				}
+				time.Sleep(time.Millisecond)
+			}
+			mu.Lock()
+			if len(delivered) != 2 || len(delivered["group"]) != 1 ||
+				delivered["group"][0] != "still-firing" ||
+				len(delivered["recovery"]) != 1 || delivered["recovery"][0] != "healthy" {
+				t.Errorf("replayed payloads = %v, want only still-firing and genuine recovery", delivered)
+			}
+			mu.Unlock()
+			var failures int
+			if err := q.db.QueryRow("SELECT count(*) FROM notification_audit WHERE success = 0").Scan(&failures); err != nil {
+				t.Fatal(err)
+			}
+			if failures != 3 {
+				t.Errorf("retained failed attempts = %d, want 3", failures)
+			}
+		})
+	}
+}
+
+// A stale per-item retry request must not bypass resolution or resend a
+// successful notification. The same scheduler handles transient send failures.
+func TestScheduleRetryRejectsCancelledAndSent(t *testing.T) {
+	q, err := NewNotificationQueue(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = q.Stop() }()
+	for _, status := range []NotificationQueueStatus{
+		QueueStatusCancelled, QueueStatusSent, QueueStatusPending,
+		QueueStatusSending, QueueStatusFailed, QueueStatusDLQ,
+	} {
+		t.Run(string(status), func(t *testing.T) {
+			id := string(status)
+			n := &QueuedNotification{ID: id, Type: "webhook", Status: QueueStatusPending, Config: []byte("{}")}
+			if err := q.Enqueue(n); err != nil {
+				t.Fatal(err)
+			}
+			if err := q.UpdateStatus(id, status, ""); err != nil {
+				t.Fatal(err)
+			}
+			err := q.ScheduleRetry(id, 0)
+			blocked := status == QueueStatusCancelled || status == QueueStatusSent
+			if (err != nil) != blocked {
+				t.Fatalf("retry error = %v, blocked = %v", err, blocked)
+			}
+			want := QueueStatusPending
+			if blocked {
+				want = status
+			}
+			var got NotificationQueueStatus
+			if err := q.db.QueryRow("SELECT status FROM notification_queue WHERE id = ?", id).Scan(&got); err != nil {
+				t.Fatal(err)
+			}
+			if got != want {
+				t.Errorf("status = %s, want %s", got, want)
+			}
+		})
+	}
+}