Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Comment thread
nvasiu marked this conversation as resolved.
)
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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Comment thread
nvasiu marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The latest revision changes EventInput.from_start_durable_execution_input(), but no repository code or test calls this method. So all tests still pass if someone removes the new if include else None branch. CONTRIBUTING rule 4 requires each new branch to have a test that fails without it.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a test for this.

truncated: bool = not include
return cls(input, truncated)

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -693,17 +696,21 @@ 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(
cls,
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)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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():
Expand Down
Loading
Loading