From 7f9c010e1af587268b651c2bf34171e06cdeaa5e Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 3 Oct 2026 20:52:42 +1000 Subject: [PATCH] Drop agent auto-revert; skip agent apply while a try is pending --- internal/agent/agent.go | 71 +++++++++++----------------------- internal/agent/agent_test.go | 75 ++---------------------------------- 2 files changed, 26 insertions(+), 120 deletions(-) diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 03d97d5..868c4e0 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -2,21 +2,19 @@ package agent import ( "context" - "errors" "fmt" "log/slog" - "net/url" "time" "git.unkin.net/unkin/tomswall/internal/config" "git.unkin.net/unkin/tomswall/internal/nftables" + "git.unkin.net/unkin/tomswall/internal/tryapply" ) -// Applier applies a translated config to the firewall and returns a func that -// restores the previous ruleset. Abstracted so the run loop is testable without -// touching the kernel. +// Applier applies a translated config to the firewall. Abstracted so the run +// loop is testable without touching the kernel. type Applier interface { - Apply(ctx context.Context, cfg *config.Config) (revert func() error, err error) + Apply(ctx context.Context, cfg *config.Config) error } // Agent runs the pull-apply-report loop for one device. @@ -27,10 +25,6 @@ type Agent struct { Applier Applier // Resolver overrides the DNS resolver (tests); nil derives it per-config. Resolver *Resolver - - // reverted is the last generation rolled back for cutting off the API; it - // is not re-applied until a newer generation is published. - reverted int64 } // Run loops until ctx is cancelled, applying one cycle per Interval (and once @@ -68,20 +62,16 @@ func (a *Agent) RunOnce(ctx context.Context) error { return fmt.Errorf("control plane unreachable and no cached config: %w", err) } // Re-apply last known-good; do not report a generation we didn't fetch. - return a.applyConfig(ctx, cached, nil) + return a.applyConfig(ctx, cached, false) } - if a.reverted != 0 && rc.Generation == a.reverted { - slog.Warn("agent: skipping reverted generation", "generation", rc.Generation) - return nil + if err := a.Cache.Write(raw); err != nil { + slog.Warn("agent: caching config failed", "err", err) } - return a.applyConfig(ctx, rc, raw) + return a.applyConfig(ctx, rc, true) } -// applyConfig applies rc. A freshly fetched config (raw != nil) is verified by -// reporting status over a new connection through the new ruleset; if the API is -// unreachable the previous ruleset is restored. Only verified configs are cached. -func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, raw []byte) error { +func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, report bool) error { resolver := a.Resolver if resolver == nil { resolver = NewResolver(rc.Resolver) @@ -92,28 +82,15 @@ func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, raw []byte) if err != nil { return fmt.Errorf("translate: %w", err) } - revert, err := a.Applier.Apply(ctx, cfg) - if err != nil { + if err := a.Applier.Apply(ctx, cfg); err != nil { return fmt.Errorf("apply: %w", err) } slog.Info("agent: applied config", "generation", rc.Generation, "rules", len(cfg.Rules)) - if raw != nil { - a.Client.HTTP.CloseIdleConnections() + if report { if err := a.Client.ReportStatus(ctx, rc.Generation); err != nil { - var uerr *url.Error - if errors.As(err, &uerr) { - if rerr := revert(); rerr != nil { - return fmt.Errorf("API unreachable after applying generation %d (%v); revert failed: %w", rc.Generation, err, rerr) - } - a.reverted = rc.Generation - return fmt.Errorf("reverted generation %d: API unreachable through new ruleset: %w", rc.Generation, err) - } slog.Warn("agent: reporting status failed", "err", err) } - if err := a.Cache.Write(raw); err != nil { - slog.Warn("agent: caching config failed", "err", err) - } // Report the FIB so the control plane can scope router enforcement. if fib := CollectFIB(ctx); len(fib) > 0 { if err := a.Client.ReportRoutes(ctx, fib); err != nil { @@ -127,26 +104,24 @@ func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, raw []byte) // EngineApplier applies via the real nftables differential engine. type EngineApplier struct{} -// Apply snapshots the live table, then computes and applies the differential -// change set for cfg. The returned func atomically restores the snapshot. -func (EngineApplier) Apply(_ context.Context, cfg *config.Config) (func() error, error) { +// Apply computes and applies the differential change set for cfg. It refuses +// while a 'tomswall try' awaits confirmation. +func (EngineApplier) Apply(_ context.Context, cfg *config.Config) error { + unlock, err := tryapply.Acquire() + if err != nil { + return err + } + defer unlock() engine, err := nftables.NewEngine(cfg) if err != nil { - return nil, fmt.Errorf("initializing nftables: %w", err) + return fmt.Errorf("initializing nftables: %w", err) } changes, err := engine.Plan() if err != nil { - return nil, fmt.Errorf("computing changes: %w", err) + return fmt.Errorf("computing changes: %w", err) } if changes.Empty() { - return func() error { return nil }, nil + return nil } - snap, err := engine.Snapshot() - if err != nil { - return nil, fmt.Errorf("snapshotting ruleset: %w", err) - } - if err := engine.Apply(changes); err != nil { - return nil, err - } - return func() error { return engine.Restore(snap) }, nil + return engine.Apply(changes) } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index cda54f7..3019815 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -126,17 +126,16 @@ func TestTranslateRejectsUnknownAction(t *testing.T) { } } -// fakeApplier records applied configs and reverts. +// fakeApplier records applied configs. type fakeApplier struct { count int32 - reverts int32 lastGen int } -func (f *fakeApplier) Apply(_ context.Context, cfg *config.Config) (func() error, error) { +func (f *fakeApplier) Apply(_ context.Context, cfg *config.Config) error { atomic.AddInt32(&f.count, 1) f.lastGen = len(cfg.Rules) - return func() error { atomic.AddInt32(&f.reverts, 1); return nil }, nil + return nil } const renderedYAML = `generation: 7 @@ -239,71 +238,3 @@ func TestRunOnceNoCacheReturnsError(t *testing.T) { t.Fatal("expected error when unreachable and no cache exists") } } - -// statusServer serves renderedYAML and answers status reports with handler. -func statusServer(t *testing.T, status http.HandlerFunc) *httptest.Server { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method == http.MethodGet { - _, _ = w.Write([]byte(renderedYAML)) - return - } - status(w, r) - })) - t.Cleanup(srv.Close) - return srv -} - -func TestRunOnceRevertsWhenAPIUnreachableAfterApply(t *testing.T) { - // Dropping the connection mimics a pushed rule that severs the API path. - srv := statusServer(t, func(w http.ResponseWriter, r *http.Request) { - conn, _, _ := w.(http.Hijacker).Hijack() - conn.Close() - }) - - applier := &fakeApplier{} - a := &Agent{ - Client: NewClient(srv.URL, "fw-a", "tok"), - Cache: Cache{Path: filepath.Join(t.TempDir(), "cache.yaml")}, - Applier: applier, - } - if err := a.RunOnce(context.Background()); err == nil { - t.Fatal("expected revert error") - } - if applier.reverts != 1 { - t.Errorf("expected 1 revert, got %d", applier.reverts) - } - if cached, _ := a.Cache.Read(); cached != nil { - t.Error("reverted config must not be cached as known-good") - } - - // The same generation is not re-applied on the next cycle. - if err := a.RunOnce(context.Background()); err != nil { - t.Fatalf("second RunOnce: %v", err) - } - if applier.count != 1 { - t.Errorf("reverted generation re-applied: %d applies", applier.count) - } -} - -func TestRunOnceKeepsConfigOnAPIErrorStatus(t *testing.T) { - // An HTTP error still proves the API is reachable — no revert. - srv := statusServer(t, func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusInternalServerError) - }) - - applier := &fakeApplier{} - a := &Agent{ - Client: NewClient(srv.URL, "fw-a", "tok"), - Cache: Cache{Path: filepath.Join(t.TempDir(), "cache.yaml")}, - Applier: applier, - } - if err := a.RunOnce(context.Background()); err != nil { - t.Fatalf("RunOnce: %v", err) - } - if applier.reverts != 0 { - t.Errorf("unexpected revert") - } - if cached, _ := a.Cache.Read(); cached == nil { - t.Error("expected config cached") - } -}