PR Feedback

Author: andystaples

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-ignore = E501,C901
+ignore = E501,C901,W503
 exclude = 
     .git
     *_pb2*

docs/supported-patterns.md

@@ -64,12 +64,12 @@ def purchase_order_workflow(ctx: task.OrchestrationContext, order: Order):
     # Orders of $1000 or more require manager approval
     yield ctx.call_activity(send_approval_request, input=order)
 
-    # Approvals must be received within 24 hours or they will be canceled.
+    # Approvals must be received within 24 hours or they will be cancelled.
     approval_event = ctx.wait_for_external_event("approval_received")
     timeout_event = ctx.create_timer(timedelta(hours=24))
     winner = yield task.when_any([approval_event, timeout_event])
     if winner == timeout_event:
-        return "Canceled"
+        return "Cancelled"
 
     # The order was approved
     yield ctx.call_activity(place_order, input=order)

durabletask/task.py

@@ -228,7 +228,7 @@ def call_sub_orchestrator(self, orchestrator: Union[Orchestrator[TInput, TOutput
         """
         pass
 
-    # TOOD: Add a timeout parameter, which allows the task to be canceled if the event is
+    # TOOD: Add a timeout parameter, which allows the task to be cancelled if the event is
     # not received within the specified timeout. This requires support for task cancellation.
     @abstractmethod
     def wait_for_external_event(self, name: str) -> CancellableTask:
@@ -324,8 +324,8 @@ class OrchestrationStateError(Exception):
     pass
 
 
-class TaskCanceledError(Exception):
-    """Exception type for canceled orchestration tasks."""
+class TaskCancelledError(Exception):
+    """Exception type for cancelled orchestration tasks."""
 
 
 class Task(ABC, Generic[T]):
@@ -440,7 +440,7 @@ def fail(self, message: str, details: Union[Exception, pb.TaskFailureDetails]):
 
 
 class CancellableTask(CompletableTask[T]):
-    """A completable task that can be canceled before it finishes."""
+    """A completable task that can be cancelled before it finishes."""
 
     def __init__(self) -> None:
         super().__init__()
@@ -449,12 +449,12 @@ def __init__(self) -> None:
 
     @property
     def is_cancelled(self) -> bool:
-        """Returns True if the task was canceled, False otherwise."""
+        """Returns True if the task was cancelled, False otherwise."""
         return self._is_cancelled
 
     def get_result(self) -> T:
         if self._is_cancelled:
-            raise TaskCanceledError('The task was canceled.')
+            raise TaskCancelledError('The task was cancelled.')
         return super().get_result()
 
     def set_cancel_handler(self, cancel_handler: Callable[[], None]) -> None:
@@ -524,27 +524,26 @@ class TimerTask(CancellableTask[None]):
     def set_retryable_parent(self, retryable_task: RetryableTask):
         self._retryable_parent = retryable_task
 
-    def complete(self, *args, **kwargs):
+    def complete(self, _: datetime) -> None:
         super().complete(None)
 
-
 class LongTimerTask(TimerTask):
-    def __init__(self, final_fire_at: datetime, maximum_timer_duration: timedelta):
+    def __init__(self, final_fire_at: datetime, maximum_timer_interval: timedelta):
         super().__init__()
         self._final_fire_at = final_fire_at
-        self._maximum_timer_duration = maximum_timer_duration
+        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 complete(self, current_utc_datetime: datetime):
+    def complete(self, current_utc_datetime: datetime) -> Optional[datetime]:
         if current_utc_datetime < self._final_fire_at:
             return self._get_next_fire_at(current_utc_datetime)
-        super().complete(None)
+        return super().complete(current_utc_datetime)
 
     def _get_next_fire_at(self, current_utc_datetime: datetime) -> datetime:
-        if current_utc_datetime + self._maximum_timer_duration < self._final_fire_at:
-            return current_utc_datetime + self._maximum_timer_duration
+        if current_utc_datetime + self._maximum_timer_interval < self._final_fire_at:
+            return current_utc_datetime + self._maximum_timer_interval
         return self._final_fire_at
 
 

durabletask/worker.py

@@ -646,7 +646,9 @@ def _execute_orchestrator(
             completionToken,
     ):
         try:
-            executor = _OrchestrationExecutor(self._registry, self._logger)
+            executor = _OrchestrationExecutor(self._registry,
+                                              self._logger,
+                                              self.maximum_timer_interval)
             result = executor.execute(req.instanceId, req.pastEvents, req.newEvents)
             res = pb.OrchestratorResponse(
                 instanceId=req.instanceId,
@@ -1029,7 +1031,7 @@ def set_custom_status(self, custom_status: Any) -> None:
             shared.to_json(custom_status) if custom_status is not None else None
         )
 
-    def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
+    def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.CancellableTask:
         return self.create_timer_internal(fire_at)
 
     def create_timer_internal(
@@ -1045,7 +1047,11 @@ def create_timer_internal(
 
         next_fire_at: datetime = final_fire_at
 
-        if self._maximum_timer_interval is not None and self.current_utc_datetime + self._maximum_timer_interval < final_fire_at:
+        if (
+            self._maximum_timer_interval is not None
+            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)
         else:

examples/human_interaction.py

@@ -48,7 +48,7 @@ def purchase_order_workflow(ctx: task.OrchestrationContext, order: Order):
     # Orders of $1000 or more require manager approval
     yield ctx.call_activity(send_approval_request, input=order)
 
-    # Approvals must be received within 24 hours or they will be canceled.
+    # Approvals must be received within 24 hours or they will be cancelled.
     approval_event = ctx.wait_for_external_event("approval_received")
     timeout_event = ctx.create_timer(timedelta(hours=24))
     winner = yield task.when_any([approval_event, timeout_event])

tests/durabletask/test_orchestration_e2e.py

@@ -654,3 +654,45 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert state.failure_details is None
     expected = "approved" if winning_event == "Approve" else "rejected"
     assert state.serialized_output == json.dumps(expected)
+
+
+def test_long_timer_chunking():
+    """Verify that a timer longer than maximum_timer_interval is broken into
+    intermediate chunks and that the orchestration completes correctly.
+
+    The worker is configured with a 2-second maximum_timer_interval.  The
+    orchestrator requests a 5-second timer, which requires 3 chunks
+    (0→2s, 2→4s, 4→5s).  Each chunk causes a full orchestrator replay, so
+    the orchestrator function is invoked once for the initial scheduling and
+    once more for each timerFired event — 4 invocations in total.  Asserting
+    invocation_count >= 4 confirms that intermediate chunks actually fired
+    rather than the timer being scheduled as a single unit.
+    """
+
+    invocation_count = 0
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        nonlocal invocation_count
+        invocation_count += 1
+        yield ctx.create_timer(timedelta(seconds=5))
+        return "done"
+
+    with worker.TaskHubGrpcWorker(
+        host_address=HOST,
+        maximum_timer_interval=timedelta(seconds=2),
+    ) as w:
+        w.add_orchestrator(orchestrator)
+        w.start()
+
+        task_hub_client = client.TaskHubGrpcClient(host_address=HOST)
+        id = task_hub_client.schedule_new_orchestration(orchestrator)
+        state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
+
+    assert state is not None
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.failure_details is None
+    assert state.serialized_output == json.dumps("done")
+    # 3 chunks (0→2s, 2→4s, 4→5s) produce 4 total orchestrator invocations
+    # (initial scheduling + one replay per timerFired).  >= 4 proves that at
+    # least two intermediate chunk timers fired rather than one direct timer.
+    assert invocation_count >= 4

tests/durabletask/test_orchestration_executor.py

@@ -380,15 +380,15 @@ def dummy_activity(ctx, _):
     assert not hasattr(activity_task, "is_cancelled")
 
 
-def test_cancelled_task_get_result_raises_task_canceled_error():
-    """Tests that canceled cancellable tasks raise TaskCanceledError from get_result."""
+def test_cancelled_task_get_result_raises_task_cancelled_error():
+    """Tests that cancelled cancellable tasks raise TaskCancelledError from get_result."""
 
     cancellable_task = task.CancellableTask()
 
     assert cancellable_task.cancel() is True
     assert cancellable_task.is_cancelled is True
 
-    with pytest.raises(task.TaskCanceledError):
+    with pytest.raises(task.TaskCancelledError):
         cancellable_task.get_result()