Controlling Parallelism with Wolverine Background Processing

A couple weeks back I started a new blog series meant to explore Wolverine’s capabilities for background processing. Working in very small steps and only one new concept at a time, the first time out just showed how to set up Wolverine inside a new ASP.Net Core web api service and immediately use it for offloading some processing from HTTP endpoints to background processing by using Wolverine’s local queues and message handlers for background processing. In the follow up post, I added durability to the background processing so that our work being executed in the background would be durable even in the face of application restarts.

In this post, let’s look at how Wolverine allows you to either control the parallelism of your background processing, or restrict the processing to be strictly sequential.

To review, in previous posts we were “publishing” a SignupRequest message from a Minimal API endpoint to Wolverine like so:

app.MapPost("/signup", (SignUpRequest request, IMessageBus bus) 
    => bus.PublishAsync(request));

In this particular case, our application has a message handler for SignupRequest, so Wolverine has a sensible default behavior of publishing the message to a local, in memory queue where each message will be processed in a separate thread from the original HTTP request, and do so asynchronously in the background.

So far, so good? By default, each message type gets its own local, in memory queue, with a default “maximum degree of parallelism” equal to the number of detected processors (Environment.ProcessorCount). In addition, the local queues do not enforce strict ordering by default.

But now, what if you do need to strict sequential ordering? Or if you want to restrict or expand the number of parallel messages that can be processed? Or the get really wild, constrain some messages to running sequentially while other messages run in parallel?

First, let’s see how we could alter the parallelism of our SignUpRequest to an absurd degree and say that up to 20 messages could theoretically be processed at one time by the system. We’ll do that by breaking into the UseWolverine() configuration and adding this:

builder.Host.UseWolverine(opts =>
{
    // The other stuff...

    // Make the SignUpRequest messages be published with even 
    // more parallelization!
    opts.LocalQueueFor<SignUpRequest>()
        
        // A maximum of 20 at a time because why not!
        .MaximumParallelMessages(20);
});

Easy enough, but now let’s say that we want all logical event messages in our system to be handled in the sequential order that our process publishes these messages. An easy way to do that with Wolverine is to have each event message type implement Wolverine’s IEvent marker interface like so:

public record Event1 : IEvent;
public record Event2 : IEvent;
public record Event3 : IEvent;

To be honest, the IEvent and corresponding IMessage and ICommand interfaces were added to Wolverine originally just to make it easier to transition a codebase from using NServiceBus to Wolverine, but those types have little actual meaning to Wolverine. The only way that Wolverine even uses them is for the purpose of “knowing” that a type is an outbound message so that Wolverine can preview the message routing for a type implementing one of these interfaces automatically in diagnostics.

Revisiting our UseWolverine() code block again, we’ll add that publishing rule like this:

builder.Host.UseWolverine(opts =>
{
    // Other stuff...

    opts.Publish(x =>
    {
        x.MessagesImplementing<IEvent>();
        x.ToLocalQueue("events")
            // Force every event message to be processed in the 
            // strict order they are enqueued, and one at a 
            // time
            .Sequential();
        });
});

With the code above, our application would be publishing every single message where the message type implements IEvent to that one local queue named “events” that has been configured to process messages in strict sequential order.

Summary and What’s Next

Wolverine makes it very easy to do work in background processing within your application, and even to easily control the desired parallelism in your application, or to make a subset of messages be processed in strict sequential order when that’s valuable instead.

To be honest, this series is what I go to when I feel like I need to write more Critter Stack content for the week, so it might be a minute or two before there’s a follow up. There’ll be at least two more posts, one on scheduling message execution and an example of using the local processing capabilities in Wolverine to implement the producer/consumer pattern.

Recent Marten & Wolverine Improvements and Roadmap Update

I’d love any feedback on any of this of course. And from something I wrote in a survey of sorts about the commercial product ideas down below yesterday (which is partially a response to a recent query wanting to know how Marten stacks up against AxonIQ from the JVM world):

There’s definitely an opportunity for a full blown Event Driven Architecture stack in the .NET ecosystem – and frankly, Jeremy really wants the Critter Stack to grow into the very best Event Driven Architecture toolset on the planet to the point where shops will purposely adopt .NET just because of the Critter Stack

I’m honestly just thinking out loud in this post, but a lot has been released for both Marten and Wolverine since the big Marten 7.0 release and the last time I published a roadmap update for the two big toolsets.

Here’s some recent highlights you might have missed from the past two months:

What’s Next?

Getting Marten 7.0 and the accompanying Wolverine 2.0 release across the finish line enabled a flurry of follow up features the past two month — largely driven by a couple JasperFx Software client projects (yeah!). Moving forward, I think these are the remaining strategic features that will hopefully go in soon:

  • Marten will get the ability to use PostgreSQL read replicas for read-only queries very soon as a way to scale applications
  • A new, alternative “Quick Append Event” workflow to Marten. The current internal mechanism in Marten for appending events is built for maximal support for “Inline” projections. This new model would simplify the runtime mechanics for appending events and hopefully make the Marten event store more robust in the face of concurrent requests than it is today. This model should also allow for faster performance if users opt into this mechanism.
  • Some ability to efficiently raise or append events (or side effects of some sort) from running projections. This has been in the backlog for a long time. I’d certainly feel better about this if we had some concrete use cases that folks want to do here. The “Quick Append Event” workflow would be a prerequisite
  • Using PostgreSQL partitioning on the Marten streams and events tables. This is the oldest item in the Marten backlog that’s been kicked down the road forever, but I think this is potentially huge for Marten scalability. This would probably be done in conjunction with some tactical improvements to the Marten projection model and the Wolverine aggregate handler workflow to make the archiving more accessible. The biggest issue has always been in how to handle the database migration model for this feature to convert brownfield applications
  • Wolverine 3.0
    • Try to eliminate the hard dependency on Lamar as the IoC container for Wolverine. Most people don’t care, but the folks who do care really don’t like that. So far from my research it looks like the answer is going to be supporting the built in .NET DI container or Lamar with the current Wolverine model — and we can maybe think about supporting other IoC containers with a step back in the runtime optimizations that Wolverine can do today with Lamar. I think it’s quickly coming to the point where all other IoC libraries besides the built in ServiceProvider container from Microsoft die off — even though there are still plenty of areas where that container is lacking compared to alternatives. Alas.
    • Try to apply the Wolverine error handling policies that today only work for Wolverine message handlers to HTTP endpoints

Critter Stack Pro

The Marten & Wolverine community is helping Babu, Jeffry Gonzalez & I brainstorm ideas for the future “Critter Stack Pro” suite of commercial add on tools. The goal is to (make money) make the “Critter Stack” be much more manageable in production environments, help troubleshoot production support issues, heal the system from runtime problems, and understand resource utilization. We don’t have the exact roadmap or exact technical approach locked down yet.

Right now that looks like:

  • A headless library to better distribute Marten projections and subscriptions across a running cluster of processes. This is “ready for testing” by a JasperFx customer
  • A management console application that will be offered both as an ASP.Net Core add on library for a single application or distributed as a standalone Docker image for managing multiple systems from one console
    • Analyze system configuration
    • Manage Wolverine’s “dead letter queue” for messages, including the ability to replay messages
    • Some integration with Open Telemetry and metrics data emitted from Marten and/or Wolverine applications, probably at a summary level with navigation to the “real” observability platform (Prometheus? Grafana? Something totally different?)
    • Management for Marten Asynchronous Projections and Subscriptions
      • Performance information
      • Triggering rebuilds or replays
      • Pausing/restarting projections or subscriptions
    • Tenant Management
      • Dynamically add or remove tenant databases
      • Pause tenants
      • Understand resource utilization and performance on a tenant by tenant basis
    • Marten Event Store Explorer — and we’re collecting several ideas for this
    • Wolverine Message Processing Explorer — ditto
    • Wolverine Scheduled Message Dashboard

