Queues & Pub/Sub

Queues & Pub/Sub

A busy checkout service should not wait for every receipt, report, and warehouse integration. It can place work on a queue or announce an event to subscribers. The choice affects who receives a copy, how work accumulates, and what a retry means.

This lesson builds four examples: a polled SQS worker, SNS fanout with queue permissions, FIFO order updates, and Lambda batch processing. It also covers visibility, dead-letter queues, payload limits, filtering, and request costs. Examples assume one AWS account in us-east-1, current boto3 or AWS SDK for JavaScript v3, configured credentials, and appropriate IAM permissions. Replace illustrative resource names and ARNs with your resources.

1. Choose the delivery shape

                         SNS Standard topic
                       /          |          \
                payment queue  stock queue  audit queue
                 /      \          |            |
              worker A worker B  worker C     worker D

Payment workers A and B share payment work. Stock and audit receive their own copies. This does not guarantee that all four workers finish together, or that an individual message is processed only once. A worker can repeat an effect after failing before acknowledgement.

SQS is polled and retains unexpired messages until deletion or redrive. SNS pushes and has protocol-specific retries; it is not “deliver immediately or lose the message.” An SNS subscription DLQ handles SNS delivery failures, while a source SQS queue's DLQ handles messages repeatedly received without successful removal. These are different failure boundaries. SQS overview, SNS delivery retries, SNS DLQs.

2. Standard and FIFO are different tradeoffs

FIFO does not make a charge exactly once. A worker can charge, crash before deletion, and receive the message again after visibility expires. Standard queues can also deliver duplicates during visibility. Use stable business operation IDs and atomic or recoverable idempotency mechanisms in either case. Standard queues, FIFO deduplication, outage recovery.

The familiar 300 transactions/second and 3,000 messages/second with full ten-message batches describe an important non-high-throughput FIFO case, not every FIFO configuration. High-throughput FIFO has Region-dependent quotas and requires appropriate group distribution. Batching also reduces request overhead; it does not remove ordering constraints. FIFO quotas, high-throughput FIFO.

3. Visibility, polling, retention, and redrive

Receiving does not delete a message. A custom worker calls DeleteMessage using the receipt handle from the latest receive after successful processing. Receipt handles change on subsequent receives; a transport message ID is not a receipt handle. An old handle can produce a successful API response without deleting the message, so do not rely on an error as protection against stale acknowledgements. DeleteMessage, queue attributes, message timers.

0s   ReceiveMessage; this attempt gets receipt handle H1
30s  visibility expires if the worker has not deleted or extended it
35s  another receive may return the message with a new handle H2
45s  first worker finishes, but its effect may now have been repeated

Expiration makes another receive possible; it does not guarantee one happens at exactly 30 seconds. Choose a worker timeout based on processing variation, not just the average, and renew visibility for long jobs. Calling ChangeMessageVisibility with 300 seconds starts that remaining interval from the call, rather than adding 300 seconds to the previous deadline. Extensions cannot exceed twelve hours from the original receive. Visibility timeout, change visibility.

Long polling returns when messages are available or the wait expires. A response can contain fewer than the requested number, including zero. Set the client's HTTP response/read timeout longer than the long-poll wait. Twenty seconds often reduces empty responses for idle queues; it does not impose a twenty-second delay on every arriving message. Long polling.

A redrive policy's maxReceiveCount is based on receives, not a log of business exceptions. Monitor the DLQ and investigate before redriving. Its retention remains finite. For Standard queues, the original enqueue timestamp is retained when a message moves to the DLQ, so set DLQ retention longer than the source's. FIFO enqueue timestamps reset on that move. Removing a FIFO message can break the business sequence, so decide whether to block and repair instead. SQS DLQs.

4. Application: a producer and a polled worker

Suppose order-work is an existing Standard queue. This JavaScript module publishes a persisted logical event. The event ID and payload remain unchanged across retries. Import it from an application that awaits the returned promise and records/retries failed publication; a database write followed by this call still needs an outbox if neither may be lost.

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({ region: "us-east-1" });
const queueUrl = "https://sqs.us-east-1.amazonaws.com/123456789012/order-work";

export async function sendOrder(orderEvent) {
  return sqs.send(new SendMessageCommand({
    QueueUrl: queueUrl,
    MessageBody: JSON.stringify(orderEvent),
    MessageAttributes: {
      eventType: { DataType: "String", StringValue: "OrderPlaced" }
    }
  }));
}

One example orderEvent is:

{"eventId":"evt-1001-created","eventType":"OrderPlaced","orderId":"1001","amountCents":9999,"currency":"USD"}

