From 5c928741fe41104da67740c9dda3bc8d2679aefb Mon Sep 17 00:00:00 2001 From: Asher Feldman <59994+asher@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:03:46 -0700 Subject: [PATCH] fix(pull): retry stalled or dropped downloads with backoff from the .part file --- CHANGELOG.md | 6 ++ docs/cli.md | 8 +- docs/env-vars.md | 2 + docs/troubleshooting.md | 6 ++ gmlx/commands/manage.py | 108 ++++++++++++++++++-- tests/commands/test_manage.py | 184 +++++++++++++++++++++++++++++++++- 6 files changed, 299 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c13ef790..c026ecf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ adhere to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Fixed + +- A stalled or dropped read no longer abandons a `pull`: the transfer + retries with backoff from the bytes already on disk, tunable by + `GMLX_PULL_RETRIES` and `GMLX_PULL_TIMEOUT`. + ## [0.4.13] - 2026-09-12 ### Changed diff --git a/docs/cli.md b/docs/cli.md index 3b1aceb6..a29f4c38 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -637,9 +637,11 @@ gmlx pull hf:org/gemma-3-27b-GGUF/gemma-3-27b-Q4_K_M.gguf mmproj-F16.gguf Inside a `model_dirs` root, downloads nest under `__/` so that a model's siblings stay together. Before the first byte, `pull` checks that the volume has space for every shard, and it notes, without refusing, a -model that will not fit this Mac's RAM. An interrupted download resumes -from its `.part` file. Gated repositories need `HF_TOKEN` in the -environment. +model that will not fit this Mac's RAM. A stalled or dropped read retries +with backoff, resuming from the `.part` file. `GMLX_PULL_RETRIES` and +`GMLX_PULL_TIMEOUT` tune the retry logic. An interrupted `pull` also resumes +from the `.part` file on the next run. Gated repositories need `HF_TOKEN` in +the environment. ## gmlx validate diff --git a/docs/env-vars.md b/docs/env-vars.md index 4eabd331..aef48436 100644 --- a/docs/env-vars.md +++ b/docs/env-vars.md @@ -132,5 +132,7 @@ explained in [streaming.md](streaming.md) and | `GMLX_MTP_WIDTH_CAP` | Speculate only while at most this many requests decode together, with `0` uncapped. Overrides each model's `speculative_width_cap` and is read on each round. | | `GMLX_IGNORE_EOS=1` | Never stop on end-of-sequence in `serve`. Same as `--ignore-eos`, for forced-length benchmarking. | | `GMLX_API_KEY` | Client-side default key for `ps` when `--api-key` is not passed. The server reads its key only from `server.api_key`. | +| `GMLX_PULL_RETRIES` | Consecutive failed attempts `pull` accepts on one file, default `10`. An attempt that moves bytes resets the count. `0` fails on the first error. | +| `GMLX_PULL_TIMEOUT` | Socket timeout in seconds for a `pull` transfer, default `60`. It is also the budget for one stalled read. | | `HF_TOKEN`, `HUGGING_FACE_HUB_TOKEN` | Hugging Face auth for `validate` and `pull` on gated or private repos. | | `XDG_CACHE_HOME` | The root of the `gmlx/` cache directory, which holds `chat`'s prompt history, backgrounded servers' runfiles and logs, and the models `talk` downloads. | diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 8726318b..37b71359 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -62,6 +62,12 @@ never has this problem. `gmlx pull` stopped mid-download, or refused to start with `error: not enough disk space`. +A stalled or dropped read retries by itself, with backoff, from the bytes +already on disk. Raise `GMLX_PULL_RETRIES` above its default of 10 when the +remote host is having a bad day, and `GMLX_PULL_TIMEOUT` above its default +of 60 seconds when the transfer is slow but alive. A permanent failure, such +as a 404 or a full disk, still stops at once. + Interrupted downloads resume: re-run the same `pull` and it continues from where it stopped, shard by shard for sharded files. The disk-space refusal is a preflight check that names how much the file needs and how much is diff --git a/gmlx/commands/manage.py b/gmlx/commands/manage.py index 60721cec..3de506db 100644 --- a/gmlx/commands/manage.py +++ b/gmlx/commands/manage.py @@ -16,7 +16,9 @@ (under ``/__/`` for hf refs, so ``serve`` discovery / ``sync-models`` find it), or into ``--to DIR`` exactly. Several files fetch in one go - multipart GGUFs expand automatically, and extra bare filenames resolve in the first ref's -repo (an mmproj companion, a second quant). Interrupted transfers resume. +repo (an mmproj companion, a second quant). Interrupted transfers resume: a +stalled or dropped read retries with backoff from the bytes already on disk +(``GMLX_PULL_RETRIES``, ``GMLX_PULL_TIMEOUT``). ``list`` tables the local GGUFs a directory holds (the same header-only scan ``serve`` discovery uses); ``ps`` shows the models resident in a running server. @@ -24,6 +26,8 @@ from __future__ import annotations import argparse +import errno +import http.client import json import os import re @@ -33,6 +37,7 @@ import urllib.error import urllib.request +from gmlx.envflags import env_float, env_int from gmlx.textfmt import plural_s import gmlx.load.remote as remote from gmlx.load.preflight import ( @@ -504,15 +509,100 @@ def _hf_download(repo: str, filename: str, revision: str, dest_dir: str) -> str: return _url_download(url, dest_path) -def _url_download(url: str, dest_path: str) -> str: - """Stream a URL to ``dest_path``, resuming an interrupted transfer. +# Pull streams tens of GB off a CDN that stalls, resets and rate-limits, so one +# bad read must not discard the whole transfer. Each retry resumes from the +# .part file, and the socket timeout doubles as the per-read stall budget. +_PULL_TIMEOUT_S = 60.0 +_PULL_RETRIES = 10 +_BACKOFF_CAP_S = 30.0 +_RETRY_STATUS = frozenset({408, 425, 429, 500, 502, 503, 504}) +# Local filesystem failures never clear by waiting. +_FATAL_ERRNOS = frozenset({errno.ENOSPC, errno.EDQUOT, errno.EACCES, + errno.EPERM, errno.EROFS, errno.ENOENT, + errno.EISDIR, errno.EFBIG}) + + +def _retryable(e: BaseException) -> bool: + """True when a later attempt may get past ``e``.""" + if isinstance(e, urllib.error.HTTPError): # before OSError: a subclass + return e.code in _RETRY_STATUS + if isinstance(e, remote.RemoteError): + # A truncated body resumes; a stale/oversized .part needs the user. + return "connection closed early" in str(e) + if isinstance(e, http.client.HTTPException): + return True + if isinstance(e, OSError): # timeouts, resets, DNS + return e.errno not in _FATAL_ERRNOS + return False + + +def _retry_after_s(e: BaseException) -> float | None: + """``Retry-After`` seconds from a 429/503, else None (a date form falls + back to the backoff).""" + headers = getattr(e, "headers", None) + try: + return max(0.0, float(str(headers["Retry-After"]).strip())) + except (TypeError, ValueError, KeyError): + return None + + +def _retry_delay(failures: int) -> float: + return min(_BACKOFF_CAP_S, 2.0 ** failures) + + +def _pull_sleep(seconds: float) -> None: + """Backoff sleep. Module-level seam: monkeypatched in tests.""" + time.sleep(seconds) + + +def _url_download(url: str, dest_path: str, *, retries: int | None = None, + timeout: float | None = None) -> str: + """Stream a URL to ``dest_path``, retrying a stalled or dropped transfer. + + Every attempt resumes from the ``.part`` file, so a retry re-reads nothing + already on disk. ``retries`` bounds *consecutive* failures: an attempt that + moves bytes resets the count, so one download survives many separate stalls + (and, against a server that drops after every chunk, still ends -- each + attempt starts further into the file). A failure that waiting cannot fix + (an unlisted 4xx, a full disk, a stale ``.part``) raises at once. + ``GMLX_PULL_RETRIES`` and ``GMLX_PULL_TIMEOUT`` set the defaults; 0 retries + restores single-shot behaviour. Module-level seam: monkeypatched in tests.""" + if retries is None: + retries = max(0, env_int("GMLX_PULL_RETRIES", _PULL_RETRIES)) + if timeout is None: + timeout = env_float("GMLX_PULL_TIMEOUT", _PULL_TIMEOUT_S) + part = dest_path + ".part" + fname = os.path.basename(dest_path) + failures = 0 + while True: + before = os.path.getsize(part) if os.path.exists(part) else 0 + try: + return _url_download_once(url, dest_path, timeout=timeout) + except Exception as e: + after = os.path.getsize(part) if os.path.exists(part) else 0 + failures = 0 if after > before else failures + 1 + print(file=sys.stderr) # close the unterminated progress line + if retries <= 0 or failures > retries or not _retryable(e): + raise + delay = _retry_after_s(e) + if delay is None: + delay = _retry_delay(max(0, failures - 1)) + # failures == 0 means the attempt still moved bytes, so the budget + # reset and quoting it would read as "retry 0 of 10". + budget = "resuming" if not failures else f"retry {failures}/{retries}" + print(f" {fname}: {e} - {budget} in {delay:.0f}s", file=sys.stderr) + _pull_sleep(delay) + + +def _url_download_once(url: str, dest_path: str, *, timeout: float) -> str: + """One transfer attempt for :func:`_url_download`. Downloads to a sibling ``.part`` file, requesting a byte ``Range`` to continue - where a previous run left off (the server must honour it; a ``200`` answer means - it didn't, so we restart from byte 0). The ``.part`` is renamed into place only - on completion -- a failure leaves it behind so a re-run resumes rather than - restarts. A finished ``dest_path`` short-circuits (idempotent re-pull). - Module-level seam: monkeypatched in tests.""" + where a previous attempt left off (the server must honour it; a ``200`` answer + means it didn't, so we restart from byte 0). The ``.part`` is renamed into place + only on completion -- a failure leaves it behind so the next attempt resumes + rather than restarts. A finished ``dest_path`` short-circuits (idempotent + re-pull).""" if os.path.exists(dest_path): return dest_path part = dest_path + ".part" @@ -522,7 +612,7 @@ def _url_download(url: str, dest_path: str) -> str: headers["Range"] = f"bytes={have}-" req = urllib.request.Request(url, headers=headers) try: - resp = remote.http_open(req, timeout=30) + resp = remote.http_open(req, timeout=timeout) except urllib.error.HTTPError as e: if e.code == 416 and have > 0: # 416 = our .part already covers the remote range. Only an exact diff --git a/tests/commands/test_manage.py b/tests/commands/test_manage.py index c9264832..901e9808 100644 --- a/tests/commands/test_manage.py +++ b/tests/commands/test_manage.py @@ -6,6 +6,7 @@ from __future__ import annotations +import errno import json import urllib.error @@ -687,7 +688,7 @@ def fake_open(req, *, timeout): out = manage._url_download("https://example.com/m.gguf", str(dest)) assert out == str(dest) assert dest.read_bytes() == b"GGUFbytes" - assert seen["timeout"] == 30 + assert seen["timeout"] == 60 assert seen["url"] == "https://example.com/m.gguf" @@ -715,7 +716,7 @@ def read(self, n=-1): monkeypatch.setattr(remote, "http_open", lambda req, *, timeout: Drops()) with pytest.raises(OSError, match="connection dropped"): - manage._url_download("https://example.com/m.gguf", str(dest)) + manage._url_download("https://example.com/m.gguf", str(dest), retries=0) assert not dest.exists() # no half file at the final path assert (tmp_path / "m.gguf.part").read_bytes() == b"partial" # kept for resume @@ -809,11 +810,188 @@ def getheader(self, name): monkeypatch.setattr(remote, "http_open", lambda req, *, timeout: Truncated()) with pytest.raises(remote.RemoteError, match="closed early"): - manage._url_download("https://example.com/m.gguf", str(dest)) + manage._url_download("https://example.com/m.gguf", str(dest), retries=0) assert not dest.exists() # nothing promoted assert (tmp_path / "m.gguf.part").read_bytes() == b"x" * 50 # resumable +class _Attempt: + """One scripted transfer: hand back ``data``, then drop or close cleanly.""" + + def __init__(self, data, drop, have, total): + self.data, self.drop, self.have, self.total = data, drop, have, total + self.status = 206 if have else 200 + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self, n=-1): + if self.data is not None: + d, self.data = self.data, None + return d + if self.drop: + raise TimeoutError("The read operation timed out") + return b"" + + def getheader(self, name): + if name != "Content-Range": + return None + return f"bytes {self.have}-{self.total - 1}/{self.total}" + + +def _script(scripted, ranges): + """An ``http_open`` fake driving ``scripted`` [(data, drop)] attempts.""" + def fake_open(req, *, timeout): + rng = req.get_header("Range") + ranges.append(rng) + have = int(rng.split("=")[1].rstrip("-")) if rng else 0 + data, drop = scripted.pop(0) + return _Attempt(data, drop, have, 6) + return fake_open + + +def test_url_download_retries_and_resumes(tmp_path, monkeypatch): + # A mid-stream read timeout must not discard the bytes already written: the + # next attempt asks for a Range covering them and finishes the file. + dest = tmp_path / "m.gguf" + ranges, slept = [], [] + monkeypatch.setattr(remote, "http_open", + _script([(b"aabb", True), (b"cc", False)], ranges)) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + out = manage._url_download("https://example.com/m.gguf", str(dest)) + assert out == str(dest) + assert dest.read_bytes() == b"aabbcc" + assert ranges == [None, "bytes=4-"] + assert slept == [1.0] + assert not (tmp_path / "m.gguf.part").exists() + + +def test_url_download_progress_resets_failure_budget(tmp_path, monkeypatch): + # A server dropping after every chunk still finishes on a budget of 1: each + # attempt moves bytes, so the *consecutive* failure count never reaches it. + dest = tmp_path / "m.gguf" + ranges, slept = [], [] + monkeypatch.setattr( + remote, "http_open", + _script([(b"aa", True), (b"bb", True), (b"cc", False)], ranges)) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + out = manage._url_download("https://example.com/m.gguf", str(dest), + retries=1) + assert out == str(dest) + assert dest.read_bytes() == b"aabbcc" + assert ranges == [None, "bytes=2-", "bytes=4-"] + assert slept == [1.0, 1.0] # no escalation: the budget keeps resetting + + +def test_url_download_gives_up_after_retries(tmp_path, monkeypatch): + # No attempt moves a byte, so the budget runs out and the last error is the + # one the caller sees. The backoff doubles between tries. + dest = tmp_path / "m.gguf" + calls, slept = [], [] + + def fake_open(req, *, timeout): + calls.append(1) + raise TimeoutError("The read operation timed out") + + monkeypatch.setattr(remote, "http_open", fake_open) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + with pytest.raises(OSError, match="timed out"): + manage._url_download("https://example.com/m.gguf", str(dest), retries=3) + assert len(calls) == 4 # first try plus 3 retries + assert slept == [1.0, 2.0, 4.0] + + +def test_url_download_honours_retry_after(tmp_path, monkeypatch): + # A rate-limited host names the wait it wants; the backoff defers to it. + dest = tmp_path / "m.gguf" + slept, ranges = [], [] + inner = _script([(b"aabbcc", False)], ranges) + first = [True] + + def fake_open(req, *, timeout): + if first[0]: + first[0] = False + raise urllib.error.HTTPError( + req.full_url, 429, "Too Many Requests", + {"Retry-After": "7"}, None) + return inner(req, timeout=timeout) + + monkeypatch.setattr(remote, "http_open", fake_open) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + assert manage._url_download("https://example.com/m.gguf", + str(dest)) == str(dest) + assert slept == [7.0] + assert dest.read_bytes() == b"aabbcc" + + +def test_url_download_does_not_retry_client_error(tmp_path, monkeypatch): + # A 404 is permanent; retrying only spends the user's time. + dest = tmp_path / "m.gguf" + calls, slept = [], [] + + def fake_open(req, *, timeout): + calls.append(1) + raise urllib.error.HTTPError(req.full_url, 404, "Not Found", {}, None) + + monkeypatch.setattr(remote, "http_open", fake_open) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + with pytest.raises(urllib.error.HTTPError): + manage._url_download("https://example.com/m.gguf", str(dest)) + assert len(calls) == 1 + assert slept == [] + + +def test_url_download_does_not_retry_full_disk(tmp_path, monkeypatch): + # ENOSPC never clears by waiting - the write side, not the wire. + dest = tmp_path / "m.gguf" + calls, slept = [], [] + + class Full: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self, n=-1): + raise OSError(errno.ENOSPC, "No space left on device") + + def fake_open(req, *, timeout): + calls.append(1) + return Full() + + monkeypatch.setattr(remote, "http_open", fake_open) + monkeypatch.setattr(manage, "_pull_sleep", slept.append) + with pytest.raises(OSError, match="No space left"): + manage._url_download("https://example.com/m.gguf", str(dest)) + assert len(calls) == 1 + assert slept == [] + + +def test_url_download_env_overrides_timeout_and_retries(tmp_path, monkeypatch): + dest = tmp_path / "m.gguf" + seen, calls = {}, [] + monkeypatch.setenv("GMLX_PULL_TIMEOUT", "5") + monkeypatch.setenv("GMLX_PULL_RETRIES", "1") + monkeypatch.setattr(manage, "_pull_sleep", lambda s: None) + + def fake_open(req, *, timeout): + seen["timeout"] = timeout + calls.append(1) + raise TimeoutError("stalled") + + monkeypatch.setattr(remote, "http_open", fake_open) + with pytest.raises(OSError): + manage._url_download("https://example.com/m.gguf", str(dest)) + assert seen["timeout"] == 5.0 + assert len(calls) == 2 # first try plus 1 retry + + def test_hf_download_delegates_to_url_download(tmp_path, monkeypatch): # _hf_download resolves the HF URL and delegates to _url_download. seen = {}