Idempotent Consumer ve Inbox: Aynı Mesaj İki Kez Geldiğinde

Sürüm notu: Bu yazıdaki kodlar .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 ve MassTransit 8.5.12 ile derlenip denenmiştir.

Önce Tahmin Et

Billing consumer'ı bir OrderPlaced mesajı için fatura kesip ödemeyi alıyor. Aynı mesaj (aynı MessageId, aynı sipariş) iki kez geliyor; consumer channel'ı aynı anda 4 mesaj işleyebiliyor. Handler'ın ilk satırı şu kontrol:

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

İşlem bittiğinde veritabanında bu sipariş için kaç fatura var ve müşteriden kaç kez para çekildi?

Aynı mesajın iki kez gelmesi bir hata değil, at-least-once teslimatın doğal sonucu: consumer işi bitirip ack'i göndermeden çöker, publisher confirm gelmeyince aynı mesajı yeniden yayınlar, outbox relay'i yayınladıktan sonra satırı işaretleyemeden durur, bir operatör DLQ'dan (Dead-Letter Queue) replay eder. Bu serinin önceki iki yazısında retry ve replay'in her birinin "en az bir kopya fazla" ürettiğini gördük. Idempotent consumer'ın kavramsal tarafı Messaging ve Event-Driven Architecture serisinin "Idempotent Consumer" yazısında; burada EF Core ve PostgreSQL ile kodunu yazıyoruz.

Check-Then-Act Neden Yetmiyor?

Yukarıdaki kontrolün tamamı:

public sealed class NaiveInvoiceHandler(BillingDbContext db, FakePaymentGateway payments)
{
    public async Task<string> HandleAsync(OrderPlaced message)
    {
        // check-then-act: iki kopya aynı anda buradan geçerse ikisi de "yok" görür
        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";
    }
}

Sahte ödeme servisi 200 ms sürüyor. Consumer channel'ı consumerDispatchConcurrency: 4 ile açıldı ve aynı mesajı iki kez yayınladım (bu demo için OrderId üzerindeki unique index'i kaldırdım):

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

İki fatura, iki ödeme. İki teslim de AnyAsync'i aynı milisaniyelerde çalıştırdı, ikisi de "fatura yok" gördü ve ikisi de devam etti. Concurrency 1 olsaydı bu demo şanslı çıkardı, ama üretimde aynı queue'yu dinleyen iki instance olduğunda tek channel'ın sırası seni korumaz. Kontrol ile yazma arasındaki boşluğu kapatabilen tek yer veritabanının kendisi: bir unique constraint.

Inbox Tablosu: MessageId ile Dedupe

İlk çözüm, işlenen her mesajı bir tabloya yazmak. Anahtar MessageId ile consumer adının birleşimi; aynı mesaj Billing'de ve Notification'da ayrı ayrı işlenmeli, biri diğerinin kaydını görüp atlamamalı:

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);
        });
    }
}

Handler, inbox satırını iş verisiyle aynı transaction'da ve işten önce yazıyor:

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
        {
            // inbox satırını önce yaz: ikinci kopya burada ilk transaction bitene kadar bekler, sonra unique violation alır
            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

Sıralamanın önemi şurada: ikinci teslim aynı primary key'i eklemeye çalıştığında PostgreSQL onu hemen reddetmez, ilk transaction'ın sonucunu bekler. İlki commit edince ikincisi 23505 (unique violation) alır; ilki rollback etseydi ikincisi devam eder ve işi o yapardı. Bu, check-then-act'in kapatamadığı boşluğu kapatan şey. Duplicate durumunda consumer mesajı ack'ler: iş zaten yapılmış, mesajı DLQ'ya göndermenin anlamı yok. Transaction'ın kendisi EF Core serisinin "EF Core'da Transaction'lar" yazısında anlatılan BeginTransactionAsync; iki SaveChangesAsync çağrısı aynı transaction'da kalıyor.

Bu yaklaşımın bir bedeli var: ödeme çağrısı süresince transaction ve inbox satırının kilidi açık kalıyor. Dış çağrı uzarsa veritabanı bağlantıları ve kilitler de uzar. Ödeme gibi kritik yan etkilerde daha sağlam yol, sağlayıcıya bir idempotency key göndermek; bunu aşağıda konuşacağız.

Doğal Idempotency: Unique İş Anahtarı ve ON CONFLICT

Inbox tablosu "aynı mesajı" yakalar. Ama Order API (Application Programming Interface), HTTP (Hypertext Transfer Protocol) isteği tekrarlandığı için aynı siparişi yeni bir MessageId'yle iki kez yayınlarsa inbox iki farklı mesaj görür. İşin kendisinin doğal bir anahtarı varsa (bir siparişin tek faturası olur) onu kullanmak daha güçlü bir koruma:

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 };
        // doğal idempotency: OrderId unique, çakışmada hiçbir şey yapma
        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 interpolasyonları parametreye çevirir, yani SQL (Structured Query Language) injection riski yok. ON CONFLICT DO NOTHING exception fırlatmaz, etkilenen satır sayısını 0 döndürür; kontrol akışı exception'a dayanmıyor. SQL Server'da aynı fikri unique index ve 2627/2601 hata kodlarını yakalayarak ya da MERGE ile kurarsın. Ödeme çağrısı yine transaction'ın içinde: ChargeAsync exception fırlatırsa transaction rollback olur, fatura satırı kaybolur ve retry geldiğinde iş baştan yapılır.

