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:

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)

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

// ❌ 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.

// 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:

⚠ 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

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)

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

// 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! });
    } catch (e) {
      // Do NOT delete → SQS redelivers after visibility timeout; RedrivePolicy routes to DLQ
      // once ApproximateReceiveCount > maxReceiveCount. No manual DLQ code needed.
    }
  }));
}

Gotchas.

• ⚠ Visibility timeout must exceed worst-case processing time, or SQS redelivers a message you're still working on → duplicate processing. For long jobs, extend visibility with ChangeMessageVisibility heartbeats.
• ⚠ SQS Standard is at-least-once AND unordered. ApproximateReceiveCount is approximate. FIFO gives ordering per MessageGroupId + in-queue dedup via MessageDeduplicationId, at lower throughput.
• Long polling (WaitTimeSeconds=20) — always set it; short polling burns API calls and money and returns empty on sparsely-populated queues.
• prefetch/batch size is your backpressure knob: too high and a slow consumer hoards messages it can't process before the visibility timeout.

Event Sourcing

Store the sequence of state-changing events as the source of truth; derive current state by replaying them. Not for every domain — reach for it when you need a full audit trail, temporal ("what did it look like at T?") queries, or event-driven integration by construction. It's overkill for CRUD.

// Append with OPTIMISTIC CONCURRENCY: expected version guards against lost updates.
// A UNIQUE(aggregate_id, version) constraint is the real enforcement — the SELECT is advisory.
async function append(events: DomainEvent[]): Promise<void> {
  await db.tx(async (t) => {
    for (const e of events) {
      const { max } = await t.one(
        "SELECT COALESCE(MAX(version),0) AS max FROM events WHERE aggregate_id=$1", [e.aggregateId]);
      if (e.version !== max + 1) throw new ConcurrencyError(`expected ${max + 1}, got ${e.version}`);
      await t.none(
        `INSERT INTO events(id,aggregate_id,type,version,data,metadata,ts)
         VALUES($1,$2,$3,$4,$5,$6,$7)`,                 // UNIQUE(aggregate_id,version) catches races
        [e.id, e.aggregateId, e.type, e.version, e.data, e.metadata, e.ts]);
    }
    // ⚠ Do NOT publish to the broker here (dual-write). Write to an OUTBOX in this same tx,
    // OR let a projector/relay tail the events table (which IS your outbox in ES).
  });
}

Aggregate replay is a fold: apply(command) validates invariants and emits an event; when(event) mutates in-memory state; loadFromHistory(events) replays when over the stream to rebuild. Uncommitted events are appended, then cleared after a successful append.

Event sourcing traps

• ⚠ Events are immutable facts. Never edit or delete an event. A mistake is corrected by appending a compensating event (OrderCorrected), not by mutating history. Editing history silently corrupts every projection that already consumed it.
• Rebuild cost is O(stream length). Long-lived aggregates get slow to load → snapshots: persist state every N events, load latest snapshot + events after it. Snapshots are a cache, never the source of truth (you must be able to delete all snapshots and rebuild).
• Schema evolution via upcasting, never rewriting stored events. On read, transform old event versions forward to the current shape (v1 → v2 → v3). Old versions live in the log forever, so upcasters live forever too.
• Projections are eventually consistent. The write side commits before read models catch up — don't read-your-own-write from a projection right after a command. Return the new version/state from the command, or subscribe/poll for the version to appear.
• ⚠ Immutable log vs. GDPR / right-to-erasure. You can't delete a person's events without breaking the chain. Crypto-shredding: encrypt PII with a per-subject key held outside the log; to "erase", delete the key — the ciphertext in the events becomes permanently unreadable while the log stays intact. Design this in from day one; retrofitting is brutal.
• Not everything is an aggregate. Cross-aggregate invariants can't be enforced in one atomic append — use a saga/process manager and accept eventual consistency, or reconsider the boundaries.

CQRS (Command Query Responsibility Segregation)

Split the write model (commands → validate invariants → emit events) from the read model (denormalized projections optimized for queries). A command bus routes commands to handlers; projections subscribe to events and maintain query-shaped views.

// Projection: subscribe to domain events, maintain a denormalized read row.
// Enrichment (join in customer/product names) happens HERE, so reads are single-fetch.
class OrderProjection {
  async onOrderCreated(e: DomainEvent) {
    const d = e.data as OrderCreated;
    await this.read.orders.upsert({
      id: e.aggregateId, status: "pending", total: d.total,
      customerName: (await this.customers.get(d.customerId)).name,   // denormalize at write-time
      version: e.version, updatedAt: e.ts,
    });
  }
  async onOrderConfirmed(e: DomainEvent) {
    // ⚠ Idempotent + ordered: guard with version so a replayed/out-of-order event can't regress state.
    await this.read.orders.updateIf(e.aggregateId, { version_lt: e.version },
      { status: "confirmed", version: e.version, updatedAt: e.ts });
  }
}

Gotchas.

