From 50d5b6e5986aca1428526ac93f8e4fe1ec20c289 Mon Sep 17 00:00:00 2001 From: Roey Berman Date: Fri, 14 Aug 2026 15:49:09 -0700 Subject: [PATCH 1/3] Report non-retryable PayloadValidationError as BAD_REQUEST A payload codec or payload converter can signal that a Nexus operation's input is invalid by raising a non-retryable ApplicationError of type PayloadValidationError while decoding the input. Such an error is now translated into a BAD_REQUEST HandlerError retaining the original error as its cause, instead of the INTERNAL handler error any other application error produces. Application errors of any other type, and retryable PayloadValidationError errors, keep their existing behavior. --- CHANGELOG.md | 5 + temporalio/worker/_nexus.py | 38 ++++ tests/nexus/test_workflow_caller_errors.py | 212 +++++++++++++++++++++ 3 files changed, 255 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f5e22ce98..4db413890 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,11 @@ to include examples, links to docs, or any other relevant information. ``PydanticJSONPlainPayloadConverter``) from a nullary subclass used as the ``DataConverter.payload_converter_class``; ``None`` makes the cache unbounded and zero disables caching. +- A data converter can now report that it understood a Nexus operation's input + but considers it invalid by raising a non-retryable `ApplicationError` of type + `PayloadValidationError` while decoding it. Such a failure is reported to the + caller as a `BAD_REQUEST` Nexus handler error instead of a handler-side + `INTERNAL` error. Any other decode failure keeps its existing treatment. ### Deprecated diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index 131e50862..9f21811a1 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -14,6 +14,7 @@ Any, NoReturn, ParamSpec, + TypeGuard, TypeVar, cast, ) @@ -51,6 +52,15 @@ _TEMPORAL_FAILURE_PROTO_TYPE = "temporal.api.failure.v1.Failure" +_PAYLOAD_VALIDATION_ERROR_TYPE = "PayloadValidationError" +""":py:attr:`temporalio.exceptions.ApplicationError.type` a data converter uses to +say that it understood a Nexus operation's input but considers it invalid. + +When non-retryable, such an error is reported as a +:py:attr:`nexusrpc.HandlerErrorType.BAD_REQUEST` handler error rather than as a +handler-side :py:attr:`nexusrpc.HandlerErrorType.INTERNAL` error. +""" + @dataclass class _RunningNexusTask: @@ -505,6 +515,17 @@ async def visit_payloads(self, payloads: PayloadSequence) -> None: payloads.extend(new_payloads) +def _is_payload_validation_error(err: BaseException) -> TypeGuard[ApplicationError]: + """Whether err is a non-retryable :py:class:`ApplicationError` whose type is + exactly :py:data:`_PAYLOAD_VALIDATION_ERROR_TYPE`. + """ + return ( + isinstance(err, ApplicationError) + and err.non_retryable + and err.type == _PAYLOAD_VALIDATION_ERROR_TYPE + ) + + @dataclass class _DummyPayloadSerializer: data_converter: temporalio.converter.DataConverter @@ -542,6 +563,14 @@ async def deserialize( _PayloadTransformVisitor(dc._decode_payload_sequence), payload ) except Exception as err: + if _is_payload_validation_error(err): + # The data converter decoded the input and rejected it, so this + # is the caller's fault rather than a handler-side error. + raise nexusrpc.HandlerError( + "Invalid operation input", + type=nexusrpc.HandlerErrorType.BAD_REQUEST, + retryable_override=False, + ) from err raise nexusrpc.HandlerError( "Payload codec failed to decode Nexus operation input", type=nexusrpc.HandlerErrorType.INTERNAL, @@ -554,6 +583,15 @@ async def deserialize( ) return input except Exception as err: + if _is_payload_validation_error(err): + # The data converter converted the input and rejected it, so + # distinguish it from input that will never decode into the + # expected type. + raise nexusrpc.HandlerError( + "Invalid operation input", + type=nexusrpc.HandlerErrorType.BAD_REQUEST, + retryable_override=False, + ) from err raise nexusrpc.HandlerError( "Payload converter failed to decode Nexus operation input", type=nexusrpc.HandlerErrorType.BAD_REQUEST, diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py index 0f1b6a789..a02c19cc9 100644 --- a/tests/nexus/test_workflow_caller_errors.py +++ b/tests/nexus/test_workflow_caller_errors.py @@ -39,6 +39,7 @@ from temporalio.service import RPCError, RPCStatusCode from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker +from temporalio.worker._nexus import _DummyPayloadSerializer from tests.helpers import LogCapturer, assert_eq_eventually from tests.helpers.nexus import make_nexus_endpoint_name @@ -841,3 +842,214 @@ async def test_nexus_operation_fails_without_retry_on_converter_failure( ) else: pytest.fail("Expected WorkflowFailureError") + + +_PAYLOAD_VALIDATION_FAILURE_MESSAGE = "Nexus operation input failed validation" + + +class RaiseOnDecodeCodec(PayloadCodec): + def __init__(self, error: Exception) -> None: + self.error = error + + async def encode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return list(payloads) + + async def decode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + raise self.error + + +def _converter_class_raising(error: Exception) -> type[DefaultPayloadConverter]: + class RaiseOnFromPayloadsConverter(DefaultPayloadConverter): + def from_payloads( + self, + payloads: Sequence[temporalio.api.common.v1.Payload], + type_hints: list[type] | None = None, + ) -> list[Any]: + raise error + + return RaiseOnFromPayloadsConverter + + +async def _deserialize_input(data_converter: DataConverter) -> Any: + [payload] = DataConverter.default.payload_converter.to_payloads(["input"]) + serializer = _DummyPayloadSerializer(data_converter=data_converter, payload=payload) + return await serializer.deserialize(nexusrpc.Content(headers={}, data=b"")) + + +async def _deserialize_input_with_codec_error(error: Exception) -> Any: + return await _deserialize_input( + DataConverter(payload_codec=RaiseOnDecodeCodec(error)) + ) + + +async def _deserialize_input_with_converter_error(error: Exception) -> Any: + return await _deserialize_input( + DataConverter(payload_converter_class=_converter_class_raising(error)) + ) + + +async def test_codec_input_payload_validation_failure_is_bad_request(): + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + non_retryable=True, + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_codec_error(validation_error) + assert err.value.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not err.value.retryable + assert err.value.message == "Invalid operation input" + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is validation_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_codec_input_decode_failure_of_other_error_type_is_internal(): + decode_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="SomeOtherErrorType", + non_retryable=True, + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_codec_error(decode_error) + assert err.value.type == nexusrpc.HandlerErrorType.INTERNAL + assert "Payload codec failed to decode Nexus operation input" in str(err.value) + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is decode_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_retryable_codec_input_payload_validation_failure_is_internal(): + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_codec_error(validation_error) + assert err.value.type == nexusrpc.HandlerErrorType.INTERNAL + assert err.value.retryable + assert "Payload codec failed to decode Nexus operation input" in str(err.value) + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is validation_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_converter_input_payload_validation_failure_is_bad_request(): + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + non_retryable=True, + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_converter_error(validation_error) + assert err.value.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not err.value.retryable + assert err.value.message == "Invalid operation input" + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is validation_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_converter_input_failure_of_other_error_type_keeps_generic_message(): + convert_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="SomeOtherErrorType", + non_retryable=True, + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_converter_error(convert_error) + assert err.value.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not err.value.retryable + assert ( + err.value.message == "Payload converter failed to decode Nexus operation input" + ) + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is convert_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_retryable_converter_input_payload_validation_failure_keeps_generic_message(): + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + ) + with pytest.raises(nexusrpc.HandlerError) as err: + await _deserialize_input_with_converter_error(validation_error) + assert err.value.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not err.value.retryable + assert ( + err.value.message == "Payload converter failed to decode Nexus operation input" + ) + cause = err.value.__cause__ + assert isinstance(cause, ApplicationError) + assert cause is validation_error + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_nexus_operation_fails_without_retry_on_input_validation_failure( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with time-skipping server") + + task_queue = str(uuid.uuid4()) + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + non_retryable=True, + ) + handler_client = Client( + client.service_client, + namespace=client.namespace, + data_converter=DataConverter( + payload_codec=RaiseOnDecodeCodec(validation_error) + ), + ) + input = ErrorTestInput( + service_name="DataConverterTestService", + operation_name="succeed", + task_queue=task_queue, + id=str(uuid.uuid4()), + ) + async with ( + Worker( + client, + workflows=[CallerWorkflow], + task_queue=task_queue, + ), + Worker( + handler_client, + nexus_service_handlers=[DataConverterTestService()], + nexus_task_executor=concurrent.futures.ThreadPoolExecutor(max_workers=1), + task_queue=task_queue, + ), + ): + await env.create_nexus_endpoint( + make_nexus_endpoint_name(input.task_queue), input.task_queue + ) + with pytest.raises(WorkflowFailureError) as err: + await client.execute_workflow( + CallerWorkflow.run, + input, + id=str(uuid.uuid4()), + task_queue=task_queue, + ) + assert isinstance(err.value.__cause__, NexusOperationError) + handler_error = err.value.__cause__.__cause__ + assert isinstance(handler_error, nexusrpc.HandlerError) + assert handler_error.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not handler_error.retryable + assert handler_error.message == "Invalid operation input" + cause = handler_error.__cause__ + assert isinstance(cause, ApplicationError) + assert cause.type == "PayloadValidationError" + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE From 1ab6ee94f5fa931ee542fe0dd5549a0d5b573f6d Mon Sep 17 00:00:00 2001 From: Roey Berman Date: Fri, 14 Aug 2026 15:59:56 -0700 Subject: [PATCH 2/3] Add a functional test for converter input payload validation failures --- tests/nexus/test_workflow_caller_errors.py | 63 +++++++++++++++++++++- 1 file changed, 62 insertions(+), 1 deletion(-) diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py index a02c19cc9..622b3140a 100644 --- a/tests/nexus/test_workflow_caller_errors.py +++ b/tests/nexus/test_workflow_caller_errors.py @@ -995,7 +995,7 @@ async def test_retryable_converter_input_payload_validation_failure_keeps_generi assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE -async def test_nexus_operation_fails_without_retry_on_input_validation_failure( +async def test_nexus_operation_fails_without_retry_on_codec_input_validation_failure( client: Client, env: WorkflowEnvironment ): if env.supports_time_skipping: @@ -1053,3 +1053,64 @@ async def test_nexus_operation_fails_without_retry_on_input_validation_failure( assert isinstance(cause, ApplicationError) assert cause.type == "PayloadValidationError" assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE + + +async def test_nexus_operation_fails_without_retry_on_converter_input_validation_failure( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with time-skipping server") + + task_queue = str(uuid.uuid4()) + validation_error = ApplicationError( + _PAYLOAD_VALIDATION_FAILURE_MESSAGE, + type="PayloadValidationError", + non_retryable=True, + ) + handler_client = Client( + client.service_client, + namespace=client.namespace, + data_converter=DataConverter( + payload_converter_class=_converter_class_raising(validation_error) + ), + ) + input = ErrorTestInput( + service_name="DataConverterTestService", + operation_name="succeed", + task_queue=task_queue, + id=str(uuid.uuid4()), + ) + async with ( + Worker( + client, + workflows=[CallerWorkflow], + task_queue=task_queue, + ), + Worker( + handler_client, + nexus_service_handlers=[DataConverterTestService()], + nexus_task_executor=concurrent.futures.ThreadPoolExecutor(max_workers=1), + task_queue=task_queue, + ), + ): + await env.create_nexus_endpoint( + make_nexus_endpoint_name(input.task_queue), input.task_queue + ) + with pytest.raises(WorkflowFailureError) as err: + await client.execute_workflow( + CallerWorkflow.run, + input, + id=str(uuid.uuid4()), + task_queue=task_queue, + ) + assert isinstance(err.value.__cause__, NexusOperationError) + handler_error = err.value.__cause__.__cause__ + assert isinstance(handler_error, nexusrpc.HandlerError) + assert handler_error.type == nexusrpc.HandlerErrorType.BAD_REQUEST + assert not handler_error.retryable + # A validation failure gets its own message, distinct from the generic decode failure. + assert handler_error.message == "Invalid operation input" + cause = handler_error.__cause__ + assert isinstance(cause, ApplicationError) + assert cause.type == "PayloadValidationError" + assert cause.message == _PAYLOAD_VALIDATION_FAILURE_MESSAGE From 4c6a95e423c272bb1796b97e9474102c1eb7fec0 Mon Sep 17 00:00:00 2001 From: Roey Berman Date: Mon, 17 Aug 2026 09:09:59 -0700 Subject: [PATCH 3/3] Clarify the changelog entry for converter payload validation failures --- CHANGELOG.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4db413890..b9d70e0cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,8 +48,12 @@ to include examples, links to docs, or any other relevant information. - A data converter can now report that it understood a Nexus operation's input but considers it invalid by raising a non-retryable `ApplicationError` of type `PayloadValidationError` while decoding it. Such a failure is reported to the - caller as a `BAD_REQUEST` Nexus handler error instead of a handler-side - `INTERNAL` error. Any other decode failure keeps its existing treatment. + caller as a `BAD_REQUEST` Nexus handler error with the message + `Invalid operation input`, retaining the original error as its cause. Raised + from a payload codec, that replaces a handler-side `INTERNAL` error; raised + from a payload converter, the type was already `BAD_REQUEST` and only the + message becomes specific to validation. Any other decode failure, and a + retryable `PayloadValidationError`, keep their existing treatment. ### Deprecated