The worker below performs one polling pass. Its caller supplies an idempotent processOrder function and decides how to loop, back off, stop, and report errors. Receiving one message keeps this teaching example from holding ten messages invisible while processing them sequentially. The client read timeout must exceed the twenty-second poll.

import {
  SQSClient, ReceiveMessageCommand, DeleteMessageCommand
} from "@aws-sdk/client-sqs";

const sqs = new SQSClient({ region: "us-east-1" });
const queueUrl = "https://sqs.us-east-1.amazonaws.com/123456789012/order-work";

export async function pollOnce(processOrder) {
  const result = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: queueUrl, MaxNumberOfMessages: 1,
    WaitTimeSeconds: 20, VisibilityTimeout: 120,
    MessageAttributeNames: ["All"]
  }));
  for (const message of result.Messages ?? []) {
    const event = JSON.parse(message.Body);
    await processOrder(event);
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: queueUrl, ReceiptHandle: message.ReceiptHandle
    }));
  }
}

If parsing or processing throws, deletion is skipped; the caller should log the failure and the message can become visible again. A DLQ policy is needed for repeatedly bad messages. If deletion fails after processing succeeds, the business effect is already committed and a later retry must recover that result. The 120-second visibility value is an example budget, not a measured guarantee; longer work needs renewal or another design. JavaScript SQS examples.

Try it: inject a failure immediately after processOrder returns. Verify that a second attempt does not charge again, reserve stock again, or overwrite a later order state.

5. Application: SNS fanout that can actually reach its queues

Creating a subscription does not by itself authorize SNS to send to SQS. Each queue needs a resource policy allowing sns.amazonaws.com to call sqs:SendMessage, constrained to the intended topic. The following one-time setup creates a Standard topic and three dedicated Standard queues in the same account. It sets a complete policy for these new queues; merge policies carefully when adapting it to existing shared queues. SNS-to-SQS permissions.

import json
import time
import boto3

sns = boto3.client("sns", region_name="us-east-1")
sqs = boto3.client("sqs", region_name="us-east-1")

def create_fanout():
    topic = sns.create_topic(Name="order-events")["TopicArn"]
    subscriptions = {}
    for name in ("payment-events", "stock-events", "audit-events"):
        url = sqs.create_queue(QueueName=name)["QueueUrl"]
        time.sleep(1)  # Minimum documented wait before using a new queue.
        arn = sqs.get_queue_attributes(
            QueueUrl=url, AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]
        policy = {
            "Version": "2012-10-17",
            "Statement": [{
                "Effect": "Allow",
                "Principal": {"Service": "sns.amazonaws.com"},
                "Action": "sqs:SendMessage", "Resource": arn,
                "Condition": {"ArnEquals": {"aws:SourceArn": topic}}
            }]
        }
        sqs.set_queue_attributes(
            QueueUrl=url, Attributes={"Policy": json.dumps(policy)})
        subscription = sns.subscribe(
            TopicArn=topic, Protocol="sqs", Endpoint=arn,
            Attributes={"RawMessageDelivery": "true"},
            ReturnSubscriptionArn=True)["SubscriptionArn"]
        subscriptions[name] = {"queueUrl": url, "subscriptionArn": subscription}
    return topic, subscriptions

The helper waits the documented minimum of one second before using each new queue. Deployment automation should also use bounded retries for transient readiness failures, wait for policy propagation, and test each subscription before enabling the producer. CreateQueue readiness. Production configuration also needs worker permissions, redrive policies, alarms, and applicable KMS permissions when using customer-managed encryption keys. Publisher permission to call SNS is separate from the queue resource policy.

These subscriptions enable raw delivery, so JSON.parse(sqsRecord.body) gives the published order object. With default raw delivery disabled, first parse the SNS envelope, then parse its Message string. Raw delivery to SQS supports at most ten message attributes; SNS discards messages with more than ten attributes for those raw subscriptions. Raw delivery details.

Here is a filter for the payment subscription that uses the JSON body, followed by a matching publication. topic and subscriptions are returned by the setup function.

def configure_payment_filter(topic, subscriptions):
    subscription_arn = subscriptions["payment-events"]["subscriptionArn"]
    sns.set_subscription_attributes(
        SubscriptionArn=subscription_arn,
        AttributeName="FilterPolicyScope", AttributeValue="MessageBody")
    sns.set_subscription_attributes(
        SubscriptionArn=subscription_arn, AttributeName="FilterPolicy",
        AttributeValue=json.dumps({
            "eventType": ["OrderPlaced"],
            "currency": ["USD"],
            "amountCents": [{"numeric": [">", 0]}]
        }))

