.NET client

Celeriant.Client on NuGet, currently 0.8.0. It targets net8.0, net9.0 and net10.0, and depends on Celeriant.Transport (same version), MessagePack and ZstdSharp.Port.

dotnet add package Celeriant.Client

Your app holds one CeleriantPool and shares it. CeleriantClient is a single connection with no failover; use it for scripts and tests.

Connect

using Celeriant.Client;

await using var pool = new CeleriantPool(new CeleriantPoolOptions
{
    Address       = "node1:10000",
    SeedAddresses = ["node2:10000", "node3:10000"],
});

Option	Default	Notes
`Address`	required	First leader candidate.
`SeedAddresses`	none	Other nodes, for failover and follower reads. Nodes named in leader redirects are added as they appear.
`MaxConnections`	10	Per node, not per pool.
`ConnectionTimeout`	5 s	Bounds the dial and TLS handshake. A watch connect spends it again on Identify and again on the subscribe ack.
`RequestTimeout`	30 s	Per request. In 0.8.0 it also bounds a pooled connection's Identify.
`MaxRequestSize`	10_000_000	Below the server's 16 MiB `--max-request-size`. In 0.8.0 it is checked against the compressed length only. See below.
`MaxResponseSize`	64 MiB	One response page, matching the server's `--max-response-size`. Set it lower and a read whose first page is bigger throws `ProtocolException`, with no smaller page to fall back to.
`IdleTimeout`	25 s	Must stay below the server's `--client-connection-timeout-ms` (30 s).
`RouteReadsToFollowers`	false	Opt in to follower reads.
`TlsConfig`	null	Plain TCP when null.
`IdentityConfig`	null	No Identify handshake when null, which also means no compression.

Why IdleTimeout sits under 30 s. The server waits at most --client-connection-timeout-ms for the next request on a connection, then closes it. The pool evicts idle connections lazily, on checkout. If the server's clock runs out first, the next checkout gets a dead socket and a ConnectionFailedException. Raise one, raise the other.

MaxRequestSize and compression. The server rejects a frame whose compressed or uncompressed length exceeds --max-request-size. The 0.8.0 client checks only the length after compression, so a payload that compresses under 10 MB but expands past 16 MiB leaves the client and dies at the server as a dropped connection. 0.9.0 checks the uncompressed length first. An oversized request throws ArgumentException, not a Celeriant exception.

Compression needs IdentityConfig. The server's zstd dictionary arrives in the Identify response. With no identity configured the pool never identifies, and every request goes out uncompressed. With one, writes whose event payloads total at least 1 KiB, and schema registrations that size, are compressed with the dictionary, and compressed responses are decompressed. There is no flag to set.

Routing

Writes, deletes, trims and schema registration go to the leader. A NotLeader answer carries the leader's address; the pool switches to it and resends, and the caller never sees it.
Reads, aggregate details, lists and watches also go to the leader, so a read sees your own writes.
RouteReadsToFollowers = true rotates reads across followers, gives up read-your-writes, and keeps the leader as last resort when every follower fails.

A crashed leader is different from a leader that stepped down. Until a new one is elected there is nobody to redirect to, and leader operations throw ConnectionFailedException. Retry with backoff. The pool does not wait out the election for you.

Dependency injection

builder.Services.AddCeleriantPool(options =>
{
    options.Address        = "node1:10000";
    options.SeedAddresses  = ["node2:10000", "node3:10000"];
    options.IdentityConfig = ClientIdentityConfig.FromApiKey(apiKeyBase64);
});

This registers ICeleriantPool as a singleton. The delegate configures a CeleriantPoolOptionsBuilder, which has the same properties and defaults as CeleriantPoolOptions but mutable; Build() throws InvalidOperationException when Address is empty. Inject ICeleriantPool.

Operations

All on ICeleriantPool. Every method also takes a trailing CancellationToken ct = default.

