Keep failed PBS namespace discovery out of complete backup evidence (#3157)

**Keep failed PBS namespace discovery out of complete backup evidence**

Carry namespace discovery errors with the immutable datastore snapshot
into background backup polling. Readable root history still updates, but
unknown child namespaces cannot delete durable facts or claim complete
protection. Cover real HTTP discovery through ingest, replacement state,
repeated failures, scoped removal, recovery and legacy root-only
support.

**Initialize the connection outcome map in namespace poll regressions**

The fixture drives the real connection poll and therefore needs its
ordinary poll-status map, in addition to backup scheduling and recovery
state. Keep the new lifecycle proof independent of fixture setup panics.

**Account for accepted new archives beside retained partial PBS
history**

The namespace lifecycle test advances the root archive timestamp. A
partial instance-wide scan must retain the two historical points and
accept that new root point, not replace all root history. Assert every
historical point remains byte-identical and exactly one independently
accepted new point is present.

**Distinguish failed PBS discovery from an empty backup scope**

Carry failed and accepted-empty discovery through ordinary serialized
backup polling. Preserve unreadable history, reconcile only confirmed
absence, and reject malformed datastore envelopes before destructive
reconciliation.

Contract-Neutral: Enforces existing PBS scope-evidence semantics; no
API, model, policy or permission change.

**Model an existing PBS interval in the skipped-scope regression**

The live interval setter deliberately schedules a catch-up poll. Test
not-due admission with an already-configured interval and its retained
completed-poll timestamp instead.

---
Fleet candidate `20261023T220242Z-core-runtime` from the `core-runtime`
lane. Reviewed tip `d1ecdcb93916f236f966897230541a66850fe3a5` on `main`.

[How Pulse handles
triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)
This commit is contained in:
pulse-triage[bot]
2026-10-23 23:18:11 +00:00
committed by GitHub
13 changed files with 725 additions and 104 deletions
@@ -6907,6 +6907,21 @@ without deleting retained artifacts. The observation is persisted before
reconciliation, so cached backup files cannot continue presenting current
protection after collection becomes incomplete or unauthorized.
PBS namespace discovery is part of that collection evidence. An unavailable,
malformed or denied namespace listing must travel with the matching datastore
snapshot into background backup polling. A readable root namespace may still
supply accepted artifacts, but cannot turn failed child-namespace discovery
into a complete root-only inventory. Such a poll records partial history with
unknown or partial access, preserves unseen durable history and never guesses
child namespace requests. Recovery and a later successful namespace removal
resume ordinary complete, connection-scoped reconciliation. The PBS client's
existing legacy root-only response for an absent namespace endpoint is unchanged.
`TestPBSNamespaceDiscoveryFailureRetainsRecoveryHistory` crosses real HTTP
namespace discovery, background polling, durable ingest and canonical posture
on warm and replacement inventories; recovery, scoped removal, first-read and
legacy controls cover both safe retention and authoritative absence. No public
model, endpoint, policy, guest operation or restorable-backup claim is added.
Direct PBS backup enumeration emits two separate storage/recovery inputs:
subject-linked recovery points and one typed provider observation for the
polled PBS instance. A complete poll records complete history with sufficient
@@ -6917,6 +6932,29 @@ denied access. Retained backup points survive failed enumeration, but the new
provider observation immediately prevents those cached points from being
presented as current protection truth.
Datastore discovery itself is part of that backup-scope result, including
the connectivity fallback. An admitted failed or malformed listing records
unavailable history while retaining existing raw artifacts, durable points and
snapshot-cache ages; only HTTP 401/403 establishes denied access. An explicit
successful empty array (including a fully excluded scope) instead reconciles
the connection's backup points, live rows and cache timestamps. Missing/null
`data` and unnamed datastore entries are rejected before any detail request,
not interpreted as authoritative absence. Both outcomes use the existing
serialized background backup admission, cadence and runtime deadline.
Disabled, not-due, in-flight and cancelled work is not an accepted empty scope;
backup availability remains separate from node connectivity. Ordinary complete
recovery resumes collection and protection without credential or guest changes.
`TestGetDatastoresRequiresExplicitNamedScope` and
`TestGetDatastoresAcceptsExplicitEmptyScope` pin the client envelope boundary.
`TestPBSDatastoreDiscoveryFailureQualifiesRetainedHistory` crosses real HTTP,
ordinary background polling, warm/rebuilt live state, durable ingestion and
canonical posture for repeated 401/403/404/503 and malformed listings.
`TestPBSDatastoreDiscoveryEmptyReconcilesOnlyAcceptedScope`,
`TestPBSDatastoreDiscoverySkippedScopeIsNotEmpty` and
`TestPBSDatastoreDiscoveryFailedConnectivityFallback` cover scoped removal,
exclusions, reappearance, admission/cancellation and failed fallback. These are
synthetic source checks, not installed PBS or guest/restore acceptance.
PBS mapping attaches provider scope and a typed evidence envelope to every
successfully enumerated recovery point. Identity correlation is confirmed only
for direct canonical identity and inferred only for an auditable unique
+10 -1
View File
@@ -1579,7 +1579,7 @@ type pbsBackupFetchRequest struct {
}
// pollPBSBackups fetches all backups from PBS datastores
func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, client *pbs.Client, datastores []models.PBSDatastore) {
func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, client *pbs.Client, datastores []models.PBSDatastore, namespaceErrors map[string]error) {
log.Debug().Str("instance", instanceName).Msg("polling PBS backups")
// Cache existing PBS backups so we can avoid redundant API calls when no changes occurred.
@@ -1615,6 +1615,15 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
datastoreHadSuccess := false
datastoreNamespaceErrors := 0
datastoreTerminalNamespaceErrors := 0
// The root namespace remains readable when discovering child namespaces
// fails. Its accepted rows are useful, but they cannot establish that
// undiscovered namespaces are empty or renew complete-provider evidence.
if err := namespaceErrors[ds.Name]; err != nil {
datastoreNamespaceErrors++
if !shouldReuseCachedPBSBackups(err) {
datastoreTerminalNamespaceErrors++
}
}
groupsReused := 0
groupsRequested := 0
@@ -348,7 +348,7 @@ func TestPollPBSBackupsFetchesRealSnapshotsForLargeGroups(t *testing.T) {
}
m := &Monitor{state: models.NewState()}
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil)
if got := atomic.LoadInt64(&snapshotCalls); got != 1 {
t.Fatalf("large PBS backup group fetched snapshots %d times, want 1", got)
@@ -383,7 +383,7 @@ func TestPollPBSBackupsFetchesRealSnapshotsForLargeGroups(t *testing.T) {
}
// A second poll must reuse the bounded cache instead of refetching.
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil)
if got := atomic.LoadInt64(&snapshotCalls); got != 1 {
t.Fatalf("second poll refetched snapshots (%d calls), want cached reuse", got)
}
@@ -448,7 +448,7 @@ func TestPollPBSBackupsBoundsLargeTopologyAcrossPolls(t *testing.T) {
m := &Monitor{state: models.NewState()}
for cycle := 0; cycle < 3; cycle++ {
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil)
snapshot := m.state.GetSnapshot()
if got := len(snapshot.PBSBackups); got != pbsBackupLiveStateLimit {
@@ -615,7 +615,7 @@ func TestPollPBSBackupsPrunesCacheTimesForDeletedGroups(t *testing.T) {
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{
{Name: "archive"},
})
}, nil)
m.mu.RLock()
perGroup := m.pbsBackupCacheTime["pbs1"]
@@ -1039,7 +1039,7 @@ func TestPollPBSBackups_DropsStaleCacheOnTerminalDatastoreError(t *testing.T) {
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{
{Name: "archive"},
})
}, nil)
snapshot := m.state.GetSnapshot()
for _, backup := range snapshot.PBSBackups {
@@ -1097,7 +1097,7 @@ func TestPollPBSBackups_PreservesCacheOnTransientDatastoreError(t *testing.T) {
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{
{Name: "archive"},
})
}, nil)
snapshot := m.state.GetSnapshot()
var found bool
@@ -2394,7 +2394,7 @@ func TestMonitor_PollPBSBackups_Extra(t *testing.T) {
t.Fatal(err)
}
ds := []models.PBSDatastore{{Name: "ds1"}}
m.pollPBSBackups(context.Background(), "pbs1", client, ds)
m.pollPBSBackups(context.Background(), "pbs1", client, ds, nil)
}
func TestMonitor_RetryPVEPortFallback_Detailed_Extra(t *testing.T) {
@@ -222,7 +222,7 @@ func TestPollPBSBackups(t *testing.T) {
}
// Execute
m.pollPBSBackups(context.Background(), "pbs1", client, datastores)
m.pollPBSBackups(context.Background(), "pbs1", client, datastores, nil)
// Verify
found := false
@@ -1115,7 +1115,7 @@ func TestPollPBSBackupsCachedIncompleteSnapshotLosesStaleTaskObservation(t *test
WriteActive: false,
}})
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil)
if snapshotRequests.Load() != 1 {
t.Fatalf("snapshot requests = %d, want 1 cached fallback attempt", snapshotRequests.Load())
@@ -3,6 +3,7 @@ package monitoring
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"os"
@@ -15,6 +16,7 @@ import (
"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"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
@@ -809,3 +811,38 @@ func TestPBSPolledCapacityRequiresObservedRecovery(t *testing.T) {
t.Fatalf("raised policy did not clear high-usage alert: %+v", active)
}
}
func TestPBSNamespaceDiscoveryFirstPollDoesNotInventScope(t *testing.T) {
for _, status := range []int{http.StatusServiceUnavailable, http.StatusForbidden} {
t.Run(fmt.Sprintf("status-%d", status), func(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.namespaceStatus.Store(int32(status))
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 1 || f.childReads.Load() != 0 {
t.Fatal("failed first discovery invented a child namespace or lost readable root history")
}
for _, p := range f.postures(t) {
if p.State == recovery.ProtectionStateProtected {
t.Errorf("incomplete first discovery claimed current protection: %+v", p)
}
}
})
}
}
func TestPBSNamespaceDiscoveryLegacyRootOnlyRemainsComplete(t *testing.T) {
f := newPBSNamespacePollFixture(t)
// The existing PBS client treats an absent namespace endpoint on legacy
// versions as a successful root-only inventory. Preserve that support.
f.namespaceStatus.Store(http.StatusNotFound)
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 1 || f.childReads.Load() != 0 {
t.Fatal("legacy root-only inventory was not collected")
}
for _, p := range f.postures(t) {
if p.SubjectResourceID == f.ids[0] && (p.State != recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 ||
p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryComplete) {
t.Errorf("legacy root-only support was marked incomplete: %+v", p)
}
}
}
+93 -87
View File
@@ -13,6 +13,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/logging"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors"
"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
"github.com/rcourtman/pulse-go-rewrite/pkg/pmg"
@@ -319,6 +320,11 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) {
m.recordAuthFailure(instanceName, "pbs")
}
// The connectivity fallback also failed datastore discovery. Qualify
// retained backup evidence through the ordinary serialized admission.
if !stderrors.Is(datastoreErr, context.Canceled) {
m.pollPBSBackupsAsync(instanceName, client, instanceCfg, nil, nil, datastoreErr)
}
return
}
}
@@ -358,10 +364,15 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
Msg("PBS node status retrieved")
}
// Keep discovery quality with the immutable input to the background backup
// poll. Public datastore presentation is not an authoritative scope receipt.
namespaceErrors := make(map[string]error)
var datastoreDiscoveryErr error
// Poll datastores if enabled
if instanceCfg.MonitorDatastores {
datastores, err := client.GetDatastoresFiltered(ctx, includeDatastore)
if err != nil {
datastoreDiscoveryErr = err
monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0)
log.Error().Err(monErr).Str("instance", instanceName).Msg("failed to get datastores")
} else {
@@ -417,6 +428,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0)
if err != nil {
namespaceErrors[ds.Store] = err
log.Warn().Err(err).
Str("instance", instanceName).
Str("datastore", ds.Store).
@@ -520,96 +532,90 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
Msg("Added PBS datastores to unified storage view")
}
// Poll backups if enabled
if instanceCfg.MonitorBackups {
if len(pbsInst.Datastores) == 0 {
log.Debug().
Str("instance", instanceName).
Msg("No PBS datastores available for backup polling")
} else if !m.backupPollingEnabledSetting() {
log.Debug().
Str("instance", instanceName).
Msg("Skipping PBS backup polling - globally disabled")
} else {
now := time.Now()
m.mu.Lock()
lastPoll := m.lastPBSBackupPoll[instanceName]
if m.pbsBackupPollers == nil {
m.pbsBackupPollers = make(map[string]bool)
}
inProgress := m.pbsBackupPollers[instanceName]
m.mu.Unlock()
shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now)
if !shouldPoll {
if reason != "" {
log.Debug().
Str("instance", instanceName).
Str("reason", reason).
Msg("Skipping PBS backup polling this cycle")
}
} else if inProgress {
log.Debug().
Str("instance", instanceName).
Msg("PBS backup polling already in progress")
} else {
datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores))
copy(datastoreSnapshot, pbsInst.Datastores)
// Atomically check and set poller flag
m.mu.Lock()
if m.pbsBackupPollers[instanceName] {
// Race: another goroutine started between our check and lock
m.mu.Unlock()
log.Debug().
Str("instance", instanceName).
Msg("PBS backup polling started by another goroutine")
} else {
m.pbsBackupPollers[instanceName] = true
m.lastPBSBackupPoll[instanceName] = newLast
m.mu.Unlock()
go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) {
defer recoverFromPanic(fmt.Sprintf("pollPBSBackups-%s", inst))
defer func() {
m.mu.Lock()
delete(m.pbsBackupPollers, inst)
m.lastPBSBackupPoll[inst] = time.Now()
m.mu.Unlock()
}()
log.Info().
Str("instance", inst).
Int("datastores", len(ds)).
Msg("Starting background PBS backup polling")
// The per-cycle ctx is canceled as soon as the main polling loop finishes,
// so derive the backup poll context from the long-lived runtime context instead.
parentCtx := m.getRuntimeContext()
if parentCtx == nil {
parentCtx = context.Background()
}
backupCtx, cancel := context.WithTimeout(parentCtx, 5*time.Minute)
defer cancel()
m.pollPBSBackups(backupCtx, inst, pbsClient, ds)
log.Info().
Str("instance", inst).
Dur("duration", time.Since(start)).
Msg("Completed background PBS backup polling")
}(datastoreSnapshot, instanceName, now, client)
}
}
}
} else {
log.Debug().
Str("instance", instanceName).
Msg("PBS backup monitoring disabled")
// A successful empty scope and a failed discovery both need an accepted
// backup-poll result; neither may silently keep the last good protection.
if !stderrors.Is(datastoreDiscoveryErr, context.Canceled) {
m.pollPBSBackupsAsync(instanceName, client, instanceCfg, pbsInst.Datastores, namespaceErrors, datastoreDiscoveryErr)
}
}
// pollPBSBackupsAsync admits all backup-scope outcomes through one worker.
// Failed discovery never becomes an empty enumeration, and a skipped/disabled
// discovery is not absence evidence. Keeping failure/empty publication inside
// this admission prevents it from racing an older in-flight backup worker.
func (m *Monitor) pollPBSBackupsAsync(instanceName string, client *pbs.Client, instanceCfg *config.PBSInstance, datastores []models.PBSDatastore, namespaceErrors map[string]error, discoveryErr error) {
if !instanceCfg.MonitorBackups || !instanceCfg.MonitorDatastores || !m.backupPollingEnabledSetting() {
return
}
now := time.Now()
m.mu.RLock()
lastPoll := m.lastPBSBackupPoll[instanceName]
m.mu.RUnlock()
shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now)
if !shouldPoll {
log.Debug().Str("instance", instanceName).Str("reason", reason).Msg("Skipping PBS backup polling this cycle")
return
}
datastoreSnapshot := append([]models.PBSDatastore(nil), datastores...)
m.mu.Lock()
if m.pbsBackupPollers[instanceName] {
m.mu.Unlock()
log.Debug().Str("instance", instanceName).Msg("PBS backup polling already in progress")
return
}
if m.pbsBackupPollers == nil {
m.pbsBackupPollers = make(map[string]bool)
}
if m.lastPBSBackupPoll == nil {
m.lastPBSBackupPoll = make(map[string]time.Time)
}
m.pbsBackupPollers[instanceName] = true
m.lastPBSBackupPoll[instanceName] = newLast
m.mu.Unlock()
go func() {
defer recoverFromPanic(fmt.Sprintf("pollPBSBackups-%s", instanceName))
defer func() {
m.mu.Lock()
delete(m.pbsBackupPollers, instanceName)
m.lastPBSBackupPoll[instanceName] = time.Now()
m.mu.Unlock()
}()
// The main polling loop cancels its per-cycle context after returning.
// Backup work keeps the existing independent, runtime-scoped deadline.
parentCtx := m.getRuntimeContext()
if parentCtx == nil {
parentCtx = context.Background()
}
backupCtx, cancel := context.WithTimeout(parentCtx, 5*time.Minute)
defer cancel()
if backupCtx.Err() != nil {
return
}
if discoveryErr != nil {
// One unreadable discovery scope, not a guessed datastore count.
// Only actual authentication/authorization status establishes denial.
denied := 0
if status, ok := pbs.HTTPStatus(discoveryErr); ok && (status == 401 || status == 403) {
denied = 1
}
observation, err := buildPBSProtectionProviderObservation(instanceName, 1, 0, 1, denied, time.Now().UTC())
if err != nil {
log.Warn().Err(err).Str("instance", instanceName).Msg("Failed to build PBS discovery failure observation")
return
}
m.ingestProtectionProviderObservationsAsync([]recovery.ProtectionProviderObservation{observation})
} else {
// The explicit empty list is authoritative and reconciles only this
// connection; the ordinary poller also drops its live rows/cache.
m.pollPBSBackups(backupCtx, instanceName, client, datastoreSnapshot, namespaceErrors)
}
log.Info().Str("instance", instanceName).Dur("duration", time.Since(now)).Msg("Completed background PBS backup polling")
}()
}
// recordPBSNodeHistory writes the API-observed CPU and memory measurements to
// the PBS service's existing metrics target (agent:<PBS source ID>). When no
// host agent can be corroborated, the Backups drawer still has useful history
@@ -0,0 +1,199 @@
package monitoring
import (
"context"
"fmt"
"net/http"
"reflect"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
)
func TestPBSDatastoreDiscoveryFailureQualifiesRetainedHistory(t *testing.T) {
for _, scenario := range []struct {
name string
status int
body string
denied bool
}{
{name: "server failure", status: 503},
{name: "unauthenticated", status: 401, denied: true},
{name: "forbidden", status: 403, denied: true},
{name: "missing endpoint is not permission evidence", status: 404},
{name: "malformed envelope", body: `{"data":`},
{name: "missing data", body: `{}`},
{name: "null data", body: `{"data":null}`},
{name: "nameless datastore", body: `{"data":[{}]}`},
} {
for _, cold := range []bool{false, true} {
t.Run(fmt.Sprintf("%s/cold-%t", scenario.name, cold), func(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-main")
before := f.points(t, "pbs-main")
if len(before) != 2 {
t.Fatal("healthy control did not ingest root and tenant")
}
if cold {
vms := f.m.state.GetSnapshot().VMs
f.m.state = models.NewState()
f.m.state.UpdateVMs(vms)
}
raw := f.m.state.GetSnapshot().PBSBackups
key := pbsBackupGroupKey{datastore: "archive", backupType: "vm", backupID: "100"}
cacheAt := f.m.pbsBackupCacheTimeFor("pbs-main", key)
reads := f.backupReads.Load()
f.datastoreStatus.Store(int32(scenario.status))
if scenario.body != "" {
f.datastoreBody.Store(&scenario.body)
}
for range 2 {
f.poll(t, "pbs-main")
if !reflect.DeepEqual(before, f.points(t, "pbs-main")) || !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PBSBackups) {
t.Error("failed discovery deleted or renewed retained artifacts")
}
if f.backupReads.Load() != reads || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(cacheAt) {
t.Error("failed discovery guessed a backup read or renewed cache")
}
permissions := operationaltrust.EvidencePermissionsUnknown
if scenario.denied {
permissions = operationaltrust.EvidencePermissionsDenied
}
for _, p := range f.postures(t) {
if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 {
t.Errorf("failed discovery left protection healthy: %+v", p)
continue
}
provider := p.ProviderStates[0]
if provider.HistoryCompleteness != recovery.ProtectionHistoryUnavailable || provider.Permissions != permissions || provider.JobState != recovery.OutcomeFailed {
t.Errorf("failure quality lost: %+v", provider)
}
}
// Backup history availability is not node connectivity.
inst := pbsInstanceByName(t, f.m.state.GetSnapshot(), "pbs-main")
if inst.Status != "online" || inst.ConnectionHealth != "healthy" {
t.Errorf("discovery failure rewrote connectivity: %+v", inst)
}
}
f.datastoreStatus.Store(0)
f.datastoreBody.Store(nil)
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 2 {
t.Error("ordinary recovery did not retain both archives")
}
for _, p := range f.postures(t) {
if p.State != recovery.ProtectionStateProtected {
t.Errorf("complete recovery did not restore protection: %+v", p)
}
}
})
}
}
}
func TestPBSDatastoreDiscoveryEmptyReconcilesOnlyAcceptedScope(t *testing.T) {
for _, excluded := range []bool{false, true} {
t.Run(fmt.Sprintf("excluded-%t", excluded), func(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-other")
other := f.points(t, "pbs-other")
f.poll(t, "pbs-main")
reads := f.backupReads.Load()
if excluded {
f.m.config.PBSInstances[0].ExcludeDatastores = []string{"archive"}
} else {
empty := `{"data":[]}`
f.datastoreBody.Store(&empty)
}
for range 2 {
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 0 {
t.Error("accepted empty scope retained nonexistent/removed archives")
}
for _, row := range f.m.state.GetSnapshot().PBSBackups {
if row.Instance == "pbs-main" {
t.Error("accepted empty scope retained a live archive")
}
}
if !reflect.DeepEqual(other, f.points(t, "pbs-other")) || f.backupReads.Load() != reads {
t.Error("empty reconciliation crossed scope or guessed a backup request")
}
f.m.mu.RLock()
cacheSize := len(f.m.pbsBackupCacheTime["pbs-main"])
f.m.mu.RUnlock()
if cacheSize != 0 {
t.Error("confirmed empty scope retained snapshot cache timestamps")
}
}
f.datastoreBody.Store(nil)
f.m.config.PBSInstances[0].ExcludeDatastores = nil
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 2 || !reflect.DeepEqual(other, f.points(t, "pbs-other")) {
t.Error("scope restoration did not rediscover archives independently")
}
})
}
}
func TestPBSDatastoreDiscoverySkippedScopeIsNotEmpty(t *testing.T) {
for _, skip := range []string{"backups disabled", "datastores disabled", "global disabled", "not due", "in flight", "cancelled cycle", "cancelled runtime"} {
t.Run(skip, func(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-main")
before, raw := f.points(t, "pbs-main"), f.m.state.GetSnapshot().PBSBackups
empty := `{"data":[]}`
f.datastoreBody.Store(&empty)
ctx := context.Background()
switch skip {
case "backups disabled":
f.m.config.PBSInstances[0].MonitorBackups = false
case "datastores disabled":
f.m.config.PBSInstances[0].MonitorDatastores = false
case "global disabled":
f.m.SetBackupPollingEnabled(false)
case "not due":
// Model an already-configured cadence. Changing cadence via the
// live setter deliberately clears timestamps for a catch-up poll.
f.m.config.BackupPollingInterval = time.Hour
case "in flight":
f.m.pbsBackupPollers["pbs-main"] = true
case "cancelled cycle":
cancelled, cancel := context.WithCancel(ctx)
cancel()
ctx = cancelled
case "cancelled runtime":
cancelled, cancel := context.WithCancel(ctx)
cancel()
f.m.runtimeCtx = cancelled
}
f.m.pollPBSInstance(ctx, "pbs-main", f.client)
if skip == "in flight" {
delete(f.m.pbsBackupPollers, "pbs-main")
}
f.wait(t)
if !reflect.DeepEqual(before, f.points(t, "pbs-main")) || !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PBSBackups) {
t.Error("skipped/cancelled discovery was treated as accepted empty scope")
}
})
}
}
func TestPBSDatastoreDiscoveryFailedConnectivityFallback(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-main")
before := f.points(t, "pbs-main")
f.versionStatus.Store(http.StatusServiceUnavailable)
f.datastoreStatus.Store(http.StatusServiceUnavailable)
f.poll(t, "pbs-main")
if !reflect.DeepEqual(before, f.points(t, "pbs-main")) {
t.Error("failed fallback destroyed history")
}
for _, p := range f.postures(t) {
if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryUnavailable {
t.Errorf("failed fallback left last good protection: %+v", p)
}
}
}
+4 -4
View File
@@ -74,7 +74,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) {
case "during-task-correlation":
datastores = []models.PBSDatastore{{Name: "archive"}}
}
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil)
f.wait(t)
f.assertProtected(t, "pbs-main")
key := pbsBackupGroupKey{datastore: "archive", backupType: "vm", backupID: "100"}
@@ -86,7 +86,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) {
done := make(chan struct{})
go func() {
defer close(done)
f.m.pollPBSBackups(ctx, "pbs-main", f.client, datastores)
f.m.pollPBSBackups(ctx, "pbs-main", f.client, datastores, nil)
}()
select {
case <-blocked:
@@ -116,7 +116,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) {
// Do not clear or expire cache again. The next ordinary poll must
// independently fetch the verification change the aborted poll lost.
phase.Store(2)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil)
f.wait(t)
raw = f.m.state.GetSnapshot().PBSBackups
mapped, err := proxmoxmapper.FromPBSBackupsWithEvidence(raw, buildPBSGuestCandidates(f.m.GetUnifiedReadStateOrSnapshot()), time.Now())
@@ -137,7 +137,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) {
}
// Once the matching failed row has been published, normal TTL reuse
// must avoid an unnecessary fourth read without restoring verified.
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil)
f.wait(t)
if snapshotReads.Load() != 3 || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(publishedAt) || f.postures(t)[0].Verification != recovery.ProtectionVerificationUnverified {
t.Error("cache-hit poll reread or renewed snapshots, or restored stale verification")
+2 -2
View File
@@ -131,7 +131,7 @@ func (f *pbsPosturePollFixture) poll(t *testing.T, instance string, mode int32,
f.m.mu.Lock()
clear(f.m.pbsBackupCacheTime)
f.m.mu.Unlock()
f.m.pollPBSBackups(context.Background(), instance, f.client, datastores)
f.m.pollPBSBackups(context.Background(), instance, f.client, datastores, nil)
f.wait(t)
}
@@ -289,7 +289,7 @@ func TestPBSOrdinaryPollSnapshotFailuresDoNotClaimCompleteHistory(t *testing.T)
}
// No TTL manipulation on the next poll: a failed refresh must not
// become a healthy cached read while snapshot access is still lost.
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds, nil)
f.wait(t)
}
f.poll(t, "pbs-main", 0, ds)
@@ -8,14 +8,18 @@ import (
"net/http"
"net/http/httptest"
"os"
"reflect"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/recovery"
recoverystore "github.com/rcourtman/pulse-go-rewrite/internal/recovery/store"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
)
@@ -186,7 +190,7 @@ func TestPBSOrdinaryPollNewestRemovalInvalidatesCappedCache(t *testing.T) {
phase.Store(mode)
// No cache clearing, TTL expiry or monitor replacement: these
// are consecutive ordinary polls of the same connection.
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds)
f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds, nil)
f.wait(t)
}
frames := map[string]any{}
@@ -284,3 +288,273 @@ func TestPBSOrdinaryPollNewestRemovalInvalidatesCappedCache(t *testing.T) {
}
}
}
// Namespace discovery is part of backup enumeration, not just optional drawer
// metadata. Drive the public poll entry point and its background backup worker
// so a missing namespace cannot masquerade as an authoritative root-only scan.
type pbsNamespacePollFixture struct {
m *Monitor
store *recoverystore.Store
client *pbs.Client
datastoreStatus atomic.Int32
datastoreBody atomic.Pointer[string]
versionStatus atomic.Int32
backupReads atomic.Int32
namespaceStatus atomic.Int32
namespaceRemoved atomic.Bool
rootEmpty atomic.Bool
backupAt atomic.Int64
childReads atomic.Int32
ids []string
}
func newPBSNamespacePollFixture(t *testing.T) *pbsNamespacePollFixture {
t.Helper()
m, manager := recoveryIngestTestMonitor(t)
f := &pbsNamespacePollFixture{m: m}
f.backupAt.Store(time.Now().UTC().Add(-time.Hour).Truncate(time.Second).Unix())
m.state = models.NewState()
m.state.UpdateVMs([]models.VM{
{ID: makeGuestID("cluster", "node-a", 100), VMID: 100, Instance: "cluster", Node: "node-a", Name: "guest-100", Status: "running"},
{ID: makeGuestID("cluster", "node-b", 101), VMID: 101, Instance: "cluster", Node: "node-b", Name: "guest-101", Status: "running"},
})
for _, id := range []int{100, 101} {
f.ids = append(f.ids, unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeVM, "cluster", id))
}
var err error
f.store, err = manager.StoreForOrg("default")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { f.wait(t); _ = f.store.Close() })
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
var data any = []any{}
switch r.URL.Path {
case "/api2/json/version":
if status := f.versionStatus.Load(); status != 0 {
http.Error(w, "fixture version unavailable", int(status))
return
}
data = map[string]any{"version": "3.4.2"}
case "/api2/json/admin/datastore":
if status := f.datastoreStatus.Load(); status != 0 {
http.Error(w, "fixture discovery unavailable", int(status))
return
}
if body := f.datastoreBody.Load(); body != nil {
_, _ = w.Write([]byte(*body))
return
}
data = []map[string]any{{"store": "archive"}}
case "/api2/json/admin/datastore/archive/status":
data = map[string]any{"total": 4096, "used": 1024, "avail": 3072}
case "/api2/json/admin/datastore/archive/namespace":
status := int(f.namespaceStatus.Load())
if status == 200 {
_, _ = w.Write([]byte(`{"data":`)) // incomplete, not a complete empty list
return
}
if status != 0 {
http.Error(w, "fixture namespace discovery unavailable", status)
return
}
if !f.namespaceRemoved.Load() {
data = []map[string]any{{"ns": "tenant"}}
}
case "/api2/json/admin/datastore/archive/groups":
f.backupReads.Add(1)
id := "100"
if r.URL.Query().Get("ns") == "tenant" {
id = "101"
f.childReads.Add(1)
}
if id != "100" || !f.rootEmpty.Load() {
data = []map[string]any{{"backup-type": "vm", "backup-id": id, "last-backup": f.backupAt.Load(), "backup-count": 1}}
}
case "/api2/json/admin/datastore/archive/snapshots":
f.backupReads.Add(1)
id := r.URL.Query().Get("backup-id")
if r.URL.Query().Get("ns") == "tenant" {
f.childReads.Add(1)
}
data = []map[string]any{{"backup-type": "vm", "backup-id": id, "backup-time": f.backupAt.Load(), "size": 1024,
"files": []string{"index.json.blob"}, "verification": map[string]any{"state": "ok"}, "comment": "guest-" + id}}
}
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
}))
t.Cleanup(server.Close)
f.client, err = pbs.NewClient(pbs.ClientConfig{Host: server.URL, TokenName: "fixture@pbs!reader", TokenValue: "fixture-only"})
if err != nil {
t.Fatal(err)
}
m.config = &config.Config{EnableBackupPolling: true, BackupPollingCycles: 1, PBSInstances: []config.PBSInstance{
{Name: "pbs-main", Host: server.URL, MonitorDatastores: true, MonitorBackups: true},
{Name: "pbs-other", Host: server.URL, MonitorDatastores: true, MonitorBackups: true},
}}
m.lastPBSBackupPoll = make(map[string]time.Time)
m.pollStatusMap = make(map[string]*pollStatus)
return f
}
func (f *pbsNamespacePollFixture) wait(t *testing.T) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for {
f.m.mu.RLock()
polling := len(f.m.pbsBackupPollers) > 0
f.m.mu.RUnlock()
f.m.recoveryIngestMu.Lock()
ingesting := f.m.recoveryIngestRunning
f.m.recoveryIngestMu.Unlock()
if !polling && !ingesting {
return
}
if time.Now().After(deadline) {
t.Fatal("PBS namespace poll/ingest did not finish")
}
time.Sleep(time.Millisecond)
}
}
func (f *pbsNamespacePollFixture) poll(t *testing.T, instance string) {
t.Helper()
f.m.pollPBSInstance(context.Background(), instance, f.client)
f.wait(t)
}
func (f *pbsNamespacePollFixture) points(t *testing.T, instance string) map[string]recovery.RecoveryPoint {
t.Helper()
points, _, err := f.store.ListPoints(context.Background(), recovery.ListPointsOptions{Limit: 1000})
if err != nil {
t.Fatal(err)
}
out := make(map[string]recovery.RecoveryPoint)
for _, point := range points {
if point.RepositoryRef != nil && point.RepositoryRef.Namespace == instance {
out[point.ID] = point
}
}
return out
}
func (f *pbsNamespacePollFixture) postures(t *testing.T) []recovery.ProtectionPosture {
t.Helper()
postures, total, err := f.store.ListProtectionPostures(context.Background(), recovery.ProtectionPostureQuery{SubjectResourceIDs: f.ids})
if err != nil || total != len(f.ids) {
t.Fatalf("postures total=%d err=%v", total, err)
}
return postures
}
func TestPBSNamespaceDiscoveryFailureRetainsRecoveryHistory(t *testing.T) {
for _, status := range []int{http.StatusServiceUnavailable, http.StatusForbidden, http.StatusOK} {
for _, cold := range []bool{false, true} {
t.Run(fmt.Sprintf("status-%d/cold-%t", status, cold), func(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-main")
before := f.points(t, "pbs-main")
if len(before) != 2 {
t.Fatalf("initial points=%d, want root and tenant", len(before))
}
for _, p := range f.postures(t) {
if p.State != recovery.ProtectionStateProtected {
t.Fatalf("healthy control=%+v", p)
}
}
if cold {
// Monitor replacement removes live inventory, not durable history.
vms := f.m.state.GetSnapshot().VMs
f.m.state = models.NewState()
f.m.state.UpdateVMs(vms)
}
f.namespaceStatus.Store(int32(status))
f.backupAt.Add(15 * 60)
reads := f.childReads.Load()
for range 2 {
f.poll(t, "pbs-main")
after := f.points(t, "pbs-main")
// A new root archive is accepted while the source-wide scan
// is partial. Retain both prior historical points, not only
// the unreadable child; reconcile old root history only when
// an ordinary complete enumeration is authoritative again.
if len(after) != len(before)+1 {
t.Errorf("partial discovery points=%d, want retained history plus one new root point (%d)", len(after), len(before)+1)
}
for id, point := range before {
if !reflect.DeepEqual(after[id], point) {
t.Error("historical point was deleted or renewed without an observation")
}
}
freshRootPoints := 0
for id, point := range after {
if _, historical := before[id]; historical {
continue
}
if point.Details["namespace"] != "" || point.CompletedAt == nil || point.CompletedAt.Unix() != f.backupAt.Load() {
t.Error("new point did not come from the accepted fresh root read")
}
freshRootPoints++
}
if freshRootPoints != 1 {
t.Errorf("accepted fresh root points=%d, want 1", freshRootPoints)
}
permissions := operationaltrust.EvidencePermissionsUnknown
if status == http.StatusForbidden {
permissions = operationaltrust.EvidencePermissionsPartial
}
for _, p := range f.postures(t) {
if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 ||
p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryPartial ||
p.ProviderStates[0].Permissions != permissions || p.ProviderStates[0].JobState != recovery.OutcomeWarning {
t.Errorf("discovery failure presented as complete/healthy: %+v", p)
}
}
if f.childReads.Load() != reads {
t.Error("failed discovery guessed a hidden namespace read")
}
}
})
}
}
}
func TestPBSNamespaceDiscoveryRecoveryAndScopedRemoval(t *testing.T) {
f := newPBSNamespacePollFixture(t)
f.poll(t, "pbs-other")
other := f.points(t, "pbs-other")
f.poll(t, "pbs-main")
f.namespaceStatus.Store(http.StatusServiceUnavailable)
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 2 {
t.Error("transient discovery failure removed tenant history")
}
f.namespaceStatus.Store(0)
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 2 {
t.Fatal("recovery did not restore namespace scope")
}
for _, p := range f.postures(t) {
for _, provider := range p.ProviderStates {
if provider.HistoryCompleteness != recovery.ProtectionHistoryComplete {
t.Errorf("recovered discovery still incomplete: %+v", p)
}
}
}
f.namespaceRemoved.Store(true)
f.poll(t, "pbs-main")
if got := len(f.points(t, "pbs-main")); got != 1 {
t.Errorf("authoritative namespace removal retained %d points, want root only", got)
}
if !reflect.DeepEqual(other, f.points(t, "pbs-other")) {
t.Error("namespace removal crossed connection scope")
}
f.rootEmpty.Store(true)
f.poll(t, "pbs-main")
if len(f.points(t, "pbs-main")) != 0 {
t.Error("authoritative complete empty scan retained points")
}
if !reflect.DeepEqual(other, f.points(t, "pbs-other")) {
t.Error("empty scan crossed connection scope")
}
}
+11
View File
@@ -933,6 +933,17 @@ func (c *Client) GetDatastoresFiltered(
Msg("Failed to parse PBS datastore list response")
return nil, fmt.Errorf("failed to parse datastore list: %w", err)
}
// Only an explicit array establishes the datastore scope. In particular,
// missing/null data and nameless rows are not an authoritative empty list:
// callers use a successful empty enumeration to reconcile backup history.
if datastoreList.Data == nil {
return nil, fmt.Errorf("datastore list response contains no data array")
}
for _, ds := range datastoreList.Data {
if strings.TrimSpace(ds.Store) == "" {
return nil, fmt.Errorf("datastore list response contains an unnamed datastore")
}
}
// Now get status for each datastore
var datastores []Datastore
@@ -0,0 +1,47 @@
package pbs
import (
"context"
"net/http"
"testing"
)
func TestGetDatastoresRequiresExplicitNamedScope(t *testing.T) {
for _, body := range []string{
`null`, `{}`, `{"data":null}`, `{"data":{}}`, `{"data":`,
`{"data":[null]}`, `{"data":[{}]}`, `{"data":[{"store":" "}]}`,
`{"data":[{"store":"valid"},{}]}`,
} {
t.Run(body, func(t *testing.T) {
detailReads := 0
client, server := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api2/json/admin/datastore" {
detailReads++
t.Error("malformed scope caused a datastore detail read")
}
_, _ = w.Write([]byte(body))
})
defer server.Close()
// Validate the whole scope before filtering or issuing detail reads;
// an exclusion must not turn malformed provider data into absence.
rows, err := client.GetDatastoresFiltered(context.Background(), func(string) bool { return false })
if err == nil || len(rows) != 0 || detailReads != 0 {
t.Fatalf("invalid scope accepted: rows=%v err=%v detailReads=%d", rows, err, detailReads)
}
})
}
}
func TestGetDatastoresAcceptsExplicitEmptyScope(t *testing.T) {
client, server := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api2/json/admin/datastore" {
t.Errorf("empty scope caused a detail read: %s", r.URL.Path)
}
_, _ = w.Write([]byte(`{"data":[]}`))
})
defer server.Close()
rows, err := client.GetDatastores(context.Background())
if err != nil || len(rows) != 0 {
t.Fatalf("explicit empty scope rejected: rows=%v err=%v", rows, err)
}
}