spb/ultra-sharp-agent-skills Public

1<!--2Author: Simon-Pierre Boucher3Contact: contact@spboucher.ai4-->56# Patterns — Async Messaging78## Contents9- Event envelope10- Transactional outbox11- Idempotent consumer with dedup table12- Ack discipline13- Schema evolution14- DLQ routing15- Gotchas1617## Event envelope1819```json20{21  "event_id": "01J4QG8Z3V9K6W2N8P5R7T1X4C",22  "event_type": "order_placed",23  "schema_version": 1,24  "occurred_at": "2026-08-05T14:03:22Z",25  "producer": "orders-service",26  "payload": {27    "order_id": "ord_8842",28    "customer_id": "cus_311",29    "total_cents": 12900,30    "currency": "USD"31  }32}33```3435ULIDs for `event_id`: sortable by creation time, globally unique.3637## Transactional outbox3839```sql40CREATE TABLE outbox (41  id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,42  event_id     text NOT NULL UNIQUE,43  topic        text NOT NULL,44  envelope     jsonb NOT NULL,45  published_at timestamptz          -- NULL = pending46);47```4849```python50# Producer: business write + outbox row, one transaction51with db.transaction():52    create_order(order)53    db.execute("INSERT INTO outbox (event_id, topic, envelope) VALUES (%s,%s,%s)",54               (evt.event_id, "orders", evt.json()))5556# Relay: poll-and-publish (or CDC/Debezium as the escape hatch at scale)57BATCH = 100  # small batches keep publish latency and crash re-sends bounded58rows = db.fetch("""SELECT * FROM outbox WHERE published_at IS NULL59                   ORDER BY id LIMIT %s FOR UPDATE SKIP LOCKED""", (BATCH,))60for r in rows:61    broker.publish(r.topic, r.envelope, key=r.envelope["payload"]["order_id"])62    db.execute("UPDATE outbox SET published_at = now() WHERE id = %s", (r.id,))63```6465The relay is at-least-once (crash between publish and UPDATE → re-publish);66consumer dedup absorbs it.6768## Idempotent consumer with dedup table6970```python71def handle(envelope):72    with db.transaction():73        inserted = db.execute(74            """INSERT INTO consumed_events (consumer, event_id)75               VALUES (%s, %s) ON CONFLICT DO NOTHING""",76            ("shipping-service", envelope["event_id"])).rowcount77        if not inserted:78            return                      # duplicate or replay — already applied79        apply_side_effects(envelope)    # same transaction where possible80    ack()                               # only after commit81```8283Prune `consumed_events` older than the broker's retention window.8485## Ack discipline8687```python88# ❌ ack-then-process: crash after ack = event lost forever89msg = consume(); ack(msg); process(msg)9091# ✅ process-then-ack: crash before ack = redelivery, dedup absorbs it92msg = consume(); process(msg); ack(msg)93```9495## Schema evolution9697Additive (same topic, bump minor): add optional field with a default.9899Breaking (new topic): publish both during migration.100101```python102broker.publish("order_placed",    v1_envelope)  # until last v1 consumer migrates103broker.publish("order_placed.v2", v2_envelope)104```105106Never: rename/retype a field in place, or repurpose an existing field.107108## DLQ routing109110```python111MAX_ATTEMPTS = 5  # transient issues resolve well before 5 spaced retries112113def consume_loop(msg):114    try:115        handle(msg)116    except Exception:117        if msg.delivery_count >= MAX_ATTEMPTS:118            broker.publish("orders.dlq", msg.envelope,119                           headers={"error": traceback.format_exc(limit=3)})120            ack(msg)          # remove poison message from the main stream121            alert("orders.dlq received a message")122        else:123            nack(msg)         # broker redelivers with backoff124```125126## Gotchas127128- **Outbox relay + `UPDATE` in one transaction with the publish** is129  impossible — the broker isn't in your DB transaction. Accept relay130  at-least-once; dedup downstream.131- **Keying by random UUID** destroys per-entity ordering; key by the entity132  whose sequence matters.133- **Consumer group rebalances** redeliver in-flight messages — another134  duplicate source the dedup table must absorb.135- **Fat events as API snapshots** rot: consumers act on stale fields.136  Carry IDs + the facts of the event; refetch the rest.137- **Retention < replay need**: if history matters, size retention (or an138  archive) before the first consumer bug, not after.139- **One shared DLQ for all topics** makes triage impossible; one DLQ per140  topic, each with its own alert.141