Fix boot race: retry on any startup error and clean up paho threads

When HA and Victron boot simultaneously (power outage, no UPS), the
integration can fail permanently because:

1. Hub.start() only catches CannotConnectError, but connect() can also
   raise NotConnectedError (if connection drops mid-setup) or other
   exceptions. Uncaught exceptions cause HA to treat the failure as
   permanent (no retry). Fix: catch Exception (except AuthenticationError)
   and convert to ConfigEntryNotReady so HA retries with backoff.

2. connect() calls loop_start() early, creating a paho background thread.
   If any later step fails, loop_stop() is never called, leaking the
   thread. Each HA retry leaks another thread. Fix: try/except around
   post-loop_start() code that calls loop_stop() on failure.

3. disconnect() calls client.disconnect() but not loop_stop(), so the
   background thread is never explicitly stopped. Fix: add loop_stop().

Fixes #461

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

@@ -211,7 +211,6 @@ def __init__(
         self._fallback_placeholders: dict[str, FallbackPlaceholder] = {}
         self._all_metrics: dict[str, Metric] = {}
         self._first_connect = True
-        self._subscribed = False
         self._first_full_publish = True
         self._notified_device_ids: set[str] = set()
         self._connect_failed_attempts = 0
@@ -368,14 +367,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] = []
@@ -415,24 +421,12 @@ def _on_connect(
         reason_code: ReasonCode,
         properties: Properties | None = None,
     ) -> None:
-        # Capture before _on_connect_internal changes it so we can distinguish
-        # first connect (where we must surface errors) from auto-reconnect
-        # (where we must NOT call disconnect, or paho stops retrying).
-        is_initial_connect = self._first_connect
         try:
             self._on_connect_internal(client, userdata, flags, reason_code, properties)
         except Exception as exc:
             _LOGGER.exception("_on_connect exception %s: %s", type(exc), exc)
-            if is_initial_connect:
-                # First connect: propagate the error so the caller can handle it
-                self._connect_failed_reason = exc
-                client.disconnect()
-            else:
-                # Reconnect: do NOT disconnect — let paho keep auto-reconnecting.
-                # Subscriptions will be retried from the keepalive loop.
-                _LOGGER.warning(
-                    "Subscription setup failed during reconnect, will retry from keepalive loop"
-                )
+            self._connect_failed_reason = exc
+            client.disconnect()
 
         try:
             self._schedule_threadsafe(self._connected_event.set)
@@ -464,14 +458,7 @@ def _on_connect_internal(
                 f"Failed to connect to MQTT broker: {self.host}:{self.port}. Error: {connack_string(reason_code)}"
             )
 
-        is_reconnect = not self._first_connect
-        if is_reconnect:
-            _LOGGER.info(
-                "Reconnected to MQTT broker (session_present=%s). Re-establishing subscriptions...",
-                flags.session_present,
-            )
-        else:
-            _LOGGER.info("Connected to MQTT broker successfully")
+        _LOGGER.info("Connected to MQTT broker successfully")
         self._setup_subscriptions()
 
     def _on_disconnect(
@@ -483,7 +470,6 @@ def _on_disconnect(
         _properties: Properties | None = None,
     ) -> None:
         """Handle disconnection callback."""
-        self._subscribed = False
         if reason_code != 0:
             _LOGGER.warning(
                 "Unexpected disconnection from MQTT broker. Error: %s. flags: %s, Reconnecting...",
@@ -839,6 +825,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)
@@ -873,15 +860,6 @@ async def _keepalive_loop(self) -> None:
         count = 0
         while True:
             try:
-                # If subscriptions were lost (e.g. failed during reconnect), retry them
-                if not self._subscribed and self._client is not None and self._client.is_connected():
-                    _LOGGER.info("Subscriptions not established, retrying subscription setup...")
-                    try:
-                        self._setup_subscriptions()
-                        _LOGGER.info("Subscription retry succeeded")
-                    except Exception as sub_exc:
-                        _LOGGER.warning("Subscription retry failed, will try again: %s", sub_exc)
-
                 self._keepalive()
                 await asyncio.sleep(30)
                 # We should keep alive all metrics every 60 seconds
@@ -1070,7 +1048,6 @@ def _setup_subscriptions(self) -> None:
         assert self.installation_id is not None
         self._subscribe(f"N/{self.installation_id}/full_publish_completed")
         _LOGGER.info("Subscribed to full_publish_completed notification")
-        self._subscribed = True
         self._keepalive(True)
 
     async def _wait_for_connect(self) -> None:

custom_components/victron_mqtt/hub.py

@@ -118,7 +118,7 @@ async def start(self) -> None:
             raise ConfigEntryAuthFailed(
                 f"Authentication failed for {self.host}: {auth_error}"
             ) from auth_error
-        except CannotConnectError as connect_error:
+        except Exception as connect_error:
             raise ConfigEntryNotReady(
                 f"Cannot connect to the hub: {connect_error}"
             ) from connect_error