Event-Driven & Messaging

Event-Driven & Messaging

An online shop accepts an order. Payment, stock reservation, analytics, and customer notifications all need to react, but they do not need to finish inside the HTTP request. Event-driven architecture separates the fact that something happened from the work that follows. That separation improves flexibility, while making retries, ordering, and incomplete work part of the application's design.

This lesson develops those tradeoffs through four applications: order processing, operational alerts, a travel-booking saga, and an account event store. Examples use Python with boto3 or modern JavaScript. AWS resources, credentials, Region, and least-privilege permissions must be prepared separately unless creation is shown. Monetary examples use integer cents with an explicit currency; they are simplified domain models.

1. Events, commands, queues, and streams

An event reports a fact: OrderPlaced. A command asks an owner to do something: ReserveStock. Naming a command as if it were a completed event makes consumers act on work that has not happened yet. After committing a durable publication intent, a producer can report that the operation is durably accepted with publication pending. It should report broker publication only after the broker accepts the event. The downstream work can then proceed asynchronously.

Request → Order service → durable order + publication intent
                                ↓ relay
                          event distributor
                         ↙        ↓        ↘
                    payments   inventory   analytics

The arrows represent separate deliveries, not simultaneous completion. A distributor does not automatically join the results of independent consumers.

These services can be combined. SNS → separate SQS queues gives multiple applications their own backlog. Putting several worker instances on one queue distributes that queue's work among them; it does not provide each worker with a full copy. Kinesis suits independently replayable stream processing, while a queue suits work acknowledgement and redrive. SQS overview, SNS overview, Kinesis terminology.

2. Delivery guarantees have boundaries

SQS Standard provides at-least-once delivery with best-effort ordering. FIFO queues preserve order within a message group and suppress repeated sends with the same deduplication ID within a five-minute interval. Neither arrangement makes a payment or database update happen exactly once automatically. A consumer can commit an effect, crash before deleting the message, and receive it again. Standard queues, FIFO deduplication, outage recovery.

t=0   worker receives message; visibility timer starts
t=8   business change commits
t=9   worker crashes before DeleteMessage
t=30  visibility expires; message can be received again

The second attempt must recognize the already-completed operation. For Standard queues, visibility also is not an absolute guarantee against duplicate delivery during the timeout. SQS retains messages only for the configured retention period: four days by default, up to fourteen days. Visibility, message quotas.

Idempotency means repeating an operation has the same intended effect as applying it once. Give each logical operation a stable ID, retained across publication retries and redrive. A transport message ID may change when the application republishes the event. Hashing the body can accidentally merge two legitimate, identical-looking operations; adding a new timestamp on every retry can do the opposite.

This sketch is unsafe:

if event ID is not in Redis:
    charge customer
    save event ID in Redis

Two workers can both pass the check, and a crash can occur after the charge but before saving the ID. An atomic claim closes the first race but still needs an in-progress state and recovery for the crash. Prefer an atomic transaction containing both the business mutation and its processed-event record when both are in the same database. For an external payment service, use its supported idempotency key and reconciliation contract. Keep deduplication records for the relevant retry/replay horizon; an arbitrary 24-hour expiry may be too short. Idempotent APIs.

3. SNS fanout and EventBridge routing

An SNS Standard topic can deliver to SQS, Lambda, HTTP/S, and other supported subscriptions. Delivery times and outcomes differ per subscription. SNS has protocol-dependent retry policies; it is not an “immediate delivery or loss” service. A subscription DLQ can preserve messages that SNS cannot deliver when correctly configured. Successful delivery to Lambda is distinct from successful execution of the function. SNS retries, SNS dead-letter queues, Lambda asynchronous errors.

SNS subscription filters can inspect message attributes or a JSON message body, selected by FilterPolicyScope. Raw delivery changes the SQS payload envelope; it does not enable or disable filtering. For example, a body filter on priority and region needs those properties in the body. Publishing only an event_type attribute does not satisfy that filter. Policy changes can take up to fifteen minutes to propagate. Filtering.

EventBridge adds event-bus rules over an event envelope and nested detail. Here is a valid example event as received on a bus:

{
  "version": "0",
  "id": "11111111-2222-4333-8444-555555555555",
  "source": "com.example.orders",
  "detail-type": "OrderPlaced",
  "account": "123456789012",
  "time": "2026-09-15T10:00:00Z",
  "region": "us-east-1",
  "resources": [],
  "detail": {
    "eventId": "evt-order-1001-created",
    "schemaVersion": 1,
    "orderId": "order-1001",
    "amountCents": 9999,
    "currency": "USD",
    "customerTier": "premium"
  }
}

