saga-pattern

# Saga Pattern ## What I do I provide a pattern for managing distributed transactions across multiple services without traditional two-phase commit. Sagas coordinate a sequence of local transactions, where each transaction updates data and publishes an event to trigger the next step. If a step fails, compensating transactions undo previous steps, maintaining data consistency across services. This enables long-running business processes while preserving eventual consistency. ## When to use me Use sagas when you need to coordinate actions across multiple microservices or bounded contexts, especially when traditional distributed transactions are impractical. Sagas are ideal for long-running business workflows, order processing, booking systems, or any multi-step process spanning services. Avoid sagas when ACID transactions within a single service are sufficient, or when strict immediate consistency is required across all steps. ## Core Concepts - **Saga**: Sequence of local transactions with compensating actions - **Local Transaction**: Single service operation with its own database - **Compensating Transaction**: Action that undoes a local transaction - **Choreography**: Distributed coordination via events - **Orchestration**: Central coordinator managing saga flow - **Saga State**: Tracking progress and handling failures - **Idempotency**: Safe to execute steps multiple times - **Retry Strategies**: Handling transient failures - **Timeout Management**: Preventing hung sagas - **Checkpointing**: Saving saga state for recovery ## Code Examples ### Choreography-Based Saga ```python from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime from typing import Protocol, Callable from uuid import UUID, uuid4 @dataclass class OrderCreated: order_id: UUID customer_id: UUID items: list[dict] total: float timestamp: datetime = datetime.utcnow() @dataclass class PaymentProcessed: order_id: UUID payment_id: UUID amount: float timestamp: datetime = datetime.utcnow() @dataclass class InventoryReserved: order_id: UUID reservation_id: UUID items: list[dict] timestamp: datetime = datetime.utcnow() @dataclass class OrderShipped: order_id: UUID tracking_number: str carrier: str timestamp: datetime = datetime.utcnow() @dataclass class OrderCancelled: order_id: UUID reason: str timestamp: datetime = datetime.utcnow() class EventBus(Protocol): def publish(self, event: object) -> None: pass class OrderService: def __init__(self, event_bus: EventBus): self._event_bus = event_bus self._orders: dict[UUID, dict] = {} def create_order(self, customer_id: UUID, items: list[dict]) -> OrderCreated: order_id = uuid4() total = sum(item["price"] * item["quantity"] for item in items) order = { "order_id": order_id, "customer_id": customer_id, "items": items, "total": total, "status": "created" } self._orders[order_id] = order event = OrderCreated( order_id=order_id, customer_id=customer_id, items=items, total=total ) self._event_bus.publish(event) return event class PaymentService: def __init__(self, event_bus: EventBus): self._event_bus = event_bus self._payments: dict[UUID, dict] = {} def handle_order_created(self, event: OrderCreated) -> None: payment_id = uuid4() payment = { "payment_id": payment_id, "order_id": event.order_id, "amount": event.total, "status": "processing" } self._payments[payment_id] = payment payment_event = PaymentProcessed( order_id=event.order_id, payment_id=payment_id, amount=event.total ) self._event_bus.publish(payment_event) def refund(self, order_id: UUID) -> None: for payment in self._payments.values(): if payment["order_id"] == order_id: payment["status"] = "refunded" class InventoryService: def __init__(self, event_bus: EventBus): self._event_bus = event_bus self._reservations: dict[UUID, dict] = {} self._inventory: dict[str, int] = {"P001": 100, "P002": 50} def handle_order_created(self, event: OrderCreated) -> None: reservation_id = uuid4() reserved_items = [] for item in event.items: product_id = item["product_id"] if self._inventory.get(product_id, 0) >= item["quantity"]: self._inventory[product_id] -= item["quantity"] reserved_items.append(item) reservation = { "reservation_id": reservation_id, "order_id": event.order_id, "items": reserved_items, "status": "reserved" } self._reservations[reservation_id] = reservation self._event_bus.publish(InventoryReserved( order_id=event.order_id, reservation_id=reservation_id, items=reserved_items )) def release(self, order_id: UUID) -> None: for reservation in self._reservations.values(): if reservation["order_id"] == order_id: for item in reservation["items"]: self._inventory[item["product_id"]] += item["quantity"] reservation["status"] = "released" class ShippingService: def __init__(self, event_bus: EventBus): self._event_bus = event_bus self._shipments: dict[UUID, dict] = {} def handle_inventory_reserved(self, event: InventoryReserved) -> None: shipment_id = uuid4() tracking = f"TRK-{uuid4().hex[:8].upper()}" shipment = { "shipment_id": shipment_id, "order_id": event.order_id, "tracking_number": tracking, "status": "shipped" } self._shipments[shipment_id] = shipment self._event_bus.publish(OrderShipped( order_id=event.order_id, tracking_number=tracking, carrier="FEDEX" )) ``` ### Orchestration-Based Saga ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import Protocol, Optional from uuid import UUID, uuid4 @dataclass class SagaContext: saga_id: UUID order_id: UUID customer_id: UUID items: list[dict] total: float step_results: dict[str, dict] = field(default_factory=dict) current_step: str = "" is_completed: bool = False is_compensating: bool = False class SagaStep(ABC): @property @abstractmethod def name(self) -> str: pass @abstractmethod def execute(self, context: SagaContext) -> dict: pass @abstractmethod def compensate(self, context: SagaContext, step_result: dict) -> None: pass class CreateOrderStep(SagaStep): @property def name(self) -> str: return "create_order" def execute(self, context: SagaContext) -> dict: return {"order_id": str(context.order_id), "status": "created"} def compensate(self, context: SagaContext, step_result: dict) -> None: print(f"Compensating: Order {step_result['order_id']} cancelled") class ProcessPaymentStep(SagaStep): @property def name(self) -> str: return "process_payment" def execute(self, context: SagaContext) -> dict: return {"payment_id": str(uuid4()), "amount": context.total} def compensate(self, context: SagaContext, step_result: dict) -> None: print(f"Compensating: Refunding payment {step_result['payment_id']}") class ReserveInventoryStep(SagaStep): @property def name(self) -> str: return "reserve_inventory" def execute(self, context: SagaContext) -> dict: return {"reservation_id": str(uuid4()), "items": context.items} def compensate(self, context: SagaContext, step_result: dict) -> None: print(f"Compensating: Releasing inventory {step_result['reservation_id']}") class ShipOrderStep(SagaStep): @property def name(self) -> str: return "ship_order" def execute(self, context: SagaContext) -> dict: return {"tracking_number": f"TRK-{uuid4().hex[:8].upper()}"} def compensate(self, context: SagaContext, step_result: dict) -> None: print(f"Compensating: Canceling shipment {step_result['tracking_number']}") class OrderProcessingSaga: def __init__(self): self._steps: list[SagaStep] = [ CreateOrderStep(), ProcessPaymentStep(), ReserveInventoryStep(), ShipOrderStep() ] def execute(self, context: SagaContext) -> bool: try: for step in self._steps: context.current_step = step.name result = step.execute(context) context.step_results[step.name] = result context.is_completed = True return True except Exception as e: print(f"Saga failed at step {context.current_step}: {e}") self._compensate(context) return False def _compensate(self, context: SagaContext) -> None: context.is_compensating = True completed_steps = [ name for name in self._steps if name in context.step_results ] for step_name in reversed(completed_steps): step = next(s for s in self._steps if s.name == step_name) result = context.step_results[step_name] try: step.compensate(context, result) except Exception as e: print(f"Compensation failed for {step_name}: {e}") class SagaOrchestrator: def __init__(self): self._saga_store: dict[UUID, SagaContext] = {} self._saga_factories: dict[str, callable] = { "order": lambda: OrderProcessingSaga() } def start_saga( self, saga_type: str, order_id: UUID, customer_id: UUID, items: list[dict], total: float ) -> UUID: saga_id = uuid4() context = SagaContext( saga_id=saga_id, order_id=order_id, customer_id=customer_id, items=items, total=total ) self._saga_store[saga_id] = context saga_factory = self._saga_factories.get(saga_type) if not saga_factory: raise ValueError(f"Unknown saga type: {saga_type}") saga = saga_factory() success = saga.execute(context) if success: print(f"Saga {saga_id} completed successfully") else: print(f"Saga {saga_id} failed and compensated") return saga_id ``` ### Saga with Retry and Timeout ```python from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime, timedelta from typing import Protocol from uuid import UUID @dataclass class SagaStepResult: step_name: str success: bool result: dict | None = None error: str | None = None retry_count: int = 0 class RetryableStep(ABC): @property @abstractmethod def name(self) -> str: pass @property @abstractmethod def max_retries(self) -> int: pass @property @abstractmethod def retry_delay_seconds(self) -> int: pass @abstractmethod def execute_with_retry(self, context: 'SagaContext') -> SagaStepResult: pass class ResilientSagaOrchestrator: def __init__(self): self._active_sagas: dict[UUID, 'SagaContext'] = {} self._timeout_minutes: int = 30 def execute_step( self, context: 'SagaContext', step: RetryableStep ) -> SagaStepResult: result = SagaStepResult(step.name, False) for attempt in range(step.max_retries + 1): try: result = SagaStepResult( step_name=step.name, success=True, result=step.execute_with_retry(context), retry_count=attempt ) break except TransientError as e: result.error = str(e) result.retry_count = attempt if attempt < step.max_retries: import time time.sleep(step.retry_delay_seconds) return result def start_with_timeout(self, context: 'SagaContext') -> None: self._active_sagas[context.saga_id] = context import threading timer = threading.Timer( self._timeout_minutes * 60, self._handle_timeout, args=[context.saga_id] ) timer.start() def _handle_timeout(self, saga_id: UUID) -> None: if saga_id in self._active_sagas: context = self._active_sagas[saga_id] print(f"Saga {saga_id} timed out, initiating compensation") self._compensate_all(context) del self._active_sagas[saga_id] def _compensate_all(self, context: 'SagaContext') -> None: pass class TransientError(Exception): pass ``` ### Saga Persistence and Recovery ```python from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime from typing import Protocol from uuid import UUID @dataclass class SagaState: saga_id: UUID saga_type: str status: str current_step: str step_results: dict created_at: datetime updated_at: datetime completed_at: datetime | None = None class SagaStore(Protocol):