From ac995ef565eddeab517a5d787daf1d28313bf60d Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 03:10:14 +0100 Subject: [PATCH] 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 --- .../v6/internal/subsystems/notifications.md | 27 ++ .../firing_grouping_occurrence_test.go | 359 ++++++++++++++++++ internal/notifications/notifications.go | 71 +++- 3 files changed, 452 insertions(+), 5 deletions(-) create mode 100644 internal/notifications/firing_grouping_occurrence_test.go diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 2903cb9f5a..ae96dfcbce 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/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 diff --git a/internal/notifications/firing_grouping_occurrence_test.go b/internal/notifications/firing_grouping_occurrence_test.go new file mode 100644 index 0000000000..6544adce92 --- /dev/null +++ b/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) +} diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index d508f53bbb..1741712f9d 100644 --- a/internal/notifications/notifications.go +++ b/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