Saga Pattern: Distributed Transaction Orchestration and Compensation

“

Series: S10 Design Patterns · Article 4 | Level: Advanced | Reading Time: 18 min

#TL;DR

The Saga pattern decomposes a cross-service long-running transaction into a series of local transactions, each with a corresponding compensating action. If any step fails, the system executes compensations in reverse order to undo completed steps.
In coomia-dip, the Saga pattern ensures eventual consistency when Ontology Actions execute across Layers — from Action registration in the Control Layer, to data writes in the Data Layer, to reasoning triggers in the Intelligence Layer. Any step failure triggers safe rollback.
Combined with the Temporal workflow engine, coomia-dip implements durable, recoverable, observable Saga orchestration, supporting both Orchestration and Choreography modes.

#Introduction: The Distributed Transaction Dilemma

In microservice architectures, a single business operation may span multiple services. Traditional Two-Phase Commit (2PC) has severe limitations in distributed environments:

Code

Scenario: User creates a risk control rule in coomia-dip
1. Control Layer: Register ObjectType and Action definitions
2. Data Layer: Create corresponding Iceberg table structure
3. Intelligence Layer: Initialize rule template in the reasoning engine
4. Agent Runtime: Deploy automatic monitoring Agent

What happens if step 3 fails? Steps 1 and 2 have already committed — the database already has records. Traditional 2PC simply does not work across heterogeneous systems — you cannot expect Kafka, Iceberg, and a Python reasoning engine to all implement the XA protocol.

The Saga pattern provides an elegant solution: replace global transactions with a series of compensable local transactions.

#Part 1: Core Concepts of the Saga Pattern

#1.1 Local Transactions and Compensating Actions

The core idea of Saga is to split a global transaction into a series of steps, each containing a forward action and a compensating action:

Python

from dataclasses import dataclass, field
from enum import Enum
from typing import Any

class SagaStepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    FAILED = "failed"

@dataclass
class SagaStep:
    """A single step in a Saga."""
    name: str
    service: str                    # Target service/Layer
    action: str                     # Forward action
    compensation: str               # Compensating action
    input_data: dict[str, Any] = field(default_factory=dict)
    output_data: dict[str, Any] = field(default_factory=dict)
    status: SagaStepStatus = SagaStepStatus.PENDING
    retry_count: int = 0
    max_retries: int = 3
    timeout_seconds: int = 30

@dataclass
class SagaDefinition:
    """Definition of a complete Saga."""
    saga_id: str
    name: str
    steps: list[SagaStep]
    compensation_policy: str = "backward"   # backward | forward_recovery
    timeout_seconds: int = 300

#1.2 Saga Execution Orchestrator

coomia-dip uses Orchestration as the default Saga coordination strategy. A central orchestrator drives the entire Saga execution:

Python

