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, } }