fix(pbs): evaluate configured datastore alerts on live polls

Connect fresh PBS storage observations to the existing capacity and connectivity evaluator. Exercise HTTP polling through unified sync, absent counter retention, confirmed empty recovery and recurrence with alternate counter names.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,28 @@
 
 ## Purpose
 
+### PBS datastore alert evaluation belongs to the live poll
+
+After publishing freshly polled PBS datastore storage rows, the poller invokes
+the existing storage alert evaluator with capacity-trend evidence. It must not
+depend on mock ticks or unified metric evaluation: that path does not evaluate
+PBS storage thresholds. Existing storage policy, aliases, suppression and
+connectivity semantics remain alerts-owned; this wiring adds no new policy.
+
+Absent or partial capacity counters must not resolve an existing usage
+incident. Explicit empty-store counters (positive total, zero used, free equal
+to total) may recover it. Repeated polls and subsequent unified alert sync must
+preserve the usage incident identity, including recurrence with alternate PBS
+counter names. Independent backup-posture incidents remain separate.
+
+Verification: `TestPBSPolledCapacityRequiresObservedRecovery` in
+`internal/monitoring/monitor_pbs_coverage_test.go` crosses synthetic HTTP
+decoding, live polling and unified sync. It checks an 80% threshold at 85%,
+missing counter retention, confirmed-empty recovery and 86% recurrence with an
+explicit one-point minimum delta. This is local capacity-path evidence, not
+installed recipient receipt, restart qualification or exhaustive connectivity
+validation.
+
 ### Host-local addresses are not PVE identity
 
 Automatic host/PVE network matching excludes non-global-unicast addresses,

internal/monitoring/monitor_pbs_coverage_test.go

@@ -8,12 +8,14 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"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/unifiedresources"
 	"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
 )
 
@@ -568,3 +570,82 @@ func TestPBSMetricAvailabilityAlertLifecycle(t *testing.T) {
 		t.Fatal("missing resolved history")
 	}
 }
+
+// Exercise HTTP decoding, poll evaluation and subsequent unified alert sync together.
+// This is synthetic integration evidence, not installed notification receipt.
+func TestPBSPolledCapacityRequiresObservedRecovery(t *testing.T) {
+	var response atomic.Value
+	response.Store(`{"data":{"total":1000,"used":850,"avail":150}}`)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch r.URL.Path {
+		case "/api2/json/version":
+			_, _ = w.Write([]byte(`{"data":{"version":"3.4.2"}}`))
+		case "/api2/json/nodes/localhost/status":
+			_, _ = w.Write([]byte(`{"data":{"cpu":0.1,"memory":{"used":100,"total":1000}}}`))
+		case "/api2/json/admin/datastore":
+			_, _ = w.Write([]byte(`{"data":[{"store":"backups"}]}`))
+		case "/api2/json/admin/datastore/backups/status":
+			_, _ = w.Write([]byte(response.Load().(string)))
+		default:
+			_, _ = w.Write([]byte(`{"data":[]}`))
+		}
+	}))
+	defer server.Close()
+	manager := alerts.NewManagerWithDataDir(t.TempDir())
+	defer manager.Stop()
+	manager.UpdateConfig(alerts.AlertConfig{Enabled: true, ActivationState: alerts.ActivationActive, MinimumDelta: 1,
+		TimeThresholds: map[string]int{"storage": 0}, StorageDefault: alerts.HysteresisThreshold{Trigger: 80, Clear: 70}})
+	instance := config.PBSInstance{Name: "pbs-capacity", Host: server.URL, MonitorDatastores: true}
+	monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance})
+	monitor.alertManager = manager
+	client := newPBSHealthTestClient(t, server.URL)
+	adapter := unifiedresources.NewMonitorAdapter(nil)
+	poll := func() {
+		t.Helper()
+		for range 5 {
+			monitor.pollPBSInstance(context.Background(), instance.Name, client)
+			adapter.PopulateFromSnapshot(monitor.state.GetSnapshot())
+			monitor.syncUnifiedResourceAlertsToState(adapter.GetAll())
+		}
+	}
+	poll()
+	active := manager.GetActiveAlerts()
+	if len(active) != 1 || active[0].Type != "usage" || active[0].Value != 85 {
+		t.Fatalf("expected one polled capacity incident: %+v", active)
+	}
+	original := active[0]
+	if original.ResourceID != "pbs-pbs-capacity-backups" {
+		t.Fatalf("unexpected datastore identity: %q", original.ResourceID)
+	}
+	for _, missing := range []string{
+		`{"data":null}`,
+		`{"data":{}}`,
+		`{"data":{"total":1000}}`,
+		`{"data":{"total":1000,"used":0}}`,
+	} {
+		response.Store(missing)
+		poll()
+		active = manager.GetActiveAlerts()
+		if len(active) != 1 || active[0].ID != original.ID || active[0].Value != 85 || !active[0].StartTime.Equal(original.StartTime) || manager.GetResolvedAlert(original.ID) != nil {
+			t.Fatalf("missing capacity %s changed incident: %+v", missing, active)
+		}
+	}
+	response.Store(`{"data":{"total":1000,"used":0,"avail":1000}}`)
+	poll()
+	if len(manager.GetActiveAlerts()) != 0 {
+		t.Fatal("observed empty datastore did not recover")
+	}
+	if resolved := manager.GetResolvedAlert(original.ID); resolved == nil || resolved.Value != 0 || !resolved.StartTime.Equal(original.StartTime) {
+		t.Fatalf("incorrect recovery: %+v", resolved)
+	}
+	// Alternate PBS counter names must feed the same policy and identity.
+	// Stay below the separate 90% backup-posture incident threshold; the
+	// configured minimum delta of one permits this immediate recurrence.
+	response.Store(`{"data":{"total-space":1000,"used-space":860,"avail-space":140}}`)
+	poll()
+	active = manager.GetActiveAlerts()
+	if len(active) != 1 || active[0].ID != original.ID || active[0].Value != 86 || !active[0].StartTime.After(original.StartTime) {
+		t.Fatalf("incorrect recurrent incident: %+v", active)
+	}
+}

internal/monitoring/monitor_pbs_pmg.go

@@ -502,6 +502,14 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
 			pbsStorages = append(pbsStorages, pbsStorage)
 		}
 		m.state.UpdateStorageForInstance("pbs-"+instanceName, pbsStorages)
+		// PBS storage is not evaluated by the unified metric path (which
+		// handles TrueNAS and VMware storage). Evaluate fresh poll observations
+		// here, as the PVE storage poller does, rather than relying on mock ticks.
+		if m.alertManager != nil {
+			for _, storage := range pbsStorages {
+				m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, time.Now()))
+			}
+		}
 		log.Debug().
 			Str("instance", instanceName).
 			Int("storageEntries", len(pbsStorages)).