Let an agent find the leader instead of being told one address

The second half of leadership. A standby that refuses to place but still holds
an agent's stream is worse than useless: commands travel down the stream the
agent opened, so the leader's placements fail with "no agent connected" for a
node that is plainly connected.

So a standby refuses the stream, before reading the opening message, and says
why. And an agent takes a list rather than an address, trying each in turn — a
standby refusing, a stopped control plane refusing the connection and a crashed
one never answering are the same thing from a node, and the answer to all three
is somewhere else.

`server` and `servers` are both accepted, with `server` first when both are
given: every agent configuration written before there could be a second control
plane names one address, and an operator adding a standby expects the one
already working to stay preferred. With a single address the behaviour is
exactly what it was, including the backoff — which is skipped only while there
is another address that has not just been tried, because the delay exists to
stop hammering one control plane and the next one has not been hammered.

Removed a check that asked whether one field was empty, which is how the first
run of these tests failed: the list was populated and `server` was not, and an
agent configured entirely correctly was refused by its own validation.

The tests run a real agent against a real service over h2c. The first attempt
used a plain HTTP server, and the refusal arrived as an HTTP/2 framing error
that said nothing about why — which is exactly what an operator would have seen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/agent/agent.go

@@ -34,8 +34,18 @@ import (
 
 // Config configures the agent.
 type Config struct {
-	NodeID   string            `koanf:"node_id"`
+	NodeID string `koanf:"node_id"`
+	// Server is the control plane, and Servers is the same thing when there is
+	// more than one. Both are accepted: every configuration written before
+	// there could be a second control plane names one address, and none of
+	// them should have to change.
+	//
+	// The order is tried in order, and a refusal moves to the next. Only the
+	// leader accepts an agent, so a standby is skipped in the time it takes to
+	// be told so — which is what makes failover a reconnect rather than an
+	// edit to every node's configuration.
 	Server   string            `koanf:"server"`
+	Servers  []string          `koanf:"servers"`
 	Hostname string            `koanf:"hostname"`
 	MgmtAddr string            `koanf:"mgmt_addr"`
 	Labels   map[string]string `koanf:"labels"`
@@ -108,9 +118,6 @@ func (c *Config) normalize() error {
 	if c.NodeID == "" {
 		return errors.New("node_id is required")
 	}
-	if c.Server == "" {
-		return errors.New("server is required")
-	}
 	if err := c.TLS.Validate(); err != nil {
 		return err
 	}
@@ -140,26 +147,59 @@ func (c *Config) normalize() error {
 // that says plaintext while TLS is configured. Silently connecting anyway would
 // mean a cluster believing it has mutual authentication when it has none.
 func (c *Config) normalizeServerURL() error {
-	secure := c.TLS.Enabled()
-	switch {
-	case strings.HasPrefix(c.Server, "https://"):
-		if !secure {
-			return errors.New("server is an https:// address but no tls certificates are configured")
-		}
-	case strings.HasPrefix(c.Server, "http://"):
-		if secure {
-			return errors.New("tls is configured but server is an http:// address; use https://")
-		}
-	case strings.Contains(c.Server, "://"):
-		return fmt.Errorf("server %q must be an http:// or https:// address", c.Server)
-	case secure:
-		c.Server = "https://" + c.Server
-	default:
-		c.Server = "http://" + c.Server
+	// One list, however it was written. Server first when both are given,
+	// because an operator adding a standby to an existing file expects the
+	// one already there to stay preferred.
+	all := make([]string, 0, 1+len(c.Servers))
+	if c.Server != "" {
+		all = append(all, c.Server)
 	}
+	all = append(all, c.Servers...)
+
+	seen := map[string]struct{}{}
+	out := make([]string, 0, len(all))
+	for _, s := range all {
+		fixed, err := normalizeOne(s, c.TLS.Enabled())
+		if err != nil {
+			return err
+		}
+		// A duplicate would be tried twice in a row on failover, which reads
+		// as the rotation not working.
+		if _, dup := seen[fixed]; dup {
+			continue
+		}
+		seen[fixed] = struct{}{}
+		out = append(out, fixed)
+	}
+	if len(out) == 0 {
+		return errors.New("server is required")
+	}
+	c.Servers = out
+	c.Server = out[0]
 	return nil
 }
 
+// normalizeOne applies the scheme rules to a single address.
+func normalizeOne(addr string, secure bool) (string, error) {
+	switch {
+	case strings.HasPrefix(addr, "https://"):
+		if !secure {
+			return "", fmt.Errorf("%s is an https:// address but no tls certificates are configured", addr)
+		}
+	case strings.HasPrefix(addr, "http://"):
+		if secure {
+			return "", fmt.Errorf("tls is configured but %s is an http:// address; use https://", addr)
+		}
+	case strings.Contains(addr, "://"):
+		return "", fmt.Errorf("server %q must be an http:// or https:// address", addr)
+	case secure:
+		return "https://" + addr, nil
+	default:
+		return "http://" + addr, nil
+	}
+	return addr, nil
+}
+
 // Agent runs on a GPU node.
 type Agent struct {
 	cfg  Config
@@ -230,15 +270,30 @@ func (a *Agent) Run(ctx context.Context) error {
 	a.recoverOrphans()
 
 	attempt := 0
+	next := 0
 	for ctx.Err() == nil {
-		err := a.session(ctx)
+		server := a.cfg.Servers[next%len(a.cfg.Servers)]
+		err := a.session(ctx, server)
 		if ctx.Err() != nil {
 			break
 		}
+		// On to the next address whatever went wrong. A standby refuses, a
+		// stopped control plane refuses the connection, and a crashed one
+		// never answers — from here they are the same thing, and trying
+		// somewhere else is the answer to all three.
+		next++
 		attempt++
 		delay := a.reconnectDelay(attempt)
+		// No wait when there is somewhere else to try: the delay exists to
+		// stop hammering one control plane, and the next address has not been
+		// hammered. With a single address this is the old behaviour exactly.
+		if len(a.cfg.Servers) > 1 && next%len(a.cfg.Servers) != 0 {
+			delay = 0
+		}
 		a.log.Warn("disconnected from the control plane, retrying",
-			"err", err, "in", delay.Round(time.Millisecond), "attempt", attempt)
+			"err", err, "server", server,
+			"next", a.cfg.Servers[next%len(a.cfg.Servers)],
+			"in", delay.Round(time.Millisecond), "attempt", attempt)
 
 		select {
 		case <-ctx.Done():
@@ -469,8 +524,8 @@ func (a *Agent) scan(ctx context.Context) error {
 }
 
 // session runs one connection until it breaks.
-func (a *Agent) session(ctx context.Context) error {
-	client := llmgwv1connect.NewAgentServiceClient(a.http, a.cfg.Server)
+func (a *Agent) session(ctx context.Context, server string) error {
+	client := llmgwv1connect.NewAgentServiceClient(a.http, server)
 
 	// parent is kept because the derived context is cancelled from inside this
 	// function; asking the derived one whether we are shutting down would

internal/agent/failover_test.go

@@ -0,0 +1,107 @@
+package agent
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
+)
+
+type standby bool
+
+func (s standby) IsLeader() bool { return !bool(s) }
+
+// startPlain runs an agent service over h2c, which is enough to test who
+// accepts a connection and who refuses it.
+//
+// A nil leader means none was configured, which is every deployment that
+// existed before leadership did and has to keep accepting agents.
+func startPlain(t *testing.T, leader agentsvc.Leader) (*inventory.Store, string) {
+	t.Helper()
+	inv := inventory.NewStore()
+	svc := agentsvc.New(inv, newRecorder(), false, slog.New(slog.DiscardHandler))
+	if leader != nil {
+		svc.SetLeadership(leader)
+	}
+
+	path, handler := svc.Handler()
+	mux := http.NewServeMux()
+	mux.Handle(path, handler)
+	// h2c, because the agent speaks HTTP/2 with prior knowledge and a plain
+	// server negotiates HTTP/1.1 — which turns a clear refusal into a frame
+	// parsing error that says nothing about why.
+	srv := httptest.NewServer(h2c.NewHandler(mux, &http2.Server{}))
+	t.Cleanup(srv.Close)
+	return inv, srv.URL
+}
+
+func TestAStandbyRefusesAnAgent(t *testing.T) {
+	// Commands travel down the stream the agent opened, so whichever control
+	// plane holds it is the only one that can start anything on that node. A
+	// standby holding it means the leader's placements fail with "no agent
+	// connected" for a node that is plainly connected.
+	_, addr := startPlain(t, standby(true))
+
+	ag := runAgent(t, Config{NodeID: "n1", Server: addr, MgmtAddr: "127.0.0.1"})
+	err := ag.session(context.Background(), ag.cfg.Server)
+	if err == nil {
+		t.Fatal("a standby accepted an agent")
+	}
+	if !strings.Contains(err.Error(), "standby") {
+		t.Errorf("the refusal does not say why: %v", err)
+	}
+}
+
+func TestAnAgentReachesTheLeaderPastAStandby(t *testing.T) {
+	// The point of the list. The standby is first, so the agent only registers
+	// if it moved on rather than sitting on the refusal.
+	_, standbyAddr := startPlain(t, standby(true))
+	leaderInv, leaderAddr := startPlain(t, standby(false))
+
+	ag := runAgent(t, Config{
+		NodeID: "n1", MgmtAddr: "127.0.0.1",
+		Servers: []string{standbyAddr, leaderAddr},
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	go ag.Run(ctx)
+
+	deadline := time.Now().Add(8 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, ok := leaderInv.Get("n1"); ok {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatal("the agent never reached the leader past the standby")
+}
+
+func TestALeaderStillAcceptsWithNoLeadershipConfigured(t *testing.T) {
+	// Every deployment before this had no leadership at all, and must keep
+	// accepting agents with nothing configured.
+	inv, addr := startPlain(t, nil)
+
+	ag := runAgent(t, Config{NodeID: "n1", Server: addr, MgmtAddr: "127.0.0.1"})
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	go ag.Run(ctx)
+
+	deadline := time.Now().Add(4 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, ok := inv.Get("n1"); ok {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatal("a service with no leadership configured refused an agent")
+}

internal/agent/mtls_test.go

@@ -255,7 +255,7 @@ func TestRefusedAgentLearnsWhy(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = ag.session(context.Background())
+	err = ag.session(context.Background(), ag.cfg.Server)
 	if err == nil {
 		t.Fatal("a refused session reported success")
 	}

internal/agent/servers_test.go

@@ -0,0 +1,90 @@
+package agent
+
+import (
+	"slices"
+	"strings"
+	"testing"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+func normalized(t *testing.T, c Config) Config {
+	t.Helper()
+	if err := c.normalizeServerURL(); err != nil {
+		t.Fatalf("normalising %+v: %v", c, err)
+	}
+	return c
+}
+
+func TestOneServerStillWorksExactlyAsBefore(t *testing.T) {
+	// Every agent configuration written before there could be a second control
+	// plane names one address, and none of them should have to change.
+	c := normalized(t, Config{Server: "10.0.0.140:7443"})
+	if !slices.Equal(c.Servers, []string{"http://10.0.0.140:7443"}) {
+		t.Errorf("servers = %v", c.Servers)
+	}
+	if c.Server != "http://10.0.0.140:7443" {
+		t.Errorf("server = %q", c.Server)
+	}
+}
+
+func TestTheAddressAlreadyThereStaysPreferred(t *testing.T) {
+	// An operator adding a standby to an existing file expects the control
+	// plane that was already working to keep being tried first.
+	c := normalized(t, Config{
+		Server:  "first:7443",
+		Servers: []string{"second:7443", "third:7443"},
+	})
+	want := []string{"http://first:7443", "http://second:7443", "http://third:7443"}
+	if !slices.Equal(c.Servers, want) {
+		t.Errorf("servers = %v, want %v", c.Servers, want)
+	}
+}
+
+func TestADuplicateAddressIsNotTriedTwiceInARow(t *testing.T) {
+	// Which would look exactly like the rotation not working.
+	c := normalized(t, Config{
+		Server:  "a:7443",
+		Servers: []string{"http://a:7443", "b:7443"},
+	})
+	want := []string{"http://a:7443", "http://b:7443"}
+	if !slices.Equal(c.Servers, want) {
+		t.Errorf("servers = %v, want %v", c.Servers, want)
+	}
+}
+
+func TestEveryAddressIsCheckedAgainstTheTLSSetting(t *testing.T) {
+	// The check existed for one address and has to hold for all of them: a
+	// standby reached over plaintext while the leader is authenticated would
+	// be a way around mutual TLS, not a fallback.
+	c := Config{
+		Servers: []string{"https://a:7443", "http://b:7443"},
+		TLS:     tlsconf.Config{CA: "ca.pem", Cert: "c.pem", Key: "k.pem"},
+	}
+	err := c.normalizeServerURL()
+	if err == nil {
+		t.Fatal("a plaintext standby was accepted alongside TLS")
+	}
+	if !strings.Contains(err.Error(), "b:7443") {
+		t.Errorf("the error does not say which address: %v", err)
+	}
+}
+
+func TestNoServerAtAllIsStillAnError(t *testing.T) {
+	c := Config{}
+	if err := c.normalizeServerURL(); err == nil {
+		t.Error("an agent with no control plane configured was accepted")
+	}
+}
+
+func TestSchemeIsAddedPerAddress(t *testing.T) {
+	c := normalized(t, Config{
+		Servers: []string{"a:7443", "b:7443"},
+		TLS:     tlsconf.Config{CA: "ca.pem", Cert: "c.pem", Key: "k.pem"},
+	})
+	for _, s := range c.Servers {
+		if !strings.HasPrefix(s, "https://") {
+			t.Errorf("%q did not get the secure scheme", s)
+		}
+	}
+}

internal/cli/server.go

@@ -342,15 +342,23 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live
 	// The same shape as the usage assertion above, and for the same reason:
 	// the sink is the router when the router is serving, and nothing when a
 	// control plane runs on its own.
+	var lead *control.Leadership
 	if h, ok := sink.(interface{ Store() *storepkg.SQLite }); ok {
 		if st := h.Store(); st != nil {
-			lead := control.NewLeadership(st, cfg.Control.LeaseTTL, log)
+			lead = control.NewLeadership(st, cfg.Control.LeaseTTL, log)
 			ctl.SetLeadership(lead)
 			log.Info("competing to be the control plane that places work",
 				"owner", lead.Owner())
 		}
 	}
 	svc := agentsvc.New(ctl.Inventory(), ctl.Reconciler(), requireIdentity, log)
+	// A standby must refuse agents as well as refusing to place. Commands
+	// travel down the stream the agent opened, so a standby holding one is a
+	// node the leader cannot command — and the leader's placements would fail
+	// with "no agent connected" for a node that is plainly connected.
+	if lead != nil {
+		svc.SetLeadership(lead)
+	}
 	fwd.set(svc)
 
 	ctl.Start(ctx)

internal/control/agentsvc/service.go

@@ -31,6 +31,9 @@ type StatusSink interface {
 // time the reconciler's Dispatcher: it holds the open streams, so it is the
 // only thing that can deliver a command to a node.
 type Service struct {
+	// leader gates whether this process accepts agent streams at all.
+	leader Leader
+
 	inv    *inventory.Store
 	status StatusSink
 	log    *slog.Logger
@@ -86,6 +89,19 @@ func (s *Service) Connect(
 	ctx context.Context,
 	stream *connect.BidiStream[llmgwv1.AgentMessage, llmgwv1.ControlMessage],
 ) error {
+	// A standby must not hold an agent's stream. Commands travel down the
+	// stream the agent opened, so whichever control plane holds it is the only
+	// one that can start or stop anything on that node — and if that is not
+	// the leader, the leader's placements fail with "no agent connected" for a
+	// node that is plainly connected.
+	//
+	// Refused before the opening message is read, so the agent spends nothing
+	// on a connection that cannot serve it.
+	if s.leader != nil && !s.leader.IsLeader() {
+		return connect.NewError(connect.CodeUnavailable,
+			errors.New("this control plane is a standby; connect to the leader"))
+	}
+
 	first, err := stream.Receive()
 	if err != nil {
 		return connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("no opening message: %w", err))
@@ -345,6 +361,16 @@ func (s *Service) ConnectedNodes() []string {
 
 // --- Dispatcher ---
 
+// Leader reports whether this process is the one that should be holding agent
+// streams.
+type Leader interface{ IsLeader() bool }
+
+// SetLeadership makes this service refuse agents while it is a standby.
+//
+// Nil accepts them unconditionally, which is what a single control plane wants
+// and what every deployment before leadership had.
+func (s *Service) SetLeadership(l Leader) { s.leader = l }
+
 // ErrNotConnected means no agent stream is open for that node.
 var ErrNotConnected = errors.New("no agent connected")