Home page
Back to articles
20 min readMehdi Hadeli

MassTransit Transactional Messaging with .NET Aspire, RabbitMQ, Kafka, PostgreSQL, MongoDB, Outbox, Inbox, Retry, Error Queues, and Dead-Letter Handling

dotnet-coreaspiremasstransitrabbitmqkafkamicroservicesmessaging

On this page

Table of contents

Introduction

Reliable messaging usually fails in the least interesting line of code.

The domain write succeeds, but the process crashes before the event leaves the service. A broker redelivers a message, and the consumer applies the same change twice. An internal post-commit projection needs retries and durability, so the team starts building an outbox table, duplicate detection, retry policies, error queues, and background delivery workers before the actual business feature is even finished.

This sample shows a practical way to solve the same problems with MassTransit and .NET Aspire. The design stays close to the Wolverine version: two services, PostgreSQL for transactional state, MongoDB for a read model, and a broker selected through configuration. The service shape is familiar, but the messaging model is different. In the MassTransit sample, the key building blocks are the Entity Framework bus outbox, consumer outbox and inbox state, transport-specific RabbitMQ or Kafka registration, and explicit retry and error-transport configuration.

Sample code: masstransit-transactional-messaging-aspire

Objectives

By the end of this article, you should be able to:

  1. Model a two-service sample around transactional writes, integration events, and post-commit internal processing.
  2. Configure MassTransit with PostgreSQL-backed Entity Framework outbox state.
  3. Publish broker messages through the bus outbox inside the same EF Core transaction as business data.
  4. Use MassTransit consumer outbox and inbox state to make broker consumption idempotent and transactional.
  5. Process internal post-commit work for a MongoDB read model without publishing from an uncommitted transaction.
  6. Configure retry and redelivery behavior centrally, while overriding Kafka-specific behavior when needed.
  7. Understand how MassTransit error queues differ from broker-native dead-letter queues and topics.
  8. Keep transport-specific code thin enough to switch between RabbitMQ and Kafka without forking the application design.

Sample Overview

The sample contains two microservices.

Catalogs is the write-side service.

  • It stores product aggregates in PostgreSQL.
  • It publishes MessageEnvelope<ProductCreatedV1> through IEventBus.
  • It sends ProjectProductReadModel through IInternalCommandBus after commit so MongoDB read model updates only after PostgreSQL transaction succeeds.

Orders is the downstream consumer.

  • It listens to same integration event from RabbitMQ queue or Kafka topic, based on configuration.
  • It uses MassTransit consumer outbox and inbox state backed by PostgreSQL on the RabbitMQ path.
  • It applies centralized retry and error-transport behavior, and the sample keeps Kafka-specific overrides explicit where transport features differ.
  • It stores an imported product record in its own PostgreSQL database.

The reusable pieces follow the same high-level split as the Wolverine sample:

  • src/Services/Shared contains contracts, messaging constants, and MessageEnvelope<T>.
  • src/BuildingBlocks/BuildingBlocks.Integration.MassTransit* contains integration options, transport registration, outbox configuration, and producer helpers.
  • src/BuildingBlocks/BuildingBlocks.Persistence.* contains PostgreSQL and MongoDB setup.
  • src/Aspire/ECommerce.AppHost contains Aspire orchestration.

Architecture Overview

The AppHost provisions PostgreSQL, MongoDB, and a broker. Both APIs receive PostgreSQL, MongoDB where needed, and broker connection details through Aspire resource references.

There are three distinct messaging concerns in the sample:

  1. External broker publishing from Catalogs.
  2. Internal post-commit processing inside Catalogs.
  3. Transactional broker consumption in Orders.

MassTransit covers those concerns with two related but different mechanisms:

  • Bus outbox for publish and send operations started inside an EF Core transaction.
  • Consumer outbox for message handling that must update database and emit more messages once per successful consume, with inbox state used for duplicate detection.

For internal processing, MassTransit does not have Wolverine's built-in durable local queue. The sample fills that gap with BuildingBlocks.DurableLocalQueue — a custom background worker that polls a DurableMessage table in the same PostgreSQL database. That preserves the same ordering rule: do not update the MongoDB read model before the PostgreSQL transaction commits. The durable queue also survives process restarts, unlike an in-memory mediator approach.

Rendering diagram...

MassTransit Feature Setup

The sample builds on seven MassTransit capabilities. Here is how each one maps to the problem.

