From a92d9cc6c022e28872349b80d58cff8f4cb844ec Mon Sep 17 00:00:00 2001 From: Ayushi Ahjolia Date: Tue, 14 Jul 2026 14:57:26 -0700 Subject: [PATCH] feat(errors): typed per-operation error hierarchy --- .../examples-catalog.json | 11 ++ .../error_handling/catch_typed_step_error.py | 43 +++++ .../test_catch_typed_step_error.py | 27 +++ .../test_wait_for_callback_failure.py | 2 +- .../__init__.py | 10 ++ .../concurrency/models.py | 2 +- .../exceptions.py | 113 +++++++----- .../lambda_service.py | 36 +++- .../operation/child.py | 35 ++-- .../operation/invoke.py | 6 +- .../operation/step.py | 24 +-- .../operation/wait_for_condition.py | 22 ++- .../retries.py | 8 +- .../aws_durable_execution_sdk_python/state.py | 30 ++-- .../tests/concurrency_test.py | 8 +- .../tests/e2e/checkpoint_response_int_test.py | 2 +- .../tests/e2e/error_hierarchy_int_test.py | 170 ++++++++++++++++++ .../tests/exceptions_test.py | 102 +++++++---- .../tests/lambda_service_test.py | 63 ++++++- .../tests/operation/child_test.py | 28 +-- .../tests/operation/invoke_test.py | 8 +- .../tests/operation/step_test.py | 18 +- .../operation/wait_for_condition_test.py | 51 +++++- .../tests/state_test.py | 35 ++-- 24 files changed, 669 insertions(+), 185 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-examples/src/error_handling/catch_typed_step_error.py create mode 100644 packages/aws-durable-execution-sdk-python-examples/test/error_handling/test_catch_typed_step_error.py create mode 100644 packages/aws-durable-execution-sdk-python/tests/e2e/error_hierarchy_int_test.py diff --git a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json index 40090b5a..661bdd0d 100644 --- a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json +++ b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json @@ -748,6 +748,17 @@ "FILESYSTEM_MOUNT_PATH": "/mnt/s3" }, "path": "./src/filesystem_serdes/filesystem_serdes_preview.py" + }, + { + "name": "Catch Typed Step Error", + "description": "Catch a typed StepError raised by a failed step", + "handler": "catch_typed_step_error.handler", + "integration": true, + "durableConfig": { + "RetentionPeriodInDays": 7, + "ExecutionTimeout": 300 + }, + "path": "./src/error_handling/catch_typed_step_error.py" } ] } diff --git a/packages/aws-durable-execution-sdk-python-examples/src/error_handling/catch_typed_step_error.py b/packages/aws-durable-execution-sdk-python-examples/src/error_handling/catch_typed_step_error.py new file mode 100644 index 00000000..c1c98105 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/src/error_handling/catch_typed_step_error.py @@ -0,0 +1,43 @@ +"""Demonstrates catching a typed StepError raised by a failed step. + +A failure inside a durable operation surfaces as a typed +DurableOperationError subclass (StepError, InvokeError, ChildContextError, +WaitForConditionError) rather than a single catch-all error, so callers can +handle each operation's failure distinctly. +""" + +from typing import Any + +from aws_durable_execution_sdk_python import StepError +from aws_durable_execution_sdk_python.config import StepConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.retries import ( + RetryStrategyConfig, + create_retry_strategy, +) +from aws_durable_execution_sdk_python.types import StepContext + + +@durable_execution +def handler(_event: Any, context: DurableContext) -> dict[str, Any]: + """Run a step that fails and catch the typed StepError.""" + + def failing_step(_step_context: StepContext) -> None: + raise ValueError("payment declined") + + # A single attempt so the step fails without scheduling retries. + config = StepConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) + ) + + try: + context.step(failing_step, name="charge-card", config=config) + except StepError as error: + return { + "handled": True, + "error_type": error.error_type, + "message": error.message, + } + + return {"handled": False} diff --git a/packages/aws-durable-execution-sdk-python-examples/test/error_handling/test_catch_typed_step_error.py b/packages/aws-durable-execution-sdk-python-examples/test/error_handling/test_catch_typed_step_error.py new file mode 100644 index 00000000..101adb3f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/error_handling/test_catch_typed_step_error.py @@ -0,0 +1,27 @@ +"""Tests for catch_typed_step_error.""" + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus + +from src.error_handling import catch_typed_step_error +from test.conftest import deserialize_operation_payload + + +@pytest.mark.example +@pytest.mark.durable_execution( + handler=catch_typed_step_error.handler, + lambda_function_name="Catch Typed Step Error", +) +def test_catch_typed_step_error(durable_runner): + """A failed step surfaces as a catchable StepError with a typed discriminator.""" + with durable_runner: + result = durable_runner.run(input=None, timeout=10) + + assert result.status is InvocationStatus.SUCCEEDED + + result_data = deserialize_operation_payload(result.result) + assert result_data == { + "handled": True, + "error_type": "StepError", + "message": "payment declined", + } diff --git a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_failure.py b/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_failure.py index ac8d52fb..03c5741f 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_failure.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_failure.py @@ -23,5 +23,5 @@ def test_wait_for_callback_failure(durable_runner): assert isinstance(result.error, ErrorObject) assert result.error.to_dict() == { "ErrorMessage": "my callback error", - "ErrorType": "CallableRuntimeError", + "ErrorType": "ChildContextError", } diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py index ab62838c..adb1b38e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py @@ -18,9 +18,14 @@ # Most common exceptions - users need to handle these exceptions from aws_durable_execution_sdk_python.exceptions import ( + ChildContextError, DurableExecutionsError, + DurableOperationError, InvocationError, + InvokeError, + StepError, ValidationError, + WaitForConditionError, ) # Core decorator - used in every durable function @@ -33,12 +38,17 @@ __all__ = [ "BatchResult", + "ChildContextError", "DurableContext", "DurableExecutionsError", + "DurableOperationError", "InvocationError", + "InvokeError", "ParallelBranch", "StepContext", + "StepError", "ValidationError", + "WaitForConditionError", "WithRetryConfig", "__version__", "durable_execution", diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index a8137eb2..3391a38f 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -261,7 +261,7 @@ def throw_if_error(self) -> None: None, ) if first_error: - raise first_error.to_callable_runtime_error() + raise first_error.to_durable_operation_error() def get_results(self) -> list[R]: return [ diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py index a3fe3196..f7fab026 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py @@ -6,7 +6,6 @@ from __future__ import annotations import time -from dataclasses import dataclass from enum import Enum from typing import TYPE_CHECKING, Self, TypedDict @@ -30,6 +29,11 @@ } ) +# Registry mapping a DurableOperationError subclass name to its class, used to +# reconstruct the correct typed error from a checkpointed ErrorObject on replay. +# Populated automatically via DurableOperationError.__init_subclass__. +_DURABLE_OPERATION_ERROR_REGISTRY: dict[str, type[DurableOperationError]] = {} + if TYPE_CHECKING: import datetime @@ -266,25 +270,79 @@ class InvalidStateError(DurableExecutionsError): """Raised when an operation is attempted on an object in an invalid state.""" -class UserlandError(DurableExecutionsError): - """Failure in user-land - i.e code passed into durable executions from the caller.""" +class DurableOperationError(DurableExecutionsError): + """Base class for typed, per-operation failures. + Wraps a failure that escaped a Durable Function operation (step, invoke, + child context, wait_for_condition). The original exception is preserved as + ``__cause__`` on the first run; on replay the error is reconstructed from + its checkpointed ``ErrorObject`` via :meth:`from_error_fields`. -class CallableRuntimeError(UserlandError): - """This error wraps any failure from inside the callable code that you pass to a Durable Function operation.""" + Attributes: + message: Human-readable failure message. + error_type: Discriminator identifying the concrete subclass. Defaults + to the class name and is serialized as the checkpoint ``ErrorType``. + data: Optional serialized error payload, preserved across operation + boundaries. + stack_trace: Optional stack trace lines captured from the origin. + """ def __init__( self, - message: str | None, + message: str | None = None, + error_type: str | None = None, + data: str | None = None, + stack_trace: list[str] | None = None, + ) -> None: + super().__init__(message) + self.message: str | None = message + self.error_type: str = error_type or type(self).__name__ + self.data: str | None = data + self.stack_trace: list[str] | None = stack_trace + + def __init_subclass__(cls, **kwargs: object) -> None: + super().__init_subclass__(**kwargs) + _DURABLE_OPERATION_ERROR_REGISTRY[cls.__name__] = cls + + @classmethod + def from_error_fields( + cls, error_type: str | None, + message: str | None, data: str | None, stack_trace: list[str] | None, - ) -> None: - super().__init__(message) - self.message = message - self.error_type = error_type - self.data = data - self.stack_trace = stack_trace + ) -> DurableOperationError: + """Rebuild the correct subclass from serialized checkpoint error fields. + + Looks the discriminator up in the reconstruction registry, falling back + to the base :class:`DurableOperationError` when ``error_type`` is unknown + (e.g. a downstream error surfaced by an async invoke/callback checkpoint). + """ + target_cls: type[DurableOperationError] = _DURABLE_OPERATION_ERROR_REGISTRY.get( + error_type or "", DurableOperationError + ) + return target_cls( + message=message, + error_type=error_type, + data=data, + stack_trace=stack_trace, + ) + + +class StepError(DurableOperationError): + """Raised when a step operation fails.""" + + +class InvokeError(DurableOperationError): + """Raised when a durable invoke operation fails.""" + + +class ChildContextError(DurableOperationError): + """Raised when a child context (run_in_child_context, map, parallel) fails.""" + + +class WaitForConditionError(DurableOperationError): + """Raised when a wait_for_condition operation fails.""" class StepInterruptedError(InvocationError): @@ -397,37 +455,6 @@ def __init__(self, message: str, source_exception: Exception | None = None) -> N self.source_exception: Exception | None = source_exception -@dataclass(frozen=True) -class CallableRuntimeErrorSerializableDetails: - """Serializable error details.""" - - type: str - message: str - - @classmethod - def from_exception( - cls, exception: Exception - ) -> CallableRuntimeErrorSerializableDetails: - """Create an instance from an Exception, using its type and message. - - Args: - exception: An Exception instance - - Returns: - A CallableRuntimeErrorDetails instance with the exception's type name and message - """ - return cls(type=exception.__class__.__name__, message=str(exception)) - - def __str__(self) -> str: - """ - Return a string representation of the object. - - Returns: - A string in the format "type: message" - """ - return f"{self.type}: {self.message}" - - class SerDesError(DurableExecutionsError): """Raised when serialization fails.""" diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index 0843956b..4d5cb5c1 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -13,8 +13,8 @@ from aws_durable_execution_sdk_python.__about__ import __version__ from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, CheckpointError, + DurableOperationError, GetExecutionStateError, ) @@ -235,6 +235,15 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> ErrorObject: @classmethod def from_exception(cls, exception: Exception) -> ErrorObject: + # Preserve a DurableOperationError's discriminator/data/stack_trace so the + # typed info survives across operation boundaries. + if isinstance(exception, DurableOperationError): + return cls( + message=exception.message, + type=exception.error_type, + data=cls._resolve_error_data(exception), + stack_trace=exception.stack_trace, + ) return cls( message=str(exception), type=type(exception).__name__, @@ -242,6 +251,25 @@ def from_exception(cls, exception: Exception) -> ErrorObject: stack_trace=None, ) + @staticmethod + def _resolve_error_data(error: DurableOperationError) -> str | None: + """Surface the first non-None ``data`` along the cause chain. + + Ensures error data is not lost when a DurableOperationError is re-wrapped + across nested operation boundaries (e.g. a step failure escaping through + multiple run_in_child_context calls). + """ + if error.data is not None: + return error.data + node: BaseException | None = error.__cause__ + for _ in range(10): + if node is None: + break + if isinstance(node, DurableOperationError) and node.data is not None: + return node.data + node = node.__cause__ + return None + @classmethod def from_message(cls, message: str) -> ErrorObject: return cls( @@ -263,10 +291,10 @@ def to_dict(self) -> MutableMapping[str, Any]: result["StackTrace"] = self.stack_trace return result - def to_callable_runtime_error(self) -> CallableRuntimeError: - return CallableRuntimeError( - message=self.message, + def to_durable_operation_error(self) -> DurableOperationError: + return DurableOperationError.from_error_fields( error_type=self.type, + message=self.message, data=self.data, stack_trace=self.stack_trace, ) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py index 2789301d..54cb5a94 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py @@ -7,6 +7,7 @@ from aws_durable_execution_sdk_python.config import ChildConfig from aws_durable_execution_sdk_python.exceptions import ( + ChildContextError, InvocationError, SuspendExecution, ) @@ -78,7 +79,7 @@ def check_result_status(self) -> CheckResult[T]: CheckResult indicating the next action to take Raises: - CallableRuntimeError: For FAILED operations + ChildContextError: For FAILED operations """ checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( self.operation_identifier.operation_id @@ -114,7 +115,7 @@ def check_result_status(self) -> CheckResult[T]: # Terminal failure if checkpointed_result.is_failed(): - checkpointed_result.raise_callable_error() + checkpointed_result.raise_operation_error(ChildContextError) # Create START checkpoint if not exists if not checkpointed_result.is_existent() and not self.is_virtual: @@ -145,7 +146,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: Raises: SuspendExecution: Re-raised without checkpointing InvocationError: Re-raised after checkpointing FAIL - CallableRuntimeError: Raised for other exceptions after checkpointing FAIL + ChildContextError: Raised for other exceptions after checkpointing FAIL """ logger.debug( "▶️ Executing child context for id: %s, name: %s", @@ -239,7 +240,24 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: # Don't checkpoint SuspendExecution - let it bubble up raise except Exception as e: - error_object = ErrorObject.from_exception(e) + # InvocationError and its derivatives bubble up (with their own type, + # not wrapped) so the top-level handler can make the backend retry. + if isinstance(e, InvocationError): + if not self.is_virtual: + self.state.create_checkpoint( + operation_update=OperationUpdate.create_context_fail( + identifier=self.operation_identifier, + error=ErrorObject.from_exception(e), + sub_type=self.sub_type, + ) + ) + raise + + # Wrap the failure in a ChildContextError (original kept as __cause__) + # so the checkpoint records the "ChildContextError" discriminator. + child_error: ChildContextError = ChildContextError(message=str(e)) + child_error.__cause__ = e + error_object = ErrorObject.from_exception(child_error) # Virtual deliberately does not write checkpoints, but exception still propagates below if not self.is_virtual: fail_operation: OperationUpdate = OperationUpdate.create_context_fail( @@ -252,14 +270,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: # This guarantees the error is durable and child operations won't be re-executed on replay. self.state.create_checkpoint(operation_update=fail_operation) - # InvocationError and its derivatives can be retried. - # When we encounter an invocation error (in all of its forms), we - # bubble that error upwards (with the checkpoint in place for - # non-virtual) such that we reach the execution handler at the - # very top, which will then make the backend retry. - if isinstance(e, InvocationError): - raise - raise error_object.to_callable_runtime_error() from e + raise child_error from e def child_handler( diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py index 6d181341..891fce01 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py @@ -5,7 +5,7 @@ import logging from typing import TYPE_CHECKING, TypeVar -from aws_durable_execution_sdk_python.exceptions import ExecutionError +from aws_durable_execution_sdk_python.exceptions import ExecutionError, InvokeError from aws_durable_execution_sdk_python.lambda_service import ( ChainedInvokeOptions, OperationUpdate, @@ -81,7 +81,7 @@ def check_result_status(self) -> CheckResult[R]: CheckResult indicating the next action to take Raises: - CallableRuntimeError: For FAILED, TIMED_OUT, or STOPPED operations + InvokeError: For FAILED, TIMED_OUT, or STOPPED operations SuspendExecution: For STARTED operations waiting for completion """ checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( @@ -107,7 +107,7 @@ def check_result_status(self) -> CheckResult[R]: or checkpointed_result.is_timed_out() or checkpointed_result.is_stopped() ): - checkpointed_result.raise_callable_error() + checkpointed_result.raise_operation_error(InvokeError) # Still running - ready to suspend if checkpointed_result.is_started(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py index 02f62d9a..743bf8bc 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py @@ -12,6 +12,7 @@ from aws_durable_execution_sdk_python.exceptions import ( ExecutionError, InvalidStateError, + StepError, StepInterruptedError, ) from aws_durable_execution_sdk_python.lambda_service import ( @@ -86,7 +87,7 @@ def check_result_status(self) -> CheckResult[T]: CheckResult indicating the next action to take Raises: - CallableRuntimeError: For FAILED operations + StepError: For FAILED operations StepInterruptedError: For interrupted AT_MOST_ONCE operations SuspendExecution: For PENDING operations waiting for retry """ @@ -115,7 +116,7 @@ def check_result_status(self) -> CheckResult[T]: # Terminal failure if checkpointed_result.is_failed(): # Have to throw the exact same error on replay as the checkpointed failure - checkpointed_result.raise_callable_error() + checkpointed_result.raise_operation_error(StepError) # Pending retry if checkpointed_result.is_pending(): @@ -143,7 +144,7 @@ def check_result_status(self) -> CheckResult[T]: # Step was previously interrupted in a prior invocation - handle retry msg: str = f"Step operation_id={self.operation_identifier.operation_id} name={self.operation_identifier.name} was previously interrupted" self.retry_handler(StepInterruptedError(msg), checkpointed_result) - checkpointed_result.raise_callable_error() + checkpointed_result.raise_operation_error(StepError) # Ready to execute if STARTED + AT_LEAST_ONCE if ( @@ -286,10 +287,14 @@ def retry_handler( Raises: SuspendExecution: If retry is scheduled - StepInterruptedError: If the error is a StepInterruptedError - CallableRuntimeError: If retry is exhausted or error is not retryable + StepError: If retry is exhausted or the error is not retryable + (including an exhausted StepInterruptedError) """ - error_object = ErrorObject.from_exception(error) + # Wrap the failure in a StepError (original kept as __cause__) so the + # checkpoint records the "StepError" discriminator for replay. + step_error: StepError = StepError(message=str(error)) + step_error.__cause__ = error + error_object = ErrorObject.from_exception(step_error) retry_strategy = self.config.retry_strategy or RetryPresets.default() @@ -360,7 +365,6 @@ def retry_handler( # This guarantees the error is durable and the step won't be retried on replay. self.state.create_checkpoint(operation_update=fail_operation) - if isinstance(error, StepInterruptedError): - raise error - - raise error_object.to_callable_runtime_error() + # A StepInterruptedError that has exhausted retries surfaces to the caller + # as a StepError, like any other exhausted step failure. + raise step_error from error diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py index 8771de36..c645ed25 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py @@ -7,6 +7,8 @@ from aws_durable_execution_sdk_python.exceptions import ( ExecutionError, + InvocationError, + WaitForConditionError, ) from aws_durable_execution_sdk_python.lambda_service import ( ErrorObject, @@ -84,7 +86,7 @@ def check_result_status(self) -> CheckResult[T]: CheckResult indicating the next action to take Raises: - CallableRuntimeError: For FAILED operations + WaitForConditionError: For FAILED operations SuspendExecution: For PENDING operations waiting for retry """ checkpointed_result = self.state.get_checkpoint_result( @@ -110,7 +112,7 @@ def check_result_status(self) -> CheckResult[T]: # Terminal failure if checkpointed_result.is_failed(): - checkpointed_result.raise_callable_error() + checkpointed_result.raise_operation_error(WaitForConditionError) # Pending retry if checkpointed_result.is_pending(): @@ -273,15 +275,27 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.name, ) + # Let control-flow errors (InvocationError/ExecutionError) propagate + # by type; wrap only genuine check failures as WaitForConditionError. + error_to_raise: Exception = e + if not isinstance(e, (InvocationError, ExecutionError)): + wait_error: WaitForConditionError = WaitForConditionError( + message=str(e) + ) + wait_error.__cause__ = e + error_to_raise = wait_error + fail_operation = OperationUpdate.create_wait_for_condition_fail( identifier=self.operation_identifier, - error=ErrorObject.from_exception(e), + error=ErrorObject.from_exception(error_to_raise), ) # Checkpoint FAIL operation with blocking (is_sync=True, default). # Must ensure the failure state is persisted before raising the exception. # This guarantees the error is durable and the condition won't be re-evaluated on replay. self.state.create_checkpoint(operation_update=fail_operation) - raise + if error_to_raise is e: + raise + raise error_to_raise from e msg: str = ( "wait_for_condition should never reach this point" # pragma: no cover diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/retries.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/retries.py index 6d0723d9..6b32704b 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/retries.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/retries.py @@ -296,8 +296,8 @@ class WithRetryConfig(Generic[T]): (RetryStrategyConfig defaults) is used. wrap_with_run_in_child_context: Whether to wrap the retry loop in a child context for isolation. Default True. When True, final - failure is rethrown as CallableRuntimeError with the original - exception on `cause`. When False, the original error is + failure is rethrown as ChildContextError with the original + exception on `__cause__`. When False, the original error is rethrown unchanged. child_context_config: Optional ChildConfig forwarded to run_in_child_context when wrapping is enabled. Ignored when @@ -344,8 +344,8 @@ def with_retry( exhausted or the retry strategy returns should_retry=False. When wrap_with_run_in_child_context is True (default), ChildOperationExecutor.process wraps non-InvocationError / - SuspendExecution exceptions as CallableRuntimeError with the - original error in cause. + SuspendExecution exceptions as ChildContextError with the + original error in __cause__. When wrap_with_run_in_child_context is False, the original exception propagates unchanged. SuspendExecution: Re-raised immediately (SDK control flow). diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py index 03b13030..2580b6a1 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py @@ -15,8 +15,8 @@ from aws_durable_execution_sdk_python.exceptions import ( BackgroundThreadError, - CallableRuntimeError, DurableExecutionsError, + DurableOperationError, GetExecutionStateError, OrphanedChildException, SuspendExecution, @@ -222,20 +222,30 @@ def is_replay_children(self) -> bool: return False return op.context_details.replay_children if op.context_details else False - def raise_callable_error(self, msg: str | None = None) -> None: + def raise_operation_error( + self, + operation_error_cls: type[DurableOperationError], + msg: str | None = None, + ) -> None: + """Reconstruct and raise the typed operation error for a FAILED checkpoint. + + The concrete error type is dictated by the operation being replayed and + supplied by the calling executor (e.g. ``StepError`` for a step). This + ensures async operations whose checkpoint carries a downstream error type + (invoke/callback) still surface as the correct per-operation error. + """ if self.error is None: - err_msg = ( + err_msg: str = ( msg or "Unknown error. No ErrorObject exists on the Checkpoint Operation." ) - raise CallableRuntimeError( - message=err_msg, - error_type=None, - data=None, - stack_trace=None, - ) + raise operation_error_cls(message=err_msg) - raise self.error.to_callable_runtime_error() + raise operation_error_cls( + message=self.error.message, + data=self.error.data, + stack_trace=self.error.stack_trace, + ) def get_next_attempt_timestamp(self) -> datetime.datetime | None: if self.operation and self.operation.step_details: diff --git a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py index 887e91c3..0fac2ae5 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -36,7 +36,7 @@ ExecutionContext, ) from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, + ChildContextError, InvalidStateError, SuspendExecution, TimedSuspendExecution, @@ -221,12 +221,12 @@ def test_batch_result_throw_if_error(): result = BatchResult(items, CompletionReason.ALL_COMPLETED) result.throw_if_error() # Should not raise - # Has error - error = ErrorObject("test message", "TestError", None, None) + # Has error - reconstructs the typed operation error from its discriminator + error = ErrorObject("test message", "ChildContextError", None, None) items = [BatchItem(0, BatchItemStatus.FAILED, error=error)] result = BatchResult(items, CompletionReason.ALL_COMPLETED) - with pytest.raises(CallableRuntimeError): + with pytest.raises(ChildContextError, match="test message"): result.throw_if_error() diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py index 66c9ad50..e5c74fd7 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py @@ -643,7 +643,7 @@ def test_end_to_end_child_context_error_handling(): """Test end-to-end child context error handling. Verifies that child context that raises exception creates FAIL checkpoint - and error is wrapped as CallableRuntimeError. + and error is wrapped as a typed DurableOperationError (e.g. StepError). """ def child_function_that_fails(ctx: DurableContext) -> str: diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/error_hierarchy_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/error_hierarchy_int_test.py new file mode 100644 index 00000000..7fe3e4d3 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/error_hierarchy_int_test.py @@ -0,0 +1,170 @@ +"""Integration tests for the typed per-operation error hierarchy. + +These exercise the full handler flow end-to-end (rather than mocking the +executor seams) to cover the two directions unit tests mock away: + +- serialization: a failing operation checkpoints a FAIL carrying the typed + discriminator, and the invocation result surfaces the typed error. +- reconstruction: a pre-seeded FAILED checkpoint resurfaces as the typed error + on replay without re-executing the operation. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import Mock, patch + +from aws_durable_execution_sdk_python.config import StepConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python.retries import ( + RetryStrategyConfig, + create_retry_strategy, +) +from tests.e2e.checkpoint_response_int_test import ( + create_mock_checkpoint_with_operations, +) +from tests.test_helpers import operation_id_sequence + +if TYPE_CHECKING: + from aws_durable_execution_sdk_python.types import StepContext + + +def _lambda_context() -> Mock: + lambda_context = Mock() + lambda_context.aws_request_id = "test-request-id" + lambda_context.client_context = None + lambda_context.identity = None + lambda_context._epoch_deadline_time_in_ms = 0 # noqa: SLF001 + lambda_context.invoked_function_arn = "test-arn" + lambda_context.tenant_id = None + return lambda_context + + +def test_step_failure_surfaces_typed_step_error(): + """A failing step checkpoints FAIL with the StepError discriminator and the + execution result carries the typed error (serialization path).""" + + def failing_step(_step_context: StepContext) -> str: + msg = "step boom" + raise ValueError(msg) + + @durable_execution + def my_handler(event, context: DurableContext) -> str: + # Single attempt so the step fails without scheduling retries. + config = StepConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) + ) + return context.step(failing_step, name="charge", config=config) + + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + + mock_checkpoint, checkpoint_calls = create_mock_checkpoint_with_operations() + mock_client.checkpoint = mock_checkpoint + + event = { + "DurableExecutionArn": "test-arn/execution-1", + "CheckpointToken": "test-token", + "InitialExecutionState": { + "Operations": [ + { + "Id": "execution-1", + "Type": "EXECUTION", + "Status": "STARTED", + "ExecutionDetails": {"InputPayload": "{}"}, + } + ], + "NextMarker": "", + }, + "LocalRunner": True, + } + + result = my_handler(event, _lambda_context()) + + # Execution failed and the result carries the typed discriminator. + assert result["Status"] == InvocationStatus.FAILED.value + assert result["Error"]["ErrorType"] == "StepError" + assert result["Error"]["ErrorMessage"] == "step boom" + + # The FAIL checkpoint records the discriminator, not the raw error type. + all_operations = [op for batch in checkpoint_calls for op in batch] + fail_updates = [ + op + for op in all_operations + if hasattr(op, "action") and op.action.value == "FAIL" + ] + assert len(fail_updates) == 1 + assert fail_updates[0].error is not None + assert fail_updates[0].error.type == "StepError" + + +def test_failed_step_reconstructs_step_error_on_replay(): + """A pre-seeded FAILED step checkpoint resurfaces as a typed StepError on + replay without re-executing the step (reconstruction path).""" + executed = {"count": 0} + + def maybe_step(_step_context: StepContext) -> str: + executed["count"] += 1 + return "should-not-run" + + @durable_execution + def my_handler(event, context: DurableContext) -> str: + return context.step(maybe_step, name="charge") + + # The id the first top-level step will look up on replay. + step_id = next(operation_id_sequence()) + + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + + mock_checkpoint, _checkpoint_calls = create_mock_checkpoint_with_operations() + mock_client.checkpoint = mock_checkpoint + + event = { + "DurableExecutionArn": "test-arn/execution-1", + "CheckpointToken": "test-token", + "InitialExecutionState": { + "Operations": [ + { + "Id": "execution-1", + "Type": "EXECUTION", + "Status": "STARTED", + "ExecutionDetails": {"InputPayload": "{}"}, + }, + { + "Id": step_id, + "Type": "STEP", + "Status": "FAILED", + "SubType": "Step", + "ParentId": "execution-1", + "StepDetails": { + "Error": { + "ErrorType": "StepError", + "ErrorMessage": "prior boom", + } + }, + }, + ], + "NextMarker": "", + }, + "LocalRunner": True, + } + + result = my_handler(event, _lambda_context()) + + # Reconstructed as a StepError with the checkpointed message; the step + # body is never re-executed. + assert result["Status"] == InvocationStatus.FAILED.value + assert result["Error"]["ErrorType"] == "StepError" + assert result["Error"]["ErrorMessage"] == "prior boom" + assert executed["count"] == 0 diff --git a/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py b/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py index a18d534e..55223706 100644 --- a/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py @@ -7,25 +7,28 @@ from botocore.exceptions import ClientError # type: ignore[import-untyped] from aws_durable_execution_sdk_python.exceptions import ( + _DURABLE_OPERATION_ERROR_REGISTRY, BotoClientError, - CallableRuntimeError, - CallableRuntimeErrorSerializableDetails, + ChildContextError, CheckpointError, CheckpointErrorCategory, DurableApiErrorCategory, DurableExecutionsError, + DurableOperationError, ExecutionError, GetExecutionStateError, InvocationError, + InvokeError, OrderedLockError, OrphanedChildException, + StepError, StepInterruptedError, SuspendExecution, TerminationReason, TimedSuspendExecution, UnrecoverableError, - UserlandError, ValidationError, + WaitForConditionError, ) @@ -207,33 +210,79 @@ def test_validation_error(): assert isinstance(error, DurableExecutionsError) -def test_userland_error(): - """Test UserlandError exception.""" - error = UserlandError("userland error") - assert str(error) == "userland error" +def test_durable_operation_error_defaults_error_type_to_class_name(): + """Base DurableOperationError defaults error_type to its class name.""" + error = DurableOperationError("boom") + assert str(error) == "boom" + assert error.message == "boom" + assert error.error_type == "DurableOperationError" + assert error.data is None + assert error.stack_trace is None assert isinstance(error, DurableExecutionsError) -def test_callable_runtime_error(): - """Test CallableRuntimeError exception.""" - error = CallableRuntimeError( +def test_durable_operation_error_explicit_fields(): + """DurableOperationError preserves explicitly provided fields.""" + error = DurableOperationError( "runtime error", "ValueError", "error data", ["line1", "line2"] ) - assert str(error) == "runtime error" assert error.message == "runtime error" assert error.error_type == "ValueError" assert error.data == "error data" - assert isinstance(error, UserlandError) + assert error.stack_trace == ["line1", "line2"] -def test_callable_runtime_error_with_none_values(): - """Test CallableRuntimeError with None values.""" - error = CallableRuntimeError(None, None, None, None) +def test_durable_operation_error_with_none_message(): + """DurableOperationError tolerates a None message.""" + error = DurableOperationError(None) assert error.message is None - assert error.error_type is None + assert error.error_type == "DurableOperationError" assert error.data is None +@pytest.mark.parametrize( + "error_cls", + [StepError, InvokeError, ChildContextError, WaitForConditionError], +) +def test_operation_error_subclasses(error_cls): + """Each per-operation subclass derives from DurableOperationError and self-types.""" + error = error_cls("failed") + assert isinstance(error, DurableOperationError) + assert isinstance(error, DurableExecutionsError) + assert error.error_type == error_cls.__name__ + assert error.message == "failed" + + +def test_operation_error_registry_contains_all_subclasses(): + """The reconstruction registry is keyed by class name for every subclass.""" + assert _DURABLE_OPERATION_ERROR_REGISTRY["StepError"] is StepError + assert _DURABLE_OPERATION_ERROR_REGISTRY["InvokeError"] is InvokeError + assert _DURABLE_OPERATION_ERROR_REGISTRY["ChildContextError"] is ChildContextError + assert ( + _DURABLE_OPERATION_ERROR_REGISTRY["WaitForConditionError"] + is WaitForConditionError + ) + + +def test_from_error_fields_by_name(): + """from_error_fields rebuilds the correct subclass from its discriminator.""" + error = DurableOperationError.from_error_fields( + "StepError", "step failed", "data", ["frame"] + ) + assert isinstance(error, StepError) + assert error.error_type == "StepError" + assert error.message == "step failed" + assert error.data == "data" + assert error.stack_trace == ["frame"] + + +def test_from_error_fields_unknown_falls_back_to_base(): + """Unknown discriminators fall back to the base DurableOperationError.""" + error = DurableOperationError.from_error_fields("ValueError", "boom", None, None) + assert type(error) is DurableOperationError + assert error.error_type == "ValueError" + + def test_step_interrupted_error(): """Test StepInterruptedError exception.""" error = StepInterruptedError("step interrupted", "step_123") @@ -267,27 +316,6 @@ def test_ordered_lock_error_with_source(): assert error.source_exception is source -def test_callable_runtime_error_serializable_details_from_exception(): - """Test CallableRuntimeErrorSerializableDetails.from_exception.""" - exception = ValueError("test error") - details = CallableRuntimeErrorSerializableDetails.from_exception(exception) - assert details.type == "ValueError" - assert details.message == "test error" - - -def test_callable_runtime_error_serializable_details_str(): - """Test CallableRuntimeErrorSerializableDetails.__str__.""" - details = CallableRuntimeErrorSerializableDetails("TypeError", "type error message") - assert str(details) == "TypeError: type error message" - - -def test_callable_runtime_error_serializable_details_frozen(): - """Test CallableRuntimeErrorSerializableDetails is frozen.""" - details = CallableRuntimeErrorSerializableDetails("Error", "message") - with pytest.raises(AttributeError): - details.type = "NewError" - - def test_timed_suspend_execution(): """Test TimedSuspendExecution exception.""" scheduled_time = 1234567890.0 diff --git a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py index 7c1885f0..9f22f2f1 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -8,9 +8,11 @@ from aws_durable_execution_sdk_python.__about__ import __version__ from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, CheckpointError, + ChildContextError, + DurableOperationError, GetExecutionStateError, + StepError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -228,20 +230,63 @@ def test_error_object_to_dict_all_none(): assert result == {} -def test_error_object_to_callable_runtime_error(): - """Test ErrorObject.to_callable_runtime_error method.""" +def test_error_object_to_durable_operation_error_reconstructs_subclass(): + """ErrorObject.to_durable_operation_error rebuilds the typed subclass by name.""" + error = ErrorObject( + message="Test error", + type="StepError", + data="test_data", + stack_trace=["line1"], + ) + operation_error = error.to_durable_operation_error() + assert isinstance(operation_error, StepError) + assert operation_error.message == "Test error" + assert operation_error.error_type == "StepError" + assert operation_error.data == "test_data" + assert operation_error.stack_trace == ["line1"] + + +def test_error_object_to_durable_operation_error_unknown_type_falls_back(): + """An unknown discriminator falls back to the base DurableOperationError.""" error = ErrorObject( message="Test error", type="TestError", data="test_data", stack_trace=["line1"], ) - runtime_error = error.to_callable_runtime_error() - assert isinstance(runtime_error, CallableRuntimeError) - assert runtime_error.message == "Test error" - assert runtime_error.error_type == "TestError" - assert runtime_error.data == "test_data" - assert runtime_error.stack_trace == ["line1"] + operation_error = error.to_durable_operation_error() + assert type(operation_error) is DurableOperationError + assert operation_error.error_type == "TestError" + assert operation_error.message == "Test error" + + +def test_error_object_from_exception_preserves_operation_error_fields(): + """from_exception keeps a DurableOperationError's typed discriminator/data.""" + cause = ValueError("boom") + step_error = StepError(message="step failed", data="payload") + step_error.__cause__ = cause + error_object = ErrorObject.from_exception(step_error) + assert error_object.type == "StepError" + assert error_object.message == "step failed" + assert error_object.data == "payload" + + +def test_error_object_from_exception_walks_cause_for_data(): + """from_exception surfaces nested data across operation boundaries.""" + inner = StepError(message="inner", data="inner-data") + outer = ChildContextError(message="outer") + outer.__cause__ = inner + error_object = ErrorObject.from_exception(outer) + assert error_object.type == "ChildContextError" + assert error_object.data == "inner-data" + + +def test_error_object_from_exception_plain_exception(): + """A non-operation exception still serializes its Python class name.""" + error_object = ErrorObject.from_exception(ValueError("nope")) + assert error_object.type == "ValueError" + assert error_object.message == "nope" + assert error_object.data is None def test_step_details_from_dict(): diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py index 1f8617ca..6af0b895 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py @@ -10,7 +10,7 @@ from aws_durable_execution_sdk_python.config import ChildConfig from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, + ChildContextError, InvocationError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier @@ -173,13 +173,13 @@ def test_child_handler_already_failed(): mock_result = Mock() mock_result.is_succeeded.return_value = False mock_result.is_failed.return_value = True - mock_result.raise_callable_error.side_effect = CallableRuntimeError( - "Previous failure", "TestError", None, None + mock_result.raise_operation_error.side_effect = ChildContextError( + "Previous failure" ) mock_state.get_checkpoint_result.return_value = mock_result mock_callable = Mock() - with pytest.raises(CallableRuntimeError, match="Previous failure"): + with pytest.raises(ChildContextError, match="Previous failure"): child_handler( mock_callable, mock_state, @@ -287,7 +287,7 @@ def test_child_handler_callable_exception( mock_callable = Mock(side_effect=ValueError("Test error")) mock_state.wrap_user_function.return_value = mock_callable - with pytest.raises(CallableRuntimeError): + with pytest.raises(ChildContextError): child_handler( mock_callable, mock_state, @@ -322,14 +322,18 @@ def test_child_handler_callable_exception( assert fail_operation.operation_type is OperationType.CONTEXT assert fail_operation.sub_type is expected_sub_type assert fail_operation.action is OperationAction.FAIL - assert fail_operation.error == ErrorObject.from_exception(ValueError("Test error")) + # The escaping ValueError is wrapped in ChildContextError before checkpointing, + # so the persisted ErrorType is the operation discriminator. + assert fail_operation.error == ErrorObject( + message="Test error", type="ChildContextError", data=None, stack_trace=None + ) def test_child_handler_error_wrapped(): - """Test child_handler wraps regular errors as CallableRuntimeError. + """Test child_handler wraps regular errors as ChildContextError. Verifies: - - Regular exceptions are wrapped as CallableRuntimeError + - Regular exceptions are wrapped as ChildContextError - FAIL checkpoint is created """ mock_state = Mock(spec=ExecutionState) @@ -344,7 +348,7 @@ def test_child_handler_error_wrapped(): mock_callable = Mock(side_effect=test_error) mock_state.wrap_user_function.return_value = mock_callable - with pytest.raises(CallableRuntimeError): + with pytest.raises(ChildContextError): child_handler( mock_callable, mock_state, @@ -881,7 +885,7 @@ def test_child_handler_is_virtual_with_exception(): A virtual branch emits no lifecycle entries in the execution history, so a failure inside the branch does not get its own FAIL checkpoint. The exception still propagates (wrapped as - CallableRuntimeError for non-InvocationError exceptions) so the + ChildContextError for non-InvocationError exceptions) so the concurrency executor records the failure in the BatchResult and its completion-tolerance logic still applies. """ @@ -899,7 +903,7 @@ def test_child_handler_is_virtual_with_exception(): config = ChildConfig(is_virtual=True) - with pytest.raises(CallableRuntimeError): + with pytest.raises(ChildContextError): child_handler( mock_callable, mock_state, @@ -931,7 +935,7 @@ def test_child_handler_not_is_virtual_with_exception(): config = ChildConfig(is_virtual=False) - with pytest.raises(CallableRuntimeError): + with pytest.raises(ChildContextError): child_handler( mock_callable, mock_state, diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index dda39b21..8b8fe25c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -9,8 +9,8 @@ from aws_durable_execution_sdk_python.config import InvokeConfig from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, ExecutionError, + InvokeError, SuspendExecution, TimedSuspendExecution, ) @@ -147,7 +147,7 @@ def test_invoke_handler_already_terminated(kind: OperationStatus): mock_result = CheckpointedResult.create_from_operation(operation) mock_state.get_checkpoint_result.return_value = mock_result - with pytest.raises(CallableRuntimeError): + with pytest.raises(InvokeError): invoke_handler( function_name="test_function", payload="test_input", @@ -176,7 +176,7 @@ def test_invoke_handler_already_timed_out(): mock_result = CheckpointedResult.create_from_operation(operation) mock_state.get_checkpoint_result.return_value = mock_result - with pytest.raises(CallableRuntimeError): + with pytest.raises(InvokeError): invoke_handler( function_name="test_function", payload="test_input", @@ -914,7 +914,7 @@ def test_invoke_immediate_response_immediate_failure(status: OperationStatus): mock_state.get_checkpoint_result.side_effect = [not_found, failed] # Verify error is raised without suspend - with pytest.raises(CallableRuntimeError): + with pytest.raises(InvokeError): invoke_handler( function_name="test_function", payload="test_input", diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py index e7195512..e62ccb1c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py @@ -12,8 +12,8 @@ StepSemantics, ) from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, ExecutionError, + StepError, StepInterruptedError, SuspendExecution, ) @@ -125,7 +125,7 @@ def test_step_handler_already_failed(): mock_callable = Mock() mock_logger = Mock(spec=Logger) - with pytest.raises(CallableRuntimeError): + with pytest.raises(StepError): step_handler( mock_callable, mock_state, @@ -371,7 +371,7 @@ def test_step_handler_retry_exhausted(): mock_logger = Mock(spec=Logger) mock_logger.with_log_info.return_value = mock_logger - with pytest.raises(CallableRuntimeError): + with pytest.raises(StepError): step_handler( mock_callable, mock_state, @@ -397,10 +397,13 @@ def test_step_handler_retry_exhausted(): assert fail_operation.operation_type is OperationType.STEP assert fail_operation.sub_type is OperationSubType.STEP assert fail_operation.action is OperationAction.FAIL + # The checkpointed error records the typed discriminator, not the raw error. + assert fail_operation.error is not None + assert fail_operation.error.type == "StepError" def test_step_handler_retry_interrupted_error(): - """Test step_handler with StepInterruptedError in retry.""" + """A StepInterruptedError that exhausts retries surfaces as a StepError.""" mock_state = Mock(spec=ExecutionState) mock_result = CheckpointedResult.create_not_found() mock_state.get_checkpoint_result.return_value = mock_result @@ -416,7 +419,7 @@ def test_step_handler_retry_interrupted_error(): mock_logger = Mock(spec=Logger) mock_logger.with_log_info.return_value = mock_logger - with pytest.raises(StepInterruptedError, match="Step interrupted"): + with pytest.raises(StepError, match="Step interrupted") as exc_info: step_handler( mock_callable, mock_state, @@ -425,6 +428,9 @@ def test_step_handler_retry_interrupted_error(): mock_logger, ) + # The original StepInterruptedError is preserved as the cause. + assert exc_info.value.__cause__ is interrupted_error + def test_step_handler_retry_with_existing_attempts(): """Test step_handler retry logic with existing attempt count.""" @@ -791,7 +797,7 @@ def test_step_immediate_response_immediate_failure(): ) # Verify operation raises error after executing step function - with pytest.raises(CallableRuntimeError, match="Step execution error"): + with pytest.raises(StepError, match="Step execution error"): step_handler( mock_callable, mock_state, diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py index da35a6fe..56583630 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py @@ -8,9 +8,10 @@ from aws_durable_execution_sdk_python.config import Duration from aws_durable_execution_sdk_python.exceptions import ( - CallableRuntimeError, + ExecutionError, InvocationError, SuspendExecution, + WaitForConditionError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -232,7 +233,7 @@ def check_func(state, context): wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), ) - with pytest.raises(CallableRuntimeError): + with pytest.raises(WaitForConditionError): wait_for_condition_handler( state=mock_state, operation_identifier=op_id, @@ -392,7 +393,49 @@ def check_func(state, context): wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), ) - with pytest.raises(ValueError, match="Test error"): + # A check-function failure surfaces as the typed WaitForConditionError, + # preserving the original exception as __cause__. + with pytest.raises(WaitForConditionError, match="Test error") as exc_info: + wait_for_condition_handler( + state=mock_state, + operation_identifier=op_id, + check=check_func, + config=config, + context_logger=mock_logger, + ) + + assert isinstance(exc_info.value.__cause__, ValueError) + assert mock_state.create_checkpoint.call_count == 2 # START and FAIL + + +@pytest.mark.parametrize("error_cls", [InvocationError, ExecutionError]) +def test_wait_for_condition_control_flow_error_not_wrapped(error_cls): + """Control-flow errors from the check function propagate by type, unwrapped.""" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "arn:aws:test" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + op_id = OperationIdentifier( + "op1", OperationSubType.WAIT_FOR_CONDITION, None, "test_wait" + ) + + def check_func(state, context): + raise error_cls("control-flow failure") + + mock_state.wrap_user_function.return_value = check_func + + config = WaitForConditionConfig( + initial_state=5, + wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), + ) + + # Not wrapped in WaitForConditionError - propagates with its own type. + with pytest.raises(error_cls, match="control-flow failure"): wait_for_condition_handler( state=mock_state, operation_identifier=op_id, @@ -1190,7 +1233,7 @@ def check_func(state, context): ) # Verify error raised without executing check function - with pytest.raises(CallableRuntimeError): + with pytest.raises(WaitForConditionError): wait_for_condition_handler( state=mock_state, operation_identifier=op_id, diff --git a/packages/aws-durable-execution-sdk-python/tests/state_test.py b/packages/aws-durable-execution-sdk-python/tests/state_test.py index f2ea641c..2ca76963 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -15,10 +15,10 @@ from aws_durable_execution_sdk_python.exceptions import ( BackgroundThreadError, - CallableRuntimeError, DurableApiErrorCategory, GetExecutionStateError, OrphanedChildException, + StepError, TimedSuspendExecution, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier @@ -368,32 +368,35 @@ def test_checkpointed_result_is_started(): assert result_no_op.is_started() is False -def test_checkpointed_result_raise_callable_error(): - """Test CheckpointedResult.raise_callable_error method.""" - error = Mock(spec=ErrorObject) - error.to_callable_runtime_error.return_value = RuntimeError("Test error") +def test_checkpointed_result_raise_operation_error(): + """raise_operation_error wraps the checkpoint error in the given operation type.""" + error = ErrorObject( + message="Test error", type="ValueError", data="d", stack_trace=["l"] + ) result = CheckpointedResult(error=error) - with pytest.raises(RuntimeError, match="Test error"): - result.raise_callable_error() + with pytest.raises(StepError, match="Test error") as exc_info: + result.raise_operation_error(StepError) - error.to_callable_runtime_error.assert_called_once() + assert exc_info.value.error_type == "StepError" + assert exc_info.value.data == "d" + assert exc_info.value.stack_trace == ["l"] -def test_checkpointed_result_raise_callable_error_no_error(): - """Test CheckpointedResult.raise_callable_error with no error.""" +def test_checkpointed_result_raise_operation_error_no_error(): + """raise_operation_error with no error raises the operation type with a default message.""" result = CheckpointedResult() - with pytest.raises(CallableRuntimeError, match="Unknown error"): - result.raise_callable_error() + with pytest.raises(StepError, match="Unknown error"): + result.raise_operation_error(StepError) -def test_checkpointed_result_raise_callable_error_no_error_with_message(): - """Test CheckpointedResult.raise_callable_error with no error and custom message.""" +def test_checkpointed_result_raise_operation_error_no_error_with_message(): + """raise_operation_error with no error uses the provided custom message.""" result = CheckpointedResult() - with pytest.raises(CallableRuntimeError, match="Custom error message"): - result.raise_callable_error("Custom error message") + with pytest.raises(StepError, match="Custom error message"): + result.raise_operation_error(StepError, "Custom error message") def test_checkpointed_result_immutable():