Keep pending notification groups occurrence-qualified

Coalesce repeated firing callbacks before delivery into the newest complete occurrence snapshot, retaining distinct and unknown identities and the original grouping deadline. Discard the transient index on cancellation, flush, disable and stop; cancel abandoned firing timers before queue shutdown. Verify ordering, concurrency, routing, persistent restart and actual loopback firing/recovery delivery without changing escalation or retry policy.

Change-source: pulse-maintainer

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

@@ -15,6 +15,33 @@
 
 ## Purpose
 
+### Pending firing groups count occurrences, not callbacks
+
+Before a grouping window expires, repeated firing callbacks with the same full
+alert ID and start instant occupy one slot. The newest observed snapshot owns
+the whole payload and operational linkage, including a genuine later severity
+downgrade. Equal observation times prefer higher severity; an unknown time
+cannot replace a known later observation. Missing ID/start identity remains
+separate, as do different occurrences of a reusable ID. The index is scoped to
+the pending window and discarded on flush, disable, cancellation or stop; callbacks
+do not extend the original timer or establish delivery.
+
+`firing_grouping_occurrence_test.go` verifies timestamp/severity ordering,
+instant equality across time-zone/monotonic representations, cloned payloads,
+concurrent callbacks, occurrence-qualified cancellation, all-destination and
+webhook-only disable/zero-window flushes, and notification disable/re-enable.
+Stopping the manager cancels its volatile firing timer and snapshots, rather
+than letting an abandoned timer send directly after its queue is closed.
+Committed queue work retains its existing restart semantics.
+`TestPendingFiringGroupStopCancelsTimer` checks actual timer/HTTP silence after stop.
+The connected `TestPendingFiringGroupHTTPAcrossRestart` uses actual timer expiry,
+persistent queue reopen, the autonomous sender and two loopback receivers: one
+snapshot per occurrence/destination, truthful counts/latest content, ordinary
+cooldown and independent firing/recovery receipts. Ungrouped admissions,
+scheduled escalation, retry budgets, queue schema and delivery policy are
+unchanged. This is source acceptance, not a native destination, natural schedule
+cycle or a reproduced cause of the existing flood reports.
+
 ### Completion order cannot replace a newer occurrence's cooldown
 
 Firing completions still record an independent delivery receipt for their own

internal/notifications/firing_grouping_occurrence_test.go

