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

Templates

Template 1: Saga Orchestrator Base

```python 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: int = 0 created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow)

class SagaOrchestrator(ABC): """Base class for saga orchestrators."""

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

@abstractmethod
def define_steps(self, data: Dict) -> List[SagaStep]:
    """Define the saga steps."""
    pass

@property
@abstractmethod
def saga_type(self) -> str:
    """Unique saga type identifier."""
    pass

async def start(self, data: Dict) -> Saga:
    """Start a new saga."""
    saga = Saga(
        saga_id=str(uuid.uuid4()),
        saga_type=self.saga_type,
        state=SagaState.STARTED,
        data=data,
        steps=self.define_steps(data)
    )
    await self.saga_store.save(saga)
    await self._execute_next_step(saga)
    return saga

async def handle_step_completed(self, saga_id: str, step_name: str, result: Dict):
    """Handle successful step completion."""
    saga = await self.saga_store.get(saga_id)

    # Update step
    for step in saga.steps:
        if step.name == step_name:
            step.status = "completed"
            step.result = result
            step.executed_at = datetime.utcnow()
            break

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

    # Check if saga is complete
    if saga.current_step >= len(saga.steps):
        saga.state = SagaState.COMPLETED
        await self.saga_store.save(saga)
        await self._on_saga_completed(saga)
    else:
        saga.state = SagaState.PENDING
        await self.saga_store.save(saga)
        await self._execute_next_step(saga)

async def handle_step_failed(self, saga_id: str, step_name: str, error: str):
    """Handle step failure - start compensation."""
    saga = await self.saga_store.get(saga_id)

    # Mark step as failed
    for step in saga.steps:
        if step.name == step_name:
            step.status = "failed"
            step.error = error
            break

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

    # Start compensation from current step backwards
    await self._compensate(saga)

async