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

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:

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

//! Sequenced NATS JetStream example — per-key ordering.
//!
//! Spins up a NATS JetStream testcontainer automatically (requires a running
//! Docker daemon):
//!
//!     cargo run -q --example nats_sequenced --features nats
//!
//! Note: per-key FIFO consumption (`run_fifo`) isn't yet surfaced on the
//! generic `Broker<B>` / `ConsumerSupervisor<B>` / `ConsumerGroup<B>`
//! wrappers — this example therefore keeps using the backend-specific
//! `NatsConsumer::run_fifo` directly.
 
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::nats::{NatsClient, NatsConfig, NatsConsumer, NatsPublisher, NatsTopologyDeclarer};
use shove::{
    ConsumerOptions, MessageHandler, MessageMetadata, Nats, Outcome, SequenceFailure,
    SequencedTopic, Topic, TopologyBuilder,
};
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::nats::{Nats as NatsImage, NatsServerCmd};
use tokio_util::sync::CancellationToken;
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct UserEvent {
    user_id: String,
    action: String,
    seq: u32,
}
 
shove::define_sequenced_topic!(
    UserEventTopic,
    UserEvent,
    |msg: &UserEvent| msg.user_id.clone(),
    TopologyBuilder::new("user-events")
        .sequenced(SequenceFailure::Skip)
        .routing_shards(4)
        .hold_queue(Duration::from_secs(1))
        .dlq()
        .build()
);
 
struct UserEventHandler;
 
impl MessageHandler<UserEventTopic> for UserEventHandler {
    type Context = ();
    async fn handle(&self, message: UserEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[user={}] action={} seq={} (retry={})",
            message.user_id, message.action, message.seq, metadata.retry_count
        );
        Outcome::Ack
    }
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();
 
    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 config = NatsConfig::new(&url);
    let client = NatsClient::connect(&config).await?;
 
    // Declare topology
    let declarer = NatsTopologyDeclarer::new(client.clone());
    declarer.declare(UserEventTopic::topology()).await?;
 
    // Publish events for two users
    let publisher = NatsPublisher::new(client.clone()).await?;
    for seq in 0..5u32 {
        for user in &["alice", "bob"] {
            publisher
                .publish::<UserEventTopic>(&UserEvent {
                    user_id: user.to_string(),
                    action: format!("action-{seq}"),
                    seq,
                })
                .await?;
        }
    }
    println!("Published 10 events (5 per user)");
 
    // Consume with FIFO ordering
    let shutdown = CancellationToken::new();
    let shutdown_clone = shutdown.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(5)).await;
        shutdown_clone.cancel();
    });
 
    let consumer = NatsConsumer::new(client.clone());
    let options = ConsumerOptions::<Nats>::new().with_shutdown(shutdown);
    consumer
        .run_fifo::<UserEventTopic, _>(UserEventHandler, (), options)
        .await?;
 
    client.shutdown().await;
    println!("Done.");
    drop(container);
    Ok(())
}

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\<Nats\> API, the example constructs NatsConsumer::new(client) and calls run_fifo::<UserEventTopic, _>(handler, ctx, opts) directly. ConsumerOptions::<Nats>::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) to stop as soon as all 10 events are processed.
Read Guides: Sequenced for the full explanation of shard routing and failure modes.