如何在 Saga 中处理补偿逻辑

2026-10-24支付阿杰

结论

这个设计的整体方向没有问题：用 Saga 协调注册流程，把支付和 ExternalXYZ 调用拆成独立的消息处理步骤。不过，有两点需要先说清楚：

Saga 能保证消息与本地持久化状态的一致性，但不能把数据库、银行卡扣款和第三方 HTTP 服务纳入同一个可回滚事务。
“请求超时”不代表”支付失败”。防止重复扣款的重点不是禁止重试，而是让每次重试保持幂等，并在支付结果不确定时主动查询状态。

DoPayment 失败后，此前已经完成的 ProcessRegistration Saga 事务不会回滚。通常只有当前 DoPayment 消息处理范围内，且确实加入当前事务的本地数据库操作和消息发送会回滚。已经发生的外部扣款，NServiceBus 无法自动撤销。

建议的流程设计

Saga 更适合充当流程协调器，不要在一个 Handler 中依次同步完成所有操作：

BeginRegistration
        |
        v
RegistrationSaga
        |
        +-- Send ChargePayment
        |
        v
Payment Handler ----> Payment Service
        |
        +-- PaymentSucceeded
        +-- PaymentFailed
        +-- PaymentStatusUnknown
        |
        v
RegistrationSaga
        |
        +-- Send RegisterWithExternalXYZ
        |
        v
ExternalXYZ Handler
        |
        +-- ExternalXYZSucceeded
        +-- ExternalXYZFailed
        |
        v
RegistrationSaga

Saga 应保存流程所需的业务状态，例如：

RegistrationId
PaymentOperationId
PaymentStatus
ExternalXYZOperationId
ExternalXYZStatus
已经发起或完成了哪些补偿操作

不要让 Saga 依靠一张日志表推断当前业务状态。日志适合审计和排障，决定下一步动作的状态则应明确保存在 Saga 数据或专门的业务状态表中。

如何避免超时后重复扣款

支付请求必须携带稳定且唯一的幂等键，例如：

PaymentOperationId = RegistrationId + ":initial-payment"

无论 NServiceBus 重试多少次，都要使用同一个 PaymentOperationId，不能在 Handler 每次执行时重新生成。

支付服务需要支持类似下面的语义：

POST /payments
Idempotency-Key: registration-8f7c:initial-payment

同一个幂等键被重复提交时，支付服务应返回首次操作的结果，而不是再次扣款。

如果第三方支付服务不支持幂等键，可靠性会明显下降。此时可以采用以下次优方案：

为支付请求生成唯一的 merchant reference。
调用前持久化该 reference。
超时后通过 reference 查询支付状态。
只有明确查到”未受理”时才重新提交。
如果无法查询，就进入人工处理或延迟轮询状态，不要直接再次扣款。

支付结果至少要区分以下状态：

public enum PaymentStatus
{
    NotStarted,
    Pending,
    Succeeded,
    Failed,
    StatusUnknown,
    RefundPending,
    Refunded
}

超时通常应映射为 StatusUnknown，而不是 Failed。

推荐的支付 Handler 写法

下面是一个简化示例。NServiceBus 的具体 API 会因版本和持久化组件而有所不同：

