diff --git a/CHANGELOG.md b/CHANGELOG.md index a64f204..2b62626 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,14 @@ is only released once its **exit criterion** is met, not when its code lands. Until v1.0.0 the minor version is the breaking one: configuration formats and Go APIs change between minors without a deprecation period. +## Unreleased + +### Fixed + +- **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. + ## v0.16.1 — 2026-10-30 The mirror page's Strata tab, Strata's ready-made packs served with Strata fetching them, and two fixes found diff --git a/internal/agent/weights/fetch.go b/internal/agent/weights/fetch.go index 7246ce7..bc99793 100644 --- a/internal/agent/weights/fetch.go +++ b/internal/agent/weights/fetch.go @@ -202,7 +202,7 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w // fetch downloads a source to a target, resuming and verifying. func (f *Fetcher) fetch(ctx context.Context, source, target string) (string, error) { - return f.fetchAs(ctx, source, source, target) + return f.fetchAs(ctx, source, source, target, setShare{}) } // fetchChecked is fetch for a file whose sha256 is known: a download that @@ -287,7 +287,7 @@ func ChecksumPath(file string) string { return checksumPath(file) } // // Each file is a "download" span, with how many bytes it took and where a // resumed one started from. -func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) (_ string, err error) { +func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string, share setShare) (_ string, err error) { ctx, span := obs.Tracer(tracerName).Start(ctx, "download", trace.WithSpanKind(trace.SpanKindClient), trace.WithAttributes(attribute.String("llmgw.weights.source", redact(source)))) @@ -351,8 +351,8 @@ func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) } total := existing + resp.ContentLength - f.report(Progress{Source: reportAs, BytesDone: existing, BytesTotal: total}) - written, err := f.copyWithProgress(ctx, out, resp.Body, reportAs, existing, total) + f.report(share.of(Progress{Source: reportAs, BytesDone: existing, BytesTotal: total})) + written, err := f.copyWithProgress(ctx, out, resp.Body, reportAs, existing, total, share) span.SetAttributes( attribute.Int64("llmgw.weights.resumed_from", existing), attribute.Int64("llmgw.weights.bytes", written), @@ -376,7 +376,7 @@ func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) } func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Reader, - source string, done, total int64, + source string, done, total int64, share setShare, ) (int64, error) { buf := make([]byte, 4<<20) last := time.Now() @@ -394,12 +394,12 @@ func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Re // hundred gigabytes is twenty-five thousand callbacks, and nobody // needs to be told that often. if time.Since(last) > 2*time.Second { - f.report(Progress{Source: source, BytesDone: done, BytesTotal: total}) + f.report(share.of(Progress{Source: source, BytesDone: done, BytesTotal: total})) last = time.Now() } } if errors.Is(err, io.EOF) { - f.report(Progress{Source: source, BytesDone: done, BytesTotal: total}) + f.report(share.of(Progress{Source: source, BytesDone: done, BytesTotal: total})) return done, nil } if err != nil { @@ -408,6 +408,18 @@ func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Re } } +// setShare places one file's download within its set's: the bytes of the +// files before it, and the whole set's. Zero for a file on its own. +type setShare struct{ before, total int64 } + +// of is a file's progress as the set's. +func (s setShare) of(p Progress) Progress { + if s.total > 0 { + p.BytesDone, p.BytesTotal = s.before+p.BytesDone, s.total + } + return p +} + func (f *Fetcher) report(p Progress) { if f.OnProgress != nil { f.OnProgress(p) diff --git a/internal/agent/weights/set.go b/internal/agent/weights/set.go index 6393acd..31641b0 100644 --- a/internal/agent/weights/set.go +++ b/internal/agent/weights/set.go @@ -89,6 +89,12 @@ func (f *Fetcher) ensureSet(ctx context.Context, source string, allowPickle bool if err := os.MkdirAll(dir, 0o755); err != nil { return "", err } + // One number for the whole set, as the shards waiting on it see it: the + // file being fetched was shown alone, a 46 GB part of 188. + var setTotal, before int64 + for _, file := range m.Files { + setTotal += file.Bytes + } for _, file := range m.Files { // Rejected rather than joined: a name with a separator in it would // write outside the directory the mirror said it was in, and the @@ -98,15 +104,17 @@ func (f *Fetcher) ensureSet(ctx context.Context, source string, allowPickle bool } target := filepath.Join(dir, file.Name) if st, err := os.Stat(target); err == nil && st.Size() == file.Bytes { + before += file.Bytes continue } // Reported under the set's own address, because that is what the // shards are waiting on. Reusing fetch rather than writing a second // downloader keeps resume, the .part rename and the progress callback // in one place. - if _, err := f.fetchAs(ctx, source+file.Name, source, target); err != nil { + if _, err := f.fetchAs(ctx, source+file.Name, source, target, setShare{before: before, total: setTotal}); err != nil { return "", err } + before += file.Bytes } // Each file against the checksum the mirror took when it fetched it, diff --git a/internal/agent/weights/set_test.go b/internal/agent/weights/set_test.go index f32e052..2b5b778 100644 --- a/internal/agent/weights/set_test.go +++ b/internal/agent/weights/set_test.go @@ -204,3 +204,33 @@ func TestASetStillDownloadingOnTheMirrorSaysSo(t *testing.T) { t.Fatalf("got %v", err) } } + +// A set's download is one number climbing against the whole set, not each +// part's own: a 46 GB part of 188 was shown as the download. +func TestASetsProgressIsTheWholeSets(t *testing.T) { + f, srv, _ := mirrorOf(t, map[string]string{ + "m-00001-of-00002.gguf": strings.Repeat("a", 600), + "m-00002-of-00002.gguf": strings.Repeat("b", 400), + }) + var seen []Progress + f.OnProgress = func(p Progress) { seen = append(seen, p) } + if _, err := f.Ensure(t.Context(), srv.URL+"/m/v/", false); err != nil { + t.Fatal(err) + } + if len(seen) == 0 { + t.Fatal("no progress reported") + } + last := int64(-1) + for _, p := range seen { + if p.BytesTotal != 1000 { + t.Fatalf("reported against %d, not the set's 1000: %+v", p.BytesTotal, seen) + } + if p.BytesDone < last { + t.Fatalf("went back: %+v", seen) + } + last = p.BytesDone + } + if last != 1000 { + t.Errorf("ended at %d", last) + } +}