Saga'yı .NET'te Uygulamak: Persist Edilen State Machine, Timeout ve Compensation

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile derlenip denenmiştir; saga state'i Entity Framework Core (EF Core) 10 ve SQLite ile saklanıyor, karşılaştırma örnekleri MassTransit 8.5.12 ve Wolverine 6.62.0 ile çalıştırıldı.

Önce Tahmin Et

Bir önceki yazının orchestrator'ı bütün kararları doğru veriyordu, ama state'ini şurada tutuyordu:

    private readonly Dictionary<OrderId, (SagaStep Step, decimal Total)> _sagas = []; // 09'da veritabanına taşınacak

Ve ödeme adımına geçişi şöyle yapıyordu:

            case StockReserved e:
                MoveTo(e.OrderId, SagaStep.CapturingPayment);
                ordering.MarkPaymentPending(e.OrderId);
                bus.Publish(new CapturePayment(e.OrderId, _sagas[e.OrderId].Total));
                break;

Üç soru. Saga CapturingPayment adımındayken uygulama yeniden başlarsa ödeme sağlayıcısından gelecek cevap neyle eşleşecek? Ödeme sağlayıcısı hiç cevap vermezse saga ne kadar bekleyecek ve sonra stoğu bırakıp siparişi iptal etmeli mi? İki instance aynı anda, biri PaymentCaptured'ı diğeri bir timeout'u işlerse hangisi kazanır?

Process Manager'ın State'i: Saga Tablosu

Hohpe ve Woolf'un Process Manager'ı ile Garcia-Molina'nın saga execution coordinator'ı aynı şeyi söylüyor: koordinatörün durumu çökmeye dayanmalı. İlk adım, her saga instance'ı için veritabanında bir satır:

// Saga'nın persist edilen state'i: domain modeli değil, bir iş akışının kaydı
public sealed class OrderSagaState
{
    public Guid OrderId { get; private set; }            // correlation id ve primary key
    public SagaStep Step { get; private set; }
    public decimal Total { get; private set; }
    public DateTime? DeadlineUtc { get; private set; }   // aktif adımın timeout'u
    public int Attempts { get; private set; }           // aynı adımda kaçıncı deneme
    public int Version { get; private set; }            // optimistic concurrency token
    public DateTime UpdatedUtc { get; private set; }

    public static OrderSagaState Start(OrderPlaced placed, DateTime nowUtc) => new()
    {
        OrderId = placed.OrderId, Step = SagaStep.ReservingStock, Total = placed.Total,
        DeadlineUtc = nowUtc + OrderSagaMachine.StockTimeout, Attempts = 1, UpdatedUtc = nowUtc
    };

    public void Apply(Decision decision, DateTime nowUtc)
    {
        Attempts = decision.Next == Step ? Attempts + 1 : 1;
        Step = decision.Next;
        DeadlineUtc = decision.Timeout is { } t ? nowUtc + t : null;
        UpdatedUtc = nowUtc;
        Version++;
    }
}

        b.Entity<OrderSagaState>(s =>
        {
            s.ToTable("OrderSagas");
            s.HasKey(x => x.OrderId);
            s.Property(x => x.Step).HasConversion<string>();
            s.Property(x => x.Version).IsConcurrencyToken();
            s.HasIndex(x => x.DeadlineUtc); // timeout taraması için
        });

Dört karar var. Correlation id olarak OrderId seçildi ve primary key yapıldı: bir siparişin tek bir saga'sı olur, her message'da OrderId taşınır ve saga'yı bulmak bir primary key okuması. Aynı OrderPlaced iki kez gelirse ikinci INSERT primary key ihlaliyle reddedilir, yani başlangıç kendiliğinden idempotent. Step string olarak saklanıyor, çünkü bu tabloya bir gün bir destek mühendisi SQL (Structured Query Language) ile bakacak ve CapturingPayment görmek, 1 görmekten çok daha faydalı. Version bir concurrency token, her Apply'da artıyor; birazdan neden gerektiğini göreceğiz. DeadlineUtc aktif adımın ne zamana kadar cevap beklediğini söylüyor ve index'li; timeout taraması ona göre çalışacak.

