From 5774b6bc4b221c1f5dd10f02dbf2666647a5f7b4 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 4 Oct 2026 15:53:32 +0100 Subject: [PATCH 1/2] Keep backup-age evaluation fast and failures visible For #2465, select latest provider evidence with the existing scope index rather than ranking every retained payload. Preserve history and evidence validation. Share complete, failure-preserving backup alert evaluation across PVE/PBS/mock polls and expose unavailable reads through the existing deduplicated system warning; keep existing age alerts until a complete read. Contract-Neutral: Application query and existing system-alert behaviour only; no API, schema, resource identity, agent lifecycle, entitlement or notification contract shape changes. The new warning uses the existing open system-alert type, identity and metadata contract. Semantic contracts and connected verification are included. Change-source: pulse-maintainer --- .../v6/internal/subsystems/alerts.md | 16 ++ .../v6/internal/subsystems/monitoring.md | 22 ++ .../internal/subsystems/storage-recovery.md | 18 ++ .../alerts/backup_evaluation_system_test.go | 28 +++ internal/alerts/system_alert.go | 4 + .../backup_alert_evaluation_test.go | 209 ++++++++++++++++++ .../monitoring/canonical_guardrails_test.go | 16 ++ internal/monitoring/monitor.go | 1 + internal/monitoring/monitor_alerts.go | 8 +- internal/monitoring/monitor_backups.go | 20 +- internal/monitoring/recovery_rollups.go | 51 +++++ .../store/store_observation_history_test.go | 132 +++++++++++ internal/recovery/store/store_posture.go | 35 ++- internal/recovery/store/store_posture_test.go | 101 +++++++++ .../recovery/store/store_queryplan_test.go | 15 ++ 15 files changed, 639 insertions(+), 37 deletions(-) create mode 100644 internal/alerts/backup_evaluation_system_test.go create mode 100644 internal/monitoring/backup_alert_evaluation_test.go create mode 100644 internal/recovery/store/store_observation_history_test.go diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 5a0febf39b..c0943f501d 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -15,6 +15,22 @@ ## Purpose +### Backup-age evaluation failures use the existing system-alert lifecycle + +`pulse-system-backup-evaluation` is a warning about Pulse's inability to read +complete backup rollups, not a claim that a workload's backup failed. It has no +resource target and carries the standard system metadata. A stable fingerprint +keeps repeated failed polls on one alert without repeat notification; existing +backup-age alerts are retained. Only complete evaluation or deliberate disablement +clears the warning. The public message names the lost evaluation and points to +Pulse logs without including raw database paths or provider errors. + +`backup_evaluation_system_test.go` verifies stable identity, idempotence, node +cleanup survival and recovery. Monitoring's connected evaluator tests exercise +persisted stale/fresh data and failures through this helper. No new route, +navigation entry, destination or monitored-resource health signal is added. + + ### RAID spare-count lifecycle — issue #2369 Canonical host RAID alerts use the shared required-member health assessment, diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 46490d7506..4dbf5cb604 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,28 @@ ## Purpose +### Unavailable backup-age evaluation is not silent recovery + +PVE, PBS and mock backup checks share one failure-preserving evaluator. The +existing two-second complete rollup-read budget includes connection waiting; +it is not extended or bypassed. A failed or partial read does not invoke backup +age evaluation with an empty result, clear existing age alerts or assert backup +health. A fixed system warning makes the skipped evaluation visible in the +existing alert list and served state, with no raw database/provider/path error. +A complete subsequent evaluation clears that warning; normal cancellation or a +missing manager does not stand in for read recovery. Deliberately disabled checks +clear their irrelevant warning without reading recovery history. + +Overlapping global checks coalesce instead of queueing. The single evaluation +owner orders failure/recovery publication so an older success cannot erase a +newer failure. `backup_alert_evaluation_test.go` exercises actual persisted +rollups, stale/fresh backups, read failure/reopening, served-state visibility, +repeated failures, missing dependencies, cancellation/deadlines and concurrent +coalescing. `TestBackupAlertEvaluationCallersShareFailureVisibility` binds all +three callers to that path. These are synthetic source controls, not native +backup thaw, Raspberry Pi relief or published availability. + + ### Platform connection discovery follows tenant storage identity TrueNAS and VMware pollers enumerate validated, sorted persistence directory diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 854284a70e..1395d26b1f 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -21,6 +21,24 @@ ## Purpose +### Latest provider observations do not scan retained evidence payloads + +The protection-posture reader enumerates provider/scope keys through the existing +covering index, then seeks the latest observation by observed time and ID. Only +that row's evidence payload is read for each scope. The 90-day retention, schema, +single-connection policy and evidence validation are unchanged; malformed or +invalid latest evidence never falls back to older reassuring facts. Narrow index +key enumeration still scales with retained history, but wide historical payloads +are not ranked on every ingest or posture read. + +`store_queryplan_test.go` checks the production SQL's covering enumeration and +indexed latest-row lookup. `store_posture_test.go` covers timestamp/ID ties, +provider/scope separation, corrupt latest evidence, cancellation, retained row +count and connection-pool deadline/recovery. The fixed synthetic history +benchmark uses 1,000/70,949/180,000 rows and two scopes; it is not native Pi +performance, backup restore acceptance or release qualification. + + ### Backup inventory ownership is source-scoped and unambiguous The existing Backups Coverage and By date views do not select a workload from diff --git a/internal/alerts/backup_evaluation_system_test.go b/internal/alerts/backup_evaluation_system_test.go new file mode 100644 index 0000000000..536838451f --- /dev/null +++ b/internal/alerts/backup_evaluation_system_test.go @@ -0,0 +1,28 @@ +package alerts + +import "testing" + +func TestBackupEvaluationSystemAlertUsesExistingIdentityAndDeduplication(t *testing.T) { + m := NewManagerWithDataDir(t.TempDir()) + defer m.Stop() + input := SystemAlertInput{Type: BackupEvaluationAlertType, Level: AlertLevelWarning, + Fingerprint: "recovery-rollups-unavailable", Message: "Backup-age alerts were not evaluated."} + if !m.RaiseSystemAlert(input) { + t.Fatal("first failure not raised") + } + if m.RaiseSystemAlert(input) { + t.Fatal("repeated failure re-notified") + } + active := m.GetActiveAlerts() + if len(active) != 1 || active[0].ID != SystemAlertID(BackupEvaluationAlertType) || + active[0].ResourceID != "" || active[0].Metadata["systemAlert"] != true { + t.Fatalf("system projection=%+v", active) + } + m.CleanupAlertsForNodes(map[string]bool{}) + if len(m.GetActiveAlerts()) != 1 { + t.Fatal("node cleanup erased the evaluator outage") + } + if !m.ClearSystemAlert(BackupEvaluationAlertType) || len(m.GetActiveAlerts()) != 0 { + t.Fatal("recovery did not clear the warning") + } +} diff --git a/internal/alerts/system_alert.go b/internal/alerts/system_alert.go index 98c3ce9b69..5bd11fffa3 100644 --- a/internal/alerts/system_alert.go +++ b/internal/alerts/system_alert.go @@ -28,6 +28,10 @@ const ( // notification destinations are not delivering. NotificationDeliveryAlertType = "notification-delivery" + // BackupEvaluationAlertType reports unavailable backup-age evaluation, + // independently of whether backup artifacts themselves are stale. + BackupEvaluationAlertType = "backup-evaluation" + // DeadManDeliveryAlertType reports that Pulse is healthy but cannot reach // the configured external watchdog. DeadManDeliveryAlertType = "deadman-delivery" diff --git a/internal/monitoring/backup_alert_evaluation_test.go b/internal/monitoring/backup_alert_evaluation_test.go new file mode 100644 index 0000000000..fe667dc9e9 --- /dev/null +++ b/internal/monitoring/backup_alert_evaluation_test.go @@ -0,0 +1,209 @@ +package monitoring + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" + recoverymanager "github.com/rcourtman/pulse-go-rewrite/internal/recovery/manager" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +func backupEvaluationMonitor(t *testing.T) (*Monitor, *config.MultiTenantPersistence) { + t.Helper() + mtp := config.NewMultiTenantPersistence(t.TempDir()) + manager := recoverymanager.New(mtp) + am := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(am.Stop) + cfg := am.GetConfig() + cfg.Enabled = true + cfg.ActivationState = alerts.ActivationActive + cfg.BackupDefaults.Enabled = true + cfg.BackupDefaults.WarningDays = 1 + cfg.BackupDefaults.CriticalDays = 2 + am.UpdateConfig(cfg) + m := &Monitor{ + state: models.NewState(), alertManager: am, recoveryManager: manager, + resourceStore: backupReadStateResourceStore([]unifiedresources.Resource{ + {ID: "vm-100", Name: "database", Type: unifiedresources.ResourceTypeVM, + Status: unifiedresources.StatusOnline, + Proxmox: &unifiedresources.ProxmoxData{Instance: "pve1", NodeName: "node1", VMID: 100}}, + }), + } + return m, mtp +} + +func backupEvaluationPoint(at time.Time) recovery.RecoveryPoint { + return recovery.RecoveryPoint{ + ID: "pve-backup:fixture-100", Provider: recovery.ProviderProxmoxPVE, + Kind: recovery.KindBackup, Mode: recovery.ModeLocal, Outcome: recovery.OutcomeSuccess, + CompletedAt: &at, SubjectResourceID: "vm-100", + SubjectRef: &recovery.ExternalRef{Type: "proxmox-vm", ID: "pve1:node1:100", Name: "database"}, + } +} + +func backupEvaluationAlerts(t *testing.T, m *Monitor) (self *alerts.Alert, ages int) { + t.Helper() + for _, a := range m.alertManager.GetActiveAlerts() { + switch a.Type { + case alerts.BackupEvaluationAlertType: + copy := a + self = © + case "backup-age": + ages++ + } + } + return +} + +func TestBackupAlertEvaluationFailureAndFreshRecovery(t *testing.T) { + m, mtp := backupEvaluationMonitor(t) + store, err := m.recoveryManager.StoreForOrg("default") + if err != nil { + t.Fatal(err) + } + old := time.Now().UTC().Add(-4 * 24 * time.Hour) + if err := store.UpsertPoints(context.Background(), []recovery.RecoveryPoint{backupEvaluationPoint(old)}); err != nil { + t.Fatal(err) + } + m.checkBackupAlerts(context.Background()) + if self, ages := backupEvaluationAlerts(t, m); self != nil || ages != 1 { + t.Fatalf("initial evaluation self=%+v age alerts=%d", self, ages) + } + // A read failure cannot be interpreted as an empty, healthy inventory. + if err := store.Close(); err != nil { + t.Fatal(err) + } + m.checkBackupAlerts(context.Background()) + first, ages := backupEvaluationAlerts(t, m) + if first == nil || ages != 1 { + t.Fatalf("failure self=%+v age alerts=%d", first, ages) + } + if first.ID != alerts.SystemAlertID(alerts.BackupEvaluationAlertType) || first.ResourceID != "" || first.Level != alerts.AlertLevelWarning { + t.Fatalf("incorrect system identity %+v", first) + } + if !strings.Contains(first.Message, "not evaluated") || strings.Contains(first.Message, "recovery.db") { + t.Fatalf("unsafe or unclear message %q", first.Message) + } + for i := 0; i < 3; i++ { + m.checkBackupAlerts(context.Background()) + } + repeated, ages := backupEvaluationAlerts(t, m) + if repeated == nil || ages != 1 || !repeated.StartTime.Equal(first.StartTime) || repeated.Metadata["systemAlertFingerprint"] != first.Metadata["systemAlertFingerprint"] { + t.Fatal("repeated failure replaced the condition or lost existing age alerts") + } + // The operator-visible snapshot uses the live manager, not stale arrays. + found := false + for _, a := range m.GetState().ActiveAlerts { + if a.ID == first.ID { + found = true + } + } + if !found { + t.Fatal("evaluation warning absent from served state") + } + // Reopening the same synthetic persisted data is recovery; success alone + // clears only the self-alert while still detecting the stale backup. + m.recoveryManager = recoverymanager.New(mtp) + recovered, err := m.recoveryManager.StoreForOrg("default") + if err != nil { + t.Fatal(err) + } + defer recovered.Close() + m.checkBackupAlerts(context.Background()) + if self, ages := backupEvaluationAlerts(t, m); self != nil || ages != 1 { + t.Fatalf("read recovery self=%+v age alerts=%d", self, ages) + } + fresh := time.Now().UTC() + if err := recovered.UpsertPoints(context.Background(), []recovery.RecoveryPoint{backupEvaluationPoint(fresh)}); err != nil { + t.Fatal(err) + } + m.checkBackupAlerts(context.Background()) + if self, ages := backupEvaluationAlerts(t, m); self != nil || ages != 0 { + t.Fatalf("fresh backup self=%+v age alerts=%d", self, ages) + } +} + +func TestBackupAlertEvaluationUnavailableIsNotRecovery(t *testing.T) { + m, _ := backupEvaluationMonitor(t) + m.recoveryManager = recoverymanager.New(nil) + m.checkBackupAlerts(context.Background()) + self, _ := backupEvaluationAlerts(t, m) + if self == nil { + t.Fatal("missing persistence did not raise an evaluation warning") + } + m.recoveryManager = nil + m.checkBackupAlerts(context.Background()) + if current, _ := backupEvaluationAlerts(t, m); current == nil { + t.Fatal("missing manager falsely cleared unavailable evaluation") + } + cfg := m.alertManager.GetConfig() + cfg.BackupDefaults.Enabled = false + m.alertManager.UpdateConfig(cfg) + m.checkBackupAlerts(context.Background()) + if current, _ := backupEvaluationAlerts(t, m); current != nil { + t.Fatal("disabled backup checks retain an irrelevant evaluation warning") + } +} + +func TestBackupAlertEvaluationCancellationAndDeadline(t *testing.T) { + m, _ := backupEvaluationMonitor(t) + store, err := m.recoveryManager.StoreForOrg("default") + if err != nil { + t.Fatal(err) + } + defer store.Close() + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + m.checkBackupAlerts(cancelled) + if self, _ := backupEvaluationAlerts(t, m); self != nil { + t.Fatal("normal cancellation generated an outage") + } + deadline, stop := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer stop() + m.checkBackupAlerts(deadline) + if self, _ := backupEvaluationAlerts(t, m); self == nil { + t.Fatal("deadline skip stayed invisible") + } + // Another normal cancellation is not positive evaluation evidence. + m.checkBackupAlerts(cancelled) + if self, _ := backupEvaluationAlerts(t, m); self == nil { + t.Fatal("cancellation falsely cleared an outage") + } + m.checkBackupAlerts(nil) + if self, _ := backupEvaluationAlerts(t, m); self != nil { + t.Fatal("complete empty read did not recover the evaluator") + } +} + +func TestBackupAlertEvaluationCoalescesInFlightReads(t *testing.T) { + m, _ := backupEvaluationMonitor(t) + m.recoveryManager = recoverymanager.New(nil) + m.backupAlertEvalMu.Lock() + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { defer wg.Done(); m.checkBackupAlerts(context.Background()) }() + } + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("overlapping global reads queued behind in-flight evaluation") + } + m.backupAlertEvalMu.Unlock() + if self, _ := backupEvaluationAlerts(t, m); self != nil { + t.Fatal("coalesced read published failure without evaluating") + } + m.checkBackupAlerts(context.Background()) + if self, _ := backupEvaluationAlerts(t, m); self == nil { + t.Fatal("subsequent global check did not run") + } +} diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 38865b21a9..5a35da2af8 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -3070,3 +3070,19 @@ func TestGuestAgentTransportMonitoringContract(t *testing.T) { } testGuestAgentTransportDeferralKeepsLastKnownHistory(t) } + +func TestBackupAlertEvaluationCallersShareFailureVisibility(t *testing.T) { + for path, want := range map[string]int{"monitor_backups.go": 2, "monitor_alerts.go": 1} { + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + source := string(raw) + if strings.Contains(source, "listBackupRollupsForAlerts(") || strings.Contains(source, "CheckBackupsWithInventory(") { + t.Errorf("%s bypasses shared failure-preserving evaluation", path) + } + if got := strings.Count(source, "m.checkBackupAlerts("); got != want { + t.Errorf("%s has %d shared backup evaluations, want %d", path, got, want) + } + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 4d97279cda..7a1f8b2405 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1199,6 +1199,7 @@ type Monitor struct { pveBackupInventoryReady map[string]map[string]bool // Track PVE guest inventory readiness for backup orphan detection pveBackupTemplateSubjects map[string]map[string]struct{} // Track template VMIDs excluded from runtime workloads but valid for backups backupPermissionWarnings map[string]string // Track backup permission issues per instance (instance -> warning message) + backupAlertEvalMu sync.Mutex // Coalesce global rollup reads and order failure/recovery publication. persistence *config.ConfigPersistence // Add persistence for saving updated configs pbsBackupPollers map[string]bool // Track PBS backup polling goroutines per instance pbsBackupCacheTime map[string]map[pbsBackupGroupKey]time.Time // Track when each PBS backup group was last fetched diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index 826f9971b7..46445c3ec7 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -658,13 +658,7 @@ func (m *Monitor) checkMockAlerts() { Msg("Collecting resources for alert cleanup in mock mode") m.alertManager.CleanupAlertsForNodes(existingNodes) - guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.GetUnifiedReadStateOrSnapshot(), m.guestMetadataStore) - rollups, err := m.listBackupRollupsForAlerts(context.Background()) - if err != nil { - log.Warn().Err(err).Msg("Failed to list recovery rollups for backup alerts") - } else { - m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) - } + m.checkBackupAlerts(context.Background()) // Limit how many guests we check per cycle to prevent blocking with large datasets const maxGuestsPerCycle = 50 diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index 2e15c9bc69..1dc59f1184 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -520,15 +520,7 @@ func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName // Sync backup times to VMs/Containers and republish them to canonical resources. m.syncGuestBackupTimesAndResourceStore() - if m.alertManager != nil { - guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.GetUnifiedReadStateOrSnapshot(), m.guestMetadataStore) - rollups, err := m.listBackupRollupsForAlerts(ctx) - if err != nil { - log.Warn().Err(err).Msg("Failed to list recovery rollups for backup alerts") - } else { - m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) - } - } + m.checkBackupAlerts(ctx) // Clear permission warning if no permission errors occurred this cycle if !hadPermissionError { @@ -1881,15 +1873,7 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien // Sync backup times to VMs/Containers and republish them to canonical resources. m.syncGuestBackupTimesAndResourceStore() - if m.alertManager != nil { - guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.GetUnifiedReadStateOrSnapshot(), m.guestMetadataStore) - rollups, err := m.listBackupRollupsForAlerts(context.Background()) - if err != nil { - log.Warn().Err(err).Msg("Failed to list recovery rollups for backup alerts") - } else { - m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) - } - } + m.checkBackupAlerts(context.Background()) // Immediately broadcast the updated state so frontend sees new backups m.broadcastStateUpdate() diff --git a/internal/monitoring/recovery_rollups.go b/internal/monitoring/recovery_rollups.go index ac71067335..34120189d6 100644 --- a/internal/monitoring/recovery_rollups.go +++ b/internal/monitoring/recovery_rollups.go @@ -2,11 +2,14 @@ package monitoring import ( "context" + "errors" "fmt" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/recovery" + "github.com/rs/zerolog/log" ) const ( @@ -80,7 +83,55 @@ func (m *Monitor) listRecoveryRollupsForAlerts(ctx context.Context, kind recover } func (m *Monitor) listBackupRollupsForAlerts(ctx context.Context) ([]recovery.ProtectionRollup, error) { + if ctx == nil { + ctx = context.Background() + } ctx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() return m.listRecoveryRollupsForAlerts(ctx, recovery.KindBackup) } + +// checkBackupAlerts is shared by PVE, PBS and mock polls. A failed/partial +// rollup read is not an empty inventory and cannot clear existing age alerts. +// Put the lost evaluation in the existing alert list, with one stable identity +// and no raw database/path/provider errors in its public message. +func (m *Monitor) checkBackupAlerts(ctx context.Context) { + if m == nil || m.alertManager == nil || !m.backupAlertEvalMu.TryLock() { + return + } + defer m.backupAlertEvalMu.Unlock() + + cfg := m.alertManager.GetConfig() + if !cfg.Enabled || !cfg.BackupDefaults.Enabled || + (cfg.BackupDefaults.WarningDays <= 0 && cfg.BackupDefaults.CriticalDays <= 0) { + // Disabling evaluation is a deliberate policy change, not a read failure. + m.alertManager.CheckBackupsWithInventory(nil, nil, nil, nil) + m.alertManager.ClearSystemAlert(alerts.BackupEvaluationAlertType) + return + } + if m.recoveryManager == nil && !mock.IsMockEnabled() { + // Missing dependencies cannot stand in for affirmative read recovery. + return + } + rollups, err := m.listBackupRollupsForAlerts(ctx) + if err != nil { + // Normal shutdown/cancellation is not a new evaluation outage. A read + // deadline, including connection-pool waiting, still needs visibility. + if !errors.Is(err, context.Canceled) { + log.Warn().Err(err).Msg("Failed to list recovery rollups for backup alerts") + m.alertManager.RaiseSystemAlert(alerts.SystemAlertInput{ + Type: alerts.BackupEvaluationAlertType, + Level: alerts.AlertLevelWarning, + Fingerprint: "recovery-rollups-unavailable", + Message: "Backup-age alerts were not evaluated because recovery data could not be read. Existing backup alerts have been kept; check Pulse's logs.", + }) + } + return + } + guestsByKey, guestsByVMID := buildGuestLookupsFromReadState(m.GetUnifiedReadStateOrSnapshot(), m.guestMetadataStore) + m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts()) + // Only a complete read/evaluation (including a genuinely empty result) + // resolves the outage. In-flight checks are coalesced, not queued, so an + // older successful read cannot later erase a newer failure. + m.alertManager.ClearSystemAlert(alerts.BackupEvaluationAlertType) +} diff --git a/internal/recovery/store/store_observation_history_test.go b/internal/recovery/store/store_observation_history_test.go new file mode 100644 index 0000000000..dc7e21ce82 --- /dev/null +++ b/internal/recovery/store/store_observation_history_test.go @@ -0,0 +1,132 @@ +package store + +import ( + "context" + "fmt" + "go/ast" + "go/parser" + "go/token" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" +) + +// Read the SQL used by the real reader, not a duplicate that can drift from it. +func latestProtectionObservationSQL(t testing.TB) string { + t.Helper() + file, err := parser.ParseFile(token.NewFileSet(), "store_posture.go", nil, 0) + if err != nil { + t.Fatal(err) + } + var query string + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Name.Name != "listLatestProtectionProviderObservations" { + continue + } + ast.Inspect(fn.Body, func(node ast.Node) bool { + call, ok := node.(*ast.CallExpr) + if !ok || len(call.Args) < 2 { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "QueryContext" { + return true + } + literal, ok := call.Args[1].(*ast.BasicLit) + if !ok || literal.Kind != token.STRING { + return true + } + query, err = strconv.Unquote(literal.Value) + if err != nil { + t.Fatal(err) + } + return false + }) + } + if query == "" { + t.Fatal("could not locate the production observation query") + } + return query +} + +func observationHistoryStore(t testing.TB, rows int) *Store { + t.Helper() + s, err := Open(filepath.Join(t.TempDir(), "recovery.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + now := time.Now().UTC().Truncate(time.Millisecond) + latest := []recovery.ProtectionProviderObservation{ + observationForHistoryTest(t, recovery.ProviderProxmoxPVE, "pve-main", now), + observationForHistoryTest(t, recovery.ProviderProxmoxPBS, "pbs-main", now), + } + if err := s.UpsertProtectionProviderObservations(context.Background(), latest); err != nil { + t.Fatal(err) + } + // Synthetic retained rows are deliberately irrelevant to the latest facts. + // Their 1 KiB payload models the cost of reading wide historical evidence; + // no reporter database or driver instrumentation is involved. + tx, err := s.db.Begin() + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + stmt, err := tx.Prepare(`INSERT INTO protection_provider_observations ( + id, provider, source, scope, job_state, history_completeness, permissions, + verification_expected, observed_at_ms, ingested_at_ms, evidence_json, + created_at_ms, updated_at_ms + ) VALUES (?, ?, 'history-fixture', ?, 'success', 'complete', 'sufficient', + 0, ?, ?, ?, ?, ?)`) + if err != nil { + t.Fatal(err) + } + defer stmt.Close() + raw := strings.Repeat(" ", 1024) + "{}" + for i := 0; i < rows-2; i++ { + seed := latest[i%2] + at := now.UnixMilli() - int64(i+1)*1000 + if _, err := stmt.Exec(fmt.Sprintf("history-%09d", i), string(seed.Provider), seed.Scope, at, at, raw, at, at); err != nil { + t.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return s +} + +func observationForHistoryTest(t testing.TB, provider recovery.Provider, scope string, at time.Time) recovery.ProtectionProviderObservation { + t.Helper() + o, err := recovery.NewProtectionProviderObservation(provider, "history-fixture", scope, + recovery.OutcomeSuccess, recovery.ProtectionHistoryComplete, + operationaltrust.EvidencePermissionsSufficient, false, at, at, nil) + if err != nil { + t.Fatal(err) + } + return o +} + +// Fixed comparisons selected before execution: small history, the reporter's +// row count, and a larger retained history. Timings are evidence, not CI gates. +func BenchmarkLatestProtectionObservationHistory(b *testing.B) { + for _, n := range []int{1000, 70949, 180000} { + b.Run(fmt.Sprintf("rows-%d", n), func(b *testing.B) { + s := observationHistoryStore(b, n) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + obs, err := s.listLatestProtectionProviderObservations(context.Background()) + if err != nil || len(obs) != 2 { + b.Fatalf("latest observations=%d error=%v", len(obs), err) + } + } + }) + } +} diff --git a/internal/recovery/store/store_posture.go b/internal/recovery/store/store_posture.go index db988ac197..4666cd3e4a 100644 --- a/internal/recovery/store/store_posture.go +++ b/internal/recovery/store/store_posture.go @@ -455,21 +455,32 @@ func (s *Store) loadProtectionPointsForKeys( func (s *Store) listLatestProtectionProviderObservations( ctx context.Context, ) ([]recovery.ProtectionProviderObservation, error) { + // Enumerate narrow scope keys from the covering index, then seek the latest + // row in each scope. Ranking SELECT * over 90 days of evidence holds the + // sole recovery connection while reading every historical payload (#2465). + // Keep the existing timestamp/ID tie-break and do not fall back from an + // invalid latest observation to older, potentially reassuring evidence. rows, err := s.db.QueryContext(ctx, ` - SELECT - id, provider, source, scope, job_state, - history_completeness, permissions, verification_expected, - observed_at_ms, ingested_at_ms, evidence_json - FROM ( - SELECT *, - ROW_NUMBER() OVER ( - PARTITION BY provider, scope - ORDER BY observed_at_ms DESC, id DESC - ) AS rn + WITH scopes AS ( + SELECT DISTINCT provider, scope FROM protection_provider_observations ) - WHERE rn = 1 - ORDER BY provider, scope + SELECT + observation.id, observation.provider, observation.source, + observation.scope, observation.job_state, + observation.history_completeness, observation.permissions, + observation.verification_expected, observation.observed_at_ms, + observation.ingested_at_ms, observation.evidence_json + FROM scopes + JOIN protection_provider_observations observation + ON observation.id = ( + SELECT candidate.id + FROM protection_provider_observations candidate + WHERE candidate.provider = scopes.provider AND candidate.scope = scopes.scope + ORDER BY candidate.observed_at_ms DESC, candidate.id DESC + LIMIT 1 + ) + ORDER BY observation.provider, observation.scope `) if err != nil { return nil, err diff --git a/internal/recovery/store/store_posture_test.go b/internal/recovery/store/store_posture_test.go index 9ab682759e..c6b054ec7b 100644 --- a/internal/recovery/store/store_posture_test.go +++ b/internal/recovery/store/store_posture_test.go @@ -649,3 +649,104 @@ func TestStoreProtectionMetadataBackfillAddsScopeAndTypedEvidence(t *testing.T) t.Fatalf("ingestedAt = %s, want %s", evidence.IngestedAt, updatedAt) } } + +func TestLatestProtectionObservationsKeepScopeAndTieSemantics(t *testing.T) { + s, err := Open(filepath.Join(t.TempDir(), "recovery.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + now := time.Now().UTC().Truncate(time.Millisecond) + old := observationForHistoryTest(t, recovery.ProviderProxmoxPVE, "same-scope", now.Add(-time.Hour)) + latest := observationForHistoryTest(t, recovery.ProviderProxmoxPVE, "same-scope", now) + tie := latest + tie.ID = "zz-latest-tie" + tie.Evidence.ID = tie.ID + tie.Evidence.PayloadRef = &operationaltrust.EvidencePayloadRef{Kind: "protection-provider-observation", ID: tie.ID} + otherProvider := observationForHistoryTest(t, recovery.ProviderProxmoxPBS, "same-scope", now.Add(-time.Minute)) + otherScope := observationForHistoryTest(t, recovery.ProviderProxmoxPVE, "same-scope-2", now.Add(-time.Minute)) + if err := s.UpsertProtectionProviderObservations(context.Background(), []recovery.ProtectionProviderObservation{tie, otherScope, old, latest, otherProvider}); err != nil { + t.Fatal(err) + } + got, err := s.listLatestProtectionProviderObservations(context.Background()) + if err != nil || len(got) != 3 { + t.Fatalf("observations=%+v error=%v", got, err) + } + want := map[string]string{ + string(tie.Provider) + "/" + tie.Scope: tie.ID, + string(otherScope.Provider) + "/" + otherScope.Scope: otherScope.ID, + string(otherProvider.Provider) + "/" + otherProvider.Scope: otherProvider.ID, + } + previous := "" + for _, o := range got { + key := string(o.Provider) + "/" + o.Scope + if want[key] != o.ID { + t.Errorf("scope %q chose %q, want %q", key, o.ID, want[key]) + } + if previous > key { + t.Errorf("provider/scope ordering lost: %q > %q", previous, key) + } + previous = key + } + // A corrupt latest row must not resurrect an older reassuring observation. + if _, err := s.db.Exec("UPDATE protection_provider_observations SET evidence_json='{' WHERE id=?", tie.ID); err != nil { + t.Fatal(err) + } + got, err = s.listLatestProtectionProviderObservations(context.Background()) + if err != nil || len(got) != 2 { + t.Fatalf("corrupt latest observations=%+v error=%v", got, err) + } + for _, o := range got { + if o.Provider == tie.Provider && o.Scope == tie.Scope { + t.Fatal("fell back from corrupt latest to older evidence") + } + } + // Empty and cancelled reads retain their existing meanings. + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := s.listLatestProtectionProviderObservations(cancelled); err == nil { + t.Fatal("cancelled read succeeded") + } + if _, err := s.db.Exec("DELETE FROM protection_provider_observations"); err != nil { + t.Fatal(err) + } + got, err = s.listLatestProtectionProviderObservations(context.Background()) + if err != nil || len(got) != 0 { + t.Fatalf("empty observations=%+v error=%v", got, err) + } +} + +func TestLatestProtectionObservationHistoryRemainsRetained(t *testing.T) { + s := observationHistoryStore(t, 70949) + got, err := s.listLatestProtectionProviderObservations(context.Background()) + if err != nil || len(got) != 2 { + t.Fatalf("observations=%d error=%v", len(got), err) + } + var count int + if err := s.db.QueryRow("SELECT COUNT(*) FROM protection_provider_observations").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 70949 { + t.Fatalf("retained observations=%d, want 70949", count) + } +} + +func TestBackupRollupConnectionWaitingHonoursDeadline(t *testing.T) { + s := observationHistoryStore(t, 1000) + conn, err := s.db.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, _, err := s.ListRollups(ctx, recovery.ListPointsOptions{Kind: recovery.KindBackup}); err != context.DeadlineExceeded { + t.Fatalf("pool wait error=%v, want deadline exceeded", err) + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } + if _, _, err := s.ListRollups(context.Background(), recovery.ListPointsOptions{Kind: recovery.KindBackup}); err != nil { + t.Fatalf("rollups after releasing held connection: %v", err) + } +} diff --git a/internal/recovery/store/store_queryplan_test.go b/internal/recovery/store/store_queryplan_test.go index 04d0f1b9e6..ff40f55e3d 100644 --- a/internal/recovery/store/store_queryplan_test.go +++ b/internal/recovery/store/store_queryplan_test.go @@ -296,3 +296,18 @@ func recoveryOutcomeForIndex(i int) recovery.Outcome { func twoDigit(v int) string { return fmt.Sprintf("%02d", v) } + +// #2465: enumerate narrow indexed scope keys and seek each latest row. Ranking +// every historical evidence payload can monopolise the one recovery connection +// and make the otherwise quick backup-age rollup read miss its deadline. +func TestLatestProtectionObservationPlanAvoidsHistoricalPayloadScan(t *testing.T) { + s := observationHistoryStore(t, 1000) + plan := explainRecoveryQueryPlan(t, s.db, latestProtectionObservationSQL(t)) + t.Logf("production latest-observation plan:\n%s", plan) + if !strings.Contains(plan, "USING COVERING INDEX idx_protection_provider_observations_scope") { + t.Error("scope enumeration must read the narrow covering index, not every historical payload") + } + if !strings.Contains(plan, "SEARCH candidate USING INDEX idx_protection_provider_observations_scope (provider=? AND scope=?)") { + t.Error("latest evidence must be selected through a provider/scope index seek") + } +} From 7a19a06631a1ccdff19ac8f082ec50c1da8c0a7a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 4 Oct 2026 16:01:54 +0100 Subject: [PATCH 2/2] Keep backup inventory guard bound to the shared evaluator The runtime preserves canonical backup inventory scope in recovery_rollups.go; update the existing ownership guard to follow that call and require both PVE/PBS callers rather than the former duplicated literal. Change-source: pulse-maintainer --- internal/monitoring/canonical_guardrails_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 5a35da2af8..30eeb870b5 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1261,6 +1261,10 @@ func TestBackupOrphanDetectionUsesCanonicalInventoryReadinessScope(t *testing.T) "func (m *Monitor) updatePVEBackupTemplateSubjectsForType(instanceName, guestType string, subjects map[string]struct{}) {", "func (m *Monitor) updatePVEBackupTemplateSubjectsFromClusterResources(instanceName string, resources []proxmox.ClusterResource) {", "func (m *Monitor) backupInventoryScopeForAlerts() *alerts.BackupInventoryScope {", + "m.checkBackupAlerts(ctx)", + "m.checkBackupAlerts(context.Background())", + }, + "recovery_rollups.go": { "m.alertManager.CheckBackupsWithInventory(rollups, guestsByKey, guestsByVMID, m.backupInventoryScopeForAlerts())", }, "monitor_pve_guest_poll.go": {