Outbox Pattern com MassTransit no .NET: mensagens confiáveis sem perda de eventos

Imagine que você processa um pedido, salva no banco de dados e depois tenta publicar um evento OrderCreated no RabbitMQ — mas o broker está fora por um segundo. O pedido foi salvo, o evento nunca chegou. O estoque não foi reservado. O e-mail de confirmação não foi enviado. Você acabou de criar um dado inconsistente silencioso.

O Outbox Pattern resolve isso com uma premissa simples: nunca publique mensagens diretamente para o broker dentro de uma transação de negócio. Salve a mensagem em uma tabela de outbox na mesma transação do banco de dados e deixe um processo separado fazer a entrega para o broker. Ou o broker recebe a mensagem, ou ela continua na tabela — nunca dados salvos sem evento.

Como o Outbox Pattern funciona

O fluxo tem três etapas: na transação de negócio, além de salvar as entidades, você insere a mensagem na tabela OutboxMessages na mesma transação. Se a transação falhar, a mensagem não existe — consistência garantida. Um processo em background (o outbox relay) lê as mensagens pendentes da tabela e as entrega ao broker. Quando o broker confirma o recebimento, o relay marca a mensagem como entregue.

O MassTransit tem suporte nativo ao Outbox Pattern com EF Core e oferece tudo isso pronto: a tabela de outbox, o relay de entrega, o controle de idempotência e a integração com os consumers.

Setup: MassTransit com Outbox e EF Core

# Pacotes necessários
dotnet add package MassTransit
dotnet add package MassTransit.RabbitMQ
dotnet add package MassTransit.EntityFrameworkCore
dotnet add package MassTransit.PostgreSQL  # se usar PostgreSQL

// Program.cs
builder.Services.AddMassTransit(x =>
{
    // Registra consumers automaticamente por assembly
    x.AddConsumers(typeof(Program).Assembly);

    // Habilita o Outbox com EF Core — chave de tudo
    x.AddEntityFrameworkOutbox<AppDbContext>(o =>
    {
        // Usa PostgreSQL como storage do outbox
        o.UsePostgres();

        // Limpa mensagens entregues após 24h
        o.UseBusOutbox(b =>
        {
            b.MessageDeliveryLimit = 10;
        });

        // Intervalo do relay (busca mensagens pendentes)
        o.QueryDelay = TimeSpan.FromSeconds(5);

        // Para evitar que múltiplos pods entreguem a mesma mensagem
        o.LockDuration = TimeSpan.FromSeconds(30);
    });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration["RabbitMQ:Host"], h =>
        {
            h.Username(builder.Configuration["RabbitMQ:User"]);
            h.Password(builder.Configuration["RabbitMQ:Password"]);
        });

        cfg.ConfigureEndpoints(context);
    });
});

Configurando o DbContext com as tabelas do Outbox

// src/Infrastructure/Persistence/AppDbContext.cs
public class AppDbContext : DbContext
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

    public DbSet<Order> Orders => Set<Order>();
    public DbSet<OrderItem> OrderItems => Set<OrderItem>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.ApplyConfigurationsFromAssembly(typeof(AppDbContext).Assembly);

        // Adiciona as tabelas do MassTransit Outbox ao schema
        // Cria: OutboxMessage, OutboxState, InboxState
        modelBuilder.AddInboxStateEntity();
        modelBuilder.AddOutboxMessageEntity();
        modelBuilder.AddOutboxStateEntity();
    }
}

// Migration gerada automaticamente incluirá as tabelas:
// OutboxMessage  — mensagens aguardando entrega
// OutboxState    — estado de entrega por consumer
// InboxState     — controle de idempotência no consumer

# Gera a migration que inclui as tabelas do outbox
dotnet ef migrations add AddMassTransitOutbox
dotnet ef database update

Publicando com garantia: transação atômica

// src/Application/UseCases/CreateOrderUseCase.cs
public class CreateOrderUseCase
{
    private readonly AppDbContext _context;
    private readonly IPublishEndpoint _publishEndpoint;

    public CreateOrderUseCase(AppDbContext context, IPublishEndpoint publishEndpoint)
    {
        _context = context;
        _publishEndpoint = publishEndpoint;
    }

