Home page
Back to articles
25 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:

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, the sample maps the MassTransit outbox directly inside the write-side DbContext:

src/Services/Catalogs/ECommerce.Services.Catalogs/Shared/Data/CatalogsDbContext.cs
using BuildingBlocks.DurableLocalQueue;
using ECommerce.Services.Catalogs.Products.Models;
using MassTransit;
using Microsoft.EntityFrameworkCore;
 
namespace ECommerce.Services.Catalogs.Shared.Data;
 
public class CatalogsDbContext(DbContextOptions<CatalogsDbContext> options) : DbContext(options)
{
    public DbSet<Product> Products => Set<Product>();
 
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);
        modelBuilder.AddTransactionalOutboxEntities();
        modelBuilder.AddDurableLocalQueueEntities();
    }
}

Focus here:

  • AddTransactionalOutboxEntities() is the MassTransit part: it adds relational outbox and delivery-state tables.
  • AddDurableLocalQueueEntities() is the sample-specific addition that preserves restart-safe internal processing for MongoDB projection.

Then register the shared MassTransit layer with the real sample options object. The sample does not wire AddEntityFrameworkOutbox(...) ad hoc at every service call site; it centralizes that in MassTransitServiceCollectionExtensions.cs:

src/BuildingBlocks/BuildingBlocks.Integration.MassTransit/Extensions/MassTransitServiceCollectionExtensions.cs
public static class MassTransitServiceCollectionExtensions
{
    public static IServiceCollection AddMassTransitMessaging<TDbContext>(
        this IServiceCollection services,
        MassTransitOptions options,
        Action<IBusRegistrationConfigurator> configure,
        Action<IMediatorRegistrationConfigurator>? configureMediator = null,
        Action<IServiceCollection>? configureServices = null,
        IHostEnvironment? environment = null
    )
        where TDbContext : DbContext
    {
        if (options.Bus.UsePostCommitMediator)
        {
            services.AddMediator(x =>
            {
                configureMediator?.Invoke(x);
            });
        }
 
        if (environment?.IsEnvironment("Test") == true)
        {
            services.AddMassTransitTestHarness(cfg =>
            {
                if (options.Bus.UseBusOutbox)
                {
                    cfg.AddEntityFrameworkOutbox<TDbContext>(o =>
                    {
                        o.UsePostgres();
                        o.UseBusOutbox();
                    });
                }
 
                configure(cfg);
            });
        }
        else
        {
            services.AddMassTransit(x =>
            {
                if (options.Bus.UseBusOutbox)
                {
                    x.AddEntityFrameworkOutbox<TDbContext>(o =>
                    {
                        o.UsePostgres();
                        o.UseBusOutbox();
                    });
                }
 
                configure(x);
            });
        }
 
        if (options.Bus.UseEnvelopePublisher)
        {
            services.AddScoped<IMassTransitMessagePublisher, MassTransitEnvelopePublisher>();
        }
        else
        {
            services.AddScoped<IMassTransitMessagePublisher, MassTransitPublishEndpointPublisher>();
        }

Focus here:

  • lines 18-27 decide whether post-commit mediator support is even registered.
  • lines 31-39 and 44-52 show the same outbox behavior applied consistently in both test and runtime registrations.
  • lines 54-64 decide which publishing abstraction the application layer gets, so feature code does not branch on transport.

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.

The actual sample-backed endpoint is CreateProductEndpoint.cs:

src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/CreatingProduct/v1/CreateProductEndpoint.cs
using BuildingBlocks.Integration.MassTransit.Abstractions;
using ECommerce.Services.Catalogs.Products.Models;
using ECommerce.Services.Catalogs.Shared.Data;
using ECommerce.Services.Shared.Contracts.IntegrationEvents;
using ECommerce.Services.Shared.Contracts.InternalCommands;
using ECommerce.Services.Shared.Contracts.MessageEnvelope;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Http.HttpResults;
using Microsoft.AspNetCore.Mvc;
using Microsoft.AspNetCore.Routing;
 
namespace ECommerce.Services.Catalogs.Products.Features.CreatingProduct.v1;
 
internal static class CreateProductEndpoint
{
    internal static RouteHandlerBuilder MapCreateProductEndpoint(
        this IEndpointRouteBuilder endpoints
    )
    {
        return endpoints.MapPost("/products", Handle).WithName("CreateProduct");
    }
 
    private static async Task<CreatedAtRoute<CreateProductResponse>> Handle(
        [FromBody] 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.EnqueueAsync(
            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 }
        );
    }
}

Focus here:

