validator: add --validator.cycle_offset_minutes for wall-clock lane scheduling

neurons/validator.py

@@ -118,40 +118,58 @@ 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:
             target_config = HIGH_FREQUENCY
         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."""
@@ -209,17 +227,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
 

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,14 @@ 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:
+        return self.subtensor.metagraph(self.config.netuid)
+
     @abstractmethod
     async def forward_miner(self, synapse: bt.Synapse) -> bt.Synapse: ...
 

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,

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]

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

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()

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()