diff --git a/connito/validator/aggregator.py b/connito/validator/aggregator.py index b618316..f5284e7 100644 --- a/connito/validator/aggregator.py +++ b/connito/validator/aggregator.py @@ -15,6 +15,21 @@ # so a restart mid-round can drop the in-flight round's partial scores. SCHEMA_VERSION = 2 +# The only score values a *finalized* round ever writes (see +# `finalize_round_scores`): the geometric rank mapping 2.25/1.5/1.0 for the +# top-3 plus 0.0 for everyone else. Any other value in the window is a *raw* +# in-cycle score (`delta ** 1.2`) that finalize is supposed to replace via +# `drop_round`. A raw value that outlives its round (an "orphan", e.g. from a +# late background eval that landed after finalize, or a round whose finalize +# never completed) rides the rolling avg as an illegal non-rank score and +# distorts weights. `sweep_orphan_raw_scores` removes such stale points. +_RANK_SCORE_VALUES: tuple[float, ...] = (0.0, 1.0, 1.5, 2.25) +_RANK_SCORE_EPS: float = 1e-9 + + +def _is_rank_score(value: float) -> bool: + return any(abs(float(value) - r) <= _RANK_SCORE_EPS for r in _RANK_SCORE_VALUES) + def _utc_now() -> datetime: return datetime.now(timezone.utc) @@ -478,6 +493,43 @@ def prune_before_round(self, min_round_id: int) -> int: dropped += state.series.prune_before_round(min_round_id) return dropped + def sweep_orphan_raw_scores(self) -> int: + """Drop every *raw* (non-rank) score point across all miners. + + A finalized round only ever stores rank values (2.25/1.5/1.0/0.0). + Any other value is a raw in-cycle `delta ** 1.2` point that + `finalize_round_scores`'s `drop_round` was supposed to replace. One + that outlives its round is an orphan: it rides the rolling avg as an + illegal non-rank score and inflates the on-chain weight derived from + it (e.g. a single 5.179 point pinning a uid's avg at ~1.29). + + Every non-rank point is dropped unconditionally — including one tagged + with the miner's highest round_id. We do NOT special-case a + "still in-flight" raw: this runs at startup right before the + round-journal recovery pass (see run.py), which replays any + unfinalized round through `finalize_round_scores`. That re-derives the + round's rank scores from the journal (`drop_round` + re-add), entirely + independent of the aggregator's pre-existing points, so a genuinely + in-flight raw is reconstructed there rather than preserved here. + Keeping the in-flight raw instead would leave a hole: an orphan that + happens to sit at the miner's newest round_id (e.g. a miner not + evaluated since) would never be swept. + + Belt-and-suspenders cleanup for orphans persisted before the + finalized-round race fix landed. Returns the number of points dropped. + """ + dropped = 0 + with self._lock: + for state in self._miners.values(): + pts = state.series.points + if not pts: + continue + kept = [p for p in pts if _is_rank_score(p[1])] + if len(kept) != len(pts): + dropped += len(pts) - len(kept) + state.series.points = kept + return dropped + def record_count(self, uid: int) -> int: """Number of recorded score points for ``uid`` (0 if unknown).""" uid = int(uid) diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 2dde2b7..b530fb1 100644 --- a/connito/validator/evaluator.py +++ b/connito/validator/evaluator.py @@ -205,12 +205,24 @@ def finalize_round_scores( """ # Snapshot all sets under the round's lock so the worker threads # cannot race a mark_scored / mark_failed against the read. - scored, _failed = round_obj.processed_uids_snapshot() + # Mark the round finalized FIRST, atomically with snapshotting its + # scores. `mark_scored` checks `round.finalized` under this same lock and + # drops any later (e.g. background) eval result, so no in-flight eval can + # add a fresh raw point to the aggregator after the `drop_round` below. + # Without this gate a late bg-eval writes a raw `delta ** 1.2` tagged with + # this round_id *after* drop_round has run; the point is never re-ranked + # and rides the rolling avg as an illegal non-rank score (the root cause + # of the inflated `validator_miner_score_avg` / weight distortion). + # `_RecoveryRound` (startup replay) has no `finalized` field; setattr + # just creates it, and recovery is single-threaded so the gate is a no-op + # there. # `round.scores` is mutated under the same lock; copy it explicitly # rather than alias. with round_obj._lock: # noqa: SLF001 — same module family + round_obj.finalized = True round_scores = dict(round_obj.scores) validation_failed = set(round_obj.validation_failed_uids) + scored, _failed = round_obj.processed_uids_snapshot() freeze_zero = set(round_obj.freeze_zero_uids) freeze_hotkeys = dict(round_obj.freeze_zero_hotkeys) diff --git a/connito/validator/round.py b/connito/validator/round.py index def4b7c..06826c4 100644 --- a/connito/validator/round.py +++ b/connito/validator/round.py @@ -93,6 +93,18 @@ class Round: freeze_zero_uids: set[int] = field(default_factory=set) freeze_zero_hotkeys: dict[int, str] = field(default_factory=dict) weights_submitted: bool = False + # Set True by `finalize_round_scores` (under `_lock`, atomically with its + # score snapshot) the instant it begins finalizing this round. Once set, + # `mark_scored` drops any further result for the round instead of + # recording it. This closes the late-eval race: a background eval that + # finishes after the eval window closed and `finalize_round_scores` has + # already run would otherwise write a *raw* in-cycle score + # (`delta ** 1.2`) to the aggregator tagged with this round_id. Finalize's + # `drop_round(round_id)` has already passed, so that point is never + # re-ranked — it survives as an illegal non-rank value in the rolling + # window, inflating the miner's `score_avg` and its on-chain weight for up + # to the full history window. See `finalize_round_scores` and `mark_scored`. + finalized: bool = False # Round-group construction scheme (gated by # `config.evaluation.enable_round_group_construction`). All default @@ -622,6 +634,11 @@ def _record_in_cycle_score(self, uid: int, hotkey: str, score: float) -> None: """ if self.score_aggregator is None: return + if self.finalized: + # Defense-in-depth: never feed the aggregator once the round is + # finalized. `mark_scored` already gates on this, but guard here + # too so any future caller cannot reintroduce an orphan raw point. + return try: self.score_aggregator.add_score( uid=uid, hotkey=hotkey, score=float(score), round_id=self.round_id, @@ -648,6 +665,21 @@ def mark_scored(self, uid: int, score: float = 0.0) -> None: """ score_f = float(score) with self._lock: + if self.finalized: + # The round was finalized (scores snapshotted, ranks written, + # drop_round already run) while this evaluation was still in + # flight. Recording it now would leave an un-ranked raw + # `delta ** 1.2` point in the aggregator that no drop_round + # ever removes — it would ride the rolling avg as an illegal + # non-rank score and inflate the miner's on-chain weight. + # Drop the stale result; release the claim so bookkeeping is + # consistent. The miner keeps its prior, correctly-ranked EMA. + self.claimed_uids.discard(uid) + logger.warning( + "Round: dropping late score for finalized round", + uid=uid, round_id=self.round_id, score=round(score_f, 6), + ) + return self.scored_uids.add(uid) self.scores[uid] = score_f self.claimed_uids.discard(uid) diff --git a/connito/validator/run.py b/connito/validator/run.py index d6d6ff8..aa004d2 100644 --- a/connito/validator/run.py +++ b/connito/validator/run.py @@ -799,6 +799,23 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = max_points=score_window, max_history_points=score_history_window, ) + # Sweep stale raw (non-rank) score points left on disk — orphans + # from a late background eval that landed after finalize, or from a + # round whose finalize never completed (e.g. a scoring failure + + # restart). These ride the rolling avg as illegal non-rank scores + # and inflate on-chain weights until they age out of the window. + _swept = score_aggregator.sweep_orphan_raw_scores() + if _swept: + logger.warning( + "Dropped orphan raw (non-rank) score points on load", + dropped=_swept, + ) + try: + score_aggregator.persist_atomic(score_path) + except Exception as e: + logger.warning( + "persist_atomic after orphan sweep failed", error=str(e), + ) _loaded_latest = score_aggregator.uid_score_pairs(how="latest") _loaded_avg = score_aggregator.uid_score_pairs(how="avg") logger.info(