Merge pull request #376 from agessaman/fix/more-things

repeater/main.py

@@ -5,6 +5,7 @@ import os
 import signal
 import socket
 import sys
+import threading
 import time
 
 from openhop_core.companion.radio_capabilities import resolve_max_tx_power_dbm
@@ -120,6 +121,9 @@ class RepeaterDaemon:
         self._companion_specs: list[IdentitySpec] | None = None
         self._shutdown_started = False
         self._main_task = None
+        # Set by the first shutdown signal so a second one is ignored while
+        # run() is still unwinding.
+        self._stop_requested = False
         self.radio_status = "unknown"
         self.radio_error = None
 
@@ -850,6 +854,15 @@ class RepeaterDaemon:
                 # its own flood replies to the region the request arrived under.
                 bridge.region_map = self._region_map
 
+                # Feed this bridge every pre-dedup copy of a flood reply so its
+                # return-path teacher can pick the best-received route rather than
+                # the first-arrived one. The router hands a bridge only the first
+                # copy (later ones are dropped by the engine's seen-table) and the
+                # pre-dedup firehose lives on the dispatcher, which the bridge does
+                # not own -- so the host has to wire it.
+                if self.dispatcher:
+                    self.dispatcher.add_raw_packet_subscriber(bridge.note_flood_copy)
+
                 # Restore persisted state (contacts/channels/messages) from SQLite.
                 # Raises CompanionStateLoadError instead of continuing with an
                 # empty store when persisted rows exist but cannot be loaded.
@@ -1088,6 +1101,15 @@ class RepeaterDaemon:
         # re-scopes its flood replies to the region the request arrived under.
         bridge.region_map = self._region_map
 
+        # Feed this bridge every pre-dedup copy of a flood reply so its
+        # return-path teacher can pick the best-received route rather than
+        # the first-arrived one. The router hands a bridge only the first
+        # copy (later ones are dropped by the engine's seen-table) and the
+        # pre-dedup firehose lives on the dispatcher, which the bridge does
+        # not own -- so the host has to wire it.
+        if self.dispatcher:
+            self.dispatcher.add_raw_packet_subscriber(bridge.note_flood_copy)
+
         # Restore persisted state; raises CompanionStateLoadError when persisted
         # rows exist but cannot be loaded (hot-reload callers surface the error).
         if sqlite_handler:
@@ -1529,91 +1551,111 @@ class RepeaterDaemon:
 
     def _signal_shutdown(self, sig, loop):
         """Handle SIGTERM/SIGINT by scheduling async shutdown."""
-        if self._shutdown_started:
+        if self._shutdown_started or self._stop_requested:
             logger.info(f"Received signal {sig.name}, shutdown already in progress")
             return
         logger.info(f"Received signal {sig.name}, shutting down...")
-        loop.create_task(self._shutdown())
-        # Cancel run() so dispatcher.run_forever() unwinds cleanly.
-        if self._main_task and not self._main_task.done():
+        self._stop_requested = True
+        # Unwind run() *cooperatively* rather than cancelling it: stopping the
+        # dispatcher makes run_forever() return, and run()'s finally then does
+        # the cleanup inside a task that is not being cancelled.
+        #
+        # Cancelling run() instead — the previous behaviour — meant its finally
+        # awaited _shutdown() from inside an already-cancelled task, so the first
+        # await raised CancelledError, run() returned, and asyncio.run() tore the
+        # loop down before any cleanup happened. Observed on SIGTERM: not one
+        # shutdown step logged, all three companion listen sockets still bound
+        # and the serial port still held, and the process then hung
+        # indefinitely in interpreter finalization. Running cleanup in a sibling
+        # task does not fix it either: run() returns as soon as the dispatcher
+        # stops, and asyncio.run() cancels every leftover task on the way out.
+        if self.dispatcher is not None and hasattr(self.dispatcher, "stop"):
+            loop.create_task(self.dispatcher.stop())
+        elif self._main_task and not self._main_task.done():
+            # No dispatcher to stop (a failure before startup finished): fall
+            # back to cancelling, and accept the reduced cleanup.
             self._main_task.cancel()
 
