Two-Tier Fan-Out: SNS to Per-Consumer SQS Queues (Python + boto3)

You have one event — say, order.created — and three services that care about it: the invoicing worker, the notification worker, and the analytics ingester. The naive move is one SQS queue that everyone reads from. That breaks immediately, because SQS is a work queue, not a broadcast channel: once one consumer receives and deletes a message, it’s gone for everyone else.

The pattern that actually works is a two-tier fan-out. Producers publish once to an SNS topic. Each consumer subscribes its own private SQS queue to that topic. SNS handles the broadcast; each queue handles buffering, retries, and failure isolation for exactly one consumer.

                        ┌──> [SQS: invoicing]  ──> invoicing worker ──> [DLQ]
producer ──> [SNS topic]├──> [SQS: notify]     ──> notify worker    ──> [DLQ]
                        └──> [SQS: analytics]  ──> analytics worker ──> [DLQ]

The payoff is operational, not just architectural:

• Deployments decouple. The notify worker can be down for 20 minutes; its queue buffers everything. Invoicing never notices.
• Retry policies decouple. Analytics can tolerate 10 retries over an hour; invoicing might want 3 fast retries then a DLQ page. Each queue sets its own RedrivePolicy.
• Failures decouple. A poison message that crashes the notify worker lands in the notify DLQ. It never touches the other consumers.
• Adding a consumer is zero-touch for producers. New service? New queue, new subscription. The publisher code doesn’t change.

Everything below runs in ap-south-1 under a named profile. All resources are prefixed awspractice- so cleanup is a grep away.

Create the topic and queues

import json
import boto3

session = boto3.Session(profile_name="aws-practice", region_name="ap-south-1")
sns = session.client("sns")
sqs = session.client("sqs")

PREFIX = "awspractice-orders"

topic_arn = sns.create_topic(Name=f"{PREFIX}-events")["TopicArn"]

CONSUMERS = ["invoicing", "notify", "analytics"]

queues = {}
for name in CONSUMERS:
    # DLQ first
    dlq_url = sqs.create_queue(
        QueueName=f"{PREFIX}-{name}-dlq",
        Attributes={"MessageRetentionPeriod": "1209600"},  # 14 days, the max
    )["QueueUrl"]
    dlq_arn = sqs.get_queue_attributes(
        QueueUrl=dlq_url, AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    # Main queue with its own redrive policy
    q_url = sqs.create_queue(
        QueueName=f"{PREFIX}-{name}",
        Attributes={
            "VisibilityTimeout": "60",
            "ReceiveMessageWaitTimeSeconds": "20",  # long polling by default
            "RedrivePolicy": json.dumps({
                "deadLetterTargetArn": dlq_arn,
                "maxReceiveCount": "3" if name == "invoicing" else "10",
            }),
        },
    )["QueueUrl"]
    q_arn = sqs.get_queue_attributes(
        QueueUrl=q_url, AttributeNames=["QueueArn"]
    )["Attributes"]["QueueArn"]

    queues[name] = {"url": q_url, "arn": q_arn}

Two deliberate choices here. ReceiveMessageWaitTimeSeconds: 20 makes long polling the queue default so no consumer accidentally busy-polls. And the redrive policy differs per consumer — that’s the entire point of the pattern. Invoicing fails fast to a DLQ you alert on; analytics grinds through transient failures on its own.

The queue policy — the step everyone misses

Creating a subscription is not enough. SQS queues reject messages from SNS unless the queue’s resource policy explicitly allows sqs:SendMessage from your topic. Skip this and everything looks wired up in the console while zero messages arrive.

for name, q in queues.items():
    policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "AllowSNSPublish",
            "Effect": "Allow",
            "Principal": {"Service": "sns.amazonaws.com"},
            "Action": "sqs:SendMessage",
            "Resource": q["arn"],
            "Condition": {"ArnEquals": {"aws:SourceArn": topic_arn}},
        }],
    }
    sqs.set_queue_attributes(
        QueueUrl=q["url"],
        Attributes={"Policy": json.dumps(policy)},
    )

The Condition on aws:SourceArn matters: without it, any SNS topic in any account could push into your queue. Scope it to the one topic.

Subscribe with raw message delivery

for name, q in queues.items():
    sns.subscribe(
        TopicArn=topic_arn,
        Protocol="sqs",
        Endpoint=q["arn"],
        Attributes={"RawMessageDelivery": "true"},
    )

RawMessageDelivery: true is not optional in practice. Without it, SNS wraps your payload in its own JSON envelope, and your consumers end up doing json.loads(json.loads(body)["Message"]) — the double-decode that shows up in every SNS/SQS codebase that skipped this flag. With raw delivery, the queue body is exactly what the producer published.

The producer publishes once

def publish_order_created(order_id: str, amount: int) -> None:
    sns.publish(
        TopicArn=topic_arn,
        Message=json.dumps({
            "event": "order.created",
            "order_id": order_id,
            "amount": amount,
        }),
        MessageAttributes={
            "event_type": {"DataType": "String", "StringValue": "order.created"}
        },
    )

publish_order_created("ord_1042", 2599)

That’s the full producer contract. It doesn’t know how many consumers exist, which queues they use, or whether they’re currently up. The MessageAttributes entry isn’t decorative — it’s what SNS filter policies match on later, if a consumer only wants a subset of events (e.g., the notify queue subscribing only to order.created and order.shipped while analytics takes everything).

