Building a Critter Stack Application: Durable Outbox Messaging and Why You Care!

Hey, did you know that JasperFx Software is ready for formal support plans for Marten and Wolverine? Not only are we trying to make the “Critter Stack” tools be viable long term options for your shop, we’re also interested in hearing your opinions about the tools and how they should change. We’re also certainly open to help you succeed with your software development projects on a consulting basis whether you’re using any part of the Critter Stack or any other .NET server side tooling.

Let’s build a small web service application using the whole “Critter Stack” and their friends, one small step at a time. For right now, the “finished” code is at CritterStackHelpDesk on GitHub.

The posts in this series are:

Event Storming
Marten as Event Store
Marten Projections
Integrating Marten into Our Application
Wolverine as Mediator
Web Service Query Endpoints with Marten
Dealing with Concurrency
Wolverine’s Aggregate Handler Workflow FTW!
Command Line Diagnostics with Oakton
Integration Testing Harness
Marten as Document Database
Asynchronous Processing with Wolverine
Durable Outbox Messaging and Why You Care! (this post)
Wolverine HTTP Endpoints
Easy Unit Testing with Pure Functions
Vertical Slice Architecture
Messaging with Rabbit MQ
The “Stateful Resource” Model
Resiliency

As we layer in new technical concepts from both Wolverine and Marten to build out incident tracking, help desk API, we looked at this message handler in the last post that both saved data, and published a message to an asynchronous, local queue that would act upon the newly saved data at some point.

public static class CategoriseIncidentHandler
{
    public static readonly Guid SystemId = Guid.NewGuid();
     
    [AggregateHandler]
    // The object? as return value will be interpreted
    // by Wolverine as appending one or zero events
    public static async Task<object?> Handle(
        CategoriseIncident command, 
        IncidentDetails existing,
        IMessageBus bus)
    {
        if (existing.Category != command.Category)
        {
            // Send the message to any and all subscribers to this message
            await bus.PublishAsync(
                new TryAssignPriority { IncidentId = existing.Id });
            return new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };
        }
 
        // Wolverine will interpret this as "do no work"
        return null;
    }
}

To recap, that message handler is potentially appending an IncidentCategorised event to an Incident event stream and publishing a command message named TryAssignPriority that will trigger a downstream action to try to assign a new priority to our Incident.

This relatively simple message handler (and we’ll make it even simpler in a later post in this series) creates a host of potential problems for our system:

In a naive usage of messaging tools, there’s a race condition between the outbound `TryAssignPriority` message being picked up by its handler and the database changes getting committed to the database. I have seen this cause nasty, hard to reproduce bugs through in real life production applications when once in awhile the message is processed before the database changes are made, and the system behaves incorrectly because the expected data is not yet committed by the original command finishing.
Maybe the actual message sending fails, but the database changes succeed, so the system is in an inconsistent state.
Maybe the outgoing message is happily published successfully, but the database changes fail, so that when the TryAssignPriority message is handled, it’s working against old system state.
Event if everything succeeds perfectly, the outgoing message should never actually be published until the transaction is complete.

To be clear, even without the usage of the outbox feature we’re about to use, Wolverine will apply an “in memory outbox” in message handlers such that all the messages published through IMessageBus.PublishAsync()/SendAsync()/etc. will be held in memory until the successful completion of the message handler. That by itself is enough to prevent the race condition between the database changes and the outgoing messages.

At this point, let’s introduce Wolverine’s transactional outbox support that was built specifically to solve or prevent the potential problems I listed up above. In this case, Wolverine has a transactional outbox & inbox support built into its integrations with PostgreSQL and Marten.

To rewind a little bit, in an earlier post where we first introduced the Marten + Wolverine integration, I had added a call to IntegrateWithWolverine() to the Marten configuration in our Program file:

using Wolverine.Marten;
 
var builder = WebApplication.CreateBuilder(args);
 
builder.Services.AddMarten(opts =>
{
    // This would be from your configuration file in typical usage
    opts.Connection(Servers.PostgresConnectionString);
    opts.DatabaseSchemaName = "wolverine_middleware";
})
    // This is the wolverine integration for the outbox/inbox,
    // transactional middleware, saga persistence we don't care about
    // yet
    .IntegrateWithWolverine()
     
    // Just letting Marten build out known database schema elements upfront
    // Helps with Wolverine integration in development
    .ApplyAllDatabaseChangesOnStartup();

Among other things, the call to IntegrateWithWolverine() up above directs Wolverine to use the PostgreSQL database for Marten as the durable storage for incoming and outgoing messages as part of Wolverine’s transactional inbox and outbox. The basic goal of this subsystem is to create consistency (really “eventual consistency“) between database transactions and outgoing messages without having to resort to endlessly painful distributed transactions.

Now, we’ve got another step to make. As of right now, Wolverine makes a determination of whether or not to use the durable outbox storage based on the destination of the outgoing message — with the theory that teams might easily want to mix and match durable messaging and less resource intensive “fire and forget” messaging within the same application. In this help desk service, we’ll make that easy and just say that all message processing in local queues (we set up TryAssignPriority to be handled through a local queue in the previous post) to be durable. In the UseWolverine() configuration, I’ll add this line of code to do that:

builder.Host.UseWolverine(opts =>
{
    // More configuration...

    // Automatic transactional middleware
    opts.Policies.AutoApplyTransactions();
    
    // Opt into the transactional inbox for local 
    // queues
    opts.Policies.UseDurableLocalQueues();
    
    // Opt into the transactional inbox/outbox on all messaging
    // endpoints
    opts.Policies.UseDurableOutboxOnAllSendingEndpoints();

    // Set up from the previous post
    opts.LocalQueueFor<TryAssignPriority>()
        // By default, local queues allow for parallel processing with a maximum
        // parallel count equal to the number of processors on the executing
        // machine, but you can override the queue to be sequential and single file
        .Sequential()

        // Or add more to the maximum parallel count!
        .MaximumParallelMessages(10);
});

I (Jeremy) may very well declare this “endpoint by endpoint” declaration of durability to have been a big mistake because confused some users and vote to change this in a later version of Wolverine.

With this outbox functionality in place, the messaging and transaction workflow behind the scenes of that handler shown above is to:

When the outgoing TryAssignPriority message is published, Wolverine will “route” that message into its internal Envelope structure that includes the message itself and all the necessary metadata and information Wolverine would need to actually send the message later
The outbox integration will append the outgoing message as a pending operation to the current Marten session
The IncidentCategorised event will be appended to the current Marten session
The Marten session is committed (IDocumentSession.SaveChangesAsync()), which will persist the new event and a copy of the outgoing Envelope into the outbox or inbox (scheduled messages or messages to local queues will be persisted in the incoming table) tables in one single, batched database command and by a native PostgreSQL transaction.
Assuming the database transaction succeeds, the outgoing messages are “released” to Wolverine’s outgoing message publishing in memory (we’re coming back to that last point in a bit)
Once Wolverine is able to successfully publish the message to the outgoing transport, it will delete the database table record for that outgoing message.

The 4th point is important I think. The close integration between Marten & Wolverine allows for more efficient processing by combining the database operations to minimize database round trips. In cases where the outgoing message transport is also batched (Azure Service Bus or AWS SQS for example), the database command to delete messages is also optimized for one call using PostgreSQL array support. I guess the main point of bringing this up is just to say there’s been quite a bit of thought and outright micro-optimizations done to this infrastructure.

But what about…?

the process is shut down cleanly? Wolverine tries to “drain” all in flight work first, and then “release” that process’s ownership of the persisted messages
the process crashes before messages floating around the local queues or outgoing message publishing finishes? Wolverine is able to detect a “dormant node” and reassign the persisted incoming and outgoing messages to be processed by another node. Or in the case of a single node, restart that work when the process is restarted.
the Wolverine tables don’t yet exist in the database? Wolverine has similar database management to Marten (it’s all the shared Weasel library doing that behind the scenes) and will happily build out missing tables in its default setting
an application using a database per tenant multi-tenancy strategy? Wolverine creates separate inbox or outbox storage in each tenant database. It’s complicated and took quite awhile to build, but it works. If no tenant is specified, the inbox/outbox in a “default” database is used
I need to use the outbox approach for consistency outside of a message handler, like when handling an HTTP request that happens to make both database changes and publish messages? That’s a really good question, and arguably one of the best reasons to use Wolverine over other .NET messaging tools because as we’ll see in later posts, that’s perfectly possible and quite easy. There is a recipe for using the Wolverine outbox functionality with MVC Core or Minimal API shown here.

Summary and What’s Next

The outbox (and closely related inbox) support is hugely important inside of any system that uses asynchronous messaging as a way of creating consistency and resiliency. Wolverine’s implementation is significantly different (and honestly more complicated) than typical implementations that depend on just polling from an outbound database table. That’s a positive in some ways because we believe that Wolverine’s approach is more efficient and will lead to greater throughput.

There is also similar inbox/outbox functionality and optimizations for Wolverine with EF Core using either PostgreSQL or Sql Server as the backing storage. In the future, I hope to see the EF Core and Sql Server support improve, but for right now, the Marten integration is getting the most attention and usage. I’d also love to see Wolverine grow to include support for alternative databases, with Azure CosmosDb and AWS Dynamo Db being leading contenders. We’ll see.

As for what’s next, let me figure out what sounds easy for the next post in January. In the meantime, Happy New Year’s everybody!

Wolverine’s HTTP Gets a Lot Better at OpenAPI (Swagger)

I just published Wolverine 1.13.0 this evening with some significant improvements (see the release notes here). Beyond the normal scattering of bug fixes (and some significant improvements to the MQTT support in Wolverine for a JasperFx Software client who we’re helping build an IoT system), the main headline is that Wolverine does a substantially better job generating OpenAPI documentation for its HTTP endpoint model.

When I’m building web services of any kind I tend to lean very hard into doing integration testing with Alba, and because of that, I also tend not to use Swashbuckle or an equivalent tool very often during development and that has apparently been a blind spot for me in building Wolverine.HTTP so far. To play out a typical conversation I frequently have with other server side .NET developers talking about tooling for web services, I think:

MVC Core by itself — but this is hugely acerbated by unfortunately popular prescriptive architectural patterns that organize code around NounController / NounService / NounRepository code organization — can easily lead to unmaintainable code in bloated controller classes and plenty of work for software consultants who get brought in later to clean up after the system wildly outgrew the original team’s “Clean Architecture” approach
I’m not convinced that Minimal API is any better for larger applications
The MVC Core controllers delegating to an inner “mediator” tool strategy may help divide the code into more maintainable code, but it adds what I think is an unacceptable level of extra code ceremony. Also acerbated by prescriptive architectures
You should use Wolverine.HTTP! It’s much lower ceremony code than the “controllers + mediator” strategy, but still sets you up for a vertical slice architecture! And it integrates well with Marten or Wolverine messaging!

Other developers: This all sounds great! Pause. Hey, the web services with this thing seem to work just fine, but man, the Swashbuckle/NSwag/Angular client generation is all kinds of not good! I’m going back to “Wolverine as MediatR”.

To which I reply:

But no more of that after today because the Wolverine HTTP OpenAPI generation just took a huge leap forward after the 1.13 release!

Here’s a sample of what I mean. From the Wolverine.HTTP test suite, here’s an endpoint method that uses Marten to load an Invoice document, modify it, then save it:

    [WolverinePost("/invoices/{invoiceId}/pay")]
    public static IMartenOp Pay([Document] Invoice invoice)
    {
        invoice.Paid = true;
        return MartenOps.Store(invoice);
    }

The [Document] attribute tells Wolverine to load the Invoice from Marten, and part of its convention will match on the invoiceId route argument from the route pattern. That failed before in a couple ways:

Swashbuckle can’t be convinced that the Invoice argument isn’t the request body
If you omit an Guid invoiceId argument from the route, Swashbuckle wasn’t seeing invoiceId as a route parameter and didn’t let you specify that in the Swashbuckle page.
Swashbuckle definitely didn’t get that IMartenOp is a specialized Wolverine side effect that shouldn’t be used as the response body.

Now though, that endpoint looks like this in Swashbuckle:

Which is now correct and actually usable! (The 404 is valid because there’s a route argument and that status is returned if the Invoice referred to by the invoiceId route argument does not exist).

To call out some improvements for Wolverine.HTTP users, at least the Swashbuckle generation handles:

Route arguments that are used by Wolverine, but not necessarily in the main method signature. So no stupid, unused [FromRoute] string id method parameters
Querystring arguments are reflected in the Swashbuckle page
[FromHeader] arguments are reflected in Swashbuckle
HTTP endpoints that return some kind of tuple correctly show the response body if there is one — and that’s a commonly used and powerful capability of Wolverine’s HTTP endpoints that previously fouled up the OpenAPI generation
The usage of [EmptyResponse] correctly sets up the 204 status code behavior with no extraneous 200 or 404 status codes coming in by default
Ignoring method injected service parameters in the main method

For a little background, after getting plenty of helpful feedback from Wolverine users, I finally took some more serious time to go investigate the problems and root causes. After digging in much deeper to the AspNetCore and Swashbuckle internals, I came to the conclusion that the OpenAPI internals in AspNetCore are ~~batshit crazy~~ far too hard coded to MVC Core and that Wolverine absolutely had to have its own provider for generating OpenAPI documents off of its own semantic model. Fortunately, AspNetCore and Swashbuckle are both open source, so I could easily get to the source code to reverse engineer what they do under the covers (plus JetBrains Rider is a rock star at disassembling code on the fly). Wolverine.HTTP 1.13 now registers its own strategy for generating the OpenAPI documentation for Wolverine endpoints and keeps the built in MVC Core-centric strategy from applying to the same Wolverine endpoints.

I’m sure there will be other issues over time, but so far, this has addressed every known issue with our OpenAPI generation. I’m hoping this goes a long way toward removing impediments to more users adopting Wolverine.HTTP because as I’ve said before, I think the Wolverine model leads to much lower ceremony code, better testability over all, and potentially to significantly better maintainability of larger systems that today turn into huge messes with MVC Core.

Building a Critter Stack Application: Asynchronous Processing with Wolverine

The posts in this series are:

As we continue to add new functionality to our incident tracking, help desk system, we have been using Marten for persistence and Wolverine for command execution within MVC Core controllers (with cameos from Alba for testing support and Oakton for command line utilities).

In the workflow we’ve built out so far for the little system shown below, we’ve created a command called CategorizeIncident that for the moment is only sent to the system through HTTP calls from a user interface.

Let’s say that in our system that we may have some domain logic rules based on customer data that we could use to try to prioritize an incident automatically once the incident is categorized. To that end, let’s create a new command named `TryAssignPriority` like this:

public class TryAssignPriority
{
    public Guid IncidentId { get; set; }
}

We’d like to kick off this work any time an incident is categorized, but we might not want to necessarily do that work within the scope of the web request that’s capturing the CategorizeIncident command. Partially this would be a potential scalability issue to potentially offload work from the web server, partially to make the user interface as responsive as possible by not making it wait for slower web service responses, but mostly because I want an excuse to introduce Wolverine’s ability to asynchronously process work through local, in memory queues.

Most of the code in this post is an intermediate form that I’m using just to introduce concepts in the simplest way I can think of. In later posts I’ll show more idiomatic Wolverine ways to do things to arrive at the final version that is in GitHub.

Alright, now that we’ve got our new command class above, let’s publish that locally through Wolverine by breaking into our earlier CategoriseIncidentHandler that I’ll show here in a “before” state:

