diff --git a/cacheon/chain/weight_reveal_monitor.py b/cacheon/chain/weight_reveal_monitor.py index 6d961140..3621f248 100644 --- a/cacheon/chain/weight_reveal_monitor.py +++ b/cacheon/chain/weight_reveal_monitor.py @@ -22,6 +22,19 @@ def send_discord(webhook: str, message: str) -> None: response.read() +def _send_info(state, notify): + """Retry informational deliveries without losing successful sends or stopping checks.""" + queue = state.setdefault("info_messages", []) + while queue: + try: + notify(queue[0]) + except Exception as exc: + print(f"weight-reveal-monitor info delivery failed: {type(exc).__name__}", flush=True) + break + queue.pop(0) + return state + + def check_reveals(subtensor, journal: Path, state: dict, *, deadline_blocks: int, notify) -> dict: """Read commit receipts and weights; write only the caller's monitor state.""" state = json.loads(json.dumps(state)) @@ -45,26 +58,46 @@ def check_reveals(subtensor, journal: Path, state: dict, *, deadline_blocks: int "block": record["submit_block"]}) state["cursor"] = sequence if not pending: - return state + return _send_info(state, notify) from cacheon.chain.weight_projection import WeightProjection latest = WeightProjection.from_dict(pending[-1]["projection"]) live = chain.fetch_metagraph(subtensor, latest.netuid) observed = chain.read_validator_weight_snapshot( subtensor, latest.netuid, latest.validator_hotkey, metagraph_view=live) + known = set(state.setdefault("known_hotkeys", sorted(observed.weights))) + awaiting = state.setdefault("hotkeys_awaiting_activation", {}) + info = state.setdefault("info_messages", []) matched = 0 + matched_projection = None for item in pending: projection = WeightProjection.from_dict(item["projection"]) if (projection.netuid != latest.netuid or projection.validator_hotkey != latest.validator_hotkey): raise ValueError("monitor journal contains multiple signers or subnets") # Chain weights use uint16 precision and omit zero-rounded dust recipients. expected = projection.weights + for hotkey in sorted(set(expected) - known): + info.append(f"Cacheon SN{projection.netuid} INFO: weights committed for new hotkey " + f"{hotkey} (UID {live.uid_of(hotkey)}), allocation {expected[hotkey]:.4%}. " + f"Commit submitted at block {item['block']}; awaiting activation.") + known.add(hotkey) + awaiting[hotkey] = item["block"] if (observed.last_update_block >= item["block"] and all(math.isclose(observed.weights.get(k, 0), expected.get(k, 0), rel_tol=2e-5, abs_tol=2e-5) for k in set(expected) | set(observed.weights))): matched = item["sequence"] + matched_projection = projection + state["known_hotkeys"] = sorted(known) if matched: + for hotkey, block in list(awaiting.items()): + if (hotkey in matched_projection.weights and observed.weights.get(hotkey, 0) > 0 + and observed.last_update_block >= block): + info.append(f"Cacheon SN{latest.netuid} INFO: weights are now active for hotkey " + f"{hotkey} (UID {live.uid_of(hotkey)}), observed share " + f"{observed.weights[hotkey]:.4%} at block {live.block}. " + "This confirms this validator's weights, not network-wide emissions.") + del awaiting[hotkey] # A later revealed allocation supersedes earlier commits. They must not # raise a false alarm just because polling never saw their intermediate row. pending[:] = [item for item in pending if item["sequence"] > matched] @@ -81,7 +114,7 @@ def check_reveals(subtensor, journal: Path, state: dict, *, deadline_blocks: int f"last weight update {observed.last_update_block}. Publication is NOT paused.") state["alerted"] = True state["checked_block"] = live.block - return state + return _send_info(state, notify) def main() -> None: diff --git a/docs/validator-guide/settlement-and-weights.md b/docs/validator-guide/settlement-and-weights.md index c5fa58ad..0eb3d49f 100644 --- a/docs/validator-guide/settlement-and-weights.md +++ b/docs/validator-guide/settlement-and-weights.md @@ -602,6 +602,13 @@ revealed allocation superseding an intermediate one. On first start it checks the latest accepted commit; persisted state preserves outstanding checks across restarts. Repeated check failures also alert. Neither reveal failures nor Discord delivery failures stop, hold, release, or otherwise control the publisher. +For a newly weighted hotkey it sends an informational message after an accepted +commit, then another when the matching allocation is observed active on-chain. +Each message names the hotkey, UID, share, and block. Already active recipients +seed its initial inventory; subsequent share changes and ordinary restarts do +not repeat the new-hotkey announcement. Undelivered info messages remain queued +in the monitor's own state. Activation describes this validator's weight row, +not network-wide emissions. The legacy combined `set-weights` reconciler described above remains separate. When `serve-weights` is deliberately run without push credentials, raw local diff --git a/scripts/loc_baseline.txt b/scripts/loc_baseline.txt index 1e8c8ce3..2a504d3d 100644 --- a/scripts/loc_baseline.txt +++ b/scripts/loc_baseline.txt @@ -4,10 +4,10 @@ (root) 1161 .github 345 bundles 2985 -cacheon 90935 +cacheon 90968 cacheon_kernels 936 dashboard 4318 -docs 12569 +docs 12576 examples 535 scripts 1894 -tests 67377 +tests 67426 diff --git a/tests/test_weight_reveal_monitor.py b/tests/test_weight_reveal_monitor.py index f4488079..511993c4 100644 --- a/tests/test_weight_reveal_monitor.py +++ b/tests/test_weight_reveal_monitor.py @@ -28,17 +28,19 @@ def test_missing_reveal_alerts_once_then_reports_recovery_without_writing_journa original = journal.read_bytes() live, messages = Chain(), [] state = check_reveals(live, journal, {}, deadline_blocks=20, notify=messages.append) - assert not messages and len(state["pending"]) == 1 + assert len(messages) == 1 and "committed for new hotkey alice" in messages[0] + assert len(state["pending"]) == 1 live.block = 120 state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) - assert len(messages) == 1 and "NOT paused" in messages[0] + assert len(messages) == 2 and "NOT paused" in messages[1] state = json.loads(json.dumps(state)) # persisted restart state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) - assert len(messages) == 1 + assert len(messages) == 2 live.install(projection.weights, update=101) state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) assert not state["pending"] and not state["alerted"] - assert "recovered" in messages[-1] and len(messages) == 2 + assert "recovered" in messages[-2] and "now active for hotkey alice" in messages[-1] + assert len(messages) == 4 assert journal.read_bytes() == original and live.submit_calls == 0 @@ -49,7 +51,9 @@ def test_chain_drops_zero_rounded_dust_without_false_alert(tmp_path): live.install({"alice": 1}, update=101) messages = [] state = check_reveals(live, journal, {}, deadline_blocks=20, notify=messages.append) - assert not state["pending"] and not messages + assert not state["pending"] and len(messages) == 1 + assert "committed for new hotkey bob" in messages[0] + assert "now active" not in messages[0] def test_later_reveal_supersedes_intermediate_commits(tmp_path): @@ -62,7 +66,8 @@ def test_later_reveal_supersedes_intermediate_commits(tmp_path): live.install({"bob": 1}, update=102) messages = [] state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) - assert not state["pending"] and not messages and state["last_revealed_sequence"] == 2 + assert not state["pending"] and state["last_revealed_sequence"] == 2 + assert len(messages) == 2 and all("hotkey bob" in message for message in messages) def test_failed_alert_delivery_can_retry_without_losing_pending_receipt(tmp_path): @@ -78,7 +83,51 @@ def unavailable(_): assert state == {} messages = [] check_reveals(Chain(block=130), journal, state, deadline_blocks=20, notify=messages.append) - assert len(messages) == 1 + assert len(messages) == 2 and "NOT paused" in messages[0] + + +def test_hotkey_info_retries_delivery_and_survives_restart_without_reannouncing(tmp_path): + journal = tmp_path / "signer.sqlite3" + projection = receipt(journal, weights=(("alice", 500_000), ("bob", 500_000))) + live = Chain() + live.install({"alice": 1}, update=90) + + def offline(_): + raise OSError("Discord offline") + + state = check_reveals(live, journal, {}, deadline_blocks=20, notify=offline) + assert len(state["info_messages"]) == 1 + live.block = 101 + live.install(projection.weights, update=101) + messages = [] + + def deliver_commit_only(message): + if "now active" in message: + raise OSError("Discord offline again") + messages.append(message) + + state = check_reveals(live, journal, state, deadline_blocks=20, notify=deliver_commit_only) + assert len(messages) == 1 and "committed for new hotkey bob" in messages[0] + assert not state["pending"] and len(state["info_messages"]) == 1 + state = json.loads(json.dumps(state)) + state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) + assert len(messages) == 2 and "now active for hotkey bob" in messages[1] + assert not state["info_messages"] + receipt(journal, sequence=2, block=102, weights=(("alice", 250_000), ("bob", 750_000))) + live.block = 102 + live.install({"alice": .25, "bob": .75}, update=102) + state = check_reveals(live, journal, state, deadline_blocks=20, notify=messages.append) + assert len(messages) == 2 and not state["pending"] + + +def test_initial_active_recipients_do_not_generate_new_hotkey_info(tmp_path): + journal = tmp_path / "signer.sqlite3" + projection = receipt(journal) + live = Chain(block=101) + live.install(projection.weights, update=101) + messages = [] + state = check_reveals(live, journal, {}, deadline_blocks=20, notify=messages.append) + assert not messages and state["known_hotkeys"] == ["alice"] def test_discord_alert_disables_mentions_and_has_bounded_timeout(monkeypatch):