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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -152,33 +152,27 @@ def execute(self, checkpointed_result: CheckpointedResult) -> T:
Suspends if condition not met
Raises error if check function fails
"""
# Determine current state from checkpoint
if checkpointed_result.is_started_or_ready() and checkpointed_result.result:
try:
try:
# Determine current state from checkpoint
if checkpointed_result.is_started_or_ready() and checkpointed_result.result:
current_state = deserialize(
serdes=self.config.serdes,
data=checkpointed_result.result,
operation_id=self.operation_identifier.operation_id,
durable_execution_arn=self.state.durable_execution_arn,
)
except Exception:
# Default to initial state if there's an error getting checkpointed state
logger.exception(
"⚠️ wait_for_condition failed to deserialize state for id: %s, name: %s. Using initial state.",
self.operation_identifier.operation_id,
self.operation_identifier.name,
)
else:
current_state = self.config.initial_state
else:
current_state = self.config.initial_state

# Get attempt number - current attempt is checkpointed attempts + 1
# The checkpoint stores completed attempts, so the current attempt being executed is one more
attempt: int = 1
if checkpointed_result.operation and checkpointed_result.operation.step_details:
attempt = checkpointed_result.operation.step_details.attempt + 1
# Get attempt number - current attempt is checkpointed attempts + 1
# The checkpoint stores completed attempts, so the current attempt being executed is one more
attempt: int = 1
if (
checkpointed_result.operation
and checkpointed_result.operation.step_details
):
attempt = checkpointed_result.operation.step_details.attempt + 1

try:
# Execute the check function with the injected logger
check_context = WaitForConditionCheckContext(
logger=self.context_logger.with_log_info(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -332,8 +332,8 @@ def check_func(state, context):
assert result == 6 # 5 (initial) + 1


def test_wait_for_condition_retry_invalid_json_state():
"""Test wait_for_condition on retry with invalid JSON state."""
def test_wait_for_condition_retry_invalid_json_state_fails():
"""Test invalid checkpointed state fails instead of restarting polling."""
mock_state = Mock(spec=ExecutionState)
mock_state.durable_execution_arn = "arn:aws:test"
operation = Operation(
Expand Down Expand Up @@ -362,15 +362,21 @@ def check_func(state, context):
wait_strategy=lambda s, a: WaitForConditionDecision.stop_polling(),
)

result = wait_for_condition_handler(
state=mock_state,
operation_identifier=op_id,
check=check_func,
config=config,
context_logger=mock_logger,
)
with pytest.raises(WaitForConditionError) as exc_info:
wait_for_condition_handler(
state=mock_state,
operation_identifier=op_id,
check=check_func,
config=config,
context_logger=mock_logger,
)

assert result == 6 # Falls back to initial state
assert exc_info.value.error_type == "ExecutionError"
mock_state.create_checkpoint.assert_called_once()
assert (
mock_state.create_checkpoint.call_args.kwargs["operation_update"].action
== OperationAction.FAIL
)


def test_wait_for_condition_check_function_exception():
Expand Down
Loading