Broker API Reference

The broker’s data-plane API: what each operation does on the wire, what it returns, and how it fails. Message shapes are shown as JSON control messages; the binary framing that carries the hot paths is in the wire protocol.

Connection Model

QUIC Connection Lifecycle

Clients establish QUIC connections to the broker over TLS 1.3:

sequenceDiagram
    participant C as Client
    participant B as Broker
    
    C->>B: QUIC ClientHello
    B-->>C: QUIC ServerHello + TLS Certificate
    C->>B: TLS Finished
    Note over C,B: Connection established
    C->>B: Open streams for operations

Default broker endpoint: 0.0.0.0:5000 (configurable via quic_bind)

The connection uses TLS 1.3, and the client always verifies the broker’s certificate. There is no switch to turn verification off.

Connection Pooling

For optimal performance, clients should maintain connection pools:

// Rust client example
use felix_client::{ClientConfig, ClusterClient};
use std::net::SocketAddr;

let quinn = quinn::ClientConfig::with_platform_verifier();
let config = ClientConfig {
    event_conn_pool: 8,      // Pool for pub/sub operations
    cache_conn_pool: 8,      // Pool for cache operations
    ..ClientConfig::optimized_defaults(quinn)
};

let seeds: Vec<SocketAddr> = vec!["127.0.0.1:5000".parse()?];
let client = ClusterClient::connect(&seeds, "localhost", config).await?;

Connection setup costs a TLS handshake, so the client keeps long-lived pools and the hot paths never pay it. Pool sizes are workload-dependent; start with the defaults and resize off a measurement.

Authentication (Felix Tokens)

Brokers require a tenant-scoped Felix token for authorization. Tokens are obtained from the control plane using an upstream OIDC JWT.

End-to-end flow:

Obtain an OIDC token from your identity provider.
Exchange it for a Felix token via the control plane.
Connect to the broker and present tenant_id + felix_token in the auth frame.

Broker validation:

Verifies the token signature using tenant JWKS from the control plane.
Checks iss = felix-auth, aud = felix-broker, exp/nbf, and tid matches the connection tenant.
Parses perms once and enforces per operation using wildcard matching.

If authentication fails, the broker rejects the connection or returns an unauthorized error for the operation.

Publish Operations

The JSON publish and publish_batch messages below are the compatibility path. Current clients publish with binary frames and fall back to JSON only against a broker that did not advertise the frame they need. The binary forms also carry a routing key, a request id for acks, and an idempotent producer’s sequence. See Binary PublishBatch and the keyed, acked and idempotent variants after it in docs/protocol.md.

Single Message Publish

Publish a single message to a stream.

Request:

{
  "type": "publish",
  "tenant_id": "string",
  "namespace": "string",
  "stream": "string",
  "payload": "base64-encoded-bytes",
  "key": "base64-encoded-bytes",
  "request_id": 1,
  "ack": "none" | "per_message"
}

Parameters:

Parameter	Type	Required	Description
`tenant_id`	string	Yes	Tenant identifier (must exist in broker)
`namespace`	string	Yes	Namespace within tenant
`stream`	string	Yes	Target stream name
`payload`	base64	Yes	Message payload (base64-encoded binary)
`key`	base64	No	Routing key; picks the shard on a sharded stream
`request_id`	u64	With an ack	Echoed in the answer
`ack`	enum	No	Acknowledgement mode (default: `none`)

With ack: "none" the broker sends nothing back. With per_message it answers once the publish is committed, or with the reason it was refused:

{ "type": "publish_ok", "request_id": 1 }

{
  "type": "publish_error",
  "request_id": 1,
  "message": "stream not found: tenant=acme namespace=prod stream=events",
  "code": "not_found",
  "retry": "retry_after",
  "detail": null
}

code, retry and detail are sent only to a client that offered FEATURE_ERROR_CODES. See Error Handling.

Example usage:

use felix_client::{AckMode, ClientConfig, ClusterClient};
use std::net::SocketAddr;

let quinn = quinn::ClientConfig::with_platform_verifier();
let config = ClientConfig::optimized_defaults(quinn);
let seeds: Vec<SocketAddr> = vec!["127.0.0.1:5000".parse()?];
let client = ClusterClient::connect(&seeds, "localhost", config).await?;

// Fire-and-forget publish
client
    .publish("acme", "prod", "events", b"Hello Felix".to_vec(), AckMode::None)
    .await?;

// With acknowledgement
client
    .publish(
        "acme",
        "prod",
        "events",
        b"Important message".to_vec(),
        AckMode::PerMessage,
    )
    .await?;

Every message costs a frame of its own, so batch when throughput matters. Measured numbers are on the Benchmarks page.

Batch Publish

Publish multiple messages in a single operation.

Request:

{
  "type": "publish_batch",
  "tenant_id": "string",
  "namespace": "string",
  "stream": "string",
  "payloads": ["base64-1", "base64-2", "base64-n"],
  "ack": "none" | "per_batch"
}

Parameters:

