# Patterns — Async Messaging

# Contents

Event envelope
Transactional outbox
Idempotent consumer with dedup table
Ack discipline
Schema evolution
DLQ routing
Gotchas

# Event envelope

json

{
  "event_id": "01J4QG8Z3V9K6W2N8P5R7T1X4C",
  "event_type": "order_placed",
  "schema_version": 1,
  "occurred_at": "2026-08-05T14:03:22Z",
  "producer": "orders-service",
  "payload": {
    "order_id": "ord_8842",
    "customer_id": "cus_311",
    "total_cents": 12900,
    "currency": "USD"
  }
}

ULIDs for event_id: sortable by creation time, globally unique.

# Transactional outbox

sql

CREATE TABLE outbox (
  id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  event_id     text NOT NULL UNIQUE,
  topic        text NOT NULL,
  envelope     jsonb NOT NULL,
  published_at timestamptz          -- NULL = pending
);

python

# Producer: business write + outbox row, one transaction
with db.transaction():
    create_order(order)
    db.execute("INSERT INTO outbox (event_id, topic, envelope) VALUES (%s,%s,%s)",
               (evt.event_id, "orders", evt.json()))

# Relay: poll-and-publish (or CDC/Debezium as the escape hatch at scale)
BATCH = 100  # small batches keep publish latency and crash re-sends bounded
rows = db.fetch("""SELECT * FROM outbox WHERE published_at IS NULL
                   ORDER BY id LIMIT %s FOR UPDATE SKIP LOCKED""", (BATCH,))
for r in rows:
    broker.publish(r.topic, r.envelope, key=r.envelope["payload"]["order_id"])
    db.execute("UPDATE outbox SET published_at = now() WHERE id = %s", (r.id,))

The relay is at-least-once (crash between publish and UPDATE → re-publish); consumer dedup absorbs it.

# Idempotent consumer with dedup table

python

def handle(envelope):
    with db.transaction():
        inserted = db.execute(
            """INSERT INTO consumed_events (consumer, event_id)
               VALUES (%s, %s) ON CONFLICT DO NOTHING""",
            ("shipping-service", envelope["event_id"])).rowcount
        if not inserted:
            return                      # duplicate or replay — already applied
        apply_side_effects(envelope)    # same transaction where possible
    ack()                               # only after commit

Prune consumed_events older than the broker's retention window.

# Ack discipline

python

# ❌ ack-then-process: crash after ack = event lost forever
msg = consume(); ack(msg); process(msg)

# ✅ process-then-ack: crash before ack = redelivery, dedup absorbs it
msg = consume(); process(msg); ack(msg)

# Schema evolution

Additive (same topic, bump minor): add optional field with a default.

Breaking (new topic): publish both during migration.

python

broker.publish("order_placed",    v1_envelope)  # until last v1 consumer migrates
broker.publish("order_placed.v2", v2_envelope)

Never: rename/retype a field in place, or repurpose an existing field.

# DLQ routing

python

MAX_ATTEMPTS = 5  # transient issues resolve well before 5 spaced retries

def consume_loop(msg):
    try:
        handle(msg)
    except Exception:
        if msg.delivery_count >= MAX_ATTEMPTS:
            broker.publish("orders.dlq", msg.envelope,
                           headers={"error": traceback.format_exc(limit=3)})
            ack(msg)          # remove poison message from the main stream
            alert("orders.dlq received a message")
        else:
            nack(msg)         # broker redelivers with backoff

# Gotchas

Outbox relay + UPDATE in one transaction with the publish is impossible — the broker isn't in your DB transaction. Accept relay at-least-once; dedup downstream.
Keying by random UUID destroys per-entity ordering; key by the entity whose sequence matters.
Consumer group rebalances redeliver in-flight messages — another duplicate source the dedup table must absorb.
Fat events as API snapshots rot: consumers act on stale fields. Carry IDs + the facts of the event; refetch the rest.
Retention < replay need: if history matters, size retention (or an archive) before the first consumer bug, not after.
One shared DLQ for all topics makes triage impossible; one DLQ per topic, each with its own alert.