[EventHub] Fix transport leak when Connection.close() hits its exception path

[MuhammadAliShahzad]

Summary

Fixes #46830.

Connection._disconnect() early-returned when state was already END. Connection.close() wraps the AMQP close handshake in try/except/finally; on any error during the handshake (timeout, network error, peer already closed, etc.) the except block sets state to END and the finally then calls _disconnect(). With the old code, _disconnect() saw state == END and early-returned, so self._transport.close() was never called.

With TransportType.AmqpOverWebsocket this leaked the aiohttp.ClientSession. We observed bursts of Unclosed client session warnings on instance shutdown — one per partition — because each partition's close-handshake races a network teardown.

Why the early return existed

It's not arbitrary — it prevents a double-close in the happy path. Tracing Connection.close(wait=True):

1. Send Close frame, state → CLOSE_SENT
2. _wait_for_response(wait=True, END) blocks
3. Peer sends Close back; the frame loop invokes _incoming_close(), which calls _disconnect() while state is CLOSE_SENT → transport correctly closed
4. State is now END; _wait_for_response returns; control returns to close()
5. finally calls _disconnect() again → previously early-returned (correct, but for the wrong reason)

The bug was only on the exception path, where _disconnect() had not yet been called.

Fix

Track close at the connection level via a self._transport_closed flag, so transport.close() runs exactly once regardless of which code path drives shutdown. Set the state to END if it isn't already; log + swallow errors from transport.close() (we're shutting down — propagating the error helps no one).

Applied to both:

• azure/eventhub/_pyamqp/aio/_connection_async.py
• azure/eventhub/_pyamqp/_connection.py

The sync sibling has the identical pattern; the user-visible leak is less severe for sync (no aiohttp.ClientSession to leak) but the logic flaw is the same.

Relationship to PR #46829

PR #46829 fixes the transport-level cleanup (WebSocketTransportAsync.close() and .connect()) so transport-level resources are properly released and transport.close() becomes idempotent. This PR fixes the connection-level code path that was skipping transport.close() entirely in the error case.

The two PRs are independent — either can land without the other — but together they close the full leak path observed in production. With both merged, leaked aiohttp.ClientSession warnings should disappear from EventHub websocket consumers.

Test plan

• New unit tests fail against current main for the right reasons (state == END causes early-return → transport.close() never called)
• All 8 new unit tests pass against the fixed code (4 async + 4 sync, covering: regression scenario, idempotency, normal state transition, transport.close() error resilience)
• Pylint clean for both modified files (rated 9.99/10 and 10/10 locally; only pre-existing unknown-option-value warnings from the pylint-guidelines-checker plugin in my env)
• Pyright reports zero new errors on the modified code; 2 pre-existing errors on unchanged _process_incoming_frame calls remain
• Mypy reports zero new errors on the modified code
• CI (azpysdk checks) — to be validated on PR run

🤖 Generated with Claude Code

[MuhammadAliShahzad]

@microsoft-github-policy-service agree

[github-actions]

Thank you for your contribution @MuhammadAliShahzad! We will review the pull request and get back to you soon.

[Copilot]

Pull request overview

Fixes a shutdown/cleanup bug in the pyAMQP Connection.close() exception path where _disconnect() could skip transport.close() (because state was already END), leaking transport resources—most notably aiohttp.ClientSession when using AmqpOverWebsocket.

Changes:

• Add a connection-level _transport_closed flag in both async and sync Connection implementations to ensure transport.close() is executed exactly once across all shutdown paths.
• Update _disconnect() to always transition to END (if needed) and to swallow/log transport close errors during shutdown.
• Add new unit tests (async + sync) covering the regression case, idempotency, state transition, and close-error resilience; document the fix in CHANGELOG.md.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py	Ensures transport closure occurs exactly once (even when close() hits exception paths) via _transport_closed gating.
sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_connection.py	Sync equivalent of the async fix to prevent skipped transport closure on shutdown error paths.
sdk/eventhub/azure-eventhub/tests/pyamqp_tests/asynctests/test_connection_disconnect_async_unit.py	New async unit tests validating _disconnect() behavior (regression + idempotency + error swallowing).
sdk/eventhub/azure-eventhub/tests/pyamqp_tests/synctests/test_connection_disconnect_unit.py	New sync unit tests mirroring the async coverage for _disconnect().
sdk/eventhub/azure-eventhub/CHANGELOG.md	Adds an unreleased bugfix entry describing the transport leak fix.

[MuhammadAliShahzad]

Pushed 49467b6 to address Build Analyze failures:

• Verify ChangeLogEntries: bumped VERSION in _version.py from 5.15.1 → 5.15.2 (so the verifier matches my new top-of-file entry) and trimmed the empty ### Features Added / ### Breaking Changes / ### Other Changes subsections.
• Run MyPy: the failures are all pre-existing on main in files not touched by this PR (_common.py, _producer.py, _consumer.py, _consumer_async.py). Verified by running the same mypy command locally against upstream/main. Happy to defer to maintainers on whether those should be fixed in a separate PR or whether the CI mypy check needs to be relaxed.

Note: both PRs target the same package version (5.15.2). Whichever lands first, the other can be merged afterwards without further version changes.

[j7nw4r]

Fix looks right. The flag splits what state END was overloading (logical state vs "cleanup ran"), and the four call sites (L227, L490, L506, L879) all behave correctly with it.

One ask: the unit tests exercise _disconnect() directly but never drive Connection.close() through its try/except/finally with _outgoing_close raising, which is the actual regression path. A test like:

connection._outgoing_close = AsyncMock(side_effect=RuntimeError("close-handshake failed"))
await connection.close()
connection._transport.close.assert_awaited_once()

would lock in the contract.

Minor: the docstring describes the original bug ("would early-return on the state check"), but the new code has no state check. Reads better describing the flag's role instead.

[j7nw4r]

Mypy failures in Run MyPy are pre-existing on main and unrelated to this change. I put up #46988 to fix all 16 of them (the _common.py setters / annotation accesses, the keep_alive_interval: Optional[int] widening in the transport layer, and the event_position_selector / create_source signature changes).

Path to green here: once #46988 lands, rebase this branch onto main. Happy to push the rebase from the maintainer side when that happens.

The earlier ask (a test that drives Connection.close() through its try/except/finally with _outgoing_close raising, so the regression case is locked in at the public API rather than just at _disconnect()) still stands.

[j7nw4r]

@MuhammadAliShahzad A release will pull in my changes which fix the failing CI tests.

[MuhammadAliShahzad]

Rebased onto current main now that #46988 has merged, which clears the pre-existing mypy failures.

Also added the requested public-API regression test (sync + async): it drives Connection.close() through its try/except/finally with _outgoing_close raising and asserts the transport is still closed exactly once. This locks the leak fix at the public API surface rather than only at _disconnect().
@j7nw4r