My fervent hope is that this tooling will be demonstrable for friendly early adopters at the end of the 2nd quarter, and looking good in the 4th quarter to try to make a serious push for sales in the all important 1st quarter of next year.

And Beyond!

I’m very interested in porting just the event store functionality from Marten to a new library targeting SQL Server as the backing store. The goal here would be to give it the same Wolverine support as the existing Marten functionality. This would be pending some of the Marten projection model stabilizing up above.

Maybe adding CosmosDb and/or DynamoDb support to Wolverine.

And who knows? It’s likely something I’m not even aware of now will be the highest priority in the 3rd and 4th quarters!

Critter Stack Improvements for Event Driven Architecture

JasperFx Software is open for business and offering consulting services (like helping you craft modular monolith strategies!) and support contracts for both Marten and Wolverine so you know you can feel secure taking a big technical bet on these tools and reap all the advantages they give for productive and maintainable server side .NET development.

As a follow on post from First Class Event Subscriptions in Marten last week, let’s introduce Wolverine into the mix for end to end Event Driven Architecture approaches. Using Wolverine’s new Event Subscriptions model, “Critter Stack” systems can automatically process Marten event data with Wolverine message handlers:

If all we want to do is publish Marten event data through Wolverine’s message publishing (which remember, can be either to local queues or external message brokers), we have this simple recipe:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.Services
            .AddMarten()
            
            // Just pulling the connection information from 
            // the IoC container at runtime.
            .UseNpgsqlDataSource()
            
            // You don't absolutely have to have the Wolverine
            // integration active here for subscriptions, but it's
            // more than likely that you will want this anyway
            .IntegrateWithWolverine()
            
            // The Marten async daemon most be active
            .AddAsyncDaemon(DaemonMode.HotCold)
            
            // This would attempt to publish every non-archived event
            // from Marten to Wolverine subscribers
            .PublishEventsToWolverine("Everything")
            
            // You wouldn't do this *and* the above option, but just to show
            // the filtering
            .PublishEventsToWolverine("Orders", relay =>
            {
                // Filtering 
                relay.FilterIncomingEventsOnStreamType(typeof(Order));

                // Optionally, tell Marten to only subscribe to new
                // events whenever this subscription is first activated
                relay.Options.SubscribeFromPresent();
            });
    }).StartAsync();

First off, what’s a “subscriber?” That would mean any event that Wolverine recognizes as having:

  • A local message handler in the application for the specific event type, which would effectively direct Wolverine to publish the event data to a local queue
  • A local message handler in the application for the specific IEvent<T> type, which would effectively direct Wolverine to publish the event with its IEvent Marten metadata wrapper to a local queue
  • Any event type where Wolverine can discover subscribers through routing rules

All the Wolverine subscription is doing is effectively calling IMessageBus.PublishAsync() against the event data or the IEvent<T> wrapper. You can make the subscription run more efficiently by applying event or stream type filters for the subscription.

If you need to do a transformation of the raw IEvent<T> or the internal event type to some kind of external event type for publishing to external systems when you want to avoid directly coupling other subscribers to your system’s internals, you can accomplish that by just building a message handler that does the transformation and publishes a cascading message like so:

public record OrderCreated(string OrderNumber, Guid CustomerId);

// I wouldn't use this kind of suffix in real life, but it helps
// document *what* this is for the sample in the docs:)
public record OrderCreatedIntegrationEvent(string OrderNumber, string CustomerName, DateTimeOffset Timestamp);

// We're going to use the Marten IEvent metadata and some other Marten reference
// data to transform the internal OrderCreated event
// to an OrderCreatedIntegrationEvent that will be more appropriate for publishing to
// external systems
public static class InternalOrderCreatedHandler
{
    public static Task<Customer?> LoadAsync(IEvent<OrderCreated> e, IQuerySession session,
        CancellationToken cancellationToken)
        => session.LoadAsync<Customer>(e.Data.CustomerId, cancellationToken);
    
    
    public static OrderCreatedIntegrationEvent Handle(IEvent<OrderCreated> e, Customer customer)
    {
        return new OrderCreatedIntegrationEvent(e.Data.OrderNumber, customer.Name, e.Timestamp);
    }
}

Process Events as Messages in Strict Order

In some cases you may want the events to be executed by Wolverine message handlers in strict order. With the recipe below:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.Services
            .AddMarten(o =>
            {
                // This is the default setting, but just showing
                // you that Wolverine subscriptions will be able
                // to skip over messages that fail without
                // shutting down the subscription
                o.Projections.Errors.SkipApplyErrors = true;
            })

            // Just pulling the connection information from 
            // the IoC container at runtime.
            .UseNpgsqlDataSource()

            // You don't absolutely have to have the Wolverine
            // integration active here for subscriptions, but it's
            // more than likely that you will want this anyway
            .IntegrateWithWolverine()
            
            // The Marten async daemon most be active
            .AddAsyncDaemon(DaemonMode.HotCold)
            
            // Notice the allow list filtering of event types and the possibility of overriding
            // the starting point for this subscription at runtime
            .ProcessEventsWithWolverineHandlersInStrictOrder("Orders", o =>
            {
                // It's more important to create an allow list of event types that can be processed
                o.IncludeType<OrderCreated>();

                // Optionally mark the subscription as only starting from events from a certain time
                o.Options.SubscribeFromTime(new DateTimeOffset(new DateTime(2023, 12, 1)));
            });
    }).StartAsync();

In this recipe, Marten & Wolverine are working together to call IMessageBus.InvokeAsync() on each event in order. You can use both the actual event type (OrderCreated) or the wrapped Marten event type (IEvent<OrderCreated>) as the message type for your message handler.

In the case of exceptions from processing the event with Wolverine:

  1. Any built in “retry” error handling will kick in to retry the event processing inline
  2. If the retries are exhausted, and the Marten setting for StoreOptions.Projections.Errors.SkipApplyErrors is true, Wolverine will persist the event to its PostgreSQL backed dead letter queue and proceed to the next event. This setting is the default with Marten when the daemon is running continuously in the background, but false in rebuilds or replays
  3. If the retries are exhausted, and SkipApplyErrors = false, Wolverine will tell Marten to pause the subscription at the last event sequence that succeeded

Custom, Batched Subscriptions

The base type for all Wolverine subscriptions is the Wolverine.Marten.Subscriptions.BatchSubscription class. If you need to do something completely custom, or just to take action on a batch of events at one time, subclass that type. Here is an example usage where I’m using event carried state transfer to publish batches of reference data about customers being activated or deactivated within our system:

public record CompanyActivated(string Name);

public record CompanyDeactivated();

public record NewCompany(Guid Id, string Name);

// Message type we're going to publish to external
// systems to keep them up to date on new companies
public class CompanyActivations
{
    public List<NewCompany> Additions { get; set; } = new();
    public List<Guid> Removals { get; set; } = new();

    public void Add(Guid companyId, string name)
    {
        Removals.Remove(companyId);
        
        // Fill is an extension method in JasperFx.Core that adds the 
        // record to a list if the value does not already exist
        Additions.Fill(new NewCompany(companyId, name));
    }

    public void Remove(Guid companyId)
    {
        Removals.Fill(companyId);

        Additions.RemoveAll(x => x.Id == companyId);
    }
}

public class CompanyTransferSubscription : BatchSubscription
{
    public CompanyTransferSubscription() : base("CompanyTransfer")
    {
        IncludeType<CompanyActivated>();
        IncludeType<CompanyDeactivated>();
    }

