Skip to main content

Your first aggregate

The Quickstart appended one event with no guards. Real writers need two: a conditional write, so a stale decision cannot land, and idempotency, so a retry cannot land twice. The examples are .NET; the other clients take the same fields under their own names.

Setup, as in the quickstart:

using System.Text;
using Celeriant.Client;
using Celeriant.Client.Errors;
using Celeriant.Client.Requests;
using Celeriant.Client.Responses;

await using var pool = new CeleriantPool(new CeleriantPoolOptions { Address = "localhost:10000" });

var key      = new AggregateKey(Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid());
var writerId = Guid.Parse("44444444-4444-4444-4444-444444444444"); // stable per writer, across restarts

Create, and only create​

An aggregate is addressed by org / type / id and exists from its first write. Its version counts writes: each successful write adds one batch and bumps the version by 1, however many events it carries. A new aggregate is at version 0.

So expectedVersion: 0 means "this aggregate must not exist yet":

var created = await pool.WriteAsync(key,
    [new AggregateEvent
    {
        ClientSeq      = 1,
        EventTypeMajor = 1,
        EventTimestamp = DateTimeOffset.UtcNow,
        EventValue     = Encoding.UTF8.GetBytes("""{ "sku": "A-1", "qty": 2 }"""),
    }],
    clientId: writerId,
    expectedVersion: 0,
    enforceClientIdempotency: true);

long version = created.MaxAggregateVersion!.Value;   // 1

Two writers racing to create the same order: one wins, the other gets a conflict. Without expectedVersion, both append.

Append conditionally​

Pass the version your decision was based on. The server appends only if the aggregate is still there:

try
{
    var appended = await pool.WriteAsync(key,
        [new AggregateEvent
        {
            ClientSeq      = 2,
            EventTypeMajor = 2,
            EventTimestamp = DateTimeOffset.UtcNow,
            EventValue     = Encoding.UTF8.GetBytes("""{ "event": "shipped" }"""),
        }],
        clientId: writerId,
        expectedVersion: version,
        enforceClientIdempotency: true);

    version = appended.MaxAggregateVersion!.Value;   // 2
}
catch (WriteOccException ex)
{
    // someone else wrote first; nothing was appended.
    // read from version + 1 up to ex.CurrentAggregateVersion, re-decide, retry.
}

WriteOccException is error 2003. It carries ExpectedVersion and CurrentAggregateVersion. Holding no version? AggregateDetailsAsync returns it as MaxAggregateVersion.

Idempotency​

A timeout does not tell you whether the write landed. Retrying blind can double-append; giving up can lose it. ClientSeq settles it.

With enforceClientIdempotency: true, the server keeps the highest ClientSeq it has accepted per (aggregate, clientId) and rejects a write whose lowest ClientSeq is at or below it. So:

  • Number your events in increasing order, per aggregate, and never reuse a number.
  • Retry with the same ClientSeq. The retry either lands once or is rejected as a duplicate.
  • Keep clientId stable across restarts. A new clientId is a new sequence space, and the old sequences stop protecting you.
Code.NET exceptionWhat it meansDo
2003WriteOccExceptionthe aggregate moved past expectedVersionre-read, re-decide
2002IdempotencyViolationExceptionthis ClientSeq was already accepted for this clientIdread back and check it was yours
2013InflightDuplicateWriteExceptionthe earlier attempt with this ClientSeq is still in flight, not yet durableback off, retry with the same ClientSeq

Two traps:

  1. The version check runs first. Retry a conditional write that already landed and you get 2003, not 2002: the aggregate is now past your expectedVersion. Read from version + 1 and look for your ClientSeq before treating it as a conflict.
  2. A 2002 proves the sequence was used, not that your event used it. Concurrent requests sharing a clientId can take each other's numbers. Read the sequence back and compare.

Implementing idempotent writes has the full retry loop that handles both.

Next​