hub v0.56.0: PBS-DR self-heal reconciler (re-stage a consumable secret)

Implements SPIKE-pbsdr-selfheal-2026-07-15 (e8f8c44). A box re-installed/rolled back onto its stable host_id loses its agent-side converged marker; the hub keeps the enabled descriptor + a CONSUMED one-time secret, the WG peer persists (changed==false, cascade can't re-fire), so the agent sits in waiting_secret forever. The missing piece is a consumable secret, not the descriptor. New internal/pbsdrheal reconciler (5m, wgsync shape): for enabled+provisioned hosts whose latest report pbs_dr.state is a stuck state past a >=2-distinct-report debounce, re-stage the stored secret (store.RestageHostPBSSecret: clear consumed_at, no ep0 call, NO generation bump); escalate to Re-issue (web ReissuePBSDR) only when no secret is stored or the agent reports consumed_failed. Converged/disabled/verify_failed/DR-OFF = no-op. PBSDRHEAL_ONLY_HOST scopes a supervised rollout. Scenarios A-F + all six red-proofs verified. No agent change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HEPuEwyyGDJdcsXLFsTWJn
2026-07-15 18:27:39 +02:00
parent e8f8c441fa
commit 6218e7919d
9 changed files with 878 additions and 43 deletions
@@ -0,0 +1,276 @@
+package pbsdrheal
+
+// Non-hollow reconciler tests mapping 1:1 to the TASK's integration scenarios A–F. A REAL store
+// (t.TempDir sqlite) supplies the work-set + reports; a FAKE Actions seam counts restage/reissue
+// calls without any SSH/ep0. reconcileOnce is driven directly (same package) for determinism. Audit
+// events are asserted against the real store. Each test is the durable guard for its §10 red-proof.
+
+import (
+	"context"
+	"io"
+	"log"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+type fakeActions struct {
+	mu            sync.Mutex
+	restageHosts  []string
+	reissueCusts  []string
+	restageResult bool // what Restage returns (does a stored row exist?)
+	restageErr    error
+	reissueErr    error
+}
+
+func (f *fakeActions) Restage(hostID string) (bool, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.restageHosts = append(f.restageHosts, hostID)
+	return f.restageResult, f.restageErr
+}
+
+func (f *fakeActions) Reissue(ctx context.Context, customerID string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.reissueCusts = append(f.reissueCusts, customerID)
+	return f.reissueErr
+}
+
+func (f *fakeActions) restages() int { f.mu.Lock(); defer f.mu.Unlock(); return len(f.restageHosts) }
+func (f *fakeActions) reissues() int { f.mu.Lock(); defer f.mu.Unlock(); return len(f.reissueCusts) }
+
+func newHealStore(t *testing.T) *store.Store {
+	t.Helper()
+	st, err := store.New(filepath.Join(t.TempDir(), "test.db"), log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatalf("store.New: %v", err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+// seedHost writes a host with a pbs_dr descriptor of the requested enable/provision shape.
+func seedHost(t *testing.T, st *store.Store, hostID, customerID string, enabled, provisioned bool) {
+	t.Helper()
+	if err := st.UpsertHost(&store.Host{HostID: hostID, CustomerID: customerID, APIKey: "k-" + hostID}); err != nil {
+		t.Fatalf("UpsertHost %s: %v", hostID, err)
+	}
+	ns := ""
+	if provisioned {
+		ns = customerID
+	}
+	desc := `{"pbs_dr":{"enabled":` + boolStr(enabled) + `,"namespace":"` + ns + `","storage_id":"felhom-pbs"}}`
+	if _, err := st.SetHostDesired(hostID, []byte(desc)); err != nil {
+		t.Fatalf("SetHostDesired %s: %v", hostID, err)
+	}
+}
+
+func seedReport(t *testing.T, st *store.Store, hostID, customerID, state string) {
+	t.Helper()
+	if err := st.SaveHostReport(hostID, customerID, []byte(`{"pbs_dr":{"state":"`+state+`"}}`), store.HostReportDenorm{AgentVersion: "0.88.0"}); err != nil {
+		t.Fatalf("SaveHostReport %s: %v", hostID, err)
+	}
+}
+
+func boolStr(b bool) string {
+	if b {
+		return "true"
+	}
+	return "false"
+}
+
+func newRec(st *store.Store, act Actions) *Reconciler {
+	r := NewReconciler(st, act, log.New(io.Discard, "", 0))
+	r.debounceReports = 1 // fire on the first stuck report unless a test overrides
+	return r
+}
+
+func genOf(t *testing.T, st *store.Store, hostID string) int64 {
+	t.Helper()
+	h, err := st.GetHost(hostID)
+	if err != nil || h == nil {
+		t.Fatalf("GetHost %s: (%v, %v)", hostID, h, err)
+	}
+	return h.DesiredGeneration
+}
+
+func eventTypes(t *testing.T, st *store.Store, customerID string) []string {
+	t.Helper()
+	evs, err := st.GetRecentEvents(customerID, 50)
+	if err != nil {
+		t.Fatalf("GetRecentEvents: %v", err)
+	}
+	var out []string
+	for _, e := range evs {
+		out = append(out, e.EventType)
+	}
+	return out
+}
+
+// A — the headline: a re-installed box (descriptor enabled+provisioned, stored CONSUMED secret,
+// waiting_secret) is re-staged — NOT reissued, and NO generation bump.
+func TestScenarioA_WaitingSecretRestages(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hA", "cA", true, true)
+	seedReport(t, st, "hA", "cA", stateWaitingSecret)
+	genBefore := genOf(t, st, "hA")
+	fake := &fakeActions{restageResult: true} // a stored secret exists
+	r := newRec(st, fake)
+
+	r.reconcileOnce(context.Background())
+
+	if fake.restages() != 1 || fake.reissues() != 0 {
+		t.Fatalf("restages=%d reissues=%d, want 1/0 (re-stage, not re-issue)", fake.restages(), fake.reissues())
+	}
+	if fake.restageHosts[0] != "hA" {
+		t.Errorf("restaged host = %s, want hA", fake.restageHosts[0])
+	}
+	if g := genOf(t, st, "hA"); g != genBefore {
+		t.Errorf("generation bumped by a re-stage heal: %d -> %d", genBefore, g)
+	}
+	if got := eventTypes(t, st, "cA"); len(got) != 1 || got[0] != eventRestaged {
+		t.Errorf("events = %v, want [%s]", got, eventRestaged)
+	}
+}
+
+// B — no stored secret → escalate to Re-issue exactly once.
+func TestScenarioB_NoStoredSecretReissues(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hB", "cB", true, true)
+	seedReport(t, st, "hB", "cB", stateWaitingSecret)
+	fake := &fakeActions{restageResult: false} // no row to re-stage
+	r := newRec(st, fake)
+
+	r.reconcileOnce(context.Background())
+
+	if fake.restages() != 1 || fake.reissues() != 1 {
+		t.Fatalf("restages=%d reissues=%d, want 1/1 (tried re-stage, then escalated)", fake.restages(), fake.reissues())
+	}
+	if fake.reissueCusts[0] != "cB" {
+		t.Errorf("reissued customer = %s, want cB", fake.reissueCusts[0])
+	}
+	if got := eventTypes(t, st, "cB"); len(got) != 1 || got[0] != eventReissued {
+		t.Errorf("events = %v, want [%s]", got, eventReissued)
+	}
+}
+
+// C — a converged/healthy host is a pure no-op (idempotency; the gen-thrash guard).
+func TestScenarioC_ConvergedHostNoOp(t *testing.T) {
+	for _, state := range []string{"applied", "adopted", "disabled", "verify_failed"} {
+		t.Run(state, func(t *testing.T) {
+			st := newHealStore(t)
+			seedHost(t, st, "hC", "cC", true, true)
+			seedReport(t, st, "hC", "cC", state)
+			genBefore := genOf(t, st, "hC")
+			fake := &fakeActions{restageResult: true}
+			r := newRec(st, fake)
+
+			r.reconcileOnce(context.Background())
+
+			if fake.restages() != 0 || fake.reissues() != 0 {
+				t.Fatalf("state %s: restages=%d reissues=%d, want 0/0 (no-op)", state, fake.restages(), fake.reissues())
+			}
+			if g := genOf(t, st, "hC"); g != genBefore {
+				t.Errorf("state %s: generation changed on a healthy host", state)
+			}
+			if got := eventTypes(t, st, "cC"); len(got) != 0 {
+				t.Errorf("state %s: events = %v, want none", state, got)
+			}
+		})
+	}
+}
+
+// D — debounce: a single waiting_secret report does NOT heal; a second DISTINCT one does.
+func TestScenarioD_DebounceRequiresTwoDistinctReports(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hD", "cD", true, true)
+	fake := &fakeActions{restageResult: true}
+	r := NewReconciler(st, fake, log.New(io.Discard, "", 0)) // default debounceReports = 2
+
+	seedReport(t, st, "hD", "cD", stateWaitingSecret) // report #1
+	r.reconcileOnce(context.Background())
+	if fake.restages() != 0 {
+		t.Fatalf("healed on the FIRST waiting_secret report (restages=%d) — debounce broken", fake.restages())
+	}
+	// Re-observing the SAME report must also not advance the debounce.
+	r.reconcileOnce(context.Background())
+	if fake.restages() != 0 {
+		t.Fatalf("healed on a RE-OBSERVED same report (restages=%d) — debounce must count distinct reports", fake.restages())
+	}
+	seedReport(t, st, "hD", "cD", stateWaitingSecret) // report #2 (distinct)
+	r.reconcileOnce(context.Background())
+	if fake.restages() != 1 {
+		t.Fatalf("did not heal after 2 distinct waiting_secret reports (restages=%d)", fake.restages())
+	}
+}
+
+// E — consumed_failed is escalated (Re-issue), never re-staged (a re-stage re-feeds a burned secret).
+func TestScenarioE_ConsumedFailedEscalates(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hE", "cE", true, true)
+	seedReport(t, st, "hE", "cE", stateConsumedFailed)
+	fake := &fakeActions{restageResult: true} // even if a secret were re-stageable, must NOT re-stage
+	r := newRec(st, fake)
+
+	r.reconcileOnce(context.Background())
+
+	if fake.restages() != 0 || fake.reissues() != 1 {
+		t.Fatalf("restages=%d reissues=%d, want 0/1 (consumed_failed → re-issue only)", fake.restages(), fake.reissues())
+	}
+	if got := eventTypes(t, st, "cE"); len(got) != 1 || got[0] != eventConsumedFailed {
+		t.Errorf("events = %v, want [%s] (distinct consumed_failed signal)", got, eventConsumedFailed)
+	}
+}
+
+// F — DR-OFF (disabled descriptor) and enabled-but-unprovisioned hosts are never in the work set.
+func TestScenarioF_OutOfScopeHostsUntouched(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hOff", "cOff", false, true)   // descriptor disabled
+	seedReport(t, st, "hOff", "cOff", stateWaitingSecret)
+	seedHost(t, st, "hUnprov", "cUnprov", true, false) // enabled but never provisioned
+	seedReport(t, st, "hUnprov", "cUnprov", stateWaitingSecret)
+	fake := &fakeActions{restageResult: true}
+	r := newRec(st, fake)
+
+	r.reconcileOnce(context.Background())
+
+	if fake.restages() != 0 || fake.reissues() != 0 {
+		t.Fatalf("out-of-scope hosts acted on: restages=%d reissues=%d, want 0/0", fake.restages(), fake.reissues())
+	}
+}
+
+// RestrictToHost scopes the work set: a stuck host OTHER than the allowed one is untouched.
+func TestRestrictToHost_ScopesWorkSet(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hAllowed", "cAllowed", true, true)
+	seedReport(t, st, "hAllowed", "cAllowed", stateWaitingSecret)
+	seedHost(t, st, "hOther", "cOther", true, true)
+	seedReport(t, st, "hOther", "cOther", stateWaitingSecret)
+	fake := &fakeActions{restageResult: true}
+	r := newRec(st, fake)
+	r.RestrictToHost("hAllowed")
+
+	r.reconcileOnce(context.Background())
+
+	if fake.restages() != 1 || fake.restageHosts[0] != "hAllowed" {
+		t.Fatalf("restaged hosts = %v, want exactly [hAllowed] (scope restriction)", fake.restageHosts)
+	}
+}
+
+// After a heal, the SAME stuck report must not re-heal on the next tick (only a fresh report does).
+func TestNoReHealOnSameReport(t *testing.T) {
+	st := newHealStore(t)
+	seedHost(t, st, "hR", "cR", true, true)
+	seedReport(t, st, "hR", "cR", stateWaitingSecret)
+	fake := &fakeActions{restageResult: true}
+	r := newRec(st, fake)
+
+	r.reconcileOnce(context.Background())
+	r.reconcileOnce(context.Background()) // same report, no new evidence
+	if fake.restages() != 1 {
+		t.Fatalf("re-healed the same stuck report: restages=%d, want 1", fake.restages())
+	}
+}