Timeout from Saga with RequestClient after publishing and receiving event

[boris1dev]

Hi,
hopefully somebody can help me as I’m looking now for almost 2 weeks for a solution for the following problem:

I have a saga which I am starting with a request client.
Inside the saga I am publishing an event, transition into a new state waiting for the answer of a consumer and then return a result to the request client.

The issue: the new state receives the answer from the consumer - and then the request client (which started the saga) gets a timeout.

I don’t find any solution because

• without publishing and receiving the event inside the saga I can return a result from the saga to the request client
• the publishing and receiving event inside the sage also work on their own
• I have also enabled the MassTransit outbox

The most relevant code (just an excerpt):

The saga itself

public class BarStateMachine : MassTransitStateMachine<BarState>
{
    public State WaitForFirstNameCreated { get; }

    public Event<StartEvent> Start { get; }

    public Event<FirstNameCreatedEvent> FirstNameCreated { get; }

    public BarStateMachine(ILogger<BarStateMachine> logger)
    {
        InstanceState(state => state.CurrentState);

        ConfigureEvents();
        ConfigureInitialState();
        ConfigureWaitForFirstNameCreated();
    }

    private void ConfigureEvents()
    {
        Event(() => Start);
        Event(() => FirstNameCreated);
    }

    private void ConfigureInitialState() =>
        Initially(
            When(Start)
                .Then(context =>
                {
                    context.Saga.FirstName = context.Message.FirstName;
                })
                .PublishAsync(context => context.Init<CreateFirstNameEvent>(new {
                    MessageId = Guid.NewGuid(),
                    DateTime = SystemTimeProvider.UtcNow(),
                    context.Saga.FirstName,
                    context.Saga.CorrelationId}))
                .TransitionTo(WaitForFirstNameCreated)
        );

    private void ConfigureWaitForFirstNameCreated()
        => During(
            WaitForFirstNameCreated,
            Ignore(Start),
            When(FirstNameCreated)
                .Then(context =>
                {
                    context.Saga.CorrelationId = context.Message.CorrelationId;
                    context.Saga.FirstNameId = context.Message.FirstNameId;
                    context.Saga.LastUpdated = DateTimeOffset.UtcNow;
                })
			    .RespondAsync(context => context.Init<SagaMachineResponse>(new {
			        MessageId = new MessageId(Guid.NewGuid()),
			        Message = "Saga: Say hello!",
			        context.Saga.CorrelationId
			    }))

Calling the saga

Response<SagaMachineResponse> state = await sagaClient.GetResponse<SagaMachineResponse>(
    new StartEvent(
        new MessageId(Guid.NewGuid()),
        "Foo",
        Guid.NewGuid()), cancellationToken);

Contracts

public interface SagaMachineResponse
{
    public MessageId Id { get; set; }

    public string Message { get; set; }

    public Guid CorrelationId { get; set; }
}

public record StartEvent(
    MessageId Id,
    string FirstName,
    Guid CorrelationId);

public interface CreateFirstNameEvent
{
    public MessageId Id { get; set; }

    public DateTime OccurredOnUtc { get; set; }

    public string FirstName { get; set; }

    public Guid CorrelationId { get; set; }
}

public sealed record FirstNameCreatedEvent(
    Guid Id,
    DateTime OccurredOnUtc,
    Guid FirstNameId,
    Guid CorrelationId);

The consumer

public sealed class CreateFirstNameConsumer : IConsumer<CreateFirstNameEvent>
{
    public async Task Consume(ConsumeContext<CreateFirstNameEvent> context)
    {
        var payload = new FirstNameCreatedSagaEvent(
            Guid.NewGuid(),
            DateTime.UtcNow,
            Guid.NewGuid(),
            context.Message.CorrelationId);

        await context.Publish(payload);
    }
}

The configuration

.AddMassTransit(busConfigurator =>
{
    busConfigurator.AddRequestClient<StartEvent>();

    busConfigurator
        .AddSagaStateMachine<BarStateMachine, BarState>(sagaConfigurator
            => sagaConfigurator.UseInMemoryOutbox())
        .MongoDbRepository(r =>
        {
            MongoDbSettings? mongoDbSettings = configuration
                .GetSection(nameof(MongoDbSettings))
                .Get<MongoDbSettings>();

            ArgumentNullException.ThrowIfNull(mongoDbSettings);

            r.Connection = mongoDbSettings.ConnectionString;
            r.DatabaseName = mongoDbSettings.Database;
        });


	busConfigurator.AddConsumer<CreateFirstNameConsumer>();

    busConfigurator.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.UseMessageRetry(retryConfig =>
        {
            retryConfig.Interval(3, TimeSpan.FromSeconds(5));
        });

        cfg.Host("localhost", 5672, "/", c =>
        {
            c.Username("guest");
            c.Password("guest");
        });
        cfg.ConfigureEndpoints(ctx);
    });
})

Thanks for any help

[phatboyg]

https://stackoverflow.com/questions/51589028/masstransit-saga-statemachine-not-returning-the-request-response-to-the-controll