Wolverine Transactional Messaging with .NET Aspire, RabbitMQ, PostgreSQL, MongoDB, Inbox, Outbox, and Durable Local Processing

Introduction

Reliable messaging usually breaks in the dullest part of the request flow.

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 ends up building an outbox table, an inbox table, a background worker, retry logic, and operational dashboards to move data from one store to another.

This sample shows a smaller way to solve those problems with Wolverine and .NET Aspire. The example uses two services, PostgreSQL for durable messaging and write models, MongoDB for a read model, and a broker selected through configuration. The AppHost provisions RabbitMQ or Kafka based on WolverineBusOptions:TransportType, and each service uses a thin application-level wrapper so the transport-specific code stays small.

Sample code: wolverine-transactional-messaging-aspire

Objectives

This article shows how to:

1. Model a two-service sample around write-side persistence, integration events, and internal post-commit processing.
2. Configure Wolverine with PostgreSQL durability and EF Core transaction integration.
3. Publish broker messages through a transactional outbox.
4. Use durable local processing for an internal MongoDB read-model projection.
5. Consume broker messages with inbox semantics: inline listeners or the durable inbox, depending on transport.
6. Configure global retry policies and dead-letter queue behavior for RabbitMQ and Kafka listeners.
7. Keep transport-specific code thin enough to switch between RabbitMQ and Kafka without forking the service design.

Sample overview

The sample contains two microservices.

Catalogs is the write-side service.

• It stores product aggregates in PostgreSQL.
• It publishes MessageEnvelope<ProductCreatedV1> to RabbitMQ or Kafka based on configuration, using explicit per-message-type topology.
• Its publish topology also declares MessageEnvelope<OrderSubmittedV1> as a second integration event, routed through Wolverine conventional routing (RabbitMQ) or explicit topic publishing (Kafka) to prove both approaches can coexist in one topology file.
• It sends an internal ProjectProductReadModel command through Wolverine so the MongoDB read model updates only after the PostgreSQL transaction commits.

Orders is the downstream consumer.

• It listens to MessageEnvelope<ProductCreatedV1> via explicit queue binding from RabbitMQ or Kafka based on configuration.
• It listens to MessageEnvelope<OrderSubmittedV1> via Wolverine conventional routing (RabbitMQ) or explicit listener (Kafka), auto-creating queues and bindings at startup.
• It consumes with inline RabbitMQ listeners (the consumer pipeline itself is the inbox); the Kafka listener path supports durable inbox per listener via UseDurableInboxOnAllListeners.
• It applies configurable retry policies and native dead-letter queues.
• It stores imported product records in its own PostgreSQL database.

The reusable pieces live in a small set of shared and building-block projects:

• src/Services/Shared contains integration events, internal commands, and messaging constants.
• src/BuildingBlocks/BuildingBlocks.Core/Messages contains the reusable envelope abstraction (IMessageEnvelope, MessageEnvelope<T>, MessageEnvelopeMetadata).
• src/Services/Catalogs/ECommerce.Services.Catalogs contains per-service topology extension files (WolverineRabbitMqCatalogsTopologyExtensions.cs, WolverineKafkaCatalogsTopologyExtensions.cs) that keep transport wiring explicit and close to the service that owns it.
• src/Services/Orders/ECommerce.Services.Orders contains the corresponding consumer-side topology extensions (WolverineRabbitMqOrdersTopologyExtensions.cs, WolverineKafkaOrdersTopologyExtensions.cs).
• src/BuildingBlocks/BuildingBlocks.Integration.Wolverine* contains the sample Wolverine event bus, durable persistence service, and transport helpers.
• src/BuildingBlocks/BuildingBlocks.Persistence.* contains the PostgreSQL and MongoDB setup the sample needs.
• tests/Shared/Tests.Shared contains shared integration-test fixtures and sample test infrastructure.

Everything else was trimmed on purpose so the focus stays on Wolverine messaging behavior.

Architecture overview

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

There are three distinct messaging concerns in the sample:

1. External broker publishing from Catalogs.
2. Internal durable processing inside Catalogs.
3. Durable broker consumption in Orders.

Wolverine covers all three with one programming model, but the sample does not wire everything directly inside Program.cs. Instead, Catalogs and Orders call AddWolverineRabbitMq(...) or AddWolverineKafka(...), which both flow through a shared AddWolverineMessaging(...) extension. That shared extension applies PostgreSQL durability, EF Core transaction integration, durable local queues, durable inbox policy, handler discovery, and retry behavior from one place.

flowchart TB
    subgraph DockerEnvironment["Docker Environment"]
        subgraph Catalogs["Catalogs Service"]
            CatalogsApi["Catalogs API"]
            CatalogsPg["PostgreSQL: product aggregate and Wolverine envelopes"]
            WolverineOutbox["Wolverine transactional outbox"]
            WolverineLocal["Wolverine durable local queue"]
            CatalogsMongo["MongoDB read model"]
        end

        subgraph Orders["Orders Service"]
            OrdersApi["Orders API"]
            WolverineInbox["Wolverine durable inbox"]
            OrdersPg["PostgreSQL: imported products and Wolverine inbox"]
        end

        Broker["RabbitMQ or Kafka: ProductCreatedV1 + OrderSubmittedV1"]
    end

    Client["Client"] -->|HTTP| CatalogsApi
    Client["Client"] -->|HTTP| OrdersApi

    CatalogsApi -->|write product| CatalogsPg
    CatalogsApi -->|publish events| WolverineOutbox
    WolverineOutbox -.-> CatalogsPg
    WolverineOutbox -->|ProductCreatedV1 (explicit)| Broker
    WolverineOutbox -->|OrderSubmittedV1 (conventional)| Broker

    CatalogsApi -->|enqueue read model| WolverineLocal
    WolverineLocal -.-> CatalogsPg
    WolverineLocal -->|update read model| CatalogsMongo

    Broker -->|consume| WolverineInbox
    WolverineInbox -->|handle| OrdersApi
    WolverineInbox -.-> OrdersPg
    OrdersApi -->|store product| OrdersPg

Wolverine feature setup

The sample uses eight Wolverine capabilities. Each section below shows the minimal setup, the role it plays in the sample, and a link to the official docs.

1. Durable messaging with PostgreSQL

Wolverine stores message envelopes in PostgreSQL so that outgoing and internal messages survive process restarts. In this sample, both services point Wolverine durability at their own service database, which is also where the outbox and inbox records live.

Minimal Wolverine setup:

using Wolverine;
using Wolverine.Postgresql;

builder.Host.UseWolverine(options =>
{
    options.PersistMessagesWithPostgresql(connectionString);
});

Docs: Durable messaging

In the sample, this logic lives in Extensions.cs, which exposes AddWolverineMessaging(...) as the shared extension that all broker registrations use:

public static class Extensions
{
    public static IHostApplicationBuilder AddWolverineMessaging(
        this IHostApplicationBuilder builder,
        Action<WolverineBusOptions>? wolverineBusOptionsConfigure = null,
        Action<WolverineOptions, WolverineBusOptions>? configure = null,
        IReadOnlyCollection<Assembly>? assemblies = null
    )
    {
        var wolverineBusOptions = builder.Configuration.BindOptions<WolverineBusOptions>(
            nameof(WolverineBusOptions)
        );
        wolverineBusOptionsConfigure?.Invoke(wolverineBusOptions);

        builder.Services.AddWolverine(options =>
        {
            // Persistence is optional: when no durable-storage connection string is
            // configured (e.g. isolated building-block tests), Wolverine falls back to
            // its in-memory message store so no Postgres polling agents are started.
            if (!string.IsNullOrWhiteSpace(wolverineBusOptions.DurableStorageConnectionString))
            {
                options.PersistMessagesWithPostgresql(
                    connectionString: wolverineBusOptions.DurableStorageConnectionString,
                    schemaName: null
                );
            }

            if (wolverineBusOptions.UseEntityFrameworkCoreTransactions)
            {
                options.UseEntityFrameworkCoreTransactions();
            }

            if (wolverineBusOptions.UseDurableLocalQueues)
            {
                options.Policies.UseDurableLocalQueues();
            }

            if (wolverineBusOptions.UseDurableInboxOnAllListeners)
            {
                options.Policies.UseDurableInboxOnAllListeners();
            }

            foreach (var assembly in assemblies ?? Array.Empty<Assembly>())
            {
                // Explicitly register each handler assembly for Wolverine
                // discovery, equivalent to decorating the assembly with
                // [assembly: WolverineModule].
                options.Discovery.IncludeAssembly(assembly);
            }

            configure?.Invoke(options, wolverineBusOptions);

            if (wolverineBusOptions.Retry is { MaximumAttempts: > 0 })
            {
                var immediateRetries = wolverineBusOptions.Retry.MaximumAttempts - 1;
                if (immediateRetries > 0)
                {
                    options
                        .OnException<Exception>()
                        .RetryTimes(immediateRetries)
                        .Then.MoveToErrorQueue();
                }
                else
                {
                    options.OnException<Exception>().MoveToErrorQueue();
                }
            }
        });

        // Five scoped application-facing services back the messaging flows:
        // metadata access, external publishing, direct publishing, durable
        // persistence (outbox + local queue), and background job scheduling.
        builder.Services.AddScoped<IMessageMetadataAccessor, MessageMetadataAccessor>();
        builder.Services.AddScoped<IExternalEventBus, WolverineExternalEventBus>();
        builder.Services.AddScoped<IBusDirectPublisher, WolverineDirectPublisher>();
        builder.Services.AddScoped<IMessagePersistenceService, WolverineMessagePersistenceService>();
        builder.Services.AddScoped<IBackgroundJobScheduler, WolverineBackgroundJobScheduler>();

        return builder;
    }
}

2. EF Core transaction integration

Call UseEntityFrameworkCoreTransactions(). Wolverine middleware can participate in the same EF Core transaction boundary as the application write. In this sample, that option is enabled from configuration and applied centrally in the shared extension.

Minimal Wolverine setup:

using Wolverine.EntityFrameworkCore;

options.UseEntityFrameworkCoreTransactions();

Docs: Transactional middleware with EF Core

The sample turns this on conditionally inside the shared AddWolverineMessaging(...), gated by WolverineBusOptions.UseEntityFrameworkCoreTransactions.

3. Transactional outbox

Inside a handler or endpoint, publish a message through Wolverine while an EF Core transaction is open. Wolverine writes the envelope to the outbox table, an implementation of the at-least-once delivery pattern. After the transaction commits, the persisted envelopes are delivered to the configured transport.

Minimal Wolverine setup with manual outbox enrollment:

await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken);

outbox.Enroll(dbContext);
await bus.PublishAsync(message, cancellationToken);

await transaction.CommitAsync(cancellationToken);
await outbox.FlushOutgoingMessagesAsync();

Docs: Durable messaging

The sample uses the same pattern in CreateProduct.cs. The call passes the bare ProductCreatedV1 to the generic IExternalEventBus.PublishAsync<T> overload, which wraps it in a MessageEnvelope<T> internally:

await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken);

outbox.Enroll(dbContext);
dbContext.Products.Add(product);
await dbContext.SaveChangesAsync(cancellationToken);

await externalEventBus.PublishAsync(
    new ProductCreatedV1(
        product.Id,
        product.Code,
        product.Name,
        product.Price,
        product.CreatedAtUtc
    ),
    cancellationToken
);

await messagePersistence.EnqueueLocalAsync(new ProjectProductReadModel(...), cancellationToken);

await jobScheduler.ScheduleAsync(
    new SyncProductToExternalSystem(product.Id),
    TimeSpan.FromMinutes(5),
    cancellationToken
);

await transaction.CommitAsync(cancellationToken);
await outbox.FlushOutgoingMessagesAsync();

