📊 telemetry: publish round progress from the foreground path too

connito/shared/telemetry.py

@@ -552,6 +552,36 @@ def note_round_series(round_id: int) -> None:
         pass
 
 
+def set_round_progress(
+    round_id: int, *, scored: int, failed: int, pending: int
+) -> None:
+    """Publish a round's evaluation-progress counters.
+
+    Shared by every path that advances a round: the freeze-time initial
+    publish, foreground eval (during Submission), and the background eval
+    worker (after Merge). This previously lived as a private static method
+    on `BackgroundEvalWorker`, so the counters only started moving once the
+    background eval window opened at Merge — foreground evals accumulated
+    in `scored_uids` with nothing publishing them, and the dashboard's
+    "Evaluated N of M" panel sat on the *previous* round's final value for
+    the first ~14 minutes of every round, then jumped straight to the
+    foreground total.
+
+    Registers the round_id via `note_round_series` so these labelsets are
+    evicted on the normal cutoff.
+
+    Best-effort — never raises. Telemetry must not influence scoring.
+    """
+    try:
+        note_round_series(int(round_id))
+        rid = str(int(round_id))
+        VALIDATOR_ROUND_MINERS_SCORED.labels(round_id=rid).set(float(scored))
+        VALIDATOR_ROUND_MINERS_FAILED.labels(round_id=rid).set(float(failed))
+        VALIDATOR_ROUND_MINERS_PENDING.labels(round_id=rid).set(float(pending))
+    except Exception:
+        pass
+
+
 def set_baseline_loss(round_id: int, baseline_loss: float) -> None:
     """Publish a round's foreground-eval baseline loss to BOTH the unlabeled
     gauge (backward compat) and the per-round labeled family.

connito/test/test_round_progress_publish.py

@@ -0,0 +1,152 @@
+"""Tests for round progress-counter publishing.
+
+`validator_round_miners_{scored,failed,pending}` used to be written only by
+`BackgroundEvalWorker._record_metrics`, whose loop starts when the eval
+window opens at Merge — so foreground evaluations (which run during
+Submission) accumulated in `scored_uids` with nothing publishing them, and
+the dashboard's "Evaluated N of M" panel sat on the previous round's final
+value for the first ~14 minutes of every round.
+
+The publish now lives in `telemetry.set_round_progress` / `Round.publish_progress`
+and is called from freeze, the foreground path, and the background worker.
+
+The prometheus registry is process-global, so these tests use round ids in a
+dedicated 93xxxx range to avoid colliding with other test modules.
+"""
+from __future__ import annotations
+
+import threading
+
+from connito.shared import telemetry as T
+from connito.validator.round import Round
+
+
+def _round_gauge(gauge, family_name: str, round_id: int) -> float | None:
+    for metric in gauge.collect():
+        for s in metric.samples:
+            if s.name == family_name and s.labels.get("round_id") == str(round_id):
+                return s.value
+    return None
+
+
+def _progress(round_id: int) -> tuple[float | None, float | None, float | None]:
+    return (
+        _round_gauge(T.VALIDATOR_ROUND_MINERS_SCORED, "validator_round_miners_scored", round_id),
+        _round_gauge(T.VALIDATOR_ROUND_MINERS_FAILED, "validator_round_miners_failed", round_id),
+        _round_gauge(T.VALIDATOR_ROUND_MINERS_PENDING, "validator_round_miners_pending", round_id),
+    )
+
+
+def _make_round(round_id: int, *, foreground=(1, 2), background=(3, 4, 5)) -> Round:
+    # Built directly rather than via `Round.freeze` (which needs chain +
+    # a model); `journal_path` / `score_aggregator` stay None so the
+    # `mark_*` helpers skip their persistence side effects.
+    return Round(
+        round_id=round_id,
+        seed="0" * 64,
+        validator_miner_assignment={},
+        foreground_uids=tuple(foreground),
+        background_uids=tuple(background),
+        uid_to_hotkey={u: f"hk{u}" for u in (*foreground, *background)},
+        model_snapshot_cpu={},
+    )
+
+
+# ---------------------------------------------------------------------------
+# set_round_progress
+# ---------------------------------------------------------------------------
+
+def test_set_round_progress_sets_all_three_gauges():
+    rid = 930_001
+    T.set_round_progress(rid, scored=7, failed=2, pending=11)
+    assert _progress(rid) == (7.0, 2.0, 11.0)
+
+
+def test_set_round_progress_registers_round_for_eviction():
+    rid = 930_002
+    T.set_round_progress(rid, scored=1, failed=0, pending=3)
+    assert rid in T._EMITTED_ROUND_IDS
+    # And the normal cutoff removes the labelsets it just created.
+    T.evict_round_series_before(rid + 1)
+    assert _progress(rid) == (None, None, None)
+
+
+def test_set_round_progress_never_raises_on_bad_input():
+    # Telemetry must not be able to break scoring.
+    T.set_round_progress(None, scored="x", failed=None, pending=object())  # type: ignore[arg-type]
+
+
+# ---------------------------------------------------------------------------
+# Round.publish_progress
+# ---------------------------------------------------------------------------
+
+def test_publish_progress_at_freeze_seeds_zero_scored_full_pending():
+    rid = 930_010
+    r = _make_round(rid)  # roster = 5
+    r.publish_progress()
+    assert _progress(rid) == (0.0, 0.0, 5.0)
+
+
+def test_publish_progress_tracks_foreground_marks():
+    """The regression this change exists for: a foreground `mark_scored`
+    followed by a publish must move the counter, with no background worker
+    involved."""
+    rid = 930_011
+    r = _make_round(rid)  # roster = 5
+    r.publish_progress()
+    assert _progress(rid) == (0.0, 0.0, 5.0)
+
+    r.mark_scored(1, 2.5)
+    r.publish_progress()
+    assert _progress(rid) == (1.0, 0.0, 4.0)
+
+    r.mark_scored(2, 1.5)
+    r.publish_progress()
+    assert _progress(rid) == (2.0, 0.0, 3.0)
+
+
+def test_publish_progress_counts_failures_and_validation_failures():
+    rid = 930_012
+    r = _make_round(rid)  # roster = 5
+    r.mark_failed(3)
+    r.publish_progress()
+    assert _progress(rid) == (0.0, 1.0, 4.0)
+
+    # `mark_validation_failed` lands in `failed_uids` too, so pending drops.
+    r.mark_validation_failed(4)
+    r.publish_progress()
+    assert _progress(rid) == (0.0, 2.0, 3.0)
+
+
+def test_publish_progress_matches_stats():
+    rid = 930_013
+    r = _make_round(rid)
+    r.mark_scored(1, 1.0)
+    r.mark_failed(3)
+    r.publish_progress()
+    stats = r.stats()
+    assert _progress(rid) == (
+        float(stats["scored"]), float(stats["failed"]), float(stats["pending"]),
+    )
+
+
+def test_publish_progress_does_not_hold_round_lock():
+    """`stats()` acquires `Round._lock`, which is a plain (non-reentrant)
+    Lock — so `publish_progress` must be callable without the caller
+    holding it, and must leave it released."""
+    rid = 930_014
+    r = _make_round(rid)
+    r.publish_progress()
+    assert r._lock.acquire(blocking=False) is True
+    r._lock.release()
+
+
+def test_publish_progress_is_isolated_per_round():
+    rid_a, rid_b = 930_020, 930_021
+    ra = _make_round(rid_a, foreground=(1,), background=(2,))
+    rb = _make_round(rid_b, foreground=(1,), background=(2, 3))
+    ra.mark_scored(1, 1.0)
+    ra.publish_progress()
+    rb.publish_progress()
+    assert _progress(rid_a) == (1.0, 0.0, 1.0)
+    assert _progress(rid_b) == (0.0, 0.0, 3.0)

connito/validator/background_eval_worker.py

@@ -28,9 +28,6 @@
     VALIDATOR_BG_EVAL_RECYCLE_TOTAL,
     VALIDATOR_BG_EVAL_STUCK_LOCK_ITERATIONS,
     VALIDATOR_BG_WORKER_PAUSED,
-    VALIDATOR_ROUND_MINERS_FAILED,
-    VALIDATOR_ROUND_MINERS_PENDING,
-    VALIDATOR_ROUND_MINERS_SCORED,
     note_round_series,
 )
 from connito.validator.evaluator import (
@@ -435,7 +432,7 @@ async def _evaluate_one(self, round_obj, *, uid: int, hotkey: str) -> None:
             # so finalize writes score=0 for it. Operational failures
             # below use plain `mark_failed` and leave the EMA alone.
             round_obj.mark_validation_failed(uid)
-            self._record_metrics(round_obj, scored_inc=False)
+            round_obj.publish_progress()
             self._prune_non_top(round_obj)
             return
 
@@ -517,7 +514,7 @@ def _run_eval():
 
         if evaluated is None:
             round_obj.mark_failed(uid)
-            self._record_metrics(round_obj, scored_inc=False)
+            round_obj.publish_progress()
             self._prune_non_top(round_obj)
             return
 
@@ -528,7 +525,7 @@ def _run_eval():
             uid=uid, hotkey=hotkey[:6],
             score=round(evaluated.score, 6),
         )
-        self._record_metrics(round_obj, scored_inc=True)
+        round_obj.publish_progress()
         self._prune_non_top(round_obj)
 
     def _stuck_lock_check_and_maybe_recycle(self) -> bool:
@@ -624,13 +621,3 @@ def _prune_non_top(self, round_obj) -> None:
                 files=deleted,
             )
 
-    @staticmethod
-    def _record_metrics(round_obj, *, scored_inc: bool) -> None:
-        try:
-            stats = round_obj.stats()
-            note_round_series(round_obj.round_id)
-            VALIDATOR_ROUND_MINERS_SCORED.labels(round_id=str(round_obj.round_id)).set(stats["scored"])
-            VALIDATOR_ROUND_MINERS_FAILED.labels(round_id=str(round_obj.round_id)).set(stats["failed"])
-            VALIDATOR_ROUND_MINERS_PENDING.labels(round_id=str(round_obj.round_id)).set(stats["pending"])
-        except Exception:
-            pass

connito/validator/evaluator.py

@@ -1127,6 +1127,7 @@ async def evaluate_foreground_round(
                 inc_error(component="foreground_eval", kind="validation")
                 _record_eval_failure(int(uid), _VALIDATION_FAIL_TO_REASON.get(fail_reason, "unknown"))
                 round_obj.mark_validation_failed(uid)
+                round_obj.publish_progress()
                 _prune_non_top_after_eval(
                     config=config,
                     round_obj=round_obj,
@@ -1167,6 +1168,7 @@ async def evaluate_foreground_round(
                 )
                 _record_eval_failure(int(uid), "timeout")
                 round_obj.mark_failed(uid)
+                round_obj.publish_progress()
                 _prune_non_top_after_eval(
                     config=config,
                     round_obj=round_obj,
@@ -1176,6 +1178,7 @@ async def evaluate_foreground_round(
                 logger.exception("foreground eval: unexpected failure", uid=uid, error=str(e))
                 _record_eval_failure(int(uid), "unknown")
                 round_obj.mark_failed(uid)
+                round_obj.publish_progress()
                 _prune_non_top_after_eval(
                     config=config,
                     round_obj=round_obj,
@@ -1184,12 +1187,14 @@ async def evaluate_foreground_round(
 
             if evaluated is None:
                 round_obj.mark_failed(uid)
+                round_obj.publish_progress()
                 _prune_non_top_after_eval(
                     config=config,
                     round_obj=round_obj,
                 )
                 continue
             round_obj.mark_scored(uid, evaluated.score)
+            round_obj.publish_progress()
             completed.append(evaluated)
             _prune_non_top_after_eval(
                 config=config,

connito/validator/round.py

@@ -19,6 +19,7 @@
 import torch.nn as nn
 
 from connito.shared.app_logging import structlog
+from connito.shared.telemetry import set_round_progress
 
 if TYPE_CHECKING:
     from connito.shared.checkpoints import ChainCheckpoint
@@ -823,6 +824,28 @@ def stats(self) -> dict[str, int]:
                 "pending": roster_size - len(self.scored_uids) - len(self.failed_uids),
             }
 
+    def publish_progress(self) -> None:
+        """Publish this round's progress counters (scored / failed / pending)
+        to Prometheus.
+
+        Called from every path that advances the round — once at freeze, after
+        each foreground evaluation, and after each background evaluation — so
+        the counters move as soon as evaluation starts rather than only once
+        the background eval window opens at Merge.
+
+        `stats()` takes `self._lock`, so callers must NOT hold it. Every call
+        site is outside the lock (the `mark_*` helpers release it before
+        returning). Best-effort: `set_round_progress` swallows telemetry
+        errors so this can never affect scoring.
+        """
+        stats = self.stats()
+        set_round_progress(
+            self.round_id,
+            scored=stats["scored"],
+            failed=stats["failed"],
+            pending=stats["pending"],
+        )
+
 
 @dataclass
 class RoundRef:

connito/validator/run.py

@@ -1417,6 +1417,12 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str =
             try:
                 note_round_series(new_round.round_id)
                 VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(0)
+                # Seed the progress counters at freeze so the round exists in
+                # the metric from the moment it is frozen (scored=0, failed=0,
+                # pending=roster) instead of appearing only once the first
+                # evaluation lands. Consumers that switch on "newest round with
+                # a non-zero scored value" are unaffected by the zero.
+                new_round.publish_progress()
             except Exception:
                 pass