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

cargo run --example nats_audited_consumer --features nats,audit

Expected output

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

//! Audited consumer example — custom `AuditHandler` that writes to stdout (NATS backend).
//!
//! Demonstrates: `MessageHandlerExt::audited` wrapper, custom `AuditHandler`
//! implementation, trace ID propagation across retries.
//!
//! Spins up a NATS JetStream testcontainer automatically (requires a running
//! Docker daemon):
//!
//!     cargo run --example nats_audited_consumer --features nats,audit
 
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::nats::{NatsConfig, NatsConsumerGroupConfig};
use shove::*;
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::nats::{Nats as NatsImage, NatsServerCmd};
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PaymentEvent {
    payment_id: String,
    amount_cents: u64,
}
 
// ─── Topic ──────────────────────────────────────────────────────────────────
 
define_topic!(
    Payments,
    PaymentEvent,
    TopologyBuilder::new("nats-audited-payments")
        .hold_queue(Duration::from_secs(5))
        .dlq()
        .build()
);
 
// ─── Handler ────────────────────────────────────────────────────────────────
 
struct PaymentHandler;
 
impl MessageHandler<Payments> for PaymentHandler {
    type Context = ();
    async fn handle(&self, msg: PaymentEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[handler] payment={} amount=${:.2} attempt={}",
            msg.payment_id,
            msg.amount_cents as f64 / 100.0,
            metadata.retry_count + 1,
        );
        // Simulate a transient failure on first attempt to show trace ID
        // persisting across retries.
        if metadata.retry_count == 0 {
            Outcome::Retry
        } else {
            Outcome::Ack
        }
    }
}
 
// ─── Custom audit handler ───────────────────────────────────────────────────
 
/// Prints every audit record to stdout as JSON. Clone-able so a fresh instance
/// can be handed to each spawned consumer.
#[derive(Clone, Default)]
struct StdoutAuditHandler;
 
impl AuditHandler<Payments> for StdoutAuditHandler {
    async fn audit(&self, record: &AuditRecord<PaymentEvent>) -> Result<(), ShoveError> {
        let json = serde_json::to_string_pretty(record).map_err(ShoveError::Serialization)?;
        println!("[audit] {json}");
        Ok(())
    }
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let cmd = NatsServerCmd::default().with_jetstream();
    let container = NatsImage::default().with_cmd(&cmd).start().await?;
    let port = container.get_host_port_ipv4(4222).await?;
    let url = format!("nats://localhost:{port}");
 
    let broker = Broker::<Nats>::new(NatsConfig::new(&url)).await?;
    broker.topology().declare::<Payments>().await?;
    println!("topology declared\n");
 
    // ── Publish a payment ──
    let publisher = broker.publisher().await?;
    let event = PaymentEvent {
        payment_id: "PAY-001".into(),
        amount_cents: 4999,
    };
    publisher.publish::<Payments>(&event).await?;
    println!("published payment\n");
 
    // ── Start audited consumer via a consumer group ──
    //
    // `audited(audit)` wraps the handler in the `Audited` decorator — the
    // group sees a normal `MessageHandler<Payments>`, no API changes needed.
    let mut group = broker.consumer_group();
    group
        .register::<Payments, _>(
            ConsumerGroupConfig::new(NatsConsumerGroupConfig::new(1..=1)),
            || PaymentHandler.audited(StdoutAuditHandler),
        )
        .await?;
 
    // Let it process (first attempt retries, second acks), then shut down.
    let outcome = group
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(10)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(10),
        )
        .await;
 
    println!("done");
    std::process::exit(outcome.exit_code());
}

Walkthrough

Consumer group path for audited consumers

Unlike the NATS sequenced example (which drops to NatsConsumer::run_fifo), this example uses the generic Broker\<Nats\> consumer group API. broker.consumer_group() creates a ConsumerGroup\<Nats\>, 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 for the complete AuditRecord schema and production wiring patterns.