Fix a bug in the algo that determines if a route is possible

Change-Id: I0747c35d07bac85feded4ff41a59d1551a0f7652
Closes-Bug: #1636169

mistral/tests/unit/engine/test_join.py

@@ -922,6 +922,61 @@ class JoinEngineTest(base.EngineTestCase):
             len(db_api.get_delayed_calls(target_method_name=mtd_name)) == 0
         )
 
+    def test_join_with_deep_dependencies_tree(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task_a_1:
+              on-success:
+                - task_with_join
+
+            task_b_1:
+              action: std.fail
+              on-success:
+                - task_b_2
+
+            task_b_2:
+              on-success:
+                - task_b_3
+
+            task_b_3:
+              on-success:
+                - task_with_join
+
+            task_with_join:
+              join: all
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', {})
+
+        self.await_workflow_error(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            task_execs = wf_ex.task_executions
+
+        self.assertEqual(3, len(task_execs))
+        self._assert_single_item(
+            task_execs,
+            name='task_a_1',
+            state=states.SUCCESS
+        )
+        self._assert_single_item(
+            task_execs,
+            name='task_b_1',
+            state=states.ERROR
+        )
+        self._assert_single_item(
+            task_execs,
+            name='task_with_join',
+            state=states.ERROR
+        )
+
     def test_no_workflow_error_after_inbound_error(self):
         wf_text = """---
         version: "2.0"

mistral/workflow/direct_workflow.py

@@ -426,7 +426,9 @@ class DirectWorkflowController(base.WorkflowController):
             t_ex = self._find_task_execution_by_name(t_s.get_name())
 
             if not t_ex:
-                if self._possible_route(t_s, depth + 1):
+                possible, depth = self._possible_route(t_s, depth + 1)
+
+                if possible:
                     return True, depth
             else:
                 t_name = task_spec.get_name()