1. Entity Framework transactional outbox with PostgreSQL

MassTransit stores outgoing messages and consumer delivery state in relational tables so business writes and message dispatch can share a transaction boundary.

At the EF Core level, add MassTransit transactional outbox entities to your DbContext model:

using MassTransit;
using Microsoft.EntityFrameworkCore;

public class CatalogsDbContext : DbContext
{
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);

        modelBuilder.AddTransactionalOutboxEntities();
    }
}

Then register the Entity Framework outbox and enable the bus outbox. The latest sample wraps that in a shared registration layer and drives it with MassTransitOptions:

var options = new MassTransitOptions
{
    DurableStorageConnectionString = connectionString,
    RabbitMqConnectionString = builder.Configuration.GetConnectionString("rabbitmq"),
    KafkaConnectionString = builder.Configuration.GetConnectionString("kafka"),
    Bus = new MassTransitBusOptions
    {
        UseBusOutbox = true,
        UsePostCommitMediator = true,
    },
};

builder.Services.AddMassTransitRabbitMq<CatalogsDbContext>(
    options,
    new MassTransitRabbitMqRegistrationOptions
    {
        ConfigureBus = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
        ConfigureMediator = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
    },
    builder.Environment);

Sample paths: src/BuildingBlocks/BuildingBlocks.Integration.MassTransit/Extensions/MassTransitServiceCollectionExtensions.cs, src/BuildingBlocks/BuildingBlocks.Integration.MassTransit.RabbitMQ/MassTransitRabbitMqExtensions.cs

What this gives you:

  • outgoing publish and send operations are written to PostgreSQL first
  • messages are dispatched only after the transaction commits
  • the application does not publish directly from an uncommitted write-side transaction

At the lower layer, the shared registration code calls AddEntityFrameworkOutbox<TDbContext>(o => { o.UsePostgres(); o.UseBusOutbox(); }) only when options.Bus.UseBusOutbox is enabled.

2. Bus outbox for transactional publishing in Catalogs

The MassTransit bus outbox is the publisher-side equivalent of a transactional outbox. In the sample, the HTTP endpoint saves the aggregate and calls IEventBus. The RabbitMQ path uses IPublishEndpoint through MassTransitEnvelopePublisher. The Kafka path swaps the publisher implementation through AddKafkaMessagePublisher<MessageEnvelope<ProductCreatedV1>>(). With the bus outbox enabled, MassTransit captures the publish operation in outbox tables instead of sending it immediately.

Minimal flow:

app.MapPost("/api/v1/products", async (
    CreateProductRequest request,
    CatalogsDbContext dbContext,
    IEventBus eventBus,
    IInternalCommandBus internalCommandBus,
    CancellationToken cancellationToken) =>
{
    var product = Product.Create(request.Code, request.Name, request.Price);

    dbContext.Products.Add(product);

    var integrationEvent = MessageEnvelope.Create(
        new ProductCreatedV1(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc)
    );

    await eventBus.PublishAsync(integrationEvent, cancellationToken);

    await dbContext.SaveChangesAsync(cancellationToken);

    await internalCommandBus.SendAsync(
        new ProjectProductReadModel(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc),
        cancellationToken);

    return TypedResults.CreatedAtRoute(
        new CreateProductResponse(product.Id, product.Code, product.Name, product.Price),
        "CreateProduct",
        new { id = product.Id });
});

Sample path: src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/CreatingProduct/v1/CreateProductEndpoint.cs

The important part is not the syntax. The important part is the ordering guarantee:

  • SaveChangesAsync commits both business data and outbox records.
  • MassTransit delivers the broker message after that commit succeeds.
  • If the transaction fails, neither the row nor the outgoing event is published.

3. Consumer outbox and inbox state in Orders

For consumers, MassTransit uses the Entity Framework outbox differently. When the sample applies UseEntityFrameworkOutbox<OrdersDbContext>(context) on the endpoint, MassTransit wraps the consumer in a database transaction, tracks inbox state for duplicate detection, stores outgoing messages in outbox state, and commits only after the consumer succeeds.

RabbitMQ endpoint example:

cfg.ReceiveEndpointWithPolicies<OrdersDbContext, ProductCreatedConsumer>(
    context,
    options.Bus,
    MessagingConstants.ProductCreatedQueue);

This is the closest MassTransit equivalent to a transactional inbox plus outbox pipeline.

