Handlers & Context

A handler is your business logic. It receives a message, does work — writes a record, calls an API, updates state — and returns an Outcome that tells the consumer what to do next. The library calls handle() for each delivered message and routes the message based on the returned Outcome. Handlers are parameterized on a Topic, which prevents accidentally sharing a handler between two topics that happen to carry the same message type.

The MessageHandler trait

Every handler implements MessageHandler<T> where T: Topic:

pub trait MessageHandler<T: Topic>: Send + Sync + 'static {
    type Context: Clone + Send + Sync + 'static;

    fn handle(
        &self,
        message: T::Message,
        metadata: MessageMetadata,
        ctx: &Self::Context,
    ) -> impl Future<Output = Outcome> + Send;

    fn handle_dead(
        &self,
        message: T::Message,
        metadata: DeadMessageMetadata,
        ctx: &Self::Context,
    ) -> impl Future<Output = ()> + Send {
        // default: log a warning and ack
    }
}

The handle method is required. handle_dead has a default implementation that logs a warning at WARN level and acks the dead-lettered message. Override it if you need alerting or investigation logic.

Context — shared state without a registry

Most handlers need access to shared resources: a database connection pool, an HTTP client, a cache, a reference to application configuration. Context is how you inject those resources cleanly.

You supply the context once when registering the handler; the harness clones it into each consumer task. Because Clone is required, the idiomatic pattern is Arc<AppState> where cloning is a cheap reference-count increment rather than a deep copy.

When there is no shared state to inject, use ():

struct Handler;

impl MessageHandler<Orders> for Handler {
    type Context = ();

    async fn handle(&self, msg: Order, _: MessageMetadata, _: &()) -> Outcome {
        println!("order received: {:?}", msg.id);
        Outcome::Ack
    }
}

When your handler depends on a database pool or other shared resource, declare a Context type:

#[derive(Clone)]
struct AppState {
    db: Arc<Pool>,
}

struct Handler;

impl MessageHandler<Orders> for Handler {
    type Context = AppState;

    async fn handle(&self, msg: Order, _: MessageMetadata, ctx: &AppState) -> Outcome {
        match ctx.db.insert(&msg).await {
            Ok(_) => Outcome::Ack,
            Err(_) => Outcome::Retry,
        }
    }
}

// Registration:
let mut group = broker.consumer_group().with_context(state.clone());
group.register::<Orders, _>(cfg, || Handler).await?;

Context should be cheap to clone. Arc<AppState> is the canonical pattern; the clone is a refcount bump and does not copy any data.

Handler timeouts

Every handler call runs under a wall-clock deadline. If it does not return within the timeout, the future is dropped, Outcome::Retry is recorded by default, and metrics::record_failed(..., FailReason::Timeout) is emitted. The default is 30 seconds.

The outcome a timeout resolves to is configurable — see Choosing what a timeout means below.

Three layers control the timeout, in priority order — the first one set wins:

1. Per-group override — <Backend>ConsumerGroupConfig::with_handler_timeout(Duration). Lives on the group config; rebuilt per register call.
2. Registry default — <Backend>ConsumerGroupRegistry::with_default_handler_timeout(Duration). Applies to every group that did not set its own. Use this when most groups in a service share a sensible deadline and a few outliers override it.
3. Library default — DEFAULT_HANDLER_TIMEOUT = 30s.

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

group
    .register::<Orders, _>(
        ConsumerGroupConfig::new(
            KafkaConsumerGroupConfig::new(1..=4)
                .with_handler_timeout(Duration::from_secs(60)),  // overrides the registry default for this topic
        ),
        || OrderHandler,
    )
    .await?;

Per-supervisor consumers (ConsumerSupervisor::register) use ConsumerOptions::<Backend>::new().with_handler_timeout(Duration) instead — the same three-layer resolution applies, with the supervisor's registry default sitting between per-consumer overrides and the library default.

The timeout enforces deadlines on the handler future itself. It does not cancel work that has escaped into a spawned task or a long-running blocking call; if you need to bound those, you must wire the cancellation yourself. The default is generous because the library cannot tell whether your handler is doing IO or compute — set a shorter one when you know the work is bounded.

Choosing what a timeout means

A timeout is the library's verdict on a slow consumer, not evidence that the message itself is bad — but from the outside the two are indistinguishable, so the choice belongs to you. with_handler_timeout_outcome sets it, on ConsumerOptions or on a <Backend>ConsumerGroupConfig:

group
    .register::<Orders, _>(
        ConsumerGroupConfig::new(
            KafkaConsumerGroupConfig::new(1..=4)
                .with_handler_timeout(Duration::from_secs(60))
                .with_handler_timeout_outcome(Outcome::Defer),
        ),
        || OrderHandler,
    )
    .await?;

