Last week I helped a JasperFx Software client with a use case where they get a steady stream of related events from an upstream system into a downstream system where order of processing is important, but the messages might arrive out of order.

Once again referring to the venerable Enterprise Integration Patterns book, that scenario requires a Resequencer:

How can we get a stream of related but out-of-sequence messages back into the correct order?

EIP ReSequencer

To solve the message ordering challenge, we introduced the new Resequencer Saga feature into Wolverine, and combined that with the existing “Partitioned Sequential Messaging” feature.

For the new built in re-sequencing, we do need you to implement this interface on any message types in that related stream so that Wolverine “knows” what order the message is inside of a related stream:

public interface SequencedMessage
{
    int? Order { get; }
}

The next step is to use a special kind of new Wolverine Saga called ResequencerSaga<T>, where the T is just some sort of common interface for all the message types that are part of this ordered stream and also implements the SequencedMessage shown above. Here’s a simple example I used for the testing:

public record StartMyWorkflow(Guid Id);
public record MySequencedCommand(Guid SagaId, int? Order) : SequencedMessage;
public class MyWorkflowSaga : ResequencerSaga<MySequencedCommand>
{
    public Guid Id { get; set; }
    public static MyWorkflowSaga Start(StartMyWorkflow cmd)
    {
        return new MyWorkflowSaga { Id = cmd.Id };
    }
    public void Handle(MySequencedCommand cmd)
    {
        // This will only be called when messages arrive in the correct order,
        // or when out-of-order messages are replayed after gaps are filled
    }
}

At runtime, when Wolverine gets a message that is handled by that MyWorkflowSaga, there is some middleware that first compares the declared order of that message against the recorded state of the saga so far. In more concrete terms, if…

• It’s the first message in the sequence, Wolverine just processes it as normal and records in the saga state what the last processed message order was so that it “knows” what message sequence should be next
• It’s a later message in the sequence compared to the last message sequence processed, the saga state will just store the current message, persist the saga state, and otherwise skip the normal message processing
• The message is the next in the sequence according to what the saga state says should be processed next, it processes normally. If there are any previously out of order messages that the saga state already knows about that are sequentially next after the current message, Wolverine will re-publish those messages locally — but with the normal Wolverine message sequencing these cascading messages will not go anywhere until the initiating message completes

With this mechanism, Wolverine is able to put the messages arriving from the outside world back into the correct sequential order in its own processing.

Of course though, this processing is very stateful and somewhat likely to be vulnerable to concurrent access problems. Most of the saga storage mechanisms in Wolverine happily support optimistic concurrency around saving saga state, so you could just use some selective retries on concurrency violations. Or better yet, Wolverine users can just about completely side step issues with concurrency by utilizing our newest improvement to partitioned messaging we’re calling “Global Partitioning.”

Let’s say that you have a great deal of operations in your system that have to modify a resource of some sort like an entity, a file, a saga in this case, or an event stream that might be a little bit sensitive to concurrent access. Let’s also say that you have a mix of messages that impact these sensitive resources that come from both external, upstream systems and from cascaded messages within your own system.

The syntax for this next feature was added just today in Wolverine 5.21 as I realized the previous syntax was basically unusable in the course of trying to write this blog post. So it goes.

A “global partitioning” allows you to create a guarantee that messages impacting those resources can be processed sequentially within a message group while allowing for parallel processing between message groups throughout the entire cluster.

Imagine it like this (but know I drew this diagram for someone using Kafka even though the next example is using Rabbit MQ queues):

And with this configuration:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        // You'd *also* supply credentials here of course!
        opts.UseRabbitMq();
        // Do something to add Saga storage too!
        opts
            .MessagePartitioning
            // This tells Wolverine to "just" use implied
            // message grouping based on Saga identity among other things
            .UseInferredMessageGrouping()
            .GlobalPartitioned(topology =>
            {
                // Creates 5 sharded RabbitMQ queues named "sequenced1" through "sequenced5"
                // with matching companion local queues for sequential processing
                topology.UseShardedRabbitQueues("sequenced", 5);
                topology.MessagesImplementing<MySequencedCommand>();
            });
    }).StartAsync();

