Delivery Guarantee'leri: At-Most-Once, At-Least-Once ve "Exactly-Once" Gerçeği

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile, RabbitMQ 4.4 (Docker container'ı) üzerinde derlenip denenmiştir.

Önce Tahmin Et

Notification'ın worker'ı onay e-postası message'larını bir RabbitMQ queue'sundan alıyor. Kural basit ve doğru görünüyor: önce e-postayı gönder, sonra message'ı ack'le. Deneyi kuralım; e-posta sağlayıcısı yerine gönderilenleri listeye yazan küçük bir class kullanıyoruz:

namespace Bookstore.Notifications;

// Dış yan etki: bir kez gönderilen e-posta geri alınamaz
public sealed class EmailSender
{
    public List<string> Sent { get; } = [];
    public void Send(string message) => Sent.Add(message);
}

using System.Text;
using Bookstore.Notifications;
using RabbitMQ.Client;
using RabbitMQ.Client.Exceptions;

ConnectionFactory factory = new() { HostName = "localhost" };
await using IConnection admin = await factory.CreateConnectionAsync();
// Publisher confirm açık: SendAsync, broker message'ı kabul edene kadar bekler
await using IChannel setup = await admin.CreateChannelAsync(
    new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true));
Dictionary<string, object?> quorum = new() { ["x-queue-type"] = "quorum" };

async Task<string> FreshQueueAsync(string name, Dictionary<string, object?> args)
{
    await setup.QueueDeleteAsync(name);
    await setup.QueueDeclareAsync(name, durable: true, exclusive: false, autoDelete: false, arguments: args);
    return name;
}

async Task SendAsync(string queue, string text) =>
    await setup.BasicPublishAsync("", queue, mandatory: false,
        new BasicProperties { MessageId = Guid.CreateVersion7().ToString(), DeliveryMode = DeliveryModes.Persistent },
        Encoding.UTF8.GetBytes(text));

Worker, Alan'ın message'ını alıyor, e-postayı gönderiyor ve tam ack'leyecekken process çöküyor (bağlantıyı AbortAsync ile kopararak taklit ediyoruz). Worker yeniden ayağa kalkıp queue'yu okuyor:

// At-least-once: önce iş, sonra ack. Ack'ten önce çökersek message tekrar gelir.
{
    string queue = await FreshQueueAsync("eda04.at-least-once", quorum);
    await SendAsync(queue, "confirm order for Alan");
    EmailSender email = new();

    await using (IConnection consumer = await factory.CreateConnectionAsync())
    {
        IChannel channel = await consumer.CreateChannelAsync();
        BasicGetResult? got = await channel.BasicGetAsync(queue, autoAck: false);
        email.Send(Encoding.UTF8.GetString(got!.Body.Span));
        await consumer.AbortAsync();   // e-posta gitti, ack gitmedi
    }

    await using (IConnection consumer = await factory.CreateConnectionAsync())
    {
        IChannel channel = await consumer.CreateChannelAsync();
        BasicGetResult? got = await channel.BasicGetAsync(queue, autoAck: false);
        object? count = got!.BasicProperties.Headers?["x-delivery-count"];
        Console.WriteLine($"[at-least-once] redelivered: {got.Redelivered}, x-delivery-count: {count}");
        email.Send(Encoding.UTF8.GetString(got.Body.Span));
        await channel.BasicAckAsync(got.DeliveryTag, multiple: false);
    }
    Console.WriteLine($"[at-least-once] emails sent: {email.Sent.Count} -> {string.Join(" | ", email.Sent)}");
}

Broker message'ı yeniden verir mi? Verirse worker bunun ikinci teslim olduğunu anlayabilir mi? Alan kaç e-posta alır? Ve kuralı tersine çevirip önce ack'leyip sonra e-postayı göndermek sorunu çözer mi?

Üç Garanti, İki Taraf

Delivery guarantee, bir message'ın kaç kez işleneceğine dair verilen sözdür ve üç seviyesi var. At-most-once: message en fazla bir kez işlenir, kaybolabilir ama tekrarlanmaz. At-least-once: message kaybolmaz ama birden fazla işlenebilir. Exactly-once: her message tam bir kez işlenir. Apache Kafka'nın tasarım dokümanı, bu sorunun aslında iki ayrı sorundan oluştuğunu vurguluyor: producer'ın gönderdiği message'ın broker'a kalıcı olarak ulaşması ve consumer'ın onu işlemesi. Bir zincir en zayıf halkası kadar güçlü; producer tarafı at-least-once, consumer tarafı at-most-once ise uçtan uca garanti at-most-once olur.