public static class CategoriseIncidentHandler
{
    public static readonly Guid SystemId = Guid.NewGuid();
    
    [AggregateHandler]
    public static IEnumerable<object> Handle(CategoriseIncident command, IncidentDetails existing)
    {
        if (existing.Category != command.Category)
        {
            yield return new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };
        }
    }
}

In this next version, I’m going to add a single call to Wolverine’s main IMessageBus entry point to publish the new TryAssignPriority command message:

public static class CategoriseIncidentHandler
{
    public static readonly Guid SystemId = Guid.NewGuid();
    
    [AggregateHandler]
    // The object? as return value will be interpreted
    // by Wolverine as appending one or zero events
    public static async Task<object?> Handle(
        CategoriseIncident command, 
        IncidentDetails existing,
        IMessageBus bus)
    {
        if (existing.Category != command.Category)
        {
            // Send the message to any and all subscribers to this message
            await bus.PublishAsync(new TryAssignPriority { IncidentId = existing.Id });
            return new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };
        }

        // Wolverine will interpret this as "do no work"
        return null;
    }
}

I didn’t do anything that is necessarily out of order here. We haven’t built a message handler for TryAssignPriority or done anything to register subscribers, but that can come later because the PublishAsync() call up above will quietly do nothing if there are no known subscribers for the message.

For asynchronous messaging veterans out there, I will discuss Wolverine’s support for a transactional outbox for a later post. For now, just know that there’s at the very least an in-memory outbox around any message handler that will not send out any pending published messages until after the original message is successfully handled. If you’re not familiar with the “transactional outbox” pattern, please come back to read the follow up post on that later because you absolutely need to understand that to use asynchronous messaging infrastructure like Wolverine.

Next, let’s just add a skeleton message handler for our TryAssignPriority command message in the root API projection:

public static class TryAssignPriorityHandler
{
    public static void Handle(TryAssignPriority command)
    {
        Console.WriteLine("Hey, somebody wants me to prioritize incident " + command.IncidentId);
    }
}

Switching to the command line (you may need to have the PostgreSQL database running for this next thing to work #sadtrombone), I’m going to call dotnet run -- describe to preview my help desk API a little bit.

Under the section of the textual output with the header “Wolverine Message Routing”, you’ll see the message routing tree for Wolverine’s known message types:

┌─────────────────────────────────┬──────────────────────────────────────────┬──────────────────┐
│ Message Type                    │ Destination                              │ Content Type     │
├─────────────────────────────────┼──────────────────────────────────────────┼──────────────────┤
│ Helpdesk.Api.CategoriseIncident │ local://helpdesk.api.categoriseincident/ │ application/json │
│ Helpdesk.Api.TryAssignPriority  │ local://helpdesk.api.tryassignpriority/  │ application/json │
└─────────────────────────────────┴──────────────────────────────────────────┴──────────────────┘

As you can hopefully see in that table up above, just by the fact that Wolverine “knows” there is a handler in the local application for the TryAssignPriority message type, it’s going to route messages of that type to a local queue where it will be executed later in a separate thread.

Don’t worry, this conventional routing, the parallelization settings, and just about anything you can think of is configurable, but let’s mostly stay with defaults for right now.

Switching to the Wolverine configuration in the Program file, here’s a little taste of some of the ways we could control the exact parameters of the asynchronous processing for this local, in memory queue:

builder.Host.UseWolverine(opts =>
{
    // more configuration...

    // Adding a single Rabbit MQ messaging rule
    opts.PublishMessage<RingAllTheAlarms>()
        .ToRabbitExchange("notifications");

    opts.LocalQueueFor<TryAssignPriority>()
        // By default, local queues allow for parallel processing with a maximum
        // parallel count equal to the number of processors on the executing
        // machine, but you can override the queue to be sequential and single file
        .Sequential()

        // Or add more to the maximum parallel count!
        .MaximumParallelMessages(10);
    
    // Or if so desired, you can route specific messages to 
    // specific local queues when ordering is important
    opts.Policies.DisableConventionalLocalRouting();
    opts.Publish(x =>
    {
        x.Message<TryAssignPriority>();
        x.Message<CategoriseIncident>();

        x.ToLocalQueue("commands").Sequential();
    });
});

Summary and What’s Next

Through its local queues function, Wolverine has very strong support for managing asynchronous work within a local process. Any of Wolverine’s message handling capability is usable within these local queues. You also have complete control over the parallelization of the messages being handled in these local queues.

This functionality does raise a lot of questions that I will try to answer in subsequent posts in this series:

For the sake of system consistency, we absolutely have to talk about Wolverine’s transactional outbox support
How we can use Wolverine’s integration testing support to test our system even when it is spawning additional messages that may be handled asynchronously
Wolverine’s ability to automatically forward captured events in Marten to message handlers for side effects
How to utilize Wolverine’s “special sauce” to craft message handlers as pure functions that are more easily unit tested than what we have so far
Wolverine’s built in Open Telemetry support to trace the asynchronous work end to end
Wolverine’s error handling policies to make our system as resilient as possible

Thanks for reading! I’ve been pleasantly surprised how well this series has been received so far. I think this will be the last entry until after Christmas, but I think I will write at least 7-8 more just to keep introducing bits of Critter Stack capabilities in small bites. In the meantime, Merry Christmas and Happy Holidays to you all!

Building a Critter Stack Application: Marten as Document Database

The posts in this series are:

So far, we’ve been completely focused on using Marten as an Event Store. While the Marten team is very committed to the event sourcing feature set, it’s pretty likely that you’ll have other data persistence needs in your system that won’t fit the event sourcing paradigm. Not to worry though, because Marten also has a very robust “PostgreSQL as Document Database” feature set that’s perfect for low friction data persistence outside of the event storage. We’ve even used it in earlier posts as Marten projections utilize Marten’s document database features when projections are running Inline or Async (i.e., not Live).

Since we’ve already got Marten integrated into our help desk application at this point, let’s just start with a document to represent customers:

public class Customer
{
    public Guid Id { get; set; }

    // We'll use this later for some "logic" about how incidents
    // can be automatically prioritized
    public Dictionary<IncidentCategory, IncidentPriority> Priorities { get; set; }
        = new();
    
    public string? Region { get; set; }
    
    public ContractDuration Duration { get; set; } 
}

public record ContractDuration(DateOnly Start, DateOnly End);

To be honest, I’m guessing at what a Customer might involve in the end, but it’s okay that I don’t know that upfront per se because as we’ll see soon, Marten makes it very easy to evolve your persisted documents.

Having built the integration test harness for our application in the last post, let’s drop right into an integration test that persists a new Customer document object, and reloads a copy from the persisted data:

public class using_customer_document : IntegrationContext
{
    public using_customer_document(AppFixture fixture) : base(fixture)
    {
    }

    [Fact]
    public async Task persist_and_load_customer_data()
    {
        var customer = new Customer
        {
            Duration = new ContractDuration(new DateOnly(2023, 12, 1), new DateOnly(2024, 12, 1)),
            Region = "West Coast",
            Priorities = new Dictionary<IncidentCategory, IncidentPriority>
            {
                { IncidentCategory.Database, IncidentPriority.High }
            }
        };
        
        // As a convenience just because you'll use it so often in tests,
        // I made a property named "Store" on the base class for quick access to
        // the DocumentStore for the application
        // ALWAYS remember to dispose any sessions you open in tests!
        await using var session = Store.LightweightSession();
        
        // Tell Marten to save the new document
        session.Store(customer);

        // commit any pending changes
        await session.SaveChangesAsync();

        // Marten is assigning an Id for you when one doesn't already
        // exist, so that's where that value comes from
        var copy = await session.LoadAsync<Customer>(customer.Id);
        
        // Just proving to you that it's not the same object
        copy.ShouldNotBeSameAs(customer);
        
        copy.Duration.ShouldBe(customer.Duration);
    }
}

As long as the configured database for our help desk API is available, the test above will happily pass. I’d like to draw your attention to a couple things about that test above:

Notice that I didn’t have to make any changes to our application’s AddMarten() configuration in the Program file first because Marten is able to create storage for the new Customer document type on the fly when it first encounters it with its default settings
Marten is able to infer that the Id property of the new Customer type is the identity (that can be overridden), and when you add a new Customer document to the session that has an empty Guid as its Id, Marten will quickly assign and set a sequential Guid value for its identity. If you’re wondering, Marten can do this even if the property is scoped as private.
The Store() method is effectively an “upsert,” that takes advantage of PostgreSQL’s very efficient, built in upsert syntax. Marten does also support Insert and Update operations, but Store is just an easy default

Behind the scenes, Marten is just serializing our document to JSON and storing that data in a PostgreSQL JSONB column type that will allow for efficient querying within the JSON body later (if you’re immediately asking “why isn’t this thing supporting Sql Server?!?, it’s because only PostgreSQL has the JSONB type). If your document type can be round-tripped by either the venerable Newtonsoft.Json library or the newer System.Text.Json library, that document type can be persisted by Marten with zero explicit mapping.

In many cases, Marten’s approach to object persistence can lead to far less friction and boilerplate code than the equivalent functionality using EF Core, the .NET developer tool of choice. Moreover, using Marten requires a lot fewer database migrations as you change and evolve your document structure, giving developers far more ability to iterate over the shape of their persisted types as opposed to an ORM + Relational Database combination.

And of course, this is .NET, so Marten does come with LINQ support, so we can do queries like this:

        var results = await session.Query<Customer>()
            .Where(x => x.Region == "West Coast")
            .OrderByDescending(x => x.Duration.End)
            .ToListAsync();

As you’ll already know if you happen to follow me on Mastodon, we’re hopefully nearing the end of some very substantial improvements to the LINQ support for the forthcoming Marten v7 release.

While the document database feature set in Marten is pretty deep, the last thing I want to show in this post is that yes, you can create indexes within the JSON body for faster querying as needed. This time, I am going to the AddMarten() configuration in the Program file and add a little bit of code to index the Customer document on its Region field:

builder.Services.AddMarten(opts =>
{
    // other configuration...

    // This will create a btree index within the JSONB data
    opts.Schema.For<Customer>().Index(x => x.Region);
});

Summary and What’s Next

Once upon a time, Marten started with a pressing need to have a reliable, ACID-compliant document database feature set, and we originally chose PostgreSQL because of its unique JSON feature set. Almost on a lark, I added a nascent event sourcing capability before the original Marten 1.0 release. To my surprise, the event sourcing feature set is the main driver of Marten adoption by far, but Marten still has its original feature set to make the rock solid PostgreSQL database engine function as a document database for .NET developers.

Even in a system using event sourcing, there’s almost always some kind of relatively static reference data that’s better suited for Marten’s document database feature set or even going back to using PostgreSQL as the outstanding relational database engine that it is.

In the next post, now that we also know how to store and retrieve customer documents with Marten, we’re going to introduce Wolverine’s “compound handler” capability and see how that can help us factor our code into being very testable.

Building a Critter Stack Application: Wolverine’s Aggregate Handler Workflow FTW!

TL;DR: The full critter stack combo can make CQRS command handler code much simpler and easier to test than any other framework on the planet. Fight me.

The posts in this series are:

Event Storming
Marten as Event Store
Marten Projections
Integrating Marten into Our Application
Wolverine as Mediator
Web Service Query Endpoints with Marten
Dealing with Concurrency
Wolverine’s Aggregate Handler Workflow FTW! (this post)
Command Line Diagnostics with Oakton
Integration Testing Harness
Marten as Document Database
Asynchronous Processing with Wolverine
Durable Outbox Messaging and Why You Care!
Wolverine HTTP Endpoints
Easy Unit Testing with Pure Functions
Vertical Slice Architecture
Messaging with Rabbit MQ
The “Stateful Resource” Model
Resiliency

This series has been written partially in response to some constructive criticism that my writings on the “Critter Stack” suffered from introducing too many libraries or concepts all at once. As a reaction to that, this series is trying to only introduce one new capability or library at a time — which brought on some constructive criticism from someone else that the series isn’t making it obvious why anyone should care about the “Critter Stack” in the first place. So especially for Rob Conery, I give you:

Last time out we talked using Marten’s facilities for optimistic concurrency or exclusive locking to protect our system from inconsistencies due to concurrent commands being processed against the same incident event stream. In the process of that post, I showed the code for a command handler for the CategoriseIncident command shown below that I purposely wrote in a long hand form as explicitly as possible to avoid introducing too many new concepts at once:

public static class LongHandCategoriseIncidentHandler
{
    public static async Task Handle(
        CategoriseIncident command, 
        IDocumentSession session, 
        CancellationToken cancellationToken)
    {
        var stream = await session
            .Events
            .FetchForWriting<IncidentDetails>(command.Id, cancellationToken);

        // Don't worry, we're going to clean this up later
        if (stream.Aggregate == null)
        {
            throw new ArgumentOutOfRangeException(nameof(command), "Unknown incident id " + command.Id);
        }
        
        // We need to validate whether this command actually 
        // should do anything
        if (stream.Aggregate.Category != command.Category)
        {
            var categorised = new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };

            stream.AppendOne(categorised);
            
            await session.SaveChangesAsync(cancellationToken);
        }
    }

Hopefully that code is relatively easy to follow, but it’s still pretty busy and there’s a mixture of business logic and fiddling with infrastructure code that’s not particularly helpful when the code inevitably gets more complicated than that as the requirements grow. As we’ll learn about later in this series, both Marten and Wolverine have some built in tooling to enable effective automated integration testing and do so much more effectively than just about any other tool out there. All the same though, you just don’t want to be testing the business logic by trudging through integration tests if you don’t have to (see my only rule of testing).

So let’s definitely look at how Wolverine plays nicely with Marten using its aggregate handler workflow recipe to simplify our handler for easier unit testing and just flat out cleaner code.

First off, I’m going to add the WolverineFx.Marten Nuget to our application:

dotnet add package WolverineFx.Marten

Next, break into our application’s Program file and add one call to the Marten configuration to incorporate some Wolverine goodness into Marten in our application:

builder.Services.AddMarten(opts =>
{
    // Existing Marten configuration...
})
    // This is a mild optimization
    .UseLightweightSessions()

    // Use this directive to add Wolverine transactional middleware for Marten
    // and the Wolverine transactional outbox support as well
    .IntegrateWithWolverine();

And now, let’s rewrite our CategoriseIncident command handler with a completely equivalent implementation using the “aggregate handler workflow” recipe:

public static class CategoriseIncidentHandler
{
    // Kinda faked, don't pay any attention to this please!
    public static readonly Guid SystemId = Guid.Parse("4773f679-dcf2-4f99-bc2d-ce196815dd29");

    // This Wolverine handler appends an IncidentCategorised event to an event stream
    // for the related IncidentDetails aggregate referred to by the CategoriseIncident.IncidentId
    // value from the command
    [AggregateHandler]
    public static IEnumerable<object> Handle(CategoriseIncident command, IncidentDetails existing)
    {
        if (existing.Category != command.Category)
        {
            // This event will be appended to the incident
            // stream after this method is called
            yield return new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };
        }
    }
}

In the handler method above, the presence of the[AggregateHandler]attribute directs Wolverine to wrap some middleware around the execution of our Handle() method that:

