saga-orchestration

name	saga-orchestration
description	Implement saga patterns for distributed transactions and cross-aggregate workflows. Use when coordinating multi-step business processes, handling compensating transactions, or managing long-running workflows.
license	MIT
metadata	{"version":"1.0.0","domain":"architecture","triggers":"saga pattern, distributed transactions, compensating transactions, long-running workflows, orchestration","role":"architect","scope":"system-design","output-format":"code","related-skills":"event-sourcing-architect, cqrs-implementation, microservices-patterns"}

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[],
                
            )

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

Resources

• Saga Pattern
• Designing Data-Intensive Applications