Avoid duplicate task yields (#292)

Author: davidmrdavid

azure/durable_functions/models/Task.py

@@ -26,6 +26,7 @@ def __init__(self, is_completed, is_faulted, action,
         self._id = id_
         self._exception = exc
         self._is_played = is_played
+        self._is_yielded: bool = False
 
     @property
     def is_completed(self) -> bool:

azure/durable_functions/models/TaskSet.py

@@ -25,6 +25,7 @@ def __init__(self, is_completed, actions, result, is_faulted=False,
         self._timestamp: datetime = timestamp
         self._exception = exception
         self._is_played = is_played
+        self._is_yielded: bool = False
 
     @property
     def is_completed(self) -> bool:

azure/durable_functions/orchestrator.py

@@ -134,12 +134,14 @@ def _add_to_actions(self, generation_state):
         # Do not add new tasks to action if continue_as_new was called
         if self.durable_context.will_continue_as_new:
             return
-        if (isinstance(generation_state, Task)
-                and hasattr(generation_state, "action")):
-            self.durable_context.actions.append([generation_state.action])
-        elif (isinstance(generation_state, TaskSet)
-              and hasattr(generation_state, "actions")):
-            self.durable_context.actions.append(generation_state.actions)
+        if not generation_state._is_yielded:
+            if (isinstance(generation_state, Task)
+                    and hasattr(generation_state, "action")):
+                self.durable_context.actions.append([generation_state.action])
+            elif (isinstance(generation_state, TaskSet)
+                    and hasattr(generation_state, "actions")):
+                self.durable_context.actions.append(generation_state.actions)
+            generation_state._is_yielded = True
 
     def _update_timestamp(self):
         last_timestamp = self.durable_context.decision_started_event.timestamp

tests/orchestrator/test_sequential_orchestrator.py

@@ -21,6 +21,13 @@ def generator_function(context):
 
     return outputs
 
+def generator_function_duplicate_yield(context):
+    task1 = context.call_activity("Hello", "Tokyo")
+    yield task1
+    yield task1
+
+    return ""
+
 def generator_function_time_is_not_none(context):
     outputs = []
 
@@ -385,3 +392,19 @@ def test_new_guid_orchestrator():
     # The two GUID lists should be the same
     assert outputs1 == outputs2
 
+def test_duplicate_yields_do_not_add_duplicate_actions():
+    """Tests that yield'ing a Task twice does not double the task's actions"""
+    context_builder = ContextBuilder('test_guid_orchestrator')
+    add_hello_completed_events(context_builder, 0, "\"Hello Tokyo!\"")
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_duplicate_yield)
+
+    expected_state = base_expected_state("")
+    add_hello_action(expected_state, 'Tokyo')
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_valid_schema(result)
+    assert_orchestration_state_equals(expected, result)
+