Saga 编排实战在 agents24 仓库的 saga-orchestration 技能中实现跨服务分布式事务与补偿【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents导读本文围绕 agents24Multi-harness agentic plugin marketplace仓库中 backend-development 插件的saga-orchestration技能展开系统讲解如何在无法使用两阶段提交2PC的微服务架构中用 Saga 模式编排跨聚合的分布式事务与长时业务工作流。读完本文你将掌握编排式与编舞式两类 Saga 的选型与实现模板、幂等步骤守卫、严格逆序补偿、分步骤超时配置以及卡死 Saga 检测与 DLQ 恢复等生产级可观测性方案——全部内容均以 SKILL.md 及其两个参考文献 details.md、advanced-patterns.md 为准并补充实现层面的细节说明。一、这个技能是做什么的定位、输入与产出在 agents 仓库中技能以plugins/{plugin-name}/skills/{skill-name}/SKILL.md的约定组织见 docs/architecture.md每个技能用 frontmatter 声明name与description以便 Agent 在遇到分布式事务需求时被正确唤起。saga-orchestration技能位于 saga-orchestration 目录SKILL.md的 frontmatter 描述明确了它的适用信号例如在跨微服务实现分布式事务、且 2PC 不可用为横跨库存inventory、支付payment、物流shipping等服务的失败订单流设计补偿动作为旅行预订系统酒店、航班、租车需原子回滚构建事件驱动的 Saga 协调器排查生产环境中补偿步骤永远无法完成、状态卡死的 Saga。技能输入开始前你应提供的信息输入项含义与作用服务边界与归属每个业务步骤由哪个服务拥有、哪个服务执行事务需求哪些步骤必须原子、哪些可以最终一致每步失败模式瞬时故障可重试还是永久故障需补偿以及重试策略每步 SLA为超时配置提供依据不同步骤延迟差异很大现有消息/事件基础设施Kafka、RabbitMQ、SQS 等决定采用何种事件总线技能产出交付什么带有序步骤、action 命令与 compensation 命令的 Saga 定义所选模式编排或编舞的协调器或实现每个参与服务的补偿逻辑幂等、始终成功语义带分步截止时间的步骤超时配置监控体系状态机指标、卡死 Saga 检测、DLQ 恢复。从仓库源码结构可以推断该技能被组织为“导航层SKILL.md 详述层references/”两级核心决策、最佳实践与排障放在 SKILL.md而完整模板与进阶实现被拆到references/下的两个文件以便 Agent 按需渐进加载、节省上下文。二、何时使用这个技能技能的适用场景判断如下需要在不加分布式锁的前提下协调多服务事务需要为部分失败实现补偿式事务管理耗时数分钟到数小时的长时业务工作流订单履约、审批、预订等在需要原子性的分布式系统中优雅处理失败希望用异步补偿替换脆弱的 2PC。需要警惕的是该技能同时提供了一条反向判断并非所有跨服务流程都需要 Saga——纯 CRUD、无状态请求/响应以及实时流式处理更适合普通 API 或消息引擎。相关的workflow-orchestration-patterns技能在讨论 Temporal 等更高层工作流引擎时也明确指出简单的数据加工流水线与统计型查询不应引入重量级编排详见 workflow-orchestration-patterns/SKILL.md。三、核心概念两种编排形态与执行状态机3.1 Saga 模式的两大类型原文档用一张 ASCII 图直观对比了两种拓扑Choreography Orchestration ┌─────┐ ┌─────┐ ┌─────┐ ┌─────────────┐ │Svc A│─►│Svc B│─►│Svc C│ │ Orchestrator│ └─────┘ └─────┘ └─────┘ └──────┬──────┘ │ │ │ │ ▼ ▼ ▼ ┌─────┼─────┐ Event Event Event ▼ ▼ ▼ ┌────┐┌────┐┌────┐ Each service reacts to the │Svc1││Svc2││Svc3│ previous services event. └────┘└────┘└────┘ No central coordinator. Central coordinator sends commands and tracks state.何时选编排Orchestration你需要显式的步骤跟踪、重试与集中的可视化能力且更易于调试。由唯一的协调器向各参与服务下发命令、推进状态机实现细节见SagaOrchestrator基类下文第五节。何时选编舞Choreography你希望服务间松耦合、可独立演进但要接受链路更难追踪的代价。每个服务只订阅上一个服务发布的事件并作出反应失败通过事件反向传播触发补偿。3.2 Saga 执行状态状态描述StartedSaga 已启动首个步骤已派发Pending等待参与服务返回某一步的回复Compensating某步骤失败正在回滚已完成步骤Completed全部正向步骤成功FailedSaga 失败且所有补偿均已结束在 advanced-patterns.md 的基类实现中这五个状态被建模为SagaState枚举与Saga聚合对象的state字段一一对应每个状态转换后都会调用saga_store.save()落盘。3.3 补偿规则决策表场景处理方式步骤从未启动无需补偿跳过步骤已成功完成执行补偿命令步骤完成前失败无需补偿直接标记失败补偿本身失败退避重试 → DLQ → 人工介入告警步骤结果已不存在将补偿视为成功幂等这张表是后续编写补偿处理器时最关键的决策依据不是每个已派发步骤都需要补偿只有“已成功完成”的步骤才进入逆序回滚。四、三大实战模板details.md提供了三个可直接参考的核心模板构成“定义 Saga → 参与服务响应 → 防重放保护”的完整闭环。4.1 模板一订单履约编排式 Saga做法是继承抽象协调器并实现saga_type与define_steps()声明四个步骤及其反向补偿命令from saga_orchestrator import SagaOrchestrator, SagaStep from typing import Dict, List class OrderFulfillmentSaga(SagaOrchestrator): Orchestrates order fulfillment across four participant services. property def saga_type(self) - str: return OrderFulfillment def define_steps(self, data: Dict) - List[SagaStep]: return [ SagaStep( namereserve_inventory, actionInventoryService.ReserveItems, compensationInventoryService.ReleaseReservation ), SagaStep( nameprocess_payment, actionPaymentService.ProcessPayment, compensationPaymentService.RefundPayment ), SagaStep( namecreate_shipment, actionShippingService.CreateShipment, compensationShippingService.CancelShipment ), SagaStep( namesend_confirmation, actionNotificationService.SendOrderConfirmation, compensationNotificationService.SendCancellationNotice ), ]启动一个 Saga 实例async def create_order(order_data: Dict, saga_store, event_publisher): saga OrderFulfillmentSaga(saga_store, event_publisher) return await saga.start({ order_id: order_data[order_id], customer_id: order_data[customer_id], items: order_data[items], payment_method: order_data[payment_method], shipping_address: order_data[shipping_address], })参与服务收到命令后处理业务并通过事件总线发布SagaStepCompleted或SagaStepFailed作为回复事件中必须携带saga_id与step_nameclass InventoryService: async def handle_reserve_items(self, command: Dict): try: reservation await self.reserve(command[items], command[order_id]) await self.event_publisher.publish(SagaStepCompleted, { saga_id: command[saga_id], step_name: reserve_inventory, result: {reservation_id: reservation.id} }) except InsufficientInventoryError as e: await self.event_publisher.publish(SagaStepFailed, { saga_id: command[saga_id], step_name: reserve_inventory, error: str(e) })需要特别说明的是文档中反复强调的一条补偿纪律补偿处理器即使发现资源已被回滚也必须“视为成功”并照常发布SagaCompensationCompleted否则协调器的终态判断会一直悬在 Compensatingasync def handle_release_reservation(self, command: Dict): Compensation — idempotent, always publishes completion. try: await self.release_reservation( command[original_result][reservation_id] ) except ReservationNotFoundError: pass # Already released — treat as success await self.event_publisher.publish(SagaCompensationCompleted, { saga_id: command[saga_id], step_name: reserve_inventory })4.2 模板二编舞式 Saga编舞式实现没有中心协调器。核心是两件事一个随事件流传递的SagaContext携带saga_id、当前step、业务data与completed_steps以及每个服务上的事件订阅与发布逻辑from dataclasses import dataclass from typing import Dict, Any dataclass class SagaContext: Carried through all events in a choreographed saga. saga_id: str step: int data: Dict[str, Any] completed_steps: list class OrderChoreographySaga: Choreography-based saga — services react to each others events. def __init__(self, event_bus): self.event_bus event_bus self._register_handlers() def _register_handlers(self): # Forward path self.event_bus.subscribe(OrderCreated, self._on_order_created) self.event_bus.subscribe(InventoryReserved, self._on_inventory_reserved) self.event_bus.subscribe(PaymentProcessed, self._on_payment_processed) self.event_bus.subscribe(ShipmentCreated, self._on_shipment_created) # Compensation path self.event_bus.subscribe(PaymentFailed, self._on_payment_failed) self.event_bus.subscribe(ShipmentFailed, self._on_shipment_failed)正向路径上每个处理器消费上一环的领域事件并发布下一环的命令事件ReserveInventory → ProcessPayment → CreateShipment → OrderFulfilled补偿路径上失败事件会触发级联反向操作。例如ShipmentFailed同时触发退款与释放库存async def _on_payment_failed(self, event: Dict): Payment failed — release inventory and mark order failed. await self.event_bus.publish(ReleaseInventory, { saga_id: event[saga_id], reservation_id: event[reservation_id], }) await self.event_bus.publish(OrderFailed, { order_id: event[order_id], reason: Payment failed, }) async def _on_shipment_failed(self, event: Dict): Shipment failed — refund payment and release inventory. await self.event_bus.publish(RefundPayment, { saga_id: event[saga_id], payment_id: event[payment_id], }) await self.event_bus.publish(ReleaseInventory, { saga_id: event[saga_id], reservation_id: event[reservation_id], })编舞式结构的优点是新增服务只需订阅相关事件但其固有的追踪难度意味着你必须依赖可靠的saga_id关联与后文的可观测性手段。4.3 模板三幂等步骤守卫由于 Broker 重连会导致命令被重放每个参与方都必须用幂等键抵御重复投递执行前先按idempotency_key查询命中则直接返回缓存结果、不产生副作用未命中才真正执行并保存结果async def handle_reserve_items(self, command: Dict): Idempotency-guarded reservation step. idempotency_key freserve-{command[order_id]} existing await self.reservation_store.find_by_key(idempotency_key) if existing: # Already executed — return the previous result without side effects await self.event_publisher.publish(SagaStepCompleted, { saga_id: command[saga_id], step_name: reserve_inventory, result: {reservation_id: existing.id} }) return # First execution reservation await self.reserve( itemscommand[items], order_idcommand[order_id], idempotency_keyidempotency_key ) await self.event_publisher.publish(SagaStepCompleted, { saga_id: command[saga_id], step_name: reserve_inventory, result: {reservation_id: reservation.id} })在原文档的排障章节中该模板被明确引用为“编排器重启后防止 Saga 重复执行”的标准解法。五、进阶模式从基类到生产落地advanced-patterns.md 明确声明其内容是“从核心技能中抽取的复杂实现供大多数 Saga 不需要的更深度参考”。以下逐一拆解。5.1SagaOrchestrator抽象基类状态机的引擎基类承担全部状态转换、补偿排序与事件发布任何 Saga 类型只需子类化并实现两个抽象成员。先看核心数据模型from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Any, Optional from datetime import datetime, timedelta import uuid class SagaState(Enum): STARTED started PENDING pending COMPENSATING compensating COMPLETED completed FAILED failed dataclass class SagaStep: name: str action: str compensation: str status: str pending result: Optional[Dict] None error: Optional[str] None executed_at: Optional[datetime] None compensated_at: Optional[datetime] None timeout_at: Optional[datetime] None dataclass class Saga: saga_id: str saga_type: str state: SagaState data: Dict[str, Any] steps: List[SagaStep] current_step: int 0 created_at: datetime field(default_factorydatetime.utcnow) updated_at: datetime field(default_factorydatetime.utcnow)从源码结构可以推断基类职责被设计为四点通过异步命令消息顺序执行步骤任何失败都触发逆序补偿每次状态转换后持久化 Saga完成与失败时发布领域事件。关键方法如下start(data)生成uuid4作为saga_id构造STARTED状态的 Saga 并落库随即派发第一个步骤handle_step_completed(saga_id, step_name, result)将对应步骤标记为completed、保存结果与时间戳current_step加一若已遍历完所有步骤则置为COMPLETED并调用_on_saga_completed否则置为PENDING并派发下一步handle_step_failed(saga_id, step_name, error)标记失败步骤并置 Saga 为COMPENSATING随后调用_compensate_execute_next_step(saga)将当前步骤标记为executing并发布step.action命令消息体携带saga_id、step_name与全部业务data_compensate(saga)从current_step - 1倒推到0仅对状态为completed的步骤发布其compensation命令并把原步骤的result以original_result字段带给补偿处理器handle_compensation_completed(...)当所有已执行步骤都处于compensated或pending/failed时将 Saga 置为FAILED并发布失败事件。async def _compensate(self, saga: Saga): Execute compensation steps in reverse order. for i in range(saga.current_step - 1, -1, -1): step saga.steps[i] if step.status completed: step.status compensating await self.saga_store.save(saga) await self.event_publisher.publish( step.compensation, { saga_id: saga.saga_id, step_name: step.name, original_result: step.result, **saga.data } )这条倒序循环正是“补偿顺序与执行顺序严格相反”这一不变式的代码级保证也是原文档排障章节要求用集成测试逐步骤注入失败来验证的落点。5.2 带分步超时的TimeoutSagaOrchestrator各步骤参与方的 SLA 差异极大示例中支付约 30 秒、物流面单创建可能 15 分钟因此原文档明确反对“全局超时”。TimeoutSagaOrchestrator通过一个STEP_TIMEOUTS字典为每个步骤名配置独立截止时间并支持子类覆盖class TimeoutSagaOrchestrator(SagaOrchestrator): Extends the base orchestrator with configurable per-step timeouts. # Override per saga subclass as needed STEP_TIMEOUTS: Dict[str, timedelta] { reserve_inventory: timedelta(minutes2), process_payment: timedelta(minutes1), create_shipment: timedelta(minutes15), send_confirmation: timedelta(minutes2), }其_execute_next_step在派发命令前计算timeout_at now STEP_TIMEOUTS[step.name]未命中的步骤回退到 5 分钟默认值并向调度器注册一个看门狗任务saga_timeout_{saga_id}_{step_name}。调度器到点回调_check_timeout若该步骤仍处于executing状态就通过handle_step_failed触发补偿而handle_step_completed会先取消对应超时任务再处理成功回复避免“慢但有效”的步骤被误判失败。5.3 完整的银行转账补偿链示例以跨账户转账为例展示补偿链如何“成对反转”debit_source的补偿是credit_sourcecredit_destination的补偿是debit_destination通知步骤则以失败通知作为对成功通知的“软补偿”class BankTransferSaga(SagaOrchestrator): Saga for transferring funds between accounts across services. property def saga_type(self) - str: return BankTransfer def define_steps(self, data: Dict) - List[SagaStep]: return [ SagaStep( namedebit_source, actionAccountService.DebitAccount, compensationAccountService.CreditAccount # reverse the debit ), SagaStep( namecreate_transfer_record, actionLedgerService.CreateTransfer, compensationLedgerService.VoidTransfer ), SagaStep( namecredit_destination, actionAccountService.CreditDestinationAccount, compensationAccountService.DebitAccount # reverse the credit ), SagaStep( namenotify_parties, actionNotificationService.SendTransferConfirmation, compensationNotificationService.SendTransferFailureNotice ), ]扣款步骤的处理器同时演示了“幂等键 区分成功/失败 补偿总是发布结果事件”三件事class AccountService: async def handle_debit_account(self, command: Dict): idempotency_key fdebit-{command[saga_id]}-{command[account_id]} existing await self.ledger.find_by_key(idempotency_key) if existing: await self._publish_completed(command, {transaction_id: existing.id}) return try: txn await self.ledger.debit( account_idcommand[source_account_id], amountcommand[amount], idempotency_keyidempotency_key ) await self._publish_completed(command, {transaction_id: txn.id}) except InsufficientFundsError as e: await self._publish_failed(command, str(e)) async def handle_credit_account(self, command: Dict): Compensation: credit back a previously debited account. idempotency_key fcredit-comp-{command[saga_id]}-{command[account_id]} existing await self.ledger.find_by_key(idempotency_key) if not existing: await self.ledger.credit( account_idcommand[source_account_id], amountcommand[amount], idempotency_keyidempotency_key ) # Always publish — even if already credited await self.event_publisher.publish(SagaCompensationCompleted, { saga_id: command[saga_id], step_name: debit_source })注意补偿侧的幂等键与正向侧相互独立credit-comp-...vsdebit-...这样即使补偿被重复执行也不会产生两次入账。5.4 生产监控Prometheus 指标与卡死检测advanced-patterns.md提供了用prometheus_client暴露 saga 健康指标的完整方案核心指标按saga_type打标签saga_started_total/saga_completed_total/saga_failed_total/saga_compensating_total四类计数刻画开始、成功、失败、进入补偿的数量saga_duration_secondsHistogrambuckets 为[1, 5, 15, 30, 60, 300, 600]秒刻画不同结果下的执行时长分布saga_stuck_countGauge按saga_type与state打标签跟踪卡在 Compensating/Pending 超过阈值的实例数。InstrumentedSagaOrchestrator通过覆写start、_on_saga_completed、_on_saga_failed完成埋点并用time.monotonic()记录开始时刻计算真实耗时。文档同时给出两条可直接用于告警的 PromQL# Alert: saga stuck in compensation for 10 min increase(saga_compensating_total[10m]) - increase(saga_failed_total[10m]) 0 # Alert: saga completion rate drops below 95% ( rate(saga_completed_total[5m]) / (rate(saga_completed_total[5m]) rate(saga_failed_total[5m])) ) 0.955.5 DLQ 恢复工作器当补偿处理器抛出未捕获异常时消息会落入死信队列。恢复工作器SagaDLQRecovery以指数退避重放 DLQ 消息每次重试延迟为BASE_DELAY_SECONDS * (2 ** attempt)基数为 10 秒达到MAX_RETRIES 5次后转入毒信队列并触发人工告警class SagaDLQRecovery: Replays failed compensation messages from the dead-letter queue. MAX_RETRIES 5 BASE_DELAY_SECONDS 10 async def process_dlq_message(self, message: Dict, attempt: int): delay self.BASE_DELAY_SECONDS * (2 ** attempt) if attempt self.MAX_RETRIES: await self._move_to_poison_queue(message) await self._alert_on_call(message) return await asyncio.sleep(delay) try: await self.event_publisher.publish(message[original_topic], message[payload]) except Exception as e: await self.process_dlq_message(message, attempt 1)六、最佳实践Dos 与 Donts原文档将实践规范浓缩为一组对称的准则应当做到Dos让每一步幂等——命令可能在 Broker 重连后被重放精心设计补偿——补偿是系统中最关键的代码路径贯穿使用关联 ID——saga_id必须流经每个事件与每条日志实现分步超时——绝不要无限期等待参与方回复记录状态转换日志——每次变更都要带上saga_id、step_name、old_state → new_state显式测试补偿路径——在集成测试中于每个步骤下标注入失败。不要做Donts不要假设立即完成——Saga 是异步的可能持续数分钟不要跳过补偿测试——回滚路径最难写对不要让服务直接耦合——Saga 步骤内部使用异步消息绝不使用同步调用不要忽略部分失败——某步骤已部分执行也必须补偿不要使用全局超时——各步骤延迟特征不同。对照 workflow-orchestration-patterns/SKILL.md 中的 Temporal 实践可以发现这份清单幂等、超时、可重试/不可重试错误区分与更高层工作流引擎对 activity 的要求一脉相承属于分布式编排领域的通用铁律。七、排障手册五种生产高发问题7.1 Saga 卡在 COMPENSATING 状态现象Saga 进入补偿却始终到不了 FAILED。根因通常是某个补偿处理器抛出了未捕获异常且从未发布SagaCompensationCompleted。对策给补偿消费者增加 DLQ 处理并保证每条补偿动作即使在底层已被回滚的情况下也照常发布结果事件代码示例见 4.1 节的handle_release_reservation。7.2 重启后 Saga 重复执行现象编排器在 Saga 中途重启后重放事件导致已完成步骤被再次执行。对策用幂等键守卫每个步骤动作——即前文模板三的“先查幂等存储、命中则返回缓存结果”。原文档明确将此场景指向details.md的 Template 3。7.3 编舞式 Saga 丢失事件现象下游服务离线期间错过了发布的事件。对策使用具备持久化的可靠消息中间件如开启副本的 Kafka、持久化 RabbitMQ并将当前 Saga 状态落盘到专门的saga_log表以便从“最后一个已知良好的步骤”重放。这与event-store-design技能中“事件应 append-only、可订阅、可重放”的事件库思想完全一致。7.4 超时先于“慢但有效”的步骤触发现象create_shipment高峰时可能耗时 15 分钟但全局超时只有 5 分钟导致误触发补偿。对策按步骤类型配置超时——即前文 5.2 节TimeoutSagaOrchestrator及其STEP_TIMEOUTS字典实现。7.5 补偿顺序与执行顺序不匹配现象两步都完成、随后才检测到失败时若补偿未按严格逆序执行会造成数据不一致。对策核验_compensate()是否从current_step - 1迭代到0见 5.1 节源码并添加一个在每个步骤下标处故意注入失败的集成测试以确认回滚顺序正确。八、在技能生态中的位置与相邻技能如何协作原文档在末尾列出了三个相关技能明确它们与 Saga 的组合方式cqrs-implementationSKILL.md在每一步完成后用 Saga 事件驱动 CQRS 读模型更新——读模型由 projector 消费事件做反规范化正适合承接 Saga 逐步提交的业务结果event-store-designSKILL.md把 Saga 事件存入事件库以获得完整审计轨迹与重放能力这与排障 7.3 中“saga_log表 可靠重放”的做法互为印证workflow-orchestration-patternsSKILL.mdTemporal、Conductor 等更高层工作流引擎建立在 Saga 概念之上其中“workflow 负责编排、activity 负责执行、活动必须幂等”的分层与本章节的实践准则相通。从仓库整体看这些技能共同归属backend-development插件该插件目录还包含architecture-patterns、microservices-patterns、temporal-python-testing等兄弟技能目录结构见 backend-development。当需要更系统的事件溯源读模型或云端持久工作流时即可从本技能平滑外延到上述模块。九、结语agents24仓库的saga-orchestration技能以“导航层 参考文献”的分层组织为 Agent 提供了一条从模式选型编排 vs 编舞、状态机建模Started → Completed / Failed、模板落地订单履约、编舞订阅、幂等守卫到生产加固分步超时、Prometheus 告警、DLQ 恢复、五种排障的完整路径。其核心方法论可概括为三句话正向步骤顺序前进失败后按严格逆序补偿所有参与方幂等、补偿永远发布结果事件每个步骤独立超时并以关联 ID 贯穿全程。无论你是要替换脆弱的 2PC还是为跨服务订单、预订与转账构建可靠流程这套模式都值得作为首选参考。【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考