BrighterCommand / BrighterCommand/Brighter
RmqMessageGateway disposal evicts shared pooled IConnection, breaking concurrent consumers/producers
- Dominant language
- C#
- Stars
- 2.5k
- Forks
- 296
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 21
Description
## Summary
`RmqMessageGateway.DisposeAsync` (and sync `Dispose`) call `RmqMessageGatewayConnectionPool.RemoveConnectionAsync`, which evicts and disposes the **shared static** pooled `IConnection`. When multiple consumer/producer instances share that pooled connection (the design intent), disposing any one of them kills the broker connection that the others are still using.
## Repro
Two `RmqMessageProducer` / `RmqMessageConsumer` instances against the same broker URL, used concurrently. Dispose one while the other is mid-`Send`/`Receive`. The surviving instance's operations land on a torn-down `IConnection`/channel — pending publisher confirms never resolve, basic.consume callbacks stop, and consumers fall through to MT_NONE / MT_UNACCEPTABLE.
## Where
- `src/Paramore.Brighter.MessagingGateway.RMQ.Async/RmqMessageGateway.cs:203`
- `src/Paramore.Brighter.MessagingGateway.RMQ.Async/RmqMessageGateway.cs:214`
- equivalent in `src/Paramore.Brighter.MessagingGateway.RMQ.Sync/RmqMessageGateway.cs`
```csharp
public virtual async ValueTask DisposeAsync()
{
...
if (Channel != null) { /* dispose channel — fine, channel is per-instance */ }
// problem: this disposes the *shared* IConnection, not just our channel
await new RmqMessageGatewayConnectionPool(Connection.Name, Connection.Heartbeat)
.RemoveConnectionAsync(_connectionFactory);
}
```
## Why it's been latent
Production gateway disposal happens at app shutdown, when nothing else is publishing — the cross-tenant kill is invisible. The bug becomes visible the moment any two consumers or producers have overlapping disposal/usage lifetimes (e.g. a `Dispatcher` shutting down a single subscriber, parallel integration tests).
## Surfaced by
Migration of the test suite from xUnit to TUnit (`feature/tunit-migration`, PR #4065). xUnit's `[Collection(\"RMQ\")]` previously serialized the RMQ test class so only one gateway existed at a time. TUnit runs tests in parallel by default; `RmqMessageProducerSendMessageTests`, `RmqMessageProducerSendPersistentMessageTests`, `RmqMessageProducerDelayedMessageTests` etc. now overlap and trip the bug:
- `result.Body.Value == \"\"` instead of the published payload
- `result.Persist == false` despite producer's `PersistMessages = true`
- `MessageType == MT_UNACCEPTABLE` on requeue tests
- failing tests exit in 5–7ms (way below their 1–10s `Receive` timeout) because the dead channel returns immediately
## Expected behaviour
`IConnection` is intended to be long-lived and shared by many channels. Individual consumer/producer disposal should release **its own channel**, not tear down the connection used by sibling consumers/producers. The pool already self-heals via `GetConnectionAsync` (it re-creates if `Connection.IsOpen == false`).
## Proposed fix
Prefer making the connection pool lease-based/ref-counted rather than letting each gateway own the lifetime of a shared `IConnection`.
Suggested shape:
- `RmqMessageGatewayConnectionPool.AcquireConnectionAsync(...)` returns a lightweight lease/handle containing the shared `IConnection`.
- Acquiring a lease increments a reference count for the pooled connection key.
- Each producer/consumer owns and disposes only its per-instance channel.
- Disposing a gateway releases its lease; the shared `IConnection` is disposed only when the final lease for that key is released.
- `ResetConnectionAsync` should invalidate/swap the pooled connection deliberately and safely for all leases using that key, instead of acting like ordinary per-gateway disposal.
- If tests or hosts need to force process-level cleanup, expose an explicit pool shutdown/reset API that is not coupled to individual gateway disposal.
A simpler stopgap is to remove the `RemoveConnectionAsync` / `RemoveConnection` call from `RmqMessageGateway.DisposeAsync` and `Dispose(bool)` in both async and sync gateways, leaving shared broker connections alive until process exit or explicit reset. That likely restores the intended long-lived `IConnection` behaviour, but a lease/ref-counted pool is a clearer ownership model and avoids turning this into an unbounded connection lifetime decision.
## Risk
Medium-low. The intended production model is already a long-lived shared `IConnection` with per-gateway channels. The lease/ref-counted approach preserves that model while making ownership explicit. The main care point is `ResetConnectionAsync`: it must not leave existing leases with silently torn-down channels without a well-defined reconnect/error path.
Contributor guide
Assessment
This issue has not been assessed yet.