diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index db4a561206..782ffeeb9d 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2642,7 +2642,14 @@ the newest bounded set of real fetched snapshots, and the per-instance PBS backup list must not grow without an explicit monitoring-owned limit. Bounding must never synthesize placeholder backup entries from group metadata: a placeholder drops verification, size, file, and per-snapshot time data, which -users read as broken discovery and failed verification. PBS backup group cache metadata +users read as broken discovery and failed verification. Reusing a PBS snapshot +cache requires the group's newest backup time to equal the cached newest time, +not merely be no later. Pruning the newest snapshot can move that time backwards +while both the old and new inventories still fill the capped cache; the ordinary +poll must refresh then, withdraw the deleted archive after an accepted read and +derive posture from the remaining backups. Failed refreshes retain the existing +partial/unavailable evidence rules, and unchanged groups retain bounded TTL reuse. +PBS backup group cache metadata must be pruned to the retained group set after a completed poll, while preserving cache metadata only for groups still observed or intentionally reused after transient datastore failures. Recovery-point ingestion started by backup diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index b667611fad..b64eb24ef1 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -1695,13 +1695,15 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien } } - // Only re-fetch when the backup count changes, the most recent backup - // is newer, or the cache TTL has expired (to pick up verification changes). + // Reuse only while both retained count and newest time still match. + // Pruning can move last-backup backwards without changing the + // capped cache size; that is removal evidence, not an unchanged group. + // The TTL also bounds reuse when only verification changes. if hasCachedData && cacheStillFresh && cacheCountMatches && !cachedHasInProgress && - !lastBackupTime.After(cached.latest) { + lastBackupTime.Equal(cached.latest) { allBackups = appendPBSBackupsWithinLimit(allBackups, cached.snapshots) groupsReused++ diff --git a/internal/monitoring/pbs_protection_observation_test.go b/internal/monitoring/pbs_protection_observation_test.go index 73e6253aed..4667265c3c 100644 --- a/internal/monitoring/pbs_protection_observation_test.go +++ b/internal/monitoring/pbs_protection_observation_test.go @@ -1,11 +1,22 @@ package monitoring import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" "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" + "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" ) func TestBuildPBSProtectionProviderObservationMapsEvidenceQuality(t *testing.T) { @@ -115,3 +126,161 @@ func TestBuildPBSProtectionProviderObservationMapsEvidenceQuality(t *testing.T) }) } } + +// The bounded cache has 100 rows both before and after deleting the newest +// of 101 snapshots. A backwards last-backup value must invalidate it even +// though its retained count still matches and its TTL has not expired. +func TestPBSOrdinaryPollNewestRemovalInvalidatesCappedCache(t *testing.T) { + for _, kind := range []string{"vm", "ct"} { + for _, failure := range []int{0, http.StatusServiceUnavailable, http.StatusForbidden} { + t.Run(fmt.Sprintf("%s/refresh-status-%d", kind, failure), func(t *testing.T) { + f := newPBSPosturePollFixture(t, 1, kind) + var phase, snapshotReads atomic.Int32 + oldAt := time.Unix(f.backupAt, 0).Add(-8 * 24 * time.Hour).Unix() + restoredAt := f.backupAt + 3600 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mode := phase.Load() + latest, count := f.backupAt, pbsBackupSnapshotsPerGroupLimit+1 + if mode == 1 || mode == 2 { + latest, count = oldAt, pbsBackupSnapshotsPerGroupLimit + } else if mode == 3 { + latest = restoredAt + } else if mode == 4 { + latest, count = 0, 0 + } + switch { + case strings.HasSuffix(r.URL.Path, "/groups"): + _ = json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{ + "backup-type": kind, "backup-id": "100", "last-backup": latest, "backup-count": count, + }}}) + case strings.HasSuffix(r.URL.Path, "/snapshots"): + snapshotReads.Add(1) + if mode == 2 { + http.Error(w, "synthetic unreadable inventory", failure) + return + } + rows := make([]map[string]any, 0, count) + for i := 0; i < count; i++ { + at := oldAt - int64(i)*3600 + if i == 0 && (mode == 0 || mode == 3) { + at = latest + } else if mode == 0 || mode == 3 { + at += 3600 + } + rows = append(rows, map[string]any{"backup-type": kind, "backup-id": "100", "backup-time": at, + "size": 1024, "files": []string{"index.json.blob"}, "verification": map[string]any{"state": "ok"}, "comment": "guest-100"}) + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": rows}) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(server.Close) + var err error + f.client, err = pbs.NewClient(pbs.ClientConfig{Host: server.URL, TokenName: "fixture@pbs!reader", TokenValue: "fixture-only"}) + if err != nil { + t.Fatal(err) + } + ds := []models.PBSDatastore{{Name: "archive"}} + poll := func(mode int32) { + 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.wait(t) + } + frames := map[string]any{} + capture := func(name string) { + points := f.points(t, "pbs-main") + frames[name] = map[string]any{ + "pbs": map[string]any{"data": map[string]any{"backups": f.m.state.GetSnapshot().PBSBackups}}, + "postures": map[string]any{"data": f.postures(t), "policy": recovery.DefaultProtectionPosturePolicy.Payload()}, + "point_count": len(points), "snapshot_reads": snapshotReads.Load(), + } + } + assertLatest := func(at int64, state recovery.ProtectionState) { + t.Helper() + raw, points, postures := f.m.state.GetSnapshot().PBSBackups, f.points(t, "pbs-main"), f.postures(t) + if len(raw) != pbsBackupSnapshotsPerGroupLimit || len(points) != pbsBackupSnapshotsPerGroupLimit || len(postures) != 1 { + t.Fatalf("raw/persisted/postures=%d/%d/%d, want %d/%d/1", len(raw), len(points), len(postures), pbsBackupSnapshotsPerGroupLimit, pbsBackupSnapshotsPerGroupLimit) + } + for _, row := range raw { + if row.BackupTime.Unix() > at { + t.Errorf("removed newest archive is still offered: %s > %s", row.BackupTime, time.Unix(at, 0)) + } + } + for _, point := range points { + if point.CompletedAt == nil || point.CompletedAt.Unix() > at || point.SubjectResourceID != f.ids[0] { + t.Errorf("persisted point is removed or belongs to the wrong subject: %s at %v", point.ID, point.CompletedAt) + } + } + p := postures[0] + if p.State != state || p.LastSuccessfulPointAt == nil || p.LastSuccessfulPointAt.Unix() != at { + t.Errorf("posture=%s latest=%v, want %s at %s", p.State, p.LastSuccessfulPointAt, state, time.Unix(at, 0)) + } + } + key := pbsBackupGroupKey{datastore: "archive", backupType: kind, backupID: "100"} + poll(0) + assertLatest(f.backupAt, recovery.ProtectionStateProtected) + capture("observed") + readAt := f.m.pbsBackupCacheTimeFor("pbs-main", key) + poll(0) + if snapshotReads.Load() != 1 || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(readAt) { + t.Error("unchanged healthy group reread or renewed its snapshot cache") + } + if failure != 0 { + for range 2 { + before := snapshotReads.Load() + poll(2) + if snapshotReads.Load() <= before { + t.Error("backwards group metadata did not trigger a snapshot refresh") + } + p := f.postures(t)[0] + if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryPartial { + t.Errorf("failed prune refresh claimed protected/complete: %+v", p) + } + if len(f.points(t, "pbs-main")) != pbsBackupSnapshotsPerGroupLimit || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(readAt) { + t.Error("failed prune refresh deleted history or renewed its cache") + } + } + capture("unreadable") + } + before := snapshotReads.Load() + poll(1) + if snapshotReads.Load() != before+1 { + t.Errorf("pruned newest did not refresh snapshots: %d -> %d", before, snapshotReads.Load()) + } + assertLatest(oldAt, recovery.ProtectionStateAttention) + capture("removed") + readAt = f.m.pbsBackupCacheTimeFor("pbs-main", key) + before = snapshotReads.Load() + poll(1) + assertLatest(oldAt, recovery.ProtectionStateAttention) + if snapshotReads.Load() != before || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(readAt) { + t.Error("unchanged pruned inventory did not resume cheap TTL reuse") + } + poll(3) + assertLatest(restoredAt, recovery.ProtectionStateProtected) + capture("recovered") + poll(4) + if len(f.m.state.GetSnapshot().PBSBackups) != 0 || len(f.points(t, "pbs-main")) != 0 || !f.m.pbsBackupCacheTimeFor("pbs-main", key).IsZero() { + t.Error("confirmed empty group did not clear its inventory, recovery points and cache") + } + capture("empty") + t.Logf("kind=%s failure=%d retained-limit=%d snapshot-reads=%d; newest removal, failed refresh, stale posture and recovery checked", kind, failure, pbsBackupSnapshotsPerGroupLimit, snapshotReads.Load()) + if os.Getenv("PULSE_PBS_PRUNING_FRAMES") == "1" && kind == "vm" && failure == http.StatusServiceUnavailable { + source, err := os.ReadFile("monitor_backups.go") + if err != nil { + t.Fatal(err) + } + data, err := json.Marshal(map[string]any{"runtime_sha256": fmt.Sprintf("%x", sha256.Sum256(source)), + "observed_at": time.Now().UTC(), "resource_ids": f.ids, "frames": frames}) + if err != nil { + t.Fatal(err) + } + fmt.Printf("PBS_PRUNING_FRAMES=%s\n", data) + } + }) + } + } +}