The control plane uses whichever mirror answers

mirror.peers names the mirrors kept in step with mirror.api. The
mirror's management calls go to the first that can be reached, and
each mirror is asked every fifteen seconds whether it answers: a shard
is sent its model's and adapters' address on one that does while the
mirror the catalog names is down.

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

docs/reference.md

@@ -180,6 +180,7 @@ rather than repeated.
 | `parallel` | `int` | [LlamaCppConfig](#llamacppconfig) |
 | `threads` | `int` | [LlamaCppConfig](#llamacppconfig) |
 | `api` | `string` | [Mirror](#mirror) |
+| `peers` | `[]string` | [Mirror](#mirror) |
 | `tls` | `tlsconf.Config` | [Mirror](#mirror) |
 | `registry` | `string` | [Mirror](#mirror) |
 | `timeout` | `time.Duration` | [Mirror](#mirror) |
@@ -1722,6 +1723,14 @@ The management port, not the one nodes fetch bytes from. Those are two
 listeners on the mirror precisely so this one can be locked down without
 re-certifying every agent's fetcher.
 
+### `peers`
+
+`[]string`
+
+Peers are the management addresses of mirrors kept in step with this
+one (llmgw mirror serve --peer): asked when it cannot be reached, and
+a node given the same file on one of them while it is down.
+
 ### `tls`
 
 `tlsconf.Config`

internal/cli/node.go

@@ -186,7 +186,7 @@ func (na *nodeAdd) prepare() error {
 	}
 	// Where the mirror serves model files, for the node to be asked to reach.
 	if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil && cfg.Mirror.API != "" {
-		if mc, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 5*time.Second); err == nil {
+		if mc, err := mirror.NewFailoverClient(append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...), cfg.Mirror.TLS, 5*time.Second); err == nil {
 			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 			if list, err := mc.List(ctx); err == nil {
 				o.MirrorBase = list.Base

internal/cli/server.go

@@ -567,10 +567,21 @@ func newServerCmd() *cobra.Command {
 			// rather than turning into a broken admin page hours later when
 			// somebody finally opens it.
 			if cfg.Mirror.API != "" {
-				mc, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, cfg.Mirror.Timeout)
+				mirrors := append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...)
+				mc, err := mirror.NewFailoverClient(mirrors, cfg.Mirror.TLS, cfg.Mirror.Timeout)
 				if err != nil {
 					return fmt.Errorf("the model mirror: %w", err)
 				}
+				// Mirrors kept in step: a node is given a model's file on one
+				// that answers while the one the catalog names is down.
+				if len(cfg.Mirror.Peers) > 0 {
+					src, err := mirror.NewSources(mirrors, cfg.Mirror.TLS, log)
+					if err != nil {
+						return fmt.Errorf("the model mirrors: %w", err)
+					}
+					go src.Run(ctx)
+					agentsvc.SetSourceRewrite(src.Rewrite)
+				}
 				rt.SetMirror(mc)
 				if fwPub != nil {
 					fwPub.setMirror(mc)
@@ -578,7 +589,7 @@ func newServerCmd() *cobra.Command {
 				// Adapters trained on the nodes are kept there too; sending one
 				// takes longer than the mirror's usual answers.
 				if adminAPI != nil {
-					if up, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 10*time.Minute); err == nil {
+					if up, err := mirror.NewFailoverClient(mirrors, cfg.Mirror.TLS, 10*time.Minute); err == nil {
 						adminAPI.SetTrainMirror(up)
 					}
 				}

internal/cli/tui_mirror.go

@@ -32,7 +32,7 @@ func (m consoleModel) mirrorClient() (*mirror.Client, error) {
 	if cfg.Mirror.API == "" {
 		return nil, errors.New("no mirror: mirror.api in server.yaml is empty")
 	}
-	return mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 10*time.Second)
+	return mirror.NewFailoverClient(append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...), cfg.Mirror.TLS, 10*time.Second)
 }
 
 func (m consoleModel) fetchMirror() tea.Cmd {

internal/config/config.go

@@ -345,6 +345,10 @@ type Mirror struct {
 	// listeners on the mirror precisely so this one can be locked down without
 	// re-certifying every agent's fetcher.
 	API string `koanf:"api"`
+	// Peers are the management addresses of mirrors kept in step with this
+	// one (llmgw mirror serve --peer): asked when it cannot be reached, and
+	// a node given the same file on one of them while it is down.
+	Peers []string `koanf:"peers"`
 	// TLS authenticates to the mirror's management API, with the same CA the
 	// agents use. Empty leaves the connection in plain HTTP, which is only
 	// acceptable on a network where nothing untrusted can reach that port —

internal/control/agentsvc/convert.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"maps"
 	"runtime"
+	"slices"
+	"sync/atomic"
 	"time"
 
 	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
@@ -189,8 +191,31 @@ func AdoptedFromRegister(nodeID string, shards []*llmgwv1.RunningShard) []reconc
 }
 
 // StartShardFromCommand converts a reconciler command into a wire message.
+// sourceRewrite is where a model is fetched while its mirror is down: set
+// when mirrors are kept in step, else nil and the catalog's address as it is.
+var sourceRewrite atomic.Pointer[func(string) string]
+
+// SetSourceRewrite has every shard started from now on fetch its model, and
+// adapters, from the address f gives: one on a mirror that answers.
+func SetSourceRewrite(f func(string) string) { sourceRewrite.Store(&f) }
+
+func rewriteSource(src string) string {
+	if f := sourceRewrite.Load(); f != nil && src != "" {
+		return (*f)(src)
+	}
+	return src
+}
+
 func StartShardFromCommand(cmd reconciler.ShardCommand) *llmgwv1.StartShard {
 	p := cmd.Spec.Profile
+	p.ModelPath = rewriteSource(p.ModelPath)
+	if len(p.Adapters) > 0 {
+		ads := slices.Clone(p.Adapters)
+		for i := range ads {
+			ads[i].Path = rewriteSource(ads[i].Path)
+		}
+		p.Adapters = ads
+	}
 	return &llmgwv1.StartShard{
 		ReplicaId:     cmd.ReplicaID,
 		ShardId:       cmd.ShardID,

internal/control/agentsvc/convert_test.go

@@ -1,6 +1,7 @@
 package agentsvc
 
 import (
+	"strings"
 	"testing"
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
@@ -18,3 +19,20 @@ func TestAdaptersTravelToTheAgentInOrder(t *testing.T) {
 		t.Errorf("adapters %+v", got)
 	}
 }
+
+// While its mirror is down, a shard is sent its model and adapters on the
+// mirror the rewrite names, and the command it was made from is left as it is.
+func TestAShardIsSentItsModelOnAMirrorThatAnswers(t *testing.T) {
+	SetSourceRewrite(func(s string) string { return strings.Replace(s, "http://mirror-a", "http://mirror-b", 1) })
+	t.Cleanup(func() { SetSourceRewrite(func(s string) string { return s }) })
+	ads := []engines.Adapter{{Name: "sql", Path: "http://mirror-a/sql.gguf"}}
+	cmd := reconciler.ShardCommand{ReplicaID: "r", ShardID: "s", Spec: engines.ShardSpec{
+		Profile: engines.Profile{Name: "cpu", Engine: "llamacpp", ModelPath: "http://mirror-a/qwen/v1/", Adapters: ads}}}
+	spec, _ := ShardSpecFromStart(StartShardFromCommand(cmd))
+	if spec.Profile.ModelPath != "http://mirror-b/qwen/v1/" || spec.Profile.Adapters[0].Path != "http://mirror-b/sql.gguf" {
+		t.Errorf("sent %+v", spec.Profile)
+	}
+	if ads[0].Path != "http://mirror-a/sql.gguf" {
+		t.Error("the catalog's adapter was changed in place")
+	}
+}

internal/docgen/reference_gen.go

@@ -230,6 +230,7 @@ var Reference = []Block{
 	}},
 	{Package: "config", Name: "Mirror", Title: "Mirror", Doc: "", Fields: []Field{
 		{Key: "api", Type: "string", Doc: "API is the mirror's management address, scheme included, for example\nhttps://models.internal:8082. Empty means no mirror is configured.\n\nThe management port, not the one nodes fetch bytes from. Those are two\nlisteners on the mirror precisely so this one can be locked down without\nre-certifying every agent's fetcher.", Nested: ""},
+		{Key: "peers", Type: "[]string", Doc: "Peers are the management addresses of mirrors kept in step with this\none (llmgw mirror serve --peer): asked when it cannot be reached, and\na node given the same file on one of them while it is down.", Nested: ""},
 		{Key: "tls", Type: "tlsconf.Config", Doc: "TLS authenticates to the mirror's management API, with the same CA the\nagents use. Empty leaves the connection in plain HTTP, which is only\nacceptable on a network where nothing untrusted can reach that port —\nthe mirror says as much in its own log when it starts that way.", Nested: "tlsconf.Config"},
 		{Key: "registry", Type: "string", Doc: "Registry is the model registry the mirror page browses, empty for the\npublic Hugging Face. Set it to an internal mirror of the registry where\nthe control plane cannot reach the internet — or leave the whole mirror\nblock out, which turns the explorer off rather than giving it a search\nbox that can only time out.", Nested: ""},
 		{Key: "timeout", Type: "time.Duration", Doc: "Timeout bounds a management call. It does not bound a download: the\nmirror answers a fetch as soon as it has started one, and the bytes keep\narriving long after this deadline has passed.", Nested: ""},

internal/mirror/client.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -24,6 +25,29 @@ import (
 type Client struct {
 	base string
 	hc   *http.Client
+	// others are the mirrors kept in step with this one, asked in turn when
+	// it cannot be reached: the same models, on another machine.
+	others []string
+}
+
+// NewFailoverClient is NewClient over several mirrors kept in step: the
+// first answers, and the next is asked when it cannot be reached.
+func NewFailoverClient(addrs []string, tc tlsconf.Config, timeout time.Duration) (*Client, error) {
+	if len(addrs) == 0 {
+		return nil, fmt.Errorf("no mirror address")
+	}
+	c, err := NewClient(addrs[0], tc, timeout)
+	if err != nil {
+		return nil, err
+	}
+	for _, a := range addrs[1:] {
+		u, err := url.Parse(a)
+		if err != nil || u.Host == "" {
+			return nil, fmt.Errorf("the mirror address %q is not a URL with a host", a)
+		}
+		c.others = append(c.others, strings.TrimSuffix(a, "/"))
+	}
+	return c, nil
 }
 
 // NewClient dials a mirror's management API, with mutual TLS when configured.
@@ -149,28 +173,42 @@ func (c *Client) Remove(ctx context.Context, model, version string) ([]string, e
 }
 
 func (c *Client) do(ctx context.Context, method, path string, body, out any) error {
-	var rdr io.Reader
+	var b []byte
 	if body != nil {
-		b, err := json.Marshal(body)
+		var err error
+		if b, err = json.Marshal(body); err != nil {
+			return err
+		}
+	}
+	// The mirror first, then those kept in step with it, while each cannot
+	// be reached at all: one that answers, even with a refusal, is the one.
+	var resp *http.Response
+	var errs []error
+	for _, base := range append([]string{c.base}, c.others...) {
+		var rdr io.Reader
+		if body != nil {
+			rdr = bytes.NewReader(b)
+		}
+		req, err := http.NewRequestWithContext(ctx, method, base+path, rdr)
 		if err != nil {
 			return err
 		}
-		rdr = bytes.NewReader(b)
+		if body != nil {
+			req.Header.Set("Content-Type", "application/json")
+		}
+		if resp, err = c.hc.Do(req); err == nil {
+			break
+		}
+		// Wrapped with the address because the failure an operator sees on
+		// a page is otherwise "connection refused" with no clue which of
+		// the machines refused it.
+		errs = append(errs, fmt.Errorf("reaching the mirror at %s: %w", base, err))
+		if ctx.Err() != nil {
+			break
+		}
 	}
-	req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr)
-	if err != nil {
-		return err
-	}
-	if body != nil {
-		req.Header.Set("Content-Type", "application/json")
-	}
-
-	resp, err := c.hc.Do(req)
-	if err != nil {
-		// Wrapped with the address because the failure an operator sees on a
-		// page is otherwise "connection refused" with no clue which of the two
-		// machines refused it.
-		return fmt.Errorf("reaching the mirror at %s: %w", c.base, err)
+	if resp == nil {
+		return errors.Join(errs...)
 	}
 	defer resp.Body.Close()
 

internal/mirror/peer_test.go

@@ -125,3 +125,50 @@ func TestACopyWhoseChecksumIsNotThePeersIsNotKept(t *testing.T) {
 		t.Errorf("a copy with the wrong checksum was kept: %v", err)
 	}
 }
+
+// A mirror that cannot be reached is passed over for one kept in step with it.
+func TestTheNextMirrorIsAskedWhenTheFirstCannotBeReached(t *testing.T) {
+	a, b := newTestMirror(t), newTestMirror(t)
+	c, err := NewFailoverClient([]string{a.api.URL, b.api.URL}, tlsconf.Config{}, 2*time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if list, err := c.List(t.Context()); err != nil || list.Base != a.files.URL {
+		t.Fatalf("both up: %+v %v", list, err)
+	}
+	a.api.Close()
+	if list, err := c.List(t.Context()); err != nil || list.Base != b.files.URL {
+		t.Errorf("the first down: %+v %v", list, err)
+	}
+	b.api.Close()
+	if _, err := c.List(t.Context()); err == nil {
+		t.Error("both down, and an answer")
+	}
+}
+
+// A node is given a model's file on a mirror that answers.
+func TestAModelsAddressIsMovedToAMirrorThatAnswers(t *testing.T) {
+	a, b := newTestMirror(t), newTestMirror(t)
+	s, err := NewSources([]string{a.api.URL, b.api.URL}, tlsconf.Config{}, discardLog())
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.Check(t.Context())
+	src := a.files.URL + "/qwen/v1/m.gguf"
+	if got := s.Rewrite(src); got != src {
+		t.Errorf("its own mirror up: %s", got)
+	}
+	a.api.Close()
+	s.Check(t.Context())
+	if got := s.Rewrite(src); got != b.files.URL+"/qwen/v1/m.gguf" {
+		t.Errorf("its own mirror down: %s", got)
+	}
+	if got := s.Rewrite("https://huggingface.co/x/y"); got != "https://huggingface.co/x/y" {
+		t.Errorf("an address on no mirror: %s", got)
+	}
+	b.api.Close()
+	s.Check(t.Context())
+	if got := s.Rewrite(src); got != src {
+		t.Errorf("none up: %s", got)
+	}
+}

internal/mirror/sources.go

@@ -0,0 +1,102 @@
+package mirror
+
+import (
+	"context"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+// Sources is where a node fetches a model while one mirror is down. A model
+// in the catalog names one mirror's address; when that mirror does not
+// answer, the address a node is given is the same file on one that does.
+
+// SourcesInterval is how often each mirror is asked whether it answers.
+const SourcesInterval = 15 * time.Second
+
+// Sources watches mirrors kept in step, and rewrites a model's address to
+// one that answers.
+type Sources struct {
+	clients []*Client
+	log     *slog.Logger
+
+	mu      sync.Mutex
+	bases   []string
+	healthy []bool
+}
+
+// NewSources watches the mirrors at these management addresses.
+func NewSources(addrs []string, tc tlsconf.Config, log *slog.Logger) (*Sources, error) {
+	s := &Sources{log: log, bases: make([]string, len(addrs)), healthy: make([]bool, len(addrs))}
+	for _, a := range addrs {
+		c, err := NewClient(a, tc, 5*time.Second)
+		if err != nil {
+			return nil, err
+		}
+		s.clients = append(s.clients, c)
+	}
+	return s, nil
+}
+
+// Run asks each mirror, every while, until ctx ends.
+func (s *Sources) Run(ctx context.Context) {
+	t := time.NewTicker(SourcesInterval)
+	defer t.Stop()
+	for {
+		s.Check(ctx)
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		}
+	}
+}
+
+// Check asks each mirror once: whether it answers, and the address its
+// files are served at.
+func (s *Sources) Check(ctx context.Context) {
+	for i, c := range s.clients {
+		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		list, err := c.List(ctx)
+		cancel()
+		s.mu.Lock()
+		was := s.healthy[i]
+		s.healthy[i] = err == nil
+		if err == nil && list.Base != "" {
+			s.bases[i] = strings.TrimSuffix(list.Base, "/")
+		}
+		s.mu.Unlock()
+		if was && err != nil {
+			s.log.Warn("a mirror does not answer; models it holds are fetched from another", "mirror", c.Addr(), "err", err)
+		} else if !was && err == nil {
+			s.log.Info("a mirror answers", "mirror", c.Addr(), "files", list.Base)
+		}
+	}
+}
+
+// Rewrite is a model's address on a mirror that answers: as it is when its
+// own mirror answers or no other does, else the same path on the first that
+// does.
+func (s *Sources) Rewrite(src string) string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	from := -1
+	for i, b := range s.bases {
+		if b != "" && (src == b || strings.HasPrefix(src, b+"/")) {
+			from = i
+			break
+		}
+	}
+	if from < 0 || s.healthy[from] {
+		return src
+	}
+	for i, b := range s.bases {
+		if i != from && b != "" && s.healthy[i] {
+			return b + strings.TrimPrefix(src, s.bases[from])
+		}
+	}
+	return src
+}