Linked Lists in Real Life

I’ve been occasionally writing posts about old design patterns or techniques that are still occasionally useful despite the decades long backlash to the old “Gang of Four” book:

A linked list structure is a simple data structure where each element has a reference to the next element. At its simplest, it’s no more than this single linked list implementation:

public abstract class Item
{
    public Item Next { get; set; }
}

Now, I got into software development through Shadow IT rather than a formal Computer Science degree, so I can’t really get into what Big O notation stuff a linked list gives you for sorting or finding or whatnot (and don’t really care either). What I can tell you that I have occasionally used linked list structures to great effect, with at least two samples within the greater “Critter Stack” / JasperFx codebases.

For the first example, consider the complex SQL generation from this LINQ query in Marten using Marten’s custom “Include Related Documents” feature:

        // theSession is an IDocumentSession from Marten
        var holders = await theSession.Query<TargetHolder>()
            .Include<Target>(x => x.TargetId, list, t => t.Color == Colors.Green)
            .ToListAsync();

Behind the scenes, Marten is generating this pile of lovely SQL:

drop table if exists mt_temp_id_list1; 
create temp table mt_temp_id_list1 as (select d.id, d.data from public.mt_doc_targetholder as d);
 select d.id, d.data from public.mt_doc_target as d where (d.id in (select CAST(d.data ->> 'TargetId' as uuid) from mt_temp_id_list1 as d) and CAST(d.data ->> 'Color' as integer) = $1);
  $1: 2

If you squint really hard, you can notice that Marten is actually executing four different SQL statements in one logical query. Internally, Marten is using a linked list structure to “plan” and then generate the SQL from the raw LINQ Expression tree using the Statement type, partially shown below:

public abstract partial class Statement: ISqlFragment
{
    public Statement Next { get; set; }
    public Statement Previous { get; set; }

    public StatementMode Mode { get; set; } = StatementMode.Select;

    /// <summary>
    ///     For common table expressions
    /// </summary>
    public string ExportName { get; protected internal set; }

    public bool SingleValue { get; set; }
    public bool ReturnDefaultWhenEmpty { get; set; }
    public bool CanBeMultiples { get; set; }

    public void Apply(ICommandBuilder builder)
    {
        configure(builder);
        if (Next != null)
        {
            if (Mode == StatementMode.Select)
            {
                builder.StartNewCommand();
            }

            builder.Append(" ");
            Next.Apply(builder);
        }
    }

    public void InsertAfter(Statement descendent)
    {
        if (Next != null)
        {
            Next.Previous = descendent;
            descendent.Next = Next;
        }

        if (object.ReferenceEquals(this, descendent))
        {
            throw new InvalidOperationException(
                "Whoa pardner, you cannot set Next to yourself, that's a stack overflow!");
        }

        Next = descendent;
        descendent.Previous = this;
    }

    /// <summary>
    ///     Place the descendent at the very end
    /// </summary>
    /// <param name="descendent"></param>
    public void AddToEnd(Statement descendent)
    {
        if (Next != null)
        {
            Next.AddToEnd(descendent);
        }
        else
        {
            if (object.ReferenceEquals(this, descendent))
            {
                return;
            }

            Next = descendent;
            descendent.Previous = this;
        }
    }

    public void InsertBefore(Statement antecedent)
    {
        if (Previous != null)
        {
            Previous.Next = antecedent;
            antecedent.Previous = Previous;
        }

        antecedent.Next = this;
        Previous = antecedent;
    }

    public Statement Top()
    {
        return Previous == null ? this : Previous.Top();
    }

    // Find the selector statement at the very end of 
    // the linked list
    public SelectorStatement SelectorStatement()
    {
        return (Next == null ? this : Next.SelectorStatement()).As<SelectorStatement>();
    }

    // And a some other stuff...
}

The Statement model is a “double linked list,” meaning that each element is aware of both its direct ancestor (Previous) and the next descendent (Descendent). With the Statement model, I’d like to call out a couple wrinkles that made the linked list strategy a great fit for complex SQL generation.

First off, the SQL generation itself frequently requires multiple statements from the top level statement down to the very last statement, and you can see that happening in the Apply() method above that writes out the SQL for the current Statement, then calls Next.Apply() all the way down to the end of the chain — all while helping Marten’s (really Weasel’s) batch SQL generation “know” when it should start a new command (see NpgsqlBatch for a little more context on what I mean there).