Bu satır Order aggregate'i değil. Sipariş, domain'in kendi kavramı ve DDD (Domain-Driven Design) serisinde kurduğumuz kurallarıyla yaşıyor; saga state'i ise "bu sipariş için hangi message'ı bekliyorum, ne zamana kadar" sorusunun cevabı, yani bir iş akışının kaydı. İkisini aynı class'ta birleştirmek, Order'a PaymentTimeoutTokenId gibi domain expert'in hiç duymadığı alanlar ekler. Saga, siparişin durumunu MarkOrderPaid, CancelOrder gibi command'larla değiştirir; Bookstore'da saga Ordering'in içinde yaşadığı için bunlar aynı veritabanında çalışan handler'lar.

Saf Bir State Machine: switch Expression ile Karar

Kararı veren kodu I/O'dan ayırmak, saga'yı test edilebilir yapan en önemli adım. Karar, mevcut state ve gelen message'tan yeni adımı, yeni timeout'u ve gönderilecek command'ları üreten saf bir fonksiyon:

public sealed record Decision(SagaStep Next, TimeSpan? Timeout, IReadOnlyList<IOrderMessage> Commands);

// Saf fonksiyon: (mevcut state, gelen message) -> karar. I/O yok, veritabanı yok; test'i kolay.
public static class OrderSagaMachine
{
    public static readonly TimeSpan StockTimeout = TimeSpan.FromMinutes(1);
    public static readonly TimeSpan PaymentTimeout = TimeSpan.FromSeconds(30);
    public static readonly TimeSpan ShipmentTimeout = TimeSpan.FromMinutes(5);
    public const int MaxShipmentAttempts = 3;

    public static Decision? Decide(OrderSagaState saga, IOrderMessage message)
    {
        // Eski bir adımın timeout'u geç geldiyse yok say
        if (message is SagaTimedOut timedOut && timedOut.Step != saga.Step)
            return null;

        Guid id = saga.OrderId;
        return (saga.Step, message) switch
        {
            (SagaStep.ReservingStock, StockReserved) =>
                Go(SagaStep.CapturingPayment, PaymentTimeout, new CapturePayment(id, saga.Total)),
            (SagaStep.ReservingStock, OutOfStock) =>
                Go(SagaStep.Cancelled, null, new CancelOrder(id, "OutOfStock")),
            (SagaStep.ReservingStock, SagaTimedOut) =>
                Go(SagaStep.Cancelled, null, new ReleaseStock(id), new CancelOrder(id, "StockTimeout")),

            // Pivot: sonucu bilinmeyen ödemeyi körlemesine geri alma, önce sor
            (SagaStep.CapturingPayment or SagaStep.CheckingPayment, PaymentCaptured) =>
                Go(SagaStep.CreatingShipment, ShipmentTimeout, new MarkOrderPaid(id), new CreateShipment(id)),
            (SagaStep.CapturingPayment or SagaStep.CheckingPayment, PaymentFailed) =>
                Go(SagaStep.ReleasingStock, StockTimeout, new ReleaseStock(id)),
            (SagaStep.CapturingPayment or SagaStep.CheckingPayment, SagaTimedOut) =>
                Go(SagaStep.CheckingPayment, PaymentTimeout, new CheckPaymentStatus(id)),

            // Retryable: geri sarma yok, tekrar dene; tükenince bir insana haber ver
            (SagaStep.CreatingShipment, ShipmentCreated) =>
                Go(SagaStep.Completed, null),
            (SagaStep.CreatingShipment, SagaTimedOut) when saga.Attempts < MaxShipmentAttempts =>
                Go(SagaStep.CreatingShipment, ShipmentTimeout, new CreateShipment(id)),
            (SagaStep.CreatingShipment, SagaTimedOut) =>
                Go(SagaStep.Failed, null, new AlertOperations(id, "Shipment could not be created")),

            (SagaStep.ReleasingStock, StockReleased) =>
                Go(SagaStep.Cancelled, null, new CancelOrder(id, "PaymentFailed")),
            (SagaStep.ReleasingStock, SagaTimedOut) =>
                Go(SagaStep.ReleasingStock, StockTimeout, new ReleaseStock(id)),

            // Saga kapandıktan sonra gelen ödeme: para iade edilmeli
            (SagaStep.Cancelled, PaymentCaptured captured) =>
                Go(SagaStep.Refunding, PaymentTimeout, new RefundPayment(id, captured.Amount)),
            (SagaStep.Refunding, PaymentRefunded) =>
                Go(SagaStep.Cancelled, null),
            (SagaStep.Refunding, RefundFailed failed) =>
                Go(SagaStep.Failed, null, new AlertOperations(id, $"Refund failed: {failed.Reason}")),

            // Diğer her şey (duplicate, beklenmeyen sıra): state değişmez
            _ => null
        };
    }