    public async Task<Guid> ExecuteAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // Tudo dentro de uma única transação de banco de dados
        await using var transaction = await _context.Database
            .BeginTransactionAsync(ct);

        try
        {
            // 1. Salva o pedido no banco
            var order = Order.Create(request.CustomerId, request.Items);
            _context.Orders.Add(order);
            await _context.SaveChangesAsync(ct);

            // 2. Publica o evento — com Outbox habilitado, o MassTransit
            //    NÃO envia para o RabbitMQ agora. Insere na tabela OutboxMessage
            //    na mesma transação do banco de dados.
            await _publishEndpoint.Publish(new OrderCreatedEvent
            {
                OrderId = order.Id,
                CustomerId = order.CustomerId,
                Total = order.Total,
                Items = order.Items.Select(i => new OrderItemDto
                {
                    ProductId = i.ProductId,
                    Quantity = i.Quantity,
                    UnitPrice = i.UnitPrice
                }).ToList(),
                CreatedAt = order.CreatedAt
            }, ct);

            // 3. Commit: salva o pedido E a mensagem no outbox atomicamente.
            //    Se o commit falhar, nenhum dos dois persiste.
            //    Se o commit funcionar, a mensagem está garantida para entrega.
            await transaction.CommitAsync(ct);

            return order.Id;
        }
        catch
        {
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

Consumer com idempotência via InboxState

// src/Application/Consumers/OrderCreatedConsumer.cs
// O MassTransit Outbox garante "at-least-once delivery" — a mensagem pode
// chegar mais de uma vez em caso de retry. Use o InboxState para idempotência.

public class OrderCreatedConsumer : IConsumer<OrderCreatedEvent>
{
    private readonly IStockService _stockService;
    private readonly IEmailService _emailService;
    private readonly ILogger<OrderCreatedConsumer> _logger;

    public OrderCreatedConsumer(
        IStockService stockService,
        IEmailService emailService,
        ILogger<OrderCreatedConsumer> logger)
    {
        _stockService = stockService;
        _emailService = emailService;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<OrderCreatedEvent> context)
    {
        var orderId = context.Message.OrderId;

        _logger.LogInformation(
            "Processando OrderCreated para pedido {OrderId}", orderId);

        // O MassTransit InboxState registra que essa mensagem já foi processada.
        // Se a mensagem chegar novamente (retry do broker), o consumer não executa
        // de novo — o framework garante idempotência automaticamente
        // quando AddEntityFrameworkOutbox está configurado no consumer também.

        await _stockService.ReserveItemsAsync(context.Message.Items, context.CancellationToken);
        await _emailService.SendOrderConfirmationAsync(orderId, context.CancellationToken);

        _logger.LogInformation(
            "OrderCreated processado com sucesso para pedido {OrderId}", orderId);
    }
}

// Definição do evento — use record para imutabilidade
public record OrderCreatedEvent
{
    public Guid OrderId { get; init; }
    public Guid CustomerId { get; init; }
    public decimal Total { get; init; }
    public List<OrderItemDto> Items { get; init; } = [];
    public DateTime CreatedAt { get; init; }
}

public record OrderItemDto
{
    public Guid ProductId { get; init; }
    public int Quantity { get; init; }
    public decimal UnitPrice { get; init; }
}

Configuração do consumer com Outbox no lado do recebimento

// Para habilitar o InboxState (idempotência) no serviço que recebe a mensagem,
// o serviço consumidor também precisa do Outbox configurado.

// No microsserviço de estoque (StockService):
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderCreatedConsumer>();

    // Outbox no consumer: habilita InboxState para idempotência
    x.AddEntityFrameworkOutbox<StockDbContext>(o =>
    {
        o.UsePostgres();
        o.UseBusOutbox();
    });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(/* ... */);

        // Configura o consumer com InboxState
        cfg.ReceiveEndpoint("order-created", e =>
        {
            e.UseMessageRetry(r => r.Interval(3, TimeSpan.FromSeconds(5)));
            e.UseEntityFrameworkOutbox<StockDbContext>(context);
            e.ConfigureConsumer<OrderCreatedConsumer>(context);
        });
    });
});

Saga com Outbox: fluxos de negócio longe de falhas parciais

// O Outbox funciona perfeitamente com State Machines (Sagas) do MassTransit.
// Cada transição de estado e cada mensagem publicada fica no outbox.

