From 0de8299693598f5d8306f5af58e9ed37a525a2db Mon Sep 17 00:00:00 2001 From: Patrick Assuied Date: Mon, 13 Jul 2026 18:37:32 -0700 Subject: [PATCH] Fix when_all hang when some children complete before construction. WhenAllTask.__init__ was resetting _completed_tasks after CompositeTask had already counted pre-completed children, so deferred when_all hung forever in the partial-completion case. Signed-off-by: Patrick Assuied Co-authored-by: Cursor --- dapr/ext/workflow/_durabletask/task.py | 7 ++- tests/ext/workflow/durabletask/test_task.py | 57 +++++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/dapr/ext/workflow/_durabletask/task.py b/dapr/ext/workflow/_durabletask/task.py index 21bebfabf..26813c5a5 100644 --- a/dapr/ext/workflow/_durabletask/task.py +++ b/dapr/ext/workflow/_durabletask/task.py @@ -385,9 +385,12 @@ class WhenAllTask(CompositeTask[list[T]]): """A task that completes when all of its child tasks complete.""" def __init__(self, tasks: list[Task[T]]): + # Do not reset _completed_tasks / _failed_tasks after super().__init__. + # CompositeTask already initializes them and counts any children that are + # already complete via on_child_completed(). Resetting the counters here + # drops those completions, so deferred when_all(children) hangs forever + # when some (but not all) children finished before when_all was constructed. super().__init__(tasks) - self._completed_tasks = 0 - self._failed_tasks = 0 # If there are no child tasks, this composite should complete immediately if len(self._tasks) == 0: self._result = [] # type: ignore[assignment] diff --git a/tests/ext/workflow/durabletask/test_task.py b/tests/ext/workflow/durabletask/test_task.py index aeb2d2b3b..230d9b54f 100644 --- a/tests/ext/workflow/durabletask/test_task.py +++ b/tests/ext/workflow/durabletask/test_task.py @@ -62,6 +62,63 @@ def test_when_all_happy_path_returns_ordered_results_and_completes_last(): assert all_task.get_result() == ['one', 'two', 'three'] +def test_when_all_counts_already_complete_children(): + """Deferred when_all must retain completions that happened before construction. + + Pattern: schedule children, do other work (timer/activity), then when_all. + Children that finish during that gap must still be counted; otherwise + when_all hangs forever when some but not all children finished early. + """ + children = [task.CompletableTask() for _ in range(5)] + for child in children[:3]: + child.complete(f'done-{id(child)}') + + all_task = task.when_all(children) + + assert not all_task.is_complete + assert all_task.get_completed_tasks() == 3 + + children[3].complete('four') + assert not all_task.is_complete + + children[4].complete('five') + assert all_task.is_complete + assert all_task.get_completed_tasks() == 5 + assert len(all_task.get_result()) == 5 + + +def test_when_all_all_children_already_complete(): + """when_all of already-complete children should complete immediately.""" + children = [task.CompletableTask() for _ in range(3)] + for i, child in enumerate(children): + child.complete(f'v{i}') + + all_task = task.when_all(children) + + assert all_task.is_complete + assert all_task.get_result() == ['v0', 'v1', 'v2'] + + +def test_when_any_of_when_all_with_precompleted_children(): + """when_any([when_all(children), timeout]) must unblock when children finish early.""" + children = [task.CompletableTask() for _ in range(3)] + children[0].complete('a') + children[1].complete('b') + + timeout = task.CompletableTask() + all_task = task.when_all(children) + any_task = task.when_any([all_task, timeout]) + + assert not any_task.is_complete + assert all_task.get_completed_tasks() == 2 + + children[2].complete('c') + + assert all_task.is_complete + assert any_task.is_complete + assert any_task.get_result() is all_task + + def test_when_all_is_composable_with_when_any(): c1 = task.CompletableTask() c2 = task.CompletableTask()