    private static Decision Go(SagaStep next, TimeSpan? timeout, params IOrderMessage[] commands) =>
        new(next, timeout, commands);
}

Bu, Çözüm Desenleri serisinin "Status Field'ı Her Yerden Güncelleniyorsa" yazısındaki transition table'ın kodla yazılmış hâli: her satır bir (adım, message) çiftine izin veriyor, listede olmayan her şey reddediliyor. Design Patterns serisinin "State Pattern" yazısındaki state object'leri yerine tek bir switch seçtim, çünkü burada davranış adımlara göre değil geçişlere göre değişiyor ve bütün geçişleri tek ekranda görmek, saga'yı gözden geçirmenin en kolay yolu. Son satırdaki _ => null sıradan görünüyor ama işin yarısını yapıyor: CreatingShipment adımındaki bir saga'ya ikinci kez gelen PaymentCaptured hiçbir şeyi değiştirmiyor. Saga'nın idempotency'si ayrı bir inbox tablosundan değil, "bu adımda bu message'ın bir anlamı var mı?" sorusundan geliyor. Bu kod veritabanı olmadan, yalnız Decide(state, message) çağrılarıyla test edilir.

Handler: Yükle, Karar Ver, Aynı Transaction'da Kaydet

Saf fonksiyonun etrafındaki kabuk, her message için aynı dört adımı izliyor:

public sealed class OrderSagaHandler(SagaDbContext db, TimeProvider clock)
{
    public async Task HandleAsync(IOrderMessage message, CancellationToken ct)
    {
        DateTime now = clock.GetUtcNow().UtcDateTime;
        OrderSagaState? saga = await db.Sagas.SingleOrDefaultAsync(s => s.OrderId == message.OrderId, ct);

        if (saga is null)
        {
            if (message is not OrderPlaced placed)
            {
                Trace.Write($"no saga for {message.GetType().Name}, ignored");
                return;
            }
            saga = OrderSagaState.Start(placed, now);
            db.Sagas.Add(saga);
            Enqueue(new ReserveStock(placed.OrderId, placed.Books));
            Trace.Write($"OrderPlaced: started -> {saga.Step}");
            await db.SaveChangesAsync(ct); // aynı OrderId ikinci kez gelirse primary key ihlali
            return;
        }

        SagaStep before = saga.Step;
        Decision? decision = OrderSagaMachine.Decide(saga, message);
        if (decision is null)
        {
            Trace.Write($"{message.GetType().Name} in {saga.Step}: ignored");
            return;
        }

        saga.Apply(decision, now);
        foreach (IOrderMessage command in decision.Commands)
            Enqueue(command);
        Trace.Write($"{message.GetType().Name}: {before} -> {saga.Step}" +
                    (decision.Commands.Count > 0 ? $" => {string.Join(", ", decision.Commands.Select(c => c.GetType().Name))}" : ""));

        // State ve command'lar aynı transaction'da (07. yazıdaki outbox); Version eşleşmezse exception
        await db.SaveChangesAsync(ct);
    }

    private void Enqueue(IOrderMessage command) => db.Outbox.Add(new OutboxCommand
    {
        Type = command.GetType().Name,
        Payload = JsonSerializer.Serialize(command, command.GetType())
    });
}

Command'lar broker'a doğrudan gitmiyor, saga state'iyle aynı SaveChangesAsync'te bir outbox tablosuna yazılıyor. Bu, serinin "Transactional Outbox ve Inbox" yazısındaki dual write problemi; saga'da daha da tehlikeli, çünkü state "ödeme bekleniyor" deyip CapturePayment hiç gönderilmezse saga sonsuza kadar bekler, gönderilip state yazılmazsa da gelen cevabı tanımaz. Handler hiçbir dış sistemi çağırmıyor; ödeme sağlayıcısına HTTP (Hypertext Transfer Protocol) isteğini Payments'ın kendi consumer'ı atıyor. Saga'nın transaction'ı birkaç milisaniye sürüyor ve kimsenin ağını beklemiyor. Mutlu yol ve sonradan gelen bir duplicate:

A) Happy path, then a duplicate PaymentCaptured
  [saga] OrderPlaced: started -> ReservingStock
  [saga] StockReserved: ReservingStock -> CapturingPayment => CapturePayment
  [saga] PaymentCaptured: CapturingPayment -> CreatingShipment => MarkOrderPaid, CreateShipment
  [ordering] order -> Paid
  [saga] ShipmentCreated: CreatingShipment -> Completed
  [saga] PaymentCaptured in Completed: ignored
  final: Completed, version 3, deadline none

