saga-orchestration

name	saga-orchestration
description	Patterns for managing distributed transactions and long-running business processes.
risk	unknown
source	community
date_added	2026-02-27

Saga Orchestration

Patterns for managing distributed transactions and long-running business processes.

Do not use this skill when

• The task is unrelated to saga orchestration
• You need a different domain or tool outside this scope

Instructions

• Clarify goals, constraints, and required inputs.
• Apply relevant best practices and validate outcomes.
• Provide actionable steps and verification.
• If detailed examples are required, open resources/implementation-playbook.md.

Use this skill when

• Coordinating multi-service transactions
• Implementing compensating transactions
• Managing long-running business workflows
• Handling failures in distributed systems
• Building order fulfillment processes
• Implementing approval workflows

Core Concepts

1. Saga Types

Choreography                    Orchestration
┌─────┐  ┌─────┐  ┌─────┐     ┌─────────────┐
│Svc A│─►│Svc B│─►│Svc C│     │ Orchestrator│
└─────┘  └─────┘  └─────┘     └──────┬──────┘
   │        │        │               │
   ▼        ▼        ▼         ┌─────┼─────┐
 Event    Event    Event       ▼     ▼     ▼
                            ┌────┐┌────┐┌────┐
                            │Svc1││Svc2││Svc3│
                            └────┘└────┘└────┘

2. Saga Execution States

State	Description
Started	Saga initiated
Pending	Waiting for step completion
Compensating	Rolling back due to failure
Completed	All steps succeeded
Failed	Saga failed after compensation

Templates

Template 1: Saga Orchestrator Base

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Any, Optional
from datetime import datetime
import uuid

class SagaState(Enum):
    STARTED = "started"
    PENDING = "pending"
    COMPENSATING = "compensating"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class SagaStep:
    name: str
    action: str
    compensation: str
    status: str = "pending"
    result: Optional[Dict] = None
    error: Optional[str] = None
    executed_at: Optional[datetime] = None
    compensated_at: Optional[datetime] = None


@dataclass
class Saga:
    saga_id: str
    saga_type: str
    state: SagaState
    data: Dict[str, Any]
    steps: List[SagaStep]
    current_step:  = 
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)


 ():
    

     ():
        .saga_store = saga_store
        .event_publisher = event_publisher


     () -> [SagaStep]:
        
        



     () -> :
        
        

      () -> Saga:
        
        saga = Saga(
            saga_id=(uuid.uuid4()),
            saga_type=.saga_type,
            state=SagaState.STARTED,
            data=data,
            steps=.define_steps(data)
        )
         .saga_store.save(saga)
         ._execute_next_step(saga)
         saga

      ():
        
        saga =  .saga_store.get(saga_id)

        
         step  saga.steps:
             step.name == step_name:
                step.status = 
                step.result = result
                step.executed_at = datetime.utcnow()
                

        saga.current_step += 
        saga.updated_at = datetime.utcnow()

        
         saga.current_step >= (saga.steps):
            saga.state = SagaState.COMPLETED
             .saga_store.save(saga)
             ._on_saga_completed(saga)
        :
            saga.state = SagaState.PENDING
             .saga_store.save(saga)
             ._execute_next_step(saga)

      ():
        
        saga =  .saga_store.get(saga_id)

        
         step  saga.steps:
             step.name == step_name:
                step.status = 
                step.error = error
                

        saga.state = SagaState.COMPENSATING
        saga.updated_at = datetime.utcnow()
         .saga_store.save(saga)

        
         ._compensate(saga)

      ():
        
         saga.current_step >= (saga.steps):
            

        step = saga.steps[saga.current_step]
        step.status = 
         .saga_store.save(saga)

        
         .event_publisher.publish(
            step.action,
            {
                : saga.saga_id,
                : step.name,
                **saga.data
            }
        )

      ():
        
        
         i  (saga.current_step - , -, -):
            step = saga.steps[i]
             step.status == :
                step.status = 
                 .saga_store.save(saga)

                 .event_publisher.publish(
                    step.compensation,
                    {
                        : saga.saga_id,
                        : step.name,
                        : step.result,
                        **saga.data
                    }
                )

      ():
        
        saga =  .saga_store.get(saga_id)

         step  saga.steps:
             step.name == step_name:
                step.status = 
                step.compensated_at = datetime.utcnow()
                

        
        all_compensated = (
            s.status  (, , )
             s  saga.steps
        )

         all_compensated:
            saga.state = SagaState.FAILED
             ._on_saga_failed(saga)

         .saga_store.save(saga)

      ():
        
         .event_publisher.publish(
            ,
            {: saga.saga_id, **saga.data}
        )

      ():
        
         .event_publisher.publish(
            ,
            {: saga.saga_id, : , **saga.data}
        )

Template 2: Order Fulfillment Saga

class OrderFulfillmentSaga(SagaOrchestrator):
    """Orchestrates order fulfillment across services."""

    @property
    def saga_type(self) -> str:
        return "OrderFulfillment"

    def define_steps(self, data: Dict) -> List[SagaStep]:
        return [
            SagaStep(
                name="reserve_inventory",
                action="InventoryService.ReserveItems",
                compensation="InventoryService.ReleaseReservation"
            ),
            SagaStep(
                name="process_payment",
                action="PaymentService.ProcessPayment",
                compensation="PaymentService.RefundPayment"
            ),
            SagaStep(
                name="create_shipment",
                action="ShippingService.CreateShipment",
                compensation="ShippingService.CancelShipment"
            ),
            SagaStep(
                name="send_confirmation",
                action="NotificationService.SendOrderConfirmation",
                compensation="NotificationService.SendCancellationNotice"
            )
        ]


