paulpas Domain Service Orchestration
Implements domain services that coordinate multi-aggregate business operations while preserving encapsulation boundaries. When loaded, the model acts as a DDD architect designing orchestrations where each aggregate root maintains its own invariants and the domain service coordinates the sequence without becoming a God object.
TL;DR Checklist
- Define domain services as ABCs that coordinate aggregates via repository interfaces — never directly call other domains
- Implement compensation actions for every irreversible operation (undo pattern)
- Use explicit transaction boundaries with commit/rollback in the orchestration flow
- Decide between orchestration (domain service) vs. choreography (events) based on coupling and visibility requirements
- Keep each domain service focused on a single business process — extract if it handles more than 3 aggregate types
When to Use
- A business operation spans multiple aggregate roots (e.g., transfer funds between accounts, place an order that creates Order + reserves Inventory + charges PaymentMethod)
- The operation has explicit compensation logic required on failure (undo a debit, release a reservation, refund a charge)
- You need deterministic, synchronous coordination of multiple domain models within one business transaction
- Auditing requirements demand logging the entire orchestration flow as a single trace
When NOT to Use
- The operation involves only one aggregate root — put the logic in the aggregate itself
- Cross-bounded-context communication that crosses organizational/team boundaries — use
cross-domain-workflow-sagasinstead - Long-running processes (hours/days) that require persistence between steps — these are sagas, not domain services
- Simple read operations or queries — use a CQRS read model / query handler instead
Core Workflow
Identify the Business Process Boundary — Determine which aggregate roots participate in this operation and what each one must do. Draw the flow: AccountA.debit() → PaymentService.charge() → Inventory.reserve() → Order.create(). Checkpoint: Each step should be independently completable or compensable. If any step cannot be undone, you need a saga pattern instead.
Define the Domain Service ABC — Create an abstract interface for the orchestration that declares the business operation as a single method (e.g.,
transfer_funds). The service depends on repository interfaces, not concrete implementations. Checkpoint: Verify the ABC has no imports from infrastructure packages — it must be domain-pure.Implement Compensation Actions — For every irreversible operation in the flow, define a compensating action that reverses it (e.g.,
credit_accountto undodebit_account). Store them in a list so they can be executed in reverse order on failure. Checkpoint: Each compensating action must have its own error handling — a failed compensation should be logged and escalated, not silently swallowed.Write the Concrete Service Implementation — Execute operations sequentially, collecting compensations. On success, commit all changes. On failure, execute compensations in reverse order, then raise. Checkpoint: Ensure no state is modified between repository calls if atomicity is required — use a Unit of Work (see
domain-repository-pattern).Apply Orchestration vs. Choreography Decision — Before finalizing, evaluate whether events-driven choreography would work better. Use orchestration when you need synchronous guarantees and deterministic ordering; use choreography for loose coupling across bounded contexts. Checkpoint: If the process crosses team-owned bounded contexts, default to choreography with sagas.
Implementation Patterns
Pattern 1: Domain Service Interface — ABC-Based Orchestration
Domain services are defined as abstract base classes that declare business operations spanning multiple aggregates. The service depends on repository interfaces and communicates with aggregates through their public domain methods only. No infrastructure concerns leak into the service interface.
from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
@dataclass(frozen=True)
class TransferRequest:
"""Immutable request object for a fund transfer operation."""
from_account_id: uuid.UUID
to_account_id: uuid.UUID
amount: float
currency: str = "USD"
reference: Optional[str] = None
@dataclass(frozen=True)
class TransferResult:
"""Result of a completed transfer with all side-effect details."""
success: bool
from_account_balance_after: float
to_account_balance_after: float
transaction_id: uuid.UUID
compensation_performed: bool = False
class DomainService(ABC):
"""Base class for domain services that coordinate cross-aggregate operations.
Each concrete implementation represents one business process that spans
multiple aggregate roots. The service handles orchestration logic,
compensation on failure, and transaction boundaries — while each
aggregate root remains responsible only for its own invariants.
"""
@abstractmethod
async def transfer_funds(self, request: TransferRequest) -> TransferResult:
"""Execute a fund transfer between two accounts.
This is the primary orchestration method. It coordinates debit
from the source account and credit to the destination account,
handling compensation if any step fails.
Args:
request: The immutable transfer parameters.
Returns:
TransferResult with final balances and transaction details.
Raises:
InsufficientFundsError: If source account lacks sufficient balance.
AccountNotFoundError: If either account does not exist.
ValueError: If amount is non-positive or accounts are identical.
"""
class CompensationStep:
"""Wraps an operation and its compensating action for undo-on-failure.
Usage in orchestration:
compensation_steps = [
CompensationStep(
action=lambda: account_a.debit(amount),
compensate=lambda: account_b.credit(amount),
)
]
On success, `action` executes and the step is discarded.
On failure, all previously executed steps' `compensate` methods run in reverse order.
"""
def __init__(
self,
action: Callable[[], None],
compensate: Callable[[], None],
description: str = "",
) -> None:
"""Initialize a compensable operation step.
Args:
action: The primary operation to execute (e.g., debit account).
compensate: The undo operation if the primary fails later in the flow.
description: Human-readable label for audit logging.
"""
self._action = action
self._compensate = compensate
self._description = description
self._executed = False
@property
def executed(self) -> bool:
"""Return True if this step's action has been executed."""
return self._executed
async def execute_action(self) -> None:
"""Execute the primary operation and mark as done.
Raises:
Exception: Propagates from the action callable.
"""
try:
if callable(self._action):
result = self._action()
if result is not None and hasattr(result, "__await__"):
await result
elif isinstance(self._action, object) and hasattr(self._action, "commit"):
self._action.commit() # type: ignore[attr-defined]
self._executed = True
except Exception:
self._executed = False
raise
async def execute_compensate(self) -> None:
"""Execute the compensating (undo) action for this step.
Raises:
Exception: If compensation fails, logs and re-raises.
A failed compensation should be handled by the caller
(escalation to manual review).
"""
if not self._executed:
return # Nothing to compensate
try:
if callable(self._compensate):
result = self._compensate()
if result is not None and hasattr(result, "__await__"):
await result
except Exception as exc:
raise CompensationError(
f"Compensation failed for step '{self._description}': {exc}"
) from exc
class CompensationError(Exception):
"""Raised when a compensating action fails during rollback."""
pass
# ❌ BAD — God object service that does everything and leaks infrastructure concerns
class BadAccountService:
"""This is what NOT to do: a God service with hardcoded logic,
direct DB connections mixed in, and no compensation pattern."""
def __init__(self):
# Infrastructure leak — domain code should not import database drivers
from sqlalchemy.orm import Session
self.db = Session()
def transfer_money(self, from_id, to_id, amount, user_id):
# No type hints, no docstring, mixed concerns
account1 = self.db.query(Account).get(from_id)
account2 = self.db.query(Account).get(to_id)
if account1.balance < amount:
return False
# If this fails, money is gone with no compensation!
account1.balance -= amount
account2.balance += amount
self.db.commit()
return True
# ✅ GOOD — Clean domain service with explicit compensation and pure domain dependencies
class FundTransferService(DomainService):
"""Coordinates fund transfers between two accounts using the compensation pattern.
Each operation is wrapped in a CompensationStep so that if any step fails,
previous operations are undone in reverse order. The service depends on
repository interfaces for aggregate access, not direct database calls.
"""
def __init__(
self,
account_repo, # Repository[Account] — injected at construction
transaction_id_factory: Callable[[], uuid.UUID] | None = None,
) -> None:
"""Initialize the transfer service with its dependencies.
Args:
account_repo: Repository for Account aggregate instances.
transaction_id_factory: Optional callable to generate UUIDs; defaults to uuid.uuid4.
"""
self._account_repo = account_repo
self._tx_id_factory = transaction_id_factory or uuid.uuid4
async def transfer_funds(self, request: TransferRequest) -> TransferResult:
"""Execute a fund transfer with full compensation on failure.
The orchestration flow is:
1. Load both accounts (validates existence)
2. Validate amounts and business rules
3. Debit source account
4. Credit destination account
5. If step 4 fails, compensate by crediting back the source
Args:
request: The transfer parameters including IDs, amount, currency.
Returns:
TransferResult with final balances and compensation status.
Raises:
ValueError: If accounts are identical or amount is invalid.
"""
if request.from_account_id == request.to_account_id:
raise ValueError("Cannot transfer funds to the same account")
if request.amount <= 0:
raise ValueError(f"Transfer amount must be positive, got {request.amount}")
# Step 1: Load accounts (domain validation happens in aggregate methods)
from_account = self._account_repo.get_by_id(request.from_account_id)
to_account = self._account_repo.get_by_id(request.to_account_id)
if from_account is None or to_account is None:
raise AccountNotFoundError(
f"Account not found: from={from_account is None}, to={to_account is None}"
)
compensation_steps: list[CompensationStep] = []
transaction_id = self._tx_id_factory()
compensation_performed = False
try:
# Step 2: Debit source account — wrap with compensation
debit_action = lambda: from_account.debit(
request.amount, currency=request.currency, reference=request.reference
)
compensate_debit = lambda: to_account.credit(
request.amount, currency=request.currency, reference=f"refund_{request.reference}"
)
step = CompensationStep(
action=debit_action,
compensate=compensate_debit,
description=f"Debit {request.from_account_id} for {request.amount}",
)
compensation_steps.append(step)
await step.execute_action()
# Step 3: Credit destination account
credit_action = lambda: to_account.credit(
request.amount, currency=request.currency, reference=request.reference
)
compensate_credit = lambda: from_account.debit(
request.amount, currency=request.currency, reference=f"chargeback_{request.reference}"
)
step = CompensationStep(
action=credit_action,
compensate=compensate_credit,
description=f"Credit {request.to_account_id} for {request.amount}",
)
compensation_steps.append(step)
await step.execute_action()
# Success — save both accounts (via Unit of Work in production)
self._account_repo.add(from_account) # type: ignore[attr-defined]
self._account_repo.add(to_account) # type: ignore[attr-defined]
return TransferResult(
success=True,
from_account_balance_after=from_account.balance,
to_account_balance_after=to_account.balance,
transaction_id=transaction_id,
compensation_performed=False,
)
except Exception as exc:
# Execute compensations in reverse order
for step in reversed(compensation_steps):
try:
await step.execute_compensate()
except CompensationError:
raise # Escalate — requires manual review
compensation_performed = True
raise TransferFailedError(
f"Transfer failed after compensations: {exc}"
) from exc
class AccountNotFoundError(ValueError):
"""Raised when a requested account does not exist."""
pass
class TransferFailedError(RuntimeError):
"""Raised when a transfer operation fails, potentially after compensation."""
pass
Pattern 2: Compensation Pattern — Saga-Style Rollback in Domain Services
The compensation pattern ensures that if any step in an orchestration flow fails, all previously executed steps are undone. Unlike traditional database transactions (which use ACID rollback), this pattern works even when different aggregates live in different databases or message queues.
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional
logger = logging.getLogger(__name__)
class OrchestrationOutcome(Enum):
"""Tracks the final state of an orchestration flow."""
COMPLETED_SUCCESSFULLY = "completed_successfully"
COMPLETED_WITH_COMPENSATION = "completed_with_compensation"
FAILED_PERMANENT = "failed_permanent"
@dataclass
class OrchestrationContext:
"""Holds transient state for a single orchestration execution.
This context is created at the start of each orchestration call and
discarded when the method returns. It does NOT persist to any database.
"""
outcome: OrchestrationOutcome = OrchestrationOutcome.COMPLETED_SUCCESSFULLY
compensation_errors: list[str] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
class CompensableStep(ABC):
"""Abstract base for steps that have both forward and compensating actions.
Subclasses implement execute() for the primary action and compensate()
for the undo action. The base class handles execution state tracking
and error propagation.
"""
@abstractmethod
async def execute(self) -> None:
"""Execute the primary business operation.
Raises:
Exception: If the operation fails, triggering compensation of prior steps.
"""
...
@abstractmethod
async def compensate(self) -> None:
"""Undo the effect of this step's execute().
This must be idempotent — calling it multiple times should have the
same effect as calling it once (safe to retry).
Raises:
CompensationError: If compensation cannot complete.
"""
...
class CompositeStep(CompensableStep):
"""A step that wraps multiple sub-steps executed sequentially.
If any sub-step fails, the composite's compensate() method undoes all
sub-steps that were successfully executed up to the failure point.
Usage:
order_step = CompositeStep("create_order", [
SaveOrderStep(order),
ReserveInventoryStep(inventory_items),
])
"""
def __init__(self, name: str, sub_steps: list[CompensableStep]) -> None:
"""Initialize a composite step with its sub-steps.
Args:
name: Human-readable name for logging and debugging.
sub_steps: List of CompensableStep instances to execute sequentially.
"""
self._name = name
self._sub_steps = sub_steps
self._executed_indices: list[int] = []
@property
def name(self) -> str:
return self._name
async def execute(self) -> None:
"""Execute sub-steps sequentially. On failure, re-raises with context.
Tracks which sub-steps completed successfully so that compensate()
can undo exactly those steps on failure.
"""
for i, step in enumerate(self._sub_steps):
try:
await step.execute()
self._executed_indices.append(i)
except Exception as exc:
raise OrchestrationStepError(
f"Step '{self._name}' sub-step {i} ({step.__class__.__name__}) failed: {exc}"
) from exc
async def compensate(self) -> None:
"""Undo all executed sub-steps in reverse order.
Each sub-step's compensate() is called from last-executed to first,
matching the reverse of execute()'s forward progression. Failed
compensations are logged but not raised — they require manual review.
"""
for idx in reversed(self._executed_indices):
step = self._sub_steps[idx]
try:
await step.compensate()
except Exception as exc:
logger.error(
"Compensation failed for %s sub-step %d (%s): %s",
self._name,
idx,
step.__class__.__name__,
exc,
)
# ❌ BAD — Manual compensation logic duplicated across every service method
def bad_transfer_with_manual_compensation(from_account, to_account, amount):
"""Each new orchestration repeats the same compensation pattern manually."""
try:
from_account.withdraw(amount)
to_account.deposit(amount)
except Exception:
# What if deposit succeeded but withdraw fails? Reverse order matters!
to_account.withdraw(amount) # Hope this doesn't fail too
from_account.deposit(amount) # Hope the original withdraw was caught
# ✅ GOOD — Centralized orchestration engine with automatic reverse compensation
class OrchestrationEngine:
"""Coordinates a sequence of compensable steps with automatic rollback.
Usage:
engine = OrchestrationEngine()
result = await engine.run([
StepA(...),
StepB(...),
StepC(...),
])
On any step failure, Compensate is called on Steps A and B in reverse
order (B first, then A). The result includes details about what was
compensated.
"""
def __init__(self) -> None:
"""Initialize the engine with an empty context tracker."""
self._context = OrchestrationContext()
async def run(self, steps: list[CompensableStep]) -> OrchestrationResult:
"""Execute all steps sequentially with compensation on failure.
Args:
steps: List of compensable operations to execute in order.
Returns:
OrchestrationResult detailing success/failure and any compensation performed.
"""
completed_steps: list[CompensableStep] = []
try:
for step in steps:
await step.execute()
completed_steps.append(step)
except Exception as exc:
self._context.outcome = OrchestrationOutcome.COMPLETED_WITH_COMPENSATION
# Compensate in reverse order of execution
for step in reversed(completed_steps):
try:
await step.compensate()
except Exception as comp_exc:
self._context.compensation_errors.append(
f"Compensation failed for {step.__class__.__name__}: {comp_exc}"
)
logger.error("Failed to compensate %s: %s", step.name, comp_exc)
self._context.outcome = OrchestrationOutcome.FAILED_PERMANENT
raise
return OrchestrationResult(
context=self._context,
steps_executed=len(completed_steps),
total_steps=len(steps),
)
@dataclass
class OrchestrationResult:
"""Summarizes the outcome of an orchestration execution."""
context: OrchestrationContext
steps_executed: int
total_steps: int
@property
def is_successful(self) -> bool:
return self.context.outcome == OrchestrationOutcome.COMPLETED_SUCCESSFULLY
@property
def compensation_errors_count(self) -> int:
return len(self.context.compensation_errors)
class OrchestrationStepError(RuntimeError):
"""Raised when a compensable step fails during execution."""
pass
Pattern 3: Orchestration vs. Choreography Decision Framework
Use this decision framework to determine whether a domain service (orchestration) or event-driven choreography is the right approach for coordinating business operations across aggregates.
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
@dataclass
class OrchestrationDecision:
"""Structured decision output from the orchestration/choreography evaluation.
Attributes:
recommended_pattern: Either "orchestration" or "choreography".
reasoning: Human-readable explanation of why this pattern was selected.
tradeoffs: List of considerations to evaluate before committing.
"""
recommended_pattern: str # "orchestration" or "choreography"
reasoning: str
tradeoffs: list[str]
def decide_orchestration_vs_choreography(
*,
spans_multiple_bounded_contexts: bool,
requires_synchronous_guarantee: bool,
involves_compensation_logic: bool,
business_owners_same_team: bool,
process_duration_seconds: float,
) -> OrchestrationDecision:
"""Evaluate whether to use domain service orchestration or event choreography.
This function implements a decision matrix that considers the key factors
driving the choice between centralized orchestration (domain service) and
decentralized choreography (event-driven).
Args:
spans_multiple_bounded_contexts: True if aggregates belong to different
bounded contexts (different teams/domains).
requires_synchronous_guarantee: True if the operation must complete
atomically and return immediately.
involves_compensation_logic: True if failed steps need undo operations.
business_owners_same_team: True if all participating aggregates share
the same team owner.
process_duration_seconds: Expected maximum duration of the entire process.
Returns:
OrchestrationDecision with recommended pattern and reasoning.
Decision matrix:
- Same team + short-lived + needs sync guarantee → ORCHESTRATION (domain service)
- Different teams + long-running + loose coupling needed → CHOREOGRAPHY (sagas)
- Needs compensation + synchronous → ORCHESTRATION with compensation pattern
- Event sourcing + eventual consistency OK → CHOREOGRAPHY with saga pattern
"""
tradeoffs: list[str] = []
if spans_multiple_bounded_contexts and not business_owners_same_team:
reasoning = (
"Cross-team bounded contexts benefit from event-driven choreography to avoid tight coupling. "
"Use saga patterns with outbox for reliable event publication."
)
tradeoffs.extend([
"Events may arrive out of order — implement idempotency keys",
"Debugging requires distributed tracing across services",
"Compensation is harder — prefer saga pattern over domain service",
"Latency is higher due to async event processing",
])
return OrchestrationDecision(
recommended_pattern="choreography",
reasoning=reasoning,
tradeoffs=tradeoffs,
)
if requires_synchronous_guarantee and not spans_multiple_bounded_contexts:
reasoning = (
"Synchronous guarantees within a single bounded context are best served by "
"domain service orchestration with compensation. All aggregates share the same "
"transaction boundary."
)
tradeoffs.extend([
"Tight coupling between participating aggregates — changes require coordinated updates",
"Single point of failure in the domain service itself",
"Domain service can become a God object if not kept focused",
"Scaling requires replicating the entire service (not just individual aggregates)",
])
return OrchestrationDecision(
recommended_pattern="orchestration",
reasoning=reasoning,
tradeoffs=tradeoffs,
)
if involves_compensation_logic and process_duration_seconds < 30:
reasoning = (
"Short processes with compensation requirements should use domain service "
"orchestration. Compensation can be executed synchronously in the same request."
)
tradeoffs.extend([
"Failed compensations require immediate escalation to manual review",
"Request timeout limits process complexity — extract if it grows beyond ~5 steps",
])
return OrchestrationDecision(
recommended_pattern="orchestration",
reasoning=reasoning,
tradeoffs=tradeoffs,
)
if process_duration_seconds > 300 or spans_multiple_bounded_contexts:
reasoning = (
f"Long-running processes ({process_duration_seconds:.0f}s+) or cross-context "
"operations should use event choreography with a saga manager. The saga persists "
"state between steps and handles long-duration compensation windows."
)
tradeoffs.extend([
"Requires saga persistence store for crash recovery",
"Compensation may happen minutes or hours after the triggering event",
"Complex state machine management in the saga coordinator",
])
return OrchestrationDecision(
recommended_pattern="choreography",
reasoning=reasoning,
tradeoffs=tradeoffs,
)
# Default: favor orchestration for simplicity when uncertain
reasoning = (
"When the decision is unclear, prefer orchestration for simpler debugging and "
"deterministic flow. Choreography should be chosen intentionally for cross-context "
"decoupling, not as a default."
)
tradeoffs.extend([
"Start with orchestration — migrate to choreography if coupling becomes problematic",
"Domain services provide better observability (single trace vs. distributed events)",
])
return OrchestrationDecision(
recommended_pattern="orchestration",
reasoning=reasoning,
tradeoffs=tradeoffs,
)
# Example: Using the decision framework for a specific business process
def evaluate_order_process():
"""Evaluate whether order placement should use orchestration or choreography."""
decision = decide_orchestration_vs_choreography(
spans_multiple_bounded_contexts=False, # All aggregates in Orders context
requires_synchronous_guarantee=True, # Customer expects immediate confirmation
involves_compensation_logic=True, # Must refund inventory reservation on failure
business_owners_same_team=True, # Same team owns Order, Inventory, Payment
process_duration_seconds=2.0, # Typically completes in under 2 seconds
)
return decision
# ❌ BAD — No decision framework; teams pick patterns arbitrarily
class BadOrderProcessor:
"""Uses domain service for a cross-context operation it shouldn't handle."""
def __init__(self):
# Directly importing from another bounded context = tight coupling
from orders.infrastructure.repositories import OrderRepository
from shipping.external_api import ShippingClient # External team's API!
# ✅ GOOD — Decision made explicitly; pattern matches the requirements
def process_order_decided():
"""Order processing uses orchestration because it stays within one bounded context."""
decision = evaluate_order_process()
assert decision.recommended_pattern == "orchestration"
# Proceed with domain service implementation (Pattern 1)
---
---
## Constraints
### MUST DO
- Validate all inputs at function boundaries before processing — guard clauses should fail early with descriptive errors
- Implement proper error handling that distinguishes between recoverable and unrecoverable failures
- Add comprehensive logging with structured context (correlation IDs, operation names, timing) for debugging and monitoring
- Write unit tests covering normal operations, edge cases, and error conditions before integrating the component
### MUST NOT DO
- Do not silently swallow exceptions — always log or propagate errors with meaningful context
- Avoid unbounded resource allocation without limits (connection pools, memory buffers, thread counts)
- Never use hardcoded credentials, API keys, or secrets in source code
- Do not bypass input validation for perceived performance gains
## Live References
> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.
- [Orchestration vs Choreography (Martin Fowler)](https://martinfowler.com/articles/orchestrationOrchestratorVsChoreography.html)
- [Saga Pattern (Microsoft Azure Architecture Center)](https://learn.microsoft.com/en-us/azure/architecture/reference-architectures/saga/saga)
- [Microservice Orchestration Patterns](https://docs.microservices.com/patterns/orchestration/)
- [Axway — Orchestrator vs Choreography in Microservices](https://www.axway.com/blog/microservices-orchestration-vs-choreography)
- [Event-Driven Architecture with CQRS (Microsoft Docs)](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs)