Reading and replaying a stream

Read an aggregate's batches in version order and fold them into state. See Reads and ordering for the model.

Replay the whole stream

using Celeriant.Client.Errors;
using Celeriant.Client.Requests;

var order = OrderState.Empty;
long version = 0;

await foreach (var batch in pool.ReadAllAsync(key, ReadFilters.From(1)))
{
    foreach (var e in batch.Events)
        order = order.Apply(e);
    version = batch.AggregateVersion;
}

ReadAllAsync follows the page cursor for you on one leased connection. Keep version: it is the value a conditional write guards on, and ReadFilters.From(version + 1) catches up from there later. Reading past the tip returns nothing, not an error.

A missing aggregate throws AggregateNotFoundException (1001). For a fold, that usually means "empty state, version 0".

One page at a time

ReadAsync returns a single page. The server sizes pages by bytes, capped by --max-response-size, so a long stream spans several. NextAggregateVersion is the next From value; null means you reached the end.

long? from = 1;
while (from is long start)
{
    var page = await pool.ReadAsync(new ReadRequest
    {
        AggregateKey = key,
        Filters = ReadFilters.From(start),
    });
    foreach (var batch in page.EventBatches)
        Handle(batch);
    from = page.NextAggregateVersion;
}

Use this when you checkpoint between pages. Otherwise use ReadAllAsync.

Filters

ReadFilters is a record struct. Start with ReadFilters.From(n) (values below 1 read as 1) and narrow with with:

var filters = ReadFilters.From(1) with
{
    ToAggregateVersion = 100,            // inclusive
    IncludeEventTypes  = [1, 2, 3],      // matched against EventTypeMajor only
    MinEventTimestamp  = since,          // your EventTimestamp; MinServerTimestamp for commit time
    ExcludeClientId    = myWriterId,     // skip batches this writer made
};

The other filters: MaxEventTimestamp, MaxServerTimestamp, IncludeClientId, IncludeUserId, ExcludeUserId, Min/MaxClientSeq, Min/MaxEventSeq.

A filtered read is not a full replay. Do not fold state from it unless the fold only cares about the events you kept.

Trimmed history

After a trim, batches below the aggregate's minimum version are gone. A read that starts below it throws BatchIndexUnavailableException (1000), and MinimumAvailableVersion says where history now begins:

try
{
    await foreach (var batch in pool.ReadAllAsync(key, ReadFilters.From(1)))
        Handle(batch);
}
catch (BatchIndexUnavailableException ex)
{
    // everything before ex.MinimumAvailableVersion is gone for good
    await foreach (var batch in pool.ReadAllAsync(key, ReadFilters.From(ex.MinimumAvailableVersion)))
        Handle(batch);
}

Restarting from the floor only gives correct state if you can seed it: a snapshot you kept, or a snapshot event written before the trim. AggregateDetailsAsync returns MinAggregateVersion if you want the floor up front.

Which node answers

Reads go to the leader by default, so a read after your own write sees it. Set RouteReadsToFollowers = true on CeleriantPoolOptions to send reads, details, lists and watches to followers instead. That sheds leader load and gives up read-your-writes: a lagging follower can return an old tip, or AggregateNotFoundException for an aggregate you just created. If every follower fails, the pool falls back to the leader.

To stay current after the replay, follow the live tail with a watch. For the full catch-up-then-follow pattern, see Building a read model.