# Usage
async def create_order(order_data: Dict):
    saga = OrderFulfillmentSaga(saga_store, event_publisher)
    return await saga.start({
        "order_id": order_data["order_id"],
        : order_data[],
        : order_data[],
        : order_data[],
        : order_data[]
    })



 :
      ():
        :
            
            reservation =  .reserve(
                command[],
                command[]
            )
            
             .event_publisher.publish(
                ,
                {
                    : command[],
                    : ,
                    : {: reservation.}
                }
            )
         InsufficientInventoryError  e:
             .event_publisher.publish(
                ,
                {
                    : command[],
                    : ,
                    : (e)
                }
            )

      ():
        
         .release_reservation(
            command[][]
        )
         .event_publisher.publish(
            ,
            {
                : command[],
                : 
            }
        )

Template 3: Choreography-Based Saga

from dataclasses import dataclass
from typing import Dict, Any
import asyncio

@dataclass
class SagaContext:
    """Passed through choreographed saga events."""
    saga_id: str
    step: int
    data: Dict[str, Any]
    completed_steps: list


class OrderChoreographySaga:
    """Choreography-based saga using events."""

    def __init__(self, event_bus):
        self.event_bus = event_bus
        self._register_handlers()

    def _register_handlers(self):
        self.event_bus.subscribe("OrderCreated", self._on_order_created)
        self.event_bus.subscribe("InventoryReserved", self._on_inventory_reserved)
        self.event_bus.subscribe("PaymentProcessed", self._on_payment_processed)
        self.event_bus.subscribe("ShipmentCreated", self._on_shipment_created)

        # Compensation handlers
        self.event_bus.subscribe("PaymentFailed", self._on_payment_failed)
        self.event_bus.subscribe(, ._on_shipment_failed)

      ():
        
         .event_bus.publish(, {
            : event[],
            : event[],
            : event[]
        })

      ():
        
         .event_bus.publish(, {
            : event[],
            : event[],
            : event[],
            : event[]
        })

      ():
        
         .event_bus.publish(, {
            : event[],
            : event[],
            : event[]
        })

      ():
        
         .event_bus.publish(, {
            : event[],
            : event[],
            : event[]
        })

    
      ():
        
         .event_bus.publish(, {
            : event[],
            : event[]
        })
         .event_bus.publish(, {
            : event[],
            : 
        })

      ():
        
         .event_bus.publish(, {
            : event[],
            : event[]
        })
         .event_bus.publish(, {
            : event[],
            : event[]
        })

Template 4: Saga with Timeouts

class TimeoutSagaOrchestrator(SagaOrchestrator):
    """Saga orchestrator with step timeouts."""

    def __init__(self, saga_store, event_publisher, scheduler):
        super().__init__(saga_store, event_publisher)
        self.scheduler = scheduler

    async def _execute_next_step(self, saga: Saga):
        if saga.current_step >= len(saga.steps):
            return

        step = saga.steps[saga.current_step]
        step.status = "executing"
        step.timeout_at = datetime.utcnow() + timedelta(minutes=5)
        await self.saga_store.save(saga)

        # Schedule timeout check
        await self.scheduler.schedule(
            f"saga_timeout_{saga.saga_id}_{step.name}",
            self._check_timeout,
            {"saga_id": saga.saga_id, "step_name": step.name},
            run_at=step.timeout_at
        )

        await self.event_publisher.publish(
            step.action,
            {"saga_id": saga.saga_id, "step_name": step.name, **saga.data}
        )

    async def _check_timeout(self, data: Dict):
        """Check if step has timed out."""
        saga = await self.saga_store.get(data[])
        step = (s  s  saga.steps  s.name == data[])

         step.status == :
            
             .handle_step_failed(
                data[],
                data[],
                
            )

Durable Execution Alternative

The templates above build saga infrastructure from scratch — saga stores, event publishers, compensation tracking. Durable execution frameworks (like DBOS) eliminate much of this boilerplate: the workflow runtime automatically persists state to a database, retries failed steps, and resumes from the last checkpoint after crashes. Instead of building a SagaOrchestrator base class, you write a workflow function with steps — the framework handles persistence, crash recovery, and exactly-once execution semantics. Consider durable execution when you want saga-like reliability without managing the coordination infrastructure yourself.

Best Practices

Do's

• Make steps idempotent - Safe to retry
• Design compensations carefully - They must work
• Use correlation IDs - For tracing across services
• Implement timeouts - Don't wait forever
• Log everything - For debugging failures

Don'ts

• Don't assume instant completion - Sagas take time
• Don't skip compensation testing - Most critical part
• Don't couple services - Use async messaging
• Don't ignore partial failures - Handle gracefully

Related Skills

Works well with: event-sourcing-architect, workflow-automation, dbos-*

Resources

• Saga Pattern
• Designing Data-Intensive Applications

Limitations

• Use this skill only when the task clearly matches the scope described above.
• Do not treat the output as a substitute for enprojectnment-specific validation, testing, or expert review.
• Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.