Orchestration Saga — Centralized Distributed Transaction Coordinator
In this tutorial, you will learn about Orchestration Saga. We cover key concepts, practical examples, and best practices to help you master this topic.
Orchestration saga uses a central orchestrator service that directs each participant through the distributed transaction steps and invokes compensating actions when any step fails, providing clear control flow.
What You'll Learn
By the end of this lesson you will implement an orchestrator saga, send commands and handle responses from participant services, design compensating actions for rollback, handle orchestrator failures, and choose between orchestration and choreography.
Why It Matters
Unlike choreography where event flow is implicit, orchestration makes the transaction flow explicit and visible. This is critical for complex business processes with branching, parallel execution, or strict ordering requirements where tracing event chains becomes unmanageable.
Real-World Use
DodaZIP's account Migration flow uses an orchestration saga. The orchestrator commands each step in sequence: export data from old storage, import to new storage, update DNS, verify integrity, notify user. If any step fails, the orchestrator runs compensating actions for all completed steps.
flowchart TB
O[Orchestrator] -->|1. Reserve Inventory| I[Inventory Service]
O -->|2. Process Payment| P[Payment Service]
O -->|3. Create Shipment| S[Shipping Service]
O -->|4. Send Email| N[Notification Service]
I -->|Response| O
P -->|Response| O
S -->|Response| O
N -->|Response| O
style O fill:#2d3748,color:#fff
Orchestrator Pattern
How the central coordinator works.
# orchestrator_pattern.py
# Orchestrator saga fundamentals
def orchestrator_pattern():
print("Orchestration Saga Fundamentals")
print("=" * 40)
print()
concepts = [
{
"concept": "Orchestrator",
"desc": "A service that manages the entire saga. It knows all steps, order, and compensations.",
"example": "OrderOrchestrator service with a state machine"
},
{
"concept": "Command",
"desc": "The orchestrator sends commands to participant services telling them what to do.",
"example": "orchestrator.commands.processPayment { orderId, amount }"
},
{
"concept": "Reply",
"desc": "Participants reply with success or failure. The orchestrator decides next step.",
"example": "payment.success { transactionId } or payment.failed { reason }"
},
{
"concept": "Saga State",
"desc": "The orchestrator maintains the saga state in its own database for recovery.",
"example": "Pending, ProcessingPayment, PaymentFailed, Compensating"
},
{
"concept": "Compensation",
"desc": "When a step fails, the orchestrator sends compensating commands for all completed steps.",
"example": "compensate.reserveInventory, compensate.processPayment (refund)"
},
]
for c in concepts:
print(f"{c['concept']:20s}")
print(f" {c['desc']}")
print(f" Example: {c['example']}")
print()
orchestrator_pattern()
Implementing the Orchestrator
State machine-based orchestrator.
# orchestrator_impl.py
# Orchestrator implementation
def orchestrator_impl():
print("Orchestrator Implementation")
print("=" * 40)
print()
code = """
from enum import Enum
import json
class SagaState(Enum):
PENDING = "pending"
RESERVING_INVENTORY = "reserving_inventory"
PROCESSING_PAYMENT = "processing_payment"
CREATING_SHIPMENT = "creating_shipment"
COMPLETED = "completed"
COMPENSATING = "compensating"
FAILED = "failed"
class OrderOrchestrator:
def __init__(self, command_bus, saga_store):
self.command_bus = command_bus
self.saga_store = saga_store
def start_saga(self, order_data):
saga = {
"id": str(uuid.uuid4()),
"order_id": order_data["order_id"],
"state": SagaState.RESERVING_INVENTORY.value,
"data": order_data,
"completed_steps": [],
"created_at": datetime.now(timezone.utc).isoformat()
}
self.saga_store.save(saga)
self.command_bus.send("inventory.reserve", {
"saga_id": saga["id"],
"items": order_data["items"]
})
return saga
def handle_reply(self, reply):
saga = self.saga_store.get(reply["saga_id"])
if saga["state"] == SagaState.COMPENSATING.value:
return # Already compensating
step = reply["step"]
result = reply["result"]
if result == "success":
saga["completed_steps"].append(step)
if step == "inventory.reserve":
saga["state"] = SagaState.PROCESSING_PAYMENT.value
self.command_bus.send("payment.process", {
"saga_id": saga["id"],
"customer_id": saga["data"]["customer_id"],
"amount": saga["data"]["total"]
})
elif step == "payment.process":
saga["state"] = SagaState.CREATING_SHIPMENT.value
self.command_bus.send("shipping.create", {
"saga_id": saga["id"],
"order_id": saga["order_id"],
"address": saga["data"]["shipping_address"]
})
elif step == "shipping.create":
saga["state"] = SagaState.COMPLETED.value
self.notify_completion(saga)
else:
self.compensate(saga, step, reply.get("reason"))
self.saga_store.update(saga)
def compensate(self, saga, failed_step, reason):
saga["state"] = SagaState.COMPENSATING.value
saga["failure_reason"] = reason
# Send compensating commands in reverse order
for step in reversed(saga["completed_steps"]):
if step == "inventory.reserve":
self.command_bus.send("inventory.release", {
"saga_id": saga["id"],
"items": saga["data"]["items"]
})
elif step == "payment.process":
self.command_bus.send("payment.refund", {
"saga_id": saga["id"],
"amount": saga["data"]["total"]
})
saga["state"] = SagaState.FAILED.value
"""
print(code)
orchestrator_impl()
Participant Services
How services handle orchestrator commands.
# participant_service.py
# Service participating in orchestrated saga
def participant_service():
print("Saga Participant Service")
print("=" * 40)
print()
participant_code = """
class InventoryParticipant:
"""Participates in the order saga."""
def handle_reserve_command(self, command):
saga_id = command["saga_id"]
try:
for item in command["items"]:
self.reserve_stock(item["product_id"], item["quantity"])
self.command_bus.reply({
"saga_id": saga_id,
"step": "inventory.reserve",
"result": "success"
})
except OutOfStockError as e:
self.command_bus.reply({
"saga_id": saga_id,
"step": "inventory.reserve",
"result": "failure",
"reason": str(e)
})
def handle_release_command(self, command):
"""Compensating action."""
for item in command["items"]:
self.release_stock(item["product_id"], item["quantity"])
print(f"Released stock for saga {command['saga_id']}")
class PaymentParticipant:
def handle_process_command(self, command):
try:
transaction = self.charge(
command["customer_id"],
command["amount"]
)
self.command_bus.reply({
"saga_id": command["saga_id"],
"step": "payment.process",
"result": "success",
"transaction_id": transaction.id
})
except PaymentError as e:
self.command_bus.reply({
"saga_id": command["saga_id"],
"step": "payment.process",
"result": "failure",
"reason": str(e)
})
def handle_refund_command(self, command):
"""Compensating action."""
self.refund(command["transaction_id"])
print(f"Refunded saga {command['saga_id']}")
"""
print(participant_code)
participant_service()
Saga State Persistence
Persisting saga state for recovery.
# saga_persistence.py
# Saga state persistence
def saga_persistence():
print("Saga State Persistence and Recovery")
print("=" * 45)
print()
persistence_code = """
import psycopg2
class SagaStore:
"""Persists saga state to PostgreSQL."""
def save(self, saga):
with self.conn.cursor() as cur:
cur.execute("""
INSERT INTO sagas (id, order_id, state, data,
completed_steps, created_at)
VALUES (%s, %s, %s, %s, %s, %s)
""", (
saga["id"], saga["order_id"], saga["state"],
json.dumps(saga["data"]),
json.dumps(saga["completed_steps"]),
saga["created_at"]
))
self.conn.commit()
def get(self, saga_id):
with self.conn.cursor() as cur:
cur.execute(
"SELECT * FROM sagas WHERE id = %s", (saga_id,)
)
row = cur.fetchone()
if row:
return {
"id": row[0],
"order_id": row[1],
"state": row[2],
"data": json.loads(row[3]),
"completed_steps": json.loads(row[4]),
"created_at": row[5]
}
return None
def get_pending_sagas(self):
"""For recovery after orchestrator restart."""
with self.conn.cursor() as cur:
cur.execute("""
SELECT * FROM sagas
WHERE state NOT IN ('completed', 'failed')
""")
return [self._row_to_dict(row) for row in cur.fetchall()]
def update(self, saga):
with self.conn.cursor() as cur:
cur.execute("""
UPDATE sagas SET state = %s, completed_steps = %s
WHERE id = %s
""", (
saga["state"],
json.dumps(saga["completed_steps"]),
saga["id"]
))
self.conn.commit()
def recover(self):
"""After restart, resume pending sagas."""
pending = self.get_pending_sagas()
for saga in pending:
print(f"Resuming saga {saga['id']} in state {saga['state']}")
if saga["state"] == "processing_payment":
# Re-send payment command
self.command_bus.send("payment.process", {...})
"""
print(persistence_code)
saga_persistence()
Common Mistakes
Orchestrator becoming a monolith: The orchestrator can grow into a god service that knows everything. Keep it focused on coordination only, not business logic.
No saga persistence: If the orchestrator crashes without persisting state, all in-flight sagas are lost. Always persist saga state before sending commands.
Blocking the orchestrator: The orchestrator should be async. Blocking on participant replies ties up resources and limits throughput.
Tight coupling to participants: The orchestrator should use commands, not service-specific APIs. Abstract participant interaction behind command interfaces.
Missing timeout for replies: A participant may never reply. Implement timeouts with automatic compensation for unresponsive participants.
Practice Questions
What is the role of the orchestrator in an orchestration saga? It coordinates the distributed transaction by sending commands to participants, collecting replies, and deciding the next step.
How does the orchestrator handle failures? It runs compensating actions in reverse order for all completed steps, then marks the saga as failed.
Why must saga state be persisted in a database? So the orchestrator can recover in-flight sagas after a restart or crash without losing progress.
What is the difference between a command and an event in sagas? Commands tell a service what to do (imperative). Events announce something that happened (declarative). Orchestration uses commands; choreography uses events.
Challenge: Design an orchestration saga for a multi-step account deletion Process. Steps: verify identity, backup user data, delete from primary DB, delete from analytics, send confirmation. Include compensating actions and timeout handling.
FAQ
Mini Project
Build an orchestration saga for a cloud service provisioning flow. The orchestrator coordinates: validate payment method, provision VMs, configure networking, deploy application, update DNS. Each step has a compensating action. Persist saga state and implement recovery.
def cloud_provisioning_saga():
print("Cloud Provisioning Orchestration Saga")
print("=" * 45)
print()
print("Orchestrator Steps:")
print()
print(" 1. Validate Payment Method")
print(" Command: billing.validate")
print(" Comp: N/A (no side effect)")
print()
print(" 2. Provision VMs")
print(" Command: compute.provision")
print(" Comp: compute.deprovision")
print()
print(" 3. Configure Networking")
print(" Command: network.configure")
print(" Comp: network.deconfigure")
print()
print(" 4. Deploy Application")
print(" Command: deploy.run")
print(" Comp: deploy.rollback")
print()
print(" 5. Update DNS")
print(" Command: dns.update")
print(" Comp: dns.restore")
print()
print("Recovery:")
print(" Persist saga in 'provisioning_sagas' table")
print(" On restart, resume any sagas in 'processing_*' state")
cloud_provisioning_saga()
What's Next
Next: Service Discovery for dynamic service location in Microservices.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro