From cdae7556d505f3421f99e48e1a8265d542fb26b3 Mon Sep 17 00:00:00 2001 From: vyrnsynx Date: Fri, 9 Oct 2026 14:05:52 +0800 Subject: [PATCH 1/6] fix(sdk): filter batch items on status alone succeeded() and get_results() also required a non-None result, so a branch that returned None counted towards success_count but was missing from both. failed() and get_errors() had the same check on the error. Select items by status only. When replay rebuilds a FAILED item from a child context checkpoint without an error, record the generic ChildContextError that replaying that child context raises, so every FAILED item still has an error. Closes #760 --- .../concurrency/executor.py | 22 ++++- .../concurrency/models.py | 27 ++---- .../tests/concurrency_test.py | 92 ++++++++++++++++++- 3 files changed, 114 insertions(+), 27 deletions(-) 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 2a39c4eff..0dfa6ea26 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,6 +31,7 @@ NestingType, ) from aws_durable_execution_sdk_python.exceptions import ( + ChildContextError, DurableOperationError, ExecutionError, InvalidStateError, @@ -88,6 +89,21 @@ def _branch_error_object(err: Exception) -> ErrorObject: return ErrorObject.from_exception(err) +def _failed_checkpoint_error_object(checkpoint: CheckpointedResult) -> ErrorObject: + """Return the error of a branch whose child context checkpoint FAILED. + + The checkpoint error is optional. Without one, record the generic + ChildContextError that replaying the child context raises, so every + FAILED item carries an error. + """ + if checkpoint.error is not None: + return checkpoint.error + try: + checkpoint.raise_operation_error(ChildContextError) + except ChildContextError as e: + return _branch_error_object(e) + + @dataclass class ResumeWave(Generic[CallableType, ResultType]): """Branches suspended until one time, and the refresh that resumes them. @@ -783,7 +799,9 @@ def _replay_terminal_item( return BatchItem(executable.index, BatchItemStatus.SUCCEEDED, result) if checkpoint.is_failed(): return BatchItem( - executable.index, BatchItemStatus.FAILED, error=checkpoint.error + executable.index, + BatchItemStatus.FAILED, + error=_failed_checkpoint_error_object(checkpoint), ) if self.nesting_type is NestingType.FLAT: try: @@ -836,7 +854,7 @@ def _replay_from_checkpoints( ) elif checkpoint.is_failed(): - error = checkpoint.error + error = _failed_checkpoint_error_object(checkpoint) status = BatchItemStatus.FAILED else: status = BatchItemStatus.STARTED diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index 3c657e570..c56c6d857 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -7,7 +7,7 @@ from collections import Counter from dataclasses import dataclass from enum import Enum -from typing import TYPE_CHECKING, Generic, TypeVar +from typing import TYPE_CHECKING, Generic, TypeVar, cast from aws_durable_execution_sdk_python.exceptions import ( ChildContextError, @@ -489,18 +489,10 @@ def to_dict(self) -> dict: } def succeeded(self) -> list[BatchItem[R]]: - return [ - item - for item in self.all - if item.status is BatchItemStatus.SUCCEEDED and item.result is not None - ] + return [item for item in self.all if item.status is BatchItemStatus.SUCCEEDED] def failed(self) -> list[BatchItem[R]]: - return [ - item - for item in self.all - if item.status is BatchItemStatus.FAILED and item.error is not None - ] + return [item for item in self.all if item.status is BatchItemStatus.FAILED] def started(self) -> list[BatchItem[R]]: return [item for item in self.all if item.status is BatchItemStatus.STARTED] @@ -533,18 +525,11 @@ def throw_if_error(self) -> None: raise BatchCompletionError(self.completion_reason) def get_results(self) -> list[R]: - return [ - item.result - for item in self.all - if item.status is BatchItemStatus.SUCCEEDED and item.result is not None - ] + # None is a valid result for a branch that returns nothing. + return [cast("R", item.result) for item in self.succeeded()] def get_errors(self) -> list[ErrorObject]: - return [ - item.error - for item in self.all - if item.status is BatchItemStatus.FAILED and item.error is not None - ] + return [cast("ErrorObject", item.error) for item in self.failed()] @property def success_count(self) -> int: 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 a93135663..b06f3146b 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -306,6 +306,23 @@ def test_batch_result_get_errors(): assert error2 in errors +def test_batch_result_keeps_succeeded_items_with_none_result(): + """A branch that returns None still counts as succeeded.""" + items = [ + BatchItem(0, BatchItemStatus.SUCCEEDED, None), + BatchItem(1, BatchItemStatus.SUCCEEDED, 10), + BatchItem( + 2, BatchItemStatus.FAILED, error=ErrorObject("msg", "Error", None, None) + ), + BatchItem(3, BatchItemStatus.SUCCEEDED, None), + ] + result = BatchResult(items, CompletionReason.ALL_COMPLETED) + + assert [item.index for item in result.succeeded()] == [0, 1, 3] + assert len(result.succeeded()) == result.success_count + assert result.get_results() == [None, 10, None] + + def test_batch_result_counts(): """Test BatchResult count properties.""" items = [ @@ -592,9 +609,9 @@ def test_func(): def test_batch_result_failed_with_none_error(): - """Test BatchResult failed method filters out None errors.""" + """Test BatchResult failed method selects FAILED items by status alone.""" items = [ - BatchItem(0, BatchItemStatus.FAILED, error=None), # Should be filtered out + BatchItem(0, BatchItemStatus.FAILED, error=None), BatchItem( 1, BatchItemStatus.FAILED, error=ErrorObject("msg", "Error", None, None) ), @@ -602,8 +619,7 @@ def test_batch_result_failed_with_none_error(): result = BatchResult(items, CompletionReason.ALL_COMPLETED) failed = result.failed() - assert len(failed) == 1 - assert failed[0].error is not None + assert len(failed) == result.failure_count == 2 def test_concurrent_executor_nesting_type_parameter(): @@ -1031,6 +1047,58 @@ def test_flat_replay_rejects_failed_nested_container_checkpoint(replay_path): executor_context.create_child_context.assert_not_called() +@pytest.mark.parametrize("replay_path", ["recorded-terminal", "fallback"]) +def test_replay_failed_checkpoint_without_error_records_child_context_error( + replay_path, +): + """A FAILED branch checkpoint may carry no error; the item still gets one.""" + executor = _RecordingExecutor( + executables=[Executable(0, lambda: "must-not-run")], + max_concurrency=1, + completion_config=CompletionConfig(tolerated_failure_count=1), + sub_type_top=OperationSubType.PARALLEL, + sub_type_iteration=OperationSubType.PARALLEL_BRANCH, + name_prefix="parallel-branch-", + serdes=None, + nesting_type=NestingType.NESTED, + operation_id_namespace=_StubNamespace(), + ) + executor_context = Mock() + executor_context._parent_id = "parallel-op" # noqa: SLF001 + execution_state = Mock() + execution_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_from_operation( + Operation( + operation_id="op_0", + operation_type=OperationType.CONTEXT, + status=OperationStatus.FAILED, + parent_id="parallel-op", + sub_type=OperationSubType.PARALLEL_BRANCH, + name="parallel-branch-0", + ) + ) + ) + + if replay_path == "recorded-terminal": + item = executor._replay_terminal_item( # noqa: SLF001 + execution_state, executor_context, executor.executables[0] + ) + result = BatchResult([item], CompletionReason.ALL_COMPLETED) + else: + result = executor._replay_from_checkpoints( # noqa: SLF001 + execution_state, executor_context + ) + + assert [item.index for item in result.failed()] == [0] + errors = result.get_errors() + assert len(errors) == 1 + assert errors[0].type == ( + "aws_durable_execution_sdk_python.exceptions.ChildContextError" + ) + with pytest.raises(ChildContextError, match="No ErrorObject exists"): + result.throw_if_error() + + @pytest.mark.parametrize( ("nesting_type", "checkpoint_name", "mismatch"), [ @@ -4003,6 +4071,22 @@ def suspend(): assert replayed.all[2].result == "replayed_2" +def test_execute_keeps_branches_that_return_none(): + """Branches returning None appear in succeeded() and get_results().""" + executables = [ + Executable(index, partial(lambda value: value, value)) + for index, value in enumerate([None, 10, None, 30]) + ] + executor = _make_executor(executables, max_concurrency=2) + execution_state, executor_context = _make_executor_mocks() + + result: BatchResult = executor.execute(execution_state, executor_context) + + assert result.success_count == 4 + assert len(result.succeeded()) == 4 + assert result.get_results() == [None, 10, None, 30] + + def test_replay_without_record_falls_back_to_checkpoint_derivation(): """No record: every executable is represented, old-style.""" executables = [Executable(i, lambda: "x") for i in range(3)] From 1bae9f75fb396558a15c048f90c68124e9e3d76c Mon Sep 17 00:00:00 2001 From: vyrnsynx <153433026+vyrnsynx@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:10:57 +0800 Subject: [PATCH 2/6] fix(sdk): normalize missing FAILED batch-item errors FAILED BatchItems constructed or deserialized without an ErrorObject now receive the same generic ChildContextError the executor records for a FAILED child-context checkpoint with no error. Extends the regression test to cover get_errors() after from_dict and ExtendedTypeSerDes. Addresses review feedback on #765. --- .../concurrency/models.py | 29 ++++++++- .../tests/concurrency_test.py | 60 ++++++++++++++++++- 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index c56c6d857..17c0e63d6 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -42,6 +42,26 @@ # region Result models +_MISSING_FAILED_ITEM_ERROR_MESSAGE = ( + "Unknown error. No ErrorObject exists on the Checkpoint Operation." +) + + +def _generic_failed_item_error() -> ErrorObject: + """ErrorObject for a FAILED batch item that carried no checkpoint error. + + Matches ``_failed_checkpoint_error_object`` when the child-context + checkpoint has no ErrorObject: the same generic ChildContextError + replaying that child would raise. + """ + err = ChildContextError(message=_MISSING_FAILED_ITEM_ERROR_MESSAGE) + return ErrorObject( + message=err.message, + type=err.error_type, + data=err.data, + stack_trace=err.stack_trace, + ) + class CompletionReason(Enum): ALL_COMPLETED = "ALL_COMPLETED" @@ -401,6 +421,10 @@ class BatchItem(Generic[R]): result: R | None = None error: ErrorObject | None = None + def __post_init__(self) -> None: + if self.status is BatchItemStatus.FAILED and self.error is None: + object.__setattr__(self, "error", _generic_failed_item_error()) + def to_dict(self) -> dict: return { "index": self.index, @@ -529,7 +553,10 @@ def get_results(self) -> list[R]: return [cast("R", item.result) for item in self.succeeded()] def get_errors(self) -> list[ErrorObject]: - return [cast("ErrorObject", item.error) for item in self.failed()] + return [ + item.error if item.error is not None else _generic_failed_item_error() + for item in self.failed() + ] @property def success_count(self) -> int: 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 6a9fbdf0a..3b1efe24f 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -61,6 +61,7 @@ SuspendExecution, TimedSuspendExecution, ) +from aws_durable_execution_sdk_python.serdes import ExtendedTypeSerDes from aws_durable_execution_sdk_python.lambda_service import ( DurableServiceClient, ErrorObject, @@ -613,7 +614,15 @@ def test_func(): def test_batch_result_failed_with_none_error(): - """Test BatchResult failed method selects FAILED items by status alone.""" + """FAILED items without an error normalize to ChildContextError. + + Construction and deserialization both fill the same generic error the + executor records for a FAILED child-context checkpoint with no + ErrorObject, so get_errors() never returns None. + """ + child_context_type = ( + "aws_durable_execution_sdk_python.exceptions.ChildContextError" + ) items = [ BatchItem(0, BatchItemStatus.FAILED, error=None), BatchItem( @@ -624,6 +633,55 @@ def test_batch_result_failed_with_none_error(): failed = result.failed() assert len(failed) == result.failure_count == 2 + assert failed[0].error is not None + assert failed[0].error.type == child_context_type + assert failed[0].error.message == ( + "Unknown error. No ErrorObject exists on the Checkpoint Operation." + ) + + errors = result.get_errors() + assert len(errors) == 2 + assert errors[0].type == child_context_type + assert errors[1].type == "Error" + + # Wire payload may still omit error; from_dict must normalize too. + wire = { + "all": [ + {"index": 0, "status": "FAILED", "result": None, "error": None}, + { + "index": 1, + "status": "FAILED", + "result": None, + "error": ErrorObject("msg", "Error", None, None).to_dict(), + }, + ], + "completionReason": "ALL_COMPLETED", + } + round_tripped = BatchResult.from_dict(wire) + round_trip_errors = round_tripped.get_errors() + assert len(round_trip_errors) == 2 + assert round_trip_errors[0].type == child_context_type + assert all(error is not None for error in round_trip_errors) + assert round_trip_errors[0].type is not None + # Accessing error.type must not raise after deserialization. + assert [error.type for error in round_trip_errors] == [ + child_context_type, + "Error", + ] + + # Same gap after ExtendedTypeSerDes round-trip of a wire payload with + # error=null (the path that previously yielded AttributeError on .type). + serdes_payload = ( + '{"t":"br","v":{"all":{"t":"l","v":[{"t":"m","v":{"index":{"t":"i","v":0},' + '"status":{"t":"s","v":"FAILED"},"result":{"t":"n","v":null},' + '"error":{"t":"n","v":null}}}]},"completionReason":{"t":"s","v":"ALL_COMPLETED"}}}' + ) + serdes_result = ExtendedTypeSerDes().deserialize(serdes_payload) + assert isinstance(serdes_result, BatchResult) + serdes_errors = serdes_result.get_errors() + assert len(serdes_errors) == 1 + assert serdes_errors[0].type == child_context_type + def test_concurrent_executor_nesting_type_parameter(): From 32316a24d9b4450716e37104b27ce57b7027391a Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Sat, 24 Oct 2026 11:20:52 +0000 Subject: [PATCH 3/6] fix: simplify batch result accessors Keep missing checkpoint errors at the replay boundary. Scan batch items once and cover None results through replay. --- .../concurrency/executor.py | 19 +- .../concurrency/models.py | 56 +++--- .../aws_durable_execution_sdk_python/state.py | 37 +++- .../tests/concurrency_test.py | 172 ++++++++++-------- .../e2e/batch_result_none_replay_test.py | 83 +++++++++ .../tests/state_test.py | 31 +++- 6 files changed, 270 insertions(+), 128 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python/tests/e2e/batch_result_none_replay_test.py 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 f261be9f0..3a8bcb98b 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 @@ -90,21 +90,6 @@ def _branch_error_object(err: Exception) -> ErrorObject: return ErrorObject.from_exception(err) -def _failed_checkpoint_error_object(checkpoint: CheckpointedResult) -> ErrorObject: - """Return the error of a branch whose child context checkpoint FAILED. - - The checkpoint error is optional. Without one, record the generic - ChildContextError that replaying the child context raises, so every - FAILED item carries an error. - """ - if checkpoint.error is not None: - return checkpoint.error - try: - checkpoint.raise_operation_error(ChildContextError) - except ChildContextError as e: - return _branch_error_object(e) - - @dataclass class ResumeWave(Generic[CallableType, ResultType]): """Branches suspended until one time, and the refresh that resumes them. @@ -815,7 +800,7 @@ def _replay_terminal_item( return BatchItem( executable.index, BatchItemStatus.FAILED, - error=_failed_checkpoint_error_object(checkpoint), + error=checkpoint.to_error_object(ChildContextError), ) if self.nesting_type is NestingType.FLAT: try: @@ -868,7 +853,7 @@ def _replay_from_checkpoints( ) elif checkpoint.is_failed(): - error = _failed_checkpoint_error_object(checkpoint) + error = checkpoint.to_error_object(ChildContextError) status = BatchItemStatus.FAILED else: status = BatchItemStatus.STARTED diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index 17c0e63d6..215701d3e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -42,26 +42,6 @@ # region Result models -_MISSING_FAILED_ITEM_ERROR_MESSAGE = ( - "Unknown error. No ErrorObject exists on the Checkpoint Operation." -) - - -def _generic_failed_item_error() -> ErrorObject: - """ErrorObject for a FAILED batch item that carried no checkpoint error. - - Matches ``_failed_checkpoint_error_object`` when the child-context - checkpoint has no ErrorObject: the same generic ChildContextError - replaying that child would raise. - """ - err = ChildContextError(message=_MISSING_FAILED_ITEM_ERROR_MESSAGE) - return ErrorObject( - message=err.message, - type=err.error_type, - data=err.data, - stack_trace=err.stack_trace, - ) - class CompletionReason(Enum): ALL_COMPLETED = "ALL_COMPLETED" @@ -421,10 +401,6 @@ class BatchItem(Generic[R]): result: R | None = None error: ErrorObject | None = None - def __post_init__(self) -> None: - if self.status is BatchItemStatus.FAILED and self.error is None: - object.__setattr__(self, "error", _generic_failed_item_error()) - def to_dict(self) -> dict: return { "index": self.index, @@ -534,28 +510,44 @@ def has_failure(self) -> bool: return any(item.status is BatchItemStatus.FAILED for item in self.all) def throw_if_error(self) -> None: - first_error = next( - (item.error for item in self.all if item.status is BatchItemStatus.FAILED), + first_failed = next( + (item for item in self.all if item.status is BatchItemStatus.FAILED), None, ) - if first_error is not None: + if first_failed is not None: + first_error = first_failed.error + if first_error is None: + raise ChildContextError( + message=( + f"Batch item {first_failed.index} failed " + "without an ErrorObject." + ) + ) + # Each item runs in a child context, so a failed item surfaces as a # ChildContextError wrapping the escaping error (reconstructed as its # typed form where possible, with __cause__ set). Remaining errors # stay in get_errors(). first_error.raise_as_operation_error(ChildContextError) if self.completion_reason is CompletionReason.CUSTOM_COMPLETION_FAILED: - # A custom decision marked the batch failed with no item error. + # A custom decision marked the batch failed with no failed item. raise BatchCompletionError(self.completion_reason) def get_results(self) -> list[R]: # None is a valid result for a branch that returns nothing. - return [cast("R", item.result) for item in self.succeeded()] + return [ + cast(R, item.result) + for item in self.all + if item.status is BatchItemStatus.SUCCEEDED + ] + + def get_errors(self) -> list[ErrorObject | None]: + """Return one entry per FAILED item, in branch order. - def get_errors(self) -> list[ErrorObject]: + An entry may be None when a deserialized BatchItem contained no error. + """ return [ - item.error if item.error is not None else _generic_failed_item_error() - for item in self.failed() + item.error for item in self.all if item.status is BatchItemStatus.FAILED ] @property 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 f77d5d156..b2f90e467 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 @@ -315,6 +315,33 @@ def is_replay_children(self) -> bool: return False return op.context_details.replay_children if op.context_details else False + def to_error_object( + self, + operation_error_cls: type[DurableOperationError], + msg: str | None = None, + ) -> ErrorObject: + """Adapt a failed checkpoint to an ErrorObject at a replay boundary. + + Return the checkpoint's ErrorObject unchanged when one exists. The + service permits a FAILED operation without an ErrorObject. Synthesize + an operation-specific fallback when the service omitted it. + """ + if self.error is not None: + return self.error + + fallback_error = operation_error_cls( + message=( + msg + or "Unknown error. No ErrorObject exists on the Checkpoint Operation." + ) + ) + return ErrorObject( + message=fallback_error.message, + type=fallback_error.error_type, + data=fallback_error.data, + stack_trace=fallback_error.stack_trace, + ) + def raise_operation_error( self, operation_error_cls: type[DurableOperationError], @@ -328,11 +355,13 @@ def raise_operation_error( (invoke/callback) still surface as the correct per-operation error. """ if self.error is None: - err_msg: str = ( - msg - or "Unknown error. No ErrorObject exists on the Checkpoint Operation." + error = self.to_error_object(operation_error_cls, msg) + raise operation_error_cls( + message=error.message, + error_type=error.type, + data=error.data, + stack_trace=error.stack_trace, ) - raise operation_error_cls(message=err_msg) # Reconstruct from the checkpointed ErrorObject. This is the same path the # handlers use on the first-run failure, so first run and replay surface 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 465ab3f84..b85b18ab0 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -63,7 +63,6 @@ SuspendExecution, TimedSuspendExecution, ) -from aws_durable_execution_sdk_python.serdes import ExtendedTypeSerDes from aws_durable_execution_sdk_python.lambda_service import ( DurableServiceClient, ErrorObject, @@ -85,6 +84,7 @@ DurableInstrumentationPlugin, PluginExecutor, ) +from aws_durable_execution_sdk_python.serdes import ExtendedTypeSerDes from aws_durable_execution_sdk_python.state import ( CheckpointedResult, ExecutionState, @@ -313,6 +313,96 @@ def test_batch_result_get_errors(): assert error2 in errors +def test_batch_result_preserves_missing_errors_through_serdes(): + """Deserialization preserves one optional error per FAILED item.""" + real_error = ErrorObject("real error", "RealError", "data", ["line"]) + wire = { + "all": [ + {"index": 2, "status": "FAILED", "result": None, "error": None}, + { + "index": 5, + "status": "FAILED", + "result": None, + "error": real_error.to_dict(), + }, + ], + "completionReason": "ALL_COMPLETED", + } + + restored = BatchResult.from_dict(wire) + assert restored.get_errors() == [None, real_error] + assert len(restored.get_errors()) == restored.failure_count == 2 + + serdes = ExtendedTypeSerDes() + round_tripped = serdes.deserialize(serdes.serialize(restored)) + + assert isinstance(round_tripped, BatchResult) + assert round_tripped.get_errors() == [None, real_error] + assert len(round_tripped.get_errors()) == round_tripped.failure_count == 2 + + +def test_batch_result_throw_if_error_without_error_uses_item_index(): + """A FAILED item without an ErrorObject raises a deterministic error.""" + result = BatchResult( + [BatchItem(7, BatchItemStatus.FAILED)], + CompletionReason.ALL_COMPLETED, + ) + + with pytest.raises(ChildContextError) as exc_info: + result.throw_if_error() + + assert exc_info.value.message == "Batch item 7 failed without an ErrorObject." + assert exc_info.value.error_type == ( + "aws_durable_execution_sdk_python.exceptions.ChildContextError" + ) + assert exc_info.value.__cause__ is None + + +def test_batch_result_throw_if_error_uses_first_failed_item(): + """A later real error does not replace the first FAILED item.""" + later_error = ErrorObject("later error", "LaterError", None, None) + result = BatchResult( + [ + BatchItem(3, BatchItemStatus.FAILED), + BatchItem(4, BatchItemStatus.FAILED, error=later_error), + ], + CompletionReason.ALL_COMPLETED, + ) + + assert result.get_errors() == [None, later_error] + with pytest.raises( + ChildContextError, + match=r"Batch item 3 failed without an ErrorObject\.", + ): + result.throw_if_error() + + +def test_batch_result_throw_if_error_preserves_real_first_error(): + """A real error on the first FAILED item retains all error fields.""" + first_error = ErrorObject( + "first error", + "OriginalError", + "error data", + ["stack line"], + ) + result = BatchResult( + [ + BatchItem(3, BatchItemStatus.FAILED, error=first_error), + BatchItem(4, BatchItemStatus.FAILED), + ], + CompletionReason.ALL_COMPLETED, + ) + + assert result.get_errors()[0] is first_error + with pytest.raises(ChildContextError) as exc_info: + result.throw_if_error() + + assert exc_info.value.message == first_error.message + assert exc_info.value.error_type == first_error.type + assert exc_info.value.data == first_error.data + assert exc_info.value.stack_trace == first_error.stack_trace + + def test_batch_result_keeps_succeeded_items_with_none_result(): """A branch that returns None still counts as succeeded.""" items = [ @@ -615,77 +705,6 @@ def test_func(): assert executable.func == test_func -def test_batch_result_failed_with_none_error(): - """FAILED items without an error normalize to ChildContextError. - - Construction and deserialization both fill the same generic error the - executor records for a FAILED child-context checkpoint with no - ErrorObject, so get_errors() never returns None. - """ - child_context_type = ( - "aws_durable_execution_sdk_python.exceptions.ChildContextError" - ) - items = [ - BatchItem(0, BatchItemStatus.FAILED, error=None), - BatchItem( - 1, BatchItemStatus.FAILED, error=ErrorObject("msg", "Error", None, None) - ), - ] - result = BatchResult(items, CompletionReason.ALL_COMPLETED) - - failed = result.failed() - assert len(failed) == result.failure_count == 2 - assert failed[0].error is not None - assert failed[0].error.type == child_context_type - assert failed[0].error.message == ( - "Unknown error. No ErrorObject exists on the Checkpoint Operation." - ) - - errors = result.get_errors() - assert len(errors) == 2 - assert errors[0].type == child_context_type - assert errors[1].type == "Error" - - # Wire payload may still omit error; from_dict must normalize too. - wire = { - "all": [ - {"index": 0, "status": "FAILED", "result": None, "error": None}, - { - "index": 1, - "status": "FAILED", - "result": None, - "error": ErrorObject("msg", "Error", None, None).to_dict(), - }, - ], - "completionReason": "ALL_COMPLETED", - } - round_tripped = BatchResult.from_dict(wire) - round_trip_errors = round_tripped.get_errors() - assert len(round_trip_errors) == 2 - assert round_trip_errors[0].type == child_context_type - assert all(error is not None for error in round_trip_errors) - assert round_trip_errors[0].type is not None - # Accessing error.type must not raise after deserialization. - assert [error.type for error in round_trip_errors] == [ - child_context_type, - "Error", - ] - - # Same gap after ExtendedTypeSerDes round-trip of a wire payload with - # error=null (the path that previously yielded AttributeError on .type). - serdes_payload = ( - '{"t":"br","v":{"all":{"t":"l","v":[{"t":"m","v":{"index":{"t":"i","v":0},' - '"status":{"t":"s","v":"FAILED"},"result":{"t":"n","v":null},' - '"error":{"t":"n","v":null}}}]},"completionReason":{"t":"s","v":"ALL_COMPLETED"}}}' - ) - serdes_result = ExtendedTypeSerDes().deserialize(serdes_payload) - assert isinstance(serdes_result, BatchResult) - serdes_errors = serdes_result.get_errors() - assert len(serdes_errors) == 1 - assert serdes_errors[0].type == child_context_type - - - def test_concurrent_executor_nesting_type_parameter(): """Test ConcurrentExecutor nesting_type parameter.""" @@ -1156,6 +1175,7 @@ def test_replay_failed_checkpoint_without_error_records_child_context_error( assert [item.index for item in result.failed()] == [0] errors = result.get_errors() assert len(errors) == 1 + assert errors[0] is not None assert errors[0].type == ( "aws_durable_execution_sdk_python.exceptions.ChildContextError" ) @@ -1742,6 +1762,7 @@ def make_executor(): failed_checkpoint.error = ErrorObject( message="Test error", type="ValueError", data=None, stack_trace=None ) + failed_checkpoint.to_error_object.return_value = failed_checkpoint.error replay_state = Mock() replay_state.get_checkpoint_result = Mock(return_value=failed_checkpoint) replay_context = Mock() @@ -4112,7 +4133,7 @@ def suspend(): # (a terminal checkpoint can land after the completion decision). replay_state, replay_context = _make_replay_mocks( { - "op_0": _succeeded_checkpoint('"replayed_0"'), + "op_0": _succeeded_checkpoint("null"), "op_1": _succeeded_checkpoint('"raced_1"'), "op_2": _succeeded_checkpoint('"replayed_2"'), } @@ -4130,9 +4151,12 @@ def suspend(): (i.index, i.status) for i in live.all ] assert replayed.completion_reason is live.completion_reason - assert replayed.all[0].result == "replayed_0" + assert replayed.all[0].result is None assert replayed.all[1].result is None assert replayed.all[2].result == "replayed_2" + assert replayed.success_count == 2 + assert [item.index for item in replayed.succeeded()] == [0, 2] + assert replayed.get_results() == [None, "replayed_2"] def test_execute_keeps_branches_that_return_none(): diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/batch_result_none_replay_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/batch_result_none_replay_test.py new file mode 100644 index 000000000..427f68cb6 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/batch_result_none_replay_test.py @@ -0,0 +1,83 @@ +"""None-valued batch results remain visible across checkpoint replay.""" + +from __future__ import annotations + +import json +from collections.abc import Callable, Sequence +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import ( + BatchResult, + DurableContext, + durable_execution, +) +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +def _map_item( + _context: DurableContext, + item: int | None, + _index: int, + _items: Sequence[int | None], +) -> int | None: + return item + + +def _parallel_branch(value: int | None) -> Callable[[DurableContext], int | None]: + def branch(_context: DurableContext) -> int | None: + return value + + return branch + + +@durable_execution +def _batch_handler(event: dict[str, str], context: DurableContext) -> dict[str, Any]: + batch: BatchResult[int | None] + if event["kind"] == "map": + batch = context.map( + [None, 10, None, 30], + _map_item, + name="none-result-map", + ) + else: + batch = context.parallel( + [ + _parallel_branch(None), + _parallel_branch(10), + _parallel_branch(None), + _parallel_branch(30), + ], + name="none-result-parallel", + ) + + # Suspend after the batch. The next invocation reads the BatchResult from + # its parent-context checkpoint before these accessors run. + context.wait(Duration.from_seconds(1), name="after-batch") + return { + "success_count": batch.success_count, + "succeeded_indexes": [item.index for item in batch.succeeded()], + "results": batch.get_results(), + } + + +@pytest.mark.parametrize("kind", ["map", "parallel"]) +def test_none_results_survive_checkpoint_replay(kind: str) -> None: + with DurableFunctionTestRunner(handler=_batch_handler) as runner: + result: DurableFunctionTestResult = runner.run( + input=json.dumps({"kind": kind}), + execution_timeout=15, + ) + + assert result.status is InvocationStatus.SUCCEEDED + assert result.result is not None + assert json.loads(result.result) == { + "success_count": 4, + "succeeded_indexes": [0, 1, 2, 3], + "results": [None, 10, None, 30], + } 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 128aa33cd..b1c8dadb7 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -381,6 +381,30 @@ def test_checkpointed_result_is_started(): assert result_no_op.is_started() is False +def test_checkpointed_result_to_error_object_returns_checkpoint_error(): + """to_error_object preserves an ErrorObject already on the checkpoint.""" + error = ErrorObject( + message="Test error", type="ValueError", data="d", stack_trace=["l"] + ) + result = CheckpointedResult(error=error) + + assert result.to_error_object(StepError) is error + + +def test_checkpointed_result_to_error_object_without_error(): + """to_error_object represents the typed error direct replay would raise.""" + result = CheckpointedResult() + + error = result.to_error_object(StepError) + + assert error == ErrorObject( + message="Unknown error. No ErrorObject exists on the Checkpoint Operation.", + type="aws_durable_execution_sdk_python.exceptions.StepError", + data=None, + stack_trace=None, + ) + + def test_checkpointed_result_raise_operation_error(): """raise_operation_error wraps the checkpoint error in the given operation type.""" error = ErrorObject( @@ -404,9 +428,14 @@ def test_checkpointed_result_raise_operation_error_no_error(): """raise_operation_error with no error raises the operation type with a default message.""" result = CheckpointedResult() - with pytest.raises(StepError, match="Unknown error"): + with pytest.raises(StepError, match="Unknown error") as exc_info: result.raise_operation_error(StepError) + assert exc_info.value.error_type == ( + "aws_durable_execution_sdk_python.exceptions.StepError" + ) + assert exc_info.value.__cause__ is None + def test_checkpointed_result_raise_operation_error_no_error_with_message(): """raise_operation_error with no error uses the provided custom message.""" From eb6ba3ae09ddfc7b4e38b77f113f57a2266d9ae7 Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Sat, 24 Oct 2026 21:38:45 +0000 Subject: [PATCH 4/6] fix(sdk): fill in missing batch item errors on read A FAILED batch item read back from a payload with no error kept error=None. get_errors() then returned None entries, so its return type had been widened to list[ErrorObject | None], which breaks typed callers that read fields on each entry. BatchItem.from_dict now fills in the same generic ChildContextError that replay records for a FAILED checkpoint with no error. Both paths build it through ErrorObject.for_missing_error, so a missing error has one message. get_errors() returns list[ErrorObject] again, and throw_if_error() no longer needs a second fallback message. --- .../concurrency/models.py | 45 +++++---- .../lambda_service.py | 20 ++++ .../aws_durable_execution_sdk_python/state.py | 18 +--- .../tests/concurrency_test.py | 94 +++++++++++++------ .../tests/lambda_service_test.py | 20 ++++ 5 files changed, 134 insertions(+), 63 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index 215701d3e..dbc1dfff4 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -396,6 +396,14 @@ def generate(result: BatchResult) -> str: @dataclass(frozen=True) class BatchItem(Generic[R]): + """One branch of a map or parallel batch. + + A FAILED item carries an error. The executor records the branch's error, + or the generic ChildContextError when a FAILED child-context checkpoint + has none, and :meth:`from_dict` fills in that same error when a payload + has a FAILED status and no error. + """ + index: int status: BatchItemStatus result: R | None = None @@ -411,11 +419,16 @@ def to_dict(self) -> dict: @classmethod def from_dict(cls, data: dict) -> BatchItem[R]: + status = BatchItemStatus(data["status"]) + error = ErrorObject.from_dict(data["error"]) if data.get("error") else None + if status is BatchItemStatus.FAILED and error is None: + # The same error replay records for a FAILED checkpoint with none. + error = ErrorObject.for_missing_error(ChildContextError) return cls( index=data["index"], - status=BatchItemStatus(data["status"]), + status=status, result=data.get("result"), - error=ErrorObject.from_dict(data["error"]) if data.get("error") else None, + error=error, ) @@ -510,20 +523,11 @@ def has_failure(self) -> bool: return any(item.status is BatchItemStatus.FAILED for item in self.all) def throw_if_error(self) -> None: - first_failed = next( - (item for item in self.all if item.status is BatchItemStatus.FAILED), + first_error = next( + (item.error for item in self.all if item.status is BatchItemStatus.FAILED), None, ) - if first_failed is not None: - first_error = first_failed.error - if first_error is None: - raise ChildContextError( - message=( - f"Batch item {first_failed.index} failed " - "without an ErrorObject." - ) - ) - + if first_error is not None: # Each item runs in a child context, so a failed item surfaces as a # ChildContextError wrapping the escaping error (reconstructed as its # typed form where possible, with __cause__ set). Remaining errors @@ -541,13 +545,14 @@ def get_results(self) -> list[R]: if item.status is BatchItemStatus.SUCCEEDED ] - def get_errors(self) -> list[ErrorObject | None]: - """Return one entry per FAILED item, in branch order. - - An entry may be None when a deserialized BatchItem contained no error. - """ + def get_errors(self) -> list[ErrorObject]: + """Return one error per FAILED item, in branch order.""" + # A FAILED item always carries an error (see BatchItem), so the cast + # states that invariant for the type checker. return [ - item.error for item in self.all if item.status is BatchItemStatus.FAILED + cast(ErrorObject, item.error) + for item in self.all + if item.status is BatchItemStatus.FAILED ] @property diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index c7c567fa3..d548d5cbd 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -265,6 +265,26 @@ def from_message(cls, message: str) -> ErrorObject: stack_trace=None, ) + @classmethod + def for_missing_error( + cls, + operation_error_cls: builtins.type[DurableOperationError], + msg: str | None = None, + ) -> ErrorObject: + """Build the error for a FAILED operation that recorded none. + + The service permits a FAILED operation without an ErrorObject. Replay + of such a checkpoint and deserialization of such a batch item both + call this, so a missing error has one representation: the operation's + own error type carrying a generic message. + """ + return cls.from_exception( + operation_error_cls( + message=msg + or "Unknown error. No ErrorObject exists on the Checkpoint Operation." + ) + ) + def to_dict(self) -> MutableMapping[str, Any]: result: MutableMapping[str, Any] = {} if self.message is not None: 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 f10b931f2..c2a4f2ab1 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 @@ -325,24 +325,12 @@ def to_error_object( """Adapt a failed checkpoint to an ErrorObject at a replay boundary. Return the checkpoint's ErrorObject unchanged when one exists. The - service permits a FAILED operation without an ErrorObject. Synthesize - an operation-specific fallback when the service omitted it. + service permits a FAILED operation without an ErrorObject; for that + case return ``ErrorObject.for_missing_error`` for the operation. """ if self.error is not None: return self.error - - fallback_error = operation_error_cls( - message=( - msg - or "Unknown error. No ErrorObject exists on the Checkpoint Operation." - ) - ) - return ErrorObject( - message=fallback_error.message, - type=fallback_error.error_type, - data=fallback_error.data, - stack_trace=fallback_error.stack_trace, - ) + return ErrorObject.for_missing_error(operation_error_cls, msg) def raise_operation_error( self, 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 b85b18ab0..f86448e12 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -313,8 +313,14 @@ def test_batch_result_get_errors(): assert error2 in errors -def test_batch_result_preserves_missing_errors_through_serdes(): - """Deserialization preserves one optional error per FAILED item.""" +def test_batch_result_from_dict_fills_in_missing_failed_item_error(): + """A FAILED payload with no error gets the replay fallback error. + + The same payload can arrive through ``BatchResult.from_dict`` directly or + through the default ``ExtendedTypeSerDes``. Both must yield an item whose + error equals the one replay records for a FAILED checkpoint with none, so + ``get_errors()`` keeps its ``list[ErrorObject]`` contract. + """ real_error = ErrorObject("real error", "RealError", "data", ["line"]) wire = { "all": [ @@ -328,53 +334,84 @@ def test_batch_result_preserves_missing_errors_through_serdes(): ], "completionReason": "ALL_COMPLETED", } + expected_fallback = CheckpointedResult().to_error_object(ChildContextError) restored = BatchResult.from_dict(wire) - assert restored.get_errors() == [None, real_error] + assert restored.get_errors() == [expected_fallback, real_error] assert len(restored.get_errors()) == restored.failure_count == 2 serdes = ExtendedTypeSerDes() round_tripped = serdes.deserialize(serdes.serialize(restored)) - assert isinstance(round_tripped, BatchResult) - assert round_tripped.get_errors() == [None, real_error] - assert len(round_tripped.get_errors()) == round_tripped.failure_count == 2 + assert round_tripped.get_errors() == [expected_fallback, real_error] + + # The typed-caller pattern from the migration guide works on every entry. + assert [error.type for error in round_tripped.get_errors()] == [ + "aws_durable_execution_sdk_python.exceptions.ChildContextError", + "RealError", + ] -def test_batch_result_throw_if_error_without_error_uses_item_index(): - """A FAILED item without an ErrorObject raises a deterministic error.""" - result = BatchResult( - [BatchItem(7, BatchItemStatus.FAILED)], - CompletionReason.ALL_COMPLETED, +def test_batch_result_from_dict_keeps_absent_error_on_non_failed_items(): + """Only FAILED items get the fallback; SUCCEEDED and STARTED keep None.""" + wire = { + "all": [ + {"index": 0, "status": "SUCCEEDED", "result": None, "error": None}, + {"index": 1, "status": "STARTED", "result": None, "error": None}, + ], + "completionReason": "ALL_COMPLETED", + } + + restored = BatchResult.from_dict(wire) + + assert [item.error for item in restored.all] == [None, None] + assert restored.get_errors() == [] + + +def test_batch_result_throw_if_error_raises_filled_in_error(): + """A deserialized FAILED item with no error raises the generic error.""" + restored = BatchResult.from_dict( + { + "all": [{"index": 7, "status": "FAILED", "result": None, "error": None}], + "completionReason": "ALL_COMPLETED", + } ) with pytest.raises(ChildContextError) as exc_info: - result.throw_if_error() + restored.throw_if_error() - assert exc_info.value.message == "Batch item 7 failed without an ErrorObject." + assert exc_info.value.message == ( + "Unknown error. No ErrorObject exists on the Checkpoint Operation." + ) assert exc_info.value.error_type == ( "aws_durable_execution_sdk_python.exceptions.ChildContextError" ) - assert exc_info.value.__cause__ is None def test_batch_result_throw_if_error_uses_first_failed_item(): """A later real error does not replace the first FAILED item.""" later_error = ErrorObject("later error", "LaterError", None, None) - result = BatchResult( - [ - BatchItem(3, BatchItemStatus.FAILED), - BatchItem(4, BatchItemStatus.FAILED, error=later_error), - ], - CompletionReason.ALL_COMPLETED, + restored = BatchResult.from_dict( + { + "all": [ + {"index": 3, "status": "FAILED", "result": None, "error": None}, + { + "index": 4, + "status": "FAILED", + "result": None, + "error": later_error.to_dict(), + }, + ], + "completionReason": "ALL_COMPLETED", + } ) - assert result.get_errors() == [None, later_error] - with pytest.raises( - ChildContextError, - match=r"Batch item 3 failed without an ErrorObject\.", - ): - result.throw_if_error() + assert [error.type for error in restored.get_errors()] == [ + "aws_durable_execution_sdk_python.exceptions.ChildContextError", + "LaterError", + ] + with pytest.raises(ChildContextError, match="No ErrorObject exists"): + restored.throw_if_error() def test_batch_result_throw_if_error_preserves_real_first_error(): @@ -388,7 +425,9 @@ def test_batch_result_throw_if_error_preserves_real_first_error(): result = BatchResult( [ BatchItem(3, BatchItemStatus.FAILED, error=first_error), - BatchItem(4, BatchItemStatus.FAILED), + BatchItem( + 4, BatchItemStatus.FAILED, error=ErrorObject("x", "X", None, None) + ), ], CompletionReason.ALL_COMPLETED, ) @@ -1175,7 +1214,6 @@ def test_replay_failed_checkpoint_without_error_records_child_context_error( assert [item.index for item in result.failed()] == [0] errors = result.get_errors() assert len(errors) == 1 - assert errors[0] is not None assert errors[0].type == ( "aws_durable_execution_sdk_python.exceptions.ChildContextError" ) diff --git a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py index 8ae3d6111..614b16769 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -227,6 +227,26 @@ def test_error_object_from_message_empty(): assert error.stack_trace is None +def test_error_object_for_missing_error_uses_operation_type_and_default_message(): + """A missing error is represented as the operation's own type.""" + error = ErrorObject.for_missing_error(StepError) + + assert error == ErrorObject( + message="Unknown error. No ErrorObject exists on the Checkpoint Operation.", + type="aws_durable_execution_sdk_python.exceptions.StepError", + data=None, + stack_trace=None, + ) + + +def test_error_object_for_missing_error_takes_custom_message(): + """A caller-supplied message replaces the default text.""" + error = ErrorObject.for_missing_error(StepError, "Custom error message") + + assert error.message == "Custom error message" + assert error.type == "aws_durable_execution_sdk_python.exceptions.StepError" + + def test_invocation_output_create_pending_is_bare_pending(): """A pending response contains neither a result nor an error.""" output = DurableExecutionInvocationOutput.create_pending() From 042e36896dc621142195ae6d15a687796a9ebe24 Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Sat, 24 Oct 2026 22:34:07 +0000 Subject: [PATCH 5/6] fix(sdk): keep a missing batch item error as None A FAILED branch whose checkpoint carried no error was given a manufactured ErrorObject on replay, and the unpushed follow-up did the same on deserialization. Lambda permits a FAILED operation without an Error, so the SDK now preserves that state instead of inventing data. Replay and from_dict store the error as received, including None. get_errors() returns list[ErrorObject | None], one entry per failed item in branch order. throw_if_error() raises a generic ChildContextError when the first failed item has no error, so a failure is never silently ignored. The migration guide example logs every failed item, including one with no ErrorObject. --- docs/migration-1.x-to-2.0.md | 14 ++- .../concurrency/executor.py | 7 +- .../concurrency/models.py | 43 +++---- .../lambda_service.py | 20 --- .../aws_durable_execution_sdk_python/state.py | 25 +--- .../tests/concurrency_test.py | 117 ++++++------------ .../tests/lambda_service_test.py | 20 --- .../tests/state_test.py | 31 +---- 8 files changed, 81 insertions(+), 196 deletions(-) diff --git a/docs/migration-1.x-to-2.0.md b/docs/migration-1.x-to-2.0.md index cd9fb64ef..b041c6909 100644 --- a/docs/migration-1.x-to-2.0.md +++ b/docs/migration-1.x-to-2.0.md @@ -348,8 +348,18 @@ result = context.map(items, process_item) try: result.throw_if_error() except (ChildContextError, SerDesError, BatchCompletionError): - for err in result.get_errors(): # every failed item's ErrorObject - context.logger.error("%s: %s", err.type, err.message) + for item in result.failed(): + if item.error is None: + context.logger.error( + "Batch item %s failed without an ErrorObject", + item.index, + ) + else: + context.logger.error( + "%s: %s", + item.error.type, + item.error.message, + ) ``` ## Serialize and Deserialize Round Trip 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 3a8bcb98b..fd049d3bf 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 @@ -32,7 +32,6 @@ NestingType, ) from aws_durable_execution_sdk_python.exceptions import ( - ChildContextError, DurableOperationError, ExecutionError, InvalidStateError, @@ -798,9 +797,7 @@ def _replay_terminal_item( return BatchItem(executable.index, BatchItemStatus.SUCCEEDED, result) if checkpoint.is_failed(): return BatchItem( - executable.index, - BatchItemStatus.FAILED, - error=checkpoint.to_error_object(ChildContextError), + executable.index, BatchItemStatus.FAILED, error=checkpoint.error ) if self.nesting_type is NestingType.FLAT: try: @@ -853,7 +850,7 @@ def _replay_from_checkpoints( ) elif checkpoint.is_failed(): - error = checkpoint.to_error_object(ChildContextError) + error = checkpoint.error status = BatchItemStatus.FAILED else: status = BatchItemStatus.STARTED diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py index dbc1dfff4..6b4053d04 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/models.py @@ -398,10 +398,8 @@ def generate(result: BatchResult) -> str: class BatchItem(Generic[R]): """One branch of a map or parallel batch. - A FAILED item carries an error. The executor records the branch's error, - or the generic ChildContextError when a FAILED child-context checkpoint - has none, and :meth:`from_dict` fills in that same error when a payload - has a FAILED status and no error. + ``error`` may be None on a FAILED item when its source contained no + ErrorObject. ``BatchResult.throw_if_error`` still raises for it. """ index: int @@ -419,16 +417,11 @@ def to_dict(self) -> dict: @classmethod def from_dict(cls, data: dict) -> BatchItem[R]: - status = BatchItemStatus(data["status"]) - error = ErrorObject.from_dict(data["error"]) if data.get("error") else None - if status is BatchItemStatus.FAILED and error is None: - # The same error replay records for a FAILED checkpoint with none. - error = ErrorObject.for_missing_error(ChildContextError) return cls( index=data["index"], - status=status, + status=BatchItemStatus(data["status"]), result=data.get("result"), - error=error, + error=ErrorObject.from_dict(data["error"]) if data.get("error") else None, ) @@ -523,11 +516,20 @@ def has_failure(self) -> bool: return any(item.status is BatchItemStatus.FAILED for item in self.all) def throw_if_error(self) -> None: - first_error = next( - (item.error for item in self.all if item.status is BatchItemStatus.FAILED), + first_failed = next( + (item for item in self.all if item.status is BatchItemStatus.FAILED), None, ) - if first_error is not None: + if first_failed is not None: + first_error = first_failed.error + if first_error is None: + raise ChildContextError( + message=( + f"Batch item {first_failed.index} failed " + "without an ErrorObject." + ) + ) + # Each item runs in a child context, so a failed item surfaces as a # ChildContextError wrapping the escaping error (reconstructed as its # typed form where possible, with __cause__ set). Remaining errors @@ -545,14 +547,13 @@ def get_results(self) -> list[R]: if item.status is BatchItemStatus.SUCCEEDED ] - def get_errors(self) -> list[ErrorObject]: - """Return one error per FAILED item, in branch order.""" - # A FAILED item always carries an error (see BatchItem), so the cast - # states that invariant for the type checker. + def get_errors(self) -> list[ErrorObject | None]: + """Return one entry per FAILED item, in branch order. + + An entry is None when the corresponding item has no ErrorObject. + """ return [ - cast(ErrorObject, item.error) - for item in self.all - if item.status is BatchItemStatus.FAILED + item.error for item in self.all if item.status is BatchItemStatus.FAILED ] @property diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index d548d5cbd..c7c567fa3 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -265,26 +265,6 @@ def from_message(cls, message: str) -> ErrorObject: stack_trace=None, ) - @classmethod - def for_missing_error( - cls, - operation_error_cls: builtins.type[DurableOperationError], - msg: str | None = None, - ) -> ErrorObject: - """Build the error for a FAILED operation that recorded none. - - The service permits a FAILED operation without an ErrorObject. Replay - of such a checkpoint and deserialization of such a batch item both - call this, so a missing error has one representation: the operation's - own error type carrying a generic message. - """ - return cls.from_exception( - operation_error_cls( - message=msg - or "Unknown error. No ErrorObject exists on the Checkpoint Operation." - ) - ) - def to_dict(self) -> MutableMapping[str, Any]: result: MutableMapping[str, Any] = {} if self.message is not None: 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 c2a4f2ab1..e9a7588d2 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 @@ -317,21 +317,6 @@ def is_replay_children(self) -> bool: return False return op.context_details.replay_children if op.context_details else False - def to_error_object( - self, - operation_error_cls: type[DurableOperationError], - msg: str | None = None, - ) -> ErrorObject: - """Adapt a failed checkpoint to an ErrorObject at a replay boundary. - - Return the checkpoint's ErrorObject unchanged when one exists. The - service permits a FAILED operation without an ErrorObject; for that - case return ``ErrorObject.for_missing_error`` for the operation. - """ - if self.error is not None: - return self.error - return ErrorObject.for_missing_error(operation_error_cls, msg) - def raise_operation_error( self, operation_error_cls: type[DurableOperationError], @@ -345,13 +330,11 @@ def raise_operation_error( (invoke/callback) still surface as the correct per-operation error. """ if self.error is None: - error = self.to_error_object(operation_error_cls, msg) - raise operation_error_cls( - message=error.message, - error_type=error.type, - data=error.data, - stack_trace=error.stack_trace, + err_msg: str = ( + msg + or "Unknown error. No ErrorObject exists on the Checkpoint Operation." ) + raise operation_error_cls(message=err_msg) # Reconstruct from the checkpointed ErrorObject. This is the same path the # handlers use on the first-run failure, so first run and replay surface 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 f86448e12..bf0c96ebc 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -313,14 +313,8 @@ def test_batch_result_get_errors(): assert error2 in errors -def test_batch_result_from_dict_fills_in_missing_failed_item_error(): - """A FAILED payload with no error gets the replay fallback error. - - The same payload can arrive through ``BatchResult.from_dict`` directly or - through the default ``ExtendedTypeSerDes``. Both must yield an item whose - error equals the one replay records for a FAILED checkpoint with none, so - ``get_errors()`` keeps its ``list[ErrorObject]`` contract. - """ +def test_batch_result_preserves_missing_errors_through_serdes(): + """Deserialization preserves one optional error per FAILED item.""" real_error = ErrorObject("real error", "RealError", "data", ["line"]) wire = { "all": [ @@ -334,84 +328,53 @@ def test_batch_result_from_dict_fills_in_missing_failed_item_error(): ], "completionReason": "ALL_COMPLETED", } - expected_fallback = CheckpointedResult().to_error_object(ChildContextError) restored = BatchResult.from_dict(wire) - assert restored.get_errors() == [expected_fallback, real_error] + assert restored.get_errors() == [None, real_error] assert len(restored.get_errors()) == restored.failure_count == 2 serdes = ExtendedTypeSerDes() round_tripped = serdes.deserialize(serdes.serialize(restored)) - assert isinstance(round_tripped, BatchResult) - assert round_tripped.get_errors() == [expected_fallback, real_error] - - # The typed-caller pattern from the migration guide works on every entry. - assert [error.type for error in round_tripped.get_errors()] == [ - "aws_durable_execution_sdk_python.exceptions.ChildContextError", - "RealError", - ] - - -def test_batch_result_from_dict_keeps_absent_error_on_non_failed_items(): - """Only FAILED items get the fallback; SUCCEEDED and STARTED keep None.""" - wire = { - "all": [ - {"index": 0, "status": "SUCCEEDED", "result": None, "error": None}, - {"index": 1, "status": "STARTED", "result": None, "error": None}, - ], - "completionReason": "ALL_COMPLETED", - } - - restored = BatchResult.from_dict(wire) - assert [item.error for item in restored.all] == [None, None] - assert restored.get_errors() == [] + assert isinstance(round_tripped, BatchResult) + assert round_tripped.get_errors() == [None, real_error] + assert len(round_tripped.get_errors()) == round_tripped.failure_count == 2 -def test_batch_result_throw_if_error_raises_filled_in_error(): - """A deserialized FAILED item with no error raises the generic error.""" - restored = BatchResult.from_dict( - { - "all": [{"index": 7, "status": "FAILED", "result": None, "error": None}], - "completionReason": "ALL_COMPLETED", - } +def test_batch_result_throw_if_error_without_error_uses_item_index(): + """A FAILED item without an ErrorObject raises a deterministic error.""" + result = BatchResult( + [BatchItem(7, BatchItemStatus.FAILED)], + CompletionReason.ALL_COMPLETED, ) with pytest.raises(ChildContextError) as exc_info: - restored.throw_if_error() + result.throw_if_error() - assert exc_info.value.message == ( - "Unknown error. No ErrorObject exists on the Checkpoint Operation." - ) + assert exc_info.value.message == "Batch item 7 failed without an ErrorObject." assert exc_info.value.error_type == ( "aws_durable_execution_sdk_python.exceptions.ChildContextError" ) + assert exc_info.value.__cause__ is None def test_batch_result_throw_if_error_uses_first_failed_item(): """A later real error does not replace the first FAILED item.""" later_error = ErrorObject("later error", "LaterError", None, None) - restored = BatchResult.from_dict( - { - "all": [ - {"index": 3, "status": "FAILED", "result": None, "error": None}, - { - "index": 4, - "status": "FAILED", - "result": None, - "error": later_error.to_dict(), - }, - ], - "completionReason": "ALL_COMPLETED", - } + result = BatchResult( + [ + BatchItem(3, BatchItemStatus.FAILED), + BatchItem(4, BatchItemStatus.FAILED, error=later_error), + ], + CompletionReason.ALL_COMPLETED, ) - assert [error.type for error in restored.get_errors()] == [ - "aws_durable_execution_sdk_python.exceptions.ChildContextError", - "LaterError", - ] - with pytest.raises(ChildContextError, match="No ErrorObject exists"): - restored.throw_if_error() + assert result.get_errors() == [None, later_error] + with pytest.raises( + ChildContextError, + match=r"Batch item 3 failed without an ErrorObject\.", + ): + result.throw_if_error() def test_batch_result_throw_if_error_preserves_real_first_error(): @@ -425,9 +388,7 @@ def test_batch_result_throw_if_error_preserves_real_first_error(): result = BatchResult( [ BatchItem(3, BatchItemStatus.FAILED, error=first_error), - BatchItem( - 4, BatchItemStatus.FAILED, error=ErrorObject("x", "X", None, None) - ), + BatchItem(4, BatchItemStatus.FAILED), ], CompletionReason.ALL_COMPLETED, ) @@ -1170,10 +1131,12 @@ def test_flat_replay_rejects_failed_nested_container_checkpoint(replay_path): @pytest.mark.parametrize("replay_path", ["recorded-terminal", "fallback"]) -def test_replay_failed_checkpoint_without_error_records_child_context_error( - replay_path, -): - """A FAILED branch checkpoint may carry no error; the item still gets one.""" +def test_replay_failed_checkpoint_without_error_keeps_none(replay_path): + """A FAILED branch checkpoint may carry no error; the item keeps None. + + Replay stores the checkpoint error as recorded. The item stays FAILED, + get_errors() keeps its position, and throw_if_error() still raises. + """ executor = _RecordingExecutor( executables=[Executable(0, lambda: "must-not-run")], max_concurrency=1, @@ -1212,14 +1175,15 @@ def test_replay_failed_checkpoint_without_error_records_child_context_error( ) assert [item.index for item in result.failed()] == [0] - errors = result.get_errors() - assert len(errors) == 1 - assert errors[0].type == ( - "aws_durable_execution_sdk_python.exceptions.ChildContextError" - ) - with pytest.raises(ChildContextError, match="No ErrorObject exists"): + assert result.failed()[0].error is None + assert result.get_errors() == [None] + assert result.has_failure + + with pytest.raises(ChildContextError, match="without an ErrorObject") as exc_info: result.throw_if_error() + assert exc_info.value.__cause__ is None + @pytest.mark.parametrize( ("nesting_type", "checkpoint_name", "mismatch"), @@ -1800,7 +1764,6 @@ def make_executor(): failed_checkpoint.error = ErrorObject( message="Test error", type="ValueError", data=None, stack_trace=None ) - failed_checkpoint.to_error_object.return_value = failed_checkpoint.error replay_state = Mock() replay_state.get_checkpoint_result = Mock(return_value=failed_checkpoint) replay_context = Mock() diff --git a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py index 614b16769..8ae3d6111 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -227,26 +227,6 @@ def test_error_object_from_message_empty(): assert error.stack_trace is None -def test_error_object_for_missing_error_uses_operation_type_and_default_message(): - """A missing error is represented as the operation's own type.""" - error = ErrorObject.for_missing_error(StepError) - - assert error == ErrorObject( - message="Unknown error. No ErrorObject exists on the Checkpoint Operation.", - type="aws_durable_execution_sdk_python.exceptions.StepError", - data=None, - stack_trace=None, - ) - - -def test_error_object_for_missing_error_takes_custom_message(): - """A caller-supplied message replaces the default text.""" - error = ErrorObject.for_missing_error(StepError, "Custom error message") - - assert error.message == "Custom error message" - assert error.type == "aws_durable_execution_sdk_python.exceptions.StepError" - - def test_invocation_output_create_pending_is_bare_pending(): """A pending response contains neither a result nor an error.""" output = DurableExecutionInvocationOutput.create_pending() 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 b1c8dadb7..128aa33cd 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -381,30 +381,6 @@ def test_checkpointed_result_is_started(): assert result_no_op.is_started() is False -def test_checkpointed_result_to_error_object_returns_checkpoint_error(): - """to_error_object preserves an ErrorObject already on the checkpoint.""" - error = ErrorObject( - message="Test error", type="ValueError", data="d", stack_trace=["l"] - ) - result = CheckpointedResult(error=error) - - assert result.to_error_object(StepError) is error - - -def test_checkpointed_result_to_error_object_without_error(): - """to_error_object represents the typed error direct replay would raise.""" - result = CheckpointedResult() - - error = result.to_error_object(StepError) - - assert error == ErrorObject( - message="Unknown error. No ErrorObject exists on the Checkpoint Operation.", - type="aws_durable_execution_sdk_python.exceptions.StepError", - data=None, - stack_trace=None, - ) - - def test_checkpointed_result_raise_operation_error(): """raise_operation_error wraps the checkpoint error in the given operation type.""" error = ErrorObject( @@ -428,14 +404,9 @@ def test_checkpointed_result_raise_operation_error_no_error(): """raise_operation_error with no error raises the operation type with a default message.""" result = CheckpointedResult() - with pytest.raises(StepError, match="Unknown error") as exc_info: + with pytest.raises(StepError, match="Unknown error"): result.raise_operation_error(StepError) - assert exc_info.value.error_type == ( - "aws_durable_execution_sdk_python.exceptions.StepError" - ) - assert exc_info.value.__cause__ is None - def test_checkpointed_result_raise_operation_error_no_error_with_message(): """raise_operation_error with no error uses the provided custom message.""" From 81ee0f994ff8ebce265c0e0098ed70ddd2ca2fa9 Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Sat, 24 Oct 2026 23:29:15 +0000 Subject: [PATCH 6/6] docs: leave the 1.x to 2.0 guide describing 2.0 The get_errors() example there was correct for 2.0, where every entry was an ErrorObject. The nullable entry is a 3.0 change, so it belongs in the 2.x to 3.0 migration guide, not here. --- docs/migration-1.x-to-2.0.md | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/docs/migration-1.x-to-2.0.md b/docs/migration-1.x-to-2.0.md index b041c6909..cd9fb64ef 100644 --- a/docs/migration-1.x-to-2.0.md +++ b/docs/migration-1.x-to-2.0.md @@ -348,18 +348,8 @@ result = context.map(items, process_item) try: result.throw_if_error() except (ChildContextError, SerDesError, BatchCompletionError): - for item in result.failed(): - if item.error is None: - context.logger.error( - "Batch item %s failed without an ErrorObject", - item.index, - ) - else: - context.logger.error( - "%s: %s", - item.error.type, - item.error.message, - ) + for err in result.get_errors(): # every failed item's ErrorObject + context.logger.error("%s: %s", err.type, err.message) ``` ## Serialize and Deserialize Round Trip