4. Durable local queues

Call UseDurableLocalQueues(). Commands sent to a local queue are written to PostgreSQL and processed by the same process, a form of internal processing. This gives retry semantics and survives restarts without adding an external broker for internal work.

Minimal Wolverine setup:

options.Policies.UseDurableLocalQueues();

Docs: Durable local queues

The sample turns this on for Catalogs through its appsettings.json (UseDurableLocalQueues: true), and the shared extension applies it to every Wolverine options object:

if (wolverineBusOptions.UseDurableLocalQueues)
{
    options.Policies.UseDurableLocalQueues();
}

Then CreateProduct.cs enqueues the projection command inside the same transaction:

await messagePersistence.EnqueueLocalAsync(
    new ProjectProductReadModel(...),
    cancellationToken
);

The internal command lives in ProjectProductReadModel.cs, and the handler in ProjectProductReadModelHandler.cs is intentionally small. It turns the internal command into a MongoDB upsert for the read model:

public static Task Handle(
    ProjectProductReadModel command,
    IProductReadRepository repository,
    CancellationToken cancellationToken
)
{
    return repository.UpsertAsync(
        new ProductReadModel(
            command.ProductId,
            command.Code,
            command.Name,
            command.Price,
            command.CreatedAtUtc,
            DateTime.UtcNow
        ),
        cancellationToken
    );
}

5. RabbitMQ transport

Call UseRabbitMqTransport(...) to enable RabbitMQ. The sample does not wire this directly at each service call site. Instead, Catalogs and Orders use AddWolverineRabbitMq(...), which internally delegates to AddWolverineMessaging(...) and activates the RabbitMQ transport with connection setup, topology provisioning, publishing, listeners, and dead-letter configuration.

Minimal Wolverine setup:

using Wolverine.RabbitMQ;

options.UseRabbitMqTransport(rabbitConnectionString)
       .PublishMessage<MyEvent>().ToRabbitQueue("my-queue")
       .ListenToRabbitQueue("my-queue");

Docs: RabbitMQ integration

Instead of wiring topology inline at each call site, each service delegates to a per-service topology extension file. Catalogs uses WolverineRabbitMqCatalogsTopologyExtensions.cs and registers its publishing topology through a single method call.

Each service passes its options through a delegate pattern that sets properties on WolverineBusOptions directly, rather than nesting wrapper objects:

builder.AddWolverineRabbitMq(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "rabbitmq";
        wolverineBusOptions.ConnectionString = rabbitMqConnectionString;
        wolverineBusOptions.DurableStorageConnectionString = connectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : rabbitMq => rabbitMq.ConfigureCatalogsPublishTopology(),
    assemblies: [typeof(CatalogsMetadata).Assembly]
);

The topology file demonstrates both approaches at the same time: explicit per-message-type binding for ProductCreatedV1 and conventional routing for a second message type, OrderSubmittedV1. This proves two routing strategies can coexist in the same service:

// ── Explicit: ProductCreatedV1 → Topic exchange ─────────────
builder.PublishToExchange<MessageEnvelope<ProductCreatedV1>>(
    nameof(ProductCreatedV1).Underscore()
);
builder.DeclareExchange(
    nameof(ProductCreatedV1).Underscore(),
    ex => ex.ExchangeType = ExchangeType.Topic
);

// ── Conventional routing: handles OrderSubmittedV1 ──────────
builder.UseSnakeCaseConventions(conventions =>
{
    conventions.IncludeTypes(type =>
        typeof(IMessageEnvelope).IsAssignableFrom(type)
        && !(
            type.IsGenericType
            && type.GetGenericTypeDefinition() == typeof(MessageEnvelope<>)
            && type.GetGenericArguments()[0] == typeof(ProductCreatedV1)
        )
    );
    conventions.ConfigureSending((ex, _) => ex.ExchangeType(ExchangeType.Topic));
});

ProductCreatedV1 uses a Topic exchange (replacing the earlier Direct) with an explicit PublishToExchange and DeclareExchange call, so the routing is transparent and easy to debug. OrderSubmittedV1 is handled by Wolverine conventional routing, which auto-discovers the snake_case naming convention (order_submitted_v1) and creates exchanges, queues, and bindings at startup. The IncludeTypes filter prevents the convention from also claiming ProductCreatedV1, which would produce a duplicate publisher declaration.

Note the exchange type is Topic for both. A Topic exchange routes messages by a routing key pattern, which gives more flexibility than Direct without adding meaningful complexity for event-driven topologies.

Pitfall: Topic exchanges need a non-empty routing key. PublishToExchange<T>() does not use ToRabbitExchange(exchangeName). That API sends with an empty routing key, which never matches a binding like product_created_v1, so the message is silently dropped (no error, no dead-letter entry, just nothing arrives). Instead, the builder uses ToRabbitRoutingKey(exchangeName, exchangeName). The exchange name doubles as the routing key, matching the listener binding queue.BindExchange(exchangeName, exchangeName):

_options.PublishMessage<T>().ToRabbitRoutingKey(exchangeName, exchangeName);

The WolverineRabbitMqRegistrationBuilder applies this rule for both its generic and reflection-based PublishToExchange overloads, so every explicit publish in the sample carries a resolvable routing key.

Orders registers its consumer topology the same way, delegating to WolverineRabbitMqOrdersTopologyExtensions.cs:

builder.AddWolverineRabbitMq(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "rabbitmq";
        wolverineBusOptions.ConnectionString = rabbitMqConnectionString;
        wolverineBusOptions.UseDurableLocalQueues = false;
        wolverineBusOptions.DeadLetterQueueName ??= MessagingConstants.DeadLetterQueueName;
        wolverineBusOptions.DurableStorageConnectionString = durableStorageConnectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : rabbitMq => rabbitMq.ConfigureOrdersConsumeTopology(),
    assemblies: [typeof(OrdersMetadata).Assembly]
);

Inside the Orders topology extension, the same dual approach applies: an explicit listener for ProductCreatedV1 and conventional routing for OrderSubmittedV1:

// ── Explicit: ProductCreatedV1 listener ─────────────────────
// Queue: product_created_v1, bound to Topic exchange.
builder.Listen<MessageEnvelope<ProductCreatedV1>>(
    nameof(ProductCreatedV1).Underscore(),
    listener => listener.ListenerCount(1)
);

// ── Conventional routing: auto-discovers OrderSubmittedV1 ───
// Creates queue order_submitted_v1 + binds to Topic exchange.
builder.UseSnakeCaseConventions(conventions =>
{
    conventions.IncludeTypes(type =>
        typeof(IMessageEnvelope).IsAssignableFrom(type)
        && !(
            type.IsGenericType
            && type.GetGenericTypeDefinition() == typeof(MessageEnvelope<>)
            && type.GetGenericArguments()[0] == typeof(ProductCreatedV1)
        )
    );
    conventions.ConfigureListeners((listener, _) => listener.ListenerCount(1));
});

The IncludeTypes filter mirrors the Catalogs one: it excludes ProductCreatedV1 so the explicit listener is the sole owner of that binding, while the convention handles only OrderSubmittedV1.

Under the hood, the shared RabbitMQ entry point uses UseRabbitMqUsingNamedConnection(connectionName).AutoProvision() and optionally customizes the dead-letter queue name. The builder exposes PublishToExchange<T>(), DeclareExchange(), UseConventionalRouting(), UseSnakeCaseConventions(), and a WithTransport() escape hatch for advanced scenarios.

6. Kafka transport

Call UseKafkaTransport(...) to enable Kafka. The shared Kafka entry point provides naming conventions, explicit topic binding, topic creation specifications, bulk auto-routing, and a WithTransport() escape hatch.

Both message types are published and consumed over RabbitMQ and Kafka. ProductCreatedV1 uses explicit per-message-type topology on both transports. OrderSubmittedV1 uses Wolverine conventional routing on RabbitMQ (auto-discovers queues and bindings) and explicit topic publishing on Kafka (mirrors the same PublishToTopic<T>() pattern).

Minimal Wolverine setup:

using Wolverine.Kafka;

options.UseKafkaTransport(kafkaConfig)
       .PublishMessage<MyEvent>().ToKafkaTopic("my-topic")
       .ListenToKafkaTopic("my-topic", "my-consumer-group");

Docs: Kafka integration

Catalogs publishes to Kafka through WolverineKafkaCatalogsTopologyExtensions.cs, delegating all topology to a single method:

builder.AddWolverineKafka(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "kafka";
        wolverineBusOptions.DurableStorageConnectionString = connectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : kafka => kafka.ConfigureCatalogsPublishTopology(),
    assemblies: [typeof(CatalogsMetadata).Assembly]
);

The topology file uses the explicit per-message-type approach for both integration events. It enables snake_case naming and binds each message type to its own topic with optional topic creation parameters:

// ── Explicit per-message-type topology ──
// Repeat per message type. Names derived via Humanizer.Underscore().
builder.UseSnakeCaseConventions();

builder.PublishToTopic<MessageEnvelope<ProductCreatedV1>>(
    nameof(ProductCreatedV1).Underscore(),
    spec =>
    {
        spec.NumPartitions = 3;
        spec.ReplicationFactor = 1;
    }
);

builder.PublishToTopic<MessageEnvelope<OrderSubmittedV1>>(
    nameof(OrderSubmittedV1).Underscore(),
    spec =>
    {
        spec.NumPartitions = 3;
        spec.ReplicationFactor = 1;
    }
);

Here UseSnakeCaseConventions() registers a naming function that uses Humanizer Underscore() to convert type names to snake_case. When you call PublishToTopic<T>(name, spec), the name argument is explicit and the optional Action<TopicSpecification> callback configures topic creation properties, a one-call pattern that replaces a separate DeclareTopic() step.

Docs: Kafka topic specification

The alternative uses Wolverine's built-in bulk auto-routing, which routes ALL published message types to Kafka topics by type name automatically with no per-type configuration:

// ── Option 2: Wolverine bulk auto-routing ──
// builder.PublishAllMessages();

Orders consumes from Kafka through WolverineKafkaOrdersTopologyExtensions.cs. The explicit per-message-type approach enables snake_case naming and listens with auto-derived topic names but explicit consumer groups:

builder.AddWolverineKafka(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "kafka";
        wolverineBusOptions.UseDurableLocalQueues = false;
        wolverineBusOptions.DeadLetterQueueName ??= MessagingConstants.DeadLetterQueueName;
        wolverineBusOptions.DurableStorageConnectionString = durableStorageConnectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : kafka => kafka.ConfigureOrdersConsumeTopology(),
    assemblies: [typeof(OrdersMetadata).Assembly]
);

Inside the Orders topology extension, the same explicit pattern is repeated for both message types, each with its own consumer group:

// ── Explicit per-message-type listeners ──
// Topic auto-derived via snake_case convention; consumer group explicit.
builder.UseSnakeCaseConventions();

builder.Listen<MessageEnvelope<ProductCreatedV1>>(
    topicName: null, // auto-derived: product_created_v1
    MessagingConstants.OrdersProductsConsumerGroup
);

builder.Listen<MessageEnvelope<OrderSubmittedV1>>(
    topicName: null, // auto-derived: order_submitted_v1
    MessagingConstants.OrdersOrdersConsumerGroup
);

The Kafka builder also provides WithNamingConvention() for custom naming rules, Publish<T>() / Listen<T>() overloads that auto-derive topic names from the convention, and a WithTransport() escape hatch that gives direct access to the underlying KafkaTransportExpression for advanced client configuration.

Docs:

• Kafka publish all messages
• Kafka listener configuration

7. Durable inbox

Call UseDurableInboxOnAllListeners(). Incoming broker deliveries are tracked in PostgreSQL so Wolverine can apply exactly-once delivery inbox semantics instead of treating every redelivery as new work.