Leaving it unset preserves the existing behaviour on every backend.

Pick Defer when a timeout means "too slow right now", not "poisoned". That is the case a service hits when its handler stalls on a downstream dependency: under Retry the stall burns the whole retry budget and the message is dead-lettered — or discarded, if the topology declares no DLQ — even though nothing was ever wrong with it. Defer is the outcome that survives a slow dependency without losing valid data.

That discard is no longer silent: it increments shove_messages_discarded_total, which is worth alerting on whichever outcome you pick here. See Alerting on silent discards.

Three caveats before you reach for it, all inherited from Outcome::Defer itself:

• Nothing bounds the redeliveries. Defer has no circuit breaker, so a handler that times out forever is redelivered forever. Pair it with your own attempt ceiling — or with monitoring on FailReason::Timeout — if the work can genuinely never succeed.
• On sequenced consumers, Defer is not uniform across backends. Kafka, NATS, SNS/SQS and the in-memory backend refuse it — deferral would violate the ordering guarantee, so it is downgraded to Retry with a warning and the override does not spare the retry budget. RabbitMQ's sharded consumer and Redis Streams instead route it to a hold queue without incrementing the retry count, so it can defer indefinitely. Check which half your backend falls in before choosing Defer for an ordered topic.
• With no hold queues configured, Defer falls back to requeue-without-delay, which turns a slow handler into a hot redelivery loop. Declare at least one hold queue on the topic before choosing Defer.

Redis Streams starts from a different default. With no override, a timed-out entry is not mapped to an outcome at all: it is left in the PEL and reclaimed by XAUTOCLAIM after the idle deadline, which already redelivers without consuming retry budget. Setting this option makes Redis route the given outcome instead — so Defer there means the configured hold-queue delay rather than the idle deadline.

That override also puts the consumer and the XAUTOCLAIM reaper on the same deadline: one wants to route an outcome at the handler timeout, the other wants to reclaim the entry and redeliver it. Redis consumers with the override set therefore hold a lease on the entry they are working on, re-asserted every half handler-timeout, which keeps the entry's idle clock away from any reaper's threshold — including a reaper in another process, which shove cannot otherwise see.

This is the one place the timeout configuration has to be consistent to be correct. Keep handler timeouts within 2× of each other across every consumer of a stream and group. A consumer whose timeout is under half of another's reclaims inside the renewal gap and can take an entry its owner was about to resolve. Should that happen the owner does not route its outcome at all — the reclaiming consumer redelivers instead, exactly as it would have with the override unset, so the message is never both dead-lettered and redelivered.

Handler panics are unaffected by this setting and always resolve to Outcome::Retry — a panic is a failed attempt, not a slow one.

With without_handler_timeout() the setting is inert, because no timeout ever fires. That includes shutdown: RabbitMQ and SNS/SQS bound their shutdown drain with a backstop — on both their standard and their sequenced consumers — so a handler that never returns cannot hang the process. That backstop bounds shutdown, not the handler. With deadlines disabled the handler is still running when it expires, so the drain resolves to Outcome::Retry and lets the broker redeliver — it will not Ack away or dead-letter work that is still in flight.

MessageMetadata

Every handle() invocation receives a MessageMetadata describing the delivery:

• retry_count: u32 — how many times this message has been retried; 0 on first delivery.
• delivery_id: String — opaque delivery identifier (AMQP delivery tag, SQS receipt handle, etc.). Useful for logging.
• redelivered: bool — whether the broker flagged this as a redelivery at the transport layer.
• delivery_count: Option<u32> — how many times the broker has delivered this message, including the current delivery. None where the backend has no such counter.
• headers: HashMap<String, String> — string-valued headers attached to the delivery.
• partition: Option<i32> - the partition the record was read from, on Kafka only, and None on every other backend.
• offset: Option<i64> - the record's position in its log: the offset inside its partition on Kafka, the stream sequence on NATS, and None elsewhere.
• timestamp_ms: Option<i64> - when the broker took the message, in milliseconds since the Unix epoch, and None where the backend has no such time.

Each field is filled where the backend has the data, and each has its own availability:

The coordinates name a position in a log, which is what an audit trail or a replay request needs. They are not an attempt counter: the same record redelivered after a rebalance carries the same coordinates. On a dead-letter drain the coordinates are the dead letter's own, the copy's position in the dead-letter topic, while retry_count and delivery_id describe the original message.

