From 29365d00c3d5607a9cc3716a7ec503f17447e20b Mon Sep 17 00:00:00 2001 From: Thykof Date: Mon, 13 Jul 2026 14:24:38 +0200 Subject: [PATCH 1/3] validator: add --validator.cycle_offset_minutes for wall-clock lane scheduling When set, the sequential scheduler anchors prompt cycles to minute boundaries where minutes-since-epoch % cycle_interval_minutes equals the offset, instead of spacing cycles relative to the previous one. This lets several validator processes cover the same asset on interleaved lanes, and an overrunning cycle re-locks to its own lane instead of drifting. Unset keeps the existing relative schedule. Only applies to the prompt cycles; scoring ignores it. Co-Authored-By: Claude Fable 5 --- neurons/validator.py | 64 ++++++++++++-------- synth/utils/config.py | 6 ++ synth/utils/sequential_scheduler.py | 34 +++++++++++ synth/validator/prompt_config.py | 1 + tests/test_sequential_scheduler.py | 91 +++++++++++++++++++++++++++++ 5 files changed, 173 insertions(+), 23 deletions(-) diff --git a/neurons/validator.py b/neurons/validator.py index fa65a54b..355d87aa 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -118,18 +118,17 @@ def __init__(self, config=None, cycle_name: str = CYCLE_LOW_FREQUENCY): ) self.cycle_name = self.config.validator.cycle_name - self._apply_assets_filter() + self._apply_cycle_overrides() - def _apply_assets_filter(self): - """Narrow the active cycle's asset_list to --validator.assets, if set. + def _apply_cycle_overrides(self): + """Apply the per-process CLI overrides to the active cycle's + PromptConfig: narrow asset_list to --validator.assets and anchor + the schedule to a wall-clock lane (--validator.cycle_offset_minutes). - No-op for the scoring cycle or when the flag is unset. Raises - ValueError on assets not in the active cycle's asset_list. + No-op for the scoring cycle or when the flags are unset. Raises + ValueError on assets not in the active cycle's asset_list, or an + offset outside [0, cycle_interval_minutes). """ - assets_filter = self.config.validator.assets - if assets_filter == "": - return - if self.cycle_name == CYCLE_LOW_FREQUENCY: target_config: PromptConfig = LOW_FREQUENCY elif self.cycle_name == CYCLE_HIGH_FREQUENCY: @@ -137,21 +136,40 @@ def _apply_assets_filter(self): else: return - requested = [a.strip() for a in assets_filter.split(",") if a.strip()] - unknown = set(requested) - set(target_config.asset_list) - if unknown: - raise ValueError( - f"--validator.assets contains unknown assets {sorted(unknown)} " - f"for {target_config.label}-frequency cycle. " - f"Valid: {target_config.asset_list}" + assets_filter = self.config.validator.assets + if assets_filter != "": + requested = [ + a.strip() for a in assets_filter.split(",") if a.strip() + ] + unknown = set(requested) - set(target_config.asset_list) + if unknown: + raise ValueError( + f"--validator.assets contains unknown assets " + f"{sorted(unknown)} for {target_config.label}-frequency " + f"cycle. Valid: {target_config.asset_list}" + ) + target_config.asset_list = [ + a for a in target_config.asset_list if a in requested + ] + bt.logging.info( + f"Restricting {target_config.label}-frequency cycle to " + f"assets: {target_config.asset_list}" + ) + + offset = self.config.validator.cycle_offset_minutes + if offset is not None: + if not 0 <= offset < target_config.cycle_interval_minutes: + raise ValueError( + f"--validator.cycle_offset_minutes {offset} must be in " + f"[0, {target_config.cycle_interval_minutes}) for the " + f"{target_config.label}-frequency cycle" + ) + target_config.cycle_offset_minutes = offset + bt.logging.info( + f"Anchoring {target_config.label}-frequency cycle to " + f"wall-clock lane {offset} " + f"(mod {target_config.cycle_interval_minutes})" ) - target_config.asset_list = [ - a for a in target_config.asset_list if a in requested - ] - bt.logging.info( - f"Restricting {target_config.label}-frequency cycle to assets: " - f"{target_config.asset_list}" - ) def forward_validator(self): """Boot metagraph refresh, then run this process's cycle forever.""" diff --git a/synth/utils/config.py b/synth/utils/config.py index f9be0930..bf5b329e 100644 --- a/synth/utils/config.py +++ b/synth/utils/config.py @@ -223,6 +223,12 @@ def add_validator_args(_, parser: argparse.ArgumentParser): default="", ) + parser.add_argument( + "--validator.cycle_offset_minutes", + type=int, + default=None, + ) + parser.add_argument( "--validator.mode", type=str, diff --git a/synth/utils/sequential_scheduler.py b/synth/utils/sequential_scheduler.py index 60165372..2c87855a 100644 --- a/synth/utils/sequential_scheduler.py +++ b/synth/utils/sequential_scheduler.py @@ -66,6 +66,12 @@ def select_delay( prompt_config: PromptConfig, first_run: bool = False, ) -> int: + offset = prompt_config.cycle_offset_minutes + if offset is not None: + return SequentialScheduler.select_grid_delay( + cycle_start_time, prompt_config, offset, first_run + ) + next_cycle = cycle_start_time next_cycle = round_time_to_minutes(next_cycle) if not first_run: @@ -83,6 +89,34 @@ def select_delay( return delay + @staticmethod + def select_grid_delay( + cycle_start_time: datetime, + prompt_config: PromptConfig, + offset: int, + first_run: bool = False, + ) -> int: + """Delay until the next minute boundary T where + minutes-since-epoch(T) % cycle_interval_minutes == offset. + """ + interval = prompt_config.cycle_interval_minutes + + now = get_current_time() + next_boundary = round_time_to_minutes(now) + boundary_minutes = int(next_boundary.timestamp()) // 60 + next_cycle = next_boundary + timedelta( + minutes=(offset - boundary_minutes) % interval + ) + + if not first_run and next_cycle - cycle_start_time > timedelta( + minutes=interval + ): + bt.logging.warning( + "Cycle overran its slot, skipping to the next one in its lane" + ) + + return int((next_cycle - now).total_seconds()) + @staticmethod def select_asset(latest_asset: str | None, asset_list: list[str]) -> str: asset = asset_list[0] diff --git a/synth/validator/prompt_config.py b/synth/validator/prompt_config.py index 6ffe34a5..62468e44 100644 --- a/synth/validator/prompt_config.py +++ b/synth/validator/prompt_config.py @@ -9,6 +9,7 @@ class PromptConfig: time_increment: int cycle_interval_minutes: int timeout_extra_seconds: int + cycle_offset_minutes: int | None = None num_simulations: int = 1000 data_retention_days: int = 30 # Density tapering: predictions on validator_requests whose start_time is diff --git a/tests/test_sequential_scheduler.py b/tests/test_sequential_scheduler.py index 7d6a4bcc..a5cfd90e 100644 --- a/tests/test_sequential_scheduler.py +++ b/tests/test_sequential_scheduler.py @@ -116,5 +116,96 @@ def test_overrun_falls_back_to_next_minute(self): warn_mock.assert_called_once() +def _grid_config(cycle_interval_minutes: int, offset: int) -> PromptConfig: + cfg = _config(cycle_interval_minutes=cycle_interval_minutes) + cfg.cycle_offset_minutes = offset + return cfg + + +class TestSelectGridDelay(unittest.TestCase): + # 2026-05-14 14:00 UTC is minute 29646120 since epoch: even, and ≡ 0 mod 5. + + def test_lanes_interleave(self): + # interval 2 at 14:00:30 → lane 0 fires at 14:02, lane 1 at 14:01. + # Together the two lanes cover every minute. + now = datetime(2026, 5, 14, 14, 0, 30, tzinfo=timezone.utc) + with patch( + "synth.utils.sequential_scheduler.get_current_time", + return_value=now, + ): + delay_lane_0 = SequentialScheduler.select_delay( + cycle_start_time=now, + prompt_config=_grid_config(cycle_interval_minutes=2, offset=0), + first_run=True, + ) + delay_lane_1 = SequentialScheduler.select_delay( + cycle_start_time=now, + prompt_config=_grid_config(cycle_interval_minutes=2, offset=1), + first_run=True, + ) + self.assertEqual(delay_lane_1, 30) + self.assertEqual(delay_lane_0, 90) + + def test_steady_state_keeps_lane(self): + # Previous cycle fired on its lane slot (14:00, lane 0 mod 2) and + # finished 70 s in → next fire is 14:02, no overrun warning. + cycle_start_time = datetime(2026, 5, 14, 14, 0, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 14, 14, 1, 10, tzinfo=timezone.utc) + with ( + patch( + "synth.utils.sequential_scheduler.get_current_time", + return_value=now, + ), + patch( + "synth.utils.sequential_scheduler.bt.logging.warning" + ) as warn_mock, + ): + delay = SequentialScheduler.select_delay( + cycle_start_time=cycle_start_time, + prompt_config=_grid_config(cycle_interval_minutes=2, offset=0), + first_run=False, + ) + self.assertEqual(delay, 50) + warn_mock.assert_not_called() + + def test_overrun_relocks_to_own_lane(self): + # Cycle started at 14:00 (lane 0 mod 2) but ran 130 s — past its + # 14:02 slot. It must skip to 14:04 (its own lane), not drift to + # the odd-minute lane, and warn about the missed slot. + cycle_start_time = datetime(2026, 5, 14, 14, 0, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 14, 14, 2, 10, tzinfo=timezone.utc) + with ( + patch( + "synth.utils.sequential_scheduler.get_current_time", + return_value=now, + ), + patch( + "synth.utils.sequential_scheduler.bt.logging.warning" + ) as warn_mock, + ): + delay = SequentialScheduler.select_delay( + cycle_start_time=cycle_start_time, + prompt_config=_grid_config(cycle_interval_minutes=2, offset=0), + first_run=False, + ) + self.assertEqual(delay, 110) + warn_mock.assert_called_once() + + def test_offset_none_keeps_relative_schedule(self): + # Guard: default configs (offset unset) must not take the grid path — + # delay stays relative to the previous cycle, not the wall clock. + now = datetime(2026, 5, 14, 14, 1, 0, tzinfo=timezone.utc) + with patch( + "synth.utils.sequential_scheduler.get_current_time", + return_value=now, + ): + delay = SequentialScheduler.select_delay( + cycle_start_time=now, + prompt_config=_config(cycle_interval_minutes=5), + first_run=False, + ) + self.assertEqual(delay, 300) + + if __name__ == "__main__": unittest.main() From 9d08fe69228b42e23958c489877366c1b85db269 Mon Sep 17 00:00:00 2001 From: Thykof Date: Mon, 13 Jul 2026 16:31:40 +0200 Subject: [PATCH 2/3] validator: survive chain-endpoint rate limiting instead of crashing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two unprotected chain-sync paths could kill the process when the public endpoint rate-limits the heavy metagraph runtime call per IP (websocket handshake rejected with HTTP 429): - BaseNeuron.__init__'s first metagraph fetch — now retried with jittered exponential backoff (tenacity, same pattern as save_responses), so a boot-time burst is waited out in-process instead of crash-looping, which kept re-triggering the limit. - Validator.refresh_metagraph — a raising sync is now swallowed: the previous miner set stays in use and the unstamped timer makes the next cycle retry. This also protects the scoring loop, which calls it outside any try/except. Co-Authored-By: Claude Fable 5 --- neurons/validator.py | 24 +++++++++++++---- synth/base/neuron.py | 25 +++++++++++++++++- tests/test_refresh_metagraph.py | 46 +++++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 6 deletions(-) create mode 100644 tests/test_refresh_metagraph.py diff --git a/neurons/validator.py b/neurons/validator.py index fa65a54b..227d7026 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -209,17 +209,31 @@ def refresh_metagraph(self, interval_minutes: int): An empty result (degenerate chain response) is applied — the shared metagraph object was just emptied by the sync, so stale uids must not outlive it — but not stamped, so the next cycle retries. + + A sync that raises (e.g. the chain endpoint rate-limiting with + HTTP 429) is swallowed: the previous miner set stays in use and, + because the timer is not stamped, the next cycle retries. This + keeps a transient chain outage from killing the process — the + boot call and the scoring loop have no other safety net. """ now = get_current_time() if not metagraph_refresh_due( self.last_metagraph_refresh_at, now, interval_minutes ): return - self.miner_uids = get_available_miners_and_update_metagraph_history( - base_neuron=self, - miner_data_handler=self.miner_data_handler, - save_snapshot=self.cycle_name == CYCLE_SCORING, - ) + try: + self.miner_uids = ( + get_available_miners_and_update_metagraph_history( + base_neuron=self, + miner_data_handler=self.miner_data_handler, + save_snapshot=self.cycle_name == CYCLE_SCORING, + ) + ) + except Exception: + bt.logging.exception( + "Metagraph refresh failed; keeping the previous miner set" + ) + return if self.miner_uids: self.last_metagraph_refresh_at = now diff --git a/synth/base/neuron.py b/synth/base/neuron.py index f0061139..def5ad47 100644 --- a/synth/base/neuron.py +++ b/synth/base/neuron.py @@ -19,6 +19,11 @@ import bittensor as bt from bittensor.core.metagraph import MetagraphMixin +from tenacity import ( + retry, + stop_after_attempt, + wait_random_exponential, +) from abc import ABC, abstractmethod @@ -80,7 +85,7 @@ def __init__(self, config=None): # The wallet holds the cryptographic key pairs for the miner. self.wallet = bt.Wallet(config=self.config) self.subtensor = bt.Subtensor(config=self.config) - self.metagraph = self.subtensor.metagraph(self.config.netuid) + self.metagraph = self._initial_metagraph_fetch() bt.logging.info(f"Wallet: {self.wallet}") bt.logging.info(f"Subtensor: {self.subtensor}") @@ -98,6 +103,24 @@ def __init__(self, config=None): ) self.step = 0 + @retry( + stop=stop_after_attempt(10), + wait=wait_random_exponential(multiplier=2, max=120), + reraise=True, + ) + def _initial_metagraph_fetch(self) -> MetagraphMixin: + """First full metagraph sync of the process. + + The public chain endpoints rate-limit the heavy metagraph runtime + call per source IP (the websocket handshake is rejected with + HTTP 429). When several neurons (re)start at once behind one IP, + an unretried fetch crashes the process and the synchronized + restarts keep re-triggering the limit. Jittered exponential + backoff waits out the burst in-process and desynchronizes the + callers instead. + """ + return self.subtensor.metagraph(self.config.netuid) + @abstractmethod async def forward_miner(self, synapse: bt.Synapse) -> bt.Synapse: ... diff --git a/tests/test_refresh_metagraph.py b/tests/test_refresh_metagraph.py new file mode 100644 index 00000000..5796375e --- /dev/null +++ b/tests/test_refresh_metagraph.py @@ -0,0 +1,46 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from neurons.validator import Validator + + +def _fake_validator(): + return SimpleNamespace( + last_metagraph_refresh_at=None, + miner_uids=[1, 2], + cycle_name="high_frequency", + miner_data_handler=None, + ) + + +class TestRefreshMetagraph(unittest.TestCase): + def test_sync_failure_keeps_previous_miner_set(self): + # A chain sync failure (e.g. endpoint rate limit) must not + # propagate — the boot call and the scoring loop have no other + # safety net — and must not stamp the timer, so the next cycle + # retries. + fake = _fake_validator() + with patch( + "neurons.validator." + "get_available_miners_and_update_metagraph_history", + side_effect=ConnectionError("server rejected: HTTP 429"), + ): + Validator.refresh_metagraph(fake, 15) + self.assertEqual(fake.miner_uids, [1, 2]) + self.assertIsNone(fake.last_metagraph_refresh_at) + + def test_successful_sync_stamps_timer(self): + fake = _fake_validator() + with patch( + "neurons.validator." + "get_available_miners_and_update_metagraph_history", + return_value=[3, 4], + ): + Validator.refresh_metagraph(fake, 15) + self.assertEqual(fake.miner_uids, [3, 4]) + self.assertIsNotNone(fake.last_metagraph_refresh_at) + + +if __name__ == "__main__": + unittest.main() From 504850a72efd3412b2ea8851ddfbebc8cbf4de39 Mon Sep 17 00:00:00 2001 From: Nathan Seva Date: Mon, 13 Jul 2026 16:36:38 +0200 Subject: [PATCH 3/3] remove commend --- synth/base/neuron.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/synth/base/neuron.py b/synth/base/neuron.py index def5ad47..6d1c8ad5 100644 --- a/synth/base/neuron.py +++ b/synth/base/neuron.py @@ -109,16 +109,6 @@ def __init__(self, config=None): reraise=True, ) def _initial_metagraph_fetch(self) -> MetagraphMixin: - """First full metagraph sync of the process. - - The public chain endpoints rate-limit the heavy metagraph runtime - call per source IP (the websocket handshake is rejected with - HTTP 429). When several neurons (re)start at once behind one IP, - an unretried fetch crashes the process and the synchronized - restarts keep re-triggering the limit. Jittered exponential - backoff waits out the burst in-process and desynchronizes the - callers instead. - """ return self.subtensor.metagraph(self.config.netuid) @abstractmethod