| title | Saga |
|---|---|
| description | 用无状态 Saga 把已提交事件映射为 0..N 条跨聚合命令，并明确上下文、业务补偿、等待与失败边界。 |
| outline | deep |
Wow 的 @StatelessSaga 是无状态事件编排器：它接收领域事件或状态事件并生成下一步命令。每个目标聚合仍在自己的本地事务中处理命令；Saga 不创建跨聚合 ACID 事务。
Stateless Saga 把一个已发生的事实转换为 0..N 条顺序发送的后续命令。
flowchart TB
Source["源聚合"] -->|领域事件| EventBus["DomainEventBus"]
EventBus -->|分发到匹配 Saga| Notifier["SagaHandledNotifierFilter"]
Notifier -->|调用 Saga 函数| Saga["Stateless Saga"]
Saga --> Commands{"没有后续命令？"}
Commands -->|是| Done["记录 commandIds"]
Commands -->|否：顺序发送| Gateway["CommandGateway"]
Gateway --> CommandBus["CommandBus"]
CommandBus --> Sent["发送完成；目标聚合处理未保证"]
Sent --> Done
Done --> Signal["SAGA_HANDLED + commandIds"]
Signal --> WaitNotifier["CommandWaitNotifier"]
当一个已提交事件需要驱动其他聚合的业务行为时使用 Saga，例如转账准备完成后向目标账户发送入账命令。只需通知、审计或调用外部集成时使用事件处理器；不要用 Saga 包装一个不生成命令的普通副作用。
无状态 Saga 不保存流程实例。每个事件函数必须仅依赖当前事件及可注入上下文，明确返回本次需要发送的 0..N 条命令。
已提交事件 -> 匹配 Saga 函数 -> 0..N 条命令 -> CommandGateway.send
Saga 与普通 Processor 共用事件函数解析、类型/topic 匹配、领域事件/状态事件分发和响应式 filter 链。区别在于 StatelessSagaFunction 会把函数结果转换为命令、按顺序发送，并把命令流记录到 exchange 供 SAGA_HANDLED 通知使用。
源事件在 Saga 运行前已经提交。函数或命令发送失败都不能回滚源事件。
@StatelessSaga 同时是 Spring 组件标记。函数可使用 onEvent / onStateEvent 约定名，或显式使用 @OnEvent / @OnStateEvent；第一个参数也可以是事件体、DomainEvent<T> 或 DomainEventExchange<T>。
@StatelessSaga
class CartSaga {
@OnEvent
fun onOrderCreated(event: DomainEvent<OrderCreated>): CommandBuilder? {
if (!event.body.fromCart) return null
return RemoveCartItem(
productIds = event.body.items.map { it.productId }.toSet(),
).commandBuilder()
.aggregateId(event.ownerId)
}
}只有需要聚合最新状态时才使用 @OnStateEvent 并把状态声明为后续注入参数。需要显式选择目标聚合 ID、请求 ID 或其他命令字段时返回 CommandBuilder。
函数可以同步、挂起或响应式返回以下结果：
| 返回结果 | 发送行为 |
|---|---|
null / Mono.empty() |
发送 0 条命令 |
| 命令体 | 转换为 CommandMessage 后发送 1 条 |
CommandBuilder |
按 builder 的目标与显式字段创建并发送 1 条 |
CommandMessage<*> |
保留现有消息并发送 1 条 |
Iterable<*>、Flux、Publisher 或 Flow |
收集并按结果顺序发送 N 条 |
StatelessSagaFunction 使用 concatMap 逐条创建并发送命令：前一条 CommandGateway.send 完成后，才开始创建下一条。保持返回顺序稳定；顺序变化不仅改变业务流程，也会改变默认 request ID。
对命令体或 CommandBuilder，框架只补充缺失字段：
- 默认
requestId为${domainEvent.id}-${index}，索引从0开始； - 保留显式
requestId； - 缺失的
tenantId、spaceId从源事件传播； - 设置源事件为 upstream，并传播消息 header。
预构造的 CommandMessage 保留自己的消息与 requestId，同时传播源事件 header。
同一事件以相同顺序重投时，默认 request ID 保持稳定，可与命令网关的幂等检查协作。它不保证外部副作用幂等，也不能保护不同事件生成的语义重复命令。
即时重试会重新执行 Saga 函数并再次发送返回的命令。新创建的命令消息继续使用原有全局唯一 commandId 生成规则，默认 requestId 保持稳定以配合幂等检查。单条发送返回 DuplicateRequestIdException 时，Saga 跳过该次重复发送并继续后续命令；其他错误仍向上传播。Saga 不缓存发送进度。
Saga 的业务补偿是一个明确的领域动作。例如 EntryFailed 可以生成 UnlockAmount：
@OnEvent
fun onEntryFailed(event: EntryFailed): UnlockAmount =
UnlockAmount(event.sourceId, event.amount)这条命令表达业务上如何抵消先前效果。它不会删除 Prepared、EntryFailed 等已提交事件，也不是数据库回滚。
不要把业务补偿与处理失败恢复混为一谈。Saga 决定“失败事实之后应发什么业务命令”；Compensation 决定“同一处理函数失败后如何持久记录并重投”。后者的权威说明见事件补偿。
Saga 函数完成并且生成命令的 CommandGateway.send 全部完成后，运行时产生 SAGA_HANDLED。该信号包含本次命令流的 commandId，只证明发送流程已完成（包括跳过重复请求），不证明目标聚合已经成功处理命令。
调用方只关心 Saga 已发出命令时等待匹配的 SAGA_HANDLED。若还必须等待每条后续命令的某个阶段，使用 CommandWait.chain(...) 指定 Saga 函数与 tail stage/function。等待状态位于进程内，受调用方超时和取消限制；Saga 重试不保证原等待仍然有效或能够恢复。完整阶段、函数匹配和提前到达信号处理见完成语义。
使用 SagaSpec 直接验证事件到命令的映射，不需要启动消息代理：
class CartSagaSpec : SagaSpec<CartSaga>({
on {
whenEvent(orderCreatedFromCart, ownerId = ownerId) {
expectNoError()
expectCommandType(RemoveCartItem::class)
expectCommand<RemoveCartItem> {
aggregateId.id.assert().isEqualTo(ownerId)
}
}
}
on {
whenEvent(orderCreatedNotFromCart, ownerId = ownerId) {
expectNoCommand()
}
}
})至少覆盖正常命令、0 命令分支和每条业务补偿分支。依赖默认 request ID、上下文传播、多命令顺序或预构造消息时，增加针对完整 CommandMessage 的断言；依赖真实发送、链式等待或失败恢复时，再增加集成测试。
Saga 函数错误或 DuplicateRequestIdException 以外的 CommandGateway.send 错误都会沿响应式链传播，随后才由即时重试或已启用的持久 Compensation 处理。即使失败发生在部分命令已被总线接受之后，源事件和已接受命令也不会自动撤销；因此命令处理与业务补偿都必须可安全重投。
完成标志： 0..N 映射、命令顺序、request ID/上下文传播和业务补偿分支均有测试；等待合同没有把 SAGA_HANDLED 误写成后续命令完成。