Working with an Event Store: Append, Optimistic Concurrency, and Streams with Marten and PostgreSQL

Version note: The code in this post was built and run with C# 14 and .NET 10.

Guess First

It is the first morning of the back-to-school campaign. One copy of Dune is left in the warehouse, and Ada and Alan place their orders in the same second. The two requests run on two threads, in two separate Marten sessions. Both read the stock item's stream, both see "1 copy available", and both get the same decision from the StockDecider of post 05: StockReserved.

StockItemId dune = await NewStockItem(quantity: 1);

// Ada and Alan order the last copy of Dune at the same moment
await using IDocumentSession ada = store.LightweightSession();
await using IDocumentSession alan = store.LightweightSession();
IEventStream<StockItem> adaStream = await ada.Events.FetchForWriting<StockItem>(dune.Value);
IEventStream<StockItem> alanStream = await alan.Events.FetchForWriting<StockItem>(dune.Value);
Console.WriteLine($"Both read: version={adaStream.StartingVersion} available={adaStream.Aggregate!.State.Available}");

adaStream.AppendMany(StockDecider.Decide(
    new ReserveStock(dune, new OrderId(Guid.CreateVersion7()), 1, now), adaStream.Aggregate.State));
OrderId alanOrder = new(Guid.CreateVersion7());
alanStream.AppendMany(StockDecider.Decide(
    new ReserveStock(dune, alanOrder, 1, now), alanStream.Aggregate!.State));

await ada.SaveChangesAsync();
Console.WriteLine("Ada:  saved");
try
{
    await alan.SaveChangesAsync();
    Console.WriteLine("Alan: saved");
}
catch (ConcurrencyException ex)
{
    Console.WriteLine($"Alan: {ex.GetType().Name}");
    Console.WriteLine($"      {ex.Message}");
}

// Retry: same command, a fresh decision on the current state
IReadOnlyList<StockEvent> retry = await handler.HandleAsync(new ReserveStock(dune, alanOrder, 1, now));
Console.WriteLine(retry.Single() is OutOfStock o
    ? $"Alan retry: OutOfStock (requested {o.Requested}, available {o.Available})"
    : $"Alan retry: {retry.Single().GetType().Name}");

What happens after the two SaveChangesAsync calls? There are three options: both reservations are written and we have sold the same copy to two people; the second write is rejected with an exception; or the second request waits for the first to finish and then gets written. Before answering, let's look at why an event store should already have an answer to this question.

What Do We Need from an Event Store?

In post 04 we defined an event store as "a database that keeps events in streams, in order, and immutably", and we separated it from a message broker. From an application's point of view that definition turns into four concrete capabilities:

• Appending events to a stream, and doing so only while the stream is at the version you expect (optimistic concurrency with an expected version).
• Reading a stream from start to end in the order it was written, optionally up to a given version or point in time.
• Reading in a global order across all streams; projections and subscriptions are fed from this (post 07).
• Events written in one transaction become visible all together or not at all.

The first item protects the aggregate's consistency boundary. The rule from the "Aggregate Design: Consistency Boundaries, Small Aggregates, and Referencing by Id" post of the Domain-Driven Design (DDD) series holds here too: a command changes a single aggregate, that is, a single stream, and that stream's version is the only piece of information that stops two writers from overwriting each other's decisions. In the classic relational model the equivalent was the concurrency token from the "Concurrency in EF Core: Tokens, rowversion and DbUpdateConcurrencyException" post of the Entity Framework Core (EF Core) series; in an event store the token is the stream's last version number.

Setting Up Marten: An Event Store on PostgreSQL

The event store in this series is Marten. Marten is a .NET library that uses PostgreSQL as both a document database and an event store; the current version on NuGet is 9.58.0 (November 5, 2026), its license is the MIT (Massachusetts Institute of Technology) License, a permissive open source license, and it targets net9.0 and net10.0. Marten 9 shipped in May 2026 and changed several defaults: the serializer is now System.Text.Json, events are written in the "Quick" append mode by default, and runtime code generation is gone. Marten 8 now only gets critical fixes and goes out of support in November 2026; start a new project on 9.

