Are you an LLM? Read llms.txt for a summary of the docs, or llms-full.txt for the full context.
Skip to content

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

LOCALSTACK_AUTH_TOKEN=<token> cargo run --example sqs_concurrent_pubsub --features aws-sns-sqs

Expected output

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

//! Concurrent consumption example (SQS backend).
//!
//! Demonstrates `ConsumerOptions::with_concurrent_processing` for non-blocking
//! message processing with in-order acknowledgement. Compares sequential vs
//! concurrent throughput with a slow handler, both driven through the generic
//! `Broker<Sqs>::consumer_supervisor()` path.
//!
//! Spins up a LocalStack testcontainer automatically. Requires a running
//! Docker daemon and the `LOCALSTACK_AUTH_TOKEN` environment variable:
//!
//!     LOCALSTACK_AUTH_TOKEN=... cargo run --example sqs_concurrent_pubsub --features aws-sns-sqs
 
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};
 
use serde::{Deserialize, Serialize};
use shove::sns::SnsConfig;
use shove::{
    Broker, ConsumerOptions, MessageHandler, MessageMetadata, Outcome, Publisher, Sqs,
    TopologyBuilder, define_topic,
};
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::localstack::LocalStack;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Task {
    id: u32,
    payload: String,
}
 
// ─── Topic ──────────────────────────────────────────────────────────────────
 
define_topic!(
    SlowTasks,
    Task,
    TopologyBuilder::new("sqs-concurrent-tasks").dlq().build()
);
 
// ─── Handler ────────────────────────────────────────────────────────────────
 
#[derive(Clone)]
struct SlowTaskHandler {
    processed: Arc<AtomicU32>,
    signal: Arc<tokio::sync::Notify>,
    delay: Duration,
}
 
impl SlowTaskHandler {
    fn new(delay: Duration) -> Self {
        Self {
            processed: Arc::new(AtomicU32::new(0)),
            signal: Arc::new(tokio::sync::Notify::new()),
            delay,
        }
    }
 
    fn count(&self) -> u32 {
        self.processed.load(Ordering::Relaxed)
    }
 
    async fn wait_for(&self, target: u32, timeout: Duration) -> bool {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            if self.count() >= target {
                return true;
            }
            tokio::select! {
                _ = self.signal.notified() => {}
                _ = tokio::time::sleep_until(deadline) => {
                    return self.count() >= target;
                }
            }
        }
    }
}
 
impl MessageHandler<SlowTasks> for SlowTaskHandler {
    type Context = ();
    async fn handle(&self, task: Task, _meta: MessageMetadata, _: &()) -> Outcome {
        // Simulate slow I/O (HTTP call, database query, etc.)
        tokio::time::sleep(self.delay).await;
        eprintln!("  processed task {}", task.id);
        self.processed.fetch_add(1, Ordering::Relaxed);
        self.signal.notify_waiters();
        Outcome::Ack
    }
}
 
// ─── Helpers ────────────────────────────────────────────────────────────────
 
async fn setup(
    endpoint: String,
) -> Result<(Broker<Sqs>, Publisher<Sqs>), Box<dyn std::error::Error>> {
    let broker = Broker::<Sqs>::new(SnsConfig {
        region: Some("us-east-1".into()),
        endpoint_url: Some(endpoint),
    })
    .await?;
    broker.topology().declare::<SlowTasks>().await?;
    let publisher = broker.publisher().await?;
    Ok((broker, publisher))
}
 
async fn publish_tasks(publisher: &Publisher<Sqs>, count: u32) -> Result<(), shove::ShoveError> {
    let messages: Vec<Task> = (0..count)
        .map(|id| Task {
            id,
            payload: format!("task-{id}"),
        })
        .collect();
    publisher.publish_batch::<SlowTasks>(&messages).await
}
 