Also note all the methods for inserting a new Statement directly before or after the current Statement. Linked lists are perfect for when you frequently need to insert new elements before or after or even at the very end of the chain rather than at a known index. Here’s an example from Marten that kicks in sometimes when a user uses the LINQ Distinct() operator:

For more context if you’ve lived a more charmed life and have never run across them, here’s an explanation of Common Table Expressions from PostgreSQL (this is commonly supported in other databases).

internal class DistinctSelectionStatement: Statement
{
    public DistinctSelectionStatement(SelectorStatement parent, ICountClause selectClause, IMartenSession session)
    {
        parent.ConvertToCommonTableExpression(session);

        ConvertToCommonTableExpression(session);

        parent.InsertAfter(this);

        selectClause.OverrideFromObject(this);
        var selector = new SelectorStatement { SelectClause = selectClause };
        InsertAfter(selector);
    }

    protected override void configure(ICommandBuilder sql)
    {
        startCommonTableExpression(sql);
        sql.Append("select distinct(data) from ");
        sql.Append(Previous.ExportName);
        sql.Append(" as d");
        endCommonTableExpression(sql);
    }
}

When Marten has to apply this DistinctSelectionStatement, it modifies its immediate parent statement that probably started out life as a simple select some_field from some_table query into a common table expression query, and appends the DistinctSelectionStatement behind its parent to do the actual SQL distinct mechanics.

In this particular case of the SQL generation, it’s been frequently necessary for a Statement to “know” about its immediate ancestor, as you can see in the sample code above where the DistinctSelectStatement picks off the ExportName (the CTE name of the preceding statement) to use in generating the right SQL in DistinctSelectStatement.Apply().

For another example, both Marten and Wolverine use a runtime code generation model to build a lot of their “glue” code between the framework and user’s application code (you can see an example of that runtime code generation in this post). One of the core conceptual abstractions in the shared code generation model is the abstract Frame class which roughly equates to logical step in the generated code — usually just a single line of code. During the code generation process, the Frame objects are assembled in a single linked list structure so each Frame.

When actually writing out the generated source code, a typical Frame will write its code, then tell the next Frame to write out its code and so on — as shown by this sample class:

// This is from Wolverine, and weaves in code
// to add selected tags from incoming messages to
// message handlers into the current Open Telemetry Activity
public class AuditToActivityFrame : SyncFrame
{
    private readonly Type _inputType;
    private readonly List<AuditedMember> _members;
    private Variable? _input;

    public AuditToActivityFrame(IChain chain)
    {
        _inputType = chain.InputType()!;
        _members = chain.AuditedMembers;
    }

    public override IEnumerable<Variable> FindVariables(IMethodVariables chain)
    {
        _input = chain.FindVariable(_inputType);
        yield return _input;
    }

    public override void GenerateCode(GeneratedMethod method, ISourceWriter writer)
    {
        writer.WriteComment("Application-specific Open Telemetry auditing");
        foreach (var member in _members)
            writer.WriteLine(
                $"{typeof(Activity).FullNameInCode()}.{nameof(Activity.Current)}?.{nameof(Activity.SetTag)}(\"{member.OpenTelemetryName}\", {_input!.Usage}.{member.Member.Name});");

        // Tell the next frame to write its code too!
        Next?.GenerateCode(method, writer);
    }
}

Where the linked list structure really comes into play with the source generation is when you need to wrap the inner Next Frame with some kind of coding construct like a using block or a try/finally block maybe. Here’s an example of doing just that where the following CatchStreamCollisionFrame places a try/catch block around the code generated by the Next frame (and all of the other frames after the Next frame as well):

