From 522d3fce634edabffaae034d8111024253567507 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:01:36 -0700 Subject: [PATCH 1/2] test: reproduce missing OTel replay barrier --- .../test_chained_invoke_replay_barrier.py | 199 ++++++++++++++++++ .../tests/test_otel_examples.py | 29 +++ 2 files changed, 228 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_replay_barrier.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_replay_barrier.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_replay_barrier.py new file mode 100644 index 000000000..0430a2990 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_replay_barrier.py @@ -0,0 +1,199 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""The real case11 handler starts one child, then replays its saved result.""" + +import importlib.util +import json +import logging +from pathlib import Path +import sys +from threading import Event +import time +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.types import LambdaContext +from aws_durable_execution_sdk_python_testing.model import ( + GetDurableExecutionHistoryResponse, +) +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 aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +SRC = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_case11_uses_one_real_child_and_replays_after_barrier( + view: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) + monkeypatch.setenv("OTEL_INVOKE_TARGET_FUNCTION_NAME", "case11-child") + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + common_spec = importlib.util.spec_from_file_location("common", SRC / "common.py") + assert common_spec is not None and common_spec.loader is not None + common = importlib.util.module_from_spec(common_spec) + monkeypatch.setitem(sys.modules, "common", common) + common_spec.loader.exec_module(common) + # Configure the real bundled plugin/exporter; no propagation hook, model, + # header, operation status or checkpoint response is injected. + monkeypatch.setattr( + common, + "otel_plugin_factory", + lambda: view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ), + ) + spec = importlib.util.spec_from_file_location( + "otel_11_chained_invoke", SRC / "otel_11_chained_invoke.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + target_entered, admit_target = Event(), Event() + target_calls: list[str] = [] + payload = {"scenario": "chained-invoke", "value": {"preserved": [1, "same-result"]}} + + def target_entrypoint( + event: dict[str, Any], context: LambdaContext + ) -> dict[str, Any]: + # A harness admission gate makes the first caller suspension observable. + # The unchanged production echo target still executes exactly once. + target_calls.append(context.aws_request_id) + target_entered.set() + assert admit_target.wait(10) + return module.target_handler(event, context) + + def wait_for_invocations( + runner: DurableFunctionTestRunner, arn: str, count: int + ) -> GetDurableExecutionHistoryResponse: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + history = runner.get_execution_history(arn, include_execution_data=True) + if ( + sum(e.event_type == "InvocationCompleted" for e in history.events) + >= count + ): + return history + time.sleep(0.01) + raise AssertionError("Caller did not reach the expected suspension") + + try: + with DurableFunctionTestRunner( + handler=module.handler, + skip_time=False, + poll_interval=0.01, + execution_timeout=30, + ) as runner: + runner.register_durable_function("case11-child", target_entrypoint) + arn = runner.run_async(input=json.dumps(payload)) + assert target_entered.wait(10) + first = wait_for_invocations(runner, arn, 1) + assert not any( + e.event_type == "ChainedInvokeSucceeded" for e in first.events + ) + admit_target.set() + callback_id = runner.wait_for_callback( + arn, name="otel-invoke-replay-barrier create callback id", timeout=10 + ) + middle = wait_for_invocations(runner, arn, 2) + assert any(e.event_type == "ChainedInvokeSucceeded" for e in middle.events) + assert not any(e.event_type == "CallbackSucceeded" for e in middle.events) + runner.send_callback_success(callback_id, result=b'"resume"') + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history(arn, include_execution_data=True) + started = [ + e for e in history.events if e.event_type == "ChainedInvokeStarted" + ] + succeeded = [ + e for e in history.events if e.event_type == "ChainedInvokeSucceeded" + ] + assert len(started) == len(succeeded) == 1 + assert started[0].operation_id == succeeded[0].operation_id + assert started[0].name == "otel-invoke" + child_arn = started[0].chained_invoke_started_details.durable_execution_arn + assert child_arn and child_arn != arn + child_history = runner.get_execution_history( + child_arn, include_execution_data=True + ) + + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == payload + assert len(target_calls) == 1 + assert sum(e.event_type == "InvocationCompleted" for e in history.events) == 3 + assert ( + sum(e.event_type == "InvocationCompleted" for e in child_history.events) + == 1 + ) + assert ( + sum(e.event_type == "ExecutionStarted" for e in child_history.events) == 1 + ) + assert ( + sum(e.event_type == "ExecutionSucceeded" for e in child_history.events) == 1 + ) + assert ( + sum( + e.event_type == "StepSucceeded" + and e.name == "otel-invoke-replay-barrier submitter" + for e in history.events + ) + == 1 + ) + spans = exporter.get_finished_spans() + caller = sorted( + [ + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == arn + ], + key=lambda s: s.start_time, + ) + child = [ + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == child_arn + ] + assert [s.attributes["durable.invocation.status"] for s in caller] == [ + "PENDING", + "PENDING", + "SUCCEEDED", + ] + assert [s.attributes["durable.invocation.first"] for s in caller] == [ + True, + False, + False, + ] + assert ( + len(child) == 1 + and child[0].attributes["durable.invocation.status"] == "SUCCEEDED" + ) + assert not [ + r + for r in caplog.records + if r.levelno >= logging.ERROR and ("plugin" in r.name or "otel" in r.name) + ] + finally: + admit_target.set() + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py index 2d086d7c0..3ccd0d72d 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_otel_examples.py @@ -259,6 +259,35 @@ def test_templates_map_every_otel_requirement( assert requirement_mappings(load_template(template_path)) == expected_mappings +@pytest.mark.parametrize( + ("logical_id", "suite"), + [("Otel1Success", "otel-invocation"), ("OtelExecution1Success", "otel-execution")], +) +def test_deferred_requirements_are_explicit(logical_id: str, suite: str) -> None: + template = load_template(TEMPLATE_PATH) + gaps = { + entry["id"]: entry["reason"] + for entry in template["Resources"][logical_id]["TestingMetadata"][ + "NotImplemented" + ] + } + assert set(gaps) == {f"{suite}-{case}" for case in range(21, 28)} + for case in (21, 22, 23, 24, 27): + assert "pull/796" in gaps[f"{suite}-{case}"] + assert "issues/853" in gaps[f"{suite}-25"] + assert "issues/862" in gaps[f"{suite}-25"] + assert "issues/854" in gaps[f"{suite}-26"] + assert "issues/860" in gaps[f"{suite}-26"] + assert "issues/861" in gaps[f"{suite}-26"] + covered = { + requirement + for _, requirement in requirement_mappings(template) + if requirement.startswith(f"{suite}-") + } + assert covered.isdisjoint(gaps) + assert covered | gaps.keys() == {f"{suite}-{case}" for case in range(1, 28)} + + @pytest.mark.parametrize( ("template_path", "expected_mappings", "expected_targets"), [ From a8baa73c32904cfd6a1c03155899a3753c954249 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:03:06 -0700 Subject: [PATCH 2/2] fix(otel): restore case 11 replay coverage --- .../src/otel_11_chained_invoke.py | 16 ++++- .../template.yaml | 64 +++++++++++++++++++ 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_11_chained_invoke.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_11_chained_invoke.py index 777d222cc..27706f244 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_11_chained_invoke.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_11_chained_invoke.py @@ -9,20 +9,34 @@ from typing import Any from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import WaitForCallbackConfig +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.types import WaitForCallbackContext from common import otel_plugin_factory, require_scenario +def _submit(_callback_id: str, _context: WaitForCallbackContext) -> None: + """The conformance driver completes this callback after suspension.""" + return None + + @durable_execution(plugins=[otel_plugin_factory()]) def handler( event: dict[str, Any], context: DurableContext, ) -> dict[str, Any]: require_scenario(event, "chained-invoke") - return context.invoke( + result: dict[str, Any] = context.invoke( function_name=os.environ["OTEL_INVOKE_TARGET_FUNCTION_NAME"], payload=event, name="otel-invoke", ) + context.wait_for_callback( + _submit, + name="otel-invoke-replay-barrier", + config=WaitForCallbackConfig(serdes=JsonSerDes()), + ) + return result @durable_execution(plugins=[otel_plugin_factory()]) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml index be70e122d..a99f19a73 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/template.yaml @@ -106,6 +106,38 @@ Resources: TestingMetadata: TestDescription: - otel-invocation-1 + NotImplemented: + - id: otel-invocation-21 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-invocation-22 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-invocation-23 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-invocation-24 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-invocation-25 + reason: >- + Deferred until terminal failure status and empty callback error history + are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/853 + and https://github.com/aws/aws-durable-execution-sdk-python/issues/862. + - id: otel-invocation-26 + reason: >- + Deferred until callback completion delivery and replay acknowledgment + are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/854, + https://github.com/aws/aws-durable-execution-sdk-python/issues/860, + and https://github.com/aws/aws-durable-execution-sdk-python/issues/861. + - id: otel-invocation-27 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. Properties: CodeUri: src/ Handler: otel_1_success.handler @@ -417,6 +449,38 @@ Resources: TestingMetadata: TestDescription: - otel-execution-1 + NotImplemented: + - id: otel-execution-21 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-execution-22 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-execution-23 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-execution-24 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. + - id: otel-execution-25 + reason: >- + Deferred until terminal failure status and empty callback error history + are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/853 + and https://github.com/aws/aws-durable-execution-sdk-python/issues/862. + - id: otel-execution-26 + reason: >- + Deferred until callback completion delivery and replay acknowledgment + are implemented: https://github.com/aws/aws-durable-execution-sdk-python/issues/854, + https://github.com/aws/aws-durable-execution-sdk-python/issues/860, + and https://github.com/aws/aws-durable-execution-sdk-python/issues/861. + - id: otel-execution-27 + reason: >- + Conformance handler coverage is pending in + https://github.com/aws/aws-durable-execution-sdk-python/pull/796. Properties: CodeUri: src/ Handler: otel_1_success.handler