fix: MQTT reconnection and reliability hardening

- Widen reconnection backoff loop exception handling: auth and MQTT
  errors (Nwp500Error) now count as failed attempts and are retried
  instead of silently killing the reconnect task; only
  InvalidCredentialsError is fatal and emits reconnection_failed
- Make disconnect() effective while the connection is interrupted:
  always disable automatic reconnection and stop periodic tasks, and
  close the SDK connection even when not connected
- Flush the command queue after active/deep reconnection; the SDK's
  on_connection_resumed callback never fires for a rebuilt connection,
  so queued commands were silently lost
- Log exceptions from thread-scheduled coroutines instead of dropping
  the run_coroutine_threadsafe future
- Widen periodic request loop exception handling so MQTT publish
  failures no longer permanently kill polling tasks
- Re-raise CancelledError in reconnection and periodic loops so
  cancelled tasks are actually marked cancelled
- Add operation_timeout (default 30s) for MQTT connect/publish/
  subscribe/unsubscribe/disconnect acknowledgements to prevent hangs
  on half-open TCP connections
- Add +/-50% jitter to reconnect backoff to avoid synchronized
  fleet-wide reconnect storms

Author: emmanuel

CHANGELOG.rst

@@ -5,6 +5,59 @@ Changelog
 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.
+
+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.
+
 Version 8.1.3 (2026-06-15)
 ==========================
 

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,23 @@
 _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:
+        _logger.error("Scheduled coroutine failed: %s", exc, exc_info=exc)
+
+
 class NavienMqttClient(EventEmitter):
     """
     Async MQTT client for Navien device communication over AWS IoT.
@@ -256,10 +274,15 @@ def _schedule_coroutine(self, coro: Any) -> None:
 
         # 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)
+        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 +499,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 +616,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 +708,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 +829,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
 

src/nwp500/mqtt/connection.py

@@ -96,6 +96,49 @@ def __init__(
             f"Initialized connection manager with client ID: {config.client_id}"
         )
 
+    async def _await_ack(
+        self, future: asyncio.Future[Any], operation: str
+    ) -> Any:
+        """Await a broker acknowledgement future with a timeout.
+
+        The awscrt future is shielded so cancellation (or a timeout) never
+        propagates into the SDK future — the underlying operation completes
+        independently, preventing InvalidStateError in AWS CRT callbacks.
+
+        A timeout guards against half-open TCP connections where the
+        acknowledgement never arrives; without it callers can hang until
+        the keep-alive expires (20+ minutes).
+
+        Args:
+            future: concurrent.futures.Future returned by the SDK
+            operation: Human-readable operation name for error messages
+
+        Returns:
+            The acknowledgement result
+
+        Raises:
+            TimeoutError: If no acknowledgement within
+                config.operation_timeout seconds
+            asyncio.CancelledError: If the awaiting task is cancelled
+        """
+        try:
+            return await asyncio.wait_for(
+                asyncio.shield(asyncio.wrap_future(future)),
+                timeout=self.config.operation_timeout,
+            )
+        except asyncio.CancelledError:
+            _logger.debug(
+                "%s was cancelled but will complete in background", operation
+            )
+            raise
+        except TimeoutError:
+            _logger.error(
+                "%s not acknowledged within %.1fs",
+                operation,
+                self.config.operation_timeout,
+            )
+            raise
+
     async def connect(self) -> bool:
         """
         Establish connection to AWS IoT Core.
@@ -151,19 +194,7 @@ async def connect(self) -> bool:
             connect_future = cast(
                 asyncio.Future[Any], self._connection.connect()
             )