Method	Returns	Routed to
`WriteAsync(WriteRequest)`	`Task<WriteResponse>`	leader
`WriteAsync(AggregateKey key, AggregateEvent[] events, Guid clientId, bool allowCreate = true, long? expectedVersion = null, bool enforceClientIdempotency = false)`	`Task<WriteResponse>`	leader
`ReadAsync(ReadRequest)`	`Task<ReadResponse>`, one page	read
`ReadAllAsync(AggregateKey key, ReadFilters? filters = null)`	`IAsyncEnumerable<AggregateEventBatch>`	read
`AggregateDetailsAsync(AggregateDetailsRequest)`	`Task<AggregateDetailsResponse>`	read
`DeleteAsync(DeleteRequest)`	`Task<SuccessResponse>`	leader
`TrimStartAsync(TrimStartRequest)`	`Task<SuccessResponse>`	leader
`RegisterSchemaAsync(RegisterSchemaRequest)`	`Task<SuccessResponse>`	leader
`ListOrgsAsync(ListOptions? options = null)`	`IAsyncEnumerable<OrgListItem>`	read
`ListAggregateTypesAsync(Guid? orgId = null, ListOptions? options = null)`	`IAsyncEnumerable<AggregateTypeListItem>`	read
`ListAggregatesAsync(Guid? orgId = null, Guid? aggregateTypeId = null, ListOptions? options = null)`	`IAsyncEnumerable<AggregateStats>`	read
`WatchAsync(WatchRequest request, WatchOptions? options = null)`	`Task<WatchConnection>`	read
`GetConnectionAsync()`	`Task<PooledConnection>`	read

ReadAllAsync and the list methods lease one connection for the whole enumeration and follow the server's cursors across pages and shards. ListOptions.IncludeDeleted includes deleted aggregates.

A worked write and read is in the Quickstart; the Guides cover each operation as a recipe.

Writing

clientId is required on every write, and the client never invents one. Keep it stable per logical writer; a fleet of replicas is one writer with one id (see scaling out).

enforceClientIdempotency defaults to false. Without it the server does not dedupe on ClientSeq, and a retried write lands twice. Turn it on for anything you retry. See the idempotency guide.

Events are AggregateEvents. Build them from domain objects with a serializer:

var serializer = JsonEventSerializer.Default;   // System.Text.Json

var evt = AggregateEventExtensions.Create(
    eventTypeMajor: 1, new Deposited(amountCents), serializer,
    clientSeq: nextSeq, eventId: requestId);

var back = evt.GetValue<Deposited>(serializer);

new JsonEventSerializer(options) takes your own JsonSerializerOptions. For another format, implement IEventSerializer (byte[] Serialize<T>(T value) and T Deserialize<T>(ReadOnlySpan<byte> data)), or set EventValue bytes directly.

Create does not number events; clientSeq defaults to 1. Several events in one write need distinct seqs. The client throws ArgumentException before sending for a duplicate seq when idempotency is enforced, a null EventValue, EventTypeMajor 0, a timestamp before 1970, an empty event list, or a negative ExpectedVersion.

Watch

await using var watch = await pool.WatchAsync(new WatchRequest
{
    Orgs             = [orgId],
    OperationTypes   = [WatchOperationType.Write, WatchOperationType.Create],
    RequestedLatency = TimeSpan.FromMilliseconds(100),
});

while (!ct.IsCancellationRequested)
{
    var resp = await watch.NextAsync(TimeSpan.FromSeconds(30), ct);
    if (resp is null)
    {
        if (watch.IsDesynchronised) break;   // dispose and reconnect
        continue;                            // idle window, not the end
    }
    foreach (var e in resp.Events) { /* e.AggregateId, e.Operation, e.ToAggregateVersion */ }
}

A watch gets its own connection, outside the pool; dispose it. It tells you what changed, not the events: re-read the aggregate from your cursor. It never reconnects and never replays a gap. When it fails, NextAsync throws, and a new subscription starts from that node's tip. WatchConnection.Address tells you which node you were on, so you notice when a reconnect moved.

Scope with Orgs, AggregateTypes, Aggregates (bare aggregate ids) and OperationTypes. A scope that spans shards fans out one connection per shard behind the same WatchConnection. RequestedLatency above the server's --max-requested-latency-ms (2000 by default) fails with WatchErrorException (8001). The ordering that avoids gaps is in Watch.

