MassTransit：如何清理已取消状态的JobSaga条目

问题分析 #

发布CancelJob消息后，作业进入Canceled(9)状态，后续收到JobSlotWaitElapsed事件时触发状态机异常，且现有操作（发布CompleteJob、等待超时）无法将这类Saga记录从表中移除。核心原因是Canceled状态的Saga未被正确纳入清理流程，且默认状态机不处理该状态下的JobSlotWaitElapsed事件。

解决方法 #

1. 手动触发Saga终结逻辑 #

• 通过Saga仓库定位目标实例，调用终结方法：
  若使用EF Core作为Saga存储，可直接查询并更新记录：

  using var scope = serviceProvider.CreateScope();
  var sagaRepo = scope.ServiceProvider.GetRequiredService<ISagaRepository<JobSaga>>();
  
  var saga = await sagaRepo.Find(context, jobId);
  if (saga != null && saga.CurrentState == JobState.Canceled)
  {
      saga.CurrentState = JobState.Completed;
      saga.CompletedTimestamp = DateTime.UtcNow;
      await sagaRepo.Update(context, saga);
  }
  

• 也可发布匹配的JobCompleted消息，确保消息的JobId与目标作业一致，且Contract版本与JobService使用的版本兼容。

2. 配置Saga自动清理策略 #

• 在JobService配置中明确启用已取消/失败作业的清理：

  services.AddMassTransit(cfg =>
  {
      cfg.SetKebabCaseEndpointNameFormatter();
  
      cfg.AddJobService(options =>
      {
          // 清理已完成作业，保留24小时
          options.CleanupCompletedJobs(TimeSpan.FromHours(24));
          // 清理已取消/失败作业，保留12小时
          options.CleanupFaultedJobs(TimeSpan.FromHours(12));
      });
  });
  

• 确保MassTransit的后台清理服务正常运行，该服务默认随JobService启动，若手动托管需确认服务已注册。

3. 扩展JobSaga状态机处理异常事件 #

• 自定义JobSaga状态机，添加Canceled状态下对JobSlotWaitElapsed事件的处理，直接终结Saga：

  public class CustomJobSagaStateMachine : JobSagaStateMachine
  {
      public CustomJobSagaStateMachine()
      {
          When(Events.JobSlotWaitElapsed, InState(States.Canceled), x => x.Finalize());
      }
  }
  

• 注册自定义状态机替代默认实现：

  services.AddMassTransit(cfg =>
  {
      cfg.AddSagaStateMachine<CustomJobSagaStateMachine, JobSaga>()
          .EntityFrameworkRepository(r =>
          {
              r.ExistingDbContext<JobSagaDbContext>();
              r.UseSqlServer();
          });
  });
  

4. 数据库层面手动清理（应急方案） #

• 针对SQL Server数据库，可执行以下语句更新或删除Canceled状态的Saga记录：

  -- 更新为Completed状态，触发自动清理
  UPDATE job_sagas 
  SET CurrentState = 10, CompletedTimestamp = GETUTCDATE() 
  WHERE CurrentState = 9;
  
  -- 或直接删除记录
  DELETE FROM job_sagas 
  WHERE CurrentState = 9;
  

• 操作前务必备份数据库，避免数据丢失。

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