// This is the actual middleware that's injecting some code
// into the runtime code generation
internal class CatchStreamCollisionFrame : AsyncFrame
{
    public override void GenerateCode(GeneratedMethod method, ISourceWriter writer)
    {
        writer.WriteComment("Catches any existing stream id collision exceptions");
        writer.Write("BLOCK:try");
        
        // Write the inner code here
        Next?.GenerateCode(method, writer);
        
        writer.FinishBlock();
        writer.Write($@"
BLOCK:catch({typeof(ExistingStreamIdCollisionException).FullNameInCode()} e)
await {typeof(StreamCollisionExceptionPolicy).FullNameInCode()}.{nameof(StreamCollisionExceptionPolicy.RespondWithProblemDetails)}(e, httpContext);
return;
END

");
    }

The ugly generated code above will catch a Marten exception named ExistingStreamIdCollisionException in the generated code for a Wolverine.HTTP endpoint and return a ProblemDetails result explaining the problem and an HTTP status code of 400 (Invalid) instead of letting the exception bubble up.

By having the linked list structure where each Frame is aware of the next Frame, it makes it relatively easy to generate code when you need to wrap the inner code in some kind of C# block structure.

Summary

I spent a lot of time as a kid helping my Dad on his construction crew and helping my grandfather on his farm (you can’t possibly imagine how often farming equipment breaks). Both of them obviously had pretty large toolboxes — but there’s some tools that don’t come out very often, but man you were glad you had them when you did need them. The average developer probably isn’t going to use linked lists by hand very often, but I’ve found them to be very helpful when you need to either model a problem as outer/inner handlers or ancestor/descendents. Linked lists are also great when you need to be able to easily insert items into the greater collection relative to another item.

Anyway, dunno if these examples are too involved or too specious, but there the only two times I’ve used a linked list in the past decade.

I hope it’s obvious, but the JasperFx Software logo is meant to be a tractor tire around a cattle brand. The company name and logo is a little tribute to my family heritage, such as it is:)

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.

Marten, PostgreSQL, and .NET Aspire walk into a bar…

This is somewhat a follow up from yesterday’s post on Marten, Metrics, and Open Telemetry Support. I was very hopeful about the defunct Project Tye, and have been curious about .NET Aspire as a more ambitious offering since it was announced. As part of the massive Marten V7 release, we took some steps to ensure that Marten could use PostgreSQL databases controlled by .NET Aspire.

I finally got a chance to put together a sample Marten system named using .NET Aspire called MartenWithProjectAspire on GitHub. Simplified from some longstanding Marten test projects, consider this little system:

At runtime, the EventPublisher service continuously appends events that represent progress in a Trip event stream to the Marten-ized PostgreSQL database. The TripBuildingService is running Marten’s async daemon subsystem that constantly reads in new events to the PostgreSQL database and builds or updates projected documents back to the database to represent the current state of the event store.

The end result was a single project named AspireHost that when executed, will use .NET Aspire to start a new PostgreSQL docker container and start up the EventPublisher and TripBuildingService services while passing the connection string to the new PostgreSQL database to these services at runtime with a little bit of Environment variable sleight of hand.

You can see the various projects and containers from the Aspire dashboard:

and even see some of the Open Telemetry activity traced by Marten and visualized through Aspire:

Honestly, it took me a bit of trial and error to get this all working together. First, we need to configure Marten to use an NpgsqlDataSource connection to the PostgreSQL database that will be loaded from each service’s IoC container — then tell Marten to use that NpgsqlDataSource.

After adding Nuget references for Aspire.Npgsql and Marten itself, I added the second line of code shown below to the top of the Program file for both services using Marten:

var builder = Host.CreateApplicationBuilder();

// Register the NpgsqlDataSource in the IoC container using
// connection string named "marten" from IConfiguration
builder.AddNpgsqlDataSource("marten");

That’s really just a hook to add a registration for the NpgsqlDataSource type to the application’s IoC container with the expectation that the connection string will be in the application’s configuration connection string collection with the key “marten.”

One of the major efforts with Marten 7 was rewiring Marten’s internals (then Wolverine’s) to strictly use the new NpgsqlDataSource concept for database connectivity. If you maybe caught me being less than polite about Npgsql on what’s left of Twitter, just know that the process was very painful but it’s completely done now and working well outside of the absurd noisiness of built in Npgsql logging.

Next, I have to explicitly tell Marten itself to load the NpgsqlDataSource object from the application’s IoC container instead of the older, idiomatic approach of passing a connection string directly to Marten as shown below:

builder.Services.AddMarten(opts =>
    {
        // Other configuration, but *not* the connection
    })
    
