fix: MQTT reconnection and reliability hardening

CHANGELOG.rst

@@ -7,6 +7,44 @@ Unreleased
 
 Bug Fixes
 ---------
+- **Fix reconnection loop dying on authentication errors**: the backoff
+  loop caught only ``AwsCrtError`` and ``RuntimeError``, so
+  ``TokenRefreshError``, ``AuthenticationError``, and
+  ``MqttCredentialsError`` raised during quick/deep reconnection escaped
+  and silently killed the reconnect task. A routine outage coinciding
+  with token expiry left the client permanently offline despite unlimited
+  retries. All library errors (``Nwp500Error``) and operation timeouts
+  are now treated as failed attempts and retried; only
+  ``InvalidCredentialsError`` is fatal and stops the loop with a
+  ``reconnection_failed`` event.
+- **Fix disconnect() being a no-op while the connection is interrupted**:
+  calling ``disconnect()`` during an interruption returned early without
+  disabling automatic reconnection or stopping periodic tasks, so the
+  backoff loop would resurrect the connection after the application shut
+  the client down. ``disconnect()`` now always disables reconnection and
+  stops periodic tasks, and tears down the SDK connection even when not
+  connected.
+- **Fix queued commands being lost after active/deep reconnection**: the
+  command queue was only flushed from the SDK's ``on_connection_resumed``
+  callback, which never fires for the new connection built by
+  active/deep reconnection. Commands queued while offline were silently
+  dropped. Both reconnect paths now flush the queue after subscriptions
+  are restored.
+- **Fix periodic request tasks dying on MQTT errors**: the periodic loop
+  caught only ``AwsCrtError`` and ``RuntimeError``;
+  ``MqttNotConnectedError``/``MqttPublishError`` raised by a publish
+  racing a disconnection permanently killed the polling task while it
+  still appeared active. The loop now survives all library errors.
+- **Fix silent failures in thread-scheduled coroutines**: futures
+  returned by ``run_coroutine_threadsafe`` were discarded, so exceptions
+  from scheduled work (e.g. a failed resubscribe after a clean-session
+  resume, leaving the client connected but deaf) vanished. A done
+  callback now logs them.
+- **Fix CancelledError being swallowed in reconnection and periodic
+  loops**: both loops caught ``asyncio.CancelledError`` and ``break``-ed,
+  so cancelled tasks ended "successfully" (and the reconnection loop
+  could emit ``reconnection_failed`` during a manual disconnect).
+  Cancellation now propagates correctly.
 - **Fix AttributeError in configure_reservation_water_program**: The
   ``NavienMqttClient`` proxy referenced ``self._control``, which is never
   assigned (the attribute is ``_device_controller``), so every call raised
@@ -33,6 +71,18 @@ Bug Fixes
   imported ``MqttConnectionConfig`` from the nonexistent ``nwp500.mqtt_utils``
   module and used the deprecated ``datetime.utcnow()``.
 
+Improvements
+------------
+- **MQTT operation acknowledgement timeouts**: connect, publish,
+  subscribe, unsubscribe, and disconnect acknowledgements are now awaited
+  with a timeout (``MqttConnectionConfig.operation_timeout``, default 30
+  seconds). Previously a half-open TCP connection could hang callers
+  until the 20-minute keep-alive expired.
+- **Reconnection backoff jitter**: reconnect delays are now randomized
+  (±50%, capped at ``max_reconnect_delay``) so fleets of clients
+  disconnected simultaneously (e.g. the AWS IoT 24-hour disconnect) no
+  longer reconnect in synchronized waves.
+
 Security
 --------
 - **Restrict token cache file permissions**: ``~/.nwp500_tokens.json``

src/nwp500/mqtt/client.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import asyncio
+import concurrent.futures
 import json
 import logging
 import uuid
@@ -69,6 +70,31 @@
 _logger = logging.getLogger(__name__)
 
 
