saga-orchestration

name	saga-orchestration
description	Coordinates multi-service workflows as orchestration or choreography sagas with compensating transactions, durable saga_id state, and step timeouts. Use when order fulfillment, approvals, or distributed rollback spans separate databases and two-phase commit is not viable. Not for single-service local transactions. Never treat sagas as ACID across services.
version	1.0.1
risk	unknown
source	community
date_added	2026-02-27

Saga Orchestration

Patterns and templates for managing distributed transactions and long-running business processes using orchestration or choreography approaches, with compensating transactions for rollback.

When to Use

Use this skill when:

• Coordinating multi-service transactions that span separate databases or bounded contexts
• Implementing compensating transactions for distributed rollback
• Managing long-running business workflows (order fulfillment, approval pipelines, booking flows)
• Handling failures in distributed systems where two-phase commit is not viable
• Building order fulfillment processes across inventory, payment, shipping, and notification services
• Implementing approval workflows with multi-step state transitions
• Designing event-driven choreography sagas where no central coordinator is desired

Do Not Use This Skill When

• The task is unrelated to saga orchestration or distributed transactions
• You need a different domain or tool outside this scope
• A single-service local transaction suffices (use standard database transactions)
• You need ACID guarantees across all services (sagas provide eventual consistency only)

Prerequisites

• Familiarity with asynchronous messaging / event-driven architecture
• A message broker or event bus (e.g., Kafka, RabbitMQ, AWS SNS/SQS, Azure Service Bus)
• A persistence store for saga state (e.g., PostgreSQL, DynamoDB, Cosmos DB)
• Python 3.8+ if using the provided templates directly (templates are language-agnostic in concept but written in Python)
• On Windows host (PowerShell), ensure Python is on PATH:

python --version

Overview

Saga Types

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

Choreography: Each service reacts to events and emits the next event. No central coordinator. Best for simpler flows with few steps.

Orchestration: A central orchestrator issues commands to services and tracks state. Best for complex flows with many steps, conditional logic, or strict ordering.

Saga Execution States

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

Procedure

Step 1: Choose Saga Type

1. Choreography — Use when the flow is linear, has ≤ 3–4 steps, and services are loosely coupled. Each service subscribes to events and emits the next event.
2. Orchestration — Use when the flow has conditional branching, many steps, or you need a single source of truth for saga state. A central orchestrator manages step execution and compensation.

Step 2: Define Saga Steps and Compensations

For each step, identify:

• Action: The command or event that triggers the step
• Compensation: The reverse operation that undoes the step's effects
• Idempotency key: Ensure the action can be safely retried

Example for order fulfillment:

Step	Action	Compensation
Reserve inventory	InventoryService.ReserveItems	InventoryService.ReleaseReservation
Process payment	PaymentService.ProcessPayment	PaymentService.RefundPayment
Create shipment	ShippingService.CreateShipment	ShippingService.CancelShipment
Send confirmation	NotificationService.SendConfirmation	NotificationService.SendCancellation

Step 3: Implement the Orchestrator (Orchestration Pattern)

Use the base orchestrator template below. The orchestrator:

1. Creates a saga with a unique saga_id
2. Persists saga state to a durable store
3. Dispatches step actions via event publisher
4. Handles step completion / failure callbacks
5. Compensates in reverse order on failure

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}
        )

Step 4: Implement a Concrete 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"
            )
        ]


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

Step 5: Implement Service-Side Handlers

Each service handles action commands and reports success or failure back to the orchestrator:

class InventoryService:
    async def handle_reserve_items(self, command: Dict):
        try:
            reservation = await self.reserve(
                command["items"],
                command["order_id"]
            )
            await self.event_publisher.publish(
                "SagaStepCompleted",
                {
                    "saga_id": command["saga_id"],
                    "step_name": "reserve_inventory",
                    "result": {"reservation_id": reservation.id}
                }
            )
        except InsufficientInventoryError as e:
            await self.event_publisher.publish(
                "SagaStepFailed",
                {
                    "saga_id": command["saga_id"],
                    "step_name": "reserve_inventory",
                    "error": str(e)
                }
            )

    async def handle_release_reservation(self, command: Dict):
        await self.release_reservation(
            command["original_result"]["reservation_id"]
        )
        await self.event_publisher.publish(
            ,
            {
                : command[],
                : 
            }
        )

Step 6: Implement Choreography-Based Saga (Alternative)

When no central orchestrator is desired, use event chaining:

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("ShipmentFailed", self._on_shipment_failed)

    async def _on_order_created(self, event: Dict):
        await self.event_bus.publish("ReserveInventory", {
            "saga_id": event["order_id"],
            "order_id": event["order_id"],
            "items": 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[]
        })