Each consumer owns its loop

def run_worker(queue_url: str, handler) -> None:
    while True:
        resp = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20,
        )
        for msg in resp.get("Messages", []):
            payload = json.loads(msg["Body"])
            handler(payload)  # raise on failure — do NOT catch-and-pass
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=msg["ReceiptHandle"],
            )

The retry mechanism is the absence of delete_message. If handler raises, the message is never deleted; after the visibility timeout it reappears, ApproximateReceiveCount increments, and once it crosses maxReceiveCount SQS moves it to the DLQ automatically. You don’t write retry code — you write a handler that fails loudly and a redrive policy that decides what “too many failures” means for this consumer.

The tempting anti-pattern is wrapping handler(payload) in a bare try/except that logs and continues to the delete. That converts your at-least-once pipeline into “at most once, silently.”

The guarantees you actually get

Worth being precise, because this is where architectures quietly overpromise:

• At-least-once, not exactly-once. SQS standard queues can deliver a message twice, and a worker crash after processing but before delete_message re-delivers legitimately. Handlers must be idempotent — an INSERT ... ON CONFLICT DO NOTHING keyed on order_id, a Redis SET NX dedupe key, or a state check before side effects.
• No ordering. Standard queues make no ordering promise, and fan-out through SNS doesn’t add one. If a consumer needs ordering per entity, that’s SNS FIFO → SQS FIFO with MessageGroupId — a different (lower-throughput, higher-cost) setup, and most consumers here don’t need it.
• Independent lag. Each queue’s ApproximateNumberOfMessagesVisible is a per-consumer health metric. One dashboard line per consumer tells you exactly who is falling behind, which a shared queue can never tell you.

Prove the decoupling

The whole claim of this pattern is testable in two minutes. Publish five events. Run only the invoicing worker. It processes five messages; meanwhile awspractice-orders-notify shows five messages sitting in it, untouched. Start the notify worker ten minutes later — it drains the backlog with no producer involvement, no replay tooling, nothing. That’s the property you’re buying.

Then test failure isolation: make the notify handler raise unconditionally and publish one event. Watch the message cycle visible → in flight → visible while ApproximateReceiveCount climbs, then land in the notify DLQ after the tenth receive. Invoicing and analytics processed the same event on their first attempt.

Troubleshooting

Messages publish successfully but never arrive in any queue. Queue policy is missing or wrong. sns.publish returns 200 because SNS accepted the message; delivery to the queue then fails silently against the queue’s resource policy. Check the policy’s Resource is the queue ARN (not URL) and aws:SourceArn matches the topic ARN exactly.

json.loads(msg["Body"]) returns a dict with Type, MessageId, and Message keys instead of your payload. Raw message delivery is off on that subscription. Set RawMessageDelivery: "true" — it’s per-subscription, so check each one.

Messages go straight to the DLQ without the worker seeing them fail. Your maxReceiveCount is counting receives, not failures. A worker that receives a message, takes longer than VisibilityTimeout to process it, and then can’t delete it (the receipt handle expired) burns a receive each cycle. Raise the visibility timeout above your worst-case processing time — a common rule of thumb is 6× the average handler duration.

One message processed twice. Not a bug. At-least-once delivery working as documented. Fix the handler’s idempotency, not the infrastructure.

AccessDenied on sqs.set_queue_attributes. Your IAM identity needs sqs:SetQueueAttributes on the queue — the SNS→SQS permission (the queue policy) and your own permission to install that policy are two different things.

Where this pattern stops

Honest limits, because “scales down too” cuts both ways:

• Two consumers, one team, one deploy cadence? A single queue with one worker is fine. Fan-out earns its complexity when consumers have different owners, different failure tolerances, or different deploy schedules.
• Need replay of historical events? SNS is fire-and-forget — messages that fanned out yesterday are gone. If consumers need to re-read history, that’s Kinesis or Kafka territory, a genuinely different tool.
• Strict per-entity ordering across all consumers? FIFO everywhere, with the throughput ceiling that implies. Don’t retrofit ordering onto standard queues with timestamps and hope.
• Cost is near-zero at lab scale (SNS: first 1M publishes free; SQS: first 1M requests free, and long polling keeps empty receives cheap) but at serious volume, per-request pricing across N queues multiplies — every event costs one publish plus N deliveries plus N receives.

Cleanup

for sub in sns.list_subscriptions_by_topic(TopicArn=topic_arn)["Subscriptions"]:
    sns.unsubscribe(SubscriptionArn=sub["SubscriptionArn"])
sns.delete_topic(TopicArn=topic_arn)

for name in CONSUMERS:
    for suffix in ("", "-dlq"):
        url = sqs.get_queue_url(QueueName=f"{PREFIX}-{name}{suffix}")["QueueUrl"]
        sqs.delete_queue(QueueUrl=url)

SQS queue deletion takes up to 60 seconds to propagate, and you can’t recreate a queue with the same name for that window — relevant if you’re tearing down and re-running the lab in a loop.

---

The producer publishes once. Every consumer gets its own buffer, its own retry budget, and its own failure boundary. That’s the entire pattern — and it’s the difference between an outage in one worker being a queue-depth graph versus a production incident.