Drop agent auto-revert; skip agent apply while a try is pending

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)
 }

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")
-	}
-}