Minimal Wolverine setup:

options.Policies.UseDurableInboxOnAllListeners();

Docs: Durable messaging

The shared AddWolverineMessaging(...) in Extensions.cs applies it through the WolverineBusOptions.UseDurableInboxOnAllListeners flag. The sample does not enable it globally: Orders consumes with inline RabbitMQ listeners (the consumer pipeline itself is the inbox), and the Kafka listener path opts in per listener via busOptions?.UseDurableInboxOnAllListeners ?? true.

There is one transport-specific nuance here. For Kafka, the shared listener helper ListenToKafkaTopicTransport (in WolverineOptionsExtensions.cs) applies durable inbox conditionally through the same option (busOptions?.UseDurableInboxOnAllListeners ?? true) and enables the native dead-letter topic unless UseNativeDeadLetterQueue is false. It is called from the Kafka registration builder's Listen<T>(topicName, consumerGroupId), so every explicitly declared listener inherits this behavior:

var listener = options
    .ListenToKafkaTopic(topicName)
    .ConfigureConsumer(config =>
    {
        config.GroupId = consumerGroupId;
    });

// Durable inbox: default true, opt-out via UseDurableInboxOnAllListeners = false
if (busOptions?.UseDurableInboxOnAllListeners ?? true)
{
    listener.UseDurableInbox();
}

if (busOptions?.UseNativeDeadLetterQueue != false)
{
    listener.EnableNativeDeadLetterQueue();
}
else
{
    listener.DisableNativeDeadLetterQueue();
}

That keeps Kafka listener behavior configurable, while RabbitMQ can rely on the shared policy when UseDurableInboxOnAllListeners is enabled.

8. Retry policies and dead-letter queues

Add a global retry policy so failed messages are retried before Wolverine gives up and routes them to an error or dead-letter destination.

Minimal Wolverine setup:

using Wolverine.ErrorHandling;

options
    .OnException<Exception>()
    .RetryTimes(2)
    .Then.MoveToErrorQueue();

That retries the message twice after the first failure, for a total of three attempts, then moves it to the error queue.

Docs: Error handling

The sample applies this policy from WolverineBusOptions.Retry inside Extensions.cs:

if (busOptions.Retry is { MaximumAttempts: > 0 })
{
    var immediateRetries = busOptions.Retry.MaximumAttempts - 1;
    if (immediateRetries > 0)
    {
        options
            .OnException<Exception>()
            .RetryTimes(immediateRetries)
            .Then.MoveToErrorQueue();
    }
    else
    {
        options.OnException<Exception>().MoveToErrorQueue();
    }
}

The retry count is controlled from configuration via the WolverineBusOptions section:

{
  "WolverineBusOptions": {
    "Retry": {
      "MaximumAttempts": 3
    }
  }
}

MaximumAttempts defaults to 3, so the sample keeps Wolverine's normal retry behavior unless you override it.

The interesting part starts after retries are exhausted.

• RabbitMQ uses native dead-lettering by default. The sample can override the dead-letter queue name through WolverineBusOptions.DeadLetterQueueName.
• Kafka needs an explicit opt-in on the listener, which the sample does with EnableNativeDeadLetterQueue().
• Both transports share the same UseNativeDeadLetterQueue and DeadLetterQueueName options, so the application-level configuration stays stable while the transport-specific wiring stays small.

The transport-specific dead-letter wiring happens inside the shared AddWolverineRabbitMq(...) and AddWolverineKafka(...) entry points. RabbitMQ applies dead-letter queue customization via UseRabbitMqUsingNamedConnection(connectionName).AutoProvision() and conditionally calls CustomizeDeadLetterQueueing(...) when DeadLetterQueueName is set. Kafka applies EnableNativeDeadLetterQueue() on each listener when UseNativeDeadLetterQueue is true.

RabbitMQ listener with the optional dead-letter queue configuration:

var listener = options.ListenToRabbitQueue(queueName);

if (busOptions?.UseNativeDeadLetterQueue == false)
{
    listener.DisableDeadLetterQueueing();
}
else if (!string.IsNullOrWhiteSpace(busOptions?.DeadLetterQueueName))
{
    listener.DeadLetterQueueing(new DeadLetterQueue(busOptions.DeadLetterQueueName));
}

Kafka listener with native dead-letter queue enabled:

if (busOptions?.UseNativeDeadLetterQueue != false)
{
    listener.EnableNativeDeadLetterQueue();
}
else
{
    listener.DisableNativeDeadLetterQueue();
}

The options are bound from the WolverineBusOptions configuration section:

public sealed class WolverineBusOptions
{
    public bool UseDurableInboxOnAllListeners { get; set; }
    public string DurableStorageConnectionString { get; set; } = string.Empty;

    public bool UseDurableLocalQueues { get; set; } = true;

    public bool UseEntityFrameworkCoreTransactions { get; set; } = true;

    public bool UseNativeDeadLetterQueue { get; set; } = true;

    public string? DeadLetterQueueName { get; set; }

    /// <summary>
    /// When true (default), topology is auto-discovered by scanning provided
    /// assemblies for IIntegrationEvent types with snake_case naming. No
    /// per-service topology file needed. Applies to both RabbitMQ and Kafka
    /// transports. Requires passing assemblies to the registration call;
    /// if omitted, neither auto nor manual topology runs.
    /// </summary>
    public bool AutoConfigMessagesTopology { get; set; } = true;

    public WolverineRetryOptions Retry { get; set; } = new();

    public MessagingTransportType TransportType { get; set; } = MessagingTransportType.RabbitMq;
    public string ConnectionName { get; set; } = string.Empty;
    public string? ConnectionString { get; set; }
}

MessagingTransportType is an enum with RabbitMq and Kafka values, letting each service switch transports through configuration alone.

That gives the sample one place to express retry policy and one place to express transport-specific dead-letter behavior.

Messaging patterns at a glance

Wolverine supports five distinct messaging patterns out of the box. The sample wires up all of them. The table below lists each pattern, what it does, the DB table that backs it, and where to see it running.

One database holds two schemas. Each service has a single PostgreSQL database (catalogsdb for Catalogs, ordersdb for Orders) that serves double duty. The business tables (products, imported_products) live in the default public schema, created by EF Core with the snake-case naming convention:

modelBuilder.Entity<Product>(builder =>
{
    builder.ToTable("products");
    ...
});

Wolverine's envelope tables (wolverine_outgoing_envelopes, wolverine_incoming_envelopes, wolverine_dead_letter_envelopes) live in their own wolverine schema inside that same database. PersistMessagesWithPostgresql takes schemaName: null, which means "use Wolverine's default schema name", i.e. wolverine:

options.PersistMessagesWithPostgresql(
    connectionString: wolverineBusOptions.DurableStorageConnectionString,
    schemaName: null
);

The detail that matters is that DurableStorageConnectionString is the same connection string as the business catalogsdb / ordersdb. InfrastructureExtensions resolves it once and assigns it to both the EF context and Wolverine's durable storage. There is no separate messaging database; the split is purely schema-level, so a single container per service holds both the domain data and the outbox/inbox tables. The business DbContext also maps the envelope storage into the same wolverine schema via MapWolverineEnvelopeStorage(), which is how EF EnsureCreated in the integration tests ends up creating both schemas.

Transactional Outbox

The outbox guarantees at-least-once delivery to the broker without two-phase commits or distributed transactions. The handler writes the domain data and the outgoing message inside a single EF Core transaction. Wolverine stores the message in its PostgreSQL outbox table (wolverine.wolverine_outgoing_envelopes). After the transaction commits, Wolverine flushes the pending messages to the configured broker. If the process crashes after the transaction commits but before the broker acknowledges delivery, Wolverine retries the flush on the next startup.

// Inside CreateProduct handler: outbox in action
outbox.Enroll(dbContext);
dbContext.Products.Add(product);
await dbContext.SaveChangesAsync(cancellationToken);

await externalEventBus.PublishAsync(
    new ProductCreatedV1(product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc),
    cancellationToken
);

await transaction.CommitAsync(cancellationToken);
await outbox.FlushOutgoingMessagesAsync();

Wolverine docs: Durable messaging · EF Core transactional middleware

What matters here: if the transaction rolls back, the outgoing message rolls back with it. There are no orphan events and no manual cleanup.

Durable Inbox

The inbox guarantees exactly-once processing on the consumer side. When a broker message arrives, Wolverine stores it in wolverine.wolverine_incoming_envelopes before acknowledging the broker. The handler processes the stored message. If the broker redelivers (because the ack was delayed), Wolverine detects the duplicate via the message ID and skips it.

The sample does not enable the policy globally; see 7. Durable inbox. Orders consumes with inline RabbitMQ listeners (the consumer pipeline itself is the inbox), and the Kafka listener path opts in per listener via ListenToKafkaTopicTransport.

Without a durable inbox, a broker redelivery during a transient handler failure would process the same event twice. The inbox makes the handler idempotent by default because it deduplicates for you.

Durable Local Queue (Internal Command Processor)

Not every post-commit task needs a broker. The durable local queue lets you enqueue work that executes inside the same process, persisted in PostgreSQL so it survives restarts. This sample uses it for the MongoDB read-model projection after a product is created.

The abstraction is IMessagePersistenceService.EnqueueLocalAsync:

await messagePersistence.EnqueueLocalAsync(
    new ProjectProductReadModel(
        product.Id, product.Code, product.Name, product.Price, product.CreatedAtUtc
    ),
    cancellationToken
);

The handler is a static Wolverine method discovered at startup:

public static Task Handle(
    ProjectProductReadModel command,
    IProductReadRepository repository,
    CancellationToken cancellationToken)
{
    return repository.UpsertAsync(
        new ProductReadModel(command.ProductId, command.Code, command.Name,
            command.Price, command.CreatedAtUtc, DateTime.UtcNow),
        cancellationToken);
}

Wolverine docs: Durable local queues

The local queue is immediate. The command executes as soon as the transaction commits and the outbox flushes. It is not delayed or scheduled. If you need a delay, see the scheduler below.

Durable Background Job Scheduler

Some work should happen later, not right after the commit. Wolverine's IMessageScheduler stores scheduled message entries in the same wolverine.wolverine_outgoing_envelopes table with a future delivery time. There is no separate table. A background scheduling agent polls the table every 10 seconds (configurable via ScheduledJob.PollingInterval) and dispatches due entries to the same handler pipeline as any other message, with the same handler discovery, retry policy, and DLQ behaviour.

Because the entries live in PostgreSQL, scheduled work survives process restarts. On startup, the durability agent checks for overdue entries and dispatches those too. The schedule entry is also transactional. If the handler that scheduled it participates in an outbox transaction and that transaction rolls back, the schedule entry never reaches the database.

The abstraction wraps Wolverine's IMessageScheduler:

public interface IBackgroundJobScheduler
{
    ValueTask ScheduleAsync<T>(T message, DateTimeOffset scheduledTime, CancellationToken ct = default)
        where T : class, IMessage;

    ValueTask ScheduleAsync<T>(T message, TimeSpan delay, CancellationToken ct = default)
        where T : class, IMessage;
}

Inside a handler: scheduling with transactionality

await jobScheduler.ScheduleAsync(
    new SyncProductToExternalSystem(product.Id),
    TimeSpan.FromMinutes(5),
    cancellationToken
);

Schedule to a specific endpoint (ScheduleSendAsync)

Wolverine also provides ScheduleSendAsync to route the scheduled message to a named transport endpoint instead of a local handler:

await scheduler.ScheduleSendAsync(
    new ProductAuditEvent(productId, changedBy),
    TimeSpan.FromHours(1),
    "audit-queue"
);

