JasperFx Software offers custom consulting engagements or ongoing support contracts for any part of the Critter Stack. Some of the features in this post were either directly part of client engagements or inspired by our work with JasperFx clients.

This week brought out some new functionality and inevitably some new bug fixes in Marten 7.38 and Wolverine 3.10. I’m actually hopeful this is about the last Marten 7.* release, and Marten 8.0 is heavily underway. Likewise, Wolverine 3.* is probably about played out, and Wolverine 4.0 will come out at the same time. For now though, here’s some highlights of new functionality.

Delete All Marten Data for a Single Tenant

A JasperFx client has a need to occasionally remove all data for a single named tenant across their entire system. Some of their Marten documents and the events themselves are multi-tenanted, while others are global documents. In their particular case, they’re using Marten’s support for managed table partitions by tenant, but other folks might not. To make the process of cleaning out all data for a single tenant as easy as possible regardless of your particular Marten storage configuration, Marten 7.38 added this API:

public static async Task delete_all_tenant_data(IDocumentStore store, CancellationToken token)
{
    await store.Advanced.DeleteAllTenantDataAsync("AAA", token);
}

Rabbit MQ Quorum Queues or Streams with Wolverine

At the request of another JasperFx Software customer, Wolverine has the ability to declare Rabbit MQ quorum queues or streams like so:

var builder = Host.CreateApplicationBuilder();
builder.UseWolverine(opts =>
{
    opts
        .UseRabbitMq(builder.Configuration.GetConnectionString("rabbit"))
        
        // You can configure the queue type for declaration with this
        // usage as well
        .DeclareQueue("stream", q => q.QueueType = QueueType.stream)

        // Use quorum queues by default as a policy
        .UseQuorumQueues()

        // Or instead use streams
        .UseStreamsAsQueues();

    opts.ListenToRabbitQueue("quorum1")
        // Override the queue type in declarations for a
        // single queue, and the explicit configuration will win
        // out over any policy or convention
        .QueueType(QueueType.quorum);
   
    
});

Note that nothing in Wolverine changed other than giving you the ability to make Wolverine declare Rabbit MQ queues as quorum queues or as streams.

Easy Access to Marten Event Sourced Aggregation Data in Wolverine

While the Wolverine + Marten “aggregate handler workflow” is a popular feature for command handlers that may need to append events, sometimes you just want a read only version of an event sourced aggregate. Marten has its FetchLatest API that lets you retrieve the current state of an aggregated projection consistent with the current event store data regardless of the lifecycle of the projection (live, inline, or async). Wolverine now has a quick short cut for accessing that data as a value “pushed” into your HTTP endpoints by decorating a parameter of your handler method with the new [ReadAggregate] attribute like so:

[WolverineGet("/orders/latest/{id}")]
public static Order GetLatest(Guid id, [ReadAggregate] Order order) => order;

or injected into a message handler similarly like this:

public record FindAggregate(Guid Id);

public static class FindLettersHandler
{
    // This is admittedly just some weak sauce testing support code
    public static LetterAggregateEnvelope Handle(
        FindAggregate command, 
        [ReadAggregate] LetterAggregate aggregate)
    
        => new LetterAggregateEnvelope(aggregate);
}

This feature was inspired by a session with a JasperFx Software client where their HTTP endpoints frequently needed to access projected aggregate data for multiple event streams, but only append events to one stream. This functionality was probably already overdue anyway as a way to quickly get projection data any time you just need to read that data as part of a command or query handler.

Coaching my daughter’s 1st/2nd grade basketball team is a trip. I don’t know that the girls are necessarily learning much, but one thing I’d love for them to understand is to “follow your shot” and try for a rebuild for a second shot if the ball doesn’t go in on their first shot. That’s the tortured metaphor/excuse for the marten playing basketball for this post:-)

