Multi-Tenancy: Marten’s “Conjoined” Model

This is continuing a series about multi-tenancy with Marten, Wolverine, and ASP.Net Core:

  1. What is it and why do you care?
  2. Marten’s “Conjoined” Model (this post)

Let’s say that you definitely have the need for multi-tenanted storage in your system, but don’t expect enough data to justify splitting the tenant data over multiple databases, or maybe you just really don’t want to mess with all the extra overhead of multiple databases.

“Conjoined” is a term I personally coined for Marten years ago and isn’t anything that’s an “official” term in the industry. I’m not aware of any widely used pattern name for this strategy, but there surely is somewhere since this is so common.

This is where Marten’s “Conjoined” multi-tenancy model comes into play. Let’s say that we have a little document in our system named User just to store information about our users:

public class User
{
    public User()
    {
        Id = Guid.NewGuid();
    }

    public List<Friend> Friends { get; set; }

    public string[] Roles { get; set; }
    public Guid Id { get; set; }
    public string UserName { get; set; }
    public string FirstName { get; set; }
    public string LastName { get; set; }

    public string? Nickname { get; set; }
    public bool Internal { get; set; }
    public string Department { get; set; } = "";
    public string FullName => $"{FirstName} {LastName}";
    public int Age { get; set; }

    public DateTimeOffset ModifiedAt { get; set; }

    public void From(User user)
    {
        Id = user.Id;
    }

    public override string ToString()
    {
        return $"{nameof(FirstName)}: {FirstName}, {nameof(LastName)}: {LastName}";
    }
}

Now, the User document certainly needs to be tracked within a single logical tenant, so I’m going to tell Marten to do exactly that:

        // This is the same syntax to configuring Marten
        // by IServiceCollection.AddMarten()
        using var store = DocumentStore.For(opts =>
        {
            // other configuration

            // Make *only* the User document be stored by tenant
            opts.Schema.For<User>().MultiTenanted();
        });

In the case above, I am only telling Marten to make the User document be multi-tenanted as it’s frequently valuable — and certainly possible — for some reference documents to be common for all tenants. If instead we just wanted to say “all documents and the event store should be multi-tenanted,” we can do this:

        using var store = DocumentStore.For(opts =>
        {
            // other configuration

            opts.Policies.AllDocumentsAreMultiTenanted();
            opts.Events.TenancyStyle = TenancyStyle.Conjoined;
        });

Either way, if we’ve established that User should be multi-tenanted, Marten will add a tenant_id column to the storage table for the User document like this:

DROP TABLE IF EXISTS public.mt_doc_user CASCADE;
CREATE TABLE public.mt_doc_user (
    tenant_id           varchar                     NOT NULL DEFAULT '*DEFAULT*',
    id                  uuid                        NOT NULL,
    data                jsonb                       NOT NULL,
    mt_last_modified    timestamp with time zone    NULL DEFAULT (transaction_timestamp()),
    mt_version          uuid                        NOT NULL DEFAULT (md5(random()::text || clock_timestamp()::text)::uuid),
    mt_dotnet_type      varchar                     NULL,
CONSTRAINT pkey_mt_doc_user_tenant_id_id PRIMARY KEY (tenant_id, id)
);

As of Marten 7, Marten also places the tenant_id first in the primary key for more efficient index usage when querying large data tables.

You might also notice that Marten adds tenant_id to the primary key for the table. Marten will happily allow you to use the same identity for documents in different tenants. And even though that’s unlikely with a Guid as the identity, it’s very certainly possible with other identity strategies and early Marten users hit that occasionally.

