Projection'lar: Inline, Async ve Live Projection ile Read Model Üretmek

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile derlenip denenmiştir.

Önce Tahmin Et

Eylül'ün ilk günü. Depoda Dune ve Emma'nın stok kalemleri var; Ağustos sonunda nüsha girmiş, Ada'nın siparişi kargoya verilmiş, Eylül'de yeni nüshalar gelmiş, Alan ve Grace rezervasyon yapmış, bir Dune nüshası da hasarlı çıkmış. Bütün bunlar 06. yazıdaki StockCommandHandler ile event olarak yazıldı:

DateTimeOffset aug = new(2026, 8, 28, 10, 0, 0, TimeSpan.Zero);
DateTimeOffset sep = new(2026, 9, 1, 9, 0, 0, TimeSpan.Zero);
StockItemId dune = await Register("Dune", 20, aug);
StockItemId emma = await Register("Emma", 8, aug);

OrderId ada = new(Guid.CreateVersion7()), alan = new(Guid.CreateVersion7()), grace = new(Guid.CreateVersion7());
await handler.HandleAsync(new ReserveStock(dune, ada, 2, aug.AddHours(1)));
await handler.HandleAsync(new ShipReservedStock(dune, ada, aug.AddDays(1)));
await handler.HandleAsync(new ReceiveStock(dune, 10, sep));
await handler.HandleAsync(new ReserveStock(dune, alan, 1, sep.AddMinutes(10)));
await handler.HandleAsync(new ReserveStock(emma, alan, 1, sep.AddMinutes(10)));
await handler.HandleAsync(new ShipReservedStock(emma, alan, sep.AddHours(2)));
await handler.HandleAsync(new ReserveStock(emma, grace, 3, sep.AddHours(3)));
await handler.HandleAsync(new AdjustStock(dune, -1, AdjustmentReason.Damaged, sep.AddHours(4)));

Hemen ardından iki read model'e bakıyoruz: katalog sayfasının "stokta kaç adet var" rozeti ve depo müdürünün aylık stok hareketi raporu.

await using (IQuerySession query = store.QuerySession())
{
    StockLevel? duneLevel = await query.LoadAsync<StockLevel>(dune.Value);
    int monthlyRows = await query.Query<MonthlyStockMovement>().CountAsync();
    Console.WriteLine($"Inline  StockLevel(Dune): OnHand={duneLevel!.OnHand} Reserved={duneLevel.Reserved} Available={duneLevel.Available}");
    Console.WriteLine($"Async   MonthlyStockMovement rows before daemon: {monthlyRows}");
}

İki satır ne yazar? Rozet doğru sayıyı gösterir mi? Aylık raporda kaç satır vardır? Event'ler aynı, sorgular aynı veritabanına gidiyor; cevabı belirleyen tek şey bu read model'lerin hangi lifecycle ile kaydedildiği.

Projection: Event'lerden Soruya Göre Şekillenmiş Read Model

06. yazıda tek bir stream'den state kurduk: AggregateStreamAsync stream'in bütün event'lerini sırayla Evolve'dan geçirdi. Projection aynı fikrin genelleştirilmiş hâli: event'leri alıp okuma için tasarlanmış bir document'e (ya da tabloya) dönüştüren fonksiyon. CQRS (Command Query Responsibility Segregation) serisinin 03. yazısında read model'i domain event'lerinden besledik; event sourcing'de durum tersine döner, event'ler tek source of truth olduğu için her read model bir projection'dır ve istediğin zaman sıfırdan yeniden üretilebilir.

Marten'de iki ana tür var. Single-stream projection bir stream'den bir document üretir; document'in id'si stream id'sidir. Multi-stream projection ise event'leri kendi belirlediğin bir anahtara göre gruplar: "stok kalemi + ay", "sipariş", "kitap kategorisi" gibi. Bookstore'un stok defterinde üç soru var ve her biri ayrı bir read model'e karşılık geliyor:

Inline: Event'lerle Aynı Transaction'da

Katalog rozeti bir stok kalemine bakar ve her zaman güncel olmalı: müşteri "stokta 1 adet" görüp sepete eklediğinde rakamın gerçek olması gerekir. Bu bir single-stream projection ve Inline çalışıyor:

