Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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"
}
]
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""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.

The handler also crosses a wait/replay boundary after catching the error. This
shows the typed StepError is deterministic across replay: on the resume
invocation the handler replays from the top, the failed step is reconstructed
from its checkpoint (its body is NOT re-executed), and it surfaces the identical
StepError, which the same ``except`` catches again.
"""

from typing import Any

from aws_durable_execution_sdk_python import StepError
from aws_durable_execution_sdk_python.config import Duration, 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, catch the typed StepError, then replay across a wait."""

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))
)

outcome: dict[str, Any] = {"handled": False}
try:
context.step(failing_step, name="charge-card", config=config)
except StepError as error:
# The class identifies the operation kind (StepError); error_type carries
# the original failure type (ValueError).
outcome = {
"handled": True,
"caught": type(error).__name__,
"error_type": error.error_type,
"message": error.message,
}

# Replay boundary: the execution suspends here and resumes as a new
# invocation that replays from the top. On resume, the failed step above is
# reconstructed from its checkpoint and raises the same StepError, which the
# except catches again — so `outcome` is identical on first run and replay.
context.wait(duration=Duration.from_seconds(1), name="post-failure-cooldown")

return outcome
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
"""Tests for catch_typed_step_error."""

import pytest
from aws_durable_execution_sdk_python.execution import InvocationStatus
from aws_durable_execution_sdk_python.lambda_service import (
OperationStatus,
OperationType,
)

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, identical across replay.

The handler catches the StepError and then crosses a wait/replay boundary, so
the returned ``outcome`` is produced on the resume (replay) invocation, while
the STEP FAILED checkpoint was written on the first invocation. Asserting the
surfaced error matches the persisted checkpoint error shows the typed error is
reconstructed identically across replays.
"""
with durable_runner:
result = durable_runner.run(input=None, timeout=10)

assert result.status is InvocationStatus.SUCCEEDED

# The error caught on the replay pass (returned as the execution result).
result_data = deserialize_operation_payload(result.result)
assert result_data == {
"handled": True,
"caught": "StepError",
"error_type": "ValueError",
"message": "payment declined",
}

# A WAIT operation proves the execution suspended and replayed after the
# error was caught, so the result above reflects the replay invocation.
wait_ops = [
op for op in result.operations if op.operation_type == OperationType.WAIT
]
assert len(wait_ops) >= 1

# The step's own checkpoint (written on the first run) is FAILED and records
# the raw escaping error type.
step_ops = [
op for op in result.operations if op.operation_type == OperationType.STEP
]
assert len(step_ops) == 1
checkpoint_error = step_ops[0].error
assert step_ops[0].status is OperationStatus.FAILED
assert checkpoint_error is not None

# Cross-replay identity: the error surfaced on the replay pass (result_data)
# matches the error persisted on the first run (the checkpoint).
assert result_data["error_type"] == checkpoint_error.type == "ValueError"
assert result_data["message"] == checkpoint_error.message == "payment declined"
Original file line number Diff line number Diff line change
Expand Up @@ -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": "CallbackError",
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -33,12 +38,17 @@

__all__ = [
"BatchResult",
"ChildContextError",
"DurableContext",
"DurableExecutionsError",
"DurableOperationError",
"InvocationError",
"InvokeError",
"ParallelBranch",
"StepContext",
"StepError",
"ValidationError",
"WaitForConditionError",
"WithRetryConfig",
"__version__",
"durable_execution",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
from typing import TYPE_CHECKING, Generic, TypeVar

from aws_durable_execution_sdk_python.exceptions import (
ChildContextError,
DurableOperationError,
InvalidStateError,
SuspendExecution,
)
Expand Down Expand Up @@ -260,8 +262,17 @@ def throw_if_error(self) -> None:
(item.error for item in self.all if item.status is BatchItemStatus.FAILED),
None,
)
if first_error:
raise first_error.to_callable_runtime_error()
if first_error is not None:
# Each item runs in a child context, so a failed item surfaces as a
# ChildContextError wrapping the escaping error (reconstructed as its
# typed form where possible). Remaining errors stay in get_errors().
cause: DurableOperationError = first_error.to_durable_operation_error()
raise ChildContextError(
message=first_error.message,
error_type=first_error.type,
data=first_error.data,
stack_trace=first_error.stack_trace,
) from cause

def get_results(self) -> list[R]:
return [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down Expand Up @@ -266,25 +270,81 @@ 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 concrete class identifies the
operation kind (so callers can ``except StepError``); the escaping error is
preserved as ``__cause__`` on the first run and reconstructed on replay.

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: Type name of the error that escaped the operation (the
original/inner error, e.g. ``"ValueError"``), not the operation
class. Defaults to the class name only when no escaping error is
known.
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):
Expand Down Expand Up @@ -397,37 +457,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."""

Expand Down
Loading
Loading