.NET'te Transactional Outbox: EF Core, BackgroundService Relay ve MassTransit/Wolverine Outbox

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

Order API'nin (Application Programming Interface) iki instance'ı çalışıyor. RabbitMQ'yu durduruyorsun ve broker 11 saniye kapalı kalıyor. Bu sırada POST /orders'a 5 istek atılıyor. İstemci ne görür: 201 mi, 500 mü, 503 mü? Broker geri geldiğinde bu 5 siparişin OrderPlaced mesajları ne olur, kim gönderir ve kaç saniye içinde?

Senaryo, serinin bütün yazılarının altında yatan soru: sipariş veritabanına yazıldı ama mesaj yayınlanamadı (ya da tersi). İki ayrı sisteme tek bir atomik işlem gibi yazmanın mümkün olmadığı "dual write" problemi ve transactional outbox'ın teorisi, Messaging ve Event-Driven Architecture serisinin "Transactional Outbox ve Inbox" yazısında. Bu yazıda outbox'ı EF Core, PostgreSQL ve bir BackgroundService relay'i ile kendimiz yazıyor, iki instance ve kapalı broker ile deniyor, sonra MassTransit'in hazır outbox'ına bakıyoruz.

Outbox Tablosu ve Aynı Transaction

Fikir basit: mesajı broker'a değil, iş verisiyle aynı veritabanına, aynı transaction'da bir tabloya yaz. Sipariş varsa mesajı da vardır; ikisi birlikte commit edilir ya da ikisi birlikte yok olur.

public sealed class Order
{
    public Guid Id { get; init; } = Guid.CreateVersion7();
    public Guid CustomerId { get; init; }
    public decimal Total { get; init; }
    public required string Currency { get; init; }
    public DateTimeOffset PlacedAt { get; init; } = DateTimeOffset.UtcNow;
}

public sealed class OutboxMessage
{
    public Guid Id { get; init; } = Guid.CreateVersion7();
    public required string Type { get; init; }
    public required string RoutingKey { get; init; }
    public required string Payload { get; init; }
    public string? CorrelationId { get; init; }
    public DateTimeOffset OccurredAt { get; init; } = DateTimeOffset.UtcNow;
    public DateTimeOffset? ProcessedAt { get; set; }
}

public sealed class OrderingDbContext(DbContextOptions<OrderingDbContext> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();
    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<OutboxMessage>(b =>
        {
            b.Property(m => m.Payload).HasColumnType("jsonb");
            b.Property(m => m.Type).HasMaxLength(200);
            b.Property(m => m.RoutingKey).HasMaxLength(200);
            // relay yalnız işlenmemiş satırları tarar: partial index
            b.HasIndex(m => m.OccurredAt).HasFilter("\"ProcessedAt\" IS NULL");
        });
    }
}

OutboxMessage.Id broker'a MessageId olarak gidecek; satır bir kez oluşturuluyor ve relay onu kaç kez yayınlarsa yayınlasın aynı kalıyor, bu da bir önceki yazıdaki inbox'ın çalışmasını sağlıyor. Guid.CreateVersion7() zamana göre sıralı olduğu için index'e de dosttur. Partial index, relay'in her saniye çalıştıracağı sorguyu milyonlarca işlenmiş satırın arasından değil, yalnız bekleyenlerin arasından yapmasını sağlar. Endpoint:

app.MapPost("/orders", async (PlaceOrderRequest request, OrderingDbContext db, CancellationToken ct) =>
{
    var order = new Order
    {
        CustomerId = request.CustomerId,
        Total = request.Items.Sum(i => i.Quantity * i.UnitPrice),
        Currency = request.Currency
    };
    var @event = new OrderPlaced(order.Id, order.CustomerId, order.Total, order.Currency, order.PlacedAt,
        [.. request.Items.Select(i => new OrderPlacedItem(i.BookId, i.Quantity, i.UnitPrice))]);

    db.Orders.Add(order);
    db.OutboxMessages.Add(new OutboxMessage
    {
        Type = "bookstore.order-placed.v1",
        RoutingKey = "order.placed",
        Payload = JsonSerializer.Serialize(@event, JsonSerializerOptions.Web),
        CorrelationId = order.Id.ToString()
    });
    // tek SaveChanges = tek transaction: ya ikisi birden yazılır ya hiçbiri
    await db.SaveChangesAsync(ct);
    return Results.Created($"/orders/{order.Id}", new { order.Id, order.Total });
});

