diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 58b21e793..cae654f2d 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -172,6 +172,20 @@ flow, pagination, and a map of the code. ### Error Handling +A callback failure with omitted diagnostics or an empty `ErrorObject` retains an +empty error object in SDK-facing operation state, in both memory and filesystem +stores. Detailed callback-failure history preserves `Error.Payload: {}` with +`Truncated: false`; history requested without execution data omits the payload and +sets `Truncated: true`. Reconstructing operations from detailed history preserves +that empty object. Missing or null errors remain `None` when decoded. Present +diagnostic fields remain intact, including explicit empty strings and empty +stack-trace lists. The callback still fails and raises the existing +`CallbackExternalError`, including its default message when diagnostics are absent. +Plugin operation information omits wholly empty diagnostics without changing the +stored error. Failed step, child-context, and invoke replay use their existing +default exception message when the stored error has no fields. Timeout history +retains the same data-inclusion rules. + The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see: - [Error Response Documentation](docs/error-responses.md) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 22da52618..052399a1c 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -675,7 +675,8 @@ class EventError: @classmethod def from_dict(cls, data: dict) -> EventError: payload = None - if payload_data := data.get("Payload"): + payload_data = data.get("Payload") + if payload_data is not None: payload = ErrorObject.from_dict(payload_data) return cls( diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/empty_callback_error_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/empty_callback_error_test.py new file mode 100644 index 000000000..52d9add1c --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/empty_callback_error_test.py @@ -0,0 +1,194 @@ +"""Real callback failure keeps runtime diagnostics and history consistent.""" + +from datetime import datetime +from pathlib import Path +import time +from typing import Any + +import pytest + +from aws_durable_execution_sdk_python import ( + DurableContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.exceptions import ( + CallbackError, + CallbackExternalError, +) +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + Operation, + OperationStatus, +) +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationInfo, +) +from aws_durable_execution_sdk_python_testing.execution import Execution +from aws_durable_execution_sdk_python_testing.model import ( + GetDurableExecutionHistoryResponse, + events_to_operations, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("error_kind", ["omitted", "empty", "rich"]) +def test_live_callback_failure_preserves_error_through_persistence( + filesystem: bool, + error_kind: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store: ExecutionStore = ( + FileSystemExecutionStore.create(tmp_path / "store") + if filesystem + else InMemoryExecutionStore() + ) + supplied_error = ( + ErrorObject( + message="rejected", + type="RejectedError", + data='{"reason":"test"}', + stack_trace=["original frame"], + ) + if error_kind == "rich" + else ErrorObject.from_dict({}) + ) + expected_error = supplied_error + completed_operations: list[Operation] = [] + original_complete = Execution.complete_callback_failure + + def capture_live_completion( + execution: Execution, + callback_id: str, + error: ErrorObject, + now: datetime | None = None, + ) -> Operation: + # Capture the real API transition before the executor persists it. + # The same error fields must survive both memory and filesystem storage. + operation = original_complete(execution, callback_id, error, now=now) + completed_operations.append(operation) + return operation + + monkeypatch.setattr(Execution, "complete_callback_failure", capture_live_completion) + step_calls: list[str] = [] + caught_errors: list[CallbackError] = [] + callback_infos: list[OperationInfo] = [] + + class ObserveCallback(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + callback_infos.extend( + operation + for operation in info.operations.values() + if operation.name == "failed-callback" + ) + + class ObserveFactory: + def create_plugin(self, _info: InvocationStartInfo) -> ObserveCallback: + return ObserveCallback() + + @durable_step + def saved_step(_context: Any) -> str: + step_calls.append("saved") + return "saved" + + def durable_handler(_event: Any, context: DurableContext) -> str: + saved = context.step(saved_step(), name="saved-step") + callback = context.create_callback(name="failed-callback") + try: + callback.result() + except CallbackError as error: + caught_errors.append(error) + return saved + raise AssertionError("Callback failure was not raised") + + handler = durable_execution(durable_handler, plugins=[ObserveFactory()]) + with DurableFunctionTestRunner( + handler=handler, + store=store, + skip_time=False, + execution_timeout=15, + poll_interval=0.01, + ) as runner: + arn = runner.run_async(input="{}") + callback_id = runner.wait_for_callback(arn, name="failed-callback", timeout=5) + deadline = time.monotonic() + 5 + while not store.load(arn).invocation_completions: + assert time.monotonic() < deadline, "Initial invocation did not suspend" + time.sleep(0.01) + if error_kind == "omitted": + runner.send_callback_failure(callback_id) + else: + runner.send_callback_failure(callback_id, supplied_error) + result = runner.wait_for_result(arn, timeout=10) + detailed = runner.get_execution_history(arn, include_execution_data=True) + metadata_only = runner.get_execution_history(arn, include_execution_data=False) + saved_execution = store.load(arn) + + assert result.status is InvocationStatus.SUCCEEDED + assert result.result == '"saved"' + assert step_calls == ["saved"] + assert len(saved_execution.invocation_completions) == 2 + assert len(caught_errors) == 1 + caught = caught_errors[0] + assert isinstance(caught, CallbackExternalError) + assert caught.message == (expected_error.message or "Callback failed") + assert caught.data == expected_error.data + assert caught.stack_trace == expected_error.stack_trace + assert callback_infos + for info in callback_infos: + assert info.status is OperationStatus.FAILED + assert info.error == (expected_error if error_kind == "rich" else None) + + assert len(completed_operations) == 1 + live = completed_operations[0] + persisted = next( + op for op in saved_execution.operations if op.operation_id == live.operation_id + ) + reloaded = Execution.from_json_dict(saved_execution.to_json_dict()) + roundtripped = next( + op for op in reloaded.operations if op.operation_id == live.operation_id + ) + for operation in (live, persisted, roundtripped): + assert operation.status is OperationStatus.FAILED + assert operation.callback_details is not None + assert operation.callback_details.error == expected_error + + for history, include_data in ((detailed, True), (metadata_only, False)): + wire = history.to_dict() + failed = next( + event for event in wire["Events"] if event["EventType"] == "CallbackFailed" + ) + expected: dict[str, Any] = {"Truncated": not include_data} + if include_data: + expected["Payload"] = expected_error.to_dict() + assert failed["CallbackFailedDetails"]["Error"] == expected + decoded = GetDurableExecutionHistoryResponse.from_dict(wire) + decoded_failed = next( + event for event in decoded.events if event.event_type == "CallbackFailed" + ) + assert decoded_failed.callback_failed_details is not None + assert decoded_failed.callback_failed_details.error is not None + assert decoded_failed.callback_failed_details.error.to_dict() == expected + for snapshot in (history, decoded): + reconstructed = next( + op + for op in events_to_operations(snapshot.events) + if op.operation_id == live.operation_id + ) + assert reconstructed.status is OperationStatus.FAILED + assert reconstructed.callback_details is not None + assert reconstructed.callback_details.error == ( + expected_error if include_data else None + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/empty_error_history_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/empty_error_history_test.py new file mode 100644 index 000000000..42dcbee08 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/empty_error_history_test.py @@ -0,0 +1,231 @@ +"""History and SDK operation state preserve empty error payload presence.""" + +from datetime import UTC, datetime +from typing import Any, cast + +import pytest + +from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, + ErrorObject, + Operation, + OperationStatus, + OperationType, +) +from aws_durable_execution_sdk_python_testing.model import ( + CallbackFailedDetails, + ChainedInvokeFailedDetails, + ContextFailedDetails, + Event, + EventCreationContext, + EventError, + GetDurableExecutionHistoryResponse, + GetDurableExecutionResponse, + StartDurableExecutionInput, + StepFailedDetails, + events_to_operations, +) +from aws_durable_execution_sdk_python_testing.runner import ( + CallbackOperation, + ContextOperation, + DurableFunctionTestResult, + InvokeOperation, + StepOperation, +) + + +RICH_ERROR = { + "ErrorType": "RejectedError", + "ErrorMessage": "rejected", + "ErrorData": '{"reason":"test"}', + "StackTrace": ["original frame"], +} +PAYLOADS = [ + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + RICH_ERROR, +] + + +@pytest.mark.parametrize("payload", PAYLOADS) +@pytest.mark.parametrize("truncated", [False, True]) +def test_event_error_roundtrip_preserves_present_payload( + payload: dict[str, Any], truncated: bool +) -> None: + wire = {"Payload": payload, "Truncated": truncated} + assert EventError.from_dict(wire).to_dict() == wire + + +@pytest.mark.parametrize("payload_field", [{}, {"Payload": None}]) +@pytest.mark.parametrize("truncated", [False, True]) +def test_event_error_omitted_and_null_payload_remain_absent( + payload_field: dict[str, Any], truncated: bool +) -> None: + parsed = EventError.from_dict({**payload_field, "Truncated": truncated}) + assert parsed.payload is None + assert parsed.to_dict() == {"Truncated": truncated} + + +@pytest.mark.parametrize( + "event_type,details_field,details_type,operation_field,wrapper_type", + [ + ( + "CallbackFailed", + "callback_failed_details", + CallbackFailedDetails, + "callback_details", + CallbackOperation, + ), + ( + "StepFailed", + "step_failed_details", + StepFailedDetails, + "step_details", + StepOperation, + ), + ( + "ContextFailed", + "context_failed_details", + ContextFailedDetails, + "context_details", + ContextOperation, + ), + ( + "ChainedInvokeFailed", + "chained_invoke_failed_details", + ChainedInvokeFailedDetails, + "chained_invoke_details", + InvokeOperation, + ), + ], +) +@pytest.mark.parametrize("payload", [None, *PAYLOADS]) +@pytest.mark.parametrize("decode_wire", [False, True]) +def test_failure_history_reconstruction_preserves_error_presence( + event_type: str, + details_field: str, + details_type: type[ + CallbackFailedDetails + | StepFailedDetails + | ContextFailedDetails + | ChainedInvokeFailedDetails + ], + operation_field: str, + wrapper_type: type[ContextOperation | InvokeOperation], + payload: dict[str, Any] | None, + decode_wire: bool, +) -> None: + timestamp = datetime(2026, 10, 27, tzinfo=UTC) + error = ErrorObject.from_dict(payload) if payload is not None else None + event_details: dict[str, Any] = { + details_field: details_type(error=EventError(payload=error, truncated=False)) + } + history = GetDurableExecutionHistoryResponse( + events=[ + Event( + event_type=event_type, + event_id=1, + event_timestamp=timestamp, + operation_id="operation", + name="failed-operation", + **event_details, + ) + ] + ) + if decode_wire: + expected_wire = history.to_dict() + history = GetDurableExecutionHistoryResponse.from_dict(expected_wire) + assert history.to_dict() == expected_wire + expected_error = error + (operation,) = events_to_operations(history.events) + assert operation.status is OperationStatus.FAILED + assert getattr(operation, operation_field).error == expected_error + if expected_error is not None: + roundtripped = Operation.from_json_dict(operation.to_json_dict()) + assert roundtripped.status is OperationStatus.FAILED + assert getattr(roundtripped, operation_field).error == expected_error + + execution = GetDurableExecutionResponse( + durable_execution_arn="arn:execution", + durable_execution_name="execution", + function_arn="arn:function", + status="FAILED", + start_timestamp=timestamp, + ) + result = DurableFunctionTestResult.from_execution_history(execution, history) + result_operation = cast( + ContextOperation | InvokeOperation, + result.get_operation_by_name("failed-operation"), + ) + assert type(result_operation) is wrapper_type + assert result_operation.error == expected_error + + +@pytest.mark.parametrize("include_data", [False, True]) +@pytest.mark.parametrize("status", [OperationStatus.FAILED, OperationStatus.TIMED_OUT]) +@pytest.mark.parametrize("payload", [None, {}, RICH_ERROR]) +def test_callback_history_projection_preserves_state_and_data_inclusion( + include_data: bool, status: OperationStatus, payload: dict[str, Any] | None +) -> None: + error = ErrorObject.from_dict(payload) if payload is not None else None + operation = Operation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=status, + callback_details=CallbackDetails(callback_id="callback-id", error=error), + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123456789012", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=30, + execution_retention_period_days=1, + ), + include_execution_data=include_data, + ) + event = Event.create_callback_event(context) + details = ( + event.callback_failed_details + if status is OperationStatus.FAILED + else event.callback_timed_out_details + ) + assert details is not None and details.error is not None + expected: dict[str, Any] = {"Truncated": not include_data} + if include_data and payload is not None: + expected["Payload"] = payload + assert details.error.to_dict() == expected + assert operation.callback_details is not None + assert operation.callback_details.error is error + + +def test_missing_callback_details_do_not_invent_error_payload() -> None: + operation = Operation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.FAILED, + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123456789012", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=30, + execution_retention_period_days=1, + ), + include_execution_data=True, + ) + event = Event.create_callback_event_failed(context) + assert event.callback_failed_details is not None + assert event.callback_failed_details.error is None 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 233812312..e43c400cf 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 @@ -209,7 +209,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> ContextDetails: return cls( replay_children=data.get("ReplayChildren", False), result=data.get("Result"), - error=ErrorObject.from_dict(error_raw) if error_raw else None, + error=ErrorObject.from_dict(error_raw) if error_raw is not None else None, ) @@ -346,7 +346,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> StepDetails: attempt=data.get("Attempt", 0), next_attempt_timestamp=data.get("NextAttemptTimestamp"), result=data.get("Result"), - error=ErrorObject.from_dict(error_raw) if error_raw else None, + error=ErrorObject.from_dict(error_raw) if error_raw is not None else None, ) @@ -371,7 +371,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> CallbackDetails: return cls( callback_id=data["CallbackId"], result=data.get("Result"), - error=ErrorObject.from_dict(error_raw) if error_raw else None, + error=ErrorObject.from_dict(error_raw) if error_raw is not None else None, ) @@ -385,7 +385,7 @@ def from_dict(cls, data: MutableMapping[str, Any]) -> ChainedInvokeDetails: error_raw = data.get("Error") return cls( result=data.get("Result"), - error=ErrorObject.from_dict(error_raw) if error_raw else None, + error=ErrorObject.from_dict(error_raw) if error_raw is not None else None, ) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py index 6cd688e07..d23b6aad9 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py @@ -73,14 +73,14 @@ def _extract_result(operation: Operation) -> str | None: def _extract_error(operation: Operation) -> ErrorObject | None: - if operation.step_details and operation.step_details.error: - return operation.step_details.error - if operation.callback_details and operation.callback_details.error: - return operation.callback_details.error - if operation.chained_invoke_details and operation.chained_invoke_details.error: - return operation.chained_invoke_details.error - if operation.context_details and operation.context_details.error: - return operation.context_details.error + for details in ( + operation.step_details, + operation.callback_details, + operation.chained_invoke_details, + operation.context_details, + ): + if details and details.error is not None and details.error.to_dict(): + return details.error return 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 a9883b3ea..ecf4e1b7f 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 @@ -335,7 +335,7 @@ def raise_operation_error( ensures async operations whose checkpoint carries a downstream error type (invoke/callback) still surface as the correct per-operation error. """ - if self.error is None: + if self.error is None or not self.error.to_dict(): err_msg: str = ( msg or "Unknown error. No ErrorObject exists on the Checkpoint Operation." 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 757036d75..975d5302a 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -1,8 +1,10 @@ """Tests for the service module.""" +import dataclasses import datetime import json from datetime import UTC +from typing import Any from unittest.mock import Mock, patch import pytest @@ -376,6 +378,93 @@ def test_step_details_minimal(): assert details.error is None +@pytest.mark.parametrize( + "details_type,required_fields", + [ + (CallbackDetails, {"CallbackId": "callback-id"}), + (StepDetails, {}), + (ContextDetails, {}), + (ChainedInvokeDetails, {}), + ], +) +@pytest.mark.parametrize( + "error_field", + [ + {}, + {"Error": None}, + {"Error": {}}, + {"Error": {"ErrorMessage": ""}}, + {"Error": {"ErrorType": ""}}, + {"Error": {"ErrorData": ""}}, + {"Error": {"StackTrace": []}}, + { + "Error": { + "ErrorMessage": "rejected", + "ErrorType": "RejectedError", + "ErrorData": '{"reason":"test"}', + "StackTrace": ["original frame"], + } + }, + ], +) +def test_operation_details_preserve_error_presence( + details_type: type[ + CallbackDetails | StepDetails | ContextDetails | ChainedInvokeDetails + ], + required_fields: dict[str, Any], + error_field: dict[str, Any], +) -> None: + details = details_type.from_dict({**required_fields, **error_field}) + payload = error_field.get("Error") + if payload is None: + assert details.error is None + else: + assert details.error is not None + assert details.error.to_dict() == payload + + +@pytest.mark.parametrize( + "operation_type,details_field,details", + [ + (OperationType.CALLBACK, "callback_details", CallbackDetails("callback-id")), + (OperationType.STEP, "step_details", StepDetails()), + (OperationType.CONTEXT, "context_details", ContextDetails()), + ( + OperationType.CHAINED_INVOKE, + "chained_invoke_details", + ChainedInvokeDetails(), + ), + ], +) +@pytest.mark.parametrize("json_roundtrip", [False, True]) +def test_operation_roundtrip_preserves_empty_error( + operation_type: OperationType, + details_field: str, + details: CallbackDetails | StepDetails | ContextDetails | ChainedInvokeDetails, + json_roundtrip: bool, +) -> None: + error = ErrorObject(message=None, type=None, data=None, stack_trace=None) + operation_details: dict[str, Any] = { + details_field: dataclasses.replace(details, error=error) + } + operation = Operation( + operation_id="failed-operation", + operation_type=operation_type, + status=OperationStatus.FAILED, + **operation_details, + ) + + if json_roundtrip: + restored = Operation.from_json_dict( + json.loads(json.dumps(operation.to_json_dict())) + ) + else: + restored = Operation.from_dict(operation.to_dict()) + + assert restored == operation + assert getattr(restored, details_field).error == error + + def test_wait_details_from_dict(): """Test WaitDetails.from_dict method.""" timestamp = datetime.datetime(2023, 1, 1, 12, 0, 0, tzinfo=datetime.UTC) diff --git a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py index c6399a7bb..ab3996335 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -12,6 +12,8 @@ from itertools import product from unittest.mock import MagicMock, patch +import pytest + from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( CallbackDetails, @@ -2691,6 +2693,129 @@ def test_extract_error_step_details_no_error(self): self.assertEqual(result.message, "boom") +@pytest.mark.parametrize( + "operation_type", + [ + ServiceOperationType.STEP, + ServiceOperationType.CALLBACK, + ServiceOperationType.CHAINED_INVOKE, + ServiceOperationType.CONTEXT, + ], +) +@pytest.mark.parametrize( + "payload", + [ + None, + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + { + "ErrorMessage": "stored failure", + "ErrorType": "ExternalError", + "ErrorData": "details", + "StackTrace": ["original frame"], + }, + ], +) +def test_plugin_error_projection_preserves_checkpoint_state( + operation_type: ServiceOperationType, payload: dict[str, str | list[str]] | None +) -> None: + error = ErrorObject.from_dict(payload) if payload is not None else None + operation = Operation( + operation_id="op-1", + operation_type=operation_type, + status=OperationStatus.FAILED, + step_details=( + StepDetails(error=error) + if operation_type is ServiceOperationType.STEP + else None + ), + callback_details=( + CallbackDetails(callback_id="callback-1", error=error) + if operation_type is ServiceOperationType.CALLBACK + else None + ), + chained_invoke_details=( + ChainedInvokeDetails(error=error) + if operation_type is ServiceOperationType.CHAINED_INVOKE + else None + ), + context_details=( + ContextDetails(error=error) + if operation_type is ServiceOperationType.CONTEXT + else None + ), + ) + checkpoint = deepcopy(operation.to_dict()) + observed: list[OperationInfo | OperationEndInfo] = [] + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + observed.extend([info.operations["op-1"], info.updated_operations["op-1"]]) + + def on_operation_end(self, info: OperationEndInfo) -> None: + observed.append(info) + + executor = PluginExecutor([plugin_factory(Observe())]) + with executor.run(): + executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=False, + operations_provider=lambda: {"op-1": operation}, + updated_operation_ids=["op-1"], + ) + executor.on_operation_update(operation) + + assert len(observed) == 3 + for info in observed: + assert info.status is OperationStatus.FAILED + assert info.error == (error if payload else None) + assert operation.to_dict() == checkpoint + if error is not None: + assert error.to_dict() == payload + + +def test_empty_wire_invoke_details_keep_plugin_operation_maps() -> None: + operation = Operation.from_dict( + { + "Id": "op-1", + "Type": "CHAINED_INVOKE", + "Status": "FAILED", + "ChainedInvokeDetails": {}, + } + ) + observed: list[OperationInfo] = [OperationInfo.from_operation(operation)] + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + observed.extend([info.operations["op-1"], info.updated_operations["op-1"]]) + + def on_operation_end(self, info: OperationEndInfo) -> None: + observed.append(info) + + executor = PluginExecutor([plugin_factory(Observe())]) + with executor.run(): + executor.on_invocation_start( + execution_arn="arn:exec", + lambda_context=LAMBDA_CTX, + execution_start_time=START_TS, + is_first_invocation=False, + operations_provider=lambda: {"op-1": operation}, + updated_operation_ids=["op-1"], + ) + executor.on_operation_update(operation) + + assert len(observed) == 4 + for info in observed: + assert info.operation_id == "op-1" + assert info.status is OperationStatus.FAILED + assert info.error is None + + class TestPluginExecutorIsTerminalStatus(unittest.TestCase): """Tests for PluginExecutor._is_terminal_status static method.""" 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 45c41e53c..d9860a3da 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -19,8 +19,11 @@ from aws_durable_execution_sdk_python.exceptions import ( BackgroundThreadError, + ChildContextError, DurableApiErrorCategory, + DurableOperationError, GetExecutionStateError, + InvokeError, NonDeterministicExecutionError, OrphanedChildException, StepError, @@ -417,6 +420,84 @@ def test_checkpointed_result_raise_operation_error_no_error_with_message(): result.raise_operation_error(StepError, "Custom error message") +@pytest.mark.parametrize( + "operation_type,error_cls", + [ + (OperationType.STEP, StepError), + (OperationType.CONTEXT, ChildContextError), + (OperationType.CHAINED_INVOKE, InvokeError), + ], +) +@pytest.mark.parametrize( + "payload", + [ + None, + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + { + "ErrorMessage": "stored failure", + "ErrorType": "ExternalError", + "ErrorData": "details", + "StackTrace": ["original frame"], + }, + ], +) +@pytest.mark.parametrize("message", [None, "Custom error message"]) +def test_checkpointed_error_uses_only_available_diagnostics( + operation_type: OperationType, + error_cls: type[DurableOperationError], + payload: dict[str, str | list[str]] | None, + message: str | None, +) -> None: + error = ErrorObject.from_dict(payload) if payload is not None else None + operation = Operation( + operation_id="failed-operation", + operation_type=operation_type, + status=OperationStatus.FAILED, + step_details=( + StepDetails(error=error) if operation_type is OperationType.STEP else None + ), + context_details=( + ContextDetails(error=error) + if operation_type is OperationType.CONTEXT + else None + ), + chained_invoke_details=( + ChainedInvokeDetails(error=error) + if operation_type is OperationType.CHAINED_INVOKE + else None + ), + ) + result = CheckpointedResult.create_from_operation(operation) + + with pytest.raises(error_cls) as raised: + result.raise_operation_error(error_cls, message) + + assert type(raised.value) is error_cls + if payload: + assert raised.value.message == payload.get("ErrorMessage") + assert raised.value.error_type == ( + payload.get("ErrorType") + or f"{error_cls.__module__}.{error_cls.__qualname__}" + ) + assert raised.value.data == payload.get("ErrorData") + assert raised.value.stack_trace == payload.get("StackTrace") + assert raised.value.__cause__ is not None + else: + assert raised.value.message == ( + message + or "Unknown error. No ErrorObject exists on the Checkpoint Operation." + ) + assert raised.value.__cause__ is None + assert result.error is error + assert result.status is OperationStatus.FAILED + if error is not None: + assert error.to_dict() == payload + + def test_checkpointed_result_immutable(): """Test that CheckpointedResult is immutable.""" result = CheckpointedResult(status=OperationStatus.SUCCEEDED)