“Knows” the aggregate type in question is the second argument to the handler method, so in this case, IncidentDetails
Scans the CategoriseIncident type looking for a property that identifies the IncidentDetails (which will make it utilize the Id property in this case, but the docs spell this convention in detail)
Does all the work to delegate and coordinate work in the logical command flow between the Marten infrastructure and our little bitty Handle() method

To visualize this, Wolverine is generating its own internal message handler for CategoriseIncident that has this simplified workflow:

And as a preview to a topic I’ll dive into in much more detail in a later post, here’s part of the (admittedly ugly in the way that only auto-generated code can be) C# code that Wolverine generates around our handler method:

public override async System.Threading.Tasks.Task HandleAsync(Wolverine.Runtime.MessageContext context, System.Threading.CancellationToken cancellation)
{
    // The actual message body
    var categoriseIncident = (Helpdesk.Api.CategoriseIncident)context.Envelope.Message;

    await using var documentSession = _outboxedSessionFactory.OpenSession(context);
    var eventStore = documentSession.Events;
    
    // Loading Marten aggregate
    var eventStream = await eventStore.FetchForWriting<Helpdesk.Api.IncidentDetails>(categoriseIncident.Id, categoriseIncident.Version, cancellation).ConfigureAwait(false);

    
    // The actual message execution
    var outgoing1 = Helpdesk.Api.CategoriseIncidentHandler.Handle(categoriseIncident, eventStream.Aggregate);

    if (outgoing1 != null)
    {
        
        // Capturing any possible events returned from the command handlers
        eventStream.AppendMany(outgoing1);

    }

    await documentSession.SaveChangesAsync(cancellation).ConfigureAwait(false);
}

And lastly, we’ve now reduced our CategoriseIncident command handler to the point where the code that we are actually having to write is a pure function, meaning that it’s a simple matter of inputs and outputs with no dependency on any kind of stateful infrastructure. You absolutely care about isolating any kind of business logic into pure functions because that code becomes much easier to unit test.

And to prove that last statement, here’s what the unit tests for our Handle(CategoriseIncident, IncidentDetails) could look like using xUnit.Net and Shouldly:

public class CategoriseIncidentTests
{
    [Fact]
    public void raise_categorized_event_if_changed()
    {
        // Arrange
        var command = new CategoriseIncident
        {
            Category = IncidentCategory.Database
        };

        var details = new IncidentDetails(
            Guid.NewGuid(), 
            Guid.NewGuid(), 
            IncidentStatus.Closed, 
            new IncidentNote[0],
            IncidentCategory.Hardware);

        // Act
        var events = CategoriseIncidentEndpoint.Post(command, details);

        // Assert
        var categorised = events.Single().ShouldBeOfType<IncidentCategorised>();
        categorised
            .Category.ShouldBe(IncidentCategory.Database);
    }

    [Fact]
    public void do_not_raise_event_if_the_category_would_not_change()
    {
        // Arrange
        var command = new CategoriseIncident
        {
            Category = IncidentCategory.Database
        };

        var details = new IncidentDetails(Guid.NewGuid(), Guid.NewGuid(), IncidentStatus.Closed, new IncidentNote[0],
            IncidentCategory.Database);

        // Act
        var events = CategoriseIncidentEndpoint.Post(command, details);
        
        // Assert no events were appended
        events.ShouldBeEmpty();
    }
}

In the unit test code above, we were able to exercise the decision about what events (if any) should be appended to the incident event stream without any dependency whatsoever on any kind of infrastructure. The easiest kind of unit test to write and to read later is a test that has a clear relationship between the test inputs and outputs with minimal noise code for setting up state — and that’s exactly what we have up above. No message mock object set up, no need to setup database state, nothing. Just, “here’s the existing state and this command, now tell me what events should be appended.”

Summary and What’s Next

The full Critter Stack “aggregate handler workflow” recipe leads to very low ceremony code to implement command handlers within a CQRS style architecture. This recipe also leads to a code structure where your business logic is relatively easy to test with fast running unit testing. And we arrived at that point without having to watch umpteen hours of “Clean Architecture” YouTube snake oil videos, introducing a ton of “Ports and Adapter” style abstractions to clutter up our code, or scattering our code for the single CategoriseIncident message handler across 3-4 “Onion Architecture” projects within a massive .NET solution.

This approach was heavily inspired by the Decider pattern that originated for Event Sourcing within the F# community. But whereas the F# approach uses language tricks (and I don’t mean that pejoratively here), Wolverine is getting to a lower ceremony approach by doing that runtime code generation around our code.

If you look back to the sequence diagram up above that tries to explain the control flow, Wolverine is purposely using Jim Shore’s idea of the “A-Frame Architecture” (it’s not really an architectural style despite the name, so don’t even try to do an apples to apples comparison between it and something more prescriptive like the Clean Architecture). In this approach, Wolverine is purposely decoupling the Marten infrastructure away from the CategoriseIncident handler logic that is implementing the business logic that “decides” what to do next by mediating between Marten and the handler. The “A-Frame” name comes from visualizing that mediation like this (Wolverine calls into the infrastructure services like Marten and the business logic so the domain logic doesn’t have to):

Now, there’s a lot more stuff that our command handlers may very well need to implement, including:

Message input validation
Instrumentation and observability
Error handling and resiliency protections ’cause it’s an imperfect world!
Publishing the new events to some other internal message handler that will take additional actions after our first command has “decided” what to do next
Publishing the new events as some kind of external message to another process
Enrolling in a transactional outbox of some sort or another to keep the system in a consistent state — and you really need to care about this capability!!!

And oh, yeah, do all that with minimal code ceremony, be testable with unit tests as much as possible, and be feasible to do automated integration testing when we have to.

We’ll get to all the items in that list above in this series, but I think in the next post I’d like to introduce Wolverine’s HTTP handler recipe and build out more aggregate command handlers, but this time with an HTTP endpoint. Until next time…

Building a Critter Stack Application: Wolverine as Mediator

The posts in this series are:

In the previous posts I’ve been focused on Marten as a persistence tool. Today I want to introduce Wolverine into the mix, but strictly as a “Mediator” tool within the commonly used MVC Core or Minimal API tools for web service development.

While Wolverine does much, much more than what we’re going to use today, let’s stay with the them of keeping these posts short and just dip our toes into the Wolverine water with a simple usage.

Using our web service project from previous posts, I’m going to add a reference to the main Wolverine nuget through:

dotnet add package WolverineFx

Next, let’s add Wolverine to our application with this one line of code within our Program file:

builder.Host.UseWolverine(opts =>
{
    // We'll add more here later, but the defaults are all
    // good enough for now
});

As a quick aside, Wolverine is added directly to the IHostBuilder instead of IServiceCollection through a “Use****()” method because it’s also quietly sliding in Lamar as the underlying IoC container. Some folks have been upset at that, so let’s be upfront with that right now. While I may talk about Lamar diagnostics as part of this series, it’s unlikely that that will ever be an issue for most users in any way. Lamar has some specific functionality that was built specifically for Wolverine and utilized quite heavily.

This time out, let’s move into the “C(ommand)” part of our CQRS architecture and build some handling for the CategoriseIncident command we’d initially discovered in our Event Storming session:

public class CategoriseIncident
{
    public Guid Id { get; set; }
    public IncidentCategory Category { get; set; }
    public int Version { get; set; }
}

And next, let’s build our very first ever Wolverine message handler for this command that will load the existing IncidentDetails for the designated incident, decide if the category is being changed, and add a new event to the event stream using Marten’s IDocumentSession service. That handler in code done purposely in an explicit, “long hand” style could be this — but in later posts we will use other Wolverine capabilities to make this code much simpler while even introducing a lot more robust set of validations:

