How to Prevent Duplicate Message Processing with Inbox Pattern

Share this post

Some of the most expensive bugs in distributed systems happen when the same message is processed twice.

At-least-once delivery is not a flaw in your messaging infrastructure. It’s a deliberate trade-off: the broker guarantees your message will arrive, but it doesn’t guarantee it will arrive exactly once. The same message can be delivered twice — after a consumer crash before ack, after a network hiccup, after a broker restart. That’s the contract.

The problem is that most business operations aren’t safe to run twice. Booking a carrier pickup slot, charging a payment, sending a notification email — these have real-world consequences. Run them twice and you have a double-booked slot, a double charge, an annoyed customer.

This is the gap the Inbox Pattern fills. Before processing any message, you record its ID in a database table. If you’ve seen it before, you discard it. If you haven’t, you process it — atomically, in the same transaction as the record. The result: at-least-once delivery in, exactly-once side effects out.

If you’re coming from the previous post on the Outbox Pattern, this is the consumer-side complement: the Outbox guarantees your events leave the database; the Inbox guarantees they’re processed exactly once when they arrive. Together, they form a complete reliability story.

Throughout this post we’ll use a Parcel Tracking domain as the concrete example — a service that creates parcels, publishes a ParcelCreatedEvent when one is registered, and has downstream consumers that react to it. The stack is .NET 10, EF Core, PostgreSQL, RabbitMQ, and MassTransit.

Introduction: Why Idempotency is Hard and What Goes Wrong Without It

Duplicate messages are not an edge case. They’re a structural property of at-least-once delivery, and at-least-once delivery is what every real broker gives you by default — RabbitMQ, Kafka, Azure Service Bus, all of them.

Here’s how duplicates happen even with a correct implementation. Say a parcel is registered in your system and a ParcelCreatedEvent is published to RabbitMQ — a message that tells downstream consumers a new parcel exists and work needs to happen:

ParcelCreatedEvent arrives at the consumer.
The consumer processes it successfully — books a carrier pickup slot, queues a notification email.
Before the consumer can ack the message, the process crashes.
RabbitMQ redelivers the unacknowledged message to the same or a different consumer instance.
The consumer processes it again.

From the broker’s perspective, this is correct behavior. From yours, you’ve just booked two carrier pickup slots for the same parcel.

The carrier API charges per booking. It’s not idempotent. You’ve now got two open pickup slots, two charges on the account, and a support ticket incoming.

Idempotency means: processing the same message twice produces the same outcome as processing it once. No double charges. No duplicate notifications. No phantom state transitions.

The naive approach is to add a guard in your consumer: check whether the work has already been done before doing it. Sometimes that’s possible — if you can inspect existing state to infer whether a message was already handled. But often you can’t. The carrier API doesn’t expose a “did I already book this?” query. The notification service doesn’t know whether an email was sent. And even when you can check, the check-then-act window introduces a race condition: two concurrent deliveries of the same message can both pass the check and both do the work.

The Inbox Pattern solves this at the infrastructure level, before your consumer code even runs.

How the Inbox Pattern Works

When a message arrives, check whether its ID already exists in an inbox_messages table. If it does, the message was already processed — acknowledge it to the broker and stop. If it doesn’t, insert the ID and execute your business logic in the same database transaction. Commit both together or neither.

In steps:

Message arrives at the consumer
Look up message_id in inbox_messages
Found → ack to broker, discard, done
Not found → open a transaction
Insert the inbox_messages record
Execute the business logic
Commit
Ack to broker

The delivery guarantee this buys you: at-least-once delivery in → exactly-once side effects out.

The steps look simple. The hard part is step 7: the commit must include both the inbox record and the business operation, or the whole thing falls apart.

The Atomic Requirement

The insert into inbox_messages and the business operation must happen in the same database transaction. This is non-negotiable.

Here’s what happens if you split them:

You insert the inbox record first, then process. Your process crashes between the two. On redelivery, the inbox record exists — so you skip processing. The business operation never happened.
You process first, then insert. Your process crashes between the two. On redelivery, there’s no inbox record — so you process again. Duplicate.

The only safe sequence: open transaction → insert inbox record → do business work → commit → ack.

If the commit fails, the broker redelivers. The consumer tries again, finds no inbox record (the transaction rolled back), and processes normally. If the commit succeeds, any subsequent redelivery finds the inbox record and discards the duplicate. There’s no window where state is inconsistent.

