Validate and process Handler.listen only once (#83400)

Fixes #83392

changelogs/fragments/83392-fix-memory-issues-handlers.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - Fix rapid memory usage growth when notifying handlers using the ``listen`` keyword (https://github.com/ansible/ansible/issues/83392)

lib/ansible/playbook/handler.py

@@ -37,6 +37,16 @@ class Handler(Task):
         ''' returns a human-readable representation of the handler '''
         return "HANDLER: %s" % self.get_name()
 
+    def _validate_listen(self, attr, name, value):
+        new_value = self.get_validated_value(name, attr, value, None)
+        if self._role is not None:
+            for listener in new_value.copy():
+                new_value.extend([
+                    f"{self._role.get_name(include_role_fqcn=True)} : {listener}",
+                    f"{self._role.get_name(include_role_fqcn=False)} : {listener}",
+                ])
+        setattr(self, name, new_value)
+
     @staticmethod
     def load(data, block=None, role=None, task_include=None, variable_manager=None, loader=None):
         t = Handler(block=block, role=role, task_include=task_include)

lib/ansible/plugins/strategy/__init__.py

@@ -549,27 +549,13 @@ class StrategyBase:
                 yield handler
                 break
 
-        templar.available_variables = {}
-        seen = []
+        seen = set()
         for handler in handlers:
-            if listeners := handler.listen:
-                listeners = handler.get_validated_value(
-                    'listen',
-                    handler.fattributes.get('listen'),
-                    listeners,
-                    templar,
-                )
-                if handler._role is not None:
-                    for listener in listeners.copy():
-                        listeners.extend([
-                            handler._role.get_name(include_role_fqcn=True) + ' : ' + listener,
-                            handler._role.get_name(include_role_fqcn=False) + ' : ' + listener
-                        ])
-                if notification in listeners:
-                    if handler.name and handler.name in seen:
-                        continue
-                    seen.append(handler.name)
-                    yield handler
+            if notification in handler.listen:
+                if handler.name and handler.name in seen:
+                    continue
+                seen.add(handler.name)
+                yield handler
 
     @debug_closure
     def _process_pending_results(self, iterator, one_pass=False, max_passes=None):