fix: keep paho network thread alive when MQTT-info refresh fails on disconnect

CHANGELOG.rst

@@ -5,6 +5,17 @@ Changelog
 Unreleased
 ==========
 
+- Fix the MQTT client never recovering from an unexpected disconnect when the
+  cloud API is unreachable at that moment (for example a DNS timeout during a
+  router reboot). ``on_disconnect`` blocked paho's network thread on the event
+  loop and let ``DeyeCloudApiCannotConnectError`` escape, which terminated the
+  thread: the client never reconnected while ``is_connected()`` still returned
+  True and published commands were silently dropped. The MQTT-info refresh now
+  runs in the background with a bounded timeout, failures are logged, and paho
+  keeps reconnecting on its own (`#67 <https://github.com/stackia/libdeye/issues/67>`_).
+- Never let a malformed MQTT payload raise inside paho's network thread; log
+  and ignore the message instead.
+
 Version 3.0.3
 =============
 

src/libdeye/mqtt_client.py

@@ -4,7 +4,9 @@
 import asyncio
 from asyncio import Future, get_running_loop
 from collections.abc import Callable
+import concurrent.futures
 import json
+import logging
 from ssl import SSLContext
 from typing import Any, cast, override
 
@@ -21,6 +23,14 @@
 from .device_command import DeyeDeviceCommand, fog_combo_frames_from_properties
 from .device_state import DeyeDeviceState
 
+_LOGGER = logging.getLogger(__name__)
+
+# Upper bound for re-fetching MQTT credentials after an unexpected disconnect.
+# The cloud is often unreachable at that moment (the disconnect and the cloud
+# outage usually share a cause), so this must not wait for aiohttp's much
+# longer default total timeout.
+MQTT_INFO_REFRESH_TIMEOUT = 30
+
 FogCommandBaseline = DeyeDeviceCommand | DeyeDeviceState
 
 # FogDeviceManager.sendCommand JSON keys, plus GET→SET aliases.