Endpoint'te broker'a dair tek satır yok; RabbitMQ kapalı olsa da çalışır. Tek bir SaveChangesAsync çağrısı EF Core'da zaten bir transaction içinde çalışır; ayrıca BeginTransactionAsync gerekmiyor. Birden fazla SaveChangesAsync gerekirse EF Core serisinin "EF Core'da Transaction'lar" yazısındaki gibi açık transaction'a geçersin.

Relay: BackgroundService ve FOR UPDATE SKIP LOCKED

Outbox satırlarını broker'a taşıyan, aynı uygulamada çalışan bir BackgroundService:

public sealed class OutboxRelay(
    IServiceScopeFactory scopeFactory, IConnection connection, ILogger<OutboxRelay> logger) : BackgroundService
{
    private const int BatchSize = 50;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using IChannel channel = await connection.CreateChannelAsync(
            new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true),
            stoppingToken);
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
        do
        {
            try
            {
                int published;
                do
                {
                    published = await PublishBatchAsync(channel, stoppingToken);
                } while (published == BatchSize);
            }
            catch (Exception ex) when (ex is not OperationCanceledException || !stoppingToken.IsCancellationRequested)
            {
                // satırlar ProcessedAt = null kaldı; bir sonraki tick'te yeniden denenir
                logger.LogWarning("Outbox relay batch failed: {Error}", ex.Message);
            }
        } while (await timer.WaitForNextTickAsync(stoppingToken));
    }

    private async Task<int> PublishBatchAsync(IChannel channel, CancellationToken ct)
    {
        await using AsyncServiceScope scope = scopeFactory.CreateAsyncScope();
        OrderingDbContext db = scope.ServiceProvider.GetRequiredService<OrderingDbContext>();
        await using var tx = await db.Database.BeginTransactionAsync(ct);

        // SKIP LOCKED: başka bir instance'ın kilitlediği satırları beklemeden atla
        List<OutboxMessage> batch = await db.OutboxMessages
            .FromSql($"""
                SELECT * FROM "OutboxMessages"
                WHERE "ProcessedAt" IS NULL
                ORDER BY "OccurredAt"
                LIMIT {BatchSize}
                FOR UPDATE SKIP LOCKED
                """)
            .ToListAsync(ct);
        if (batch.Count == 0)
            return 0;

        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
        timeout.CancelAfter(TimeSpan.FromSeconds(10));
        foreach (OutboxMessage message in batch)
        {
            var props = new BasicProperties
            {
                MessageId = message.Id.ToString(),
                CorrelationId = message.CorrelationId,
                Type = message.Type,
                ContentType = "application/json",
                Persistent = true
            };
            // confirm tracking açık: broker onaylamadan await dönmez
            await channel.BasicPublishAsync("bookstore.ordering", message.RoutingKey, mandatory: true, props,
                Encoding.UTF8.GetBytes(message.Payload), timeout.Token);
            message.ProcessedAt = DateTimeOffset.UtcNow;
        }
        await db.SaveChangesAsync(ct);
        await tx.CommitAsync(ct);
        logger.LogInformation("Relay published {Count} outbox messages", batch.Count);
        return batch.Count;
    }
}

Her tick'te relay bir transaction açar ve en eski 50 bekleyen satırı FOR UPDATE SKIP LOCKED ile kilitler. FOR UPDATE satırları bu transaction bitene kadar başkasına kapatır; SKIP LOCKED ise başka bir instance'ın kilitlediği satırlarda beklemek yerine onları atlamasını sağlar. Böylece iki relay aynı tabloda çalışırken birbirini bloklamaz ve aynı satırı iki kez almaz. Her mesaj publisher confirm ile gönderilir; broker onaylamadan satır işaretlenmez. Batch bitince ProcessedAt yazılır ve transaction commit edilir. Bir mesajda hata olursa (broker kapalı, confirm gelmedi, 10 saniyelik timeout) exception batch'i keser, transaction rollback olur, kilitler bırakılır ve satırlar bir sonraki tick'te yeniden denenir. FromSql'deki {BatchSize} bir parametreye dönüşür; bu bir SQL (Structured Query Language) string birleştirmesi değil.