    // Use PostgreSQL data source from the IoC container
    .UseNpgsqlDataSource();

Now, switching to the AspireHost, I needed to add a Nuget reference to Aspire.Hosting.PostgreSQL in order to be able to bootstrap the PostgreSQL database at runtime. I also made project references from AspireHost to EventPublisher and TripBuildingService — which is important because Aspire does some source generation build a strong typed enumeration representing your projects that we’ll use next. That last step confused me when I was first playing with Aspire, so hopefully now you get to bypass that confusion. Maybe.

In the Program file for AspireHost, it’s just this:

var builder = DistributedApplication.CreateBuilder(args);

var postgresdb = builder.AddPostgres("marten");

builder.AddProject<Projects.EventPublisher>("publisher")
    .WithReference(postgresdb);

builder.AddProject<Projects.TripBuildingService>("trip-building")
    .WithReference(postgresdb);

builder.Build().Run();

Now, run the AspireHost project and you are able to run the two other services with the newly activated PostgreSQL Docker container, which you can see from the Docker Desktop dashboard:

Ship it!

Summary

Is .NET Aspire actually useful (I think so, even if it’s just for local development and testing maybe)? Can I explain the new Open Telemetry data exported from Marten? Would I use this instead of a dirt simple Docker Compose file like I do today (probably not to be honest)? Is this all fake?

All these questions and more will be somewhat addressed tomorrow-ish when I try to launch a new YouTube channel for JasperFx Software using the sample from this blog post as the subject for my first ever solo YouTube video.

One more thing…

I did alter the launchSettings.json file of the Aspire host project so it didn’t need to care about HTTPS to this:

{
  "$schema": "https://json.schemastore.org/launchsettings.json",
  "profiles": {
    "http": {
      "commandName": "Project",
      "dotnetRunMessages": true,
      "launchBrowser": true,
      "applicationUrl": "http://localhost:15242",
      "environmentVariables": {
        "ASPNETCORE_ENVIRONMENT": "Development",
        "DOTNET_ENVIRONMENT": "Development",
        "DOTNET_DASHBOARD_OTLP_ENDPOINT_URL": "http://localhost:19076",
        "DOTNET_RESOURCE_SERVICE_ENDPOINT_URL": "http://localhost:20101",
        "ASPIRE_ALLOW_UNSECURED_TRANSPORT": "true"
      }
    }
  }
}

Note the usage of the ASPIRE_ALLOW_UNSECURED_TRANSPORT environment variable.

Marten, Metrics, and Open Telemetry Support

Marten 7.10 was released today, and (finally) brings some built in support for monitoring Marten performance by exporting Open Telemetry and Metrics about Marten activity and performance within your system.

To use a little example, there’s a sample application in the Marten codebase called EventPublisher that we use to manually test some of the command line tooling. All that EventPublisher does is to continuously publish randomized events to a Marten event store when it runs. That made it a good place to start with a test harness for our new Open Telemetry support and performance related metrics.

For testing, I was just using the Project Aspire dashboard for viewing metrics and Open Telemetry tracing. First off, I enabled the “opt in” connection tracing for Marten, and put it into a verbose mode that’s probably only suitable for debugging or performance optimization work:

        // builder is a HostApplicationBuilder object
        builder.Services.AddMarten(opts =>
        {
            // Other Marten configuration...
            
            // Turn on Otel tracing for connection activity, and
            // also tag events to each span for all the Marten "write"
            // operations
            opts.OpenTelemetry.TrackConnections = TrackLevel.Verbose;

            // This opts into exporting a counter just on the number
            // of events being appended. Kinda a duplication
            opts.OpenTelemetry.TrackEventCounters();
            );
        });

That’s just the Marten side of things, so to hook up an Open Telemetry exporter for the Aspire dashboard, I added (really copy/pasted) this code (note that you’ll need the OpenTelemetry.Extensions.Hosting and OpenTelemetry.Exporter.OpenTelemetryProtocol Nugets added to your project):

        builder.Logging.AddOpenTelemetry(logging =>
        {
            logging.IncludeFormattedMessage = true;
            logging.IncludeScopes = true;
        });

        builder.Services.AddOpenTelemetry()
            .WithMetrics(metrics =>
            {
                metrics
                    .AddRuntimeInstrumentation().AddMeter("EventPublisher");
            })
            .WithTracing(tracing =>
            {
                tracing.AddAspNetCoreInstrumentation()
                    .AddHttpClientInstrumentation();
            });

        var endpointUri = builder.Configuration["OTEL_EXPORTER_OTLP_ENDPOINT"];
        builder.Services.AddOpenTelemetry().UseOtlpExporter();

        builder.Services.AddOpenTelemetry()
            // Enable exports of Open Telemetry activities
            .WithTracing(tracing =>
            {
                tracing.AddSource("Marten");
            })
            
            // Enable exports of metrics
            .WithMetrics(metrics =>
            {
                metrics.AddMeter("Marten");
            });

