diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 33d5684..7b6695d 100644 --- a/CHANGELOG.rst +++ b/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 `_). +- 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 c7b6455..5282964 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: @@ -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, diff --git a/tests/test_mqtt_client.py b/tests/test_mqtt_client.py index 5ba2400..29d1df8 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.""" @@ -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"