Skip to main content

Building a read model

Celeriant is the write side. To answer queries you fold the event log into a read store and query that. This is where the eventual-consistency tradeoff actually lives, and most of it can be engineered away.

The pattern here follows Celeriant.Reference/ in the .NET client repo: a banking API with a Postgres projection, run as a fleet of replicas.

Two rules​

A projector folds events into a store and remembers how far it got: a cursor, the last aggregate version applied.

  1. Fold in order, from the cursor. Read from cursor + 1, apply, advance.
  2. Write the projection and the cursor in one statement, guarded so the cursor never moves backwards. A crash commits both or neither. A second folder that got there first turns yours into a no-op.

The reference's persist step:

INSERT INTO account_balances (account_id, account_name, balance_cents, last_batch_index, last_client_event_index, updated_at)
VALUES (@id, @name, @balance, @batchIndex, @clientSeq, now())
ON CONFLICT (account_id) DO UPDATE
SET balance_cents = @balance, last_batch_index = @batchIndex,
    last_client_event_index = @clientSeq, updated_at = now()
WHERE account_balances.last_batch_index < @batchIndex

It writes absolute values computed by the fold, never balance = balance + @amount. An increment applied twice is wrong; the same absolute value written twice is not.

Rule 2 buys self-healing for free. If the projection write fails, do nothing clever: the events are in Celeriant, the cursor did not move, and the next catch-up replays them. The reference logs the failure and carries on.

Lazy catch-up​

The reference runs no background projector. Every read and write catches up the one aggregate it touches, on the request path (AccountService.CatchUpAsync, trimmed here):

// 1. projection row and cursor, one query
var (balance, lastBatchIndex) = await LoadRow(accountId);

// already fresh enough for this caller?
if (minBatchIndex is long min && lastBatchIndex >= min)
    return (balance, lastBatchIndex);

// 2. fold whatever is newer than the cursor
var newBalance = balance;
var newBatchIndex = lastBatchIndex;
try
{
    await foreach (var batch in pool.ReadAllAsync(Constants.AccountKey(accountId),
                                                  ReadFilters.From(lastBatchIndex + 1), ct))
    {
        foreach (var evt in batch.Events)
            newBalance = AccountEvents.ReplayEvent(newBalance, evt);
        newBatchIndex = batch.AggregateVersion;
    }
}
catch (AggregateNotFoundException)
{
    return (balance, lastBatchIndex);   // no events yet
}

// 3. persist with the guarded upsert above
if (newBatchIndex > lastBatchIndex)
    await PersistRow(accountId, newBalance, newBatchIndex);
return (newBalance, newBatchIndex);

What this buys:

  • Read-your-writes per aggregate. A read after a write catches up first, so it sees the write. That holds while reads go to the leader, the pool's default; with RouteReadsToFollowers a lagging follower can serve an older tip.
  • Nothing to run. No projector process to deploy, monitor, or restart.
  • Fleet safety. Any replica can catch up any aggregate. The guard makes concurrent catch-ups on one aggregate duplicated work, not duplicated effects.

What it costs: every request pays a Celeriant read, which comes back empty when the row is current. And a row is only as fresh as its last visit, so a query across many aggregates ("every account over a threshold") reads stale rows for accounts nobody touched. That query needs a background projector.

The watch as a freshness hint​

The reference's watch does not drive the projection. It streams notifications to the browser over SSE, and the browser refetches, which runs a catch-up. The balance endpoint also takes minBatchIndex: pass the notification's ToAggregateVersion and catch-up returns early when the row is already there, so a notification the projection has already covered costs one indexed lookup. The reference's own frontend does not pass it.

A background projector​

When queries span aggregates, keep the whole table current. The ordering is the one from Subscribing to live events: subscribe first, then catch up, then follow. Reverse the first two and a write landing in between reaches neither path.

while (!ct.IsCancellationRequested)
{
    try
    {
        // 1. subscribe
        await using var watch = await pool.WatchAsync(new WatchRequest
        {
            AggregateTypes = [Constants.AccountTypeId],
            OperationTypes = [WatchOperationType.Write],
        }, ct: ct);

        // 2. catch up, including aggregates first written while nobody was subscribed
        await foreach (var agg in pool.ListAggregatesAsync(Constants.OrgId, Constants.AccountTypeId, ct: ct))
            await CatchUpAsync(agg.AggregateId, minBatchIndex: agg.MaxAggregateVersion, ct: ct);

        // 3. follow; minBatchIndex drops what catch-up already covered
        while (true)
        {
            var response = await watch.NextAsync(ct);
            foreach (var change in response.Events)
                await CatchUpAsync(change.AggregateId, minBatchIndex: change.ToAggregateVersion, ct: ct);
        }
    }
    catch (Exception e) when (e is not OperationCanceledException)
    {
        await Task.Delay(TimeSpan.FromSeconds(2), ct);   // watch died: resubscribe and catch up again
    }
}

CatchUpAsync is the lazy one above, reused. minBatchIndex is the dedupe: a notification at or below the row's cursor skips the read entirely.

The listing in step 2 is not optional. An aggregate whose first write landed before the subscribe has no row and no cursor, so nothing else would ever find it. Listing walks the whole scope, so it is a start-up and reconnect cost.

The outer loop is not optional either. A watch never reconnects on its own, and one that falls behind is cut off by the server. Each reconnect runs the full subscribe-then-catch-up again, because nothing was notified while it was down.

Run the projector on several replicas if you like. The guarded upsert keeps them correct; they just repeat each other's reads.

Rebuilding​

The log is the source of truth, so a new read model is a new fold over the same events: drop the projection and its cursors, and replay from version 1.

One catch: a trimmed aggregate has no version 1. ReadAllAsync from below the trim point throws BatchIndexUnavailableException (1000), and its MinimumAvailableVersion says where history now starts. A fold can only start there if it can be seeded: a snapshot you kept, or a snapshot event written before the trim. See Reading and replaying.