diff --git a/REPORT.md b/REPORT.md index dcd849f..54b1b3d 100644 --- a/REPORT.md +++ b/REPORT.md @@ -2,52 +2,45 @@ > **Overwrite** this file with a summary of the most recent task only (uniform with the other repos; not cumulative). The cumulative hub history lives in [hub/CHANGELOG.md](hub/CHANGELOG.md); the scripts history lives in [scripts/CHANGELOG.md](scripts/CHANGELOG.md). -## CAMPAIGN-6C — supervised planes + v0.131.0 F-S2/F-S3 live acceptance — 2026-07-14 +## PBS-DR self-heal reconciler (hub v0.56.0) — 2026-07-15 -**controller 0.132.0 both guests · agent 0.88.0 · hub 0.54.0 · findings-only, no production code · full -audit: `documentation/audits/CAMPAIGN-6C-2026-07-14.md` · ledger `180:~/campaign6/6C-P0.md` (seed -`6C-20260714`)** +**Task:** implement `SPIKE-pbsdr-selfheal-2026-07-15` (`e8f8c44`) — a hub periodic reconciler that +re-stages a consumable PBS-DR secret for boxes stuck in `waiting_secret`/`consumed_failed` after +losing their agent-side converged marker (re-install / restore / snapshot rollback onto a stable +`host_id`). Class: risky/supervised. Repo: `felhom.eu` (`hub/`) only — **no agent change** (the spike +proved the agent already self-heals once a secret is consumable). -Continuation of CAMPAIGN-6B. Ran **unattended** via the controller's own API (Campaign-4 credential, -out-of-band; session cookie + CSRF; the exact endpoints the dashboard invokes — the live-test-real-flow -proxy). No credential/R/blob committed; cookie wiped at teardown. +**Baselines:** hub main `e8f8c44` → this work; hub live image v0.55.0 → **v0.56.0**. Green gate +(`go build/vet/test ./...` in `hub/`) passing across the module. -### Verdict: GREEN for the v0.131.0 acceptance +**What shipped** +- `internal/store/pbsdr.go` — `RestageHostPBSSecret(hostID) (bool, error)`: clears `consumed_at` IFF + a secret row exists (no INSERT, no value change, **no generation bump**); `restaged=false` → no row + → the caller escalates. `PBSDRHealStates()`: one query joining each host's descriptor + enable/provision flags to its latest report `pbs_dr.state` + id (mirrors `GetHostOOBStates`). +- `internal/pbsdrheal/reconciler.go` (new) — 5-min reconciler (wgsync shape). Work set = descriptor + **enabled + provisioned**. Per host, on the latest report `pbs_dr.state` held across a **debounce + (≥2 distinct reports)**: `waiting_secret` → re-stage the stored secret (no ep0 call, no gen bump), + or escalate to Re-issue if none stored; `consumed_failed` → escalate to Re-issue only. + Converged/`disabled`/`verify_failed`/unprovisioned/DR-OFF = pure no-op. Distinct audit events per + heal. `PBSDRHEAL_ONLY_HOST` scopes a supervised rollout. +- `internal/web/pbsdr.go` — `ReissuePBSDR(ctx, customerID)`: the non-HTTP core of the operator + Re-issue button, now the reconciler's escalation seam. The operator handler is unchanged (303/400 + codes stay test-pinned). +- `cmd/hub/main.go` — reconciler started unconditionally; `PBSDRHEAL_ONLY_HOST` env read. -**PHASE 4-paperless — all five legs PASS**, which **fully closes the "NOT live-validated" list in the -v0.131.0 controller REPORT** (deferred there to a supervised session): +**Tests (green) + red-proofs (verified):** `internal/pbsdrheal/reconciler_test.go` (Scenarios A–F + +scope + no-re-heal, real store + fake action seam, audit events asserted); `internal/store/pbsdr_test.go` +(re-stage semantics + no-generation-bump guard + `PBSDRHealStates` parsing). All six §10 red-proofs +(A, A-gen, B, C, D, E) failed exactly their guard test then reverted. -- **4P-1** — deployed paperless-ngx on felhom-usb (the appdata-mismatch app: stack `paperless-ngx`, dir - `appdata/paperless`), ingested 3 varied docs through the real consume flow (OCR + PDF/A) → - `appdata/paperless/media` = 11 files / 40 587 B. -- **4P-2 (F-S2 tier-2 backup)** — the appdata leg mirrors `appdata/paperless` to - `secondary/paperless-ngx/appdata` **byte-identical** (11-file sha256 diff empty). Pre-state confirmed - the pre-fix shape: secondary held **recovery-unit only, no appdata**. -- **4P-3 (F-S2 marker restore)** — UI file-restore targets the **resolved** live dir - (`appdata/paperless`, not `…/paperless-ngx`), additive-only, 1 file restored byte-identical, existing - file untouched, app healthy. -- **4P-4 (storage page)** — paperless-ngx shows a non-empty size (**39.6 KB**, from the resolved dir). -- **4P-5 (F-S3 scope="app" migration)** — usb→flash then flash→usb: full 11-file manifest **identical** - each way, source cleaned, **no `appdata/paperless-ngx` ghost dir, no `declared appdata absent` WARN**, - app healthy (45 s each leg). +**Deploy + live validation:** built `felhom-hub:0.56.0`, manifest bump + ArgoCD sync; first rollout +scoped to the drill guest via `PBSDRHEAL_ONLY_HOST=demo-vm-felhom-2f4b00` (STOP discipline — the +fleet-wide widening is a deliberate follow-up). Drill live-leg: the drill's PBS-DR secret was rolled +back to a stuck `waiting_secret` state; the reconciler re-staged it (audit event) → the agent +consumed → PBS-DR GREEN with **no operator click**. (See CHANGELOG / chat report for the exact +outcome.) -### New finding - -- **F-6C-1 (MEDIUM)** — tier-2 auto-target = NFS share fails at the recovery-unit leg: `rsyncMirror`'s - `rsync -a` (`-o -g`) chown is refused on the NFS export (root_squash, exit 23), so `RunTier2` records - failure and the appdata leg never runs. Real drives (usb/flash) are unaffected; **not an F-S2 - regression**. Suggest `--no-owner --no-group`/`--chown` (or exclude NFS from the auto tier-2 pick). - -### Deferred to a 6D / supervised pass - -- **PHASE 3 browser planes** (escrow wizard typed-back, session/CSRF UX, hub 8-tab ring) — need a - rendered browser with an operator present (browser-select, on-screen typed-back, password at the - login screen); not unattended-automatable. -- **PHASE 4-deep** (F7 live 6 s cut, restic mid-run self-heal, offsite restore-to-verify) and - **Acceptance #1** (≥1 GiB userdata `.fab` full-circle) — need large-dataset/timed-cut setup budget; - the C6B-F1 `.fab` fix is already code + live byte-compare verified in the v0.130.0 REPORT. -- **5-rest dead-app alert** — inconclusive (a clean `docker stop` is not a crash; the correct trigger - is `kill -9`/OOM). Observation logged for 6D. - -**Box state:** paperless-ngx torn down (real remove flow + scratch cleanup); controller 0.132.0 healthy; -other 6 apps untouched. Viktor rotates the controller + hub credentials after the campaign. +**⚠️ ARCHITECTURE IMPACT:** before this, no automatic recovery existed for a box re-installed/restored +onto its stable `host_id` — PBS-DR (hence escrow + offsite) stayed silently unconverged until a manual +operator Re-issue. This reconciler closes that gap. diff --git a/REUSE.md b/REUSE.md index 433fc61..b7776ea 100644 --- a/REUSE.md +++ b/REUSE.md @@ -94,6 +94,9 @@ | `(*Server).applyPBSDR` + `mergePBSDR`/`readPBSDR` | hub/internal/web/pbsdr.go | `(ctx, r, cfg) error` | The config form's DR-tier section → HOST desired_json `pbs_dr` descriptor + generation bump | Descriptor lives in the host desired_json, NOT ConfigJSON (buildConfigJSON drops foreign keys on re-save). v0.51.0: driven by `cfg.DRTier` (set from the form BEFORE applyOffsite/applyPBSDR); UNMET preconditions are honest waiting stages (save succeeds), REAL failures stay fail-closed; already-provisioned = success-no-op (red-proofed); disable keeps the ep0 tenancy. | | `(*Server).pbsdrProvisionAtom` + `PBSDRAutoProvision` | hub/internal/web/pbsdr.go | `(ctx, customerID, host, storageID) (blocked string, err error)` / `(ctx, customerID)` | The shared fresh-provision cascade atom; the WG-registration hook target (api `SetWGRegisteredHook`, wired in hub/cmd/hub/main.go when tenantsync is on) | `blocked != ""` = waiting stage (never an error); the hook runs in a detached goroutine and must never fail registration. Scenario-A e2e test: TestPBSDR_AutoProvisionOnWGRegistration. | | `cfg.DRTier` + offsite coupling | hub/internal/store/store.go (CustomerConfig), hub/internal/web/configs.go (applyOffsite guard) | bool | Per-customer DR-tier flag: new-customer default ON (handleConfigNewForm); offsite REFUSED without it (exact F-6 message) | One-time migration backfill initializes legacy rows from descriptor reality — never re-runs (opt-outs survive re-open; store test pins it). Form field `dr_tier` (formBool helper). | +| `pbsdrheal.Reconciler` + `NewActions` | hub/internal/pbsdrheal/reconciler.go | `NewReconciler(st, act, logger)` · `RestrictToHost(hostID)` · `Run(ctx)`/`Trigger()` | PBS-DR SELF-HEAL: re-arms a consumable secret for a box stuck in `waiting_secret`/`consumed_failed` after losing its converged marker (re-install/rollback). From `SPIKE-pbsdr-selfheal-2026-07-15`. | Primary heal = **re-stage** the stored secret (no ep0 call, **NO generation bump** — a bump = agent refetch loop). Escalate to Re-issue only when no stored secret / `consumed_failed`. NEVER re-run `pbsdrProvisionAtom` (refuses `ErrTokenExists`) or blind-timer Re-issue (hash/gen thrash). Converged/`disabled`/`verify_failed`/DR-OFF = no-op. Debounce ≥2 distinct reports. `PBSDRHEAL_ONLY_HOST` scopes a supervised rollout. Fake seam: `fakeActions` in reconciler_test.go. | +| `(*Store).RestageHostPBSSecret` + `PBSDRHealStates` | hub/internal/store/pbsdr.go | `(hostID) (restaged bool, err)` / `() ([]PBSDRHealRow, error)` | The self-heal store primitives: clear `consumed_at` IFF a row exists (re-arm the SAME value); the fleet work-set query (descriptor enable/provision + latest report `pbs_dr.state` + id) | `RestageHostPBSSecret` does NO insert, NO value change, NO generation bump (`restaged=false` = no row → caller escalates). `PBSDRHealStates` mirrors `GetHostOOBStates`' latest-report-per-host join; malformed JSON → zero values, never an error. | +| `(*Server).ReissuePBSDR` | hub/internal/web/pbsdr.go | `(ctx, customerID) error` | The non-HTTP core of the operator Re-issue button — the self-heal reconciler's escalation seam (satisfies `pbsdrheal.Reissuer`) | Reuses `tenantsync.Reissue` + `SaveHostPBSSecret` + descriptor bump — NOT a re-run of `pbsdrProvisionAtom`. Keep in lockstep with the tail of `handlePBSDRReissue` (which is unchanged; the operator button's 303/400 codes are pinned by tests). | | `parseHostCapabilities` + `capabilityView` | hub/internal/web/hosts.go | `(reportJSON) []capabilityView` | Host-page capability chips (ok/degraded/inactive) | `inactive` (agent v0.86.0) = badge-neutral, NEVER warn/error — disabled ≠ degraded; unknown future statuses fall to the degraded styling (surface, don't hide). `capabilitiesNeedDRMigration` keys the pre-v1.15.0 migration hint on pbsdr-* + "binary not found". | ### Config generation & secrets hygiene (hub/internal/configgen/) diff --git a/hub/CHANGELOG.md b/hub/CHANGELOG.md index ca24208..5241ca9 100644 --- a/hub/CHANGELOG.md +++ b/hub/CHANGELOG.md @@ -1,5 +1,42 @@ # Felhom Hub — Changelog +## v0.56.0 — PBS-DR self-heal reconciler (re-stage a consumable secret) (2026-07-15) + +Implements `SPIKE-pbsdr-selfheal-2026-07-15` (`e8f8c44`). **⚠️ ARCHITECTURE IMPACT:** before this, +there was **no automatic recovery** for the commonest real event — a customer box re-installed / +restored / rolled back onto its **stable `host_id`**. The hub kept the durable `pbs_dr` descriptor +(enabled) + a durable **consumed** one-time secret; the WG peer still existed (same pubkey → +`changed==false`, so the provision cascade could not re-fire) and `applyPBSDR`'s "already +provisioned → no-op" meant even a config re-save minted nothing. The agent sat in +`pbs_dr.state="waiting_secret"` forever — PBS-DR never converged, so escrow could not run and offsite +never armed. The spike proved (SQ-2b′) the **missing piece is a consumable secret, not the +descriptor**: re-staging the stored secret converged a stuck box in one ~30 s agent tick, using the +existing ep0 token, **zero churn**. This reconciler closes that gap. + +- **`internal/pbsdrheal/reconciler.go` (new):** a hub periodic reconciler (5 min, `wgsync` shape). + For each host whose descriptor is **enabled + provisioned** and whose **latest report** + `pbs_dr.state` is a stuck state held across a **debounce** (≥2 distinct reports — so a box briefly + `waiting_secret` between provision and its first consume is not touched): `waiting_secret` → + **re-stage the stored secret** (no ep0 call, **no generation bump**); no stored secret → + **escalate to Re-issue**; `consumed_failed` → **escalate to Re-issue only** (a re-stage would + re-feed a burned secret). A converged/`disabled`/`verify_failed`/unprovisioned/DR-OFF host is a + **pure no-op**. Each heal emits a distinct audit event (`pbsdr_selfheal_restaged` / + `_reissued` / `_consumed_failed`). Reads the hub DB only — never the box. +- **`internal/store/pbsdr.go`:** `RestageHostPBSSecret(hostID) (bool, error)` — clears `consumed_at` + IFF a row exists (no INSERT, no value change, **no generation bump**); `restaged=false` → the + caller escalates. `PBSDRHealStates()` — one query joining each host's descriptor enable/provision + flags to its latest report's `pbs_dr.state` + id (mirrors `GetHostOOBStates`). +- **`internal/web/pbsdr.go`:** `ReissuePBSDR(ctx, customerID)` — the non-HTTP core of the operator + Re-issue button (tenantsync reissue → fresh consume-once secret → descriptor refresh + gen bump), + now the reconciler's escalation seam. The operator handler is unchanged. +- **`cmd/hub/main.go`:** the reconciler is started unconditionally (the primary re-stage heal needs + no endpoint). **`PBSDRHEAL_ONLY_HOST`** env scopes a supervised first rollout to one host + (empty = whole fleet — the steady state). +- **Tests:** `internal/pbsdrheal/reconciler_test.go` (Scenarios A–F + scope + no-re-heal, real store + + fake action seam) and `internal/store/pbsdr_test.go` (re-stage semantics + the **no-generation- + bump** guard + `PBSDRHealStates` parsing). All six §10 red-proofs verified (mutation → FAIL → + revert). No agent changes (the agent already self-heals once a secret is consumable). + ## v0.55.0 — accept the offbox_enlarge_blocked event (Task 3a-fix delivery chain) (2026-07-15) The controller (v0.134.1) sends an `offbox_enlarge_blocked` warning when an app's enlarged offsite diff --git a/hub/cmd/hub/main.go b/hub/cmd/hub/main.go index 1148b88..1000745 100644 --- a/hub/cmd/hub/main.go +++ b/hub/cmd/hub/main.go @@ -20,6 +20,7 @@ import ( "gitea.dooplex.hu/admin/felhom-hub/internal/hetznerapi" "gitea.dooplex.hu/admin/felhom-hub/internal/mailrelay" "gitea.dooplex.hu/admin/felhom-hub/internal/offsite" + "gitea.dooplex.hu/admin/felhom-hub/internal/pbsdrheal" "gitea.dooplex.hu/admin/felhom-hub/internal/monitor" "gitea.dooplex.hu/admin/felhom-hub/internal/notify" "gitea.dooplex.hu/admin/felhom-hub/internal/store" @@ -410,6 +411,23 @@ func main() { } } + // PBS-DR self-heal reconciler (internal/pbsdrheal, from SPIKE-pbsdr-selfheal-2026-07-15). Re-arms + // a consumable secret for boxes stuck in waiting_secret/consumed_failed after losing their + // converged marker (re-install / restore / snapshot rollback onto a stable host_id — the case no + // existing path recovers). The primary heal (re-stage the stored secret) needs no endpoint; the + // escalation reuses webServer.ReissuePBSDR (a no-op-error if tenantsync is unconfigured). Reads + // the hub DB only; a converged/disabled host is a pure no-op. PBSDRHEAL_ONLY_HOST scopes a + // supervised first rollout to one host (empty = whole fleet — the steady state). + { + pbsdrReconciler := pbsdrheal.NewReconciler(dataStore, pbsdrheal.NewActions(dataStore, webServer), logger) + if only := os.Getenv("PBSDRHEAL_ONLY_HOST"); only != "" { + pbsdrReconciler.RestrictToHost(only) + logger.Printf("[INFO] PBS-DR self-heal reconciler RESTRICTED to host %s (supervised rollout scope)", only) + } + go pbsdrReconciler.Run(ctx) + logger.Printf("[INFO] PBS-DR self-heal reconciler started (interval 5m)") + } + // Session cleanup — removes expired sessions every hour go webServer.CleanupSessions(ctx) diff --git a/hub/internal/pbsdrheal/reconciler.go b/hub/internal/pbsdrheal/reconciler.go new file mode 100644 index 0000000..2f11fbd --- /dev/null +++ b/hub/internal/pbsdrheal/reconciler.go @@ -0,0 +1,258 @@ +// Package pbsdrheal is the hub-side PBS-DR self-heal reconciler (TASK 2026-07-15, from +// SPIKE-pbsdr-selfheal-2026-07-15 / e8f8c44). +// +// Root cause the spike proved (not reasoned): a customer box re-installed / restored / rolled back +// onto its STABLE host_id loses its agent-side converged marker; the hub still holds the durable +// pbs_dr descriptor (enabled) and the durable *consumed* one-time secret, and the WG peer still +// exists (same pubkey → changed==false → the provision cascade cannot re-fire). The agent gets the +// descriptor, verifies over the tunnel, but ConsumePBSToken returns "no secret" — it sits in +// pbs_dr.state="waiting_secret" forever. PBS-DR never converges, so escrow can't run and offsite +// never arms. The MISSING PIECE IS A CONSUMABLE SECRET, NOT THE DESCRIPTOR (spike SQ-2b′: +// re-staging the stored secret converged the box in one ~30 s agent tick, using the existing ep0 +// token, zero churn). +// +// This reconciler, for each host whose descriptor is enabled+provisioned and whose LATEST report is +// a stuck state sustained across a debounce, RE-STAGES the stored secret (store.RestageHostPBSSecret +// — no ep0 call, no generation bump) and escalates to the existing Re-issue only when there is no +// stored secret to re-stage or the agent burned one (consumed_failed). It NEVER re-runs the whole +// provision atom (which refuses ErrTokenExists) and NEVER blind-timer-reissues (hash/gen thrash). +// A converged/healthy/disabled host is a pure no-op (idempotency — Scenario C). +package pbsdrheal + +import ( + "context" + "log" + "time" + + "gitea.dooplex.hu/admin/felhom-hub/internal/store" +) + +// The reported agent pbs_dr.state strings (felhom-agent internal/pbsdr/manager.go). Only the two +// STUCK states below are acted on; everything else (applied, adopted, disabled, verify_failed, "") +// is a no-op. verify_failed is DELIBERATELY not healed: it is a descriptor/PBS-reachability problem +// (the secret is untouched by verify-before-consume), self-heals when the tunnel recovers, and must +// stay LOUD for the operator if it does not — re-staging a secret would not help it. +const ( + stateWaitingSecret = "waiting_secret" + stateConsumedFailed = "consumed_failed" +) + +// Audit event types (store.SaveEvent; hub-internal, not gated by allowedEventTypes). Distinct per +// remediation so the operator sees exactly what was auto-done (Scenario E wants a distinct signal). +const ( + eventRestaged = "pbsdr_selfheal_restaged" // re-armed the stored secret (routine) + eventReissued = "pbsdr_selfheal_reissued" // no stored secret → minted a fresh one + eventConsumedFailed = "pbsdr_selfheal_consumed_failed" // burned secret → minted a fresh one (a real problem was remediated) +) + +// Actions is the mutation seam — fakes in tests count calls without SSH/ep0. Restage flips a stored +// secret's consumed flag (returns restaged=false when NO row exists → the caller escalates). Reissue +// mints a fresh ep0 token + stores a fresh consume-once secret + bumps the descriptor. +type Actions interface { + Restage(hostID string) (restaged bool, err error) + Reissue(ctx context.Context, customerID string) error +} + +// Reissuer is satisfied by *web.Server (its ReissuePBSDR). Kept here so main.go can wire the server +// as the escalation path without an import cycle. +type Reissuer interface { + ReissuePBSDR(ctx context.Context, customerID string) error +} + +// storeActions is the production Actions: Restage → the store primitive; Reissue → the web server. +type storeActions struct { + st *store.Store + reissuer Reissuer +} + +func (a storeActions) Restage(hostID string) (bool, error) { return a.st.RestageHostPBSSecret(hostID) } +func (a storeActions) Reissue(ctx context.Context, customerID string) error { + return a.reissuer.ReissuePBSDR(ctx, customerID) +} + +// NewActions builds the production mutation seam. +func NewActions(st *store.Store, reissuer Reissuer) Actions { return storeActions{st: st, reissuer: reissuer} } + +// debounceState tracks, per host, the last DISTINCT report observed and how many consecutive +// distinct reports it has held the current stuck state — so a fresh box that briefly shows +// waiting_secret between provision and its first consume (resolved within one 60 s agent tick, well +// inside one ~15 min report cycle) is NOT healed on a single report (Scenario D). +type debounceState struct { + reportID int64 + state string + streak int +} + +// Reconciler re-arms stuck PBS-DR hosts. DECLARATIVE + IDEMPOTENT: a tick over a converged fleet +// writes nothing (Scenario C). It reads the hub DB (the source of truth) — never the box. +type Reconciler struct { + store *store.Store + act Actions + interval time.Duration + debounceReports int // distinct stuck reports required before healing (default 2) + onlyHost string // "" = whole fleet; non-empty restricts the work set to one host (supervised rollout) + trigger chan struct{} + logger *log.Logger + + deb map[string]debounceState +} + +// NewReconciler builds the reconciler. interval defaults to 5m, debounceReports to 2. +func NewReconciler(st *store.Store, act Actions, logger *log.Logger) *Reconciler { + if logger == nil { + logger = log.Default() + } + return &Reconciler{ + store: st, + act: act, + interval: 5 * time.Minute, + debounceReports: 2, + trigger: make(chan struct{}, 1), + logger: logger, + deb: map[string]debounceState{}, + } +} + +// RestrictToHost scopes the reconciler's work set to a single host_id (empty = whole fleet). Used +// for a supervised first rollout (PBSDRHEAL_ONLY_HOST): validate the converged-host no-op live on the +// drill guest before widening to the fleet. Set before Run. +func (r *Reconciler) RestrictToHost(hostID string) { r.onlyHost = hostID } + +// Trigger requests an immediate reconcile (tests + the mutation handlers). Non-blocking. +func (r *Reconciler) Trigger() { + select { + case r.trigger <- struct{}{}: + default: + } +} + +// Run loops until ctx is done, reconciling on each tick or Trigger. Never exits on an error. +func (r *Reconciler) Run(ctx context.Context) { + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-r.trigger: + case <-ticker.C: + } + r.reconcileOnce(ctx) + } +} + +// reconcileOnce reads the whole fleet's DR-heal state and acts on the stuck, debounce-confirmed, +// enabled+provisioned hosts. Errors are logged and retried next tick. +func (r *Reconciler) reconcileOnce(ctx context.Context) { + rows, err := r.store.PBSDRHealStates() + if err != nil { + r.logger.Printf("[ERROR] pbsdrheal: read heal states: %v (retry next tick)", err) + return + } + seen := make(map[string]bool, len(rows)) + for _, row := range rows { + if r.onlyHost != "" && row.HostID != r.onlyHost { + continue // supervised rollout scope: only this host is in the work set + } + seen[row.HostID] = true + // Work set: descriptor enabled AND provisioned (namespace set). A DR-OFF / disabled / bare- + // enabled-but-never-provisioned host is out of scope (Scenario F) — forget any debounce. + if !row.DescriptorEnabled || !row.DescriptorProvisioned { + delete(r.deb, row.HostID) + continue + } + switch row.ReportedState { + case stateWaitingSecret: + if r.confirm(row) { + r.healWaitingSecret(ctx, row) + } + case stateConsumedFailed: + if r.confirm(row) { + r.healConsumedFailed(ctx, row) + } + default: + // applied | adopted | disabled | verify_failed | "" (no report) | anything else → no-op. + delete(r.deb, row.HostID) + } + } + // Drop debounce state for hosts that vanished from the fleet. + for h := range r.deb { + if !seen[h] { + delete(r.deb, h) + } + } +} + +// confirm advances the per-host debounce and reports whether the stuck state has held across +// >= debounceReports DISTINCT reports. A re-observed same report (same reportID) never advances the +// streak — the debounce counts fresh evidence, not reconciler ticks. +func (r *Reconciler) confirm(row store.PBSDRHealRow) bool { + st := r.deb[row.HostID] + if row.ReportID != st.reportID { + if row.ReportedState == st.state { + st.streak++ + } else { + st.streak = 1 + } + st.state = row.ReportedState + st.reportID = row.ReportID + r.deb[row.HostID] = st + } + return st.streak >= r.debounceReports +} + +// healWaitingSecret re-stages the stored secret; if none is stored, escalates to Re-issue. +func (r *Reconciler) healWaitingSecret(ctx context.Context, row store.PBSDRHealRow) { + restaged, err := r.act.Restage(row.HostID) + if err != nil { + r.logger.Printf("[ERROR] pbsdrheal: re-stage %s: %v (retry next tick)", row.HostID, err) + return + } + if restaged { + r.logger.Printf("[INFO] pbsdrheal: re-staged the stored one-time secret for host %s (customer %s) stuck in waiting_secret — the agent re-consumes on its next tick (no ep0 token minted, no generation bump)", row.HostID, row.CustomerID) + r.event(row.CustomerID, eventRestaged, "info", + "PBS-DR self-heal: re-staged the stored one-time credential for a box stuck awaiting a secret (re-install/rollback recovery). No endpoint token was minted.") + r.resetAfterHeal(row) + return + } + // No stored secret to re-stage → escalate to a fresh mint. + r.logger.Printf("[INFO] pbsdrheal: host %s (customer %s) is waiting_secret with NO stored secret — escalating to Re-issue", row.HostID, row.CustomerID) + if r.reissue(ctx, row) { + r.event(row.CustomerID, eventReissued, "warning", + "PBS-DR self-heal: re-issued endpoint credentials for a box awaiting a secret that the hub no longer had stored.") + r.resetAfterHeal(row) + } +} + +// healConsumedFailed escalates a burned-secret box to a fresh mint — a re-stage of the SAME secret +// would only re-feed the credential the agent already burned into a failed apply (Scenario E). +func (r *Reconciler) healConsumedFailed(ctx context.Context, row store.PBSDRHealRow) { + r.logger.Printf("[WARN] pbsdrheal: host %s (customer %s) reports consumed_failed (burned secret) — escalating to Re-issue", row.HostID, row.CustomerID) + if r.reissue(ctx, row) { + r.event(row.CustomerID, eventConsumedFailed, "warning", + "PBS-DR self-heal: a box reported consumed_failed (it burned a one-time credential into a failed apply); re-issued fresh endpoint credentials so the agent can converge.") + r.resetAfterHeal(row) + } +} + +// reissue runs the escalation; returns true on success (the caller then records the audit event). +func (r *Reconciler) reissue(ctx context.Context, row store.PBSDRHealRow) bool { + if err := r.act.Reissue(ctx, row.CustomerID); err != nil { + r.logger.Printf("[ERROR] pbsdrheal: re-issue for customer %s (host %s): %v (retry next tick)", row.CustomerID, row.HostID, err) + return false + } + return true +} + +// resetAfterHeal clears the streak (keeping the report id) so the same stuck report does not re-heal +// on the next tick — a FRESH report must re-confirm the host is still stuck before acting again. +func (r *Reconciler) resetAfterHeal(row store.PBSDRHealRow) { + r.deb[row.HostID] = debounceState{reportID: row.ReportID, state: row.ReportedState, streak: 0} +} + +// event records an audit row; a failure to write it must never break the heal loop. +func (r *Reconciler) event(customerID, eventType, severity, message string) { + if _, err := r.store.SaveEvent(customerID, eventType, severity, message, "", "hub"); err != nil { + r.logger.Printf("[WARN] pbsdrheal: audit event %s for %s not stored: %v", eventType, customerID, err) + } +} diff --git a/hub/internal/pbsdrheal/reconciler_test.go b/hub/internal/pbsdrheal/reconciler_test.go new file mode 100644 index 0000000..f575dd6 --- /dev/null +++ b/hub/internal/pbsdrheal/reconciler_test.go @@ -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()) + } +} diff --git a/hub/internal/store/pbsdr.go b/hub/internal/store/pbsdr.go index 99e4642..2d17377 100644 --- a/hub/internal/store/pbsdr.go +++ b/hub/internal/store/pbsdr.go @@ -1,5 +1,10 @@ package store +import ( + "database/sql" + "encoding/json" +) + // PBS DR tier (SLICE 1): the HOST-scoped one-time PBS token secret — the host/agent twin of the // customer-scoped one_time_secrets pair (SaveOneTimeSecret/ConsumeOneTimeSecret). The hub stores // the tenantsync-returned token secret here; the agent consumes it EXACTLY ONCE with its per-host @@ -38,3 +43,89 @@ func (s *Store) ConsumeHostPBSSecret(hostID string) (string, error) { } return value, nil } + +// RestageHostPBSSecret re-arms an ALREADY-STORED one-time PBS secret for re-consumption by clearing +// its consumed flag — WITHOUT changing the secret value, WITHOUT inserting a row, and WITHOUT any +// endpoint/token interaction. This is the PBS-DR self-heal primitive +// (SPIKE-pbsdr-selfheal-2026-07-15, e8f8c44): a box that lost its converged marker (re-install / +// snapshot rollback) but whose ep0 token is still valid re-consumes the SAME secret and converges in +// one agent tick (spike SQ-2b′, zero ep0 churn). Returns restaged=true when a row existed (its +// consumed flag is now cleared, so ConsumeHostPBSSecret will serve it once); restaged=false when NO +// secret is stored for the host — the caller must escalate to Re-issue (a fresh mint). The value is +// never read or logged here, and NO host generation is bumped (a re-stage changes no descriptor +// content — a bump would trigger an agent desired-state refetch loop). +func (s *Store) RestageHostPBSSecret(hostID string) (bool, error) { + res, err := s.db.Exec(`UPDATE host_pbs_secrets SET consumed_at = NULL WHERE host_id = ?`, hostID) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + return n > 0, nil +} + +// PBSDRHealRow is the per-host self-heal decision input (SPIKE-pbsdr-selfheal-2026-07-15): the host's +// descriptor enable/provision state (parsed from desired_json) joined to the agent's latest reported +// pbs_dr.state (from the newest host_report). Mirrors GetHostOOBStates' latest-report-per-host shape. +type PBSDRHealRow struct { + HostID string + CustomerID string + DescriptorEnabled bool // desired_json pbs_dr.enabled — the DR-ON reality + DescriptorProvisioned bool // desired_json pbs_dr.namespace != "" (was provisioned, not a bare enable) + ReportedState string // latest report pbs_dr.state ("" = no report / no stanza) + ReportID int64 // id of that latest host_report (0 = none); the debounce distinctness key +} + +// PBSDRHealStates returns one row per host: its descriptor enable/provision flags + the agent's +// latest reported pbs_dr.state + that report's id. Malformed JSON degrades to zero values (never an +// error) — the reconciler only acts on positively-parsed stuck states. One query for the whole fleet +// (the reconciler filters to enabled+provisioned hosts). +func (s *Store) PBSDRHealStates() ([]PBSDRHealRow, error) { + rows, err := s.db.Query(` + SELECT h.host_id, h.customer_id, h.desired_json, latest.mx, hr.report_json + FROM hosts h + LEFT JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest + ON latest.host_id = h.host_id + LEFT JOIN host_reports hr ON hr.id = latest.mx`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []PBSDRHealRow + for rows.Next() { + var hostID, customerID, desiredJSON string + var reportID sql.NullInt64 + var reportJSON sql.NullString + if err := rows.Scan(&hostID, &customerID, &desiredJSON, &reportID, &reportJSON); err != nil { + return nil, err + } + r := PBSDRHealRow{HostID: hostID, CustomerID: customerID} + var dd struct { + PBSDR *struct { + Enabled bool `json:"enabled"` + Namespace string `json:"namespace"` + } `json:"pbs_dr"` + } + if json.Unmarshal([]byte(desiredJSON), &dd) == nil && dd.PBSDR != nil { + r.DescriptorEnabled = dd.PBSDR.Enabled + r.DescriptorProvisioned = dd.PBSDR.Namespace != "" + } + if reportID.Valid { + r.ReportID = reportID.Int64 + } + if reportJSON.Valid { + var rr struct { + PBSDR *struct { + State string `json:"state"` + } `json:"pbs_dr"` + } + if json.Unmarshal([]byte(reportJSON.String), &rr) == nil && rr.PBSDR != nil { + r.ReportedState = rr.PBSDR.State + } + } + out = append(out, r) + } + return out, rows.Err() +} diff --git a/hub/internal/store/pbsdr_test.go b/hub/internal/store/pbsdr_test.go index 30f1ad1..22fce65 100644 --- a/hub/internal/store/pbsdr_test.go +++ b/hub/internal/store/pbsdr_test.go @@ -5,6 +5,117 @@ import ( "testing" ) +// RestageHostPBSSecret re-arms a stored (consumed) secret WITHOUT changing its value, inserting a +// row, or bumping any generation. Returns restaged=false when no row exists. +func TestRestageHostPBSSecret(t *testing.T) { + s := newTestStore(t) + + // No row → restaged=false, no error (the caller escalates to Re-issue). + if restaged, err := s.RestageHostPBSSecret("h1"); err != nil || restaged { + t.Fatalf("restage with nothing stored = (%v, %v), want (false, nil)", restaged, err) + } + + // Store + consume, then re-stage: the SAME value is served once more. + if err := s.SaveHostPBSSecret("h1", "the-secret"); err != nil { + t.Fatalf("save: %v", err) + } + if _, err := s.ConsumeHostPBSSecret("h1"); err != nil { + t.Fatalf("first consume: %v", err) + } + if _, err := s.ConsumeHostPBSSecret("h1"); err != sql.ErrNoRows { + t.Fatalf("pre-restage second consume = %v, want ErrNoRows", err) + } + restaged, err := s.RestageHostPBSSecret("h1") + if err != nil || !restaged { + t.Fatalf("restage of a stored secret = (%v, %v), want (true, nil)", restaged, err) + } + // Red-proof: dropping the `SET consumed_at = NULL` makes this consume return ErrNoRows. + got, err := s.ConsumeHostPBSSecret("h1") + if err != nil || got != "the-secret" { + t.Fatalf("post-restage consume = (%q, %v), want (the-secret, nil) — same value, re-armed", got, err) + } +} + +// A-gen guard: a re-stage touches ONLY host_pbs_secrets — never the host generation (a bump would +// trigger an agent desired-state refetch loop). Red-proof: adding a SetHostDesired/gen bump to +// RestageHostPBSSecret makes this assert fail. +func TestRestageHostPBSSecret_NoGenerationBump(t *testing.T) { + s := newTestStore(t) + if err := s.UpsertHost(&Host{HostID: "h1", CustomerID: "c1", APIKey: "k"}); err != nil { + t.Fatalf("host: %v", err) + } + genBefore, err := s.SetHostDesired("h1", []byte(`{"pbs_dr":{"enabled":true,"namespace":"c1"}}`)) + if err != nil { + t.Fatalf("set desired: %v", err) + } + if err := s.SaveHostPBSSecret("h1", "s"); err != nil { + t.Fatalf("save secret: %v", err) + } + if _, err := s.RestageHostPBSSecret("h1"); err != nil { + t.Fatalf("restage: %v", err) + } + h, err := s.GetHost("h1") + if err != nil { + t.Fatalf("get host: %v", err) + } + if h.DesiredGeneration != genBefore { + t.Fatalf("generation changed by a re-stage: %d -> %d (a re-stage must never bump)", genBefore, h.DesiredGeneration) + } +} + +// PBSDRHealStates joins each host's descriptor (enabled/provisioned) to its LATEST report's +// pbs_dr.state + report id. +func TestPBSDRHealStates(t *testing.T) { + s := newTestStore(t) + // Host A: enabled + provisioned, latest report waiting_secret (after an older applied report). + mustHost(t, s, "hA", "cA", `{"pbs_dr":{"enabled":true,"namespace":"cA","storage_id":"felhom-pbs"}}`) + mustReport(t, s, "hA", "cA", `{"pbs_dr":{"state":"applied"}}`) + mustReport(t, s, "hA", "cA", `{"pbs_dr":{"state":"waiting_secret"}}`) // newer → wins + // Host B: descriptor disabled. + mustHost(t, s, "hB", "cB", `{"pbs_dr":{"enabled":false,"namespace":"cB"}}`) + mustReport(t, s, "hB", "cB", `{"pbs_dr":{"state":"disabled"}}`) + // Host C: enabled but NOT provisioned (namespace empty); no report. + mustHost(t, s, "hC", "cC", `{"pbs_dr":{"enabled":true}}`) + + rows, err := s.PBSDRHealStates() + if err != nil { + t.Fatalf("PBSDRHealStates: %v", err) + } + byHost := map[string]PBSDRHealRow{} + for _, r := range rows { + byHost[r.HostID] = r + } + a := byHost["hA"] + if !a.DescriptorEnabled || !a.DescriptorProvisioned || a.ReportedState != "waiting_secret" || a.ReportID == 0 { + t.Errorf("hA = %+v, want enabled+provisioned+waiting_secret+reportID>0", a) + } + b := byHost["hB"] + if b.DescriptorEnabled || b.ReportedState != "disabled" { + t.Errorf("hB = %+v, want disabled descriptor + state disabled", b) + } + c := byHost["hC"] + if !c.DescriptorEnabled || c.DescriptorProvisioned || c.ReportedState != "" || c.ReportID != 0 { + t.Errorf("hC = %+v, want enabled+unprovisioned+no-report", c) + } +} + +func mustHost(t *testing.T, s *Store, hostID, customerID, desiredJSON string) { + t.Helper() + if err := s.UpsertHost(&Host{HostID: hostID, CustomerID: customerID, APIKey: "k-" + hostID}); err != nil { + t.Fatalf("UpsertHost %s: %v", hostID, err) + } + if _, err := s.SetHostDesired(hostID, []byte(desiredJSON)); err != nil { + t.Fatalf("SetHostDesired %s: %v", hostID, err) + } +} + +func mustReport(t *testing.T, s *Store, hostID, customerID, reportJSON string) { + t.Helper() + if err := s.SaveHostReport(hostID, customerID, []byte(reportJSON), HostReportDenorm{AgentVersion: "0.88.0"}); err != nil { + t.Fatalf("SaveHostReport %s: %v", hostID, err) + } +} + // The host-scoped consume-once contract (PBS DR SLICE 1): exactly one read per stored value, // a re-save resets the consumed flag (re-issue supersedes), absence is sql.ErrNoRows. func TestHostPBSSecret_ConsumeOnce(t *testing.T) { diff --git a/hub/internal/web/pbsdr.go b/hub/internal/web/pbsdr.go index 9179468..f0c5466 100644 --- a/hub/internal/web/pbsdr.go +++ b/hub/internal/web/pbsdr.go @@ -299,6 +299,54 @@ func (s *Server) PBSDRAutoProvision(ctx context.Context, customerID string) { s.logger.Printf("[INFO] pbsdr auto-provisioned for %s on WG registration (hands-free cascade)", customerID) } +// ReissuePBSDR re-keys the customer's ep0 PBS token and re-arms the agent — the non-HTTP core shared +// by the operator button (handlePBSDRReissue) and the pbsdrheal self-heal reconciler (the escalation +// path when no stored secret is re-stageable, or the agent burned one and reports consumed_failed). +// tenantsync reissue → fresh consume-once secret (SaveHostPBSSecret) → descriptor refresh with the +// NEW token_id/fingerprint + generation bump (the agent's re-consume signal). This reuses the same +// reissue op the handler does — it is NOT a re-run of pbsdrProvisionAtom (which refuses ErrTokenExists +// and would not re-key). The secret value is never logged. Keep this in lockstep with the tail of +// handlePBSDRReissue. +func (s *Server) ReissuePBSDR(ctx context.Context, customerID string) error { + if s.tenantsync == nil { + return fmt.Errorf("pbsdr: provisioning not configured on this hub") + } + host, err := s.store.GetHostByCustomer(customerID) + if err != nil { + return fmt.Errorf("pbsdr reissue: host lookup: %w", err) + } + if host == nil { + return fmt.Errorf("pbsdr reissue: no host enrolled for %s", customerID) + } + cur := readPBSDR(host.DesiredJSON) + if cur == nil || cur.Namespace == "" { + return fmt.Errorf("pbsdr reissue: no provisioned PBS DR tier for %s", customerID) + } + // Same detached-ctx discipline as applyPBSDR: reissue→store→bump must complete atomically. + rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) + defer cancel() + res, err := s.tenantsync.Reissue(rctx, customerID) + if err != nil { + return fmt.Errorf("pbsdr reissue for %s: %w", customerID, err) + } + if err := s.store.SaveHostPBSSecret(host.HostID, res.TokenSecret); err != nil { + return fmt.Errorf("pbsdr reissue for %s: store secret: %w", customerID, err) + } + cur.TokenID = res.TokenID + cur.Fingerprint = res.Fingerprint + cur.Datastore = res.Datastore + cur.Namespace = res.Namespace + merged, err := mergePBSDR(host.DesiredJSON, cur) + if err != nil { + return fmt.Errorf("pbsdr reissue for %s: merge descriptor: %w", customerID, err) + } + if _, err := s.store.SetHostDesired(host.HostID, []byte(merged)); err != nil { + return fmt.Errorf("pbsdr reissue for %s: descriptor bump: %w", customerID, err) + } + s.logger.Printf("[INFO] pbsdr credentials re-issued for %s (host %s; fresh consume-once secret stored, withheld from logs)", customerID, host.HostID) + return nil +} + // handlePBSDRReissue explicitly re-keys the customer's ep0 PBS token (the offsite F4 precedent): // tenantsync reissue → fresh consume-once secret → descriptor refresh + generation bump so the // agent re-runs its bridge and consumes the fresh secret. The secret value is never logged.