[Messenger][AmqpExt] Add a retry mechanism for AMQP messages #27008

sroze · 2018-04-22T20:12:05Z

Q	A
Branch?	master
Bug fix?	no
New feature?	yes
BC breaks?	no
Deprecations?	no
Tests pass?	yes
Fixed tickets	ø
License	MIT
Doc PR	symfony/symfony-docs#9756

One of the most asked features (and required to have an enterprise-grade management of errors). I chose the approach of an AmqpExt-only retry mechanism to ensure we have a great user experience with this one at least. There are discussions on-going (#26945) to have more generic things like that across adapters.

So what's the point?

When processing some messages, some error might happen. It might be that one of our 3rd party is down, that something is wrong with the last deployment, etc... All of these things will throw exceptions in the message handlers. The way of tackling this problem is very well described in @odolbeau's Swarrot Retry provider real-life example.

How to configure it

Same than for the rest, this can be configured both via the DSN or the options when creating the adapter. Here is a bunch of working DSNs with their corresponding effect:

Enable 3 retries before having an exception

amqp://guest:guest@localhost:5672/%2f/messages
    ?retry[attempts]=3

Enable 3 retries before message to be queue in a "dead queue" (i.e. messages to be looked at manually later)

amqp://guest:guest@localhost:5672/%2f/messages
    ?retry[attempts]=3
    &retry[dead_queue]=dead_queue

Enable 3 retries, dead queue and custom settings

amqp://guest:guest@localhost:5672/%2f/messages
    ?retry[attempts]=3
    &retry[exchange_name]=retry
    &retry[routing_key_pattern]=retry_attempt_%25attempt%25
    &retry[queue_name_pattern]=retry_queue_%attempt%
    &retry[dead_queue]=dead_queue
    &retry[dead_routing_key]=dead_routing_key

Enable 3 retries, with custom TTLs per attempt. Waits 10 seconds before retrying for the first attempt, 30 seconds for the 2nd attempt and 5 minutes for the 3rd one.

amqp://guest:guest@localhost:5672/%2f/messages
    ?retry[attempts]=3
    &retry[ttl][0]=10000
    &retry[ttl][1]=30000
    &retry[ttl][2]=300000

sroze · 2018-04-22T20:12:37Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+        $queue->setName(str_replace('%attempt%', $attemptNumber, $name));
+        // $queue->setFlags(AMQP_DURABLE);
+        $queue->setArguments(array(
+            'x-message-ttl' => 30000, // 30 seconds


This TTL should be configurable per attempt number.

sroze · 2018-04-22T20:12:44Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+        $exchange = $this->amqpFactory->createExchange($this->channel());
+        $exchange->setName($retryConfiguration['name'] ?? 'retry');
+        $exchange->setType(AMQP_EX_TYPE_DIRECT);
+        // $exchange->setFlags(AMQP_DURABLE);


To be removed.

sroze · 2018-04-22T20:12:54Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+    {
+        $queue = $this->amqpFactory->createQueue($this->channel());
+        $queue->setName(str_replace('%attempt%', $attemptNumber, $name));
+        // $queue->setFlags(AMQP_DURABLE);


To be removed.

sroze · 2018-04-22T20:14:56Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+                return false;
+            }
+
+            $routingKey = 'dead';


Should be a configurable dead_routing_key

dead is a very strong word for humans. Could we use expired instead? Thanks!

Well, that’s the term used by AMQP unfortunately. Maybe too_many_attempts if really you want to change? 🤔

sroze · 2018-04-22T20:15:19Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+            $queue = $this->amqpFactory->createQueue($this->channel());
+            $queue->setName($retryConfiguration['dead_queue']);
+            $queue->declareQueue();
+            $queue->bind($exchange->getName(), 'dead');


Should be using the dead_routing_key configuration.

lyrixx

IMHO, this really need to be tested against a real implementation of AMQP (RabbitMQ).

I read very quickly the code and it seems this does not work well. I guess I should test it (by hand) to ensure work, but it looks like many case are not supported:

retrying a message that come from an exchange in a fan out mode => the message will be dispatched is all queue that are bound. It should not
retrying a message that come from an exchange in an header mode => the message will not be dispatched at all.

Then, I would be nice to be able to configure the type of retry strategy (linear / backoff), and offset (the 3/4 first iteration may be too close, and so we want to skip it) and of course the delay

Another things: all header should be casted to string.
From PHP, with the PECL, it's not really possible to create a typed header (in AMQP you can tell if the header is a string or an int). So it's better to use string everywhere to avoid edge case. I some version of rabbitmq, we hit this issue. The exchange were created with "string" type. But the message where created with "int" type. So there were not matches. Really hard to find bugs ;)

Edited: I forgot to say, you can grad some code from #23315 where the retry mechanisme is really advanced / configurable. But the more I read code about Messenger, The more I think we should reconsider #23315. How I see things: I should remove everything about worker, keep only AMQP stuff, and update the Messenger to use AMQP. (ping @fabpot)

lyrixx · 2018-04-23T08:37:29Z

src/Symfony/Component/Messenger/Adapter/AmqpExt/Connection.php

+        $queue = $this->amqpFactory->createQueue($this->channel());
+        $queue->setName(str_replace('%attempt%', $attemptNumber, $retryConfiguration['queue_name_pattern'] ?? 'retry_queue_%attempt%'));
+        $queue->setArguments(array(
+            'x-message-ttl' => $retryConfiguration['ttls'][$attemptNumber - 1] ?? 30000, // 30 seconds by default


why did you add a s in the key name ?

sroze · 2018-04-23T09:13:50Z

retrying a message that come from an exchange in a fan out mode => the message will be dispatched is all queue that are bound. It should not

That's the setup I used to test this retry mechanism and it works well. I'd love you to try and see really in which case it doesn't work :)

I think we should reconsider #23315. How I see things: I should remove everything about worker, keep only AMQP stuff, and update the Messenger to use AMQP.

If you believe you can't make the AMQP adapter much better by doing so, it is probably an interesting option (lots of work on your side though 😄).

Though, there is a point in keeping the AMQP adapter simple and not having it "fully blown featured". Obviously, there is the maintenance cost point. But also, most people will use the default auto-setuped queues, right? I also believe that everybody willing to configure the queues and exchanges very precisely will very rarely be fully satisfied by auto-configuration (regardless of the amount of effort we put into it) and end up configuring their AMQP bits manually, that's what everybody has been doing so far at the end, and this retry mechanism PR allows you to configure the routing key pattern without auto-configuring it, so this use-case would work.

kbond · 2018-04-26T15:11:28Z

Question about retries in general: If a consumed message is handled by multiple handlers but only one throws an exception - how do you handle this? It probably isn't desirable to retry the message on handlers that were successful.

nicolas-grekas · 2018-05-04T00:37:28Z

Rebase needed after #27129

sroze · 2018-05-09T08:14:28Z

If a consumed message is handled by multiple handlers but only one throws an exception - how do you handle this? It probably isn't desirable to retry the message on handlers that were successful.

That's a good question. You are right, it might not be desirable to retry on successful handlers but I believe that's the only (without a crazy amount of code) way to make it simple. On the other hand, I'd argue that something "critical" (a command in the CQRS point-of-view) will only have one handler. The things that will have multiple handlers are basically event subscribers and will be projections (or similar) and there should be no problem with having them running multiple times.

sroze · 2018-05-09T08:27:15Z

PR rebased.

sroze · 2018-05-09T08:47:10Z

I suggest we go ahead with this simple (optional) retry implementation for 4.1 and replace it with the one in the AMQP component later when introduced.

ogizanagi

Not much to say. Looks good to me.

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php

ogizanagi · 2018-05-09T12:33:20Z

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php

+        }
+
+        $maximumAttempts = $retryConfiguration['attempts'] ?? 3;
+        $routingKey = str_replace('%attempt%', $attemptNumber, $retryConfiguration['routing_key_pattern'] ?? 'attempt_%attempt%');


Could we normalize all the options values before using it, in the constructor for instance?

Yep, it's a good idea.

kbond · 2018-05-09T12:34:18Z

@sroze another potential problem could be an exception being thrown at a point where retrying the message would cause additional problems.

What about an optional mechanism to have the subscriber tell the message bus/receiver to retry the message? This could also tell the bus to retry just this specific handler.

// in your handler/subscriber
try {
    // ... some logic that throws an exception
} catch (\Throwable $e) {
    throw new RetryException(__CLASS__, $e); // or instead of __CLASS__ we can get the class from the exception's trace?
}

This is probably a discussion for another place as it is a more general retry - I can open an issue if we want to discuss further.

sroze · 2018-05-09T19:04:27Z

This is probably a discussion for another place as it is a more general retry - I can open an issue if we want to discuss further.

I'd say so. Though a very good point; I'd be great if you can open an issue, describe the issue and, cherry on top, if you have an idea on how to implement, propose something :)

sroze · 2018-05-10T15:20:33Z

I don't think that fabbot's PoV is relevant here :)

sroze · 2018-05-10T15:30:19Z

Documentation PR is here: symfony/symfony-docs/pull/9756

Tobion · 2018-05-10T16:08:12Z

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php

+    private $amqpRetryExchange;
+
+    /**
+     * Available options:


With so many options, it's IMO better to use a configuration class. Then the code self-explaining and more explicit.

But it would mean that we have something creating the configuration object from the DSNs. I'd say that keeping these options as they are now is good and that we don't need more code to maintain just for this configuration details.

Yes, I think it's fine like this (but not ideal, true). Otherwise we'll need something similar as the SecurityFactoryInterface in order to make it extendable to any transport with similar use-cases.

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php

src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpReceiver.php

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php

Add some tests and remove the intrication between the receiver and the connection Configure each TTL individually and the dead routing key The `ttls` array is 0-indexed Update the retry based on feedback (`ttls` -> `ttl`, add options to documentation and normalises the default values) Catches failed retries and forward other messages' attributes

sroze · 2018-08-12T17:33:30Z

This pull-request is welcoming another round of review 💚

Koc · 2018-08-12T20:04:24Z

@sroze how can we programatically requeue message with specified delay from handler? For example for cases, when we cann't process this message now.

sroze · 2018-08-24T09:45:36Z

can we programatically requeue message with specified delay from handler?

In the given implementation we can't. I believe that this is out of scope for now anyway and that it will be handled with the AMQP component (#27140) which will provide more features such as this one 👍

fabpot · 2018-10-10T09:45:43Z

@sroze Can we talk about this one this week IRL?

ragboyjr · 2018-12-09T04:51:52Z

@sroze do you think it would be possible to implement this in away so that it would standardize the retrying logic across the various transports?

obsolete review

weaverryan · 2019-03-15T01:45:25Z

I've brought the AMQP retry logic from Sam and put it into #30557 - but also fitting it into a system where retrying has been made generic, so it can be implemented/included in future transports.

I would LOVE review on that - I'm running full speed to get the features Messenger needs/deserves, but I need queue experts to help me avoid pitfalls.

lyrixx · 2019-03-16T01:04:12Z

IMHO this one could be closed in favor or #30557

fabpot · 2019-03-17T06:58:35Z

Closing in favor of #30557

… (weaverryan) This PR was merged into the 4.3-dev branch. Discussion ---------- [Messenger] Worker events + global retry functionality | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | yes, on Messenger only | Deprecations? | no | Tests pass? | NEEDED | Fixed tickets | #29132, #27008, #27215 and part of #30540 | License | MIT | Doc PR | TODO This is an alternative to #29132 and #27008. There are several big things: 1) The `messenger:consume` does not die if a handler has an error 2) Events are dispatched before, after and on error a message being handled 3) Logic is moved out of Amqp and into the Worker so that we can have some consistent features, like error handling. 4) A generic retry system was added, which works with Amqp and future transports should support. It will work out of the box for users. Retrying works by putting the received `Envelope` back into the bus, but with the `ReceivedStamp` removed. The retry functionality has an integration test for AMQP. 5) Added a new `MessageDecodingFailedException` that transport Serializers should throw if `decode()` fails. It allows us to reject a message in this situation, as allowing it to fail but remain on the queue causes it to be retried forever. 6) A new `DelayStamp` was added, which is the first of (later) more stamps for configuring the transport layer (see #30558). BC breaks are documented in the CHANGELOG. Thanks! Commits ------- a989384 Adding global retry support, events & more to messenger transport

…ndler (keulinho, sroze) This PR was merged into the 4.3-dev branch. Discussion ---------- [Messenger] Ensure message is handled only once per handler Add check to ensure that a message is only handled once per handler Add try...catch to run all handlers before throwing exception | Q | A | ------------- | --- | Branch? | master | Bug fix? | no | New feature? | yes | BC breaks? | no | Deprecations? |no | Tests pass? | yes | Fixed tickets | #27215 | License | MIT | Doc PR | Todo This would make error handling and retrying of messages much more easier. As statet here #27008 (comment) there is currently no way to retry a for all failed handlers if there are mutliple handlers and just some throw an exception. Also if an Exception in an handler occurs the execution chain is disrupted and the other handlers are never invoked. With this change it is easily possible to create an userland middleware that catches the `ChainedHandlerFailedException` and does some custom retry logic. If you ensure that the `HandledStamps` on the `Envelope` are preserved the message will be handled just by the failed handlers Commits ------- 2e5e910 Rename exception, add change log and a few other things e6e4cde Ensure message is handled only once per handler

carsonbot added Status: Needs Review Feature labels Apr 22, 2018

sroze added the Messenger label Apr 22, 2018

sroze commented Apr 22, 2018

View reviewed changes

sroze added this to the 4.1 milestone Apr 23, 2018

lyrixx reviewed Apr 23, 2018

View reviewed changes

nicolas-grekas changed the base branch from master to 4.1 May 7, 2018 15:06

sroze force-pushed the retry-amqp-messages branch 2 times, most recently from ca47435 to dfdb38c Compare May 9, 2018 08:27

ogizanagi previously approved these changes May 9, 2018

View reviewed changes

carsonbot added Status: Reviewed and removed Status: Needs Review labels May 9, 2018

kbond mentioned this pull request May 9, 2018

[RFC][Messenger] General message retry mechanism #27215

Closed

sroze force-pushed the retry-amqp-messages branch 2 times, most recently from 624ae8d to 4ae654b Compare May 10, 2018 15:13

sroze mentioned this pull request May 10, 2018

[Messenger] Add the transports documentation symfony/symfony-docs#9756

Closed

Tobion reviewed May 10, 2018

View reviewed changes

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php Outdated Show resolved Hide resolved

Tobion reviewed May 10, 2018

View reviewed changes

src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpReceiver.php Outdated Show resolved Hide resolved

sroze added this to the next milestone May 21, 2018

nicolas-grekas reviewed Jul 7, 2018

View reviewed changes

src/Symfony/Component/Messenger/Transport/AmqpExt/Connection.php Show resolved Hide resolved

sroze added 2 commits August 12, 2018 18:28

Add the method name change in the CHANGELOG

d32d7b6

sroze force-pushed the retry-amqp-messages branch from 88594f9 to d32d7b6 Compare August 12, 2018 17:32

sroze requested review from dunglas and xabbuh as code owners August 12, 2018 17:32

carsonbot added Status: Needs Review and removed Status: Reviewed labels Aug 12, 2018

sroze changed the base branch from 4.1 to master August 12, 2018 17:32

Tobion mentioned this pull request Nov 21, 2018

[Messenger] Add "--force-consumption" option to force the consumption of messages #29132

Closed

keulinho mentioned this pull request Jan 29, 2019

[Messenger] Ensure message is handled only once per handler #30020

Merged

weaverryan mentioned this pull request Mar 12, 2019

[Messenger] Making it Shine #30540

Closed

36 tasks

nicolas-grekas added Status: Needs Work and removed Status: Needs Review labels Mar 13, 2019

weaverryan mentioned this pull request Mar 15, 2019

[Messenger] Worker events + global retry functionality #30557

Merged

fabpot closed this Mar 17, 2019

nicolas-grekas modified the milestones: next, 4.3 Apr 30, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Messenger][AmqpExt] Add a retry mechanism for AMQP messages #27008

[Messenger][AmqpExt] Add a retry mechanism for AMQP messages #27008

sroze commented Apr 22, 2018 •

edited

Loading

sroze Apr 22, 2018

sroze Apr 22, 2018

sroze Apr 22, 2018

sroze Apr 22, 2018

javiereguiluz Apr 23, 2018

sroze Apr 23, 2018

sroze Apr 22, 2018

lyrixx left a comment •

edited

Loading

lyrixx Apr 23, 2018

sroze commented Apr 23, 2018

kbond commented Apr 26, 2018

nicolas-grekas commented May 4, 2018

sroze commented May 9, 2018

sroze commented May 9, 2018

sroze commented May 9, 2018

ogizanagi left a comment

ogizanagi May 9, 2018

sroze May 9, 2018

kbond commented May 9, 2018 •

edited

Loading

sroze commented May 9, 2018

sroze commented May 10, 2018

sroze commented May 10, 2018

Tobion May 10, 2018

sroze May 11, 2018

ogizanagi May 11, 2018 •

edited

Loading

sroze commented Aug 12, 2018

Koc commented Aug 12, 2018

sroze commented Aug 24, 2018

fabpot commented Oct 10, 2018

ragboyjr commented Dec 9, 2018

weaverryan commented Mar 15, 2019

lyrixx commented Mar 16, 2019

fabpot commented Mar 17, 2019

[Messenger][AmqpExt] Add a retry mechanism for AMQP messages #27008

[Messenger][AmqpExt] Add a retry mechanism for AMQP messages #27008

Conversation

sroze commented Apr 22, 2018 • edited Loading

So what's the point?

How to configure it

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lyrixx left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sroze commented Apr 23, 2018

kbond commented Apr 26, 2018

nicolas-grekas commented May 4, 2018

sroze commented May 9, 2018

sroze commented May 9, 2018

sroze commented May 9, 2018

ogizanagi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kbond commented May 9, 2018 • edited Loading

sroze commented May 9, 2018

sroze commented May 10, 2018

sroze commented May 10, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ogizanagi May 11, 2018 • edited Loading

Choose a reason for hiding this comment

sroze commented Aug 12, 2018

Koc commented Aug 12, 2018

sroze commented Aug 24, 2018

fabpot commented Oct 10, 2018

ragboyjr commented Dec 9, 2018

weaverryan commented Mar 15, 2019

lyrixx commented Mar 16, 2019

fabpot commented Mar 17, 2019

sroze commented Apr 22, 2018 •

edited

Loading

lyrixx left a comment •

edited

Loading

kbond commented May 9, 2018 •

edited

Loading

ogizanagi May 11, 2018 •

edited

Loading