Marten Event Sourcing in Production: Streams, Aggregates & Projections

Marten event sourcing is a stream of facts in PostgreSQL. Marten projections are the read model built from that stream. Both have to be operable the morning after the first deploy.

This guide is the write side and the read side of one system. A stream is identified, an aggregate decides, two writers collide, and an old event has to change shape without a rewritten history. On the other side, a projection is either inside the commit or behind it, and a rebuild is an operation, not a button you hope nobody presses.

Storing the current document, with no event stream, is Marten as a Document Database. Handlers that publish the next message are in CQRS in .NET with Wolverine. The commit that must include that message is in Transactional Outbox with Wolverine. The design of that context is Marten, Polecat, and Wolverine architecture. The hands-on version, on your domain, is the Marten and Wolverine workshop.

The stream is the identity

A stream is the sequence of events for one thing. The stream id is the identity of that thing. I use one stream per aggregate instance, and I decide the id before the first event is stored. A Guid the caller already has is a fine id. A string such as order/{id} is also fine. Inventing the id inside the database is how two requests create two streams for one order.

StartStream is the first write. A later decision appends. Starting the same id twice fails the commit with ExistingStreamIdCollisionException. That failure is the one I want. A second history for the same order is not.

session.Events.StartStream<Order>(
    orderId,
    new OrderPlaced(orderId, customerId));

await session.SaveChangesAsync();

The events land in mt_events. The stream row is in mt_streams. Both only grow. A projection can be dropped and built again. These two tables cannot.

The aggregate decides

The aggregate is a function of the stream. Create folds the first event. Apply folds each event after that. A command is allowed to say no before any event is appended. After the event is stored, the decision has already happened.

public class Order
{
    public Guid Id { get; set; }
    public bool Shipped { get; private set; }

    public static Order Create(OrderPlaced placed) =>
        new() { Id = placed.OrderId };

    public void Apply(OrderShipped _) => Shipped = true;
}

FetchForWriting<Order> needs that type registered as a single-stream snapshot. Without it, the call has no aggregate to load. I keep the write snapshot inline, so the version I decide against is the version just committed.

opts.Projections.Snapshot<Order>(SnapshotLifecycle.Inline);

The aggregate may refuse a command. It may not reach into another stream to do it. If shipping an order depends on a reservation that lives elsewhere, that fact is already in this stream, or it is a different decision. Cross-stream business rules are where event sourcing turns into a distributed lock with extra steps.

The snapshot document is the current fold, stored so the next decision does not replay the stream. It is not a screen. A list of open orders is a projection further down. Editing the snapshot row by hand skips the stream, and the next rebuild puts the stream back.

Two writes at once

Two decisions will land on the same stream. The write carries the version FetchForWriting just read. Marten rejects the append when that version is no longer the tip. The caller loads again, decides again, and appends again. A lost update that silently sticks is worse than a failed request.

var stream = await session.Events.FetchForWriting<Order>(orderId);

if (stream.Aggregate is null)
    throw new InvalidOperationException($"Order {orderId} does not exist.");

if (stream.Aggregate.Shipped)
    throw new InvalidOperationException("Already shipped.");

stream.AppendOne(new OrderShipped(orderId));
await session.SaveChangesAsync();

The check happens in SaveChangesAsync. The exception is ConcurrencyException. I read stream.Aggregate. I do not mutate it. With an inline snapshot, a field changed on that object can be applied twice: once by the mutation, once by Apply. The event is the change.

FetchForWriting works in both append modes. Passing an expected version to Append yourself requires the rich append mode, and the current default is quick. I stay on FetchForWriting.

If the conflict is common, the stream is too wide. A stream that takes every click in a tenant will conflict all day. Split the identity until a conflict means two real decisions about the same thing.

Changing the event without rewriting history

A bad event stays in the stream. I do not update the row. Four changes, and they are not the same:

• The decision was wrong. Append a compensating event. The history still shows the mistake.
• The class moved, or the class was renamed. Marten stores the event type name in mt_events.type, separate from the CLR name in mt_dotnet_type. AddEventType covers a namespace move when the class name is unchanged. MapEventType covers a renamed class. Stored JSON stays. This is not an upcaster.
• The JSON shape changed. An upcaster runs when the event is read and produces the CLR type the aggregate knows. Stored bytes stay. Old and new code can run a rebuild.

opts.Events.Upcast<OrderPlaced, OrderPlacedWithCurrency>(
    placed => new OrderPlacedWithCurrency(
        placed.OrderId,
        placed.CustomerId,
        "unknown"));

The default in that function is a business decision. An upcaster that calls a service on every read turns a rebuild into an outage. I keep it a pure function.

This overload deserializes the stored JSON into OrderPlaced, then calls the function. OrderPlaced stays in the codebase for as long as those bytes are read. A raw JSON upcast is the registration that does not keep the old CLR type.

• The event should never have been visible. That is an operational delete of a stream, with a reason, not a quiet UPDATE. Treat it as an incident.

A new business fact is a new event. A rename of a property inside the JSON is the upcaster, or a serializer name map. I version the CLR type when readers must see a different shape. I do not version it because a property was renamed and the bytes still mean the same fact.

