Saga 模式：长事务的补偿与回滚设计
当一个决策需要跨越多个微服务执行多步操作时，传统的分布式事务（2PC）无法满足高可用和性能要求。coomia-dip 采用 Saga 模式 配合 Temporal 工作流引擎，将长事务拆解为一系列可补偿的本地事务。每个步骤定义正向操作和补偿操作，当某一步失败时按反序执行补偿，确保系统最终一致性。本文深入剖析 Saga 编排的架构设计、补偿策略分类、死信队列处理、超时控制以及与 ActionEngine 的集成方式。
Coomia发布于 2025年9月4日16 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 13 篇 | 难度：高级 | 阅读时间：20 分钟
Saga 模式：长事务的补偿与回滚设计
#TL;DR
当一个决策需要跨越多个微服务执行多步操作时，传统的分布式事务（2PC）无法满足高可用和性能要求。coomia-dip 采用 Saga 模式 配合 Temporal 工作流引擎，将长事务拆解为一系列可补偿的本地事务。每个步骤定义正向操作和补偿操作，当某一步失败时按反序执行补偿，确保系统最终一致性。本文深入剖析 Saga 编排的架构设计、补偿策略分类、死信队列处理、超时控制以及与 ActionEngine 的集成方式。
#1. 分布式事务的挑战
#1.1 为什么 2PC 不适合微服务
Code
两阶段提交（2PC）的问题：
┌──────────┐ Prepare ┌──────────┐
│Coordinat.│ ───────────────→ │ Service A │ 持有锁
│ │ ───────────────→ │ Service B │ 持有锁
│ │ ───────────────→ │ Service C │ 持有锁
└──────────┘ └──────────┘
│
│ 如果 Coordinator 崩溃？
│ → 所有参与者无限等待
│ → 资源锁无法释放
│ → 系统可用性骤降
▼
单点故障 + 性能瓶颈
| 对比维度 | 2PC | Saga |
|---|---|---|
| 一致性模型 | 强一致 | 最终一致 |
| 锁持有时间 | 整个事务期间 | 仅本地事务 |
| 可用性 | 低（协调者单点） | 高（无全局锁） |
| 性能 | 差（同步阻塞） | 好（异步执行） |
| 复杂度 | 低（框架内置） | 高（需设计补偿） |
| 适用场景 | 单数据库 | 跨服务编排 |
#1.2 Saga 的核心思想
Saga 最早由 Hector Garcia-Molina 在 1987 年提出，核心思想是：
“将一个长事务 T 分解为 n 个子事务 T1, T2, ..., Tn，每个 Ti 有对应的补偿事务 Ci。如果 Ti 失败，则执行 Ci-1, Ci-2, ..., C1 进行回滚。
Code
正向执行：
T1 ──→ T2 ──→ T3 ──→ T4 ──→ T5
✗ 失败！
补偿回滚：
C3 ←── C2 ←── C1
(反序执行补偿)
#2. coomia-dip 的 Saga 架构
#2.1 整体架构
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 核心数据模型
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" # 补偿也失败
class StepStatus(str, Enum):
PENDING = "pending"
EXECUTING = "executing"
SUCCEEDED = "succeeded"
FAILED = "failed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
COMPENSATION_FAILED = "compensation_failed"
class SagaStep(BaseModel):
"""Saga 的一个步骤"""
step_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
step_index: int
action_request: dict # ActionRequest 序列化
compensation_request: dict | None = None
status: StepStatus = StepStatus.PENDING
result: dict | None = None
error: str | None = None
snapshot: dict | None = None # 执行前的状态快照
started_at: datetime | None = None
completed_at: datetime | None = None
class SagaDefinition(BaseModel):
"""Saga 定义"""
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 编排模式
#3.1 编排式 vs 协同式
coomia-dip 选择 编排式 Saga（Orchestration），由中央编排器控制流程：
Code
编排式 (Orchestration) — coomia-dip 采用:
┌──────────────┐
│ Orchestrator │
│ (Temporal) │
└──────┬───────┘
│
┌────┼────┬────────┐
▼ ▼ ▼ ▼
Svc Svc Svc Svc
A B C D
优点：流程集中可见、易调试
缺点：编排器是关键路径
协同式 (Choreography) — 未采用:
Svc A ──event──→ Svc B ──event──→ Svc C
▲ │
└──────────event────────────────────┘
优点：去中心化
缺点：流程分散、难追踪
#3.2 Temporal 集成
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:
"""执行单个 Saga 步骤"""
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:
"""执行补偿操作"""
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.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:
"""反序执行补偿"""
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 {max_retries} "
f"attempts: {e}"
)
await workflow.execute_activity(
enqueue_dead_letter,
arg=step_record,
start_to_close_timeout=timedelta(seconds=10),
)
#4. 补偿策略分类
#4.1 四种补偿策略
Code
┌────────────────────────────────────────────────────────────┐
│ 补偿策略分类 │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 精确反向 │ │ 语义反向 │ │
│ │ Exact Inverse│ │ Semantic Inv. │ │
│ │ │ │ │ │
│ │ Create → Del │ │ Approve → │ │
│ │ Add → Remove │ │ Reject │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 快照恢复 │ │ 无操作 │ │
│ │ Snapshot │ │ No-Op │ │
│ │ Restore │ │ │ │
│ │ │ │ 只读操作 │ │
│ │ 恢复到执行前 │ │ 无需补偿 │ │
│ │ 的状态快照 │ │ │ │
│ └──────────────┘ └──────────────┘ │
└────────────────────────────────────────────────────────────┘
#4.2 各执行器的补偿策略
| 执行器 | 补偿策略 | 实现方式 | 幂等性 |
|---|---|---|---|
| CreateObject | 精确反向 | 删除已创建对象 | 幂等（对象不存在则跳过） |
| UpdateObject | 快照恢复 | 恢复为执行前的属性快照 | 幂等（版本检查） |
| DeleteObject | 快照恢复 | 从快照重建对象 | 幂等（对象存在则跳过） |
| CreateRelation | 精确反向 | 删除已创建关系 | 幂等 |
| DeleteRelation | 快照恢复 | 从快照重建关系 | 幂等 |
| InvokeFunction | 自定义 | 调用用户定义的补偿函数 | 依赖实现 |
| Webhook | 无操作/自定义 | 发送取消 Webhook 或无操作 | 依赖外部系统 |
| Notification | 无操作 | 已发送通知无法撤回 | N/A |
| SimpleOp | 无操作 | 内联操作无副作用 | N/A |
| CompositeOp | 递归补偿 | 补偿所有子步骤 | 依赖子步骤 |
#4.3 快照机制
Python
class SnapshotManager:
"""执行前状态快照管理"""
async def capture(self, request: ActionRequest) -> dict:
"""在执行前捕获对象当前状态"""
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:
"""从快照恢复状态"""
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, # 强制覆盖
)
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"], # 使用原 ID
)
for rel in snapshot.get("relations", []):
await self.ontology.create_relation(**rel)
#5. 超时与重试策略
#5.1 多层超时控制
Code
┌─────────────────────────────────────────────────────┐
│ 超时层级 │
│ │
│ Saga 全局超时：300s │
│ ┌───────────────────────────────────────────────┐ │
│ │ │ │
│ │ Step 超时：60s Step 超时：60s │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ │ │ │ │ │
│ │ │ Activity：30s │ │ Activity：30s │ │ │
│ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │
│ │ │ │ gRPC: 10s │ │ │ │ HTTP: 15s │ │ │ │
│ │ │ └───────────┘ │ │ └───────────┘ │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ └───────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
#5.2 重试策略配置
Python
class RetryConfig(BaseModel):
"""重试配置"""
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 步骤重试策略"""
@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. 死信队列与人工介入
#6.1 死信队列设计
Code
┌────────────────────────────────────────────────────────┐
│ Dead Letter Queue (DLQ) │
│ │
│ ┌──────────┐ │
│ │ 补偿失败 │──→ DLQ Entry: │
│ │ 的步骤 │ { │
│ └──────────┘ 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 处理器
Python
class DeadLetterProcessor:
"""死信队列处理器"""
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:
"""手动重试补偿"""
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:
"""跳过补偿（接受数据不一致）"""
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. 实战案例：订单审批 Saga
#7.1 业务场景
一个完整的订单审批流程需要跨越 4 个服务：
Code
订单审批 Saga：
Step 1: 创建审批记录（Ontology CreateObject）
补偿: 删除审批记录
Step 2: 冻结库存（InvokeFunction → 库存服务）
补偿: 释放已冻结库存
Step 3: 预扣款（Webhook → 财务系统）
补偿: 退回预扣款
Step 4: 发送审批通知（Notification → 审批人）
补偿: 无（通知不可撤回）
#7.2 Saga 定义
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 执行时序
Code
时间 ──→
T0 T1 T2 T3 T4 T5
│ │ │ │ │ │
│ Step 0: CreateObject │ │ │
│ ┌──────────────┐ │ │ │
│ │ 创建审批记录 │ │ │ │
│ └──────┬───────┘ │ │ │
│ │ │ │ │
│ Step 1: InvokeFunction │ │
│ ┌──────────────┐│ │ │
│ │ 冻结库存 ││ │ │
│ └──────┬───────┘│ │ │
│ │ │ │ │
│ Step 2: Webhook │ │
│ ┌────────────────┐ │
│ │ 预扣款 ✗ 失败！ │
│ └────────┬──────┘ │
│ │ │
│ ← 触发补偿链 ────────── │
│ │ │
│ Comp 1: 释放库存 │
│ ┌────────────────┐ │
│ │ unfreeze() │ │
│ └────────┬──────┘ │
│ │ │
│ Comp 0: 删除审批记录 │
│ ┌──────────────┐ │
│ │ DeleteObject │ │
│ └──────────────┘ │
│ │
│ Saga 状态: COMPENSATED │
#8. 可观测性与监控
#8.1 关键指标
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 分布式追踪
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. 最佳实践
#9.1 补偿设计原则
| 原则 | 说明 | 示例 |
|---|---|---|
| 幂等补偿 | 同一补偿可安全执行多次 | 删除前检查对象是否存在 |
| 快照优先 | 执行前记录状态快照 | UpdateObject 前保存旧属性 |
| 补偿解耦 | 补偿逻辑独立于正向逻辑 | 独立的 compensate() 方法 |
| 超时保守 | 补偿超时应大于正向超时 | 正向 30s → 补偿 60s |
| DLQ 兜底 | 补偿失败必须有人工兜底 | 告警 + 审批仪表板 |
#9.2 反模式清单
| 反模式 | 问题 | 正确做法 |
|---|---|---|
| 忽略补偿设计 | 失败后数据不一致 | 每个步骤必须定义补偿 |
| 补偿非幂等 | 重试导致重复操作 | 所有补偿操作必须幂等 |
| 无超时控制 | 死锁或资源泄漏 | 每层设置合理超时 |
| 无 DLQ | 补偿失败静默丢失 | 必须配置死信队列 |
| 同步等待 | 阻塞调用方 | 异步 Saga + 回调通知 |
#Key Takeaways
- Saga 优于 2PC：微服务架构下 Saga 的最终一致性模型比 2PC 强一致性更适合，解决了锁等待和协调者单点问题
- 编排式架构：coomia-dip 通过 Temporal 实现编排式 Saga，流程集中可见、易于调试和监控
- 四种补偿策略：精确反向、语义反向、快照恢复、无操作 — 根据操作类型选择最合适的策略
- 多层超时控制：Saga 全局超时 → Step 超时 → Activity 超时 → 网络超时，层层保护
- DLQ 兜底：补偿失败不能静默丢失，死信队列 + 告警 + 人工审批仪表板是最后防线
- 快照机制：执行前捕获状态快照是可靠补偿的基础，特别是 Update 和 Delete 操作
#Next Article
下一篇 S5-14 Mutation Rules：声明式状态变更编排 将介绍如何用声明式规则定义对象状态变更逻辑，将业务人员可理解的规则转化为自动化的 Action 序列。
tags: saga, temporal, compensation, distributed-transaction, dead-letter-queue, coomia-dip