Render a root-cause exception for dependency and join errors

parsl/dataflow/errors.py

@@ -1,4 +1,4 @@
-from typing import Optional, Sequence, Tuple
+from typing import List, Sequence, Tuple
 
 from parsl.errors import ParslError
 
@@ -29,35 +29,77 @@ def __str__(self) -> str:
         return self.reason
 
 
-class DependencyError(DataFlowException):
-    """Error raised if an app cannot run because there was an error
-       in a dependency.
+class PropagatedException(DataFlowException):
+    """Error raised if an app fails because there was an error
+       in a related task. This is intended to be subclassed for
+       dependency and join_app errors.
 
     Args:
-         - dependent_exceptions_tids: List of exceptions and identifiers for
-           dependencies which failed. The identifier might be a task ID or
-           the repr of a non-DFK Future.
+         - dependent_exceptions_tids: List of exceptions and brief descriptions
+           for dependencies which failed. The description might be a task ID or
+           the repr of a non-AppFuture.
          - task_id: Task ID of the task that failed because of the dependency error
     """
 
-    def __init__(self, dependent_exceptions_tids: Sequence[Tuple[Exception, str]], task_id: int) -> None:
+    def __init__(self,
+                 dependent_exceptions_tids: Sequence[Tuple[BaseException, str]],
+                 task_id: int,
+                 *,
+                 failure_description: str) -> None:
         self.dependent_exceptions_tids = dependent_exceptions_tids
         self.task_id = task_id
+        self._failure_description = failure_description
+
+        (cause, cause_sequence) = self._find_any_root_cause()
+        self.__cause__ = cause
+        self._cause_sequence = cause_sequence
 
     def __str__(self) -> str:
-        deps = ", ".join(tid for _exc, tid in self.dependent_exceptions_tids)
-        return f"Dependency failure for task {self.task_id} with failed dependencies from {deps}"
+        sequence_text = " <- ".join(self._cause_sequence)
+        return f"{self._failure_description} for task {self.task_id}. " \
+               f"The representative cause is via {sequence_text}"
+
+    def _find_any_root_cause(self) -> Tuple[BaseException, List[str]]:
+        """Looks recursively through self.dependent_exceptions_tids to find
+        an exception that caused this propagated error, that is not itself
+        a propagated error.
+        """
+        e: BaseException = self
+        dep_ids = []
+        while isinstance(e, PropagatedException) and len(e.dependent_exceptions_tids) >= 1:
+            id_txt = e.dependent_exceptions_tids[0][1]
+            assert isinstance(id_txt, str)
+            # if there are several causes for this exception, label that
+            # there are more so that we know that the representative fail
+            # sequence is not the full story.
+            if len(e.dependent_exceptions_tids) > 1:
+                id_txt += " (+ others)"
+            dep_ids.append(id_txt)
+            e = e.dependent_exceptions_tids[0][0]
+        return e, dep_ids
+
+
+class DependencyError(PropagatedException):
+    """Error raised if an app cannot run because there was an error
+       in a dependency. There can be several exceptions (one from each
+       dependency) and DependencyError collects them all together.
 
+    Args:
+         - dependent_exceptions_tids: List of exceptions and brief descriptions
+           for dependencies which failed. The description might be a task ID or
+           the repr of a non-AppFuture.
+         - task_id: Task ID of the task that failed because of the dependency error
+    """
+    def __init__(self, dependent_exceptions_tids: Sequence[Tuple[BaseException, str]], task_id: int) -> None:
+        super().__init__(dependent_exceptions_tids, task_id,
+                         failure_description="Dependency failure")
 
-class JoinError(DataFlowException):
+
+class JoinError(PropagatedException):
     """Error raised if apps joining into a join_app raise exceptions.
        There can be several exceptions (one from each joining app),
        and JoinError collects them all together.
     """
-    def __init__(self, dependent_exceptions_tids: Sequence[Tuple[BaseException, Optional[str]]], task_id: int) -> None:
-        self.dependent_exceptions_tids = dependent_exceptions_tids
-        self.task_id = task_id
-
-    def __str__(self) -> str:
-        dep_tids = [tid for (exception, tid) in self.dependent_exceptions_tids]
-        return "Join failure for task {} with failed join dependencies from tasks {}".format(self.task_id, dep_tids)
+    def __init__(self, dependent_exceptions_tids: Sequence[Tuple[BaseException, str]], task_id: int) -> None:
+        super().__init__(dependent_exceptions_tids, task_id,
+                         failure_description="Join failure")

parsl/tests/test_python_apps/test_fail.py

@@ -39,6 +39,9 @@ def test_fail_sequence_first():
     assert isinstance(t_final.exception().dependent_exceptions_tids[0][0], DependencyError)
     assert t_final.exception().dependent_exceptions_tids[0][1].startswith("task ")
 
+    assert hasattr(t_final.exception(), '__cause__')
+    assert t_final.exception().__cause__ == t1.exception()
+
 
 def test_fail_sequence_middle():
     t1 = random_fail(fail_prob=0)
@@ -50,3 +53,6 @@ def test_fail_sequence_middle():
 
     assert len(t_final.exception().dependent_exceptions_tids) == 1
     assert isinstance(t_final.exception().dependent_exceptions_tids[0][0], ManufacturedTestFailure)
+
+    assert hasattr(t_final.exception(), '__cause__')
+    assert t_final.exception().__cause__ == t2.exception()