Parameter	Type	Required	Description
`tenant_id`	string	Yes	Tenant identifier
`namespace`	string	Yes	Namespace within tenant
`stream`	string	Yes	Target stream name
`payloads`	array	Yes	Array of base64-encoded payloads
`ack`	enum	No	Acknowledgement mode (default: `none`)

With ack: "per_batch" and a request_id, the answer is one publish_ok or publish_error for the whole batch, shaped as above.

Example usage:

let messages = vec![
    b"Event 1".to_vec(),
    b"Event 2".to_vec(),
    b"Event 3".to_vec(),
];

use felix_client::AckMode;
client
    .publish_batch("acme", "prod", "events", messages, AckMode::PerBatch)
    .await?;

Larger batches raise throughput and add latency. See Benchmarks for measured trade-offs.

Binary Batch Publish

For maximum throughput, use binary encoding.

Frame flags: Set bit 0 (flags | 0x0001)

Binary format:

[tenant_len: u16][tenant_id: bytes]
[namespace_len: u16][namespace: bytes]
[stream_len: u16][stream: bytes]
[count: u32]
[payload_1_len: u32][payload_1: bytes]
[payload_2_len: u32][payload_2: bytes]
...

Example (Rust client handles encoding automatically; Publisher is the single-connection Client’s API, and applications should use ClusterClient, whose publish_batch sends this frame):

// Encode a batch directly as binary
let messages = vec![large_payload_1, large_payload_2, /* ... */];
let publisher = client.publisher().await?;
publisher
    .publish_batch_binary("acme", "prod", "stream", &messages)
    .await?;

The Rust client uses this frame for every publish once the broker advertises it, so publish and publish_batch already take this path.

Conditional Publish

Appends a batch only if it would start at exactly expected_offset, the shard’s next offset. Sent only to a broker that advertised FEATURE_PUBLISH_CONDITIONAL.

{
  "type": "publish_if",
  "tenant_id": "acme",
  "namespace": "games",
  "stream": "match-7",
  "payloads": ["<base64>"],
  "expected_offset": 120,
  "request_id": 9
}

key is optional and routes the batch like publish_batch. Written, the answer is publish_ok with offset, once the batch is durable. Refused, it is publish_refused with {"offset_mismatch": {"tail": <u64>}}, and nothing was written. A broker that does not lead the shard answers not_leader. commit takes the same optional expected_offset.

use felix_client::ConditionalWrite;

match client
    .publish_if("acme", "games", "match-7", None, vec![b"tick".to_vec()], 120)
    .await?
{
    ConditionalWrite::Written { offset } => assert_eq!(offset, 120),
    ConditionalWrite::Refused { tail } => println!("the shard is at {tail}"),
}

Publish Pipeline Configuration

Broker-side tuning for publish pipeline:

# Broker config.yml
pub_workers_per_conn: 4        # Workers per connection
pub_queue_depth: 512           # Publish queue bound
publish_queue_wait_timeout_ms: 2000  # Queue full timeout

Worker sizing:

pub_workers_per_conn ≤ active_publish_streams

Over-sizing workers creates contention without benefit.

Creating a Subscription

Subscribe to a stream to receive events.

Request:

{
  "type": "subscribe",
  "tenant_id": "string",
  "namespace": "string",
  "stream": "string",
  "subscription_id": 7,
  "start": "latest" | "earliest" | { "offset": 1200 },
  "shard": 0
}

start defaults to latest. earliest means the oldest record retained, and {"offset": n} resumes at offset n on a durable stream. shard defaults to 0, and a subscription reads one shard. An offset the broker cannot serve is answered with subscribe_cursor_error (too_old or in_future) instead of a silent restart at the tail.

Response:

{ "type": "subscribed", "subscription_id": 7, "start_offset": 1200, "live_offset": 1350 }

start_offset is where the subscription starts and live_offset is the stream’s tail when the subscriber was registered. Both leave out generation-start records the log ends with, since those are never delivered, and for latest they are equal. Both are sent only for a subscribe with a start, on a durable stream, to a client that negotiated FLAG_EVENT_BATCH_OFFSETS.

Broker behavior:

Broker validates tenant/namespace/stream
Broker sends subscribed on control stream
Broker opens new unidirectional stream for events
Broker sends event_stream_hello as first frame on event stream
Broker begins streaming events

Example usage:

let mut subscription = client.subscribe("acme", "prod", "events").await?;

// Receive events
while let Some(event) = subscription.next_event().await? {
    println!("Received: {:?}", event.payload);
}

Receiving Events

Events arrive on a dedicated unidirectional stream per subscription.

Event frame:

{
  "type": "event",
  "tenant_id": "acme",
  "namespace": "prod",
  "stream": "events",
  "payload": "base64-encoded-bytes",
  "offset": 1200
}

offset is present on durable streams only.

Event batch frame:

{
  "type": "event_batch",
  "tenant_id": "acme",
  "namespace": "prod",
  "stream": "events",
  "payloads": ["base64-1", "base64-2", "base64-n"]
}

In practice events arrive in binary batches. With FLAG_EVENT_BATCH_OFFSETS negotiated, each batch starts with a base_offset, and event i sits at base_offset + i. A gap between batches means the subscriber’s queue dropped events. With FLAG_EVENT_BATCH_SKIPPED as well, a batch also carries skipped_before, the count of offsets before it that hold no event, so a gap can be told apart from a record that was never an event. See Event batch offsets.

