perf(validator): schedule metagraph sync and metagraph_history snapshots

neurons/validator.py

@@ -28,6 +28,7 @@
 from synth.simulation_input import SimulationInput
 from synth.utils.helpers import (
     get_current_time,
+    metagraph_refresh_due,
     round_time_to_minutes,
 )
 from synth.utils.logging import print_execution_time, setup_gcp_logging
@@ -56,6 +57,14 @@
 CYCLE_SCORING = "scoring"
 CYCLE_FULL = "full"
 
+# Metagraph schedule (each cycle runs as its own process, with its own
+# timer). The chain sync keeps miner_uids/axon addresses/miner identities
+# fresh for querying; the metagraph_history snapshot feeds downstream
+# analytics that read it hourly, so minute-level snapshots are pure table
+# growth. Only the scoring process writes snapshots.
+METAGRAPH_SYNC_INTERVAL_MINUTES = 15
+METAGRAPH_SNAPSHOT_INTERVAL_MINUTES = 30
+
 
 class Validator(BaseValidatorNeuron):
     """
@@ -90,6 +99,7 @@ def __init__(self, config=None, cycle_name: str = CYCLE_LOW_FREQUENCY):
         self.price_data_provider = PriceDataProvider()
 
         self.miner_uids: list[int] = []
+        self.last_metagraph_refresh_at: datetime | None = None
         LOW_FREQUENCY.data_retention_days = self.config.retention.low.days
         HIGH_FREQUENCY.data_retention_days = self.config.retention.high.days
         LOW_FREQUENCY.cycle_interval_minutes = (
@@ -143,13 +153,9 @@ def _apply_assets_filter(self):
             f"{target_config.asset_list}"
         )
 
-    # Keep sync method for backward compatibility if needed
     def forward_validator(self):
-        """Sync entry point - runs the async version"""
-        self.miner_uids = get_available_miners_and_update_metagraph_history(
-            base_neuron=self,
-            miner_data_handler=self.miner_data_handler,
-        )
+        """Boot metagraph refresh, then run this process's cycle forever."""
+        self.refresh_metagraph(interval_minutes=0)  # boot: always due
         if self.cycle_name == CYCLE_LOW_FREQUENCY:
             SequentialScheduler(
                 prompt_config=LOW_FREQUENCY,
@@ -172,19 +178,17 @@ def cycle_low_frequency(self, asset: str):
         bt.logging.info(
             "starting the low frequency cycle", "cycle_low_frequency"
         )
-
-        # update the miners, also for the high frequency prompt that will use the same list
         self.forward_prompt(asset, LOW_FREQUENCY)
-        self.miner_uids = get_available_miners_and_update_metagraph_history(
-            base_neuron=self,
-            miner_data_handler=self.miner_data_handler,
-        )
+        # Refresh after the prompt: the scheduler fires just before a minute
+        # boundary, and a chain sync beforehand would push start_time late.
+        self.refresh_metagraph(METAGRAPH_SYNC_INTERVAL_MINUTES)
 
     @print_execution_time
     def cycle_scoring(self):
         bt.logging.info("starting the scoring cycle", "cycle_scoring")
         while True:
             self.forward_score()
+            self.refresh_metagraph(METAGRAPH_SNAPSHOT_INTERVAL_MINUTES)
             time.sleep(5)
 
     @print_execution_time
@@ -193,11 +197,31 @@ def cycle_high_frequency(self, asset: str):
             "starting the high frequency cycle", "cycle_high_frequency"
         )
         self.forward_prompt(asset, HIGH_FREQUENCY)
+        self.refresh_metagraph(METAGRAPH_SYNC_INTERVAL_MINUTES)
+
+    def refresh_metagraph(self, interval_minutes: int):
+        """Sync the chain metagraph — refreshing miner_uids and upserting
+        miner identities in the same call, so a newly queryable uid always
+        has its miners row — at most every interval_minutes. Only the
+        scoring process also appends a metagraph_history snapshot. Each
+        cycle runs as its own process, so one timestamp suffices.
+
+        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.
+        """
+        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=False,
+            save_snapshot=self.cycle_name == CYCLE_SCORING,
         )