What this does is spread the work out for handling MySequencedCommand messages through five different Rabbit MQ + Local queue pairs, with each pair active on only one single node within your application. Even inside each local queue in this partitioning scheme, Wolverine is parallelizing between message groups.

Now, let’s talk about receiving any message that can be cast to MySequencedCommand. If the message is received at a completely different listener than the “sequenced1/2/3/4/5” queues defined above, like from an external system that knows absolutely nothing about your message partitioning, Wolverine is going to immediately determine the message group identity by inferring that from the saga message handler rules (that’s what the UseInferredMessageGrouping() option does for us), then forwards that message to the proper node that is currently handling that group id. If the current node happens to be assigned that message group id, Wolverine forwards the message directly to the right local queue.

Likewise, if you publish a cascading message inside one of your handlers, Wolverine will determine the message group id for that message type, then try to either route that message locally if that group happens to be assigned to the current node (and it probably would be if you were cascading from your own handlers) or sends it remotely to the right messaging endpoint (Rabbit MQ queue or a Kafka topic or an AWS SQS queue maybe).

The point being, this guarantees that related messages are processed sequentially across the entire application cluster while allowing parallel processing between unrelated messages.

Summary

These are hopefully two powerful new features that will benefit Wolverine users in the near future. Both of these features were built at the behest of JasperFx Software clients to directly support their current work. I’m very happy to just quietly fold in reasonably sized new features for JasperFx support clients without extra cost when those features likely benefit the community as a whole. Contact us at sales@jasperfx.net to find out what we can do to help your software development efforts be more successful.

And just for bragging rights tonight, I did some poking around (okay, I asked Claude to do it for me) to see if any other asynchronous messaging tools offer anything similar to what our global partitioning option does for Wolverine users. While you can certainly achieve the same goals through actor frameworks like AkkaDotNet or Orleans (I consider actor frameworks to be such a different paradigm that I don’t really think of them as direct competitors to Wolverine), it doesn’t appear that there are any equivalents out there to this feature in the .NET space. MassTransit and NServiceBus both have more limited versions of this capability, but nothing that is as easy or flexible as what Wolverine has at this point. Now, granted, we’re at this point because Marten event stream appends can be sensitive to concurrent access so we’ve had to take concurrency maybe a little more seriously than the pure play asynchronous messaging tools that don’t really have an event sourcing component.

Wolverine — the event-driven messaging and HTTP framework for .NET — provides a rich, layered set of options for validating incoming data. Whether you are building HTTP endpoints or message handlers, Wolverine meets you where you are: from zero-configuration inline checks to full Fluent Validation or Data Annotation middleware support for both command handlers and HTTP endpoints.

Let’s maybe over simplify validation scenarios say they’ll fall into two buckets:

1. Run of the mill field level validation rules like required fields or value ranges. These rules are the bread and butter of dedicated validation frameworks like Fluent Validation or Microsoft’s Data Annotations markup.
2. Custom validation rules that are custom to your business domain and might involve checks against the existing state of your system beyond the command messages.

Let’s first look at Wolverine’s Data Annotation integration that is completely baked into the core WolverineFx Nuget. To get started, just opt into the Data Annotations middleware for message handlers like this:

using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        // Apply the validation middleware
        opts.UseDataAnnotationsValidation();
    }).StartAsync();

In message handlers, this middleware will kick in for any message type that has any validation attributes as this example:

public record CreateCustomer(
    // you can use the attributes on a record, but you need to
    // add the `property` modifier to the attribute
    [property: Required] string FirstName,
    [property: MinLength(5)] string LastName,
    [property: PostalCodeValidator] string PostalCode
) : IValidatableObject
{
    public IEnumerable<ValidationResult> Validate(ValidationContext validationContext)
    {
        // you can implement `IValidatableObject` for custom
        // validation logic
        yield break;
    }
};
public class PostalCodeValidatorAttribute : ValidationAttribute
{
    public override bool IsValid(object? value)
    {
        // custom attributes are supported
        return true;
    }
}
public static class CreateCustomerHandler
{
    public static void Handle(CreateCustomer customer)
    {
        // do whatever you'd do here, but this won't be called
        // at all if the DataAnnotations Validation rules fail
    }
}

