Azure / Azure/azure-sdk-for-python
WebSocketTransportAsync.close() leaks aiohttp ClientSession when sock.close() raises
- Dominant language
- Python
- Stars
- 5.6k
- Forks
- 3.4k
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 193
Description
### Package
`azure-eventhub` 5.15.1 (present since pyamqp was introduced in 5.8.0)
### Describe the bug
`WebSocketTransportAsync.close()` calls `self.sock.close()` and then `self.session.close()` sequentially without `try/finally`. If `sock.close()` raises (e.g. the WebSocket is already disconnected or timed out), `session.close()` never executes and the `aiohttp.ClientSession` leaks.
Each leaked session holds TCP connections, SSL state, and read buffers. In a long-running Cloud Run service processing EventHub messages, this manifests as a steady stream of `Unclosed client session` errors logged by aiohttp's garbage collector, and growing memory usage.
### Current code
https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py
```python
# WebSocketTransportAsync.close() — NOT exception-safe
async def close(self):
"""Do any preliminary work in shutting down the connection."""
async with self.socket_lock:
await self.sock.close() # if this raises...
await self.session.close() # ...this never runs → session leaks
self.connected = False
```
### Note: AsyncTransport.close() in the same file IS exception-safe
The sibling `AsyncTransport.close()` handles this correctly:
```python
# AsyncTransport.close() — exception-safe ✓
async def close(self):
if self.sock is not None:
try:
os.close(self.sock)
except OSError:
pass
self.sock = None
self.connected = False
```
### Secondary issue: connect() also leaks
`WebSocketTransportAsync.connect()` creates a new `ClientSession()` on line 451 without closing any previously assigned `self.session`. On reconnection, the old session leaks. Additionally, if `ws_connect()` raises anything other than `ClientConnectorError`, the newly created session is not cleaned up.
### Suggested fix
```python
async def close(self):
"""Do any preliminary work in shutting down the connection."""
async with self.socket_lock:
try:
await self.sock.close()
except Exception:
pass
try:
await self.session.close()
except Exception:
pass
self.connected = False
```
For `connect()`, close any existing session before creating a new one, and add a `try/except` around the `ws_connect()` call to clean up on failure:
```python
async def connect(self):
# ... existing setup code ...
# Close previous session if reconnecting
if self.session is not None:
try:
await self.session.close()
except Exception:
pass
self.session = None
self.session = ClientSession()
try:
self.sock = await self.session.ws_connect(...)
except Exception:
try:
await self.session.close()
except Exception:
pass
self.session = None
raise
self.connected = True
```
### Reproduction
Run an EventHub consumer/producer over WebSocket transport (`TransportType.AmqpOverWebsocket`) for an extended period with intermittent network disruptions. The `Unclosed client session` errors appear in logs on each reconnection cycle and on shutdown.
### Environment
- Python 3.12
- `azure-eventhub==5.15.1`
- `aiohttp==3.13.5`
- Cloud Run (long-lived instances)
- Transport: `TransportType.AmqpOverWebsocket`
Contributor guide
Assessment
This issue has not been assessed yet.