Add long timers to Python

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from azure.durable_functions.models.actions.SignalEntityAction import SignalEntityAction
 from azure.durable_functions.models.actions.CallEntityAction import CallEntityAction
-from azure.durable_functions.models.Task import TaskBase, TimerTask
+from azure.durable_functions.models.Task import LongTimerTask, TaskBase, TimerTask
 from azure.durable_functions.models.actions.CallHttpAction import CallHttpAction
 from azure.durable_functions.models.DurableHttpRequest import DurableHttpRequest
 from azure.durable_functions.models.actions.CallSubOrchestratorWithRetryAction import \
@@ -26,6 +26,8 @@
 from uuid import UUID, uuid5, NAMESPACE_URL, NAMESPACE_OID
 from datetime import timezone
 
+from azure.durable_functions.models.utils.json_utils import parse_timespan_attrib
+
 from .RetryOptions import RetryOptions
 from .FunctionContext import FunctionContext
 from .history import HistoryEvent, HistoryEventType
@@ -48,11 +50,22 @@ class DurableOrchestrationContext:
     # parameter names are as defined by JSON schema and do not conform to PEP8 naming conventions
     def __init__(self,
                  history: List[Dict[Any, Any]], instanceId: str, isReplaying: bool,
-                 parentInstanceId: str, input: Any = None, upperSchemaVersion: int = 0, **kwargs):
+                 parentInstanceId: str, input: Any = None, upperSchemaVersion: int = 0,
+                 maximumShortTimerDuration: str = None,
+                 longRunningTimerIntervalDuration: str = None, upperSchemaVersionNew: int = None,
+                 **kwargs):
         self._histories: List[HistoryEvent] = [HistoryEvent(**he) for he in history]
         self._instance_id: str = instanceId
         self._is_replaying: bool = isReplaying
         self._parent_instance_id: str = parentInstanceId
+        self._maximum_short_timer_duration: datetime.timedelta = None
+        if maximumShortTimerDuration is not None:
+            max_short_duration = parse_timespan_attrib(maximumShortTimerDuration)
+            self._maximum_short_timer_duration = max_short_duration
+        self._long_timer_interval_duration: datetime.timedelta = None
+        if longRunningTimerIntervalDuration is not None:
+            long_interval_duration = parse_timespan_attrib(longRunningTimerIntervalDuration)
+            self._long_timer_interval_duration = long_interval_duration
         self._custom_status: Any = None
         self._new_uuid_counter: int = 0
         self._sub_orchestrator_counter: int = 0