Why `consumer_type` Matters

The same ParcelCreatedEvent might be consumed by multiple services — a NotificationConsumer that sends an SMS, and a CarrierAssignmentConsumer that calls the carrier API. Idempotency needs to be scoped per consumer, not per message.

If you deduplicate only on message ID, and both consumers share an inbox table, the first consumer to process a message “claims” the ID for everyone. The second consumer finds the record and skips its own processing. That’s wrong.

The composite key (message_id, consumer_type) gives each consumer its own independent idempotency scope.

Broker Independence

The pattern is broker-agnostic. Whether your duplicates come from RabbitMQ redelivery, Kafka, Azure Service Bus, or any other — the inbox doesn’t care. It’s a database-level guarantee, and the database is always in scope.

Implementation in .Net

Project Structure

Before writing any code, let’s be explicit about where everything lives.

This post builds on an existing solution with the following projects already in place:

ParcelTracker.Domain/          ← domain events, entities
ParcelTracker.Infrastructure/  ← DbContext, EF Core configurations, migrations
ParcelTracker.Api/             ← HTTP endpoints, producer side
ParcelTracker.Worker/          ← background services
ParcelTracker.Consumer/        ← new in this post: MassTransit consumers

The inbox-related code is split across two projects, following the same convention the rest of the solution uses:

ParcelTracker.Infrastructure/ — anything EF Core touches goes here. The InboxMessage entity, its EF configuration, and the AppDbContext addition all live alongside your existing configurations:

ParcelTracker.Infrastructure/
└── Configurations/
    ├── InboxMessage.cs                     // the entity
    └── InboxMessageConfiguration.cs        // EF Core mapping

ParcelTracker.Consumer/ — MassTransit consumers, the inbox filter, and the service abstractions live here:

ParcelTracker.Consumer/
├── Consumers/
│   ├── CarrierAssignmentConsumer.cs
│   └── NotificationConsumer.cs
├── Inbox/
│   └── IdempotencyFilter.cs                // MassTransit pipeline filter
├── Services/
│   ├── ICarrierApiClient.cs
│   ├── INotificationService.cs
│   ├── StubCarrierApiClient.cs
│   ├── StubNotificationService.cs
│   └── CarrierSlot.cs
└── Cleanup/
    └── InboxCleanupService.cs  // background service, deletes old inbox records

Service Interfaces and Stubs

The two consumers depend on INotificationService and ICarrierApiClient. This post is about idempotency, not about building real carrier or notification integrations — so we’ll define the interfaces and provide stub implementations that are enough to compile, run, and verify the pattern works.

In production you’d replace the stubs with real implementations: an HTTP client wrapping your carrier’s API, an email/SMS provider, and so on. The consumers don’t change — only the registrations do.

Services/INotificationService.cs

public interface INotificationService
{
    Task SendParcelCreatedAsync(
        Guid   parcelId,
        string trackingNumber,
        string destination,
        CancellationToken ct = default);
}

Services/ICarrierApiClient.cs

public sealed record CarrierSlot(Guid SlotId, DateTimeOffset PickupTime);

public interface ICarrierApiClient
{
    Task<CarrierSlot> ReservePickupSlotAsync(
        Guid   parcelId,
        string destination,
        CancellationToken ct = default);
}

Services/StubNotificationService.cs

public sealed class StubNotificationService : INotificationService
{
    private readonly ILogger<StubNotificationService> _logger;

    public StubNotificationService(ILogger<StubNotificationService> logger)
        => _logger = logger;

    public Task SendParcelCreatedAsync(
        Guid   parcelId,
        string trackingNumber,
        string destination,
        CancellationToken ct = default)
    {
// In production: send an SMS or email via your provider of choice.
// Here we log so the behaviour is visible in tests and local runs.
        _logger.LogInformation(
            "[STUB] Notification sent. ParcelId={ParcelId} TrackingNumber={TrackingNumber} Destination={Destination}",
            parcelId, trackingNumber, destination);

        return Task.CompletedTask;
    }
}

Services/StubCarrierApiClient.cs

public sealed class StubCarrierApiClient : ICarrierApiClient
{
    private readonly ILogger<StubCarrierApiClient> _logger;

    public StubCarrierApiClient(ILogger<StubCarrierApiClient> logger)
        => _logger = logger;