+        if self.miner_uids:
+            self.last_metagraph_refresh_at = now
 
     @print_execution_time
     def forward_prompt(self, asset: str, prompt_config: PromptConfig):

synth/utils/helpers.py

@@ -109,6 +109,21 @@ def round_time_to_minutes(dt: datetime, extra_seconds=0) -> datetime:
     ) + timedelta(seconds=extra_seconds)
 
 
+def metagraph_refresh_due(
+    last_refresh: datetime | None,
+    now: datetime,
+    interval_minutes: int,
+) -> bool:
+    """Whether a scheduled metagraph refresh is due.
+
+    True when no refresh happened yet or interval_minutes have passed
+    since the last one. The caller stamps its own last-refresh time.
+    """
+    if last_refresh is None:
+        return True
+    return now - last_refresh >= timedelta(minutes=interval_minutes)
+
+
 def from_iso_to_unix_time(iso_time: str):
     # Convert to a datetime object
     dt = datetime.fromisoformat(iso_time).replace(tzinfo=timezone.utc)

synth/validator/forward.py

@@ -259,7 +259,7 @@ def query_available_miners_and_save_responses(
 def get_available_miners_and_update_metagraph_history(
     base_neuron: BaseValidatorNeuron,
     miner_data_handler: MinerDataHandler,
-    save: bool = True,
+    save_snapshot: bool,
 ):
     # Sync metagraph to get latest miner addresses
     base_neuron.metagraph.sync(subtensor=base_neuron.subtensor)
@@ -301,10 +301,18 @@ def get_available_miners_and_update_metagraph_history(
             }
             metagraph_info.append(metagraph_item)
 
-    if len(miners) > 0 and save:
+    # Always upsert miner identities: save_responses maps uid -> miner_id
+    # through the miners table, so a new/re-registered uid must have its row
+    # before its first queried predictions are saved (else they are dropped
+    # or attributed to the uid's previous identity). The on-conflict
+    # updated_at bump is load-bearing: it keeps a re-current identity the
+    # newest row for its uid.
+    if len(miners) > 0:
         miner_data_handler.insert_new_miners(miners)
 
-    if len(metagraph_info) > 0 and save:
+    # The metagraph_history snapshot is appended on the caller's schedule,
+    # since downstream analytics read it hourly.
+    if len(metagraph_info) > 0 and save_snapshot:
         miner_data_handler.update_metagraph_history(metagraph_info)
 
     random.shuffle(miner_uids)

tests/test_helpers.py

@@ -1,10 +1,11 @@
 import unittest
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 
 
 from synth.utils.helpers import (
     convert_prices_to_time_format,
     get_intersecting_arrays,
+    metagraph_refresh_due,
     round_time_to_minutes,
     from_iso_to_unix_time,
     get_current_time,
@@ -139,3 +140,22 @@ def test_from_iso_to_unix_time(self):
         self.assertEqual(
             from_iso_to_unix_time("2025-08-05T14:56:00+00:00"), 1754405760
         )
+
+    def test_metagraph_refresh_due(self):
+        now = datetime(2026, 7, 9, 12, 0, tzinfo=timezone.utc)
+
+        # never refreshed -> due
+        self.assertTrue(metagraph_refresh_due(None, now, 15))
+
+        # refreshed less than the interval ago -> not due
+        self.assertFalse(
+            metagraph_refresh_due(now - timedelta(minutes=14), now, 15)
+        )
+
+        # exactly the interval ago (or more) -> due
+        self.assertTrue(
+            metagraph_refresh_due(now - timedelta(minutes=15), now, 15)
+        )
+        self.assertTrue(
+            metagraph_refresh_due(now - timedelta(hours=2), now, 60)
+        )