Idempotent Consumers and the Inbox: When the Same Message Arrives Twice

Version note: The code in this post was built and run with .NET 10, EF (Entity Framework) Core 10.0.14, Npgsql.EntityFrameworkCore.PostgreSQL 10.0.3 (PostgreSQL 17), RabbitMQ 4.4, RabbitMQ.Client 7.2.2 and MassTransit 8.5.12.

Guess First

The Billing consumer issues an invoice and takes the payment for an OrderPlaced message. The same message (same MessageId, same order) arrives twice; the consumer channel can process 4 messages at once. The first line of the handler is this check:

if (await db.Invoices.AnyAsync(i => i.OrderId == message.OrderId))
    return "duplicate, skipped";

When it is all over, how many invoices does the database hold for this order, and how many times was the customer charged?

The same message arriving twice is not a bug, it is the natural consequence of at-least-once delivery: the consumer finishes the work and crashes before sending the ack, the publisher republishes because no confirm came back, the outbox relay publishes and stops before it can mark the row, an operator replays from the DLQ (Dead-Letter Queue). In the previous two posts we saw that retry and replay each produce "at least one extra copy". The conceptual side of the idempotent consumer is in the "Idempotent Consumer" post of the Messaging and Event-Driven Architecture series; here we write the code with EF Core and PostgreSQL.

Why Check-Then-Act Is Not Enough

The full version of the check above:

public sealed class NaiveInvoiceHandler(BillingDbContext db, FakePaymentGateway payments)
{
    public async Task<string> HandleAsync(OrderPlaced message)
    {
        // check-then-act: if two copies pass here at the same time, both see "none"
        if (await db.Invoices.AnyAsync(i => i.OrderId == message.OrderId))
            return "duplicate, skipped";

        await payments.ChargeAsync(message);
        db.Invoices.Add(new Invoice { OrderId = message.OrderId, Amount = message.Total, Currency = message.Currency });
        await db.SaveChangesAsync();
        return "invoice created";
    }
}

The fake payment service takes 200 ms. The consumer channel was opened with consumerDispatchConcurrency: 4, and I published the same message twice (for this demo I dropped the unique index on OrderId):

delivery tag 1 message msg-7f3a: invoice created
delivery tag 2 message msg-7f3a: invoice created
invoices for order: 2, charges: 2

Two invoices, two charges. Both deliveries ran AnyAsync within the same few milliseconds, both saw "no invoice" and both carried on. With concurrency 1 this demo would have been lucky, but in production, when two instances listen to the same queue, the ordering of a single channel does not protect you. The only place that can close the gap between the check and the write is the database itself: a unique constraint.

The Inbox Table: Dedupe by MessageId

The first solution is to record every processed message in a table. The key combines the MessageId and the consumer name; the same message must be processed separately by Billing and Notification, and one must not skip it because it sees the other's record:

public sealed class ProcessedMessage
{
    public required string MessageId { get; init; }
    public required string Consumer { get; init; }
    public DateTimeOffset ProcessedAt { get; init; } = DateTimeOffset.UtcNow;
}

public sealed class BillingDbContext(DbContextOptions<BillingDbContext> options) : DbContext(options)
{
    public DbSet<Invoice> Invoices => Set<Invoice>();
    public DbSet<ProcessedMessage> ProcessedMessages => Set<ProcessedMessage>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<Invoice>().HasIndex(i => i.OrderId).IsUnique();
        modelBuilder.Entity<ProcessedMessage>(b =>
        {
            b.HasKey(p => new { p.MessageId, p.Consumer });
            b.Property(p => p.MessageId).HasMaxLength(100);
            b.Property(p => p.Consumer).HasMaxLength(200);
            b.HasIndex(p => p.ProcessedAt);
        });
    }
}

The handler writes the inbox row in the same transaction as the business data, and before the work:

public sealed class InboxInvoiceHandler(BillingDbContext db, FakePaymentGateway payments)
{
    private const string ConsumerName = "billing.invoice";

