A set's files checked four at once, the check said as it goes and not taken for a stuck download

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

CHANGELOG.md

@@ -33,6 +33,10 @@ APIs change between minors without a deprecation period.
 - **A model of several files showed one file's download as the whole.** A node fetching a set reported the part it
   was on — "59% of 46.1 GB" for a 188 GB model — instead of the set: it now reports the set's bytes against its
   whole size.
+- **A set's files checked with nothing said, one at a time.** After the last byte arrived a node checked each file
+  against the mirror's checksum, 195 GB on one core for eight minutes, while the replica still read "fetching 100%".
+  It now checks four at once and says so: `checking 40% of 195 GB` in `ctl replicas`, "Checking the weights" in the
+  chat page; and a check that takes long is not taken for a stuck download.
 
 ## v0.16.1 — 2026-10-30
 

api/gen/llmgw/v1/agent.pb.go

@@ -3586,7 +3586,10 @@ type FetchProgress struct {
 	// bytes_total is 0 when the server did not say how large the file is, which
 	// is a real answer: a percentage would then be invented, and an invented
 	// percentage that stops moving is worse than a byte count that keeps rising.
-	BytesTotal    int64 `protobuf:"varint,3,opt,name=bytes_total,json=bytesTotal,proto3" json:"bytes_total,omitempty"`
+	BytesTotal int64 `protobuf:"varint,3,opt,name=bytes_total,json=bytesTotal,proto3" json:"bytes_total,omitempty"`
+	// phase is what the bytes are: empty while they arrive, "checking" while the
+	// node checks a set's files against the mirror's checksums.
+	Phase         string `protobuf:"bytes,4,opt,name=phase,proto3" json:"phase,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -3642,6 +3645,13 @@ func (x *FetchProgress) GetBytesTotal() int64 {
 	return 0
 }
 
+func (x *FetchProgress) GetPhase() string {
+	if x != nil {
+		return x.Phase
+	}
+	return ""
+}
+
 // StartShard launches one shard. It is idempotent by shard_id: the control
 // plane retries, and a repeat must not produce a second process.
 type StartShard struct {
@@ -5515,13 +5525,14 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\vSTATE_READY\x10\x04\x12\x12\n" +
 	"\x0eSTATE_DRAINING\x10\x05\x12\x11\n" +
 	"\rSTATE_STOPPED\x10\x06\x12\x10\n" +
-	"\fSTATE_FAILED\x10\a\"g\n" +
+	"\fSTATE_FAILED\x10\a\"}\n" +
 	"\rFetchProgress\x12\x16\n" +
 	"\x06source\x18\x01 \x01(\tR\x06source\x12\x1d\n" +
 	"\n" +
 	"bytes_done\x18\x02 \x01(\x03R\tbytesDone\x12\x1f\n" +
 	"\vbytes_total\x18\x03 \x01(\x03R\n" +
-	"bytesTotal\"\x93\x04\n" +
+	"bytesTotal\x12\x14\n" +
+	"\x05phase\x18\x04 \x01(\tR\x05phase\"\x93\x04\n" +
 	"\n" +
 	"StartShard\x12\x1d\n" +
 	"\n" +

api/proto/llmgw/v1/agent.proto

@@ -705,6 +705,9 @@ message FetchProgress {
   // is a real answer: a percentage would then be invented, and an invented
   // percentage that stops moving is worse than a byte count that keeps rising.
   int64 bytes_total = 3;
+  // phase is what the bytes are: empty while they arrive, "checking" while the
+  // node checks a set's files against the mirror's checksums.
+  string phase = 4;
 }
 
 // StartShard launches one shard. It is idempotent by shard_id: the control

internal/agent/agent.go

@@ -1132,7 +1132,7 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan<
 		send(&llmgwv1.ShardStatus{
 			State: agentsvc.StateToWire(supervisor.Starting),
 			Fetch: &llmgwv1.FetchProgress{
-				Source: p.Source, BytesDone: p.BytesDone, BytesTotal: p.BytesTotal,
+				Source: p.Source, BytesDone: p.BytesDone, BytesTotal: p.BytesTotal, Phase: p.Phase,
 			},
 		})
 	})
@@ -1155,7 +1155,8 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan<
 		for i, m := range ms {
 			stop := a.watchFetch(m["source"], func(p weights.Progress) {
 				send(&llmgwv1.ShardStatus{State: agentsvc.StateToWire(supervisor.Starting),
-					Fetch: &llmgwv1.FetchProgress{Source: p.Source, BytesDone: p.BytesDone, BytesTotal: p.BytesTotal}})
+					Fetch: &llmgwv1.FetchProgress{Source: p.Source, BytesDone: p.BytesDone, BytesTotal: p.BytesTotal,
+						Phase: p.Phase}})
 			})
 			path, err := a.weights().Ensure(ctx, m["source"], m["allow_pickle"] == "true")
 			stop()

internal/agent/agent_test.go

@@ -68,7 +68,7 @@ func (r *statusRecorder) OnShardCompute(replicaID, shardID string, c reconciler.
 	r.computes = append(r.computes, c)
 }
 
-func (r *statusRecorder) OnFetchProgress(replicaID, shardID, source string, done, total int64) {
+func (r *statusRecorder) OnFetchProgress(replicaID, shardID, source, phase string, done, total int64) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.fetches = append(r.fetches, fetchReport{replicaID, shardID, source, done, total})

internal/agent/weights/fetch.go

@@ -38,8 +38,14 @@ type Progress struct {
 	Source     string
 	BytesDone  int64
 	BytesTotal int64
+	// Phase is empty while the bytes arrive, PhaseChecking while a set's
+	// files are checked against the mirror's checksums.
+	Phase string
 }
 
+// PhaseChecking is a set's files being checked, after they arrived.
+const PhaseChecking = "checking"
+
 // Percent is how far along, or -1 when the total is unknown.
 func (p Progress) Percent() int {
 	if p.BytesTotal <= 0 {

internal/agent/weights/set.go

@@ -5,12 +5,16 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
 )
 
 // IsSet reports whether a source names a set of files rather than one.
@@ -120,20 +124,8 @@ func (f *Fetcher) ensureSet(ctx context.Context, source string, allowPickle bool
 	// Each file against the checksum the mirror took when it fetched it,
 	// before the set is called complete: a shard damaged on the way loads as
 	// a model that answers nonsense, not as an error.
-	for _, file := range m.Files {
-		if file.SHA256 == "" {
-			continue
-		}
-		target := filepath.Join(dir, file.Name)
-		got, err := fileSHA256(target)
-		if err != nil {
-			return "", err
-		}
-		if !strings.EqualFold(got, file.SHA256) {
-			_ = os.Remove(target)
-			return "", fmt.Errorf("%s%s: sha256 %s, the mirror has %s; removed, to be fetched again",
-				source, file.Name, got, file.SHA256)
-		}
+	if err := f.checkSet(ctx, source, dir, m); err != nil {
+		return "", err
 	}
 
 	if err := os.WriteFile(done, []byte(source+"\n"), 0o644); err != nil {
@@ -195,3 +187,101 @@ func fileSHA256(path string) (string, error) {
 	}
 	return hex.EncodeToString(h.Sum(nil)), nil
 }
+
+// checkSetAtOnce is how many of a set's files are checked at once: a set of
+// six 50 GB files took eight minutes on one core, with nothing said.
+const checkSetAtOnce = 4
+
+// checkSet checks each file of a set against the checksum the mirror took
+// when it fetched it, several at once, saying how far it has got.
+func (f *Fetcher) checkSet(ctx context.Context, source, dir string, m manifest) error {
+	var total int64
+	for _, file := range m.Files {
+		if file.SHA256 != "" {
+			total += file.Bytes
+		}
+	}
+	if total == 0 {
+		return nil // nothing the mirror took a checksum of
+	}
+	var done atomic.Int64
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		t := time.NewTicker(2 * time.Second)
+		defer t.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-t.C:
+				f.report(Progress{Source: source, BytesDone: done.Load(), BytesTotal: total, Phase: PhaseChecking})
+			}
+		}
+	}()
+	f.report(Progress{Source: source, BytesTotal: total, Phase: PhaseChecking})
+
+	var (
+		mu    sync.Mutex
+		first error
+		wg    sync.WaitGroup
+	)
+	slots := make(chan struct{}, checkSetAtOnce)
+	for _, file := range m.Files {
+		if file.SHA256 == "" {
+			continue
+		}
+		wg.Add(1)
+		slots <- struct{}{}
+		go func() {
+			defer wg.Done()
+			defer func() { <-slots }()
+			target := filepath.Join(dir, file.Name)
+			got, err := countedSHA256(ctx, target, &done)
+			if err == nil && !strings.EqualFold(got, file.SHA256) {
+				_ = os.Remove(target)
+				err = fmt.Errorf("%s%s: sha256 %s, the mirror has %s; removed, to be fetched again",
+					source, file.Name, got, file.SHA256)
+			}
+			if err != nil {
+				mu.Lock()
+				if first == nil {
+					first = err
+				}
+				mu.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+	if first == nil {
+		f.report(Progress{Source: source, BytesDone: total, BytesTotal: total, Phase: PhaseChecking})
+	}
+	return first
+}
+
+// countedSHA256 is a file's checksum, adding what it has read to done.
+func countedSHA256(ctx context.Context, path string, done *atomic.Int64) (string, error) {
+	file, err := os.Open(path) //nolint:gosec // a file this fetcher wrote
+	if err != nil {
+		return "", err
+	}
+	defer file.Close()
+	h := sha256.New()
+	buf := make([]byte, 4<<20)
+	for {
+		if err := ctx.Err(); err != nil {
+			return "", err
+		}
+		n, err := file.Read(buf)
+		if n > 0 {
+			h.Write(buf[:n])
+			done.Add(int64(n))
+		}
+		if errors.Is(err, io.EOF) {
+			return hex.EncodeToString(h.Sum(nil)), nil
+		}
+		if err != nil {
+			return "", err
+		}
+	}
+}

internal/agent/weights/set_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -234,3 +235,38 @@ func TestASetsProgressIsTheWholeSets(t *testing.T) {
 		t.Errorf("ended at %d", last)
 	}
 }
+
+// A set's files checked after they arrive say so, as far as they have got,
+// rather than leaving the download at 100% for minutes.
+func TestASetBeingCheckedSaysSo(t *testing.T) {
+	parts := map[string]string{"a.gguf": strings.Repeat("a", 300), "b.gguf": strings.Repeat("b", 200)}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/") {
+			var files []string
+			for n, body := range parts {
+				sum := sha256.Sum256([]byte(body))
+				files = append(files, fmt.Sprintf(`{"name":%q,"bytes":%d,"sha256":%q}`, n, len(body), hex.EncodeToString(sum[:])))
+			}
+			fmt.Fprintf(w, `{"model":"m","version":"v","files":[%s]}`, strings.Join(files, ","))
+			return
+		}
+		_, _ = w.Write([]byte(parts[strings.TrimPrefix(r.URL.Path, "/m/v/")]))
+	}))
+	t.Cleanup(srv.Close)
+	var mu sync.Mutex
+	var checking []Progress
+	f := &Fetcher{CacheDir: t.TempDir(), OnProgress: func(p Progress) {
+		if p.Phase == PhaseChecking {
+			mu.Lock()
+			checking = append(checking, p)
+			mu.Unlock()
+		}
+	}}
+	if _, err := f.Ensure(t.Context(), srv.URL+"/m/v/", false); err != nil {
+		t.Fatal(err)
+	}
+	if len(checking) < 2 || checking[0].BytesDone != 0 || checking[len(checking)-1].BytesDone != 500 ||
+		checking[len(checking)-1].BytesTotal != 500 {
+		t.Errorf("%+v", checking)
+	}
+}

internal/cli/ctl.go

@@ -499,8 +499,12 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
 					if s.Fetch == nil {
 						continue
 					}
-					fmt.Fprintf(w, "  %s: fetching %s\t\t\t\t\t\t\t\t\n",
-						s.Node, fetchColumn(s.Fetch))
+					what := "fetching"
+					if s.Fetch.Phase != "" {
+						what = s.Fetch.Phase
+					}
+					fmt.Fprintf(w, "  %s: %s %s\t\t\t\t\t\t\t\t\n",
+						s.Node, what, fetchColumn(s.Fetch))
 				}
 			}
 			_ = w.Flush()

internal/control/adminapi/api.go

@@ -608,6 +608,9 @@ type FetchView struct {
 	// invented percentage that stops moving is worse than a byte count that
 	// keeps rising.
 	Percent *int `json:"percent,omitempty"`
+	// Phase is empty while the bytes arrive, "checking" while a set's files
+	// are checked against the mirror's checksums.
+	Phase string `json:"phase,omitempty"`
 }
 
 // ExplainView is why a model could not be placed.
@@ -922,7 +925,7 @@ func (a *API) Replicas(model string) []ReplicaView {
 			}
 			if f := s.Fetch; f != nil {
 				sv.Fetch = &FetchView{
-					Source: f.Source, BytesDone: f.BytesDone, BytesTotal: f.BytesTotal,
+					Source: f.Source, BytesDone: f.BytesDone, BytesTotal: f.BytesTotal, Phase: f.Phase,
 				}
 				if p, ok := f.Percent(); ok {
 					sv.Fetch.Percent = &p

internal/control/agentsvc/service.go

@@ -35,7 +35,7 @@ type StatusSink interface {
 	// OnFetchProgress is separate because it is not a transition: the shard is
 	// still starting, and treating each report as a state change would reset
 	// the timers that decide when a slow start has become a failed one.
-	OnFetchProgress(replicaID, shardID, source string, done, total int64)
+	OnFetchProgress(replicaID, shardID, source, phase string, done, total int64)
 	// OnShardCompute records where a shard's engine said it runs the model.
 	OnShardCompute(replicaID, shardID string, c reconciler.Compute)
 	Adopt(nodeID string, shards []reconciler.AdoptedShard) (unclaimed []reconciler.AdoptedShard)
@@ -461,7 +461,7 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
 		// download that takes an hour would keep a start timer at zero.
 		if f := m.Shard.GetFetch(); f != nil {
 			s.status.OnFetchProgress(m.Shard.GetReplicaId(), m.Shard.GetShardId(),
-				f.GetSource(), f.GetBytesDone(), f.GetBytesTotal())
+				f.GetSource(), f.GetPhase(), f.GetBytesDone(), f.GetBytesTotal())
 			return
 		}
 		if c := m.Shard.GetCompute(); c != nil {

internal/control/reconciler/fetch_test.go

@@ -23,7 +23,7 @@ func TestProgressIsVisibleWithoutBeingATransition(t *testing.T) {
 	}
 	rep, shard := reps[0].ID, reps[0].Shards[0].ID
 
-	h.r.OnFetchProgress(rep, shard, "hf:org/repo/model.gguf", 3<<30, 12<<30)
+	h.r.OnFetchProgress(rep, shard, "hf:org/repo/model.gguf", "", 3<<30, 12<<30)
 
 	got := h.r.Replicas()[0].Shards[0]
 	if got.Fetch == nil {
@@ -54,7 +54,7 @@ func TestATransitionClearsTheDownload(t *testing.T) {
 	reps := h.r.Replicas()
 	rep, shard := reps[0].ID, reps[0].Shards[0].ID
 
-	h.r.OnFetchProgress(rep, shard, "hf:org/repo/model.gguf", 1, 2)
+	h.r.OnFetchProgress(rep, shard, "hf:org/repo/model.gguf", "", 1, 2)
 	h.r.OnShardStatus(rep, shard, supervisor.Ready, "")
 
 	if got := h.r.Replicas()[0].Shards[0]; got.Fetch != nil {
@@ -66,5 +66,5 @@ func TestProgressForAReplicaThatIsGoneIsIgnored(t *testing.T) {
 	// A download that finishes after its replica was torn down should not
 	// resurrect anything or panic.
 	h := newHarness(t, testModel(1), gpuNode("n1", 8))
-	h.r.OnFetchProgress("no-such-replica", "no-such-shard", "x", 1, 2)
+	h.r.OnFetchProgress("no-such-replica", "no-such-shard", "x", "", 1, 2)
 }

internal/control/reconciler/reconciler.go

@@ -1320,7 +1320,7 @@ func (r *Reconciler) stopReplica(ctx context.Context, id, why string) {
 // Separate from OnShardStatus because it is not a transition: the shard is
 // still starting, and treating each progress report as a state change would
 // reset the timers that decide when a slow start becomes a failed one.
-func (r *Reconciler) OnFetchProgress(replicaID, shardID, source string, done, total int64) {
+func (r *Reconciler) OnFetchProgress(replicaID, shardID, source, phase string, done, total int64) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -1332,11 +1332,13 @@ func (r *Reconciler) OnFetchProgress(replicaID, shardID, source string, done, to
 		if rep.Shards[i].ID != shardID {
 			continue
 		}
-		if prev := rep.Shards[i].Fetch; prev == nil || done > prev.BytesDone {
+		// Moving: more bytes, or on to the next phase, whose count starts
+		// again from nothing — the files checked after they arrived.
+		if prev := rep.Shards[i].Fetch; prev == nil || done > prev.BytesDone || phase != prev.Phase {
 			rep.fetchedAt = r.now()
 		}
 		rep.Shards[i].Fetch = &FetchProgress{
-			Source: source, BytesDone: done, BytesTotal: total,
+			Source: source, BytesDone: done, BytesTotal: total, Phase: phase,
 		}
 	}
 }

internal/control/reconciler/replica.go

@@ -101,6 +101,9 @@ type FetchProgress struct {
 	// invented, and an invented one that stops moving is worse than a byte
 	// count that keeps rising.
 	BytesTotal int64
+	// Phase is empty while the bytes arrive, "checking" while they are
+	// checked against the mirror's checksums.
+	Phase string
 }
 
 // Percent returns progress as a percentage, and false when the total is

internal/control/reconciler/stuck_test.go

@@ -422,7 +422,7 @@ func TestAReplicaStillDownloadingIsNotFailed(t *testing.T) {
 	// Forty minutes of download, moving all the while.
 	for i := int64(1); i <= 8; i++ {
 		h.advance(5 * time.Minute)
-		h.r.OnFetchProgress(rep.ID, shard, "mirror", i<<35, 400<<30)
+		h.r.OnFetchProgress(rep.ID, shard, "mirror", "", i<<35, 400<<30)
 		h.r.expireStuckReplicas()
 	}
 	if got := h.r.Replicas()[0].State; got == Failed {
@@ -430,7 +430,7 @@ func TestAReplicaStillDownloadingIsNotFailed(t *testing.T) {
 	}
 	// The same figure again is not progress: stalled, it fails a timeout on.
 	h.advance(10 * time.Minute)
-	h.r.OnFetchProgress(rep.ID, shard, "mirror", 8<<35, 400<<30)
+	h.r.OnFetchProgress(rep.ID, shard, "mirror", "", 8<<35, 400<<30)
 	h.advance(11 * time.Minute)
 	h.r.expireStuckReplicas()
 	if got := h.r.Replicas()[0].State; got != Failed {

internal/router/chat_load.go

@@ -49,10 +49,13 @@ func (r *Router) loadStatus(model string) modelStatus {
 					continue
 				}
 				st.State = "downloading"
+				if sh.Fetch.Phase == "checking" {
+					st.State = "checking"
+				}
 				st.Done += sh.Fetch.BytesDone
 				st.Total += sh.Fetch.BytesTotal
 			}
-			if st.State == "downloading" && st.Total > 0 {
+			if (st.State == "downloading" || st.State == "checking") && st.Total > 0 {
 				p := int(st.Done * 100 / st.Total)
 				st.Percent = &p
 			}

internal/web/locales/en/chat.json

@@ -23,6 +23,7 @@
   "chat.js.download_failed": "download failed",
   "chat.js.stage.waiting": "Waiting for a node with room",
   "chat.js.stage.downloading": "Downloading weights",
+  "chat.js.stage.checking": "Checking the weights",
   "chat.js.stage.starting": "Starting the engine",
   "chat.js.stage.ready": "Ready",
   "chat.js.of": "{done} of {total}",

internal/web/locales/tr/chat.json

@@ -23,6 +23,7 @@
   "chat.js.download_failed": "indirme başarısız oldu",
   "chat.js.stage.waiting": "Yeri olan bir düğüm bekleniyor",
   "chat.js.stage.downloading": "Ağırlıklar indiriliyor",
+  "chat.js.stage.checking": "Ağırlıklar denetleniyor",
   "chat.js.stage.starting": "Motor başlatılıyor",
   "chat.js.stage.ready": "Hazır",
   "chat.js.of": "{done} / {total}",

internal/web/templates/chat.html

@@ -82,6 +82,7 @@
     downloadFailed: {{t "chat.js.download_failed"}},
     stageWaiting: {{t "chat.js.stage.waiting"}},
     stageDownloading: {{t "chat.js.stage.downloading"}},
+    stageChecking: {{t "chat.js.stage.checking"}},
     stageStarting: {{t "chat.js.stage.starting"}},
     stageReady: {{t "chat.js.stage.ready"}},
     of: {{t "chat.js.of"}},
@@ -288,6 +289,7 @@
   const stages = {
     waiting: say.stageWaiting,
     downloading: say.stageDownloading,
+    checking: say.stageChecking,
     starting: say.stageStarting,
     ready: say.stageReady,
   };
@@ -322,7 +324,7 @@
       for (;;) {
         const secs = Math.round((Date.now() - started) / 1000);
         let text = (stages[st.state] || st.state) + " — " + fill(say.seconds, { n: secs });
-        if (st.state === "downloading" && st.total) text += " · " + fill(say.of, { done: mib(st.done), total: mib(st.total) });
+        if ((st.state === "downloading" || st.state === "checking") && st.total) text += " · " + fill(say.of, { done: mib(st.done), total: mib(st.total) });
         label.textContent = text;
         // Indeterminate (no value) unless there is a real number: a bar that
         // sits at 0% for the engine start says it is stuck when it is not.