Redis / Valkey Streams

If Redis or Valkey is already in your stack, this backend gives you stream-based messaging with FNV-1a shard routing for per-key ordering, consumer groups, hold-queue retries, and queue-depth autoscaling — without adding another broker to your infrastructure.

What you need

Redis 6.2+ or any Valkey release. shove uses ZRANGE … BYSCORE (introduced in Redis 6.2) for hold-queue polling. The version is validated at connection time: Broker::new returns an error immediately if the server is older rather than failing later with a cryptic protocol error.

For local dev:

docker run --rm -p 6379:6379 redis:7-alpine

Install

cargo add shove --features redis-streams

Connect

use shove::redis::{RedisConfig, RedisMode};
use shove::{Broker, Redis};
 
let broker = Broker::<Redis>::new(RedisConfig {
    mode: RedisMode::Standalone {
        url: "redis://127.0.0.1:6379/".into(),
    },
    group: None,
})
.await?;

RedisMode::Standalone accepts redis:// and rediss:// (TLS) URLs. For Redis Cluster, use RedisMode::Cluster { urls: vec![...] } with one or more seed node URLs.

group sets the consumer group name shared by all consumers. Defaults to "shove" if None.

RedisConfig exposes three builder methods on top of the struct fields:

use shove::redis::{RedisConfig, RedisMode};
use std::time::Duration;

let config = RedisConfig::new(RedisMode::Standalone {
    url: "redis://127.0.0.1:6379/".into(),
})
.with_group("billing")                                // overrides the default "shove" group name
.with_connection_timeout(Duration::from_secs(10))     // TCP connect deadline; default 10s
.with_response_timeout(Duration::from_secs(30));      // per-command deadline; default 30s

response_timeout must be longer than the consumer's XREADGROUP BLOCK window (the consumer derives this from prefetch_count and the polling interval) — otherwise the client cancels the long-poll mid-block and treats the resulting timeout as a connection error. The defaults are sized to leave headroom.

Declare topology

broker.topology().declare::<Orders>().await?;

declare creates the Redis Streams and Sorted Sets required by the topic topology: the main shard streams, hold-queue sorted sets, and DLQ stream when .dlq() is configured. The call is idempotent and safe to run on every startup.

Publish

let publisher = broker.publisher().await?;
publisher.publish::<Orders>(&OrderPaid { order_id: "ORD-1".into() }).await?;

publish routes to a shard stream via FNV-1a hash of the message key and appends the entry with XADD.

Consume

use shove::redis::RedisConsumerGroupConfig;
use shove::{ConsumerGroupConfig};
 
let mut group = broker.consumer_group();
group
    .register::<Orders, _>(
        ConsumerGroupConfig::new(
            RedisConsumerGroupConfig::new(1..=4)
                .with_prefetch_count(10)
                .with_concurrent_processing(true),
        ),
        || Handler,
    )
    .await?;

RedisConsumerGroupConfig::new(min..=max) mirrors the other coordinated-group backends. The registry starts min_consumers tasks; max_consumers is the ceiling the autoscaler scales up to based on stream backlog (enable it with enable_autoscaling). Each task polls its assigned shard stream with XREADGROUP BLOCK COUNT prefetch_count.

.with_prefetch_count(N) — the COUNT argument to XREADGROUP. With concurrent processing on, also the in-flight cap per consumer task. Default: 10.
.with_concurrent_processing(true) — dispatch each fetched message to its own tokio task with a per-task multiplexed connection for outcome routing. A semaphore caps in-flight handlers at prefetch_count. Default: false (handlers serialize via prefetch clamped to 1).
.with_handler_timeout(Duration) — per-message handler deadline. Timed-out messages are left in the PEL for XAUTOCLAIM to reclaim. Default 30s; the registry-level default can be set via RedisConsumerGroupRegistry::with_default_handler_timeout(Duration). The timeout also sets the crash-recovery reclaim deadline for the whole consumer group, so use the same value for every consumer of a stream and group, including other processes (see Stream maintenance).
.with_max_retries(u32) — retry budget before a message is dead-lettered. Default 10.
.with_max_pending_per_key(usize) — caps the per-key pending buffer for sequenced topics. Default 100.
.with_max_message_size(usize) — payload ceiling in bytes; oversize messages route to the DLQ without invoking the handler. Default 1 MiB.

ConsumerOptions::<Redis>::new().with_max_reconnect_attempts(N) bounds how many times a consumer redials Redis after a connection drop before surfacing a Connection error. Default: unlimited.

Sequenced delivery

Messages for the same key stay in order. The publisher hashes the sequence key to a shard stream; each shard has exactly one active consumer task, so messages for the same key are never processed concurrently. Different keys on different shards are fully independent.

use shove::consumer_group::ConsumerGroupConfig;
use shove::redis::RedisConsumerGroupConfig;
 
group
    .register_fifo::<Ledger, _>(
        ConsumerGroupConfig::new(RedisConsumerGroupConfig::default()),
        || Handler,
    )
    .await?;

register_fifo wires consumer tasks for sequenced delivery. RedisConsumerGroupConfig::default() spawns one consumer task per shard; pass RedisConsumerGroupConfig::new(min..=max) to run more replicas. with_concurrent_processing(true) is rejected with a Topology error on this path — it would break per-key ordering within a shard. See Sequenced Topics for the full ordering model.

Stream maintenance

While a consumer runs, a background sidecar keeps its stream healthy. Entries that every consumer group on the stream has acknowledged are trimmed away, so stream memory tracks outstanding work rather than lifetime throughput. Entries stuck pending past the handler timeout — for example after a consumer crash — are reclaimed and redelivered. One sidecar runs per stream and group per process, no matter how many consumers you scale to, and it applies to consumers started through the group registry and through ConsumerSupervisor alike.

Three behaviours worth knowing:

Trimming waits for the slowest reader. If several consumer groups share a stream (fan-out via distinct with_group names), entries survive until every group has acknowledged them. A group that has not consumed anything yet blocks trimming entirely.
DLQ streams are never trimmed or reclaimed. Dead letters are an audit record and stay put until you handle them.
Disabling the handler timeout also disables reclaim. ConsumerOptions::without_handler_timeout() means in-flight work is never presumed dead; trimming still runs. Configure the same timeout for every consumer of a stream and group, including consumers in other processes — a process with a shorter timeout reclaims entries held by longer-running consumers elsewhere.

Gotchas

Redis 6.2+ is required. Connecting to an older server returns an error at startup. Any Valkey release is compatible.
Standalone vs cluster. Cluster mode routes XADD and XREADGROUP commands through the cluster client. All shard keys for a given topic should hash to the same slot if you rely on Lua-based atomicity; shove does not use Lua, so cross-slot operations are safe.
Consumer group names must be unique per application. The group name (default "shove") is shared by all consumers on the client. If multiple unrelated applications share a Redis instance, set a distinct group name per application via RedisConfig { group: Some("myapp".into()), .. }.
Hold queues are Sorted Sets, not streams. Retried messages are stored in a Redis Sorted Set with a redeliver-at timestamp as the score. A background requeuer task polls the set and XADDs due entries back. If two instances run concurrently during a rolling restart, the same message may be requeued twice — this is expected at-least-once behaviour; handlers must be idempotent.

Examples

Basic — publish/consume round trip with hold queues and DLQ
Sequenced — FNV-1a shard routing for per-key ordering
Consumer Groups — coordinated min..=max group with a burst-and-drain workload
Autoscaler — consumer-lag-driven scale up/down within a min..=max range
Audited Consumer — MessageHandlerExt::audited wrapping
Stress — throughput benchmarking