Who published an event

A subscriber that offers FLAG_EVENT_BATCH_PUBLISHER (0x2000) is told the principal that published each batch: the sub of the token the broker accepted the write from. A consumer that offers FEATURE_GROUP_PUBLISHER gets the same value as publisher on each group record. Nobody else’s frames change.

It proves that the broker accepted the write from a connection authenticated as that principal, with permission to publish to the stream. It does not say who wrote the payload’s contents. An in-memory stream reports the publishing connection’s principal. A durable stream reports what it stored with the record, so live delivery, replay from any offset, a promoted replica and a consumer group agree, and it stores one only once enabled: by finalizing the publisher_principal fleet feature in a cluster, or with FELIX_RECORD_PUBLISHERS=true on a single broker. See Event batch publisher.

When an event was written

A subscriber that offers FLAG_EVENT_BATCH_TIMESTAMPS (0x4000) gets each event’s append time, in microseconds since the Unix epoch, on every batch from a durable stream. A consumer that offers FEATURE_RECORD_TIMESTAMPS gets it as timestamp_micros on each group record. Live delivery and replay report the same time, because the broker stamps it once, when it appends the publish batch. It is the leading broker’s clock, not the publisher’s. In-memory streams store no time and send none. See Event batch timestamps.

Starting from a time

To replay from a point in time, ask for the offset first and subscribe at it:

{ "type": "offset_for_time", "tenant_id": "acme", "namespace": "prod",
  "stream": "events", "shard": 0, "at_micros": 1767225600000000, "request_id": 9 }

{ "type": "offset_value", "offset": 48210, "request_id": 9 }

The answer is the first offset on that shard appended at or after at_micros. offset is absent when nothing that recent is committed yet, so subscribe at latest; a time older than the shard’s oldest record answers with that record. It needs stream.subscribe, only the shard’s leader answers, and the broker must advertise FEATURE_RECORD_TIMESTAMPS. The lookup is a binary search over append times, so if the leader’s clock stepped back the answer is close to the right record rather than exactly it. See OffsetForTime.

Reading a range without subscribing

A subscription has no end: replay runs on into live delivery. To read a slice of a durable stream and stop, send stream_read:

{ "type": "stream_read", "tenant_id": "acme", "namespace": "prod",
  "stream": "events", "shard": 0, "from": 48210, "end": 48300,
  "max_records": 50, "request_id": 10 }

{ "type": "stream_records",
  "records": [{ "offset": 48210, "payload": "eyJ...", "timestamp_micros": 1767225600000000 }],
  "next_offset": 48260, "request_id": 10 }

Read the next page from next_offset. The range is done once it reaches end. Leave end out to read up to what is committed. The broker caps a page at FELIX_DURABLE_MAX_RECORDS_PER_READ records and 4 MiB of payload, and returns only committed records without waiting for more, so a page can come back short or empty. A from below what retention kept, or past the tail, gets a subscribe_cursor_error. No subscriber is registered. It needs stream.subscribe, only the shard’s leader answers, and the broker must advertise FEATURE_STREAM_READ. See StreamRead.

Event stream lifecycle:

sequenceDiagram
    participant C as Client
    participant B as Broker
    
    C->>B: subscribe
    B-->>C: subscribed
    Note over B: Open event stream
    B->>C: event_stream_hello
    loop Event delivery
        B->>C: event or event_batch
    end
    Note over C: Client closes connection
    Note over B: Broker closes event stream

Subscription Configuration

Client-side configuration:

let quinn = quinn::ClientConfig::with_platform_verifier();
let config = ClientConfig {
    event_conn_pool: 8,              // Connection pool size
    event_router_max_pending: 1024,  // Max pending events in client router
    ..ClientConfig::optimized_defaults(quinn)
};

Broker-side configuration:

subscriber_queue_capacity: 512       # Per-subscriber broker-core buffer
subscriber_writer_lanes: 4           # Outbound writer lanes
subscriber_lane_queue_depth: 64      # Per-lane queue depth
max_subscriber_writer_lanes: 8       # Safety clamp
subscriber_lane_shard: auto          # auto|subscriber_id_hash|connection_id_hash|round_robin_pin
event_batch_max_events: 64           # Max events per batch
event_batch_max_bytes: 65536         # Max batch size (64 KB)
event_batch_max_delay_us: 250        # Max batching delay under load (250 µs)

Batching trade-offs:

Parameter	Effect on Latency	Effect on Throughput
Increase `max_events`	Higher	Higher
Increase `max_delay_us`	Higher	Higher
Decrease `max_events`	Lower	Lower
Decrease `max_delay_us`	Lower	Lower

Multiple Subscriptions

Clients can maintain multiple concurrent subscriptions:

// Subscribe to multiple streams
let mut sub1 = client.subscribe("acme", "prod", "orders").await?;
let mut sub2 = client.subscribe("acme", "prod", "inventory").await?;
let mut sub3 = client.subscribe("acme", "staging", "logs").await?;