Most handlers ignore all of these fields. They become useful when you need to trace a message (headers["x-trace-id"]), implement idempotency checks (headers["x-message-id"] on RabbitMQ), or adjust behavior on redelivery. The retry_count in particular is more reliable than redelivered for detecting retries, because shove defines how it advances. redelivered only reflects the broker's view of the transport. Under RetryStrategy::Republish, shove increments the count itself and carries it in a header on the copy it publishes. Under RetryStrategy::InPlace no copy is published. Kafka then keeps the count in memory for the redelivery. An external NATS stream reads it from the JetStream delivery information, which counts every redelivery.

retry_count vs delivery_count

The two counters answer different questions, and neither is a total-attempts-ever number.

retry_count is shove's retry budget, the field max_retries and DLQ routing are measured against. How it advances depends on the retry strategy and the backend. Under RetryStrategy::Republish, the default on a shove-owned topology, it advances only when Outcome::Retry publishes an incremented copy. Redis, RabbitMQ, SQS and the in-process broker route the copy through a hold queue for the tier's delay. Kafka has no hold queue and republishes the copy into the consumed topic after the delay. NATS has no hold stream either. It holds the original un-acked for the delay, with progress acknowledgements, then publishes the copy to the same subject and acks the original. In every case the count survives the hop because shove carries it in a header. Under RetryStrategy::InPlace on Kafka, implied by an external topology, no copy is published. The count advances on each Retry the handler's task waits out, and is kept in memory for the redelivery. Under RetryStrategy::InPlace on an external NATS stream, the count is the number of redeliveries JetStream reports. A Defer nak and an ack_wait expiry are redeliveries there, so they advance it too.

delivery_count is the broker's own attempt counter, so it sees deliveries shove never initiated — a consumer crash, an ack that never landed, a visibility timeout expiring. It is reported where the backend supplies one:

Treat None as "unknown", never as zero. Where it is reported, a first delivery is Some(1).

Because the count belongs to a broker-level message, anything that creates a new one restarts it at 1. Under RetryStrategy::Republish, the default on a shove-owned topology, Retry publishes an incremented copy and so resets delivery_count; retry_count is the field that survives a retry. Under RetryStrategy::InPlace, Kafka on an external or opted-in topology and NATS on an external stream, no copy is published and the same record comes back: Kafka keeps retry_count in memory, and NATS reads it from this very count, which climbs with every redelivery there. Defer resets delivery_count where deferring re-sends the message (SQS) and preserves it where deferring naks in place (NATS, and the in-process broker, which models NATS here).

That last case is what delivery_count is for; see Bounding a Defer loop.

A consumer running with RetryStrategy::InPlace publishes no copy on Retry or Defer: on Kafka, where an external topology implies it and a shove-owned one may opt into it, the handler's task waits and hands the record back; on an external NATS stream the broker redelivers after a delayed Nak. The same record comes back in place. On Kafka its retry_count is kept in memory for the redelivery. On an external NATS stream it is read from the JetStream delivery information, which counts every redelivery.

Dead-letter handling

When a message exhausts its retries or is permanently rejected, it lands in the DLQ. The handle_dead method is your hook for what happens next: send an alert, write to a quarantine store, push to an investigation queue, page an on-call engineer.

impl MessageHandler<Orders> for Handler {
    type Context = AppState;

    async fn handle(&self, msg: Order, _: MessageMetadata, _: &AppState) -> Outcome {
        // ... normal processing ...
        Outcome::Ack
    }

    async fn handle_dead(&self, msg: Order, meta: DeadMessageMetadata, ctx: &AppState) {
        tracing::error!(
            order_id = %msg.id,
            reason = meta.reason.as_deref().unwrap_or("unknown"),
            death_count = meta.death_count,
            "dead-lettered order",
        );
        // alert, write to quarantine table, etc.
    }
}

DeadMessageMetadata wraps the base MessageMetadata (accessible via meta.message) and adds reason, original_queue, and death_count.

The dead-letter consumer loop is driven separately from the main consumer loop: consumer.run_dlq::<T>(). The message is always acked from the DLQ after handle_dead returns.

Audited handlers

For compliance, debugging, or fraud investigation, you may need a record of every message delivery: what it contained, how long the handler ran, what the outcome was. MessageHandlerExt::audited(audit_handler) wraps any handler in an audit layer. The wrapped handler has the same Outcome contract and the same Context type as the original. Auditing is purely compositional — it does not change behaviour, only adds a side-channel write per invocation.

use shove::MessageHandlerExt;

let handler = OrderHandler.audited(MyAuditSink);
group.register::<Orders, _>(cfg, move || handler.clone()).await?;

See Audit Logging for a full explanation of the AuditHandler trait and audit record schema.

For the other side of the registration — how broker.consumer_group() and with_context work — see The Broker<B> Pattern.