Azure / Azure/azure-sdk-for-python

WebSocketTransportAsync.close() leaks aiohttp ClientSession when sock.close() raises

Open
#46,828 1 comment 0 reactions 0 assignees View on GitHub
bug Client customer-reported Event Hubs needs-team-attention Service Attention
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.