I’m currently helping a JasperFx Software client to retrofit in some concurrency protection to their existing system that uses Marten for event sourcing by utilizing Marten’s FetchForWriting API deep in the guts of a custom repository to prevent their system from being put into an inconsistent state.

Great, right! Except that there’s not a very real possibility that their application will throw Marten’s ConcurrencyException when an operation fails in Marten’s optimistic concurrency checks.

Our next trick is building in some selective retries for the commands that could probably succeed if they just started over from the new system state after first triggering the concurrency check — and that’s an absolutely perfect use case for the built in Wolverine error handling policies!

This particular system was built around MediatR that doesn’t have any built in error handling policies, and we’ll probably end up rigging up some kind of pipeline behavior or even a flat out decorator MediatR. I did call out the error handling in Wolverine as an advantage in the Wolverine for MediatR Users guide.

In the ubiquitous “Incident Service” example we use in documentation here and there, we have a message handler for trying to automatically assign a priority to an in flight customer reported “Incident” like this:

public static class TryAssignPriorityHandler
{
    // Wolverine will call this method before the "real" Handler method,
    // and it can "magically" connect that the Customer object should be delivered
    // to the Handle() method at runtime
    public static Task<Customer?> LoadAsync(Incident details, IDocumentSession session)
    {
        return session.LoadAsync<Customer>(details.CustomerId);
    }

    // There's some database lookup at runtime, but I've isolated that above, so the
    // behavioral logic that "decides" what to do is a pure function below. 
    [AggregateHandler]
    public static (Events, OutgoingMessages) Handle(
        TryAssignPriority command, 
        Incident details,
        Customer customer)
    {
        var events = new Events();
        var messages = new OutgoingMessages();

        if (details.Category.HasValue && customer.Priorities.TryGetValue(details.Category.Value, out var priority))
        {
            if (details.Priority != priority)
            {
                events.Add(new IncidentPrioritised(priority, command.UserId));

                if (priority == IncidentPriority.Critical)
                {
                    messages.Add(new RingAllTheAlarms(command.IncidentId));
                }
            }
        }

        return (events, messages);
    }
}

The handler above depends on the current state of the Incident in the system, and it’s somewhat possible that two or more people or transactions are happily trying to modify the same Incident at the same time. The Wolverine aggregate handler workflow triggered by the [AggregateHandler] usage up above happily builds in optimistic concurrency protection such that an attempt to save the pending transaction will throw an exception if something else has modified that Incident between the command starting and the call to persist all changes.

Now, depending on the command, you may want to either:

1. Immediately discard the command message because it’s not obsolete
2. Just have the command message retried from scratch, either immediately, with a little delay, or even scheduled for a much later time

Wolverine will happily do that for you. While you can happily set global error handling, you can also fine tune the specific error handling for specific message handlers, exception types, and even exception details as shown below:

public static class TryAssignPriorityHandler
{
    public static void Configure(HandlerChain chain)
    {
        // It's a fall through, so you would only do *one*
        // of these options!

        // It can never succeed, so just discard it instead of wasting
        // time on retries or dead letter queues
        chain.OnException<ConcurrencyException>().Discard();

        // Do some selective retries with a progressive wait
        // in between tries, and if that fails, move it to the dead
        // letter storage
        chain.OnException<ConcurrencyException>()
            .RetryWithCooldown(50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds())
            .Then
            .MoveToErrorQueue();
        
        // Or throw it away after a few tries...
        chain.OnException<ConcurrencyException>()
            .RetryWithCooldown(50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds())
            .Then
            .Discard();
    }
    
