saga-pattern

name	Saga Pattern
description	Managing distributed transactions through coordinated sequences of local transactions with compensating actions
category	software-development

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

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 :
     ():
        ._event_bus = event_bus
        ._orders: [UUID, ] = {}
    
     () -> OrderCreated:
        order_id = uuid4()
        total = (item[] * item[]  item  items)
        
        order = {
            : order_id,
            : customer_id,
            : items,
            : total,
            : 
        }
        ._orders[order_id] = order
        
        event = OrderCreated(
            order_id=order_id,
            customer_id=customer_id,
            items=items,
            total=total
        )
        ._event_bus.publish(event)
         event

 :
     ():
        ._event_bus = event_bus
        ._payments: [UUID, ] = {}
    
     () -> :
        payment_id = uuid4()
        payment = {
            : payment_id,
            : event.order_id,
            : event.total,
            : 
        }
        ._payments[payment_id] = payment
        
        payment_event = PaymentProcessed(
            order_id=event.order_id,
            payment_id=payment_id,
            amount=event.total
        )
        ._event_bus.publish(payment_event)
    
     () -> :
         payment  ._payments.values():
             payment[] == order_id:
                payment[] = 

 :
     ():
        ._event_bus = event_bus
        ._reservations: [UUID, ] = {}
        ._inventory: [, ] = {: , : }
    
     () -> :
        reservation_id = uuid4()
        reserved_items = []
        
         item  event.items:
            product_id = item[]
             ._inventory.get(product_id, ) >= item[]:
                ._inventory[product_id] -= item[]
                reserved_items.append(item)
        
        reservation = {
            : reservation_id,
            : event.order_id,
            : reserved_items,
            : 
        }
        ._reservations[reservation_id] = reservation
        
        ._event_bus.publish(InventoryReserved(
            order_id=event.order_id,
            reservation_id=reservation_id,
            items=reserved_items
        ))
    
     () -> :
         reservation  ._reservations.values():
             reservation[] == order_id:
                 item  reservation[]:
                    ._inventory[item[]] += item[]
                reservation[] = 

 :
     ():
        ._event_bus = event_bus
        ._shipments: [UUID, ] = {}
    
     () -> :
        shipment_id = uuid4()
        tracking = 
        
        shipment = {
            : shipment_id,
            : event.order_id,
            : tracking,
            : 
        }
        ._shipments[shipment_id] = shipment
        
        ._event_bus.publish(OrderShipped(
            order_id=event.order_id,
            tracking_number=tracking,
            carrier=
        ))

Orchestration-Based Saga

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

 ():

     () -> :
         
    
     () -> :
         {: (context.order_id), : }
    
     () -> :
        ()

 ():

     () -> :
         
    
     () -> :
         {: (uuid4()), : context.total}
    
     () -> :
        ()

 ():

     () -> :
         
    
     () -> :
         {: (uuid4()), : context.items}
    
     () -> :
        ()

 ():

     () -> :
         
    
     () -> :
         {: }
    
     () -> :
        ()

 :
     ():
        ._steps: [SagaStep] = [
            CreateOrderStep(),
            ProcessPaymentStep(),
            ReserveInventoryStep(),
            ShipOrderStep()
        ]
    
     () -> :
        :
             step  ._steps:
                context.current_step = step.name
                result = step.execute(context)
                context.step_results[step.name] = result
            
            context.is_completed = 
             
        
         Exception  e:
            ()
            ._compensate(context)
             
    
     () -> :
        context.is_compensating = 
        completed_steps = [
            name  name  ._steps
             name  context.step_results
        ]
        
         step_name  (completed_steps):
            step = (s  s  ._steps  s.name == step_name)
            result = context.step_results[step_name]
            :
                step.compensate(context, result)
             Exception  e:
                ()

 :
     ():
        ._saga_store: [UUID, SagaContext] = {}
        ._saga_factories: [, ] = {
            : : OrderProcessingSaga()
        }
    
     () -> UUID:
        saga_id = uuid4()
        context = SagaContext(
            saga_id=saga_id,
            order_id=order_id,
            customer_id=customer_id,
            items=items,
            total=total
        )
        ._saga_store[saga_id] = context
        
        saga_factory = ._saga_factories.get(saga_type)
          saga_factory:
             ValueError()
        
        saga = saga_factory()
        success = saga.execute(context)
        
         success:
            ()
        :
            ()
        
         saga_id

