diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 2cc850585..4d69f4034 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/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. diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index ef311465c..b9e962abc 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -1205,7 +1205,8 @@ func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotificati return ¬if, 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 } } diff --git a/internal/notifications/queue_resolution_retry_test.go b/internal/notifications/queue_resolution_retry_test.go new file mode 100644 index 000000000..96cedecee --- /dev/null +++ b/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) + } + }) + } +}