.NET'te MongoDB İleri: Aggregation, Bulk Write, Transaction, Change Stream ve Testcontainers

Sürüm notu: Bu yazıdaki kodlar .NET 10, MongoDB.Driver 3.14, Microsoft.Extensions.Hosting 10.0, Testcontainers.MongoDb 4.16 ve xunit.v3 4.0 ile derlenip MongoDB 9.0.8 Community Server üzerinde çalıştırılmıştır. Aynı komutlar MongoDB 8.0.38'de de aynı çıktıyı verdi; yalnız transaction gövdesinin kaç kez çalıştığı her koşuda değişiyor.

Önce Tahmin Et

Bookstore'da (sitedeki serilerin ortak örnek uygulaması, bir online kitapçı) sipariş vermek birden çok document'e dokunuyor: siparişteki her kitabın stoğu düşmeli ve sipariş kaydedilmeli; biri olmazsa hiçbiri olmamalı. 4. yazıda bunun tek document'e sığmadığı yer için transaction'ı işaret etmiştik. Sipariş yerleştirme sınıfı transaction gövdesinin kaç kez çalıştığını sayıyor. 20 müşteri aynı anda birer Dune alıyor:

// 20 müşteri aynı anda Dune alıyor: write conflict'leri WithTransactionAsync yeniden dener
int before = placement.Attempts;
await Task.WhenAll(Enumerable.Range(0, 20).Select(i => placement.PlaceAsync(SampleOrder(400 + i, (26, 1)))));
await ShowAsync("20 concurrent orders");
Console.WriteLine($"transaction bodies run: {placement.Attempts - before} for 20 orders");

Yirmi transaction, yirmi gövde çalıştırması. Öyle mi?

20 concurrent orders         Dune 29, 1984 51, new orders 21
transaction bodies run: 62 for 20 orders

Stok doğru (49'dan 29'a, 20 adet), siparişlerin hepsi kaydedildi, ama gövde 62 kez çalıştı; aynı koşu 8.0.38'de 59, bir başka denemede 88 verdi. Yirmi transaction aynı kitap document'ini değiştirmeye çalışınca MongoDB ilk yazanı bırakıp diğerlerini WriteConflict hatasıyla iptal ediyor. Bu hata TransientTransactionError etiketini taşıyor ve WithTransactionAsync böyle bir hatada transaction'ı baştan, gövdeyle birlikte yeniden çalıştırıyor. Sonuç doğru, çünkü gövde yalnız veritabanına yazıyor; gövdenin içinde bir e-posta gönderseydin müşteri üç e-posta alabilirdi. Bu yazıda .NET driver'ının ileri tarafını bu türden ayrıntılarla ele alıyoruz: fluent Aggregate(), bulk write, transaction, change stream ve Testcontainers ile integration testi. Komutlar 9. yazıdaki konsol uygulamasında (aggregate, bulk, transaction, watch) ve 6. yazının 06-bookstore-indexes.js index'lerini varsayıyor.

Aggregation: Fluent API, LINQ ve BsonDocument Stage'leri

7. yazıdaki aylık ciroyu .NET'ten yazalım. Aggregate() tipli bir fluent API veriyor: Match ve Group C# ifadelerinden çevriliyor. Tipli API'nin kapsamadığı bir stage gerektiğinde AppendStage ile BsonDocument olarak ekleniyor:

// Tipli stage'ler: Match ve Group C# ifadelerinden çevrilir
var monthly = orders.Aggregate()
    .Match(o => o.Status != OrderStatus.Cancelled && o.OrderedAt >= from && o.OrderedAt < to)
    .Group(o => new { o.OrderedAt.Year, o.OrderedAt.Month }, g => new { g.Key.Year, g.Key.Month, Revenue = g.Sum(o => o.Total) })
    .SortBy(m => m.Year).ThenBy(m => m.Month);
Console.WriteLine(monthly.ToString());