async fn run_round(
    broker: &Broker<Sqs>,
    handler: SlowTaskHandler,
    prefetch: u16,
    concurrent: bool,
    msg_count: u32,
) -> Duration {
    let mut supervisor = broker.consumer_supervisor();
    supervisor
        .register::<SlowTasks, _>(
            handler.clone(),
            ConsumerOptions::<Sqs>::new()
                .with_prefetch_count(prefetch)
                .with_concurrent_processing(concurrent),
        )
        .expect("register");
 
    let start = Instant::now();
 
    let signal_handler = handler.clone();
    let outcome = supervisor
        .run_until_timeout(
            async move {
                signal_handler
                    .wait_for(msg_count, Duration::from_secs(120))
                    .await;
            },
            Duration::from_secs(10),
        )
        .await;
    let duration = start.elapsed();
    assert!(
        outcome.is_clean(),
        "supervisor reported errors: {outcome:?}"
    );
    duration
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let auth_token = match std::env::var("LOCALSTACK_AUTH_TOKEN") {
        Ok(t) => t,
        Err(_) => {
            eprintln!(
                "LOCALSTACK_AUTH_TOKEN is not set. This example requires a LocalStack Pro auth \
                 token:\n\n    export LOCALSTACK_AUTH_TOKEN=...\n"
            );
            std::process::exit(1);
        }
    };
 
    // SAFETY: called before any concurrent env access in this process.
    unsafe {
        std::env::set_var("AWS_ACCESS_KEY_ID", "test");
        std::env::set_var("AWS_SECRET_ACCESS_KEY", "test");
        std::env::set_var("AWS_REGION", "us-east-1");
    }
 
    let container = LocalStack::default()
        .with_env_var("LOCALSTACK_AUTH_TOKEN", auth_token)
        .start()
        .await?;
    let port = container.get_host_port_ipv4(4566).await?;
    let endpoint = format!("http://localhost:{port}");
 
    let msg_count = 20u32;
    let handler_delay = Duration::from_millis(100);
    let prefetch = 20u16;
 
    let (broker, publisher) = setup(endpoint).await?;
 
    // ── Sequential run ──────────────────────────────────────────────────
 
    eprintln!("--- Sequential (prefetch_count=1) ---");
    eprintln!("  {msg_count} messages, {handler_delay:?} handler delay, prefetch=1");
 
    publish_tasks(&publisher, msg_count).await?;
 
    let handler = SlowTaskHandler::new(handler_delay);
    let sequential_dur = run_round(&broker, handler, 1, false, msg_count).await;
 
    let seq_throughput = msg_count as f64 / sequential_dur.as_secs_f64();
    eprintln!(
        "  done: {:.1}s ({:.0} msg/s)\n",
        sequential_dur.as_secs_f64(),
        seq_throughput
    );
 
    // ── Concurrent run ──────────────────────────────────────────────────
    // The sequential run drained all messages, so we publish a fresh batch.
 
    eprintln!("--- Concurrent (prefetch_count={prefetch}) ---");
    eprintln!("  {msg_count} messages, {handler_delay:?} handler delay, prefetch={prefetch}");
 
    publish_tasks(&publisher, msg_count).await?;
 
    let handler = SlowTaskHandler::new(handler_delay);
    let concurrent_dur = run_round(&broker, handler, prefetch, true, msg_count).await;
 
    let conc_throughput = msg_count as f64 / concurrent_dur.as_secs_f64();
    eprintln!(
        "  done: {:.1}s ({:.0} msg/s)\n",
        concurrent_dur.as_secs_f64(),
        conc_throughput
    );
 
    // ── Summary ─────────────────────────────────────────────────────────
 
    let speedup = sequential_dur.as_secs_f64() / concurrent_dur.as_secs_f64();
    eprintln!("--- Summary ---");
    eprintln!(
        "  sequential:  {:.1}s ({:.0} msg/s)",
        sequential_dur.as_secs_f64(),
        seq_throughput
    );
    eprintln!(
        "  concurrent:  {:.1}s ({:.0} msg/s)",
        concurrent_dur.as_secs_f64(),
        conc_throughput
    );
    eprintln!("  speedup:     {speedup:.1}x");
    eprintln!();
    eprintln!("  Messages are always acked in delivery order.");
    eprintln!("  Concurrent mode overlaps handler I/O within a single consumer.");
 
    broker.close().await;
    drop(container);
    Ok(())
}

Walkthrough

Handler with notification signal

SlowTaskHandler uses an Arc<tokio::sync::Notify> 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:

ConsumerOptions::<Sqs>::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 for higher-volume concurrency measurements.