Bu tasarım bilerek yavaş: mesajları tek tek ve confirm bekleyerek gönderiyor. Saniyede birkaç bin mesajdan fazlası gerekiyorsa confirm'leri toplu beklemek (tracking'i kapatıp batch sonunda onay beklemek) ya da hazır bir kütüphaneye geçmek gerekir. Bir de PeriodicTimer 1 saniye: sipariş ile mesaj arasında ortalama yarım saniyelik bir gecikme kabul ediyorsun.

İki Instance, Tek Yayın

API'nin iki kopyasını aynı veritabanı ve aynı broker'la başlattım ve 200 siparişi 8'erli paralel isteklerle iki instance'a dağıttım. 200 istek 0,4 saniyede bitti. Relay log'ları:

api1 (fail lines: 0)
05:47:04.097 info: Bookstore.OrderApi.OutboxRelay[0] Relay published 50 outbox messages
api2 (fail lines: 0)
05:47:03.913 info: Bookstore.OrderApi.OutboxRelay[0] Relay published 50 outbox messages
05:47:03.972 info: Bookstore.OrderApi.OutboxRelay[0] Relay published 50 outbox messages
05:47:04.028 info: Bookstore.OrderApi.OutboxRelay[0] Relay published 50 outbox messages

order.#'a bağlı bir kontrol queue'sundaki mesajları sayan küçük bir araç:

messages in lab11.audit: 200, distinct MessageId: 200

İkinci instance üç batch, birincisi bir batch yayınladı; toplam 200, her MessageId tek. İş bölümü eşit değil ve olması da gerekmiyor: hangi relay'in tick'i önce gelirse o kilitler. Önemli olan, hiçbir satırın iki relay tarafından birlikte alınmaması. Bu yaklaşım Background Job'lar serisinde anlatılacak distributed lock'a da ihtiyaç duymuyor; kilit, satır seviyesinde veritabanının kendisi.

Broker Kapalıyken

Şimdi baştaki soru. RabbitMQ'yu durdurdum, beş sipariş attım, 11 saniye sonra broker'ı başlattım:

05:47:17 rabbit stopped
201 201 201 201 201
05:47:20 {"pending":5}
05:47:28 rabbit started
05:47:53 {"pending":0}

Relay'lerin log'u (her biri bu sürede 15 uyarı yazdı; ilk iki ve son satırları bırakıp uzun satırları kısalttım):

api2:
05:47:17.807 warn: Bookstore.OrderApi.OutboxRelay[0] Outbox relay batch failed: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=320, text='CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'' ...
05:47:18.807 warn: Bookstore.OrderApi.OutboxRelay[0] Outbox relay batch failed: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=320, text='CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'' ...
05:47:32.833 info: Bookstore.OrderApi.OutboxRelay[0] Relay published 5 outbox messages

İstemci beş kez 201 aldı; sipariş kaydedildi, broker'ın durumu istemciyi ilgilendirmiyor. Beş mesaj outbox'ta bekledi. Broker 05:47:28'de ayağa kalktı, RabbitMQ.Client'ın automatic recovery'si connection'ı ve channel'ı yaklaşık 5 saniyede yeniden kurdu ve 05:47:32'de ikinci instance'ın relay'i beşini birden yayınladı. Outbox olmasaydı aynı beş istek ya 500 dönerdi (sipariş kaydedilmiş ama istemci hata görmüş) ya da sipariş kaydedilip mesaj sessizce kaybolurdu.

Outbox'ın Garanti Etmedikleri

Relay de at-least-once. BasicPublishAsync confirm aldıktan sonra, CommitAsync'ten önce process çökerse satırlar ProcessedAt = null olarak kalır ve bir sonraki relay onları tekrar yayınlar. Outbox, "mesaj kaybolmaz" garantisi verir; "mesaj bir kez gider" garantisi vermez. Bu yüzden outbox ile bir önceki yazıdaki inbox bir çifttir: biri kaybı, diğeri tekrarı karşılar. Aynı nedenle batch büyüdükçe bir çöküşte tekrar gönderilecek mesaj sayısı da büyür.

Sıralama da garanti değil. Tek relay ve tek channel ile mesajlar OccurredAt sırasıyla gider. İki relay olduğunda birinci batch'i alan relay yavaş kalırsa, ikinci batch ondan önce broker'a ulaşabilir. Aynı siparişin OrderPlaced ve OrderCancelled mesajları farklı batch'lere düştüyse consumer onları ters sırada görebilir. Aggregate başına sıra gerekiyorsa ya tek relay çalıştırırsın, ya satırları aggregate id'sine göre bölüp her bölümü tek bir relay'e verirsin, ya da consumer'ı sıradan bağımsız kılarsın (bir önceki yazıdaki state machine kontrolü). Messaging serisinin "Message Ordering ve Partitioning" yazısındaki tartışma burada birebir geçerli.