+    # Per-step ceiling for shutdown. A best-effort shutdown must never be able to
+    # hang: one stuck step used to strand the whole sequence, leaving sockets
+    # bound and the serial port held.
+    SHUTDOWN_STEP_TIMEOUT_S = 5.0
+    # Grace period after cleanup before the process is forced down. Interpreter
+    # finalization joins non-daemon threads with no timeout of its own, so a
+    # single library thread that never returns hangs SIGTERM forever.
+    SHUTDOWN_EXIT_GRACE_S = 5.0
+
+    async def _shutdown_step(self, name: str, awaitable, timeout: float = None) -> None:
+        """Await one shutdown step, bounded and logged; never raise."""
+        try:
+            await asyncio.wait_for(
+                awaitable, timeout=self.SHUTDOWN_STEP_TIMEOUT_S if timeout is None else timeout
+            )
+        except asyncio.TimeoutError:
+            logger.warning("Shutdown step '%s' timed out; continuing", name)
+        except asyncio.CancelledError:
+            logger.warning("Shutdown step '%s' was cancelled; continuing", name)
+        except Exception as e:
+            logger.warning("Shutdown step '%s' failed: %s", name, e)
+
     async def _shutdown(self):
         """Best-effort shutdown: stop background services and release hardware."""
         if self._shutdown_started:
             return
         self._shutdown_started = True
+        logger.info("Shutdown: stopping services and releasing hardware")
+
+        # Stop the dispatcher first so RX stops before its radio is taken away.
+        if self.dispatcher is not None and hasattr(self.dispatcher, "stop"):
+            await self._shutdown_step("dispatcher", self.dispatcher.stop())
 
         # Stop companion frame servers first to close client sockets and child workers.
         for frame_server in getattr(self, "companion_frame_servers", []):
-            try:
-                await frame_server.stop()
-            except Exception as e:
-                logger.warning(f"Companion frame server stop error: {e}")
+            await self._shutdown_step(
+                f"frame server :{getattr(frame_server, 'port', '?')}", frame_server.stop()
+            )
 
         # Stop companion bridges to flush/persist state.
         if hasattr(self, "companion_bridges"):
-            for bridge in self.companion_bridges.values():
+            for companion_hash, bridge in self.companion_bridges.items():
                 if hasattr(bridge, "stop"):
-                    try:
-                        await bridge.stop()
-                    except Exception as e:
-                        logger.warning(f"Companion bridge stop error: {e}")
+                    await self._shutdown_step(f"bridge 0x{companion_hash:02X}", bridge.stop())
 
         # Stop router
         if self.router:
-            try:
-                await self.router.stop()
-            except Exception as e:
-                logger.warning(f"Error stopping router: {e}")
+            await self._shutdown_step("router", self.router.stop())
 
-        # Stop HTTP server
+        # Stop HTTP server. Sync stop() runs off-loop so a wedged handler thread
+        # cannot block the sequence.
         if self.http_server:
-            try:
-                await asyncio.wait_for(asyncio.to_thread(self.http_server.stop), timeout=3)
-            except asyncio.TimeoutError:
-                logger.warning("Timeout stopping HTTP server")
-            except Exception as e:
-                logger.warning(f"Error stopping HTTP server: {e}")
+            await self._shutdown_step(
+                "http server", asyncio.to_thread(self.http_server.stop), timeout=3
+            )
 
         # Stop Glass inform loop
         if self.glass_handler:
-            try:
-                await self.glass_handler.stop()
-            except Exception as e:
-                logger.warning(f"Error stopping Glass handler: {e}")
+            await self._shutdown_step("glass handler", self.glass_handler.stop())
 
         # Stop sensor manager.
         if self.sensor_manager:
-            try:
-                self.sensor_manager.stop()
-            except Exception as e:
-                logger.warning(f"Error stopping sensor manager: {e}")
+            await self._shutdown_step("sensor manager", asyncio.to_thread(self.sensor_manager.stop))
 
         # Stop GPS diagnostics.
         if self.gps_service:
-            try:
-                self.gps_service.stop()
-            except Exception as e:
-                logger.warning(f"Error stopping GPS diagnostics: {e}")
+            await self._shutdown_step("gps service", asyncio.to_thread(self.gps_service.stop))
 
         # Close storage publishers (MQTT/LetsMesh) to stop their worker threads.
