Saga handling undeliverable messages

Question

From documentation here:

If the configuration of an endpoint changes, or if a message is mistakenly sent to an endpoint, it is possible that a message type is received that does not have any connected consumers. If this occurs, the message is moved to a _skipped queue (prefixed by the original queue name). The original message content is retained, and additional headers are added to indicate the host which moved the message.

Should I expect the same behavior for a saga when some message tried to be delivered to the saga but saga instance is not created yet? In other words, if want to handle message in saga but that message is not a trigger for saga creation. I would expect to see that messages in _skipped queue, but in reality I don't see them either in _skipped or _error queues. Also I can see that the massage successfully was delivered to the saga queue and successfully was consumed somehow without any warnings or errors. Who has consumed the message?

UPDATED

State machine:

public class TestState : SagaStateMachineInstance, IVersionedSaga
{
    public Guid CorrelationId { get; set; }
    public Guid Id { get; set; }
    public string CurrentState { get; set; }
    public int Version { get; set; }
    public DateTime TimeStart { get; set; }
    public int Progress { get; set; }
}

public class TestStateMachine : MassTransitStateMachine<TestState>
{
    public TestStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => ProcessingStarted, x => x.CorrelateById(context => context.Message.Id));
        Event(() => ProgressUpdated, x => x.CorrelateById(context => context.Message.Id));
        Event(() => ProcessingFinished, x => x.CorrelateById(context => context.Message.Id));

        Initially(
            When(ProcessingStarted)
                .Then(context =>
                {
                    context.Instance.TimeStart = DateTime.Now;
                })
                .TransitionTo(Processing)
        );

        During(Processing,
            When(ProgressUpdated)
                .Then(context =>
                {
                    context.Instance.Progress = context.Data.Progress;
                }),
            When(ProcessingFinished)
                .TransitionTo(Processed)
                .ThenAsync(async context =>
                {
                    await context.Raise(AllDone);
                })
        );

        During(Processed,
            When(AllDone)
                .Then(context =>
                {
                    Log.Information($"Saga {context.Instance.Id} finished");
                })
                .Finalize());

        SetCompletedWhenFinalized();
    }

    public State Processing { get; set; }
    public State Processed { get; set; }

    public Event AllDone { get; set; }

    public Event<ProcessingStarted> ProcessingStarted { get; set; }
    public Event<ProgressUpdated> ProgressUpdated { get; set; }
    public Event<ProcessingFinished> ProcessingFinished { get; set; }
}

public interface ProcessingStarted
{
    Guid Id { get; }
}

public interface ProgressUpdated
{
    Guid Id { get; }
    int Progress { get; }
}

public interface ProcessingFinished
{
    Guid Id { get; }
}

Configuration

var repository = new MongoDbSagaRepository<TestState>(Environment.ExpandEnvironmentVariables(configuration["MongoDb:ConnectionString"]), "sagas");

var services = new ServiceCollection();

services.AddSingleton(context => Bus.Factory.CreateUsingRabbitMq(x =>
{
    IRabbitMqHost host = x.Host(new Uri(Environment.ExpandEnvironmentVariables(configuration["MassTransit:ConnectionString"])), h => { });

    x.ReceiveEndpoint(host, "receiver_saga_queue", e =>
    {
        e.StateMachineSaga(new TestStateMachine(), repository);
    });

    x.UseRetry(r =>
    {
        r.Incremental(10, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50));
        r.Handle<MongoDbConcurrencyException>();
        r.Handle<UnhandledEventException>();
    });

    x.UseSerilog();
}));

var container = services.BuildServiceProvider();

var busControl = container.GetRequiredService<IBusControl>();

busControl.Start();

Later, when I publish ProgressUpdated event

busControl.Publish<ProgressUpdated>(new
{
    Id = id,
    Progress = 50
});

I would expect it will raise UnhandledEventException and move the massage to the _error queue but in reality I can see that message appeared in the queue, consumed somehow and I don't see it in either _error or _skipped queues. MongoDB Saga storage also doesn't have any new saga instances created.

What am I doing wrong? I need to configure saga the way that if it get ProgressUpdated event when the saga instance is not created it would be retried later when the instance is ready to accept it.

SOLUTION

Reconfigure the event ProgressUpdated and ProcessingFinished as

Event(() => ProgressUpdated, x =>
{
    x.CorrelateById(context => context.Message.Id);

    x.OnMissingInstance(m => m.Fault());
});
Event(() => ProcessingFinished, x =>
{
    x.CorrelateById(context => context.Message.Id);

    x.OnMissingInstance(m => m.Fault());
});

and catching SagaException in UseRetry later

r.Handle<SagaException>();

did the work!

Thanks @Alexey Zimarev for sharing the knowledge!

Answer 1

If a message should be handled by a saga but there is no matching instance, the OnMissingInstance callback is called.

Reconfigure ProgressUpdated and ProcessingFinished:

Event(() => ProgressUpdated, x =>
{
    x.CorrelateById(context => context.Message.Id);
    x.OnMissingInstance(m => m.Fault());
});
Event(() => ProcessingFinished, x =>
{
    x.CorrelateById(context => context.Message.Id);
    x.OnMissingInstance(m => m.Fault());
});

and catch SagaException in the retry filter:

r.Handle<SagaException>();