Fix: isReplaying flag (#185)

davidmrdavid · web-flow · commit c99c4722c956 · 2020-09-09T15:49:03.000-07:00
diff --git a/azure/durable_functions/models/Task.py b/azure/durable_functions/models/Task.py
@@ -17,14 +17,15 @@ class Task:
     """
 
     def __init__(self, is_completed, is_faulted, action,
-                 result=None, timestamp=None, id_=None, exc=None):
+                 result=None, timestamp=None, id_=None, exc=None, is_played=False):
         self._is_completed: bool = is_completed
         self._is_faulted: bool = is_faulted
         self._action: Action = action
         self._result = result
         self._timestamp: datetime = timestamp
         self._id = id_
         self._exception = exc
+        self._is_played = is_played
 
     @property
     def is_completed(self) -> bool:
diff --git a/azure/durable_functions/models/TaskSet.py b/azure/durable_functions/models/TaskSet.py
@@ -17,13 +17,14 @@ class TaskSet:
     """
 
     def __init__(self, is_completed, actions, result, is_faulted=False,
-                 timestamp=None, exception=None):
+                 timestamp=None, exception=None, is_played=False):
         self._is_completed: bool = is_completed
         self._actions: List[Action] = actions
         self._result = result
         self._is_faulted: bool = is_faulted
         self._timestamp: datetime = timestamp
         self._exception = exception
+        self._is_played = is_played
 
     @property
     def is_completed(self) -> bool:
diff --git a/azure/durable_functions/orchestrator.py b/azure/durable_functions/orchestrator.py
@@ -90,6 +90,7 @@ def handle(self, context: DurableOrchestrationContext):
                         continue
 
                     self._reset_timestamp()
+                    self.durable_context._is_replaying = generation_state._is_played
                     generation_state = self._generate_next(generation_state)
 
             except StopIteration as sie:
diff --git a/azure/durable_functions/tasks/call_activity.py b/azure/durable_functions/tasks/call_activity.py
@@ -40,6 +40,7 @@ def call_activity_task(
             is_completed=True,
             is_faulted=False,
             action=new_action,
+            is_played=task_completed._is_played,
             result=parse_history_event(task_completed),
             timestamp=task_completed.timestamp,
             id_=task_completed.TaskScheduledId)