By default for message handlers, any validation errors are logged, then the current execution is stopped through the usage of the HandlerContinuation value we’ll discuss later.

For Wolverine.HTTP integration with Data Annotations, use:

app.MapWolverineEndpoints(opts =>
{
    // Use Data Annotations that are built
    // into the Wolverine.HTTP library
    opts.UseDataAnnotationsValidationProblemDetailMiddleware();
});

Likewise, this middleware will only apply to HTTP endpoints that have a request input model that contains data annotation attributes. In this case though, Wolverine is using the ProblemDetails specification to report validation errors back to the caller with a status code of 400 by default.

Fluent Validation Middleware

Similarly, the Fluent Validation integration works more or less the same, but requires the WolverineFx.FluentValidation package for message handlers and the WolverineFx.Http.FluentValidation package for HTTP endpoints. There are some Wolverine helpers for discovering and registering FluentValidation validators in a way that applies some Wolverine-specific performance optimizations by trying to register most validators with a Singleton lifetime just to allow Wolverine to generate more optimized code.

It is possible to override how Wolverine handles validation failures, but I’d personally recommend just using the ProblemDetails default in most cases.

I would like to note that the way that Wolverine generates code for the Fluent Validation middleware is generally going to be more efficient at runtime than the typical IoC dependent equivalents you’ll frequently find in the MediatR space.

Explicit Validation

Let’s move on to validation rules that are more specific to your own problem domain, and especially the type of validation rules that would require you to examine the state of your system by exercising some kind of data access. These kinds of rules certainly can be done with custom Fluent Validation validators, but I strongly recommend you put that kind of validation directly into your message handlers or HTTP endpoints to colocate business logic together with the actual message handler or HTTP endpoint happy path.

One of the unique features of Wolverine in comparison to the typical “IHandler of T” application frameworks in .NET is Wolverine’s built in support for a type of low code ceremony Railway Programming, and this turns out to be perfect for one off validation rules.

In message handlers we’ve long had support for returning the HandlerContinuation enum from Validate() or Before() methods as a way to signal to Wolverine to conditionally stop all additional processing:

public static class ShipOrderHandler
{
    // This would be called first
    public static async Task<(HandlerContinuation, Order?, Customer?)> LoadAsync(ShipOrder command, IDocumentSession session)
    {
        var order = await session.LoadAsync<Order>(command.OrderId);
        if (order == null)
        {
            return (HandlerContinuation.Stop, null, null);
        }
        var customer = await session.LoadAsync<Customer>(command.CustomerId);
        return (HandlerContinuation.Continue, order, customer);
    }
    // The main method becomes the "happy path", which also helps simplify it
    public static IEnumerable<object> Handle(ShipOrder command, Order order, Customer customer)
    {
        // use the command data, plus the related Order & Customer data to
        // "decide" what action to take next
        yield return new MailOvernight(order.Id);
    }
}

But of course, with the example above, you could also write that with Wolverine’s declarative persistence like this:

public static class ShipOrderHandler
{
    // The main method becomes the "happy path", which also helps simplify it
    public static IEnumerable<object> Handle(
      ShipOrder command, 
      // This is loaded by the OrderId on the ShipOrder command
      [Entity(Required = true)]
      Order order, 
      // This is loaded by the CustomerId value on the ShipOrder command
      [Entity(Required = true)]
      Customer customer)
    {
        // use the command data, plus the related Order & Customer data to
        // "decide" what action to take next
        yield return new MailOvernight(order.Id);
    }
}

In the code above, Wolverine would stop the processing if either the Order or Customer entity referenced by the command message is missing. Similarly, if this code were in an HTTP endpoint instead, Wolverine would emit a ProblemDetails with a 400 status code and a message stating the data that is missing.

If you were using the code above with the integration with Marten or Polecat, Wolverine can even emit code that uses Marten or Polecat’s batch querying functionality to make your system more efficient by eliminating database round trips.

Likewise in the HTTP space, you could also return a ProblemDetails object directly from a Validate() method like:

public class ProblemDetailsUsageEndpoint
{
    public ProblemDetails Validate(NumberMessage message)
    {
        if (message.Number > 5)
            return new ProblemDetails
            {
                Detail = "Number is bigger than 5",
                Status = 400
            };
        // All good — continue!
        return WolverineContinue.NoProblems;
    }
    [WolverinePost("/problems")]
    public static string Post(NumberMessage message) => "Ok";
}

Even More Lightweight Validation!

When reviewing client code that uses the HandlerContinuation or ProblemDetails syntax, I definitely noticed the code can become verbose and noisy, especially compared to just embedding throw new InvalidOperationException("something is not right here"); code directly in the main methods — which isn’t something I’d like to see people tempted to do.

Instead, Wolverine 5.18 added a more lightweight approach that allows you to just return an array of strings from a Before/Validation() method:

    public static IEnumerable<string> Validate(SimpleValidateEnumerableMessage message)
    {
        if (message.Number > 10)
        {
            yield return "Number must be 10 or less";
        }
    }

    // or

    public static string[] Validate(SimpleValidateStringArrayMessage message)
    {
        if (message.Number > 10)
        {
            return ["Number must be 10 or less"];
        }

        return [];
    }

At runtime, Wolverine will stop a handler if there are any messages or emit a ProblemDetails response in HTTP endpoints.

Summary

Hopefully, Wolverine has you covered no matter what with options. A few practical takeaways:

• Reach for Validate() / ValidateAsync() first whenever IoC services or database queries are involved or the validation logic is just specific to your message handler or HTTP endpoint.
• Use Data Annotations middleware when your model types are already decorated with attributes and you want zero validator classes.
• Use Fluent Validation middleware when you want reusable, composable validators shared across multiple handlers or endpoints.

All three strategies generate efficient, ahead-of-time compiled middleware via Wolverine’s code generation engine, keeping the runtime overhead minimal regardless of which path you choose.

The Wolverine community fields a lot of questions from people who are moving to Wolverine from their previous MediatR usage. A quite natural response is to try to use Wolverine as a pure drop in replacement for MediatR and even try to use the existing MediatR idioms they’re already used to. However, Wolverine comes from a different philosophy than MediatR and most of the other “mediator” tools it’s inspired and using Wolverine with its idioms might lead to much simpler code or more efficient execution. Inspired by a conversation I had online today, let’s just into an example that I think shows quite a bit of contrast between the tools.

We’ve tried to lay out some of the differences between the tools in our Wolverine for MediatR Users guide, including the section this post is taken from.

Here’s an example of MediatR usage I borrowed from this blog post that shows the usage of MediatR within a shopping cart subsystem:

public class AddToCartRequest : IRequest<Result>
{
    public int ProductId { get; set; }
    public int Quantity { get; set; }
}
public class AddToCartHandler : IRequestHandler<AddToCartRequest, Result>
{
    private readonly ICartService _cartService;
    public AddToCartHandler(ICartService cartService)
    {
        _cartService = cartService;
    }
    public async Task<Result> Handle(AddToCartRequest request, CancellationToken cancellationToken)
    {
        // Logic to add the product to the cart using the cart service
        bool addToCartResult = await _cartService.AddToCart(request.ProductId, request.Quantity);
        bool isAddToCartSuccessful = addToCartResult; // Check if adding the product to the cart was successful.
        return Result.SuccessIf(isAddToCartSuccessful, "Failed to add the product to the cart."); // Return failure if adding to cart fails.
    }
}
    
public class CartController : ControllerBase
{
    private readonly IMediator _mediator;
    public CartController(IMediator mediator)
    {
        _mediator = mediator;
    }
    [HttpPost]
    public async Task<IActionResult> AddToCart([FromBody] AddToCartRequest request)
    {
        var result = await _mediator.Send(request);
        if (result.IsSuccess)
        {
            return Ok("Product added to the cart successfully.");
        }
        else
        {
            return BadRequest(result.ErrorMessage);
        }
    }
}

Note the usage of the custom Result<T> type from the message handler. Folks using MediatR love using these custom Result types when you’re passing information between logical layers because it avoids the usage of throwing exceptions and communicates failure cases more clearly.