    public override async Task ProcessEventsAsync(EventRange page, ISubscriptionController controller, IDocumentOperations operations,
        IMessageBus bus, CancellationToken cancellationToken)
    {
        var activations = new CompanyActivations();
        foreach (var e in page.Events)
        {
            switch (e)
            {
                // In all cases, I'm assuming that the Marten stream id is the identifier for a customer
                case IEvent<CompanyActivated> activated:
                    activations.Add(activated.StreamId, activated.Data.Name);
                    break;
                case IEvent<CompanyDeactivated> deactivated:
                    activations.Remove(deactivated.StreamId);
                    break;
            }
        }
        
        // At the end of all of this, publish a single message
        // In case you're wondering, this will opt into Wolverine's
        // transactional outbox with the same transaction as any changes
        // made by Marten's IDocumentOperations passed in, including Marten's
        // own work to track the progression of this subscription
        await bus.PublishAsync(activations);
    }
}

And the related code to register this subscription:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.UseRabbitMq(); 
        
        // There needs to be *some* kind of subscriber for CompanyActivations
        // for this to work at all
        opts.PublishMessage<CompanyActivations>()
            .ToRabbitExchange("activations");
        
        opts.Services
            .AddMarten()

            // Just pulling the connection information from 
            // the IoC container at runtime.
            .UseNpgsqlDataSource()
            
            .IntegrateWithWolverine()
            
            // The Marten async daemon most be active
            .AddAsyncDaemon(DaemonMode.HotCold)

                                
            // Register the new subscription
            .SubscribeToEvents(new CompanyTransferSubscription());
    }).StartAsync();

Summary

The feature set shown here has been a very long planned set of capabilities to truly extend the “Critter Stack” into the realm of supporting Event Driven Architecture approaches from soup to nuts. Using the Wolverine subscriptions automatically gets you support to publish Marten events to any transport supported by Wolverine itself, and does so in a much more robust way than you can easily roll by hand like folks did previously with Marten’s IProjection interface. I’m currently helping a JasperFx Software client utilize this functionality for data exchange that has strict ordering and at least once delivery guarantees.

First Class Event Subscriptions in Marten

This feature has been planned for Marten for years, but finally happened this month because a JasperFx Software client had a complicated multi-tenanted integration need for this as part of a complicated multi-tenanted and order sensitive data integration.

Marten recently (these samples are pulled from Marten 7.9) got a first class “event subscription” feature that allows users to take action upon events being appended to Marten’s event store in strict sequential order in a background process. While you’ve long been able to integrate Marten with other systems by using Marten’s older projection model, the newer subscription model is leaner and more efficient for background processing.

Before I get to “what” it is, let’s say that you need to carry out some kind of background processing on these events as they are captured? For example, maybe you need to:

  • Publish events to an external system as some kind of integration?
  • Carry out background processing based on a captured event
  • Build a view representation of the events in something outside of the current PostgreSQL database, like maybe an Elastic Search view for better searching

With this recently added feature, you can utilize Marten’s ISubscription model that runs within Marten’s async daemon subsystem to “push” events into your subscriptions as events flow into your system. Note that this is a background process within your application, and happen in a completely different thread than the initial work of appending and saving events to the Marten event storage.

Subscriptions will always be an implementation of the ISubscription interface shown below:

/// <summary>
/// Basic abstraction for custom subscriptions to Marten events through the async daemon. Use this in
/// order to do custom processing against an ordered stream of the events
/// </summary>
public interface ISubscription : IAsyncDisposable
{
    /// <summary>
    /// Processes a page of events at a time
    /// </summary>
    /// <param name="page"></param>
    /// <param name="controller">Use to log dead letter events that are skipped or to stop the subscription from processing based on an exception</param>
    /// <param name="operations">Access to Marten queries and writes that will be committed with the progress update for this subscription</param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    Task<IChangeListener> ProcessEventsAsync(EventRange page, ISubscriptionController controller,
        IDocumentOperations operations,
        CancellationToken cancellationToken);
}

So far, the subscription model gives you these abilities:

  • Access to the Marten IDocumentOperations service that is scoped to the processing of a single page and can be used to either query additional data or to make database writes within the context of the same transaction that Marten will use to record the current progress of the subscription to the database
  • Error handling abilities via the ISubscriptionController interface argument that can be used to record events that were skipped by the subscription or to completely stop all further processing
  • By returning an IChangeListener, the subscription can be notified right before and right after Marten commits the database transaction for any changes including recording the current progress of the subscription for the current page. This was done purposely to enable transactional outbox approaches like the one in Wolverine. See the async daemon diagnostics for more information.
  • The ability to filter the event types or stream types that the subscription is interested in as a way to greatly optimize the runtime performance by preventing Marten from having to fetch events that the subscription will not process
  • The ability to create the actual subscription objects from the application’s IoC container when that is necessary
  • Flexible control over where or when the subscription starts when it is first applied to an existing event store
  • Some facility to “rewind and replay” subscriptions

To make this concrete, here’s the simplest possible subscription you can make to simply write out a console message for every event:

public class ConsoleSubscription: ISubscription
{
    public Task<IChangeListener> ProcessEventsAsync(EventRange page, ISubscriptionController controller, IDocumentOperations operations,
        CancellationToken cancellationToken)
    {
        Console.WriteLine($"Starting to process events from {page.SequenceFloor} to {page.SequenceCeiling}");
        foreach (var e in page.Events)
        {
            Console.WriteLine($"Got event of type {e.Data.GetType().NameInCode()} from stream {e.StreamId}");
        }

        // If you don't care about being signaled for
        return Task.FromResult(NullChangeListener.Instance);
    }

    public ValueTask DisposeAsync()
    {
        return new ValueTask();
    }
}

And to register that with our Marten store:

var builder = Host.CreateApplicationBuilder();
builder.Services.AddMarten(opts =>
    {
        opts.Connection(builder.Configuration.GetConnectionString("marten"));

        // Because this subscription has no service dependencies, we
        // can use this simple mechanism
        opts.Events.Subscribe(new ConsoleSubscription());

        // Or with additional configuration like:
        opts.Events.Subscribe(new ConsoleSubscription(), s =>
        {
            s.SubscriptionName = "Console"; // Override Marten's naming
            s.SubscriptionVersion = 2; // Potentially version as an all new subscription

            // Optionally create an allow list of
            // event types to subscribe to
            s.IncludeType<InvoiceApproved>();
            s.IncludeType<InvoiceCreated>();

            // Only subscribe to new events, and don't try
            // to apply this subscription to existing events
            s.Options.SubscribeFromPresent();
        });
    })
    .AddAsyncDaemon(DaemonMode.HotCold);

using var host = builder.Build();
await host.StartAsync();

Here’s a slightly more complicated sample that publishes events to a configured Kafka topic:

public class KafkaSubscription: SubscriptionBase
{
    private readonly KafkaProducerConfig _config;

    public KafkaSubscription(KafkaProducerConfig config)
    {
        _config = config;

        SubscriptionName = "Kafka";

        // Access to any or all filtering rules
        IncludeType<InvoiceApproved>();

        // Fine grained control over how the subscription runs
        // in the async daemon
        Options.BatchSize = 1000;
        Options.MaximumHopperSize = 10000;

        // Effectively run as a hot observable
        Options.SubscribeFromPresent();
    }