Her iki tarafta da garantiyi belirleyen şey aynı: onay (acknowledgement) ile işin sırası.

Consumer Tarafı: Ack Nerede Duruyor?

Tahminin cevabı:

[at-least-once] redelivered: True, x-delivery-count: 1
[at-least-once] emails sent: 2 -> confirm order for Alan | confirm order for Alan

Broker message'ı yeniden verdi, Redelivered bayrağını true yaptı ve quorum queue x-delivery-count header'ında başarısız teslim sayısını 1 olarak gösterdi. Alan iki e-posta aldı. Bu at-least-once: broker, ack almadığı message'ın işlenip işlenmediğini bilemez; güvenli tarafta kalıp yeniden teslim eder. Çökme, e-postayla ack arasındaki milisaniyelik boşluğa denk geldiyse sonuç bir duplicate'tir.

Peki Redelivered bayrağı duplicate'i yakalamak için yeterli mi? Hayır. Bayrak "bu message daha önce birine verildi" diyor, "işlendi" demiyor; worker, e-posta gönderilmeden önce de çökmüş olabilirdi. Üstelik birazdan göreceğimiz gibi duplicate'lerin bir kısmı broker'a iki ayrı message olarak gelir ve onlarda bayrak hiç yanmaz.

Kuralı tersine çevirelim: önce ack (burada autoAck: true, yani message teslim edildiği anda ack'lenmiş sayılıyor), sonra iş:

// At-most-once: önce ack (autoAck), sonra iş. Arada çökersek message kaybolur.
{
    string queue = await FreshQueueAsync("eda04.at-most-once", quorum);
    await SendAsync(queue, "confirm order for Ada");
    EmailSender email = new();

    await using (IConnection consumer = await factory.CreateConnectionAsync())
    {
        IChannel channel = await consumer.CreateChannelAsync();
        BasicGetResult? got = await channel.BasicGetAsync(queue, autoAck: true);
        Console.WriteLine($"[at-most-once]  received: {Encoding.UTF8.GetString(got!.Body.Span)}");
        await consumer.AbortAsync();   // e-postayı göndermeden "çöktü"
    }
    Console.WriteLine($"[at-most-once]  emails sent: {email.Sent.Count}, left in queue: {await setup.MessageCountAsync(queue)}");
}

[at-most-once]  received: confirm order for Ada
[at-most-once]  emails sent: 0, left in queue: 0

Duplicate gitti, onunla birlikte message da gitti: Ada'nın e-postası hiç gönderilmedi ve queue boş. At-most-once bu. Kafka'daki karşılığı da aynı mantık: consumer offset'ini işten önce commit ederse at-most-once, işten sonra commit ederse at-least-once olur.

Broker'ların "message şu an birinde, ama henüz bitmedi" durumunu tutma biçimleri farklı, fikir aynı. RabbitMQ'da ack'lenmemiş message, consumer'ın channel'ı ya da bağlantısı kapandığında queue'ya geri döner. Azure Service Bus'ın peek-lock modunda message bir süreliğine kilitlenir; varsayılan lock süresi 1 dakika, en fazla 5 dakika, ve consumer süre dolmadan Complete çağırmazsa message yeniden görünür olur. Amazon SQS'te (Simple Queue Service) aynı şeyin adı visibility timeout. Her birinde aynı soru var: işin süresi kilidin süresini aşarsa, birinci consumer hâlâ çalışırken ikinci consumer aynı message'ı alır. Bu, çökme olmadan da duplicate üretmenin en sinsi yolu.

Producer Tarafı: Confirm ve Retry

Producer'ın sorusu "broker message'ımı aldı mı?". RabbitMQ'da cevap publisher confirm: channel'da confirm açıldığında broker, message'ı kalıcı olarak kabul ettiğinde ya da reddettiğinde haber verir. RabbitMQ.Client 7'de BasicPublishAsync'in döndürdüğü task bu cevabı bekliyor. mandatory: true ise "hiçbir queue'ya yönlendiremezsen bana geri ver" demek:

// Producer tarafı: publisher confirm + mandatory, broker message'ı yönlendiremezse haber verir
{
    await using IChannel confirmed = await admin.CreateChannelAsync(
        new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true));
    try
    {
        await confirmed.BasicPublishAsync("", "eda04.no-such-queue", mandatory: true,
            new BasicProperties(), Encoding.UTF8.GetBytes("confirm order for Grace"));
        Console.WriteLine("[confirm]       published");
    }
    catch (PublishReturnException ex)
    {
        Console.WriteLine($"[confirm]       returned by broker: {ex.ReplyCode} {ex.ReplyText}");
    }
}

