Skip to content
Open
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 @@ -406,7 +406,9 @@ def complete_success(self, result: str | None, now: datetime | None = None) -> N
self.close_status = ExecutionStatus.SUCCEEDED
self._end_execution(OperationStatus.SUCCEEDED, now)

def complete_fail(self, error: ErrorObject, now: datetime | None = None) -> None:
def complete_fail(
self, error: ErrorObject | None, now: datetime | None = None
) -> None:
"""Complete execution with failure (DecisionType.FAIL_WORKFLOW_EXECUTION)."""
self.result = DurableExecutionInvocationOutput(
status=InvocationStatus.FAILED, error=error
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1285,9 +1285,7 @@ def _validate_invocation_response_and_store(
)
raise InvalidParameterValueException(msg_failed_result)
logger.info("[%s] Execution failed", execution_arn)
self._complete_workflow(
execution_arn, result=None, error=response.error
)
self._fail_workflow(execution_arn, response.error)

This comment was marked as outdated.

@hln33 hln33 Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is one small obstacle to doing that. There currently isn't anything in the public API of DurableFunctionTestRunner or DurableFunctionTestResult that exposes an execution's status in the emulator. The public runner result only exposes the returned durable invocation status/result, not execution.

We could use private internals, like DurableFunctionTestRunner._executor, to get execution information, but that feels hacky to me for an e2e style test.

If we want to add an e2e test for this case, then I think we should first add a small public runner accessor for execution details/status, then use that in the test.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel like we should expose a method for the overall execution status on the test runner here. Do we have this in the TS testing library? Thinking something like result.execution_status?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice idea @bchampp :-)

result.execution_status sourced from close_status


