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 facade. See 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 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:

[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:

use metrics_exporter_prometheus::PrometheusBuilder;
use std::net::Ipv4Addr;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    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:

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

• 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:

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:

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:

• "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.

• "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 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 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 elapsed; aborting surviving tasks" (warn) — run_until_timeout deadline elapsed. Fields: timeout_ms.

Header propagation

MessageMetadata::headers is a HashMap<String, String> 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:

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::<Orders>(&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:

[dependencies]
tracing-opentelemetry = "0.x"
opentelemetry = "0.x"

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:

For RabbitMQ specifically: x-message-id is stamped by RabbitMqPublisher on every outgoing message. Handlers can read it for deduplication:

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) 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
• Handlers & Context
• Shutdown & Exit Codes