// Katalog sayfasındaki "Stokta 3 adet" rozeti: tek stream, Inline
public sealed record StockLevel(Guid Id, int OnHand, int Reserved)
{
    public int Available => OnHand - Reserved;
}

public sealed class StockLevelProjection : SingleStreamProjection<StockLevel, Guid>
{
    public override StockLevel Evolve(StockLevel? level, Guid id, IEvent e)
    {
        level ??= new StockLevel(id, 0, 0);
        return e.Data switch
        {
            StockReceived r => level with { OnHand = level.OnHand + r.Quantity },
            StockReserved r => level with { Reserved = level.Reserved + r.Quantity },
            ReservationReleased r => level with { Reserved = level.Reserved - r.Quantity },
            ReservedStockShipped s => level with { OnHand = level.OnHand - s.Quantity, Reserved = level.Reserved - s.Quantity },
            StockAdjusted a => level with { OnHand = level.OnHand + a.Delta },
            _ => level
        };
    }
}

Store'a dört projection'ı da lifecycle'larıyla kaydettik:

await using DocumentStore store = DocumentStore.For(opts =>
{
    opts.Connection(connectionString);
    opts.DatabaseSchemaName = "inventory";

    // Write model: hiçbir yerde saklanmaz, her command'da event'lerden kurulur
    opts.Projections.Add(new StockItemProjection(), ProjectionLifecycle.Live);
    // Event'lerle aynı transaction'da güncellenir
    opts.Projections.Add(new StockLevelProjection(), ProjectionLifecycle.Inline);
    // Async daemon arka planda günceller
    opts.Projections.Add(new MonthlyStockMovementProjection(), ProjectionLifecycle.Async);
    opts.Projections.Add(new PendingReservationProjection(), ProjectionLifecycle.Async);
});

Inline projection, SaveChangesAsync çağrısının içinde, event'lerin eklendiği PostgreSQL transaction'ında çalışır. Event commit edildiyse read model da güncellenmiştir; arada bir pencere yok. Bunun iki bedeli var. Birincisi her command biraz daha uzun sürer: event'in yanında bir document da yazılır. İkincisi daha önemli: projection'daki bir exception transaction'ı geri alır, yani bir read model'deki hata command'ın kendisini düşürür. "Rapor tablosunda bir bug var" diye sipariş alamaz hâle gelmek istemezsin. Inline'ı az sayıda, basit ve tek stream'e bakan projection'a sakla.

Async: Daemon ile Arka Planda

Aylık rapor birden fazla event tipini stok kalemi ve aya göre grupluyor. Anahtar event'in iş zamanından (At) geliyor, Marten'in event'i kaydettiği andan (Timestamp) değil: Ağustos'un son günü gelen ama gece yarısından sonra sisteme girilen bir sevkiyat Ağustos'a yazılmalı.

// Aylık stok hareketi: stok kalemi + ay başına bir document, Async
public sealed record MonthlyStockMovement(string Id, Guid StockItemId, string Month, int Received, int Shipped, int Adjusted);

public sealed class MonthlyStockMovementProjection : MultiStreamProjection<MonthlyStockMovement, string>
{
    public MonthlyStockMovementProjection()
    {
        Identity<StockReceived>(e => Key(e));
        Identity<ReservedStockShipped>(e => Key(e));
        Identity<StockAdjusted>(e => Key(e));
    }

    // Ay, event'in iş zamanından (At) gelir; kaydın yazıldığı andan değil
    private static string Key(StockEvent e) => $"{e.StockItemId.Value}:{e.At:yyyy-MM}";

    public override MonthlyStockMovement Evolve(MonthlyStockMovement? row, string id, IEvent e)
    {
        StockEvent stockEvent = (StockEvent)e.Data;
        row ??= new MonthlyStockMovement(id, stockEvent.StockItemId.Value, $"{stockEvent.At:yyyy-MM}", 0, 0, 0);
        return e.Data switch
        {
            StockReceived r => row with { Received = row.Received + r.Quantity },
            ReservedStockShipped s => row with { Shipped = row.Shipped + s.Quantity },
            StockAdjusted a => row with { Adjusted = row.Adjusted + a.Delta },
            _ => row
        };
    }
}