  • lines 15-19 keep the endpoint shape minimal: one route, one handler.
  • lines 21-28 create the aggregate and build the integration-event envelope inside the same HTTP flow.
  • lines 30-40 are the transactional messaging core: publish through IEventBus, commit with SaveChangesAsync, then enqueue internal work.
  • lines 42-50 show that the internal MongoDB projection path is no longer mediator SendAsync; the current sample uses durable EnqueueAsync.

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:

src/Services/Orders/ECommerce.Services.Orders/ApplicationConfiguration.cs
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.

The shared RabbitMQ helper stays small, and the service configuration decides queue topology and DLQ wiring:

src/BuildingBlocks/BuildingBlocks.Integration.MassTransit.RabbitMQ/MassTransitRabbitMqExtensions.cs
public static IServiceCollection AddMassTransitRabbitMq<TDbContext>(
    this IServiceCollection services,
    MassTransitOptions options,
    MassTransitRabbitMqRegistrationOptions? registrationOptions = null,
    IHostEnvironment? environment = null
)
    where TDbContext : DbContext
{
    registrationOptions ??= new MassTransitRabbitMqRegistrationOptions();
 
    services.AddMassTransitMessaging<TDbContext>(
        options,
        x =>
        {
            registrationOptions.ConfigureBus?.Invoke(x);
            x.UsingRabbitMq(
                (context, cfg) =>
                {
                    cfg.Host(
                        new Uri(
                            options.RabbitMqConnectionString
                                ?? throw new InvalidOperationException(
                                    "Missing connection string 'rabbitmq'."
                                )
                        )
                    );
                    registrationOptions.ConfigureTransport?.Invoke(context, cfg);
                    cfg.ConfigureEndpoints(context);
                }
            );
        },
        registrationOptions.ConfigureMediator,
        serviceCollection =>
        {
            registrationOptions.ConfigureServices?.Invoke(serviceCollection);
        },
        environment
    );
 
    return services;
}
 
public static void ReceiveEndpointWithPolicies<TDbContext, TConsumer>(
    this IRabbitMqBusFactoryConfigurator cfg,
    IRegistrationContext context,
    MassTransitBusOptions busOptions,
    string queueName,
    Action<IRabbitMqReceiveEndpointConfigurator>? configureEndpoint = null
)
    where TDbContext : DbContext
    where TConsumer : class, IConsumer
{
    cfg.ReceiveEndpoint(
        queueName,
        endpoint =>
        {
            endpoint.ApplyEndpointPolicies<TDbContext>(context, busOptions);
            endpoint.ConfigureConsumer<TConsumer>(context);
            configureEndpoint?.Invoke(endpoint);
        }
    );
}

Focus here:

  • the helper owns transport bootstrapping and endpoint policy application.
  • service modules stay responsible only for message-specific topology and consumer registration.

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.

The Kafka path is explicit about rider setup and the in-memory base bus that hosts the rider:

src/BuildingBlocks/BuildingBlocks.Integration.MassTransit.Kafka/MassTransitKafkaExtensions.cs
public static IServiceCollection AddMassTransitKafka<TDbContext>(
    this IServiceCollection services,
    MassTransitOptions options,
    MassTransitKafkaRegistrationOptions? registrationOptions = null,
    IHostEnvironment? environment = null
)
    where TDbContext : DbContext
{
    registrationOptions ??= new MassTransitKafkaRegistrationOptions();
 
    services.AddMassTransitMessaging<TDbContext>(
        options,
        x =>
        {
            registrationOptions.ConfigureBus?.Invoke(x);
            x.AddRider(rider =>
            {
                registrationOptions.ConfigureRider?.Invoke(rider);
                rider.UsingKafka(
                    (context, kafka) =>
                    {
                        kafka.Host(
                            options.KafkaConnectionString
                                ?? throw new InvalidOperationException(
                                    "Missing connection string 'kafka'."
                                )
                        );
                        registrationOptions.ConfigureTransport?.Invoke(context, kafka);
                    }
                );
            });
            x.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context));
        },
        registrationOptions.ConfigureMediator,
        serviceCollection =>
        {
            registrationOptions.ConfigurePublisher?.Invoke(serviceCollection);
            registrationOptions.ConfigureServices?.Invoke(serviceCollection);
        },
        environment
    );
 
    return services;
}
 
