Mirrors kept in step: llmgw mirror serve --peer

Each mirror reads its peers every thirty seconds and copies the
versions a peer holds whole that it lacks, keeping a copy only if its
checksum is the peer's. A removal is kept with its time (.removed.json)
and listed for peers: a version removed while a peer was down is removed
there when it is back, not copied back from it; one added again after
its removal stands on both. Neither mirror leads.

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

internal/cli/mirror.go

@@ -255,6 +255,7 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 	var tlsc tlsconf.Config
 	var allow, clients []string
 	var allowPrivate bool
+	var peers []string
 
 	c := &cobra.Command{
 		Use:   "serve",
@@ -317,6 +318,21 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 				ReadTimeout: time.Minute,
 			})
 
+			// Another mirror kept in step with this one, each copying what
+			// the other has: the cluster's models while one machine is down.
+			if len(peers) > 0 {
+				p := &mirror.Peers{M: mgr, Log: log}
+				for _, addr := range peers {
+					c, err := mirror.NewClient(addr, tlsc, 30*time.Second)
+					if err != nil {
+						return fmt.Errorf("--peer %s: %w", addr, err)
+					}
+					p.Clients = append(p.Clients, c)
+				}
+				go p.Run(cmd.Context())
+				fmt.Fprintf(out, "kept in step with %s, every %s\n", strings.Join(peers, ", "), mirror.PeerInterval)
+			}
+
 			errs := make(chan error, 2)
 			go func() { errs <- fmt.Errorf("serving weights: %w", files.ListenAndServe()) }()
 			fmt.Fprintf(out, "serving %s on %s\n", *dir, listen)
@@ -343,6 +359,9 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 		},
 	}
 	c.Flags().StringVar(&listen, "listen", ":8081", "address to serve the weights on")
+	c.Flags().StringArrayVar(&peers, "peer", nil,
+		"another mirror's management API to keep in step with, each copying what the other has; repeatable. "+
+			"Give each the other's address in --api-allow (or its certificate in --api-client)")
 	c.Flags().StringVar(&apiListen, "api-listen", ":8082", "address to serve the management API on")
 	c.Flags().StringVar(&base, "base", "http://localhost:8081",
 		"the address nodes use to reach this mirror, as it should appear in catalog sources")

internal/mirror/api.go

@@ -52,6 +52,9 @@ type ModelsResponse struct {
 	// when it is not.
 	Base   string   `json:"base"`
 	Models []Status `json:"models"`
+	// Removed are the versions removed here, model/version to when (ms),
+	// for a peer mirror to remove too.
+	Removed map[string]int64 `json:"removed,omitempty"`
 }
 
 func (a *API) list(w http.ResponseWriter, _ *http.Request) {
@@ -60,7 +63,7 @@ func (a *API) list(w http.ResponseWriter, _ *http.Request) {
 		fail(w, http.StatusInternalServerError, err)
 		return
 	}
-	writeJSON(w, http.StatusOK, ModelsResponse{Base: a.base, Models: all})
+	writeJSON(w, http.StatusOK, ModelsResponse{Base: a.base, Models: all, Removed: a.m.Tombstones()})
 }
 
 // FetchRequest asks the mirror to download a model.

internal/mirror/manager.go

@@ -43,6 +43,9 @@ type Status struct {
 
 	StartedAt  time.Time `json:"started_at,omitzero"`
 	FinishedAt time.Time `json:"finished_at,omitzero"`
+	// AddedAt is when the file came to this mirror, as its modification
+	// time says: set against a removal elsewhere, the later stands.
+	AddedAt time.Time `json:"added_at,omitzero"`
 }
 
 // Percent reports progress, and false when the total is unknown.
