v0.39.0 — DR-recipe completion: live PBS coord + drop role/restic_repo_coord from v1 drive shape

Live PBS coord: new internal/pbs/live_reporter.go (LiveSnapshotReporter implements hub.PBSReporter via the cheap Client.Snapshots() list with last-known-good fallback, bounded by an 8s timeout, list-only — never triggers a verify). Closes the gap where the recipe's pbs block was omitted whenever the verify-loop SnapshotStore was empty (one-shot collect + the first ~6h after a daemon restart). SnapshotStore.Get added (per-datastore LKG). Wired into the collector in both runDaemon and runSelftestHub; the verify loop keeps Recording into the SAME shared store via one hoisted pbsTargets. v1 host-half drive shape: dropped drives[].role (hub/operator-owned manifest concept, not host-derivable) and drives[].restic_repo_coord (named a backup tier that doesn't exist). Drive shape is now {durable_id, mount_path, intent, fs_type?, total_bytes}. Hub reads drives as json.RawMessage → no hub struct change; goldens re-pinned byte-identical (agent + hub copies). Tests: live_reporter_test.go (T1 load-bearing coord-without-verify + T2..T6), TestDRRecipeHostHalf_V1DriveShape; each companion demonstrated to fail pre-fix then reverted. go build/vet/test green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-16 20:25:09 +02:00
parent 8abc1b8852
commit fbe113011d
8 changed files with 414 additions and 23 deletions
@@ -0,0 +1,97 @@
+package pbs
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// DefaultLiveSnapshotTimeout bounds the per-collect live PBS reads so a slow/hung PBS can never
+// stall the host-report (and thus the heartbeat). A few seconds is ample for the cheap list GET on
+// the LAN; on overrun the reporter falls back to last-known-good rather than blocking.
+const DefaultLiveSnapshotTimeout = 8 * time.Second
+
+// LiveSnapshotReporter resolves the PBS snapshot inventory LIVE on each collect (the cheap
+// Snapshots() list), so the host-report — and the DR recipe's pbs coord derived from it — is present
+// whenever PBS is reachable, INDEPENDENT of the 6 h verify cadence. This closes the gap where a
+// one-shot collect (selftest=hub) and the first window of every daemon after a restart saw the
+// verify-loop's SnapshotStore still empty (→ no pbs coord, the restore SOURCE missing).
+//
+// On a per-datastore live error/timeout it falls back to the store's last-known-good; a successful
+// (even empty) response is authoritative and updates the store. If targets cannot be resolved at all
+// it returns the store's full aggregate. It shares the same *SnapshotStore the verify loop Records
+// into, so the two keep each other's last-known-good warm. It NEVER triggers a verify — list only.
+type LiveSnapshotReporter struct {
+	targets Targets        // pbsTargetsFromPVE(...) closure — re-resolved each collect
+	store   *SnapshotStore // shared last-known-good cache (verify loop writes it too)
+	timeout time.Duration  // per-collect bound on the live reads
+	log     *slog.Logger
+
+	// listSnapshots is the production→PBS seam, overridable in tests so no live PBS is needed.
+	// Default = liveListSnapshots (one Snapshots() GET, converted via Snapshot.ToHub()).
+	listSnapshots func(ctx context.Context, t Target) ([]hub.PBSSnapshot, error)
+}
+
+// NewLiveSnapshotReporter builds a live reporter sharing store with the verify loop. A zero timeout
+// falls back to DefaultLiveSnapshotTimeout; a nil logger to slog.Default.
+func NewLiveSnapshotReporter(targets Targets, store *SnapshotStore, timeout time.Duration, log *slog.Logger) *LiveSnapshotReporter {
+	if timeout <= 0 {
+		timeout = DefaultLiveSnapshotTimeout
+	}
+	if log == nil {
+		log = slog.Default()
+	}
+	return &LiveSnapshotReporter{
+		targets:       targets,
+		store:         store,
+		timeout:       timeout,
+		log:           log,
+		listSnapshots: liveListSnapshots,
+	}
+}
+
+// liveListSnapshots does ONE cheap snapshot list for a target and converts each record to the hub
+// wire shape (reusing Snapshot.ToHub — RFC3339 backup_time, verify_state). List only; no verify.
+func liveListSnapshots(ctx context.Context, t Target) ([]hub.PBSSnapshot, error) {
+	snaps, err := t.Client.Snapshots(ctx, t.Datastore)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]hub.PBSSnapshot, 0, len(snaps))
+	for _, s := range snaps {
+		out = append(out, s.ToHub())
+	}
+	return out, nil
+}
+
+// PBSSnapshots implements hub.PBSReporter with a single bounded, live pass (last-known-good
+// fallback). Non-nil result so it marshals as [].
+func (r *LiveSnapshotReporter) PBSSnapshots(ctx context.Context) []hub.PBSSnapshot {
+	childCtx, cancel := context.WithTimeout(ctx, r.timeout)
+	defer cancel()
+
+	targets, err := r.targets(childCtx)
+	if err != nil {
+		// Cannot even enumerate datastores → fall back to the full last-known-good aggregate.
+		r.log.Debug("pbs: live targets resolution failed; using last-known-good aggregate", "err", err)
+		return r.store.PBSSnapshots(ctx)
+	}
+
+	out := []hub.PBSSnapshot{}
+	for _, t := range targets {
+		snaps, err := r.listSnapshots(childCtx, t)
+		if err != nil {
+			// Per-datastore live failure → that datastore's last-known-good (does NOT clobber it).
+			r.log.Debug("pbs: live list failed; using last-known-good", "datastore", t.Datastore, "err", err)
+			out = append(out, r.store.Get(t.Datastore)...)
+			continue
+		}
+		// A successful response (INCLUDING empty) is authoritative — record it as the new
+		// last-known-good so a later error reuses fresh truth, not a stale set.
+		r.store.Record(t.Datastore, snaps)
+		out = append(out, snaps...)
+	}
+	return out
+}
@@ -0,0 +1,186 @@
+package pbs
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// fakeReporter builds a LiveSnapshotReporter with an injected snapshot-lister seam and a targets
+// closure, so no live PBS is needed. listFn is keyed by datastore; targetsErr forces a targets-
+// resolution failure. The Target.Client is a throwaway &Client{} the seam never dereferences.
+func fakeReporter(store *SnapshotStore, datastores []string, targetsErr error,
+	listFn func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error)) *LiveSnapshotReporter {
+	targets := func(ctx context.Context) ([]Target, error) {
+		if targetsErr != nil {
+			return nil, targetsErr
+		}
+		out := make([]Target, 0, len(datastores))
+		for _, ds := range datastores {
+			out = append(out, Target{Datastore: ds, Client: &Client{}})
+		}
+		return out, nil
+	}
+	r := NewLiveSnapshotReporter(targets, store, 2*time.Second, nil)
+	r.listSnapshots = func(ctx context.Context, t Target) ([]hub.PBSSnapshot, error) {
+		return listFn(ctx, t.Datastore)
+	}
+	return r
+}
+
+func snap(ds, backupID, backupTime string) hub.PBSSnapshot {
+	return hub.PBSSnapshot{Namespace: "root", BackupType: "ct", BackupID: backupID, BackupTime: backupTime}
+}
+
+// T1 (load-bearing): coord present with NO prior verify. A fresh store (verify never ran) + a live
+// lister returning 2 snapshots → PBSSnapshots returns both, and feeding them through
+// BuildDRRecipeHostHalf yields a non-nil pbs whose latest_snapshot_id is the LATER snapshot.
+func TestLiveReporter_CoordPresentWithoutPriorVerify(t *testing.T) {
+	store := NewSnapshotStore() // empty — simulates a just-restarted daemon, no verify yet
+	r := fakeReporter(store, []string{"felhom-spike"}, nil,
+		func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error) {
+			return []hub.PBSSnapshot{
+				snap(ds, "9201", "2026-06-13T20:00:00Z"),
+				snap(ds, "9201", "2026-06-16T17:00:00Z"), // latest
+			}, nil
+		})
+
+	got := r.PBSSnapshots(context.Background())
+	if len(got) != 2 {
+		t.Fatalf("want 2 live snapshots, got %d", len(got))
+	}
+
+	// The whole point: a recipe built from the live read carries the pbs coord.
+	h := hub.BuildDRRecipeHostHalf(nil,
+		[]hub.StorageTarget{{Name: "felhom-pbs", Type: hub.StorageTypePBS, Content: "backup"}}, got)
+	if h.PBS == nil {
+		t.Fatal("pbs coord absent despite a reachable PBS — the gap this fixes")
+	}
+	if h.PBS.RepoID != "felhom-pbs" || h.PBS.Namespace != "root" || h.PBS.LatestSnapshotID != "9201" {
+		t.Errorf("pbs coord = %+v, want felhom-pbs/root/9201", h.PBS)
+	}
+
+	// COMPANION (pre-fix): the bare SnapshotStore (no live read) with an empty store omits pbs.
+	bare := NewSnapshotStore()
+	h2 := hub.BuildDRRecipeHostHalf(nil,
+		[]hub.StorageTarget{{Name: "felhom-pbs", Type: hub.StorageTypePBS, Content: "backup"}},
+		bare.PBSSnapshots(context.Background()))
+	if h2.PBS != nil {
+		t.Fatal("companion sanity: the bare store should yield NO pbs coord (proves the live read is load-bearing)")
+	}
+}
+
+// T2: live error → last-known-good fallback (and the store is NOT clobbered).
+func TestLiveReporter_ErrorFallsBackToLastKnownGood(t *testing.T) {
+	store := NewSnapshotStore()
+	store.Record("D", []hub.PBSSnapshot{snap("D", "9001", "2026-06-10T00:00:00Z")})
+
+	r := fakeReporter(store, []string{"D"}, nil,
+		func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error) {
+			return nil, errors.New("pbs unreachable")
+		})
+
+	got := r.PBSSnapshots(context.Background())
+	if len(got) != 1 || got[0].BackupID != "9001" {
+		t.Fatalf("want the last-known-good snapshot on live error, got %+v", got)
+	}
+	// The store must still hold the prior set (a failed live read must not wipe it).
+	if lkg := store.Get("D"); len(lkg) != 1 || lkg[0].BackupID != "9001" {
+		t.Errorf("live error clobbered the store: %+v", lkg)
+	}
+
+	// COMPANION (pre-fix): without the fallback branch (return empty on error) the result is empty.
+	noFallback := func(ctx context.Context) []hub.PBSSnapshot {
+		targets, _ := r.targets(ctx)
+		out := []hub.PBSSnapshot{}
+		for _, tg := range targets {
+			if _, err := r.listSnapshots(ctx, tg); err != nil {
+				continue // the mutation: drop the LKG-append branch
+			}
+		}
+		return out
+	}
+	if len(noFallback(context.Background())) != 0 {
+		t.Fatal("companion sanity: the no-fallback variant should return empty")
+	}
+}
+
+// T3: a successful response updates the store (last-known-good warm for a later error).
+func TestLiveReporter_SuccessUpdatesStore(t *testing.T) {
+	store := NewSnapshotStore() // empty
+	r := fakeReporter(store, []string{"D"}, nil,
+		func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error) {
+			return []hub.PBSSnapshot{snap("D", "9201", "2026-06-16T00:00:00Z")}, nil
+		})
+	_ = r.PBSSnapshots(context.Background())
+	if lkg := store.Get("D"); len(lkg) != 1 || lkg[0].BackupID != "9201" {
+		t.Errorf("store not warmed by a successful live read: %+v", lkg)
+	}
+}
+
+// T4: targets-resolution error → the full last-known-good aggregate (not empty).
+func TestLiveReporter_TargetsErrorReturnsAggregate(t *testing.T) {
+	store := NewSnapshotStore()
+	store.Record("D1", []hub.PBSSnapshot{snap("D1", "1", "2026-06-10T00:00:00Z")})
+	store.Record("D2", []hub.PBSSnapshot{snap("D2", "2", "2026-06-11T00:00:00Z")})
+
+	r := fakeReporter(store, nil, errors.New("ListStorage failed"),
+		func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error) {
+			t.Fatal("listSnapshots must not be called when targets fail to resolve")
+			return nil, nil
+		})
+	got := r.PBSSnapshots(context.Background())
+	if len(got) != 2 {
+		t.Fatalf("want the 2-snapshot aggregate on a targets-resolution error, got %d (%+v)", len(got), got)
+	}
+}
+
+// T5: bounded. A lister that blocks until ctx is done must return within ~timeout (the child-ctx
+// deadline) and fall back, not hang. A short timeout keeps the test fast + deterministic.
+func TestLiveReporter_BoundedByTimeout(t *testing.T) {
+	store := NewSnapshotStore()
+	store.Record("D", []hub.PBSSnapshot{snap("D", "9001", "2026-06-10T00:00:00Z")})
+
+	targets := func(ctx context.Context) ([]Target, error) {
+		return []Target{{Datastore: "D", Client: &Client{}}}, nil
+	}
+	r := NewLiveSnapshotReporter(targets, store, 100*time.Millisecond, nil)
+	r.listSnapshots = func(ctx context.Context, t Target) ([]hub.PBSSnapshot, error) {
+		<-ctx.Done() // block until the child-ctx deadline fires
+		return nil, ctx.Err()
+	}
+
+	done := make(chan []hub.PBSSnapshot, 1)
+	go func() { done <- r.PBSSnapshots(context.Background()) }()
+	select {
+	case got := <-done:
+		// Fell back to last-known-good rather than hanging.
+		if len(got) != 1 || got[0].BackupID != "9001" {
+			t.Errorf("want last-known-good after the deadline, got %+v", got)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("PBSSnapshots hung past the timeout — not bounded by the child ctx")
+	}
+}
+
+// T6: an empty-but-successful response is authoritative — it OVERWRITES a prior non-empty set.
+// (Documents the chosen semantics: a real empty datastore must be reflected, not masked by stale LKG.)
+func TestLiveReporter_EmptySuccessIsAuthoritative(t *testing.T) {
+	store := NewSnapshotStore()
+	store.Record("D", []hub.PBSSnapshot{snap("D", "9001", "2026-06-10T00:00:00Z")})
+
+	r := fakeReporter(store, []string{"D"}, nil,
+		func(ctx context.Context, ds string) ([]hub.PBSSnapshot, error) {
+			return []hub.PBSSnapshot{}, nil // success, but the datastore is now empty
+		})
+	got := r.PBSSnapshots(context.Background())
+	if len(got) != 0 {
+		t.Errorf("empty-but-successful should yield an empty result, got %+v", got)
+	}
+	if lkg := store.Get("D"); len(lkg) != 0 {
+		t.Errorf("empty success should overwrite the store to empty, got %+v", lkg)
+	}
+}
@@ -72,6 +72,21 @@ func (s *SnapshotStore) Record(datastore string, snaps []hub.PBSSnapshot) {
 	s.byDatastore[datastore] = snaps
 }

+// Get returns a copy of the last-known-good snapshot set for ONE datastore (nil when absent or
+// empty). It is the per-datastore fallback the live reporter reaches for when a live list fails —
+// distinct from PBSSnapshots, which aggregates every datastore.
+func (s *SnapshotStore) Get(datastore string) []hub.PBSSnapshot {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	src := s.byDatastore[datastore]
+	if len(src) == 0 {
+		return nil
+	}
+	out := make([]hub.PBSSnapshot, len(src))
+	copy(out, src)
+	return out
+}
+
 // PBSSnapshots implements hub.PBSReporter — all known snapshots across datastores.
 func (s *SnapshotStore) PBSSnapshots(context.Context) []hub.PBSSnapshot {
 	s.mu.Lock()