@@ -401,6 +411,7 @@ def __init__(
         self._mqtt.on_disconnect = self._mqtt_on_disconnect
         self._subscribers: dict[str, set[Callable[[Any], None]]] = {}
         self._pending_commands: list[tuple[str, bytes]] = []
+        self._mqtt_info_refresh: concurrent.futures.Future[None] | None = None
 
     @abstractmethod
     async def _set_mqtt_info(self) -> None:
@@ -415,6 +426,9 @@ async def connect(self) -> None:
 
     def disconnect(self) -> None:
         """Disconnect the MQTT client to the server."""
+        if self._mqtt_info_refresh is not None:
+            self._mqtt_info_refresh.cancel()
+            self._mqtt_info_refresh = None
         self._mqtt.disconnect()
         self._mqtt.loop_stop()
 
@@ -445,9 +459,38 @@ def _mqtt_on_disconnect(
         if reason_code == 0:  # User initiated disconnect
             return
 
-        # Update MQTT info and wait for it to complete before reconnecting
-        # (reconnect is automatically handled by paho-mqtt by default)
-        asyncio.run_coroutine_threadsafe(self._set_mqtt_info(), self._loop).result()
+        # This callback runs on paho's network thread, which also performs the
+        # automatic reconnect. It must neither block on the event loop (the
+        # loop may be inside ``disconnect()`` joining this very thread) nor
+        # raise: an exception escaping a paho callback terminates the network
+        # thread, so the client would never reconnect while ``is_connected()``
+        # keeps returning True. Refresh the credentials in the background;
+        # paho reconnects with the previous ones meanwhile, and a refused
+        # CONNACK simply brings us back here.
+        if self._mqtt_info_refresh is not None and not self._mqtt_info_refresh.done():
+            return
+        refresh = self._refresh_mqtt_info_after_disconnect(reason_code)
+        try:
+            self._mqtt_info_refresh = asyncio.run_coroutine_threadsafe(
+                refresh, self._loop
+            )
+        except RuntimeError:  # Event loop is closed
+            refresh.close()
+
+    async def _refresh_mqtt_info_after_disconnect(
+        self, reason_code: mqtt.ReasonCode
+    ) -> None:
+        """Re-fetch MQTT credentials so paho's automatic reconnect can use them."""
+        try:
+            async with asyncio.timeout(MQTT_INFO_REFRESH_TIMEOUT):
+                await self._set_mqtt_info()
+        except Exception as err:  # noqa: BLE001
+            _LOGGER.warning(
+                "MQTT disconnected (%s) and refreshing the MQTT connection info "
+                "failed (%s); reconnecting with the previous credentials",
+                reason_code,
+                str(err.__cause__ or err) or type(err).__name__,
+            )
 
     @abstractmethod
     def _process_message_payload(self, msg: mqtt.MQTTMessage) -> Any:
@@ -461,12 +504,13 @@ def _mqtt_on_message(
             return
         callbacks = self._subscribers[msg.topic]
         try:
-            for callback in callbacks.copy():
-                self._loop.call_soon_threadsafe(
-                    callback, self._process_message_payload(msg)
-                )
-        except json.JSONDecodeError, KeyError:
-            pass
+            payload = self._process_message_payload(msg)
+        except Exception as err:  # noqa: BLE001
+            # Same rule as above: nothing may escape into paho's network thread.
+            _LOGGER.warning("Ignoring malformed MQTT message on %s: %r", msg.topic, err)
+            return
+        for callback in callbacks.copy():
+            self._loop.call_soon_threadsafe(callback, payload)
 
     def _subscribe_topic(
         self,

tests/test_mqtt_client.py

@@ -17,6 +17,7 @@
     DeyeApiResponseFogPlatformMqttInfo,
     DeyeApiResponseFogPlatformMqttTopics,
     DeyeCloudApi,
+    DeyeCloudApiCannotConnectError,
     DeyeIotPlatform,
 )
 from libdeye.const import (
@@ -178,19 +179,153 @@ def test_mqtt_on_disconnect_user_initiated(
             )
             mock_run_coroutine_threadsafe.assert_not_called()
 
-    def test_mqtt_on_disconnect_unexpected(
+    @staticmethod
+    async def _unexpected_disconnect(client: MockBaseDeyeMqttClient) -> None:
+        """Invoke on_disconnect from a worker thread, like paho's network loop."""
+        await asyncio.to_thread(
+            client._mqtt_on_disconnect,
+            client._mqtt,
+            None,
+            mqtt.DisconnectFlags(False),
+            mqtt.ReasonCode(mqtt.PacketTypes.DISCONNECT, "Unspecified error"),
+            None,
+        )
+
+    @pytest.mark.asyncio
+    async def test_mqtt_on_disconnect_unexpected_refreshes_in_background(
         self, base_client: MockBaseDeyeMqttClient
     ) -> None:
-        """Test _mqtt_on_disconnect method with unexpected disconnect."""
-        with patch("asyncio.run_coroutine_threadsafe") as mock_run_coroutine_threadsafe:
+        """Test an unexpected disconnect refreshes MQTT info without blocking paho."""
+        release = asyncio.Event()
+
+        async def slow_set_mqtt_info() -> None:
+            await release.wait()
+
+        with patch.object(
+            base_client, "_set_mqtt_info", AsyncMock(side_effect=slow_set_mqtt_info)
+        ) as mock_set_mqtt_info:
+            await self._unexpected_disconnect(base_client)
+
+            # The callback returned while the refresh is still pending.
+            assert base_client._mqtt_info_refresh is not None
+            assert not base_client._mqtt_info_refresh.done()
+
+            release.set()
+            await asyncio.wrap_future(base_client._mqtt_info_refresh)
+            mock_set_mqtt_info.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_mqtt_on_disconnect_refresh_failure_is_logged(
+        self, base_client: MockBaseDeyeMqttClient, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test a cloud failure during refresh never reaches paho's thread."""
+        error = DeyeCloudApiCannotConnectError()
+        error.__cause__ = OSError("Timeout while contacting DNS servers")
+        with patch.object(base_client, "_set_mqtt_info", AsyncMock(side_effect=error)):
+            await self._unexpected_disconnect(base_client)
+            assert base_client._mqtt_info_refresh is not None
+            await asyncio.wrap_future(base_client._mqtt_info_refresh)
+
+        assert base_client._mqtt_info_refresh.exception() is None
+        assert "Timeout while contacting DNS servers" in caplog.text
+        assert "reconnecting with the previous credentials" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_mqtt_on_disconnect_refresh_timeout_is_logged(
+        self, base_client: MockBaseDeyeMqttClient, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test a hanging cloud request is bounded and logged."""
+
+        async def hang() -> None:
+            await asyncio.sleep(60)
+
+        with (
+            patch("libdeye.mqtt_client.MQTT_INFO_REFRESH_TIMEOUT", 0.01),
+            patch.object(base_client, "_set_mqtt_info", AsyncMock(side_effect=hang)),
+        ):
+            await self._unexpected_disconnect(base_client)
+            assert base_client._mqtt_info_refresh is not None
+            await asyncio.wrap_future(base_client._mqtt_info_refresh)
+
+        assert base_client._mqtt_info_refresh.exception() is None
+        assert "TimeoutError" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_mqtt_on_disconnect_skips_duplicate_refresh(
+        self, base_client: MockBaseDeyeMqttClient
+    ) -> None:
+        """Test repeated disconnects share one in-flight refresh."""
+        release = asyncio.Event()
+
+        async def slow_set_mqtt_info() -> None:
+            await release.wait()
+
+        with patch.object(
+            base_client, "_set_mqtt_info", AsyncMock(side_effect=slow_set_mqtt_info)
+        ) as mock_set_mqtt_info:
+            await self._unexpected_disconnect(base_client)
+            first = base_client._mqtt_info_refresh
+            await self._unexpected_disconnect(base_client)
+            assert base_client._mqtt_info_refresh is first
+
+            release.set()
+            assert first is not None
+            await asyncio.wrap_future(first)
+            mock_set_mqtt_info.assert_awaited_once()
+
+            # Once finished, the next disconnect refreshes again.
+            await self._unexpected_disconnect(base_client)
+            second = base_client._mqtt_info_refresh
+            assert second is not None
+            assert second is not first
+            await asyncio.wrap_future(second)
+            assert mock_set_mqtt_info.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_disconnect_cancels_pending_refresh(
+        self, base_client: MockBaseDeyeMqttClient
+    ) -> None:
+        """Test disconnect() cancels a refresh that is still in flight."""
+
+        async def hang() -> None:
+            await asyncio.sleep(60)
+
+        with (
+            patch.object(base_client, "_set_mqtt_info", AsyncMock(side_effect=hang)),
+            patch.object(base_client._mqtt, "disconnect") as mock_disconnect,
+            patch.object(base_client._mqtt, "loop_stop") as mock_loop_stop,
+        ):
+            await self._unexpected_disconnect(base_client)
+            pending = base_client._mqtt_info_refresh
+            assert pending is not None
+
+            base_client.disconnect()
+            await asyncio.sleep(0)
+
+        assert pending.cancelled()
+        assert base_client._mqtt_info_refresh is None
+        mock_disconnect.assert_called_once()
+        mock_loop_stop.assert_called_once()
+
+    def test_mqtt_on_disconnect_with_closed_event_loop(
+        self, base_client: MockBaseDeyeMqttClient
+    ) -> None:
+        """Test a disconnect after the event loop closed is a no-op."""
+        closed_loop = asyncio.new_event_loop()
+        closed_loop.close()
+        base_client._loop = closed_loop
+
+        with patch.object(base_client, "_set_mqtt_info") as mock_set_mqtt_info:
             base_client._mqtt_on_disconnect(
                 base_client._mqtt,
                 None,
                 mqtt.DisconnectFlags(False),
                 mqtt.ReasonCode(mqtt.PacketTypes.DISCONNECT, "Unspecified error"),
                 None,
             )
-            mock_run_coroutine_threadsafe.assert_called_once()
+
+        assert base_client._mqtt_info_refresh is None
+        mock_set_mqtt_info.assert_not_called()
 
     def test_mqtt_on_message(self, base_client: MockBaseDeyeMqttClient) -> None:
         """Test _mqtt_on_message method."""
@@ -248,6 +383,32 @@ def test_mqtt_on_message_json_error(
             base_client._mqtt_on_message(base_client._mqtt, None, message)
             mock_call_soon_threadsafe.assert_not_called()
 
+    def test_mqtt_on_message_payload_error_is_logged(
+        self, base_client: MockBaseDeyeMqttClient, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test any payload processing error is swallowed, not raised into paho."""
+        topic = "test/topic"
+        callback = MagicMock()
+        base_client._subscribers = {topic: {callback}}
+        message = MagicMock(spec=mqtt.MQTTMessage)
+        message.topic = topic
+        message.payload = b"{}"
+
+        with (
+            patch.object(
+                base_client,
+                "_process_message_payload",
+                side_effect=TypeError("unexpected payload shape"),
+            ),
+            patch.object(
+                base_client._loop, "call_soon_threadsafe"
+            ) as mock_call_soon_threadsafe,
+        ):
+            base_client._mqtt_on_message(base_client._mqtt, None, message)
+
+        mock_call_soon_threadsafe.assert_not_called()
+        assert "Ignoring malformed MQTT message on test/topic" in caplog.text
+
     def test_subscribe_topic(self, base_client: MockBaseDeyeMqttClient) -> None:
         """Test _subscribe_topic method."""
         topic = "test/topic"