Fix when_all to wait for all tasks before surfacing failures

CHANGELOG.md

@@ -11,6 +11,10 @@ ADDED
 
 - Added `TaskHubGrpcClient.rewind_orchestration()` to rewind a failed orchestration instance to its last known good state. Failed activity and sub-orchestration results are removed from the history and the orchestration replays from the last successful checkpoint, retrying only the failed work. The in-memory testing backend supports rewind as well.
 
+FIXED
+
+- Fixed `task.when_all()` failing fast when one of its child tasks failed. It now waits for every child task to complete before surfacing the first failure, matching the semantics of .NET's `Task.WhenAll`.
+
 ## v1.7.2
 
 FIXED

durabletask/task.py

@@ -552,6 +552,7 @@ def __init__(self, tasks: list[Task[T]]):
         super().__init__(cast(list[Task[Any]], tasks))
         self._completed_tasks = 0
         self._failed_tasks = 0
+        self._pending_exception: TaskFailedError | None = None
 
     @property
     def pending_tasks(self) -> int:
@@ -562,13 +563,26 @@ def on_child_completed(self, task: Task[Any]) -> None:
         if self.is_complete:
             raise ValueError('The task has already completed.')
         self._completed_tasks += 1
-        if task.is_failed and self._exception is None:
-            self._exception = task.get_exception()
-            self._is_complete = True
+        if task.is_failed:
+            self._failed_tasks += 1
+            if self._pending_exception is None:
+                # Stage the first failure but do NOT expose it via `_exception`
+                # yet. Exposing it now would make `is_failed` return True while
+                # `is_complete` is still False, diverging from .NET's
+                # Task.WhenAll (which does not fault until all children finish).
+                self._pending_exception = task.get_exception()
         if self._completed_tasks == len(self._tasks):
-            # The order of the result MUST match the order of the tasks provided to the constructor.
-            self._result = [child.get_result() for child in self._tasks]
+            # Only complete once every child task has completed. This matches the
+            # semantics of .NET's Task.WhenAll: the composite task waits for all
+            # children to finish and, if any failed, surfaces the first failure
+            # rather than failing fast on the first error.
             self._is_complete = True
+            if self._pending_exception is not None:
+                self._exception = self._pending_exception
+            else:
+                # The order of the result MUST match the order of the tasks
+                # provided to the constructor.
+                self._result = [child.get_result() for child in self._tasks]
 
     def get_completed_tasks(self) -> int:
         return self._completed_tasks

tests/durabletask/test_orchestration_executor.py

@@ -1615,15 +1615,27 @@ def orchestrator(ctx: task.OrchestrationContext, _):
             i + 1, activity_name, encoded_input=str(i)))
 
     # 5 of the tasks complete successfully, 1 fails, and 4 are still running.
-    # The expectation is that the orchestration will fail immediately.
-    new_events = []
+    # when_all must NOT fail fast: it waits for every child task to complete
+    # before surfacing a failure (matching .NET's Task.WhenAll semantics), so
+    # the orchestration is expected to still be running with zero new actions.
+    ex = Exception("Kah-BOOOOM!!!")
+    partial_events = []
     for i in range(5):
+        partial_events.append(helpers.new_task_completed_event(
+            i + 1, encoded_output=print_int(None, i)))
+    partial_events.append(helpers.new_task_failed_event(6, ex))
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
+    result = executor.execute(TEST_INSTANCE_ID, old_events, partial_events)
+    assert len(result.actions) == 0
+
+    # Once the remaining 4 tasks also complete, the orchestration fails and
+    # surfaces the first task failure.
+    new_events = list(partial_events)
+    for i in range(6, 10):
         new_events.append(helpers.new_task_completed_event(
             i + 1, encoded_output=print_int(None, i)))
-    ex = Exception("Kah-BOOOOM!!!")
-    new_events.append(helpers.new_task_failed_event(6, ex))
 
-    # Now test with the full set of new events. We expect the orchestration to complete.
     executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter())
     result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
     actions = result.actions
@@ -1634,6 +1646,40 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert str(ex) in complete_action.failureDetails.errorMessage
 
 
+def test_when_all_defers_failure_until_all_children_complete():
+    """when_all must not report is_failed until every child task completes.
+
+    This mirrors .NET's Task.WhenAll, where the returned task does not fault
+    (failed => complete) until all children have finished. A composite that
+    exposed is_failed while still incomplete would surprise consumers that
+    assume failed implies complete.
+    """
+    t1 = task.CompletableTask()
+    t2 = task.CompletableTask()
+    t3 = task.CompletableTask()
+    when_all = task.when_all([t1, t2, t3])
+
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # First child fails: the composite must stay pending and must NOT report
+    # failure yet.
+    t1.fail("boom", Exception("boom"))
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # A later child completes successfully: still pending, still not failed.
+    t2.complete("ok")
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # Once the final child completes, the composite completes and only then
+    # surfaces the first failure.
+    t3.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):