Fix boot race condition: catch NotConnectedError and stop leaked paho threads

When HA and Victron boot simultaneously after a power outage, several
failure paths can occur:

1. NotConnectedError escapes Hub.start() uncaught — HA treats this as a
   permanent failure (no retry), so the integration never recovers.
   Fix: catch NotConnectedError in Hub.start() and raise ConfigEntryNotReady
   so HA retries with exponential backoff.

2. Paho MQTT thread leak — connect() calls loop_start() early, but if any
   subsequent step fails (timeout, connection drop, etc.), loop_stop() is
   never called. Each HA retry creates a new Hub, leaking the old thread.
   Fix: wrap post-loop_start() code in try/except that calls loop_stop()
   on failure.

3. disconnect() doesn't call loop_stop() — the background thread started
   by loop_start() is never explicitly stopped.
   Fix: add loop_stop() call in disconnect().

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 6374e2f1-1fb3-453d-a1c2-dddf69a62f61

Author: tomer-w

custom_components/victron_mqtt/_vendor/victron_mqtt/hub.py

@@ -368,14 +368,21 @@ async def connect(self) -> None:
 
         _LOGGER.info("Starting paho mqtt")
         self._client.loop_start()
-        _LOGGER.info("Connecting")
-        self._client.connect_async(self.host, self.port)
-        _LOGGER.info("Waiting for connection event")
-        await self._wait_for_connect()
-        if self._connect_failed_reason is not None:
-            reraise_same_exception(self._connect_failed_reason)
-        _LOGGER.info("Successfully connected to MQTT broker at %s:%d", self.host, self.port)
-        await self._wait_for_installation_id(expected_id=self._expected_installation_id)
+        try:
+            _LOGGER.info("Connecting")
+            self._client.connect_async(self.host, self.port)
+            _LOGGER.info("Waiting for connection event")
+            await self._wait_for_connect()
+            if self._connect_failed_reason is not None:
+                reraise_same_exception(self._connect_failed_reason)
+            _LOGGER.info("Successfully connected to MQTT broker at %s:%d", self.host, self.port)
+            await self._wait_for_installation_id(expected_id=self._expected_installation_id)
+        except Exception:
+            # If anything fails after loop_start(), stop the paho thread to avoid leaking it.
+            # On HA retry a new Hub and paho client will be created.
+            _LOGGER.info("Connection setup failed, stopping paho mqtt loop")
+            self._client.loop_stop()
+            raise
         assert self._installation_id is not None
         # First we need to replace the installation ID in the subscription topics
         new_list: list[str] = []
@@ -839,6 +846,7 @@ async def disconnect(self) -> None:
         self._stop_keepalive_loop()
         await asyncio.sleep(0.1)
         self._client.disconnect()  # need to call disconnect so the paho thread will terminate
+        self._client.loop_stop()   # stop the background thread started by loop_start()
         _LOGGER.info("Disconnected from MQTT broker")
         # Give a small delay to allow any pending MQTT messages to be processed
         await asyncio.sleep(0.1)

custom_components/victron_mqtt/hub.py

@@ -11,6 +11,7 @@
     Hub as VictronVenusHub,
     Metric as VictronVenusMetric,
     MetricKind,
+    NotConnectedError,
     OperationMode,
 )
 
@@ -122,6 +123,10 @@ async def start(self) -> None:
             raise ConfigEntryNotReady(
                 f"Cannot connect to the hub: {connect_error}"
             ) from connect_error
+        except NotConnectedError as connect_error:
+            raise ConfigEntryNotReady(
+                f"Connection lost during setup (possible boot race): {connect_error}"
+            ) from connect_error
 
     async def stop(self) -> None:
         """Stop the Victron MQTT hub."""