From 0de45b3e82037d483ec171dc69670654dbde808a Mon Sep 17 00:00:00 2001 From: emmanuel Date: Mon, 6 Jul 2026 08:36:37 -0700 Subject: [PATCH 1/2] Consume MQTT ack futures on abandonment to prevent asyncio warnings Fixes #97. _await_ack() (mqtt/connection.py) and the inline subscribe/unsubscribe waits (mqtt/subscriptions.py) shield the AWS CRT acknowledgement future so a timeout/cancellation never propagates into the SDK future. If that shielded future later completed with an exception (e.g. AwsCrtError from clean-session cancellation during reconnect) after the caller had already given up, nobody retrieved the exception and asyncio logged an unhandled 'Future exception was never retrieved' / 'exception in shielded future' warning at garbage-collection time. Attach a done callback whenever a wait is abandoned (cancelled or timed out) so the eventual result/exception is retrieved and logged at debug level, instead of leaking as an asyncio-level warning that integrations had to suppress globally. --- CHANGELOG.rst | 16 ++++++ src/nwp500/mqtt/connection.py | 40 +++++++++++++++ src/nwp500/mqtt/subscriptions.py | 62 ++++++++++++++++++++++- tests/test_mqtt_reliability.py | 84 ++++++++++++++++++++++++++++++++ 4 files changed, 200 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 93e5a1fc..93b0ae01 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -5,6 +5,22 @@ Changelog Unreleased ========== +Fixed +----- +- **MQTT ack futures now consumed on abandonment** (`#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) ========================== diff --git a/src/nwp500/mqtt/connection.py b/src/nwp500/mqtt/connection.py index 2bc3f4f4..41bb97a0 100644 --- a/src/nwp500/mqtt/connection.py +++ b/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,6 +169,11 @@ 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( @@ -146,6 +181,11 @@ async def _await_ack( operation, self.config.operation_timeout, ) + awaitable_future.add_done_callback( + functools.partial( + self._consume_abandoned_future, operation=operation + ) + ) raise async def connect(self) -> bool: diff --git a/src/nwp500/mqtt/subscriptions.py b/src/nwp500/mqtt/subscriptions.py index b64ce165..5e5fc7eb 100644 --- a/src/nwp500/mqtt/subscriptions.py +++ b/src/nwp500/mqtt/subscriptions.py @@ -176,6 +176,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,9 +248,10 @@ 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: @@ -234,6 +259,22 @@ async def subscribe( f"Subscribe to '{redact_topic(topic)}' was cancelled " "but will complete in background" ) + awaitable_future.add_done_callback( + lambda f: self._consume_abandoned_future( + f, 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( + lambda f: self._consume_abandoned_future( + f, f"Subscribe to '{redact_topic(topic)}'" + ) + ) raise _logger.info( @@ -303,9 +344,10 @@ 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: @@ -313,6 +355,22 @@ async def unsubscribe( "Unsubscribe from topic (redacted) was " "cancelled but will complete in background" ) + awaitable_future.add_done_callback( + lambda f: self._consume_abandoned_future( + f, "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( + lambda f: self._consume_abandoned_future( + f, "Unsubscribe from topic (redacted)" + ) + ) raise # Remove from tracking diff --git a/tests/test_mqtt_reliability.py b/tests/test_mqtt_reliability.py index 2b410993..ab1bc617 100644 --- a/tests/test_mqtt_reliability.py +++ b/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.""" From 7f7b6bc31bdd8058922b20ba370a75cec9163ee8 Mon Sep 17 00:00:00 2001 From: emmanuel Date: Mon, 6 Jul 2026 08:49:04 -0700 Subject: [PATCH 2/2] Avoid capturing self in abandoned-future done callbacks Use functools.partial with the static _consume_abandoned_future helper referenced via the class, rather than a lambda closing over self, so an abandoned subscribe/unsubscribe ack future doesn't keep the MqttSubscriptionManager instance alive until it eventually completes. --- src/nwp500/mqtt/subscriptions.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/src/nwp500/mqtt/subscriptions.py b/src/nwp500/mqtt/subscriptions.py index 5e5fc7eb..4f718faa 100644 --- a/src/nwp500/mqtt/subscriptions.py +++ b/src/nwp500/mqtt/subscriptions.py @@ -10,6 +10,7 @@ """ import asyncio +import functools import json import logging from collections.abc import Callable @@ -260,8 +261,9 @@ async def subscribe( "but will complete in background" ) awaitable_future.add_done_callback( - lambda f: self._consume_abandoned_future( - f, f"Subscribe to '{redact_topic(topic)}'" + functools.partial( + MqttSubscriptionManager._consume_abandoned_future, + operation=f"Subscribe to '{redact_topic(topic)}'", ) ) raise @@ -271,8 +273,9 @@ async def subscribe( f"acknowledged within {self._operation_timeout:.1f}s" ) awaitable_future.add_done_callback( - lambda f: self._consume_abandoned_future( - f, f"Subscribe to '{redact_topic(topic)}'" + functools.partial( + MqttSubscriptionManager._consume_abandoned_future, + operation=f"Subscribe to '{redact_topic(topic)}'", ) ) raise @@ -356,8 +359,9 @@ async def unsubscribe( "cancelled but will complete in background" ) awaitable_future.add_done_callback( - lambda f: self._consume_abandoned_future( - f, "Unsubscribe from topic (redacted)" + functools.partial( + MqttSubscriptionManager._consume_abandoned_future, + operation="Unsubscribe from topic (redacted)", ) ) raise @@ -367,8 +371,9 @@ async def unsubscribe( f"acknowledged within {self._operation_timeout:.1f}s" ) awaitable_future.add_done_callback( - lambda f: self._consume_abandoned_future( - f, "Unsubscribe from topic (redacted)" + functools.partial( + MqttSubscriptionManager._consume_abandoned_future, + operation="Unsubscribe from topic (redacted)", ) ) raise