Fix(validator): HF Hub flake in foreground baseline must not kill the process

connito/test/test_foreground_baseline_resilience.py

@@ -0,0 +1,143 @@
+"""The 2026-07-10 crash: a transient HF Hub failure inside the foreground
+baseline dataloader build (`_evaluate_on_fresh_loader_sync` →
+`load_streaming_shard` → `requests.ConnectionError`) escaped
+`evaluate_foreground_round`, reached run()'s top-level handler, and killed
+the validator — losing round 8590274 entirely.
+
+These tests pin the fix: the baseline build retries with backoff and, on
+exhaustion, `evaluate_foreground_round` returns cleanly (skipping the
+foreground pass) instead of raising.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import types
+from types import SimpleNamespace
+
+import pytest
+
+
+def _install_stub_if_unavailable(mod_path: str, attrs: dict) -> None:
+    try:
+        __import__(mod_path)
+        return
+    except Exception:
+        pass
+    mod = types.ModuleType(mod_path)
+    for name, value in attrs.items():
+        setattr(mod, name, value)
+    sys.modules[mod_path] = mod
+
+
+_install_stub_if_unavailable(
+    "connito.shared.dataloader",
+    {"get_dataloader": lambda **k: None, "materialize_batches": lambda dl, n: []},
+)
+_install_stub_if_unavailable(
+    "connito.shared.evaluate",
+    {"evaluate_model": lambda *a, **k: {"val_loss": 100.0}},
+)
+
+import connito.validator.evaluator as evaluator  # noqa: E402
+from connito.validator.round import Round  # noqa: E402
+
+
+def _make_round(*, foreground_uids: tuple[int, ...] = ()) -> Round:
+    return Round(
+        round_id=42,
+        seed="test-seed",
+        validator_miner_assignment={},
+        foreground_uids=foreground_uids,
+        background_uids=(),
+        uid_to_hotkey={uid: f"HK{uid}" for uid in foreground_uids},
+        model_snapshot_cpu={},
+        journal_path=None,
+        score_aggregator=None,
+        score_path=None,
+    )
+
+
+def _call(monkeypatch, *, baseline_fn, completed_out=None):
+    """Invoke evaluate_foreground_round with minimal stubs and no sleeps."""
+    monkeypatch.setattr(
+        evaluator, "FOREGROUND_BASELINE_RETRY_DELAYS_SEC", (0.0, 0.0, 0.0),
+    )
+    monkeypatch.setattr(evaluator, "_evaluate_on_fresh_loader_sync", baseline_fn)
+    # subtensor.block > end_block so the polling loop never runs on the
+    # success path; the failure path returns before either is consulted.
+    subtensor = SimpleNamespace(block=1_001)
+    return asyncio.run(
+        evaluator.evaluate_foreground_round(
+            config=SimpleNamespace(),
+            round_obj=_make_round(),
+            subtensor=subtensor,
+            step=1,
+            device="cpu",
+            base_model=None,
+            tokenizer=None,
+            end_block=1_000,
+            expert_group_assignment={},
+            per_miner_eval_timeout_sec=5.0,
+            completed_out=completed_out,
+        )
+    )
+
+
+def test_baseline_hf_failure_degrades_instead_of_raising(monkeypatch) -> None:
+    """Repro of the crash: every baseline attempt raises ConnectionError.
+    Before the fix this propagated out (process-fatal); now the round's
+    foreground pass is skipped and an empty result is returned."""
+    calls = {"n": 0}
+
+    def _always_fails(**kwargs):
+        calls["n"] += 1
+        raise ConnectionError("huggingface.co Read timed out")
+
+    result = _call(monkeypatch, baseline_fn=_always_fails)
+    assert result == []
+    assert calls["n"] == 3  # one attempt per configured delay
+
+
+def test_baseline_failure_returns_completed_out_alias(monkeypatch) -> None:
+    """When the caller pre-allocates `completed_out` (run.py does, so
+    partial scores survive cancellation), the degraded path must return
+    that same list, not a fresh one."""
+    sink: list = []
+
+    def _always_fails(**kwargs):
+        raise ConnectionError("boom")
+
+    result = _call(monkeypatch, baseline_fn=_always_fails, completed_out=sink)
+    assert result is sink
+
+
+def test_baseline_transient_failure_recovers_on_retry(monkeypatch) -> None:
+    """First attempt fails (the transient blip), second succeeds — the
+    round proceeds with a real baseline instead of being skipped."""
+    calls = {"n": 0}
+
+    def _flaky(**kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise ConnectionError("transient")
+        return {"val_loss": 4.0}
+
+    result = _call(monkeypatch, baseline_fn=_flaky)
+    # Empty foreground set + block past end_block → completes with no jobs,
+    # but crucially it got PAST the baseline (no skip, no raise).
+    assert result == []
+    assert calls["n"] == 2
+
+
+def test_baseline_success_unchanged(monkeypatch) -> None:
+    calls = {"n": 0}
+
+    def _ok(**kwargs):
+        calls["n"] += 1
+        return {"val_loss": 4.2}
+
+    result = _call(monkeypatch, baseline_fn=_ok)
+    assert result == []
+    assert calls["n"] == 1  # no spurious retries on success

connito/validator/evaluator.py

@@ -584,6 +584,13 @@ class MinerEvalJob:
 EVAL_WORKERS = 1
 DOWNLOAD_TIMEOUT_SEC = 60
 EVAL_MAX_BATCHES = 50
+# Backoff delays between foreground-baseline dataloader build attempts.
+# Same policy (and rationale) as the background worker's
+# DATALOADER_BUILD_RETRY_DELAYS_SEC: total budget ~40 s absorbs transient
+# HF Hub blips without eating meaningfully into the eval window; after
+# exhaustion the foreground pass degrades (skipped for the round) instead
+# of raising — an HF flake must never be process-fatal.
+FOREGROUND_BASELINE_RETRY_DELAYS_SEC: tuple[float, ...] = (0.0, 10.0, 30.0)
 # ------------------------------------------------------------------------------
 
 # def load_model_from_path(path: str, base_model, device: torch.device) -> nn.Module:
@@ -957,16 +964,66 @@ async def evaluate_foreground_round(
     # e692cc7, which moved the GPU work off the event loop); wrap it in
     # `asyncio.to_thread` so this `async def evaluate_foreground_round`
     # doesn't block the event loop during the baseline pass.
-    baseline_metrics = await asyncio.to_thread(
-        _evaluate_on_fresh_loader_sync,
-        config=config,
-        tokenizer=tokenizer,
-        combinded_seed=round_obj.seed,
-        step=step,
-        model=base_model,
-        device=device,
-        max_eval_batches=EVAL_MAX_BATCHES,
-    )
+    #
+    # The baseline build streams eval shards from HF Hub, and a transient
+    # Hub blip here used to be PROCESS-FATAL: the exception propagated
+    # through run()'s foreground loop into the top-level "Quit training"
+    # handler and took the whole validator down (seen 2026-07-10, round
+    # 8590274: `requests.ConnectionError: huggingface.co Read timed out`
+    # inside `load_streaming_shard` → validator exit → round lost →
+    # fallback weights). Mirror the background worker's policy: bounded
+    # retries with backoff, then degrade — skip foreground eval for the
+    # round instead of raising. Bg-eval still covers the roster and
+    # finalize proceeds normally; a skipped foreground pass costs one
+    # round of foreground coverage, not the process.
+    baseline_metrics: dict | None = None
+    last_error: Exception | None = None
+    for attempt, delay in enumerate(FOREGROUND_BASELINE_RETRY_DELAYS_SEC):
+        if delay > 0:
+            logger.info(
+                "foreground eval: backing off before retrying baseline build",
+                attempt=attempt + 1,
+                of=len(FOREGROUND_BASELINE_RETRY_DELAYS_SEC),
+                delay_sec=delay,
+                round_id=round_obj.round_id,
+            )
+            await asyncio.sleep(delay)
+        try:
+            baseline_metrics = await asyncio.to_thread(
+                _evaluate_on_fresh_loader_sync,
+                config=config,
+                tokenizer=tokenizer,
+                combinded_seed=round_obj.seed,
+                step=step,
+                model=base_model,
+                device=device,
+                max_eval_batches=EVAL_MAX_BATCHES,
+            )
+            last_error = None
+            break
+        except asyncio.CancelledError:
+            # The outer wait_for deadline — not ours to swallow.
+            raise
+        except Exception as e:
+            last_error = e
+            logger.warning(
+                "foreground eval: baseline build/eval failed",
+                attempt=attempt + 1,
+                of=len(FOREGROUND_BASELINE_RETRY_DELAYS_SEC),
+                error=str(e),
+                round_id=round_obj.round_id,
+            )
+
+    if baseline_metrics is None:
+        logger.error(
+            "foreground eval: baseline exhausted retries — skipping "
+            "foreground eval this round (bg-eval still covers the roster)",
+            attempts=len(FOREGROUND_BASELINE_RETRY_DELAYS_SEC),
+            error=str(last_error),
+            round_id=round_obj.round_id,
+        )
+        return completed_out if completed_out is not None else []
+
     baseline_loss = float(baseline_metrics.get("val_loss", 100))
     del baseline_metrics
     gc.collect()

connito/validator/run.py

@@ -1486,6 +1486,22 @@ async def _bounded_foreground_eval():
                     end_block=phase_response.phase_end_block,
                     completed_count=len(miner_jobs),
                 )
+            except Exception:
+                # Foreground eval is best-effort: bg-eval covers the same
+                # roster and finalize does not require a foreground pass.
+                # Before this catch, any non-timeout exception here (e.g.
+                # an HF Hub ConnectionError leaking from the baseline
+                # dataloader build, 2026-07-10 round 8590274) escaped to
+                # run()'s top-level handler and KILLED the validator —
+                # losing the whole round instead of one foreground pass.
+                # evaluate_foreground_round now retries/degrades its own
+                # HF path; this is the safety net for anything else.
+                logger.exception(
+                    "Foreground evaluation failed — continuing round with "
+                    "partial scores (bg-eval unaffected)",
+                    round_id=new_round.round_id,
+                    completed_count=len(miner_jobs),
+                )
             finally:
                 foreground_loop.close()