feat: Enhance asyncio event loop handling across modules

- Introduced a mechanism to utilize the main event loop for scheduling coroutines, preventing deadlocks when the main loop is running.
- Updated the JWT renewal interval to 12 hours, with tokens now valid for 24 hours, improving token management.
- Refactored various async function calls in the MessageScheduler, WeatherService, and PacketCaptureService to ensure consistent event loop usage.
- Improved error handling and logging for scheduled tasks, enhancing robustness and maintainability.

modules/core.py

@@ -882,6 +882,9 @@ use_zulu_time = false
         """
         self.logger.info("Starting MeshCore Bot...")
         
+        # Store reference to main event loop for scheduler thread access
+        self.main_event_loop = asyncio.get_running_loop()
+        
         # Connect to MeshCore node
         if not await self.connect():
             self.logger.error("Failed to connect to MeshCore node")

modules/scheduler.py

@@ -103,15 +103,29 @@ class MessageScheduler:
         
         import asyncio
         
-        # Create a new event loop for this thread if one doesn't exist
-        try:
-            loop = asyncio.get_event_loop()
-        except RuntimeError:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-        
-        # Run the async function in the event loop
-        loop.run_until_complete(self._send_scheduled_message_async(channel, message))
+        # Use the main event loop if available, otherwise create a new one
+        # This prevents deadlock when the main loop is already running
+        if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+            # Schedule coroutine in the running main event loop
+            future = asyncio.run_coroutine_threadsafe(
+                self._send_scheduled_message_async(channel, message),
+                self.bot.main_event_loop
+            )
+            # Wait for completion (with timeout to prevent indefinite blocking)
+            try:
+                future.result(timeout=60)  # 60 second timeout
+            except Exception as e:
+                self.logger.error(f"Error sending scheduled message: {e}")
+        else:
+            # Fallback: create new event loop if main loop not available
+            try:
+                loop = asyncio.get_event_loop()
+            except RuntimeError:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+            
+            # Run the async function in the event loop
+            loop.run_until_complete(self._send_scheduled_message_async(channel, message))
     
     async def _get_mesh_info(self) -> Dict[str, Any]:
         """Get mesh network information for scheduled messages"""
@@ -332,18 +346,30 @@ class MessageScheduler:
                     hasattr(self.bot, 'connected') and self.bot.connected):
                     # Run feed polling in async context
                     import asyncio
-                    try:
-                        loop = asyncio.get_event_loop()
-                    except RuntimeError:
-                        loop = asyncio.new_event_loop()
-                        asyncio.set_event_loop(loop)
-                    
-                    # Schedule feed polling
-                    try:
-                        loop.run_until_complete(self.bot.feed_manager.poll_all_feeds())
-                        self.logger.debug("Feed polling cycle completed")
-                    except Exception as e:
-                        self.logger.error(f"Error in feed polling cycle: {e}")
+                    if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+                        # Schedule coroutine in the running main event loop
+                        future = asyncio.run_coroutine_threadsafe(
+                            self.bot.feed_manager.poll_all_feeds(),
+                            self.bot.main_event_loop
+                        )
+                        try:
+                            future.result(timeout=120)  # 2 minute timeout for feed polling
+                            self.logger.debug("Feed polling cycle completed")
+                        except Exception as e:
+                            self.logger.error(f"Error in feed polling cycle: {e}")
+                    else:
+                        # Fallback: create new event loop if main loop not available
+                        try:
+                            loop = asyncio.get_event_loop()
+                        except RuntimeError:
+                            loop = asyncio.new_event_loop()
+                            asyncio.set_event_loop(loop)
+                        
+                        try:
+                            loop.run_until_complete(self.bot.feed_manager.poll_all_feeds())
+                            self.logger.debug("Feed polling cycle completed")
+                        except Exception as e:
+                            self.logger.error(f"Error in feed polling cycle: {e}")
                     last_feed_poll_time = time.time()
             
             # Channels are fetched once on launch only - no periodic refresh
@@ -357,13 +383,25 @@ class MessageScheduler:
                 if (hasattr(self.bot, 'channel_manager') and self.bot.channel_manager and 
                     hasattr(self.bot, 'connected') and self.bot.connected):
                     import asyncio
-                    try:
-                        loop = asyncio.get_event_loop()
-                    except RuntimeError:
-                        loop = asyncio.new_event_loop()
-                        asyncio.set_event_loop(loop)
-                    
-                    loop.run_until_complete(self._process_channel_operations())
+                    if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+                        # Schedule coroutine in the running main event loop
+                        future = asyncio.run_coroutine_threadsafe(
+                            self._process_channel_operations(),
+                            self.bot.main_event_loop
+                        )
+                        try:
+                            future.result(timeout=30)  # 30 second timeout
+                        except Exception as e:
+                            self.logger.error(f"Error processing channel operations: {e}")
+                    else:
+                        # Fallback: create new event loop if main loop not available
+                        try:
+                            loop = asyncio.get_event_loop()
+                        except RuntimeError:
+                            loop = asyncio.new_event_loop()
+                            asyncio.set_event_loop(loop)
+                        
+                        loop.run_until_complete(self._process_channel_operations())
                     self.last_channel_ops_check_time = time.time()
             
             # Process feed message queue (every 2 seconds)
@@ -374,13 +412,25 @@ class MessageScheduler:
                 if (hasattr(self.bot, 'feed_manager') and self.bot.feed_manager and 
                     hasattr(self.bot, 'connected') and self.bot.connected):
                     import asyncio
-                    try:
-                        loop = asyncio.get_event_loop()
-                    except RuntimeError:
-                        loop = asyncio.new_event_loop()
-                        asyncio.set_event_loop(loop)
-                    
-                    loop.run_until_complete(self.bot.feed_manager.process_message_queue())
+                    if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+                        # Schedule coroutine in the running main event loop
+                        future = asyncio.run_coroutine_threadsafe(
+                            self.bot.feed_manager.process_message_queue(),
+                            self.bot.main_event_loop
+                        )
+                        try:
+                            future.result(timeout=30)  # 30 second timeout
+                        except Exception as e:
+                            self.logger.error(f"Error processing message queue: {e}")
+                    else:
+                        # Fallback: create new event loop if main loop not available
+                        try:
+                            loop = asyncio.get_event_loop()
+                        except RuntimeError:
+                            loop = asyncio.new_event_loop()
+                            asyncio.set_event_loop(loop)
+                        
+                        loop.run_until_complete(self.bot.feed_manager.process_message_queue())
                     self.last_message_queue_check_time = time.time()
             
             schedule.run_pending()
@@ -421,15 +471,29 @@ class MessageScheduler:
         
         import asyncio
         
-        # Create a new event loop for this thread if one doesn't exist
-        try:
-            loop = asyncio.get_event_loop()
-        except RuntimeError:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-        
-        # Run the async function in the event loop
-        loop.run_until_complete(self._send_interval_advert_async())
+        # Use the main event loop if available, otherwise create a new one
+        # This prevents deadlock when the main loop is already running
+        if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+            # Schedule coroutine in the running main event loop
+            future = asyncio.run_coroutine_threadsafe(
+                self._send_interval_advert_async(),
+                self.bot.main_event_loop
+            )
+            # Wait for completion (with timeout to prevent indefinite blocking)
+            try:
+                future.result(timeout=60)  # 60 second timeout
+            except Exception as e:
+                self.logger.error(f"Error sending interval advert: {e}")
+        else:
+            # Fallback: create new event loop if main loop not available
+            try:
+                loop = asyncio.get_event_loop()
+            except RuntimeError:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+            
+            # Run the async function in the event loop
+            loop.run_until_complete(self._send_interval_advert_async())
     
     async def _send_interval_advert_async(self):
         """Send an interval-based advert (async implementation)"""

modules/service_plugins/packet_capture_service.py

@@ -144,8 +144,8 @@ class PacketCaptureService(BaseServicePlugin):
         self.background_tasks: List[asyncio.Task] = []
         self.should_exit = False
         
-        # JWT renewal
-        self.jwt_renewal_interval = self.get_config_int('jwt_renewal_interval', 3600)
+        # JWT renewal (default: 12 hours, tokens valid for 24 hours)
+        self.jwt_renewal_interval = self.get_config_int('jwt_renewal_interval', 43200)
         
         # Health check
         self.health_check_interval = self.get_config_int('health_check_interval', 30)
@@ -1063,7 +1063,7 @@ class PacketCaptureService(BaseServicePlugin):
                             password = token
                             self.logger.debug(
                                 f"Created auth token for {broker_config['host']} "
-                                f"(username: {username}, expires in 1 hour) "
+                                f"(username: {username}, valid for 24 hours) "
                                 f"using {'device' if use_device else 'Python'} signing"
                             )
                         else:
@@ -1714,18 +1714,87 @@ class PacketCaptureService(BaseServicePlugin):
                 self.logger.error(f"Error publishing status to MQTT: {e}")
     
     async def jwt_renewal_scheduler(self) -> None:
-        """Background task to check and renew JWT tokens.
-        
-        Periodically checks if JWT tokens need renewal.
+        """Background task to proactively renew JWT tokens before expiration.
+
+        Renews auth tokens for all MQTT brokers that use token authentication.
+        Runs every jwt_renewal_interval seconds (default: 12 hours).
+        Tokens are valid for 24 hours, so this provides a 12-hour buffer.
         """
         if self.jwt_renewal_interval <= 0:
             return