// Process events from all subscriptions concurrently
tokio::select! {
    Ok(Some(event)) = sub1.next_event() => handle_order(event),
    Ok(Some(event)) = sub2.next_event() => handle_inventory(event),
    Ok(Some(event)) = sub3.next_event() => handle_log(event),
}

Each subscription gets:

Independent event stream
Independent buffer
Independent flow control

Subscription Isolation

Slow subscribers don’t affect fast subscribers:

// Fast subscriber
let mut fast_sub = client.subscribe("acme", "prod", "stream").await?;
tokio::spawn(async move {
    while let Ok(Some(event)) = fast_sub.next_event().await {
        process_quickly(event).await;  // ~1ms processing
    }
});

// Slow subscriber
let mut slow_sub = client.subscribe("acme", "prod", "stream").await?;
tokio::spawn(async move {
    while let Ok(Some(event)) = slow_sub.next_event().await {
        process_slowly(event).await;  // ~100ms processing
    }
});

// Fast subscriber continues at full rate even if slow subscriber falls behind

Cache Operations

Cache Put

Store a key-value pair with optional TTL.

Request:

{
  "type": "cache_put",
  "tenant_id": "acme",
  "namespace": "prod",
  "cache": "sessions",
  "request_id": 1,
  "key": "string",
  "value": "base64-encoded-bytes",
  "ttl_ms": number | null
}

Parameters:

Parameter	Type	Required	Description
`tenant_id`, `namespace`, `cache`	string	Yes	Which cache
`request_id`	u64	Yes	Client-provided correlation ID
`key`	string	Yes	Cache key
`value`	base64	Yes	Value to store (base64-encoded)
`ttl_ms`	number	No	Time-to-live in milliseconds (null = no expiration)

Response:

{ "type": "cache_ok", "request_id": 1 }

Example usage:

// Store session with 1-hour TTL
use bytes::Bytes;
client
    .cache_put(
        "acme",
        "prod",
        "sessions",
        session_id,
        Bytes::from(session_data),
        Some(3600_000),
    )
    .await?;

// Store config without expiration
client
    .cache_put(
        "acme",
        "prod",
        "config",
        "app-settings",
        Bytes::from(config_data),
        None,
    )
    .await?;

Measured latency and throughput are on the Benchmarks page.

Cache Get

Retrieve a value from the cache.

Request:

{
  "type": "cache_get",
  "request_id": "unique-id",
  "key": "string"
}

Response:

{
  "type": "cache_value",
  "request_id": "unique-id",
  "key": "string",
  "value": "base64-encoded-bytes" | null
}

Value is null when:

Key doesn’t exist
Key has expired (TTL elapsed)
Key was evicted under memory pressure

Example usage:

match client.cache_get("acme", "prod", "sessions", session_id).await? {
    Some(session_data) => validate_session(session_data)?,
    None => anyhow::bail!("session expired or unknown"),
}

On a replicated Quorum cache, the broker confirms it still leads the shard before answering a get. FELIX_QUORUM_READS picks how: majority (the default) runs a round its replicas answer, and lease trusts the lease instead. Before the fleet has finalized lease_free_reads, every read uses the lease.

Cache Delete

Remove a key, and find out whether it was there.

Request:

{
  "type": "cache_delete",
  "request_id": "unique-id",
  "tenant_id": "acme",
  "namespace": "prod",
  "cache": "sessions",
  "key": "string"
}

Response: a cache_value carrying the value that was removed, or a null value if the key was not there. A caller can tell a delete that did something from one that did not.

Sent only to a broker that advertised FEATURE_CACHE_DELETE. A delete is an append like a put: it writes a tombstone record carrying the key, and compaction reclaims it along with the superseded values later.

match client.cache_delete("acme", "prod", "sessions", session_id).await? {
    Some(removed) => audit_log(session_id, removed),
    None => { /* already gone, or never there */ }
}

Cache Conditional Writes

{
  "type": "cache_put_if",
  "request_id": 7,
  "tenant_id": "acme",
  "namespace": "prod",
  "cache": "leases",
  "key": "endpoint-7",
  "value": "base64-encoded-bytes",
  "ttl_ms": 30000,
  "condition": "absent"
}

condition is "absent" (no live entry) or {"version": 41} (the live entry is at that version). cache_delete_if names the key and a version instead of a value and condition. Both are answered with:

{ "type": "cache_condition_result", "applied": false, "version": 41, "request_id": 7 }

Applied, version is the version the put wrote. Refused, it is the key’s current version, left out when the key has none. The check and the write are atomic on the key’s owner. A cache_value answering a get carries the value’s version too, for a client that offered the bit. Sent only to a broker that advertised FEATURE_CACHE_CONDITIONAL.

use felix_client::CacheCondition;

let taken = client
    .cache_put_if("acme", "prod", "leases", "endpoint-7", me, Some(30_000), CacheCondition::Absent)
    .await?;

Cache Request Pipelining

Cache streams support pipelining multiple requests:

// Three requests in flight at once on the client's cache streams
let (val1, val2, val3) = tokio::join!(
    client.cache_get("acme", "prod", "config", "key1"),
    client.cache_get("acme", "prod", "config", "key2"),
    client.cache_get("acme", "prod", "config", "key3"),
);

