unsloth/studio/backend/tests/test_gguf_xet_fallback_integration.py
Daniel Han 58c2ec1ebd
Studio: Xet-primary model downloads with automatic HTTP fallback on stall (#6372)
* Studio: add shared Xet-primary download helper with HTTP stall fallback

Xet is the fast default transport in huggingface_hub, but a stalled Xet
transfer hangs with no progress and no exception, and a blocked native thread
cannot be killed. The safetensors inference path already recovers (subprocess
watchdog + respawn with HF_HUB_DISABLE_XET=1); the GGUF and training paths do
not. Add a reusable helper that the in-process paths can adopt.

utils/hf_xet_fallback.py:
- DownloadStallError (moved here from core/inference/orchestrator.py, which now
  imports it; behavior unchanged, still a RuntimeError subclass).
- get_hf_download_state / start_watchdog: a no-progress watchdog built on the
  sparse-aware hub.utils.hf_cache_state helpers; fires only while a .incomplete
  is present and the on-disk byte total is unchanged for stall_timeout.
- hf_hub_download_with_xet_fallback: cached files short-circuit; otherwise the
  download runs in a spawn child (own process group) supervised by the watchdog.
  On a stall it kills the child, makes the partial safe for HTTP via
  prepare_cache_for_transport, and respawns once with HF_HUB_DISABLE_XET=1. Cancel
  and deterministic errors (auth/missing/disk) propagate without a fallback.

Tests cover the watchdog state machine, the transport decision logic, and a
regression lock that HF_HUB_DISABLE_XET is honored in a fresh interpreter.

* Studio: route GGUF Chat-Mode downloads through the Xet->HTTP fallback

The GGUF load path (_download_gguf main+shards, _download_companion_gguf for
mmproj/MTP) called a bare blocking hf_hub_download with no recovery, so a Xet
stall hung the Chat-Mode load with no fallback. Route those three calls through
hf_hub_download_with_xet_fallback: Xet stays primary, HTTP is used only if Xet
stalls, per-file so finished shards stay cached. The existing _cancel_event is
threaded through, the Cancelled sentinel is preserved, and companions stay
best-effort (a terminal stall is swallowed to None). Cached files short-circuit
in the helper with no subprocess, so the fast path is unchanged.

The two offline mmproj tests are repointed from huggingface_hub.hf_hub_download
to the new call boundary (the helper) since the download now goes through it.

* Studio: recover a stalled training model-load via Xet->HTTP respawn

Training runs in a spawn subprocess and FastModel.from_pretrained downloads
internally, so the download cannot be wrapped per-file like GGUF. Instead the
worker now watches the HF cache during the model-load phase (emitting
model_load_started / model_load_completed and a stall event), and the parent
recovers a stall by terminating the worker and respawning it once with
HF_HUB_DISABLE_XET=1.

worker.py: set HF_HUB_DISABLE_XET=1 before any HF import when the parent passes
disable_xet (respawn), and wrap trainer.load_model with start_watchdog.

training.py: plumb disable_xet through the config; track the model-load window;
on a first-load stall arm a one-shot respawn (handled on the exiting pump thread,
so no pump self-join) that preserves the DB run row (history is not duplicated)
and re-runs the load over HTTP. A second stall, or a stall outside model-load,
surfaces as a normal error. W&B init happens after model-load, so a pre-load
respawn cannot duplicate it; the dataset is re-formatted in the new worker.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: add gated test-only fault-injection hook for the Xet stall path

UNSLOTH_HF_XET_FORCE_STALL=1 makes the Xet download attempt write a partial
blob and hang, so the no-progress watchdog and the HTTP fallback can be
exercised end to end against a real repo (never set in production). Used to
verify recovery on real models: a forced Xet stall on a 5.37GB Qwen3.5-35B-A3B
shard triggered the watchdog and the HTTP retry downloaded the correct file
(sha256 verified).

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: tighten Xet-fallback comments and consolidate its tests

Trim docstrings and inline comments across the Xet->HTTP fallback code to
the non-obvious why (spawn-not-thread, killpg-not-getpgid, the sparse-partial
HTTP-resume hazard); drop comments that merely restate the code. Verified
comment-only with an AST signature check.

Merge the three helper-level test files (watchdog, transport policy, and the
HF_HUB_DISABLE_XET regression lock) into tests/test_hf_xet_fallback.py, and
prefer the real structlog over a bare stub so test collection order cannot
leak an incomplete module to others that log at import.

Full backend suite: 3455 passed, 14 pre-existing flash-attn failures only.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2026-06-16 06:17:54 -07:00

160 lines
4.9 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Integration: GGUF Chat-Mode downloads route through the Xet->HTTP helper,
preserving cancellation and the best-effort companion contract. No GPU, no
network, no real subprocess (the helper is patched).
"""
from __future__ import annotations
import sys
import threading
import types as _types
from pathlib import Path
from unittest.mock import patch
import pytest
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
if _BACKEND_DIR not in sys.path:
sys.path.insert(0, _BACKEND_DIR)
# Heavy-dep stubbing; prefer the real structlog so a bare stub never leaks to
# later modules that log at import time.
_loggers_stub = _types.ModuleType("loggers")
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
sys.modules.setdefault("loggers", _loggers_stub)
try:
import structlog # noqa: F401
except ImportError:
sys.modules["structlog"] = _types.ModuleType("structlog")
try:
import httpx # noqa: F401
except ImportError:
_httpx_stub = _types.ModuleType("httpx")
for _exc in (
"ConnectError",
"TimeoutException",
"ReadTimeout",
"ReadError",
"RemoteProtocolError",
"CloseError",
"HTTPError",
"RequestError",
"HTTPStatusError",
):
setattr(_httpx_stub, _exc, type(_exc, (Exception,), {}))
_httpx_stub.Response = type("Response", (), {})
_httpx_stub.Request = type("Request", (), {})
_httpx_stub.Timeout = type("Timeout", (), {"__init__": lambda self, *a, **k: None})
_httpx_stub.Client = type(
"Client",
(),
{
"__init__": lambda self, **k: None,
"__enter__": lambda self: self,
"__exit__": lambda self, *a: None,
},
)
sys.modules.setdefault("httpx", _httpx_stub)
from huggingface_hub import constants as hf_constants
from core.inference.llama_cpp import LlamaCppBackend
from utils.hf_xet_fallback import DownloadStallError
REPO = "unsloth/vision-GGUF"
@pytest.fixture
def hf_cache(tmp_path, monkeypatch):
monkeypatch.setattr(hf_constants, "HF_HUB_CACHE", str(tmp_path))
return tmp_path
def _build_cache(
root: Path,
repo_id: str,
files: dict[str, int],
sha: str = "a" * 40,
) -> Path:
repo_dir = root / f"models--{repo_id.replace('/', '--')}"
(repo_dir / "blobs").mkdir(parents = True, exist_ok = True)
snap = repo_dir / "snapshots" / sha
snap.mkdir(parents = True, exist_ok = True)
for rel, size in files.items():
(snap / rel).write_bytes(b"\0" * size)
return snap
def test_companion_routes_through_helper(hf_cache):
_build_cache(hf_cache, REPO, {"mmproj-vision-F16.gguf": 1})
backend = LlamaCppBackend()
captured = {}
def fake_helper(
repo_id,
filename,
token = None,
**kwargs,
):
captured["filename"] = filename
captured["cancel_event"] = kwargs.get("cancel_event")
return f"/fake/{filename}"
with (
patch("huggingface_hub.list_repo_files", lambda *a, **k: ["mmproj-vision-F16.gguf"]),
patch("core.inference.llama_cpp.hf_hub_download_with_xet_fallback", fake_helper),
):
out = backend._download_mmproj(hf_repo = REPO, hf_token = None)
assert out == "/fake/mmproj-vision-F16.gguf"
# _cancel_event must be threaded through so /unload can abort the download.
assert captured["cancel_event"] is backend._cancel_event
def test_companion_swallows_terminal_stall_to_none(hf_cache):
_build_cache(hf_cache, REPO, {"mmproj-vision-F16.gguf": 1})
backend = LlamaCppBackend()
def stalling_helper(
repo_id,
filename,
token = None,
**kwargs,
):
raise DownloadStallError("both transports stalled")
with (
patch("huggingface_hub.list_repo_files", lambda *a, **k: ["mmproj-vision-F16.gguf"]),
patch("core.inference.llama_cpp.hf_hub_download_with_xet_fallback", stalling_helper),
):
out = backend._download_mmproj(hf_repo = REPO, hf_token = None)
assert out is None, "a companion download is best-effort; a terminal stall must not raise"
def test_companion_cancelled_skips_download(hf_cache):
_build_cache(hf_cache, REPO, {"mmproj-vision-F16.gguf": 1})
backend = LlamaCppBackend()
backend._cancel_event.set()
called = {"n": 0}
def helper(
repo_id,
filename,
token = None,
**kwargs,
):
called["n"] += 1
return "/should-not-happen"
with (
patch("huggingface_hub.list_repo_files", lambda *a, **k: ["mmproj-vision-F16.gguf"]),
patch("core.inference.llama_cpp.hf_hub_download_with_xet_fallback", helper),
):
out = backend._download_mmproj(hf_repo = REPO, hf_token = None)
assert out is None
assert called["n"] == 0, "a cancelled load must not start a companion download"