Advanced MongoDB in .NET: Aggregation, Bulk Writes, Transactions, Change Streams and Testcontainers

Version note: The code in this post was built with .NET 10, MongoDB.Driver 3.14, Microsoft.Extensions.Hosting 10.0, Testcontainers.MongoDb 4.16 and xunit.v3 4.0 and run against MongoDB 9.0.8 Community Server. The same commands gave the same output on MongoDB 8.0.38; only the number of times the transaction body runs changes from run to run.

Guess First

In Bookstore, the online bookstore used as the shared example across this site's series, placing an order touches several documents: the stock of every book in the order must go down and the order must be saved; if one fails, none should happen. In post 4 we pointed to transactions for the cases where this does not fit in a single document. The order placement class counts how many times the transaction body runs. 20 customers each buy one copy of Dune at the same moment:

// 20 customers buy Dune at the same moment: write conflicts are retried by WithTransactionAsync
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");

Twenty transactions, twenty runs of the body. Right?

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

The stock is right (from 49 down to 29, 20 copies) and all orders were saved, but the body ran 62 times; the same run gave 59 on 8.0.38 and 88 in another attempt. When twenty transactions try to change the same book document, MongoDB lets the first writer through and aborts the others with a WriteConflict error. This error carries the TransientTransactionError label, and on such an error WithTransactionAsync reruns the transaction from the start, body included. The result is correct because the body only writes to the database; had you sent an e-mail inside the body, the customer could have received three. In this post we cover the advanced side of the .NET driver with details of this kind: fluent Aggregate(), bulk writes, transactions, change streams and integration tests with Testcontainers. The commands are in the console application from post 9 (aggregate, bulk, transaction, watch) and assume the indexes from post 6's 06-bookstore-indexes.js.

Aggregation: the Fluent API, LINQ and BsonDocument Stages

Let's write the monthly revenue from post 7 in .NET. Aggregate() gives a typed fluent API: Match and Group are translated from C# expressions. When you need a stage the typed API does not cover, you append it as a BsonDocument with AppendStage:

// Typed stages: Match and Group are translated from C# expressions
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());

// A stage the typed API does not cover: append it as BsonDocument
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

The numbers are the same as in posts 7 and 8. Watch the translation: the enum became the string "Cancelled" (the convention from post 9), OrderedAt.Year became a $year expression, and the anonymous type's fields became PascalCase names (Year, Revenue). The BsonDocument stage that follows has to use these names; the contract between the typed part and the hand-written part is the field names, and the compiler does not check it. A practical rule: write the body of the pipeline typed, add stages with weak typed support such as $setWindowFields, $densify or $graphLookup as BsonDocument, and check the ToString() output by eye once. The same query can also be written as LINQ with AsQueryable(); as we saw in post 9, LINQ produces a pipeline in the end as well.

Bulk Writes: Ordered, Unordered and Client-Level

Importing a supplier's price list means hundreds of updates; instead of sending each in its own request, BulkWriteAsync carries them in a few commands. Let's see the ordered/unordered difference from post 3's insertMany with mixed operations here: the second model is an insert with an existing _id and fails.

// A price list import: the second model is a duplicate _id and fails
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+: one bulkWrite command across several collections (not a transaction)
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 (the default) stopped at the first error: the first price was updated, the third model was never processed and stayed in UnprocessedRequests. Unordered reported the error, carried on and updated both prices. Both throw MongoBulkWriteException<Book>; it contains both the result counts and the error list, so a "partly successful" bulk can be picked apart. If order matters (insert the book first, then update its price), use ordered; if the operations are independent, unordered, which also lets the server parallelize the work.

The collection's BulkWriteAsync writes to a single collection. IMongoClient.BulkWriteAsync, which uses the bulkWrite command that came with MongoDB 8.0, sends operations going to different collections and databases in one request; above, the order insert and the stock update went in a single command. It is not a transaction: if one operation fails, the others may still be written. It is a tool to cut network round trips, not an atomicity tool. The documentation states that this method needs MongoDB 8.0 or later; it does not work in an application connected to 7.0.

Transactions: WithTransactionAsync and Automatic Retries

This is the whole class from "Guess First". The stock decrement is the conditional $inc from post 3; the transaction only ties several books and the order together:

// Places an order: all stock decrements and the order insert succeed together or not at all
public sealed class OrderPlacement(IMongoClient client, IMongoCollection<Book> books, IMongoCollection<Order> orders)
{
    private int _attempts;

    // How many times the transaction body ran (retries included)
    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);   // aborts the transaction
                }
            }
            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

In order 381 Dune's stock went down first, then an exception was thrown because 100 copies of 1984 were requested, and the transaction was aborted: Dune stayed at 49 and the order was not written. Passing the session (s) to every operation is a must; an UpdateOneAsync without the session writes outside the transaction, immediately and permanently, and nobody notices. WithTransactionAsync has two retries: on errors labelled TransientTransactionError it retries the whole body, and when the outcome of the commit is unknown (UnknownTransactionCommitResult) it retries only the commit. Your own exception (here InsufficientStockException) carries no label, so there is no retry; the transaction is aborted and the exception comes to you. A warning from the driver documentation: if you catch an exception inside the body, rethrow it; if you swallow it, you can see behavior that goes as far as an infinite loop.