This rule pattern selects premium orders above $100, using the example's USD contract:

{
  "source": ["com.example.orders"],
  "detail-type": ["OrderPlaced"],
  "detail": {
    "currency": ["USD"],
    "customerTier": ["premium"],
    "amountCents": [{"numeric": [">", 10000]}]
  }
}

The $99.99 event does not match. Neither does exactly $100. A JSON string such as "12500" is not the numeric value 12500. Field names and ordinary string matches are case-sensitive. Event patterns, comparison operators.

The following publisher sends one already-persisted event to an existing custom bus. The caller owns durable retry scheduling and retains the same eventId. A transport exception is allowed to propagate; a per-entry failure is turned into an exception as well.

import json
import boto3

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

def publish_order(detail, bus_name="orders-bus"):
    response = events.put_events(Entries=[{
        "EventBusName": bus_name,
        "Source": "com.example.orders",
        "DetailType": "OrderPlaced",
        "Detail": json.dumps(detail, separators=(",", ":"))
    }])
    result = response["Entries"][0]
    if response["FailedEntryCount"]:
        raise RuntimeError(
            f"Event publication failed: {result.get('ErrorCode')}")
    return result["EventId"]

For batches, PutEvents accepts up to ten entries; inspect each result and retry only failed entries. The currently documented total entry size per request must be less than 1 MiB, calculated using the API's entry-size rules. An HTTP 200 response is insufficient evidence of complete success. A nonexistent bus is a special trap: the service can drop the event without increasing FailedEntryCount. Verify the bus, rule, target, and permissions during deployment, and test delivery. PutEvents results and size calculation.

EventBridge replay does not promise the original event order. SNS FIFO archiving/replay is an optional feature, so “SNS has no replay” is too broad. A schema registry documents structures and supports tooling; creating a schema is not automatic runtime validation of every publication. Validate your event contract in producers and consumers. EventBridge archives, SNS FIFO archives, schemas.

4. Application: order processing without a missing handoff

The order service should not write an order and then assume a separate SNS call will succeed. If the process dies between those operations, the order exists but no consumer learns about it. Store the order and an outbox row in the same database transaction, then let a relay publish committed outbox rows. Transactional outbox.

Database transaction: Order(PENDING) + Outbox(OrderPlaced, stable eventId)
                                   ↓ relay
                         SNS Standard order topic
                       ↙              ↘
                payment queue     inventory queue
                       ↓              ↓
                PaymentAccepted   StockReserved
                       ↘              ↙
                  durable order coordinator
                           ↓ only after both succeed
                    OrderReady → fulfillment / confirmation

The coordinator persists each result keyed by order and operation ID. It emits OrderReady once the required conditions are met, using its own transactional publication intent. A failed payment or stock reservation follows an explicit failure path; an order acknowledgement sent immediately after checkout should say “received,” not “paid and ready.” SNS fanout alone does not implement this join.

An outbox relay can publish and crash before marking the row sent. The next run may publish again. Therefore payment and inventory operations remain idempotent. A payment key should identify the charge operation, for example charge:order-1001:v1; a refund is a distinct operation with a distinct key. Use integer cents (9999), not a floating-point multiplication of 99.99, when the downstream contract expects integer minor units.

For this design, subscriptions use raw delivery so workers receive the order JSON directly. With default SNS-to-SQS delivery, workers instead parse the SQS body as an SNS envelope and then parse its Message field. Do not mix those two payload contracts. The next lesson builds the queue policy and subscription settings explicitly. Raw delivery.

Try it: a payment succeeds, inventory is unavailable, and the payment worker times out before acknowledging its message. The retry must recover the same payment result. The coordinator must still wait for inventory or take its defined cancellation/refund path.

5. Application: route operational alarms without guessing their schema

CloudWatch alarm state-change events can reach EventBridge. Match the documented source and detail-type, then distinguish the new alarm state. Avoid assuming every alarm contains metrics[0].metricStat.metric: metric math and composite alarms have different configuration shapes. Alarm events.

