Consumer Groups & Autoscaling

When your topic fills up faster than one consumer can drain it, you add consumers. When traffic drops off at night and you don't need ten idle workers, the autoscaler removes them. broker.consumer_group() is shove's answer to both: a coordinated, min/max-bounded group of consumers that the autoscaler can grow or shrink based on queue depth. Available on RabbitMQ, NATS JetStream, Apache Kafka, Redis Streams, and InMemory.

ConsumerGroupConfig shape

Each backend has its own concrete ConsumerGroupConfig type. The generic ConsumerGroupConfig\<B\> wrapper in the top-level API holds the backend-specific inner config:

ConsumerGroupConfig::new(inner_config)

For the InMemory backend:

use shove::inmemory::InMemoryConsumerGroupConfig;
use shove::ConsumerGroupConfig;

ConsumerGroupConfig::new(
    InMemoryConsumerGroupConfig::new(1..=8)   // min=1, max=8 consumers
        .with_prefetch_count(10)
        .with_max_retries(5)
        .with_handler_timeout(Duration::from_secs(30)),
)

For the RabbitMQ backend:

use shove::rabbitmq::ConsumerGroupConfig as RabbitMqConsumerGroupConfig;

ConsumerGroupConfig::new(
    RabbitMqConsumerGroupConfig::new(1..=8)
        .with_prefetch_count(20)
        .with_max_retries(5)
        .with_handler_timeout(Duration::from_secs(30))
        .with_concurrent_processing(true),
)

Methods available on all backend configs (per-group; the registry-level default is documented below):

• .new(min..=max) — sets the consumer count range. The group starts at min consumers and the autoscaler can add up to max. Panics if min > max.
• .with_prefetch_count(N) — number of unacknowledged messages the broker delivers to each consumer. Default: 10.
• .with_max_retries(N) — retry budget per message. Default: 10.
• .with_handler_timeout(Duration) — maximum time a handler may run before the message is retried. Default: 30s.
• .with_concurrent_processing(bool) — when true, each consumer processes up to prefetch_count messages concurrently. When false, one message at a time. Default varies by backend (RabbitMQ defaults to false; InMemory defaults to sequential via prefetch clamping).

Autoscaling — the signal

The autoscaler polls queue/consumer metrics on a configurable interval and adjusts the consumer count within the configured range. The metric used as the scaling signal differs by backend:

The autoscaler uses a ThresholdStrategy by default, wrapped in Stabilized to prevent flapping:

• Scale up when messages_ready > capacity × scale_up_multiplier (default: 2.0) and the condition has been sustained for at least hysteresis_duration (default: 10s).
• Scale down when messages_ready < capacity × scale_down_multiplier (default: 0.5) and the condition has been sustained for hysteresis_duration.
• A scaling action is not taken again until cooldown_duration (default: 30s) has elapsed since the last action.

capacity is active_consumers × prefetch_count.

AutoscalerConfig

AutoscalerConfig controls the polling and decision-making behavior:

use shove::autoscaler::AutoscalerConfig;

let config = AutoscalerConfig {
    poll_interval: Duration::from_secs(5),       // how often to check queue depths
    scale_up_multiplier: 2.0,                    // scale up when ready > capacity × 2.0
    scale_down_multiplier: 0.5,                  // scale down when ready < capacity × 0.5
    hysteresis_duration: Duration::from_secs(10), // condition must hold for 10s
    cooldown_duration: Duration::from_secs(30),  // wait 30s between scaling actions
};

Reasonable defaults work for most workloads. When to tune:

• Lower poll_interval — if queue depths spike and drain quickly (bursty workloads). Lower values increase autoscaler overhead but improve responsiveness.
• Higher scale_up_multiplier — if you're scaling up too aggressively for shallow spikes. Higher values require a proportionally larger queue depth before adding consumers.
• Lower scale_down_multiplier — if you want to keep more consumers alive during lulls (avoids scale-up latency). Set it to 0.0 to never scale down automatically.
• Longer hysteresis_duration — if the autoscaler is flapping (scaling up then immediately down). Longer hysteresis requires sustained conditions before acting.
• Longer cooldown_duration — if scaling actions are happening too frequently. Prevents rapid oscillation.

Enabling autoscaling

On the coordinated backends (InMemory, Redis Streams, Apache Kafka, NATS JetStream, RabbitMQ) one builder call turns autoscaling on. enable_autoscaling(config) stashes the AutoscalerConfig on the group; run_until_timeout then owns the entire lifecycle — it starts the consumers, spawns the autoscaler against the group's own registry, stops and joins the autoscaler when the signal fires, and drains the consumers:

use shove::autoscaler::AutoscalerConfig;

let outcome = group
    .enable_autoscaling(AutoscalerConfig::default())
    .run_until_timeout(
        tokio::signal::ctrl_c().map(drop),
        Duration::from_secs(30),
    )
    .await;

std::process::exit(outcome.exit_code());

The consumer count stays within the per-group ConsumerGroupConfig min/max range; the autoscaler uses Stabilized<ThresholdStrategy> derived from the AutoscalerConfig. A panic in the autoscaler task (observed before its stop deadline) surfaces in the returned SupervisorOutcome as a panic. SQS has no consumer_group() entry point, so it wires the autoscaler manually — see the SQS autoscaler example.