def publish_order(topic, persisted_event):
    return sns.publish(TopicArn=topic, Message=json.dumps(persisted_event))

After the filter has propagated, the earlier event with eventType: OrderPlaced, currency: USD, and numeric amountCents: 9999 matches. A zero-value order does not. All named filter properties must satisfy the policy. Attribute-based filtering is also supported and is the default scope; in that case publish the corresponding SNS message attributes. Filter changes can take up to fifteen minutes, and filtering is independent of raw delivery. Applying filters, numeric matching.

SNS publication acceptance is not business completion. Its retries reduce delivery loss, but SQS adds an application-controlled work backlog, polling rate, visibility, and redrive. Direct SNS-to-Lambda can still be appropriate when that queue behavior is unnecessary. Optional SNS FIFO archives also support replay; do not infer that all SNS topics have permanent message history. FIFO archives.

6. Application: order updates using FIFO groups

Use one group per order when an order's updates must remain in sequence and different orders can proceed independently. Grouping by customer instead is valid when the customer's orders must be serialized together, but it reduces independence. Grouping everything under one ID serializes the entire workload.

def create_fifo_queue():
    queue_url = sqs.create_queue(
        QueueName="order-updates.fifo",
        Attributes={
            "FifoQueue": "true",
            "ContentBasedDeduplication": "false",
            "DeduplicationScope": "messageGroup",
            "FifoThroughputLimit": "perMessageGroupId"
        })["QueueUrl"]
    time.sleep(1)
    return queue_url

def send_update(queue_url, persisted_event):
    return sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps(persisted_event),
        MessageGroupId=persisted_event["orderId"],
        MessageDeduplicationId=persisted_event["eventId"])

Both functions reuse the boto3 clients from the previous example. The high-throughput settings are paired intentionally. The publisher must send an order's updates in the intended sequence; FIFO cannot infer a business version that the producer submitted out of order. Each distinct transition gets a distinct persisted event ID; retrying that transition reuses its ID. An ID such as orderId-status may collapse two legitimate visits to the same status within the deduplication window.

With ContentBasedDeduplication=true, SQS can derive a SHA-256 deduplication ID from the message body, excluding message attributes. An explicit MessageDeduplicationId overrides that generated value. Suppression lasts five minutes; neither explicit nor content-based deduplication is a permanent processed-event ledger. FIFO identifiers, SendMessage.

FIFO also cannot make invalid business operations valid. Starting at $0, deposit $100, withdraw $100, then withdraw $50 still attempts an overdraft even in perfect order. The account service must enforce its balance rule. Ordering and validation solve different problems.

7. Application: Lambda processing with partial batch responses

With an SQS event source mapping, Lambda polls and deletes successfully processed messages. The handler normally does not call DeleteMessage. By default, a failed invocation makes the whole batch eligible for retry after visibility expires, including records whose effects already succeeded. Lambda with SQS.

The handler below records immutable events in a DynamoDB table OrderEventLedger, partition key eventId (string), with no expiry for this exercise. It is deliberately a ledger, not a payment processor or a mutable order-status table. A conditional insert prevents a duplicate event from overwriting an existing row. On a duplicate ID, it verifies that the payload agrees; a conflicting reuse of an ID is a failure.

The contract requires a nonempty event ID, order ID, OrderPlaced, USD, and positive integer cents. The UTF-8 event ID must fit DynamoDB's 2,048-byte partition-key limit, and the entire two-string-attribute item must fit its 400 KiB limit, including attribute names and values. Extra fields are retained in the canonical payload and count toward that limit. A message fitting SQS can still be too large for this ledger; use a small immutable S3 reference for larger data. DynamoDB constraints. allow_nan=False rejects non-JSON numeric values; floating-point amounts are rejected by the integer-cents check.

import json
import boto3
from botocore.exceptions import ClientError

ledger = boto3.resource("dynamodb", region_name="us-east-1").Table("OrderEventLedger")

def record_event(payload):
    if not isinstance(payload, dict):
        raise ValueError("Expected an object")
    for field in ("eventId", "orderId"):
        if not isinstance(payload.get(field), str) or not payload[field]:
            raise ValueError("Missing event identity")
    if (payload.get("eventType") != "OrderPlaced" or
        payload.get("currency") != "USD" or
        type(payload.get("amountCents")) is not int or
        payload["amountCents"] <= 0):
        raise ValueError("Invalid order contract")
    if len(payload["eventId"].encode("utf-8")) > 2048:
        raise ValueError("Event ID exceeds partition-key limit")
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"),
                           ensure_ascii=False, allow_nan=False)
    key = {"eventId": payload["eventId"]}
    item = {**key, "payload": canonical}
    item_bytes = sum(len(k.encode("utf-8")) + len(v.encode("utf-8"))
                     for k, v in item.items())
    if item_bytes > 400 * 1024:
        raise ValueError("Event exceeds ledger item limit; use a payload reference")
    try:
        ledger.put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(eventId)")
    except ClientError as error:
        if error.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise
        existing = ledger.get_item(Key=key, ConsistentRead=True).get("Item")
        if existing is None or existing.get("payload") != canonical:
            raise ValueError("Conflicting or unavailable event identity")