Doğal idempotency'nin bir başka hâli state machine kontrolü. Bir ödemenin durumu Pending'den Paid'e geçiyorsa, güncellemeyi WHERE Status = 'Pending' koşuluyla yazarsın; ikinci PaymentCaptured mesajı 0 satır günceller ve bu "zaten yapıldı" demektir. Bu kontrol aynı zamanda bir önceki yazıda değindiğimiz "eski mesaj yeniyi ezmesin" sorununu da çözer: Shipped durumundaki bir siparişe geç gelen bir OrderPlaced hiçbir şeyi değiştirmez.

Hangisi Ne Zaman?

İkisi birbirinin alternatifi değil. "Stoktan 1 düş" gibi doğal anahtarı olmayan, toplamsal işlerde inbox tek koruman; fatura gibi doğal anahtarı olan işlerde unique constraint'i her durumda koy, inbox'ı da istersen üstüne ekle.

Yan Etkiler Transaction'a Girmez

Veritabanı transaction'ı yalnız veritabanını korur. Ödeme sağlayıcısına giden HTTP isteği, gönderilen e-posta ya da başka bir servise atılan çağrı rollback olmaz. Inbox örneğinde ödeme alındıktan sonra commit başarısız olursa, mesaj tekrar gelir, inbox satırı yoktur ve ödeme ikinci kez alınır. Bunun üç cevabı var. Birincisi, dış servise bir idempotency key göndermek: ödeme isteğine OrderId'yi koyarsan, sağlayıcı aynı anahtarla gelen ikinci isteği yeni bir ödeme olarak işlemez (sağlayıcının bunu desteklemesi gerekir). İkincisi, yan etkiyi kendi transaction'ından sonra ve ayrı bir mesajla tetiklemek: fatura satırıyla birlikte bir outbox satırı yazarsın, e-postayı o mesajın consumer'ı gönderir; bu bir sonraki yazının konusu. Üçüncüsü, iş açısından kabul edilebiliyorsa "en fazla bir kez" seçmek: bir "siparişiniz alındı" e-postasının nadiren hiç gitmemesi, iki kez gitmesinden daha iyi olabilir. Hangisinin kabul edilebilir olduğu teknik değil, iş kararı.

Dedupe Kayıtlarını Temizlemek

Inbox tablosu sonsuza kadar büyüyemez. Bir kaydı ne kadar tutman gerektiği, aynı mesajın en geç ne zaman tekrar gelebileceğine bağlı: retry zinciri, DLQ'da bekleme süresi ve replay penceresi. Bir önceki yazıda DLQ'ya 14 gün TTL (Time-To-Live) verdik; inbox kaydını bundan kısa tutarsan, 10. günde replay edilen bir mesaj inbox'ta bulunamaz ve tekrar işlenir. Temizliği küçük parçalarla yapan bir job:

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
        {
            // küçük parçalar: tek dev DELETE tabloyu ve WAL'ı uzun süre meşgul etmesin
            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}");
    }
}

Üçü 10 gün, ikisi 1 gün önce işlenmiş gibi yazılmış örnek kayıtlarla, 7 günlük saklama süresiyle çalıştırdım (yukarıdaki demo için retention 7 gün; DLQ TTL'i 14 gün olan bir sistemde bunu en az 14 gün yapmalısın):

      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, Take içeren bir ExecuteDeleteAsync'i PostgreSQL'de WHERE EXISTS ve LIMIT'li bir alt sorguya çeviriyor; ProcessedAt üzerindeki index bu sorgunun hızlı kalmasını sağlıyor. Parça parça silmek, tek bir dev DELETE'in tabloyu uzun süre kilitlemesini ve WAL'ı (Write-Ahead Log) şişirmesini önlüyor. Bu job'ı birden fazla instance'ta aynı anda çalıştırmamak için ilerideki Background Job'lar serisinin "Çok Instance'lı Ortamda Job'lar" yazısındaki distributed lock yaklaşımlarını kullanabilirsin.

MassTransit Inbox

MassTransit, aynı fikri hazır veriyor: EF Core outbox'ının consumer tarafı bir inbox tutar ve aynı MessageId aynı consumer'a ikinci kez geldiğinde consumer'ı çalıştırmaz. Kurulum üç parça:

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: aynı MessageId ikinci kez gelirse consumer çalışmaz
            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 });
        // SaveChanges'i outbox middleware çağırır: invoice + InboxState aynı transaction'da
        logger.LogInformation("Invoice for order {OrderId}, MessageId {MessageId}",
            context.Message.OrderId.ToString()[^8..], context.MessageId?.ToString()[..8]);
        await Task.CompletedTask;
    }
}

