Run RabbitMQ deliveries through the core DeliveryDispatcher, with one scope per consumer attempt and no reply on shutdown - #423
Merged
Conversation
… scope per consumer attempt and no reply on shutdown
Code reviewNo issues found. Checked for bugs and CLAUDE.md compliance. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Moves the RabbitMQ transport onto the core
DeliveryDispatcherfrom #422. The worker keeps only the AMQP parts: ack,nack for dead-lettering, the retry-queue re-publish, the fault route and the reply route.
Production behavior changes
so a failed consumer's scoped state (for example an unsaved
DbContext) could reach the next consumer.again. Before, it replied with an
RpcFault("A task was canceled").OriginalContextis the context of the attempt that failed for good, so itsRetryCountis the round theconsumer failed on. For a one-way envelope delivery its
RequestIdis now empty; it held the AMQP correlation id.The fault's
Messageis still the payload as delivered.a logging scope with the queue, the routing key and the message type, so those entries keep that context.
receivespan.Fix
WithRetryCountand the retry re-publish copied the delivery's AMQP properties, but the copy shared the headerdictionary with the delivery, so setting
x-retry-countalso changed the original delivery. Both now copy theheaders.
Code
MessageHandler,RabbitMqHandlerFactoryandGatedPublisherare deleted.QueueDispatchPlansbuildsDeliveryHandlers withDeliveryHandlerFactory.RabbitMqBustakes theDeliveryDispatcherfrom DI instead of anIServiceScopeFactory.Tests
ConsumeFilterPipelineTests,PolymorphicDispatchTestsandRecordingGatedPublisherare deleted, because theytested pipeline and dispatch rules the core now owns. A core test pins that a request's filters finish before its
reply is sent, and a plan test pins which consumers a polymorphic plan holds.
dispatcher's entries, the fault round, and
WithRetryCountleaving the delivery unchanged.Docs
OriginalContextis the failing attempt. Request/Reply: noreply on shutdown. Observability: the dispatcher's log events, the delivery scope and the span events.
Verification
Vulthil.IntegrationTests(Testcontainers with a realbroker, 68 tests) and the Aspire messaging integration tests (net10.0, 13 tests).
dotnet packofVulthil.Messaging.RabbitMqpasses package validation against the 1.2.0 baseline.Backport to v1.0: no