diff --git a/scripts/gen_payload_visitor.py b/scripts/gen_payload_visitor.py index 53d61611b..7fab85dcd 100644 --- a/scripts/gen_payload_visitor.py +++ b/scripts/gen_payload_visitor.py @@ -124,12 +124,20 @@ def generate(self, roots: list[Descriptor]) -> str: The generated code defines async visitor functions for each reachable protobuf message type starting from WorkflowActivation, including support - for repeated fields and map entries, and a convenience entrypoint - function `visit`. + for repeated fields and map entries. Payload-free roots get no-op methods + so the `visit` entrypoint recognizes them as supported. """ - for r in roots: - self.walk(r) + for root in roots: + if not self.walk(root): + self.methods.append( + f"""\ + async def _visit_{name_for(root)}( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass +""" + ) header = """ from __future__ import annotations diff --git a/temporalio/bridge/_visitor.py b/temporalio/bridge/_visitor.py index 5882a1de2..d8fa05edf 100644 --- a/temporalio/bridge/_visitor.py +++ b/temporalio/bridge/_visitor.py @@ -645,3 +645,8 @@ async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutio await self._visit_temporal_api_common_v1_Header(fs, o.header) if o.HasField("user_metadata"): await self._visit_temporal_api_sdk_v1_UserMetadata(fs, o.user_metadata) + + async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutionResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 730dd4258..1f8f7c6d2 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -15,6 +15,7 @@ import temporalio.api.common.v1 import temporalio.common import temporalio.converter +import temporalio.exceptions from temporalio.bridge._visitor_functions import VisitorFunctions from temporalio.converter import BinaryProtoPayloadConverter, CompositePayloadConverter from temporalio.converter._payload_converter import ( @@ -154,7 +155,14 @@ async def maybe_visit_payload( payload_visitor = PayloadVisitor(skip_search_attributes=skip_search_attributes) checkpoint = visitor_functions.checkpoint() - await payload_visitor.visit(visitor_functions, value) + try: + await payload_visitor.visit(visitor_functions, value) + except ValueError as err: + if not str(err).startswith("Unknown root message type: "): + raise + raise temporalio.exceptions.ApplicationError( + f"Unknown Temporal system payload: {value.DESCRIPTOR.full_name}" + ) from err if checkpoint is not None: await visitor_functions.drain_since(checkpoint) return payload_converter.to_payload(value) diff --git a/tests/__init__.py b/tests/__init__.py index af97849fe..eff2c8adb 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1 +1 @@ -DEV_SERVER_DOWNLOAD_VERSION = "v1.7.4-standalone-nexus-operations" +DEV_SERVER_DOWNLOAD_VERSION = "v1.8.3-server-1.32.0-162.0" diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 983e9ea03..66bc2290b 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -44,9 +44,6 @@ # See https://github.com/temporalio/sdk-python/issues/1704. pytestmark = pytest.mark.requires_local_server -# Query response links require a newer server than the shared test environment. -_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION = "v1.8.3-server-1.32.0-162.0" - @dataclass class Input: @@ -854,14 +851,7 @@ async def run(self, input: Input) -> bool: return await client.execute_operation(TestService.query_op, input.value) -async def test_temporal_operation_query_workflow() -> None: - async with await WorkflowEnvironment.start_local( - dev_server_download_version=_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION - ) as env: - await _assert_temporal_operation_query_workflow(env.client, env) - - -async def _assert_temporal_operation_query_workflow( +async def test_temporal_operation_query_workflow( client: Client, env: WorkflowEnvironment ) -> None: task_queue = str(uuid.uuid4()) @@ -900,15 +890,17 @@ async def _assert_temporal_operation_query_workflow( target_history = await target_handle.fetch_history() assert not any(event.links for event in target_history.events) - assert target_handle.result_run_id is not None - assert Link( - workflow=Link.Workflow( - namespace=client.namespace, - workflow_id=target_workflow_id, - run_id=target_handle.result_run_id, - reason="Query processed", - ) - ) in list(completed_event.links) + # The Java time-skipping test server does not return Nexus operation links. + if not env.supports_time_skipping: + assert target_handle.result_run_id is not None + assert Link( + workflow=Link.Workflow( + namespace=client.namespace, + workflow_id=target_workflow_id, + run_id=target_handle.result_run_id, + reason="Query processed", + ) + ) in list(completed_event.links) finally: await target_handle.cancel() @@ -1314,14 +1306,8 @@ async def test_temporal_operation_start_activity_raises_error( id=str(uuid.uuid4()), ) - operation_err = err.value.__cause__ - assert isinstance(operation_err, temporalio.exceptions.ApplicationError) - assert operation_err.type == "OperationError" - assert "nexus operation completed unsuccessfully" in str(operation_err) - - application_err = operation_err.__cause__ + application_err = err.value.__cause__ assert isinstance(application_err, temporalio.exceptions.ApplicationError) - assert application_err.type == "test-activity-error-type" assert "test-activity-error-message" in str(application_err) assert application_err.__cause__ is None diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 76916aaa7..362288f34 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -9,6 +9,7 @@ import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2 import temporalio.bridge.worker import temporalio.converter +import temporalio.exceptions import temporalio.nexus.system as nexus_system from temporalio.api.common.v1.message_pb2 import ( Payload, @@ -266,6 +267,75 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: assert visitor.system_envelope_count == 1 +async def test_system_nexus_envelope_without_payloads_is_visited(): + class SystemNexusVisitor(Visitor): + def __init__(self) -> None: + self.system_envelope_count = 0 + + async def visit_system_nexus_envelope(self, payload: Payload) -> None: + _ = payload + self.system_envelope_count += 1 + + response = workflowservice_pb2.SignalWithStartWorkflowExecutionResponse( + run_id="test-run-id" + ) + data_converter = temporalio.converter.default() + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + system_payload = payload_converter.to_payload(response) + assert system_payload is not None + completion = WorkflowActivationCompletion( + run_id="3", + successful=Success( + commands=[ + WorkflowCommand( + update_response=UpdateResponse(completed=system_payload), + ) + ] + ), + ) + visitor = SystemNexusVisitor() + + await PayloadVisitor().visit(visitor, completion) + + completed = completion.successful.commands[0].update_response.completed + assert payload_converter.from_payload(completed) == response + assert visitor.system_envelope_count == 1 + + +async def test_unknown_system_nexus_payload_raises_application_error(): + data_converter = temporalio.converter.default() + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + system_payload = payload_converter.to_payload( + workflowservice_pb2.StartWorkflowExecutionResponse(run_id="test-run-id") + ) + assert system_payload is not None + completion = WorkflowActivationCompletion( + run_id="3", + successful=Success( + commands=[ + WorkflowCommand( + update_response=UpdateResponse(completed=system_payload), + ) + ] + ), + ) + + with pytest.raises(temporalio.exceptions.ApplicationError) as err: + await PayloadVisitor().visit(Visitor(), completion) + + assert ( + err.value.message == "Unknown Temporal system payload: " + "temporal.api.workflowservice.v1.StartWorkflowExecutionResponse" + ) + assert not err.value.non_retryable + + async def test_concurrent_throughput(): """Demonstrate that concurrent visitation is faster than serialized for I/O-bound codecs.""" N_CMDS = 10