• ⚠ Read-model lag is inherent — the write commits, the projection updates milliseconds-to-seconds later. Design UX for it: optimistic UI, return the new state from the command, or expose "processing". Never assume a query immediately reflects a just-issued command.
• CQRS ≠ event sourcing. You can do CQRS with plain read replicas or maintained materialized views; ES is one way to feed projections, not a prerequisite.
• Projections are disposable and rebuildable — that's the point. Version the projection code; to change a read model's shape, rebuild it from the event log rather than migrating in place.
• ⚠ Idempotent, order-tolerant projection updates. At-least-once + possible reordering means a projection handler must be safe to run twice and must not regress on a stale event — guard writes by version/sequence.
• Don't apply CQRS to simple CRUD; the operational cost (two models, sync, eventual consistency) only pays off for read/write asymmetry or complex domains.

Sagas — Distributed Transactions Without 2PC

A business transaction spanning services can't hold one ACID transaction. A saga is a sequence of local transactions; if step k fails, run compensating transactions for steps k-1…1 in reverse.

Why not two-phase commit (2PC/XA)

• Blocking + locks held across services for the whole transaction → terrible availability and throughput.
• Coordinator is a SPOF; if it dies mid-commit, participants are stuck holding locks (in-doubt).
• Poor fit for the modern stack — most brokers and many datastores don't support XA; it doesn't scale.
• Sagas trade atomicity for availability: you accept a window of visible intermediate state, and converge via compensation. That trade is almost always correct for microservices.

Choreography vs. orchestration

// Orchestrated saga: linear steps, reverse-order compensation. Persist state after EVERY step
// so a crash resumes/compensates instead of losing the transaction.
async function runSaga(steps: SagaStep[], data: SagaData, store: SagaStore) {
  const done: SagaStep[] = [];
  const inst = await store.start(data);
  for (const step of steps) {
    try {
      await step.execute(data);                 // must be idempotent (may re-run after crash)
      done.push(step);
      await store.advance(inst, step.name, data);
    } catch (err) {
      for (const s of done.reverse()) {         // compensate in reverse
        try { await s.compensate(data); }       // compensations MUST be idempotent + retryable
        catch (ce) { await store.flagCompensationFailure(inst, s.name, ce); /* alert; keep going */ }
      }
      await store.fail(inst, err);
      return;
    }
  }
  await store.complete(inst);
}

Saga gotchas.

• ⚠ Compensation is semantic undo, not rollback. You can't un-charge a card — you refund. Un-send an email — you send a correction. Design compensations as forward business actions.
• Some steps aren't compensatable (email sent, physical shipment). Order steps so retriable-only actions come after the last compensatable one — the "pivot transaction". Everything before the pivot can be undone; everything after must be driven forward with retries.
• Compensations must be idempotent and retryable — they run during failure handling, exactly when the network is flaky. A failed compensation is a manual-intervention incident; alert, don't swallow.
• Persist saga state after each step. A saga is a state machine; on crash it must resume forward or resume compensating. In-memory-only sagas lose transactions on restart.
• Counter-compensation / semantic locks: during a saga, a resource is in a provisional state (PENDING). Concurrent readers must know not to treat it as final; use a status flag rather than assuming isolation you don't have.

Streaming (aggregations, joins, windows)

For stateful stream processing — windowed aggregations, stream-stream/stream-table joins — prefer a real stream-processing engine (Kafka Streams, ksqlDB, Flink, Spark Structured Streaming) over hand-rolling in a consumer. Hand-rolled windowing with setTimeout + in-memory Map (as older versions of this skill showed) has fatal gaps: state is lost on restart (not checkpointed), doesn't survive rebalance, no watermarks/late-data handling, and doesn't scale past one process. Roll your own only for trivial, loss-tolerant cases.

What the engines give you that a raw consumer doesn't:

• State stores backed by changelog topics → survive crashes/rebalances.
• Event-time windows + watermarks → correct results with out-of-order/late data (tumbling/hopping/session windows).
• Repartitioning (groupByKey) so aggregation keys land co-partitioned.
• Exactly-once processing within the Kafka boundary (processing.guarantee=exactly_once_v2).

⚠ Event-time vs. processing-time. Windowing on wall-clock (processing time) miscounts when events are delayed or replayed. Window on the embedded event timestamp and configure allowed lateness; otherwise a backfill or a lagging consumer silently corrupts aggregates.

Schema Evolution & Versioning

Events, once published, are consumed by code you don't control and (in event sourcing) stored forever. Schema is a permanent contract.

• Use a schema registry (Confluent Schema Registry with Avro/Protobuf/JSON Schema) to enforce compatibility at publish time. Compatibility modes: BACKWARD (new consumer reads old events — the usual default), FORWARD (old consumer reads new events), FULL (both).
• Only make additive, optional changes: add fields with defaults. Never remove/rename a field, change a type, or repurpose semantics in place — that breaks existing consumers and stored history.
• Tolerant reader: consumers ignore unknown fields and tolerate missing optional ones, so producers can evolve without lock-step deploys.
• Version explicitly — carry a schema version in the event (v/schemaVersion). In event sourcing, upcast old versions on read (see traps). For breaking changes, publish a new event type (OrderCreatedV2) and run both until consumers migrate.
• ⚠ Deploy order matters: with BACKWARD compat, roll out consumers before producers; with FORWARD, producers first. Ship the wrong order and you break in prod during the deploy window.