    // rest of the handler code...

If you’re processing messages through the asynchronous messaging in Wolverine — and this includes from local, in memory queues too — you have the full set of error policies. If you’re consuming Wolverine as a “Mediator” tool where you may be delegating to Wolverine like so:

public static async Task delegate_to_wolverine(IMessageBus bus, TryAssignPriority command)
{
    await bus.InvokeAsync(command);
}

Wolverine can still use any “Retry” or “Discard” error handling policies, and if Wolverine does a retry, it effectively starts from a completely clean slate so you don’t have to worry about any dirty state from scoped services used by the initial failed attempt to process the message.

Summary

Wolverine puts a ton of emphasis on allowing our users to build low ceremony code that’s highly testable, but we also aren’t compromising on resiliency or observability. While being a “mediator” isn’t really what our hopes and dreams for Wolverine were originally, it does it quite credibly and even brings some of the error handling resiliency that you may be used to in asynchronous messaging frameworks but aren’t always a feature of smaller “mediator” tools.

JasperFx Software works with our customers to help wring the absolute best results out of our customer’s usage of the “Critter Stack.” We build several improvements in collaboration with our customers last year to both Marten and Wolverine specifically to improve scalability of large systems using Event Sourcing. If you’re concerned about whether or not your approach to Event Sourcing will actually scale, definitely look at the Critter Stack, and give JasperFx a shout for help making it all work.

Alright, you’re using Event Sourcing with the whole Critter Stack, and you want to get the best scalability possible in the face of an expected onslaught of incoming events. There’s some “opt in” features in Marten especially that you can take advantage of to get your system going a little bit faster and handle bigger databases.

Using the near ubiquitous “Incident Service” example originally built by Oskar Dudycz, the “Critter Stack” community is building out a new version in the Wolverine codebase that when (and if) finished, will hopefully show off an end to end example of using an event sourced workflow.

In this application we’ll need to track common events for the workflow of a customer reported Incident like when it’s logged, categorised, collects notes, and hopefully gets closed. Coming into this, we think it’s going to get very heavy usage so we expect to have tons of events streaming into the database. We’ve also been told by our business partners that we only need to retain closed incidents in the active views of the user interface for a certain amount of time — but we never want to lose data permanently.

All that being said, let’s look at a few options we can enable in Marten right off the bat:

builder.Services.AddMarten(opts =>
{
    var connectionString = builder.Configuration.GetConnectionString("Marten");
    opts.Connection(connectionString);
    opts.DatabaseSchemaName = "incidents";
    
    // We're going to refer to this one soon
    opts.Projections.Snapshot<Incident>(SnapshotLifecycle.Inline);

    // Use PostgreSQL partitioning for hot/cold event storage
    opts.Events.UseArchivedStreamPartitioning = true;
    
    // Recent optimization that will specifically make command processing
    // with the Wolverine "aggregate handler workflow" a bit more efficient
    opts.Projections.UseIdentityMapForAggregates = true;

    // This is big, use this by default with all new development
    // Long story
    opts.Events.AppendMode = EventAppendMode.Quick;
})
    
// Another performance optimization if you're starting from
// scratch
.UseLightweightSessions()
    
// Run projections in the background
.AddAsyncDaemon(DaemonMode.HotCold)

// This adds configuration with Wolverine's transactional outbox and
// Marten middleware support to Wolverine
.IntegrateWithWolverine();

There are three options here I want to bring to your attention:

1. UseLightweightSessions() directs Marten to use IDocumentSession sessions by default (what’s injected by your DI container) to avoid any performance overhead from identity map tracking in the session. Don’t use this of course if you really do want or need the identity map tracking.
2. opts.Events.UseArchivedStreamPartitioning = true sets us up for Marten’s “hot/cold” event storage scheme using PostgreSQL native partitioning. More on this in the section on stream archiving below. Read more about this feature in the Marten documentation.
3. Setting UseIdentityMapForAggregates = true opts into some recent performance optimizations for updating Inline aggregates through Marten’s FetchForWriting API. More detail on this here. Long story short, this makes Marten and Wolverine do less work and make fewer database round trips to support the aggregate handler workflow I’m going to demonstrate below.
4. Events.AppendMode = EventAppendMode.Quick makes the event appending operations upon saving a Marten session a lot faster, like 50% faster in our testing. It also makes Marten’s “async daemon” feature work smoothly. The downside is that you lose access to some event metadata during Inline projections — which most people won’t care about, but again, we try not to break existing users.

The “Aggregate Handler Workflow”

I have typically described this as Wolverine’s version of the Decider Pattern, but no, I’m now saying that this is a significantly different approach that I believe will lead to better results in larger systems than the “Decider” in that it manages complexity better and handles several technical details that the “Decider” pattern does not. Plus you won’t end up with the humongous switch statements with the Wolverine “Aggregate Handler Workflow” that a Decider function can easily become with any level of domain complexity.

Using Wolverine’s aggregate handler workflow, a command handler that may result in a new event being appended to Marten will look like this one for categorizing an incident:

public static class CategoriseIncidentEndpoint
{
    // This is Wolverine's form of "Railway Programming"
    // Wolverine will execute this before the main endpoint,
    // and stop all processing if the ProblemDetails is *not*
    // "NoProblems"
    public static ProblemDetails Validate(Incident incident)
    {
        return incident.Status == IncidentStatus.Closed 
            ? new ProblemDetails { Detail = "Incident is already closed" } 
            
            // All good, keep going!
            : WolverineContinue.NoProblems;
    }
    
