A set wanted by several replicas on one node is fetched once

Single files were shared between the shards and replicas asking for them
at the same time; sets were not, and four replicas on one node's four cards
wrote the same .part files over one another until all but one failed.

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

CHANGELOG.md

@@ -80,6 +80,12 @@ APIs change between minors without a deprecation period.
 
 ### Fixed
 
+- **Several replicas of a multi-file model on one node fetched it once
+  each, over one another.** Four replicas placed on a node's four cards
+  wanted the same safetensors set at the same moment, each wrote the same
+  `.part` files, and all but one failed to start ("rename … no such file")
+  until a retry found the set finished. A set is now fetched once per node
+  and every replica waiting on it gets it, as single files always were.
 - **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.

internal/agent/weights/fetch.go

@@ -146,7 +146,11 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w
 	// file, and a set has to be handled before any of it.
 	if IsSet(source) {
 		outcome(ctx, "set")
-		return f.ensureSet(ctx, source, allowPickle)
+		// Shared like a single file: four replicas placed on one node's
+		// four cards want the same set at the same instant, and each
+		// fetching it wrote the same .part files over one another — three
+		// failed renaming a file another had already renamed.
+		return f.shared(ctx, source, func() (string, error) { return f.ensureSet(ctx, source, allowPickle) })
 	}
 	if !isRemote(source) {
 		if !isPath(source) {
@@ -178,6 +182,15 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w
 		// Not the file the catalog names: fetched again below, and checked.
 	}
 
+	return f.shared(ctx, source, func() (string, error) {
+		outcome(ctx, "downloaded")
+		return f.fetchChecked(ctx, source, target, want)
+	})
+}
+
+// shared runs fetch once for every caller asking for source at the same
+// time; the others wait for its answer.
+func (f *Fetcher) shared(ctx context.Context, source string, fetch func() (string, error)) (string, error) {
 	f.mu.Lock()
 	if f.inflight == nil {
 		f.inflight = map[string]*fetchCall{}
@@ -196,8 +209,7 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w
 	f.inflight[source] = call
 	f.mu.Unlock()
 
-	outcome(ctx, "downloaded")
-	call.path, call.err = f.fetchChecked(ctx, source, target, want)
+	call.path, call.err = fetch()
 	close(call.done)
 
 	f.mu.Lock()

internal/agent/weights/set_test.go

@@ -10,7 +10,9 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 )
 
 // mirrorOf serves a manifest at "/m/v/" and the files beneath it.
@@ -270,3 +272,38 @@ func TestASetBeingCheckedSaysSo(t *testing.T) {
 		t.Errorf("%+v", checking)
 	}
 }
+
+// Four replicas on one node's four cards start together and want the same
+// set: it is fetched once, and every one of them gets it. Each fetching it
+// alone wrote the same .part files over one another, and three failed.
+func TestASetWantedByFourReplicasAtOnceIsFetchedOnce(t *testing.T) {
+	body := strings.Repeat("a", 1<<20)
+	var hits atomic.Int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/") {
+			fmt.Fprintf(w, `{"model":"m","version":"v","files":[{"name":"a.safetensors","bytes":%d},{"name":"b.safetensors","bytes":%d}]}`,
+				len(body), len(body))
+			return
+		}
+		hits.Add(1)
+		time.Sleep(100 * time.Millisecond) // long enough for all four to ask
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(srv.Close)
+	f := &Fetcher{CacheDir: t.TempDir()}
+
+	var wg sync.WaitGroup
+	errs := make([]error, 4)
+	for i := range errs {
+		wg.Go(func() { _, errs[i] = f.Ensure(t.Context(), srv.URL+"/m/v/", false) })
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err != nil {
+			t.Errorf("replica %d: %v", i, err)
+		}
+	}
+	if n := hits.Load(); n != 2 {
+		t.Errorf("%d downloads of a set of two files", n)
+	}
+}