public static class CategoriseIncidentHandler
{
    public static async Task Handle(
        CategoriseIncident command, 
        IDocumentSession session, 
        CancellationToken cancellationToken)
    {
        // Find the existing state of the referenced Incident
        var existing = await session
            .Events
            .AggregateStreamAsync<IncidentDetails>(command.Id, token: cancellationToken);

        // Don't worry, we're going to clean this up later
        if (existing == null)
        {
            throw new ArgumentOutOfRangeException(nameof(command), "Unknown incident id " + command.Id);
        }
        
        // We need to validate whether this command actually 
        // should do anything
        if (existing.Category != command.Category)
        {
            var categorised = new IncidentCategorised
            {
                Category = command.Category,
                UserId = SystemId
            };

            session.Events.Append(command.Id, categorised);
            await session.SaveChangesAsync(cancellationToken);
        }
    }
    
    // This is kinda faked out, nothing to see here!
    public static readonly Guid SystemId = Guid.NewGuid();
}

There’s a couple things I want you to note about the handler class above:

We’re not going to make any kind of explicit configuration to help Wolverine discover and use that handler class. Instead, Wolverine is going to discover that within our main service assembly because it’s a public, concrete class suffixed with the name “Handler” (there are other alternatives for this discovery if you don’t like that approach).
Wolverine “knows” that the Handle() method is a handler for the CategoriseIncident command because the method is named “Handle” and the first argument is that command type
Note that this handler is a static type. It doesn’t have to be, but doing so helps Wolverine shave off some object allocations at runtime.
Also note that Wolverine message handlers happily support “method injection” and allow you to inject IoC service dependencies like the Marten IDocumentSession through method arguments. You can also do the more traditional .NET approach of pulling everything through a constructor and setting instance fields, but hey, why not write simpler code?
While it’s perfectly legal to handle multiple message types in the same handler class, I typically recommend making that a one to one relationship in most cases

And next, let’s put this into context by having an MVC Core controller expose an HTTP route for this command type, then pass the command on to Wolverine where it will mediate between the HTTP outer world and the inner world of the application services like Marten:

// I'm doing it this way for now because this is 
// a common usage, but we'll move away from 
// this later into more of a "vertical slice"
// approach of organizing code
public class IncidentController : ControllerBase
{
    [HttpPost("/api/incidents/categorize")]
    public Task Categorize(
        [FromBody] CategoriseIncident command,
        [FromServices] IMessageBus bus)

        // IMessageBus is the main entry point into
        // using Wolverine
        => bus.InvokeAsync(command);
}

Summary and Next Time

In this post we looked at the very simplest usage of Wolverine, how to integrate that into your codebase, and how to get started writing command handlers with Wolverine. What I’d like you to take away is that Wolverine is a very different animal from “IHandler of T” frameworks like MediatR, NServiceBus, MassTransit, or Brighter that require mandatory interface signatures and/or base classes. Even when writing long hand code as I did, I hope you can notice already how much lower code ceremony Wolverine requires compared to more typical .NET frameworks that solve similar problems to Wolverine.

I very purposely wrote the message handlers in a very explicit way, and left out some significant use cases like concurrency protection, user input validation, and cross cutting concerns. I’m not 100% sure where I want to go next, but in this next week we’ll look at concurrency protections with Marten, highly efficient GET HTTP endpoints with Marten and ASP.Net Core, and start getting into Wolverine’s HTTP endpoint model.

Why you might ask are all the Wolverine nugets suffixed with “Fx?” The Marten core team and some of our closest collaborators really liked the name “Wolverine” for this project and instantly came up with the project graphics, but when we tried to start publishing Nuget packages, we found out that someone is squatting on the name “Wolverine” in Nuget and we weren’t able to get the rights to that name. Rather than change course, we stubbornly went full speed ahead with the “WolverineFx” naming scheme just for the published Nugets.

Let’s Get Controversial for (only) a Minute!

When my wife and I watched the Silicon Valley show, I think she was bemused when I told her there was a pretty heated debate in development circles over “tabs vs spaces.”

I don’t want this to detract too much from the actual content of this series, but I have very mixed feelings about ASP.Net MVC Core as a framework and the whole idea of using a “mediator” as popularized by the MediatR library within an MVC Core application.

I’ve gone back and forth on both ASP.Net MVC in its various incarnations and also on MediatR both alone and as a complement to MVC Core. Where I’ve landed at right now is the opinion that MVC Core used by itself is a very flawed framework that can easily lead to unmaintainable code over time as an enterprise system grows over time as typical interpretations of the “Clean Architecture” style in concert with MVC Core’s routing rules lead unwary developers to creating bloated MVC controller classes.

While I was admittedly unimpressed with MediatR as I first encountered it on its own merits in isolation, what I will happily admit is that the usage of MediatR is helpful within MVC Core controllers as a way to offload operation specific code into more manageable pieces as opposed to the bloated controllers that frequently result from using MVC Core. I have since occasionally recommended the usage of MediatR within MVC Core codebases to my consulting clients as a way to help make their code easier to maintain over time.

If you’re interested, I touched on this theme somewhat in my talk A Contrarian View of Software Architecture from NDC Oslo 2023. And yes, I absolutely think you can build maintainable systems with MVC Core over time even without the MediatR crutch, but I think you have to veer away from the typical usage of MVC Core to do so and be very mindful of how you’re using the framework. In other words, MVC Core does not by itself lead teams to a “pit of success” for maintainable code in the long run. I think that MediatR or Wolverine with MVC Core can help, but I think we can do better in the long run by moving away from MVC Core. a

By the time this series is over, I will be leaning very hard into organizing code in a vertical slice architecture style and seeing how to use the Critter Stack to create maintainability and testability without the typically complex “Ports and Adapter” style architecture that well meaning server side development teams have been trying to use in the past decade or two.

While I introduced Wolverine today as a “mediator” tool within MVC Core, by the time this series is done we’ll move away from MVC Core with or without MediatR or “Wolverine as MediatR” and use Wolverine’s HTTP endpoint model by itself as simpler alternative with less code ceremony — and I’m going to try hard to make the case that that simpler model is a superior way to build systems.

Building a Critter Stack Application: Integrating Marten into Our Application

The posts in this series are:

In the previous couple posts I’ve introduced Marten as a standalone library and some of its capabilities for persisting events and creating projected views off those events within an event sourcing persistence strategy. Today I want to end the week by simply talking about how to integrate Marten into an ASP.Net Core application.

Oskar’s Introduction to Event Sourcing – Self Paced Kit has a world of information for folks getting started with event sourcing.

Let’s start a shell of a new web service project and add a Nuget reference to Marten through:

dotnet new webapi
dotnet add package Marten

If you’ll open up the Program.cs file in your new application, find this line of code at the top where it’s just starting to configure your application:

using Marten;
// Many other using statements

var builder = WebApplication.CreateBuilder(args);

Right underneath that (it doesn’t actually matter most times what order this all happens inside the Program code, but I’m giving Marten the seat at the head of the table so to speak), add this code:

// "AddTool()" is now the common .NET idiom
// for integrating tools into .NET applications
builder.Services.AddMarten(opts =>
{
    // You always have to tell Marten what the connection string to the underlying
    // PostgreSQL database is, but this is the only mandatory piece of 
    // configuration
    var connectionString = builder.Configuration.GetConnectionString("marten");
    opts.Connection(connectionString);
    
    // We have to tell Marten about the projection we built in the previous post
    // so that Marten will "know" how to project events to the IncidentDetails
    // projected view
    opts.Projections.Add<IncidentDetailsProjection>(ProjectionLifecycle.Inline);
})
    // This is a mild optimization
    .UseLightweightSessions();;

That little bit of code is adding the necessary Marten services to your application’s underlying IoC container with the correct scoping. The main services you’ll care about are:

Service	Description	Lifetime
`IDocumentStore`	Root configuration of the Marten database	`Singleton`
`IQuerySession`	Read-only subset of the `IDocumentSession`	`Scoped`
`IDocumentSession`	Marten’s unit of work service that also exposes capabilities for querying and the event store	`Scoped`

