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..622b3140a 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,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