// Tipli API'nin kapsamadığı bir stage: BsonDocument olarak ekle
List<BsonDocument> withYtd = await monthly
    .AppendStage<BsonDocument>(new BsonDocument("$setWindowFields", new BsonDocument
    {
        { "sortBy", new BsonDocument { { "Year", 1 }, { "Month", 1 } } },
        { "output", new BsonDocument("Ytd", new BsonDocument
            {
                { "$sum", "$Revenue" },
                { "window", new BsonDocument("documents", new BsonArray { "unbounded", "current" }) }
            }) }
    }))
    .ToListAsync();

aggregate([{ "$match" : { "status" : { "$ne" : "Cancelled" }, "orderedAt" : { "$gte" : { "$date" : "2025-01-01T00:00:00Z" }, "$lt" : { "$date" : "2025-04-01T00:00:00Z" } } } }, { "$group" : { "_id" : { "Year" : { "$year" : "$orderedAt" }, "Month" : { "$month" : "$orderedAt" } }, "__agg0" : { "$sum" : "$total" } } }, { "$project" : { "Year" : "$_id.Year", "Month" : "$_id.Month", "Revenue" : "$__agg0", "_id" : 0 } }, { "$sort" : { "Year" : 1, "Month" : 1 } }])
  2025-01  revenue   8959.20  ytd   8959.20
  2025-02  revenue  13727.50  ytd  22686.70
  2025-03  revenue  20367.60  ytd  43054.30

