如何在MassTransit状态机中处理所有事件的Fault消息？

当前基于MassTransit的状态机可正常处理Requested、Started、ArchiveFinished事件，并通过复合事件Finished完成流程终止。但存在以下问题：

• 当上述任一事件的消费者处理出错时，系统会生成对应的Fault消息，但状态机无法捕获这些消息
• 需要在状态机中捕获所有事件的Fault消息，并发布FinishedFault消息终止流程
• 尝试使用Catch方法无效，自行添加Fault事件定义后仍未实现正确逻辑

初始状态机代码 #

public Event<Requested>? Requested { get; }
public Event<Started>? Started { get; }
public Event<ArchiveFinished>? ArchiveFinished { get; }
public Event? Finished { get; }

public Test() {
    InstanceState(x => x.CurrentState);

    Event(() => Requested, x => x.CorrelateById(context => context.Message.RequestId));
    Event(() => Started, x => x.CorrelateById(context => context.Message.RequestId));
    Event(() => ArchiveFinished, x => x.CorrelateById(context => context.Message.RequestId));

    Initially(
        When(Requested),
        When(Started),
        When(ArchiveFinished)
    );

    CompositeEvent(() => Finished,
        x => x.ReadyEventStatus,
        CompositeEventOptions.IncludeInitial,
        Requested,
        Started,
        ArchiveFinished
    );

    DuringAny(
        When(Finished)
            .Publish(context => new Finished() { RequestId = context.Instance.RequestId })
            .Finalize()
    );

    SetCompletedWhenFinalized();
}

尝试的修改代码 #

public Event<Requested>? Requested { get; }
public Event<Fault<Requested>>? RequestedFault { get; }
public Event<Started>? Started { get; }
public Event<Fault<Started>>? StartedFault { get; }
public Event<ArchiveFinished>? ArchiveFinished { get; }
public Event<Fault<ArchiveFinished>>? ArchiveFinishedFault { get; }
public Event? Finished { get; }

public Test() {
    InstanceState(x => x.CurrentState);

    Event(() => Requested, x => x.CorrelateById(context => context.Message.RequestId));
    Event(() => RequestedFault, x => x.CorrelateById(context => context.Message.Message.RequestId));
    Event(() => Started, x => x.CorrelateById(context => context.Message.RequestId));
    Event(() => StartedFault, x => x.CorrelateById(context => context.Message.Message.RequestId));
    Event(() => ArchiveFinished, x => x.CorrelateById(context => context.Message.RequestId));
    Event(() => ArchiveFinishedFault, x => x.CorrelateById(context => context.Message.Message.RequestId));

    Initially(
        When(Requested),
        When(RequestedFault)
            .Publish(context => new FinishedFault { RequestId = context.Message.Message.RequestId, ErrorMessage = context.Message.Exceptions.First().Message }),
        When(Started),
        When(StartedFault)
            .Publish(context => new FinishedFault { RequestId = context.Message.Message.RequestId, ErrorMessage = context.Message.Exceptions.First().Message }),
        When(ArchiveFinished),
        When(ArchiveFinishedFault)
            .Publish(context => new FinishedFault { RequestId = context.Message.Message.RequestId, ErrorMessage = context.Message.Exceptions.First().Message })
    );

    CompositeEvent(() => Finished,
        x => x.ReadyEventStatus,
        CompositeEventOptions.IncludeInitial,
        Requested,
        Started,
        ArchiveFinished
    );

    DuringAny(
        When(Finished)
            .Publish(context => new Finished() { RequestId = context.Instance.RequestId })
            .Finalize()
    );

    SetCompletedWhenFinalized();
}

正确实现方案 #

核心改进点 #

1. 统一Fault事件处理范围：使用DuringAny确保状态机在任何状态下都能捕获Fault消息
2. 简化重复代码：提取通用Fault处理逻辑，避免冗余
3. 确保关联正确性：严格从Fault<T>的嵌套消息中获取关联ID

最终实现代码 #

public Event<Requested>? Requested { get; }
public Event<Fault<Requested>>? RequestedFault { get; }
public Event<Started>? Started { get; }
public Event<Fault<Started>>? StartedFault { get; }
public Event<ArchiveFinished>? ArchiveFinished { get; }
public Event<Fault<ArchiveFinished>>? ArchiveFinishedFault { get; }
public Event? Finished { get; }

public Test() {
    InstanceState(x => x.CurrentState);

    // 配置正常事件的关联规则
    Event(() => Requested, x => x.CorrelateById(c => c.Message.RequestId));
    Event(() => Started, x => x.CorrelateById(c => c.Message.RequestId));
    Event(() => ArchiveFinished, x => x.CorrelateById(c => c.Message.RequestId));

    // 配置Fault事件的关联规则：从嵌套的原始消息中获取RequestId
    Event(() => RequestedFault, x => x.CorrelateById(c => c.Message.Message.RequestId));
    Event(() => StartedFault, x => x.CorrelateById(c => c.Message.Message.RequestId));
    Event(() => ArchiveFinishedFault, x => x.CorrelateById(c => c.Message.Message.RequestId));

    // 初始状态仅处理正常事件
    Initially(
        When(Requested),
        When(Started),
        When(ArchiveFinished)
    );

    // 复合事件配置：三个正常事件全部触发后生成Finished事件
    CompositeEvent(() => Finished,
        x => x.ReadyEventStatus,
        CompositeEventOptions.IncludeInitial,
        Requested,
        Started,
        ArchiveFinished
    );

    // 统一处理所有完成/错误场景
    DuringAny(
        // 正常完成流程
        When(Finished)
            .Publish(c => new Finished { RequestId = c.Instance.RequestId })
            .Finalize(),
        // 处理Requested事件的错误
        HandleFault(RequestedFault),
        // 处理Started事件的错误
        HandleFault(StartedFault),
        // 处理ArchiveFinished事件的错误
        HandleFault(ArchiveFinishedFault)
    );

    SetCompletedWhenFinalized();
}

// 通用Fault处理方法，复用逻辑
private EventActivityBinder<TestState> HandleFault<T>(Event<Fault<T>> faultEvent) where T : class
{
    return When(faultEvent)
        .Publish(c => new FinishedFault { 
            RequestId = c.Message.Message.RequestId, 
            ErrorDetails = c.Message.Exceptions.Select(e => $"{e.Message}\n{e.StackTrace}").ToList() 
        })
        .Finalize();
}

关键说明 #

• 关联ID正确性：Fault<T>消息的Message属性是原始事件对象，必须通过context.Message.Message.RequestId获取关联ID，才能匹配到对应的状态机实例
• 全局状态覆盖：DuringAny替代仅在Initially中处理Fault，确保即使状态机已处理部分事件，后续的Fault消息仍能被捕获并终止流程
• 资源清理：每个Fault处理后调用Finalize()，结合SetCompletedWhenFinalized()确保状态机实例被标记为完成，避免资源泄漏

内容的提问来源于stack exchange，提问作者Lubos Hnedy