Rust client

celeriant_client_tokio is the client for async Rust services. Version 0.2.0 is on crates.io. AggregateKey lives in celeriant_wal and the request and response types in celeriant_msg, so add those too:

cargo add celeriant_client_tokio celeriant_msg celeriant_wal

The workspace also has celeriant_client_glommio. That is the server's own node-to-node client for replication traffic. It is not published and not something to build an application on.

Connect

CeleriantPool is the entry point. Build it from a PoolOptions; it dials nothing until the first operation. The pool is not Clone, so wrap it in an Arc to share it across tasks.

use celeriant_client_tokio::{CeleriantPool, PoolOptions};
use std::time::Duration;

let pool = CeleriantPool::new(
    PoolOptions::new("node1:10000")
        .with_seed_addresses(vec!["node2:10000".into()])
        .with_request_timeout(Duration::from_secs(10)),
);

Builder	Default	Notes
with_seed_addresses	none	Extra nodes for failover and follower reads.
with_max_connections	10	Per node.
with_connection_timeout	5 s	TCP connect plus the TLS handshake. Also the default dial timeout for a watch.
with_request_timeout	30 s	Per request.
with_max_request_size	10 MB (10,000,000 bytes)	Checked on the uncompressed body, before anything is sent.
with_max_response_size	64 MiB	Matches the server's --max-response-size.
with_idle_timeout	25 s	Keep it below the server's 30 s --client-connection-timeout-ms.
with_route_reads_to_followers	false	See routing below.
with_max_leader_retries	3	Hops a write may take hunting for the leader.
with_tls / with_identity	none	See TLS and identity.

The timeouts match the .NET client. The request size cap does not match the server: the server accepts up to 16 MiB (--max-request-size), so a default client refuses writes between 10 MB and 16 MiB the server would take. The refusal is ClientError::WireError(MessageTooLarge), raised before a byte leaves. Raise the cap with with_max_request_size if your events are that big.

Operations

Most calls take a request struct from celeriant_msg::request::requests and return Result<_, ClientError>. The trait CeleriantPoolApi covers the ones you would mock:

pool.read(ReadRequest).await
pool.write(WriteRequest).await
pool.delete(DeleteRequest).await
pool.trim_start(TrimStartRequest).await
pool.aggregate_details(AggregateDetailsRequest).await
pool.register_schema(RegisterSchemaRequest).await
pool.watch(WatchRequest, WatchOptions).await                          // -> WatchConnection
pool.write_events(AggregateKey, Vec<DatablockAggregateEvent>, client_id).await
pool.write_events_with(AggregateKey, events, client_id, WriteEventsOptions).await

The streaming calls live on CeleriantPool only:

pool.read_all(AggregateKey, Option<ReadFilters>).await                // pages to the tip
pool.list_orgs(ListOptions).await
pool.list_aggregate_types(Option<u128>, ListOptions).await
pool.list_aggregates(Option<u128>, Option<u128>, ListOptions).await

Each returns an iterator with async fn next() and collect(). The iterator holds one pooled connection for its whole life, so every page comes from the node that answered the first.

A single-aggregate write with the correctness controls on:

use celeriant_client_tokio::{json_event, WriteEventsOptions};
use celeriant_wal::aggregate_key::AggregateKey;

let key = AggregateKey::new(org_id, order_type_id, order_id);
let mut event = json_event(1, &order_placed)?;
event.client_seq = next_seq;

pool.write_events_with(key, vec![event], client_id, WriteEventsOptions {
    allow_create: false,
    expected_version: Some(4),         // optimistic concurrency
    enforce_client_idempotency: true,  // replayed client_seq is rejected, not appended
}).await?;

WriteEventsOptions::default() (what write_events uses) is allow_create: true, no expected version, idempotency off. client_id is a u128 and every write takes it explicitly; the client never invents one. Keep it stable per logical writer, see idempotency. IDs are u128 wherever .NET uses Guid.

Routing

Writes (write, delete, trim, register schema) start at the cached leader. A NotLeader answer with a leader address jumps to that node; without one the pool tries the next seed. Hops share the max_leader_retries budget. When the budget or the seed list runs out, the pool returns ConnectionFailed naming the last node and its error, not NotLeader.

Once a write request is on the wire, the pool never re-sends it. ConnectionLostAfterSend, RequestTimeout, and wire or read errors go straight back to you, because the node may have applied the request.