And now after running that with Aspire, you can see the output:

By itself, these spans, especially when shown in context of being nested within an HTTP request or a message being handled in a service bus framework, can point out where you may have performance issues from chattiness between the application server and the database — which I have found to be a very common source of performance problems out in the real world.

This is an opt in mode, but there are metrics and Open Telemetry tracing that are automatic for the background, async daemon subsystem. Skipping ahead a little bit, here’s a preview of some performance metrics in a related application that shows the “health” of a projection running in Marten’s async daemon subsystem by visualizing the “gap” between the projection’s current progression and the “high water mark” of Marten’s event store (how far the projection is sequentially compared to how far the whole event store itself is):

Summary

This is a short blog post, but I hope even this is enough to demonstrate how useful this new tracing is going to be in this new world order of Open Telemetry tracing tools.

The Decorator Pattern is sometimes helpful

I’ve been occasionally writing posts about old design patterns that are still occasionally useful despite the decades long backlash to the old “Gang of Four” book:

According to the original Gang of Four book, the “Decorator Pattern”:

…dynamically adds/overrides behavior in an existing method of an object.

Or more concretely, a decorator is a wrapper for an inner object that intercepts all calls to the inner object and potentially does some kind of work before or after the inner call. As a simple example, consider this ancient interface from the testing suite in StructureMap & Lamar:

    public interface IWidget
    {
        void DoSomething();
    }

And here’s a potential decorator for the IWidget service:

    public class ConsoleWritingWidgetDecorator : IWidget
    {
        private readonly IWidget _inner;

        public ConsoleWritingWidgetDecorator(IWidget inner)
        {
            _inner = inner;
        }

        public void DoSomething()
        {
            Console.WriteLine("I'm about to do something!");
            _inner.DoSomething();
            Console.WriteLine("I did something!");
        }
    }

The mechanics are simple enough, so let’s dive into some more complex use cases from the Marten 7.* codebase.

The most common usage of a decorator for me has been to separate out some kind of infrastructural concern like logging, error handling, or security from the core behavior. Just think on this. Instrumentation, security, and error handling are all very important elements of successful production code, but how many times in your career have you struggled to comprehend, modify, or debug code that is almost completely obfuscated by technical concerns.

Instead, I’ve sometimes found it helpful to separate out some of the technical concerns to a wrapping decorator just to allow the core functionality code to be easier to write, read later, and definitely to test. As an example from Marten 7.*, we have this interface for a service within Marten’s async daemon subsystem that’s used to fetch a page of event data at a time for a given projection or subscription:

public class EventRequest
{
    public long Floor { get; init; }
    public long HighWater { get; init; }
    public int BatchSize { get; init; }

    public ShardName Name { get; init; }

    public ErrorHandlingOptions ErrorOptions { get; init; }

    // other stuff...
}

public interface IEventLoader
{
    Task<EventPage> LoadAsync(EventRequest request, CancellationToken token);
}

This is for an asynchronous, background process, and we fully expect for there to be occasional issues with database connectivity, network hiccups, command timeouts when the database is stressed, and who knows what else. It’s obviously very important for this code to be as resilient as possible and be able to do some selected retries on transient errors at runtime. At the same time though, the actual functionality of that one LoadAsync() method was busy enough all by itself, so I opted to write the “real” functionality first with this — then test the heck out of that first:

internal class EventLoader: IEventLoader
{
    private readonly int _aggregateIndex;
    private readonly int _batchSize;
    private readonly NpgsqlParameter _ceiling;
    private readonly NpgsqlCommand _command;
    private readonly NpgsqlParameter _floor;
    private readonly IEventStorage _storage;
    private readonly IDocumentStore _store;

    public EventLoader(DocumentStore store, MartenDatabase database, AsyncProjectionShard shard, AsyncOptions options) : this(store, database, options, shard.BuildFilters(store).ToArray())
    {

    }