@@ -66,6 +79,13 @@ def __init__(self,
         self._function_context: FunctionContext = FunctionContext(**kwargs)
         self._sequence_number = 0
         self._replay_schema = ReplaySchema(upperSchemaVersion)
+        if (upperSchemaVersionNew is not None
+                and upperSchemaVersionNew > self._replay_schema.value):
+            valid_schema_values = [enum_member.value for enum_member in ReplaySchema]
+            if upperSchemaVersionNew in valid_schema_values:
+                self._replay_schema = ReplaySchema(upperSchemaVersionNew)
+            else:
+                self._replay_schema = ReplaySchema(max(valid_schema_values))
 
         self._action_payload_v1: List[List[Action]] = []
         self._action_payload_v2: List[Action] = []
@@ -471,6 +491,37 @@ def parent_instance_id(self) -> str:
         """
         return self._parent_instance_id
 
+    @property
+    def maximum_short_timer_duration(self) -> datetime.timedelta:
+        """Get the maximum duration for a short timer.
+
+        The maximum length of a "short timer" is defined by the storage backend.
+        Some storage backends have a maximum future date for scheduled tasks, and
+        so for timers longer than this duration, we must simulate a long timer by
+        waiting in chunks.
+
+        Returns
+        -------
+        str
+            Maximum allowable duration for a short timer in Durable
+        """
+        return self._maximum_short_timer_duration
+
+    @property
+    def long_timer_interval_duration(self) -> datetime.timedelta:
+        """Get the interval for long timers.
+
+        When a timer is scheduled for a duration longer than the maximum short timer
+        duration, the timer is set to run in chunks of time. The long running timer
+        interval duration defines how long these chunks of time should be.
+
+        Returns
+        -------
+        str
+            Duration for intervals of a long-running timer
+        """
+        return self._long_timer_interval_duration
+
     @property
     def current_utc_datetime(self) -> datetime.datetime:
         """Get the current date/time.
@@ -532,10 +583,10 @@ def _record_fire_and_forget_action(self, action: Action):
             The action to append
         """
         new_action: Union[List[Action], Action]
-        if self._replay_schema is ReplaySchema.V2:
-            new_action = action
-        else:
+        if self._replay_schema is ReplaySchema.V1:
             new_action = [action]
+        else:
+            new_action = action
         self._add_to_actions(new_action)
         self._sequence_number += 1
 
@@ -580,6 +631,23 @@ def create_timer(self, fire_at: datetime.datetime) -> TaskBase:
         TaskBase
             A Durable Timer Task that schedules the timer to wake up the activity
         """
+        if self._replay_schema.value >= ReplaySchema.V3.value:
+            if not self.maximum_short_timer_duration or not self.long_timer_interval_duration:
+                raise Exception(
+                    "A framework-internal error was detected: "
+                    "replay schema version >= V3 is being used, "
+                    "but one or more of the properties `maximumShortTimerDuration`"
+                    "and `longRunningTimerIntervalDuration` are not defined. "
+                    "This is likely an issue with the Durable Functions Extension. "
+                    "Please report this bug here: "
+                    "https://github.com/Azure/azure-functions-durable-python/issues\n"
+                    f"maximumShortTimerDuration: {self.maximum_short_timer_duration}\n"
+                    f"longRunningTimerIntervalDuration: {self.long_timer_interval_duration}"
+                )
+            if fire_at > self.current_utc_datetime + self.maximum_short_timer_duration:
+                action = CreateTimerAction(fire_at)
+                return LongTimerTask(None, action, self)
+
         action = CreateTimerAction(fire_at)
         task = self._generate_task(action, task_constructor=TimerTask)
         return task
@@ -656,7 +724,8 @@ def _add_to_actions(self, action_repr: Union[List[Action], Action]):
 
         if self._replay_schema is ReplaySchema.V1 and isinstance(action_repr, list):
             self._action_payload_v1.append(action_repr)
-        elif self._replay_schema is ReplaySchema.V2 and isinstance(action_repr, Action):
+        elif (self._replay_schema.value >= ReplaySchema.V2.value
+              and isinstance(action_repr, Action)):
             self._action_payload_v2.append(action_repr)
         else:
             raise Exception(f"DF-internal exception: ActionRepr of signature {type(action_repr)}"

azure/durable_functions/models/ReplaySchema.py

@@ -6,3 +6,4 @@ class ReplaySchema(Enum):
 
     V1 = 0
     V2 = 1
+    V3 = 2

azure/durable_functions/models/Task.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from azure.durable_functions.models.actions.NoOpAction import NoOpAction
 from azure.durable_functions.models.actions.CompoundAction import CompoundAction
 from azure.durable_functions.models.RetryOptions import RetryOptions
@@ -170,7 +171,7 @@ def __init__(self, tasks: List[TaskBase], compound_action_constructor=None):
                     child_actions.append(action_repr)
         if compound_action_constructor is None:
             self.action_repr = child_actions
-        else:  # replay_schema is ReplaySchema.V2
+        else:  # replay_schema >= ReplaySchema.V2
             self.action_repr = compound_action_constructor(child_actions)
         self._first_error: Optional[Exception] = None
         self.pending_tasks: Set[TaskBase] = set(tasks)
@@ -292,7 +293,7 @@ def __init__(self, task: List[TaskBase], replay_schema: ReplaySchema):
             The ReplaySchema, which determines the inner action payload representation
         """
         compound_action_constructor = None
-        if replay_schema is ReplaySchema.V2:
+        if replay_schema.value >= ReplaySchema.V2.value:
             compound_action_constructor = WhenAllAction
         super().__init__(task, compound_action_constructor)
 
@@ -317,6 +318,119 @@ def try_set_value(self, child: TaskBase):
                 self.set_value(is_error=True, value=self._first_error)
 
 
+class LongTimerTask(WhenAllTask):
+    """A Timer Task for intervals longer than supported by the storage backend."""
+
+    def __init__(self, id, action: CreateTimerAction, orchestration_context):
+        """Initialize a LongTimerTask.
+
+        Parameters
+        ----------
+        id_ : int
+            An ID for the task
+        action : CreateTimerAction
+            The action this task represents
+        orchestration_context: DurableOrchestrationContext
+            The orchestration context this task was created in
+        """
+        current_time = orchestration_context.current_utc_datetime
+        final_fire_time = action.fire_at
+        duration_until_fire = final_fire_time - current_time
+
+        if duration_until_fire > orchestration_context.maximum_short_timer_duration:
+            next_fire_time = current_time + orchestration_context.long_timer_interval_duration
+        else:
+            next_fire_time = final_fire_time
+
+        next_timer_action = CreateTimerAction(next_fire_time)
+        next_timer_task = TimerTask(None, next_timer_action)
+        super().__init__([next_timer_task], orchestration_context._replay_schema)
+
+        self.id = id
+        self.action = action
+        self.orchestration_context = orchestration_context
+        self.maximum_short_timer_duration = self.orchestration_context.maximum_short_timer_duration
+        self.long_timer_interval_duration = self.orchestration_context.long_timer_interval_duration
+
+    def is_canceled(self) -> bool:
+        """Check if the LongTimer is cancelled.
+
+        Returns
+        -------
+        bool
+            Returns whether the timer has been cancelled or not
+        """
+        return self.action.is_cancelled
+
+    def cancel(self):
+        """Cancel a timer.
+
+        Raises
+        ------
+        ValueError
+            Raises an error if the task is already completed and an attempt is made to cancel it
+        """
+        if (self.result):
+            raise Exception("Cannot cancel a completed task.")
+        self.action.is_cancelled = True
+
+    def try_set_value(self, child: TimerTask):
+        """Transition this LongTimer Task to a terminal state and set its value.
+
+        If the LongTimer has not yet reached the designated completion time, starts a new
+        TimerTask for the next interval and does not close.
+
+        Parameters
+        ----------
+        child : TimerTask
+            A timer sub-task that just completed
+        """
+        current_time = self.orchestration_context.current_utc_datetime
+        final_fire_time = self.action.fire_at
+        if final_fire_time > current_time:
+            next_timer = self.get_next_timer_task(final_fire_time, current_time)
+            self.add_new_child(next_timer)
+        return super().try_set_value(child)
+
+    def get_next_timer_task(self, final_fire_time: datetime, current_time: datetime) -> TimerTask:
+        """Create a TimerTask to represent the next interval of the LongTimer.
+
+        Parameters
+        ----------
+        final_fire_time : datetime.datetime
+            The final firing time of the LongTimer
+        current_time : datetime.datetime
+            The current time
+
+        Returns
+        -------
+        TimerTask
+            A TimerTask representing the next interval of the LongTimer
+        """
+        duration_until_fire = final_fire_time - current_time
+        if duration_until_fire > self.maximum_short_timer_duration:
+            next_fire_time = current_time + self.long_timer_interval_duration
+        else:
+            next_fire_time = final_fire_time
+        return TimerTask(None, CreateTimerAction(next_fire_time))
+
+    def add_new_child(self, child_timer: TimerTask):
+        """Add the TimerTask to this task's children.
+
+        Also register the TimerTask with the orchestration context.
+
+        Parameters
+        ----------
+        child_timer : TimerTask
+            The newly created TimerTask to add
+        """
+        child_timer.parent = self
+        self.pending_tasks.add(child_timer)
+        self.orchestration_context._add_to_open_tasks(child_timer)
+        self.orchestration_context._add_to_actions(child_timer.action_repr)
+        child_timer._set_is_scheduled(True)
+
+
 class WhenAnyTask(CompoundTask):
     """A Task representing `when_any` scenarios."""
 
@@ -331,7 +445,7 @@ def __init__(self, task: List[TaskBase], replay_schema: ReplaySchema):
             The ReplaySchema, which determines the inner action payload representation
         """
         compound_action_constructor = None
-        if replay_schema is ReplaySchema.V2:
+        if replay_schema.value >= ReplaySchema.V2.value:
             compound_action_constructor = WhenAnyAction
         super().__init__(task, compound_action_constructor)
 

azure/durable_functions/models/utils/json_utils.py

@@ -1,3 +1,5 @@
+import datetime
+import re
 from typing import Dict, Any
 
 from ...constants import DATETIME_STRING_FORMAT
@@ -37,6 +39,44 @@ def add_datetime_attrib(json_dict: Dict[str, Any], object_,
             getattr(object_, attribute_name).strftime(DATETIME_STRING_FORMAT)
 
 
+# When we recieve properties from WebJobs extension originally parsed as
+#   TimeSpan objects through Newtonsoft, the format complies with the constant
+#   format specifier for TimeSpan in .NET.
+#   Python offers no convenient way to parse these back into timedeltas,
+#   so we use this regex method instead
+def parse_timespan_attrib(from_str: str) -> datetime.timedelta:
+    """Convert a string representing TimeSpan.ToString("c") in .NET to a python timedelta.
+
+    Parameters
+    ----------
+    from_str: The string format of the TimeSpan to convert
+
+    Returns
+    -------
+    timespan.timedelta
+        The TimeSpan expressed as a Python datetime.timedelta
+
+    """
+    match = re.match(r"^(?P<negative>-)?(?:(?P<days>[0-9]*)\.)?"
+                     r"(?P<hours>[0-9]{2}):(?P<minutes>[0-9]{2})"
+                     r":(?P<seconds>[0-9]{2})(?:\.(?P<ticks>[0-9]{7}))?$",
+                     from_str)
+    if match:
+        groups = match.groupdict()
+        span = datetime.timedelta(
+            days=int(groups['days'] or "0"),
+            hours=int(groups['hours']),
+            minutes=int(groups['minutes']),
+            seconds=int(groups['seconds']),
+            microseconds=int(groups['ticks'] or "0") // 10)
+
+        if groups['negative'] == '-':
+            span = -span
+        return span
+    else:
+        raise Exception(f"Format of TimeSpan failed attempted conversion to timedelta: {from_str}")
+
+
 def add_json_attrib(json_dict: Dict[str, Any], object_,
                     attribute_name: str, alt_name: str = None):
     """Add the results of the to_json() function call of the attribute from the object to the dict.