--- name: event-driven-architecture-saga-patterns description: Designing resilient, eventual-consistency distributed transactions using the Saga Pattern (Choreography and Orchestration). Includes Outbox Pattern, Change Data Capture (CDC), idempotent consumers, compensating transactions, dual-write prevention, and saga recovery mechanisms. Use when implementing distributed multi-service workflows without two-phase commit (2PC). --- # Event-Driven Architecture: Saga Patterns & Eventual Consistency Guide ## 1. Architectural Foundations: BASE over ACID In microservices architectures, each service maintains its own private database to ensure loose coupling and independent scalability. This renders distributed ACID transactions impossible without two-phase commit (2PC). However, 2PC introduces high latency, synchronous blocking, single points of failure, and deadlocks across service boundaries. The **Saga Pattern** manages distributed transactions as a sequence of local transactions. Each local transaction updates a single service's database and emits an event or message to trigger the next step. If a step fails, the Saga executes **compensating transactions** in reverse order to undo changes and restore data consistency. ### 1.1 Transaction Classification ``` [ Local Tx 1: Compensable ] ──► [ Local Tx 2: Pivot ] ──► [ Local Tx 3: Retryable ] │ │ │ (Rollbackable) (Point of No Return) (Guaranteed Success) ``` - **Compensable Transactions**: Transactions that can be undone by running an explicit compensating transaction. - **Pivot Transaction**: The critical decision point of the Saga. Once the pivot transaction succeeds, the Saga MUST go to completion. It cannot be compensated. - **Retryable Transactions**: Transactions that follow the pivot point and are guaranteed to eventually succeed through retries. --- ## 2. Saga Execution Models: Choreography vs. Orchestration ``` CHOREOGRAPHY: [ Order Service ] ── OrderCreated ──► [ Payment Service ] ── PaymentProcessed ──► [ Inventory Service ] ORCHESTRATION: ┌──► [ Payment Service ] │ [ Saga Orchestrator ] ──┼──► [ Inventory Service ] │ └──► [ Shipping Service ] ``` ### 2.1 Comparative Analysis | Dimension | Choreography Saga | Orchestration Saga | | :--- | :--- | :--- | | **Control Model** | Decentralized; services react to domain events. | Centralized; dedicated orchestrator directs participants. | | **Coupling** | Loosely coupled; services know events, not handlers. | Moderate coupling; orchestrator knows participant APIs. | | **State Visibility** | Distributed across service logs; hard to query global status. | Centralized state store; easy to track saga status. | | **Complexity** | Simple for small flows (2-3 steps); cyclic dependencies in large flows. | High initial setup; manages complex branching & parallel paths easily. | | **Failure Handling** | Complex cascading compensation logic across events. | Explicit compensation routines directed by orchestrator. | ### 2.2 Decision Matrix - **Choose Choreography** when workflows are simple (2-3 microservices), linear, and teams want maximum domain event decoupling. - **Choose Orchestration** when workflows involve 4+ services, complex conditional branching, parallel execution steps, or require strict operational auditing. --- ## 3. Dual-Write Prevention: Transactional Outbox & CDC A major vulnerability in event-driven systems is the **Dual-Write Problem**: attempting to update a database and publish a message to a broker in a single code block without atomic guarantees. ``` WRONG (Dual Write Risk): db.orders.insert(order); // Success broker.publish("OrderCreated"); // Network Failure! -> DB & Broker state out of sync! ``` ### 3.1 Transactional Outbox Pattern Architecture ``` [ Business Logic ] │ ▼ (Single ACID Local DB Transaction) ┌──────────────────────────────────────────────┐ │ Service Database │ │ ├── Orders Table: [ Insert Order ] │ │ └── Outbox Table: [ Insert Outbox Event ] │ └──────────────────────────────────────────────┘ │ ▼ (Async Log Tailing / CDC) [ Debezium / Outbox Publisher ] ──► [ Message Broker (Kafka / RabbitMQ) ] ``` 1. Business record update and message metadata insertion are committed into the service's local database within a **single local ACID transaction**. 2. A separate process (e.g., Debezium CDC or a dedicated polling worker) reads the `outbox` table and safely forwards events to the message broker with **at-least-once** delivery guarantees. --- ## 4. Idempotency & De-duplication Engineering Because event delivery is **at-least-once**, consumers WILL receive duplicate events. Consumers must be designed to be completely **idempotent**. ### 4.1 Idempotent Consumer Design Patterns 1. **Natural Unique Key Constraints**: Insert incoming event processing results using database unique constraints (e.g., `PRIMARY KEY (payment_id)`). 2. **Idempotency Key / Deduplication Table**: ```sql CREATE TABLE processed_events ( event_id VARCHAR(128) PRIMARY KEY, consumer_group VARCHAR(64) NOT NULL, processed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ); ``` Before processing logic begins, attempt atomic insertion into `processed_events`. If duplicate key violation occurs, skip processing and immediately ACK message. 3. **State Machine Guardrails**: Only allow state transitions that are valid from the current state (e.g., `UPDATE orders SET status = 'PAID' WHERE id = :id AND status = 'PENDING'`). --- ## 5. Compensating Transactions & Failure Recovery Compensating transactions are inverse operations that semantically undo completed local transactions. ### 5.1 Rules for Compensating Actions - **Must Be Idempotent**: A compensating step may be retried multiple times during recovery. - **Must Never Fail Permanently**: If a compensating action encounters transient errors, it must retry until successful. If business invariants prevent auto-compensation, route to a **Human-in-the-Loop (HITL)** DLQ queue. - **Semantic Rollback**: Compensation does not necessarily mean "deleting" rows. It often means applying a counter-action (e.g., issuing a credit refund rather than un-writing a credit charge). --- ## 6. Critical Anti-Patterns & Production Traps 1. **Dual Writes without Outbox Pattern**: Updating local DB state and calling `broker.publish()` directly without outbox tables or transactional CDC. 2. **Missing Compensating Triggers**: Implementing forward execution steps without defining corresponding inverse compensation handlers. 3. **Cyclic Choreography Dependencies**: Service A triggers Service B, which triggers Service C, which triggers Service A again, creating un-traceable infinite loops. 4. **Non-Idempotent Compensations**: Compensating transactions that double-credit or double-refund when retried. 5. **Synchronous REST Calls inside Sagas**: Mixing blocking HTTP calls within asynchronous event-driven sagas, leading to thread pool exhaustion and cascading failures. --- ## 7. Production Code Reference ### 7.1 Transactional Outbox Pattern in Python (SQLAlchemy) ```python import uuid import json from datetime import datetime, timezone from sqlalchemy import create_engine, Column, String, Float, DateTime, Text from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() class Order(Base): __tablename__ = 'orders' id = Column(String(64), primary_key=True) customer_id = Column(String(64), nullable=False) total_amount = Column(Float, nullable=False) status = Column(String(32), nullable=False) class OutboxEvent(Base): __tablename__ = 'outbox' id = Column(String(64), primary_key=True) aggregate_type = Column(String(64), nullable=False) aggregate_id = Column(String(64), nullable=False) event_type = Column(String(64), nullable=False) payload = Column(Text, nullable=False) created_at = Column(DateTime(timezone=True), nullable=False) engine = create_engine("postgresql://app_user:secret@db.prod.internal:5432/orders_db") SessionLocal = sessionmaker(bind=engine) def create_order_with_outbox(customer_id: str, amount: float) -> str: session = SessionLocal() order_id = f"ord-{uuid.uuid4()}" event_id = f"evt-{uuid.uuid4()}" order = Order( id=order_id, customer_id=customer_id, total_amount=amount, status="PENDING_PAYMENT" ) event_payload = json.dumps({ "event_id": event_id, "order_id": order_id, "customer_id": customer_id, "amount": amount, "timestamp": datetime.now(timezone.utc).isoformat() }) outbox_entry = OutboxEvent( id=event_id, aggregate_type="Order", aggregate_id=order_id, event_type="OrderCreated", payload=event_payload, created_at=datetime.now(timezone.utc) ) try: # Atomic single local ACID transaction session.add(order) session.add(outbox_entry) session.commit() return order_id except Exception as e: session.rollback() raise RuntimeError(f"Failed to create order atomically: {str(e)}") finally: session.close() ``` ### 7.2 Saga Orchestrator State Machine in TypeScript / Node.js ```typescript type SagaState = 'STARTED' | 'PAYMENT_RESERVED' | 'INVENTORY_RESERVED' | 'COMPLETED' | 'FAILED' | 'COMPENSATING'; interface SagaContext { sagaId: string; orderId: string; customerId: string; amount: number; items: Array<{ sku: string; qty: number }>; currentState: SagaState; failureReason?: string; } export class OrderSagaOrchestrator { private context: SagaContext; constructor(context: SagaContext) { this.context = context; } async executeSaga(): Promise { try { // Step 1: Reserve Payment (Compensable) await this.executeStep('PAYMENT_RESERVED', () => this.callPaymentService('/payments/reserve', { amount: this.context.amount }), () => this.callPaymentService('/payments/refund', { amount: this.context.amount }) ); // Step 2: Reserve Inventory (Compensable) await this.executeStep('INVENTORY_RESERVED', () => this.callInventoryService('/inventory/reserve', { items: this.context.items }), () => this.callInventoryService('/inventory/release', { items: this.context.items }) ); // Step 3: Pivot Point - Approve Order (Retryable) this.context.currentState = 'COMPLETED'; await this.updateSagaState('COMPLETED'); await this.callOrderService(`/orders/${this.context.orderId}/approve`, {}); } catch (error: any) { console.error(`Saga ${this.context.sagaId} failed at state ${this.context.currentState}: ${error.message}`); this.context.failureReason = error.message; await this.compensateSaga(); } } private async executeStep( targetState: SagaState, action: () => Promise, compensation: () => Promise ): Promise { try { await action(); this.context.currentState = targetState; await this.updateSagaState(targetState); } catch (err) { // Register compensation step for reverse execution this.compensations.push(compensation); throw err; } } private compensations: Array<() => Promise> = []; private async compensateSaga(): Promise { this.context.currentState = 'COMPENSATING'; await this.updateSagaState('COMPENSATING'); // Run compensations in reverse order for (const compensation of this.compensations.reverse()) { let retryCount = 0; let success = false; while (!success && retryCount < 5) { try { await compensation(); success = true; } catch (compErr) { retryCount++; console.error(`Compensation retry ${retryCount} failed:`, compErr); await new Promise(res => setTimeout(res, 1000 * Math.pow(2, retryCount))); } } if (!success) { // Escalate to Human-In-The-Loop / DLQ Alerting console.error(`CRITICAL: Compensation failed permanently for Saga ${this.context.sagaId}`); await this.routeToHumanInterventionDLQ(this.context); } } this.context.currentState = 'FAILED'; await this.updateSagaState('FAILED'); } private async updateSagaState(state: SagaState): Promise {/* DB state update */} private async callPaymentService(path: string, payload: any): Promise {/* HTTP/gRPC */} private async callInventoryService(path: string, payload: any): Promise {/* HTTP/gRPC */} private async callOrderService(path: string, payload: any): Promise {/* HTTP/gRPC */} private async routeToHumanInterventionDLQ(context: SagaContext): Promise {/* DLQ Alert */} } ```