fix: keep MQTT network loop alive on disconnect

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:

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