Correlation ve Optimistic Concurrency

Message'lar at-least-once ve birden çok consumer instance'ıyla işleniyorsa aynı saga'ya ait iki message aynı anda iki farklı thread'e düşebilir. En tehlikeli ikili, ödeme cevabı ile ödeme timeout'u: cevap tam 30. saniyede geldiyse ikisi aynı state'i okur. Bunu iki DbContext ile adım adım oynadım:

    await using SagaDbContext first = createDb();
    await using SagaDbContext second = createDb();
    OrderSagaState a = await first.Sagas.SingleAsync(s => s.OrderId == id);
    OrderSagaState b = await second.Sagas.SingleAsync(s => s.OrderId == id);
    Console.WriteLine($"  both handlers loaded version {a.Version}, step {a.Step}");

    a.Apply(OrderSagaMachine.Decide(a, new PaymentCaptured(id, 60m))!, now);
    await first.SaveChangesAsync();
    Console.WriteLine($"  handler 1 (PaymentCaptured) saved: {a.Step}, version {a.Version}");

    b.Apply(OrderSagaMachine.Decide(b, new SagaTimedOut(id, SagaStep.CapturingPayment))!, now);
    try
    {
        await second.SaveChangesAsync();
    }
    catch (DbUpdateConcurrencyException)
    {
        Console.WriteLine("  handler 2 (SagaTimedOut) lost the race: DbUpdateConcurrencyException");
    }
    await inbox.DeliverAsync(new SagaTimedOut(id, SagaStep.CapturingPayment), ct);

D) PaymentCaptured and the payment timeout race
  [saga] OrderPlaced: started -> ReservingStock
  [saga] StockReserved: ReservingStock -> CapturingPayment => CapturePayment
  [payments] captured, but the reply was lost
  both handlers loaded version 1, step CapturingPayment
  handler 1 (PaymentCaptured) saved: CreatingShipment, version 2
  handler 2 (SagaTimedOut) lost the race: DbUpdateConcurrencyException
  [saga] SagaTimedOut in CreatingShipment: ignored
  final: CreatingShipment, version 2, deadline 10:05:31

Version concurrency token olduğu için EF Core ikinci UPDATE'i WHERE Version = 1 ile gönderdi; satır artık 2 olduğundan etkilenen satır sayısı sıfır ve DbUpdateConcurrencyException fırladı. Concurrency token'ın genel mekanizması EF Core serisinin "EF Core'da Concurrency" yazısında anlatıldı. Optimistic concurrency olmasaydı ikinci handler CheckingPayment'ı yazar, CreatingShipment'ın üstüne geçer ve zaten çekilmiş bir ödemeyi tekrar sorgulayan, gönderiyi unutan bir saga kalırdı: klasik bir lost update. Kaybeden taraf için doğru tepki message'ı atmak değil, taze state ile yeniden denemek:

// Saga'ya message teslim eden giriş noktası: concurrency çakışmasında taze state ile tekrar dener
public sealed class SagaInbox(Func<SagaDbContext> createDb, TimeProvider clock)
{
    public async Task DeliverAsync(IOrderMessage message, CancellationToken ct)
    {
        for (int attempt = 1; ; attempt++)
        {
            await using SagaDbContext db = createDb();
            try
            {
                await new OrderSagaHandler(db, clock).HandleAsync(message, ct);
                return;
            }
            catch (DbUpdateConcurrencyException) when (attempt < 3)
            {
                Trace.Write($"{message.GetType().Name}: concurrency conflict, retrying with fresh state");
            }
        }
    }
}

