MassTransit 7.3升级至8.0.15后Saga定时事件手动触发失效

问题描述 #

将MassTransit从7.3版本升级至8.0.15版本后，Saga中配置的30秒调度任务（RestartConversion）无法通过手动发送RestartConversionSchedule消息触发，仅能等待30秒自动触发。此前7.3版本中该手动触发逻辑可正常执行When(RestartConversion.Received)分支。

Saga调度配置：

Schedule(() => RestartConversion, instance => instance.RestartConversionTokenId, s =>
{
    s.Delay = TimeSpan.FromSeconds(30);
    s.Received = r => r.CorrelateById(context => context.Message.ItemId);
});

原测试触发代码（升级后失效）：

var endPoint = await Harness.Bus.GetSendEndpoint(Harness.InputQueueAddress);
await endPoint.Send(new DocumentConversionSaga.RestartConversionSchedule
{
    ItemId = Guid.Parse(uploadDocumentInfo.ItemId),
    ConversionJobInstanceId = conversionInstanceId,
    PreviewJobInstanceId = previewInstanceId
});

原因分析 #

MassTransit 8对Saga调度消息的处理逻辑做了变更，直接发送RestartConversionSchedule这类内部调度类型的消息不再被Saga的调度接收器识别。调度任务的触发需要通过框架提供的调度器API来完成，而非手动发送原始调度消息类型。

解决方案 #

方法一：使用TestHarness的TestScheduler手动触发调度任务 #

在单元测试中，利用MassTransit测试套件提供的TestScheduler来直接触发指定的调度任务，无需等待延迟时间：

// 获取目标Saga实例
var sagaInstance = await Harness.Saga<DocumentConversionState>.GetInstance(Guid.Parse(uploadDocumentInfo.ItemId));

// 通过TestScheduler触发该调度任务
await Harness.TestScheduler.ScheduleJob(sagaInstance.RestartConversionTokenId);

方法二：发送调度绑定的事件类型（若适用） #

如果RestartConversion.Received分支绑定的是自定义事件类型，可直接发送该事件消息触发逻辑：

await Harness.Bus.Publish(new RestartConversionEvent
{
    ItemId = Guid.Parse(uploadDocumentInfo.ItemId),
    ConversionJobInstanceId = conversionInstanceId,
    PreviewJobInstanceId = previewInstanceId
});

配置验证 #

你的现有TestHarness配置已正确启用UseDelayedMessageScheduler，InMemoryTestHarness默认集成了TestScheduler，无需额外调整配置。

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