Rust client

tephra-client is the reference client. It ships a blocking Client and, behind the async feature, a multiplexing AsyncClient. The snippets below are compiled and run against a real server as part of the site’s test suite.

cargo add tephra-client

Optional features: async (the AsyncClient), tls (TLS for the blocking client), and async-tls (TLS for the async client).

tephra-client = { version = "0.4", features = ["async", "tls"] }

Connect

Client::connect takes anything that resolves to a socket address and opens one TCP connection. A blocking Client carries a single request at a time and is not shared between threads: give each thread its own, or reach for the AsyncClient below.

let mut client = Client::connect(addr)?;

Append

An event is a type, a set of tags, and an opaque payload. append takes an iterator of events and an optional condition (None here, so the write is unconditional) and returns the position range the batch landed at.

let event = Event::new(
    "CourseOpened",
    ["course:c1"],
    br#"{"course":"c1","seats":30}"#.to_vec(),
)?;
let result = client.append([event], None)?;
println!("recorded positions {} to {}", result.first, result.last);

An AppendCondition guards the write as a dynamic consistency boundary. Decision models builds the read-fold-decide-append cycle, Handling conflicts covers the same-batch versus durable retry contract, and its optional fail_if_exists clause makes an append idempotent, returning ErrorCode::AlreadyExists on a duplicate (Idempotency).

Read

read and read_all take a query, an after position, and a limit. Position::ZERO reads from the start and a None limit reads the whole matching history, ascending.

let query = Query::item(QueryItem::with_tags(Tags::new([Tag::new("course:c1")?])?));
// `None` reads the whole matching history; pass `Some(n)` to cap the number of events.
let (events, _watermark) = client.read_all(query, Position::ZERO, None)?;
for seq in &events {
    println!("{} {}", seq.position(), seq.event().event_type());
}

Paginate

Pass Some(n) to cap a read, and pair it with after to page through a large result: each page resumes at the last position seen, with no gap and no duplicate at the seam.

// `limit` caps how many events a read returns. Pair it with `after` (an exclusive lower
// bound) to page through a large result: each page resumes at the last position seen, with
// no gap and no duplicate at the seam. A page shorter than the limit is the last one.
let query = Query::item(QueryItem::with_tags(Tags::new([Tag::new("course:c1")?])?));
let page_size = 2;
let mut after = Position::ZERO;
let mut seen = 0usize;
loop {
    let (page, _watermark) = client.read_all(query.clone(), after, Some(page_size))?;
    let short = (page.len() as u64) < page_size;
    if let Some(last) = page.last() {
        after = last.position();
    }
    seen += page.len();
    if short {
        break;
    }
}

Read newest-first

read_back and read_all_back are the newest-first duals. They take before (an exclusive upper bound), so Position::MAX starts at the tip.

// `read_back` / `read_all_back` are the newest-first duals of `read` / `read_all`: they
// return matching events in descending position order. `before` is an exclusive upper bound,
// so `Position::MAX` starts at the tip. Pair it with a `limit`, then pass the oldest position
// of a page back as the next `before`, to page an event explorer newest-first.
let query = Query::item(QueryItem::with_tags(Tags::new([Tag::new("course:c1")?])?));
let (newest, _watermark) = client.read_all_back(query, Position::MAX, Some(2))?;
let newest_positions: Vec<u64> = newest.iter().map(|seq| seq.position().get()).collect();

Server stats

stats returns a point-in-time snapshot of the server’s operational state.

// A point-in-time snapshot of the server: the durable event count (also the tip position),
// the on-disk size, uptime, and the live connection and subscription counts.
let stats = client.stats()?;
println!(
    "{} events across {} segments, {} bytes on disk",
    stats.event_count, stats.segment_count, stats.disk_bytes
);

Subscribe

A subscription catches up on history and then tails the live edge as one loop, yielding a CaughtUp marker each time it reaches the live edge.

// Subscribe from the start: catch up on history, then a CaughtUp marker at the live edge.
let query = Query::item(QueryItem::with_tags(Tags::new([Tag::new("course:c1")?])?));
let (mut stream, cancel) = client.subscribe(query, Position::ZERO)?;
for item in &mut stream {
    match item? {
        SubEvent::Event(seq) => {
            println!("live {} {}", seq.position(), seq.event().event_type());
        }
        // Reached the live edge. A long-running consumer keeps going; this example stops.
        SubEvent::CaughtUp(_) => break,
    }
}
cancel.cancel();

Async client

With the async feature, AsyncClient multiplexes many concurrent requests over one control socket plus a pool of bulk read sockets. Its methods take &self, so a single client drives concurrent work. It needs a Tokio runtime.

use tephra_client::{AsyncClient, Event, Position, Query};

let client = AsyncClient::connect("127.0.0.1:9000").await?;

// Both futures borrow the same client; the requests are multiplexed on one connection.
let (a, b) = tokio::join!(
    client.append([Event::new("A", ["k:1"], b"{}".to_vec())?], None),
    client.append([Event::new("B", ["k:2"], b"{}".to_vec())?], None),
);
a?;
b?;

let (events, _watermark) = client.read_all(Query::all(), Position::ZERO, None).await?;

TLS

With the tls feature, Client::connect_tls verifies the server certificate. Build the client config from the system roots, or from a custom CA for a self-signed certificate.

use tephra_client::{Client, tls};

// Verify against the system roots (a public CA):
let config = tls::config_with_native_roots()?;
let mut client = Client::connect_tls("tephra.example.com:9000", "tephra.example.com", config)?;

// Or trust a private CA for a self-signed certificate:
let config = tls::config_with_custom_ca("ca.pem".as_ref())?;
let mut client = Client::connect_tls("tephra.internal:9000", "tephra.internal", config)?;

Authentication

When the server requires a bearer token, pass it to a *_with connect variant, so a rejected token fails the connect rather than the first request.

// Blocking, over TLS (the token should not cross an unencrypted hop):
let config = tls::config_with_native_roots()?;
let mut client = Client::connect_tls_with(
    "tephra.example.com:9000",
    "tephra.example.com",
    config,
    Some("a-long-random-secret"),
)?;

The async client carries the token on its config, and every socket in the control-plus-bulk pool authenticates independently.

use tephra_client::{AsyncClient, AsyncClientConfig};

let config = AsyncClientConfig {
    auth_token: Some("a-long-random-secret".into()),
    ..Default::default()
};
let client = AsyncClient::connect_tls_with(addr, server_name, tls_config, config).await?;

Errors

A method returns ClientError on failure. The Server variant carries the wire code, a message, a retryable flag (set for an advisory same-batch append conflict), and a conflict_position for a durable one; Protocol, UnexpectedEof, and Frame cover transport and framing failures. The client does no automatic retries. See Handling conflicts for the append retry loop.