public static void TopicEndpointWithPolicies<TDbContext, TMessage, TConsumer>(
    this IKafkaFactoryConfigurator kafka,
    IRegistrationContext context,
    MassTransitBusOptions busOptions,
    string topicName,
    string groupId,
    Action<IKafkaTopicReceiveEndpointConfigurator<Ignore, TMessage>>? configure = null
)
    where TDbContext : DbContext
    where TMessage : class
    where TConsumer : class, IConsumer<TMessage>
{
    kafka.TopicEndpoint<TMessage>(
        topicName,
        groupId,
        endpoint =>
        {
            endpoint.ApplyEndpointPolicies<TDbContext>(context, busOptions);
            endpoint.ConfigureConsumer<TConsumer>(context);
            configure?.Invoke(endpoint);
        }
    );
}

Focus here:

  • x.UsingInMemory(...) does not mean the messaging workflow is in-memory only; it hosts the base bus while the rider handles Kafka traffic.
  • TopicEndpointWithPolicies(...) keeps Kafka consumption aligned with the same retry and policy model as RabbitMQ, while still allowing Kafka-specific overrides.

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 from MassTransitServiceCollectionExtensions.cs:

src/BuildingBlocks/BuildingBlocks.Integration.MassTransit/Extensions/MassTransitServiceCollectionExtensions.cs
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);
    }
 
    if (!busOptions.UseErrorTransport)
    {
        endpoint.DiscardFaultedMessages();
    }
}

Focus here:

  • retry, delayed redelivery, consumer outbox, and error-transport behavior are applied in one reusable policy point.
  • DiscardFaultedMessages() makes the error-transport toggle explicit instead of implicit.

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 real AppHost keeps that switch narrow in AppHost.cs:

src/Aspire/ECommerce.AppHost/AppHost.cs
var builder = DistributedApplication.CreateBuilder(args);
var transport =
    builder.Configuration["Messaging:Transport"]?.Trim().ToLowerInvariant() ?? "rabbitmq";
 
var postgres = builder.AddPostgres("postgres").WithImage(PostgresImage).WithImageTag(PostgresTag);
var catalogsDb = postgres.AddDatabase("catalogsdb");
var ordersDb = postgres.AddDatabase("ordersdb");
 
var mongo = builder.AddMongoDB("mongo").WithImage(MongoImage).WithImageTag(MongoTag);
var catalogsMongo = mongo.AddDatabase("catalogs-mongo");
 
var catalogsApi = builder
    .AddProject<Projects.ECommerce_Services_Catalogs_Api>("catalogs-api")
    .WithReference(catalogsDb)
    .WithReference(catalogsMongo)
    .WithEnvironment("Messaging__Transport", transport);
 
var ordersApi = builder
    .AddProject<Projects.ECommerce_Services_Orders_Api>("orders-api")
    .WithReference(ordersDb)
    .WithEnvironment("Messaging__Transport", transport);
 
switch (transport)
{
    case "rabbitmq":
        var rabbitMq = builder
            .AddRabbitMQ("rabbitmq")
            .WithImage(RabbitMqImage)
            .WithImageTag(RabbitMqTag);
        catalogsApi.WithReference(rabbitMq);
        ordersApi.WithReference(rabbitMq);
        break;
    case "kafka":
        var kafka = builder.AddKafka("kafka").WithImage(KafkaImage).WithImageTag(KafkaTag);
        catalogsApi.WithReference(kafka);
        ordersApi.WithReference(kafka);
        break;
}

Focus here:

  • the databases and APIs are provisioned once.
  • only the broker resource and references change with transport selection.

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. The real envelope lives in MessageEnvelope.cs:

src/Services/Shared/ECommerce.Services.Shared/Contracts/MessageEnvelope/MessageEnvelope.cs
public sealed record MessageEnvelope<TMessage>(
    Guid MessageId,
    Guid CorrelationId,
    DateTime OccurredAtUtc,
    TMessage Message
) : IMessageEnvelopeMetadata
    where TMessage : IMessage
{
    Type IMessageEnvelopeMetadata.MessageType => typeof(TMessage);
}
 
public static class MessageEnvelope
{
    public static MessageEnvelope<TMessage> Create<TMessage>(
        TMessage message,
        Guid? correlationId = null,
        Guid? messageId = null,
        DateTime? occurredAtUtc = null
    )
        where TMessage : IMessage
    {
        return new MessageEnvelope<TMessage>(
            messageId ?? Guid.NewGuid(),
            correlationId ?? Guid.NewGuid(),
            occurredAtUtc ?? DateTime.UtcNow,
            message
        );
    }
}

Factory method keeps call sites small:

src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/CreatingProduct/v1/CreateProductEndpoint.cs
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:

src/Services/Catalogs/ECommerce.Services.Catalogs/ApplicationConfiguration.cs
switch (transport)
{
    case MessagingTransportType.RabbitMq:
        builder.Services.AddMassTransitRabbitMq<CatalogsDbContext>(
            options,
            new MassTransitRabbitMqRegistrationOptions
            {
                ConfigureTransport = (_, cfg) =>
                {
                    cfg.PublishToRabbitQueue<MessageEnvelope<ProductCreatedV1>>(
                        MessagingConstants.ProductCreatedQueue
                    );
                },
            },
            builder.Environment
        );
        break;
}

Kafka-oriented setup in latest sample:

src/Services/Catalogs/ECommerce.Services.Catalogs/ApplicationConfiguration.cs
case MessagingTransportType.Kafka:
    builder.Services.AddMassTransitKafka<CatalogsDbContext>(
        options,
        new MassTransitKafkaRegistrationOptions
        {
            ConfigureRider = rider =>
            {
                MassTransit.KafkaProducerRegistrationExtensions.AddProducer<
                    MessageEnvelope<ProductCreatedV1>
                >(rider, MessagingConstants.ProductCreatedTopic);
            },
            ConfigurePublisher = services =>
            {
                services.AddKafkaMessagePublisher<MessageEnvelope<ProductCreatedV1>>();
            },
        },
        builder.Environment
    );
    break;

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:

src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/CreatingProduct/v1/CreateProductEndpoint.cs
private static async Task<CreatedAtRoute<CreateProductResponse>> Handle(
    [FromBody] 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.EnqueueAsync(
        new ProjectProductReadModel(
            product.Id,
            product.Code,
            product.Name,
            product.Price,
            product.CreatedAtUtc
        ),
        cancellationToken
    );

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 the MassTransit sample intentionally diverges from Wolverine.

MassTransit still does not provide Wolverine-style durable local queues as a built-in abstraction, but the current sample no longer uses mediator-only post-commit handling. It fills the gap with BuildingBlocks.DurableLocalQueue, which persists internal commands in PostgreSQL and processes them in the background after commit.

The write-side registration happens in ApplicationConfiguration.cs:

src/Services/Catalogs/ECommerce.Services.Catalogs/ApplicationConfiguration.cs
var options = new MassTransitOptions
{
    DurableStorageConnectionString = connectionString,
    RabbitMqConnectionString = builder.Configuration.GetConnectionString("rabbitmq"),
    KafkaConnectionString = builder.Configuration.GetConnectionString("kafka"),
    Bus = new MassTransitBusOptions { UseBusOutbox = true, UsePostCommitMediator = true },
};
 
switch (transport)
{
    case MessagingTransportType.RabbitMq:
        builder.Services.AddMassTransitRabbitMq<CatalogsDbContext>(
            options,
            new MassTransitRabbitMqRegistrationOptions
            {
                ConfigureTransport = (_, cfg) =>
                {
                    cfg.PublishToRabbitQueue<MessageEnvelope<ProductCreatedV1>>(
                        MessagingConstants.ProductCreatedQueue
                    );
                },
            },
            builder.Environment
        );
        break;
}
 
builder.Services.AddDurableLocalQueue<CatalogsDbContext>();
 
builder.Services.AddDurableCommandHandler<ProjectProductReadModel>(
    async (command, sp, ct) =>
    {
        var repository = sp.GetRequiredService<IProductReadRepository>();
        await repository.UpsertAsync(
            new ProductReadModel(
                command.ProductId,
                command.Code,
                command.Name,
                command.Price,
                command.CreatedAtUtc
            ),
            ct
        );
    }
);

Focus here:

  • lines 27-36 still configure MassTransit itself, including bus outbox and transport selection.
  • lines 64-82 are the important divergence from the Wolverine article wording: the sample explicitly registers a durable local queue and a durable command handler, not only an in-memory mediator path.

The actual projection consumer stays small in ProjectProductReadModelConsumer.cs:

src/Services/Catalogs/ECommerce.Services.Catalogs/Products/Features/ProjectingProductReadModel/v1/ProjectProductReadModelConsumer.cs
using ECommerce.Services.Catalogs.Shared.Contracts;
using ECommerce.Services.Catalogs.Shared.ReadModels;
using ECommerce.Services.Shared.Contracts.InternalCommands;
using MassTransit;
 
namespace ECommerce.Services.Catalogs.Products.Features.ProjectingProductReadModel.v1;
 
public sealed class ProjectProductReadModelConsumer(IProductReadRepository repository)
    : IConsumer<ProjectProductReadModel>
{
    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
        );
    }
}

This is the same teaching point as in the Wolverine article: the handler logic stays small because durability and polling mechanics live below it.

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 from the real sample:

src/Services/Orders/ECommerce.Services.Orders/Products/Features/ConsumingProductCreated/v1/ProductCreatedConsumer.cs
using ECommerce.Services.Orders.Products.Models;
using ECommerce.Services.Orders.Shared.Data;
using ECommerce.Services.Shared.Contracts.IntegrationEvents;
using ECommerce.Services.Shared.Contracts.MessageEnvelope;
using MassTransit;
using Microsoft.EntityFrameworkCore;
 
namespace ECommerce.Services.Orders.Products.Features.ConsumingProductCreated.v1;
 
public sealed class ProductCreatedConsumer(OrdersDbContext dbContext)
    : IConsumer<MessageEnvelope<ProductCreatedV1>>
{
    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 existing = await dbContext.ImportedProducts.SingleOrDefaultAsync(
            x => x.Id == message.ProductId,
            context.CancellationToken
        );
 
        if (existing is null)
        {
            dbContext.ImportedProducts.Add(
                ImportedProduct.Create(
                    message.ProductId,
                    message.Code,
                    message.Name,
                    message.Price,
                    message.CreatedAtUtc
                )
            );
        }
        else
        {
            existing.Update(message.Code, message.Name, message.Price, message.CreatedAtUtc);
        }
 
        await dbContext.SaveChangesAsync(context.CancellationToken);
    }
}

Focus here:

  • lines 10-14 show the actual consumer contract: consume the envelope, then unwrap the business payload.
  • lines 15-24 show the built-in fault injection seam the tests use to verify retry and dead-letter behavior.
  • lines 26-40 show the imported-product upsert path and the final SaveChangesAsync, which is the side effect that the consumer outbox protects transactionally.

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:

tests/Services/Catalogs/ECommerce.Services.Catalogs.IntegrationTests/Products/Features/CreatingProduct/v1/CreateProductTests.cs
[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 created = await response.Content.ReadFromJsonAsync<CreateProductResult>(
        cancellationToken
    );
    Assert.NotNull(created);
 
    var published = await WaitForOutboxToDrainAsync(
        TimeSpan.FromSeconds(30),
        cancellationToken
    );
 
    Assert.True(
        published,
        "Expected the product-created event to be delivered by the bus outbox."
    );
 
    var mongoClient = new MongoClient(MongoConnectionString);
    var database = mongoClient.GetDatabase("catalogs-mongo");
    var collection = database.GetCollection<ProductReadModel>("product-read-models");
 
    ProductReadModel? readModel = null;
    for (var attempt = 0; attempt < 20; attempt++)
    {
        var readModels = await collection
            .Find(Builders<ProductReadModel>.Filter.Empty)
            .ToListAsync(cancellationToken);
        readModel = readModels.SingleOrDefault(x => x.Id == created!.Id);
        if (readModel is not null)
        {
            break;
        }
 
        await Task.Delay(250, cancellationToken);
    }
}

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 queuescustom durable local queue in current sample; otherwise mediator or dedicated durable endpoint
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. Wolverine ships durable local processing as a first-class feature. In this MassTransit sample, that gap is filled explicitly with a custom durable local queue. Without that extra building block, 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. This sample uses a custom durable local queue to keep MongoDB projection restart-safe. If you do not build that extra layer, mediator can be enough for post-commit ordering, but it is not a substitute for durable local processing.

That clarity is what keeps the architecture honest.