Context
#571 rebuilt the subscription lifecycle around a supervisor loop: each connect attempt gets a SubscriptionRun, and a transport signals its death via run.Fail(...) so the supervisor can tear the run down and resubscribe. Every transport follows this contract except RabbitMQ.
RabbitMqSubscription.Connect wires only consumer.ReceivedAsync. Consumer cancellation, channel shutdown, and connection shutdown are never mapped to run.Fail, so when the broker cancels the consumer or the channel/connection dies for good (notably with automatic recovery disabled, or when recovery gives up), run.Ended never completes: the supervisor stays parked, no drop is reported, health stays green, and the subscription consumes nothing until the host restarts.
This is not a regression from #571 — the previous implementation never called Dropped for RabbitMQ either, which is why it was split out of that PR (it surfaced during its review). c061dd7 in #571 fixed the adjacent case of handler failures being swallowed under ThrowOnError; this issue covers transport-level death.
Suggested direction
In Connect, map the client's shutdown signals to the run:
IChannel.ChannelShutdownAsyncIConnection.ConnectionShutdownAsync- consumer cancellation/unregistration on
AsyncEventingBasicConsumer
each calling run.Fail(DropReason.ServerError, ...) — but only when the shutdown isn't the run's own teardown (guard on run.Token.IsCancellationRequested, as the other transports do), so a clean stop isn't double-reported.
The interaction with the client's automatic recovery (on by default in ConnectionFactory) needs a decision: a recoverable shutdown may be healed by the client without a resubscribe, so either
- treat any shutdown as a drop and let the supervisor own recovery (simple, predictable, matches the other transports), or
- only fail the run on shutdowns the client reports as non-recoverable, and keep relying on auto-recovery for the rest.
Option 1 is likely the right default; relying on auto-recovery means the supervisor never learns the transport flapped, and per-run state (sequence, checkpoint handler) silently spans reconnects.
Refs: #571
Context
#571 rebuilt the subscription lifecycle around a supervisor loop: each connect attempt gets a
SubscriptionRun, and a transport signals its death viarun.Fail(...)so the supervisor can tear the run down and resubscribe. Every transport follows this contract except RabbitMQ.RabbitMqSubscription.Connectwires onlyconsumer.ReceivedAsync. Consumer cancellation, channel shutdown, and connection shutdown are never mapped torun.Fail, so when the broker cancels the consumer or the channel/connection dies for good (notably with automatic recovery disabled, or when recovery gives up),run.Endednever completes: the supervisor stays parked, no drop is reported, health stays green, and the subscription consumes nothing until the host restarts.This is not a regression from #571 — the previous implementation never called
Droppedfor RabbitMQ either, which is why it was split out of that PR (it surfaced during its review). c061dd7 in #571 fixed the adjacent case of handler failures being swallowed underThrowOnError; this issue covers transport-level death.Suggested direction
In
Connect, map the client's shutdown signals to the run:IChannel.ChannelShutdownAsyncIConnection.ConnectionShutdownAsyncAsyncEventingBasicConsumereach calling
run.Fail(DropReason.ServerError, ...)— but only when the shutdown isn't the run's own teardown (guard onrun.Token.IsCancellationRequested, as the other transports do), so a clean stop isn't double-reported.The interaction with the client's automatic recovery (on by default in
ConnectionFactory) needs a decision: a recoverable shutdown may be healed by the client without a resubscribe, so eitherOption 1 is likely the right default; relying on auto-recovery means the supervisor never learns the transport flapped, and per-run state (sequence, checkpoint handler) silently spans reconnects.
Refs: #571