A PostgreSQL instance is all you need to try it. You can start a temporary container on your machine and remove it when you are done:

docker run -d --name bookstore-pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:17
dotnet add package Marten --version 9.58.0

Setting up the store takes a few lines. I use DocumentStore.For so the examples run in a console app instead of ASP.NET Core; in a web app the same settings go inside builder.Services.AddMarten(...).

string connectionString = Environment.GetEnvironmentVariable("BOOKSTORE_PG")
    ?? "Host=localhost;Port=5432;Database=bookstore;Username=postgres;Password=postgres";

await using DocumentStore store = DocumentStore.For(opts =>
{
    opts.Connection(connectionString);
    opts.DatabaseSchemaName = "inventory";
    opts.Projections.Add(new StockItemProjection(), ProjectionLifecycle.Live);
});

StockItemProjection comes below; ProjectionLifecycle.Live means "do not store this aggregate anywhere, rebuild it from the events on every read". Marten creates the schema itself on the first SaveChangesAsync. The two main tables on the event side are mt_streams (one row per stream: id, type, last version) and mt_events:

SELECT column_name, data_type FROM information_schema.columns
WHERE table_schema = 'inventory' AND table_name = 'mt_events' ORDER BY ordinal_position;

  column_name   |        data_type
----------------+--------------------------
 seq_id         | bigint
 id             | uuid
 stream_id      | uuid
 version        | bigint
 data           | jsonb
 bdata          | bytea
 type           | character varying
 timestamp      | timestamp with time zone
 tenant_id      | character varying
 mt_dotnet_type | character varying
 is_archived    | boolean

SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'inventory' AND tablename = 'mt_events';

 pkey_mt_events_seq_id           | CREATE UNIQUE INDEX pkey_mt_events_seq_id ON inventory.mt_events USING btree (seq_id)
 pk_mt_events_stream_and_version | CREATE UNIQUE INDEX pk_mt_events_stream_and_version ON inventory.mt_events USING btree (stream_id, version)

Note two lines: seq_id is the global order across the whole store, and the unique index on (stream_id, version) is optimistic concurrency's last line of defense in the database. Two events cannot be written to the same stream with the same version. bdata is for binary serialization, tenant_id for multi-tenancy, and is_archived for the archiving we will see in post 08.

StartStream, Append, and Reading

Let's open a stock item, receive five copies, and reserve two:

StockItemId dune = new StockItemId(Guid.CreateVersion7());
await using (IDocumentSession session = store.LightweightSession())
{
    // New stream: known type and id, together with the first events
    session.Events.StartStream<StockItem>(dune.Value,
        new StockItemRegistered(dune, new BookId(Guid.CreateVersion7()), now),
        new StockReceived(dune, 5, now));
    await session.SaveChangesAsync();
}

await using (IDocumentSession session = store.LightweightSession())
{
    // Append to an existing stream (no version check)
    session.Events.Append(dune.Value, new StockReserved(dune, new OrderId(Guid.CreateVersion7()), 2, now.AddMinutes(5)));
    await session.SaveChangesAsync();
}

await using IQuerySession query = store.QuerySession();
foreach (IEvent e in await query.Events.FetchStreamAsync(dune.Value))
    Console.WriteLine($"v{e.Version} seq={e.Sequence} {e.EventTypeName}");

StockItem? current = await query.Events.AggregateStreamAsync<StockItem>(dune.Value);
StockItem? atVersion2 = await query.Events.AggregateStreamAsync<StockItem>(dune.Value, version: 2);
Console.WriteLine($"Current: OnHand={current!.State.OnHand} Available={current.State.Available}");
Console.WriteLine($"At v2:   OnHand={atVersion2!.State.OnHand} Available={atVersion2.State.Available}");

v1 seq=1 stock_item_registered
v2 seq=2 stock_received
v3 seq=3 stock_reserved
Current: OnHand=5 Available=3
At v2:   OnHand=5 Available=5

StartStream<StockItem> opens a new stream and records the stream's type in mt_streams; call it a second time with the same id and you get an ExistingStreamIdCollisionException. Append also opens the stream if it does not exist, and it does not check the version, which is why it is not used on its own in command handlers. FetchStreamAsync returns the events in version order, and AggregateStreamAsync builds state by running them through the projection. The version: 2 parameter is the cheapest way to travel back in time: the answer to "what was the stock before the reservation?" comes without a separate audit table.

