Masstransit: Retry filter should cancel when the bus is stopped

[andersnm]

When a retry policy is waiting for a retry delay to elapsed, or if the filter is checking whether or not a retry can be performed, it should obey the cancellationToken of the endpoint being stopped so that an subsequent retries will be avoided and the message should remain on the queue.

It should _not_ be moved to the error or skipped queues.

Original Issue

In the following example, pressing a key causes serviceBus.Stop() to hang for a while (until all retries are exhausted?) and then prints:

MassTransit.Util.TaskSupervisor Error: 0 : Failed to close scope MassTransit.RabbitMqTransport.Pipeline.RabbitMqBasicConsumer - rabbitmq://localhost/RetryBreakRepro/Test_Send, System.OperationCanceledException: The operation was canceled.

```c#
public class MyConsumer : IConsumer {
public Task Consume(ConsumeContext context) {
throw new NotImplementedException();
}
}

...

var serviceBus = Bus.Factory.CreateUsingRabbitMq(config => {
    var host = config.Host(new Uri("rabbitmq://localhost/RetryBreakRepro"), h => {
        h.Username("guest");
        h.Password("guest");
    });

    config.ReceiveEndpoint(host, "Test_Send", e => {
        e.UseRetry(r => r.Interval(10, TimeSpan.FromSeconds(10)) );
        e.Consumer<MyConsumer>();
    });
});

serviceBus.Start();
serviceBus.Publish(new My());
Console.ReadKey();
serviceBus.Stop();

Is this expected?

To workaround I tried switching to using .UseDelayedRedelivery() as described in issue #493, but can't get it to compile no matter how I type it out:

```c#
    config.UseDelayedExchangeMessageScheduler();

    ...

    e.Consumer<MyConsumer>(c => {
        c.UseDelayedRedelivery<IMy>(Retry.Interval(10, TimeSpan.FromSeconds(10)));
    });

I noticed the DelayRetry_Specs.cs test uses .Handler(...) instead of .Consumer(...) and indeed that compiles, runs, and throws no exceptions in serviceBus.Stop().

Should .UseDelayedRedelivery() work with consumers too?

Dependencies:
"MassTransit": "3.5.4",
"MassTransit.RabbitMQ": "3.5.4"

[phatboyg]

For delayed redelivery, use:

e.Consumer<MyConsumer>(c =>
{
    c.Message<IMy>(m => m.UseDelayedRedelivery(...));
});

[phatboyg]

As to the other items, I'll have to check into them to see what's happening.

[paveldayneko]

Hello, I ran into exactly the same issue and using UseDelayedRedelivery() looks like a great workaround. But is there any update regarding that?

[alexeyzimarev]

How people would like it to work? When bus is being stopped, all consumers are being waited to finish processing and since retries are part of the pipeline, messages that are being retried, only finish the pipeline when all retries are exhausted. So this is by design, the question is - do we need to do anything about it? If yes - it will not be easy since all retry policies will need to get some sort of semaphore saying - ok, give up now. But also, does this mean that messages need to be unACKed or moved to the error queue?

[andersnm]

I would expect the retry policy to use a cancellationtoken when doing the sleep Task.Delay() so MT can shut down quickly and cleanly in case the service is stopped. F.ex if the server is rebooted during retries, I'd rather my service shut down cleanly than being killed by the OS after a timeout. I would also expect the message to be put back on the queue.

[ushenkodmitry]

+1 for me

On ср, 6 сент. 2017 г., 10:23 andersnm [e-mail] wrote:

I would expect the retry policy to use a cancellationtoken when doing the
sleep Task.Delay() so MT can shut down quickly and cleanly in case the
service is stopped. F.ex if the server is rebooted during retries, I'd
rather my service shut down cleanly than being killed by the OS after a
timeout. I would also expect the message to be put back on the queue.

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
https://github.com/MassTransit/MassTransit/issues/780#issuecomment-327398185,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AP7WH3lr3AOpYVng70vYLNHENfzscniGks5sfkhcgaJpZM4LrIh_
.

[MatteoSp]

+1

I would expect the retry policy to use a cancellationtoken when doing the sleep Task.Delay() so MT can shut down quickly and cleanly in case the service is stopped. F.ex if the server is rebooted during retries, I'd rather my service shut down cleanly than being killed by the OS after a timeout. I would also expect the message to be put back on the queue.

[alexeyzimarev]

expect the message to be put back on the queue

well, messages are not being "removed" from the queue to start with, they are being consumed whilst retrying and not ACKed. Unless we ACK the message, they will be kept in the queue.

[phatboyg]

I'm considering this change, but I need some feedback.

When the bus is stopped, I changed it so that the CancellationToken on every ReceiveContext is signaled. If a consumer is in a retry loop, it exits immediately and the message remains on the queue in the current position (it is a NACK essentially).

This also means that if a consumer is executing, and makes an asynchronous call using the same token, that async operation would also be cancelled. Since previously the consumers would complete gracefully and then the consumer would exit, this is a pretty serious change, and honestly, I don't think I like it at all. But, in reality, it does make sense in some ways.

The other option is to keep the CancellationToken on the ConsumeContext _separate_ from the one on the ReceiveContext, but again, not sure what value that adds other than confusing things.

Or I can add a "Stopping" token to the receive context, and use that in the retry filter, but that isn't easy since the retry filter uses the PipeContext CancellationToken, so that's sort of not going to work...

I dont' want to make it weird, I want it to be consistent.

[ushenkodmitry]

I personally vote for this change. Stopping the bus should cancel all
activities, including retries.
I my case i only stop the bus when the application is going to close. I
don't know any scenarios, when the bus must be stopped and then started
again, during the lifetime of the host process.
But, if such scenarios do exist, then it is desirable that retry attempts
are not going to be lost during bus restart.

вт, 27 февр. 2018 г. в 22:27, Chris Patterson [e-mail]:

I'm considering this change, but I need some feedback.

When the bus is stopped, I changed it so that the CancellationToken on
every ReceiveContext is signaled. If a consumer is in a retry loop, it
exits immediately and the message remains on the queue in the current
position (it is a NACK essentially).

This also means that if a consumer is executing, and makes an asynchronous
call using the same token, that async operation would also be cancelled.
Since previously the consumers would complete gracefully and then the
consumer would exit, this is a pretty serious change, and honestly, I don't
think I like it at all. But, in reality, it does make sense in some ways.

The other option is to keep the CancellationToken on the ConsumeContext
separate from the one on the ReceiveContext, but again, not sure what
value that adds other than confusing things.

Or I can add a "Stopping" token to the receive context, and use that in
the retry filter, but that isn't easy since the retry filter uses the
PipeContext CancellationToken, so that's sort of not going to work...

I dont' want to make it weird, I want it to be consistent.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/MassTransit/MassTransit/issues/780#issuecomment-368996203,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AP7WHx2v2sKTH5Gz2Fpx83YVLl-4eAB1ks5tZFbPgaJpZM4LrIh_
.