Scheduling agent configuration

options.ScheduledJob.PollingInterval = 5.Seconds();      // check every 5s instead of default 10s
options.ScheduledJob.FirstExecution = 2.Seconds();       // stagger initial poll after startup

What to remember:

• IMessageScheduler is built into WolverineFx.PostgreSQL, the same package already referenced for outbox and inbox durability. No extra table is created: scheduled entries are rows in wolverine_outgoing_envelopes carrying a delivery time.
• Retry and DLQ work the same as all other patterns, with one global policy governing everything.
• Overdue entries are dispatched on startup by the durability agent.
• Scheduling is one-shot only. For cron-style recurring jobs, use WolverineFx.Quartz (out of scope for this sample).

Distinction from EnqueueLocalAsync

Wolverine docs: IMessageScheduler · Scheduled job configuration

Retry + Dead-Letter Queue

When any handler throws, Wolverine retries according to the configured policy, then moves the permanently failed message to an error queue, either a broker-native DLQ or a PostgreSQL error table.

The sample configures a global policy inside AddWolverineMessaging(...):

if (busOptions.Retry is { MaximumAttempts: > 0 })
{
    var immediateRetries = busOptions.Retry.MaximumAttempts - 1;
    if (immediateRetries > 0)
    {
        options
            .OnException<Exception>()
            .RetryTimes(immediateRetries)
            .Then.MoveToErrorQueue();
    }
    else
    {
        options.OnException<Exception>().MoveToErrorQueue();
    }
}

The dead-letter destination is transport-specific:

• RabbitMQ: native dead-letter exchange + queue (configured via DeadLetterQueueName).
• Kafka: native dead-letter topic (enabled via EnableNativeDeadLetterQueue()).
• Fallback: Wolverine's PostgreSQL dead-letter store (wolverine.wolverine_dead_letter_envelopes), used when the transport has no native DLQ.

Wolverine docs: Error handling · RabbitMQ dead letter · Kafka native DLQ

Retry and DLQ apply to all the patterns above: outbox publishing, inbox processing, local queue commands, and scheduled jobs. One policy governs everything.

How the patterns compose

%%{init: {
  'theme': 'base',
  'themeVariables': {
    'background': '#ffffff',
    'primaryColor': '#ffffff',
    'primaryTextColor': '#2c3e50',
    'primaryBorderColor': '#bdc3c7',
    'lineColor': '#34495e',
    'secondaryColor': '#f8f9fa',
    'tertiaryColor': '#f1f3f5',
    'fontFamily': 'Segoe UI, Arial, sans-serif',
    'fontSize': '14px',
    'edgeLabelBackground': '#ffffff',
    'nodeBorder': '2px',
    'mainBkg': '#ffffff',
    'nodePadding': '12px',
    'subGraphBkg': '#fafbfc',
    'subGraphBorder': '#dfe6e9'
  }
}}%%
flowchart TB
    subgraph Request["HTTP POST /products"]
        Handler["CreateProductHandler<br/>(MediatR)"]
    end

    subgraph Transaction["EF Core Transaction"]
        direction TB
        Save["Save product to PostgreSQL"]
        Outbox["Transactional Outbox<br/>enroll → publish → enqueue"]
    end

    subgraph AfterCommit["After Commit"]
        Flush["Flush Outgoing Messages"]
    end

    Handler -->|1. Begin transaction| Transaction
    Transaction -->|2. Commit| AfterCommit
    AfterCommit -->|3. Publish| Broker["RabbitMQ / Kafka"]
    AfterCommit -->|4. EnqueueLocal| LocalQueue["Durable Local Queue<br/>PostgreSQL"]
    AfterCommit -->|5. ScheduleAsync| Scheduler["Background Job Scheduler<br/>PostgreSQL outgoing_envelopes<br/>(scheduled delivery time)"]

    Broker -->|6. Consume| Inbox["Durable Inbox<br/>PostgreSQL"]
    Inbox -->|7. Handle| OrdersHandler["Orders Handler"]
    OrdersHandler -.->|8. On failure| Retry["Retry N times"]
    Retry -.->|9. Exhausted| DLQ["Dead-Letter Queue"]

    LocalQueue -->|Handle| Projection["Project Read Model → MongoDB"]

    Scheduler -->|10. Time elapses| ScheduledHandler["Scheduled Job Handler"]

    classDef db fill:#e3f2fd,stroke:#1976d2,color:#0d47a1
    classDef broker fill:#f3e5f5,stroke:#7b1fa2,color:#4a148c
    classDef compute fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef error fill:#ffebee,stroke:#c62828,color:#b71c1c
    classDef messaging fill:#e0f2f1,stroke:#00796b,color:#004d40
    classDef gateway fill:#e8eaf6,stroke:#3949ab,color:#1a237e

    class Outbox,Inbox,LocalQueue,Scheduler messaging
    class Broker broker
    class Retry,DLQ error
    class Handler,OrdersHandler,Projection,ScheduledHandler compute
    class Request,Transaction,AfterCommit gateway

When to use each pattern: a decision guide

Comparison table: patterns and our abstractions

How the patterns relate

%%{init: {
  'theme': 'base',
  'themeVariables': {
    'background': '#ffffff',
    'primaryColor': '#ffffff',
    'primaryTextColor': '#2c3e50',
    'primaryBorderColor': '#bdc3c7',
    'lineColor': '#34495e',
    'secondaryColor': '#f8f9fa',
    'tertiaryColor': '#f1f3f5',
    'fontFamily': 'Segoe UI, Arial, sans-serif',
    'fontSize': '14px',
    'edgeLabelBackground': '#ffffff',
    'nodeBorder': '2px',
    'mainBkg': '#ffffff',
    'nodePadding': '12px',
    'subGraphBkg': '#fafbfc',
    'subGraphBorder': '#dfe6e9'
  }
}}%%
flowchart TB
    Request["HTTP POST /products<br/>CreateProductHandler (MediatR)"]

    subgraph Transaction["EF Core Transaction"]
        direction TB
        Enroll["outbox.Enroll(dbContext)<br/>dbContext.Products.Add(product)<br/>dbContext.SaveChangesAsync()"]

        subgraph Outbox["OUTBOX"]
            Publish["externalEventBus.PublishAsync<br/>→ ProductCreatedV1 (broker)"]
            Enqueue["messagePersistence.EnqueueLocalAsync<br/>→ ProjectProductReadModel<br/>(immediate local)"]
            Schedule["jobScheduler.ScheduleAsync<br/>→ SyncToExternalSystem<br/>(5 min delayed)"]
        end
    end

    subgraph AfterCommit["After Commit"]
        Commit["commit transaction<br/>flush outbox"]
    end

    Broker["RabbitMQ / Kafka<br/>ProductCreatedV1"]
    LocalQueue["PostgreSQL<br/>local queue<br/>ProjectReadModel"]
    ScheduledMsgs["PostgreSQL<br/>outgoing_envelopes<br/>(delivery time)<br/>SyncToExternal<br/>(5 min later)"]

    InboxDurable["INBOX (durable)<br/>Orders"]
    LocalHandler["local handler<br/>(upserts MongoDB)"]
    BackgroundJob["background job handler<br/>(calls external API)"]

    Request --> Enroll
    Enroll --> Outbox
    Outbox --> Commit
    Commit --> Broker
    Commit --> LocalQueue
    Commit --> ScheduledMsgs

    Broker --> InboxDurable
    LocalQueue --> LocalHandler
    ScheduledMsgs --> BackgroundJob

    classDef db fill:#e3f2fd,stroke:#1976d2,color:#0d47a1
    classDef broker fill:#f3e5f5,stroke:#7b1fa2,color:#4a148c
    classDef compute fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef error fill:#ffebee,stroke:#c62828,color:#b71c1c
    classDef messaging fill:#e0f2f1,stroke:#00796b,color:#004d40
    classDef gateway fill:#e8eaf6,stroke:#3949ab,color:#1a237e

    class Request,Enroll,Commit gateway
    class Outbox,Publish,Enqueue,Schedule messaging
    class Broker broker
    class LocalQueue,ScheduledMsgs db
    class InboxDurable,LocalHandler,BackgroundJob compute

Key distinction: internal command vs background job

Both are "background processors" from the app's perspective. The difference is timing: immediate vs scheduled. Wolverine does not use a separate table. Scheduled messages live in wolverine_outgoing_envelopes with a delivery time and are checked periodically (every 10s by default) by a polling agent rather than flushed immediately.

Proving retry and dead-letter behavior

The retry policy and the transport dead-letter wiring are fully in place, but the dedicated dead-letter integration tests are not part of the sample yet. This section separates what exists today from what is planned, so the prose matches the code.

What exists today:

• The global retry policy in AddWolverineMessaging (OnException<Exception>().RetryTimes(...).Then.MoveToErrorQueue()), driven by WolverineBusOptions.Retry.
• RabbitMQ native dead-lettering: AddWolverineRabbitMq calls CustomizeDeadLetterQueueing(...) when DeadLetterQueueName is set, and the shared listener helper applies DeadLetterQueueing(new DeadLetterQueue(name)), or DisableDeadLetterQueueing() when UseNativeDeadLetterQueue is false.
• Kafka native dead-lettering: ListenToKafkaTopicTransport calls EnableNativeDeadLetterQueue() unless UseNativeDeadLetterQueue is false.
• A fault-injection seam: ProductCreatedFaultyHandler.cs throws when the product code is FaultyProductCreated, and ProductCreatedHandler.cs calls it before doing its upsert. Publishing a product with that code makes the consumer fail through the normal handler path.

What does not exist yet:

• No DeadLetterQueueTests / KafkaDeadLetterQueueTests integration test classes. The tasks.json entries that reference them target classes that are not in the repository.

So the retry → dead-letter wiring is demonstrable by reading the code, but the end-to-end broker test that proves a failed message lands in the RabbitMQ dead-letter queue or Kafka dead-letter topic still needs to be written. The ProductCreatedFaultyHandler seam is the intended trigger for those tests.

Official Wolverine docs for this area:

• Error handling
• RabbitMQ dead letter queues
• Kafka native dead letter queue

Prerequisites

You need:

1. .NET 10 SDK
2. Docker Desktop or another supported OCI runtime
3. A recent .NET Aspire workload and tooling setup

AppHost infrastructure

The infrastructure entry point is AppHost.cs.

The AppHost does four useful things:

1. Provisions one PostgreSQL server and creates catalogsdb and ordersdb.
2. Provisions one MongoDB server and creates separate databases for each service.
3. Provisions either RabbitMQ or Kafka based on the WolverineBusOptions__TransportType environment variable.
4. Passes the selected transport to both APIs.

The full AppHost is compact. The important part is how WithReference() and WithEnvironment() work together:

const string PostgresImage = "postgres";
const string PostgresTag = "17";
const string MongoImage = "mongo";
const string MongoTag = "8.0";
const string RabbitMqImage = "rabbitmq";
const string RabbitMqTag = "4-management";
const string KafkaImage = "confluentinc/cp-kafka";
const string KafkaTag = "7.5.12";

var builder = DistributedApplication.CreateBuilder(args);
var transport =
    builder.Configuration["WolverineBusOptions:TransportType"]?.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 ordersMongo = mongo.AddDatabase("orders-mongo");

var catalogsApi = builder
    .AddProject<Projects.ECommerce_Services_Catalogs_Api>("catalogs-api")
    .WithReference(catalogsDb)
    .WithReference(catalogsMongo)
    .WithEnvironment("WolverineBusOptions__TransportType", transport);

