diff --git a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json index 030fce1a..f74d2607 100644 --- a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json +++ b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json @@ -46,15 +46,15 @@ "path": "./src/step/step_with_retry.py" }, { - "name": "Step with Custom SerDes", - "description": "Step with a non-identity custom SerDes; the step returns the canonical round-tripped value, consistent across first run and replay", - "handler": "step_with_custom_serdes.handler", + "name": "Custom SerDes Round-Trip", + "description": "One function proving the first-run/replay result-equality guarantee with a non-identity SerDes across step, wait_for_condition, and run_in_child_context (normal, virtual, large-payload)", + "handler": "serdes_roundtrip.handler", "integration": true, "durableConfig": { "RetentionPeriodInDays": 7, "ExecutionTimeout": 300 }, - "path": "./src/step/step_with_custom_serdes.py" + "path": "./src/serdes_roundtrip/serdes_roundtrip.py" }, { "name": "Wait State", diff --git a/packages/aws-durable-execution-sdk-python-examples/src/serdes_roundtrip/serdes_roundtrip.py b/packages/aws-durable-execution-sdk-python-examples/src/serdes_roundtrip/serdes_roundtrip.py new file mode 100644 index 00000000..685f2fb7 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/src/serdes_roundtrip/serdes_roundtrip.py @@ -0,0 +1,128 @@ +"""Custom (non-identity) SerDes round-trip across every operation. + +Demonstrates the first-run / replay result-equality guarantee: with a +non-identity ``SerDes``, each durable operation returns the *round-tripped* +value - the value produced by ``serialize`` then ``deserialize`` - on the first +run, which is exactly the value a replay reconstructs from the checkpoint. +Returning the raw in-memory result on the first run would make the first run and +replay disagree whenever the serdes is not a perfect round-trip. + +A single deployed function exercises the guarantee across every operation that +checkpoints a serialized result, so there is no need to deploy a separate +example per operation: + +* ``step`` +* ``wait_for_condition`` +* ``run_in_child_context`` - normal, virtual, and large-payload (ReplayChildren) + +The shared ``MarkerSerDes`` strips a ``round_tripped`` marker on ``serialize`` +and re-adds it on ``deserialize``, so ``deserialize(serialize(x)) != x``. Every +operation's result therefore carries ``round_tripped=True`` only if the SDK +handed back the canonical round-tripped value rather than the raw one. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import ChildConfig, StepConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) + +# Results larger than this are not checkpointed in full; the child context +# switches to ReplayChildren mode (a compact summary is checkpointed and the +# child re-executes on replay). Kept in sync with the SDK constant. +CHECKPOINT_SIZE_LIMIT_BYTES = 256 * 1024 + + +class MarkerSerDes(SerDes[dict[str, Any]]): + """Non-identity serdes: ``deserialize`` re-adds a marker ``serialize`` strips. + + ``deserialize(serialize(value)) == {**value, "round_tripped": True}``, which + differs from ``value`` whenever ``value`` lacks the marker. That makes the + round-trip observable in the value each operation returns. + """ + + def serialize(self, value: dict[str, Any], _: SerDesContext) -> str: + payload = {k: v for k, v in dict(value).items() if k != "round_tripped"} + return json.dumps(payload) + + def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: + return {**json.loads(data), "round_tripped": True} + + +def _large_child_summary(_result: dict[str, Any]) -> str: + """Compact summary checkpointed in place of the large child result.""" + return json.dumps({"type": "large-child-result"}) + + +def _stop_immediately( + _state: dict[str, Any], _attempt: int +) -> WaitForConditionDecision: + """Meet the wait_for_condition on the first check.""" + return WaitForConditionDecision.stop_polling() + + +@durable_execution +def handler(_event: Any, context: DurableContext) -> dict[str, bool]: + """Run every operation with a non-identity serdes and report the round-trip. + + Each returned value carries ``round_tripped=True`` only because the SDK + returned ``deserialize(serialize(result))`` on the first run - the same + value the corresponding replay path produces. + """ + serdes = MarkerSerDes() + + step_result = context.step( + lambda _step_ctx: {"op": "step"}, + name="step", + config=StepConfig(serdes=serdes), + ) + + child_result = context.run_in_child_context( + lambda _child_ctx: {"op": "child"}, + name="child", + config=ChildConfig(serdes=serdes), + ) + + virtual_result = context.run_in_child_context( + lambda _child_ctx: {"op": "virtual-child"}, + name="virtual-child", + config=ChildConfig(serdes=serdes, is_virtual=True), + ) + + # A result larger than the checkpoint limit forces ReplayChildren mode: only + # the summary is checkpointed, yet the returned value is still the full + # round-tripped result. + large_blob = "x" * (CHECKPOINT_SIZE_LIMIT_BYTES + 1) + large_result = context.run_in_child_context( + lambda _child_ctx: {"op": "large-child", "blob": large_blob}, + name="large-child", + config=ChildConfig(serdes=serdes, summary_generator=_large_child_summary), + ) + + wait_for_condition_result = context.wait_for_condition( + check=lambda _state, _ctx: {"op": "wait-for-condition"}, + config=WaitForConditionConfig( + initial_state={}, + wait_strategy=_stop_immediately, + serdes=serdes, + ), + name="wait-for-condition", + ) + + # Only the marker booleans are returned (never the large blob) so the + # handler result stays small and easy to assert on. + return { + "step_round_tripped": step_result.get("round_tripped", False), + "child_round_tripped": child_result.get("round_tripped", False), + "virtual_child_round_tripped": virtual_result.get("round_tripped", False), + "large_child_round_tripped": large_result.get("round_tripped", False), + "wait_for_condition_round_tripped": wait_for_condition_result.get( + "round_tripped", False + ), + } diff --git a/packages/aws-durable-execution-sdk-python-examples/src/step/step_with_custom_serdes.py b/packages/aws-durable-execution-sdk-python-examples/src/step/step_with_custom_serdes.py deleted file mode 100644 index 7657c644..00000000 --- a/packages/aws-durable-execution-sdk-python-examples/src/step/step_with_custom_serdes.py +++ /dev/null @@ -1,65 +0,0 @@ -"""Step with a custom (non-identity) SerDes. - -Demonstrates that a step returns the *canonical, round-tripped* value — the -value produced by ``serialize`` then ``deserialize`` — rather than the raw -in-memory object the step function returned. Because a step's result is -deserialized from the checkpoint on replay, returning the round-tripped value -on the first run keeps the result identical across the first run and every -replay. - -Here the custom SerDes normalizes the order on the way to the checkpoint: it -persists only the canonical fields, sorts the tags, and drops a transient, -non-persisted field. The handler uses the step result immediately, so the value -observed on the first run is already the canonical form (sorted tags, transient -field dropped) — the same value a replay would deserialize from the checkpoint. -""" - -import json -from typing import Any - -from aws_durable_execution_sdk_python.config import StepConfig -from aws_durable_execution_sdk_python.context import DurableContext -from aws_durable_execution_sdk_python.execution import durable_execution -from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext - - -class OrderSerDes(SerDes[dict[str, Any]]): - """Persist a canonical order: keep stable fields, sort tags, drop transients. - - ``deserialize(serialize(order))`` is intentionally not the identity — the - transient field is dropped and the tags are sorted — so the difference - between the raw result and the round-tripped result is observable. - """ - - def serialize(self, value: dict[str, Any], _: SerDesContext) -> str: - canonical: dict[str, Any] = { - "order_id": value["order_id"], - "tags": sorted(value["tags"]), - } - return json.dumps(canonical) - - def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: - return json.loads(data) - - -@durable_execution -def handler(_event: Any, context: DurableContext) -> dict[str, Any]: - """Build an order in a step and use its round-tripped result immediately.""" - order = context.step( - lambda _: { - "order_id": "ORD-42", - "tags": ["priority", "gift", "fragile"], # unsorted on purpose - "transient_score": 0.99, # not persisted by OrderSerDes - }, - name="build_order", - config=StepConfig(serdes=OrderSerDes()), - ) - - # `order` is the canonical, round-tripped value even on the first run: - # tags are sorted and the transient field is gone. A replay would produce - # the exact same value by deserializing the checkpoint. - return { - "order_id": order["order_id"], - "tags": order["tags"], # canonical: sorted - "has_transient": "transient_score" in order, # False after round-trip - } diff --git a/packages/aws-durable-execution-sdk-python-examples/test/serdes_roundtrip/test_serdes_roundtrip.py b/packages/aws-durable-execution-sdk-python-examples/test/serdes_roundtrip/test_serdes_roundtrip.py new file mode 100644 index 00000000..d52f660a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/serdes_roundtrip/test_serdes_roundtrip.py @@ -0,0 +1,37 @@ +"""Tests for the combined custom-serdes round-trip example.""" + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus + +from src.serdes_roundtrip import serdes_roundtrip +from test.conftest import deserialize_operation_payload + + +@pytest.mark.example +@pytest.mark.durable_execution( + handler=serdes_roundtrip.handler, + lambda_function_name="Custom SerDes Round-Trip", +) +def test_serdes_roundtrip_all_operations(durable_runner): + """Every operation returns the canonical, round-tripped value on first run. + + With a non-identity serdes, each operation (step, wait_for_condition, and + every run_in_child_context variant - normal, virtual, large-payload) must + return ``deserialize(serialize(result))`` on the first run, so the returned + value matches what a replay reconstructs. The handler reports a boolean per + operation that is True only when the marker added by ``deserialize`` is + present. + """ + with durable_runner: + result = durable_runner.run(input="test", timeout=15) + + assert result.status is InvocationStatus.SUCCEEDED + + result_data = deserialize_operation_payload(result.result) + assert result_data == { + "step_round_tripped": True, + "child_round_tripped": True, + "virtual_child_round_tripped": True, + "large_child_round_tripped": True, + "wait_for_condition_round_tripped": True, + } diff --git a/packages/aws-durable-execution-sdk-python-examples/test/step/test_step_with_custom_serdes.py b/packages/aws-durable-execution-sdk-python-examples/test/step/test_step_with_custom_serdes.py deleted file mode 100644 index 8cc3a66b..00000000 --- a/packages/aws-durable-execution-sdk-python-examples/test/step/test_step_with_custom_serdes.py +++ /dev/null @@ -1,44 +0,0 @@ -"""Tests for the step-with-custom-serdes example.""" - -import pytest -from aws_durable_execution_sdk_python.execution import InvocationStatus - -from src.step import step_with_custom_serdes -from src.step.step_with_custom_serdes import OrderSerDes -from test.conftest import deserialize_operation_payload - - -@pytest.mark.example -@pytest.mark.durable_execution( - handler=step_with_custom_serdes.handler, - lambda_function_name="Step with Custom SerDes", -) -def test_step_with_custom_serdes(durable_runner): - """The handler returns the canonical, round-tripped step result. - - Even on the first run the step returns the value produced by - OrderSerDes.serialize then OrderSerDes.deserialize, so the result reflects - the canonical form (sorted tags, transient field dropped) rather than the - raw value the step function produced. This is the same value a replay would - deserialize from the checkpoint. - """ - with durable_runner: - result = durable_runner.run(input="test", timeout=10) - - assert result.status is InvocationStatus.SUCCEEDED - - result_data = deserialize_operation_payload(result.result) - assert result_data == { - "order_id": "ORD-42", - "tags": ["fragile", "gift", "priority"], # sorted by the serdes - "has_transient": False, # transient field dropped on serialize - } - - # The step checkpoint stores the canonical payload, and deserializing it - # yields exactly what the handler returned for the order fields. - step_result = result.get_step("build_order") - canonical = deserialize_operation_payload(step_result.result, OrderSerDes()) - assert canonical == { - "order_id": "ORD-42", - "tags": ["fragile", "gift", "priority"], - } diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py index 6871c5d3..86c944fb 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py @@ -28,6 +28,8 @@ ExecutionError, InvocationError, InvokeError, + RetryableSerDesError, + SerDesError, StepError, ValidationError, WaitForConditionError, @@ -55,6 +57,8 @@ "InvocationError", "InvokeError", "ParallelBranch", + "RetryableSerDesError", + "SerDesError", "StepContext", "StepError", "ValidationError", diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py index 46c23356..247fdce9 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py @@ -329,6 +329,16 @@ class WaitForConditionError(DurableOperationError): """Raised when a wait_for_condition operation fails.""" +class SerDesError(DurableOperationError): + """Raised when serializing or deserializing an operation result fails. + + Surfaced to user code so a serialization problem can be caught intentionally + (``except SerDesError``); a plain ``except StepError``/``ChildContextError`` + will not catch it. Signals a permanent failure - use + :class:`RetryableSerDesError` for a transient one that should retry. + """ + + class CallbackError(DurableOperationError): """Base class for callback operation failures; catches all of them. @@ -367,6 +377,7 @@ class CallbackSubmitterError(CallbackError): "CallbackExternalError": CallbackExternalError, "CallbackTimeoutError": CallbackTimeoutError, "CallbackSubmitterError": CallbackSubmitterError, + "SerDesError": SerDesError, } @@ -378,6 +389,22 @@ def __init__(self, message: str, step_id: str | None = None): self.step_id = step_id +class RetryableSerDesError(InvocationError): + """Raised by a SerDes to signal a transient failure (e.g. an offloading + serdes whose network call timed out). + + Fails the invocation so the backend retries, rather than surfacing to user + code. Use :class:`SerDesError` for a permanent failure. + """ + + def __init__( + self, + message: str, + termination_reason: TerminationReason = TerminationReason.SERIALIZATION_ERROR, + ): + super().__init__(message, termination_reason) + + class BackgroundThreadError(BaseException): """Critical error from background checkpoint thread. @@ -480,10 +507,6 @@ def __init__(self, message: str, source_exception: Exception | None = None) -> N self.source_exception: Exception | None = source_exception -class SerDesError(DurableExecutionsError): - """Raised when serialization fails.""" - - class OrphanedChildException(BaseException): """Raised when a child operation attempts to checkpoint after its parent context has completed. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index 1db42ec3..5efe7526 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -17,6 +17,7 @@ CheckpointError, DurableOperationError, GetExecutionStateError, + SerDesError, ) @@ -278,10 +279,10 @@ def raise_as_operation_error( Used by both the first-run terminal-failure path and replay, so the surfaced error is identical (a durable-execution determinism guarantee): - the wrapper is ``operation_error_cls`` carrying this object's - ``error_type``/``data``/``stack_trace``, and ``__cause__`` is the escaping - error rebuilt via the registry (a typed subclass when known, else the - base ``DurableOperationError``). + the wrapper is ``operation_error_cls`` (or ``SerDesError`` for a serdes + failure) carrying this object's ``error_type``/``data``/``stack_trace``, + and ``__cause__`` is the escaping error rebuilt via the registry (a typed + subclass when known, else the base ``DurableOperationError``). """ cause: DurableOperationError = DurableOperationError.from_error_fields( error_type=self.type, @@ -289,7 +290,12 @@ def raise_as_operation_error( data=self.data, stack_trace=self.stack_trace, ) - raise operation_error_cls( + # A serdes failure surfaces as SerDesError regardless of the operation + # kind, so it is catchable as itself on both first run and replay. + surfaced_cls: builtins.type[DurableOperationError] = ( + SerDesError if self.type == SerDesError.__name__ else operation_error_cls + ) + raise surfaced_cls( message=self.message, error_type=self.type, data=self.data, @@ -587,7 +593,7 @@ def create_context_start( def create_context_succeed( cls, identifier: OperationIdentifier, - payload: str, + payload: str | None, sub_type: OperationSubType, context_options: ContextOptions | None = None, ) -> OperationUpdate: @@ -649,7 +655,7 @@ def create_execution_fail(cls, error: ErrorObject) -> OperationUpdate: # region step @classmethod def create_step_succeed( - cls, identifier: OperationIdentifier, payload: str + cls, identifier: OperationIdentifier, payload: str | None ) -> OperationUpdate: """Create an instance of OperationUpdate for type: STEP, action: SUCCEED.""" return cls( @@ -751,7 +757,7 @@ def create_wait_for_condition_start( @classmethod def create_wait_for_condition_succeed( - cls, identifier: OperationIdentifier, payload: str + cls, identifier: OperationIdentifier, payload: str | None ) -> OperationUpdate: """Create an instance of OperationUpdate for type: STEP, action: SUCCEED.""" return cls( @@ -768,7 +774,7 @@ def create_wait_for_condition_succeed( def create_wait_for_condition_retry( cls, identifier: OperationIdentifier, - payload: str, + payload: str | None, next_attempt_delay_seconds: int, ) -> OperationUpdate: """Create an instance of OperationUpdate for type: STEP, action: RETRY.""" diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py index f4967338..f8ca3557 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/child.py @@ -134,6 +134,20 @@ def check_result_status(self) -> CheckResult[T]: # Ready to execute (checkpoint exists or was just created) return CheckResult.create_is_ready_to_execute(checkpointed_result) + def _deserialize_payload(self, serialized: str | None) -> T: + """Return the round-tripped value, so the first run matches replay. + + A None payload is returned as-is. + """ + if serialized is None: + return None # type: ignore[return-value] + return deserialize( + serdes=self.config.serdes, + data=serialized, + operation_id=self.operation_identifier.operation_id, + durable_execution_arn=self.state.durable_execution_arn, + ) + def execute(self, checkpointed_result: CheckpointedResult) -> T: """Execute child context function with error handling and large payload support. @@ -163,13 +177,30 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: ) raw_result: T = wrapped_user_func() + # Serialize once: used as the round-tripped return value in every + # mode, and as the checkpoint payload on the normal path. A custom + # serdes may serialize to None, which is handled below. + serialized_result: str | None = serialize( + serdes=self.config.serdes, + value=raw_result, + operation_id=self.operation_identifier.operation_id, + durable_execution_arn=self.state.durable_execution_arn, + ) + + # Round-trip before any SUCCEED checkpoint so a SUCCEEDED context is + # always reconstructable and the first run matches replay in every + # mode. A permanent serdes failure here fails before any SUCCEED is + # written. + return_value: T = self._deserialize_payload(serialized_result) + if self.is_virtual: logger.debug( "Virtual context: Exiting child context without creating another checkpoint. id: %s, name: %s", self.operation_identifier.operation_id, self.operation_identifier.name, ) - return raw_result + # Virtual contexts never checkpoint and re-execute on replay. + return return_value # If in replay_children mode, return without checkpointing if checkpointed_result.is_replay_children(): @@ -178,30 +209,19 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) - return raw_result - - # Serialize result - serialized_result: str = serialize( - serdes=self.config.serdes, - value=raw_result, - operation_id=self.operation_identifier.operation_id, - durable_execution_arn=self.state.durable_execution_arn, - ) - - # Check payload size and use ReplayChildren mode if needed - # Summary Generator Logic: - # When the serialized result exceeds 256KB, we use ReplayChildren mode to avoid - # checkpointing large payloads. Instead, we checkpoint a compact summary and mark - # the operation for replay. This matches the TypeScript implementation behavior. - # - # See TypeScript reference: - # - aws-durable-execution-sdk-js/src/handlers/run-in-child-context-handler/run-in-child-context-handler.ts (lines ~200-220) - # - # The summary generator creates a JSON summary with metadata (type, counts, status) - # instead of the full BatchResult. During replay, the child context is re-executed - # to reconstruct the full result rather than deserializing from the checkpoint. + # Large payloads re-execute on replay; the checkpoint stays + # small (summary only). + return return_value + + # Large results checkpoint a compact summary and use ReplayChildren + # so replay re-executes instead of deserializing. The returned value + # always uses the full serialized_result, never the summary. + payload_to_checkpoint: str | None = serialized_result replay_children: bool = False - if len(serialized_result) > CHECKPOINT_SIZE_LIMIT_BYTES: + if ( + serialized_result is not None + and len(serialized_result) > CHECKPOINT_SIZE_LIMIT_BYTES + ): logger.debug( "Large payload detected, using ReplayChildren mode: id: %s, name: %s, payload_size: %d, limit: %d", self.operation_identifier.operation_id, @@ -210,8 +230,8 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: CHECKPOINT_SIZE_LIMIT_BYTES, ) replay_children = True - # Use summary generator if provided, otherwise use empty string (matches TypeScript) - serialized_result = ( + # Use the summary generator if provided, otherwise an empty string. + payload_to_checkpoint = ( self.config.summary_generator(raw_result) if self.config.summary_generator else "" @@ -220,7 +240,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: # Checkpoint SUCCEED success_operation: OperationUpdate = OperationUpdate.create_context_succeed( identifier=self.operation_identifier, - payload=serialized_result, + payload=payload_to_checkpoint, sub_type=self.sub_type, context_options=ContextOptions(replay_children=replay_children), ) @@ -235,7 +255,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) - return raw_result # noqa: TRY300 + return return_value # noqa: TRY300 except SuspendExecution: # Don't checkpoint SuspendExecution - let it bubble up raise diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py index 28a6368f..5bb72f34 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/step.py @@ -12,6 +12,8 @@ from aws_durable_execution_sdk_python.exceptions import ( ExecutionError, InvalidStateError, + RetryableSerDesError, + SerDesError, StepError, StepInterruptedError, ) @@ -229,13 +231,29 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: ) raw_result: T = wrapped_user_func(step_context) - serialized_result: str = serialize( + # A custom serdes may serialize to None, which is handled below. + serialized_result: str | None = serialize( serdes=self.config.serdes, value=raw_result, operation_id=self.operation_identifier.operation_id, durable_execution_arn=self.state.durable_execution_arn, ) + # Round-trip before the SUCCEED checkpoint so a SUCCEEDED step is + # always reconstructable and the first run matches replay. A None + # payload is returned as-is, mirroring the replay path. A permanent + # serdes failure here fails before any SUCCEED is written. + return_value: T = ( + None # type: ignore[assignment] + if serialized_result is None + else deserialize( + serdes=self.config.serdes, + data=serialized_result, + operation_id=self.operation_identifier.operation_id, + durable_execution_arn=self.state.durable_execution_arn, + ) + ) + success_operation: OperationUpdate = OperationUpdate.create_step_succeed( identifier=self.operation_identifier, payload=serialized_result, @@ -251,17 +269,25 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) - # Return the round-tripped value so the first run matches replay, - # which reconstructs the result by deserializing the checkpoint. - # A None payload is returned as-is, mirroring the replay path. - if serialized_result is None: - return None # type: ignore[return-value] - return deserialize( # noqa: TRY300 - serdes=self.config.serdes, - data=serialized_result, - operation_id=self.operation_identifier.operation_id, - durable_execution_arn=self.state.durable_execution_arn, + return return_value + except RetryableSerDesError: + # Transient serdes failure: fail the invocation for backend retry, + # bypassing the step retry strategy. + raise + except SerDesError as e: + # Permanent serdes failure: terminal FAIL surfaced as SerDesError, + # without the step retry strategy. + logger.exception( + "❌ serdes failed for step id: %s, name: %s", + self.operation_identifier.operation_id, + self.operation_identifier.name, + ) + error_object = ErrorObject.from_exception(e) + fail_operation: OperationUpdate = OperationUpdate.create_step_fail( + identifier=self.operation_identifier, error=error_object ) + self.state.create_checkpoint(operation_update=fail_operation) + error_object.raise_as_operation_error(StepError) except Exception as e: if isinstance(e, ExecutionError): # No retry on fatal - e.g checkpoint exception diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py index 3e7bd900..6ad8d61e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/wait_for_condition.py @@ -213,7 +213,21 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: ) if not decision.should_continue: - # Condition is met - complete successfully + # Condition is met - complete successfully. + # Round-trip before the SUCCEED checkpoint so a SUCCEEDED + # operation is always reconstructable and the first run matches + # replay. A None payload is returned as-is. A permanent serdes + # failure here fails before any SUCCEED is written. + return_value: T = ( + None # type: ignore[assignment] + if serialized_state is None + else deserialize( + serdes=self.config.serdes, + data=serialized_state, + operation_id=self.operation_identifier.operation_id, + durable_execution_arn=self.state.durable_execution_arn, + ) + ) success_operation = OperationUpdate.create_wait_for_condition_succeed( identifier=self.operation_identifier, payload=serialized_state, @@ -228,7 +242,7 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T: self.operation_identifier.operation_id, self.operation_identifier.name, ) - return new_state + return return_value # Condition not met - schedule retry # We enforce a minimum delay second of 1, to match model behaviour. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/serdes.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/serdes.py index 2cf41200..15a81288 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/serdes.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/serdes.py @@ -35,7 +35,7 @@ from aws_durable_execution_sdk_python.concurrency.models import BatchResult from aws_durable_execution_sdk_python.exceptions import ( DurableExecutionsError, - ExecutionError, + RetryableSerDesError, SerDesError, ) @@ -462,13 +462,16 @@ def serialize( active_serdes: SerDes[T] = serdes or EXTENDED_TYPES_SERDES try: return active_serdes.serialize(value, serdes_context) + except RetryableSerDesError: + # Transient failure: propagate so it retries the invocation. + raise except Exception as e: logger.exception( "⚠️ Serialization failed for id: %s", operation_id, ) msg = f"Serialization failed for id: {operation_id}, error: {e}." - raise ExecutionError(msg) from e + raise SerDesError(msg) from e def deserialize( @@ -492,7 +495,10 @@ def deserialize( active_serdes: SerDes[T] = serdes or EXTENDED_TYPES_SERDES try: return active_serdes.deserialize(data, serdes_context) + except RetryableSerDesError: + # Transient failure: propagate so it retries the invocation. + raise except Exception as e: logger.exception("⚠️ Deserialization failed for id: %s", operation_id) msg = f"Deserialization failed for id: {operation_id}" - raise ExecutionError(msg) from e + raise SerDesError(msg) from e diff --git a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py index 0cf8f0ab..96199589 100644 --- a/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/concurrency_test.py @@ -1433,9 +1433,16 @@ def success_callable(): execution_state = Mock() execution_state.create_checkpoint = Mock() + # wrap_user_function must return the real function so the branch result is + # the (serializable) "result_N" string that the child round-trips. + child_context = Mock() + child_context.state.wrap_user_function = lambda func, *a, **k: func + executor_context = Mock() executor_context._create_step_id_for_logical_step = lambda *args: "1" - executor_context.create_child_context = lambda *args, **kwargs: Mock() + # Return the configured child_context so its pass-through wrap_user_function + # is used and the branch result is the serializable "result_N" string. + executor_context.create_child_context = lambda *args, **kwargs: child_context result = executor.execute(execution_state, executor_context) diff --git a/packages/aws-durable-execution-sdk-python/tests/context_test.py b/packages/aws-durable-execution-sdk-python/tests/context_test.py index 6f7b2b13..c5b8ceeb 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -1840,7 +1840,9 @@ def test_function(context, item, index, items): with patch( "aws_durable_execution_sdk_python.context.map_handler" ) as mock_map_handler: - mock_map_handler.return_value = Mock() + # The wrapping child context round-trips this result, so it must be + # serializable. + mock_map_handler.return_value = {"result": "value"} with patch.object(context, "run_in_child_context") as mock_run_in_child: # Set up the mock to call the nested function @@ -1926,7 +1928,9 @@ def test_callable_2(context): with patch( "aws_durable_execution_sdk_python.context.parallel_handler" ) as mock_parallel_handler: - mock_parallel_handler.return_value = Mock() + # The wrapping child context round-trips this result, so it must be + # serializable. + mock_parallel_handler.return_value = {"result": "value"} with patch.object(context, "run_in_child_context") as mock_run_in_child: # Set up the mock to call the nested function diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py index 08d74352..83f7c146 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/custom_serdes_roundtrip_int_test.py @@ -1,10 +1,14 @@ """Integration tests for the first-run/replay result equality guarantee. With a non-identity custom ``SerDes`` (serialize/deserialize is not a round-trip -identity), a step must return the *round-tripped* value on the first run so that -it matches the value returned on replay, where the result is deserialized from -the checkpoint. Returning the raw in-memory result on the first run would make -the first-run and replay results diverge. +identity), an operation must return the *round-tripped* value on the first run +so that it matches the value returned on replay, where the result is +deserialized from the checkpoint. Returning the raw in-memory result on the +first run would make the first-run and replay results diverge. + +This is exercised below for ``step``, ``wait_for_condition``, and +``run_in_child_context`` in all three modes: normal, virtual, and large-payload +(ReplayChildren). The serdes used here strips a marker key on ``serialize`` and re-adds it on ``deserialize`` so that ``deserialize(serialize(x)) != x``. That makes the bug @@ -15,7 +19,7 @@ from typing import Any from unittest.mock import Mock, patch -from aws_durable_execution_sdk_python.config import StepConfig +from aws_durable_execution_sdk_python.config import ChildConfig, StepConfig from aws_durable_execution_sdk_python.context import DurableContext from aws_durable_execution_sdk_python.execution import ( InvocationStatus, @@ -27,6 +31,10 @@ OperationAction, ) from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) class MarkerSerDes(SerDes[Any]): @@ -183,7 +191,251 @@ def handler(event, context: DurableContext) -> dict[str, Any]: assert replay_result["Status"] == InvocationStatus.SUCCEEDED.value replay_data = json.loads(replay_result["Result"]) - # The core invariant: first-run output equals replay output, and both carry - # the marker added by deserialize() (i.e. the round-tripped value, not raw). + # First run equals replay, and both carry the round-trip marker. + assert first_data == replay_data + assert first_data == {"order_id": "ORD-123", "round_tripped": True} + + +def test_child_first_run_matches_replay_with_non_identity_serdes(): + """First-run result equals replay result for run_in_child_context.""" + + @durable_execution + def handler(event, context: DurableContext) -> dict[str, Any]: + return context.run_in_child_context( + lambda _child_ctx: {"order_id": "ORD-123"}, + name="process_order", + config=ChildConfig(serdes=MarkerSerDes()), + ) + + # --- First invocation --- + checkpoint_calls, mock_checkpoint = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint + + first_result = handler(_create_initial_event(), _create_lambda_context()) + + assert first_result["Status"] == InvocationStatus.SUCCEEDED.value + first_data = json.loads(first_result["Result"]) + + # The child's SUCCEED payload is the *serialized* value (marker stripped). + all_operations = [op for batch in checkpoint_calls for op in batch] + child_succeed_ops = [ + op + for op in all_operations + if op.action == OperationAction.SUCCEED and op.operation_id != "execution-1" + ] + assert len(child_succeed_ops) == 1 + child_payload = child_succeed_ops[0].payload + assert json.loads(child_payload) == {"order_id": "ORD-123"} # no marker + + # --- Replay: rebuild the execution with the child already SUCCEEDED --- + from tests.test_helpers import operation_id_sequence + + child_id = next(operation_id_sequence()) + + checkpoint_calls_replay, mock_checkpoint_replay = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint_replay + + replay_event = _create_replay_event( + [ + { + "Id": child_id, + "Type": "CONTEXT", + "Status": "SUCCEEDED", + "ParentId": "execution-1", + "ContextDetails": {"Result": child_payload}, + } + ] + ) + replay_result = handler(replay_event, _create_lambda_context()) + + assert replay_result["Status"] == InvocationStatus.SUCCEEDED.value + replay_data = json.loads(replay_result["Result"]) + + # First run equals replay, and both carry the round-trip marker. + assert first_data == replay_data + assert first_data == {"order_id": "ORD-123", "round_tripped": True} + + +def test_wait_for_condition_first_run_matches_replay_with_non_identity_serdes(): + """First-run result equals replay result for wait_for_condition.""" + + @durable_execution + def handler(event, context: DurableContext) -> dict[str, Any]: + return context.wait_for_condition( + check=lambda _state, _ctx: {"order_id": "ORD-123"}, + config=WaitForConditionConfig( + initial_state={}, + wait_strategy=lambda _s, _a: WaitForConditionDecision.stop_polling(), + serdes=MarkerSerDes(), + ), + name="process_order", + ) + + # --- First invocation --- + checkpoint_calls, mock_checkpoint = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint + + first_result = handler(_create_initial_event(), _create_lambda_context()) + + assert first_result["Status"] == InvocationStatus.SUCCEEDED.value + first_data = json.loads(first_result["Result"]) + + # The SUCCEED payload is the *serialized* state (marker stripped). A + # wait_for_condition operation is checkpointed as a STEP-typed operation. + all_operations = [op for batch in checkpoint_calls for op in batch] + succeed_ops = [ + op + for op in all_operations + if op.action == OperationAction.SUCCEED and op.operation_id != "execution-1" + ] + assert len(succeed_ops) == 1 + wfc_payload = succeed_ops[0].payload + assert json.loads(wfc_payload) == {"order_id": "ORD-123"} # no marker + + # --- Replay: rebuild the execution with the operation already SUCCEEDED --- + from tests.test_helpers import operation_id_sequence + + wfc_id = next(operation_id_sequence()) + + checkpoint_calls_replay, mock_checkpoint_replay = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint_replay + + replay_event = _create_replay_event( + [ + { + "Id": wfc_id, + "Type": "STEP", + "Status": "SUCCEEDED", + "ParentId": "execution-1", + "StepDetails": {"Result": wfc_payload}, + } + ] + ) + replay_result = handler(replay_event, _create_lambda_context()) + + assert replay_result["Status"] == InvocationStatus.SUCCEEDED.value + replay_data = json.loads(replay_result["Result"]) + + # First run equals replay, and both carry the round-trip marker. assert first_data == replay_data assert first_data == {"order_id": "ORD-123", "round_tripped": True} + + +def test_virtual_child_first_run_returns_round_tripped_result(): + """A virtual child returns the round-tripped value and writes no checkpoint.""" + + @durable_execution + def handler(event, context: DurableContext) -> dict[str, Any]: + return context.run_in_child_context( + lambda _child_ctx: {"order_id": "ORD-123"}, + name="process_order", + config=ChildConfig(serdes=MarkerSerDes(), is_virtual=True), + ) + + checkpoint_calls, mock_checkpoint = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint + + first_result = handler(_create_initial_event(), _create_lambda_context()) + + assert first_result["Status"] == InvocationStatus.SUCCEEDED.value + first_data = json.loads(first_result["Result"]) + + # Virtual contexts re-execute on replay and never checkpoint, so the + # round-tripped first-run result is what every replay reproduces. + assert first_data == {"order_id": "ORD-123", "round_tripped": True} + all_operations = [op for batch in checkpoint_calls for op in batch] + assert not [op for op in all_operations if op.operation_id != "execution-1"] + + +def test_large_child_first_run_matches_replay_with_non_identity_serdes(): + """First-run result equals replay result for a large-payload child.""" + blob = "x" * (256 * 1024 + 100) + + @durable_execution + def handler(event, context: DurableContext) -> dict[str, Any]: + return context.run_in_child_context( + lambda _child_ctx: {"order_id": "ORD-123", "blob": blob}, + name="process_order", + config=ChildConfig(serdes=MarkerSerDes()), + ) + + # --- First invocation: large result -> ReplayChildren + summary checkpoint --- + checkpoint_calls, mock_checkpoint = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint + + first_result = handler(_create_initial_event(), _create_lambda_context()) + + assert first_result["Status"] == InvocationStatus.SUCCEEDED.value + first_data = json.loads(first_result["Result"]) + + all_operations = [op for batch in checkpoint_calls for op in batch] + child_succeed_ops = [ + op + for op in all_operations + if op.action == OperationAction.SUCCEED and op.operation_id != "execution-1" + ] + assert len(child_succeed_ops) == 1 + # The full result is not checkpointed; only a summary with ReplayChildren. + assert child_succeed_ops[0].context_options.replay_children is True + + # --- Replay: child re-executes (ReplayChildren) and round-trips --- + from tests.test_helpers import operation_id_sequence + + child_id = next(operation_id_sequence()) + + checkpoint_calls_replay, mock_checkpoint_replay = _patched_client() + with patch( + "aws_durable_execution_sdk_python.execution.LambdaClient" + ) as mock_client_class: + mock_client = Mock() + mock_client_class.initialize_client.return_value = mock_client + mock_client.checkpoint = mock_checkpoint_replay + + replay_event = _create_replay_event( + [ + { + "Id": child_id, + "Type": "CONTEXT", + "Status": "SUCCEEDED", + "ParentId": "execution-1", + "ContextDetails": {"Result": "", "ReplayChildren": True}, + } + ] + ) + replay_result = handler(replay_event, _create_lambda_context()) + + assert replay_result["Status"] == InvocationStatus.SUCCEEDED.value + replay_data = json.loads(replay_result["Result"]) + + assert first_data == replay_data + assert first_data == {"order_id": "ORD-123", "blob": blob, "round_tripped": True} diff --git a/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py b/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py index b1c439cc..6fb12b0a 100644 --- a/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/exceptions_test.py @@ -25,6 +25,8 @@ InvokeError, OrderedLockError, OrphanedChildException, + RetryableSerDesError, + SerDesError, StepError, StepInterruptedError, SuspendExecution, @@ -288,6 +290,45 @@ def test_operation_error_registry_contains_all_subclasses(): _DURABLE_OPERATION_ERROR_REGISTRY["CallbackSubmitterError"] is CallbackSubmitterError ) + assert _DURABLE_OPERATION_ERROR_REGISTRY["SerDesError"] is SerDesError + + +# ============================================================================= +# SerDes error hierarchy +# ============================================================================= + + +def test_serdes_error_is_catchable_operation_error(): + """SerDesError is a catchable operation error, not a step failure or termination.""" + error = SerDesError("serdes boom") + assert isinstance(error, DurableOperationError) + assert isinstance(error, DurableExecutionsError) + # Distinct from other operation errors so `except StepError` won't catch it. + assert not isinstance(error, StepError) + # Not in the termination tree - it does not fail or retry the invocation. + assert not isinstance(error, UnrecoverableError) + assert error.error_type == "SerDesError" + + +def test_retryable_serdes_error_is_retryable_invocation_error(): + """RetryableSerDesError fails the invocation for backend retry.""" + error = RetryableSerDesError("transient boom") + assert isinstance(error, InvocationError) + assert error.is_retryable() is True + # Not an operation error, so it is not caught as a per-operation failure. + assert not isinstance(error, DurableOperationError) + + +def test_from_error_fields_reconstructs_serdes_error(): + """from_error_fields rebuilds SerDesError from its discriminator (replay).""" + error = DurableOperationError.from_error_fields( + "SerDesError", "serdes boom", "payload", ["frame"] + ) + assert isinstance(error, SerDesError) + assert error.error_type == "SerDesError" + assert error.message == "serdes boom" + assert error.data == "payload" + assert error.stack_trace == ["frame"] # ============================================================================= diff --git a/packages/aws-durable-execution-sdk-python/tests/filesystem_serdes_test.py b/packages/aws-durable-execution-sdk-python/tests/filesystem_serdes_test.py index 61c8b370..ecde61a9 100644 --- a/packages/aws-durable-execution-sdk-python/tests/filesystem_serdes_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/filesystem_serdes_test.py @@ -632,25 +632,25 @@ def test_works_with_top_level_serialize_function(self, tmp_path): assert deserialized == value def test_serialization_error_handling(self, tmp_path): - """Serialize should raise ExecutionError on failure via top-level API.""" - from aws_durable_execution_sdk_python.exceptions import ExecutionError + """Serialize should raise SerDesError on failure via top-level API.""" + from aws_durable_execution_sdk_python.exceptions import SerDesError from aws_durable_execution_sdk_python.serdes import serialize fs_serdes = FileSystemSerDes("/nonexistent/readonly/path") - with pytest.raises(ExecutionError, match="Serialization failed"): + with pytest.raises(SerDesError, match="Serialization failed"): serialize(fs_serdes, {"data": "test"}, TEST_OPERATION_ID, TEST_ARN) def test_deserialization_error_handling(self, tmp_path): - """Deserialize should raise ExecutionError on failure via top-level API.""" - from aws_durable_execution_sdk_python.exceptions import ExecutionError + """Deserialize should raise SerDesError on failure via top-level API.""" + from aws_durable_execution_sdk_python.exceptions import SerDesError from aws_durable_execution_sdk_python.serdes import deserialize fs_serdes = FileSystemSerDes(str(tmp_path)) # Invalid envelope pointing to nonexistent file invalid_data = json.dumps({"file": "/nonexistent/file.json"}) - with pytest.raises(ExecutionError, match="Deserialization failed"): + with pytest.raises(SerDesError, match="Deserialization failed"): deserialize(fs_serdes, invalid_data, TEST_OPERATION_ID, TEST_ARN) diff --git a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py index 2d0ff829..20ddef04 100644 --- a/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/lambda_service_test.py @@ -11,6 +11,7 @@ CheckpointError, DurableOperationError, GetExecutionStateError, + SerDesError, StepError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier @@ -179,6 +180,28 @@ def test_error_object_from_exception_empty_message(): assert error.stack_trace is None +def test_raise_as_operation_error_wraps_in_passed_class(): + """A non-serdes error surfaces as the operation's own error type.""" + error_object = ErrorObject.from_exception(ValueError("boom")) + with pytest.raises(StepError) as exc_info: + error_object.raise_as_operation_error(StepError) + assert exc_info.value.error_type == "ValueError" + + +def test_raise_as_operation_error_surfaces_serdes_error_directly(): + """A serdes failure surfaces as SerDesError regardless of the operation kind. + + Both the first-run FAIL path and replay call this method, so this covers + both: a SerDesError is catchable as itself, not wrapped in the operation's + own error type. + """ + error_object = ErrorObject.from_exception(SerDesError("serdes boom")) + with pytest.raises(SerDesError) as exc_info: + error_object.raise_as_operation_error(StepError) + assert not isinstance(exc_info.value, StepError) + assert exc_info.value.error_type == "SerDesError" + + def test_error_object_from_message_regular(): """Test ErrorObject.from_message with regular message.""" error = ErrorObject.from_message("Test error message") diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py index 07c7150e..0da58975 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/child_test.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from typing import cast +from typing import Any, cast from unittest.mock import Mock import pytest @@ -15,6 +15,8 @@ DurableApiErrorCategory, ExecutionError, InvocationError, + RetryableSerDesError, + SerDesError, StepError, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier @@ -25,9 +27,31 @@ OperationType, ) from aws_durable_execution_sdk_python.operation.child import child_handler +from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext from aws_durable_execution_sdk_python.state import ExecutionState from aws_durable_execution_sdk_python.types import SummaryGenerator -from tests.serdes_test import CustomDictSerDes +from tests.serdes_test import ( + CustomDictSerDes, + PermanentDeserializeSerDes, + RetryableDeserializeSerDes, +) + + +class _NonIdentitySerDes(SerDes[Any]): + """deserialize() adds a marker that serialize() never removes. + + Makes ``deserialize(serialize(x)) != x`` so first-run vs replay divergence + is observable. + """ + + def serialize(self, value: Any, _: SerDesContext) -> str: + payload = dict(value) + payload.pop("deserialized", None) + return json.dumps(payload) + + def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: + parsed = json.loads(data) + return {**parsed, "deserialized": True} # region child_handler @@ -1150,3 +1174,235 @@ def setup_mocks(): assert result2 == "test_result" assert mock_state2.create_checkpoint.call_count == 0 # No checkpoints + + +# region first-run round-trip +def test_child_handler_first_run_returns_round_tripped_result(): + """First-run result must equal the replay (deserialized-from-checkpoint) result. + + With a non-identity SerDes, returning the raw child result on the first run + diverges from the value replay reconstructs by deserializing the checkpoint. + The first run must return the serialize-then-deserialize value so both agree. + """ + serdes = _NonIdentitySerDes() + raw_result = {"key": "value"} + config = ChildConfig(serdes=serdes) + op_id = OperationIdentifier( + "child_rt", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ) + + # --- First run --- + first_state = Mock(spec=ExecutionState) + first_state.durable_execution_arn = "test_arn" + first_result = Mock() + first_result.is_succeeded.return_value = False + first_result.is_failed.return_value = False + first_result.is_started.return_value = False + first_result.is_replay_children.return_value = False + first_result.is_existent.return_value = False + first_state.get_checkpoint_result.return_value = first_result + mock_callable = Mock(return_value=raw_result) + first_state.wrap_user_function.return_value = mock_callable + + first_run_result = child_handler(mock_callable, first_state, op_id, config) + + # Grab the payload actually checkpointed (START is call 0, SUCCEED is call 1). + success_call = first_state.create_checkpoint.call_args_list[1] + checkpointed_payload = success_call[1]["operation_update"].payload + + # --- Replay: checkpoint already SUCCEEDED (not replay_children) --- + replay_state = Mock(spec=ExecutionState) + replay_state.durable_execution_arn = "test_arn" + replay_result = Mock() + replay_result.is_succeeded.return_value = True + replay_result.is_replay_children.return_value = False + replay_result.result = checkpointed_payload + replay_state.get_checkpoint_result.return_value = replay_result + + replayed = child_handler( + Mock(return_value="should_not_call"), replay_state, op_id, config + ) + + assert first_run_result == {"key": "value", "deserialized": True} + assert first_run_result == replayed + assert first_run_result != raw_result + + +def test_child_handler_first_run_none_payload_skips_deserialize(): + """A None serialized payload is returned as-is without deserializing. + + Mirrors the replay path, which returns None (without deserializing) when the + checkpointed result is None. + """ + + class NonePayloadSerDes(SerDes[Any]): + """serialize() yields None; deserialize() must never be called for None.""" + + def serialize(self, _value: Any, _ctx: SerDesContext) -> str: + return None # type: ignore[return-value] + + def deserialize(self, _data: str, _ctx: SerDesContext) -> Any: + msg = "deserialize should not be called for a None payload" + raise AssertionError(msg) + + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_result = Mock() + mock_result.is_succeeded.return_value = False + mock_result.is_failed.return_value = False + mock_result.is_started.return_value = False + mock_result.is_replay_children.return_value = False + mock_result.is_existent.return_value = False + mock_state.get_checkpoint_result.return_value = mock_result + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + + result = child_handler( + mock_callable, + mock_state, + OperationIdentifier( + "child_none", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ), + ChildConfig(serdes=NonePayloadSerDes()), + ) + + assert result is None + + +def test_child_handler_virtual_returns_round_tripped_result(): + """A virtual child returns the serdes round-tripped result. + + Virtual contexts write no checkpoint and re-execute on replay, so the + round-trip keeps the returned value identical across first run and replay. + """ + serdes = _NonIdentitySerDes() + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_result = Mock() + mock_result.is_succeeded.return_value = False + mock_result.is_failed.return_value = False + mock_result.is_started.return_value = False + mock_result.is_replay_children.return_value = False + mock_result.is_existent.return_value = False + mock_state.get_checkpoint_result.return_value = mock_result + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + + result = child_handler( + mock_callable, + mock_state, + OperationIdentifier( + "child_virtual_rt", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ), + ChildConfig(serdes=serdes, is_virtual=True), + ) + + # Round-tripped result (deserialize adds the marker). + assert result == {"key": "value", "deserialized": True} + # Virtual contexts still write no checkpoints. + assert mock_state.create_checkpoint.call_count == 0 + + +def test_child_handler_replay_children_returns_round_tripped_result(): + """A large-payload (ReplayChildren) child round-trips its re-executed result. + + In ReplayChildren mode only a summary is checkpointed and the child + re-executes on replay. The re-executed result is round-tripped so it matches + the value the first run returned; only the checkpoint payload stays small. + """ + serdes = _NonIdentitySerDes() + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_result = Mock() + mock_result.is_succeeded.return_value = True + mock_result.is_failed.return_value = False + mock_result.is_started.return_value = True + mock_result.is_replay_children.return_value = True + mock_result.is_existent.return_value = True + mock_state.get_checkpoint_result.return_value = mock_result + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + + result = child_handler( + mock_callable, + mock_state, + OperationIdentifier( + "child_replay_rt", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ), + ChildConfig(serdes=serdes), + ) + + # Round-tripped result (deserialize adds the marker). + assert result == {"key": "value", "deserialized": True} + # ReplayChildren re-executes the function and writes no new checkpoint. + mock_callable.assert_called_once() + mock_state.create_checkpoint.assert_not_called() + + +# endregion first-run round-trip + + +def test_child_handler_permanent_serdes_error_surfaces_without_double_checkpoint(): + """A permanent round-trip failure surfaces SerDesError and never double-checkpoints. + + Deserialization runs before the SUCCEED checkpoint, so a permanent failure + writes FAIL and never SUCCEED for the same operation. + """ + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_result = Mock() + mock_result.is_succeeded.return_value = False + mock_result.is_failed.return_value = False + mock_result.is_replay_children.return_value = False + mock_result.is_existent.return_value = False + mock_state.get_checkpoint_result.return_value = mock_result + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + + with pytest.raises(SerDesError): + child_handler( + mock_callable, + mock_state, + OperationIdentifier( + "childP", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ), + ChildConfig(serdes=PermanentDeserializeSerDes()), + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL in actions + assert OperationAction.SUCCEED not in actions + + +def test_child_handler_transient_serdes_error_reraised(): + """A transient serdes failure re-raises for backend retry, writing no FAIL.""" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_result = Mock() + mock_result.is_succeeded.return_value = False + mock_result.is_failed.return_value = False + mock_result.is_replay_children.return_value = False + mock_result.is_existent.return_value = False + mock_state.get_checkpoint_result.return_value = mock_result + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + + with pytest.raises(RetryableSerDesError): + child_handler( + mock_callable, + mock_state, + OperationIdentifier( + "childT", OperationSubType.RUN_IN_CHILD_CONTEXT, None, "test_name" + ), + ChildConfig(serdes=RetryableDeserializeSerDes()), + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL not in actions + assert OperationAction.SUCCEED not in actions diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py index 02f69a7f..3abf5c96 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py @@ -481,8 +481,10 @@ def get_checkpoint_result(self, operation_id): operation_id_namespace=_StubNamespace(), ) - # Verify execute was called - assert result.all[0].result == "RESULT_TEST_ITEM" + # Verify execute was called. The item result is the serdes round-tripped + # value: CustomStrSerDes uppercases on serialize and lowercases on + # deserialize, so "RESULT_TEST_ITEM" round-trips to "result_test_item". + assert result.all[0].result == "result_test_item" def test_map_handler_with_summary_generator(): @@ -499,7 +501,9 @@ def mock_summary_generator(result): executor_context = Mock() executor_context._create_step_id_for_logical_step = Mock(side_effect=["1", "2"]) - executor_context.create_child_context = Mock(return_value=Mock()) + _child_ctx = Mock() + _child_ctx.state.wrap_user_function = lambda func, *a, **k: func + executor_context.create_child_context = Mock(return_value=_child_ctx) class MockExecutionState: def register_branch_pool(self, pool): @@ -598,7 +602,9 @@ def get_checkpoint_result(self, operation_id): executor_context._create_step_id_for_logical_step = Mock( side_effect=["1", "2", "3"] ) - executor_context.create_child_context = Mock(return_value=Mock()) + _child_ctx = Mock() + _child_ctx.state.wrap_user_function = lambda func, *a, **k: func + executor_context.create_child_context = Mock(return_value=_child_ctx) # Call map_handler map_handler( @@ -1004,9 +1010,13 @@ def create_id(self, i): call.kwargs["operation_id"]: call.kwargs for call in mock_deserialize.call_args_list } - assert set(deserialize_calls) == {"child-0", "child-1"} + # Children deserialize from their checkpoints with the item serdes; the + # parent also deserializes its BatchResult (the first-run round-trip) with + # the batch serdes. + assert set(deserialize_calls) == {"child-0", "child-1", "parent"} assert deserialize_calls["child-0"]["serdes"] is expected assert deserialize_calls["child-1"]["serdes"] is expected + assert deserialize_calls["parent"]["serdes"] is batch_serdes def test_map_result_serialization_roundtrip(): @@ -1126,7 +1136,14 @@ def create_id(self, i): assert len(mock_serdes_serialize.call_args_list) == 3 parent_call = mock_serdes_serialize.call_args_list[2] - assert parent_call[1]["value"] is result + # The value serialized (checkpointed) at the parent level is the raw + # BatchResult. + assert isinstance(parent_call[1]["value"], BatchResult) + # The first run returns the round-trip of that checkpointed payload, + # matching what replay would deserialize from the checkpoint (with + # serialize mocked to a plain string, the round-trip yields that + # string). + assert result == "serialized" finally: importlib.reload(child) @@ -1195,7 +1212,10 @@ def create_id(self, i): parent_call = mock_serialize.call_args_list[2] assert parent_call[1]["serdes"] is None assert isinstance(parent_call[1]["value"], BatchResult) - assert parent_call[1]["value"] is result + # First run returns the round-tripped BatchResult, which equals the + # value serialized into the checkpoint (default serdes round-trips + # as identity). + assert parent_call[1]["value"] == result finally: importlib.reload(child) @@ -1204,12 +1224,22 @@ def test_map_custom_serdes_serializes_batch_result(): """Verify custom serdes is used for BatchResult serialization.""" custom_serdes = CustomStrSerDes() + round_tripped: BatchResult = BatchResult( + all=[BatchItem(index=0, status=BatchItemStatus.SUCCEEDED, result="test")], + completion_reason=CompletionReason.ALL_COMPLETED, + ) try: - with patch( - "aws_durable_execution_sdk_python.serdes.serialize" - ) as mock_serialize: + with ( + patch( + "aws_durable_execution_sdk_python.serdes.serialize" + ) as mock_serialize, + patch( + "aws_durable_execution_sdk_python.serdes.deserialize" + ) as mock_deserialize, + ): mock_serialize.return_value = '"serialized"' + mock_deserialize.return_value = round_tripped importlib.reload(child) parent_checkpoint = Mock() @@ -1267,12 +1297,14 @@ def create_id(self, i): config=MapConfig(serdes=custom_serdes), ) - assert isinstance(result, BatchResult) assert len(mock_serialize.call_args_list) == 3 parent_call = mock_serialize.call_args_list[2] assert parent_call[1]["serdes"] is custom_serdes + # The parent serializes a BatchResult with the custom serdes and + # returns the round-tripped BatchResult. assert isinstance(parent_call[1]["value"], BatchResult) - assert parent_call[1]["value"] is result + assert isinstance(result, BatchResult) + assert result is round_tripped finally: importlib.reload(child) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/parallel_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/parallel_test.py index a19f87e8..f333cb08 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/parallel_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/parallel_test.py @@ -490,7 +490,10 @@ def get_checkpoint_result(self, operation_id): operation_id_namespace=_StubNamespace(), ) - assert result.all[0].result == "RESULT1" + # The branch result is the serdes round-tripped value: CustomStrSerDes + # uppercases on serialize and lowercases on deserialize, so "RESULT1" + # round-trips to "result1". + assert result.all[0].result == "result1" def test_parallel_handler_with_summary_generator(): @@ -564,7 +567,9 @@ def get_checkpoint_result(self, operation_id): executor_context = Mock() executor_context._create_step_id_for_logical_step = Mock(side_effect=["1", "2"]) - executor_context.create_child_context = Mock(return_value=Mock()) + _child_ctx = Mock() + _child_ctx.state.wrap_user_function = lambda func, *a, **k: func + executor_context.create_child_context = Mock(return_value=_child_ctx) # Call parallel_handler with None config (should use default) parallel_handler( @@ -614,7 +619,9 @@ def get_checkpoint_result(self, operation_id): executor_context._create_step_id_for_logical_step = Mock( side_effect=["1", "2", "3"] ) - executor_context.create_child_context = Mock(return_value=Mock()) + _child_ctx = Mock() + _child_ctx.state.wrap_user_function = lambda func, *a, **k: func + executor_context.create_child_context = Mock(return_value=_child_ctx) # Call parallel_handler parallel_handler( @@ -1017,9 +1024,13 @@ def create_id(self, i): expected = item_serdes or batch_serdes calls_by_operation_id = _mock_call_kwargs_by_operation_id(mock_deserialize) - assert set(calls_by_operation_id) == {"child-0", "child-1"} + # Branches deserialize from their checkpoints with the item serdes; the + # parent also deserializes its BatchResult (the first-run round-trip) with + # the batch serdes. + assert set(calls_by_operation_id) == {"child-0", "child-1", "parent"} assert calls_by_operation_id["child-0"]["serdes"] is expected assert calls_by_operation_id["child-1"]["serdes"] is expected + assert calls_by_operation_id["parent"]["serdes"] is batch_serdes def test_parallel_result_serialization_roundtrip(): @@ -1146,7 +1157,14 @@ def create_id(self, i): assert len(mock_serdes_serialize.call_args_list) == 3 parent_call = mock_serdes_serialize.call_args_list[2] - assert parent_call[1]["value"] is result + # The value serialized (checkpointed) at the parent level is the raw + # BatchResult. + assert isinstance(parent_call[1]["value"], BatchResult) + # The first run returns the round-trip of that checkpointed payload, + # matching what replay would deserialize from the checkpoint (with + # serialize mocked to a plain string, the round-trip yields that + # string). + assert result == "serialized" finally: importlib.reload(child) @@ -1215,7 +1233,10 @@ def create_id(self, i): parent_call = mock_serialize.call_args_list[2] assert parent_call[1]["serdes"] is None assert isinstance(parent_call[1]["value"], BatchResult) - assert parent_call[1]["value"] is result + # First run returns the round-tripped BatchResult, which equals the + # value serialized into the checkpoint (default serdes round-trips + # as identity). + assert parent_call[1]["value"] == result finally: importlib.reload(child) @@ -1224,12 +1245,22 @@ def test_parallel_custom_serdes_serializes_batch_result(): """Verify custom serdes is used for BatchResult serialization.""" custom_serdes = CustomStrSerDes() + round_tripped: BatchResult = BatchResult( + all=[BatchItem(index=0, status=BatchItemStatus.SUCCEEDED, result="test")], + completion_reason=CompletionReason.ALL_COMPLETED, + ) try: - with patch( - "aws_durable_execution_sdk_python.serdes.serialize" - ) as mock_serialize: + with ( + patch( + "aws_durable_execution_sdk_python.serdes.serialize" + ) as mock_serialize, + patch( + "aws_durable_execution_sdk_python.serdes.deserialize" + ) as mock_deserialize, + ): mock_serialize.return_value = '"serialized"' + mock_deserialize.return_value = round_tripped importlib.reload(child) parent_checkpoint = Mock() @@ -1286,12 +1317,14 @@ def create_id(self, i): config=ParallelConfig(serdes=custom_serdes), ) - assert isinstance(result, BatchResult) assert len(mock_serialize.call_args_list) == 3 parent_call = mock_serialize.call_args_list[2] assert parent_call[1]["serdes"] is custom_serdes + # The parent serializes a BatchResult with the custom serdes and + # returns the round-tripped BatchResult. assert isinstance(parent_call[1]["value"], BatchResult) - assert parent_call[1]["value"] is result + assert isinstance(result, BatchResult) + assert result is round_tripped finally: importlib.reload(child) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py index 6021da18..4ee8d9fb 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/step_test.py @@ -15,6 +15,8 @@ from aws_durable_execution_sdk_python.exceptions import ( DurableOperationError, ExecutionError, + RetryableSerDesError, + SerDesError, StepError, StepInterruptedError, SuspendExecution, @@ -39,7 +41,11 @@ ) from aws_durable_execution_sdk_python.state import CheckpointedResult, ExecutionState from aws_durable_execution_sdk_python.types import StepContext -from tests.serdes_test import CustomDictSerDes +from tests.serdes_test import ( + CustomDictSerDes, + PermanentDeserializeSerDes, + RetryableDeserializeSerDes, +) # Test helper - maintains old handler signature for backward compatibility in tests @@ -1227,3 +1233,75 @@ def test_step_creates_start_checkpoint_when_status_is_ready(): success_call = mock_state.create_checkpoint.call_args_list[1] success_operation = success_call[1]["operation_update"] assert success_operation.action is OperationAction.SUCCEED + + +def test_step_handler_permanent_serdes_error_surfaces_without_double_checkpoint(): + """A permanent round-trip failure surfaces SerDesError and never double-checkpoints. + + Deserialization runs before the SUCCEED checkpoint, so a permanent failure + writes FAIL and never SUCCEED for the same operation. + """ + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + with pytest.raises(SerDesError): + step_handler( + mock_callable, + mock_state, + OperationIdentifier("stepP", OperationSubType.STEP, None, "test_step"), + StepConfig( + step_semantics=StepSemantics.AT_LEAST_ONCE_PER_RETRY, + serdes=PermanentDeserializeSerDes(), + ), + mock_logger, + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL in actions + assert OperationAction.SUCCEED not in actions + + +def test_step_handler_transient_serdes_error_reraised(): + """A transient serdes failure re-raises for backend retry, with no checkpoint. + + RetryableSerDesError bypasses the step retry strategy and writes neither + SUCCEED nor FAIL, so the invocation fails and the backend retries. + """ + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + mock_callable = Mock(return_value={"key": "value"}) + mock_state.wrap_user_function.return_value = mock_callable + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + with pytest.raises(RetryableSerDesError): + step_handler( + mock_callable, + mock_state, + OperationIdentifier("stepT", OperationSubType.STEP, None, "test_step"), + StepConfig( + step_semantics=StepSemantics.AT_LEAST_ONCE_PER_RETRY, + serdes=RetryableDeserializeSerDes(), + ), + mock_logger, + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL not in actions + assert OperationAction.SUCCEED not in actions diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py index 40f97a0d..3ad61cc7 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/wait_for_condition_test.py @@ -2,6 +2,7 @@ import datetime import json +from typing import Any from unittest.mock import Mock import pytest @@ -13,6 +14,8 @@ DurableOperationError, ExecutionError, InvocationError, + RetryableSerDesError, + SerDesError, SuspendExecution, WaitForConditionError, ) @@ -30,6 +33,7 @@ from aws_durable_execution_sdk_python.operation.wait_for_condition import ( WaitForConditionOperationExecutor, ) +from aws_durable_execution_sdk_python.serdes import SerDes, SerDesContext from aws_durable_execution_sdk_python.state import CheckpointedResult, ExecutionState from aws_durable_execution_sdk_python.types import WaitForConditionCheckContext from aws_durable_execution_sdk_python.waits import ( @@ -38,7 +42,11 @@ WaitStrategyConfig, create_wait_strategy, ) -from tests.serdes_test import CustomDictSerDes +from tests.serdes_test import ( + CustomDictSerDes, + PermanentDeserializeSerDes, + RetryableDeserializeSerDes, +) # Test helper - maintains old handler signature for backward compatibility in tests @@ -362,7 +370,7 @@ def check_func(state, context): wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), ) - with pytest.raises(WaitForConditionError) as exc_info: + with pytest.raises(SerDesError) as exc_info: wait_for_condition_handler( state=mock_state, operation_identifier=op_id, @@ -371,7 +379,7 @@ def check_func(state, context): context_logger=mock_logger, ) - assert exc_info.value.error_type == "ExecutionError" + assert exc_info.value.error_type == "SerDesError" mock_state.create_checkpoint.assert_called_once() assert ( mock_state.create_checkpoint.call_args.kwargs["operation_update"].action @@ -1710,3 +1718,225 @@ def mock_wait_strategy(state, attempt): assert result == "final_state" assert mock_state.get_checkpoint_result.call_count == 1 # Single check (async) assert mock_state.create_checkpoint.call_count == 2 # START + SUCCESS checkpoints + + +def test_wait_for_condition_first_run_returns_round_tripped_result(): + """First-run result must match the replay (deserialized-from-checkpoint) result. + + With a non-identity SerDes whose serialize/deserialize is not a round-trip + identity, returning the raw check-function result on the first run diverges + from the value returned on replay. The first run must return the value + obtained by serializing then deserializing, so both runs agree. + """ + + class NonIdentitySerDes(SerDes[Any]): + """deserialize() adds a marker that serialize() never removes.""" + + def serialize(self, value: Any, _: SerDesContext) -> str: + payload = dict(value) + payload.pop("deserialized", None) + return json.dumps(payload) + + def deserialize(self, data: str, _: SerDesContext) -> dict[str, Any]: + parsed = json.loads(data) + return {**parsed, "deserialized": True} + + serdes = NonIdentitySerDes() + raw_new_state = {"key": "value"} + + def check_func(_state, _context): + return raw_new_state + + config = WaitForConditionConfig( + initial_state={}, + wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), + serdes=serdes, + ) + + # --- First run --- + first_run_state = Mock(spec=ExecutionState) + first_run_state.durable_execution_arn = "test_arn" + first_run_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + first_run_state.wrap_user_function.return_value = check_func + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + op_id = OperationIdentifier( + "wfc_rt", OperationSubType.WAIT_FOR_CONDITION, None, "test_wait" + ) + first_run_result = wait_for_condition_handler( + state=first_run_state, + operation_identifier=op_id, + check=check_func, + config=config, + context_logger=mock_logger, + ) + + # Grab the payload that was actually checkpointed (START is call 0, SUCCEED is call 1). + success_call = first_run_state.create_checkpoint.call_args_list[1] + checkpointed_payload = success_call[1]["operation_update"].payload + + # --- Replay: checkpoint already SUCCEEDED with the serialized payload --- + replay_state = Mock(spec=ExecutionState) + replay_state.durable_execution_arn = "test_arn" + succeeded_op = Operation( + operation_id="wfc_rt", + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + step_details=StepDetails(result=checkpointed_payload), + ) + replay_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_from_operation(succeeded_op) + ) + replay_result = wait_for_condition_handler( + state=replay_state, + operation_identifier=op_id, + check=Mock(return_value="should_not_call"), + config=config, + context_logger=Mock(spec=Logger), + ) + + # First run must return the round-tripped value, which equals the replay value, + # and must NOT equal the raw check-function result. + assert first_run_result == {"key": "value", "deserialized": True} + assert first_run_result == replay_result + assert first_run_result != raw_new_state + + +def test_wait_for_condition_first_run_none_payload_skips_deserialize(): + """A None serialized payload is returned as-is without deserializing. + + This mirrors the replay path, which returns None without calling deserialize + when the checkpointed result is None. A serdes whose serialize returns None + must therefore see its deserialize skipped on the first run too. + """ + + class NonePayloadSerDes(SerDes[Any]): + """serialize() yields None; deserialize() must never be called for None.""" + + def serialize(self, _value: Any, _ctx: SerDesContext) -> str: + return None # type: ignore[return-value] + + def deserialize(self, _data: str, _ctx: SerDesContext) -> Any: + msg = "deserialize should not be called for a None payload" + raise AssertionError(msg) + + def check_func(_state, _context): + return {"key": "value"} + + config = WaitForConditionConfig( + initial_state={}, + wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), + serdes=NonePayloadSerDes(), + ) + + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "test_arn" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + mock_state.wrap_user_function.return_value = check_func + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + result = wait_for_condition_handler( + state=mock_state, + operation_identifier=OperationIdentifier( + "wfc_none", OperationSubType.WAIT_FOR_CONDITION, None, "test_wait" + ), + check=check_func, + config=config, + context_logger=mock_logger, + ) + + assert result is None + + +def test_wait_for_condition_permanent_serdes_error_surfaces_without_double_checkpoint(): + """A permanent round-trip failure surfaces SerDesError and never double-checkpoints. + + Deserialization runs before the SUCCEED checkpoint, so a permanent failure + writes FAIL and never SUCCEED for the same operation. + """ + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "arn:aws:test" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + op_id = OperationIdentifier( + "op_perm", OperationSubType.WAIT_FOR_CONDITION, None, "test_wait" + ) + + def check_func(state, context): + return state + 1 + + mock_state.wrap_user_function.return_value = check_func + + config = WaitForConditionConfig( + initial_state=5, + wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), + serdes=PermanentDeserializeSerDes(), + ) + + with pytest.raises(SerDesError): + wait_for_condition_handler( + state=mock_state, + operation_identifier=op_id, + check=check_func, + config=config, + context_logger=mock_logger, + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL in actions + assert OperationAction.SUCCEED not in actions + + +def test_wait_for_condition_transient_serdes_error_reraised(): + """A transient serdes failure re-raises for backend retry, writing no FAIL.""" + mock_state = Mock(spec=ExecutionState) + mock_state.durable_execution_arn = "arn:aws:test" + mock_state.get_checkpoint_result.return_value = ( + CheckpointedResult.create_not_found() + ) + mock_logger = Mock(spec=Logger) + mock_logger.with_log_info.return_value = mock_logger + + op_id = OperationIdentifier( + "op_transient", OperationSubType.WAIT_FOR_CONDITION, None, "test_wait" + ) + + def check_func(state, context): + return state + 1 + + mock_state.wrap_user_function.return_value = check_func + + config = WaitForConditionConfig( + initial_state=5, + wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(), + serdes=RetryableDeserializeSerDes(), + ) + + with pytest.raises(RetryableSerDesError): + wait_for_condition_handler( + state=mock_state, + operation_identifier=op_id, + check=check_func, + config=config, + context_logger=mock_logger, + ) + + actions: list[OperationAction] = [ + call.kwargs["operation_update"].action + for call in mock_state.create_checkpoint.call_args_list + ] + assert OperationAction.FAIL not in actions + assert OperationAction.SUCCEED not in actions diff --git a/packages/aws-durable-execution-sdk-python/tests/serdes_test.py b/packages/aws-durable-execution-sdk-python/tests/serdes_test.py index d511918a..9df11215 100644 --- a/packages/aws-durable-execution-sdk-python/tests/serdes_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/serdes_test.py @@ -16,7 +16,7 @@ ) from aws_durable_execution_sdk_python.exceptions import ( DurableExecutionsError, - ExecutionError, + RetryableSerDesError, SerDesError, ) from aws_durable_execution_sdk_python.lambda_service import ErrorObject @@ -83,6 +83,28 @@ def _rec_deserialize(self, value: Any) -> Any: return value +class PermanentDeserializeSerDes(SerDes[Any]): + """Serializes fine but always fails to deserialize (permanent failure).""" + + def serialize(self, value: Any, serdes_context: SerDesContext) -> str: + return json.dumps(value) + + def deserialize(self, data: str, serdes_context: SerDesContext) -> Any: + msg = "cannot deserialize" + raise ValueError(msg) + + +class RetryableDeserializeSerDes(SerDes[Any]): + """Serializes fine but signals a transient failure on deserialize.""" + + def serialize(self, value: Any, serdes_context: SerDesContext) -> str: + return json.dumps(value) + + def deserialize(self, data: str, serdes_context: SerDesContext) -> Any: + msg = "transient serdes failure" + raise RetryableSerDesError(msg) + + # region Abstract SerDes Tests def test_serdes_abstract(): """Test SerDes abstract base class.""" @@ -131,17 +153,38 @@ def test_serialize_invalid_json(): circular_ref = {"a": 1} circular_ref["self"] = circular_ref - with pytest.raises(ExecutionError) as exc_info: + with pytest.raises(SerDesError) as exc_info: serialize(None, circular_ref, "test-op", "test-arn") assert "Serialization failed" in str(exc_info.value) def test_deserialize_invalid_json(): - with pytest.raises(ExecutionError) as exc_info: + with pytest.raises(SerDesError) as exc_info: deserialize(None, "invalid json", "test-op", "test-arn") assert "Deserialization failed" in str(exc_info.value) +def test_serialize_propagates_retryable_serdes_error(): + """A transient failure on serialize propagates unchanged (not wrapped as SerDesError).""" + + class RaiseOnSerialize(SerDes[Any]): + def serialize(self, value: Any, serdes_context: SerDesContext) -> str: + msg = "transient serialize failure" + raise RetryableSerDesError(msg) + + def deserialize(self, data: str, serdes_context: SerDesContext) -> Any: + return json.loads(data) + + with pytest.raises(RetryableSerDesError): + serialize(RaiseOnSerialize(), {"a": 1}, "test-op", "test-arn") + + +def test_deserialize_propagates_retryable_serdes_error(): + """A transient failure on deserialize propagates unchanged (not wrapped as SerDesError).""" + with pytest.raises(RetryableSerDesError): + deserialize(RetryableDeserializeSerDes(), "{}", "test-op", "test-arn") + + def test_none_serdes_context(): data = {"test": "value"} result = serialize(None, data, None, None) @@ -591,7 +634,7 @@ def test_envelope_handles_json_incompatible_types(): } # JsonSerDes should fail - with pytest.raises(ExecutionError): + with pytest.raises(SerDesError): serialize(json_serdes, complex_data, "test-op", "test-arn") # EnvelopeSerDes should succeed @@ -606,11 +649,11 @@ def test_envelope_error_handling_with_main_api(): envelope_serdes = ExtendedTypeSerDes() # Test serialization error - with pytest.raises(ExecutionError, match="Serialization failed"): + with pytest.raises(SerDesError, match="Serialization failed"): serialize(envelope_serdes, object(), "test-op", "test-arn") # Test deserialization error - with pytest.raises(ExecutionError, match="Deserialization failed"): + with pytest.raises(SerDesError, match="Deserialization failed"): deserialize(envelope_serdes, "invalid json", "test-op", "test-arn")