From bea763bc4c110d0c38ffd83ca490ae698ccd08a2 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:07:20 +0100 Subject: [PATCH] Revalidate restored backup alerts before startup delivery Restored critical backup alerts could notify before the first successful backup check removed obsolete rollup identities. Hold firing and escalation until valid backup evidence resolves or confirms the retained occurrence. Confirmed overdue backups resume normal notification policy. Cover JSON and SQLite restore, delayed or invalid evidence, overdue delivery, acknowledgement, snooze, cooldown and quiet-hours replay in the registered canonical stateful regression proof. Contract-Neutral: Correct startup notification ordering without changing alert truth, identity, restart authority, or delivery API. --- internal/alerts/active_cleanup.go | 1 + internal/alerts/active_lifecycle.go | 3 + internal/alerts/active_persistence.go | 10 +- internal/alerts/canonical_lifecycle.go | 10 +- internal/alerts/canonical_stateful_test.go | 311 +++++++++++++++++++++ internal/alerts/escalation.go | 6 + internal/alerts/manager.go | 3 + internal/alerts/notification_policy.go | 33 +++ 8 files changed, 374 insertions(+), 3 deletions(-) diff --git a/internal/alerts/active_cleanup.go b/internal/alerts/active_cleanup.go index 9d0a19af68..ea78202cf7 100644 --- a/internal/alerts/active_cleanup.go +++ b/internal/alerts/active_cleanup.go @@ -341,6 +341,7 @@ func (m *Manager) ClearActiveAlerts() { return } m.activeAlerts = make(map[string]*Alert) + m.restoredBackupNotifications = nil m.activeAlertAlias = make(map[string]string) m.core.Reset() m.intentPending = make(map[string]IntentPendingState) diff --git a/internal/alerts/active_lifecycle.go b/internal/alerts/active_lifecycle.go index 3771f98205..c2d150d324 100644 --- a/internal/alerts/active_lifecycle.go +++ b/internal/alerts/active_lifecycle.go @@ -663,6 +663,9 @@ func (m *Manager) removeActiveAlertNoLock(alertID string) { } if exists { delete(m.activeAlerts, key) + if currentAlert != nil { + delete(m.restoredBackupNotifications, canonicalTrackingKeyForAlert(currentAlert)) + } m.removeActiveRecoveryAlert(currentAlert, key) } diff --git a/internal/alerts/active_persistence.go b/internal/alerts/active_persistence.go index d5c0c122a1..66c7e35f25 100644 --- a/internal/alerts/active_persistence.go +++ b/internal/alerts/active_persistence.go @@ -431,6 +431,7 @@ func (m *Manager) restoreActiveAlertSnapshots(alerts []*Alert, source string, re m.mu.Lock() if replace { m.activeAlerts = make(map[string]*Alert) + m.restoredBackupNotifications = nil m.activeAlertAlias = make(map[string]string) m.ackState = make(map[string]ackRecord) m.ackStateByCanonical = make(map[string]ackRecord) @@ -526,7 +527,14 @@ func (m *Manager) restoreActiveAlertSnapshots(alerts []*Alert, source string, re restoredCount++ if alert.Level == AlertLevelCritical { - restoredCritical = append(restoredCritical, alert.Clone()) + if alert.Type == "backup-age" { + if m.restoredBackupNotifications == nil { + m.restoredBackupNotifications = make(map[string]time.Time) + } + m.restoredBackupNotifications[canonicalTrackingKeyForAlert(alert)] = alert.StartTime + } else { + restoredCritical = append(restoredCritical, alert.Clone()) + } } } if err := m.loadIntentPendingNoLock(); err != nil { diff --git a/internal/alerts/canonical_lifecycle.go b/internal/alerts/canonical_lifecycle.go index b6f8386273..f4a7c614f4 100644 --- a/internal/alerts/canonical_lifecycle.go +++ b/internal/alerts/canonical_lifecycle.go @@ -823,8 +823,14 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa return result, true } - if primary == reducer.EventSeverityChanged && params.NotifyOnSeverityChange { - if params.AddToHistoryOnSeverityChange { + // Restored backup incidents do not dispatch on a startup timer. Their + // first valid overdue evaluation must resume delivery even when the + // reducer keeps the same occurrence and severity. + backupRevalidated := m.revalidateBackupNotificationNoLock(alert) + severityChanged := primary == reducer.EventSeverityChanged && params.NotifyOnSeverityChange + if severityChanged || + (backupRevalidated && m.shouldNotifyAfterCooldown(alert)) { + if severityChanged && params.AddToHistoryOnSeverityChange { m.historyManager.AddAlertTransition(*alert) } if params.RateLimit && !m.checkRateLimit(trackingKey) { diff --git a/internal/alerts/canonical_stateful_test.go b/internal/alerts/canonical_stateful_test.go index d76ced7073..d5e7118854 100644 --- a/internal/alerts/canonical_stateful_test.go +++ b/internal/alerts/canonical_stateful_test.go @@ -1,12 +1,14 @@ package alerts import ( + "fmt" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog" alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" "github.com/rcourtman/pulse-go-rewrite/internal/storagehealth" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" @@ -448,3 +450,312 @@ func TestCheckDiskHealthWearoutFlappingDoesNotRepeatNotifications(t *testing.T) t.Fatalf("genuine recurrence produced %d total dispatches, want two", got) } } + +// Replays obsolete PVE backup identities and timestamps. The live +// PBS inventory is deliberately evaluated after the production restore timer, +// matching a restart whose first successful backup check is still pending. +func TestRestoredBackupRequiresFreshEvaluation(t *testing.T) { + for _, sqlite := range []bool{false, true} { + name := "recovery-mirror" + if sqlite { + name = "sqlite-authority" + } + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + now := time.Now().UTC() + lastSeen := now.Add(-4 * 24 * time.Hour) + var fixtures []*Alert + for i, subject := range []string{"res-vm-legacy-1", "res-vm-legacy-2"} { + resourceID := "backup-subject:" + subject + specID := resourceID + "-backup-age" + backupTime := now.Add(-94*24*time.Hour + time.Duration(i)*15*time.Minute) + fixtures = append(fixtures, &Alert{ + ID: resourceID + "::" + specID, Type: "backup-age", Level: AlertLevelCritical, + ResourceID: resourceID, ResourceName: fmt.Sprintf("backup-server-%d backup", i+1), + CanonicalSpecID: specID, CanonicalKind: "posture-threshold", + CanonicalState: resourceID + "::" + specID, + Message: "Backup via proxmox-pve is 90.0 days old", Value: 90, Threshold: 84, + StartTime: backupTime.Add(84 * 24 * time.Hour), LastSeen: lastSeen, + Metadata: map[string]interface{}{ + "source": "PVE", "providers": []string{"proxmox-pve"}, + "lastBackupTime": backupTime, "ageDays": float64(90), "thresholdDays": float64(84), + "guestVmid": 0, "guestInstance": "", "guestNode": "", "guestType": "", + "guestName": fmt.Sprintf("backup-server-%d", i+1), "orphaned": false, + "resourceType": "vm", "rollupId": "res:" + subject[4:], + "canonicalAlertKind": "posture-threshold", "canonicalSpecID": specID, + }, + }) + } + writeActiveRecoveryFixture(t, dir, fixtures) + seed := NewManagerWithDataDir(dir) + if sqlite { + seed.EnableEventLog() + if !seed.activeStateAuthoritative.Load() { + t.Fatal("failed to establish SQLite authority") + } + } + if err := seed.SaveActiveAlerts(); err != nil { + t.Fatal(err) + } + seed.Stop() + + m := NewManagerWithDataDir(dir) + t.Cleanup(m.Stop) + if sqlite { + m.EnableEventLog() + } + m.UpdateConfig(AlertConfig{Enabled: true, ActivationState: ActivationActive, + TimeThresholds: map[string]int{}, + BackupDefaults: BackupAlertConfig{Enabled: true, WarningDays: 70, CriticalDays: 84}}) + delivered := make(chan *Alert, 8) + m.SetAlertCallback(func(a *Alert) { delivered <- a.Clone() }) + for _, a := range fixtures { + if !testHasActiveAlert(t, m, a.ID) { + t.Fatalf("the historical alert was not restored: %s", a.ID) + } + } + + // Exercise the real ten-second startup timer without changing source. + select { + case a := <-delivered: + t.Errorf("restored stale backup warning dispatched before any fresh backup evaluation: id=%s lastSeen=%s lastBackupTime=%v", + a.ID, a.LastSeen.Format(time.RFC3339), a.Metadata["lastBackupTime"]) + case <-time.After(12 * time.Second): + } + + // The current PBS rollups use different resource identities. + var rollups []recovery.ProtectionRollup + guests := make(map[string]GuestLookup) + byVMID := make(map[string][]GuestLookup) + for i, resourceID := range []string{"vm-current-1", "vm-current-2"} { + vmid := 101 + i + node := fmt.Sprintf("pve%02d", i+1) + guest := GuestLookup{Name: fmt.Sprintf("backup-server-%d", i+1), Instance: "pve01", Node: node, + VMID: vmid, Type: "qemu", ResourceID: resourceID} + guests[BuildGuestKey("pve01", node, vmid)] = guest + byVMID[fmt.Sprint(vmid)] = []GuestLookup{guest} + success := now.Add(-2*24*time.Hour + time.Duration(i)*15*time.Minute) + rollups = append(rollups, recovery.ProtectionRollup{RollupID: "res:" + resourceID, + SubjectRef: &recovery.ExternalRef{Type: "proxmox-vm", Namespace: "pve01", Name: guest.Name, + ID: fmt.Sprintf("pve01:%s:%d", node, vmid), Class: node}, + LastSuccessAt: &success, LastOutcome: recovery.OutcomeSuccess, + Providers: []recovery.Provider{recovery.ProviderProxmoxPBS}}) + } + m.CheckBackups(rollups, guests, byVMID) + for _, a := range fixtures { + if testHasActiveAlert(t, m, a.ID) { + t.Errorf("fresh current PBS inventory did not clear historical warning: %s", a.ID) + } + } + select { + case a := <-delivered: + t.Fatalf("cleared historical backup emitted a notification: %s", a.ID) + case <-time.After(100 * time.Millisecond): + } + }) + } +} + +func startupBackupConfig() AlertConfig { + return AlertConfig{Enabled: true, ActivationState: ActivationActive, + TimeThresholds: map[string]int{}, + BackupDefaults: BackupAlertConfig{Enabled: true, WarningDays: 70, CriticalDays: 84}} +} + +// Seed through normal backup evaluation so revalidation must retain the same +// canonical identity, acknowledgement, occurrence, and notification receipt. +func restartOverdueBackup(t *testing.T, sqlite bool, edit func(*Alert)) (*Manager, []recovery.ProtectionRollup, *Alert) { + t.Helper() + dir := t.TempDir() + success := time.Now().Add(-94 * 24 * time.Hour) + rollups := []recovery.ProtectionRollup{{RollupID: "res:vm-startup-test", + SubjectRef: &recovery.ExternalRef{Type: "proxmox-vm", Namespace: "test", ID: "test:node:101"}, + LastSuccessAt: &success, LastOutcome: recovery.OutcomeSuccess, + Providers: []recovery.Provider{recovery.ProviderProxmoxPBS}}} + seed := NewManagerWithDataDir(dir) + if sqlite { + seed.EnableEventLog() + } + seed.UpdateConfig(startupBackupConfig()) + seed.CheckBackups(rollups, nil, nil) + active := seed.GetActiveAlerts() + if len(active) != 1 || active[0].Level != AlertLevelCritical { + seed.Stop() + t.Fatalf("expected one overdue backup incident, got %+v", active) + } + if edit != nil { + seed.mu.Lock() + current, _ := seed.getActiveAlertNoLock(active[0].ID) + edit(current) + seed.setActiveAlertNoLock(current.ID, current) + seed.mu.Unlock() + } + if err := seed.SaveActiveAlerts(); err != nil { + seed.Stop() + t.Fatal(err) + } + seed.Stop() + m := NewManagerWithDataDir(dir) + t.Cleanup(m.Stop) + if sqlite { + m.EnableEventLog() + if !m.activeStateAuthoritative.Load() { + t.Fatal("failed to restore SQLite active authority") + } + } + m.UpdateConfig(startupBackupConfig()) + restored := m.GetActiveAlerts() + if len(restored) != 1 { + t.Fatalf("expected retained overdue backup incident, got %+v", restored) + } + return m, rollups, restored[0].Clone() +} + +func TestRestoredBackupRevalidationDeliversOverdueOccurrence(t *testing.T) { + for _, sqlite := range []bool{false, true} { + t.Run(fmt.Sprintf("sqlite=%t", sqlite), func(t *testing.T) { + m, rollups, restored := restartOverdueBackup(t, sqlite, nil) + delivered := make(chan *Alert, 4) + m.SetAlertCallback(func(a *Alert) { delivered <- a }) + if !m.ShouldSuppressNotification(restored) { + t.Fatal("restored snapshot can notify before a backup evaluation") + } + m.CheckBackups(rollups, nil, nil) + select { + case a := <-delivered: + if a.ID != restored.ID || !a.StartTime.Equal(restored.StartTime) || !a.LastSeen.After(restored.LastSeen) { + t.Fatalf("revalidation did not preserve the occurrence with fresh evidence: %+v", a) + } + case <-time.After(2 * time.Second): + t.Fatal("genuinely overdue backup did not notify after revalidation") + } + m.CheckBackups(rollups, nil, nil) + select { + case a := <-delivered: + t.Fatalf("unchanged overdue occurrence notified twice: %s", a.ID) + case <-time.After(100 * time.Millisecond): + } + if active := m.GetActiveAlerts(); len(active) != 1 || active[0].LastNotified == nil { + t.Fatalf("revalidated notification receipt was not retained: %+v", active) + } + }) + } +} + +func TestRestoredBackupWithoutValidEvidenceRetainsIncidentAndEscalation(t *testing.T) { + m, rollups, restored := restartOverdueBackup(t, true, nil) + delivered := make(chan *Alert, 4) + m.SetAlertCallback(func(a *Alert) { delivered <- a }) + m.SetEscalateCallback(func(a *Alert, _ int) { delivered <- a }) + cfg := startupBackupConfig() + cfg.Schedule.Escalation = EscalationConfig{Enabled: true, Levels: []EscalationLevel{{After: 5}}} + m.UpdateConfig(cfg) + spec, err := buildCanonicalPostureThresholdSpec(restored.CanonicalSpecID, restored.ResourceID, + restored.ResourceName, "vm", "backup-age-days", 70, 84, "", 0, 0, false) + if err != nil { + t.Fatal(err) + } + if _, valid := m.evaluateCanonicalStatefulAlert(canonicalStatefulAlertParams{ + Spec: spec, AlertID: restored.ID, AlertType: "backup-age", + Evidence: alertspecs.AlertEvidence{ObservedAt: time.Now()}, + }); valid { + t.Fatal("invalid backup evidence was accepted") + } + m.mu.Lock() + if m.dispatchAlert(restored, false) { + m.mu.Unlock() + t.Fatal("restored snapshot dispatched without valid evidence") + } + m.mu.Unlock() + m.checkEscalations() + if _, _, eligible := m.PrepareEscalationNotification(restored, 1); eligible { + t.Fatal("restored backup snapshot bypassed revalidation through escalation") + } + diagnosis, found := m.DiagnoseAlertDelivery(restored.ID) + if !found || diagnosis.Reason != alertDeliveryReasonBackupRevalidation { + t.Fatalf("delivery diagnosis did not explain the startup hold: %+v", diagnosis) + } + active := m.GetActiveAlerts() + if len(active) != 1 || !active[0].StartTime.Equal(restored.StartTime) || active[0].LastEscalation != 0 || active[0].LastNotified != nil { + t.Fatalf("missing evidence mutated or resolved the retained incident: %+v", active) + } + select { + case <-delivered: + t.Fatal("missing evidence sent a firing or escalation notification") + case <-time.After(100 * time.Millisecond): + } + // Once collection succeeds, neither the overdue notification nor the + // previously held escalation level may be permanently lost. + m.CheckBackups(rollups, nil, nil) + select { + case <-delivered: + case <-time.After(2 * time.Second): + t.Fatal("overdue notification was lost after collection succeeded") + } + m.checkEscalations() + if _, _, eligible := m.PrepareEscalationNotification(restored, 1); !eligible { + t.Fatal("freshly revalidated backup cannot escalate") + } +} + +func TestRestoredBackupRevalidationRespectsAcknowledgementAndCooldown(t *testing.T) { + for _, policy := range []string{"acknowledged", "cooldown", "first-notification-only", "snoozed"} { + t.Run(policy, func(t *testing.T) { + m, rollups, restored := restartOverdueBackup(t, false, func(a *Alert) { + if policy == "acknowledged" { + a.Acknowledged = true + a.AckUser = "operator" + } else if policy != "snoozed" { + now := time.Now() + a.LastNotified = &now + } + }) + if policy == "snoozed" { + if err := m.SnoozeAlert(restored.ID, "operator", time.Now().Add(time.Hour)); err != nil { + t.Fatal(err) + } + } + cfg := startupBackupConfig() + if policy == "cooldown" { + cfg.Schedule.Cooldown = 60 + } + m.UpdateConfig(cfg) + delivered := make(chan *Alert, 2) + m.SetAlertCallback(func(a *Alert) { delivered <- a }) + m.CheckBackups(rollups, nil, nil) + select { + case a := <-delivered: + t.Fatalf("revalidation bypassed %s: %+v", policy, a) + case <-time.After(100 * time.Millisecond): + } + }) + } +} + +func TestRestoredBackupRevalidationPreservesQuietHoursReplay(t *testing.T) { + m, rollups, restored := restartOverdueBackup(t, false, nil) + cfg := startupBackupConfig() + cfg.Schedule.QuietHours = QuietHours{Enabled: true, Start: "00:00", End: "23:59", Timezone: "UTC", + Days: map[string]bool{"monday": true, "tuesday": true, "wednesday": true, "thursday": true, + "friday": true, "saturday": true, "sunday": true}, + Suppress: QuietHoursSuppression{Storage: true}} + m.UpdateConfig(cfg) + if !m.ShouldSuppressNotification(restored) || hasQuietHoursNotificationReplay(restored) { + t.Fatal("startup hold queued stale backup evidence for quiet-hours replay") + } + // Fresh evidence downgrades this retained occurrence to a warning, which + // normal quiet-hours policy defers. Critical backup-age alerts bypass it. + warningBackupTime := time.Now().Add(-75 * 24 * time.Hour) + rollups[0].LastSuccessAt = &warningBackupTime + delivered := make(chan *Alert, 2) + m.SetAlertCallback(func(a *Alert) { delivered <- a }) + m.CheckBackups(rollups, nil, nil) + select { + case a := <-delivered: + if a.Level != AlertLevelWarning || !hasQuietHoursNotificationReplay(a) { + t.Fatal("revalidated overdue backup lost normal quiet-hours replay metadata") + } + case <-time.After(2 * time.Second): + t.Fatal("revalidated backup did not enter normal quiet-hours delivery") + } +} diff --git a/internal/alerts/escalation.go b/internal/alerts/escalation.go index 5e747492ed..8e8bbcc59b 100644 --- a/internal/alerts/escalation.go +++ b/internal/alerts/escalation.go @@ -49,6 +49,9 @@ func (m *Manager) checkEscalations() { if alert == nil || alert.Acknowledged || isMonitorOnlyAlert(alert) || isSupportedInfrastructureSymptom(alert) { continue } + if m.awaitsBackupRevalidationNoLock(alert) { + continue + } if _, snoozed := alertSnoozeUntil(alert, now); snoozed { continue } @@ -133,6 +136,9 @@ func (m *Manager) PrepareEscalationNotification(snapshot *Alert, level int) (*Al if !ok || active == nil || !active.StartTime.Equal(snapshot.StartTime) || active.Acknowledged || isMonitorOnlyAlert(active) || isSupportedInfrastructureSymptom(active) { return nil, EscalationLevel{}, false } + if m.awaitsBackupRevalidationNoLock(active) { + return nil, EscalationLevel{}, false + } if _, snoozed := alertSnoozeUntil(active, m.policyNow()); snoozed { return nil, EscalationLevel{}, false } diff --git a/internal/alerts/manager.go b/internal/alerts/manager.go index 79aa6dc464..ef7021c080 100644 --- a/internal/alerts/manager.go +++ b/internal/alerts/manager.go @@ -133,6 +133,9 @@ type Manager struct { activeRecoveryReadable atomic.Bool activeRecoveryWriteBlock atomic.Bool restoredAlertEpoch atomic.Uint64 + // Restored backup occurrences stay visible, but cannot notify until a + // successful backup evaluation confirms they are still overdue. Under mu. + restoredBackupNotifications map[string]time.Time // Shadow-mode reducer feed (Phase 1 capstone). Nil until // EnableShadowFeed; all access is under m.mu. diff --git a/internal/alerts/notification_policy.go b/internal/alerts/notification_policy.go index 2fc298450e..ddb9032c64 100644 --- a/internal/alerts/notification_policy.go +++ b/internal/alerts/notification_policy.go @@ -172,11 +172,39 @@ func (m *Manager) expireSuppressionNoLock(trackingKey string, now time.Time) boo return true } +const alertDeliveryReasonBackupRevalidation = "awaiting_backup_evaluation" + +// Caller holds mu. The occurrence check prevents a deferred startup snapshot +// from suppressing a later alert that happens to reuse the same identity. +func (m *Manager) awaitsBackupRevalidationNoLock(alert *Alert) bool { + if alert == nil || alert.Type != "backup-age" { + return false + } + startedAt, pending := m.restoredBackupNotifications[canonicalTrackingKeyForAlert(alert)] + return pending && startedAt.Equal(alert.StartTime) +} + +// A valid, matched canonical evaluation releases the startup delivery hold. +// Recovery/removal retires the hold through the active lifecycle owner instead. +func (m *Manager) revalidateBackupNotificationNoLock(alert *Alert) bool { + if !m.awaitsBackupRevalidationNoLock(alert) { + return false + } + delete(m.restoredBackupNotifications, canonicalTrackingKeyForAlert(alert)) + return true +} + func (m *Manager) dispatchAlert(alert *Alert, async bool) bool { callbacks := m.getAlertCallbacks() if len(callbacks) == 0 || alert == nil { return false } + if m.awaitsBackupRevalidationNoLock(alert) { + m.recordAlertEvent(eventlog.TypeNotificationSuppressed, alert, "", + alertDeliveryReasonBackupRevalidation, + "Restored backup alert is awaiting a successful backup evaluation.", nil) + return false + } // Don't dispatch notifications for acknowledged alerts if alert.Acknowledged { @@ -640,6 +668,9 @@ func (m *Manager) ShouldSuppressNotification(alert *Alert) bool { m.mu.RLock() defer m.mu.RUnlock() + if m.awaitsBackupRevalidationNoLock(alert) { + return true + } if _, snoozed := alertSnoozeUntil(alert, m.policyNow()); snoozed { clearQuietHoursNotificationReplay(alert) return true @@ -858,6 +889,8 @@ func (m *Manager) diagnoseActiveAlertLocked(alert *Alert) AlertDeliveryDiagnosis diagnosis.RecentAlertsInHour = m.rateLimitCountLocked(trackingKey, now) switch { + case m.awaitsBackupRevalidationNoLock(alert): + diagnosis.setSuppressed(alertDeliveryReasonBackupRevalidation, "Restored backup alert is awaiting a successful backup evaluation.") case alert.Acknowledged: diagnosis.setSuppressed(AlertDeliveryReasonAcknowledged, "Alert is acknowledged, so firing notifications are suppressed.") case isSupportedInfrastructureSymptom(alert):