From 26536b6d843dfea4e63502b2fdb2653e899d7c0f Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:33:23 -0700 Subject: [PATCH 1/2] test: cover terminal OTel status and event times --- .../e2e/test_terminal_failure_bounds_int.py | 133 ++++++++++++++++++ .../tests/test_status_mapping.py | 81 ++++++++++- 2 files changed, 207 insertions(+), 7 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py new file mode 100644 index 000000000..c2d14256b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py @@ -0,0 +1,133 @@ +"""Caught operation failures preserve telemetry through real suspension/replay.""" + +import json +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import ChildConfig, Duration +from aws_durable_execution_sdk_python.exceptions import ChildContextError, InvokeError +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize("failure_kind", ["child", "virtual", "invoke"]) +@pytest.mark.parametrize("resume", [False, True]) +def test_caught_failure_keeps_events_bounded_and_workflow_successful( + view: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + failure_kind: str, + resume: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + factory = view( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + saved_calls: list[str] = [] + failure_calls: list[str] = [] + observations: list[OperationEndInfo] = [] + + class Observe(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "caught-failure": + observations.append(info) + + class ObserveFactory: + def create_plugin(self, _info: InvocationStartInfo) -> Observe: + return Observe() + + def saved_step(_context: Any) -> str: + saved_calls.append("saved") + return "saved" + + def failing_child(_context: DurableContext) -> str: + failure_calls.append("failed") + raise ValueError("actual child failure") + + def failing_function(_event: Any, _context: Any) -> None: + failure_calls.append("failed") + raise ValueError("actual invoke failure") + + def body(_event: Any, context: DurableContext) -> str: + saved = context.step(saved_step, name="saved-step") + try: + if failure_kind == "invoke": + context.invoke("failing-target", {}, name="caught-failure") + else: + context.run_in_child_context( + failing_child, + name="caught-failure", + config=ChildConfig(is_virtual=failure_kind == "virtual"), + ) + except (ChildContextError, InvokeError): + if resume: + context.wait(Duration.from_seconds(1), name="replay-wait") + return saved + raise AssertionError("Expected the original operation failure") + + try: + with DurableFunctionTestRunner( + handler=durable_execution(body, plugins=[factory, ObserveFactory()]), + poll_interval=0.01, + ) as runner: + runner.register_function("failing-target", failing_function) + result = runner.run(input="{}", timeout=15) + assert result.status.value == "SUCCEEDED" + assert result.result is not None and json.loads(result.result) == "saved" + assert saved_calls == ["saved"] + # Virtual contexts intentionally execute on replay; checkpointed child + # failures and invoke outcomes must return their stored failure instead. + assert len(failure_calls) == (2 if resume and failure_kind == "virtual" else 1) + assert observations + if failure_kind == "virtual": + assert all(info.start_time is None for info in observations) + spans = exporter.get_finished_spans() + failed = [ + span + for span in spans + if span.name == "caught-failure" + and span.attributes is not None + and span.attributes.get("durable.operation.status") == "FAILED" + ] + assert failed + for span in failed: + assert span.status.status_code is StatusCode.ERROR + (event,) = span.events + assert span.start_time is not None and span.end_time is not None + assert span.start_time <= event.timestamp == span.end_time + assert event.attributes is not None + assert event.attributes["exception.message"] in { + "actual child failure", + "actual invoke failure", + } + invocations = [span for span in spans if span.name == "Invocation"] + assert len(invocations) == (2 if resume else 1) + assert all(span.status.status_code is StatusCode.OK for span in invocations) + (workflow,) = [span for span in spans if span.name == "Workflow"] + assert workflow.status.status_code is StatusCode.OK + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py index 0b4bcc324..57cdc3c9d 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py @@ -1,6 +1,7 @@ """Shared status regressions through both public plugin lifecycles.""" from collections.abc import Iterator +from dataclasses import replace from datetime import UTC, datetime, timedelta from typing import Any @@ -69,13 +70,7 @@ def instrumented_view(request: pytest.FixtureRequest) -> Iterator[InstrumentedVi @pytest.mark.parametrize( "status", - [ - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.CANCELLED, - OperationStatus.TIMED_OUT, - OperationStatus.STOPPED, - ], + list(OperationStatus), ) @pytest.mark.parametrize("with_error", [False, True]) @pytest.mark.parametrize("started_here", [False, True]) @@ -129,6 +124,8 @@ def test_operation_status_and_exception( expected = ( StatusCode.ERROR if with_error + or status + in {OperationStatus.FAILED, OperationStatus.TIMED_OUT, OperationStatus.STOPPED} else ( StatusCode.OK if status is OperationStatus.SUCCEEDED else StatusCode.UNSET ) @@ -144,6 +141,76 @@ def test_operation_status_and_exception( assert not span.events +@pytest.mark.parametrize("start_time", [None, START, END]) +@pytest.mark.parametrize("end_time", [END, None]) +def test_terminal_exception_uses_normalized_span_end( + instrumented_view: InstrumentedView, + start_time: datetime | None, + end_time: datetime | None, +) -> None: + """Delayed and checkpointless completions retain bounded diagnostic times.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + plugin.on_operation_end( + replace( + _operation_end_info(ErrorObject("stored failure", "ValueError", None, None)), + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + start_time=start_time, + end_time=end_time, + ) + ) + _end_invocation(plugin) + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + (event,) = span.events + assert span.start_time is not None and span.end_time is not None + assert span.start_time <= event.timestamp == span.end_time + + +@pytest.mark.parametrize("with_error", [False, True]) +@pytest.mark.parametrize("end_time", [START, END, None]) +def test_failed_attempt_exception_uses_normalized_span_end( + instrumented_view: InstrumentedView, + with_error: bool, + end_time: datetime | None, +) -> None: + """An attempt failure reports its outcome without fabricating diagnostics.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + operation: dict[str, Any] = dict( + operation_id="attempt", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="attempt", + parent_id=None, + start_time=START, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + plugin.on_user_function_start(UserFunctionStartInfo(**operation)) + plugin.on_user_function_end( + UserFunctionEndInfo( + **operation, + outcome=UserFunctionOutcome.FAILED, + end_time=end_time, + error=ErrorObject("stored failure", "ValueError", None, None) + if with_error + else None, + ) + ) + _end_invocation(plugin) + span = next(s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1") + assert span.status.status_code is StatusCode.ERROR + if with_error: + (event,) = span.events + assert span.start_time is not None and span.end_time is not None + assert span.start_time <= event.timestamp == span.end_time + else: + assert not span.events + + def test_retry_then_resume_exports_retrying( instrumented_view: InstrumentedView, ) -> None: From 82fc8767225cbbf4504a7c89ea033afc2b306b37 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:36:08 -0700 Subject: [PATCH 2/2] fix: align terminal OTel status and event times --- .../README.md | 27 +++++++++++++---- .../exception_events.py | 11 ++++--- .../execution_plugin.py | 21 ++++++++++---- .../invocation_plugin.py | 29 +++++++++++++++---- .../e2e/test_terminal_failure_bounds_int.py | 5 ++-- .../tests/test_status_mapping.py | 8 +++-- 6 files changed, 77 insertions(+), 24 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 8e9ed0816..d9c503187 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -401,11 +401,28 @@ each durable span in the same invocation. ### Status attributes `durable.invocation.status` uses `RETRYING` when the core plugin hook reports -`InvocationStatus.RETRY`. The core enum remains unchanged. Operation spans use -OTel `OK` only for `SUCCEEDED`, `ERROR` when error details are delivered, and -`UNSET` for other outcomes without error details, including `FAILED`, -`CANCELLED`, `TIMED_OUT`, and `STOPPED`. The original durable operation status -remains in `durable.operation.status`. +`InvocationStatus.RETRY`, with OTel `UNSET`. The core enum remains unchanged. +Operation spans use the following mapping in both views: + +| Durable operation outcome | Error details | OTel status | Exception event | +| --- | --- | --- | --- | +| `SUCCEEDED` | Absent | `OK` | None | +| `FAILED`, `TIMED_OUT`, `STOPPED` | Absent | `ERROR` | None | +| `CANCELLED` and other non-success outcomes | Absent | `UNSET` | None | +| Any outcome | Present | `ERROR` | Available diagnostics | + +The original durable operation status remains in `durable.operation.status`. +Failure status does not fabricate an exception when diagnostics are absent, +including for failed step attempts. A caught operation failure leaves the +enclosing Invocation and Workflow spans successful if the handler succeeds. + +Terminal operation and attempt exception events use the same normalized end +timestamp as their span, including delayed or replayed completion delivery. +Execution view retains durable timing and normalizes an end at or before its +start; Invocation view retains its invocation-local timestamp ordering. A +checkpointless context with no durable start uses its completion time as the +lower bound in Execution view. When an end timestamp is absent, the current time +is chosen once for both the event and span end. ### Invocation context isolation diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py index 872a7bedf..67156b63b 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py @@ -40,16 +40,19 @@ def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]: return attributes -def record_error_object_exception(span: Any, error: ErrorObject | None) -> None: +def record_error_object_exception( + span: Any, error: ErrorObject | None, *, timestamp: int | None = None +) -> None: """Record a span exception that preserves the SDK's original error type. - ``error`` may be ``None`` for a failed outcome that carried no metadata; a - generic exception is recorded in that case. + A failed outcome without an error object has no diagnostic event. Terminal + callers supply the normalized span-end timestamp so delayed delivery does + not place an event outside the span. """ if error is None: - span.record_exception(Exception(_DEFAULT_MESSAGE)) return span.record_exception( exception_from_error_object(error), attributes=error_object_exception_attributes(error), + timestamp=timestamp, ) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index 356990ede..2c52e294b 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -892,16 +892,25 @@ def on_operation_end(self, info: OperationEndInfo) -> None: span_id_override=span_id_override, ) + end_timestamp = _to_otel_timestamp( + end_time or datetime.datetime.now(datetime.UTC) + ) if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) + elif info.status in { + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.STOPPED, + }: + span.set_status(StatusCode.ERROR) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) self._note_parent_end(info.parent_id, end_time) popped = self._pop_span(info.operation_id) if popped is not None: - popped.end(end_time=_to_otel_timestamp(end_time)) + popped.end(end_time=end_timestamp) def _start_span( self, @@ -1024,18 +1033,20 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: and info.outcome is not UserFunctionOutcome.INCOMPLETE ): span.set_attributes(self._operation_attributes(info)) + end_timestamp = _to_otel_timestamp( + end_time or datetime.datetime.now(datetime.UTC) + ) if info.outcome is UserFunctionOutcome.FAILED: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) else: span.set_status(StatusCode.OK) - end_time = ensure_end_after_start(info.start_time, info.end_time) popped = self._pop_span(key) if popped is not None: - popped.end(end_time=_to_otel_timestamp(end_time)) + popped.end(end_time=end_timestamp) # Restore the enclosing context by releasing the scope this user # function attached, so the parent operation (or, at the top level, the diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 4eab6c7c3..81a2e00d3 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -7,6 +7,7 @@ import threading from typing import Any, TYPE_CHECKING, cast +from aws_durable_execution_sdk_python.lambda_service import ErrorObject from aws_durable_execution_sdk_python.plugin import ( DurableInstrumentationPlugin, InvocationEndInfo, @@ -563,8 +564,12 @@ def _start_span( return span def _end_span( - self, operation_id: str | None, end_timestamp: datetime.datetime | None = None - ): + self, + operation_id: str | None, + end_timestamp: datetime.datetime | None = None, + *, + error: ErrorObject | None = None, + ) -> None: """End and unregister the active span for an operation ID. Args: @@ -572,6 +577,7 @@ def _end_span( span. end_timestamp: Optional durable end timestamp to use as the span end time. When omitted, OpenTelemetry uses the current time. + error: Diagnostic details recorded at the normalized span end. """ logger.debug("Ending OTel span: operation_id=%s", operation_id) with self._operation_spans_lock: @@ -579,6 +585,7 @@ def _end_span( if span: # the span is not going to be populated if it has the same end_time and start_time end_time = self._next_ordered_timestamp(end_timestamp) + record_error_object_exception(span, error, timestamp=end_time) span.end(end_time=end_time) logger.debug("Ended OTel span: %s", span) @@ -965,11 +972,16 @@ def on_operation_end(self, info: OperationEndInfo) -> None: if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - record_error_object_exception(span, info.error) + elif info.status in { + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.STOPPED, + }: + span.set_status(StatusCode.ERROR) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) - self._end_span(info.operation_id, info.end_time) + self._end_span(info.operation_id, info.end_time, error=info.error) def on_user_function_start(self, info: UserFunctionStartInfo) -> None: """Called when a context or step operation starts user code. @@ -1088,14 +1100,19 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - record_error_object_exception(span, info.error) else: span.set_status(StatusCode.OK) end_timestamp = info.end_time if end_timestamp is not None and end_timestamp == info.start_time: end_timestamp += datetime.timedelta(microseconds=1) - self._end_span(span_key, end_timestamp) + self._end_span( + span_key, + end_timestamp, + error=info.error + if info.outcome is UserFunctionOutcome.FAILED + else None, + ) # Restore the enclosing context by releasing the scope this user # function attached. Code that runs after this operation (e.g. between # steps in a child context) correlates to its enclosing operation diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py index c2d14256b..79dd2eda4 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py @@ -95,7 +95,7 @@ def body(_event: Any, context: DurableContext) -> str: poll_interval=0.01, ) as runner: runner.register_function("failing-target", failing_function) - result = runner.run(input="{}", timeout=15) + result = runner.run(input="{}", execution_timeout=15) assert result.status.value == "SUCCEEDED" assert result.result is not None and json.loads(result.result) == "saved" assert saved_calls == ["saved"] @@ -125,7 +125,8 @@ def body(_event: Any, context: DurableContext) -> str: "actual invoke failure", } invocations = [span for span in spans if span.name == "Invocation"] - assert len(invocations) == (2 if resume else 1) + # An invoke can add its own suspension before the explicit replay wait. + assert len(invocations) >= (2 if resume else 1) assert all(span.status.status_code is StatusCode.OK for span in invocations) (workflow,) = [span for span in spans if span.name == "Workflow"] assert workflow.status.status_code is StatusCode.OK diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py index 57cdc3c9d..eb37d24f8 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py @@ -153,7 +153,9 @@ def test_terminal_exception_uses_normalized_span_end( _start_invocation(plugin) plugin.on_operation_end( replace( - _operation_end_info(ErrorObject("stored failure", "ValueError", None, None)), + _operation_end_info( + ErrorObject("stored failure", "ValueError", None, None) + ), operation_type=OperationType.CONTEXT, sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, start_time=start_time, @@ -201,7 +203,9 @@ def test_failed_attempt_exception_uses_normalized_span_end( ) ) _end_invocation(plugin) - span = next(s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1") + span = next( + s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1" + ) assert span.status.status_code is StatusCode.ERROR if with_error: (event,) = span.events