Saga Pattern: Compensation and Rollback for Long Transactions
When a decision requires multi-step operations spanning multiple microservices, traditional distributed transactions (2PC) cannot meet availability and performance requirements. coomia-dip adopts the Saga pattern with the Temporal workflow engine, decomposing long transactions into a series of compensable local transactions. Each step defines a forward operation and a compensation operation; when any step fails, compensations execute in reverse order to ensure eventual consistency. This article provides a deep analysis of Saga orchestration architecture, compensation strategy taxonomy, dead-letter queue handling, timeout control, and integration with ActionEngine.
CoomiaPublished on September 4, 202514 min read
Share this articleTwitter / X
“Series: S5 Intelligent Decisions · Article 13 | Level: Advanced | Reading Time: 20 min
Saga Pattern: Compensation and Rollback for Long Transactions
#TL;DR
When a decision requires multi-step operations spanning multiple microservices, traditional distributed transactions (2PC) cannot meet availability and performance requirements. coomia-dip adopts the Saga pattern with the Temporal workflow engine, decomposing long transactions into a series of compensable local transactions. Each step defines a forward operation and a compensation operation; when any step fails, compensations execute in reverse order to ensure eventual consistency. This article provides a deep analysis of Saga orchestration architecture, compensation strategy taxonomy, dead-letter queue handling, timeout control, and integration with ActionEngine.
#1. Challenges of Distributed Transactions
#1.1 Why 2PC Does Not Fit Microservices
Code
Two-Phase Commit (2PC) Problems:
┌──────────┐ Prepare ┌──────────┐
│Coordinat.│ ───────────────→ │ Service A │ holding lock
│ │ ───────────────→ │ Service B │ holding lock
│ │ ───────────────→ │ Service C │ holding lock
└──────────┘ └──────────┘
│
│ What if the Coordinator crashes?
│ → All participants wait indefinitely
│ → Resource locks cannot be released
│ → System availability plummets
▼
Single point of failure + performance bottleneck
| Dimension | 2PC | Saga |
|---|---|---|
| Consistency model | Strong | Eventual |
| Lock duration | Entire transaction | Local transaction only |
| Availability | Low (coordinator SPOF) | High (no global lock) |
| Performance | Poor (synchronous blocking) | Good (async execution) |
| Complexity | Low (framework built-in) | High (compensation design) |
| Use case | Single database | Cross-service orchestration |
#1.2 Core Idea of Saga
The Saga pattern was first proposed by Hector Garcia-Molina in 1987. Its core idea:
“Decompose a long transaction T into n sub-transactions T1, T2, ..., Tn, each Ti having a corresponding compensation Ci. If Ti fails, execute Ci-1, Ci-2, ..., C1 to roll back.
Code
Forward execution:
T1 ──→ T2 ──→ T3 ──→ T4 ──→ T5
✗ Failed!
Compensation rollback:
C3 ←── C2 ←── C1
(reverse-order compensation)
#2. coomia-dip Saga Architecture
#2.1 Overall Architecture
Code
┌─────────────────────────────────────────────────────────────────┐
│ Saga Orchestrator │
│ (Temporal Workflow) │
│ │
│ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ Step 1 │───→│ Step 2 │───→│ Step 3 │───→│ Step 4 │ │
│ │CreateOb│ │CreateRe│ │InvokeFn│ │Webhook │ │
│ └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘ │
│ │ │ │ │ │
│ ┌───┴────┐ ┌───┴────┐ ┌───┴────┐ ┌───┴────┐ │
│ │Comp. 1 │ │Comp. 2 │ │Comp. 3 │ │Comp. 4 │ │
│ │DeleteOb│ │DeleteRe│ │RevokeFn│ │ N/A │ │
│ └────────┘ └────────┘ └────────┘ └────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Saga State Store (PostgreSQL / Redis) │ │
│ │ saga_id | step | status | snapshot | compensation │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Dead Letter Queue (DLQ) │ │
│ │ failed compensations → manual review │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
#2.2 Core Data Models
Python
from enum import Enum
from pydantic import BaseModel, Field
from datetime import datetime
import uuid
class SagaStatus(str, Enum):
RUNNING = "running"
SUCCEEDED = "succeeded"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
FAILED = "failed" # Compensation also failed
class StepStatus(str, Enum):
PENDING = "pending"
EXECUTING = "executing"
SUCCEEDED = "succeeded"
FAILED = "failed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
COMPENSATION_FAILED = "compensation_failed"
class SagaStep(BaseModel):
"""A single step in a Saga"""
step_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
step_index: int
action_request: dict # Serialized ActionRequest
compensation_request: dict | None = None
status: StepStatus = StepStatus.PENDING
result: dict | None = None
error: str | None = None
snapshot: dict | None = None # Pre-execution state snapshot
started_at: datetime | None = None
completed_at: datetime | None = None
class SagaDefinition(BaseModel):
"""Saga definition"""
saga_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
name: str
steps: list[SagaStep]
status: SagaStatus = SagaStatus.RUNNING
timeout_seconds: int = 300
max_compensation_retries: int = 3
created_by: str = "system"
created_at: datetime = Field(default_factory=datetime.utcnow)
completed_at: datetime | None = None
metadata: dict = Field(default_factory=dict)
#3. Saga Orchestration Patterns
#3.1 Orchestration vs. Choreography
coomia-dip uses Orchestration-based Saga, with a central orchestrator controlling the flow:
Code
Orchestration (coomia-dip choice):
┌──────────────┐
│ Orchestrator │
│ (Temporal) │
└──────┬───────┘
│
┌────┼────┬────────┐
▼ ▼ ▼ ▼
Svc Svc Svc Svc
A B C D
Pros: Centralized visibility, easy debugging
Cons: Orchestrator is on the critical path
Choreography (not adopted):
Svc A ──event──→ Svc B ──event──→ Svc C
▲ │
└──────────event────────────────────┘
Pros: Decentralized
Cons: Scattered flow, hard to trace
#3.2 Temporal Integration
Python
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from datetime import timedelta
@activity.defn
async def execute_saga_step(step_data: dict) -> dict:
"""Execute a single Saga step"""
from action_engine import ActionScheduler
request = ActionRequest(**step_data["action_request"])
scheduler = ActionScheduler.get_instance()
result = await scheduler.dispatch(request)
return result.model_dump()
@activity.defn
async def compensate_saga_step(step_data: dict) -> dict:
"""Execute a compensation operation"""
from action_engine import ActionScheduler
original_request = ActionRequest(**step_data["action_request"])
original_result = ActionResult(**step_data["result"])
scheduler = ActionScheduler.get_instance()
executor = scheduler.executors.get(original_request.executor_type)
comp_result = await executor.compensate(
original_request, original_result
)
return comp_result.model_dump()
@workflow.defn
class SagaWorkflow:
"""Temporal Saga Workflow"""
@workflow.run
async def run(self, saga_def: dict) -> dict:
saga = SagaDefinition(**saga_def)
completed_steps: list[dict] = []
for step in saga.steps:
try:
result = await workflow.execute_activity(
execute_saga_step,
arg=step.model_dump(),
start_to_close_timeout=timedelta(
seconds=saga.timeout_seconds
),
retry_policy=RetryPolicy(
maximum_attempts=3,
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
),
)
step_record = {
"step": step.model_dump(),
"result": result,
}
completed_steps.append(step_record)
if result["status"] == "failed":
await self._compensate(
completed_steps,
saga.max_compensation_retries,
)
return {
"saga_id": saga.saga_id,
"status": "compensated",
"failed_step": step.step_index,
}
except Exception as e:
await self._compensate(
completed_steps,
saga.max_compensation_retries,
)
return {
"saga_id": saga.saga_id,
"status": "compensated",
"error": str(e),
}
return {
"saga_id": saga.saga_id,
"status": "succeeded",
"steps_completed": len(completed_steps),
}
async def _compensate(self, completed: list[dict],
max_retries: int) -> None:
"""Execute compensations in reverse order"""
for step_record in reversed(completed):
for attempt in range(max_retries):
try:
await workflow.execute_activity(
compensate_saga_step,
arg=step_record,
start_to_close_timeout=timedelta(seconds=60),
retry_policy=RetryPolicy(maximum_attempts=1),
)
break
except Exception as e:
if attempt == max_retries - 1:
workflow.logger.error(
f"Compensation failed after "
f"{max_retries} attempts: {e}"
)
await workflow.execute_activity(
enqueue_dead_letter,
arg=step_record,
start_to_close_timeout=timedelta(seconds=10),
)
#4. Compensation Strategy Taxonomy
#4.1 Four Compensation Strategies
Code
┌────────────────────────────────────────────────────────────┐
│ Compensation Strategy Taxonomy │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Exact Inverse│ │ Semantic │ │
│ │ │ │ Inverse │ │
│ │ Create → Del │ │ Approve → │ │
│ │ Add → Remove │ │ Reject │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Snapshot │ │ No-Op │ │
│ │ Restore │ │ │ │
│ │ │ │ Read-only │ │
│ │ Restore pre- │ │ operations │ │
│ │ execution │ │ need no comp │ │
│ │ state │ │ │ │
│ └──────────────┘ └──────────────┘ │
└────────────────────────────────────────────────────────────┘
#4.2 Compensation Strategy Per Executor
| Executor | Strategy | Implementation | Idempotency |
|---|---|---|---|
| CreateObject | Exact inverse | Delete created object | Idempotent (skip if absent) |
| UpdateObject | Snapshot restore | Restore pre-execution properties | Idempotent (version check) |
| DeleteObject | Snapshot restore | Recreate object from snapshot | Idempotent (skip if exists) |
| CreateRelation | Exact inverse | Delete created relation | Idempotent |
| DeleteRelation | Snapshot restore | Recreate relation from snapshot | Idempotent |
| InvokeFunction | Custom | Invoke user-defined compensation function | Implementation-dependent |
| Webhook | No-op/Custom | Send cancel webhook or no-op | External-system-dependent |
| Notification | No-op | Sent notifications cannot be recalled | N/A |
| SimpleOp | No-op | Inline operations have no side effects | N/A |
| CompositeOp | Recursive | Compensate all sub-steps | Sub-step-dependent |
#4.3 Snapshot Mechanism
Python
class SnapshotManager:
"""Pre-execution state snapshot manager"""
async def capture(self, request: ActionRequest) -> dict:
"""Capture current object state before execution"""
match request.executor_type:
case ExecutorType.UPDATE_OBJECT:
obj = await self.ontology.get_object(
object_type=request.target_object_type,
object_id=request.target_object_id,
)
return {
"type": "object_snapshot",
"object_type": request.target_object_type,
"object_id": request.target_object_id,
"properties": obj.properties,
"version": obj.version,
"captured_at": datetime.utcnow().isoformat(),
}
case ExecutorType.DELETE_OBJECT:
obj = await self.ontology.get_object(
object_type=request.target_object_type,
object_id=request.target_object_id,
)
return {
"type": "full_object_snapshot",
"object_type": request.target_object_type,
"object_data": obj.model_dump(),
"relations": await self._capture_relations(obj),
"captured_at": datetime.utcnow().isoformat(),
}
case _:
return {}
async def restore(self, snapshot: dict) -> None:
"""Restore state from snapshot"""
match snapshot.get("type"):
case "object_snapshot":
await self.ontology.update_object(
object_type=snapshot["object_type"],
object_id=snapshot["object_id"],
properties=snapshot["properties"],
expected_version=None, # Force overwrite
)
case "full_object_snapshot":
await self.ontology.create_object(
object_type=snapshot["object_type"],
properties=snapshot["object_data"]["properties"],
object_id=snapshot["object_data"]["id"],
)
for rel in snapshot.get("relations", []):
await self.ontology.create_relation(**rel)
#5. Timeout and Retry Strategies
#5.1 Multi-Layer Timeout Control
Code
┌─────────────────────────────────────────────────────┐
│ Timeout Hierarchy │
│ │
│ Saga global timeout: 300s │
│ ┌───────────────────────────────────────────────┐ │
│ │ │ │
│ │ Step timeout: 60s Step timeout: 60s │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ │ │ │ │ │
│ │ │ Activity: 30s │ │ Activity: 30s │ │ │
│ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │
│ │ │ │ gRPC: 10s │ │ │ │ HTTP: 15s │ │ │ │
│ │ │ └───────────┘ │ │ └───────────┘ │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ └───────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
#5.2 Retry Configuration
Python
class RetryConfig(BaseModel):
"""Retry configuration"""
max_attempts: int = 3
initial_interval_seconds: float = 1.0
backoff_coefficient: float = 2.0
max_interval_seconds: float = 60.0
non_retryable_errors: list[str] = Field(default_factory=lambda: [
"ValidationError",
"PermissionError",
"ObjectNotFoundError",
])
class SagaRetryPolicy:
"""Saga step retry policy"""
@staticmethod
def should_retry(error: Exception, config: RetryConfig,
attempt: int) -> bool:
error_type = type(error).__name__
if error_type in config.non_retryable_errors:
return False
if attempt >= config.max_attempts:
return False
return True
@staticmethod
def get_delay(config: RetryConfig, attempt: int) -> float:
delay = config.initial_interval_seconds * (
config.backoff_coefficient ** attempt
)
return min(delay, config.max_interval_seconds)
#6. Dead-Letter Queue and Manual Intervention
#6.1 Dead-Letter Queue Design
Code
┌────────────────────────────────────────────────────────┐
│ Dead Letter Queue (DLQ) │
│ │
│ ┌──────────┐ │
│ │ Failed │──→ DLQ Entry: │
│ │ compens. │ { │
│ └──────────┘ saga_id: "...", │
│ step_index: 2, │
│ action_request: {...}, │
│ original_result: {...}, │
│ compensation_error: "...", │
│ retry_count: 3, │
│ created_at: "2026-03-24T...", │
│ status: "pending_review" │
│ } │
│ │
│ ┌──────────────────────────────────────┐ │
│ │ DLQ Dashboard │ │
│ │ │ │
│ │ Pending: 3 │ Resolved: 47 │ │
│ │ ───────────────────────── │ │
│ │ saga-abc step-2 [Retry] [Skip] │ │
│ │ saga-def step-1 [Retry] [Manual] │ │
│ │ saga-ghi step-3 [Retry] [Skip] │ │
│ └──────────────────────────────────────┘ │
└────────────────────────────────────────────────────────┘
#6.2 DLQ Processor
Python
class DeadLetterProcessor:
"""Dead-letter queue processor"""
async def enqueue(self, step_record: dict,
error: Exception) -> str:
entry = {
"dlq_id": str(uuid.uuid4()),
"saga_id": step_record["step"]["saga_id"],
"step_index": step_record["step"]["step_index"],
"action_request": step_record["step"]["action_request"],
"original_result": step_record["result"],
"compensation_error": str(error),
"retry_count": 0,
"status": "pending_review",
"created_at": datetime.utcnow().isoformat(),
}
await self.store.insert("dead_letter_queue", entry)
await self._send_alert(entry)
return entry["dlq_id"]
async def retry(self, dlq_id: str) -> dict:
"""Manually retry compensation"""
entry = await self.store.get("dead_letter_queue", dlq_id)
try:
result = await self.action_scheduler.compensate_step(entry)
entry["status"] = "resolved"
entry["resolved_at"] = datetime.utcnow().isoformat()
await self.store.update("dead_letter_queue", dlq_id, entry)
return {"status": "resolved", "result": result}
except Exception as e:
entry["retry_count"] += 1
entry["last_error"] = str(e)
await self.store.update("dead_letter_queue", dlq_id, entry)
return {"status": "still_failed", "error": str(e)}
async def skip(self, dlq_id: str, reason: str) -> None:
"""Skip compensation (accept data inconsistency)"""
entry = await self.store.get("dead_letter_queue", dlq_id)
entry["status"] = "skipped"
entry["skip_reason"] = reason
entry["skipped_at"] = datetime.utcnow().isoformat()
await self.store.update("dead_letter_queue", dlq_id, entry)
await self.audit_log.record_skip(entry, reason)
#7. Real-World Example: Order Approval Saga
#7.1 Business Scenario
A complete order approval flow spans 4 services:
Code
Order Approval Saga:
Step 1: Create approval record (Ontology CreateObject)
Compensation: Delete approval record
Step 2: Freeze inventory (InvokeFunction → Inventory Service)
Compensation: Release frozen inventory
Step 3: Pre-charge (Webhook → Finance System)
Compensation: Refund pre-charge
Step 4: Send approval notification (Notification → Approver)
Compensation: None (notifications are irrevocable)
#7.2 Saga Definition
Python
order_approval_saga = SagaDefinition(
name="order_approval",
timeout_seconds=300,
max_compensation_retries=5,
steps=[
SagaStep(
step_index=0,
action_request={
"executor_type": "CreateObject",
"target_object_type": "ApprovalRecord",
"parameters": {
"properties": {
"order_id": "ORD-2026-001",
"status": "pending",
"requested_by": "user-abc",
"amount": 50000.00,
}
},
},
),
SagaStep(
step_index=1,
action_request={
"executor_type": "InvokeFunction",
"parameters": {
"function_rid": "ri.function.inventory.freeze",
"arguments": {
"order_id": "ORD-2026-001",
"items": [
{"sku": "SKU-001", "quantity": 10},
{"sku": "SKU-002", "quantity": 5},
],
},
"compensate_function_rid":
"ri.function.inventory.unfreeze",
},
},
),
SagaStep(
step_index=2,
action_request={
"executor_type": "Webhook",
"parameters": {
"url": "https://finance.internal/api/pre-charge",
"method": "POST",
"payload": {
"order_id": "ORD-2026-001",
"amount": 50000.00,
"currency": "CNY",
},
"timeout_seconds": 30,
},
},
compensation_request={
"executor_type": "Webhook",
"parameters": {
"url": "https://finance.internal/api/refund",
"method": "POST",
"payload": {
"order_id": "ORD-2026-001",
"amount": 50000.00,
},
},
},
),
SagaStep(
step_index=3,
action_request={
"executor_type": "Notification",
"parameters": {
"channel": "feishu",
"template_id": "tpl_order_approval",
"recipients": ["approver@company.com"],
"variables": {
"order_id": "ORD-2026-001",
"amount": "50,000.00 CNY",
},
},
},
),
],
)
#7.3 Execution Timeline
Code
Time ──→
T0 T1 T2 T3 T4 T5
│ │ │ │ │ │
│ Step 0: CreateObject │ │ │
│ ┌──────────────┐ │ │ │
│ │ Create record │ │ │ │
│ └──────┬───────┘ │ │ │
│ │ │ │ │
│ Step 1: InvokeFunction │ │
│ ┌──────────────┐│ │ │
│ │ Freeze inv. ││ │ │
│ └──────┬───────┘│ │ │
│ │ │ │ │
│ Step 2: Webhook │ │
│ ┌────────────────┐ │
│ │ Pre-charge ✗ FAIL! │
│ └────────┬──────┘ │
│ │ │
│ ← Trigger compensation chain ───── │
│ │ │
│ Comp 1: Release inventory │
│ ┌────────────────┐ │
│ │ unfreeze() │ │
│ └────────┬──────┘ │
│ │ │
│ Comp 0: Delete record │
│ ┌──────────────┐ │
│ │ DeleteObject │ │
│ └──────────────┘ │
│ │
│ Saga status: COMPENSATED │
#8. Observability and Monitoring
#8.1 Key Metrics
Python
SAGA_METRICS = {
"saga_started_total": Counter(
"saga_started_total",
"Total sagas started",
labels=["saga_name"],
),
"saga_completed_total": Counter(
"saga_completed_total",
"Total sagas completed",
labels=["saga_name", "status"],
# status: succeeded | compensated | failed
),
"saga_duration_seconds": Histogram(
"saga_duration_seconds",
"Saga total duration",
labels=["saga_name", "status"],
buckets=[1, 5, 10, 30, 60, 120, 300],
),
"saga_step_duration_seconds": Histogram(
"saga_step_duration_seconds",
"Individual step duration",
labels=["saga_name", "step_index", "executor_type"],
),
"saga_compensation_total": Counter(
"saga_compensation_total",
"Total compensation operations",
labels=["saga_name", "step_index"],
),
"saga_dlq_total": Counter(
"saga_dlq_total",
"Dead letter queue entries",
labels=["saga_name"],
),
}
#8.2 Distributed Tracing
Code
Trace: saga_order_approval_abc123
│
├── Span: saga.execute (300ms)
│ ├── Span: step[0].create_object (45ms)
│ │ └── Span: ontology.create (38ms)
│ ├── Span: step[1].invoke_function (120ms)
│ │ ├── Span: function.resolve (5ms)
│ │ └── Span: function.execute (110ms)
│ ├── Span: step[2].webhook (FAILED, 80ms)
│ │ └── Span: http.post finance/pre-charge (timeout)
│ │
│ ├── Span: compensate[1] (60ms)
│ │ └── Span: function.execute unfreeze (55ms)
│ └── Span: compensate[0] (30ms)
│ └── Span: ontology.delete (25ms)
#9. Best Practices
#9.1 Compensation Design Principles
| Principle | Description | Example |
|---|---|---|
| Idempotent compensation | Same compensation can safely run multiple times | Check object existence before delete |
| Snapshot first | Record state snapshot before execution | Save old properties before UpdateObject |
| Decouple compensation | Compensation logic is independent of forward logic | Separate compensate() method |
| Conservative timeouts | Compensation timeout should exceed forward timeout | Forward 30s → Compensation 60s |
| DLQ safety net | Failed compensations must have manual fallback | Alert + approval dashboard |
#9.2 Anti-Pattern Checklist
| Anti-Pattern | Problem | Correct Approach |
|---|---|---|
| Ignoring compensation design | Data inconsistency after failure | Every step must define compensation |
| Non-idempotent compensation | Retries cause duplicate operations | All compensations must be idempotent |
| No timeout control | Deadlocks or resource leaks | Set reasonable timeouts at every layer |
| No DLQ | Failed compensations silently lost | Always configure a dead-letter queue |
| Synchronous waiting | Blocks the caller | Async Saga + callback notification |
#Key Takeaways
- Saga beats 2PC: In microservice architectures, Saga's eventual consistency model is more appropriate than 2PC's strong consistency, eliminating lock-waiting and coordinator SPOF issues
- Orchestration architecture: coomia-dip uses Temporal for orchestration-based Saga, providing centralized visibility, easy debugging, and monitoring
- Four compensation strategies: Exact inverse, semantic inverse, snapshot restore, and no-op — choose based on operation type
- Multi-layer timeout control: Saga global → Step → Activity → Network timeouts provide defense in depth
- DLQ safety net: Failed compensations must not be silently lost; dead-letter queue + alerting + manual approval dashboard is the last line of defense
- Snapshot mechanism: Capturing pre-execution state snapshots is the foundation of reliable compensation, especially for Update and Delete operations
#Next Article
The next article, S5-14 Mutation Rules: Declarative State Change Orchestration, introduces how to define object state change logic through declarative rules, transforming business-readable rules into automated Action sequences.
tags: saga, temporal, compensation, distributed-transaction, dead-letter-queue, coomia-dip