Step 7: Add Timeouts

Never let a saga step wait indefinitely. Schedule timeout checks:

from datetime import timedelta

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)

        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):
        saga = await self.saga_store.get(data["saga_id"])
        step = (s  s  saga.steps  s.name == data[])

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

Step 8: Consider Durable Execution Frameworks

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.

Step 9: Load Detailed Examples

If detailed examples or extended implementation patterns are required, open resources/implementation-playbook.md for additional walkthroughs, edge cases, and testing strategies.

Examples

Minimal Saga Step Definition

steps = [
    SagaStep(
        name="reserve_inventory",
        action="InventoryService.ReserveItems",
        compensation="InventoryService.ReleaseReservation"
    ),
    SagaStep(
        name="process_payment",
        action="PaymentService.ProcessPayment",
        compensation="PaymentService.RefundPayment"
    ),
]

Compensation Flow

1. Step 3 (create_shipment) fails
2. Orchestrator sets state to COMPENSATING
3. Compensate step 2: PaymentService.RefundPayment
4. Compensate step 1: InventoryService.ReleaseReservation
5. All compensations complete → state becomes FAILED
6. OrderFulfillmentFailed event published

Pitfalls

• Do not assume instant completion — Sagas are long-running by nature. Steps may take seconds to minutes. Always persist state and handle async callbacks.
• Do not skip compensation testing — Compensations are the most critical part. A failed compensation leaves the system in an inconsistent state. Test every compensation path explicitly.
• Do not couple services synchronously — Use async messaging. Synchronous calls between services in a saga create cascading failures and tight coupling.
• Do not ignore partial failures — A step may partially succeed (e.g., payment charged but response lost). Design steps to be idempotent so retries are safe.
• Do not forget correlation IDs — Without a saga_id propagated through all events, tracing and debugging distributed failures becomes nearly impossible.
• Do not omit timeouts — A step that never responds will hang the saga forever. Always schedule timeout checks.
• Do not use sagas when ACID is required — Sagas provide eventual consistency. If you need strict ACID across services, reconsider your service boundaries or use a shared database.
• Do not compensate steps that were never completed — The _compensate method checks step.status == "completed" before compensating. Ensure your implementation preserves this guard.
• Do not lose saga state on crash — The saga store must be durable. In-memory state is unacceptable for production sagas. Persist after every state transition.

Verification

Verify Saga State Machine

Check that all state transitions are valid:

# Valid transitions:
# STARTED → PENDING (first step dispatched)
# PENDING → PENDING (step completed, next dispatched)
# PENDING → COMPLETED (last step completed)
# PENDING → COMPENSATING (step failed)
# COMPENSATING → FAILED (all compensations done)

Verify Idempotency

Each step action and compensation must be safely retriable:

# Test: calling reserve_inventory twice with same order_id
# should not create two reservations.
reservation = await inventory_service.reserve(items, order_id)
reservation_2 = await inventory_service.reserve(items, order_id)
assert reservation.id == reservation_2.id  # Idempotent

Verify Compensation Reverses Action

# Test: reserve then release should restore inventory to original state
original_count = await get_inventory(item_id)
await reserve(item_id, quantity=5)
await release_reservation(reservation_id)
assert await get_inventory(item_id) == original_count

Verify Timeout Handling

# Test: step that never responds should trigger timeout
saga = await timeout_orchestrator.start(test_data)
# Wait longer than timeout period
await asyncio.sleep(timeout_minutes * 60 + 1)
saga = await saga_store.get(saga.saga_id)
assert saga.state == SagaState.FAILED

Verify Saga Persistence

# On Windows PowerShell, verify saga records exist in your store
# Example for PostgreSQL:
psql -U YOUR_USER -d YOUR_DB -c "SELECT saga_id, saga_type, state FROM sagas WHERE saga_type = 'OrderFulfillment';"

Expected output should show saga records with states completed, failed, or pending.

Best Practices

Do's

• Make steps idempotent — Safe to retry without side effects
• Design compensations carefully — They must reliably reverse the action
• Use correlation IDs — Propagate saga_id through all events for tracing
• Implement timeouts — Never wait forever for a step response
• Log everything — Every state transition, step dispatch, and compensation for debugging
• Persist state after every transition — Crash recovery depends on durable state
• Version your saga definitions — Changes to step order or compensation logic require migration strategy for in-flight sagas

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 — microservices.io
• Designing Data-Intensive Applications
• resources/implementation-playbook.md — Load when detailed examples or extended implementation patterns are needed

Limitations

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