diff --git a/src/libdeye/mqtt_client.py b/src/libdeye/mqtt_client.py index fbab33c..f85e001 100644 --- a/src/libdeye/mqtt_client.py +++ b/src/libdeye/mqtt_client.py @@ -2,9 +2,11 @@ import asyncio import json +import logging from abc import ABC, abstractmethod from asyncio import Future, get_running_loop from collections.abc import Callable +from concurrent.futures import Future as ConcurrentFuture from ssl import SSLContext from typing import Any, cast @@ -19,6 +21,10 @@ from .device_command import DeyeDeviceCommand from .device_state import DeyeDeviceState +_LOGGER = logging.getLogger(__name__) + +MQTT_INFO_REFRESH_TIMEOUT = 15 + class BaseDeyeMqttClient(ABC): """Base class for MQTT clients connected to Deye MQTT servers.""" @@ -43,6 +49,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_future: ConcurrentFuture[None] | None = None @abstractmethod async def _set_mqtt_info(self) -> None: @@ -57,6 +64,11 @@ async def connect(self) -> None: def disconnect(self) -> None: """Disconnect the MQTT client to the server.""" + if ( + self._mqtt_info_refresh_future is not None + and not self._mqtt_info_refresh_future.done() + ): + self._mqtt_info_refresh_future.cancel() self._mqtt.disconnect() self._mqtt.loop_stop() @@ -81,9 +93,49 @@ def _mqtt_on_disconnect( if result_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() + _LOGGER.warning( + "Deye MQTT disconnected unexpectedly, result_code=%s", result_code + ) + + # This callback runs on Paho's network thread. Do not block it while a cloud + # request refreshes the MQTT credentials, and never let a cloud/DNS failure + # escape the callback and terminate that thread. + if ( + self._mqtt_info_refresh_future is not None + and not self._mqtt_info_refresh_future.done() + ): + return + + refresh_coro = self._refresh_mqtt_info_after_disconnect(result_code) + try: + self._mqtt_info_refresh_future = asyncio.run_coroutine_threadsafe( + refresh_coro, self._loop + ) + except Exception: + refresh_coro.close() + _LOGGER.exception( + "Failed to schedule a Deye MQTT information refresh after disconnect" + ) + + async def _refresh_mqtt_info_after_disconnect(self, result_code: int) -> None: + """Refresh MQTT credentials without risking the Paho network thread.""" + try: + await asyncio.wait_for( + self._set_mqtt_info(), timeout=MQTT_INFO_REFRESH_TIMEOUT + ) + except TimeoutError: + _LOGGER.error( + "Timed out after %ss while refreshing Deye MQTT information " + "following disconnect (result_code=%s)", + MQTT_INFO_REFRESH_TIMEOUT, + result_code, + ) + except Exception: + _LOGGER.exception( + "Failed to refresh Deye MQTT information following disconnect " + "(result_code=%s)", + result_code, + ) @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 d16a39b..383ef21 100644 --- a/tests/test_mqtt_client.py +++ b/tests/test_mqtt_client.py @@ -20,6 +20,7 @@ from libdeye.device_command import DeyeDeviceCommand from libdeye.device_state import DeyeDeviceState from libdeye.mqtt_client import ( + MQTT_INFO_REFRESH_TIMEOUT, BaseDeyeMqttClient, DeyeClassicMqttClient, DeyeFogMqttClient, @@ -155,9 +156,60 @@ def test_mqtt_on_disconnect_unexpected( self, base_client: MockBaseDeyeMqttClient ) -> None: """Test _mqtt_on_disconnect method with unexpected disconnect.""" - with patch("asyncio.run_coroutine_threadsafe") as mock_run_coroutine_threadsafe: + refresh_future = MagicMock() + with patch( + "asyncio.run_coroutine_threadsafe", return_value=refresh_future + ) as mock_run_coroutine_threadsafe: base_client._mqtt_on_disconnect(base_client._mqtt, None, 1) mock_run_coroutine_threadsafe.assert_called_once() + refresh_future.result.assert_not_called() + + scheduled_coro = mock_run_coroutine_threadsafe.call_args.args[0] + scheduled_coro.close() + + def test_mqtt_on_disconnect_does_not_schedule_duplicate_refresh( + self, base_client: MockBaseDeyeMqttClient + ) -> None: + """Test repeated disconnect callbacks share an in-flight refresh.""" + refresh_future = MagicMock() + refresh_future.done.return_value = False + base_client._mqtt_info_refresh_future = refresh_future + + with patch("asyncio.run_coroutine_threadsafe") as mock_run_coroutine_threadsafe: + base_client._mqtt_on_disconnect(base_client._mqtt, None, 1) + + mock_run_coroutine_threadsafe.assert_not_called() + + @pytest.mark.asyncio + async def test_mqtt_info_refresh_failure_is_caught( + self, base_client: MockBaseDeyeMqttClient, caplog: pytest.LogCaptureFixture + ) -> None: + """Test a cloud failure cannot escape into Paho's network thread.""" + with patch.object( + base_client, + "_set_mqtt_info", + AsyncMock(side_effect=RuntimeError("cloud unavailable")), + ): + await base_client._refresh_mqtt_info_after_disconnect(1) + + assert "Failed to refresh Deye MQTT information" in caplog.text + + @pytest.mark.asyncio + async def test_mqtt_info_refresh_timeout_is_caught( + self, base_client: MockBaseDeyeMqttClient, caplog: pytest.LogCaptureFixture + ) -> None: + """Test a slow cloud refresh times out without raising.""" + with patch( + "libdeye.mqtt_client.asyncio.wait_for", + AsyncMock(side_effect=TimeoutError), + ) as mock_wait_for: + await base_client._refresh_mqtt_info_after_disconnect(1) + + assert mock_wait_for.call_args.kwargs["timeout"] == MQTT_INFO_REFRESH_TIMEOUT + assert "Timed out after 15s" in caplog.text + + refresh_coro = mock_wait_for.call_args.args[0] + refresh_coro.close() def test_mqtt_on_message(self, base_client: MockBaseDeyeMqttClient) -> None: """Test _mqtt_on_message method."""