diff --git a/.github/scripts/tests/test_opentelemetry_conformance_workflow.py b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py index 034be671e..37841a888 100644 --- a/.github/scripts/tests/test_opentelemetry_conformance_workflow.py +++ b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py @@ -1,8 +1,12 @@ +import re from pathlib import Path import yaml +SHARED_WORKFLOW_REF = "main" +CONFORMANCE_TEST_REF = "main" + WORKFLOW_PATH = ( Path(__file__).parents[2] / "workflows" / "opentelemetry-conformance-tests.yml" ) @@ -18,7 +22,10 @@ def test_opentelemetry_conformance_caller_uses_current_workflow_contract() -> No "uses: aws/aws-durable-execution-conformance-tests/.github/workflows/" "opentelemetry-orchestrator.yml@" ) - assert orchestrator in workflow + shared_ref = re.search(re.escape(orchestrator) + r"(\S+)", workflow) + assert shared_ref is not None + assert shared_ref.group(1) == SHARED_WORKFLOW_REF + assert f"default: {CONFORMANCE_TEST_REF}" in workflow assert "python-opentelemetry.yml@" not in workflow assert "\n otlp_endpoint:" not in workflow @@ -28,7 +35,7 @@ def test_opentelemetry_conformance_caller_uses_current_workflow_contract() -> No "resource_prefix: p", "sdk_repository: aws/aws-durable-execution-sdk-python", "sdk_ref: ${{ github.event.pull_request.head.sha || github.sha }}", - "conformance_test_ref: ${{ inputs.conformance_test_ref || 'main' }}", + f"conformance_test_ref: ${{{{ inputs.conformance_test_ref || '{CONFORMANCE_TEST_REF}' }}}}", "checkout_sdk: true", f"examples_dir: {EXAMPLES_DIR}", "adot_release_repository: aws-observability/aws-otel-python-instrumentation", diff --git a/.github/workflows/opentelemetry-conformance-tests.yml b/.github/workflows/opentelemetry-conformance-tests.yml index aed97f6e8..fca1b1f40 100644 --- a/.github/workflows/opentelemetry-conformance-tests.yml +++ b/.github/workflows/opentelemetry-conformance-tests.yml @@ -66,7 +66,7 @@ jobs: actions: write contents: read id-token: write - uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@2387bb48968f7420ddf55aaf6dc83a9f15778332 + uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@main with: language: python runs_on: codebuild-github-actions-runner-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md index 9c23dd9b6..a5ac7ef4a 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/README.md @@ -12,7 +12,7 @@ language-agnostic OTel conformance runner in which invokes each function, collects its spans from the configured backend, and asserts they match the shared requirement specification. That repository owns the runner, the requirement specifications, and the orchestration (backend matrix, -ADOT layer resolution, collector build, long-running cycle); this package owns +ADOT layer resolution, collector build); this package owns the Python handlers and SAM templates. This mirrors the non-OTel @@ -33,7 +33,7 @@ template-long-running.yaml # otel-long-running suite tests/ # contract tests for the templates and handlers ``` -The 20 invocation and 20 execution requirements reuse the same scenario +The invocation and execution requirements reuse the same scenario handlers; the view is selected per function through the `OTEL_PLUGIN_MODE` environment variable, which `common.otel_plugin_factory()` reads to pick `InvocationOtelPluginFactory` or `ExecutionOtelPluginFactory`. `template.yaml` @@ -63,6 +63,11 @@ deploys only the view named by its `OtelSuite` parameter. | `otel-invocation-18` | `otel_18_chained_invoke_failure.handler` | Verifies failed chained-invoke telemetry. | | `otel-invocation-19` | `otel_19_execution_failure.handler` | Verifies telemetry for a direct handler failure. | | `otel-invocation-20` | `otel_20_virtual_context.handler` | Verifies a virtual child-context span without context checkpoints. | +| `otel-invocation-21` | `otel_21_completed_step_replay.handler` | Replays a completed step after a successful wait/resume; each step body runs once. | +| `otel-invocation-22` | `otel_22_user_function_context.handler` | Creates ordinary user spans under the active handler, attempt, child, branch, and iteration contexts. | +| `otel-invocation-23` | `otel_23_callback_function_context.handler` | Verifies retry/check attempts, callback submitter, wrapped retry helper, and virtual-child callback parents. | +| `otel-invocation-24` | `otel_24_invocation_retry_status.handler` | Raises a retryable invocation error after a completed step, then resumes with its saved result. | +| `otel-invocation-27` | `otel_27_chained_invoke_suspend.handler` | Verifies caller and child invocation spans when the child suspends for a durable wait and resumes. | | `otel-execution-1` | `otel_1_success.handler` | Verifies the execution-view workflow, step, and attempt hierarchy. | | `otel-execution-2` | `otel_2_wait_resume.handler` | Verifies the execution view across a resumed invocation. | | `otel-execution-3` | `otel_3_retry.handler` | Verifies the execution view across retry attempts. | @@ -83,6 +88,11 @@ deploys only the view named by its `OtelSuite` parameter. | `otel-execution-18` | `otel_18_chained_invoke_failure.handler` | Verifies source and target failed workflow roots. | | `otel-execution-19` | `otel_19_execution_failure.handler` | Verifies a failed invocation without a completed workflow. | | `otel-execution-20` | `otel_20_virtual_context.handler` | Verifies a virtual child-context span under the workflow root. | +| `otel-execution-21` | `otel_21_completed_step_replay.handler` | Verifies completed-operation spans are exported once across normal successful replay. | +| `otel-execution-22` | `otel_22_user_function_context.handler` | Observes active execution-view callback contexts without supplying or repairing parents. | +| `otel-execution-23` | `otel_23_callback_function_context.handler` | Verifies the same SDK-owned callback lifecycle parents in the execution view. | +| `otel-execution-24` | `otel_24_invocation_retry_status.handler` | Verifies invocation retry status independently of step retry and recovery re-exports. | +| `otel-execution-27` | `otel_27_chained_invoke_suspend.handler` | Verifies caller/child workflow parentage and continuation links across child suspension. | | `otel-long-running-1` | `otel_long_running_1_wait.handler` | Verifies wait and resume telemetry across a long durable suspension. | | `otel-long-running-2` | `otel_long_running_2_retry.handler` | Verifies retry telemetry across a long durable backoff. | | `otel-long-running-3` | `otel_long_running_3_callback.handler` | Verifies callback telemetry when completion arrives after a long delay. | @@ -91,6 +101,54 @@ deploys only the view named by its `OtelSuite` parameter. The runner discovers each mapping from `TestingMetadata.TestDescription` on the functions in the templates. +## Deferred callback coverage + +Cases 25 and 26 are declared `NOT_IMPLEMENTED` in both views. Their handlers, +SAM functions, build targets and local integration tests are deferred until +the required SDK and testing-library behavior lands. The runner reports these +as explicit gaps; the assertions for the active cases remain unchanged. + +- Case 25 needs terminal failure status without diagnostic details + ([#853](https://github.com/aws/aws-durable-execution-sdk-python/issues/853)) + and consistent empty callback error history + ([#862](https://github.com/aws/aws-durable-execution-sdk-python/issues/862)). +- Case 26 needs completion discovery and export + ([#854](https://github.com/aws/aws-durable-execution-sdk-python/issues/854)), + callback update production + ([#860](https://github.com/aws/aws-durable-execution-sdk-python/issues/860)), + and invocation-input acknowledgment + ([#861](https://github.com/aws/aws-durable-execution-sdk-python/issues/861)). +- The additional case 17 callback-error regression also needs snapshot isolation + ([#852](https://github.com/aws/aws-durable-execution-sdk-python/issues/852)). + The existing case 17 handler and supplied-error coverage remain active. + +The deferred source and assertions are preserved in +[commit `6484e577`](https://github.com/aws/aws-durable-execution-sdk-python/commit/6484e577b478aeb6345e346515b40bd516cacb1f) +for follow-up. Before restoring them, correct the history assertions to respect +`IncludeExecutionData`: detailed errors are untruncated, while metadata-only +history omits payloads and cannot reconstruct diagnostic details. The claimed +empty history `Payload` shape remains **unverified** against AWS Lambda and +requires service evidence. Case 26 must still assert that the terminal callback +has reached the exporter before the observed step, then check the raw export +count across both later resumes. + +Case 24 remains active and covers invocation `RETRY` becoming `RETRYING`/`UNSET`. + +## Callback coverage boundary + +Cases 22 and 23 create normal user spans from the active context, without an +explicit parent or a copied execution ARN. They cover handler context, step and +condition attempts, child and branch bodies, callback submission, wrapped +`with_retry` body/strategy callbacks, and a virtual child. Case 23 places the +retry helper and virtual child after callback completion so ordinary successful +replay does not repeat their probes. + +This does not promise an operation/attempt parent for every arbitrary callback. +General retry/wait policies, serializers, summary generators, and item naming +have phase-specific caller scopes outside the wrapped user-function lifecycle; +these cases do not impose a new ownership policy on them. Instrumentation +extensions and user-created threads are outside this business-callback contract. + ## How a handler maps to a requirement ```yaml @@ -197,6 +255,15 @@ write access; the runner identity needs list, read, and cleanup access. `.github/workflows/opentelemetry-conformance-tests.yml` calls the shared orchestrator in the conformance repository and points it at this package with `examples_dir`, so orchestration stays centralized while the handlers run from -the commit under test. Pull requests run the invocation and execution suites plus -a short (60-second) long-running cycle; the full multi-hour long-running cycle is -driven by `workflow_dispatch` with `phase: launch` and `phase: check`. +the commit under test. The shared workflow tracks the conformance repository's +`main` branch, and the test catalog defaults to `main`. Manual runs can override +the catalog with the `conformance_test_ref` input. Pull requests run the +invocation and execution suites, with cases 25 and 26 reported as +`NOT_IMPLEMENTED`. The orchestrator no longer runs the legacy long-running +suite; its templates and local contract tests remain in this package. + +Case 27 (`chained-invoke-suspend`) invokes a child that performs a one-second +durable wait and returns the original payload after resuming. Both views +check two caller invocations, two child invocations and the child wait's +parentage and continuation links. It replaces the suspended-child scenario +formerly exercised by the short mode of `otel-long-running-4`. diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile index 320577e7f..32d55bdee 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/Makefile @@ -1,22 +1,25 @@ .PHONY: build-Otel1Success build-Otel2WaitResume build-Otel3Retry build-Otel4TerminalFailure \ build-Otel5ChildContext build-Otel6Parallel build-Otel7Map build-Otel8HandledFailure \ build-Otel9WaitForCondition build-Otel10WaitForCallback build-Otel11ChainedInvoke \ - build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ + build-Otel27ChainedInvokeSuspend build-Otel27InvokeTarget build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ build-Otel14MapFailure build-Otel15WaitInterrupted build-Otel16WaitForConditionFailure \ build-Otel17WaitForCallbackFailure build-Otel18ChainedInvokeFailure \ build-Otel18InvokeTarget build-Otel19ExecutionFailure build-Otel20VirtualContext \ + build-Otel21CompletedStepReplay build-Otel22UserFunctionContext build-Otel23CallbackFunctionContext build-Otel24InvocationRetryStatus \ build-OtelExecution1Success \ build-OtelExecution2WaitResume build-OtelExecution3Retry \ build-OtelExecution4TerminalFailure build-OtelExecution5ChildContext \ build-OtelExecution6Parallel build-OtelExecution7Map build-OtelExecution8HandledFailure \ build-OtelExecution9WaitForCondition build-OtelExecution10WaitForCallback \ - build-OtelExecution11ChainedInvoke build-OtelExecution11InvokeTarget \ + build-OtelExecution11ChainedInvoke build-OtelExecution27ChainedInvokeSuspend build-OtelExecution27InvokeTarget build-OtelExecution11InvokeTarget \ build-OtelExecution12ChildContextFailure build-OtelExecution13ParallelFailure \ build-OtelExecution14MapFailure build-OtelExecution15WaitInterrupted \ build-OtelExecution16WaitForConditionFailure \ build-OtelExecution17WaitForCallbackFailure build-OtelExecution18ChainedInvokeFailure \ build-OtelExecution18InvokeTarget build-OtelExecution19ExecutionFailure \ build-OtelExecution20VirtualContext \ + build-OtelExecution21CompletedStepReplay build-OtelExecution22UserFunctionContext build-OtelExecution23CallbackFunctionContext \ + build-OtelExecution24InvocationRetryStatus \ build-OtelLongRunning1Wait \ build-OtelLongRunning2Retry build-OtelLongRunning3Callback \ build-OtelLongRunning4ChainedInvoke build-OtelLongRunning4InvokeTarget @@ -24,22 +27,25 @@ build-Otel1Success build-Otel2WaitResume build-Otel3Retry build-Otel4TerminalFailure \ build-Otel5ChildContext build-Otel6Parallel build-Otel7Map build-Otel8HandledFailure \ build-Otel9WaitForCondition build-Otel10WaitForCallback build-Otel11ChainedInvoke \ -build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ +build-Otel27ChainedInvokeSuspend build-Otel27InvokeTarget build-Otel11InvokeTarget build-Otel12ChildContextFailure build-Otel13ParallelFailure \ build-Otel14MapFailure build-Otel15WaitInterrupted build-Otel16WaitForConditionFailure \ build-Otel17WaitForCallbackFailure build-Otel18ChainedInvokeFailure \ build-Otel18InvokeTarget build-Otel19ExecutionFailure build-Otel20VirtualContext \ +build-Otel21CompletedStepReplay build-Otel22UserFunctionContext build-Otel23CallbackFunctionContext build-Otel24InvocationRetryStatus \ build-OtelExecution1Success \ build-OtelExecution2WaitResume build-OtelExecution3Retry \ build-OtelExecution4TerminalFailure build-OtelExecution5ChildContext \ build-OtelExecution6Parallel build-OtelExecution7Map build-OtelExecution8HandledFailure \ build-OtelExecution9WaitForCondition build-OtelExecution10WaitForCallback \ -build-OtelExecution11ChainedInvoke build-OtelExecution11InvokeTarget \ +build-OtelExecution11ChainedInvoke build-OtelExecution27ChainedInvokeSuspend build-OtelExecution27InvokeTarget build-OtelExecution11InvokeTarget \ build-OtelExecution12ChildContextFailure build-OtelExecution13ParallelFailure \ build-OtelExecution14MapFailure build-OtelExecution15WaitInterrupted \ build-OtelExecution16WaitForConditionFailure \ build-OtelExecution17WaitForCallbackFailure build-OtelExecution18ChainedInvokeFailure \ build-OtelExecution18InvokeTarget build-OtelExecution19ExecutionFailure \ build-OtelExecution20VirtualContext \ +build-OtelExecution21CompletedStepReplay build-OtelExecution22UserFunctionContext build-OtelExecution23CallbackFunctionContext \ +build-OtelExecution24InvocationRetryStatus \ build-OtelLongRunning1Wait \ build-OtelLongRunning2Retry build-OtelLongRunning3Callback \ build-OtelLongRunning4ChainedInvoke build-OtelLongRunning4InvokeTarget: 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/src/otel_21_completed_step_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_21_completed_step_replay.py new file mode 100644 index 000000000..2204a82d5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_21_completed_step_replay.py @@ -0,0 +1,36 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Normal successful replay of a completed step for OTel case 21.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import Duration +from common import otel_plugin_factory, require_scenario + + +@durable_step +def before_wait(_step_context: StepContext) -> str: + return "before" + + +@durable_step +def after_wait(_step_context: StepContext) -> str: + return "after" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "completed-step-replay") + before = context.step(before_wait(), name="otel-before-wait") + context.wait(Duration.from_seconds(1), name="otel-replay-wait") + after = context.step(after_wait(), name="otel-after-wait") + return f"{before}-{after}" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py new file mode 100644 index 000000000..7e983dfc3 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_22_user_function_context.py @@ -0,0 +1,113 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Observe the real active SDK context inside public user callbacks.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, +) +from aws_durable_execution_sdk_python.config import ( + Duration, + MapConfig, + ParallelBranch, + ParallelConfig, +) +from common import otel_plugin_factory, require_scenario +from opentelemetry import trace + + +def probe(label: str) -> None: + if not trace.get_current_span().get_span_context().is_valid: + raise RuntimeError(f"No active span for conformance.{label}") + span = trace.get_tracer("aws-durable-execution-conformance").start_span( + f"conformance.{label}", attributes={"conformance.callback": label} + ) + span.end() + + +def step_body(_context: StepContext) -> str: + probe("step") + return "step" + + +def child_step(_context: StepContext) -> str: + probe("child-step") + return "child-step" + + +def child_body(context: DurableContext) -> str: + probe("child") + context.step(child_step, name="otel-context-child-step") + probe("child-restored") + return "child" + + +def parallel_step_a(_context: StepContext) -> str: + probe("parallel-step-a") + return "a" + + +def parallel_step_b(_context: StepContext) -> str: + probe("parallel-step-b") + return "b" + + +def parallel_a(context: DurableContext) -> str: + probe("parallel-a") + return context.step(parallel_step_a, name="otel-context-branch-step-a") + + +def parallel_b(context: DurableContext) -> str: + probe("parallel-b") + return context.step(parallel_step_b, name="otel-context-branch-step-b") + + +def iteration_name(_item: int, index: int) -> str: + return ("otel-context-iteration-0", "otel-context-iteration-1")[index] + + +def mapper( + context: DurableContext, item: int, index: int, _items: Sequence[int] +) -> int: + probe(("map-0", "map-1")[index]) + + def map_step(_step_context: StepContext) -> int: + probe(("map-step-0", "map-step-1")[index]) + return item + + return context.step( + map_step, name=("otel-context-map-step-0", "otel-context-map-step-1")[index] + ) + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "user-function-context") + probe("handler") + context.step(step_body, name="otel-context-step") + context.run_in_child_context(child_body, name="otel-context-child") + context.parallel( + [ + ParallelBranch(parallel_a, name="otel-context-branch-a"), + ParallelBranch(parallel_b, name="otel-context-branch-b"), + ], + name="otel-context-parallel", + config=ParallelConfig(max_concurrency=2), + ) + context.map( + [0, 1], + mapper, + name="otel-context-map", + config=MapConfig(max_concurrency=2, item_namer=iteration_name), + ) + probe("handler-restored") + context.wait(Duration.from_seconds(1), name="otel-context-resume") + probe("handler-after-resume") + return "context-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py new file mode 100644 index 000000000..d6b1bbf06 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_23_callback_function_context.py @@ -0,0 +1,136 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Observe callback contexts at SDK-owned user-function lifecycle boundaries.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, +) +from aws_durable_execution_sdk_python.config import ( + ChildConfig, + Duration, + JitterStrategy, + StepConfig, +) +from aws_durable_execution_sdk_python.exceptions import ChildContextError +from aws_durable_execution_sdk_python.retries import ( + RetryDecision, + RetryStrategyConfig, + WithRetryConfig, + create_retry_strategy, + with_retry, +) +from aws_durable_execution_sdk_python.types import ( + WaitForCallbackContext, + WaitForConditionCheckContext, +) +from aws_durable_execution_sdk_python.waits import ( + WaitForConditionConfig, + WaitForConditionDecision, +) +from common import otel_plugin_factory, require_scenario +from opentelemetry import trace + + +HELPER_FAILURE = "intentional-helper-failure" + + +def probe(label: str) -> None: + if not trace.get_current_span().get_span_context().is_valid: + raise RuntimeError(f"No active span for conformance.{label}") + span = trace.get_tracer("aws-durable-execution-conformance").start_span( + f"conformance.{label}", attributes={"conformance.callback": label} + ) + span.end() + + +def retry_step(context: StepContext) -> str: + probe(f"retry-attempt-{context.attempt}") + if context.attempt == 1: + raise RuntimeError("intentional-step-retry") + return "retried" + + +def check_condition(state: int, _context: WaitForConditionCheckContext) -> int: + next_state = state + 1 + probe(f"condition-check-{next_state}") + return next_state + + +def wait_strategy(state: int, _attempt: int) -> WaitForConditionDecision: + if state >= 2: + return WaitForConditionDecision.stop_polling() + return WaitForConditionDecision.continue_waiting(Duration.from_seconds(1)) + + +def submit_callback(_callback_id: str, _context: WaitForCallbackContext) -> None: + probe("callback-submitter") + + +def helper_body(_context: DurableContext, _attempt: int) -> str: + probe("with-retry-body") + raise RuntimeError(HELPER_FAILURE) + + +def helper_retry_strategy(_error: Exception, _attempt: int) -> RetryDecision: + probe("with-retry-strategy") + return RetryDecision.no_retry() + + +def virtual_child(_context: DurableContext) -> str: + probe("virtual-child") + return "virtual" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "callback-function-context") + context.step( + retry_step, + name="otel-context-retry-step", + config=StepConfig( + retry_strategy=create_retry_strategy( + RetryStrategyConfig( + max_attempts=2, + initial_delay=Duration.from_seconds(1), + max_delay=Duration.from_seconds(1), + backoff_rate=1.0, + jitter_strategy=JitterStrategy.NONE, + retryable_error_types=[RuntimeError], + ) + ) + ), + ) + context.wait_for_condition( + check_condition, + name="otel-context-condition", + config=WaitForConditionConfig(initial_state=0, wait_strategy=wait_strategy), + ) + context.wait_for_callback(submit_callback, name="otel-context-callback") + + # These callbacks run after the last asynchronous wait, so normal replay + # does not repeat the helper body or the checkpointless virtual child. + try: + with_retry( + context, + helper_body, + WithRetryConfig(retry_strategy=helper_retry_strategy), + name="otel-context-with-retry", + ) + except ChildContextError as error: + if error.message != HELPER_FAILURE: + raise + else: + raise RuntimeError("Expected the intentional helper failure") + context.run_in_child_context( + virtual_child, + name="otel-context-virtual", + config=ChildConfig(is_virtual=True), + ) + return "callback-context-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py new file mode 100644 index 000000000..382b7bb4a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_24_invocation_retry_status.py @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""A real invocation retry for OTel status-mapping case 24.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.exceptions import InvocationError +from common import otel_plugin_factory, require_scenario + + +@durable_step +def before_invocation_retry(_step_context: StepContext) -> str: + return "saved" + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler(event: dict[str, Any], context: DurableContext) -> str: + require_scenario(event, "invocation-retry-status") + # Capture this at entry: consuming the completed step can end replay mode. + entered_replay = context.is_replaying() + context.step(before_invocation_retry(), name="otel-before-invocation-retry") + if not entered_replay: + raise InvocationError("Conformance invocation retry after a completed step") + return "retry-complete" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py new file mode 100644 index 000000000..d3137cf29 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/src/otel_27_chained_invoke_suspend.py @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Chained invoke with child suspension for both OTel view case 27 requirements.""" + +from __future__ import annotations + +import os +from typing import Any + +from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import Duration +from common import otel_plugin_factory, require_scenario + + +@durable_execution(plugins=[otel_plugin_factory()]) +def handler( + event: dict[str, Any], + context: DurableContext, +) -> dict[str, Any]: + require_scenario(event, "chained-invoke-suspend") + return context.invoke( + function_name=os.environ["OTEL_INVOKE_TARGET_FUNCTION_NAME"], + payload=event, + name="otel-suspended-invoke", + ) + + +@durable_execution(plugins=[otel_plugin_factory()]) +def target_handler( + event: dict[str, Any], + context: DurableContext, +) -> dict[str, Any]: + require_scenario(event, "chained-invoke-suspend") + context.wait( + Duration.from_seconds(1), + name="otel-suspended-target-wait", + ) + return event 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..05120fc86 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,18 @@ Resources: TestingMetadata: TestDescription: - otel-invocation-1 + NotImplemented: + - 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. Properties: CodeUri: src/ Handler: otel_1_success.handler @@ -409,6 +421,90 @@ Resources: FunctionName: !Sub "${AWS::StackName}-otel-invocation-20" Role: !Ref LambdaExecutionRoleArn + Otel21CompletedStepReplay: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-21 + Properties: + CodeUri: src/ + Handler: otel_21_completed_step_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-21" + Role: !Ref LambdaExecutionRoleArn + + Otel22UserFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-22 + Properties: + CodeUri: src/ + Handler: otel_22_user_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-22" + Role: !Ref LambdaExecutionRoleArn + + Otel23CallbackFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-23 + Properties: + CodeUri: src/ + Handler: otel_23_callback_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-23" + Role: !Ref LambdaExecutionRoleArn + + Otel24InvocationRetryStatus: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-24 + Properties: + CodeUri: src/ + Handler: otel_24_invocation_retry_status.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-24" + Role: !Ref LambdaExecutionRoleArn + + Otel27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-invocation-27 + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${Otel27InvokeTarget.Arn}:$LATEST" + + Otel27InvokeTarget: + Type: AWS::Serverless::Function + Condition: DeployInvocationView + Metadata: + BuildMethod: makefile + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.target_handler + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27-target" + Role: !Ref LambdaExecutionRoleArn + OtelExecution1Success: Type: AWS::Serverless::Function Condition: DeployExecutionView @@ -417,6 +513,18 @@ Resources: TestingMetadata: TestDescription: - otel-execution-1 + NotImplemented: + - 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. Properties: CodeUri: src/ Handler: otel_1_success.handler @@ -781,3 +889,103 @@ Resources: Environment: Variables: OTEL_PLUGIN_MODE: execution + + OtelExecution21CompletedStepReplay: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-21 + Properties: + CodeUri: src/ + Handler: otel_21_completed_step_replay.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-21" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution22UserFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-22 + Properties: + CodeUri: src/ + Handler: otel_22_user_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-22" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution23CallbackFunctionContext: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-23 + Properties: + CodeUri: src/ + Handler: otel_23_callback_function_context.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-23" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution24InvocationRetryStatus: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-24 + Properties: + CodeUri: src/ + Handler: otel_24_invocation_retry_status.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-24" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + + OtelExecution27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + TestingMetadata: + TestDescription: + - otel-execution-27 + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-27" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${OtelExecution27InvokeTarget.Arn}:$LATEST" + + OtelExecution27InvokeTarget: + Type: AWS::Serverless::Function + Condition: DeployExecutionView + Metadata: + BuildMethod: makefile + Properties: + CodeUri: src/ + Handler: otel_27_chained_invoke_suspend.target_handler + FunctionName: !Sub "${AWS::StackName}-otel-execution-27-target" + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py new file mode 100644 index 000000000..ea2e035c1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_callback_function_context.py @@ -0,0 +1,124 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Verify actual SDK callback parents across retries and external completion.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import time +from collections import Counter +from pathlib import Path + +import pytest +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context, trace +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_DIR = Path(__file__).resolve().parents[1] / "src" +EXPECTED_PARENTS = { + "retry-attempt-1": "otel-context-retry-step attempt 1", + "retry-attempt-2": "otel-context-retry-step attempt 2", + "condition-check-1": "otel-context-condition attempt 1", + "condition-check-2": "otel-context-condition attempt 2", + "callback-submitter": "otel-context-callback submitter attempt 1", + "with-retry-body": "otel-context-with-retry", + "with-retry-strategy": "otel-context-with-retry", + "virtual-child": "otel-context-virtual", +} + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_callback_probes_keep_their_actual_sdk_parent( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "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) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_23_callback_function_context", + SRC_DIR / "otel_23_callback_function_context.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "callback-function-context"}), timeout=30 + ) + callback_id = runner.wait_for_callback( + arn, name="otel-context-callback create callback id", timeout=10 + ) + # External delivery delay, outside the durable handler. + time.sleep(1) + runner.send_callback_success(callback_id, result=b"callback-complete") + result = runner.wait_for_result(arn, timeout=15) + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "callback-context-complete" + spans = exporter.get_finished_spans() + probes = [span for span in spans if span.name.startswith("conformance.")] + assert Counter(span.name for span in probes) == Counter( + f"conformance.{label}" for label in EXPECTED_PARENTS + ) + spans_by_id = { + (span.context.trace_id, span.context.span_id): span + for span in spans + if span.context is not None + } + sdk_traces = { + span.context.trace_id + for span in spans + if span.context is not None + and span.attributes is not None + and span.attributes.get("durable.execution.arn") == arn + } + for probe_span in probes: + assert probe_span.context is not None + assert probe_span.parent is not None + assert probe_span.attributes is not None + label = str(probe_span.attributes["conformance.callback"]) + assert "durable.execution.arn" not in probe_span.attributes + assert probe_span.context.trace_id in sdk_traces + parent = spans_by_id[ + (probe_span.context.trace_id, probe_span.parent.span_id) + ] + assert parent.name == EXPECTED_PARENTS[label] + assert parent.attributes is not None + assert parent.attributes["durable.execution.arn"] == arn + assert context.get_current() == before_context + finally: + provider.shutdown() 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_chained_invoke_suspend.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_suspend.py new file mode 100644 index 000000000..9e39b54b8 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_chained_invoke_suspend.py @@ -0,0 +1,135 @@ +"""The real case 27 child suspends and resumes before its caller completes.""" + +import importlib.util +import json +from pathlib import Path +import sys + +import pytest +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_case27_resumes_the_real_child_and_preserves_its_result(view, monkeypatch): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) + monkeypatch.setenv("OTEL_INVOKE_TARGET_FUNCTION_NAME", "case27-child") + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + spec = importlib.util.spec_from_file_location("common", SRC / "common.py") + common = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "common", common) + spec.loader.exec_module(common) + 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_27_chained_invoke_suspend", SRC / "otel_27_chained_invoke_suspend.py" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + payload = {"scenario": "chained-invoke-suspend", "nested": {"value": 42}} + try: + with DurableFunctionTestRunner( + handler=module.handler, + skip_time=False, + poll_interval=0.01, + execution_timeout=30, + ) as runner: + runner.register_durable_function("case27-child", module.target_handler) + arn = runner.run_async(input=json.dumps(payload)) + result = runner.wait_for_result(arn, timeout=15) + history = runner.get_execution_history(arn, include_execution_data=True) + starts = [ + e for e in history.events if e.event_type == "ChainedInvokeStarted" + ] + ends = [ + e for e in history.events if e.event_type == "ChainedInvokeSucceeded" + ] + assert len(starts) == len(ends) == 1 + assert starts[0].operation_id == ends[0].operation_id + child_arn = starts[0].chained_invoke_started_details.durable_execution_arn + child_history = runner.get_execution_history( + child_arn, include_execution_data=True + ) + + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == payload + for execution_history in (history, child_history): + assert ( + sum( + e.event_type == "InvocationCompleted" + for e in execution_history.events + ) + == 2 + ) + assert ( + sum( + e.event_type == "ExecutionStarted" for e in execution_history.events + ) + == 1 + ) + assert sum(e.event_type == "WaitStarted" for e in child_history.events) == 1 + assert sum(e.event_type == "WaitSucceeded" for e in child_history.events) == 1 + spans = exporter.get_finished_spans() + for execution_arn in (arn, child_arn): + invocations = sorted( + [ + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == execution_arn + ], + key=lambda s: s.start_time, + ) + assert [s.attributes["durable.invocation.status"] for s in invocations] == [ + "PENDING", + "SUCCEEDED", + ] + assert [s.attributes["durable.invocation.first"] for s in invocations] == [ + True, + False, + ] + waits = sorted( + [s for s in spans if s.name == "otel-suspended-target-wait"], + key=lambda s: s.start_time, + ) + assert len(waits) == (2 if view is InvocationOtelPluginFactory else 1) + assert len({s.attributes["durable.operation.id"] for s in waits}) == 1 + assert all(s.attributes["durable.execution.arn"] == child_arn for s in waits) + if view is InvocationOtelPluginFactory: + assert waits[-1].links[0].context.span_id == waits[0].context.span_id + else: + resumed = next( + s + for s in spans + if s.name == "Invocation" + and s.attributes.get("durable.execution.arn") == child_arn + and not s.attributes["durable.invocation.first"] + ) + assert waits[0].links[0].context.span_id == resumed.context.span_id + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py new file mode 100644 index 000000000..7cf5c7353 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_completed_step_replay.py @@ -0,0 +1,102 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise case 21 through the public decorator and local durable runner.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context +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_DIR = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_completed_step_replay_uses_saved_result( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "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) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_21_completed_step_replay", SRC_DIR / "otel_21_completed_step_replay.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + calls: list[str] = [] + + def track(factory: Any, label: str) -> Any: + def make_step() -> Any: + step = factory() + + def counted(step_context: Any) -> str: + calls.append(label) + return step(step_context) + + return counted + + return make_step + + monkeypatch.setattr(module, "before_wait", track(module.before_wait, "before")) + monkeypatch.setattr(module, "after_wait", track(module.after_wait, "after")) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + result = runner.run( + input=json.dumps({"scenario": "completed-step-replay"}), timeout=15 + ) + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "before-after" + assert calls == ["before", "after"] + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert [ + span.attributes["durable.invocation.status"] + for span in invocations + if span.attributes is not None + ] == ["PENDING", "SUCCEEDED"] + if plugin_type is ExecutionOtelPluginFactory: + # Count raw exports, including duplicates that reuse a span ID. + for name in ("otel-before-wait", "otel-replay-wait", "otel-after-wait"): + assert len([span for span in spans if span.name == name]) == 1 + assert context.get_current() == before_context + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py new file mode 100644 index 000000000..b9e0d2793 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_invocation_retry_status.py @@ -0,0 +1,123 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise a real invocation retry through the public decorator and local runner.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, +) +from aws_durable_execution_sdk_python.plugin import InvocationEndInfo, InvocationStatus +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context +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_DIR = Path(__file__).resolve().parents[1] / "src" + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +def test_invocation_retry_preserves_completed_step( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + statuses: list[InvocationStatus] = [] + original_create = plugin.create_plugin + + def create_plugin(info: InvocationStartInfo) -> DurableInstrumentationPlugin: + instance = original_create(info) + original_end = instance.on_invocation_end + + def observe_end(info: InvocationEndInfo) -> None: + statuses.append(info.status) + original_end(info) + + monkeypatch.setattr(instance, "on_invocation_end", observe_end) + return instance + + monkeypatch.setattr(plugin, "create_plugin", create_plugin) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "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) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_24_invocation_retry_status", + SRC_DIR / "otel_24_invocation_retry_status.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + calls: list[str] = [] + + def track(factory: Any, label: str) -> Any: + def make_step() -> Any: + step = factory() + + def counted(step_context: Any) -> str: + calls.append(label) + return step(step_context) + + return counted + + return make_step + + monkeypatch.setattr( + module, + "before_invocation_retry", + track(module.before_invocation_retry, "saved"), + ) + try: + with DurableFunctionTestRunner(handler=module.handler) as runner: + result = runner.run( + input=json.dumps({"scenario": "invocation-retry-status"}), timeout=15 + ) + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == "retry-complete" + assert calls == ["saved"] + assert statuses == [InvocationStatus.RETRY, InvocationStatus.SUCCEEDED] + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert len(invocations) == 2 + assert [span.status.status_code.name for span in invocations] == ["UNSET", "OK"] + workflows = [span for span in spans if span.name == "Workflow"] + assert len(workflows) == 1 + assert workflows[0].status.status_code.name == "OK" + assert context.get_current() == before_context + finally: + 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..af5a01bde 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 @@ -50,6 +50,11 @@ ("Otel18ChainedInvokeFailure", "otel-invocation-18"), ("Otel19ExecutionFailure", "otel-invocation-19"), ("Otel20VirtualContext", "otel-invocation-20"), + ("Otel21CompletedStepReplay", "otel-invocation-21"), + ("Otel22UserFunctionContext", "otel-invocation-22"), + ("Otel23CallbackFunctionContext", "otel-invocation-23"), + ("Otel24InvocationRetryStatus", "otel-invocation-24"), + ("Otel27ChainedInvokeSuspend", "otel-invocation-27"), ("OtelExecution1Success", "otel-execution-1"), ("OtelExecution2WaitResume", "otel-execution-2"), ("OtelExecution3Retry", "otel-execution-3"), @@ -70,6 +75,11 @@ ("OtelExecution18ChainedInvokeFailure", "otel-execution-18"), ("OtelExecution19ExecutionFailure", "otel-execution-19"), ("OtelExecution20VirtualContext", "otel-execution-20"), + ("OtelExecution21CompletedStepReplay", "otel-execution-21"), + ("OtelExecution22UserFunctionContext", "otel-execution-22"), + ("OtelExecution23CallbackFunctionContext", "otel-execution-23"), + ("OtelExecution24InvocationRetryStatus", "otel-execution-24"), + ("OtelExecution27ChainedInvokeSuspend", "otel-execution-27"), ] EXPECTED_LONG_RUNNING_MAPPINGS: list[tuple[str, str]] = [ ("OtelLongRunning1Wait", "otel-long-running-1"), @@ -80,6 +90,8 @@ # Invoke targets carry no requirement of their own; the chained-invoke cases # assert on the telemetry the target produces. EXPECTED_TARGETS: dict[str, str] = { + "Otel27InvokeTarget": "otel-invocation-27-target", + "OtelExecution27InvokeTarget": "otel-execution-27-target", "Otel11InvokeTarget": "otel-invocation-11-target", "Otel18InvokeTarget": "otel-invocation-18-target", "OtelExecution11InvokeTarget": "otel-execution-11-target", @@ -89,6 +101,12 @@ "OtelLongRunning4InvokeTarget": "otel-long-running-4-target", } CHAINED_INVOKE_PAIRS: list[tuple[Path, str, str]] = [ + (TEMPLATE_PATH, "Otel27ChainedInvokeSuspend", "Otel27InvokeTarget"), + ( + TEMPLATE_PATH, + "OtelExecution27ChainedInvokeSuspend", + "OtelExecution27InvokeTarget", + ), (TEMPLATE_PATH, "Otel11ChainedInvoke", "Otel11InvokeTarget"), (TEMPLATE_PATH, "Otel18ChainedInvokeFailure", "Otel18InvokeTarget"), (TEMPLATE_PATH, "OtelExecution11ChainedInvoke", "OtelExecution11InvokeTarget"), @@ -143,6 +161,7 @@ "otel_9_wait_for_condition", "otel_10_wait_for_callback", "otel_11_chained_invoke", + "otel_27_chained_invoke_suspend", "otel_12_child_context_failure", "otel_13_parallel_failure", "otel_14_map_failure", @@ -152,6 +171,10 @@ "otel_18_chained_invoke_failure", "otel_19_execution_failure", "otel_20_virtual_context", + "otel_21_completed_step_replay", + "otel_22_user_function_context", + "otel_23_callback_function_context", + "otel_24_invocation_retry_status", "otel_long_running_1_wait", "otel_long_running_2_retry", "otel_long_running_3_callback", @@ -259,6 +282,33 @@ 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}-25", f"{suite}-26"} + 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"), [ @@ -391,6 +441,19 @@ def test_every_registered_handler_exists( ) +def test_active_callback_and_retry_handlers_keep_their_mapping( + template: dict[str, Any], +) -> None: + resources = template["Resources"] + for prefix in ("Otel", "OtelExecution"): + assert resources[prefix + "17WaitForCallbackFailure"]["Properties"][ + "Handler" + ] == ("otel_17_wait_for_callback_failure.handler") + assert resources[prefix + "24InvocationRetryStatus"]["Properties"][ + "Handler" + ] == ("otel_24_invocation_retry_status.handler") + + def test_template_deploys_only_the_selected_view(template: dict[str, Any]) -> None: assert template["Parameters"]["OtelSuite"]["AllowedValues"] == [ "all", diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py new file mode 100644 index 000000000..a54f3fa87 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests-otel/tests/test_user_function_context.py @@ -0,0 +1,262 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Exercise case 22's callback contexts through real concurrent work and resume.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from collections import Counter +from contextlib import nullcontext +from pathlib import Path +from threading import Barrier +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.execution import DurableExecutionInvocationInput +from aws_durable_execution_sdk_python.types import LambdaContext +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import context, trace +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_DIR = Path(__file__).resolve().parents[1] / "src" +CALLBACK_PARENTS = { + "step": "otel-context-step attempt 1", + "child": "otel-context-child", + "child-step": "otel-context-child-step attempt 1", + "child-restored": "otel-context-child", + "parallel-a": "otel-context-branch-a", + "parallel-step-a": "otel-context-branch-step-a attempt 1", + "parallel-b": "otel-context-branch-b", + "parallel-step-b": "otel-context-branch-step-b attempt 1", + "map-0": "otel-context-iteration-0", + "map-step-0": "otel-context-map-step-0 attempt 1", + "map-1": "otel-context-iteration-1", + "map-step-1": "otel-context-map-step-1 attempt 1", +} +HANDLER_COUNTS = {"handler": 2, "handler-restored": 2, "handler-after-resume": 1} + + +@pytest.mark.parametrize( + "plugin_type", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize( + "ambient", [False, True], ids=["no-ambient", "unrelated-ambient"] +) +def test_user_function_probes_keep_sdk_context_across_resume( + monkeypatch: pytest.MonkeyPatch, + plugin_type: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + ambient: bool, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before_context = context.get_current() + common_spec = importlib.util.spec_from_file_location( + "common", SRC_DIR / "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) + monkeypatch.setattr(common, "otel_plugin_factory", lambda: plugin) + spec = importlib.util.spec_from_file_location( + "otel_22_user_function_context", SRC_DIR / "otel_22_user_function_context.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + barriers: dict[str, Barrier] = {} + for pair in ( + ("parallel-a", "parallel-b"), + ("parallel-step-a", "parallel-step-b"), + ("map-0", "map-1"), + ("map-step-0", "map-step-1"), + ): + barrier = Barrier(2, timeout=5) + barriers.update(dict.fromkeys(pair, barrier)) + original_probe = module.probe + + def synchronized_probe(label: str) -> None: + callback_context = context.get_current() + # Both sibling callbacks must be active before either observes its + # parent. Synchronize only in the test; never supply a tracing context. + if label in barriers: + barriers[label].wait() + original_probe(label) + if label in barriers: + barriers[label].wait() + assert context.get_current() is callback_context + + monkeypatch.setattr(module, "probe", synchronized_probe) + invocation_contexts: list[tuple[context.Context, context.Context]] = [] + + def entry( + event: DurableExecutionInvocationInput, lambda_context: LambdaContext + ) -> dict[str, Any]: + # Simulate host instrumentation outside the unmodified durable handler. + # With no extracted parent, its unrelated trace must not become the + # parent of any SDK operation or handler probe. + host_context = context.get_current() + scope = ( + provider.get_tracer("host-instrumentation").start_as_current_span( + "ambient-invocation" + ) + if ambient + else nullcontext() + ) + try: + with scope: + invocation_context = context.get_current() + try: + return module.handler(event, lambda_context) + finally: + invocation_contexts.append( + (invocation_context, context.get_current()) + ) + finally: + assert context.get_current() is host_context + + try: + with DurableFunctionTestRunner(handler=entry, skip_time=False) as runner: + arn = runner.run_async( + input=json.dumps({"scenario": "user-function-context"}), + execution_timeout=30, + ) + result = runner.wait_for_result(arn, timeout=30) + history = runner.get_execution_history(arn) + assert result.status.value == "SUCCEEDED", result.error + assert result.result is not None + assert json.loads(result.result) == "context-complete" + assert [ + event.event_type + for event in history.events + if event.event_type.startswith("Wait") + ] == ["WaitStarted", "WaitSucceeded"] + + spans = exporter.get_finished_spans() + invocations = [span for span in spans if span.name == "Invocation"] + assert len(invocations) == 2 + assert [ + span.attributes["durable.invocation.status"] + for span in invocations + if span.attributes is not None + ] == ["PENDING", "SUCCEEDED"] + assert [ + span.attributes["durable.invocation.first"] + for span in invocations + if span.attributes is not None + ] == [True, False] + workflows = [span for span in spans if span.name == "Workflow"] + assert len(workflows) == 1 + assert workflows[0].context is not None + canonical_trace_id = workflows[0].context.trace_id + assert { + span.context.trace_id + for span in spans + if span.context is not None + and span.attributes is not None + and span.attributes.get("durable.execution.arn") == arn + } == {canonical_trace_id} + + probes = [span for span in spans if span.name.startswith("conformance.")] + # Count raw exports so replayed callback bodies and duplicate exports + # fail even if they reuse a span ID. + assert Counter(span.name for span in probes) == Counter( + { + f"conformance.{label}": count + for label, count in ( + {**dict.fromkeys(CALLBACK_PARENTS, 1), **HANDLER_COUNTS} + ).items() + } + ) + spans_by_id = { + (span.context.trace_id, span.context.span_id): span + for span in spans + if span.context is not None + } + handler_parent = ( + "Workflow" if plugin_type is ExecutionOtelPluginFactory else "Invocation" + ) + for probe_span in probes: + assert probe_span.context is not None + assert probe_span.context.trace_id == canonical_trace_id + assert probe_span.parent is not None + assert probe_span.parent.trace_id == canonical_trace_id + assert probe_span.attributes is not None + assert "durable.execution.arn" not in probe_span.attributes + label = str(probe_span.attributes["conformance.callback"]) + assert probe_span.name == f"conformance.{label}" + parent = spans_by_id[(canonical_trace_id, probe_span.parent.span_id)] + assert parent.name == CALLBACK_PARENTS.get(label, handler_parent) + assert parent.attributes is not None + assert parent.attributes["durable.execution.arn"] == arn + + for index, invocation in enumerate(invocations): + assert invocation.start_time is not None and invocation.end_time is not None + invocation_probes = [ + span + for span in probes + if span.start_time is not None + and invocation.start_time <= span.start_time <= invocation.end_time + ] + expected = ( + { + **dict.fromkeys(CALLBACK_PARENTS, 1), + "handler": 1, + "handler-restored": 1, + } + if index == 0 + else dict.fromkeys(HANDLER_COUNTS, 1) + ) + assert Counter(span.name for span in invocation_probes) == Counter( + {f"conformance.{label}": count for label, count in expected.items()} + ) + handler_probes = [ + span + for span in invocation_probes + if span.attributes is not None + and span.attributes["conformance.callback"] in HANDLER_COUNTS + ] + expected_parent = ( + workflows[0] + if plugin_type is ExecutionOtelPluginFactory + else invocation + ) + assert all( + span.parent == expected_parent.context for span in handler_probes + ) + + ambient_spans = [span for span in spans if span.name == "ambient-invocation"] + assert len(ambient_spans) == (2 if ambient else 0) + for span in ambient_spans: + assert span.context is not None + assert span.context.trace_id != canonical_trace_id + # Observe restoration on the runner's invocation threads, including + # PENDING, rather than checking only the pytest caller's context. + assert len(invocation_contexts) == 2 + assert all(before is after for before, after in invocation_contexts) + assert context.get_current() is before_context + finally: + provider.shutdown()