Shutdown semantics

ConsumerGroup exposes run_until_timeout for clean shutdown:

let outcome = group
    .run_until_timeout(
        tokio::signal::ctrl_c().map(drop),
        Duration::from_secs(30),
    )
    .await;

std::process::exit(outcome.exit_code());

• signal_future — when this future resolves, the group stops accepting new deliveries and begins draining in-flight handlers.
• drain_timeout — how long to wait for in-flight handlers to finish before forcibly aborting them. Handlers still running at the deadline are abandoned (the tokio task is aborted, not cancelled).
• SupervisorOutcome — captures errors, panics, and whether a timeout occurred. .exit_code() maps these to 0 (clean), 1 (errors), 2 (panics), 3 (drain timeout).

Pick a drain timeout longer than your longest-expected handler. If handlers routinely take up to 10 seconds, use 30 seconds. Too short a drain timeout results in abandoned in-flight messages that the broker will redeliver — handlers must be idempotent. See Shutdown & Exit Codes for the full contract.

Per-backend config knob differences

Each backend's consumer group config has backend-specific knobs beyond the shared set:

RabbitMQ (rabbitmq::ConsumerGroupConfig) — adds .with_concurrent_processing(bool). When true, each consumer processes up to prefetch_count messages concurrently. RabbitMQ's Single Active Consumer mode is enabled automatically for sequenced topics.

NATS (nats::NatsConsumerGroupConfig) — see NatsConsumerGroupConfig for NATS-specific options. NATS pull consumers use the max_ack_pending setting derived from prefetch_count.

Kafka (kafka::KafkaConsumerGroupConfig) — adds .with_concurrent_processing(bool). Kafka's consumer-group protocol handles partition assignment across instances automatically.

Redis (redis::RedisConsumerGroupConfig) — adds .with_prefetch_count(u16), .with_concurrent_processing(bool), and .with_handler_timeout(Duration). Construct via RedisConsumerGroupConfig::new(min..=max); the registry starts min_consumers tasks and the autoscaler scales up to max_consumers based on stream backlog. With concurrent_processing(true) each consumer dispatches up to prefetch_count handlers in parallel via spawned tasks and per-task multiplexed connections; register_fifo rejects this flag to preserve per-key ordering.

InMemory (inmemory::InMemoryConsumerGroupConfig) — adds knobs for max_pending_per_key (for sequenced topics) and max_message_size. Useful for test-environment configuration.

SQS — has no broker-level coordinated-group primitive. consumer_group() is not available on Broker\<Sqs\>. Refer to the SQS overview for scaling on SQS.

For exact method signatures and backend-specific options, refer to the sidebar links to each backend's overview page.

Registry-level default handler timeout

ConsumerGroup (returned by broker.consumer_group()) exposes a registry-level default that applies to every group registered through that registry whose per-group config did not call .with_handler_timeout(...):

let mut group = broker
    .consumer_group()
    .with_default_handler_timeout(Duration::from_secs(60));

group
    .register::<OrdersTopic, _>(
        ConsumerGroupConfig::new(RabbitMqConsumerGroupConfig::new(1..=4)),
        || OrdersHandler,
    )
    .await?;

Resolution order, from highest priority to lowest:

1. Per-group explicit — the config called .with_handler_timeout(d). This always wins.
2. Registry default — broker.consumer_group().with_default_handler_timeout(d).
3. Library default — 30 seconds, the value of DEFAULT_HANDLER_TIMEOUT.

Call .with_default_handler_timeout(...) before register / register_fifo. Each backend resolves the effective timeout at registration time, so a call after the first registration has no effect on the already-registered group. broker.consumer_group() also returns a fresh registry on every call — a default set on one handle does not propagate to subsequent broker.consumer_group() results.

Both .with_handler_timeout(d) and .with_default_handler_timeout(d) panic if d is Duration::ZERO — a zero timeout would otherwise perpetuate an infinite retry loop.

Available on every backend that implements coordinated consumer groups (RabbitMQ, NATS, Kafka, Redis, InMemory). SQS uses the supervisor model and has no consumer_group() entry point; SQS users set timeouts via ConsumerOptions::<Sqs>::with_handler_timeout instead.

Full walkthrough

[!include ~/examples/inmemory/consumer_groups.rs]

The key lines are 63–70: group.register::<WorkTopic, _>(...) wraps the config in ConsumerGroupConfig::new(InMemoryConsumerGroupConfig::new(3..=6).with_prefetch_count(4)) — three to six consumers, each with a prefetch of four. The factory closure move || Worker { count: c.clone() } runs once per consumer instance, returning a fresh (but Clone-cheap) handler each time. Lines 93–99 show the canonical shutdown: a CancellationToken flipped by a background waiter task, passed into run_until_timeout, followed by std::process::exit(outcome.exit_code()).

What's next

• The Broker<B> Pattern
• Shutdown & Exit Codes
• InMemory consumer groups example
• SQS autoscaler example