[Messenger] Move container resetting after receiver acknowledging #43133

upyx · 2021-09-22T09:11:48Z

Q	A
Branch?	5.4
Bug fix?	yes
New feature?	no
Deprecations?	no
Tickets	Fix #43114
License	MIT
Doc PR	nope

It fixes the #43114 issue and similar potential bugs. The ResetServicesListener lean on events sequence: the AbstractWorkerMessageEvent to save the actual receiver name, then the WorkerRunningEvent to check that name and reset the container. It may look fragile, so let's discuss it in the issue.

okwinza · 2021-09-22T11:53:13Z

Waiting until the worker is completely done handling a message before actually resetting services sounds good to me. 👍

Things to keep in mind:

Services should only be reset after the worker is done with a message.
Services should not be reset more once between each message.

As for the implementation part, this could also benefit from #42335 to easily check the worker transport name.
So you could just do something like this:

$event->getWorker()->getWorkerMetadata()->getTransportName() === $this->transportName;

…kwinza) This PR was merged into the 5.4 branch. Discussion ---------- [Messenger] Add `WorkerMetadata` to `Worker` class. | Q | A | ------------- | --- | Branch? | 5.4 | Bug fix? | no | New feature? | yes | Deprecations? | no | Tickets | Fixes #37736 | License | MIT | Doc PR | - At the moment, there is no clean way to access the values of `transportNames` or recently introduced `queueNames` that the worker was configured with, although such data might be quite useful for logging/monitoring or other tasks. This PR attempts to fix that by adding a new and extensible way to provide additional information about a particular `Worker` object. So far, the following PRs could benefit from this change: - #43133 - #42723 **Use case example:** ---- - As I developer - When a message was consumed from transport with name `async`. - And the worker state is `idle`. - Then I want to reset services. **Before this PR**, the only solution not relying on using Reflection API would look like this: ```php private $servicesResetter; private $receiversName; private $actualReceiverName = null; public function __construct(ServicesResetter $servicesResetter, array $receiversName) { $this->servicesResetter = $servicesResetter; $this->receiversName = $receiversName; } public function saveReceiverName(AbstractWorkerMessageEvent $event): void { $this->actualReceiverName = $event->getReceiverName(); } public function resetServices(WorkerRunningEvent $event): void { if (!$event->isWorkerIdle() && \in_array($this->actualReceiverName, $this->receiversName, true)) { $this->servicesResetter->reset(); } $this->actualReceiverName = null; } public static function getSubscribedEvents(): array { return [ WorkerMessageHandledEvent::class => ['saveReceiverName'], WorkerMessageFailedEvent::class => ['saveReceiverName'], WorkerRunningEvent::class => ['resetServices'], ]; } ``` **With this PR**, one could simply use this to retrieve the transport name. ```php $event->getWorker()->getWorkerMetadata()->getTransportName() === $this->transportName; ``` So the whole solution would look like this: ```php private $servicesResetter; private $receiversName; public function __construct(ServicesResetter $servicesResetter, array $receiversName) { $this->servicesResetter = $servicesResetter; $this->receiversName = $receiversName; } public function resetServices(WorkerRunningEvent $event): void { $actualTransportName = $event->getWorker()->getWorkerMetadata()->getTransportName(); if (!$event->isWorkerIdle() || !in_array($actualTransportName, $this->receiversName, true)) { return; } $this->servicesResetter->reset(); } public static function getSubscribedEvents(): array { return [ WorkerRunningEvent::class => ['resetServices'], ]; } ``` Commits ------- 583f85b [Messenger] Add WorkerMetadata to Worker class

lyrixx · 2021-09-22T14:56:56Z

I just merged #42335. Can you rebase to use the new feature? Thanks

And please, fix fabpot issue :)

upyx · 2021-09-24T10:46:37Z

$event->getWorker()->getWorkerMetadata()->getTransportName() === $this->transportName;

@okwinza
It won't work as expected because the WorkerMetadata has an array of transport names. The part of transports can be not resettable. We can either restrict using resettable and not resettable transports in the same worker or rely on the event's sequence. I'd prefer to use new metadata, and I think it will be good to restrict resetting services on a receiver when the different one doesn't expect that behavior.