@@ -106,6 +109,16 @@ var ErrAlreadyFetching = errors.New("a download for that version is already runn
 // the API call that asked for it, which is the point of doing it here rather
 // than making the caller hold a connection open for an hour.
 func (m *Manager) Start(source string, e Entry) (Status, error) {
+	return m.start(source, e, "", m.client)
+}
+
+// startCopy fetches a file a peer mirror holds, from a private address,
+// and keeps it only if its checksum is the peer's.
+func (m *Manager) startCopy(source string, e Entry, sha string) (Status, error) {
+	return m.start(source, e, sha, fetchClient(true))
+}
+
+func (m *Manager) start(source string, e Entry, expect string, client *http.Client) (Status, error) {
 	if source == "" {
 		return Status{}, fmt.Errorf("a source URL is required")
 	}
@@ -132,16 +145,16 @@ func (m *Manager) Start(source string, e Entry) (Status, error) {
 	out := *st
 	m.mu.Unlock()
 
-	go m.run(source, e, key)
+	go m.run(source, e, key, expect, client)
 	return out, nil
 }
 
-func (m *Manager) run(source string, e Entry, key string) {
+func (m *Manager) run(source string, e Entry, key, expect string, client *http.Client) {
 	// Its own context, deliberately unbounded by any request. A download is
 	// minutes to hours and the thing that asked for it is long gone.
 	ctx := context.Background()
 
-	added, err := addWith(ctx, m.client, m.root, source, e, func(done, total int64) {
+	added, err := addWith(ctx, client, m.root, source, e, func(done, total int64) {
 		m.mu.Lock()
 		if st, ok := m.running[key]; ok {
 			st.BytesDone, st.BytesTotal = done, total
@@ -149,6 +162,16 @@ func (m *Manager) run(source string, e Entry, key string) {
 		m.mu.Unlock()
 	})
 
+	if err == nil && expect != "" && added.SHA256 != expect {
+		// Not what the peer holds: a file changed or cut short on the way.
+		_ = Remove(m.root, e)
+		err = fmt.Errorf("the copy's sha256 %s is not the peer's %s", added.SHA256, expect)
+	}
+	if err == nil {
+		// Added again after it was removed: the new one stands.
+		m.clearTombstone(e.Model, e.Version)
+	}
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	st, ok := m.running[key]
@@ -284,6 +307,9 @@ func (m *Manager) All() ([]Status, error) {
 			Model: e.Model, Version: e.Version, File: e.File,
 			State: Present, BytesDone: e.Bytes, BytesTotal: e.Bytes,
 		}
+		if fi, err := os.Stat(filepath.Join(m.root, e.Model, e.Version, e.File)); err == nil {
+			st.AddedAt = fi.ModTime()
+		}
 		// The checksum and the source are not recoverable from a directory
 		// listing, and the checksum is the one field worth having: it is what
 		// an operator compares against what upstream published. Dropping it in
@@ -364,6 +390,9 @@ func (m *Manager) RemoveVersion(model, version string) ([]string, error) {
 		return nil, fmt.Errorf("%s/%s: %w", model, version, os.ErrNotExist)
 	}
 	m.Forget(model, version)
+	// Kept, so a peer mirror that still has it removes it too, and does
+	// not copy it back here.
+	m.recordTombstone(model, version, time.Now())
 	m.log.Info("mirror removed a version", "model", model, "version", version, "files", len(removed))
 	return removed, nil
 }

internal/mirror/peer.go

@@ -0,0 +1,124 @@
+package mirror
+
+import (
+	"context"
+	"log/slog"
+	"strings"
+	"time"
+)
+
+// Mirrors kept in step, so the cluster has its models while one machine is
+// down: each reads its peers every while, copies the versions a peer has
+// that it lacks, checksum checked, and removes those a peer removed after
+// they came here. Neither leads; what was done last stands.
+
+// PeerInterval is how often each peer is read.
+const PeerInterval = 30 * time.Second
+
+// Peers keeps one mirror in step with others.
+type Peers struct {
+	M       *Manager
+	Clients []*Client
+	Log     *slog.Logger
+}
+
+// Run reads the peers until ctx ends.
+func (p *Peers) Run(ctx context.Context) {
+	t := time.NewTicker(PeerInterval)
+	defer t.Stop()
+	for {
+		for _, c := range p.Clients {
+			p.SyncOnce(ctx, c)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		}
+	}
+}
+
+// SyncOnce brings this mirror in step with one peer.
+func (p *Peers) SyncOnce(ctx context.Context, c *Client) {
+	ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	theirs, err := c.List(ctx)
+	if err != nil {
+		p.Log.Warn("could not read a peer mirror; trying again shortly", "peer", c.Addr(), "err", err)
+		return
+	}
+	mine, err := p.M.All()
+	if err != nil {
+		p.Log.Warn("could not read this mirror", "err", err)
+		return
+	}
+	here := map[string]Status{}
+	for _, st := range mine {
+		here[st.Model+"/"+st.Version+"/"+st.File] = st
+	}
+	removed := p.M.Tombstones()
+
+	// What the peer removed after it came here goes here too.
+	for key, at := range theirs.Removed {
+		model, version, ok := strings.Cut(key, "/")
+		if !ok {
+			continue
+		}
+		newest := time.Time{}
+		has := false
+		for _, st := range mine {
+			if st.Model == model && st.Version == version {
+				has = true
+				if st.AddedAt.After(newest) {
+					newest = st.AddedAt
+				}
+			}
+		}
+		if has && newest.UnixMilli() < at {
+			if _, err := p.M.RemoveVersion(model, version); err == nil {
+				p.Log.Info("removed a version a peer mirror removed", "peer", c.Addr(), "model", model, "version", version)
+			}
+		} else if !has && at > removed[key] {
+			p.M.recordTombstone(model, version, time.UnixMilli(at))
+		}
+	}
+
+	// Versions whole there and missing here: every file present, each with
+	// its checksum, so a copy can be checked.
+	type version struct {
+		files []Status
+		whole bool
+	}
+	versions := map[string]*version{}
+	for _, st := range theirs.Models {
+		key := st.Model + "/" + st.Version
+		v := versions[key]
+		if v == nil {
+			v = &version{whole: true}
+			versions[key] = v
+		}
+		v.files = append(v.files, st)
+		v.whole = v.whole && st.State == Present && st.SHA256 != ""
+	}
+	for key, v := range versions {
+		if !v.whole {
+			continue
+		}
+		for _, st := range v.files {
+			if _, ok := here[key+"/"+st.File]; ok {
+				continue
+			}
+			// Removed here after the peer had it: not copied back. The
+			// peer removes its own when it reads this one.
+			if at, gone := removed[key]; gone && st.AddedAt.UnixMilli() < at {
+				continue
+			}
+			source := strings.TrimRight(theirs.Base, "/") + "/" + st.Model + "/" + st.Version + "/" + st.File
+			e := Entry{Model: st.Model, Version: st.Version, File: st.File}
+			if _, err := p.M.startCopy(source, e, st.SHA256); err == nil {
+				p.Log.Info("copying a version from a peer mirror", "peer", c.Addr(), "model", st.Model,
+					"version", st.Version, "file", st.File)
+			}
+		}
+	}
+}

internal/mirror/peer_test.go

@@ -0,0 +1,127 @@
+package mirror
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+// testMirror is a mirror with its file server and management API, as a peer
+// reaches it.
+type testMirror struct {
+	root  string
+	m     *Manager
+	files *httptest.Server
+	api   *httptest.Server
+}
+
+func newTestMirror(t *testing.T) *testMirror {
+	t.Helper()
+	root := t.TempDir()
+	tm := &testMirror{root: root, m: NewManager(root, discardLog())}
+	tm.m.AllowPrivateSources()
+	tm.files = httptest.NewServer(http.FileServer(http.Dir(root)))
+	tm.api = httptest.NewServer(NewAPI(tm.m, tm.files.URL).Handler())
+	t.Cleanup(func() { tm.files.Close(); tm.api.Close() })
+	return tm
+}
+
+func (tm *testMirror) client(t *testing.T) *Client {
+	c, err := NewClient(tm.api.URL, tlsconf.Config{}, 5*time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c
+}
+
+func (tm *testMirror) has(model, version string) bool {
+	st, ok := tm.m.Status(model, version)
+	return ok && st.State == Present
+}
+
+func waitPresent(t *testing.T, tm *testMirror, model, version string) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if tm.has(model, version) {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	st, _ := tm.m.Status(model, version)
+	t.Fatalf("%s/%s never arrived: %+v", model, version, st)
+}
+
+func TestTwoMirrorsKeepInStepAndAgreeOnWhatWasRemoved(t *testing.T) {
+	blob := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte("weights of a small model"))
+	}))
+	defer blob.Close()
+	a, b := newTestMirror(t), newTestMirror(t)
+	if _, err := a.m.Start(blob.URL+"/m.gguf", Entry{Model: "qwen", Version: "v1", File: "m.gguf"}); err != nil {
+		t.Fatal(err)
+	}
+	waitPresent(t, a, "qwen", "v1")
+
+	// b copies what a has, its checksum checked.
+	pb := &Peers{M: b.m, Log: discardLog()}
+	pb.SyncOnce(t.Context(), a.client(t))
+	waitPresent(t, b, "qwen", "v1")
+	got, _ := os.ReadFile(filepath.Join(b.root, "qwen", "v1", "m.gguf"))
+	if string(got) != "weights of a small model" {
+		t.Errorf("b holds %q", got)
+	}
+
+	// Removed on a: a does not copy it back from b, and b removes it too.
+	time.Sleep(10 * time.Millisecond)
+	if _, err := a.m.RemoveVersion("qwen", "v1"); err != nil {
+		t.Fatal(err)
+	}
+	pa := &Peers{M: a.m, Log: discardLog()}
+	pa.SyncOnce(t.Context(), b.client(t))
+	time.Sleep(200 * time.Millisecond)
+	if a.has("qwen", "v1") {
+		t.Fatal("a copied back from b what it had removed")
+	}
+	pb.SyncOnce(t.Context(), a.client(t))
+	if b.has("qwen", "v1") {
+		t.Error("b kept what a removed")
+	}
+
+	// Added again on a, after the removal: the new one stands, on both.
+	time.Sleep(10 * time.Millisecond)
+	if _, err := a.m.Start(blob.URL+"/m.gguf", Entry{Model: "qwen", Version: "v1", File: "m.gguf"}); err != nil {
+		t.Fatal(err)
+	}
+	waitPresent(t, a, "qwen", "v1")
+	pb.SyncOnce(t.Context(), a.client(t))
+	waitPresent(t, b, "qwen", "v1")
+}
+
+// A file that is not what the peer says it is, is not kept.
+func TestACopyWhoseChecksumIsNotThePeersIsNotKept(t *testing.T) {
+	served := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte("something else"))
+	}))
+	defer served.Close()
+	b := newTestMirror(t)
+	e := Entry{Model: "qwen", Version: "v1", File: "m.gguf"}
+	if _, err := b.m.startCopy(served.URL+"/m.gguf", e, "0000"); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if st, _ := b.m.Status("qwen", "v1"); st.State == Failed {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	if _, err := os.Stat(e.Path(b.root)); !os.IsNotExist(err) {
+		t.Errorf("a copy with the wrong checksum was kept: %v", err)
+	}
+}

internal/mirror/tombstone.go

@@ -0,0 +1,59 @@
+package mirror
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// Removals kept so mirrors in step agree on them: a version removed here
+// while a peer was down is removed there when it is back, rather than copied
+// back here from it. Kept in the mirror's own bookkeeping, a dot file the
+// file server never serves as a model.
+
+const tombstoneFile = ".removed.json"
+
+// Tombstones are the versions removed, model/version to when, in ms.
+func (m *Manager) Tombstones() map[string]int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.readTombstones()
+}
+
+func (m *Manager) readTombstones() map[string]int64 {
+	out := map[string]int64{}
+	b, err := os.ReadFile(filepath.Join(m.root, tombstoneFile))
+	if err == nil {
+		_ = json.Unmarshal(b, &out)
+	}
+	return out
+}
+
+func (m *Manager) writeTombstones(t map[string]int64) {
+	b, _ := json.Marshal(t)
+	tmp := filepath.Join(m.root, tombstoneFile+".tmp")
+	if os.WriteFile(tmp, b, 0o644) == nil {
+		_ = os.Rename(tmp, filepath.Join(m.root, tombstoneFile))
+	}
+}
+
+func (m *Manager) recordTombstone(model, version string, at time.Time) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	t := m.readTombstones()
+	if at.UnixMilli() > t[model+"/"+version] {
+		t[model+"/"+version] = at.UnixMilli()
+		m.writeTombstones(t)
+	}
+}
+
+func (m *Manager) clearTombstone(model, version string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	t := m.readTombstones()
+	if _, ok := t[model+"/"+version]; ok {
+		delete(t, model+"/"+version)
+		m.writeTombstones(t)
+	}
+}