Fix InvalidStateError when cancelling MQTT futures during disconnect

[eman]

Problem

When disconnect() was called while MQTT operations (publish, subscribe, etc.) were in progress, the following error would appear:

Exception ignored in: <class 'concurrent.futures._base.InvalidStateError'>
Traceback (most recent call last):
  File "/usr/local/lib/python3.13/site-packages/awscrt/mqtt.py", line 825, in puback
    future.set_exception(awscrt.exceptions.from_code(error_code))
  File "/usr/local/lib/python3.13/concurrent/futures/_base.py", line 559, in set_exception
    raise InvalidStateError('{}: {!r}'.format(self._state, self))
concurrent.futures._base.InvalidStateError: CANCELLED: <Future at 0x7f64ac296150 state=cancelled>

Root Cause

asyncio.wrap_future() automatically cancels the underlying concurrent.futures.Future when the asyncio task is cancelled. When AWS CRT callbacks later tried to call set_exception() on these already-cancelled futures, Python raised InvalidStateError.

Solution

Wrap all asyncio.wrap_future() calls with asyncio.shield() to prevent cancellation from propagating to the underlying concurrent.futures.Future objects. This allows:

• ✅ The outer asyncio context to be cancelled cleanly
• ✅ AWS CRT futures to complete independently
• ✅ No InvalidStateError exceptions
• ✅ Graceful shutdown behavior

Changes

Modified Files

• src/nwp500/mqtt_connection.py: Shield connect(), disconnect(), subscribe(), unsubscribe(), and publish() operations
• src/nwp500/mqtt_subscriptions.py: Shield subscribe() and unsubscribe() operations

Technical Details

The fix uses asyncio.shield() to create a protective barrier that:

1. Allows the caller to be cancelled
2. Prevents cancellation from reaching AWS CRT futures
3. Lets AWS CRT operations complete independently
4. Prevents InvalidStateError when callbacks execute

Testing

• ✅ All 123 tests pass
• ✅ Type checking passes (mypy)
• ✅ Linting passes (make ci-lint)
• ✅ No InvalidStateError exceptions during disconnect

Impact

This is a bug fix that eliminates spurious error messages during normal disconnect operations. No breaking changes or API modifications.

[Copilot]

Pull Request Overview

This PR fixes an InvalidStateError that occurred when MQTT operations were cancelled during disconnect. The issue arose because asyncio.wrap_future() automatically cancels the underlying concurrent.futures.Future when the asyncio task is cancelled, causing AWS CRT callbacks to fail when they later tried to set exceptions on already-cancelled futures.

Key changes:

• Wrapped all asyncio.wrap_future() calls with asyncio.shield() to prevent cancellation propagation
• Added explicit cancellation handling in publish() with debug logging
• Updated docstring to document the asyncio.CancelledError exception

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
src/nwp500/mqtt_connection.py	Shields connect(), disconnect(), subscribe(), unsubscribe(), and publish() operations from cancellation; adds explicit error handling in publish()
src/nwp500/mqtt_subscriptions.py	Shields subscribe() and unsubscribe() operations from cancellation

[eman]

✅ Updated to address code review feedback

Applied consistent CancelledError handling across all shielded operations (connect(), disconnect(), subscribe(), unsubscribe(), publish()).

Changes in latest commit:

• All operations now have uniform cancellation handling with debug logging
• Makes it easier to diagnose cancellation-related issues
• Maintains the InvalidStateError fix while improving debugging capabilities

Validation:

• ✅ All 123 tests pass
• ✅ Type checking passes (mypy)
• ✅ Linting passes (make ci-lint)