DbContext'e MassTransit'in tablolarını eklemek gerekiyor:

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

Aynı mesajı aynı MessageId ile 1,5 saniye arayla iki kez yayınladım:

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

Consumer bir kez çalıştı, ikinci teslim sessizce ack'lendi, InboxState tablosunda mesajın ne zaman tüketildiği ve teslim edildiği duruyor. Consumer'ın içinde SaveChangesAsync çağırmadığına dikkat et: fatura ve inbox kaydını aynı transaction'da kaydeden, middleware. DuplicateDetectionWindow, inbox kayıtlarının ne kadar tutulacağını belirliyor; yukarıdaki saklama süresi kuralı burada da geçerli. Consumer'ın yayınladığı mesajlar da aynı transaction'daki outbox tablosuna yazılır ve commit'ten sonra gönderilir; o kısım bir sonraki yazıda. Bu paketin 8.5.x sürümleri EF Core 9'u hedefliyor; biz EF Core 10.0.14 ve Npgsql 10.0.3 ile sorunsuz çalıştırdık.

Üretimde Dikkat

MessageId'nin kaynağını kontrol et. Inbox, publisher'ın her mesaja benzersiz ve retry'lar boyunca sabit bir MessageId vermesine dayanıyor. Publisher retry'da yeni bir Guid üretiyorsa inbox hiçbir şeyi yakalamaz. Kontrat tarafı serinin altıncı yazısında; özet: MessageId mesaj oluşturulurken bir kez üretilir, yeniden gönderimde değişmez.

Unique violation'ı doğru yerde yakala. Yalnız inbox tablosunun primary key ihlali "duplicate" demek; başka bir tablodaki unique ihlal gerçek bir hata olabilir. PostgreSQL'de PostgresException.ConstraintName ile hangi constraint olduğunu kontrol etmek daha sağlam.

Bağlantı havuzunu say. Consumer concurrency'si, her mesaj için açılan transaction ve dış çağrının süresi çarpıldığında veritabanı bağlantı sayısı hızla artar. Concurrency 20 ve ödeme çağrısı 2 saniye ise yük altında 20 bağlantı sürekli dolu kalır.

Ölç. Duplicate sayısını bir metric olarak yayınla. Normalde düşük ve sabit olan bu sayının aniden artması, bir publisher'ın retry döngüsüne girdiğini ya da bir replay'in fazla geniş yapıldığını gösterir.

Sık Düşülen Tuzaklar

Check-then-act ile idempotency. Yukarıdaki gibi iki fatura; unique constraint olmadan veritabanı seviyesinde garanti yok.

Inbox kaydını ayrı bir transaction'da yazmak. Önce inbox'a yazıp commit edip sonra işi yapmak, iş başarısız olduğunda mesajın "işlendi" sayılıp bir daha hiç denenmemesi demek. Inbox satırı ile iş verisi aynı transaction'da olmalı.

Inbox'ı yalnız MessageId ile anahtarlamak. Aynı mesaj iki farklı consumer'da işlenirken birinin kaydı diğerini atlatır. Anahtara consumer adını ekle.

Dedupe kaydını replay penceresinden kısa tutmak. 14 günlük DLQ ile 1 günlük inbox, 2. günden sonra replay edilen her mesajın yeniden işlenmesi demek.

Dış yan etkiyi transaction'ın koruduğunu sanmak. Rollback, gönderilmiş bir e-postayı geri almaz; idempotency key, outbox ya da bilinçli bir at-most-once kararı gerekir.

Küçük Bir Deneme

★☆☆ Inbox handler'ında yakalanan exception'ı ConstraintName ile daralt: yalnız ProcessedMessages tablosunun primary key'i ise duplicate say, değilse exception'ı yeniden fırlat. Fatura tablosundaki unique index'i ihlal eden bir test mesajıyla farkı gör.

★★☆ Order API'nin aynı siparişi iki farklı MessageId ile yayınladığı senaryoyu kur. Inbox handler'ı ve upsert handler'ı ayrı ayrı çalıştır ve hangisinin iki fatura kestiğini çıktıyla doğrula.

★★★ Bir Payment entity'si ekle (Pending, Captured, Refunded) ve PaymentCaptured consumer'ını durum geçişini ExecuteUpdateAsync ile WHERE Status = Pending koşuluna bağlayarak yaz. Aynı mesajı üç kez, ayrıca sırası bozulmuş bir PaymentRefunded mesajını gönder ve son durumun doğru kaldığını göster.

Consumer tarafını sağlamlaştırdık. Ama soru bir de öbür uçta var: Order API siparişi veritabanına yazdı, ardından mesajı yayınlarken broker'a ulaşamadı. Sipariş var, mesaj yok. Bir sonraki yazıda transactional outbox'ı EF Core ve bir BackgroundService relay'i ile kendimiz yazıyoruz, sonra MassTransit'in hazır outbox'ına bakıyoruz.