The read side is a projection

Marten projections are how anything except the aggregate gets a query. The aggregate answers what is true for this decision. The projection answers what this screen lists. A projection that is a copy of the aggregate, one row per stream, is a snapshot you already have.

A single-stream projection folds one stream into one document.

public class OrderSummaryProjection
    : SingleStreamProjection<OrderSummary, Guid>
{
    public OrderSummary Create(OrderPlaced placed) =>
        new() { Id = placed.OrderId, CustomerId = placed.CustomerId };

    public void Apply(OrderShipped shipped, OrderSummary summary) =>
        summary.Shipped = true;
}

A projection is allowed to ignore events. It is not allowed to invent facts that are not in the stream. If the screen needs a fact the stream does not have, the event is missing, and the write side is the place to add it.

Inline, or the async daemon

Three lifecycles. Live folds the stream when something reads it and stores nothing. I use it for a short stream and a rare read. A list screen that live-folds every order on the page will read mt_events for every row.

ProjectionLifecycle.Inline, and SnapshotLifecycle.Inline for the aggregate, run in the same transaction as the append. The document is updated before SaveChangesAsync returns. The cost sits on the write. I use it when the caller cannot leave without that row, and when the fold is cheap. The order snapshot is this case.

ProjectionLifecycle.Async is the daemon. The append commits. The projection catches up from mt_event_progression. Lag is normal. A screen that treats lag as a bug will page you for ordinary load. The order list is this case.

builder.Services.AddMarten(opts =>
{
    opts.Connection(connectionString);
    opts.Projections.Snapshot<Order>(SnapshotLifecycle.Inline);
    opts.Projections.Add<OrderSummaryProjection>(ProjectionLifecycle.Async);
})
.AddAsyncDaemon(DaemonMode.HotCold);

AddAsyncDaemon is what starts the daemon. Registering a projection as Async and never calling it leaves the list frozen at its last rebuild. HotCold elects one leader per projection. Solo runs every shard on that process, which is right when you already know the process is the only one. Solo on two nodes means two daemons write the same progression.

If a stale row changes a decision, the projection is inline, or the screen reads the stream. If the screen can be a few seconds late, it is async, and the lateness is shown.

Rebuilds and lag

A rebuild replays the stream into a projection after Apply changes. While it runs, that read model is behind the write model on purpose.

RebuildProjectionAsync rebuilds the projection you already serve. The default teardown runs TRUNCATE on that document table, then the replay fills it. Until the replay finishes, a query reads an empty table or a partial one. I do that when the screen can be late, or when I can take it offline. UseOptimizedProjectionRebuilds rebuilds one stream at a time and is not this from-zero truncate. I fix a projection that throws before I rebuild it. A rebuild of a projection that still throws fails louder.

A shape change that has to stay up uses a new projection version. [ProjectionVersion(2)] makes Marten write different tables. The new version runs Async while it catches up, even when the previous version was inline. The old table keeps serving. The cutover is the moment the query uses the new document. It is not the moment the replay starts.

Know three numbers before you start: how far behind mt_event_progression is, how long a full replay takes on this database, and which queries are allowed to read the old shape during that window. A rebuild of a projection that walks every event in a large tenant is a maintenance window.

Lag after a deploy is not a failed projection. Lag that grows for an hour is. If the progression mark does not move, the daemon is down or the projection is throwing. The projections commands are the same JasperFx host as db-apply in the document guide. A process that only calls Run() will not rebuild anything.

PostgreSQL is still the database

Event sourcing does not remove vacuum, indexes, or a backup you have restored once. Back up the database, and test a restore, before you argue about snapshot frequency. The event tables are still Marten schema. Production applies them with AutoCreate.None and db-apply, the same step as the document tables.

Inline projections add write latency. Async projections add a process you operate. Either way the events and the documents are in the same PostgreSQL instance unless the write rate says otherwise. A second database for the read model is a replication problem you do not have yet.

The test is the conflict

A test that calls AppendOne and does not commit has not tested the stream. The fact I want is that the second writer keeps the event, and the first writer's commit throws ConcurrencyException.

The store in the fixture is the application store, against PostgreSQL, with the inline order snapshot. I build it once.

public async Task a_stale_append_leaves_the_first_event()
{
    var orderId = Guid.NewGuid();
    var customerId = "cust-1";

    await using (var setup = store.LightweightSession())
    {
        setup.Events.StartStream<Order>(
            orderId,
            new OrderPlaced(orderId, customerId));
        await setup.SaveChangesAsync();
    }

    await using var first = store.LightweightSession();
    await using var second = store.LightweightSession();

    var byFirst = await first.Events.FetchForWriting<Order>(orderId);
    var bySecond = await second.Events.FetchForWriting<Order>(orderId);

    bySecond.AppendOne(new OrderShipped(orderId));
    await second.SaveChangesAsync();

    byFirst.AppendOne(new OrderShipped(orderId));
    await Should.ThrowAsync<ConcurrencyException>(
        () => first.SaveChangesAsync());
}

Testcontainers belongs in this test. After the throw, the stream has one OrderShipped. The first session still holds the version it read, and that version is no longer the tip.