-        try:
-            if self.repeater_handler and self.repeater_handler.storage:
-                await asyncio.wait_for(
-                    asyncio.to_thread(self.repeater_handler.storage.close), timeout=5
-                )
-        except asyncio.TimeoutError:
-            logger.warning("Timeout closing storage publishers")
-        except Exception as e:
-            logger.warning(f"Error closing storage: {e}")
+        if self.repeater_handler and self.repeater_handler.storage:
+            await self._shutdown_step(
+                "storage publishers",
+                asyncio.to_thread(self.repeater_handler.storage.close),
+                timeout=5,
+            )
 
-        # Release radio resources
+        # Release radio resources. Off-loop for the same reason as the HTTP
+        # server: closing a serial port can block on a stuck driver.
         if self.radio and hasattr(self.radio, "cleanup"):
-            try:
-                self.radio.cleanup()
-            except Exception as e:
-                logger.warning(f"Error cleaning up radio: {e}")
+            await self._shutdown_step("radio cleanup", asyncio.to_thread(self.radio.cleanup))
 
         # Release CH341 USB device if in use
         try:
@@ -1627,6 +1669,60 @@ class RepeaterDaemon:
             logger.debug(f"CH341 reset skipped/failed: {e}")
 
         # Do not force-stop the event loop here; asyncio.run() owns loop lifecycle.
+        logger.info("Shutdown: services stopped")
+        self._report_lingering_threads()
+        self._arm_exit_watchdog()
+
+    @staticmethod
+    def _report_lingering_threads() -> None:
+        """Name any non-daemon threads that will block interpreter exit.
+
+        Python joins non-daemon threads at finalization with no timeout, so one
+        library thread that never returns hangs SIGTERM indefinitely — observed
+        here for 18 minutes, the main thread parked in
+        ``Py_FinalizeEx -> wait_for_thread_shutdown``. The watchdog below stops
+        that from holding up a restart; this names the culprit so it can be
+        fixed at the source rather than papered over every time.
+        """
+        current = threading.current_thread()
+        expected, unexpected = [], []
+        for t in threading.enumerate():
+            if t is current or t is threading.main_thread() or t.daemon or not t.is_alive():
+                continue
+            # asyncio names its default-executor workers "asyncio_N". Those are
+            # joined by asyncio.run() under its own timeout, so they are not the
+            # unbounded kind; warning about them every shutdown would bury the
+            # thread that actually matters.
+            (expected if t.name.startswith("asyncio_") else unexpected).append(t.name)
+        if unexpected:
+            logger.warning(
+                "Non-daemon threads still alive; these block interpreter exit: %s",
+                ", ".join(sorted(unexpected)),
+            )
+        if expected:
+            logger.debug("Executor threads still winding down: %s", ", ".join(sorted(expected)))
+
+    def _arm_exit_watchdog(self) -> None:
+        """Force the process down if finalization does not complete promptly.
+
+        Cleanup is finished by the time this is armed, so exiting hard costs
+        nothing and guarantees SIGTERM is honoured. A daemon timer, so it never
+        becomes the thing holding the process open.
+        """
+        grace = self.SHUTDOWN_EXIT_GRACE_S
+
+        def _force_exit() -> None:
+            logger.warning(
+                "Shutdown did not complete within %.0fs of cleanup finishing; exiting now",
+                grace,
+            )
+            logging.shutdown()
+            os._exit(0)
+
+        watchdog = threading.Timer(grace, _force_exit)
+        watchdog.name = "shutdown-watchdog"
+        watchdog.daemon = True
+        watchdog.start()
 
     @staticmethod
     def _detect_container() -> bool:

tests/test_main_py_coverage.py

@@ -1,14 +1,10 @@
+import asyncio
 import logging
