diff --git a/CHANGELOG.md b/CHANGELOG.md index d0d80bc71..ee61e055d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Cancelling an activity from a signal while the workflow itself is cancelled + no longer causes a nondeterminism error from duplicate activity-cancellation + commands. - `StrandsPlugin` now disables Botocore retries for its default Bedrock model so model request retries are handled exclusively by Temporal. - `temporalio.contrib.openai_agents` now honors the `retry-after-ms` and diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index fc5a6efe3..f7cd66cfe 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2035,7 +2035,7 @@ async def run_activity() -> Any: try: return await self._await_temporal_operation( handle._result_fut, - lambda _err, command: handle._apply_cancel_command(command), + lambda _err: handle._request_cancel(), completed_cancellation_flag=_WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY, ) except _ActivityDoBackoffError as err: @@ -2106,8 +2106,8 @@ async def _outbound_start_child_workflow( # Common code for handling cancel for start and run def apply_child_cancel_error( err: asyncio.CancelledError, - cancel_command: temporalio.bridge.proto.workflow_commands.WorkflowCommand, ) -> None: + cancel_command = self._add_command() # Send a cancel request to the child, forwarding the msg passed to # Task.cancel(msg) (if any) as the cancellation reason. reason = err.args[0] if err.args and isinstance(err.args[0], str) else "" @@ -2171,7 +2171,7 @@ async def operation_handle_fn() -> OutputT: OutputT, await self._await_temporal_operation( handle._result_fut, - lambda _err, command: handle._apply_cancel_command(command), + lambda _err: handle._apply_cancel_command(self._add_command()), ), ) @@ -2194,7 +2194,7 @@ async def operation_handle_fn() -> OutputT: await self._await_temporal_operation( handle._start_fut, - lambda _err, command: handle._apply_cancel_command(command), + lambda _err: handle._apply_cancel_command(self._add_command()), reraise_on_workflow_cancellation=True, ) return handle @@ -2252,10 +2252,7 @@ async def _await_temporal_operation( self, fut: asyncio.Future[_T], apply_cancel: Callable[ - [ - asyncio.CancelledError, - temporalio.bridge.proto.workflow_commands.WorkflowCommand, - ], + [asyncio.CancelledError], None, ], *, @@ -2283,7 +2280,7 @@ async def _await_temporal_operation( ) raise - apply_cancel(err, self._add_command()) + apply_cancel(err) # Clear the cancellation counter on Python 3.11+ so the next # await does not immediately re-raise CancelledError. @@ -2798,8 +2795,8 @@ async def _signal_external_workflow( def apply_cancel( _err: asyncio.CancelledError, - command: temporalio.bridge.proto.workflow_commands.WorkflowCommand, ) -> None: + command = self._add_command() command.cancel_signal_workflow.seq = seq # Wait until completed or cancelled @@ -3281,6 +3278,7 @@ def __init__( self._input = input self._result_fut = instance.create_future() self._started = False + self._cancel_command_seq: int | None = None instance._register_task(self, name=f"activity: {input.activity}") self._payload_converter = self._instance._payload_converter_with_context( temporalio.converter.ActivitySerializationContext( @@ -3307,9 +3305,15 @@ def cancel(self, msg: Any | None = None) -> bool: # to send a cancel command because the async function won't run to trap # the cancel (i.e. cancelled before started) if not self._started and not self.done(): - self._apply_cancel_command(self._instance._add_command()) + self._request_cancel() return super().cancel(msg) + def _request_cancel(self) -> None: + if self._cancel_command_seq == self._seq: + return + self._cancel_command_seq = self._seq + self._apply_cancel_command(self._instance._add_command()) + def _resolve_success(self, result: Any) -> None: # We intentionally let this error if already done self._result_fut.set_result(result) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 14394dd98..fe720564a 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -1045,6 +1045,96 @@ async def activity_result() -> str: await activity_inst.wait_cancel_complete.wait() +@workflow.defn +class CancelActivityDuringWorkflowCancellationWorkflow: + def __init__(self) -> None: + self._activity_started = False + self._cancel_activity = False + + @workflow.run + async def run(self) -> str: + handle = workflow.start_activity( + wait_cancel, + start_to_close_timeout=timedelta(minutes=1), + heartbeat_timeout=timedelta(seconds=1), + ) + self._activity_started = True + + async def cancel_activity() -> None: + await workflow.wait_condition(lambda: self._cancel_activity) + handle.cancel() + + cancel_task = asyncio.create_task(cancel_activity()) + try: + await handle + except ActivityError: + pass + finally: + cancel_task.cancel() + return "activity cancelled" + + @workflow.signal + def cancel_activity(self) -> None: + self._cancel_activity = True + + @workflow.query + def activity_started(self) -> bool: + return self._activity_started + + +async def test_workflow_cancel_activity_while_workflow_cancelled(client: Client): + task_queue = str(uuid.uuid4()) + runner = CustomWorkflowRunner() + handle = await client.start_workflow( + CancelActivityDuringWorkflowCancellationWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + + async with new_worker(client, activities=[wait_cancel], task_queue=task_queue): + async with new_worker( + client, + CancelActivityDuringWorkflowCancellationWorkflow, + task_queue=task_queue, + workflow_runner=runner, + max_cached_workflows=0, + ): + + async def activity_started() -> bool: + return await handle.query( + CancelActivityDuringWorkflowCancellationWorkflow.activity_started + ) + + await assert_eq_eventually(True, activity_started) + + # Keep the workflow worker offline so the signal and cancellation are + # delivered in the same activation when it resumes. + await handle.signal( + CancelActivityDuringWorkflowCancellationWorkflow.cancel_activity + ) + await handle.cancel() + + async with new_worker( + client, + CancelActivityDuringWorkflowCancellationWorkflow, + task_queue=task_queue, + workflow_runner=runner, + ): + assert await handle.result() == "activity cancelled" + + assert not [ + event + async for event in handle.fetch_history_events() + if event.HasField("workflow_task_failed_event_attributes") + ] + assert any( + {"signal_workflow", "cancel_workflow"}.issubset( + {job.WhichOneof("variant") for job in activation.jobs} + ) + for activation, _ in runner._pairs + ) + + @workflow.defn class SimpleChildWorkflow: @workflow.run