Skip to main content

Optimistic concurrency

A write to an aggregate can be conditional: append these events only if the aggregate is still at the version I read. If another writer moved it in the meantime, the write is rejected whole and nothing is appended. This is the operation event sourcing needs and the one Kafka cannot do.

How it works​

Every aggregate has a version: the index of its latest batch, starting at 1. Version 0 means the aggregate does not exist. You learn the version when you read, and you pass it back as expectedVersion when you write:

await pool.WriteAsync(
    key,
    events: [orderShipped],
    clientId: writerId,
    expectedVersion: 4,        // commit only if still at version 4
    enforceClientIdempotency: true);

At version 4, the write lands and the aggregate moves to 5. At 5 because someone else appended first, the server returns OptimisticConcurrencyViolation (error 2003), the .NET client throws WriteOccException, and the log is untouched. The exception carries CurrentAggregateVersion, so you know where to catch up from.

A single-aggregate write returns the version it committed (MaxAggregateVersion); a multi-aggregate write leaves it unset.

Leave expectedVersion out (null) and the write is unconditional: it appends to the end whatever the version is. Use that only when contention genuinely does not matter.

Creating is the same check. allowCreate defaults to true on the single-aggregate WriteAsync, so a write to a missing aggregate creates it. Pass expectedVersion: 0 to say "only if nobody created it first": it succeeds only while the aggregate does not exist. With allowCreate: false, a write to a missing aggregate fails with AggregateNotExists (2005) before the version is even compared.

The check runs against the shard's write-side state, which includes writes that are fsynced but not yet replicated and so not yet readable. You can get a 2003 against a version your last read could not show you. The fix is the same: catch up and decide again.

The read-retry loop​

Read the aggregate, fold your state, decide, write with the version you read. On a conflict, do it again:

while (true)
{
    var state = await LoadAggregate(key);   // your code: read the stream, fold state, keep its version
    if (!state.CanShip()) throw new InvalidOperationException();

    var shipped = AggregateEventExtensions.Create(
        OrderShippedType, new OrderShipped(), JsonEventSerializer.Default,
        clientSeq: state.NextClientSeq);        // fresh decision, fresh sequence

    try
    {
        await pool.WriteAsync(key, [shipped],
            clientId: writerId,
            expectedVersion: state.Version,
            enforceClientIdempotency: true);
        break;                                   // committed
    }
    catch (WriteOccException)
    {
        // the aggregate moved; loop and re-decide against the new state
    }
}

A conflict is not an error in your domain. The world changed and your decision needs re-checking. The server never retries a conflict for you; only your domain logic knows whether the write still makes sense. See Handling concurrency conflicts.

Re-deciding is only safe if your own earlier attempt did not land. After a timeout it might have, and the server will not tell you through this path: it checks the version guard before idempotency. With both expectedVersion and enforceClientIdempotency set, resending a write that already landed returns 2003, not 2002, because your own write moved the version. While the earlier attempt is still in flight you get 2013 instead. Idempotent retries has the recipe: catch up, keep the ClientSeq, send with the fresh version.

Across several aggregates​

One write request can carry conditional writes to several aggregates, each with its own ExpectedVersion, and they commit atomically: all or none. The aggregates must route to the same shard; otherwise the request is rejected with ShardRoutingMultipleShards (9001). That is how you enforce an invariant spanning aggregates, such as a transfer that debits one account and credits another, without a distributed transaction. See Consistency boundaries.

Cross-shard atomic writes are not supported, on purpose. If you need them, your routing is drawing the wrong boundary.