PR feedback - merge TimerTask/LongTimerTask

Author: andystaples

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -81,5 +81,6 @@ def __init__(self, *,
             log_formatter=log_formatter,
             interceptors=interceptors,
             concurrency_options=concurrency_options,
-            maximum_timer_interval=None  # DTS allows timers of indefinite length
+            # DTS natively supports long timers so chunking is unnecessary
+            maximum_timer_interval=None
         )

durabletask/task.py

@@ -521,26 +521,22 @@ def compute_next_delay(self) -> Optional[timedelta]:
 
 
 class TimerTask(CancellableTask[None]):
-    def set_retryable_parent(self, retryable_task: RetryableTask):
-        self._retryable_parent = retryable_task
-
-    def complete(self, _: datetime) -> None:
-        super().complete(None)
-
-
-class LongTimerTask(TimerTask):
-    def __init__(self, final_fire_at: datetime, maximum_timer_interval: timedelta):
+    def __init__(self, final_fire_at: Optional[datetime] = None,
+                 maximum_timer_interval: Optional[timedelta] = None):
         super().__init__()
         self._final_fire_at = final_fire_at
         self._maximum_timer_interval = maximum_timer_interval
 
-    def start(self, current_utc_datetime: datetime) -> datetime:
-        return self._get_next_fire_at(current_utc_datetime)
+    def set_retryable_parent(self, retryable_task: RetryableTask):
+        self._retryable_parent = retryable_task
 
-    def complete(self, current_utc_datetime: datetime) -> Optional[datetime]:
-        if current_utc_datetime < self._final_fire_at:
+    def _handle_timer_fired(self, current_utc_datetime: datetime) -> Optional[datetime]:
+        if (self._final_fire_at is not None
+                and self._maximum_timer_interval is not None
+                and current_utc_datetime < self._final_fire_at):
             return self._get_next_fire_at(current_utc_datetime)
-        return super().complete(current_utc_datetime)
+        super().complete(None)
+        return None
 
     def _get_next_fire_at(self, current_utc_datetime: datetime) -> datetime:
         if current_utc_datetime + self._maximum_timer_interval < self._final_fire_at:

durabletask/worker.py

@@ -1158,8 +1158,8 @@ def create_timer_internal(
             and self._maximum_timer_interval > timedelta(0)
             and self.current_utc_datetime + self._maximum_timer_interval < final_fire_at
         ):
-            timer_task = task.LongTimerTask(final_fire_at, self._maximum_timer_interval)
-            next_fire_at = timer_task.start(self.current_utc_datetime)
+            timer_task = task.TimerTask(final_fire_at, self._maximum_timer_interval)
+            next_fire_at = timer_task._get_next_fire_at(self.current_utc_datetime)
         else:
             timer_task = task.TimerTask()
 
@@ -1675,7 +1675,7 @@ def process_event(
                             f"{ctx.instance_id}: Ignoring unexpected timerFired event with ID = {timer_id}."
                         )
                     return
-                if not (isinstance(timer_task, task.TimerTask) or isinstance(timer_task, task.LongTimerTask)):
+                if not isinstance(timer_task, task.TimerTask):
                     if not ctx._is_replaying:
                         self._logger.warning(
                             f"{ctx.instance_id}: Ignoring timerFired event with non-timer task ID = {timer_id}."
@@ -1692,7 +1692,7 @@ def process_event(
                             scheduled_time_ns=created_ns,
                             parent_trace_context=ctx._orchestration_trace_context or ctx._parent_trace_context,
                         )
-                next_fire_at = timer_task.complete(event.timerFired.fireAt.ToDatetime())
+                next_fire_at = timer_task._handle_timer_fired(event.timerFired.fireAt.ToDatetime())
                 if next_fire_at is not None:
                     id = ctx.next_sequence_number()
                     new_action = ph.new_create_timer_action(id, next_fire_at)