Subscribing to live events

Open a watch to react to writes as they land. A watch delivers change notifications, not payloads: which aggregate moved, and to which version. You read the events yourself.

Open a watch

await using var watch = await pool.WatchAsync(new WatchRequest
{
    AggregateTypes = [ordersType],               // scope: Orgs, AggregateTypes, or Aggregates
    OperationTypes = [WatchOperationType.Write],
});

while (true)
{
    var response = await watch.NextAsync(ct);
    foreach (var change in response.Events)
    {
        // change.OrgId / AggregateTypeId / AggregateId identify the aggregate;
        // change.FromAggregateVersion .. ToAggregateVersion is what changed.
        await ReadAndApply(change);
    }
}

WatchAsync returns once the server has acknowledged the subscription, and the server registers the subscriber before it sends that ack. From that moment every write in scope is queued for you. The connection is dedicated, not pooled; dispose it when done. Some frames arrive with no events: the ack and idle heartbeats. The loop handles them for free.

The other clients have the same shape: pool.watch(request, WatchOptions::default()) then watch.next() in Rust.

A creating write produces two notifications: Create, with no version range, and Write, with one. Advance cursors on Write only.

Shards are handled for you

A watch runs per shard, and your scope often spans several. The client probes with one connection; if the server answers 9001 (scope spans shards) or 9002 (filters do not match the routing rule), it opens one connection per shard and merges them behind NextAsync.

Only a raw protocol client sees those errors. There, a watch without a shard_id must route to exactly one shard, so the scope has to name the routing key: an Orgs filter when routed by org_id, AggregateTypes by aggregate_type_id, Aggregates by aggregate_id.

Latency and limits

RequestedLatency is how much coalescing you tolerate. Leave it null and the server flushes each change as it lands. Set it and the server merges bursts inside that window into fewer notifications. Merging widens the version range; it never drops a change, because ToAggregateVersion only advances and you re-read from your cursor anyway.

Limit	Flag	Default	Error
Max RequestedLatency	--max-requested-latency-ms	2000	8001 LatencyTooHigh
Watch subscriptions per shard	--max-watch-subscribers	16384	8005 WatchTooManySubscribers

Both come back as WatchErrorException from WatchAsync.

A watch ends, and never reconnects

A watch dies with its connection: a leader failover, a network drop, or falling behind. A subscriber whose buffer fills is cut off by the server rather than skipped past, so a slow consumer sees its watch fail, not a silent gap.

The client does not reconnect for you. A fresh subscription starts at the node's current tip, so anything written while you were disconnected produces no notification. Recovery is the same routine as a cold start: subscribe again, then catch up from your cursor.

Subscribe, catch up, follow

A watch covers the live tail, not the past. The order that leaves no gap:

1. Subscribe. Everything written from here on is queued on the watch.
2. Catch up from your cursor with a read. This covers everything before the subscribe.
3. Follow the watch, skipping notifications your catch-up already covered.

Reverse steps 1 and 2 and a write that lands between the end of the read and the subscribe reaches neither path. It sits in the log, un-notified, until some later write to the same aggregate happens to drag it in.

long cursor = LoadCheckpoint(key);   // last aggregate version you applied

// 1. subscribe
await using var watch = await pool.WatchAsync(new WatchRequest
{
    Aggregates = [key.AggregateId],
    OperationTypes = [WatchOperationType.Write],
});

// 2. catch up
await Drain();

// 3. follow
while (true)
{
    var response = await watch.NextAsync(ct);
    foreach (var change in response.Events)
    {
        if (change.ToAggregateVersion <= cursor) continue;   // catch-up already applied it
        await Drain();
    }
}

async Task Drain()
{
    await foreach (var batch in pool.ReadAllAsync(key, ReadFilters.From(cursor + 1), ct))
    {
        foreach (var e in batch.Events) Apply(e);
        cursor = batch.AggregateVersion;
        SaveCheckpoint(key, cursor);
    }
}

The overlap this order creates is the easy problem. Writes that land during catch-up are both read and notified; ToAggregateVersion at or below your cursor means you already have them. Reading by cursor rather than trusting the notification's range also makes coalesced or repeated notifications harmless: the re-read finds nothing new.

ReadAllAsync throws AggregateNotFoundException for an aggregate with no events yet. On the catch-up, that just means nothing to catch up.

Watching a scope, not a key

A cursor per known aggregate misses one case: an aggregate whose first write landed while you were not subscribed. You hold no cursor for it, so catch-up never reads it, and its notification went nowhere.

Discover those by listing the same scope the watch names, after subscribing:

await using var watch = await pool.WatchAsync(new WatchRequest { AggregateTypes = [ordersType] });

await foreach (var agg in pool.ListAggregatesAsync(orgId, ordersType, ct: ct))
{
    if (agg.MaxAggregateVersion > CursorFor(agg.AggregateId))   // 0 for one you have never seen
        await Drain(new AggregateKey(agg.OrgId, agg.AggregateTypeId, agg.AggregateId));
}
// then follow, as above, draining whichever aggregate each notification names

The listing walks every aggregate in scope, so this is a start-up and reconnect cost, not a per-notification one. If your key set is fixed and known, skip it.

For turning this into a read model, see Building a read model.