Yeniden denemede timeout, saga'yı CreatingShipment'ta buldu ve Decide'ın ilk satırı devreye girdi: timeout message'ı hangi adım için kurulduğunu (Step) taşıyor, adım değiştiyse eski timeout anlamsız. MassTransit'in schedule token'ı da aynı işi görüyor. Çakışmalar sık oluyorsa (bir saga'ya saniyede çok sayıda message geliyorsa) bunun bir alternatifi, serinin "Message Ordering ve Partitioning" yazısındaki gibi aynı OrderId'nin message'larını tek bir partition'a ya da session'a yönlendirip saga başına tek tüketici sağlamak. Bookstore'da bir siparişe dakikada birkaç message geliyor; optimistic concurrency yeterli.

Timeout'lar: Deadline Kolonu mu Scheduled Message mı?

Saga'nın asıl düşmanı hata değil sessizlik: başarısız olan bir adım en azından bir event üretir, cevap vermeyen bir adım ise saga'yı sonsuza kadar bekletir. Her bekleme adımının bir deadline'ı olmalı. İki yaygın uygulama var. Birincisi, saga state'indeki deadline'ı periyodik olarak tarayan bir polling:

// Timeout'lar için polling: süresi dolan saga'lara SagaTimedOut teslim eder
public sealed class SagaTimeoutScanner(Func<SagaDbContext> createDb, SagaInbox inbox, TimeProvider clock)
{
    public async Task<int> FireDueAsync(CancellationToken ct)
    {
        DateTime now = clock.GetUtcNow().UtcDateTime;
        List<SagaTimedOut> due;
        await using (SagaDbContext db = createDb())
        {
            due = await db.Sagas
                .Where(s => s.DeadlineUtc != null && s.DeadlineUtc <= now)
                .Select(s => new SagaTimedOut(s.OrderId, s.Step))
                .ToListAsync(ct);
        }
        foreach (SagaTimedOut timedOut in due)
            await inbox.DeliverAsync(timedOut, ct);
        return due.Count;
    }
}

İkincisi, adıma girerken broker'a gecikmeli bir message bırakmak: Azure Service Bus'ın scheduled message'ları ya da Quartz.NET, Hangfire gibi bir scheduler. MassTransit'in Schedule ve Wolverine'in TimeoutMessage'ı bu yolu soyutluyor. Polling'in avantajı, deadline'ın saga state'iyle aynı satırda ve aynı transaction'da olması: adım değişince deadline da değişir, iptal edilecek ayrı bir scheduled message yoktur. Dezavantajı, çözünürlüğün tarama aralığına bağlı olması ve tablonun index'lenmesi gereği. Saniye hassasiyeti gerekmeyen sipariş saga'sında polling basit ve yeterli; test'te de FakeTimeProvider ile zamanı ileri sarmak bir satır.

Asıl soru timeout'ta ne yapılacağı. Pivot adımında körlemesine compensation yapmak, saga'nın yapabileceği en pahalı hata: "ödeme 30 saniyede cevap vermedi, stoğu bırak ve siparişi iptal et" diyen saga, ödemenin aslında çekildiği ama cevabın yolda kaybolduğu durumda müşterinin parasını alıp kitabını iptal eder. Örnekteki state machine pivot'ta timeout olunca CheckingPayment adımına geçiyor ve ödeme sağlayıcısına durumu soruyor:

B) The provider captured, but its reply was lost
  [saga] OrderPlaced: started -> ReservingStock
  [saga] StockReserved: ReservingStock -> CapturingPayment => CapturePayment
  [payments] captured, but the reply was lost
  31 s later the timeout scanner runs
  [saga] SagaTimedOut: CapturingPayment -> CheckingPayment => CheckPaymentStatus
  [saga] PaymentCaptured: CheckingPayment -> CreatingShipment => MarkOrderPaid, CreateShipment
  [ordering] order -> Paid
  [saga] ShipmentCreated: CreatingShipment -> Completed
  final: Completed, version 4, deadline none

Bunun çalışması için ödeme isteğinin idempotency key taşıması gerekiyor: sağlayıcı aynı sipariş için ikinci bir çekim yapmamalı ve durum sorgusunda o key'le cevap verebilmeli. Çözüm Desenleri serisinin "Idempotency Key" yazısında HTTP tarafını anlatmıştık; saga'nın OrderId'si o key için doğal bir aday. Mimariye Giriş serisinin quality attribute senaryolarındaki "ödeme sağlayıcısı 5 saniye yanıt vermezse sipariş PaymentPending olur, kullanıcıya hata dönmez" hedefinin saga'daki karşılığı da bu: kullanıcı beklemeye alınır, belirsizliği saga çözer.

Compensation da Başarısız Olabilir

Garcia-Molina ve Salem'in makalesinin "Other Errors" bölümü, bugün de çoğu tasarımın atladığı soruyu soruyor: compensating transaction'ın kendisi başarısız olursa? Tekrar denemek aynı hatayı verecekse sistem ne geri alabilir ne ileri gidebilir. Örnekteki en kötü senaryo tam olarak bu: ödeme isteği sağlayıcıya ulaşmamış görünüyor, saga stoğu bırakıp siparişi iptal ediyor, sonra sağlayıcı yoldaki isteği yine de işliyor ve iade de reddediliyor.

