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:

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::<T>().

Install

cargo add shove --features nats

Connect

    let broker = Broker::<Nats>::new(NatsConfig::new(&url)).await?;

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:

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

    broker.topology().declare::<OrderTopic>().await?;

topology().declare::<T>() 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::<T>() 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:

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():

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::<T>() 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

    let publisher = broker.publisher().await?;
    for i in 0..3 {
        publisher
            .publish::<OrderTopic>(&OrderCreated {
                order_id: format!("ORD-{i}"),
                amount: 99.99 + i as f64,
            })
            .await?;
        println!("Published order ORD-{i}");
    }

publisher().await? returns a Publisher<Nats> that publishes to the JetStream stream. Each message is acknowledged by JetStream before publish returns.

Consume

    let mut group = broker.consumer_group();
    group
        .register::<OrderTopic, _>(
            ConsumerGroupConfig::new(NatsConsumerGroupConfig::new(1..=1)),
            || OrderHandler,
        )
        .await?;
 
    let outcome = group
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(3)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(10),
        )
        .await;

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 for the full ordering model, and Sequenced example 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:

use shove::nats::NatsConsumerGroupConfig;
use shove::{Broker, ConsumerGroupConfig, Nats};

let mut group = broker.consumer_group();
group
    .register::<OrderTopic, _>(
        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 for a complete runnable walkthrough.

Gotchas

Stream provisioning at declaration time. topology().declare::<T>() creates JetStream streams with default storage limits unless you pass nats_stream_config. 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 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 or bind to an infra-owned stream with nats_external_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::<T>() 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::<Nats>::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 (or skipped entirely with nats_external_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 — publish/consume round trip with hold queues and DLQ
Sequenced — subject-shard ordering
Audited Consumer — MessageHandlerExt::audited wrapping
Stress — throughput benchmarking