paulpas Cross-Domain Workflow Coordination with Sagas
Orchestrates and choreographs long-running business transactions across multiple bounded contexts using saga patterns. Implements compensating actions for failure recovery, timeout handling for unresponsive contexts, persistent saga state for restart resilience, and both orchestration (centralized coordinator) and choreography (decentralized event-driven) coordination styles — enabling eventual consistency without distributed locking or two-phase commit across service boundaries.
TL;DR Checklist
- Map every bounded context that participates in the business transaction before writing any code
- For each forward action that changes state in another context, define its compensating action first
- Choose orchestration for complex sagas with explicit error handling; choose choreography for simple linear flows where loose coupling matters more than traceability
- Persist saga instance state to durable storage after each step — never only at completion
- Use a single correlation ID that links all events and commands within one saga instance
- Make every compensating action idempotent — calling it twice must not cause double-refunds or double-releases
- Design compensations as domain-specific business rules, not generic "undo" operations
- Implement timeout detection per step and recover pending sagas on system restart
When to Use
Use this skill when:
- A single business operation must modify state in two or more bounded contexts that have independent databases and cannot share a transaction
- You need to maintain consistency across contexts where the full transaction may take seconds, minutes, or even hours (order placement with inventory reservation, payment processing, and shipping coordination)
- One of the participating contexts is external or third-party and you have no control over its availability or rollback capabilities
- You are designing an order management, booking, onboarding, or any multi-step business workflow that spans teams and services
- The system requires eventual consistency rather than strong consistency — you can tolerate a brief window where state is partially committed
When NOT to Use
Avoid this skill for:
- Single-context operations within one bounded context — use in-process domain events (
domain-events) or direct method calls instead. Sagas add orchestration overhead that is unnecessary inside one service boundary - Operations requiring strong consistency (ACID) across contexts — sagas provide eventual consistency only. If the business rule requires all-or-nothing atomicity with zero tolerance for intermediate states, redesign to keep everything within a single context or use a compensatable saga with extremely short execution time
- Simple request-response workflows where the entire operation completes in milliseconds — the state persistence and recovery overhead of a saga is not justified for fast operations
- When any participant context cannot implement compensating actions — if you cannot define a business-meaningful compensation for every forward step, do not use a saga; redesign the workflow or negotiate with the other team
Core Workflow
Map the Cross-Domain Transaction — List every bounded context that must participate in this business operation. For each context, identify: what command it receives from the saga, what state it modifies, and what compensating action reverses that modification. Draw the forward step sequence left-to-right and the compensation sequence right-to-left on a shared diagram. Checkpoint: Every context has exactly one forward action and one compensating action defined before proceeding. No context is "we'll figure out compensation later."
Choose Orchestration vs Choreography — Apply these decision criteria: choose orchestration when the saga has more than three steps, requires explicit error handling for individual steps, or when business operators need a central view of progress; choose choreography when the flow is strictly linear (each step triggers exactly one next step), loose coupling between contexts is more important than traceability, and you do not need a coordinator to make routing decisions. Checkpoint: Document the decision rationale — if switching to the alternative pattern later, you need to understand why you chose this one.
Define Saga State Schema — Design a persistent schema that tracks:
saga_instance_id(the saga's own ID),current_step_index(which step has been reached most recently),context_ids(correlation IDs for each participant context),status(RUNNING, COMPLETED, COMPENSATING, FAILED, TIMEOUT), andcreated_at/updated_attimestamps. This schema enables recovery after system restarts — the saga must resume or compensate from where it left off. Checkpoint: Every field in the schema is necessary for either resumption logic or observability. Remove fields that no reader or resumption handler uses.Implement Forward Actions with Compensation Pairs — For every step, implement both the forward command and its compensating action before wiring them together. Both must be idempotent: sending the same forward command twice should not create duplicate orders; sending the same compensation command twice should not issue double refunds. Store each pair as a
(forward_action, compensate_action)tuple in a step registry that the saga executor iterates through sequentially. Checkpoint: After implementation, verify that callingcompensate()afterexecute()restores the participant context to its pre-step state (within eventual consistency bounds).Wire Context Communication — Set up the messaging layer between contexts. For orchestration: configure a command bus that sends commands from the saga coordinator to each context's command handler. For choreography: configure an event bus where each context publishes events that other contexts listen for. In both cases, each context exposes only the specific commands or events needed by the saga — not its entire API surface. Checkpoint: The communication channel uses correlation IDs that link all messages within a single saga instance. Verify this by tracing a message's correlation ID through the entire step sequence.
Add Timeout and Recovery Mechanisms — Implement timeout detection for each step with configurable duration (e.g., 30 seconds for internal services, 5 minutes for external APIs). Store pending sagas in the durable store with
last_heartbeat_atandtimeout_secondsmetadata. On system restart or new deployment, scan the saga store for sagas that are neither COMPLETED nor FAILED nor COMPENSATING — resume those that have not timed out, compensate those that have. Checkpoint: The recovery handler processes pending sagas in deterministic order (e.g., oldest-first) and logs every resume/compensate decision with the correlation ID for audit purposes.
Implementation Patterns / Reference Guide
Pattern 1: Orchestration-Based Saga Coordinator
An orchestration-based saga has a single coordinator class that owns the entire workflow lifecycle. The coordinator sends commands to each context in sequence, tracks which steps succeeded, and runs compensating actions in exact reverse order when any step fails. This pattern is preferred for complex multi-step sagas (4+ steps) where explicit error handling and debugging visibility matter more than loose coupling.
The key insight: the coordinator does NOT hold participant state — it only tracks progress through the saga. Each participant context owns its own data, and the coordinator communicates via commands. The coordinator maintains a stack of compensation functions that get popped and executed in LIFO order when a failure occurs.
from __future__ import annotations
import logging
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum, auto
from typing import Awaitable, Callable, Protocol, Sequence
logger = logging.getLogger(__name__)
# ── Domain Types ──────────────────────────────────────────────────────────────
class SagaStatus(Enum):
"""Lifecycle status of a saga instance."""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
COMPENSATING = "compensating"
FAILED = "failed"
TIMEOUT = "timeout"
@dataclass(frozen=True)
class SagaResult:
"""Immutable result of a saga execution."""
saga_instance_id: str
status: SagaStatus
completed_at: datetime | None = None
failure_reason: str | None = None
@classmethod
def completed(cls, saga_id: str) -> SagaResult:
return cls(
saga_instance_id=saga_id,
status=SagaStatus.COMPLETED,
completed_at=datetime.now(timezone.utc),
)
@classmethod
def failed(cls, saga_id: str, reason: str) -> SagaResult:
return cls(
saga_instance_id=saga_id,
status=SagaStatus.FAILED,
failure_reason=reason,
)
@dataclass(frozen=True)
class SagaStep[A, B]:
"""A single step in a saga: forward action and its compensating counterpart.
Args:
name: Human-readable step name for logging (e.g., "Reserve Inventory")
forward: Async callable that performs the primary business action
compensate: Async callable that reverses the forward action's effects
"""
name: str
forward: Callable[..., Awaitable[B]]
compensate: Callable[..., Awaitable[None]]
# ── Coordinator Protocol ──────────────────────────────────────────────────────
class SagaCoordinatorProtocol(Protocol):
"""Interface for a saga coordinator that manages cross-context transactions."""
async def execute(self, saga_instance_id: str, *args: object, **kwargs: object) -> SagaResult: ...
async def compensate(self, saga_instance_id: str, completed_steps: list[SagaStep]) -> None: ...
# ── Core Saga Orchestrator ────────────────────────────────────────────────────
class SagaOrchestrator:
"""Orchestrates a sequence of forward/compensate steps across bounded contexts.
The orchestrator is the single point of control for cross-domain transactions.
It executes steps sequentially, tracking each successful step in a compensation stack.
When any step raises an exception, it pops compensations off the stack and executes
them in LIFO (last-in-first-out) order — the exact reverse of how forward actions were applied.
Each compensating action must be idempotent: calling it twice produces the same
end state as calling it once. This handles retries from message brokers and
recovery after system restarts during compensation.
"""
def __init__(self, saga_instance_id: str = "") -> None:
self._saga_id = saga_instance_id or str(uuid.uuid4())
self._correlation_id: str = str(uuid.uuid4()) # Links all commands/events in this saga
self._status: SagaStatus = SagaStatus.PENDING
self._step_history: list[dict] = []
@property
def correlation_id(self) -> str:
"""Returns the correlation ID that links all messages within this saga.
Every command sent to a participant context and every event published by
that context must include this correlation_id so operators can trace the
entire business transaction in distributed tracing tools.
"""
return self._correlation_id
@property
def status(self) -> SagaStatus:
return self._status
async def execute(
self,
steps: Sequence[SagaStep],
saga_context: dict[str, object],
) -> SagaResult:
"""Execute all saga steps sequentially with compensation on any failure.
Args:
steps: Ordered list of forward/compensate step pairs defining the workflow.
saga_context: Shared context dict passed to each step (order_id, customer_id, etc.).
Returns:
SagaResult indicating COMPLETED or FAILED with reason.
Algorithm:
1. Iterate through steps in order
2. Execute forward action — if it succeeds, push compensation onto stack
3. If any forward action raises, pop compensations off the stack and execute them LIFO
4. Update status atomically after all compensations complete
"""
self._status = SagaStatus.RUNNING
compensation_stack: list[SagaStep] = []
try:
for step in steps:
logger.info(
"[saga:%s] Executing step %s (correlation_id=%s)",
self._saga_id,
step.name,
self._correlation_id,
)
result = await step.forward(**saga_context)
compensation_stack.append(step)
self._step_history.append({
"step": step.name,
"action": "forward",
"success": True,
"timestamp": datetime.now(timezone.utc).isoformat(),
})
# All steps succeeded — mark completed
self._status = SagaStatus.COMPLETED
logger.info(
"[saga:%s] All %d steps completed successfully (correlation_id=%s)",
self._saga_id,
len(steps),
self._correlation_id,
)
return SagaResult.completed(self._saga_id)
except Exception as exc:
# Forward execution failed — compensate in reverse order
self._status = SagaStatus.COMPENSATING
logger.warning(
"[saga:%s] Step %s failed with %s. Starting compensation (%d pending).",
self._saga_id,
step.name if "step" in dir() else "unknown",
exc,
len(compensation_stack),
)
await self._run_compensations(compensation_stack, saga_context)
return SagaResult.failed(self._saga_id, str(exc))
async def _run_compensations(
self,
compensation_stack: list[SagaStep],
saga_context: dict[str, object],
) -> None:
"""Execute compensating actions in reverse (LIFO) order.
This is the critical safety mechanism of saga orchestration. Every forward
action that changed state must be reversed, and it must be reversed in the
exact opposite order to prevent inconsistent intermediate states.
For example: if step 1 reserved inventory and step 2 charged payment,
compensation must first refund payment (step 2's inverse) then release
inventory (step 1's inverse). Refunding after releasing inventory would
leave the customer charged but the items still committed to another order.
Each compensation is wrapped in try/except so one compensation failure does
not abort the entire compensation sequence — we log and continue.
"""
for step in reversed(compensation_stack):
try:
logger.warning(
"[saga:%s] Compensating step %s (correlation_id=%s)",
self._saga_id,
step.name,
self._correlation_id,
)
await step.compensate(**saga_context)
self._step_history.append({
"step": step.name,
"action": "compensate",
"success": True,
"timestamp": datetime.now(timezone.utc).isoformat(),
})
except Exception as comp_exc:
# Compensation itself failed — this is the hardest problem in saga design.
# The state may be inconsistent. This MUST trigger manual review.
logger.error(
"[saga:%s] COMPENSATION FAILED for step %s: %s — MANUAL REVIEW REQUIRED",
self._saga_id,
step.name,
comp_exc,
exc_info=True,
)
self._step_history.append({
"step": step.name,
"action": "compensate",
"success": False,
"error": str(comp_exc),
"timestamp": datetime.now(timezone.utc).isoformat(),
})
# After all compensations attempt, mark saga as failed
self._status = SagaStatus.FAILED
def get_step_history(self) -> list[dict]:
"""Returns the complete step execution history for debugging and audit."""
return list(self._step_history)
# ❌ BAD: Manual orchestration with no compensation tracking — failures are silently ignored
class BadSagaManual:
"""No coordination, no compensation, no recovery. This is what happens when you
write cross-context transactions by hand without a saga pattern."""
def __init__(self) -> None:
self._order_id: str = ""
async def place_order(self, order_data: dict) -> bool:
"""Multiple contexts called directly with no coordination or compensation."""
try:
# Step 1: Reserve inventory
await self._reserve_inventory(order_data["items"])
except Exception as e:
logger.error("Inventory reserve failed: %s", e)
return False # ❌ Caller gets False but inventory may still be partially reserved
try:
# Step 2: Charge payment (after inventory is already committed!)
await self._charge_payment(order_data["total"])
except Exception as e:
logger.error("Payment failed: %s", e)
return False # ❌ Inventory was reserved but nobody compensates — leaked reservation
try:
await self._confirm_order(order_data["order_id"])
except Exception as e:
logger.error("Order confirmation failed: %s", e)
return False # ❌ Two failures now, zero compensation ever triggered
return True # No one knows this partially failed saga is sitting in a zombie state
# ✅ GOOD: Structured orchestration with full compensation chain and logging
async def demonstrate_order_fulfillment_saga() -> tuple[SagaOrchestrator, list[SagaStep]]:
"""Demonstrate a properly orchestrated order fulfillment saga.
This is the canonical example: an order must pass through Inventory (reserve),
Payment (charge), and Shipping (schedule) contexts. If any step fails, all
completed steps are compensated in reverse order.
"""
orchestrator = SagaOrchestrator(saga_instance_id="SAGA-ORDER-001")
# Each context exposes a command handler as an async callable
inventory_client: InventoryClient = InventoryClient() # type: ignore[name-defined]
payment_client: PaymentClient = PaymentClient() # type: ignore[name-defined]
shipping_client: ShippingClient = ShippingClient() # type: ignore[name-defined]
steps: list[SagaStep] = [
SagaStep(
name="Reserve Inventory",
forward=inventory_client.reserve_items,
compensate=inventory_client.release_items,
),
SagaStep(
name="Charge Payment",
forward=payment_client.charge_card,
compensate=payment_client.refund_charge,
),
SagaStep(
name="Confirm Order",
forward=lambda **ctx: _confirm_order_in_order_context(ctx["order_id"]),
compensate=lambda **ctx: _cancel_order_in_order_context(ctx["order_id"]),
),
SagaStep(
name="Schedule Shipping",
forward=shipping_client.schedule_delivery,
compensate=shipping_client.cancel_shipment,
),
]
saga_context = {
"order_id": "ORD-2026-0042",
"customer_id": "CUST-789",
"items": [{"sku": "WIDGET-A", "quantity": 3}],
"total_amount_cents": 2997,
}
result = await orchestrator.execute(steps, saga_context)
# Production: persist the SagaResult to the saga state store
if result.status == SagaStatus.COMPLETED:
print(f"✅ Order fulfillment completed (correlation_id={orchestrator.correlation_id})")
elif result.status == SagaStatus.FAILED:
print(f"❌ Order fulfillment failed: {result.failure_reason}")
print(f" Step history: {len(orchestrator.get_step_history())} entries recorded for review")
return orchestrator, steps
async def _confirm_order_in_order_context(order_id: str) -> dict:
return {"order_id": order_id, "status": "CONFIRMED"}
async def _cancel_order_in_order_context(order_id: str) -> None:
logger.info("Compensating: cancelling order %s", order_id)
Pattern 2: Choreography-Based Event-Driven Sagas
In a choreographed saga, there is no central coordinator. Each context listens for events from other contexts and reacts by executing its own step, then publishing the next event in the chain. The event sequence defines the forward flow; a separate set of compensation events defines the reverse flow. This pattern produces tighter coupling between contexts than orchestration — every context must know about the specific events that trigger its step — but it has better loose-coupling properties within each context since no coordinator exists as a single point of failure.
The key difference from orchestration: in choreography, there is NO explicit compensation stack managed by a central object. Instead, when any context detects a failure (e.g., "payment charge declined"), it publishes a CompensationRequested event that triggers the reverse chain. Each context listens for its own compensation event and runs its compensating action locally.
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum, auto
from typing import Any, Callable
logger = logging.getLogger(__name__)
# ── Event Types for Choreographed Saga ────────────────────────────────────────
@dataclass(frozen=True)
class SagaEvent:
"""Base class for all saga coordination events.
Every event carries a correlation_id that links it to a specific saga instance,
allowing operators to trace the entire event chain in distributed tracing tools.
"""
correlation_id: str
occurred_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@property
def event_name(self) -> str:
return self.__class__.__name__
# Forward events (trigger next step in the saga flow)
@dataclass(frozen=True)
class OrderCreated(SagaEvent):
order_id: str
customer_id: str
items: list[dict[str, Any]]
total_cents: int
@dataclass(frozen=True)
class StockReserved(SagaEvent):
order_id: str
reservation_id: str
item_sku: str
quantity_reserved: int
@dataclass(frozen=True)
class PaymentCharged(SagaEvent):
order_id: str
payment_id: str
amount_cents: int
@dataclass(frozen=True)
class ShipmentScheduled(SagaEvent):
order_id: str
shipment_id: str
estimated_delivery_date: str
# Compensation events (trigger reverse flow when something fails)
@dataclass(frozen=True)
class PaymentFailed(SagaEvent):
order_id: str
failure_reason: str
@dataclass(frozen=True)
class StockReservationCancelled(SagaEvent):
order_id: str
reservation_id: str | None = None
# ── Event Router (Choreography Coordinator — Decentralized) ───────────────────
class SagaEventRouter:
"""Routes saga events between bounded contexts in a choreographed architecture.
Unlike an orchestrator that actively sends commands, this router is passive: it
receives events and delivers them to registered handlers. Each handler belongs to
a specific bounded context and reacts by executing its step (forward or compensation).
This is the closest thing to a coordinator in choreography — but it does not own
the saga state, does not track progress, and cannot make routing decisions beyond
delivering events to their subscribers. The business logic lives entirely within
each context's event handlers.
"""
def __init__(self) -> None:
# event_type -> list of handler callables
self._subscribers: dict[str, list[Callable[[SagaEvent], Any]]] = {}
def subscribe(self, event_type: type[SagaEvent], handler: Callable[[SagaEvent], Any]) -> None:
"""Register a handler for a specific saga event type.
In choreography, each context registers handlers for the events it listens to.
For example, the InventoryContext subscribes to OrderCreated so it can reserve stock.
"""
key = event_type.__name__
if key not in self._subscribers:
self._subscribers[key] = []
self._subscribers[key].append(handler)
async def publish(self, event: SagaEvent) -> None:
"""Deliver an event to all registered handlers for its type.
Handlers are called sequentially. If a handler raises, the remaining handlers
still execute (fail-fast is NOT automatic — each handler must handle errors).
This enables the event chain: OrderCreated → StockReserved → PaymentCharged → ShipmentScheduled
"""
key = event.event_name
handlers = self._subscribers.get(key, [])
for handler in handlers:
try:
await handler(event)
except Exception as exc:
logger.error(
"Handler failed for event %s (correlation_id=%s): %s",
event.event_name,
getattr(event, "correlation_id", ""),
exc,
exc_info=True,
)
# In production: route to dead letter queue after max retries
# ── Context Event Handlers (Each lives in its own bounded context) ────────────
class InventoryEventHandler:
"""Inventory context handler — listens for OrderCreated and PaymentFailed events.
When an order is created, this handler reserves stock and publishes StockReserved.
When payment fails, it releases the previously reserved stock and publishes
StockReservationCancelled (triggering the next compensation step).
"""
def __init__(self, router: SagaEventRouter, inventory_service: Any = None) -> None: # type: ignore[name-defined]
self._router = router
self._inventory_service = inventory_service or InventoryService() # type: ignore[name-defined]
# Register this context's handlers
router.subscribe(OrderCreated, self._handle_order_created)
router.subscribe(PaymentFailed, self._handle_payment_failed)
async def _handle_order_created(self, event: OrderCreated) -> None:
"""Reserve inventory when an order is created. Publishes StockReserved on success."""
logger.info(
"[INVENTORY] Reserving stock for order %s (correlation_id=%s)",
event.order_id,
event.correlation_id,
)
try:
reservation = await self._inventory_service.reserve(
items=event.items,
correlation_id=event.correlation_id,
)
# Publish next event in the chain — PaymentContext will listen for this
await self._router.publish(StockReserved(
correlation_id=event.correlation_id,
order_id=event.order_id,
reservation_id=reservation.reservation_id, # type: ignore[attr-defined]
item_sku=event.items[0].get("sku", "unknown") if event.items else "unknown", # type: ignore[index]
quantity_reserved=sum(item.get("quantity", 1) for item in event.items), # type: ignore[misc]
))
except Exception as exc:
# Inventory cannot fulfill — trigger compensation chain
await self._router.publish(PaymentFailed(
correlation_id=event.correlation_id,
order_id=event.order_id,
failure_reason=f"Inventory unavailable: {exc}",
))
async def _handle_payment_failed(self, event: PaymentFailed) -> None:
"""Release reserved stock when payment fails. Publishes StockReservationCancelled."""
logger.warning(
"[INVENTORY] Releasing stock — payment failed for order %s",
event.order_id,
)
await self._inventory_service.release(event.correlation_id)
await self._router.publish(StockReservationCancelled(
correlation_id=event.correlation_id,
order_id=event.order_id,
))
class PaymentEventHandler:
"""Payment context handler — listens for StockReserved and publishes PaymentCharged or triggers failure."""
def __init__(self, router: SagaEventRouter, payment_service: Any = None) -> None: # type: ignore[name-defined]
self._router = router
self._payment_service = payment_service or PaymentService() # type: ignore[name-defined]
router.subscribe(StockReserved, self._handle_stock_reserved)
async def _handle_stock_reserved(self, event: StockReserved) -> None:
"""Charge payment when inventory is reserved. Publishes ShipmentScheduled or triggers failure."""
logger.info(
"[PAYMENT] Charging payment for order %s (correlation_id=%s)",
event.order_id,
event.correlation_id,
)
try:
charge = await self._payment_service.charge(event.order_id) # type: ignore[attr-defined]
await self._router.publish(PaymentCharged(
correlation_id=event.correlation_id,
order_id=event.order_id,
payment_id=charge.payment_id, # type: ignore[attr-defined]
amount_cents=0, # In real code, would be charge.amount_cents
))
except Exception as exc:
await self._router.publish(PaymentFailed(
correlation_id=event.correlation_id,
order_id=event.order_id,
failure_reason=f"Payment failed: {exc}",
))
# ❌ BAD: No event-based choreography — contexts call each other directly
class BadChoreography:
"""Direct synchronous calls between contexts — tight coupling, no saga pattern."""
def __init__(self) -> None:
self.inventory = InventoryClient() # Tight coupling to another context's service
self.payment = PaymentClient() # Direct dependency creates a deployment chain
async def process_order(self, order_data: dict) -> bool:
"""Synchronous cascade with no event trail and no compensation.
If this fails halfway through, there is no record of what happened
and no automated way to undo partial work.
"""
reservation = await self.inventory.reserve(order_data["items"])
# ❌ No event published — other contexts don't know what's happening
# ❌ If payment fails below, nobody compensates the inventory reservation
charge = await self.payment.charge(reservation.order_id)
return True
# ✅ GOOD: Decentralized choreography — each context reacts to events independently
async def demonstrate_choreography() -> SagaEventRouter:
"""Set up a choreographed saga with event handlers in each bounded context.
The event chain flows like this:
1. OrderCreated (from Orders context) → triggers InventoryContext
2. StockReserved (from InventoryContext) → triggers PaymentContext
3. PaymentCharged (from PaymentContext) → triggers ShippingContext
4. ShipmentScheduled (from ShippingContext) → saga complete
Compensation chain (triggered by any failure):
1. PaymentFailed → triggers InventoryContext to release stock
2. StockReservationCancelled → no further action needed — done
"""
router = SagaEventRouter()
# Register each context's event handlers
inventory_events = InventoryEventHandler(router) # type: ignore[name-defined]
payment_events = PaymentEventHandler(router) # type: ignore[name-defined]
# shipping_events = ShippingEventHandler(router) # Would listen for PaymentCharged
# Publish the initial event from the Orders context
initial_event = OrderCreated(
correlation_id="corr-choreo-001",
order_id="ORD-2026-0043",
customer_id="CUST-790",
items=[{"sku": "WIDGET-B", "quantity": 2}],
total_cents=4998,
)
await router.publish(initial_event)
# The event chain now executes through all contexts automatically
return router
# Placeholder services for demonstration — in real code, these are context-owned
class InventoryService:
async def reserve(self, items: list, correlation_id: str = "") -> Any: # type: ignore[name-defined]
raise NotImplementedError
async def release(self, correlation_id: str) -> None:
logger.info("[INVENTORY] Released stock (correlation_id=%s)", correlation_id)
class PaymentService:
async def charge(self, order_id: str) -> Any: # type: ignore[name-defined]
raise NotImplementedError
class InventoryClient:
async def reserve(self, items: list[dict]) -> Any: # type: ignore[name-defined]
return type("Reservation", (), {"reservation_id": "RES-001"})()
async def release(self, correlation_id: str) -> None:
logger.info("[INVENTORY] Released items (correlation_id=%s)", correlation_id)
class PaymentClient:
async def charge_card(self, total_cents: int, **kwargs: Any) -> dict: # type: ignore[name-defined]
return {"charge_id": "CHG-001", "status": "charged"}
async def refund_charge(self, order_id: str, **kwargs: Any) -> None: # type: ignore[name-defined]
logger.info("[PAYMENT] Refunded charge for order %s", order_id)
class ShippingClient:
async def schedule_delivery(self, order_id: str, **kwargs: Any) -> dict: # type: ignore[name-defined]
return {"shipment_id": "SHIP-001"}
async def cancel_shipment(self, order_id: str, **kwargs: Any) -> None: # type: ignore[name-defined]
logger.info("[SHIPPING] Cancelled shipment for order %s", order_id)
Pattern 3: Timeout and Cancellation Handling
Long-running sagas face a critical problem: what happens when one context becomes unresponsive? The coordinator must detect timeouts and take action — typically compensating all previously completed steps. This pattern shows how to implement per-step timeout detection with configurable durations, heartbeat tracking, and automated cleanup of timed-out sagas.
The timeout mechanism has two parts: (1) step-level timeout on individual commands (how long to wait for a response from a context), and (2) saga-level timeout (total elapsed time for the entire saga). Step timeouts trigger immediate compensation; saga-level timeouts trigger full cancellation when the system detects that no progress is being made.
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import timedelta
from enum import Enum, auto
from typing import Any
logger = logging.getLogger(__name__)
class TimeoutAction(Enum):
"""What to do when a saga step exceeds its timeout."""
COMPENSATE_AND_STOP = "compensate_and_stop" # Default — roll back completed work
RETRY_WITH_BACKOFF = "retry_with_backoff" # Retry the failed step with exponential backoff
NOTIFY_AND_HOLD = "notify_and_hold" # Alert operators but keep waiting (manual intervention)
@dataclass(frozen=True)
class StepTimeoutConfig:
"""Configuration for timeout detection on a single saga step.
Args:
duration: Maximum time to wait for the step's forward action to complete
max_retries: Number of retries before treating as timeout (for RETRY_WITH_BACKOFF)
backoff_base_seconds: Base delay between retries (doubles each retry)
action: What to do when the timeout is exceeded
"""
duration: timedelta = field(default=timedelta(seconds=30))
max_retries: int = 1
backoff_base_seconds: float = 2.0
action: TimeoutAction = TimeoutAction.COMPENSATE_AND_STOP
@dataclass
class PendingSagaState:
"""Persisted state for a saga that is still running (needs recovery after restart).
This dataclass represents what must be stored in the saga state store so that
a system restart can resume or compensate the saga from where it left off.
"""
saga_instance_id: str
correlation_id: str
status: str # RUNNING, COMPENSATING, TIMEOUT
current_step_index: int
completed_steps: list[str] = field(default_factory=list)
timeout_config: dict[str, Any] = field(default_factory=dict)
created_at: str = "" # ISO 8601 timestamp string for JSON persistence
last_updated_at: str = "" # Updated after each step completion
last_heartbeat_at: str = "" # Updated on every step boundary
def has_timed_out(self, total_timeout_seconds: int) -> bool:
"""Check if the saga has exceeded its total timeout from creation.
Args:
total_timeout_seconds: Maximum allowed elapsed time for the entire saga.
Returns:
True if the saga should be cancelled and compensated due to age.
"""
import datetime
created = datetime.datetime.fromisoformat(self.created_at)
elapsed = datetime.datetime.now(datetime.timezone.utc) - created
return elapsed.total_seconds() > total_timeout_seconds
class TimeoutManager:
"""Manages timeout detection and response for saga steps.
In production, this integrates with the orchestration layer to wrap each step's
forward action with asyncio.wait_for() and triggers compensation when a timeout
fires. It also manages saga-level timeouts for detecting stuck or zombie sagas.
For long-running sagas that span minutes or hours (e.g., manual approval workflows),
use the NOTIFY_AND_HOLD action: pause execution, alert operators, and wait for
manual resumption rather than automatically compensating.
"""
def __init__(self, saga_state_store: Any = None) -> None: # type: ignore[name-defined]
self._saga_state_store = saga_state_store or InMemorySagaStateStore()
async def execute_with_timeout(
self,
step_name: str,
action_fn, # Callable that performs the forward action
timeout_config: StepTimeoutConfig,
correlation_id: str,
**context_kwargs: Any,
) -> Any:
"""Execute an action with timeout detection and retry logic.
Args:
step_name: Human-readable name for logging
action_fn: Async callable to execute (the forward action)
timeout_config: Timeout settings for this specific step
correlation_id: Links this action to its saga instance
**context_kwargs: Arguments passed to the action function
Returns:
The result of the successful forward action, or None on timeout/compensation.
"""
last_exception: Exception | None = None
for attempt in range(timeout_config.max_retries + 1):
try:
logger.info(
"[timeout] Step %s attempt %d/%d (correlation_id=%s)",
step_name,
attempt + 1,
timeout_config.max_retries + 1,
correlation_id,
)
# asyncio.wait_for enforces the deadline — raises TimeoutError if exceeded
result = await asyncio.wait_for(
action_fn(**context_kwargs),
timeout=timeout_config.duration.total_seconds(),
)
# Update heartbeat on successful step completion
self._saga_state_store.update_heartbeat(correlation_id)
…(truncated)