AsyncDefaultBasicConsumer state is unsynchronised under ConsumerDispatchConcurrency > 1 · Issue #2016 · rabbitmq/rabbitmq-dotnet-client · GitHub
Skip to content

AsyncDefaultBasicConsumer state is unsynchronised under ConsumerDispatchConcurrency > 1 #2016

Description

@lukebakken

Note

This issue was written by Claude (Anthropic's Claude Code) under the direction of @lukebakken. It was found by a code review of the fix for #2006. The code paths and types below were read directly; no corruption has been reproduced against a running process, so the second symptom in particular is a code-level finding rather than a field report.

AsyncDefaultBasicConsumer keeps three pieces of mutable state - IsRunning, ShutdownReason, and _consumerTags - and mutates all of them from consumer dispatcher work items. That is safe with the default ConsumerDispatchConcurrency of 1, because ConsumerDispatcherChannelBase then runs a single worker. Above 1 it starts one task per unit of concurrency, all reading from the same unbounded channel:

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/main/projects/RabbitMQ.Client/ConsumerDispatching/ConsumerDispatcherChannelBase.cs#L68-L74

if (_concurrency == 1)
{
    _worker = Task.Run(loopStart);
}
else
{
    var tasks = new Task[_concurrency];

SingleReader is set to _concurrency == 1, confirming multiple readers above 1. Enqueue order therefore does not imply execution order, and none of the three fields is synchronised.

Symptom 1: the two health signals can disagree

HandleChannelShutdownAsync sets ShutdownReason and clears IsRunning; HandleBasicConsumeOkAsync clears ShutdownReason and sets IsRunning. With concurrent workers a ConsumeOk item enqueued just before the channel died can be executed after the Shutdown item, leaving ShutdownReason == null and IsRunning == true on a channel that is gone. On a connection without automatic recovery nothing later corrects it, so both signals say healthy forever.

This ordering hazard predates the #2006 fix for IsRunning. That fix, which clears ShutdownReason on registration so it stops reporting a shutdown that recovery already dealt with, extends the same hazard to ShutdownReason, so the two can now be wrong together rather than only one of them. The trade is deliberate: the pre-fix behaviour was wrong on every recovery, whereas this needs a registration to race a shutdown with concurrency above 1.

A guard on Channel.CloseReason was considered and rejected while fixing #2006, because AutorecoveringChannel.CloseReason reads InnerChannel, whose getter calls ThrowIfDisposed(). Reading it from a dispatcher work item during teardown risks an ObjectDisposedException inside the dispatcher, which is worse than the inconsistency it would prevent.

Symptom 2: _consumerTags is a plain HashSet<string>

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/main/projects/RabbitMQ.Client/AsyncDefaultBasicConsumer.cs#L12

private readonly HashSet<string> _consumerTags = new HashSet<string>();

It is Added in HandleBasicConsumeOkAsync, Removed in OnCancelAsync, and enumerated by ConsumerTags via ToArray(), which application code calls on its own thread. With concurrency above 1, a ConsumeOk for one tag can run alongside a CancelOk for another on the same consumer instance, so two workers mutate the set at once. Concurrent mutation of HashSet<T> can corrupt its buckets or throw from enumeration, and the ToArray() from an application thread is unsynchronised against both.

A single consumer instance holding several tags is the normal way to hit this: the same consumer object registered against more than one queue.

Possible directions, not settled

  1. Serialise the consumer's own state rather than the dispatch: a lock around the three fields, or Interlocked/volatile for the two flags plus a concurrent collection for the tags. Local to this class and does not change dispatch semantics.
  2. Guarantee per-consumer ordering in the dispatcher, so work items for one consumer never execute concurrently. Stronger and fixes the ordering symptom properly, but a real change to the dispatcher's concurrency model.
  3. Document ConsumerDispatchConcurrency > 1 as giving no ordering guarantee for consumer state, and leave the fields alone. Cheapest, but leaves HashSet corruption on the table, which is a correctness problem rather than a semantics one.

Symptom 2 seems worth fixing on its own regardless of what is decided about ordering.

Related

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

A-consumer-dispatchArea: Consumer dispatching and concurrency.C-bugCategory: This is a bug.

Type

No type

Projects

No projects

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions