Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Messenger] Adding failure transport support #30970

Merged
merged 1 commit into from May 1, 2019

Conversation

@weaverryan
Copy link
Member

commented Apr 7, 2019

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

This adds "failure" transport support for messenger, so that messages that fail on all their retries can be collected in one spot and retried later if wanted:

framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: 'amqp://'
            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\SmsNotification': async

In this setup, SmsNotification would be retried 3 times on the async transport (current behavior) and then finally sent to the failed transport. The failed transport can be consumed like a normal transport, but should usually be handled & consumed by one of the new commands:

> bin/console messenger:failed:show
Screen Shot 2019-04-10 at 3 15 45 PM

> bin/console messenger:failed:show 217
Screen Shot 2019-04-10 at 3 15 55 PM

> bin/console messenger:failed:purge 217
Screen Shot 2019-04-10 at 3 16 07 PM

> bin/console messenger:failed:retry 217
Screen Shot 2019-04-10 at 3 16 29 PM

> bin/console messenger:failed:retry 218 -vv
Screen Shot 2019-04-10 at 3 20 39 PM

Note (This screenshot is ugly - need to make the dump of the message and the exception more attractive)

Or you can run bin/console messenger:failed:retry without any argument, and it will consume the failed messages one-by-one and ask you if you want to retry/handle each. By passing

Cheers!

@weaverryan weaverryan requested a review from sroze as a code owner Apr 7, 2019

@weaverryan weaverryan referenced this pull request Apr 7, 2019

Open

[Messenger][4.3] Tracker for changes #11236

1 of 25 tasks complete

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch from e63e80f to 3a6080f Apr 7, 2019

*/
public function __construct(array $senders, array $sendAndHandle = [])
public function __construct(array $sendersMap, ContainerInterface $sendersLocator, array $sendAndHandle = [])

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 7, 2019

Author Member

Instead of each sender class having its own service locator object of senders, this $sendersMap is now a simple array of string sender aliases (or service ids, if no alias) and $sendersLocator is a service locator that contains all senders.

@nicolas-grekas nicolas-grekas added this to the next milestone Apr 7, 2019

Show resolved Hide resolved src/Symfony/Component/Messenger/Event/WorkerMessageReceivedEvent.php Outdated
Show resolved Hide resolved src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php Outdated
*
* @experimental in 4.3
*/
class FailedMessagesRetryCommand extends Command

This comment has been minimized.

Copy link
@OskarStark

OskarStark Apr 8, 2019

Contributor

Should we make the commands final?

Show resolved Hide resolved ...t/Messenger/EventListener/SendFailedMessageToFailedTransportListener.php Outdated
Show resolved Hide resolved src/Symfony/Component/Messenger/Transport/Sender/SendersLocator.php Outdated

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch 3 times, most recently from 8e203fc to e151d3c Apr 10, 2019

@weaverryan weaverryan changed the title [Messenger][WIP] Adding failed transport support [Messenger] Adding failed transport support Apr 10, 2019

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch from 9085c63 to 9ccf3bc Apr 10, 2019

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented Apr 10, 2019

This is ready! And it works awesome. Here are some technical notes:

  1. SendersLocatorInterface changed... because for the first time, we need to look up the sender with more logic than just the message class - we need to look up by a specific alias name

  2. Sender aliases are still optional... but the system is increasingly dependent on them being available - retry, failure transport, etc.

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch 3 times, most recently from c7e613d to 16f2b32 Apr 10, 2019

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch from 16f2b32 to 04a985a Apr 13, 2019

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch 2 times, most recently from 2c448d8 to 905febb Apr 23, 2019

@weaverryan
Copy link
Member Author

left a comment

This is ready! I think it's important to get into 4.3. We're beyond feature freeze, but this component is experimental and this contains one more BC break. This also contains various bug fixes, which I've outlined.

$throwable = $throwable->getNestedExceptions()[0];
}
$flattenedException = \class_exists(FlattenException::class) ? FlattenException::createFromThrowable($throwable) : null;

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Using FlattenException was my way of making sure we had an Exception that was serializable. The original exception would be even better (as we already have logic to render this very "pretty" in the console), but I think this is the only safe way.

This comment has been minimized.

Copy link
@nicolas-grekas

nicolas-grekas Apr 23, 2019

Member

maybe at least pass file+line info when the Debug component is not found?
or maybe remove this and replace by the throwable cast as string?

public static function getSubscribedEvents()
{
return [
WorkerMessageFailedEvent::class => ['onMessageFailed', -100],

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Low priority so that users could handle it first if they wanted.

$envelope = $envelope->with(new BusNameStamp($this->busName));
if (null === $envelope->last(BusNameStamp::class)) {
$envelope = $envelope->with(new BusNameStamp($this->busName));
}

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Fixes an unrelated "bug" where retries would cause many BusNameStamp to be added, which was just unnecessary and wasteful. Test was added for this.

{
if (null !== $redeliveryStamp) {
return [
$redeliveryStamp->getSenderClassOrAlias() => $this->sendersLocator->getSenderByAlias($redeliveryStamp->getSenderClassOrAlias()),

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

For redelivery, the sender "alias" really becomes more important. This is communicated on the phpdoc of SendersLocatorInterface::getSenders(), which mentions that the iterator should be keyed by a sender alias, when possible.

@@ -66,7 +66,7 @@ public function __construct(int $maxRetries = 3, int $delayMilliseconds = 1000,
public function isRetryable(Envelope $message): bool
{
if (0 === $this->maxRetries) {
if (null === $this->maxRetries) {

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Unrelated: changes behavior slightly so that 0 can be used to mean "don't retry on this transport". Previously, there was no easy way to say "don't retry" as 0 meant "retry infinitely".

Test added for this.

public function getMessageCount(): int
{
return ($this->receiver ?? $this->getReceiver())->getMessageCount();
}

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Unrelated: the interface & this method were just missing.

Show resolved Hide resolved ...Symfony/Component/Messenger/Transport/Receiver/SingleMessageReceiver.php
@trigger_error(sprintf('The 2nd argument to "%s::__construct()" requires ContainerInterface 2nd argument. Not passing that was deprecated in Symfony 4.3 and will be required in Symfony 5.0.', __CLASS__), E_USER_DEPRECATED);
$this->sendersLocator = new ServiceLocator([]);
$this->sendAndHandle = $sendersLocator;
$this->useLegacyLookup = true;

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

BC added so that FW bundle 4.2 can work with Messenger 4.3 (otherwise cross-version tests fail)

*/
public function __construct(array $senders, array $sendAndHandle = [])
public function __construct(array $sendersMap, /*ContainerInterface*/ $sendersLocator = null, array $sendAndHandle = [])

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

This is really the "messiest" part of the PR. Because we need to send to the failure transport, identified by its "name", we need a way to ask the system: "please give me the sender named my_failed_transport". The new getSenderByAlias() accomplishes this. And it works great, but it was a "shift" in the system, as - until now - we only ever asked "Give me the senders for this envelope/message".

@@ -35,7 +36,8 @@
"symfony/var-dumper": "~3.4|~4.0"
},
"conflict": {
"symfony/event-dispatcher": "<4.3"
"symfony/event-dispatcher": "<4.3",
"symfony/debug": "<4.1"

This comment has been minimized.

Copy link
@weaverryan

weaverryan Apr 23, 2019

Author Member

Because, IF symfony/debug is available, we need the FlattenException::createFromThrowable() method.

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented Apr 24, 2019

Ready again!

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch 2 times, most recently from e6bb425 to 1692281 Apr 29, 2019

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented Apr 29, 2019

Thanks for the review @chalasr!

This is once again ready to go!

@chalasr
Copy link
Member

left a comment

Retested locally, works well now.
👍 once my last comment addressed

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch from 1692281 to a8923a9 Apr 30, 2019

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented Apr 30, 2019

Comment addressed!

@weaverryan weaverryan modified the milestones: next, 4.3 Apr 30, 2019

@weaverryan weaverryan force-pushed the weaverryan:messenger-reject-transport branch from a8923a9 to dd1edf3 Apr 30, 2019

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented Apr 30, 2019

Test failures are unrelated. Ready for review!

@fabpot fabpot force-pushed the weaverryan:messenger-reject-transport branch from dd1edf3 to 36487e5 May 1, 2019

@fabpot

This comment has been minimized.

Copy link
Member

commented May 1, 2019

Thank you @weaverryan.

@fabpot fabpot merged commit 36487e5 into symfony:master May 1, 2019

1 of 3 checks passed

continuous-integration/appveyor/pr Waiting for AppVeyor build to complete
Details
continuous-integration/travis-ci/pr The Travis CI build is in progress
Details
fabbot.io Your code looks good.
Details

fabpot added a commit that referenced this pull request May 1, 2019

feature #30970 [Messenger] Adding failure transport support (weaverryan)
This PR was squashed before being merged into the 4.3-dev branch (closes #30970).

Discussion
----------

[Messenger] Adding failure transport support

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

This adds "failure" transport support for messenger, so that messages that fail on *all* their retries can be collected in one spot and retried later if wanted:

```yml
framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: 'amqp://'
            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\SmsNotification': async
```

In this setup, `SmsNotification` would be retried 3 times on the `async` transport (current behavior) and then finally sent to the `failed` transport. The `failed` transport can be consumed like a normal transport, but should usually be handled & consumed by one of the new commands:

**> bin/console messenger:failed:show**
<img width="861" alt="Screen Shot 2019-04-10 at 3 15 45 PM" src="https://user-images.githubusercontent.com/121003/55917329-ddc54280-5ba3-11e9-878c-af3c653643de.png">

**> bin/console messenger:failed:show 217**
<img width="804" alt="Screen Shot 2019-04-10 at 3 15 55 PM" src="https://user-images.githubusercontent.com/121003/55917360-f33a6c80-5ba3-11e9-9f12-a8c57a9a7a4b.png">

**> bin/console messenger:failed:purge 217**
<img width="835" alt="Screen Shot 2019-04-10 at 3 16 07 PM" src="https://user-images.githubusercontent.com/121003/55917383-ff262e80-5ba3-11e9-9720-e24176b834f7.png">

**> bin/console messenger:failed:retry 217**
<img width="737" alt="Screen Shot 2019-04-10 at 3 16 29 PM" src="https://user-images.githubusercontent.com/121003/55917396-09482d00-5ba4-11e9-8d51-0bbe2b4ffc14.png">

**> bin/console messenger:failed:retry 218 -vv**
<img width="1011" alt="Screen Shot 2019-04-10 at 3 20 39 PM" src="https://user-images.githubusercontent.com/121003/55917503-6512b600-5ba4-11e9-9365-4ac87d858541.png">

**Note** (This screenshot is ugly - need to make the dump of the message and the exception more attractive)

Or you can run `bin/console messenger:failed:retry` without any argument, and it will consume the failed messages one-by-one and ask you if you want to retry/handle each. By passing

Cheers!

Commits
-------

36487e5 [Messenger] Adding failure transport support

@weaverryan weaverryan deleted the weaverryan:messenger-reject-transport branch May 1, 2019

@Tobion

This comment has been minimized.

Copy link
Member

commented May 5, 2019

If I see it correctly, the messenger:failed:show command cannot show messages when rabbitmq is used as a failed transport because it does not implement the listing and getting individual messages.
But in theory that could still be possible by consuming all messages and then putting them back into the queue. Was that considered?

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented May 5, 2019

Hi @Tobion!

Yes, it’s something i talked about with Sam, but wasn’t ultimately sure about. The problem is if there are many messages. Consuming them all, just to list them and put them back could be a big operation. I don’t have any experience in how rabbit behaves in this way. I’d love to hear any experiences related to this.

Another consideration is that (from feedback I’ve heard) people could theoretically have many failed messages (due to a 3rd party service failure, for example) and want to retry all of them, or maybe all failed messages within a date range. Retrying all of them would work fine, but doing it by date range would require the “consume all” strategy.

Finally, one other unrelated thing from feedback: retry should perhaps just requeue. This is especially important if we offer some sort of “bulk” retry/requeue: requeuing them and allowing their (potentially many) workers on the original queue do their work normally makes more sense then retrying them one-by-one of the server where someone is manually looking at and retrying the failed messages.

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented May 5, 2019

Also, generally speaking, I’m not sure if Amqp is a good candidate for the failure transport. The failure transport could get “large” - even in a normal app - if you rarely check it and handle those messages. That could hurt Amqp performance - it doesn’t (as I understand it) like to have queues with many messages.

@Tobion

This comment has been minimized.

Copy link
Member

commented May 5, 2019

Hey @weaverryan. I love this feature.I think if we provide a way to configure the failed storage, then we should also try our best to make them all feature complete.

The problem is if there are many messages. Consuming them all, just to list them and put them back could be a big operation.

I don't think it's very common to have many failed messages after retry. But as a sanity check, we could just do the listing if the amount of messages is below a threshold and ask for confirmation if not.

retry should perhaps just requeue

There are use-cases for both approaches in my opinion. You either just want to retry one and see if it works now or why it doesn't work. Or the queue is full and you don't want to wait.
But requeing also makes sense when there are too many messages and you want to leverage your workers.
So ideally both solutions would be offered.

@weaverryan

This comment has been minimized.

Copy link
Member Author

commented May 5, 2019

I'm going to work on a PR for the requeue. When I've open that, let's continue discussing there to make sure we're happy with everything. I also like the "retry" for at least some situations (just retry it right now and show me if it worked!).

public function get(): iterable
{
if ($this->hasReceived) {
return [];

This comment has been minimized.

Copy link
@Tobion

Tobion May 5, 2019

Member

since this method uses a generator, it should not return array. but just return;

This comment has been minimized.

Copy link
@Tobion

Tobion May 5, 2019

Member

actually it does not seem worth to use a generator here, but just return the envolope in an array.

This comment has been minimized.

Copy link
@Tobion
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.