From 0dcd592ef2f8bc3a7fec0b46cc581e1d878b361f Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 02:28:48 +0100 Subject: [PATCH] Keep notification cooldowns ordered by alert occurrence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A late provider completion must retain its own firing receipt without replacing a newer occurrence’s repeat-delivery cooldown. Preserve the latest completion and highest delivered severity within one occurrence, keeping genuine new occurrences and severity increases eligible. Exercise reverse HTTP completion, duplicate admission and durable recovery receipts alongside legacy ordering controls. Change-source: pulse-maintainer --- .../v6/internal/subsystems/notifications.md | 18 ++ .../cooldown_receipt_order_test.go | 163 ++++++++++++++++++ internal/notifications/notifications.go | 25 ++- 3 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 internal/notifications/cooldown_receipt_order_test.go diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 068c67b93a..1ddb0616ee 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -15,6 +15,24 @@ ## Purpose +### Completion order cannot replace a newer occurrence's cooldown + +Firing completions still record an independent delivery receipt for their own +occurrence and destination. The reusable-ID cooldown, however, belongs to the +newest known occurrence: a late older completion or unknown legacy start cannot +evict it. Within the same occurrence, successful completion times and delivered +severity only advance. A genuinely newer occurrence begins its own severity +history. No queue schema, retry budget, recovery selection or schedule changes. + +`TestNotificationCooldownReceiptOrdering` verifies older, unknown and equal-start +completion ordering, including legacy zero starts. The connected +`TestLateFiringHTTPReceiptPreservesCurrentCooldown` holds an older HTTP response +while the newer firing completes through the persistent sender. It verifies that +the ordinary repeat adds no queue row, a current severity increase still reaches +the receiver, and both occurrence/destination receipts survive queue reopen. +This is loopback source acceptance, not native delivery, restart cooldown +persistence, a natural schedule cycle or resolution of the original flood reports. + ### Queue recovery handler ownership after reload Router monitor replacement must refresh the existing queue/DLQ handler as well diff --git a/internal/notifications/cooldown_receipt_order_test.go b/internal/notifications/cooldown_receipt_order_test.go new file mode 100644 index 0000000000..6650194479 --- /dev/null +++ b/internal/notifications/cooldown_receipt_order_test.go @@ -0,0 +1,163 @@ +package notifications + +import ( + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/stretchr/testify/require" +) + +func TestNotificationCooldownReceiptOrdering(t *testing.T) { + start := time.Date(2026, 10, 3, 1, 0, 0, 123, time.UTC) + for _, tc := range []struct { + name string + priorStart time.Time + lateStart time.Time + lateSent time.Time + }{ + {"older_occurrence", start, start.Add(-time.Nanosecond), start.Add(3 * time.Minute)}, + {"unknown_legacy_occurrence", start, time.Time{}, start.Add(3 * time.Minute)}, + {"earlier_same_occurrence_completion", start, start, start.Add(time.Minute)}, + {"earlier_legacy_completion", time.Time{}, time.Time{}, start.Add(time.Minute)}, + } { + t.Run(tc.name, func(t *testing.T) { + m := NewNotificationManagerWithDeferredQueue("", t.TempDir()) + t.Cleanup(m.Stop) + m.SetCooldown(0) // No ordinary repeats for the delivered occurrence. + m.SetGroupingConfig(true, 3600, false, false) + current := &alerts.Alert{ID: "cpu-order", Level: alerts.AlertLevelCritical, StartTime: tc.priorStart} + m.markAlertsNotified([]*alerts.Alert{current}, start.Add(2*time.Minute)) + m.mu.RLock() + before := m.lastNotified[current.ID] + m.mu.RUnlock() + late := current.Clone() + late.StartTime = tc.lateStart + late.Level = alerts.AlertLevelWarning + m.markAlertsNotified([]*alerts.Alert{nil, late}, tc.lateSent) + m.mu.RLock() + after := m.lastNotified[current.ID] + m.mu.RUnlock() + require.Equal(t, before, after, "late receipt replaced the current cooldown") + m.SendAlert(current) + m.mu.RLock() + pending := len(m.pendingAlerts) + m.mu.RUnlock() + require.Zero(t, pending, "delivered current occurrence was admitted again") + }) + } +} + +// Use the actual persistent sender and loopback receiver. Two firing callbacks +// for a reusable ID can complete in reverse occurrence order; no fabricated +// successful-delivery marker or modified scheduling clock is needed here. +func TestLateFiringHTTPReceiptPreservesCurrentCooldown(t *testing.T) { + oldArrived := make(chan struct{}, 1) + releaseOld := make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(releaseOld) }) } + var mu sync.Mutex + var bodies []string + server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read firing: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + mu.Lock() + bodies = append(bodies, string(body)) + mu.Unlock() + if strings.Contains(string(body), "older firing") { + select { + case oldArrived <- struct{}{}: + default: + } + select { + case <-releaseOld: + case <-r.Context().Done(): + return + } + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(server.Close) + dir := t.TempDir() + webhook := WebhookConfig{ID: "ops", URL: server.URL + "/hook", Enabled: true, Service: "generic"} + m := NewNotificationManagerWithDataDir("", dir) + t.Cleanup(m.Stop) + t.Cleanup(unblock) // Release an in-flight provider before waiting for shutdown. + m.webhookClient = server.Client() + require.NoError(t, m.UpdateAllowedPrivateCIDRs("127.0.0.1/32")) + m.SetGroupingConfig(false, 0, false, false) + m.SetCooldown(60) + m.AddWebhook(webhook) + queue := m.GetQueue() + old := &alerts.Alert{ID: "recurring-cpu", ResourceName: "cpu-node", Type: "cpu", + Level: alerts.AlertLevelCritical, Message: "older firing", StartTime: time.Now().UTC().Add(-time.Minute)} + m.SendAlert(old) + select { + case <-oldArrived: + case <-time.After(10 * time.Second): + t.Fatal("old firing did not reach the HTTP receiver") + } + current := old.Clone() + current.StartTime = time.Now().UTC() + current.Level = alerts.AlertLevelWarning + current.Message = "current warning firing" + m.SendAlert(current) + wait := func(condition func() bool) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for !condition() { + if time.Now().After(deadline) { + t.Fatal("notification completion timed out") + } + time.Sleep(time.Millisecond) + } + } + wait(func() bool { + m.mu.RLock() + defer m.mu.RUnlock() + return m.lastNotified[current.ID].alertStart.Equal(current.StartTime) + }) + unblock() + wait(func() bool { + stats, err := queue.GetQueueStats() + require.NoError(t, err) + return stats["sent"] == 2 + }) + // Sending synchronously admits queue work unless the cooldown rejects it. + // Assert admission itself, rather than hoping an excess send stays silent. + m.SendAlert(current) + stats, err := queue.GetQueueStats() + require.NoError(t, err) + require.Equal(t, map[string]int{"sent": 2}, stats, "late old completion admitted a duplicate current firing") + // Old critical severity must not leak into the newer warning occurrence. + current.Level = alerts.AlertLevelCritical + current.Message = "current critical firing" + m.SendAlert(current) + wait(func() bool { + stats, err := queue.GetQueueStats() + require.NoError(t, err) + return stats["sent"] == 3 + }) + mu.Lock() + count := len(bodies) + mu.Unlock() + require.Equal(t, 3, count) + m.Stop() + // The old completion still owns a durable recovery receipt, independently + // of which occurrence owns the transient repeat-delivery cooldown. + reopened := NewNotificationManagerWithDeferredQueue("", dir) + t.Cleanup(reopened.Stop) + for _, alert := range []*alerts.Alert{old, current} { + jobs := buildNotificationDeliveryJobs(EmailConfig{}, []WebhookConfig{webhook}, AppriseConfig{}, []*alerts.Alert{alert}, eventResolved, time.Now()) + require.Len(t, reopened.filterResolvedJobsByDeliveryReceipt(jobs), 1, "firing receipt lost across reopen") + } + t.Log("Two reverse-order HTTP 200 completions retain the current cooldown; repeat adds no row, severity increase delivers, both occurrence receipts survive queue reopen.") +} diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 7b93dbf590..d508f53bbb 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -1291,14 +1291,29 @@ func (n *NotificationManager) markAlertsNotified(alertsToSend []*alerts.Alert, s continue } level := alert.Level - // Destinations and retries can complete out of order. A late warning - // receipt must not re-enable critical delivery for this occurrence. - if previous, ok := n.lastNotified[alert.ID]; ok && previous.alertStart.Equal(alert.StartTime) && notificationSeverityRank(previous.level) > notificationSeverityRank(level) { - level = previous.level + lastSent := sentAt + if previous, ok := n.lastNotified[alert.ID]; ok { + // Destinations and retries can complete out of order. An older + // occurrence still owns its delivery receipt, but must not replace + // the current occurrence's repeat-delivery cooldown. Unknown legacy + // starts likewise cannot evict a known occurrence. + if previous.alertStart.After(alert.StartTime) { + continue + } + if previous.alertStart.Equal(alert.StartTime) { + // A late warning cannot re-enable critical delivery, and an + // earlier completion cannot shorten the same occurrence's hold. + if notificationSeverityRank(previous.level) > notificationSeverityRank(level) { + level = previous.level + } + if previous.lastSent.After(lastSent) { + lastSent = previous.lastSent + } + } } n.lastNotified[alert.ID] = notificationRecord{ level: level, - lastSent: sentAt, + lastSent: lastSent, alertStart: alert.StartTime, } }