diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 1783a9637..38cbd4a5d 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -2908,3 +2908,18 @@ real history hook and administration card: load a row, start a pending range read, confirm clear, then release the obsolete response at desktop and phone widths. Scripted API responses establish component behaviour, not installed backend deletion or destination delivery. + +### Escalation callback admission uses current policy and occurrence + +`PrepareEscalationNotification` rejects asynchronous snapshots after escalation +or global alert disablement, inactive activation, recovery, acknowledgement, +snooze, or replacement by another occurrence of the same alert ID. It returns +the current alert and detached current per-level routing, not mutable manager +state. The monitoring callback must use this admission before enqueueing. +This check does not recall provider-accepted messages or make the subsequent +notification enqueue atomic with alert resolution. + +`escalation_policy_change_test.go` pins a short-lived alert and disabled policy +before the former 180-minute deadline, plus current payload/routing selection, +copy isolation, snooze and invalid-level rejection. These deterministic component tests +establish neither installed delivery timing nor the cause of a delayed email. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 1514be7b2..20e26637a 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4008,3 +4008,13 @@ without a request consumer, and after acceptance, plus late completion and success/error propagation using channel-only fixtures without starting monitors or storage. This is shutdown handshake proof, not installed notification receipt or full server reload acceptance. + +### Stale escalation callback snapshots are not delivery authority + +`handleAlertEscalated` obtains current occurrence/policy admission from the +alert manager before dispatch. `monitor_escalation_stale_test.go` invokes the +callback after disablement, inactive activation, recovery, recurrence, +and acknowledgement, requiring no queued notification. Existing routing, +quiet-hours and cooldown tests now use real active manager occurrences rather +than fabricated inactive alerts. This is callback-boundary proof, not a claim +that already admitted or provider-accepted deliveries can be recalled. diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 1a548abe6..01c107521 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -899,3 +899,14 @@ wraps its returned error and checks both transport retry policy and the shared class predicate used by the queue. `TestClassFromHTTPStatus` pins the reason classes. This proves the classification boundary without starting queue/storage workers; it does not establish installed receipt or queue scheduling execution. + +### Initial routing edits and previously queued firing work + +Changing the initial target affects new sends; it does not rewrite a previously +queued email or revoke independent per-level escalation routing. Queue delivery +uses the admitted destination configuration while checking current global and +destination enablement. Resolution independently cancels obsolete firing work. +`queue_policy_change_test.go` verifies retained email routing after a webhook +initial-target edit with an in-memory SMTP acceptance, and prevents a previously +fetched pending item from sending after cancellation. This does not establish +recipient inbox timing or recall mail already accepted by a provider. diff --git a/internal/alerts/escalation.go b/internal/alerts/escalation.go index cb6c7973f..5edc3920d 100644 --- a/internal/alerts/escalation.go +++ b/internal/alerts/escalation.go @@ -114,3 +114,28 @@ func (m *Manager) checkEscalations() { m.safeCallEscalationRepeatCallback(alert, len(escalation.Levels)) } } + +// PrepareEscalationNotification revalidates an asynchronously dispatched +// escalation against the current policy and occurrence. The callback snapshot +// is not authority to send after recovery, acknowledgement or a policy edit. +// This does not recall a notification already handed to a destination. +func (m *Manager) PrepareEscalationNotification(snapshot *Alert, level int) (*Alert, EscalationLevel, bool) { + if m == nil || snapshot == nil { + return nil, EscalationLevel{}, false + } + m.mu.Lock() + defer m.mu.Unlock() + if !m.config.Enabled || m.config.ActivationState != ActivationActive || !m.config.Schedule.Escalation.Enabled || level <= 0 || level > len(m.config.Schedule.Escalation.Levels) { + return nil, EscalationLevel{}, false + } + active, ok := m.getActiveAlertNoLock(snapshot.ID) + if !ok || active == nil || !active.StartTime.Equal(snapshot.StartTime) || active.Acknowledged { + return nil, EscalationLevel{}, false + } + if _, snoozed := alertSnoozeUntil(active, m.policyNow()); snoozed { + return nil, EscalationLevel{}, false + } + target := m.config.Schedule.Escalation.Levels[level-1] + target.DestinationIDs = append([]string(nil), target.DestinationIDs...) + return cloneAlertForOutput(active), target, true +} diff --git a/internal/alerts/escalation_policy_change_test.go b/internal/alerts/escalation_policy_change_test.go new file mode 100644 index 000000000..1e503084d --- /dev/null +++ b/internal/alerts/escalation_policy_change_test.go @@ -0,0 +1,80 @@ +package alerts + +import ( + "testing" + "time" +) + +func TestEscalationPolicyChangeBeforeFormerDeadline(t *testing.T) { + for _, action := range []string{"disable", "resolve"} { + t.Run(action, func(t *testing.T) { + m := newTestManager(t) + now := time.Date(2026, 9, 10, 1, 0, 0, 0, time.UTC) + m.now = func() time.Time { return now } + cfg := m.GetConfig() + cfg.Enabled = true + cfg.ActivationState = ActivationActive + cfg.Schedule.Escalation = EscalationConfig{Enabled: true, Levels: []EscalationLevel{{After: 180, Notify: "email"}}} + m.UpdateConfig(cfg) + a := &Alert{ID: "short-lived", StartTime: now, Level: AlertLevelWarning} + m.mu.Lock() + m.setActiveAlertNoLock(a.ID, a) + m.mu.Unlock() + now = now.Add(time.Minute) + m.checkEscalations() + if action == "disable" { + cfg.Schedule.Escalation.Enabled = false + m.UpdateConfig(cfg) + } else if !m.ClearAlert(a.ID) { + t.Fatal("clear failed") + } + now = now.Add(180 * time.Minute) + m.checkEscalations() + // checkEscalations records admission synchronously before spawning delivery. + if a.LastEscalation != 0 || len(a.EscalationTimes) != 0 { + t.Fatalf("obsolete escalation admitted: %+v", a) + } + }) + } +} + +func TestPrepareEscalationNotificationUsesCurrentOccurrenceAndDetachedTarget(t *testing.T) { + m := newTestManager(t) + now := time.Now() + m.mu.Lock() + m.config.Enabled = true + m.config.ActivationState = ActivationActive + m.config.Schedule.Escalation = EscalationConfig{Enabled: true, Levels: []EscalationLevel{{After: 180, Notify: "email", DestinationIDs: []string{"email"}}}} + a := &Alert{ID: "current-escalation", Type: "cpu", ResourceID: "node/test", StartTime: now, Level: AlertLevelWarning, Value: 90} + m.setActiveAlertNoLock(a.ID, a) + snapshot := cloneAlertForOutput(a) + a.Value = 95 + m.config.Schedule.Escalation.Levels[0].Notify = "webhook" + m.config.Schedule.Escalation.Levels[0].DestinationIDs = []string{"webhook:ops"} + m.mu.Unlock() + current, target, ok := m.PrepareEscalationNotification(snapshot, 1) + if !ok || current.Value != 95 || target.Notify != "webhook" || target.DestinationIDs[0] != "webhook:ops" { + t.Fatalf("current=%+v target=%+v ok=%v", current, target, ok) + } + current.Value = 0 + target.DestinationIDs[0] = "mutated" + current, target, ok = m.PrepareEscalationNotification(snapshot, 1) + if !ok || current.Value != 95 || target.DestinationIDs[0] != "webhook:ops" { + t.Fatal("returned data aliases manager state") + } + for _, level := range []int{0, -1, 2} { + if _, _, ok := m.PrepareEscalationNotification(snapshot, level); ok { + t.Fatalf("invalid level %d admitted", level) + } + } + if _, _, ok := m.PrepareEscalationNotification(nil, 1); ok { + t.Fatal("nil snapshot admitted") + } + if err := m.SnoozeAlert(snapshot.ID, "test", time.Now().Add(time.Hour)); err != nil { + t.Fatal(err) + } + if _, _, ok := m.PrepareEscalationNotification(snapshot, 1); ok { + t.Fatal("snoozed occurrence admitted") + } + +} diff --git a/internal/monitoring/monitor_alert_handling_test.go b/internal/monitoring/monitor_alert_handling_test.go index d8ae378fa..4185215ea 100644 --- a/internal/monitoring/monitor_alert_handling_test.go +++ b/internal/monitoring/monitor_alert_handling_test.go @@ -957,17 +957,7 @@ func TestMonitor_HandleAlertEscalated_QuietHoursSuppressesNotification(t *testin alertManager: mgr, } - alert := &alerts.Alert{ - ID: "escalated-offline", - Type: "connectivity", - Level: alerts.AlertLevelCritical, - ResourceID: "node/pve-1", - ResourceName: "pve-1", - Node: "pve-1", - Instance: "pve", - Message: "Node offline", - StartTime: time.Now(), - } + alert := activeEscalationFixture(t, mgr, "connectivity", alerts.AlertLevelCritical) m.handleAlertEscalated(nil, alert, 1) @@ -1013,17 +1003,7 @@ func TestMonitor_HandleAlertEscalated_SendsNotificationWhenNotSuppressed(t *test alertManager: mgr, } - alert := &alerts.Alert{ - ID: "escalated-normal", - Type: "connectivity", - Level: alerts.AlertLevelCritical, - ResourceID: "node/pve-1", - ResourceName: "pve-1", - Node: "pve-1", - Instance: "pve", - Message: "Node offline", - StartTime: time.Now(), - } + alert := activeEscalationFixture(t, mgr, "connectivity", alerts.AlertLevelCritical) m.handleAlertEscalated(nil, alert, 1) @@ -1069,17 +1049,7 @@ func TestMonitor_HandleAlertEscalated_BypassesDeliveryCooldown(t *testing.T) { alertManager: mgr, } - alert := &alerts.Alert{ - ID: "escalated-with-cooldown", - Type: "memory", - Level: alerts.AlertLevelWarning, - ResourceID: "vm/100", - ResourceName: "vm-100", - Node: "pve-1", - Instance: "pve", - Message: "Memory threshold crossed", - StartTime: time.Now().Add(-10 * time.Minute), - } + alert := activeEscalationFixture(t, mgr, "memory", alerts.AlertLevelWarning) notifMgr.SendAlert(alert) select { diff --git a/internal/monitoring/monitor_alert_override_migration_test.go b/internal/monitoring/monitor_alert_override_migration_test.go index 3dbe71cd7..87aca1c90 100644 --- a/internal/monitoring/monitor_alert_override_migration_test.go +++ b/internal/monitoring/monitor_alert_override_migration_test.go @@ -62,10 +62,7 @@ func TestHandleAlertEscalatedPreservesLegacyAppriseAndExactWebhookRouting(t *tes cfg.Schedule.QuietHours.Enabled = false manager.UpdateConfig(cfg) - (&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, &alerts.Alert{ - ID: "apprise-escalation", Type: "connectivity", Level: alerts.AlertLevelCritical, - ResourceID: "node/pve-1", ResourceName: "pve-1", StartTime: time.Now(), - }, 1) + (&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, activeEscalationFixture(t, manager, "connectivity", alerts.AlertLevelCritical), 1) select { case <-requests: @@ -106,10 +103,7 @@ func TestHandleAlertEscalatedPreservesLegacyAppriseAndExactWebhookRouting(t *tes cfg.Schedule.QuietHours.Enabled = false manager.UpdateConfig(cfg) - (&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, &alerts.Alert{ - ID: "exact-webhook-escalation", Type: "connectivity", Level: alerts.AlertLevelCritical, - ResourceID: "node/pve-1", ResourceName: "pve-1", StartTime: time.Now(), - }, 1) + (&Monitor{notificationMgr: notifMgr, alertManager: manager}).handleAlertEscalated(nil, activeEscalationFixture(t, manager, "connectivity", alerts.AlertLevelCritical), 1) select { case <-selectedRequests: diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index e106a7a7a..df5209e7e 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -280,11 +280,13 @@ func (m *Monitor) handleAlertEscalated(hub *websocket.Hub, alert *alerts.Alert, Int("level", level). Msg("Alert escalated") - config := m.alertManager.GetConfig() - if level <= 0 || level > len(config.Schedule.Escalation.Levels) { + current, escalationLevel, eligible := m.alertManager.PrepareEscalationNotification(alert, level) + if !eligible { return } + alert = current + if m.alertManager.ShouldSuppressNotification(alert) { log.Info(). Str("alertID", alert.ID). @@ -295,7 +297,6 @@ func (m *Monitor) handleAlertEscalated(hub *websocket.Hub, alert *alerts.Alert, } if m.notificationMgr != nil { - escalationLevel := config.Schedule.Escalation.Levels[level-1] if len(escalationLevel.DestinationIDs) > 0 { m.notificationMgr.SendEscalatedAlertToDestinations(alert, escalationLevel.Notify, escalationLevel.DestinationIDs) m.broadcastEscalatedAlert(hub, alert) diff --git a/internal/monitoring/monitor_escalation_stale_test.go b/internal/monitoring/monitor_escalation_stale_test.go new file mode 100644 index 000000000..02d6635d2 --- /dev/null +++ b/internal/monitoring/monitor_escalation_stale_test.go @@ -0,0 +1,83 @@ +package monitoring + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/notifications" +) + +func activeEscalationFixture(t *testing.T, m *alerts.Manager, kind string, level alerts.AlertLevel) *alerts.Alert { + t.Helper() + t.Cleanup(m.Stop) + cfg := m.GetConfig() + cfg.Enabled = true + cfg.ActivationState = alerts.ActivationActive + cfg.Schedule.Escalation.Enabled = true + m.UpdateConfig(cfg) + m.RaiseSystemAlert(alerts.SystemAlertInput{Type: kind, Level: level, Message: "escalation fixture"}) + active := m.GetActiveAlerts() + if len(active) != 1 { + t.Fatalf("active fixtures = %d", len(active)) + } + return active[0].Clone() +} + +// The callback carries a snapshot and can run after a config edit or recovery. +// Invoke it after that boundary, rather than relying on goroutine timing. +func TestHandleAlertEscalatedRejectsStaleAdmission(t *testing.T) { + for _, action := range []string{"disabled", "inactive", "alerts-disabled", "resolved", "new-occurrence", "acknowledged"} { + t.Run(action, func(t *testing.T) { + t.Setenv("PULSE_DATA_DIR", t.TempDir()) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) + defer server.Close() + n := notifications.NewNotificationManager("https://pulse.example.test") + defer n.Stop() + if err := n.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + n.AddWebhook(notifications.WebhookConfig{ID: "test", URL: server.URL, Enabled: true}) + manager := alerts.NewManager() + cfg := manager.GetConfig() + cfg.Schedule.Escalation.Levels = []alerts.EscalationLevel{{After: 180, Notify: "webhook"}} + cfg.Schedule.QuietHours.Enabled = false + manager.UpdateConfig(cfg) + snapshot := activeEscalationFixture(t, manager, "stale-escalation", alerts.AlertLevelCritical) + cfg = manager.GetConfig() + switch action { + case "disabled": + cfg.Schedule.Escalation.Enabled = false + manager.UpdateConfig(cfg) + case "inactive": + cfg.ActivationState = alerts.ActivationPending + manager.UpdateConfig(cfg) + case "alerts-disabled": + cfg.Enabled = false + manager.UpdateConfig(cfg) + case "resolved", "new-occurrence": + if !manager.ClearAlert(snapshot.ID) { + t.Fatal("clear failed") + } + if action == "new-occurrence" { + activeEscalationFixture(t, manager, "stale-escalation", alerts.AlertLevelCritical) + } + case "acknowledged": + if err := manager.AcknowledgeAlert(snapshot.ID, "test"); err != nil { + t.Fatal(err) + } + } + (&Monitor{alertManager: manager, notificationMgr: n}).handleAlertEscalated(nil, snapshot, 1) + stats, err := n.GetQueue().GetQueueStats() + if err != nil { + t.Fatal(err) + } + for status, count := range stats { + if count != 0 { + t.Errorf("stale callback queued %s=%d", status, count) + } + } + }) + } +} diff --git a/internal/notifications/queue_policy_change_test.go b/internal/notifications/queue_policy_change_test.go new file mode 100644 index 000000000..c3b3c4b4f --- /dev/null +++ b/internal/notifications/queue_policy_change_test.go @@ -0,0 +1,54 @@ +package notifications + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" +) + +// Initial routing changes apply to new sends, not already admitted work. +// Resolution, independently, must retire that old firing work before retry. +func TestQueuedEmailRoutingChangeAndResolution(t *testing.T) { + for _, resolve := range []bool{false, true} { + name := "still-active" + if resolve { + name = "resolved" + } + t.Run(name, func(t *testing.T) { + var deliveries int32 + stubSMTPDialSuccess(t, &deliveries) + q, err := NewNotificationQueue(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer q.Stop() + n := &NotificationManager{enabled: true, queue: q, initialTarget: notificationDeliveryTargetEmail, + emailConfig: EmailConfig{Enabled: true, SMTPHost: "smtp.example.com", SMTPPort: 25, From: "pulse@example.test", To: []string{"ops@example.test"}}, + lastNotified: make(map[string]notificationRecord), deliveryReceipts: make(map[string]struct{}), + } + alert := &alerts.Alert{ID: "routing-change", ResourceName: "test-node", Level: alerts.AlertLevelWarning, StartTime: time.Now().Add(-time.Minute)} + n.SendAlert(alert) + pending, err := q.GetPending(10) + if err != nil || len(pending) != 1 || pending[0].Type != "email" { + t.Fatalf("queued email = %v, %v", pending, err) + } + old := pending[0] + n.SetInitialNotifyTarget("webhook") + // A fetched worker item must not defeat resolution cancellation. + if resolve { + n.CancelAlert(alert.ID) + } + q.SetProcessor(n.ProcessQueuedNotification) + q.processNotification(old) + want := int32(1) + if resolve { + want = 0 + } + if got := atomic.LoadInt32(&deliveries); got != want { + t.Fatalf("SMTP acceptances = %d, want %d", got, want) + } + }) + } +}