Reads (read, aggregate details, the streaming calls, watch) are pinned to the leader for read-your-writes. with_route_reads_to_followers(true) rotates across followers instead and uses the leader last, if every follower fails. Followers can lag the leader, so you give up read-your-writes. A read is safe to retry, so a connection that dies mid-read moves on to the next candidate.

Errors

ClientError is #[non_exhaustive]. The variants that change what you do next:

Variant	Meaning
Server(ServerError)	The server answered with an error code. ServerError groups codes by operation: Write { kind: WriteError, .. }, Read, Delete, Trim, Schema, Watch, Details, Auth, plus List, Replication, ShardRouting and Unknown carrying the raw code.
ConnectionLostAfterSend	The request left and no answer came back. Outcome unknown.
RequestTimeout	Same ambiguity: the node may have applied it.
PoolTimeout { address }	No pool slot freed up. Nothing was sent; retrying is safe.
ConnectionFailed / ConnectionTimeout	Never reached a node, or the leader walk ran out.
ServerBusy	Load shed. On a leader-pinned read it comes straight back; on a write the pool moves on to the next candidate.
NotLeader	Only seen from a bare CeleriantClient; the pool handles it.
IdentityRequired	The server wants an Identify first (10004). Configure with_identity.

use celeriant_client_tokio::{ClientError, ServerError, WriteError};

match pool.write(request).await {
    Ok(_) => {}
    Err(ClientError::Server(ServerError::Write { kind: WriteError::OptimisticConcurrencyViolation { .. }, .. })) => {
        // someone else wrote first: re-read, re-decide
    }
    Err(ClientError::Server(ServerError::Write { kind: WriteError::ClientIdempotencyViolation { .. }, .. })) => {
        // the client_seq is consumed; find out by whom before doing anything
    }
    Err(ClientError::Server(ServerError::Write { kind: WriteError::InflightDuplicateWrite { .. }, .. })) => {
        // fsynced, replication not yet confirmed: retry unchanged
    }
    Err(ClientError::ConnectionLostAfterSend(_) | ClientError::RequestTimeout) => {
        // outcome unknown: retry with the same client_seq
    }
    Err(e) => return Err(e.into()),
}

Deletes carry their own DeleteError::OptimisticConcurrencyViolation. The codes are in the error codes reference.

TLS and identity

use celeriant_client_tokio::{ClientIdentityConfig, ClientTlsConfig};
use std::path::Path;

let tls = ClientTlsConfig::from_paths(
    Path::new("ca.crt"),
    Some((Path::new("client.crt"), Path::new("client.key"))),  // None for plain TLS
    "node1.example",                                            // SNI
)?;

let pool = CeleriantPool::new(
    PoolOptions::new("node1.example:10000")
        .with_tls(tls)
        .with_identity(ClientIdentityConfig::from_api_key(api_key_base64)),
);

from_paths builds a TLS 1.3 config from PEM files. If you already hold a rustls::ClientConfig, use ClientTlsConfig::new(Arc<ClientConfig>, ServerName). sni_host("host:port") pulls the host out of an address for the SNI.

Identity is either ClientIdentityConfig::from_api_key(key) or ::from_key_pair(public, private), where both keys are base64 DER strings (SPKI public, PKCS#8 private). The pool runs Identify on every new connection when an identity is set. Without one it skips Identify entirely, which also means the connection never learns the cluster's compression dictionary and sends requests uncompressed. See Identity and authentication.

Watch

pool.watch(request, WatchOptions::default()) opens a dedicated connection outside the pool, applying the pool's TLS, identity, and size limits. It dials the leader, or a follower when follower routing is on. A watch that spans shards opens one connection per shard and merges them into one stream.

let mut watch = pool.watch(request, WatchOptions::default()).await?;
loop {
    let resp = watch.next().await?;   // or next_timeout(d) -> Ok(None) on timeout
    for ev in resp.events {
        // ev.org_id, ev.aggregate_type_id, ev.aggregate_id, ev.operation,
        // ev.from_aggregate_version, ev.to_aggregate_version
    }
}

It does not reconnect. Once a multi-shard watch returns an error, every later next() returns ProtocolError; drop it, subscribe again, then catch up with a read. Watch and subscribe covers the order that leaves no gap.