diff --git a/azure/durable_functions/models/Task.py b/azure/durable_functions/models/Task.py index e5667001..447ee540 100644 --- a/azure/durable_functions/models/Task.py +++ b/azure/durable_functions/models/Task.py @@ -380,6 +380,8 @@ def cancel(self): if (self.result): raise Exception("Cannot cancel a completed task.") self.action.is_cancelled = True + for timer in self.pending_tasks: + timer.cancel() def try_set_value(self, child: TimerTask): """Transition this LongTimer Task to a terminal state and set its value. @@ -394,7 +396,7 @@ def try_set_value(self, child: TimerTask): """ current_time = self._orchestration_context.current_utc_datetime final_fire_time = self.action.fire_at - if final_fire_time > current_time: + if not self.is_canceled() and 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) diff --git a/tests/tasks/test_long_timers.py b/tests/tasks/test_long_timers.py index e3dceb23..4f16379f 100644 --- a/tests/tasks/test_long_timers.py +++ b/tests/tasks/test_long_timers.py @@ -1,9 +1,13 @@ import datetime +import json import pytest from azure.durable_functions.models.DurableOrchestrationContext import DurableOrchestrationContext from azure.durable_functions.models.Task import LongTimerTask, TaskState, TimerTask from azure.durable_functions.models.actions.CreateTimerAction import CreateTimerAction +from azure.durable_functions.models.ReplaySchema import ReplaySchema +from azure.durable_functions.orchestrator import Orchestrator +from tests.test_utils.ContextBuilder import ContextBuilder @pytest.fixture @@ -68,3 +72,81 @@ def test_long_timer_fires_appropriately(starting_context_v3): # Ensure the LongTimerTask finished assert len(long_timer.pending_tasks) == 0 assert long_timer.state == TaskState.SUCCEEDED + + +@pytest.mark.parametrize("completed_chunk", [False, True]) +def test_cancel_long_timer_cancels_pending_chunk(starting_context_v3, completed_chunk): + context = starting_context_v3 + timer = context.create_timer(context.current_utc_datetime + datetime.timedelta(days=2)) + first_chunk = next(iter(timer.pending_tasks)) + if completed_chunk: + context.current_utc_datetime = first_chunk.action_repr.fire_at + first_chunk.set_value(False, None) + + pending_chunk = next(iter(timer.pending_tasks)) + timer.cancel() + timer.cancel() # Like short timers, cancellation is idempotent before completion. + + assert timer.is_canceled() + assert pending_chunk.is_cancelled + if completed_chunk: + assert not first_chunk.is_cancelled + + +@pytest.mark.parametrize("completed_chunk", [False, True]) +def test_approval_cancels_serialized_long_timer(completed_chunk): + def approval(context): + event = context.wait_for_external_event("approved") + timer = context.create_timer(context.current_utc_datetime + datetime.timedelta(days=2)) + winner = yield context.task_any([event, timer]) + if winner == event: + timer.cancel() + return True + return False + + builder = ContextBuilder("approval", increase_time=False, replay_schema=ReplaySchema.V3) + first_fire = builder.current_datetime + datetime.timedelta(hours=8) + builder.add_timer_created_event(0, first_fire.isoformat()) + builder.add_orchestrator_completed_event() + if completed_chunk: + builder.current_datetime = first_fire + builder.add_orchestrator_started_event() + builder.add_timer_fired_event(0, first_fire.isoformat()) + second_fire = first_fire + datetime.timedelta(hours=8) + builder.add_timer_created_event(1, second_fire.isoformat()) + builder.add_orchestrator_completed_event() + builder.add_orchestrator_started_event() + builder.add_event_raised_event("approved", -1, "true") + context = DurableOrchestrationContext.from_json(builder.to_json_string( + maximumShortTimerDuration="0.16:00:00", + longRunningTimerIntervalDuration="0.08:00:00")) + + result = json.loads(Orchestrator(approval).handle(context)) + assert result["isDone"] + assert result["output"] is True + + def timer_actions(actions): + for action in actions: + if "isCanceled" in action: + yield action + yield from timer_actions(action.get("compoundActions", [])) + + timers = list(timer_actions([action for group in result["actions"] for action in group])) + assert len(timers) == (2 if completed_chunk else 1) + assert timers[-1]["isCanceled"] + if completed_chunk: + assert not timers[0]["isCanceled"] + + +def test_cancelled_long_timer_does_not_schedule_another_chunk(starting_context_v3): + context = starting_context_v3 + timer = context.create_timer(context.current_utc_datetime + datetime.timedelta(days=2)) + chunk = next(iter(timer.pending_tasks)) + timer.cancel() + + # A firing already in history must not restart the cancelled timer. + context.current_utc_datetime = chunk.action_repr.fire_at + chunk.set_value(False, None) + + assert not timer.pending_tasks + assert not any(context._actions)