feat: Adding replay mode for child execution

- During child execution, if the result exceeded checkpoint api size limit, set replay mode to True
- When replay, if replay mode is true, re-execute func instead of return the result from the payload

Author: Alex Wang

src/aws_durable_execution_sdk_python/config.py

@@ -104,7 +104,7 @@ class CheckpointMode(Enum):
 
 
 @dataclass(frozen=True)
-class ChildConfig:
+class ChildConfig(Generic[T]):
     """Options when running inside a child context."""
 
     # checkpoint_mode: CheckpointMode = CheckpointMode.CHECKPOINT_AT_START_AND_FINISH

src/aws_durable_execution_sdk_python/operation/child.py

@@ -79,7 +79,7 @@ def child_handler(
         raw_result: T = func()
         if checkpointed_result.is_replay_children():
             logger.debug(
-                "ReplayChildren mode: Re-executing child context due to large payload: id: %s, name: %s",
+                "ReplayChildren mode: Executed child context again on replay due to large payload. Exiting child context without creating another checkpoint. id: %s, name: %s",
                 operation_identifier.operation_id,
                 operation_identifier.name,
             )
@@ -90,22 +90,21 @@ def child_handler(
             operation_id=operation_identifier.operation_id,
             durable_execution_arn=state.durable_execution_arn,
         )
-        payload_to_checkpoint = serialized_result
-        replay_children = False
+        replay_children: bool = False
         if len(serialized_result) > CHECKPOINT_SIZE_LIMIT:
             logger.debug(
                 "Large payload detected, using ReplayChildren mode: id: %s, name: %s",
                 operation_identifier.operation_id,
                 operation_identifier.name,
             )
             replay_children = True
-            payload_to_checkpoint = (
+            serialized_result = (
                 config.summary_generator(raw_result) if config.summary_generator else ""
             )
 
         success_operation = OperationUpdate.create_context_succeed(
             identifier=operation_identifier,
-            payload=payload_to_checkpoint,
+            payload=serialized_result,
             sub_type=sub_type,
             context_options=ContextOptions(replay_children=replay_children),
         )

src/aws_durable_execution_sdk_python/state.py

@@ -117,10 +117,7 @@ def is_replay_children(self) -> bool:
         op = self.operation
         if not op:
             return False
-        context_details = op.context_details
-        if not context_details:
-            return False
-        return context_details.replay_children
+        return op.context_details.replay_children if op.context_details else False
 
     def raise_callable_error(self) -> None:
         if self.error is None:

tests/operation/child_test.py

@@ -42,6 +42,7 @@ def test_child_handler_not_started(
     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_replay_children.return_value = False
     mock_state.get_checkpoint_result.return_value = mock_result
     mock_callable = Mock(return_value="fresh_result")
 
@@ -374,7 +375,11 @@ def test_child_handler_large_payload_with_summary_generator():
     mock_state.get_checkpoint_result.return_value = mock_result
     large_result = "large" * 256 * 1024
     mock_callable = Mock(return_value=large_result)
-    child_config: ChildConfig = ChildConfig(summary_generator=lambda x: "summary")
+
+    def my_summary(result: str) -> str:
+        return "summary"
+
+    child_config: ChildConfig = ChildConfig[str](summary_generator=my_summary)
 
     actual_result = child_handler(
         mock_callable,