class SagaOrchestrator:
    """Central Saga orchestrator for coomia-dip cross-Layer transactions."""

    def __init__(
        self,
        event_store: "EventStore",
        step_executors: dict[str, "StepExecutor"],
    ):
        self._event_store = event_store
        self._executors = step_executors

    async def execute(self, saga: SagaDefinition) -> SagaResult:
        """Execute a Saga with automatic compensation on failure."""
        completed_steps: list[SagaStep] = []

        for step in saga.steps:
            try:
                step.status = SagaStepStatus.RUNNING
                executor = self._executors[step.service]

                result = await self._execute_with_retry(
                    executor, step, saga.timeout_seconds
                )

                step.output_data = result
                step.status = SagaStepStatus.COMPLETED
                completed_steps.append(step)

                await self._event_store.append_saga_event(
                    saga.saga_id, step.name, "completed", result
                )

            except Exception as exc:
                step.status = SagaStepStatus.FAILED
                await self._event_store.append_saga_event(
                    saga.saga_id, step.name, "failed", {"error": str(exc)}
                )

                # Initiate compensation
                await self._compensate(saga, completed_steps)
                return SagaResult(
                    saga_id=saga.saga_id,
                    status="compensated",
                    failed_step=step.name,
                    error=str(exc),
                )

        return SagaResult(saga_id=saga.saga_id, status="completed")

    async def _compensate(
        self, saga: SagaDefinition, completed_steps: list[SagaStep]
    ) -> None:
        """Execute compensation in reverse order."""
        for step in reversed(completed_steps):
            try:
                step.status = SagaStepStatus.COMPENSATING
                executor = self._executors[step.service]

                await executor.compensate(step.compensation, step.output_data)

                step.status = SagaStepStatus.COMPENSATED
                await self._event_store.append_saga_event(
                    saga.saga_id, step.name, "compensated", {}
                )
            except Exception as comp_exc:
                step.status = SagaStepStatus.FAILED
                await self._event_store.append_saga_event(
                    saga.saga_id,
                    step.name,
                    "compensation_failed",
                    {"error": str(comp_exc)},
                )
                # Compensation failure requires manual intervention
                await self._alert_manual_intervention(saga, step, comp_exc)

    async def _execute_with_retry(
        self, executor: "StepExecutor", step: SagaStep, timeout: int
    ) -> dict[str, Any]:
        """Execute a step with retry logic."""
        import asyncio

        last_error: Exception | None = None
        for attempt in range(step.max_retries + 1):
            try:
                return await asyncio.wait_for(
                    executor.execute(step.action, step.input_data),
                    timeout=min(step.timeout_seconds, timeout),
                )
            except asyncio.TimeoutError:
                last_error = TimeoutError(
                    f"Step {step.name} timed out after {step.timeout_seconds}s"
                )
            except Exception as e:
                last_error = e
                if attempt < step.max_retries:
                    await asyncio.sleep(2 ** attempt)  # Exponential backoff
        raise last_error  # type: ignore[misc]

    async def _alert_manual_intervention(
        self, saga: SagaDefinition, step: SagaStep, error: Exception
    ) -> None:
        """Alert for manual intervention when compensation fails."""
        pass

#1.3 Orchestration vs Choreography

Saga has two coordination modes. coomia-dip supports both but defaults to Orchestration:

Dimension	Orchestration	Choreography
Coordination	Central orchestrator drives	Event-driven, services listen independently
Coupling	Orchestrator knows all steps	Services only know their own events
Observability	High — orchestrator tracks global state	Low — requires aggregating multi-source events
Complexity	Centralized flow management, easy to understand	Scattered across services, hard to trace
Use Cases	Cross-Layer transactions, complex flows	Simple notification chains, loose coupling
coomia-dip Usage	Action execution, rule deployment	Event notifications, monitoring triggers

#Part 2: Saga Practices in coomia-dip

#2.1 Cross-Layer Action Execution Saga

When a user invokes an Ontology Action through the SDK, the Action may need to coordinate across multiple Layers. Taking "Create Risk Rule" as an example:

Python

class CreateRiskRuleSaga:
    """Saga for creating a risk control rule across Layers."""

    def build(self, rule_config: dict) -> SagaDefinition:
        return SagaDefinition(
            saga_id=generate_id(),
            name="create_risk_rule",
            steps=[
                SagaStep(
                    name="register_object_type",
                    service="control-Layer",
                    action="register_risk_rule_type",
                    compensation="unregister_risk_rule_type",
                    input_data={"schema": rule_config["schema"]},
                ),
                SagaStep(
                    name="create_storage",
                    service="data-Layer",
                    action="create_iceberg_table",
                    compensation="drop_iceberg_table",
                    input_data={"table_spec": rule_config["storage"]},
                ),
                SagaStep(
                    name="init_reasoning_template",
                    service="intelligence-Layer",
                    action="deploy_rule_template",
                    compensation="undeploy_rule_template",
                    input_data={"template": rule_config["reasoning"]},
                ),
                SagaStep(
                    name="setup_monitoring",
                    service="agent-runtime",
                    action="create_monitor_agent",
                    compensation="destroy_monitor_agent",
                    input_data={"monitor": rule_config["monitoring"]},
                ),
            ],
            timeout_seconds=120,
        )

