From 44dd5cfdff75e57c2fb52cfddd80e6ffcba4edaf Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 3 Sep 2026 10:57:35 -0700 Subject: [PATCH 1/4] Fix duplicate activity cancellation commands --- CHANGELOG.md | 3 + temporalio/worker/_workflow_instance.py | 26 ++++--- tests/worker/test_workflow.py | 90 +++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 11 deletions(-) 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 From 88361f0c0cc4697fad5e669ac442de64b93530ea Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 9 Sep 2026 11:39:10 -0700 Subject: [PATCH 2/4] Fix System Nexus payload converter context --- temporalio/worker/_workflow_instance.py | 18 ++- tests/nexus/test_temporal_system_nexus.py | 137 ++++++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index f7cd66cfe..80d77f103 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2176,9 +2176,23 @@ async def operation_handle_fn() -> OutputT: ) if temporalio.nexus.system.is_system_endpoint(input.endpoint): + serialization_context = temporalio.nexus.system._get_serialization_context( + input.service, + input.operation_name, + input.input, + ) + user_payload_converter = self._workflow_context_payload_converter + user_failure_converter = self._workflow_context_failure_converter + if serialization_context is not None: + user_payload_converter = self._payload_converter_with_context( + serialization_context + ) + user_failure_converter = self._failure_converter_with_context( + serialization_context + ) payload_converter = temporalio.nexus.system._get_payload_converter( - self._workflow_context_payload_converter, - self._workflow_context_failure_converter, + user_payload_converter, + user_failure_converter, ) else: payload_converter = self._context_free_payload_converter diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 5fee9c5b3..c944ce8a0 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -27,7 +27,11 @@ ) from temporalio.client import Client from temporalio.converter import ( + CompositePayloadConverter, + DefaultPayloadConverter, + EncodingPayloadConverter, ExternalStorage, + JSONPlainPayloadConverter, PayloadCodec, SerializationContext, WithSerializationContext, @@ -238,6 +242,91 @@ async def decode( return list(payloads) +@dataclasses.dataclass +class ContextValue: + value: str + + +converter_contexts: list[SerializationContext | None] = [] + + +class ContextPayloadConverter(EncodingPayloadConverter, WithSerializationContext): + def __init__(self, context: SerializationContext | None = None) -> None: + self.context = context + + @property + def encoding(self) -> str: + return "test-context" + + def with_context(self, context: SerializationContext) -> ContextPayloadConverter: + return ContextPayloadConverter(context) + + def to_payload(self, value: Any) -> temporalio.api.common.v1.Payload | None: + if not isinstance(value, ContextValue): + return None + converter_contexts.append(self.context) + payload = JSONPlainPayloadConverter().to_payload(value) + assert payload is not None + payload.metadata["encoding"] = self.encoding.encode() + return payload + + def from_payload( + self, + payload: temporalio.api.common.v1.Payload, + type_hint: type | None = None, + ) -> Any: + return JSONPlainPayloadConverter().from_payload(payload, type_hint) + + +class ContextPayloadConverterSet(CompositePayloadConverter): + def __init__(self) -> None: + super().__init__( + ContextPayloadConverter(), + *DefaultPayloadConverter.default_encoding_payload_converters, + ) + + +class ContextPayloadCodec(PayloadCodec, WithSerializationContext): + def __init__( + self, + contexts: list[SerializationContext | None], + context: SerializationContext | None = None, + ) -> None: + self.contexts = contexts + self.context = context + + def with_context(self, context: SerializationContext) -> ContextPayloadCodec: + return ContextPayloadCodec(self.contexts, context) + + async def encode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + for payload in payloads: + if payload.metadata.get("encoding") == b"test-context": + self.contexts.append(self.context) + return list(payloads) + + async def decode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return list(payloads) + + +@workflow.defn +class ContextSignalWithStartWorkflowCaller: + @workflow.run + async def run(self, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + "test-workflow", + ContextValue("workflow-input"), + id="system-nexus-workflow-id", + task_queue=task_queue, + signal="test-signal", + signal_args=[ContextValue("signal-input")], + ) + return handle.id + + class TracingWorkflowInterceptor(Interceptor): def workflow_interceptor_class( self, input: WorkflowInterceptorClassInput @@ -785,3 +874,51 @@ def capture_get_payload_converter( and context.workflow_id == target_workflow_id for context in captured_contexts ) + + +@pytest.mark.requires_local_server +async def test_signal_with_start_uses_target_context_for_converter_and_codec( + env: WorkflowEnvironment, +) -> None: + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + converter_contexts.clear() + codec_contexts: list[SerializationContext | None] = [] + caller_config = env.client.config() + caller_config["data_converter"] = dataclasses.replace( + temporalio.converter.default(), + payload_converter_class=ContextPayloadConverterSet, + payload_codec=ContextPayloadCodec(codec_contexts), + ) + caller_client = Client(**caller_config) + caller_task_queue = str(uuid.uuid4()) + target_workflow_id = "system-nexus-workflow-id" + + async with Worker( + caller_client, + task_queue=caller_task_queue, + workflows=[ContextSignalWithStartWorkflowCaller], + workflow_runner=UnsandboxedWorkflowRunner(), + ): + result = await caller_client.execute_workflow( + ContextSignalWithStartWorkflowCaller.run, + caller_task_queue, + id=str(uuid.uuid4()), + task_queue=caller_task_queue, + execution_timeout=timedelta(seconds=5), + ) + + assert result == target_workflow_id + assert len(converter_contexts) >= 2 + assert all( + isinstance(context, WorkflowSerializationContext) + and context.workflow_id == target_workflow_id + for context in converter_contexts + ), converter_contexts + assert len(codec_contexts) >= 2 + assert all( + isinstance(context, WorkflowSerializationContext) + and context.workflow_id == target_workflow_id + for context in codec_contexts + ), codec_contexts From 4940e62b1112715e6b2d4ff7430b5773a556382e Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 9 Sep 2026 12:00:24 -0700 Subject: [PATCH 3/4] Add System Nexus context changelog entry --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 14b7eef6e..b256e3e03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,9 @@ to include examples, links to docs, or any other relevant information. - **Experimental**: External storage metrics now report the wall-clock time storage was in flight. Previously each batch's duration was summed, over-reporting the time whenever storage operations ran concurrently. +- System Nexus Signal-with-Start workflow operations now give custom payload + converters the target workflow's serialization context when encoding their + inner request payloads. - Cancelling an activity from a signal while the workflow itself is cancelled no longer causes a nondeterminism error from duplicate activity-cancellation commands. From 41328c29c176ffd4c611bdc03179d226f315edae Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 9 Sep 2026 15:13:47 -0700 Subject: [PATCH 4/4] Keep System Nexus test context in converter --- tests/nexus/test_temporal_system_nexus.py | 28 +++++++++++++---------- 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index c944ce8a0..49200aeab 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -33,6 +33,7 @@ ExternalStorage, JSONPlainPayloadConverter, PayloadCodec, + PayloadConverter, SerializationContext, WithSerializationContext, WorkflowSerializationContext, @@ -247,11 +248,13 @@ class ContextValue: value: str -converter_contexts: list[SerializationContext | None] = [] - - class ContextPayloadConverter(EncodingPayloadConverter, WithSerializationContext): - def __init__(self, context: SerializationContext | None = None) -> None: + def __init__( + self, + contexts: list[SerializationContext | None], + context: SerializationContext | None = None, + ) -> None: + self.contexts = contexts self.context = context @property @@ -259,12 +262,12 @@ def encoding(self) -> str: return "test-context" def with_context(self, context: SerializationContext) -> ContextPayloadConverter: - return ContextPayloadConverter(context) + return ContextPayloadConverter(self.contexts, context) def to_payload(self, value: Any) -> temporalio.api.common.v1.Payload | None: if not isinstance(value, ContextValue): return None - converter_contexts.append(self.context) + self.contexts.append(self.context) payload = JSONPlainPayloadConverter().to_payload(value) assert payload is not None payload.metadata["encoding"] = self.encoding.encode() @@ -280,8 +283,9 @@ def from_payload( class ContextPayloadConverterSet(CompositePayloadConverter): def __init__(self) -> None: + self.contexts: list[SerializationContext | None] = [] super().__init__( - ContextPayloadConverter(), + ContextPayloadConverter(self.contexts), *DefaultPayloadConverter.default_encoding_payload_converters, ) @@ -883,12 +887,12 @@ async def test_signal_with_start_uses_target_context_for_converter_and_codec( if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") - converter_contexts.clear() codec_contexts: list[SerializationContext | None] = [] + payload_converter = ContextPayloadConverterSet() caller_config = env.client.config() caller_config["data_converter"] = dataclasses.replace( temporalio.converter.default(), - payload_converter_class=ContextPayloadConverterSet, + payload_converter_class=cast(type[PayloadConverter], lambda: payload_converter), payload_codec=ContextPayloadCodec(codec_contexts), ) caller_client = Client(**caller_config) @@ -910,12 +914,12 @@ async def test_signal_with_start_uses_target_context_for_converter_and_codec( ) assert result == target_workflow_id - assert len(converter_contexts) >= 2 + assert len(payload_converter.contexts) >= 2 assert all( isinstance(context, WorkflowSerializationContext) and context.workflow_id == target_workflow_id - for context in converter_contexts - ), converter_contexts + for context in payload_converter.contexts + ), payload_converter.contexts assert len(codec_contexts) >= 2 assert all( isinstance(context, WorkflowSerializationContext)