linear: fix included handlers executing in lockstep

Fixes ansible#83019

lib/ansible/plugins/strategy/linear.py

@@ -46,12 +46,6 @@
 
 class StrategyModule(StrategyBase):
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        # used for the lockstep to indicate to run handlers
-        self._in_handlers = False
-
     def _get_next_task_lockstep(self, hosts, iterator):
         '''
         Returns a list of (host, task) tuples, where the task may
@@ -73,52 +67,35 @@ def _get_next_task_lockstep(self, hosts, iterator):
         if not state_task_per_host:
             return [(h, None) for h in hosts]
 
-        if self._in_handlers and not any(filter(
-            lambda rs: rs == IteratingStates.HANDLERS,
-            (s.run_state for s, dummy in state_task_per_host.values()))
-        ):
-            self._in_handlers = False
-
-        if self._in_handlers:
-            lowest_cur_handler = min(
-                s.cur_handlers_task for s, t in state_task_per_host.values()
-                if s.run_state == IteratingStates.HANDLERS
-            )
-        else:
-            task_uuids = [t._uuid for s, t in state_task_per_host.values()]
-            _loop_cnt = 0
-            while _loop_cnt <= 1:
-                try:
-                    cur_task = iterator.all_tasks[iterator.cur_task]
-                except IndexError:
-                    # pick up any tasks left after clear_host_errors
-                    iterator.cur_task = 0
-                    _loop_cnt += 1
-                else:
-                    iterator.cur_task += 1
-                    if cur_task._uuid in task_uuids:
-                        break
+        task_uuids = {t._uuid for s, t in state_task_per_host.values()}
+        _loop_cnt = 0
+        while _loop_cnt <= 1:
+            try:
+                cur_task = iterator.all_tasks[iterator.cur_task]
+            except IndexError:
+                # pick up any tasks left after clear_host_errors
+                iterator.cur_task = 0
+                _loop_cnt += 1
             else:
-                # prevent infinite loop
-                raise AnsibleAssertionError(
-                    'BUG: There seems to be a mismatch between tasks in PlayIterator and HostStates.'
-                )
+                iterator.cur_task += 1
+                if cur_task._uuid in task_uuids:
+                    break
+        else:
+            # prevent infinite loop
+            raise AnsibleAssertionError(
+                'BUG: There seems to be a mismatch between tasks in PlayIterator and HostStates.'
+            )
 
         host_tasks = []
         for host, (state, task) in state_task_per_host.items():
-            if ((self._in_handlers and lowest_cur_handler == state.cur_handlers_task) or
-                    (not self._in_handlers and cur_task._uuid == task._uuid)):
+            if cur_task._uuid == task._uuid:
                 iterator.set_state_for_host(host.name, state)
                 host_tasks.append((host, task))
             else:
                 host_tasks.append((host, noop_task))
 
-        # once hosts synchronize on 'flush_handlers' lockstep enters
-        # '_in_handlers' phase where handlers are run instead of tasks
-        # until at least one host is in IteratingStates.HANDLERS
-        if (not self._in_handlers and cur_task.action in C._ACTION_META and
-                cur_task.args.get('_raw_params') == 'flush_handlers'):
-            self._in_handlers = True
+        if cur_task.action in C._ACTION_META and cur_task.args.get('_raw_params') == 'flush_handlers':
+            iterator.all_tasks[iterator.cur_task:iterator.cur_task] = [h for b in iterator._play.handlers for h in b.block]
 
         return host_tasks
 
@@ -319,7 +296,7 @@ def run(self, iterator, play_context):
                                     final_block = new_block.filter_tagged_tasks(task_vars)
                                     display.debug("done filtering new block on tags")
 
-                                    included_tasks.extend(final_block.get_tasks())
+                                included_tasks.extend(final_block.get_tasks())
 
                                 for host in hosts_left:
                                     if host in included_file._hosts: