diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index 02052f56b3c..1e7775d3e22 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -275,7 +275,7 @@ class PlayIterator:
         return (s, task)
 
 
-    def _get_next_task_from_state(self, state, host, peek):
+    def _get_next_task_from_state(self, state, host, peek, in_child=False):
 
         task = None
 
@@ -341,7 +341,7 @@ class PlayIterator:
                 # have one recurse into it for the next task. If we're done with the child
                 # state, we clear it and drop back to geting the next task from the list.
                 if state.tasks_child_state:
-                    (state.tasks_child_state, task) = self._get_next_task_from_state(state.tasks_child_state, host=host, peek=peek)
+                    (state.tasks_child_state, task) = self._get_next_task_from_state(state.tasks_child_state, host=host, peek=peek, in_child=True)
                     if self._check_failed_state(state.tasks_child_state):
                         # failed child state, so clear it and move into the rescue portion
                         state.tasks_child_state = None
@@ -379,7 +379,7 @@ class PlayIterator:
                 # The process here is identical to ITERATING_TASKS, except instead
                 # we move into the always portion of the block.
                 if state.rescue_child_state:
-                    (state.rescue_child_state, task) = self._get_next_task_from_state(state.rescue_child_state, host=host, peek=peek)
+                    (state.rescue_child_state, task) = self._get_next_task_from_state(state.rescue_child_state, host=host, peek=peek, in_child=True)
                     if self._check_failed_state(state.rescue_child_state):
                         state.rescue_child_state = None
                         self._set_failed_state(state)
@@ -409,7 +409,7 @@ class PlayIterator:
                 # run state to ITERATING_COMPLETE in the event of any errors, or when we
                 # have hit the end of the list of blocks.
                 if state.always_child_state:
-                    (state.always_child_state, task) = self._get_next_task_from_state(state.always_child_state, host=host, peek=peek)
+                    (state.always_child_state, task) = self._get_next_task_from_state(state.always_child_state, host=host, peek=peek, in_child=True)
                     if self._check_failed_state(state.always_child_state):
                         state.always_child_state = None
                         self._set_failed_state(state)
@@ -434,7 +434,7 @@ class PlayIterator:
 
                             # we're advancing blocks, so if this was an end-of-role block we
                             # mark the current role complete
-                            if block._eor and host.name in block._role._had_task_run:
+                            if block._eor and host.name in block._role._had_task_run and not in_child:
                                 block._role._completed[host.name] = True
                     else:
                         task = block.always[state.cur_always_task]
diff --git a/test/units/executor/test_play_iterator.py b/test/units/executor/test_play_iterator.py
index e11772031f6..337342a3496 100644
--- a/test/units/executor/test_play_iterator.py
+++ b/test/units/executor/test_play_iterator.py
@@ -98,6 +98,29 @@ class TestPlayIterator(unittest.TestCase):
             - include: foo.yml
             - name: role task after include
               debug: msg="after include in role"
+            - block:
+              - name: starting role nested block 1
+                debug:
+              - block:
+                - name: role nested block 1 task 1
+                  debug:
+                - name: role nested block 1 task 2
+                  debug:
+                - name: role nested block 1 task 3
+                  debug:
+              - name: end of role nested block 1
+                debug:
+              - name: starting role nested block 2
+                debug:
+              - block:
+                - name: role nested block 2 task 1
+                  debug:
+                - name: role nested block 2 task 2
+                  debug:
+                - name: role nested block 2 task 3
+                  debug:
+              - name: end of role nested block 2
+                debug:
             """,
             '/etc/ansible/roles/test_role/tasks/foo.yml': """
             - name: role included task
@@ -160,13 +183,11 @@ class TestPlayIterator(unittest.TestCase):
         # role block task
         (host_state, task) = itr.get_next_task_for_host(hosts[0])
         self.assertIsNotNone(task)
-        self.assertEqual(task.action, 'debug')
         self.assertEqual(task.name, "role block task")
         self.assertIsNotNone(task._role)
         # role block always task
         (host_state, task) = itr.get_next_task_for_host(hosts[0])
         self.assertIsNotNone(task)
-        self.assertEqual(task.action, 'debug')
         self.assertEqual(task.name, "role always task")
         self.assertIsNotNone(task._role)
         # role include task
@@ -178,9 +199,49 @@ class TestPlayIterator(unittest.TestCase):
         # role task after include
         (host_state, task) = itr.get_next_task_for_host(hosts[0])
         self.assertIsNotNone(task)
-        self.assertEqual(task.action, 'debug')
         self.assertEqual(task.name, "role task after include")
         self.assertIsNotNone(task._role)
+        # role nested block tasks
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "starting role nested block 1")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 1 task 1")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 1 task 2")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 1 task 3")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "end of role nested block 1")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "starting role nested block 2")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 2 task 1")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 2 task 2")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "role nested block 2 task 3")
+        self.assertIsNotNone(task._role)
+        (host_state, task) = itr.get_next_task_for_host(hosts[0])
+        self.assertIsNotNone(task)
+        self.assertEqual(task.name, "end of role nested block 2")
+        self.assertIsNotNone(task._role)
         # regular play task
         (host_state, task) = itr.get_next_task_for_host(hosts[0])
         self.assertIsNotNone(task)