    public EventLoader(DocumentStore store, MartenDatabase database, AsyncOptions options, ISqlFragment[] filters)
    {
        _store = store;
        Database = database;

        _storage = (IEventStorage)store.Options.Providers.StorageFor<IEvent>().QueryOnly;
        _batchSize = options.BatchSize;

        var schemaName = store.Options.Events.DatabaseSchemaName;

        var builder = new CommandBuilder();
        builder.Append($"select {_storage.SelectFields().Select(x => "d." + x).Join(", ")}, s.type as stream_type");
        builder.Append(
            $" from {schemaName}.mt_events as d inner join {schemaName}.mt_streams as s on d.stream_id = s.id");

        if (_store.Options.Events.TenancyStyle == TenancyStyle.Conjoined)
        {
            builder.Append(" and d.tenant_id = s.tenant_id");
        }

        var parameters = builder.AppendWithParameters(" where d.seq_id > ? and d.seq_id <= ?");
        _floor = parameters[0];
        _ceiling = parameters[1];
        _floor.NpgsqlDbType = _ceiling.NpgsqlDbType = NpgsqlDbType.Bigint;

        foreach (var filter in filters)
        {
            builder.Append(" and ");
            filter.Apply(builder);
        }

        builder.Append(" order by d.seq_id limit ");
        builder.Append(_batchSize);

        _command = builder.Compile();
        _aggregateIndex = _storage.SelectFields().Length;
    }

    public IMartenDatabase Database { get; }

    public async Task<EventPage> LoadAsync(EventRequest request,
        CancellationToken token)
    {
        // There's an assumption here that this method is only called sequentially
        // and never at the same time on the same instance
        var page = new EventPage(request.Floor);

        await using var session = (QuerySession)_store.QuerySession(SessionOptions.ForDatabase(Database));
        _floor.Value = request.Floor;
        _ceiling.Value = request.HighWater;

        await using var reader = await session.ExecuteReaderAsync(_command, token).ConfigureAwait(false);
        while (await reader.ReadAsync(token).ConfigureAwait(false))
        {
            try
            {
                // as a decorator
                var @event = await _storage.ResolveAsync(reader, token).ConfigureAwait(false);

                if (!await reader.IsDBNullAsync(_aggregateIndex, token).ConfigureAwait(false))
                {
                    @event.AggregateTypeName =
                        await reader.GetFieldValueAsync<string>(_aggregateIndex, token).ConfigureAwait(false);
                }

                page.Add(@event);
            }
            catch (UnknownEventTypeException e)
            {
                if (request.ErrorOptions.SkipUnknownEvents)
                {
                    request.Runtime.Logger.LogWarning("Skipping unknown event type '{EventTypeName}'", e.EventTypeName);
                }
                else
                {
                    // Let any other exception throw
                    throw;
                }
            }
            catch (EventDeserializationFailureException e)
            {
                if (request.ErrorOptions.SkipSerializationErrors)
                {
                    await request.Runtime.RecordDeadLetterEventAsync(e.ToDeadLetterEvent(request.Name)).ConfigureAwait(false);
                }
                else
                {
                    // Let any other exception throw
                    throw;
                }
            }
        }

        page.CalculateCeiling(_batchSize, request.HighWater);

        return page;
    }

At runtime, that type is wrapped by a decorator that adds resiliency through the Polly library like so:

internal class ResilientEventLoader: IEventLoader
{
    private readonly ResiliencePipeline _pipeline;
    private readonly EventLoader _inner;

    internal record EventLoadExecution(EventRequest Request, IEventLoader Loader)
    {
        public async ValueTask<EventPage> ExecuteAsync(CancellationToken token)
        {
            var results = await Loader.LoadAsync(Request, token).ConfigureAwait(false);
            return results;
        }
    }

    public ResilientEventLoader(ResiliencePipeline pipeline, EventLoader inner)
    {
        _pipeline = pipeline;
        _inner = inner;
    }