    public async Task<string> HandleAsync(string messageId, OrderPlaced message)
    {
        await using var tx = await db.Database.BeginTransactionAsync();
        db.ProcessedMessages.Add(new ProcessedMessage { MessageId = messageId, Consumer = ConsumerName });
        try
        {
            // write the inbox row first: the second copy waits here until the first transaction ends, then gets a unique violation
            await db.SaveChangesAsync();
        }
        catch (DbUpdateException ex) when (ex.InnerException is Npgsql.PostgresException
                                           { SqlState: Npgsql.PostgresErrorCodes.UniqueViolation })
        {
            await tx.RollbackAsync();
            return "duplicate (inbox), skipped";
        }

        await payments.ChargeAsync(message);
        db.Invoices.Add(new Invoice { OrderId = message.OrderId, Amount = message.Total, Currency = message.Currency });
        await db.SaveChangesAsync();
        await tx.CommitAsync();
        return "invoice created";
    }
}

delivery tag 1 message msg-7f3a: invoice created
delivery tag 2 message msg-7f3a: duplicate (inbox), skipped
invoices for order: 1, charges: 1

Here is why the order matters: when the second delivery tries to insert the same primary key, PostgreSQL does not reject it immediately, it waits for the outcome of the first transaction. When the first commits, the second gets 23505 (unique violation); had the first rolled back, the second would have carried on and done the work. That is what closes the gap check-then-act could not. On a duplicate the consumer acks the message: the work is already done, there is no point in sending it to the DLQ. The transaction itself is the BeginTransactionAsync covered in the "Transactions in EF Core" post of the EF Core series; both SaveChangesAsync calls stay in the same transaction.

This approach has a price: the transaction and the lock on the inbox row stay open for the duration of the payment call. If the outgoing call takes long, database connections and locks are held long too. For critical side effects such as payments, the more robust way is to send the provider an idempotency key; we will discuss it below.

Natural Idempotency: A Unique Business Key and ON CONFLICT

The inbox table catches "the same message". But if the Order API (Application Programming Interface) publishes the same order twice with a new MessageId because an HTTP (Hypertext Transfer Protocol) request was retried, the inbox sees two different messages. If the work itself has a natural key (an order has exactly one invoice), using it is a stronger protection:

public sealed class UpsertInvoiceHandler(BillingDbContext db, FakePaymentGateway payments)
{
    public async Task<string> HandleAsync(OrderPlaced message)
    {
        await using var tx = await db.Database.BeginTransactionAsync();
        var invoice = new Invoice { OrderId = message.OrderId, Amount = message.Total, Currency = message.Currency };
        // natural idempotency: OrderId is unique, do nothing on conflict
        int inserted = await db.Database.ExecuteSqlInterpolatedAsync($"""
            INSERT INTO "Invoices" ("Id", "OrderId", "Amount", "Currency", "CreatedAt")
            VALUES ({invoice.Id}, {invoice.OrderId}, {invoice.Amount}, {invoice.Currency}, {invoice.CreatedAt})
            ON CONFLICT ("OrderId") DO NOTHING
            """);
        if (inserted == 0)
            return "duplicate (unique OrderId), skipped";

        await payments.ChargeAsync(message);
        await tx.CommitAsync();
        return "invoice created";
    }
}

delivery tag 1 message msg-7f3a: invoice created
delivery tag 2 message msg-7f3a: duplicate (unique OrderId), skipped
invoices for order: 1, charges: 1

ExecuteSqlInterpolatedAsync turns the interpolations into parameters, so there is no SQL (Structured Query Language) injection risk. ON CONFLICT DO NOTHING throws no exception and returns 0 affected rows; the control flow does not rely on exceptions. In SQL Server you build the same idea by catching error codes 2627/2601 on a unique index, or with MERGE. The payment call is still inside the transaction: if ChargeAsync throws, the transaction rolls back, the invoice row disappears and when the retry arrives the work starts from scratch.

Another form of natural idempotency is a state machine check. If a payment moves from Pending to Paid, you write the update with a WHERE Status = 'Pending' condition; a second PaymentCaptured message updates 0 rows, which means "already done". The same check also solves the "old must not overwrite new" problem from the previous post: a late OrderPlaced for an order that is already Shipped changes nothing.

Which One When?

