From efcbca0e72af2aa440bafdcd927d64e11238e0c7 Mon Sep 17 00:00:00 2001 From: Aditya Ganti Date: Mon, 6 Apr 2026 22:54:30 -0400 Subject: [PATCH 1/3] fix: remove timeout from Invoke Config (v2 breaking change) --- .../config.py | 12 +-- .../operation/invoke.py | 2 +- .../tests/config_test.py | 1 - .../tests/context_test.py | 3 +- .../tests/operation/invoke_test.py | 90 ++++--------------- 5 files changed, 18 insertions(+), 90 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py index 6763a53e..71eb01ea 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py @@ -479,13 +479,9 @@ class InvokeConfig(Generic[P, R]): Configuration for invoke operations. This class configures how function invocations are executed, including - timeout behavior, serialization, and tenant isolation. + serialization and tenant isolation. Args: - timeout: Maximum duration to wait for the invoked function to complete. - Default is no timeout. Use this to prevent long-running invocations - from blocking execution indefinitely. - serdes_payload: Custom serialization/deserialization for the payload sent to the invoked function. Defaults to DEFAULT_JSON_SERDES when not set. @@ -499,16 +495,10 @@ class InvokeConfig(Generic[P, R]): """ # retry_strategy: Callable[[Exception, int], RetryDecision] | None = None - timeout: Duration = field(default_factory=Duration) serdes_payload: SerDes[P] | None = None serdes_result: SerDes[R] | None = None tenant_id: str | None = None - @property - def timeout_seconds(self) -> int: - """Get timeout in seconds.""" - return self.timeout.to_seconds() - @dataclass(frozen=True) class CallbackConfig: diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py index 9288c983..6d181341 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/operation/invoke.py @@ -166,7 +166,7 @@ def execute(self, _checkpointed_result: CheckpointedResult) -> R: ExecutionError: If suspend doesn't raise (should never happen) """ msg: str = f"Invoke {self.operation_identifier.operation_id} started, suspending for completion" - suspend_with_optional_resume_delay(msg, self.config.timeout_seconds) + suspend_with_optional_resume_delay(msg) # This line should never be reached since suspend_with_optional_resume_delay always raises error_msg: str = "suspend_with_optional_resume_delay should have raised an exception, but did not." raise ExecutionError(error_msg) from None diff --git a/packages/aws-durable-execution-sdk-python/tests/config_test.py b/packages/aws-durable-execution-sdk-python/tests/config_test.py index a8af487d..17069355 100644 --- a/packages/aws-durable-execution-sdk-python/tests/config_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/config_test.py @@ -271,7 +271,6 @@ def test_invoke_config_defaults(): """Test InvokeConfig defaults.""" config = InvokeConfig() assert config.tenant_id is None - assert config.timeout_seconds == 0 def test_invoke_config_with_tenant_id(): 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 d0061aa9..d91c0951 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -644,7 +644,7 @@ def test_invoke_with_name_and_config(mock_executor_class): mock_state.durable_execution_arn = ( "arn:aws:durable:us-east-1:123456789012:execution/test" ) - config = InvokeConfig[str, str](timeout=Duration.from_seconds(30)) + config = InvokeConfig[str, str]() context = create_test_context(state=mock_state) [context._create_step_id() for _ in range(5)] # Set counter to 5 # noqa: SLF001 @@ -790,7 +790,6 @@ def test_invoke_with_custom_serdes(mock_executor_class): config = InvokeConfig[dict, dict]( serdes_payload=payload_serdes, serdes_result=result_serdes, - timeout=Duration.from_minutes(1), ) context = create_test_context(state=mock_state) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index c6a247f1..81d1a283 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -7,7 +7,7 @@ import pytest -from aws_durable_execution_sdk_python.config import Duration, InvokeConfig +from aws_durable_execution_sdk_python.config import InvokeConfig from aws_durable_execution_sdk_python.exceptions import ( CallableRuntimeError, ExecutionError, @@ -218,8 +218,8 @@ def test_invoke_handler_already_started(status): @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.""" +def test_invoke_handler_already_started_suspends(status): + """Test invoke_handler when operation is already started suspends indefinitely.""" mock_state = Mock(spec=ExecutionState) mock_state.durable_execution_arn = "test_arn" @@ -232,9 +232,9 @@ def test_invoke_handler_already_started_with_timeout(status): mock_result = CheckpointedResult.create_from_operation(operation) mock_state.get_checkpoint_result.return_value = mock_result - config = InvokeConfig[str, str](timeout=Duration.from_seconds(30)) + config = InvokeConfig[str, str]() - with pytest.raises(TimedSuspendExecution): + with pytest.raises(SuspendExecution): invoke_handler( function_name="test_function", payload="test_input", @@ -261,7 +261,7 @@ def test_invoke_handler_new_operation(): started = CheckpointedResult.create_from_operation(started_op) mock_state.get_checkpoint_result.side_effect = [not_found, started] - config = InvokeConfig[str, str](timeout=Duration.from_minutes(1)) + config = InvokeConfig[str, str]() with pytest.raises( SuspendExecution, match="Invoke invoke8 started, suspending for completion" @@ -288,62 +288,6 @@ def test_invoke_handler_new_operation(): assert operation_update.chained_invoke_options.function_name == "test_function" -def test_invoke_handler_new_operation_with_timeout(): - """Test invoke_handler when starting a new operation with timeout.""" - mock_state = Mock(spec=ExecutionState) - mock_state.durable_execution_arn = "test_arn" - - not_found = CheckpointedResult.create_not_found() - started_op = Operation( - operation_id="invoke_test", - operation_type=OperationType.CHAINED_INVOKE, - status=OperationStatus.STARTED, - ) - started = CheckpointedResult.create_from_operation(started_op) - mock_state.get_checkpoint_result.side_effect = [not_found, started] - - config = InvokeConfig[str, str](timeout=Duration.from_seconds(30)) - - with pytest.raises(TimedSuspendExecution): - invoke_handler( - function_name="test_function", - payload="test_input", - state=mock_state, - operation_identifier=OperationIdentifier( - "invoke9", OperationSubType.CHAINED_INVOKE, None, "test_invoke" - ), - config=config, - ) - - -def test_invoke_handler_new_operation_no_timeout(): - """Test invoke_handler when starting a new operation without timeout.""" - mock_state = Mock(spec=ExecutionState) - mock_state.durable_execution_arn = "test_arn" - - not_found = CheckpointedResult.create_not_found() - started_op = Operation( - operation_id="invoke_test", - operation_type=OperationType.CHAINED_INVOKE, - status=OperationStatus.STARTED, - ) - started = CheckpointedResult.create_from_operation(started_op) - mock_state.get_checkpoint_result.side_effect = [not_found, started] - - config = InvokeConfig[str, str](timeout=Duration.from_seconds(0)) - - with pytest.raises(SuspendExecution): - invoke_handler( - function_name="test_function", - payload="test_input", - state=mock_state, - operation_identifier=OperationIdentifier( - "invoke10", OperationSubType.CHAINED_INVOKE, None, "test_invoke" - ), - config=config, - ) - - def test_invoke_handler_no_config(): """Test invoke_handler when no config is provided.""" mock_state = Mock(spec=ExecutionState) @@ -1067,8 +1011,8 @@ def test_invoke_immediate_response_already_completed(): assert mock_state.get_checkpoint_result.call_count == 1 -def test_invoke_immediate_response_with_timeout_immediate_success(): - """Test immediate success with timeout configuration.""" +def test_invoke_immediate_response_immediate_success(): + """Test immediate success response.""" mock_state = Mock(spec=ExecutionState) mock_state.durable_execution_arn = "test_arn" @@ -1079,13 +1023,13 @@ def test_invoke_immediate_response_with_timeout_immediate_success(): operation_type=OperationType.CHAINED_INVOKE, status=OperationStatus.SUCCEEDED, chained_invoke_details=ChainedInvokeDetails( - result=json.dumps("timeout_result") + result=json.dumps("immediate_result") ), ) succeeded = CheckpointedResult.create_from_operation(succeeded_op) mock_state.get_checkpoint_result.side_effect = [not_found, succeeded] - config = InvokeConfig[str, str](timeout=Duration.from_seconds(30)) + config = InvokeConfig[str, str]() result = invoke_handler( function_name="test_function", @@ -1098,15 +1042,12 @@ def test_invoke_immediate_response_with_timeout_immediate_success(): ) # Verify result was returned without suspend - assert result == "timeout_result" + assert result == "immediate_result" assert mock_state.get_checkpoint_result.call_count == 2 -def test_invoke_immediate_response_with_timeout_no_immediate_response(): - """Test no immediate response with timeout configuration. - - When no immediate response, operation should suspend with timeout. - """ +def test_invoke_immediate_response_no_immediate_response(): + """Test no immediate response — operation suspends indefinitely.""" mock_state = Mock(spec=ExecutionState) mock_state.durable_execution_arn = "test_arn" @@ -1120,10 +1061,9 @@ def test_invoke_immediate_response_with_timeout_no_immediate_response(): started = CheckpointedResult.create_from_operation(started_op) mock_state.get_checkpoint_result.side_effect = [not_found, started] - config = InvokeConfig[str, str](timeout=Duration.from_seconds(30)) + config = InvokeConfig[str, str]() - # Verify operation suspends with timeout - with pytest.raises(TimedSuspendExecution): + with pytest.raises(SuspendExecution): invoke_handler( function_name="test_function", payload="test_input", From 7c0bb96c022d59f72a4f9427f3b2bb7ce539b060 Mon Sep 17 00:00:00 2001 From: Aditya Ganti Date: Mon, 13 Jul 2026 18:48:06 -0400 Subject: [PATCH 2/3] fix: remove duplicate invoke immediate response tests --- .../tests/operation/invoke_test.py | 69 ------------------- 1 file changed, 69 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index 81d1a283..5d43863c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -1011,75 +1011,6 @@ def test_invoke_immediate_response_already_completed(): assert mock_state.get_checkpoint_result.call_count == 1 -def test_invoke_immediate_response_immediate_success(): - """Test immediate success response.""" - mock_state = Mock(spec=ExecutionState) - mock_state.durable_execution_arn = "test_arn" - - # First call: not found, second call: succeeded - not_found = CheckpointedResult.create_not_found() - succeeded_op = Operation( - operation_id="invoke_immediate_8", - operation_type=OperationType.CHAINED_INVOKE, - status=OperationStatus.SUCCEEDED, - chained_invoke_details=ChainedInvokeDetails( - result=json.dumps("immediate_result") - ), - ) - succeeded = CheckpointedResult.create_from_operation(succeeded_op) - mock_state.get_checkpoint_result.side_effect = [not_found, succeeded] - - config = InvokeConfig[str, str]() - - result = invoke_handler( - function_name="test_function", - payload="test_input", - state=mock_state, - operation_identifier=OperationIdentifier( - "invoke_immediate_8", OperationSubType.CHAINED_INVOKE, None, "test_invoke" - ), - config=config, - ) - - # Verify result was returned without suspend - assert result == "immediate_result" - assert mock_state.get_checkpoint_result.call_count == 2 - - -def test_invoke_immediate_response_no_immediate_response(): - """Test no immediate response — operation suspends indefinitely.""" - mock_state = Mock(spec=ExecutionState) - mock_state.durable_execution_arn = "test_arn" - - # First call: not found, second call: started - not_found = CheckpointedResult.create_not_found() - started_op = Operation( - operation_id="invoke_immediate_9", - operation_type=OperationType.CHAINED_INVOKE, - status=OperationStatus.STARTED, - ) - started = CheckpointedResult.create_from_operation(started_op) - mock_state.get_checkpoint_result.side_effect = [not_found, started] - - config = InvokeConfig[str, str]() - - with pytest.raises(SuspendExecution): - invoke_handler( - function_name="test_function", - payload="test_input", - state=mock_state, - operation_identifier=OperationIdentifier( - "invoke_immediate_9", - OperationSubType.CHAINED_INVOKE, - None, - "test_invoke", - ), - config=config, - ) - - assert mock_state.get_checkpoint_result.call_count == 2 - - def test_invoke_immediate_response_with_custom_serdes(): """Test immediate success with custom serialization.""" mock_state = Mock(spec=ExecutionState) From d973096283a765fe7ac273a1d7bd76cd398ed78d Mon Sep 17 00:00:00 2001 From: Aditya Ganti Date: Mon, 13 Jul 2026 23:17:49 -0400 Subject: [PATCH 3/3] test: assert exact SuspendExecution type in already-started test Address review feedback: pytest.raises(SuspendExecution) alone would still pass if we regressed to TimedSuspendExecution, since it is a subclass. Assert the exact class instead. --- .../tests/operation/invoke_test.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index 5d43863c..dda39b21 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -234,7 +234,7 @@ def test_invoke_handler_already_started_suspends(status): config = InvokeConfig[str, str]() - with pytest.raises(SuspendExecution): + with pytest.raises(SuspendExecution) as exc_info: invoke_handler( function_name="test_function", payload="test_input", @@ -244,6 +244,7 @@ def test_invoke_handler_already_started_suspends(status): ), config=config, ) + assert type(exc_info.value) is SuspendExecution def test_invoke_handler_new_operation():