From f5a025ea87477ab4c8dbc6b36a4fb9b865e7aee8 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Mon, 27 Jul 2026 15:43:29 -0700 Subject: [PATCH 1/5] fix(otel): avoid duplicate spans for terminal replays --- .../execution_plugin.py | 15 ++++++ .../invocation_plugin.py | 24 ++++++++- .../tests/test_execution_plugin.py | 51 ++++++++++++++++++- .../tests/test_invocation_plugin.py | 32 +++++++----- 4 files changed, 104 insertions(+), 18 deletions(-) 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 b326ec77..5a651b8b 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 @@ -27,6 +27,7 @@ from aws_durable_execution_sdk_python.lambda_service import ( InvocationStatus, + OperationStatus, OperationType, ) from aws_durable_execution_sdk_python.plugin import ( @@ -76,6 +77,16 @@ {InvocationStatus.SUCCEEDED, InvocationStatus.FAILED} ) +_TERMINAL_OPERATION_STATUSES = frozenset( + { + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + } +) + # Registry key for the invocation span (operations use their operation_id). _INVOCATION_KEY = "__invocation__" @@ -346,6 +357,8 @@ def _reset_state(self) -> None: # ------------------------------------------------------------------ def on_operation_start(self, info: OperationStartInfo) -> None: logger.debug("Durable operation started: %s", info) + if info.is_replayed and info.status in _TERMINAL_OPERATION_STATUSES: + return if info.operation_type is OperationType.CONTEXT: return # tracked via on_user_function_start parent = self._resolve_parent(info.parent_id) @@ -359,6 +372,8 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: logger.debug("Durable operation ended: %s", info) + if info.is_replayed: + return if info.operation_type is OperationType.CONTEXT: return span = self._get_span(info.operation_id) 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 675bc027..b3f1fec6 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 @@ -9,6 +9,7 @@ from aws_durable_execution_sdk_python.lambda_service import ( InvocationStatus, + OperationStatus, OperationType, ) from aws_durable_execution_sdk_python.plugin import ( @@ -49,6 +50,16 @@ _SpanAttributes = dict[str, str | bool | int] +_TERMINAL_OPERATION_STATUSES = frozenset( + { + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + } +) + def _to_otel_timestamp(dt: datetime.datetime | None) -> int | None: """Convert a datetime to OTel timestamp (nanoseconds since epoch), or None.""" @@ -68,8 +79,9 @@ class InvocationOtelPlugin(DurableInstrumentationPlugin): Operation IDs are converted into deterministic span IDs. The first observed span for an operation uses that deterministic ID; later continuation spans use newly generated span IDs and link back to the deterministic span ID so - trace viewers can relate retries and replay-created terminal spans to the - original logical operation. + trace viewers can relate retries and cross-invocation completions to the + original logical operation. Terminal operations encountered only as replay + history do not emit another span. Args: trace_provider: OpenTelemetry tracer provider used to create spans. @@ -373,6 +385,8 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: def on_operation_start(self, info: OperationStartInfo) -> None: """Called when an operation begins. Creates a span for the operation.""" logger.debug("Durable operation started: %s", info) + if info.is_replayed and info.status in _TERMINAL_OPERATION_STATUSES: + return if info.operation_type is OperationType.CONTEXT: # Context operations are tracked using on_user_function_start. return @@ -396,8 +410,14 @@ def on_operation_end(self, info: OperationEndInfo) -> None: invocation is completing an operation that began earlier, so a short continuation span is created and linked to the deterministic logical operation span before being ended. + + Terminal replay callbacks describe history that completed before this + invocation and do not produce another span. Operations completed by the + backend while suspended are delivered with ``is_replayed=False``. """ logger.debug("Durable operation ended: %s", info) + if info.is_replayed: + return if info.operation_type is OperationType.CONTEXT: # Context operations are tracked using on_user_function_end. return diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 7cd544eb..9f46eceb 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -209,7 +209,7 @@ def test_cross_invocation_operation_end_uses_deterministic_span_id(): plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) - # Operation end with no matching start (started in a prior invocation). + # Backend-updated completion for an operation started in a prior invocation. plugin.on_operation_end( OperationEndInfo( operation_id="step-earlier", @@ -218,7 +218,7 @@ def test_cross_invocation_operation_end_uses_deterministic_span_id(): name="earlier-step", parent_id=None, start_time=START_TIME, - is_replayed=True, + is_replayed=False, status=OperationStatus.SUCCEEDED, end_time=END_TIME, error=None, @@ -235,6 +235,53 @@ def test_cross_invocation_operation_end_uses_deterministic_span_id(): ) +@pytest.mark.parametrize( + "operation_status", + [ + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + ], +) +def test_terminal_replayed_operation_does_not_emit_duplicate_span( + operation_status: OperationStatus, +): + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + + plugin.on_operation_start( + OperationStartInfo( + operation_id="step-earlier", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="earlier-step", + parent_id=None, + start_time=START_TIME, + is_replayed=True, + status=operation_status, + ) + ) + plugin.on_operation_end( + OperationEndInfo( + operation_id="step-earlier", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="earlier-step", + parent_id=None, + start_time=START_TIME, + is_replayed=True, + status=operation_status, + end_time=END_TIME, + error=None, + ) + ) + plugin.on_invocation_end(_invocation_end_info()) + + assert "earlier-step" not in {span.name for span in exporter.get_finished_spans()} + + # --------------------------------------------------------------------------- # Default-provider mode: invocation span # --------------------------------------------------------------------------- diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 8a243503..b1d24eff 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -326,15 +326,23 @@ def test_continuation_span_uses_recorded_start_and_end_times(): assert span.end_time >= span.start_time -def test_replayed_operation_start_emits_continuation_span_with_link(): - """Replayed operation spans should not reuse the original deterministic span ID.""" +@pytest.mark.parametrize( + "operation_status", + [ + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + ], +) +def test_terminal_replayed_operation_does_not_emit_duplicate_span( + operation_status: OperationStatus, +): + """Terminal operations completed before this invocation are not re-emitted.""" plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) operation_id = "wait-replayed" - random_span_id = int("abcdef1234567890", 16) - plugin._id_generator._fallback_id_generator.generate_span_id = lambda: ( - random_span_id - ) plugin.on_operation_start( OperationStartInfo( @@ -345,7 +353,7 @@ def test_replayed_operation_start_emits_continuation_span_with_link(): parent_id=None, start_time=START_TIME, is_replayed=True, - status=OperationStatus.SUCCEEDED, + status=operation_status, ) ) plugin.on_operation_end( @@ -357,18 +365,14 @@ def test_replayed_operation_start_emits_continuation_span_with_link(): parent_id=None, start_time=START_TIME, is_replayed=True, - status=OperationStatus.SUCCEEDED, + status=operation_status, end_time=END_TIME, error=None, ) ) + plugin.on_invocation_end(_invocation_end_info()) - span = exporter.get_finished_spans()[0] - assert span.name == "replayed-wait" - assert span.context.span_id == random_span_id - assert span.links[0].context.span_id == operation_id_to_span_id( - EXECUTION_ARN, operation_id - ) + assert [span.name for span in exporter.get_finished_spans()] == ["invocation"] def test_retried_operation_start_emits_continuation_span_with_link(): From d038e492392b0586e1699e643e928f4d6bedcc9c Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Mon, 27 Jul 2026 16:46:20 -0700 Subject: [PATCH 2/5] fix(otel): start operation spans at current time --- .../aws_durable_execution_sdk_python_otel/invocation_plugin.py | 2 +- .../tests/test_invocation_plugin.py | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) 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 b3f1fec6..8783a77e 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 @@ -397,7 +397,7 @@ def on_operation_start(self, info: OperationStartInfo) -> None: operation_id=info.operation_id, name=info.name or info.operation_id, attributes=attributes, - start_time=info.start_time, + start_time=datetime.datetime.now(datetime.UTC), parent_span=parent_span, existed=info.is_replayed, ) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index b1d24eff..658dc9dd 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -218,6 +218,9 @@ def test_operation_callbacks_emit_child_span_with_deterministic_span_id(): ) active_wait_span = plugin._get_span(operation_id) assert active_wait_span is not None + invocation_span = plugin._get_span(None) + assert invocation_span is not None + assert invocation_span.start_time <= active_wait_span.start_time assert ( active_wait_span.attributes["durable.operation.status"] == OperationStatus.STARTED.value From dba1fb4bea410bf7d4ecc5c55d4eb490e9f080ac Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 28 Jul 2026 00:14:34 +0000 Subject: [PATCH 3/5] fix(otel): normalize continuation span timestamps --- .../invocation_plugin.py | 7 ++++--- .../tests/test_invocation_plugin.py | 17 +++++++++-------- 2 files changed, 13 insertions(+), 11 deletions(-) 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 8783a77e..ca0db5b5 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 @@ -422,7 +422,8 @@ def on_operation_end(self, info: OperationEndInfo) -> None: # Context operations are tracked using on_user_function_end. return span = self._get_span(info.operation_id) - if not span: + is_continuation = span is None + if span is None: # the span was not started in the current invocation, so we need to # create a new one that links to the previous one parent_span = self._resolve_parent_span(info.parent_id) @@ -431,7 +432,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: operation_id=info.operation_id, name=info.name or info.operation_id, attributes=attributes, - start_time=info.start_time, + start_time=datetime.datetime.now(datetime.UTC), parent_span=parent_span, existed=True, ) @@ -446,7 +447,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: else: span.set_status(StatusCode.OK) - end_timestamp = info.end_time + end_timestamp = None if is_continuation else info.end_time if end_timestamp is not None and end_timestamp == info.start_time: end_timestamp += datetime.timedelta(microseconds=1) self._end_span(info.operation_id, end_timestamp) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 658dc9dd..58d36166 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime @@ -300,10 +301,13 @@ def test_operation_end_without_start_emits_continuation_span_with_link(): ) -def test_continuation_span_uses_recorded_start_and_end_times(): - """Continuation spans use the recorded operation start/end times.""" +def test_continuation_span_uses_current_start_and_end_times(): + """Continuation spans use current times within the invocation.""" plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) + invocation_span = plugin._get_span(None) + assert invocation_span is not None + before_callback = time.time_ns() plugin.on_operation_end( OperationEndInfo( @@ -319,14 +323,11 @@ def test_continuation_span_uses_recorded_start_and_end_times(): error=None, ) ) + after_callback = time.time_ns() span = exporter.get_finished_spans()[0] - expected_start = int(START_TIME.timestamp() * 1_000_000_000) - expected_end = int(END_TIME.timestamp() * 1_000_000_000) - assert span.start_time == expected_start - assert span.end_time == expected_end - # Duration must be non-negative. - assert span.end_time >= span.start_time + assert invocation_span.start_time <= span.start_time + assert before_callback <= span.start_time <= span.end_time <= after_callback @pytest.mark.parametrize( From 9835af36d66571473ef359071082844681e121ec Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 28 Jul 2026 01:12:12 +0000 Subject: [PATCH 4/5] fix(otel): use current operation end times --- .../invocation_plugin.py | 6 +----- .../tests/test_invocation_plugin.py | 6 ++++++ 2 files changed, 7 insertions(+), 5 deletions(-) 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 ca0db5b5..81421464 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 @@ -422,7 +422,6 @@ def on_operation_end(self, info: OperationEndInfo) -> None: # Context operations are tracked using on_user_function_end. return span = self._get_span(info.operation_id) - is_continuation = span is None if span is None: # the span was not started in the current invocation, so we need to # create a new one that links to the previous one @@ -447,10 +446,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: else: span.set_status(StatusCode.OK) - end_timestamp = None if is_continuation else info.end_time - if end_timestamp is not None and end_timestamp == info.start_time: - end_timestamp += datetime.timedelta(microseconds=1) - self._end_span(info.operation_id, end_timestamp) + self._end_span(info.operation_id) def on_user_function_start(self, info: UserFunctionStartInfo) -> None: """Called when a context or step operation starts user code. diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 58d36166..20215f4e 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -254,6 +254,12 @@ def test_operation_callbacks_emit_child_span_with_deterministic_span_id(): EXECUTION_ARN, operation_id ) assert wait_span.parent.span_id == invocation_span.context.span_id + assert ( + invocation_span.start_time + <= wait_span.start_time + <= wait_span.end_time + <= invocation_span.end_time + ) assert wait_span.attributes["durable.operation.id"] == operation_id assert wait_span.attributes["durable.operation.type"] == OperationType.WAIT.value assert ( From a36e2ed5c5085c6c526c21150b53f24f2d78cb01 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 28 Jul 2026 11:02:51 -0700 Subject: [PATCH 5/5] fix(plugin): skip terminal operation replay callbacks --- .../execution_plugin.py | 15 ------ .../invocation_plugin.py | 22 +-------- .../tests/test_execution_plugin.py | 47 ------------------ .../tests/test_invocation_plugin.py | 49 ------------------- .../plugin.py | 39 ++++----------- .../tests/plugin_test.py | 42 ++++++++++++++++ .../tests/state_test.py | 23 ++++++--- 7 files changed, 69 insertions(+), 168 deletions(-) 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 5a651b8b..b326ec77 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 @@ -27,7 +27,6 @@ from aws_durable_execution_sdk_python.lambda_service import ( InvocationStatus, - OperationStatus, OperationType, ) from aws_durable_execution_sdk_python.plugin import ( @@ -77,16 +76,6 @@ {InvocationStatus.SUCCEEDED, InvocationStatus.FAILED} ) -_TERMINAL_OPERATION_STATUSES = frozenset( - { - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.TIMED_OUT, - OperationStatus.CANCELLED, - OperationStatus.STOPPED, - } -) - # Registry key for the invocation span (operations use their operation_id). _INVOCATION_KEY = "__invocation__" @@ -357,8 +346,6 @@ def _reset_state(self) -> None: # ------------------------------------------------------------------ def on_operation_start(self, info: OperationStartInfo) -> None: logger.debug("Durable operation started: %s", info) - if info.is_replayed and info.status in _TERMINAL_OPERATION_STATUSES: - return if info.operation_type is OperationType.CONTEXT: return # tracked via on_user_function_start parent = self._resolve_parent(info.parent_id) @@ -372,8 +359,6 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: logger.debug("Durable operation ended: %s", info) - if info.is_replayed: - return if info.operation_type is OperationType.CONTEXT: return span = self._get_span(info.operation_id) 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 81421464..7110a21d 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 @@ -9,7 +9,6 @@ from aws_durable_execution_sdk_python.lambda_service import ( InvocationStatus, - OperationStatus, OperationType, ) from aws_durable_execution_sdk_python.plugin import ( @@ -50,16 +49,6 @@ _SpanAttributes = dict[str, str | bool | int] -_TERMINAL_OPERATION_STATUSES = frozenset( - { - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.TIMED_OUT, - OperationStatus.CANCELLED, - OperationStatus.STOPPED, - } -) - def _to_otel_timestamp(dt: datetime.datetime | None) -> int | None: """Convert a datetime to OTel timestamp (nanoseconds since epoch), or None.""" @@ -80,8 +69,7 @@ class InvocationOtelPlugin(DurableInstrumentationPlugin): span for an operation uses that deterministic ID; later continuation spans use newly generated span IDs and link back to the deterministic span ID so trace viewers can relate retries and cross-invocation completions to the - original logical operation. Terminal operations encountered only as replay - history do not emit another span. + original logical operation. Args: trace_provider: OpenTelemetry tracer provider used to create spans. @@ -385,8 +373,6 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: def on_operation_start(self, info: OperationStartInfo) -> None: """Called when an operation begins. Creates a span for the operation.""" logger.debug("Durable operation started: %s", info) - if info.is_replayed and info.status in _TERMINAL_OPERATION_STATUSES: - return if info.operation_type is OperationType.CONTEXT: # Context operations are tracked using on_user_function_start. return @@ -410,14 +396,8 @@ def on_operation_end(self, info: OperationEndInfo) -> None: invocation is completing an operation that began earlier, so a short continuation span is created and linked to the deterministic logical operation span before being ended. - - Terminal replay callbacks describe history that completed before this - invocation and do not produce another span. Operations completed by the - backend while suspended are delivered with ``is_replayed=False``. """ logger.debug("Durable operation ended: %s", info) - if info.is_replayed: - return if info.operation_type is OperationType.CONTEXT: # Context operations are tracked using on_user_function_end. return diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 9f46eceb..68753472 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -235,53 +235,6 @@ def test_cross_invocation_operation_end_uses_deterministic_span_id(): ) -@pytest.mark.parametrize( - "operation_status", - [ - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.TIMED_OUT, - OperationStatus.CANCELLED, - OperationStatus.STOPPED, - ], -) -def test_terminal_replayed_operation_does_not_emit_duplicate_span( - operation_status: OperationStatus, -): - plugin, exporter = _create_plugin() - plugin.on_invocation_start(_invocation_start_info()) - - plugin.on_operation_start( - OperationStartInfo( - operation_id="step-earlier", - operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, - name="earlier-step", - parent_id=None, - start_time=START_TIME, - is_replayed=True, - status=operation_status, - ) - ) - plugin.on_operation_end( - OperationEndInfo( - operation_id="step-earlier", - operation_type=OperationType.STEP, - sub_type=OperationSubType.STEP, - name="earlier-step", - parent_id=None, - start_time=START_TIME, - is_replayed=True, - status=operation_status, - end_time=END_TIME, - error=None, - ) - ) - plugin.on_invocation_end(_invocation_end_info()) - - assert "earlier-step" not in {span.name for span in exporter.get_finished_spans()} - - # --------------------------------------------------------------------------- # Default-provider mode: invocation span # --------------------------------------------------------------------------- diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 20215f4e..3711d32d 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -336,55 +336,6 @@ def test_continuation_span_uses_current_start_and_end_times(): assert before_callback <= span.start_time <= span.end_time <= after_callback -@pytest.mark.parametrize( - "operation_status", - [ - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.TIMED_OUT, - OperationStatus.CANCELLED, - OperationStatus.STOPPED, - ], -) -def test_terminal_replayed_operation_does_not_emit_duplicate_span( - operation_status: OperationStatus, -): - """Terminal operations completed before this invocation are not re-emitted.""" - plugin, exporter = _create_plugin() - plugin.on_invocation_start(_invocation_start_info()) - operation_id = "wait-replayed" - - plugin.on_operation_start( - OperationStartInfo( - operation_id=operation_id, - operation_type=OperationType.WAIT, - sub_type=OperationSubType.WAIT, - name="replayed-wait", - parent_id=None, - start_time=START_TIME, - is_replayed=True, - status=operation_status, - ) - ) - plugin.on_operation_end( - OperationEndInfo( - operation_id=operation_id, - operation_type=OperationType.WAIT, - sub_type=OperationSubType.WAIT, - name="replayed-wait", - parent_id=None, - start_time=START_TIME, - is_replayed=True, - status=operation_status, - end_time=END_TIME, - error=None, - ) - ) - plugin.on_invocation_end(_invocation_end_info()) - - assert [span.name for span in exporter.get_finished_spans()] == ["invocation"] - - def test_retried_operation_start_emits_continuation_span_with_link(): """Retried operation spans should not reuse the original deterministic span ID.""" plugin, exporter = _create_plugin() diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py index 5294f1c7..c2d24750 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py @@ -213,9 +213,9 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: def on_operation_start(self, info: OperationStartInfo) -> None: """ Called before an operation's START checkpoint is queued, or when a - prior operation is replayed. This guarantees that it strictly precedes - ``on_user_function_start``. This is called NOT within the thread that - runs operation. + prior non-terminal operation is replayed. This guarantees that it + strictly precedes ``on_user_function_start``. This is called NOT within + the thread that runs operation. Args: info: Information about the operation. @@ -225,8 +225,9 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: """ - Called when an operation checkpoints a terminal status, or when a prior - terminal operation is replayed. This is called NOT within the thread that runs operation. + Called when an operation checkpoints a terminal status. Terminal + operations are not emitted again during replay. This is called NOT + within the thread that runs operation. Args: info: Information about the operation. @@ -407,7 +408,10 @@ def on_operation_action( ) def on_operation_replay(self, operation: Operation) -> None: - """Execute plugins for a checkpointed operation observed during replay.""" + """Execute plugins for a non-terminal operation observed during replay.""" + if self._is_terminal_status(operation.status): + return + start_info = OperationStartInfo( operation_id=operation.operation_id, operation_type=operation.operation_type, @@ -420,29 +424,6 @@ def on_operation_replay(self, operation: Operation) -> None: ) self.execute_plugins(start_info, sync=True) - if self._is_terminal_status(operation.status): - self.execute_plugins( - OperationEndInfo( - operation_id=operation.operation_id, - operation_type=operation.operation_type, - sub_type=operation.sub_type, - name=operation.name, - parent_id=operation.parent_id, - start_time=operation.start_timestamp, - end_time=operation.end_timestamp, - result=_extract_result(operation), - status=operation.status, - error=self._extract_error(operation), - attempt=( - operation.step_details.attempt - if operation.step_details - else None - ), - is_replayed=True, - ), - sync=True, - ) - def on_operation_update( self, operation_or_operations: Operation | Sequence[Operation] | None, diff --git a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py index b1b9a6a1..0dff5f5b 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_test.py @@ -642,6 +642,48 @@ def test_fail_action_does_not_fire(self): self.assertEqual(self.plugin.calls, []) +class TestPluginExecutorOnOperationReplay(unittest.TestCase): + """Tests for PluginExecutor.on_operation_replay.""" + + def test_terminal_operation_does_not_fire_callbacks(self): + terminal_statuses = [ + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + ] + + for status in terminal_statuses: + with self.subTest(status=status): + plugin = _TrackingPlugin() + executor = PluginExecutor(plugins=[plugin]) + operation = Operation( + operation_id="op-1", + operation_type=OperationType.STEP, + status=status, + ) + + with executor.run(): + executor.on_operation_replay(operation) + + self.assertEqual(plugin.calls, []) + + def test_non_terminal_operation_fires_operation_start(self): + plugin = _TrackingPlugin() + executor = PluginExecutor(plugins=[plugin]) + operation = Operation( + operation_id="op-1", + operation_type=OperationType.WAIT, + status=OperationStatus.STARTED, + ) + + with executor.run(): + executor.on_operation_replay(operation) + + self.assertEqual(plugin.calls, ["operation_start:op-1"]) + + class TestPluginExecutorOnOperationUpdate(unittest.TestCase): """Tests for PluginExecutor.on_operation_update.""" diff --git a/packages/aws-durable-execution-sdk-python/tests/state_test.py b/packages/aws-durable-execution-sdk-python/tests/state_test.py index f0a576e4..94d6d56f 100644 --- a/packages/aws-durable-execution-sdk-python/tests/state_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/state_test.py @@ -4921,14 +4921,26 @@ def test_plugin_executor_not_called_for_pending_operations(): assert len(operation_end_calls) == 0 -def test_emit_operation_replay_hook_fires_start_and_end_for_terminal_operation(): - """emit_operation_replay_hook emits start+end (is_replayed=True) for terminal ops.""" +@pytest.mark.parametrize( + "terminal_status", + [ + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + ], +) +def test_emit_operation_replay_hook_skips_terminal_operation( + terminal_status: OperationStatus, +): + """Terminal operations do not emit plugin callbacks during replay.""" start_time = datetime.datetime(2025, 1, 1, tzinfo=datetime.UTC) end_time = datetime.datetime(2025, 1, 2, tzinfo=datetime.UTC) operation = Operation( operation_id="step-1", operation_type=OperationType.STEP, - status=OperationStatus.SUCCEEDED, + status=terminal_status, parent_id="parent-1", name="my-step", start_timestamp=start_time, @@ -4959,10 +4971,7 @@ def on_operation_end(self, info): state.emit_operation_replay_hook(operation) state.emit_operation_replay_hook(operation) - assert captured == [ - ("start", "step-1", True, OperationStatus.SUCCEEDED), - ("end", "step-1", True, OperationStatus.SUCCEEDED), - ] + assert captured == [] def test_emit_operation_replay_hook_fires_only_start_for_non_terminal_operation():