diff --git a/changelog/fixed/snoozing-sloth.md b/changelog/fixed/snoozing-sloth.md new file mode 100644 index 0000000000..5bc672cf75 --- /dev/null +++ b/changelog/fixed/snoozing-sloth.md @@ -0,0 +1 @@ +Cancelling a workflow or local activity while the local activity waits on its retry backoff timer now cancels the timer, instead of failing every workflow task with a `Missing associated machine for LocalActivity` nondeterminism error. diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 1088f80a46..31b655b90c 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2087,10 +2087,8 @@ def _outbound_schedule_activity( # Function that runs in the handle async def run_activity() -> Any: + handle._started = True while True: - # Mark it as started each loop because backoff could cause it to - # be marked as unstarted - handle._started = True try: return await self._await_temporal_operation( handle._result_fut, @@ -3455,8 +3453,8 @@ def _resolve_backoff( self._result_fut.set_exception(_ActivityDoBackoffError(backoff)) # Replace the result future since that one is resolved now self._result_fut = self._instance.create_future() - # Mark this as not started since it's a new future - self._started = False + # Leave _started set: run_activity is running and traps a cancel in its + # backoff sleep, and Core has no command for the new seq to cancel yet def _apply_schedule_command( self, diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 01a2fcea89..876755d729 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -3487,6 +3487,72 @@ async def test_workflow_local_activity_backoff(client: Client): ) +@activity.defn +async def retry_later_activity() -> None: + # A delay above the default 60s local_retry_threshold makes Core back off + # with a workflow timer instead of retrying in memory + raise ApplicationError("Retry later", next_retry_delay=timedelta(seconds=90)) + + +@workflow.defn +class LocalActivityBackoffCancelWorkflow: + def __init__(self) -> None: + self._cancel_activity = False + + @workflow.run + async def run(self, cancel_from_workflow: bool) -> str: + handle = workflow.start_local_activity( + retry_later_activity, start_to_close_timeout=timedelta(seconds=10) + ) + if cancel_from_workflow: + await workflow.wait_condition(lambda: self._cancel_activity) + handle.cancel() + try: + await handle + except asyncio.CancelledError: + return "activity cancelled" + await handle + return "activity completed" + + @workflow.signal + def cancel_activity(self) -> None: + self._cancel_activity = True + + +@pytest.mark.parametrize("cancel_from_workflow", [False, True]) +async def test_workflow_local_activity_cancel_during_backoff( + client: Client, cancel_from_workflow: bool +): + async with new_worker( + client, LocalActivityBackoffCancelWorkflow, activities=[retry_later_activity] + ) as worker: + handle = await client.start_workflow( + LocalActivityBackoffCancelWorkflow.run, + cancel_from_workflow, + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + + async def backoff_timer_started() -> None: + async for e in handle.fetch_history_events(): + if e.event_type == EventType.EVENT_TYPE_TIMER_STARTED: + return + assert False, "Backoff timer not started" + + await assert_eventually(backoff_timer_started) + if cancel_from_workflow: + await handle.signal(LocalActivityBackoffCancelWorkflow.cancel_activity) + assert "activity cancelled" == await handle.result() + else: + await handle.cancel() + with pytest.raises(WorkflowFailureError) as err: + await handle.result() + assert isinstance(err.value.cause, CancelledError) + event_types = [e.event_type async for e in handle.fetch_history_events()] + assert EventType.EVENT_TYPE_TIMER_CANCELED in event_types + assert EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types + + deadlock_thread_event = threading.Event()