C) Request never arrived; the capture shows up late and the refund fails
  [saga] OrderPlaced: started -> ReservingStock
  [saga] StockReserved: ReservingStock -> CapturingPayment => CapturePayment
  [payments] request never reached the provider
  31 s later the timeout scanner runs
  [saga] SagaTimedOut: CapturingPayment -> CheckingPayment => CheckPaymentStatus
  [saga] PaymentFailed: CheckingPayment -> ReleasingStock => ReleaseStock
  [inventory] stock released
  [saga] StockReleased: ReleasingStock -> Cancelled => CancelOrder
  [ordering] order -> Cancelled (PaymentFailed)
  ...the provider processes the original request after all
  [saga] PaymentCaptured: Cancelled -> Refunding => RefundPayment
  [saga] RefundFailed: Refunding -> Failed => AlertOperations
  [ops] ALERT: Refund failed: card account closed
  final: Failed, version 6, deadline none

Burada iki tasarım kararı görünüyor. Birincisi, saga Cancelled'a geçtikten sonra da gelen message'ları dinliyor; kapanmış bir saga'yı tablodan hemen silmek, geç gelen PaymentCaptured'ı "saga yok, yok say" diye kaybetmek ve müşterinin parasını sessizce tutmak demek. Bu yüzden tamamlanmış saga satırları, geç message'ların gelebileceği süre boyunca (ödeme sağlayıcısının en uzun işlem süresi, DLQ (dead-letter queue) replay penceresi) saklanmalı. İkincisi, Failed bir son durum ve bir insan için: AlertOperations command'ı bir alarm üretiyor, destek ekibi müşteriyle iletişime geçip parayı başka yolla iade ediyor. Retryable adımın tükenmesi de aynı yere çıkıyor:

E) The carrier API stays down
  [saga] OrderPlaced: started -> ReservingStock
  [saga] StockReserved: ReservingStock -> CapturingPayment => CapturePayment
  [saga] PaymentCaptured: CapturingPayment -> CreatingShipment => MarkOrderPaid, CreateShipment
  [ordering] order -> Paid
  [shipping] carrier API timed out
  [saga] SagaTimedOut: CreatingShipment -> CreatingShipment => CreateShipment
  [shipping] carrier API timed out
  [saga] SagaTimedOut: CreatingShipment -> CreatingShipment => CreateShipment
  [shipping] carrier API timed out
  [saga] SagaTimedOut: CreatingShipment -> Failed => AlertOperations
  [ops] ALERT: Shipment could not be created
  final: Failed, version 5, deadline none

Pivot geçildiği için saga geri sarılmıyor; üç denemeden sonra bir insana devrediyor. Gerçek bir sistemde denemeler arası süre üstel olarak artar ve "üç" sayısı kargo firmasının tipik kesinti süresine göre seçilir. Failed adımındaki saga'ların sayısı, serinin son yazısında konuşacağımız mimari sinyallerden biri.

Aynı Saga MassTransit ve Wolverine'de

Kütüphaneler bu iskeleti hazır veriyor. MassTransit'in state machine'i aynı saga'yı bir DSL (domain-specific language) ile yazıyor:

// Persist edilen state: CorrelationId ve CurrentState MassTransit'in sözleşmesi
public sealed class OrderSagaState : SagaStateMachineInstance, ISagaVersion
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = "";
    public decimal Total { get; set; }
    public Guid? PaymentTimeoutTokenId { get; set; }
    public int Version { get; set; }
}

        During(ReservingStock,
            When(Reserved)
                .Publish(ctx => new CapturePayment(ctx.Saga.CorrelationId, ctx.Saga.Total))
                .Schedule(PaymentTimeout, ctx => new PaymentTimeoutExpired(ctx.Saga.CorrelationId))
                .TransitionTo(CapturingPayment));

        During(CapturingPayment, CheckingPayment,
            When(Captured)
                .Unschedule(PaymentTimeout)
                .Publish(ctx => new CreateShipment(ctx.Saga.CorrelationId))
                .TransitionTo(CreatingShipment),
            When(Declined)
                .Unschedule(PaymentTimeout)
                .Publish(ctx => new ReleaseStock(ctx.Saga.CorrelationId))
                .TransitionTo(ReleasingStock));

        // Pivot'ta timeout: körlemesine compensation yok, önce ödeme durumunu sor
        During(CapturingPayment,
            When(PaymentTimeout.Received)
                .Publish(ctx => new CheckPaymentStatus(ctx.Saga.CorrelationId))
                .TransitionTo(CheckingPayment));

