Consume MQTT ack futures on abandonment to prevent asyncio warnings

CHANGELOG.rst

@@ -5,6 +5,22 @@ Changelog
 Unreleased
 ==========
 
+Fixed
+-----
+- **MQTT ack futures now consumed on abandonment** (`#97
+  <https://github.com/eman/nwp500-python/issues/97>`_): ``_await_ack()``
+  in ``mqtt/connection.py`` and the inline subscribe/unsubscribe
+  acknowledgement waits in ``mqtt/subscriptions.py`` shield the AWS CRT
+  future so a timeout or cancellation doesn't propagate into the SDK
+  future. Previously, if the shielded future later completed with an
+  exception (e.g. ``AwsCrtError`` from clean-session cancellation during
+  reconnect) after the awaiting task had already given up, nobody
+  retrieved that exception, and asyncio logged a "Future exception was
+  never retrieved" warning at garbage-collection time. A done callback
+  is now attached whenever a wait is abandoned so the eventual
+  result/exception is always retrieved and logged at debug level
+  instead of leaking as an unhandled asyncio warning.
+
 Version 9.0.0 (2026-07-05)
 ==========================
 

src/nwp500/mqtt/connection.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import concurrent.futures
+import functools
 import json
 import logging
 from collections.abc import Callable
@@ -95,6 +96,30 @@ def __init__(
             f"Initialized connection manager with client ID: {config.client_id}"
         )
 
