diff --git a/README.md b/README.md index 08fca04..a6017f2 100644 --- a/README.md +++ b/README.md @@ -214,6 +214,14 @@ action: apply on reload. - The identity is stored in `config/reticulum/identity` — back it up to keep the same address. +- RNS is a standalone-daemon-style library. To embed it cleanly the integration + suppresses RNS's own signal handlers (so Home Assistant keeps ownership of + SIGINT/SIGTERM — important under Kubernetes) and unregisters its blocking + `atexit` persistence, persisting RNS/LXMF state itself, off the event loop, + on the `homeassistant_stop` event. RNS still spawns a couple of non-daemon + worker threads, so on shutdown you may see a "non-daemonic threads" notice and + a short delay before the process exits (within the pod's termination grace + period). - Assist replies use the standard conversation pipeline, so whichever agent you select (built-in intents, a local LLM, a cloud LLM, …) is what answers. diff --git a/custom_components/reticulum/__init__.py b/custom_components/reticulum/__init__.py index 7ef57ad..302c504 100644 --- a/custom_components/reticulum/__init__.py +++ b/custom_components/reticulum/__init__.py @@ -9,7 +9,7 @@ from typing import Any import voluptuous as vol from homeassistant.config_entries import ConfigEntry -from homeassistant.const import Platform +from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform from homeassistant.core import ( HomeAssistant, ServiceCall, @@ -121,6 +121,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ) ) + # Persist RNS/LXMF state off-loop when Home Assistant stops (we unregister + # RNS's own atexit handlers, which would otherwise do this with blocking I/O + # on the event loop and detach HA's stdout). + async def _on_ha_stop(_event) -> None: + await manager.async_persist() + + runtime.unsubs.append( + hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _on_ha_stop) + ) + entry.async_on_unload(entry.add_update_listener(_async_options_updated)) _async_register_services(hass) diff --git a/custom_components/reticulum/manifest.json b/custom_components/reticulum/manifest.json index 56bf04a..64d62a1 100644 --- a/custom_components/reticulum/manifest.json +++ b/custom_components/reticulum/manifest.json @@ -12,5 +12,5 @@ "loggers": ["RNS", "LXMF"], "requirements": ["rns>=0.9.0", "lxmf>=0.6.0"], "single_config_entry": true, - "version": "1.0.1" + "version": "1.0.2" } diff --git a/custom_components/reticulum/reticulum_client.py b/custom_components/reticulum/reticulum_client.py index ed7f635..648b0d4 100644 --- a/custom_components/reticulum/reticulum_client.py +++ b/custom_components/reticulum/reticulum_client.py @@ -13,8 +13,11 @@ run a standalone (non-shared) RNS instance inside the Home Assistant process. from __future__ import annotations import asyncio +import atexit import logging import os +import signal +import threading import time from collections import deque from dataclasses import dataclass, field @@ -172,22 +175,9 @@ class ReticulumManager: _RNS_INSTANCE = existing else: _LOGGER.debug("Initialising Reticulum instance at %s", self.storage_dir) - try: - _RNS_INSTANCE = RNS.Reticulum( - configdir=self.storage_dir, loglevel=self.loglevel - ) - except OSError as err: - # Lost a race (or divergent state): adopt whatever is running. - adopted = None - try: - adopted = RNS.Reticulum.get_instance() - except Exception: # noqa: BLE001 - adopted = None - if adopted is None: - raise - _LOGGER.debug("Adopted existing Reticulum instance after %s", err) - _RNS_INSTANCE = adopted + _RNS_INSTANCE = self._create_rns_instance(RNS) self._rns = _RNS_INSTANCE + self._tame_exit_handler(RNS.Reticulum) # Stable identity so our LXMF address survives restarts. if _IDENTITY is None: @@ -213,6 +203,7 @@ class ReticulumManager: self._router = _LXM_ROUTER self._local_destination = _LOCAL_DESTINATION self._router.register_delivery_callback(self._delivery_callback) + self._tame_exit_handler(self._router) # Discover peers via their LXMF delivery announces. Deregister any # previous handler (bound to a stale manager) before installing ours. @@ -255,6 +246,75 @@ class ReticulumManager: with open(config_path, "w", encoding="utf-8") as handle: handle.write(contents) + def _create_rns_instance(self, rns: Any) -> Any: + """Create the RNS instance from an executor thread. + + ``Reticulum.__init__`` calls ``signal.signal()`` with no main-thread + guard, but signal handlers can only be installed from the main thread. + Since we (correctly) initialise off the event loop, we temporarily + neutralise ``signal.signal`` so init completes — and as a bonus this + stops RNS from hijacking Home Assistant's own SIGINT/SIGTERM handling, + which HA needs for clean shutdown (important under Kubernetes). + """ + saved_signal = signal.signal + if threading.current_thread() is not threading.main_thread(): + signal.signal = lambda *args, **kwargs: None # type: ignore[assignment] + try: + return rns.Reticulum( + configdir=self.storage_dir, loglevel=self.loglevel + ) + except OSError as err: + # Divergent state / race: adopt whatever is already running. + adopted = None + try: + adopted = rns.Reticulum.get_instance() + except Exception: # noqa: BLE001 + adopted = None + if adopted is None: + raise + _LOGGER.debug("Adopted existing Reticulum instance after %s", err) + return adopted + finally: + signal.signal = saved_signal # type: ignore[assignment] + + @staticmethod + def _tame_exit_handler(obj: Any) -> None: + """Unregister an RNS/LXMF ``atexit`` exit handler. + + RNS and LXMF register ``atexit`` handlers that persist state with + blocking file I/O on the main (event-loop) thread and detach + stdout/stderr — both of which trip HA's loop-protection and can suppress + HA's own shutdown logging. We persist state ourselves off-loop at HA + stop instead (see :meth:`async_persist`). + """ + handler = getattr(obj, "exit_handler", None) + if handler is None: + return + try: + atexit.unregister(handler) + except Exception: # noqa: BLE001 - best effort + _LOGGER.debug("Could not unregister exit handler for %s", obj) + + async def async_persist(self) -> None: + """Persist RNS/LXMF state off the event loop (call at HA stop).""" + if self._rns is None: + return + await self.hass.async_add_executor_job(self._persist) + + def _persist(self) -> None: + import RNS # noqa: PLC0415 + + for label, fn in ( + ("transport", getattr(RNS.Transport, "persist_data", None)), + ("identity", getattr(RNS.Identity, "persist_data", None)), + ): + if fn is None: + continue + try: + fn() + except Exception: # noqa: BLE001 - best effort + _LOGGER.debug("Failed to persist RNS %s data", label, exc_info=True) + async def async_stop(self) -> None: """Detach our callbacks. The RNS instance itself lives for the process.