In the database the events sit as JSON (JavaScript Object Notation):

 seq_id | version |      type      | data
--------+---------+----------------+------------------------------------------------------------------------------
      2 |       2 | stock_received | {"At": "2026-09-01T09:00:00+00:00", "Quantity": 5, "StockItemId": {"Value": "01a1b850-8783-7b96-9a18-c1bd6bde6e3a"}}
      3 |       3 | stock_reserved | {"At": "2026-09-01T09:05:00+00:00", "OrderId": {"Value": "01a1b850-884d-71dc-bc23-325d6e7b7510"}, "Quantity": 2, "StockItemId": {"Value": "01a1b850-8783-7b96-9a18-c1bd6bde6e3a"}}

The type column is the event's name in Marten (its alias), and mt_dotnet_type is the full name of the .NET type. When reading, Marten looks at the alias first and falls back to the .NET type name if it cannot find it. In the pitfalls section this detail turns into a bug that bites on Turkish machines.

FetchForWriting: Connecting the Decider to the Event Store

The decider in post 05 was made of pure functions: Decide(command, state) produces events and Evolve(state, event) moves the state forward. We do not want to give up that purity when connecting it to the event store. Marten has a single demand: every document it stores or builds must have an Id. Instead of adding an id to StockState, I put it in a thin envelope:

using JasperFx.Events;
using Marten;
using Marten.Events.Aggregation;

namespace Bookstore.Inventory;

// Marten wants an Id on every document; a thin envelope that leaves the decider's state alone
public sealed record StockItem(Guid Id, StockState State);

// Adapter that introduces the decider's Evolve to Marten
public sealed class StockItemProjection : SingleStreamProjection<StockItem, Guid>
{
    public override StockItem Evolve(StockItem? snapshot, Guid id, IEvent e) =>
        e.Data is StockEvent stockEvent
            ? new StockItem(id, StockDecider.Evolve(snapshot?.State ?? StockDecider.InitialState, stockEvent))
            : snapshot ?? new StockItem(id, StockDecider.InitialState);
}

public sealed class StockCommandHandler(IDocumentStore store)
{
    public async Task<IReadOnlyList<StockEvent>> HandleAsync(StockCommand command, CancellationToken ct = default)
    {
        await using IDocumentSession session = store.LightweightSession();

        // Reads the stream together with its current version
        IEventStream<StockItem> stream = await session.Events.FetchForWriting<StockItem>(command.StockItemId.Value, ct);
        StockState state = stream.Aggregate?.State ?? StockDecider.InitialState;

        IReadOnlyList<StockEvent> events = StockDecider.Decide(command, state);
        stream.AppendMany(events);

        // ConcurrencyException here if someone wrote after the version we read
        await session.SaveChangesAsync(ct);
        return events;
    }
}

The Evolve override of SingleStreamProjection<TDoc, TId> has used the same name as the decider terminology since Marten 8; inside it we just call our own StockDecider.Evolve. The StockCommandHandler in post 05 read the stream from the in-memory store, folded it, and appended with the expected version; in the Marten version FetchForWriting takes over two of those three jobs, and we did not touch the decider at all. The steps are the same for every command: read the stream and its current version with FetchForWriting, decide with Decide, add with AppendMany, and save with SaveChangesAsync. Marten's documentation explicitly recommends FetchForWriting for new code, because the same code works whether you keep the aggregate with a Live, Inline, or Async projection. When we move from Live to a snapshot in post 08 we will not touch the handler.

A note: in Marten 9 UseIdentityMapForAggregates is on by default. It reuses the instance FetchForWriting returned for Inline projections, and it is only safe when the handler does not modify that instance. With a decider the state is an immutable record, so the problem never reaches us; with classic classes that modify the aggregate in place, a change can be applied twice.

Optimistic Concurrency: Two Writers on One Stream

Now back to the opening question. The output of a real run:

Both read: version=2 available=1
Ada:  saved
Alan: EventStreamUnexpectedMaxEventIdException
      Unexpected starting version number for event stream '01a1b850-89db-7833-889d-d27079230926', expected 2 but was -1
Alan retry: OutOfStock (requested 1, available 0)

The second option: Alan's write was rejected. Both sessions read the stream at version 2. Ada's event was written as version 3; Alan's session arrived saying "I am writing what comes after version 2" and was refused. EventStreamUnexpectedMaxEventIdException derives from JasperFx's ConcurrencyException, so a single catch is enough in the handler. The "but was -1" in the message may look odd: in Quick append mode, the Marten 9 default, the version check happens in a PostgreSQL function on the server and the actual version does not make it into the message. Running the same scenario with EventAppendMode.Rich turns the message into "expected 2 but was 3".

How the retry is done matters as well. Resending Alan's request with the same events would be wrong, because those events were produced from a now stale state. The right way is to run the command from scratch: read the stream again, decide again. This time the decider saw that no copies were left and, following the rule from post 05, produced an OutOfStock event instead of an exception. Doing this retry once, or a few times with a short delay, is usually enough; if you see constant conflicts on the same stream, the problem is not the retry but the stream's design (see the single global stream pitfall below).

Sometimes it is the client, not the server, that knows the expected version. When the warehouse clerk opened the stock correction screen, the stock was at v2; while they were on that screen, an order made a reservation. The clerk's "3 copies missing" correction is a decision made on a state they no longer see. If you pass the version the screen sends to FetchForWriting, the check happens at read time. In a web API (Application Programming Interface) this version usually goes out as an ETag (entity tag) in the HTTP (Hypertext Transfer Protocol) response and comes back in the If-Match header of the next request:

StockItemId dune = await NewStockItem(quantity: 10);

// The warehouse clerk opened the stock count screen at v2; a reservation came in meanwhile
long versionOnScreen = 2;
await handler.HandleAsync(new ReserveStock(dune, new OrderId(Guid.CreateVersion7()), 1, now));

await using IDocumentSession session = store.LightweightSession();
try
{
    IEventStream<StockItem> stream = await session.Events.FetchForWriting<StockItem>(dune.Value, versionOnScreen);
    stream.AppendOne(new StockAdjusted(dune, -3, AdjustmentReason.CountCorrection, now));
    await session.SaveChangesAsync();
}
catch (ConcurrencyException ex)
{
    Console.WriteLine($"{ex.GetType().Name}: {ex.Message}");
}

ConcurrencyException: Expected the existing version to be 2, but was 3

This time the exception did not come from SaveChangesAsync but from the FetchForWriting call, which is exactly the right place to tell the screen "this record changed while you were looking at it, refresh". On top of that, Marten offers FetchForExclusiveWriting, which takes a PostgreSQL row lock on the stream; that is the "second request waits" option. Except for the rare cases where conflicts are very frequent and retries are expensive, the optimistic approach scales better. The old Append(streamId, expectedVersion, events) overload still exists, but in Marten 9 it only works in Rich mode; the documentation recommends FetchForWriting instead of it as well.

Stream Id Strategies

In Marten a stream id is either a Guid or a string (opts.Events.StreamIdentity = StreamIdentity.AsString). The choice depends on how you reach the aggregate:

Bookstore has a single StockItem per book, so a Guid v7 is enough, and in post 09 we will see how to protect the uniqueness of the ISBN separately. If you use a natural key, make sure the key really never changes; a stream id cannot be changed later. Also know the opts.Events.UseMandatoryStreamTypeDeclaration = true setting: it rejects untyped StartStream calls and Append to a stream that does not exist; it prevents a typo from silently opening a new stream, and Marten's projection rebuild optimizations need it too.

Alternatives: KurrentDB and Your Own Table

Marten is not the only option for an event store. The best known is KurrentDB: EventStoreDB, which Greg Young started, took this name with version 25.0 in March 2025. It is a database with its own server, written for event sourcing: streams, expected versions, subscriptions, and projections are first-class concepts. Its .NET client is KurrentDB.Client (currently 1.6.1, Apache 2.0). The server's license is different: Kurrent License v1, whose source is open but which is not OSI (Open Source Initiative) approved. Since 26.2, released in September 2026, a single-node deployment, the connectors, and the authentication options are free; clusters, read-only replicas, archiving, and encryption at rest need a license key. If the team already runs PostgreSQL, Marten does not add a new infrastructure component; being able to write events and read models in the same transaction also makes the Inline projection we will see in post 07 possible.