Trace zinciri kopar. Relay mesajı ayrı bir arka plan işinde gönderdiği için HTTP (Hypertext Transfer Protocol) isteğinin trace context'i mesaja kendiliğinden geçmez. Bunu bizzat gördüm: bir sonraki yazının demosunda bu relay'in yayınladığı mesajlar consumer'da parent'sız, yepyeni trace'ler olarak göründü. Çözümü basit: traceparent'ı da outbox satırına bir kolon olarak yazıp relay'de header'a koymak; ayrıntısı serinin on ikinci yazısında.

Hazır Çözümler: MassTransit, Wolverine ve CDC

MassTransit'in EF Core outbox'ının "bus outbox" modu, publish'i doğrudan DbContext'e bağlar. Bir önceki yazıdaki AddEntityFrameworkOutbox<BookstoreDbContext>(o => { o.UsePostgres(); o.UseBusOutbox(); }) kurulumuyla, scoped IPublishEndpoint mesajı broker'a değil DbContext'e ekler:

public sealed class OrderService(BookstoreDbContext db, IPublishEndpoint publishEndpoint)
{
    public async Task<Guid> PlaceOrderAsync(Guid customerId, decimal total, bool saveChanges)
    {
        var order = new Order { CustomerId = customerId, Total = total };
        db.Orders.Add(order);
        // bus outbox: Publish broker'a gitmez, DbContext'e OutboxMessage olarak eklenir
        await publishEndpoint.Publish(new OrderPlaced(order.Id, customerId, total, "EUR", order.PlacedAt, []));
        if (saveChanges)
            await db.SaveChangesAsync();
        return order.Id;
    }
}

İki sipariş verdim: birinde SaveChangesAsync çağrıldı, diğerinde çağrılmadı:

05:45:04.333 info: Lab[0] Order 42d45070 saved, OutboxMessage rows: 1
05:45:04.335 info: Lab[0] Order 3c4e4650 published but SaveChanges NOT called
05:45:04.415 info: Bookstore.Billing.InvoiceConsumer[0] Invoice for order 42d45070, MessageId be8d0000
05:45:07.345 info: Lab[0] After delivery: OutboxMessage rows 0, invoices 1

Kaydedilen sipariş, SaveChangesAsync'ten hemen sonra OutboxMessage tablosunda bir satır olarak duruyordu; MassTransit'in delivery service'i onu yaklaşık 80 ms sonra broker'a taşıdı, Billing faturayı kesti ve satır silindi. SaveChangesAsync çağrılmayan siparişin Publish'i hiçbir yere gitmedi: transaction yoksa mesaj da yok. Kendi relay'imizdeki bütün kararlar (kilitleme, batch, temizlik, tek instance koordinasyonu) burada hazır geliyor. MassTransit v9 ticari lisanslı; v8'de kalma kararını serinin son yazısında konuşuyoruz.

Wolverine (MIT, yani Massachusetts Institute of Technology lisanslı) aynı işi "durable outbox" adıyla yapar: giden mesajları uygulamanın veritabanında iş verisiyle aynı transaction'da saklar, bir arka plan ajanı bunları gönderir; bir node çökerse saklanan mesajlar yeniden yüklenip gönderilir. PostgreSQL ve SQL Server'ı destekler, EF Core entegrasyonu vardır. Hangisini seçeceğin serinin yedinci yazısındaki karar tablosuna bağlı.

Bir de CDC (Change Data Capture) yolu var: Debezium gibi bir araç, veritabanının transaction log'unu (PostgreSQL'de logical replication) okur ve outbox tablosuna eklenen satırları Kafka'ya yayınlar. Uygulamada relay kodu kalmaz ve polling gecikmesi düşer, ama işletmen gereken yeni bir altyapı parçası (Kafka Connect, Debezium connector'ı, replication slot'ları) gelir. Kurulumu DevOps tarafının konusu.

Üretimde Dikkat