Pipelining hides the round trip, so concurrent requests finish sooner than the same requests sent one after another.

Request_id requirement:

Each request must have a unique request_id within a stream. The broker may respond out of order; clients use request_id to correlate responses.

Cache Stream Pooling

For high-concurrency cache workloads, use stream pooling:

# Client config
cache_conn_pool: 8                # Number of connections
cache_streams_per_conn: 4         # Streams per connection
# Total concurrent cache operations: 8 × 4 = 32

More streams allow more requests in flight. See Benchmarks for measured numbers.

Cache Configuration

Broker-side cache tuning. Cache requests arrive on the same listener as publishes, so the receive windows are the publish ones:

pub_conn_recv_window: 16777216       # 16 MiB per connection
pub_stream_recv_window: 16777216     # 16 MiB per stream
cache_send_window: 268435456         # 256 MiB send window

Consumer Group Operations

The third way to read a stream. Where subscribe pushes every record to every subscriber, a consumer group hands each record to one consumer and takes it back if nobody says it was handled. See Queues for the semantics.

Every request below goes only to the broker that leads the shard, and only to one that advertised FEATURE_CONSUMER_GROUP. A poll is refused rather than forwarded: relaying would put the claim and the acknowledgement on different brokers, and a queue’s whole promise is that one consumer holds a record at a time.

They travel on the control stream.

Group Poll

Claim records to work on.

Request:

{
  "type": "group_poll",
  "tenant_id": "acme", "namespace": "prod", "stream": "jobs",
  "shard": 0, "group": "fulfilment",
  "max_records": 32, "wait_ms": 5000,
  "request_id": 1
}

Response:

{
  "type": "group_records",
  "records": [{ "offset": 41, "payload": "base64-encoded-bytes", "attempts": 1 }],
  "request_id": 1
}

wait_ms is how long the broker may hold the request open waiting for work, so an idle consumer costs one open request rather than a round trip per attempt. The broker caps it at FELIX_GROUP_MAX_WAIT_MS. An empty records after the wait means nothing was available, and is not an error. That includes a group at its in-flight cap (FELIX_GROUP_MAX_IN_FLIGHT): it gets nothing more until some of what it holds is acknowledged, handed back or lapses.

attempts counts deliveries including this one, so 1 is a first attempt and anything higher is a redelivery. Absent means the broker did not report it, which is not the same as a first attempt.

Group Ack / Nack

{ "type": "group_ack",  "tenant_id": "acme", "namespace": "prod", "stream": "jobs",
  "shard": 0, "group": "fulfilment", "offset": 41, "request_id": 2 }
{ "type": "group_nack", "...": "same shape" }

An ack finishes a record. A nack hands it back for immediate redelivery, rather than waiting out the visibility timeout.

A record that is neither is redelivered once FELIX_GROUP_VISIBILITY_TIMEOUT_MS (30s) lapses. The group’s cursor advances only over a contiguous run of acks: acknowledging offset 42 while 41 is still in flight leaves the cursor at 41, which is what makes it safe to restart from.

What a group has handed out is kept in memory. After a move or a failover, or once an idle group’s state is dropped, the new state still takes an ack or nack for a claim made before it: anything below the log tail it first saw. An offset written after that and not yet handed out is answered stale_claim (retryable): nothing was applied and the record comes round.

Claim Control

{ "type": "group_poll", "...": "as above", "visibility_ms": 300000 }
{ "type": "group_extend", "...": "scope", "offset": 41, "attempts": 1,
  "extend_ms": 60000, "request_id": 9 }
{ "type": "group_extended", "visible_ms": 60000, "request_id": 9 }
{ "type": "group_nack", "...": "scope", "offset": 41, "delay_ms": 30000, "attempts": 1,
  "request_id": 10 }
{ "type": "group_dead_letter", "...": "scope", "offset": 41, "request_id": 11 }

Sent only to a broker that advertised FEATURE_GROUP_CLAIM_CONTROL. That includes the two new fields: an older broker ignores visibility_ms and delay_ms and would claim for its own timeout or nack at once.

visibility_ms sets how long one poll’s claims stand. group_extend keeps a claim standing for extend_ms from now and answers with what it granted. It is refused with stale_claim once the claim named by offset and attempts no longer stands, because it lapsed or the record went out again. delay_ms on a nack makes the record owed only after the delay, holding a place under the in-flight cap until then. attempts on a nack names the delivery the same way and is refused the same way, so a late nack cannot take the record from whoever holds it now; left out, the nack hands back whatever claim stands. The clients name the delivery on every nack, delayed or not. group_dead_letter lists the record as a dead letter and finishes it, answered with cache_ok; a record already finished is invalid_request. The broker caps every duration at FELIX_GROUP_MAX_VISIBILITY_MS. All of these need group.consume. Extensions and delays are the leader’s memory, so a failover redelivers sooner.

Dead Letters

Past FELIX_GROUP_MAX_ATTEMPTS (5) a record is dead-lettered, so one poison record cannot stall the queue behind it.