    public Task<CarrierSlot> ReservePickupSlotAsync(
        Guid   parcelId,
        string destination,
        CancellationToken ct = default)
    {
// In production: POST to the carrier API and parse the response.
// This stub returns a deterministic slot so tests can assert on it.
        var slot = new CarrierSlot(
            SlotId:      Guid.NewGuid(),
            PickupTime:  DateTimeOffset.UtcNow.AddHours(24));

        _logger.LogInformation(
            "[STUB] Carrier slot reserved. ParcelId={ParcelId} SlotId={SlotId}",
            parcelId, slot.SlotId);

        return Task.FromResult(slot);
    }
}

builder.Services.AddScoped<INotificationService, StubNotificationService>();
builder.Services.AddScoped<ICarrierApiClient, StubCarrierApiClient>();

With the interfaces and stubs in place, the consumers compile and run. When you’re ready to replace a stub with a real implementation, swap the registration — nothing else changes.

The ParcelCreatedEvent contract already lives in ParcelTracker.Domain/Events/ — both the producer and consumer reference it from there. No duplication needed.

With the structure clear, let’s build it out.

The InboxMessage Schema

CREATE TABLE inbox_messages (
    id            UUID          NOT NULL,
    consumer_type VARCHAR(500)  NOT NULL,
    received_at   TIMESTAMPTZ   NOT NULL,
    processed_at  TIMESTAMPTZ   NULL,

    CONSTRAINT pk_inbox_messages PRIMARY KEY (id, consumer_type)
);

CREATE INDEX ix_inbox_messages_unprocessed
    ON inbox_messages (received_at)
    WHERE processed_at IS NULL;

A few deliberate decisions worth noting:

The primary key is (id, consumer_type), not just id. This enforces per-consumer uniqueness at the database level. Even if your application logic has a bug, the constraint won’t let a duplicate through.

processed_at is a timestamp, not a boolean. A timestamp tells you when it was processed; a boolean tells you nothing. Useful for auditing and for the cleanup job that runs later.

id is the broker’s message ID, not a surrogate. You don’t generate this — MassTransit provides it via ConsumeContext.MessageId. This is what makes deduplication work: the broker’s ID is stable across redeliveries of the same message.

The partial index on processed_at IS NULL is for the cleanup job, not for deduplication. The deduplication path relies on the primary key constraint. The partial index keeps the cleanup query fast as the table grows.

The EF Core Model

Create InboxMessage in ParcelTracker.Infrastructure/Configurations/:

public class InboxMessage
{
    public Guid Id { get; private set; }
    public string ConsumerType { get; private set; } = string.Empty;
    public DateTime ReceivedAt { get; private set; }
    public DateTime? ProcessedAt { get; private set; }

    public InboxMessage() { }

    // New constructor for external creation
    public InboxMessage(Guid id, string consumerType)
    {
        Id = id;
        ConsumerType = consumerType;
        ReceivedAt = DateTime.UtcNow;
    }

    public static InboxMessage For<TConsumer>(Guid messageId)
    {
        return new InboxMessage(messageId, typeof(TConsumer).FullName ?? throw new InvalidOperationException($"Cannot determine type name for {typeof(TConsumer).Name}."));
    }

    public void MarkAsProcessed() => ProcessedAt = DateTime.UtcNow;
}

The constructor takes the message ID and consumer type directly — both are supplied by the filter, which derives them from the consume context. The entity stays ignorant of how it’s being used.

The EF Core Configuration

internal sealed class InboxMessageConfiguration : IEntityTypeConfiguration<InboxMessage>
{
    public void Configure(EntityTypeBuilder<InboxMessage> builder)
    {
        builder.ToTable("inbox_messages");

        builder.HasKey(x => new { x.Id, x.ConsumerType });

        builder.Property(x => x.Id)
            .HasColumnName("id")
            .IsRequired();

        builder.Property(x => x.ConsumerType)
            .HasColumnName("consumer_type")
            .HasMaxLength(500)
            .IsRequired();

        builder.Property(x => x.ReceivedAt)
            .HasColumnName("received_at")
            .IsRequired();

        builder.Property(x => x.ProcessedAt)
            .HasColumnName("processed_at")
            .IsRequired(false);

        builder.HasIndex(x => x.ReceivedAt)
            .HasFilter("processed_at IS NULL")
            .HasDatabaseName("ix_inbox_messages_unprocessed");
    }
}

The composite key maps to the composite primary key in the schema. EF Core handles the uniqueness constraint automatically — no need to add a separate .HasAlternateKey().

