diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py index 207d37cd..e0270eaf 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py @@ -31,8 +31,10 @@ ) from aws_durable_execution_sdk_python.exceptions import ( DurableOperationError, + ExecutionError, InvalidStateError, InvocationError, + NonDeterministicExecutionError, OrphanedChildException, SuspendExecution, TimedSuspendExecution, @@ -154,6 +156,21 @@ def get_iteration_name(self, index: int) -> str: name: str | None = self.executables[index].name return name if name is not None else f"{self.name_prefix}{index}" + def _get_iteration_operation_identifier( + self, + executor_context: DurableContext, + executable: Executable[CallableType], + ) -> OperationIdentifier: + """Build the stable operation identity for one branch or iteration.""" + return OperationIdentifier( + operation_id=self.operation_id_namespace.create_id_for_step( + executable.index + ), + sub_type=self.sub_type_iteration, + parent_id=executor_context._parent_id, # noqa: SLF001 + name=self.get_iteration_name(executable.index), + ) + def _build_items_snapshot(self) -> tuple[CompletionItemStatus, ...]: """Build the per-branch status snapshot for the custom predicate. @@ -239,7 +256,11 @@ def execute( def submit(branch: Branch[CallableType, ResultType]) -> None: branch.start() pool.submit( - self._branch_worker, executor_context, events, branch.executable + self._branch_worker, + execution_state, + executor_context, + events, + branch.executable, ) try: @@ -479,15 +500,17 @@ def _create_result( def _branch_worker( self, + execution_state: ExecutionState, executor_context: DurableContext, events: queue.Queue[BranchEvent[ResultType]], executable: Executable[CallableType], ) -> None: """Worker-thread body: run one branch and report its outcome. - Converts every outcome into a :class:`BranchEvent` on the queue and - never raises into the pool. The coordinator loop is the sole - consumer of the events. + Converts every outcome into a :class:`BranchEvent` on the queue. Fatal + errors are also re-raised into the pool after posting their event; the + coordinator loop consumes the event and propagates the error on the + calling thread. """ try: result: ResultType = self._execute_item_in_child_context( @@ -507,6 +530,22 @@ def _branch_worker( executable.index, ) events.put(BranchEvent.orphaned(executable.index)) + except ExecutionError as e: + # Execution-terminal SDK errors (including nondeterminism) must + # bypass branch failure tolerance and custom completion policies. + parent_operation_id: str | None = executor_context._parent_id # noqa: SLF001 + if ( + parent_operation_id is not None + and execution_state.record_branch_fatal_error(parent_operation_id, e) + is False + ): + logger.debug( + "Ignoring fatal error from orphaned branch %s", + executable.index, + ) + return + events.put(BranchEvent.fatal(executable.index, e)) + raise except Exception as e: # noqa: BLE001 # A retryable error (e.g. RetryableSerDesError) escapes the batch: # the coordinator re-raises it so the invocation fails and the @@ -518,6 +557,17 @@ def _branch_worker( # Post a fatal event so the coordinator re-raises it on the # calling thread instead of blocking forever on the queue, then # let the exception propagate to the worker thread. + parent_operation_id = executor_context._parent_id # noqa: SLF001 + if ( + parent_operation_id is not None + and execution_state.record_branch_fatal_error(parent_operation_id, e) + is False + ): + logger.debug( + "Ignoring fatal error from orphaned branch %s", + executable.index, + ) + return events.put(BranchEvent.fatal(executable.index, e)) raise else: @@ -542,10 +592,10 @@ def _execute_item_in_child_context( and execution-order invariant. """ - operation_id: str = self.operation_id_namespace.create_id_for_step( - executable.index + operation_identifier = self._get_iteration_operation_identifier( + executor_context, executable ) - name: str = self.get_iteration_name(executable.index) + operation_id = operation_identifier.operation_id is_virtual: bool = self.nesting_type is NestingType.FLAT child_context: DurableContext = executor_context.create_child_context( @@ -554,13 +604,6 @@ def _execute_item_in_child_context( # For NESTED this is for branch's START/SUCCEED/FAIL checkpoints (not the children of the branch). # For FLAT `child_handler` skips checkpoints, so not used. # Construct it unconditionally to keep the call simple. - operation_identifier = OperationIdentifier( - operation_id=operation_id, - sub_type=self.sub_type_iteration, - parent_id=executor_context._parent_id, # noqa: SLF001 - name=name, - ) - # The branch/iteration container op is resolved here via child_handler, # bypassing context.run_in_child_context and therefore the parent's # `_replay_aware`. Replicate the two things `_replay_aware` would have @@ -571,13 +614,31 @@ def _execute_item_in_child_context( # de-duplicated during a map/parallel replay. # 2. Replay hook: a branch that already has a checkpoint was observed # in a prior invocation, so emit the plugin replay hook (once). - # Virtual (FLAT) branches do not checkpoint themselves, so neither - # applies; their inner operations still self-correct via `_replay_aware`. - if not is_virtual and child_context.is_replaying(): - branch_checkpoint = child_context.state.get_checkpoint_result(operation_id) - if not branch_checkpoint.is_existent(): + # Virtual (FLAT) branches do not checkpoint themselves. Therefore an + # existing branch-container checkpoint proves that replay changed from + # NESTED and must be rejected before child_handler can consume it. + if child_context.is_replaying(): + branch_checkpoint = child_context.state.get_checkpoint_result( + operation_identifier.operation_id + ) + if is_virtual: + if branch_checkpoint.is_existent(): + operation_identifier.validate_checkpoint( + branch_checkpoint.operation + ) + msg = ( + "Non-deterministic branch nesting at " + f"id={operation_identifier.operation_id!r}: " + "checkpoint contains a NESTED branch context but current " + "nesting is FLAT" + ) + raise NonDeterministicExecutionError( + msg, step_id=operation_identifier.operation_id + ) + elif not branch_checkpoint.is_existent(): child_context._set_replay_status_new() # noqa: SLF001 elif branch_checkpoint.operation is not None: + operation_identifier.validate_checkpoint(branch_checkpoint.operation) child_context.state.emit_operation_replay_hook( branch_checkpoint.operation ) @@ -650,12 +711,26 @@ def _replay_terminal_item( themselves, so re-executing the branch body over its inner operations' checkpoints discriminates success from failure. """ - operation_id: str = self.operation_id_namespace.create_id_for_step( - executable.index + operation_identifier = self._get_iteration_operation_identifier( + executor_context, executable ) checkpoint: CheckpointedResult = execution_state.get_checkpoint_result( - operation_id + operation_identifier.operation_id ) + operation_identifier.validate_checkpoint(checkpoint.operation) + if self.nesting_type is NestingType.NESTED and not checkpoint.is_terminal(): + checkpoint_status = ( + checkpoint.status.value if checkpoint.status is not None else None + ) + msg = ( + "Non-deterministic branch nesting at " + f"id={operation_identifier.operation_id!r}: " + "recorded terminal branch requires a terminal NESTED branch " + f"context checkpoint, got status={checkpoint_status!r}" + ) + raise NonDeterministicExecutionError( + msg, step_id=operation_identifier.operation_id + ) if checkpoint.is_succeeded(): result: ResultType = self._execute_item_in_child_context( executor_context, executable @@ -670,6 +745,10 @@ def _replay_terminal_item( flat_result: ResultType = self._execute_item_in_child_context( executor_context, executable ) + except ExecutionError: + # Nondeterminism and other execution-terminal SDK errors must + # not be downgraded to a failed FLAT item. + raise except Exception as e: # noqa: BLE001 if isinstance(e, InvocationError) and e.is_retryable(): # Escape the batch so the invocation fails and the backend @@ -694,10 +773,13 @@ def _replay_from_checkpoints( """ items: list[BatchItem[ResultType]] = [] for executable in self.executables: - operation_id = self.operation_id_namespace.create_id_for_step( - executable.index + operation_identifier = self._get_iteration_operation_identifier( + executor_context, executable + ) + checkpoint = execution_state.get_checkpoint_result( + operation_identifier.operation_id ) - checkpoint = execution_state.get_checkpoint_result(operation_id) + operation_identifier.validate_checkpoint(checkpoint.operation) result: ResultType | None = None error = None 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 ff1ab6bc..d1129760 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 @@ -83,7 +83,7 @@ if TYPE_CHECKING: - from collections.abc import Callable, Sequence + from collections.abc import Callable, Iterator, Sequence from aws_durable_execution_sdk_python.concurrency.models import BatchResult from aws_durable_execution_sdk_python.lambda_service import ErrorObject @@ -514,7 +514,9 @@ def _next_operation_exists(self) -> bool: return self._peek_next_checkpoint().is_existent() @contextmanager - def _replay_aware(self): + def _replay_aware( + self, operation_identifier: OperationIdentifier | None = None + ) -> Iterator[None]: """Wrap a single operation with replay-boundary detection. The operation kind is inferred from its own checkpoint (when one @@ -539,27 +541,33 @@ def _replay_aware(self): next_checkpoint: CheckpointedResult | None = ( self._peek_next_checkpoint() if was_replaying else None ) - next_exists: bool = ( - next_checkpoint.is_existent() if next_checkpoint is not None else False + next_operation = ( + next_checkpoint.operation if next_checkpoint is not None else None ) - next_terminal: bool = next_exists and next_checkpoint.is_terminal() - - next_is_step: bool = ( - next_exists - and next_checkpoint.operation.operation_type is OperationType.STEP - ) - # While replaying, an operation that already has a checkpoint was - # observed in a prior invocation. If the backend says the operation - # changed since the last invocation, notify plugins as an update rather - # than replayed history; otherwise notify as replayed history. State - # owns the dedup; the context owns the "only while replaying" gate. - if was_replaying and next_exists: + next_exists: bool = next_operation is not None + if was_replaying and next_operation is not None: + if operation_identifier is not None: + operation_identifier.validate_checkpoint(next_operation) + + # While replaying, an operation that already has a checkpoint was + # observed in a prior invocation. If the backend says the operation + # changed since the last invocation, notify plugins as an update rather + # than replayed history; otherwise notify as replayed history. State + # owns the dedup; the context owns the "only while replaying" gate. if self.state.is_operation_updated_since_last_invocation( - next_checkpoint.operation.operation_id + next_operation.operation_id ): - self.state.emit_operation_update_hook(next_checkpoint.operation) + self.state.emit_operation_update_hook(next_operation) else: - self.state.emit_operation_replay_hook(next_checkpoint.operation) + self.state.emit_operation_replay_hook(next_operation) + + next_terminal: bool = ( + next_checkpoint.is_terminal() if next_checkpoint is not None else False + ) + next_is_step: bool = ( + next_operation is not None + and next_operation.operation_type is OperationType.STEP + ) # Deferred flip applies only to non-step resume points. For step ops we # flip before instead, so don't defer. flip_after: bool = ( @@ -579,6 +587,21 @@ def _replay_aware(self): elif self.is_replaying() and not self._next_operation_exists(): self._set_replay_status_new() + @contextmanager + def _operation_replay_aware( + self, sub_type: OperationSubType, name: str | None = None + ) -> Iterator[OperationIdentifier]: + """Allocate an operation and validate its replay identity before hooks.""" + operation_identifier = OperationIdentifier( + operation_id=self._peek_next_operation_id(), + sub_type=sub_type, + parent_id=self._parent_id, + name=name, + ) + with self._replay_aware(operation_identifier): + self._create_step_id() + yield operation_identifier + # endregion replay status # region Operations @@ -602,22 +625,18 @@ def create_callback( """ if not config: config = CallbackConfig() - with self._replay_aware(): - operation_id: str = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.CALLBACK, name + ) as operation_identifier: executor: CallbackOperationExecutor = CallbackOperationExecutor( state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.CALLBACK, - parent_id=self._parent_id, - name=name, - ), + operation_identifier=operation_identifier, config=config, ) callback_id: str = executor.process() return Callback( callback_id=callback_id, - operation_id=operation_id, + operation_id=operation_identifier.operation_id, state=self.state, serdes=config.serdes, ) @@ -642,18 +661,14 @@ def invoke( """ if not config: config = InvokeConfig[P, R]() - with self._replay_aware(): - operation_id = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.CHAINED_INVOKE, name + ) as operation_identifier: executor: InvokeOperationExecutor[R] = InvokeOperationExecutor( function_name=function_name, payload=payload, state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.CHAINED_INVOKE, - parent_id=self._parent_id, - name=name, - ), + operation_identifier=operation_identifier, config=config, ) return executor.process() @@ -674,14 +689,10 @@ def map( if config is not None: config.completion_config._validate_for_total(len(inputs)) - with self._replay_aware(): - operation_id = self._create_step_id() - operation_identifier = OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.MAP, - parent_id=self._parent_id, - name=map_name, - ) + with self._operation_replay_aware( + OperationSubType.MAP, map_name + ) as operation_identifier: + operation_id = operation_identifier.operation_id map_context = self.create_child_context(operation_id=operation_id) def map_in_child_context() -> BatchResult[R]: @@ -729,16 +740,11 @@ def parallel( if config is not None: config.completion_config._validate_for_total(len(functions)) - with self._replay_aware(): - # _create_step_id() is thread-safe. rest of method is safe, since using local copy of parent id - operation_id = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.PARALLEL, name + ) as operation_identifier: + operation_id = operation_identifier.operation_id parallel_context = self.create_child_context(operation_id=operation_id) - operation_identifier = OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.PARALLEL, - parent_id=self._parent_id, - name=name, - ) def parallel_in_child_context() -> BatchResult[T]: # parallel_context is a child_context of the context upon which `.map` @@ -790,15 +796,14 @@ def run_in_child_context( T: The result of the callable. """ step_name: str | None = self._resolve_step_name(name, func) - with self._replay_aware(): - # _create_step_id() is thread-safe. rest of method is safe, since using local copy of parent id - operation_id = self._create_step_id() - sub_type = ( - config.sub_type - if config and config.sub_type - else OperationSubType.RUN_IN_CHILD_CONTEXT - ) + sub_type = ( + config.sub_type + if config and config.sub_type + else OperationSubType.RUN_IN_CHILD_CONTEXT + ) + with self._operation_replay_aware(sub_type, step_name) as operation_identifier: + operation_id = operation_identifier.operation_id is_virtual: bool = config.is_virtual if config else False def callable_with_child_context(): @@ -811,12 +816,7 @@ def callable_with_child_context(): return child_handler( func=callable_with_child_context, state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=sub_type, - parent_id=self._parent_id, - name=step_name, - ), + operation_identifier=operation_identifier, config=config, ) @@ -830,18 +830,14 @@ def step( logger.debug("Step name: %s", step_name) if not config: config = StepConfig() - with self._replay_aware(): - operation_id = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.STEP, step_name + ) as operation_identifier: executor: StepOperationExecutor[T] = StepOperationExecutor( func=func, config=config, state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.STEP, - parent_id=self._parent_id, - name=step_name, - ), + operation_identifier=operation_identifier, context_logger=self.logger, ) return executor.process() @@ -857,18 +853,14 @@ def wait(self, duration: Duration, name: str | None = None) -> None: if seconds < 1: msg = "duration must be at least 1 second" raise ValidationError(msg) - with self._replay_aware(): - operation_id = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.WAIT, name + ) as operation_identifier: wait_seconds = duration.seconds executor: WaitOperationExecutor = WaitOperationExecutor( seconds=wait_seconds, state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.WAIT, - parent_id=self._parent_id, - name=name, - ), + operation_identifier=operation_identifier, ) executor.process() @@ -931,19 +923,15 @@ def wait_for_condition( msg = "`config` is required for wait_for_condition" raise ValidationError(msg) - with self._replay_aware(): - operation_id = self._create_step_id() + with self._operation_replay_aware( + OperationSubType.WAIT_FOR_CONDITION, name + ) as operation_identifier: executor: WaitForConditionOperationExecutor[T] = ( WaitForConditionOperationExecutor( check=check, config=config, state=self.state, - operation_identifier=OperationIdentifier( - operation_id=operation_id, - sub_type=OperationSubType.WAIT_FOR_CONDITION, - parent_id=self._parent_id, - name=name, - ), + operation_identifier=operation_identifier, context_logger=self.logger, ) ) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py index 9a46bae1..0a7f1623 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/identifier.py @@ -5,7 +5,11 @@ import hashlib from dataclasses import dataclass +from aws_durable_execution_sdk_python.exceptions import ( + NonDeterministicExecutionError, +) from aws_durable_execution_sdk_python.lambda_service import ( + Operation, OperationType, OperationSubType, ) @@ -40,3 +44,42 @@ class OperationIdentifier: @property def type(self) -> OperationType: return OperationType.from_sub_type(self.sub_type) + + def validate_checkpoint(self, checkpoint: Operation | None) -> None: + """Ensure replay history belongs to this operation before it is consumed.""" + if not isinstance(checkpoint, Operation): + return + + expected_name = self.name or None + checkpoint_name = checkpoint.name or None + expected_parent_id = self.parent_id or None + checkpoint_parent_id = checkpoint.parent_id or None + mismatches: list[str] = [] + + if checkpoint.operation_type is not self.type: + mismatches.append( + f"type checkpoint={checkpoint.operation_type.value!r} current={self.type.value!r}" + ) + if checkpoint.sub_type is not self.sub_type: + checkpoint_sub_type = ( + checkpoint.sub_type.value if checkpoint.sub_type is not None else None + ) + mismatches.append( + f"subtype checkpoint={checkpoint_sub_type!r} current={self.sub_type.value!r}" + ) + if checkpoint_name != expected_name: + mismatches.append( + f"name checkpoint={checkpoint_name!r} current={expected_name!r}" + ) + if checkpoint_parent_id != expected_parent_id: + mismatches.append( + f"parent_id checkpoint={checkpoint_parent_id!r} current={expected_parent_id!r}" + ) + + if mismatches: + mismatch_details = ", ".join(mismatches) + msg = ( + "Non-deterministic operation identity at " + f"id={self.operation_id!r}: {mismatch_details}" + ) + raise NonDeterministicExecutionError(msg, step_id=self.operation_id) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/base.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/base.py index 5836cda8..126f63d5 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/base.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/base.py @@ -9,7 +9,11 @@ from aws_durable_execution_sdk_python.exceptions import InvalidStateError if TYPE_CHECKING: - from aws_durable_execution_sdk_python.state import CheckpointedResult + from aws_durable_execution_sdk_python.identifier import OperationIdentifier + from aws_durable_execution_sdk_python.state import ( + CheckpointedResult, + ExecutionState, + ) T = TypeVar("T") @@ -103,6 +107,17 @@ class OperationExecutor(ABC, Generic[T]): - execute(): Execute the operation logic with checkpoint data """ + state: ExecutionState + operation_identifier: OperationIdentifier + + def _get_checkpoint_result(self) -> CheckpointedResult: + """Return this operation's checkpoint after validating replay identity.""" + checkpointed_result = self.state.get_checkpoint_result( + self.operation_identifier.operation_id + ) + self.operation_identifier.validate_checkpoint(checkpointed_result.operation) + return checkpointed_result + @abstractmethod def check_result_status(self) -> CheckResult[T]: """Check operation status and create START checkpoint if needed. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/callback.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/callback.py index 67c51ebc..d579b3ce 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/callback.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/callback.py @@ -82,9 +82,7 @@ def check_result_status(self) -> CheckResult[str]: Raises: CallbackError: If callback_details are missing from checkpoint """ - checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # CRITICAL: Do NOT raise on FAILED - defer error to Callback.result() # If checkpoint exists (any status including FAILED), return ready to execute diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py index 4c048a3c..8287be64 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py @@ -8,6 +8,7 @@ from aws_durable_execution_sdk_python.config import ChildConfig from aws_durable_execution_sdk_python.exceptions import ( ChildContextError, + ExecutionError, InvocationError, SuspendExecution, ) @@ -82,9 +83,7 @@ def check_result_status(self) -> CheckResult[T]: Raises: ChildContextError: For FAILED operations """ - checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # Terminal success without replay_children - deserialize and return if ( @@ -271,6 +270,11 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: except SuspendExecution: # Don't checkpoint SuspendExecution - let it bubble up raise + except ExecutionError: + # Execution-terminal SDK errors (including nondeterminism) must + # escape unchanged without mutating history or being wrapped as a + # child failure. + raise except Exception as e: # Retryable InvocationError: re-raise with no FAIL checkpoint so the # backend retry re-runs. Non-retryable falls through to FAIL + wrap. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py index 891fce01..c6401397 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py @@ -84,9 +84,7 @@ def check_result_status(self) -> CheckResult[R]: InvokeError: For FAILED, TIMED_OUT, or STOPPED operations SuspendExecution: For STARTED operations waiting for completion """ - checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # Terminal success - deserialize and return if checkpointed_result.is_succeeded(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py index c6b6a102..ad99d3b5 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/map.py @@ -124,6 +124,7 @@ def map_handler( checkpoint: CheckpointedResult = execution_state.get_checkpoint_result( operation_identifier.operation_id ) + operation_identifier.validate_checkpoint(checkpoint.operation) if checkpoint.is_succeeded(): # if we've reached this point, then not only is the step succeeded, but it is also `replay_children`. return executor.replay(execution_state, map_context, checkpoint) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py index d4cb2703..5aad4da0 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/parallel.py @@ -114,6 +114,7 @@ def parallel_handler( checkpoint = execution_state.get_checkpoint_result( operation_identifier.operation_id ) + operation_identifier.validate_checkpoint(checkpoint.operation) if checkpoint.is_succeeded(): return executor.replay(execution_state, parallel_context, checkpoint) return executor.execute(execution_state, executor_context=parallel_context) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py index ac50f9bc..a9b9151e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py @@ -93,9 +93,7 @@ def check_result_status(self) -> CheckResult[T]: StepInterruptedError: For interrupted AT_MOST_ONCE operations SuspendExecution: For PENDING operations waiting for retry """ - checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # Terminal success - deserialize and return if checkpointed_result.is_succeeded(): @@ -175,9 +173,7 @@ def check_result_status(self) -> CheckResult[T]: # After creating sync checkpoint, check the status if is_sync: # Refresh checkpoint result to check for immediate response - refreshed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + refreshed_result = self._get_checkpoint_result() # START checkpoint only returns STARTED status # Any errors would be thrown as runtime exceptions during checkpoint creation diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait.py index fc16e664..48729dd7 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait.py @@ -58,9 +58,7 @@ def check_result_status(self) -> CheckResult[None]: Raises: SuspendExecution: When wait timer has not completed """ - checkpointed_result: CheckpointedResult = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # Terminal success - wait completed if checkpointed_result.is_succeeded(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py index 1076d842..03253203 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py @@ -109,9 +109,7 @@ def check_result_status(self) -> CheckResult[T]: WaitForConditionError: For FAILED operations SuspendExecution: For PENDING operations waiting for retry """ - checkpointed_result = self.state.get_checkpoint_result( - self.operation_identifier.operation_id - ) + checkpointed_result = self._get_checkpoint_result() # Check if already completed if checkpointed_result.is_succeeded(): diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py index 8642c0ff..0b6a5fcf 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py @@ -307,6 +307,11 @@ def __init__( # Operations whose parent has completed self._parent_done: set[str] = set() + # Map/parallel contexts whose terminal checkpoint has started. Fatal + # branch errors race terminal parent checkpoints under + # _parent_done_lock so a detected replay mismatch cannot be lost. + self._terminal_contexts: set[str] = set() + self._branch_fatal_errors: dict[str, BaseException] = {} # Protects parent_to_children and parent_done. When both state locks are # required, acquire _completion_lock before _parent_done_lock. @@ -646,6 +651,11 @@ def create_checkpoint( and operation_update.action in {OperationAction.SUCCEED, OperationAction.FAIL} ): + if fatal_error := self._branch_fatal_errors.get( + operation_update.operation_id + ): + raise fatal_error + self._terminal_contexts.add(operation_update.operation_id) self._mark_orphans(operation_update.operation_id) # Check if this operation's parent is done @@ -991,6 +1001,22 @@ def register_branch_pool(self, pool: ThreadPoolExecutor) -> None: with self._branch_pools_lock: self._branch_pools.append(pool) + def record_branch_fatal_error( + self, parent_operation_id: str, error: BaseException + ) -> bool: + """Retain a fatal branch error until its parent context checkpoints. + + Returns False when the parent terminal checkpoint already won the + race, making the reporting branch an orphan. Otherwise the first fatal + error is retained and raised before the parent can checkpoint a + terminal result. + """ + with self._parent_done_lock: + if parent_operation_id in self._terminal_contexts: + return False + self._branch_fatal_errors.setdefault(parent_operation_id, error) + return True + def _collect_checkpoint_batch(self) -> list[QueuedOperation]: """Collect multiple checkpoint operations into a batch for API efficiency. diff --git a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py index 5bc73c8c..a1af7f39 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -55,25 +55,33 @@ SerDesError, ValidationError, InvalidStateError, + NonDeterministicExecutionError, OrphanedChildException, SuspendExecution, TimedSuspendExecution, ) from aws_durable_execution_sdk_python.lambda_service import ( + DurableServiceClient, ErrorObject, Operation, OperationStatus, OperationSubType, OperationType, ) -from aws_durable_execution_sdk_python.identifier import OperationIdNamespace - - +from aws_durable_execution_sdk_python.identifier import ( + OperationIdentifier, + OperationIdNamespace, +) +from aws_durable_execution_sdk_python.operation.child import child_handler from aws_durable_execution_sdk_python.operation.map import MapExecutor from aws_durable_execution_sdk_python.operation.parallel import ( ParallelExecutor, ) -from aws_durable_execution_sdk_python.state import CheckpointedResult +from aws_durable_execution_sdk_python.plugin import PluginExecutor +from aws_durable_execution_sdk_python.state import ( + CheckpointedResult, + ExecutionState, +) class _StubNamespace(OperationIdNamespace): @@ -862,7 +870,7 @@ def execute_item(self, child_context, executable): tolerated_failure_percentage=None, ), sub_type_top="TOP", - sub_type_iteration="ITER", + sub_type_iteration=OperationSubType.PARALLEL_BRANCH, name_prefix="test_", serdes=None, nesting_type=NestingType.NESTED, @@ -879,7 +887,9 @@ def execute_item(self, child_context, executable): operation_id="branch-1", operation_type=OperationType.CONTEXT, status=OperationStatus.SUCCEEDED, + parent_id="parent", sub_type=OperationSubType.PARALLEL_BRANCH, + name="test_0", ) existing = CheckpointedResult.create_from_operation(branch_op) child_context = Mock() @@ -894,7 +904,7 @@ def execute_item(self, child_context, executable): def test_execute_item_virtual_branch_skips_replay_status_handling(): - """FLAT (virtual) branches don't checkpoint, so no flip or hook is attempted.""" + """A normal FLAT replay verifies branch-container checkpoint absence.""" class TestExecutor(ConcurrentExecutor): def execute_item(self, child_context, executable): @@ -923,15 +933,61 @@ def execute_item(self, child_context, executable): executor_context._parent_id = "parent" # noqa: SLF001 child_context = Mock() + child_context.is_replaying.return_value = True + child_context.state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) child_context.state.wrap_user_function = lambda func, *args, **kwargs: func executor_context.create_child_context = lambda *args, **kwargs: child_context executor._execute_item_in_child_context(executor_context, executables[0]) # noqa: SLF001 + assert child_context.state.get_checkpoint_result.call_count == 2 + child_context.state.get_checkpoint_result.assert_called_with("op_0") child_context.state.emit_operation_replay_hook.assert_not_called() child_context._set_replay_status_new.assert_not_called() # noqa: SLF001 +def test_execute_item_flat_branch_rejects_nested_container_checkpoint(): + """FLAT replay rejects a branch container left by NESTED history.""" + + executor = _RecordingExecutor( + executables=[Executable(0, lambda: "must-not-run")], + max_concurrency=1, + completion_config=CompletionConfig(min_successful=1), + sub_type_top=OperationSubType.PARALLEL, + sub_type_iteration=OperationSubType.PARALLEL_BRANCH, + name_prefix="parallel-branch-", + serdes=None, + nesting_type=NestingType.FLAT, + operation_id_namespace=_StubNamespace(), + ) + executor_context = Mock() + executor_context._parent_id = "parallel-op" # noqa: SLF001 + child_context = Mock() + child_context.is_replaying.return_value = True + child_context.state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_from_operation( + Operation( + operation_id="op_0", + operation_type=OperationType.CONTEXT, + status=OperationStatus.SUCCEEDED, + parent_id="parallel-op", + sub_type=OperationSubType.PARALLEL_BRANCH, + name="parallel-branch-0", + ) + ) + ) + executor_context.create_child_context.return_value = child_context + + with pytest.raises(NonDeterministicExecutionError, match="nesting is FLAT"): + executor._execute_item_in_child_context( # noqa: SLF001 + executor_context, executor.executables[0] + ) + + child_context.state.wrap_user_function.assert_not_called() + + def test_concurrent_executor_create_result_failure_tolerance_exceeded(): """Test ConcurrentExecutor with failure tolerance exceeded using public execute method.""" @@ -1470,6 +1526,7 @@ def _serdes_branch_test_context() -> tuple[Mock, Mock]: execution_state: Mock = Mock() execution_state.create_checkpoint = Mock() child_context: Mock = Mock() + child_context.is_replaying.return_value = False child_context.state.wrap_user_function = lambda func, *a, **k: func executor_context: Mock = Mock() executor_context._create_step_id_for_logical_step = lambda *args: "1" @@ -1587,6 +1644,37 @@ def execute_item(self, child_context, executable): ) +def test_replay_flat_branch_nondeterminism_escapes_batch(): + """FLAT replay must not convert nondeterminism into a failed batch item.""" + executor = _RecordingExecutor( + executables=[Executable(0, lambda: "x")], + max_concurrency=1, + completion_config=CompletionConfig(tolerated_failure_count=1), + sub_type_top="TOP", + sub_type_iteration="ITER", + name_prefix="test_", + serdes=None, + nesting_type=NestingType.FLAT, + operation_id_namespace=_StubNamespace(), + ) + executor._execute_item_in_child_context = Mock( + side_effect=NonDeterministicExecutionError("branch history drift") + ) + + checkpoint: Mock = Mock() + checkpoint.operation = None + checkpoint.is_succeeded.return_value = False + checkpoint.is_failed.return_value = False + execution_state: Mock = Mock() + execution_state.get_checkpoint_result.return_value = checkpoint + executor_context: Mock = Mock() + + with pytest.raises(NonDeterministicExecutionError, match="branch history drift"): + executor._replay_terminal_item( + execution_state, executor_context, Executable(0, lambda: "x") + ) + + def test_create_result_with_suspended_executable(): """Test with suspended executable using public execute method.""" @@ -1839,6 +1927,7 @@ def create_step_id(index): def create_child_context(operation_id, *, is_virtual=False): child_ctx = Mock() + child_ctx.is_replaying.return_value = False child_ctx.state = execution_state return child_ctx @@ -3063,9 +3152,13 @@ def _make_executor_mocks(): executor_context = Mock() executor_context._create_step_id_for_logical_step = lambda idx: f"step_{idx}" executor_context._parent_id = "parent" # noqa: SLF001 - executor_context.create_child_context = lambda op_id, *, is_virtual=False: Mock( - state=execution_state - ) + + def create_child_context(op_id, *, is_virtual=False): + child_context = Mock(state=execution_state) + child_context.is_replaying.return_value = False + return child_context + + executor_context.create_child_context = create_child_context return execution_state, executor_context @@ -3343,7 +3436,9 @@ def execute_item(self, child_context, executable): events: queue.Queue = queue.Queue() for executable in executables: - executor._branch_worker(executor_context, events, executable) # noqa: SLF001 + executor._branch_worker( # noqa: SLF001 + execution_state, executor_context, events, executable + ) collected = {} while not events.empty(): @@ -3720,6 +3815,120 @@ def checkpoint_dead(): executor.execute(execution_state, executor_context) +@pytest.mark.parametrize("nesting_type", [NestingType.NESTED, NestingType.FLAT]) +def test_nondeterminism_in_branch_escapes_batch(nesting_type): + """Branch completion policies cannot downgrade nondeterminism to failure.""" + + class TestExecutor(ConcurrentExecutor): + def execute_item(self, child_context, executable): + raise NonDeterministicExecutionError("branch history drift") + + executor = TestExecutor( + executables=[Executable(0, lambda: "x")], + max_concurrency=1, + completion_config=CompletionConfig(tolerated_failure_count=1), + sub_type_top="TOP", + sub_type_iteration=OperationSubType.PARALLEL_BRANCH, + name_prefix="test_", + serdes=None, + nesting_type=nesting_type, + operation_id_namespace=_StubNamespace(), + ) + execution_state, executor_context = _serdes_branch_test_context() + + with pytest.raises(NonDeterministicExecutionError, match="branch history drift"): + executor.execute(execution_state, executor_context) + + +def test_late_nondeterminism_blocks_early_completion_checkpoint(): + """A fatal branch race is retained until the parent terminal checkpoint.""" + barrier = threading.Barrier(2, timeout=5.0) + release_fatal = threading.Event() + fatal_recorded = threading.Event() + + def fast_branch() -> str: + barrier.wait() + return "fast" + + def late_fatal_branch() -> str: + barrier.wait() + assert release_fatal.wait(timeout=5.0) + raise NonDeterministicExecutionError("late branch history drift") + + class LateFatalExecutor(_RecordingExecutor): + def _create_result( + self, completion_reason: CompletionReason | None = None + ) -> BatchResult: + # execute() has already made its early-completion decision and + # drained the event queue. Let the straggler report afterward. + release_fatal.set() + assert fatal_recorded.wait(timeout=5.0) + return super()._create_result(completion_reason) + + parent_id = "parallel-op" + state = ExecutionState( + durable_execution_arn="arn:test:execution/exec1", + initial_checkpoint_token="token", # noqa: S106 + operations={}, + service_client=Mock(spec=DurableServiceClient), + plugin_executor=PluginExecutor(plugins=None), + ) + executor_context = DurableContext( + state=state, + execution_context=ExecutionContext( + durable_execution_arn=state.durable_execution_arn + ), + parent_id=parent_id, + ) + executor = LateFatalExecutor( + executables=[ + Executable(0, fast_branch), + Executable(1, late_fatal_branch), + ], + max_concurrency=2, + completion_config=CompletionConfig(min_successful=1), + sub_type_top=OperationSubType.PARALLEL, + sub_type_iteration=OperationSubType.PARALLEL_BRANCH, + name_prefix="parallel-branch-", + serdes=None, + nesting_type=NestingType.FLAT, + operation_id_namespace=_StubNamespace(), + ) + original_record = state.record_branch_fatal_error + + def record_fatal(parent_operation_id: str, error: BaseException) -> bool: + accepted = original_record(parent_operation_id, error) + fatal_recorded.set() + return accepted + + try: + with ( + patch.object( + state, + "record_branch_fatal_error", + side_effect=record_fatal, + ), + pytest.raises( + NonDeterministicExecutionError, + match="late branch history drift", + ), + ): + child_handler( + lambda: executor.execute(state, executor_context), + state, + OperationIdentifier( + operation_id=parent_id, + sub_type=OperationSubType.PARALLEL, + name="parallel", + ), + ChildConfig(sub_type=OperationSubType.PARALLEL), + ) + finally: + state.close() + + state._service_client.checkpoint.assert_not_called() # noqa: SLF001 + + def test_unlimited_concurrency_starts_all_items(): """max_concurrency=None keeps the previous start-everything behavior.""" barrier = threading.Barrier(3, timeout=5.0) 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 c5b8ceeb..99d76974 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -32,6 +32,7 @@ CallbackTimeoutError, ChildContextError, InvokeError, + NonDeterministicExecutionError, StepError, SuspendExecution, ValidationError, @@ -44,6 +45,7 @@ OperationStatus, OperationSubType, OperationType, + StepDetails, ) from aws_durable_execution_sdk_python.plugin import ( DurableInstrumentationPlugin, @@ -2380,6 +2382,36 @@ def test_should_propagate_outer_parent_id_when_virtual_is_nested_in_virtual(): assert inner_branch._create_step_id_for_logical_step(1) == expected +def test_flat_branch_rejects_nested_inner_checkpoint_parent(): + """Changing NESTED to FLAT keeps the inner id but changes its parent.""" + branch_id = "branch-op" + inner_id = hashlib.blake2b(f"{branch_id}-1".encode()).hexdigest()[:64] + checkpoint = Operation( + operation_id=inner_id, + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + parent_id=branch_id, + sub_type=OperationSubType.STEP, + name="inner-step", + step_details=StepDetails(result=json.dumps("cached")), + ) + state = _replay_state({inner_id: checkpoint}) + executor_context = DurableContext( + state=state, + execution_context=ExecutionContext( + durable_execution_arn=state.durable_execution_arn + ), + parent_id="parallel-op", + replay_status=ReplayStatus.REPLAY, + ) + flat_branch = executor_context.create_child_context(branch_id, is_virtual=True) + + with pytest.raises(NonDeterministicExecutionError, match="parent_id"): + flat_branch.step(lambda _ctx: "must-not-run", name="inner-step") + + state.close() + + # endregion Virtual-context identity tests @@ -2871,6 +2903,64 @@ def test_replay_aware_does_not_emit_replay_hook_when_not_replaying(): assert emitted == [] +@pytest.mark.parametrize( + ("checkpoint_status", "updated"), + [ + (OperationStatus.STARTED, False), + (OperationStatus.SUCCEEDED, True), + ], +) +def test_operation_identity_is_validated_before_replay_hooks( + checkpoint_status: OperationStatus, + updated: bool, +): + """Mismatched history fails before replay/update plugin hooks are dispatched.""" + captured: list[str] = [] + + class _CapturingPlugin(DurableInstrumentationPlugin): + def on_operation_start(self, info): + captured.append(f"start:{info.operation_id}") + + def on_operation_end(self, info): + captured.append(f"end:{info.operation_id}") + + plugin_executor = PluginExecutor(plugins=[_CapturingPlugin()]) + step_body_calls: list[bool] = [] + with plugin_executor.run(): + state = ExecutionState( + durable_execution_arn="arn", + initial_checkpoint_token="token", # noqa: S106 + operations={}, + service_client=Mock(), + plugin_executor=plugin_executor, + updated_operation_ids=[], + ) + ctx = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + next_id = ctx._peek_next_operation_id() # noqa: SLF001 + state._operations[next_id] = Operation( # noqa: SLF001 + operation_id=next_id, + operation_type=OperationType.WAIT, + status=checkpoint_status, + sub_type=OperationSubType.WAIT, + name="stale-wait", + ) + if updated: + state._updated_operation_ids.add(next_id) # noqa: SLF001 + + with pytest.raises(NonDeterministicExecutionError): + ctx.step( + lambda _step_context: step_body_calls.append(True), + name="current-step", + ) + + assert captured == [] + assert step_body_calls == [] + + def test_replay_aware_emits_update_hook_for_operation_updated_since_last_invocation(): """Updated terminal operations emit operation_end, not replay start+end.""" captured: list[tuple[str, str, bool, OperationStatus]] = [] diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py index 5c6eafb3..81c36306 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/checkpoint_response_int_test.py @@ -65,6 +65,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) operations.append(op) @@ -369,6 +371,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, callback_details=CallbackDetails( callback_id=f"cb-{update.operation_id[:8]}" ), @@ -379,6 +383,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) operations.append(op) @@ -589,6 +595,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) operations.append(op) diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py index 83f7c146..9768ad85 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py @@ -180,8 +180,9 @@ def handler(event, context: DurableContext) -> dict[str, Any]: { "Id": step_id, "Type": "STEP", + "SubType": "Step", + "Name": "process_order", "Status": "SUCCEEDED", - "ParentId": "execution-1", "StepDetails": {"Result": step_payload}, } ] @@ -250,8 +251,9 @@ def handler(event, context: DurableContext) -> dict[str, Any]: { "Id": child_id, "Type": "CONTEXT", + "SubType": "RunInChildContext", + "Name": "process_order", "Status": "SUCCEEDED", - "ParentId": "execution-1", "ContextDetails": {"Result": child_payload}, } ] @@ -325,8 +327,9 @@ def handler(event, context: DurableContext) -> dict[str, Any]: { "Id": wfc_id, "Type": "STEP", + "SubType": "WaitForCondition", + "Name": "process_order", "Status": "SUCCEEDED", - "ParentId": "execution-1", "StepDetails": {"Result": wfc_payload}, } ] @@ -426,8 +429,9 @@ def handler(event, context: DurableContext) -> dict[str, Any]: { "Id": child_id, "Type": "CONTEXT", + "SubType": "RunInChildContext", + "Name": "process_order", "Status": "SUCCEEDED", - "ParentId": "execution-1", "ContextDetails": {"Result": "", "ReplayChildren": True}, } ] 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 f3ada5e3..292a1707 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 @@ -156,7 +156,7 @@ def my_handler(event, context: DurableContext) -> str: "Type": "STEP", "Status": "FAILED", "SubType": "Step", - "ParentId": "execution-1", + "Name": "charge", # A failed step records the raw escaping error type. "StepDetails": { "Error": { @@ -244,7 +244,7 @@ def my_handler(event, context: DurableContext) -> str: "Type": "CONTEXT", "Status": "FAILED", "SubType": "WaitForCallback", - "ParentId": "execution-1", + "Name": "await-external", "ContextDetails": { "Error": { "ErrorType": checkpointed_type, @@ -306,7 +306,7 @@ def my_handler(event, context: DurableContext) -> str: "Type": "CONTEXT", "Status": "FAILED", "SubType": "WaitForCallback", - "ParentId": "execution-1", + "Name": "await-external", "ContextDetails": { "Error": { "ErrorType": "aws_durable_execution_sdk_python.exceptions.StepError", diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/execution_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/execution_int_test.py index ed774632..40ddcb48 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/execution_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/execution_int_test.py @@ -64,6 +64,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, # New operations start as STARTED parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) operations.append(op) @@ -540,8 +542,11 @@ def mock_checkpoint( operations = [ Operation( operation_id=update.operation_id, - operation_type=OperationType.CALLBACK, + operation_type=update.operation_type, status=OperationStatus.STARTED, + parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, callback_details=CallbackDetails( callback_id=f"callback-{update.operation_id[:8]}" ), diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/filesystem_serdes_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/filesystem_serdes_int_test.py index 08f1fa41..12c2efd8 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/filesystem_serdes_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/filesystem_serdes_int_test.py @@ -229,8 +229,9 @@ def mock_checkpoint( { "Id": step_id, "Type": "STEP", + "SubType": "Step", + "Name": "process_order", "Status": "SUCCEEDED", - "ParentId": "execution-1", "StepDetails": {"Result": envelope}, } ] diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_operations_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_operations_int_test.py index 26f1b77a..3cad3412 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_operations_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_operations_int_test.py @@ -96,6 +96,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) ) return CheckpointOutput( @@ -187,6 +189,7 @@ def replayed_handler(event: Any, context: DurableContext) -> str: # noqa: ARG00 completed_wait = { "Id": wait_id, "Type": OperationType.WAIT.value, + "SubType": "Wait", "Status": OperationStatus.SUCCEEDED.value, } diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_payload_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_payload_int_test.py index 4055b8f1..6b934423 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_payload_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_invocation_payload_int_test.py @@ -98,6 +98,8 @@ def mock_checkpoint( operation_type=update.operation_type, status=OperationStatus.STARTED, parent_id=update.parent_id, + name=update.name, + sub_type=update.sub_type, ) ) return CheckpointOutput( @@ -182,6 +184,7 @@ def my_handler(event: Any, context: DurableContext) -> str: completed_wait = { "Id": next(operation_id_sequence()), "Type": OperationType.WAIT.value, + "SubType": "Wait", "Status": OperationStatus.SUCCEEDED.value, } diff --git a/packages/aws-durable-execution-sdk-python/tests/execution_test.py b/packages/aws-durable-execution-sdk-python/tests/execution_test.py index ee04ce30..3e6d3aca 100644 --- a/packages/aws-durable-execution-sdk-python/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/execution_test.py @@ -4,13 +4,20 @@ import json import time import warnings +from collections.abc import Sequence from copy import deepcopy from typing import Any from unittest.mock import Mock, patch import pytest -from aws_durable_execution_sdk_python.config import StepConfig, StepSemantics +from aws_durable_execution_sdk_python.config import ( + MapConfig, + NestingType, + ParallelConfig, + StepConfig, + StepSemantics, +) from aws_durable_execution_sdk_python.context import DurableContext from aws_durable_execution_sdk_python.exceptions import ( BotoClientError, @@ -29,6 +36,7 @@ InvocationStatus, durable_execution, ) +from aws_durable_execution_sdk_python.identifier import OperationIdNamespace # LambdaContext no longer needed - using duck typing from aws_durable_execution_sdk_python.lambda_service import ( @@ -43,6 +51,7 @@ Operation, OperationAction, OperationStatus, + OperationSubType, OperationType, OperationUpdate, StateOutput, @@ -2760,6 +2769,356 @@ def test_handler(event: Any, context: DurableContext) -> dict: ) +@pytest.mark.parametrize( + ("checkpoint_type", "checkpoint_sub_type", "checkpoint_name", "mismatch"), + [ + ( + OperationType.WAIT, + OperationSubType.WAIT, + "current-name", + "type", + ), + ( + OperationType.STEP, + OperationSubType.WAIT_FOR_CONDITION, + "current-name", + "subtype", + ), + ( + OperationType.STEP, + OperationSubType.STEP, + "checkpoint-name", + "name", + ), + ], +) +def test_durable_execution_fails_replay_operation_identity_mismatch( + checkpoint_type: OperationType, + checkpoint_sub_type: OperationSubType, + checkpoint_name: str, + mismatch: str, +): + """Mismatched replay history must fail instead of skipping current work.""" + mock_client = Mock(spec=DurableServiceClient) + operation_id = OperationIdNamespace().create_id_for_step(1) + execution_operation = Operation( + operation_id="exec1", + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + execution_details=ExecutionDetails(input_payload="{}"), + ) + checkpoint_operation = Operation( + operation_id=operation_id, + operation_type=checkpoint_type, + status=OperationStatus.SUCCEEDED, + sub_type=checkpoint_sub_type, + name=checkpoint_name, + step_details=StepDetails(result=json.dumps("checkpoint-result")), + ) + invocation_input = DurableExecutionInvocationInputWithClient( + durable_execution_arn="arn:test:execution/exec1", + checkpoint_token="token123", # noqa: S106 + initial_execution_state=InitialExecutionState( + operations=[execution_operation, checkpoint_operation], + next_marker="", + ), + service_client=mock_client, + ) + step_body_calls: list[bool] = [] + + def step_body(_step_context) -> str: + step_body_calls.append(True) + return "executed" + + @durable_execution + def test_handler(event: Any, context: DurableContext) -> str: + return context.step(step_body, name="current-name") + + result = test_handler(invocation_input, _make_lambda_context()) + + assert result["Status"] == InvocationStatus.FAILED.value + assert ( + result["Error"]["ErrorType"] + == "aws_durable_execution_sdk_python.exceptions.NonDeterministicExecutionError" + ) + assert mismatch in result["Error"]["ErrorMessage"] + assert step_body_calls == [] + mock_client.checkpoint.assert_not_called() + + +def test_durable_execution_preserves_nested_nondeterminism_error(): + """A nested mismatch fails directly without checkpointing child failure.""" + mock_client = Mock(spec=DurableServiceClient) + outer_id = OperationIdNamespace().create_id_for_step(1) + inner_id = OperationIdNamespace(outer_id).create_id_for_step(1) + execution_operation = Operation( + operation_id="exec1", + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + execution_details=ExecutionDetails(input_payload="{}"), + ) + outer_operation = Operation( + operation_id=outer_id, + operation_type=OperationType.CONTEXT, + status=OperationStatus.STARTED, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + name="outer", + ) + mismatched_inner_operation = Operation( + operation_id=inner_id, + operation_type=OperationType.WAIT, + status=OperationStatus.SUCCEEDED, + parent_id=outer_id, + sub_type=OperationSubType.WAIT, + name="inner-step", + ) + invocation_input = DurableExecutionInvocationInputWithClient( + durable_execution_arn="arn:test:execution/exec1", + checkpoint_token="token123", # noqa: S106 + initial_execution_state=InitialExecutionState( + operations=[ + execution_operation, + outer_operation, + mismatched_inner_operation, + ], + next_marker="", + ), + service_client=mock_client, + ) + step_body_calls: list[bool] = [] + + def step_body(_step_context) -> str: + step_body_calls.append(True) + return "executed" + + @durable_execution + def test_handler(event: Any, context: DurableContext) -> str: + return context.run_in_child_context( + lambda child: child.step(step_body, name="inner-step"), + name="outer", + ) + + result = test_handler(invocation_input, _make_lambda_context()) + + assert result["Status"] == InvocationStatus.FAILED.value + assert ( + result["Error"]["ErrorType"] + == "aws_durable_execution_sdk_python.exceptions.NonDeterministicExecutionError" + ) + assert step_body_calls == [] + mock_client.checkpoint.assert_not_called() + + +@pytest.mark.parametrize("operation_kind", ["map", "parallel"]) +@pytest.mark.parametrize("parent_replay_children", [False, True]) +def test_durable_execution_rejects_nested_branch_checkpoint_in_flat_replay( + operation_kind: str, + parent_replay_children: bool, +): + """Real STARTED and ReplayChildren replays reject NESTED-to-FLAT drift.""" + mock_client = Mock(spec=DurableServiceClient) + parent_id = OperationIdNamespace().create_id_for_step(1) + branch_id = OperationIdNamespace(parent_id).create_id_for_step(0) + is_map = operation_kind == "map" + parent_sub_type = OperationSubType.MAP if is_map else OperationSubType.PARALLEL + branch_sub_type = ( + OperationSubType.MAP_ITERATION if is_map else OperationSubType.PARALLEL_BRANCH + ) + branch_name = "map-item-0" if is_map else "parallel-branch-0" + execution_operation = Operation( + operation_id="exec1", + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + execution_details=ExecutionDetails(input_payload="{}"), + ) + parent_operation = Operation( + operation_id=parent_id, + operation_type=OperationType.CONTEXT, + status=( + OperationStatus.SUCCEEDED + if parent_replay_children + else OperationStatus.STARTED + ), + sub_type=parent_sub_type, + name="batch", + context_details=( + ContextDetails( + replay_children=True, + result=json.dumps( + { + "totalCount": 1, + "completionReason": "ALL_COMPLETED", + "startedIndexes": [], + } + ), + ) + if parent_replay_children + else None + ), + ) + nested_branch_operation = Operation( + operation_id=branch_id, + operation_type=OperationType.CONTEXT, + status=OperationStatus.SUCCEEDED, + parent_id=parent_id, + sub_type=branch_sub_type, + name=branch_name, + context_details=ContextDetails(result=json.dumps("cached")), + ) + invocation_input = DurableExecutionInvocationInputWithClient( + durable_execution_arn="arn:test:execution/exec1", + checkpoint_token="token123", # noqa: S106 + initial_execution_state=InitialExecutionState( + operations=[ + execution_operation, + parent_operation, + nested_branch_operation, + ], + next_marker="", + ), + service_client=mock_client, + ) + branch_body_calls: list[bool] = [] + + def map_body( + _child: DurableContext, + _item: int, + _index: int, + _items: Sequence[int], + ) -> str: + branch_body_calls.append(True) + return "executed" + + def parallel_body(_child: DurableContext) -> str: + branch_body_calls.append(True) + return "executed" + + @durable_execution + def test_handler(event: Any, context: DurableContext) -> Any: + if is_map: + return context.map( + [1], + map_body, + name="batch", + config=MapConfig(nesting_type=NestingType.FLAT), + ) + return context.parallel( + [parallel_body], + name="batch", + config=ParallelConfig(nesting_type=NestingType.FLAT), + ) + + result = test_handler(invocation_input, _make_lambda_context()) + + assert result["Status"] == InvocationStatus.FAILED.value + assert ( + result["Error"]["ErrorType"] + == "aws_durable_execution_sdk_python.exceptions.NonDeterministicExecutionError" + ) + assert "nesting is FLAT" in result["Error"]["ErrorMessage"] + assert branch_body_calls == [] + mock_client.checkpoint.assert_not_called() + + +@pytest.mark.parametrize("operation_kind", ["map", "parallel"]) +def test_durable_execution_rejects_flat_terminal_branch_in_nested_replay( + operation_kind: str, +): + """ReplayChildren reconstruction rejects FLAT-to-NESTED branch drift.""" + mock_client = Mock(spec=DurableServiceClient) + parent_id = OperationIdNamespace().create_id_for_step(1) + branch_id = OperationIdNamespace(parent_id).create_id_for_step(0) + inner_step_id = OperationIdNamespace(branch_id).create_id_for_step(1) + is_map = operation_kind == "map" + parent_sub_type = OperationSubType.MAP if is_map else OperationSubType.PARALLEL + execution_operation = Operation( + operation_id="exec1", + operation_type=OperationType.EXECUTION, + status=OperationStatus.STARTED, + execution_details=ExecutionDetails(input_payload="{}"), + ) + parent_operation = Operation( + operation_id=parent_id, + operation_type=OperationType.CONTEXT, + status=OperationStatus.SUCCEEDED, + sub_type=parent_sub_type, + name="batch", + context_details=ContextDetails( + replay_children=True, + result=json.dumps( + { + "totalCount": 1, + "completionReason": "ALL_COMPLETED", + "startedIndexes": [], + } + ), + ), + ) + flat_inner_step = Operation( + operation_id=inner_step_id, + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + parent_id=parent_id, + sub_type=OperationSubType.STEP, + name="inner-step", + step_details=StepDetails(result=json.dumps("cached")), + ) + invocation_input = DurableExecutionInvocationInputWithClient( + durable_execution_arn="arn:test:execution/exec1", + checkpoint_token="token123", # noqa: S106 + initial_execution_state=InitialExecutionState( + operations=[ + execution_operation, + parent_operation, + flat_inner_step, + ], + next_marker="", + ), + service_client=mock_client, + ) + branch_body_calls: list[bool] = [] + + def branch_result(child: DurableContext) -> str: + branch_body_calls.append(True) + return child.step(lambda _step_context: "executed", name="inner-step") + + def map_body( + child: DurableContext, + _item: int, + _index: int, + _items: Sequence[int], + ) -> str: + return branch_result(child) + + @durable_execution + def test_handler(event: Any, context: DurableContext) -> Any: + if is_map: + return context.map( + [1], + map_body, + name="batch", + config=MapConfig(nesting_type=NestingType.NESTED), + ) + return context.parallel( + [branch_result], + name="batch", + config=ParallelConfig(nesting_type=NestingType.NESTED), + ) + + result = test_handler(invocation_input, _make_lambda_context()) + + assert result["Status"] == InvocationStatus.FAILED.value + assert ( + result["Error"]["ErrorType"] + == "aws_durable_execution_sdk_python.exceptions.NonDeterministicExecutionError" + ) + assert ( + "terminal NESTED branch context checkpoint" in result["Error"]["ErrorMessage"] + ) + assert branch_body_calls == [] + mock_client.checkpoint.assert_not_called() + + def test_durable_execution_non_retryable_invocation_error_returns_failed(): """Test that non-retryable InvocationError returns FAILED instead of retrying.""" mock_client = Mock(spec=DurableServiceClient) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/base_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/base_test.py index 4b208187..fa378c22 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/base_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/base_test.py @@ -2,19 +2,26 @@ from __future__ import annotations +from unittest.mock import Mock + import pytest -from aws_durable_execution_sdk_python.exceptions import InvalidStateError +from aws_durable_execution_sdk_python.exceptions import ( + InvalidStateError, + NonDeterministicExecutionError, +) +from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( Operation, OperationStatus, + OperationSubType, OperationType, ) from aws_durable_execution_sdk_python.operation.base import ( CheckResult, OperationExecutor, ) -from aws_durable_execution_sdk_python.state import CheckpointedResult +from aws_durable_execution_sdk_python.state import CheckpointedResult, ExecutionState # Test fixtures and helpers @@ -52,6 +59,30 @@ def create_mock_checkpoint(status: OperationStatus) -> CheckpointedResult: return CheckpointedResult.create_from_operation(operation) +def test_get_checkpoint_result_validates_operation_identity(): + """The base helper validates identity before returning checkpoint data.""" + executor = ConcreteOperationExecutor() + executor.state = Mock(spec=ExecutionState) + executor.operation_identifier = OperationIdentifier( + "test_op", OperationSubType.STEP, name="current-step" + ) + checkpoint = Operation( + operation_id="test_op", + operation_type=OperationType.WAIT, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.WAIT, + name="current-step", + ) + executor.state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_from_operation(checkpoint) + ) + + with pytest.raises(NonDeterministicExecutionError, match="type"): + executor._get_checkpoint_result() # noqa: SLF001 + + executor.state.get_checkpoint_result.assert_called_once_with("test_op") + + # Tests for CheckResult factory methods diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/callback_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/callback_test.py index a5844d35..92191299 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/callback_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/callback_test.py @@ -61,6 +61,8 @@ def test_create_callback_handler_new_operation_with_config(): operation = Operation( operation_id="callback1", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, + name="test_callback", status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -107,6 +109,7 @@ def test_create_callback_handler_new_operation_without_config(): operation = Operation( operation_id="callback2", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -145,6 +148,7 @@ def test_create_callback_handler_existing_started_operation(): operation = Operation( operation_id="callback3", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -173,6 +177,7 @@ def test_create_callback_handler_existing_failed_operation(): failed_op = Operation( operation_id="callback4", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=CallbackDetails(callback_id="failed_cb4"), ) @@ -198,6 +203,7 @@ def test_create_callback_handler_existing_started_missing_callback_details(): operation = Operation( operation_id="callback5", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=None, ) @@ -220,6 +226,7 @@ def test_create_callback_handler_new_operation_missing_callback_details_after_ch operation = Operation( operation_id="callback6", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=None, ) @@ -245,6 +252,7 @@ def test_create_callback_handler_existing_timed_out_operation(): operation = Operation( operation_id="callback_timed_out", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.TIMED_OUT, callback_details=callback_details, ) @@ -269,6 +277,7 @@ def test_create_callback_handler_existing_timed_out_missing_callback_details(): operation = Operation( operation_id="callback_timed_out_no_details", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.TIMED_OUT, callback_details=None, ) @@ -483,6 +492,7 @@ def test_create_callback_handler_existing_succeeded_operation(): operation = Operation( operation_id="callback_succeeded", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=callback_details, ) @@ -507,6 +517,7 @@ def test_create_callback_handler_existing_succeeded_missing_callback_details(): operation = Operation( operation_id="callback_succeeded_no_details", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=None, ) @@ -530,6 +541,7 @@ def test_create_callback_handler_config_with_zero_timeouts(): operation = Operation( operation_id="callback_zero", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -574,6 +586,7 @@ def test_create_callback_handler_config_with_large_timeouts(): operation = Operation( operation_id="callback_large", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -619,6 +632,7 @@ def test_create_callback_handler_empty_operation_id(): operation = Operation( operation_id="", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -810,6 +824,7 @@ def test_callback_lifecycle_complete_flow(): operation = Operation( operation_id="lifecycle_callback", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -863,6 +878,7 @@ def test_callback_retry_scenario(): operation = Operation( operation_id="retry_callback", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -900,6 +916,7 @@ def test_callback_timeout_configuration(): operation = Operation( operation_id=f"timeout_callback_{timeout_seconds}", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -932,6 +949,7 @@ def test_callback_error_propagation(): failed_op = Operation( operation_id="error_callback", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=CallbackDetails(callback_id="failed_cb"), ) @@ -996,12 +1014,14 @@ def test_callback_state_consistency(): started_operation = Operation( operation_id="consistent_callback", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) succeeded_operation = Operation( operation_id="consistent_callback", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=callback_details, ) @@ -1074,6 +1094,7 @@ def test_callback_operation_update_creation(mock_operation_update): operation = Operation( operation_id="update_test", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -1117,6 +1138,7 @@ def test_callback_immediate_response_get_checkpoint_result_called_twice(): started_op = Operation( operation_id="callback_immediate_1", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -1147,6 +1169,7 @@ def test_callback_immediate_response_create_checkpoint_with_is_sync_true(): started_op = Operation( operation_id="callback_immediate_2", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -1183,6 +1206,7 @@ def test_callback_immediate_response_immediate_success(): succeeded_op = Operation( operation_id="callback_immediate_3", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=callback_details, ) @@ -1220,6 +1244,7 @@ def test_callback_immediate_response_immediate_failure_deferred(): failed_op = Operation( operation_id="callback_immediate_4", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=callback_details, ) @@ -1263,6 +1288,7 @@ def test_callback_result_raises_error_for_failed_callbacks(): failed_op = Operation( operation_id="callback_failed_result", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=callback_details, ) @@ -1300,6 +1326,7 @@ def test_callback_result_raises_error_for_timed_out_callbacks(): timed_out_op = Operation( operation_id="callback_timed_out_result", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.TIMED_OUT, callback_details=callback_details, ) @@ -1333,6 +1360,7 @@ def test_callback_immediate_response_no_immediate_response(): started_op = Operation( operation_id="callback_immediate_5", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.STARTED, callback_details=callback_details, ) @@ -1368,6 +1396,7 @@ def test_callback_immediate_response_already_completed(): succeeded_op = Operation( operation_id="callback_immediate_6", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=callback_details, ) @@ -1403,6 +1432,7 @@ def test_callback_immediate_response_already_failed(): failed_op = Operation( operation_id="callback_immediate_7", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=callback_details, ) @@ -1445,6 +1475,7 @@ def test_callback_deferred_error_handling_code_execution_between_create_and_resu failed_op = Operation( operation_id="callback_deferred_error", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.FAILED, callback_details=callback_details, ) @@ -1495,6 +1526,7 @@ def test_callback_immediate_response_with_config(): succeeded_op = Operation( operation_id="callback_with_config", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, status=OperationStatus.SUCCEEDED, callback_details=callback_details, ) @@ -1541,6 +1573,8 @@ def test_callback_returns_id_when_second_check_returns_started(): Operation( operation_id="callback-1", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, + name="test_callback", status=OperationStatus.STARTED, callback_details=CallbackDetails(callback_id="cb-123"), ) @@ -1575,6 +1609,8 @@ def test_callback_returns_id_when_second_check_returns_started_duplicate(): started_op = Operation( operation_id="callback-1", operation_type=OperationType.CALLBACK, + sub_type=OperationSubType.CALLBACK, + name="test_callback", status=OperationStatus.STARTED, callback_details=CallbackDetails(callback_id="cb-123"), ) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py index a5f66a5f..0ca91d69 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py @@ -427,8 +427,8 @@ def test_child_handler_invocation_error_reraised(): mock_state.emit_child_context_end_hook.assert_not_called() -def test_child_handler_execution_error_wrapped(): - """ExecutionError is wrapped as ChildContextError (replay-safe), not raised raw.""" +def test_child_handler_execution_error_reraised_without_fail_checkpoint(): + """ExecutionError escapes unchanged without mutating child history.""" mock_state = Mock(spec=ExecutionState) mock_state.durable_execution_arn = "test_arn" mock_result = Mock() @@ -441,7 +441,7 @@ def test_child_handler_execution_error_wrapped(): mock_callable = Mock(side_effect=test_error) mock_state.wrap_user_function.return_value = mock_callable - with pytest.raises(ChildContextError) as exc_info: + with pytest.raises(ExecutionError) as exc_info: child_handler( mock_callable, mock_state, @@ -451,18 +451,15 @@ def test_child_handler_execution_error_wrapped(): None, ) - # Not the raw ExecutionError; the original type survives on error_type. - assert not isinstance(exc_info.value, ExecutionError) - assert ( - exc_info.value.error_type - == "aws_durable_execution_sdk_python.exceptions.ExecutionError" - ) + assert exc_info.value is test_error - # Verify FAIL checkpoint was created - assert mock_state.create_checkpoint.call_count == 2 # start and fail - fail_call = mock_state.create_checkpoint.call_args_list[1] - fail_operation = fail_call[1]["operation_update"] - assert fail_operation.action is OperationAction.FAIL + # Only the async START checkpoint is written - no FAIL. + assert mock_state.create_checkpoint.call_count == 1 + actions = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL not in actions def test_child_handler_non_retryable_invocation_error_wrapped(): diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index 212d8528..6fca520b 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -51,6 +51,8 @@ def test_invoke_handler_already_succeeded(): operation = Operation( operation_id="invoke1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails(result=json.dumps("test_result")), ) @@ -79,6 +81,8 @@ def test_invoke_handler_already_succeeded_none_result(): operation = Operation( operation_id="invoke2", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails(result=None), ) @@ -106,6 +110,8 @@ def test_invoke_handler_already_succeeded_no_chained_invoke_details(): operation = Operation( operation_id="invoke3", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=None, ) @@ -139,6 +145,8 @@ def test_invoke_handler_already_terminated(kind: OperationStatus): operation = Operation( operation_id="invoke4", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=kind, chained_invoke_details=ChainedInvokeDetails(error=error), ) @@ -168,6 +176,8 @@ def test_invoke_handler_already_timed_out(): operation = Operation( operation_id="invoke5", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.TIMED_OUT, chained_invoke_details=ChainedInvokeDetails(error=error), ) @@ -195,6 +205,8 @@ def test_invoke_handler_already_started(status): operation = Operation( operation_id="invoke6", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=status, chained_invoke_details=ChainedInvokeDetails(), ) @@ -224,6 +236,8 @@ def test_invoke_handler_already_started_suspends(status): operation = Operation( operation_id="invoke7", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=status, chained_invoke_details=ChainedInvokeDetails(), ) @@ -255,6 +269,8 @@ def test_invoke_handler_new_operation(): started_op = Operation( operation_id="invoke8", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -296,6 +312,8 @@ def test_invoke_handler_no_config(): started_op = Operation( operation_id="invoke_test", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -328,6 +346,8 @@ def test_invoke_handler_custom_serdes(): operation = Operation( operation_id="invoke12", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}', @@ -363,6 +383,8 @@ def test_invoke_handler_custom_serdes_new_operation(): started_op = Operation( operation_id="invoke_test", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -399,6 +421,8 @@ def test_invoke_handler_with_operation_name(status: OperationStatus): operation = Operation( operation_id="invoke14", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="named_invoke", status=status, chained_invoke_details=ChainedInvokeDetails(), ) @@ -426,6 +450,8 @@ def test_invoke_handler_without_operation_name(status: OperationStatus): operation = Operation( operation_id="invoke15", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=status, chained_invoke_details=ChainedInvokeDetails(), ) @@ -453,6 +479,8 @@ def test_invoke_handler_with_none_payload(): started_op = Operation( operation_id="invoke_test", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -483,6 +511,8 @@ def test_invoke_handler_already_succeeded_with_none_payload(): operation = Operation( operation_id="invoke17", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails(result=json.dumps("test_result")), ) @@ -516,6 +546,8 @@ def test_invoke_handler_suspend_does_not_raise(mock_suspend): started_op = Operation( operation_id="invoke_test", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -550,6 +582,8 @@ def test_invoke_handler_with_tenant_id(): started_op = Operation( operation_id="invoke1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -585,6 +619,8 @@ def test_invoke_handler_without_tenant_id(): started_op = Operation( operation_id="invoke1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -620,6 +656,8 @@ def test_invoke_handler_default_config_no_tenant_id(): started_op = Operation( operation_id="invoke1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -653,6 +691,8 @@ def test_invoke_handler_defaults_to_json_serdes(): started_op = Operation( operation_id="invoke1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -686,6 +726,8 @@ def test_invoke_handler_result_defaults_to_json_serdes(): operation = Operation( operation_id="invoke_result_json", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name=None, status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails(result=json.dumps(result_data)), ) @@ -723,6 +765,8 @@ def test_invoke_immediate_response_get_checkpoint_result_called_twice(): started_op = Operation( operation_id="invoke_immediate_1", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -756,6 +800,8 @@ def test_invoke_immediate_response_create_checkpoint_with_is_sync_true(): started_op = Operation( operation_id="invoke_immediate_2", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -795,6 +841,8 @@ def test_invoke_immediate_response_immediate_success(): succeeded_op = Operation( operation_id="invoke_immediate_3", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails( result=json.dumps("immediate_result") @@ -831,6 +879,8 @@ def test_invoke_immediate_response_immediate_success_with_none_result(): succeeded_op = Operation( operation_id="invoke_immediate_4", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails(result=None), ) @@ -873,6 +923,8 @@ def test_invoke_immediate_response_immediate_failure(status: OperationStatus): failed_op = Operation( operation_id="invoke_immediate_5", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=status, chained_invoke_details=ChainedInvokeDetails(error=error), ) @@ -913,6 +965,8 @@ def test_invoke_immediate_response_no_immediate_response(): started_op = Operation( operation_id="invoke_immediate_6", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) @@ -952,6 +1006,8 @@ def test_invoke_immediate_response_already_completed(): succeeded_op = Operation( operation_id="invoke_immediate_7", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails( result=json.dumps("existing_result") @@ -988,6 +1044,8 @@ def test_invoke_immediate_response_with_custom_serdes(): succeeded_op = Operation( operation_id="invoke_immediate_10", operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}' @@ -1031,7 +1089,9 @@ def test_invoke_suspends_when_second_check_returns_started(): CheckpointedResult.create_from_operation( Operation( operation_id="invoke-1", - operation_type=OperationType.STEP, + operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) ), @@ -1067,7 +1127,9 @@ def test_invoke_suspends_when_second_check_returns_started_duplicate(): not_found = CheckpointedResult.create_not_found() started_op = Operation( operation_id="invoke-1", - operation_type=OperationType.STEP, + operation_type=OperationType.CHAINED_INVOKE, + sub_type=OperationSubType.CHAINED_INVOKE, + name="test_invoke", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py index a0d53b8a..bba7b1da 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py @@ -70,6 +70,8 @@ def test_step_handler_already_succeeded(): operation = Operation( operation_id="step1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=json.dumps("test_result")), ) @@ -99,6 +101,8 @@ def test_step_handler_already_succeeded_none_result(): operation = Operation( operation_id="step2", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=None), ) @@ -130,6 +134,8 @@ def test_step_handler_already_failed(): operation = Operation( operation_id="step3", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.FAILED, step_details=StepDetails(error=error), ) @@ -158,6 +164,8 @@ def test_step_handler_started_at_most_once(): operation = Operation( operation_id="step4", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -188,6 +196,8 @@ def test_step_handler_started_at_least_once(): operation = Operation( operation_id="step5", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(error=error), ) @@ -267,6 +277,8 @@ def test_step_context_exposes_current_attempt( operation: Operation = Operation( operation_id="step-attempt", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=checkpointed_attempts), ) @@ -307,6 +319,8 @@ def test_step_handler_success_at_most_once(): started_op = Operation( operation_id="step7", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -511,6 +525,8 @@ def test_step_handler_retry_with_existing_attempts(): operation = Operation( operation_id="step12", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.PENDING, step_details=StepDetails( attempt=2, @@ -552,6 +568,8 @@ def test_step_handler_pending_without_existing_attempts(): operation = Operation( operation_id="step12", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.PENDING, step_details=StepDetails(attempt=2), ) @@ -593,6 +611,8 @@ def test_step_handler_retry_handler_no_exception(mock_retry_handler): started_op = Operation( operation_id="step13", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -660,6 +680,8 @@ def test_step_handler_custom_serdes_already_succeeded(): operation = Operation( operation_id="step1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}' @@ -738,6 +760,8 @@ def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: succeeded_op = Operation( operation_id="step_rt", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=checkpointed_payload), ) @@ -841,6 +865,8 @@ def test_step_handler_already_succeeded_empty_string_result(): operation = Operation( operation_id="step1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=""), ) @@ -876,6 +902,8 @@ def test_step_immediate_response_get_checkpoint_called_twice(): started_op = Operation( operation_id="step_immediate_1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -913,6 +941,8 @@ def test_step_immediate_response_create_checkpoint_sync_at_most_once(): started_op = Operation( operation_id="step_immediate_2", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -985,6 +1015,8 @@ def test_step_immediate_response_immediate_success(): started_op = Operation( operation_id="step_immediate_4", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -1025,6 +1057,8 @@ def test_step_immediate_response_immediate_failure(): started_op = Operation( operation_id="step_immediate_5", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -1075,6 +1109,8 @@ def test_step_immediate_response_no_immediate_response(): started_op = Operation( operation_id="step_immediate_6", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) @@ -1113,6 +1149,8 @@ def test_step_immediate_response_already_completed(): succeeded_op = Operation( operation_id="step_immediate_7", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=json.dumps("already_completed_result")), ) @@ -1154,6 +1192,8 @@ def test_step_executes_function_when_second_check_returns_started(): started_op = Operation( operation_id="step-1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=1), ) @@ -1194,6 +1234,8 @@ def test_step_creates_start_checkpoint_when_status_is_ready(): ready_op = Operation( operation_id="step_ready_1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.READY, step_details=StepDetails(attempt=0), ) @@ -1203,6 +1245,8 @@ def test_step_creates_start_checkpoint_when_status_is_ready(): started_op = Operation( operation_id="step_ready_1", operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="test_step", status=OperationStatus.STARTED, step_details=StepDetails(attempt=0), ) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py index 541be016..29831acd 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py @@ -145,6 +145,8 @@ def test_wait_for_condition_already_succeeded(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=json.dumps(42)), ) @@ -185,6 +187,8 @@ def test_wait_for_condition_already_succeeded_none_result(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=None), ) @@ -224,6 +228,8 @@ def test_wait_for_condition_already_failed(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.FAILED, step_details=StepDetails( error=ErrorObject("Test error", "TestError", None, None) @@ -264,6 +270,8 @@ def test_wait_for_condition_retry_with_state(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=2), ) @@ -310,6 +318,8 @@ def test_wait_for_condition_retry_restores_none_state(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(None), attempt=2), ) @@ -356,6 +366,8 @@ def test_wait_for_condition_retry_without_state(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=None, attempt=2), ) @@ -397,6 +409,8 @@ def test_wait_for_condition_retry_invalid_json_state_fails(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result="invalid json", attempt=2), ) @@ -751,6 +765,8 @@ def test_wait_for_condition_operation_no_step_details(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=None, ) @@ -839,6 +855,8 @@ def test_wait_for_condition_attempt_number_passed_to_strategy(): operation: Operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=3), ) @@ -884,6 +902,8 @@ def test_wait_for_condition_context_exposes_current_attempt(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=3), ) @@ -973,6 +993,8 @@ def wait_strategy(state, attempt): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=1), ) @@ -995,6 +1017,8 @@ def wait_strategy(state, attempt): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=2), ) @@ -1017,6 +1041,8 @@ def wait_strategy(state, attempt): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.STARTED, step_details=StepDetails(result=json.dumps(10), attempt=3), ) @@ -1209,6 +1235,8 @@ def test_wait_for_condition_custom_serdes_already_succeeded(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}' @@ -1248,6 +1276,8 @@ def test_wait_for_condition_pending(): operation = Operation( operation_id="XXX", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.PENDING, step_details=StepDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}', @@ -1294,6 +1324,8 @@ def test_wait_for_condition_pending_without_next_attempt(): operation = Operation( operation_id="XXX", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.PENDING, step_details=StepDetails( result='{"key": "VALUE", "number": "84", "list": [1, 2, 3]}', @@ -1384,6 +1416,8 @@ def test_wait_for_condition_immediate_success_without_executing_check(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=json.dumps(42)), ) @@ -1426,6 +1460,8 @@ def test_wait_for_condition_immediate_failure_without_executing_check(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.FAILED, step_details=StepDetails( error=ErrorObject("Test error", "TestError", None, None) @@ -1470,6 +1506,8 @@ def test_wait_for_condition_pending_suspends_without_executing_check(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.PENDING, step_details=StepDetails( result=json.dumps(10), @@ -1566,6 +1604,8 @@ def test_wait_for_condition_already_completed_no_checkpoint_created(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=json.dumps(42)), ) @@ -1703,6 +1743,8 @@ def test_wait_for_condition_exhaustion_surfaces_on_replay(): operation = Operation( operation_id="op1", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.FAILED, step_details=StepDetails( error=ErrorObject( @@ -1854,6 +1896,8 @@ def check_func(_state, _context): succeeded_op = Operation( operation_id="wfc_rt", operation_type=OperationType.STEP, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="test_wait", status=OperationStatus.SUCCEEDED, step_details=StepDetails(result=checkpointed_payload), ) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/wait_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/wait_test.py index 07274826..cb572bdc 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/wait_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/wait_test.py @@ -382,6 +382,8 @@ def test_wait_suspends_when_second_check_returns_started(): Operation( operation_id="wait-1", operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + name="test_wait", status=OperationStatus.STARTED, ) ), @@ -416,6 +418,8 @@ def test_wait_suspends_when_second_check_returns_started_duplicate(): started_op = Operation( operation_id="wait-1", operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + name="test_wait", status=OperationStatus.STARTED, ) started = CheckpointedResult.create_from_operation(started_op) diff --git a/packages/aws-durable-execution-sdk-python/tests/state_test.py b/packages/aws-durable-execution-sdk-python/tests/state_test.py index bb5f841d..92b07f58 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -18,9 +18,11 @@ CheckpointError, DurableApiErrorCategory, GetExecutionStateError, + NonDeterministicExecutionError, OrphanedChildException, StepError, SuspendExecution, + TerminationReason, TimedSuspendExecution, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier @@ -614,6 +616,112 @@ def test_get_checkpoint_result_operation_not_found(): assert result.operation is None +@pytest.mark.parametrize( + ("checkpoint", "operation_identifier", "mismatch"), + [ + ( + Operation( + operation_id="op1", + operation_type=OperationType.WAIT, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.WAIT, + name="current-name", + ), + OperationIdentifier("op1", OperationSubType.STEP, name="current-name"), + "type", + ), + ( + Operation( + operation_id="op1", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.WAIT_FOR_CONDITION, + name="current-name", + ), + OperationIdentifier("op1", OperationSubType.STEP, name="current-name"), + "subtype", + ), + ( + Operation( + operation_id="op1", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.STEP, + name="checkpoint-name", + ), + OperationIdentifier("op1", OperationSubType.STEP, name="current-name"), + "name", + ), + ( + Operation( + operation_id="op1", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + parent_id="old-parent", + sub_type=OperationSubType.STEP, + name="current-name", + ), + OperationIdentifier( + "op1", + OperationSubType.STEP, + parent_id="current-parent", + name="current-name", + ), + "parent_id", + ), + ], +) +def test_operation_identifier_rejects_mismatched_checkpoint_identity( + checkpoint: Operation, + operation_identifier: OperationIdentifier, + mismatch: str, +): + """Replay checkpoints must match the current operation before use.""" + state = ExecutionState( + durable_execution_arn="test_arn", + initial_checkpoint_token="token123", # noqa: S106 + operations={"op1": checkpoint}, + service_client=Mock(spec=LambdaClient), + plugin_executor=PluginExecutor(plugins=None), + ) + + with pytest.raises(NonDeterministicExecutionError, match=mismatch) as error_info: + operation_identifier.validate_checkpoint( + state.get_checkpoint_result(operation_identifier.operation_id).operation + ) + + assert error_info.value.step_id == "op1" + assert ( + error_info.value.termination_reason + is TerminationReason.NON_DETERMINISTIC_EXECUTION + ) + + +def test_operation_identifier_normalizes_empty_name_to_wire_identity(): + """An empty emitted name is omitted on the wire and replays as None.""" + operation = Operation( + operation_id="op1", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.STEP, + name=None, + ) + state = ExecutionState( + durable_execution_arn="test_arn", + initial_checkpoint_token="token123", # noqa: S106 + operations={"op1": operation}, + service_client=Mock(spec=LambdaClient), + plugin_executor=PluginExecutor(plugins=None), + ) + + result = state.get_checkpoint_result("op1") + OperationIdentifier("op1", OperationSubType.STEP, name="").validate_checkpoint( + result.operation + ) + + assert result.operation is operation + + def test_create_checkpoint(): """Test create_checkpoint method enqueues operations asynchronously.""" mock_lambda_client = Mock(spec=LambdaClient)