var ordersApi = builder
    .AddProject<Projects.ECommerce_Services_Orders_Api>("orders-api")
    .WithReference(ordersDb)
    .WithReference(ordersMongo)
    .WithEnvironment("WolverineBusOptions__TransportType", 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;

    default:
        throw new InvalidOperationException(
            $"Unsupported messaging transport '{transport}'. Use 'rabbitmq' or 'kafka'."
        );
}

builder.Build().Run();

The transport switch stays small because Aspire's resource references handle all connection-string injection automatically.

How Aspire injects connection strings

Aspire uses WithReference() to bind a provisioned container resource to a project. Under the hood it adds environment variables of the form ConnectionStrings__{resourceName} to the target project. For example:

• catalogsApi.WithReference(rabbitMq) → Aspire adds ConnectionStrings__rabbitmq=amqp://guest:guest@localhost:5672 (resolved at runtime).
• catalogsApi.WithReference(kafka) → Aspire adds ConnectionStrings__kafka=localhost:9092.
• catalogsApi.WithReference(catalogsDb) → Aspire adds ConnectionStrings__catalogsdb=Host=localhost;Port=5432;Database=catalogsdb;....

The services consume these via GetConnectionString() in their infrastructure extension. For example, Catalogs reads them at startup:

var connectionString =
    builder.Configuration.GetConnectionString("catalogsdb");
var rabbitMqConnectionString = builder.Configuration.GetConnectionString("rabbitmq");
var wolverineOptions = builder.Configuration.BindOptions<WolverineBusOptions>(
    nameof(WolverineBusOptions));

GetConnectionString("catalogsdb") reads through to Configuration.GetConnectionString("catalogsdb"), which resolves Aspire's ConnectionStrings__catalogsdb environment variable. The value from GetConnectionString("catalogsdb") is then used to chain into WolverineBusOptions via the delegate:

AddWolverineRabbitMq(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "rabbitmq";
        wolverineBusOptions.ConnectionString = rabbitMqConnectionString;
        wolverineBusOptions.DurableStorageConnectionString = connectionString;
    },
    ...
);

When Aspire provisions RabbitMQ, ConnectionStrings__rabbitmq gets the full AMQP URI. When it provisions Kafka, ConnectionStrings__kafka gets the bootstrap server address. The key insight: ConnectionString and DurableStorageConnectionString are not set in appsettings.json. They are resolved from Aspire's environment variables at runtime and carried into the options delegate.

The WolverineBusOptions__TransportType environment variable is set explicitly with WithEnvironment() rather than through config, because it controls which transport the AppHost itself provisions. The services read it back through the same WolverineBusOptions:TransportType config path.

The default branch throws InvalidOperationException for unknown transport values, which keeps the AppHost honest. If someone forgets to set TransportType, they get a clear error at startup instead of silent fallback behavior.

The ECommerce.ServiceDefaults project follows the same idea as the Aspire Shop defaults project: centralize OpenTelemetry wiring, service discovery, resilient HTTP defaults, and health endpoints once, then reuse that baseline from each service.

Shared messaging contracts

The shared contracts live under src/Services/Shared.

The integration events are ProductCreatedV1.cs and OrderSubmittedV1.cs.

The envelope abstraction lives in BuildingBlocks.Core/Messages. The core interfaces are IMessageEnvelope.cs and MessageEnvelope.cs, plus the metadata record in MessageEnvelopeMetadata.cs.

public sealed record MessageEnvelope<T>(
    T Message,
    MessageEnvelopeMetadata Metadata
) : IMessageEnvelope
    where T : class, IMessage;

The MessageEnvelopeMetadata record carries the correlation data:

public record MessageEnvelopeMetadata(
    Guid MessageId,
    Guid CorrelationId,
    string MessageType,
    string Name,
    Guid? CausationId
)
{
    public IDictionary<string, object?> Headers { get; init; } = new Dictionary<string, object?>();

    public DateTime Created { get; init; } = DateTime.UtcNow;

    public long? CreatedUnixTime { get; init; } = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
}

That envelope gives the sample stable metadata for correlation, tracing, and test fixtures without pushing those fields into every event type.

Transport names stay in one place inside MessagingConstants.cs:

• ProductCreatedQueue → product_created_v1 (RabbitMQ queue)
• ProductCreatedTopic → catalogs-products-created (vestigial; see note below)
• OrdersProductsConsumerGroup → orders-products (Kafka consumer group)
• OrdersOrdersConsumerGroup → orders-orders (Kafka consumer group)
• DeadLetterQueueName → wolverine-dead-letter-queue (RabbitMQ DLQ)

Kafka topic names are derived automatically from the message type via Humanizer's Underscore(). ProductCreatedV1 becomes product_created_v1, so no separate topic constant is needed. ProductCreatedTopic exists in MessagingConstants.cs but is unused: the Kafka topology listens with topicName: null so the naming convention derives the topic.

The transport selection is driven by the WolverineBusOptions:TransportType config key. Each service binds its WolverineBusOptions from this section and switches on TransportType (rabbitmq or kafka). The AppHost uses the same key (WolverineBusOptions:TransportType) to provision the right broker container and pass it to both APIs via the WolverineBusOptions__TransportType environment variable.

Configuring Wolverine in Catalogs

The main write-side Wolverine configuration lives in InfrastructureExtensions.cs. It binds the WolverineBusOptions section, resolves the service connection strings, and switches on TransportType to pick the RabbitMQ or Kafka registration. Instead of wiring topology inline, it delegates to per-service topology extension methods that keep the transport configuration visible and close to the service:

switch (wolverineOptions.TransportType)
{
    case MessagingTransportType.RabbitMq:
        builder.AddWolverineRabbitMq(
            wolverineBusOptions =>
            {
                wolverineBusOptions.ConnectionName = "rabbitmq";
                wolverineBusOptions.ConnectionString = rabbitMqConnectionString;
                wolverineBusOptions.DurableStorageConnectionString = connectionString;
            },
            configure: wolverineOptions.AutoConfigMessagesTopology
                ? null
                : rabbitMq => rabbitMq.ConfigureCatalogsPublishTopology(),
            assemblies: [typeof(CatalogsMetadata).Assembly]
        );
        break;

    case MessagingTransportType.Kafka:
        builder.AddWolverineKafka(
            wolverineBusOptions =>
            {
                wolverineBusOptions.ConnectionName = "kafka";
                wolverineBusOptions.DurableStorageConnectionString = connectionString;
            },
            configure: wolverineOptions.AutoConfigMessagesTopology
                ? null
                : kafka => kafka.ConfigureCatalogsPublishTopology(),
            assemblies: [typeof(CatalogsMetadata).Assembly]
        );
        break;
}

The topology extension files (WolverineRabbitMqCatalogsTopologyExtensions.cs and WolverineKafkaCatalogsTopologyExtensions.cs) contain the actual topology wiring. Each file documents two approaches: an explicit per-message-type configuration (preferred) and a Wolverine auto-discovery approach (alternative). Names are derived through Humanizer's Underscore() to produce snake_case identifiers like product_created_v1.

Three lines matter most in practice:

1. PersistMessagesWithPostgresql(connectionString: ..., schemaName: null), applied only when a durable-storage connection string is configured; without one, Wolverine falls back to its in-memory message store (no Postgres polling agents)
2. UseEntityFrameworkCoreTransactions()
3. UseDurableLocalQueues()

Together they tell Wolverine to:

• store durable envelopes in PostgreSQL
• participate in the same transaction boundary as EF Core
• persist internal asynchronous commands instead of treating them as in-memory work

The shared WolverineBusOptions defaults assume you want durability unless you opt out:

public sealed class WolverineBusOptions
{
    public bool UseDurableInboxOnAllListeners { get; set; }
    public string DurableStorageConnectionString { get; set; } = string.Empty;

    public bool UseDurableLocalQueues { get; set; } = true;

    public bool UseEntityFrameworkCoreTransactions { get; set; } = true;

    public bool UseNativeDeadLetterQueue { get; set; } = true;

    public string? DeadLetterQueueName { get; set; }

    /// <summary>
    /// When true (default), topology is auto-discovered by scanning provided
    /// assemblies for IIntegrationEvent types with snake_case naming. No
    /// per-service topology file needed. Applies to both RabbitMQ and Kafka
    /// transports. Requires passing assemblies to the registration call;
    /// if omitted, neither auto nor manual topology runs.
    /// </summary>
    public bool AutoConfigMessagesTopology { get; set; } = true;

    public WolverineRetryOptions Retry { get; set; } = new();

    public MessagingTransportType TransportType { get; set; } = MessagingTransportType.RabbitMq;
    public string ConnectionName { get; set; } = string.Empty;
    public string? ConnectionString { get; set; }
}

Catalogs keeps the defaults, so EF Core transaction integration and durable local queues both stay on. Orders disables durable local queues (UseDurableLocalQueues = false) and defaults the dead-letter queue name to wolverine-dead-letter-queue, because its only messaging job is to consume broker events; it does not enqueue local projection commands.

The sample integration layer also splits responsibilities more clearly:

1. IExternalEventBus for application-facing external event publishing
2. IMessagePersistenceService for durable publish and durable local enqueue
3. transport-specific RabbitMQ and Kafka registration helpers

Transactional outbox in the Create Product slice

The business flow lives in the split slice files: CreateProductEndpoint.cs (HTTP boundary) and CreateProduct.cs (MediatR handler).

The endpoint maps the HTTP request to a CreateProduct command and sends it through MediatR:

private static async Task<CreatedAtRoute<CreateProductResponse>> Handle(
    [FromBody] CreateProductRequest request,
    ISender sender,
    CancellationToken cancellationToken
)
{
    var command = new CreateProduct(request.Code, request.Name, request.Price);
    var result = await sender.Send(command, cancellationToken);

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

The handler contains the core transactional outbox sequence:

var product = Product.Create(command.Code, command.Name, command.Price);

await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken);

outbox.Enroll(dbContext);
dbContext.Products.Add(product);
await dbContext.SaveChangesAsync(cancellationToken);

await externalEventBus.PublishAsync(integrationEvent, cancellationToken);
await messagePersistence.EnqueueLocalAsync(
    new ProjectProductReadModel(...), cancellationToken);

await jobScheduler.ScheduleAsync(
    new SyncProductToExternalSystem(product.Id),
    TimeSpan.FromMinutes(5),
    cancellationToken
);

await transaction.CommitAsync(cancellationToken);
await outbox.FlushOutgoingMessagesAsync();

This gives you a real transactional outbox flow.

• The product write model and Wolverine durable envelopes are part of the same PostgreSQL transaction.
• If the transaction rolls back, the outgoing integration event, the internal projection command, and the scheduled sync job all roll back with it.
• If the transaction commits, Wolverine can dispatch the external and internal messages afterward, while the scheduled sync job stays in the envelope table until its delivery time.

That is the gap most hand-rolled messaging designs struggle to close cleanly.

sequenceDiagram
    autonumber
    participant Client
    participant CatalogsAPI as Catalogs API
    participant MediatR
    participant CatalogsDB as Catalogs DB
    participant Outbox
    participant LocalQueue as Local Queue
    participant RabbitMQ
    participant OrdersAPI as Orders API
    participant OrdersDB as Orders DB
    participant ReadStore as Read Store

    Client->>CatalogsAPI: POST /products
    CatalogsAPI->>MediatR: Send(CreateProduct command)

    rect rgb(230, 245, 255)
        Note over MediatR,LocalQueue: Transaction
        MediatR->>CatalogsDB: Save product
        MediatR->>Outbox: Enroll transaction
        MediatR->>Outbox: Queue integration event
        MediatR->>LocalQueue: Queue projection command
        CatalogsDB-->>MediatR: Commit
    end

    CatalogsAPI-->>Client: 201 Created

    Note over Outbox,OrdersDB: Async delivery after commit
    Outbox->>RabbitMQ: Flush ProductCreatedV1
    LocalQueue->>ReadStore: Upsert read model
    RabbitMQ->>OrdersAPI: Deliver event
    OrdersAPI->>OrdersDB: Store imported product