The third path is writing your own table. The core really is small:

await using NpgsqlDataSource dataSource = NpgsqlDataSource.Create(connectionString);
await using (NpgsqlCommand ddl = dataSource.CreateCommand("""
    CREATE TABLE IF NOT EXISTS stock_events (
        global_position bigint GENERATED ALWAYS AS IDENTITY,
        stream_id uuid NOT NULL,
        version integer NOT NULL,
        event_type text NOT NULL,
        data jsonb NOT NULL,
        recorded_at timestamptz NOT NULL DEFAULT now(),
        PRIMARY KEY (stream_id, version)
    )
    """))
    await ddl.ExecuteNonQueryAsync();

Guid streamId = Guid.CreateVersion7();
await Append(dataSource, streamId, expectedVersion: 0, "stock_received", """{"quantity": 1}""");

// Both requests read v1 and try to write v2
await Append(dataSource, streamId, expectedVersion: 1, "stock_reserved", """{"quantity": 1}""");
try
{
    await Append(dataSource, streamId, expectedVersion: 1, "stock_reserved", """{"quantity": 1}""");
}
catch (PostgresException ex) when (ex.SqlState == PostgresErrorCodes.UniqueViolation)
{
    Console.WriteLine($"Conflict: {ex.SqlState} on {ex.ConstraintName}");
}

static async Task Append(NpgsqlDataSource dataSource, Guid streamId, int expectedVersion, string type, string json)
{
    await using NpgsqlCommand insert = dataSource.CreateCommand("""
        INSERT INTO stock_events (stream_id, version, event_type, data)
        VALUES ($1, $2, $3, $4::jsonb)
        """);
    insert.Parameters.Add(new() { Value = streamId });
    insert.Parameters.Add(new() { Value = expectedVersion + 1 });
    insert.Parameters.Add(new() { Value = type });
    insert.Parameters.Add(new() { Value = json });
    await insert.ExecuteNonQueryAsync();
    Console.WriteLine($"Appended v{expectedVersion + 1} {type}");
}

Appended v1 stock_received
Appended v2 stock_reserved
Conflict: 23505 on stock_events_pkey

The primary key gives you optimistic concurrency: the same (stream_id, version) cannot be written twice, and PostgreSQL returns 23505 (unique_violation). The hard part starts afterwards. global_position is an identity column; its values are taken at the start of the transaction, but the commit order can differ. If event 105 commits before event 104, a projection that says "the last position I read is 105" misses 104 forever. Marten's async daemon handles these gaps with its "high water mark" mechanism and tombstone events; with your own table you would have to reinvent that. Then come the projection runner, checkpoints, rebuilds, snapshots, event versioning, and serialization. Your own table is reasonable when: there is a single small aggregate type, its projection is one or two tables updated in the same transaction, there is no async reader, and the team has a clear reason not to add a library dependency. On the day an async projection is needed, moving to a ready-made event store is cheaper.

Common Pitfalls

A single global stream

"Let's keep all stock movements in one inventory stream, the order is easy to read" looks tidy at first sight. But optimistic concurrency works per stream: a single stream means only one writer in the whole system can succeed at a time. When 120 orders per minute arrive during a campaign, the conflict from the opening stops being an exception and becomes the rule. The stream is the consistency boundary itself; it should be as small as the aggregate. To read in global order there is already seq_id.

Huge streams

A Live aggregate reads the whole stream on every read. I measured it on the same machine (the second read after a warm-up read):

    10 events -> OnHand=    9, rehydrated in 0 ms
  1000 events -> OnHand=  999, rehydrated in 7 ms
 20000 events -> OnHand=19999, rehydrated in 113 ms

Twenty thousand events means 113 ms on every command. The stock item of a very popular book can reach that number over the years. The fix is either a snapshot or splitting the stream into periods; both are the subject of post 08. What matters here is thinking about stream length at design time: ask "how many events will this stream get in a year?"