    // This tells Wolverine that the first "return value" is NOT the response
    // body
    [EmptyResponse]
    [WolverinePost("/api/incidents/{incidentId:guid}/category")]
    public static IncidentCategorised Post(
        // the actual command
        CategoriseIncident command, 
        
        // Wolverine is generating code to look up the Incident aggregate
        // data for the event stream with this id
        [Aggregate("incidentId")] Incident incident)
    {
        // This is a simple case where we're just appending a single event to
        // the stream.
        return new IncidentCategorised(incident.Id, command.Category, command.CategorisedBy);
    }
}

The UseIdentityMapForAggregates = true flag optimizes the code above by allowing Marten to use the exact same Incident aggregate object that was originally passed into the Post() method above as the starting basis for updating the Incident data stored in the database. The application of the Inline projection to update the Incident will start with our originally fetched value, apply any new events on top of that, and update the Incident in the same transaction as the events being captured. Without that flag, Marten would have to fetch the Incident starting data from the database all over again when it applies the projection updates while committing the Marten unit of work containing the events.

There’s plenty of rocket science and sophisticated techniques to improving performance, but one simple thing that almost always works out is not repetitively fetching the exact same data from the database if you don’t have to — and that’s the point of the UseIdentityMapForAggregates optimization.

Hot/Cold Storage

Here’s an exciting, relatively new feature in Marten that was planned for years before JasperFx was able to build this for a client late last year. The UseArchivedStreamPartitioning flag sets up your Marten database for “hot / code storage”:

Again, it might require some brain surgery to really improve performance sometimes, but an absolute no-brainer that’s frequently helpful is to just keep your transactional database tables as small and sprightly as possible over time by moving out obsolete or archived data — and that’s exactly what we’re going to do here.

When an Incident event stream is closed, we want to keep that Incident data shown in the user interface for 3 days, then we’d like all the data for that Incident to get archived. Here’s the sample command handler for the CloseIncident command:

public record CloseIncident(
    Guid ClosedBy,
    int Version
);

public static class CloseIncidentEndpoint
{
    [WolverinePost("/api/incidents/close/{id}")]
    public static (UpdatedAggregate, Events, OutgoingMessages) Handle(
        CloseIncident command, 
        [Aggregate]
        Incident incident)
    {
        /* More logic for later
        if (current.Status is not IncidentStatus.ResolutionAcknowledgedByCustomer)
               throw new InvalidOperationException("Only incident with acknowledged resolution can be closed");

           if (current.HasOutstandingResponseToCustomer)
               throw new InvalidOperationException("Cannot close incident that has outstanding responses to customer");

         */
        
        
        if (incident.Status == IncidentStatus.Closed)
        {
            return (new UpdatedAggregate(), [], []);
        }

        return (

            // Returning the latest view of
            // the Incident as the actual response body
            new UpdatedAggregate(),

            // New event to be appended to the Incident stream
            [new IncidentClosed(command.ClosedBy)],

            // Getting fancy here, telling Wolverine to schedule a 
            // command message for three days from now
            [new ArchiveIncident(incident.Id).DelayedFor(3.Days())]);
    }
}

The ArchiveIncident message is being published by this handler using Wolverine’s scheduled message capability so that it will be executed in exactly 3 days time from the current time (you could get fancier and set an exact time to end of business on that day if you wanted).

Note that even when doing the message scheduling, we can still use Wolverine’s cascading message feature. The point of doing this is to keep our handler a pure function that doesn’t have to invoke services, create side effects, or do anything that would force us into asynchronous methods and all of the inherent complexity and noise that inevitably causes.

The ArchiveIncident command handler might just be this:

public record ArchiveIncident(Guid IncidentId);

public static class ArchiveIncidentHandler
{
    // Just going to code this one pretty crudely
    // I'm assuming that we have "auto-transactions"
    // turned on in Wolverine so we don't have to much
    // with the asynchronous IDocumentSession.SaveChangesAsync()
    public static void Handle(ArchiveIncident command, IDocumentSession session)
    {
        session.Events.Append(command.IncidentId, new Archived("It'd done baby!"));
        session.Delete<Incident>(command.IncidentId);
    }
}

When that command executes in three days time, it will delete the projected Incident document from the database and mark the event stream as archived, which will cause PostgreSQL to move that data into the “cold” archived storage.

To close the loop, all normal database operations in Marten specifically filter out archived data with a SQL filter so that they will always be querying directly against the much smaller “active” partition table.

To sum this up, if you use the event archival partitioning and are able to be aggressive about archiving event streams, you can hugely improve the performance of your event sourced application even after you’ve captured a huge number of events because the actual table that Marten is reading and writing from will be relatively stable in side.

As the late, great Stuart Scott would have told us, that’s cooler than the other side of the pillow!

Why aren’t these all defaults?!?

It’s an imperfect world. Every one of the three flags I showed here either subtly change underlying behavior or force additive changes to your application database. The UseIdentityMapForAggregates flag has to be an “opt in” because using that will absolutely give unexpected results for Marten users who mutate the projected aggregate inside of their command handlers (basically anyone doing any type of AggregateRoot base class approach).

Likewise, Marten was originally built using a session with the somewhat more expensive identity map mechanics built in to mimic the commercial tool we were originally trying to replace. I’ve always regretted this decision, but once this has escaped into real systems, changing the underlying behavior absolutely breaks some existing code.

Lastly, introducing the hot/cold partitioning of the event & stream tables to an existing database will cause an expensive database migration, and we certainly don’t want to be inflicting that on unsuspecting users doing an upgrade.

It’s a lot of overhead and compromise, but we’ve chosen to maintain backward compatibility for existing users over enabling out of the box performance improvements.

But wait, there’s more!

Marten has been able to grow quite a bit in capability after I started JasperFx Software as a company to support it. Doing that has allowed us to partner with shops pushing the limits on Marten and Wolverine, and the feedback, collaboration, and yes, compensation has allowed us to push the Critter Stack’s capabilities a lot in the last 18 months.

Wolverine now has the ability to better spread the work of running projections and event subscriptions from Marten over an application cluster.

Sometime in the current quarter, we’re also going to be building and releasing a new “Stream Compacting” feature as another way to deal with archiving data from very long event streams. And yes, a lot of the Event Sourcing community will lecture you about how you should “keep your streams” short, and while there may be some truth to that, that advice is partially around using less capable technical event sourcing solutions. We strive to make Marten & Wolverine more robust so you don’t have to be omniscient and perfect in your upfront modeling.