public class OrderStateMachine : MassTransitStateMachine<OrderSagaState>
{
    public State WaitingPayment { get; private set; }
    public State WaitingShipment { get; private set; }
    public State Completed { get; private set; }
    public State Cancelled { get; private set; }

    public Event<OrderCreatedEvent> OrderCreated { get; private set; }
    public Event<PaymentConfirmedEvent> PaymentConfirmed { get; private set; }
    public Event<PaymentFailedEvent> PaymentFailed { get; private set; }
    public Event<ShipmentSentEvent> ShipmentSent { get; private set; }

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => OrderCreated, x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentConfirmed, x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentFailed, x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => ShipmentSent, x => x.CorrelateById(ctx => ctx.Message.OrderId));

        Initially(
            When(OrderCreated)
                .TransitionTo(WaitingPayment)
                .Publish(ctx => new RequestPaymentCommand
                {
                    OrderId = ctx.Message.OrderId,
                    Total = ctx.Message.Total
                }));  // ← publicado via outbox

        During(WaitingPayment,
            When(PaymentConfirmed)
                .TransitionTo(WaitingShipment)
                .Publish(ctx => new RequestShipmentCommand
                {
                    OrderId = ctx.Message.OrderId
                }),
            When(PaymentFailed)
                .TransitionTo(Cancelled)
                .Publish(ctx => new CancelOrderCommand
                {
                    OrderId = ctx.Message.OrderId,
                    Reason = "Payment failed"
                }));

        During(WaitingShipment,
            When(ShipmentSent)
                .TransitionTo(Completed)
                .Finalize());

        SetCompletedWhenFinalized();
    }
}

Monitorando o Outbox: mensagens presas e alertas

// Health check para detectar mensagens paradas no outbox
public class OutboxHealthCheck : IHealthCheck
{
    private readonly AppDbContext _context;

    public OutboxHealthCheck(AppDbContext context)
        => _context = context;

    public async Task<HealthCheckResult> CheckHealthAsync(
        HealthCheckContext context,
        CancellationToken ct = default)
    {
        // Mensagem com mais de 5 minutos sem entrega indica problema
        var stuckThreshold = DateTime.UtcNow.AddMinutes(-5);

        var stuckCount = await _context.Set<OutboxMessage>()
            .CountAsync(m =>
                m.DeliveredAt == null &&
                m.EnqueueTime < stuckThreshold, ct);

        if (stuckCount == 0)
            return HealthCheckResult.Healthy("Outbox: sem mensagens presas");

        if (stuckCount < 10)
            return HealthCheckResult.Degraded(
                $"Outbox: {stuckCount} mensagens com entrega atrasada");

        return HealthCheckResult.Unhealthy(
            $"Outbox: {stuckCount} mensagens presas — verificar conectividade com broker");
    }
}

// Registro no Program.cs
builder.Services.AddHealthChecks()
    .AddCheck<OutboxHealthCheck>("outbox",
        failureStatus: HealthStatus.Degraded,
        tags: ["messaging", "outbox"]);

Considerações de produção

Tamanho da tabela de outbox: configure a limpeza automática de mensagens entregues. O MassTransit já faz isso com CleanupInterval, mas monitore o crescimento da tabela se o volume de mensagens for alto. Em ambientes com milhares de mensagens por minuto, considere uma tabela particionada por data.

Múltiplos pods: o MassTransit usa locking otimista para evitar que dois pods entreguem a mesma mensagem ao mesmo tempo. O LockDuration configura por quanto tempo um pod "segura" uma mensagem. Se o pod morrer durante a entrega, outro pod assume após o timeout.

Ordering: o Outbox garante entrega, não ordem. Se a ordem de eventos importa para o consumer, use a mesma chave de partição no RabbitMQ/Kafka e projete os consumers para serem idempotentes e tolerantes a mensagens fora de ordem.

Dead letter: após N retries, o MassTransit move a mensagem para uma fila de erro. Monitore essa fila — mensagens na DLQ indicam um problema que precisa de intervenção humana.

---

Mensageria confiável em microsserviços exige mais do que configurar um broker. Se você quer implementar arquitetura orientada a eventos com garantias de entrega no seu sistema .NET, a Neryx pode ajudar.