diff --git a/CHANGELOG.rst b/CHANGELOG.rst index f45e57ef..2500cda6 100644 --- a/CHANGELOG.rst +++ b/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`` diff --git a/src/nwp500/mqtt/client.py b/src/nwp500/mqtt/client.py index 72bc1bcf..38a63053 100644 --- a/src/nwp500/mqtt/client.py +++ b/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 diff --git a/src/nwp500/mqtt/connection.py b/src/nwp500/mqtt/connection.py index f48ecc8d..61d2af5d 100644 --- a/src/nwp500/mqtt/connection.py +++ b/src/nwp500/mqtt/connection.py @@ -9,10 +9,11 @@ from __future__ import annotations import asyncio +import concurrent.futures import json import logging from collections.abc import Callable -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any from awscrt import mqtt from awscrt.exceptions import AwsCrtError @@ -96,6 +97,59 @@ def __init__( f"Initialized connection manager with client ID: {config.client_id}" ) + async def _await_ack( + self, + future: concurrent.futures.Future[Any] | 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: Acknowledgement future — the + ``concurrent.futures.Future`` returned by the AWS CRT SDK + (wrapped via ``asyncio.wrap_future``), or an already + loop-bound ``asyncio.Future`` + 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 + """ + awaitable_future: asyncio.Future[Any] = ( + asyncio.wrap_future(future) + if isinstance(future, concurrent.futures.Future) + else future + ) + try: + return await asyncio.wait_for( + asyncio.shield(awaitable_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. @@ -148,22 +202,8 @@ async def connect(self) -> bool: # underlying future if not self._connection: raise RuntimeError("Connection not initialized") - 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_future = self._connection.connect() + connect_result = await self._await_ack(connect_future, "Connect") self._connected = True _logger.info( @@ -223,20 +263,8 @@ async def disconnect(self) -> None: # Convert concurrent.futures.Future to asyncio.Future and await # Use shield to prevent cancellation from propagating to # underlying future - 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 + disconnect_future = self._connection.disconnect() + await self._await_ack(disconnect_future, "Disconnect") self._connected = False self._connection = None @@ -268,12 +296,10 @@ async def close(self) -> None: _logger.debug("Closing underlying SDK connection...") try: - disconnect_future = cast( - asyncio.Future[Any], connection.disconnect() - ) - await asyncio.shield(asyncio.wrap_future(disconnect_future)) + disconnect_future = connection.disconnect() + 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: @@ -314,20 +340,10 @@ async def subscribe( subscribe_future_raw, packet_id_raw = self._connection.subscribe( topic=topic, qos=qos, callback=callback ) - subscribe_future = cast(asyncio.Future[Any], subscribe_future_raw) + subscribe_future = 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) @@ -356,20 +372,10 @@ async def unsubscribe(self, topic: str) -> int: unsubscribe_future_raw, packet_id_raw = self._connection.unsubscribe( topic=topic ) - unsubscribe_future = cast(asyncio.Future[Any], unsubscribe_future_raw) + unsubscribe_future = 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 @@ -411,7 +417,7 @@ async def publish( publish_future_raw, packet_id_raw = self._connection.publish( topic=topic, payload=payload_bytes, qos=qos ) - publish_future = cast(asyncio.Future[Any], publish_future_raw) + publish_future = publish_future_raw packet_id = int(packet_id_raw) # Shield the operation to prevent cancellation from propagating to @@ -419,16 +425,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 diff --git a/src/nwp500/mqtt/periodic.py b/src/nwp500/mqtt/periodic.py index 24b6f7dd..7ead5186 100644 --- a/src/nwp500/mqtt/periodic.py +++ b/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) diff --git a/src/nwp500/mqtt/reconnection.py b/src/nwp500/mqtt/reconnection.py index 6f30c6ba..732902ee 100644 --- a/src/nwp500/mqtt/reconnection.py +++ b/src/nwp500/mqtt/reconnection.py @@ -10,11 +10,14 @@ import asyncio import contextlib import logging +import random from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING, Any from awscrt.exceptions import AwsCrtError +from ..exceptions import InvalidCredentialsError, Nwp500Error + if TYPE_CHECKING: from .utils import MqttConnectionConfig @@ -197,8 +200,16 @@ async def _reconnect_with_backoff(self) -> None: Uses a two-tier strategy: - Quick reconnects (attempts 1-N): Fast reconnection with existing setup - Deep reconnects (attempts N+): Full rebuild including token refresh + + All library errors (``Nwp500Error``, including authentication and + MQTT errors raised by the reconnect functions), transient runtime + errors, and operation timeouts count as failed attempts and are + retried. Only ``InvalidCredentialsError`` is fatal: retrying with + rejected credentials can never succeed, so the loop stops and + emits ``reconnection_failed``. """ unlimited_retries = self.config.max_reconnect_attempts < 0 + fatal_error: Exception | None = None while ( not self._is_connected_func() @@ -223,7 +234,10 @@ async def _reconnect_with_backoff(self) -> None: has_deep_reconnect and is_at_threshold and is_threshold_multiple ) - # Calculate delay with exponential backoff + # Calculate delay with exponential backoff, then apply random + # jitter so a fleet of clients disconnected at the same moment + # (e.g. AWS IoT 24-hour disconnect) doesn't reconnect in + # synchronized waves. delay = min( self.config.initial_reconnect_delay * ( @@ -232,6 +246,10 @@ async def _reconnect_with_backoff(self) -> None: ), self.config.max_reconnect_delay, ) + delay = min( + delay * random.uniform(0.5, 1.5), # noqa: S311 + self.config.max_reconnect_delay, + ) if unlimited_retries: reconnect_type = "deep" if use_deep_reconnect else "quick" @@ -272,7 +290,15 @@ async def _reconnect_with_backoff(self) -> None: "Successfully reconnected via deep reconnection" ) break - except (AwsCrtError, RuntimeError, ValueError) as e: + except InvalidCredentialsError: + raise + except ( + AwsCrtError, + Nwp500Error, + RuntimeError, + ValueError, + TimeoutError, + ) as e: _logger.warning( f"Deep reconnection failed: {e}. Will retry..." ) @@ -286,30 +312,63 @@ async def _reconnect_with_backoff(self) -> None: "quick reconnection" ) break - except (AwsCrtError, RuntimeError) as e: + except InvalidCredentialsError: + raise + except ( + AwsCrtError, + Nwp500Error, + RuntimeError, + TimeoutError, + ) as e: _logger.warning( f"Quick reconnection failed: {e}. Will retry..." ) except asyncio.CancelledError: + # Re-raise so the task is actually marked as cancelled; + # swallowing it would let execution continue past the loop + # (e.g. emitting reconnection_failed during a manual + # disconnect) and break cancellation semantics. _logger.info("Reconnection task cancelled") + raise + except InvalidCredentialsError as e: + _logger.error( + "Credentials rejected during reconnection; " + "stopping automatic reconnection: %s", + e, + ) + fatal_error = e break - except (AwsCrtError, RuntimeError) as e: + except ( + AwsCrtError, + Nwp500Error, + RuntimeError, + TimeoutError, + ) as e: _logger.error( f"Error during reconnection attempt: {e}", exc_info=True ) - # Check final state (only if not unlimited retries) - if ( + # Check final state: report failure when retries are exhausted + # (limited mode) or a fatal error stopped the loop. + attempts_exhausted = ( not unlimited_retries and self._reconnect_attempts >= self.config.max_reconnect_attempts - and not self._is_connected_func() - ): - _logger.error( - f"Failed to reconnect after " - f"{self.config.max_reconnect_attempts} attempts. " - "Manual reconnection required." - ) + ) + if ( + fatal_error is not None or attempts_exhausted + ) and not self._is_connected_func(): + if fatal_error is not None: + _logger.error( + "Reconnection stopped due to fatal error. " + "Manual reconnection required." + ) + else: + _logger.error( + f"Failed to reconnect after " + f"{self.config.max_reconnect_attempts} attempts. " + "Manual reconnection required." + ) # Emit reconnection_failed event if event emitter is available if self._emit_event: try: diff --git a/src/nwp500/mqtt/subscriptions.py b/src/nwp500/mqtt/subscriptions.py index 52cf1d82..088df320 100644 --- a/src/nwp500/mqtt/subscriptions.py +++ b/src/nwp500/mqtt/subscriptions.py @@ -65,6 +65,7 @@ def __init__( event_emitter: EventEmitter, schedule_coroutine: Callable[[Any], None], device_info_cache: MqttDeviceInfoCache | None = None, + operation_timeout: float = 30.0, ): """ Initialize subscription manager. @@ -76,12 +77,15 @@ def __init__( schedule_coroutine: Function to schedule async tasks device_info_cache: Optional MqttDeviceInfoCache for caching device features + operation_timeout: Maximum seconds to wait for broker + acknowledgement of subscribe/unsubscribe operations """ self._connection = connection self._client_id = client_id self._event_emitter = event_emitter self._schedule_coroutine = schedule_coroutine self._device_info_cache = device_info_cache + self._operation_timeout = operation_timeout # Track subscriptions and handlers self._subscriptions: dict[str, mqtt.QoS] = {} @@ -191,20 +195,19 @@ async def subscribe( _logger.info(f"Subscribing to topic: {redact_topic(topic)}") try: - # Convert concurrent.futures.Future to asyncio.Future and await - # Use shield to prevent cancellation from propagating to - # underlying future + # Await the broker acknowledgement with a timeout. Shield the + # SDK future so a timeout/cancel never propagates into AWS CRT + # callbacks (avoids InvalidStateError); the underlying + # subscribe completes independently. subscribe_future, packet_id = self._connection.subscribe( topic=topic, qos=qos, callback=self._on_message_received ) try: - subscribe_result = await asyncio.shield( - asyncio.wrap_future(subscribe_future) + subscribe_result = await asyncio.wait_for( + asyncio.shield(asyncio.wrap_future(subscribe_future)), + timeout=self._operation_timeout, ) except asyncio.CancelledError: - # Shield was cancelled - the underlying subscribe will - # complete independently, preventing InvalidStateError - # in AWS CRT callbacks _logger.debug( f"Subscribe to '{redact_topic(topic)}' was cancelled " "but will complete in background" @@ -221,7 +224,7 @@ async def subscribe( return int(packet_id) - except (AwsCrtError, RuntimeError) as e: + except (AwsCrtError, RuntimeError, TimeoutError) as e: # Clean up handler on failure if this was the first one if (h := self._message_handlers.get(topic)) and callback in h: h.remove(callback) @@ -275,16 +278,15 @@ async def unsubscribe( _logger.info("Unsubscribing from topic (redacted)") try: - # Convert concurrent.futures.Future to asyncio.Future and await - # Use shield to prevent cancellation from propagating to - # underlying future + # Await the broker acknowledgement with a timeout (see + # subscribe() for shielding rationale). unsubscribe_future, packet_id = self._connection.unsubscribe(topic) try: - await asyncio.shield(asyncio.wrap_future(unsubscribe_future)) + await asyncio.wait_for( + asyncio.shield(asyncio.wrap_future(unsubscribe_future)), + timeout=self._operation_timeout, + ) except asyncio.CancelledError: - # Shield was cancelled - the underlying unsubscribe will - # complete independently, preventing InvalidStateError - # in AWS CRT callbacks _logger.debug( "Unsubscribe from topic (redacted) was " "cancelled but will complete in background" @@ -299,7 +301,7 @@ async def unsubscribe( return int(packet_id) - except (AwsCrtError, RuntimeError) as e: + except (AwsCrtError, RuntimeError, TimeoutError) as e: _logger.error(f"Failed to unsubscribe from topic (redacted): {e}") raise diff --git a/src/nwp500/mqtt/utils.py b/src/nwp500/mqtt/utils.py index 8e126e33..6f6f76f6 100644 --- a/src/nwp500/mqtt/utils.py +++ b/src/nwp500/mqtt/utils.py @@ -208,6 +208,11 @@ class MqttConnectionConfig: deep_reconnect_threshold: Attempt count to trigger full connection rebuild + operation_timeout: Maximum seconds to wait for a broker + acknowledgement of an MQTT operation (connect, publish, + subscribe, unsubscribe). Prevents callers from hanging + forever on half-open TCP connections. + enable_command_queue: Enable command queueing when disconnected max_queued_commands: Maximum number of queued commands """ @@ -228,6 +233,9 @@ class MqttConnectionConfig: 10 # Switch to full rebuild after N attempts ) + # MQTT operation acknowledgement timeout + operation_timeout: float = 30.0 # seconds + # Command queue settings enable_command_queue: bool = True max_queued_commands: int = 100 diff --git a/tests/test_mqtt_reliability.py b/tests/test_mqtt_reliability.py new file mode 100644 index 00000000..2b410993 --- /dev/null +++ b/tests/test_mqtt_reliability.py @@ -0,0 +1,630 @@ +"""Regression tests for MQTT reliability fixes. + +Covers: +- Reconnection backoff loop surviving auth/library exceptions +- InvalidCredentialsError stopping reconnection with reconnection_failed +- CancelledError propagation (tasks are actually marked cancelled) +- disconnect() during an interruption disabling reconnection/periodic tasks +- Command queue flush after active/deep reconnect +- Periodic loop surviving MqttError +- Scheduled-coroutine exceptions being logged instead of dropped +- MQTT operation acknowledgement timeouts +- Backoff jitter staying within the configured maximum delay +""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import logging +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from awscrt import mqtt + +from nwp500.auth import AuthenticationResponse, AuthTokens, UserInfo +from nwp500.exceptions import ( + InvalidCredentialsError, + MqttNotConnectedError, + TokenRefreshError, +) +from nwp500.mqtt.connection import MqttConnection +from nwp500.mqtt.periodic import MqttPeriodicRequestManager +from nwp500.mqtt.reconnection import MqttReconnectionHandler +from nwp500.mqtt.utils import MqttConnectionConfig, PeriodicRequestType + + +@pytest.fixture +def mock_auth_client(): + """Create a mock auth client with valid tokens.""" + from nwp500.auth import NavienAuthClient + + client = NavienAuthClient("test@example.com", "password") + valid_tokens = AuthTokens( + id_token="test_id", + access_token="test_access", + refresh_token="test_refresh", + authentication_expires_in=3600, + access_key_id="test_key_id", + secret_key="test_secret_key", + session_token="test_session", + authorization_expires_in=3600, + ) + client._auth_response = AuthenticationResponse( + user_info=UserInfo(user_first_name="Test", user_last_name="User"), + tokens=valid_tokens, + ) + return client + + +def _fast_config(**overrides): + defaults = { + "client_id": "test-client", + "initial_reconnect_delay": 0.001, + "max_reconnect_delay": 0.005, + "max_reconnect_attempts": 3, + } + defaults.update(overrides) + return MqttConnectionConfig(**defaults) + + +def _make_handler(config, is_connected, reconnect_func, **kwargs): + return MqttReconnectionHandler( + config=config, + is_connected_func=is_connected, + schedule_coroutine_func=lambda coro: None, + reconnect_func=reconnect_func, + **kwargs, + ) + + +class TestReconnectLoopExceptionHandling: + """The backoff loop must not be killed by library exceptions.""" + + @pytest.mark.asyncio + async def test_auth_error_counts_as_failed_attempt_and_retries(self): + """Regression: TokenRefreshError escaped the loop's except clauses + (it is not a RuntimeError), killing the backoff task silently.""" + attempts = [] + connected = False + + async def failing_reconnect(): + attempts.append(1) + if len(attempts) >= 2: + nonlocal connected + connected = True + return + raise TokenRefreshError("transient network error during refresh") + + handler = _make_handler( + _fast_config(), lambda: connected, failing_reconnect + ) + handler.enable() + + await handler._reconnect_with_backoff() + + # First attempt raised, second succeeded — the loop survived + assert len(attempts) == 2 + assert connected + + @pytest.mark.asyncio + async def test_mqtt_error_counts_as_failed_attempt_and_retries(self): + """MqttError subclasses (Nwp500Error, not RuntimeError) must be + retried too.""" + attempts = [] + connected = False + + async def failing_reconnect(): + attempts.append(1) + if len(attempts) >= 2: + nonlocal connected + connected = True + return + raise MqttNotConnectedError("not connected") + + handler = _make_handler( + _fast_config(), lambda: connected, failing_reconnect + ) + handler.enable() + + await handler._reconnect_with_backoff() + + assert len(attempts) == 2 + + @pytest.mark.asyncio + async def test_invalid_credentials_stops_loop_and_emits_failed(self): + """Rejected credentials can never succeed: stop retrying and emit + reconnection_failed even with unlimited retries.""" + attempts = [] + events = [] + + async def failing_reconnect(): + attempts.append(1) + raise InvalidCredentialsError("bad password") + + async def emit(event, *args): + events.append((event, args)) + + handler = _make_handler( + _fast_config(max_reconnect_attempts=-1), + lambda: False, + failing_reconnect, + emit_event_func=emit, + ) + handler.enable() + + await handler._reconnect_with_backoff() + + assert len(attempts) == 1 # no retry + assert events and events[0][0] == "reconnection_failed" + + @pytest.mark.asyncio + async def test_cancelled_task_is_marked_cancelled(self): + """Regression: CancelledError was swallowed with `break`, so the + task ended 'successfully' and could emit reconnection_failed + during a manual disconnect.""" + started = asyncio.Event() + + async def never_reconnect(): + raise MqttNotConnectedError("still down") + + events = [] + + async def emit(event, *args): + events.append(event) + + config = _fast_config( + initial_reconnect_delay=5.0, + max_reconnect_delay=10.0, + max_reconnect_attempts=2, + ) + handler = _make_handler( + config, lambda: False, never_reconnect, emit_event_func=emit + ) + handler.enable() + + async def run(): + started.set() + await handler._reconnect_with_backoff() + + task = asyncio.create_task(run()) + await started.wait() + await asyncio.sleep(0.01) # let it enter the backoff sleep + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + assert task.cancelled() + assert events == [] # no reconnection_failed on cancellation + + @pytest.mark.asyncio + async def test_backoff_delay_never_exceeds_max(self): + """Jitter must stay within max_reconnect_delay.""" + sleeps = [] + connected = False + + async def failing_reconnect(): + raise MqttNotConnectedError("down") + + config = _fast_config( + initial_reconnect_delay=1.0, + max_reconnect_delay=2.0, + reconnect_backoff_multiplier=10.0, + max_reconnect_attempts=4, + ) + handler = _make_handler(config, lambda: connected, failing_reconnect) + handler.enable() + + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + sleeps.append(delay) + await real_sleep(0) + + with patch( + "nwp500.mqtt.reconnection.asyncio.sleep", side_effect=fake_sleep + ): + await handler._reconnect_with_backoff() + + assert len(sleeps) == 4 + assert all(d <= config.max_reconnect_delay for d in sleeps) + assert all(d > 0 for d in sleeps) + + +class TestDisconnectDuringInterruption: + """disconnect() must fully shut down even when not connected.""" + + @pytest.mark.asyncio + async def test_disconnect_while_interrupted_stops_everything( + self, mock_auth_client + ): + """Regression: disconnect() returned early when _connected was + False, leaving the reconnect loop and periodic tasks running to + resurrect the connection after shutdown.""" + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + + client._reconnection_handler = MagicMock() + client._reconnection_handler.disable = MagicMock() + client._reconnection_handler.cancel = AsyncMock() + client._periodic_manager = MagicMock() + client._periodic_manager.stop_all_periodic_tasks = AsyncMock() + client._connection_manager = MagicMock() + client._connection_manager.disconnect = AsyncMock() + client._connection_manager.close = AsyncMock() + + client._connected = False # interrupted state + + await client.disconnect() + + client._reconnection_handler.disable.assert_called_once() + client._reconnection_handler.cancel.assert_awaited_once() + periodic = client._periodic_manager + periodic.stop_all_periodic_tasks.assert_awaited_once() + # No broker session to end, but the SDK connection is torn down + client._connection_manager.close.assert_awaited_once() + client._connection_manager.disconnect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_disconnect_while_connected_uses_graceful_path( + self, mock_auth_client + ): + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + + client._reconnection_handler = MagicMock() + client._reconnection_handler.disable = MagicMock() + client._reconnection_handler.cancel = AsyncMock() + client._periodic_manager = None + client._connection_manager = MagicMock() + client._connection_manager.disconnect = AsyncMock() + client._connection_manager.close = AsyncMock() + + client._connected = True + + await client.disconnect() + + client._connection_manager.disconnect.assert_awaited_once() + client._connection_manager.close.assert_not_awaited() + assert client._connected is False + + +class TestQueueFlushAfterReconnect: + """Queued commands must be sent after active/deep reconnects.""" + + @pytest.mark.asyncio + async def test_active_reconnect_flushes_command_queue( + self, mock_auth_client + ): + """Regression: a rebuilt connection never fires the SDK's + on_connection_resumed callback, so queued commands were only sent + on SDK auto-resume — never after an active reconnect.""" + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + client._connected = False + client._loop = asyncio.get_running_loop() + + flushed = [] + + async def fake_send_all(publish_func, is_connected): + flushed.append(1) + + client._command_queue = MagicMock() + client._command_queue.send_all = AsyncMock(side_effect=fake_send_all) + client._subscription_manager = MagicMock() + client._subscription_manager.update_connection = MagicMock() + client._subscription_manager.resubscribe_all = AsyncMock() + + old_manager = MagicMock() + old_manager.close = AsyncMock() + client._connection_manager = old_manager + + new_manager = MagicMock() + new_manager.connect = AsyncMock(return_value=True) + new_manager.connection = MagicMock() + + with ( + patch.object( + mock_auth_client, "ensure_valid_token", new=AsyncMock() + ), + patch( + "nwp500.mqtt.client.MqttConnection", + return_value=new_manager, + ), + ): + await client._active_reconnect() + + assert client._connected is True + sub_manager = client._subscription_manager + sub_manager.resubscribe_all.assert_awaited_once() + client._command_queue.send_all.assert_awaited_once() + + @pytest.mark.asyncio + async def test_deep_reconnect_flushes_command_queue(self, mock_auth_client): + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + client._connected = False + client._loop = asyncio.get_running_loop() + + client._command_queue = MagicMock() + client._command_queue.send_all = AsyncMock() + client._subscription_manager = MagicMock() + client._subscription_manager.update_connection = MagicMock() + client._subscription_manager.resubscribe_all = AsyncMock() + + old_manager = MagicMock() + old_manager.close = AsyncMock() + client._connection_manager = old_manager + + new_manager = MagicMock() + new_manager.connect = AsyncMock(return_value=True) + new_manager.connection = MagicMock() + + with ( + patch.object( + mock_auth_client, + "refresh_token", + new=AsyncMock(), + ), + patch( + "nwp500.mqtt.client.MqttConnection", + return_value=new_manager, + ), + ): + await client._deep_reconnect() + + assert client._connected is True + client._command_queue.send_all.assert_awaited_once() + + +class TestPeriodicLoopResilience: + """The periodic request loop must survive library errors.""" + + @pytest.mark.asyncio + async def test_periodic_task_survives_mqtt_error(self): + """Regression: MqttNotConnectedError (Nwp500Error, not + RuntimeError) escaped the loop's except clause and permanently, + silently killed the periodic task.""" + calls = [] + + async def request_status(device): + calls.append(1) + if len(calls) == 1: + raise MqttNotConnectedError("connection dropped mid-loop") + + manager = MqttPeriodicRequestManager( + is_connected_func=lambda: True, + request_device_info_func=AsyncMock(), + request_device_status_func=request_status, + ) + + device = MagicMock() + device.device_info.mac_address = "aa:bb:cc:dd:ee:ff" + + await manager.start_periodic_requests( + device, + period_seconds=0.01, + request_type=PeriodicRequestType.DEVICE_STATUS, + ) + # Give the loop time for several iterations, including the failure + await asyncio.sleep(0.06) + await manager.stop_all_periodic_tasks() + + # The loop continued after the first (failing) request + assert len(calls) >= 2 + + @pytest.mark.asyncio + async def test_periodic_task_cancellation_marks_task_cancelled(self): + manager = MqttPeriodicRequestManager( + is_connected_func=lambda: True, + request_device_info_func=AsyncMock(), + request_device_status_func=AsyncMock(), + ) + + device = MagicMock() + device.device_info.mac_address = "aa:bb:cc:dd:ee:ff" + + await manager.start_periodic_requests( + device, + period_seconds=10.0, + request_type=PeriodicRequestType.DEVICE_STATUS, + ) + await asyncio.sleep(0.01) + + task = next(iter(manager._periodic_tasks.values())) + await manager.stop_all_periodic_tasks() + + assert task.cancelled() + + +class TestScheduledCoroutineExceptions: + """Exceptions from scheduled coroutines must be logged, not dropped.""" + + def test_failed_future_is_logged(self, caplog): + from nwp500.mqtt.client import _log_scheduled_coroutine_result + + future = concurrent.futures.Future() + future.set_exception(MqttNotConnectedError("resubscribe failed")) + + with caplog.at_level(logging.ERROR, logger="nwp500.mqtt.client"): + _log_scheduled_coroutine_result(future) + + assert "Scheduled coroutine failed" in caplog.text + assert "resubscribe failed" in caplog.text + + def test_cancelled_future_is_silent(self, caplog): + from nwp500.mqtt.client import _log_scheduled_coroutine_result + + future = concurrent.futures.Future() + future.cancel() + + with caplog.at_level(logging.ERROR, logger="nwp500.mqtt.client"): + _log_scheduled_coroutine_result(future) + + assert caplog.text == "" + + @pytest.mark.asyncio + async def test_schedule_coroutine_attaches_done_callback( + self, mock_auth_client, caplog + ): + """End-to-end: a scheduled coroutine that raises produces a log.""" + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + client._loop = asyncio.get_running_loop() + + async def boom(): + raise MqttNotConnectedError("scheduled failure") + + with caplog.at_level(logging.ERROR, logger="nwp500.mqtt.client"): + client._schedule_coroutine(boom()) + # Let the scheduled coroutine and its callback run + for _ in range(5): + await asyncio.sleep(0) + + assert "Scheduled coroutine failed" in caplog.text + + +class TestOperationTimeouts: + """MQTT operations must not hang forever on half-open connections.""" + + @pytest.mark.asyncio + async def test_publish_times_out_without_ack(self, mock_auth_client): + config = MqttConnectionConfig( + client_id="test-client", operation_timeout=0.05 + ) + conn = MqttConnection(config, mock_auth_client) + + never_acked = concurrent.futures.Future() # never resolves + sdk_conn = MagicMock() + sdk_conn.publish.return_value = (never_acked, 42) + conn._connection = sdk_conn + conn._connected = True + + with pytest.raises(TimeoutError): + await conn.publish("test/topic", {"key": "value"}) + + @pytest.mark.asyncio + async def test_subscribe_times_out_without_ack(self, mock_auth_client): + config = MqttConnectionConfig( + client_id="test-client", operation_timeout=0.05 + ) + conn = MqttConnection(config, mock_auth_client) + + never_acked = concurrent.futures.Future() + sdk_conn = MagicMock() + sdk_conn.subscribe.return_value = (never_acked, 7) + conn._connection = sdk_conn + conn._connected = True + + with pytest.raises(TimeoutError): + await conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE) + + @pytest.mark.asyncio + async def test_publish_resolves_before_timeout(self, mock_auth_client): + config = MqttConnectionConfig( + client_id="test-client", operation_timeout=1.0 + ) + conn = MqttConnection(config, mock_auth_client) + + acked = concurrent.futures.Future() + acked.set_result({"packet_id": 42}) + sdk_conn = MagicMock() + sdk_conn.publish.return_value = (acked, 42) + conn._connection = sdk_conn + conn._connected = True + + packet_id = await conn.publish("test/topic", {"key": "value"}) + assert packet_id == 42 + + +class TestScheduleCoroutineCleanup: + """Scheduling failures must not leak un-awaited coroutines.""" + + def test_coroutine_closed_when_no_loop_available(self, mock_auth_client): + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + client._loop = None # no captured loop, and no running loop here + + async def never_runs(): + raise AssertionError("must not execute") + + coro = never_runs() + client._schedule_coroutine(coro) # must not raise or warn + + # A closed coroutine raises RuntimeError when sent to + with pytest.raises(RuntimeError, match="cannot reuse already"): + coro.send(None) + + def test_coroutine_closed_when_loop_is_closed(self, mock_auth_client): + from nwp500.mqtt import NavienMqttClient + + client = NavienMqttClient(mock_auth_client) + loop = asyncio.new_event_loop() + loop.close() + client._loop = loop + + async def never_runs(): + raise AssertionError("must not execute") + + coro = never_runs() + client._schedule_coroutine(coro) + + with pytest.raises(RuntimeError, match="cannot reuse already"): + coro.send(None) + + def test_done_callback_logs_traceback(self, caplog): + """The log record must carry the coroutine's traceback.""" + import logging + + from nwp500.mqtt.client import _log_scheduled_coroutine_result + + def boom(): + raise MqttNotConnectedError("scheduled failure") + + future: concurrent.futures.Future = concurrent.futures.Future() + try: + boom() + except MqttNotConnectedError as e: + future.set_exception(e) + + with caplog.at_level(logging.ERROR, logger="nwp500.mqtt.client"): + _log_scheduled_coroutine_result(future) + + record = caplog.records[-1] + assert record.exc_info is not None + assert record.exc_info[0] is MqttNotConnectedError + # Traceback present and points at the raising function + assert "boom" in caplog.text + + +class TestAwaitAckFutureTypes: + """_await_ack must accept both concurrent and asyncio futures.""" + + @pytest.mark.asyncio + async def test_concurrent_future(self, mock_auth_client): + config = MqttConnectionConfig(client_id="t", operation_timeout=1.0) + conn = MqttConnection(config, mock_auth_client) + + future: concurrent.futures.Future = concurrent.futures.Future() + future.set_result({"ok": True}) + + result = await conn._await_ack(future, "Test") + assert result == {"ok": True} + + @pytest.mark.asyncio + async def test_asyncio_future(self, mock_auth_client): + config = MqttConnectionConfig(client_id="t", operation_timeout=1.0) + conn = MqttConnection(config, mock_auth_client) + + future = asyncio.get_running_loop().create_future() + future.set_result({"ok": True}) + + result = await conn._await_ack(future, "Test") + assert result == {"ok": True}