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

AWS SNS + SQS

If you're already on AWS and don't want to run a broker, this is your path. SNS fans messages out to SQS queues; shove handles polling, retries, DLQ routing, and consumer supervision. No broker process to provision or maintain — pay AWS for what you use.

What you need

An AWS account with IAM credentials that have SNS and SQS permissions, and a region selected. For local dev, LocalStack Pro emulates both services:

docker run --rm -p 4566:4566 \
  -e LOCALSTACK_AUTH_TOKEN=$LOCALSTACK_AUTH_TOKEN \
  localstack/localstack-pro

Set the endpoint_url field in SnsConfig to http://localhost:4566 when targeting LocalStack. Omit it entirely for real AWS.

Install

cargo add shove --features aws-sns-sqs

Connect

    let broker = Broker::<Sqs>::new(SnsConfig {
        region: Some("us-east-1".into()),
        endpoint_url: Some(endpoint),
    })
    .await?;

SnsConfig holds the AWS region and an optional endpoint_url override for local dev. AWS credentials are loaded from the standard environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION) or from the default credential chain.

Declare topology

    let topology = broker.topology();
    topology.declare::<MinimalOrder>().await?;
    topology.declare::<DlqOrder>().await?;
    topology.declare::<RetryOrder>().await?;

topology().declare::<T>() creates the SNS topic and SQS queue(s), including hold queues and the DLQ. Idempotent — safe to call on every startup.

Publish

    let publisher = broker.publisher().await?;
 
    // Single publish
    let order = OrderEvent {
        order_id: "ORD-001".into(),
        amount_cents: 5000,
    };
    publisher.publish::<MinimalOrder>(&order).await?;
    publisher.publish::<DlqOrder>(&order).await?;
    publisher.publish::<RetryOrder>(&order).await?;
 
    // Publish with custom headers
    let mut headers = HashMap::new();
    headers.insert("x-source".into(), "example".into());
    headers.insert("x-priority".into(), "high".into());
    let order2 = OrderEvent {
        order_id: "ORD-002".into(),
        amount_cents: 9900,
    };
    publisher
        .publish_with_headers::<MinimalOrder>(&order2, headers)
        .await?;
 
    // Batch publish
    let batch = vec![
        OrderEvent {
            order_id: "ORD-003".into(),
            amount_cents: 1000,
        },
        OrderEvent {
            order_id: "ORD-004".into(),
            amount_cents: 2500,
        },
        OrderEvent {
            order_id: "ORD-005".into(),
            amount_cents: 7777,
        },
    ];
    publisher.publish_batch::<MinimalOrder>(&batch).await?;

publisher().await? returns a Publisher<Sqs> that publishes via SNS. SNS fans the message out to all subscribed SQS queues.

Consume

    let mut supervisor = broker.consumer_supervisor();
    supervisor.register::<MinimalOrder, _>(AckHandler, ConsumerOptions::<Sqs>::new())?;
    supervisor.register::<DlqOrder, _>(RejectHandler, ConsumerOptions::<Sqs>::new())?;
    supervisor.register::<RetryOrder, _>(
        RetryHandler,
        ConsumerOptions::<Sqs>::new().with_max_retries(3),
    )?;
 
    // Let everything run, then shut down.
    let outcome = supervisor
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(10)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(5),
        )
        .await;

SQS uses consumer_supervisor() — N independent pollers per registered topic. There is no broker-level coordinated-group primitive on SQS, so consumer_group() is not available on Broker\<Sqs\>. Use ConsumerOptions::<Sqs>::new() or ConsumerOptions::<Sqs>::preset(prefetch_count) to configure polling behavior.

Sequenced delivery

Messages for the same group key stay in order. SQS FIFO topics enforce ordering natively: when you declare a sequenced topic, shove provisions FIFO SNS topics and SQS queues and sets the MessageGroupId from T::sequence_key(); SQS delivers messages in publish order within each group.

FIFO queue names must end in .fifo — shove appends this suffix automatically when sequenced topology is declared. The default SQS FIFO throughput quota is 300 messages/second per queue; for higher throughput, contact AWS to raise the quota or switch to a standard (unordered) topic.

See Sequenced Topics for the full ordering model, and Sequenced Pubsub example for a runnable walkthrough.

Consumer supervision + autoscaling

SQS has no broker-level coordinated-group primitive, so Broker\<Sqs\> only exposes consumer_supervisor() (not consumer_group()). The supervisor fans out one polling task per registered topic with a fixed ConsumerOptions<Sqs>:

use shove::{Broker, ConsumerOptions, Sqs};

let mut supervisor = broker.consumer_supervisor();
supervisor.register::<OrderTopic, _>(
    MyHandler,
    ConsumerOptions::<Sqs>::preset(10),
)?;
let outcome = supervisor.run_until_timeout(signal, drain_timeout).await;

consumer_supervisor() does not scale poller count on its own — register accepts one set of ConsumerOptions and runs exactly one consumer per topic.

Tuning ConsumerOptions

use shove::{ConsumerOptions, Sqs};
use std::time::Duration;

let opts = ConsumerOptions::<Sqs>::new()
    .with_prefetch_count(10)
    .with_receive_batch_size(10)                         // SQS max is 10
    .with_handler_timeout(Duration::from_secs(30))
    .with_max_reconnect_attempts(5);
  • with_receive_batch_size(u16) — messages requested per ReceiveMessage call. SQS's hard max is 10; SQS returns batches up to this size. Default: 0 (= use the prefetch count).
  • with_handler_timeout(Duration) — per-message wall-clock deadline; must be less than the queue's VisibilityTimeout so a slow handler doesn't race the broker re-delivering the same message. Default: 30s.
  • with_max_reconnect_attempts(u32) — cap on SQS client redials before surfacing a Connection error. Default: unlimited.

SqsConsumerGroupConfig exposes the same knobs for the group registry (with_prefetch_count, with_max_retries, with_handler_timeout, with_concurrent_processing).

Autoscaling

For autoscaling within a min..=max range driven by ApproximateNumberOfMessages, use the backend-specific SqsConsumerGroupRegistry directly. It accepts an SqsConsumerGroupConfig::new(min..=max) and spawns/retires independent pollers within that range as queue depth changes. The registry also accepts a service-wide handler timeout default via SqsConsumerGroupRegistry::with_default_handler_timeout(Duration) — applied to every group that did not set its own. See the Consumer Groups example for the full path.

Gotchas

  • Visibility timeout must be greater than handler timeout. If the handler takes longer than the visibility timeout, the message becomes re-visible on the queue and will be delivered again while the original handler is still running. Set both values consciously.
  • FIFO 300 msg/s default cap. High-throughput sequenced topics may require quota increases via the AWS console. Standard (non-FIFO) SQS queues have no ordering guarantee and much higher throughput limits.
  • LocalStack endpoint is for local dev only. The SnsConfig::endpoint_url field should be omitted in production; it exists solely to redirect traffic to a local emulator.
  • Region matters. SQS queues are regional. Cross-region fan-out requires explicit SNS topic ARN configuration — shove does not automatically route across regions.
  • IAM permissions required: sns:CreateTopic, sns:Subscribe, sns:Publish, sqs:CreateQueue, sqs:ReceiveMessage, sqs:DeleteMessage, sqs:GetQueueAttributes, and sqs:SetQueueAttributes at minimum.

Examples

See also