# 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.