So it'll take a while to implement.

lyrixx · 2021-09-24T12:01:37Z

What about moving the ack before the dispatch ?

Edit: this is a bad idea. But I think the best solution is to create a new event (same for nack)

upyx · 2021-09-24T20:59:15Z

@lyrixx:

What about moving the ack before the dispatch ?

Edit: this is a bad idea. But I think the best solution is to create a new event (same for nack)

We have the event already. We have metadata, transport names, and other things.

I've been investigating the topic more. I've realized that resetting services is a worker's job rather than a transport's because service leak is a consumer problem and isn't dependent on transport. And there aren't any consumers on the "sync" transport.

In my opinion, the best alternative is resetting services after every message. The only drawback is decreasing performance, so we can provide an option to disable it. It also will break compatibility, so we should enable that option by default and add an inversed one.

I consider removing the reset_on_message parameter from a configuration and adding two options to the messenger:consume command: --reset-services-on-message and --no-reset-services-on-message. It's also good to deprecate running the command without explicitly indicating which one to use and change defaults in 6.0.

upyx · 2021-09-24T23:52:26Z

I've made a working draft as an example. If it's OK, I'll continue.

lyrixx

I like it. But I realize we can simplify the work for end user.
We can move the command configuration to the bundle:
Like that, user don't have to update all consumer recipies (ansible, helm, etc). Only the code needs to be updated.

framework:
    messenger:
        reset_on_message: true # add a deprecation if not set
        transports: ....

WDYT?

I agree with you on the fact that is should be the default, and I don't see a use case where a transport should be reset but not another one.

And BTW Fab wanted to reset the container all the time too: #41163 (comment)

src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php

okwinza · 2021-09-25T09:48:17Z

framework:
    messenger:
        reset_on_message: true # add a deprecation if not set
        transports: ....

Can't we just set it to false by default for BC and then change it to true and make it mandatory in 6.0?
Also it might be useful to add a runtime --no-reset option to messenger:consume as well.
WDYT?

lyrixx · 2021-09-25T11:08:35Z

Can't we just set it to false by default for BC and then change it to true and make it mandatory in 6.0?

it's not the best way to handle migration path.
5.4: deprecation if not set, telling it'll be true in 6.0, default to false for BC
6.0: default to true, no deprecation, optin

Also it might be useful to add a runtime --no-reset option to messenger:consume as well.

do you have a use case for overriding the value? (less code == less bug)

okwinza · 2021-09-25T12:59:55Z

5.4: deprecation if not set, telling it'll be true in 6.0, default to false for BC
6.0: default to true, no deprecation, optin

👍 Agreed.

Also it might be useful to add a runtime --no-reset option to messenger:consume as well.

do you have a use case for overriding the value? (less code == less bug)

Since container resetting affects conformance, it might make sense to be able to override it for additional performance gain in some particular consumers.

But this is just an idea and it's totally out of scope of this PR and should be added later down the road, if at all.

lyrixx · 2021-09-25T13:50:01Z

Okay let's go 👍 do you wanna to finish the pr ?

Thanks for discussion 😊

upyx · 2021-09-25T15:52:28Z

@lyrixx
The only use case for disabling service resetting is to gain performance for specific consumers; therefore, the option should be in a consumer but not in configuration.

less code == less bug

Absolutely! So we do not need a configuration option (in prospect). What we need are backward compatibility and migration strategy. Some users may use "service leak" as a side effect for their scenarios; however, it's a poor solution. So the decision we should make: how to migrate it.

I think you are right about a deployment. A new plan:
5.4:

add the messenger.reset_on_message config with default false
deprecate empty messenger.reset_on_message with hint about defaults
add --no-reset (or --no-reset-services or --no-reset-services-on-message ?) option to messenger:consume command.

6.0:

change default of messenger.reset_on_message to true
deprecate messenger.reset_on_message to remove it in 7.0

@lyrixx @okwinza do you agree?

… command)

upyx · 2021-09-26T13:54:45Z

I've realized that services must not be reset on idle. It's not optional because of counters and sequences that can be used. Possible memory leaks in transports should be fixed in transports.

lyrixx · 2021-09-27T08:36:41Z

I've realized that services must not be reset on idle. It's not optional because of counters and sequences that can be used.

What are you refering to?

I think something is not clear here. you (@upyx and @okwinza) talks about performance penalty. Even if this could be true, this is not always the case. I'm not sure, but it looks like you think all symfony (and app) services will be unset. This is not the case. Service will be reset, it means the ServicesResetter service will take all services tagged kernel.reset and call reset() method on it. On the current application I'm working one (SF 5.3, APIP, Doctrine). There are 41 services impacted in dev.

For reference:

 ---------------------------------------------------------- ---------------------- ----------------------------------------------------------------------------------------
  Service ID                                                 method                 Class name                                                                             
 ---------------------------------------------------------- ---------------------- ----------------------------------------------------------------------------------------
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx        reset                  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  Symfony\Bridge\Monolog\Processor\ConsoleCommandProcessor   reset                  Symfony\Bridge\Monolog\Processor\ConsoleCommandProcessor
  Symfony\Component\Cache\Psr16Cache                         reset                  Symfony\Component\Cache\Psr16Cache
  cache.app                                                  reset                  Symfony\Component\Cache\Adapter\FilesystemAdapter
  cache.system                                               reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  cache.validator                                            reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  cache.serializer                                           reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  cache.annotations                                          reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  cache.property_info                                        reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  cache.messenger.restart_workers_signal                     reset                  Symfony\Component\Cache\Adapter\FilesystemAdapter
  agent_rules.cache                                          reset                  Symfony\Component\Cache\Adapter\RedisAdapter
  cache.array                                                reset                  Symfony\Component\Cache\Adapter\ArrayAdapter
  doctrine.result_cache_pool                                 reset                  Symfony\Component\Cache\Adapter\FilesystemAdapter
  doctrine.system_cache_pool                                 reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  form.choice_list_factory.cached                            reset                  Symfony\Component\Form\ChoiceList\Factory\CachingFactoryDecorator
  messenger.transport.in_memory.factory                      reset                  Symfony\Component\Messenger\Transport\InMemoryTransportFactory
  cache.validator_expression_language                        reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  debug.stopwatch                                            reset                  Symfony\Component\Stopwatch\Stopwatch
  security.token_storage                                     disableUsageTracking   Symfony\Component\Security\Core\Authentication\Token\Storage\UsageTrackingTokenStorage
   (same service as previous, another tag)                   setToken
  cache.security_expression_language                         reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  monolog.handler.main                                       reset                  Monolog\Handler\FingersCrossedHandler
  monolog.handler.console                                    reset                  Symfony\Bridge\Monolog\Handler\ConsoleHandler
  webpack_encore.tag_renderer                                reset                  Symfony\WebpackEncoreBundle\Asset\TagRenderer
  cache.webpack_encore                                       reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  webpack_encore.entrypoint_lookup[_default]                 reset                  Symfony\WebpackEncoreBundle\Asset\EntrypointLookup
  doctrine                                                   reset                  Doctrine\Bundle\DoctrineBundle\Registry
  form.type.entity                                           reset                  Symfony\Bridge\Doctrine\Form\Type\EntityType
  cache.easyadmin                                            reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  api_platform.cache.route_name_resolver                     reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  api_platform.cache.identifiers_extractor                   reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  api_platform.cache.subresource_operation_factory           reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  api_platform.cache.metadata.resource                       reset                  Symfony\Component\Cache\Adapter\AdapterInterface
  api_platform.cache.metadata.property                       reset                  Symfony\Component\Cache\Adapter\AdapterInterface
 ---------------------------------------------------------- ---------------------- ----------------------------------------------------------------------------------------

upyx · 2021-09-27T11:25:40Z

@lyrixx
I meant that what services do on reset. It can be a custom service that increases some counter on every message. For example, Monolog's UidProcessor generates random UID on every reset. I'm not too fond of randoms (because of collisions), so I'd like to store a sequence in APCu or Redis. It isn't a hot queue (5 messages per day), and I'd like to make numbers as short as possible, and the sequence is short. So forwarding sequence on idle would quickly lead to sequence overflow. It's a simplified example just for clarity. Performance isn't an issue when the queue is idle. We mustn't reset services on idle not because of the cost but because of side effects.