-            try:
-                connect_result = await asyncio.shield(
-                    asyncio.wrap_future(connect_future)
-                )
-            except asyncio.CancelledError:
-                # Shield was cancelled - the underlying connect will
-                # complete independently, preventing InvalidStateError
-                # in AWS CRT callbacks
-                _logger.debug(
-                    "Connect operation was cancelled but will complete "
-                    "in background"
-                )
-                raise
+            connect_result = await self._await_ack(connect_future, "Connect")
 
             self._connected = True
             _logger.info(
@@ -226,17 +257,7 @@ async def disconnect(self) -> None:
             disconnect_future = cast(
                 asyncio.Future[Any], self._connection.disconnect()
             )
-            try:
-                await asyncio.shield(asyncio.wrap_future(disconnect_future))
-            except asyncio.CancelledError:
-                # Shield was cancelled - the underlying disconnect will
-                # complete independently, preventing InvalidStateError
-                # in AWS CRT callbacks
-                _logger.debug(
-                    "Disconnect operation was cancelled but will complete "
-                    "in background"
-                )
-                raise
+            await self._await_ack(disconnect_future, "Disconnect")
 
             self._connected = False
             self._connection = None
@@ -271,9 +292,9 @@ async def close(self) -> None:
             disconnect_future = cast(
                 asyncio.Future[Any], connection.disconnect()
             )
-            await asyncio.shield(asyncio.wrap_future(disconnect_future))
+            await self._await_ack(disconnect_future, "Close")
             _logger.debug("SDK connection closed")
-        except (AwsCrtError, RuntimeError) as e:
+        except (AwsCrtError, RuntimeError, TimeoutError) as e:
             # Expected when connection is already dead or in bad state
             _logger.debug(f"SDK connection close (benign): {e}")
         except asyncio.CancelledError:
@@ -317,17 +338,7 @@ async def subscribe(
         subscribe_future = cast(asyncio.Future[Any], subscribe_future_raw)
         packet_id = packet_id_raw
 
-        try:
-            await asyncio.shield(asyncio.wrap_future(subscribe_future))
-        except asyncio.CancelledError:
-            # Shield was cancelled - the underlying subscribe will
-            # complete independently, preventing InvalidStateError
-            # in AWS CRT callbacks
-            _logger.debug(
-                f"Subscribe to '{topic}' was cancelled but will complete "
-                "in background"
-            )
-            raise
+        await self._await_ack(subscribe_future, f"Subscribe to '{topic}'")
 
         _logger.info(f"Subscribed to '{topic}' with packet_id {packet_id}")
         return (subscribe_future, packet_id)
@@ -359,17 +370,7 @@ async def unsubscribe(self, topic: str) -> int:
         unsubscribe_future = cast(asyncio.Future[Any], unsubscribe_future_raw)
         packet_id = int(packet_id_raw)
 
-        try:
-            await asyncio.shield(asyncio.wrap_future(unsubscribe_future))
-        except asyncio.CancelledError:
-            # Shield was cancelled - the underlying unsubscribe will
-            # complete independently, preventing InvalidStateError
-            # in AWS CRT callbacks
-            _logger.debug(
-                f"Unsubscribe from '{topic}' was cancelled but will "
-                "complete in background"
-            )
-            raise
+        await self._await_ack(unsubscribe_future, f"Unsubscribe from '{topic}'")
 
         _logger.info(f"Unsubscribed from '{topic}' with packet_id {packet_id}")
         return packet_id
@@ -419,16 +420,7 @@ async def publish(
         # InvalidStateError when AWS CRT tries to set exception on a
         # cancelled future.
         try:
-            await asyncio.shield(asyncio.wrap_future(publish_future))
-        except asyncio.CancelledError:
-            # Shield was cancelled - the underlying publish will complete
-            # independently, preventing InvalidStateError in AWS CRT
-            # callbacks
-            _logger.debug(
-                f"Publish to '{topic}' was cancelled but will complete "
-                "in background"
-            )
-            raise
+            await self._await_ack(publish_future, f"Publish to '{topic}'")
         except AwsCrtError as e:
             # Handle connection destruction during publish
             # This can happen when AWS IoT Core disconnects (e.g., 24-hour

src/nwp500/mqtt/periodic.py

@@ -19,6 +19,7 @@
 
 from awscrt.exceptions import AwsCrtError
 
+from ..exceptions import Nwp500Error
 from ..models import Device
 from .utils import PeriodicRequestType
 
@@ -184,13 +185,24 @@ async def periodic_request() -> None:
                     await asyncio.sleep(period_seconds)
 
                 except asyncio.CancelledError:
+                    # Re-raise so the task is actually marked as cancelled
+                    # instead of ending "successfully".
                     _logger.info(
                         f"Periodic {request_type.value} requests cancelled "
                         f"for {redacted_device_id}"
                     )
-                    break
-                except (AwsCrtError, RuntimeError) as e:
-                    # Handle known MQTT errors gracefully
+                    raise
+                except (
+                    AwsCrtError,
+                    Nwp500Error,
+                    RuntimeError,
+                    TimeoutError,
+                ) as e:
+                    # Handle known MQTT/library errors gracefully. This must
+                    # include Nwp500Error: publish failures raise
+                    # MqttNotConnectedError/MqttPublishError (not
+                    # RuntimeError), which would otherwise silently kill
+                    # the periodic task while it still looks alive.
                     error_name = (
                         getattr(e, "name", None)
                         if isinstance(e, AwsCrtError)