Mass transit saga handling transition when multiple events of the same type have occurred

[elenacalugaru]

Hello,

I'm starting to learn how to use mass transit with rabbit mq, ef core, .net 6 and saga state machines.
Currently I have this saga UsersAssignedToCourseState with the following UsersAssignedToCourseStateMachine.
The flow is the following from a microservice a message/command is send to the orchestrator microservice to assign a list of users to a course and when receiving this message I setup the saga and i publish for each user an event . Then I can receive back for each user
an event with status completed or failure. After that I can publish another event for the user with the completed orfailure status.
When all the users have been assigned with success i want to publish another event , this will be used to
notify the person how assigned the students that all the users are assigned.
The main issue is the events for the users are all coming in the same time and the all success event
raised by state machine is not fired because the processedUserIds and SuccessfullyProcessedUserIds are getting reseted between events.
What can i use to mitigate this behaviour?

x.AddSagaStateMachine<UsersAssignedToCourseStateMachine, UsersAssignedToCourseState>((context, cfg) =>
               {
                   cfg.UseInMemoryOutbox(context);
               })
           .EntityFrameworkRepository(opt =>
           {
               opt.AddDbContext<DbContext, StateMachineDbContext>((provider, builder) =>
               {
                   builder.UseNpgsql(Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") != null
       && Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") == "Production" ?
       Environment.GetEnvironmentVariable("CONNECTIONSTRING") : configuration.GetConnectionString("PostgreSQL"), x => x.MigrationsHistoryTable("__MigrationHistorySaga", "Trial"));
               });

               opt.ConcurrencyMode = ConcurrencyMode.Optimistic;
           });
   		
   		
   public class UsersAssignedToCourseState : SagaStateMachineInstance
   {
       public Guid CorrelationId { get; set; }
       public Guid MessageId { get; set; }
       public Guid CourseId { get; set; }
       public string CourseName { get; set; }
       public string CurrentState { get; set; }
       public ICollection<AssignUserCourse> AssignedUsers { get; set; }
       public ICollection<Guid> ProcessedUserIds { get; set; }
       public ICollection<Guid> SuccessfullyProcessedUserIds { get; set; }
       public ICollection<UserFailure> FailedProcessedUserIds { get; set; }

       // If using Optimistic concurrency, this property is required
       public byte[] RowVersion { get; set; }



   }
   
   public class UserAssignedToCourseStateMap : SagaClassMap<UsersAssignedToCourseState>
   {
       protected override void Configure(EntityTypeBuilder<UsersAssignedToCourseState> entity, ModelBuilder model)
       {
           

           entity.Property(e => e.ProcessedUserIds)
               .HasConversion(
                   v => JsonConvert.SerializeObject(v),
                   v => JsonConvert.DeserializeObject<List<Guid>>(v) ?? new List<Guid>());

           entity.Property(e => e.SuccessfullyProcessedUserIds)
               .HasConversion(
                   v => JsonConvert.SerializeObject(v),
                   v => JsonConvert.DeserializeObject<List<Guid>>(v) ?? new List<Guid>());

           entity.Property(e => e.FailedProcessedUserIds)
               .HasConversion(
                   v => JsonConvert.SerializeObject(v),
                   v => JsonConvert.DeserializeObject<List<UserFailure>>(v) ?? new List<UserFailure>());

           entity.Property(e => e.AssignedUsers)
               .HasConversion(
             v => JsonConvert.SerializeObject(v),
             v => JsonConvert.DeserializeObject<List<AssignUserAsset>>(v) ?? new List<AssignUserCourse>());


       }
   }
   
   public class UsersAssignedToCourseStateMachine : MassTransitStateMachine<UsersAssignedToCourseState>
   {
      

       // Commands
       private Event<IAssignUsersToCourseMessage> AssignUsersToCourseMessage { get; }

       // Events
       public Event<IUserCourseScopesUpdateCompletedEvent> UserCourseScopesUpdateCompletedEvent { get; set; }
       public Event<IUserCourseScopesUpdateFailedEvent> UserCourseScopesUpdateFailedEvent { get; set; }
       public Event<IUsersCourseAssignmentSuccessEvent> AssignmentAllSuccess { get; private set; }
       public Event<IUsersCourseAssignmentFailureEvent> AssignmentAllFailure { get; private set; }
       public Event<IUsersCourseAssignmentMixedEvent> AssignmentMixed { get; private set; }

       // States
       public State ProcessingCourseUsers { get; private set; }
       public State ProcessedCourseUser { get; private set; }


       public UsersAssignedToCourseStateMachine()
       {

           InstanceState(x => x.CurrentState);

           Event(() => AssignUsersToCourseMessage,
               y => y.CorrelateBy((instance, context) => instance.MessageId == context.Message.MessageId)
               .SelectId(context => Guid.NewGuid()));


           Event(() => UserCourseScopesUpdateFailedEvent, x => x.CorrelateById(y => y.Message.CorrelationId));
           Event(() => UserCourseScopesUpdateCompletedEvent, x => x.CorrelateById(y => y.Message.CorrelationId));
           Event(() => AssignmentAllSuccess, x => x.CorrelateById(x => x.Message.CorrelationId));
           Event(() => AssignmentAllFailure, x => x.CorrelateById(y => y.Message.CorrelationId));
           Event(() => AssignmentMixed, x => x.CorrelateById(y => y.Message.CorrelationId));


           Initially(
               When(AssignUsersToCourseMessage)
                   .Then(context =>
                   {
                       context.Saga.CourseId = context.Message.CourseId;
                       context.Saga.CourseName = context.Message.CourseName;
                       context.Saga.MessageId = context.Message.MessageId;
                       context.Saga.CreatedDate = DateTime.UtcNow;
                       context.Saga.AssignedUsers = context.Message.AssignedUsers;
                       context.Saga.ProcessedUserIds = new List<Guid>();
                       context.Saga.SuccessfullyProcessedUserIds = new List<Guid>();
                       context.Saga.FailedProcessedUserIds = new List<UserFailure>();

                       foreach (var CourseUser in context.Saga.AssignedUsers)
                       {

                           context.Publish(new AssignedUserToCourseEvent
                           {
                               CorrelationId = context.Saga.CorrelationId,
                               UserId = CourseUser.UserId,
                               UserName = CourseUser.UserName,
                               CourseId = context.Saga.CourseId,
                               CourseName = context.Saga.CourseName,
                               RoleId = CourseUser.RoleId,
                               Timestamp = DateTime.UtcNow,
                               EventId = Guid.NewGuid()
                           });
                       }
                   })
                   .TransitionTo(ProcessingCourseUsers));


           During(ProcessingCourseUsers,
               When(UserCourseScopesUpdateCompletedEvent)
                   .Then(context => HandleUserScopeUpdated(context))
                   .TransitionTo(ProcessedCourseUser)
                   .ThenAsync(async context => await CheckCompletionAsync(context)),



              When(UserCourseScopesUpdateFailedEvent)
               .Then(context => HandleUserScopeUpdateFailed(context))
               .TransitionTo(ProcessedCourseUser)
               .ThenAsync(async context => await CheckCompletionAsync(context)),

              When(AssignmentMixed)
              .Publish(context => new UsersCourseAssignmentMixedEvent
              {
                  CorrelationId = context.Saga.CorrelationId,
                  CourseId = context.Saga.CourseId,
                  CourseName = context.Saga.CourseName,
                  AssignedUsersSuccess = PopulateAssignedUsersSuccess(context.Saga.AssignedUsers, context.Saga.SuccessfullyProcessedUserIds),
                  AssignedUsersFailure = PopulateAssignedUsersFail(context.Saga.AssignedUsers, context.Saga.FailedProcessedUserIds),
                  Timestamp = DateTime.UtcNow,
                  EventId = Guid.NewGuid()
              })
              .Finalize(),

              When(AssignmentAllSuccess)
              .Publish(context => new UsersCourseAssignmentSuccessEvent
              {
                  CorrelationId = context.Saga.CorrelationId,
                  CourseId = context.Saga.CourseId,
                  CourseName = context.Saga.CourseName,
                  AssignedUsersSuccess = context.Saga.AssignedUsers,
                  Timestamp = DateTime.UtcNow,
                  EventId = Guid.NewGuid()
              })
              .Finalize(),

              When(AssignmentAllFailure)
              .Publish(context => new UsersCourseAssignmentFailureEvent
              {
                  CorrelationId = context.Saga.CorrelationId,
                  CourseId = context.Saga.CourseId,
                  CourseName = context.Saga.CourseName,
                  AssignedUsersFailure = PopulateAssignedUsersFail(context.Saga.AssignedUsers, context.Saga.FailedProcessedUserIds),
                  Timestamp = DateTime.UtcNow,
                  EventId = Guid.NewGuid()
              })
              .Finalize()

           );


       }

       private ICollection<AssignUserCourse> PopulateAssignedUsersSuccess(ICollection<AssignUserCourse> assignedUsers, ICollection<Guid> successfullyProcessedUserIds)
       {
           var result = new List<AssignUserCourse>();
           foreach (var user in assignedUsers)
           {
               var okUser = successfullyProcessedUserIds.FirstOrDefault(u => u == user.UserId);
               if (okUser != Guid.Empty)
               {
                   result.Add(user);
               }

           }

           return result;
       }

       private ICollection<AssignUserCourse> PopulateAssignedUsersFail(ICollection<AssignUserCourse> assignedUsers, ICollection<UserFailure> failedProcessedUserIds)
       {
           var result = new List<AssignUserCourse>();
           foreach (var user in assignedUsers)
           {
               var failedUser = failedProcessedUserIds.FirstOrDefault(u => u.UserId == user.UserId);
               if (failedUser != null)
               {
                   user.ErrorMessage = failedUser?.ErrorMessage ?? string.Empty;
                   result.Add(user);
               }

           }

           return result;
       }

       private async Task CheckCompletionAsync(BehaviorContext<UsersAssignedToCourseState> context)
       {
           if (AllProcessedUsersSuccessful(context.Saga))
           {
               var allSuccess = new UsersCourseAssignmentSuccessEvent { CorrelationId = context.Saga.CorrelationId };
               await context.Raise(AssignmentAllSuccess, allSuccess);

           }
           else if (AllProcessedUsersFailed(context.Saga))
           {
               await context.Raise(AssignmentAllFailure);
           }
           else if (context.Saga.ProcessedUserIds.Count == context.Saga.AssignedUsers.Count)
           {
               await context.Raise(AssignmentMixed);
           }
           else
           {
               await context.TransitionToState(ProcessingCourseUsers);
           }
       }

       private void HandleUserScopeUpdateFailed(BehaviorContext<UsersAssignedToCourseState, IUserCourseScopesUpdateFailedEvent> context)
       {


           context.Saga.ProcessedUserIds.Add(context.Message.UserId);

           context.Saga.FailedProcessedUserIds.Add(new UserFailure() { UserId = context.Message.UserId, ErrorMessage = context.Message.ErrorMessage });

           context.Publish(new AssignedUserToCourseFailedEvent
           {
               CorrelationId = context.Saga.CorrelationId,
               UserId = context.Message.UserId,
               UserName = context.Message.UserName,
               CourseName = context.Saga.CourseName,
               CourseId = context.Saga.CourseId,
               RoleId = context.Message.RoleId,
               Timestamp = DateTime.UtcNow,
               ErrorMessage = context.Message.ErrorMessage,
               EventId = Guid.NewGuid()

           });


       }

       private void HandleUserScopeUpdated(BehaviorContext<UsersAssignedToCourseState, IUserCourseScopesUpdateCompletedEvent> context)
       {


           context.Saga.ProcessedUserIds.Add(context.Message.UserId);

           context.Saga.SuccessfullyProcessedUserIds.Add(context.Message.UserId);

           context.Publish(new AssignedUserToCourseCompletedEvent
           {
               CorrelationId = context.Saga.CorrelationId,
               UserId = context.Message.UserId,
               UserName = context.Message.UserName,
               CourseId = context.Saga.CourseId,
               CourseName = context.Saga.CourseName,
               Timestamp = DateTime.UtcNow,
               EventId = Guid.NewGuid()

           });


       }

       private static bool AllProcessedUsersSuccessful(UsersAssignedToCourseState state)
       {
           return state.ProcessedUserIds.All(id => state.SuccessfullyProcessedUserIds.Contains(id)) && state.ProcessedUserIds.Count == state.AssignedUsers.Count;
       }

       private static bool AllProcessedUsersFailed(UsersAssignedToCourseState state)
       {
           return state.ProcessedUserIds.All(id => state.FailedProcessedUserIds.Any(u => u.UserId == id)) && state.ProcessedUserIds.Count == state.AssignedUsers.Count;
       }
   }

[phatboyg]

This is too much code to read, if you have a standalone repository, share it. Also, don't use optimistic concurrency if you're expecting concurrent events to the same saga instance.

[elenacalugaru]

I've changed to Pesimistic as opt.ConcurrencyMode = ConcurrencyMode.Pessimistic; opt.UsePostgres(); and I'm using SemaphoreSlim when I update the processed users succcess users or failed users but sometimes I get the some messages in the error queue with the reason

40001: could not serialize access due to concurrent update -- What should I do to avoid this kind of errors? Thanks a lot.

[phatboyg]

Conflicts happen, you should configure UseMessageRetry() to handle the concurrency exceptions and retry those messages.

[elenacalugaru]

I have configured following and then I noticed that the error changed to The UserCourseScopesUpdateFailedEvent event is not handled during the Final state for the UsersAssignedToCourseStateMachine state machine. Should this error be related to the message retry?
e.UseMessageRetry(r => r.Interval(5, 1000));
e.PrefetchCount = 20;

var partition = new Partitioner(20, new Murmur3UnsafeHashGenerator());

e.ConfigureSaga(context);

e.UsePartitioner(partition, p => p.Message.CorrelationId);
e.UsePartitioner(partition, p => p.Message.CorrelationId);