I think it would be good to add documentation that describes what the "services resetting" is and what it is for, how to use the kernel.reset tag and the ResetInterface, where to enable autoconfigure, and where don't.

lyrixx · 2021-09-27T12:16:44Z

so I'd like to store a sequence in APCu or Redis

if the sequence is in APCu or Redis, it's safe.

IMHO, store a state between 2 messages in PHP memory (so in a service) is a bad idea. It does not work when a worker restart or when there are many worker.

So this is not a valid use case since it's harmful.

The plan you propose in this comment is really good. Let's stick to it.

lyrixx

Very good work. Thanks 👍🏼

fabpot · 2021-09-27T15:29:09Z

Thank you @upyx.

upyx · 2021-09-27T15:46:54Z

@lyrixx @fabpot you are welcome 🤝 🤝

Should I make another PR to the 6.0 branch in which deprecations are removed?

kbond · 2021-09-27T16:00:48Z

src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php

+        if (null === $config['reset_on_message']) {
+            trigger_deprecation('symfony/framework-bundle', '5.4', 'Not setting the "framework.messenger.reset_on_message" configuration option is deprecated, it will default to "true" in version 6.0.');
+
+            $container->getDefinition('console.command.messenger_consume_messages')->replaceArgument(5, null);


I will debug further and create a proper issue/PR later but this service may not exist:

Symfony\Component\DependencyInjection\Exception\ServiceNotFoundException: You have requested a non-existent service "console.command.messenger_consume_messages".

Yep, I've found where it is. It's possible when messenger is installed without console. I will fix it, but I don't know how to test that fix because of class_exists(Application::class) expression.

lyrixx · 2021-09-27T17:06:04Z

Should I make another PR to the 6.0 branch in which deprecations are removed?

yes, and a PR on the doc too :)

upyx · 2021-09-27T18:11:19Z

Should I make another PR to the 6.0 branch in which deprecations are removed?

yes, and a PR on the doc too :)

#43202 has some fixes
#43203 is a draft for 6.0

…nsole (upyx) This PR was squashed before being merged into the 5.4 branch. Discussion ---------- Fix framework configuration when messenger uses without console | Q | A | ------------- | --- | Branch? | 5.4 | Bug fix? | yes | New feature? | no | Deprecations? | yes | Tickets | Fix #43133 (review) | License | MIT | Doc PR | later We hurried, an addition to #43133 It fixes #43133 (review) It adds the forgotten CHANGELOG & UPGRADE I'm not sure about the CHANGELOG format. I've done as people do but... [it says](https://symfony.com/doc/current/contributing/code/conventions.html#writing-a-changelog-entry) "New entries must be added on top of the list." However, nobody does it 🤷 ~I don't know how to test the fix, because of `class_exists()` call that cannot be easily mocked.~ `src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php:245`: ```php if (class_exists(Application::class)) { $loader->load('console.php'); /* ... */ } ``` Commits ------- d098ef8 Fix framework configuration when messenger uses without console

upyx requested a review from sroze as a code owner September 22, 2021 09:11

carsonbot added Status: Needs Review Bug Messenger labels Sep 22, 2021

carsonbot added this to the 5.4 milestone Sep 22, 2021

okwinza mentioned this pull request Sep 22, 2021

[Messenger] Add WorkerMetadata to Worker class. #42335

Merged

upyx marked this pull request as draft September 24, 2021 21:15

upyx force-pushed the fix-container-resetting branch 2 times, most recently from 987baef to 214a63a Compare September 24, 2021 23:31

lyrixx reviewed Sep 25, 2021

View reviewed changes

src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php Outdated Show resolved Hide resolved

upyx force-pushed the fix-container-resetting branch from 214a63a to baa42f9 Compare September 25, 2021 18:11

upyx marked this pull request as ready for review September 25, 2021 18:25

upyx force-pushed the fix-container-resetting branch from baa42f9 to b75dd03 Compare September 25, 2021 18:25

[Messenger] Move container resetting after receiver acknowledging (in…

fc85aef

… command)

