Masstransit: UseInMemoryOutbox doesn't work as expected in case of combining Retry and DelayedRedelivery policies

[pshenichnov]

Is this a bug report?

Yes

Can you also reproduce the problem with the lastest version?

Yes

Environment

1. Operating system: Windows 10 Pro
2. Visual Studio version: 2017
3. Dotnet version: Net Core 2.0

Steps to Reproduce

(Write your steps here:)

1. Configure UseDelayedRedelivery and UseMessageRetry
2. Add UseInMemoryOutbox

Expected Behavior

Messages published before the consumer has failed should not be issued.

Actual Behavior

In the example below I don't expect message InnerCommand to be consumed at all, but in reality I get consumer hit twice

using GreenPipes;
using MassTransit;
using MassTransit.RabbitMqTransport;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Serilog;
using System;
using System.Threading.Tasks;

namespace Receiver
{
    class Program
    {
        static void Main(string[] args)
        {
            var builder = new ConfigurationBuilder()
                    .AddJsonFile("appsettings.json", true, true);

            var configuration = builder.Build();

            Log.Logger = new LoggerConfiguration()
                .MinimumLevel.Debug()
                .WriteTo.Console()
                .ReadFrom.Configuration(configuration)
                .CreateLogger();

            Log.Information("Starting Receiver...");

            var services = new ServiceCollection();

            services.AddSingleton(context => Bus.Factory.CreateUsingRabbitMq(x =>
            {
                IRabbitMqHost host = x.Host(new Uri("rabbitmq://guest:guest@localhost:5672/test"), h => { });

                x.UseDelayedExchangeMessageScheduler();

                //x.UseInMemoryOutbox();

                x.ReceiveEndpoint(host, $"receiver_queue", e =>
                {
                    x.UseInMemoryOutbox();

                    e.Consumer<TestHandler>();

                    e.UseDelayedRedelivery(r =>
                    {
                        r.Interval(1, TimeSpan.FromMilliseconds(100));
                        r.Handle<Exception>();
                    });

                    e.UseMessageRetry(r =>
                    {
                        r.Interval(1, TimeSpan.FromMilliseconds(100));
                        r.Handle<Exception>();
                    });
                });

                x.UseSerilog();
            }));

            var container = services.BuildServiceProvider();

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

            busControl.Start();

            busControl.Publish<TestCommand>(new
            {
                Id = NewId.NextGuid()
            });

            Log.Information("Receiver started...");
        }
    }

    public interface TestCommand
    {
        Guid Id { get; }
    }

    public interface InnerCommand
    {
        Guid Id { get; }
    }

    public class TestHandler : IConsumer<TestCommand>, IConsumer<InnerCommand>
    {
        static int count = 0;

        public Task Consume(ConsumeContext<TestCommand> context)
        {
            context.Publish<InnerCommand>(new
            {
                Id = context.Message.Id
            });

            var redeliveryCount = context.Headers.Get<string>("MT-Redelivery-Count");

            Log.Information($"MT-Redelivery-Count: {redeliveryCount}; Total: {++count}");

            throw new Exception("something went wrong...");
        }

        public Task Consume(ConsumeContext<InnerCommand> context)
        {
            Log.Information($"Inner command: {context.Message.Id}");

            return Task.CompletedTask;
        }
    }
}

Have I missed something and it should be configured some other way?

Reproducible Demo

https://github.com/pshenichnov/TestInMemoryOutbox

[phatboyg]

So, order matters, and your configuration is all out of order:

    x.ReceiveEndpoint(host, $"receiver_queue", e =>
                {
                    e.UseDelayedRedelivery(r =>
                    {
                        r.Interval(1, TimeSpan.FromMilliseconds(100));
                        r.Handle<Exception>();
                    });

                    e.UseMessageRetry(r =>
                    {
                        r.Interval(1, TimeSpan.FromMilliseconds(100));
                        r.Handle<Exception>();
                    });

                    x.UseInMemoryOutbox();

                    e.Consumer<TestHandler>();
                });

[pshenichnov]

Unfortunately it didn't help much... still getting InnerCommand consumer called twice

[phatboyg]

Oh, and change x.UseInMemoryOutbox() to e.UseInMemoryOutbox() - right now it isn't on the endpoint, it's on the bus.

[phatboyg]

Long story short, the new extensions created for delayed retry and message retry don't play nicely with the pipe configuration when you use things like the outbox, since it's earlier in the pipeline. If you were to dump the pipeline, you would see how the per-message redelivery and retry pipe filters are further down the chain, making the outbox too early to behave properly.

To that end, I've had to add a new extension method and related types to setup a per-message-type outbox, and that is also making me think about how other message-pipeline filters are implemented and suspect they need to be configured the same way.

That said, the following works with the new extension methods I've added to develop.

configurator.Consumer<TestHandler>(x =>
{
    x.Message<TestCommand>(m =>
    {
        m.UseDelayedRedelivery(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
        m.UseRetry(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
        m.UseInMemoryOutbox();
    });
});

[pshenichnov]

changing x.UseInMemoryOutbox() to e.UseInMemoryOutbox() didn't help... (missed that part from the beginning)

will wait while you implement new extension methods

[pshenichnov]

Hi Chris, thanks for fixing it!

the code below now works as expected

configurator.Consumer<TestHandler>(x =>
{
    x.Message<TestCommand>(m =>
    {
        m.UseDelayedRedelivery(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
        m.UseRetry(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
        m.UseInMemoryOutbox();
    });
});

Just one question to double check... this is only the option now to use consumer configurator and setup this behavior for every message handled by consumer? There is no any other option to make it more generic (per consumer etc.)?

[phatboyg]

With 5.0.1 you'll be able to put it before the consumer and it will automatically work for all message types.

[pshenichnov]

Great! Looking forward 5.0.1!

[pshenichnov]

Wow! Just noticed that 5.0.1 been released! So fast, guys!

And configuration below now works as expected

                x.ReceiveEndpoint(host, $"receiver_queue", e =>
                {
                    e.UseDelayedRedelivery(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
                    e.UseMessageRetry(r => r.Interval(1, TimeSpan.FromMilliseconds(100)));
                    e.UseInMemoryOutbox();

                    e.Consumer<TestHandler>();
                });

thanks much!