cntryl-fitz

The Rust SDK for Fitz. The crate uses the cntryl_fitz import name, treats tokens and routes as opaque values, and supports TCP and binary WebSocket transports.

[dependencies]
cntryl-fitz = "0.2"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

use cntryl_fitz::{Client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::builder("tcp://127.0.0.1:4091", || async {
        // Fetch a fresh token here. The provider is called again on reconnect.
        Ok(std::env::var("FITZ_TOKEN").unwrap_or_default())
    })
    .build()?;

    client.connect().await?;
    println!("state: {:?}", client.state());
    client.close().await?;
    Ok(())
}

For a broker that permits anonymous sessions, use Client::anonymous("ws://127.0.0.1:4190/ws").

The canonical protocol, acceptance criteria, and cross-language scenarios live in the Fitz server repository under docs/clients. Production code in this crate never creates or inspects JWTs.

Subscription registrations

All domain operations are async. Domain accessors such as client.kv() and client.notice() return typed clients, while subscription and RPC worker handles implement futures_core::Stream. Wire registration IDs remain private and are replaced transparently after reconnect. Dropping a one-shot future is cancellation-safe; slow bounded streams terminate with a typed backpressure error instead of stalling the receive loop.

KV, Queue, Stream, Notice, RPC worker, and Schedule registrations accept exact routes and whole-segment * or ** patterns, including wildcard realms. KV, Queue, and Stream patterns must be capable of matching three segments; Schedule patterns must match four; Notice and RPC have flexible depth. The broker permits 128 wildcard registrations per domain and session, while exact registrations do not consume the quota. Lease subscriptions accept only an exact lease://realm/area/resource route.

Notifications expose the exact concrete route. Queue availability notifications additionally report ready, delayed, and inflight message counts. Queue reserves accept general whole-segment patterns capable of matching three segments. Stream READ and SUBSCRIBE accept the complete documented selector matrix; Stream LAST is concrete-route only. Each returned QueueItem and StreamReadItem exposes the concrete matched route, including StreamRecord::route for event records. Route-less reserve/read/last responses are not supported. If any item contains an invalid concrete route, the entire response fails closed; the client never returns a partial reservation or read batch.

Schedule listing uses list(offset, limit) on message 702 and returns total_count. Global stream continuations reuse the returned fingerprint and captured-watermark pair.

Local broker

docker compose up -d

This starts ghcr.io/cntryl/fitz:latest as an authenticated broker on 127.0.0.1:4090/4091 and an anonymous broker on 127.0.0.1:4190/4191. Both are loopback-only and use local storage volumes. The development JWT secret is dev-test-secret, the audience is fitz, and tenant dev maps to identity 1.

Run the broker-backed tests explicitly:

cargo test --test conformance -- --ignored
CONFORMANCE_TRANSPORT=tcp CONFORMANCE_AUTH_MODE=anonymous cargo test --test conformance -- --ignored --nocapture
docker compose down --volumes

The broker-backed tests cover KV, Queue, RPC, Lease, Notice, Stream, and Schedule lifecycles. The conformance runner covers shared scenarios CS-001 through CS-017, including a real relay-induced transport loss that must recover on the same Client instance.

Managed leases

LeaseClient::with_lease and with_lease_with_options supervise acquisition, renewal, callback cancellation, and release without blocking Tokio executor threads. Their original one-argument callback shape remains supported. Use with_lease_authority or with_lease_authority_with_options when application code also needs the broker-issued admission fence:

# use cntryl_fitz::Client;
# async fn run(client: &Client) {
client
    .lease()
    .expect("lease client")
    .with_lease_authority(
        "lease://my-realm/locks/leader",
        "worker-1",
        30,
        |cancellation, authority| async move {
            while !cancellation.is_cancelled() {
                perform_fenced_step(authority.fencing_token).await?;
            }
            Ok::<(), &'static str>(())
        },
    )
    .await
    .expect("managed lease");
# }
# async fn perform_fenced_step(_fencing_token: u64) -> Result<(), &'static str> { Ok(()) }

LeaseAuthority::fencing_token is copied from the final successful ACQUIRE response and stays fixed for that callback even when managed renewal rotates the handle's live credential. Tokens are ordered only across successive ownership of the same exact lease route. External stores should atomically retain the greatest accepted token and reject lower values; do not compare the admission snapshot for equality with a later live broker token. Low-level lease handles rotate their private live credential after every successful extend and become stale after an uncertain renewal.

Development

cargo fmt --all -- --check
cargo clippy --locked --workspace --all-targets --all-features -- -D warnings -D clippy::pedantic
cargo test --locked --workspace --all-targets --all-features
RUSTDOCFLAGS="-D warnings" cargo doc --no-deps
cargo package --allow-dirty

Licensed under Apache-2.0.