TLS and identity

var options = new CeleriantPoolOptions
{
    Address        = "node1:10010",
    TlsConfig      = ClientTlsConfig.WithClientCertificateFromPem("node1", "client.crt", "client.key"),
    IdentityConfig = ClientIdentityConfig.FromRsaKeyPair(publicKeyBase64, privateKeyBase64),
};

ClientTlsConfig factories:

Create(targetHost): server-only TLS.
WithClientCertificate(targetHost, X509Certificate2): mTLS with a certificate that holds its key.
WithClientCertificate(targetHost, X509Certificate2, AsymmetricAlgorithm): mTLS with the key elsewhere, a KMS or HSM behind an RSA or ECDsa subclass.
WithClientCertificateFromPem(targetHost, certPath, keyPath): mTLS from PEM files.
FromSslOptions(SslClientAuthenticationOptions): everything else.

ClientIdentityConfig factories:

FromApiKey(base64Key): a 32-byte key. Sets the connection's access level. It is not a client id.
FromRsaKeyPair(publicKeyBase64, privateKeyBase64): DER SubjectPublicKeyInfo and DER PKCS#8, base64. Proves a client id derived from the public key. Get that id from CeleriantCrypto.GenerateClientIdentity(publicKeyBase64) and use it as your write clientId; the server rejects a mismatch with 10003.
FromClientId(Guid): sends the Guid's 16 bytes in the API key field. That proves nothing. A server with no api_keys.toml ignores the field, and one with keys rejects it with 10006, because a key must be 32 bytes.

Set an API key and a key pair on one config and only the API key is sent. Keys must be RSA-2048; anything larger overflows the fixed-size Identify frame, the server drops the socket, and every connect fails with ConnectionFailedException (see Clients overview). What the server actually enforces is in Identity and authentication.

Errors

Everything the client throws derives from CeleriantClientException:

Type	Base	What it means
`CeleriantErrorException`	`CeleriantClientException`	A server error response. `Error.ErrorCode` holds the code; unmapped codes arrive as this type.
`WriteOccException`	`WriteErrorException`	2003. `ExpectedVersion`, `CurrentAggregateVersion`.
`IdempotencyViolationException`	`WriteErrorException`	2002. The seq landed durably, maybe not yours.
`InflightDuplicateWriteException`	`WriteErrorException`	2013. The seq is in flight, not yet durable. Hold it and retry.
`AggregateRecreateNotAllowedException`	`WriteErrorException`	The aggregate was deleted for good.
`AggregateNotFoundException`	`CeleriantErrorException`	Missing aggregate, on read, write, trim, delete or details.
`BatchIndexUnavailableException`	`ReadErrorException`	1000. Trimmed; re-read from `MinimumAvailableVersion`.
`DeleteOccException`	`DeleteErrorException`	Expected version mismatch on delete.
`TrimIndexOutOfRangeException`	`TrimErrorException`	Trim past the end.
`SchemaValidationException`	`SchemaErrorException`	2022. `FailedEventIndex` is the position in your events array.
`WatchErrorException`	`CeleriantErrorException`	Watch refused: 8000 (invalid request), 8001 (latency too high), 8005 (too many subscribers).
`AuthErrorException`	`CeleriantErrorException`	Identify and API-key rejections.
`ShardRoutingException`	`CeleriantErrorException`	A multi-aggregate write spans shards.
`ServerInternalErrorException`	`CeleriantErrorException`	Server-side IO, fsync or replication failure.
`NotLeaderException`	`CeleriantClientException`	Absorbed by the pool while it can follow the redirect.
`ServerBusyException`	`CeleriantClientException`	Backpressure. The pool moves a leader operation to the next known node; with none left, it throws. Back off.
`IdentityRequiredException`	`CeleriantClientException`	10004.
`ConnectionFailedException`	`CeleriantClientException`	No node answered.
`CeleriantTimeoutException`	`CeleriantClientException`	A request timed out.
`ConnectionTimeoutException`	`CeleriantTimeoutException`	Dial or TLS handshake timed out.
`ProtocolException`	`CeleriantClientException`	A malformed or mismatched response; the connection is retired.

