Integrate reviewed backup evaluation query and outage repair

Change-source: pulse-maintainer

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,

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

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

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")
+	}
+}

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"

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 = &copy
+		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")
+	}
+}

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": {
@@ -3070,3 +3074,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)
+		}
+	}
+}

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

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

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()

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)
+}

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)
+				}
+			}
+		})
+	}
+}

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

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)
+	}
+}

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")
+	}
+}