Kargo bekleyen rezervasyonlar ise sipariş başına bir document. Rezervasyon gelince satır eklenir, sevkiyat ya da iptal gelince satır çıkar; satır kalmayınca Evolve null döner ve Marten document'i siler:

// Kargoya verilmeyi bekleyen rezervasyonlar: sipariş başına bir document, Async
public sealed record PendingLine(Guid StockItemId, int Quantity, DateTimeOffset ReservedAt);
public sealed record PendingReservation(Guid Id, ImmutableList<PendingLine> Lines)
{
    public DateTimeOffset OldestReservedAt => Lines.Min(l => l.ReservedAt);
}

public sealed class PendingReservationProjection : MultiStreamProjection<PendingReservation, Guid>
{
    public PendingReservationProjection()
    {
        Identity<StockReserved>(e => e.OrderId.Value);
        Identity<ReservationReleased>(e => e.OrderId.Value);
        Identity<ReservedStockShipped>(e => e.OrderId.Value);
    }

    public override PendingReservation? Evolve(PendingReservation? pending, Guid orderId, IEvent e)
    {
        switch (e.Data)
        {
            case StockReserved r:
                pending ??= new PendingReservation(orderId, []);
                return pending with { Lines = pending.Lines.Add(new PendingLine(r.StockItemId.Value, r.Quantity, r.At)) };
            case ReservationReleased or ReservedStockShipped:
                if (pending is null)
                    throw new InvalidOperationException($"No pending reservation for order {orderId}");
                Guid stockItemId = ((StockEvent)e.Data).StockItemId.Value;
                ImmutableList<PendingLine> rest = pending.Lines.RemoveAll(l => l.StockItemId == stockItemId);
                // null dönmek document'i siler
                return rest.IsEmpty ? null : pending with { Lines = rest };
            default:
                return pending;
        }
    }
}

Bu iki projection Async. Komutlar çalıştığında Marten onlara dokunmaz; bir arka plan süreci olan async daemon, mt_events'i seq_id sırasıyla okuyup projection'ları besler. Açılıştaki sorunun cevabı burada:

Warning: The async daemon is disabled.
Projections Bookstore.Inventory.ReadModels.MonthlyStockMovementProjection, Bookstore.Inventory.ReadModels.PendingReservationProjection will not be executed without the async daemon enabled
Inline  StockLevel(Dune): OnHand=27 Reserved=1 Available=26
Async   MonthlyStockMovement rows before daemon: 0

Rozet doğru: 20 + 10 giriş, 2 sevk, 1 hasar, Alan'ın 1 rezervasyonu. Aylık raporda ise tek satır yok. Marten de daemon'un kapalı olduğunu baştan bir uyarıyla söylüyor. Bir console uygulamasında daemon'u kendimiz başlatıyoruz:

using IProjectionDaemon daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();
await daemon.WaitForNonStaleData(TimeSpan.FromSeconds(30));

Dune  2026-08  received=20 shipped=2 adjusted=0
Dune  2026-09  received=10 shipped=0 adjusted=-1
Emma  2026-08  received= 8 shipped=0 adjusted=0
Emma  2026-09  received= 0 shipped=1 adjusted=0
Pending order Alan: Dune x1, waiting 290 min
Pending order Grace: Emma x3, waiting 120 min

ASP.NET Core uygulamasında bunu builder.Services.AddMarten(...).AddAsyncDaemon(DaemonMode.HotCold) ile yaparsın. HotCold modunda birden fazla instance çalışsa bile her projection PostgreSQL advisory lock'u ile seçilen tek bir process'te koşar; Solo ise tek instance varsayımıdır. WaitForNonStaleData test ve demo içindir; production'da kullanıcıya eventual consistency ile nasıl yaşanacağını 03. yazıda konuşmuştuk. Marten'in QueryForNonStaleData<T>(timeout) sorgusu, gerektiğinde bir read model'in belli bir anın event'lerine yetişmesini beklemeni sağlıyor, ama her sorguda bunu yapıyorsan o projection aslında Inline olmak istiyordur.