NotLeaderException, ServerBusyException and IdentityRequiredException are server responses but do not derive from CeleriantErrorException. Catch CeleriantClientException to get everything. Codes and meanings: error codes reference.

try { await pool.WriteAsync(key, events, clientId, expectedVersion: tip, enforceClientIdempotency: true); }
catch (WriteOccException)                { /* 2003: re-read, re-decide, new seq */ }
catch (IdempotencyViolationException)    { /* 2002: find out whose event holds the seq, below */ }
catch (InflightDuplicateWriteException)  { /* 2013: same seq, retry after a short backoff */ }
catch (SchemaValidationException)        { /* 2022: fix the payload; retrying cannot help */ }

0.8.0 can resend a write that already landed

In 0.8.0, a request timeout or a dropped connection on a leader operation makes the pool try the next known node, even when the request was already sent. With more than one known node, a write that already landed is sent again. What you see depends on the guard. With expectedVersion the resend fails 2003 against your own write, so a naive OCC loop re-decides on state it already changed. With a stable clientId, enforceClientIdempotency: true and a held ClientSeq it fails 2002, which you resolve below. With neither, the duplicate lands. 0.9.0 walks only on failures before the request is sent; a loss after sending is RequestOutcomeUnknownException and is never resent.

What changes in 0.9.0

0.9.0 is not on NuGet yet. It splits the ambiguous failures so you can tell a safe retry from an unsafe one:

PoolUnavailableException (Address, Reason): this process's own pool refused before contacting any node, because the node's circuit breaker is open after a failed dial or the pool was disposed. Always safe to retry. A write gets it straight back.
RequestOutcomeUnknownException: the request was fully written and no answer came back. The node may have applied it. The pool never resends it. Make the write idempotent and retry, or read back and decide.
An exhausted leader walk throws ConnectionFailedException("No leader found; last attempt to <address> failed: ...") with the last error as InnerException, instead of rethrowing NotLeaderException.
An Identify timeout is ConnectionTimeoutException, not CeleriantTimeoutException.
MaxRequestSize is checked on the uncompressed length before compression. A leader operation over the cap throws ArgumentException before any node is dialled.
A negative ListOptions.StartShard throws ArgumentOutOfRangeException at the call, instead of returning an empty stream.
WatchOptions gains MaxRequestSize and MaxResponseSize, and WatchAsync fills them from the pool options. In 0.8.0 a pooled watch runs at the client defaults whatever the pool says.

Resolving a 2002

A 2002 proves the sequence was consumed, not that your event consumed it. With concurrent requests sharing one client id, a sibling may have taken the number first. The stream settles it: point-read the contested sequence and compare the EventId.

enum SeqOwnership { Unwritten, Ours, Sibling }

async Task<SeqOwnership> WhoOwnsSeq(
    Guid orgId, Guid aggregateTypeId, Guid accountId,
    Guid serviceClientId, long clientSeq, Guid requestId)
{
    var resp = await pool.ReadAsync(new ReadRequest
    {
        AggregateKey = new AggregateKey(orgId, aggregateTypeId, accountId),
        Filters = ReadFilters.From(1) with
        {
            MinClientSeq    = clientSeq,
            MaxClientSeq    = clientSeq,
            IncludeClientId = serviceClientId,
        },
    });
    var evt = resp.EventBatches.SelectMany(b => b.Events)
        .FirstOrDefault(e => e.ClientSeq == clientSeq);
    if (evt is null) return SeqOwnership.Unwritten;
    return evt.EventId == requestId ? SeqOwnership.Ours : SeqOwnership.Sibling;
}

The client id and sequence filters are checked against each batch's metadata, so the server skips every other batch without reading its events. Yours means the earlier attempt landed: success. A sibling's means yours never landed: take a fresh sequence and write again. The retry loop this slots into is in the idempotency guide. Celeriant.Reference in the .NET repo has a working version in Verify.cs.

Connect​

Routing​

Dependency injection​

Operations​

Writing​

Watch​

TLS and identity​

Errors​

What changes in 0.9.0​

Resolving a 2002​