diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py index 222a246..0151658 100644 --- a/src/meshcore_hub/collector/subscriber.py +++ b/src/meshcore_hub/collector/subscriber.py @@ -78,7 +78,6 @@ class Subscriber(LetsMeshNormalizer): self._running = False self._shutdown_event = threading.Event() self._handlers: dict[str, EventHandler] = {} - self._mqtt_connected = False self._db_connected = False self._health_reporter: Optional[HealthReporter] = None # Webhook processing @@ -131,7 +130,7 @@ class Subscriber(LetsMeshNormalizer): Returns: True if MQTT and database are connected """ - return self._running and self._mqtt_connected and self._db_connected + return self._running and self.mqtt.is_connected and self._db_connected def _load_channel_keys_from_db(self) -> bool: """Load channel keys from the database (synchronous). @@ -204,7 +203,7 @@ class Subscriber(LetsMeshNormalizer): return { "healthy": self.is_healthy, "running": self._running, - "mqtt_connected": self._mqtt_connected, + "mqtt_connected": self.mqtt.is_connected, "database_connected": self._db_connected, } @@ -769,11 +768,11 @@ class Subscriber(LetsMeshNormalizer): try: self.mqtt.connect() self.mqtt.start_background() - self._mqtt_connected = True - logger.info("Connected to MQTT broker") + logger.info( + "MQTT connection initiated (awaiting broker acknowledgement)" + ) break except Exception as e: - self._mqtt_connected = False if attempt < max_retries: logger.warning( "MQTT connection attempt %d/%d failed: %s. Retrying in %.1fs...", @@ -880,7 +879,6 @@ class Subscriber(LetsMeshNormalizer): # Stop MQTT self.mqtt.stop() self.mqtt.disconnect() - self._mqtt_connected = False logger.info("Collector subscriber stopped") diff --git a/src/meshcore_hub/common/mqtt.py b/src/meshcore_hub/common/mqtt.py index b4e4489..6be0fef 100644 --- a/src/meshcore_hub/common/mqtt.py +++ b/src/meshcore_hub/common/mqtt.py @@ -2,6 +2,7 @@ import json import logging +import socket from dataclasses import dataclass from typing import Any, Callable, Optional @@ -174,6 +175,7 @@ class MQTTClient: ) self._connected = False self._message_handlers: dict[str, list[MessageHandler]] = {} + self._topic_qos: dict[str, int] = {} # Set WebSocket path when using MQTT over WebSockets. if transport == "websockets": @@ -193,6 +195,67 @@ class MQTTClient: self._client.on_connect = self._on_connect self._client.on_disconnect = self._on_disconnect self._client.on_message = self._on_message + self._client.on_subscribe = self._on_subscribe + + def _apply_socket_keepalive(self) -> None: + """Enable TCP keepalive on the active socket. + + Some brokers destroy connections without sending a FIN/RST, which + leaves a half-open socket that paho never sees as disconnected. TCP + keepalive lets the kernel probe the peer and fail the socket so paho + can reconnect and resubscribe. + + With transport=websockets, paho's socket() returns a _WebsocketWrapper; + the raw socket is unwrapped from its ``_socket`` attribute. Best-effort: + failures (unsupported option, no socket yet) are logged at DEBUG and + never fatal. + """ + sock = getattr(self._client, "socket", lambda: None)() + if sock is None: + logger.debug("No MQTT socket available for TCP keepalive setup") + return + if not hasattr(sock, "setsockopt"): + raw = getattr(sock, "_socket", None) + if raw is None or not hasattr(raw, "setsockopt"): + logger.debug( + "MQTT socket does not expose setsockopt; skipping TCP keepalive" + ) + return + sock = raw + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + if hasattr(socket, "TCP_KEEPIDLE"): + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60) + if hasattr(socket, "TCP_KEEPINTVL"): + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 30) + if hasattr(socket, "TCP_KEEPCNT"): + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3) + logger.debug("TCP keepalive enabled on MQTT socket") + except (OSError, AttributeError) as e: + logger.debug("Could not enable TCP keepalive on MQTT socket: %s", e) + + def _on_subscribe( + self, + client: mqtt.Client, + userdata: Any, + mid: Any, + reason_code_list: Any, + properties: Any = None, + ) -> None: + """Handle subscribe callback: verify broker granted the subscription.""" + reason_codes = ( + list(reason_code_list) + if reason_code_list is not None and not isinstance(reason_code_list, int) + else [reason_code_list] + ) + for rc in reason_codes: + value = getattr(rc, "value", rc) + if value == 0x80 or (isinstance(value, str) and "0x80" in value): + logger.error( + "MQTT broker refused subscription (mid=%s, reason=%s)", mid, rc + ) + else: + logger.debug("Subscription acknowledged (mid=%s, granted=%s)", mid, rc) def _on_connect( self, @@ -208,10 +271,13 @@ class MQTTClient: logger.info( f"Connected to MQTT broker at {self.config.host}:{self.config.port}" ) + # Enable TCP keepalive on the fresh socket so half-open + # connections are detected and paho can reconnect. + self._apply_socket_keepalive() # Resubscribe to topics on reconnect - for topic in self._message_handlers.keys(): - self._client.subscribe(topic) - logger.debug(f"Resubscribed to topic: {topic}") + for topic, qos in self._topic_qos.items(): + self._client.subscribe(topic, qos) + logger.debug(f"Resubscribed to topic: {topic} (qos={qos})") else: logger.error(f"Failed to connect to MQTT broker: {reason_code}") @@ -317,6 +383,7 @@ class MQTTClient: """ if topic not in self._message_handlers: self._message_handlers[topic] = [] + self._topic_qos[topic] = qos if self._connected: self._client.subscribe(topic, qos) logger.debug(f"Subscribed to topic: {topic}") @@ -331,6 +398,7 @@ class MQTTClient: """ if topic in self._message_handlers: del self._message_handlers[topic] + self._topic_qos.pop(topic, None) self._client.unsubscribe(topic) logger.debug(f"Unsubscribed from topic: {topic}") diff --git a/tests/test_collector/test_subscriber.py b/tests/test_collector/test_subscriber.py index 1882651..b080a98 100644 --- a/tests/test_collector/test_subscriber.py +++ b/tests/test_collector/test_subscriber.py @@ -14,6 +14,7 @@ class TestSubscriber: def mock_mqtt_client(self): """Create a mock MQTT client.""" client = MagicMock() + client.is_connected = False client.topic_builder = MagicMock() client.topic_builder.prefix = "meshcore" client.topic_builder.all_events_topic.return_value = "meshcore/+/event/#" @@ -1224,6 +1225,7 @@ class TestSubscriberDispatch: def mock_mqtt_client(self): """Create a mock MQTT client.""" client = MagicMock() + client.is_connected = False client.topic_builder = MagicMock() client.topic_builder.prefix = "meshcore" client.topic_builder.all_events_topic.return_value = "meshcore/+/event/#" @@ -1298,9 +1300,23 @@ class TestSubscriberDispatch: subscriber.start() assert mock_mqtt_client.connect.call_count == 2 - assert subscriber._mqtt_connected is True + assert subscriber._running is True subscriber.stop() + def test_health_reflects_mqtt_connection_state(self, mock_mqtt_client, db_manager): + """Health tracks the client's real connection state, not an optimistic flag.""" + subscriber = Subscriber(mock_mqtt_client, db_manager) + subscriber._running = True + subscriber._db_connected = True + + mock_mqtt_client.is_connected = False + assert subscriber.is_healthy is False + assert subscriber.get_health_status()["mqtt_connected"] is False + + mock_mqtt_client.is_connected = True + assert subscriber.is_healthy is True + assert subscriber.get_health_status()["mqtt_connected"] is True + def test_start_mqtt_all_retries_exhausted(self, mock_mqtt_client, db_manager): """Subscriber raises when all MQTT retries fail.""" mock_mqtt_client.connect.side_effect = ConnectionError("fail") diff --git a/tests/test_common/test_mqtt.py b/tests/test_common/test_mqtt.py index 58ab9c7..32e0478 100644 --- a/tests/test_common/test_mqtt.py +++ b/tests/test_common/test_mqtt.py @@ -1,6 +1,13 @@ """Tests for MQTT topic parsing utilities.""" -from meshcore_hub.common.mqtt import TopicBuilder +import logging +import socket +from types import SimpleNamespace +from unittest.mock import MagicMock, call + +import pytest + +from meshcore_hub.common.mqtt import MQTTClient, MQTTConfig, TopicBuilder class TestTopicBuilder: @@ -55,3 +62,172 @@ class TestTopicBuilder: ) assert parsed is None + + +class TestMQTTClient: + """Tests for the MQTTClient wrapper (keepalive, SUBACK, resubscribe).""" + + @pytest.fixture + def wired(self) -> tuple[MQTTClient, MagicMock]: + """Create an MQTTClient whose paho client is a MagicMock.""" + instance = MQTTClient(MQTTConfig()) + paho_mock = MagicMock() + instance._client = paho_mock + return instance, paho_mock + + def test_apply_socket_keepalive_sets_sockopts( + self, wired: tuple[MQTTClient, MagicMock] + ) -> None: + """TCP keepalive options are applied to the active socket.""" + client, paho_mock = wired + sock = MagicMock() + paho_mock.socket.return_value = sock + + client._apply_socket_keepalive() + + assert sock.setsockopt.call_args_list == [ + call(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), + call(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60), + call(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 30), + call(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3), + ] + + def test_apply_socket_keepalive_without_socket( + self, wired: tuple[MQTTClient, MagicMock] + ) -> None: + """Missing socket (not yet connected) is tolerated silently.""" + client, paho_mock = wired + paho_mock.socket.return_value = None + + client._apply_socket_keepalive() # must not raise + + paho_mock.socket.assert_called_once() + + def test_apply_socket_keepalive_swallows_oserror( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """OSError from setsockopt is swallowed (best-effort).""" + client, paho_mock = wired + sock = MagicMock() + sock.setsockopt.side_effect = OSError("unsupported") + paho_mock.socket.return_value = sock + + with caplog.at_level(logging.DEBUG, logger="meshcore_hub.common.mqtt"): + client._apply_socket_keepalive() # must not raise + + assert any("TCP keepalive" in r.message for r in caplog.records) + + def test_on_connect_applies_keepalive_and_resubscribes_with_qos( + self, wired: tuple[MQTTClient, MagicMock] + ) -> None: + """Successful CONNACK enables keepalive and resubscribes at stored QoS.""" + client, paho_mock = wired + client.subscribe("meshcore/+/+/packets", handler=lambda *a: None, qos=1) + client.subscribe("meshcore/+/+/status", handler=lambda *a: None, qos=1) + paho_mock.subscribe.reset_mock() + paho_mock.socket.return_value = MagicMock() + + client._on_connect(paho_mock, None, None, 0) + + assert client.is_connected is True + assert paho_mock.socket.return_value.setsockopt.call_count >= 1 + assert paho_mock.subscribe.call_args_list == [ + call("meshcore/+/+/packets", 1), + call("meshcore/+/+/status", 1), + ] + + def test_on_connect_failure_stays_disconnected( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """Refused CONNACK keeps the client disconnected and logs an error.""" + client, paho_mock = wired + + client._on_connect(paho_mock, None, None, SimpleNamespace(value=135)) + + assert client.is_connected is False + paho_mock.subscribe.assert_not_called() + assert any("Failed to connect" in r.message for r in caplog.records) + + def test_apply_socket_keepalive_unwraps_websocket_wrapper( + self, wired: tuple[MQTTClient, MagicMock] + ) -> None: + """Keepalive is applied to the raw socket inside paho's websocket wrapper.""" + client, paho_mock = wired + raw_sock = MagicMock() + wrapper = MagicMock(spec=["recv", "send", "close", "_socket"]) + wrapper._socket = raw_sock + paho_mock.socket.return_value = wrapper + + client._apply_socket_keepalive() + + assert raw_sock.setsockopt.call_args_list == [ + call(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), + call(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60), + call(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 30), + call(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3), + ] + + def test_apply_socket_keepalive_wrapper_without_raw_socket( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """A wrapper exposing no raw socket is skipped without raising.""" + client, paho_mock = wired + wrapper = MagicMock(spec=["recv", "send", "close"]) + paho_mock.socket.return_value = wrapper + + with caplog.at_level(logging.DEBUG, logger="meshcore_hub.common.mqtt"): + client._apply_socket_keepalive() # must not raise + + assert any("setsockopt" in r.message for r in caplog.records) + + def test_on_subscribe_granted_is_debug_only( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """Granted SUBACKs are logged at DEBUG, not ERROR.""" + client, paho_mock = wired + with caplog.at_level(logging.DEBUG, logger="meshcore_hub.common.mqtt"): + client._on_subscribe( + paho_mock, None, mid=7, reason_code_list=[SimpleNamespace(value=0)] + ) + + assert not any(r.levelno == logging.ERROR for r in caplog.records) + assert any("acknowledged" in r.message for r in caplog.records) + + def test_on_subscribe_refusal_logs_error( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """A 0x80 SUBACK refusal is surfaced as an ERROR log.""" + client, paho_mock = wired + client._on_subscribe( + paho_mock, None, mid=7, reason_code_list=[SimpleNamespace(value=0x80)] + ) + + assert any( + r.levelno == logging.ERROR and "refused" in r.message + for r in caplog.records + ) + + def test_on_subscribe_accepts_plain_int_reason_code( + self, wired: tuple[MQTTClient, MagicMock], caplog: pytest.LogCaptureFixture + ) -> None: + """A bare int reason code (defensive path) is handled.""" + client, paho_mock = wired + client._on_subscribe(paho_mock, None, mid=1, reason_code_list=0x80) + + assert any( + r.levelno == logging.ERROR and "refused" in r.message + for r in caplog.records + ) + + def test_subscribe_records_qos_for_resubscribe( + self, wired: tuple[MQTTClient, MagicMock] + ) -> None: + """Registered QoS is stored and removed with the subscription.""" + client, _ = wired + client.subscribe("meshcore/+/+/packets", handler=lambda *a: None, qos=1) + + assert client._topic_qos == {"meshcore/+/+/packets": 1} + + client.unsubscribe("meshcore/+/+/packets") + + assert client._topic_qos == {}