From 8ecf8929fb88ce711dfc59daecfd251ca5752ff2 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Wed, 10 Jun 2026 19:02:59 +0200 Subject: [PATCH] slice 10A: activate the control envelope (Down channel) + hub-backed desired provider (v0.15.0) The control envelope becomes live: the agent caches the hub's desired-state + generation and re-fetches GET /hosts/{id}/desired-state only when the generation advances. A new internal/desired Syncer maps the wire shape into a reconcile.CachingProvider feeding the engine; benign deltas reconcile, an explicit guest decommission is gated pending_signature (exec is 10B). Adds the DesiredStateResponse/WireDesiredState wire types + Client.FetchDesiredState + the loop EnvelopeObserver seam. Cross-repo golden (envelope + desired-state) byte-identical with the hub. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 16 +++ REPORT.md | 84 +++++++------ cmd/felhom-agent/main.go | 14 ++- internal/desired/syncer.go | 97 +++++++++++++++ internal/desired/syncer_test.go | 117 ++++++++++++++++++ internal/hub/client.go | 41 +++++- internal/hub/desired_client_test.go | 77 ++++++++++++ internal/hub/desired_contract_test.go | 75 +++++++++++ internal/hub/loop.go | 25 +++- internal/hub/loop_test.go | 41 ++++++ internal/hub/mock_test.go | 2 +- internal/hub/report.go | 61 ++++++++- .../hub/testdata/control-envelope.golden.json | 7 ++ .../hub/testdata/desired-state.golden.json | 23 ++++ internal/reconcile/classify.go | 2 + internal/reconcile/engine.go | 28 +++-- internal/reconcile/plan.go | 19 +++ internal/reconcile/slice10_test.go | 110 ++++++++++++++++ internal/reconcile/state.go | 56 +++++++++ 19 files changed, 836 insertions(+), 59 deletions(-) create mode 100644 internal/desired/syncer.go create mode 100644 internal/desired/syncer_test.go create mode 100644 internal/hub/desired_client_test.go create mode 100644 internal/hub/desired_contract_test.go create mode 100644 internal/hub/testdata/control-envelope.golden.json create mode 100644 internal/hub/testdata/desired-state.golden.json create mode 100644 internal/reconcile/slice10_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f0c77e8..1965ff3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,22 @@ All notable changes to **felhom-agent** are recorded here. Update on every code change that gets pushed. +## v0.15.0 — slice 10A: hub desired-state serving — the "Down" channel (2026-06-10) + +The agent half of slice 10A. The control envelope (`hub.ControlEnvelope`) stops being "reserved — ignored" and becomes the live **Down channel**: a cheap change-notification on every heartbeat. The agent caches the hub's desired-state + its generation; only when **`DesiredGeneration` advances** does it fetch the full state (the heartbeat stays light, the heavy state moves on change). The engine then reconciles **benign** deltas and the gate marks an explicit **destructive** delta `pending_signature` (no signer in 10A → never executed; signed execution is 10B). Pairs with hub v0.9.0. + +### Added / changed +- **`internal/reconcile`**: `DesiredGuest.Decommission` — the canonical **destructive desired-state delta** (an EXPLICIT flag, not "absent from the list", so a partial hub list can never mass-destroy). The planner emits `ActionDecommission` → `ClassDecommission` → Destructive → the gate refuses it `pending_signature`. `Reconcile` now counts a `pending_signature` refusal as **`Result.Pending`** (expected, logged INFO) rather than a failure; any other refusal stays a real failure. `ActionDecommission` has **no executor** (slice 10B) — a defensive guard refuses to run it. New **`CachingProvider`** (thread-safe DesiredState + generation cache; `Desired`/`Update`/`Generation`) — the production `DesiredProvider`, replacing `EmptyProvider` in the daemon engine (empty until the hub serves intent → cold-start is a live no-op, unchanged). +- **`internal/hub`**: the **`ControlEnvelope`** fields are now active (DesiredGeneration drives the fetch, HasSignedOps noted). New wire types **`DesiredStateResponse`** + **`WireDesiredState`** (guests + forward-compat `restore_directive` (10D) / `pbs_namespace` / opaque `storage_manifest`+`backup_policy`) + **`WireDesiredGuest`** (vmid/run/spec/description/decommission). New **`Client.FetchDesiredState`** (GET `/api/v1/hosts/{host_id}/desired-state`, self-scoped to the client's own host). New **`EnvelopeObserver`** loop seam + `SetEnvelopeObserver` — the loop hands the envelope to the sync layer each cycle (hub does not import reconcile/desired). +- **`internal/desired`** (new): the **`Syncer`** — implements `hub.EnvelopeObserver`, fetches desired-state on a generation advance, maps the wire shape to the reconcile domain, and updates the `CachingProvider`. Caches the **fetched** generation (robust to a generation that advanced mid-fetch); a fetch failure keeps the last-known state. `restore_directive` is carried + logged, not acted on (10D). Wired in `cmd/felhom-agent` (daemon): provider → engine, syncer → loop. + +### Tests +- reconcile: a desired-state with one benign + one decommission delta → **benign applied, destructive gated pending (not executed)**; `Plan` emits decommission-only for a decommissioned guest + classifies Destructive; `CachingProvider` update/isolation. +- desired: **fetch-once-on-advance** (no re-fetch on an unchanged generation), fetch-failure-keeps-cache, caches-the-fetched-generation. +- hub client: `FetchDesiredState` hits the self-scoped path with the bearer + decodes (incl. `restore_directive`); a 403 is a typed `HTTPError`. +- loop: the cycle notifies the observer + adopts `PollIntervalSeconds`; a report error skips the observer. +- cross-repo golden: `testdata/desired-state.golden.json` + `control-envelope.golden.json` decode + key-set guard, **byte-identical** with felhom.eu/hub. + ## v0.14.0 — slice 9: host metrics to the controller (`GET /host/metrics` + CPU-temp collector) (2026-06-10) The de-privileged controller (slice 8C) sees only its own cgroup, so it can't read host health itself. Slice 9 **re-serves** the slice-4 collector's host + per-storage view to the customer over the local API, plus the one missing collector — CPU/chassis temperature — so the customer sees their box's health in the controller. Host-wide, token-authed, fresh (a live collect, not the 15-min hub snapshot). Assumption: **one customer per host** (the home-server model); if a host ever serves multiple customers, host-wide CPU/mem would leak cross-customer load → revisit then. diff --git a/REPORT.md b/REPORT.md index fe490a2..5d377e6 100644 --- a/REPORT.md +++ b/REPORT.md @@ -1,53 +1,61 @@ -# REPORT — slice 9 (agent half): host metrics to the controller (v0.14.0) (2026-06-10) +# REPORT — slice 10A (agent half): hub desired-state serving — the "Down" channel (v0.15.0) (2026-06-10) > Overwrite-latest report. Cumulative history: [CHANGELOG.md](CHANGELOG.md). ## What was implemented -The agent half of **slice 9** — re-serving the host's health to the customer's controller, plus the -one new collector (CPU/chassis temperature). The de-privileged controller (slice 8C) sees only its -own cgroup, so it cannot read host metrics itself; the agent already collects host CPU/mem/loadavg/ -uptime + per-storage targets for the hub, and slice 9 exposes that to the customer over the local API. +The agent half of **slice 10A**: activate the control envelope as the live **Down channel** and feed +a hub-backed desired-state into the reconcile engine. Pairs with hub v0.9.0. -### `internal/hub/cputemp.go` — CPU/chassis-temp collector (new) -- `TempReader` seam + `SysfsTempReader`: reads the CPU package temperature from **sysfs** — hwmon - (`coretemp`/`k10temp`/`zenpower`/`cpu_thermal`, preferring the `Package id 0` input) then thermal - zones (preferring `x86_pkg_temp`/`coretemp`/`cpu-thermal`, falling back to `acpitz`). -- **No external binary, no privilege** (sysfs is world-readable) → the root-CLI fence is untouched. -- **Graceful-null**: a missing sensor, an unsupported board, an implausible reading (outside - 5–150 °C), or any read error all degrade to `nil` ("n/a") — never fails the report. Same nullable - contract as the per-disk `SmartSummary.TemperatureC`. +### The control loop (now live) +report (heartbeat) → control envelope → (DesiredGeneration advanced past cache? fetch desired-state) +→ reconcile benign / gate destructive → report. The heartbeat stays light; the heavy desired-state is +fetched **only on a generation advance**. -### `internal/hub` — shared wire field + collector reuse -- `HostMetrics` gains **`CPUTempC *int` (`cpu_temp_c`)** — nullable, on the **shared** struct, so the - **hub report carries it too** (operator freebie). Cross-repo host-report golden updated - **byte-identical** with the hub's copy. -- `Collector` gains a nil-safe `temp TempReader` (defaults to the real `SysfsTempReader`; - `SetTempReader` injects a fake in tests). `Collect()` now sets `cpu_temp_c` on the report. -- **`Collector.HostMetricsNow(ctx)`** — a fresh `NodeStatus` + CPU-temp read returning just the host - block; the source for the local API (current cpu%/temp, not the 15-min hub snapshot). +### `internal/reconcile` +- **`DesiredGuest.Decommission`** — the canonical **destructive** desired-state delta (an EXPLICIT + flag, never "absent from the list", so a partial hub list can't mass-destroy). Planner emits + `ActionDecommission` → `ClassDecommission` → Destructive → the gate refuses `pending_signature`. +- **`Reconcile`** now counts a `pending_signature` gate refusal as **`Result.Pending`** (expected, + INFO-logged), not a failure; any other refusal stays a real failure. `ActionDecommission` has **no + executor** (10B) — a defensive guard refuses to run it. +- **`CachingProvider`** — thread-safe DesiredState + generation cache (`Desired`/`Update`/ + `Generation`); the production provider, replacing `EmptyProvider` in the daemon engine. Empty until + the hub serves intent → cold-start is a live no-op (unchanged behaviour). -### `internal/localapi` — `GET /host/metrics` (new endpoint) -- `host_metrics.go`: host-wide health (cpu%/mem/load/uptime/`cpu_temp_c`) + per-storage capacity - (total/used/fraction, thin-pool, SMART temp+wear). Token-authed via `withGuest` (host-wide data; a - cross-guest `?vmid=` still 403). Best-effort on storage (a view error still returns the host - block). Served only when the `HostMetrics` provider (the shared collector) is wired in - `buildLocalAPIServer` — else 503 "not configured". +### `internal/hub` +- `ControlEnvelope` fields are now active. New wire types **`DesiredStateResponse`** + + **`WireDesiredState`** (guests + forward-compat `restore_directive` (10D) / `pbs_namespace` / opaque + `storage_manifest`+`backup_policy`) + **`WireDesiredGuest`**. New **`Client.FetchDesiredState`** + (GET `/api/v1/hosts/{host_id}/desired-state`, self-scoped to the client's own host). New + **`EnvelopeObserver`** loop seam + `SetEnvelopeObserver` (hub does not import reconcile/desired). + +### `internal/desired` (new) + wiring +- **`Syncer`** — implements `hub.EnvelopeObserver`; fetches on a generation advance, maps wire→domain, + updates the `CachingProvider`. Caches the **fetched** generation (race-robust); a fetch failure keeps + the last-known state. `restore_directive` carried + logged, not acted on (10D). Wired in + `cmd/felhom-agent`: provider → engine, syncer → loop. ## Tests (all green) -- `cputemp_test.go`, `hostmetrics_test.go`, `host_metrics_test.go`: hwmon/thermal-zone selection + - **graceful-null**, `HostMetricsNow` populate/null/hard-error, endpoint populated + `cpu_temp_c:null` - serialization + **401 without a token** + 403 cross-guest + 503 not-configured. -- `go test ./...` green; `go vet ./internal/hub ./internal/localapi` clean. +- reconcile: **benign applied + destructive decommission gated pending (not executed)**; Plan + decommission-only + classifies Destructive; CachingProvider update/isolation. +- desired: **fetch-once-on-advance** / no-refetch-on-unchanged / fetch-failure-keeps-cache / + caches-the-fetched-generation. +- hub: `FetchDesiredState` path+auth+decode (incl. `restore_directive`) + typed 403; loop notifies the + observer + adopts `PollIntervalSeconds`, skips the observer on a report error. +- cross-repo golden (`desired-state` + `control-envelope`) decode + key-set guard, byte-identical with + felhom.eu/hub. `go test ./...` green. ## Versioning / docs -- Version `0.13.0 → 0.14.0`; `CHANGELOG.md` updated. Doc 03 §6 (local-API surface) + §9 (roadmap + - changelog) updated. +- Version `0.14.0 → 0.15.0`; `CHANGELOG.md` updated. Doc 03 §4 (control loop live) + §9 (slice table: + 10A done, 10B/10C/10D pending) updated. -## Assumption (noted, not built) -- **One customer per host** (home-server model): `/host/metrics` is host-wide. A multi-customer host - would leak cross-customer CPU/mem → revisit then. +## Out of scope (per the task) +- Signed-op **execution** (verify + run the gated destructive op) → 10B (10A marks it pending only). +- **Restore-mode / re-enroll** consumption (a new box's first directive) → 10D; 10A serves + already-authenticated hosts only. ## Pending -- **Live validation** on the demo (build + deploy agent v0.14.0; controller monitoring page → real - N100 CPU%/temp + per-storage, cross-checked vs `pvesh`/`free`/`df`). +- **Live validation** on the demo: build+deploy agent v0.15.0 + hub v0.9.0; admin-set a desired-state + with a benign + a decommission delta → generation bumps → agent fetches → reconciles benign + gates + the decommission; change `poll_interval_seconds`; confirm a host can't fetch another host's state. diff --git a/cmd/felhom-agent/main.go b/cmd/felhom-agent/main.go index 4490ae6..bfb1e48 100644 --- a/cmd/felhom-agent/main.go +++ b/cmd/felhom-agent/main.go @@ -27,6 +27,7 @@ import ( "gitea.dooplex.hu/admin/felhom-agent/internal/authz" "gitea.dooplex.hu/admin/felhom-agent/internal/backup" "gitea.dooplex.hu/admin/felhom-agent/internal/config" + "gitea.dooplex.hu/admin/felhom-agent/internal/desired" "gitea.dooplex.hu/admin/felhom-agent/internal/escrow" "gitea.dooplex.hu/admin/felhom-agent/internal/hub" "gitea.dooplex.hu/admin/felhom-agent/internal/localapi" @@ -40,7 +41,7 @@ import ( // version is the agent version. Overridable at build time with // -ldflags "-X main.version="; defaults to the in-repo CHANGELOG version. -var version = "0.14.0" +var version = "0.15.0" func main() { var ( @@ -237,6 +238,15 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int { loop := hub.NewLoop(collector, client, time.Duration(hcfg.PollSeconds)*time.Second, logger) interval := time.Duration(hcfg.PollSeconds) * time.Second + // Desired-state provider (slice 10A): the hub-served target the reconcile engine converges + // toward. Starts EMPTY (generation 0) — reconcile is a live no-op until the hub serves intent, + // exactly like the slice-4 EmptyProvider. The desired.Syncer (wired below) fills it when the + // control envelope's generation advances. Replaces EmptyProvider in the engine. + desiredProvider := reconcile.NewCachingProvider() + // The "Down" channel sync hook: on each heartbeat, fetch desired-state when the generation + // advances. The loop calls it via the EnvelopeObserver seam (hub does not import desired). + loop.SetEnvelopeObserver(desired.NewSyncer(client, desiredProvider, logger)) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() logger.Info("felhom-agent daemon starting", @@ -304,7 +314,7 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int { API: px, Queue: queue, Journal: journal, - Provider: reconcile.EmptyProvider{}, // slice 4: no live desired-state source + Provider: desiredProvider, // slice 10A: hub-served desired-state (empty until the hub serves intent) Gate: gate, HostID: cfg.Hub.HostID, Logger: logger, diff --git a/internal/desired/syncer.go b/internal/desired/syncer.go new file mode 100644 index 0000000..3ce12f3 --- /dev/null +++ b/internal/desired/syncer.go @@ -0,0 +1,97 @@ +// Package desired bridges the hub's "Down" channel (the control-envelope generation signal + +// the desired-state fetch) to the reconcile engine's provider (slice 10A). It implements +// hub.EnvelopeObserver: on each heartbeat it inspects the envelope's DesiredGeneration and, only +// when it has ADVANCED past the cached one, fetches the full desired-state and updates the +// engine's CachingProvider. So the heartbeat stays light; the heavy state moves on change. +// +// It lives in its own package because it imports BOTH hub (the wire client + types) and reconcile +// (the domain DesiredState + CachingProvider). hub does not import it (the loop sees only the +// hub.EnvelopeObserver seam) and reconcile does not import it — so there is no import cycle. +package desired + +import ( + "context" + "log/slog" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/reconcile" +) + +// Fetcher fetches this host's desired-state from the hub. Satisfied by *hub.Client. +type Fetcher interface { + FetchDesiredState(ctx context.Context) (*hub.DesiredStateResponse, error) +} + +// Syncer keeps the engine's CachingProvider in step with the hub's authoritative desired-state. +type Syncer struct { + fetcher Fetcher + provider *reconcile.CachingProvider + logger *slog.Logger +} + +// NewSyncer builds a Syncer over the hub fetcher and the engine's provider. +func NewSyncer(fetcher Fetcher, provider *reconcile.CachingProvider, logger *slog.Logger) *Syncer { + if logger == nil { + logger = slog.Default() + } + return &Syncer{fetcher: fetcher, provider: provider, logger: logger} +} + +// OnEnvelope implements hub.EnvelopeObserver. It fetches + caches the desired-state ONLY when the +// envelope's generation advances past the provider's cached generation — otherwise it is a no-op +// (the cached state is already current). A fetch failure keeps the last-known state (the engine +// keeps reconciling toward it) and is retried on the next advance signal. +func (s *Syncer) OnEnvelope(ctx context.Context, env *hub.ControlEnvelope) { + if env == nil || s.provider == nil { + return + } + have := s.provider.Generation() + if env.DesiredGeneration <= have { + return // cached: the heavy desired-state moves only on a generation advance + } + resp, err := s.fetcher.FetchDesiredState(ctx) + if err != nil { + s.logger.Warn("desired: fetch failed; keeping cached desired-state", + "have_generation", have, "envelope_generation", env.DesiredGeneration, "err", err) + return + } + state := mapWire(resp.DesiredState, s.logger) + // Cache against the FETCHED generation (not the envelope's) — robust to a generation that + // advanced again between the heartbeat and this fetch (we won't re-fetch the same state). + s.provider.Update(resp.Generation, state) + s.logger.Info("desired: updated from hub", + "generation", resp.Generation, "guests", len(state.Guests)) + if env.HasSignedOps { + // 10A only notes the flag; fetching + verifying + executing signed ops is slice 10B. + s.logger.Info("desired: hub reports pending signed ops (fetch/execute is slice 10B)") + } +} + +// mapWire maps the hub wire desired-state to the reconcile domain. 10A acts only on guests; the +// forward-compat fields (restore_directive — 10D — etc.) are carried on the wire and logged, but +// not translated into actions here. +func mapWire(w hub.WireDesiredState, logger *slog.Logger) reconcile.DesiredState { + guests := make(map[int]reconcile.DesiredGuest, len(w.Guests)) + for _, g := range w.Guests { + dg := reconcile.DesiredGuest{ + VMID: g.VMID, + Spec: g.Spec, + Description: g.Description, + Decommission: g.Decommission, + } + switch g.Run { + case "running": + dg.Run = reconcile.RunRunning + case "stopped": + dg.Run = reconcile.RunStopped + default: + dg.Run = reconcile.RunUnspecified // unknown/empty → unmanaged (planner leaves run alone) + } + guests[g.VMID] = dg + } + if w.RestoreDirective != nil { + logger.Info("desired: restore_directive present (consumed in slice 10D — ignored in 10A)", + "mode", w.RestoreDirective.Mode) + } + return reconcile.DesiredState{Guests: guests} +} diff --git a/internal/desired/syncer_test.go b/internal/desired/syncer_test.go new file mode 100644 index 0000000..3364512 --- /dev/null +++ b/internal/desired/syncer_test.go @@ -0,0 +1,117 @@ +package desired + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/reconcile" +) + +func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +// fakeFetcher counts FetchDesiredState calls and returns a fixed response (or error). +type fakeFetcher struct { + resp *hub.DesiredStateResponse + err error + calls int +} + +func (f *fakeFetcher) FetchDesiredState(context.Context) (*hub.DesiredStateResponse, error) { + f.calls++ + return f.resp, f.err +} + +func env(gen int64, signed bool) *hub.ControlEnvelope { + return &hub.ControlEnvelope{DesiredGeneration: gen, HasSignedOps: signed} +} + +// The headline caching behaviour: desired-state is fetched ONCE when the generation advances, and +// NOT re-fetched while the generation is unchanged (the heartbeat stays light). +func TestSyncer_FetchesOnceOnGenerationAdvance(t *testing.T) { + run := "running" + f := &fakeFetcher{resp: &hub.DesiredStateResponse{ + Generation: 1, + DesiredState: hub.WireDesiredState{Guests: []hub.WireDesiredGuest{ + {VMID: 100, Run: run}, + {VMID: 200, Decommission: true}, + }}, + }} + p := reconcile.NewCachingProvider() + s := NewSyncer(f, p, quiet()) + ctx := context.Background() + + // Generation 0 (fresh host, no intent) → NO fetch. + s.OnEnvelope(ctx, env(0, false)) + if f.calls != 0 { + t.Fatalf("fetched %d times at generation 0, want 0", f.calls) + } + + // Generation advances to 1 → fetch exactly once, cache updated. + s.OnEnvelope(ctx, env(1, false)) + if f.calls != 1 { + t.Fatalf("fetched %d times on advance, want 1", f.calls) + } + if p.Generation() != 1 { + t.Errorf("cached generation = %d, want 1", p.Generation()) + } + st, _ := p.Desired(ctx) + if st.Guests[100].Run != reconcile.RunRunning { + t.Errorf("guest 100 run = %q, want running", st.Guests[100].Run) + } + if !st.Guests[200].Decommission { + t.Errorf("guest 200 decommission = false, want true") + } + + // Same generation on the next heartbeats → NO re-fetch (cached). + s.OnEnvelope(ctx, env(1, false)) + s.OnEnvelope(ctx, env(1, false)) + if f.calls != 1 { + t.Errorf("re-fetched on an unchanged generation (calls=%d, want 1)", f.calls) + } + + // A further advance → one more fetch. + f.resp = &hub.DesiredStateResponse{Generation: 2, DesiredState: hub.WireDesiredState{}} + s.OnEnvelope(ctx, env(2, false)) + if f.calls != 2 || p.Generation() != 2 { + t.Errorf("second advance: calls=%d gen=%d, want 2/2", f.calls, p.Generation()) + } +} + +// A fetch failure keeps the last-known cache (the engine keeps reconciling toward it) and does not +// advance the cached generation (so the next heartbeat retries). +func TestSyncer_FetchFailureKeepsCache(t *testing.T) { + p := reconcile.NewCachingProvider() + p.Update(1, reconcile.DesiredState{Guests: map[int]reconcile.DesiredGuest{100: {VMID: 100, Run: reconcile.RunRunning}}}) + f := &fakeFetcher{err: errors.New("hub down")} + s := NewSyncer(f, p, quiet()) + + s.OnEnvelope(context.Background(), env(5, false)) // generation jumped, but fetch fails + if p.Generation() != 1 { + t.Errorf("generation advanced to %d despite fetch failure, want kept at 1", p.Generation()) + } + st, _ := p.Desired(context.Background()) + if st.Guests[100].Run != reconcile.RunRunning { + t.Errorf("cache lost on fetch failure: %+v", st.Guests) + } +} + +// The fetched generation (not the envelope's) is what gets cached — robust to a generation that +// advanced again between the heartbeat and the fetch. +func TestSyncer_CachesFetchedGeneration(t *testing.T) { + f := &fakeFetcher{resp: &hub.DesiredStateResponse{Generation: 7, DesiredState: hub.WireDesiredState{}}} + p := reconcile.NewCachingProvider() + s := NewSyncer(f, p, quiet()) + s.OnEnvelope(context.Background(), env(5, false)) // envelope said 5, fetch returned 7 + if p.Generation() != 7 { + t.Errorf("cached generation = %d, want 7 (the fetched generation)", p.Generation()) + } + // A later envelope at generation 6 must NOT trigger a re-fetch (we already have 7). + s.OnEnvelope(context.Background(), env(6, false)) + if f.calls != 1 { + t.Errorf("re-fetched at generation 6 when cache is 7 (calls=%d)", f.calls) + } +} diff --git a/internal/hub/client.go b/internal/hub/client.go index 9708ff0..89abf49 100644 --- a/internal/hub/client.go +++ b/internal/hub/client.go @@ -25,6 +25,7 @@ const reportPath = "/api/v1/host-report" type Client struct { baseURL string apiKey string + hostID string // for the slice-10A desired-state/jobs paths (/hosts/{hostID}/…) hc *http.Client logger *slog.Logger } @@ -51,12 +52,12 @@ func NewClient(cfg config.HubConfig, logger *slog.Logger) (*Client, error) { Timeout: time.Duration(cfg.TimeoutSeconds) * time.Second, Transport: &http.Transport{TLSClientConfig: tlsCfg}, } - return newClient(cfg.URL, cfg.APIKey, hc, logger), nil + return newClient(cfg.URL, cfg.APIKey, cfg.HostID, hc, logger), nil } // newClient is the shared constructor (tests inject a mock-transport *http.Client). -func newClient(baseURL, apiKey string, hc *http.Client, logger *slog.Logger) *Client { - return &Client{baseURL: strings.TrimRight(baseURL, "/"), apiKey: apiKey, hc: hc, logger: logger} +func newClient(baseURL, apiKey, hostID string, hc *http.Client, logger *slog.Logger) *Client { + return &Client{baseURL: strings.TrimRight(baseURL, "/"), apiKey: apiKey, hostID: hostID, hc: hc, logger: logger} } // TransportError is a network/connection failure (no HTTP response). It never @@ -109,6 +110,40 @@ func (c *Client) Report(ctx context.Context, r *HostReport) (*ControlEnvelope, e return &env, nil } +// FetchDesiredState GETs the host's authoritative desired-state (slice 10A — the "Down" channel's +// heavy payload). The agent calls this ONLY when the heartbeat envelope's DesiredGeneration has +// advanced past its cached one (the heartbeat stays light; the state moves on change). It is +// self-scoped server-side: the per-host key only ever reads ITS OWN host (the client uses its +// configured hostID). Errors are typed (transport vs HTTP) and never include the bearer token. +func (c *Client) FetchDesiredState(ctx context.Context) (*DesiredStateResponse, error) { + if c.hostID == "" { + return nil, fmt.Errorf("hub: FetchDesiredState requires a configured host_id") + } + url := c.baseURL + "/api/v1/hosts/" + c.hostID + "/desired-state" + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("hub: building desired-state request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+c.apiKey) + req.Header.Set("Accept", "application/json") + + resp, err := c.hc.Do(req) + if err != nil { + return nil, &TransportError{Err: err} + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, &HTTPError{StatusCode: resp.StatusCode, BodyTail: tail(raw, 256)} + } + var out DesiredStateResponse + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("hub: decoding desired-state: %w", err) + } + return &out, nil +} + func tail(b []byte, max int) string { s := strings.TrimSpace(string(b)) if len(s) > max { diff --git a/internal/hub/desired_client_test.go b/internal/hub/desired_client_test.go new file mode 100644 index 0000000..cff8704 --- /dev/null +++ b/internal/hub/desired_client_test.go @@ -0,0 +1,77 @@ +package hub + +import ( + "context" + "net/http" + "testing" +) + +// FetchDesiredState GETs the SELF-SCOPED path with the bearer token and decodes the response. +func TestFetchDesiredState_PathAuthAndDecode(t *testing.T) { + var gotPath, gotAuth, gotMethod string + c := testClient(func(r *http.Request) (*http.Response, error) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + gotMethod = r.Method + return httpResp(200, `{"generation":4,"desired_state":{"guests":[ + {"vmid":100,"run":"running"}, + {"vmid":200,"decommission":true} + ],"restore_directive":{"mode":"guest_loss","archive":"local:backup/x","vmid":100}}}`), nil + }) + + resp, err := c.FetchDesiredState(context.Background()) + if err != nil { + t.Fatalf("FetchDesiredState: %v", err) + } + // Self-scoped: the client only ever fetches ITS OWN host (the configured host_id). + if gotMethod != http.MethodGet || gotPath != "/api/v1/hosts/demo-host-01/desired-state" { + t.Errorf("request = %s %s, want GET /api/v1/hosts/demo-host-01/desired-state", gotMethod, gotPath) + } + if gotAuth != "Bearer super-secret-bearer-key" { + t.Errorf("auth header = %q, want the per-host bearer", gotAuth) + } + if resp.Generation != 4 { + t.Errorf("generation = %d, want 4", resp.Generation) + } + if len(resp.DesiredState.Guests) != 2 { + t.Fatalf("guests = %d, want 2", len(resp.DesiredState.Guests)) + } + if resp.DesiredState.Guests[0].Run != "running" || !resp.DesiredState.Guests[1].Decommission { + t.Errorf("guests = %+v", resp.DesiredState.Guests) + } + // Forward-compat restore_directive is carried through (consumed in 10D). + if resp.DesiredState.RestoreDirective == nil || resp.DesiredState.RestoreDirective.Mode != "guest_loss" { + t.Errorf("restore_directive = %+v, want carried (mode guest_loss)", resp.DesiredState.RestoreDirective) + } +} + +// A non-2xx response is a typed HTTPError (e.g. a 403 self-scope refusal from the hub). +func TestFetchDesiredState_HTTPError(t *testing.T) { + c := testClient(func(r *http.Request) (*http.Response, error) { + return httpResp(403, `Forbidden: host_id mismatch`), nil + }) + _, err := c.FetchDesiredState(context.Background()) + if err == nil { + t.Fatal("expected an error on 403") + } + var he *HTTPError + if !asHTTPError(err, &he) || he.StatusCode != 403 { + t.Errorf("err = %v, want HTTPError 403", err) + } +} + +func asHTTPError(err error, target **HTTPError) bool { + for err != nil { + if he, ok := err.(*HTTPError); ok { + *target = he + return true + } + type unwrapper interface{ Unwrap() error } + if u, ok := err.(unwrapper); ok { + err = u.Unwrap() + } else { + return false + } + } + return false +} diff --git a/internal/hub/desired_contract_test.go b/internal/hub/desired_contract_test.go new file mode 100644 index 0000000..107412c --- /dev/null +++ b/internal/hub/desired_contract_test.go @@ -0,0 +1,75 @@ +package hub + +import ( + "encoding/json" + "os" + "testing" +) + +// The desired-state wire is a contract DUPLICATED across two repos (no shared types module yet). +// testdata/desired-state.golden.json and testdata/control-envelope.golden.json MUST be kept +// byte-identical with felhom.eu/hub's copies; these tests decode them through the agent structs +// and key-set-compare, catching drift between the struct and the served shape. +func TestDesiredStateGolden_DecodesAndKeySet(t *testing.T) { + raw, err := os.ReadFile("testdata/desired-state.golden.json") + if err != nil { + t.Fatal(err) + } + var resp DesiredStateResponse + if err := json.Unmarshal(raw, &resp); err != nil { + t.Fatalf("golden does not decode into DesiredStateResponse: %v", err) + } + if resp.Generation != 4 { + t.Errorf("generation = %d, want 4", resp.Generation) + } + if len(resp.DesiredState.Guests) != 2 { + t.Fatalf("guests = %d, want 2", len(resp.DesiredState.Guests)) + } + benign := resp.DesiredState.Guests[0] + if benign.VMID != 100 || benign.Run != "running" || benign.Spec == nil || benign.Spec.Cores != 2 || benign.Description == nil { + t.Errorf("benign guest = %+v", benign) + } + destructive := resp.DesiredState.Guests[1] + if destructive.VMID != 200 || !destructive.Decommission { + t.Errorf("destructive guest = %+v, want vmid 200 decommission", destructive) + } + if resp.DesiredState.PBSNamespace != "felhom-cust-acme" { + t.Errorf("pbs_namespace = %q", resp.DesiredState.PBSNamespace) + } + if resp.DesiredState.RestoreDirective == nil || resp.DesiredState.RestoreDirective.Mode != "guest_loss" { + t.Errorf("restore_directive = %+v, want carried (guest_loss)", resp.DesiredState.RestoreDirective) + } + + // Bidirectional key-set drift guard: the marshaled struct's keys must match the golden's + // (top-level, the desired_state object, and a guest element). + var golden map[string]any + json.Unmarshal(raw, &golden) + b, _ := json.Marshal(resp) + var got map[string]any + json.Unmarshal(b, &got) + assertSameKeys(t, "", golden, got) + assertSameKeys(t, "desired_state", golden["desired_state"], got["desired_state"]) + assertSameKeys(t, "desired_state.guests[0]", + firstElem(golden["desired_state"].(map[string]any)["guests"]), + firstElem(got["desired_state"].(map[string]any)["guests"])) + assertSameKeys(t, "desired_state.restore_directive", + golden["desired_state"].(map[string]any)["restore_directive"], + got["desired_state"].(map[string]any)["restore_directive"]) +} + +func TestControlEnvelopeGolden_Decodes(t *testing.T) { + raw, err := os.ReadFile("testdata/control-envelope.golden.json") + if err != nil { + t.Fatal(err) + } + var env ControlEnvelope + if err := json.Unmarshal(raw, &env); err != nil { + t.Fatalf("golden does not decode into ControlEnvelope: %v", err) + } + if env.Status != "ok" || env.PollIntervalSeconds == nil || *env.PollIntervalSeconds != 900 { + t.Errorf("envelope poll/status = %+v", env) + } + if env.DesiredGeneration != 4 || !env.HasSignedOps || env.Blocked { + t.Errorf("envelope flags = gen %d signed %v blocked %v", env.DesiredGeneration, env.HasSignedOps, env.Blocked) + } +} diff --git a/internal/hub/loop.go b/internal/hub/loop.go index 1051e61..6a234bc 100644 --- a/internal/hub/loop.go +++ b/internal/hub/loop.go @@ -20,6 +20,15 @@ type collectorIface interface { Collect(ctx context.Context) (*HostReport, error) } +// EnvelopeObserver is notified of the hub's control envelope on every heartbeat (slice 10A). +// The desired-state sync layer (internal/desired) implements it: when DesiredGeneration advances +// past its cache it fetches the full desired-state and updates the engine's provider. Defined +// here (consumer-side) so hub does NOT import the desired/reconcile packages — same seam pattern +// as the collector's StorageObserver. A nil observer (no desired-state wiring) is a clean no-op. +type EnvelopeObserver interface { + OnEnvelope(ctx context.Context, env *ControlEnvelope) +} + // Loop is the agent's first daemon run loop: collect a host-report, POST it, adopt // the hub's cadence, repeat. It is resilient — a collect or report error is logged // and the loop continues (the data plane is independent of the agent; a hub outage @@ -30,7 +39,8 @@ type Loop struct { client reporter interval time.Duration logger *slog.Logger - trigger <-chan struct{} // optional: an out-of-band report request (storage watchdog) + trigger <-chan struct{} // optional: an out-of-band report request (storage watchdog) + observer EnvelopeObserver // optional: the slice-10A desired-state sync hook } // NewLoop builds the loop. interval is the starting cadence (the hub may override it @@ -48,6 +58,11 @@ func NewLoop(collector collectorIface, client reporter, interval time.Duration, // so this fires at most once per debounce window. func (l *Loop) SetTrigger(ch <-chan struct{}) { l.trigger = ch } +// SetEnvelopeObserver wires the slice-10A desired-state sync hook. It is called once per cycle +// with the hub's control envelope (after the interval is adopted), so the sync layer can fetch +// desired-state when the generation advances. Optional — unset is a clean no-op. +func (l *Loop) SetEnvelopeObserver(o EnvelopeObserver) { l.observer = o } + // Run reports immediately, then on each tick, until ctx is cancelled (then nil). func (l *Loop) Run(ctx context.Context) error { interval := l.interval @@ -97,9 +112,15 @@ func (l *Loop) cycle(ctx context.Context, current time.Duration) time.Duration { } l.logger.Debug("hub: report sent", "guests", len(report.Guests), - // reserved/forward-compat envelope fields — logged only, never acted on (slice 4). "blocked", env.Blocked, "desired_generation", env.DesiredGeneration, "has_signed_ops", env.HasSignedOps) + // Slice 10A: hand the envelope to the desired-state sync hook (fetch desired-state on a + // generation advance). Done off the report's critical path semantics — a sync/fetch failure + // is the observer's concern and never affects the heartbeat cadence below. + if l.observer != nil { + l.observer.OnEnvelope(ctx, env) + } + if env.PollIntervalSeconds == nil { return current } diff --git a/internal/hub/loop_test.go b/internal/hub/loop_test.go index d78ee57..9e2a9c8 100644 --- a/internal/hub/loop_test.go +++ b/internal/hub/loop_test.go @@ -35,6 +35,47 @@ func (r *fakeReporter) Report(ctx context.Context, _ *HostReport) (*ControlEnvel return r.env, nil } +// recordingObserver records the envelopes the loop hands it (slice 10A EnvelopeObserver seam). +type recordingObserver struct{ envs []*ControlEnvelope } + +func (o *recordingObserver) OnEnvelope(_ context.Context, e *ControlEnvelope) { o.envs = append(o.envs, e) } + +// The loop notifies the EnvelopeObserver once per successful cycle (with the envelope) AND still +// adopts PollIntervalSeconds — the two are independent. +func TestLoop_CycleNotifiesObserverAndAdoptsInterval(t *testing.T) { + var cn, rn int32 + env := &ControlEnvelope{DesiredGeneration: 3, HasSignedOps: true, PollIntervalSeconds: intPtr(120)} + loop := NewLoop( + &fakeCollector{report: &HostReport{}, n: &cn}, + &fakeReporter{env: env, n: &rn}, + 900*time.Second, quietLogger()) + obs := &recordingObserver{} + loop.SetEnvelopeObserver(obs) + + got := loop.cycle(context.Background(), 900*time.Second) + if len(obs.envs) != 1 || obs.envs[0].DesiredGeneration != 3 || !obs.envs[0].HasSignedOps { + t.Fatalf("observer envelopes = %+v, want 1 with gen 3 + has_signed_ops", obs.envs) + } + if got != 120*time.Second { + t.Errorf("poll interval = %v, want 120s adopted alongside the observer notify", got) + } +} + +// On a report failure the observer is NOT notified (there is no envelope to act on). +func TestLoop_ReportErrorSkipsObserver(t *testing.T) { + var cn, rn int32 + loop := NewLoop( + &fakeCollector{report: &HostReport{}, n: &cn}, + &fakeReporter{env: &ControlEnvelope{}, errSeq: []error{errors.New("hub 5xx")}, n: &rn}, + 900*time.Second, quietLogger()) + obs := &recordingObserver{} + loop.SetEnvelopeObserver(obs) + loop.cycle(context.Background(), 900*time.Second) + if len(obs.envs) != 0 { + t.Errorf("observer notified on a report error: %+v", obs.envs) + } +} + func TestClampInterval(t *testing.T) { cases := []struct { in int diff --git a/internal/hub/mock_test.go b/internal/hub/mock_test.go index 3147159..627ea6f 100644 --- a/internal/hub/mock_test.go +++ b/internal/hub/mock_test.go @@ -19,7 +19,7 @@ func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { retu // testClient builds a hub Client over a mock transport (no network). func testClient(rt roundTripFunc) *Client { - return newClient("https://hub.example.test", "super-secret-bearer-key", &http.Client{Transport: rt}, quietLogger()) + return newClient("https://hub.example.test", "super-secret-bearer-key", "demo-host-01", &http.Client{Transport: rt}, quietLogger()) } func httpResp(code int, body string) *http.Response { diff --git a/internal/hub/report.go b/internal/hub/report.go index c6106ac..197da85 100644 --- a/internal/hub/report.go +++ b/internal/hub/report.go @@ -1,5 +1,7 @@ package hub +import "encoding/json" + // HostReport is the wire contract shared with the hub's ingest // (felhom.eu TASK-slice3-hub-ingest). Field NAMES must match the hub // field-for-field. Encoding is ordinary encoding/json (no canonicalization — @@ -233,15 +235,62 @@ type PBSSnapshot struct { type AuditEntry struct{} // audit-log tail entry fields TBD -// ControlEnvelope is the hub's 200 response to a host-report. This slice the agent -// adopts ONLY PollIntervalSeconds; the rest are reserved/forward-compat fields it -// logs at most and never acts on (reconcile, slice 4, consumes them). +// ControlEnvelope is the hub's 200 response to a host-report — the "Down" channel (slice 10A). +// It is a cheap change-notification on every heartbeat: the agent adopts PollIntervalSeconds, +// and when DesiredGeneration ADVANCES past its cached one it fetches the full desired-state from +// GET /hosts/{id}/desired-state (the heavy state moves only on change). HasSignedOps flags a +// non-empty signed-jobs queue (the agent fetches/executes them in 10B). Blocked stays reserved. type ControlEnvelope struct { Status string `json:"status"` // PollIntervalSeconds is a pointer so a missing field (keep current interval) is // distinguishable from an explicit 0. PollIntervalSeconds *int `json:"poll_interval_seconds"` - Blocked bool `json:"blocked"` // reserved — ignored (slice 4) - DesiredGeneration int64 `json:"desired_generation"` // reserved — ignored (slice 4) - HasSignedOps bool `json:"has_signed_ops"` // reserved — ignored (slice 4) + Blocked bool `json:"blocked"` // reserved — ignored + DesiredGeneration int64 `json:"desired_generation"` // slice 10A: the cached-vs-current change signal + HasSignedOps bool `json:"has_signed_ops"` // slice 10A: signed-jobs queue non-empty (exec 10B) +} + +// DesiredStateResponse is GET /hosts/{host_id}/desired-state (slice 10A — the "Down" channel's +// heavy payload, fetched only when the envelope's generation advances). Generation is the +// generation this state corresponds to, so the agent caches state+generation atomically. This is +// a cross-repo wire contract (DUPLICATED in felhom.eu/hub until a shared module exists); the +// desired-state golden stays byte-identical across the two repos. +type DesiredStateResponse struct { + Generation int64 `json:"generation"` + DesiredState WireDesiredState `json:"desired_state"` +} + +// WireDesiredState is the hub's authoritative per-host target (slice 10A). The agent reconciles the +// parts it can today (guests: benign deltas reconciled, an explicit decommission gated +// pending_signature); the rest are FORWARD-COMPAT — carried + cached, NOT acted on in 10A. The +// restore_directive is consumed in 10D (host/guest-loss DR); storage_manifest / backup_policy / +// pbs_namespace are placeholders kept opaque so the wire is stable as those land. +type WireDesiredState struct { + Guests []WireDesiredGuest `json:"guests"` + + StorageManifest json.RawMessage `json:"storage_manifest,omitempty"` + BackupPolicy json.RawMessage `json:"backup_policy,omitempty"` + PBSNamespace string `json:"pbs_namespace,omitempty"` + RestoreDirective *WireRestoreDirective `json:"restore_directive,omitempty"` // slice 10D (forward-compat) +} + +// WireDesiredGuest is one guest's target (slice 10A). Every field is optional ("unmanaged"); the +// agent's planner acts only on the fields that are set. Run is running|stopped|""; Spec reuses +// GuestSpec (cores/memory_bytes/disk_bytes); Decommission is the EXPLICIT destructive delta (gated +// pending_signature in 10A — executor is 10B). +type WireDesiredGuest struct { + VMID int `json:"vmid"` + Run string `json:"run,omitempty"` + Spec *GuestSpec `json:"spec,omitempty"` + Description *string `json:"description,omitempty"` + Decommission bool `json:"decommission,omitempty"` +} + +// WireRestoreDirective is the forward-compat restore directive (slice 10D — host/guest-loss DR). +// Defined now so the wire contract is stable; 10A carries it through to the cache but does NOT +// consume it (no restore is initiated from desired-state in 10A). +type WireRestoreDirective struct { + Mode string `json:"mode,omitempty"` // guest_loss | host_loss (10D vocabulary) + Archive string `json:"archive,omitempty"` // source archive/snapshot to restore from + VMID int `json:"vmid,omitempty"` } diff --git a/internal/hub/testdata/control-envelope.golden.json b/internal/hub/testdata/control-envelope.golden.json new file mode 100644 index 0000000..a935393 --- /dev/null +++ b/internal/hub/testdata/control-envelope.golden.json @@ -0,0 +1,7 @@ +{ + "status": "ok", + "poll_interval_seconds": 900, + "blocked": false, + "desired_generation": 4, + "has_signed_ops": true +} diff --git a/internal/hub/testdata/desired-state.golden.json b/internal/hub/testdata/desired-state.golden.json new file mode 100644 index 0000000..1df911c --- /dev/null +++ b/internal/hub/testdata/desired-state.golden.json @@ -0,0 +1,23 @@ +{ + "generation": 4, + "desired_state": { + "guests": [ + { + "vmid": 100, + "run": "running", + "spec": { "cores": 2, "memory_bytes": 2147483648, "disk_bytes": 21474836480 }, + "description": "felhom: acme prod" + }, + { + "vmid": 200, + "decommission": true + } + ], + "pbs_namespace": "felhom-cust-acme", + "restore_directive": { + "mode": "guest_loss", + "archive": "local:backup/vzdump-lxc-200-2026_06_09-11_00_00.tar.zst", + "vmid": 200 + } + } +} diff --git a/internal/reconcile/classify.go b/internal/reconcile/classify.go index f639d30..48335e3 100644 --- a/internal/reconcile/classify.go +++ b/internal/reconcile/classify.go @@ -107,6 +107,8 @@ func classOfAction(k ActionKind) OpClass { return ClassSetConfig case ActionResize: return ClassResize + case ActionDecommission: + return ClassDecommission default: return OpClass(k) } diff --git a/internal/reconcile/engine.go b/internal/reconcile/engine.go index ed95436..9885c45 100644 --- a/internal/reconcile/engine.go +++ b/internal/reconcile/engine.go @@ -86,6 +86,7 @@ type Result struct { Planned int Executed int // succeeded Failed int // errored + Pending int // destructive actions gated pending_signature (slice 10A — expected, not failed) Errors []error // one per failed action } @@ -110,11 +111,12 @@ func (e *Engine) Reconcile(ctx context.Context) (Result, error) { return res, nil } - // Every mutation passes the reversibility gate before the queue (doc 03 §4). - // Reconcile only produces benign actions, so each is allowed unsigned — but the - // gate is genuinely in the path: a destructive class here would be refused - // (pending_signature) and never dispatched. A gate refusal counts as a failed - // action (it should not happen for the benign reconcile set). + // Every mutation passes the reversibility gate before the queue (doc 03 §4). Benign actions + // are allowed unsigned; a DESTRUCTIVE delta (slice 10A: an explicit decommission) is refused + // `pending_signature` when no operator signature is present — that is EXPECTED, not a failure: + // 10A serves destructive intent but never executes it (the signed-op execution is 10B). So a + // pending_signature refusal is counted as Pending and logged at INFO; any OTHER refusal (a + // benign action denied, or a destructive one rejected for a different reason) is a real failure. type dispatched struct { act Action ch <-chan error @@ -124,10 +126,16 @@ func (e *Engine) Reconcile(ctx context.Context) (Result, error) { act := actions[i] dec := e.gate.Authorize(intentForAction(e.hostID, act), nil) if !dec.Allowed { + if dec.Disposition == Destructive && dec.Reason == ReasonPendingSignature { + res.Pending++ + e.logger.Info("reconcile: destructive action gated pending operator signature (slice 10B)", + "vmid", act.VMID, "kind", act.Kind, "reason", dec.Reason) + continue + } res.Failed++ res.Errors = append(res.Errors, fmt.Errorf("reconcile: gate refused %s vmid %d: %s", act.Kind, act.VMID, dec.Reason)) - e.logger.Error("reconcile: gate refused a benign action (unexpected)", + e.logger.Error("reconcile: gate refused an action unexpectedly", "vmid", act.VMID, "kind", act.Kind, "reason", dec.Reason) continue } @@ -174,6 +182,12 @@ func (e *Engine) execute(ctx context.Context, act Action) error { } else { upid, err = e.api.ResizeLXC(ctx, act.VMID, disk, size) } + case ActionDecommission: + // Reaching here means a destructive decommission passed the gate (a verified signature) — + // which only happens once 10B wires the signed-op executor. In 10A there is no signer, so + // the gate refuses it before dispatch and this branch is unreachable. Fail safe loudly + // rather than silently no-op, so a future signed path can't accidentally execute here. + err = fmt.Errorf("reconcile: decommission executor is slice 10B (refusing to execute vmid %d)", act.VMID) default: err = fmt.Errorf("reconcile: unknown action kind %q", act.Kind) } @@ -261,7 +275,7 @@ func (e *Engine) reconcileOnce(ctx context.Context) { } if res.Planned > 0 { e.logger.Info("reconcile: pass complete", - "planned", res.Planned, "executed", res.Executed, "failed", res.Failed) + "planned", res.Planned, "executed", res.Executed, "failed", res.Failed, "pending", res.Pending) } } diff --git a/internal/reconcile/plan.go b/internal/reconcile/plan.go index 2bce053..39953ba 100644 --- a/internal/reconcile/plan.go +++ b/internal/reconcile/plan.go @@ -24,6 +24,11 @@ const ( // emits it only when desired DiskBytes > actual; a shrink is data-losing and is refused // (never silently applied as a grow). Slice 5 Phase B; unfed live until slice 10. ActionResize ActionKind = "resize" + // ActionDecommission tears a guest down — the canonical DESTRUCTIVE delta (slice 10A). The + // planner emits it for an explicit DesiredGuest.Decommission; it classifies ClassDecommission + // → Destructive, so the gate refuses it `pending_signature` (no signer in 10A → never + // executed). Its EXECUTOR is slice 10B; 10A only plans + gates it. + ActionDecommission ActionKind = "decommission" ) // growRoundMiB rounds a positive byte delta UP to whole MiB for the Proxmox `+M` grow @@ -88,6 +93,20 @@ func Plan(desired DesiredState, actual ActualState, norm FieldNormalizers) []Act continue } + // EXPLICIT decommission (slice 10A) — the destructive delta. Emit it as a single + // ActionDecommission and emit NOTHING else for this guest (no point reconciling cores + // on a guest the operator wants torn down). It is classified Destructive downstream, so + // the gate refuses it pending_signature in 10A (executor is 10B). Only emitted when the + // guest actually exists (decommissioning an absent guest is a no-op). + if d.Decommission { + actions = append(actions, Action{ + VMID: vmid, + Kind: ActionDecommission, + Reason: "decommission requested (destructive — requires operator signature)", + }) + continue + } + // Benign spec/description changes → a single SetConfig, only when we could // read the current config (else we'd write blind). if a.SpecKnown { diff --git a/internal/reconcile/slice10_test.go b/internal/reconcile/slice10_test.go new file mode 100644 index 0000000..0d10e2c --- /dev/null +++ b/internal/reconcile/slice10_test.go @@ -0,0 +1,110 @@ +package reconcile + +import ( + "context" + "testing" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +// The headline slice-10A reconcile behaviour: a desired-state carrying ONE benign delta and ONE +// destructive (decommission) delta → the benign op is applied, the destructive one is GATED +// pending_signature (counted Pending, NOT Failed) and is NEVER executed (no signer in 10A). +func TestReconcile_BenignAppliedDestructiveGated(t *testing.T) { + api := &fakeAPI{ + lxc: []proxmox.Guest{ + {VMID: 100, Status: "stopped"}, // benign: desired running → Start + {VMID: 200, Status: "running"}, // destructive: decommission → gated + }, + cfg: map[int]proxmox.GuestConfig{100: {Cores: 2}, 200: {Cores: 2}}, + } + provider := StaticProvider{State: DesiredState{Guests: map[int]DesiredGuest{ + 100: {VMID: 100, Run: RunRunning}, + 200: {VMID: 200, Decommission: true}, + }}} + e, _, _ := newEngine(t, api, provider) + + res, err := e.Reconcile(context.Background()) + if err != nil { + t.Fatalf("Reconcile: %v", err) + } + if res.Planned != 2 { + t.Errorf("planned = %d, want 2 (one benign + one destructive)", res.Planned) + } + // Benign Start(100) applied. + if res.Executed != 1 || len(api.starts) != 1 || api.starts[0] != 100 { + t.Errorf("benign delta not applied: executed=%d starts=%v", res.Executed, api.starts) + } + // Destructive decommission GATED pending (not a failure). + if res.Pending != 1 { + t.Errorf("pending = %d, want 1 (decommission gated pending_signature)", res.Pending) + } + if res.Failed != 0 { + t.Errorf("failed = %d, want 0 (a pending_signature gate is expected, not a failure)", res.Failed) + } + // And it was NEVER executed: no destroy/decommission op reached Proxmox. + if len(api.destroys) != 0 { + t.Errorf("destructive decommission EXECUTED (destroys=%v) — it must be gated, not run", api.destroys) + } +} + +// Plan unit: an explicit Decommission emits exactly one ActionDecommission and suppresses any +// other delta for that guest (no point reconciling cores on a guest being torn down). +func TestPlan_DecommissionEmitsDestructiveActionOnly(t *testing.T) { + desired := DesiredState{Guests: map[int]DesiredGuest{ + // Decommission set AND a spec drift — only the decommission should be emitted. + 7: {VMID: 7, Decommission: true, Spec: &hub.GuestSpec{Cores: 9, MemoryBytes: 9 << 20}, Run: RunStopped}, + }} + actual := ActualState{Guests: map[int]ActualGuest{ + 7: {VMID: 7, Run: RunRunning, SpecKnown: true, Cores: 2}, + }} + actions := Plan(desired, actual, DefaultNormalizers()) + if len(actions) != 1 { + t.Fatalf("actions = %d (%+v), want exactly 1 (decommission only)", len(actions), actions) + } + if actions[0].Kind != ActionDecommission || actions[0].VMID != 7 { + t.Errorf("action = %+v, want decommission of vmid 7", actions[0]) + } + // And it classifies destructive. + if Classify(classOfAction(ActionDecommission), Provenance{}) != Destructive { + t.Error("ActionDecommission must classify Destructive (no provenance)") + } +} + +// Decommission of an ABSENT guest is a no-op (nothing to tear down). +func TestPlan_DecommissionAbsentGuestNoop(t *testing.T) { + desired := DesiredState{Guests: map[int]DesiredGuest{7: {VMID: 7, Decommission: true}}} + actual := ActualState{Guests: map[int]ActualGuest{}} // guest 7 not present + if actions := Plan(desired, actual, DefaultNormalizers()); len(actions) != 0 { + t.Errorf("decommission of absent guest emitted %+v, want none", actions) + } +} + +// CachingProvider: empty until Update, then serves the cached state + generation, and isolates +// the cache from caller mutation. +func TestCachingProvider_UpdateAndIsolation(t *testing.T) { + p := NewCachingProvider() + if p.Generation() != 0 { + t.Fatalf("fresh generation = %d, want 0", p.Generation()) + } + if st, _ := p.Desired(context.Background()); len(st.Guests) != 0 { + t.Fatalf("fresh provider should be empty, got %+v", st.Guests) + } + + p.Update(3, DesiredState{Guests: map[int]DesiredGuest{5: {VMID: 5, Run: RunRunning}}}) + if p.Generation() != 3 { + t.Errorf("generation after update = %d, want 3", p.Generation()) + } + st, _ := p.Desired(context.Background()) + if len(st.Guests) != 1 || st.Guests[5].Run != RunRunning { + t.Fatalf("cached state = %+v", st.Guests) + } + // Mutating the returned copy must NOT affect the cache. + st.Guests[5] = DesiredGuest{VMID: 5, Run: RunStopped} + st.Guests[99] = DesiredGuest{VMID: 99} + st2, _ := p.Desired(context.Background()) + if len(st2.Guests) != 1 || st2.Guests[5].Run != RunRunning { + t.Errorf("cache was mutated by a caller: %+v", st2.Guests) + } +} diff --git a/internal/reconcile/state.go b/internal/reconcile/state.go index d917463..46118d4 100644 --- a/internal/reconcile/state.go +++ b/internal/reconcile/state.go @@ -3,6 +3,7 @@ package reconcile import ( "context" "encoding/json" + "sync" "gitea.dooplex.hu/admin/felhom-agent/internal/hub" "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" @@ -50,6 +51,14 @@ type DesiredGuest struct { // Description, when non-nil, manages the cosmetic `description` field (the first // proven SetConfig round-trip, slice-4 pre-check). Nil = unmanaged. Description *string + // Decommission, when true, is an EXPLICIT destructive intent to tear the guest down + // (slice 10A). It is the canonical destructive desired-state delta: the planner emits an + // ActionDecommission, which classifies ClassDecommission → Destructive → the gate refuses it + // `pending_signature` unless a verified operator signature is present. 10A never has a signer, + // so a decommission is always gated (never executed); the signed execution path is 10B. An + // explicit flag (not "absent from the desired list") is the safe design — a partial/empty hub + // list can never silently mass-destroy guests. + Decommission bool } // DesiredState is the vmid-keyed target for this host. At slice 4 the only live @@ -101,6 +110,53 @@ type StaticProvider struct{ State DesiredState } // Desired returns the static state. func (p StaticProvider) Desired(context.Context) (DesiredState, error) { return p.State, nil } +// CachingProvider is the slice-10A production provider: a thread-safe cache of the hub-served +// DesiredState plus the generation it corresponds to. The hub-sync layer (internal/desired) calls +// Update when the heartbeat envelope's generation advances and a fresh fetch arrives; the engine +// reads the cache via Desired each reconcile tick. Until the first Update it returns an empty +// state (generation 0) — so reconcile is a live no-op exactly like EmptyProvider, with zero +// mutations, which is the correct cold-start behaviour. +type CachingProvider struct { + mu sync.RWMutex + state DesiredState + gen int64 +} + +// NewCachingProvider builds an empty provider (generation 0, no guests). +func NewCachingProvider() *CachingProvider { + return &CachingProvider{state: DesiredState{Guests: map[int]DesiredGuest{}}} +} + +// Desired returns the cached state (a shallow copy of the guest map so a caller can't mutate the +// cache, and a concurrent Update can't race the read). +func (p *CachingProvider) Desired(context.Context) (DesiredState, error) { + p.mu.RLock() + defer p.mu.RUnlock() + out := DesiredState{Guests: make(map[int]DesiredGuest, len(p.state.Guests))} + for k, v := range p.state.Guests { + out.Guests[k] = v + } + return out, nil +} + +// Update replaces the cached state + generation (called by the sync layer on a generation advance). +func (p *CachingProvider) Update(generation int64, state DesiredState) { + p.mu.Lock() + defer p.mu.Unlock() + if state.Guests == nil { + state.Guests = map[int]DesiredGuest{} + } + p.state = state + p.gen = generation +} + +// Generation returns the cached generation (the agent's view of "what I have applied from"). +func (p *CachingProvider) Generation() int64 { + p.mu.RLock() + defer p.mu.RUnlock() + return p.gen +} + // GuestAPI is the narrow Proxmox surface the engine needs: read actual state and // dispatch the benign-on-existing-guest mutations. *proxmox.Client satisfies it; a // fake satisfies it in tests. Every mutating call returns a UPID (or "" for the