parity: Suspend when invoking a pending function

When invoking a function that is currently pending, we should treat it
is as if it has just started, and suspend execution.

Author: FullyTyped

src/aws_durable_execution_sdk_python/operation/invoke.py

@@ -69,7 +69,7 @@ def invoke_handler(
         # Operation failed, throw the exact same error on replay as the checkpointed failure
         checkpointed_result.raise_callable_error()
 
-    if checkpointed_result.is_started():
+    if checkpointed_result.is_started() or checkpointed_result.is_pending():
         # Operation is still running, suspend until completion
         logger.debug(
             "⏳ Invoke %s still in progress, suspending",

tests/operation/invoke_test.py

@@ -164,15 +164,16 @@ def test_invoke_handler_already_timed_out():
         )
 
 
-def test_invoke_handler_already_started():
+@pytest.mark.parametrize("status", [OperationStatus.STARTED, OperationStatus.PENDING])
+def test_invoke_handler_already_started(status):
     """Test invoke_handler when operation is already started."""
     mock_state = Mock(spec=ExecutionState)
     mock_state.durable_execution_arn = "test_arn"
 
     operation = Operation(
         operation_id="invoke6",
         operation_type=OperationType.CHAINED_INVOKE,
-        status=OperationStatus.STARTED,
+        status=status,
         chained_invoke_details=ChainedInvokeDetails(),
     )
     mock_result = CheckpointedResult.create_from_operation(operation)
@@ -188,15 +189,16 @@ def test_invoke_handler_already_started():
         )
 
 
-def test_invoke_handler_already_started_with_timeout():
+@pytest.mark.parametrize("status", [OperationStatus.STARTED, OperationStatus.PENDING])
+def test_invoke_handler_already_started_with_timeout(status):
     """Test invoke_handler when operation is already started with timeout config."""
     mock_state = Mock(spec=ExecutionState)
     mock_state.durable_execution_arn = "test_arn"
 
     operation = Operation(
         operation_id="invoke7",
         operation_type=OperationType.CHAINED_INVOKE,
-        status=OperationStatus.STARTED,
+        status=status,
         chained_invoke_details=ChainedInvokeDetails(),
     )
     mock_result = CheckpointedResult.create_from_operation(operation)
@@ -402,15 +404,16 @@ def test_suspend_with_optional_timeout_negative_timeout():
     assert "test message" in str(exc_info.value)
 
 
-def test_invoke_handler_with_operation_name():
+@pytest.mark.parametrize("status", [OperationStatus.STARTED, OperationStatus.PENDING])
+def test_invoke_handler_with_operation_name(status: OperationStatus):
     """Test invoke_handler uses operation name in logs when available."""
     mock_state = Mock(spec=ExecutionState)
     mock_state.durable_execution_arn = "test_arn"
 
     operation = Operation(
         operation_id="invoke14",
         operation_type=OperationType.CHAINED_INVOKE,
-        status=OperationStatus.STARTED,
+        status=status,
         chained_invoke_details=ChainedInvokeDetails(),
     )
     mock_result = CheckpointedResult.create_from_operation(operation)
@@ -426,15 +429,16 @@ def test_invoke_handler_with_operation_name():
         )
 
 
-def test_invoke_handler_without_operation_name():
+@pytest.mark.parametrize("status", [OperationStatus.STARTED, OperationStatus.PENDING])
+def test_invoke_handler_without_operation_name(status: OperationStatus):
     """Test invoke_handler uses function name in logs when no operation name."""
     mock_state = Mock(spec=ExecutionState)
     mock_state.durable_execution_arn = "test_arn"
 
     operation = Operation(
         operation_id="invoke15",
         operation_type=OperationType.CHAINED_INVOKE,
-        status=OperationStatus.STARTED,
+        status=status,
         chained_invoke_details=ChainedInvokeDetails(),
     )
     mock_result = CheckpointedResult.create_from_operation(operation)