MassTransit Routing Slip 异常为何不触发 Compensate

结论

Send() 只能保证消息已经成功交给消息代理，不会等待目标 consumer 执行完毕。只要 TakeProductActivity.Execute 返回 context.Completed(...)，Routing Slip 就会认为该 activity 已成功完成。此后，即使另一个进程中的 consumer 抛出异常，也只会被视为独立消息的消费故障，无法改变已经完成的 activity 状态，自然也不会触发 Routing Slip 补偿。

要让异常进入 Routing Slip 的 fault 和 compensation 流程，异常必须发生在当前 activity 的执行范围内，或者通过 request/reply 返回给 activity，使 Execute 以失败结束。

为什么 Send() 无法触发补偿

下面的代码等待的只是消息发送完成：

await sendEndpoint.Send<TakeProductTransactionMessage>(new
{
    ProductBaskets = context.Arguments.Baskets
});

return context.Completed(new
{
    Baskets = context.Arguments.Baskets,
    OrderId = context.Arguments.OrderId
});

这里有两个相互独立的时间点：

1. RabbitMQ 接收消息，Send() 返回。
2. TakeProductTransactionConsumer 随后开始消费消息。

如果异常发生在第二步，对应的 Routing Slip activity 早已返回 Completed。MassTransit 可以重试失败的消费、将消息移入 _error 队列，并产生 Fault<T>，但这个 fault 已经不在结束的 Routing Slip activity 调用链中。

问题不在于 RabbitMQ 无法“获取 Exception”，而在于 fire-and-forget 消息没有调用方等待执行结果，消费失败也没有绑定到 Routing Slip 当前的 activity。

方案一：把业务处理直接实现为 Routing Slip Activity

这是最符合 Routing Slip 工作方式的方案。Activity 可以部署在独立 micro-service 中，不必与 saga 或 Routing Slip 发起方运行在同一个进程。

可以把库存扣减逻辑放进远程 activity 的 Execute：

public class TakeProductActivity :
    IActivity<TakeProductArgument, TakeProductLog>
{
    private readonly IProductService _productService;
    private readonly IPublishEndpoint _publishEndpoint;

    public TakeProductActivity(
        IProductService productService,
        IPublishEndpoint publishEndpoint)
    {
        _productService = productService;
        _publishEndpoint = publishEndpoint;
    }

    public async Task<ExecutionResult> Execute(
        ExecuteContext<TakeProductArgument> context)
    {
        if (context.Arguments.Baskets.Count > 0)
            throw new Exception("Process Failed!");

        var productCounts = context.Arguments.Baskets.ToDictionary(
            x => x.ProductId,
            x => x.Count);

        var products = await _productService.TakeProducts(productCounts);

        await _publishEndpoint.Publish<ProductUpdated>(new
        {
            ProductUpdatedEvents = products.Select(p => new
            {
                ProductId = p.Id,
                p.Price,
                p.Count
            }).ToList()
        });

        return context.Completed<TakeProductLog>(new
        {
            context.Arguments.OrderId,
            Baskets = context.Arguments.Baskets
        });
    }

    public async Task<CompensationResult> Compensate(
        CompensateContext<TakeProductLog> context)
    {
        await _productService.RestoreProducts(context.Log.Baskets);
        return context.Compensated();
    }
}

该 activity 的执行端点仍可放在库存 micro-service 中：

cfg.AddActivityEndpoint<TakeProductActivity, TakeProductArgument, TakeProductLog>(
    provider,
    QueueNames.GetActivityUri(nameof(TakeProductActivity)));

具体的注册 API 可能随 MassTransit 版本变化，应以项目所用版本的 activity endpoint 注册方式为准。重点不在注册方法的名称，而在于 RoutingSlipBuilder.AddActivity 必须指向真正执行库存业务的 activity endpoint。

这样一来，只要 Execute 抛出异常，Routing Slip 就会进入 fault 流程，并按相反顺序补偿此前已经成功完成且生成了 compensation log 的 activities。

方案二：在 Activity 中使用 request/reply 等待远程结果

如果必须保留普通 consumer，可以让 activity 发送请求，并等待明确的成功响应：

public record TakeProductRequest(
    Guid OrderId,
    IReadOnlyList<ProductBasket> ProductBaskets);

public record TakeProductSucceeded(Guid OrderId);

Activity：

public async Task<ExecutionResult> Execute(
    ExecuteContext<TakeProductArgument> context)
{
    var response = await _requestClient.GetResponse<TakeProductSucceeded>(
        new TakeProductRequest(
            context.Arguments.OrderId,
            context.Arguments.Baskets),
        context.CancellationToken);

    return context.Completed<TakeProductLog>(new
    {
        context.Arguments.OrderId,
        Baskets = context.Arguments.Baskets
    });
}

