package pbs import ( "context" "errors" "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) // probeAuth is the R-39 credential probe seam (default = (*Client).ProbeAuth). Overridable so the // auth-honesty path is testable with no PBS. probeAuth func(ctx context.Context, t Target) error // authSink receives every probe result. nil = nobody is listening (the probe is then skipped // entirely — no point paying for a request nothing consumes). authSink AuthSink } // AuthSink receives the result of each per-storage credential probe (R-39 leg c). // // It exists so the DR bridge can turn a 401 into a LOUD `auth_failed` state instead of the Warn-and- // skip that made an applied-but-dead tier invisible. Deliberately a plain interface taking a bool // rather than the error: the consumer (internal/pbsdr) must not have to import this package just to // test a sentinel. type AuthSink interface { // NoteAuthResult reports one storage's credential health. unauthorized=true means PBS REJECTED // the credential (401) — terminal until it is replaced. A transport error is unauthorized=false // with a non-empty detail: unknown, not dead, and never a reason to re-key. NoteAuthResult(storageID string, unauthorized bool, detail string) } // SetAuthSink wires the credential-probe consumer. Without it the reporter does not probe at all. func (r *LiveSnapshotReporter) SetAuthSink(s AuthSink) { r.authSink = s } // 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, probeAuth: func(ctx context.Context, t Target) error { return t.Client.ProbeAuth(ctx) }, } } // 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 } // probeAuthAndReport runs the credential probe for one target and forwards the verdict to the sink. // No sink → no probe (nothing would consume it). Never fails the collect: a report must still go out. func (r *LiveSnapshotReporter) probeAuthAndReport(ctx context.Context, t Target) { if r.authSink == nil || r.probeAuth == nil { return } err := r.probeAuth(ctx, t) switch { case err == nil: // Logged even on success: a safety mechanism that is silent when healthy cannot be shown to be // RUNNING, and "no auth_failed" is indistinguishable from "never probed". Debug level, so it // costs nothing in normal operation but is one log-level away when it matters. r.log.Debug("pbs: credential probe OK", "storage", t.StorageID, "datastore", t.Datastore) r.authSink.NoteAuthResult(t.StorageID, false, "") case errors.Is(err, ErrUnauthorized): // The one case that is TERMINAL and actionable: the credential is rejected, not the network. r.log.Error("pbs: the DR endpoint REJECTED this box's credential (401) — the storage entry is "+ "pinned to a superseded secret and every backup/restore against it will fail", "storage", t.StorageID, "datastore", t.Datastore) r.authSink.NoteAuthResult(t.StorageID, true, "PBS rejected the stored credential (401)") default: // Unreachable/timeout/TLS — UNKNOWN, not dead. Reporting this as unauthorized would re-key a // perfectly good credential on every network blip. r.log.Debug("pbs: credential probe inconclusive (not a rejection)", "storage", t.StorageID, "err", err) r.authSink.NoteAuthResult(t.StorageID, false, err.Error()) } } // 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 { // R-39 leg (c): prove the credential BEFORE interpreting anything else about this datastore. // A snapshot list that fails with 401 used to look identical to "PBS is busy" — which is how // an applied-and-dead tier stayed green. The probe runs on this 15-minute collect path (not // the 6 h verify cadence) because that is how fast the hub can react. r.probeAuthAndReport(childCtx, t) 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 }