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

Use this when you want to see queue-depth-based autoscaling on SQS end-to-end. Sixty tasks are published at once; with one slow consumer the queue fills faster than it drains, triggering scale-up to five consumers. The autoscaler then scales back once the queue is empty. Run this before production to tune the hysteresis and cooldown settings for your traffic pattern.

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_autoscaler --features aws-sns-sqs

Expected output

Consumer count climbs from 1 to 5 as the queue fills, then falls back as it drains. Exact timing depends on LocalStack latency and system load:

published 60 tasks

consumer group started (min=1, max=5)

autoscaler running — watching consumer count change

[monitor] consumers=1 messages_ready=58 in_flight=2
[monitor] consumers=3 messages_ready=40 in_flight=15
[monitor] consumers=5 messages_ready=18 in_flight=25
[monitor] consumers=5 messages_ready=0 in_flight=5
[monitor] consumers=2 messages_ready=0 in_flight=0
[monitor] consumers=1 messages_ready=0 in_flight=0
...
shutting down…
done

Source

//! SQS Autoscaler example.
//!
//! Demonstrates: `SqsAutoscaler`, `AutoscalerConfig`, `SqsConsumerGroupRegistry`,
//! and dynamic scaling of SQS consumer groups based on queue depth.
//!
//! Note: the autoscaler + consumer-group registry machinery isn't yet
//! surfaced on the generic `Broker<Sqs>` wrapper, so this example stays on
//! the backend-specific `SqsConsumerGroupRegistry` / `SqsAutoscalerBackend`
//! path (same as the stress harness).
//!
//! The autoscaler polls SQS queue attributes on a configurable interval and
//! scales consumer groups up or down using hysteresis (condition must be
//! sustained for `hysteresis_duration`) and cooldown (minimum gap between
//! consecutive scaling actions) to prevent flapping.
//!
//! 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_autoscaler --features aws-sns-sqs
 
use std::sync::Arc;
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::sns::*;
use shove::*;
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::localstack::LocalStack;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct TaskEvent {
    task_id: String,
    payload: String,
}
 
// ─── Topic ──────────────────────────────────────────────────────────────────
 
define_topic!(
    WorkQueue,
    TaskEvent,
    TopologyBuilder::new("sqs-autoscale-work").dlq().build()
);
 
// ─── Handler ────────────────────────────────────────────────────────────────
 
#[derive(Clone)]
struct TaskHandler;
 
