dotnet-messaging-patterns

name	dotnet-messaging-patterns
description	Builds event-driven systems. Pub/sub, competing consumers, DLQ, sagas, delivery guarantees.

dotnet-messaging-patterns

Durable messaging patterns for .NET event-driven architectures. Covers publish/subscribe, competing consumers, dead-letter queues, saga/process manager orchestration, and delivery guarantee strategies using Azure Service Bus, RabbitMQ, and MassTransit.

Scope

• Publish/subscribe and competing consumer patterns
• Dead-letter queues and poison message handling
• Saga/process manager orchestration
• Delivery guarantee strategies (at-least-once, exactly-once)
• Azure Service Bus, RabbitMQ, and MassTransit integration

Out of scope

• Background service lifecycle and IHostedService registration -- see [skill:dotnet-background-services]
• Resilience pipelines and retry policies -- see [skill:dotnet-resilience]
• JSON/binary serialization configuration -- see [skill:dotnet-serialization]
• In-process producer/consumer queues with Channel -- see [skill:dotnet-channels]

Cross-references: [skill:dotnet-background-services] for hosting message consumers, [skill:dotnet-resilience] for fault tolerance around message handlers, [skill:dotnet-serialization] for message envelope serialization, [skill:dotnet-channels] for in-process queuing patterns.

---

Messaging Fundamentals

Message Types

Type	Purpose	Example
Command	Request an action (one recipient)	PlaceOrder, ShipPackage
Event	Notify something happened (many recipients)	OrderPlaced, PaymentReceived
Document	Transfer data between systems	CustomerProfile, ProductCatalog

Commands are sent to a specific queue; events are published to a topic/exchange and delivered to all subscribers. This distinction drives the choice between point-to-point and pub/sub topologies.

Delivery Guarantees

Guarantee	Behavior	Implementation
At-most-once	Fire and forget; message may be lost	No ack, no retry
At-least-once	Message retried until acknowledged; duplicates possible	Ack after processing + retry on failure
Exactly-once	Each message processed exactly once	At-least-once + idempotent consumer

At-least-once with idempotent consumers is the standard approach for durable messaging. True exactly-once requires distributed transactions (which most brokers do not support) or consumer-side deduplication.

---

Publish/Subscribe

Azure Service Bus Topics

// Publisher -- send event to a topic
await using var client = new ServiceBusClient(connectionString);
await using var sender = client.CreateSender("order-events");

var message = new ServiceBusMessage(
    JsonSerializer.SerializeToUtf8Bytes(new OrderPlaced(orderId, total)))
{
    Subject = nameof(OrderPlaced),
    ContentType = "application/json",
    MessageId = Guid.NewGuid().ToString()
};

await sender.SendMessageAsync(message, cancellationToken);

```text

```csharp

// Subscriber -- process events from a subscription
await using var processor = client.CreateProcessor(
    topicName: "order-events",
    subscriptionName: "billing-service",
    new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = 10,
        AutoCompleteMessages = false
    });

processor.ProcessMessageAsync += async args =>
{
    var body = args.Message.Body.ToObjectFromJson<OrderPlaced>();
    await HandleOrderPlacedAsync(body);
    await args.CompleteMessageAsync(args.Message);
};

processor.ProcessErrorAsync += args =>
{
    logger.LogError(args.Exception, "Error processing message");
    return Task.CompletedTask;
};

await processor.StartProcessingAsync(cancellationToken);

```text

**Key packages:**

```xml

<PackageReference Include="Azure.Messaging.ServiceBus" Version="7.*" />

```xml



```csharp


 factory =  ConnectionFactory { HostName =  };
   connection =  factory.CreateConnectionAsync();
   channel =  connection.CreateChannelAsync();

 channel.ExchangeDeclareAsync(
    exchange: ,
    type: ExchangeType.Fanout,
    durable: );

 body = JsonSerializer.SerializeToUtf8Bytes(
     OrderPlaced(orderId, total));

 channel.BasicPublishAsync(
    exchange: ,
    routingKey: .Empty,
    body: body);

```text

**Key packages:**

```xml

<PackageReference Include= Version= />

```xml



MassTransit abstracts the broker, providing a unified API  Azure Service Bus, RabbitMQ, Amazon SQS,  -memory
transport.

```csharp


builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(, , h =>
        {
            h.Username();
            h.Password();
        });
        cfg.ConfigureEndpoints(context);
    });
});



{
    
    {
        
         publishEndpoint.Publish(
             OrderPlaced(orderId, total), ct);
    }
}



{
    
    {
        logger.LogInformation(
            , context.Message.OrderId);
         ProcessAsync(context.Message);
    }
}


;

```text

**Key packages:**

```xml

<PackageReference Include= Version= />
<!-- Pick ONE transport: -->
<PackageReference Include= Version= />
<!-- OR -->
<PackageReference Include= Version= />

```text

---



Multiple consumer instances process messages  the same queue  parallel. The broker delivers each message to
exactly one consumer, distributing load across instances.



```text

Queue: order-processing
  ├──  = client.CreateProcessor(,
     ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = ,
        PrefetchCount = ,
        AutoCompleteMessages = 
    });

```text



```csharp

x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseConcurrentMessageLimit();
});

```text



Competing consumers sacrifice strict ordering  throughput. When order matters:

- **Azure Service Bus**: ;


   dlqReceiver = client.CreateReceiver(
    ,
     ServiceBusReceiverOptions
    {
        SubQueue = SubQueue.DeadLetter
    });

 ()
{
     message =  dlqReceiver.ReceiveMessageAsync(
        TimeSpan.FromSeconds(), cancellationToken);
     (message  ) ;

    logger.LogWarning(
        ,
        message.DeadLetterReason,
        message.DeadLetterErrorDescription);

    
     dlqReceiver.CompleteMessageAsync(message);
}

```text



MassTransit automatically creates `_error`  `_skipped` queues. Failed messages after retry exhaustion move to the
error queue  fault metadata.

```csharp


x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseMessageRetry(r => r.Intervals(
        TimeSpan.FromSeconds(),
        TimeSpan.FromSeconds(),
        TimeSpan.FromSeconds()));
});

```text



Always monitor DLQ depth  alerts. Unmonitored DLQs accumulate silently until data  lost  stale.

---



Sagas coordinate multi-step business processes across services. Each step publishes events that trigger the next step,
 compensation logic  failures.



| Style             | How it works                                                   | Use                                                 |
| ----------------- | -------------------------------------------------------------- | ------------------------------------------------------- |
| **Choreography**  | Services react to events independently; no central coordinator | Simple flows, few steps, loosely coupled                |
| **Orchestration** | A saga/process manager directs each step                       | Complex flows, compensation needed, visibility  |



```csharp


   : 
{
     Guid CorrelationId { ; ; }
      CurrentState { ; ; } = !;
     Guid OrderId { ; ; }
      Total { ; ; }
     DateTime? PaymentReceivedAt { ; ; }
}


    : <>
{
     State Submitted { ;  ; } = !;
     State PaymentPending { ;  ; } = !;
     State Completed { ;  ; } = !;
     State Faulted { ;  ; } = !;

     Event<OrderSubmitted> OrderSubmitted { ;  ; } = !;
     Event<PaymentReceived> PaymentReceived { ;  ; } = !;
     Event<PaymentFailed> PaymentFailed { ;  ; } = !;

    
    {
        InstanceState(x => x.CurrentState);

        Event(() => OrderSubmitted,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentReceived,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentFailed,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));

        Initially(
            When(OrderSubmitted)
                .Then(ctx =>
                {
                    ctx.Saga.OrderId = ctx.Message.OrderId;
                    ctx.Saga.Total = ctx.Message.Total;
                })
                .Publish(ctx =>  RequestPayment(
                    ctx.Saga.OrderId, ctx.Saga.Total))
                .TransitionTo(PaymentPending));

        During(PaymentPending,
            When(PaymentReceived)
                .Then(ctx =>
                    ctx.Saga.PaymentReceivedAt = DateTime.UtcNow)
                .Publish(ctx =>  FulfillOrder(ctx.Saga.OrderId))
                .TransitionTo(Completed),
            When(PaymentFailed)
                .Publish(ctx =>  CancelOrder(ctx.Saga.OrderId))
                .TransitionTo(Faulted));
    }
}



builder.Services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .EntityFrameworkRepository(r =>
        {
            r.ExistingDbContext<SagaDbContext>();
            r.UsePostgres();
        });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });
});

```text



| Store                 | Package                           | Use                                  |
| --------------------- | --------------------------------- | ---------------------------------------- |
| Entity Framework Core | `MassTransit.EntityFrameworkCore` | Already  EF Core; need transactions |
| MongoDB               | `MassTransit.MongoDb`             | Document-oriented state; high throughput |
| Redis                 | `MassTransit.Redis`               | Ephemeral sagas; low latency             |
| In-Memory             | Built-                          | Testing only -- state lost  restart    |



When a saga step fails, publish compensating commands to undo prior steps:

```bash

OrderSubmitted -> RequestPayment -> PaymentReceived -> ReserveInventory
                                                          |
                                                     InventoryFailed
                                                          |
                                                    RefundPayment (compensation)
                                                          |
                                                    CancelOrder (compensation)

```text

---



At-least-once delivery means consumers may receive the same message multiple times. Idempotent consumers ensure repeated
processing produces the same result.



```
{
    
    {
         messageId = context.MessageId
            ??   InvalidOperationException();

        
         exists =  db.ProcessedMessages
            .AnyAsync(m => m.MessageId == messageId);

         (exists)
        {
            logger.LogInformation(
                , messageId);
            ;
        }

        
         ProcessOrderAsync(context.Message);

        
        db.ProcessedMessages.Add( ProcessedMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow,
            ConsumerType = (IdempotentOrderConsumer)
        });

         db.SaveChangesAsync();
    }
}

```text



Prefer operations that are naturally idempotent:

- **Upserts** (`INSERT ... ON CONFLICT UPDATE`) instead of blind inserts
- **Conditional updates** (`UPDATE ... WHERE Status = `) instead of unconditional
- **Deterministic IDs** derived  message content instead of auto-generated

---



Wrap message payloads  a standard envelope  metadata  tracing, versioning,  routing.

```;

```text

MassTransit provides  automatically via `ConsumeContext` (MessageId, CorrelationId, Headers). When  raw broker
clients, implement envelopes explicitly.

---



 **Do  use auto-complete  Azure Service Bus** --  `AutoCompleteMessages = `  call
   `CompleteMessageAsync` after successful processing. Auto-complete acknowledges before processing finishes, risking
   data loss  failure.
 **Do  forget to handle poison messages** -- always configure max delivery count  DLQ monitoring. Without these,
   a single bad message blocks the entire queue indefinitely.
 **Do  use -memory saga persistence  production** -- saga state  lost  restart, leaving business processes
    unknown states. Use Entity Framework, MongoDB,  Redis persistence.
 **Do  assume message ordering across partitions** -- competing consumers  topic subscriptions deliver messages
    of order  . Use sessions  partitioning  order matters.
 **Do  skip idempotency  at-least-once consumers** -- brokers may redeliver  timeout, network glitch, 
   consumer restart. Every consumer must handle duplicate messages safely.
 **Do  hardcode connection strings** -- use environment variables  Azure Key Vault references. For local
   development, use user secrets  `.env` files excluded  source control.

---



- [Azure Service Bus documentation](https:
- [Azure Service Bus client library  .NET](https:
- [RabbitMQ .NET client documentation](https:
- [MassTransit documentation](https:
- [MassTransit sagas](https:
- [Enterprise Integration Patterns](https:

Code Navigation (Serena MCP)

Primary approach: Use Serena symbol operations for efficient code navigation:

1. Find definitions: serena_find_symbol instead of text search
2. Understand structure: serena_get_symbols_overview for file organization
3. Track references: serena_find_referencing_symbols for impact analysis
4. Precise edits: serena_replace_symbol_body for clean modifications

When to use Serena vs traditional tools:

• Use Serena: Navigation, refactoring, dependency analysis, precise edits
• Use Read/Grep: Reading full files, pattern matching, simple text operations
• Fallback: If Serena unavailable, traditional tools work fine

Example workflow:

# Instead of:
Read: src/Services/OrderService.cs
Grep: "public void ProcessOrder"

# Use:
serena_find_symbol: "OrderService/ProcessOrder"
serena_get_symbols_overview: "src/Services/OrderService.cs"