diff --git a/changelogs/fragments/81532-fix-nested-flush_handlers.yml b/changelogs/fragments/81532-fix-nested-flush_handlers.yml
new file mode 100644
index 00000000000..e43aa2e1df7
--- /dev/null
+++ b/changelogs/fragments/81532-fix-nested-flush_handlers.yml
@@ -0,0 +1,2 @@
+bugfixes:
+  - flush_handlers - properly handle a handler failure in a nested block when ``force_handlers`` is set (http://github.com/ansible/ansible/issues/81532)
diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index f36be5af52c..07eb523f4d1 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -52,7 +52,7 @@ class FailedStates(IntFlag):
     TASKS = 2
     RESCUE = 4
     ALWAYS = 8
-    HANDLERS = 16
+    HANDLERS = 16  # NOTE not in use anymore
 
 
 class HostState:
@@ -438,22 +438,18 @@ class PlayIterator:
                     state.update_handlers = False
                     state.cur_handlers_task = 0
 
-                if state.fail_state & FailedStates.HANDLERS == FailedStates.HANDLERS:
-                    state.update_handlers = True
-                    state.run_state = IteratingStates.COMPLETE
-                else:
-                    while True:
-                        try:
-                            task = state.handlers[state.cur_handlers_task]
-                        except IndexError:
-                            task = None
-                            state.run_state = state.pre_flushing_run_state
-                            state.update_handlers = True
+                while True:
+                    try:
+                        task = state.handlers[state.cur_handlers_task]
+                    except IndexError:
+                        task = None
+                        state.run_state = state.pre_flushing_run_state
+                        state.update_handlers = True
+                        break
+                    else:
+                        state.cur_handlers_task += 1
+                        if task.is_host_notified(host):
                             break
-                        else:
-                            state.cur_handlers_task += 1
-                            if task.is_host_notified(host):
-                                break
 
             elif state.run_state == IteratingStates.COMPLETE:
                 return (state, None)
@@ -494,20 +490,16 @@ class PlayIterator:
             else:
                 state.fail_state |= FailedStates.ALWAYS
                 state.run_state = IteratingStates.COMPLETE
-        elif state.run_state == IteratingStates.HANDLERS:
-            state.fail_state |= FailedStates.HANDLERS
-            state.update_handlers = True
-            if state._blocks[state.cur_block].rescue:
-                state.run_state = IteratingStates.RESCUE
-            elif state._blocks[state.cur_block].always:
-                state.run_state = IteratingStates.ALWAYS
-            else:
-                state.run_state = IteratingStates.COMPLETE
         return state
 
     def mark_host_failed(self, host):
         s = self.get_host_state(host)
         display.debug("marking host %s failed, current state: %s" % (host, s))
+        if s.run_state == IteratingStates.HANDLERS:
+            # we are failing `meta: flush_handlers`, so just reset the state to whatever
+            # it was before and let `_set_failed_state` figure out the next state
+            s.run_state = s.pre_flushing_run_state
+            s.update_handlers = True
         s = self._set_failed_state(s)
         display.debug("^ failed state is now: %s" % s)
         self.set_state_for_host(host.name, s)
@@ -523,8 +515,6 @@ class PlayIterator:
             return True
         elif state.run_state == IteratingStates.ALWAYS and self._check_failed_state(state.always_child_state):
             return True
-        elif state.run_state == IteratingStates.HANDLERS and state.fail_state & FailedStates.HANDLERS == FailedStates.HANDLERS:
-            return True
         elif state.fail_state != FailedStates.NONE:
             if state.run_state == IteratingStates.RESCUE and state.fail_state & FailedStates.RESCUE == 0:
                 return False
diff --git a/test/integration/targets/handlers/nested_flush_handlers_failure_force.yml b/test/integration/targets/handlers/nested_flush_handlers_failure_force.yml
new file mode 100644
index 00000000000..7380923e99a
--- /dev/null
+++ b/test/integration/targets/handlers/nested_flush_handlers_failure_force.yml
@@ -0,0 +1,19 @@
+- hosts: A,B
+  gather_facts: false
+  force_handlers: true
+  tasks:
+    - block:
+        - command: echo
+          notify: h
+
+        - meta: flush_handlers
+      rescue:
+        - debug:
+            msg: flush_handlers_rescued
+      always:
+        - debug:
+            msg: flush_handlers_always
+  handlers:
+    - name: h
+      fail:
+      when: inventory_hostname == "A"
diff --git a/test/integration/targets/handlers/runme.sh b/test/integration/targets/handlers/runme.sh
index 757200de385..fa62e0a330c 100755
--- a/test/integration/targets/handlers/runme.sh
+++ b/test/integration/targets/handlers/runme.sh
@@ -198,3 +198,7 @@ ansible-playbook test_include_tasks_in_include_role.yml "$@" 2>&1 | tee out.txt
 
 ansible-playbook test_run_once.yml -i inventory.handlers "$@" 2>&1 | tee out.txt
 [ "$(grep out.txt -ce 'handler ran once')" = "1" ]
+
+ansible-playbook nested_flush_handlers_failure_force.yml -i inventory.handlers "$@" 2>&1 | tee out.txt
+[ "$(grep out.txt -ce 'flush_handlers_rescued')" = "1" ]
+[ "$(grep out.txt -ce 'flush_handlers_always')" = "2" ]