def lambda_handler(event, context):
    failures = []
    for record in event["Records"]:
        try:
            record_event(json.loads(record["body"]))
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

The broad catch exists at the per-record batch boundary, so a failure can be reported without retrying successful Standard-queue records. Add structured error logging and alarms in deployment; do not log sensitive whole payloads. A function crash or timeout can still fail the whole invocation. Catching an exception does not mean the operation failed before making an effect; the conditional ledger handles that uncertainty on retry. Adding an external charge after the insert would reintroduce a separate transaction boundary.

Configure the event source mapping to interpret this response. For the example, the function has a 30-second timeout, the Standard queue has a 185-second visibility timeout, its redrive policy uses a compatible DLQ with maxReceiveCount at least 5, and the function role has SQS receive/delete/get-attributes plus ledger write/read permissions. Applicable KMS permissions are also required.

lambda_client = boto3.client("lambda", region_name="us-east-1")

def attach_queue(function_arn, queue_arn):
    return lambda_client.create_event_source_mapping(
        FunctionName=function_arn,
        EventSourceArn=queue_arn,
        BatchSize=10,
        MaximumBatchingWindowInSeconds=5,
        FunctionResponseTypes=["ReportBatchItemFailures"],
        Enabled=True)

The recommendation gives 6 × 30 + 5 = 185 seconds. It uses the configured function timeout, not average processing duration. Function and queue must be in the same Region. The timeout must not exceed queue visibility; AWS validates that condition. Mapping configuration.

The invocation's 6 MB payload limit includes record metadata, so the effective record count can be smaller. At low traffic, Lambda may wait up to twenty seconds even with a shorter batching window. The handler shown is for Standard queues; do not copy its continue-after-failure loop to FIFO unchanged. Returning batchItemFailures works as intended only when ReportBatchItemFailures is enabled. Mapping parameters, partial failures.

8. Payloads and costs: check the actual bottleneck

SQS currently permits a message size up to 1 MiB (1,048,576 bytes); the queue's configured MaximumMessageSize can be lower. SNS publish messages are limited to 256 KiB (262,144 bytes) by default, including applicable attributes; a topic's MaximumMessageSize attribute can raise that limit, up to 1 MiB. A 500 KiB payload may fit SQS directly but cannot simply pass through SNS in this fanout while the topic keeps that default. A serialized 2 MiB payload exceeds the SQS ceiling. Measure encoded bytes, including applicable message attributes, not character count or an in-memory object estimate. SQS size limits, SNS Publish.

For larger payloads, store an immutable or versioned object in S3 and send a small reference:

{"eventId":"evt-report-1001","payload":{"bucket":"example-order-payloads","key":"reports/evt-report-1001.json","versionId":"example-version-id"}}

The consumer must have permission to read the specified version and, where applicable, decrypt it. Retain the object for the entire retry/redrive/replay horizon. Overwriting an unversioned key can cause the same queued pointer to produce different data later. A checksum can support integrity checking. This is an application payload-reference contract; it does not cause SQS to fetch S3 automatically. Extended client libraries implement their own pointer and cleanup conventions for supported languages. SQS Extended Client, S3 object versions.

For an idle queue, compare 10,000 empty short-poll requests/hour with one continuously repeated twenty-second long poll: 3,600 / 20 = 180 requests/hour, ignoring small client overhead. That is a 98.2% reduction in empty receive request count under these assumptions. Busy queues return early, and other requests still cost money.

At an illustrative marginal rate of $0.40 per million requests, those empty receives would cost $0.004/hour versus $0.000072/hour, or $2.88 versus $0.05184 over 720 hours. This is arithmetic, not a current account quote: Region, queue type, tier, free allowances, payload billing units, and additional services affect charges. SQS bills each 64 KB payload chunk as a request unit; a single large-message API call can use multiple units. Retention does not have a separate per-message storage fee on the SQS pricing model. SQS pricing.

Practice: choose a failure boundary and test it deliberately: wrong queue policy, malformed JSON, a duplicate ID with different data, a ledger write whose response is lost, or a timeout before deletion. Confirm which component retries, what appears in the DLQ, and whether the business state remains correct.