diff --git a/packages/aws-durable-execution-sdk-python/README.md b/packages/aws-durable-execution-sdk-python/README.md index 0589e9a4e..dc84924b9 100644 --- a/packages/aws-durable-execution-sdk-python/README.md +++ b/packages/aws-durable-execution-sdk-python/README.md @@ -63,6 +63,12 @@ earlier invocation), or put it on the factory, which outlives every invocation keyed by execution ARN, and pruned by the owner, because the SDK will not tell the factory when an execution ends for good. +Invocation operation-map snapshots and terminal `OperationEndInfo` own a copy of +the error's mutable stack-trace list, including virtual child completions without +a checkpoint. A plugin may modify that list without changing the checkpointed +error, the child error raised to the caller, or a later snapshot. Plugins receiving +the same hook object still share that snapshot; it is not copied for each plugin. + A factory is an object with a method rather than a plain callable so the registration type can grow a second, optional member later -- a process-level flush on execution-environment shutdown, for example -- without a second breaking 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 43b578f36..6cd688e07 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 @@ -1805,7 +1805,7 @@ def on_child_context_end( start_time=None, end_time=now, status=status, - error=error, + error=_copy_error(error), is_replayed=is_replayed, ), ) @@ -1866,7 +1866,7 @@ def _end_info(self, operation: Operation) -> OperationEndInfo: end_time=operation.end_timestamp, result=_extract_result(operation), status=operation.status, - error=self._extract_error(operation), + error=_copy_error(self._extract_error(operation)), attempt=( operation.step_details.attempt if operation.step_details else None ), diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py new file mode 100644 index 000000000..85370634f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_error_snapshot_int_test.py @@ -0,0 +1,172 @@ +"""A terminal plugin cannot change the failed checkpoint replay raises.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import Mock, patch + +from aws_durable_execution_sdk_python.config import ChildConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.exceptions import ( + ChildContextError, + DurableOperationError, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + OperationEndInfo, +) +from aws_durable_execution_sdk_python.types import StepContext +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from tests.e2e.checkpoint_response_int_test import ( + create_mock_checkpoint_with_operations, +) +from tests.test_helpers import operation_id_sequence + + +def test_terminal_error_mutation_preserves_failure_on_handler_replay() -> None: + step_id = next(operation_id_sequence()) + terminal_errors: list[list[str]] = [] + invocation_errors: list[list[str]] = [] + executed: list[str] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_id == step_id: + assert info.error is not None + assert info.error.stack_trace is not None + terminal_errors.append(list(info.error.stack_trace)) + info.error.stack_trace.clear() + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + error = info.operations[step_id].error + assert error is not None + assert error.stack_trace is not None + invocation_errors.append(list(error.stack_trace)) + + class Factory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return MutatingPlugin() + + def charge(_context: StepContext) -> str: + executed.append("charged") + return "unexpected execution" + + def handler(_event: Any, context: DurableContext) -> str: + return context.step(charge, name="charge") + + wrapped = durable_execution(handler, plugins=[Factory()]) + stored_error = { + "ErrorType": "ValueError", + "ErrorMessage": "original failure", + "StackTrace": ["original frame"], + } + event: dict[str, Any] = { + "DurableExecutionArn": "test-arn/execution-1", + "CheckpointToken": "test-token", + "InitialExecutionState": { + "Operations": [ + { + "Id": "execution-1", + "Type": "EXECUTION", + "Status": "STARTED", + "ExecutionDetails": {"InputPayload": "{}"}, + }, + { + "Id": step_id, + "Type": "STEP", + "Status": "FAILED", + "SubType": "Step", + "Name": "charge", + "StepDetails": {"Error": stored_error}, + }, + ], + "NextMarker": "", + }, + "UpdatedOperationIds": [step_id], + "LocalRunner": True, + } + lambda_context = Mock() + lambda_context.aws_request_id = "request-1" + lambda_context.client_context = None + lambda_context.identity = None + lambda_context._epoch_deadline_time_in_ms = 0 # noqa: SLF001 + lambda_context.invoked_function_arn = "test-arn" + lambda_context.tenant_id = None + + with patch("aws_durable_execution_sdk_python.execution.LambdaClient") as client: + checkpoint, _calls = create_mock_checkpoint_with_operations() + client.initialize_client.return_value.checkpoint = checkpoint + first = wrapped(event, lambda_context) + event["UpdatedOperationIds"] = [] + lambda_context.aws_request_id = "request-2" + replay = wrapped(event, lambda_context) + + for result in (first, replay): + assert result["Status"] == "FAILED" + assert result["Error"]["ErrorType"] == ( + "aws_durable_execution_sdk_python.exceptions.StepError" + ) + assert result["Error"]["ErrorMessage"] == "original failure" + assert result["Error"]["StackTrace"] == ["original frame"] + assert terminal_errors == [["original frame"]] + assert invocation_errors == [["original frame"], ["original frame"]] + assert stored_error["StackTrace"] == ["original frame"] + assert executed == [] + + +def test_virtual_child_terminal_snapshot_preserves_raised_error() -> None: + original = DurableOperationError( + message="original failure", data="details", stack_trace=["original frame"] + ) + terminal_infos: list[OperationEndInfo] = [] + caught: list[ChildContextError] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "virtual-child": + assert info.error is not None + assert info.error.stack_trace is not None + terminal_infos.append(info) + info.error.stack_trace.clear() + + class Factory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return MutatingPlugin() + + def fail(_context: DurableContext) -> None: + raise original + + def handler(_event: Any, context: DurableContext) -> str: + try: + context.run_in_child_context( + fail, name="virtual-child", config=ChildConfig(is_virtual=True) + ) + except ChildContextError as error: + caught.append(error) + return "caught" + raise AssertionError("Expected the virtual child's original failure") + + with DurableFunctionTestRunner( + handler=durable_execution(handler, plugins=[Factory()]) + ) as runner: + result = runner.run(input="{}", execution_timeout=15) + + assert result.status.value == "SUCCEEDED" + assert result.result == '"caught"' + (terminal_info,) = terminal_infos + assert terminal_info.start_time is None + assert terminal_info.error is not None + assert terminal_info.error.stack_trace == [] + (raised,) = caught + assert raised.message == "original failure" + assert raised.error_type == original.error_type + assert raised.data == "details" + assert raised.stack_trace == ["original frame"] + assert original.stack_trace == ["original frame"] 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 cb277e4a5..c6399a7bb 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -9,10 +9,14 @@ from collections.abc import Iterator from copy import deepcopy from dataclasses import asdict, fields +from itertools import product from unittest.mock import MagicMock, patch from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, + ChainedInvokeDetails, + ContextDetails, DurableExecutionInvocationOutput, ErrorObject, InvocationStatus as ServiceInvocationStatus, @@ -582,6 +586,57 @@ def test_user_function_end_info(self): self.assertEqual(USER_FUNCTION_END_INFO.end_time, END_TS) self.assertEqual(USER_FUNCTION_END_INFO.error.message, "boom") + def test_terminal_snapshot_preserves_operation_fields(self) -> None: + operation = Operation( + operation_id="step-1", + operation_type=ServiceOperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type="CustomStep", + name="saved-step", + parent_id="parent-1", + start_timestamp=START_TS, + end_timestamp=END_TS, + step_details=StepDetails(attempt=2, result='"saved"', error=ERROR), + ) + + info = PluginExecutor(plugins=[])._end_info(operation) + + self.assertIs(type(info), OperationEndInfo) + self.assertIs(info.operation_type, OperationType.STEP) + self.assertEqual(info.operation_id, "step-1") + self.assertEqual(info.sub_type, "CustomStep") + self.assertEqual(info.name, "saved-step") + self.assertEqual(info.parent_id, "parent-1") + self.assertEqual(info.status, OperationStatus.SUCCEEDED) + self.assertEqual(info.result, '"saved"') + self.assertEqual(info.attempt, 2) + self.assertEqual(info.error, ERROR) + self.assertIsNot(info.error, ERROR) + self.assertEqual(info.start_time, START_TS) + self.assertEqual(info.end_time, END_TS) + self.assertFalse(info.is_replayed) + + def test_terminal_snapshot_preserves_absent_and_empty_errors(self) -> None: + for error in ( + None, + ERROR, + ErrorObject(message="boom", type="Error", data=None, stack_trace=[]), + ): + with self.subTest(error=error): + operation = Operation( + operation_id="step-1", + operation_type=ServiceOperationType.STEP, + status=OperationStatus.FAILED, + step_details=StepDetails(error=error), + ) + info = PluginExecutor(plugins=[])._end_info(operation) + self.assertEqual(info.error, error) + if error is not None: + self.assertIsNot(info.error, error) + assert info.error is not None + if error.stack_trace is not None: + self.assertIsNot(info.error.stack_trace, error.stack_trace) + # endregion Dataclass Tests @@ -1928,6 +1983,107 @@ def test_provider_is_released_when_the_run_scope_exits(self): self.assertIsNone(self.executor._operations_provider) # noqa: SLF001 + def test_plugin_error_mutations_do_not_change_checkpoint_or_later_hooks( + self, + ) -> None: + """Snapshots and terminal hooks must own their mutable error stack traces.""" + for operation_type, checkpoint_delivery in product( + ( + ServiceOperationType.STEP, + ServiceOperationType.CALLBACK, + ServiceOperationType.CHAINED_INVOKE, + ServiceOperationType.CONTEXT, + ), + (False, True), + ): + with self.subTest( + operation_type=operation_type, checkpoint_delivery=checkpoint_delivery + ): + error = ErrorObject( + message="saved error", + type="ExternalError", + data="details", + stack_trace=["original frame"], + ) + operation = Operation( + operation_id="op-1", + operation_type=operation_type, + status=OperationStatus.FAILED, + step_details=( + StepDetails(attempt=2, 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 + ), + ) + seen: list[list[str]] = [] + + class MutatingPlugin(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + snapshot_error = info.updated_operations["op-1"].error + assert snapshot_error is not None + assert snapshot_error.stack_trace is not None + snapshot_error.stack_trace.clear() + + def on_operation_end(self, info: OperationEndInfo) -> None: + assert info.error is not None + assert info.error.stack_trace is not None + seen.append(list(info.error.stack_trace)) + info.error.stack_trace.clear() + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + snapshot_error = info.operations["op-1"].error + assert snapshot_error is not None + assert snapshot_error.stack_trace is not None + seen.append(list(snapshot_error.stack_trace)) + + executor = PluginExecutor([plugin_factory(MutatingPlugin())]) + 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"], + ) + self.assertEqual(error.stack_trace, ["original frame"]) + self.assertEqual(seen, []) + # Both delivery paths must snapshot errors before hooks run. + if checkpoint_delivery: + executor.record_end_hooks([operation]) + executor.drain_end_hooks() + else: + executor.on_operation_update(operation) + # A later observation must not repeat the terminal hook. + executor.on_operation_update( + [operation], + operations={"op-1": operation}, + previous_operations={"op-1": operation}, + ) + executor.on_invocation_end( + DurableExecutionInvocationOutput( + status=ServiceInvocationStatus.SUCCEEDED, + ) + ) + + self.assertEqual(seen, [["original frame"]] * 2) + self.assertEqual(error.stack_trace, ["original frame"]) + class TestInvocationInfoCopySafety(unittest.TestCase): """Invocation infos carrying the lazy operation map must stay copyable. @@ -2193,6 +2349,7 @@ class TestPluginExecutorOnChildContextEnd(unittest.TestCase): def test_builds_terminal_operation_info(self): captured: list[OperationEndInfo] = [] + error = ErrorObject("boom", "Error", None, ["original frame"]) class _CapturingPlugin(_TrackingPlugin): def on_operation_end(self, info: OperationEndInfo) -> None: @@ -2214,7 +2371,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: executor.on_child_context_end( identifier, OperationStatus.FAILED, - error=ERROR, + error=error, is_replayed=True, ) @@ -2228,7 +2385,11 @@ def on_operation_end(self, info: OperationEndInfo) -> None: self.assertEqual(info.parent_id, "parent-1") self.assertEqual(info.name, "book-trip") self.assertEqual(info.status, OperationStatus.FAILED) - self.assertEqual(info.error, ERROR) + self.assertEqual(info.error, error) + assert info.error is not None + assert info.error.stack_trace is not None + info.error.stack_trace.clear() + self.assertEqual(error.stack_trace, ["original frame"]) self.assertTrue(info.is_replayed) self.assertIsNone(info.start_time) assert info.end_time is not None