@@ -49,6 +50,7 @@ def call_activity_task(
             is_completed=True,
             is_faulted=True,
             action=new_action,
+            is_played=task_failed._is_played,
             result=task_failed.Reason,
             timestamp=task_failed.timestamp,
             id_=task_failed.TaskScheduledId,
diff --git a/azure/durable_functions/tasks/call_activity_with_retry.py b/azure/durable_functions/tasks/call_activity_with_retry.py
@@ -55,6 +55,7 @@ def call_activity_with_retry_task(
                 is_completed=True,
                 is_faulted=False,
                 action=new_action,
+                is_played=task_completed._is_played,
                 result=parse_history_event(task_completed),
                 timestamp=task_completed.timestamp,
                 id_=task_completed.TaskScheduledId)
@@ -65,6 +66,7 @@ def call_activity_with_retry_task(
                 is_completed=True,
                 is_faulted=True,
                 action=new_action,
+                is_played=task_failed._is_played,
                 timestamp=task_failed.timestamp,
                 id_=task_failed.TaskScheduledId,
                 exc=Exception(
diff --git a/azure/durable_functions/tasks/call_http.py b/azure/durable_functions/tasks/call_http.py
@@ -57,6 +57,7 @@ def call_http(state: List[HistoryEvent], method: str, uri: str, content: Optiona
             is_completed=True,
             is_faulted=False,
             action=new_action,
+            is_played=task_completed._is_played,
             result=parse_history_event(task_completed),
             timestamp=task_completed.timestamp,
             id_=task_completed.TaskScheduledId)
@@ -66,6 +67,7 @@ def call_http(state: List[HistoryEvent], method: str, uri: str, content: Optiona
             is_completed=True,
             is_faulted=True,
             action=new_action,
+            is_played=task_failed._is_played,
             result=task_failed.Reason,
             timestamp=task_failed.timestamp,
             id_=task_failed.TaskScheduledId,
diff --git a/azure/durable_functions/tasks/call_suborchestrator.py b/azure/durable_functions/tasks/call_suborchestrator.py
@@ -48,6 +48,7 @@ def call_sub_orchestrator_task(
             is_completed=True,
             is_faulted=False,
             action=new_action,
+            is_played=task_completed._is_played,
             result=parse_history_event(task_completed),
             timestamp=task_completed.timestamp,
             id_=task_completed.TaskScheduledId)
@@ -57,6 +58,7 @@ def call_sub_orchestrator_task(
             is_completed=True,
             is_faulted=True,
             action=new_action,
+            is_played=task_failed._is_played,
             result=task_failed.Reason,
             timestamp=task_failed.timestamp,
             id_=task_failed.TaskScheduledId,
diff --git a/azure/durable_functions/tasks/call_suborchestrator_with_retry.py b/azure/durable_functions/tasks/call_suborchestrator_with_retry.py
@@ -59,6 +59,7 @@ def call_sub_orchestrator_with_retry_task(
                 is_completed=True,
                 is_faulted=False,
                 action=new_action,
+                is_played=task_completed._is_played,
                 result=parse_history_event(task_completed),
                 timestamp=task_completed.timestamp,
                 id_=task_completed.TaskScheduledId)
@@ -69,6 +70,7 @@ def call_sub_orchestrator_with_retry_task(
                 is_completed=True,
                 is_faulted=True,
                 action=new_action,
+                is_played=task_failed._is_played,
                 result=task_failed.Reason,
                 timestamp=task_failed.timestamp,
                 id_=task_failed.TaskScheduledId,
diff --git a/azure/durable_functions/tasks/create_timer.py b/azure/durable_functions/tasks/create_timer.py
@@ -33,7 +33,8 @@ def create_timer_task(state: List[HistoryEvent],
         return TimerTask(
             is_completed=True, action=new_action,
             timestamp=timer_fired.timestamp,
-            id_=timer_fired.event_id)
+            id_=timer_fired.event_id,
+            is_played=timer_fired.is_played)
     else:
         return TimerTask(
             is_completed=False, action=new_action,
diff --git a/azure/durable_functions/tasks/task_all.py b/azure/durable_functions/tasks/task_all.py
@@ -1,6 +1,9 @@
-from typing import List
+from datetime import datetime
+from typing import List, Optional, Any
+
 from ..models.Task import Task
 from ..models.TaskSet import TaskSet
+from ..models.actions import Action
 
 
 def task_all(tasks: List[Task]):
@@ -16,31 +19,57 @@ def task_all(tasks: List[Task]):
     TaskSet
         A Durable Task Set that reports the state of running all of the tasks within it.
     """
-    all_actions = []
-    results = []
+    # Args for constructing the output TaskSet
+    is_played = True
+    is_faulted = False
     is_completed = True
-    complete_time = None
-    faulted = []
+
+    actions: List[Action] = []
+    results: List[Any] = []
+
+    exception: Optional[str] = None
+    end_time: Optional[datetime] = None
+
     for task in tasks:
+        # Add actions and results
         if isinstance(task, TaskSet):
-            for action in task.actions:
-                all_actions.append(action)
+            actions.extend(task.actions)
         else:
-            all_actions.append(task.action)
+            # We know it's an atomic Task
+            actions.append(task.action)
         results.append(task.result)
 
-        if task.is_faulted:
-            faulted.append(task.exception)
+        # Record first exception, if it exists
+        if task.is_faulted and not is_faulted:
+            is_faulted = True
+            exception = task.exception
 
+        # If any task is not played, TaskSet is not played
+        if not task._is_played:
+            is_played = False
+
+        # If any task is incomplete, TaskSet is incomplete
+        # If the task is complete, we can update the end_time
         if not task.is_completed:
             is_completed = False
+        elif end_time is None:
+            end_time = task.timestamp
         else:
-            complete_time = task.timestamp if complete_time is None \
-                else max([task.timestamp, complete_time])
-
-    if len(faulted) > 0:
-        return TaskSet(is_completed, all_actions, results, is_faulted=True, exception=faulted[0])
-    if is_completed:
-        return TaskSet(is_completed, all_actions, results, False, complete_time)
-    else:
-        return TaskSet(is_completed, all_actions, None)
+            end_time = max([task.timestamp, end_time])
+
+    # Incomplete TaskSets do not have results or end-time
+    if not is_completed:
+        results = None
+        end_time = None
+
+    # Construct TaskSet
+    taskset = TaskSet(
+        is_completed=is_completed,
+        actions=actions,
+        result=results,
+        is_faulted=is_faulted,
+        timestamp=end_time,
+        exception=exception,
+        is_played=is_played
+    )
+    return taskset
diff --git a/azure/durable_functions/tasks/timer_task.py b/azure/durable_functions/tasks/timer_task.py
@@ -15,14 +15,15 @@ class TimerTask(Task):
     ```
     """
 
-    def __init__(self, action: CreateTimerAction, is_completed, timestamp, id_):
+    def __init__(self, action: CreateTimerAction, is_completed, timestamp, id_, is_played=False):
         self._action: CreateTimerAction = action
         self._is_completed = is_completed
         self._timestamp = timestamp
         self._id = id_
 
         super().__init__(self._is_completed, False,
                          self._action, None, self._timestamp, self._id, None)
+        self._is_played = is_played
 
     def is_cancelled(self) -> bool:
         """Check of a timer is cancelled.
diff --git a/azure/durable_functions/tasks/wait_for_external_event.py b/azure/durable_functions/tasks/wait_for_external_event.py
@@ -34,6 +34,7 @@ def wait_for_external_event_task(
             is_completed=True,
             is_faulted=False,
             action=new_action,
+            is_played=event_raised._is_played,
             result=parse_history_event(event_raised),
             timestamp=event_raised.timestamp,
             id_=event_raised.event_id)
diff --git a/tests/orchestrator/orchestrator_test_utils.py b/tests/orchestrator/orchestrator_test_utils.py
@@ -50,6 +50,17 @@ def get_orchestration_state_result(
     result = json.loads(result_of_handle)
     return result
 
+def get_orchestration_property(
+        context_builder,
+        activity_func: Callable[[DurableOrchestrationContext], Iterator[Any]],
+        prop: str):
+    context_as_string = context_builder.to_json_string()
+    orchestrator = Orchestrator(activity_func)
+    result_of_handle = orchestrator.handle(
+        DurableOrchestrationContext.from_json(context_as_string))
+    result = getattr(orchestrator, prop)
+    return result
+
 
 def assert_valid_schema(orchestration_state):
     validation_results = validate(instance=orchestration_state, schema=schema)
diff --git a/tests/orchestrator/test_create_timer.py b/tests/orchestrator/test_create_timer.py
@@ -28,8 +28,8 @@ def generator_function(context):
     return "Done!"
 
 def add_timer_action(state: OrchestratorState, fire_at: datetime):
-    action = CreateTimerAction(fire_at= fire_at)
-    state._actions.append([action]) # Todo: brackets?
+    action = CreateTimerAction(fire_at=fire_at)
+    state._actions.append([action])
 
 def test_timers_comparison_with_relaxed_precision():
     """Test if that two `datetime` different but equivalent
diff --git a/tests/orchestrator/test_is_replaying_flag.py b/tests/orchestrator/test_is_replaying_flag.py
@@ -0,0 +1,72 @@
+from tests.test_utils.ContextBuilder import ContextBuilder
+from .orchestrator_test_utils \
+    import get_orchestration_property, assert_orchestration_state_equals, assert_valid_schema
+from azure.durable_functions.models.actions.CreateTimerAction import CreateTimerAction
+from azure.durable_functions.models.OrchestratorState import OrchestratorState
+from azure.durable_functions.constants import DATETIME_STRING_FORMAT
+from datetime import datetime, timedelta, timezone
+
+def generator_function(context):
+    # Create a timezone aware datetime object, just like a normal
+    # call to `context.current_utc_datetime` would create
+    timestamp = "2020-07-23T21:56:54.936700Z"
+    deadline = datetime.strptime(timestamp, DATETIME_STRING_FORMAT)
+    deadline = deadline.replace(tzinfo=timezone.utc)
+
+    for _ in range(0, 3):
+        deadline = deadline + timedelta(seconds=30)
+        yield context.create_timer(deadline)
+
+def base_expected_state(output=None) -> OrchestratorState:
+    return OrchestratorState(is_done=False, actions=[], output=output)
+
+def add_timer_fired_events(context_builder: ContextBuilder, id_: int, timestamp: str,
+        is_played: bool = True):
+    fire_at: str = context_builder.add_timer_created_event(id_, timestamp)
+    context_builder.add_orchestrator_completed_event()
+    context_builder.add_orchestrator_started_event()
+    context_builder.add_timer_fired_event(id_=id_, fire_at=fire_at, is_played=is_played)
+
+def add_timer_action(state: OrchestratorState, fire_at: datetime):
+    action = CreateTimerAction(fire_at=fire_at)
+    state._actions.append([action])
+
+def test_is_replaying_initial_value():
+
+    context_builder = ContextBuilder("")
+    result = get_orchestration_property(
+        context_builder, generator_function, "durable_context")
+
+    assert result.is_replaying == False
+
+def test_is_replaying_one_replayed_event():
+
+    timestamp = "2020-07-23T21:56:54.9367Z"
+    fire_at = datetime.strptime(timestamp, DATETIME_STRING_FORMAT) + timedelta(seconds=30)
+    fire_at_str = fire_at.strftime(DATETIME_STRING_FORMAT)
+
+    context_builder = ContextBuilder("")
+    add_timer_fired_events(context_builder, 0, fire_at_str, is_played=True)
+
+    result = get_orchestration_property(
+        context_builder, generator_function, "durable_context")
+
+    assert result.is_replaying == True
+
+def test_is_replaying_one_replayed_one_not():
+
+    timestamp = "2020-07-23T21:56:54.9367Z"
+    fire_at = datetime.strptime(timestamp, DATETIME_STRING_FORMAT) + timedelta(seconds=30)
+    fire_at_str = fire_at.strftime(DATETIME_STRING_FORMAT)
+    fire_at2 = datetime.strptime(timestamp, DATETIME_STRING_FORMAT) + timedelta(seconds=60)
+    fire_at_str2 = fire_at2.strftime(DATETIME_STRING_FORMAT)
+
+    context_builder = ContextBuilder("")
+    add_timer_fired_events(context_builder, 0, fire_at_str, is_played=True)
+    add_timer_fired_events(context_builder, 1, fire_at_str2, is_played=False)
+
+
+    result = get_orchestration_property(
+        context_builder, generator_function, "durable_context")
+
+    assert result.is_replaying == False
diff --git a/tests/test_utils/ContextBuilder.py b/tests/test_utils/ContextBuilder.py
@@ -93,8 +93,8 @@ def add_timer_created_event(self, id_: int, timestamp: str = None):
         self.history_events.append(event)
         return fire_at
 
-    def add_timer_fired_event(self, id_: int, fire_at: str):
-        event = self.get_base_event(HistoryEventType.TIMER_FIRED, is_played=True)
+    def add_timer_fired_event(self, id_: int, fire_at: str, is_played: bool = True):
+        event = self.get_base_event(HistoryEventType.TIMER_FIRED, is_played=is_played)
         event.TimerId = id_
         event.FireAt = fire_at
         self.history_events.append(event)