What it buys you:

  • duplicate broker deliveries are tracked in inbox state
  • the imported product write and any follow-up publish/send happen in one transaction scope
  • retries do not duplicate outgoing messages after a successful commit

This is stronger and more precise than saying "exactly once" at the transport level. The broker still delivers at least once. MassTransit uses inbox state and transaction boundaries so your consumer side effects stay idempotent and transactional.

4. RabbitMQ transport

RabbitMQ is the simpler transport path. The latest sample uses shared helper methods that keep Catalogs and Orders thin while centralizing endpoint policy setup.

Minimal setup:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ProductCreatedConsumer, ProductCreatedConsumerDefinition>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("rabbitmq"));

        cfg.Message<MessageEnvelope<ProductCreatedV1>>(m =>
        {
            m.SetEntityName("catalogs-products-created");
        });

        cfg.ReceiveEndpoint("catalogs-products-created", endpoint =>
        {
            endpoint.ApplyEndpointPolicies<OrdersDbContext>(context, options.Bus);
            endpoint.ConfigureConsumer<ProductCreatedConsumer>(context);
        });
    });
});

Sample paths: src/BuildingBlocks/BuildingBlocks.Integration.MassTransit.RabbitMQ/MassTransitRabbitMqExtensions.cs, src/Services/Orders/ECommerce.Services.Orders/ApplicationConfiguration.cs

In the actual sample, the RabbitMQ producer sets the MessageEnvelope<ProductCreatedV1> entity name to catalogs-products-created. The consumer side uses an explicit receive endpoint with the same queue name. The endpoint also configures broker dead-letter arguments and binds orders-products-dlq explicitly. MassTransit error transport remains a separate concern from the broker-native DLQ path.

5. Kafka transport with Rider

MassTransit handles Kafka through the Rider API. Publishing uses ITopicProducer<T>, and consumption uses TopicEndpoint with a consumer group. The sample still uses UsingInMemory for the base bus while the Kafka rider handles broker traffic.

Minimal setup:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ProductCreatedConsumer, ProductCreatedConsumerDefinition>();

    x.UsingInMemory((_, _) => { });

    x.AddRider(rider =>
    {
        rider.AddConsumer<ProductCreatedConsumer>();

        rider.AddProducer<MessageEnvelope<ProductCreatedV1>>("catalogs-products-created");

        rider.UsingKafka((context, kafka) =>
        {
            kafka.Host(builder.Configuration["ConnectionStrings:kafka"]);

            kafka.TopicEndpoint<MessageEnvelope<ProductCreatedV1>>(
                "catalogs-products-created",
                "orders-products",
                endpoint =>
                {
                    endpoint.ApplyEndpointPolicies<OrdersDbContext>(context, kafkaBusOptions);
                    endpoint.ConfigureConsumer<ProductCreatedConsumer>(context);
                });
        });
    });
});

And publishing to Kafka:

public static class MassTransitKafkaPublisherRegistrationExtensions
{
    public static IServiceCollection AddKafkaMessagePublisher<TMessage>(
        this IServiceCollection services)
        where TMessage : class
    {
        services.AddScoped<IMassTransitMessagePublisher, KafkaTopicMessagePublisher<TMessage>>();
        return services;
    }
}

internal sealed class KafkaTopicMessagePublisher<TMessage>(ITopicProducer<TMessage> producer)
    : IMassTransitMessagePublisher
    where TMessage : class
{
    public Task PublishAsync<TRequestedMessage>(
        TRequestedMessage message,
        CancellationToken cancellationToken)
        where TRequestedMessage : class
    {
        if (message is not TMessage typedMessage)
        {
            throw new InvalidOperationException(
                $"Kafka publisher does not support message type '{typeof(TRequestedMessage).Name}'. Expected '{typeof(TMessage).Name}'.");
        }

        return producer.Produce(typedMessage, cancellationToken);
    }
}

Sample paths: src/BuildingBlocks/BuildingBlocks.Integration.MassTransit.Kafka/MassTransitKafkaExtensions.cs, src/Services/Orders/ECommerce.Services.Orders/ApplicationConfiguration.cs, and src/Services/Catalogs/ECommerce.Services.Catalogs/ApplicationConfiguration.cs

Kafka changes the transport wiring, but not the transactional design on the publisher side. The write model still commits before the event is dispatched. On the consumer side, this sample keeps Kafka-specific behavior explicit by disabling consumer outbox and delayed redelivery in kafkaBusOptions, while still using retry and a dedicated dead-letter forwarding consumer.

