diff --git a/connito/shared/telemetry.py b/connito/shared/telemetry.py index 6aa6623..c987e5c 100644 --- a/connito/shared/telemetry.py +++ b/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. diff --git a/connito/test/test_round_progress_publish.py b/connito/test/test_round_progress_publish.py new file mode 100644 index 0000000..588c0e9 --- /dev/null +++ b/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) diff --git a/connito/validator/background_eval_worker.py b/connito/validator/background_eval_worker.py index da1c34a..d198a22 100644 --- a/connito/validator/background_eval_worker.py +++ b/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 diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 93a9467..9123264 100644 --- a/connito/validator/evaluator.py +++ b/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, diff --git a/connito/validator/round.py b/connito/validator/round.py index a3c310d..aa16db3 100644 --- a/connito/validator/round.py +++ b/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: diff --git a/connito/validator/run.py b/connito/validator/run.py index 7edc947..1f3ffd2 100644 --- a/connito/validator/run.py +++ b/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