adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-07-26 19:42:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(repeater): make SIGTERM actually shut the daemon down

Browse Source 
SIGTERM did not stop the daemon. Observed: the process stayed alive 18+
minutes with all three companion listen sockets still bound and the serial
port still held, so a restart could not reopen the radio. Two defects
compounded.

Cleanup never ran. The signal handler cancelled run(), whose finally then
awaited _shutdown() from inside an already-cancelled task -- so the first
await raised CancelledError, run() returned, and asyncio.run() tore the loop
down before a single shutdown step executed. Not one step logged. Running
cleanup in a sibling task instead does not fix it either: run() returns as
soon as the dispatcher stops and asyncio.run() cancels every leftover task
on the way out, which showed up as 'frame server :5050' being cancelled
mid-stop. The handler now unwinds run() cooperatively by stopping the
dispatcher, so run_forever() returns on its own and cleanup runs in a task
that is not being cancelled. Cancelling is kept only as a fallback for a
failure before the dispatcher exists.

Nothing bounded the steps. Frame servers, bridges, router and Glass had no
timeout, so one stuck step stranded every step after it -- including
releasing the radio. Each step is now bounded by SHUTDOWN_STEP_TIMEOUT_S and
logged by name, so a hang is both survivable and diagnosable, and the sync
steps (HTTP stop, sensor manager, GPS, radio cleanup) run off-loop so a
blocking close cannot stall the sequence. The dispatcher is stopped first so
RX ends before its radio is released.

Even with cleanup fixed, a single non-daemon thread that never returns hangs
SIGTERM forever: interpreter finalization joins them with no timeout, which
is where the original 18-minute hang sat (main thread parked in
Py_FinalizeEx -> wait_for_thread_shutdown). Report any that are still alive
by name so the offender can be fixed at the source, excluding asyncio's own
executor workers since asyncio.run() joins those under its own timeout. Then
arm a daemon watchdog that forces the process down SHUTDOWN_EXIT_GRACE_S
after cleanup finishes, so a future stray thread costs a delayed exit rather
than a stuck service.

Verified on hardware: SIGTERM and SIGINT both exit in 1s with zero warnings,
all ports and the serial device released, and MQTT publishing its offline
status before disconnecting.
This commit is contained in:
agessaman
2026-07-24 15:29:40 -07:00
parent 05dae761e7
commit 8310ccdee6
2 changed files with 271 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/main.py
+128 -50
View File
@@ -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
@@ -1547,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:
@@ -1645,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
+143 -9
View File
@@ -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):