Microservices transactions with saga orchestrator

Question

I'm trying to build microservices application. I want to make auth/register/login gateway service checks every request and modify headers to pass through to internal services and user profile service.

I made an saga orchestrator with mass transit in ASP.NET Core.

The problem is: while I make waiting request to saga I can't get a response if on not first event step. I getting only timeout exceptions.

The algorithm is: user makes register request to auth service -> service makes waiting request to saga -> saga tries create empty profile in user service -> if no - user gets answer. if yes - saga tries create auth profile with gotten user profile id. if yes - user gets answer etc.

The request client is auth service and he answers to self. The auth, user, saga orchestrator are all different services

Auth service request code:

 var correlationId = Guid.NewGuid();
 var response = await requestClient.GetResponse<RegistrationInfoMessage>(new StartRegistrationCommand
 {
 CorrelationId = correlationId,
 UserEmail = email,
 }, timeout: TimeSpan.FromSeconds(30));

Auth mass transit register:

builder.Services.AddMassTransit(x =>
{
 x.AddConsumer<AuthProfileConsumer>();
 x.UsingRabbitMq((ctx, cfg) =>
 {
 cfg.Host(builder.Configuration["AUTH_SAGA_RABBIT_MQ_HOST"], h =>
 {
 h.Username(builder.Configuration["AUTH_SAGA_RABBIT_MQ_USER"]);
 h.Password(builder.Configuration["AUTH_SAGA_RABBIT_MQ_PASSWORD"]);
 });
 cfg.ReceiveEndpoint("auth_profile_queue", e =>
 {
 e.ConfigureConsumer<AuthProfileConsumer>(ctx);
 });
 });
 
 x.AddRequestClient<RegistrationInfoMessage>(timeout: TimeSpan.FromMinutes(5));
});

User service event code:

public async Task Consume(ConsumeContext<CreateUserCommand> context)
{
 try
 {
 logger.LogInformation($"Creating empty user profile of saga registration process. Correlation: {context.CorrelationId}");
 var result = await userService.CreateEmptyProfileAsync(context.Message.Email);
 if (!result.Success)
 {
 logger.LogError($"Failed to create profile: {context.Message.Email} with error: {result.ErrorMessage} and code: {result.ErrorCode}");
 await context.Publish(new UserRejectedCommand(context.Message.CorrelationId));
 return;
 }
 await context.RespondAsync(new UserCreatedCommand(context.Message.CorrelationId, result.ProfileId));
 // await context.Publish(new UserCreatedCommand(context.Message.CorrelationId, result.ProfileId));
 }
 catch (Exception e)
 {
 logger.LogError($"Failed to create profile: {context.Message.Email} with error: {e.Message}");
 // await context.Publish(new UserRejectedCommand(context.Message.CorrelationId));
 await context.RespondAsync(new UserRejectedCommand(context.Message.CorrelationId));
 }
}

Saga orchestrator register code:

builder.Services.AddMassTransit(x =>
{
 x.AddSagaStateMachine<RegistrationSaga, RegistrationState>()
 .EntityFrameworkRepository(r =>
 {
 r.ExistingDbContext<SagaContext>();
 r.ConcurrencyMode = ConcurrencyMode.Optimistic;
 });
 x.UsingRabbitMq((ctx, cfg) =>
 {
 cfg.Host(builder.Configuration["SAGA_RABBIT_MQ_HOST"], h =>
 {
 h.Username(builder.Configuration["SAGA_RABBIT_MQ_USER"]);
 h.Password(builder.Configuration["SAGA_RABBIT_MQ_PASSWORD"]);
 });
 cfg.UseJsonSerializer();
 cfg.ReceiveEndpoint("saga_queue", e =>
 {
 e.ConfigureSaga<RegistrationState>(ctx);
 });
 cfg.UseMessageRetry(r => r.Intervals(100, 500, 1000));
 });
});

The saga orchestrator:

public class RegistrationSaga : MassTransitStateMachine<RegistrationState>
{
 private readonly ILogger<RegistrationSaga> _logger;
 
 public State ProfileCreating { get; private set; }
 public State AuthCreating { get; private set; }
 public State Completed { get; private set; }
 public State Failed { get; private set; }
 
 public Event<StartRegistrationCommand> StartRegistration { get; private set; }
 
 public Event<UserCreatedCommand> ProfileCreated { get; private set; }
 public Event<UserRejectedCommand> ProfileRejected { get; private set; }
 
 public Event<AuthCreatedCommand> AuthCreated { get; private set; }
 public Event<AuthRejectedCommand> AuthRejected { get; private set; }
 