For this example, an EventBridge rule selects aws.cloudwatch, CloudWatch Alarm State Change, and detail.state.value = ALARM, and targets this Lambda handler. The topic exists; the handler role permits publishing to it, and the rule is authorized to invoke Lambda. The severity mapping is our team's policy, not an AWS inference from an alarm's name.

import json
import boto3

sns = boto3.client("sns", region_name="us-east-1")
ALERT_TOPIC = "arn:aws:sns:us-east-1:123456789012:ops-alerts"
SEVERITIES = {"CheckoutErrorRateHigh": "critical"}

def alarm_handler(event, context):
    if (event.get("source") != "aws.cloudwatch" or
        event.get("detail-type") != "CloudWatch Alarm State Change"):
        return
    detail = event["detail"]
    if detail["state"]["value"] != "ALARM":
        return
    alarm_name = detail["alarmName"]
    severity = SEVERITIES.get(alarm_name, "warning")
    sns.publish(
        TopicArn=ALERT_TOPIC,
        Message=json.dumps({
            "eventId": event["id"], "alarmName": alarm_name,
            "severity": severity, "reason": detail["state"]["reason"]
        }),
        MessageAttributes={"severity": {
            "DataType": "String", "StringValue": severity
        }})

Subscribers may filter on the severity attribute. An incident-management integration must actually subscribe and be configured; this code does not create an incident by itself. Duplicate notifications are possible, so an incident consumer should correlate using a stable alarm-transition identity. Delivery failure policies exist at separate stages: EventBridge to Lambda, Lambda execution, and SNS to its subscription. Configure and monitor each relevant stage.

For retriable EventBridge target-delivery failures, the default policy retries for up to 24 hours and up to 185 attempts. A target DLQ needs its own permissions; some non-retriable errors can go directly there. After Lambda accepts an asynchronous event, function failures are handled by Lambda's asynchronous processing policy, not by replaying EventBridge's target-delivery retry count. Target retries, EventBridge DLQs.

6. Application: a travel-booking saga

A flight, hotel, and car booking span independent systems. One database transaction cannot roll them all back. A saga coordinates local transactions and uses compensating business actions when the overall operation cannot complete. A compensation may cost money, take time, or itself fail; it is not a time machine. Step Functions can maintain orchestration state and drive retries and catches. Saga orchestration.

Book flight → Book hotel → Book car → Trip confirmed
    │              │          │
 known failure  known failure known failure
    ↓              ↓          ↓
   fail       cancel flight  cancel hotel → cancel flight

The table and diagram describe known outcomes. If the car service accepted the booking but its response was lost, immediately declaring “car failed” could leave a live car reservation outside the compensation list. Every step needs a stable request ID, a supported way to query/retry uncertain results, and durable records of acquired booking IDs. Successful cancellations must also tolerate retry.

A useful persisted saga record contains tripId, current state, per-step operation IDs, confirmed booking IDs, compensation status, and the next retry/reconciliation time. Parallel compensation is only appropriate when business dependencies permit it. A failed cancellation should enter a visible recovery state with an alert, not be silently treated as a successful rollback.

Keep publication separate from booking outcome. If all bookings succeed but publishing TripConfirmed fails, retry the publication from its outbox. Do not catch both failures in one broad block and publish TripFailed simply because the broker was unavailable. Choreography using events is another option, but it still needs explicit correlation, timeouts, and ownership of the saga outcome.

7. Application: rebuild an account from its event history

Event sourcing stores the authoritative sequence of domain changes and derives current state by applying those events. An EventBridge archive is useful for redelivery, but its replay order is not a substitute for an account's versioned event store.

Consider an append-only DynamoDB table AccountEvents with partition key accountId (string) and sort key version (number). Each row contains kind and a JSON-string data field. A separate AccountHeads table contains the latest committed version. The writer must use a transaction that conditionally advances the expected head version and inserts the next event only if that key is absent. This serializes competing writes for one account; a timestamp alone is not a safe sequence number. DynamoDB transactions, conditional writes.

The reader below assumes that writer contract, immutable events, a pre-existing USD account, and no deletion/expiry of its history. Capturing the head first gives a fixed version boundary; querying every page avoids truncating larger histories. This uses a base-table query with strong consistency, not a global secondary index.

import json
import boto3
from boto3.dynamodb.conditions import Key

db = boto3.resource("dynamodb", region_name="us-east-1")
heads = db.Table("AccountHeads")
history = db.Table("AccountEvents")

def load_account(account_id):
    head = heads.get_item(
        Key={"accountId": account_id}, ConsistentRead=True).get("Item")
    if head is None:
        raise KeyError("Unknown account")
    through_version = int(head["version"])
    args = {
        "KeyConditionExpression": Key("accountId").eq(account_id) &
                                  Key("version").between(1, through_version),
        "ConsistentRead": True,
        "ScanIndexForward": True
    }
    records = []
    while True:
        page = history.query(**args)
        records.extend(page.get("Items", []))
        if not page.get("LastEvaluatedKey"):
            break
        args["ExclusiveStartKey"] = page["LastEvaluatedKey"]
    return rebuild(records, through_version)

def rebuild(records, through_version):
    balance = None
    expected = 1
    for row in records:
        if int(row["version"]) != expected:
            raise ValueError("Missing or out-of-order event")
        data = json.loads(row["data"])
        kind = row["kind"]
        if expected == 1:
            if kind != "AccountOpened" or data != {"currency": "USD"}:
                raise ValueError("Expected USD account opening")
            balance = 0
        else:
            cents = data.get("amountCents")
            if type(cents) is not int or cents <= 0:
                raise ValueError("Expected positive integer cents")
            if kind == "MoneyDeposited":
                balance += cents
            elif kind == "MoneyWithdrawn" and cents <= balance:
                balance -= cents
            else:
                raise ValueError("Unknown event or invalid withdrawal")
        expected += 1
    if expected != through_version + 1 or balance is None:
        raise ValueError("Incomplete history")
    return {"currency": "USD", "balanceCents": balance,
            "version": through_version}

For opening, deposit 50,000 cents, withdrawal 10,000 cents, then deposit 2,500 cents, the balance is 42,500 cents ($425). The model forbids overdrafts: the writer must validate a withdrawal against the expected-version state before committing it. The reader's checks detect invalid history; they do not implement the missing writer transaction. A competing writer requires reloading state and revalidating the command, not simply forcing a new version. Query pagination, read consistency.

For publication, use committed immutable event rows as a durable outbox, with a relay/checkpoint design. DynamoDB Streams can trigger low-latency relay work, but stream records expire after 24 hours; retain the event store and a reconciliation path for longer outages. Relay retries may duplicate publications. Preserve the domain eventId and account version so consumers can deduplicate and detect gaps. A successful PutItem followed by a separate PutEvents call alone does not make the two operations atomic. Streams retention.

8. Operational choices that preserve correctness

Use SQS long polling with a positive wait up to twenty seconds to reduce empty responses; twenty is the maximum, not the only valid setting. Standard queues have very high throughput, while FIFO throughput depends on mode, batching, message-group distribution, and Region quotas. Do not memorize 300 transactions/second as a universal FIFO ceiling. Long polling, FIFO throughput.

For ordinary workers, choose visibility from the processing-time distribution plus recovery needs, and renew it if necessary. ChangeMessageVisibility resets the remaining timeout from that call; it does not append seconds to the original deadline. The twelve-hour overall visibility limit is measured from the original receive. For Lambda SQS mappings, AWS recommends at least six times the configured function timeout plus the batching window. For example, a 30-second function and five-second window suggest at least 185 seconds. Visibility API, Lambda SQS configuration.

A queue's maxReceiveCount governs receives before redrive; it is not a counter of confirmed business exceptions. A DLQ stores diagnostic/recovery work until its retention rules expire it. For Standard queues, expiry still uses the original enqueue timestamp, so time in the source queue counts; configure longer DLQ retention than source retention. FIFO resets the enqueue timestamp when the message moves to the DLQ. Alarm on its backlog and inspect failures before redriving; putting a poison message straight back into the source can repeat the incident. For FIFO workloads, removing a message to a DLQ can break the business sequence. SQS DLQs.

Parallelize independent work with bounded concurrency that respects downstream capacity. In JavaScript, await Promise.all(records.map(processRecord)) awaits the mapped promises; Promise.all([records.map(processRecord)]) merely receives a nested array and does not await those inner tasks. FIFO records in the same group need ordered processing. Lambda partial batch responses require the event source mapping option ReportBatchItemFailures; returning the JSON shape alone is insufficient. Partial batch handling.

Practice: trace one operation from database commit through publication, delivery, side effect, and acknowledgement. At each boundary, ask what a timeout leaves unknown and how the next attempt recovers. That exercise reveals more about the design than a diagram whose every arrow is labelled “success.”