The DB Context

The inbox table must live in the same DbContext as your business data. This is the atomicity requirement from the previous section made concrete: EF Core can only wrap the inbox insert and the business operation in a single transaction if they share a context. A separate InboxDbContext breaks that guarantee.

If you’re starting fresh, here’s the full context. If you’re continuing from the Outbox Pattern post, add the InboxMessages line to what you already have.

public class AppDbContext : DbContext
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

    // Domain entities — your business data
    public DbSet<Parcel> Parcels => Set<Parcel>();

    // Outbox table — written by the producer side to guarantee durable publishing
    // (covered in the Outbox Pattern post; include if you're building the full pipeline)
    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    // Inbox table — written by the consumer side to guarantee idempotent processing
    public DbSet<InboxMessage> InboxMessages => Set<InboxMessage>();

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

The Idempotency Filter

The idempotency check shouldn’t live inside each consumer — that’s a cross-cutting concern, and duplicating the logic across every consumer guarantees someone forgets it. Instead, we implement it as a MassTransit pipeline filter that wraps every consumer transparently.

How MassTransit’s Pipeline Works

MassTransit is a .NET message bus abstraction that sits on top of brokers like RabbitMQ, Azure Service Bus, and others. When a message arrives on a queue, MassTransit doesn’t hand it directly to your consumer — it runs it through a configurable middleware pipeline first. That pipeline is where cross-cutting concerns live: retry logic, scoped DI, error handling, and — in our case — idempotency.

Each middleware step implements IFilter<ConsumeContext<T>>. You get the raw consume context (which carries the message, its headers, the broker acknowledgement handle, and attached scoped services), a reference to the next step in the chain, and full control over whether processing continues. Call next.Send() to pass the message forward. Don’t call it and the consumer never runs.

The idempotency filter sits at the front of that pipeline. If the message is a duplicate, it short-circuits. If it’s new, it opens a transaction, inserts the inbox record, calls next.Send() to run the consumer, and commits — all in one unit of work.

public sealed class IdempotencyFilter<TMessage> : IFilter<ConsumeContext<TMessage>>
where TMessage : class
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<IdempotencyFilter<TMessage>> _logger;

    public IdempotencyFilter(ILogger<IdempotencyFilter<TMessage>> logger, IServiceScopeFactory scopeFactory)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
    }

    public async Task Send(
        ConsumeContext<TMessage> context,
        IPipe<ConsumeContext<TMessage>> next)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        var messageId = context.MessageId
            ?? throw new InvalidOperationException(
                   "Message does not carry a MessageId. Cannot enforce idempotency.");

        // Get consumer type from the context
        var consumerTypeName = context.ReceiveContext.InputAddress?.Segments.Last() ?? "Unknown";

        var alreadyProcessed = await db.InboxMessages.AnyAsync(
            m => m.Id == messageId &&
                 m.ConsumerType == consumerTypeName,
            context.CancellationToken);

        if (alreadyProcessed)
        {
            _logger.LogWarning(
                "Duplicate message detected. MessageId={MessageId} Consumer={Consumer}. Discarding.",
                messageId, consumerTypeName);
            return;
        }

        await using var transaction =
            await db.Database.BeginTransactionAsync(context.CancellationToken);

        try
        {
            var inboxMessage = new InboxMessage(messageId, consumerTypeName);
            db.InboxMessages.Add(inboxMessage);

            await next.Send(context);

            await db.SaveChangesAsync(context.CancellationToken);
            await transaction.CommitAsync(context.CancellationToken);
        }
        catch
        {
            await transaction.RollbackAsync(context.CancellationToken);
            throw;
        }
    }

    public void Probe(ProbeContext context) =>
        context.CreateFilterScope("idempotency");
}

A few deliberate decisions worth noting:

The filter takes IServiceScopeFactory, not AppDbContext directly. MassTransit filters are registered as singletons by default. Injecting a scoped AppDbContext directly would cause a captive dependency — the same context instance reused across messages. IServiceScopeFactory creates a fresh scope per message, which means a fresh AppDbContext per message.

The duplicate check uses AnyAsync, not FindAsync. We’re checking existence, not retrieving an entity to work with. AnyAsync lets the database stop scanning the moment it finds a match and returns a boolean — no unnecessary materialization.

If next.Send() throws, the transaction rolls back. The inbox record is never persisted. On redelivery, there’s no record — the consumer tries again. This is correct: a message that failed processing is not a duplicate, it’s a retry candidate.

