| 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"
)
]
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)
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)
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