public sealed class ChargePaymentHandler :
    IHandleMessages<ChargePayment>
{
    private readonly IPaymentClient paymentClient;
    private readonly IPaymentOperationStore operationStore;

    public ChargePaymentHandler(
        IPaymentClient paymentClient,
        IPaymentOperationStore operationStore)
    {
        this.paymentClient = paymentClient;
        this.operationStore = operationStore;
    }

    public async Task Handle(
        ChargePayment message,
        IMessageHandlerContext context)
    {
        var operation = await operationStore.Find(
            message.PaymentOperationId,
            context.CancellationToken);

        if (operation?.Status == PaymentStatus.Succeeded)
        {
            await context.SendLocal(new PaymentSucceeded
            {
                RegistrationId = message.RegistrationId,
                PaymentOperationId = message.PaymentOperationId,
                ProviderPaymentId = operation.ProviderPaymentId
            });

            return;
        }

        await operationStore.EnsureCreated(
            message.PaymentOperationId,
            message.RegistrationId,
            context.CancellationToken);

        try
        {
            var response = await paymentClient.Charge(
                amount: message.Amount,
                idempotencyKey: message.PaymentOperationId,
                cancellationToken: context.CancellationToken);

            await operationStore.MarkSucceeded(
                message.PaymentOperationId,
                response.ProviderPaymentId,
                context.CancellationToken);

            await context.SendLocal(new PaymentSucceeded
            {
                RegistrationId = message.RegistrationId,
                PaymentOperationId = message.PaymentOperationId,
                ProviderPaymentId = response.ProviderPaymentId
            });
        }
        catch (PaymentDeclinedException ex)
        {
            await operationStore.MarkFailed(
                message.PaymentOperationId,
                ex.Message,
                context.CancellationToken);

            await context.SendLocal(new PaymentFailed
            {
                RegistrationId = message.RegistrationId,
                PaymentOperationId = message.PaymentOperationId,
                Reason = ex.Message
            });
        }
        catch (TimeoutException)
        {
            await context.SendLocal(new VerifyPaymentStatus
            {
                RegistrationId = message.RegistrationId,
                PaymentOperationId = message.PaymentOperationId
            });

            // 结果未知，不应立即创建一个新的支付请求。
        }
    }
}

实际实现还要处理”数据库已经更新，但成功消息未发送”的双写问题。通常可以结合 NServiceBus Outbox，或者确保业务数据库操作与 NServiceBus 持久化加入同一个本地事务。

超时后的状态确认

查询状态时，仍要使用原来的支付操作标识：

public sealed class VerifyPaymentStatusHandler :
    IHandleMessages<VerifyPaymentStatus>
{
    private readonly IPaymentClient paymentClient;

    public VerifyPaymentStatusHandler(IPaymentClient paymentClient)
    {
        this.paymentClient = paymentClient;
    }

    public async Task Handle(
        VerifyPaymentStatus message,
        IMessageHandlerContext context)
    {
        var result = await paymentClient.FindByIdempotencyKey(
            message.PaymentOperationId,
            context.CancellationToken);

        switch (result.Status)
        {
            case ProviderPaymentStatus.Succeeded:
                await context.SendLocal(new PaymentSucceeded
                {
                    RegistrationId = message.RegistrationId,
                    PaymentOperationId = message.PaymentOperationId,
                    ProviderPaymentId = result.ProviderPaymentId
                });
                break;

            case ProviderPaymentStatus.Failed:
            case ProviderPaymentStatus.NotFound:
                await context.SendLocal(new PaymentFailed
                {
                    RegistrationId = message.RegistrationId,
                    PaymentOperationId = message.PaymentOperationId,
                    Reason = result.Status.ToString()
                });
                break;

            default:
                // 可通过 Saga timeout 或延迟投递再次查询。
                await context.SendLocal(new VerifyPaymentStatus
                {
                    RegistrationId = message.RegistrationId,
                    PaymentOperationId = message.PaymentOperationId
                });
                break;
        }
    }
}

轮询需要设置最大持续时间和退避策略，并保留人工处理出口，避免无限发送消息。

`DoPayment` 失败时会回滚什么

ProcessRegistration 和 DoPayment 处理的是两条不同消息，分别属于两个独立的消息处理事务。

假设流程如下：

ProcessRegistration 更新 Saga 状态。
ProcessRegistration 发送 DoPayment。
当前消息处理成功并提交。
DoPaymentHandler 开始处理。
DoPaymentHandler 抛出异常。

第 5 步不会回滚第 1 步，因为 ProcessRegistration 的事务已经提交。

至于 DoPaymentHandler 内部的数据库操作是否回滚，要看这些操作是否真正加入了当前事务：

使用兼容的事务集成并加入 NServiceBus 当前事务的操作，通常会在 Handler 失败时回滚。
使用独立连接、独立事务或不支持事务的数据存储，不会自动回滚。
已经提交的独立事务不会回滚。
HTTP、SOAP 和支付网关调用通常不属于数据库事务，因此无法自动回滚。
Handler 产生的外部副作用，即使随后抛出异常，也可能已经生效。