case InvocationStatus.SUCCEEDED:
if response.error is not None:
Expand Down Expand Up @@ -1591,7 +1589,7 @@ def _complete_workflow(
else:
self.complete_execution(execution_arn, result)

def _fail_workflow(self, execution_arn: str, error: ErrorObject):
def _fail_workflow(self, execution_arn: str, error: ErrorObject | None):
"""Fail workflow with terminal state validation."""
execution = self._store.load(execution_arn)

Expand Down Expand Up @@ -1671,8 +1669,8 @@ def complete_execution(self, execution_arn: str, result: str | None = None) -> N
raise IllegalStateException(msg)
self._complete_events(execution_arn=execution_arn)

def fail_execution(self, execution_arn: str, error: ErrorObject) -> None:
"""Fail execution with error (FAIL_WORKFLOW_EXECUTION decision)."""
def fail_execution(self, execution_arn: str, error: ErrorObject | None) -> None:
"""Fail execution with optional error (FAIL_WORKFLOW_EXECUTION decision)."""
logger.error("[%s] Completing execution with error: %s", execution_arn, error)
execution: Execution = self._store.load(execution_arn=execution_arn)
execution.complete_fail(error=error, now=self._clock.now())
Expand All @@ -1688,7 +1686,7 @@ def on_completed(self, execution_arn: str, result: str | None = None) -> None:
"""Complete execution successfully. Observer method triggered by notifier."""
self.complete_execution(execution_arn, result)

def on_failed(self, execution_arn: str, error: ErrorObject) -> None:
def on_failed(self, execution_arn: str, error: ErrorObject | None) -> None:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This type widening is right, and it exposes a pre-existing emulator divergence... Common typing W.

The checkpoint EXECUTION FAIL processor (checkpoint/processors/execution.py) synthesizes a generic ErrorObject when the update has no error, but it should instead preserve null there.

Proposal: remove the synthetic fallback, and widen ExecutionNotifier.notify_failed + ExecutionObserver.on_failed to ErrorObject | None to match the widened null signature.

"""Fail execution. Observer method triggered by notifier."""
self.fail_execution(execution_arn, error)

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
"""End-to-end child context failure handling through the test runner."""

import json
from typing import Any

from aws_durable_execution_sdk_python.config import StepConfig
from aws_durable_execution_sdk_python.context import (
DurableContext,
durable_step,
durable_with_child_context,
)
from aws_durable_execution_sdk_python.execution import durable_execution
from aws_durable_execution_sdk_python.lambda_service import (
InvocationStatus,
OperationStatus,
)
from aws_durable_execution_sdk_python.retries import RetryPresets
from aws_durable_execution_sdk_python.types import StepContext

from aws_durable_execution_sdk_python_testing.runner import (
ContextOperation,
DurableFunctionTestResult,
DurableFunctionTestRunner,
)


def test_caught_child_context_failure_does_not_fail_root_execution() -> None:
@durable_step
def failing_step(step_context: StepContext) -> str: # noqa: ARG001
msg = "Child step failed"
raise RuntimeError(msg)

@durable_with_child_context
def failing_child(ctx: DurableContext) -> str:
return ctx.step(
failing_step(),
config=StepConfig(retry_strategy=RetryPresets.none()),
)

@durable_step
def recovery_step(step_context: StepContext, value: str) -> str: # noqa: ARG001
return value

@durable_execution
def handler(event: Any, context: DurableContext) -> str: # noqa: ARG001
try:
context.run_in_child_context(failing_child(), name="failing-child")
except Exception:
pass

return context.step(recovery_step("handled"))

with DurableFunctionTestRunner(handler=handler, execution_timeout=10) as runner:
result: DurableFunctionTestResult = runner.run(input="input str")

assert result.status is InvocationStatus.SUCCEEDED
assert result.result == json.dumps("handled")

child_op: ContextOperation = result.get_context("failing-child")
assert child_op.status == OperationStatus.FAILED

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

child_op.status is OperationStatus.FAILED

assert child_op.error is not None
assert child_op.error.message is not None
assert "Child step failed" in child_op.error.message
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,41 @@ def test_create_execution_failed():
assert event.execution_failed_details.error.payload.message == "Execution failed"


def test_create_execution_failed_without_error_payload():
from aws_durable_execution_sdk_python.execution import (
DurableExecutionInvocationOutput,
InvocationStatus,
)

operation = create_mock_operation("op-1", status=OperationStatus.FAILED)
operation.end_timestamp = datetime.now(UTC)

error_result = DurableExecutionInvocationOutput(
status=InvocationStatus.FAILED,
error=None,
)
context = EventCreationContext.create(
operation=operation,
event_id=3,
durable_execution_arn="arn:test",
start_input=StartDurableExecutionInput(
account_id="123",
function_name="test",
function_qualifier="$LATEST",
execution_name="test",
execution_timeout_seconds=300,
execution_retention_period_days=7,
),
result=error_result,
include_execution_data=True,
)
event = Event.create_execution_event(context)

assert event.event_type == "ExecutionFailed"
assert event.execution_failed_details.error is not None
assert event.execution_failed_details.error.payload is None


def test_create_execution_timed_out():
from aws_durable_execution_sdk_python.execution import (
DurableExecutionInvocationOutput,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -507,6 +507,26 @@ def test_complete_fail():
assert execution.result.error == error


def test_complete_fail_without_error():
"""Test complete_fail preserves a missing error payload."""
start_input = StartDurableExecutionInput(
account_id="123456789012",
function_name="test-function",
function_qualifier="$LATEST",
execution_name="test-execution",
execution_timeout_seconds=300,
execution_retention_period_days=7,
invocation_id="test-invocation-id",
)
execution = Execution("test-arn", start_input, [Mock()])

execution.complete_fail(None)

assert execution.is_complete is True
assert execution.result.status == InvocationStatus.FAILED

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

execution.result.status is InvocationStatus.FAILED

assert execution.result.error is None


def test_find_operation_exists():
"""Test find_operation method when operation exists."""
start_input = StartDurableExecutionInput(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -310,6 +310,42 @@ def test_should_complete_workflow_with_error_when_invocation_fails(
mock_fail.assert_called_once_with("test-arn", failed_response.error)


def test_validate_invocation_response_failed_without_error_still_fails():
"""FAILED without an error must fail (not succeed), preserving the null error."""

store = InMemoryExecutionStore()
executor = Executor(store, Mock(), Mock(), Mock())

start_input = StartDurableExecutionInput(
account_id="123456789012",
function_name="test-function",
function_qualifier="$LATEST",
execution_name="test-execution",
execution_timeout_seconds=300,
execution_retention_period_days=7,
invocation_id="test-invocation-id",
)
execution = Execution.new(start_input)
execution.start()
store.save(execution)

response = DurableExecutionInvocationOutput(
status=InvocationStatus.FAILED, error=None
)

executor._validate_invocation_response_and_store( # noqa: SLF001
execution.durable_execution_arn, response, execution
)

stored = store.load(execution.durable_execution_arn)
assert stored.is_complete is True
assert stored.close_status is not None
assert stored.close_status.value == "FAILED"

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you can compare enums directly

assert stored.close_status is ExecutionStatus.FAILED

assert stored.result is not None
assert stored.result.status == InvocationStatus.FAILED

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

assert stored.result.status is InvocationStatus.FAILED

assert stored.result.error is None


def test_should_complete_workflow_with_result_when_invocation_succeeds(
executor, mock_store, mock_scheduler, mock_invoker, start_input
):
Expand Down
Loading