diff --git a/connito/shared/telemetry.py b/connito/shared/telemetry.py index c70847a..c57bcf7 100644 --- a/connito/shared/telemetry.py +++ b/connito/shared/telemetry.py @@ -243,6 +243,39 @@ def set_validator_identity(*, hotkey: str, uid: int | None, version: str, netuid ["miner_uid"], ) +# --- Cycle-consistent per-miner attribution (dashboard contract) ----------- +# The gateway attributes every per-miner sample to the exact evaluation +# round via these three families. All are set from `finalize_round_scores` +# (including its journal-recovery replay path) so a validator restart +# re-publishes them without waiting for a fresh round. +VALIDATOR_MINER_LAST_SCORED_ROUND_ID = Gauge( + "validator_miner_last_scored_round_id", + "round_id of the last round in which this validator wrote a finalize " + "verdict (scored, tie-zeroed, validation-failed, or freeze-zero) for the miner", + ["miner_uid"], +) +VALIDATOR_MINER_ROUND_DELTA = Gauge( + "validator_miner_round_delta", + "Raw per-round improvement signal ((baseline - val_loss) ** 1.2, >= 0) " + "from the miner's most recent evaluated round. Distinct from " + "validator_miner_score_latest, which is the finalized podium rank score.", + ["miner_uid"], +) +VALIDATOR_MINER_EVALUATED_COMMIT_INFO = Gauge( + "validator_miner_evaluated_commit_info", + "round_id in which the labeled (hf_repo_id, hf_revision) was frozen and " + "evaluated for the miner. At most one labelset per miner_uid (old " + "labelsets are evicted on change).", + ["miner_uid", "hf_repo_id", "hf_revision"], +) +# uid -> (hf_repo_id, hf_revision) currently exposed on +# VALIDATOR_MINER_EVALUATED_COMMIT_INFO. Guarded by _COMMIT_INFO_LOCK; used +# to evict the previous labelset when a miner's commit changes, keeping the +# "<= 1 labelset per uid" invariant. After a restart both this dict and the +# registry start empty, so correctness holds without persistence. +_COMMIT_INFO_LOCK = threading.Lock() +_COMMIT_INFO_LABELS: dict[str, tuple[str, str]] = {} + # Per-round lifecycle (background submission validation) VALIDATOR_ROUND_LIFECYCLE_STEP = Gauge( "validator_round_lifecycle_step", @@ -264,6 +297,13 @@ def set_validator_identity(*, hotkey: str, uid: int | None, version: str, netuid "Roster miners that failed download/eval for the round", ["round_id"], ) +# round_id label values ever emitted on the per-round families above (and on +# VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL). Call sites register via +# `note_round_series`; `evict_round_series_before` removes stale labelsets on +# the same cutoff run.py already uses to prune journals/aggregator entries — +# without this, every round leaves four-plus permanent series behind. +_ROUND_SERIES_LOCK = threading.Lock() +_EMITTED_ROUND_IDS: set[int] = set() VALIDATOR_BG_WORKER_PAUSED = Gauge( "validator_bg_worker_paused", "1 while a background worker is paused on merge_phase_active / eval_window / download_window", @@ -425,6 +465,114 @@ def set_miner_eval_status(miner_uid: int | str, reason: EvalFailureReason | str pass +def set_miner_last_scored_round(miner_uid: int | str, round_id: int) -> None: + """Record the round_id of the last finalize verdict for this miner. + + Best-effort — never raises. Telemetry must not influence scoring. + """ + try: + VALIDATOR_MINER_LAST_SCORED_ROUND_ID.labels(miner_uid=str(miner_uid)).set( + float(int(round_id)) + ) + except Exception: + pass + + +def set_miner_round_delta(miner_uid: int | str, delta: float) -> None: + """Record the raw per-round improvement signal for an evaluated miner. + + Best-effort — never raises. + """ + try: + VALIDATOR_MINER_ROUND_DELTA.labels(miner_uid=str(miner_uid)).set(float(delta)) + except Exception: + pass + + +def set_miner_evaluated_commit( + miner_uid: int | str, hf_repo_id: str, hf_revision: str, round_id: int +) -> None: + """Expose which (hf_repo_id, hf_revision) was frozen + evaluated for the + miner, valued with the round_id it belongs to. + + Enforces at most ONE labelset per miner_uid: when the commit changes, the + previous labelset is removed from the registry before the new one is set, + so the gateway never sees two competing commit rows for a uid. The + KeyError guard covers the post-restart case (tracking dict repopulated + while the registry series was already re-created) and double-eviction. + + Best-effort — never raises. + """ + try: + uid = str(miner_uid) + repo = str(hf_repo_id or "") + rev = str(hf_revision or "") + if not repo or not rev: + return + with _COMMIT_INFO_LOCK: + prev = _COMMIT_INFO_LABELS.get(uid) + if prev is not None and prev != (repo, rev): + try: + VALIDATOR_MINER_EVALUATED_COMMIT_INFO.remove(uid, prev[0], prev[1]) + except KeyError: + pass + VALIDATOR_MINER_EVALUATED_COMMIT_INFO.labels( + miner_uid=uid, hf_repo_id=repo, hf_revision=rev + ).set(float(int(round_id))) + _COMMIT_INFO_LABELS[uid] = (repo, rev) + except Exception: + pass + + +def note_round_series(round_id: int) -> None: + """Register a round_id whose label value was emitted on a per-round + family, so `evict_round_series_before` can remove it later. + + Best-effort — never raises. + """ + try: + with _ROUND_SERIES_LOCK: + _EMITTED_ROUND_IDS.add(int(round_id)) + except Exception: + pass + + +def evict_round_series_before(min_round_id: int) -> int: + """Remove per-round labelsets for every tracked round_id below the + cutoff. Called from run.py's journal/aggregator prune block with the + same cutoff, so metric retention matches on-disk retention. + + Only rounds emitted by THIS process are tracked (the set is in-memory); + series left over from a previous process incarnation don't exist in the + fresh registry either, so nothing is leaked across restarts. + + Returns the number of round_ids evicted. Best-effort — never raises. + """ + removed = 0 + try: + cutoff = int(min_round_id) + with _ROUND_SERIES_LOCK: + stale = [r for r in _EMITTED_ROUND_IDS if r < cutoff] + for r in stale: + rid = str(r) + for family in ( + VALIDATOR_ROUND_LIFECYCLE_STEP, + VALIDATOR_ROUND_MINERS_PENDING, + VALIDATOR_ROUND_MINERS_SCORED, + VALIDATOR_ROUND_MINERS_FAILED, + VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL, + ): + try: + family.remove(rid) + except KeyError: + pass + _EMITTED_ROUND_IDS.discard(r) + removed += 1 + except Exception: + return removed + return removed + + def set_miner_score_snapshot( miner_uid: int | str, *, diff --git a/connito/test/test_cycle_consistent_telemetry.py b/connito/test/test_cycle_consistent_telemetry.py new file mode 100644 index 0000000..0673030 --- /dev/null +++ b/connito/test/test_cycle_consistent_telemetry.py @@ -0,0 +1,291 @@ +"""Tests for the cycle-consistent miner-telemetry contract. + +Covers: + - `set_miner_evaluated_commit` eviction invariant (≤ 1 labelset per uid, + restart-simulation safe); + - RoundJournal v2 round-trip + v1 backward compatibility; + - `finalize_round_scores` emitting last_scored_round_id / score snapshots + for every verdict uid, round deltas for scored uids, and exactly one + evaluated-commit labelset per checkpointed uid — both on a live-shaped + round and again via the journal-recovery path; + - `evict_round_series_before` removing tracked per-round labelsets. + +The prometheus registry is process-global, so every test uses uids in a +dedicated 9xxx range to avoid colliding with other test modules. +""" +from __future__ import annotations + +import json +from types import SimpleNamespace + +from connito.shared import telemetry as T +from connito.validator import round_journal as RJ +from connito.validator.evaluator import finalize_round_scores + + +def _samples_for(gauge, family_name: str) -> list: + for metric in gauge.collect(): + return [s for s in metric.samples if s.name == family_name] + return [] + + +def _labelsets_for_uid(uid: int) -> list: + return [ + s + for s in _samples_for( + T.VALIDATOR_MINER_EVALUATED_COMMIT_INFO, + "validator_miner_evaluated_commit_info", + ) + if s.labels["miner_uid"] == str(uid) + ] + + +# --------------------------------------------------------------------------- +# Commit-info eviction invariant +# --------------------------------------------------------------------------- + +def test_evaluated_commit_evicts_previous_labelset(): + uid = 9001 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-a", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-b", 200) + rows = _labelsets_for_uid(uid) + assert len(rows) == 1 + assert rows[0].labels["hf_revision"] == "rev-b" + assert rows[0].value == 200.0 + + +def test_evaluated_commit_same_labels_updates_value_in_place(): + uid = 9002 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-x", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-x", 300) + rows = _labelsets_for_uid(uid) + assert len(rows) == 1 + assert rows[0].value == 300.0 + + +def test_evaluated_commit_restart_simulation_is_safe(): + uid = 9003 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-a", 100) + # Simulate a restart of the tracking dict only (registry keeps series in + # a real restart neither survives; here the tracking dict being empty + # while the registry still has the old row exercises the KeyError-free + # first write, and the follow-up change exercises eviction rebuilt from + # scratch). + with T._COMMIT_INFO_LOCK: + T._COMMIT_INFO_LABELS.clear() + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-b", 200) + # Old row may linger (tracking was lost) but the invariant re-establishes + # on the NEXT change: + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-c", 300) + rows = _labelsets_for_uid(uid) + revs = {r.labels["hf_revision"] for r in rows} + assert "rev-c" in revs + assert "rev-b" not in revs # evicted by the rev-b -> rev-c change + + +def test_evaluated_commit_skips_empty_repo_or_revision(): + uid = 9004 + T.set_miner_evaluated_commit(uid, "", "rev-a", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", None, 100) # type: ignore[arg-type] + assert _labelsets_for_uid(uid) == [] + + +# --------------------------------------------------------------------------- +# Journal v2 / v1 compatibility +# --------------------------------------------------------------------------- + +def test_journal_v2_roundtrip(tmp_path): + j = RJ.RoundJournal( + round_id=123, + uid_to_hotkey={1: "hk1"}, + scores={1: 2.5}, + scored_uids=(1,), + uid_to_commit={1: ("acme/repo", "deadbeef")}, + finalized=True, + ) + p = tmp_path / "round_123.json" + RJ.write_atomic(p, j) + loaded = RJ.load(p) + assert loaded is not None + assert loaded.schema_version == RJ.SCHEMA_VERSION + assert loaded.uid_to_commit == {1: ("acme/repo", "deadbeef")} + assert loaded.scores == {1: 2.5} + + +def test_journal_v1_loads_with_empty_commit_map(tmp_path): + v1 = { + "round_id": 77, + "uid_to_hotkey": {"5": "hk5"}, + "scores": {"5": 1.25}, + "scored_uids": [5], + "failed_uids": [], + "validation_failed_uids": [], + "freeze_zero_uids": [], + "freeze_zero_hotkeys": {}, + "finalized": False, + "schema_version": 1, + } + p = tmp_path / "round_77.json" + p.write_text(json.dumps(v1), encoding="utf-8") + loaded = RJ.load(p) + assert loaded is not None + assert loaded.uid_to_commit == {} + assert loaded.scores == {5: 1.25} + + +def test_journal_future_version_rejected(tmp_path): + p = tmp_path / "round_88.json" + p.write_text( + json.dumps({"round_id": 88, "schema_version": RJ.SCHEMA_VERSION + 1}), + encoding="utf-8", + ) + try: + RJ.load(p) + except ValueError: + return + raise AssertionError("future schema_version must raise") + + +def test_commit_map_from_checkpoints_skips_incomplete(): + m = RJ.commit_map_from_checkpoints({ + 1: SimpleNamespace(hf_repo_id="a/r", hf_revision="v1"), + 2: SimpleNamespace(hf_repo_id=None, hf_revision="v2"), + 3: SimpleNamespace(hf_repo_id="a/r3", hf_revision=""), + }) + assert m == {1: ("a/r", "v1")} + + +# --------------------------------------------------------------------------- +# finalize_round_scores emission (live-shaped round + recovery path) +# --------------------------------------------------------------------------- + + +class _FakeAggregator: + def __init__(self): + self.rows: dict[int, list[float]] = {} + + def drop_round(self, round_id): + pass + + def add_score(self, *, uid, hotkey, score, round_id): + self.rows.setdefault(int(uid), []).append(float(score)) + + def uid_score_pairs(self, how="avg"): + if how == "latest": + return {u: v[-1] for u, v in self.rows.items()} + return {u: sum(v) / len(v) for u, v in self.rows.items()} + + def record_count(self, uid): + return len(self.rows.get(int(uid), [])) + + +def _make_round(round_id: int, journal_path=None): + return RJ._RecoveryRound( + round_id=round_id, + scores={9101: 2.5, 9102: 1.5, 9103: 0.0}, + scored_uids={9101, 9102, 9103}, + failed_uids=set(), + validation_failed_uids={9104}, + freeze_zero_uids={9105}, + freeze_zero_hotkeys={9105: "hk9105"}, + uid_to_hotkey={u: f"hk{u}" for u in (9101, 9102, 9103, 9104)}, + journal_path=journal_path, + uid_to_chain_checkpoint={ + 9101: SimpleNamespace(hf_repo_id="miner/one", hf_revision="aaa1"), + 9102: SimpleNamespace(hf_repo_id="miner/two", hf_revision="bbb2"), + }, + ) + + +def _gauge_value(gauge, family_name, uid): + for s in _samples_for(gauge, family_name): + if s.labels.get("miner_uid") == str(uid): + return s.value + return None + + +def test_finalize_emits_cycle_consistent_series(tmp_path): + rid = 524_000 + round_obj = _make_round(rid, journal_path=tmp_path / f"round_{rid}.json") + written = finalize_round_scores( + round_obj=round_obj, score_aggregator=_FakeAggregator(), score_path=None, + ) + # Verdict uids: 3 scored + 1 validation_failed + 1 freeze_zero. + assert set(written) == {9101, 9102, 9103, 9104, 9105} + for uid in written: + assert _gauge_value( + T.VALIDATOR_MINER_LAST_SCORED_ROUND_ID, + "validator_miner_last_scored_round_id", uid, + ) == float(rid) + assert _gauge_value( + T.VALIDATOR_MINER_SCORE_LATEST, "validator_miner_score_latest", uid, + ) is not None + # Deltas only for evaluated uids, raw values preserved. + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9101 + ) == 2.5 + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9103 + ) == 0.0 + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9105 + ) is None + # Exactly one commit labelset per checkpointed uid. + assert len(_labelsets_for_uid(9101)) == 1 + assert _labelsets_for_uid(9101)[0].value == float(rid) + assert len(_labelsets_for_uid(9102)) == 1 + + +def test_finalize_recovery_path_reemits_commits(tmp_path): + rid = 524_524 + # Write a v2 journal, hydrate a recovery round from it, finalize. + live = _make_round(rid, journal_path=tmp_path / f"round_{rid}.json") + RJ.write_atomic( + live.journal_path, + RJ.RoundJournal( + round_id=rid, + uid_to_hotkey=dict(live.uid_to_hotkey), + scores=dict(live.scores), + scored_uids=tuple(sorted(live.scored_uids)), + validation_failed_uids=tuple(sorted(live.validation_failed_uids)), + freeze_zero_uids=tuple(sorted(live.freeze_zero_uids)), + freeze_zero_hotkeys=dict(live.freeze_zero_hotkeys), + uid_to_commit=RJ.commit_map_from_checkpoints(live.uid_to_chain_checkpoint), + ), + ) + journal = RJ.load(live.journal_path) + assert journal is not None + recovered = RJ._RecoveryRound.from_journal(journal, live.journal_path) + finalize_round_scores( + round_obj=recovered, score_aggregator=_FakeAggregator(), score_path=None, + ) + rows = _labelsets_for_uid(9101) + assert len(rows) == 1 + assert rows[0].value == float(rid) + assert _gauge_value( + T.VALIDATOR_MINER_LAST_SCORED_ROUND_ID, + "validator_miner_last_scored_round_id", 9105, + ) == float(rid) + + +# --------------------------------------------------------------------------- +# Per-round series eviction +# --------------------------------------------------------------------------- + +def test_evict_round_series_before(): + old_rid, new_rid = 910_000, 910_524 + for rid in (old_rid, new_rid): + T.note_round_series(rid) + T.VALIDATOR_ROUND_MINERS_SCORED.labels(round_id=str(rid)).set(5) + removed = T.evict_round_series_before(new_rid) + assert removed >= 1 + rids = { + s.labels["round_id"] + for s in _samples_for( + T.VALIDATOR_ROUND_MINERS_SCORED, "validator_round_miners_scored" + ) + } + assert str(old_rid) not in rids + assert str(new_rid) in rids + # Idempotent / KeyError-safe on repeat. + assert T.evict_round_series_before(new_rid) == 0 diff --git a/connito/validator/background_eval_worker.py b/connito/validator/background_eval_worker.py index 8963045..da1c34a 100644 --- a/connito/validator/background_eval_worker.py +++ b/connito/validator/background_eval_worker.py @@ -31,6 +31,7 @@ VALIDATOR_ROUND_MINERS_FAILED, VALIDATOR_ROUND_MINERS_PENDING, VALIDATOR_ROUND_MINERS_SCORED, + note_round_series, ) from connito.validator.evaluator import ( EVAL_MAX_BATCHES, @@ -503,6 +504,7 @@ def _run_eval(): round_id=round_obj.round_id, ) try: + note_round_series(round_obj.round_id) VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL.labels( round_id=str(round_obj.round_id) ).inc() @@ -626,6 +628,7 @@ def _prune_non_top(self, round_obj) -> None: 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"]) diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 4209003..df2f5f9 100644 --- a/connito/validator/evaluator.py +++ b/connito/validator/evaluator.py @@ -344,6 +344,9 @@ def finalize_round_scores( validation_failed_uids=tuple(sorted(validation_failed)), freeze_zero_uids=tuple(sorted(freeze_zero)), freeze_zero_hotkeys=dict(freeze_hotkeys), + uid_to_commit=_rj.commit_map_from_checkpoints( + getattr(round_obj, "uid_to_chain_checkpoint", None) or {} + ), finalized=True, ), ) @@ -353,6 +356,54 @@ def finalize_round_scores( round_id=round_obj.round_id, error=str(e), ) + # --- Cycle-consistent per-miner telemetry (dashboard contract). ------- + # Emitted HERE — not from run.py's weight loop — for two reasons: + # (1) the weight loop only iterates weight recipients, so the dashboard + # previously saw score snapshots for ~1 uid; every verdict uid gets + # one now; (2) the journal-recovery replay calls this function too, + # so a restart re-publishes the series without waiting for a fresh + # round. Best-effort throughout — telemetry must never block + # finalize or scoring. + try: + from connito.shared.telemetry import ( + set_miner_evaluated_commit, + set_miner_last_scored_round, + set_miner_round_delta, + set_miner_score_snapshot, + ) + + _rid = int(round_obj.round_id) + try: + _latest_scores = score_aggregator.uid_score_pairs(how="latest") + _avg_scores = score_aggregator.uid_score_pairs(how="avg") + except Exception: + _latest_scores, _avg_scores = {}, {} + for uid in written: + set_miner_last_scored_round(int(uid), _rid) + try: + _samples = score_aggregator.record_count(int(uid)) + except Exception: + _samples = None + set_miner_score_snapshot( + int(uid), + latest=_latest_scores.get(int(uid)), + avg=_avg_scores.get(int(uid)), + samples=_samples, + ) + for uid in scored: + set_miner_round_delta(int(uid), float(round_scores.get(uid, 0.0))) + _ckpt_map = getattr(round_obj, "uid_to_chain_checkpoint", None) or {} + for uid, _ckpt in _ckpt_map.items(): + _repo = getattr(_ckpt, "hf_repo_id", None) + _rev = getattr(_ckpt, "hf_revision", None) + if _repo and _rev: + set_miner_evaluated_commit(int(uid), str(_repo), str(_rev), _rid) + except Exception as e: + logger.warning( + "finalize_round_scores: telemetry emission failed", + round_id=round_obj.round_id, error=str(e), + ) + logger.info( "finalize_round_scores: round scored by rank", round_id=round_obj.round_id, diff --git a/connito/validator/round.py b/connito/validator/round.py index 9670264..a3c310d 100644 --- a/connito/validator/round.py +++ b/connito/validator/round.py @@ -575,6 +575,8 @@ def _journal_snapshot_locked(self) -> dict | None: """ if self.journal_path is None: return None + from connito.validator.round_journal import commit_map_from_checkpoints + return { "round_id": self.round_id, "uid_to_hotkey": dict(self.uid_to_hotkey), @@ -584,6 +586,7 @@ def _journal_snapshot_locked(self) -> dict | None: "validation_failed_uids": tuple(sorted(self.validation_failed_uids)), "freeze_zero_uids": tuple(sorted(self.freeze_zero_uids)), "freeze_zero_hotkeys": dict(self.freeze_zero_hotkeys), + "uid_to_commit": commit_map_from_checkpoints(self.uid_to_chain_checkpoint), "finalized": False, } diff --git a/connito/validator/round_journal.py b/connito/validator/round_journal.py index 2757922..41e2250 100644 --- a/connito/validator/round_journal.py +++ b/connito/validator/round_journal.py @@ -25,7 +25,11 @@ from dataclasses import asdict, dataclass, field from pathlib import Path -SCHEMA_VERSION = 1 +# v2 adds `uid_to_commit` (uid -> (hf_repo_id, hf_revision)) so the +# journal-recovery finalize can re-publish evaluated-commit telemetry. +# `from_json` accepts v1 files (missing map -> empty) so leftover journals +# written by an older build still recover. +SCHEMA_VERSION = 2 JOURNAL_DIR_NAME = "round_journal" JOURNAL_FILENAME_PREFIX = "round_" JOURNAL_FILENAME_SUFFIX = ".json" @@ -52,6 +56,9 @@ class RoundJournal: validation_failed_uids: tuple[int, ...] = () freeze_zero_uids: tuple[int, ...] = () freeze_zero_hotkeys: dict[int, str] = field(default_factory=dict) + # v2: uid -> (hf_repo_id, hf_revision) evaluated for the round. Only + # uids whose chain checkpoint carried BOTH values are recorded. + uid_to_commit: dict[int, tuple[str, str]] = field(default_factory=dict) finalized: bool = False schema_version: int = SCHEMA_VERSION @@ -65,16 +72,23 @@ def to_json(self) -> str: payload["failed_uids"] = list(self.failed_uids) payload["validation_failed_uids"] = list(self.validation_failed_uids) payload["freeze_zero_uids"] = list(self.freeze_zero_uids) + payload["uid_to_commit"] = { + str(k): [str(v[0]), str(v[1])] for k, v in self.uid_to_commit.items() + } return json.dumps(payload) @classmethod def from_json(cls, data: str) -> "RoundJournal": raw = json.loads(data) version = int(raw.get("schema_version", 1)) - if version != SCHEMA_VERSION: + # Accept every version up to the current one: v1 files simply lack + # `uid_to_commit` (defaults to empty). Reject only FUTURE versions — + # fields this build doesn't understand could change recovery + # semantics silently. + if version > SCHEMA_VERSION: raise ValueError( f"Unsupported RoundJournal schema_version={version}; " - f"expected {SCHEMA_VERSION}" + f"this build supports <= {SCHEMA_VERSION}" ) return cls( round_id=int(raw["round_id"]), @@ -87,11 +101,31 @@ def from_json(cls, data: str) -> "RoundJournal": freeze_zero_hotkeys={ int(k): str(v) for k, v in raw.get("freeze_zero_hotkeys", {}).items() }, + uid_to_commit={ + int(k): (str(v[0]), str(v[1])) + for k, v in raw.get("uid_to_commit", {}).items() + if isinstance(v, (list, tuple)) and len(v) == 2 + }, finalized=bool(raw.get("finalized", False)), schema_version=version, ) +def commit_map_from_checkpoints( + uid_to_chain_checkpoint: dict[int, object], +) -> dict[int, tuple[str, str]]: + """Extract the journal's `uid_to_commit` map from a round's + `uid_to_chain_checkpoint`. Skips uids missing either field. + """ + out: dict[int, tuple[str, str]] = {} + for uid, ckpt in (uid_to_chain_checkpoint or {}).items(): + repo = getattr(ckpt, "hf_repo_id", None) + rev = getattr(ckpt, "hf_revision", None) + if repo and rev: + out[int(uid)] = (str(repo), str(rev)) + return out + + def journal_dir(checkpoint_path: str | os.PathLike) -> Path: """Directory holding all per-round journal files.""" return Path(checkpoint_path) / JOURNAL_DIR_NAME @@ -179,10 +213,17 @@ class _RecoveryRound: freeze_zero_hotkeys: dict[int, str] uid_to_hotkey: dict[int, str] journal_path: Path + # Same shape finalize reads off a live Round: objects exposing + # `.hf_repo_id` / `.hf_revision`. Hydrated from the journal's v2 + # `uid_to_commit` map (empty for v1 journals) so a recovered finalize + # re-publishes evaluated-commit telemetry too. + uid_to_chain_checkpoint: dict[int, object] = field(default_factory=dict) _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) @classmethod def from_journal(cls, journal: "RoundJournal", journal_path: str | os.PathLike) -> "_RecoveryRound": + from types import SimpleNamespace + return cls( round_id=int(journal.round_id), scores=dict(journal.scores), @@ -193,6 +234,10 @@ def from_journal(cls, journal: "RoundJournal", journal_path: str | os.PathLike) freeze_zero_hotkeys=dict(journal.freeze_zero_hotkeys), uid_to_hotkey=dict(journal.uid_to_hotkey), journal_path=Path(journal_path), + uid_to_chain_checkpoint={ + int(uid): SimpleNamespace(hf_repo_id=repo, hf_revision=rev) + for uid, (repo, rev) in journal.uid_to_commit.items() + }, ) def processed_uids_snapshot(self) -> tuple[set[int], set[int]]: diff --git a/connito/validator/run.py b/connito/validator/run.py index d6d6ff8..7edc947 100644 --- a/connito/validator/run.py +++ b/connito/validator/run.py @@ -187,8 +187,9 @@ def validate_hf_distribution_config(config: ValidatorConfig) -> tuple[str | None set_miner_assignment_role, set_miner_cohort_group, set_miner_last_observed_commit_block, - set_miner_score_snapshot, set_validator_identity, + note_round_series, + evict_round_series_before, track_metagraph_sync_latency, ) from datetime import datetime @@ -1092,6 +1093,22 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = "round_journal.prune_before_round failed", error=str(e), ) + # Evict per-round Prometheus labelsets on the same cutoff so + # metric retention matches on-disk retention (without this, + # every round leaves permanent {round_id} series behind). + try: + _series_evicted = evict_round_series_before(_min_round_id) + if _series_evicted: + logger.info( + "telemetry: evicted stale per-round series", + rounds=_series_evicted, + min_round_id=_min_round_id, + ) + except Exception as e: + logger.warning( + "telemetry.evict_round_series_before failed", + error=str(e), + ) logger.info( "(4) Handing weight submission to background submitter", round_id=pending_round.round_id, @@ -1120,26 +1137,16 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = ) # Mirror the about-to-submit weights into Prometheus so # external aggregators don't have to scrape `/v1/state.json` - # to learn what each validator votes on chain. Mirrors the - # semantics of `score_aggregator.uid_score_pairs(how="avg")` - # — entries are written only for UIDs we actually weight, - # so a miner the validator has never scored has *no* sample - # rather than a zero (preserves prior EMA semantics). + # to learn what each validator votes on chain. Entries are + # written only for UIDs we actually weight, so a miner the + # validator has never scored has *no* sample rather than a + # zero (preserves prior EMA semantics). # - # Same scrape also publishes the aggregator snapshot - # (latest / avg / sample count) for every UID we weight, - # so the gateway can render miner-facing telemetry without - # re-deriving from per-round samples. Best-effort throughout - # — a Prometheus failure must not block weight submission. - try: - _latest_scores = score_aggregator.uid_score_pairs(how="latest") - _avg_scores = score_aggregator.uid_score_pairs(how="avg") - except Exception as _e: - logger.warning( - "Failed to read aggregator snapshot for telemetry", - error=str(_e), - ) - _latest_scores, _avg_scores = {}, {} + # The per-miner score snapshots (latest / avg / samples / + # emitted_at) are NOT published here anymore — they moved to + # `finalize_round_scores`, which covers every verdict uid + # (not just weight recipients) and re-publishes via the + # journal-recovery replay after a restart. for _uid, _weight in uid_weights.items(): try: VALIDATOR_MINER_WEIGHT_SUBMITTED.labels( @@ -1147,16 +1154,6 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = ).set(float(_weight)) except Exception: pass - try: - _samples = score_aggregator.record_count(int(_uid)) - except Exception: - _samples = None - set_miner_score_snapshot( - int(_uid), - latest=_latest_scores.get(int(_uid)), - avg=_avg_scores.get(int(_uid)), - samples=_samples, - ) # Fire-and-forget. ChainSubmitter sets # pending_round.weights_submitted once the chain accepts the call. chain_submitter.async_submit_weight(pending_round, uid_weights) @@ -1418,6 +1415,7 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = round_ref.swap(new_current=new_round) download_window_closed.clear() try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(0) except Exception: pass @@ -1497,6 +1495,7 @@ async def _bounded_foreground_eval(): if eval_worker is not None and not eval_worker.has_eval_base_model(): eval_worker.set_eval_base_model(copy.deepcopy(global_model)) try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(2) except Exception: pass @@ -1728,6 +1727,7 @@ async def _bounded_foreground_eval(): # the post-Merge mutation of global_model does not affect it. eval_window_active.set() try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(3) except Exception: pass