6. Retry and delayed redelivery

MassTransit separates immediate retry from delayed redelivery.

  • UseMessageRetry handles short transient failures in-memory while the message is being consumed.
  • UseDelayedRedelivery re-enqueues the message for a later attempt.
  • If the consumer still fails after those policies are exhausted, the message moves to the error transport.

Shared policy example:

public static void ApplyEndpointPolicies<TDbContext>(
    this IReceiveEndpointConfigurator endpoint,
    IRegistrationContext context,
    MassTransitBusOptions busOptions)
    where TDbContext : DbContext
{
    var immediateRetries = busOptions.Retry.MaximumAttempts - 1;
    if (immediateRetries > 0)
    {
        endpoint.UseMessageRetry(r => r.Immediate(immediateRetries));
    }

    if (busOptions.Retry.UseDelayedRedelivery)
    {
        endpoint.UseDelayedRedelivery(r =>
            r.Intervals(busOptions.Retry.DelayedRedeliveryIntervals));
    }

    if (busOptions.UseConsumerOutbox)
    {
        endpoint.UseEntityFrameworkOutbox<TDbContext>(context);
    }
}

That gives you a clear operational flow:

  1. try a few immediate retries for short-lived failures
  2. schedule redelivery for longer transient problems
  3. move the message to the configured error transport if it still cannot be processed

The sample applies that shared policy to both transports, but the Kafka path overrides two settings intentionally:

  • UseConsumerOutbox = false
  • Retry.UseDelayedRedelivery = false

That keeps the Kafka sample aligned with the current rider behavior and avoids treating both transports as operationally identical.

7. Error queues versus broker-native dead-letter queues

This distinction matters because Wolverine and MassTransit present it differently.

MassTransit has a built-in error transport. In the sample, Orders enables it with UseErrorTransport = true. When a message faults and retry plus redelivery are exhausted, it moves to an error endpoint such as orders-products_error for a queue-based transport. That is MassTransit behavior, not the same thing as a RabbitMQ dead-letter exchange or a Kafka dead-letter topic.

For most applications, MassTransit error queues are enough because they are consistent across transports and easy to inspect. If you specifically need broker-native dead-letter topology for RabbitMQ or a Kafka dead-letter topic strategy, treat that as an extra transport concern layered on top of the MassTransit defaults.

The sample now shows both paths explicitly in configuration:

  • MassTransit default: retries plus MassTransit error transport
  • broker-native strategy: RabbitMQ DLX and DLQ wiring, plus Kafka dead-letter topic forwarding

Do not conflate those two.

RabbitMQ dead-letter support is configured on the receive queue with x-dead-letter-exchange and x-dead-letter-routing-key, then bound to orders-products-dlq.

Kafka dead-letter support is modeled differently. The sample listens for Fault<MessageEnvelope<ProductCreatedV1>> on the Kafka path and forwards the original envelope to orders-products-dead-letter. That keeps the broker-specific dead-letter story explicit instead of implying Kafka gives you the same queue semantics as RabbitMQ.

Operational comparison: _error versus DLQ and DLT

At runtime, these paths answer different operational questions.

  • orders-products_error tells you MassTransit gave up after retry and redelivery and moved the failed message to its error transport.
  • orders-products-dlq tells you RabbitMQ dead-letter topology captured the message at the broker layer.
  • orders-products-dead-letter tells you the Kafka flow forwarded a faulted message into an explicit dead-letter topic strategy.

Use the MassTransit _error endpoint when you want transport-agnostic failure handling and a consistent place to inspect failed consumes across transports. Use a broker-native DLQ or DLT when your operators already work in RabbitMQ or Kafka tooling and want the failed-message story to live in broker topology.

In this sample, that usually means:

  • inspect orders-products_error first when validating MassTransit policy behavior
  • inspect orders-products-dlq when validating RabbitMQ DLX and DLQ wiring
  • inspect orders-products-dead-letter when validating Kafka dead-letter forwarding

That split keeps policy concerns and broker concerns visible instead of mixing them into one failure bucket.

AppHost Infrastructure

The AppHost setup stays very close to the Wolverine sample.

It should:

  1. Provision one PostgreSQL server and create catalogsdb and ordersdb.
  2. Provision one MongoDB server and create the read-model database.
  3. Provision either RabbitMQ or Kafka based on Messaging__Transport.
  4. Pass the selected transport and connection details to both APIs.

The transport switch stays small:

var transport =
    builder.Configuration["Messaging:Transport"]?.Trim().ToLowerInvariant() ?? "rabbitmq";

switch (transport)
{
    case "rabbitmq":
        // add rabbitmq resource
        break;
    case "kafka":
        // add kafka resource
        break;
}

That keeps orchestration transport-aware without forking the business design.

Shared Messaging Contracts

The shared contracts stay close to the earlier sample shape, but the latest sample adds richer envelope metadata.

The key integration event is still ProductCreatedV1, and the envelope stays useful:

public sealed record MessageEnvelope<TMessage>(
    Guid MessageId,
    Guid CorrelationId,
    DateTime OccurredAtUtc,
    TMessage Message)
    where TMessage : IMessage;

Factory method keeps call sites small:

var integrationEvent = MessageEnvelope.Create(
    new ProductCreatedV1(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc)
);

That envelope gives the sample stable metadata for correlation, observability, and test fixtures without duplicating headers into every event contract.

Transport names stay centralized in MessagingConstants:

  • catalogs-products-created queue name for RabbitMQ receive endpoint and topic/entity name for publish topology
  • catalogs-products-created topic name for Kafka
  • orders-products consumer group name for Kafka
  • orders-products_error MassTransit error queue name when queue-based transport faults exhaust policies
  • orders-products-dlq RabbitMQ dead-letter queue name
  • orders-products-dead-letter Kafka dead-letter topic name

Configuring MassTransit in Catalogs

The write-side MassTransit configuration has three responsibilities:

  1. register the EF outbox with PostgreSQL
  2. choose RabbitMQ or Kafka
  3. expose a small publishing abstraction to the application layer

RabbitMQ-oriented setup in latest sample:

builder.Services.AddMassTransitRabbitMq<CatalogsDbContext>(
    options,
    new MassTransitRabbitMqRegistrationOptions
    {
        ConfigureBus = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
        ConfigureMediator = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
        ConfigureTransport = (_, cfg) =>
        {
            cfg.PublishToRabbitQueue<MessageEnvelope<ProductCreatedV1>>(
                MessagingConstants.ProductCreatedQueue);
        },
    },
    builder.Environment);

Kafka-oriented setup in latest sample:

builder.Services.AddMassTransitKafka<CatalogsDbContext>(
    options,
    new MassTransitKafkaRegistrationOptions
    {
        ConfigureBus = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
        ConfigureMediator = x => x.AddConsumer<ProjectProductReadModelConsumer>(),
        ConfigureRider = rider =>
        {
            rider.AddProducer<MessageEnvelope<ProductCreatedV1>>(
                MessagingConstants.ProductCreatedTopic);
        },
        ConfigurePublisher = services =>
        {
            services.AddKafkaMessagePublisher<MessageEnvelope<ProductCreatedV1>>();
        },
    },
    builder.Environment);

The application code should not care which transport is active. It depends on IEventBus, backed by IPublishEndpoint for RabbitMQ or a Kafka-specific publisher implementation for ITopicProducer<T>.

Transactional Outbox in the Create Product Slice

The business flow in CreateProductEndpoint is still the center of the sample.

A sample-backed version looks like this:

app.MapPost("/api/v1/products", async (
    CreateProductRequest request,
    CatalogsDbContext dbContext,
    IEventBus eventBus,
    IInternalCommandBus internalCommandBus,
    CancellationToken cancellationToken) =>
{
    var product = Product.Create(request.Code, request.Name, request.Price);

    dbContext.Products.Add(product);

    var integrationEvent = MessageEnvelope.Create(
        new ProductCreatedV1(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc)
    );

    await eventBus.PublishAsync(integrationEvent, cancellationToken);

    await dbContext.SaveChangesAsync(cancellationToken);

    await internalCommandBus.SendAsync(
        new ProjectProductReadModel(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc),
        cancellationToken);

    return TypedResults.CreatedAtRoute(
        new CreateProductResponse(product.Id, product.Code, product.Name, product.Price),
        "CreateProduct",
        new { id = product.Id });
});

The critical rule is the same as before:

  • do not publish directly to the broker outside a transaction-aware outbox
  • do not trigger MongoDB projection before the PostgreSQL commit succeeds

The only real difference is which MassTransit-backed publisher implementation handles the outbound message.

Internal Post-Commit Processing for MongoDB Read Models

This is the one area where MassTransit is not a one-for-one replacement for Wolverine.

