Fix duplicate activity cancellation commands

Author: tconley1428

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

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)

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