+def _log_scheduled_coroutine_result(
+    future: concurrent.futures.Future[Any],
+) -> None:
+    """Surface exceptions from scheduled coroutines.
+
+    Attached to futures returned by ``run_coroutine_threadsafe``.
+    Without this callback the returned future is discarded and any
+    exception (e.g. resubscribe failure after a clean-session resume) is
+    silently swallowed.
+    """
+    if future.cancelled():
+        return
+    exc = future.exception()
+    if exc is not None:
+        # Pass an explicit (type, value, traceback) tuple: a bare
+        # exception instance is treated by logging as a truthy flag,
+        # which would log the *current* exception context (empty in a
+        # done-callback) instead of the coroutine's traceback.
+        _logger.error(
+            "Scheduled coroutine failed: %s",
+            exc,
+            exc_info=(type(exc), exc, exc.__traceback__),
+        )
+
+
 class NavienMqttClient(EventEmitter):
     """
     Async MQTT client for Navien device communication over AWS IoT.
@@ -252,14 +278,24 @@ def _schedule_coroutine(self, coro: Any) -> None:
                 self._loop = asyncio.get_running_loop()
             except RuntimeError:
                 _logger.warning("No event loop available to schedule coroutine")
+                # Close the never-scheduled coroutine to avoid a
+                # "coroutine was never awaited" warning and release any
+                # resources it holds.
+                coro.close()
                 return
 
         # Schedule the coroutine in the stored loop using thread-safe method
         try:
-            asyncio.run_coroutine_threadsafe(coro, self._loop)
+            future = asyncio.run_coroutine_threadsafe(coro, self._loop)
         except RuntimeError as e:
             # Event loop is closed or not running
             _logger.error(f"Failed to schedule coroutine: {e}", exc_info=True)
+            coro.close()
+        else:
+            # Never drop the future: exceptions from scheduled coroutines
+            # (e.g. a failed resubscribe after a clean-session resume) would
+            # otherwise be stored on the discarded future and vanish.
+            future.add_done_callback(_log_scheduled_coroutine_result)
 
     def _on_connection_interrupted_internal(
         self, connection: mqtt.Connection, error: AwsCrtError, **kwargs: Any
@@ -476,6 +512,12 @@ async def _active_reconnect(self) -> None:
                         )
                         await self._subscription_manager.resubscribe_all()
 
+                    # A brand-new connection never fires the SDK's
+                    # on_connection_resumed callback, so queued commands
+                    # must be flushed here or they would be lost.
+                    if self.config.enable_command_queue and self._command_queue:
+                        await self._send_queued_commands_internal()
+
                     _logger.info("Active reconnection successful")
                 else:
                     _logger.warning("Active reconnection failed")
@@ -587,6 +629,12 @@ async def _deep_reconnect(self) -> None:
                     )
                     await self._subscription_manager.resubscribe_all()
 
+                # A brand-new connection never fires the SDK's
+                # on_connection_resumed callback, so queued commands must
+                # be flushed here or they would be lost.
+                if self.config.enable_command_queue and self._command_queue:
+                    await self._send_queued_commands_internal()
+
                 _logger.info(
                     "Deep reconnection successful - fully rebuilt connection"
                 )
@@ -673,6 +721,7 @@ async def connect(self) -> bool:
                     event_emitter=self,
                     schedule_coroutine=self._schedule_coroutine,
                     device_info_cache=device_info_cache,
+                    operation_timeout=self.config.operation_timeout,
                 )
 
                 # Update device controller cache
@@ -793,32 +842,46 @@ def _create_credentials_provider(self) -> Any:
         )
 
     async def disconnect(self) -> None:
-        """Disconnect from AWS IoT Core and stop all periodic tasks."""
-        if not self._connected or not self._connection_manager:
-            _logger.warning("Not connected")
-            return
-
-        _logger.info("Disconnecting from AWS IoT...")
+        """Disconnect from AWS IoT Core and stop all periodic tasks.
 
-        # Disable automatic reconnection
+        Safe to call in any state. Even when the connection is currently
+        interrupted (``_connected`` is False with a reconnection loop
+        running), this method disables automatic reconnection and stops
+        periodic tasks so a backoff loop cannot resurrect the connection
+        after the application has shut the client down.
+        """
+        # Always disable automatic reconnection first, regardless of the
+        # current connection state.
         if self._reconnection_handler:
             self._reconnection_handler.disable()
             await self._reconnection_handler.cancel()
 
-        # Stop all periodic tasks first
+        # Stop all periodic tasks
         if self._periodic_manager:
             await self._periodic_manager.stop_all_periodic_tasks()
 
+        if not self._connection_manager:
+            _logger.warning("Not connected")
+            return
+
+        _logger.info("Disconnecting from AWS IoT...")
+
         try:
-            # Delegate disconnection to connection manager
-            await self._connection_manager.disconnect()
+            if self._connected:
+                # Delegate graceful disconnection to connection manager
+                await self._connection_manager.disconnect()
+            else:
+                # Interrupted state: no broker session to end gracefully,
+                # but the SDK connection object must still be torn down so
+                # its auto-reconnect cannot fire after shutdown.
+                await self._connection_manager.close()
 
             # Clear connection state
             self._connected = False
             self._connection = None
 
             _logger.info("Disconnected successfully")
-        except (AwsCrtError, RuntimeError) as e:
+        except (AwsCrtError, RuntimeError, TimeoutError) as e:
             _logger.error(f"Error during disconnect: {e}")
             raise