Wolverine has durable local queues built in. MassTransit does not provide the same local durable queue abstraction as a first-class equivalent. Because of that, you need to choose one of three designs:

  1. Use a post-commit in-process mediator for simple internal commands.
  2. Use a dedicated durable queue or topic and consume it through MassTransit like any other message flow.
  3. Use a background projector that reacts to the same integration event and updates MongoDB asynchronously.

The latest sample chooses option 1 and enables it explicitly through MassTransitBusOptions.UsePostCommitMediator = true. That gives you post-commit ordering, but not restart-safe durability.

Minimal mediator setup:

if (options.Bus.UsePostCommitMediator)
{
    services.AddMediator(x =>
    {
        configureMediator?.Invoke(x);
    });
}

Trigger after the write transaction succeeds:

await dbContext.SaveChangesAsync(cancellationToken);
await internalCommandBus.SendAsync(
    new ProjectProductReadModel(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc),
    cancellationToken);

And the consumer stays small:

public sealed class ProjectProductReadModelConsumer : IConsumer<ProjectProductReadModel>
{
    private readonly IProductReadRepository _repository;

    public ProjectProductReadModelConsumer(IProductReadRepository repository)
    {
        _repository = repository;
    }

    public Task Consume(ConsumeContext<ProjectProductReadModel> context)
    {
        var message = context.Message;

        return _repository.UpsertAsync(
            new ProductReadModel(message.ProductId, message.Code, message.Name, message.Price, message.CreatedAtUtc),
            context.CancellationToken);
    }
}

Sample paths: src/Services/Catalogs/ECommerce.Services.Catalogs/ApplicationConfiguration.cs, src/BuildingBlocks/BuildingBlocks.Integration.MassTransit/Extensions/MassTransitServiceCollectionExtensions.cs, src/BuildingBlocks/BuildingBlocks.Integration.MassTransit/MassTransitInternalCommandBus.cs, and src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/ProjectingProductReadModel/v1/ProjectProductReadModelConsumer.cs

If you need restart-safe durability for that internal work, prefer option 2 instead: route the command through a real MassTransit receive endpoint backed by RabbitMQ or Kafka, or persist your own internal work table and process it separately.

That tradeoff should be explicit in the article, because otherwise readers will assume MassTransit has the same local durability feature as Wolverine.

Transactional Consumption in Orders

The consumer-side MassTransit configuration lives in Orders and focuses on one policy combination:

  • retry and redelivery for transient failures
  • Entity Framework consumer outbox for inbox state and transactional side effects

Consumer example:

public class ProductCreatedConsumer : IConsumer<MessageEnvelope<ProductCreatedV1>>
{
    private readonly OrdersDbContext _dbContext;

    public async Task Consume(ConsumeContext<MessageEnvelope<ProductCreatedV1>> context)
    {
        var message = context.Message.Message;

        if (string.Equals(message.Code, "faulty-product-created", StringComparison.OrdinalIgnoreCase))
        {
            throw new InvalidOperationException(
                "Intentional consumer failure for retry and dead-letter tests.");
        }

        var imported = await _dbContext.ImportedProducts.FindAsync(
            new object[] { message.ProductId },
            context.CancellationToken);

        if (imported is null)
        {
            imported = new ImportedProduct(message.ProductId);
            _dbContext.ImportedProducts.Add(imported);
        }

        imported.Name = message.Name;
        imported.Price = message.Price;
    }
}

With UseEntityFrameworkOutbox<OrdersDbContext>(context) on the RabbitMQ endpoint, MassTransit handles the transaction and inbox state around the consumer. The consumer itself stays focused on business logic.

Validation

The current test suite proves the important parts of the sample, but it no longer tries to scrape every broker artifact end to end.

The strongest coverage is now in five places:

  1. CreateProductTests posts to /api/v1/catalogs/products, verifies the PostgreSQL write model, then polls MongoDB until the read model appears.
  2. CreateProductTests also waits for the PostgreSQL OutboxMessage table to drain, proving the bus outbox delivered the integration event.
  3. ErrorQueueTests resolves the real ProductCreatedConsumer from DI, invokes it with a mocked ConsumeContext, asserts the expected failure, and verifies no imported product row is persisted for the faulty message.
  4. KafkaErrorQueueTests verifies the same failure semantics for ProductCreatedConsumer, then tests KafkaDeadLetterForwardingConsumer directly by asserting it forwards the original envelope to ITopicProducer<MessageEnvelope<ProductCreatedV1>>.
  5. Unit tests cover ProductCreatedConsumer and ProjectProductReadModelConsumer so the write-side and read-model update logic stay small and focused.