#2.2 Temporal-Based Saga Persistence

coomia-dip uses Temporal as the persistence and recovery engine for Sagas. Temporal provides workflow state persistence — even if a process crashes, execution resumes from the interruption point:

Python

from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def register_object_type(input_data: dict) -> dict:
    """Register ObjectType in Control Layer via gRPC."""
    async with grpc_channel("control-Layer:50051") as channel:
        stub = OntologyServiceStub(channel)
        response = await stub.RegisterObjectType(
            RegisterObjectTypeRequest(**input_data)
        )
        return {"type_id": response.type_id, "version": response.version}

@activity.defn
async def unregister_object_type(output_data: dict) -> None:
    """Compensate: unregister the ObjectType."""
    async with grpc_channel("control-Layer:50051") as channel:
        stub = OntologyServiceStub(channel)
        await stub.UnregisterObjectType(
            UnregisterObjectTypeRequest(type_id=output_data["type_id"])
        )

@activity.defn
async def create_iceberg_table(input_data: dict) -> dict:
    """Create Iceberg table in Data Layer."""
    async with grpc_channel("data-Layer:50052") as channel:
        stub = DataServiceStub(channel)
        response = await stub.CreateTable(
            CreateTableRequest(**input_data)
        )
        return {"table_id": response.table_id, "location": response.location}

@activity.defn
async def drop_iceberg_table(output_data: dict) -> None:
    """Compensate: drop the Iceberg table."""
    async with grpc_channel("data-Layer:50052") as channel:
        stub = DataServiceStub(channel)
        await stub.DropTable(
            DropTableRequest(table_id=output_data["table_id"])
        )

