Are there limitation on short durations when scheduling events from the Initially block in a MassTransit saga?

Question

I'm working on a POC for using MassTransit sagas to handle state changes in a system for grant applications. I'm using MassTransit 8.0.0-develop.394, .Net 6, EF Core 6.0.2 and ActiveMQ Artemis 1.19.0.

In the final solution the applicants can register their application and prepare the data for several weeks. A few days before the deadline another external system will be populated with data that will be used to validate the application data. Application data entered before the validation data is populated should just be scheduled for later validation, but data entered after should be validated immediately. I think MassTransit sagas with scheduled events looks like a good fit for this.

In the POC I just schedule the validation start time for some 10 seconds into the future from the program starts, and uses a shorter and shorter delay in the schedule until I just schedule it with a delay of TimeSpan.Zero.

From looking in the database I noticed that some of the schedule events somehow get lost when I run the POC with an empty saga repository, but everything works fine when I rerun the the program with existing sagas in the database. I use the same scheduling code in Initially and in DuringAny, which make me think that there might be some limitations on how short delay its safe to use when scheduling saga events?

Note 1: I've switched to not schedule the event in the saga when its less than 1 second to the valdation can be started, then I just publish the validation message directly, so this issue is not blocking me at the moment.

Note 2: I noticed this when running the POC from the command line and checking the database manually. I've tried to reproduce it in a test using the TestHarness, and also using ActiveMQ Artemis and InMemoryRepository, but with no luck. I've been able to reproduce it (more or less consistently) with a test using Artemis and EF Core Repository. I must admit that the test got quite complex with a lot of Task.Delay and other stuff, so it might be hard to follow the logic, but I can post it here if anyone think it's of any help.

Update 2 using Chris Pattersons recommendation about cfg.UseMessageRetry and cfg.UseInMemoryOutbox in the SagaDefinition and not on the bus.

Here is the updated code where MassTransit is configured

private static ServiceProvider BuildServiceProvider()
{
    return new ServiceCollection()
        .AddDbContext<MySagaDbContext>(builder =>
        {
            MySagaDbContextFactory.Apply(builder);
        })
        .AddMassTransit(cfg =>
        {
            cfg.AddDelayedMessageScheduler();
            cfg.UsingActiveMq((context, config) =>
            {
                config.Host("artemis", 61616, configureHost =>
                {
                    configureHost.Username("admin");
                    configureHost.Password("admin");
                });
                config.EnableArtemisCompatibility();
                config.UseDelayedMessageScheduler();
                config.ConfigureEndpoints(context);
            });
            cfg.AddSagaStateMachine<MyStateMachine, MySaga, MySagaDefinition<MySaga>>()
                .EntityFrameworkRepository(x =>
                {
                    x.ConcurrencyMode = ConcurrencyMode.Optimistic;
                    x.ExistingDbContext<MySagaDbContext>();
                });
        })
        .AddLogging(configure =>
        {
            configure.AddFilter("MassTransit", LogLevel.Error); // Filter out all retry warnings
            configure.AddFilter("Microsoft", LogLevel.None);
            configure.AddSimpleConsole(options =>
            {
                options.UseUtcTimestamp = true;
                options.TimestampFormat = "HH:mm:ss.fff ";
            });
        })
        .BuildServiceProvider(true);
}

Here is the updated saga definition code

public class MySagaDefinition<TSaga> : SagaDefinition<TSaga> where TSaga : class, ISaga
{
    protected override void ConfigureSaga(IReceiveEndpointConfigurator endpointConfigurator, ISagaConfigurator<TSaga> consumerConfigurator)
    {
        endpointConfigurator.UseMessageRetry(r => r.Intervals(10, 50, 100, 500, 1000));
        endpointConfigurator.UseInMemoryOutbox();
    }
}

Answer 1

If you are scheduling messages from a saga, or really producing any messages from a saga, you should always have the following middleware components configured:

cfg.UseMessageRetry(r => r.Intervals(50,100,1000));
cfg.UseInMemoryOutbox();

That will ensure that messages produced by the saga are:

• Only produced if the saga is successfully saved to the repository
• Produced after the saga has been saved to the repository

More details are available in the documentation.

The reason being, a short delay is likely delivering the message before it has been saved, and the scheduled event isn't correlating to an existing saga instance because it hasn't saved yet.