after OrderPlaced: in ReservingStock = True, ReserveStock published: True
after StockReserved: in CapturingPayment = True
after PaymentFailed: in ReleasingStock = True, ReleaseStock published: True
after StockReleased: finalized = True

Çıktı MassTransit'in in-memory test harness'ından. ISagaVersion optimistic concurrency'yi, Schedule/Unschedule timeout token'ını, During(...) ise bizim switch'teki adım eşleşmesini karşılıyor; tanımlanmamış bir (adım, event) çifti varsayılan olarak hata sayılıyor, Ignore(...) ile açıkça yok sayılıyor. Lisans notu: bu kod 8.5.12 (Apache 2.0) ile derlendi; MassTransit 9.x ticari lisanslı ve Massient'in sayfasına göre v8 desteği 31 Aralık 2026'da bitiyor. 8'den 9'a geçiş Backend serisinin konusu.

Wolverine'de (MIT lisansı) saga bir class, state property'leri ve handler method'ları bir arada:

// Wolverine saga: state ve handler'lar aynı class'ta; persistence'ı Wolverine yönetir
public sealed class OrderSaga : Saga
{
    public Guid Id { get; set; }
    public SagaStep Step { get; set; }
    public decimal Total { get; set; }

    public static (OrderSaga, ReserveStock) Start(OrderPlaced placed) =>
        (new OrderSaga { Id = placed.OrderId, Step = SagaStep.ReservingStock, Total = placed.Total },
         new ReserveStock(placed.OrderId, placed.Books));

    public (CapturePayment, PaymentTimeout) Handle([SagaIdentityFrom(nameof(StockReserved.OrderId))] StockReserved reserved)
    {
        Step = SagaStep.CapturingPayment;
        return (new CapturePayment(Id, Total), new PaymentTimeout(Id));
    }

    public CheckPaymentStatus? Handle([SagaIdentityFrom(nameof(PaymentTimeout.OrderId))] PaymentTimeout timeout)
    {
        if (Step != SagaStep.CapturingPayment)
            return null; // ödeme cevabı zaten geldi
        Step = SagaStep.CheckingPayment;
        return new CheckPaymentStatus(Id);
    }

Happy path: 60.00 EUR
  [inventory] reserved 2 x Dune
  [payments] capture 60.00 EUR
  [shipping] shipment created
  handled: ReserveStock -> StockReserved -> CapturePayment -> PaymentCaptured -> CreateShipment -> ShipmentCreated
Payment declined: 135.00 EUR
  [inventory] reserved 2 x Dune
  [payments] capture 135.00 EUR
  [inventory] released
  handled: ReserveStock -> StockReserved -> CapturePayment -> PaymentFailed -> ReleaseStock -> StockReleased

Handler'ın döndürdüğü değerler (tuple dahil) Wolverine'de cascading message olarak gönderiliyor; PaymentTimeout, TimeoutMessage(30.Seconds())'tan türediği için gecikmeli teslim ediliyor. Message'larımızda saga id'si OrderId adında olduğu için [SagaIdentityFrom] ile hangi property'nin kullanılacağını söyledik; Wolverine'in varsayılan kuralı OrderSagaId, SagaId ya da Id arıyor. MarkCompleted() saga satırını siliyor; yukarıdaki geç message tartışmasını hatırlarsan bunun ne zaman çağrılacağı bir tasarım kararı. Örnek, Wolverine'in in-memory saga persistence'ıyla çalıştı; production'da EF Core, Marten ya da Wolverine'in SQL Server/PostgreSQL üzerindeki JSON (JavaScript Object Notation) saga tablosu kullanılır. Kullandığım 6.62.0 sürümünde çekirdek paket runtime compiler'ı artık taşımıyor; WolverineFx.RuntimeCompilation paketini eklemek ya da kodu önceden üretmek gerekiyor, ilk denemede bunu çalışma anındaki hata mesajından öğrendim.

Hangisi? Sıfırdan yazmak birkaç yüz satır ve her satırını sen biliyorsun; kütüphane ise outbox, retry, scheduler entegrasyonu ve test harness'ıyla geliyor. Saga sayısı bir ya da iki olan bir sistemde sıfırdan kod, onlarca saga'sı olan bir sistemde kütüphane mantıklı.

Ne Zaman Workflow Engine?

Saga'nın adımları çoğaldıkça, saatler ya da günler süren bekleyişler (müşteri iade kargosunu ne zaman gönderecek?), insan onayları ve çalışan instance'lar varken akışın sürümünü değiştirme ihtiyacı ortaya çıkar. Bu noktada elle yazılmış bir state machine yerine bir workflow engine düşünülür. Temporal'ın .NET SDK'sı (Software Development Kit; MIT lisansı) ve Microsoft'un Durable Task ailesi (Azure Durable Functions ve açık kaynak Durable Task .NET SDK'sı, MIT lisansı) iş akışını sıradan async C# kodu gibi yazdırıyor; engine her await'in sonucunu bir event geçmişine kaydedip çökmeden sonra kodu baştan "replay" ederek kaldığı yere dönüyor. Bedeli de bu modelden geliyor: workflow kodu deterministik olmak zorunda (doğrudan DateTime.UtcNow, rastgele sayı ya da I/O yok), çalışan workflow'lar varken kodu değiştirmek sürümleme gerektiriyor ve engine'in kendisi işletilmesi gereken yeni bir altyapı. Bookstore'un üç adımlı sipariş saga'sı için bu fazla; iade süreci gibi günlerce süren, insan adımları olan bir akış gelirse karar yeniden değerlendirilmeli.