    // The daemon will "push" a page of events at a time to this subscription
    public override async Task<IChangeListener> ProcessEventsAsync(
        EventRange page,
        ISubscriptionController controller,
        IDocumentOperations operations,
        CancellationToken cancellationToken)
    {
        using var kafkaProducer =
            new ProducerBuilder<string, string>(_config.ProducerConfig).Build();

        foreach (var @event in page.Events)
        {
            await kafkaProducer.ProduceAsync(_config.Topic,
                new Message<string, string>
                {
                    // store event type name in message Key
                    Key = @event.Data.GetType().Name,
                    // serialize event to message Value
                    Value = JsonConvert.SerializeObject(@event.Data)
                }, cancellationToken);

        }

        // We don't need any kind of callback, so the nullo is fine
        return NullChangeListener.Instance;
    }

}

// Just assume this is registered in your IoC container
public class KafkaProducerConfig
{
    public ProducerConfig? ProducerConfig { get; set; }
    public string? Topic { get; set; }
}

This time, it’s requiring IoC services injected through its constructor, so we’re going to use this mechanism to add it to Marten:

var builder = Host.CreateApplicationBuilder();
builder.Services.AddMarten(opts =>
    {
        opts.Connection(builder.Configuration.GetConnectionString("marten"));
    })
    // Marten also supports a Scoped lifecycle, and quietly forward Transient
    // to Scoped
    .AddSubscriptionWithServices<KafkaSubscription>(ServiceLifetime.Singleton, o =>
    {
        // This is a default, but just showing what's possible
        o.IncludeArchivedEvents = false;

        o.FilterIncomingEventsOnStreamType(typeof(Invoice));

        // Process no more than 10 events at a time
        o.Options.BatchSize = 10;
    })
    .AddAsyncDaemon(DaemonMode.HotCold);

using var host = builder.Build();
await host.StartAsync();

But there’s more!

The subscriptions run with Marten’s async daemon process, which just got a world of improvements in the Marten V7 release, including the ability to distribute work across running nodes in your application at runtime.

I didn’t show it in this blog post, but there are also facilities to configure whether a new subscription will start by working through all the events from the beginning of the system, or whether the subscription should start from the current sequence of the event store, or even go back to an explicitly stated sequence or timestamp, then play forward. Marten also has support — similar to its projection rebuild functionality — to rewind and replay subscriptions.

Wolverine already has specific integrations to utilize Marten event subscriptions to process events with Wolverine message handlers, or to forward events as messages through Wolverine publishing (Kafka? Rabbit MQ? Azure Service Bus?), or to do something completely custom with batches of events at a time (which I’ll demonstrate in the next couple weeks). I’ll post about that soon after that functionality gets fully documented with decent examples.

Lastly, and this is strictly in the hopefully near term future, there will be specific support for Marten subscriptions in the planned “Critter Stack Pro” add on product to Marten & Wolverine to:

  • Distribute subscription work across running nodes within your system — which actually exists in a crude, but effective form, and will absolutely be in Critter Stack Pro V1!
  • User interface monitoring and control pane to manually turn on and off subscriptions, review performance, and manually “rewind” subscriptions

Hopefully much more on this soon. It’s taken much longer than I’d hoped, but it’s still coming.

Strict Ordered Message Handling wth Wolverine

The feature was built for a current JasperFx Software client, and came with a wave of developments across both Marten and Wolverine to support a fairly complex, mission critical set of application integrations. The PostgreSQL transport new to Wolverine was part of this wave. Some time next week I’ll be blogging about the Marten event subscription capabilities that were built into Marten & Wolverine to support this client as well. The point being, JasperFx is wide open for business and we can help your shop succeed with challenging project work!

Wolverine now has the ability to support strict messaging order with its message listeners. Given any random listening endpoint in Wolverine, just add this directive below to make the message processing be strictly sequential (with the proviso that your error handling policies may impact the order on failures):

var host = await Host.CreateDefaultBuilder().UseWolverine(opts =>
{
    opts.UseRabbitMq().EnableWolverineControlQueues();
    
opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, "listeners");

    opts.ListenToRabbitQueue("ordered")
        
        // This option is available on all types of Wolverine
        // endpoints that can be configured to be a listener
        .ListenWithStrictOrdering();
}).StartAsync();

Some notes about the ListenWithStrictOrdering() directive you might have:

  1. It’s supported with every external messaging broker that Wolverine supports, including Kafka, Azure Service Bus, AWS SQS, and Rabbit MQ. It is also supported with the two database backed transports (we have both kinds, Sql Server and PostgreSQL!)
  2. When this directive is applied, Wolverine will only make the listener for each endpoint (in the case above, the Rabbit MQ named “ordered”) be active on a single node within your application. Today that distribution is just crudely spreading out the “exclusive listeners” evenly across the whole application cluster. Definitely note that the strict ordering comes at the cost of reduced throughput, so use this feature wisely! Did I mention that JasperFx Software is here and ready to work with your company on Critter Stack projects?
  3. Every exclusive listener will quickly start up on a single node if WolverineOptions.Durability.Mode = DurabilityMode.Solo, and you may want to do that for local testing and development just to be a little quicker on cold starts
  4. The ListenWithStrictOrdering will make the internal worker queue (Wolverine uses an internal TPL Dataflow ActionBlock in these cases) for “buffered” or “durable” endpoints be strictly sequential
  5. You will have to have a durable message store configured for your application in order for Wolverine to perform the leadership election and “agent tracking” (what’s running where)

Summary

This is a powerful tool in the continually growing Wolverine tool belt. The strict ordering may also be used to alleviate some concurrency issues that some users have hit with event sourcing using Marten when a single stream may be receiving bursts of commands that impact the event stream. The leadership election and agent distribution in Wolverine, in conjunction with this “sticky” listener assignment, gives Wolverine a nascent ability for virtual actors that we will continue to exploit. More soon-ish!

Wolverine’s New PostgreSQL Messaging Transport

Wolverine just got a new PostgreSQL-backed messaging transport (with the work sponsored by a JasperFx Software client!). The use case is just this, say you’re already using Wolverine to build a system with PostgreSQL as your backing database, and want to introduce some asynchronous, background processing in your system — which you could already do with just a database backed, local queue. Going farther though, let’s say that we’d like to have a competing consumers setup for our queueing for load balancing between active nodes and we’d like to do that without having to introduce some kind of new message broker infrastructure into our existing architecture.

That’s time to bring in Wolverine’s new option for asynchronous messaging just using our existing PostgreSQL database. To set that up by itself (without using Marten, but we’ll get to that in a second), it’s these couple lines of code:

var builder = WebApplication.CreateBuilder(args);
var connectionString = builder.Configuration.GetConnectionString("postgres");

builder.Host.UseWolverine(opts =>
{
    // Setting up Postgresql-backed message storage
    // This requires a reference to Wolverine.Postgresql
    opts.PersistMessagesWithPostgresql(connectionString);

    // Other Wolverine configuration
});

Of course, you’d want to setup PostgreSQL queues for Wolverine to send to and to listen to for messages to process. That’s shown below:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine((context, opts) =>
    {
        var connectionString = context.Configuration.GetConnectionString("postgres");
        opts.UsePostgresqlPersistenceAndTransport(connectionString, "myapp")
            
            // Tell Wolverine to build out all necessary queue or scheduled message
            // tables on demand as needed
            .AutoProvision()
            
            // Optional that may be helpful in testing, but probably bad
            // in production!
            .AutoPurgeOnStartup();

        // Use this extension method to create subscriber rules
        opts.PublishAllMessages().ToPostgresqlQueue("outbound");

        // Use this to set up queue listeners
        opts.ListenToPostgresqlQueue("inbound")
            
            .CircuitBreaker(cb =>
            {
                // fine tune the circuit breaker
                // policies here
            })
            
            // Optionally specify how many messages to 
            // fetch into the listener at any one time
            .MaximumMessagesToReceive(50);
    }).StartAsync();