    public Task<EventPage> LoadAsync(EventRequest request, CancellationToken token)
    {
        try
        {
            var execution = new EventLoadExecution(request, _inner);
            return _pipeline.ExecuteAsync(static (x, t) => x.ExecuteAsync(t),
                execution, token).AsTask();
        }
        catch (Exception e)
        {
            // This would only happen after a chain of repeated
            // failures -- which can of course happen!
            throw new EventLoaderException(request.Name, _inner.Database, e);
        }
    }
}

In the case above, using a decorator allowed me to focus on one set of concerns at a time and punt the Polly usage for resiliency to something else. The “something else” being a decorator that only really deals with the error handling and resiliency while letting the inner IEventFetcher “know” how to fetch the requested event data and turn that into the right .NET objects.

Here’s a more recent example written by Sean Farrow where we’re purposely using a decorator to add extra functionality to a core bit of the Marten command execution. If you go spelunking around in the Marten codebase, you’ll fine an interface called IConnectionLifetime that is used to actually execute database commands or queries within most common Marten operations (it was actually featured in my post on the State pattern) partially shown below:

public interface IConnectionLifetime: IAsyncDisposable, IDisposable
{
    // Other stuff...

    Task<DbDataReader> ExecuteReaderAsync(NpgsqlCommand command,
        CancellationToken token = default);
}

As we’re adding Open Telemetry support into Marten for the 7.10 release, we know that some folks will want some control to turn up or down the telemetry data emitted by Marten (more can be noise, and sometimes less can mean better performance anyway). One possible data collection element is to track the number of database requests in a given session and the number of subsequent database exceptions. That’s being accomplished with a decorator around the IConnectionLifetime like this:

internal class EventTracingConnectionLifetime:
    IConnectionLifetime
{
    private const string MartenCommandExecutionStarted = "marten.command.execution.started";
    private const string MartenBatchExecutionStarted = "marten.batch.execution.started";
    private const string MartenBatchPagesExecutionStarted = "marten.batch.pages.execution.started";
    private readonly IConnectionLifetime _innerConnectionLifetime;
    private readonly Activity? _databaseActivity;

    public EventTracingConnectionLifetime(IConnectionLifetime innerConnectionLifetime, string tenantId)
    {
        if (innerConnectionLifetime == null)
        {
            throw new ArgumentNullException(nameof(innerConnectionLifetime));
        }

        if (string.IsNullOrWhiteSpace(tenantId))
        {
            throw new ArgumentException("The tenant id cannot be null, an empty string or whitespace.", nameof(tenantId));
        }

        Logger = innerConnectionLifetime.Logger;
        CommandTimeout = innerConnectionLifetime.CommandTimeout;
        _innerConnectionLifetime = innerConnectionLifetime;

        var currentActivity = Activity.Current ?? null;
        var tags = new ActivityTagsCollection(new[] { new KeyValuePair<string, object?>(MartenTracing.MartenTenantId, tenantId) });
        _databaseActivity = MartenTracing.StartConnectionActivity(currentActivity, tags);
    }

    public ValueTask DisposeAsync()
    {
        _databaseActivity?.Stop();
        return _innerConnectionLifetime.DisposeAsync();
    }

    public void Dispose()
    {
        _databaseActivity?.Stop();
        _innerConnectionLifetime.Dispose();
    }

    public IMartenSessionLogger Logger { get; set; }
    public int CommandTimeout { get; }
    public int Execute(NpgsqlCommand cmd)
    {
        _databaseActivity?.AddEvent(new ActivityEvent(MartenCommandExecutionStarted));

        try
        {
            return _innerConnectionLifetime.Execute(cmd);
        }
        catch (Exception e)
        {
            _databaseActivity?.RecordException(e);

            throw;
        }
    }

    public async Task<DbDataReader> ExecuteReaderAsync(NpgsqlCommand command, CancellationToken token = default)
    {
        _databaseActivity?.AddEvent(new ActivityEvent(MartenCommandExecutionStarted));

        try
        {
            return await _innerConnectionLifetime.ExecuteReaderAsync(command, token).ConfigureAwait(false);
        }
        catch (Exception e)
        {
            _databaseActivity?.RecordException(e);

            throw;
        }
    }

    // And much more...
}

That decorator is only selectively applied depending on whether or not the system developers have opted into this tracing and also if there’s an active listener for the data (no sense wasting extra CPU time on emitting data into the void!):