The duplicate is logged at Warning, not Debug. A steady trickle of duplicates is normal. A sudden spike is a sign something upstream is wrong — bad retry configuration, a retry loop, or a bug in the publisher retrying messages that were already delivered. You want that signal to surface.

Registering the Filter

builder.Services.AddScoped(typeof(IdempotencyFilter<>));

Wiring up to MassTransit

MassTransit provides UseConsumeFilter to attach a filter to a specific consumer’s pipeline. Register it in your AddMassTransit configuration:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<NotificationConsumer>();
    x.AddConsumer<CarrierAssignmentConsumer>();

    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host(builder.Configuration["RabbitMq:Host"], "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.ReceiveEndpoint("parcel-created-notification", e =>
				{
				    e.UseMessageScope(ctx);
				    e.UseConsumeFilter(typeof(IdempotencyFilter<ParcelCreatedEvent>), ctx);
				    e.ConfigureConsumer<NotificationConsumer>(ctx);
				});
				
				cfg.ReceiveEndpoint("parcel-created-carrier", e =>
				{
				    e.UseMessageScope(ctx);
				    e.UseMessageRetry(r => r.Interval(3, TimeSpan.FromSeconds(5))); // retry 3 times with 5 seconds interval
				    e.UseConsumeFilter(typeof(IdempotencyFilter<ParcelCreatedEvent>), ctx);
				    e.ConfigureConsumer<CarrierAssignmentConsumer>(ctx);
				});
    });
});

The two receive endpoints correspond to the two consumers — each gets its own queue and its own idempotency scope in the inbox table.

Filter Pipeline Order

Order matters in MassTransit’s pipeline. The general rule:

UseMessageScope → UseMessageRetry → UseConsumeFilter (idempotency) → Consumer

UseMessageScope must come first — it sets up the DI scope. UseMessageRetry should come before the idempotency filter so that transient failures (database blip, lock timeout) can retry without being logged as duplicates. The idempotency filter sits closest to the consumer.

The Consumers

With the filter in place, the consumers themselves have no idempotency code at all. They do their job. The infrastructure handles the rest.

Both consumers implement the IConsumer interface and simply return Task:

public sealed class NotificationConsumer : IConsumer<ParcelCreatedEvent>
{
    private readonly INotificationService _notifications;
    private readonly ILogger<NotificationConsumer> _logger;

    public NotificationConsumer(
        INotificationService notifications,
        ILogger<NotificationConsumer> logger)
    {
        _notifications = notifications;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<ParcelCreatedEvent> context)
    {
        var evt = context.Message;

        await _notifications.SendParcelCreatedAsync(
            parcelId: evt.ParcelId,
            trackingNumber: evt.TrackingNumber,
            destination: evt.Destination,
            ct: context.CancellationToken);

        _logger.LogInformation(
            "Notification sent for parcel {ParcelId}.", evt.ParcelId);
    }
}

public sealed class CarrierAssignmentConsumer : IConsumer<ParcelCreatedEvent>
{
    private readonly ICarrierApiClient _carrierApi;
    private readonly ILogger<CarrierAssignmentConsumer> _logger;

    public CarrierAssignmentConsumer(
        ICarrierApiClient carrierApi,
        ILogger<CarrierAssignmentConsumer> logger)
    {
        _carrierApi = carrierApi;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<ParcelCreatedEvent> context)
    {
        var evt = context.Message;

        // The carrier API is not idempotent — calling it twice books two slots.
        // The inbox filter guarantees this method is called at most once per MessageId.
        var slot = await _carrierApi.ReservePickupSlotAsync(
            parcelId: evt.ParcelId,
            destination: evt.Destination,
            ct: context.CancellationToken);

        _logger.LogInformation(
            "Carrier pickup slot {SlotId} reserved for parcel {ParcelId}.",
            slot.SlotId, evt.ParcelId);
    }
}

CarrierAssignmentConsumer would be broken without the inbox filter. With it, the comment in the method body is informational, not a warning.

Production Considerations

Inbox Cleanup

Processed inbox rows accumulate. They don’t need to stay around for long — once a message is processed and the broker has moved on, the record’s only purpose is to catch late redeliveries of the same message. The right retention window is roughly: how long can the broker redeliver a message after it’s been acked?

