diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/validators/checkpoint.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/validators/checkpoint.py index f5ae3acff..2592b42cb 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/validators/checkpoint.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/validators/checkpoint.py @@ -51,7 +51,7 @@ from aws_durable_execution_sdk_python_testing.execution import Execution -MAX_ERROR_PAYLOAD_SIZE_BYTES = 32768 +MAX_ERROR_PAYLOAD_SIZE_BYTES = 1048576 class CheckpointValidator: @@ -120,9 +120,11 @@ def _validate_payload_sizes(update: OperationUpdate, execution: Execution) -> No execution returns to its parent, are each capped at 1 MiB. """ if update.error is not None: - payload = json.dumps(update.error.to_dict()) - if len(payload) > MAX_ERROR_PAYLOAD_SIZE_BYTES: - msg: str = f"Error object size must be less than {MAX_ERROR_PAYLOAD_SIZE_BYTES} bytes." + payload = json.dumps( + update.to_dict(), separators=(",", ":"), ensure_ascii=False + ) + if payload_size_bytes(payload) > MAX_ERROR_PAYLOAD_SIZE_BYTES: + msg: str = f"Error object size must be less than or equal to {MAX_ERROR_PAYLOAD_SIZE_BYTES} bytes." raise InvalidParameterValueException(msg) if update.payload is None: return diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/child_dispatcher.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/child_dispatcher.py index 39ce8c9ba..7bed8a10d 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/child_dispatcher.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/child_dispatcher.py @@ -136,7 +136,7 @@ def chained_invoke_timeout_error(timeout_seconds: int) -> ErrorObject: def payload_size_bytes(payload: str) -> int: """Size of ``payload`` as the service measures it: UTF-8 bytes.""" - return len(payload.encode("utf-8")) + return len(payload.encode("utf-8", errors="replace")) @dataclass(frozen=True) 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 3735c7d09..22da52618 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 @@ -616,7 +616,7 @@ def from_start_durable_execution_input( start_durable_execution_input: StartDurableExecutionInput, include: bool = False, # noqa: FBT001, FBT002 ) -> EventInput: - input: str | None = start_durable_execution_input.input + input: str | None = start_durable_execution_input.input if include else None truncated: bool = not include return cls(input, truncated) @@ -658,8 +658,11 @@ def from_durable_execution_invocation_output( durable_execution_invocation_output: DurableExecutionInvocationOutput, include: bool = False, # noqa: FBT001, FBT002 ) -> EventResult: + payload: str | None = ( + durable_execution_invocation_output.result if include else None + ) truncated: bool = not include - return cls(durable_execution_invocation_output.result, truncated) + return cls(payload, truncated) @dataclass(frozen=True) @@ -693,8 +696,9 @@ def from_details( include: bool = False, # noqa: FBT001, FBT002 ) -> EventError: error_object: ErrorObject | None = details.error if details else None + payload: ErrorObject | None = error_object if include else None truncated: bool = not include - return cls(error_object, truncated) + return cls(payload, truncated) @classmethod def from_durable_execution_invocation_output( @@ -702,8 +706,11 @@ def from_durable_execution_invocation_output( durable_execution_invocation_output: DurableExecutionInvocationOutput, include: bool = False, # noqa: FBT001, FBT002 ) -> EventError: + payload: ErrorObject | None = ( + durable_execution_invocation_output.error if include else None + ) truncated: bool = not include - return cls(durable_execution_invocation_output.error, truncated) + return cls(payload, truncated) @dataclass(frozen=True) @@ -1800,7 +1807,11 @@ def create_context_event_succeeded(cls, context: EventCreationContext) -> Event: def create_context_event_failed(cls, context: EventCreationContext) -> Event: context_details: ContextDetails | None = context.operation.context_details event_error: EventError | None = ( - EventError.from_details(context_details) if context_details else None + EventError.from_details( + context_details, include=context.include_execution_data + ) + if context_details + else None ) return cls( event_type=EventType.CONTEXT_FAILED.value, @@ -2232,7 +2243,11 @@ def create_callback_event_succeeded(cls, context: EventCreationContext) -> Event def create_callback_event_failed(cls, context: EventCreationContext) -> Event: callback_details: CallbackDetails | None = context.operation.callback_details event_error: EventError | None = ( - EventError.from_details(callback_details) if callback_details else None + EventError.from_details( + callback_details, include=context.include_execution_data + ) + if callback_details + else None ) return cls( event_type=EventType.CALLBACK_FAILED.value, @@ -2249,7 +2264,11 @@ def create_callback_event_failed(cls, context: EventCreationContext) -> Event: def create_callback_event_timed_out(cls, context: EventCreationContext) -> Event: callback_details: CallbackDetails | None = context.operation.callback_details event_error: EventError | None = ( - EventError.from_details(callback_details) if callback_details else None + EventError.from_details( + callback_details, include=context.include_execution_data + ) + if callback_details + else None ) return cls( event_type=EventType.CALLBACK_TIMED_OUT.value, diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/validators/checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/validators/checkpoint_test.py index 956ee34ed..ef5c8d896 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/validators/checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/validators/checkpoint_test.py @@ -19,6 +19,7 @@ ) from aws_durable_execution_sdk_python_testing.child_dispatcher import ( MAX_CHAINED_INVOKE_PAYLOAD_BYTES, + payload_size_bytes, ) from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, @@ -127,29 +128,110 @@ def test_validate_execution_update_as_last(): CheckpointValidator.validate_input(updates, execution) -def test_validate_payload_sizes_error_too_large(): - """Test validation fails when error payload is too large.""" +_ERROR_LIMIT_BYTES = 1048576 +_STEP_FAIL_ENVELOPE = ( + '{"Id":"step-1","Type":"STEP","Action":"FAIL","Error":{"ErrorMessage":""}}' +) +_ERROR_TOO_LARGE_MESSAGE = ( + f"Error object size must be less than or equal to {_ERROR_LIMIT_BYTES} bytes." +) + + +def _step_fail(message: str, name: str | None = None) -> OperationUpdate: + return OperationUpdate( + operation_id="step-1", + operation_type=OperationType.STEP, + action=OperationAction.FAIL, + name=name, + error=ErrorObject.from_message(message), + ) + + +def test_error_size_limit_is_one_mebibyte(): + assert MAX_ERROR_PAYLOAD_SIZE_BYTES == _ERROR_LIMIT_BYTES + + +def test_validate_payload_sizes_error_at_limit_is_accepted(): + """The whole update, serialized compactly, may be exactly the limit.""" execution = _create_test_execution() + message = "x" * (_ERROR_LIMIT_BYTES - len(_STEP_FAIL_ENVELOPE)) + CheckpointValidator.validate_input([_step_fail(message)], execution) + + +def test_validate_payload_sizes_error_over_limit_is_rejected(): + execution = _create_test_execution() + message = "x" * (_ERROR_LIMIT_BYTES - len(_STEP_FAIL_ENVELOPE) + 1) + with pytest.raises(InvalidParameterValueException) as exc_info: + CheckpointValidator.validate_input([_step_fail(message)], execution) + assert str(exc_info.value) == _ERROR_TOO_LARGE_MESSAGE + + +def test_validate_payload_sizes_error_multibyte_at_limit_is_accepted(): + """Non-ASCII is counted as raw UTF-8 bytes, not as escaped characters.""" + execution = _create_test_execution() + remaining = _ERROR_LIMIT_BYTES - len(_STEP_FAIL_ENVELOPE) + message = "\u00e9" * (remaining // 2) + "x" * (remaining % 2) + assert len(message.encode("utf-8")) == remaining + CheckpointValidator.validate_input([_step_fail(message)], execution) + + +def test_validate_payload_sizes_error_multibyte_over_limit_is_rejected(): + """A message under the limit in characters is rejected when over it in bytes.""" + execution = _create_test_execution() + remaining = _ERROR_LIMIT_BYTES - len(_STEP_FAIL_ENVELOPE) + message = "\u00e9" * (remaining // 2 + 1) + assert len(message) < remaining + with pytest.raises(InvalidParameterValueException) as exc_info: + CheckpointValidator.validate_input([_step_fail(message)], execution) + assert str(exc_info.value) == _ERROR_TOO_LARGE_MESSAGE + - large_message = "x" * (MAX_ERROR_PAYLOAD_SIZE_BYTES + 1) - large_error = ErrorObject( - message=large_message, type="TestError", data=None, stack_trace=None +def test_validate_payload_sizes_error_unpaired_surrogate_is_accepted(): + """A lone surrogate counts as one byte instead of failing to encode.""" + execution = _create_test_execution() + CheckpointValidator.validate_input([_step_fail("No such file: \udcff")], execution) + assert payload_size_bytes("\udcff") == 1 + + +def test_validate_payload_sizes_error_counts_whole_update(): + """Fields outside the Error object count toward the limit.""" + execution = _create_test_execution() + name_field = ',"Name":""' + name = "n" * (_ERROR_LIMIT_BYTES - len(_STEP_FAIL_ENVELOPE)) + with pytest.raises(InvalidParameterValueException) as exc_info: + CheckpointValidator.validate_input([_step_fail("", name=name)], execution) + assert str(exc_info.value) == _ERROR_TOO_LARGE_MESSAGE + + within = name[: -len(name_field)] + CheckpointValidator.validate_input( + [_step_fail("", name=within)], _create_test_execution() ) - updates = [ - OperationUpdate( - operation_id="step-1", - operation_type=OperationType.STEP, - action=OperationAction.FAIL, - error=large_error, - ) - ] - with pytest.raises( - InvalidParameterValueException, - match=f"Error object size must be less than {MAX_ERROR_PAYLOAD_SIZE_BYTES} bytes", - ): - CheckpointValidator.validate_input(updates, execution) +@pytest.mark.parametrize( + ("operation_type", "action"), + [ + (OperationType.STEP, OperationAction.RETRY), + (OperationType.CONTEXT, OperationAction.FAIL), + (OperationType.WAIT, OperationAction.CANCEL), + (OperationType.CALLBACK, OperationAction.START), + (OperationType.CHAINED_INVOKE, OperationAction.START), + (OperationType.EXECUTION, OperationAction.FAIL), + ], +) +def test_validate_payload_sizes_error_limit_applies_to_every_type( + operation_type, action +): + execution = _create_test_execution() + update = OperationUpdate( + operation_id="op-1", + operation_type=operation_type, + action=action, + error=ErrorObject.from_message("x" * _ERROR_LIMIT_BYTES), + ) + with pytest.raises(InvalidParameterValueException) as exc_info: + CheckpointValidator.validate_input([update], execution) + assert str(exc_info.value) == _ERROR_TOO_LARGE_MESSAGE def test_validate_payload_sizes_error_within_limit(): diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py index cd2fd8743..578e69434 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py @@ -359,6 +359,7 @@ def test_create_context_failed(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_context_event(context) @@ -572,6 +573,7 @@ def test_create_step_failed(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_step_event(context) @@ -681,6 +683,7 @@ def test_create_chained_invoke_failed(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_chained_invoke_event(context) @@ -706,6 +709,7 @@ def test_create_chained_invoke_timed_out(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_chained_invoke_event(context) @@ -734,6 +738,7 @@ def test_create_chained_invoke_stopped(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_chained_invoke_event(context) @@ -842,6 +847,7 @@ def test_create_callback_failed(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_callback_event(context) @@ -867,6 +873,7 @@ def test_create_callback_timed_out(): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_callback_event(context) @@ -986,7 +993,7 @@ def test_event_error_from_details(): error_obj = ErrorObject.from_message("Test error") details = StepDetails(error=error_obj) - event_error = EventError.from_details(details) + event_error = EventError.from_details(details, include=True) assert event_error.payload.message == "Test error" @@ -1482,6 +1489,7 @@ def test_from_operation_finished_step_with_error(self): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_event_terminated(context) @@ -1615,6 +1623,7 @@ def test_from_operation_finished_callback_timed_out(self): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_event_terminated(context) @@ -1688,6 +1697,7 @@ def test_from_operation_finished_chained_invoke_stopped(self): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_event_terminated(context) @@ -1762,6 +1772,7 @@ def test_from_operation_finished_context_failed(self): execution_timeout_seconds=300, execution_retention_period_days=7, ), + include_execution_data=True, ) event = Event.create_event_terminated(context) @@ -2097,3 +2108,190 @@ def test_create_chained_invoke_event_started_omits_the_tenant_id(): assert event.chained_invoke_started_details is not None assert event.chained_invoke_started_details.tenant_id is None assert "TenantId" not in event.to_dict()["ChainedInvokeStartedDetails"] + + +# region error-redaction-tests +_REDACTION_START_INPUT = StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, +) + +_OPERATION_ERROR_CASES = [ + (OperationType.STEP, OperationStatus.FAILED, "step_details", "step_failed_details"), + ( + OperationType.CONTEXT, + OperationStatus.FAILED, + "context_details", + "context_failed_details", + ), + ( + OperationType.CHAINED_INVOKE, + OperationStatus.FAILED, + "chained_invoke_details", + "chained_invoke_failed_details", + ), + ( + OperationType.CHAINED_INVOKE, + OperationStatus.TIMED_OUT, + "chained_invoke_details", + "chained_invoke_timed_out_details", + ), + ( + OperationType.CHAINED_INVOKE, + OperationStatus.STOPPED, + "chained_invoke_details", + "chained_invoke_stopped_details", + ), + ( + OperationType.CALLBACK, + OperationStatus.FAILED, + "callback_details", + "callback_failed_details", + ), + ( + OperationType.CALLBACK, + OperationStatus.TIMED_OUT, + "callback_details", + "callback_timed_out_details", + ), +] + + +def _assert_error_redaction(error: EventError | None, include: bool) -> None: + """Assert the payload is present only when execution data is included.""" + assert error is not None + assert error.truncated is (not include) + if include: + assert error.payload is not None + assert error.payload.message == "boom" + else: + assert error.payload is None + assert "Payload" not in error.to_dict() + + +@pytest.mark.parametrize("include", [True, False]) +@pytest.mark.parametrize( + ("operation_type", "status", "details_attr", "event_details_attr"), + _OPERATION_ERROR_CASES, +) +def test_operation_error_event_redacts_payload_unless_included( + operation_type, status, details_attr, event_details_attr, include +): + operation = create_mock_operation("op-1", status=status) + operation.operation_type = operation_type + operation.end_timestamp = datetime.now(UTC) + details = Mock() + details.result = None + details.error = ErrorObject.from_message("boom") + setattr(operation, details_attr, details) + context = EventCreationContext.create( + operation=operation, + event_id=1, + durable_execution_arn="arn:test", + start_input=_REDACTION_START_INPUT, + include_execution_data=include, + ) + + event = Event.create_event_terminated(context) + + _assert_error_redaction(getattr(event, event_details_attr).error, include) + + +@pytest.mark.parametrize("include", [True, False]) +@pytest.mark.parametrize( + ("status", "event_details_attr"), + [ + (OperationStatus.FAILED, "execution_failed_details"), + (OperationStatus.TIMED_OUT, "execution_timed_out_details"), + (OperationStatus.STOPPED, "execution_stopped_details"), + ], +) +def test_execution_error_event_redacts_payload_unless_included( + status, event_details_attr, include +): + from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, + ) + + operation = create_mock_operation("exec-1", status=status) + operation.end_timestamp = datetime.now(UTC) + context = EventCreationContext.create( + operation=operation, + event_id=1, + durable_execution_arn="arn:test", + start_input=_REDACTION_START_INPUT, + result=DurableExecutionInvocationOutput( + status=InvocationStatus.FAILED, error=ErrorObject.from_message("boom") + ), + include_execution_data=include, + ) + + event = Event.create_execution_event(context) + + _assert_error_redaction(getattr(event, event_details_attr).error, include) + + +@pytest.mark.parametrize("include", [True, False]) +def test_event_error_factories_redact_payload_unless_included(include): + from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, + ) + + error = ErrorObject.from_message("boom") + output = DurableExecutionInvocationOutput( + status=InvocationStatus.FAILED, error=error + ) + + for event_error in ( + EventError.from_details(StepDetails(error=error), include=include), + EventError.from_durable_execution_invocation_output(output, include=include), + ): + _assert_error_redaction(event_error, include) + + +@pytest.mark.parametrize("include", [True, False]) +def test_event_result_factory_redacts_payload_unless_included(include): + from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, + ) + + output = DurableExecutionInvocationOutput( + status=InvocationStatus.SUCCEEDED, result="ok" + ) + + result = EventResult.from_durable_execution_invocation_output( + output, include=include + ) + + assert result.payload == ("ok" if include else None) + assert result.truncated is (not include) + + +@pytest.mark.parametrize("include", [True, False]) +def test_event_input_factory_redacts_payload_unless_included(include): + start_input = StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + input='{"test": "data"}', + ) + + event_input = EventInput.from_start_durable_execution_input( + start_input, include=include + ) + + assert event_input.payload == ('{"test": "data"}' if include else None) + assert event_input.truncated is (not include) + + +# endregion error-redaction-tests 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 89a5618dd..21993649e 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 @@ -2491,6 +2491,113 @@ def test_get_execution_history_invalid_marker(executor, mock_store): assert result.next_marker is None +def _history_executor(mock_scheduler, mock_invoker, mock_checkpoint_processor): + store = InMemoryExecutionStore() + executor = Executor(store, mock_scheduler, mock_invoker, mock_checkpoint_processor) + execution = Execution.new( + StartDurableExecutionInput( + account_id="123456789012", + function_name="test-function", + function_qualifier="$LATEST", + execution_name="test-execution", + execution_timeout_seconds=300, + execution_retention_period_days=7, + invocation_id="test-inv-id", + ) + ) + execution.start() + store.save(execution) + return executor, execution.durable_execution_arn + + +@pytest.mark.parametrize("include", [True, False]) +def test_get_execution_history_wait_cancelled_has_no_error( + mock_scheduler, mock_invoker, mock_checkpoint_processor, include +): + from aws_durable_execution_sdk_python.lambda_service import WaitOptions + + executor, arn = _history_executor( + mock_scheduler, mock_invoker, mock_checkpoint_processor + ) + executor._set_invocation_gate(arn, InvocationState.INVOKING) # noqa: SLF001 + token = CheckpointToken(execution_arn=arn, token_sequence=0).to_str() + token = executor.checkpoint_execution( + arn, + token, + [ + OperationUpdate( + operation_id="wait-1", + operation_type=OperationType.WAIT, + action=OperationAction.START, + wait_options=WaitOptions(wait_seconds=60), + ) + ], + ).checkpoint_token + executor.checkpoint_execution( + arn, + token, + [ + OperationUpdate( + operation_id="wait-1", + operation_type=OperationType.WAIT, + action=OperationAction.CANCEL, + error=ErrorObject.from_message("boom"), + ) + ], + ) + + events = executor.get_execution_history(arn, include_execution_data=include).events + + cancelled = [e for e in events if e.event_type == "WaitCancelled"] + assert len(cancelled) == 1 + assert cancelled[0].wait_cancelled_details is not None + assert cancelled[0].wait_cancelled_details.error is None + + +@pytest.mark.parametrize("include", [True, False]) +def test_get_execution_history_redacts_execution_result_unless_included( + mock_scheduler, mock_invoker, mock_checkpoint_processor, include +): + executor, arn = _history_executor( + mock_scheduler, mock_invoker, mock_checkpoint_processor + ) + executor.complete_execution(arn, "ok") + + events = executor.get_execution_history(arn, include_execution_data=include).events + + succeeded = [e for e in events if e.event_type == "ExecutionSucceeded"] + assert len(succeeded) == 1 + details = succeeded[0].execution_succeeded_details + assert details is not None + assert details.result is not None + assert details.result.payload == ("ok" if include else None) + assert details.result.truncated is (not include) + + +@pytest.mark.parametrize("include", [True, False]) +def test_get_execution_history_redacts_execution_error_unless_included( + mock_scheduler, mock_invoker, mock_checkpoint_processor, include +): + executor, arn = _history_executor( + mock_scheduler, mock_invoker, mock_checkpoint_processor + ) + executor.fail_execution(arn, ErrorObject.from_message("boom")) + + events = executor.get_execution_history(arn, include_execution_data=include).events + + failed = [e for e in events if e.event_type == "ExecutionFailed"] + assert len(failed) == 1 + details = failed[0].execution_failed_details + assert details is not None + assert details.error is not None + assert details.error.truncated is (not include) + if include: + assert details.error.payload is not None + assert details.error.payload.message == "boom" + else: + assert details.error.payload is None + + def test_checkpoint_execution(mock_scheduler, mock_invoker, mock_checkpoint_processor): """Checkpoint with an empty update list returns a non-null new_execution_state with an empty operations list and a freshly