fix(realtime): properly dispose of stale MQTT clients before reconnect

.github/workflows/validate-pr.yml

@@ -37,6 +37,27 @@ jobs:
             didn't match the configured pattern. Please ensure that the subject
             doesn't start with an uppercase character.
 
+  validate-tests:
+    runs-on: ubuntu-latest
+    name: Run tests
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+
+      - name: Setup Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: '3.11'
+
+      - name: Install test dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install pytest
+
+      - name: Run unit tests
+        run: python -m pytest -q tests
+
   validate-hacs:
     runs-on: ubuntu-latest
     name: Validate HACS

custom_components/minibrew/realtime.py

@@ -121,8 +121,26 @@ async def async_start(self):
         self._stopped = False
         await self._async_do_connect()
 
+    def _dispose_mqtt_client(self, mqtt=None):
+        """Disconnect an MQTT client and drop our reference to it.
+
+        This avoids leaking stale paho sockets and threads when the broker is
+        unavailable or a connection attempt fails and we immediately retry.
+        """
+        mqtt = self._mqtt if mqtt is None else mqtt
+        if mqtt is None:
+            return
+        try:
+            mqtt.disconnect()
+        except Exception as err:  # noqa: BLE001
+            _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err)
+        if self._mqtt is mqtt:
+            self._mqtt = None
+
     async def _async_do_connect(self):
         """Internal: create the MQTT client and connect. Called from async_start and reconnect."""
+        if self._mqtt is not None:
+            self._dispose_mqtt_client()
         try:
             self._mqtt = await self.hass.async_add_executor_job(self._client.create_mqtt_client)
         except Exception as err:  # noqa: BLE001 - never break REST polling
@@ -142,6 +160,7 @@ async def _async_do_connect(self):
             _LOGGER.info("MiniBrew realtime: MQTT connect() returned (waiting for on_connected callback)")
         except Exception as err:  # noqa: BLE001 - never break REST polling
             _LOGGER.warning("MiniBrew realtime: could not connect to MQTT broker: %s", err)
+            self._dispose_mqtt_client(self._mqtt)
             self._schedule_reconnect()
 
     def async_ensure_subscribed(self, serials):
@@ -184,13 +203,13 @@ async def async_stop(self):
         if self._reconnect_task is not None:
             self._reconnect_task.cancel()
             self._reconnect_task = None
-        if self._mqtt is None:
-            return
-        mqtt, self._mqtt = self._mqtt, None
-        try:
-            await self.hass.async_add_executor_job(mqtt.disconnect)
-        except Exception as err:  # noqa: BLE001
-            _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err)
+        if self._mqtt is not None:
+            mqtt = self._mqtt
+            self._mqtt = None
+            try:
+                await self.hass.async_add_executor_job(mqtt.disconnect)
+            except Exception as err:  # noqa: BLE001
+                _LOGGER.debug("MiniBrew realtime: error during disconnect: %s", err)
         self._connected = False
 
     def _schedule_reconnect(self):
@@ -266,6 +285,7 @@ def _handle_disconnected(self):
         """Mark disconnected, refresh entity availability, and schedule reconnect (paho thread)."""
         self._connected = False
         _LOGGER.warning("MiniBrew realtime: MQTT disconnected — will retry in %s s", _RECONNECT_DELAY)
+        self._dispose_mqtt_client()
         self.hass.loop.call_soon_threadsafe(self.coordinator.async_update_listeners)
         self._schedule_reconnect()
 

tests/test_realtime_overlay.py

@@ -184,6 +184,63 @@ def call_soon_threadsafe(self, callback):
 
     assert len(calls) == 1
 
+
+def test_realtime_manager_disconnects_stale_client_before_reconnect():
+    import asyncio
+
+    class _FakeMqtt:
+        def __init__(self, name):
+            self.name = name
+            self.disconnected = 0
+            self.connected = 0
+
+        def on_device_log(self, *args, **kwargs):
+            pass
+
+        def on_connected(self, *args, **kwargs):
+            pass
+
+        def on_disconnected(self, *args, **kwargs):
+            pass
+
+        def on_error(self, *args, **kwargs):
+            pass
+
+        def connect(self):
+            self.connected += 1
+
+        def disconnect(self):
+            self.disconnected += 1
+
+    class _FakeLoop:
+        def call_soon_threadsafe(self, callback):
+            callback()
+
+    class _FakeHass:
+        def __init__(self):
+            self.loop = _FakeLoop()
+
+        async def async_add_executor_job(self, func, *args):
+            return func(*args)
+
+    class _FakeCoordinator:
+        def async_update_listeners(self):
+            return None
+
+    existing = _FakeMqtt("old")
+    new_client = _FakeMqtt("new")
+    hass = _FakeHass()
+    coordinator = _FakeCoordinator()
+    client = SimpleNamespace(create_mqtt_client=lambda: new_client)
+    manager = realtime.MiniBrewRealtimeManager(hass, coordinator, client)
+    manager._mqtt = existing
+
+    asyncio.run(manager._async_do_connect())
+
+    assert existing.disconnected == 1
+    assert manager._mqtt is new_client
+    assert new_client.connected == 1
+
 if __name__ == "__main__":
     failures = 0
     for name, fn in sorted(globals().items()):