From 1f2768be45e6bcba58b35b04db4e38704c0c50df Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:20:17 -0700 Subject: [PATCH] fix(testing): match replay state to the service --- .github/scripts/install_otel_test_wheels.py | 5 +- .../otel_lifecycle_compatibility_test.py | 256 +++++++++++++++ .github/workflows/ci.yml | 5 +- .github/workflows/cloud-tests.yml | 5 +- .github/workflows/conformance-tests.yml | 5 +- .github/workflows/js-examples.yml | 5 +- .../opentelemetry-conformance-tests.yml | 5 +- CONTRIBUTING.md | 2 +- .../README.md | 4 +- .../test_cached_child_input_delivery_int.py | 295 ++++++++++++++++++ .../e2e/test_callback_input_delivery_int.py | 207 ++++++++++++ .../tests/e2e/test_late_child_parent_int.py | 245 +++++++++++++++ .../e2e/test_parent_creation_order_int.py | 226 ++++++++++++++ .../tests/e2e/test_terminal_status_int.py | 181 +++++++++++ .../README.md | 25 ++ .../docs/architecture.md | 16 +- .../checkpoint/processor.py | 5 +- .../checkpoint/processors/base.py | 8 +- .../execution.py | 58 ++-- .../executor.py | 10 +- .../invoker.py | 10 +- .../model.py | 35 ++- .../operation_state.py | 64 ++++ .../e2e/callback_updated_operations_test.py | 284 +++++++++++++++++ .../e2e/invocation_input_delivery_test.py | 154 +++++++++ ...replay_children_service_parity_int_test.py | 256 +++++++++++++++ .../tests/event_factory_test.py | 61 ++++ .../tests/execution_test.py | 114 +++++++ .../tests/executor_checkpoint_test.py | 44 +++ .../tests/executor_test.py | 4 + .../tests/invocation_input_delivery_test.py | 98 ++++++ .../tests/model_test.py | 114 +++++++ .../replay_children_service_parity_test.py | 239 ++++++++++++++ .../tests/runner_test.py | 76 +++++ .../e2e/plugin_operation_type_int_test.py | 12 +- 35 files changed, 3086 insertions(+), 47 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_cached_child_input_delivery_int.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_callback_input_delivery_int.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_late_child_parent_int.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_parent_creation_order_int.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_status_int.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py diff --git a/.github/scripts/install_otel_test_wheels.py b/.github/scripts/install_otel_test_wheels.py index 938996aca..d4f136867 100644 --- a/.github/scripts/install_otel_test_wheels.py +++ b/.github/scripts/install_otel_test_wheels.py @@ -16,6 +16,7 @@ ROOT = Path(__file__).resolve().parents[2] CORE = "aws-durable-execution-sdk-python" OTEL = CORE + "-otel" +TESTING = CORE + "-testing" def built_wheel(package: str) -> Path: @@ -72,7 +73,9 @@ def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--legacy-plugin", action="store_true") args = parser.parse_args() - packages = [CORE] if args.legacy_plugin else [CORE, OTEL] + # Both lanes use the built local runner, including its real callback-update + # producer. The legacy lane keeps the actual published OTel plugin unchanged. + packages = [CORE, TESTING] if args.legacy_plugin else [CORE, OTEL, TESTING] wheels = [built_wheel(package) for package in packages] subprocess.run( [ diff --git a/.github/tests/otel_lifecycle_compatibility_test.py b/.github/tests/otel_lifecycle_compatibility_test.py index e315464ba..ae40189d4 100644 --- a/.github/tests/otel_lifecycle_compatibility_test.py +++ b/.github/tests/otel_lifecycle_compatibility_test.py @@ -18,6 +18,7 @@ OperationStatus, OperationType, ) +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin from aws_durable_execution_sdk_python_otel.plugin_factory import ( ExecutionOtelPluginFactory, @@ -171,3 +172,258 @@ def handler(event, durable_context): provider.shutdown() contextvars.Context().run(run) + + +@pytest.mark.parametrize( + "view", [InvocationOtelPluginFactory, ExecutionOtelPluginFactory] +) +def test_installed_pair_preserves_nested_callback_terminal_on_resume(view): + """The installed Core 3 / OTel 2 pair retains callback completion on resume.""" + from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestRunner, + ) + import aws_durable_execution_sdk_python_testing as installed_testing + + import aws_durable_execution_sdk_python_otel as installed_otel + + assert "site-packages" in Path(installed_testing.__file__).resolve().parts + assert "site-packages" in Path(core_execution.__file__).resolve().parts + assert "site-packages" in Path(installed_otel.__file__).resolve().parts + assert version("aws-durable-execution-sdk-python-otel") == "2.0.0" + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + submitted = threading.Event() + suspended = threading.Event() + callback_ids = [] + updated_ids = [] + invocation_statuses = [] + side_effects = [] + + class Observer(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + updated_ids.append(list(info.updated_operations)) + + def on_invocation_end(self, info): + invocation_statuses.append(info.status.value) + if info.status.value == "PENDING": + suspended.set() + + class ObserverFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observer() + + def submit(callback_id, _context): + side_effects.append("submit") + callback_ids.append(callback_id) + submitted.set() + + def saved(_context): + side_effects.append("saved") + return "saved" + + def body(_event, context): + context.step(saved, name="saved-before") + return context.wait_for_callback(submit, name="await-external") + + handler = durable_execution(body, plugins=[plugin, ObserverFactory()]) + try: + with DurableFunctionTestRunner( + handler=handler, skip_time=False, poll_interval=0.01, execution_timeout=20 + ) as runner: + arn = runner.run_async(input="{}") + assert submitted.wait(8) and suspended.wait(8) + runner.send_callback_success(callback_ids[0], result=b"done") + result = runner.wait_for_result(arn, timeout=10) + assert result.status.value == "SUCCEEDED" + assert result.result == '"done"' + assert invocation_statuses == ["PENDING", "SUCCEEDED"] + assert side_effects == ["saved", "submit"] + assert updated_ids[0] == [] and len(updated_ids[1]) == 1 + spans = exporter.get_finished_spans() + callback_spans = [ + span for span in spans if span.name == "await-external create callback id" + ] + terminal = [ + span + for span in callback_spans + if span.attributes["durable.operation.status"] == "SUCCEEDED" + ] + assert len(terminal) == 1 + assert terminal[0].attributes["durable.operation.id"] == updated_ids[1][0] + parent_spans = [span for span in spans if span.name == "await-external"] + assert terminal[0].parent in [span.context for span in parent_spans] + assert all( + span.context.trace_id == terminal[0].context.trace_id + for span in callback_spans + ) + finally: + provider.shutdown() + + +@pytest.mark.parametrize( + "view", [InvocationOtelPluginFactory, ExecutionOtelPluginFactory] +) +def test_installed_pair_waits_for_completed_replaychildren_parent(view): + """A real large child result retains and replays its callback's parent.""" + import json + import time + + from aws_durable_execution_sdk_python.config import CallbackConfig + from aws_durable_execution_sdk_python.serdes import JsonSerDes + from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestRunner, + ) + import aws_durable_execution_sdk_python_otel as installed_otel + import aws_durable_execution_sdk_python_testing as installed_testing + + for module in (core_execution, installed_otel, installed_testing): + assert "site-packages" in Path(module.__file__).resolve().parts + assert version("aws-durable-execution-sdk-python-otel") == "2.0.0" + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + starts, ends, input_parents = [], [], [] + child_calls, saved_calls, marker_calls = [], [], [] + exports_at_marker = [] + + class Observer(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_invocation_end(self, info): + ends.append(info.status.value) + + class ObserverFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observer() + + def callback_spans(): + return [ + span + for span in exporter.get_finished_spans() + if span.name == "b" + and span.attributes.get("durable.operation.status") == "SUCCEEDED" + ] + + def saved(_context): + saved_calls.append(1) + return "saved" + + def child(context): + child_calls.append(1) + callback = context.create_callback( + name="b", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert context.step(saved, name="saved-child") == "saved" + return {"id": callback.callback_id, "large": "x" * (300 * 1024)} + + def marker(_context): + marker_calls.append(1) + exports_at_marker.extend(callback_spans()) + return "marked" + + def body(_event, context): + a = context.create_callback( + name="a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b_id = context.run_in_child_context(child, name="large-parent")["id"] + result = a.result() + assert context.step(marker, name="after-a") == "marked" + return {"a": result, "b_id": b_id} + + wrapped = durable_execution(body, plugins=[ObserverFactory(), plugin]) + + def entrypoint(event, context): + input_parents.append( + [ + (op.operation_id, op.status.value, op.context_details.replay_children) + for op in event.initial_execution_state.operations + if op.name == "large-parent" and op.context_details is not None + ] + ) + return wrapped(event, context) + + def completed(runner, arn, count): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + len(ends) >= count + and sum(e.event_type == "InvocationCompleted" for e in events) >= count + ): + return + time.sleep(0.01) + raise AssertionError("Invocation did not finish") + + try: + with DurableFunctionTestRunner( + handler=entrypoint, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + completed(runner, arn, 1) + runner.send_callback_success(b_id, result=b'"B"') + completed(runner, arn, 2) + middle_exports = callback_spans() + runner.send_callback_success(a_id, result=b'"A"') + result = runner.wait_for_result(arn, timeout=10) + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == {"a": "A", "b_id": b_id} + assert ends == ["PENDING", "PENDING", "SUCCEEDED"] + assert child_calls == [1, 1, 1] and saved_calls == [1] and marker_calls == [1] + b_info = next( + info for info in starts[1].updated_operations.values() if info.name == "b" + ) + assert input_parents[1] == [(b_info.parent_id, "SUCCEEDED", True)] + terminal = callback_spans() + assert len(middle_exports) == len(terminal) == 1 + assert exports_at_marker == middle_exports == terminal + assert terminal[0].attributes["durable.operation.id"] == b_info.operation_id + parents = [ + span + for span in exporter.get_finished_spans() + if span.attributes.get("durable.operation.id") == b_info.parent_id + and span.attributes.get("durable.operation.status") == "SUCCEEDED" + ] + assert terminal[0].parent in [span.context for span in parents] + assert all( + span.context.trace_id == terminal[0].context.trace_id for span in parents + ) + if view is InvocationOtelPluginFactory: + invocations = [ + span + for span in exporter.get_finished_spans() + if span.name == "Invocation" + ] + assert len(parents) == len(invocations) == 3 + assert terminal[0].parent == parents[1].context + assert parents[1].parent == invocations[1].context + else: + assert len(parents) == 1 + finally: + provider.shutdown() diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dfc01a16e..c9ee85ce9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,10 @@ on: branches: [ main ] pull_request: - branches: [ main ] + branches: + - main + - test/otel-conformance-feasibility + - fix/python-testing-replay-state-20261021 jobs: lint-commits: diff --git a/.github/workflows/cloud-tests.yml b/.github/workflows/cloud-tests.yml index 80b5051ed..c3c6142a3 100644 --- a/.github/workflows/cloud-tests.yml +++ b/.github/workflows/cloud-tests.yml @@ -2,7 +2,10 @@ name: Cloud tests on: pull_request: - branches: ["main"] + branches: + - main + - test/otel-conformance-feasibility + - fix/python-testing-replay-state-20261021 workflow_dispatch: inputs: lambda_endpoint: diff --git a/.github/workflows/conformance-tests.yml b/.github/workflows/conformance-tests.yml index b3d70c205..9b201a14a 100644 --- a/.github/workflows/conformance-tests.yml +++ b/.github/workflows/conformance-tests.yml @@ -8,7 +8,10 @@ on: push: branches: ["main"] pull_request: - branches: ["main"] + branches: + - main + - test/otel-conformance-feasibility + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-conformance-tests/**" diff --git a/.github/workflows/js-examples.yml b/.github/workflows/js-examples.yml index 1b911c7bb..3439d3be9 100644 --- a/.github/workflows/js-examples.yml +++ b/.github/workflows/js-examples.yml @@ -20,7 +20,10 @@ on: - ".github/workflows/js-examples.yml" - "pyproject.toml" pull_request: - branches: ["main"] + branches: + - main + - test/otel-conformance-feasibility + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python-testing/**" - ".github/scripts/js_examples/**" diff --git a/.github/workflows/opentelemetry-conformance-tests.yml b/.github/workflows/opentelemetry-conformance-tests.yml index aed97f6e8..097fed102 100644 --- a/.github/workflows/opentelemetry-conformance-tests.yml +++ b/.github/workflows/opentelemetry-conformance-tests.yml @@ -8,7 +8,10 @@ name: OpenTelemetry Conformance Tests on: pull_request: - branches: ["main"] + branches: + - main + - test/otel-conformance-feasibility + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-otel/**" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index ce5faaf30..2f90f9d3c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -72,7 +72,7 @@ hatch run dev-examples:test # run examples tests only ### Installed package compatibility testing -Build the core and OTel distributions with `hatch build` in each package, then run +Build the core, OTel and testing-library distributions with `hatch build` in each package, then run these commands from the repository root: ```bash diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index fd99e5cb4..7078bd39a 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -489,7 +489,9 @@ durable checkpoint observations still do not close or re-export that segment. The service omits descendants of completed contexts unless their ancestors retain children through `ReplayChildren`. An `UpdatedOperationIds` entry alone does not provide missing operation metadata, and the plugin does not synthesize an -operation span from that missing record. +operation span from that missing record. Public input-delivery tests use a real +large child result to retain metadata, while still asserting one saved-step +execution and terminal export with the correct parent in each view. Known completed context metadata is retained from the full visible input snapshot for descendants delivered by later checkpoint responses. Both views retain actual diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_cached_child_input_delivery_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_cached_child_input_delivery_int.py new file mode 100644 index 000000000..8185da3a7 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_cached_child_input_delivery_int.py @@ -0,0 +1,295 @@ +"""Visible terminal input updates export once with real retained children.""" + +import json +import logging +import os +from pathlib import Path +import time + +import pytest +from aws_durable_execution_sdk_python import durable_execution, durable_step +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +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, +) + + +@pytest.mark.parametrize( + ("view", "filesystem", "failed_delivery"), + [ + (ExecutionOtelPluginFactory, False, False), + (InvocationOtelPluginFactory, False, False), + (ExecutionOtelPluginFactory, True, False), + (InvocationOtelPluginFactory, True, False), + (ExecutionOtelPluginFactory, False, True), + ], +) +def test_retained_child_callback_notifies_before_input_ack( + view, filesystem, failed_delivery, tmp_path, monkeypatch, caplog +): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + store = ( + FileSystemExecutionStore.create(tmp_path / "store") + if filesystem + else InMemoryExecutionStore() + ) + starts, ends, notifications, child_calls, marker_calls = [], [], [], [], [] + attempts = [] + saved_child_calls = [] + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_invocation_end(self, info): + ends.append(info) + + def on_operation_end(self, info): + if info.name in ("a", "b"): + notifications.append((len(starts), info)) + + class ObserveFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observe() + + def saved_child(_step): + saved_child_calls.append(1) + return "saved" + + def child(context): + child_calls.append(1) + callback_id = context.create_callback( + name="b", config=CallbackConfig(serdes=JsonSerDes()) + ).callback_id + assert context.step(saved_child, name="saved-child") == "saved" + # The real service hides descendants of a completed ordinary context. + # A real large result makes this context use ReplayChildren, retaining + # B's metadata without injecting input state or hook notifications. + return {"id": callback_id, "large": "x" * (300 * 1024)} + + @durable_step + def marker(_context, result): + marker_calls.append(1) + return result + + def handler(_event, context): + a = context.create_callback( + name="a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b_id = context.run_in_child_context(child, name="cached-child")["id"] + result = context.step(marker(a.result()), name="after-a") + return {"a": result, "b_id": b_id} + + wrapped = durable_execution(handler, plugins=[ObserveFactory(), plugin]) + + def entrypoint(event, context): + attempts.append(1) + if failed_delivery and len(attempts) == 2: + raise RuntimeError("invocation failed before SDK input delivery") + return wrapped(event, context) + + def wait_for_completed(runner, arn, count): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + len(ends) >= count + and sum(e.event_type == "InvocationCompleted" for e in events) >= count + ): + return events + time.sleep(0.01) + raise AssertionError("Expected invocation did not finish") + + try: + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + wait_for_completed(runner, arn, 1) + runner.send_callback_success(b_id, result=b'"B"') + # A pre-SDK failure is a physical attempt, not an SDK End hook. + middle_history = wait_for_completed(runner, arn, 2) + middle_state = store.load(arn) + middle_updated_ids = list(middle_state.updated_operation_ids) + runner.send_callback_success(a_id, result=b'"A"') + result = runner.wait_for_result(arn, timeout=10) + final_history = runner.get_execution_history( + arn, include_execution_data=True + ).events + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == {"a": "A", "b_id": b_id} + assert [i.status.value for i in ends] == ["PENDING", "PENDING", "SUCCEEDED"] + assert child_calls == [1, 1, 1] and saved_child_calls == [1] + assert marker_calls == [1] + assert len(attempts) == (4 if failed_delivery else 3) + b_info = next( + info for info in starts[1].updated_operations.values() if info.name == "b" + ) + parent = starts[1].operations[b_info.parent_id] + assert parent.name == "cached-child" and parent.status.value == "SUCCEEDED" + b_ends = [ + (invocation, info) for invocation, info in notifications if info.name == "b" + ] + b_inputs = [ + (invocation, info) + for invocation, start in enumerate(starts, start=1) + for info in start.updated_operations.values() + if info.name == "b" + ] + terminal_spans = [ + s + for s in exporter.get_finished_spans() + if s.attributes.get("durable.operation.name") == "b" + and s.attributes.get("durable.operation.status") == "SUCCEEDED" + ] + parent_spans = [ + s + for s in exporter.get_finished_spans() + if s.attributes.get("durable.operation.id") == b_info.parent_id + and s.attributes.get("durable.operation.type") == "CONTEXT" + and s.attributes.get("durable.operation.status") == "SUCCEEDED" + ] + hook_errors = [ + r.getMessage() + for r in caplog.records + if r.levelno >= logging.ERROR and ("plugin" in r.name or "otel" in r.name) + ] + proof_dir = os.environ.get("CACHED_CHILD_PROOF_DIR") + if proof_dir: + destination = Path(proof_dir) + destination.mkdir(parents=True, exist_ok=True) + ( + destination + / f"{view.__name__}-{'fs' if filesystem else 'memory'}-failed{failed_delivery}.json" + ).write_text( + json.dumps( + { + "business_succeeded": True, + "result_b_matches_issued": True, + "invocation_statuses": [i.status.value for i in ends], + "child_calls": len(child_calls), + "saved_child_calls": len(saved_child_calls), + "retention_via_public_large_result": True, + "marker_calls": len(marker_calls), + "physical_attempts": len(attempts), + "failed_before_SDK": failed_delivery, + "input_updated_b": b_info.operation_id, + "input_parent_id": b_info.parent_id, + "parent_status": parent.status.value, + "middle_updated_ids": middle_updated_ids, + "b_notifications": [ + (n, i.operation_id, i.parent_id, i.is_replayed) + for n, i in b_ends + ], + "b_input_deliveries": [ + (n, i.operation_id, i.parent_id, i.is_replayed) + for n, i in b_inputs + ], + "b_terminal_spans": len(terminal_spans), + "hook_errors": hook_errors, + "parent_segment_span_ids": [ + s.context.span_id for s in parent_spans + ], + "b_terminal_parent_ids": [ + s.parent.span_id if s.parent else None + for s in terminal_spans + ], + "middle_history_events": [ + (e.event_id, e.event_type, e.name) for e in middle_history + ], + "final_history_events": [ + (e.event_id, e.event_type, e.name) for e in final_history + ], + }, + indent=2, + ) + + "\n" + ) + # Input-only completions arrive in the invocation snapshot; the core + # does not synthesize an End hook for an unvisited callback result. + expected_delivery = ( + 2, + b_info.operation_id, + b_info.parent_id, + "SUCCEEDED", + False, + ) + assert [ + ( + invocation, + info.operation_id, + info.parent_id, + info.status.value, + info.is_replayed, + ) + for invocation, info in b_inputs + ] == [expected_delivery] + # Any natural replay/checkpoint End notifications may repeat only in + # this invocation. Neither delivery path may report B on invocation 3. + assert { + ( + invocation, + info.operation_id, + info.parent_id, + info.status.value, + info.is_replayed, + ) + for invocation, info in b_ends + } <= {expected_delivery} + assert len(terminal_spans) == 1 + assert terminal_spans[0].parent is not None + if view is ExecutionOtelPluginFactory: + assert len(parent_spans) == 1 + assert terminal_spans[0].parent.span_id == parent_spans[0].context.span_id + else: + assert len(parent_spans) == 3 + assert len({span.context.span_id for span in parent_spans}) == 3 + invocations = [ + span + for span in exporter.get_finished_spans() + if span.name == "Invocation" + ] + assert len(invocations) == 3 + assert [span.parent.span_id for span in parent_spans] == [ + span.context.span_id for span in invocations + ] + assert terminal_spans[0].parent.span_id == parent_spans[1].context.span_id + assert not hook_errors + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_callback_input_delivery_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_callback_input_delivery_int.py new file mode 100644 index 000000000..17a943209 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_callback_input_delivery_int.py @@ -0,0 +1,207 @@ +"""Input-only callback resumes export each terminal outcome once.""" + +import json +import logging +import threading +import time + +import pytest +from aws_durable_execution_sdk_python import ( + DurableContext, + durable_execution, + durable_step, + durable_with_child_context, +) +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin +from aws_durable_execution_sdk_python.serdes import JsonSerDes +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, +) + + +@pytest.mark.parametrize( + "view,nested,page_bytes,fresh_executor,concurrent_completion", + [ + (ExecutionOtelPluginFactory, False, None, False, False), + (InvocationOtelPluginFactory, False, None, False, False), + (ExecutionOtelPluginFactory, True, None, False, False), + (InvocationOtelPluginFactory, True, None, False, False), + (InvocationOtelPluginFactory, True, 1, False, False), + (ExecutionOtelPluginFactory, False, None, True, False), + (InvocationOtelPluginFactory, True, None, False, True), + ], + ids=[ + "execution", + "invocation", + "execution-nested", + "invocation-nested", + "paged", + "fresh-executor", + "concurrent", + ], +) +def test_two_callbacks_do_not_repeat_terminal_spans_after_input_only_resume( + view, nested, page_bytes, fresh_executor, concurrent_completion, caplog, monkeypatch +): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + starts = [] + ends = [] + marker_calls = [] + second_end = threading.Event() + release_end = threading.Event() + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_operation_end(self, info): + if info.name in ("a", "b"): + ends.append((len(starts), info.name, info.status.value)) + + def on_invocation_end(self, info): + if concurrent_completion and len(starts) == 2: + second_end.set() + assert release_end.wait(6) + + class ObserveFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observe() + + observer = ObserveFactory() + + def make_plugin(): + return view( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + + @durable_step + def marker(context, value): + marker_calls.append(1) + return value + + def callbacks(context): + a = context.create_callback( + name="a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b = context.create_callback( + name="b", config=CallbackConfig(serdes=JsonSerDes()) + ) + return [a.result(), b.result()] + + @durable_with_child_context + def child(context): + return callbacks(context) + + def body(event, context: DurableContext): + result = ( + context.run_in_child_context(child(), name="group") + if nested + else callbacks(context) + ) + return context.step(marker(result), name="after-b") + + wrapped = durable_execution(body, plugins=[make_plugin(), observer]) + + def entrypoint(event, context): + target = ( + durable_execution(body, plugins=[make_plugin(), observer]) + if fresh_executor + else wrapped + ) + return target(event, context) + + try: + with DurableFunctionTestRunner( + handler=entrypoint, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + max_invocation_page_bytes=page_bytes, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + runner.send_callback_success(a_id, result=b'"A"') + if concurrent_completion: + assert second_end.wait(5) + runner.send_callback_success(b_id, result=b'"B"') + release_end.set() + else: + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + sum( + event.event_type == "InvocationCompleted" + for event in events + ) + >= 2 + ): + break + time.sleep(0.01) + else: + raise AssertionError("The second invocation did not suspend") + runner.send_callback_success(b_id, result=b'"B"') + result = runner.wait_for_result(arn, timeout=10) + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == ["A", "B"] + assert len(starts) == 3 + assert marker_calls == [1] + # Replay reads and a later checkpoint may both report the same terminal + # operation. It must not be delivered again in a later invocation. + assert set(ends) == {(2, "a", "SUCCEEDED"), (3, "b", "SUCCEEDED")} + spans = exporter.get_finished_spans() + callback_spans = [ + span + for span in spans + if (span.attributes or {}).get("durable.operation.type") == "CALLBACK" + ] + # Retain the nonterminal segments as well as the terminal segments; + # filtering down to successful spans would hide the invocation story. + expected = [("a", "SUCCEEDED"), ("b", "SUCCEEDED")] + if view is InvocationOtelPluginFactory: + expected = [ + ("b", "STARTED"), + ("a", "STARTED"), + ("a", "SUCCEEDED"), + ("b", "STARTED"), + ("b", "SUCCEEDED"), + ] + assert [ + (span.name, span.attributes["durable.operation.status"]) + for span in callback_spans + ] == expected + exported_ids = {span.context.span_id for span in spans} + assert all( + span.parent is not None and span.parent.span_id in exported_ids + for span in callback_spans + ) + assert not [ + record + for record in caplog.records + if record.levelno >= logging.ERROR + and ("plugin" in record.name or "otel" in record.name) + ] + finally: + release_end.set() + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_late_child_parent_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_late_child_parent_int.py new file mode 100644 index 000000000..25268edfe --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_late_child_parent_int.py @@ -0,0 +1,245 @@ +"""Late callback completions keep the actual parent for their invocation.""" + +import json +import logging +import os +import queue +from pathlib import Path +import threading +import time + +import pytest +from aws_durable_execution_sdk_python import durable_execution +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin +from aws_durable_execution_sdk_python.serdes import JsonSerDes +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, +) + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize( + ("large_result", "resume_before_completion"), + [(False, False), (True, False), (False, True)], +) +def test_late_callback_keeps_completed_or_replayed_parent( + view, large_result, resume_before_completion, monkeypatch, caplog +): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + starts, ends, child_calls, saved_calls, root_calls = [], [], [], [], [] + root_step_started, release_root_step = threading.Event(), threading.Event() + callback_ids = queue.Queue(maxsize=1) + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_invocation_end(self, info): + ends.append(info) + + class ObserveFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observe() + + def saved(_step): + saved_calls.append(1) + return "saved" + + def child(ctx): + child_calls.append(1) + cb = ctx.create_callback(name="b", config=CallbackConfig(serdes=JsonSerDes())) + assert ctx.step(saved, name="save-child") == "saved" + return dict( + b_id=cb.callback_id, large="x" * (300 * 1024) if large_result else "small" + ) + + def root_step(a_id, b_id): + callback_ids.put((a_id, b_id)) + root_calls.append(1) + root_step_started.set() + assert release_root_step.wait(10) + return "root-saved" + + def handler(_event, ctx): + a = ctx.create_callback(name="a", config=CallbackConfig(serdes=JsonSerDes())) + result = ctx.run_in_child_context(child, name="returned-child") + if resume_before_completion: + gate = ctx.create_callback( + name="resume-gate", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert gate.result() == "resume" + if not large_result: + assert ( + ctx.step( + lambda _: root_step(a.callback_id, result["b_id"]), + name="root-checkpoint", + ) + == "root-saved" + ) + return dict(a=a.result(), b_id=result["b_id"], length=len(result["large"])) + + wrapped = durable_execution(handler, plugins=[ObserveFactory(), plugin]) + + def completed(runner, arn, count): + deadline = time.monotonic() + 12 + while time.monotonic() < deadline: + history = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + len(ends) >= count + and sum(e.event_type == "InvocationCompleted" for e in history) >= count + ): + return history + time.sleep(0.01) + raise AssertionError("invocation completion not observed") + + def spans(name, status=None): + return [ + s + for s in exporter.get_finished_spans() + if s.attributes.get("durable.operation.name") == name + and ( + status is None or s.attributes.get("durable.operation.status") == status + ) + ] + + try: + with DurableFunctionTestRunner( + handler=wrapped, skip_time=False, poll_interval=0.01, execution_timeout=35 + ) as runner: + arn = runner.run_async(input="{}") + if large_result: + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + completed(runner, arn, 1) + else: + if resume_before_completion: + gate_id = runner.wait_for_callback( + arn, name="resume-gate", timeout=5 + ) + completed(runner, arn, 1) + runner.send_callback_success(gate_id, result=b'"resume"') + # The runner wait_for_callback helper waits for suspension; + # this case deliberately completes during the active root step. + a_id, b_id = callback_ids.get(timeout=5) + assert root_step_started.wait(5) + runner.send_callback_success(b_id, result=b'"B"') + release_root_step.set() + completed(runner, arn, 2 if large_result or resume_before_completion else 1) + middle_callbacks = spans("b", "SUCCEEDED") + middle_parents = spans("returned-child", "SUCCEEDED") + invocation_spans = [ + s for s in exporter.get_finished_spans() if s.name == "Invocation" + ] + runner.send_callback_success(a_id, result=b'"A"') + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history( + arn, include_execution_data=True + ).events + expected_invocations = 3 if large_result or resume_before_completion else 2 + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == dict( + a="A", b_id=b_id, length=300 * 1024 if large_result else 5 + ) + assert len(starts) == expected_invocations and [ + x.status.value for x in ends + ] == ["PENDING"] * (expected_invocations - 1) + ["SUCCEEDED"] + assert saved_calls == [1] and root_calls == ([] if large_result else [1]) + assert len(child_calls) == (3 if large_result else 1) + if resume_before_completion: + assert all( + info.name != "b" for info in starts[1].updated_operations.values() + ) + # Service input prunes B even while it is STARTED. Its later + # checkpoint response still carries the actual completion unchanged. + assert all(info.name != "b" for info in starts[1].operations.values()) + completed_parent = next( + info + for info in starts[1].operations.values() + if info.name == "returned-child" + ) + assert completed_parent.status.value == "SUCCEEDED" + hook_errors = [ + r.getMessage() + for r in caplog.records + if r.levelno >= logging.ERROR and ("plugin" in r.name or "otel" in r.name) + ] + proof = dict( + view=view.__name__, + large_result=large_result, + resume_before_completion=resume_before_completion, + statuses=[x.status.value for x in ends], + saved_step_calls=len(saved_calls), + root_step_calls=len(root_calls), + child_calls=len(child_calls), + middle_input_updated_names=[ + info.name for info in starts[1].updated_operations.values() + ] + if resume_before_completion + else None, + middle_callback_spans=[ + dict(span_id=s.context.span_id, parent_id=s.parent.span_id) + for s in middle_callbacks + ], + middle_parent_spans=[ + dict(span_id=s.context.span_id, parent_id=s.parent.span_id) + for s in middle_parents + ], + middle_invocation_ids=[s.context.span_id for s in invocation_spans], + final_callback_spans=len(spans("b", "SUCCEEDED")), + hook_errors=hook_errors, + history=[(e.event_id, e.event_type, e.name) for e in history], + ) + if os.environ.get("LATE_PARENT_PROOF_DIR"): + out = Path(os.environ["LATE_PARENT_PROOF_DIR"]) + out.mkdir(parents=True, exist_ok=True) + ( + out + / f"{view.__name__}-large{large_result}-resume{resume_before_completion}.json" + ).write_text(json.dumps(proof, indent=2) + "\n") + assert len(middle_callbacks) == 1 + assert len(spans("b", "SUCCEEDED")) == 1 + if large_result and view is InvocationOtelPluginFactory: + assert len(middle_parents) == 2 + assert ( + middle_parents[-1].parent.span_id + == invocation_spans[-1].context.span_id + ) + assert ( + middle_callbacks[0].parent.span_id == middle_parents[-1].context.span_id + ) + assert ( + middle_callbacks[0].parent.span_id != middle_parents[0].context.span_id + ) + else: + assert ( + middle_callbacks[0].parent.span_id == middle_parents[0].context.span_id + ) + assert not hook_errors + finally: + release_root_step.set() + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_parent_creation_order_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_parent_creation_order_int.py new file mode 100644 index 000000000..7d5701115 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_parent_creation_order_int.py @@ -0,0 +1,226 @@ +"""A real completion can precede its replayed sibling's parent span.""" + +from __future__ import annotations + +import json +import threading +import time +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import ( + CallbackConfig, + Duration, + ParallelConfig, +) +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + OperationType, + UserFunctionStartInfo, +) +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry import trace +from opentelemetry.sdk.trace import ReadableSpan, 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 ( + InvocationOtelPlugin, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +def _span_start(span: ReadableSpan) -> int: + assert span.start_time is not None + return span.start_time + + +def test_early_sibling_callback_exports_under_real_parent_before_user_observation( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + instances: list[InvocationOtelPlugin] = [] + + class RecordingFactory(InvocationOtelPluginFactory): + def create_plugin(self, info: InvocationStartInfo) -> InvocationOtelPlugin: + instance = super().create_plugin(info) + instances.append(instance) + return instance + + plugin = RecordingFactory( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + second_start = threading.Event() + both_completed = threading.Event() + end_observed = threading.Event() + parent_entered = threading.Event() + generations = [0] + observed: list[tuple[int, int]] = [] + hook_order: list[str] = [] + + class Gate(DurableInstrumentationPlugin): + def on_invocation_start(self, _info: InvocationStartInfo) -> None: + generations[0] += 1 + if generations[0] == 2: + second_start.set() + assert both_completed.wait(10) + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if generations[0] == 2 and info.name == "parallel-branch-1": + hook_order.append("parent-gated") + assert end_observed.wait(10) + parent_entered.set() + + class GateFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Gate() + + class ObserveEnd(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if ( + generations[0] == 2 + and info.name == "target-1" + and info.operation_type is OperationType.CALLBACK + ): + # This observer follows OTel in the actual SDK dispatch order. + assert not parent_entered.is_set() + hook_order.append("early-completion-observed") + end_observed.set() + + class ObserveEndFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return ObserveEnd() + + def branch(child: DurableContext, index: int) -> Any: + value = child.create_callback( + name=f"target-{index}", config=CallbackConfig(serdes=JsonSerDes()) + ).result() + if index == 1: + finished = [ + span + for span in exporter.get_finished_spans() + if span.name == "target-1" + ] + observed.append( + (len(finished), trace.get_current_span().get_span_context().span_id) + ) + with provider.get_tracer("customer").start_as_current_span( + "after-target-1" + ): + pass + return value + + def handler(_event: Any, durable: DurableContext) -> list[Any]: + result = durable.parallel( + [lambda child: branch(child, 0), lambda child: branch(child, 1)], + name="parallel", + config=ParallelConfig(max_concurrency=2), + ) + durable.wait(Duration.from_seconds(1), name="later-one") + durable.wait(Duration.from_seconds(1), name="later-two") + return result.get_results() + + wrapped = durable_execution( + handler, plugins=[GateFactory(), plugin, ObserveEndFactory()] + ) + try: + with DurableFunctionTestRunner( + handler=wrapped, poll_interval=0.01, skip_time=False, execution_timeout=25 + ) as runner: + arn = runner.run_async(input="{}") + deadline = time.monotonic() + 10 + callbacks: dict[str, str] = {} + while time.monotonic() < deadline: + history = runner.get_execution_history( + arn, include_execution_data=True + ).events + callbacks = {} + for event in history: + if event.event_type == "CallbackStarted": + assert event.name is not None + assert event.callback_started_details is not None + assert event.callback_started_details.callback_id is not None + callbacks[event.name] = ( + event.callback_started_details.callback_id + ) + if len(callbacks) == 2 and any( + event.event_type == "InvocationCompleted" for event in history + ): + break + time.sleep(0.01) + assert set(callbacks) == {"target-0", "target-1"} + runner.send_callback_success( + callbacks["target-0"], result=json.dumps("left").encode() + ) + assert second_start.wait(10) + # The second callback completes after the invocation input snapshot, + # so another branch's checkpoint returns this genuine new completion. + runner.send_callback_success( + callbacks["target-1"], result=json.dumps("right").encode() + ) + both_completed.set() + result = runner.wait_for_result(arn, timeout=15) + history = runner.get_execution_history( + arn, include_execution_data=True + ).events + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == ["left", "right"] + assert end_observed.is_set() and parent_entered.is_set() + assert len(observed) == 1 and observed[0][0] == 2 + spans = exporter.get_finished_spans() + # Keep every span: Invocation view exports a nonterminal segment in + # invocation one and a distinct terminal continuation in invocation two. + # Neither segment may stand in for the other or reappear on later replays. + for index in (0, 1): + segments = sorted( + [span for span in spans if span.name == f"target-{index}"], + key=_span_start, + ) + assert len(segments) == 2 + initial, terminal = segments + assert initial.attributes is not None and terminal.attributes is not None + assert initial.attributes["durable.operation.status"] == "STARTED" + assert terminal.attributes["durable.operation.status"] == "SUCCEEDED" + assert initial.context.span_id != terminal.context.span_id + parents = sorted( + [span for span in spans if span.name == f"parallel-branch-{index}"], + key=_span_start, + ) + assert len(parents) == 2 + assert initial.parent is not None + assert initial.parent.span_id == parents[0].context.span_id + assert terminal.parent is not None + assert terminal.parent.span_id == parents[1].context.span_id + assert initial.end_time is not None and terminal.start_time is not None + assert initial.end_time <= terminal.start_time + if index == 1: + assert terminal.parent.span_id == observed[0][1] + (marker,) = [span for span in spans if span.name == "after-target-1"] + assert terminal.end_time is not None and marker.start_time is not None + assert terminal.end_time <= marker.start_time + assert sum(event.event_type == "InvocationCompleted" for event in history) == 4 + assert not [record for record in caplog.records if record.exc_info] + assert len(instances) == 4 + assert len({id(instance) for instance in instances}) == 4 + assert all(instance._context_tokens == {} for instance in instances) + finally: + both_completed.set() + end_observed.set() + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_status_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_status_int.py new file mode 100644 index 000000000..f7368d22a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_status_int.py @@ -0,0 +1,181 @@ +"""A caught durable failure remains an error span without invented diagnostics.""" + +import json +import threading + +import pytest +from aws_durable_execution_sdk_python import durable_execution, durable_step +from aws_durable_execution_sdk_python.exceptions import CallbackError +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStatus, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize("completion", ["initial", "resume"]) +@pytest.mark.parametrize("outcome", ["empty-failure", "rich-failure", "success"]) +def test_caught_callback_status_survives_two_later_replays( + view, completion, outcome, monkeypatch +): + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = view( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + created = threading.Event() + completed = threading.Event() + suspended = threading.Event() + finished = threading.Condition() + completed_invocations = [] + invocations = [] + target_ends = [] + side_effects = [] + first_exports = [] + caught = [] + callback_ids = [] + + class Observer(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + invocations.append(info) + + def on_operation_end(self, info): + if info.name == "target": + target_ends.append((len(invocations), info)) + + def on_invocation_end(self, info): + if info.status is InvocationStatus.PENDING: + suspended.set() + with finished: + completed_invocations.append(info) + finished.notify_all() + + class ObserverFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observer() + + @durable_step + def saved(_context, name, value): + side_effects.append(name) + if name == "observed": + first_exports.extend( + span for span in exporter.get_finished_spans() if span.name == "target" + ) + return value + + def body(_event, context): + callback = context.create_callback(name="target") + context.step(saved("before", "saved"), name="before") + if completion == "initial" and len(invocations) == 1: + callback_ids.append(callback.callback_id) + created.set() + assert completed.wait(5), "Driver did not complete the callback" + # A real checkpoint refreshes an external completion during the active + # invocation; on resume this completed step is consumed from history. + context.step(saved("refresh", "saved"), name="refresh") + try: + callback.result() + result = "success" + except CallbackError as error: + result = type(error).__name__ + caught.append(result) + result = context.step(saved("observed", result), name="observed") + for name in ("barrier-1", "barrier-2"): + context.create_callback(name=name).result() + return result + + handler = durable_execution(body, plugins=[plugin, ObserverFactory()]) + error = ( + ErrorObject( + message="supplied detail", type="External", data=None, stack_trace=None + ) + if outcome == "rich-failure" + else None + ) + try: + with DurableFunctionTestRunner( + handler=handler, skip_time=False, poll_interval=0.01, execution_timeout=25 + ) as runner: + arn = runner.run_async(input="{}") + if completion == "initial": + assert created.wait(5) + callback_id = callback_ids[0] + else: + callback_id = runner.wait_for_callback(arn, name="target", timeout=5) + assert suspended.wait(5) + if outcome == "success": + runner.send_callback_success(callback_id) + else: + runner.send_callback_failure(callback_id, error=error) + completed.set() + first_completion = 1 if completion == "initial" else 2 + for index, name in enumerate(("barrier-1", "barrier-2")): + callback_id = runner.wait_for_callback(arn, name=name, timeout=5) + with finished: + assert finished.wait_for( + lambda: len(completed_invocations) >= first_completion + index, + timeout=5, + ) + runner.send_callback_success(callback_id) + result = runner.wait_for_result(arn, timeout=10) + assert result.status.value == "SUCCEEDED" + expected_result = "success" if outcome == "success" else "CallbackExternalError" + assert json.loads(result.result) == expected_result + assert side_effects == ["before", "refresh", "observed"] + assert len(invocations) == (3 if completion == "initial" else 4) + assert caught == ([] if outcome == "success" else [expected_result] * 3) + # Core exposes natural replay/checkpoint observations. OTel owns export + # deduplication, while the runner must not redeliver in a later invocation. + assert target_ends + expected_status = "SUCCEEDED" if outcome == "success" else "FAILED" + for invocation_number, info in target_ends: + assert invocation_number == (1 if completion == "initial" else 2) + assert info.status.value == expected_status + assert info.error == error + spans = [ + span for span in exporter.get_finished_spans() if span.name == "target" + ] + expected_statuses = [expected_status] + if view is InvocationOtelPluginFactory and completion == "resume": + expected_statuses.insert(0, "STARTED") + assert [ + span.attributes["durable.operation.status"] for span in spans + ] == expected_statuses + assert first_exports == spans + terminal = spans[-1] + assert terminal.status.status_code is ( + StatusCode.OK if outcome == "success" else StatusCode.ERROR + ) + assert len(terminal.events) == (1 if error is not None else 0) + if error is not None: + assert terminal.events[0].attributes["exception.message"] == error.message + workflow = next( + s for s in exporter.get_finished_spans() if s.name == "Workflow" + ) + assert workflow.status.status_code is StatusCode.OK + finally: + completed.set() + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 58b21e793..fc31dbefc 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -170,6 +170,31 @@ flow, pagination, and a map of the code. ## Documentation +### Operation state on replay + +Invocation input and `GetDurableExecutionState` follow the service's retained +operation view. An operation is omitted when any ancestor is a `CONTEXT` that +has `SUCCEEDED` or `FAILED` without `ReplayChildren: true`. The walk includes +running intermediate contexts; a nested context's own retention flag cannot +override a completed ancestor that prunes it. Roots and operations whose parent +is absent are kept, and ordering does not change the rule. + +`ContextOptions.ReplayChildren` on START is ignored. A closing update records +true only when it explicitly supplies true; otherwise it records false. Testing +service responses include that closing false, while Core's public model and +generic serialization contract remain unchanged. + +Checkpoint responses still contain all updated operations. `UpdatedOperationIds` +can include an ID whose operation record is hidden from invocation input and +GetState. Raw stores, history, results and nested operation inspection retain the +complete execution. An updated ID alone does not restore missing metadata for a +plugin. A real large child result retains its children through ReplayChildren +when a test needs that public replay behavior. + +This corrects local service parity; it does not fix duplicate execution-view +spans for virtual FLAT map iterations while their map is still running. + + ### Error Handling The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see: diff --git a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md index 0a33e4ec1..4780b5a97 100644 --- a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md +++ b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md @@ -234,10 +234,18 @@ list in a checkpoint response. The delta for the next response is the operations whose `operation_last_touched_seq` is greater than `handler_seen_seq`. -The watermark advances only when a checkpoint response is built. It never -advances when invocation input is built. That is what makes retries -correct. A failed invocation never moves the watermark. So the retry -delivers the same state again. +The watermark advances when a checkpoint response is built or an invocation +response passes validation, including a `PENDING` response. Building invocation +input does not acknowledge it: input updates remain pending until delivery is +confirmed. A failed call or rejected response does not consume that input, so a +retry can receive it again. + +The invocation acknowledgment uses the sequence captured when its input was +built, on the execution's worker. It never moves a higher checkpoint watermark +backward or consumes operations that changed after the captured snapshot. This +also prevents an input-only callback resume from receiving the same terminal +update again at a later checkpoint. The acknowledgment concerns runner state +delivery, not telemetry export. ``` seq_counter handler_seen_seq delta next checkpoint diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py index 42ea9333b..f1b1d6da6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py @@ -21,6 +21,9 @@ CheckpointRequestDispatcher, ) from aws_durable_execution_sdk_python_testing.clock import RealClock +from aws_durable_execution_sdk_python_testing.operation_state import ( + prune_replay_children, +) from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) @@ -153,7 +156,7 @@ def get_execution_state( token = CheckpointToken.from_str(checkpoint_token) execution = self._store.load(token.execution_arn) return StateOutput( - operations=execution.get_navigable_operations(), + operations=prune_replay_children(execution.get_navigable_operations()), next_marker=None, ) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py index d2eaa4ec4..9ea3cb256 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py @@ -100,9 +100,11 @@ def _create_context_details(self, update: OperationUpdate) -> ContextDetails | N ContextDetails( result=update.payload, error=update.error, - replay_children=update.context_options.replay_children - if update.context_options - else False, + replay_children=( + update.action in (OperationAction.SUCCEED, OperationAction.FAIL) + and update.context_options is not None + and update.context_options.replay_children is True + ), ) if update.operation_type == OperationType.CONTEXT else None diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py index 5190b9b69..9da824e1a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py @@ -24,6 +24,9 @@ ) from aws_durable_execution_sdk_python_testing.clock import real_now +from aws_durable_execution_sdk_python_testing.operation_state import ( + prune_replay_children, +) from aws_durable_execution_sdk_python_testing.exceptions import ( IllegalStateException, InvalidParameterValueException, @@ -479,15 +482,11 @@ def record_invocation_completion( ) def mark_state_delivered(self) -> None: - """Reset the list of changed operations once an invocation's input - carries them. - - The service reports, on each invocation, the operations that - changed after the last state the handler observed. The runner - delivers state in the invocation input. So the list is reset when - that input is built, not when the invocation completes: an - operation that completes while the handler is still running is - reported on the next invocation. + """Explicitly clear all pending operation IDs after their delivery. + + The live executor uses versioned ``advance_handler_seen`` acknowledgments + so failed invocations retain their input updates and concurrent changes + newer than the acknowledged input snapshot remain pending. """ self.updated_operation_ids = [] @@ -692,6 +691,7 @@ def complete_callback_success( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_callback_failure( @@ -708,8 +708,11 @@ def complete_callback_failure( self.touch_operation(operation.operation_id) updated_callback_details = None if operation.callback_details: + # Match CallbackDetails.from_dict without depending on a store + # serialization round trip: an empty wire Error has no details. updated_callback_details = replace( - operation.callback_details, error=error + operation.callback_details, + error=error if error is not None and error.to_dict() else None, ) self.operations[index] = replace( @@ -718,6 +721,7 @@ def complete_callback_failure( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_callback_timeout( @@ -744,6 +748,7 @@ def complete_callback_timeout( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_chained_invoke( @@ -897,10 +902,9 @@ def page( ``max_size_bytes`` and, when given, ``max_items``. Second element of the tuple is a marker for the next page, or ``None`` when the page fits everything.""" - start_idx = self._resolve_marker(marker) - return self._walk_page( - self.snapshot_operations, start_idx, max_size_bytes, max_items - ) + operations = prune_replay_children(self.snapshot_operations) + start_idx = self._resolve_marker(marker, len(operations)) + return self._walk_page(operations, start_idx, max_size_bytes, max_items) def unseen_operations(self) -> list[Operation]: """Operations whose ``operation_last_touched_seq`` is strictly @@ -920,11 +924,20 @@ def unseen_operations(self) -> list[Operation]: ] def advance_handler_seen(self, seq: int) -> None: - """Advance the wrapped execution's ``handler_seen_seq`` to - ``seq`` if that represents forward progress. Monotonic: - smaller or equal values are ignored.""" + """Acknowledge delivered updates without retreating the watermark. + + Equal acknowledgments also consume old-format, unversioned update IDs. + Only versions covered by this delivery may be removed. + """ if seq > self.execution.handler_seen_seq: self.execution.handler_seen_seq = seq + # A checkpoint response or completed invocation delivered this state. + # Retain changes newer than that acknowledged snapshot. + self.execution.updated_operation_ids = [ + operation_id + for operation_id in self.execution.updated_operation_ids + if self.execution.operation_last_touched_seq.get(operation_id, 0) > seq + ] # --- internals ------------------------------------------------- @@ -953,14 +966,21 @@ def _size_for(self, op: Operation) -> int: # floor at 1 to guarantee pagination always advances. return max(self.execution.operation_size_bytes.get(op.operation_id, 0), 1) - def _resolve_marker(self, marker: str | None) -> int: + def _resolve_marker( + self, marker: str | None, operation_count: int | None = None + ) -> int: if marker is None: return 0 seq, idx = self._decode_marker(marker) if seq != self.pinned_token_sequence: msg = "Invalid marker" raise InvalidParameterValueException(msg) - if idx < 0 or idx > len(self.snapshot_operations): + count = ( + len(self.snapshot_operations) + if operation_count is None + else operation_count + ) + if idx < 0 or idx > count: msg = "Invalid marker" raise InvalidParameterValueException(msg) return idx diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index 4c3a0a71f..1092291db 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -1551,7 +1551,8 @@ def _begin_invocation( self._set_invocation_gate(execution_arn, InvocationState.INVOKING) execution.begin_new_invocation() invocation_input = self._invoker.create_invocation_input(execution=execution) - execution.mark_state_delivered() + # Keep input updates until a valid response acknowledges this snapshot. + # A failed invocation must not consume their delivery before a retry. self._store.save(execution) return execution, invocation_input, execution.seq_counter @@ -1618,6 +1619,13 @@ def _finish_invocation( # A clean invocation resets the retry counter. reset_target = self._store.load(execution_arn) reset_target.consecutive_failed_invocation_attempts = 0 + if invocation_seq is not None: + # A valid response acknowledges the input snapshot, even when this + # invocation suspended without another checkpoint. Newer changes + # must remain available for the next invocation or checkpoint. + OperationPaginatorState.pin(reset_target).advance_handler_seen( + invocation_seq + ) self._store.save(reset_target) # Release the gate: terminal -> COMPLETED, otherwise PRE_INVOKE diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 4992212f4..aec058fe5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -27,6 +27,9 @@ ResourceNotFoundException, ) from aws_durable_execution_sdk_python_testing.execution import OperationPaginatorState +from aws_durable_execution_sdk_python_testing.operation_state import ( + operation_state_to_dict, +) from aws_durable_execution_sdk_python_testing.model import ( LambdaContext, executed_version, @@ -511,10 +514,15 @@ def invoke( input.durable_execution_arn, endpoint_url, region_name ) + invocation_payload = input.to_json_dict() + invocation_payload["InitialExecutionState"]["Operations"] = [ + operation_state_to_dict(operation, json_timestamps=True) + for operation in input.initial_execution_state.operations + ] invoke_kwargs: dict[str, Any] = { "FunctionName": function_name, "InvocationType": "RequestResponse", # Synchronous invocation - "Payload": json.dumps(input.to_json_dict()), + "Payload": json.dumps(invocation_payload), } if tenant_id is not None: invoke_kwargs["TenantId"] = tenant_id diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 5c82e4d75..df41a8441 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -1,6 +1,9 @@ """Model classes for the web API.""" from __future__ import annotations +from aws_durable_execution_sdk_python_testing.operation_state import ( + operation_state_to_dict, +) import datetime @@ -572,7 +575,7 @@ def from_dict(cls, data: dict) -> GetDurableExecutionStateResponse: def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { - "Operations": [op.to_dict() for op in self.operations] + "Operations": [operation_state_to_dict(op) for op in self.operations] } if self.next_marker is not None: result["NextMarker"] = self.next_marker @@ -677,7 +680,8 @@ class EventError: @classmethod def from_dict(cls, data: dict) -> EventError: payload = None - if payload_data := data.get("Payload"): + payload_data = data.get("Payload") + if payload_data is not None: payload = ErrorObject.from_dict(payload_data) return cls( @@ -2239,6 +2243,14 @@ def create_callback_event_failed(cls, context: EventCreationContext) -> Event: event_error: EventError | None = ( EventError.from_details(callback_details) if callback_details else None ) + if ( + context.include_execution_data + and callback_details is not None + and callback_details.error is None + ): + # Detailed service history retains an empty Error.Payload object. + # This projection must not turn the SDK-facing absent error into one. + event_error = EventError(payload=ErrorObject.from_dict({}), truncated=False) return cls( event_type=EventType.CALLBACK_FAILED.value, event_timestamp=context.end_timestamp, @@ -2592,6 +2604,11 @@ class HistoryEventTypeConfig: } +def _operation_error(error: ErrorObject | None) -> ErrorObject | None: + # History retains present Payload={}, while SDK state treats Error={} as absent. + return error if error is not None and error.to_dict() else None + + def events_to_operations(events: list[Event]) -> list[Operation]: """Convert a list of history events into operations. @@ -2744,7 +2761,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: callback_details=CallbackDetails( callback_id=callback_id, result=result, - error=error, + error=_operation_error(error), ), ) @@ -2785,7 +2802,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, step_details=StepDetails( result=result_val, - error=error_val, + error=_operation_error(error_val), attempt=attempt, next_attempt_timestamp=next_attempt_ts, ), @@ -2818,7 +2835,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, context_details=ContextDetails( result=None, - error=event.context_failed_details.error.payload, + error=_operation_error( + event.context_failed_details.error.payload + ), ), ) @@ -2843,7 +2862,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, chained_invoke_details=ChainedInvokeDetails( result=None, - error=event.chained_invoke_failed_details.error.payload, + error=_operation_error( + event.chained_invoke_failed_details.error.payload + ), ), ) @@ -3139,7 +3160,7 @@ def from_dict(cls, data: dict) -> CheckpointUpdatedExecutionState: def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { - "Operations": [op.to_dict() for op in self.operations] + "Operations": [operation_state_to_dict(op) for op in self.operations] } if self.next_marker is not None: result["NextMarker"] = self.next_marker diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py new file mode 100644 index 000000000..850eea00b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py @@ -0,0 +1,64 @@ +"""Service views of retained operation state; raw execution storage stays complete.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ( + Operation, + OperationStatus, + OperationType, +) + + +_CLOSED_CONTEXT = (OperationStatus.SUCCEEDED, OperationStatus.FAILED) + + +def prune_replay_children(operations: Sequence[Operation]) -> list[Operation]: + """Omit a descendant when any known completed context prunes its children.""" + by_id = {operation.operation_id: operation for operation in operations} + visible = [] + for operation in operations: + parent_id = operation.parent_id + visited = {operation.operation_id} + hidden = False + while parent_id is not None and parent_id not in visited: + parent = by_id.get(parent_id) + if parent is None: + break + if ( + parent.operation_type is OperationType.CONTEXT + and parent.status in _CLOSED_CONTEXT + and not ( + parent.context_details + and parent.context_details.replay_children is True + ) + ): + hidden = True + break + visited.add(parent_id) + parent_id = parent.parent_id + if not hidden: + visible.append(operation) + return visible + + +def operation_state_to_dict( + operation: Operation, *, json_timestamps: bool = False +) -> dict[str, Any]: + """Project service ReplayChildren presence without changing Core's model API.""" + data = dict(operation.to_json_dict() if json_timestamps else operation.to_dict()) + if ( + operation.operation_type is OperationType.CONTEXT + and operation.context_details is not None + ): + details = dict(data.get("ContextDetails", {})) + if operation.status is OperationStatus.STARTED: + details.pop("ReplayChildren", None) + elif operation.status in _CLOSED_CONTEXT: + details["ReplayChildren"] = ( + operation.context_details.replay_children is True + ) + data["ContextDetails"] = details + return data diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py new file mode 100644 index 000000000..aded098fa --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py @@ -0,0 +1,284 @@ +"""Callback completions are observed before resumed code without later redelivery.""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from queue import Queue +from threading import Event +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import ( + CallbackConfig, + Duration, + WaitForCallbackConfig, +) +from aws_durable_execution_sdk_python.exceptions import CallbackError +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + UserFunctionStartInfo, +) +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.types import WaitForCallbackContext + +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +class CallbackObserverFactory: + def __init__(self) -> None: + self.invocations: list[InvocationStartInfo] = [] + self.ends: list[tuple[int, OperationEndInfo]] = [] + self.order: list[str] = [] + + def create_plugin(self, info: InvocationStartInfo) -> DurableInstrumentationPlugin: + return CallbackObserver(self) + + +class CallbackObserver(DurableInstrumentationPlugin): + def __init__(self, records: CallbackObserverFactory) -> None: + self.invocations = records.invocations + self.ends = records.ends + self.order = records.order + self.invocation_number = 0 + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self.invocations.append(info) + self.invocation_number = len(self.invocations) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "target": + self.ends.append((self.invocation_number, info)) + self.order.append("target-end") + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if info.name == "observed": + self.order.append("observed-start") + + +def _submit(_callback_id: str, _context: WaitForCallbackContext) -> None: + return None + + +def _suspended_callback(runner: DurableFunctionTestRunner, arn: str, name: str) -> str: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history(arn, include_execution_data=True).events + starts = [ + event + for event in events + if event.event_type == "CallbackStarted" and event.name == name + ] + if starts and any( + event.event_type == "InvocationCompleted" + and event.event_id > starts[0].event_id + for event in events + ): + details = starts[0].callback_started_details + assert details is not None + assert details.callback_id is not None + return details.callback_id + time.sleep(0.01) + raise AssertionError(f"Callback {name} did not suspend") + + +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +@pytest.mark.parametrize("filesystem", [False, True]) +def test_callback_update_is_consumed_before_user_code( + outcome: str, filesystem: bool, tmp_path: Path +) -> None: + observer = CallbackObserverFactory() + marker_calls: list[str] = [] + + @durable_step + def observed(_context: StepContext, value: str) -> str: + marker_calls.append(value) + return value + + def handler(_event: Any, context: DurableContext) -> str: + config = CallbackConfig( + timeout=Duration.from_seconds(1 if outcome == "timeout" else 30), + serdes=JsonSerDes(), + ) + try: + target = context.create_callback(name="target", config=config).result() + except CallbackError: + target = outcome + assert isinstance(target, str) + saved = context.step(observed(target), name="observed") + callback_config = WaitForCallbackConfig(serdes=JsonSerDes()) + one = context.wait_for_callback(_submit, name="one", config=callback_config) + two = context.wait_for_callback(_submit, name="two", config=callback_config) + return "/".join((saved, one, two)) + + wrapped = durable_execution(handler, plugins=[observer]) + store = FileSystemExecutionStore.create(tmp_path) if filesystem else None + with DurableFunctionTestRunner( + handler=wrapped, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + callback_id = _suspended_callback(runner, arn, "target") + if outcome == "success": + runner.send_callback_success( + callback_id, result=json.dumps("target").encode() + ) + elif outcome == "failure": + runner.send_callback_failure( + callback_id, error=ErrorObject.from_message("explicit callback failure") + ) + # The timeout case uses the runner's real scheduled callback deadline. + for name in ["one", "two"]: + callback_id = _suspended_callback(runner, arn, name + " create callback id") + runner.send_callback_success(callback_id, result=json.dumps(name).encode()) + result = runner.wait_for_result(arn, timeout=10) + + expected = "target" if outcome == "success" else outcome + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == expected + "/one/two" + assert marker_calls == [expected] + assert len(observer.invocations) == 4 + # Multiple natural hooks in the delivery invocation are allowed, but an + # acknowledged callback must never be delivered in later invocations. + assert {invocation for invocation, _ in observer.ends} == {2} + terminal = observer.ends[0][1] + assert ( + terminal.status.value + == {"success": "SUCCEEDED", "failure": "FAILED", "timeout": "TIMED_OUT"}[ + outcome + ] + ) + assert all( + end.operation_id == terminal.operation_id + and end.status is terminal.status + and end.is_replayed is False + and end.result == terminal.result + and end.error == terminal.error + for _, end in observer.ends + ) + assert set(observer.invocations[1].updated_operations) == {terminal.operation_id} + assert all( + terminal.operation_id not in invocation.updated_operations + for invocation in observer.invocations[2:] + ) + assert observer.order.count("observed-start") == 1 + assert observer.order.index("target-end") < observer.order.index("observed-start") + if outcome == "failure": + assert terminal.error is not None + assert terminal.error.message == "explicit callback failure" + + +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +@pytest.mark.parametrize("filesystem", [False, True]) +def test_callback_delivered_by_checkpoint_is_not_updated_on_later_resume( + outcome: str, filesystem: bool, tmp_path: Path +) -> None: + observer = CallbackObserverFactory() + submitter_entered = Event() + release_submitter = Event() + submission_calls: list[str] = [] + published_callbacks: Queue[str] = Queue() + + @durable_step + def submit(_context: StepContext, callback_id: str) -> str: + submission_calls.append(callback_id) + published_callbacks.put(callback_id) + submitter_entered.set() + assert release_submitter.wait(10) + return "submitted" + + def handler(_event: Any, context: DurableContext) -> str: + callback = context.create_callback( + name="target", + config=CallbackConfig( + timeout=Duration.from_seconds(1 if outcome == "timeout" else 30), + serdes=JsonSerDes(), + ), + ) + context.step(submit(callback.callback_id), name="submit") + try: + result = callback.result() + except CallbackError: + result = outcome + context.wait(Duration.from_seconds(1), name="first-replay") + context.wait(Duration.from_seconds(1), name="second-replay") + assert isinstance(result, str) + return result + + store = FileSystemExecutionStore.create(tmp_path) if filesystem else None + with DurableFunctionTestRunner( + handler=durable_execution(handler, plugins=[observer]), + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + try: + assert submitter_entered.wait(5) + callback_id = published_callbacks.get(timeout=5) + if outcome == "success": + runner.send_callback_success(callback_id, result=b'"target"') + elif outcome == "failure": + runner.send_callback_failure( + callback_id, error=ErrorObject.from_message("early failure") + ) + expected_event = { + "success": "CallbackSucceeded", + "failure": "CallbackFailed", + "timeout": "CallbackTimedOut", + }[outcome] + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if any(event.event_type == expected_event for event in events): + break + time.sleep(0.01) + else: + raise AssertionError("Callback did not complete during submission") + assert not any( + event.event_type == "InvocationCompleted" for event in events + ) + finally: + release_submitter.set() + result = runner.wait_for_result(arn, timeout=15) + + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == ("target" if outcome == "success" else outcome) + assert len(submission_calls) == 1 + assert len(observer.invocations) >= 3 + assert {invocation for invocation, _ in observer.ends} == {1} + terminal = observer.ends[0][1] + target_id = terminal.operation_id + assert all( + end.operation_id == target_id + and end.status is terminal.status + and end.is_replayed is False + and end.result == terminal.result + and end.error == terminal.error + for _, end in observer.ends + ) + assert all( + target_id not in invocation.updated_operations + for invocation in observer.invocations[1:] + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py new file mode 100644 index 000000000..6eaacf5bf --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py @@ -0,0 +1,154 @@ +"""An input-only callback resume must not redeliver its terminal update later.""" + +import json +import time + +import pytest +from aws_durable_execution_sdk_python import ( + DurableContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin +from aws_durable_execution_sdk_python_testing.execution import Execution +from aws_durable_execution_sdk_python_testing.executor import Executor, InvocationState +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("failed_delivery", [False, True]) +@pytest.mark.parametrize("legacy_format", [False, True]) +def test_callback_input_acknowledgment_survives_checkpoint_free_resume( + filesystem, failed_delivery, legacy_format, tmp_path, monkeypatch +): + store = ( + FileSystemExecutionStore.create(tmp_path / "store") + if filesystem + else InMemoryExecutionStore() + ) + starts = [] + invocation_ends = [] + operation_ends = [] + attempts = [] + marker_calls = [] + legacy_restores = [] + + original_begin = Executor._begin_invocation # noqa: SLF001 + + def begin_with_old_store(self, arn): + # Rehydrate a real completed callback using the persisted format that + # predates sequence accounting. No SDK hook or invocation input is faked. + if ( + legacy_format + and not legacy_restores + and self._invocation_gate(arn) is InvocationState.PRE_INVOKE # noqa: SLF001 + ): + execution = store.load(arn) + if execution.updated_operation_ids: + snapshot = execution.to_json_dict() + for field in ( + "SeqCounter", + "HandlerSeenSeq", + "OperationLastTouchedSeq", + "OperationSizeBytes", + ): + snapshot.pop(field) + restored = Execution.from_json_dict(snapshot) + assert restored.seq_counter == restored.handler_seen_seq == 0 + assert restored.updated_operation_ids == execution.updated_operation_ids + store.save(restored) + legacy_restores.append(snapshot) + return original_begin(self, arn) + + monkeypatch.setattr(Executor, "_begin_invocation", begin_with_old_store) + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_invocation_end(self, info): + invocation_ends.append(info) + + def on_operation_end(self, info): + if info.name in ("a", "b"): + operation_ends.append((len(starts), info.name, info.is_replayed)) + + class ObserveFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observe() + + @durable_step + def marker(context, value): + marker_calls.append(1) + return value + + @durable_execution(plugins=[ObserveFactory()]) + def handler(event, context: DurableContext): + a = context.create_callback( + name="a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b = context.create_callback( + name="b", config=CallbackConfig(serdes=JsonSerDes()) + ) + result = [a.result(), b.result()] + return context.step(marker(result), name="after-b") + + def entrypoint(event, context): + attempts.append(1) + if failed_delivery and len(attempts) == 2: + raise RuntimeError("invocation failed before the SDK received its input") + return handler(event, context) + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + runner.send_callback_success(a_id, result=b'"A"') + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + len(invocation_ends) >= 2 + and sum(event.event_type == "InvocationCompleted" for event in events) + >= 2 + ): + break + time.sleep(0.01) + else: + raise AssertionError("The second SDK invocation did not suspend") + runner.send_callback_success(b_id, result=b'"B"') + result = runner.wait_for_result(arn, timeout=10) + + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == ["A", "B"] + assert [info.status.value for info in invocation_ends] == [ + "PENDING", + "PENDING", + "SUCCEEDED", + ] + # Replay and checkpoint processing can each report the same terminal state. + # Acknowledgment prevents either callback from being redelivered to a later + # invocation; deduplicating exports within an invocation belongs to plugins. + assert set(operation_ends) == {(2, "a", False), (3, "b", False)} + assert marker_calls == [1] + assert len(attempts) == (4 if failed_delivery else 3) + assert len(legacy_restores) == int(legacy_format) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py new file mode 100644 index 000000000..3c31a9412 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py @@ -0,0 +1,256 @@ +"""Real public handlers replay through the service's retained-operation view.""" + +import json +import os +from pathlib import Path + +import pytest +from aws_durable_execution_sdk_python import durable_execution +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.exceptions import ChildContextError +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("mode", ["completed", "retained", "nested-retained", "failed"]) +def test_public_resume_uses_service_state_but_inspection_remains_complete( + filesystem, mode, tmp_path +): + store = ( + FileSystemExecutionStore.create(tmp_path / "state") + if filesystem + else InMemoryExecutionStore() + ) + snapshots = [] + child_calls = [] + step_calls = [] + + def save(_step): + step_calls.append(1) + return "saved" + + def child(ctx): + child_calls.append(1) + saved = ctx.step(save, name="saved-child-step") + if mode == "failed": + raise ValueError("stored-child-failure") + return { + "saved": saved, + "large": "x" * (300 * 1024) + if mode in ["retained", "nested-retained"] + else "", + } + + def outer(ctx): + value = ctx.run_in_child_context(child, name="inner-retained") + return {"saved": value["saved"], "large": ""} + + def handler(_event, ctx): + try: + value = ctx.run_in_child_context( + outer if mode == "nested-retained" else child, name="outer" + ) + result = {"saved": value["saved"], "length": len(value["large"])} + except ChildContextError: + assert mode == "failed" + result = {"failed": True} + gate = ctx.create_callback( + name="resume", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert gate.result() == "resume" + return result + + wrapped = durable_execution(handler) + + def entrypoint(event, context): + initial = event.initial_execution_state.operations + state = event.service_client.get_execution_state( + event.durable_execution_arn, event.checkpoint_token, "" + ) + snapshots.append( + { + "input_names": [op.name for op in initial], + "state_names": [op.name for op in state.operations], + "input_ids": [op.operation_id for op in initial], + "state_ids": [op.operation_id for op in state.operations], + "updated_ids": list(event.updated_operation_ids), + } + ) + return wrapped(event, context) + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=20, + ) as runner: + arn = runner.run_async(input="{}") + callback_id = runner.wait_for_callback(arn, name="resume", timeout=5) + 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).events + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == ( + {"failed": True} + if mode == "failed" + else {"saved": "saved", "length": 300 * 1024 if mode == "retained" else 0} + ) + assert ( + len(snapshots) == 2 and snapshots[0]["input_ids"] == snapshots[0]["state_ids"] + ) + assert snapshots[1]["input_ids"] == snapshots[1]["state_ids"] + visible = "saved-child-step" in snapshots[1]["input_names"] + assert visible is (mode == "retained") + if mode == "nested-retained": + assert "inner-retained" not in snapshots[1]["input_names"] + assert "outer" in snapshots[1]["input_names"] + assert step_calls == [1] and len(child_calls) == (2 if mode == "retained" else 1) + raw = store.load(arn) + assert any(op.name == "saved-child-step" for op in raw.operations) + inspected = result.get_operation_by_name("outer") + if mode == "nested-retained": + inspected = inspected.get_operation_by_name("inner-retained") + assert ( + inspected.get_operation_by_name("saved-child-step").name == "saved-child-step" + ) + assert ( + sum( + e.event_type == "StepSucceeded" and e.name == "saved-child-step" + for e in history + ) + == 1 + ) + assert any(e.name == "saved-child-step" for e in history) + if os.environ.get("SERVICE_PRUNING_PROOF_DIR"): + out = Path(os.environ["SERVICE_PRUNING_PROOF_DIR"]) + out.mkdir(parents=True, exist_ok=True) + (out / f"{mode}-fs{filesystem}.json").write_text( + json.dumps( + dict( + mode=mode, + filesystem=filesystem, + snapshots=snapshots, + result=json.loads(result.result), + step_calls=len(step_calls), + child_calls=len(child_calls), + history_names=[e.name for e in history], + ), + indent=2, + ) + + "\n" + ) + + +@pytest.mark.parametrize("filesystem", [False, True]) +def test_hidden_callback_updated_id_does_not_make_its_metadata_visible( + filesystem, tmp_path +): + import time + + store = ( + FileSystemExecutionStore.create(tmp_path / "hidden-state") + if filesystem + else InMemoryExecutionStore() + ) + snapshots = [] + child_calls = [] + saved_calls = [] + + def saved(_step): + saved_calls.append(1) + return "saved" + + def child(ctx): + child_calls.append(1) + callback = ctx.create_callback( + name="hidden-b", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert ctx.step(saved, name="hidden-saved") == "saved" + return callback.callback_id + + def handler(_event, ctx): + a = ctx.create_callback( + name="root-a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b_id = ctx.run_in_child_context(child, name="closed-parent") + return {"a": a.result(), "b_id": b_id} + + wrapped = durable_execution(handler) + + def entrypoint(event, context): + state = event.service_client.get_execution_state( + event.durable_execution_arn, event.checkpoint_token, "" + ) + snapshots.append( + { + "input_ids": [ + op.operation_id for op in event.initial_execution_state.operations + ], + "state_ids": [op.operation_id for op in state.operations], + "updated_ids": list(event.updated_operation_ids), + } + ) + return wrapped(event, context) + + def completed(runner, arn, count): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if sum(e.event_type == "InvocationCompleted" for e in events) >= count: + return events + time.sleep(0.01) + raise AssertionError("invocation completion not observed") + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="root-a", timeout=5) + b_id = runner.wait_for_callback(arn, name="hidden-b", timeout=5) + first = completed(runner, arn, 1) + b_operation_id = next( + e.operation_id + for e in first + if e.name == "hidden-b" and e.event_type == "CallbackStarted" + ) + runner.send_callback_success(b_id, result=b'"B"') + completed(runner, arn, 2) + runner.send_callback_success(a_id, result=b'"A"') + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history(arn, include_execution_data=True).events + assert result.status.value == "SUCCEEDED" and json.loads(result.result) == { + "a": "A", + "b_id": b_id, + } + assert len(snapshots) == 3 and child_calls == [1] and saved_calls == [1] + assert b_operation_id in snapshots[1]["updated_ids"] + assert b_operation_id not in snapshots[1]["input_ids"] + assert b_operation_id not in snapshots[1]["state_ids"] + assert any(op.operation_id == b_operation_id for op in store.load(arn).operations) + assert ( + result.get_operation_by_name("closed-parent") + .get_operation_by_name("hidden-b") + .result + == '"B"' + ) + assert ( + sum( + e.event_type == "CallbackSucceeded" and e.operation_id == b_operation_id + for e in history + ) + == 1 + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py index cd2fd8743..04334b6db 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py @@ -9,8 +9,10 @@ import pytest from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, ChainedInvokeOptions, ErrorObject, + Operation as ServiceOperation, OperationStatus, OperationType, StepDetails, @@ -849,6 +851,65 @@ def test_create_callback_failed(): assert event.callback_failed_details.error.payload.message == "Callback failed" +@pytest.mark.parametrize("include_data", [False, True]) +@pytest.mark.parametrize("has_error", [False, True]) +def test_callback_failure_history_projection_preserves_sdk_state( + include_data, has_error +): + error = ErrorObject.from_message("details") if has_error else None + operation = ServiceOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.FAILED, + callback_details=CallbackDetails(callback_id="callback-id", error=error), + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + ), + include_execution_data=include_data, + ) + event = Event.create_callback_event(context) + expected = {"Truncated": not (include_data and not has_error)} + if has_error or include_data: + expected["Payload"] = error.to_dict() if error else {} + assert event.callback_failed_details.error.to_dict() == expected + assert operation.callback_details.error is error + + +def test_callback_timeout_history_projection_is_unchanged(): + operation = ServiceOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.TIMED_OUT, + callback_details=CallbackDetails(callback_id="callback-id", error=None), + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + ), + include_execution_data=True, + ) + event = Event.create_callback_event(context) + assert event.callback_timed_out_details.error.to_dict() == {"Truncated": True} + + def test_create_callback_timed_out(): operation = create_mock_operation("callback-1", status=OperationStatus.TIMED_OUT) error_obj = ErrorObject.from_message("Callback timed out") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py index bde5e3d4b..c9457a6dd 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py @@ -3,6 +3,7 @@ import json from dataclasses import replace from datetime import datetime, timezone +from threading import Event, Thread from unittest.mock import patch, Mock import pytest @@ -955,6 +956,38 @@ def test_from_dict_with_none_result(): # region callback +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +def test_callback_completion_records_only_successful_state_changes(outcome): + """Preserve payloads, token versions and consumed metadata on rejection.""" + operation = Operation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id="callback-id"), + ) + execution = Execution("test-arn", _make_start_input(), [operation]) + complete = getattr(execution, f"complete_callback_{outcome}") + payload = b'"result"' if outcome == "success" else ErrorObject.from_message("error") + token_version = execution.token_sequence + result = complete("callback-id", payload) + + assert execution.updated_operation_ids == ["callback"] + assert execution.token_sequence == token_version + assert execution.seq_counter == 1 + assert result.callback_details.result == ( + '"result"' if outcome == "success" else None + ) + assert result.callback_details.error == (None if outcome == "success" else payload) + restored = Execution.from_json_dict(execution.to_json_dict()) + assert restored.updated_operation_ids == ["callback"] + execution.mark_state_delivered() + with pytest.raises(IllegalStateException, match="not in STARTED state"): + complete("callback-id", payload) + assert execution.updated_operation_ids == [] + assert execution.seq_counter == 1 + assert execution.token_sequence == token_version + + def test_find_callback_operation_not_found(): """Test find_callback_operation raises exception when callback not found.""" execution = Execution("test-arn", Mock(), []) @@ -1690,6 +1723,87 @@ def test_record_invocation_completion_keeps_updated_operation_ids(): assert execution.updated_operation_ids == [] +@pytest.mark.parametrize("complete_before_advance", [False, True]) +def test_checkpoint_consumes_only_updates_covered_by_its_watermark( + complete_before_advance, +): + """A late update survives reads and retries of an older state delivery.""" + execution = Execution( + "test-arn", + _make_start_input(), + [ + Operation( + operation_id=name, + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id=name), + ) + for name in ["delivered", "later"] + ], + ) + execution.complete_callback_success("delivered", b"first") + response = OperationPaginatorState.pin(execution) + ready = Event() + finished = Event() + + def complete_later(): + assert ready.wait(5) + execution.complete_callback_failure("later", ErrorObject.from_message("later")) + finished.set() + + worker = Thread(target=complete_later) + worker.start() + if complete_before_advance: + ready.set() + assert finished.wait(5) + expected = ["delivered", "later"] if complete_before_advance else ["delivered"] + assert execution.updated_operation_ids == expected + response.page(None, max_size_bytes=1024 * 1024) + assert execution.updated_operation_ids == expected + + response.advance_handler_seen(1) + ready.set() + assert finished.wait(5) + worker.join(timeout=5) + assert not worker.is_alive() + assert execution.updated_operation_ids == ["later"] + assert execution.handler_seen_seq == 1 + assert execution.token_sequence == 0 + assert execution.seq_counter == 2 + # An idempotent/older delivery cannot consume the later completion. + response.advance_handler_seen(1) + response.advance_handler_seen(0) + assert execution.updated_operation_ids == ["later"] + OperationPaginatorState.pin(execution).advance_handler_seen(2) + assert execution.updated_operation_ids == [] + + +@pytest.mark.parametrize("seen,ack", [(0, 0), (5, 5), (5, 3)]) +def test_acknowledged_updates_are_pruned_without_watermark_progress(seen, ack): + execution = Execution("test-arn", _make_start_input(), []) + snapshot = execution.to_json_dict() + for field in ( + "SeqCounter", + "HandlerSeenSeq", + "OperationLastTouchedSeq", + "OperationSizeBytes", + ): + snapshot.pop(field) + snapshot["UpdatedOperationIds"] = ["legacy", "newer"] + restored = Execution.from_json_dict(snapshot) + restored.handler_seen_seq = seen + restored.operation_last_touched_seq["newer"] = ack + 1 + paginator = OperationPaginatorState.pin(restored) + + paginator.advance_handler_seen(ack) + + assert restored.updated_operation_ids == ["newer"] + assert restored.handler_seen_seq == seen + assert restored.token_sequence == restored.seq_counter == 0 + paginator.advance_handler_seen(ack) + assert restored.updated_operation_ids == ["newer"] + + def test_function_arn_is_qualified_with_the_executed_version(): execution = Execution.new(_make_start_input()) execution.region = "ap-southeast-2" diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index 08edbac70..3f2af3677 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -20,6 +20,7 @@ InvocationStatus, ) from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, ErrorObject, Operation as SvcOperation, OperationAction, @@ -133,6 +134,49 @@ def test_empty_poll_returns_empty_operations_and_advances_token(): assert CheckpointToken.from_str(response.checkpoint_token).token_sequence == 1 +@pytest.mark.parametrize("rejection", ["token", "operation"]) +def test_rejected_checkpoint_retains_undelivered_callback_update(rejection): + executor, store, execution, token = _make_executor_with_started_execution() + execution.operations.append( + SvcOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id="callback-id"), + ) + ) + execution.complete_callback_success("callback-id", b"result") + store.save(execution) + before = ( + execution.token_sequence, + execution.handler_seen_seq, + execution.seq_counter, + ) + assert execution.updated_operation_ids == ["callback"] + with pytest.raises(InvalidParameterValueException): + executor.checkpoint_execution( + execution_arn=execution.durable_execution_arn, + checkpoint_token="invalid" if rejection == "token" else token, + updates=( + [ + OperationUpdate( + operation_id="callback", + operation_type=OperationType.CALLBACK, + action=OperationAction.SUCCEED, + ) + ] + if rejection == "operation" + else [] + ), + ) + assert execution.updated_operation_ids == ["callback"] + assert ( + execution.token_sequence, + execution.handler_seen_seq, + execution.seq_counter, + ) == before + + # endregion # region: Non-empty checkpoint returns only the delta diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py index 89a5618dd..392ad864b 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py @@ -736,6 +736,10 @@ def test_pending_response_is_valid_when_an_operation_completed_after_the_handler mock_execution.consecutive_failed_invocation_attempts = 0 mock_execution.seq_counter = 4 mock_execution.handler_seen_seq = 2 + mock_execution.token_sequence = 0 + mock_execution.operations = [] + mock_execution.updated_operation_ids = [] + mock_execution.operation_last_touched_seq = {} mock_execution.has_pending_operations.return_value = False mock_execution.has_changes_after.side_effect = lambda seq: seq == 4 mock_execution.has_unseen_changes.return_value = True diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py new file mode 100644 index 000000000..42c64a35b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py @@ -0,0 +1,98 @@ +"""Invocation acknowledgments preserve newer checkpoint and completion state.""" + +from datetime import UTC, datetime +from unittest.mock import Mock, patch + +import pytest + +from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, +) +from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, + Operation, + OperationStatus, + OperationType, +) +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + OperationPaginatorState, +) +from aws_durable_execution_sdk_python_testing.executor import Executor +from aws_durable_execution_sdk_python_testing.invoker import InvokeResponse +from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.token import CallbackToken + + +@pytest.mark.parametrize("checkpoint_during_invocation", [False, True]) +def test_input_ack_preserves_newer_completion_and_checkpoint_watermark( + checkpoint_during_invocation, +): + store = InMemoryExecutionStore() + execution = Execution.new( + StartDurableExecutionInput( + account_id="123456789012", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=30, + execution_retention_period_days=1, + invocation_id="invocation", + ) + ) + execution.start() + arn = execution.durable_execution_arn + tokens = {} + for name in ("a", "b", "c"): + tokens[name] = CallbackToken(execution_arn=arn, operation_id=name).to_str() + execution.operations.append( + Operation( + operation_id=name, + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id=tokens[name]), + ) + ) + execution.touch_operation(name) + OperationPaginatorState.pin(execution).advance_handler_seen(execution.seq_counter) + execution.complete_callback_success(tokens["a"], b'"A"') + input_sequence = execution.seq_counter + if checkpoint_during_invocation: + execution.complete_callback_success(tokens["c"], b'"C"') + OperationPaginatorState.pin(execution).advance_handler_seen( + execution.seq_counter + ) + already_seen = execution.handler_seen_seq + execution.complete_callback_success(tokens["b"], b'"B"') + execution.needs_reinvoke = True + store.save(execution) + executor = Executor(store, Mock(), Mock(), Mock()) + try: + with patch.object(executor, "_invoke_execution") as reinvoke: + executor._finish_invocation( # noqa: SLF001 + arn, + InvokeResponse( + invocation_output=DurableExecutionInvocationOutput( + status=InvocationStatus.PENDING + ), + request_id="request", + ), + datetime.now(UTC), + datetime.now(UTC), + input_sequence, + ) + saved = store.load(arn) + assert saved.handler_seen_seq == max(already_seen, input_sequence) + assert saved.updated_operation_ids == ["b"] + assert [ + op.operation_id + for op in OperationPaginatorState.pin(saved).unseen_operations() + ] == ["b"] + reinvoke.assert_called_once_with(arn) + finally: + executor.shutdown() + executor._registry.shutdown() # noqa: SLF001 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py index 746f083d6..9885003a5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py @@ -7,8 +7,11 @@ import pytest from aws_durable_execution_sdk_python.lambda_service import ( + ChainedInvokeDetails as ServiceChainedInvokeDetails, + ContextDetails as ServiceContextDetails, OperationStatus, OperationType, + StepDetails as ServiceStepDetails, ) from aws_durable_execution_sdk_python_testing.exceptions import ( @@ -1461,6 +1464,117 @@ def test_event_error_with_payload_only(): } +@pytest.mark.parametrize( + "payload", + [ + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + ], +) +def test_event_error_roundtrip_preserves_present_payload(payload): + wire = {"Payload": payload, "Truncated": False} + assert EventError.from_dict(wire).to_dict() == wire + + +@pytest.mark.parametrize("truncated", [False, True]) +@pytest.mark.parametrize("payload_field", [{}, {"Payload": None}]) +def test_event_error_absent_or_null_payload_retains_existing_meaning( + payload_field, truncated +): + parsed = EventError.from_dict({**payload_field, "Truncated": truncated}) + assert parsed.payload is None + assert parsed.to_dict() == {"Truncated": truncated} + + +def test_event_error_empty_payload_retains_truncation_flag(): + wire = {"Payload": {}, "Truncated": True} + assert EventError.from_dict(wire).to_dict() == wire + + +@pytest.mark.parametrize( + "event_type,details_attribute,service_details", + [ + ("StepFailed", "step_details", ServiceStepDetails), + ("ContextFailed", "context_details", ServiceContextDetails), + ("ChainedInvokeFailed", "chained_invoke_details", ServiceChainedInvokeDetails), + ], +) +@pytest.mark.parametrize( + "wire_error", + [ + {"Truncated": False}, + {"Payload": None, "Truncated": False}, + {"Payload": {}, "Truncated": False}, + {"Payload": {}, "Truncated": True}, + {"Truncated": True}, + {"Payload": {"ErrorMessage": ""}, "Truncated": False}, + {"Payload": {"ErrorType": ""}, "Truncated": False}, + {"Payload": {"ErrorData": ""}, "Truncated": False}, + {"Payload": {"StackTrace": []}, "Truncated": False}, + { + "Payload": { + "ErrorMessage": "failure", + "ErrorType": "RuntimeError", + "ErrorData": "details", + "StackTrace": [], + }, + "Truncated": False, + }, + ], +) +def test_failed_history_preserves_payload_and_matches_service_error_semantics( + event_type, details_attribute, service_details, wire_error +): + from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + ) + + history = GetDurableExecutionHistoryResponse.from_dict( + { + "Events": [ + { + "EventType": event_type, + "EventId": 1, + "Id": "operation", + "Name": "failed-operation", + "EventTimestamp": TIMESTAMP_2023_01_01_00_01, + event_type + "Details": {"Error": wire_error}, + } + ] + } + ) + history_wire = history.to_dict() + expected_history_error = dict(wire_error) + if expected_history_error.get("Payload") is None: + expected_history_error.pop("Payload", None) + assert history_wire["Events"][0][event_type + "Details"]["Error"] == ( + expected_history_error + ) + assert GetDurableExecutionHistoryResponse.from_dict(history_wire).to_dict() == ( + history_wire + ) + + expected = service_details.from_dict({"Error": wire_error.get("Payload")}).error + (operation,) = events_to_operations(history.events) + assert operation.status is OperationStatus.FAILED + + execution = GetDurableExecutionResponse( + durable_execution_arn="arn:aws:lambda:us-west-2:123456789012:function:test:execution:one", + durable_execution_name="one", + function_arn="arn:aws:lambda:us-west-2:123456789012:function:test", + status="FAILED", + start_timestamp=TIMESTAMP_2023_01_01_00_00, + ) + result = DurableFunctionTestResult.from_execution_history(execution, history) + assert result.get_operation_by_name("failed-operation").error == expected + assert getattr(operation, details_attribute).error == expected + # Reconstructing SDK-facing state must not mutate detailed history payloads. + assert history.to_dict() == history_wire + + # Tests for RetryDetails def test_retry_details_serialization(): """Test RetryDetails from_dict/to_dict round-trip.""" diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py new file mode 100644 index 000000000..65302dbec --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py @@ -0,0 +1,239 @@ +"""The service's retained-operation tree applies only to invocation/state reads.""" + +from datetime import UTC, datetime +from unittest.mock import Mock + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + ContextDetails, + ContextOptions, + Operation, + OperationAction, + OperationStatus, + OperationType, + OperationUpdate, +) +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + OperationPaginatorState, +) +from aws_durable_execution_sdk_python_testing.invoker import ( + InProcessInvoker, + LambdaInvoker, +) +from aws_durable_execution_sdk_python_testing.model import ( + StartDurableExecutionInput, + GetDurableExecutionStateResponse, + CheckpointUpdatedExecutionState, +) +from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( + CheckpointProcessor, +) +from aws_durable_execution_sdk_python_testing.checkpoint.processors.context import ( + ContextProcessor, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +def step(identifier, parent=None, status=OperationStatus.SUCCEEDED): + return Operation( + operation_id=identifier, + parent_id=parent, + operation_type=OperationType.STEP, + status=status, + ) + + +def ctx(identifier, parent=None, status=OperationStatus.SUCCEEDED, replay=None): + return Operation( + operation_id=identifier, + parent_id=parent, + operation_type=OperationType.CONTEXT, + status=status, + context_details=ContextDetails.from_dict( + {} if replay is None else {"ReplayChildren": replay} + ), + ) + + +TREE = [ + step("1-step"), + ctx("2-context", status=OperationStatus.STARTED), + step("2.1-step", "2-context", OperationStatus.PENDING), + step("3-step", "missing"), + ctx("4-context", status=OperationStatus.FAILED), + ctx("4.1-context", "4-context", OperationStatus.FAILED, True), + step("4.1.1-step", "4.1-context", OperationStatus.FAILED), + step("4.1.2-step", "4.1-context", OperationStatus.FAILED), + ctx("5-context", replay=True), + ctx("5.1-context", "5-context", replay=False), + step("5.1.1-step", "5.1-context"), + step("5.1-step", "5-context"), + ctx("map"), + ctx("running-branch", "map", OperationStatus.STARTED), + step("pending-branch-step", "running-branch", OperationStatus.PENDING), + ctx("retained-inside-pruned", "running-branch", replay=True), + step("nested-step", "retained-inside-pruned"), +] +KEPT = [ + "1-step", + "2-context", + "2.1-step", + "3-step", + "4-context", + "5-context", + "5.1-context", + "5.1-step", + "map", +] + + +def execution_with(operations): + start = StartDurableExecutionInput( + account_id="123456789012", + function_name="parity", + function_qualifier="$LATEST", + execution_name="parity", + execution_timeout_seconds=60, + execution_retention_period_days=1, + invocation_id="invocation", + ) + execution = Execution( + "arn:aws:lambda:us-west-2:123456789012:function:parity:$LATEST/durable-execution/parity/id", + start, + list(operations), + ) + for op in operations: + execution.touch_operation(op.operation_id) + execution.updated_operation_ids = ["4.1.1-step", "pending-branch-step"] + return execution + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("reverse", [False, True]) +def test_service_tree_paging_inputs_and_get_state_keep_raw_inspection( + filesystem, reverse, tmp_path +): + operations = list(reversed(TREE)) if reverse else list(TREE) + expected = list(reversed(KEPT)) if reverse else KEPT + execution = execution_with(operations) + store = ( + FileSystemExecutionStore.create(tmp_path / "state") + if filesystem + else InMemoryExecutionStore() + ) + store.save(execution) + restored = store.load(execution.durable_execution_arn) + paginator = OperationPaginatorState.pin(restored) + paged = [] + marker = None + while True: + page, marker = paginator.page(marker, 100000, max_items=2) + paged.extend(op.operation_id for op in page) + if marker is None: + break + assert paged == expected + for invoker in [InProcessInvoker(Mock(), Mock()), LambdaInvoker(Mock())]: + invocation = invoker.create_invocation_input(restored) + assert [ + o.operation_id for o in invocation.initial_execution_state.operations + ] == expected + assert invocation.updated_operation_ids == ["4.1.1-step", "pending-branch-step"] + state = CheckpointProcessor(store, Mock()).get_execution_state( + restored.get_new_checkpoint_token(), "" + ) + assert [o.operation_id for o in state.operations] == expected + # Checkpoint deltas, raw store, inspection, and hidden updated IDs remain full. + assert paginator.unseen_operations() == operations + assert ( + restored.operations == operations + and restored.get_assertable_operations() == operations + ) + assert restored.updated_operation_ids == ["4.1.1-step", "pending-branch-step"] + + +@pytest.mark.parametrize("action", [OperationAction.SUCCEED, OperationAction.FAIL]) +@pytest.mark.parametrize( + "close_options", + [ + None, + ContextOptions(replay_children=True), + ContextOptions(replay_children=False), + ContextOptions(), + ], +) +def test_replay_children_is_recorded_only_by_closing_update(action, close_options): + processor = ContextProcessor() + now = datetime.now(UTC) + notifier = Mock() + start = OperationUpdate( + operation_id="ctx", + operation_type=OperationType.CONTEXT, + action=OperationAction.START, + context_options=ContextOptions(replay_children=True), + ) + opened = processor.process(start, None, notifier, "execution", now) + assert ( + opened.context_details is not None + and opened.context_details.replay_children is False + ) + assert "ReplayChildren" not in opened.to_json_dict()["ContextDetails"] + close = OperationUpdate( + operation_id="ctx", + operation_type=OperationType.CONTEXT, + action=action, + context_options=close_options, + payload="saved", + ) + completed = processor.process(close, opened, notifier, "execution", now) + expected = close_options is not None and close_options.replay_children is True + assert completed.context_details.replay_children is expected + assert ( + GetDurableExecutionStateResponse([completed]).to_dict()["Operations"][0][ + "ContextDetails" + ]["ReplayChildren"] + is expected + ) + assert ( + CheckpointUpdatedExecutionState([completed]).to_dict()["Operations"][0][ + "ContextDetails" + ]["ReplayChildren"] + is expected + ) + assert completed.context_details.result == "saved" + + +@pytest.mark.parametrize( + "operations", [[step("a", "a")], [step("a", "b"), step("b", "a")]] +) +def test_invalid_cycles_terminate_without_dropping_unrelated_operations(operations): + execution = execution_with(operations) + assert OperationPaginatorState.pin(execution).page(None, 100000)[0] == operations + + +def test_lambda_invocation_projects_close_false_and_start_absence(): + from io import BytesIO + import json + + execution = execution_with(TREE) + client = Mock() + client.invoke.return_value = { + "StatusCode": 200, + "Payload": BytesIO(b'{"Status":"SUCCEEDED","Result":"null"}'), + } + invoker = LambdaInvoker(client) + invocation = invoker.create_invocation_input(execution) + invoker.invoke("target", invocation) + payload = json.loads(client.invoke.call_args.kwargs["Payload"]) + operations = {op["Id"]: op for op in payload["InitialExecutionState"]["Operations"]} + assert list(operations) == KEPT + assert "ReplayChildren" not in operations["2-context"]["ContextDetails"] + assert operations["4-context"]["ContextDetails"]["ReplayChildren"] is False + assert operations["5-context"]["ContextDetails"]["ReplayChildren"] is True + assert operations["5.1-context"]["ContextDetails"]["ReplayChildren"] is False + assert payload["UpdatedOperationIds"] == ["4.1.1-step", "pending-branch-step"] diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py index a7d2576ca..cb57f7a10 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py @@ -1116,6 +1116,82 @@ def test_durable_child_context_test_runner_init_with_args( # Tests for DurableFunctionCloudTestRunner and from_execution_history +@pytest.mark.parametrize("decode_wire", [False, True]) +@pytest.mark.parametrize( + "payload", + [ + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + {"ErrorMessage": "failure"}, + ], +) +def test_callback_history_result_preserves_wire_and_canonical_error( + payload, decode_wire +): + from aws_durable_execution_sdk_python.lambda_service import ErrorObject + from aws_durable_execution_sdk_python_testing.model import ( + CallbackFailedDetails, + CallbackStartedDetails, + Event, + EventError, + GetDurableExecutionResponse, + ) + + timestamp = datetime.datetime(2026, 10, 7, tzinfo=datetime.UTC) + caller_error = ErrorObject.from_message("Callback failed") + execution = GetDurableExecutionResponse( + durable_execution_arn="arn:execution", + durable_execution_name="execution", + function_arn="arn:function", + status="FAILED", + start_timestamp=timestamp, + error=caller_error, + ) + history = GetDurableExecutionHistoryResponse( + events=[ + Event( + event_type="CallbackStarted", + event_timestamp=timestamp, + event_id=1, + operation_id="callback", + name="callback", + callback_started_details=CallbackStartedDetails( + callback_id="callback-id" + ), + ), + Event( + event_type="CallbackFailed", + event_timestamp=timestamp, + event_id=2, + operation_id="callback", + name="callback", + callback_failed_details=CallbackFailedDetails( + error=EventError( + payload=ErrorObject.from_dict(payload), truncated=False + ) + ), + ), + ] + ) + if decode_wire: + history = GetDurableExecutionHistoryResponse.from_dict(history.to_dict()) + assert history.events[-1].callback_failed_details.error.to_dict() == { + "Payload": payload, + "Truncated": False, + } + result = DurableFunctionTestResult.from_execution_history(execution, history) + callback = result.get_callback("callback") + assert callback.status is OperationStatus.FAILED + assert (callback.error.to_dict() if callback.error is not None else None) == ( + payload if payload else None + ) + assert result.status is InvocationStatus.FAILED + assert result.error is caller_error + + def test_durable_function_test_result_from_execution_history(): """Test DurableFunctionTestResult.from_execution_history factory method.""" import datetime diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py index ffabc11bf..bc4b1d65e 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py @@ -123,10 +123,16 @@ def handler(_event: Any, context: DurableContext) -> str: assert {info.sub_type for info in plugin.operation_starts} == set(expected) assert {info.sub_type for info in plugin.operation_ends} == set(expected) - # The final replay starts with every earlier operation checkpointed, - # including the execution root, whose service record has no subtype. + # Live hooks above cover every category. The service prunes descendants of + # the completed callback/map/parallel contexts from the replay input; the + # surviving records, including the subtype-less execution root, keep type. replay_operations = plugin.invocation_starts[-1].operations.values() - assert {info.sub_type for info in replay_operations} == {*expected, None} + retained_subtypes = set(expected) - { + OperationSubType.CALLBACK, + OperationSubType.MAP_ITERATION, + OperationSubType.PARALLEL_BRANCH, + } + assert {info.sub_type for info in replay_operations} == {*retained_subtypes, None} all_infos: list[OperationInfo] = [ *plugin.operation_starts, *plugin.operation_ends,