From be54f1617b2bec02b33e64dfe03a3c341ddf09eb Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Fri, 30 Oct 2026 17:47:32 +0100 Subject: [PATCH] A mirror download cut off carries on from where it got to, and is tried again Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 + internal/mirror/add.go | 161 +++++++++++++++++++++++++++------ internal/mirror/resume_test.go | 94 +++++++++++++++++++ 3 files changed, 229 insertions(+), 30 deletions(-) create mode 100644 internal/mirror/resume_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b0bd058..1af98d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,10 @@ APIs change between minors without a deprecation period. 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. +- **A mirror download the registry cut off started again from nothing, or not at all.** Hugging Face's CDN ended a + 60 GB transfer at 23 GB (`stream error … CANCEL`), the mirror deleted what had arrived, and `ctl model serve` + failed. The mirror now tries again up to six times, carrying on from where it got to (`Range`), keeps what arrived + when it gives up, and carries on from it after a restart too. ## v0.16.1 — 2026-10-30 diff --git a/internal/mirror/add.go b/internal/mirror/add.go index a5d4fb5..688386f 100644 --- a/internal/mirror/add.go +++ b/internal/mirror/add.go @@ -2,11 +2,15 @@ package mirror import ( "context" + "errors" "fmt" "io" + "net" "net/http" "os" "path/filepath" + "strings" + "syscall" "time" ) @@ -54,40 +58,11 @@ func addWith(ctx context.Context, client *http.Client, root, source string, e En return Added{}, err } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, source, nil) - if err != nil { - return Added{}, err - } - resp, err := client.Do(req) - if err != nil { - return Added{}, fmt.Errorf("fetching %s: %w", source, err) - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return Added{}, fmt.Errorf("%s answered %s", source, resp.Status) - } - part := final + partSuffix - f, err := os.OpenFile(part, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644) //nolint:gosec // inside the mirror root + written, err := downloadResuming(ctx, client, source, part, progress) if err != nil { return Added{}, err } - written, err := copyWithProgress(f, resp.Body, resp.ContentLength, progress) - closeErr := f.Close() - if err != nil { - _ = os.Remove(part) - return Added{}, err - } - if closeErr != nil { - _ = os.Remove(part) - return Added{}, closeErr - } - // A truncated download that the server ended cleanly would otherwise be - // renamed into place and copied to every node. - if resp.ContentLength > 0 && written != resp.ContentLength { - _ = os.Remove(part) - return Added{}, fmt.Errorf("%s: got %d bytes of %d", source, written, resp.ContentLength) - } sum, err := sha256File(part) if err != nil { @@ -112,6 +87,132 @@ func addWith(ctx context.Context, client *http.Client, root, source string, e En return Added{Entry: e, SHA256: sum}, nil } +// downloadAttempts is how many times a download is tried before it fails: +// a registry's CDN ends a long transfer now and then ("stream error … +// CANCEL"), and a 60 GB file failed at 23 GB and started again from nothing. +var downloadAttempts = 6 + +// downloadPause is the wait before the next attempt, doubled each time. +var downloadPause = 2 * time.Second + +// downloadResuming fetches source into part, carrying on from what part +// already holds — a download cut short, or one a restarted mirror began — +// and trying again, from where it got to, when the transfer breaks. The part +// is kept when it fails, for the next fetch of the same file to carry on. +func downloadResuming(ctx context.Context, client *http.Client, source, part string, + progress func(done, total int64), +) (int64, error) { + var have int64 + if st, err := os.Stat(part); err == nil { + have = st.Size() + } + pause := downloadPause + var last error + for attempt := 0; attempt < downloadAttempts; attempt++ { + if attempt > 0 { + select { + case <-ctx.Done(): + return have, ctx.Err() + case <-time.After(pause): + } + pause = min(pause*2, 30*time.Second) + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, source, nil) + if err != nil { + return have, err + } + if have > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", have)) + } + resp, err := client.Do(req) + if err != nil { + if ctx.Err() != nil { + return have, ctx.Err() + } + // A refused address, a name that does not resolve: asking + // again changes nothing. + if !transient(err) { + return have, fmt.Errorf("fetching %s: %w", source, err) + } + last = fmt.Errorf("fetching %s: %w", source, err) + continue + } + flags := os.O_CREATE | os.O_WRONLY + switch { + case resp.StatusCode == http.StatusPartialContent && have > 0: + flags |= os.O_APPEND + case resp.StatusCode == http.StatusOK: + have = 0 // the server ignored the range: from the start + flags |= os.O_TRUNC + case resp.StatusCode == http.StatusRequestedRangeNotSatisfiable && have > 0: + // Everything was there already. + resp.Body.Close() + return have, nil + case resp.StatusCode >= 500 || resp.StatusCode == http.StatusTooManyRequests: + resp.Body.Close() + last = fmt.Errorf("%s answered %s", source, resp.Status) + continue + default: + resp.Body.Close() + return have, fmt.Errorf("%s answered %s", source, resp.Status) + } + want := resp.ContentLength + f, err := os.OpenFile(part, flags, 0o644) //nolint:gosec // inside the mirror root + if err != nil { + resp.Body.Close() + return have, err + } + offset := have + var report func(done, total int64) + if progress != nil { + report = func(done, total int64) { + if total > 0 { + total += offset + } + progress(offset+done, total) + } + } + n, err := copyWithProgress(f, resp.Body, want, report) + resp.Body.Close() + closeErr := f.Close() + have += n + if err == nil && closeErr != nil { + return have, closeErr + } + if err != nil { + if ctx.Err() != nil { + return have, ctx.Err() + } + last = fmt.Errorf("fetching %s after %d bytes: %w", source, have, err) + continue + } + // Ended cleanly but short: the rest is asked for again. + if want > 0 && n != want { + last = fmt.Errorf("%s: got %d bytes of %d", source, n, want) + continue + } + return have, nil + } + return have, fmt.Errorf("%w (tried %d times; what arrived is kept and the next fetch carries on from it)", + last, downloadAttempts) +} + +// transient says whether a failure to reach a server may pass if asked +// again: a timeout, a connection the far end reset or ended, an HTTP/2 +// stream it cancelled. +func transient(err error) bool { + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + return true + } + if errors.Is(err, syscall.ECONNRESET) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) { + return true + } + msg := err.Error() + return strings.Contains(msg, "stream error") || strings.Contains(msg, "connection reset") || + strings.Contains(msg, "GOAWAY") +} + // copyWithProgress copies and reports occasionally. // // Occasionally rather than continuously: a forty gigabyte download is a diff --git a/internal/mirror/resume_test.go b/internal/mirror/resume_test.go new file mode 100644 index 0000000..1e701b9 --- /dev/null +++ b/internal/mirror/resume_test.go @@ -0,0 +1,94 @@ +package mirror + +import ( + "fmt" + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" +) + +// cutServer serves body, honouring Range, and cuts the first transfer off +// halfway, as a registry's CDN does now and then. +func cutServer(t *testing.T, body string) (*httptest.Server, *atomic.Int32, *atomic.Value) { + t.Helper() + var calls atomic.Int32 + var ranges atomic.Value + ranges.Store("") + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := calls.Add(1) + from := 0 + if rg := r.Header.Get("Range"); rg != "" { + ranges.Store(rg) + from, _ = strconv.Atoi(strings.TrimSuffix(strings.TrimPrefix(rg, "bytes="), "-")) + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", from, len(body)-1, len(body))) + w.Header().Set("Content-Length", strconv.Itoa(len(body)-from)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + } + if n == 1 { + _, _ = w.Write([]byte(body[from : from+len(body)/2])) + w.(http.Flusher).Flush() + hj, _ := w.(http.Hijacker) + conn, _, _ := hj.Hijack() + _ = conn.Close() + return + } + _, _ = w.Write([]byte(body[from:])) + })) + t.Cleanup(srv.Close) + return srv, &calls, &ranges +} + +func TestADownloadCutOffCarriesOnFromWhereItGot(t *testing.T) { + old := downloadPause + downloadPause = 10 * time.Millisecond + t.Cleanup(func() { downloadPause = old }) + body := strings.Repeat("weights ", 1000) + srv, calls, ranges := cutServer(t, body) + root := t.TempDir() + e := Entry{Model: "m", Version: "v", File: "w.gguf"} + added, err := addWith(t.Context(), http.DefaultClient, root, srv.URL+"/w.gguf", e, nil) + if err != nil { + t.Fatal(err) + } + got, _ := os.ReadFile(e.Path(root)) + if string(got) != body || added.Entry.Bytes != int64(len(body)) { + t.Fatalf("got %d bytes", len(got)) + } + if calls.Load() != 2 || ranges.Load().(string) != fmt.Sprintf("bytes=%d-", len(body)/2) { + t.Errorf("calls %d, range %q", calls.Load(), ranges.Load()) + } +} + +func TestADownloadARestartedMirrorBeganIsCarriedOn(t *testing.T) { + body := strings.Repeat("x", 4000) + var asked atomic.Value + asked.Store("") + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rg := r.Header.Get("Range") + asked.Store(rg) + from, _ := strconv.Atoi(strings.TrimSuffix(strings.TrimPrefix(rg, "bytes="), "-")) + w.Header().Set("Content-Length", strconv.Itoa(len(body)-from)) + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write([]byte(body[from:])) + })) + defer srv.Close() + root := t.TempDir() + e := Entry{Model: "m", Version: "v", File: "w.gguf"} + _ = os.MkdirAll(root+"/m/v", 0o755) + if err := os.WriteFile(e.Path(root)+partSuffix, []byte(body[:1500]), 0o644); err != nil { + t.Fatal(err) + } + if _, err := addWith(t.Context(), http.DefaultClient, root, srv.URL+"/w.gguf", e, nil); err != nil { + t.Fatal(err) + } + if got, _ := os.ReadFile(e.Path(root)); string(got) != body || asked.Load().(string) != "bytes=1500-" { + t.Errorf("%d bytes, asked %q", len(got), asked.Load()) + } +}