grenade/rustingface
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b41964fae03cf2c003d5cfba632e285741b97483
rustingface/test/conformance/conformance.py
T
grenadeandClaude Opus 5 cfa53e14bd fix(data): a hidden 30s body timeout capped every proxied blob 
`object_store` puts a 30-second `reqwest` timeout on the whole request,
body included. On a proxied blob that is not a timeout at all: the bucket
is read at the pace the client drains it, so it is a maximum servable
file size.

The reason it never looked like one is that object_store hides its own
consequences. It catches every response-body error and silently retries
with a resumed Range, so a large read does not fail at 30s -- it
reconnects every 30s and keeps going, until RetryConfig's 180s
retry_timeout is spent. Then the body simply ends, with a 200 already on
the wire and nothing logged anywhere. The ceiling is 180s x bandwidth: at
the 8MB/s a GPU host gets to caveman, no file over ~1.4GB can ever be
served, and no client retry converges because the next attempt is no
faster. Measured against the live bucket, 500MiB at 1MB/s: cut at 184.1s,
193,340,351 of 524,288,000 bytes. With the timeout disabled, 499.6s and
complete -- the same wall time curl takes reading the same object
straight from MinIO.

Disable it, and guard the stream the way a stream should be guarded:
`storage.connect_timeout` and an idle `storage.read_timeout` that bounds
silence rather than progress. A test asserts the total cap stays off.

Issue #1 named `server.request_timeout` as the cause. It was not -- that
key was parsed and then read by nothing, which is its own problem and is
why the hunt started 300 seconds away from a 180-second bug. Remove it.

Also log when a blob response ends before its Content-Length. Once the
headers are out the bytes leave through hyper, so a cut mid-body was
invisible on this side: the service logged a clean 200 while the client
saw a short read and retried forever.

Refs #1

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L
2026-09-02 13:03:34 +03:00