Sayılar 7. ve 8. yazıdakilerle aynı. Çeviriye dikkat: enum "Cancelled" string'ine (9. yazıdaki convention), OrderedAt.Year $year ifadesine, anonim tipin alanları PascalCase adlara (Year, Revenue) çevrildi. Sonraki BsonDocument stage'i bu adları kullanmak zorunda; tipli kısımla elle yazılan kısım arasındaki sözleşme alan adları ve bunu derleyici kontrol etmiyor. Pratik kural: pipeline'ın gövdesini tipli yaz, $setWindowFields, $densify ya da $graphLookup gibi tipli karşılığı zayıf stage'leri BsonDocument ile ekle ve ToString() çıktısını bir kez gözle kontrol et. Aynı sorgu AsQueryable() ile LINQ (C#'ta sorguyu dilin içinde yazmayı sağlayan özellik) olarak da yazılabiliyor; 9. yazıda gördüğümüz gibi LINQ da sonunda bir pipeline üretiyor.

Bulk Write: Ordered, Unordered ve Client-Level

Tedarikçiden gelen bir fiyat listesini içeri almak yüzlerce update demek; her birini ayrı istekle göndermek yerine BulkWriteAsync hepsini birkaç komutta götürüyor. 3. yazıdaki insertMany'nin ordered/unordered farkını burada karışık operasyonlarla görelim: ikinci model aynı _id'li bir insert ve hata veriyor.

// Fiyat listesi aktarımı: ikinci model aynı _id'li ve hata verir
List<WriteModel<Book>> models =
[
    new UpdateOneModel<Book>(Builders<Book>.Filter.Eq(b => b.Id, 26), Builders<Book>.Update.Set(b => b.ListPrice, 399.90m)),
    new InsertOneModel<Book>(dune),
    new UpdateOneModel<Book>(Builders<Book>.Filter.Eq(b => b.Id, 35), Builders<Book>.Update.Set(b => b.ListPrice, 529.90m))
];

    await books.BulkWriteAsync(models, new BulkWriteOptions { IsOrdered = ordered });

// MongoDB 8.0+: birden çok collection'a tek bulkWrite komutu (transaction değil)
ClientBulkWriteResult clientResult = await client.BulkWriteAsync(
[
    new BulkWriteInsertOneModel<Order>("bookstore.orders", order),
    new BulkWriteUpdateOneModel<Book>("bookstore.books",
        Builders<Book>.Filter.Eq(b => b.Id, 26), Builders<Book>.Update.Inc(b => b.Stock!.OnHand, -1))
]);

IsOrdered = True : modified 1, error at index 1 (code 11000), not processed 1
IsOrdered = False: modified 2, error at index 1 (code 11000), not processed 0
client bulk write: inserted 1, modified 1

Ordered (varsayılan) ilk hatada durdu: birinci fiyat güncellendi, üçüncü model hiç işlenmedi ve UnprocessedRequests'te kaldı. Unordered hatayı raporlayıp devam etti ve iki fiyatı da güncelledi. İkisinde de MongoBulkWriteException<Book> geliyor; içinde hem sonuç sayıları hem hata listesi var, yani "kısmen başarılı" bir bulk'ı ayrıştırmak mümkün. Sıra önemliyse (önce kitabı ekle, sonra fiyatını güncelle) ordered, operasyonlar bağımsızsa unordered; unordered sunucunun işi paralelleştirmesine de izin veriyor.

Collection'ın BulkWriteAsync'i tek bir collection'a yazıyor. MongoDB 8.0 ile gelen bulkWrite komutunu kullanan IMongoClient.BulkWriteAsync ise farklı collection'lara ve database'lere giden operasyonları tek istekte gönderiyor; yukarıda sipariş insert'i ve stok update'i tek komutla gitti. Bu bir transaction değil: operasyonlardan biri hata verirse diğerleri yine yazılmış olabilir. Ağ turunu azaltan bir araç, atomiklik aracı değil. Doküman bu metodun MongoDB 8.0 ve sonrasını istediğini belirtiyor; 7.0'a bağlanan bir uygulamada çalışmaz.

Transaction: WithTransactionAsync ve Otomatik Retry

"Önce Tahmin Et"teki sınıfın tamamı bu. Stok düşürme 3. yazıdaki koşullu $inc; transaction yalnız birden çok kitabı ve siparişi birbirine bağlıyor:

// Sipariş verir: bütün stok düşürmeleri ve sipariş insert'i ya birlikte başarılı olur ya hiçbiri
public sealed class OrderPlacement(IMongoClient client, IMongoCollection<Book> books, IMongoCollection<Order> orders)
{
    private int _attempts;

    // Transaction gövdesinin kaç kez çalıştığı (retry'lar dahil)
    public int Attempts => _attempts;

    public async Task<Order> PlaceAsync(Order order, CancellationToken cancellationToken = default)
    {
        using IClientSessionHandle session = await client.StartSessionAsync(cancellationToken: cancellationToken);
        return await session.WithTransactionAsync(async (s, ct) =>
        {
            Interlocked.Increment(ref _attempts);
            foreach (OrderItem item in order.Items)
            {
                UpdateResult result = await books.UpdateOneAsync(s,
                    b => b.Id == item.BookId && b.Stock!.OnHand >= item.Quantity,
                    Builders<Book>.Update.Inc(b => b.Stock!.OnHand, -item.Quantity),
                    cancellationToken: ct);
                if (result.ModifiedCount == 0)
                {
                    throw new InsufficientStockException(item.BookId);   // transaction'ı iptal eder
                }
            }
            await orders.InsertOneAsync(s, order, cancellationToken: ct);
            return order;
        }, cancellationToken: cancellationToken);
    }
}

start                        Dune 51, 1984 52, new orders 0
order 380 (Dune 2, 1984 1)   Dune 49, 1984 51, new orders 1
order 381: Not enough stock for book 35
after order 381              Dune 49, 1984 51, new orders 1

Sipariş 381'de Dune'un stoğu önce düştü, sonra 1984'ten 100 adet istendiği için exception fırladı ve transaction iptal edildi: Dune 49'da kaldı, sipariş yazılmadı. Her operasyona session'ı (s) vermek şart; session'sız bir UpdateOneAsync transaction'ın dışında, hemen ve kalıcı olarak yazar ve kimse fark etmez. WithTransactionAsync'in iki retry'ı var: TransientTransactionError etiketli hatalarda bütün gövdeyi, commit sonucu belirsiz kaldığında (UnknownTransactionCommitResult) yalnız commit'i yeniden deniyor. Kendi exception'ın (burada InsufficientStockException) etiket taşımadığı için retry yok, transaction iptal edilip exception sana geliyor. Driver dokümanının uyarısı: gövdede bir exception'ı yakalarsan yeniden fırlat; yutarsan sonsuz döngüye kadar giden davranışlar görebilirsin.

Transaction'ın da bedeli var: varsayılan ömür 60 saniye, kilitler ve snapshot'lar o süre boyunca tutuluyor, 9.0'da aynı anda açık transaction sayısına bir üst sınır geldi (12. yazı). Bu yüzden önce modele bak: 4. yazıdaki gibi ödeme ve durum değişikliği tek document'teyse transaction gerekmiyor, tek bir update zaten atomik. Tek operasyonluk yazmalar için de transaction yerine retryable writes yeterli; driver varsayılan olarak açık tutuyor ve bir ağ hatasında aynı yazmayı bir kez, sunucunun tekrarı tanıyacağı şekilde yeniden gönderiyor. Aynı kalıp 5. yazıdaki yorum + kitaptaki özet ikilisi ve hepsi ya da hiçbiri yazılması gereken bir InsertManyAsync için de geçerli: aynı session'la WithTransactionAsync içine al.

Change Stream: BackgroundService ve Resume Token

4. yazıda yazar adını kitaplara kopyaladık ve kopyanın bedelinin güncelleme olduğunu söyledik. Change stream bir collection'daki değişiklikleri oplog (replica set'in değişiklik günlüğü) üzerinden sırayla dinlemeyi sağlıyor; lab'ı baştan replica set kurmamızın bir nedeni bu. Yazar adı değişince kitaplardaki kopyaları güncelleyen bir BackgroundService:

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    IMongoCollection<BsonDocument> tokens = database.GetCollection<BsonDocument>("resume_tokens");
    BsonDocument? saved = await tokens.Find(new BsonDocument("_id", TokenId)).FirstOrDefaultAsync(stoppingToken);

    PipelineDefinition<ChangeStreamDocument<Author>, ChangeStreamDocument<Author>> pipeline =
        new EmptyPipelineDefinition<ChangeStreamDocument<Author>>()
            .Match(c => c.OperationType == ChangeStreamOperationType.Update);
    ChangeStreamOptions options = new()
    {
        FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
        ResumeAfter = saved?["token"].AsBsonDocument      // ilk çalışmada null: şu andan itibaren başla
    };

    using IChangeStreamCursor<ChangeStreamDocument<Author>> cursor =
        await authors.WatchAsync(pipeline, options, stoppingToken);

    while (await cursor.MoveNextAsync(stoppingToken))
    {
        foreach (ChangeStreamDocument<Author> change in cursor.Current)
        {
            Author author = change.FullDocument;
            string name = $"{author.FirstName} {author.LastName}";
            UpdateResult result = await books.UpdateManyAsync(
                Builders<Book>.Filter.ElemMatch(b => b.Authors, a => a.AuthorId == author.Id),
                Builders<Book>.Update.Set("authors.$[a].name", name),
                new UpdateOptions
                {
                    ArrayFilters = [new BsonDocumentArrayFilterDefinition<BsonDocument>(new BsonDocument("a.authorId", author.Id))]
                },
                stoppingToken);

            // Token'ı yalnız iş bittikten sonra kaydet: restart en fazla bu tek değişikliği tekrarlar
            await tokens.ReplaceOneAsync(new BsonDocument("_id", TokenId),
                new BsonDocument { { "_id", TokenId }, { "token", change.ResumeToken } },
                new ReplaceOptions { IsUpsert = true }, stoppingToken);
            logger.LogInformation("Author {AuthorId} is now {Name}: {Books} book(s) updated", author.Id, name, result.ModifiedCount);
        }
    }
}