They are not alternatives to each other. For additive work without a natural key, such as "take 1 off the stock", the inbox is your only protection; for work with a natural key, such as an invoice, always add the unique constraint and, if you like, the inbox on top.

Side Effects Do Not Join the Transaction

A database transaction protects only the database. An HTTP request to the payment provider, an e-mail that has been sent or a call to another service is not rolled back. In the inbox example, if the commit fails after the payment was taken, the message comes again, there is no inbox row, and the payment is taken a second time. There are three answers. First, send an idempotency key to the external service: if you put OrderId into the payment request, the provider does not treat a second request with the same key as a new payment (the provider has to support this). Second, trigger the side effect after your own transaction, through a separate message: you write an outbox row together with the invoice row, and the consumer of that message sends the e-mail; that is the topic of the next post. Third, if the business can accept it, choose "at most once": an "order received" e-mail that occasionally never goes out may be better than one that goes out twice. Which of these is acceptable is a business decision, not a technical one.

Cleaning Up Dedupe Records

The inbox table cannot grow forever. How long you need to keep a record depends on how late the same message can arrive again: the retry chain, time spent waiting in the DLQ and the replay window. In the previous post we gave the DLQ a 14-day TTL (Time-To-Live); if you keep inbox records shorter than that, a message replayed on day 10 is not found in the inbox and is processed again. A job that cleans up in small batches:

public sealed class ProcessedMessageCleaner(BillingDbContext db)
{
    public async Task RunAsync(TimeSpan retention, int batchSize, CancellationToken ct = default)
    {
        DateTimeOffset cutoff = DateTimeOffset.UtcNow - retention;
        int total = 0, deleted;
        do
        {
            // small batches: one huge DELETE should not keep the table and the WAL busy for long
            deleted = await db.ProcessedMessages
                .Where(p => p.ProcessedAt < cutoff)
                .OrderBy(p => p.ProcessedAt)
                .Take(batchSize)
                .ExecuteDeleteAsync(ct);
            total += deleted;
        } while (deleted == batchSize);
        Console.WriteLine($"deleted {total} processed-message rows older than {cutoff:yyyy-MM-dd}");
    }
}

I ran it with a 7-day retention against sample records written as if three had been processed 10 days ago and two 1 day ago (retention is 7 days for this demo; in a system with a 14-day DLQ TTL make it at least 14 days):

      DELETE FROM "ProcessedMessages" AS p
      WHERE EXISTS (
          SELECT 1
          FROM (
              SELECT p0."MessageId", p0."Consumer"
              FROM "ProcessedMessages" AS p0
              WHERE p0."ProcessedAt" < @cutoff
              ORDER BY p0."ProcessedAt"
              LIMIT @p
          ) AS p1
          WHERE p1."MessageId" = p."MessageId" AND p1."Consumer" = p."Consumer")
deleted 3 processed-message rows older than 2026-11-01

EF Core 10 translates an ExecuteDeleteAsync with Take on PostgreSQL into a WHERE EXISTS with a LIMITed subquery; the index on ProcessedAt keeps that query fast. Deleting in batches prevents a single huge DELETE from locking the table for a long time and bloating the WAL (Write-Ahead Log). To avoid running this job on several instances at once, you can use the distributed lock approaches from the "Jobs Across Multiple Instances" post of the upcoming Background Jobs series.

The MassTransit Inbox

MassTransit offers the same idea ready-made: the consumer side of its EF Core outbox keeps an inbox and does not run the consumer when the same MessageId reaches the same consumer a second time. The setup has three parts:

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<BookstoreDbContext>(o =>
    {
        o.UsePostgres();
        o.UseBusOutbox();
        o.QueryDelay = TimeSpan.FromSeconds(1);
        o.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
    });
    x.AddConsumer<InvoiceConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", 17301, "/", h => { h.Username("guest"); h.Password("guest"); });
        cfg.ReceiveEndpoint("billing-invoice", e =>
        {
            // inbox + consumer outbox: if the same MessageId arrives again, the consumer does not run
            e.UseEntityFrameworkOutbox<BookstoreDbContext>(context);
            e.ConfigureConsumer<InvoiceConsumer>(context);
        });
    });
});