And that’s that, we’re completely set up for messaging via the PostgreSQL database we already have with our Wolverine application!

Just a couple things to note before you run off and try to use this:

  • Like I alluded to earlier, the PostgreSQL queueing mechanism supports competing consumers, so different nodes at runtime can be pulling and processing messages from the PostgreSQL queues
  • There is a separate set of tables for each named queue (one for the actual inbound/outbound messages, and a separate table to segregate “scheduled” messages). Utilize that separation for better performance as needed by effectively sharding the message transfers
  • As that previous bullet point implies, the PostgreSQL transport is able to support scheduled message delivery
  • As in most cases, Wolverine is able to detect whether or not the necessary tables all exist in your database, and create any missing tables for you at runtime
  • In the case of using Wolverine with Marten multi-tenancy through separate databases, the queue tables will exist in all tenant databases
  • There’s some optimizations and integration between these queues and the transactional inbox/outbox support in Wolverine for performance by reducing database chattiness whenever possible

Summary

I’m not sure I’d recommend this approach over dedicated messaging infrastructure for high volumes of messages, but it’s a way to get things done with less infrastructure in some cases and it’s a valuable tool in the Wolverine toolbox.

Modular Monoliths and the “Critter Stack”

JasperFx Software is open for business and offering consulting services (like helping you craft modular monolith strategies!) and support contracts for both Marten and Wolverine so you know you can feel secure taking a big technical bet on these tools and reap all the advantages they give for productive and maintainable server side .NET development.

I’ve been thinking, discussing, and writing a bit lately about the whole “modular monolith” idea, starting with Thoughts on “Modular Monoliths” and continuing onto Actually Talking about Modular Monoliths. This time out I think I’d like to just put out some demos and thoughts about where Marten and Wolverine fit well into the modular monolith idea — and also some areas where I think there’s room for improvement.

First off, let’s talk about…

Modular Configuration

Both tools use the idea of a “configuration model” (what Marten Fowler coined a Semantic Model years ago) that is compiled and built from a combination of attributes in the code, explicit configuration, user supplied policies, and built in policies in baseline Marten or Wolverine as shown below with an indication of the order of precedence:

In code, when you as a user configure Marten and Wolverine inside of the Program file for your system like so:

builder.Services.AddMarten(opts =>
{
    var connectionString = builder.Configuration.GetConnectionString("marten");
    opts.Connection(connectionString);

    // This will create a btree index within the JSONB data
    opts.Schema.For<Customer>().Index(x => x.Region);
})
    // Adds Wolverine transactional middleware for Marten
    // and the Wolverine transactional outbox support as well
    .IntegrateWithWolverine();

builder.Host.UseWolverine(opts =>
{
    opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Static;
    
    // Let's build in some durability for transient errors
    opts.OnException<NpgsqlException>().Or<MartenCommandException>()
        .RetryWithCooldown(50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds());

    // Shut down the listener for whatever queue experienced this exception
    // for 5 minutes, and put the message back on the queue
    opts.OnException<MakeBelieveSubsystemIsDownException>()
        .PauseThenRequeue(5.Minutes());

    // Log the bad message sure, but otherwise throw away this message because
    // it can never be processed
    opts.OnException<InvalidInputThatCouldNeverBeProcessedException>()
        .Discard();
    
    
    
    // Apply the validation middleware *and* discover and register
    // Fluent Validation validators
    opts.UseFluentValidation();
    
    // 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();
    
    // Connecting to a local Rabbit MQ broker
    // at the default port
    opts.UseRabbitMq();

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

        // Pause processing on this local queue for 1 minute if there's
        // more than 20% failures for a period of 2 minutes
        .CircuitBreaker(cb =>
        {
            cb.PauseTime = 1.Minutes();
            cb.SamplingPeriod = 2.Minutes();
            cb.FailurePercentageThreshold = 20;
            
            // Definitely worry about this type of exception
            cb.Include<TimeoutException>();
            
            // Don't worry about this type of exception
            cb.Exclude<InvalidInputThatCouldNeverBeProcessedException>();
        });
    
    // 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();
    });
});

The nested lambdas in AddMarten() and UseWolverine() are configuring the MartenOptions and WolverineOptions models respectively (the “configuration model” in that diagram above).

I’m not aware of any one commonly used .NET idiom for building modular configuration, but I do commonly see folks using extension methods for IServiceCollection or IHostBuilder to segregate configuration that’s specific to a single module, and that’s what I think I’d propose. Assuming that we have a module in our modular monolith system for handling workflow around “incidents”, there might be an extension method something like this:

public static class IncidentsConfigurationExtensions
{
    public static WebApplicationBuilder AddIncidentModule(this WebApplicationBuilder builder)
    {
        // Whatever other configuration, services, et al
        // we need for just the Incidents module
        
        // Extra Marten configuration
        builder.Services.ConfigureMarten(opts =>
        {
            // I'm just adding an index for a document type within
            // this module
            opts.Schema.For<IncidentDetails>()
                .Index(x => x.Priority);
            
            // Purposely segregating all document types in this module's assembly
            // to a separate database schema
            opts.Policies.ForAllDocuments(m =>
            {
                if (m.DocumentType.Assembly == typeof(IncidentsConfigurationExtensions).Assembly)
                {
                    m.DatabaseSchemaName = "incidents";
                }
            });
        });
        
        return builder;
    }
}

Which would be called from the overall system’s Program file like so:

var builder = WebApplication.CreateBuilder(args);

builder.AddIncidentModule();

// Much more below...

Between them, the two main Critter Stack tools have a lot of support for modularity through:

All of the facilities I described above can be used to separate specific configuration for different modules within the module code itself.

Modular Monoliths and Backing Persistence

In every single experience report you’ll ever find about a team trying to break up and modernize a large monolithic application the authors will invariably say that breaking apart the database was the single most challenging task. If we are really doing things better this time with the modular monolith approach, we’d probably better take steps ahead of time to make it easier to extract services by attempting to keep the persistence for each module at least somewhat decoupled from the persistence of other modules.

Going even farther in terms of separation, it’s not unlikely that some modules have quite different persistence needs and might be better served by using a completely different style of persistence than the other modules. Just as an example, one of our current JasperFx Software clients has a large monolithic application where some workflow-centric modules would be a good fit for an event sourcing approach, while other modules are more CRUD centric or reporting-centric where a straight up RDBMS approach is probably much more appropriate.

So let’s finally bring Marten and Wolverine into the mix and talk about the Good, the Bad, and the (sigh) Ugly of how the Critter Stack fits into modular monoliths:

wah, wah, wah…

Let’s start with a positive. Marten sits on top of the very robust PostgreSQL database. So in addition to Marten’s ability to use PostgreSQL as a document database and as an event store, PostgreSQL out of the box is a rock solid relational database. Heck, PostgreSQL even has some ability to be used as either a graph database! The point is that using the Marten + PostgreSQL combination gives you a lot of flexibility in terms of persistence style between different modules in a modular monolith without introducing a lot more infrastructure. Moreover, Wolverine can happily utilize its PostgreSQL-backed transactional outbox with both Entity Framework Core and Marten targeting the same PostgreSQL database in the same application.