571 lines
22 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Conformance suite: test against real clients, not against the document.
The spec is only as good as the client compatibility, so this drives an
unmodified ``huggingface_hub`` (and the ``hf`` CLI) through rustingface and
checks the behaviour a stack pinned to a model revision actually depends on.
The test that matters is the offline one. Its steps here mirror spec §11:
1. fresh bucket, ``HF_ENDPOINT`` set, pull a model, confirm bytes landed;
2. cut rustingface off from upstream;
3. wipe every client-side cache;
4. pull again -- it must succeed identically, with no upstream traffic;
5. kill rustingface, start a *new* process against the same bucket, repeat.
Step 2 differs from the spec in one honest respect. The spec null-routes
``huggingface.co`` at the site router and verifies there, which CI cannot do.
This suite instead does something strictly stronger for the sealed run --
``upstream.enabled = false`` means no upstream client is constructed at all, so
there is no code path that *could* reach the network -- and, for the
"upstream enabled but unreachable" run, points the endpoint at a closed port
and asserts that a miss fails loudly rather than hanging. Run the real
null-route test by hand before trusting a deployment.
Usage::
python3 test/conformance/conformance.py --binary target/release/rustingface
Set ``--s3-endpoint``/``--s3-bucket`` (plus ``AWS_ACCESS_KEY_ID`` and
``AWS_SECRET_ACCESS_KEY``) to run against a real bucket instead of a directory.
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
# A small, ungated, stable model. Small enough that CI can pull it repeatedly,
# real enough that it exercises the LFS bridge and a safetensors file.
REPO = os.environ.get("CONFORMANCE_REPO", "hf-internal-testing/tiny-random-gpt2")
REVISION = os.environ.get("CONFORMANCE_REVISION", "main")
# The X-Error-Code values rustingface emits. These are client-internal
# constants upstream and are re-checked against the installed client below.
EXPECTED_ERROR_CODES = {
"RepoNotFound",
"RevisionNotFound",
"EntryNotFound",
"GatedRepo",
}
# A disabled repo is the odd one out: the client dispatches DisabledRepoError
# on this exact X-Error-Message, not on an X-Error-Code, so rustingface emits
# the message verbatim. If this string moves, the mapping in
# rustingface-api/src/error.rs moves with it.
DISABLED_REPO_MESSAGE = "Access to this resource is disabled."
class Failure(Exception):
"""A conformance check that did not hold."""
@dataclass
class Results:
passed: list[str] = field(default_factory=list)
failed: list[tuple[str, str]] = field(default_factory=list)
skipped: list[tuple[str, str]] = field(default_factory=list)
def skip(self, name: str, why: str) -> None:
print(f"\n=== {name}\n skipped: {why}")
self.skipped.append((name, why))
def check(self, name: str, fn) -> None:
print(f"\n=== {name}")
try:
fn()
except Exception as err: # noqa: BLE001 - a suite reports, it does not raise
print(f" FAILED: {err}")
self.failed.append((name, str(err)))
else:
print(" ok")
self.passed.append(name)
def report(self) -> int:
print(
f"\n{len(self.passed)} passed, {len(self.failed)} failed, "
f"{len(self.skipped)} skipped"
)
for name, err in self.failed:
print(f" FAILED {name}: {err}")
# Surfaced in the summary, not just where it happened: a skipped check
# reads exactly like a passing one in a scrollback, and this suite runs
# unattended on a timer.
for name, why in self.skipped:
print(f" SKIPPED {name}: {why}")
return 1 if self.failed else 0
def free_port() -> int:
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
def closed_port() -> int:
"""A port nothing is listening on, standing in for an unreachable Hub."""
return free_port()
class Instance:
"""A running rustingface, configured for one phase of the suite."""
def __init__(self, binary: Path, workdir: Path, storage: dict, upstream: dict):
self.binary = binary
self.workdir = workdir
self.port = free_port()
self.config_path = workdir / f"config-{self.port}.toml"
self.process: subprocess.Popen | None = None
storage_lines = "\n".join(f"{k} = {v}" for k, v in storage.items())
upstream_lines = "\n".join(f"{k} = {v}" for k, v in upstream.items())
self.config_path.write_text(
f"""
[server]
listen = "127.0.0.1:{self.port}"
[storage]
{storage_lines}
[upstream]
{upstream_lines}
[policy]
ref_resolution = "freeze"
[observability]
metrics = true
log_format = "text"
"""
)
@property
def endpoint(self) -> str:
return f"http://127.0.0.1:{self.port}"
def __enter__(self) -> "Instance":
self.process = subprocess.Popen(
[str(self.binary), "--config", str(self.config_path), "serve"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
for _ in range(100):
if self.process.poll() is not None:
raise Failure(f"rustingface exited early:\n{self.process.stdout.read()}")
try:
with urllib.request.urlopen(f"{self.endpoint}/healthz", timeout=1):
return self
except (urllib.error.URLError, ConnectionError, TimeoutError):
time.sleep(0.1)
raise Failure("rustingface did not become healthy")
def __exit__(self, *_exc) -> None:
if self.process is not None:
self.process.terminate()
try:
self.process.wait(timeout=30)
except subprocess.TimeoutExpired:
self.process.kill()
def metrics(self) -> str:
with urllib.request.urlopen(f"{self.endpoint}/metrics", timeout=5) as response:
return response.read().decode()
def cli(self, *args: str) -> str:
result = subprocess.run(
[str(self.binary), "--config", str(self.config_path), *args],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise Failure(f"`rustingface {' '.join(args)}` failed:\n{result.stdout}\n{result.stderr}")
return result.stdout
def wipe_client_caches(home: Path) -> None:
"""Remove every client-side cache, so a pull cannot be served locally."""
for sub in ("hub", "assets", "xet"):
shutil.rmtree(home / sub, ignore_errors=True)
shutil.rmtree(home.parent / ".cache" / "huggingface", ignore_errors=True)
def client_env(endpoint: str, home: Path) -> dict:
env = dict(os.environ)
env.update(
{
"HF_ENDPOINT": endpoint,
"HF_HOME": str(home),
"HF_HUB_CACHE": str(home / "hub"),
# rustingface implements no Xet client and never advertises the
# capability; setting this makes the client's side explicit too.
"HF_HUB_DISABLE_XET": "1",
"HF_HUB_DISABLE_TELEMETRY": "1",
"HF_HUB_DISABLE_PROGRESS_BARS": "1",
}
)
env.pop("HF_TOKEN", None)
return env
def run_client(script: str, endpoint: str, home: Path) -> str:
"""Run a snippet in a fresh interpreter, so no client state leaks between phases."""
result = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
env=client_env(endpoint, home),
)
if result.returncode != 0:
raise Failure(f"client failed:\n{result.stdout}\n{result.stderr}")
return result.stdout
DOWNLOAD_ONE = """
import hashlib, json, sys
from huggingface_hub import hf_hub_download
path = hf_hub_download({repo!r}, "config.json", revision={revision!r})
data = open(path, "rb").read()
print(json.dumps({{"sha256": hashlib.sha256(data).hexdigest(), "len": len(data)}}))
"""
SNAPSHOT = """
import hashlib, json, os
from huggingface_hub import snapshot_download
root = snapshot_download({repo!r}, revision={revision!r})
files = {{}}
for base, _dirs, names in os.walk(root):
for name in names:
full = os.path.join(base, name)
if os.path.islink(full) and not os.path.exists(full):
continue
rel = os.path.relpath(full, root)
files[rel] = hashlib.sha256(open(full, "rb").read()).hexdigest()
print(json.dumps(files, sort_keys=True))
"""
MODEL_INFO = """
import json
from huggingface_hub import HfApi
info = HfApi().model_info({repo!r}, revision={revision!r})
print(json.dumps({{"sha": info.sha, "siblings": sorted(s.rfilename for s in info.siblings)}}))
"""
FILE_METADATA = """
import json
from huggingface_hub import get_hf_file_metadata, hf_hub_url
url = hf_hub_url({repo!r}, "config.json", revision={revision!r})
meta = get_hf_file_metadata(url)
print(json.dumps({{"commit": meta.commit_hash, "etag": meta.etag, "size": meta.size}}))
"""
MISSING_FILE = """
import json
from huggingface_hub import hf_hub_download
from huggingface_hub.errors import EntryNotFoundError, RepositoryNotFoundError, RevisionNotFoundError
try:
hf_hub_download({repo!r}, "no-such-file.bin", revision={revision!r})
except EntryNotFoundError:
print(json.dumps({{"raised": "EntryNotFoundError"}}))
except (RepositoryNotFoundError, RevisionNotFoundError) as err:
print(json.dumps({{"raised": type(err).__name__}}))
else:
print(json.dumps({{"raised": None}}))
"""
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--binary", type=Path, default=Path("target/release/rustingface"))
parser.add_argument("--s3-endpoint")
parser.add_argument("--s3-bucket")
parser.add_argument("--keep", action="store_true", help="keep the work directory")
args = parser.parse_args()
if not args.binary.exists():
print(f"{args.binary} does not exist; build it with `cargo build --release`", file=sys.stderr)
return 2
results = Results()
workdir = Path(tempfile.mkdtemp(prefix="rustingface-conformance-"))
print(f"work directory: {workdir}")
bucket_dir = workdir / "bucket"
bucket_dir.mkdir()
hf_home = workdir / "hf"
if args.s3_endpoint:
storage = {
"endpoint": json.dumps(args.s3_endpoint),
"bucket": json.dumps(args.s3_bucket or "rustingface-conformance"),
"region": '"us-east-1"',
"path_style": "true",
"multipart_part_size": '"5MiB"',
}
for var, key in (("AWS_ACCESS_KEY_ID", "access_key_id_file"),
("AWS_SECRET_ACCESS_KEY", "secret_access_key_file")):
value = os.environ.get(var)
if not value:
print(f"{var} must be set to run against S3", file=sys.stderr)
return 2
secret = workdir / key
secret.write_text(value)
storage[key] = json.dumps(str(secret))
else:
storage = {"local_path": json.dumps(str(bucket_dir)), "multipart_part_size": '"5MiB"'}
open_upstream = {"enabled": "true", "endpoint": '"https://huggingface.co"'}
unreachable_upstream = {"enabled": "true", "endpoint": f'"http://127.0.0.1:{closed_port()}"',
"connect_timeout": '"2s"'}
sealed_upstream = {"enabled": "false"}
# ------------------------------------------------------------------
# The client's own constants, re-checked rather than assumed.
# ------------------------------------------------------------------
def check_error_codes() -> None:
import huggingface_hub
print(f" huggingface_hub {huggingface_hub.__version__}")
source = Path(huggingface_hub.__file__).parent
found = set()
for path in source.rglob("*.py"):
text = path.read_text(errors="ignore")
for code in EXPECTED_ERROR_CODES:
if f'"{code}"' in text or f"'{code}'" in text:
found.add(code)
missing = EXPECTED_ERROR_CODES - found
if missing:
raise Failure(
f"this client does not mention {sorted(missing)}; the X-Error-Code strings "
"rustingface emits are version-coupled and need review"
)
http = (source / "utils" / "_http.py").read_text()
if DISABLED_REPO_MESSAGE not in http:
raise Failure(
f"this client no longer dispatches DisabledRepoError on "
f"{DISABLED_REPO_MESSAGE!r}; rustingface-api/src/error.rs needs review"
)
print(f" {len(found)} codes plus the disabled-repo message all match")
results.check("X-Error-Code strings match the pinned client", check_error_codes)
baseline: dict = {}
snapshot_baseline: dict = {}
# ------------------------------------------------------------------
# 1. Cold pull through rustingface.
# ------------------------------------------------------------------
def cold_pull() -> None:
nonlocal baseline, snapshot_baseline
wipe_client_caches(hf_home)
with Instance(args.binary, workdir, storage, open_upstream) as rf:
baseline = json.loads(
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
snapshot_baseline = json.loads(
run_client(SNAPSHOT.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
print(f" {len(snapshot_baseline)} files in the snapshot")
listing = rf.cli("list")
if REPO not in listing:
raise Failure(f"`rustingface list` does not mention {REPO}:\n{listing}")
if "rustingface_bytes_served_total" not in rf.metrics():
raise Failure("/metrics did not expose the served-bytes counter")
results.check("a cold pull succeeds and lands in the bucket", cold_pull)
# ------------------------------------------------------------------
# 2-4. Cut off upstream, wipe the client, pull again.
# ------------------------------------------------------------------
def sealed_pull() -> None:
wipe_client_caches(hf_home)
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
again = json.loads(
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if again != baseline:
raise Failure(f"sealed pull differs from the first: {again} != {baseline}")
snapshot_again = json.loads(
run_client(SNAPSHOT.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if snapshot_again != snapshot_baseline:
missing = set(snapshot_baseline) - set(snapshot_again)
raise Failure(f"sealed snapshot differs; missing {sorted(missing)}")
results.check("a sealed instance serves the identical snapshot", sealed_pull)
# ------------------------------------------------------------------
# 5. A brand-new process against the same bucket.
# ------------------------------------------------------------------
def portable_state() -> None:
wipe_client_caches(hf_home)
# Nothing carries over: a fresh process, a fresh client cache, only the
# bucket in common. This is the step that tests sovereignty rather
# than a cache.
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
again = json.loads(
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if again != baseline:
raise Failure(f"reinstalled instance differs: {again} != {baseline}")
results.check("a reinstalled instance serves from the bucket alone", portable_state)
# ------------------------------------------------------------------
# Metadata replay.
# ------------------------------------------------------------------
def metadata_replay() -> None:
with Instance(args.binary, workdir, storage, open_upstream) as rf:
live = json.loads(
run_client(FILE_METADATA.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
replayed = json.loads(
run_client(FILE_METADATA.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if live != replayed:
raise Failure(f"file metadata changed once sealed: {live} != {replayed}")
print(f" commit {live['commit']} etag {live['etag']} size {live['size']}")
results.check("file metadata replays byte-identically when sealed", metadata_replay)
def model_info() -> None:
with Instance(args.binary, workdir, storage, open_upstream) as rf:
info = json.loads(
run_client(MODEL_INFO.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if not info["sha"]:
raise Failure("model_info returned no commit sha")
print(f" {info['sha']} with {len(info['siblings'])} siblings")
results.check("model_info answers through rustingface", model_info)
# ------------------------------------------------------------------
# Failures are loud.
# ------------------------------------------------------------------
def missing_file_raises() -> None:
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
result = json.loads(
run_client(MISSING_FILE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if result["raised"] is None:
raise Failure("a missing file returned successfully instead of raising")
print(f" client raised {result['raised']}")
results.check("a missing file raises a typed client exception", missing_file_raises)
def unreachable_upstream_fails_loudly() -> None:
wipe_client_caches(hf_home)
with Instance(args.binary, workdir, storage, unreachable_upstream) as rf:
# Something already stored still serves.
served = json.loads(
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if served != baseline:
raise Failure("a stored file stopped serving when upstream became unreachable")
# Something not stored fails, rather than returning an empty file.
result = json.loads(
run_client(MISSING_FILE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
)
if result["raised"] is None:
raise Failure("an unfetchable file returned successfully")
results.check(
"an unreachable upstream serves what is stored and refuses what is not",
unreachable_upstream_fails_loudly,
)
# ------------------------------------------------------------------
# The hf CLI.
# ------------------------------------------------------------------
def hf_cli() -> None:
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
env = client_env(rf.endpoint, hf_home)
whoami = subprocess.run(
["hf", "auth", "whoami"], capture_output=True, text=True, env=env
)
# `hf auth whoami` without a token is allowed to report "not logged
# in"; what must not happen is a crash against /api/whoami-v2.
if "Traceback" in whoami.stderr:
raise Failure(f"hf auth whoami crashed:\n{whoami.stderr}")
download = subprocess.run(
["hf", "download", REPO, "config.json", "--revision", REVISION],
capture_output=True,
text=True,
env=env,
)
if download.returncode != 0:
raise Failure(f"hf download failed:\n{download.stdout}\n{download.stderr}")
if shutil.which("hf"):
results.check("the hf CLI downloads through rustingface", hf_cli)
else:
results.skip(
"the hf CLI downloads through rustingface",
"hf not on PATH (is the venv's bin directory exported?)",
)
# ------------------------------------------------------------------
# safetensors, the path an inference stack actually takes.
# ------------------------------------------------------------------
def safetensors_load() -> None:
script = """
import glob, json, os
from huggingface_hub import snapshot_download
from safetensors import safe_open
root = snapshot_download({repo!r}, revision={revision!r})
files = glob.glob(os.path.join(root, "**", "*.safetensors"), recursive=True)
if not files:
print(json.dumps({{"tensors": 0, "note": "repo has no safetensors"}}))
else:
total = 0
for path in files:
with safe_open(path, framework="np") as handle:
total += len(handle.keys())
print(json.dumps({{"tensors": total}}))
""".format(repo=REPO, revision=REVISION)
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
out = json.loads(run_client(script, rf.endpoint, hf_home))
print(f" {out}")
results.check("safetensors files load from a sealed snapshot", safetensors_load)
# ------------------------------------------------------------------
# Admin operations against the same bucket.
# ------------------------------------------------------------------
def admin_operations() -> None:
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
show = rf.cli("show", REPO)
manifest = json.loads(show)
if not manifest["entries"]:
raise Failure("the manifest records no entries")
verify = rf.cli("verify", REPO)
if "0 corrupt" not in verify:
raise Failure(f"verify reported problems:\n{verify}")
dry = rf.cli("gc", "--dry-run")
if "0 reclaimable" not in dry:
raise Failure(f"gc found reclaimable blobs while everything is referenced:\n{dry}")
results.check("show, verify and gc agree with what was stored", admin_operations)
if not args.keep:
shutil.rmtree(workdir, ignore_errors=True)
else:
print(f"\nwork directory kept at {workdir}")
return results.report()
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink