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) <noreply@anthropic.com>
This commit is contained in:
@@ -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}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user