Outbox tablosu şişer. Biz işlenen satırı ProcessedAt ile işaretliyoruz; bir temizlik job'ı olmadan tablo her gün sipariş sayısı kadar büyür. Bir önceki yazıdaki ExecuteDeleteAsync'li parça parça temizlik burada da geçerli; işlenmiş satırları birkaç gün tutmak, "bu mesaj gerçekten gitti mi" sorusunun cevabını verir.

Bekleyen satırları izle. İki metric yeter: bekleyen satır sayısı ve en eski bekleyen satırın yaşı. Yaş birkaç saniyeyi geçiyorsa relay çalışmıyor, broker'a ulaşamıyor ya da yetişemiyor demektir. Bu, kullanıcının "siparişimin faturası neden gelmedi" sorusundan önce gelen sinyal.

Poison satır relay'i kilitler. Broker'ın her seferinde reddettiği bir mesaj (ör. olmayan bir exchange'e routing) batch'in her denemede baştan düşmesine yol açar ve arkasındaki satırlar hiç gitmez. Satıra bir deneme sayacı ve son hata kolonu ekleyip belirli bir sayıdan sonra satırı "failed" olarak ayırmak gerekir.

SQL Server'da aynı sorgu. FOR UPDATE SKIP LOCKED PostgreSQL'e özgü. SQL Server'da karşılığı SELECT TOP (50) ... WITH (UPDLOCK, READPAST, ROWLOCK) tablo ipuçları ve bir transaction'dır; mantık aynı.

Polling yerine bildirim. 1 saniyelik polling çoğu sistem için yeterli. Daha düşük gecikme gerekiyorsa endpoint commit'ten sonra relay'i bir Channel ile uyandırabilir ya da PostgreSQL'in LISTEN/NOTIFY'ı kullanılabilir; polling yine de güvenlik ağı olarak kalmalı.

Sık Düşülen Tuzaklar

Commit'ten sonra doğrudan publish etmek. SaveChangesAsync'in hemen ardından BasicPublishAsync çağırmak dual write'ın kendisidir: arada process çökerse ya da broker kapalıysa sipariş var, mesaj yok.

Relay'de SKIP LOCKED'u unutmak. Sadece FOR UPDATE ile ikinci instance birinci instance'ın kilidini bekler; çift yayın olmaz ama relay'ler sırayla çalışır. Kilit hiç yoksa iki relay aynı satırları okur ve her mesaj iki kez gider.

Outbox'ın exactly-once sağladığını sanmak. Relay at-least-once; consumer'da inbox ya da doğal idempotency olmadan duplicate'lar iş verisine ulaşır.

Relay'in sırayı koruduğunu varsaymak. Birden fazla relay ile batch'ler paralel gider; aggregate başına sıra gerekiyorsa bunu açıkça tasarla.

Her publish'te yeni bir MessageId üretmek. Relay bir satırı iki kez gönderirse iki kopyanın aynı MessageId'yi taşıması gerekir; kimliği satır oluşturulurken üret, relay'de değil.

İşlenmiş satırları hiç silmemek. Partial index sorguyu hızlı tutar, ama tablo ve yedekler büyümeye devam eder.

Küçük Bir Deneme

★☆☆ Outbox tablosuna Attempts ve LastError kolonları ekle. Publish başarısız olduğunda satır bazında artır ve 5 denemeden sonra satırı relay sorgusunun dışında bırak. Olmayan bir exchange'e routing ile test et.

★★☆ En eski bekleyen satırın yaşını (now() - min("OccurredAt")) veren bir health check yaz; 10 saniyeyi geçince Degraded dönsün. RabbitMQ'yu durdurup değeri izle.

★★★ Relay'i iki instance'la çalıştır ve aynı sipariş için OrderPlaced ile hemen ardından OrderCancelled yaz. Consumer'da varış sırasını log'la ve sıranın bozulabildiği bir durumu yakalamaya çalış (batch boyutunu 1'e indirmek ve relay'e yapay gecikme eklemek yardımcı olur). Sonra satırları CorrelationId'ye göre iki bölüme ayırıp her bölümü tek bir relay'e veren bir çözüm kur.

Mesaj artık API'den çıkıp outbox'tan, relay'den, broker'dan geçerek Billing'e ve oradan Notification'a gidiyor. Bu yolculuğun neresinde ne kadar zaman geçtiğini, hangi mesajın nerede takıldığını görmek için bir sonraki yazıda OpenTelemetry ile trace context'i mesaj header'ında taşıyor, queue depth ve consumer lag'i ölçüyoruz.