watch komutu servisi başlatıyor, Frank Herbert'in adını değiştiriyor, servisi durduruyor, servis kapalıyken adı bir kez daha değiştiriyor ve servisi yeniden başlatıyor:

info: Bookstore.MongoLab.AuthorNameSync[0] Watching authors (from now)
info: Bookstore.MongoLab.AuthorNameSync[0] Author 18 is now Frank P. Herbert: 2 book(s) updated
service stopped; author 18 renamed to Franklin
info: Bookstore.MongoLab.AuthorNameSync[0] Watching authors (resuming after saved token)
info: Bookstore.MongoLab.AuthorNameSync[0] Author 18 is now Franklin Herbert: 2 book(s) updated
books: Dune by Franklin Herbert, Dune Messiah by Franklin Herbert

Servis kapalıyken olan değişiklik kaybolmadı: yeniden başlayınca kayıtlı resume token'dan sonrasını okudu ve iki kitabı güncelledi. Token'ı işi bitirdikten sonra kaydetmek bilinçli bir seçim; servis ikisinin arasında çökerse aynı değişiklik bir kez daha işlenir, bu yüzden işin kendisi idempotent olmalı (adı aynı değere iki kez yazmak zararsız). UpdateLookup bir ayrıntıyı saklıyor: tam document değişiklik anındaki hâliyle değil, change stream'in onu işlediği andaki hâliyle geliyor. Servis uzun süre geride kalırsa ara değişiklikler için bile son adı görürsün; burada istediğimiz bu, ama her senaryoda değil. Resume token'ın bir son kullanma tarihi de var: oplog'un tuttuğu geçmiş kadar geriye dönebilirsin. Servis oplog penceresinden uzun süre kapalı kalırsa resume başarısız olur ve tam bir yeniden eşitleme gerekir; oplog boyutu 13. yazıda.

Aynı yapı kopyalanmış alanların ötesine de taşınıyor: sipariş değişikliğinde cache'teki kaydı silmek Caching serisinin "Cache Invalidation Stratejileri" yazısındaki event ile invalidation'ın, değişikliği bir message broker'a aktarmak ise Message Queue'larla Çalışmak serisindeki ".NET'te Transactional Outbox" yazısının MongoDB karşılığı. Outbox'tan farkı, olayların ayrı bir tabloya değil oplog'a yazılması; ama teslim garantisi yine "en az bir kez" ve tüketici yine idempotent olmalı. 8. yazıdaki materialized view'ı yenilemek de aynı kalıba oturuyor.

Testcontainers ile Integration Testi

Transaction ve change stream kodunu mock'larla test etmek anlamlı değil; asıl soru gerçek bir MongoDB'nin ne yaptığı. Testcontainers test sırasında Docker'da gerçek bir MongoDB başlatıyor. Transaction replica set istediği için WithReplicaSet ile tek node'lu bir replica set:

// Test koşusu başına tek node'lu bir replica set: transaction ve change stream replica set ister
public sealed class MongoFixture : IAsyncLifetime
{
    private readonly MongoDbContainer _container = new MongoDbBuilder("mongodb/mongodb-community-server:9.0-ubuntu2204")
        .WithReplicaSet("rs0")
        .Build();

    public IMongoClient Client { get; private set; } = null!;

    public async ValueTask InitializeAsync()
    {
        await _container.StartAsync();
        BookstoreMongo.RegisterSerialization();
        Client = new MongoClient(_container.GetConnectionString());
    }

    public ValueTask DisposeAsync() => _container.DisposeAsync();
}

[Fact]
public async Task Insufficient_stock_rolls_back_the_whole_order()
{
    (OrderPlacement placement, IMongoCollection<Book> books, IMongoCollection<Order> orders) = await SeedAsync();

    await Assert.ThrowsAsync<InsufficientStockException>(() =>
        placement.PlaceAsync(NewOrder(2, (26, 1), (35, 5)), TestContext.Current.CancellationToken));

    // Önce Dune'un stoğu düştü, sonra transaction iptal edildi
    Assert.Equal(3, (await books.Find(b => b.Id == 26).FirstAsync(TestContext.Current.CancellationToken)).Stock!.OnHand);
    Assert.Equal(0, await orders.CountDocumentsAsync(o => o.Id == 2, cancellationToken: TestContext.Current.CancellationToken));
}

dotnet test

Test run summary: Passed!
  total: 2
  failed: 0
  succeeded: 2
  skipped: 0

Fixture container'ı bütün test sınıfı için bir kez başlatıyor; her test SeedAsync içinde kendi adı rastgele bir database'ini kuruyor, böylece testler birbirinin verisini görmüyor. Container'ın image'ı lab'daki ile aynı, yani test 9.0'ın davranışını sınıyor. İki ayar gerekti. BookstoreMongo.RegisterSerialization artık birden çok kez çağrılabiliyor (ilk çağrıdan sonra hiçbir şey yapmıyor), çünkü fixture ve uygulama aynı process'te global kayıtları ikinci kez yapmaya kalkınca driver aynı tip için ikinci serializer'ı reddediyor. Repo köküne de test.runner değeri Microsoft.Testing.Platform olan bir global.json girdi: xunit.v3 4.x ile .NET 10 SDK'da dotnet test, bu ayar olmadan VSTest yolunun artık desteklenmediğini söyleyen bir hatayla duruyor. Test için makinede Docker'ın çalışıyor olması yeterli; Windows'ta Docker Desktop.

Üretimde Dikkat

Change stream servisini tek kopya çalıştır ya da işi paylaştır. Uygulama üç instance'la çalışıyorsa üç BackgroundService aynı değişikliği üç kez işler. Ya servisi ayrı, tek instance'lı bir worker'a taşı ya da bir lider seçimi (örneğin bir lock document'i) kur. Token'ı hangi instance'ın kaydettiği de aynı soru.

Transaction'ları kısa tut ve ölç. Gövdede ağ çağrısı, dosya işlemi ya da kullanıcı beklemesi olmamalı; bunlar hem süreyi uzatır hem retry'da tekrar eder. "Önce Tahmin Et"teki gibi aynı document'e yoğun yazılan yerlerde write conflict ve retry sayısını izle: sık retry, modelin aynı document'te bir hot spot yarattığının işareti.

Bulk işlerini parçala. Driver büyük bir model listesini sunucunun mesaj sınırlarına göre komutlara böler, ama yüz binlerce modeli tek listede bellekte tutmak ve hata durumunda nereden devam edeceğini bilmemek uygulamanın sorunu. Birkaç bin modellik parçalar ve her parçadan sonra ilerlemeyi kaydetmek, 5. yazıdaki migration'daki gibi kaldığı yerden devam etmeyi mümkün kılar.

Sık Düşülen Tuzaklar

Resume token'ı kaydetmeden restart. Token yalnız bellekte tutulursa her deploy'da servis "şu andan itibaren" başlar ve aradaki değişiklikler sessizce kaybolur; yukarıdaki "Franklin" değişikliği kitaplara hiç ulaşmazdı. Token'ı kalıcı bir yere, işi bitirdikten sonra yaz.

Uzun transaction. Bir rapor ya da toplu güncellemeyi transaction'a sarmak, binlerce document'i kilitleyip 60 saniyelik sınıra yaklaşmak ve aynı document'lere yazan diğer işlemleri write conflict'e sokmak demek. Transaction birkaç document'i bağlamak için; toplu işler bulk write ve idempotent adımlarla yapılır.

Gövdede yan etki. Transaction gövdesinde e-posta, HTTP çağrısı ya da bir message broker'a publish, retry'da tekrar eder ve transaction iptal edilse bile geri alınmaz. Yan etkiyi commit'ten sonra yap ya da outbox ile değişikliğin parçası hâline getir.

Küçük Bir Deneme

★☆☆ OrderPlacement'ta UpdateOneAsync çağrısından session parametresini (s) kaldır ve "Insufficient stock" testini çalıştır. Hangi assertion neden kırılıyor?

★★☆ 5. yazıdaki "yorum ekle + kitaptaki özeti güncelle" işini WithTransactionAsync ile yazan bir ReviewService ve bunun için iki integration testi yaz: başarılı yol ve geçersiz puanla (rating: 6) iptal edilen yol.

★★★ AuthorNameSync'i kategori adları için de genişlet ve 8. yazıdaki daily_sales materialized view'ını sipariş değişikliklerinde yalnız etkilenen günler için yenileyen ikinci bir change stream servisi yaz. Servisi iki instance'la çalıştırıp aynı değişikliğin iki kez işlenmesini engelleyen basit bir lock document'i tasarla.

Bir sonraki yazıda aynı Bookstore verisine MongoDB'nin Entity Framework Core (EF Core) provider'ıyla bakıyoruz: UseMongoDB ile DbContext, owned type'ların embed olarak eşlenmesi, change tracking ve provider'ın resmî sınırları; hangi işte native driver, hangisinde EF Core.