public sealed class InvoiceConsumer(BookstoreDbContext db, ILogger<InvoiceConsumer> logger) : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        db.Invoices.Add(new Invoice { OrderId = context.Message.OrderId, Amount = context.Message.Total });
        // the outbox middleware calls SaveChanges: invoice + InboxState in the same transaction
        logger.LogInformation("Invoice for order {OrderId}, MessageId {MessageId}",
            context.Message.OrderId.ToString()[^8..], context.MessageId?.ToString()[..8]);
        await Task.CompletedTask;
    }
}

The DbContext needs MassTransit's tables:

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.AddInboxStateEntity();
    modelBuilder.AddOutboxMessageEntity();
    modelBuilder.AddOutboxStateEntity();
}

I published the same message twice with the same MessageId, 1.5 seconds apart:

05:45:08.368 info: Bookstore.Billing.InvoiceConsumer[0] Invoice for order 589e1d70, MessageId d28d0000
05:45:11.257 info: Lab[0] Invoices 1; InboxState MessageId=d28d0000 Consumed=2026-11-08T02:45:08.3700370Z Delivered=2026-11-08T02:45:08.4079860Z

The consumer ran once, the second delivery was acked silently, and the InboxState table records when the message was consumed and delivered. Note that the consumer does not call SaveChangesAsync: it is the middleware that saves the invoice and the inbox record in the same transaction. DuplicateDetectionWindow determines how long inbox records are kept; the retention rule above applies here too. Messages the consumer publishes are written to the outbox table in the same transaction and sent after the commit; that part is in the next post. The 8.5.x versions of this package target EF Core 9; we ran it with EF Core 10.0.14 and Npgsql 10.0.3 without problems.

Production Notes

Check where the MessageId comes from. The inbox relies on the publisher giving every message a unique MessageId that stays stable across retries. If the publisher generates a new Guid on retry, the inbox catches nothing. The contract side is in the sixth post of the series; in short: the MessageId is generated once when the message is created and never changes on resend.

Catch the unique violation in the right place. Only a primary key violation on the inbox table means "duplicate"; a unique violation on another table may be a real bug. In PostgreSQL, checking which constraint it was with PostgresException.ConstraintName is more robust.

Count your connection pool. Consumer concurrency multiplied by a transaction per message and the duration of the outgoing call quickly drives up the number of database connections. With concurrency 20 and a 2-second payment call, 20 connections stay busy under load.

Measure it. Publish the duplicate count as a metric. A sudden jump in this normally low and steady number shows that a publisher has entered a retry loop or that a replay was too broad.

Common Pitfalls

Idempotency via check-then-act. Two invoices, as above; without a unique constraint there is no guarantee at the database level.

Writing the inbox record in a separate transaction. Writing to the inbox, committing and then doing the work means that when the work fails the message counts as "processed" and is never tried again. The inbox row and the business data must be in the same transaction.

Keying the inbox on MessageId alone. When the same message is processed by two different consumers, one's record makes the other skip it. Add the consumer name to the key.

Keeping dedupe records shorter than the replay window. A 14-day DLQ with a 1-day inbox means every message replayed after day 2 is processed again.

Assuming the transaction protects external side effects. A rollback does not unsend an e-mail; you need an idempotency key, an outbox or a deliberate at-most-once decision.

A Small Experiment

★☆☆ Narrow the exception caught in the inbox handler with ConstraintName: treat it as a duplicate only if it is the primary key of the ProcessedMessages table, otherwise rethrow. See the difference with a test message that violates the unique index on the invoice table.

★★☆ Set up the scenario where the Order API publishes the same order with two different MessageIds. Run the inbox handler and the upsert handler separately and verify from the output which one issues two invoices.

★★★ Add a Payment entity (Pending, Captured, Refunded) and write the PaymentCaptured consumer by tying the state transition to a WHERE Status = Pending condition with ExecuteUpdateAsync. Send the same message three times, plus an out-of-order PaymentRefunded, and show that the final state stays correct.

We have hardened the consumer side. But there is a question at the other end too: the Order API wrote the order to the database, then could not reach the broker while publishing the message. The order exists, the message does not. In the next post we write a transactional outbox ourselves with EF Core and a BackgroundService relay, then look at MassTransit's built-in outbox.