foreach (string q in new[] { "eda04.at-most-once", "eda04.at-least-once", "eda04.emails", "eda04.emails.dlq" })
    await setup.QueueDeleteAsync(q);

[confirm]       returned by broker: 312 NO_ROUTE

Broker message'ı aldı ama gidecek bir queue bulamadı ve 312 NO_ROUTE ile geri verdi. Confirm olmadan bu message sessizce kaybolurdu: bir önceki yazıda gördüğümüz, bağlı queue'su olmayan exchange davranışı. Azure Service Bus'ın client'larında send zaten her zaman broker'ın kabulünü bekliyor. Kafka'da ise producer'ın acks ayarı belirleyici; Java producer'ında Kafka 3.0'dan beri, .NET'in kullandığı librdkafka'da da varsayılan acks=all, yani kayıt bütün senkron replica'lara yazılmadan başarılı sayılmıyor.

Confirm, kaybı önler ama yeni bir soru açar. Producer message'ı gönderdi, broker yazdı, ama confirm ağda kayboldu ya da timeout'a düştü. Producer ne yapmalı? Tekrar göndermezse message kaybolabilir; tekrar gönderirse ve ilki aslında yazılmışsa broker'da iki kopya olur. Güvenli seçim tekrar göndermek, yani producer tarafında da at-least-once. Bu ikinci kopya broker'a ayrı bir message olarak girer; consumer'da Redelivered bayrağı yanmaz.

Bazı broker'lar bu kaynağı kendi tarafında kapatmaya çalışıyor. Kafka'nın idempotent producer'ı (0.11'den beri var; Java producer'ında 3.0'dan beri varsayılan olarak açık, ama 3.0 ve 3.1'deki bir hata yüzünden varsayılan ancak düzeltilmiş sürümlerde gerçekten uygulanıyor; Confluent.Kafka'nın altındaki librdkafka'da ise varsayılan hâlâ kapalı, EnableIdempotence = true ile açılıyor) her producer'a bir id verir ve kayıtları sıra numarasıyla gönderir; broker aynı numarayı ikinci kez gördüğünde yazmaz. Azure Service Bus'ın duplicate detection özelliği, aynı MessageId ile gelen ikinci message'ı belirli bir pencere içinde (varsayılan 10 dakika, en fazla 7 gün) sessizce düşürür. İkisi de yalnızca producer'ın tekrar göndermesini kapsar.

Duplicate'lerin Üç Kaynağı

Microsoft'un Idempotent Consumer rehberi duplicate'lerin kaynağını üç başlıkta topluyor ve yukarıdaki deneyler üçünü de gösteriyor:

• Producer retry: gönderim başarılı oldu ama onay ulaşmadı, producer tekrar gönderdi. Broker'da iki ayrı message var.
• Kayıp ack: consumer işledi ama ack broker'a ulaşmadı (bağlantı koptu, lock süresi doldu). Broker aynı message'ı yeniden teslim ediyor.
• İşlem sonrası çökme: consumer veritabanına yazdı ya da e-postayı gönderdi, ack'lemeden çöktü. Alan'ın iki e-postası bu.

Üçünün ortak noktası, karşı tarafın "oldu mu, olmadı mı?" sorusunun cevabını kesin bilememesi. Bunun teorik bir sınır olduğunu, iki tarafın güvenilmez bir kanal üzerinden kesin bir uzlaşmaya varamayacağını gösteren Two Generals problemi ilerideki Distributed Systems Temelleri serisinin konusu. Pratik sonucu şu: ağ üzerinden konuşan iki taraf arasında, kayıp olmasın istiyorsan duplicate'e katlanmak zorundasın.

"Exactly-Once" Gerçeği

