Constructor ordering to prevent AttributeError

Author: andystaples

durabletask/task.py

@@ -526,6 +526,8 @@ def get_exception(self) -> TaskFailedError:
 class CompositeTask(Task[T]):
     """A task that is composed of other tasks."""
     _tasks: list[Task[Any]]
+    _completed_tasks: int
+    _failed_tasks: int
 
     def __init__(self, tasks: list[Task[Any]]):
         super().__init__()
@@ -549,10 +551,14 @@ class WhenAllTask(CompositeTask[list[T]]):
     """A task that completes when all of its child tasks complete."""
 
     def __init__(self, tasks: list[Task[T]]):
-        super().__init__(cast(list[Task[Any]], tasks))
-        self._completed_tasks = 0
-        self._failed_tasks = 0
+        # Initialize state that on_child_completed() reads BEFORE invoking the
+        # base constructor: CompositeTask.__init__ calls on_child_completed()
+        # for any children that are already complete, so `_pending_exception`
+        # must exist first. The base constructor also initializes
+        # `_completed_tasks`/`_failed_tasks` to 0 and then accounts for
+        # pre-completed children, so they must not be reset afterwards.
         self._pending_exception: TaskFailedError | None = None
+        super().__init__(cast(list[Task[Any]], tasks))
 
     @property
     def pending_tasks(self) -> int:

tests/durabletask/test_orchestration_executor.py

@@ -1680,6 +1680,33 @@ def test_when_all_defers_failure_until_all_children_complete():
     assert when_all.is_failed
 
 
+def test_when_all_handles_pre_completed_children():
+    """when_all must account for children that are already complete/failed.
+
+    CompositeTask.__init__ invokes on_child_completed() for children that are
+    already complete, so when_all must be constructible from an already-failed
+    child without raising (regression test for AttributeError on
+    `_pending_exception`).
+    """
+    # An already-failed child plus a still-pending child.
+    failed = task.CompletableTask()
+    failed.fail("boom", Exception("boom"))
+    pending = task.CompletableTask()
+
+    when_all = task.when_all([failed, pending])
+
+    # The pre-completed failure is accounted for but not yet surfaced.
+    assert when_all.get_completed_tasks() == 1
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # Completing the remaining child finishes the composite and surfaces the
+    # first failure.
+    pending.complete("ok")
+    assert when_all.is_complete
+    assert when_all.is_failed
+
+
 def test_when_any():
     """Tests that a when_any pattern works correctly"""
     def hello(_, name: str):