From d18bba71ceb72dc08aea72999a1c53fc939997a2 Mon Sep 17 00:00:00 2001 From: Yashvardhan Thanvi Date: Thu, 15 Oct 2026 16:35:46 +0530 Subject: [PATCH 1/3] fix(testing): fail fast on invalid handler output The local runner retried every invalid invocation output up to five times, five seconds apart. The service fails the execution on the first such output, with an InvalidParameterValueException error, and retries only PENDING with no pending operations, up to three attempts. - Fail the execution at once for FAILED with a Result, SUCCEEDED with an Error, a missing or unknown Status, and an output that cannot be parsed. The error has ErrorType InvalidParameterValueException and the validation message as ErrorMessage. - Keep retrying PENDING with no pending operations, capped at three consecutive attempts. Other retries keep the five-attempt budget. - Parse handler output in InvokeResponse.from_output and from_payload, which raise InvalidInvocationOutputError, so the executor can tell an output it cannot parse from a failed invocation. --- .../docs/architecture.md | 13 +- .../exceptions.py | 4 + .../executor.py | 101 +++++++- .../invoker.py | 48 +++- .../e2e/invalid_invocation_output_test.py | 76 ++++++ .../tests/exceptions_test.py | 8 + .../tests/executor_checkpoint_test.py | 95 ++++++- .../tests/executor_test.py | 240 +++++++++++++----- .../tests/invoker_test.py | 99 ++++++++ 9 files changed, 595 insertions(+), 89 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py diff --git a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md index c46f076a1..0a33e4ec1 100644 --- a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md +++ b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md @@ -430,8 +430,8 @@ retry budget cannot be bypassed. ### Retry budget -A handler exception or an output validation failure increases the -execution's `consecutive_failed_invocation_attempts`. At +A handler exception increases the execution's +`consecutive_failed_invocation_attempts`. At `MAX_CONSECUTIVE_FAILED_ATTEMPTS` the execution is failed with the last observed error. Below that, a retry is scheduled through `_invoke_execution` after `RETRY_BACKOFF_SECONDS`. Any clean invocation @@ -439,6 +439,15 @@ resets the counter. A clean invocation is one that succeeded or returned PENDING. So a transient failure followed by a success does not count against future budget. +An invalid output is handled as the service handles it. A FAILED output +with a Result, a SUCCEEDED output with an Error, a missing or unknown +Status, or an output that cannot be parsed fails the execution at once. +The error has type `InvalidParameterValueException` and the validation +message. The only invalid output that is retried is PENDING with no +pending operations, because an SDK race that resolves on replay can +produce it. It counts against the same counter, with a cap of +`MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS`. + ### Invocation timeout `execution_timeout` and `invocation_timeout` are set on the runner diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py index 8d51e2f18..90917bfdc 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py @@ -109,6 +109,10 @@ class DurableFunctionsTestError(Exception): """Base class for testing errors.""" +class InvalidInvocationOutputError(DurableFunctionsTestError): + """The handler's output cannot be parsed as an invocation output.""" + + # endregion Testing diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index 44a01bc01..dd250fe87 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -54,6 +54,7 @@ ) from aws_durable_execution_sdk_python_testing.exceptions import ( IllegalStateException, + InvalidInvocationOutputError, InvalidParameterValueException, ResourceNotFoundException, ) @@ -126,6 +127,11 @@ class Executor(ExecutionObserver): # target and creating a durable child. Each takes milliseconds. QUICK_DISPATCH_WORKERS: int = 8 MAX_CONSECUTIVE_FAILED_ATTEMPTS: int = 5 + # The service fails an execution on an invalid invocation output at + # once, except PENDING with no pending operations. An SDK race that + # resolves on replay can produce that one, so it gets this many + # consecutive attempts. + MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS: int = 3 RETRY_BACKOFF_SECONDS: int = 5 # GetDurableExecutionState page-count bounds, mirroring the service # (default 100, hard cap 1000 operations per page). @@ -1540,8 +1546,8 @@ def _finish_invocation( ) -> None: """Apply a completed handler invocation. - Records the attempt, validates the response (completing or - retrying), resets the retry counter on a clean run, and releases + Records the attempt, validates the response (completing, failing + or retrying), resets the retry counter on a clean run, and releases the gate to PRE_INVOKE (re-invoking if a trigger was deferred) or COMPLETED. """ @@ -1574,11 +1580,20 @@ def _finish_invocation( logger.warning( "[%s] Lambda output validation failure: %s", execution_arn, e ) - error_obj = ErrorObject.from_exception(e) + error_obj = self._invalid_output_error(str(e)) # Release the gate before retry scheduling so the retry's # invoke finds PRE_INVOKE. self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) - self._retry_invocation(execution, error_obj) + # Only PENDING with no pending operations is retried; every + # other invalid output fails the execution at once. + if response.status is InvocationStatus.PENDING: + self._retry_invocation( + execution, + error_obj, + max_attempts=self.MAX_CONSECUTIVE_INVALID_PENDING_ATTEMPTS, + ) + else: + self._fail_workflow(execution_arn, error_obj) return # A clean invocation resets the retry counter. @@ -1616,6 +1631,36 @@ def _fail_invocation_not_found( self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) self._fail_workflow(execution_arn, error) + @staticmethod + def _invalid_output_error(message: str) -> ErrorObject: + """The error the service fails an execution with for an invalid output.""" + return ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + def _fail_after_invalid_output( + self, + execution_arn: str, + error: ErrorObject, + invocation_start: datetime, + invocation_end: datetime, + ) -> None: + """Fail the execution for an output the invoker could not parse.""" + execution = self._store.load(execution_arn) + execution.record_invocation_completion( + invocation_start, invocation_end, str(uuid.uuid4()) + ) + self._store.save(execution) + if execution.is_complete: + # A stop or timeout landed during the invocation and stands. + self._set_invocation_gate(execution_arn, InvocationState.COMPLETED) + return + self._set_invocation_gate(execution_arn, InvocationState.PRE_INVOKE) + self._fail_workflow(execution_arn, error) + def _retry_after_timeout( self, execution_arn: str, @@ -1721,6 +1766,30 @@ async def invoke() -> None: ) ) + except InvalidInvocationOutputError as err: + if self._closing: + return + # The handler ran, but the service cannot parse its output, + # so it fails the execution without invoking it again. + invocation_end = self._clock.now() + logger.warning( + "[%s] Lambda output validation failure: %s", execution_arn, err + ) + error_obj = self._invalid_output_error(str(err)) + await asyncio.wrap_future( + self._registry.submit( + execution_arn, + CallableTask( + lambda: self._fail_after_invalid_output( + execution_arn, + error_obj, + invocation_start, + invocation_end, + ) + ), + ) + ) + except asyncio.TimeoutError: if self._closing: return @@ -1803,22 +1872,26 @@ def _fail_workflow(self, execution_arn: str, error: ErrorObject | None): self.fail_execution(execution_arn, error) - def _retry_invocation(self, execution: Execution, error: ErrorObject): + def _retry_invocation( + self, + execution: Execution, + error: ErrorObject, + max_attempts: int | None = None, + ) -> None: """Handle retry logic or fail execution if retries exhausted. - Budget: ``MAX_CONSECUTIVE_FAILED_ATTEMPTS`` attempts - per ``RETRY_BACKOFF_SECONDS`` backoff. Increments the counter - BEFORE the threshold check so an attempt-N failure that - reaches the ceiling fails the execution rather than scheduling - a pointless retry. + Budget: ``max_attempts`` attempts, by default + ``MAX_CONSECUTIVE_FAILED_ATTEMPTS``, per ``RETRY_BACKOFF_SECONDS`` + backoff. Increments the counter BEFORE the threshold check so an + attempt-N failure that reaches the ceiling fails the execution + rather than scheduling a pointless retry. """ + if max_attempts is None: + max_attempts = self.MAX_CONSECUTIVE_FAILED_ATTEMPTS execution.consecutive_failed_invocation_attempts += 1 self._store.save(execution) - if ( - execution.consecutive_failed_invocation_attempts - >= self.MAX_CONSECUTIVE_FAILED_ATTEMPTS - ): + if execution.consecutive_failed_invocation_attempts >= max_attempts: # Budget exhausted — fail the execution with the last # observed error. self._fail_workflow( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 47e7ac5ed..1aeaf7d4d 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -21,6 +21,7 @@ ) from aws_durable_execution_sdk_python_testing.exceptions import ( DurableFunctionsTestError, + InvalidInvocationOutputError, InvalidParameterValueException, ResourceNotFoundException, ) @@ -107,6 +108,36 @@ class InvokeResponse: invocation_output: DurableExecutionInvocationOutput request_id: str + @classmethod + def from_output(cls, output: Any, request_id: str) -> InvokeResponse: + """Create from the value a handler returned. + + Raises: + InvalidInvocationOutputError: If the value is not an object with + a known ``Status`` and, if present, an object ``Error``. + """ + try: + invocation_output = DurableExecutionInvocationOutput.from_dict(output) + except (AttributeError, TypeError, ValueError) as e: + msg: str = f"Invalid invocation output: {e}" + raise InvalidInvocationOutputError(msg) from e + return cls(invocation_output=invocation_output, request_id=request_id) + + @classmethod + def from_payload(cls, payload: str, request_id: str) -> InvokeResponse: + """Create from the JSON payload a Lambda invocation returned. + + Raises: + InvalidInvocationOutputError: If the payload is not JSON or not + a valid invocation output. + """ + try: + output: Any = json.loads(payload) + except ValueError as e: + msg: str = f"Invalid invocation output: {e}" + raise InvalidInvocationOutputError(msg) from e + return cls.from_output(output, request_id) + # Values the in-process Lambda context reports when the caller gives none. DEFAULT_TEST_REGION = "us-west-2" @@ -261,10 +292,7 @@ def invoke( ) handler: Callable = self._resolve_handler(function_name) response_dict = handler(input_with_client, context) - output = DurableExecutionInvocationOutput.from_dict(response_dict) - return InvokeResponse( - invocation_output=output, request_id=context.aws_request_id - ) + return InvokeResponse.from_output(response_dict, context.aws_request_id) def update_endpoint(self, endpoint_url: str, region_name: str) -> None: """No-op for in-process invoker.""" @@ -459,6 +487,7 @@ def invoke( Raises: ResourceNotFoundException: If function does not exist InvalidParameterValueException: If parameters are invalid + InvalidInvocationOutputError: If the function's output cannot be parsed DurableFunctionsTestError: For other invocation failures """ @@ -496,9 +525,8 @@ def invoke( msg = f"Lambda invocation failed with status {status_code}: {error_payload}" raise DurableFunctionsTestError(msg) - # Parse response payload + # Read response payload response_payload = response["Payload"].read().decode("utf-8") - response_dict = json.loads(response_payload) # Extract request ID from response headers (x-amzn-RequestId or x-amzn-request-id) headers = response.get("ResponseMetadata", {}).get("HTTPHeaders", {}) @@ -508,10 +536,6 @@ def invoke( or f"local-{uuid4()}" ) - # Convert to DurableExecutionInvocationOutput - output = DurableExecutionInvocationOutput.from_dict(response_dict) - return InvokeResponse(invocation_output=output, request_id=request_id) - except client.exceptions.ResourceNotFoundException as e: msg = f"Function not found: {function_name}" raise ResourceNotFoundException(msg) from e @@ -565,3 +589,7 @@ def invoke( raise DurableFunctionsTestError(msg) from e msg = f"Unexpected error during Lambda invocation: {e}" raise DurableFunctionsTestError(msg) from e + + # Parsed outside the try: an output the service cannot parse fails + # the execution, so it must not be reported as a failed invocation. + return InvokeResponse.from_payload(response_payload, request_id) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py new file mode 100644 index 000000000..02a661a91 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py @@ -0,0 +1,76 @@ +"""End-to-end handling of invalid invocation outputs through the test runner.""" + +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.lambda_service import ErrorObject + +from aws_durable_execution_sdk_python_testing.execution import ExecutionStatus +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +@pytest.mark.parametrize( + ("output", "message"), + [ + pytest.param( + {"Status": "FAILED", "Result": "result"}, + "Cannot provide a Result for FAILED status.", + id="failed-with-result", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": {"ErrorMessage": "boom"}}, + "Cannot provide an Error for SUCCEEDED status.", + id="succeeded-with-error", + ), + pytest.param( + {"Result": "result"}, + "Invalid invocation output: None is not a valid InvocationStatus", + id="missing-status", + ), + pytest.param( + {"Status": "DONE"}, + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + {"Status": "RETRY"}, + "Unexpected invocation status: InvocationStatus.RETRY", + id="internal-status", + ), + pytest.param( + ["SUCCEEDED"], + "Invalid invocation output: 'list' object has no attribute 'get'", + id="not-an-object", + ), + ], +) +def test_invalid_output_fails_the_execution_after_one_invocation( + output: Any, message: str +) -> None: + """The service fails the execution on the first invalid output, with an + InvalidParameterValueException error, and does not invoke again.""" + invocations: list[Any] = [] + + def handler(event: Any, context: Any) -> Any: # noqa: ARG001 + invocations.append(event) + return output + + with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner: + execution_arn = runner.run_async(input="input str") + result: DurableFunctionTestResult = runner.wait_for_result( + execution_arn, timeout=10 + ) + + assert len(invocations) == 1 + assert result.status is InvocationStatus.FAILED + assert result.execution_status is ExecutionStatus.FAILED + assert result.error == ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py index 1f1ea5509..1866fe150 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py @@ -25,6 +25,14 @@ def test_durable_functions_test_error_base_exception() -> None: assert isinstance(error, Exception) +def test_invalid_invocation_output_error() -> None: + """Test InvalidInvocationOutputError for handler outputs that cannot be parsed.""" + error = exceptions.InvalidInvocationOutputError("Invalid invocation output: bad") + + assert str(error) == "Invalid invocation output: bad" + assert isinstance(error, exceptions.DurableFunctionsTestError) + + def test_durable_functions_local_runner_error_base_exception() -> None: """Test DurableFunctionsLocalRunnerError base exception.""" error = exceptions.DurableFunctionsLocalRunnerError("Local runner error") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index c789dcd74..9dfe96f21 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -15,7 +15,10 @@ from unittest.mock import Mock import pytest -from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, +) from aws_durable_execution_sdk_python.lambda_service import ( ErrorObject, Operation as SvcOperation, @@ -31,8 +34,12 @@ from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) -from aws_durable_execution_sdk_python_testing.execution import Execution +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + ExecutionStatus, +) from aws_durable_execution_sdk_python_testing.executor import Executor, InvocationState +from aws_durable_execution_sdk_python_testing.invoker import InvokeResponse from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput from aws_durable_execution_sdk_python_testing.stores.memory import ( InMemoryExecutionStore, @@ -583,6 +590,90 @@ def test_exhausted_retries_fail_the_execution(): assert final.close_status.value == "FAILED" +def _invoke_response( + status: InvocationStatus, result: str | None = None +) -> InvokeResponse: + return InvokeResponse( + invocation_output=DurableExecutionInvocationOutput( + status=status, result=result + ), + request_id="req", + ) + + +def test_pending_without_pending_operations_fails_after_three_attempts(): + """PENDING with nothing pending is the one invalid output the service + retries, because an SDK race that a replay resolves can produce it. + It allows three consecutive attempts, then fails the execution with + an InvalidParameterValueException error.""" + store = InMemoryExecutionStore() + scheduler = Mock() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.return_value = _invoke_response(InvocationStatus.PENDING) + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, scheduler, invoker, Mock()) + + for attempt in (1, 2): + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + retrying = store.load(arn) + assert retrying.is_complete is False + assert retrying.consecutive_failed_invocation_attempts == attempt + assert scheduler.call_later.call_count == attempt + assert ( + scheduler.call_later.call_args.kwargs["delay"] + == Executor.RETRY_BACKOFF_SECONDS + ) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + + final = store.load(arn) + assert invoker.invoke.call_count == 3 + assert scheduler.call_later.call_count == 2 + assert final.close_status is ExecutionStatus.FAILED + assert final.result is not None + assert final.result.error == ErrorObject( + message="Cannot return PENDING status with no pending operations.", + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + +def test_pending_without_pending_operations_recovers_on_a_valid_output(): + """A retried PENDING followed by a valid output completes the + execution and resets the attempt counter.""" + store = InMemoryExecutionStore() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.side_effect = [ + _invoke_response(InvocationStatus.PENDING), + _invoke_response(InvocationStatus.SUCCEEDED, result="ok"), + ] + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, Mock(), invoker, Mock()) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + assert store.load(arn).consecutive_failed_invocation_attempts == 1 + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + final = store.load(arn) + assert final.close_status is ExecutionStatus.SUCCEEDED + assert final.result is not None + assert final.result.result == "ok" + assert final.consecutive_failed_invocation_attempts == 0 + + # endregion # region: earliest pending diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py index bb6b71f4f..41dab117d 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py @@ -181,6 +181,16 @@ def mock_execution(): return execution +def _invalid_output_error(message: str) -> ErrorObject: + """The error the service fails an execution with for an invalid output.""" + return ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) + + def test_init(mock_store, mock_scheduler, mock_invoker, mock_checkpoint_processor): # Test that Executor can be constructed with dependencies # Dependency injection is implementation detail - test behavior instead @@ -504,10 +514,10 @@ def test_should_ignore_response_when_execution_already_complete( mock_invoker.invoke.assert_not_called() -def test_should_retry_when_response_has_no_status( +def test_should_fail_when_response_has_no_status( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that invocation responses without status trigger retry logic.""" + """Test that invocation responses without status fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -530,27 +540,30 @@ def test_should_retry_when_response_has_no_status( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", _invalid_output_error("Response status is required") + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 -def test_should_retry_when_failed_response_has_result( +def test_should_fail_when_failed_response_has_result( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that failed responses with result trigger retry logic.""" + """Test that failed responses with result fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -575,27 +588,31 @@ def test_should_retry_when_failed_response_has_result( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Cannot provide a Result for FAILED status."), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 -def test_should_retry_when_success_response_has_error( +def test_should_fail_when_success_response_has_error( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that successful responses with error trigger retry logic.""" + """Test that successful responses with error fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -621,21 +638,25 @@ def test_should_retry_when_success_response_has_error( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Cannot provide an Error for SUCCEEDED status."), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 def test_should_retry_when_pending_response_has_no_operations( @@ -648,6 +669,7 @@ def test_should_retry_when_pending_response_has_no_operations( mock_execution.is_complete = False mock_execution.start_input = start_input mock_execution.consecutive_failed_invocation_attempts = 0 + mock_execution.seq_counter = 0 mock_execution.has_pending_operations.return_value = False # No pending operations mock_execution.handler_seen_seq = 0 mock_execution.has_changes_after.return_value = False @@ -1236,7 +1258,12 @@ def test_should_retry_invocation_when_under_limit_through_public_api( mock_execution.durable_execution_arn = "test-arn" mock_execution.is_complete = False mock_execution.start_input = start_input - mock_execution.consecutive_failed_invocation_attempts = 3 # Under limit (5 is max) + # Under limit (3 is max for PENDING with no pending operations) + mock_execution.consecutive_failed_invocation_attempts = 1 + mock_execution.seq_counter = 0 + mock_execution.has_pending_operations.return_value = False + mock_execution.handler_seen_seq = 0 + mock_execution.has_changes_after.return_value = False # Configure invoker to fail initially with validation error, then succeed on retry mock_invocation_input = Mock() @@ -1244,8 +1271,7 @@ def test_should_retry_invocation_when_under_limit_through_public_api( # First invocation: invalid response triggers retry invalid_response = DurableExecutionInvocationOutput( - status=InvocationStatus.FAILED, - result="should not have result", # Invalid: failed response with result + status=InvocationStatus.PENDING, # Invalid: PENDING with nothing pending ) # Second invocation: valid success response success_response = DurableExecutionInvocationOutput( @@ -1289,8 +1315,8 @@ def test_should_retry_invocation_when_under_limit_through_public_api( # Assert - verify final outcome after retry sequence # Counter resets to 0 on a clean invocation. The first - # attempt failed validation (3 -> 4), the second attempt succeeded - # (4 -> 0). Previously this was 4 because the counter only ever + # attempt failed validation (1 -> 2), the second attempt succeeded + # (2 -> 0). Previously this was 2 because the counter only ever # grew; that was a bug. assert mock_execution.consecutive_failed_invocation_attempts == 0 assert retry_delay == Executor.RETRY_BACKOFF_SECONDS # Correct backoff delay used @@ -1522,10 +1548,10 @@ def test_on_failed(executor): mock_fail.assert_called_once_with("test-arn", error) -def test_should_retry_when_response_has_unexpected_status( +def test_should_fail_when_response_has_unexpected_status( executor, mock_store, mock_scheduler, mock_invoker, start_input ): - """Test that responses with unexpected status trigger retry logic.""" + """Test that responses with unexpected status fail the execution without a retry.""" # Arrange mock_execution = Mock() mock_execution.durable_execution_arn = "test-arn" @@ -1549,21 +1575,113 @@ def test_should_retry_when_response_has_unexpected_status( mock_execution_class.new.return_value = mock_execution mock_store.load.return_value = mock_execution - # Act - trigger invocation through public start_execution method - executor.start_execution(start_input) + with patch.object(executor, "fail_execution") as mock_fail: + # Act - trigger invocation through public start_execution method + executor.start_execution(start_input) - # Get the handler that was passed to the scheduler and execute it manually - assert mock_scheduler.call_later.call_count >= 1 - handler = mock_scheduler.call_later.call_args_list[-1][0][0] + # Get the handler that was passed to the scheduler and execute it manually + assert mock_scheduler.call_later.call_count >= 1 + handler = mock_scheduler.call_later.call_args_list[-1][0][0] - # Execute the handler to trigger the invocation logic - asyncio.run(handler()) + # Execute the handler to trigger the invocation logic + asyncio.run(handler()) - # Assert - verify retry was triggered due to validation error - assert mock_execution.consecutive_failed_invocation_attempts == 1 - mock_store.save.assert_called_with(mock_execution) - # Verify retry was scheduled (call_later should be called 3 times: timeout + initial + retry) - assert mock_scheduler.call_later.call_count == 3 + # Assert - verify the execution failed at once, as the service does + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error("Unexpected invocation status: UNKNOWN_STATUS"), + ) + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # Verify no retry was scheduled (call_later is called 2 times: timeout + initial) + assert mock_scheduler.call_later.call_count == 2 + + +def test_invoke_handler_fails_execution_when_output_cannot_be_parsed( + executor, mock_store, mock_scheduler, mock_invoker, start_input +): + """An output the invoker cannot parse fails the execution at once, + without a retry, and the invocation is still recorded.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + mock_execution = Mock() + mock_execution.durable_execution_arn = "test-arn" + mock_execution.is_complete = False + mock_execution.start_input = start_input + mock_execution.consecutive_failed_invocation_attempts = 0 + + mock_invoker.create_invocation_input.return_value = Mock() + mock_invoker.invoke.side_effect = InvalidInvocationOutputError( + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)" + ) + + with patch( + "aws_durable_execution_sdk_python_testing.executor.Execution" + ) as mock_execution_class: + mock_execution_class.new.return_value = mock_execution + mock_store.load.return_value = mock_execution + + with patch.object(executor, "fail_execution") as mock_fail: + executor.start_execution(start_input) + handler = mock_scheduler.call_later.call_args_list[-1][0][0] + asyncio.run(handler()) + + mock_fail.assert_called_once_with( + "test-arn", + _invalid_output_error( + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)" + ), + ) + mock_execution.record_invocation_completion.assert_called_once() + assert mock_execution.consecutive_failed_invocation_attempts == 0 + # timeout + initial invocation, no retry + assert mock_scheduler.call_later.call_count == 2 + + +def test_invoke_handler_unparsed_output_leaves_a_stop_during_invocation( + executor, mock_store, mock_scheduler, mock_invoker, start_input +): + """A stop that lands while the handler runs stands when its output + cannot be parsed: the execution is not failed over it.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + incomplete_execution = Mock(spec=Execution) + incomplete_execution.is_complete = False + incomplete_execution.start_input = start_input + incomplete_execution.consecutive_failed_invocation_attempts = 0 + incomplete_execution.durable_execution_arn = "test-arn" + incomplete_execution.seq_counter = 0 + + stopped_execution = Mock(spec=Execution) + stopped_execution.is_complete = True + + # Never let the side_effect iterator exhaust: see + # test_invoke_handler_execution_completed_during_invocation_async. + _loads = iter([incomplete_execution]) + mock_store.load.side_effect = lambda *a, **k: next(_loads, stopped_execution) + + mock_invoker.create_invocation_input.return_value = Mock() + mock_invoker.invoke.side_effect = InvalidInvocationOutputError( + "Invalid invocation output: 'str' object has no attribute 'get'" + ) + + with patch( + "aws_durable_execution_sdk_python_testing.executor.Execution" + ) as mock_execution_class: + mock_execution_class.new.return_value = incomplete_execution + + with patch.object(executor, "fail_execution") as mock_fail: + executor.start_execution(start_input) + handler = mock_scheduler.call_later.call_args_list[-1][0][0] + asyncio.run(handler()) + + mock_fail.assert_not_called() + stopped_execution.record_invocation_completion.assert_called_once() + # No retry is scheduled either: timeout + initial invocation only. + assert mock_scheduler.call_later.call_count == 2 def test_invoke_handler_execution_completed_during_invocation_async( diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py index 14e6f2344..02d6e073b 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py @@ -271,6 +271,56 @@ def test_in_process_invoker_invoke(): assert isinstance(call_args[1], LambdaContext) +@pytest.mark.parametrize( + ("output", "message"), + [ + pytest.param( + {"Result": "result"}, + "Invalid invocation output: None is not a valid InvocationStatus", + id="missing-status", + ), + pytest.param( + {"Status": "DONE"}, + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + {"Status": "FAILED", "Error": "boom"}, + "Invalid invocation output: 'str' object has no attribute 'get'", + id="error-not-an-object", + ), + pytest.param( + "SUCCEEDED", + "Invalid invocation output: 'str' object has no attribute 'get'", + id="not-an-object", + ), + pytest.param( + None, + "Invalid invocation output: 'NoneType' object has no attribute 'get'", + id="none", + ), + ], +) +def test_in_process_invoker_rejects_an_output_it_cannot_parse(output, message): + """An output the service cannot parse raises InvalidInvocationOutputError, + which the executor fails the execution on instead of retrying.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + invoker = InProcessInvoker(Mock(return_value=output), Mock()) + input_data = DurableExecutionInvocationInput( + durable_execution_arn="test-arn", + checkpoint_token="test-token", # noqa: S106 + initial_execution_state=InitialExecutionState(operations=[], next_marker=""), + ) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", input_data) + + assert str(exc_info.value) == message + + def test_lambda_invoker_init(): """Test LambdaInvoker initialization.""" lambda_client = Mock() @@ -839,6 +889,55 @@ def test_lambda_invoker_invoke_function_error(): invoker.invoke("test-function", input_data) +@pytest.mark.parametrize( + ("payload", "message"), + [ + pytest.param( + b"not json", + "Invalid invocation output: Expecting value: line 1 column 1 (char 0)", + id="not-json", + ), + pytest.param( + b'{"Status": "DONE"}', + "Invalid invocation output: 'DONE' is not a valid InvocationStatus", + id="unknown-status", + ), + pytest.param( + b'["SUCCEEDED"]', + "Invalid invocation output: 'list' object has no attribute 'get'", + id="not-an-object", + ), + ], +) +def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): + """A payload the service cannot parse raises InvalidInvocationOutputError, + not the DurableFunctionsTestError of a failed invocation.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + lambda_client, _ = _create_mock_lambda_client_with_exceptions() + mock_payload = Mock() + mock_payload.read.return_value = payload + lambda_client.invoke.return_value = { + "StatusCode": 200, + "Payload": mock_payload, + "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "test-request-id"}}, + } + + invoker = LambdaInvoker(lambda_client) + input_data = DurableExecutionInvocationInput( + durable_execution_arn="test-arn", + checkpoint_token="test-token", # noqa: S106 + initial_execution_state=InitialExecutionState(operations=[], next_marker=""), + ) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", input_data) + + assert str(exc_info.value) == message + + def _create_mock_lambda_client_with_exceptions(): """Helper to create mock lambda client with all exception types.""" lambda_client = Mock() From 3d66c45651d8526e3ca7948bc340e04569e57f72 Mon Sep 17 00:00:00 2001 From: Yashvardhan Thanvi Date: Sat, 17 Oct 2026 01:00:04 +0530 Subject: [PATCH 2/3] fix(testing): keep the request ID for malformed handler output InvokeResponse.from_output and from_payload get the invocation's request ID, but dropped it when they raised InvalidInvocationOutputError. The executor then recorded the invocation completion under a new UUID, so a malformed-output failure could not be matched to the invocation that produced it. InvalidInvocationOutputError now carries an optional request_id, both factories set it, and the executor records it. An error raised without one still gets a generated ID. --- .../exceptions.py | 10 ++++++ .../executor.py | 5 ++- .../invoker.py | 4 +-- .../tests/exceptions_test.py | 8 +++++ .../tests/executor_checkpoint_test.py | 35 +++++++++++++++++++ .../tests/invoker_test.py | 2 ++ 6 files changed, 61 insertions(+), 3 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py index 90917bfdc..a243ec914 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/exceptions.py @@ -112,6 +112,16 @@ class DurableFunctionsTestError(Exception): class InvalidInvocationOutputError(DurableFunctionsTestError): """The handler's output cannot be parsed as an invocation output.""" + def __init__(self, message: str, request_id: str | None = None) -> None: + """Initialize with the message and the invocation's request ID. + + Args: + message: Why the output cannot be parsed + request_id: Request ID of the invocation that returned the output + """ + self.request_id = request_id + super().__init__(message) + # endregion Testing diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index dd250fe87..21d43da6e 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -1647,11 +1647,12 @@ def _fail_after_invalid_output( error: ErrorObject, invocation_start: datetime, invocation_end: datetime, + request_id: str | None, ) -> None: """Fail the execution for an output the invoker could not parse.""" execution = self._store.load(execution_arn) execution.record_invocation_completion( - invocation_start, invocation_end, str(uuid.uuid4()) + invocation_start, invocation_end, request_id or str(uuid.uuid4()) ) self._store.save(execution) if execution.is_complete: @@ -1776,6 +1777,7 @@ async def invoke() -> None: "[%s] Lambda output validation failure: %s", execution_arn, err ) error_obj = self._invalid_output_error(str(err)) + request_id = err.request_id await asyncio.wrap_future( self._registry.submit( execution_arn, @@ -1785,6 +1787,7 @@ async def invoke() -> None: error_obj, invocation_start, invocation_end, + request_id, ) ), ) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 1aeaf7d4d..8ba95815c 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -120,7 +120,7 @@ def from_output(cls, output: Any, request_id: str) -> InvokeResponse: invocation_output = DurableExecutionInvocationOutput.from_dict(output) except (AttributeError, TypeError, ValueError) as e: msg: str = f"Invalid invocation output: {e}" - raise InvalidInvocationOutputError(msg) from e + raise InvalidInvocationOutputError(msg, request_id) from e return cls(invocation_output=invocation_output, request_id=request_id) @classmethod @@ -135,7 +135,7 @@ def from_payload(cls, payload: str, request_id: str) -> InvokeResponse: output: Any = json.loads(payload) except ValueError as e: msg: str = f"Invalid invocation output: {e}" - raise InvalidInvocationOutputError(msg) from e + raise InvalidInvocationOutputError(msg, request_id) from e return cls.from_output(output, request_id) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py index 1866fe150..57a95366a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/exceptions_test.py @@ -30,8 +30,16 @@ def test_invalid_invocation_output_error() -> None: error = exceptions.InvalidInvocationOutputError("Invalid invocation output: bad") assert str(error) == "Invalid invocation output: bad" + assert error.request_id is None assert isinstance(error, exceptions.DurableFunctionsTestError) + error = exceptions.InvalidInvocationOutputError( + "Invalid invocation output: bad", request_id="request-1" + ) + + assert str(error) == "Invalid invocation output: bad" + assert error.request_id == "request-1" + def test_durable_functions_local_runner_error_base_exception() -> None: """Test DurableFunctionsLocalRunnerError base exception.""" diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index 9dfe96f21..bc1c24bd7 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -674,6 +674,41 @@ def test_pending_without_pending_operations_recovers_on_a_valid_output(): assert final.consecutive_failed_invocation_attempts == 0 +@pytest.mark.parametrize( + "parse", + [ + pytest.param( + lambda request_id: InvokeResponse.from_output(["SUCCEEDED"], request_id), + id="from-output", + ), + pytest.param( + lambda request_id: InvokeResponse.from_payload("not json", request_id), + id="from-payload", + ), + ], +) +def test_unparsed_output_keeps_the_request_id(parse): + """An output the invoker cannot parse fails the execution, and the + recorded invocation keeps the request ID of that invocation.""" + store = InMemoryExecutionStore() + invoker = Mock() + invoker.create_invocation_input.return_value = Mock() + invoker.invoke.side_effect = lambda *_args, **_kwargs: parse("request-1") + + execution = Execution.new(_make_start_input()) + execution.start() + store.save(execution) + arn = execution.durable_execution_arn + + executor = Executor(store, Mock(), invoker, Mock()) + + asyncio.run(executor._invoke_handler(arn)()) # noqa: SLF001 + + final = store.load(arn) + assert final.close_status is ExecutionStatus.FAILED + assert [c.request_id for c in final.invocation_completions] == ["request-1"] + + # endregion # region: earliest pending diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py index 02d6e073b..3b4edc0ad 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py @@ -319,6 +319,7 @@ def test_in_process_invoker_rejects_an_output_it_cannot_parse(output, message): invoker.invoke("test-function", input_data) assert str(exc_info.value) == message + assert exc_info.value.request_id == "test-invoke-12345" def test_lambda_invoker_init(): @@ -936,6 +937,7 @@ def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): invoker.invoke("test-function", input_data) assert str(exc_info.value) == message + assert exc_info.value.request_id == "test-request-id" def _create_mock_lambda_client_with_exceptions(): From f16a4107aca119b22a0271454bda56eaaafd2a8b Mon Sep 17 00:00:00 2001 From: Yashvardhan Thanvi Date: Sat, 17 Oct 2026 02:59:45 +0530 Subject: [PATCH 3/3] fix(testing): close the remaining invalid-output gaps Three kinds of output still got past the fail-fast path. - DurableExecutionInvocationOutput.from_dict treats any falsey Error as absent, so SUCCEEDED with an empty Error object completed the execution, and a falsey Error that is not an object ([], 0, false) was never checked. InvokeResponse.from_output now keeps any Error that is present and not null, so an empty object fails SUCCEEDED validation and a value that is not an object raises InvalidInvocationOutputError. - LambdaInvoker decoded the payload inside its try, so a payload that is not UTF-8 was reported as a failed invocation and retried. from_payload now takes the bytes and decodes them itself, so undecodable output raises InvalidInvocationOutputError with the request ID. - botocore returns the request ID as ResponseMetadata.RequestId and the header as x-amzn-requestid. The mixed-case header lookups missed both and fell back to a generated ID. The invoker now reads RequestId first, then the headers by lowercased name. --- .../invoker.py | 42 +++-- .../e2e/invalid_invocation_output_test.py | 87 ++++++++++ .../tests/executor_checkpoint_test.py | 6 +- .../tests/invoker_test.py | 161 +++++++++++++++++- 4 files changed, 276 insertions(+), 20 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 8ba95815c..4992212f4 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from threading import Lock from typing import TYPE_CHECKING, Any, Protocol from uuid import uuid4 @@ -15,6 +15,7 @@ DurableExecutionInvocationOutput, InitialExecutionState, ) +from aws_durable_execution_sdk_python.lambda_service import ErrorObject from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( DEFAULT_MAX_INVOCATION_PAGE_BYTES, @@ -118,21 +119,30 @@ def from_output(cls, output: Any, request_id: str) -> InvokeResponse: """ try: invocation_output = DurableExecutionInvocationOutput.from_dict(output) + # from_dict treats any falsey Error as absent, but only a missing + # or null Error is. An empty object is kept, so SUCCEEDED with it + # fails validation; a value that is not an object raises here. + error: Any = output.get("Error") + if error is not None and invocation_output.error is None: + invocation_output = replace( + invocation_output, error=ErrorObject.from_dict(error) + ) except (AttributeError, TypeError, ValueError) as e: msg: str = f"Invalid invocation output: {e}" raise InvalidInvocationOutputError(msg, request_id) from e return cls(invocation_output=invocation_output, request_id=request_id) @classmethod - def from_payload(cls, payload: str, request_id: str) -> InvokeResponse: - """Create from the JSON payload a Lambda invocation returned. + def from_payload(cls, payload: bytes, request_id: str) -> InvokeResponse: + """Create from the payload a Lambda invocation returned. Raises: - InvalidInvocationOutputError: If the payload is not JSON or not - a valid invocation output. + InvalidInvocationOutputError: If the payload is not UTF-8 JSON or + not a valid invocation output. """ try: - output: Any = json.loads(payload) + # UnicodeDecodeError is a ValueError. + output: Any = json.loads(payload.decode("utf-8")) except ValueError as e: msg: str = f"Invalid invocation output: {e}" raise InvalidInvocationOutputError(msg, request_id) from e @@ -525,13 +535,19 @@ def invoke( msg = f"Lambda invocation failed with status {status_code}: {error_payload}" raise DurableFunctionsTestError(msg) - # Read response payload - response_payload = response["Payload"].read().decode("utf-8") - - # Extract request ID from response headers (x-amzn-RequestId or x-amzn-request-id) - headers = response.get("ResponseMetadata", {}).get("HTTPHeaders", {}) - request_id = ( - headers.get("x-amzn-RequestId") + # Read response payload; from_payload decodes it, below + response_payload: bytes = response["Payload"].read() + + # botocore puts the request ID in ResponseMetadata and keeps the + # response headers with lowercased names. + metadata: dict[str, Any] = response.get("ResponseMetadata", {}) + headers: dict[str, str] = { + name.lower(): value + for name, value in metadata.get("HTTPHeaders", {}).items() + } + request_id: str = ( + metadata.get("RequestId") + or headers.get("x-amzn-requestid") or headers.get("x-amzn-request-id") or f"local-{uuid4()}" ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py index 02a661a91..9aea9c713 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invalid_invocation_output_test.py @@ -1,5 +1,7 @@ """End-to-end handling of invalid invocation outputs through the test runner.""" +from collections.abc import Callable +from pathlib import Path from typing import Any import pytest @@ -11,6 +13,16 @@ DurableFunctionTestResult, 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, +) +from aws_durable_execution_sdk_python_testing.stores.sqlite import ( + SQLiteExecutionStore, +) @pytest.mark.parametrize( @@ -74,3 +86,78 @@ def handler(event: Any, context: Any) -> Any: # noqa: ARG001 data=None, stack_trace=None, ) + + +@pytest.mark.parametrize( + "store_factory", + [ + pytest.param(lambda _tmp: InMemoryExecutionStore(), id="memory"), + pytest.param( + lambda tmp: FileSystemExecutionStore.create(tmp / "fs-store"), + id="filesystem", + ), + pytest.param( + lambda tmp: SQLiteExecutionStore.create_and_initialize( + str(tmp / "exec.db") + ), + id="sqlite", + ), + ], +) +@pytest.mark.parametrize( + ("error", "message"), + [ + pytest.param( + {}, + "Cannot provide an Error for SUCCEEDED status.", + id="empty-object", + ), + pytest.param( + [], + "Invalid invocation output: 'list' object has no attribute 'get'", + id="empty-list", + ), + pytest.param( + 0, + "Invalid invocation output: 'int' object has no attribute 'get'", + id="zero", + ), + pytest.param( + False, + "Invalid invocation output: 'bool' object has no attribute 'get'", + id="false", + ), + ], +) +def test_falsey_error_fails_a_succeeded_output_on_each_store( + store_factory: Callable[[Path], ExecutionStore], + error: Any, + message: str, + tmp_path: Path, +) -> None: + """The SDK reads any falsey Error as no Error. An empty Error object is + still an Error, and the other falsey values are not objects, so each + fails the execution after one invocation, on every store.""" + invocations: list[Any] = [] + + def handler(event: Any, context: Any) -> Any: # noqa: ARG001 + invocations.append(event) + return {"Status": "SUCCEEDED", "Error": error} + + with DurableFunctionTestRunner( + handler=handler, store=store_factory(tmp_path), execution_timeout=10 + ) as runner: + execution_arn = runner.run_async(input="input str") + result: DurableFunctionTestResult = runner.wait_for_result( + execution_arn, timeout=10 + ) + + assert len(invocations) == 1 + assert result.status is InvocationStatus.FAILED + assert result.execution_status is ExecutionStatus.FAILED + assert result.error == ErrorObject( + message=message, + type="InvalidParameterValueException", + data=None, + stack_trace=None, + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index bc1c24bd7..08edbac70 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -682,9 +682,13 @@ def test_pending_without_pending_operations_recovers_on_a_valid_output(): id="from-output", ), pytest.param( - lambda request_id: InvokeResponse.from_payload("not json", request_id), + lambda request_id: InvokeResponse.from_payload(b"not json", request_id), id="from-payload", ), + pytest.param( + lambda request_id: InvokeResponse.from_payload(b"\xff", request_id), + id="from-payload-not-utf-8", + ), ], ) def test_unparsed_output_keeps_the_request_id(parse): diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py index 3b4edc0ad..0286c1c00 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invoker_test.py @@ -15,6 +15,7 @@ ) from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, ExecutionDetails, Operation, OperationStatus, @@ -179,6 +180,20 @@ def handler(event, context): # noqa: ARG001 assert seen[0][1].endswith(":function:child:prod") +def _response_metadata(request_id: str) -> dict: + """ResponseMetadata as boto3 returns it: the request ID, and the + response headers with lowercased names.""" + return { + "RequestId": request_id, + "HTTPStatusCode": 200, + "HTTPHeaders": { + "content-type": "application/json", + "x-amzn-requestid": request_id, + }, + "RetryAttempts": 0, + } + + def _stub_success_client() -> Mock: client = Mock() payload = Mock() @@ -289,6 +304,21 @@ def test_in_process_invoker_invoke(): "Invalid invocation output: 'str' object has no attribute 'get'", id="error-not-an-object", ), + pytest.param( + {"Status": "SUCCEEDED", "Error": []}, + "Invalid invocation output: 'list' object has no attribute 'get'", + id="error-empty-list", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": 0}, + "Invalid invocation output: 'int' object has no attribute 'get'", + id="error-zero", + ), + pytest.param( + {"Status": "SUCCEEDED", "Error": False}, + "Invalid invocation output: 'bool' object has no attribute 'get'", + id="error-false", + ), pytest.param( "SUCCEEDED", "Invalid invocation output: 'str' object has no attribute 'get'", @@ -322,6 +352,27 @@ def test_in_process_invoker_rejects_an_output_it_cannot_parse(output, message): assert exc_info.value.request_id == "test-invoke-12345" +@pytest.mark.parametrize( + ("error", "expected"), + [ + pytest.param( + {}, ErrorObject(None, None, None, None), id="empty-object-is-an-error" + ), + pytest.param(None, None, id="null-is-no-error"), + ], +) +def test_in_process_invoker_keeps_an_empty_error_object(error, expected): + """An empty Error object is still an Error, so the executor rejects it on + SUCCEEDED; only a missing or null Error means there is none.""" + invoker = InProcessInvoker( + Mock(return_value={"Status": "SUCCEEDED", "Error": error}), Mock() + ) + + response = invoker.invoke("test-function", _minimal_input()) + + assert response.invocation_output.error == expected + + def test_lambda_invoker_init(): """Test LambdaInvoker initialization.""" lambda_client = Mock() @@ -541,7 +592,7 @@ def new_client(*_a, **_k): client.invoke.return_value = { "StatusCode": 200, "Payload": payload, - "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "r"}}, + "ResponseMetadata": _response_metadata("r"), } return client @@ -663,7 +714,7 @@ def test_lambda_invoker_invoke_success(): lambda_client.invoke.return_value = { "StatusCode": 200, "Payload": mock_payload, - "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "test-request-id"}}, + "ResponseMetadata": _response_metadata("test-request-id"), } invoker = LambdaInvoker(lambda_client) @@ -841,9 +892,7 @@ def test_lambda_invoker_invoke_status_202(): lambda_client.invoke.return_value = { "StatusCode": 202, "Payload": mock_payload, - "ResponseMetadata": { - "HTTPHeaders": {"x-amzn-RequestId": "test-request-id-202"} - }, + "ResponseMetadata": _response_metadata("test-request-id-202"), } invoker = LambdaInvoker(lambda_client) @@ -908,6 +957,12 @@ def test_lambda_invoker_invoke_function_error(): "Invalid invocation output: 'list' object has no attribute 'get'", id="not-an-object", ), + pytest.param( + b"\xff", + "Invalid invocation output: 'utf-8' codec can't decode byte 0xff" + " in position 0: invalid start byte", + id="not-utf-8", + ), ], ) def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): @@ -923,7 +978,7 @@ def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): lambda_client.invoke.return_value = { "StatusCode": 200, "Payload": mock_payload, - "ResponseMetadata": {"HTTPHeaders": {"x-amzn-RequestId": "test-request-id"}}, + "ResponseMetadata": _response_metadata("test-request-id"), } invoker = LambdaInvoker(lambda_client) @@ -940,6 +995,100 @@ def test_lambda_invoker_rejects_an_output_it_cannot_parse(payload, message): assert exc_info.value.request_id == "test-request-id" +@pytest.mark.parametrize( + ("metadata", "request_id"), + [ + pytest.param( + { + "RequestId": "from-metadata", + "HTTPHeaders": {"x-amzn-requestid": "from-header"}, + }, + "from-metadata", + id="response-metadata", + ), + pytest.param( + {"HTTPHeaders": {"x-amzn-requestid": "from-header"}}, + "from-header", + id="lowercase-header", + ), + pytest.param( + {"HTTPHeaders": {"X-Amzn-RequestId": "from-header"}}, + "from-header", + id="mixed-case-header", + ), + pytest.param( + {"HTTPHeaders": {"x-amzn-request-id": "from-header"}}, + "from-header", + id="request-id-header", + ), + ], +) +def test_lambda_invoker_reads_the_request_id_from_the_response(metadata, request_id): + """The request ID comes from ResponseMetadata, then from the response + headers, whatever the case of their names.""" + client = _stub_success_client() + client.invoke.return_value["ResponseMetadata"] = metadata + + response = LambdaInvoker(client).invoke("test-function", _minimal_input()) + + assert response.request_id == request_id + + +def _botocore_answers_invoke(monkeypatch, body: bytes) -> LambdaInvoker: + """A LambdaInvoker on a real client whose Invoke returns ``body``. + + The response, with header ``x-amzn-RequestId: request-1``, is + returned at ``before-send``, so botocore parses it as it would one + from the endpoint. + """ + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test") + client = create_lambda_client("http://localhost:3001", "us-west-2") + + def answer(request, **_kwargs): + return AWSResponse( + request.url, + 200, + {"Content-Type": "application/json", "x-amzn-RequestId": "request-1"}, + _RawBody(body), + ) + + client.meta.events.register("before-send.lambda.Invoke", answer) + return LambdaInvoker(client) + + +def test_lambda_invoker_keeps_the_request_id_botocore_parses(monkeypatch): + """The response carries the request ID the endpoint sent, not a generated one.""" + invoker = _botocore_answers_invoke(monkeypatch, b'{"Status": "SUCCEEDED"}') + + response = invoker.invoke("test-function", _minimal_input()) + + assert response.request_id == "request-1" + + +@pytest.mark.parametrize( + "body", + [ + pytest.param(b"not json", id="not-json"), + pytest.param(b'{"Status": "DONE"}', id="unknown-status"), + pytest.param(b"\xff", id="not-utf-8"), + ], +) +def test_lambda_invoker_rejects_with_the_request_id_botocore_parses(monkeypatch, body): + """An output that cannot be parsed, undecodable bytes included, raises + InvalidInvocationOutputError with the request ID the endpoint sent.""" + from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidInvocationOutputError, + ) + + invoker = _botocore_answers_invoke(monkeypatch, body) + + with pytest.raises(InvalidInvocationOutputError) as exc_info: + invoker.invoke("test-function", _minimal_input()) + + assert exc_info.value.request_id == "request-1" + + def _create_mock_lambda_client_with_exceptions(): """Helper to create mock lambda client with all exception types.""" lambda_client = Mock()