 public RegistrationSaga(ILogger<RegistrationSaga> logger)
 {
 _logger = logger;
 InstanceState(x => x.CurrentState);
 
 Event(() => StartRegistration, x => x.CorrelateById(context => context.Message.CorrelationId));
 
 Event(() => ProfileCreated, x => x.CorrelateById(context => context.Message.CorrelationId));
 Event(() => ProfileRejected, x => x.CorrelateById(context => context.Message.CorrelationId));
 
 Event(() => AuthCreated, x => x.CorrelateById(context => context.Message.CorrelationId));
 Event(() => AuthRejected, x => x.CorrelateById(context => context.Message.CorrelationId));
 
 
 Initially(
 When(StartRegistration)
 .Then(ctx =>
 {
 ctx.Saga.CorrelationId = ctx.Message.CorrelationId;
 ctx.Saga.Email = ctx.Message.UserEmail;
 
 ctx.Saga.RequestId = ctx.RequestId;
 ctx.Saga.ResponseAddress = ctx.ResponseAddress;
 
 logger.LogInformation($"Registration started: {ctx.Saga.CorrelationId}, RequestId: {ctx.RequestId}, ResponseAddress: {ctx.ResponseAddress}");
 })
 .Publish(ctx => new CreateUserCommand(ctx.Saga.CorrelationId, ctx.Saga.Email))
 .TransitionTo(ProfileCreating)
 ); 
 
 During(ProfileCreating,
 When(ProfileCreated)
 .Then(ctx =>
 { 
 ctx.Saga.CorrelationId = ctx.Message.CorrelationId;
 ctx.Saga.UserProfileId = ctx.Message.UserProfileId;
 
 logger.LogInformation($"Profile created for registration: {ctx.Saga.CorrelationId}");
 })
 .Publish(ctx => new CreateAuthCommand(ctx.Saga.CorrelationId, ctx.Saga.Email, "password_placeholder", ctx.Saga.UserProfileId))
 .TransitionTo(AuthCreating),
 
 When(ProfileRejected)
 .Then(ctx =>
 {
 ctx.Saga.CorrelationId = ctx.Message.CorrelationId;
 logger.LogError($"Registration profile rejected: {ctx.Saga.RequestId}");
 })
 .RespondAsync(ctx => ctx.Init<RegistrationInfoMessage>(new {
 CorrelationId = ctx.Message.CorrelationId,
 Error = "Profile creation failed"
 }))
 .TransitionTo(Failed)
 .Finalize()
 );
 
 During(AuthCreating,
 When(AuthCreated)
 .Then(ctx =>
 {
 ctx.Saga.CorrelationId = ctx.Message.CorrelationId;
 logger.LogInformation($"Registration completed: {ctx.Saga.CorrelationId}");
 })
 .Respond(ctx => new RegistrationInfoMessage
 {
 CorrelationId = ctx.Message.CorrelationId,
 UserProfileId = ctx.Saga.UserProfileId,
 })
 .TransitionTo(Completed)
 .Finalize(),
 
 When(AuthRejected)
 .Then(ctx =>
 {
 ctx.Saga.CorrelationId = ctx.Message.CorrelationId;
 logger.LogError($"Registration auth rejected: {ctx.Saga.CorrelationId}");
 })
 // Компенсация - удаляем профиль
 .Publish(ctx => new DeleteUserCommand(ctx.Saga.CorrelationId, ctx.Saga.UserProfileId))
 .Respond(ctx => new RegistrationInfoMessage
 {
 CorrelationId = ctx.Message.CorrelationId,
 Error = "Auth creation failed"
 })
 .TransitionTo(Failed)
 .Finalize()
 );
 
 SetCompletedWhenFinalized();
 }
}

• JonasH: The main benefits of microservices are at large scales, both in team size and load. Are you sure that the benefits outweigh the increased complexity for your system? Or is this mainly a learning exercise?
• Leopold95: its all of that. ill learning microservices, planning to have best pet project for future job and perfect diploma.
• JonasH: This is just my opinion, but if you want to learn about micro services I would take work at some large organization that have the kinds of problems micro services is intended to solve. There is a lot of overhead work required to ensure each micro service is truly independent. If you are the sole developer it will be tempting to sacrifice some of that independence to get something that works, and there is a risk this will result in a distributed monolith.

Answer 1

To fix this, you need to change your approach. The Auth service shouldn't be using GetResponse to wait for a final result from a long-running saga. Instead, the saga should be responsible for sending the final result back to the Auth service on a dedicated endpoint.

Here's a revised and more robust algorithm:

1. Auth Service Initiates: The Auth service sends a StartRegistrationCommand to the saga queue using await bus.Send or await bus.Publish. This is a fire-and-forget message, not a request that expects an immediate response. It should not use GetResponse.

2. Saga Process Continues: The saga receives the StartRegistrationCommand, saves the ResponseAddress and RequestId from the message context, and proceeds with its business logic (User service, Auth service, etc.).

3. Saga Completes or Fails: Once the saga reaches its Completed or Failed state, it uses the saved ResponseAddressand RequestId to send the final result message back to the Auth service.

4. Auth Service Consumes the Final Result: The Auth service now needs a consumer to listen for this final result on a dedicated endpoint. This consumer would then handle the final response, such as sending a success or error message back to the original HTTP client.

• Leopold95: Yeah, i know about this solution with consumer in auth. But i, in case of my not senjor c# knowlage i dont know how can make this with code. Its two independent entities, consumer and auth service(as class). And there is seems no solutions in inernet. So, my final question is how can i make architecure of this in code. Maybe, it should be something like list of waiting tasks in separete class? (smt like js delayed promises) or something else? The separate class with wainting tasks easy to implement, but i think it wont be right way to do this. Ofcourse thank you for your an