Add syncing metagraph and remove manual script

Author: stephen-ridges

api/src/endpoints/scoring.py

@@ -17,7 +17,7 @@
 from api.src.backend.queries.agents import get_top_agent, ban_agents as db_ban_agents, approve_agent_version, get_agent_by_version_id as db_get_agent_by_version_id
 from api.src.backend.entities import MinerAgentScored
 from api.src.backend.db_manager import get_transaction, new_db, get_db_connection
-from api.src.utils.refresh_subnet_hotkeys import check_if_hotkey_is_registered
+from api.src.utils.subtensor import get_subnet_hotkeys, check_if_hotkey_is_registered
 from api.src.utils.slack import notify_unregistered_top_miner, notify_unregistered_treasury_hotkey
 from api.src.backend.internal_tools import InternalTools
 from api.src.backend.entities import TreasuryTransaction
@@ -71,7 +71,7 @@ async def get_treasury_hotkey():
             raise ValueError("No active treasury wallets found in database")
         treasury_hotkey = treasury_hotkey_data[0]["hotkey"]
 
-        if not check_if_hotkey_is_registered(treasury_hotkey):
+        if not await check_if_hotkey_is_registered(treasury_hotkey):
             logger.error(f"Treasury hotkey {treasury_hotkey} not registered on subnet")
             await notify_unregistered_treasury_hotkey(treasury_hotkey)
 
@@ -100,7 +100,7 @@ async def weights() -> Dict[str, float]:
     if top_agent.miner_hotkey.startswith("open-"):
         weights[treasury_hotkey] = weight_left
     else:
-        if check_if_hotkey_is_registered(top_agent.miner_hotkey):
+        if await check_if_hotkey_is_registered(top_agent.miner_hotkey):
             weights[top_agent.miner_hotkey] = weight_left
         else:
             logger.error(f"Top agent {top_agent.miner_hotkey} not registered on subnet")

api/src/main.py

@@ -24,6 +24,7 @@
 from api.src.endpoints.benchmarks import router as benchmarks_router
 from api.src.utils.slack import send_slack_message
 from api.src.utils.config import WHITELISTED_VALIDATOR_IPS
+from api.src.utils.hotkey_subscription import start_hotkey_subscription, stop_hotkey_subscription
 
 logger = get_logger(__name__)
 
@@ -56,8 +57,13 @@ async def lifespan(app: FastAPI):
     # Start background tasks
     asyncio.create_task(run_weight_setting_loop(30))
 
+    # Start hotkey subscription service
+    asyncio.create_task(start_hotkey_subscription())
+    
     yield
 
+    # Stop hotkey subscription service
+    await stop_hotkey_subscription()
     await new_db.close()
 
 app = FastAPI(lifespan=lifespan)

api/src/utils/hotkey_subscription.py

