Report non-retryable PayloadValidationError as BAD_REQUEST

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
 

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,

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,275 @@ 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_codec_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
+
+
+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