Marten services in the IoC container

You can read more about the bootstrapping options in Marten in the documentation. If you’re wondering what “Lightweight Session” means to Marten, you can learn more about the different flavors of sessions in the documentation, but treat that as an advanced subject that’s not terribly relevant to this post.

And also, the usage of AddMarten() should feel familiar to .NET developers now as it follows the common idioms for integrating external tools into .NET applications through the generic host infrastructure that came with .NET Core. As a long term .NET developer, I cannot exaggerate how valuable I think this standardization of application bootstrapping has been for the OSS community in .NET.

Using Marten in an MVC Controller

For right now, I want to assume that many of you are already familiar with ASP.Net MVC Core, so let’s start by showing the usage of Marten within a simple controller to build the first couple endpoints to log a new incident and fetch the current state of an incident in our new incident tracking, help desk service:

public class IncidentController : ControllerBase
{
    private readonly IDocumentSession _session;

    public IncidentController(IDocumentSession session)
    {
        _session = session;
    }

    [HttpPost("/api/incidents")]
    public async Task<IResult> Log(
        [FromBody] LogIncident command
        )
    {
        var userId = currentUserId();
        var logged = new IncidentLogged(command.CustomerId, command.Contact, command.Description, userId);

        var incidentId = _session.Events.StartStream(logged).Id;
        await _session.SaveChangesAsync(HttpContext.RequestAborted);

        return Results.Created("/incidents/" + incidentId, incidentId);
    }

    [HttpGet("/api/incidents/{incidentId}")]
    public async Task<IResult> Get(Guid incidentId)
    {
        // In this case, the IncidentDetails are projected
        // "inline", meaning we can load the pre-built projected
        // view
        var details = await _session.LoadAsync<IncidentDetails>(incidentId);

        return details != null
            ? Results.Json(details)
            : Results.NotFound();
    }

    private Guid currentUserId()
    {
        // let's say that we do something here that "finds" the
        // user id as a Guid from the ClaimsPrincipal
        var userIdClaim = User.FindFirst("user-id");
        if (userIdClaim != null && Guid.TryParse(userIdClaim.Value, out var id))
        {
            return id;
        }

        throw new UnauthorizedAccessException("No user");
    }
}

It’s important to note at this point (that might change in Marten 7) that the IDocumentSession should be disposed when you’re doing using it to tell Marten to close down any open database connections. In the usage above, the scoped IoC container mechanics in ASP.Net Core are handling all the necessary object disposal for you.

Summary and What’s Next

Today we strictly just looked at how to integrate Marten services into a .NET application using the AddMarten() mechanism. Using a simple MVC Core Controller, we saw how Marten services are available and managed in the application’s IoC container and how to perform basic event sourcing actions in the context of little web service endpoints.

In later posts in this series we’ll actually replace this IncidentController with Wolverine endpoints and some “special sauce” with Marten to be much more efficient.

In the next post, I think I want to talk through the CQRS architectural style with Marten. Bear with me, but I’ll still be using explicit code with MVC Core controllers that folks are probably already familiar with to talk over the requirements and Marten capabilities in isolation. Don’t worry though, I will eventually introduce Wolverine into the mix to show how the Wolverine + Marten integration can make your code very clean and testable.

Critter Stack at .NET Conf 2023

JasperFx Software will be shortly announcing the availability of official support plans for Marten, Wolverine, and other JasperFx open source tools. We’re working hard to build a sustainable ecosystem around these tools so that companies can feel confident in making a technical bet on these high productivity tools for .NET server side development.

I’ll be presenting a short talk at .NET Conf 2023 entitled “CQRS with Event Sourcing using the Critter Stack.” It’s going to be a quick dive into how to use Marten and Wolverine to build a very small system utilizing a CQRS Architecture with Event Sourcing as the persistence strategy.

Hopefully, I’ll be showing off:

How Wolverine’s runtime architecture is significantly different than other .NET tools and why its approach leads to much lower code ceremony and potentially higher performance
Marten and PostgreSQL providing a great local developer story both in development and in integration testing
How the Wolverine + Marten integration makes your domain logic easily unit testable without resorting to complicated Clean/Onion/Hexagonal Architectures
Wolverine’s built in integration testing support that you’ll wish you had today in other .NET messaging tools
The built in tooling for unraveling Wolverine or Marten’s “conventional magic”

Here’s the talk abstract:

CQRS with Event Sourcing using the “Critter Stack”

Do you have a system where you think would be a good fit for a CQRS architecture that also uses Event Sourcing for at least part of its persistence strategy? Are you intimidated by the potential complexity of that kind of approach? Fear not, using a combination of the PostgreSQL-backed Marten library for event sourcing and its newer friend Wolverine for command handling and asynchronous messaging, I’ll show you how you can quickly get started with both CQRS and Event Sourcing. Once we get past the quick start, I’ll show you how the Critter Stack’s unique approach to the “Decider” pattern will help you create robust command handlers with very little code ceremony while still enjoying easy testability. Moving beyond basic command handling, I’ll show you how to reliably subscribe to and publish the events or other messages created by your command handlers through Wolverine’s durable outbox and direct subscriptions to Marten’s event storage.

Wolverine now does Event Streaming with Kafka or MQTT

As part of an ongoing JasperFx client engagement, Wolverine (1.9.0) just added some new options for event streaming from Wolverine applications. The immediate need was to support messaging with the MQTT protocol for usage inside of a new system in the “Internet of Things” problem space. Knowing that a different JasperFx client is going to need to support event subscriptions with Apache Kafka, it was also convenient to finally add the much requested option for Kafka support within Wolverine while the similar MQTT work was still fresh in my mind.

While the new MQTT transport option is documented, the Kafka transport documentation is still on the way, so I’m going to focus on that first.

To get started with Kafka within a Wolverine application, add the WolverineFx.Kafka Nuget to your project. Next, add the Kafka transport option, any messaging subscription rules, and the topics you want your application to listen to with code like this:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseKafka("localhost:29092");

        // Just publish all messages to Kafka topics
        // based on the message type (or message attributes)
        // This will get fancier in the near future
        opts.PublishAllMessages().ToKafkaTopics();
        
        // Or explicitly make subscription rules
        opts.PublishMessage<ColorMessage>()
            .ToKafkaTopic("colors");
        
        // Listen to topics
        opts.ListenToKafkaTopic("red")
            .ProcessInline();

        opts.ListenToKafkaTopic("green")
            .BufferedInMemory();
        

        // This will direct Wolverine to try to ensure that all
        // referenced Kafka topics exist at application start up 
        // time
        opts.Services.AddResourceSetupOnStartup();
    }).StartAsync();

I’m very sure that these two transports (and shortly a third option for Apache Pulsar) will need to be enhanced when they meet real users and unexpected use cases, but I think there’s a solid foundation ready to go.

In the near future, JasperFx Software will be ready to start offering official support contracts and relationships for both Marten and Wolverine. In the slightly longer term, we’re hoping to create some paid add on products (with support!) for Wolverine for “big, serious enterprise usage.” One of the first use cases I’d like us to tackle with that initiative will be a more robust event subscription capability from Marten’s event sourcing through Wolverine’s messaging capabilities. Adding options especially for Kafka messaging and also for MQTT, Pulsar, and maybe SignalR is an obvious foundational piece to make that a reality.

Important Patterns Lurking in Your Persistence Tooling

I was foolish enough to glance at my speaker feedback from my talk at KCDC this summer where I gave an updated version of my Concurrency and Parallelism talk. Out of a combination of time constraints and the desire to shamelessly promote the Critter Stack tools, I mostly used sample problems and solutions from my own work with Marten and Wolverine. One red card evaluation complained that the talk was useless to him (and I have no doubt it was a “him”) because it didn’t focus on mainstream .NET tools. I’m going to do the same thing here, mostly out of time constraints, but I would hope you would take away some understanding of the conceptual patterns I’m discussing here rather than being hung up on the exact choice of tooling.