Consumer 成功后返回响应：

public async Task Consume(
    ConsumeContext<TakeProductRequest> context)
{
    if (context.Message.ProductBaskets.Count > 0)
        throw new Exception("Process Failed!");

    var productCounts = context.Message.ProductBaskets.ToDictionary(
        x => x.ProductId,
        x => x.Count);

    await _productService.TakeProducts(productCounts);

    await context.RespondAsync(
        new TakeProductSucceeded(context.Message.OrderId));
}

Consumer 抛出异常后，MassTransit 的 request client 通常会收到 fault response，并抛出 RequestFaultException。只要 activity 没有吞掉这个异常，Execute 就不会返回 Completed，Routing Slip 随后会进入 fault 和 compensation 流程。

如果需要先记录日志，再继续向上抛出异常，可以这样处理：

try
{
    await _requestClient.GetResponse<TakeProductSucceeded>(
        request,
        context.CancellationToken);
}
catch (RequestFaultException exception)
{
    _logger.LogError(
        exception,
        "Take product failed for order {OrderId}",
        context.Arguments.OrderId);

    throw;
}

不要在 catch 中继续返回 context.Completed(...)，否则 Routing Slip 仍会将该 activity 判定为成功。

方案三：保留 fire-and-forget，由 Saga 显式处理失败

如果调用必须采用异步 fire-and-forget，就不能再依赖 Routing Slip 自动补偿，而应将其设计为明确的业务流程：

• 命令携带 OrderId、CorrelationId 或 Routing Slip TrackingNumber。
• consumer 成功后发布 TakeProductSucceeded。
• consumer 最终失败后发布 TakeProductFailed，也可以由专门的 fault consumer 处理 Fault<TakeProductTransactionMessage>。
• saga 等待成功或失败事件。
• saga 收到失败事件后，显式发送退款、恢复库存等补偿命令。

例如监听 fault：

public class TakeProductFaultConsumer :
    IConsumer<Fault<TakeProductTransactionMessage>>
{
    public async Task Consume(
        ConsumeContext<Fault<TakeProductTransactionMessage>> context)
    {
        await context.Publish<TakeProductFailed>(new
        {
            context.Message.Message.OrderId,
            Reason = context.Message.Exceptions.FirstOrDefault()?.Message
        });
    }
}

Saga 再通过 OrderId 关联 TakeProductFailed，进入失败状态并执行相应的恢复操作。

这种做法属于 saga 编排的业务补偿，并不是让一个已经完成的 Routing Slip 自动“回到过去”触发 compensation。也不要手工伪造 Routing Slip fault 事件来推动其内部状态，否则会绕过 Routing Slip 自己维护的 activity log 和执行状态。

为什么 Payment 自身失败时不会调用自己的 Compensate

Routing Slip 只会补偿已经成功完成并生成 compensation log 的 activities。

假设执行顺序如下：

SubmitOrder -> Payment -> TakeProduct

对应行为是：

• Payment.Execute 失败时，只补偿已经完成的 SubmitOrder，不会调用 Payment.Compensate。
• Payment.Execute 成功，而 TakeProduct.Execute 失败时，先调用 Payment.Compensate，再补偿更早完成的 activity。
• 如果 Payment 只是通过 Send() 发送一条消息，然后立即返回 Completed，下游 Payment consumer 随后发生的异常不会自动导致 Routing Slip 失败。

这是 Routing Slip 所需的语义。Payment.Execute 如果没有成功结束，Routing Slip 通常无法获得可靠的 compensation log，也不能确认支付产生了哪些副作用。

如果支付操作可能出现“扣款成功但响应发送失败”这类不确定状态，需要使用幂等键并查询支付状态，不能只根据是否抛出异常来判断结果。

设计时需要注意

• Activity 和 compensation 操作都应支持幂等，因为消息重试或网络故障可能造成重复执行。
• 使用稳定的业务幂等键，例如 OrderId 或独立的 PaymentTransactionId。
• request/reply 需要设置合理的超时时间。超时只表示结果未知，不代表远程操作一定没有执行。
• consumer 的消息重试会延迟 fault response。Routing Slip 只有在重试最终失败后，才会收到请求失败的结果。
• Fault<T> 表示某条消息消费失败，不会自动关联并回滚任意 Routing Slip。
• Compensation 不是数据库事务回滚，而是一次新的消息驱动业务操作。它同样可能失败，因此需要配套监控和人工修复机制。
• 如果某项远程操作会影响 Routing Slip 对成败的判断，优先将其实现为远程 Routing Slip Activity。如果业务需要长期异步等待，通常更适合由 saga 显式编排成功、失败和补偿事件。

备注：内容仅供参考。