Last handler defined runs, fix for roles (#79558) (#80495)

Fixes #73643
* clear_notification method and simplify ifs
* Deduplicate code
* Limit number of Templar creations
* Fix sanity
* Preserve handler callbacks order as they were notified

(cherry picked from commit 09dd80b4ec7563caea095b4213794dd64ce8bed4)

8 files changed, 126 insertions, 84 deletions

diff --git a/changelogs/fragments/73643-handlers-prevent-multiple-runs.yml b/changelogs/fragments/73643-handlers-prevent-multiple-runs.yml
new file mode 100644
index 0000000000..2cb132ddf9
--- /dev/null
+++ b/changelogs/fragments/73643-handlers-prevent-multiple-runs.yml
@@ -0,0 +1,2 @@
+bugfixes:
+  - Prevent running same handler multiple times when included via ``include_role`` (https://github.com/ansible/ansible/issues/73643)
diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index 2449782165..f36be5af52 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -60,6 +60,8 @@ class HostState:
         self._blocks = blocks[:]
         self.handlers = []
 
+        self.handler_notifications = []
+
         self.cur_block = 0
         self.cur_regular_task = 0
         self.cur_rescue_task = 0
@@ -120,6 +122,7 @@ class HostState:
     def copy(self):
         new_state = HostState(self._blocks)
         new_state.handlers = self.handlers[:]
+        new_state.handler_notifications = self.handler_notifications[:]
         new_state.cur_block = self.cur_block
         new_state.cur_regular_task = self.cur_regular_task
         new_state.cur_rescue_task = self.cur_rescue_task
@@ -650,3 +653,12 @@ class PlayIterator:
         if not isinstance(fail_state, FailedStates):
             raise AnsibleAssertionError('Expected fail_state to be a FailedStates but was %s' % (type(fail_state)))
         self._host_states[hostname].fail_state = fail_state
+
+    def add_notification(self, hostname: str, notification: str) -> None:
+        # preserve order
+        host_state = self._host_states[hostname]
+        if notification not in host_state.handler_notifications:
+            host_state.handler_notifications.append(notification)
+
+    def clear_notification(self, hostname: str, notification: str) -> None:
+        self._host_states[hostname].handler_notifications.remove(notification)
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index cce387b57a..7b80890003 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -27,6 +27,7 @@ import queue
 import sys
 import threading
 import time
+import typing as t
 
 from collections import deque
 from multiprocessing import Lock
@@ -37,7 +38,7 @@ from ansible import constants as C
 from ansible import context
 from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleUndefinedVariable, AnsibleParserError
 from ansible.executor import action_write_locks
-from ansible.executor.play_iterator import IteratingStates
+from ansible.executor.play_iterator import IteratingStates, PlayIterator
 from ansible.executor.process.worker import WorkerProcess
 from ansible.executor.task_result import TaskResult
 from ansible.executor.task_queue_manager import CallbackSend, DisplaySend, PromptSend
@@ -506,6 +507,57 @@ class StrategyBase:
 
         return task_result
 
+    def search_handlers_by_notification(self, notification: str, iterator: PlayIterator) -> t.Generator[Handler, None, None]:
+        templar = Templar(None)
+        # iterate in reversed order since last handler loaded with the same name wins
+        for handler in (h for b in reversed(iterator._play.handlers) for h in b.block if h.name):
+            if not handler.cached_name:
+                if templar.is_template(handler.name):
+                    templar.available_variables = self._variable_manager.get_vars(
+                        play=iterator._play,
+                        task=handler,
+                        _hosts=self._hosts_cache,
+                        _hosts_all=self._hosts_cache_all
+                    )
+                    try:
+                        handler.name = templar.template(handler.name)
+                    except (UndefinedError, AnsibleUndefinedVariable) as e:
+                        # We skip this handler due to the fact that it may be using
+                        # a variable in the name that was conditionally included via
+                        # set_fact or some other method, and we don't want to error
+                        # out unnecessarily
+                        if not handler.listen:
+                            display.warning(
+                                "Handler '%s' is unusable because it has no listen topics and "
+                                "the name could not be templated (host-specific variables are "
+                                "not supported in handler names). The error: %s" % (handler.name, to_text(e))
+                            )
+                        continue
+                handler.cached_name = True
+
+            # first we check with the full result of get_name(), which may
+            # include the role name (if the handler is from a role). If that
+            # is not found, we resort to the simple name field, which doesn't
+            # have anything extra added to it.
+            if notification in {
+                handler.name,
+                handler.get_name(include_role_fqcn=False),
+                handler.get_name(include_role_fqcn=True),
+            }:
+                yield handler
+                break
+
+        templar.available_variables = {}
+        for handler in (h for b in iterator._play.handlers for h in b.block):
+            if listeners := handler.listen:
+                if notification in handler.get_validated_value(
+                    'listen',
+                    handler.fattributes.get('listen'),
+                    listeners,
+                    templar,
+                ):
+                    yield handler
+
     @debug_closure
     def _process_pending_results(self, iterator, one_pass=False, max_passes=None):
         '''
@@ -516,46 +568,6 @@ class StrategyBase:
         ret_results = []
         handler_templar = Templar(self._loader)
 
-        def search_handler_blocks_by_name(handler_name, handler_blocks):
-            # iterate in reversed order since last handler loaded with the same name wins
-            for handler_block in reversed(handler_blocks):
-                for handler_task in handler_block.block:
-                    if handler_task.name:
-                        try:
-                            if not handler_task.cached_name:
-                                if handler_templar.is_template(handler_task.name):
-                                    handler_templar.available_variables = self._variable_manager.get_vars(play=iterator._play,
-                                                                                                          task=handler_task,
-                                                                                                          _hosts=self._hosts_cache,
-                                                                                                          _hosts_all=self._hosts_cache_all)
-                                    handler_task.name = handler_templar.template(handler_task.name)
-                                handler_task.cached_name = True
-
-                            # first we check with the full result of get_name(), which may
-                            # include the role name (if the handler is from a role). If that
-                            # is not found, we resort to the simple name field, which doesn't
-                            # have anything extra added to it.
-                            candidates = (
-                                handler_task.name,
-                                handler_task.get_name(include_role_fqcn=False),
-                                handler_task.get_name(include_role_fqcn=True),
-                            )
-
-                            if handler_name in candidates:
-                                return handler_task
-                        except (UndefinedError, AnsibleUndefinedVariable) as e:
-                            # We skip this handler due to the fact that it may be using
-                            # a variable in the name that was conditionally included via
-                            # set_fact or some other method, and we don't want to error
-                            # out unnecessarily
-                            if not handler_task.listen:
-                                display.warning(
-                                    "Handler '%s' is unusable because it has no listen topics and "
-                                    "the name could not be templated (host-specific variables are "
-                                    "not supported in handler names). The error: %s" % (handler_task.name, to_text(e))
-                                )
-                            continue
-
         cur_pass = 0
         while True:
             try:
@@ -636,49 +648,24 @@ class StrategyBase:
                     result_items = [task_result._result]
 
                 for result_item in result_items:
-                    if '_ansible_notify' in result_item:
-                        if task_result.is_changed():
-                            # The shared dictionary for notified handlers is a proxy, which
-                            # does not detect when sub-objects within the proxy are modified.
-                            # So, per the docs, we reassign the list so the proxy picks up and
-                            # notifies all other threads
-                            for handler_name in result_item['_ansible_notify']:
-                                found = False
-                                # Find the handler using the above helper.  First we look up the
-                                # dependency chain of the current task (if it's from a role), otherwise
-                                # we just look through the list of handlers in the current play/all
-                                # roles and use the first one that matches the notify name
-                                target_handler = search_handler_blocks_by_name(handler_name, iterator._play.handlers)
-                                if target_handler is not None:
-                                    found = True
-                                    if target_handler.notify_host(original_host):
-                                        self._tqm.send_callback('v2_playbook_on_notify', target_handler, original_host)
-
-                                for listening_handler_block in iterator._play.handlers:
-                                    for listening_handler in listening_handler_block.block:
-                                        listeners = getattr(listening_handler, 'listen', []) or []
-                                        if not listeners:
-                                            continue
-
-                                        listeners = listening_handler.get_validated_value(
-                                            'listen', listening_handler.fattributes.get('listen'), listeners, handler_templar
-                                        )
-                                        if handler_name not in listeners:
-                                            continue
-                                        else:
-                                            found = True
-
-                                        if listening_handler.notify_host(original_host):
-                                            self._tqm.send_callback('v2_playbook_on_notify', listening_handler, original_host)
-
-                                # and if none were found, then we raise an error
-                                if not found:
-                                    msg = ("The requested handler '%s' was not found in either the main handlers list nor in the listening "
-                                           "handlers list" % handler_name)
-                                    if C.ERROR_ON_MISSING_HANDLER:
-                                        raise AnsibleError(msg)
-                                    else:
-                                        display.warning(msg)
+                    if '_ansible_notify' in result_item and task_result.is_changed():
+                        # only ensure that notified handlers exist, if so save the notifications for when
+                        # handlers are actually flushed so the last defined handlers are exexcuted,
+                        # otherwise depending on the setting either error or warn
+                        for notification in result_item['_ansible_notify']:
+                            if any(self.search_handlers_by_notification(notification, iterator)):
+                                iterator.add_notification(original_host.name, notification)
+                                display.vv(f"Notification for handler {notification} has been saved.")
+                                continue
+
+                            msg = (
+                                f"The requested handler '{notification}' was not found in either the main handlers"
+                                " list nor in the listening handlers list"
+                            )
+                            if C.ERROR_ON_MISSING_HANDLER:
+                                raise AnsibleError(msg)
+                            else:
+                                display.warning(msg)
 
                     if 'add_host' in result_item:
                         # this task added a new host (add_host module)
@@ -957,6 +944,15 @@ class StrategyBase:
         elif meta_action == 'flush_handlers':
             if _evaluate_conditional(target_host):
                 host_state = iterator.get_state_for_host(target_host.name)
+                # actually notify proper handlers based on all notifications up to this point
+                for notification in list(host_state.handler_notifications):
+                    for handler in self.search_handlers_by_notification(notification, iterator):
+                        if not handler.notify_host(target_host):
+                            # NOTE even with notifications deduplicated this can still happen in case of handlers being
+                            # notified multiple times using different names, like role name or fqcn
+                            self._tqm.send_callback('v2_playbook_on_notify', handler, target_host)
+                    iterator.clear_notification(target_host.name, notification)
+
                 if host_state.run_state == IteratingStates.HANDLERS:
                     raise AnsibleError('flush_handlers cannot be used as a handler')
                 if target_host.name not in self._tqm._unreachable_hosts:
diff --git a/test/integration/targets/handlers/roles/two_tasks_files_role/handlers/main.yml b/test/integration/targets/handlers/roles/two_tasks_files_role/handlers/main.yml
new file mode 100644
index 0000000000..3fd1318713
--- /dev/null
+++ b/test/integration/targets/handlers/roles/two_tasks_files_role/handlers/main.yml
@@ -0,0 +1,3 @@
+- name: handler
+  debug:
+    msg: handler ran
diff --git a/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/main.yml b/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/main.yml
new file mode 100644
index 0000000000..e6c12397bb
--- /dev/null
+++ b/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/main.yml
@@ -0,0 +1,3 @@
+- name: main.yml task
+  command: echo
+  notify: handler
diff --git a/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/other.yml b/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/other.yml
new file mode 100644
index 0000000000..d90d46e00b
--- /dev/null
+++ b/test/integration/targets/handlers/roles/two_tasks_files_role/tasks/other.yml
@@ -0,0 +1,3 @@
+- name: other.yml task
+  command: echo
+  notify: handler
diff --git a/test/integration/targets/handlers/runme.sh b/test/integration/targets/handlers/runme.sh
index 76fc99d85b..cc0997a5c2 100755
--- a/test/integration/targets/handlers/runme.sh
+++ b/test/integration/targets/handlers/runme.sh
@@ -181,3 +181,6 @@ grep out.txt -e "ERROR! Using a block as a handler is not supported."
 
 ansible-playbook test_block_as_handler-import.yml "$@"  2>&1 | tee out.txt
 grep out.txt -e "ERROR! Using a block as a handler is not supported."
+
+ansible-playbook test_include_role_handler_once.yml -i inventory.handlers "$@" 2>&1 | tee out.txt
+[ "$(grep out.txt -ce 'handler ran')" = "1" ]
diff --git a/test/integration/targets/handlers/test_include_role_handler_once.yml b/test/integration/targets/handlers/test_include_role_handler_once.yml
new file mode 100644
index 0000000000..764aef6490
--- /dev/null
+++ b/test/integration/targets/handlers/test_include_role_handler_once.yml
@@ -0,0 +1,20 @@
+- hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: "Call main entry point"
+      include_role:
+        name: two_tasks_files_role
+
+    - name: "Call main entry point again"
+      include_role:
+        name: two_tasks_files_role
+
+    - name: "Call other entry point"
+      include_role:
+        name: two_tasks_files_role
+        tasks_from: other
+
+    - name: "Call other entry point again"
+      include_role:
+        name: two_tasks_files_role
+        tasks_from: other