MassTransit, from contract to outbox

This note follows one event through MassTransit, the framework compared in the previous noteMessaging frameworks for .NETWhat MassTransit, NServiceBus, Wolverine and CAP add on top of a raw broker client.Opens in a new tab. Orders publishes OrderPlaced, and Inventory reserves stock when it arrives. The code uses the MassTransit, MassTransit.RabbitMQ and MassTransit.EntityFrameworkCore packages at version 8.5. Version 9 is a commercial release, and its documentation shows the same configuration calls used here.

The contract

A contract is a plain type shared by both services, usually from a small shared project. MassTransit identifies a message by its full type name, namespace included, so the same type declared in two projects under different namespaces is two different messages and will not be consumed. A record is the recommended shape.

namespace Shop.Contracts; public record OrderPlaced(Guid OrderId, decimal Total);

A consumer

A consumer implements IConsumer<T> for the message it handles. It is registered with the container, so it can take services through its constructor (see dependency injectionDependency injection and service lifetimesHow ASP.NET Core hands each class what it needs, and how long each instance lives.Opens in a new tab).

public class ReserveStockConsumer(IStock stock, ILogger<ReserveStockConsumer> log)    : IConsumer<OrderPlaced>{    public async Task Consume(ConsumeContext<OrderPlaced> context)    {        log.LogInformation("Reserving stock for {OrderId}", context.Message.OrderId);        await stock.ReserveAsync(context.Message.OrderId, context.CancellationToken);    }}

Configuration

AddMassTransit registers the consumer and the bus. ConfigureEndpoints creates a receive endpoint for each registered consumer, and it should be the last call, after the middleware.

builder.Services.AddMassTransit(x =>{    x.AddConsumer<ReserveStockConsumer>();     x.UsingRabbitMq((context, cfg) =>    {        cfg.Host("localhost", "/", h =>        {            h.Username("guest");            h.Password("guest");        });         cfg.UseMessageRetry(r => r.Exponential(            5, TimeSpan.FromSeconds(1), TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(2)));         cfg.ConfigureEndpoints(context);    });});

UsingRabbitMq selects the transport. On RabbitMQ, MassTransit declares a fanout exchange named after the message type, a queue for the consumer, and the bindings between them. A published message goes to the type’s exchange, which copies it into every subscribed queue. That is the topic-to-queue shape from queue versus topicQueue vs topicA queue gives each message to one consumer, a topic gives every subscriber its own copy.Opens in a new tab.

UseMessageRetry retries a failing consumer in memory, here up to five times with growing delays (see retries and backoffRetries and backoffRun a failed message again after a growing, randomised delay, and stop after a limit.Opens in a new tab). When the retries run out, the message is moved to a queue named after the receive queue with an _error suffix, and a Fault<OrderPlaced> event is published. That error queue is MassTransit’s dead-letter queueDead-letter queuesWhere a message goes when it cannot be processed, so the rest of the traffic keeps moving.Opens in a new tab. Retry middleware must be configured before ConfigureEndpoints, or the endpoints will not get it.

Publishing

Orders publishes through IPublishEndpoint. Publish sends the event to every subscriber, while Send addresses a single queue and suits commands.

app.MapPost("/orders", async (    PlaceOrder cmd, OrdersDbContext db, IPublishEndpoint publish, CancellationToken ct) =>{    var order = Order.Create(cmd.CustomerId, cmd.Total);    db.Orders.Add(order);     await publish.Publish(new OrderPlaced(order.Id, order.Total), ct);    await db.SaveChangesAsync(ct);     return Results.Accepted($"/orders/{order.Id}");});

As written, lines 7 and 8 are two separate writes to two systems, the database and the broker. If the process stops between them, the order exists and the event does not, or the reverse. This is the dual-write problem from the outbox noteOutbox and inboxWrite the event in the same transaction as the data, publish it from there, and deduplicate on the way in.Opens in a new tab.

The transactional outbox

MassTransit’s Entity Framework outbox stores the outgoing message in the same database as the order. The DbContext gets three extra tables.

public class OrdersDbContext(DbContextOptions<OrdersDbContext> options) : DbContext(options){    public DbSet<Order> Orders => Set<Order>();     protected override void OnModelCreating(ModelBuilder modelBuilder)    {        modelBuilder.AddInboxStateEntity();        modelBuilder.AddOutboxMessageEntity();        modelBuilder.AddOutboxStateEntity();    }}

Then the bus outbox is switched on. With it, Publish writes the message into the outbox table instead of the broker, and SaveChangesAsync commits the order and the message in one transaction. The code in the endpoint does not change.

builder.Services.AddMassTransit(x =>{    x.AddEntityFrameworkOutbox<OrdersDbContext>(o =>    {        o.UsePostgres();        o.UseBusOutbox();    });     // ... AddConsumer and UsingRabbitMq as above ...});

After the commit, a delivery service in the application reads the outbox table and sends the messages to RabbitMQ, in the order they were published. There are two variants. The bus outbox covers messages published from an API request. The consumer outbox, enabled per receive endpoint with UseEntityFrameworkOutbox, also adds an inbox, which records received message ids so that a redelivered message is not processed twice. This follows the outbox and inboxOutbox and inboxWrite the event in the same transaction as the data, publish it from there, and deduplicate on the way in.Opens in a new tab pair.

Delivery is still at-least-once. The outbox service can send a message and fail before marking it sent, so a consumer without an inbox must be idempotentIdempotent consumersA consumer that gives the same result whether a message arrives once or five times.Opens in a new tab.

Sagas

A workflow that spans several services, such as reserve stock, charge payment, book shipping, is a sagaSagasA workflow made of local transactions, with an undo step for each one.Opens in a new tab. MassTransit models it as a state machine: a class derived from MassTransitStateMachine<TState> declares states and events, and each event moves the saga to the next state with TransitionTo. The saga’s state is kept in a repository, and Entity Framework, MongoDB and Redis are among the supported ones. The shop’s workflow is covered in the orchestrationChoreography vs orchestrationServices reacting to each other's events, or one coordinator telling them what to do.Opens in a new tab note, and a state machine is MassTransit’s way to orchestrate.