MassTransit状态机Faulted处理器未触发问题咨询

问题描述 #

我正在使用MassTransit配置状态机，发现当Activity中抛出异常时，Saga管道并未触发Faulted处理器。
为便于测试和说明，我采用了MassTransit示例库，但遇到了相同问题。
修改后的NotifyMemberActivity代码如下：

public class NotifyMemberActivity :
        IStateMachineActivity<CheckOut>
    {
        readonly IMemberRegistry _memberRegistry;

        public NotifyMemberActivity(IMemberRegistry memberRegistry)
        {
            _memberRegistry = memberRegistry;
        }

        public void Probe(ProbeContext context)
        {
            context.CreateScope("notifyMember");
        }

        public void Accept(StateMachineVisitor visitor)
        {
            visitor.Visit(this);
        }

        public async Task Execute(BehaviorContext<CheckOut> context, IBehavior<CheckOut> next)
        {
            await Execute(context);
            throw new NotImplementedException();

            await next.Execute(context);
        }

        public async Task Execute<T>(BehaviorContext<CheckOut, T> context, IBehavior<CheckOut, T> next)
            where T : class
        {
            await Execute(context);
            throw new NotImplementedException();

            await next.Execute(context);
        }

        public Task Faulted<TException>(BehaviorExceptionContext<CheckOut, TException> context, IBehavior<CheckOut> next)
            where TException : Exception
        {
            return next.Faulted(context);
        }

        public Task Faulted<T, TException>(BehaviorExceptionContext<CheckOut, T, TException> context, IBehavior<CheckOut, T> next)
            where TException : Exception
            where T : class
        {
            return next.Faulted(context);
        }

        async Task Execute(BehaviorContext<CheckOut> context)
        {
            var isValid = await _memberRegistry.IsMemberValid(context.Saga.MemberId);
            if (!isValid)
                throw new InvalidOperationException("Invalid memberId");

            throw new NotImplementedException();

            var consumeContext = context.GetPayload<ConsumeContext>();

            await consumeContext.Publish<NotifyMemberDueDate>(new
            {
                context.Saga.MemberId,
                context.Saga.DueDate
            });
        }
    }

问题原因与解决方案 #

核心误解 #

Activity自身的Faulted方法并非用来捕获当前Activity执行抛出的异常，它的职责是处理后续行为链中抛出的异常，并将异常传递给下一个行为的Faulted处理器。这是你遇到问题的核心原因。

正确处理方式 #

1. 在Activity内部捕获异常处理

直接在Execute方法中通过try/catch捕获异常，执行补偿逻辑，示例如下：

public async Task Execute(BehaviorContext<CheckOut> context, IBehavior<CheckOut> next)
{
    try
    {
        await Execute(context);
        await next.Execute(context);
    }
    catch (Exception ex)
    {
        // 执行故障补偿逻辑，比如回滚操作、通知等
        await HandleCompensation(context, ex);
        
        // 若需要让异常继续向上传递（触发Saga全局故障处理），可重新抛出
        throw;
    }
}

// 新增补偿处理方法
async Task HandleCompensation(BehaviorContext<CheckOut> context, Exception ex)
{
    // 这里编写你的补偿逻辑，比如发布补偿事件、更新Saga状态等
}

2. 配置状态机的全局/状态级Fault事件处理

在状态机定义中，针对特定Activity或全局配置Fault事件，当Activity抛出异常时，MassTransit会自动触发对应的Fault事件，你可以在事件处理中执行补偿：

public class CheckOutStateMachine : MassTransitStateMachine<CheckOut>
{
    public CheckOutStateMachine()
    {
        InstanceState(x => x.CurrentState);

        // 定义状态和事件
        public State CheckingOut { get; private set; }
        public State FailedState { get; private set; }
        public Event<Fault<CheckOutBook>> FaultedCheckOut { get; private set; }
        public Event<Fault<NotifyMemberActivity>> FaultedNotifyMember { get; private set; }

        // 针对NotifyMemberActivity的异常处理
        During(CheckingOut,
            When(CheckOutBook)
                .Activity(x => x.OfType<NotifyMemberActivity>()),
            When(FaultedNotifyMember)
                .Then(context =>
                {
                    // 执行针对该Activity异常的补偿逻辑
                    var exception = context.Data.Exceptions.First();
                    // 比如更新Saga状态、记录错误信息等
                })
                .TransitionTo(FailedState));
    }
}

额外说明 #

你代码中Execute方法抛出异常后未调用await next.Execute(context)，但这不是Faulted不触发的原因——核心问题还是对Activity的Faulted方法职责范围的误解。

内容的提问来源于stack exchange，提问作者Paweł Mrówczyński