Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 35 additions & 2 deletions cacheon/chain/weight_reveal_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand All @@ -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]
Expand All @@ -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:
Expand Down
7 changes: 7 additions & 0 deletions docs/validator-guide/settlement-and-weights.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 3 additions & 3 deletions scripts/loc_baseline.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
63 changes: 56 additions & 7 deletions tests/test_weight_reveal_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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):
Expand Down
Loading