[cw-jlu]

Closing a listener after a delivery currently returns while RabbitMQ still has an active consumer. listen() does not await closure of its nested merged generator, so cancellation of the pump tasks and queue iterators depends on a later async-generator finalizer. This leaves a window for the abandoned listener to receive/prefetch more deliveries.

Wrap the merged generator in contextlib.aclosing so explicit close or an exception thrown at the outer yield waits for the existing pump/queue cleanup. The shared broker channels remain available for subsequent listeners. There is no change to routing, acknowledgements, retries, or public signatures.

Fixes #60. Assignment has been requested; no assignee has been set.

Validation actually run

Environment: Windows, Python 3.14.7, locked aio-pika 9.5.8 / aiormq 6.9.2, and a real isolated RabbitMQ 4.3.0 Docker container over loopback AMQP.

• pytest tests/test_listener.py -q: 6 passed on the patch. The same tests on unmodified master f63aa6057475f0c83605245b5255d6a9ffc9c589 failed 6/6 in an isolated sequential run. Explicit close/throw fail on the first server-side consumer-count check; cancellation already cleans the first listener on baseline, but its subsequent explicitly closed listener fails the recovery check.
• Tests cover one/two queues, aclose, athrow, cancellation of a pending receive, server-side consumer counts after close returns, acknowledgement, and reuse of the shared channel to receive/ack the next delivery. No AMQP client/server is mocked.
• Linux / Python 3.12 in a disposable Python container sharing the real RabbitMQ network: the same 6 regressions failed on the unmodified baseline and 6 passed on the patch. aio-pika 9.5.8 / aiormq 6.9.2 and Taskiq 0.12.6 were pinned. The read-only source mounts produced pytest-cache warnings only.
• Broader suite with -n 2, excluding two tests that require the unavailable delayed-message plugin: 36 passed, 2 failed. The exact two existing failure nodes also fail in the same environment on unmodified master: test_when_delivery_limit_exceeded__message_is_dead_lettered (QueueEmpty not raised), and TestStartup::test_when_declare_flag_not_passed_to_queue__broker_does_not_declare_queue_on_startup (RabbitMQ 4.3 rejects deprecated non-durable, non-exclusive queues). The baseline broader run, including the copied new tests, was 31 passed, 7 failed: these same two existing failures plus five new regression failures; the cancellation/recovery timing case can already pass on baseline due to deferred finalization.
• Whole-tree Black, Ruff and ty check ./taskiq_aio_pika ./tests passed. All repository prek run --all-files hooks passed after staging both changed files, including typos and detect-secrets. git diff --cached --check passed.

Not run locally: the two delayed-message-plugin tests, a complete Taskiq CLI worker-process end-to-end test, broker outage/reconnection, or the Python/OS CI matrix. Existing full-suite failures remain unresolved by this focused patch; CI needs its configured RabbitMQ/plugin environment.

Review status

AI-assisted investigation, implementation, and testing. Please keep this PR in Draft pending independent review and account-owner review. No completed human review is claimed.