Bekleyen rezervasyon read model'i bir de iş akışı besliyor. 05. yazıdaki kurala göre rezervasyonlar 30 dakikada düşer; ReleaseExpiredReservations command'ını çalıştıracak background job'ın "hangi siparişler 30 dakikadan eski?" sorusunu bütün stream'leri taramadan cevaplaması gerekir. Alan'ın rezervasyonu 290 dakikadır bekliyor; job onu bu tablodan bulur, kararı ise yine stok kaleminin kendi stream'inde, decider verir.

Live: Hiç Saklamamak

Üçüncü lifecycle 06. yazıdan tanıdık: StockItemProjection Live. Hiçbir yerde saklanmaz, FetchForWriting ve AggregateStreamAsync her çağrıldığında event'lerden kurulur. Write model için doğal tercih, çünkü command'ın kararı her zaman en güncel event'e göre verilmeli ve okuma zaten tek stream'le sınırlı. Read model olarak Live'ı nadiren seçersin: bir yöneticinin ayda bir açtığı, tek stream'e bakan bir ekran için ayrı bir tablo tutmaya değmeyebilir. 06. yazıdaki ölçüm de sınırı gösteriyor: stream uzadıkça her okuma pahalılaşır.

Checkpoint ve Rebuild

Daemon nerede kaldığını her projection için mt_event_progression tablosunda tutar:

await using NpgsqlDataSource dataSource = NpgsqlDataSource.Create(connectionString);
await using NpgsqlCommand command = dataSource.CreateCommand(
    "SELECT name, last_seq_id FROM inventory.mt_event_progression ORDER BY name");
await using NpgsqlDataReader reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
    Console.WriteLine($"{reader.GetString(0),-34} {reader.GetInt64(1)}");

HighWaterAllocationFence           12
HighWaterMark                      12
MonthlyStockMovement:All           12
PendingReservation:All             12

HighWaterMark, daemon'un "bundan öncesinde boşluk yok, güvenle işlenebilir" dediği en yüksek seq_id. 06. yazıdaki kendi tablomuzda yaşadığımız commit sırası problemini Marten bu işaretle çözüyor. Her projection'ın satırı da o projection'ın işlediği son event. Process yeniden başladığında daemon buradan devam eder; projection'ı yeniden kurmak istediğinde ise sıfırdan başlatırsın.

Rebuild'in asıl değeri, read model'in tasarımını sonradan değiştirebilmek. Depo müdürü bir ay sonra geliyor: "Düzeltmeleri tek kalemde görmek işime yaramıyor; sayım farkıyla kayıp ve hasarı ayır." İlişkisel bir rapor tablosunda bu, geçmiş ay satırları için veriyi tahmin etmek demekti. Burada geçmişin tamamı event'lerde duruyor. Projection'ın yeni hâli:

// v2: düzeltmeyi sayım farkı ve kayıp/hasar olarak ayır
public sealed record MonthlyStockMovement(
    string Id, Guid StockItemId, string Month, int Received, int Shipped, int Corrected, int Lost);

public sealed class MonthlyStockMovementProjection : MultiStreamProjection<MonthlyStockMovement, string>
{
    public MonthlyStockMovementProjection()
    {
        Identity<StockReceived>(e => Key(e));
        Identity<ReservedStockShipped>(e => Key(e));
        Identity<StockAdjusted>(e => Key(e));
    }

    private static string Key(StockEvent e) => $"{e.StockItemId.Value}:{e.At:yyyy-MM}";

    public override MonthlyStockMovement Evolve(MonthlyStockMovement? row, string id, IEvent e)
    {
        StockEvent stockEvent = (StockEvent)e.Data;
        row ??= new MonthlyStockMovement(id, stockEvent.StockItemId.Value, $"{stockEvent.At:yyyy-MM}", 0, 0, 0, 0);
        return e.Data switch
        {
            StockReceived r => row with { Received = row.Received + r.Quantity },
            ReservedStockShipped s => row with { Shipped = row.Shipped + s.Quantity },
            StockAdjusted { Reason: AdjustmentReason.CountCorrection } a => row with { Corrected = row.Corrected + a.Delta },
            StockAdjusted a => row with { Lost = row.Lost - a.Delta },
            _ => row
        };
    }
}

Yeni kodu deploy edip rebuild çalıştırıyoruz:

using IProjectionDaemon daemon = await store.BuildProjectionDaemonAsync();
await daemon.RebuildProjectionAsync<MonthlyStockMovementProjection>(CancellationToken.None);

Before rebuild (old documents, new code):
Dune  2026-08  received=20 shipped=2 corrected=0 lost=0
Dune  2026-09  received=10 shipped=0 corrected=0 lost=0
Emma  2026-08  received= 8 shipped=0 corrected=0 lost=0
Emma  2026-09  received= 0 shipped=1 corrected=0 lost=0
After rebuild:
Dune  2026-08  received=20 shipped=2 corrected=0 lost=0
Dune  2026-09  received=10 shipped=0 corrected=0 lost=1
Emma  2026-08  received= 8 shipped=0 corrected=0 lost=0
Emma  2026-09  received= 0 shipped=1 corrected=0 lost=0

İlk dört satıra dikkat: rebuild'den önce yeni kod eski document'leri okuyor ve yeni field'lar sıfır görünüyor. Kod ile read model arasındaki bu tutarsızlık, rebuild bitene kadar kullanıcıya yanlış rapor gösterir. Marten'in bunun için önerdiği yol blue/green: projection class'ındaki ProjectionVersion'ı artırırsın, yeni sürüm ayrı tablolara yazar ve eski sürüm trafiğe hizmet etmeye devam ederken arka planda yetişir; yetişince trafiği yeni sürüme çevirirsin.

Sık Düşülen Tuzaklar

Projection'dan dış servis çağırmak

Projection'a "rezervasyon gelince depoya e-posta at" ya da "kitabın fiyatını Catalog API'sinden (Application Programming Interface) çek" gibi bir çağrı eklemek cazip gelir. İki sorun var. Rebuild bütün geçmişi baştan oynatır; e-posta gönderen bir projection o gün yüzlerce eski e-postayı yeniden gönderir. Dış servis yavaşsa ya da düşerse, projection'ın ilerlemesi ona bağlanır ve rebuild süresi servisin cevap süresiyle çarpılır. Projection saf kalmalı: event'ten read model'e bir fonksiyon. Dışarıya bir şey söylemek gerekiyorsa bu ayrı bir subscription'ın ya da outbox'ın işi; ilerideki Messaging ve Event-Driven Architecture serisinin konusu. Marten'in RaiseSideEffects hook'u da tam bu yüzden rebuild sırasında çalışmaz.

Projection hatasının daemon'u ve read model'i durdurması

Eski sistemden veri aktarılırken rezervasyonu olmayan bir sevkiyat event'i stream'e eklendi. PendingReservationProjection böyle bir durumda exception fırlatıyor:

// Eski sistemden aktarılan bir kayıt: rezervasyonu olmayan bir sevkiyat
StockItemId hobbit = await Register("The Hobbit", 5, DateTimeOffset.UtcNow);
await using (IDocumentSession session = store.LightweightSession())
{
    session.Events.Append(hobbit.Value,
        new ReservedStockShipped(hobbit, new OrderId(Guid.CreateVersion7()), 1, DateTimeOffset.UtcNow));
    await session.SaveChangesAsync();
}

using IProjectionDaemon daemon = await store.BuildProjectionDaemonAsync();
await daemon.StartAllAsync();
await daemon.WaitForNonStaleData(TimeSpan.FromSeconds(30));
await daemon.StopAllAsync();

await using (IQuerySession query = store.QuerySession())
{
    IReadOnlyList<DeadLetterEvent> dead = await query.Query<DeadLetterEvent>().ToListAsync();
    foreach (DeadLetterEvent d in dead)
        Console.WriteLine($"Dead letter: {d.ProjectionName} seq={d.EventSequence} {d.ExceptionType}");
    Console.WriteLine($"Pending documents before rebuild: {await query.Query<PendingReservation>().CountAsync()}");
}

try
{
    await daemon.RebuildProjectionAsync<PendingReservationProjection>(CancellationToken.None);
}
catch (Exception ex)
{
    Console.WriteLine($"Rebuild failed: {ex.GetType().Name}: {ex.Message.Split('\n')[0]}");
}
await using (IQuerySession query = store.QuerySession())
    Console.WriteLine($"Pending documents after failed rebuild: {await query.Query<PendingReservation>().CountAsync()}");

