fix(collector): survive broker dropping MQTT connections without FIN

The meshcore-mqtt-broker destroys subscriber connections (keep-alive
timeout) without sending FIN/RST, leaving a half-open socket that paho
never sees as disconnected: the collector stays "healthy" but silently
stops ingesting forever.

- Enable TCP keepalive (idle 60s / intvl 30s / count 3) on every
  connect, unwrapping paho's _WebsocketWrapper to reach the raw socket;
  kernel now fails zombie sockets so paho reconnects and resubscribes
- Verify SUBACKs via on_subscribe; log broker refusals (0x80) as ERROR
- Resubscribe with the registered QoS instead of paho's QoS 0 default
- Drop the optimistic _mqtt_connected flag; health now reports
  mqtt.is_connected, driven by real CONNACK/disconnect callbacks

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")
 

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}")
 

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")

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 == {}