    internal IConnectionLifetime Initialize(DocumentStore store, CommandRunnerMode mode)
    {
        Mode = mode;
        Tenant ??= TenantId != Tenancy.DefaultTenantId ? store.Tenancy.GetTenant(TenantId) : store.Tenancy.Default;

        if (!AllowAnyTenant && !store.Options.Advanced.DefaultTenantUsageEnabled &&
            Tenant.TenantId == Tenancy.DefaultTenantId)
        {
            throw new DefaultTenantUsageDisabledException();
        }

        var innerConnectionLifetime = GetInnerConnectionLifetime(store, mode);

        return !OpenTelemetryOptions.TrackConnectionEvents || !MartenTracing.ActivitySource.HasListeners()
            ? innerConnectionLifetime
            : new EventTracingConnectionLifetime(innerConnectionLifetime, Tenant.TenantId);
    }

Summary

I showed off a couple examples where I feel like the decorator pattern is adding value to the Marten code by helping us expose extra functionality or just to separate concerns a little more cleanly in these particular cases. I’ve absolutely seen codebases where the code was dreadfully hard to follow because of the copious usage of decorators. Using decorators can also help blow up your object allocations (potential performance issue) and lead to some extraordinarily noisy exception stack traces from failures in the inner most objects. That being said, I’d still rather deal with nested decorators where you can at least see the boundaries between object responsibilities than wrestle with deep inheritance relationships.

As with all patterns, the decorator pattern is sometimes helpful and sometimes harmful. Just be cautious with its usage on a case by case basis and always filter it through the lens of “is using this making the code easier to understand or harder?”

But regardless, decorators are commonly used, and it’s just good to recognize the pattern when you see it and understand what the original author was trying to do.

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.

Embedding Database Migrations with Weasel

A woodworking weasel building a table, of course!

Let’s say that you’re building a system that needs to directly work with a handful of database tables. Or maybe more aptly, you’re building a redistributable class library that will expect to interact with a small number of database tables, functions, view, or sequences — and you’d love to make that class library be responsible for building those database objects as necessary at least at development time so your users can just get to work without any kind of laborious setup before hand.

If you’ve worked with the main “Critter Stack” tools (Marten and Wolverine), you’re familiar with how they can quietly set up your development or even your production database as necessary to reflect your system’s configuration. The actual work of database migrations built into these tools is done by the third member of the “Critter Stack, ” a helper library named Weasel.

You can also use Weasel in your own class library to do the same kind of automatic database migration — as long as you’re using either PostgreSQL or Sql Server (for now).

With Weasel, you can define the requirements for a new database table with a class originally named Table in the Weasel.Postgresql Nuget which exposes an API for just about anything you could do to configure a table including columns, primary keys, foreign key relationships to other tables, and indexes:

        var table = new Table("tables.people");
        table.AddColumn<int>("id").AsPrimaryKey();
        table.AddColumn<string>("first_name");
        table.AddColumn<string>("last_name");

Inside your code, you can at any time migrate the existing database to reflect your Table object with this convenience extension method added in Weasel 7.4:

var table = new Table("tables.people");
table.AddColumn<int>("id").AsPrimaryKey();
table.AddColumn<string>("first_name");
table.AddColumn<string>("last_name");

await using var conn = new NpgsqlConnection("some connection string");
await conn.OpenAsync();

// This will apply any necessary changes to make
// the database reflect the specified table structure
await table.MigrateAsync(conn);

Behind the scenes, Weasel reaches into the database to find the current status — if any — of the specified table. If the table doesn’t exist, Weasel creates it based on the in memory specification. If the table does already exist in the database, Weasel can figure out if there is any “delta” between the expected table from the Table specification and the actual database table. Weasel can issue SQL patches to:

  • Add missing columns
  • Remove columns in the database that are not part of the specification
  • Modify the primary key if necessary
  • Add missing indexes
  • Remove indexes that are not reflected in the specification
  • Deal with foreign keys

And of course, Weasel will do absolutely nothing else if it does not find any differences between the tables.

Likewise, Weasel supports functions and sequences for PostgreSQL. The Weasel.SqlServer has similar support for tables, stored procedures, and custom types (Wolverine uses quite a few user defined table types as an optimization to batch up updates and inserts with its Sql Server integration).

So Weasel definitely isn’t the best documented or visible library in the Critter Stack, but it is useful outside of Marten and Wolverine, and the documentation story might improve dramatically if there’s more demand for that.

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.