@workflow.defn
class CreateRiskRuleWorkflow:
    """Temporal workflow implementing the Create Risk Rule Saga."""

    @workflow.run
    async def run(self, rule_config: dict) -> dict:
        completed: list[dict] = []

        try:
            # Step 1: Register ObjectType
            type_result = await workflow.execute_activity(
                register_object_type,
                {"schema": rule_config["schema"]},
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            completed.append(("unregister_object_type", type_result))

            # Step 2: Create storage
            storage_result = await workflow.execute_activity(
                create_iceberg_table,
                {"table_spec": rule_config["storage"]},
                start_to_close_timeout=timedelta(seconds=30),
            )
            completed.append(("drop_iceberg_table", storage_result))

            # Step 3: Deploy reasoning template
            reasoning_result = await workflow.execute_activity(
                deploy_rule_template,
                {"template": rule_config["reasoning"]},
                start_to_close_timeout=timedelta(seconds=60),
            )
            completed.append(("undeploy_rule_template", reasoning_result))

            # Step 4: Setup monitoring
            monitor_result = await workflow.execute_activity(
                create_monitor_agent,
                {"monitor": rule_config["monitoring"]},
                start_to_close_timeout=timedelta(seconds=30),
            )

            return {
                "status": "completed",
                "type_id": type_result["type_id"],
                "table_id": storage_result["table_id"],
            }

        except Exception as e:
            # Compensate completed steps
            for comp_action, comp_data in reversed(completed):
                try:
                    await workflow.execute_activity(
                        comp_action,
                        comp_data,
                        start_to_close_timeout=timedelta(seconds=30),
                    )
                except Exception as comp_error:
                    workflow.logger.error(
                        f"Compensation failed: {comp_action}, "
                        f"error: {comp_error}"
                    )
            raise

#2.3 Saga State Visualization

The coomia-dip Platform Console provides real-time visualization of Saga execution. Operations staff can view each Saga's progress, failure points, and compensation status:

Python

@dataclass
class SagaVisualization:
    """Saga execution visualization data."""
    saga_id: str
    name: str
    status: str
    started_at: datetime
    completed_at: datetime | None
    steps: list[StepVisualization]
    total_duration_ms: int
    compensation_duration_ms: int | None

@dataclass
class StepVisualization:
    """Individual step visualization."""
    name: str
    service: str
    status: str
    started_at: datetime
    completed_at: datetime | None
    duration_ms: int
    retry_count: int
    error: str | None

class SagaDashboardService:
    """Service for Saga monitoring dashboard."""

    async def get_saga_history(
        self,
        tenant_id: str,
        time_range: tuple[datetime, datetime],
        status_filter: list[str] | None = None,
    ) -> list[SagaVisualization]:
        """Retrieve Saga execution history with visualization data."""
        pass

    async def get_compensation_stats(
        self, tenant_id: str
    ) -> dict[str, Any]:
        """Get compensation statistics for monitoring."""
        return {
            "total_sagas": 0,
            "completed": 0,
            "compensated": 0,
            "compensation_rate": 0.0,
            "avg_compensation_time_ms": 0,
            "manual_interventions": 0,
        }

#Part 3: Compensation Strategy Design

#3.1 Semantic Compensation vs Exact Rollback

In distributed systems, compensating actions are not always exact "undos." coomia-dip distinguishes two compensation strategies:

Exact Rollback: The inverse operation that fully restores prior state. For example, deleting a just-created Iceberg table.

Semantic Compensation: Not an exact inverse, but a business-level "cancellation." For example, if an order has shipped, the compensation is not "unship" (physically impossible) but "create a return order."

Python

class CompensationStrategies:
    """Different compensation strategies for various scenarios."""

    @staticmethod
    async def exact_rollback(step_name: str, output_data: dict) -> None:
        """Exact inverse operation — fully restores prior state."""
        pass

    @staticmethod
    async def semantic_compensation(step_name: str, output_data: dict) -> None:
        """Business-level compensation — not exact inverse."""
        pass

    @staticmethod
    async def idempotent_retry(step_name: str, input_data: dict) -> None:
        """Retry with idempotency key instead of compensating."""
        pass

#3.2 Idempotency of Compensating Actions

Compensating actions must be idempotent — executing them multiple times produces the same effect as executing once. This is necessary because in distributed environments, compensation operations themselves may fail and be retried:

Python

class IdempotentCompensation:
    """Ensure compensation operations are idempotent."""

    def __init__(self, state_store: "StateStore"):
        self._state_store = state_store

    async def compensate_with_idempotency(
        self,
        saga_id: str,
        step_name: str,
        compensation_fn: callable,
        data: dict,
    ) -> None:
        """Execute compensation with idempotency guarantee."""
        idempotency_key = f"{saga_id}:{step_name}:compensation"

        # Check if already executed
        if await self._state_store.is_completed(idempotency_key):
            return  # Already executed, skip

        try:
            await compensation_fn(data)
            await self._state_store.mark_completed(idempotency_key)
        except Exception:
            await self._state_store.mark_failed(idempotency_key)
            raise

#3.3 Timeout and Dead Letter Handling

When compensating actions repeatedly fail, coomia-dip moves the Saga instance to a Dead Letter Queue (DLQ) for manual intervention:

Python

class SagaDeadLetterHandler:
    """Handle Sagas that cannot be automatically compensated."""

    async def move_to_dead_letter(
        self, saga_id: str, failed_step: str, error: str
    ) -> None:
        """Move a failed Saga to the dead letter queue."""
        await self._dead_letter_store.save({
            "saga_id": saga_id,
            "failed_step": failed_step,
            "error": error,
            "timestamp": datetime.utcnow(),
            "requires_manual_intervention": True,
        })

        await self._notification_service.send_alert(
            severity="critical",
            title=f"Saga compensation failed: {saga_id}",
            message=(
                f"Step '{failed_step}' compensation failed after all retries. "
                f"Manual intervention required. Error: {error}"
            ),
        )

    async def retry_dead_letter(self, saga_id: str) -> SagaResult:
        """Manual retry of a dead-lettered Saga."""
        saga_state = await self._dead_letter_store.get(saga_id)
        return await self._orchestrator.resume_compensation(
            saga_id, saga_state["failed_step"]
        )

#Part 4: Saga Applications Across coomia-dip Scenarios

#4.1 Ontology Schema Change Saga

When an Ontology Schema changes (e.g., adding properties, modifying relationships), cross-Layer coordination is required:

Code

Step 1: Control Layer — Validate Schema compatibility
Step 2: Control Layer — Update ObjectType definition
Step 3: Data Layer — Execute Iceberg Schema Evolution
Step 4: Intelligence Layer — Update reasoning engine property mappings
Step 5: SDK Layer — Regenerate SDK type definitions

Compensation:
Step 5 fails → Roll back SDK code generation
Step 4 fails → Restore old reasoning engine mappings
Step 3 fails → Roll back Iceberg Schema
Step 2 fails → Restore old ObjectType definition

#4.2 Data Migration Saga

Large-scale data migration must be executed in phases, with each phase rollback-capable:

Python

class DataMigrationSaga:
    """Saga for large-scale data migration."""

    def build(self, migration_plan: dict) -> SagaDefinition:
        return SagaDefinition(
            saga_id=generate_id(),
            name="data_migration",
            steps=[
                SagaStep(
                    name="validate_source",
                    service="data-Layer",
                    action="validate_source_data",
                    compensation="noop",
                    timeout_seconds=60,
                ),
                SagaStep(
                    name="create_target_schema",
                    service="data-Layer",
                    action="create_migration_target",
                    compensation="drop_migration_target",
                    timeout_seconds=30,
                ),
                SagaStep(
                    name="copy_data",
                    service="data-Layer",
                    action="copy_data_batch",
                    compensation="delete_copied_data",
                    timeout_seconds=3600,
                ),
                SagaStep(
                    name="validate_target",
                    service="data-Layer",
                    action="validate_target_data",
                    compensation="noop",
                    timeout_seconds=120,
                ),
                SagaStep(
                    name="switch_references",
                    service="control-Layer",
                    action="update_ontology_references",
                    compensation="revert_ontology_references",
                    timeout_seconds=30,
                ),
            ],
            timeout_seconds=7200,
        )

#4.3 Multi-Tenant Resource Provisioning Saga

When onboarding a new tenant, resources must be allocated across multiple Layers:

Code

Step 1: Control Layer — Create tenant metadata
Step 2: Data Layer — Allocate storage namespace (Nessie Branch)
Step 3: Intelligence Layer — Initialize reasoning engine instance
Step 4: Agent Runtime — Deploy default Agents
Step 5: SDK Layer — Generate tenant-specific API keys

Any step failure → Clean up allocated resources in reverse order

#Part 5: Combining Saga with Other Patterns

#5.1 Saga + Event Sourcing

Every Saga step execution and compensation is recorded as an event in the Event Store, providing a complete audit trail:

Python

class SagaEventStore:
    """Record Saga events for audit trail."""

    async def append_saga_event(
        self, saga_id: str, step_name: str, action: str, data: dict
    ) -> None:
        event = DomainEvent(
            event_id=generate_id(),
            event_type=f"saga.{action}",
            aggregate_id=saga_id,
            aggregate_type="saga",
            sequence_number=await self._next_sequence(saga_id),
            timestamp=datetime.utcnow(),
            payload={
                "step_name": step_name,
                "action": action,
                "data": data,
            },
            metadata=EventMetadata(
                actor_id="system",
                actor_type="saga_orchestrator",
                tenant_id=self._current_tenant_id,
                world_id=self._current_world_id,
                source_plane="orchestration",
                trace_id=self._current_trace_id,
            ),
        )
        await self._event_store.append([event])

#5.2 Saga + State Machine

Each Saga instance is essentially a state machine. coomia-dip uses state machines (detailed in the next article) to manage Saga lifecycles:

Code

CREATED → RUNNING → COMPLETED
                  ↘ COMPENSATING → COMPENSATED
                                 ↘ FAILED (requires manual intervention)

#5.3 Saga + Idempotent Consumer

To handle duplicate message delivery, each Saga step implements idempotency. Combined with the Idempotent Consumer pattern, this ensures that even if messages are processed multiple times, system state remains consistent.

#Part 6: Production Considerations

#6.1 Performance Considerations

The Saga pattern incurs additional overhead compared to monolithic transactions:

Overhead Source	Impact	Mitigation Strategy
Multiple network calls	Increased latency	Parallelize independent steps
Event persistence	I/O overhead	Batch writes, async persistence
Compensation logic	Extra computation	Only triggered on failure
State checks	Query overhead	In-memory cache for active Saga states

#6.2 Concurrency Control

Multiple Saga instances may operate on the same resource simultaneously. coomia-dip uses distributed locks to ensure mutual exclusion for critical resources:

Python

class SagaResourceLock:
    """Distributed lock for Saga resource protection."""

    async def acquire(
        self, resource_id: str, saga_id: str, timeout: int = 30
    ) -> bool:
        """Acquire a distributed lock for a resource."""
        return await self._redis.set(
            f"saga:lock:{resource_id}",
            saga_id,
            nx=True,
            ex=timeout,
        )

    async def release(self, resource_id: str, saga_id: str) -> None:
        """Release a distributed lock."""
        current = await self._redis.get(f"saga:lock:{resource_id}")
        if current == saga_id:
            await self._redis.delete(f"saga:lock:{resource_id}")

#6.3 Monitoring and Alerting

Production environments must monitor key Saga metrics:

Compensation Rate: Compensation executions / total Saga executions. A high rate indicates downstream service instability.
Average Execution Time: Mean time from Saga start to completion.
Dead Letter Queue Depth: Number of accumulated failed Sagas.
Step Failure Distribution: Which steps fail most frequently.

#Part 7: Anti-Patterns and Caveats

#7.1 Do Not Assume Isolation in Sagas

Sagas do not provide Isolation. During Saga execution, intermediate states are visible to other transactions. Design considerations include:

Dirty Reads: Other transactions may read intermediate states during Saga execution
Lost Updates: Concurrent Sagas may overwrite each other's updates
Solutions: Semantic locks, commutative updates

#7.2 Avoid Overly Long Saga Chains

More steps mean higher failure probability and more complex compensation. Rules of thumb:

Sagas with more than 7 steps should be considered for splitting
Each step's timeout should be set appropriately
Nested Sagas (sub-Sagas) can manage complex flows

#7.3 Compensations Must Not Depend on External State Changes

Compensating actions should only depend on the output data of their corresponding forward action. They should not assume the external environment has remained unchanged — by the time compensation occurs, external state may have changed due to other operations.

#Key Takeaways

Saga Replaces 2PC: In distributed microservice architectures, the Saga pattern replaces traditional Two-Phase Commit with a series of compensable local transactions
Orchestration First: coomia-dip defaults to orchestrated Sagas, using the Temporal workflow engine for persistence and recoverability
Compensation Design Is Critical: Compensating actions must be idempotent, self-contained, and independent of external state assumptions
Observability: Every Saga step records events, providing a complete audit trail and real-time monitoring
Dead Letter Handling: Sagas that cannot be automatically compensated enter a dead letter queue for manual intervention
Control Chain Length: Avoid overly long Saga chains — consider splitting into sub-Sagas beyond 7 steps

#Next Article

In the next article, we will dive deep into the State Machine pattern — how coomia-dip uses finite state machines to manage Ontology object lifecycles, and how state machines work together with Saga orchestrators.

S10-05: State Machine: Object Lifecycle Management

#Tags

#DesignPatterns #Saga #DistributedTransactions #Compensation #Temporal #Orchestration #EventualConsistency #CrossPlaneTransactions #Idempotency