Continuing with another positive, let’s say that we want to create some logical separation between our modules in the database, and one way to do so would be to simply keep Marten documents in separate database schemas for each module. Repeating a code sample from above, you can see that configuration below:

    public static WebApplicationBuilder AddIncidentModule(this WebApplicationBuilder builder)
    {
        // Whatever other configuration, services, et al
        // we need for just the Incidents module
        
        // Extra Marten configuration
        builder.Services.ConfigureMarten(opts =>
        {
            // Purposely segregating all document types in this module's assembly
            // to a separate database schema
            opts.Policies.ForAllDocuments(m =>
            {
                if (m.DocumentType.Assembly == typeof(IncidentsConfigurationExtensions).Assembly)
                {
                    m.DatabaseSchemaName = "incidents";
                }
            });
        });
        
        return builder;
    }

So, great, the storage for Marten documents could easily be segregated by schema. Especially considering there’s little or no referential integrity relationships between Marten document tables, it should be relatively easy to move these document tables to completely different databases later!

And with that, let’s move more into “Bad” or hopefully not too “Ugly” territory.

The event store data in Marten is all in one single set of tables (mt_streams and mt_events). So every module utilizing Marten’s event sourcing could be intermingling their events in just these tables through the one single, AddMarten() store for the application’s IHost. You could depend on marking event streams by their aggregate type like so:

public static async Task start_stream(IDocumentSession session)
{
    // the Incident type argument is strictly a marker for
    // Marten
    session.Events.StartStream<Incident>(new IncidentLogged());
    await session.SaveChangesAsync();
}

I think we could ameliorate this situation with a couple future changes:

  1. A new flag in Marten that would make it mandatory to mark every new event stream with an aggregate type specifically to make it easier to separate the events later to extract a service and its event storage
  2. Some kind of helper to move event streams from one database to another. It’s just not something we have in our tool belt at the moment

Of course, it would also help immeasurably if we had a way to split the event store storage for different types of event streams, but somehow that idea has never gotten any traction within Marten and never rises to the level of a high priority. Most of our discussions about sharding or partitioning the event store data has been geared around scalability — which is certainly an issue here too of course.

Marten also has its concept of “separate stores” that was meant to allow an application to interact with multiple Marten-ized databases from a single .NET process. This could be used with modular monoliths to segregate the event store data, even if targeting the same physical database in the end. The very large downside to this approach is that Wolverine’s Marten integration does not today do anything with the separate store model. So no Wolverine transactional middleware, event forwarding, transactional inbox/outbox integration, and no aggregate handler workflow. So basically everything about the full “Critter Stack” integration that makes that tooling the single most productive event sourcing development experience in all of .NET (in my obviously biased opinion). Ugly.

Randomly, I heard an NPR interview with Eli Wallach very late in his life who was the actor who played the “Ugly” character in the famous western, and I could only describe him as effusively jolly. So basically a 180 degree difference from his character!

Module to Module Communication

I’ve now spent much more time on this post than I had allotted, so it’s time to go fast…

In my last post I used this diagram to illustrate the risk of coupling modules through direct usage of internals (the red arrows):

Instead of the red arrows everywhere above, I think I’m in favor of trying to limit the module to module communication to using some mix of a “mediator” tool or an in memory message bus between modules. That’s obviously going to come with some overhead, but I think (hope) that overhead is a net positive.

For a current client, I’m recommending they further utilize MediatR as they move a little more in the direction of modularity in their current monolith. For greenfield codebases, I’d recommend Wolverine instead because I think it does much, much more.

First, Wolverine has a full set of functionality to be “just a Mediator” to decouple modules from too much of the internals of another module. Secondly, Wolverine has a lot of support for background processing through local, in memory queues that could be very advantageous in modular monoliths where Wolverine can de facto be an in memory message bus. Moreover, Wolverine’s main entry point usage is identical for messages processed locally versus messages published through external messaging brokers to external processes:

    public static async Task using_message_bus(IMessageBus bus)
    {
        // Use Wolverine as a "mediator"
        // This is normally executed inline, in process, but *could*
        // also be invoking this command in an external process
        // and waiting for the success or failure ack
        await bus.InvokeAsync(new CategoriseIncident());


        // Use Wolverine for asynchronous messaging. This could 
        // start by publishing to a local, in process queue, or
        // it could be routed to an external message broker -- but
        // the calling code doesn't have to know that
        await bus.PublishAsync(new CategoriseIncident());
    }

The point here is that Wolverine can potentially set your modular monolith architecture up so that it’s possible to extract or move functionality out into separate services later.

All that being said about messaging or mediator tools, some of the ugliest systems I’ve ever seen utilized messaging or proto-Mediatr command handlers between logical modules. Those systems had code that was almost indecipherable by introducing too many layers and far too much internal messaging. I think I’d say that some of the root cause of the poor system code was from getting the bounded context boundaries wrong so that the messaging was too chatty. Using high ceremony anti-corruption layers also adds a lot of mental overhead to follow information flow through several mapping transformations. One of these systems was using the iDesign architectural approach that I think naturally leads to very poorly factored software architectures and too much harmful code ceremony. I do not recommend.

I guess my only point here is that no matter what well intentioned advice people like me try to give, or any theory of how to make code more maintainable any of us might have, if you find yourself saying to yourself about code that “this is way harder than it should be” you should challenge the approach and look for something different — even if that just leads you right back to where you are now if the alternatives don’t look any better.

One important point here about both modular monoliths or a micro service strategy or a mix of the two: if two or more services/modules are chatty between themselves and very frequently have to be modified at the same time, they’re best described as a single bounded context and should probably be combined into a single service or module.

Summary

Anyway, that’s enough from me on this subject for now, and this took way longer than I meant to spend on it. Time to get my nose back to the grindstone. I am certainly very open to any feedback about the Critter Stack tools limitations for modular monolith construction and any suggestions or requests to improve those tools.

Durable Background Processing with Wolverine

A couple weeks back I started a new blog series meant to explore Wolverine’s capabilities for background processing. Working in very small steps and only one new concept at a time, the first time out just showed how to set up Wolverine inside a new ASP.Net Core web api service and immediately use it for offloading some processing from HTTP endpoints to background processing by using Wolverine’s local queues and message handlers for background processing.

In that previous post though, the messages held in those in memory, local queues could conceivably be lost if the application is shut down unexpectedly (Wolverine will attempt to “drain” the local queues of outstanding work on graceful process shutdowns). That’s perfectly acceptable sometimes, but in other times you really need those queued up messages to be durable so that the in flight messages can be processed even if the service process is unexpectedly killed while work is in flight — so let’s opt into Wolverine’s ability to do exactly that!

To that end, let’s just assume that we’re a very typical .NET shop and we’re already using Sql Server as our backing database for the system. Knowing that, let’s add a new Nuget reference to our project:

dotnet add package WolverineFx.SqlServer

And let’s break into our Program file for the service where all the system configuration is, and expand the Wolverine configuration within the UseWolverine() call to this:

// This is good enough for what we're trying to do
// at the moment
builder.Host.UseWolverine(opts =>
{
    // Just normal .NET stuff to get the connection string to our Sql Server database
    // for this service
    var connectionString = builder.Configuration.GetConnectionString("SqlServer");
    
    // Telling Wolverine to build out message storage with Sql Server at 
    // this database and using the "wolverine" schema to somewhat segregate the 
    // wolverine tables away from the rest of the real application
    opts.PersistMessagesWithSqlServer(connectionString, "wolverine");
    
    // In one fell swoop, let's tell Wolverine to make *all* local
    // queues be durable and backed up by Sql Server 
    opts.Policies.UseDurableLocalQueues();
});

Nothing else in our previous code needs to change. As a matter of fact, once you restart your application — assuming that your box can reach the Sql Server database in the appsettings.json file — Wolverine is going to happily see that those necessary tables are missing, and build them out for you in your database so that Wolverine “can just work” on its first usage. That automatic schema creation can of course be disabled and/or done with pure SQL through other Wolverine facilities, but for right now, we’re taking the easy road.

Before I get into the runtime mechanics, here’s a refresher about our first message handler:

public static class SendWelcomeEmailHandler
{
    public static void Handle(SignUpRequest @event, ILogger logger)
    {
        // Just logging, a real handler would obviously do something real
        // to send an email
        logger.LogInformation("Send a Send a welcome email to {Name} at {Email}", @event.Name, @event.Email);
    }
}

And the code that publishes a SignUpRequest message to a local Wolverine queue in a Minimal API endpoint:

app.MapPost("/signup", (SignUpRequest request, IMessageBus bus) 
    => bus.PublishAsync(request));

After our new configuration up above to add message durability to our local queues, when a service client posts a SignUpRequest message is published to Wolverine as a result of a client posting valid data to the /signup Url, Wolverine will:

  1. Persist all the necessary information about the new SignUpRequest message that Wolverine uses to process that message in the Sql Server database (this is using the “Envelope Wrapper” pattern from the old EIP book, which is quite originally called Envelope in the Wolverine internals).
  2. If the message is successfully processed, Wolverine will delete that stored record for the message in Sql Server
  3. If the message processing fails, and there’s some kind of retry policy in effect, Wolverine will increment the number of failed attempts in the Sql Server database (with an UPDATE statement because it’s trying to be as efficient as possible)
  4. If the process somehow fails while the message is floating around in the in memory queues, Wolverine will be able to recover that local message from the database storage later when the system is restarted. Or if the system is running in a load balanced cluster, a different Wolverine node will be able to see that the messages are orphaned in the database and will steal that work into another node so that the messages eventually get processed

Summary and What’s Next?

That’s a lot of detail about what is happening in your system, but I’d argue that was very little code necessary to make the background processing with Wolverine be durable. And all without introducing any other new infrastructure other than the Sql Server database we were probably already using. Moreover, Wolverine can do a lot to make the necessary database setup for you at runtime so there’s hopefully very little friction getting up and running after a fresh git clone.

I’ll add at least a couple more entries to this series by looking at error handling strategies, controller the parallelism or strict ordering of message processing, a simple implementation of the Producer/Consumer pattern with Wolverine, and message scheduling.

Wolverine’s Baked In Integration Testing Support

Hey, did you know that JasperFx Software offers formal support plans for Marten and Wolverine? Not only are we making 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 some completely different .NET server side tooling.

Hey, when you’re building grown up software systems in a responsible way, who likes effective automated testing? Me, too! Moreover, I like automated tests that are reliable — and anyone who has ever been remotely near a large automated test suite testing through the web application layer with any kind of asynchronous behavior knows exactly how painful “flake-y” tests are that suffer from timing issues.

Wolverine of course is an application framework for performing background processing and asynchronous messaging — meaning that there’s no end of the exact kind of asynchronous behavior that is notoriously hard to deal with in automated tests. At a minimum, what you need is a way to exercise the message handling within Wolverine (the “act” in the “act, arrange, assert” test pattern), but wait until all cascading activity is really complete before allowing the automated test to continue making assertions on expected outcomes. Fortunately, Wolverine has that very functionality baked into its core library. Here’s a fake saga that we recently used to fix a bug in Wolverine:

public class LongProcessSaga : Saga
{
    public Guid Id { get; init; }
    
    [Middleware(typeof(BeginProcessMiddleware))]
    public static (LongProcessSaga, OutgoingMessages) Start(BeginProcess message, RecordData? sourceData = null)
    {
        var outgoingMessages = new OutgoingMessages();

        var saga = new LongProcessSaga
        {
            Id = message.DataId,
        };

        if (sourceData is not null)
        {
            outgoingMessages.Add(new ContinueProcess(saga.Id, message.DataId, sourceData.Data));
        }

        return (
            saga,
            outgoingMessages
        );
    }

    public void Handle(ContinueProcess process)
    {
        Continued = true;
    }

    public bool Continued { get; set; }
}

When the BeginProcess message is handled by Wolverine, it might also spawn a ContinueProcess message. So let’s write a test that exercises the first message, but waits until the second message that we expect to be spawned while handling the first message before allowing the test to proceed:

    [Fact]
    public async Task can_compile_without_issue()
    {
        // Arrange -- and sorry, it's a bit of "Arrange" to get an IHost
        var builder = WebApplication.CreateBuilder(Array.Empty<string>());

        builder.Services
            .AddMarten(options =>
            {
                options.Connection(Servers.PostgresConnectionString);
            })
            .UseLightweightSessions()
            .IntegrateWithWolverine();

        builder.Host.UseWolverine(options =>
        {
            options.Discovery.IncludeAssembly(GetType().Assembly);
            
            options.Policies.AutoApplyTransactions();
            options.Policies.UseDurableLocalQueues();
            options.Policies.UseDurableOutboxOnAllSendingEndpoints();
        });

        builder.Services.AddScoped<IDataService, DataService>();

        // This is using Alba, which uses WebApplicationFactory under the covers
        await using var host = await AlbaHost.For(builder, app =>
        {
            app.MapWolverineEndpoints();
        });

        // Finally, the "Act"!
        var originalMessage = new BeginProcess(Guid.NewGuid());
        
        // This is a built in extension method to Wolverine to "wait" until
        // all activity triggered by this operation is completed
        var tracked = await host.InvokeMessageAndWaitAsync(originalMessage);
        
        // And now it's okay to do assertions....
        // This would have failed if there was 0 or many ContinueProcess messages
        var continueMessage = tracked.Executed.SingleMessage<ContinueProcess>();
        
        continueMessage.DataId.ShouldBe(originalMessage.DataId);

    }

The IHost.InvokeMessageAndWaitAsync() is part of Wolverine’s “tracked session” feature that’s descended from an earlier system some former colleagues and I developed and used at my then employer about a decade ago. The original mechanism was quite successful for our integration testing efforts of the time, and was built into Wolverine quite early. This “tracked session” feature is very heavily used within the Wolverine test suites to test Wolverine itself.

But wait, there’s more! Here’s a bigger sample from the documentation just showing you some more things that are possible:

public async Task using_tracked_sessions_advanced(IHost otherWolverineSystem)
{
    // The point here is just that you somehow have
    // an IHost for your application
    using var host = await Host.CreateDefaultBuilder()
        .UseWolverine().StartAsync();

    var debitAccount = new DebitAccount(111, 300);
    var session = await host
            
        // Start defining a tracked session 
        .TrackActivity()
        
        // Override the timeout period for longer tests
        .Timeout(1.Minutes())
        
        // Be careful with this one! This makes Wolverine wait on some indication
        // that messages sent externally are completed
        .IncludeExternalTransports()
        
        // Make the tracked session span across an IHost for another process
        // May not be super useful to the average user, but it's been crucial
        // to test Wolverine itself
        .AlsoTrack(otherWolverineSystem)

        // This is actually helpful if you are testing for error handling 
        // functionality in your system
        .DoNotAssertOnExceptionsDetected()
        
        // Again, this is testing against processes, with another IHost
        .WaitForMessageToBeReceivedAt<LowBalanceDetected>(otherWolverineSystem)
        
        // There are many other options as well
        .InvokeMessageAndWaitAsync(debitAccount);

    var overdrawn = session.Sent.SingleMessage<AccountOverdrawn>();
    overdrawn.AccountId.ShouldBe(debitAccount.AccountId);
}

As hopefully implied by the earlier example, the “tracked session” functionality also gives you:

  • Recursive tracking of all message activity to wait for everything to finish
  • Enforces timeouts in case of hanging tests that probably won’t finish successfully
  • The ability to probe the exact messaging activity that happened as a result of your original message
  • Visibility into any exceptions recorded by Wolverine during message processing that might otherwise be hidden from you. This functionality will re-throw these exceptions to fail a test unless explicitly told to ignore processing exceptions — which you may very well want to do to test error handling logic
  • If a test fails because of a timeout, or doesn’t reach the expected conditions, the test failure exception will show you a (hopefully) neatly formatted textual table explaining what it did observe in terms of what messages were sent, received, started, and finished executing. Again, this is to give you more visibility into test failures, because those inevitably do happen!

Last Thoughts

Supporting a complicated OSS tool like Marten or Wolverine is a little bit like being trapped in somewhere in Jurassic Park while the raptors (users, and especially creative users) are prowling around the perimeter of your tool just looking for weak spots in your tools — a genuine bug, a use case you didn’t anticipate, an awkward API, some missing documentation, or even just some wording in your documentation that isn’t clear enough. The point is, it’s exhausting and sometimes demoralizing when raptors are getting past your defenses a little too often just because you rolled out a near complete rewrite of your LINQ provider subsystem:)

Yesterday I was fielding questions from a fellow whose team was looking to move to Wolverine from one of the older .NET messaging frameworks, and he was very complimentary of the integration testing support that’s the subject of this post. My only point here is to remember to celebrate your successes to balance out the constant worry about what’s not yet great about your tool or project or codebase.

And by success, I mean a very important feature that will absolutely help teams build reliable software more productively with Wolverine that does not exist in other .NET messaging frameworks. And certainly doesn’t exist in the yet-to-be-built Microsoft eventing framework where they haven’t even considered the idea of testability.

Conventional Message Routing in Wolverine

I got a little feedback over the weekend that some folks newly encountering Wolverine for the first time think that it’s harder to use than some other tools because “it doesn’t do the message routing for you.” This fortunately isn’t true, but there’s obviously some work to do on improving documentation and samples to dispel that impression.

For the sake of this post, let’s assume that you want to use Wolverine for asynchronous messaging between processes using an external messaging broker transport (or using asynchronous messaging in a single application but queueing up work in an external message broker). And while Wolverine does indeed have support for interoperability with non-Wolverine applications, let’s assume that it’s going to be Wolverine on both sides of all the message pipes.

First though, just know that for all for external transports with Wolverine, the conventional routing is opt in, meaning that you have to explicitly turn it on when you configure Wolverine within the UseWolverine() bootstrapping. Likewise, know that you can also control exactly how Wolverine configures the listening or message sending behavior in the conventionally determined endpoints.

Now then, to just go fast and make Wolverine do all the message routing for you with predictable conventions “just” see these recipes:

For Wolverine’s Rabbit MQ integration, you can opt into conventional routing like this:

        using var host = await Host.CreateDefaultBuilder()
            .UseWolverine(opts =>
            {
                opts.UseRabbitMq()
                    // Opt into conventional Rabbit MQ routing
                    .UseConventionalRouting();
            }).StartAsync();

In this convention, message routing is to:

  1. Publish all messages to a Rabbit MQ exchange named after Wolverine’s message type name that’s effectively the alias for that message type. Most of the time that’s just the full name of the concrete .NET type.
  2. Create a Rabbit MQ queue named with Wolverine’s message type name for the message type of every known message handler within your application. Wolverine also configures a binding from a Rabbit MQ exchange with that same name to the Rabbit MQ queue for that message type.

This routing behavior was absolutely influenced by similar functionality in the older MassTransit framework. Imitation is the sincerest form of flattery, plus I just agree with what they did anyway. Wolverine adds additional value through its richer resource setup model (AutoProvision / AddResourceSetupOnStartup()) and stronger model of automatic handler discovery.

That’s the quickest possible way to get started, but you have plenty of other levers, knobs, and dials to farther control the conventions as shown below:

        using var host = await Host.CreateDefaultBuilder()
            .UseWolverine(opts =>
            {
                opts.UseRabbitMq()
                    // Opt into conventional Rabbit MQ routing
                    .UseConventionalRouting(c =>
                    {
                        // Override naming conventions
                        c.QueueNameForListener(type => type.Name);
                        c.ExchangeNameForSending(type => type.Name);

                        // Override the configuration for each listener
                        c.ConfigureListeners((l, _) =>
                        {
                            l.ProcessInline().ListenerCount(5);
                        });

                        // Override the sending configuration for each subscriber
                        c.ConfigureSending((s, _) =>
                        {
                            s.UseDurableOutbox();
                        });
                    })
                    
                    // Let Wolverine both discover all these necessary exchanges, queues,
                    // and binding, then also build them as necessary on the broker if
                    // they are missing
                    .AutoProvision();
            }).StartAsync();

The “extra” optional configuration I used above hopefully show you how to take more exacting control over the conventional routing, but the pure defaults in the first sample will help you get up and going fast.

Let’s move on.

Azure Service Bus

Wolverine’s Azure Service Bus integration comes with a pair of conventional routing options. The simpler is to let Wolverine create and utilize an Azure Service Bus queue named after each message type name for both sending the receiving with this syntax:

        using var host = await Host.CreateDefaultBuilder()
            .UseWolverine(opts =>
            {
                opts.UseAzureServiceBus("some connection string")
                    .UseConventionalRouting();
            }).StartAsync();

Slightly more complicated is another option to use Azure Service Bus topics and subscriptions like so:

    // opts is a WolverineOptions object    
    opts.UseAzureServiceBusTesting()
                
        .UseTopicAndSubscriptionConventionalRouting(convention =>
                {
                    // Optionally control every aspect of the convention and
                    // its applicability to types
                    // as well as overriding any listener, sender, topic, or subscription
                    // options
                });

In this usage:

  1. Outgoing messages are routed to an Azure Service Bus topic named after Wolverine’s message type name for that type
  2. At application startup, Wolverine will listen for an Azure Service Bus subscription for each message type from the application’s known message handlers. Likewise, that subscription will automatically be bound to a topic named after the message type name

As was the case with the Rabbit MQ conventions shown first, you have complete control over the naming conventions, the listener configuration, and the subscriber configuration.

For Wolverine’s AWS SQS integration, you can conventionally route to SQS queues named after Wolverine’s message type name (by default) like this:

        var host = await Host.CreateDefaultBuilder()
            .UseWolverine(opts =>
            {
                opts.UseAmazonSqsTransport()
                    .UseConventionalRouting();

            }).StartAsync();

As was the case with the Rabbit MQ conventions shown first, you have complete control over the naming conventions, the listener configuration, and the subscriber configuration.

Wolverine’s Kafka integration is a little bit different animal. In this case you can set up rules to publish to different Kafka topics by message type like this:

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

                opts.PublishAllMessages().ToKafkaTopics();

                opts.Services.AddResourceSetupOnStartup();
            }).StartAsync();

In this case, Wolverine will send each message type to a topic name derived from the message type that’s either Wolverine’s message type name or an explicitly configured topic name configured by attribute like:

[Topic("color.purple")]
public class PurpleMessage
{
}

What if I want a completely different routing convention?!?

There’s an extension point for programmatic message routing rules in Wolverine that’s utilized by all the capabilities shown above called IMessageRouteSource, but if you think you really want to use that, maybe just head to the Wolverine Discord room and we’ll try to help you out!

Summary

Wolverine has strong support for conventional message routing using external brokers, but you need to make at least a one line of code entry in your configuration to explicitly add this behavior to your system. In all cases, Wolverine is able to help build the necessary queues, topics, subscriptions, binding, and exchanges derived by these conventions in your message broker for an efficient developer experience. Moreover, you have complete power to fine tune the usage of this conventional routing for your application.