End-to-End Reference Flow (outbox + CQRS + saga)

// 1) HTTP → command. Return 202: this is async; the resource isn't queryable yet (read-model lag).
app.post("/orders", async (req, res) => {
  await commandBus.dispatch({ type: "CreateOrder", payload: req.body,
    metadata: { userId: req.user.id, correlationId: req.headers["x-correlation-id"] } });
  res.status(202).json({ status: "accepted" });        // NOT 200 with the order body
});

// 2) Handler: aggregate emits events; events + state committed in ONE tx via outbox (no dual write).
class CreateOrderHandler {
  async handle(cmd: CreateOrderCommand) {
    const order = Order.create(crypto.randomUUID(), cmd.payload.customerId, cmd.payload.items);
    await this.store.appendWithOutbox(order.getUncommittedEvents()); // events table + outbox, atomic
  }
}
// 3) Relay/CDC publishes outbox → Kafka (at-least-once).
// 4) Projections (idempotent, version-guarded) build read models; queries hit those.
// 5) Orchestrated saga runs reserve-inventory → charge → ship → confirm, compensating on failure.
//    Every consumer above is idempotent because delivery is at-least-once.

Best Practices (dense)

Events

• Past-tense facts (OrderCreated, not CreateOrder); immutable; one event = one business fact.
• Include id, type, version/schema-version, timestamp (event-time), correlationId, causationId, aggregate id/type.
• Prefer event-carried state transfer (fat) for integration events; version everything (see Fat vs. Thin).

Correctness

• At-least-once is the floor → every handler idempotent (dedup key or naturally idempotent side effect).
• Never dual-write DB + broker → transactional outbox (relay or CDC).
• Ordering only per partition/key → choose the key deliberately; watch hot partitions.

Reliability

• Bounded retries + backoff/jitter → DLQ; alert on DLQ arrivals and size/age; build replay tooling.
• Classify transient (retry) vs. deterministic (fast-fail to DLQ) failures.
• Visibility timeout / ack deadline > worst-case processing time; heartbeat-extend long jobs.

Observability

• Track consumer lag (Kafka offset lag, SQS ApproximateAgeOfOldestMessage) — the single most important EDA health metric; alert on sustained growth.
• Propagate correlationId/trace context through every hop (distributed tracing); it's the only way to reconstruct an async flow.
• Monitor processing latency, error rate, DLQ depth, redelivery counts, event-store/topic growth.

Architecture

• Sagas over 2PC; pick choreography (few steps) vs. orchestration (complex/auditable) deliberately.
• Event sourcing/CQRS only where audit/temporal/read-write-asymmetry justify the cost; not for CRUD.
• Don't over-broker: Kafka for streaming/replay, queue brokers for work distribution, managed (SQS/SNS) to avoid ops.

Verification Checklists

Before shipping any consumer:

• Handler is idempotent — reprocessing the same message twice yields the same state (dedup key or ON CONFLICT/naturally-idempotent write).
• Dedup/idempotency window ≥ worst-case redelivery age (retention + retry horizon + DLQ replay).
• Failures are classified: transient → bounded retry+backoff; deterministic → straight to DLQ.
• There is a DLQ, it alerts on arrival + size/age, and you can replay from it.
• Poison pills can't block the partition/queue forever (retry cap enforced).
• Ack/delete happens after successful processing (at-least-once), not before.
• Visibility/ack timeout exceeds max processing time (or heartbeat-extended).
• Correlation/trace id is read from and propagated onward.

Before shipping any producer / write path:

• No dual write — DB change and event publish are atomic (outbox in same tx, or CDC).
• Partition/message key chosen so per-entity order holds; hot-partition risk assessed.
• Kafka producer idempotent: true (or equivalent) to avoid producer-retry duplicates.
• Event includes version/schema-version, correlation id, event-time timestamp.
• Schema change is additive + registry-compatible; deploy order (consumer-vs-producer-first) is correct.

Event sourcing / CQRS specific:

• Optimistic concurrency enforced by a UNIQUE(aggregate_id, version) constraint, not just a SELECT.
• No code path edits or deletes stored events; corrections are compensating events.
• Snapshot strategy exists for long streams; system still correct with all snapshots deleted.
• Old event versions are upcast on read; upcasters covered by tests.
• PII erasure strategy (crypto-shredding) designed in, not retrofitted.
• Projections are idempotent + version-guarded and rebuildable from the log.
• Callers don't read-your-own-write from a lagging projection.

Saga specific:

• Saga state persisted after every step; crash resumes forward or compensating.
• Every compensation is a semantic forward action, idempotent, and retryable.
• Non-compensatable steps ordered after the pivot (retriable-only) transaction.
• Compensation failures alert for manual intervention (not swallowed).
• Provisional resource states are visible to concurrent readers (status flags / semantic locks).