The Catalogs integration test is the main end-to-end flow test:

[Fact]
public async Task PostProduct_ShouldCreateWriteAndReadModels_AndPublishEvent()
{
    var cancellationToken = TestContext.Current.CancellationToken;
    using var client = Factory.CreateClient();

    var request = new
    {
        code = "catalog-101",
        name = "Test Basket",
        price = 15.25m,
    };

    var response = await client.PostAsJsonAsync(
        "/api/v1/catalogs/products",
        request,
        cancellationToken);

    Assert.Equal(HttpStatusCode.Created, response.StatusCode);

    var published = await WaitForOutboxToDrainAsync(
        TimeSpan.FromSeconds(30),
        cancellationToken);

    Assert.True(published);
}

The Orders failure tests are now narrower and more deterministic. Instead of depending on broker inspection from the test host, they verify the consumer failure semantics and the Kafka forwarding logic directly.

That distinction still matters because the runtime configuration still supports different operational paths even though the automated tests now focus on application behavior:

  • RabbitMQ: the sample configures MassTransit error transport plus RabbitMQ DLX and orders-products-dlq
  • Kafka: the sample configures explicit dead-letter forwarding through KafkaDeadLetterForwardingConsumer
  • Orders tests: the sample proves consumer failure semantics and non-persistence for faulty messages
  • MongoDB projection: the sample proves the read model appears after the create flow completes

Transport Switching Strategy

The cleanest design is to isolate transport differences at the integration edge.

Shared application concerns:

  • contracts
  • envelope
  • business endpoints
  • consumer logic
  • EF outbox configuration
  • retry and outbox policies

Transport-specific concerns:

  • RabbitMQ bus host and exchange or queue topology
  • Kafka rider registration, producers, and topic endpoints
  • any broker-native dead-letter customization

That means the switch can stay narrow:

  • Messaging:Transport=rabbitmq uses UsingRabbitMq
  • Messaging:Transport=kafka uses AddRider(...UsingKafka...)

Everything above that layer stays the same.

What Changes Relative to Wolverine

This comparison is the part readers usually need most.

ConcernWolverine sampleMassTransit version
Transactional publishdurable outboxEF bus outbox
Consumer idempotencydurable inboxconsumer outbox plus inbox state
Internal post-commit workdurable local queuesmediator, dedicated durable endpoint, or background projector
Retryretry policy on listenersUseMessageRetry and UseDelayedRedelivery
Failed-message destinationerror queue or native DLQ depending on transportMassTransit error transport by default, optional RabbitMQ DLQ and Kafka dead-letter topic
Kafka integrationnative transport APIRider

The most important difference is internal durable processing. If you need restart-safe local processing inside the same service without an external broker, Wolverine has the more direct feature. With MassTransit, you either accept a lighter post-commit in-process model or make the internal work a real durable messaging flow.

That is not a weakness in itself. It is simply a different design center.

Prerequisites

You need:

  1. .NET 10 SDK
  2. Docker Desktop or another supported OCI runtime
  3. A recent .NET Aspire workload and tooling setup
  4. MassTransit packages for chosen transport and EF Core outbox support

Typical package set:

<PackageReference Include="MassTransit" Version="8.*" />
<PackageReference Include="MassTransit.EntityFrameworkCore" Version="8.*" />
<PackageReference Include="MassTransit.RabbitMQ" Version="8.*" />
<PackageReference Include="MassTransit.Kafka" Version="8.*" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.*" />

Conclusion

MassTransit can solve the same broad problem as the Wolverine sample: transactional publishing, idempotent consumption, retries, and operationally visible failure handling across RabbitMQ or Kafka.

The design is close, but not identical.

Use the EF bus outbox when publishing from the write side. Use the consumer outbox when a consumer updates PostgreSQL and may publish more messages. Treat retry and delayed redelivery as first-class policies. Treat MassTransit error queues as the default failed-message story, and add broker-native dead-letter topology only when you have a concrete operational reason.

For internal post-commit processing, be explicit. If you only need post-commit ordering, mediator can be enough. If you need durability across restarts, push that work through a real durable endpoint instead of assuming a Wolverine-style local queue exists.

That clarity is what keeps the architecture honest.