# shove > Type-safe async pub/sub for Rust on RabbitMQ, AWS SNS+SQS, NATS JetStream, Apache Kafka, or in-process. ## Getting Started You can run a real `shove` publisher and consumer in 60 seconds against the **in-process backend** — no Docker, no AWS credentials, no config. This page is a copy-paste walkthrough. ### Add the dependency ```sh cargo add shove --features inmemory cargo add tokio --features full cargo add serde --features derive cargo add tracing-subscriber cargo add tokio-util --features rt ``` ### Define a topic A topic binds a Rust message type to a queue topology. The `define_topic!` macro is the fast path; this example shows the manual `Topic` impl for clarity. ```rust // [!include ~/examples/inmemory/basic.rs:topic] ``` Reference: [Topics & Topology](/concepts/topics). ### Write a handler ```rust // [!include ~/examples/inmemory/basic.rs:handler] ``` The handler returns an [`Outcome`](/concepts/outcomes). `Ack` removes the message from the queue. ### Wire it up ```rust // [!include ~/examples/inmemory/basic.rs:main] ``` This connects to the in-process broker, declares the topology, registers a one-worker consumer group, publishes five messages, and exits cleanly when all five are processed. ### Run it ```sh cargo run --example inmemory_basic --features inmemory ``` Expected output: ```text received #0: hello 0 received #1: hello 1 received #2: hello 2 received #3: hello 3 received #4: hello 4 ``` ### What's next * **[Core concepts](/concepts/topics)** — what `Topic`, `Outcome`, and `Broker` mean. * **[Pick a real backend](/backends/choosing)** — RabbitMQ, SNS+SQS, NATS, Kafka, or Redis/Valkey. Same handler, swap one marker. * **[Sequenced topics](/guides/sequenced)** — strict per-key ordering when message order matters. * **[Audit logging](/guides/audit)** — wrap any handler with structured per-delivery records. ## Reference The full API reference for `shove` is auto-generated rustdoc on **[docs.rs/shove](https://docs.rs/shove)**. ### Headline types * [`Broker`](https://docs.rs/shove/latest/shove/struct.Broker.html) — generic entry point parameterized on the backend marker * [`Publisher`](https://docs.rs/shove/latest/shove/struct.Publisher.html) — typed publish API * [`TopologyDeclarer`](https://docs.rs/shove/latest/shove/struct.TopologyDeclarer.html) — idempotent topology setup * [`ConsumerSupervisor`](https://docs.rs/shove/latest/shove/struct.ConsumerSupervisor.html) — N-poller fan-out (SQS, all backends) * [`ConsumerGroup`](https://docs.rs/shove/latest/shove/struct.ConsumerGroup.html) — coordinated min/max-bounded groups * [`SupervisorOutcome`](https://docs.rs/shove/latest/shove/struct.SupervisorOutcome.html) — shutdown result with `exit_code()` ### Core enums and traits * [`Outcome`](https://docs.rs/shove/latest/shove/enum.Outcome.html) — `Ack`, `Retry`, `Reject`, `Defer` * [`MessageHandler`](https://docs.rs/shove/latest/shove/trait.MessageHandler.html) — handler trait, parameterized on `Topic` * [`MessageHandlerExt`](https://docs.rs/shove/latest/shove/trait.MessageHandlerExt.html) — fluent helpers (`.audited(...)`) * [`Topic`](https://docs.rs/shove/latest/shove/trait.Topic.html), [`SequencedTopic`](https://docs.rs/shove/latest/shove/trait.SequencedTopic.html) — topic traits * [`TopologyBuilder`](https://docs.rs/shove/latest/shove/struct.TopologyBuilder.html) — `.hold_queue`, `.sequenced`, `.routing_shards`, `.dlq` * [`ConsumerOptions`](https://docs.rs/shove/latest/shove/struct.ConsumerOptions.html) — per-consumer tuning ### Per-backend modules * [`shove::rabbitmq`](https://docs.rs/shove/latest/shove/rabbitmq/) — RabbitMQ backend * [`shove::sns`](https://docs.rs/shove/latest/shove/sns/) — AWS SNS+SQS backend * [`shove::nats`](https://docs.rs/shove/latest/shove/nats/) — NATS JetStream backend * [`shove::kafka`](https://docs.rs/shove/latest/shove/kafka/) — Apache Kafka backend * [`shove::redis`](https://docs.rs/shove/latest/shove/redis/) — Redis/Valkey Streams backend * [`shove::inmemory`](https://docs.rs/shove/latest/shove/inmemory/) — in-process backend > Docs on this site follow `main`. For pinned versions, use the version dropdown on docs.rs. ## Backend Ops Notes Setup is covered by the per-backend overview pages. This page covers what tends to bite in production: which knobs to set before you go live, which surprises to anticipate, and which monitoring to put in place. One section per backend. ### RabbitMQ **Management plugin required for autoscaler.** The autoscaler queries the RabbitMQ HTTP management API (`/api/queues`) for queue depth. Without the plugin, the autoscaler errors at startup. Enable it with: ```sh rabbitmq-plugins enable rabbitmq_management ``` The management API is available on port `15672` by default. If you run RabbitMQ behind a network policy that blocks that port, the autoscaler will not start. **Auto-reconnect is automatic at the consumer level.** Transient broker failures — network blips, broker restarts — recover without operator intervention. The consumer reconnects and resumes processing. Don't add wrapping retry loops at the application level; they interfere with shove's internal reconnect logic and can cause duplicate deliveries. **TLS via `rustls-ring`** is the default `lapin` feature used by shove. For environments that require OpenSSL (e.g. FIPS-compliant infrastructure) or vendored alternatives, configure the TLS backend downstream in your own `Cargo.toml` feature selection. **Cluster setups.** Point `RabbitMqConfig` at a load-balanced cluster URL (e.g. via HAProxy or a DNS-based round-robin). shove uses a single AMQP connection per `Broker\`. For HA with automatic failover, use a clustered broker with quorum queues — the connection layer handles reconnect, but the queue must be replicated to survive a node failure. **Hold queue dead-letter exchange (DLX) is configured at topology declaration.** shove sets queue arguments (including `x-dead-letter-exchange` and `x-message-ttl`) when it declares hold queues. RabbitMQ does not allow modifying queue arguments in place after creation. If you need to change hold-queue delays, you must delete and recreate the queues — or use a different queue name to force a fresh declaration. ### AWS SNS + SQS **Visibility timeout MUST exceed handler timeout.** If your handler can take 30 s, set the SQS visibility timeout to at least 60 s. If the handler is still running when the visibility timeout expires, the message becomes visible again and another consumer picks it up — producing duplicate work. Configure via the SQS console, Terraform, or CloudFormation: ```text VisibilityTimeout: 60 # seconds — set to 2× your worst-case handler time ``` **FIFO queues have a 300 msg/s default cap.** High-throughput sequenced topics that need sequenced delivery via FIFO will hit this limit before exhausting hardware capacity. Request a quota increase via AWS Support, or switch to standard SQS queues (no ordering guarantee) if ordering is not required. **`endpoint_url` is for local dev only.** The `endpoint_url` field on `SnsConfig` points shove at a custom SQS/SNS endpoint (typically LocalStack on `http://localhost:4566`). Omit it in production, or you will point all traffic at localhost. **Region matters.** SQS queues are regional. Consuming a queue in a different region than your application's default region requires explicit configuration. Cross-region fanout is handled at the SNS topic ARN level, not in shove. **Dead-letter queues are wired at topology-declaration time.** When you declare a topic with `.dlq()`, shove creates the DLQ queue (or its FIFO equivalent for sequenced topics), attaches a `RedrivePolicy` with `maxReceiveCount` set to shove's default (`DEFAULT_MAX_RECEIVE_COUNT`), and points the main queue at it. You don't need to provision the DLQ separately in Terraform/CloudFormation. If you want a different `maxReceiveCount` or a pre-existing DLQ ARN, configure it via the SQS console or IaC tooling AFTER `declare` runs — but the default wiring is automatic. **IAM permissions.** The consumer role needs the following permissions on each queue it reads from: ```text sqs:ReceiveMessage sqs:DeleteMessage sqs:ChangeMessageVisibility sqs:GetQueueUrl sqs:GetQueueAttributes ``` If the same role publishes, add `sns:Publish` on each SNS topic it publishes to. ### NATS JetStream **Stream provisioning happens at topology declaration.** When `declare_topology` runs, shove creates the JetStream stream if it does not already exist. For policy-driven setups — limits-based retention, interest-based retention, work-queue retention — pre-create the stream with the correct policy before starting the application. shove will detect the existing stream and use it as-is without overwriting configuration. **`Nats-Msg-Id` dedup window is 120 s.** shove stamps this header on every publish (a fresh UUID per call). JetStream de-duplicates within the stream's `duplicate_window` (default 120 seconds), so a publisher that retries the same logical message inside that window will not produce a duplicate delivery. If you need deduplication across application-level retries, pass a stable id via `publish_with_headers` so each retry sets the same `Nats-Msg-Id`. If your handler routinely takes longer than 120 s, raise the stream's `duplicate_window` to match: ```text duplicate_window: 600s # example: 10-minute dedup window ``` **Stream-side vs consumer-side limits.** Storage size and per-message size are set on the JetStream stream itself (configured outside shove). `max_ack_pending`, however, is set by shove on the durable pull consumer at consumer-creation time, derived from `prefetch_count` (and from `max_consumers × prefetch_count` for groups). To override per-consumer, use `ConsumerOptions::::with_max_ack_pending(...)`. **Subject naming.** shove uses `{queue}.shard.{N}` subjects for sequenced topics. Do not reuse these subjects for other applications or publishers outside of shove — you will interfere with sequencing semantics. **Consumer durability.** shove uses durable pull consumers. The consumer name is derived from the topic and group configuration. On restart, the NATS server remembers the consumer's last acknowledged sequence number, so processing resumes where it left off rather than re-reading from the start of the stream. ### Apache Kafka **Partition count is set at provisioning time** and is rarely changed after the fact. It caps the maximum group concurrency: a 4-partition topic can have at most 4 active consumers in a consumer group simultaneously. Over-partitioning wastes broker resources; under-partitioning limits scale. Plan partition count for peak load plus headroom before creating the topic. **Consumer-group rebalances during scale events cause brief delivery pauses.** When a consumer joins or leaves the group — including during autoscaling — Kafka triggers a group rebalance, during which no consumer in the group processes messages. Rebalances typically complete in under 1 s, but aggressive autoscaling that adds and removes consumers rapidly can cause near-continuous rebalance churn. Set scale cooldowns to allow the group to stabilize between scale events. **`cmake` is required on Windows.** The `rdkafka` crate links `librdkafka`, which requires `cmake` during the build on Windows. Linux and macOS use the standard `librdkafka` build path without additional prerequisites. **SASL/TLS.** Enable the `kafka-ssl` feature for TLS transport. `KafkaSasl` configuration supports SCRAM-SHA-256 and SCRAM-SHA-512. GSSAPI/Kerberos requires activating the `rdkafka/gssapi` feature downstream in your own `Cargo.toml`. For Amazon MSK with IAM authentication, enable the `kafka-msk-iam` feature instead — see the [Kafka backend page](/backends/kafka) for setup details. **`librdkafka` system dependency.** On Debian/Ubuntu CI runners, install the SASL development headers before building: ```sh apt-get install -y libsasl2-dev ``` On macOS, the dependency is typically satisfied via Homebrew. `librdkafka` is a runtime dependency — it is not statically bundled. **Consumer offsets are committed asynchronously.** `librdkafka` commits offsets in the background. On clean shutdown, shove flushes pending commits before exiting. On a crash or SIGKILL, expect at-least-once redelivery from the last committed offset when the consumer restarts. Design handlers to be idempotent or use the dedup layer to handle this. ### Redis / Valkey Streams **Redis 6.2 or newer is required.** shove uses `ZRANGE … BYSCORE` for hold-queue polling, a command introduced in Redis 6.2. The version is checked at connection time — if the server is older, `Broker::new` returns an error with a clear message rather than failing later with a cryptic protocol error. Any Valkey release is compatible (Valkey reports a 7.x-equivalent `redis_version` for backwards compatibility). **Standalone, TLS, and cluster modes are all supported.** Use `RedisMode::Standalone` with a `redis://` or `rediss://` URL for single-node and Sentinel setups. Use `RedisMode::Cluster` with a list of seed node URLs for Redis Cluster. shove opens one multiplexed connection for non-blocking operations (XADD, XACK, ZADD) and a separate dedicated connection per consumer loop for blocking reads (XREADGROUP with `BLOCK`). **Hold queues use Sorted Sets, not Streams.** Messages waiting for delayed retry are stored in a Redis Sorted Set with the redeliver-at timestamp as the score. A background requeuer task polls the set on each node and XADDs due entries back to the appropriate shard stream. If two requeuer instances run concurrently (e.g. during a rolling restart), the same entry may be XADDed twice — this is expected at-least-once behaviour; handlers must be idempotent. **Consumer groups are created automatically.** At topology declaration, shove creates an XGROUP on each shard stream if one does not already exist. The group name defaults to `"shove"` and can be overridden via `RedisConfig::group`. Do not reuse the same group name across unrelated applications that share a Redis instance. **Stream memory is bounded by outstanding work.** While consumers run, shove trims entries that every consumer group on the stream has acknowledged, so a stream's size tracks its backlog rather than its lifetime throughput — no external `XTRIM` policy is needed. Trimming waits for the slowest group, never touches pending (in-flight) entries, and never touches DLQ streams. The same background task reclaims entries left pending past the handler timeout (e.g. after a consumer crash) and redelivers them; keep the handler timeout consistent across every consumer of a stream and group, including other processes. ### In-process **No durability.** Messages live entirely in process memory. Process exit — whether clean or via a crash — means message loss. State this explicitly on every internal review of "should we use the in-process backend in production?" — the answer is almost always no for distributed or stateful workloads. **No cross-process delivery.** Only consumers running in the same OS process see messages published to an in-process broker. There is no IPC, no network transport, no persistence layer. Two separate service instances cannot share an in-process topic. **Use cases:** integration tests (fast, no broker required), single-process CLI tools that use pub/sub internally for decoupling, and prototyping before wiring up a real broker. Not suitable for distributed services or anything that must survive process restart. **Memory footprint.** Each topic holds undelivered messages in memory. A publisher that outpaces its consumers will grow the backlog without bound. shove does not enforce backpressure on publishers in the in-process backend — if this is a concern, add an external rate limiter or switch to a broker-backed backend. ### What's next * [Performance Tuning](/ops/performance) — throughput levers and measurement methodology * [Integrate with your stack](/backends/choosing) — per-broker setup overview ## Performance Tuning Most `shove` performance questions resolve to four levers: prefetch count, worker count, the concurrent-processing flag, and (negatively) transactional mode. Knowing which lever to pull first — and which to ignore — is what this page is for. Numbers below are from a single-node setup and intended as a starting baseline; always measure on your own hardware and workload. ### Headline numbers Measured on a MacBook Pro M4 Max, single RabbitMQ node via Docker, Rust 1.91. | Handler | 1 worker, prefetch=1 | 1 worker, prefetch=20 | 8 workers, prefetch=20 | 32 workers, prefetch=40 | | ---------------- | -------------------- | --------------------- | ---------------------- | ----------------------- | | Fast (1–5 ms) | 179 msg/s | 2,866 msg/s | 19,669 msg/s | 29,207 msg/s | | Slow (50–300 ms) | 6 msg/s | 75 msg/s | 544 msg/s | 4,076 msg/s | | Heavy (1–5 s) | 0.4 msg/s | 5 msg/s | 21 msg/s | 199 msg/s | Reproducible via `cargo run -q --example rabbitmq_stress --features rabbitmq`. Results vary by hardware and broker config. ### The throughput levers In order of impact: * **`prefetch_count`** (largest lever): how many unacked messages the broker pre-delivers to a consumer. Higher values mean more in-flight work and fewer round-trips waiting for acks. * **Worker count** (linear scaling on handler-bound workloads): more concurrent handlers means more throughput when handlers are CPU- or I/O-bound. Scaling is close to linear until you hit a broker or partition limit. * **`concurrent_processing` flag**: `true` allows multiple handler invocations to run concurrently within a single worker, bounded by `prefetch_count`. `false` (default) processes one message at a time per worker. * **Transactional mode** (negative lever): the `rabbitmq-transactional` feature reduces throughput roughly 10–15× per channel compared to non-transactional. The trade-off is routing safety — every ack and publish land in the same AMQP transaction. ### prefetch\_count deep-dive `prefetch_count` tells the broker how many unacked messages it may deliver to a single consumer before waiting for acks. With `prefetch_count = 1`, the broker waits for an ack before sending the next message — every handler invocation pays a full network round-trip. With higher values, the consumer has a local buffer of pre-delivered messages and can begin processing the next one immediately after finishing the previous. Why too high is also a problem: if one consumer holds a large unacked backlog, other consumers in the same group receive less work — distribution becomes uneven. Very high values also increase per-consumer memory usage and slow graceful shutdown (all unacked messages must drain). **Recommended starting point: 10–40.** Tune based on handler latency: short-latency handlers benefit more from high prefetch; long-latency handlers (where the bottleneck is the handler itself) need high worker counts instead. **Worked example.** Handler average latency 10 ms, network RTT 1 ms: * `prefetch_count = 1`: throughput capped at 1000 / (10 + 1) ≈ 90 msg/s — the handler is idle while waiting for the next delivery. * `prefetch_count = 20`: the consumer always has work ready; the handler runs at close to its maximum rate (\~100 msg/s). ### Workers — when to add them Add workers when the topic backlog grows faster than the consumer drains it. That's the signal that the handler is the bottleneck. A single shove consumer rarely saturates the network or broker on its own. The bottleneck is almost always handler latency × concurrency. Each additional worker adds a parallel handler goroutine; throughput scales linearly until you hit: * **Broker quota** — e.g. RabbitMQ per-connection channel limits. * **Partition count** (Kafka) — a 4-partition topic can only be consumed by 4 group members simultaneously. * **Sub-queue count** (RabbitMQ sequenced) — the number of consistent-hash shards limits sequenced consumer parallelism. ### with\_concurrent\_processing(true) When `concurrent_processing` is `true`, a single worker can have multiple handler invocations in-flight simultaneously, limited by `prefetch_count`. The benefit is full utilization of the pre-fetched buffer: while one handler is awaiting a database response, another can start. **Use when:** handlers are pure I/O-bound — waiting on a database query, an HTTP call, an external API. The async runtime multiplexes tasks efficiently. **Don't use when:** handlers are CPU-bound. Concurrent tasks without additional OS threads don't add parallelism — they compete for the same thread pool and can increase latency without improving throughput. For CPU-bound work, increase worker count instead. ### Measurement methodology Reproduce locally: ```sh cargo run -q --example rabbitmq_stress --features rabbitmq ``` The benchmark publishes 10,000 messages, measures elapsed time per (worker, prefetch) configuration, and prints msg/s for each combination. Vary the parameters in source and re-run. Hardware, broker version, network setup, and OS scheduler all affect results. The numbers in the headline table are a starting reference — measure on your own setup before committing to a configuration. ### What NOT to tune first Ranked by typical impact: * **Serde format** — JSON serialization overhead is negligible compared to network round-trip. Switching to bincode or MessagePack rarely moves the needle for normal message sizes. * **Connection pool size** — a single AMQP channel or Kafka producer handles thousands of msg/s. Pooling is for fault-tolerance (connection failover), not raw throughput. * **TLS** — modern TLS handshake and record overhead is negligible for the message sizes and rates shove handles. Disable only if profiling shows it in a hot path. * **Tracing** — structured logging at `info` level adds less than 1% overhead. Disable only at `debug` level if profiling shows it dominates. ### What's next * [Backend Ops Notes](/ops/backends) — per-backend production knobs and gotchas * [Consumer Groups & Autoscaling](/guides/groups) — autoscaler configuration and scaling policies * [RabbitMQ stress example](/backends/rabbitmq/examples/stress) — walk through the benchmark harness ## Audit Logging Compliance requirements, fraud investigation, and debugging all share the same need: a durable record of what happened, when, and what the outcome was. `shove` provides a transparent wrapper — `Audited\` — that captures this for every handler invocation, regardless of backend. The wrapper is a drop-in: it implements `MessageHandler\` with the same associated `Context` type as the inner handler, so it fits anywhere a handler fits. ### What audit logging captures `AuditRecord\` is the struct emitted to your audit sink on every delivery: ```rust,no_run pub struct AuditRecord { pub trace_id: String, pub topic: String, pub payload: M, pub metadata: MessageMetadata, pub outcome: Outcome, pub duration_ms: u64, pub timestamp: DateTime, } ``` Field by field: * **`trace_id`** — a UUID v4 generated per delivery, or the value of the `x-trace-id` header if one was set at publish time. Correlates retries of the same message and connects to distributed tracing systems. * **`topic`** — the queue name from the topology (e.g. `"orders"`). Useful for routing audit records from multiple topics to a single sink without losing context. * **`payload`** — the deserialized message, type `M`. The audit sink receives the full payload. For `ShoveAuditHandler`, `M` is erased to `serde_json::Value` so heterogeneous records share one topic. * **`metadata`** — the `MessageMetadata` for this delivery: `retry_count`, `delivery_id`, `redelivered` flag, and broker headers. * **`outcome`** — what the handler returned (`Ack`, `Retry`, `Reject`, or `Defer`). Lets you filter for failures without parsing logs. * **`duration_ms`** — wall-clock time the `handle()` call took, in milliseconds. Use for latency percentiles and SLO tracking. * **`timestamp`** — UTC timestamp when the handler completed. ### AuditHandler\ trait Implement `AuditHandler\` for your chosen persistence sink: ```rust,no_run pub trait AuditHandler: Send + Sync + 'static { async fn audit(&self, record: &AuditRecord) -> Result<(), ShoveError>; } ``` The trait is async and takes a shared reference — your implementation can use an `Arc`-wrapped client internally. Implementations for common sinks: ```rust,no_run // Write to a database (e.g. PostgreSQL). struct PostgresAudit { pool: Arc, } impl AuditHandler for PostgresAudit { async fn audit(&self, record: &AuditRecord) -> Result<(), ShoveError> { sqlx::query!( "INSERT INTO audit_log (trace_id, topic, outcome, duration_ms, ts) VALUES ($1, $2, $3, $4, $5)", record.trace_id, record.topic, format!("{:?}", record.outcome), record.duration_ms as i64, record.timestamp, ) .execute(&*self.pool) .await .map_err(|e| ShoveError::Connection(e.to_string()))?; Ok(()) } } ``` The `AuditHandler` bound is per-topic: a different implementation can be used for each topic, or you can implement it generically over `T: Topic` if the record schema is homogeneous. ### Strict auditing Returning `Err` from `audit()` causes the consumer to **retry the message** — the outcome of the business handler is discarded and replaced with `Outcome::Retry`. Audit records are never silently dropped. This is the right default for "every action must be logged" requirements — particularly relevant for regulated industries where a gap in the audit trail is a compliance violation, not just an inconvenience. If the audit sink is unavailable (database outage, network partition), message processing halts rather than proceeding without a record. The message will retry (with backoff if hold queues are configured) until the sink recovers. If you need **lossy auditing** — where a sink failure should not block processing — catch internal errors in your `audit()` implementation, log them, and return `Ok(())`: ```rust,no_run impl AuditHandler for BestEffortAudit { async fn audit(&self, record: &AuditRecord) -> Result<(), ShoveError> { if let Err(e) = self.inner_write(record).await { tracing::warn!(error = %e, "audit write failed (best-effort — continuing)"); } Ok(()) // never block processing } } ``` Because `Err` causes a retry, **handlers wrapped with `Audited` must be idempotent**. If the audit sink fails after the business handler runs, the message is retried and the handler runs again. Use a stable `delivery_id` (available on `metadata.delivery_id`) to deduplicate business-side effects. `Audited` also supports an optional audit timeout via `Audited::with_audit_timeout(duration)`. If the audit handler takes longer than the timeout, the original handler outcome is returned and the audit failure is logged — the message is not retried due to a slow sink: ```rust,no_run let handler = OrderHandler.audited(SlowSink).with_audit_timeout(Duration::from_secs(2)); ``` ### MessageHandlerExt::audited The `MessageHandlerExt` trait provides a fluent `.audited(sink)` method on any `MessageHandler`: ```rust,no_run use shove::MessageHandlerExt; let handler = OrderHandler::new(state).audited(PostgresAudit { pool: pool.clone() }); ``` This is equivalent to `Audited::new(OrderHandler::new(state), PostgresAudit { pool: pool.clone() })` but reads more naturally in registration chains. The wrapped handler has the same `type Context` as the inner handler — no changes are needed at the consumer group registration site. ### Trace ID propagation By default, the audit wrapper generates a fresh UUID v4 as the `trace_id` on each delivery. To propagate a trace ID from the publisher across retries — so all retries of the same message share one trace ID with the original delivery — set the `x-trace-id` header at publish time: ```rust,no_run use std::collections::HashMap; let mut headers = HashMap::new(); headers.insert("x-trace-id".to_string(), existing_trace_id.to_string()); publisher.publish_with_headers::(&order, headers).await?; ``` The consumer surfaces broker headers via `MessageMetadata::headers`, which is a `HashMap`. The audit wrapper reads `headers["x-trace-id"]` and uses it as the record's `trace_id` if present. This means all retries of the message (which preserve headers through hold-queue hops) share the same trace ID, making it straightforward to reconstruct the full delivery history from the audit log. ### ShoveAuditHandler — the built-in dogfood backend Behind the `audit` Cargo feature flag, `shove` ships `ShoveAuditHandler\`, an `AuditHandler` that publishes audit records as messages on the dedicated `shove-audit-log` topic using any backend's `Publisher\`. Enable it: ```toml [dependencies] shove = { version = "0.x", features = ["audit"] } ``` Use it: ```rust,no_run use shove::audit::ShoveAuditHandler; let audit = ShoveAuditHandler::new(publisher.clone()); // Or from a reference: let audit = ShoveAuditHandler::for_publisher(&publisher); let handler = OrderHandler::new(state).audited(audit); ``` `ShoveAuditHandler` erases the payload type to `serde_json::Value` (via `serde_json::to_value`) so that a single `shove-audit-log` topic can carry records from all topics without needing a different message type per topic. The `AuditLog` topic definition: ```rust,no_run // Defined in shove::audit when the `audit` feature is enabled. define_topic!( pub AuditLog, AuditRecord, TopologyBuilder::new("shove-audit-log").dlq().build() ); ``` The `shove-audit-log` topic is itself consumable. You can subscribe a consumer to it to fan out audit records to a persistent store, alert on specific outcomes (e.g. a spike in `Reject` outcomes), or feed them into an analytics pipeline. One guard is in place to prevent infinite recursion: `Audited` skips its audit handler when the topic being consumed is `AuditLog` itself, so wrapping an `AuditLog` handler with an `Audited` that uses `ShoveAuditHandler` does not produce a loop. ### Wiring with consumer groups Wrapping a handler with `Audited` requires no changes on the consumer group side. The wrapped handler is still a `MessageHandler\`, and it's registered identically: ```rust,no_run let mut group = broker.consumer_group(); let pool = pool.clone(); group .register::( ConsumerGroupConfig::new( InMemoryConsumerGroupConfig::new(1..=4).with_prefetch_count(10), ), move || OrderHandler::new(state.clone()).audited(PostgresAudit { pool: pool.clone() }), ) .await?; ``` The factory closure runs once per consumer instance. If your audit sink is `Clone`, clone it inside the factory. If it holds an `Arc` internally, the clone is a refcount bump. ### Full walkthrough \[!include \~/examples/inmemory/audited\_consumer.rs] The example wraps `Inner` — a simple handler that increments a counter — with `StdoutAudit`, which prints the trace ID, outcome, and duration on every delivery. The key line is line 80: `Inner { count: c.clone() }.audited(StdoutAudit)`. The resulting `Audited` is returned from the factory closure and registered directly with the consumer group. No other changes are needed anywhere. ### What's next * [Handlers & Context](/concepts/handlers) * [Observability](/guides/observability) * [InMemory audited example](/backends/inmemory/examples/audited) * [Retries, Hold Queues & DLQs](/guides/retries) ## Exactly-Once (RabbitMQ) Charging a payment card twice, sending a confirmation email to a customer twice, triggering an external API that has no idempotency tokens — these are the cases where at-least-once delivery causes real harm and exactly-once delivery matters. `shove` is at-least-once by default on every backend. RabbitMQ has an **opt-in transactional mode** that makes routing decisions atomic, eliminating the publish-then-ack race that produces duplicates. It is slower — expect roughly 10–15× lower throughput per channel compared to the default confirm-mode consumer — so reserve it for handlers with irreversible side effects where duplicates cause real harm. ### The publish-then-ack race Under default at-least-once semantics, `shove`'s consumer uses two separate broker operations to route a message that needs to be retried: 1. Handler returns `Outcome::Retry`. 2. Consumer publishes the message (with incremented retry count header) to the appropriate hold queue. 3. Consumer sends ack for the original delivery — removing it from the main queue. These two operations are not atomic. If the consumer process crashes, restarts, or loses connection between steps 2 and 3: * The broker redelivers the original message (it was never acked). * The copy in the hold queue is also in flight. Both copies will eventually be processed by the handler. The handler runs **twice** for the same logical message. For idempotent handlers, this is harmless. For handlers with irreversible side effects — charging a payment card, sending an email, calling an external API that doesn't support idempotency tokens — it produces a duplicate. The same race applies to `Outcome::Defer` and `Outcome::Reject` (reject-to-DLQ also involves a publish + ack pair). ### When you need exactly-once routing Consider enabling transactional mode when: * **Ledger writes without idempotency keys** — a double-write produces two distinct entries and corrupts the balance. * **Outbound email or SMS without dedup** — the recipient receives the message twice. * **Calls to APIs that don't support idempotency tokens** — the API has no way to detect or reject a duplicate request. * **State-machine transitions where a duplicate transition is invalid** — e.g. `approved → shipped` should happen exactly once; a duplicate ships the order again. Do NOT enable it for: * **Already-idempotent handlers** — the overhead is wasted if duplicates are safe. * **High-throughput topics** — the 10–15× throughput reduction may be unacceptable. Profile first. * **Backends other than RabbitMQ** — the feature only exists on RabbitMQ in `shove` today. ### rabbitmq-transactional feature Enable the feature in `Cargo.toml`: ```toml [dependencies] shove = { version = "0.11", features = ["rabbitmq-transactional"] } ``` With the feature enabled, `ConsumerOptions\` gains the `.with_exactly_once()` builder method. When set, the consumer channel is put in AMQP transaction mode (`tx_select`) before the first delivery. Every routing decision — publish-to-hold-queue AND ack/nack of the original — is wrapped in a `tx_commit`. The broker either applies both operations or neither. The publish-then-ack race is gone. The transactional code path is implemented in the RabbitMQ consumer and router. The observable behavior is identical to the default confirm-mode consumer: messages are acked exactly once, retries use hold queues, rejections go to the DLQ. The only difference is that these operations are now atomically grouped. ### Why RabbitMQ only Other backends don't expose equivalent broker-level transactions in `shove` today: * **SQS** has some transactional semantics via FIFO deduplication tokens, but the model is different: dedup happens at the publisher side, not the consumer routing layer. * **Kafka** has transactional producers (cross-partition atomic writes), but they are scoped to producer batches, not the consumer ack/publish pair that `shove` needs to make atomic. * **NATS JetStream** does not expose transactions. * **InMemory** is single-process; there is no network-level race to eliminate. If exactly-once routing is a hard requirement for a non-RabbitMQ backend, you must implement idempotency in the handler using a distributed store (Redis, database). Track the `delivery_id` (from `MessageMetadata::delivery_id`) and skip processing if you've already seen it. ### Wiring Once the feature is enabled, opt in per-topic via `ConsumerOptions::::with_exactly_once()`: ```rust,no_run use shove::{ConsumerOptions, RabbitMq}; let opts = ConsumerOptions::::new() .with_max_retries(3) .with_exactly_once(); supervisor.register::(PaymentHandler::new(), opts)?; ``` No other changes are required. The feature flag gates the method: if `rabbitmq-transactional` is not enabled, `with_exactly_once()` does not exist and code that calls it will not compile. `with_exactly_once()` can be combined with all other `ConsumerOptions` builder methods. The transactional mode applies only to the routing layer — prefetch count, handler timeout, and max retries work as usual. If a `tx_commit` itself fails (e.g. the broker drops the channel mid-commit), the consumer surfaces the error and reconnects rather than silently dropping the routing decision. On shutdown the consumer drains in-flight transactions inside the supervisor's drain timeout so pending commits land before the process exits. ### What it does NOT give you Transactional mode fixes the **routing race** in `shove`'s consumer layer. It does not make your business handler idempotent. Consider a handler that: 1. Charges a payment card. 2. Publishes a confirmation event to another queue. 3. Returns `Outcome::Ack`. With transactional mode, step 3 (the ack) is atomic with any hold-queue routing. But there is no transaction spanning step 1 (the card charge) and step 3 (the ack). If the handler crashes after charging the card but before returning `Outcome::Ack`, the broker redelivers the message. The card is charged again on the next attempt. **Idempotent handlers are still required.** Store idempotency keys in your payment processor or use the `delivery_id` to track which messages have already been processed: ```rust,no_run async fn handle(&self, msg: Payment, meta: MessageMetadata, ctx: &State) -> Outcome { // Idempotency check — skip if already processed. if ctx.payments.is_processed(&meta.delivery_id).await { return Outcome::Ack; } // Charge the card. ctx.payments.charge(&msg, &meta.delivery_id).await?; Outcome::Ack } ``` Transactional mode and idempotent handlers are complementary, not alternatives. Use both for the strongest guarantee. ### Full walkthrough \[!include \~/examples/rabbitmq/exactly\_once.rs] The opt-in is on lines where `ConsumerOptions::::new().with_exactly_once()` is passed to `supervisor.register`. Three topics demonstrate the three paths: ack, retry-then-ack, and reject-to-DLQ — all under transactional mode. The hold queue on `RetryPaymentTopic` has a 2-second TTL; the example waits 8 seconds to let the retry fire before draining. ### What's next * [Retries, Hold Queues & DLQs](/guides/retries) * [RabbitMQ overview](/backends/rabbitmq) * [RabbitMQ exactly-once example](/backends/rabbitmq/examples/exactly-once) ## 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\` wrapper in the top-level API holds the backend-specific inner config: ```rust,no_run ConsumerGroupConfig::new(inner_config) ``` For the InMemory backend: ```rust,no_run 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: ```rust,no_run 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: | Backend | Autoscale signal | | -------------- | ------------------------------------------------- | | RabbitMQ | Queue depth via management HTTP API | | SQS | `ApproximateNumberOfMessages` attribute | | NATS JetStream | Pending messages on the durable consumer | | Apache Kafka | Consumer lag (committed offset behind tip offset) | | InMemory | In-process queue length | 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: ```rust,no_run 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: ```rust,no_run 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` 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](/backends/sqs/examples/autoscaler). ### Shutdown semantics `ConsumerGroup` exposes `run_until_timeout` for clean shutdown: ```rust,no_run 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](/guides/shutdown) 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\`. Refer to the [SQS overview](/backends/sqs) 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(...)`: ```rust,no_run let mut group = broker .consumer_group() .with_default_handler_timeout(Duration::from_secs(60)); group .register::( 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::::with_handler_timeout` instead. ### Full walkthrough \[!include \~/examples/inmemory/consumer\_groups.rs] The key lines are 63–70: `group.register::(...)` 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\ Pattern](/concepts/broker) * [Shutdown & Exit Codes](/guides/shutdown) * [InMemory consumer groups example](/backends/inmemory/examples/consumer-groups) * [SQS autoscaler example](/backends/sqs/examples/autoscaler) ## Liveness Probes Kubernetes liveness and readiness probes need a single, fast, bounded answer to "can this process still talk to the broker?". `shove` exposes `Broker::ping` for exactly that question: one bounded RPC against the cluster, returning `Ok(())` iff it completes within the deadline. Probe policy — retries, failure thresholds, reconnect — belongs to the caller, so `ping` itself does none of those things. ### The probe contract `Broker::ping` uses the default 5-second deadline. Use `ping_with_timeout` to tighten or relax it. ```rust,no_run // Requires the `http` crate (or your web framework's re-export of it). use std::time::Duration; use shove::{Broker, Kafka}; async fn healthz(broker: &Broker) -> http::StatusCode { match broker.ping_with_timeout(Duration::from_secs(2)).await { Ok(()) => http::StatusCode::OK, Err(_) => http::StatusCode::SERVICE_UNAVAILABLE, } } ``` Wire this into your HTTP framework of choice. Retry policy, failure thresholds, and probe intervals live with your runtime (Kubernetes `failureThreshold`, an HTTP middleware, a circuit breaker). `ping` itself does not retry. ### What happens on the wire Every production backend issues a real round-trip. There are no local-state shortcuts on the network path — a half-closed socket cannot lie its way to a green probe. The InMemory backend is the exception: it performs a shutdown-token check only, since it has no underlying transport. | Backend | RPC | Side effects | | -------- | --------------------------------------------------------------------------------- | ----------------------- | | Kafka | `fetch_metadata(None, timeout)` on producer client | none | | Redis | `PING` on the multiplexed connection | none | | NATS | subscribe + unsubscribe-after(1) + flush + publish + await echo on a unique inbox | none | | RabbitMQ | transient channel open + close | none | | SQS | `ListQueues` with `max_results=1` | one AWS API call billed | | InMemory | shutdown-token check (no I/O) | none | The SQS call is the only one with a meaningful operational cost: each probe is one billable `ListQueues` request. Sized k8s probes (one every few seconds) put this well under a dollar a month per pod, but plan accordingly for very large fleets. ### Timeouts `Broker::ping` uses `DEFAULT_PING_TIMEOUT` (5 s). Pass a tighter or looser bound via `ping_with_timeout(Duration)` and align it with your platform's probe configuration. In Kubernetes, the deadline should sit comfortably below `timeoutSeconds` on the probe spec so that a slow but recovering broker can flake one probe without the kubelet timing the handler out on its end. If the underlying RPC exceeds the deadline, `ping` returns `Err(ShoveError::Connection(...))` whose message names the timeout. ### Reconnect policy `ping` does not retry a failed probe. It does, however, allow backends to transparently recover stale internal state during the probe. The RabbitMQ backend dials a fresh AMQP connection if the cached one died, librdkafka maintains its own broker pool, and async-nats heartbeats keep the underlying connection alive. A probe that succeeds after such recovery is reported as `Ok(())`; the broker is reachable now, which is what liveness asks. After `Broker::close`, most backends' `ping` returns `Err(ShoveError::Connection)` immediately via a shutdown-token check. Redis is the exception: its `close` is a no-op because connections drop on last `Arc` release, so `ping` continues to function against a closed broker until the underlying server becomes unreachable. ### What `ping` does not do * **No retries.** A single probe is a single observation. Use k8s `failureThreshold`, HTTP retries, or your circuit breaker to filter transient blips. * **No metrics.** Probe handlers are called frequently; emitting a metric per call would drown out failure signal. * **No per-topic check.** `ping` answers "can I talk to the cluster". Whether a specific topic is declared is a topology concern, handled separately. * **No `PingInfo` payload.** `Result<()>` is the only return. ### What's next * [Shutdown & Exit Codes](/guides/shutdown) — `ping` after `Broker::close` is a permanent failure state. * [Observability](/guides/observability) — surface probe failures through your tracing/metrics pipeline so a flapping probe shows up as signal rather than getting retried into silence. ## Observability When something goes wrong — a handler is slow, a topic is backing up, a retry loop is burning through the budget — you need to know where to look. `shove` integrates with the standard Rust `tracing` ecosystem. Every interesting event — handler invocation outcomes, retry routing, DLQ routing, group scaling, autoscaler decisions, connection errors — is emitted as a structured `tracing` event with named fields. Add a subscriber and you have a full observability trail without any instrumentation code in your handlers. For dashboards and alerting, `shove` also emits operational metrics through the [`metrics`](https://docs.rs/metrics) facade. See [Metrics](#metrics) below. ### Metrics `shove` is a library, not a service, so it does not expose its own scrape endpoint. Instead it emits operational metrics through the [`metrics`](https://docs.rs/metrics) facade crate. The consuming service installs a recorder of its choice — `metrics-exporter-prometheus`, `metrics-exporter-statsd`, OpenTelemetry, etc. — and exposes the endpoint itself. #### Enabling Add the `metrics` feature in your `Cargo.toml`: ```toml [dependencies] shove = { version = "0.11", features = ["rabbitmq", "metrics"] } metrics = "0.24" metrics-exporter-prometheus = "0.16" ``` #### Recorder setup A minimal Prometheus exporter that exposes `/metrics` on `:9100`: ```rust,no_run use metrics_exporter_prometheus::PrometheusBuilder; use std::net::Ipv4Addr; #[tokio::main] async fn main() -> Result<(), Box> { PrometheusBuilder::new() .with_http_listener((Ipv4Addr::UNSPECIFIED, 9100)) // Histogram bucket recommendations for shove's two duration histograms. .set_buckets_for_metric( metrics_exporter_prometheus::Matcher::Suffix( "_duration_seconds".to_string(), ), &[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0], )? .install()?; // ... your shove broker / publisher / supervisor setup ... Ok(()) } ``` #### Custom prefix By default every metric name starts with `shove_`. To override (for example to namespace under your service name) call `shove::metrics::set_prefix` once before installing the recorder: ```rust,no_run shove::metrics::set_prefix("billing_shove"); // ... then install your recorder ... ``` **Important**: `set_prefix` must be called *before any metric emission*, not just before installing the recorder. The metric-name cache is materialised on first use, so any backend or publisher activity locks in the default `shove` prefix; calling `set_prefix` after that point panics rather than silently produce mis-named metrics. The prefix string itself must match Prometheus' name grammar (`[a-zA-Z_][a-zA-Z0-9_]*`); hyphens or other special characters produce invalid metric names that the exporter will reject. #### Metric reference | Name | Type | Labels | | ------------------------------------------- | --------- | ------------------------------------ | | `shove_messages_consumed_total` | counter | `topic`, `consumer_group`, `outcome` | | `shove_messages_failed_total` | counter | `topic`, `consumer_group`, `reason` | | `shove_messages_published_total` | counter | `topic`, `outcome` | | `shove_message_processing_duration_seconds` | histogram | `topic`, `consumer_group`, `outcome` | | `shove_message_publish_duration_seconds` | histogram | `topic`, `outcome` | | `shove_message_size_bytes` | histogram | `topic`, `consumer_group` | | `shove_messages_inflight` | gauge | `topic`, `consumer_group` | | `shove_autoscaler_decisions_total` | counter | `consumer_group`, `direction` | | `shove_autoscaler_messages_ready` | gauge | `consumer_group` | | `shove_autoscaler_messages_in_flight` | gauge | `consumer_group` | | `shove_autoscaler_active_consumers` | gauge | `consumer_group` | | `shove_backend_errors_total` | counter | `backend`, `kind` | **Label values:** * `outcome` on `messages_consumed_total` and `message_processing_duration_seconds`: `ack`, `retry`, `reject`, `defer`. These mirror the `Outcome` enum returned by handlers. * `outcome` on `messages_published_total` and `message_publish_duration_seconds`: `success`, `error`. * `reason` on `messages_failed_total`: `oversize`, `deserialize`, `pending_full`, `timeout`, `max_retries_exceeded`, `rejected`. The first four cover failures that happen *before* the handler runs; `max_retries_exceeded` and `rejected` are emitted after the handler runs when the message is dead-lettered. * `direction` on `autoscaler_decisions_total`: `up`, `down`, `hold`. * `shove_autoscaler_messages_ready`, `shove_autoscaler_messages_in_flight`, and `shove_autoscaler_active_consumers` are emitted once per group on every autoscaler poll, carrying only the `consumer_group` label. They expose the raw signal the scaling decision is made from: backlog waiting, messages in flight, and the live consumer count. Watching `messages_ready` stay high while `active_consumers` is flat at `max` is the canonical "saturated / scaling commands not keeping up" signal. * `backend` on `backend_errors_total`: `inmemory`, `rabbitmq`, `kafka`, `nats`, `sns_sqs`, `redis`. * `kind` on `backend_errors_total`: `connection`, `publish`, `consume`, `topology`, `ack`. * `topic`: the queue name from the topology (`QueueTopology::queue()`). * `consumer_group`: the group name set by a coordinated group registry, or `"default"` when the consumer was registered through `ConsumerSupervisor` directly. `shove_backend_errors_total` covers the most operationally-meaningful error sites (connection drops, topology conflicts, broker NACKs, consume-stream closures, ack failures); not every possible internal `Err` propagation produces a counter increment. Treat the counter as a "something is wrong with backend X" signal rather than an exhaustive error count. `shove_messages_published_total` increments once per message in a batch publish, matching what consumers see downstream. `shove_message_publish_duration_seconds` records one sample per batch (the user-observable call latency). Empty batches are no-ops and emit no events. #### Histogram buckets `shove` does not configure histogram buckets. Set them at the recorder, as shown above. Reasonable starting points: * `shove_message_processing_duration_seconds`: `[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]` * `shove_message_publish_duration_seconds`: same as processing — most publishes are sub-100 ms but tail latency under broker stress is the interesting region. * `shove_message_size_bytes`: `[256, 1024, 4096, 16384, 65536, 262144, 1048576, 10485760]` — covers the path from tiny envelopes up to the default 10 MiB ceiling. ### tracing integration The recommended subscriber setup for a consumer binary: ```rust,no_run use tracing_subscriber::{EnvFilter, fmt}; tracing_subscriber::registry() .with(EnvFilter::from_default_env()) .with(fmt::layer()) .init(); ``` Run with `RUST_LOG=info,shove=debug` to get per-message delivery traces. For production, use `RUST_LOG=info,shove=info` to get group scaling events and errors without per-message noise. Available log levels used by `shove`: | Level | Used for | | ------- | -------------------------------------------------------------------------------------------------------------- | | `error` | Consumer task panics, unrecoverable broker errors, failed acks/nacks | | `warn` | Handler timeouts, deserialization failures, oversized messages, DLQ routing, missing DLQ, deprecated fallbacks | | `info` | Consumer group start/stop, scale up/down, autoscaler start/shutdown | | `debug` | Per-message handled (with outcome), individual scale decisions, consumer spawn | ### Key events emitted `shove` does not use structured span names in the `tracing::span!()` form. All observability is through event macros (`debug!`, `info!`, `warn!`, `error!`) with named fields. The following are the most useful events to watch for, with their actual field names from the source: **Consumer events:** * `"handler task panicked — retrying message"` (`warn`) — a handler future panicked. Fields: `error`, `ticket`. The message is retried. * `"handler timed out — retrying"` (`warn`) — handler exceeded `handler_timeout`. Fields: `timeout`. The message is retried. * `"rejecting oversized message"` (`warn`) — payload exceeds `max_message_size`. Fields: `error`. Routed to DLQ. * `"failed to deserialize message — rejecting"` (`warn`) — JSON deserialization failed. Fields: `error`. Routed to DLQ. * `"message handled (concurrent)"` (`debug`) — a concurrent consumer processed a message. Fields: `queue`, `outcome`. * `"message handled (concurrent-sequenced)"` (`debug`) — a concurrent-sequenced consumer processed a message. Fields: `queue`, `sequence_key`, `outcome`. * `"DLQ declared but not found in broker"` (`error`) — topic has a DLQ in the topology but the broker queue was not found. Fields: `queue`. Indicates a topology declaration failure. **Consumer group / scale events:** * `"starting consumer group"` (`info`) — group starts its initial consumers. Fields: `group`, `queue`, `initial_consumers`. * `"scaled up: spawned new consumer"` (`info`) — autoscaler or manual call added a consumer. Fields: `group`, `consumers`. * `"scaled down: cancelled an idle consumer"` (`info`) — autoscaler or manual call removed a consumer. Fields: `group`, `consumers`. * `"scale_up rejected: at max capacity"` (`debug`) — scale-up attempted but already at `max_consumers`. Fields: `group`, `max`. * `"scale_down rejected: at min capacity"` (`debug`) — scale-down attempted but already at `min_consumers`. Fields: `group`, `min`. * `"scale_down rejected: all consumers are busy"` (`warn`) — tried to scale down but every consumer is processing a message. Fields: `group`. * `"shutting down consumer group"` (`info`) — drain started. Fields: `group`, `consumers`. **Autoscaler events:** * `"autoscaler started"` (`info`) — autoscaler loop entered. * `"autoscaler shutting down"` (`info`) — shutdown token was cancelled. * `"failed to list groups: {e}"` (`error`) — autoscaler could not enumerate consumer groups (backend error). * `"failed to fetch metrics for {group}: {e}"` (`error`) — metrics fetch failed for one group; others continue. * `"failed to scale {group}: {e}"` (`error`) — scaling command failed. **Audit events:** * `"audit handler failed, retrying message"` (`error`) — `AuditHandler::audit()` returned `Err`. Fields: `error`, `delivery_id`. The message will be retried. * `"audit handler timed out, returning original outcome"` (`error`) — audit exceeded `audit_timeout`. Fields: `delivery_id`, `timeout_ms`. Original outcome is preserved. **Drain timeout event:** * `"drain timeout elapsed; aborting surviving tasks"` (`warn`) — `run_until_timeout` deadline elapsed. Fields: `timeout_ms`. ### Header propagation `MessageMetadata::headers` is a `HashMap` carrying broker-level headers that survive hold-queue hops. Notable headers: * **`x-trace-id`** — used by `Audited` as the `trace_id` on audit records. Publishers do not set it automatically; if absent on delivery, `Audited` generates a fresh UUID per message. Set it explicitly via `publish_with_headers` (shown below) when you want a trace ID that connects to upstream/downstream systems. Once set, it is preserved across retries — hold-queue hops forward headers. * **`x-shove-retry-count`** — the internal retry counter maintained by `shove`'s consumer routing layer. * Backend-specific identifiers (see the table below). To set `x-trace-id` at publish time: ```rust,no_run use std::collections::HashMap; let mut headers = HashMap::new(); headers.insert("x-trace-id".to_string(), your_trace_id.to_string()); publisher.publish_with_headers::(&msg, headers).await?; ``` The consumer surfaces this via `metadata.headers.get("x-trace-id")` in the handler. The audit wrapper reads it automatically. ### Connecting to OpenTelemetry Use the `tracing-opentelemetry` bridge to route `shove`'s structured events into your OTel pipeline: ```toml [dependencies] tracing-opentelemetry = "0.x" opentelemetry = "0.x" ``` ```rust,no_run use tracing_subscriber::layer::SubscriberExt; use tracing_opentelemetry::OpenTelemetryLayer; let tracer = init_otel_tracer(); // your OpenTelemetry tracer setup let otel_layer = OpenTelemetryLayer::new(tracer); tracing_subscriber::registry() .with(EnvFilter::from_default_env()) .with(fmt::layer()) // keep local logging too .with(otel_layer) // forward to OTel .init(); ``` All named fields on `shove` events (`group`, `queue`, `outcome`, `error`, `timeout_ms`, etc.) become OTel span attributes automatically via the bridge. Scale events, panics, and audit failures all become searchable attributes in your OTel backend (Jaeger, Tempo, Honeycomb, etc.). ### Per-backend identifier headers Each backend stamps deliveries with a stable per-message identifier. Access these through `MessageMetadata::headers`: | Backend | Header / identifier | Use | | -------------- | ----------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------- | | RabbitMQ | `x-message-id` header | Stable UUID per logical message, preserved through hold-queue hops. Useful for deduplication. External messages get it stamped on first retry. | | NATS JetStream | `Nats-Msg-Id` header | JetStream dedup window (120s default). Use for publisher-side idempotency. | | Apache Kafka | Partition + offset (in metadata) | Stable position in the partition log. Useful for replay and audit correlation. | | SQS | SQS Message ID (in `delivery_id`) | AWS-assigned per-message ID. Stable for the lifetime of the message. | | Redis Streams | Stream entry ID (in `delivery_id`) | Redis-assigned `-` ID per `XADD`. Stable for the lifetime of the entry in the stream. | | InMemory | Internal counter (in `delivery_id`) | Process-local, monotonically increasing. Not stable across restarts. | For RabbitMQ specifically: `x-message-id` is stamped by `RabbitMqPublisher` on every outgoing message. Handlers can read it for deduplication: ```rust,no_run if let Some(mid) = metadata.headers.get("x-message-id") { if store.already_processed(mid).await? { return Outcome::Ack; } store.mark_processed(mid).await?; } // ... business logic ... ``` ### Audit as observability Audit records are a high-resolution observability stream complementary to `tracing` events. Every delivery produces one record with the full payload, outcome, duration in milliseconds, and trace ID. Where `tracing` events are best for operational monitoring (scale events, error rates), audit records are best for business-level visibility: which messages were processed, what the outcome was, how long each one took. Wire `ShoveAuditHandler` (see [Audit Logging](/guides/audit)) and consume the `shove-audit-log` topic to build: * Per-topic outcome histograms (ratio of Ack / Retry / Reject). * Handler latency percentiles (p50 / p99 from `duration_ms`). * Per-entity message history (all records for a given `trace_id` or entity ID in the payload). * Alerting on sustained rejection spikes. ### What's next * [Audit Logging](/guides/audit) * [Handlers & Context](/concepts/handlers) * [Shutdown & Exit Codes](/guides/shutdown) ## Retries, Hold Queues & DLQs Transient failures recover. Permanent failures don't. `shove` gives you the vocabulary to express the difference: **hold queues** park a message for a configurable delay before redelivering it, giving downstream services time to recover; a **DLQ** catches messages that have exhausted all retries or are permanently invalid, so they don't disappear silently but instead wait for human inspection and replay. Together they turn a binary ack/reject model into a resilient, observable pipeline. ### The retry lifecycle Every message delivery follows the same path. The handler receives the message and returns an `Outcome`. The consumer then routes the message based on that outcome and the current retry count: ```text ┌─────────────────┐ │ Main queue │ └────────┬────────┘ │ deliver ▼ ┌─────────────────┐ Ack ────────────► removed from queue │ handler.handle │ └────────┬────────┘ Reject ──────────► DLQ (or discard if no DLQ) │ Outcome │ Retry ──────────► hold_queues[min(retry_count, len-1)] │ │ │ │ after delay │ ▼ │ main queue (retry_count+1) │ └──────────────► Defer ──────────► hold_queues[0] (no counter increment) │ │ after delay ▼ main queue (retry_count unchanged) ``` When `retry_count >= max_retries`, the consumer routes to the DLQ instead of a hold queue. If no DLQ is configured and the message hits its retry limit, it is discarded with a warning log. ### Configuring hold queues Add hold queues to a topology by calling `.hold_queue(Duration)` on the builder. Each call appends one hold queue with the given delay. Order matters: the first call defines the shortest delay, the last call defines the longest. ```rust,no_run TopologyBuilder::new("orders") .hold_queue(Duration::from_secs(5)) // hold_queues[0] → orders-hold-5s .hold_queue(Duration::from_secs(30)) // hold_queues[1] → orders-hold-30s .hold_queue(Duration::from_secs(120)) // hold_queues[2] → orders-hold-120s .dlq() .build() ``` The hold queue selected on each retry attempt is `hold_queues[min(retry_count, len - 1)]`. This gives automatic escalating backoff: | Retry count | Hold queue selected | Delay | | ----------- | ------------------- | ---------------------- | | 0 | hold\_queues\[0] | 5s | | 1 | hold\_queues\[1] | 30s | | 2+ | hold\_queues\[2] | 120s (clamped to last) | Hold-queue names are derived from the queue name and delay: `{queue}-hold-{N}s`. For the example above, the queues are `orders-hold-5s`, `orders-hold-30s`, and `orders-hold-120s`. If no hold queues are configured but the handler returns `Outcome::Retry`, the consumer falls back to broker-level nack-with-requeue (immediate redelivery, no delay). This is usually not what you want — configure at least one hold queue for any topic where transient failures are expected. If a topic has hold queues but no DLQ, a warning is logged at build time: messages that exhaust `max_retries` will be silently discarded. If a topic has a DLQ but no hold queues, another warning is logged: retries will use broker redelivery with no delay. ### max\_retries `ConsumerOptions::with_max_retries(N)` sets how many times a message can be retried before being routed to the DLQ. The default is **10**. ```rust,no_run let opts = ConsumerOptions::::new() .with_max_retries(5); // DLQ after 5 retries ``` When `retry_count >= max_retries`, the consumer routes to the DLQ regardless of the handler's `Outcome::Retry` return. The handler is not invoked again. If the topic has no DLQ and `max_retries` is exhausted, the message is discarded (nack without requeue) and a warning is logged. `shove` does not silently drop messages — you will see the warning. ### max\_reconnect\_attempts `ConsumerOptions::with_max_reconnect_attempts(N)` caps how many times the consumer will attempt to re-establish a broker connection after a connection-level failure. The default is **unlimited** (`None`) — the consumer backs off and retries forever until it succeeds or is shut down. ```rust,no_run let opts = ConsumerOptions::::new() .with_max_reconnect_attempts(50); // give up after 50 failed reconnects ``` Each failed connection attempt increments the counter. Once the limit is reached the consumer emits a `tracing::error!` and propagates the error to the caller — typically the supervisor, which may then restart the consumer or surface the failure to your alerting pipeline. Use this when a permanently-unreachable broker should fail fast and visibly rather than silently spinning in a backoff loop. For transient outages (broker restarts, network blips) the default unlimited behaviour is usually correct. > **Note:** `max_reconnect_attempts` counts connection failures, not message-handling failures. It does not interact with `max_retries`, which counts handler-level retries for individual messages. ### The retry counter is best-effort The retry counter is stored in a broker-level header (`x-shove-retry-count` or the backend equivalent). The publish-to-hold-queue and ack-of-original-delivery are two separate broker operations. There is a small window between them where a crash can produce a duplicate: 1. Handler returns `Outcome::Retry`. 2. Consumer publishes the message to the hold queue (counter incremented in the header). 3. Consumer sends ack for the original delivery. 4. **If the consumer crashes between steps 2 and 3:** the broker redelivers the original message with the OLD retry count. The hold-queue copy is also in flight. Both will eventually be delivered to the handler. For idempotent handlers — those that produce the same result regardless of how many times they are invoked for the same logical message — this is harmless. For handlers with irreversible side effects, the RabbitMQ `rabbitmq-transactional` feature makes steps 2 and 3 atomic, eliminating this race. See [Exactly-Once (RabbitMQ)](/guides/exactly-once) for details. ### Defer vs Retry `Outcome::Defer` is for messages that are **not yet ready** — the message itself is valid, but the time or conditions to process it haven't arrived. Unlike `Retry`, `Defer`: * Does NOT increment the retry counter. * Always routes to `hold_queues[0]` (the shortest delay), regardless of retry count. There is no escalating backoff for deferred messages. * Will never exhaust `max_retries` and will never be routed to the DLQ as a result of deferral alone. Use cases for `Defer`: * **Scheduled delivery** — the message carries a "not-before" timestamp that hasn't arrived yet. * **Missing dependency** — a prerequisite event or resource isn't available yet (e.g. waiting for a payment to clear before fulfilling an order). * **External rate-limiting** — an upstream API returned 429 / "retry after". The message is valid; you just can't act on it right now. Use `Retry` (not `Defer`) for genuine transient failures — network errors, database timeouts, temporary unavailability. `Retry` increments the counter and escalates through the hold queues. `Defer` is specifically for "the message is fine but not yet actionable." **Caution:** because `Defer` never increments the retry counter, a handler that always returns `Defer` will bounce the message between the main queue and `hold_queues[0]` indefinitely. There is no built-in circuit breaker. Ensure your handler has a condition (e.g. a deadline or a maximum deferral count tracked in the message body) that eventually resolves to `Ack`, `Retry`, or `Reject`. If no hold queues are configured, `Defer` falls back to nack-with-requeue (broker-level immediate redelivery) and a warning is logged. On **sequenced consumers**, `Defer` is not supported because it would block all subsequent messages for the same key indefinitely. If a handler returns `Defer` on a sequenced consumer, it is treated as `Retry` and a warning is logged. ### DLQ as a queue Dead-letter queues exist so that permanently-failed messages don't disappear. You can consume the DLQ to alert on failures, write to a quarantine store, trigger a pager, or replay messages after a fix. Override `handle_dead` in your handler to act on dead-lettered messages: ```rust,no_run impl MessageHandler for OrderHandler { type Context = AppState; async fn handle(&self, msg: Order, meta: MessageMetadata, ctx: &AppState) -> Outcome { // ... main processing ... Outcome::Ack } // Override to do something useful with dead messages. async fn handle_dead(&self, msg: Order, meta: DeadMessageMetadata, ctx: &AppState) { tracing::error!( order_id = %msg.order_id, reason = meta.reason.as_deref().unwrap_or("unknown"), deaths = meta.death_count, "Order permanently failed — alerting on-call", ); ctx.pagerduty.alert(&msg).await; } } ``` The default `handle_dead` implementation logs a warning with the delivery ID, reason, original queue, and death count. Override it for alerting, manual replay queues, quarantine workflows, or metrics. `DeadMessageMetadata` carries: * `message` — the base `MessageMetadata` (retry count, delivery ID, headers). * `reason` — why the message was dead-lettered (e.g. `"rejected"`, `"expired"`). * `original_queue` — which queue the message came from. * `death_count` — how many times this message has been dead-lettered. The message is always acked from the DLQ after `handle_dead` returns, regardless of what the handler does. DLQ messages are not retried. ### No DLQ behavior If a topic has no DLQ configured: * A handler returning `Outcome::Reject` causes the message to be discarded (nack without requeue) and a warning is logged. * A message that exhausts `max_retries` is also discarded with a warning log. `shove` does NOT silently drop messages. Every discard produces a log line. Check your logs if messages seem to be disappearing. This is an explicit design choice: if you haven't configured a DLQ, you've accepted that permanently-failing messages will be lost. Make that decision consciously. For production topics with any business significance, configure a DLQ. ### Worked example Consider a topic with three hold queues and a DLQ, and `max_retries = 10` (the default): ```rust,no_run TopologyBuilder::new("orders") .hold_queue(Duration::from_secs(5)) .hold_queue(Duration::from_secs(30)) .hold_queue(Duration::from_secs(120)) .dlq() .build() ``` **Scenario 1 — transient failure that recovers on attempt 4:** | Attempt | retry\_count | Hold queue used | Outcome | | ------- | ------------ | ----------------------------- | ------------------------ | | 1 | 0 | — | Retry → orders-hold-5s | | 2 | 1 | orders-hold-5s (5s delay) | Retry → orders-hold-30s | | 3 | 2 | orders-hold-30s (30s delay) | Retry → orders-hold-120s | | 4 | 3 | orders-hold-120s (120s delay) | Ack ✓ | The message succeeds on the fourth attempt. Total hold time: 5s + 30s + 120s = 155s. **Scenario 2 — message that always fails (11 attempts, then DLQ):** Attempts 1–10 follow the escalating pattern: 5s → 30s → 120s → 120s → 120s → ... (clamped to `orders-hold-120s` from attempt 3 onward). On attempt 11, `retry_count = 10 >= max_retries = 10`, so the message is routed to `orders-dlq` instead of another hold queue. `handle_dead` is invoked on the DLQ consumer. **Scenario 3 — Defer mid-stream:** A handler checks a "not-before" timestamp and returns `Outcome::Defer` on the first two deliveries because the scheduled time hasn't arrived yet. On the third delivery the timestamp has passed and it returns `Outcome::Ack`. | Attempt | retry\_count | Hold queue used | Outcome | | ------- | ------------ | --------------- | ------------------------------------------ | | 1 | 0 | — | Defer → orders-hold-5s (counter unchanged) | | 2 | 0 | orders-hold-5s | Defer → orders-hold-5s (counter unchanged) | | 3 | 0 | orders-hold-5s | Ack ✓ | The retry counter stayed at 0 throughout. The `max_retries` budget was never consumed. **Scenario 4 — deserialization failure (bad bytes):** A message arrives with a malformed payload that cannot be deserialized into the expected type. The consumer rejects it **before invoking the handler** — straight to the DLQ (or discards it with a warning if no DLQ is configured). `max_retries` is irrelevant here: the handler never runs, so the retry counter is never consulted. This prevents poison-pill messages from burning through the retry budget indefinitely. A message that is too large (exceeds `ConsumerOptions::max_message_size`, default 10 MiB) is handled the same way. ### What's next * [Outcomes & Delivery](/concepts/outcomes) * [Topics & Topology](/concepts/topics) * [Exactly-Once (RabbitMQ)](/guides/exactly-once) * [Sequenced Topics](/guides/sequenced) ## Sequenced Topics Some messages depend on each other. A ledger entry assumes the previous balance was applied. A state-machine transition assumes the entity is in the right prior state. An audit record references the preceding record's hash. When processing message N assumes message N−1 has already been applied, you need `shove`'s sequenced delivery: messages with the same key arrive one at a time, in strict publish order. Messages sharing the same sequence key are consumed one at a time: the next delivery only happens after the previous one has been successfully acked. ### When to use sequenced topics Use a sequenced topic when messages for the same logical entity have **causal dependencies**: * **Financial ledgers** — an account balance is updated by a stream of debit/credit entries. Applying them out of order corrupts the balance. * **State machines** — an order moving through `placed → confirmed → shipped → delivered`. A `shipped` event arriving before `confirmed` is processed leaves the state machine in an inconsistent state. * **Per-entity audit chains** — each audit entry references the previous state. Reordering entries breaks the chain. * **Event sourcing** — replaying an entity's event log in the wrong order produces the wrong aggregate state. Not every topic needs sequencing. Independent events don't benefit from it — and strict per-key ordering costs throughput. Good counter-examples: * **Search-hit events** — each hit is independent. Order between different users' search events is meaningless. * **Page-view analytics** — each view is self-contained. Ordering across millions of concurrent sessions is unnecessary and expensive. * **Notification fan-outs** — sending an email to many recipients is embarrassingly parallel. FIFO ordering per recipient adds no value. When in doubt: if you can process message B without knowing the result of message A, ordering probably doesn't matter. ### define\_sequenced\_topic! The `define_sequenced_topic!` macro is the recommended way to define a sequenced topic. It generates a unit struct, implements both `Topic` and `SequencedTopic`, and wires the sequence-key function into `Topic::SEQUENCE_KEY_FN` so the publisher can route messages correctly without requiring a `SequencedTopic` bound. ```rust,no_run use shove::{define_sequenced_topic, SequenceFailure, TopologyBuilder}; use std::time::Duration; define_sequenced_topic!( AccountLedger, LedgerEntry, |msg| msg.account_id.clone(), TopologyBuilder::new("account-ledger") .sequenced(SequenceFailure::FailAll) .routing_shards(16) .hold_queue(Duration::from_secs(5)) .dlq() .build() ); ``` The four macro arguments are: 1. **Name** — the unit struct identifier (`AccountLedger`). 2. **Message type** — the payload type that flows through the topic (`LedgerEntry`). 3. **Key function** — a non-capturing closure or bare function pointer `fn(&Message) -> String` that extracts the sequence key from a message. 4. **Topology** — a `QueueTopology` built with `TopologyBuilder`. What the macro generates — equivalent hand-written code: ```rust,no_run pub struct AccountLedger; impl Topic for AccountLedger { type Message = LedgerEntry; fn topology() -> &'static QueueTopology { static TOPOLOGY: std::sync::OnceLock = std::sync::OnceLock::new(); TOPOLOGY.get_or_init(|| { TopologyBuilder::new("account-ledger") .sequenced(SequenceFailure::FailAll) .routing_shards(16) .hold_queue(Duration::from_secs(5)) .dlq() .build() }) } // Wired automatically — lets publishers route without SequencedTopic bound. const SEQUENCE_KEY_FN: Option String> = Some(Self::sequence_key); } impl SequencedTopic for AccountLedger { fn sequence_key(message: &LedgerEntry) -> String { message.account_id.clone() } } ``` The `OnceLock` ensures the topology expression runs once and the same `&'static QueueTopology` is returned on every call. The macro accepts an optional visibility modifier (`pub`, `pub(crate)`, etc.); omitting it gives module-private visibility. The key function must be non-capturing. A closure that captures environment variables cannot be coerced to `fn(&Message) -> String` and will produce a compile error. If you need access to external state for key extraction, encode it into the message type instead. ### Sequence keys A sequence key is a `String` that identifies a logical entity. All messages with the same key form a strictly ordered stream. Messages with different keys are completely independent and can be processed concurrently across different shards. Guidelines for choosing a key: * **One key per logical entity.** Use `account_id`, `order_id`, `user_id`, `entity_id`. The key should identify the thing whose state you're trying to keep consistent. * **High cardinality is fine.** More distinct keys = more concurrency. A ledger with a million accounts and 16 shards distributes across all 16 simultaneously. * **Low cardinality causes hot spots.** If all messages share the same key (e.g. you accidentally use a constant), every message serializes through one shard, defeating the purpose of multiple consumers. * **Keys are hashed to shards.** Different keys can land on the same shard — that's expected. The consumer enforces per-key ordering within each shard, not global per-shard ordering. Two different keys on the same shard are still processed independently as long as they don't depend on each other. Avoid composite keys unless you genuinely need joint ordering. `account_id:transaction_type` serializes all transaction types for an account together, which may be unnecessary if only debits and credits need to be ordered relative to each other. ### Failure policies — Skip vs FailAll When a message is permanently rejected (the handler returns `Outcome::Reject`, or the retry budget is exhausted), the remaining messages in the same sequence are in a difficult position. The system cannot know whether they are still valid without the context of the failed message. `SequenceFailure` controls what happens next. Given a sequence for key `ACC-A` with messages `[1, 2, 3, 4, 5]` where message 3 is permanently rejected: | Policy | Acked | DLQed | | --------- | ---------- | ------- | | `Skip` | 1, 2, 4, 5 | 3 | | `FailAll` | 1, 2 | 3, 4, 5 | **`SequenceFailure::Skip`** — dead-letters the failed message and continues processing subsequent messages in the sequence as normal. Use `Skip` when messages are **independently valid** but happen to require ordered delivery. Audit log entries are a good example: if entry 3 is malformed and cannot be processed, entries 4 and 5 are still valid records that should be stored. The audit trail has a gap, but it continues. `Skip` is the right choice when gaps in the sequence are preferable to stopping cold. **`SequenceFailure::FailAll`** — dead-letters the failed message AND all subsequent messages for the same key ("poisons" the key). The key stays poisoned for the lifetime of the consumer process. Use `FailAll` when messages are **causally dependent** — each message assumes every prior message in the sequence was applied successfully. Financial ledger entries are the canonical example: if a debit entry fails, the account balance is unknown. Applying subsequent credits or debits against an unknown balance produces incorrect results. `FailAll` is the right choice when processing on a corrupted baseline would cause more damage than stopping and surfacing the problem for manual review. Both policies only affect the failing sequence key. Other keys (e.g. `ACC-B`) are completely unaffected — they continue to be processed normally regardless of what happens to `ACC-A`. ### Per-backend mapping The sequenced topic abstraction maps to different broker primitives depending on the backend in use. The consumer-visible contract is identical across all backends — per-key strict ordering with the configured failure policy — but the underlying mechanism differs. | Backend | Primitive | Ordering enforcement | | -------------- | ------------------------------------- | --------------------------------------------------- | | RabbitMQ | Consistent-hash exchange + sub-queues | Single Active Consumer + `prefetch=1` per sub-queue | | AWS SNS+SQS | FIFO topic + `MessageGroupId` | Native SQS FIFO ordering | | NATS JetStream | Subject shard | `max_ack_pending=1` per subject | | Apache Kafka | Partition key | Consumer-group partition assignment | | In-process | Per-key FIFO shards | `tokio::sync::Mutex` per key | For RabbitMQ, the consistent-hash exchange name is derived from the queue name as `{queue}-seq-hash` (visible in `SequenceConfig::exchange()`). Each routing shard gets its own sub-queue. The hold-queue names for sequenced topics use the shard index: `{queue}-seq-{shard_index}-hold-{N}s`. For AWS SNS+SQS FIFO topics, `shove` sets the `MessageGroupId` to the sequence key returned by `SequencedTopic::sequence_key`. The broker enforces ordering natively. ### Routing shards Call `.routing_shards(N)` on the `TopologyBuilder` (before `.build()`, after `.sequenced()`) to override the default shard count of **8**. It panics if called before `.sequenced()`. ```rust,no_run TopologyBuilder::new("account-ledger") .sequenced(SequenceFailure::FailAll) .routing_shards(16) // 16 concurrent shards instead of 8 .hold_queue(Duration::from_secs(5)) .dlq() .build() ``` Trade-offs: * **More shards** — more cross-key concurrency, because more keys can be processed simultaneously. Costs more broker resources (more queues, more consumers, more connections). * **Fewer shards** — lower broker overhead, but more keys share a shard, reducing concurrency. * **Default (8)** — suitable for most workloads. Change only after profiling. For Kafka, the "routing shard" concept maps to partition count. Partition count is set at topic creation time and is operationally expensive to change later (it requires partition reassignment). Think carefully before choosing a low number for a Kafka-backed sequenced topic in production. A routing shard count of 0 panics at `build()` time. The builder enforces this guard. ### Consuming sequenced topics Sequenced topics participate in the same harness as regular topics via `register_fifo` — but the plain `register` method still rejects them at runtime so per-key ordering can't be silently dropped: ```rust,no_run // This will return Err for sequenced topics: supervisor.register::(handler, options)?; // ^ error: use register_fifo instead ``` For the in-memory backend, use `InMemoryConsumer::run_fifo` for the consumer-direct path: ```rust,no_run consumer.run_fifo::(handler, ctx, opts).await ``` The type signature enforces the `SequencedTopic` bound — the call will not compile if `LedgerTopic` does not implement `SequencedTopic`. Other backends have equivalent per-backend sequenced consumer entry points: `RabbitMqConsumer::run_fifo`, `NatsConsumer::run_fifo`, `KafkaConsumer::run_fifo`, and `SqsConsumer::run_fifo`. #### Harness path: `register_fifo` When you're already running other consumers through `ConsumerSupervisor` or `ConsumerGroup`, register sequenced topics on the same harness via `register_fifo`. They drain through the same `run_until_timeout` as regular registrations, returning a single `SupervisorOutcome` for the entire harness: ```rust,no_run let mut sup = broker.consumer_supervisor().with_context(ctx); sup.register::(handler_a, options_a)?; sup.register_fifo::(handler_b, options_b).await?; let outcome = sup .run_until_timeout(shutdown.cancelled_owned(), Duration::from_secs(30)) .await; ``` For consumer groups, `register_fifo` accepts a `ConsumerGroupConfig` — the same type as `register`. On most backends the consumer count is pinned to 1 internally (ordering requires a single consumer per shard); on Redis, `consumer_count()` controls how many consumer tasks are spawned. Autoscaling is not applied to FIFO registrations: ```rust,no_run use shove::consumer_group::ConsumerGroupConfig; let mut group = broker.consumer_group().with_context(ctx); group .register_fifo::( ConsumerGroupConfig::new(BackendConsumerGroupConfig::default()), || MyHandler, ) .await?; let outcome = group.run_until_timeout(signal, Duration::from_secs(30)).await; ``` Pick the harness path when you have a mix of sequenced and unsequenced topics in one binary; pick `run_fifo_until_timeout` (below) when you only have one FIFO consumer and don't want a harness layer. #### Graceful shutdown `run_fifo` blocks until every shard exits — usually because the shutdown token in `ConsumerOptions` was cancelled. To bound the drain and surface error/panic counts to the process, use `run_fifo_until_timeout` instead: ```rust,no_run let outcome = consumer .run_fifo_until_timeout::( handler, ctx, opts, shutdown.cancelled_owned(), Duration::from_secs(30), ) .await; ``` Returns the same `SupervisorOutcome` as `ConsumerGroup::run_until_timeout`, so process-level exit-code rollups across coordinated groups and sequenced topics use one type. See [Shutdown semantics → Sequenced topics](/guides/shutdown#sequenced-topics-run_fifo_until_timeout) for the full contract. To preserve per-key ordering, both `ConsumerGroup::register` and `ConsumerSupervisor::register` reject any topic whose topology has `.sequenced(...)` configured: registration returns `ShoveError::Topology(...)` pointing at `register_fifo`. The plain `register` path spawns workers via the unsequenced `run`, which would silently drop ordering — the runtime check is there as a safety net, not a long-term limitation. `ConsumerOptions::max_pending_per_key` (default: 1,000) limits how many messages can be locally buffered per key in concurrent-sequenced consumers. When exceeded, new deliveries for that key are rejected to the DLQ. Adjust with `.with_max_pending_per_key(N)` or disable with `.without_max_pending_per_key()`. Note also that `Outcome::Defer` is not supported on sequenced consumers. A handler returning `Defer` on a sequenced consumer is treated as `Retry` and a warning is logged, because deferring a message without incrementing the retry counter would block all subsequent messages for that key indefinitely. ### Full walkthrough \[!include \~/examples/inmemory/sequenced.rs] The example hand-implements `Topic` and `SequencedTopic` directly (instead of using `define_sequenced_topic!`) to show what the macro produces. The handler simply acks every message — the interesting part is the `run_fifo` call on line 88, which selects the sequenced code path, and the four routing shards in the topology that allow `alice`, `bob`, and `carol` to be processed concurrently while their individual sequences remain ordered. ### What's next * [InMemory sequenced example](/backends/inmemory/examples/sequenced) * [RabbitMQ sequenced example](/backends/rabbitmq/examples/sequenced) * [Topics & Topology](/concepts/topics) * [Retries, Hold Queues & DLQs](/guides/retries) ## Shutdown & Exit Codes Rolling deploys, Kubernetes preStop hooks, systemd `ExecStop`, CI pipelines — all of them send a signal and expect the process to finish cleanly before they terminate it. `shove` ships a shutdown contract that fits all of these: stop accepting new work, finish in-flight handlers, and surface the result as a process exit code that operators can read and act on. `SupervisorOutcome` is a `#[must_use]` struct returned by `run_until_timeout` on both `ConsumerGroup\` and `ConsumerSupervisor\`, and its `.exit_code()` method maps the drain result to a conventional integer. ### The shutdown contract The canonical pattern for a `shove` consumer binary: ```rust,no_run use std::time::Duration; use futures::FutureExt; let outcome = group .run_until_timeout( tokio::signal::ctrl_c().map(|_| ()), Duration::from_secs(30), ) .await; std::process::exit(outcome.exit_code()); ``` What happens step by step: 1. `run_until_timeout` starts the group's consumer tasks and blocks until either `signal_future` resolves or the group's internal `CancellationToken` is cancelled. 2. When `signal_future` resolves, the shutdown token is cancelled. Consumers stop accepting new deliveries from the broker. 3. Already-in-flight handlers are given up to `drain_timeout` to complete naturally. 4. Any handler still running at the deadline is abandoned — the tokio task is aborted (not gracefully cancelled). The broker will eventually redeliver those messages. 5. `run_until_timeout` returns a `SupervisorOutcome` describing what happened. 6. `std::process::exit(outcome.exit_code())` sets the process exit code and terminates. The `CancellationToken` from `group.cancellation_token()` (or `supervisor.cancellation_token()`) can also be used to trigger shutdown from other parts of your application — e.g. when an unrecoverable error occurs in a non-consumer task: ```rust,no_run let token = group.cancellation_token(); // ... elsewhere in your app ... token.cancel(); // triggers graceful drain ``` ### SupervisorOutcome variants `SupervisorOutcome` is a simple struct (not an enum) with three fields: ```rust,no_run pub struct SupervisorOutcome { pub errors: usize, // count of consumer tasks that returned Err pub panics: usize, // count of consumer tasks that panicked pub timed_out: bool, // true if drain_timeout elapsed before all handlers finished } ``` And two methods: ```rust,no_run impl SupervisorOutcome { pub fn exit_code(&self) -> i32 { ... } pub fn is_clean(&self) -> bool { ... } } ``` * **Clean** (`errors == 0 && panics == 0 && !timed_out`) — all in-flight handlers completed and acked within the drain window. `exit_code()` returns `0`. `is_clean()` returns `true`. * **Errors** (`errors > 0`) — at least one consumer task returned a `ShoveError` from its run loop (distinct from `Outcome::Reject`, which is normal business logic). `exit_code()` returns `1`. * **Panics** (`panics > 0`) — at least one consumer task or handler panicked. Panics outrank errors. `exit_code()` returns `2` even if there are also errors. * **Drain timeout** (`timed_out == true`) — the drain deadline elapsed before all handlers finished. Drain timeout outranks everything. `exit_code()` returns `3` even if there are also errors or panics. The priority ordering for `exit_code()` is: `timed_out` (3) > `panics` (2) > `errors` (1) > clean (0). ### exit\_code() mapping | Field state | exit\_code() | Meaning | | ------------------- | ------------ | --------------------------------------------------------------------- | | All zero / false | 0 | Clean drain — all in-flight messages were acked, retried, or rejected | | `errors > 0` | 1 | At least one consumer task returned a framework-level error | | `panics > 0` | 2 | At least one task panicked (errors may also be present) | | `timed_out == true` | 3 | Drain deadline elapsed; in-flight messages were abandoned | Exit code `1` is a consumer task error, not a handler-returned `Outcome::Reject`. An `Outcome::Reject` that routes a message to the DLQ is normal operation — it does not increment `errors`. Only unrecoverable infrastructure errors (failed broker connection during drain, serialization bugs that `shove` surfaces as `Err`) increment the error count. ### Picking a drain timeout Set the drain timeout to **longer than your longest-expected handler**. Guidelines: * If your handlers complete in under 1 second, 10 seconds is plenty. * If handlers make external API calls with up to 10-second timeouts, use 30 seconds. * If handlers do complex database transactions that can take up to 30 seconds in the worst case, use 60–90 seconds. Too short a drain timeout results in `timed_out = true` and abandoned in-flight messages. Those messages are not lost — the broker will redeliver them — but they cause an extra delivery and your handlers must be idempotent to handle it correctly. Too long a drain timeout slows rolling deploys. In Kubernetes, a pod's `terminationGracePeriodSeconds` must be longer than your drain timeout, or the kubelet will SIGKILL the process before the drain finishes. A common setup: ```yaml terminationGracePeriodSeconds: 90 # longer than drain_timeout ``` ```rust,no_run group.run_until_timeout(signal, Duration::from_secs(60)).await ``` ### Signal sources **Local development:** ```rust,no_run use futures::FutureExt; group.run_until_timeout( tokio::signal::ctrl_c().map(|_| ()), Duration::from_secs(10), ).await ``` **Kubernetes preStop / SIGTERM** — the kubelet sends SIGTERM before SIGKILL. Wire it with `tokio::signal::unix`: ```rust,no_run use tokio::signal::unix::{signal, SignalKind}; let mut sigterm = signal(SignalKind::terminate()).expect("SIGTERM handler"); group.run_until_timeout( async move { sigterm.recv().await; }, Duration::from_secs(60), ).await ``` **Multiple signals (Ctrl-C + SIGTERM)** — use `tokio::select!` to race them: ```rust,no_run use tokio::signal::unix::{signal, SignalKind}; use futures::FutureExt; let mut sigterm = signal(SignalKind::terminate()).expect("SIGTERM"); let signal_future = async move { tokio::select! { _ = tokio::signal::ctrl_c() => {} _ = sigterm.recv() => {} } }; group.run_until_timeout(signal_future, Duration::from_secs(30)).await ``` **GitHub Actions / systemd** — both send SIGTERM. The SIGTERM handler above covers these cases. **Internal cancellation** — if another part of your application detects an unrecoverable condition and wants to trigger a clean shutdown, use the cancellation token: ```rust,no_run let token = group.cancellation_token(); let shutdown_handle = tokio::spawn(async move { if let Err(e) = health_check().await { tracing::error!("health check failed: {e} — initiating shutdown"); token.cancel(); } }); ``` ### What "clean" actually means A `SupervisorOutcome` with `exit_code() == 0` means every in-flight message at signal time was one of: * **Acked** — handler returned `Outcome::Ack`. Removed from the queue. * **Retried** — handler returned `Outcome::Retry` or `Outcome::Defer`. Published to hold queue and acked from main queue. * **Rejected** — handler returned `Outcome::Reject` or exhausted `max_retries`. Routed to DLQ (or discarded with warning if no DLQ) and acked from main queue. * **Discarded with warning** — message was malformed (deserialization failure, size limit exceeded). Acked from main queue after routing to DLQ or discarding. Critically: `Clean` means NO message was left partially-processed mid-flight. If you see exit code `3` (`DrainTimeout`), the handler was still running when the deadline elapsed and the tokio task was aborted. The broker will redeliver that message. Your handler must be idempotent to handle the redelivery correctly. For handlers with irreversible side effects, consider increasing the drain timeout or reducing handler latency before a rolling deploy. Abandoned in-flight messages under `DrainTimeout` are the most common source of production incidents involving `shove` consumers. ### Sequenced topics: `run_fifo_until_timeout` Sequenced topics don't go through the `ConsumerSupervisor` / `ConsumerGroup` harness — `register` rejects them so per-key ordering can't be silently dropped (see [Sequenced consumers](/guides/sequenced)). Each backend's consumer instead exposes `run_fifo`, which spawns one task per routing shard and runs until the consumer's shutdown token is cancelled. For graceful shutdown with the same `SupervisorOutcome` semantics as the harness, use `run_fifo_until_timeout`: ```rust,no_run use std::time::Duration; use shove::SupervisorOutcome; use shove::rabbitmq::consumer::RabbitMqConsumer; let consumer = RabbitMqConsumer::new(client.clone()); let opts = ConsumerOptions::::new() .with_prefetch_count(1) .with_max_retries(5) .with_shutdown(shutdown.clone()); let outcome: SupervisorOutcome = consumer .run_fifo_until_timeout::( handler, ctx, opts, shutdown.cancelled_owned(), Duration::from_secs(30), ) .await; std::process::exit(outcome.exit_code()); ``` `run_fifo_until_timeout` races `signal` against shards finishing on their own: * If shards finish before `signal`: tally errors and panics, return immediately (no drain timeout consumed). * If `signal` fires first: cancel `options.shutdown`, wait up to `drain_timeout` for shards to finish, then abort surviving shards. The timeout itself sets `timed_out = true`, which `exit_code()` maps to `3`. This is the FIFO peer of `ConsumerGroup::run_until_timeout` and shares the same `SupervisorOutcome` type, so a process-level `.max(...)` rollup across groups + FIFO drains is just: ```rust,no_run let exit_code = group_outcome .exit_code() .max(fifo_outcome.exit_code()); std::process::exit(exit_code); ``` Available on `RabbitMqConsumer`, `KafkaConsumer`, `NatsConsumer`, `InMemoryConsumer`, and `SqsConsumer`. **Handler-panic counting note.** All backends wrap each handler invocation in a `tokio::spawn` + `oneshot` boundary that catches panics and maps them to `Outcome::Retry`. As a result `outcome.panics` does not increment for panics inside `MessageHandler::handle` — those land in your retry/DLQ path instead. `outcome.panics > 0` reflects panics in shard-infrastructure code that escape the in-shard handler-spawn boundary. Aborted-during-drain shards do **not** count as panics: the harness's tally ignores `JoinError::is_cancelled()`, so a clean drain timeout returns `timed_out=true, panics=0, errors=0`. **Harness path.** If your binary already runs other consumers through `ConsumerSupervisor` / `ConsumerGroup`, prefer `register_fifo` over `run_fifo_until_timeout` — sequenced topics drain through the same `run_until_timeout` as regular topics, returning a single `SupervisorOutcome` for the entire harness. See [Sequenced consumers → Harness path](/guides/sequenced#harness-path-register_fifo). ### What's next * [Consumer Groups & Autoscaling](/guides/groups) * [The Broker\ Pattern](/concepts/broker) * [Retries, Hold Queues & DLQs](/guides/retries) ## The Broker\ Pattern The central value of `shove` is portability: write your topic definitions, handler logic, and consumer registrations once — then run the same code against an in-process broker in tests and a real broker in production. Swapping backends means changing one marker type and one config struct. Nothing else changes. `Broker` is the single entry point for every backend. The type parameter `B` is a zero-sized marker struct — `RabbitMq`, `Sqs`, `Nats`, `Kafka`, or `InMemory` — that ties together the backend's connection type, publisher, consumer, and topology implementations. ### The four-corner API Every `Broker` exposes the same four methods: ```text Broker ├─ .topology() → TopologyDeclarer ├─ .publisher().await → Publisher ├─ .consumer_supervisor() → ConsumerSupervisor (all backends) └─ .consumer_group() → ConsumerGroup (all except Sqs) ``` * `topology()` — declare queues, exchanges, and streams idempotently at startup. * `publisher().await` — obtain a typed publisher; each `publish::(&msg)` call is bound to a topic. * `consumer_supervisor()` — register multiple topic consumers in one process and manage their lifecycle together: shared shutdown signal, unified exit code, graceful drain. * `consumer_group()` — register handlers into a coordinated, min/max-bounded consumer group with autoscaling. ### consumer\_supervisor `consumer_supervisor()` is the harness for running multiple topic consumers inside a single application process. You register each topic's handler independently, and the supervisor gives you one place to start them all, one shutdown signal, and one `SupervisorOutcome` at the end. This is the right primitive when your application consumes several topics and you want them to start and stop together — for example, an order-processing service that simultaneously listens to `OrderCreated`, `PaymentConfirmed`, and `ShipmentRequested`. The supervisor handles fan-in: wait for all registered consumers to drain before returning. One constraint: `ConsumerSupervisor::register` explicitly rejects sequenced topics at runtime — sequenced topics require the coordinated group path to preserve ordering guarantees. ### consumer\_group `consumer_group()` is for horizontal scaling of a single topic. It registers a handler with a minimum and maximum consumer count; the autoscaler adjusts the live count within that range based on queue depth. Consumers coordinate through the broker (RabbitMQ Single Active Consumer, Kafka consumer groups, NATS pull consumers, in-process channel registry) so messages are not delivered to two consumers simultaneously. Available on `RabbitMq`, `Nats`, `Kafka`, and `InMemory`. Both primitives share the same shutdown contract: `run_until_timeout(signal_future, drain_timeout)` returning a `SupervisorOutcome`. See [Consumer Groups & Autoscaling](/guides/groups) for the coordination details. ### Backend-swap demo The same topic definition and handler compiles and runs against any backend. Only the marker and the config change: ```rust,no_run // Topic and handler defined once, backend-agnostic define_topic!(Orders, OrderEvent, TopologyBuilder::new("orders").dlq().build()); struct Handler; impl MessageHandler for Handler { type Context = (); async fn handle(&self, msg: OrderEvent, _: MessageMetadata, _: &()) -> Outcome { Outcome::Ack } } // In-process for tests let broker = Broker::::new(InMemoryConfig::default()).await?; // RabbitMQ for production let broker = Broker::::new( RabbitMqConfig::new("amqp://guest:guest@rabbitmq:5672"), ).await?; // Everything else is identical across backends: broker.topology().declare::().await?; let publisher = broker.publisher().await?; publisher.publish::(&event).await?; ``` Tests run against `InMemory` with no external dependencies; production runs against the real broker; the application code does not notice the difference. This makes it practical to test retry logic, sequencing, and DLQ routing in fast unit tests without a running broker. ### SupervisorOutcome and exit codes Every consumer group and supervisor returns a `SupervisorOutcome` after all tasks have drained. The `exit_code()` method maps the outcome to a uniform process exit code that Kubernetes, systemd, and shell pipelines can read: * `0` — clean shutdown; all messages processed and no tasks failed. * `1` — at least one consumer task returned an error. * `2` — at least one consumer task panicked. * `3` — drain timeout elapsed; in-flight messages were abandoned. The canonical shutdown pattern in a long-running binary: ```rust,no_run let outcome = group.run_until_timeout( tokio::signal::ctrl_c().map(drop), Duration::from_secs(30), ).await; std::process::exit(outcome.exit_code()); ``` This integrates cleanly with Kubernetes `preStop` hooks, systemd `ExecStop`, and `&&`-chained shell scripts. See [Shutdown & Exit Codes](/guides/shutdown) for the full shutdown lifecycle. For per-broker setup, see [Integrate with your stack](/backends/choosing). For handler registration and context injection, see [Handlers & Context](/concepts/handlers). ## Codecs A codec controls how a topic's messages cross the wire. Every topic carries a `Codec` associated type that decides how `T::Message` is encoded on `publish` and decoded on `consume`. The default is JSON, which means existing code keeps working without any changes; opting into a different encoding is a one-line edit on the topic definition. The codec slot is per-topic, not per-broker. Two topics on the same backend may speak entirely different wire formats. This is the right boundary: a topic models a single event flow, and the encoding is a property of that flow, not of the transport. ### When to swap the default Three situations come up in practice: * **Protobuf**, when you need a compact, schema-evolution-aware wire format, or when you share message contracts with services written in other languages. * **Raw bytes**, when the payload is already encoded by an upstream component (Confluent Schema Registry's framed Avro, opaque blobs from a legacy system, a third-party format with its own framing). * **A custom codec**, when none of the above fit — for example, CBOR or a domain-specific framing. Outside those situations, stick with JSON; it is the default, requires no extra dependencies, and is the easiest format to debug. ### The Codec trait The trait is short: ```rust,no_run pub trait Codec: Send + Sync + 'static { const NAME: &'static str; fn encode(value: &M) -> Result>; fn decode(bytes: &[u8]) -> Result; } ``` `encode` and `decode` are associated functions, not methods on a value. The codec carries no state; the macros plug the type in directly. `NAME` is a stable label used in logs. ### JsonCodec — the default `JsonCodec` is what you get when `define_topic!` is invoked without a `codec = …` clause: ```rust,no_run define_topic!( OrderSettlement, SettlementEvent, TopologyBuilder::new("order-settlement").dlq().build() ); ``` This compiles down to a topic with `type Codec = JsonCodec;`. Payloads ride the wire as `serde_json::to_vec` output, and `serde_json::from_slice` is called on the consumer side. `SettlementEvent` must implement `Serialize` and `DeserializeOwned`. ### ProtobufCodec Enable the `protobuf` cargo feature, derive `prost::Message` on the payload, and pass the codec to the macro: ```rust // [!include ~/examples/kafka/protobuf_pubsub.rs:topic] ``` `ProtobufCodec` is a single marker that implements `Codec` for every `M: prost::Message + Default`. One topic can use `codec = ProtobufCodec` for `OrderEvent`, another for `UserCreated`, and so on — each call site picks the matching impl through `>`. Encoding pre-sizes the buffer via `encoded_len`, so a single allocation covers the common case. Decoding failures surface as `ShoveError::Codec { codec: "protobuf", .. }` rather than `Serialization`, so handlers and middleware can distinguish protobuf-decode errors from JSON ones. Runnable example: [`examples/kafka/protobuf_pubsub.rs`](https://github.com/zannis/shove/blob/main/examples/kafka/protobuf_pubsub.rs). #### Generating Rust types from a `.proto` file The example above defines `OrderEvent` inline with `#[derive(prost::Message)]`. In production you usually drive the schema from a `.proto` file and let [`prost-build`](https://docs.rs/prost-build) generate the Rust types at compile time. shove sees the generated type the same as a hand-written one — anything implementing `prost::Message + Default` works. Add `prost-build` as a build dependency: ```toml # Cargo.toml [dependencies] shove = { version = "0.11", features = ["kafka", "protobuf"] } prost = "0.13" [build-dependencies] prost-build = "0.13" ``` Write the schema: ```proto // proto/order.proto syntax = "proto3"; package myservice; message OrderEvent { string order_id = 1; double amount = 2; } ``` Compile it at build time: ```rust,no_run // build.rs fn main() -> Result<(), Box> { prost_build::compile_protos(&["proto/order.proto"], &["proto/"])?; Ok(()) } ``` Then include the generated module and bind it to a topic: ```rust,no_run use shove::{define_topic, ProtobufCodec, TopologyBuilder}; // The module name matches the `.proto`'s `package` declaration. mod myservice { include!(concat!(env!("OUT_DIR"), "/myservice.rs")); } define_topic!( Orders, myservice::OrderEvent, TopologyBuilder::new("orders").dlq().build(), codec = ProtobufCodec ); ``` For gRPC services, swap `prost-build` for [`tonic-build`](https://docs.rs/tonic-build) — the generated message types still implement `prost::Message`, so `ProtobufCodec` works on them unchanged. If you already depend on a separate crate that publishes generated protobuf types, depend on it as a normal dependency and skip the `build.rs` step. shove deliberately does not bundle a `.proto` compiler — `prost-build` and `tonic-build` are the standard tools and would only duplicate them. The `protobuf` feature in shove is the runtime codec only. ### RawBytesCodec — the escape hatch `RawBytesCodec` is a passthrough for payloads that are already encoded. The topic carries `Vec` and the handler owns every wire-format decision: ```rust // [!include ~/examples/nats/raw_bytes.rs:topic] ``` The canonical use case is **Confluent Schema Registry**. Records on that wire arrive as `[magic_byte, schema_id (4 bytes), avro_payload...]`. The library doesn't ship a Schema Registry client, but `RawBytesCodec` lets the handler strip the 5-byte framing header, look up the schema, and decode the Avro payload itself. Re-publishing works the same way in reverse: prepend the framing header before publishing. Reach for `RawBytesCodec` only when the format genuinely isn't `serde`-friendly. Once the bytes leave the handler, the type system can no longer help you. Runnable example: [`examples/nats/raw_bytes.rs`](https://github.com/zannis/shove/blob/main/examples/nats/raw_bytes.rs). ### Migrating hand-rolled Topic impls If you implement `Topic` by hand instead of via `define_topic!`, you now need an explicit `type Codec` line: ```rust,no_run impl Topic for OrderSettlement { type Message = SettlementEvent; type Codec = JsonCodec; fn topology() -> &'static QueueTopology { /* ... */ } } ``` `JsonCodec` preserves the old behaviour exactly. Macro-generated topics already do this for you. ### Custom codecs Implement `Codec` for your own type when neither JSON, Protobuf, nor raw bytes fits. The contract is round-trip safety: `decode(encode(m).unwrap()).unwrap() == m` for every `m` the codec supports. Encode and decode errors should surface as `ShoveError::Codec { codec: "", source }` so consumers see a consistent error variant. ## 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` where `T: Topic`: ```rust,no_run pub trait MessageHandler: Send + Sync + 'static { type Context: Clone + Send + Sync + 'static; fn handle( &self, message: T::Message, metadata: MessageMetadata, ctx: &Self::Context, ) -> impl Future + Send; fn handle_dead( &self, message: T::Message, metadata: DeadMessageMetadata, ctx: &Self::Context, ) -> impl Future + 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` where cloning is a cheap reference-count increment rather than a deep copy. When there is no shared state to inject, use `()`: ```rust,no_run struct Handler; impl MessageHandler 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: ```rust,no_run #[derive(Clone)] struct AppState { db: Arc, } struct Handler; impl MessageHandler 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::(cfg, || Handler).await?; ``` `Context` should be cheap to clone. `Arc` 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, and `metrics::record_failed(..., FailReason::Timeout)` is emitted. The default is 30 seconds. Three layers control the timeout, in priority order — the first one set wins: 1. **Per-group override** — `ConsumerGroupConfig::with_handler_timeout(Duration)`. Lives on the group config; rebuilt per `register` call. 2. **Registry default** — `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`. ```rust,no_run let mut group = broker .consumer_group() .with_default_handler_timeout(Duration::from_secs(15)); group .register::( 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::::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 `spawn`ed 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. ### 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. * `headers: HashMap` — string-valued headers attached to the delivery. 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` increments it explicitly, whereas `redelivered` reflects the broker's view of the transport. ### 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. ```rust,no_run impl MessageHandler 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::()`. 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. ```rust,no_run use shove::MessageHandlerExt; let handler = OrderHandler.audited(MyAuditSink); group.register::(cfg, move || handler.clone()).await?; ``` See [Audit Logging](/guides/audit) 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\ Pattern](/concepts/broker). ## Outcomes & Delivery Handlers in `shove` do not write to queues directly. They return one of four `Outcome` values after processing a message, and the consumer routes the message accordingly. The `Outcome` is the only mechanism a handler has to influence what happens next. ### The four Outcome variants The `Outcome` enum has four variants, each mapped to a distinct delivery action: **`Ack`** — the message was handled successfully. The consumer removes it from the queue and moves on. Use this when your handler has done its work: the record was written, the email was sent, the downstream API responded correctly. **`Retry`** — a transient failure occurred. Network blip, downstream service temporarily down, database deadlock — something went wrong that is likely to resolve itself. The consumer routes the message to a hold queue whose index corresponds to the current retry count, waits for the configured delay, then redelivers the message to the main queue. The retry counter increments on each `Retry`. **`Reject`** — a permanent failure. The message is malformed, violates a business rule, or references data that will never exist. No amount of waiting will fix it. The consumer routes the message to the DLQ if one is configured, or discards it with a warning log if not. The message is never retried. **`Defer`** — the message is not failed, just not yet ready. The consumer routes it to `hold_queues[0]` (the shortest delay) without incrementing the retry counter. This does not burn the retry budget. See [Defer — the special case](#defer--the-special-case) below. ### At-least-once and idempotency `shove` guarantees at-least-once delivery by default. Under normal conditions each message is delivered once, but redeliveries can occur after a crash, a network blip, or an unacked timeout. Handlers must be idempotent. Per-backend deduplication aids exist — RabbitMQ stamped with `x-message-id`, NATS `Nats-Msg-Id`, Kafka offset tracking — but these are aids, not guarantees. A fully idempotent handler is the only reliable defence. ### Retry math Hold queue selection is automatic. When a handler returns `Retry`, the consumer picks: ```text index = min(retry_count, hold_queues.len() - 1) ``` With hold queues configured at `[5s, 30s, 120s]` and `max_retries = 10`: * 1st retry (`retry_count = 0`) → 5s hold queue * 2nd retry (`retry_count = 1`) → 30s hold queue * 3rd–10th retry (`retry_count = 2..9`) → 120s hold queue (clamped at last index) * 11th attempt exhausted → routed to DLQ (or discarded if no DLQ) The topic topology defines the hold queues; the consumer options define `max_retries`. See [Retries, Hold Queues & DLQs](/guides/retries) for a deeper dive. ### Reject vs Retry Choose `Reject` when the message will never succeed regardless of when it is retried: a malformed payload, a business-rule violation, a foreign-key constraint that will never be satisfied. Retrying those messages wastes resources and burns through the retry budget before the inevitable DLQ routing. Choose `Retry` when the failure is transient: a network blip, a downstream service that is temporarily down, a database deadlock. The message is valid and the handler will likely succeed once conditions improve. When in doubt, ask: "Would this handler succeed if I called it again in five minutes?" If yes, `Retry`. If no, `Reject`. ### Defer — the special case `Defer` exists for messages that are not failed — they are simply not ready yet. Examples: * A scheduled-for-the-future event that arrived early. * An order awaiting payment confirmation that has not yet appeared. * An upstream API that returned 429 (rate limited); the payload is valid, just the timing is wrong. `Defer` routes the message to `hold_queues[0]` (shortest delay) with the same retry count. It does not increment the counter and will never trigger DLQ routing as a result of deferral alone. One caution: because `Defer` never increments the retry counter, a handler that always defers will loop the message between the main queue and `hold_queues[0]` indefinitely. There is no built-in circuit breaker. Ensure your handler has a condition that eventually resolves to `Ack`, `Retry`, or `Reject`. If no hold queues are configured on the topic, `Defer` falls back to broker nack-with-requeue and logs a warning. `Defer` is not supported on sequenced consumers. If a handler returns `Defer` in that context it is treated as `Retry` and a warning is logged, because deferral would violate ordering guarantees. ### Handler timeouts and deserialization failures Two failure modes bypass the handler entirely: **Handler timeout.** Configured via `ConsumerOptions::with_handler_timeout`. If the handler future does not complete within the timeout, the consumer aborts it and treats the result as `Outcome::Retry`. The message will be retried according to normal retry math. **Deserialization failure.** A message whose bytes cannot be deserialized into `T::Message` is permanently broken — no amount of retrying will fix a corrupt payload. The consumer treats it as `Outcome::Reject` without invoking the handler at all. The message goes to the DLQ (or is discarded with a warning) and is never retried. This is by design: poison-pill messages should not burn through the retry budget indefinitely. For more on handler registration and context, see [Handlers & Context](/concepts/handlers). ## Topics & Topology A topic is how you express a domain event as a type. When you define `OrderSettlement`, you're declaring that these events flow through a specific queue, carry a specific message shape, and obey specific retry rules. Publishers, consumer registrations, and topology declarations all reference the topic; queue names and message types are derived from it. One topic always carries one message type. Two topics may share the same message type if two independent event flows happen to share a schema, but a single topic never carries mixed types. The type system enforces this: publishing `MyEvent` via the wrong topic is a compile error, not a runtime surprise. Think of topics as the unit of design: one topic per logical event flow. ### The Topic trait Every topic implements the `Topic` trait from `src/topic.rs`: ```rust,no_run pub trait Topic: Send + Sync + 'static { type Message: Serialize + DeserializeOwned + Send + Sync + 'static; fn topology() -> &'static QueueTopology; const SEQUENCE_KEY_FN: Option String> = None; } ``` `topology()` returns `&'static QueueTopology` — computed once via `OnceLock` and returned cheaply on every subsequent call. The default value of `SEQUENCE_KEY_FN` is `None`; `define_sequenced_topic!` overrides it automatically. ### define\_topic! — the fast path In practice, you almost never implement `Topic` by hand. The `define_topic!` macro handles the boilerplate: ```rust,no_run define_topic!( OrderSettlement, SettlementEvent, TopologyBuilder::new("order-settlement") .hold_queue(Duration::from_secs(5)) .dlq() .build() ); ``` The macro accepts an optional visibility modifier (`pub`, `pub(crate)`, etc.) before the name. To understand what it generates, here is the equivalent manual implementation: ```rust,no_run struct OrderSettlement; impl Topic for OrderSettlement { type Message = SettlementEvent; fn topology() -> &'static QueueTopology { static TOPOLOGY: OnceLock = OnceLock::new(); TOPOLOGY.get_or_init(|| { TopologyBuilder::new("order-settlement") .hold_queue(Duration::from_secs(5)) .dlq() .build() }) } } ``` The macro and the manual form are identical in behaviour. Use the macro; it is less error-prone. ### TopologyBuilder `TopologyBuilder` is a chained builder that describes the queues a topic needs. Call `.build()` at the end to produce a `QueueTopology`. The full chain of methods, in the order you would typically call them: * `.new(name)` — required; the primary queue name. * `.hold_queue(Duration)` — add a delayed-redelivery queue; callable multiple times for escalating backoff. * `.dlq()` — add a dead-letter queue named `{name}-dlq`. * `.dlq_named(name)` — add a dead-letter queue with a custom name. * `.sequenced(SequenceFailure)` — enable strict per-key ordering; see [Sequenced Topics](/guides/sequenced). * `.routing_shards(N)` — override the number of consistent-hash shards (default 8); must be called after `.sequenced()`. * `.allow_message_loss()` — suppress the DLQ/hold-queue guards for sequenced topics where message loss is acceptable. * `.build()` — produce the `QueueTopology`. Consider this topology: ```rust,no_run TopologyBuilder::new("order-settlement") .hold_queue(Duration::from_secs(5)) .hold_queue(Duration::from_secs(30)) .dlq() .build() ``` The builder derives all queue names from the primary name: | Component | Generated name | | ---------- | --------------------------- | | Main queue | `order-settlement` | | Hold (5s) | `order-settlement-hold-5s` | | Hold (30s) | `order-settlement-hold-30s` | | DLQ | `order-settlement-dlq` | ### Hold queues Hold queues give a message a second chance. When a handler returns `Outcome::Retry`, the consumer routes the message to a hold queue and parks it there for the configured duration before redelivering it to the main queue. This is how you recover from transient failures — a downstream service that momentarily times out, a database deadlock, a third-party API that returns 503 — without immediately sending messages to the DLQ. The hold queue selected on each retry is `min(retry_count, hold_queues.len() - 1)`. This means the first retry uses the first (shortest) hold queue, the second retry uses the second, and all subsequent retries stay on the last (longest) hold queue. Ordering matters: define hold queues from shortest to longest. For a detailed walkthrough of retry mechanics and worked examples, see [Retries, Hold Queues & DLQs](/guides/retries). ### DLQ The DLQ is where permanently-failed messages go for human inspection or some specific handling. When a message is rejected (via `Outcome::Reject`, or because `max_retries` is exhausted), it is routed to the DLQ rather than discarded. This lets an operator investigate why messages failed, fix the root cause, and replay them without needing to re-publish from the source. When no DLQ is configured, rejected messages are discarded and a warning is logged — they are not silently dropped, but they are gone. For production topics with any business significance, configure a DLQ. A DLQ is itself consumable. Call `consumer.run_dlq::()` to start a loop that processes dead-lettered messages. The [Outcomes & Delivery](/concepts/outcomes) page explains the `handle_dead` hook. ### Sequenced topics — quick mention When message order matters per-key — for example, all ledger entries for a given account must be processed in publish order — use `define_sequenced_topic!` and the `SequencedTopic` trait instead of `define_topic!`. The macro wires a key-extraction function into the `SEQUENCE_KEY_FN` constant, and the publisher routes messages with the same key to the same shard. See [Sequenced Topics](/guides/sequenced) for the full story. ### Topology declaration Before publishing or consuming, call `broker.topology().declare::().await?` once at startup. This call is idempotent — the backend creates queues, exchanges, streams, and any other infrastructure the topology requires, and does nothing if they already exist. It is safe to call on every restart. ## Integrate with your existing stack `shove` sits on top of the broker you already run. Drop it into your code, point it at your broker, and your handlers get retries, ordering, consumer groups, and audit logging without rewriting the rest of your service. ### Per-broker setup * **[RabbitMQ →](/backends/rabbitmq)** — consistent-hash routing, ordering shards via single-active-consumer, optional exactly-once via AMQP transactions. * **[AWS SNS + SQS →](/backends/sqs)** — fully managed; SNS fanout, SQS polling. Uses `consumer_supervisor()` since SQS has no coordinated-group primitive. * **[NATS JetStream →](/backends/nats)** — subject-based shard routing for per-key ordering, JetStream-backed durability. * **[Apache Kafka →](/backends/kafka)** — partition-key routing, native consumer groups with consumer-lag autoscaling, log retention for replay. * **[Redis / Valkey Streams →](/backends/redis)** — stream-based delivery with FNV-1a shard routing; requires Redis 6.2+ or any Valkey release. * **[In-process →](/backends/inmemory)** — zero-dependency, in-RAM. For tests and single-process apps; same API as the durable backends. ### Same code, every broker The `Topic` definition, the `MessageHandler` impl, and the call sites stay identical no matter which broker you point your `Broker\` at. Tests run against `InMemory`; production runs against your real broker; the application code doesn't change. It is normal to have multiple `Broker\` instances in one process. A common pattern is `Broker\` in tests and `Broker\` (or another durable backend) in production, with the same `define_topic!` and handler code feeding both. ## In-process The fastest way to get started with `shove` and the right choice for tests. No Docker, no AWS, no broker process — the broker lives in process RAM. Suitable for unit tests, integration tests, and single-process applications where you want the full pub/sub API without external infrastructure. ### What you need Nothing. No services to start, no credentials to configure. ### Install ```sh cargo add shove --features inmemory ``` ### Connect ```rust // [!include ~/examples/inmemory/basic.rs:connect] ``` `InMemoryConfig::default()` is the only config needed. The broker lives in process RAM and is dropped when the variable goes out of scope. ### Declare topology ```rust // [!include ~/examples/inmemory/basic.rs:declare] ``` `topology().declare::()` registers the topic with the in-process broker. Idempotent — safe to call multiple times. ### Publish ```rust // [!include ~/examples/inmemory/basic.rs:publish] ``` `publisher().await?` returns a `Publisher`. Messages are delivered to consumers in the same process synchronously via in-memory channels. ### Consume ```rust // [!include ~/examples/inmemory/basic.rs:consume] ``` `consumer_group()` creates an in-process consumer group. `InMemoryConsumerGroupConfig::new(min..=max)` sets the concurrency bounds. The `run_until_timeout` call blocks until the shutdown signal fires or the drain timeout elapses. ### Sequenced delivery Per-key FIFO shards in process memory. The same `T::sequence_key()` semantics as durable backends — messages for the same key are processed in order, and different keys are independent and may run concurrently. No persistence: ordering is enforced in-memory only and is lost on shutdown. See [Sequenced Topics](/guides/sequenced) for the full ordering model. ### Consumer groups + autoscaling In-process queue-depth tracking drives the autoscaler. The autoscale signal works the same shape as on durable backends — queue depth within the in-process channels drives scale-up and scale-down. Configure via `InMemoryConsumerGroupConfig`: ```rust,no_run use shove::inmemory::InMemoryConsumerGroupConfig; use shove::{Broker, ConsumerGroupConfig, InMemory}; use std::time::Duration; let mut group = broker.consumer_group(); group .register::( ConsumerGroupConfig::new( InMemoryConsumerGroupConfig::new(1..=4) .with_prefetch_count(8) .with_max_retries(5) .with_handler_timeout(Duration::from_secs(10)) .with_max_message_size(64 * 1024) .with_max_pending_per_key(50), ), || MyHandler, ) .await?; ``` All builder methods mirror the durable backends, so test code written against the in-process backend is structurally identical to production code on RabbitMQ/Kafka/Redis. `with_default_handler_timeout(Duration)` on `consumer_group()` sets a registry-wide default — useful when most test topics share a deadline. ### Gotchas * **No durability.** Messages live in process RAM and are dropped on shutdown. Do not use the in-process backend for production business-critical flows. * **No cross-process delivery.** Only consumers registered in the same process see messages. This is by design — the in-process backend is not a substitute for a real broker. * **Intended for tests and single-process apps.** It is the right default for unit tests, integration tests that don't need a broker, and single-binary applications where all producers and consumers run in the same process. ### Examples * [Basic](/backends/inmemory/examples/basic) — publish/consume round trip with hold queues and DLQ * [Sequenced](/backends/inmemory/examples/sequenced) — per-key FIFO ordering * [Consumer Groups](/backends/inmemory/examples/consumer-groups) — coordinated groups with autoscaling * [Audited Consumer](/backends/inmemory/examples/audited) — `MessageHandlerExt::audited` wrapping * [Stress](/backends/inmemory/examples/stress) — throughput benchmarking ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## Apache Kafka If you already have a Kafka cluster or need streaming semantics — replayable log, partition-key ordering, consumer-lag autoscaling — this backend fits naturally. Best for high-throughput pipelines where log retention and replay matter. ### What you need A Kafka cluster in KRaft mode or with Zookeeper. For local dev: ```sh docker run --rm -p 9092:9092 confluentinc/cp-kafka:latest ``` The integration tests use [testcontainers](https://crates.io/crates/testcontainers) with the Apache Kafka module, so any runnable example also spins up a container automatically. ### Install ```sh cargo add shove --features kafka ``` For TLS + SASL (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512): ```sh cargo add shove --features kafka-ssl ``` For AWS MSK with IAM authentication: ```sh cargo add shove --features kafka-msk-iam ``` ### Connect ```rust // [!include ~/examples/kafka/basic.rs:connect] ``` `KafkaConfig::new` takes a bootstrap server address (e.g. `localhost:9092`). TLS and SASL are configured via `KafkaTls` and `KafkaSasl` on the config — available when the `kafka-ssl` feature is enabled. ### Connecting to AWS MSK The `kafka-msk-iam` feature adds IAM-based authentication for Amazon MSK clusters. It pulls in `aws-config`, `aws-credential-types`, and `aws-msk-iam-sasl-signer`. Use it alongside `kafka-ssl`: ```sh cargo add shove --features kafka-ssl,kafka-msk-iam ``` MSK IAM clusters listen on port `9098` (SASL/IAM). SCRAM/PLAIN clusters use port `9096`. #### Minimal setup MSK brokers use publicly-signed ACM certificates. `KafkaTls::default()` is correct — the OS trust store handles validation. No custom CA path is needed. ```rust use shove::kafka::{KafkaConfig, KafkaSasl, KafkaTls}; let config = KafkaConfig::new("b-1.cluster.amazonaws.com:9098,b-2.cluster.amazonaws.com:9098") .with_tls(KafkaTls::default()) .with_sasl(KafkaSasl::msk_iam("eu-west-2")); ``` `KafkaSasl::msk_iam(region)` resolves credentials from the standard AWS provider chain: environment variables, shared credentials file, EC2 instance metadata (IMDS), EKS pod identity / IRSA, and SSO. No explicit credential configuration is needed in most deployment environments. For a non-default named profile, use `KafkaSasl::msk_iam_with_profile`: ```rust let config = KafkaConfig::new(brokers) .with_tls(KafkaTls::default()) .with_sasl(KafkaSasl::msk_iam_with_profile("eu-west-2", "production")); ``` The OAUTHBEARER mechanism and `SASL_SSL` security protocol are set automatically. Do not set `sasl.mechanism` or `security.protocol` manually, and do not set `sasl.oauthbearer.config` — token rotation is handled automatically by the library. See [`examples/kafka/msk_iam.rs`](/backends/kafka/examples/msk_iam) for a runnable walkthrough. ### Declare topology ```rust // [!include ~/examples/kafka/basic.rs:declare] ``` `topology().declare::()` creates the Kafka topics for the main topic, hold topics, and DLQ. Idempotent — safe to call on every startup. ### Publish ```rust // [!include ~/examples/kafka/basic.rs:publish] ``` `publisher().await?` returns a `Publisher`. Messages are produced via `rdkafka`. The message key is derived from the topic's partition-key logic (or from `T::sequence_key()` for sequenced topics). ### Consume ```rust // [!include ~/examples/kafka/basic.rs:consume] ``` `consumer_group()` registers a native Kafka consumer group. `KafkaConsumerGroupConfig::new(min..=max)` sets the autoscale bounds. Kafka handles partition assignment and rebalance automatically when the group membership changes. #### Group configuration ```rust,no_run use shove::kafka::{KafkaAutoOffsetReset, KafkaConsumerGroupConfig}; let cfg = KafkaConsumerGroupConfig::new(1..=8) .with_prefetch_count(20) .with_max_retries(5) .with_handler_timeout(Duration::from_secs(30)) .with_concurrent_processing(true) .with_group_id("billing-orders-consumer") // override the default `{queue}-consumer` .with_auto_offset_reset(KafkaAutoOffsetReset::Latest); ``` * **`with_prefetch_count(u16)`** — librdkafka in-flight cap per consumer task. Default `10`. * **`with_max_retries(u32)`** — retry budget before dead-lettering. Default `10`. * **`with_handler_timeout(Duration)`** — per-message wall-clock deadline. Default `30s` (see [Handlers & Context](/concepts/handlers#handler-timeouts)). * **`with_concurrent_processing(bool)`** — dispatch each fetched message to its own tokio task (rejected for sequenced topics). Default `false`. * **`with_group_id(impl Into)`** — override the broker-side consumer group ID. Defaults to `"{queue}-consumer"`. Set this when two independent services consume the same topic and must each receive every message (fan-out) — otherwise they share a group and compete for partitions. * **`with_auto_offset_reset(KafkaAutoOffsetReset)`** — `Earliest` (default, replay history), `Latest` (tail-only), or `None` (refuse silent replay/skip on a fresh group). #### Replication factor Auto-created topics get replication factor `1` by default, which is fine for single-broker dev but unsafe in production. Set a cluster-wide default on the registry: ```rust,no_run let mut group = broker .consumer_group() .with_default_replication_factor(3); // applied to every auto-created topic ``` Or set it per-declaration on the topology declarer: ```rust,no_run broker .topology() .with_replication_factor(3) .declare::() .await?; ``` `create_topic` is idempotent and will not lower an existing topic's replication. Pre-creating topics out of band (Terraform, MSK console) is also fine — the declarer is a no-op when the topic already exists. ### Sequenced delivery Messages for the same key stay in order. Kafka uses partition-key routing: `T::sequence_key()` becomes the Kafka message key, so messages with the same key always land on the same partition. A single consumer handles each partition at a time, guaranteeing that messages for the same key are never processed concurrently. Ordering is partition-scoped: two messages with different keys may land on different partitions and be processed concurrently. The partition count is fixed at topic creation time and caps the maximum degree of parallelism across the consumer group. See [Sequenced Topics](/guides/sequenced) for the full ordering model, and [Sequenced example](/backends/kafka/examples/sequenced) for a runnable walkthrough. ### Consumer groups + autoscaling Kafka consumer groups are native — `broker.consumer_group()` creates a standard Kafka consumer group. Call `register` to associate a topic with handlers and bounds: ```rust,no_run use shove::kafka::KafkaConsumerGroupConfig; use shove::{Broker, ConsumerGroupConfig, Kafka}; let mut group = broker.consumer_group(); group .register::( ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=4)), || MyHandler, ) .await?; ``` The autoscaler measures consumer lag (the offset gap between the latest produced message and the latest committed offset) and adjusts the number of active consumers within the `min..=max` range. Note: consumer-group rebalances occur during scale-up and scale-down events. Rebalances cause a brief delivery pause while Kafka reassigns partitions. Plan for this when setting autoscale bounds and drain timeouts. See the [Basic example](/backends/kafka/examples/basic) for a full runnable walkthrough. ### Offset commit semantics `Outcome::Ack` commits the offset immediately after the handler returns. `Outcome::Retry` and `Outcome::Defer` defer the offset commit until the delayed republish to the hold topic has been acknowledged by the broker. This closes the publish-then-commit race: a broker crash or process kill between handler return and republish redelivers the message on restart rather than silently dropping it. On shutdown the consumer drains the in-flight republish queue before exiting the poll loop, so pending Retry/Defer messages are persisted even if the shutdown signal arrives mid-batch. `Outcome::Reject` routes to the DLQ (if configured) and commits the offset; the original is not redelivered. Messages that exhaust `max_retries` follow the same DLQ-then-commit path. ### Schema Registry The `kafka-schema-registry` feature adds [Confluent Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html) support for Kafka consumers (decode) and producers (encode). On the consume side, each incoming message is unwrapped from the Confluent wire frame (magic byte `0x00` + 4-byte big-endian schema id; for Protobuf, also the message-index array), the schema id is resolved against the registry (cached in memory, with single-flight deduplication of concurrent cold misses and a configurable negative-TTL for permanent failures), and the inner payload is decoded via the topic's existing `Codec` (`JsonCodec` or `ProtobufCodec`). On the produce side, an opt-in publisher wraps each encoded payload in the same wire frame — see [Producer-side encoding](#producer-side-encoding). The feature works with the Confluent Schema Registry and with **Redpanda's built-in Schema Registry**, which exposes the same Confluent-compatible REST API and wire format. The e2e test suite validates both paths. #### Install ```sh cargo add shove --features kafka-schema-registry ``` #### Build the registry client ```rust,no_run use std::time::Duration; use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth}; let registry = SchemaRegistry::builder("http://schema-registry:8081") .auth(SchemaRegistryAuth::Basic { user: "sr-user".into(), pass: "sr-pass".into(), }) .timeout(Duration::from_secs(3)) .max_retries(2) .negative_cache_ttl(Duration::from_secs(60)) .build(); ``` `SchemaRegistry::builder(url)` accepts any `http://` or `https://` base URL. Auth options are: * `SchemaRegistryAuth::None` — no authentication (default) * `SchemaRegistryAuth::Bearer(token)` — `Authorization: Bearer ` * `SchemaRegistryAuth::Basic { user, pass }` — HTTP Basic auth The returned value is an `Arc`. Clone it to share the same schema cache across multiple consumers or a consumer group. #### Per-consumer configuration ```rust,no_run use std::sync::Arc; use shove::schema_registry::{SchemaEnforcement, SchemaRegistry, SchemaRegistryAuth}; use shove::consumer::ConsumerOptions; use shove::markers::Kafka; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); let opts = ConsumerOptions::::new() .with_schema_registry(Arc::clone(®istry)) .with_schema_enforcement(SchemaEnforcement::Enforce) .accept_schema_subjects(["orders-value"]); ``` #### Per-consumer-group configuration Attaching the registry on a `KafkaConsumerGroupConfig` shares the `Arc` — and therefore the same in-memory schema cache — across every consumer spawned in the autoscaling group: ```rust,no_run use std::sync::Arc; use shove::kafka::{KafkaConsumerGroupConfig}; use shove::schema_registry::{SchemaEnforcement, SchemaRegistry}; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); let cfg = KafkaConsumerGroupConfig::new(1..=8) .with_schema_registry(Arc::clone(®istry)) .with_schema_enforcement(SchemaEnforcement::Enforce) .accept_schema_subjects(["orders-value"]); ``` #### Enforcement modes `with_schema_enforcement` controls what happens when a message's registered subject is not in the accepted set: * **`SchemaEnforcement::Enforce`** (default) — the message is routed to the DLQ with death reason `schema_validation_failed`. Choose this for production where a subject mismatch is a producer misconfiguration. * **`SchemaEnforcement::Permissive`** — the mismatch is logged and counted, and the message is decoded anyway. Use this during migration windows or when multiple producers share a topic with different subject conventions. #### Accepted subjects `accept_schema_subjects([...])` pins the set of Confluent schema subjects that are accepted for a consumer. When not called, the default is derived from the queue name using the Confluent [TopicNameStrategy](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#subject-name-strategy): `"{queue}-value"`. For a topic named `orders`, that resolves to `orders-value`. #### Producer-side encoding The same feature lets a publisher emit Confluent-framed messages, symmetric with how the consumer is configured. Attach a `SchemaRegistry` to a `KafkaPublisherConfig` and obtain the publisher with `Broker::publisher_with`; the publisher then wraps each codec-encoded payload in the Confluent wire frame using the latest registered schema id for the subject. Like the consumer side, framing is a publisher-layer concern — it is **not** baked into the topic's `Codec`. ```rust,no_run use std::sync::Arc; use shove::kafka::KafkaPublisherConfig; use shove::schema_registry::SchemaRegistry; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); // `broker` is a `Broker`. let publisher = broker .publisher_with(KafkaPublisherConfig::new().with_schema_registry(Arc::clone(®istry))) .await?; // `publisher.publish::(&order).await?` now emits SR-framed bytes. ``` Details: * **Subject** — defaults to the Confluent TopicNameStrategy `"{topic}-value"`; override with `KafkaPublisherConfig::with_subject("...")`. * **Schema id** — resolved once per subject via `GET /subjects/{subject}/versions/latest` and cached. shove carries no schema text; the subject must already be registered. * **Codecs** — framing applies only to `JsonCodec` and `ProtobufCodec` topics; other codecs are published unframed. For Protobuf the message-index is `[0]` (the first message type). * **Opt-in** — a publisher built with the plain `Broker::publisher()` does no framing, so existing publishers are byte-for-byte unchanged. #### Codec requirement Registry decoding is only supported for topics using a `JsonCodec` or `ProtobufCodec`. A message arriving on a topic whose codec is neither JSON nor Protobuf is routed to the DLQ with reason `schema_unsupported_codec`. #### Redpanda compatibility The Confluent wire format (magic byte + schema id) and REST API endpoints used by `kafka-schema-registry` are fully compatible with Redpanda's built-in Schema Registry. No additional configuration is needed — point `SchemaRegistry::builder` at the Redpanda Schema Registry URL. #### DLQ consumer caveat :::warning If you enable registry decoding on the consumer that drains the DLQ (`handle_dead` path), any dead message that cannot be resolved or validated against the registry cannot be delivered to `handle_dead`. Such messages are acked-and-dropped — matching the existing behavior for undecodable dead messages. Operators concerned about this should omit the registry on the DLQ consumer, or set `SchemaEnforcement::Permissive` so subject mismatches do not block delivery. ::: ### Metrics The Kafka backend emits `shove_messages_failed_total` with the following `reason` labels (see [Observability](/guides/observability#metric-reference)): * `oversize` — payload exceeds `max_message_size` before deserialization. * `deserialize` — JSON/codec decode failed; routed to DLQ. * `timeout` — handler exceeded `handler_timeout`; retried. * `max_retries_exceeded` — retry budget exhausted; routed to DLQ. * `rejected` — handler returned `Outcome::Reject`; routed to DLQ. * `schema_validation_failed` — schema subject not in the accepted set under `Enforce` mode; routed to DLQ. Requires `kafka-schema-registry`. * `schema_unsupported_codec` — registry decoding requested but the topic uses a codec other than JSON or Protobuf; routed to DLQ. Requires `kafka-schema-registry`. `shove_backend_errors_total` carries the `backend="kafka"` label for connection drops, produce failures, and consume-stream errors. ### Gotchas * **Replication factor defaults to `1`** — safe for single-broker dev, unsafe in production. Set `with_default_replication_factor(3)` on the registry (or `with_replication_factor` on the topology declarer) before the first declaration. The default exists so the no-config dev path works; the constant lives in `kafka::constants::DEFAULT_REPLICATION`. * **Partition count is set at topic creation time** and is rarely changed after the fact. Choose carefully — partition count caps the maximum degree of parallelism a consumer group can achieve (one consumer per partition maximum). * **Consumer-group rebalances** during scale events cause brief delivery pauses. The larger the group and the more topics, the longer the rebalance. shove pins `session.timeout.ms` to 10 seconds and `max.poll.interval.ms` to 5 minutes (constants in `kafka::constants`); neither is settable via `KafkaConfig`. If handlers are slow, raise the handler timeout with `with_handler_timeout` (per group) or `with_default_handler_timeout` (registry-wide) instead. * **On Windows**, the `cmake-build` feature is required for `rdkafka` (the underlying C library). This is handled automatically via the `[target.'cfg(windows)'.dependencies]` entry in `Cargo.toml`. * **SASL and TLS** require the `kafka-ssl` feature. SCRAM-SHA-256 and SCRAM-SHA-512 are supported out of the box. GSSAPI/Kerberos requires downstream activation of `rdkafka/gssapi`. For AWS MSK IAM, use the `kafka-msk-iam` feature instead. * **`librdkafka` system dependencies** on Debian/Ubuntu CI runners: `libsasl2-dev` is required when using `kafka-ssl`. Add it to your CI runner setup (the shove `ci.yml` already does this). ### Examples * [Basic](/backends/kafka/examples/basic) — publish/consume round trip with hold queues and DLQ * [Sequenced](/backends/kafka/examples/sequenced) — partition-key ordering * [Audited Consumer](/backends/kafka/examples/audited) — `MessageHandlerExt::audited` wrapping * [Stress](/backends/kafka/examples/stress) — throughput benchmarking * [MSK IAM](/backends/kafka/examples/msk_iam) — IAM authentication against Amazon MSK * [Schema Registry](/backends/kafka/examples/schema-registry) — Confluent Schema Registry decode with subject enforcement ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## NATS JetStream If NATS is already in your stack, this backend gives you everything you need: subject-routed messaging with per-key ordering, JetStream-backed durability, coordinated pull consumers, and pending-message autoscaling. Fast, simple, and lighter to operate than AMQP brokers. ### What you need A NATS server with JetStream enabled. For local dev: ```sh docker run --rm -p 4222:4222 nats:latest -js ``` The `-js` flag enables JetStream on the server. JetStream streams are provisioned automatically when you call `topology().declare::()`. ### Install ```sh cargo add shove --features nats ``` ### Connect ```rust // [!include ~/examples/nats/basic.rs:connect] ``` `NatsConfig::new` takes a NATS server URL. Authentication and TLS are populated by writing the matching public fields on the struct after `new` — there are no setter builders, just plain field assignment: ```rust,no_run use shove::nats::NatsConfig; use std::path::PathBuf; let mut config = NatsConfig::new("tls://broker.example.com:4222"); // One of the auth slots — pick whichever the server expects: config.username = Some("svc-billing".into()); config.password = Some(std::env::var("NATS_PASSWORD")?); // or token / nkey_seed / creds_file config.token = Some(std::env::var("NATS_TOKEN")?); config.nkey_seed = Some(std::env::var("NATS_NKEY_SEED")?); config.creds_file = Some(PathBuf::from("/etc/nats/user.creds")); // TLS material: config.tls_ca_cert = Some(PathBuf::from("/etc/nats/ca.pem")); config.tls_client_cert = Some(PathBuf::from("/etc/nats/client.pem")); config.tls_client_key = Some(PathBuf::from("/etc/nats/client.key")); ``` All slots are `Option<…>` — set only the ones your deployment uses. The `Debug` impl redacts every credential and the user-info portion of the URL. ### Declare topology ```rust // [!include ~/examples/nats/basic.rs:declare] ``` `topology().declare::()` provisions JetStream streams for the topic: the main stream, plus a DLQ stream when `.dlq()` is configured. Sequenced topics declare a single main stream with subject shards (`{queue}.shard.{N}`). Durable consumers are not created here — the consumer runtime creates them on demand when you start consuming. The call is idempotent and safe to invoke on every startup: it reconciles an existing stream to the configured shape (via `create_or_update_stream`) rather than leaving a stale config in place. ### Stream management By default `declare::()` creates the main stream with shove's defaults — `WorkQueue` retention, file storage, a single replica, and **no size or age limit**. Two NATS-specific `TopologyBuilder` options change how the stream is bounded and who owns it. #### Bounded / replicated managed stream Use `nats_stream_config` when shove should own the stream but you need it bounded — so a stalled consumer can't grow the file store without limit — or replicated for durability: ```rust,no_run use shove::{NatsRetention, NatsStreamConfig, TopologyBuilder}; use std::time::Duration; shove::define_topic!( PriceChanges, PriceChange, TopologyBuilder::new("price-changes") .nats_stream_config(NatsStreamConfig { retention: NatsRetention::Limits, max_age: Some(Duration::from_secs(600)), // drop messages older than 10m max_bytes: Some(1_000_000_000), // cap the file store at ~1 GiB max_messages: None, num_replicas: 3, // R3 for production durability }) .dlq() .build() ); ``` `NatsStreamConfig::default()` is WorkQueue / unbounded / single-replica, so set only the fields you need. On re-declare, **mutable** changes (`max_age`, `max_bytes`, `max_messages`, `num_replicas`) are applied to the existing stream; **immutable** ones (`retention`, storage) make `declare()` fail loudly rather than silently no-op — recreate the stream, or hand ownership to infra with `nats_external_stream()` if you need to change those. #### Bind to an externally-provisioned stream When infra owns the stream — a Terraform module, a `nats` CLI bootstrap job, or an operator that sets retention, replication, or placement outside the app — use `nats_external_stream()`: ```rust,no_run use shove::TopologyBuilder; shove::define_topic!( PriceChanges, PriceChange, // The queue name must match the externally-provisioned stream name. TopologyBuilder::new("CLOB_PRICE_CHANGES") .nats_external_stream() .dlq() .build() ); ``` In this mode `declare::()` does **not** create the main stream — it verifies the stream (named after the queue) already exists and returns an error if it doesn't, so a missing or misnamed stream fails fast at startup instead of silently falling back to a default-configured one. shove still creates and owns its own DLQ stream and the durable consumer on the external stream. Provision the stream before the consumer starts (e.g. order a bootstrap job ahead of the deployment). `nats_external_stream()` and `nats_stream_config()` are mutually exclusive — and both are incompatible with `nats_subjects()` and `sequenced()`, which configure stream *creation* that external mode skips. `build()` panics if they are combined. ### Publish ```rust // [!include ~/examples/nats/basic.rs:publish] ``` `publisher().await?` returns a `Publisher` that publishes to the JetStream stream. Each message is acknowledged by JetStream before `publish` returns. ### Consume ```rust // [!include ~/examples/nats/basic.rs:consume] ``` `consumer_group()` creates a coordinated pull consumer backed by a JetStream durable deliver-group. `NatsConsumerGroupConfig::new(min..=max)` sets the autoscale bounds. The group coordinates via JetStream — only one consumer processes a given message at a time. ### Sequenced delivery Messages for the same key stay in order. NATS uses subject-based shard routing: the publisher derives a subject suffix from `T::sequence_key()` and routes to a dedicated shard subject. Each shard consumer is configured with `max_ack_pending: 1`, meaning at most one message per shard is in-flight at a time. Messages for the same key are never processed concurrently. Different keys map to different shards and are entirely independent — one stuck key does not block others. See [Sequenced Topics](/guides/sequenced) for the full ordering model, and [Sequenced example](/backends/nats/examples/sequenced) for a runnable walkthrough. ### Consumer groups + autoscaling NATS consumer groups are coordinated via JetStream pull consumers with a deliver group name. Call `broker.consumer_group()` and register topics with a `ConsumerGroupConfig`: ```rust,no_run use shove::nats::NatsConsumerGroupConfig; use shove::{Broker, ConsumerGroupConfig, Nats}; let mut group = broker.consumer_group(); group .register::( ConsumerGroupConfig::new(NatsConsumerGroupConfig::new(1..=4)), || MyHandler, ) .await?; ``` The autoscaler reads the JetStream pending message count and adjusts the number of active pull consumers within the `min..=max` range. See the [Basic example](/backends/nats/examples/basic) for a complete runnable walkthrough. ### Gotchas * **Stream provisioning at declaration time.** `topology().declare::()` creates JetStream streams with default storage limits unless you pass [`nats_stream_config`](#bounded--replicated-managed-stream). Ensure stream limits (max bytes, max messages, max message size) match your expected load — streams that hit their limits drop or reject new messages depending on the retention policy. When infra owns the stream, bind to it with [`nats_external_stream`](#bind-to-an-externally-provisioned-stream) instead. * **De-duplication via `Nats-Msg-Id`.** shove sets the `Nats-Msg-Id` header on every publish (a fresh UUID per call). JetStream de-duplicates within a configurable window (default 120 seconds) — if a publisher retries the same logical message inside that window, JetStream rejects the duplicate. Note that the header is generated per-publish-call, so deduplicating across application-level retries requires the caller to pass a stable id via `publish_with_headers`. * **Work-queue semantics by default.** shove provisions JetStream streams with work-queue (delete-on-ack) semantics unless you override `retention` via [`nats_stream_config`](#bounded--replicated-managed-stream) or bind to an infra-owned stream with [`nats_external_stream`](#bind-to-an-externally-provisioned-stream). Work-queue streams allow a single logical consumer over a subject — each `consumer_group()` registration gets its own durable consumer on the stream. * **JetStream must be enabled on the server.** Connecting to a server without JetStream and calling `topology().declare::()` will return an error. The `-js` flag is required on the NATS server. * **TLS scheme must match TLS options.** Setting any of `tls_ca_cert`, `tls_client_cert`, or `tls_client_key` on a plaintext `nats://` URL is rejected at connect time. Use `tls://` or `nats+tls://` to make the encrypted transport explicit — this guards against silent downgrade. * **Reconnect bounds.** `ConsumerOptions::::new().with_max_reconnect_attempts(N)` caps reconnect attempts after a connection drop; default is unlimited. Set this when you want the consumer to surface `Connection` errors after `N` failed redials rather than retry forever. * **Stream config is tunable; durable-consumer tuning is fixed.** Stream retention, storage bounds (`max_age` / `max_bytes` / `max_messages`), and `num_replicas` are configurable via [`nats_stream_config`](#bounded--replicated-managed-stream) (or skipped entirely with [`nats_external_stream`](#bind-to-an-externally-provisioned-stream)); the defaults remain `WorkQueue` / `StorageType::File` / `DiscardPolicy::New` / single-replica / unbounded. The **durable consumers** are still declared with `AckPolicy::Explicit`, `max_ack_pending = prefetch × max_consumers`, and a 120 s deduplication window — these are not yet user-configurable, so open an issue if you need them surfaced as knobs. ### Examples * [Basic](/backends/nats/examples/basic) — publish/consume round trip with hold queues and DLQ * [Sequenced](/backends/nats/examples/sequenced) — subject-shard ordering * [Audited Consumer](/backends/nats/examples/audited) — `MessageHandlerExt::audited` wrapping * [Stress](/backends/nats/examples/stress) — throughput benchmarking ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## RabbitMQ If you already run RabbitMQ, this backend slots in with zero infrastructure changes. It covers the full feature surface: hold-queue retry topologies, dead-letter routing, coordinated consumer groups with queue-depth autoscaling, per-key FIFO delivery via consistent-hash exchange routing, and optional AMQP transactional mode for exactly-once routing on handlers with irreversible side effects. ### What you need A RabbitMQ broker 3.13+ with the **management plugin** enabled. The autoscaler reads queue depth from the management HTTP API, so the plugin is required for `consumer_group()` autoscaling. For local dev: ```sh docker run --rm -p 5672:5672 -p 15672:15672 rabbitmq:3-management ``` This exposes AMQP on `5672` and the management UI / API on `15672`. ### Install ```sh cargo add shove --features rabbitmq ``` For exactly-once transactional routing (see below): ```sh cargo add shove --features rabbitmq-transactional ``` ### Connect ```rust // [!include ~/examples/rabbitmq/basic_pubsub.rs:connect] ``` `RabbitMqConfig::new` takes a standard AMQP URL. TLS and authentication are configured via the URL (e.g. `amqps://user:pass@host:5671`). #### Management API Autoscaling needs queue depth from the RabbitMQ management plugin. Attach credentials via `.with_management(...)`: ```rust,no_run use shove::rabbitmq::{ManagementConfig, RabbitMqConfig}; let config = RabbitMqConfig::new("amqp://guest:guest@localhost:5672") .with_management( ManagementConfig::new("http://localhost:15672", "guest", "guest") .with_vhost("my-vhost") // non-default vhost .with_tls_ca_cert("/path/to/ca.pem") // custom CA for self-signed mgmt certs .with_tls_skip_verify(), // dev only — disables verification ); ``` When `management` is `None`, `Broker::queue_stats_provider().snapshot(...)` and `RabbitMqAutoscalerBackend` return a `Topology` error directing the caller to configure it. The `consumer_supervisor()` path does not need the plugin; only autoscaling does. ### Declare topology ```rust // [!include ~/examples/rabbitmq/basic_pubsub.rs:declare] ``` `topology().declare::()` creates the main queue, any hold queues, and the DLQ on the broker. Idempotent — safe to call on every startup. ### Publish ```rust // [!include ~/examples/rabbitmq/basic_pubsub.rs:publish] ``` `publisher().await?` returns a `Publisher` that can publish single messages, messages with custom headers, or batches. ### Consume ```rust // [!include ~/examples/rabbitmq/basic_pubsub.rs:consume] ``` `consumer_supervisor()` fans out one consumer task per registered topic. For coordinated groups with autoscaling, use `consumer_group()` instead — see the section below. ### Sequenced delivery Messages for the same key stay in order. RabbitMQ enforces per-key ordering via a consistent-hash exchange and single-active-consumer (SAC) sub-queues. When you declare a sequenced topic, shove provisions a consistent-hash exchange (`{queue}-seq-hash`) and N sub-queues each configured with SAC and `prefetch=1`. The publisher routes each message by `T::sequence_key()`; the consumer processes one message at a time per shard, which guarantees that messages for the same key are never processed concurrently. The shard count is configurable via `routing_shards(N)` in `TopologyBuilder` (default 8). More shards increase parallelism across different keys at the cost of more queues on the broker. See [Sequenced Topics](/guides/sequenced) for the full ordering model, and [Sequenced Pubsub example](/backends/rabbitmq/examples/sequenced) for a runnable walkthrough. ### Consumer groups + autoscaling For non-sequenced topics, a RabbitMQ consumer group is N consumers competing on the main queue under standard work-queue semantics — the broker hands each message to one consumer. (Single-active-consumer is reserved for sequenced sub-queues, where one consumer per shard is required to preserve ordering — see [Sequenced delivery](#sequenced-delivery) above.) Call `broker.consumer_group()` and register topics with a `ConsumerGroupConfig`: ```rust,no_run use shove::rabbitmq::ConsumerGroupConfig as RabbitMqGroupConfig; use shove::{Broker, ConsumerGroupConfig, RabbitMq}; let mut group = broker.consumer_group(); group .register::( ConsumerGroupConfig::new( RabbitMqGroupConfig::new(1..=8) .with_prefetch_count(20) .with_max_retries(5), ), || MyHandler, ) .await?; ``` The autoscaler reads queue depth from the RabbitMQ management HTTP API and scales the number of active consumers within the `min..=max` range. The management plugin must be running and reachable for autoscaling to work. #### Connection pooling When `max_consumers > 50`, the group automatically pools AMQP connections — one per \~50 workers, rounded up. Lapin parses every inbound delivery on a connection through a single reader task, so a high-throughput group of 200 consumers on one connection bottlenecks on that task. The pool is transparent; all clones of the underlying `RabbitMqClient` share broker config and reconnect logic. #### Stuck-retry eviction If a transient broker error kills the consumer mid-republish, a delivery can land in the per-consumer `AwaitingRetry` map and never escape. `ConsumerOptions::::new().with_hold_queue_timeout(Duration)` evicts any entry that has been waiting longer than the timeout and requeues it to the main queue. The eviction ticker fires at half the configured timeout, so a 30 s setting checks every 15 s. Default: no eviction — set it explicitly to enable. #### Reconnect bounds `ConsumerOptions::::new().with_max_reconnect_attempts(N)` caps how many times the consumer redials after a connection drop before surfacing a `Connection` error. Default: unlimited (retries forever). See the [Consumer Groups example](/backends/rabbitmq/examples/consumer-groups) for a full runnable walkthrough. ### Exactly-once mode Enable with `--features rabbitmq-transactional`. This wraps the publish-to-hold-queue + ack in an AMQP transaction (`tx_select` / `tx_commit`), eliminating the publish-then-ack race condition. The trade-off is approximately 10–15× lower throughput per channel. Reserve this for handlers with irreversible side effects where duplicate delivery is unacceptable. See the [Exactly-Once example](/backends/rabbitmq/examples/exactly-once) and the [Exactly-Once guide](/guides/exactly-once). ### Gotchas * **Management plugin is required** for autoscaler queue-depth polling. Attach credentials via `RabbitMqConfig::with_management(ManagementConfig::new(...))` — when absent, `queue_stats_provider().snapshot(...)` and the autoscaler return a `Topology` error rather than silently using a no-op provider. * **TLS and authentication** are configured via the AMQP URL passed to `RabbitMqConfig::new(...)`. Use `amqps://` for TLS; embed credentials in the URL or configure them via environment variables before constructing the config. * **`lapin` handles the AMQP connection.** `rustls--ring` is the default TLS feature pulled in by shove. Downstream crates that need a different TLS stack (e.g. vendored OpenSSL) can activate alternative `rdkafka` or `lapin` feature flags without changing shove's API. * **Auto-reconnect** is built in at the consumer level. Transient broker failures — network blips, broker restarts — recover automatically without operator intervention. * **Exactly-once throughput cost** (\~10–15×) means it should only be enabled for topics that genuinely require it. The standard at-least-once mode is far faster and handlers should be made idempotent instead where possible. ### Examples * [Basic Pubsub](/backends/rabbitmq/examples/basic) — all `Outcome` variants, hold queues, DLQ, batch publish * [Sequenced Pubsub](/backends/rabbitmq/examples/sequenced) — consistent-hash exchange + SAC ordering * [Consumer Groups](/backends/rabbitmq/examples/consumer-groups) — coordinated groups with autoscaling * [Audited Consumer](/backends/rabbitmq/examples/audited) — `MessageHandlerExt::audited` wrapping * [Concurrent Pubsub](/backends/rabbitmq/examples/concurrent) — high-concurrency publish patterns * [Exactly-Once](/backends/rabbitmq/examples/exactly-once) — transactional AMQP delivery * [Stress](/backends/rabbitmq/examples/stress) — throughput benchmarking ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## 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: ```sh docker run --rm -p 6379:6379 redis:7-alpine ``` ### Install ```sh cargo add shove --features redis-streams ``` ### Connect ```rust use shove::redis::{RedisConfig, RedisMode}; use shove::{Broker, Redis}; let broker = Broker::::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: ```rust,no_run 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 ```rust broker.topology().declare::().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 ```rust let publisher = broker.publisher().await?; publisher.publish::(&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 ```rust use shove::redis::RedisConsumerGroupConfig; use shove::{ConsumerGroupConfig}; let mut group = broker.consumer_group(); group .register::( 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`](/guides/groups#enabling-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](#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::::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. ```rust use shove::consumer_group::ConsumerGroupConfig; use shove::redis::RedisConsumerGroupConfig; group .register_fifo::( 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](/guides/sequenced) 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](/backends/redis/examples/basic) — publish/consume round trip with hold queues and DLQ * [Sequenced](/backends/redis/examples/sequenced) — FNV-1a shard routing for per-key ordering * [Consumer Groups](/backends/redis/examples/consumer-groups) — coordinated `min..=max` group with a burst-and-drain workload * [Autoscaler](/backends/redis/examples/autoscaler) — consumer-lag-driven scale up/down within a `min..=max` range * [Audited Consumer](/backends/redis/examples/audited) — `MessageHandlerExt::audited` wrapping * [Stress](/backends/redis/examples/stress) — throughput benchmarking ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## AWS SNS + SQS If you're already on AWS and don't want to run a broker, this is your path. SNS fans messages out to SQS queues; shove handles polling, retries, DLQ routing, and consumer supervision. No broker process to provision or maintain — pay AWS for what you use. ### What you need An AWS account with IAM credentials that have SNS and SQS permissions, and a region selected. For local dev, [LocalStack Pro](https://localstack.cloud/) emulates both services: ```sh docker run --rm -p 4566:4566 \ -e LOCALSTACK_AUTH_TOKEN=$LOCALSTACK_AUTH_TOKEN \ localstack/localstack-pro ``` Set the `endpoint_url` field in `SnsConfig` to `http://localhost:4566` when targeting LocalStack. Omit it entirely for real AWS. ### Install ```sh cargo add shove --features aws-sns-sqs ``` ### Connect ```rust // [!include ~/examples/sqs/basic_pubsub.rs:connect] ``` `SnsConfig` holds the AWS region and an optional `endpoint_url` override for local dev. AWS credentials are loaded from the standard environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_REGION`) or from the default credential chain. ### Declare topology ```rust // [!include ~/examples/sqs/basic_pubsub.rs:declare] ``` `topology().declare::()` creates the SNS topic and SQS queue(s), including hold queues and the DLQ. Idempotent — safe to call on every startup. ### Publish ```rust // [!include ~/examples/sqs/basic_pubsub.rs:publish] ``` `publisher().await?` returns a `Publisher` that publishes via SNS. SNS fans the message out to all subscribed SQS queues. ### Consume ```rust // [!include ~/examples/sqs/basic_pubsub.rs:consume] ``` SQS uses `consumer_supervisor()` — N independent pollers per registered topic. There is no broker-level coordinated-group primitive on SQS, so `consumer_group()` is not available on `Broker\`. Use `ConsumerOptions::::new()` or `ConsumerOptions::::preset(prefetch_count)` to configure polling behavior. ### Sequenced delivery Messages for the same group key stay in order. SQS FIFO topics enforce ordering natively: when you declare a sequenced topic, shove provisions FIFO SNS topics and SQS queues and sets the `MessageGroupId` from `T::sequence_key()`; SQS delivers messages in publish order within each group. FIFO queue names must end in `.fifo` — shove appends this suffix automatically when sequenced topology is declared. The default SQS FIFO throughput quota is 300 messages/second per queue; for higher throughput, contact AWS to raise the quota or switch to a standard (unordered) topic. See [Sequenced Topics](/guides/sequenced) for the full ordering model, and [Sequenced Pubsub example](/backends/sqs/examples/sequenced) for a runnable walkthrough. ### Consumer supervision + autoscaling SQS has no broker-level coordinated-group primitive, so `Broker\` only exposes `consumer_supervisor()` (not `consumer_group()`). The supervisor fans out one polling task per registered topic with a fixed `ConsumerOptions`: ```rust,no_run use shove::{Broker, ConsumerOptions, Sqs}; let mut supervisor = broker.consumer_supervisor(); supervisor.register::( MyHandler, ConsumerOptions::::preset(10), )?; let outcome = supervisor.run_until_timeout(signal, drain_timeout).await; ``` `consumer_supervisor()` does not scale poller count on its own — `register` accepts one set of `ConsumerOptions` and runs exactly one consumer per topic. #### Tuning ConsumerOptions ```rust,no_run use shove::{ConsumerOptions, Sqs}; use std::time::Duration; let opts = ConsumerOptions::::new() .with_prefetch_count(10) .with_receive_batch_size(10) // SQS max is 10 .with_handler_timeout(Duration::from_secs(30)) .with_max_reconnect_attempts(5); ``` * **`with_receive_batch_size(u16)`** — messages requested per `ReceiveMessage` call. SQS's hard max is `10`; SQS returns batches up to this size. Default: `0` (= use the prefetch count). * **`with_handler_timeout(Duration)`** — per-message wall-clock deadline; must be less than the queue's `VisibilityTimeout` so a slow handler doesn't race the broker re-delivering the same message. Default: `30s`. * **`with_max_reconnect_attempts(u32)`** — cap on SQS client redials before surfacing a `Connection` error. Default: unlimited. `SqsConsumerGroupConfig` exposes the same knobs for the group registry (`with_prefetch_count`, `with_max_retries`, `with_handler_timeout`, `with_concurrent_processing`). #### Autoscaling For autoscaling within a `min..=max` range driven by `ApproximateNumberOfMessages`, use the backend-specific `SqsConsumerGroupRegistry` directly. It accepts an `SqsConsumerGroupConfig::new(min..=max)` and spawns/retires independent pollers within that range as queue depth changes. The registry also accepts a service-wide handler timeout default via `SqsConsumerGroupRegistry::with_default_handler_timeout(Duration)` — applied to every group that did not set its own. See the [Consumer Groups example](/backends/sqs/examples/consumer-groups) for the full path. ### Gotchas * **Visibility timeout must be greater than handler timeout.** If the handler takes longer than the visibility timeout, the message becomes re-visible on the queue and will be delivered again while the original handler is still running. Set both values consciously. * **FIFO 300 msg/s default cap.** High-throughput sequenced topics may require quota increases via the AWS console. Standard (non-FIFO) SQS queues have no ordering guarantee and much higher throughput limits. * **LocalStack endpoint is for local dev only.** The `SnsConfig::endpoint_url` field should be omitted in production; it exists solely to redirect traffic to a local emulator. * **Region matters.** SQS queues are regional. Cross-region fan-out requires explicit SNS topic ARN configuration — shove does not automatically route across regions. * **IAM permissions required:** `sns:CreateTopic`, `sns:Subscribe`, `sns:Publish`, `sqs:CreateQueue`, `sqs:ReceiveMessage`, `sqs:DeleteMessage`, `sqs:GetQueueAttributes`, and `sqs:SetQueueAttributes` at minimum. ### Examples * [Basic Pubsub](/backends/sqs/examples/basic) — all `Outcome` variants, hold queues, DLQ, batch publish * [Sequenced Pubsub](/backends/sqs/examples/sequenced) — FIFO topics + `MessageGroupId` ordering * [Concurrent Pubsub](/backends/sqs/examples/concurrent) — high-concurrency publish patterns * [Consumer Groups](/backends/sqs/examples/consumer-groups) — supervisor with autoscaling * [Audited Consumer](/backends/sqs/examples/audited) — `MessageHandlerExt::audited` wrapping * [Autoscaler](/backends/sqs/examples/autoscaler) — full autoscaler configuration walkthrough * [Stress](/backends/sqs/examples/stress) — throughput benchmarking ### See also * [Liveness Probes](/guides/liveness) — wire `Broker::ping` into a k8s health endpoint. ## SQS — Audited Consumer Use this when you want to confirm that trace IDs survive the SQS hold-queue round-trip. The same UUID appears on both the initial `Outcome::Retry` delivery and the successful second attempt — exactly the property you need when correlating retries in a payment audit log or compliance record. The `AuditHandler` API is identical across backends. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo features: `aws-sns-sqs,audit` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_audited_consumer --features aws-sns-sqs,audit ``` ### Expected output ```text topology declared published payment [handler] payment=PAY-001 amount=$49.99 attempt=1 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Retry", "duration_ms": 0, ... } [handler] payment=PAY-001 amount=$49.99 attempt=2 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Ack", "duration_ms": 0, ... } done ``` The same `trace_id` UUID appears on both delivery attempts. The 15-second run window allows the hold-queue TTL (5 s) to expire before the second delivery arrives. ### Source ```rust // [!include ~/examples/sqs/audited_consumer.rs] ``` ### Walkthrough #### Handler with deliberate retry `PaymentHandler` returns `Outcome::Retry` when `metadata.retry_count == 0` and `Outcome::Ack` on subsequent attempts. On SQS, `Outcome::Retry` extends the message's visibility timeout and routes it through the hold queue (5 s TTL), then redelivers it. The second delivery has `retry_count == 1`, so the handler acks. #### `StdoutAuditHandler` serialises the full record `StdoutAuditHandler` implements `AuditHandler\` and serialises the complete `AuditRecord\` to pretty JSON. The record carries `trace_id` — a UUID that the framework assigns once per original message and propagates across hold-queue round-trips via the SQS message attribute `shove-trace-id`. This is what allows the same `trace_id` to appear on both the retry and the final ack delivery. #### 15-second run window The supervisor runs for 15 seconds (longer than the 10-second RabbitMQ equivalent) to ensure the 5-second hold queue TTL fires and the retried message arrives before shutdown. On SQS, LocalStack's polling latency adds additional overhead beyond the bare queue TTL. #### Wiring: `.audited()` unchanged `PaymentHandler.audited(StdoutAuditHandler)` — the call is identical to the RabbitMQ version. `MessageHandlerExt::audited` is backend-agnostic: it wraps any `MessageHandler` with any `AuditHandler` into a combined `MessageHandler` that the supervisor accepts without modification. ### What to try next * Add `.with_max_retries(1)` to `ConsumerOptions` and make `PaymentHandler` always return `Retry` — observe the audit record show `outcome: "Reject"` when the retry limit is hit. * Replace `StdoutAuditHandler` with one that writes to a database or publishes to a second SQS topic. * Check the `duration_ms` field in the audit record and confirm it measures the handler's wall-clock time by adding a `tokio::time::sleep` inside `PaymentHandler::handle`. * See [Guides: Audit](/guides/audit) for the complete `AuditRecord` schema. ## SQS — Autoscaler Use this when you want to see queue-depth-based autoscaling on SQS end-to-end. Sixty tasks are published at once; with one slow consumer the queue fills faster than it drains, triggering scale-up to five consumers. The autoscaler then scales back once the queue is empty. Run this before production to tune the hysteresis and cooldown settings for your traffic pattern. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo feature: `aws-sns-sqs` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_autoscaler --features aws-sns-sqs ``` ### Expected output Consumer count climbs from 1 to 5 as the queue fills, then falls back as it drains. Exact timing depends on LocalStack latency and system load: ```text published 60 tasks consumer group started (min=1, max=5) autoscaler running — watching consumer count change [monitor] consumers=1 messages_ready=58 in_flight=2 [monitor] consumers=3 messages_ready=40 in_flight=15 [monitor] consumers=5 messages_ready=18 in_flight=25 [monitor] consumers=5 messages_ready=0 in_flight=5 [monitor] consumers=2 messages_ready=0 in_flight=0 [monitor] consumers=1 messages_ready=0 in_flight=0 ... shutting down… done ``` ### Source ```rust // [!include ~/examples/sqs/autoscaler.rs] ``` ### Walkthrough #### Consumer group registry `SqsConsumerGroupRegistry::new(client)` manages a pool of independent SQS poll workers. `registry.register` configures `WorkQueue` with `SqsConsumerGroupConfig::new(1..=5).with_prefetch_count(5)` — a capacity of `consumers × prefetch = 5–25` messages in-flight. `registry.start_all()` spawns the minimum number of consumers (1); the autoscaler adds workers as needed. #### `SqsAutoscalerBackend` and `AutoscalerConfig` `SqsAutoscalerBackend::autoscaler(stats_provider, registry, config)` wires the autoscaler to the registry. It polls `SqsQueueStatsProvider::get_queue_stats` on each `poll_interval` (2 s) and computes: * `capacity = active_consumers × prefetch_count` * scale up when `messages_ready > capacity × scale_up_multiplier` (1.5) * scale down when `messages_ready < capacity × scale_down_multiplier` (0.3) `hysteresis_duration` (4 s) requires the condition to hold across multiple polls before acting, preventing flapping caused by momentary spikes. `cooldown_duration` (8 s) enforces a minimum gap between consecutive scale actions for the same group. #### `SqsQueueStatsProvider` for dual monitoring The example creates two `SqsQueueStatsProvider` instances: one inside the autoscaler and one in the monitor loop. Both call `GetQueueAttributes` independently — the autoscaler uses the stats to make scaling decisions, while the monitor loop prints `messages_ready` and `messages_not_visible` (in-flight) so you can correlate queue depth with consumer count changes in the output. #### Shutdown sequence The autoscaler task runs on a `CancellationToken`. After the monitor loop completes, `shutdown.cancel()` stops the autoscaler, then `registry.shutdown_all()` drains all consumer workers, then `client.shutdown()` closes the AWS client connections. This ordering ensures the autoscaler does not attempt to scale the registry after its workers are already shutting down. ### What to try next * Raise `scale_up_multiplier` to 5.0 — the autoscaler waits for a much deeper backlog before adding consumers; observe the queue draining more slowly during the initial burst. * Set `hysteresis_duration` to `Duration::ZERO` — the autoscaler reacts on every poll cycle; watch for flapping between consecutive scaling decisions. * Increase `burst_size` to 200 — the drain phase becomes longer and the scale-down step is more visible in the monitor output. * See the [SQS consumer groups example](/backends/sqs/examples/consumer-groups) for the same registry without autoscaling. ## SQS — Basic Pubsub Use this as your starting point for SQS integration: it covers the three non-sequenced topology shapes (minimal queue, DLQ, escalating retry backoff), all three SQS-compatible `Outcome` variants, and all three publish variants against a LocalStack backend. The SQS counterpart to the [RabbitMQ basic example](/backends/rabbitmq/examples/basic) — same programming model, different infrastructure. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token (LocalStack Pro is required for SNS+SQS FIFO queues) * Cargo feature: `aws-sns-sqs` ```sh export LOCALSTACK_AUTH_TOKEN= ``` A LocalStack testcontainer is started automatically — no manual `docker run` needed. ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_basic_pubsub --features aws-sns-sqs ``` ### Expected output ```text topologies declared messages published [minimal] order=ORD-001 amount=$50.00 attempt=1 [minimal] order=ORD-002 amount=$99.00 attempt=1 [minimal] order=ORD-003 amount=$10.00 attempt=1 [minimal] order=ORD-004 amount=$25.00 attempt=1 [minimal] order=ORD-005 amount=$77.77 attempt=1 [dlq] rejecting order=ORD-001 → DLQ [dlq] dead-letter: order=ORD-001 reason=rejected deaths=1 [retry] order=ORD-001 attempt=1 [retry] → transient failure, will Retry [retry] order=ORD-001 attempt=2 [retry] → success on retry done ``` ### Source ```rust // [!include ~/examples/sqs/basic_pubsub.rs] ``` ### Walkthrough #### LocalStack setup The example reads `LOCALSTACK_AUTH_TOKEN` from the environment and exits with a clear message if it is missing. Dummy AWS credentials (`test`/`test`) are set via `std::env::set_var` before any AWS SDK calls — LocalStack accepts these without validation. The container exposes the LocalStack endpoint on port 4566, which is fed into `SnsConfig { endpoint_url: Some(endpoint) }`. #### `SnsConfig` and `Broker` `Broker::::new(SnsConfig { region, endpoint_url })` initialises the SNS/SQS client. Providing `endpoint_url` redirects all AWS API calls to LocalStack instead of the real AWS endpoints. In production you omit `endpoint_url` and rely on standard AWS credential/region resolution. #### Three topology shapes `MinimalOrder` has no DLQ — rejected messages are discarded. `DlqOrder` adds `.dlq()` so `Outcome::Reject` lands messages in a dead-letter queue; `RejectHandler` implements `handle_dead` to process those entries. `RetryOrder` chains `.hold_queue(5s)`, `.hold_queue(30s)`, and `.hold_queue(120s)` — on the SQS backend each stage maps to a separate SQS queue with a corresponding visibility timeout extension. `with_max_retries(3)` on the supervisor registration caps retry loops before DLQ routing. #### Consumer supervisor on SQS `broker.consumer_supervisor()` returns a `ConsumerSupervisor\`. Each `register` call spawns a long-poll SQS consumer for that topic. Unlike AMQP, SQS consumers poll `ReceiveMessage` in a loop — there is no broker-push delivery. The supervisor handles the polling loop, visibility timeout management, and graceful drain on shutdown. ### What to try next * Increase `with_max_retries(3)` to `with_max_retries(0)` — messages go directly to the DLQ on first failure. * Change `Outcome::Retry` to `Outcome::Reject` in `RetryHandler` — the message bypasses hold queues and goes straight to the DLQ. * Inspect the LocalStack management UI at `http://localhost:4566/_localstack/health` to see queues, message counts, and DLQ contents while the example runs. * See [Guides: Retries](/guides/retries) for how hold queues and max-retries interact on SQS. ## SQS — Concurrent Pubsub Use this when sizing the prefetch and concurrency configuration for SQS I/O-bound handlers. The example shows the throughput improvement from enabling concurrent processing, and highlights the SQS-specific constraint: `ReceiveMessage` returns at most 10 messages per call, so the maximum useful prefetch is lower than for AMQP backends. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo feature: `aws-sns-sqs` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_concurrent_pubsub --features aws-sns-sqs ``` ### Expected output ```text --- Sequential (prefetch_count=1) --- 20 messages, 100ms handler delay, prefetch=1 processed task 0 processed task 1 ... done: 2.3s (9 msg/s) --- Concurrent (prefetch_count=20) --- 20 messages, 100ms handler delay, prefetch=20 processed task 0 processed task 5 ... done: 0.3s (68 msg/s) --- Summary --- sequential: 2.3s (9 msg/s) concurrent: 0.3s (68 msg/s) speedup: 7.5x Messages are always acked in delivery order. Concurrent mode overlaps handler I/O within a single consumer. ``` Speedup will be lower than the RabbitMQ equivalent because SQS's `ReceiveMessage` returns at most 10 messages per call even when `prefetch=20`. ### Source ```rust // [!include ~/examples/sqs/concurrent_pubsub.rs] ``` ### Walkthrough #### Handler with notification signal `SlowTaskHandler` uses an `Arc` to wake up the waiting main task as each message completes. The `wait_for(target, timeout)` method loops, checking the atomic counter on each `notify_waiters()` signal. This avoids a busy-poll or a fixed sleep, and its 120-second timeout (longer than the RabbitMQ version's 60 s) accounts for SQS polling overhead. #### Sequential vs concurrent modes `run_round` registers a consumer with: ```rust,no_run ConsumerOptions::::new() .with_prefetch_count(prefetch) .with_concurrent_processing(concurrent) ``` For the sequential round `prefetch=1` and `concurrent=false`. For the concurrent round `prefetch=20` and `concurrent=true`. Enabling `concurrent_processing` causes the consumer to dispatch each handler invocation as a separate tokio task, overlapping the 100 ms sleeps rather than serialising them. #### No queue purge between rounds Unlike the RabbitMQ example, which calls the Management HTTP API to purge the queue between rounds, this example simply re-publishes a fresh batch. The first round's consumer drains the first batch fully before the supervisor stops; then a second batch is published for the concurrent round. This avoids SQS's strict 60-second rate limit on `PurgeQueue`. #### SQS prefetch and batch size SQS's `ReceiveMessage` API returns at most 10 messages per request regardless of the requested batch size. `prefetch_count=20` will therefore fetch two batches of 10 in parallel, giving an effective in-flight window of 20 — but each batch requires a separate API call. This is why throughput improvements taper off above `prefetch=10` on SQS compared to AMQP backends where larger prefetch counts have a flatter cost curve. ### What to try next * Set `prefetch=10` for the concurrent round — this matches SQS's natural batch size; further increases provide diminishing returns. * Increase `msg_count` to 100 — the sequential time grows proportionally while concurrent time stays nearly flat. * Set `handler_delay` to `Duration::ZERO` — sequential and concurrent perform identically because there is no I/O to overlap. * See the [SQS stress example](/backends/sqs/examples/stress) for higher-volume concurrency measurements. ## SQS — Consumer Groups Use this when you want to observe how a pool of SQS poll workers drains a queue over time. Because SQS has no broker-side coordinated-group primitive, the registry spawns independent consumers; the example shows how to monitor `messages_ready` and `messages_not_visible` as the backlog drains — useful for sizing your worker pool before adding the autoscaler. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo feature: `aws-sns-sqs` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_consumer_groups --features aws-sns-sqs ``` ### Expected output ```text published 50 tasks consumer group started (min_consumers=1) monitoring queue depth — watching backlog drain [monitor] messages_ready=48 in_flight=2 [monitor] messages_ready=38 in_flight=10 [monitor] messages_ready=10 in_flight=10 [monitor] messages_ready=0 in_flight=4 [monitor] messages_ready=0 in_flight=0 ... shutting down... done ``` Exact numbers vary by system speed and LocalStack latency. ### Source ```rust // [!include ~/examples/sqs/consumer_groups.rs] ``` ### Walkthrough #### SQS supervisors vs consumer groups SQS has no broker-side queue consumer-group primitive (unlike Kafka consumer groups or RabbitMQ coordinated groups). `SqsConsumerGroupRegistry` fills this role by spawning independent poll workers that each run their own `ReceiveMessage` loop. From the application's perspective the API mirrors the RabbitMQ `ConsumerGroupRegistry`: `registry.register(config, factory, ctx)` followed by `registry.start_all()`. The key difference is that each registered consumer is fully independent — there is no broker-side coordination when adding or removing workers. #### `SqsConsumerGroupConfig` and worker range `SqsConsumerGroupConfig::new(1..=5)` sets the minimum to 1 and the maximum to 5 consumers. `.with_prefetch_count(10)` lets each consumer hold up to 10 in-flight messages (capped by SQS's batch size limit of 10). `.with_max_retries(3)` caps hold-queue retry cycles before DLQ routing. `registry.start_all()` starts one consumer (the minimum); autoscaling would add more, but this example keeps the count fixed to focus on monitoring. #### Shared SNS client and registries `SnsClient::new(&config)` returns a client that owns the topic and queue registries. `SnsTopologyDeclarer::new(client.clone())` declares the `WorkQueue` topology and populates these registries. `SnsPublisher::new(client.clone(), client.topic_registry().clone())` and `SqsConsumerGroupRegistry::new(client.clone())` both share the same underlying registry, so the publisher can resolve the SNS topic ARN and the consumers can resolve the SQS queue URL without a second declaration step. #### Queue depth monitoring with `SqsQueueStatsProvider` `SqsQueueStatsProvider::new(client, queue_registry)` wraps the `GetQueueAttributes` API. `stats_provider.get_queue_stats(queue_name)` returns `messages_ready` (visible messages waiting to be consumed) and `messages_not_visible` (in-flight messages currently held by a consumer). Polling these every 2 seconds gives a live view of the drain rate — a production monitoring setup would feed these values to a metrics system and drive autoscaling decisions. ### What to try next * Wire `SqsAutoscalerBackend` to the registry (see the [autoscaler example](/backends/sqs/examples/autoscaler)) to scale the consumer count automatically based on `messages_ready`. * Reduce the handler delay to 20 ms — observe how quickly `messages_ready` drops to 0 with even one consumer. * Call `registry.lock().await.add_consumers::(2)` after the initial burst to manually scale up mid-flight. * See [Guides: Groups](/guides/groups) for an explanation of how independent SQS workers relate to coordinated consumer groups on other backends. ## SQS — Sequenced Pubsub Use this when you need to verify sequenced delivery semantics on SQS: `SequenceFailure::Skip` lets a sequence continue after one bad message; `SequenceFailure::FailAll` quarantines the rest. The example asserts exact totals for both policies, giving you a concrete end-to-end test you can run against LocalStack before deploying to AWS. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo feature: `aws-sns-sqs` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_sequenced_pubsub --features aws-sns-sqs ``` ### Expected output ```text sequenced topologies declared published 5 entries per topic for ACC-A published 3 entries per topic for ACC-B [skip error and continue] account=ACC-A seq=1 amount=100 attempt=1 ... [skip error and continue] account=ACC-A seq=3 amount=300 attempt=1 [skip error and continue] → Reject (seq=3 is poison for ACC-A) ... ── Results (elapsed: 15.3s) ── skip ACC-A = 1200 (expected 1200) skip ACC-B = 300 (expected 300) strict ACC-A = 300 (expected 300) strict ACC-B = 300 (expected 300) ``` ### Source ```rust // [!include ~/examples/sqs/sequenced_pubsub.rs] ``` ### Walkthrough #### SQS client and registry sharing Rather than using the `Broker\` facade, the example uses `SnsClient::new(&config)` directly to obtain a shareable client handle. `SnsTopologyDeclarer::new(client.clone())` declares queues against the same client instance that the publisher and consumers will use. This is necessary because the declarer populates the topic and queue registries (`client.topic_registry()`, `client.queue_registry()`) that `SnsPublisher` and `SqsConsumer` read at runtime to resolve ARNs and queue URLs. #### `SqsConsumer::run_fifo` for per-key FIFO Per-key FIFO is not yet available through the generic `Broker\` API, so the example uses `SqsConsumer::new(client, queue_registry).run_fifo::(handler, ctx, opts)`. Both topics use `.routing_shards(2)` — smaller than the RabbitMQ equivalent (which defaults to 8) because LocalStack's overhead per FIFO queue is higher and two shards are sufficient for two accounts. `with_prefetch_count(8)` allows up to 8 messages to be in-flight per consumer, but only one per key at a time. #### Longer processing window The shutdown timer is 15 seconds (vs 5 seconds in the RabbitMQ equivalent) to account for SQS visibility timeout behaviour. When a message is retried on SQS, the hold queue is implemented as a separate SQS queue with a matching visibility timeout extension — the round-trip to LocalStack adds latency that requires a longer window. #### Totals verification The same assertion pattern used in the RabbitMQ sequenced example applies here: both consumers accumulate per-account totals in `Arc>` and the main task asserts exact values after shutdown. The expected values are identical: `skip ACC-A = 1200`, `strict ACC-A = 300`, both `ACC-B = 300`. ### What to try next * Swap `SequenceFailure::FailAll` for `SequenceFailure::Skip` on `StrictLedger` — confirm the `strict ACC-A` total rises from 300 to 1200. * Change `.routing_shards(2)` to `.routing_shards(4)` — more FIFO queues improve parallelism for workloads with many distinct keys. * Add a third account to the publish loop to verify cross-key independence. * Read [Guides: Sequenced](/guides/sequenced) for how shards, FIFO ordering, and failure modes interact. ## SQS — Stress Use this to measure SQS throughput on your workload profile and compare it against the [InMemory baseline](/backends/inmemory/examples/stress) and [RabbitMQ results](/backends/rabbitmq/examples/stress) to quantify the HTTP round-trip cost. Each consumer is an independent poll worker, reflecting SQS's architecture — the numbers here set realistic expectations before sizing a production SQS deployment. ### Prerequisites * Docker with a running daemon * `LOCALSTACK_AUTH_TOKEN` environment variable set to a valid LocalStack Pro token * Cargo feature: `aws-sns-sqs` ### Run ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_stress --features aws-sns-sqs ``` Narrow to one tier or handler profile: ```sh LOCALSTACK_AUTH_TOKEN= cargo run --example sqs_stress --features aws-sns-sqs -- --tier moderate --handler fast ``` Release mode for representative numbers: ```sh LOCALSTACK_AUTH_TOKEN= cargo run -q --release --example sqs_stress --features aws-sns-sqs ``` ### Expected output Non-deterministic. Look for these characteristic markers: ```text shove stress benchmarks — sqs scenarios: 60 [1/60] moderate | 20000msg | 1c | fast (1-5ms) ... -> 450.2 msg/s | dispatch p50=3.1ms p99=12.8ms | e2e p50=5.2ms p99=18.4ms | cpu=30% rss=45.2MB | 44.4s ... Backend: sqs TIER MSGS C HANDLER MSG/SEC ... moderate 20000 1 fast 450 ... moderate 20000 4 fast 1600 ... ... ``` SQS throughput is significantly lower than RabbitMQ or NATS because every `ReceiveMessage`, `DeleteMessage`, and hold-queue publish is an independent HTTP round-trip to LocalStack. ### Source ```rust // [!include ~/examples/sqs/stress.rs] ``` ### Walkthrough #### `run_supervisor_scenarios` instead of `run_all_scenarios` The SQS stress binary calls `run_supervisor_scenarios` from the shared harness, not `run_all_scenarios`. The distinction is architectural: `run_all_scenarios` drives a `ConsumerGroup\` (a coordinated group backed by a `HasCoordinatedGroups` capability), while `run_supervisor_scenarios` drives independent `ConsumerOptions\` workers registered one by one on a `ConsumerSupervisor\`. SQS does not implement `HasCoordinatedGroups`, so the supervisor path is the only option. #### Prefetch cap at 10 `HarnessConfig::::new("sqs").with_prefetch_cap(SQS_PREFETCH_CAP)` sets the maximum prefetch to 10, matching SQS's `ReceiveMessage` batch limit. The harness computes a default prefetch as `(messages / consumers).clamp(1, cap)`, so any scenario with more than 10 messages per consumer will hit the cap. This is why SQS throughput does not scale linearly with prefetch the way AMQP-backed backends do. #### Queue purge between scenarios `with_purge(purge)` injects a purge closure that calls `PurgeQueue` on the main queue, the DLQ, and the FIFO shard queue between scenarios. SQS enforces a 60-second rate limit on `PurgeQueue`, so the closure ignores errors — if purge fails, the next scenario's messages are published on top of leftovers, but the harness still counts correctly once `scenario.messages` are processed. This is a deliberate trade-off: deleting and recreating queues would trigger SQS's 60-second queue-name reuse restriction, which is worse. #### `SnsConfig` for LocalStack `SnsConfig { region: Some("us-east-1".into()), endpoint_url: Some(endpoint) }` redirects all AWS SDK calls to the LocalStack container. In production remove `endpoint_url` and rely on the AWS SDK's standard credential and region resolution. ### What to try next * Compare `--tier moderate --handler zero` throughput here against the RabbitMQ and InMemory stress results to quantify HTTP vs AMQP vs in-process costs. * Add `--concurrent` for the `slow` handler profile — I/O overlap within each consumer reduces the impact of SQS's per-request latency. * Use `--output json` and import results into a spreadsheet to plot scaling curves across consumer counts. * See the [SQS autoscaler example](/backends/sqs/examples/autoscaler) for how to autoscale based on the queue depth metrics this harness measures. ## Redis — Audited Consumer Wraps a payment handler in `MessageHandlerExt::audited` and pipes audit records through a simple stdout `AuditHandler` impl. Every delivery emits one structured audit line with the payload, outcome, duration, and trace ID — exactly what the [Audit Logging guide](/guides/audit) describes, applied to the Redis backend. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo features: `redis-streams`, `audit` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_audited_consumer --features "redis-streams audit" ``` ### Expected output ```text topology declared published payment [handler] processing payment PAY-0001 [audit] {"topic":"payments","outcome":"ack","duration_ms":1,...} done ``` ### Source ```rust // [!include ~/examples/redis/audited_consumer.rs] ``` ## Redis — Autoscaler Wires the Redis backend into the autoscaler. A consumer group with `min=1, max=5` starts with a single worker; the autoscaler polls the group's consumer lag (messages awaiting delivery, as reported by Redis — already-processed history does not count as backlog) and spawns workers up to `max` while a burst of forty messages is pending, then drains back down to `min` as the backlog clears. The driver thread prints periodic stats so you can watch the scale decisions. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo feature: `redis-streams` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_autoscaler --features redis-streams ``` ### Expected output ```text consumer group started (min=1, max=5) published 40 work items [worker] handling id=0 [worker] handling id=1 [monitor] active_consumers=3 pending=37 lag=37 [worker] handling id=2 ... [monitor] active_consumers=5 pending=12 lag=12 ... [monitor] active_consumers=1 pending=0 lag=0 shutting down… ``` ### Source ```rust // [!include ~/examples/redis/autoscaler.rs] ``` ## Redis — Basic A minimal Redis Streams publish/consume round-trip. Three `OrderPaid` messages are published to a topic with a hold queue and DLQ, then consumed by a single-worker consumer group until Ctrl-C. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo feature: `redis-streams` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_basic --features redis-streams ``` ### Expected output ```text published ORD-0001 published ORD-0002 published ORD-0003 consuming — press Ctrl-C to stop received order: ORD-0001 (retry=0) received order: ORD-0002 (retry=0) received order: ORD-0003 (retry=0) ``` ### Source ```rust // [!include ~/examples/redis/basic.rs] ``` ## Redis — Consumer Groups Registers a single coordinated consumer group with `min_consumers=2, max_consumers=4`, publishes a batch of twenty work items, and lets the group drain them on Ctrl-C (or a 30 s timeout). Demonstrates the `RedisConsumerGroupConfig` builder shape that mirrors every other coordinated-group backend in shove. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo feature: `redis-streams` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_consumer_groups --features redis-streams ``` ### Expected output ```text published 20 work items consuming with 2 initial consumers — Ctrl-C to stop [worker] id=0 payload=item-0 [worker] id=1 payload=item-1 ... [worker] id=19 payload=item-19 shutting down… ``` ### Source ```rust // [!include ~/examples/redis/consumer_groups.rs] ``` ## Redis — Sequenced Demonstrates per-key ordering with `SequencedTopic` and `register_fifo`. Interleaved ledger entries for three accounts are published; entries for the same account are always consumed in strict publish order, regardless of how many shard consumer tasks run. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo feature: `redis-streams` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_sequenced --features redis-streams ``` ### Expected output ```text published account=acct-001 seq=000 published account=acct-002 seq=000 published account=acct-001 seq=001 published account=acct-003 seq=000 published account=acct-002 seq=001 published account=acct-001 seq=002 published account=acct-003 seq=001 consuming — press Ctrl-C to stop account=acct-001 seq=000 amount=+1000 account=acct-002 seq=000 amount=+2000 account=acct-001 seq=001 amount=-500 account=acct-003 seq=000 amount=+750 account=acct-002 seq=001 amount=-200 account=acct-001 seq=002 amount=+300 account=acct-003 seq=001 amount=-100 ``` ### Source ```rust // [!include ~/examples/redis/sequenced.rs] ``` ## Redis — Stress Throughput benchmark for the Redis backend. Publishes a configurable burst of messages on a non-sequenced topic and consumes them through a coordinated consumer group with concurrent processing. Useful for sanity-checking client tuning (`prefetch_count`, `max_consumers`, `response_timeout`) against your Redis or Valkey deployment before rolling production traffic. ### Prerequisites * Redis 6.2+ or Valkey running locally (see command below) * Cargo feature: `redis-streams` ### Run ```sh docker run --rm -p 6379:6379 redis:7-alpine cargo run --example redis_stress --features redis-streams --release ``` Pass `--release` — the JSON codec and XADD batching show their real cost only with optimisations on. ### Source ```rust // [!include ~/examples/redis/stress.rs] ``` ## RabbitMQ — Audited Consumer Use this when you want to verify that trace IDs are preserved across retries on RabbitMQ. The handler returns `Outcome::Retry` on the first delivery; both delivery attempts emit an audit record with the same UUID — proving that trace identity survives the hold-queue round-trip. The pattern directly applies to payment processing, compliance logging, and any workflow where correlating retries to the original delivery matters. ### Prerequisites * Docker (a RabbitMQ testcontainer is spun up automatically) * Cargo features: `rabbitmq,audit` ### Run ```sh cargo run --example rabbitmq_audited_consumer --features rabbitmq,audit ``` ### Expected output ```text topology declared published payment [handler] payment=PAY-001 amount=$49.99 attempt=1 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Retry", "duration_ms": 0, ... } [handler] payment=PAY-001 amount=$49.99 attempt=2 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Ack", "duration_ms": 0, ... } done ``` The `trace_id` is the same UUID on both delivery attempts. ### Source ```rust // [!include ~/examples/rabbitmq/audited_consumer.rs] ``` ### Walkthrough #### Handler with deliberate retry `PaymentHandler` inspects `metadata.retry_count`: when it is 0 (first delivery) it returns `Outcome::Retry`, simulating a transient failure. On the second delivery (`retry_count >= 1`) it returns `Outcome::Ack`. This two-attempt lifecycle is the minimal setup needed to observe trace-ID continuity across a retry cycle. #### `StdoutAuditHandler` and `AuditRecord` `StdoutAuditHandler` implements `AuditHandler\` and serialises the full `AuditRecord\` to pretty-printed JSON. The record includes `trace_id` (a UUID assigned when the message is first dispatched and carried through retries via AMQP headers), `outcome` (the value returned by the inner handler), `duration_ms` (wall-clock time the inner handler spent), and a copy of the original message. Printing the whole record as JSON makes it easy to inspect every field during development; swap in a DB writer or `ShoveAuditHandler` for production. #### Wiring with `.audited()` The call `PaymentHandler.audited(StdoutAuditHandler)` comes from `MessageHandlerExt`, which is blanket-implemented for all `MessageHandler` types when the `audit` feature is enabled. The resulting combinator satisfies `MessageHandler\`, so `supervisor.register` sees a single opaque handler — no API changes needed to add auditing. #### Supervisor with `max_retries` `ConsumerOptions::::new().with_max_retries(3)` ensures the consumer will retry up to three times before sending a message to the DLQ. In this example only one retry is needed, but setting a limit guards against runaway retry loops if the handler logic changes. ### What to try next * Remove `.with_max_retries(3)` and change `PaymentHandler` to always `Retry` — the message loops until `max_retries` is reached (broker default), then lands in the DLQ; verify the audit record's `outcome` changes to `"Reject"` on the final delivery. * Replace `StdoutAuditHandler` with a struct that publishes to a second topic — this is the `ShoveAuditHandler` pattern for production audit pipelines. * Add a second payment with a different `payment_id` and observe that each gets a distinct `trace_id`. * See [Guides: Audit](/guides/audit) for the full `AuditRecord` schema and production wiring patterns. ## RabbitMQ — Basic Pubsub Use this as your starting point for RabbitMQ integration: it covers every non-sequenced topology shape (minimal queue, DLQ, escalating retry backoff, deferred delivery), all four `Outcome` variants, and all three publish variants against a real broker. The canonical reference for how [topic definitions](/concepts/topics) and [handler outcomes](/concepts/outcomes) map to RabbitMQ. ### Prerequisites * Docker (a RabbitMQ testcontainer is spun up automatically — no manual `docker run` needed) * Cargo feature: `rabbitmq` ### Run ```sh cargo run --example rabbitmq_basic_pubsub --features rabbitmq ``` ### Expected output ```text topologies declared messages published [minimal] order=ORD-001 amount=$50.00 attempt=1 [minimal] order=ORD-002 amount=$99.00 attempt=1 [minimal] order=ORD-003 amount=$10.00 attempt=1 [minimal] order=ORD-004 amount=$25.00 attempt=1 [minimal] order=ORD-005 amount=$77.77 attempt=1 [dlq] rejecting order=ORD-001 → DLQ [dlq] dead-letter: order=ORD-001 reason=rejected deaths=1 [retry] order=ORD-001 attempt=1 [retry] → transient failure, will Retry [retry] order=ORD-001 attempt=2 [retry] → success on retry [defer] order=ORD-001 attempt=1 redelivered=false [defer] → not ready yet, deferring to hold queue [defer] order=ORD-001 attempt=1 redelivered=true [defer] → processing now done ``` ### Source ```rust // [!include ~/examples/rabbitmq/basic_pubsub.rs] ``` ### Walkthrough #### Four topic shapes with `define_topic!` The example defines four topics using two different approaches. `MinimalOrder`, `DlqOrder`, and `RetryOrder` all use the `define_topic!` macro — a shorthand that generates the `Topic` impl, the `OnceLock` topology cell, and a zero-sized struct in one call. `ScheduledOrder` is declared manually to show the `OnceLock` pattern explicitly; this form is needed when the topic needs to use `Outcome::Defer` (which requires the hold queue TTL to be read from the topology). The `RetryOrder` topology chains three `.hold_queue()` calls to create escalating backoff stages: 5 s → 30 s → 120 s. #### All four `Outcome` variants in action * `AckHandler` always returns `Outcome::Ack` — the message is removed from the queue. * `RejectHandler` returns `Outcome::Reject` — the message is immediately routed to the DLQ without using up a retry slot. It also implements `handle_dead` to process messages that arrive on the DLQ itself. * `RetryHandler` checks `metadata.retry_count`: on the first attempt it returns `Outcome::Retry`, which sends the message through the hold queue and back. On the second attempt it acks. * `DeferHandler` returns `Outcome::Defer` on first delivery when `metadata.redelivered == false`. Unlike `Retry`, `Defer` routes through the hold queue without incrementing the retry counter — the message comes back as if it were a fresh delivery. #### Consumer supervisor `broker.consumer_supervisor()` creates a `ConsumerSupervisor\`. Each `supervisor.register(handler, options)` call spawns one AMQP channel/consumer for that topic. The supervisor owns a shared shutdown token, so `run_until_timeout` stops every registered consumer in lock-step and drains in-flight messages before returning. `with_max_retries(3)` on `RetryOrder` ensures messages that exhaust their retries are sent to the DLQ rather than looping indefinitely. #### Publish variants `publisher.publish::(&msg)` sends a single message. `publish_with_headers::(&msg, headers)` attaches custom AMQP headers (the example uses `x-source` and `x-priority`). `publish_batch::(&[msg])` sends multiple messages in a single channel operation — more efficient for bursts. All three variants share the same type-checked routing; the compiler rejects publishing the wrong message type for a given topic. ### What to try next * Change `Outcome::Retry` to `Outcome::Defer` in `RetryHandler` — notice that `metadata.retry_count` no longer increments on re-delivery. * Add a fourth `.hold_queue(Duration::from_secs(300))` to `RetryOrder` — messages get a fourth retry stage at 5 minutes. * Remove `.dlq()` from `MinimalOrder` — rejected messages are silently discarded (you will see a warning log from the framework). * See [Guides: Retries](/guides/retries) for a full explanation of hold queues and retry limits. ## RabbitMQ — Concurrent Pubsub Use this when deciding whether to enable concurrent processing for I/O-bound handlers. A 100 ms handler processes 20 messages sequentially in \~2 s; the same handler with `with_concurrent_processing` processes them in \~0.2 s — a 10x speedup visible in the printed summary. The example also confirms that acknowledgements remain in-order regardless of which mode is active. ### Prerequisites * Docker (a RabbitMQ testcontainer is spun up automatically) * Cargo feature: `rabbitmq` ### Run ```sh cargo run --example rabbitmq_concurrent_pubsub --features rabbitmq ``` ### Expected output ```text --- Sequential (prefetch_count=1) --- 20 messages, 100ms handler delay, prefetch=1 processed task 0 processed task 1 ... done: 2.1s (10 msg/s) --- Concurrent (prefetch_count=20) --- 20 messages, 100ms handler delay, prefetch=20 processed task 0 processed task 3 ... done: 0.2s (107 msg/s) --- Summary --- sequential: 2.1s (10 msg/s) concurrent: 0.2s (107 msg/s) speedup: 10.1x Messages are always acked in delivery order. Concurrent mode overlaps handler I/O within a single consumer. ``` Exact numbers vary by system; expect a speedup close to the prefetch count (20x theoretical). ### Source ```rust // [!include ~/examples/rabbitmq/concurrent_pubsub.rs] ``` ### Walkthrough #### Handler design `SlowTaskHandler` wraps an `Arc` counter and an `Arc` signal. The handler sleeps for `self.delay` (100 ms) to simulate an I/O operation, then increments the counter and notifies any waiters. The `wait_for(target, timeout)` method polls the counter and wakes on `Notify`, letting the main task stop the supervisor as soon as the last message is processed rather than waiting on a fixed timer. #### Sequential vs concurrent modes The example calls `run_round` twice. For the sequential round: `prefetch=1` (at most one message in-flight) and `concurrent=false`. For the concurrent round: `prefetch=20` and `concurrent=true`. The key option is: ```rust,no_run ConsumerOptions::::new() .with_concurrent_processing(concurrent) ``` When `true`, the consumer spawns each handler invocation as a separate tokio task, allowing up to `prefetch` handlers to run simultaneously. Acknowledgements are still sent in delivery order, so the broker's message ordering guarantees are maintained. #### Queue management between rounds `purge_queue` calls the RabbitMQ Management HTTP API (`DELETE /api/queues/%2F/{name}/contents`) between the two rounds to ensure the second round starts from a clean slate. The hold queue (named `{queue}-hold-5s`) is also purged to avoid stale retries bleeding into the concurrent run. #### Speedup calculation After both rounds complete, the example computes `speedup = sequential_duration / concurrent_duration`. With 20 messages at 100 ms each and prefetch 20, theoretical peak is 20x; real numbers are slightly lower due to tokio scheduling overhead. `assert!(outcome.is_clean())` verifies the supervisor drained cleanly (no error exits) after each round. ### What to try next * Lower `prefetch` to 5 while keeping `concurrent=true` — the speedup should converge toward 5x since at most 5 handlers run at once. * Increase `msg_count` to 200 — the sequential time grows linearly while concurrent time stays roughly constant until the prefetch window fills. * Set `handler_delay` to 0 ms — the speedup collapses to \~1x because there is no I/O to overlap; sequential and concurrent modes perform identically for CPU-bound handlers. * See [Concepts: Handlers](/concepts/handlers) for the in-order acknowledgement guarantee and how concurrent mode interacts with prefetch. ## RabbitMQ — Consumer Groups Use this when you want to see the RabbitMQ autoscaler in action: start at one consumer, watch the count climb as queue depth rises, and watch it fall once the queue drains. The example uses the one-line `enable_autoscaling` API, showing how `shove` adapts consumer count to actual load with the autoscaler lifecycle folded into `run_until_timeout`. ### Prerequisites * Docker (a RabbitMQ testcontainer with the management plugin is spun up automatically) * Cargo feature: `rabbitmq` ### Run ```sh cargo run --example rabbitmq_consumer_groups --features rabbitmq ``` ### Expected output Consumer count climbs from 1 toward 5 as the queue fills, then falls back after the queue drains. With `RUST_LOG=shove=info` the autoscaler's scale decisions are logged as `scaled up: spawned new consumer` / `scaled down: cancelled an idle consumer`: ```text published 100 tasks; starting group with autoscaling (min=1, max=5) [worker] task=TASK-000 attempt=1 INFO shove: scaled up: spawned new consumer group=ex-work-queue consumers=2 INFO shove: scaled up: spawned new consumer group=ex-work-queue consumers=5 ... INFO shove: scaled down: cancelled an idle consumer group=ex-work-queue consumers=1 shutdown complete: SupervisorOutcome { errors: 0, panics: 0, timed_out: false } ``` Exact counts and timing depend on system load and the autoscaler's poll interval. ### Source ```rust // [!include ~/examples/rabbitmq/consumer_groups.rs] ``` ### Walkthrough #### Group and `ConsumerGroupConfig` `broker.consumer_group()` returns a coordinated group registry. `group.register::(config, factory)` wires `TaskHandler` to `WorkQueue` with `RabbitMqConsumerGroupConfig::new(1..=5)` — a range from one minimum to five maximum consumers. `.with_prefetch_count(10)` lets each consumer hold up to 10 messages at a time, giving a capacity of `consumers × 10` at any given scale. #### Enabling autoscaling `group.enable_autoscaling(auto)` switches the autoscaler on. It polls the RabbitMQ Management HTTP API every `poll_interval` (2 s). The scaling logic is pure arithmetic: scale up when `messages_ready > capacity × scale_up_multiplier` (1.5), scale down when `messages_ready < capacity × scale_down_multiplier` (0.3). `hysteresis_duration` (4 s) prevents flapping by requiring the condition to hold before acting; `cooldown_duration` (8 s) prevents consecutive scale events from interfering. #### Management API configuration The autoscaler reads its management credentials from the client, so the `RabbitMqConfig` is built with `.with_management(ManagementConfig::new(url, username, password))`, pointing at the RabbitMQ Management API on port 15672. The management plugin is enabled by default in the `testcontainers-modules` RabbitMQ image. In production this URL and credentials would come from environment variables. #### Lifecycle and shutdown `run_until_timeout(shutdown_signal(), drain_timeout)` owns the whole lifecycle: it starts the consumers, spawns the autoscaler against the group's own registry, and — when `shutdown_signal` fires — stops and joins the autoscaler before draining the consumers. The returned `SupervisorOutcome` carries the error/panic/timeout tally; `std::process::exit(outcome.exit_code())` maps it to a process exit code. ### What to try next * Lower the handler delay to 50 ms — the queue drains faster and the autoscaler may not have time to scale up before it triggers scale-down. * Raise `scale_up_multiplier` to 5.0 — the autoscaler requires a much deeper backlog before adding consumers. * Set `min..=max` to `3..=3` — pins the pool at 3 workers; with min == max the autoscaler has no room to move. * See [Guides: Groups](/guides/groups) for an explanation of how consumer groups distribute load across workers. ## RabbitMQ — Exactly-Once Use this when you need to eliminate duplicate deliveries for handlers with irreversible side effects — payment charges, outbound notifications, external API calls without idempotency tokens. The example shows how a single `with_exactly_once()` call wraps every routing decision in an AMQP transaction, covering all three routing paths: ack, retry-then-ack, and reject-to-DLQ. ### Prerequisites * Docker (a RabbitMQ testcontainer is spun up automatically) * Cargo feature: `rabbitmq-transactional` ### Run ```sh cargo run --example rabbitmq_exactly_once --features rabbitmq-transactional ``` ### Expected output ```text topologies declared messages published [ack] payment=PAY-001 amount=$10.00 (processed: 1) [ack] payment=PAY-002 amount=$20.00 (processed: 2) [ack] payment=PAY-003 amount=$30.00 (processed: 3) [ack] payment=PAY-004 amount=$40.00 (processed: 4) [ack] payment=PAY-005 amount=$50.00 (processed: 5) [retry] payment=PAY-RETRY attempt=1 retry_count=0 [retry] → simulated transient failure, retrying… [retry] payment=PAY-RETRY attempt=2 retry_count=1 [retry] → succeeded on retry [reject] payment=PAY-BAD-1 → routing to DLQ [dlq] payment=PAY-BAD-1 reason=rejected deaths=1 (dlq total: 1) [reject] payment=PAY-BAD-2 → routing to DLQ [dlq] payment=PAY-BAD-2 reason=rejected deaths=1 (dlq total: 2) [reject] payment=PAY-BAD-3 → routing to DLQ [dlq] payment=PAY-BAD-3 reason=rejected deaths=1 (dlq total: 3) done ``` ### Source ```rust // [!include ~/examples/rabbitmq/exactly_once.rs] ``` ### Walkthrough #### What exactly-once means here AMQP's `tx_select` / `tx_commit` mode makes every channel operation part of an atomic transaction. In confirm-mode (the default), publishing to the hold queue and acking the original delivery are two separate network round-trips — a broker restart between them can cause the original message to be redelivered without the hold-queue copy. Transactional mode collapses both into a single commit that either fully succeeds or fully rolls back. The observable behaviour is identical; the guarantee is stronger at the cost of roughly 10–15x lower throughput per channel. #### Three topics, three outcomes `PaymentTopic` receives five payments that `AckHandler` acks immediately — the simplest case. `RetryPaymentTopic` receives one payment that `RetryThenAckHandler` retries once via its hold queue (TTL 2 s) before acking on the second delivery. `RejectPaymentTopic` receives three payments that `RejectHandler` sends straight to the DLQ via `Outcome::Reject`; `handle_dead` then processes each dead-letter and prints the `death_count`. #### Enabling transactional mode `ConsumerOptions::::new().with_exactly_once()` is the only API change. The framework swaps the confirm-mode channel for a transactional channel and wraps the ack/nack/publish sequence in `tx_select → ... → tx_commit`. The `PaymentEvent` message type, the `Outcome` variants, and the topology declarations are all unchanged — exactly-once is an opt-in delivery guarantee, not a different programming model. #### Hold queue interaction `RetryPaymentTopic` uses a 2-second hold queue. Under transactional mode, publishing to the hold queue and acking the original delivery are committed atomically, so the consumer will not re-receive a partial state even if the broker restarts mid-commit. The example's 8-second run window is chosen to ensure the 2-second TTL fires and the retry delivery arrives before shutdown. ### What to try next * Remove `.with_exactly_once()` from one consumer and use `RUST_LOG=shove=debug` to compare how confirm-mode and transactional-mode channel lifecycles differ in the log output. * Raise the hold queue TTL to 30 s — the retry consumer will not see the second delivery within the 8-second window, and the message stays in the hold queue. * Benchmark throughput with and without `with_exactly_once()` — the \~10–15x slowdown becomes visible with hundreds of messages. * See [Guides: Exactly-once](/guides/exactly-once) for the full explanation of the transactional path and when to use it. ## RabbitMQ — Sequenced Pubsub Use this when you need to decide between `SequenceFailure::Skip` and `SequenceFailure::FailAll` for your ordering policy. Ledger entries for two accounts are published, one entry is deliberately rejected, and the example asserts the resulting totals — making the behavioral difference between the two policies concrete and verifiable against a real RabbitMQ broker. ### Prerequisites * Docker (a RabbitMQ testcontainer is spun up automatically with the `rabbitmq_consistent_hash_exchange` plugin enabled) * Cargo feature: `rabbitmq` ### Run ```sh cargo run --example rabbitmq_sequenced_pubsub --features rabbitmq ``` ### Expected output ```text sequenced topologies declared published 5 entries per topic for ACC-A published 3 entries per topic for ACC-B [skip error and continue] account=ACC-A seq=1 amount=100 attempt=1 [skip error and continue] account=ACC-B seq=1 amount=50 attempt=1 ... [skip error and continue] account=ACC-A seq=3 amount=300 attempt=1 [skip error and continue] → Reject (seq=3 is poison for ACC-A) ... [fail all after error] account=ACC-A seq=3 amount=300 attempt=1 [fail all after error] → Reject (FailAll: seq=3 + all subsequent for ACC-A will DLQ) ... ── Results (elapsed: 5.2s) ── skip ACC-A = 1200 (expected 1200) skip ACC-B = 300 (expected 300) strict ACC-A = 300 (expected 300) strict ACC-B = 300 (expected 300) ``` ### Source ```rust // [!include ~/examples/rabbitmq/sequenced_pubsub.rs] ``` ### Walkthrough #### Two topics, two failure policies `SkipLedger` uses `SequenceFailure::Skip`: when `ACC-A seq=3` is rejected, it goes to the DLQ and consumption of `ACC-A seq=4` and `seq=5` resumes normally. The total for `ACC-A` is `100+200+400+500 = 1200`. `StrictLedger` uses `SequenceFailure::FailAll`: the same rejection causes `seq=4` and `seq=5` to be automatically DLQ'd without reaching the handler. The total for `ACC-A` drops to `100+200 = 300`. In both cases `ACC-B` is unaffected — keys are independent. #### `define_sequenced_topic!` and routing shards Both topics are created with `define_sequenced_topic!(Name, MsgType, key_fn, topology)`. The key function `|msg| msg.account_id.clone()` is captured inline and used by the broker to route each `LedgerEntry` to the correct FIFO shard. `StrictLedger` adds `.routing_shards(4)` to reduce the default of 8 shards — useful when the number of distinct keys is small and you want to pack them into fewer exchanges. The consistent-hash exchange plugin (enabled at container start) is required for shard routing to work on RabbitMQ. #### Backend-specific `run_fifo` Per-key FIFO is not yet available through the generic `Broker\` API, so the example uses `RabbitMqConsumer::new(client).run_fifo::(handler, ctx, opts)` directly. `with_prefetch_count(8)` means up to 8 messages may be in-flight at once, but only one per key at a time — the consumer holds back messages for a key while the current one is being processed. #### Assertion-based verification After a 5-second processing window the example reads the `totals` maps collected by `LedgerHandler` and asserts exact values. This pattern (accumulate totals in a shared `Arc>`, assert after shutdown) is the cleanest way to write an integration test that also runs as a runnable example. ### What to try next * Swap `SequenceFailure::FailAll` for `SequenceFailure::Skip` on `StrictLedger` — the `ACC-A` total should rise from 300 to 1200. * Increase `routing_shards` to 16 on `SkipLedger` — measure whether latency changes for two-key workloads. * Add a third account `ACC-C` to the publish loop with no poison entry — confirm it processes all entries regardless of which policy is active. * Read [Guides: Sequenced](/guides/sequenced) for the full explanation of shard routing and cross-key concurrency. ## RabbitMQ — Stress Use this to measure RabbitMQ throughput on your own hardware and compare it against the [InMemory baseline](/backends/inmemory/examples/stress) to quantify the AMQP round-trip cost. The harness sweeps handler profiles and consumer counts, making it straightforward to find the prefetch and concurrency configuration that maximises throughput for your workload. ### Prerequisites * Docker (a RabbitMQ testcontainer with the `rabbitmq_consistent_hash_exchange` plugin is started automatically) * Cargo feature: `rabbitmq` ### Run ```sh cargo run --example rabbitmq_stress --features rabbitmq ``` Narrow to a single tier or handler profile: ```sh cargo run --example rabbitmq_stress --features rabbitmq -- --tier moderate --handler fast ``` Run in release mode for representative numbers: ```sh cargo run -q --release --example rabbitmq_stress --features rabbitmq ``` ### Expected output Non-deterministic. Look for these characteristic markers: ```text shove stress benchmarks — rabbitmq scenarios: 60 [1/60] moderate | 20000msg | 1c | fast (1-5ms) ... -> 3200.5 msg/s | dispatch p50=1.2ms p99=4.8ms | e2e p50=3.1ms p99=6.2ms | cpu=45% rss=28.4MB | 6.3s ... Backend: rabbitmq TIER MSGS C HANDLER MSG/SEC ... moderate 20000 1 fast 3200 ... moderate 20000 4 fast 11500 ... ... ``` Throughput is typically in the thousands of msg/s for `fast` handlers (compared to hundreds of thousands for in-memory), reflecting AMQP round-trip cost. ### Source ```rust // [!include ~/examples/rabbitmq/stress.rs] ``` ### Walkthrough #### Container setup and plugin enabling The RabbitMQ testcontainer is started with `RabbitMqImage::default().start()`. After obtaining the AMQP port, the example runs `rabbitmq-plugins enable rabbitmq_consistent_hash_exchange` inside the container via `exec(ExecCommand::new([...]))` and waits 2 seconds for the plugin to activate. The consistent-hash plugin is required for the `StressTestTopic` topology declared by the shared harness, which uses sequenced shards. #### Queue purge between scenarios `HarnessConfig::::new("rabbitmq").with_purge(purge)` injects a purge closure that connects a fresh AMQP connection and calls `channel.queue_purge(QUEUE_NAME)` between scenarios. Unlike in-memory (where a new `Broker\` is created per scenario from scratch), RabbitMQ reuses the same broker-level topology and only clears messages. This keeps scenario boot cost low — re-declaring exchanges and bindings for every scenario would add seconds of overhead. #### `ConsumerGroupConfig` with concurrent processing The `make_cfg` closure passed to `run_all_scenarios` produces: ```rust,no_run rmq::ConsumerGroupConfig::new(consumers..=consumers) .with_prefetch_count(prefetch) .with_concurrent_processing(concurrent) ``` When `--concurrent` is passed on the command line, `concurrent=true` enables the overlap mode from the [concurrent example](/backends/rabbitmq/examples/concurrent); this is especially effective for `slow` and `heavy` handler profiles where I/O dominates. #### Interpreting the results Key columns to watch: * `disp p50/p99` — the AMQP round-trip from publish to handler entry; this reflects broker and network latency, not handler work. * `scaling_efficiency` — how close to linear throughput scales. RabbitMQ often exceeds 1.0x at low consumer counts because AMQP channels parallelize well; at high counts the broker becomes the bottleneck. * `RSS(MB)` — in-process memory. RabbitMQ messages are held in the broker, so RSS stays lower than in-memory backends at equivalent message counts. ### What to try next * Compare `--tier moderate --handler zero` throughput here against `inmemory_stress` — the difference is the AMQP round-trip cost per message. * Add `--concurrent` for `--handler slow` — watch the speedup in the `MSG/SEC` column as I/O is overlapped within each consumer. * Run `--output json` and load results into a spreadsheet to plot latency percentiles across consumer counts. * See the [InMemory stress example](/backends/inmemory/examples/stress) for in-process baseline numbers. ## NATS — Audited Consumer Use this when you need to confirm that audit trace IDs survive hold-queue retries on NATS. The handler returns `Outcome::Retry` on first delivery; both attempts emit an audit record with the same UUID, demonstrating that the `trace_id` propagates correctly through JetStream's header mechanism. ### Prerequisites * Docker (a NATS JetStream testcontainer is started automatically) * Cargo features: `nats,audit` ### Run ```sh cargo run --example nats_audited_consumer --features nats,audit ``` ### Expected output ```text topology declared published payment [handler] payment=PAY-001 amount=$49.99 attempt=1 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Retry", "duration_ms": 0, ... } [handler] payment=PAY-001 amount=$49.99 attempt=2 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Ack", "duration_ms": 0, ... } done ``` The same UUID appears in both audit records. The 10-second run window allows the 5-second hold queue to fire and redeliver before shutdown. ### Source ```rust // [!include ~/examples/nats/audited_consumer.rs] ``` ### Walkthrough #### Consumer group path for audited consumers Unlike the NATS sequenced example (which drops to `NatsConsumer::run_fifo`), this example uses the generic `Broker\` consumer group API. `broker.consumer_group()` creates a `ConsumerGroup\`, and `group.register` binds the audited factory `|| PaymentHandler.audited(StdoutAuditHandler)` to `Payments`. The `NatsConsumerGroupConfig::new(1..=1)` starts exactly one worker, keeping output strictly ordered for readability. #### `StdoutAuditHandler` as a `Clone` type `StdoutAuditHandler` derives `Clone` and `Default` because `group.register` calls the factory closure once per worker, and audited handlers must be cloneable so each worker gets its own instance. The `Clone` impl here is trivially cheap — `StdoutAuditHandler` holds no state. A handler that writes to a database connection pool would clone an `Arc` instead. #### Trace ID propagation on NATS On NATS, shove embeds the `trace_id` in a JetStream message header (`shove-trace-id`) before publishing to the hold queue. When the hold queue delivers the message back, the consumer reads the header and restores the trace ID into the `AuditRecord`. This round-trip through JetStream's metadata system is what allows the same UUID to appear on both delivery attempts. #### 10-second window and hold queue TTL The supervisor runs for 10 seconds — long enough for the 5-second hold queue TTL to expire and the retried message to arrive. `run_until_timeout` then drains for up to 10 additional seconds. If you shorten the run window below 5 seconds, the second delivery will not arrive and the example will exit with only one audit record. ### What to try next * Return `Outcome::Retry` for three consecutive attempts and confirm the `trace_id` stays the same across all three deliveries. * Add a `.with_max_retries(1)` to the consumer group config and make the handler always retry — observe the final delivery with `outcome: "Reject"` after the retry limit. * Replace `StdoutAuditHandler` with a struct that appends to a `Vec` and assert after shutdown that each payment has exactly two records (one retry, one ack). * See [Guides: Audit](/guides/audit) for the complete `AuditRecord` schema and production wiring patterns. ## NATS — Basic Use this as your starting point for NATS JetStream integration. Three `OrderCreated` events are published and consumed via a coordinated consumer group, showing how `shove`'s generic `Broker\` API maps to JetStream streams and durable consumers with hold queues and DLQ. ### Prerequisites * Docker (a NATS JetStream testcontainer is started automatically — no manual `docker run` needed) * Cargo feature: `nats` ### Run ```sh cargo run --example nats_basic --features nats ``` ### Expected output ```text Published order ORD-0 Published order ORD-1 Published order ORD-2 Processing order ORD-0 ($99.99) [retry=0] Processing order ORD-1 ($100.99) [retry=0] Processing order ORD-2 ($101.99) [retry=0] Done. ``` ### Source ```rust // [!include ~/examples/nats/basic.rs] ``` ### Walkthrough #### JetStream testcontainer `NatsServerCmd::default().with_jetstream()` enables the JetStream persistence layer inside the NATS server. Without JetStream the broker is a pure pub-sub bus with no durability or consumer-group semantics — shove requires JetStream for all its delivery guarantees. The container starts on the default NATS port 4222 and is torn down when the `container` variable is dropped. #### Topic with hold queues `OrderTopic` uses `TopologyBuilder::new("orders").hold_queue(Duration::from_secs(1)).hold_queue(Duration::from_secs(5)).dlq()`. On NATS, shove provisions one JetStream stream per topic name and separate durable consumers for the main queue, each hold queue, and the DLQ. The hold queue durations determine how long a retried message waits before redelivery — 1 s then 5 s in this example. #### Consumer group via `Broker` `broker.consumer_group()` returns a `ConsumerGroup\`. `group.register::(ConsumerGroupConfig::new(NatsConsumerGroupConfig::new(1..=1)), || OrderHandler)` wires the handler factory to the topic with a fixed pool of one worker. The `NatsConsumerGroupConfig` maps to a JetStream durable consumer with `max_ack_pending` set from the computed prefetch count. The factory closure `|| OrderHandler` is called once per spawned worker. #### Run until timeout `group.run_until_timeout(signal, drain_timeout)` drives the consumer group event loop. The signal future resolves after either 3 seconds or Ctrl-C, whichever comes first; after the signal the group drains in-flight messages for up to 10 seconds before returning. `outcome.exit_code()` is 0 if the drain completed cleanly. ### What to try next * Add a second `group.register` call for a different topic to consume from multiple streams in the same process. * Change `NatsConsumerGroupConfig::new(1..=1)` to `new(1..=4)` and publish 20 messages — observe how the group scales workers to match queue depth. * Return `Outcome::Retry` from `OrderHandler` and watch the 1-second hold queue delay before redelivery. * See [Guides: Groups](/guides/groups) for how `ConsumerGroup` distributes work across workers. ## NATS — Sequenced Use this when you need per-user (or per-entity) FIFO ordering on NATS. Events for two users are published in interleaved order; each user's events arrive strictly in sequence while the two users are processed concurrently across different shards. Shows the `NatsConsumer::run_fifo` path for sequenced delivery. ### Prerequisites * Docker (a NATS JetStream testcontainer is started automatically) * Cargo feature: `nats` ### Run ```sh cargo run --example nats_sequenced --features nats ``` ### Expected output Events for each user arrive in `seq` order (0 before 1 before 2…), but events for different users may interleave: ```text Published 10 events (5 per user) [user=alice] action=action-0 seq=0 (retry=0) [user=bob] action=action-0 seq=0 (retry=0) [user=alice] action=action-1 seq=1 (retry=0) [user=bob] action=action-1 seq=1 (retry=0) ... Done. ``` ### Source ```rust // [!include ~/examples/nats/sequenced.rs] ``` ### Walkthrough #### `define_sequenced_topic!` with subject-shard routing `UserEventTopic` is declared with `define_sequenced_topic!(Name, MsgType, key_fn, topology)`. The key function `|msg: &UserEvent| msg.user_id.clone()` maps each event to its per-user FIFO lane. The topology uses `.sequenced(SequenceFailure::Skip)` — if one event for a user is rejected, it is DLQ'd and the sequence for that user continues. `.routing_shards(4)` provisions four JetStream subjects (`user-events.0` through `user-events.3`) using consistent hashing; each user always hashes to the same shard, ensuring ordering within the shard. #### Subject-shard routing on NATS On NATS, shove's sequenced delivery uses JetStream subjects as shards. Each shard is backed by a durable consumer with `max_ack_pending: 1` — only one message per shard can be in-flight at a time, guaranteeing that the next message for a key is not dispatched until the current one is acknowledged or rejected. With four shards and two users, each user lands on a different shard and their events are processed concurrently (but each user's events are strictly ordered within their shard). #### Backend-specific `NatsConsumer::run_fifo` Because `run_fifo` is not yet available on the generic `ConsumerGroup\` API, the example constructs `NatsConsumer::new(client)` and calls `run_fifo::(handler, ctx, opts)` directly. `ConsumerOptions::::new().with_shutdown(shutdown)` carries the cancellation token that stops the consumer after the 5-second timer fires. #### Deduplication window NATS JetStream supports a `Nats-Msg-Id` header for server-side deduplication within a configurable time window. The shove publisher sets this header on every publish, so messages published within the dedup window (default 2 minutes) with the same ID are deduplicated by the broker — useful for at-least-once publisher retries. This example does not exercise dedup explicitly, but the header is always present. ### What to try next * Change `SequenceFailure::Skip` to `SequenceFailure::FailAll` and make the handler reject `user=alice seq=2` — confirm that `seq=3` and `seq=4` for Alice are automatically DLQ'd while Bob's sequence is unaffected. * Increase `routing_shards(4)` to `routing_shards(8)` — more subjects spread keys more evenly but add stream setup overhead. * Replace the 5-second fixed timer with a counter-based shutdown (see [InMemory sequenced](/backends/inmemory/examples/sequenced)) to stop as soon as all 10 events are processed. * Read [Guides: Sequenced](/guides/sequenced) for the full explanation of shard routing and failure modes. ## NATS — Stress Use this to measure NATS JetStream throughput and compare it against the [InMemory baseline](/backends/inmemory/examples/stress) and [RabbitMQ results](/backends/rabbitmq/examples/stress). The numbers reveal JetStream's persistence cost versus in-process, and the protocol efficiency difference versus AMQP — useful data before choosing a backend or sizing a deployment. ### Prerequisites * Docker (a NATS JetStream testcontainer is started automatically) * Cargo feature: `nats` ### Run ```sh cargo run --example nats_stress --features nats ``` Narrow to one tier or handler profile: ```sh cargo run --example nats_stress --features nats -- --tier moderate --handler fast ``` Release mode for representative numbers: ```sh cargo run -q --release --example nats_stress --features nats ``` ### Expected output Non-deterministic. Look for these characteristic markers: ```text shove stress benchmarks — nats scenarios: 60 [1/60] moderate | 20000msg | 1c | fast (1-5ms) ... -> 12500.3 msg/s | dispatch p50=0.4ms p99=2.1ms | e2e p50=2.8ms p99=5.3ms | cpu=55% rss=38.2MB | 1.6s ... Backend: nats TIER MSGS C HANDLER MSG/SEC ... moderate 20000 1 fast 12500 ... moderate 20000 4 fast 42000 ... ... ``` NATS JetStream typically achieves higher throughput than RabbitMQ (lower protocol overhead) but lower than in-memory (no persistence or network round-trip). ### Source ```rust // [!include ~/examples/nats/stress.rs] ``` ### Walkthrough #### Stream purge strategy `with_purge(purge)` injects a closure that drops and recreates the entire JetStream stream between scenarios using `js.delete_stream(STREAM_NAME)`. This is necessary because changing `max_ack_pending` on an existing durable consumer requires an explicit `UpdateConsumer` call, which the current harness does not perform — cleanest to drop the stream and let the next scenario's `declare` recreate it fresh. The scenario boot overhead is small compared to the processing time for even the `moderate` tier. #### `NatsConsumerGroupConfig` with `with_concurrent_processing` The `make_cfg` closure produces: ```rust NatsConsumerGroupConfig::new(consumers..=consumers) .with_prefetch_count(prefetch) .with_concurrent_processing(concurrent) ``` The `prefetch_count` maps to JetStream's `max_ack_pending` — how many messages the JetStream server will push to the consumer before requiring acknowledgements. When `--concurrent` is passed, `with_concurrent_processing(true)` enables parallel handler invocations within a single consumer, which is especially effective for `slow` and `heavy` profiles. #### Interpreting the results Key metrics to compare across backends: * `disp p50/p99` — the latency from publish timestamp to handler entry. For NATS this includes JetStream replication and push-delivery overhead. Typically 0.2–2 ms at moderate load, compared to 1–5 ms for RabbitMQ. * `scaling_efficiency` — how closely throughput scales with consumer count. NATS JetStream push consumers scale well up to the stream's partition count; above that point the bottleneck shifts to the JetStream server. * `RSS(MB)` — messages are held in the JetStream server's storage, so in-process memory stays lower than for in-memory backends at equivalent message counts. #### `run_all_scenarios` — coordinated group path NATS implements `HasCoordinatedGroups`, so `run_all_scenarios` is used (not `run_supervisor_scenarios`). Each scenario creates a fresh `Broker\`, which provisions a new JetStream connection and durable consumer. The purge closure ensures the stream is empty before each scenario's publish phase. ### What to try next * Compare `--tier moderate --handler zero` throughput here against the InMemory and RabbitMQ stress results to quantify NATS's overhead relative to both extremes. * Run `--concurrent --handler slow` — the throughput improvement will be pronounced because NATS's push delivery already fills the `max_ack_pending` window; concurrent processing consumes that window faster. * Try `--tier extreme` in release mode — NATS typically sustains hundreds of thousands of msg/s for `zero` handlers before JetStream becomes the bottleneck. * See the [NATS backend overview](/backends/nats) for JetStream stream configuration options not covered by this harness. ## Kafka — Audited Consumer Use this when you need to confirm that audit trace IDs survive Kafka hold-queue retries. The handler returns `Outcome::Retry` on first delivery; both delivery attempts emit a JSON audit record with the same UUID, demonstrating that the `trace_id` propagates correctly through Kafka record headers across the hold-queue topic round-trip. ### Prerequisites * Docker (an Apache Kafka testcontainer is started automatically) * Cargo features: `kafka,audit` ### Run ```sh cargo run --example kafka_audited_consumer --features kafka,audit ``` ### Expected output ```text topology declared published payment [handler] payment=PAY-001 amount=$49.99 attempt=1 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Retry", "duration_ms": 0, ... } [handler] payment=PAY-001 amount=$49.99 attempt=2 [audit] { "trace_id": "550e8400-e29b-41d4-a716-446655440000", "outcome": "Ack", "duration_ms": 0, ... } done ``` The same UUID appears on both delivery attempts. The 10-second run window allows the 5-second hold queue topic to deliver the retried message before shutdown. ### Source ```rust // [!include ~/examples/kafka/audited_consumer.rs] ``` ### Walkthrough #### Consumer group for audited consumers Unlike the Kafka sequenced example (which drops to `KafkaConsumer::run_fifo`), this example uses the generic `Broker\` consumer group API. `group.register::(ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)), || PaymentHandler.audited(StdoutAuditHandler))` wires the audited factory to the `Payments` topic. The `1..=1` range starts exactly one consumer for simple, ordered output. #### `StdoutAuditHandler` as a `Clone` type `StdoutAuditHandler` derives `Clone` and `Default` because the factory closure `|| PaymentHandler.audited(StdoutAuditHandler)` is called once per spawned consumer, and each call produces a fresh `Audited\` value. In this example `StdoutAuditHandler` holds no state, so `Clone` is trivially cheap. A real audit handler that batches records to Kafka would clone an `Arc` instead. #### Trace ID propagation via Kafka headers On Kafka, shove embeds the `trace_id` in a record header (`shove-trace-id`). When `Outcome::Retry` causes the message to be published to the hold-queue topic, the `shove-trace-id` header is copied to the hold-queue record. When the hold-queue consumer redelivers the message back to the main topic, the header is again copied. This chain of header propagation is what allows the same UUID to appear on both the retry and the final ack delivery without any changes to the handler code. #### Hold queue as a separate Kafka topic On Kafka, `Outcome::Retry` publishes the message to `kafka-audited-payments-hold-5s` (a separate Kafka topic) and commits the offset on the main topic. After 5 seconds a framework-internal consumer reads from the hold-queue topic and republishes to the main topic. This is why the 10-second run window is needed — the 5-second TTL plus consume-and-republish latency must both fit within the window. ### What to try next * Inspect the Kafka topic list after the example exits — you should see `kafka-audited-payments`, `kafka-audited-payments-hold-5s`, and `kafka-audited-payments-dlq` all present. * Add `.with_max_retries(1)` and make `PaymentHandler` always retry — verify the audit record shows `outcome: "Reject"` on the final delivery after the retry limit. * Replace `StdoutAuditHandler` with one that publishes to a separate Kafka topic using the same `broker.publisher()`. * See [Guides: Audit](/guides/audit) for the complete `AuditRecord` schema. ## Kafka — Basic Use this as your starting point for Kafka integration. Three `OrderCreated` messages are published and consumed via a coordinated consumer group, showing how `shove`'s generic `Broker\` API maps to rdkafka producers, consumer groups, and partition assignment — including how hold queues are implemented as separate Kafka topics. ### Prerequisites * Docker (an Apache Kafka testcontainer is started automatically — no manual `docker run` needed) * Cargo feature: `kafka` ### Run ```sh cargo run --example kafka_basic --features kafka ``` ### Expected output ```text Published order ORD-0 Published order ORD-1 Published order ORD-2 Processing order ORD-0 ($99.99) [retry=0] Processing order ORD-1 ($100.99) [retry=0] Processing order ORD-2 ($101.99) [retry=0] Done. ``` ### Source ```rust // [!include ~/examples/kafka/basic.rs] ``` ### Walkthrough #### Kafka topic provisioning on declare `broker.topology().declare::()` creates the Kafka topics that back the topology: one main topic (`kafka-orders`), two hold-queue topics (`kafka-orders-hold-1s`, `kafka-orders-hold-5s`), and a DLQ topic (`kafka-orders-dlq`). On Kafka, each logical shove queue is a separate Kafka topic. The number of partitions is determined by the consumer group's `max_consumers` setting — a partition count less than the consumer count would leave some consumers idle. #### `KafkaConfig` and bootstrap servers `Broker::::new(KafkaConfig::new(&bootstrap))` initialises the rdkafka producer and consumer configuration pointing at the single-broker testcontainer. In production `bootstrap` would be a comma-separated list of broker addresses. `KafkaConfig` exposes additional methods for TLS (`with_tls`) and SASL authentication (`with_sasl`) — see the [Kafka backend overview](/backends/kafka) for the `kafka-ssl` feature. #### Consumer group with coordinated partition assignment `broker.consumer_group()` creates a `ConsumerGroup\`. `group.register::(ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)), || OrderHandler)` binds the handler factory to `OrderTopic` with a fixed pool of one consumer. On Kafka, the consumer group registers a standard rdkafka consumer group; when multiple consumers are active they receive distinct partition assignments from the Kafka coordinator — messages within a partition remain ordered. #### Hold queues as separate Kafka topics Unlike RabbitMQ (which uses AMQP dead-letter routing) or NATS (which uses JetStream subjects), Kafka hold queues are separate topics. When a handler returns `Outcome::Retry`, shove publishes the message to the appropriate hold-queue topic and acknowledges the original. A separate consumer (run internally by the framework) reads from the hold-queue topic after the TTL and republishes to the main topic. This approach keeps messages durable across broker restarts at the cost of one additional produce+consume round-trip per retry. ### What to try next * Change `new(1..=1)` to `new(1..=4)` and publish 20 messages — the consumer group assigns partitions dynamically; observe rebalance log lines during the 3-second run window. * Return `Outcome::Retry` from `OrderHandler` and verify the 1-second hold queue fires and redelivers before the 3-second window closes. * Add TLS by enabling the `kafka-ssl` feature and configuring `KafkaConfig::with_tls` — no handler code changes needed. * See [Guides: Retries](/guides/retries) for how hold queues and retry limits interact on Kafka. ## Kafka — MSK IAM Round-trips a single `Ping` message against an IAM-enabled Amazon MSK cluster, using `SASL_SSL` + `OAUTHBEARER` with auto-refreshed presigned tokens. Run this against a real MSK cluster you have access to — it's the canonical end-to-end check for the `kafka-msk-iam` feature. ### Prerequisites * A running MSK cluster with IAM authentication enabled * AWS credentials available via the standard provider chain (env vars, shared credentials file, EC2 instance metadata, EKS pod identity / IRSA, or SSO) * The caller's principal must have `kafka-cluster:Connect` on the cluster (plus `kafka-cluster:WriteData` and `ReadData` on the topic for the publish/consume round-trip) * Cargo features: `kafka-msk-iam` (implies `kafka-ssl`) ### Run ```sh MSK_BROKERS=b-1.example.kafka.us-east-1.amazonaws.com:9098,b-2.example.kafka.us-east-1.amazonaws.com:9098 \ AWS_REGION=us-east-1 \ cargo run --example kafka_msk_iam --features kafka-msk-iam ``` For a non-default named profile, set `AWS_PROFILE` alongside the rest — the AWS SDK picks it up via the standard provider chain. The example uses `KafkaSasl::msk_iam(region)`, which is the no-profile form. To pin a specific profile in code, swap in `KafkaSasl::msk_iam_with_profile(region, profile)`. ### Expected output ```text Using topic: shove-msk-iam-example-7a3f9c4e8b1d4e2a9c6f5b8d3e1a4f7b Connecting to MSK cluster at b-1.example.kafka.us-east-1.amazonaws.com:9098,... (region=us-east-1) … Connected. Published ping: id=4f2a8b3c-1d5e-4f6a-9b8c-2d7e3a5f8b1c Received ping: id=4f2a8b3c-1d5e-4f6a-9b8c-2d7e3a5f8b1c Done. ``` ### Source ```rust // [!include ~/examples/kafka/msk_iam.rs] ``` ### Walkthrough #### IAM auth on the `KafkaConfig` ```rust let config = KafkaConfig::new(&brokers) .with_tls(KafkaTls::default()) .with_sasl(KafkaSasl::msk_iam(®ion)); ``` `KafkaTls::default()` is the right TLS configuration for MSK — clusters use publicly-signed ACM certificates, so the OS trust store handles validation. There's nothing to download, no CA path to set. `KafkaSasl::msk_iam(region)` selects the OAUTHBEARER mechanism and binds the token signer to `region`. Behind the scenes shove sets `security.protocol=SASL_SSL`, `sasl.mechanism=OAUTHBEARER`, and installs a custom `ClientContext` whose token-refresh callback signs a fresh presigned URL via `aws-msk-iam-sasl-signer` whenever librdkafka asks for one (every \~12 minutes at default expiry). If you forget to call `with_tls(...)` for an MSK IAM config, `Broker::::new` returns `ShoveError::Topology` at connect time — presigned tokens over plaintext are a misconfiguration, not a runtime fallback. #### Per-run topic name ```rust let topic_name = format!("shove-msk-iam-example-{}", Uuid::new_v4().simple()); PING_TOPOLOGY .set(TopologyBuilder::new(&topic_name).build()) .expect("PING_TOPOLOGY already initialised"); ``` The `Topic::topology()` method must return a `&'static QueueTopology`, but the topic name needs to differ across runs so reruns don't collide on the cluster. The example resolves this with a `OnceLock` initialised in `main` before the broker starts — the topology is set exactly once, then returned by reference forever after. #### Topic provisioning ```rust broker.topology().declare::().await?; ``` `declare` creates the Kafka topics that back the topology (one main topic plus DLQ + hold queues if configured). The caller's IAM principal needs `kafka-cluster:CreateTopic` on the cluster ARN. In production where topology is pre-provisioned by infrastructure, this call is idempotent — it does nothing if the topics already exist with sufficient partitions. #### Single consumer in a group ```rust group .register::( ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)), || PingHandler, ) .await?; ``` `1..=1` pins exactly one consumer in the group, which keeps the example deterministic. For real workloads, see the [Basic example](/backends/kafka/examples/basic) for how the group scales partition assignment dynamically. ### Token rotation Tokens are signed locally — no network call to AWS per refresh. librdkafka schedules a refresh at \~20% of the remaining lifetime (so \~3 minutes before expiry for a 15-minute token). The refresh callback runs on librdkafka's internal poll thread; shove bridges it to the async AWS signer via a Tokio runtime handle captured at `connect()` time. Do not set `sasl.oauthbearer.config` manually. The shove integration owns this property. ### What to try next * Stop the running example mid-flight, revoke the caller's `kafka-cluster:Connect` permission for \~30 seconds, then restore it — librdkafka will report auth failures during the gap and recover on the next refresh. * Switch to `KafkaSasl::msk_iam_with_profile(region, "production")` and confirm shove reads credentials from the named profile instead of the default chain. * Wire this same code into a long-running consumer service and observe token-refresh log lines from librdkafka over a multi-hour run. * See the [Kafka backend overview](/backends/kafka) for the cargo feature matrix and other auth modes. ## Kafka — Schema Registry Configure a Kafka consumer to decode [Confluent Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html)-framed messages, and a publisher to produce them. On each consumed message the consumer strips the wire frame (magic byte `0x00` + big-endian schema id), resolves the schema from the registry (cached in memory), validates the schema subject against the accepted set, then delegates to the topic's `Codec` for the actual payload decode. On the produce side, an opt-in publisher wraps each encoded payload in the same frame — see [Producing SR-framed messages](#producing-sr-framed-messages). Works with the Confluent Schema Registry and with **Redpanda's built-in Schema Registry** — both expose the same REST API and wire format. ### Prerequisites * Cargo features: `kafka-schema-registry` (implies `kafka`) * A running Confluent-compatible Schema Registry (Confluent, Redpanda, or compatible) * Topics must use `JsonCodec` or `ProtobufCodec`; other codecs are not supported ### Install ```sh cargo add shove --features kafka-schema-registry ``` ### Build the registry client ```rust,no_run use std::time::Duration; use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth}; let registry = SchemaRegistry::builder("http://schema-registry:8081") .auth(SchemaRegistryAuth::Basic { user: "sr-user".into(), pass: "sr-pass".into(), }) .timeout(Duration::from_secs(3)) .max_retries(2) .negative_cache_ttl(Duration::from_secs(60)) .build(); ``` `build()` returns `Arc`. Clone the `Arc` to share the same schema cache across consumers or a consumer group. Auth options: | Variant | Use | | ------------------------------------------ | ------------------------------- | | `SchemaRegistryAuth::None` | No authentication (default) | | `SchemaRegistryAuth::Bearer(token)` | `Authorization: Bearer ` | | `SchemaRegistryAuth::Basic { user, pass }` | HTTP Basic auth | ### Per-consumer setup ```rust,no_run use std::sync::Arc; use shove::schema_registry::{SchemaEnforcement, SchemaRegistry}; use shove::consumer::ConsumerOptions; use shove::markers::Kafka; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); let opts = ConsumerOptions::::new() .with_schema_registry(Arc::clone(®istry)) .with_schema_enforcement(SchemaEnforcement::Enforce) .accept_schema_subjects(["orders-value"]); ``` ### Per-consumer-group setup Attaching the registry on `KafkaConsumerGroupConfig` shares one `Arc` — and therefore one schema cache — across the whole autoscaling group: ```rust,no_run use std::sync::Arc; use shove::kafka::KafkaConsumerGroupConfig; use shove::schema_registry::{SchemaEnforcement, SchemaRegistry}; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); let cfg = KafkaConsumerGroupConfig::new(1..=8) .with_schema_registry(Arc::clone(®istry)) .with_schema_enforcement(SchemaEnforcement::Enforce) .accept_schema_subjects(["orders-value"]); ``` ### Producing SR-framed messages The same registry client drives producer-side encoding. Attach it to a `KafkaPublisherConfig` and get the publisher via `Broker::publisher_with` — every publish then emits a Confluent-framed payload, wire-compatible with any standard Confluent consumer (including ClickPipes and shove's own decode path above). ```rust,no_run use std::sync::Arc; use shove::kafka::KafkaPublisherConfig; use shove::schema_registry::SchemaRegistry; let registry = SchemaRegistry::builder("http://schema-registry:8081").build(); // `broker` is a `Broker`. let publisher = broker .publisher_with(KafkaPublisherConfig::new().with_schema_registry(Arc::clone(®istry))) .await?; // Emits `0x00` + big-endian schema id + encoded payload. publisher.publish::(&order).await?; ``` The publisher looks up the latest schema id for the subject (`GET /subjects/{subject}/versions/latest`, cached per subject) — it carries no schema text, so the subject must already be registered. The subject defaults to `"{topic}-value"`; override it with `KafkaPublisherConfig::with_subject("...")`. Framing applies only to `JsonCodec` / `ProtobufCodec` topics (Protobuf uses message index `[0]`); a plain `Broker::publisher()` does no framing. ### Enforcement modes `SchemaEnforcement` controls what happens when a message's registered subject is not in the accepted set: * **`Enforce`** (default) — the message is routed to the DLQ with death reason `schema_validation_failed`. Use this in production where a subject mismatch is a producer misconfiguration. * **`Permissive`** — the mismatch is logged and counted; the message is still decoded. Use this during migration windows or when producers share a topic with different subject conventions. ### Accepted subjects `accept_schema_subjects([...])` pins the set of Confluent schema subjects the consumer will accept. When not set, the default follows the Confluent [TopicNameStrategy](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#subject-name-strategy): `"{queue}-value"`. For a topic named `orders` that resolves to `orders-value`. ### DLQ consumer caveat :::warning If you enable registry decoding on the consumer that drains the DLQ (`handle_dead` path), any dead message that cannot be resolved or validated by the registry cannot be delivered to `handle_dead`. Such messages are acked-and-dropped, matching the existing behavior for undecodable dead messages. To avoid this, omit the registry on the DLQ consumer, or set `SchemaEnforcement::Permissive` so subject mismatches do not block delivery. ::: ### What to try next * Change `SchemaEnforcement::Enforce` to `SchemaEnforcement::Permissive` and verify that subject-mismatched messages reach your handler rather than the DLQ. * Share one `Arc` across two different `ConsumerGroupConfig` registrations to confirm they hit the same cache. * Point `SchemaRegistry::builder` at a Redpanda broker URL to confirm Redpanda Schema Registry compatibility. * See the [Kafka backend overview](/backends/kafka#schema-registry) for the full configuration reference. ## Kafka — Sequenced Use this when you need per-entity FIFO ordering on Kafka. Events for two users are published in interleaved order; each user's events arrive strictly in sequence. The example also highlights the key Kafka constraint: partition count caps the maximum number of active consumers, so partition count must be planned before topic creation. ### Prerequisites * Docker (an Apache Kafka testcontainer is started automatically) * Cargo feature: `kafka` ### Run ```sh cargo run --example kafka_sequenced --features kafka ``` ### Expected output Events for each user arrive in `seq` order, but events for different users may interleave across shards: ```text Published 10 events (5 per user) [user=alice] action=action-0 seq=0 (retry=0) [user=bob] action=action-0 seq=0 (retry=0) [user=alice] action=action-1 seq=1 (retry=0) [user=bob] action=action-1 seq=1 (retry=0) ... Done. ``` ### Source ```rust // [!include ~/examples/kafka/sequenced.rs] ``` ### Walkthrough #### Partition-key routing on Kafka `UserEventTopic` uses `.routing_shards(4)` which on Kafka means 4 Kafka partitions on the main topic. The key function `|msg: &UserEvent| msg.user_id.clone()` is hashed (consistent hashing) to select a partition; all events for the same user always land on the same partition. Within a Kafka partition messages are stored in strict append order — this is the mechanism that gives shove its per-key ordering guarantee. Changing the partition count requires deleting and recreating the topic; Kafka can only expand, not shrink. #### `KafkaConsumer::run_fifo` and the partition constraint `KafkaConsumer::new(client).run_fifo::(handler, ctx, opts)` is the backend-specific path for per-key FIFO (not yet exposed on the generic `ConsumerGroup\` API). With 4 partitions and 1 consumer, the consumer owns all 4 partitions. Adding a second consumer triggers a Kafka rebalance and each consumer is assigned 2 partitions. If the consumer count exceeds the partition count, some consumers receive no partitions and sit idle — a Kafka-specific constraint that does not apply to AMQP or JetStream backends. #### `SequenceFailure::Skip` and DLQ routing `.sequenced(SequenceFailure::Skip)` means that if a message for `alice` is rejected, only that message goes to the DLQ; subsequent messages for `alice` continue to be processed. The rejected message is published to the DLQ topic by the framework before the consumer commits its offset, ensuring at-least-once delivery for the DLQ entry even across consumer restarts. #### Cancellation token shutdown `CancellationToken::new()` is cancelled after 5 seconds by a spawned task. `ConsumerOptions::::new().with_shutdown(shutdown)` passes the token to `run_fifo`, which stops polling and drains when the token fires. `client.shutdown().await` then closes the rdkafka connections cleanly. ### What to try next * Change `routing_shards(4)` to `routing_shards(8)` and `consumers` to 8 in `KafkaConsumerGroupConfig` — each partition gets exactly one consumer, maximising parallelism with no idle consumers. * Switch `SequenceFailure::Skip` to `SequenceFailure::FailAll` and make the handler reject `alice seq=2` — all subsequent messages for Alice are DLQ'd while Bob's sequence continues. * Increase the consumer count beyond the partition count — observe that the extra consumers sit idle (no partitions assigned) and log a warning. * Read [Guides: Sequenced](/guides/sequenced) for a full explanation of shard routing and failure modes across backends. ## Kafka — Stress Use this to measure Kafka throughput on your workload and compare it against the [InMemory baseline](/backends/inmemory/examples/stress) and [NATS results](/backends/nats/examples/stress). The numbers quantify Kafka's write-ahead log cost relative to lighter backends — useful data when choosing between Kafka and NATS, or when sizing partitions and consumer counts for a production deployment. ### Prerequisites * Docker (an Apache Kafka testcontainer is started automatically) * Cargo feature: `kafka` ### Run ```sh cargo run --example kafka_stress --features kafka ``` Narrow to one tier or handler profile: ```sh cargo run --example kafka_stress --features kafka -- --tier moderate --handler fast ``` Release mode for representative numbers: ```sh cargo run -q --release --example kafka_stress --features kafka ``` ### Expected output Non-deterministic. Look for these characteristic markers: ```text shove stress benchmarks — kafka scenarios: 60 [1/60] moderate | 20000msg | 1c | fast (1-5ms) ... -> 8200.4 msg/s | dispatch p50=0.3ms p99=1.8ms | e2e p50=2.9ms p99=5.1ms | cpu=62% rss=44.1MB | 2.4s ... Backend: kafka TIER MSGS C HANDLER MSG/SEC ... moderate 20000 1 fast 8200 ... moderate 20000 4 fast 25000 ... ... ``` Throughput is typically in the thousands to tens of thousands of msg/s — higher than RabbitMQ for `zero` handlers due to Kafka's batch-oriented producer, but with a hard ceiling at the partition count. ### Source ```rust // [!include ~/examples/kafka/stress.rs] ``` ### Walkthrough #### Topic delete-and-recreate between scenarios `with_purge(purge)` injects a closure that uses an rdkafka `AdminClient` to delete the main topic and DLQ topic between scenarios. Unlike NATS (which deletes the stream) or RabbitMQ (which purges the queue), Kafka deletes and recreates topics because `ensure_partitions` can only expand partition counts, not shrink them. Each scenario declares its own topology with a partition count matched to `consumers`, so the previous scenario's partition layout must be cleared to avoid idle consumers. #### Partition count and consumer ceiling The `make_cfg` closure produces: ```rust KafkaConsumerGroupConfig::new(consumers..=consumers) .with_prefetch_count(prefetch) ``` On Kafka, `declare` provisions exactly `consumers` partitions for the main topic. When the stress harness steps through consumer counts like `[1, 4, 8, 16, 32]`, each scenario first purges and then redeclares with the correct partition count for that scenario's consumer level. The consumer count ceiling for `scaling_efficiency` is thus the partition count — adding more consumers than partitions would leave some idle and cap throughput. #### `KafkaConsumerGroupConfig` with concurrent processing `with_concurrent_processing(concurrent)` enables parallel handler dispatch within a single consumer partition assignment. When `--concurrent` is passed, each consumed record spawns a tokio task, allowing up to `prefetch` records to be processed simultaneously within one Kafka consumer instance. This is effective for `slow` and `heavy` handler profiles but provides little benefit for `zero` (no-op) handlers where CPU is the bottleneck. #### Interpreting the results Key metrics: * `disp p50` — time from `published_at_ns` (embedded in the message) to handler entry. For Kafka this includes batch accumulation in the rdkafka producer, network transmission, and consumer fetch latency. Typically 0.2–2 ms at moderate load with default batch settings. * `scaling_efficiency` — near-linear scaling up to the partition count; flat beyond it because additional consumers get no partition assignment. * `CPU%` — Kafka's log-structured storage and zero-copy reads keep CPU lower than AMQP backends for high-throughput scenarios. ### What to try next * Compare `--tier moderate --handler zero` throughput here against NATS and InMemory results to see how Kafka's write-ahead log affects raw scheduling cost. * Add `--concurrent --handler slow` — measure the throughput improvement as concurrent dispatch overlaps the handler I/O within each partition. * Try `--tier extreme` with 256 consumers — throughput will plateau far below the InMemory extreme because the partition count per scenario is also 256, and Kafka consumer-group rebalance overhead becomes significant. * See the [Kafka backend overview](/backends/kafka) for TLS and SASL configuration options relevant to production deployments. ## InMemory — Audited Consumer Use this when you want to add audit logging — trace IDs, outcomes, handler duration — to an existing handler without changing its logic. The `Audited` decorator wraps any `MessageHandler` transparently; this example shows the wiring with no external services required. ### Prerequisites No external services required. The in-memory backend runs entirely in-process. * Cargo features: `inmemory,audit` ### Run ```sh cargo run --example inmemory_audited_consumer --features inmemory,audit ``` ### Expected output Each message produces two lines — one from the primary handler and one from the audit handler: ```text handled event 0 audit trace_id= outcome=Ack duration_ms=0 handled event 1 audit trace_id= outcome=Ack duration_ms=0 handled event 2 audit trace_id= outcome=Ack duration_ms=0 ``` ### Source ```rust // [!include ~/examples/inmemory/audited_consumer.rs] ``` ### Walkthrough #### Primary handler `Inner` is a straightforward `MessageHandler\` that increments a shared `Arc` counter and prints the event id. It has no knowledge of auditing — it simply processes the message and returns `Outcome::Ack`. #### Audit handler `StdoutAudit` implements `AuditHandler\`. Its single method `audit` receives an `AuditRecord\` that carries the `trace_id` (a UUID assigned per message by the framework), the `outcome` returned by the inner handler, and `duration_ms` (wall-clock time the inner handler spent). In this example the record is printed to stdout; in production you would typically publish it to a dedicated audit topic using `ShoveAuditHandler`. #### Wiring with `.audited()` The factory passed to `group.register` calls `Inner { count: c.clone() }.audited(StdoutAudit)`. The `.audited()` method comes from the `MessageHandlerExt` trait and wraps `Inner` in the framework's `Audited\` combinator. The consumer group sees only a single `MessageHandler` — the auditing happens transparently inside. #### Consumer group configuration `InMemoryConsumerGroupConfig::new(1..=1).with_prefetch_count(1)` runs a single worker with no pipelining. With `prefetch_count(1)` only one message at a time is dispatched to the handler, which makes the output strictly ordered and easier to read in this demonstration. ### What to try next * Replace `StdoutAudit` with a struct that publishes to a second topic via `broker.publisher()` — this is the pattern `ShoveAuditHandler` follows on real deployments. * Set `with_prefetch_count(4)` and `new(1..=2)` — verify that audit records are still emitted for every message even under concurrent processing. * Return `Outcome::Nack` from `Inner::handle` — confirm that `AuditRecord::outcome` reflects the nack and the message is requeued. * See [Guides: Audit](/guides/audit) for a full description of the `AuditRecord` fields. ## InMemory — Basic Use this when you want to see the full publish/consume cycle with no external services: declare a topic, publish five messages, handle them, and exit cleanly. The simplest end-to-end proof that your topic definition and handler are wired correctly. ### Prerequisites No external services required. The in-memory backend runs entirely in-process. * Cargo feature: `inmemory` ### Run ```sh cargo run --example inmemory_basic --features inmemory ``` ### Expected output ```text received #0: hello 0 received #1: hello 1 received #2: hello 2 received #3: hello 3 received #4: hello 4 ``` ### Source ```rust // [!include ~/examples/inmemory/basic.rs] ``` ### Walkthrough #### Topic definition `PingTopic` is the type-level identifier for the queue. It implements the `Topic` trait, which associates the topic with its message type (`Ping`) and its `QueueTopology`. The topology is built once via `OnceLock` and describes the queue name (`"ping"`) along with a dead-letter queue (`.dlq()`). Every shove backend consults this topology to provision the necessary queues on startup. #### Connecting and declaring `Broker::::new(InMemoryConfig::default())` creates the in-process broker. Calling `broker.topology().declare::()` then materialises the `"ping"` queue (and its DLQ) inside the broker's in-memory store. Declaration is idempotent — safe to call more than once. #### Handler and consumer group `PingHandler` wraps an `Arc` counter so the main thread can observe how many messages have been processed. The handler always returns `Outcome::Ack`, which tells the broker to remove the message from the queue. `broker.consumer_group()` creates a `ConsumerGroup\`, and `group.register` binds the handler factory to `PingTopic` with a worker range of `1..=1` and a prefetch of 4. The factory closure is called once per spawned worker. #### Publishing Five `Ping` messages are published in a loop via `publisher.publish::`. The in-memory backend delivers them immediately into the in-process queue — no serialisation round-trip to a remote broker. #### Graceful shutdown A background task polls `count` every 10 ms and cancels a `CancellationToken` once all five messages are processed (or after a 5-second safety deadline). The consumer group's `run_until_timeout` call drives the event loop until that token fires, then allows in-flight handlers to drain before returning. `std::process::exit(outcome.exit_code())` propagates a non-zero code if the drain timed out. ### What to try next * Change `InMemoryConsumerGroupConfig::new(1..=1)` to `new(1..=4)` — observe how the consumer group scales workers automatically as the queue grows. * Change `Outcome::Ack` to `Outcome::Nack` in `PingHandler::handle` — watch messages loop back into the queue and eventually land in the DLQ. * Swap the `InMemory` marker for another backend (e.g. `RabbitMq`) and supply the matching config — the handler code remains unchanged. * See [Concepts: Outcomes](/concepts/outcomes) for the full list of outcome variants and their retry semantics. ## InMemory — Consumer Groups Use this when you want to see how consumer groups scale workers to match queue depth. One hundred messages are published and processed by three to six workers in parallel, each simulating 5 ms of I/O — a practical demonstration of the min/max worker range and prefetch configuration, all in-process. ### Prerequisites No external services required. The in-memory backend runs entirely in-process. * Cargo feature: `inmemory` ### Run ```sh cargo run --example inmemory_consumer_groups --features inmemory ``` ### Expected output Lines arrive in non-deterministic order because workers process messages concurrently. After all 100 messages are consumed you will see a final summary: ```text processed work id=0 processed work id=3 processed work id=1 ... processed work id=99 processed 100 messages ``` ### Source ```rust // [!include ~/examples/inmemory/consumer_groups.rs] ``` ### Walkthrough #### Topic and worker definition `WorkTopic` has a minimal topology: a `"work"` queue with a DLQ and no sequencing. `Worker` simulates realistic I/O with `tokio::time::sleep(Duration::from_millis(5))` before returning `Outcome::Ack`. The `Arc` counter is shared across all worker instances so the main thread can observe the running total. #### Dynamic worker range `InMemoryConsumerGroupConfig::new(3..=6)` declares a minimum of 3 and a maximum of 6 concurrent workers. The consumer group starts workers up to the minimum immediately and may spawn additional workers (up to the maximum) as queue depth warrants. With 100 messages and 5 ms per handler, the group has meaningful work to justify scaling toward 6 workers. `with_prefetch_count(4)` lets each worker hold up to 4 messages at a time, reducing per-message scheduling overhead. #### Shutdown pattern A background task polls the `count` AtomicUsize every 20 ms and cancels the `CancellationToken` once all 100 messages have been processed — or after a 30-second hard deadline. The final `println!` reports the actual count consumed, which should always be 100 on a successful run. #### Factory closure and `Clone` `group.register` takes a factory `move || Worker { count: c.clone() }`. The closure is called once per spawned worker, so each worker gets its own `Worker` instance with a cloned `Arc` pointing to the shared counter. The `Clone` derive on `Worker` is what enables this — each call to the factory produces a fresh value, but all share the same underlying atomic. ### What to try next * Narrow the range to `new(1..=1)` — all 100 messages are processed sequentially and total wall-clock time grows to roughly 500 ms. * Widen the range to `new(1..=32)` — measure how throughput scales as more workers saturate the queue. * Increase `with_prefetch_count(4)` to `with_prefetch_count(16)` — observe fewer scheduling round-trips per message at the cost of head-of-line blocking if a slow message stalls a worker. * See [Guides: Groups](/guides/groups) for how the group adapts its worker count at runtime. ## InMemory — Sequenced Use this when you want to verify per-key ordering before wiring a real broker. Messages for three accounts are published in interleaved order; each account's entries arrive in strict sequence, proving that ledger-style FIFO delivery works correctly. No external infrastructure required. ### Prerequisites No external services required. The in-memory backend runs entirely in-process. * Cargo feature: `inmemory` ### Run ```sh cargo run --example inmemory_sequenced --features inmemory ``` ### Expected output Output order within a single account is always ascending (`seq=0` before `seq=1`), but entries for different accounts may interleave: ```text applied entry account=alice seq=0 amount=+100 applied entry account=bob seq=0 amount=+100 applied entry account=carol seq=0 amount=+100 applied entry account=alice seq=1 amount=+100 applied entry account=bob seq=1 amount=+100 ... ``` ### Source ```rust // [!include ~/examples/inmemory/sequenced.rs] ``` ### Walkthrough #### Topic definition with sequence key `LedgerTopic` implements both `Topic` and `SequencedTopic`. The topology is built with `.sequenced(SequenceFailure::FailAll)` — if any message for an account fails, all subsequent messages for that account are also failed immediately rather than left queued. `.routing_shards(4)` tells the broker to spread keys across four in-memory shard channels, which improves parallelism when many distinct keys are in flight. `.hold_queue(Duration::from_millis(50))` configures a 50 ms retry delay before a failed message is redelivered. #### Sequence key function `SEQUENCE_KEY_FN` is set to `Some(Self::sequence_key)`, which extracts `msg.account` as a `String`. The broker uses this function to route each `LedgerEntry` to the correct FIFO lane — messages with the same account name always land in the same shard and are processed one at a time. #### Backend-specific consumer path Because per-key FIFO (`run_fifo`) is not yet exposed on the generic `Broker\` / `ConsumerGroup\` API, this example drops down to the concrete `InMemoryConsumer` type directly. `ConsumerOptions::::new().with_prefetch_count(1)` keeps exactly one message in-flight per consumer, which is the correct setting for strict FIFO — allowing more would let the consumer pick up a later message for the same key before finishing the earlier one. #### Shutdown after all messages land The main thread busy-loops on `acked.load(Ordering::Relaxed) < 15` (five sequence numbers × three accounts) before calling `shutdown.cancel()`. This pattern is simple for a bounded benchmark; in production you would instead react to an external signal (see [Guides: Shutdown](/guides/shutdown)). ### What to try next * Change `SequenceFailure::FailAll` to `SequenceFailure::FailOne` — observe that only the individual failing message is retried; subsequent messages for the same account are not held back. * Remove `.routing_shards(4)` — all keys collapse into a single shard; per-key parallelism disappears and throughput drops proportionally. * Add a second hold queue stage with `.hold_queue(Duration::from_secs(1))` — failed entries wait longer before redelivery. * Read [Guides: Sequenced](/guides/sequenced) for a deeper explanation of shard routing and failure modes. ## InMemory — Stress Use this to establish your framework performance baseline before adding a real broker. Everything runs in-process, so results here isolate `shove`'s scheduling and dispatch overhead from network and broker costs — the ceiling against which all durable backends are measured. ### Prerequisites No external services required. The in-memory backend runs entirely in-process. * Cargo feature: `inmemory` ### Run ```sh cargo run --example inmemory_stress --features inmemory ``` You can narrow the run with CLI flags: ```sh # Single tier, single handler profile cargo run --example inmemory_stress --features inmemory -- --tier moderate --handler fast # JSON output cargo run --example inmemory_stress --features inmemory -- --output json ``` Run in release mode for representative numbers: ```sh cargo run -q --release --example inmemory_stress --features inmemory ``` ### Expected output Output is non-deterministic and machine-dependent. Look for the scenario progress lines on stderr and the final table on stdout: ```text shove stress benchmarks — inmemory scenarios: 60 [1/60] moderate | 50000msg | 1c | zero (no-op) ... -> 285000.4 msg/s | dispatch p50=0.0ms p99=0.1ms | e2e p50=0.0ms p99=0.1ms | cpu=98% rss=12.3MB | 0.2s [2/60] moderate | 50000msg | 4c | zero (no-op) ... -> 512000.1 msg/s | ... ... Backend: inmemory TIER MSGS C HANDLER MSG/SEC disp p50 disp p95 disp p99 e2e p50 e2e p95 e2e p99 SCALE RSS(MB) CPU% ... ``` Characteristic markers to look for: * `zero (no-op)` handler throughput in the hundreds of thousands of msg/s — this is the raw scheduling cost. * `scaling_efficiency` column approaching `1.0x` as consumer count grows for CPU-bound profiles. * `peak_rss_mb` staying well under 100 MB even at extreme tiers. ### Source ```rust // [!include ~/examples/inmemory/stress.rs] ``` ### Walkthrough #### Thin backend wrapper `inmemory/stress.rs` is intentionally minimal — it constructs a `HarnessConfig::` with the backend name `"inmemory"`, then calls `run_all_scenarios` from the shared harness in `examples/common/stress_test.rs`. The `make_cfg` closure maps `(consumers, prefetch, _concurrent)` directly to: ```rust InMemoryConsumerGroupConfig::new(consumers..=consumers) .with_prefetch_count(prefetch) ``` Because in-memory has no concurrent-mode distinction, the `_concurrent` parameter is intentionally ignored. #### Harness configuration `HarnessConfig::::new("inmemory")` leaves all defaults: `prefetch_cap` of 100, `publish_chunk_size` of 1000, and a no-op `purge` function. The in-memory backend does not need purging between scenarios because a fresh `Broker\` is created for each scenario — previous messages cannot bleed across. #### Scenario matrix The harness defines three tiers (`moderate`, `high`, `extreme`) and four handler profiles (`zero`, `fast`, `slow`, `heavy`). The `moderate` tier uses consumer counts `[1, 4, 8, 16, 32]`; the `extreme` tier goes up to 256 consumers. Handler latencies range from 0 ms (`zero`) to 1–5 s (`heavy`). Each combination is one scenario, run sequentially — a full `--tier all --handler all` sweep is 60 scenarios on the moderate tier alone. #### Metrics collected For each scenario the harness records per-message dispatch latency (`published_at_ns` to handler entry) and end-to-end latency (`published_at_ns` to handler completion), then reports p50/p95/p99 percentiles. It also samples peak RSS and CPU% via platform-specific APIs (`mach_task_basic_info` on macOS, `/proc/self/statm` on Linux). The `scaling_efficiency` column divides each throughput by the baseline (lowest consumer count for the same tier/handler combination) — a value of `4.0x` with 4 consumers indicates linear scaling. ### What to try next * Run with `--tier extreme --handler zero` to stress-test the scheduler at 1 M messages and 256 consumers. * Add `--concurrent` to enable pipeline processing within each consumer and compare throughput against the default sequential mode. * Pipe output through `--output json` and import into a spreadsheet to plot scaling curves. * Compare numbers against the RabbitMQ or Kafka stress examples to quantify the network + broker overhead relative to this in-process baseline.