Continuing my new series about the usefulness of old design patterns that I started with The Lowly Strategy Pattern is Still Useful. Today I want to talk about a handful of design patterns commonly implemented inside of mainstream persistence tooling that you probably already use today. In all cases, the original terminology I’m using here comes from Martin Fowler’s seminal Patterns of Enterprise Application Architecture book from the early 00’s. I’ll be using Marten (of course) for all the examples, but these patterns all exist within Entity Framework Core as well. Again, anytime I write about design pattern usage, I urge you to pay more attention to the concepts, roles, and responsibilities within code without getting too hung up on implementation details.

I seriously doubt that most of you will ever purposely sit down and write your own implementations of these patterns, but it’s always helpful to understand how the tools and technical layers directly underneath your code actually work.

The first two patterns are important for performance and sometimes even just scoping within complex system operations. In a subsequent post I think I’d like to tackle patterns for data consistency and managing concurrency.

Quick Note on Marten Mechanics

If you’re not already familiar with it, Marten is a library in .NET that turns PostgreSQL into a full featured document database and event store. When you integrate Marten into a typical .NET system, you will probably use this idiomatic approach to add Marten:

// This is the absolute, simplest way to integrate Marten into your
// .NET application with Marten's default configuration
builder.Services.AddMarten(options =>
{
    // Establish the connection string to your Marten database
   options.Connection(builder.Configuration.GetConnectionString("Marten")!);
});

Using the AddMarten() method adds a service called IDocumentSession from Marten into your application’s IoC container with a Scoped lifetime, meaning that a single IDocumentSession should be created, shared, and used within a single HTTP request or within the processing of a single message within a messaging framework. That lifetime is important to understand the Marten (and similar EF Core DbContext usage) usage of the Identity Map and Unit of Work patterns explained below.

If you are more familiar with EF Core, just translate Marten’s IDocumentSession to EF Core’s DbContext in this post if that helps.

Identity Map

Ensures that each object gets loaded only once by keeping every loaded object in a map. Looks up objects using the map when referring to them.
Martin Fowler

In enterprise software systems I’ve frequently hit code that tries to implement a single, logical transaction across multiple internal functions or services in a large call stack. This situation usually arises over time out of sheer complexity of the business rules and the build up of “special” condition handling over time for whatever cockamamie logic that additional customers require. Unfortunately, this arrangement can frequently lead to duplicated database queries for the same reference data that is needed by completely different areas of code within the single, logical transaction — which can easily lead to very poor performance in your system.

In my experience, chattiness (making many network round trips) to the database has been maybe the single most common source of poor system performance. Followed closely by chattiness between user interface clients and the backend services. The identity map mechanics shown here can be an easy way to mitigate at least the first problem.

This is where the “Unit of Work” pattern can help. Let’s say that in your code you frequently need to load information about the User entities within your system. Here’s a little demonstration of what the identity map actually does for you in terms of scoped caching:

using var store = DocumentStore.For("some connection string");

// Chiefs great Tamba Hali!
var user = new User { FirstName = "Tamba", LastName = "Hali" };

// Marten assigns the identity for the User as it 
// persists the new document
await store.BulkInsertAsync(new[] { user });

// Open a Marten session with the identity map
// functionality
await using var session = store.IdentitySession();

// First request for this document, so this call would
// hit the database
var user2 = await session.LoadAsync<User>(user.Id);

// This time it would be loaded from the identity map
// in memory
var user3 = await session.LoadAsync<User>(user.Id);

// Just to prove that
user2.ShouldBeSameAs(user3);

// And also...
var user4 = await session.Query<User>()
    .FirstAsync(x => x.FirstName == "Tamba");

// Yup, Marten has to actually query the database, but still
// finds the referenced document from the identity map when
// it resolves the results from the raw PostgreSQL data
user4.ShouldBeSameAs(user2);

With the “Identity Map” functionality, the Marten session is happily able to avoid making repeated requests to the database for the same information across multiple attempts to access that same data.

In bigger call stacks where there’s a real need to potentially access the same data at different times, the Identity Map is a great advantage. However, in smaller usages the Identity Map is nothing but extra overhead as your persistence tooling has to track the data it loads in some kind of in memory key/value storage. Especially in cases where you’re needing to load quite a bit of data at one time, the identity map can be a significant drag both in terms of memory usage and in performance.

Not to worry though, at least for Marten we can purposely create “lightweight” sessions that leave out the identity map tracking altogether like this:

using var store = DocumentStore.For("some connection string");

// Create a lightweight session without any
// identity map overhead
using var session = store.LightweightSession();

or globally within our application like so:

builder.Services.AddMarten(options =>
{
    // Establish the connection string to your Marten database
    options.Connection(builder.Configuration.GetConnectionString("Marten")!);
})
    // Tell Marten to use lightweight sessions
    // for the default IoC registration of
    // IDocumentSession
    .UseLightweightSessions();

You’re unlikely to ever purposely build your own implementation of the Identity Map pattern, but it’s in many common persistence tools and it’s still quite valuable to understand that behavior and also when you would rather bypass that usage to be more efficient.

Unit of Work

Maintains a list of objects affected by a business transaction and coordinates the writing out of changes and the resolution of concurrency problems.
Martin Fowler

Data consistency is a pretty big deal in most enterprise systems, and that makes us developers have to care about our transactional boundaries to ensure that related changes succeed or fail in one operation. This is where the Unit of Work pattern implemented by tools like Marten comes into play.

For Marten, the IDocumentSession is the unit of work as shown below:

public static async Task DoWork(IDocumentSession session)
{
    DoWorkPart1(session);
    DoWorkPart2(session);
    DoWorkPart3(session);

    // Make all the queued up persistence
    // operations in one database transactions
    await session.SaveChangesAsync();
}

public static void DoWorkPart1(IDocumentSession session)
{
    session.Store(new User{FirstName = "Travis", LastName = "Kelce"});
    session.DeleteWhere<User>(x => x.Department == "Wide Receiver");
}

public static void DoWorkPart2(IDocumentSession session)
{
    session.Store(new User{FirstName = "Patrick", LastName = "Mahomes"});
    session.Events.StartStream<Game>(new KickedOff());
}

public static void DoWorkPart3(IDocumentSession session)
{
    session.Store(new User{FirstName = "Chris", LastName = "Jones"});
}

When IDocumentSession.SaveChangesAsync() is called, it executes a database command for all the new documents stored and the deletion operation queued up across the different helper methods all at one time. Marten is letting us worry about business logic and expressing what database changes should be made while Marten handles the actual transaction boundaries for us when it writes to the databse.

A couple more things to note about the code above:

If you don’t need to read any data first, Marten doesn’t even open a database connection until you call the SaveChangesAsync() method. That’s important to know because database connections are expensive within your system, and you want them to be as short lived as possible. In a manual implementation without a unit of work tracker of some sort, you often open a connection and start a transaction that you then pass around within your code — which leads to holding onto connections longer and risking potential destabilization of your system through connection exhaustion. And don’t blow that off, because that happens quite frequently when we developers are less than perfect with our database connection hygiene.
As I said earlier, Marten registers the IDocumentSession in your IoC container as Scoped, meaning that the same session would be shared by all objects created by the same scoped container within an AspNetCore request or inside message handling frameworks like Wolverine, NServiceBus, or MassTransit. That scoping is important to make the transactional boundaries through the session’s unit of work tracking actually work across different functions within the code.
I’m not sure about other tools, but Marten also batches the various database commands into a single request to the database when SaveChangesAsync() is called. We’ve consistently found that to be a very important performance optimization.

Next time…

I’d like to dive into patterns for managing data concurrency.