Make -WithRetry APIs wait for TimerFired after final task failure (#339)

Author: davidmrdavid

azure/durable_functions/models/Task.py

@@ -347,6 +347,7 @@ def __init__(self, child: TaskBase, retry_options: RetryOptions, context):
         self.context = context
         self.actions = child.action_repr
         self.is_waiting_on_timer = False
+        self.error = None
 
     @property
     def id_(self):
@@ -373,10 +374,21 @@ def try_set_value(self, child: TaskBase):
         if self.is_waiting_on_timer:
             # timer fired, re-scheduling original task
             self.is_waiting_on_timer = False
-            rescheduled_task = self.context._generate_task(
-                action=NoOpAction("rescheduled task"), parent=self)
-            self.pending_tasks.add(rescheduled_task)
-            self.context._add_to_open_tasks(rescheduled_task)
+            # As per DTFx semantics: we need to check the number of retires only after the final
+            # timer has fired. This means we essentially have to wait for one "extra" timer after
+            # the maximum number of attempts has been reached. Removing this extra timer will cause
+            # stuck orchestrators as we need to be "in sync" with the replay logic of DTFx.
+            if self.num_attempts >= self.retry_options.max_number_of_attempts:
+                self.is_waiting_on_timer = True
+                # we have reached the maximum number of attempts, set error
+                self.set_value(is_error=True, value=self.error)
+            else:
+                rescheduled_task = self.context._generate_task(
+                    action=NoOpAction("rescheduled task"), parent=self)
+                self.pending_tasks.add(rescheduled_task)
+                self.context._add_to_open_tasks(rescheduled_task)
+                self.num_attempts += 1
+
             return
         if child.state is TaskState.SUCCEEDED:
             if len(self.pending_tasks) == 0:
@@ -386,17 +398,11 @@ def try_set_value(self, child: TaskBase):
                 self.set_value(is_error=False, value=child.result)
 
         else:  # child.state is TaskState.FAILED:
-            if self.num_attempts >= self.retry_options.max_number_of_attempts:
-                # we have reached the maximum number of attempts, set error
-                self.set_value(is_error=True, value=child.result)
-            else:
-                # still have some retries left.
-                # increase size of pending tasks by adding a timer task
-                # when it completes, we'll retry the original task
-                timer_task = self.context._generate_task(
-                    action=NoOpAction("-WithRetry timer"), parent=self)
-                self.pending_tasks.add(timer_task)
-                self.context._add_to_open_tasks(timer_task)
-                self.is_waiting_on_timer = True
-
-            self.num_attempts += 1
+            # increase size of pending tasks by adding a timer task
+            # when it completes, we'll retry the original task
+            timer_task = self.context._generate_task(
+                action=NoOpAction("-WithRetry timer"), parent=self)
+            self.pending_tasks.add(timer_task)
+            self.context._add_to_open_tasks(timer_task)
+            self.is_waiting_on_timer = True
+            self.error = child.result

tests/orchestrator/test_sequential_orchestrator_with_retry.py

@@ -29,6 +29,19 @@ def generator_function(context):
 
     return outputs
 
+def generator_function_try_catch(context):
+    outputs = []
+
+    retry_options = RETRY_OPTIONS
+    result = None
+    try:
+        result = yield context.call_activity_with_retry(
+        "Hello", retry_options, "Tokyo")
+    except:
+        result = yield context.call_activity_with_retry(
+        "Hello",  retry_options, "Seattle")
+    return result
+
 def generator_function_concurrent_retries(context):
     outputs = []
 
@@ -305,6 +318,37 @@ def test_failed_tokyo_hit_max_attempts():
         expected_error_str = f"{error_msg}{error_label}{state_str}"
         assert expected_error_str == error_str
 
+def test_failed_tokyo_hit_max_attempts_in_try_catch():
+    # This test ensures that APIs can still be invoked after a failed CallActivityWithRetry invocation
+    failed_reason = 'Reasons'
+    failed_details = 'Stuff and Things'
+    context_builder = ContextBuilder('test_simple_function')
+    
+    # events for first task: "Hello Tokyo"
+    add_hello_failed_events(context_builder, 0, failed_reason, failed_details)
+    add_retry_timer_events(context_builder, 1)
+    add_hello_failed_events(context_builder, 2, failed_reason, failed_details)
+    add_retry_timer_events(context_builder, 3)
+    add_hello_failed_events(context_builder, 4, failed_reason, failed_details)
+    # we have an "extra" timer to wait for, due to legacy behavior in DTFx.
+    add_retry_timer_events(context_builder, 5)
+
+    # events to task in except block
+    add_hello_completed_events(context_builder, 6, "\"Hello Seattle!\"")
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_try_catch)
+
+    expected_state = base_expected_state()
+    add_hello_action(expected_state, 'Tokyo')
+    add_hello_action(expected_state, 'Seattle')
+    expected_state._output = "Hello Seattle!"
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_valid_schema(result)
+    assert_orchestration_state_equals(expected, result)
+
 def test_concurrent_retriable_results():
     failed_reason = 'Reasons'
     failed_details = 'Stuff and Things'