diff --git a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_submitter_failure.py b/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_submitter_failure.py index d192dd78b..af50bac0b 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_submitter_failure.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/wait_for_callback/test_wait_for_callback_submitter_failure.py @@ -24,13 +24,10 @@ def test_fail_after_exhausting_retries_when_submitter_always_fails(durable_runne # Execution should fail after retries are exhausted assert result.status is InvocationStatus.FAILED - # A failed submitter step surfaces as CallbackSubmitterError (the submitter's - # StepError is translated at the wait_for_callback call site). The type is - # identical on first run and replay, so the execution-level record pins it. + # User code sees CallbackSubmitterError, and the execution records the + # error it wraps: the submitter's own exception. The type is identical on + # first run and replay, so the execution-level record pins it. error = result.error assert error is not None - assert ( - error.type - == "aws_durable_execution_sdk_python.exceptions.CallbackSubmitterError" - ) + assert error.type == "Exception" assert "Simulated submitter failure" in error.message diff --git a/packages/aws-durable-execution-sdk-python-extras/README.md b/packages/aws-durable-execution-sdk-python-extras/README.md index bfc9e463e..0d2804fae 100644 --- a/packages/aws-durable-execution-sdk-python-extras/README.md +++ b/packages/aws-durable-execution-sdk-python-extras/README.md @@ -225,8 +225,8 @@ pipeline child context A failed `vm.invoke` raises its MicroVM error inside the handler. The handler can catch it and send another job to the same MicroVM. An error that leaves the handler ends the session: -1. A MicroVM error that the handler raises itself reaches the caller with its own class, such as `MicrovmJobFailedError`. -2. Any other error reaches the caller as the SDK's `ChildContextError`, as from `run_in_child_context`. This includes a MicroVM error that leaves a child context that the handler created: that child context records `ChildContextError` as its error type. The caller then gets the MicroVM error's message, but not its class. Inside the handler, the `ChildContextError` of that child context has the MicroVM error as its `__cause__`. An SDK error that leaves the handler, such as the `StepError` of a failed step, also reaches the caller as `ChildContextError`, with the SDK error as its `__cause__`. This rule matches `run_in_child_context`. The JavaScript operation differs here: it passes an SDK error to the caller with its own class. +1. A MicroVM error that leaves the handler reaches the caller with its own class, such as `MicrovmJobFailedError`. This includes a MicroVM error that leaves a child context that the handler created: the SDK records the error that a `ChildContextError` wraps, not the `ChildContextError`. Inside the handler, the `ChildContextError` of that child context has the MicroVM error as its `__cause__`. +2. Any other error reaches the caller as the SDK's `ChildContextError`, as from `run_in_child_context`. An SDK error that leaves the handler, such as the `StepError` of a failed step, also reaches the caller as `ChildContextError`. Its `error_type` is the type of the error that failed the step, for example `LookupError`. This rule matches `run_in_child_context`. The JavaScript operation differs here: it passes an SDK error to the caller with its own class. An invalid job raises `ValidationError` or `TypeError` from `vm.invoke`, before the job's durable operations start. The handler can catch it too. diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py index 591048b83..695bfd680 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py @@ -113,13 +113,13 @@ def microvm_session( error that leaves the handler ends the session, and the MicroVM is terminated: - 1. A MicroVM error, such as :class:`MicrovmJobFailedError`, that the - handler raises itself reaches the caller with its own class. + 1. A MicroVM error, such as :class:`MicrovmJobFailedError`, that leaves + the handler reaches the caller with its own class. This includes a + MicroVM error that leaves a child context that the handler created: + the SDK records the error that the child context's + ``ChildContextError`` wraps. 2. Any other error reaches the caller as ``ChildContextError``, as from - ``run_in_child_context``. This includes a MicroVM error that leaves - a child context that the handler created: that child context records - ``ChildContextError`` as its error type. The caller then gets the - MicroVM error's message, but not its class. + ``run_in_child_context``. The session's value is the handler's return value, checkpointed as the child context result. So it must be serializable by the SDK's default diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py index 2e914e265..977e8a774 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py @@ -1016,14 +1016,13 @@ def body(vm: MicrovmSession, ctx: DurableContext) -> Any: ] -def test_wraps_a_job_error_that_leaves_a_child_context_of_the_handler( +def test_keeps_a_job_error_that_leaves_a_child_context_of_the_handler( run: Any, fake_client: Any, fake_endpoint: Any ) -> None: - # Only an error that the handler raises itself keeps its MicroVM class. - # The SDK records one error type per child context. So an error from the - # handler's own child context reaches the caller as ChildContextError, - # with the MicroVM error's message. Inside the handler, the - # ChildContextError's __cause__ is still the MicroVM class. + # The SDK records the error a ChildContextError wraps. So a MicroVM error + # that leaves the handler's own child context reaches the caller with its + # MicroVM class. Inside the handler, the ChildContextError's __cause__ is + # the MicroVM class. client = fake_client(endpoint=fake_endpoint().url) config = session_config(client) inside: list[str] = [] @@ -1042,14 +1041,14 @@ def body(vm: MicrovmSession, ctx: DurableContext) -> Any: try: microvm_session(context, "pipeline", config=config, handler=body) - except ChildContextError as error: - return [type(error.__cause__).__name__, error.message] + except MicrovmJobFailedError as error: + return [type(error).__name__, error.message] return None result = run(handler, None, [("build", fail_with("BuildError", "failed"))]) assert outcome(result) == [ - "ChildContextError", + "MicrovmJobFailedError", 'MicroVM "build": the job failed (BuildError): failed', ] assert "MicrovmJobFailedError" in inside @@ -1111,12 +1110,17 @@ def body(vm: MicrovmSession, ctx: DurableContext) -> Any: try: microvm_session(context, "pipeline", config=config, handler=body) except ChildContextError as error: - return [type(error.__cause__).__name__, error.message] + return [ + type(error.__cause__).__name__, + error.error_type, + error.message, + ] return None result = run(handler) - assert outcome(result) == ["StepError", "no artifact"] + # The session context records the step's LookupError, not StepError. + assert outcome(result) == ["DurableOperationError", "LookupError", "no artifact"] assert len(client.requests("terminate_microvm")) == 1 diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py index afc816dda..e23cfb2e2 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py @@ -305,25 +305,45 @@ def result(self) -> T | None: def _raise_terminal_failure( self, checkpointed_result: CheckpointedResult ) -> NoReturn: - """Raise the graded error for a terminal callback failure. + _raise_callback_terminal_failure(checkpointed_result) - TIMED_OUT -> CallbackTimeoutError; any other non-succeeded terminal - (external FAILED, cancelled, stopped) -> CallbackExternalError. - Carries the message/data/stack_trace; the external error's own type - stays on the callback operation's details. - """ - error: ErrorObject | None = checkpointed_result.error - is_timeout: bool = checkpointed_result.is_timed_out() - error_cls: type[CallbackError] = ( - CallbackTimeoutError if is_timeout else CallbackExternalError - ) - default_message: str = "Callback timed out" if is_timeout else "Callback failed" - message: str = error.message if error and error.message else default_message - raise error_cls( - message=message, - data=error.data if error else None, - stack_trace=error.stack_trace if error else None, - ) + +def _is_blank_type(error_type: str | None) -> bool: + """A missing type, or one made only of ASCII whitespace.""" + return error_type is None or all(ch in " \t\n\r\f\v" for ch in error_type) + + +def _raise_callback_terminal_failure( + checkpointed_result: CheckpointedResult, +) -> NoReturn: + """Raise the graded error for a terminal callback failure. + + TIMED_OUT -> CallbackTimeoutError; any other non-succeeded terminal + (external FAILED, cancelled, stopped) -> CallbackExternalError. + Carries the message/data/stack_trace. An external failure also carries + the sender's ErrorType as ``error_type``. + """ + error: ErrorObject | None = checkpointed_result.error + is_timeout: bool = checkpointed_result.is_timed_out() + error_cls: type[CallbackError] = ( + CallbackTimeoutError if is_timeout else CallbackExternalError + ) + default_message: str = "Callback timed out" if is_timeout else "Callback failed" + message: str = error.message if error and error.message else default_message + # An external failure keeps the sender's ErrorType, the explicit type + # "Error" included, so it is recorded and surfaced as that type. Only a + # missing or blank type falls back to the SDK class. + error_type: str | None = ( + error.type + if not is_timeout and error is not None and not _is_blank_type(error.type) + else None + ) + raise error_cls( + message=message, + error_type=error_type, + data=error.data if error else None, + stack_trace=error.stack_trace if error else None, + ) class DurableContext(DurableContextProtocol): @@ -908,9 +928,11 @@ def wait_for_callback( def wait_in_child_context(context: DurableContext): return wait_for_callback_handler(context, submitter, step_name, config) - # run_in_child_context is generic and raises ChildContextError with the - # inner failure on __cause__; translate to callback-typed errors here, at - # the call site, so it re-fires identically on replay. + # The context records the wrapped error's type, so the callback-typed + # error is recovered from which child failed: the submitter step or the + # callback. Both are read from checkpoints, so first run and replay + # raise the same error. + context_operation_id: str = self._peek_next_operation_id() try: return self.run_in_child_context( wait_in_child_context, @@ -918,19 +940,53 @@ def wait_in_child_context(context: DurableContext): ChildConfig(sub_type=OperationSubType.WAIT_FOR_CALLBACK), ) except ChildContextError as e: - inner: BaseException | None = e.__cause__ - # Callback failures (external/timeout/base) pass through. - if isinstance(inner, CallbackError): - raise inner - # A failed submitter step surfaces as CallbackSubmitterError, keeping - # the step's data/stack_trace (the StepError stays as its cause). - if isinstance(inner, StepError): - raise CallbackSubmitterError( - inner.message or "Callback submitter failed", - data=inner.data, - stack_trace=inner.stack_trace, - ) from inner - raise + self._raise_wait_for_callback_error(e, context_operation_id) + + def _raise_wait_for_callback_error( + self, error: ChildContextError, context_operation_id: str + ) -> NoReturn: + """Translate a failed wait_for_callback context to a callback error. + + A failed submitter step surfaces as CallbackSubmitterError and a + callback failure as CallbackExternalError or CallbackTimeoutError. + Checkpoints written before the context recorded the wrapped error hold + the callback error or StepError itself; those surface as before. + """ + inner: BaseException | None = error.__cause__ + # The context holds a callback error type: a timeout, a callback + # failure without a sender type, or the base CallbackError. + if isinstance(inner, CallbackError): + raise inner + if isinstance(inner, StepError): + # The context holds StepError itself. + raise CallbackSubmitterError( + inner.message or "Callback submitter failed", + data=inner.data, + stack_trace=inner.stack_trace, + ) from inner + namespace = OperationIdNamespace(prefix=context_operation_id) + # wait_for_callback_handler creates the callback first, then the step. + callback_result: CheckpointedResult = self.state.get_checkpoint_result( + namespace.create_id_for_step(1) + ) + submitter_result: CheckpointedResult = self.state.get_checkpoint_result( + namespace.create_id_for_step(2) + ) + if submitter_result.is_failed(): + raise CallbackSubmitterError( + error.message or "Callback submitter failed", + error_type=error.error_type, + data=error.data, + stack_trace=error.stack_trace, + ) from inner + if ( + callback_result.is_failed() + or callback_result.is_cancelled() + or callback_result.is_timed_out() + or callback_result.is_stopped() + ): + _raise_callback_terminal_failure(callback_result) + raise error def wait_for_condition( self, 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 ea3f163f2..a8737e740 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 @@ -414,6 +414,28 @@ class CallbackSubmitterError(CallbackError): } +# SDK wrapper types: each re-raises the failure that escaped an operation and +# carries that failure's fields (error_type, message, data, stack_trace). A +# checkpoint records the wrapped error, never the wrapper. Matched by exact +# class, so a user subclass is never unwrapped. CallbackTimeoutError, the base +# CallbackError, SerDesError, BatchCompletionError and ValidationError are SDK +# errors in their own right and are not wrappers. +SDK_WRAPPER_ERROR_TYPES: frozenset[type[DurableOperationError]] = frozenset( + { + DurableOperationError, + StepError, + InvokeError, + ChildContextError, + WaitForConditionError, + CallbackExternalError, + CallbackSubmitterError, + } +) +SDK_WRAPPER_ERROR_TYPE_NAMES: frozenset[str] = frozenset( + f"{cls.__module__}.{cls.__qualname__}" for cls in SDK_WRAPPER_ERROR_TYPES +) + + def register_operation_error(error_cls: type[DurableOperationError]) -> None: """Register an SDK-owned operation-error subclass for replay reconstruction. 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 89243e68c..24de8090c 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 @@ -17,6 +17,8 @@ CheckpointError, DurableOperationError, GetExecutionStateError, + SDK_WRAPPER_ERROR_TYPE_NAMES, + SDK_WRAPPER_ERROR_TYPES, SerDesError, ) @@ -200,6 +202,15 @@ def _qualified_error_type(exception: BaseException) -> str: return f"{module}{cls_.__qualname__}" +# Bound on the SDK wrappers unwrapped when recording an error. +_MAX_WRAPPER_UNWRAPS: int = 16 + + +def _is_sdk_wrapper(exception: BaseException) -> bool: + # Exact class match: a user subclass of an SDK wrapper is the user's error. + return type(exception) in SDK_WRAPPER_ERROR_TYPES + + @dataclass(frozen=True) class ErrorObject: message: str | None @@ -218,6 +229,46 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> ErrorObject: @classmethod def from_exception(cls, exception: Exception) -> ErrorObject: + """Build the ErrorObject that records ``exception``. + + An SDK wrapper (StepError, ChildContextError, InvokeError, + CallbackExternalError, ...) is unwrapped first, so the record carries + the type, message, data and stack trace of the error it wraps rather + than the wrapper's class. A wrapper keeps those fields itself: its + ``error_type`` names the wrapped error. When ``error_type`` names + another wrapper, the walk continues with ``__cause__``. The walk takes + at most ``_MAX_WRAPPER_UNWRAPS`` steps and stops at a cause it has + already visited; a wrapper it stops at is recorded by its + ``error_type``. + """ + current: Exception = exception + visited: set[int] = {id(current)} + for _ in range(_MAX_WRAPPER_UNWRAPS): + if not _is_sdk_wrapper(current): + break + wrapper = cast(DurableOperationError, current) + if wrapper.error_type not in SDK_WRAPPER_ERROR_TYPE_NAMES: + return cls._from_wrapper(wrapper) + cause = wrapper.__cause__ + if not isinstance(cause, Exception) or id(cause) in visited: + break + visited.add(id(cause)) + current = cause + if _is_sdk_wrapper(current): + return cls._from_wrapper(cast(DurableOperationError, current)) + return cls._from_unwrapped_exception(current) + + @classmethod + def _from_wrapper(cls, wrapper: DurableOperationError) -> ErrorObject: + return cls( + message=wrapper.message, + type=wrapper.error_type, + data=wrapper.data, + stack_trace=wrapper.stack_trace, + ) + + @classmethod + def _from_unwrapped_exception(cls, exception: Exception) -> ErrorObject: # SerDesError and subclasses pin to the base discriminator so replay # always reconstructs them as SerDesError. if isinstance(exception, SerDesError): diff --git a/packages/aws-durable-execution-sdk-python/tests/context_test.py b/packages/aws-durable-execution-sdk-python/tests/context_test.py index e3cc07acd..c50c2eabe 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -217,11 +217,12 @@ def test_callback_result_failed(): callback = Callback("callback5", "op5", mock_state) # A FAILED callback (external SendDurableExecutionCallbackFailure) surfaces - # as CallbackExternalError. The external error's own type stays on the - # callback operation, and there is no synthetic __cause__. + # as CallbackExternalError carrying the sender's type as error_type, and + # there is no synthetic __cause__. with pytest.raises(CallbackExternalError) as exc_info: callback.result() assert exc_info.value.message == "Callback failed" + assert exc_info.value.error_type == "CallbackError" assert isinstance(exc_info.value, CallbackError) assert exc_info.value.__cause__ is None @@ -1582,6 +1583,10 @@ def test_wait_for_callback_reraises_unrelated_child_context_error(): """A non-callback, non-step failure re-raises the original ChildContextError.""" mock_state = Mock(spec=ExecutionState) mock_state.durable_execution_arn = "test_arn" + # Neither the callback nor the submitter step failed. + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) context = create_test_context(state=mock_state) inner = InvokeError("nested invoke failed") diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_wrapped_error_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_wrapped_error_int_test.py new file mode 100644 index 000000000..1070e8941 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_wrapped_error_int_test.py @@ -0,0 +1,389 @@ +"""A failed context checkpoints the error it wraps, not the SDK wrapper. + +Each scenario runs through the local test runner twice: live, and replayed. +"Replayed" means the handler catches the error, waits so that a new invocation +starts, and rethrows it, so the second invocation rebuilds the error from the +checkpoint. Every scenario asserts: + +* the ErrorType checkpointed by each failed operation and by the execution; +* the error user code sees has the same shape live and on replay. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable + +import pytest + +from aws_durable_execution_sdk_python.config import ( + CompletionConfig, + Duration, + MapConfig, + ParallelConfig, + StepConfig, + WaitForCallbackConfig, +) +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.exceptions import ( + CallbackExternalError, + CallbackSubmitterError, + CallbackTimeoutError, + ChildContextError, + DurableOperationError, + StepError, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.retries import ( + RetryStrategyConfig, + create_retry_strategy, +) +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) +from aws_durable_execution_sdk_python_testing.runner import ( + ContextOperation, + DurableFunctionTestRunner, + Operation, +) + + +MESSAGE = "card declined" + + +class PaymentError(Exception): + pass + + +class FraudError(Exception): + pass + + +PAYMENT = f"{PaymentError.__module__}.PaymentError" +FRAUD = f"{FraudError.__module__}.FraudError" + + +def _sdk(cls: type) -> str: + return f"{cls.__module__}.{cls.__qualname__}" + + +NO_RETRY = StepConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) +) +NO_RETRY_SUBMITTER = WaitForCallbackConfig( + retry_strategy=create_retry_strategy(RetryStrategyConfig(max_attempts=1)) +) + + +def _raise(exc: BaseException) -> Callable[..., Any]: + def fail(*_args: Any) -> Any: + raise exc + + return fail + + +def _failing_step(ctx: DurableContext) -> Any: + return ctx.step(_raise(PaymentError(MESSAGE)), name="charge", config=NO_RETRY) + + +def _child_step(ctx: DurableContext) -> Any: + return ctx.run_in_child_context(_failing_step, name="checkout") + + +def _nested_child_step(ctx: DurableContext) -> Any: + return ctx.run_in_child_context(_child_step, name="outer") + + +def _map_item_step(ctx: DurableContext) -> Any: + def item(c: DurableContext, *_args: Any) -> Any: + return _failing_step(c) + + ctx.map(["a"], item, name="fanout").throw_if_error() + + +def _map_two_types(ctx: DurableContext) -> Any: + def item(_c: DurableContext, value: str, *_args: Any) -> Any: + raise (PaymentError if value == "a" else FraudError)(MESSAGE) + + # Both items run, one at a time, so item 0's error is the one raised. + config: MapConfig[Any] = MapConfig( + max_concurrency=1, completion_config=CompletionConfig.all_completed() + ) + ctx.map(["a", "b"], item, name="fanout", config=config).throw_if_error() + + +def _parallel_branch_step(ctx: DurableContext) -> Any: + ctx.parallel([lambda _c: "ok", _failing_step], name="branches").throw_if_error() + + +def _parallel_two_types(ctx: DurableContext) -> Any: + config = ParallelConfig( + max_concurrency=1, completion_config=CompletionConfig.all_completed() + ) + ctx.parallel( + [_raise(PaymentError(MESSAGE)), _raise(FraudError(MESSAGE))], + name="branches", + config=config, + ).throw_if_error() + + +def _child_wait_for_condition(ctx: DurableContext) -> Any: + def child(c: DurableContext) -> Any: + return c.wait_for_condition( + _raise(PaymentError(MESSAGE)), + WaitForConditionConfig( + wait_strategy=lambda _s, _n: WaitForConditionDecision.stop_polling(), + initial_state=0, + ), + name="poll", + ) + + return ctx.run_in_child_context(child, name="checkout") + + +@durable_execution +def _payments_handler(event: Any, context: DurableContext) -> Any: + raise PaymentError(MESSAGE) + + +def _child_invoke(ctx: DurableContext) -> Any: + return ctx.run_in_child_context( + lambda c: c.invoke("payments", {}, name="pay"), name="checkout" + ) + + +def _submitter_fails(ctx: DurableContext) -> Any: + return ctx.wait_for_callback( + _raise(PaymentError(MESSAGE)), name="approval", config=NO_RETRY_SUBMITTER + ) + + +def _callback(ctx: DurableContext) -> Any: + return ctx.wait_for_callback(lambda _id, _c: None, name="approval") + + +def _callback_times_out(ctx: DurableContext) -> Any: + return ctx.wait_for_callback( + lambda _id, _c: None, + name="approval", + config=WaitForCallbackConfig(timeout=Duration.from_seconds(1)), + ) + + +def _fail_callback(error_type: str | None) -> Callable[[Any, str], None]: + def drive(runner: DurableFunctionTestRunner, arn: str) -> None: + callback_id = runner.wait_for_callback(arn, timeout=30) + runner.send_callback_failure( + callback_id, + ErrorObject(message=MESSAGE, type=error_type, data=None, stack_trace=None), + ) + + return drive + + +@dataclass +class Scenario: + name: str + body: Callable[[DurableContext], Any] + # Checkpointed ErrorType by operation name (failed operations only). + operations: dict[str, str] + execution: str + # Class, error_type and cause class of the error user code sees. + raised: tuple[type, str, type | None] + message: str = MESSAGE + drive: Callable[[Any, str], None] | None = None + + +SCENARIOS = [ + Scenario( + "child-context-step", + _child_step, + {"checkout": PAYMENT, "charge": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "nested-child-contexts", + _nested_child_step, + {"outer": PAYMENT, "checkout": PAYMENT, "charge": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "map-item-step", + _map_item_step, + {"map-item-0": PAYMENT, "charge": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "map-items-two-types", + _map_two_types, + {"map-item-0": PAYMENT, "map-item-1": FRAUD}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "parallel-branch-step", + _parallel_branch_step, + {"parallel-branch-1": PAYMENT, "charge": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "parallel-branches-two-types", + _parallel_two_types, + {"parallel-branch-0": PAYMENT, "parallel-branch-1": FRAUD}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "child-context-wait-for-condition", + _child_wait_for_condition, + {"checkout": PAYMENT, "poll": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "child-context-invoke", + _child_invoke, + {"checkout": PAYMENT, "pay": PAYMENT}, + PAYMENT, + (ChildContextError, PAYMENT, DurableOperationError), + ), + Scenario( + "wait-for-callback-submitter", + _submitter_fails, + {"approval": PAYMENT, "approval submitter": PAYMENT}, + PAYMENT, + (CallbackSubmitterError, PAYMENT, DurableOperationError), + ), + Scenario( + "wait-for-callback-external-type", + _callback, + {"approval": "com.acme.PaymentError"}, + "com.acme.PaymentError", + (CallbackExternalError, "com.acme.PaymentError", None), + drive=_fail_callback("com.acme.PaymentError"), + ), + Scenario( + "wait-for-callback-external-error-type", + _callback, + {"approval": "Error"}, + "Error", + (CallbackExternalError, "Error", None), + drive=_fail_callback("Error"), + ), + Scenario( + "wait-for-callback-external-no-type", + _callback, + {"approval": _sdk(CallbackExternalError)}, + _sdk(CallbackExternalError), + (CallbackExternalError, _sdk(CallbackExternalError), None), + drive=_fail_callback(None), + ), + Scenario( + "wait-for-callback-external-blank-type", + _callback, + {"approval": _sdk(CallbackExternalError)}, + _sdk(CallbackExternalError), + (CallbackExternalError, _sdk(CallbackExternalError), None), + drive=_fail_callback(" \t"), + ), + Scenario( + "wait-for-callback-timeout", + _callback_times_out, + {"approval": _sdk(CallbackTimeoutError)}, + _sdk(CallbackTimeoutError), + (CallbackTimeoutError, _sdk(CallbackTimeoutError), None), + message="Callback timed out", + ), + Scenario( + "step", + _failing_step, + {"charge": PAYMENT}, + PAYMENT, + (StepError, PAYMENT, DurableOperationError), + ), +] + + +def _shape(exc: BaseException) -> tuple[Any, ...]: + cause = exc.__cause__ + return ( + type(exc), + getattr(exc, "error_type", None), + str(exc), + getattr(exc, "data", None), + getattr(exc, "stack_trace", None), + type(cause) if cause is not None else None, + getattr(cause, "error_type", None), + str(cause) if cause is not None else None, + ) + + +def _run(scenario: Scenario, *, replay: bool) -> tuple[Any, list[tuple[Any, ...]]]: + seen: list[tuple[Any, ...]] = [] + + @durable_execution + def handler(event: Any, ctx: DurableContext) -> Any: + try: + scenario.body(ctx) + except Exception as exc: + seen.append(_shape(exc)) + if not replay: + raise + saved = exc + else: # pragma: no cover - every scenario fails + return "no failure" + ctx.wait(Duration.from_seconds(1), name="pause") + raise saved + + with DurableFunctionTestRunner(handler=handler, execution_timeout=60) as runner: + runner.register_durable_function("payments", _payments_handler) + arn = runner.run_async(input="{}", execution_timeout=60) + if scenario.drive is not None: + scenario.drive(runner, arn) + result = runner.wait_for_result(arn, timeout=60) + return result, seen + + +def _failed_operations(operations: list[Operation]) -> dict[str, ErrorObject]: + failed: dict[str, ErrorObject] = {} + for op in operations: + error = getattr(op, "error", None) + if error is not None and op.name is not None: + failed[op.name] = error + if isinstance(op, ContextOperation): + failed.update(_failed_operations(op.child_operations)) + return failed + + +@pytest.mark.parametrize("replay", [False, True], ids=["live", "replayed"]) +@pytest.mark.parametrize("scenario", SCENARIOS, ids=[s.name for s in SCENARIOS]) +def test_failed_context_checkpoints_the_wrapped_error( + scenario: Scenario, replay: bool +) -> None: + result, seen = _run(scenario, replay=replay) + + assert result.status.value == "FAILED" + assert result.error is not None + assert result.error.type == scenario.execution + assert result.error.message == scenario.message + failed = _failed_operations(result.operations) + for name, expected in scenario.operations.items(): + assert failed[name].type == expected, name + assert failed[name].message == scenario.message, name + + raised_cls, raised_type, cause_cls = scenario.raised + assert seen, "user code saw no error" + first = seen[0] + assert first[0] is raised_cls + assert first[1] == raised_type + assert first[5] is cause_cls + # Live and replay surface the same error shape to user code. + assert all(shape == first for shape in seen) + if replay: + assert len(seen) == 2 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 index 292a1707e..6d067bdc2 100644 --- 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 @@ -94,13 +94,10 @@ def my_handler(event, context: DurableContext) -> str: result = my_handler(event, _lambda_context()) - # Execution failed; the result (one level up from the step) carries the - # StepError wrapper, and StepError.error_type is the original error type. + # Execution failed; the result records the error the StepError wraps, + # not the wrapper. assert result["Status"] == InvocationStatus.FAILED.value - assert ( - result["Error"]["ErrorType"] - == "aws_durable_execution_sdk_python.exceptions.StepError" - ) + assert result["Error"]["ErrorType"] == "ValueError" assert result["Error"]["ErrorMessage"] == "step boom" # The step's own FAIL checkpoint records the raw escaping error type. @@ -174,12 +171,9 @@ def my_handler(event, context: DurableContext) -> str: result = my_handler(event, _lambda_context()) # Reconstructed as a StepError with the checkpointed message; the step - # body is never re-executed. + # body is never re-executed. The result records the wrapped error. assert result["Status"] == InvocationStatus.FAILED.value - assert ( - result["Error"]["ErrorType"] - == "aws_durable_execution_sdk_python.exceptions.StepError" - ) + assert result["Error"]["ErrorType"] == "ValueError" assert result["Error"]["ErrorMessage"] == "prior boom" assert executed["count"] == 0 @@ -268,8 +262,10 @@ def my_handler(event, context: DurableContext) -> str: def test_failed_submitter_reconstructs_callback_submitter_error_on_replay(): - """A failed submitter records StepError at the context level but surfaces - CallbackSubmitterError to the caller on replay (console-parity contract).""" + """A context checkpoint written before contexts recorded the wrapped error + holds StepError. It still surfaces CallbackSubmitterError to the caller on + replay, and the result records the deepest wrapper type the checkpoint + holds (StepError).""" submitter_runs = {"count": 0} def submitter(_callback_id, _ctx) -> None: @@ -325,7 +321,7 @@ def my_handler(event, context: DurableContext) -> str: assert result["Status"] == InvocationStatus.FAILED.value assert ( result["Error"]["ErrorType"] - == "aws_durable_execution_sdk_python.exceptions.CallbackSubmitterError" + == "aws_durable_execution_sdk_python.exceptions.StepError" ) assert result["Error"]["ErrorMessage"] == "submitter boom" assert submitter_runs["count"] == 0 diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/invoke_propagation_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/invoke_propagation_int_test.py index c39a3c4ba..0130e8fbc 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/invoke_propagation_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/invoke_propagation_int_test.py @@ -192,7 +192,8 @@ def test_public_invoke_pending_and_terminal_replay_do_not_recollect( else: assert result["Status"] == "FAILED" assert result["Error"]["ErrorMessage"] == "saved child error" - assert result["Error"]["ErrorType"].endswith("InvokeError") + # The result records the invoked function's error, not InvokeError. + assert result["Error"]["ErrorType"] == "ChildError" @pytest.mark.parametrize( 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 1c83797a4..6155e6f5b 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 @@ -1,6 +1,7 @@ """Tests for the service module.""" import datetime +import itertools from datetime import UTC from unittest.mock import Mock, patch @@ -8,11 +9,18 @@ from aws_durable_execution_sdk_python.__about__ import __version__ from aws_durable_execution_sdk_python.exceptions import ( + CallbackError, + CallbackExternalError, + CallbackSubmitterError, + CallbackTimeoutError, CheckpointError, + ChildContextError, DurableOperationError, GetExecutionStateError, + InvokeError, SerDesError, StepError, + WaitForConditionError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -301,14 +309,139 @@ def test_error_object_to_durable_operation_error_unknown_type_falls_back(): 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") +def test_error_object_from_exception_records_the_wrapped_error(): + """An SDK wrapper is recorded as the error it wraps, not as the wrapper.""" + step_error = StepError( + message="step failed", + error_type="ValueError", + data="payload", + stack_trace=["frame"], + ) + error_object = ErrorObject.from_exception(step_error) + assert error_object == ErrorObject( + message="step failed", type="ValueError", data="payload", stack_trace=["frame"] + ) + + +@pytest.mark.parametrize( + "wrapper_cls", + [ + DurableOperationError, + StepError, + InvokeError, + ChildContextError, + WaitForConditionError, + CallbackExternalError, + CallbackSubmitterError, + ], +) +def test_error_object_from_exception_unwraps_every_sdk_wrapper(wrapper_cls): + wrapper = wrapper_cls(message="card declined", error_type="app.PaymentError") + assert ErrorObject.from_exception(wrapper).type == "app.PaymentError" + + +def test_error_object_from_exception_follows_cause_through_wrapper_types(): + """A wrapper whose error_type names another wrapper is unwrapped via __cause__.""" + step_error = StepError(message="card declined", error_type="app.PaymentError") + child_error = ChildContextError( + message="card declined", + error_type="aws_durable_execution_sdk_python.exceptions.StepError", + ) + child_error.__cause__ = step_error + assert ErrorObject.from_exception(child_error).type == "app.PaymentError" + + +def test_error_object_from_exception_unwraps_to_a_user_cause(): + """A wrapper with no wrapped type is unwrapped to its __cause__.""" step_error = StepError(message="step failed", data="payload") - step_error.__cause__ = cause + step_error.__cause__ = ValueError("boom") error_object = ErrorObject.from_exception(step_error) - assert error_object.type == "aws_durable_execution_sdk_python.exceptions.StepError" - assert error_object.message == "step failed" + assert error_object.type == "ValueError" + assert error_object.message == "boom" + assert error_object.data is None + + +def test_error_object_from_exception_keeps_deepest_wrapper_without_cause(): + """A chain that ends at a wrapper records that wrapper's type.""" + child_error = ChildContextError( + message="m", error_type="aws_durable_execution_sdk_python.exceptions.StepError" + ) + child_error.__cause__ = StepError(message="m") + assert ( + ErrorObject.from_exception(child_error).type + == "aws_durable_execution_sdk_python.exceptions.StepError" + ) + assert ( + ErrorObject.from_exception(StepError(message="m")).type + == "aws_durable_execution_sdk_python.exceptions.StepError" + ) + + +def test_error_object_from_exception_unwraps_at_most_16_wrappers(): + wrapper_classes = [ + StepError, + ChildContextError, + InvokeError, + CallbackSubmitterError, + ] + + def chain(length: int) -> Exception: + wrappers = [wrapper_classes[i % 4](message=f"w{i}") for i in range(length)] + for outer, inner in itertools.pairwise(wrappers): + outer.__cause__ = inner + wrappers[-1].__cause__ = ValueError("root") + return wrappers[0] + + assert ErrorObject.from_exception(chain(16)).type == "ValueError" + # After 16 unwraps the 17th wrapper (index 16) is still a wrapper: reported. + deep = ErrorObject.from_exception(chain(20)) + assert deep.type == "aws_durable_execution_sdk_python.exceptions.StepError" + assert deep.message == "w16" + + +def test_error_object_from_exception_stops_at_a_cause_cycle(): + """A repeat stops the walk at the last error visited before it.""" + first = StepError(message="first") + second = ChildContextError(message="second") + first.__cause__ = second + second.__cause__ = first + error_object = ErrorObject.from_exception(first) + assert error_object.message == "second" + assert ( + error_object.type + == "aws_durable_execution_sdk_python.exceptions.ChildContextError" + ) + + +class _UserStepError(StepError): + pass + + +@pytest.mark.parametrize( + "error", + [ + CallbackTimeoutError(message="Callback timed out"), + CallbackError(message="internal"), + _UserStepError(message="user subclass", error_type="ValueError"), + ], +) +def test_error_object_from_exception_does_not_unwrap_other_errors(error): + """Only the SDK's own wrapper classes are unwrapped, matched exactly.""" + error_object = ErrorObject.from_exception(error) + assert error_object.type == f"{type(error).__module__}.{type(error).__qualname__}" + assert error_object.message == error.message + + +def test_error_object_from_exception_preserves_operation_error_fields(): + """A non-wrapper DurableOperationError keeps its own type and data.""" + timeout = CallbackTimeoutError(message="timed out", data="payload") + timeout.__cause__ = ValueError("boom") + error_object = ErrorObject.from_exception(timeout) + assert ( + error_object.type + == "aws_durable_execution_sdk_python.exceptions.CallbackTimeoutError" + ) + assert error_object.message == "timed out" assert error_object.data == "payload"