See Andrew Lock on Working with the result pattern for more information about the Result pattern.

Wolverine is all about reducing code ceremony and we always strive to write application code as synchronous pure functions whenever possible, so let’s just write the exact same functionality as above using Wolverine idioms to shrink down the code:

public static class AddToCartRequestEndpoint
{
    // Remember, we can do validation in middleware, or
    // even do a custom Validate() : ProblemDetails method
    // to act as a filter so the main method is the happy path
    
    [WolverinePost("/api/cart/add"), EmptyResponse]
    public static Update<Cart> Post(
        AddToCartRequest request, 
        
        // This usage will return a 400 status code if the Cart
        // cannot be found
        [Entity(OnMissing = OnMissing.ProblemDetailsWith400)] Cart cart)
    {
        return cart.TryAddRequest(request) ? Storage.Update(cart) : Storage.Nothing(cart);
    }
}

There’s a lot going on above, so let’s dive into some of the details:

I used Wolverine.HTTP to write the HTTP endpoint so we only have one piece of code for our “vertical slice” instead of having both the Controller method and the matching message handler for the same logical command. Wolverine.HTTP embraces our Railway Programming model and direct support for the ProblemDetails specification as a means of stopping the HTTP request such that validation pre-conditions can be validated by middleware such that the main endpoint method is really the “happy path”.

The code above is using Wolverine’s “declarative data access” helpers you see in the [Entity] usage. We realized early on that a lot of message handlers or HTTP endpoints need to work on a single domain entity or a handful of entities loaded by identity values riding on either command messages, HTTP requests, or HTTP routes. At runtime, if the Cart isn’t found by loading it from your configured application persistence (which could be EF Core, Marten, or RavenDb at this time), the whole HTTP request would stop with status code 400 and a message communicated through ProblemDetails that the requested Cart cannot be found.

The key point I’m trying to prove is that idiomatic Wolverine results in potentially less repetitive code, less code ceremony, and less layering than MediatR idioms. Sure, it’s going to take a bit to get used to Wolverine idioms, but the potential payoff is code that’s easier to reason about and much easier to unit test — especially if you’ll buy into our A-Frame Architecture approach for organizing code within your slices.

Validation Middleware

As another example just to show how Wolverine’s runtime is different than MediatR’s, let’s consider the very common case of using Fluent Validation (or now DataAnnotations too!) middleware in front of message handlers or HTTP requests. With MediatR, you might use an IPipelineBehavior<T> implementation like this that will wrap all requests:

    public class ValidationBehaviour<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse> where TRequest : IRequest<TResponse>
    {
        private readonly IEnumerable<IValidator<TRequest>> _validators;
        public ValidationBehaviour(IEnumerable<IValidator<TRequest>> validators)
        {
            _validators = validators;
        }
      
        public async Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate<TResponse> next)
        {
            if (_validators.Any())
            {
                var context = new ValidationContext<TRequest>(request);
                var validationResults = await Task.WhenAll(_validators.Select(v => v.ValidateAsync(context, cancellationToken)));
                var failures = validationResults.SelectMany(r => r.Errors).Where(f => f != null).ToList();
                if (failures.Count != 0)
                    throw new ValidationException(failures);
            }
          
            return await next();
        }
    }

I’ve seen plenty of alternatives out there with slightly different implementations. In some cases folks will use service location to probe the application’s IoC container for any possible IValidator<T> implementations for the current request. In all cases though, the implementations are using runtime logic on every possible request to check if there is any validation logic. With the Wolverine version of Fluent Validation middleware, we do things a bit differently with less runtime overhead that will also result in cleaner Exception stack traces when things go wrong — don’t laugh, we really did design Wolverine quite purposely to avoid the really nasty kind of Exception stack traces you get from many other middleware or “behavior” using frameworks like Wolverine’s predecessor tool FubuMVC did 😦

Let’s say that you have a Wolverine.HTTP endpoint like so:

public record CreateCustomer
(
    string FirstName,
    string LastName,
    string PostalCode
)
{
    public class CreateCustomerValidator : AbstractValidator<CreateCustomer>
    {
        public CreateCustomerValidator()
        {
            RuleFor(x => x.FirstName).NotNull();
            RuleFor(x => x.LastName).NotNull();
            RuleFor(x => x.PostalCode).NotNull();
        }
    }
}
public static class CreateCustomerEndpoint
{
    [WolverinePost("/validate/customer")]
    public static string Post(CreateCustomer customer)
    {
        return "Got a new customer";
    }
    
    [WolverinePost("/validate/customer2")]
    public static string Post2([FromQuery] CreateCustomer customer)
    {
        return "Got a new customer";
    }
}

In the application bootstrapping, I’ve added this option:

app.MapWolverineEndpoints(opts =>
{
    // more configuration for HTTP...
    // Opting into the Fluent Validation middleware from
    // Wolverine.Http.FluentValidation
    opts.UseFluentValidationProblemDetailMiddleware();
}

Just like with MediatR, you would need to register the Fluent Validation validator types in your IoC container as part of application bootstrapping. Now, here’s how Wolverine’s model is very different from MediatR’s pipeline behaviors. While MediatR is applying that ValidationBehaviour to each and every message handler in your application whether or not that message type actually has any registered validators, Wolverine is able to peek into the IoC configuration and “know” whether there are registered validators for any given message type. If there are any registered validators, Wolverine will utilize them in the code it generates to execute the HTTP endpoint method shown above for creating a customer. If there is only one validator, and that validator is registered as a Singleton scope in the IoC container, Wolverine generates this code:

    public class POST_validate_customer : Wolverine.Http.HttpHandler
    {
        private readonly Wolverine.Http.WolverineHttpOptions _wolverineHttpOptions;
        private readonly Wolverine.Http.FluentValidation.IProblemDetailSource<WolverineWebApi.Validation.CreateCustomer> _problemDetailSource;
        private readonly FluentValidation.IValidator<WolverineWebApi.Validation.CreateCustomer> _validator;

        public POST_validate_customer(Wolverine.Http.WolverineHttpOptions wolverineHttpOptions, Wolverine.Http.FluentValidation.IProblemDetailSource<WolverineWebApi.Validation.CreateCustomer> problemDetailSource, FluentValidation.IValidator<WolverineWebApi.Validation.CreateCustomer> validator) : base(wolverineHttpOptions)
        {
            _wolverineHttpOptions = wolverineHttpOptions;
            _problemDetailSource = problemDetailSource;
            _validator = validator;
        }



        public override async System.Threading.Tasks.Task Handle(Microsoft.AspNetCore.Http.HttpContext httpContext)
        {
            // Reading the request body via JSON deserialization
            var (customer, jsonContinue) = await ReadJsonAsync<WolverineWebApi.Validation.CreateCustomer>(httpContext);
            if (jsonContinue == Wolverine.HandlerContinuation.Stop) return;
            
            // Execute FluentValidation validators
            var result1 = await Wolverine.Http.FluentValidation.Internals.FluentValidationHttpExecutor.ExecuteOne<WolverineWebApi.Validation.CreateCustomer>(_validator, _problemDetailSource, customer).ConfigureAwait(false);

            // Evaluate whether or not the execution should be stopped based on the IResult value
            if (result1 != null && !(result1 is Wolverine.Http.WolverineContinue))
            {
                await result1.ExecuteAsync(httpContext).ConfigureAwait(false);
                return;
            }


            
            // The actual HTTP request handler execution
            var result_of_Post = WolverineWebApi.Validation.ValidatedEndpoint.Post(customer);

            await WriteString(httpContext, result_of_Post);
        }

    }

I should note that Wolverine’s Fluent Validation middleware will not generate any code for any HTTP endpoint where there are no known Fluent Validation validators for the endpoint’s request model. Moreover, Wolverine can even generate slightly different code for having multiple validators versus a singular validator as a way of wringing out a little more efficiency in the common case of having only a single validator registered for the request type.

The point here is that Wolverine is trying to generate the most efficient code possible based on what it can glean from the IoC container registrations and the signature of the HTTP endpoint or message handler methods while the MediatR model has to effectively use runtime wrappers and conditional logic at runtime.