From 50724d1feddf0afde2b028581266fd1bb1b306cd Mon Sep 17 00:00:00 2001 From: agessaman Date: Fri, 2 Jan 2026 09:25:31 -0800 Subject: [PATCH] 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 | 3 + modules/scheduler.py | 152 +++++++++++++----- .../service_plugins/packet_capture_service.py | 87 ++++++++-- .../service_plugins/packet_capture_utils.py | 4 +- modules/service_plugins/weather_service.py | 30 +++- 5 files changed, 213 insertions(+), 63 deletions(-) diff --git a/modules/core.py b/modules/core.py index 1725fda..aa73044 100644 --- a/modules/core.py +++ b/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") diff --git a/modules/scheduler.py b/modules/scheduler.py index 3e09b31..ecdd03e 100644 --- a/modules/scheduler.py +++ b/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)""" diff --git a/modules/service_plugins/packet_capture_service.py b/modules/service_plugins/packet_capture_service.py index 98ffbb1..57fee62 100644 --- a/modules/service_plugins/packet_capture_service.py +++ b/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: diff --git a/modules/service_plugins/packet_capture_utils.py b/modules/service_plugins/packet_capture_utils.py index 7bac3a1..85faf25 100644 --- a/modules/service_plugins/packet_capture_utils.py +++ b/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, diff --git a/modules/service_plugins/weather_service.py b/modules/service_plugins/weather_service.py index 7030a5a..229f776 100644 --- a/modules/service_plugins/weather_service.py +++ b/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).