Putting a domain object in an event

It is easy to put another model's entity into an event as is:

// Bad example: the event carries Catalog's Book entity as is
public sealed record Book(Guid Id, string Isbn, string Title, string Author, decimal ListPrice, string Currency, string Description);
public sealed record StockReceivedWithBook(Book Book, int Quantity, DateTimeOffset At);

{"At": "2026-09-01T09:00:00+00:00", "Book": {"Id": "01a1b850-978a-7183-94a5-3e90c497fb2b", "Isbn": "978-0441172719", "Title": "Dune", "Author": "Frank Herbert", "Currency": "EUR", "ListPrice": 14.99, "Description": "Set on the desert planet Arrakis..."}, "Quantity": 5}

The event is now tied to the shape Catalog's Book class had that day. When Catalog drops Description or turns ListPrice into Money, you have to remember that old shape to deserialize events written years ago. The stock ledger does not even care about price; an event should carry only the values the decision needs and that belong to its own context (BookId, Quantity). The cost of embedding an entity that carries personal data is even higher; we will come back to that in post 09 on GDPR (General Data Protection Regulation).

Event names on a Turkish machine: stock_ıtem_registered

This is a bug I actually ran into while running the examples for this series. Marten derives the event alias from the .NET type name, and JasperFx's ToTableAlias method uses the culture-sensitive ToLower() for that. If the machine's culture is tr-TR, the capital I becomes the dotless ı:

foreach (string culture in (string[])["en-US", "tr-TR"])
{
    CultureInfo.CurrentCulture = new CultureInfo(culture);
    await using DocumentStore cultureStore = DocumentStore.For(opts =>
    {
        opts.Connection(connectionString);
        opts.DatabaseSchemaName = "inventory_culture";
    });
    await using IDocumentSession session = cultureStore.LightweightSession();
    StockItemId id = new StockItemId(Guid.CreateVersion7());
    session.Events.StartStream<StockItem>(id.Value, new StockItemRegistered(id, new BookId(Guid.CreateVersion7()), now));
    await session.SaveChangesAsync();
}

await using NpgsqlDataSource dataSource = NpgsqlDataSource.Create(connectionString);
await using NpgsqlCommand command = dataSource.CreateCommand(
    "SELECT type, count(*) FROM inventory_culture.mt_events GROUP BY type ORDER BY type");
await using NpgsqlDataReader reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
    Console.WriteLine($"{reader.GetString(0)}: {reader.GetInt64(1)}");

stock_item_registered: 1
stock_ıtem_registered: 1

The same event type sits in the database under two different names. Marten itself tolerates this: when it does not recognize the alias it falls back to the .NET type name in mt_dotnet_type; I ran the async daemon under both en-US and tr-TR, and in both cases it processed both events. The problem shows up in two places. First, every SQL (Structured Query Language) query, report, and tool that looks at the type column sees only half of the events. The second is sneakier: in post 08 we will move the event classes to another namespace; on that day, events written with the dotless-ı alias can be found neither by the alias nor by an old .NET type name that no longer exists. The fix has two steps: give event names explicitly with opts.Events.MapEventType<StockItemRegistered>("stock_item_registered") (in post 08 we will see this is right for versioning too), and run server processes with InvariantGlobalization or a fixed culture.

A Small Experiment

★☆☆ Run the opening race again with EventAppendMode.Rich and compare the exception message. Then use FetchForExclusiveWriting instead of FetchForWriting: what does Alan's session do instead of failing?

★★☆ Move the retry into StockCommandHandler: if a ConcurrencyException arrives, run the command at most three times, reading the stream again each time. Send ten parallel ReserveStock commands to the same stream and count how many retries happen.

★★★ Add a simple projection that reads by global_position to the event store you wrote with your own table. Open two transactions by hand, commit the second one first, and show that the projection misses an event. How would you prevent it?

In this post we wrote events safely and built state from a single stream. But the answers to questions like "how many copies of which books went out this month?" or "which reservations are waiting to be shipped?" are not in a single stream. In the next post we will produce read models designed for reading from events with Inline, Async, and Live projections.