You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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:
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>
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
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.
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.
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.
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.
AsyncDefaultBasicConsumerkeeps three pieces of mutable state -IsRunning,ShutdownReason, and_consumerTags- and mutates all of them from consumer dispatcher work items. That is safe with the defaultConsumerDispatchConcurrencyof 1, becauseConsumerDispatcherChannelBasethen 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
SingleReaderis 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
HandleChannelShutdownAsyncsetsShutdownReasonand clearsIsRunning;HandleBasicConsumeOkAsyncclearsShutdownReasonand setsIsRunning. With concurrent workers aConsumeOkitem enqueued just before the channel died can be executed after theShutdownitem, leavingShutdownReason == nullandIsRunning == trueon 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 clearsShutdownReasonon registration so it stops reporting a shutdown that recovery already dealt with, extends the same hazard toShutdownReason, 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.CloseReasonwas considered and rejected while fixing #2006, becauseAutorecoveringChannel.CloseReasonreadsInnerChannel, whose getter callsThrowIfDisposed(). Reading it from a dispatcher work item during teardown risks anObjectDisposedExceptioninside the dispatcher, which is worse than the inconsistency it would prevent.Symptom 2:
_consumerTagsis a plainHashSet<string>https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/main/projects/RabbitMQ.Client/AsyncDefaultBasicConsumer.cs#L12
It is
Added inHandleBasicConsumeOkAsync,Removed inOnCancelAsync, and enumerated byConsumerTagsviaToArray(), which application code calls on its own thread. With concurrency above 1, aConsumeOkfor one tag can run alongside aCancelOkfor another on the same consumer instance, so two workers mutate the set at once. Concurrent mutation ofHashSet<T>can corrupt its buckets or throw from enumeration, and theToArray()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
Interlocked/volatilefor the two flags plus a concurrent collection for the tags. Local to this class and does not change dispatch semantics.ConsumerDispatchConcurrency > 1as giving no ordering guarantee for consumer state, and leave the fields alone. Cheapest, but leavesHashSetcorruption 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
ShutdownReasonon registration, which is where this was foundConsumerDispatchConcurrencyonIConnectionFactoryandCreateChannelOptions