impl MessageHandler<WorkQueue> for TaskHandler {
    type Context = ();
    async fn handle(&self, msg: TaskEvent, meta: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[worker] task={} attempt={}",
            msg.task_id,
            meta.retry_count + 1,
        );
        // Simulate slow work so messages queue up and trigger scale-up.
        tokio::time::sleep(Duration::from_millis(500)).await;
        Outcome::Ack
    }
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "shove=debug,sqs_autoscaler=debug".parse().unwrap()),
        )
        .init();
 
    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 config = SnsConfig {
        region: Some("us-east-1".into()),
        endpoint_url: Some(endpoint),
    };
    let client = SnsClient::new(&config).await?;
 
    // ── Declare topology and publish a burst ──
    // The declarer reads the client-owned topic/queue registries shared with
    // every publisher and consumer group built from the same client.
    let declarer = SnsTopologyDeclarer::new(client.clone());
    declarer.declare(WorkQueue::topology()).await?;
 
    let publisher = SnsPublisher::new(client.clone(), client.topic_registry().clone());
    let burst_size = 60usize;
    for i in 0..burst_size {
        publisher
            .publish::<WorkQueue>(&TaskEvent {
                task_id: format!("TASK-{i:03}"),
                payload: format!("work item {i}"),
            })
            .await?;
    }
    println!("published {burst_size} tasks\n");
 
    // ── Set up consumer group registry ──
    //
    // Consumer groups manage identical consumers reading from the same queue.
    // `register` declares the topology and prepares the group; `start_all`
    // starts consumers at their minimum count (1 here).
    let mut registry = SqsConsumerGroupRegistry::new(client.clone());
 
    registry
        .register::<WorkQueue, TaskHandler>(
            SqsConsumerGroupConfig::new(1..=5) // scale between 1 and 5 consumers
                .with_prefetch_count(5) // each consumer holds up to 5 in-flight messages
                .with_max_retries(3),
            || TaskHandler,
            (),
        )
        .await?;
 
    registry.start_all();
    println!("consumer group started (min=1, max=5)\n");
 
    let registry = Arc::new(Mutex::new(registry));
 
    // ── Set up SqsAutoscaler ──
    //
    // The autoscaler polls SQS `GetQueueAttributes` and computes:
    //   capacity = active_consumers × prefetch_count
    //   scale up   when messages_ready > capacity × scale_up_multiplier
    //   scale down when messages_ready < capacity × scale_down_multiplier
    //
    // Hysteresis prevents flapping: the condition must be sustained for
    // `hysteresis_duration` before action is taken.
    // Cooldown prevents back-to-back scaling: at least `cooldown_duration`
    // must elapse between two scaling actions for the same group.
    let stats_provider =
        SqsQueueStatsProvider::new(client.clone(), client.queue_registry().clone());
 
    let auto_config = AutoscalerConfig {
        poll_interval: Duration::from_secs(2),
        // Scale up when messages_ready > capacity × 1.5
        scale_up_multiplier: 1.5,
        // Scale down when messages_ready < capacity × 0.3
        scale_down_multiplier: 0.3,
        // Condition must hold for 4 s before acting
        hysteresis_duration: Duration::from_secs(4),
        // At least 8 s between two scaling actions
        cooldown_duration: Duration::from_secs(8),
    };
 
    let mut autoscaler =
        SqsAutoscalerBackend::autoscaler(stats_provider, registry.clone(), auto_config);
    let shutdown = CancellationToken::new();
 
    let s = shutdown.clone();
    let autoscaler_task = tokio::spawn(async move {
        autoscaler.run(s).await;
    });
 
    // ── Monitor — watch consumer count change as load spikes then drains ──
    //
    // Phase 1: queue is deep → autoscaler scales up toward max_consumers.
    // Phase 2: queue drains  → autoscaler scales back down to min_consumers.
    println!("autoscaler running — watching consumer count change\n");
 
    let monitor_stats = SqsQueueStatsProvider::new(client.clone(), client.queue_registry().clone());
    for _ in 0..30 {
        tokio::time::sleep(Duration::from_secs(3)).await;
 
        let stats = monitor_stats
            .get_queue_stats(WorkQueue::topology().queue())
            .await;
        let active = {
            let reg = registry.lock().await;
            reg.groups()
                .values()
                .next()
                .map(|g| g.active_consumers())
                .unwrap_or(0)
        };
 
        match stats {
            Ok(s) => println!(
                "[monitor] consumers={active} messages_ready={} in_flight={}",
                s.messages_ready, s.messages_not_visible,
            ),
            Err(e) => eprintln!("[monitor] failed to fetch stats: {e}"),
        }
    }
 
    // ── Shutdown ──
    println!("\nshutting down…");
    shutdown.cancel();
    let _ = autoscaler_task.await;
    registry.lock().await.shutdown_all().await;
    client.shutdown().await;
    println!("done");
 
    drop(container);
    Ok(())
}

Walkthrough

Consumer group registry

SqsConsumerGroupRegistry::new(client) manages a pool of independent SQS poll workers. registry.register configures WorkQueue with SqsConsumerGroupConfig::new(1..=5).with_prefetch_count(5) — a capacity of consumers × prefetch = 5–25 messages in-flight. registry.start_all() spawns the minimum number of consumers (1); the autoscaler adds workers as needed.

SqsAutoscalerBackend and AutoscalerConfig

SqsAutoscalerBackend::autoscaler(stats_provider, registry, config) wires the autoscaler to the registry. It polls SqsQueueStatsProvider::get_queue_stats on each poll_interval (2 s) and computes:

  • capacity = active_consumers × prefetch_count
  • scale up when messages_ready > capacity × scale_up_multiplier (1.5)
  • scale down when messages_ready < capacity × scale_down_multiplier (0.3)

hysteresis_duration (4 s) requires the condition to hold across multiple polls before acting, preventing flapping caused by momentary spikes. cooldown_duration (8 s) enforces a minimum gap between consecutive scale actions for the same group.

SqsQueueStatsProvider for dual monitoring

The example creates two SqsQueueStatsProvider instances: one inside the autoscaler and one in the monitor loop. Both call GetQueueAttributes independently — the autoscaler uses the stats to make scaling decisions, while the monitor loop prints messages_ready and messages_not_visible (in-flight) so you can correlate queue depth with consumer count changes in the output.

Shutdown sequence

The autoscaler task runs on a CancellationToken. After the monitor loop completes, shutdown.cancel() stops the autoscaler, then registry.shutdown_all() drains all consumer workers, then client.shutdown() closes the AWS client connections. This ordering ensures the autoscaler does not attempt to scale the registry after its workers are already shutting down.

What to try next

  • Raise scale_up_multiplier to 5.0 — the autoscaler waits for a much deeper backlog before adding consumers; observe the queue draining more slowly during the initial burst.
  • Set hysteresis_duration to Duration::ZERO — the autoscaler reacts on every poll cycle; watch for flapping between consecutive scaling decisions.
  • Increase burst_size to 200 — the drain phase becomes longer and the scale-down step is more visible in the monitor output.
  • See the SQS consumer groups example for the same registry without autoscaling.