The CreateProductValidator runs before the handler via MediatR's pipeline. If validation fails, the ValidationBehavior throws ValidationException, which the DefaultExceptionHandler in ServiceDefaults catches and returns as 422 ProblemDetails, all without branching logic in the endpoint or handler.

Durable internal processing for MongoDB read models

Kamil Grzybek's modular monolith write-up has a good explanation of internal processing: handling an in-process command after the domain transaction commits so downstream work stays consistent without relying on an external broker.

The internal command is ProjectProductReadModel.cs.

The handler is ProjectProductReadModelHandler.cs, and it writes through MongoProductReadRepository.cs.

The handler stays small:

public static Task Handle(
    ProjectProductReadModel command,
    IProductReadRepository repository,
    CancellationToken cancellationToken
)
{
    return repository.UpsertAsync(
        new ProductReadModel(
            command.ProductId,
            command.Code,
            command.Name,
            command.Price,
            command.CreatedAtUtc,
            DateTime.UtcNow
        ),
        cancellationToken
    );
}

That simplicity works because Wolverine already gives the command durable delivery semantics. You do not need a separate polling worker to keep the MongoDB projection retryable.

Durable inbox in Orders

The consumer-side configuration lives in InfrastructureExtensions.cs. As with Catalogs, AddInfrastructure resolves the connection strings, binds WolverineBusOptions, and switches on TransportType to pick the RabbitMQ or Kafka registration, delegating topology to per-service extension methods:

builder.AddWolverineRabbitMq(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "rabbitmq";
        wolverineBusOptions.ConnectionString = rabbitMqConnectionString;
        wolverineBusOptions.UseDurableLocalQueues = false;
        wolverineBusOptions.DeadLetterQueueName ??= MessagingConstants.DeadLetterQueueName;
        wolverineBusOptions.DurableStorageConnectionString = durableStorageConnectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : rabbitMq => rabbitMq.ConfigureOrdersConsumeTopology(),
    assemblies: [typeof(OrdersMetadata).Assembly]
);

The Kafka branch uses the same common options and the same delegation pattern:

builder.AddWolverineKafka(
    wolverineBusOptions =>
    {
        wolverineBusOptions.ConnectionName = "kafka";
        wolverineBusOptions.UseDurableLocalQueues = false;
        wolverineBusOptions.DeadLetterQueueName ??= MessagingConstants.DeadLetterQueueName;
        wolverineBusOptions.DurableStorageConnectionString = durableStorageConnectionString;
    },
    configure: wolverineOptions.AutoConfigMessagesTopology
        ? null
        : kafka => kafka.ConfigureOrdersConsumeTopology(),
    assemblies: [typeof(OrdersMetadata).Assembly]
);

The consumer topology files (WolverineRabbitMqOrdersTopologyExtensions.cs and WolverineKafkaOrdersTopologyExtensions.cs) follow the same pattern as Catalogs: RabbitMQ uses the dual approach (explicit + conventional), while Kafka uses explicit per-message-type listeners. The RabbitMQ consumer registers an explicit snake_case queue:

builder.Listen<MessageEnvelope<ProductCreatedV1>>(
    nameof(ProductCreatedV1).Underscore(),
    listener => listener.ListenerCount(1)
);

The Kafka consumer listens with auto-derived topic names and explicit consumer groups, repeated per message type:

builder.UseSnakeCaseConventions();

builder.Listen<MessageEnvelope<ProductCreatedV1>>(
    topicName: null,
    MessagingConstants.OrdersProductsConsumerGroup
);

builder.Listen<MessageEnvelope<OrderSubmittedV1>>(
    topicName: null,
    MessagingConstants.OrdersOrdersConsumerGroup
);

A note on durable inbox semantics. UseDurableInboxOnAllListeners() tracks broker deliveries in PostgreSQL so Wolverine can apply exactly-once delivery inbox semantics instead of treating every redelivery as new work. Orders does not enable it globally: with inline RabbitMQ listeners the consumer pipeline itself is the inbox, and the Kafka listener path opts in per listener via busOptions?.UseDurableInboxOnAllListeners ?? true in ListenToKafkaTopicTransport. The actual handler is ProductCreatedHandler.cs, which upserts an imported product record from the event payload after first delegating to ProductCreatedFaultyHandler.cs. That helper throws when the product code is FaultyProductCreated to simulate a consumer failure for dead-letter testing.

UseNativeDeadLetterQueue and the Retry options are also active for Orders. When a message fails repeatedly, Wolverine retries it up to the configured maximum and then moves it to the transport's dead-letter destination: RabbitMQ's native dead-letter queue or Kafka's native dead-letter topic.

Transport switching and topology provisioning

The sample contains both RabbitMQ and Kafka transport branches, and the active broker is selected through configuration instead of separate service implementations.

RabbitMQ and Kafka differ only at the transport edge:

• RabbitMQ uses queue publishing and queue listeners.
• Kafka uses topic publishing and topic listeners plus a consumer group.

The durability story stays the same:

• PostgreSQL stores Wolverine durable envelopes.
• EF Core transaction integration keeps the write model and envelopes aligned.
• Durable local queues handle internal post-commit processing.
• Durable inbox policies protect consumers.

At runtime, the switch is the WolverineBusOptions:TransportType setting in the service configuration or the Aspire environment override.

Per-service topology pattern

Each service keeps its transport topology in a dedicated extension file rather than wiring it inline in InfrastructureExtensions.cs. This follows a pattern similar to Wolverine's conventional routing but also makes explicit bindings transparent and debuggable.

The sample now demonstrates both approaches coexisting in the same topology file:

• Explicit per-message-type binding is used for ProductCreatedV1. Each direction (publishing in Catalogs and listening in Orders) is declared with an explicit exchange/queue name, exchange type, and listener configuration. Names are derived through Humanizer Underscore() to produce snake_case identifiers like product_created_v1. This is fully transparent and debuggable topology.

• Wolverine conventional routing handles OrderSubmittedV1 at the same time. The convention auto-discovers the message type, applies snake_case naming, and creates the exchange, queue, and binding at startup with zero per-type boilerplate. An IncludeTypes filter prevents the convention from also claiming ProductCreatedV1, which would produce duplicate declarations.

The exchange type is Topic for all bindings, both explicit and conventional. Topic exchanges route messages by a routing key pattern, giving more flexibility than Direct exchanges for event-driven topologies where multiple consumers may subscribe with different key patterns.

Kafka follows the explicit-only approach. Kafka does not have a UseConventionalRouting() equivalent. PublishToTopic<T>() and Listen<T>() are always explicit per-message-type calls. The Kafka topology files in the sample therefore repeat the PublishToTopic / Listen call for each message type: one for ProductCreatedV1, one for OrderSubmittedV1.

The file structure for each service looks like this:

src/Services/Catalogs/ECommerce.Services.Catalogs/
├── WolverineRabbitMqCatalogsTopologyExtensions.cs   # RabbitMQ publish topology
├── WolverineKafkaCatalogsTopologyExtensions.cs       # Kafka publish topology
└── Shared/Extensions/HostApplicationBuilderExtensions/
    └── InfrastructureExtensions.cs                   # transport switch + registration

src/Services/Orders/ECommerce.Services.Orders/
├── WolverineRabbitMqOrdersTopologyExtensions.cs      # RabbitMQ consume topology
├── WolverineKafkaOrdersTopologyExtensions.cs          # Kafka consume topology
└── Shared/Extensions/HostApplicationBuilderExtensions/
    └── InfrastructureExtensions.cs                   # transport switch + registration

InfrastructureExtensions.AddInfrastructure in each service switches on MessagingTransportType.RabbitMq / Kafka and calls the appropriate topology method. This keeps the transport registration logic in one place while the topology details stay in focused, per-transport files.

Wolverine's RabbitMQ transport supports AutoProvision(). Exchanges, queues, and bindings declared in topology extensions are created automatically at startup, so no manual provisioning code is needed.

Message metadata travels through Wolverine headers. When a message implements IMessageEnvelope, WolverineDeliveryOptionsFactory.cs copies MessageId, CorrelationId, and Created onto DeliveryOptions so the envelope metadata crosses the broker with the payload.

Publish the envelope, not the raw payload

A second, more subtle trap hides behind the topology above: routes are registered for MessageEnvelope<T>, never for the raw T. Auto-config topology (AutoConfigMessagesTopology = true) builds PublishToExchange(typeof(MessageEnvelope<>).MakeGenericType(eventType), exchangeName) for every IIntegrationEvent. The wrapper type gets the route; the inner event type does not.

The low-level IBusDirectPublisher used by WolverineMessagePersistenceService initially published messageEnvelope.Message (the raw inner ProductCreatedV1) instead of the envelope object. Two things went wrong:

1. Routing fell back to conventional rules. The raw type has no registered route, so Wolverine picked the plain exchange endpoint rabbitmq://exchange/product_created_v1 (no /routing/ segment) and sent an empty routing key → silently dropped on the Topic exchange.
2. TrackActivity recorded Sent for the raw type. Tests asserting ShouldPublishing<MessageEnvelope<ProductCreatedV1>> saw nothing, because Wolverine records activity for the published type, which was the inner message, not the wrapper.

The fix is one line: publish the envelope object, not its inner payload:

// WolverineDirectPublisher.cs
return bus.PublishAsync(messageEnvelope, deliveryOptions);   // ✅ wrapper → registered route
// return bus.PublishAsync(messageEnvelope.Message, ...);    // ❌ raw → conventional fallback → empty routing key

