Avoid error when JTs deleted while task manager running

awx/main/scheduler/dag_workflow.py

@@ -42,7 +42,13 @@ def _init_graph(self, workflow_job_or_jt):
 
         wfn_by_id = dict()
 
-        for workflow_node in workflow_nodes.all():
+        # Intentionally prefetch related jobs and templates so that if they
+        # are deleted while task manager runs, it will not cause an error
+        if workflow_nodes.model == WorkflowJobNode:
+            node_qs = workflow_nodes.prefetch_related('job', 'unified_job_template').all()
+        else:
+            node_qs = workflow_nodes.prefetch_related('unified_job_template').all()
+        for workflow_node in node_qs:
             wfn_by_id[workflow_node.id] = workflow_node
             self.add_node(workflow_node)
 

awx/main/tests/functional/models/test_workflow.py

@@ -418,3 +418,23 @@ def test_workflow_ancestors_recursion_prevention(organization):
     )
     # mostly, we just care that this assertion finishes in finite time
     assert wfj.get_ancestor_workflows() == []
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize('related', ['template', 'job'])
+def test_deleted_related_race_condition(related):
+    wfj = WorkflowJob.objects.create()
+    jt = JobTemplate.objects.create(name='test-jt')
+    # FIXME: running and successful status also results in node
+    # considered failed when it has not UJT
+    job = jt.create_job(_eager_fields=dict(status='failed'))
+    WorkflowJobNode.objects.create(
+        workflow_job=wfj, unified_job_template=jt, job=job
+    )
+    dag = WorkflowDAG(workflow_job=wfj)
+    if related == 'template':
+        JobTemplate.objects.get(pk=jt.pk).delete()
+    elif related == 'job':
+        Job.objects.get(pk=job.pk).delete()
+    has_failed, reason = dag.has_workflow_failed()
+    assert has_failed is True, reason