loom-event-driven

# Event-Driven Architecture ## Overview Patterns for decoupling services via events instead of synchronous calls: message queues, pub/sub, event sourcing, CQRS, sagas, and streaming. Scope of THIS skill = the architecture and its distributed-systems traps (delivery semantics, ordering, outbox, schema evolution). For job-queue *mechanics* — worker pools, scheduling/cron, retry/backoff internals, generic DLQ plumbing — see `loom-background-jobs`; cross-reference rather than duplicate. **The one law that governs everything below:** networked delivery is *at-least-once*; therefore **every consumer must be idempotent.** Read the Delivery Semantics section first — most EDA bugs are a violation of it. Code samples are TypeScript for concreteness; the patterns are language-agnostic. Assume `crypto.randomUUID()`, a broker client, and a datastore are in scope. --- ## Delivery Semantics — "Exactly-once" Is (Mostly) a Lie Two-Generals: across an unreliable network you cannot guarantee a message is delivered *exactly* once. You get to pick a *failure mode*: | Semantic | Mechanism | Failure mode | Use when | | --- | --- | --- | --- | | At-most-once | fire-and-forget, ack before processing | **loses** messages on crash | metrics/telemetry where loss is OK | | At-least-once | ack *after* processing, redeliver on no-ack | **duplicates** on retry/redelivery | **default for everything that matters** | | "Exactly-once" | at-least-once transport **+ idempotent consumer + dedup** | none observable | any state-changing handler | **Exactly-once is an *effect* you engineer at the consumer, not a transport guarantee you buy.** The real recipe: 1. At-least-once delivery (durable broker, ack after commit). 2. Idempotent consumer: processing the same message twice == processing it once. 3. A dedup/idempotency store keyed by a stable message id or business key. ⚠️ **Kafka "exactly-once semantics (EOS)" is real but scoped to Kafka.** Idempotent producer + transactions give exactly-once for **read-topic → process → write-topic-and-offsets** *inside Kafka*. It does **not** cover external side effects — a DB write, an email, a charge. The moment a handler touches the outside world, you are back to at-least-once and must be idempotent. Never tell a stakeholder Kafka gives you exactly-once for a payment. ### Idempotent consumer (the load-bearing pattern) ```typescript // Dedup by stable id (producer-assigned messageId or a business key). // Store the result so a duplicate returns the SAME answer, not a re-execution. async function handleIdempotent<T>(messageId: string, work: () => Promise<T>): Promise<T> { const key = `dedup:${messageId}`; const cached = await redis.get(key); if (cached) return JSON.parse(cached) as T; // already processed → no-op replay const result = await work(); // side effects happen here // SETNX + TTL: window must exceed max redelivery lag (broker retention + retry horizon) await redis.set(key, JSON.stringify(result), "EX", 24 * 3600, "NX"); return result; } ``` **Gotchas.** - ⚠ **Dedup window vs. redelivery horizon.** If the TTL expires before the broker could still redeliver (retention + max retry delay + DLQ replay), a late duplicate re-executes. Size the window to the *worst-case* redelivery age, not the happy path. - ⚠ **The "check then act" race.** Two workers both miss the key and both execute. Either make the side effect itself idempotent (DB `INSERT ... ON CONFLICT DO NOTHING` on a unique business key — the strongest option, atomic with the write), or gate with a lock. A Redis dedup key is best-effort unless it shares the transaction with the side effect. - ⚠ **Prefer a natural idempotency key** (orderId, paymentIntentId) over the broker's message id. Redelivery may carry a *new* transport id for the same business fact, and a producer retry can emit two transport messages for one intent. --- ## The Dual-Write Problem & the Transactional Outbox The single most common EDA correctness bug: **updating the database and publishing an event as two separate operations.** ```typescript // ❌ BROKEN: two systems, no atomicity await db.orders.insert(order); // commits await broker.publish("order.created"); // crash here → DB has order, world never hears about it ``` Reorder it and you get the opposite: publish succeeds, DB write fails, phantom event for an order that doesn't exist. There is **no ordering of these two lines that is safe** — you cannot atomically commit across a DB and a broker without distributed transactions (which brokers don't support and you don't want; see 2PC below). **Fix: Transactional Outbox.** Write the event into an `outbox` table **in the same local DB transaction** as the state change. A separate *relay* reads the outbox and publishes, marking rows sent. One atomic commit; publishing becomes at-least-once (relay may crash after publish, before marking) → consumers idempotent. ```typescript // Producer: one transaction, two writes to the SAME db await db.tx(async (t) => { await t.orders.insert(order); await t.outbox.insert({ id: crypto.randomUUID(), aggregate_id: order.id, type: "order.created", payload: JSON.stringify(order), created_at: new Date(), published_at: null, // relay flips this }); }); ``` **Two ways to run the relay:** - **Polling publisher** — `SELECT ... WHERE published_at IS NULL ORDER BY created_at` (add `FOR UPDATE SKIP LOCKED` so multiple relays don't double-publish), publish, set `published_at`. Simple, portable; adds latency and DB load. - **CDC / log-tailing (Debezium, etc.)** — a connector tails the DB WAL/binlog and streams outbox inserts to the broker. No polling, near-real-time, no query load — but operational weight (Kafka Connect) and DB-specific. Debezium has a dedicated **Outbox Event Router**. **Gotchas.** - ⚠ Outbox is **at-least-once**, never exactly-once — relay can publish then die before marking sent. Non-negotiable: idempotent consumers. - ⚠ **Ordering:** to preserve per-aggregate order, publish with the aggregate id as the partition key and have the relay process a given aggregate's rows in `created_at`/sequence order. - ⚠ **Outbox bloat:** prune `published_at IS NOT NULL` rows on a schedule; an unbounded outbox degrades the polling query. - **Inbox pattern** is the mirror on the consumer: record processed message ids in an inbox table inside the same transaction as the write — atomic dedup without an external store. --- ## Ordering — Only Per-Partition/Per-Key, Never Global **Global total order across a topic does not scale and is not offered.** Kafka guarantees order **only within a single partition**; RabbitMQ only within a single queue with a single consumer; SQS FIFO only within a `MessageGroupId`. **Consequences you must design around:** - **Choose the partition key = the entity whose events must stay ordered** (orderId, userId, accountId). All events for one entity land on one partition → ordered. Cross-entity order is *not* guaranteed and you must not depend on it. - **Hot-partition skew.** A skewed key ("region=US", or one whale tenant) piles most traffic on one partition → that consumer lags while others idle. Pick a high-cardinality, evenly-distributed key; composite keys or salting for known-hot entities. - **Consumer parallelism is capped by partition count** per consumer group. 6 partitions → at most 6 useful consumers. Repartitioning later is disruptive (rehashes keys, breaks in-flight order). Over-provision partitions modestly up front. - **Retries reorder.** Reprocessing message N after N+1 already went breaks order. Kafka's idempotent producer preserves send order with up to `max.in.flight.requests.per.connection=5`; *without* idempotence, `>1` in-flight can reorder on retry — set it to 1 or enable idempotence. - ⚠ If handlers must be ordered per key, **do not fan a single partition out to a worker pool** — that reintroduces reordering. Parallelize across partitions/keys, serialize within a key. --- ## Fat vs. Thin Events (the coupling / staleness trade-off) How much state does an event carry? Two poles: | | Thin / notification | Fat / event-carried state transfer | | --- | --- | --- | | Payload | ids only (`{orderId}`) | full snapshot (`{orderId, items, total, status,...}`) | | Consumer action | **call back / refetch** source | read straight from the event | | Coupling | temporal (source must be up *now*) + API coupling | schema coupling to the event shape | | Autonomy | low | high (consumer needs nothing else) | | PII/size | small | larger; **spreads PII** to every subscriber | ⚠ **The thin-event stale-read race:** consumer gets `order.updated {id}`, refetches the order — but events are delivered async and can arrive *before* the source's own read replica is consistent, or after a *newer* update. You can read a **stale or wrong version**. Mitigations: include a **version/sequence number** in the thin event and refetch-then-check `version >=`, or make the source's read strongly consistent for this path, or just send a fat event. **Default to event-carried state transfer** (fat) for integration events between services — it removes temporal coupling and the refetch storm. Use thin events when payloads are huge, PII-sensitive, or consumers legitimately need the freshest value at handling time. Always version either way. --- ## Broker Selection | Broker | Model | Ordering | Delivery | Retention/replay | Reach for it when | | --- | --- | --- | --- | --- | --- | | **Kafka** | partitioned log, pull | per-partition | at-least-once; EOS *within Kafka* | long, offset-based replay | high-throughput streaming, event sourcing, replayable log, many independent consumer groups | | **RabbitMQ** | queues + exchanges, push | per-queue (single consumer) | at-least-once | consumed msgs gone (no replay) | complex routing (topic/headers/fanout), per-message TTL, RPC, priority queues | | **Pulsar** | log, segmented storage | per-partition/key-shared | at-least-once; effectively-once dedup | tiered (offload to S3), long | Kafka-like + multi-tenancy, geo-replication, unified queue+stream, decoupled compute/storage | | **SQS + SNS** | managed queue (+ fanout) | FIFO: per-MessageGroupId; Standard: none | at-least-once (Std); FIFO exactly-once *in-queue* | up to 14 days; no arbitrary replay | AWS-native, zero-ops, native DLQ redrive; SNS/EventBridge fan-out | | **NATS JetStream** | log/stream, pull | per-subject-sequence | at-least-once (+ msg-id dedup window) | limits/interest/workqueue, replay | low-latency, lightweight ops, edge/IoT, request-reply + streams | | **Redis Streams** | log | per-stream | at-least-once (consumer groups + PEL) | capped (MAXLEN) | already-have-Redis, modest scale, simple durable queue | **Selection heuristics.** - Need **replay / rebuild projections / event sourcing** → log-based (Kafka/Pulsar). Queue brokers discard consumed messages. - Need **rich routing / per-message priority / TTL** → RabbitMQ. - **On AWS and don't want to run brokers** → SQS/SNS/EventBridge; native DLQ redrive is a real ergonomics win. - Extreme throughput or multi-consumer-group replay → Kafka. Kafka is overkill (and heavy ops) for a simple work queue — don't Kafka a to-do list. - ⚠ Don't conflate SNS (fanout pub/sub, no per-subscriber durability by itself) with SQS (durable queue). The durable fanout pattern is **SNS → SQS per consumer**. --- ## Pub/Sub — Kafka (canonical producer/consumer) ```typescript import { Kafka, logLevel } from "kafkajs"; const kafka = new Kafka({ clientId: "order-service", brokers: ["localhost:9092"] }); // Producer: idempotent → no duplicates from producer-side retries, preserves order. const producer = kafka.producer({ idempotent: true, maxInFlightRequests: 5 }); await producer.connect(); await producer.send({ topic: "orders", messages: [{ key: order.id, // partition key = ordering key value: JSON.stringify({ type: "order.created", data: order, v: 1 }), headers: { "event-type": "order.created", "correlation-id": corrId }, }], }); // Consumer: one group per logically-distinct subscriber; each group gets all messages. const consumer = kafka.consumer({ groupId: "inventory-service" }); await consumer.subscribe({ topic: "orders", fromBeginning: false }); await consumer.run({ eachMessage: async ({ message }) => { const evt = JSON.parse(message.value!.toString()); await handleIdempotent(message.key!.toString() + ":" + evt.v, () => react(evt)); // throwing here does NOT auto-DLQ in Kafka — offset isn't committed, msg redelivers. // Bound retries yourself, then produce to an error topic (see Poison Pills). }, }); ``` **Kafka gotchas.** - ⚠ **A crashing handler blocks the partition.** No auto-DLQ: the offset isn't committed, so the same message redelivers forever (poison pill) and everything behind it stalls. You must implement bounded-retry-then-error-topic. - **Consumer group == subscription.** Two services that both need every event use two *different* group ids. Two instances that should *share* the load use the *same* group id (partitions split among them). - **`fromBeginning`** only matters the first time a group has no committed offset; afterwards it resumes from the committed offset. - Manual offset commit after successful processing = at-least-once. Auto-commit before processing = at-most-once (silent loss on crash). **RabbitMQ** (routing-first): declare queue with `x-dead-letter-exchange` + `x-message-ttl`, publish `persistent: true`, `prefetch(n)` for backpressure, `ack` after success / `nack(requeue=false)` to route to the DLX. `nack(requeue=true)` in a tight loop is a poison-pill amplifier — track a retry-count header and DLX after N. **NATS JetStream** (lightweight, durable): create a stream (`subjects: ["events.*"]`, retention `limits`), a **durable** pull consumer with `ack_policy: explicit` and `max_deliver: N` (built-in redelivery cap), publish with `Nats-Msg-Id` for the server-side dedup window. `msg.ack()` / `msg.nak()` / `msg.term()` (term = don't redeliver, straight to poison handling). --- ## Poison Pills & Dead-Letter Policy A **poison pill** is a message that *always* fails — malformed payload, unparseable schema, a referenced entity that will never exist. In-band infinite retry blocks the partition/queue behind it and can pin CPU. Policy: 1. **Bounded retries with backoff** (exponential + jitter), retry count carried in a header/attribute. 2. On exhaustion, **park it in a DLQ** (never drop silently) with metadata: original topic/queue, error, stack, retry count, first-seen time, correlation id. 3. **Alert** on DLQ arrivals and on DLQ size/age crossing a threshold — a growing DLQ is an incident, not a metric. 4. **Replay tooling**: inspect, edit/fix, selectively redrive, and purge. A DLQ you can't replay from is a graveyard. **Broker specifics.** - **SQS:** native `RedrivePolicy` (`maxReceiveCount` → DLQ) and console/SDK **redrive back** to source. Prefer it over hand-rolling. - **Kafka:** no native DLQ. Kafka Connect sinks have `errors.deadletterqueue.topic.name`; app consumers publish failures to an `<topic>.DLT` error topic yourself. - **RabbitMQ:** DLX + a retry queue with per-message TTL that dead-letters *back* to the work queue implements delayed retry. ⚠ **Retry vs. DLQ classification:** only retry *transient* failures (timeouts, 5xx, lock contention). A deserialization error or a 4xx/validation failure is deterministic — retrying wastes time and delays the DLQ; fast-fail those straight to the DLQ. Distinguish retryable from non-retryable in the handler. See `loom-background-jobs` for generic retry/backoff and worker-pool mechanics. --- ## Message Queues (work distribution) Competing-consumers: N workers pull from one queue, each message handled once (per successful ack). Contrast with pub/sub (every subscriber group gets every message). ```typescript // SQS competing-consumers loop. Delete = ack; not-deleting = redelivery after visibility timeout. while (running) { const { Messages = [] } = await sqs.receiveMessage({ QueueUrl: url, MaxNumberOfMessages: 10, WaitTimeSeconds: 20, // long-poll AttributeNames: ["ApproximateReceiveCount"], }); await Promise.all(Messages.map(async (m) => { try { await handleIdempotent(m.MessageId!, () => process(JSON.parse(m.Body!))); await sqs.deleteMessage({ QueueUrl: url, ReceiptHandle: m.ReceiptHandle! });