Dead letter: PendingReservation seq=15 InvalidOperationException
Pending documents before rebuild: 2
Rebuild failed: ApplyEventException: Failure to apply event #15 Id(457a6300-9b89-4e3a-81ed-c71fed0b4a0a)
Pending documents after failed rebuild: 0
HighWaterAllocationFence           15
HighWaterMark                      15
MonthlyStockMovement:All           15

Sürekli çalışmada Marten'in varsayılanı hatalı event'i atlamak (SkipApplyErrors = true): event mt_doc_deadletterevent tablosuna exception'ıyla birlikte yazıldı ve daemon devam etti. Rebuild'de varsayılan tersi (RebuildErrors.SkipApplyErrors = false) ve sonucu çıktının son satırlarında görüyorsun: rebuild durdu, read model boş kaldı, PendingReservation'ın checkpoint satırı da yok oldu. Sabah rebuild başlatan bir ekip, öğlen "bekleyen sipariş yok" ekranıyla karşılaşabilir. İki ders var: dead letter tablosunu izle ve hepsini rebuild'den önce temizle; projection kodunda beklenmeyen durumu exception yerine görünür bir şekilde ele alabiliyorsan (ör. ayrı bir "anomali" field'ı) ele al.

Rebuild süresini hesaba katmamak

Aynı makinede, Docker'daki PostgreSQL ile aylık rapor projection'ını iki farklı büyüklükte rebuild ettim:

Rebuilt MonthlyStockMovement from 100,000 events in 2.6 s
Rebuilt MonthlyStockMovement from 400,000 events in 29.5 s

İkinci koşuda 1.000 stream'in her biri 400 event taşıyor ve 14.000 document üretiliyor; süre event sayısıyla doğru orantılı değil, document sayısı ve stream başına event sayısı da belirleyici. Bookstore günde ~800 sipariş alıyor; her sipariş birkaç stok event'i üretiyorsa yılda milyonlarca event birikir. Rebuild'i bir deploy adımına koymadan önce production büyüklüğündeki veride ölç. Marten 9.13'ten beri MaxConcurrentRebuildsPerDatabase ile aynı anda çalışan rebuild sayısı da sınırlanıyor; büyük store'larda bağlantı havuzunu tüketmemek için bu ayara bak.

Her read model'i Inline yapmak

"Eventual consistency ile uğraşmayalım" diye her şeyi Inline yapmak, command'ları yavaşlatır ve yukarıdaki hatayı command'a taşır. Multi-stream projection'ları Inline çalıştırmak ayrıca risklidir: iki farklı stream'e yazan iki command aynı ayın document'ini aynı anda günceller. Marten'in dokümantasyonu da multi-stream projection'ları neredeyse her zaman Async çalıştırmayı öneriyor.

Ne Zaman Hangisi?

Küçük Bir Deneme

★☆☆ StockLevelProjection'ı Async'e çevir ve açılıştaki programı tekrar çalıştır. Rozet ne gösteriyor? Kullanıcıya bunu nasıl anlatırdın (03. yazıdaki stratejilerden birini seç)?

★★☆ PendingReservationProjection'a, sevkiyat geldiğinde siparişin kaç dakika beklediğini tutan bir ShippedAfterMinutes listesi ekle ve projection'ı rebuild et. Geçmiş siparişler için de değer çıkıyor mu? Neden?

★★★ Açılıştaki demo verisinin üstüne poison event'i ekle ve bu sefer rebuild'den önce dead letter'daki event'i düzelt: rezervasyonu olmayan sevkiyat için projection'ın exception yerine bir "anomali" satırı yazmasını sağla. Rebuild'in başarılı bittiğini ve bekleyen siparişlerin geri geldiğini göster.

Bu yazıda event'lerden istediğimiz kadar read model ürettik ve tasarımını rebuild ile değiştirdik. Ama event'lerin kendisini değiştirmedik. Sıradaki yazıda o daha zor soruya geçiyoruz: bir event tipinin şekli değişmek zorunda kaldığında, yıllar önce yazılmış event'leri kırmadan bunu nasıl yaparız ve uzayan stream'leri snapshot ile nasıl kısa tutarız.