For most RabbitMQ configurations, that’s zero — acked messages are gone. But with dead-letter queues, scheduled requeue, or external message replay, you might see redeliveries hours later. A 7-day window is a safe default for most teams.

public sealed class InboxCleanupService : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<InboxCleanupService> _logger;
    private readonly TimeSpan _interval = TimeSpan.FromHours(24);
    private readonly TimeSpan _retention = TimeSpan.FromDays(7);

    public InboxCleanupService(
        IServiceScopeFactory scopeFactory,
        ILogger<InboxCleanupService> logger)
    {
        _scopeFactory = scopeFactory;
        _logger       = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            await Task.Delay(_interval, stoppingToken);

            try
            {
                await RunCleanupAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Inbox cleanup failed.");
            }
        }
    }

    private async Task RunCleanupAsync(CancellationToken ct)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        var cutoff = DateTime.UtcNow - _retention;

        var deleted = await db.InboxMessages
            .Where(m => m.ProcessedAt != null && m.ProcessedAt < cutoff)
            .ExecuteDeleteAsync(ct);

        _logger.LogInformation("Inbox cleanup deleted {Count} processed records.", deleted);
    }
}

builder.Services.AddHostedService<InboxCleanupService>();

Observability

Duplicates are signal. A low, steady rate is expected — one or two per hour in a busy system is noise. A spike is a bug.

Log every discard at Warning with the message ID and consumer type. That’s already in the filter. Add a structured counter so you can alert on it:

// In the filter, after the duplicate detection branch:
// consumerTypeName is already derived from the receive endpoint address
_logger.LogWarning(
    "Inbox duplicate discarded. {MessageId} {ConsumerType}",
    messageId, consumerTypeName);

// Wire up your metrics provider of choice — this shows the shape:
// metrics.IncrementCounter("inbox.duplicates_discarded",
//     tags: new[] { ("consumer", consumerTypeName) });

A spike in inbox.duplicates_discarded means something upstream is misbehaving — a retry loop, a misconfigured dead-letter queue, or a publisher retrying messages that were already delivered.

Two metrics give you a complete picture of inbox health:

inbox.duplicates_discarded — a counter. Steady is fine; a spike is a problem.
inbox_messages table row count — a gauge. Growing without bound means cleanup isn’t running.

Conclusion: The full reliability loop

The Outbox Pattern ensures events are durably published from the producer — writing to the database and the outbox in a single transaction, with a background dispatcher that guarantees delivery to the broker even across crashes. The Inbox Pattern, which you now have, ensures they’re processed exactly once on the consumer side. Together, they close the loop.

[API Endpoint]
      │
      │  Single DB transaction
      ├─ INSERT parcels
      └─ INSERT outbox_messages
             │
             │  OutboxProcessor (background)
             ▼
        [RabbitMQ]
             │
             │  MassTransit consumer pipeline
             ▼
    [IdempotencyFilter]
             │
             │  Single DB transaction
             ├─ INSERT inbox_messages
             └─ Execute consumer logic

The producer side guarantees your events leave the database. The consumer side guarantees they’re processed exactly once, regardless of how many times they arrive. Neither side relies on the broker being reliable, the network being stable, or processes staying alive. The database is the only thing that has to work — and it’s already the most reliable component in your stack.

What You Now Have

End-to-end exactly-once semantics over at-least-once transport — the full reliability loop, producer to consumer.
Per-consumer idempotency — NotificationConsumer and CarrierAssignmentConsumer each have independent deduplication scope. Same message, different consumers, both process correctly.
Infrastructure-level protection — your consumer code has no idempotency logic. It can’t accidentally forget to check. The filter enforces it unconditionally.
A database-level hard stop — even if the application logic has a bug, the unique constraint on (id, consumer_type) won’t let a duplicate through.

What This Doesn’t Solve

Ordering. Neither the Outbox Pattern nor the Inbox Pattern makes ordering guarantees. If two different events for the same parcel arrive out of order, the inbox will process both — in whatever order the broker delivers them
Non-database side effects mid-transaction. The inbox filter wraps your consumer in a database transaction — but HTTP calls, emails, and other external operations aren’t transactional. If your consumer calls the carrier API and then the database commit fails, the carrier slot is booked but your local state says it isn’t. The safe approach: treat external calls as the last step, after the commit. Structure your consumer so the database work commits first, and non-transactional side effects follow. Getting this right in complex flows is the domain of saga patterns — a topic for a future post.

Cover photo by Jan van der Wolf