Saga with Retry and Timeout

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: ) -> SagaStepResult:
        

 :
     ():
        ._active_sagas: [UUID, ] = {}
        ._timeout_minutes:  = 
    
     () -> SagaStepResult:
        result = SagaStepResult(step.name, )
        
         attempt  (step.max_retries + ):
            :
                result = SagaStepResult(
                    step_name=step.name,
                    success=,
                    result=step.execute_with_retry(context),
                    retry_count=attempt
                )
                
            
             TransientError  e:
                result.error = (e)
                result.retry_count = attempt
                 attempt < step.max_retries:
                     time
                    time.sleep(step.retry_delay_seconds)
        
         result
    
     () -> :
        ._active_sagas[context.saga_id] = context
         threading
        timer = threading.Timer(
            ._timeout_minutes * ,
            ._handle_timeout,
            args=[context.saga_id]
        )
        timer.start()
    
     () -> :
         saga_id  ._active_sagas:
            context = ._active_sagas[saga_id]
            ()
            ._compensate_all(context)
             ._active_sagas[saga_id]
    
     () -> :
        

 ():
    

Saga Persistence and Recovery

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):
    @abstractmethod
    def save(self, state: SagaState) -> None:
        pass
    
    @abstractmethod
    def load(self, saga_id: UUID) -> SagaState | None:
        pass
    
    @abstractmethod
    def update_status(self, saga_id: UUID, status: str, current_step: str) -> None:
        pass
    
    @abstractmethod
    def get_pending_sagas(self) -> list[SagaState]:
        pass

class :
     ():
        ._store = saga_store
        ._active_sagas: [UUID, ] = {}
    
     () -> :
        context = SagaContext(
            saga_id=saga_id,
            order_id=initial_data.get(, uuid4()),
            customer_id=initial_data.get(, uuid4()),
            items=initial_data.get(, []),
            total=initial_data.get(, )
        )
        
        ._active_sagas[saga_id] = context
        
        state = SagaState(
            saga_id=saga_id,
            saga_type=saga_type,
            status=,
            current_step=,
            step_results={},
            created_at=datetime.utcnow(),
            updated_at=datetime.utcnow()
        )
        ._store.save(state)
    
     () -> :
        pending = ._store.get_pending_sagas()
         state  pending:
            ._resume_saga(state)
    
     () -> :
         state.saga_id   ._active_sagas:
            ()
        
        context = ._active_sagas.get(state.saga_id)
         context:
            context.step_results = state.step_results
    
     () -> :
        ._store.update_status(saga_id, , step_name)
    
     () -> :
         saga_id  ._active_sagas:
            ._active_sagas[saga_id].step_results[step_name] = result
        ._store.save(SagaState(
            saga_id=saga_id,
            saga_type=,
            status=,
            current_step=step_name,
            step_results={},
            created_at=datetime.utcnow(),
            updated_at=datetime.utcnow()
        ))

Distributed Saga with Compensation

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Protocol

@dataclass
class BookingRequest:
    hotel_id: str
    flight_id: str
    customer_id: str
    start_date: str
    end_date: str

@dataclass
class HotelBooking:
    confirmation_number: str

@dataclass
class FlightBooking:
    reservation_code: str

class HotelService(Protocol):
    def book(self, request: BookingRequest) -> HotelBooking:
        pass
    
    def cancel(self, confirmation_number: str) -> None:
        pass

class FlightService(Protocol):
    def book(self, request: BookingRequest) -> FlightBooking:
        pass
    
    def cancel(self, reservation_code: str) -> None:
        pass

class TripBookingSaga:
     ():
        ._hotel = hotel_service
        ._flight = flight_service
        ._completed_steps: [] = []
    
     () -> :
        :
            hotel_booking = ._hotel.book(request)
            ._completed_steps.append()
            
            flight_booking = ._flight.book(request)
            ._completed_steps.append()
            
             {
                : ,
                : hotel_booking.confirmation_number,
                : flight_booking.reservation_code
            }
        
         Exception  e:
            ()
            ._compensate()
            
    
     () -> :
         step  (._completed_steps):
             step == :
                 booking  ._flight_bookings:
                    ._flight.cancel(booking.reservation_code)
             step == :
                 booking  ._hotel_bookings:
                    ._hotel.cancel(booking.confirmation_number)
        ._completed_steps.clear()
    

     () -> [FlightBooking]:
         []
    

     () -> [HotelBooking]:
         []

Best Practices

1. Keep Sagas Short: Limit the number of steps
2. Idempotent Steps: Handle retries safely
3. Compensating Logic: Each step must have undo logic
4. Timeout Handling: Prevent hung sagas
5. Saga State Persistence: Recover from crashes
6. Avoid Cross-Dependencies: Steps should be independent
7. Eventual Consistency: Accept intermediate states
8. Testing: Test happy path and all failure scenarios
9. Monitoring: Track saga execution times
10. Documentation: Document saga flows clearly
11. Choreography vs Orchestration: Choose based on complexity
12. Retry Strategies: Handle transient failures with backoff