Peki Kafka'nın "exactly-once semantics" diye sunduğu özellik ne? Kafka 0.11 ile iki parça geldi: idempotent producer ve transaction'lar. Transaction, bir producer'ın birden fazla partition'a yaptığı yazmaları atomik yapıyor. Asıl önemlisi, consumer'ın offset'i de aynı transaction'a yazılabiliyor. Kafka'dan oku, işle, Kafka'ya yaz zincirinde (Kafka Streams'in yaptığı tam olarak bu) okuma pozisyonu ile çıktı birlikte commit ediliyor ya da birlikte geri alınıyor; okuyanlar da read_committed isolation seviyesiyle yarım kalan transaction'ları görmüyor.

Sınır burada çiziliyor ve Kafka'nın kendi dokümanı da bunu açıkça söylüyor: exactly-once, broker'ın kendi kontrol ettiği işlemler için geçerli. Çıktı bir dış sistemse (bir veritabanı, bir e-posta sağlayıcısı, bir ödeme API'si (Application Programming Interface)) o sistemin işbirliği gerekir; Kafka'nın önerdiği yol, consumer'ın offset'ini çıktıyla birlikte o sistemde saklamak. Alan'ın e-postasını hiçbir broker ayarı "tam bir kez" gönderemez; çünkü e-posta sağlayıcısı Kafka'nın transaction'ına katılmıyor.

Microsoft'un rehberi bu durumu bir terimle özetliyor: effectively-once. Broker at-least-once teslim eder, consumer duplicate'i fark edip etkisiz hâle getirir; sonuç, dışarıdan bakıldığında tam bir kez işlenmiş gibidir. Formül şu: at-least-once delivery + idempotent consumer = effectively-once. Formülün ikinci yarısı serinin 06. yazısının konusu.

Retry, Poison Message ve Dead-Letter Queue

At-least-once'ın bir başka yüzü var: bazı message'lar hiçbir zaman işlenemez. Bozuk bir JSON (JavaScript Object Notation), artık var olmayan bir kitaba ait bir sipariş, consumer'daki bir bug. Böyle bir message'ı yeniden denemek işe yaramaz; sonsuza kadar yeniden denemek ise queue'yu tıkar ve CPU'yu (Central Processing Unit) boşa yakar. Buna poison message deniyor. EIP'teki (Enterprise Integration Patterns) adıyla Dead Letter Channel, teslim edilemeyen ya da işlenemeyen message'ların gönderildiği ayrı bir yer. Kavramı küçük bir denemeyle görelim; limiti küçük tutalım:

// Poison message: hiçbir zaman işlenemeyen message, limit dolunca dead-letter queue'ya
{
    string dlq = await FreshQueueAsync("eda04.emails.dlq", quorum);
    string queue = await FreshQueueAsync("eda04.emails", new()
    {
        ["x-queue-type"] = "quorum",
        ["x-delivery-limit"] = 3,
        ["x-dead-letter-exchange"] = "",
        ["x-dead-letter-routing-key"] = dlq,
    });
    await SendAsync(queue, "{ not valid json");

    IChannel channel = await admin.CreateChannelAsync();
    int attempts = 0;
    while (await channel.BasicGetAsync(queue, autoAck: false) is { } got)
    {
        attempts++;
        await channel.BasicRejectAsync(got.DeliveryTag, requeue: true);   // "işleyemedim, tekrar dene"
    }
    uint dead = 0;
    for (int i = 0; i < 20 && dead == 0; i++)   // dead-lettering asenkron: kısa bir süre bekle
    {
        await Task.Delay(100);
        dead = await setup.MessageCountAsync(dlq);
    }
    Console.WriteLine($"[poison]        attempts: {attempts}, main queue: {await setup.MessageCountAsync(queue)}, " +
                      $"dead-letter queue: {dead}");
}

[poison]        attempts: 4, main queue: 0, dead-letter queue: 1

Message bir kez teslim edildi, üç kez yeniden denendi ve limit dolunca dead-letter queue'ya taşındı. Ana queue boşaldı, sıradaki message'lar artık onun arkasında beklemiyor. Taşımanın asenkron olduğunu da denemede gördüm: ilk koşuda sayaç dead-letter queue'yu boş gösterdi, bu yüzden kod kısa bir süre bekliyor.

Varsayılanlar önemli. RabbitMQ 4.0'dan beri quorum queue'larda delivery limit varsayılan olarak 20; öncesinde sınırsızdı. Dead-letter hedefi tanımlı değilse limiti aşan message silinir, bu yüzden RabbitMQ dokümanı her quorum queue için bir dead-letter düzeni kurmayı öneriyor. Sayaç da ince ayrıntılı: RabbitMQ 4.3'ten beri yalnızca gerçek başarısızlıklar (basic.reject, bağlantı kopması, çökme) limite sayılıyor, basic.nack sayılmıyor. Azure Service Bus'ta aynı işi MaxDeliveryCount yapıyor (varsayılanı 10). Gecikmeli retry, backoff ve dead-letter konfigürasyonunun ayrıntıları ilerideki Backend › Message Queue'larla Çalışmak serisinde.

Mimari açıdan önemli olan, dead-letter queue'nun bir çöp kutusu değil, bir iş listesi olması. Orada duran her message, bir müşterinin almadığı e-posta ya da hazırlanmamış bir kargo. Birinin onu izlemesi, nedenini bulması ve düzelttikten sonra yeniden göndermesi gerekiyor. Yeniden gönderilen message da tanım gereği bir duplicate adayı; idempotent consumer'ın dedup kayıtlarını ne kadar süre saklaması gerektiği sorusu buradan doğuyor.

Hangi Garantiyi Seçmeli?

Bookstore'un mesajlarının çoğu ikinci ve üçüncü satırda. Pratik kural: at-least-once ile başla, her consumer için "bu message iki kez gelirse ne olur?" sorusunu sor ve cevabı "zarar verir" olanları idempotent yap.

Sık Düşülen Tuzaklar

autoAck'i kolaylık diye açmak

Örneklerde ve ilk prototiplerde autoAck: true sık görülür, çünkü kod daha kısa. Ama bu, consumer'ın her çökmesinde elindeki message'ları kaybetmeyi kabul etmek demek. Bunu bilinçli seçmediysen manuel ack kullan ve ack'i iş bittikten sonra gönder.

Lock süresini işin süresine göre ayarlamamak

Azure Service Bus'ta 1 dakikalık varsayılan lock ile 90 saniye süren bir işi yapan consumer, her message'ı iki kez işler: lock düşer, message başka bir consumer'a gider, ilk consumer'ın Complete çağrısı da kilit kaybolduğu için başarısız olur. Çökme yok, hata log'u var, duplicate var. Lock süresini normal işlem süresinin üstünde tut, uzun işlerde lock'u yenile.

Exactly-once bayrağına güvenmek

Kafka'da transaction'ları açmak, consumer'ın veritabanına yazdığı satırları ya da gönderdiği e-postaları tekilleştirmez. "Exactly-once" ayarı gördüğünde sor: hangi işlemler bu sözün içinde, hangileri dışında?

Sonsuz retry ve dead-letter'ı izlememek

Limit olmadan yeniden kuyruğa koymak, tek bir bozuk message'ın bir consumer'ı saatlerce meşgul etmesi demek. Limit varsa da dead-letter queue izlenmiyorsa sorun yalnızca görünmez olur. Dead-letter queue derinliği, serinin son yazısında konuşacağımız mimari sinyallerin başında geliyor.

Küçük Bir Deneme

★☆☆ Poison message denemesinde BasicRejectAsync yerine BasicNackAsync(got.DeliveryTag, multiple: false, requeue: true) kullan. Kaç denemeden sonra döngü bitiyor? RabbitMQ 4.3+ kuralına göre neden? (Döngüye bir üst sınır koymayı unutma.)

★★☆ At-least-once denemesinde ikinci consumer'ın, aynı message'ı ikinci kez işlediğini MessageId üzerinden fark etmesini sağla: işlenmiş id'leri bellekteki bir HashSet<string>'te tut. Bu çözüm worker yeniden başladığında neden işe yaramaz?

★★★ Producer tarafındaki duplicate'i taklit et: aynı içeriği iki kez, her seferinde yeni bir MessageId ile gönder. Consumer bu iki message'ı birbirinden nasıl ayırt edebilir? MessageId'yi rastgele değil, iş akışından türetilmiş bir değer (sipariş id'si + message türü gibi) yapsaydın ne değişirdi?

At-least-once'ın sözü sıra hakkında hiçbir şey söylemiyor; üstelik yeniden teslim edilen message, arkasındaki message'lardan sonra işlenebiliyor. Bir sonraki yazıda sıra garantisinin nerede başlayıp nerede bittiğine, partition key ve session'larla sıranın nasıl korunduğuna bakacağız.