From 2142451e4c9b3f3f7cd38b0b3176d5d04599d076 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 00:04:54 +0000 Subject: [PATCH 1/2] fix: keep paho network thread alive when MQTT-info refresh fails on disconnect _mqtt_on_disconnect runs on paho's network thread and blocked it with run_coroutine_threadsafe(...).result(). If the cloud API was unreachable at that moment (typically a DNS timeout right after a router reboot), DeyeCloudApiCannotConnectError escaped the callback and paho terminated the thread: the client never reconnected, is_connected() kept returning True because _loop_rc_handle sets CONNECTION_LOST only after the callback, and published commands were silently dropped. Blocking on the loop could also deadlock with disconnect(), which joins that thread. Refresh the MQTT info in the background instead, bounded by a timeout, log failures, deduplicate in-flight refreshes, and cancel a pending refresh on disconnect(). paho keeps reconnecting with the previous credentials; a refused CONNACK re-enters on_disconnect. Fixes #67 Co-authored-by: Stackie Jia --- CHANGELOG.rst | 9 +++ src/libdeye/mqtt_client.py | 49 ++++++++++++- tests/test_mqtt_client.py | 143 +++++++++++++++++++++++++++++++++++-- 3 files changed, 194 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 33d5684..8ca9c74 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -5,6 +5,15 @@ 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 `_). + Version 3.0.3 ============= diff --git a/src/libdeye/mqtt_client.py b/src/libdeye/mqtt_client.py index c7b6455..f4275af 100644 --- a/src/libdeye/mqtt_client.py +++ b/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: diff --git a/tests/test_mqtt_client.py b/tests/test_mqtt_client.py index 5ba2400..a3131e4 100644 --- a/tests/test_mqtt_client.py +++ b/tests/test_mqtt_client.py @@ -17,6 +17,7 @@ DeyeApiResponseFogPlatformMqttInfo, DeyeApiResponseFogPlatformMqttTopics, DeyeCloudApi, + DeyeCloudApiCannotConnectError, DeyeIotPlatform, ) from libdeye.const import ( @@ -178,11 +179,143 @@ 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, @@ -190,7 +323,9 @@ def test_mqtt_on_disconnect_unexpected( 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.""" From 7ffa884b7a171dc909f79f730bf8d985f723591d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 4 Sep 2026 00:05:02 +0000 Subject: [PATCH 2/2] fix: never let a malformed MQTT payload raise inside paho's network thread _mqtt_on_message only caught JSONDecodeError and KeyError; any other error from payload processing (e.g. TypeError, UnicodeDecodeError) would escape the paho callback and terminate the network thread with the same symptoms as #67. Decode the payload once, log and drop it on any error. Co-authored-by: Stackie Jia --- CHANGELOG.rst | 2 ++ src/libdeye/mqtt_client.py | 13 +++++++------ tests/test_mqtt_client.py | 26 ++++++++++++++++++++++++++ 3 files changed, 35 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 8ca9c74..7b6695d 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -13,6 +13,8 @@ Unreleased 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 `_). +- Never let a malformed MQTT payload raise inside paho's network thread; log + and ignore the message instead. Version 3.0.3 ============= diff --git a/src/libdeye/mqtt_client.py b/src/libdeye/mqtt_client.py index f4275af..5282964 100644 --- a/src/libdeye/mqtt_client.py +++ b/src/libdeye/mqtt_client.py @@ -504,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, diff --git a/tests/test_mqtt_client.py b/tests/test_mqtt_client.py index a3131e4..29d1df8 100644 --- a/tests/test_mqtt_client.py +++ b/tests/test_mqtt_client.py @@ -383,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"