Sık Düşülen Tuzaklar

Saga state'ini domain modeline karıştırmak

Order entity'sine PaymentTimeoutTokenId, CurrentSagaStep, ShipmentAttempts eklemek, domain modelini bir iş akışı motorunun ayrıntılarıyla doldurur. Domain Order.Status'u bilir (PaymentPending, Paid); saga hangi message'ı beklediğini bilir. İkisi farklı sorulara cevap verir.

Handler'ın içinde dış sistemi çağırmak

Saga handler'ı ödeme sağlayıcısına doğrudan HTTP çağrısı yaparsa transaction açıkken ağ beklenir ve çağrının sonucu ile state yazımı yine iki ayrı işlem olur. Saga yalnız karar verir ve command yazar; dış çağrıyı katılımcı yapar.

Timeout'u adıma bağlamamak

Adım bilgisi taşımayan bir timeout, saga ilerledikten sonra gelirse yeni adımın timeout'u sanılır. Örnekteki SagaTimedOut(OrderId, Step) ya da MassTransit'in token'ı bunu önler.

Tamamlanan saga'yı hemen silmek

Geç gelen message'lar için saga'nın son durumu bir süre daha bilinmeli. Silmek yerine son durumu saklayıp periyodik olarak arşivlemek daha güvenli.

Her şey için saga

Tek bir veritabanında ve tek bir transaction'da bitebilecek bir işi saga'ya çevirmek, karmaşıklığı karşılığı olmadan artırır. Bir önceki yazının "Ne Zaman Kullanma" bölümü burada da geçerli.

Küçük Bir Deneme

★☆☆ OrderSagaMachine.Decide için veritabanı olmadan birkaç test yaz: CreatingShipment adımında gelen PaymentCaptured null dönmeli, CapturingPayment'ta gelen SagaTimedOut CheckPaymentStatus üretmeli. Kaç satır kod yetiyor?

★★☆ Müşterinin iptal isteğini saga'ya ekle: CancelRequested message'ı ReservingStock'ta gelirse stok bırakılıp sipariş iptal edilsin, CapturingPayment'ta gelirse istek kaydedilip ödeme sonucu beklensin, CreatingShipment'ta gelirse refund ile iptal edilsin. Önceki yazıdaki semantic lock'a göre kullanıcı deneyimi nasıl değişti?

★★★ İki SagaInbox instance'ını Task.WhenAll ile aynı saga'ya aynı anda yüzlerce message teslim edecek şekilde çalıştır (SQLite yerine örneğin PostgreSQL ile). Kaç concurrency çakışması oluyor, üç deneme yetiyor mu? Aynı OrderId'nin message'larını tek bir worker'a yönlendirince sayı nasıl değişiyor?

Saga artık çökmeye, sessizliğe ve eşzamanlı message'lara dayanıyor. Ama message'ların içeriği zamanla değişecek: PaymentCaptured'a bir para birimi alanı eklendiğinde ya da OrderPlaced'in bir alanı yeniden adlandırıldığında, eski sürümle çalışan consumer'lar ne olacak? Bir sonraki yazıda message contract'larını, schema evolution'ı, CloudEvents ve AsyncAPI'yi ele alacağız.