@@ -0,0 +1,144 @@
+import asyncio
+import json
+import os
+import time
+import threading
+from typing import Optional
+from substrateinterface import SubstrateInterface
+from pathlib import Path
+
+from loggers.logging_utils import get_logger
+
+logger = get_logger(__name__)
+
+NETUID = os.getenv("NETUID", "62")
+SUBTENSOR_URL = os.getenv("SUBTENSOR_ADDRESS", "ws://127.0.0.1:9945")
+CACHE_FILE = Path("subnet_hotkeys_cache.json")
+
+class HotkeySubscriptionManager:
+    def __init__(self):
+        self.is_running = False
+        self.main_loop: Optional[asyncio.AbstractEventLoop] = None
+        
+    async def start_subscription(self) -> None:
+        if self.is_running:
+            return
+            
+        self.is_running = True
+        self.main_loop = asyncio.get_running_loop()
+        logger.info("Starting hotkey subscription service")
+        
+        await self._update_cache()
+        
+        asyncio.create_task(self._subscription_loop())
+        
+    async def stop_subscription(self) -> None:
+        self.is_running = False
+        logger.info("Stopped hotkey subscription service")
+        
+    async def _subscription_loop(self) -> None:
+        retry_delay = 1.0
+        
+        while self.is_running:
+            try:
+                await self._run_subscription()
+                retry_delay = 1.0  # Reset on success
+            except Exception as e:
+                logger.error(f"Subscription failed: {e}")
+                if retry_delay < 60:
+                    retry_delay *= 2
+                await asyncio.sleep(retry_delay)
+                
+    async def _run_subscription(self) -> None:
+        substrate = SubstrateInterface(
+            url=SUBTENSOR_URL,
+            ss58_format=42,
+            type_registry_preset="substrate-node-template"
+        )
+        
+        try:
+            storage_key = substrate.create_storage_key("SubtensorModule", "Uids", [NETUID])
+            
+            def handler(storage_key, obj, update_nr, subscription_id):
+                if not self.is_running:
+                    return True
+                if self.main_loop and not self.main_loop.is_closed():
+                    asyncio.run_coroutine_threadsafe(self._update_cache(), self.main_loop)
+                return None
+            
+            # Run in thread to avoid blocking
+            def subscription_thread():
+                try:
+                    substrate.subscribe_storage([storage_key], handler)
+                except Exception as e:
+                    # Filters out expected errors, such as 
+                    # ERROR - Subscription error: Expecting value: line 1 column 1 (char 0)
+                    # ERROR - Subscription error: Connection closed
+                    # ERROR - Subscription error: WebSocket connection is closed
+                    if not any(x in str(e).lower() for x in ["expecting value", "json", "connection", "closed"]):
+                        logger.error(f"Subscription error: {e}")
+                finally:
+                    try:
+                        substrate.close()
+                    except:
+                        pass
+            
+            thread = threading.Thread(target=subscription_thread, daemon=True)
+            thread.start()
+            
+            while self.is_running and thread.is_alive():
+                await asyncio.sleep(1)
+                
+        finally:
+            try:
+                substrate.close()
+            except:
+                pass
+                
+    async def _update_cache(self) -> None:
+        try:
+            substrate = SubstrateInterface(
+                url=SUBTENSOR_URL,
+                ss58_format=42,
+                type_registry_preset="substrate-node-template"
+            )
+            
+            result = substrate.query_map("SubtensorModule", "Uids", [NETUID])
+            hotkeys = []
+            
+            for uid_data in result:
+                try:
+                    hotkey = uid_data[0]
+                    if hasattr(hotkey, 'value'):
+                        hotkey = hotkey.value
+                    if isinstance(hotkey, bytes):
+                        hotkey = substrate.ss58_encode(hotkey)
+                    hotkeys.append(hotkey)
+                except:
+                    continue
+                    
+            # Atomic write
+            temp_file = CACHE_FILE.with_suffix('.tmp')
+            with open(temp_file, 'w') as f:
+                json.dump({"hotkeys": hotkeys, "timestamp": time.time()}, f)
+            temp_file.replace(CACHE_FILE)
+            
+            logger.info(f"Updated cache with {len(hotkeys)} hotkeys")
+            substrate.close()
+            
+        except Exception as e:
+            logger.error(f"Failed to update cache: {e}")
+
+# Global instance
+_manager: Optional[HotkeySubscriptionManager] = None
+
+async def start_hotkey_subscription() -> None:
+    global _manager
+    if _manager is None:
+        _manager = HotkeySubscriptionManager()
+    await _manager.start_subscription()
+    
+async def stop_hotkey_subscription() -> None:
+    global _manager
+    if _manager:
+        await _manager.stop_subscription()

api/src/utils/refresh_subnet_hotkeys.py

@@ -22,8 +22,17 @@ async def get_subnet_hotkeys():
                 data = json.loads(content)
                 return data["hotkeys"]
         else:
-            logger.warning("Hotkeys cache file does not exist. Make sure refresh_subnet_hotkeys.py service is running.")
+            logger.warning("Hotkeys cache file does not exist. Make sure hotkey subscription service is running.")
             return []
     except Exception as e:
         logger.error(f"Error reading hotkeys cache: {e}")
         return []
+
+async def check_if_hotkey_is_registered(hotkey: str) -> bool:
+    """Check if a hotkey is registered on the subnet by looking in the cache."""
+    try:
+        hotkeys = await get_subnet_hotkeys()
+        return hotkey in hotkeys
+    except Exception as e:
+        logger.error(f"Error checking if hotkey is registered: {e}")
+        return False