diff --git a/CHANGELOG.md b/CHANGELOG.md index 17d90a43d..6135dd6fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -153,11 +153,18 @@ to include examples, links to docs, or any other relevant information. `temporalio.client`) instead of setting `payload_limits` on `DataConverter`. Config fields were renamed to `payloads_warn_size` and `memo_warn_size`, and the deprecated `PayloadSizeWarning` was removed. +- `WorkflowOutboundInterceptor.start_nexus_operation` no longer receives Temporal System Nexus + operations. Custom interceptors that need to observe or modify these operations must implement + `start_system_nexus_operation` instead. ### Fixed - Marked system Nexus envelope payloads so nested payloads can be detected and visited after the envelope is already stored as a payload. +- Fixed OpenTelemetry context propagation when a workflow uses signal-with-start. Trace context is + now added to the called workflow's headers instead of the System Nexus transport headers. + +### Security ## [1.30.0] - 2026-07-01 diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index ef6769fa1..fad21151e 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -167,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() diff --git a/temporalio/contrib/opentelemetry/_interceptor.py b/temporalio/contrib/opentelemetry/_interceptor.py index eb22f8be6..38321a2bc 100644 --- a/temporalio/contrib/opentelemetry/_interceptor.py +++ b/temporalio/contrib/opentelemetry/_interceptor.py @@ -830,6 +830,18 @@ async def start_nexus_operation( return await super().start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: temporalio.worker.StartSystemNexusOperationInput[Any, Any] + ) -> temporalio.workflow.NexusOperationHandle[Any]: + if hasattr(input.input, "headers"): + input.input.headers = input.input.headers or {} + self.root._completed_span( + f"StartNexusOperation:{input.service}/{input.operation_name}", + kind=opentelemetry.trace.SpanKind.CLIENT, + add_to_outbound=cast(_InputWithHeaders, input.input), + ) + return await super().start_system_nexus_operation(input) + def _carrier_to_nexus_headers( carrier: _CarrierDict, initial: Mapping[str, str] | None = None diff --git a/temporalio/contrib/opentelemetry/_otel_interceptor.py b/temporalio/contrib/opentelemetry/_otel_interceptor.py index c120fcd03..457ff982d 100644 --- a/temporalio/contrib/opentelemetry/_otel_interceptor.py +++ b/temporalio/contrib/opentelemetry/_otel_interceptor.py @@ -600,3 +600,14 @@ async def start_nexus_operation( ): input.headers = _context_to_nexus_headers(input.headers or {}) return await super().start_nexus_operation(input) + + async def start_system_nexus_operation( + self, input: temporalio.worker.StartSystemNexusOperationInput[Any, Any] + ) -> temporalio.workflow.NexusOperationHandle[Any]: + with self._workflow_maybe_span( + f"StartNexusOperation:{input.service}/{input.operation_name}", + kind=opentelemetry.trace.SpanKind.CLIENT, + ): + if hasattr(input.input, "headers"): + input.input.headers = _context_to_headers(input.input.headers or {}) + return await super().start_system_nexus_operation(input) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 730dd4258..60843aa1e 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -102,6 +102,7 @@ def __init__( self._user_converters = _SystemNexusUserConverters( user_payload_converter, user_failure_converter ) + self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap( _SystemNexusOuterPayloadConverter() ) diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index 092383b35..f757a41dc 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/__init__.py b/temporalio/nexus/system/workflow_service/_support/__init__.py index 530c33e80..8fe5fb12a 100644 --- a/temporalio/nexus/system/workflow_service/_support/__init__.py +++ b/temporalio/nexus/system/workflow_service/_support/__init__.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py index ef6769fa1..fad21151e 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py @@ -167,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index d03e7c3b4..653db044d 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations @@ -17,6 +17,8 @@ from ._support import ( duration_from_proto, duration_to_proto, + header_from_proto, + header_to_proto, memo_from_proto, memo_to_proto, payload_from_proto, @@ -133,6 +135,9 @@ def from_transfer_type( if proto.HasField("user_metadata") else None, namespace=proto.namespace, + headers=header_from_proto(proto.header) + if proto.HasField("header") + else None, ) @typing_extensions.override @@ -191,6 +196,8 @@ def to_transfer_type( ) ) message.namespace = value.namespace + if value.headers is not None: + message.header.CopyFrom(header_to_proto(value.headers)) return message @@ -226,6 +233,7 @@ class SignalWithStartWorkflowRequest: start_delay: datetime.timedelta | None = None user_metadata: UserMetadata | None = None namespace: str = dataclasses.field(default_factory=workflow_namespace) + headers: collections.abc.Mapping[str, typing.Any] | None = None class _UserMetadataTransferTypeConverter( diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index 67c9cc56b..26c5cc532 100644 --- a/temporalio/nexus/system/workflow_service/operations/__init__.py +++ b/temporalio/nexus/system/workflow_service/operations/__init__.py @@ -1,3 +1,3 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py index ac75f9fb9..77782589b 100644 --- a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py +++ b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/services.py b/temporalio/nexus/system/workflow_service/services.py index e2e901157..7b6c58f0e 100644 --- a/temporalio/nexus/system/workflow_service/services.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/worker/__init__.py b/temporalio/worker/__init__.py index 4f6efe68c..ada34ab71 100644 --- a/temporalio/worker/__init__.py +++ b/temporalio/worker/__init__.py @@ -21,6 +21,7 @@ StartChildWorkflowInput, StartLocalActivityInput, StartNexusOperationInput, + StartSystemNexusOperationInput, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, WorkflowOutboundInterceptor, @@ -100,6 +101,7 @@ "StartChildWorkflowInput", "StartLocalActivityInput", "StartNexusOperationInput", + "StartSystemNexusOperationInput", "WorkflowInterceptorClassInput", "ExecuteNexusOperationStartInput", "ExecuteNexusOperationCancelInput", diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py index 4acf3c5d1..b886a59b6 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -348,6 +348,29 @@ def operation_name(self) -> str: raise ValueError(f"Operation is not a Nexus operation: {self.operation}") +@dataclass +class StartSystemNexusOperationInput(Generic[InputT, OutputT]): + """Input for :py:meth:`WorkflowOutboundInterceptor.start_system_nexus_operation`.""" + + service: str + operation: nexusrpc.Operation[InputT, OutputT] | str | Callable[..., Any] + input: InputT + output_type: type[OutputT] | None = None + + @property + def operation_name(self) -> str: + """Get the name of the Nexus operation.""" + if isinstance(self.operation, nexusrpc.Operation): + return self.operation.name + elif isinstance(self.operation, str): + return self.operation + elif callable(self.operation): + _, op = temporalio.nexus._util.get_operation_factory(self.operation) + if isinstance(op, nexusrpc.Operation): + return op.name + raise ValueError(f"Operation is not a Nexus operation: {self.operation}") + + @dataclass class StartLocalActivityInput: """Input for :py:meth:`WorkflowOutboundInterceptor.start_local_activity`.""" @@ -481,6 +504,12 @@ async def start_nexus_operation( """Called for every :py:func:`temporalio.workflow.NexusClient.start_operation` call.""" return await self.next.start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[InputT, OutputT] + ) -> temporalio.workflow.NexusOperationHandle[OutputT]: + """Called for every Temporal System Nexus operation started by a workflow.""" + return await self.next.start_system_nexus_operation(input) + @dataclass class ExecuteNexusOperationStartInput: diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index eec5f903c..9e9a4f100 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -76,6 +76,7 @@ StartChildWorkflowInput, StartLocalActivityInput, StartNexusOperationInput, + StartSystemNexusOperationInput, WorkflowInboundInterceptor, WorkflowOutboundInterceptor, ) @@ -1724,7 +1725,18 @@ async def workflow_start_nexus_operation( headers: Mapping[str, str] | None, summary: str | None, ) -> temporalio.workflow.NexusOperationHandle[OutputT]: - # start_nexus_operation + if temporalio.nexus.system.is_system_endpoint(endpoint): + # System operations have no caller-configurable Nexus options. Do not + # expose the normal operation's scheduling, cancellation, headers, or + # summary arguments to System Nexus interceptors. + return await self._outbound.start_system_nexus_operation( + StartSystemNexusOperationInput( + service=service, + operation=operation, + input=input, + output_type=output_type, + ) + ) return await self._outbound.start_nexus_operation( StartNexusOperationInput( endpoint=endpoint, @@ -2171,6 +2183,27 @@ async def operation_handle_fn() -> OutputT: ) return handle + async def _outbound_start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, OutputT] + ) -> _NexusOperationHandle[OutputT]: + return await self._outbound_start_nexus_operation( + StartNexusOperationInput( + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + service=input.service, + operation=input.operation, + input=input.input, + output_type=input.output_type, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=( + temporalio.workflow.NexusOperationCancellationType.WAIT_COMPLETED + ), + headers=None, + summary=None, + ) + ) + #### Miscellaneous helpers #### # These are in alphabetical order. @@ -3157,6 +3190,11 @@ async def start_nexus_operation( ) -> _NexusOperationHandle[OutputT]: return await self._instance._outbound_start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, OutputT] + ) -> _NexusOperationHandle[OutputT]: + return await self._instance._outbound_start_system_nexus_operation(input) + def start_local_activity( self, input: StartLocalActivityInput ) -> temporalio.workflow.ActivityHandle[Any]: diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 1bab931ac..3253f4f2b 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -88,6 +88,34 @@ class TracingWorkflowActionActivity: fail_on_non_replay_before_complete: bool = False +@workflow.defn +class LegacySignalWithStartHeaderWorkflow: + def __init__(self) -> None: + self._signaled = False + + @workflow.run + async def run(self) -> bool: + await workflow.wait_condition(lambda: self._signaled) + return "_tracer-data" in workflow.info().headers + + @workflow.signal + def notify(self) -> None: + self._signaled = True + + +@workflow.defn +class LegacySignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + LegacySignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=LegacySignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + @dataclass class TracingWorkflowActionContinueAsNew: param: TracingWorkflowParam @@ -229,6 +257,40 @@ def update_validator(self) -> None: pass +# Cloud namespaces created by CI do not have the System Nexus dynamic config. +@pytest.mark.requires_local_server +async def test_legacy_otel_workflow_signal_with_start_propagates_trace_headers( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + provider = TracerProvider() + tracer = provider.get_tracer(__name__) + config = client.config() + config["interceptors"] = [TracingInterceptor(tracer)] + client = Client(**config) + + async with Worker( + client, + task_queue=f"signal-with-start-{uuid.uuid4()}", + workflows=[ + LegacySignalWithStartCallerWorkflow, + LegacySignalWithStartHeaderWorkflow, + ], + workflow_runner=UnsandboxedWorkflowRunner(), + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with tracer.start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + LegacySignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + + async def test_opentelemetry_tracing(client: Client, env: WorkflowEnvironment): # TODO(cretz): Fix if env.supports_time_skipping: diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 12d0c5972..8d922aeaa 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -122,6 +122,34 @@ async def run(self): return +@workflow.defn +class SignalWithStartHeaderWorkflow: + def __init__(self) -> None: + self._signaled = False + + @workflow.run + async def run(self) -> bool: + await workflow.wait_condition(lambda: self._signaled) + return "_tracer-data" in workflow.info().headers + + @workflow.signal + def notify(self) -> None: + self._signaled = True + + +@workflow.defn +class SignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + SignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=SignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: Any): # type: ignore[reportUnusedParameter] exporter = InMemorySpanExporter() provider = create_tracer_provider() @@ -169,6 +197,37 @@ async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: An ) +# Cloud namespaces created by CI do not have the System Nexus dynamic config. +@pytest.mark.requires_local_server +async def test_otel_workflow_signal_with_start_propagates_trace_headers( + client: Client, + env: WorkflowEnvironment, + reset_otel_tracer_provider: Any, # type: ignore[reportUnusedParameter] +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + provider = create_tracer_provider() + opentelemetry.trace.set_tracer_provider(provider) + config = client.config() + config["plugins"] = [OpenTelemetryPlugin()] + client = Client(**config) + + async with new_worker( + client, SignalWithStartCallerWorkflow, SignalWithStartHeaderWorkflow + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with get_tracer(__name__).start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + SignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=3), + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + + @workflow.defn class ComprehensiveWorkflow: def __init__(self) -> None: diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index c75fd64ff..d9f651f2f 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -36,6 +36,7 @@ from temporalio.worker import ( Interceptor, StartNexusOperationInput, + StartSystemNexusOperationInput, Worker, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, @@ -253,6 +254,12 @@ async def start_nexus_operation( interceptor_traces.append(("workflow.start_nexus_operation", input)) return await super().start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, Any] + ) -> workflow.NexusOperationHandle[Any]: + interceptor_traces.append(("workflow.start_system_nexus_operation", input)) + return await super().start_system_nexus_operation(input) + def _assert_stored_payloads_include( driver: InMemoryTestDriver, expected_payload_data: set[bytes] @@ -269,8 +276,8 @@ def _assert_stored_payloads_include( def _assert_start_nexus_operation_interceptor_trace() -> None: assert len(interceptor_traces) == 1 trace_name, trace_value = interceptor_traces.pop() - assert trace_name == "workflow.start_nexus_operation" - trace_input = cast(StartNexusOperationInput[Any, Any], trace_value) + assert trace_name == "workflow.start_system_nexus_operation" + trace_input = cast(StartSystemNexusOperationInput[Any, Any], trace_value) request = trace_input.input assert request.id == "system-nexus-workflow-id" assert request.signal == "test-signal"