@@ -0,0 +1,359 @@
+package notifications
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/stretchr/testify/require"
+)
+
+func pendingGroupSnapshot(id string, start, seen time.Time, level alerts.AlertLevel, message string) *alerts.Alert {
+	a := quietReplayAlert(id, level, start)
+	a.LastSeen = seen
+	a.Message = message
+	a.Metadata = map[string]interface{}{"snapshot": message}
+	a.LatestTransition.ID = "transition-" + message
+	return a
+}
+
+func TestPendingFiringGroupSnapshotOrdering(t *testing.T) {
+	start := time.Now().UTC().Add(-time.Minute)
+	first := pendingGroupSnapshot("cpu", start, start, alerts.AlertLevelWarning, "first")
+	for _, tc := range []struct {
+		name    string
+		before  func(*alerts.Alert)
+		update  func(*alerts.Alert)
+		wantNew bool
+		wantTwo bool
+	}{
+		{name: "repeat", wantNew: true},
+		{name: "newer_critical", update: func(a *alerts.Alert) { a.LastSeen = start.Add(time.Second); a.Level = alerts.AlertLevelCritical }, wantNew: true},
+		{name: "newer_downgrade", before: func(a *alerts.Alert) { a.Level = alerts.AlertLevelCritical }, update: func(a *alerts.Alert) { a.LastSeen = start.Add(time.Second) }, wantNew: true},
+		{name: "older_critical", update: func(a *alerts.Alert) { a.LastSeen = start.Add(-time.Second); a.Level = alerts.AlertLevelCritical }},
+		{name: "equal_time_critical", update: func(a *alerts.Alert) { a.Level = alerts.AlertLevelCritical }, wantNew: true},
+		{name: "equal_time_downgrade", before: func(a *alerts.Alert) { a.Level = alerts.AlertLevelCritical }},
+		{name: "unknown_observation", update: func(a *alerts.Alert) { a.LastSeen = time.Time{}; a.Level = alerts.AlertLevelCritical }},
+		{name: "known_observation", before: func(a *alerts.Alert) { a.LastSeen = time.Time{} }, wantNew: true},
+		{name: "same_instant_different_location", update: func(a *alerts.Alert) { a.StartTime = start.In(time.FixedZone("other", 3600)) }, wantNew: true},
+		{name: "same_instant_without_monotonic", before: func(a *alerts.Alert) { a.StartTime = time.Now() }, update: func(a *alerts.Alert) { a.StartTime = a.StartTime.Round(0) }, wantNew: true},
+		{name: "different_occurrence", update: func(a *alerts.Alert) { a.StartTime = start.Add(time.Nanosecond) }, wantTwo: true},
+		{name: "different_identifier", update: func(a *alerts.Alert) { a.ID = "other-cpu" }, wantTwo: true},
+		{name: "unknown_start", before: func(a *alerts.Alert) { a.StartTime = time.Time{} }, wantTwo: true},
+		{name: "unknown_identifier", before: func(a *alerts.Alert) { a.ID = "" }, wantTwo: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := NewNotificationManagerWithDeferredQueue("", t.TempDir())
+			t.Cleanup(m.Stop)
+			m.SetGroupingConfig(true, 3600, false, false)
+			a := first.Clone()
+			if tc.before != nil {
+				tc.before(a)
+			}
+			m.SendAlert(a)
+			m.mu.RLock()
+			timer := m.groupTimer
+			m.mu.RUnlock()
+			incoming := pendingGroupSnapshot(a.ID, a.StartTime, first.LastSeen, alerts.AlertLevelWarning, "incoming")
+			if tc.update != nil {
+				tc.update(incoming)
+			}
+			m.SendAlert(nil)
+			m.SendAlert(incoming)
+			m.mu.RLock()
+			defer m.mu.RUnlock()
+			require.Same(t, timer, m.groupTimer, "an update must not restart the grouping deadline")
+			require.Empty(t, m.lastNotified, "group admission is not delivery")
+			if tc.wantTwo {
+				require.Len(t, m.pendingAlerts, 2, "incomplete or distinct identities must not merge")
+				require.Equal(t, a, m.pendingAlerts[0])
+				require.Equal(t, incoming, m.pendingAlerts[1])
+				return
+			}
+			require.Len(t, m.pendingAlerts, 1, "one occurrence must occupy one group slot")
+			want := a
+			if tc.wantNew {
+				want = incoming
+			}
+			require.Equal(t, want, m.pendingAlerts[0], "payload and operational linkage must come from the same selected snapshot")
+			incoming.Metadata["snapshot"] = "caller-mutated"
+			require.NotEqual(t, incoming.Metadata, m.pendingAlerts[0].Metadata, "pending snapshot must not alias its caller")
+		})
+	}
+}
+
+func TestPendingFiringGroupConcurrentSnapshots(t *testing.T) {
+	m := NewNotificationManagerWithDeferredQueue("", t.TempDir())
+	t.Cleanup(m.Stop)
+	m.SetGroupingConfig(true, 3600, false, false)
+	start := time.Now().UTC().Add(-time.Minute)
+	var wg sync.WaitGroup
+	for i := range 64 {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			m.SendAlert(pendingGroupSnapshot("cpu", start, start.Add(time.Duration(i)*time.Second), alerts.AlertLevelWarning, strconv.Itoa(i)))
+		}(i)
+	}
+	wg.Wait()
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	require.Len(t, m.pendingAlerts, 1)
+	require.Equal(t, "63", m.pendingAlerts[0].Message)
+	require.Equal(t, "transition-63", m.pendingAlerts[0].LatestTransition.ID)
+	require.Empty(t, m.lastNotified)
+}
+
+func TestPendingFiringGroupCancellationAndFlush(t *testing.T) {
+	for _, target := range []string{"all", "webhook"} {
+		for _, flush := range []string{"disable", "zero_window", "disable_notifications"} {
+			t.Run(target+"/"+flush, func(t *testing.T) {
+				m := NewNotificationManagerWithDeferredQueue("", t.TempDir())
+				t.Cleanup(m.Stop)
+				m.SetGroupingConfig(true, 3600, false, false)
+				m.SetInitialNotifyTarget(target)
+				m.SetEmailConfig(EmailConfig{Enabled: true, SMTPHost: "smtp.example.test", SMTPPort: 587, From: "pulse@example.test", To: []string{"ops@example.test"}})
+				m.SetAppriseConfig(AppriseConfig{Enabled: true, Targets: []string{"ntfy://example.test/pulse"}})
+				m.AddWebhook(WebhookConfig{ID: "ops", URL: "https://ops.example.test", Enabled: true})
+				start := time.Now().UTC().Add(-time.Minute)
+				old := pendingGroupSnapshot("cpu", start, start, alerts.AlertLevelCritical, "old")
+				current := pendingGroupSnapshot("cpu", start.Add(time.Nanosecond), start, alerts.AlertLevelWarning, "current")
+				current.OperationalRecord.ID += "-new"
+				peer := pendingGroupSnapshot("memory", start, start, alerts.AlertLevelWarning, "peer")
+				for _, a := range []*alerts.Alert{old, current, old, peer, current} {
+					m.SendAlert(a)
+				}
+				require.True(t, m.CancelResolvedAlert(old))
+				// Rebuild after cancellation; a stale index must neither overwrite
+				// the peer nor append the current occurrence again.
+				current.LastSeen = start.Add(time.Second)
+				current.Message = "updated-current"
+				current.LatestTransition.ID = "updated-transition"
+				m.SendAlert(current)
+				m.mu.RLock()
+				pending := append([]*alerts.Alert(nil), m.pendingAlerts...)
+				m.mu.RUnlock()
+				require.Equal(t, []*alerts.Alert{current, peer}, pending)
+				switch flush {
+				case "disable":
+					m.SetGroupingConfig(false, 30, false, false)
+				case "zero_window":
+					m.SetGroupingConfig(true, 0, false, false)
+				case "disable_notifications":
+					m.SetEnabled(false)
+					m.SetEnabled(true)
+					m.SendAlert(peer)
+					m.SetGroupingConfig(false, 0, false, false)
+					pending = []*alerts.Alert{peer}
+				}
+				rows := quietReplayRows(t, m.GetQueue())
+				destinations := 3
+				if target == "webhook" {
+					destinations = 1
+				}
+				require.Len(t, rows, destinations*len(pending))
+				counts := map[string]int{}
+				for _, row := range rows {
+					require.Equal(t, QueueStatusPending, row.Status)
+					require.Zero(t, row.Attempts)
+					require.Len(t, row.Alerts, 1)
+					require.Len(t, row.Links, 1)
+					a := row.Alerts[0]
+					require.Equal(t, a.LatestTransition.ID, row.Links[0].TransitionID)
+					require.False(t, old.ID == a.ID && old.StartTime.Equal(a.StartTime), "cancelled occurrence must not reach the queue")
+					counts[row.Type+"/"+a.ID]++
+				}
+				for _, count := range counts {
+					require.Equal(t, 1, count, "flush duplicated an occurrence/destination")
+				}
+				m.mu.RLock()
+				require.Empty(t, m.pendingAlerts)
+				require.Nil(t, m.groupTimer)
+				require.Empty(t, m.lastNotified, "persisting pending work is not a successful delivery")
+				m.mu.RUnlock()
+			})
+		}
+	}
+}
+
+// The real timer admits one snapshot per occurrence into the persistent queue.
+// Bootstrap restart, autonomous processing and recovery use normal entry points;
+// no synthetic successful receipt, direct dispatcher or edited queue clock.
+func TestPendingFiringGroupHTTPAcrossRestart(t *testing.T) {
+	type message struct {
+		destination string
+		Event       string          `json:"event"`
+		Grouped     bool            `json:"grouped"`
+		Alerts      []*alerts.Alert `json:"alerts"`
+		Count       int             `json:"count"`
+	}
+	received := make(chan message, 16)
+	server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		var msg message
+		if err := json.NewDecoder(r.Body).Decode(&msg); err != nil {
+			t.Errorf("decode grouped delivery: %v", err)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		msg.destination = r.URL.Path
+		select {
+		case received <- msg:
+		default:
+			t.Error("unexpected excess grouped deliveries")
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(server.Close)
+	dir := t.TempDir()
+	hooks := []WebhookConfig{
+		{ID: "left", URL: server.URL + "/left", Enabled: true, Service: "generic"},
+		{ID: "right", URL: server.URL + "/right", Enabled: true, Service: "generic"},
+	}
+	open := func() *NotificationManager {
+		m := NewNotificationManagerWithDeferredQueue("", dir)
+		t.Cleanup(m.Stop)
+		m.webhookClient = server.Client()
+		require.NoError(t, m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"))
+		m.SetCooldown(0)
+		m.SetGroupingConfig(true, 1, false, false)
+		for _, hook := range hooks {
+			m.AddWebhook(hook)
+		}
+		return m
+	}
+	wait := func(check func() bool) {
+		t.Helper()
+		deadline := time.Now().Add(10 * time.Second)
+		for !check() {
+			if time.Now().After(deadline) {
+				t.Fatal("grouped delivery did not complete")
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}
+	m := open()
+	start := time.Now().UTC().Add(-time.Minute)
+	current := pendingGroupSnapshot("cpu", start, start, alerts.AlertLevelWarning, "warning")
+	peer := pendingGroupSnapshot("memory", start, start, alerts.AlertLevelWarning, "peer")
+	for range 32 {
+		m.SendAlert(current)
+	}
+	m.SendAlert(peer)
+	current.LastSeen = time.Now().UTC()
+	current.Level = alerts.AlertLevelCritical
+	current.Message = "current critical"
+	current.LatestTransition.ID = "current-critical-transition"
+	m.SendAlert(current)
+	// Actual timer expiry, not a call to the private flush method.
+	wait(func() bool { return len(quietReplayRows(t, m.GetQueue())) == 2 })
+	for _, row := range quietReplayRows(t, m.GetQueue()) {
+		require.Len(t, row.Alerts, 2, "callback count leaked into persisted payload")
+		require.Equal(t, current, row.Alerts[0])
+		require.Equal(t, peer, row.Alerts[1])
+		require.Len(t, row.Links, 2)
+		require.Equal(t, current.LatestTransition.ID, row.Links[0].TransitionID)
+		require.Zero(t, row.Attempts)
+	}
+	m.Stop()
+	m = open()
+	m.StartQueueProcessing()
+	wait(func() bool {
+		stats, err := m.GetQueue().GetQueueStats()
+		require.NoError(t, err)
+		return stats["sent"] == 2
+	})
+	firingDestinations := map[string]bool{}
+	for range 2 {
+		select {
+		case msg := <-received:
+			require.True(t, msg.Grouped)
+			require.Equal(t, 2, msg.Count)
+			require.Len(t, msg.Alerts, 2)
+			require.Equal(t, current.Message, msg.Alerts[0].Message)
+			require.Equal(t, current.Level, msg.Alerts[0].Level)
+			require.Equal(t, peer.ID, msg.Alerts[1].ID)
+			require.Contains(t, []string{"/left", "/right"}, msg.destination)
+			require.False(t, firingDestinations[msg.destination], "one destination cannot stand in for another")
+			firingDestinations[msg.destination] = true
+		default:
+			t.Fatal("successful queued firing has no receiver evidence")
+		}
+	}
+	m.SendAlert(current)
+	m.mu.RLock()
+	require.Empty(t, m.pendingAlerts, "delivered occurrence must keep its ordinary cooldown")
+	m.mu.RUnlock()
+	require.False(t, m.CancelResolvedAlert(current))
+	m.SetGroupingConfig(false, 0, false, false)
+	m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: current, ResolvedTime: time.Now()})
+	wait(func() bool {
+		stats, err := m.GetQueue().GetQueueStats()
+		require.NoError(t, err)
+		return stats["sent"] == 4
+	})
+	recoveryDestinations := map[string]bool{}
+	for range 2 {
+		select {
+		case msg := <-received:
+			require.Equal(t, "resolved", msg.Event)
+			require.Len(t, msg.Alerts, 1)
+			require.Equal(t, current.ID, msg.Alerts[0].ID)
+			require.True(t, firingDestinations[msg.destination])
+			require.False(t, recoveryDestinations[msg.destination])
+			recoveryDestinations[msg.destination] = true
+		default:
+			t.Fatal("successful queued recovery has no receiver evidence")
+		}
+	}
+	m.Stop()
+	m = open()
+	for _, hook := range hooks {
+		for _, tc := range []struct {
+			a    *alerts.Alert
+			want int
+		}{{current, 0}, {peer, 1}} {
+			jobs := buildNotificationDeliveryJobs(EmailConfig{}, []WebhookConfig{hook}, AppriseConfig{}, []*alerts.Alert{tc.a}, eventResolved, time.Now())
+			require.Len(t, m.filterResolvedJobsByDeliveryReceipt(jobs), tc.want, "reopen must preserve each destination/occurrence receipt independently")
+		}
+	}
+	select {
+	case extra := <-received:
+		t.Fatalf("duplicate provider message: %+v", extra)
+	default:
+	}
+	t.Log("34 callbacks become two occurrence snapshots per destination; actual timer, persisted restart, two HTTP 200 firing deliveries and two matching recoveries pass, without duplicate receiver messages.")
+}
+
+func TestPendingFiringGroupStopCancelsTimer(t *testing.T) {
+	received := make(chan struct{}, 4)
+	server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		received <- struct{}{}
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(server.Close)
+	m := NewNotificationManagerWithDeferredQueue("", t.TempDir())
+	t.Cleanup(m.Stop)
+	m.webhookClient = server.Client()
+	require.NoError(t, m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"))
+	m.AddWebhook(WebhookConfig{ID: "ops", URL: server.URL, Enabled: true})
+	m.SetGroupingConfig(true, 1, false, false)
+	start := time.Now().UTC()
+	m.SendAlert(pendingGroupSnapshot("cpu", start, start, alerts.AlertLevelWarning, "pending at stop"))
+	m.Stop()
+	// An abandoned timer must not dispatch directly after Stop clears the queue.
+	select {
+	case <-received:
+		t.Fatal("stopped manager delivered its abandoned firing group")
+	case <-time.After(1500 * time.Millisecond):
+	}
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	require.Empty(t, m.pendingAlerts)
+	require.Nil(t, m.groupTimer)
+	require.Empty(t, m.lastNotified)
+}

internal/notifications/notifications.go

@@ -248,6 +248,7 @@ type NotificationManager struct {
 	groupWindow        time.Duration
 	groupingEnabled    bool
 	pendingAlerts      []*alerts.Alert
+	pendingAlertIndex  map[pendingFiringOccurrence]int
 	groupTimer         *time.Timer
 	pendingResolved    []*alerts.Alert
 	resolvedGroupTimer *time.Timer
@@ -1007,7 +1008,9 @@ func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNod
 	if !enabled || seconds == 0 {
 		resolved = n.takePendingResolvedLocked()
 		pending = append(pending, n.pendingAlerts...)
+		clear(n.pendingAlerts)
 		n.pendingAlerts = n.pendingAlerts[:0]
+		n.pendingAlertIndex = nil
 		if n.groupTimer != nil {
 			n.groupTimer.Stop()
 			n.groupTimer = nil
@@ -1118,10 +1121,9 @@ func (n *NotificationManager) SetEnabled(enabled bool) {
 	n.enabled = enabled
 	if !enabled {
 		n.takePendingResolvedLocked()
-		for i := range n.pendingAlerts {
-			n.pendingAlerts[i] = nil
-		}
+		clear(n.pendingAlerts)
 		n.pendingAlerts = n.pendingAlerts[:0]
+		n.pendingAlertIndex = nil
 		if n.groupTimer != nil {
 			n.groupTimer.Stop()
 			n.groupTimer = nil
@@ -1251,8 +1253,9 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp
 		Bool("inCooldown", exists).
 		Msg("alert passed cooldown check - adding to pending notifications")
 
-	// Add to pending alerts for grouping
-	n.pendingAlerts = append(n.pendingAlerts, alert)
+	// Group occurrences, not callback invocations. A repeat before delivery has
+	// no cooldown receipt yet, but must not inflate the batch's count or payload.
+	n.addPendingFiringAlertLocked(alert)
 
 	// If this is the first alert in the group, start the timer
 	if n.groupTimer == nil {
@@ -1267,6 +1270,52 @@ func (n *NotificationManager) sendAlert(alert *alerts.Alert, options alertSendOp
 	n.mu.Unlock()
 }
 
+type pendingFiringOccurrence struct {
+	id    string
+	start time.Time
+}
+
+func pendingFiringKey(alert *alerts.Alert) (pendingFiringOccurrence, bool) {
+	if alert == nil || alert.ID == "" || alert.StartTime.IsZero() {
+		// Incomplete legacy identity cannot prove that two callbacks refer to
+		// the same occurrence. Preserve both rather than silently lose one.
+		return pendingFiringOccurrence{}, false
+	}
+	// time.Time's map equality includes location and monotonic clock data;
+	// occurrence identity must use instant equality, as cancellation does.
+	return pendingFiringOccurrence{alert.ID, alert.StartTime.UTC().Round(0)}, true
+}
+
+// addPendingFiringAlertLocked takes an already-cloned snapshot under n.mu.
+// The index belongs only to the current grouping window; cancellation rebuilds
+// it lazily, and every flush/disable discards it. No delivery is recorded here.
+func (n *NotificationManager) addPendingFiringAlertLocked(alert *alerts.Alert) {
+	if key, known := pendingFiringKey(alert); known {
+		if n.pendingAlertIndex == nil {
+			n.pendingAlertIndex = make(map[pendingFiringOccurrence]int, len(n.pendingAlerts)+1)
+			for index, pending := range n.pendingAlerts {
+				if key, known := pendingFiringKey(pending); known {
+					n.pendingAlertIndex[key] = index
+				}
+			}
+		}
+		if index, exists := n.pendingAlertIndex[key]; exists {
+			previous := n.pendingAlerts[index]
+			// Keep the newest observation's complete snapshot, including its
+			// operational links. Equal-time callbacks prefer higher severity;
+			// unknown observation time cannot evict a known newer observation.
+			if previous.LastSeen.After(alert.LastSeen) ||
+				(previous.LastSeen.Equal(alert.LastSeen) && notificationSeverityRank(previous.Level) > notificationSeverityRank(alert.Level)) {
+				return
+			}
+			n.pendingAlerts[index] = alert
+			return
+		}
+		n.pendingAlertIndex[key] = len(n.pendingAlerts)
+	}
+	n.pendingAlerts = append(n.pendingAlerts, alert)
+}
+
 // Unknown levels cannot opt into the severity cooldown exception.
 func notificationSeverityRank(level alerts.AlertLevel) int {
 	switch level {
@@ -1633,6 +1682,9 @@ func (n *NotificationManager) cancelAlert(alertID string, start *time.Time) bool
 		}
 
 		n.pendingAlerts = filtered
+		if removed > 0 {
+			n.pendingAlertIndex = nil
+		}
 
 		if len(n.pendingAlerts) == 0 && n.groupTimer != nil {
 			if n.groupTimer.Stop() {
@@ -1687,7 +1739,9 @@ func (n *NotificationManager) sendGroupedAlerts() {
 	copy(alertsToSend, n.pendingAlerts)
 
 	// Clear pending alerts
+	clear(n.pendingAlerts)
 	n.pendingAlerts = n.pendingAlerts[:0]
+	n.pendingAlertIndex = nil
 	if n.groupTimer != nil {
 		n.groupTimer.Stop()
 	}
@@ -4214,6 +4268,13 @@ func (n *NotificationManager) Stop() {
 		n.mu.Lock()
 		n.enabled = false
 		n.takePendingResolvedLocked()
+		clear(n.pendingAlerts)
+		n.pendingAlerts = n.pendingAlerts[:0]
+		n.pendingAlertIndex = nil
+		if n.groupTimer != nil {
+			n.groupTimer.Stop()
+			n.groupTimer = nil
+		}
 		queue := n.queue
 		cleanupDone := n.cleanupDone
 		client := n.webhookClient