+    @staticmethod
+    def _consume_abandoned_future(
+        future: asyncio.Future[Any], operation: str
+    ) -> None:
+        """Retrieve the eventual result of an abandoned acknowledgement future.
+
+        When ``_await_ack`` is cancelled or times out, the shielded future
+        keeps running in the background. If nobody retrieves its eventual
+        exception, asyncio logs a "Future exception was never retrieved"
+        warning at garbage-collection time. This callback consumes that
+        result/exception so the warning is never emitted, logging benign
+        errors (e.g. ``AwsCrtError`` from clean-session cancellation during
+        reconnect) at debug level instead.
+        """
+        if future.cancelled():
+            return
+        exc = future.exception()
+        if exc is not None:
+            _logger.debug(
+                "%s completed with exception after being abandoned: %s",
+                operation,
+                exc,
+            )
+
     async def _await_ack(
         self,
         future: concurrent.futures.Future[Any] | asyncio.Future[Any],
@@ -110,6 +135,11 @@ async def _await_ack(
         acknowledgement never arrives; without it callers can hang until
         the keep-alive expires (20+ minutes).
 
+        If the awaiting task is cancelled or times out before the shielded
+        future completes, a done callback is attached so its eventual
+        result/exception is still retrieved, preventing asyncio's "Future
+        exception was never retrieved" warnings.
+
         Args:
             future: Acknowledgement future — the
                 ``concurrent.futures.Future`` returned by the AWS CRT SDK
@@ -139,13 +169,23 @@ async def _await_ack(
             _logger.debug(
                 "%s was cancelled but will complete in background", operation
             )
+            awaitable_future.add_done_callback(
+                functools.partial(
+                    self._consume_abandoned_future, operation=operation
+                )
+            )
             raise
         except TimeoutError:
             _logger.error(
                 "%s not acknowledged within %.1fs",
                 operation,
                 self.config.operation_timeout,
             )
+            awaitable_future.add_done_callback(
+                functools.partial(
+                    self._consume_abandoned_future, operation=operation
+                )
+            )
             raise
 
     async def connect(self) -> bool:

src/nwp500/mqtt/subscriptions.py

@@ -10,6 +10,7 @@
 """
 
 import asyncio
+import functools
 import json
 import logging
 from collections.abc import Callable
@@ -176,6 +177,30 @@ async def _dispatch_message(self, topic: str, payload: bytes) -> None:
                             redact_topic(topic),
                         )
 
+    @staticmethod
+    def _consume_abandoned_future(
+        future: asyncio.Future[Any], operation: str
+    ) -> None:
+        """Retrieve the eventual result of an abandoned acknowledgement future.
+
+        When the shielded acknowledgement future outlives a cancelled or
+        timed-out awaiter, nothing else consumes its eventual result. If it
+        later completes with an exception, asyncio logs a "Future exception
+        was never retrieved" warning at garbage-collection time. This
+        callback retrieves that result/exception so the warning is never
+        emitted, logging benign errors (e.g. ``AwsCrtError`` from
+        clean-session cancellation during reconnect) at debug level instead.
+        """
+        if future.cancelled():
+            return
+        exc = future.exception()
+        if exc is not None:
+            _logger.debug(
+                "%s completed with exception after being abandoned: %s",
+                operation,
+                exc,
+            )
+
     async def subscribe(
         self,
         topic: str,
@@ -224,16 +249,35 @@ async def subscribe(
             subscribe_future, packet_id = self._connection.subscribe(
                 topic=topic, qos=qos, callback=self._on_message_received
             )
+            awaitable_future = asyncio.wrap_future(subscribe_future)
             try:
                 subscribe_result = await asyncio.wait_for(
-                    asyncio.shield(asyncio.wrap_future(subscribe_future)),
+                    asyncio.shield(awaitable_future),
                     timeout=self._operation_timeout,
                 )
             except asyncio.CancelledError:
                 _logger.debug(
                     f"Subscribe to '{redact_topic(topic)}' was cancelled "
                     "but will complete in background"
                 )
+                awaitable_future.add_done_callback(
+                    functools.partial(
+                        MqttSubscriptionManager._consume_abandoned_future,
+                        operation=f"Subscribe to '{redact_topic(topic)}'",
+                    )
+                )
+                raise
+            except TimeoutError:
+                _logger.error(
+                    f"Subscribe to '{redact_topic(topic)}' not "
+                    f"acknowledged within {self._operation_timeout:.1f}s"
+                )
+                awaitable_future.add_done_callback(
+                    functools.partial(
+                        MqttSubscriptionManager._consume_abandoned_future,
+                        operation=f"Subscribe to '{redact_topic(topic)}'",
+                    )
+                )
                 raise
 
             _logger.info(
@@ -303,16 +347,35 @@ async def unsubscribe(
             # Await the broker acknowledgement with a timeout (see
             # subscribe() for shielding rationale).
             unsubscribe_future, packet_id = self._connection.unsubscribe(topic)
+            awaitable_future = asyncio.wrap_future(unsubscribe_future)
             try:
                 await asyncio.wait_for(
-                    asyncio.shield(asyncio.wrap_future(unsubscribe_future)),
+                    asyncio.shield(awaitable_future),
                     timeout=self._operation_timeout,
                 )
             except asyncio.CancelledError:
                 _logger.debug(
                     "Unsubscribe from topic (redacted) was "
                     "cancelled but will complete in background"
                 )
+                awaitable_future.add_done_callback(
+                    functools.partial(
+                        MqttSubscriptionManager._consume_abandoned_future,
+                        operation="Unsubscribe from topic (redacted)",
+                    )
+                )
+                raise
+            except TimeoutError:
+                _logger.error(
+                    "Unsubscribe from topic (redacted) not "
+                    f"acknowledged within {self._operation_timeout:.1f}s"
+                )
+                awaitable_future.add_done_callback(
+                    functools.partial(
+                        MqttSubscriptionManager._consume_abandoned_future,
+                        operation="Unsubscribe from topic (redacted)",
+                    )
+                )
                 raise
 
             # Remove from tracking

tests/test_mqtt_reliability.py

@@ -604,6 +604,90 @@ def boom():
         assert "boom" in caplog.text
 
 
+class TestAbandonedAckFutureConsumed:
+    """Abandoned ack futures must not leak "never retrieved" warnings.
+
+    Regression test for https://github.com/eman/nwp500-python/issues/97:
+    when ``_await_ack`` times out or is cancelled, the shielded future
+    keeps running in the background. If nobody retrieves its eventual
+    exception, asyncio logs a "Future exception was never retrieved"
+    warning at garbage-collection time. We assert the fix retrieves the
+    exception itself (logged at debug level) rather than relying on GC
+    timing to observe the (absence of the) warning.
+    """
+
+    @pytest.mark.asyncio
+    async def test_timeout_then_late_exception_is_consumed(
+        self, mock_auth_client, caplog
+    ):
+        config = MqttConnectionConfig(
+            client_id="test-client", operation_timeout=0.05
+        )
+        conn = MqttConnection(config, mock_auth_client)
+
+        raw_future: concurrent.futures.Future = concurrent.futures.Future()
+        sdk_conn = MagicMock()
+        sdk_conn.subscribe.return_value = (raw_future, 7)
+        conn._connection = sdk_conn
+        conn._connected = True
+
+        with pytest.raises(TimeoutError):
+            await conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE)
+
+        from awscrt.exceptions import AwsCrtError
+
+        with caplog.at_level(logging.DEBUG, logger="nwp500.mqtt.connection"):
+            # The ack arrives late, after the caller has given up.
+            raw_future.set_exception(
+                AwsCrtError(
+                    0,
+                    "AWS_ERROR_MQTT_CANCELLED_FOR_CLEAN_SESSION",
+                    "clean session cancellation",
+                )
+            )
+            # Let the done callback attached by _await_ack run.
+            await asyncio.sleep(0)
+
+        assert "completed with exception after being abandoned" in (caplog.text)
+
+    @pytest.mark.asyncio
+    async def test_cancelled_then_late_exception_is_consumed(
+        self, mock_auth_client, caplog
+    ):
+        config = MqttConnectionConfig(
+            client_id="test-client", operation_timeout=5.0
+        )
+        conn = MqttConnection(config, mock_auth_client)
+
+        raw_future: concurrent.futures.Future = concurrent.futures.Future()
+        sdk_conn = MagicMock()
+        sdk_conn.subscribe.return_value = (raw_future, 7)
+        conn._connection = sdk_conn
+        conn._connected = True
+
+        task = asyncio.ensure_future(
+            conn.subscribe("test/topic", qos=mqtt.QoS.AT_LEAST_ONCE)
+        )
+        await asyncio.sleep(0)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+        from awscrt.exceptions import AwsCrtError
+
+        with caplog.at_level(logging.DEBUG, logger="nwp500.mqtt.connection"):
+            raw_future.set_exception(
+                AwsCrtError(
+                    0,
+                    "AWS_ERROR_MQTT_CANCELLED_FOR_CLEAN_SESSION",
+                    "clean session cancellation",
+                )
+            )
+            await asyncio.sleep(0)
+
+        assert "completed with exception after being abandoned" in (caplog.text)
+
+
 class TestAwaitAckFutureTypes:
     """_await_ack must accept both concurrent and asyncio futures."""