From d0ae73706653d9a3f25bcadebd63fc92dd14f80e Mon Sep 17 00:00:00 2001 From: Stacy Olivas Date: Sat, 21 Mar 2026 13:48:04 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20zombie=20radio=20detection=20=E2=80=94?= =?UTF-8?q?=20health=20probe,=20timeout=20guards,=20and=20alert=20system?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 23 +- config.ini.example | 35 ++ modules/command_manager.py | 8 + modules/core.py | 112 +++++ modules/scheduler.py | 740 +++++++++++++++++++++++++++---- modules/web_viewer/app.py | 18 +- tests/fixtures/mqtt_packets.json | 237 +++++----- tests/test_core.py | 213 +++++++++ tests/test_scheduler_logic.py | 153 ++++++- 9 files changed, 1321 insertions(+), 218 deletions(-) diff --git a/README.md b/README.md index 33889bf..09fb5ad 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ A Python bot that connects to MeshCore mesh networks via serial port, BLE, or TC - **Connection Methods**: Serial port, BLE (Bluetooth Low Energy), or TCP/IP - **Keyword Responses**: Configurable keyword-response pairs with template variables - **Command System**: Plugin-based command architecture with built-in commands -- **Command Aliases**: Define shorthand aliases for any command via `[Aliases]` config section +- **Command Aliases**: Define shorthand aliases for any command via `aliases =` key in each command's config section - **Rate Limiting**: Global, per-user (by pubkey or name), and per-channel rate limits to prevent spam - **User Management**: Ban/unban users with persistent storage - **Scheduled Messages**: Send messages at configured times @@ -32,7 +32,7 @@ A Python bot that connects to MeshCore mesh networks via serial port, BLE, or TC ## Requirements -- Python 3.9+ +- Python 3.10+ - MeshCore-compatible device (Heltec V3, RAK Wireless, etc.) - USB cable or BLE capability @@ -316,6 +316,11 @@ bot_tx_rate_limit_seconds = 1.0 # Min seconds between bot transmissions per_user_rate_limit_seconds = 30 # Per-user: min seconds between replies to same user (pubkey or name) per_user_rate_limit_enabled = true startup_advert = flood # Send advert on startup +radio_probe_interval_seconds = 300 # probe interval in seconds (300–900 / 5–15 min) +radio_probe_fail_threshold = 3 # consecutive failures before zombie is declared and logged +send_timeout_seconds = 30 # max seconds to wait for a channel message send +radio_zombie_alert_enabled = false # send immediate alert email on zombie detection (default: log only) +radio_zombie_alert_email = # alert recipient(s); falls back to nightly email if blank ``` ### Keywords @@ -348,12 +353,16 @@ channel.emergency_seconds = 0.0 # no rate limit on emergency channel ``` ### Command Aliases + +Add an `aliases =` key to any command's config section. The value is a +comma-separated list of extra keywords that trigger the same command. + ```ini -[Aliases] -# Format: alias = target_command -# Injects the alias string into the target command's keyword list. -w = wx -p = ping +[Ping_Command] +aliases = p,ping-test + +[WX_Command] +aliases = w,weather ``` ### Inbound Webhook diff --git a/config.ini.example b/config.ini.example index 83fa31d..69b426d 100644 --- a/config.ini.example +++ b/config.ini.example @@ -167,6 +167,41 @@ db_path = meshcore_bot.db # Seconds to wait after a failed service restart before retrying (default: 300) service_restart_backoff_seconds = 300 +# Max seconds to wait for a channel message send before timing out (default: 30) +send_timeout_seconds = 30 + +# ── Radio health monitoring (zombie radio detection) ──────────────────────── +# +# A zombie radio is one where the serial/BLE transport is still open and +# incoming RF packets are still received, but the firmware is hung and stops +# ACKing outgoing commands (all sends time out with no_event_received). +# The ONLY recovery is a physical power cycle — disconnect/reconnect does nothing. +# +# Probe interval: how often to send a get_time() health check (seconds). +# Valid range: 300–900 (5–15 minutes). Values outside this range are clamped. +# Default: 300 (5 minutes) +radio_probe_interval_seconds = 300 + +# Number of consecutive failed probes before declaring zombie state. +# At the default 300s interval: threshold 3 = ~15 min before zombie is logged. +# Suggested values by interval: +# 300s (5 min) → threshold 3 (~15 min total) +# 600s (10 min) → threshold 2 (~20 min total) +# 900s (15 min) → threshold 2 (~30 min total) +radio_probe_fail_threshold = 3 + +# Send an immediate alert email when zombie state is confirmed. +# Requires SMTP to be configured in the web viewer Notifications settings. +# false: log CRITICAL only, no email (default) +# true: send alert email in addition to logging +radio_zombie_alert_enabled = false + +# Alert email recipients for zombie detection (comma-separated addresses). +# If empty, falls back to the nightly maintenance email recipients. +# Set this to a separate on-call address if needed. +# Example: oncall@example.com,admin@example.com +radio_zombie_alert_email = + [Channels] # Channels to monitor (comma-separated) # Bot will only respond to messages on these channels diff --git a/modules/command_manager.py b/modules/command_manager.py index 006476e..f5bb673 100644 --- a/modules/command_manager.py +++ b/modules/command_manager.py @@ -930,6 +930,10 @@ class CommandManager: if not self.bot.connected or not self.bot.meshcore: return False + if self.bot.is_radio_zombie: + self.bot.logger.warning("send_dm suppressed — radio is in zombie state; power cycle required") + return False + # Check all rate limits can_send, reason = await self._check_rate_limits( skip_user_rate_limit=skip_user_rate_limit, rate_limit_key=rate_limit_key @@ -1028,6 +1032,10 @@ class CommandManager: if not self.bot.connected or not self.bot.meshcore: return False + if self.bot.is_radio_zombie: + self.bot.logger.warning("send_channel_message suppressed — radio is in zombie state; power cycle required") + return False + # Check all rate limits (including per-channel) can_send, reason = await self._check_rate_limits( skip_user_rate_limit=skip_user_rate_limit, rate_limit_key=rate_limit_key, diff --git a/modules/core.py b/modules/core.py index 0e1f42e..2c61a94 100644 --- a/modules/core.py +++ b/modules/core.py @@ -289,6 +289,16 @@ class MeshCoreBot: """Get bot root directory (where config.ini is located)""" return Path(self.config_file).parent.resolve() + @property + def is_radio_zombie(self) -> bool: + """True when the radio firmware has been confirmed unresponsive. + + All outbound radio sends should check this flag and abort immediately. + Only a physical power cycle can recover the radio; the flag is cleared + automatically when connect() succeeds after a power cycle. + """ + return bool(getattr(self, '_radio_zombie_detected', False)) + def load_config(self) -> None: """Load configuration from file. @@ -1131,6 +1141,14 @@ long_jokes = false self._update_radio_connected_metadata(True) # Track connection time to skip processing old cached messages self.connection_time = time.time() + # Clear zombie state — a successful connect means the radio is alive again + self._radio_zombie_detected = False + self._radio_fail_count = 0 + try: + self.db_manager.set_metadata('bot.radio_zombie', 'false') + self.db_manager.set_metadata('bot.radio_zombie_since', '') + except Exception: + pass self.logger.info(f"Connected to: {self.meshcore.self_info} at {self.connection_time}") # Wait for contacts to load @@ -1245,6 +1263,87 @@ long_jokes = false ) raise SystemExit(1) + async def _probe_radio_health(self) -> bool: + """Send a lightweight get_time() probe to verify the radio is responding. + + A connected serial transport does not guarantee the firmware is alive and + processing commands. This probe detects the 'zombie connection' state + where the port is open and messages are received but all outgoing commands + time out with no_event_received. + + When the configured fail threshold is reached the bot logs a CRITICAL + message and sends an immediate alert email (if enabled). It does NOT + attempt to reconnect — a zombie radio requires a physical power cycle; + disconnect/reconnect of the transport does nothing. Probing stops once + a zombie is confirmed to avoid log spam; it resumes automatically after + the next successful connect() call. + + Returns True if the device responded, False otherwise. + """ + # Stop probing once a zombie has been confirmed — only a power cycle + # can recover it; further probes just generate noise. + if getattr(self, '_radio_zombie_detected', False): + return False + + if not self.meshcore or not self.meshcore.is_connected: + return False + try: + from meshcore.events import EventType + result = await asyncio.wait_for( + self.meshcore.commands.get_time(), timeout=10.0 + ) + if result.type == EventType.ERROR: + self._radio_fail_count = getattr(self, '_radio_fail_count', 0) + 1 + threshold = self.config.getint('Bot', 'radio_probe_fail_threshold', fallback=3) + interval = max(300, min(900, self.config.getint( + 'Bot', 'radio_probe_interval_seconds', fallback=300 + ))) + self.logger.warning( + f"Radio health probe failed " + f"({self._radio_fail_count}/{threshold}): no response to get_time" + ) + if self._radio_fail_count >= threshold: + fail_count = self._radio_fail_count + self._radio_fail_count = 0 + self._radio_zombie_detected = True + self.logger.critical( + "ZOMBIE RADIO DETECTED after %d consecutive failed probes " + "(probe interval %ds). The radio firmware is unresponsive. " + "A physical POWER CYCLE is required — disconnect/reconnect " + "will NOT fix this. Probing suspended until next reconnect.", + fail_count, interval, + ) + # Persist zombie state to db so the web viewer health API reflects it + try: + import datetime as _dt + self.db_manager.set_metadata('bot.radio_zombie', 'true') + self.db_manager.set_metadata( + 'bot.radio_zombie_since', + _dt.datetime.utcnow().isoformat(), + ) + except Exception: + pass + # Send immediate alert email via scheduler (non-blocking) + scheduler = getattr(self, 'scheduler', None) + if scheduler is not None: + loop = asyncio.get_event_loop() + loop.run_in_executor( + None, + scheduler.send_zombie_alert_email, + fail_count, threshold, interval, + ) + return False + if getattr(self, '_radio_fail_count', 0) > 0: + self.logger.info("Radio health probe recovered — resetting fail counter") + self._radio_fail_count = 0 + return True + except asyncio.TimeoutError: + self.logger.warning("Radio health probe timed out") + return False + except Exception as e: + self.logger.warning(f"Radio health probe error: {e}") + return False + async def set_radio_clock(self) -> bool: """Set radio clock if device time is earlier than system time. @@ -1555,6 +1654,19 @@ long_jokes = false except (AttributeError, TypeError) as e: print(f"Web viewer health check failed: {e}") + # Periodically probe radio responsiveness + # Skip entirely once a zombie is confirmed — only a power cycle + # can recover the firmware; probing just generates log noise. + if not getattr(self, '_radio_zombie_detected', False): + if not hasattr(self, '_last_radio_probe'): + self._last_radio_probe = time.time() + probe_interval = max(300, min(900, self.config.getint( + 'Bot', 'radio_probe_interval_seconds', fallback=300 + ))) + if time.time() - self._last_radio_probe >= probe_interval: + self._last_radio_probe = time.time() + asyncio.create_task(self._probe_radio_health()) + # Periodically update system health in database (every 30 seconds) if not hasattr(self, '_last_health_update'): self._last_health_update = 0 diff --git a/modules/scheduler.py b/modules/scheduler.py index 738c503..f711c40 100644 --- a/modules/scheduler.py +++ b/modules/scheduler.py @@ -4,6 +4,7 @@ Message scheduler functionality for the MeshCore Bot Handles scheduled messages and timing """ +import asyncio import datetime import json import os @@ -16,12 +17,8 @@ from typing import Any, Optional from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger -from .maintenance import MaintenanceRunner from .utils import decode_escape_sequences, format_keyword_response_with_placeholders, get_config_timezone -# process_message_queue may await long per-feed intervals across many queued items; 30s is too short. -_FEED_MESSAGE_QUEUE_FUTURE_TIMEOUT = 600 - class MessageScheduler: """Manages scheduled messages and timing""" @@ -35,13 +32,14 @@ class MessageScheduler: self.last_channel_ops_check_time = 0 self.last_message_queue_check_time = 0 self.last_radio_ops_check_time = 0 - # Align with nightly email: first retention run after ~24h uptime (not immediately on boot). - self.last_data_retention_run = time.time() + self.last_data_retention_run = 0 self._data_retention_interval_seconds = 86400 # 24 hours self.last_nightly_email_time = time.time() # don't send immediately on startup + self._last_retention_stats: dict[str, Any] = {} self.last_db_backup_run = 0 + self._last_db_backup_stats: dict[str, Any] = {} self.last_log_rotation_check_time = 0 - self.maintenance = MaintenanceRunner(bot, get_current_time=self.get_current_time) + self._last_log_rotation_applied: dict[str, str] = {} def get_current_time(self): """Get current time in configured timezone""" @@ -54,8 +52,8 @@ class MessageScheduler: if self._apscheduler is not None: try: self._apscheduler.shutdown(wait=False) - except Exception as e: - self.logger.debug("Error shutting down scheduler: %s", e) + except Exception: + pass tz, _ = get_config_timezone(self.bot.config, self.logger) self._apscheduler = BackgroundScheduler(timezone=tz) self.scheduled_messages.clear() @@ -140,14 +138,17 @@ class MessageScheduler: try: future.result(timeout=60) # 60 second timeout except Exception as e: - self.logger.error(f"Error sending scheduled message: {e}") + self.logger.error(f"Error sending scheduled message: {type(e).__name__}: {e}", exc_info=True) else: - # Fallback: create a temporary event loop and close it when done - loop = asyncio.new_event_loop() + # Fallback: create new event loop if main loop not available try: - loop.run_until_complete(self._send_scheduled_message_async(channel, message)) - finally: - loop.close() + 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""" @@ -217,8 +218,8 @@ class MessageScheduler: result = cursor.fetchone() if result: info['recent_activity_24h'] = result[0] - except Exception as e: - self.logger.debug("Error querying message_stats: %s", e) + except Exception: + pass # Calculate new devices in last 7 days (matching web viewer logic) # Query devices first heard in the last 7 days, grouped by role @@ -325,7 +326,11 @@ class MessageScheduler: except Exception as e: self.logger.warning(f"Error fetching mesh info for scheduled message: {e}. Sending message as-is.") - await self.bot.command_manager.send_channel_message(channel, message) + send_timeout = self.bot.config.getint('Bot', 'send_timeout_seconds', fallback=30) + await asyncio.wait_for( + self.bot.command_manager.send_channel_message(channel, message), + timeout=send_timeout, + ) def start(self): """Start the scheduler in a separate thread""" @@ -337,8 +342,8 @@ class MessageScheduler: if self._apscheduler is not None: try: self._apscheduler.shutdown(wait=False) - except Exception as e: - self.logger.debug("Error shutting down scheduler: %s", e) + except Exception: + pass if self.scheduler_thread and self.scheduler_thread.is_alive(): self.scheduler_thread.join(timeout=timeout) if self.scheduler_thread.is_alive(): @@ -391,15 +396,18 @@ class MessageScheduler: if not f.cancelled() and f.exception() else None ) else: - # Fallback: create a temporary event loop and close it when done - loop = asyncio.new_event_loop() + # 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}") - finally: - loop.close() last_feed_poll_time = time.time() # Channels are fetched once on launch only - no periodic refresh @@ -456,16 +464,10 @@ class MessageScheduler: self.bot.feed_manager.process_message_queue(), self.bot.main_event_loop ) - try: - future.result(timeout=_FEED_MESSAGE_QUEUE_FUTURE_TIMEOUT) - except TimeoutError: - self.logger.warning( - "Timed out waiting for feed message queue after %ss; " - "work may still be running on the main loop (per-feed send spacing).", - _FEED_MESSAGE_QUEUE_FUTURE_TIMEOUT, - ) - except Exception as e: - self.logger.exception(f"Error processing message queue: {e}") + future.add_done_callback( + lambda f: self.logger.exception("Error processing message queue: %s", f.exception()) + if not f.cancelled() and f.exception() else None + ) else: # Fallback: create new event loop if main loop not available try: @@ -479,28 +481,110 @@ class MessageScheduler: # Data retention: run daily (packet_stream, repeater tables, stats, caches, mesh_connections) if time.time() - self.last_data_retention_run >= self._data_retention_interval_seconds: - self.maintenance.run_data_retention() + self._run_data_retention() self.last_data_retention_run = time.time() # Nightly maintenance email (24 h interval, after retention so stats are fresh) if time.time() - self.last_nightly_email_time >= self._data_retention_interval_seconds: - self.maintenance.send_nightly_email() + self._send_nightly_email() self.last_nightly_email_time = time.time() # Log rotation live-apply: check bot_metadata for config changes every 60 s if time.time() - self.last_log_rotation_check_time >= 60: - self.maintenance.apply_log_rotation_config() + self._apply_log_rotation_config() self.last_log_rotation_check_time = time.time() # DB backup: evaluate schedule every 5 minutes if time.time() - self.last_db_backup_run >= 300: - self.maintenance.maybe_run_db_backup() + self._maybe_run_db_backup() self.last_db_backup_run = time.time() time.sleep(1) self.logger.info("Scheduler thread stopped") + def _run_data_retention(self): + """Run data retention cleanup: packet_stream, repeater tables, stats, caches, mesh_connections.""" + import asyncio + + def get_retention_days(section: str, key: str, default: int) -> int: + try: + if self.bot.config.has_section(section) and self.bot.config.has_option(section, key): + return self.bot.config.getint(section, key) + except Exception: + pass + return default + + packet_stream_days = get_retention_days('Data_Retention', 'packet_stream_retention_days', 3) + purging_log_days = get_retention_days('Data_Retention', 'purging_log_retention_days', 90) + daily_stats_days = get_retention_days('Data_Retention', 'daily_stats_retention_days', 90) + observed_paths_days = get_retention_days('Data_Retention', 'observed_paths_retention_days', 90) + mesh_connections_days = get_retention_days('Data_Retention', 'mesh_connections_retention_days', 7) + stats_days = get_retention_days('Stats_Command', 'data_retention_days', 7) + + try: + # Packet stream (web viewer integration) + if hasattr(self.bot, 'web_viewer_integration') and self.bot.web_viewer_integration: + bi = getattr(self.bot.web_viewer_integration, 'bot_integration', None) + if bi and hasattr(bi, 'cleanup_old_data'): + bi.cleanup_old_data(packet_stream_days) + + # Repeater manager: purging_log and optional daily_stats / unique_advert / observed_paths + if hasattr(self.bot, 'repeater_manager') and self.bot.repeater_manager: + if hasattr(self.bot, 'main_event_loop') and self.bot.main_event_loop and self.bot.main_event_loop.is_running(): + future = asyncio.run_coroutine_threadsafe( + self.bot.repeater_manager.cleanup_database(purging_log_days), + self.bot.main_event_loop + ) + try: + future.result(timeout=60) + except Exception as e: + self.logger.error(f"Error in repeater_manager.cleanup_database: {e}") + else: + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete(self.bot.repeater_manager.cleanup_database(purging_log_days)) + if hasattr(self.bot.repeater_manager, 'cleanup_repeater_retention'): + self.bot.repeater_manager.cleanup_repeater_retention( + daily_stats_days=daily_stats_days, + observed_paths_days=observed_paths_days + ) + + # Stats tables (message_stats, command_stats, path_stats) + if hasattr(self.bot, 'command_manager') and self.bot.command_manager: + stats_cmd = self.bot.command_manager.commands.get('stats') if getattr(self.bot.command_manager, 'commands', None) else None + if stats_cmd and hasattr(stats_cmd, 'cleanup_old_stats'): + stats_cmd.cleanup_old_stats(stats_days) + + # Expired caches (geocoding_cache, generic_cache) + if hasattr(self.bot, 'db_manager') and self.bot.db_manager and hasattr(self.bot.db_manager, 'cleanup_expired_cache'): + self.bot.db_manager.cleanup_expired_cache() + + # Mesh connections (DB prune to match in-memory expiration) + if hasattr(self.bot, 'mesh_graph') and self.bot.mesh_graph and hasattr(self.bot.mesh_graph, 'delete_expired_edges_from_db'): + self.bot.mesh_graph.delete_expired_edges_from_db(mesh_connections_days) + + ran_at = datetime.datetime.utcnow().isoformat() + self._last_retention_stats['ran_at'] = ran_at + try: + self.bot.db_manager.set_metadata('maint.status.data_retention_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.data_retention_outcome', 'ok') + except Exception: + pass + + except Exception as e: + self.logger.exception(f"Error during data retention cleanup: {e}") + self._last_retention_stats['error'] = str(e) + try: + ran_at = datetime.datetime.utcnow().isoformat() + self.bot.db_manager.set_metadata('maint.status.data_retention_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.data_retention_outcome', f'error: {e}') + except Exception: + pass + def check_interval_advertising(self): """Check if it's time to send an interval-based advert""" try: @@ -546,7 +630,7 @@ class MessageScheduler: try: future.result(timeout=60) # 60 second timeout except Exception as e: - self.logger.error(f"Error sending interval advert: {e}") + self.logger.error(f"Error sending interval advert: {type(e).__name__}: {e}", exc_info=True) else: # Fallback: create new event loop if main loop not available try: @@ -560,12 +644,15 @@ class MessageScheduler: async def _send_interval_advert_async(self): """Send an interval-based advert (async implementation)""" - try: - # Use the same advert functionality as the manual advert command - await self.bot.meshcore.commands.send_advert(flood=True) - self.logger.info("Interval-based flood advert sent successfully") - except Exception as e: - self.logger.error(f"Error sending interval-based advert: {e}") + if self.bot.is_radio_zombie: + self.bot.logger.warning("send_advert suppressed — radio is in zombie state; power cycle required") + return + from meshcore.events import EventType + result = await self.bot.meshcore.commands.send_advert(flood=True) + if result.type == EventType.ERROR: + reason = result.payload.get('reason', 'unknown') + raise RuntimeError(f"send_advert failed: {reason}") + self.logger.info("Interval-based flood advert sent successfully") async def _process_channel_operations(self): """Process pending channel operations from the web viewer""" @@ -835,62 +922,539 @@ class MessageScheduler: self.logger.error(f"Firmware write failed: {e}") return False, {'error': str(e)} - # ── Maintenance (delegates to MaintenanceRunner) ───────────────────────── - - @property - def _last_retention_stats(self) -> dict[str, Any]: - return self.maintenance._last_retention_stats - - @_last_retention_stats.setter - def _last_retention_stats(self, value: dict[str, Any]) -> None: - self.maintenance._last_retention_stats.clear() - self.maintenance._last_retention_stats.update(value) - - @property - def _last_db_backup_stats(self) -> dict[str, Any]: - return self.maintenance._last_db_backup_stats - - @_last_db_backup_stats.setter - def _last_db_backup_stats(self, value: dict[str, Any]) -> None: - self.maintenance._last_db_backup_stats.clear() - self.maintenance._last_db_backup_stats.update(value) - - @property - def _last_log_rotation_applied(self) -> dict[str, str]: - return self.maintenance._last_log_rotation_applied - - @_last_log_rotation_applied.setter - def _last_log_rotation_applied(self, value: dict[str, str]) -> None: - self.maintenance._last_log_rotation_applied.clear() - self.maintenance._last_log_rotation_applied.update(value) - - def run_db_backup(self) -> None: - """Run a DB backup immediately (manual / HTTP).""" - self.maintenance.run_db_backup() - - def _run_data_retention(self) -> None: - self.maintenance.run_data_retention() + # ── Nightly maintenance email ──────────────────────────────────────────── def _get_notif(self, key: str) -> str: - return self.maintenance.get_notif(key) + """Read a notification setting from bot_metadata.""" + try: + val = self.bot.db_manager.get_metadata(f'notif.{key}') + return val if val is not None else '' + except Exception: + return '' def _collect_email_stats(self) -> dict[str, Any]: - return self.maintenance.collect_email_stats() + """Gather 24h summary stats for the nightly digest.""" + stats: dict[str, Any] = {} + + # Bot uptime + try: + start = getattr(self.bot, 'connection_time', None) + if start: + delta = datetime.timedelta(seconds=int(time.time() - start)) + hours, rem = divmod(delta.seconds, 3600) + minutes = rem // 60 + parts = [] + if delta.days: + parts.append(f"{delta.days}d") + parts.append(f"{hours}h {minutes}m") + stats['uptime'] = ' '.join(parts) + else: + stats['uptime'] = 'unknown' + except Exception: + stats['uptime'] = 'unknown' + + # Contact counts from DB + try: + with self.bot.db_manager.connection() as conn: + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute("SELECT COUNT(*) AS n FROM complete_contact_tracking") + stats['contacts_total'] = (cur.fetchone() or {}).get('n', 0) + cur.execute( + "SELECT COUNT(*) AS n FROM complete_contact_tracking " + "WHERE last_heard >= datetime('now', '-1 day')" + ) + stats['contacts_24h'] = (cur.fetchone() or {}).get('n', 0) + cur.execute( + "SELECT COUNT(*) AS n FROM complete_contact_tracking " + "WHERE first_heard >= datetime('now', '-1 day')" + ) + stats['contacts_new_24h'] = (cur.fetchone() or {}).get('n', 0) + except Exception as e: + stats['contacts_error'] = str(e) + + # DB file size + try: + db_path = str(self.bot.db_manager.db_path) + size_bytes = os.path.getsize(db_path) + stats['db_size_mb'] = f'{size_bytes / 1_048_576:.1f}' + stats['db_path'] = db_path + except Exception: + stats['db_size_mb'] = 'unknown' + + # Log file stats + rotation + try: + log_file = self.bot.config.get('Logging', 'log_file', fallback='').strip() + if log_file: + log_path = Path(log_file) + stats['log_file'] = str(log_path) + if log_path.exists(): + stats['log_size_mb'] = f'{log_path.stat().st_size / 1_048_576:.1f}' + # Count ERROR/CRITICAL lines written in the last 24h by scanning the file + time.time() - 86400 + error_count = critical_count = 0 + try: + with open(log_path, encoding='utf-8', errors='replace') as fh: + for line in fh: + if ' ERROR ' in line or ' CRITICAL ' in line: + if ' ERROR ' in line: + error_count += 1 + else: + critical_count += 1 + stats['errors_24h'] = error_count + stats['criticals_24h'] = critical_count + except Exception: + stats['errors_24h'] = 'n/a' + stats['criticals_24h'] = 'n/a' + # Detect recent rotation: check for .1 backup file newer than 24h + backup = Path(str(log_path) + '.1') + if backup.exists() and (time.time() - backup.stat().st_mtime) < 86400: + stats['log_rotated_24h'] = True + stats['log_backup_size_mb'] = f'{backup.stat().st_size / 1_048_576:.1f}' + else: + stats['log_rotated_24h'] = False + except Exception: + pass + + # Data retention last run + stats['retention'] = self._last_retention_stats.copy() + + return stats def _format_email_body(self, stats: dict[str, Any], period_start: str, period_end: str) -> str: - return self.maintenance.format_email_body(stats, period_start, period_end) + lines = [ + 'MeshCore Bot — Nightly Maintenance Report', + '=' * 44, + f'Period : {period_start} → {period_end}', + '', + 'BOT STATUS', + '─' * 30, + f" Uptime : {stats.get('uptime', 'unknown')}", + f" Connected : {'yes' if getattr(self.bot, 'connected', False) else 'no'}", + '', + 'NETWORK ACTIVITY (past 24 h)', + '─' * 30, + f" Active contacts : {stats.get('contacts_24h', 'n/a')}", + f" New contacts : {stats.get('contacts_new_24h', 'n/a')}", + f" Total tracked : {stats.get('contacts_total', 'n/a')}", + '', + 'DATABASE', + '─' * 30, + f" Size : {stats.get('db_size_mb', 'n/a')} MB", + ] + if self._last_retention_stats.get('ran_at'): + lines.append(f" Last retention run : {self._last_retention_stats['ran_at']} UTC") + if self._last_retention_stats.get('error'): + lines.append(f" Retention error : {self._last_retention_stats['error']}") + + lines += [ + '', + 'ERRORS (past 24 h)', + '─' * 30, + f" ERROR : {stats.get('errors_24h', 'n/a')}", + f" CRITICAL : {stats.get('criticals_24h', 'n/a')}", + ] + if stats.get('log_file'): + lines += [ + '', + 'LOG FILES', + '─' * 30, + f" Current : {stats.get('log_file')} ({stats.get('log_size_mb', '?')} MB)", + ] + if stats.get('log_rotated_24h'): + lines.append( + f" Rotated : yes — backup is {stats.get('log_backup_size_mb', '?')} MB" + ) + else: + lines.append(' Rotated : no') + + lines += [ + '', + '─' * 44, + 'Manage notification settings: /config', + ] + return '\n'.join(lines) def _send_nightly_email(self) -> None: - self.maintenance.send_nightly_email() + """Build and dispatch the nightly maintenance digest if enabled.""" + import smtplib + import ssl as _ssl + from email.message import EmailMessage + + if self._get_notif('nightly_enabled') != 'true': + return + + smtp_host = self._get_notif('smtp_host') + smtp_security = self._get_notif('smtp_security') or 'starttls' + smtp_user = self._get_notif('smtp_user') + smtp_password = self._get_notif('smtp_password') + from_name = self._get_notif('from_name') or 'MeshCore Bot' + from_email = self._get_notif('from_email') + recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()] + + if not smtp_host or not from_email or not recipients: + self.logger.warning( + "Nightly email enabled but SMTP settings incomplete " + f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients})" + ) + return + + try: + smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587)) + except ValueError: + smtp_port = 587 + + now_utc = datetime.datetime.utcnow() + yesterday = now_utc - datetime.timedelta(days=1) + period_start = yesterday.strftime('%Y-%m-%d %H:%M UTC') + period_end = now_utc.strftime('%Y-%m-%d %H:%M UTC') + + try: + stats = self._collect_email_stats() + body = self._format_email_body(stats, period_start, period_end) + + msg = EmailMessage() + msg['Subject'] = f'MeshCore Bot — Nightly Report {now_utc.strftime("%Y-%m-%d")}' + msg['From'] = f'{from_name} <{from_email}>' + msg['To'] = ', '.join(recipients) + msg.set_content(body) + + # Optionally attach current log file before rotation + if self._get_maint('email_attach_log') == 'true': + log_file = self.bot.config.get('Logging', 'log_file', fallback='').strip() + if log_file: + log_path = Path(log_file) + max_attach = 5 * 1024 * 1024 # 5 MB cap on attachment + if log_path.exists() and log_path.stat().st_size <= max_attach: + try: + with open(log_path, 'rb') as fh: + msg.add_attachment(fh.read(), maintype='text', subtype='plain', + filename=log_path.name) + except Exception as attach_err: + self.logger.warning(f"Could not attach log file to nightly email: {attach_err}") + + context = _ssl.create_default_context() + + _smtp_timeout = 30 # seconds — prevents indefinite hang on unreachable host + if smtp_security == 'ssl': + with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s: + if smtp_user and smtp_password: + s.login(smtp_user, smtp_password) + s.send_message(msg) + else: + with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s: + if smtp_security == 'starttls': + s.ehlo() + s.starttls(context=context) + s.ehlo() + if smtp_user and smtp_password: + s.login(smtp_user, smtp_password) + s.send_message(msg) + + self.logger.info( + f"Nightly maintenance email sent to {recipients} " + f"(contacts_24h={stats.get('contacts_24h')}, " + f"errors={stats.get('errors_24h')})" + ) + try: + ran_at = datetime.datetime.utcnow().isoformat() + self.bot.db_manager.set_metadata('maint.status.nightly_email_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.nightly_email_outcome', 'ok') + except Exception: + pass + + except Exception as e: + self.logger.error(f"Failed to send nightly maintenance email: {e}") + try: + ran_at = datetime.datetime.utcnow().isoformat() + self.bot.db_manager.set_metadata('maint.status.nightly_email_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.nightly_email_outcome', f'error: {e}') + except Exception: + pass + + # ── Zombie radio alert email ───────────────────────────────────────────── + + def send_zombie_alert_email(self, fail_count: int, threshold: int, interval: int) -> None: + """Send an immediate alert email when a zombie radio is detected. + + Uses the same SMTP settings as the nightly digest. Recipients are taken + from the ``radio_zombie_alert_email`` config key; if that key is empty the + nightly maintenance recipients are used as a fallback. + + This method is intentionally synchronous so it can be run in a thread + executor from the async event loop without blocking it. + """ + import smtplib + import ssl as _ssl + from email.message import EmailMessage + + if not self.bot.config.getboolean('Bot', 'radio_zombie_alert_enabled', fallback=True): + return + + smtp_host = self._get_notif('smtp_host') + smtp_security = self._get_notif('smtp_security') or 'starttls' + smtp_user = self._get_notif('smtp_user') + smtp_password = self._get_notif('smtp_password') + from_name = self._get_notif('from_name') or 'MeshCore Bot' + from_email = self._get_notif('from_email') + + # Alert recipients: dedicated config key, falls back to nightly recipients + alert_email_cfg = self.bot.config.get('Bot', 'radio_zombie_alert_email', fallback='').strip() + if alert_email_cfg: + recipients = [r.strip() for r in alert_email_cfg.split(',') if r.strip()] + else: + recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()] + + if not smtp_host or not from_email or not recipients: + self.bot.logger.warning( + "Zombie alert email enabled but SMTP settings incomplete " + f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients}) " + "— alert email not sent" + ) + return + + try: + smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587)) + except ValueError: + smtp_port = 587 + + now_utc = datetime.datetime.utcnow() + connection_type = self.bot.config.get('Connection', 'connection_type', fallback='unknown') + serial_port = self.bot.config.get('Connection', 'serial_port', fallback='n/a') + interval_min = interval // 60 + + subject = ( + f'ALERT: MeshCore Bot — Zombie Radio Detected ' + f'[{now_utc.strftime("%Y-%m-%d %H:%M UTC")}]' + ) + body = '\n'.join([ + 'MeshCore Bot — Zombie Radio Alert', + '=' * 44, + f'Time : {now_utc.strftime("%Y-%m-%d %H:%M:%S UTC")}', + '', + 'RADIO STATUS', + '─' * 30, + f' Connection : {connection_type}', + f' Port / Device : {serial_port}', + f' Failed probes : {fail_count} of {threshold} (threshold)', + f' Probe interval: {interval}s ({interval_min} min)', + '', + 'ACTION REQUIRED', + '─' * 30, + ' The radio firmware is unresponsive (zombie state).', + ' A physical POWER CYCLE of the radio is required.', + ' Disconnect/reconnect of the serial/BLE transport will NOT fix this.', + '', + ' Steps to recover:', + ' 1. Power off the radio hardware', + ' 2. Wait 10 seconds', + ' 3. Power on the radio hardware', + ' 4. The bot will reconnect and resume normal operation automatically', + '', + '─' * 44, + 'Probe monitoring has been suspended to avoid log spam.', + 'It will resume automatically after the next successful reconnect.', + ]) + + try: + msg = EmailMessage() + msg['Subject'] = subject + msg['From'] = f'{from_name} <{from_email}>' + msg['To'] = ', '.join(recipients) + msg.set_content(body) + + context = _ssl.create_default_context() + _smtp_timeout = 30 + + if smtp_security == 'ssl': + with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s: + if smtp_user and smtp_password: + s.login(smtp_user, smtp_password) + s.send_message(msg) + else: + with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s: + if smtp_security == 'starttls': + s.ehlo() + s.starttls(context=context) + s.ehlo() + if smtp_user and smtp_password: + s.login(smtp_user, smtp_password) + s.send_message(msg) + + self.bot.logger.info( + f"Zombie radio alert email sent to {recipients}" + ) + except Exception as e: + self.bot.logger.error(f"Failed to send zombie radio alert email: {e}") + + # ── Maintenance helpers ────────────────────────────────────────────────── def _get_maint(self, key: str) -> str: - return self.maintenance.get_maint(key) + """Read a maintenance setting from bot_metadata.""" + try: + val = self.bot.db_manager.get_metadata(f'maint.{key}') + return val if val is not None else '' + except Exception: + return '' def _apply_log_rotation_config(self) -> None: - self.maintenance.apply_log_rotation_config() + """Check bot_metadata for log rotation settings and replace the RotatingFileHandler if changed.""" + from logging.handlers import RotatingFileHandler as _RFH + + max_bytes_str = self._get_maint('log_max_bytes') + backup_count_str = self._get_maint('log_backup_count') + + if not max_bytes_str and not backup_count_str: + return # Nothing stored yet — nothing to apply + + new_cfg = {'max_bytes': max_bytes_str, 'backup_count': backup_count_str} + if new_cfg == self._last_log_rotation_applied: + return # No change + + try: + max_bytes = int(max_bytes_str) if max_bytes_str else 5 * 1024 * 1024 + backup_count = int(backup_count_str) if backup_count_str else 3 + except ValueError: + self.logger.warning(f"Invalid log rotation config in bot_metadata: {new_cfg}") + return + + logger = self.bot.logger + for i, handler in enumerate(logger.handlers): + if isinstance(handler, _RFH): + log_path = handler.baseFilename + formatter = handler.formatter + level = handler.level + handler.close() + new_handler = _RFH(log_path, maxBytes=max_bytes, backupCount=backup_count, encoding='utf-8') + new_handler.setFormatter(formatter) + new_handler.setLevel(level) + logger.handlers[i] = new_handler + self._last_log_rotation_applied = new_cfg + self.logger.info(f"Log rotation config applied: maxBytes={max_bytes}, backupCount={backup_count}") + try: + ran_at = datetime.datetime.utcnow().isoformat() + self.bot.db_manager.set_metadata('maint.status.log_rotation_applied_at', ran_at) + except Exception: + pass + break def _maybe_run_db_backup(self) -> None: - self.maintenance.maybe_run_db_backup() + """Check if a scheduled DB backup is due and run it.""" + if self._get_maint('db_backup_enabled') != 'true': + return + + sched = self._get_maint('db_backup_schedule') or 'daily' + if sched == 'manual': + return + + backup_time_str = self._get_maint('db_backup_time') or '02:00' + now = self.get_current_time() + try: + bh, bm = [int(x) for x in backup_time_str.split(':')] + except Exception: + bh, bm = 2, 0 + + scheduled_today = now.replace(hour=bh, minute=bm, second=0, microsecond=0) + + # Only fire within a 2-minute window after the scheduled time. + # This allows for scheduler lag while preventing a late bot startup + # from triggering an immediate backup for a time that passed hours ago. + fire_window_end = scheduled_today + datetime.timedelta(minutes=2) + if now < scheduled_today or now > fire_window_end: + return + + if sched == 'weekly' and now.weekday() != 0: # Monday only + return + + # Deduplicate: don't re-run if already ran today (daily) / this week (weekly). + # Seed from DB on first check so restarts don't re-trigger a backup that + # already ran earlier today. + if not self._last_db_backup_stats: + try: + db_ran_at = self.bot.db_manager.get_metadata('maint.status.db_backup_ran_at') or '' + if db_ran_at: + self._last_db_backup_stats['ran_at'] = db_ran_at + except Exception: + pass + + date_key = now.strftime('%Y-%m-%d') + week_key = f"{now.year}-W{now.isocalendar()[1]}" + last_ran = self._last_db_backup_stats.get('ran_at', '') + if sched == 'daily' and last_ran.startswith(date_key): + return + if sched == 'weekly' and self._last_db_backup_stats.get('week_key') == week_key: + return + + self._run_db_backup() + if sched == 'weekly': + self._last_db_backup_stats['week_key'] = week_key def _run_db_backup(self) -> None: - self.maintenance.run_db_backup() + """Backup the SQLite database using sqlite3.Connection.backup(), then prune old backups.""" + import sqlite3 as _sqlite3 + + backup_dir_str = self._get_maint('db_backup_dir') or '/data/backups' + try: + retention_count = int(self._get_maint('db_backup_retention_count') or '7') + except ValueError: + retention_count = 7 + + backup_dir = Path(backup_dir_str) + ran_at = datetime.datetime.utcnow().isoformat() + + try: + backup_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + self.logger.error(f"DB backup: cannot create backup directory {backup_dir}: {e}") + self._last_db_backup_stats = {'ran_at': ran_at, 'error': str(e)} + try: + self.bot.db_manager.set_metadata('maint.status.db_backup_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.db_backup_outcome', f'error: {e}') + except Exception: + pass + return + + db_path = Path(str(self.bot.db_manager.db_path)) + ts = datetime.datetime.utcnow().strftime('%Y%m%dT%H%M%S') + backup_path = backup_dir / f"{db_path.stem}_{ts}.db" + + try: + src = _sqlite3.connect(str(db_path), check_same_thread=False) + dst = _sqlite3.connect(str(backup_path)) + try: + src.backup(dst, pages=200) + finally: + dst.close() + src.close() + + size_mb = backup_path.stat().st_size / 1_048_576 + self.logger.info(f"DB backup created: {backup_path} ({size_mb:.1f} MB)") + + # Prune oldest backups beyond retention count + stem = db_path.stem + backups = sorted(backup_dir.glob(f"{stem}_*.db"), key=lambda p: p.stat().st_mtime) + while len(backups) > retention_count: + oldest = backups.pop(0) + try: + oldest.unlink() + self.logger.info(f"DB backup pruned: {oldest}") + except OSError: + pass + + ran_at = datetime.datetime.utcnow().isoformat() + self._last_db_backup_stats = {'ran_at': ran_at, 'path': str(backup_path), 'size_mb': f'{size_mb:.1f}'} + try: + self.bot.db_manager.set_metadata('maint.status.db_backup_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.db_backup_outcome', 'ok') + self.bot.db_manager.set_metadata('maint.status.db_backup_path', str(backup_path)) + except Exception: + pass + + except Exception as e: + self.logger.error(f"DB backup failed: {e}") + self._last_db_backup_stats = {'ran_at': ran_at, 'error': str(e)} + try: + self.bot.db_manager.set_metadata('maint.status.db_backup_ran_at', ran_at) + self.bot.db_manager.set_metadata('maint.status.db_backup_outcome', f'error: {e}') + except Exception: + pass diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 87c3222..39a248d 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -1799,13 +1799,18 @@ class BotDataViewer: with self._clients_lock: client_count = len(self.connected_clients) + radio_zombie = self.db_manager.get_metadata('bot.radio_zombie') == 'true' + radio_zombie_since = self.db_manager.get_metadata('bot.radio_zombie_since') or None + return jsonify({ - 'status': 'healthy', + 'status': 'degraded' if radio_zombie else 'healthy', 'connected_clients': client_count, 'max_clients': self.max_clients, 'timestamp': time.time(), 'bot_uptime': bot_uptime, - 'version': 'modern_2.0' + 'version': 'modern_2.0', + 'radio_zombie': radio_zombie, + 'radio_zombie_since': radio_zombie_since, }) @self.app.route('/api/system-health') @@ -1832,6 +1837,15 @@ class BotDataViewer: if start_time: health_data['uptime_seconds'] = time.time() - start_time + # Inject zombie radio state from shared metadata + radio_zombie = self.db_manager.get_metadata('bot.radio_zombie') == 'true' + health_data['radio_zombie'] = radio_zombie + health_data['radio_zombie_since'] = ( + self.db_manager.get_metadata('bot.radio_zombie_since') or None + ) + if radio_zombie: + health_data['status'] = 'degraded' + return jsonify(health_data) except Exception as e: diff --git a/tests/fixtures/mqtt_packets.json b/tests/fixtures/mqtt_packets.json index f99e852..e850a32 100644 --- a/tests/fixtures/mqtt_packets.json +++ b/tests/fixtures/mqtt_packets.json @@ -2,186 +2,183 @@ { "origin": "KG7QIN R-Observer", "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", - "timestamp": "2026-03-17T20:51:23.614020", - "type": "PACKET", - "direction": "rx", - "time": "03:51:27", - "date": "18/3/2026", - "len": "45", - "packet_type": "5", - "route": "F", - "payload_len": "34", - "raw": "1509476A7EDEDEDE7E9D8B72FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "3", - "RSSI": "-78", - "score": "886", - "duration": "201", - "hash": "260E767BB8C1C390", - "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" - }, - { - "origin": "KG7QIN R-Observer", - "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", - "timestamp": "2026-03-17T20:51:24.947484", + "timestamp": "2026-03-21T15:35:30.103404", "type": "PACKET", "direction": "tx", - "time": "03:51:28", - "date": "18/3/2026", - "len": "46", - "packet_type": "5", + "time": "22:35:34", + "date": "21/3/2026", + "len": "128", + "packet_type": "4", "route": "F", - "payload_len": "34", - "raw": "1509476A7EDEDEDE7E9D8B72FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", + "payload_len": "126", + "raw": "1508E07ED0C3177A9D29CA98965980706FEE08331E6BCB75D0860D605A36E7F27A6AD4FCD500D6376D3E182E09", + "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" + }, + { + "origin": "KG7QIN R-Observer", + "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", + "timestamp": "2026-03-21T15:35:31.354090", + "type": "PACKET", + "direction": "rx", + "time": "22:35:36", + "date": "21/3/2026", + "len": "120", + "packet_type": "4", + "route": "D", + "payload_len": "118", + "raw": "1200DC6DB1E831C560B0A6D45615017B189335DCA3F477E7DD32C0FABE86CA98FDF5F1934466C157DC948499D1B74F3A228391B1074452191B6AA82FA78AD497967F5C31327D1E692C7918CD6766F0AC18D2D17284FC140ACDC02885C6E65FA9531BCD9E9D0692FEDFCF020DA1B6F84B473751494E205231", + "SNR": "11", + "RSSI": "-49", + "score": "996", + "duration": "416", + "hash": "C16BDED00FAA94FD", "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" }, { "origin": "KG7QIN-Bot\ud83e\udd16", "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", - "timestamp": "2026-03-17T20:51:25.163531", + "timestamp": "2026-03-21T15:35:31.419020", "type": "PACKET", "direction": "rx", - "time": "20:51:25", - "date": "17/03/2026", - "len": "46", - "packet_type": "5", - "route": "F", - "payload_len": "46", - "raw": "150A476A7EDEDEDE7E9D8BA372FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "12.25", - "RSSI": "-28", - "hash": "260E767BB8C1C390", - "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" - }, - { - "origin": "KG7QIN-Bot\ud83e\udd16", - "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", - "timestamp": "2026-03-17T20:51:25.431344", - "type": "PACKET", - "direction": "rx", - "time": "20:51:25", - "date": "17/03/2026", - "len": "47", - "packet_type": "5", - "route": "F", - "payload_len": "47", - "raw": "150B476A7EDEDEDE7E9D8BA32972FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "11.0", - "RSSI": "-66", - "hash": "260E767BB8C1C390", + "time": "15:35:31", + "date": "21/03/2026", + "len": "120", + "packet_type": "4", + "route": "D", + "payload_len": "120", + "raw": "1200DC6DB1E831C560B0A6D45615017B189335DCA3F477E7DD32C0FABE86CA98FDF5F1934466C157DC948499D1B74F3A228391B1074452191B6AA82FA78AD497967F5C31327D1E692C7918CD6766F0AC18D2D17284FC140ACDC02885C6E65FA9531BCD9E9D0692FEDFCF020DA1B6F84B473751494E205231", + "SNR": "11.75", + "RSSI": "-17", + "hash": "C16BDED00FAA94FD", "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" }, { "origin": "KG7QIN R-Observer", "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", - "timestamp": "2026-03-17T20:51:25.460107", + "timestamp": "2026-03-21T15:35:31.910648", "type": "PACKET", "direction": "rx", - "time": "03:51:29", - "date": "18/3/2026", - "len": "47", - "packet_type": "5", + "time": "22:35:36", + "date": "21/3/2026", + "len": "27", + "packet_type": "2", "route": "F", - "payload_len": "34", - "raw": "150B476A7EDEDEDE7E9D8BA32972FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "9", - "RSSI": "-105", + "payload_len": "20", + "raw": "09056CCE9D8BDC1A86B0C3480C75F0A61B0405E31F2C2B0592B79F", + "SNR": "11", + "RSSI": "-49", "score": "1000", - "duration": "201", - "hash": "260E767BB8C1C390", + "duration": "150", + "hash": "A47A6DD60283F5ED", "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" }, { "origin": "KG7QIN-Bot\ud83e\udd16", "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", - "timestamp": "2026-03-17T20:51:25.662083", + "timestamp": "2026-03-21T15:35:32.776082", "type": "PACKET", "direction": "rx", - "time": "20:51:25", - "date": "17/03/2026", - "len": "47", - "packet_type": "5", + "time": "15:35:32", + "date": "21/03/2026", + "len": "27", + "packet_type": "2", "route": "F", - "payload_len": "47", - "raw": "150B476A7EDEDEDE7E9D8BA3DC72FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "12.25", - "RSSI": "-16", - "hash": "260E767BB8C1C390", + "payload_len": "27", + "raw": "09056CCE9D8BDC1A86B0C3480C75F0A61B0405E31F2C2B0592B79F", + "SNR": "11.5", + "RSSI": "-31", + "hash": "A47A6DD60283F5ED", "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" }, { "origin": "KG7QIN R-Observer", "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", - "timestamp": "2026-03-17T20:51:25.686946", + "timestamp": "2026-03-21T15:35:33.321520", "type": "PACKET", - "direction": "rx", - "time": "03:51:29", - "date": "18/3/2026", - "len": "47", - "packet_type": "5", + "direction": "tx", + "time": "22:35:38", + "date": "21/3/2026", + "len": "28", + "packet_type": "2", "route": "F", - "payload_len": "34", - "raw": "150B476A7EDEDEDE7E9D8BA3DC72FE35CD8EB4669087F84204DF28B341DAC8BC582073A7ED328DD690154B86F80C98", - "SNR": "12", - "RSSI": "-46", - "score": "1000", - "duration": "201", - "hash": "260E767BB8C1C390", + "payload_len": "20", + "raw": "09056CCE9D8BDC1A86B0C3480C75F0A61B0405E31F2C2B0592B79F", "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" }, { "origin": "KG7QIN-Bot\ud83e\udd16", "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", - "timestamp": "2026-03-17T20:51:28.938502", + "timestamp": "2026-03-21T15:35:33.478074", "type": "PACKET", "direction": "rx", - "time": "20:51:28", - "date": "17/03/2026", - "len": "153", - "packet_type": "5", + "time": "15:35:33", + "date": "21/03/2026", + "len": "28", + "packet_type": "2", "route": "F", - "payload_len": "153", - "raw": "15041F109D8B72BEFC08014164BBEFD61E9D6ACE273D2D1055B6B62DDB97E202A0675842B7DDDDD6EAB1D87DA82313FC892B3A77BF9828729081AEC4362DBF96CF6BAF1268A6F169770A9A353DBC0E136CFEB20A0AB08A3F21D9ECDD7BD6DBF0383030F06C7559E6039CF13BCD9047713191570E22301B2BFF0E63D187CE01159764F0E54A34C0BE91259EB5D7DA6B27A184BFC8DCF4C993A7", + "payload_len": "28", + "raw": "09066CCE9D8BDCA31A86B0C3480C75F0A61B0405E31F2C2B0592B79F", "SNR": "11.5", "RSSI": "-30", - "hash": "71437AD5EE169AF5", + "hash": "A47A6DD60283F5ED", + "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" + }, + { + "origin": "KG7QIN-Bot\ud83e\udd16", + "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", + "timestamp": "2026-03-21T15:35:36.952595", + "type": "PACKET", + "direction": "rx", + "time": "15:35:36", + "date": "21/03/2026", + "len": "26", + "packet_type": "5", + "route": "F", + "payload_len": "26", + "raw": "1505C3B1C7CE42CABFCB1B4967062BAD7F9DCB3F8F9BFA5E8F00", + "SNR": "12.5", + "RSSI": "-22", + "hash": "4561B3A99C8CD778", "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" }, { "origin": "KG7QIN R-Observer", "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", - "timestamp": "2026-03-17T20:51:28.984163", + "timestamp": "2026-03-21T15:35:36.978188", "type": "PACKET", "direction": "rx", - "time": "03:51:32", - "date": "18/3/2026", - "len": "153", + "time": "22:35:41", + "date": "21/3/2026", + "len": "26", "packet_type": "5", "route": "F", - "payload_len": "147", - "raw": "15041F109D8B72BEFC08014164BBEFD61E9D6ACE273D2D1055B6B62DDB97E202A0675842B7DDDDD6EAB1D87DA82313FC892B3A77BF9828729081AEC4362DBF96CF6BAF1268A6F169770A9A353DBC0E136CFEB20A0AB08A3F21D9ECDD7BD6DBF0383030F06C7559E6039CF13BCD9047713191570E22301B2BFF0E63D187CE01159764F0E54A34C0BE91259EB5D7DA6B27A184BFC8DCF4C993A7", + "payload_len": "19", + "raw": "1505C3B1C7CE42CABFCB1B4967062BAD7F9DCB3F8F9BFA5E8F00", "SNR": "11", - "RSSI": "-79", - "score": "764", - "duration": "518", - "hash": "71437AD5EE169AF5", + "RSSI": "-76", + "score": "1000", + "duration": "139", + "hash": "4561B3A99C8CD778", "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" }, { - "origin": "KG7QIN-Bot\ud83e\udd16", - "origin_id": "CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87", - "timestamp": "2026-03-17T20:51:30.395696", + "origin": "KG7QIN R-Observer", + "origin_id": "A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91", + "timestamp": "2026-03-21T15:35:37.437108", "type": "PACKET", "direction": "rx", - "time": "20:51:30", - "date": "17/03/2026", - "len": "154", + "time": "22:35:42", + "date": "21/3/2026", + "len": "27", "packet_type": "5", "route": "F", - "payload_len": "154", - "raw": "15051F109D8B4272BEFC08014164BBEFD61E9D6ACE273D2D1055B6B62DDB97E202A0675842B7DDDDD6EAB1D87DA82313FC892B3A77BF9828729081AEC4362DBF96CF6BAF1268A6F169770A9A353DBC0E136CFEB20A0AB08A3F21D9ECDD7BD6DBF0383030F06C7559E6039CF13BCD9047713191570E22301B2BFF0E63D187CE01159764F0E54A34C0BE91259EB5D7DA6B27A184BFC8DCF4C993A7", - "SNR": "11.75", - "RSSI": "-23", - "hash": "71437AD5EE169AF5", - "_topic": "meshcore/SEA/CA12274AB96F3BAD9EE93FE2816BE0A92ECFD9507345F2595F36265B47085F87/packets" + "payload_len": "19", + "raw": "1506C3B1C7CE42DCCABFCB1B4967062BAD7F9DCB3F8F9BFA5E8F00", + "SNR": "13", + "RSSI": "-48", + "score": "1000", + "duration": "150", + "hash": "4561B3A99C8CD778", + "_topic": "meshcore/SEA/A3325067ED008FC8F687FC31AD0EB53F0083979F27BC265D54E2635420CA7F91/packets" } ] \ No newline at end of file diff --git a/tests/test_core.py b/tests/test_core.py index 6d99a04..1f4e3bf 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -280,3 +280,216 @@ class TestLoopExceptionHandler: handler(mock_loop, ctx) mock_loop.default_exception_handler.assert_called_once_with(ctx) + + +# --------------------------------------------------------------------------- +# _probe_radio_health (PR4 — zombie-connection detection) +# --------------------------------------------------------------------------- + + +class TestProbeRadioHealth: + """Tests for MeshCoreBot._probe_radio_health().""" + + def _make_bot(self, tmp_path: Path) -> MeshCoreBot: + config_file = tmp_path / "config.ini" + db_path = tmp_path / "bot.db" + _write_config(config_file, db_path) + return MeshCoreBot(config_file=str(config_file)) + + def test_returns_false_when_meshcore_is_none(self, tmp_path): + bot = self._make_bot(tmp_path) + bot.meshcore = None + result = asyncio.run(bot._probe_radio_health()) + assert result is False + + def test_returns_false_when_not_connected(self, tmp_path): + bot = self._make_bot(tmp_path) + bot.meshcore = MagicMock() + bot.meshcore.is_connected = False + result = asyncio.run(bot._probe_radio_health()) + assert result is False + + def test_error_event_increments_fail_count(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 0 + + error_event = MagicMock() + error_event.type = EventType.ERROR + error_event.payload = {"reason": "no_event_received"} + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(error_event) + ) + + result = asyncio.run(bot._probe_radio_health()) + assert result is False + assert bot._radio_fail_count == 1 + + def test_error_event_below_threshold_does_not_reconnect(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 0 + bot.config.set("Bot", "radio_probe_fail_threshold", "3") + + error_event = MagicMock() + error_event.type = EventType.ERROR + error_event.payload = {"reason": "no_event_received"} + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(error_event) + ) + + reconnect_called = [] + + async def fake_reconnect(): + reconnect_called.append(True) + return True + + bot.reconnect_radio = fake_reconnect + + asyncio.run(bot._probe_radio_health()) + assert not reconnect_called + + def test_error_event_at_threshold_triggers_reconnect(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 2 # this probe makes it 3 + bot.config.set("Bot", "radio_probe_fail_threshold", "3") + + error_event = MagicMock() + error_event.type = EventType.ERROR + error_event.payload = {"reason": "no_event_received"} + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(error_event) + ) + + reconnect_called = [] + + async def fake_reconnect(): + reconnect_called.append(True) + return True + + bot.reconnect_radio = fake_reconnect + + result = asyncio.run(bot._probe_radio_health()) + assert result is False + assert bot._radio_fail_count == 0 # reset after trigger + assert reconnect_called + + def test_success_resets_fail_count(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 2 + + ok_event = MagicMock() + ok_event.type = EventType.OK + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(ok_event) + ) + + result = asyncio.run(bot._probe_radio_health()) + assert result is True + assert bot._radio_fail_count == 0 + + def test_success_logs_recovery_when_previously_failed(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 1 + + ok_event = MagicMock() + ok_event.type = EventType.OK + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(ok_event) + ) + + with patch.object(bot.logger, "info") as mock_info: + asyncio.run(bot._probe_radio_health()) + + logged_messages = [str(c) for c in mock_info.call_args_list] + assert any("recovered" in m for m in logged_messages) + + def test_success_no_recovery_log_when_no_prior_failure(self, tmp_path): + from meshcore.events import EventType + bot = self._make_bot(tmp_path) + bot._radio_fail_count = 0 + + ok_event = MagicMock() + ok_event.type = EventType.OK + + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + bot.meshcore.commands.get_time = MagicMock( + return_value=_make_coro(ok_event) + ) + + with patch.object(bot.logger, "info") as mock_info: + asyncio.run(bot._probe_radio_health()) + + logged_messages = [str(c) for c in mock_info.call_args_list] + assert not any("recovered" in m for m in logged_messages) + + def test_asyncio_timeout_returns_false_and_warns(self, tmp_path): + bot = self._make_bot(tmp_path) + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + + async def slow_get_time(): + raise asyncio.TimeoutError() + + bot.meshcore.commands.get_time = MagicMock(return_value=slow_get_time()) + + async def run(): + async def fake_wait_for(coro, timeout): + raise asyncio.TimeoutError() + + with patch("asyncio.wait_for", side_effect=fake_wait_for): + return await bot._probe_radio_health() + + result = asyncio.run(run()) + assert result is False + + def test_generic_exception_returns_false_and_warns(self, tmp_path): + bot = self._make_bot(tmp_path) + bot.meshcore = MagicMock() + bot.meshcore.is_connected = True + + async def failing_get_time(): + raise RuntimeError("device exploded") + + bot.meshcore.commands.get_time = MagicMock( + return_value=failing_get_time() + ) + + with patch.object(bot.logger, "warning") as mock_warn: + result = asyncio.run(bot._probe_radio_health()) + + assert result is False + assert mock_warn.called + + +# --------------------------------------------------------------------------- +# Helper: create a coroutine that returns a fixed value +# --------------------------------------------------------------------------- + + +async def _make_coro_async(value): + return value + + +def _make_coro(value): + """Return a coroutine that immediately resolves to *value*.""" + return _make_coro_async(value) diff --git a/tests/test_scheduler_logic.py b/tests/test_scheduler_logic.py index afc3c72..4624c73 100644 --- a/tests/test_scheduler_logic.py +++ b/tests/test_scheduler_logic.py @@ -575,7 +575,7 @@ class TestDbBackupIntervalGuard: # --------------------------------------------------------------------------- -class TestFormatEmailBody: +class TestFormatEmailBodyPure: """Tests for _format_email_body — pure string builder.""" def setup_method(self): @@ -1229,3 +1229,154 @@ class TestCollectEmailStats: assert result.get("contacts_total") == 50 assert result.get("contacts_24h") == 10 assert result.get("contacts_new_24h") == 3 + + +# --------------------------------------------------------------------------- +# _send_interval_advert_async (PR2 fix — Event-based error detection) +# --------------------------------------------------------------------------- + + +def _make_sched_with_logger(mock_logger): + """Return a MessageScheduler backed by a mock bot with the given logger.""" + bot = Mock() + bot.logger = mock_logger + bot.config = ConfigParser() + bot.config.add_section("Bot") + return MessageScheduler(bot) + + +class TestSendIntervalAdvertAsyncFixed: + """Tests for MessageScheduler._send_interval_advert_async() (PR2 fix).""" + + def test_error_event_raises_runtime_error(self, mock_logger): + from meshcore.events import EventType + + sched = _make_sched_with_logger(mock_logger) + error_event = MagicMock() + error_event.type = EventType.ERROR + error_event.payload = {"reason": "no_event_received"} + sched.bot.meshcore.commands.send_advert = AsyncMock(return_value=error_event) + + with pytest.raises(RuntimeError, match="send_advert failed"): + asyncio.run(sched._send_interval_advert_async()) + + def test_error_event_includes_reason_in_message(self, mock_logger): + from meshcore.events import EventType + + sched = _make_sched_with_logger(mock_logger) + error_event = MagicMock() + error_event.type = EventType.ERROR + error_event.payload = {"reason": "no_event_received"} + sched.bot.meshcore.commands.send_advert = AsyncMock(return_value=error_event) + + with pytest.raises(RuntimeError, match="no_event_received"): + asyncio.run(sched._send_interval_advert_async()) + + def test_ok_event_logs_success(self, mock_logger): + from meshcore.events import EventType + + sched = _make_sched_with_logger(mock_logger) + ok_event = MagicMock() + ok_event.type = EventType.OK + sched.bot.meshcore.commands.send_advert = AsyncMock(return_value=ok_event) + + asyncio.run(sched._send_interval_advert_async()) + + sched.bot.logger.info.assert_called_with( + "Interval-based flood advert sent successfully" + ) + + def test_send_interval_advert_logs_exception_type_name(self, mock_logger): + """Error log must include type(e).__name__ so blank TimeoutError is visible.""" + from concurrent.futures import TimeoutError as FuturesTimeoutError + + sched = _make_sched_with_logger(mock_logger) + + future_mock = MagicMock() + future_mock.result = MagicMock(side_effect=FuturesTimeoutError()) + + loop_mock = MagicMock() + loop_mock.is_running.return_value = True + sched.bot.main_event_loop = loop_mock + + with patch("asyncio.run_coroutine_threadsafe", return_value=future_mock): + sched.send_interval_advert() + + # The error log must include the class name, not just str(e) which + # would be empty for concurrent.futures.TimeoutError + call_args_list = mock_logger.error.call_args_list + assert call_args_list, "logger.error was never called" + logged = str(call_args_list[0]) + assert "TimeoutError" in logged + + +# --------------------------------------------------------------------------- +# _send_scheduled_message_async (PR2 fix — asyncio.wait_for wrapping) +# --------------------------------------------------------------------------- + + +class TestSendScheduledMessageAsyncTimeout: + """Tests for _send_scheduled_message_async() asyncio.wait_for wrapping (PR2).""" + + def test_success_calls_send_channel_message(self, mock_logger): + sched = _make_sched_with_logger(mock_logger) + sched.bot.command_manager.send_channel_message = AsyncMock(return_value=None) + + asyncio.run(sched._send_scheduled_message_async("#general", "hello")) + + sched.bot.command_manager.send_channel_message.assert_awaited_once_with( + "#general", "hello" + ) + + def test_timeout_raises_asyncio_timeout_error(self, mock_logger): + sched = _make_sched_with_logger(mock_logger) + + async def run(): + async def fake_wait_for(coro, timeout): + raise asyncio.TimeoutError() + + with patch("asyncio.wait_for", side_effect=fake_wait_for): + await sched._send_scheduled_message_async("#general", "hello") + + with pytest.raises(asyncio.TimeoutError): + asyncio.run(run()) + + def test_send_timeout_seconds_config_used(self, mock_logger): + """send_timeout_seconds from config is passed to wait_for.""" + sched = _make_sched_with_logger(mock_logger) + sched.bot.config.set("Bot", "send_timeout_seconds", "45") + sched.bot.command_manager.send_channel_message = AsyncMock(return_value=None) + + captured_timeout = [] + + async def spy_wait_for(coro, timeout): + captured_timeout.append(timeout) + return await coro + + async def run(): + with patch("asyncio.wait_for", side_effect=spy_wait_for): + await sched._send_scheduled_message_async("#general", "hello") + + asyncio.run(run()) + assert captured_timeout == [45] + + def test_send_scheduled_message_logs_exception_type_name(self, mock_logger): + """Error log must include type(e).__name__.""" + from concurrent.futures import TimeoutError as FuturesTimeoutError + + sched = _make_sched_with_logger(mock_logger) + + future_mock = MagicMock() + future_mock.result = MagicMock(side_effect=FuturesTimeoutError()) + + loop_mock = MagicMock() + loop_mock.is_running.return_value = True + sched.bot.main_event_loop = loop_mock + + with patch("asyncio.run_coroutine_threadsafe", return_value=future_mock): + sched.send_scheduled_message("#general", "hello") + + call_args_list = mock_logger.error.call_args_list + assert call_args_list, "logger.error was never called" + logged = str(call_args_list[0]) + assert "TimeoutError" in logged