From 69af454e85ac1b691641f9926c3cd759cc7966f6 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 09:27:47 -0700 Subject: [PATCH 1/3] fix: checkpoint the wrapped error for failed contexts When an SDK wrapper (StepError, ChildContextError, InvokeError, WaitForConditionError, CallbackExternalError, CallbackSubmitterError or DurableOperationError) escaped a child context, map item, parallel branch or wait_for_callback context, the context checkpointed the wrapper's class as its ErrorType. The type of the error that caused the failure was lost one level up. ErrorObject.from_exception now unwraps the SDK's own wrappers (exact class match, at most 16 steps, stopping at a repeated cause) and records the wrapped error's type, message, data and stack trace. This applies to every context checkpoint and to the execution result. CallbackTimeoutError, CallbackError, SerDesError and other SDK errors are not unwrapped. A callback failed externally now carries the sender's ErrorType on CallbackExternalError.error_type. A missing or blank type keeps the SDK class; the explicit type "Error" is kept as "Error". wait_for_callback tells a submitter failure from a callback failure by reading the checkpoints of the context's two children, so first run and replay raise the same error. Checkpoints that still hold StepError or a callback error type surface as before. --- ...est_wait_for_callback_submitter_failure.py | 11 +- .../context.py | 124 ++++-- .../exceptions.py | 22 + .../lambda_service.py | 51 +++ .../tests/context_test.py | 9 +- .../e2e/checkpoint_wrapped_error_int_test.py | 389 ++++++++++++++++++ .../tests/e2e/error_hierarchy_int_test.py | 24 +- .../tests/e2e/invoke_propagation_int_test.py | 3 +- .../tests/lambda_service_test.py | 145 ++++++- 9 files changed, 714 insertions(+), 64 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_wrapped_error_int_test.py 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/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..63a79ff80 --- /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( + 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" From 9322e9094c004036e884311a34ab25725a44c1bd Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 09:28:53 -0700 Subject: [PATCH 2/3] test: annotate the map config for mypy --- .../tests/e2e/checkpoint_wrapped_error_int_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index 63a79ff80..1070e8941 100644 --- 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 @@ -108,7 +108,7 @@ 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( + config: MapConfig[Any] = MapConfig( max_concurrency=1, completion_config=CompletionConfig.all_completed() ) ctx.map(["a", "b"], item, name="fanout", config=config).throw_if_error() From 90b12ea299dadd9aedf20efc9c20fe0dd913c34c Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Thu, 22 Oct 2026 09:38:47 -0700 Subject: [PATCH 3/3] test(extras): expect the wrapped error from a session The core now records the error a ChildContextError wraps. A MicroVM error that leaves a child context of the session handler therefore reaches the caller with its MicroVM class, and a step failure that leaves the handler carries the step's error type. Update the session docs and tests to match. --- .../README.md | 4 +-- .../microvm/session.py | 12 ++++----- .../tests/microvm/session_test.py | 26 +++++++++++-------- 3 files changed, 23 insertions(+), 19 deletions(-) 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