Fix when_all to wait for all tasks before surfacing failures

when_all previously failed fast on the first child task failure. It now waits for every child task to complete before surfacing the first failure, matching the semantics of .NET's Task.WhenAll.

Author: andystaples

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

@@ -562,13 +562,20 @@ 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._exception is None:
+                self._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._exception is None:
+                # 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