With the envelope-first publish, the sender becomes a routing endpoint (RabbitMqSender: rabbitmq://exchange/product_created_v1/routing/product_created_v1). That is the tell-tale difference when debugging: a plain exchange URI means the message type was never bound, and a /routing/ endpoint means the routing key resolves.

Rule of thumb: if your publishing abstraction wraps messages in MessageEnvelope<T>, publish the envelope object through the bus. Keep raw payload publishing only for type-based local routing (durable local queues), where bus.SendAsync(message) is correct.

To run the sample with RabbitMQ:

WolverineBusOptions__TransportType=rabbitmq dotnet run --project src/Aspire/ECommerce.AppHost/ECommerce.AppHost.csproj

To run the same code path with Kafka:

WolverineBusOptions__TransportType=kafka dotnet run --project src/Aspire/ECommerce.AppHost/ECommerce.AppHost.csproj

Validation

Build the solution:

dotnet build wolverine-transactional-messaging-aspire.slnx

Run the tests. The sample uses xUnit v3 with Microsoft Testing Platform v2, so the classic dotnet test form no longer works on the .NET 10 SDK ("Testing with VSTest target is no longer supported"). Run the MTP-native test runner directly instead:

dotnet run --project tests/Services/Orders/ECommerce.Services.Orders.IntegrationTests/ECommerce.Services.Orders.IntegrationTests.csproj
dotnet run --project tests/Services/Catalogs/ECommerce.Services.Catalogs.IntegrationTests/ECommerce.Services.Catalogs.IntegrationTests.csproj

To run one test class, pass its fully qualified class name (short names yield "Zero tests ran"):

dotnet run --project tests/Services/Orders/ECommerce.Services.Orders.IntegrationTests/ECommerce.Services.Orders.IntegrationTests.csproj \
  -- --filter-class ECommerce.Services.Orders.IntegrationTests.Products.Features.ConsumingProductCreated.v1.ProductCreatedConsumerTests

The current test suite covers:

• MessageEnvelope<T> metadata creation
• durable internal MongoDB projection handler behavior
• downstream consumer upsert behavior
• publisher-side outbox verification via TrackActivity (Sent after commit, no fault)
• consumer-side inbox verification via TrackActivity (Received + MessageSucceeded, no fault)
• integration startup tests that override WolverineBusOptions:TransportType to exercise both RabbitMQ and Kafka
• vertical-slice integration tests that keep one test class per feature and switch the broker through appsettings overrides

Dead-letter integration coverage is not part of the suite yet. See Proving Retry and Dead-Letter Behavior.

Asserting real broker round-trips with TrackActivity

The integration tests assert real cross-service delivery over the broker, not inline handler invocation. They run through one shared harness, SharedFixture<TEntryPoint> in tests/Shared/Tests.Shared/Fixtures/SharedFixture.cs, which does the same job as the MassTransit test harness (TestHarness, AssertPublished, AssertConsumed, …), but on top of Wolverine's own TrackActivity() API.

The SharedFixture test harness

Every helper funnels through one BuildSession method, which starts tracking on the application's service provider and applies the options the test asked for:

private TrackedSessionConfiguration BuildSession(
    bool includeExternalTransports,
    Func<Type, bool>? ignoreMessageTypes = null
)
{
    var session = Factory.Services.TrackActivity().Timeout(TestTimeout);

    if (includeExternalTransports)
    {
        session = session.IncludeExternalTransports();
    }

    if (ignoreMessageTypes is not null)
    {
        session = session.IgnoreMessagesMatchingType(ignoreMessageTypes);
    }

    return session;
}

TrackActivity() records a history of envelope events, each tagged with a MessageEventType. That event vocabulary is what the helpers assert on:

Timeout(TestTimeout) bounds the whole session: Wolverine wraps the tracked action in WaitAsync(Timeout) and throws a TimeoutException (including the activity grid) if the expected activity never completes, so a broken test can never hang the run.

All helpers share one assertion pattern: run the action inside the session with ExecuteAndWaitAsync, then query the history with FindEnvelopesWithMessageType<T>(eventType). A positive ShouldNotBeEmpty() proves the event happened; AutoFaultPublished must stay empty to prove no fault was raised. The five helpers differ only in which events they demand:

Every helper comes in a Func<IMessageContext, Task> form (for when the action needs the tracked message context) and a plain Func<Task> form. ShouldConsuming<T> also offers assertSideEffect variants that verify the consumer's side effect (e.g. the persisted ImportedProduct row) after the tracked session completes. All forward an optional CancellationToken.

ShouldConsuming<T> is the consumer-side assertion:

public async Task ShouldConsuming<T>(
    Func<IMessageContext, Task> action,
    bool includeExternalTransports = false,
    CancellationToken cancellationToken = default
)
    where T : class
{
    var trackedSession = await BuildSession(includeExternalTransports)
        .ExecuteAndWaitAsync(action);

    // Message arrived at the consumer listener
    trackedSession
        .FindEnvelopesWithMessageType<T>(MessageEventType.Received)
        .ShouldNotBeEmpty();

    // Handler executed successfully
    trackedSession
        .FindEnvelopesWithMessageType<T>(MessageEventType.MessageSucceeded)
        .ShouldNotBeEmpty();

    // No fault published
    trackedSession
        .FindEnvelopesWithMessageType<Fault<T>>(MessageEventType.AutoFaultPublished)
        .ShouldBeEmpty();
}

ShouldProcessingOutboxMessage<T> is the publisher-side outbox proof. It hard-codes includeExternalTransports: false, the external-transport trap explained below:

public async Task ShouldProcessingOutboxMessage<T>(
    Func<IMessageContext, Task> action,
    Func<Task>? assertOutbox = null,
    Func<Type, bool>? ignoreMessageTypes = null,
    CancellationToken cancellationToken = default
)
    where T : class
{
    var trackedSession = await BuildSession(
            includeExternalTransports: false,
            ignoreMessageTypes: ignoreMessageTypes
        )
        .ExecuteAndWaitAsync(action);

    // The outbox flush handed the envelope to the broker after the
    // business transaction committed
    trackedSession.FindEnvelopesWithMessageType<T>(MessageEventType.Sent).ShouldNotBeEmpty();

    trackedSession
        .FindEnvelopesWithMessageType<Fault<T>>(MessageEventType.AutoFaultPublished)
        .ShouldBeEmpty();

    if (assertOutbox is not null)
    {
        await assertOutbox();
    }
}

ShouldProcessingInternalCommand<T> asserts on MessageSucceeded rather than Sent, because the internal command runs locally after the main handler. No transport event is involved:

public async Task ShouldProcessingInternalCommand<T>(
    Func<IMessageContext, Task> action,
    Func<Task>? assertSideEffect = null,
    bool includeExternalTransports = false,
    Func<Type, bool>? ignoreMessageTypes = null,
    CancellationToken cancellationToken = default
)
    where T : class
{
    var trackedSession = await BuildSession(
            includeExternalTransports,
            ignoreMessageTypes: ignoreMessageTypes
        )
        .ExecuteAndWaitAsync(action);

    // The internal command executed successfully after the main handler
    trackedSession
        .FindEnvelopesWithMessageType<T>(MessageEventType.MessageSucceeded)
        .ShouldNotBeEmpty();

    trackedSession
        .FindEnvelopesWithMessageType<Fault<T>>(MessageEventType.AutoFaultPublished)
        .ShouldBeEmpty();

    if (assertSideEffect is not null)
    {
        await assertSideEffect();
    }
}

Three helpers accept a trailing assertion (assertSideEffect / assertOutbox) that runs after the tracked session completes, in the window where the handler's durable effects are visible. The Orders consumer tests use it to confirm the ImportedProduct row was persisted before declaring the consume successful.

Filtering noise with ignoreMessageTypes. Every helper takes an optional ignoreMessageTypes: Func<Type, bool> filter, forwarded to IgnoreMessagesMatchingType. A tracked session waits for all tracked activity, and the CreateProduct handler schedules SyncProductToExternalSystem five minutes into the future, a job the session will never see complete. The tests pass a filter that skips IInternalCommand types they are not asserting on, so the scheduled job cannot hold the session open until timeout.

Polling without hanging: WaitUntilConditionMet. Read models and projections appear asynchronously, so the fixture also provides a polling loop:

public async ValueTask WaitUntilConditionMet(
    Func<Task<bool>> conditionToMet,
    int? timeoutSecond = null,
    string? exception = null,
    CancellationToken cancellationToken = default
)

It evaluates the condition and, while it fails, waits 100 ms and re-checks, up to timeoutSecond (default 90 seconds). If the condition never becomes true it throws TimeoutException instead of spinning forever, and the cancellation token is honored on every iteration.

Per-test timeout and cancellation. IntegrationTestBase creates a cancellation source per test, linked to xUnit's ambient token, and cancels it after SharedFixture.TestTimeout (90 seconds):

public virtual async ValueTask InitializeAsync()
{
    _testTimeoutCts?.Dispose();
    _testTimeoutCts = CancellationTokenSource.CreateLinkedTokenSource(
        GetAmbientCancellationToken()
    );
    _testTimeoutCts.CancelAfter(SharedFixture.TestTimeout);

    await SharedFixture.ResetAsync();
    await ResetStateAsync();
}

TestCancellationToken is then threaded through every helper and polling loop, so a stuck test is stopped by three layers: the runner's own cancellation, the 90-second per-test token, and the tracked session's WaitAsync. Ninety seconds is generous headroom. The suites themselves finish in seconds (container startup is a collection-fixture concern that runs once per class, not per test), while still failing fast on a genuine hang.

The external-transport trap. Wolverine's tracking has two recorders. Without IncludeExternalTransports(), outgoing records are recorded with RecordLocally, which marks an external Sent complete the moment it is handed to the transport. With it, RecordCrossApplication applies a stricter rule: a non-scheduled Sent record only completes when a matching Received record arrives from a tracked receiver. That is the right semantic for cross-service flow tests, but if the consuming service is a separate process (as here), no Received ever arrives and the session times out. That is why ShouldProcessingOutboxMessage intentionally keeps external-transport tracking off: the Sent record still proves the outbox flush handed the message to the broker, and the Orders tests separately verify the receiving half with includeExternalTransports: true (where the receiver is in-process, so Received does arrive).

Earlier versions of ShouldPublishing<T> computed the sent-envelope list but never asserted it. That check was a silent no-op that masked regressions like the raw-payload publish bug above. The strict ShouldNotBeEmpty() is what catches them.

A producer-side test in Catalogs drives the full CreateProduct slice (DB write → outbox → publish) and asserts the envelope left the service:

[Fact]
public async Task PostProduct_ShouldProcessOutboxMessage()
{
    var request = CatalogsTestData.NewProductRequest();

    // Act + Assert: after the write transaction commits, the transactional outbox
    // flushes MessageEnvelope<ProductCreatedV1> and hands it to the broker (Sent).
    // No fault published. This is the publisher-side proof that the message went
    // through the outbox during publishing.
    await SharedFixture.ShouldProcessingOutboxMessage<MessageEnvelope<ProductCreatedV1>>(
        async () =>
        {
            var response = await SharedFixture.GuestClient.PostAsJsonAsync(
                "/api/v1/catalogs/products",
                request
            );

            Assert.Equal(HttpStatusCode.Created, response.StatusCode);
        },
        ignoreMessageTypes: IgnoreScheduledInternalCommands
    );
}

The internal-command test verifies the durable local processing half. ProjectProductReadModel runs after the transaction commits, and its side effect (the Mongo read model) becomes visible through the API:

[Fact]
public async Task PostProduct_ShouldProcessInternalCommand()
{
    var request = CatalogsTestData.NewProductRequest();
    CreateProductResult? created = null;

    await SharedFixture.ShouldProcessingInternalCommand<ProjectProductReadModel>(
        async () =>
        {
            var response = await SharedFixture.GuestClient.PostAsJsonAsync(
                "/api/v1/catalogs/products", request);
            Assert.Equal(HttpStatusCode.Created, response.StatusCode);
            created = await response.Content.ReadFromJsonAsync<CreateProductResult>();
        },
        async () =>
        {
            // poll until the read-model endpoint serves the projected document
            ProductReadModelResult? readModel = null;
            await SharedFixture.WaitUntilConditionMet(async () =>
            {
                var response = await SharedFixture.GuestClient.GetAsync(
                    $"/api/v1/catalogs/products/read-model/{created!.Id}");
                if (response.StatusCode != HttpStatusCode.OK) return false;

                readModel = await response.Content.ReadFromJsonAsync<ProductReadModelResult>();
                return readModel is not null;
            });

            Assert.Equal(request.Name, readModel!.Name);
        },
        ignoreMessageTypes: t =>
            typeof(IInternalCommand).IsAssignableFrom(t) && t != typeof(ProjectProductReadModel)
    );
}

A consumer-side test in Orders verifies the other half of the round-trip: publish the envelope through the real RabbitMQ broker (IMessageBus.PublishAsync(envelope)), let the consumer listener persist the imported product, and assert the read model appears in MongoDB.

SharedFixture.ShouldConsuming<MessageEnvelope<ProductCreatedV1>>(
    async _ => await PublishEnvelopeAsync(envelope),
    assertSideEffect: async () =>
    {
        var imported = await WaitForImportedProductAsync(message.ProductId);
        imported.ShouldNotBeNull();
    },
    includeExternalTransports: true
);

Taken together, the two test classes cover every step of the transaction flow, mapped one-to-one to the verification goals:

Building-block integration tests

The service tests above exercise whole vertical slices through the real Catalogs and Orders APIs. A second suite, under tests/BuildingBlocks, isolates the Wolverine building blocks themselves against a real broker without any microservice.

Each building-block test project owns its host configuration. The shared BuildingBlocksSharedFixture creates a WebApplicationBuilder with WebApplication.CreateBuilder, enables TestServer, invokes the derived fixture's dependency and topology configuration, and starts the resulting WebApplication. This keeps RabbitMQ and Kafka building-block tests independent from service entry points and avoids a shared test-host project.

The RabbitMQ building-block fixture, RabbitMqBuildingBlocksSharedFixture.cs, does three things:

1. Boots a real RabbitMQ container via RabbitMqContainerFixture.
2. Disables auto topology, EF Core transactions, and durable local queues, then registers the test project's handler assembly and manual topology.
3. Supplies the RabbitMQ container connection string through an in-memory configuration override and starts the standalone WebApplication owned by the fixture.

Because durable storage is disabled for these tests, Wolverine falls back to its in-memory message store. No PostgreSQL polling agents are started, so the tests focus purely on publish/consume round-trips and broker topology.

The fixture calls AddWolverineRabbitMq(..., configure: rabbitMq => rabbitMq.ConfigureTestRabbitMqTopology()) directly. The manual topology in WolverineRabbitMqTestTopology.cs exercises every major builder API:

• PublishToExchange<T>(exchange) + DeclareExchange(..., ExchangeType.Topic) + Listen<T>(queue) + BindQueue(...)
• Publish<T>(queue) + Listen<T>(queue) for direct queue round-trips
• UseSnakeCaseConventions(...) for conventional fanout routing
• DeclareExchange, DeclareQueue, BindQueue, and BindExchangeToExchange for declarative topology

The publish tests in WolverineRabbitMqPublishTests.cs use IMessageBus inside the test process, then assert that MessageEnvelope<T> is received by the matching in-process handler:

[Fact]
public async Task should_round_trip_product_created_via_topic_exchange()
{
    var envelope = MessageEnvelopeFactory.From(NewProductCreated());

    await SharedFixture.ShouldConsuming<MessageEnvelope<ProductCreatedV1>>(
        async _ => await PublishEnvelopeAsync(envelope),
        includeExternalTransports: true,
        cancellationToken: TestCancellationToken
    );
}

The topology tests in WolverineRabbitMqTopologyTests.cs query the RabbitMQ management HTTP API to verify that exchanges, queues, bindings, and exchange-to-exchange bindings were actually declared on the broker.

SharedFixture.ResetAsync() runs before every test and delegates to RabbitMqContainerFixture.ResetAsync(). That reset is narrow on purpose: it purges only the contents of each queue through the management API (DELETE /api/queues/{vhost}/{name}/contents) and leaves the topology (exchanges, queues, bindings) untouched. Wolverine's AutoProvision declared that topology once when the host started, and because the reset never removes it, the shared host stays up for the whole collection: listeners stay attached to the same queues and wait for the next message. That is why the RabbitMQ building-block tests never restart the host between tests.

Kafka building-block tests: why topics survive between tests

The Kafka building-block fixture, KafkaBuildingBlocksSharedFixture.cs, follows the same pattern: it boots a real confluentinc/cp-kafka container, passes the container's mapped BootstrapServers directly as WolverineBusOptions.ConnectionString, registers the test project's handlers and manual topology, and starts its own WebApplication through BuildingBlocksSharedFixture.

The difference is how the two brokers treat a reset. RabbitMQ separates structure (exchanges, queues, bindings) from data (messages), so a reset can purge messages without touching the structure. Kafka has no such split: a topic is the queue and the binding in one. KafkaContainerFixture.ResetAsync() deletes topics outright (DeleteTopicsAsync), which removes the very topology the running host depends on.

That is a problem because the host is long-lived across the collection: Wolverine's AutoProvision only creates topics at startup. Once a later test deletes them, listeners stay subscribed to partitions that no longer exist (Unknown topic or partition), AutoProvision never runs again, and every later round-trip times out.

The fix lives in the shared base fixture. SharedFixtureCore.ResetAsync exposes a virtual switch so a collection can opt out of broker cleanup while still resetting the databases and ensuring the host is available before broker assertions. Service-level hosts start lazily through WebApplicationFactory; building-block hosts have already been started by BuildingBlocksSharedFixture.InitializeAsync:

protected virtual bool ResetBrokerStateBetweenTests => true;

public virtual async Task ResetAsync(CancellationToken cancellationToken = default)
{
    if (Postgres is not null)
        await Postgres.ResetAsync();
    if (Mongo is not null)
        await Mongo.ResetAsync(cancellationToken);

    if (ResetBrokerStateBetweenTests)
    {
        if (Kafka is not null)
            await Kafka.ResetAsync(cancellationToken);
        if (RabbitMq is not null)
            await RabbitMq.ResetAsync(cancellationToken);
    }

    // Ensure the host is initialized so AutoProvision has created the broker topology.
    _ = ServiceProvider;
}

The Kafka fixture opts out:

protected override bool ResetBrokerStateBetweenTests => false;

The trade-off is that the Kafka tests share one static topology and messages can accumulate in the topics across tests. That is safe here: each round-trip publishes a fresh envelope with a new id and waits for its own consumption through SharedFixture.ShouldConsuming<T> (backed by TrackActivity), so the tests never assert on messages left by a previous test. If you need strict per-test isolation with Kafka, the alternatives are a fresh host per test (which re-runs AutoProvision) or unique topics per test. Both are slower than sharing the static topology.

Orders consumes with inline RabbitMQ listeners in these tests, so Wolverine does not persist a handled copy into wolverine.wolverine_incoming_envelopes. The inbox is the in-memory consumer pipeline itself. That is what TrackActivity's Received/MessageSucceeded events prove: the durable-inbox admission step is in the pipeline, and the envelope is processed exactly once (MassTransit's AssertConsumed analog). The last test also confirms the message store backing the inbox/outbox infrastructure is queryable and healthy.

The Kafka building-block tests (WolverineKafkaPublishTests.cs) cover the same round-trip shape over PublishToTopic (explicit topic), UseSnakeCaseConventions (auto-named topic/group), and WithNamingConvention (custom naming), and the topology tests (WolverineKafkaTopologyTests.cs) verify the topics exist via the Kafka admin client.

The shared integration test fixtures explicitly use the local Docker images already present in the sample environment:

• postgres:17
• rabbitmq:4-management
• confluentinc/cp-kafka:7.5.12
• mongo:8.0

The integration test structure now mirrors the application slices more closely:

• each service keeps one shared integration-test base that has both RabbitMQ and Kafka fixtures available
• the base replaces broker connection strings according to WolverineBusOptions:TransportType
• startup tests verify broker selection with explicit appsettings overrides
• feature tests stay under the same vertical-slice folders as the production code, such as Products/Features/CreatingProduct/v1 and Products/Features/GettingImportedProducts/v1

If you want to exercise the live flow manually, start the AppHost, open the Aspire dashboard, copy the catalogs-api base URL, and create a product:

curl -X POST "<catalogs-base-url>/api/v1/catalogs/products" \
  -H "Content-Type: application/json" \
  -d '{"code":"catalog-001","name":"Starter Basket","price":42.50}'

Then inspect:

• GET <catalogs-base-url>/api/v1/catalogs/products/read-model
• GET <orders-base-url>/api/v1/orders/products

The first endpoint shows the internal durable projection into MongoDB. The second shows the downstream broker-consumer result in the Orders service.

Project structure

The sample keeps a feature-first structure without adding extra infrastructure layers that do not help explain the messaging flow.

Key folders:

• src/Aspire/ECommerce.AppHost
• src/Aspire/ECommerce.ServiceDefaults
• src/BuildingBlocks/*
• src/Services/Catalogs/*
• src/Services/Orders/*
• src/Services/Shared/*
• tests/Shared/Tests.Shared
• tests/Services/Catalogs/*
• tests/Services/Orders/*

Inside each service, the features stay close to the endpoints and handlers that implement them. Each service also includes per-transport topology extension files that keep the broker wiring explicit:

Catalogs topology:
├── WolverineRabbitMqCatalogsTopologyExtensions.cs  # RabbitMQ publish topology
└── WolverineKafkaCatalogsTopologyExtensions.cs      # Kafka publish topology

Orders topology:
├── WolverineRabbitMqOrdersTopologyExtensions.cs     # RabbitMQ consume topology
└── WolverineKafkaOrdersTopologyExtensions.cs         # Kafka consume topology

In Catalogs, that means slices such as:

• Products/Dtos/v1/ProductDto.cs
• Products/Features/CreatingProduct/v1/CreateProduct.cs
• Products/Features/CreatingProduct/v1/CreateProductValidator.cs
• Products/Features/CreatingProduct/v1/CreateProductEndpoint.cs
• Products/Features/GettingProductById/v1/GetProductById.cs
• Products/Features/GettingProductById/v1/GetProductByIdEndpoint.cs
• Products/Features/GettingProductReadModels/v1/GetProductReadModels.cs
• Products/Features/GettingProductReadModels/v1/GetProductReadModelsEndpoint.cs
• Products/Features/ProjectingProductReadModel/v1/ProjectProductReadModelHandler.cs

Each slice separates the MediatR request/handler from the minimal API endpoint and, where needed, adds a dedicated FluentValidation validator. The endpoint stays thin (HTTP mapping only), and the handler owns the business logic.

On the Orders side, the same structure holds. Note the second consumer slice for OrderSubmittedV1, discovered automatically by conventional routing:

• Products/Dtos/v1/ImportedProductDto.cs
• Products/Features/ConsumingProductCreated/v1/ProductCreatedHandler.cs
• Products/Features/ConsumingOrderSubmitted/v1/OrderSubmittedHandler.cs
• Products/Features/GettingImportedProducts/v1/GetImportedProducts.cs
• Products/Features/GettingImportedProducts/v1/GetImportedProductsEndpoint.cs
• Products/Features/GettingImportedProductById/v1/GetImportedProductById.cs
• Products/Features/GettingImportedProductById/v1/GetImportedProductByIdEndpoint.cs

The test projects follow the same idea. Instead of separate test hierarchies per broker, each slice keeps a small number of tests and overrides WolverineBusOptions:TransportType when it needs to validate RabbitMQ or Kafka behavior.

That is enough structure to feel like a real microservice codebase without turning the sample into a framework demo.

Trade-offs

It helps to be explicit about the trade-offs:

1. You are relying on Wolverine runtime behavior and durability tables instead of handwritten plumbing.
2. Eventual consistency still means temporary lag between the PostgreSQL write model, MongoDB projection, and downstream service state.
3. The sample is intentionally small, so it does not cover every operational concern such as richer contract versioning, full observability dashboards, or a large distributed compatibility matrix for every broker scenario.

Those trade-offs are reasonable here because the goal is clarity around transactional messaging, not production completeness in every direction.

Conclusion

This sample shows how to build two small services around one consistent messaging model.

Catalogs writes a product once, Wolverine persists the external and internal work transactionally, MongoDB is projected through durable local processing, and Orders consumes the event through its inline RabbitMQ listeners (or Kafka listeners with per-listener durable inbox). Aspire keeps the infrastructure reproducible, and the shared test layer now proves the same slices against broker selection through configuration without burying the point under a large amount of generic scaffolding.

If your current design still depends on hand-written outbox rows, custom pollers, and ad hoc retry logic, Wolverine is a strong candidate to remove that infrastructure burden while keeping the service code small and explicit.

References

• Background Work with Wolverine, Jeremy Miller
• Durable Background Processing with Wolverine, Jeremy Miller
• Scheduled Message Delivery with Wolverine, Jeremy Miller
• Wolverine Documentation
• Wolverine Durability Guide
• Wolverine EF Core Transactional Middleware
• Wolverine RabbitMQ Transport
• Wolverine Kafka Transport