Redis Streams¶

A #[subscriber("key")] handler binds to a Redis stream key. Because Redis Streams always read through a consumer group, the bare-string form needs a broker-wide default group (.default_group):

crates/ruststream-fred/examples/fred_streams.rs

use ruststream::runtime::{App, AppInfo, HandlerResult, RustStream};
use ruststream::subscriber;
use ruststream_fred::RedisBroker;
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerResult {
    println!("got order {}", order.id);
    HandlerResult::Ack
}

Wire it onto the broker; the with_broker / include part is identical to every other broker.

crates/ruststream-fred/examples/fred_streams.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379").default_group("workers"),
        |b| {
            b.include(handle);
        },
    )
}

Payload and headers travel as stream entry fields: the body under a reserved field and each header under a h: prefix, so a round-trip through XADD / XREADGROUP preserves both.

Read modes: fresh tail vs reclaim¶

The read mode is chosen by constructor, never a runtime flag, because the two return disjoint sets of messages:

RedisStream::new(key) reads fresh entries off the tail (XREADGROUP >). This is the normal worker.
RedisStream::reclaim(key, min_idle) reclaims entries another consumer fetched but never acked (XAUTOCLAIM, idle at least min_idle). This is crash recovery, run alongside a new subscriber on the same group ("two handlers per group").

min_idle has no default and must exceed the longest legitimate handler runtime: set it too low and a healthy consumer's in-flight message gets reclaimed and processed twice.

A descriptor can sit directly in the #[subscriber(...)] decorator. The fresh-tail worker:

crates/ruststream-fred/examples/fred_reclaim.rs

// The descriptor sits directly in the decorator: a fresh-tail consumer on the `workers` group.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerResult {
    println!("processing order {}", order.id);
    HandlerResult::Ack
}

The recovery handler on the same group, reclaiming entries idle for over 30 seconds:

crates/ruststream-fred/examples/fred_reclaim.rs

// A recovery handler for the same group: reclaims entries left pending for over 30s.
#[subscriber(RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers"))]
async fn recover(order: &Order) -> HandlerResult {
    println!("recovering order {}", order.id);
    HandlerResult::Ack
}

Repositioning a group¶

A stream keeps its entries until it is trimmed, so a group can be moved back over history or forward past a region. StreamStart only chooses where a group starts when it is first created; moving a group that already exists is the Seekable capability, which the streams transport implements (the list transport is destructive and Pub/Sub keeps no history, so neither does).

A seek is group-wide. Redis keeps one cursor per consumer group, so moving it repositions every consumer of that group, not just the subscription that asked - unlike a partitioned log, where a seek is scoped to one consumer. The type names carry that scope: RedisGroupPosition and RedisGroupSeeker.

Three positions, named by constructor:

Constructor	Where the group resumes
`RedisGroupPosition::beginning()`	the oldest entry the stream still retains
`RedisGroupPosition::end()`	the tail: only entries added afterwards
`RedisGroupPosition::after(id)`	the entry following `id` (the cursor is exclusive, like `XGROUP SETID`)

A start_at(..) clause seeks the subscription before its first delivery, on every startup:

crates/ruststream-fred/examples/fred_seek.rs

// The audit trail replays from the oldest retained entry on every start: the clause takes a
// position constructor, and the subscription is sought there before its first delivery. Because
// the cursor belongs to the group, this rewinds the `auditors` group as a whole.
#[subscriber(
    RedisStream::new("audit").group("auditors"),
    start_at(RedisGroupPosition::beginning())
)]
async fn replay(order: &Order) -> HandlerResult {
    println!("audit: replayed order {}", order.id);
    HandlerResult::Ack
}

A Seek parameter injects the subscription's own seeker, so a handler can move the group while the service runs:

crates/ruststream-fred/examples/fred_seek.rs

// The worker owns its group's cursor: on the producer's poison marker it skips the group forward
// to the tail instead of grinding through the bad region. Every consumer of `workers` resumes
// there, which is the point - the region is bad for all of them.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order, Seek(seeker): Seek<RedisGroupSeeker>) -> HandlerResult {
    if order.id == 0 {
        if seeker.seek(RedisGroupPosition::end()).await.is_err() {
            return HandlerResult::retry();
        }
        println!("orders: skipped the poison region");
        return HandlerResult::Ack;
    }
    println!("orders: processed {}", order.id);
    HandlerResult::Ack
}

A delivery also reports its own position (Positioned::position), and seeking to it delivers that message again followed by the entries after it - the id is decremented automatically, since the cursor is exclusive.

What a seek does not touch:

the pending entries list. Entries already delivered and not acknowledged stay pending whichever way the cursor moved, and remain reachable through the reclaim path.
scheduled delayed retries. Copies already sitting in a ZSET delay queue are keyed by their due time, so they are appended to the stream when they fall due regardless of where the group reads.
delivery counts. A replayed entry is delivered again, so its native delivery count grows; a reclaim subscription with max_deliveries therefore counts replays towards the poison cap, while the framework retry-count header only moves on an actual nack.

The cursor changes as soon as the seek returns, but a subscription parked in a blocking XREADGROUP observes it on its next read - within one block interval. Entries selected under the old cursor are discarded rather than delivered.

Acknowledgement¶

Settlement follows the republish-retry model:

ack -> XACK (remove from the pending list).
nack(requeue = true) -> re-append a copy to the same stream, then XACK the original. The copy is reprocessed by the normal new consumer. This is at-least-once: a crash between the two leaves a duplicate.
nack(requeue = false) -> XACK to drop.

Delayed retry¶

A handler can ask for a delayed redelivery (HandlerResult::retry_after(delay)), for example to back off a transient failure. Redis Streams have no native per-message delay, so by default the runtime falls back to an in-process timer that re-publishes the message after the delay - at-most-once over that window, since a crash before the timer fires loses the deferred copy.

For a crash-safe alternative, opt a subscription into a durable ZSET delay queue. It is off by default and you name the ZSET key explicitly (the key has no sane default):

crates/ruststream-fred/examples/fred_delayed_retry.rs

// On a transient failure the handler asks for a delayed retry. The delay queue is the named ZSET,
// so the redelivery is durable: it survives a crash between the failure and the retry firing.
#[subscriber(
    RedisStream::new("orders")
        .group("workers")
        .delayed_retry(DelayedRetry::DurableZset { key: "orders.delayed".to_owned(), ttl: None })
)]
async fn handle_order(order: &Order) -> HandlerResult {
    if order.id == 0 {
        // Park the message in the ZSET for 30s instead of blocking the worker or busy-requeuing.
        return HandlerResult::retry_after(Duration::from_secs(30));
    }
    println!("processed order {}", order.id);
    HandlerResult::Ack
}

A delayed delivery is ZADDed to the named ZSET with a fire_at score, then the original is XACKed; a sweeper folded into the subscription's read loop moves due entries back onto the stream with XADD, so the retry survives a restart. The sweeper's granularity is the read block interval, and the retry-count header is incremented on each pass. An optional TTL on the ZSET key cleans up an abandoned queue, but it must exceed the longest scheduled delay or pending entries are dropped before they fire. Scores are wall-clock epoch milliseconds, so keep clocks synced (NTP).

Capabilities¶

Which of the framework's optional capability traits this broker implements natively. Streams implements the most of the three transports; the notes name where Lists and Pub/Sub differ.

Capability	Native	Notes
`Subscribe`	yes	Subscribes by stream key through a consumer group (the bare-string form needs `default_group`). Lists and Pub/Sub subscribe through their own descriptors.
`BatchSubscriber`	yes (Streams)	One batch per non-empty `XREADGROUP` / `XAUTOCLAIM` read, up to `RedisStream::count` entries, never empty. The List and Pub/Sub subscribers deliver one message at a time.
`TransactionalPublisher`	yes (Streams, standalone and sentinel)	The stream publisher buffers on the handle and commits it as one `MULTI` / `EXEC`. A cluster publisher rejects it, because a `MULTI` block cannot span hash slots. The List and Pub/Sub publishers have no transaction. See Transactions.
`OwnedTransactions`	yes (Streams, standalone and sentinel)	`publisher.transaction()` returns a buffer-owning value, so any number can be open on one handle; cluster is rejected for the same reason.
`RequestReply`	no	Redis has no request-reply primitive: nothing on the wire carries a reply address or correlates a reply with its request.
`Partitioned`	yes	All three transports read the key from the `redis-partition-key` header for the runtime's `workers(n, by_key)` lanes. The sender sets it.
`Seekable` + `Positioned`	yes (Streams)	The group cursor moves with `XGROUP SETID`, and a delivery reports the position that redelivers it. See Repositioning a group. A list is destructive and Pub/Sub keeps no history, so neither implements it.
`DescribeServer`	yes	Reports the configured address (the first seed on cluster and sentinel).