upyx force-pushed the fix-container-resetting branch from b75dd03 to fc85aef Compare September 26, 2021 13:45

lyrixx approved these changes Sep 27, 2021

View reviewed changes

carsonbot added Status: Reviewed and removed Status: Needs Review labels Sep 27, 2021

fabpot approved these changes Sep 27, 2021

View reviewed changes

fabpot merged commit 8653b33 into symfony:5.4 Sep 27, 2021

kbond reviewed Sep 27, 2021

View reviewed changes

upyx mentioned this pull request Sep 27, 2021

Fix framework configuration when messenger uses without console #43202

Merged

upyx mentioned this pull request Sep 30, 2021

[FrameworkBundle] Make the messenger.reset_on_message config option default to true #43203

Merged

lyrixx mentioned this pull request Oct 1, 2021

[RFC][Messenger] Remove ResetLoggersWorkerSubscriber #43271

Closed

upyx deleted the fix-container-resetting branch October 6, 2021 22:59

GromNaN mentioned this pull request Nov 3, 2021

[Messenger] Add completion to command messenger:consume #43891

Merged

upyx mentioned this pull request Mar 21, 2022

[FrameworkBundle] Deprecate the messenger.reset_on_message config option #45705

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Messenger] Move container resetting after receiver acknowledging #43133

[Messenger] Move container resetting after receiver acknowledging #43133

upyx commented Sep 22, 2021

okwinza commented Sep 22, 2021 •

edited

Loading

lyrixx commented Sep 22, 2021 •

edited

Loading

upyx commented Sep 24, 2021

lyrixx commented Sep 24, 2021 •

edited

Loading

upyx commented Sep 24, 2021

upyx commented Sep 24, 2021

lyrixx left a comment •

edited

Loading

okwinza commented Sep 25, 2021

lyrixx commented Sep 25, 2021

okwinza commented Sep 25, 2021 •

edited

Loading

lyrixx commented Sep 25, 2021

upyx commented Sep 25, 2021 •

edited

Loading

upyx commented Sep 26, 2021

lyrixx commented Sep 27, 2021

upyx commented Sep 27, 2021 •

edited

Loading

lyrixx commented Sep 27, 2021

lyrixx left a comment

fabpot commented Sep 27, 2021

upyx commented Sep 27, 2021

kbond Sep 27, 2021

upyx Sep 27, 2021

lyrixx commented Sep 27, 2021

upyx commented Sep 27, 2021

[Messenger] Move container resetting after receiver acknowledging #43133

[Messenger] Move container resetting after receiver acknowledging #43133

Conversation

upyx commented Sep 22, 2021

okwinza commented Sep 22, 2021 • edited Loading

lyrixx commented Sep 22, 2021 • edited Loading

upyx commented Sep 24, 2021

lyrixx commented Sep 24, 2021 • edited Loading

upyx commented Sep 24, 2021

upyx commented Sep 24, 2021

lyrixx left a comment • edited Loading

Choose a reason for hiding this comment

okwinza commented Sep 25, 2021

lyrixx commented Sep 25, 2021

okwinza commented Sep 25, 2021 • edited Loading

lyrixx commented Sep 25, 2021

upyx commented Sep 25, 2021 • edited Loading

upyx commented Sep 26, 2021

lyrixx commented Sep 27, 2021

upyx commented Sep 27, 2021 • edited Loading

lyrixx commented Sep 27, 2021

lyrixx left a comment

Choose a reason for hiding this comment

fabpot commented Sep 27, 2021

upyx commented Sep 27, 2021

kbond Sep 27, 2021

Choose a reason for hiding this comment

upyx Sep 27, 2021

Choose a reason for hiding this comment

lyrixx commented Sep 27, 2021

upyx commented Sep 27, 2021

okwinza commented Sep 22, 2021 •

edited

Loading

lyrixx commented Sep 22, 2021 •

edited

Loading

lyrixx commented Sep 24, 2021 •

edited

Loading

lyrixx left a comment •

edited

Loading

okwinza commented Sep 25, 2021 •

edited

Loading

upyx commented Sep 25, 2021 •

edited

Loading

upyx commented Sep 27, 2021 •

edited

Loading