fix(services): keep restarted services logging, and stop a taken port from killing the bot

A second review of the restart fix found more of the same class. The map uploader's stop() removes its file log handler and start() never added it back, so after one restart it logged nothing to file; start() now restores it. DARC MoWaS set _running before binding its port, so a port in use left it dead but healthy, and werkzeug's make_server calls sys.exit when the port is taken, which got past every except Exception and stopped the bot; start() now binds first and turns that exit into an OSError.

Boot-time starts go through _start_service_at_boot, which records a restart failure when start() raises, so that service also waits out the backoff, and no longer reports a disabled service as a configuration problem. The health-loop step is _restart_unhealthy_services. New tests cover both, the bridges' session cleanup and the map uploader restart; all 13 lifecycle tests fail on dev.

CHANGELOG.md

@@ -30,7 +30,7 @@ semantic versioning.
 
 ### Fixed
 
-- A service plugin that does not start because it is not configured (Discord or Telegram with no channels, the map uploader with no key) is no longer restarted every 5 seconds with a warning each time. The health check restarted any service that was not running, and the restart counted as a success, which cleared its backoff. A service that does not start, or is still unhealthy after a restart, now waits out `service_restart_backoff_seconds` before the next try, so one that was only waiting for the radio still recovers. A restarted packet capture or map uploader service also gets its background work back; `stop()` left a shutdown flag set that `start()` never cleared. Restarts also no longer subscribe a service to radio events a second time: services never unsubscribed in `stop()`, so after a restart packet capture published every packet twice and the Discord, Telegram and prefix-collision services handled every event twice.
+- A service plugin that does not start because it is not configured (Discord or Telegram with no channels, the map uploader with no key) is no longer restarted every 5 seconds with a warning each time. The health check restarted any service that was not running, and the restart counted as a success, which cleared its backoff. A service that does not start, or is still unhealthy after a restart, now waits out `service_restart_backoff_seconds` before the next try, so one that was only waiting for the radio still recovers. A restarted packet capture or map uploader service also gets its background work back (`stop()` left a shutdown flag set that `start()` never cleared), and the map uploader keeps logging to the bot's log file after a restart. The DARC MoWaS service no longer stops the whole bot when its port is taken (werkzeug called `sys.exit`), and no longer reports itself healthy when it could not listen. Restarts also no longer subscribe a service to radio events a second time: services never unsubscribed in `stop()`, so after a restart packet capture published every packet twice and the Discord, Telegram and prefix-collision services handled every event twice.
 - A delivered DM is no longer reported as failed because its ACK came back after the bot stopped waiting (#290). Each attempt waited 1.2 times the radio's `suggested_timeout`, which the firmware derives from the packet's airtime and hop count with well under a second of fixed allowance, so on fast presets a zero-hop attempt gave up after about 1.3 seconds, before the recipient's ACK delay and the radio's TX queue had even played out. Webhook DMs then returned an error for messages the recipient had. New `[Bot] dm_min_ack_timeout` (default 8 seconds) sets a floor on each attempt's wait; an ACK still ends it at once, so successful sends are no slower, but a DM that is never acknowledged now takes at least 24 seconds to report failure with the default three attempts. Values are capped at 60 seconds. Set it to 0 for the old behavior.
 - A radio that rejects `set_flood_scope` is now reported. The check compared the reply's `EventType` to the string `"ERROR"`, which never matches, so a rejected regional scope (the message then goes out at whatever scope the radio holds) and a rejected restore to global flood (the radio stays pinned to the region for every later send) were both silent.
 - Reverse geocoding for contact adverts no longer blocks the bot. `track_contact_advertisement`, background geocoding, the repeater scan and the geolocation backfill called rate-limited Nominatim synchronously from async code, so every other task (command replies, sends, the mesh reader) stalled for the length of each lookup, including the rate limiter's sleep. These calls now run in a worker thread. Advert tracking decides whether to geocode before its write block and re-checks the duplicate-packet guard inside it, so concurrent adverts still cannot double-count.

modules/core.py

@@ -2397,12 +2397,11 @@ long_jokes = false
         # link is actually ready, so it's safe to bind before we're connected.
         webhook_service = self.services.get('webhook')
         if webhook_service is not None and getattr(webhook_service, 'enabled', False):
-            try:
-                await webhook_service.start()
-                if not self._note_if_service_not_running('webhook', webhook_service):
-                    self.logger.info("Service 'webhook' started (early, before radio connect)")
-            except Exception as e:
-                self.logger.error(f"Failed to start service 'webhook' early: {e}")
+            await self._start_service_at_boot(
+                'webhook', webhook_service,
+                started="Service 'webhook' started (early, before radio connect)",
+                failed="Failed to start service 'webhook' early",
+            )
 
         # Connect to MeshCore node
         if not await self.connect():
@@ -2450,12 +2449,11 @@ long_jokes = false
         for service_name, service_instance in self.services.items():
             if service_name == 'webhook':
                 continue
-            try:
-                await service_instance.start()
-                if not self._note_if_service_not_running(service_name, service_instance):
-                    self.logger.info(f"Service '{service_name}' started")
-            except Exception as e:
-                self.logger.error(f"Failed to start service '{service_name}': {e}")
+            await self._start_service_at_boot(
+                service_name, service_instance,
+                started=f"Service '{service_name}' started",
+                failed=f"Failed to start service '{service_name}'",
+            )
 
         # Start command queue processor if needed
         if hasattr(self.command_manager, '_start_queue_processor'):
@@ -2532,14 +2530,7 @@ long_jokes = false
                     restart_backoff = self.config.getint(
                         'Bot', 'service_restart_backoff_seconds', fallback=300
                     )
-                    now = time.time()
-                    for name, service in self.services.items():
-                        if not self._service_restart_due(name, service, now, restart_backoff):
-                            continue
-                        self.logger.warning(
-                            f"Service '{name}' unhealthy, attempting restart..."
-                        )
-                        asyncio.create_task(self._restart_service(name, service))
+                    self._restart_unhealthy_services(time.time(), restart_backoff)
 
                 await asyncio.sleep(5)  # Check every 5 seconds
         except KeyboardInterrupt:
@@ -2636,6 +2627,25 @@ long_jokes = false
         finally:
             self._shutdown_complete = True
 
+    async def _start_service_at_boot(self, name: str, service: Any, *, started: str, failed: str) -> None:
+        """Start one service during startup; a failure or a declined start waits out the restart backoff."""
+        try:
+            await service.start()
+        except Exception as e:
+            self.logger.error(f"{failed}: {e}")
+            self._service_restart_failures[name] = time.time()
+            return
+        if not self._note_if_service_not_running(name, service):
+            self.logger.info(started)
+
+    def _restart_unhealthy_services(self, now: float, backoff: float) -> None:
+        """Health-loop step: start a restart task for each service that is due one."""
+        for name, service in self.services.items():
+            if not self._service_restart_due(name, service, now, backoff):
+                continue
+            self.logger.warning(f"Service '{name}' unhealthy, attempting restart...")
+            asyncio.create_task(self._restart_service(name, service))
+
     def _service_restart_due(self, name: str, service: Any, now: float, backoff: float) -> bool:
         """Whether the health loop should restart *service* now."""
         if not getattr(service, 'enabled', True):
@@ -2659,6 +2669,8 @@ long_jokes = false
         health tick cannot help; after the backoff the health loop tries again,
         which recovers the transient case.
         """
+        if not getattr(service_instance, 'enabled', True):
+            return False  # disabled on purpose; the health loop skips it anyway
         try:
             running = service_instance.is_running()
         except Exception:

modules/service_plugins/darc_mowas_service.py

@@ -108,14 +108,20 @@ class DARC_MoWaS_Service(BaseServicePlugin):
         self._setup_routes()
 
     async def start(self) -> None:
-        self._running = True
         await self._ensure_channels()
         self.logger.info(
             "MoWaSAlert service starting on %s:%s",
             self.host,
             self.port,
         )
-        self._server = make_server(self.host, self.port, self.app)
+        try:
+            self._server = make_server(self.host, self.port, self.app)
+        except SystemExit as e:
+            # werkzeug exits the process when the port is taken; that must not
+            # take the whole bot down with it.
+            raise OSError(f"MoWaSAlert could not listen on {self.host}:{self.port}") from e
+        # Only after the bind succeeds, so a port in use is not reported healthy.
+        self._running = True
         self._loop = asyncio.get_running_loop()
         self._server_future = self._loop.run_in_executor(
             None, self._server.serve_forever

modules/service_plugins/map_uploader_service.py

@@ -127,15 +127,8 @@ class MapUploaderService(BaseServicePlugin):
         self.logger.addHandler(console_handler)
 
         # Also add file handler to write to the same log file as the bot (skip if no [Logging] section)
-        log_file = (bot.config.get('Logging', 'log_file', fallback='meshcore_bot.log')
-                    if bot.config.has_section('Logging') else '')
-        if log_file:
-            # Resolve log file path (relative paths resolved from bot root, absolute paths used as-is)
-            log_file = resolve_path(log_file, bot.bot_root)
-
-            file_handler = logging.FileHandler(log_file)
-            file_handler.setFormatter(bot_formatter)
-            self.logger.addHandler(file_handler)
+        self._log_formatter = bot_formatter
+        self._add_file_handler()
 
         # Prevent propagation to root logger
         self.logger.propagate = False
@@ -221,8 +214,11 @@ class MapUploaderService(BaseServicePlugin):
             self.logger.info("Map uploader service is disabled")
             return
 
-        # stop() sets this; clear it so a restart's background work actually runs.
+        # stop() sets this and removes the file log handler; undo both so a
+        # restart's background work runs and still logs to the bot's file.
         self.should_exit = False
+        if not any(isinstance(h, logging.FileHandler) for h in self.logger.handlers):
+            self._add_file_handler()
 
         # Check dependencies
         if not AIOHTTP_AVAILABLE:
@@ -307,6 +303,19 @@ class MapUploaderService(BaseServicePlugin):
 
         self.logger.info("Map uploader service stopped")
 
+    def _add_file_handler(self) -> None:
+        """Log to the bot's log file as well (nothing without a [Logging] section)."""
+        config = self.bot.config
+        log_file = (config.get('Logging', 'log_file', fallback='meshcore_bot.log')
+                    if config.has_section('Logging') else '')
+        if log_file:
+            # Resolve log file path (relative paths resolved from bot root, absolute paths used as-is)
+            log_file = resolve_path(log_file, self.bot.bot_root)
+
+            file_handler = logging.FileHandler(log_file)
+            file_handler.setFormatter(self._log_formatter)
+            self.logger.addHandler(file_handler)
+
     async def _fetch_private_key(self) -> None:
         """Fetch private key from device if not already loaded.
 

tests/test_service_lifecycle.py

@@ -171,3 +171,124 @@ def test_a_restart_that_leaves_the_service_unhealthy_backs_off(tmp_path):
     bot = _core_bot(tmp_path)
     assert asyncio.run(bot._restart_service("flaky", _StillUnhealthy())) is False
     assert "flaky" in bot._service_restart_failures
+
+
+def _plain_bot(sections):
+    import configparser
+
+    bot = MagicMock()
+    bot.config = configparser.ConfigParser()
+    bot.config.read_dict(sections)
+    bot.bot_root = Path("/tmp")
+    return bot
+
+
+def test_bridges_close_their_http_session_when_meshcore_is_missing():
+    from modules.service_plugins.discord_bridge_service import DiscordBridgeService
+    from modules.service_plugins.telegram_bridge_service import TelegramBridgeService
+
+    for cls, section, mapping in (
+        (DiscordBridgeService, "DiscordBridge", "channel_webhooks"),
+        (TelegramBridgeService, "TelegramBridge", "channel_chat_ids"),
+    ):
+        bot = _plain_bot({section: {"enabled": "true"}})
+        service = cls(bot)
+        service.enabled = True
+        setattr(service, mapping, {"general": ["target"]})
+        bot.meshcore = None
+
+        async def run(service=service):
+            await service.start()
+            return service.http_session
+
+        assert asyncio.run(run()) is None, cls.__name__
+        assert service.is_running() is False
+
+
+def test_map_uploader_restart_restores_its_file_log_and_clears_should_exit(tmp_path):
+    import logging
+    import logging as _logging
+
+    from modules.service_plugins.map_uploader_service import MapUploaderService
+
+    bot = _plain_bot({"MapUploader": {"enabled": "true"}, "Logging": {"log_file": str(tmp_path / "bot.log")}})
+    bot.logger = _logging.getLogger("test-bot")
+    bot.connected = False
+    service = MapUploaderService(bot)
+    service.enabled = True
+
+    def file_handlers():
+        return [h for h in service.logger.handlers if isinstance(h, logging.FileHandler)]
+
+    assert len(file_handlers()) == 1
+    asyncio.run(service.stop())
+    assert file_handlers() == [] and service.should_exit is True
+
+    async def no_wait(*args, **kwargs):
+        return None
+
+    from unittest.mock import patch
+
+    with patch("asyncio.sleep", no_wait):
+        asyncio.run(service.start())  # gives up waiting for the radio
+    assert len(file_handlers()) == 1
+    assert service.should_exit is False
+    for handler in file_handlers():
+        handler.close()
+        service.logger.removeHandler(handler)
+
+
+def test_health_step_restarts_only_services_that_are_due(tmp_path):
+    from unittest.mock import AsyncMock
+
+    bot = _core_bot(tmp_path)
+    due, backing_off = _Declining(), _Declining()
+    bot.services = {"due": due, "backing_off": backing_off}
+    bot._service_restart_failures["backing_off"] = 1000.0
+    bot._restart_service = AsyncMock(return_value=False)
+
+    async def step():
+        bot._restart_unhealthy_services(now=1010.0, backoff=300)
+        await asyncio.sleep(0)
+
+    asyncio.run(step())
+    bot._restart_service.assert_awaited_once_with("due", due)
+
+
+def test_boot_start_failures_and_disabled_services(tmp_path):
+    bot = _core_bot(tmp_path)
+
+    class Raises(_Declining):
+        async def start(self):
+            raise OSError("port in use")
+
+    asyncio.run(bot._start_service_at_boot("raises", Raises(), started="ok", failed="failed"))
+    assert "raises" in bot._service_restart_failures
+
+    disabled = _Declining()
+    disabled.enabled = False
+    asyncio.run(bot._start_service_at_boot("off", disabled, started="ok", failed="failed"))
+    assert "off" not in bot._service_restart_failures
+
+
+def test_darc_mowas_is_not_running_when_its_port_is_taken():
+    import socket
+
+    from modules.service_plugins.darc_mowas_service import DARC_MoWaS_Service
+
+    holder = socket.socket()
+    holder.bind(("127.0.0.1", 0))
+    holder.listen(1)
+    port = holder.getsockname()[1]
+    try:
+        bot = _plain_bot({"DARC_MoWaS_Service": {"enabled": "true", "host": "127.0.0.1", "port": str(port)}})
+        service = DARC_MoWaS_Service(bot)
+        service.host, service.port = "127.0.0.1", port
+        service._ensure_channels = lambda: asyncio.sleep(0)
+        import pytest
+
+        with pytest.raises(OSError):  # not SystemExit: a taken port must not stop the bot
+            asyncio.run(service.start())
+        assert service.is_healthy() is False
+    finally:
+        holder.close()