{ "type": "group_dead_letters", "...": "scope", "request_id": 3 }
{ "type": "group_dead_letter_list", "offsets": [37], "request_id": 3 }
{ "type": "group_discard", "...": "scope", "offset": 37, "request_id": 4 }
{ "type": "group_redrive", "...": "scope", "offset": 37, "request_id": 5 }

Sent only to a broker that advertised FEATURE_GROUP_DEAD_LETTERS, a separate bit from FEATURE_CONSUMER_GROUP.

A dead letter is a pointer to the record, not a copy of it. The record is still in the stream’s log at that offset, readable by an ordinary replay. group_discard drops it from the list; group_redrive puts it back in play.

Group Lifecycle

{ "type": "group_seek", "...": "scope", "start": "latest", "if_new": true, "request_id": 6 }
{ "type": "group_position", "offset": 118, "moved": true, "request_id": 6 }
{ "type": "group_describe", "...": "scope", "request_id": 7 }
{ "type": "group_info", "committed": 118, "tail": 130, "in_flight": 4, "owed": 1,
  "dead_letters": 0, "request_id": 7 }
{ "type": "group_delete", "...": "scope", "request_id": 8 }
{ "type": "group_deleted", "existed": true, "request_id": 8 }

Sent only to a broker that advertised FEATURE_GROUP_ADMIN.

group_seek moves a group’s cursor on one shard to earliest (the oldest record held), latest (the committed tail) or {"offset": n} between the two; anything outside is invalid_request. With if_new it moves only a group that does not exist yet, which is how a group is created somewhere other than offset 0. A seek voids the claims standing when it lands: an ack for a record the group now owes is answered stale_claim. Dead letters are kept.

group_info leaves out committed for a group with no cursor. in_flight and owed are the leader’s memory and start again from zero when the shard changes leader. group_delete removes the cursor and dead letters; the next poll starts the group afresh.

Seek and delete need group.manage, describe needs group.consume.

Example:

let records = client
    .group_poll_wait("acme", "prod", "jobs", 0, "fulfilment", 32, Duration::from_secs(5))
    .await?;

for record in records {
    match handle(&record.payload) {
        Ok(()) => client.group_ack("acme", "prod", "jobs", 0, "fulfilment", record.offset).await?,
        Err(_) => client.group_nack("acme", "prod", "jobs", 0, "fulfilment", &record).await?,
    }
}

Requirements

Consumer groups need durable storage. A broker started without FELIX_DURABLE_STORAGE_DIR serves no groups and does not advertise the feature. A group that forgot its position on restart would redeliver everything it had already finished, which is worse than not offering queues at all.

Cluster Operations

A broker in a cluster supports the four messages below that a standalone one does not, each gated by a feature bit it advertises during the handshake. A client must not send one to a broker that did not advertise it. Unless both sides negotiated FEATURE_UNSUPPORTED, an unrecognised message type closes the stream, so probing costs the connection.

Topology

{ "type": "topology" }
{ "type": "topology_view", "brokers": [{ "node_id": "broker-1", "addr": "..." }] }

Which brokers a client may connect to. Gated by FEATURE_TOPOLOGY. An empty list is not an error. It means the cluster has told this broker of no client-reachable address, which is the normal answer on a single node.

Redirects

{ "type": "not_leader", "node_id": "broker-2", "addr": "...", "generation": 7 }

A subscribe sent to a broker that does not own the shard is answered with this, naming the one that does. Gated by FEATURE_REDIRECT, and sent only to a client that offered the bit. Everyone else gets an ordinary error, because a client that cannot decode not_leader must not be sent one.

The client should reconnect to the named broker. A publish to the wrong broker is forwarded instead and needs nothing from the client.

Stream Shards

{ "type": "stream_shards", "tenant_id": "acme", "namespace": "prod",
  "stream": "orders", "request_id": 1 }
{ "type": "stream_shards_view", "shards": 4, "request_id": 1,
  "routing": "jump_hash" }

How many shards a stream was placed with, and how it maps routing keys onto them. Gated by FEATURE_STREAM_SHARDS. routing is sent only for a stream created with jump-hash routing; absent means modulo. A client that computes a key’s shard itself must use the stream’s own mapping.

A subscription reads one shard, so a client consuming a whole stream needs this to know how many to open; nothing else on the wire says. 0 means the broker knows nothing of that stream. It does not mean one shard: a client that rounded it up would read shard 0 and call it the stream.

Cache Shards

{ "type": "cache_shards", "tenant_id": "acme", "namespace": "prod",
  "cache": "sessions", "request_id": 2 }
{ "type": "cache_shards_view", "shards": 4, "request_id": 2 }

How many shards a cache was placed with. Gated by FEATURE_CACHE_SHARDS.

A prefix watch reads one shard, so watching a prefix across a whole cache means one cache_watch per shard. 0 means the broker knows nothing of that cache.

Shard Owners

{ "type": "shard_owners", "tenant_id": "acme", "namespace": "prod",
  "name": "orders", "kind": "stream", "request_id": 3 }
{ "type": "shard_owners_view", "request_id": 3, "owners": [
  { "shard": 0, "node_id": "broker-2", "addr": "10.0.0.6:5000", "generation": 4 },
  { "shard": 1, "generation": 0, "unavailable": "not_assigned" } ] }

Which broker owns each shard of a stream or cache (kind is stream or cache). Gated by FEATURE_SHARD_OWNERS. A shard nobody can serve right now has no node_id and says why in unavailable. A broker not in a cluster answers one shard with no node_id. Client::shard_owners asks it.

ClusterClient::subscribe_sharded does all of this for you: it asks, opens one subscription per shard, and follows each shard’s own redirect.

Other Messages

The control stream carries more than this page walks through. Each is specified in docs/protocol.md:

Message	What it does
`auth` / `auth_ok`	The first round trip on every control stream. The client offers `client_flags` and features, and the broker answers with `server_flags`, `server_features` and, when granted, a `publish_window`.
`subscribe_cursor_error`	A subscribe asked for an offset the broker cannot serve: `too_old` (retention removed it) or `in_future`.
`shard_moved`	The last frame on a subscription’s or cache watch’s event stream when its shard moved, with where to resume. Sent only with `FEATURE_SHARD_MOVED`.
`stream_deleted`	The last frame on a subscription’s event stream when its stream was deleted. Sent only with `FEATURE_STREAM_DELETED`.
`subscription_lagged`	The last frame on a durable-stream subscription whose queue on the broker dropped records, with the offset to resume from. Sent only with `FEATURE_SUBSCRIPTION_LAGGED`.
`cache_watch`	Watch a cache key or prefix: current values, then every change.
`counter_add` / `counter_get`	Add to and read a cache’s counters.
`publish_width`	The width the client routed the keyed batches that follow to one stream by. The broker refuses a keyed batch routed by a width the stream does not have with `stale_width`. Sent only with `FEATURE_PUBLISH_WIDTH`.
`producer_init` / `publish_idempotent`	Take a producer id and publish numbered batches that land once when re-sent.
`unsupported`	The answer to a request type the broker does not know, when both sides negotiated `FEATURE_UNSUPPORTED`.

HTTP Endpoints

Besides QUIC, each broker serves plain HTTP on its metrics listener (FELIX_BROKER_METRICS_BIND, 0.0.0.0:8080 by default): /metrics, /live, /ready, /replication/halted (see Observability) and the one below. The listener has no authentication, so everything on it is read-only.

Backup Offsets

curl -s http://broker-1:8080/backup/offsets | jq

{
  "node_id": "broker-1",
  "shards": [
    { "tenant_id": "t1", "namespace": "ns", "name": "orders", "shard": 0,
      "kind": "stream", "generation": 7,
      "logs": { "records": 918233, "group_cursors": 12, "group_dead_letters": 0 } },
    { "tenant_id": "t1", "namespace": "ns", "name": "prices", "shard": 0,
      "kind": "cache", "generation": 3,
      "logs": { "records": 44120, "counters": 310 } }
  ],
  "skipped": [
    { "tenant_id": "t1", "namespace": "ns", "name": "orders", "shard": 3,
      "kind": "stream", "reason": "settling" }
  ]
}

Every shard this broker leads, with the committed offset of each of its logs: one past the last record a reader may see. For a Quorum shard that is the quorum mark, never past what the leader itself has made durable; for anything else, what the leader has acknowledged. A log the broker does not keep for that shard is left out. The group and dead-letter offsets are read before the records’, so neither names a record past the records offset. generation is the leadership the offsets belong to.

A shard this broker leads but has no committed answer for is under skipped, with a reason: settling (just taken, no quorum mark yet), refused (it no longer serves the shard, or its lease lapsed on a shard whose readers still need it; with lease_free_reads finalized a replicated Quorum shard does not), moved (its leadership changed while it was read), error (with a detail), or not_durable (an in-memory stream, with nothing on disk to back up). All but the last are worth asking again shortly. A broker with no cluster leads nothing and answers with empty lists.

felix-controlplane admin backup-point is what reads this; see Backup and restore.

Error Handling

Error Response Format

{
  "type": "error",
  "message": "Descriptive error message",
  "code": "shard_unavailable",
  "retry": "retry",
  "detail": { "reason": "not_ready" }
}

code, retry and detail are sent only to a client that offered FEATURE_ERROR_CODES in auth; other clients get type and message alone. A publish is not_found only when its stream does not exist; one refused because its shard cannot be served right now is shard_unavailable, with the reason in detail, and is safe to retry. publish_error carries the same fields next to its request_id. The retry class says whether the request may have been applied (outcome_unknown) or certainly was not (retry, retry_after, redirect), and an unknown code must be tolerated. The full table is in docs/protocol.md under “Error codes”.

Common Errors

Unknown stream:

{
  "type": "publish_error",
  "request_id": 1,
  "message": "stream not found: tenant=acme namespace=prod stream=events",
  "code": "not_found",
  "retry": "retry_after"
}

The stream does not exist on this broker. Brokers learn streams from the control plane, so a stream created a moment ago can answer this until the broker’s next sync. Retry after a short wait, and check the stream exists if it persists.

Publish queue full:

{
  "type": "error",
  "message": "publish queue full; retry in 10 ms",
  "code": "overloaded",
  "retry": "retry_after",
  "detail": { "reason": "publish_queue_full", "retry_after_ms": 10 }
}

Resolution: The broker’s publish queue had no room for this tenant, and nothing was queued; retry after the suggested wait. Persistent refusals mean the broker is overloaded, or this tenant is sending more than its share: check felix_tenant_publish_queue_full_total by tenant, reduce the publish rate, or raise pub_queue_depth.

Authorization failure:

{
  "type": "publish_error",
  "request_id": 1,
  "message": "forbidden",
  "code": "forbidden",
  "retry": "fatal"
}

Publish, subscribe and cache operations each check a permission against the tenant-scoped token. A forwarded publish is authorized twice (at the broker the client reached and again at the shard’s owner), so routing does not launder a credential.

Connection Errors

QUIC connection errors are surfaced as connection-level failures:

Certificate validation failure: TLS handshake error
Connection timeout: No response within the QUIC idle timeout (6 s by default, FELIX_MAX_IDLE_TIMEOUT_MS)
Connection reset: Broker restart or network issue

Retry logic:

Branch on the error’s retry class. A BrokerError is recovered from the anyhow::Error with downcast_ref. Only retry, retry_after and redirect say nothing was applied. outcome_unknown may have landed, so re-sending it is safe only through an idempotent producer.

use felix_client::{AckMode, BrokerError, ClusterClient, RetryClass};
use std::time::Duration;

async fn publish_with_retry(client: &ClusterClient, data: &[u8], retries: u32) -> anyhow::Result<()> {
    for attempt in 0..retries {
        let err = match client
            .publish("acme", "prod", "events", data.to_vec(), AckMode::PerMessage)
            .await
        {
            Ok(_) => return Ok(()),
            Err(err) => err,
        };
        match err.downcast_ref::<BrokerError>().map(|e| e.retry) {
            Some(RetryClass::Retry | RetryClass::RetryAfter) => {
                tokio::time::sleep(Duration::from_millis(100 * 2u64.pow(attempt))).await;
            }
            _ => return Err(err),
        }
    }
    anyhow::bail!("gave up after {retries} attempts")
}

Performance Tuning

Publish Performance

Maximize throughput:

# Broker config
pub_workers_per_conn: 8
pub_queue_depth: 2048
pub_inflight_bytes: 268435456
event_batch_max_events: 256
event_batch_max_delay_us: 2000

// Client: scale publish throughput via pools and sharding
use felix_client::PublishSharding;

let quinn = quinn::ClientConfig::with_platform_verifier();
let config = ClientConfig {
    publish_conn_pool: 8,
    publish_streams_per_conn: 4,
    publish_sharding: PublishSharding::HashStream,
    ..ClientConfig::optimized_defaults(quinn)
};

Minimize latency:

# Broker config
pub_workers_per_conn: 2
event_batch_max_events: 8
event_batch_max_delay_us: 100

// Client: publish immediately
client
    .publish("acme", "prod", "events", data.to_vec(), AckMode::PerMessage)
    .await?;

High fanout tuning:

subscriber_queue_capacity: 4096   # Larger per-subscriber burst buffer
subscriber_writer_lanes: 4        # Start with 4, benchmark before increasing
subscriber_lane_shard: auto
fanout_batch_size: 128            # Batch fanout operations
event_batch_max_events: 128       # Larger event batches

Low latency tuning:

subscriber_queue_capacity: 64
subscriber_writer_lanes: 2
subscriber_lane_shard: auto
event_batch_max_events: 8
event_batch_max_delay_us: 100

Cache Performance

High concurrency:

cache_conn_pool: 16
cache_streams_per_conn: 8
# Total: 128 concurrent operations

Low latency:

cache_conn_pool: 4
cache_streams_per_conn: 2
cache_conn_recv_window: 134217728  # Smaller windows for lower memory

Change these knobs based on measurements. High queue depth means too few workers; contention means too many; dropped events mean buffers too small for the workload’s bursts; high memory means the opposite. Broker telemetry and client metrics (see Observability) tell you which.

Broker API Reference

Connection Model

QUIC Connection Lifecycle

Connection Pooling

Authentication (Felix Tokens)

Publish Operations

Single Message Publish

Batch Publish

Binary Batch Publish

Conditional Publish

Publish Pipeline Configuration

Subscribe Operations

Creating a Subscription

Receiving Events

Who published an event

When an event was written

Starting from a time

Reading a range without subscribing

Subscription Configuration

Multiple Subscriptions

Subscription Isolation

Cache Operations

Cache Put

Cache Get

Cache Delete

Cache Conditional Writes

Cache Request Pipelining

Cache Stream Pooling

Cache Configuration

Consumer Group Operations

Group Poll

Group Ack / Nack

Claim Control

Dead Letters

Group Lifecycle

Requirements

Cluster Operations

Topology

Redirects

Stream Shards

Cache Shards

Shard Owners

Other Messages

HTTP Endpoints

Backup Offsets

Error Handling

Error Response Format

Common Errors

Connection Errors

Performance Tuning

Publish Performance

Subscribe Performance

Cache Performance