BrighterCommand / BrighterCommand/Brighter

RmqMessageGateway disposal evicts shared pooled IConnection, breaking concurrent consumers/producers

Open
#4,102 1 comment 0 reactions 1 assignee Claimed by @thomhurst View on GitHub
.NET 0 - Backlog Bug
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

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.