转账
在两个账户之间转移资金，并确保下游操作失败时不会留下不完整的借记或贷记。
产品需求
转账请求包含源账户、目标账户、金额和备注。流程必须：
- 检查源账户余额是否足够。余额不足时，在产生任何副作用之前拒绝 Flow。
- 按顺序创建借记 memo、借记源账户、创建贷记 memo、贷记目标账户。
- 只有在目标账户贷记成功后才完成。
- 当带副作用的 Step 在重试耗尽后仍然失败时，对转账进行补偿。
正常路径不等待审批、Channel 消息或 RPC，只会等待重试间隔。Go、Java、Python 和 TypeScript 示例会对每个带副作用的 Step 重试最多一小时，并对补偿重试最多 24 小时。Rust 示例使用三次尝试，让已发布 SDK 的集成测试可以快速完成。
Rust 示例用正数金额模拟余额检查，并把账务副作用记录为事件。其他语言的示例会调用带有明确余额检查和撤销方法的内存依赖服务。
潜在故障分为三类：
- 余额不足属于业务拒绝。因为资金尚未移动，Flow 会立即失败。
- memo、借记和贷记调用可能发生瞬时故障。Dex 会按退避策略重试，并保留当前 Step 的状态。
- 操作在重试耗尽后仍可能失败。Flow 会进入补偿路径，撤销转账副作用，并以失败结果结束。补偿本身也会重试，因为补偿不完整同样是故障。
依赖服务调用必须具备幂等性。远端系统可能已经提交调用，但 Dex 尚未记录结果，此时 Worker 会再次执行调用。
Flow 设计
这个交互式 definition graph 由可运行的 Python Flow 生成。
这个 Flow 是一个顺序 saga：
CheckBalance → CreateDebitMemo → Debit → CreateCreditMemo → Credit
它组合了以下设计模式：
- 快速失败校验让预期内的业务拒绝不进入补偿路径。
- 持久重试在拥有该操作的 Step 内处理瞬时依赖故障。
- Execute Failure Recovery 在带副作用的 Step 重试耗尽后转到 Compensate。
- Saga 补偿执行反向操作，随后强制失败 Flow，防止调用方把已撤销的转账误判为成功。
每个副作用都使用独立 Step。这样可以清楚看到重试状态和执行历史，也能避免重试 Credit 时再次执行 Debit。
核心实现
下面的标签页展示完整的 Flow 声明、必需的 Flow 方法、拓扑、重试策略和补偿路径。每个链接指向可运行的完整示例，其中还包含请求模型、依赖服务、HTTP controller 和集成测试。
def compensated_step_options(
total_duration: timedelta,
) -> StepOptions:
return StepOptions(
execute_retry=RetryPolicy(total_duration=total_duration)
).on_execute_failure_proceed_to(
Compensate,
StepOptions(execute_retry=COMPENSATE_RETRY),
)
class MoneyTransferFlow(Flow[TransferRequest]):
def __init__(self, service: MyDependencyService) -> None:
self.service = service
self.compensate = Compensate(service)
options = compensated_step_options(timedelta(hours=1))
self.credit = Credit(service, options)
self.create_credit_memo = CreateCreditMemo(service, self.credit, options)
self.debit = Debit(service, self.create_credit_memo, options)
self.create_debit_memo = CreateDebitMemo(service, self.debit, options)
self.check_balance = CheckBalance(service, self.create_debit_memo)
def get_steps(self) -> StepList[TransferRequest]:
return StepList.start_step(self.check_balance).other_steps(
self.create_debit_memo,
self.debit,
self.create_credit_memo,
self.credit,
self.compensate,
)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of()
class Compensate(Step[TransferRequest]):
def execute(self, context: Context, input: TransferRequest) -> StepDecision:
self.service.undo_credit(input.to_account, input.amount)
self.service.undo_create_credit_memo(
input.to_account,
input.amount,
input.notes,
)
self.service.undo_create_debit_memo(
input.from_account,
input.amount,
input.notes,
)
self.service.undo_debit(input.from_account, input.amount)
return force_fail(
f"transfer has failed from {input.from_account} "
f"to {input.to_account} for amount {input.amount}"
)
例子: examples/python/dex_examples/products/money-transfer/money_transfer_flow.py
演示
启动 Dex 和任意语言的示例服务，然后通过 examples playground 启动 Money transfer。下面的截图展示了一笔金额为 42 个单位的成功转账。Flow 先从 checking-account 扣除 42，再向 savings-account 增加 42，并按顺序完成余额检查、创建借记 memo、借记、创建贷记 memo 和贷记五个操作。