Transactions have a cost too: the default lifetime is 60 seconds, locks and snapshots are held for that time, and 9.0 added an upper limit on the number of transactions open at the same time (post 12). So look at the model first: if the payment and the status change live in one document, as in post 4, no transaction is needed; a single update is already atomic. For single-operation writes, retryable writes are enough instead of a transaction; the driver keeps them on by default and, after a network error, resends the same write once in a way the server recognizes as a retry. The same pattern applies to the review plus the summary in the book from post 5, and to an InsertManyAsync that must be written all or nothing: wrap it in WithTransactionAsync with the same session.

Change Streams: a BackgroundService and the Resume Token

In post 4 we copied the author name into books and said that the price of the copy is updating it. A change stream lets you listen to the changes in a collection in order, through the oplog; this is one reason we built the lab as a replica set from the start. A BackgroundService that updates the copies in books when an author's name changes:

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      // null on the first run: start from now
    };

    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);

            // Save the token only after the work is done: a restart repeats at most this one change
            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);
        }
    }
}

The watch command starts the service, changes Frank Herbert's name, stops the service, changes the name once more while the service is down and starts the service again:

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

The change made while the service was down was not lost: on restart it read everything after the saved resume token and updated the two books. Saving the token after the work is done is a deliberate choice; if the service crashes between the two, the same change is processed once more, so the work itself must be idempotent (writing the same name twice is harmless). UpdateLookup hides a detail: the full document comes as it is when the change stream processes it, not as it was at the moment of the change. If the service falls far behind, you see the latest name even for intermediate changes; that is what we want here, but not in every scenario. The resume token also has an expiry: you can go back only as far as the history the oplog keeps. If the service stays down longer than the oplog window, resuming fails and a full resync is needed; oplog sizing is in post 13.

The same structure carries beyond copied fields: removing a cached entry when an order changes is the MongoDB counterpart of event-driven invalidation in the Caching series post "Cache Invalidation Strategies", and forwarding the change to a message broker is the counterpart of the "Transactional Outbox in .NET" post in the Working with Message Queues series. The difference from an outbox is that the events are written to the oplog rather than a separate table; the delivery guarantee is still "at least once", and the consumer must still be idempotent. Refreshing the materialized view from post 8 fits the same pattern.

Integration Tests with Testcontainers

Testing transaction and change stream code with mocks makes little sense; the real question is what a real MongoDB does. Testcontainers starts a real MongoDB in Docker during the test run. Since transactions need a replica set, a single-node replica set with WithReplicaSet:

// One single-node replica set per test run: transactions and change streams need a replica set
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));

    // Dune was decremented first, then the transaction was aborted
    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

The fixture starts the container once for the whole test class; each test creates its own randomly named database in SeedAsync, so tests do not see each other's data. The container image is the same as in the lab, so the test checks the behavior of 9.0. Two adjustments were needed. BookstoreMongo.RegisterSerialization can now be called more than once (it does nothing after the first call), because when the fixture and the application try to make the global registrations a second time in the same process, the driver rejects a second serializer for the same type. And a global.json with test.runner set to Microsoft.Testing.Platform went into the repo root: with xunit.v3 4.x on the .NET 10 SDK, dotnet test stops without this setting with an error saying that the VSTest path is no longer supported. All the test needs is a running Docker on the machine; on Windows that is Docker Desktop.

In Production

Run the change stream service as a single copy, or split the work. If the application runs with three instances, three BackgroundService instances process the same change three times. Either move the service to a separate worker with a single instance or set up a leader election (a lock document, for example). Which instance saves the token is the same question.

Keep transactions short and measure them. The body should contain no network calls, file operations or waiting for users; these both prolong it and repeat on retry. Where the same document is written heavily, as in "Guess First", watch the number of write conflicts and retries: frequent retries signal that the model creates a hot spot on one document.

Split bulk jobs. The driver splits a large list of models into commands according to the server's message limits, but keeping hundreds of thousands of models in memory in one list and not knowing where to resume after an error is the application's problem. Chunks of a few thousand models, with progress saved after each one, make resuming possible, as in the migration from post 5.

Common Pitfalls

Restarting without saving the resume token. If the token is kept only in memory, the service starts "from now" on every deploy and the changes in between are silently lost; the "Franklin" change above would never have reached the books. Write the token somewhere durable, after the work is done.

Long transactions. Wrapping a report or a bulk update in a transaction means locking thousands of documents, getting close to the 60-second limit and pushing other operations that write the same documents into write conflicts. Transactions are for tying a few documents together; bulk jobs are done with bulk writes and idempotent steps.

Side effects in the body. An e-mail, an HTTP call or a publish to a message broker inside the transaction body repeats on retry and is not undone even if the transaction is aborted. Do side effects after the commit, or make them part of the change with an outbox.

A Small Experiment

★☆☆ Remove the session parameter (s) from the UpdateOneAsync call in OrderPlacement and run the "Insufficient stock" test. Which assertion breaks, and why?

★★☆ Write a ReviewService that does "add a review + update the summary in the book" from post 5 with WithTransactionAsync, and two integration tests for it: the happy path and the path aborted with an invalid rating (rating: 6).

★★★ Extend AuthorNameSync to category names, and write a second change stream service that refreshes the daily_sales materialized view from post 8 only for the affected days when orders change. Run the service with two instances and design a simple lock document that prevents the same change from being processed twice.

In the next post we look at the same Bookstore data through MongoDB's EF Core provider: a DbContext with UseMongoDB, owned types mapped as embedded documents, change tracking and the provider's official limitations; which job calls for the native driver and which for EF Core.