Let’s see the conjoined tenancy in action:

        // I'm creating a session specifically for a tenant id of
        // "tenant1"
        using var session1 = store.LightweightSession("tenant1");

        // My youngest & I just saw the Phantom Menace in the theater
        var user = new User { FirstName = "Padme", LastName = "Amidala" };

        // Marten itself assigns the identity at this point
        // if the document doesn't already have one
        session1.Store(user);
        await session1.SaveChangesAsync();

        // Let's open a session to a completely different tenant
        using var session2 = store.LightweightSession("tenant2");

        // Try to find the same user we just persisted in the other tenant...
        var user2 = await session2.LoadAsync<User>(user.Id);

        // And it shouldn't exist!
        user2.ShouldBeNull();

In the very last call to Marten to try to load the same User, but from the “tenant2” tenant used this SQL:

select d.id, d.data from public.mt_doc_user as d where id = $1 and d.tenant_id = $2
  : f746f237-ed4f-4aaa-b805-ad05f7ae2cd3
  : tenant2

If you squint really hard, you can see that Marten automatically stuck in a second WHERE filter for the current tenant id. Moreover, if we switch to LINQ and try to query that way like so:

        var user3 = await session2.Query<User>().SingleOrDefaultAsync(x => x.Id == user.Id);
        user3.ShouldBeNull();

Marten is still quietly sticking in that tenant_id == [tenant id] filter for us with this SQL:

select d.id, d.data from public.mt_doc_user as d where (d.tenant_id = $1 and d.id = $2) LIMIT $3;
  $1: tenant2
  $2: bfc53828-d56b-4fea-8d93-e8a22fe2db40
  $3: 2

If you really, really need to do this, you can query across tenants with some special Marten LINQ helpers:

        var all = await session2
            .Query<User>()
            
            // Notice AnyTenant()
            .Where(x => x.AnyTenant())
            .ToListAsync();
        
        all.ShouldContain(x => x.Id == user.Id);

Or for specific tenants:

        var all = await session2
            .Query<User>()

            // Notice the Where()
            .Where(x => x.TenantIsOneOf("tenant1", "tenant2", "tenant3"))
            .ToListAsync();

        all.ShouldContain(x => x.Id == user.Id);

Summary

While I don’t think folks should willy nilly build out the “Conjoined” model from scratch without some caution, Marten’s model is pretty robust after 8-9 years of constant use from a large, unfortunately for me the maintainer, creative user base.

I didn’t discuss the Event Sourcing functionality in this post, but do note that Marten’s conjoined tenancy model also applies to Marten’s event store and the projected documents built by Marten as well.

In the next post, we’ll branch out to using different databases for different tenants.

Multi-Tenancy: What is it and why do you care?

I’m always on the lookout for ideas about how to endlessly promote both Marten & Wolverine. Since I’ve been fielding a lot of questions, issues, and client requests around multi-tenancy support in both tools the past couple weeks, now seems to be a good time for a new series exploring the existing foundation in both critter stack tools for handling quite a few basic to advanced multi-tenancy scenarios. But first, let’s start by just talking about what the phrase “multi-tenancy” even means for architecting software systems.

In the course of building systems, you’re frequently going to have a single system that needs to serve different sets of users or clients. Some examples I’ve run across have been systems that need to segregate data for different partner companies, different regions within the same company, or just flat out different users like online email services do today.

I don’t know the origin of the terminology, but we refer to those logical separations within the system data as “tenants.”

My youngest is very quickly outgrowing Dr. Seuss books, but we still read “Because a Bug went Kachoo!” above

It’s certainly going to be important many times to keep the data accessed through the system segregated so that nobody is able to access data that they should not. For example, I probably shouldn’t be able to read you email inbox when I lot into my gmail account. For another example from my early career, I worked with an early web application system that was used to gather pricing quotes from my very large manufacturing company’s suppliers for a range of parts. Due to a series of unfortunate design decisions (because a bug went kachoo!), that application did a very poor job being able to segregate data, and I figured out that some of our suppliers were able to see the quoted prices from their competitors and get unfair advantages.

So we can all agree that mixing up the data between users who shouldn’t see each other’s data is a bad thing, so what can we do about that? The most extreme solution is to just flat out deploy a completely different set of servers for each segregated group of users as shown below:

While there are some valid reasons once in awhile to do the completely separate deployments, that’s potentially a lot of overhead and extra hosting costs. At best, this is probably only viable for a finite number of deployments (Gmail is certainly not building out a separate web server for every one of us with a Gmail account for example).

When a single deployed system is able to serve different tenants, we call that “multi-tenancy.”

According to Wikipedia:

Software multitenancy is a software architecture in which a single instance of software runs on a server and serves multiple tenants.

With multi-tenancy, we’re ensuring that one single deployment of the logical service can handle requests for multiple tenants without allowing users from one tenant to inappropriately see data from other tenants.

Roughly speaking, I’m familiar with three different ways to achieve multi-tenancy.

The first approach is to use one database for all tenant data, but to use some sort of tenant id field that just denotes which tenant the data belongs to. This is what I termed “Conjoined Tenancy” in Marten. This approach is simpler in terms of the database deployment and database change management because after all, there’s only one of them! It is potentially more complex within your codebase because your persistence layer will always need to apply filters on the data being modified and accessed by the user and whichever tenant they are part of.

    There’s some inherent risk with this approach as developers aren’t perfectly omniscient, and there’s always a chance that we miss some scenarios and let data leak out inappropriately to the wrong users. I think this approach is much more viable when using persistence tooling that has strong support (like Marten!) for this type of “conjoined multi-tenancy” and mostly takes care of the tenancy filters for you.

    The second approach is to use a separate schema for each tenant within the same database. I’ve never used this approach myself, and I’m not aware of any tooling in my own .NET ecosystem that supports this approach out of the box. I think this would be a good approach if you were building something on top of a relational database from scratch with a custom data layer — but I think it would be a lot of extra overhead managing the database schema migrations.

    The third way to do multi-tenancy is to use a separate database for each tenant, but the single deployed system is smart enough to connect to the correct database throughout its persistence layer based on the current user (or through metadata on messages as we’ll see in a later entry on Wolverine multi-tenancy). This approach is shown below:

    There’s of course some challenges to this approach as well. First off, there’s more databases to worry about, and subsequently more overhead for database migrations and management. This approach does give you rock solid data segregation between tenants, and I’ve heard of strong business or regulatory requirements to take this approach even when the data volume wouldn’t require this. As my last statement hints at, we all know that the system database is very commonly the bottleneck for performance and scalability, so segregating different tenant data into separate databases may be a good way to improve the scalability of your system.

    It’s obviously going to be more difficult to do any kind of per-tenant data rollup or summary with the separate database approach, but some cloud providers have specialized infrastructure for per tenant database multi-tenancy.

    A Note about Scalability

    I was taught very early on that an effective way to scale systems was to design for any given server to be able to handle all the possible types of operations, then add more servers to the horizontal cluster. I think at the time this was in reaction to several systems we had where teams had tried to scale bigger systems by segregating all operations for one region to one set of servers, and a different set of servers for other regions. The end result was an explosion of deployed servers and frequently having servers absolutely pegged on CPU or memory while North America factories were in full swing while the servers tasked with handling factories on the Pacific Rim were completely dormant when their factories were closed. An architecture that can spread all the work across the cluster of running nodes might often be a much cheaper solution in the end than standing up many more nodes that can only service a subset of tenants.

    Then again, you might also want to prioritize some tenants over others, so take everything I just said with a grain of “it depends” salt.

    Thar be Dragons!

    In the next set of posts, I’ll get into first Marten, then Wolverine capabilities for multi-tenancy, but just know first that there’s a significant set of challenges ahead:

    • Managing multiple database schemas if using separate databases per tenant
    • Needing to use per-tenant filters if using the conjoined storage model for query segregation — and trust me as the author of a persistence tool, there’s plenty of edge case dragons here
    • Detecting the current tenant based on HTTP requests or messaging metadata
    • Communicating the tenant information when using asynchronous messaging
    • Querying across tenants
    • Dynamically spinning up new tenant databases at runtime — or tearing them down! — or even moving them at runtime?!?
    • Isolated data processing by tenant database
    • Multi-level tenancy!?! JasperFx helped a customer build this out with Marten
    • Transactional outbox support in a multi-tenanted work — which Wolverine can do today!

    The two “Critter Stack” tools help with most of these challenges today, and I’ll get around to some discussion about future work to help fill in the more advanced usages that some real users are busy running into right now.

    Wolverine’s Test Support Diagnostics

    I’m working on fixing a reported bug with Wolverine today and its event forwarding from Marten feature. I can’t say that I yet know why this should-be-very-straightforward-and-looks-exactly-like-the-currently-passing-tests bug is happening, but it’s a good time to demonstrate Wolverine’s automated testing support and even how it can help you to understand test failures.

    First off, and I’ll admit that there’s some missing context here, I’m setting up a system such that when this message handler is executed:

    public record CreateShoppingList();
    
    public static class CreateShoppingListHandler
    {
        public static string Handle(CreateShoppingList _, IDocumentSession session)
        {
            var shoppingListId = CombGuidIdGeneration.NewGuid().ToString();
            session.Events.StartStream<ShoppingList>(shoppingListId, new ShoppingListCreated(shoppingListId));
            return shoppingListId;
        }
    }
    

    The configured Wolverine + Marten integration should be kicking in to publish the event appended in the handler above to a completely different handler shown below with the Marten IEvent wrapper so that you can use Marten event store metadata within the secondary, cascaded message:

    public static class IntegrationHandler
    {
        public static void Handle(IEvent<ShoppingListCreated> _)
        {
            // Don't need a body here, and I'll show why not
            // next
        }
    }
    

    Knowing those two things, here’s the test I wrote to reproduce the problem:

        [Fact]
        public async Task publish_ievent_of_t()
        {
            // The "Arrange"
            using var host = await Host.CreateDefaultBuilder()
                .UseWolverine(opts =>
                {
                    opts.Policies.AutoApplyTransactions();
    
                    opts.Services.AddMarten(m =>
                    {
                        m.Connection(Servers.PostgresConnectionString);
                        m.DatabaseSchemaName = "forwarding";
    
                        m.Events.StreamIdentity = StreamIdentity.AsString;
                        m.Projections.LiveStreamAggregation<ShoppingList>();
                    }).UseLightweightSessions()
                    .IntegrateWithWolverine()
                    .EventForwardingToWolverine();;
                }).StartAsync();
            
            // The "Act". This method is an extension method in Wolverine
            // specifically for facilitating integration testing that should
            // invoke the given message with Wolverine, then wait until all
            // additional "work" is complete
            var session = await host.InvokeMessageAndWaitAsync(new CreateShoppingList());
    
            // And finally, just assert that a single message of
            // type IEvent<ShoppingListCreated> was executed
            session.Executed.SingleMessage<IEvent<ShoppingListCreated>>()
                .ShouldNotBeNull();
        }
    

    And now, when I run the test — which “helpfully” reproduces reported bug from earlier today — I get this output:

    System.Exception: No messages of type Marten.Events.IEvent<MartenTests.Bugs.ShoppingListCreated> were received
    
    System.Exception
    No messages of type Marten.Events.IEvent<MartenTests.Bugs.ShoppingListCreated> were received
    Activity detected:
    
    ----------------------------------------------------------------------------------------------------------------------
    | Message Id                             | Message Type                          | Time (ms)   | Event               |
    ----------------------------------------------------------------------------------------------------------------------
    | 018f82a9-166d-4c71-919e-3bcb04a93067   | MartenTests.Bugs.CreateShoppingList   |          873| ExecutionStarted    |
    | 018f82a9-1726-47a6-b657-2a59d0a097cc   | System.String                         |         1057| NoRoutes            |
    | 018f82a9-17b1-4078-9997-f6117fd25e5c   | EventShoppingListCreated              |         1242| Sent                |
    | 018f82a9-166d-4c71-919e-3bcb04a93067   | MartenTests.Bugs.CreateShoppingList   |         1243| ExecutionFinished   |
    | 018f82a9-17b1-4078-9997-f6117fd25e5c   | EventShoppingListCreated              |         1243| Received            |
    | 018f82a9-17b1-4078-9997-f6117fd25e5c   | EventShoppingListCreated              |         1244| NoHandlers          |
    ----------------------------------------------------------------------------------------------------------------------
    

    EDIT: If I’d read this more closely before, I would have noticed that the problem was somewhere different than the routing that I first suspected from a too casual read.

    The textual table above is Wolverine telling me what it did do during the failed test. In this case, the output does tip me off that there’s some kind of issue with the internal message routing in Wolverine that should be applying some special rules for IEvent<T> wrappers, but was not in this case. While that work fixing the real bug continues for me, what I hope you get out of this is how Wolverine is trying to help you diagnose test failures by providing diagnostic information about what was actually happening internally during all the asynchronous processing. As a long veteran of test automation efforts, I will vociferously say that it’s important for test automation harnesses to be able to adequately explain the inevitable test failures. Like Wolverine helpfully does.

    Now, back to work trying to actually fix the problem…

    Wolverine’s HTTP Model Does More For You

    One of the things I’m wrestling with right now is frankly how to sell Wolverine as a server side toolset. Yes, it’s technically a message library like MassTransit or NServiceBus. It can also be used as “just” a mediator tool like MediatR. With Wolverine.HTTP, it’s even an alternative HTTP endpoint framework that’s technically an alternative to FastEndpoints, MVC Core, or Minimal API. You’ve got to categorize Wolverine somehow, and we humans naturally understand something new by comparing it to some older thing we’re already familiar with. In the case of Wolverine, it’s drastically selling the toolset short by comparing it to any of the older application frameworks I rattled off above because Wolverine fundamentally does much more to remove code ceremony, improve testability throughout your codebase, and generally just let you focus more on core application functionality than older application frameworks.

    This post was triggered by a conversation I had with a friend last week who told me he was happy with his current toolset for HTTP API creation and couldn’t imagine how Wolverine’s HTTP endpoint model could possibly reduce his efforts. Challenge accepted!

    For just this moment, consider a simplistic HTTP service that works on this little entity:

    public record Counter(Guid Id, int Count);
    

    Now, let’s build an HTTP endpoint that will:

    1. Receive route arguments for the Counter.Id and the current tenant id because of course let’s say that we’re using multi-tenancy with a separate database per tenant
    2. Try to look up the existing Counter entity by its id from the right tenant database
    3. If the entity doesn’t exist, return a status code 404 and get out of there
    4. If the entity does exist, increment the Count property and save the entity to the database and return a status code 204 for a successful request with an empty body

    Just to make it easier on me because I already had this example code, we’re going to use Marten for persistence which happens to have much stronger multi-tenancy built in than EF Core. Knowing all that, here’s a sample MVC Core controller to implement the functionality I described above:

    public class CounterController : ControllerBase
    {
        [HttpPost("/api/tenants/{tenant}/counters/{id}")]
        [ProducesResponseType(204)] // empty response
        [ProducesResponseType(404)]
        public async Task<IResult> Increment(
            Guid id, 
            string tenant, 
            [FromServices] IDocumentStore store)
        {
            // Open a Marten session for the right tenant database
            await using var session = store.LightweightSession(tenant);
            var counter = await session.LoadAsync<Counter>(id, HttpContext.RequestAborted);
            if (counter == null)
            {
                return Results.NotFound();
            }
            else
            {
                counter = counter with { Count = counter.Count + 1 };
                await session.SaveChangesAsync(HttpContext.RequestAborted);
                return Results.Empty;
            }
        }
    }
    

    I’m completely open to recreating the multi-tenancy support from the Marten + Wolverine combo for EF Core and SQL Server through Wolverine, but I’m shamelessly waiting until another company is willing to engage with JasperFx Software to deliver that.

    Alright, now let’s switch over to using Wolverine.HTTP with its WolverineFx.Http.Marten add on Nuget setup. Let’s drink some Wolverine koolaid and write a functionally identical endpoint the Wolverine way:

    You need Wolverine 2.7.0 for this by the way!

        [WolverinePost("/api/tenants/{tenant}/counters/{id}")]
        public static IMartenOp Increment([Document(Required = true)] Counter counter)
        {
            counter = counter with { Count = counter.Count + 1 };
            return MartenOps.Store(counter);
        }
    

    Seriously, this is the same functionality and even the same generated OpenAPI documentation. Some things to note:

    • Wolverine is able to derive much more of the OpenAPI documentation from the type signatures and from policies applied to the endpoint method, like…
    • The usage of the Document(Required = true) tells Wolverine that it will be trying to load a document of type Counter from Marten, and by default it’s going to do that through a route argument named “id”. The Required property tells Wolverine to return a 404 NotFound status code automatically if the Counter document doesn’t exist. This attribute usage also applies some OpenAPI smarts to tag the route as potentially returning a 404
    • The return value of the method is an IMartenOpside effect” just saying “go save this document”, which Wolverine will do as part of this endpoint execution. Using the side effect makes this method a nice, simple pure function that’s completely synchronous. No wrestling with async Task, await, or schlepping around CancellationToken every which way
    • Because Wolverine can see there will not be any kind of response body, it’s going to use a 204 status code to denote the empty body and tag the OpenAPI with that as well.
    • There is absolutely zero Reflection happening at runtime because Wolverine is generating and compiling code at runtime (or ahead of time for faster cold starts) that “bakes” in all of this knowledge for fast execution
    • Wolverine + Marten has a far more robust support for multi-tenancy all the way through the technology stack than any other application framework I know of in .NET (web frameworks, mediators, or messaging libraries), and you can see that evident in the code above where Marten & Wolverine would already know how to detect tenant usage in an HTTP request and do all the wiring for you all the way through the stack so you can focus on just writing business functionality.

    To make this all more concrete, here’s the generated code:

    // <auto-generated/>
    #pragma warning disable
    using Microsoft.AspNetCore.Routing;
    using System;
    using System.Linq;
    using Wolverine.Http;
    using Wolverine.Marten.Publishing;
    using Wolverine.Runtime;
    
    namespace Internal.Generated.WolverineHandlers
    {
        // START: POST_api_tenants_tenant_counters_id_inc2
        public class POST_api_tenants_tenant_counters_id_inc2 : Wolverine.Http.HttpHandler
        {
            private readonly Wolverine.Http.WolverineHttpOptions _wolverineHttpOptions;
            private readonly Wolverine.Runtime.IWolverineRuntime _wolverineRuntime;
            private readonly Wolverine.Marten.Publishing.OutboxedSessionFactory _outboxedSessionFactory;
    
            public POST_api_tenants_tenant_counters_id_inc2(Wolverine.Http.WolverineHttpOptions wolverineHttpOptions, Wolverine.Runtime.IWolverineRuntime wolverineRuntime, Wolverine.Marten.Publishing.OutboxedSessionFactory outboxedSessionFactory) : base(wolverineHttpOptions)
            {
                _wolverineHttpOptions = wolverineHttpOptions;
                _wolverineRuntime = wolverineRuntime;
                _outboxedSessionFactory = outboxedSessionFactory;
            }
    
    
    
            public override async System.Threading.Tasks.Task Handle(Microsoft.AspNetCore.Http.HttpContext httpContext)
            {
                var messageContext = new Wolverine.Runtime.MessageContext(_wolverineRuntime);
                // Building the Marten session
                await using var documentSession = _outboxedSessionFactory.OpenSession(messageContext);
                if (!System.Guid.TryParse((string)httpContext.GetRouteValue("id"), out var id))
                {
                    httpContext.Response.StatusCode = 404;
                    return;
                }
    
    
                var counter = await documentSession.LoadAsync<Wolverine.Http.Tests.Bugs.Counter>(id, httpContext.RequestAborted).ConfigureAwait(false);
                // 404 if this required object is null
                if (counter == null)
                {
                    httpContext.Response.StatusCode = 404;
                    return;
                }
    
                
                // The actual HTTP request handler execution
                var martenOp = Wolverine.Http.Tests.Bugs.CounterEndpoint.Increment(counter);
    
                if (martenOp != null)
                {
                    
                    // Placed by Wolverine's ISideEffect policy
                    martenOp.Execute(documentSession);
    
                }
    
                
                // Commit any outstanding Marten changes
                await documentSession.SaveChangesAsync(httpContext.RequestAborted).ConfigureAwait(false);
    
                
                // Have to flush outgoing messages just in case Marten did nothing because of https://github.com/JasperFx/wolverine/issues/536
                await messageContext.FlushOutgoingMessagesAsync().ConfigureAwait(false);
    
                // Wolverine automatically sets the status code to 204 for empty responses
                if (!httpContext.Response.HasStarted) httpContext.Response.StatusCode = 204;
            }
    
        }
    
        // END: POST_api_tenants_tenant_counters_id_inc2
        
        
    }
    
    

    Summary

    Wolverine isn’t “just another messaging library / mediator / HTTP endpoint alternative.” Rather, Wolverine is a completely different animal that while fulfilling those application framework roles for server side .NET, potentially does a helluva lot more than older frameworks to help you write systems that are maintainable, testable, and resilient. And do all of that with a lot less of the typical “Clean/Onion/Hexagonal Architecture” cruft that shines in software conference talks and YouTube videos but helps lead teams into a morass of unmaintainable code in larger systems in the real world.

    But yes, the Wolverine community needs to find a better way to communicate how Wolverine adds value above and beyond the more traditional server side application frameworks in .NET. I’m completely open to suggestions — and fully aware that some folks won’t like the “magic” in the “drank all the Wolverine Koolaid” approach I used.

    You can of course use Wolverine with 100% explicit code and none of the magic.

    Scaling Marten with PostgreSQL Read Replicas

    JasperFx Software is open for business and offering consulting services (like helping you craft scalability 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.

    First off, big thanks to Jaedyn Tonee for actually doing all the work I’m writing about here. JT recently accepted a long standing “offer” to be part of the official “Critter Stack Core Team.”

    Marten 7 embraced several new-ish features in Npgsql, including the NpgsqlDataSource concept for connection management. This opened Marten up for a couple other capabilities like integrating Marten with .NET Aspire. It also enabled Marten to utilize PostgreSQL Read Replicas for read only query usage. Read Replicas are valuable both for high availability of database systems, and to offload heavy read intensive operations off of the main database server and onto read replicas.

    To opt into read replicas with Marten, you need to utilize the new MultiHostNpgsqlDataSource in Npgsql and Marten as shown in this sample code:

    // services is an IServiceCollection collection
    services.AddMultiHostNpgsqlDataSource(ConnectionSource.ConnectionString);
    
    services.AddMarten(x =>
        {
            // Will prefer standby nodes for querying.
            x.Advanced.MultiHostSettings.ReadSessionPreference = TargetSessionAttributes.PreferStandby;
        })
        .UseLightweightSessions()
        .UseNpgsqlDataSource();
    

    Behind the scenes, if you are opting into this model, when you make a query with Marten like this:

           // theSession is an IDocumentSession 
           var users = await theSession
                .Query<User>()
                .Where(x => x.FirstName == "Sam")
                .ToListAsync();
    

    Marten will be trying to connect to a PostgreSQL read replica to service that LINQ query.

    Summary

    I hope this is an important new tool for Marten users to achieve both high availability and scalability within systems with bigger data loads.

    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.

    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.