-        
+
         while not self.should_exit:
             try:
                 await asyncio.sleep(self.jwt_renewal_interval)
-                # JWT renewal logic would go here if needed
-                # For now, tokens are created on-demand
+
+                if self.should_exit:
+                    break
+
+                # Renew tokens for all MQTT brokers that use auth tokens
+                for mqtt_client_info in self.mqtt_clients:
+                    config = mqtt_client_info['config']
+                    client = mqtt_client_info['client']
+
+                    # Only renew for brokers that use auth tokens
+                    if not config.get('use_auth_token'):
+                        continue
+
+                    try:
+                        broker_host = config.get('host', 'unknown')
+                        self.logger.debug(f"Renewing auth token for MQTT broker {broker_host}...")
+
+                        # Get device's public key
+                        device_public_key_hex = None
+                        if self.meshcore and hasattr(self.meshcore, 'self_info'):
+                            try:
+                                self_info = self.meshcore.self_info
+                                if isinstance(self_info, dict):
+                                    device_public_key_hex = self_info.get('public_key', '')
+                                elif hasattr(self_info, 'public_key'):
+                                    device_public_key_hex = self_info.public_key
+
+                                # Convert to hex string if bytes
+                                if isinstance(device_public_key_hex, bytes):
+                                    device_public_key_hex = device_public_key_hex.hex()
+                                elif isinstance(device_public_key_hex, bytearray):
+                                    device_public_key_hex = bytes(device_public_key_hex).hex()
+                            except Exception as e:
+                                self.logger.debug(f"Could not get public key from device: {e}")
+
+                        if not device_public_key_hex:
+                            self.logger.warning(f"No device public key available for token renewal (broker: {broker_host})")
+                            continue
+
+                        # Create new auth token
+                        token_audience = config.get('token_audience') or broker_host
+                        username = f"v1_{device_public_key_hex.upper()}"
+
+                        use_device = (self.auth_token_method == 'device' and
+                                     self.meshcore and
+                                     self.meshcore.is_connected)
+                        meshcore_for_key_fetch = self.meshcore if self.meshcore and self.meshcore.is_connected else None
+
+                        token = await create_auth_token_async(
+                            meshcore_instance=meshcore_for_key_fetch,
+                            public_key_hex=device_public_key_hex,
+                            private_key_hex=self.private_key_hex,
+                            iata=self.global_iata,
+                            audience=token_audience,
+                            owner_public_key=self.owner_public_key,
+                            owner_email=self.owner_email,
+                            use_device=use_device
+                        )
+
+                        if token:
+                            # Update client credentials with new token
+                            client.username_pw_set(username, token)
+                            self.logger.info(f"✓ Renewed auth token for MQTT broker {broker_host} (valid for 24 hours)")
+                        else:
+                            self.logger.warning(f"Failed to renew auth token for MQTT broker {broker_host}")
+
+                    except Exception as e:
+                        self.logger.error(f"Error renewing token for MQTT broker {config.get('host', 'unknown')}: {e}")
+
             except asyncio.CancelledError:
                 break
             except Exception as e:

modules/service_plugins/packet_capture_utils.py

@@ -540,9 +540,9 @@ async def create_auth_token_async(
     public_key_hex = public_key_hex.replace('0x', '').replace(' ', '').upper()
     
     # Create JWT payload
-    # Default expiration: 1 hour from now (3600 seconds)
+    # Default expiration: 24 hours from now (86400 seconds)
     if exp is None:
-        exp = timestamp + 3600
+        exp = timestamp + 86400
     
     payload_dict = {
         'publicKey': public_key_hex,

modules/service_plugins/weather_service.py

@@ -335,14 +335,28 @@ class WeatherService(BaseServicePlugin):
         
         self.logger.info(f"📅 Sending daily weather forecast at {datetime.now().strftime('%H:%M:%S')}")
         
-        # Run async function in event loop
-        try:
-            loop = asyncio.get_event_loop()
-        except RuntimeError:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-        
-        loop.run_until_complete(self._send_daily_forecast_async())
+        # Use the main event loop if available, otherwise create a new one
+        # This prevents deadlock when the main loop is already running
+        if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running():
+            # Schedule coroutine in the running main event loop
+            future = asyncio.run_coroutine_threadsafe(
+                self._send_daily_forecast_async(),
+                self.bot.main_event_loop
+            )
+            # Wait for completion (with timeout to prevent indefinite blocking)
+            try:
+                future.result(timeout=120)  # 2 minute timeout for weather forecast
+            except Exception as e:
+                self.logger.error(f"Error sending daily weather forecast: {e}")
+        else:
+            # Fallback: create new event loop if main loop not available
+            try:
+                loop = asyncio.get_event_loop()
+            except RuntimeError:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+            
+            loop.run_until_complete(self._send_daily_forecast_async())
     
     async def _send_daily_forecast_async(self) -> None:
         """Send daily weather forecast (async implementation).