investigate async issue

Author: eclipsevortex

subvortex/core/core_bittensor/subtensor/subtensor.py

@@ -294,6 +294,8 @@ async def watchdog():
                             f"No block received in the last {timeout} seconds"
                         )
 
+            btul.logging.trace("🚀 Starting block subscription and watchdog.", prefix="ReliableSubtensor")
+
             # Run handler + watchdog concurrently
             await asyncio.gather(
                 subtensor.substrate._get_block_handler(
@@ -313,10 +315,12 @@ async def watchdog():
                 prefix="ReliableSubtensor",
             )
 
+            btul.logging.info("🔁 Reinstantiating substrate connection...", prefix="ReliableSubtensor")
             await subtensor.substrate._reinstantiate_substrate()
 
             if attempt > 0:
                 # Wait avg time of a block
+                btul.logging.debug("⏲️ Waiting before retrying subscription...", prefix="ReliableSubtensor")
                 await asyncio.sleep(12)
 
             attempt += 1

subvortex/core/database/database.py

@@ -1,107 +1,123 @@
+import asyncio
 from redis import asyncio as aioredis
 from packaging.version import parse as parse_version
+from weakref import WeakKeyDictionary
 
 import bittensor.utils.btlogging as btul
-
 from subvortex.core.database.database_utils import decode_value
 
 
 class Database:
     def __init__(self, settings):
         self.models = {}
         self.settings = settings
-        self.database = None
+        self._clients = WeakKeyDictionary()  # Cache clients per event loop
 
-    async def connect(self):
-        self.database = aioredis.StrictRedis(
+    def _new_client(self):
+        return aioredis.StrictRedis(
             host=self.settings.database_host,
             port=self.settings.database_port,
             db=self.settings.database_index,
             password=self.settings.database_password,
         )
 
-        btul.logging.info("Connected to Redis", prefix=self.settings.logging_name)
+    def _get_loop(self):
+        return asyncio.get_running_loop()
+
+    async def get_client(self):
+        loop = self._get_loop()
+
+        if loop in self._clients:
+            return self._clients[loop]
+
+        client = self._new_client()
+        self._clients[loop] = client
+
+        btul.logging.info(
+            "Created new Redis client for event loop", prefix=self.settings.logging_name
+        )
+        return client
 
     async def is_connection_alive(self) -> bool:
+        client = await self.get_client()
+
         try:
-            pong = await self.database.ping()
+            pong = await client.ping()
             return pong is True
         except Exception as e:
-            btul.logging.warning(f"Redis connection check failed: {e}")
+            btul.logging.warning(
+                f"Redis connection check failed: {e}", prefix=self.settings.logging_name
+            )
             return False
 
     async def ensure_connection(self):
-        if self.database is None or not await self.is_connection_alive():
+        client = await self.get_client()
+
+        if not await self.is_connection_alive():
             btul.logging.warning(
-                "Reconnecting to Redis...",
+                "Redis ping failed, but client will be reused",
                 prefix=self.settings.logging_name,
             )
-            await self.connect()
+            # You may optionally recreate here if needed
 
     async def wait_until_ready(self, name: str):
-        # Ensure the connection is ip and running
         await self.ensure_connection()
 
+        client = await self.get_client()
+
         message_key = self._key(f"state:{name}")
         stream_key = self._key(f"state:{name}:stream")
         last_id = "$"
 
         try:
-            # Step 1: check the message key first
-            snapshot = await self.database.get(message_key)
+            snapshot = await client.get(message_key)
             if snapshot and snapshot.decode() == "ready":
-                btul.logging.trace(
+                btul.logging.debug(
                     f"{name} is already ready (via message key)",
                     prefix=self.settings.logging_name,
                 )
                 return
 
-            # Step 2: wait for stream messages
-            btul.logging.trace(
+            btul.logging.debug(
                 f"Waiting on stream: {stream_key}", prefix=self.settings.logging_name
             )
             while True:
-                entries = await self.database.xread({stream_key: last_id}, block=0)
+                entries = await client.xread({stream_key: last_id}, block=0)
                 if not entries:
                     continue
 
                 for stream_key, messages in entries:
-                    btul.logging.trace(
+                    btul.logging.debug(
                         f"Received stream message: {messages}",
                         prefix=self.settings.logging_name,
                     )
                     for msg_id, fields in messages:
-                        state = fields.get("state".encode(), b"").decode()
+                        state = fields.get(b"state", b"").decode()
                         if state == "ready":
-                            btul.logging.trace(
+                            btul.logging.debug(
                                 f"{name} is now ready (via stream)",
                                 prefix=self.settings.logging_name,
                             )
                             return
-                        last_id = msg_id  # move forward
+                        last_id = msg_id
         except Exception as err:
             btul.logging.warning(
                 f"Failed to read the state of {name}: {err}",
                 prefix=self.settings.logging_name,
             )
 
     async def _get_migration_status(self, model_name: str):
-        """
-        Returns:
-            - latest_version: the 'new' version
-            - active_versions: versions marked 'dual' or 'new',
-            or fallback to latest if none are active.
-        """
-        # Ensure the connection is ip and running
         await self.ensure_connection()
+        
+        client = await self.get_client()
 
         latest = None
         active = []
 
         all_versions = sorted(self.models[model_name].keys(), key=parse_version)
 
         for version in all_versions:
-            mode = await self.database.get(f"migration_mode:{version}")
+            mode = await client.get(f"migration_mode:{version}")
             mode = decode_value(mode)
 
             if mode == "new":

subvortex/core/file/file_monitor.py

@@ -16,12 +16,11 @@
 # DEALINGS IN THE SOFTWARE.
 import asyncio
 import threading
-import bittensor.utils.btlogging as btul
 from enum import Enum
 
+import bittensor.utils.btlogging as btul
 from subvortex.core.file.file_provider import FileProvider
 
-
 LOGGER_NAME = "File Monitoring"
 
 
@@ -32,21 +31,25 @@ class FileType(Enum):
 
 class FileMonitor(threading.Thread):
     def __init__(self):
-        super().__init__()
+        super().__init__(daemon=True)
         self.stop_flag = threading.Event()
         self.last_error_shown = None
         self.coroutines = []
-        self.loop = asyncio.new_event_loop()
+        self.loop = None
 
     def add_file_provider(self, file_provider: FileProvider):
-        task = self.loop.create_task(self._check_file(file_provider))
-        self.coroutines.append(task)
+        if self.loop:
+            task = asyncio.run_coroutine_threadsafe(
+                self._check_file(file_provider), self.loop
+            )
+            self.coroutines.append(task)
 
     async def _check_file(self, file: FileProvider):
-        while not self.stop_flag.is_set():
-            try:
-                # Wait a specific time before starting
+        try:
+            while not self.stop_flag.is_set():
                 await asyncio.sleep(file.check_interval)
+                if self.stop_flag.is_set():
+                    break
 
                 btul.logging.debug(
                     f"[{LOGGER_NAME}][{file.logger_name}] Checking file..."
@@ -70,45 +73,38 @@ async def _check_file(self, file: FileProvider):
 
                 # Reset the last error shown
                 self.last_error_shown = None
-            except Exception as err:
-                error_message = f"[{LOGGER_NAME}][{file.logger_name}] Failed processing file: {err} {type(err)}"
-                if error_message != self.last_error_shown:
-                    btul.logging.error(error_message)
-                    self.last_error_shown = error_message
-
-    async def _run_async(self):
-        while not self.stop_flag.is_set():
-            try:
-                # Sleep for a second before gathering tasks
-                await asyncio.sleep(1)
 
-                if self.stop_flag.is_set():
-                    # Time to stop the file monitoring
-                    # We wait until all the tasks are finished
-                    await asyncio.gather(*self.coroutines)
-            except Exception as err:
-                error_message = (
-                    f"[{LOGGER_NAME}] Failed checking files: {err} {type(err)}"
-                )
-                if error_message != self.last_error_shown:
-                    btul.logging.error(error_message)
-                    self.last_error_shown = error_message
+        except Exception as err:
+            error_message = f"[{LOGGER_NAME}][{file.logger_name}] Failed processing file: {err} {type(err)}"
+            if error_message != self.last_error_shown:
+                btul.logging.error(error_message)
+                self.last_error_shown = error_message
 
     def run(self):
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+
+        async def monitor():
+            await asyncio.Event().wait()  # Wait forever unless externally cancelled
+
         try:
-            self.loop.run_until_complete(self._run_async())
+            self.loop.run_until_complete(monitor())
+        except Exception as err:
+            btul.logging.error(f"[{LOGGER_NAME}] Loop error: {err}")
         finally:
-            self.loop.stop()
-            self.loop.run_until_complete(self.loop.shutdown_asyncgens())
+            pending = asyncio.all_tasks(loop=self.loop)
+            for task in pending:
+                task.cancel()
+            self.loop.run_until_complete(
+                asyncio.gather(*pending, return_exceptions=True)
+            )
             self.loop.close()
-
-        btul.logging.debug(f"[{LOGGER_NAME}] run ended")
-
-    def start(self):
-        super().start()
-        btul.logging.debug(f"[{LOGGER_NAME}] started")
+            btul.logging.debug(f"[{LOGGER_NAME}] Event loop closed")
 
     def stop(self):
+        btul.logging.info(f"[{LOGGER_NAME}] FileMonitor stopping")
         self.stop_flag.set()
-        super().join()
-        btul.logging.debug(f"[{LOGGER_NAME}] stopped")
+        if self.loop:
+            self.loop.call_soon_threadsafe(self.loop.stop)
+        self.join()
+        btul.logging.info(f"[{LOGGER_NAME}] FileMonitor stopped")