不能仅凭代码写在同一个 Handler 或 TransactionScope 中，就认为第三方服务调用也受事务保护。

调用前后的日志应该怎样处理

“调用前写日志，调用后更新日志”有一个容易被忽视的问题：

写入调用前日志
调用支付服务
抛出异常
本地事务回滚

如果日志写入加入了当前事务，发生异常后，日志也会一起回滚，最后看不到这次调用记录。如果日志使用独立事务，又会出现另一种不一致：日志写入成功，并不能证明外部调用确实发生了。

所以，日志只能作为审计证据，不能单独充当业务真相。更稳妥的做法是维护一条带有唯一约束的操作记录：

CREATE UNIQUE INDEX UX_PaymentOperation_OperationId
ON PaymentOperation(OperationId);

记录至少应包含：

OperationId
RegistrationId
Status
ProviderPaymentId
AttemptCount
LastAttemptAt
LastError
CreatedAt
UpdatedAt

OperationId 必须保持稳定，StatusUnknown 也要作为正式的业务状态处理。

补偿应该如何实现

补偿不是数据库事务中的 rollback，而是一个新的业务操作。例如：

支付成功，但后续注册失败：发送退款请求。
外部账号已经创建，但本地注册失败：调用外部服务禁用或删除账号。
外部操作不可逆：转入人工处理。

补偿可以由原来的 Registration Saga 协调，也可以交给单独的 Compensation Saga。流程较短时，把补偿放在原 Saga 中通常更容易保持状态清晰。只有当补偿本身包含多个步骤、需要长时间重试或有独立生命周期时，才有必要拆成单独的 Saga。

例如：

public async Task Handle(
    ExternalXYZFailed message,
    IMessageHandlerContext context)
{
    Data.ExternalXYZStatus = "Failed";

    if (Data.PaymentStatus == "Succeeded" &&
        Data.RefundStatus == "NotStarted")
    {
        Data.RefundStatus = "Requested";

        await context.Send(new RefundPayment
        {
            RegistrationId = Data.RegistrationId,
            PaymentOperationId = Data.PaymentOperationId,
            RefundOperationId = $"{Data.PaymentOperationId}:refund"
        });
    }
}

退款操作本身也必须幂等：

public sealed class RefundPaymentHandler :
    IHandleMessages<RefundPayment>
{
    private readonly IPaymentClient paymentClient;

    public RefundPaymentHandler(IPaymentClient paymentClient)
    {
        this.paymentClient = paymentClient;
    }

    public async Task Handle(
        RefundPayment message,
        IMessageHandlerContext context)
    {
        var result = await paymentClient.Refund(
            paymentOperationId: message.PaymentOperationId,
            idempotencyKey: message.RefundOperationId,
            cancellationToken: context.CancellationToken);

        await context.SendLocal(new PaymentRefunded
        {
            RegistrationId = message.RegistrationId,
            RefundOperationId = message.RefundOperationId,
            ProviderRefundId = result.ProviderRefundId
        });
    }
}

不要扫描日志表来猜测是否需要补偿。Saga 应根据已经确认的业务状态发送明确的补偿命令。日志表可以用于核对、告警和人工恢复，但不能成为唯一的状态机。

还需要注意的边界

NServiceBus 的重试机制意味着 Handler 可能执行多次。所有会产生外部副作用的 Handler，都应按照”至少一次执行”来设计。
Saga correlation id 处理的是消息归属问题，并不等同于支付幂等键。两者可以有关联，但职责不同。
Saga 数据与业务数据不一定天然处于同一个事务中，具体取决于 transport、persistence、Outbox 和数据库配置。
不建议让 Saga Handler 长时间等待第三方 HTTP 调用。更合适的设计是发送命令，由专门的 endpoint 调用第三方服务，再通过事件返回结果。
银行卡拒付这类明确的业务失败，不应依赖异常来触发 NServiceBus 重试。
网络故障、服务不可用等技术失败可以重试，但操作必须幂等。
支付结果未知时应先查询状态，不能把超时直接视为”可以重新扣款”。
补偿操作也可能失败，因此退款、撤销和删除等操作同样需要状态、幂等键、重试及人工处理机制。