+import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
-
-from repeater.companion.constants import STATS_TYPE_CORE, STATS_TYPE_PACKETS, STATS_TYPE_RADIO
-from repeater.exceptions import ConfigurationError
-from repeater.identity_manager import IdentityConfigurationError
-from repeater.main import RepeaterDaemon
-from repeater.main import main as repeater_main
 from openhop_core.node.dispatcher import Dispatcher
 from openhop_core.protocol import PacketBuilder
 from openhop_core.protocol.constants import (
@@ -18,6 +14,12 @@ from openhop_core.protocol.constants import (
     ROUTE_TYPE_TRANSPORT_FLOOD,
 )
 
+from repeater.companion.constants import STATS_TYPE_CORE, STATS_TYPE_PACKETS, STATS_TYPE_RADIO
+from repeater.exceptions import ConfigurationError
+from repeater.identity_manager import IdentityConfigurationError
+from repeater.main import RepeaterDaemon
+from repeater.main import main as repeater_main
+
 
 class _FakeIdentity:
     def __init__(self, pubkey: bytes):
@@ -467,9 +469,25 @@ def test_update_repeater_location_from_gps_branches():
     assert daemon._update_repeater_location_from_gps({"latitude": 6.5, "longitude": 7.5}) is True
 
 
-def test_signal_shutdown_idempotence_and_task_cancel():
+def test_signal_shutdown_unwinds_run_cooperatively():
+    """The handler must not cancel run(): it stops the dispatcher instead.
+
+    Cancelling run() made its ``finally`` await _shutdown() from inside an
+    already-cancelled task, so the first await raised CancelledError and no
+    cleanup ran at all -- on SIGTERM the companion listen sockets stayed bound,
+    the serial port stayed held, and the process then hung in interpreter
+    finalization. Stopping the dispatcher lets run_forever() return so cleanup
+    runs in a task that is not being cancelled.
+    """
     daemon = RepeaterDaemon(_base_config(), radio=object())
-    loop = SimpleNamespace(create_task=MagicMock(side_effect=lambda coro: coro.close()))
+    created = []
+
+    def _fake_create_task(coro):
+        created.append(coro)
+        coro.close()  # never scheduled in this unit test
+        return SimpleNamespace(done=lambda: False)
+
+    loop = SimpleNamespace(create_task=MagicMock(side_effect=_fake_create_task))
     sig = SimpleNamespace(name="SIGTERM")
 
     daemon._shutdown_started = True
@@ -477,12 +495,94 @@ def test_signal_shutdown_idempotence_and_task_cancel():
     loop.create_task.assert_not_called()
 
     daemon._shutdown_started = False
+    daemon.dispatcher = SimpleNamespace(stop=AsyncMock())
     daemon._main_task = SimpleNamespace(done=lambda: False, cancel=MagicMock())
     daemon._signal_shutdown(sig, loop)
+
     loop.create_task.assert_called_once()
+    daemon._main_task.cancel.assert_not_called()
+    assert daemon._stop_requested is True
+
+    # A second signal is ignored while the first is still unwinding.
+    daemon._signal_shutdown(sig, loop)
+    loop.create_task.assert_called_once()
+
+
+def test_signal_shutdown_falls_back_to_cancel_without_a_dispatcher():
+    """A failure before startup finished leaves nothing to stop cooperatively."""
+    daemon = RepeaterDaemon(_base_config(), radio=object())
+    loop = SimpleNamespace(create_task=MagicMock())
+    daemon.dispatcher = None
+    daemon._main_task = SimpleNamespace(done=lambda: False, cancel=MagicMock())
+
+    daemon._signal_shutdown(SimpleNamespace(name="SIGINT"), loop)
+
+    loop.create_task.assert_not_called()
     daemon._main_task.cancel.assert_called_once()
 
 
+def test_lingering_non_daemon_threads_are_named():
+    """The report names what will block interpreter exit, so it is fixable."""
+    import threading as _threading
+
+    release = _threading.Event()
+    victim = _threading.Thread(target=release.wait, name="stuck-nondaemon", daemon=False)
+    victim.start()
+    try:
+        with patch("repeater.main.logger") as log:
+            RepeaterDaemon._report_lingering_threads()
+        warned = " ".join(str(c) for c in log.warning.call_args_list)
+        assert "stuck-nondaemon" in warned
+    finally:
+        release.set()
+        victim.join(timeout=2)
+
+    # Nothing to report once it exits.
+    with patch("repeater.main.logger") as log:
+        RepeaterDaemon._report_lingering_threads()
+    assert "stuck-nondaemon" not in " ".join(str(c) for c in log.warning.call_args_list)
+
+
+def test_asyncio_executor_threads_are_not_reported_as_blockers():
+    """asyncio.run() joins its own executor workers, so they must not warn.
+
+    They are non-daemon and alive at this point on every healthy shutdown;
+    warning about them would bury the thread that actually blocks exit.
+    """
+    import threading as _threading
+
+    release = _threading.Event()
+    worker = _threading.Thread(target=release.wait, name="asyncio_0", daemon=False)
+    worker.start()
+    try:
+        with patch("repeater.main.logger") as log:
+            RepeaterDaemon._report_lingering_threads()
+        log.warning.assert_not_called()
+        assert "asyncio_0" in " ".join(str(c) for c in log.debug.call_args_list)
+    finally:
+        release.set()
+        worker.join(timeout=2)
+
+
+def test_exit_watchdog_is_a_daemon_timer_that_forces_exit():
+    """It must never itself hold the process open, and must exit hard."""
+    import threading as _threading
+
+    daemon_obj = RepeaterDaemon(_base_config(), radio=object())
+    daemon_obj.SHUTDOWN_EXIT_GRACE_S = 0.05
+    with patch("repeater.main.os._exit") as force_exit, patch("repeater.main.logging.shutdown"):
+        daemon_obj._arm_exit_watchdog()
+        timer = next(
+            (t for t in _threading.enumerate() if t.name == "shutdown-watchdog"),
+            None,
+        )
+        assert timer is not None and timer.daemon is True
+        deadline = time.monotonic() + 3
+        while not force_exit.called and time.monotonic() < deadline:
+            time.sleep(0.02)
+        force_exit.assert_called_once_with(0)
+
+
 @pytest.mark.asyncio
 async def test_shutdown_stops_components_and_handles_errors():
     daemon = RepeaterDaemon(_base_config(), radio=SimpleNamespace(cleanup=MagicMock()))
@@ -498,13 +598,47 @@ async def test_shutdown_stops_components_and_handles_errors():
     daemon.sensor_manager = SimpleNamespace(stop=MagicMock())
     daemon.gps_service = SimpleNamespace(stop=MagicMock())
     daemon.repeater_handler = SimpleNamespace(storage=SimpleNamespace(close=MagicMock()))
+    daemon.dispatcher = SimpleNamespace(stop=AsyncMock())
 
-    await daemon._shutdown()
+    with patch.object(daemon, "_arm_exit_watchdog") as watchdog:
+        await daemon._shutdown()
 
+    # RX is stopped before the radio it depends on is released.
+    daemon.dispatcher.stop.assert_awaited_once()
     frame_server.stop.assert_awaited_once()
     bridge.stop.assert_awaited_once()
     daemon.router.stop.assert_awaited_once()
     daemon.radio.cleanup.assert_called_once()
+    # The process is guaranteed to exit even if a thread lingers.
+    watchdog.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_shutdown_continues_past_a_step_that_hangs():
+    """One stuck step must not strand the rest of the sequence.
+
+    A hang here used to leave the companion listen sockets bound and the serial
+    port held, so a restart could not reopen the radio.
+    """
+    daemon = RepeaterDaemon(_base_config(), radio=SimpleNamespace(cleanup=MagicMock()))
+    daemon.config["radio_type"] = "none"
+    daemon.SHUTDOWN_STEP_TIMEOUT_S = 0.05
+
+    hang_forever = asyncio.Event()
+
+    async def _never_returns():
+        await hang_forever.wait()
+
+    daemon.companion_frame_servers = [SimpleNamespace(stop=_never_returns, port=5050)]
+    daemon.router = SimpleNamespace(stop=AsyncMock())
+    daemon.companion_bridges = {}
+
+    with patch.object(daemon, "_arm_exit_watchdog"):
+        await asyncio.wait_for(daemon._shutdown(), timeout=5)
+
+    # Everything after the wedged frame server still ran.
+    daemon.router.stop.assert_awaited_once()
+    daemon.radio.cleanup.assert_called_once()
 
 
 def test_main_entrypoint_success_and_fatal_paths(monkeypatch):