From 906dc97f6092d157486a9426350fd9ab24461fca Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 29 Jul 2026 19:33:13 +0000 Subject: [PATCH 1/4] test: add plugin-suite handlers for 10-8..10-19 --- .../plugin_external_update_on_invoke.py | 71 +++++++ .../plugin/plugin_faulty_and_healthy.py | 117 ++++++++++++ .../plugin/plugin_nested_parent_linkage.py | 72 +++++++ .../plugin/plugin_operation_change.py | 69 +++++++ .../plugin/plugin_parallel_branch_hooks.py | 93 +++++++++ .../handlers/plugin/plugin_replay_flags.py | 105 ++++++++++ .../plugin/plugin_retry_exhaustion.py | 113 +++++++++++ .../handlers/plugin/plugin_slow_hook_flush.py | 86 +++++++++ .../plugin_suspension_invocation_end.py | 60 ++++++ .../plugin/plugin_terminal_payloads.py | 83 ++++++++ .../plugin/plugin_wait_operation_hooks.py | 74 +++++++ .../plugin/plugin_wait_replay_flag.py | 94 +++++++++ .../template_plugin.yaml | 180 ++++++++++++++++++ 13 files changed, 1217 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_external_update_on_invoke.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_nested_parent_linkage.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_replay_flags.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_retry_exhaustion.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_payloads.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_operation_hooks.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_external_update_on_invoke.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_external_update_on_invoke.py new file mode 100644 index 00000000..7abbc401 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_external_update_on_invoke.py @@ -0,0 +1,71 @@ +"""10-9: Plugin observes an externally-updated wait on re-invocation. + +Python does NOT surface externally-updated operations on the invocation-start +info. Instead, when a suspended execution is re-invoked, the SDK routes an +operation that changed while the execution was suspended (its +UpdatedOperationIds) through ``on_operation_update`` -> ``on_operation_end`` +during replay (see ExecutionState.emit_operation_update_hook / +DurableContext replay gating). The requirement explicitly allows emitting the +``updated-on-invoke`` record from whichever hook carries this semantic, so this +handler emits it from ``on_operation_end`` for wait-type operations. + +A 2-second ``context.wait`` always suspends after checkpointing WaitStarted and +completes externally (the timer fires between invocations), so the wait's +terminal end only ever fires on the replay invocation via the external-update +path — never on the first invocation. The ``first`` flag is captured from the +invocation-start hook of the same invocation, so it is naturally False on replay. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class ExternalUpdatePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + self._execution_arn: str | None = None + self._is_first: bool = False + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + self._is_first = info.is_first_invocation + + def on_operation_end(self, info: OperationEndInfo) -> None: + # For a 2s wait, on_operation_end fires exclusively via the + # external-update path on the replay invocation (the wait never + # completes in-process), so this record is the externally-updated signal. + if info.operation_type.name != "WAIT": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "updated-on-invoke", + "op": info.operation_id, + "status": info.status.name, + "first": self._is_first, + }, + self._execution_arn, + ) + + +@durable_execution(plugins=[ExternalUpdatePlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + context.wait(Duration.from_seconds(2)) + return "Wait completed" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py new file mode 100644 index 00000000..4a40ed91 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py @@ -0,0 +1,117 @@ +"""10-17: Faulty plugin does not affect a healthy plugin. + +Two plugins are registered together, in order: a faulty plugin whose every hook +logs a line and then raises, and a healthy plugin that logs normally. The SDK +must isolate each plugin (the faulty plugin's exceptions are swallowed) so the +healthy plugin still receives every hook and the execution result/history are +identical to running without the faulty plugin. No mocking: the isolation +guarantee under test is the SDK's own per-plugin hook-dispatch try/except, which +we exercise by genuinely raising. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + OperationEndInfo, + OperationStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class FaultyPlugin(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-start"}, + info.execution_arn, + ) + raise RuntimeError("faulty invocation-start") + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-end"}, + info.execution_arn, + ) + raise RuntimeError("faulty invocation-end") + + +class HealthyPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "invocation-start", + "first": info.is_first_invocation, + }, + info.execution_arn, + ) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + status = info.status.name if info.status is not None else "NONE" + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "invocation-end", + "status": status, + }, + info.execution_arn, + ) + + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "operation-start", + "op": info.operation_id, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "operation-end", + "op": info.operation_id, + "status": info.status.name, + }, + self._execution_arn, + ) + + +@durable_step +def greet(_step_context: StepContext, name: str) -> str: + return f"Hello, {name}!" + + +@durable_execution(plugins=[FaultyPlugin(), HealthyPlugin()]) +def handler(event: Any, context: DurableContext) -> str: + result: str = context.step(greet(event)) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_nested_parent_linkage.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_nested_parent_linkage.py new file mode 100644 index 00000000..82f6ad8b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_nested_parent_linkage.py @@ -0,0 +1,72 @@ +"""10-11: Plugin parent linkage for nested operations. + +A child context (``run_in_child_context``) wraps a single step. The plugin emits +from the SDK's real ``on_operation_end`` hook for every operation that reaches a +terminal status, reporting the operation id and its parent id (the literal +string NONE when the info carries no parent). The inner step's parent is the +child context's operation id; the child context itself has no parent. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, + durable_with_child_context, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class ParentLinkagePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_end(self, info: OperationEndInfo) -> None: + parent = info.parent_id if info.parent_id else "NONE" + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "parent": parent, + "status": info.status.name, + }, + self._execution_arn, + ) + + +@durable_step +def greet(_step_context: StepContext, name: str) -> str: + return f"Hello, {name}!" + + +@durable_with_child_context +def child_operation(ctx: DurableContext, name: str) -> str: + return ctx.step(greet(name)) + + +@durable_execution(plugins=[ParentLinkagePlugin()]) +def handler(event: Any, context: DurableContext) -> str: + result: str = context.run_in_child_context(child_operation(str(event))) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change.py new file mode 100644 index 00000000..14c36c2a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change.py @@ -0,0 +1,69 @@ +"""10-8: Plugin operation-change hook reports updated operations + full map. + +The plugin implements the SDK's real ``on_operation_change`` hook. When the +step's terminal checkpoint response is merged the SDK fires this hook with the +step in the updated-operations delta and in the full operation map; the plugin +emits one record per step-type updated operation. The execution ARN is captured +from the invocation-start hook and stamped on every record. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationChangeInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class OperationChangePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # The operation-change info also carries the execution ARN, but per the + # requirement we capture it from invocation-start (consistent with the + # other plugin handlers) and reuse it here. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_change(self, info: OperationChangeInfo) -> None: + for op_id, op in info.updated_operations.items(): + if op.operation_type.name != "STEP": + continue + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-change", + "op": op_id, + "status": op.status.name, + "in_full_map": op_id in info.operations, + }, + self._execution_arn, + ) + + +@durable_step +def greet(_step_context: StepContext, name: str) -> str: + return f"Hello, {name}!" + + +@durable_execution(plugins=[OperationChangePlugin()]) +def handler(event: Any, context: DurableContext) -> str: + result: str = context.step(greet(event)) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py new file mode 100644 index 00000000..2475d4be --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_parallel_branch_hooks.py @@ -0,0 +1,93 @@ +"""10-12: Plugin user-function hooks for parallel branches. + +Two parallel branches each return a constant string directly (no inner step), +with max-concurrency 1 so they run sequentially in index order. The plugin emits +from the SDK's real ``on_user_function_start`` / ``on_user_function_end`` hooks, +filtering to parallel-branch sub-type operations. Each record carries the branch +operation id and its parent (the parallel operation id). These hooks run on the +same thread as the branch function, so start-before-end per branch is +deterministic. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import ParallelConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + UserFunctionEndInfo, + UserFunctionStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _is_branch(info: UserFunctionStartInfo | UserFunctionEndInfo) -> bool: + return info.sub_type is not None and info.sub_type.name == "PARALLEL_BRANCH" + + +class ParallelBranchPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # User-function hooks do not carry the execution ARN, so capture it from + # the invocation-start hook and reuse it for later emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if not _is_branch(info): + return + parent = info.parent_id if info.parent_id else "NONE" + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "fn-start", + "op": info.operation_id, + "parent": parent, + }, + self._execution_arn, + ) + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if not _is_branch(info): + return + parent = info.parent_id if info.parent_id else "NONE" + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "fn-end", + "op": info.operation_id, + "parent": parent, + "outcome": info.outcome.name, + }, + self._execution_arn, + ) + + +def branch0(_ctx: DurableContext) -> str: + return "task-1" + + +def branch1(_ctx: DurableContext) -> str: + return "task-2" + + +@durable_execution(plugins=[ParallelBranchPlugin()]) +def handler(_event: Any, context: DurableContext) -> list: + result = context.parallel( + [branch0, branch1], + name="parallel", + config=ParallelConfig(max_concurrency=1), + ) + return result.get_results() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_replay_flags.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_replay_flags.py new file mode 100644 index 00000000..12a29056 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_replay_flags.py @@ -0,0 +1,105 @@ +"""10-13: Plugin replay flag on operation-start. + +Two sequential steps: step A succeeds on its first attempt (terminal); step B +fails once then succeeds via the SDK's real retry strategy. The plugin emits +from the SDK's real ``on_operation_start`` (carrying the ``is_replayed`` flag) +and ``on_operation_end`` hooks, filtering to step-type operations. Terminal step +A is never re-emitted on replay (replay=false only); non-terminal step B is +observed with replay=true on the retry invocation. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration, StepConfig +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + OperationStartInfo, +) +from aws_durable_execution_sdk_python.retries import ( + RetryStrategyConfig, + create_retry_strategy, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class ReplayFlagPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-start", + "op": info.operation_id, + "replay": info.is_replayed, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "status": info.status.name, + }, + self._execution_arn, + ) + + +@durable_step +def step_a(_step_context: StepContext) -> str: + return "a" + + +@durable_step +def step_b(step_context: StepContext) -> str: + # Fail on the first attempt, succeed on the second, driven by the SDK's + # built-in durable attempt counter (1-based). + if step_context.attempt < 2: + msg = f"Attempt {step_context.attempt} failed" + raise RuntimeError(msg) + return "Operation succeeded" + + +@durable_execution(plugins=[ReplayFlagPlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + context.step(step_a()) + retry_config = RetryStrategyConfig( + max_attempts=3, + initial_delay=Duration.from_seconds(1), + retryable_error_types=[RuntimeError], + ) + result: str = context.step( + step_b(), + config=StepConfig(create_retry_strategy(retry_config)), + ) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_retry_exhaustion.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_retry_exhaustion.py new file mode 100644 index 00000000..e97d4528 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_retry_exhaustion.py @@ -0,0 +1,113 @@ +"""10-15: Plugin attempt hooks through retry exhaustion. + +A single step always throws, configured with the SDK's real retry strategy +allowing 2 total attempts (~1s delay). The plugin emits from the SDK's real +``on_user_function_start`` / ``on_user_function_end`` (per-attempt) and +``on_operation_end`` (terminal) hooks, filtering to step-type operations. Every +attempt fails; retries exhaust; the terminal operation-end reports FAILED. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration, StepConfig +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + UserFunctionEndInfo, + UserFunctionStartInfo, +) +from aws_durable_execution_sdk_python.retries import ( + RetryStrategyConfig, + create_retry_strategy, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _is_step(info: Any) -> bool: + return info.operation_type.name == "STEP" + + +class RetryExhaustionPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # User-function / operation hooks do not carry the execution ARN, so + # capture it from the invocation-start hook and reuse it. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if not _is_step(info): + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "attempt-start", + "n": info.attempt, + "op": info.operation_id, + }, + self._execution_arn, + ) + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if not _is_step(info): + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "attempt-end", + "n": info.attempt, + "outcome": info.outcome.name, + "op": info.operation_id, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if not _is_step(info): + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "status": info.status.name, + }, + self._execution_arn, + ) + + +@durable_step +def always_fail(_step_context: StepContext) -> str: + msg = "boom" + raise RuntimeError(msg) + + +@durable_execution(plugins=[RetryExhaustionPlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + retry_config = RetryStrategyConfig( + max_attempts=2, + initial_delay=Duration.from_seconds(1), + retryable_error_types=[RuntimeError], + ) + result: str = context.step( + always_fail(), + config=StepConfig(create_retry_strategy(retry_config)), + ) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py new file mode 100644 index 00000000..4d8ab64b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py @@ -0,0 +1,86 @@ +"""10-18: Plugin hook work is flushed before the invocation returns. + +A single step succeeds. The plugin performs ~1 second of deliberately slow work +inside its ``on_operation_end`` hook and only then logs its record. Python +dispatches operation-end synchronously as part of the checkpoint-response merge +and the plugin executor is drained (ThreadPoolExecutor.shutdown(wait=True)) in +the ``PluginExecutor.run`` context manager before the durable handler wrapper +returns the invocation response, so the ~1s hook work MUST complete and its +record MUST be emitted before the Lambda response is returned. This is the real +SDK completion contract, not a mock — the sleep genuinely blocks the hook. +""" + +import json +import time +from typing import Any + +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + OperationEndInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class SlowHookPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + # Deliberately slow hook work; the SDK must not return the invocation + # response until this completes (otherwise the environment freeze would + # drop the record). + time.sleep(1) + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "slow-operation-end", + "op": info.operation_id, + "status": info.status.name, + }, + self._execution_arn, + ) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + status = info.status.name if info.status is not None else "NONE" + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "invocation-end", + "status": status, + }, + info.execution_arn, + ) + + +@durable_step +def greet(_step_context: StepContext, name: str) -> str: + return f"Hello, {name}!" + + +@durable_execution(plugins=[SlowHookPlugin()]) +def handler(event: Any, context: DurableContext) -> str: + result: str = context.step(greet(event)) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py new file mode 100644 index 00000000..54499aed --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py @@ -0,0 +1,60 @@ +"""10-16: Plugin invocation-end on suspension. + +A single 2-second wait suspends the execution on the first invocation and +completes on replay. The plugin emits from the SDK's real ``on_invocation_start`` +/ ``on_invocation_end`` hooks. It classifies each invocation-end as terminal +(SUCCEEDED/FAILED) or not — the suspending invocation reports a non-terminal +status (e.g. PENDING) and the replay reports the terminal SUCCEEDED. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class SuspensionPlugin(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "invocation-start", + "first": info.is_first_invocation, + }, + info.execution_arn, + ) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + status = info.status.name if info.status is not None else "NONE" + terminal = status in ("SUCCEEDED", "FAILED") + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "invocation-end", + "terminal": terminal, + "status": status, + }, + info.execution_arn, + ) + + +@durable_execution(plugins=[SuspensionPlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + context.wait(Duration.from_seconds(2)) + return "Wait completed" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_payloads.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_payloads.py new file mode 100644 index 00000000..e54049c1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_terminal_payloads.py @@ -0,0 +1,83 @@ +"""10-14: Plugin operation-end payload semantics. + +Two sequential steps: step A returns the constant "task-a" and succeeds; step B +always throws "boom" with no retries so the execution fails. The plugin emits +from the SDK's real ``on_operation_end`` hook, filtering to step-type +operations, reporting the operation's checkpointed serialized result (the +literal NONE when absent) and the error message from the info's error object +(the literal NONE when absent). +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import StepConfig +from aws_durable_execution_sdk_python.context import ( + DurableContext, + StepContext, + durable_step, +) +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, +) +from aws_durable_execution_sdk_python.retries import RetryPresets + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class TerminalPayloadPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + result = info.result if info.result is not None else "NONE" + error = info.error.message if (info.error and info.error.message) else "NONE" + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "status": info.status.name, + "result": result, + "error": error, + }, + self._execution_arn, + ) + + +@durable_step +def step_a(_step_context: StepContext) -> str: + return "task-a" + + +@durable_step +def step_b(_step_context: StepContext) -> str: + msg = "boom" + raise RuntimeError(msg) + + +@durable_execution(plugins=[TerminalPayloadPlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + context.step(step_a()) + result: str = context.step( + step_b(), + config=StepConfig(retry_strategy=RetryPresets.none()), + ) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_operation_hooks.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_operation_hooks.py new file mode 100644 index 00000000..df7e79fa --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_operation_hooks.py @@ -0,0 +1,74 @@ +"""10-10: Plugin operation hooks for a wait operation. + +The plugin emits from the SDK's real ``on_operation_start`` / ``on_operation_end`` +hooks, filtering to wait-type operations only. ``operation_type.name`` is already +the upper-case token (WAIT). The wait's STARTED checkpoint fires operation-start +on the first invocation; the wait's terminal SUCCEEDED status fires operation-end +on the replay invocation (via the external-update path). operation-start may fire +again as a replayed START, so it is asserted at-least-once. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + OperationStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class WaitOperationPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "WAIT": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-start", + "op": info.operation_id, + "type": info.operation_type.name, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "WAIT": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "op": info.operation_id, + "type": info.operation_type.name, + "status": info.status.name, + }, + self._execution_arn, + ) + + +@durable_execution(plugins=[WaitOperationPlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + context.wait(Duration.from_seconds(2)) + return "Wait completed" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py new file mode 100644 index 00000000..783d34bc --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py @@ -0,0 +1,94 @@ +"""10-19: Plugin replay flag for a non-terminal wait. + +The parallel operation "waits" runs two branches concurrently (max-concurrency +2): branch 0 waits 2 seconds then returns "short-done"; branch 1 waits 8 seconds +then returns "long-done". Both waits checkpoint WaitStarted in the first +invocation, so the plugin observes ``operation-start`` with ``replay=false`` +twice. When the 2-second wait completes the execution is re-invoked while the +8-second wait is still NON-terminal, so the plugin observes it via the SDK's real +``on_operation_start`` hook with ``is_replayed`` (``replay``) true. Each wait +reaches a terminal SUCCEEDED end exactly once. + +The plugin filters to wait-type operations and emits from the SDK's real +``on_operation_start`` / ``on_operation_end`` hooks. Operation ids are +deliberately not logged: branch event ids are nondeterministic under concurrency, +and the wait type + replay flag alone identify the behavior under test. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration, ParallelConfig +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import durable_execution +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + OperationStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + # Prefix every plugin record with the execution ARN as a top-level field so + # the conformance runner's CloudWatch JSON filter can scope logs to a single + # execution. Omit the field when the ARN is unset (never invent a value). + if execution_arn: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class WaitReplayFlagPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation hooks do not carry the execution ARN, so capture it from the + # invocation-start hook and reuse it for later operation emissions. + self._execution_arn: str | None = None + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn + + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "WAIT": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-start", + "type": info.operation_type.name, + "replay": info.is_replayed, + }, + self._execution_arn, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "WAIT": + return + _emit( + { + "plugin": "CONFPLUGIN", + "hook": "operation-end", + "type": info.operation_type.name, + "status": info.status.name, + }, + self._execution_arn, + ) + + +def wait_short(ctx: DurableContext) -> str: + ctx.wait(Duration.from_seconds(2)) + return "short-done" + + +def wait_long(ctx: DurableContext) -> str: + ctx.wait(Duration.from_seconds(8)) + return "long-done" + + +@durable_execution(plugins=[WaitReplayFlagPlugin()]) +def handler(_event: Any, context: DurableContext) -> list: + result = context.parallel( + [wait_short, wait_long], + name="waits", + config=ParallelConfig(max_concurrency=2), + ) + return result.get_results() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml index 8577409f..1272523f 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml @@ -133,3 +133,183 @@ Resources: DurableConfig: RetentionPeriodInDays: 7 ExecutionTimeout: 300 + PluginOperationChange: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-8"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_operation_change.handler + Description: Plugin operation-change hook reports updated operations and full map + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginExternalUpdateOnInvoke: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-9"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_external_update_on_invoke.handler + Description: Plugin sees a wait completed externally between invocations + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginWaitOperationHooks: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-10"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_wait_operation_hooks.handler + Description: Plugin operation-start and operation-end hooks fire for wait operations + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginNestedParentLinkage: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-11"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_nested_parent_linkage.handler + Description: Plugin operation hooks report parent id for a step nested in a child context + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginParallelBranchHooks: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-12"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_parallel_branch_hooks.handler + Description: Plugin user-function hooks fire per parallel branch with parent linkage + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginReplayFlags: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-13"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_replay_flags.handler + Description: Plugin replay flag on operation-start for non-terminal operations + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginTerminalPayloads: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-14"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_terminal_payloads.handler + Description: Operation-end info carries the checkpointed result and error + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginRetryExhaustion: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-15"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_retry_exhaustion.handler + Description: Attempt hooks fire for every attempt until exhaustion, then FAILED + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginSuspensionInvocationEnd: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-16"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_suspension_invocation_end.handler + Description: Invocation-end fires non-terminal on suspension and terminal at completion + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginFaultyAndHealthy: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-17"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_faulty_and_healthy.handler + Description: A faulty plugin never blocks another plugin's hooks or the execution + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginSlowHookFlush: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-18"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_slow_hook_flush.handler + Description: Slow work in a plugin hook completes before the invocation returns + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginWaitReplayFlag: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-19"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_wait_replay_flag.handler + Description: Plugin observes a still-pending wait with replay=true on re-invocation + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 From 0377b8c441c0995b39820710dc2a26c649de88a0 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 29 Jul 2026 21:39:01 +0000 Subject: [PATCH 2/4] test: strengthen plugin conformance handlers --- .../plugin/plugin_faulty_and_healthy.py | 108 +++++++++++++++--- .../handlers/plugin/plugin_slow_hook_flush.py | 20 ++-- .../plugin/plugin_wait_replay_flag.py | 32 +++--- 3 files changed, 119 insertions(+), 41 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py index 4a40ed91..ef11896a 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py @@ -1,12 +1,18 @@ """10-17: Faulty plugin does not affect a healthy plugin. -Two plugins are registered together, in order: a faulty plugin whose every hook -logs a line and then raises, and a healthy plugin that logs normally. The SDK -must isolate each plugin (the faulty plugin's exceptions are swallowed) so the -healthy plugin still receives every hook and the execution result/history are -identical to running without the faulty plugin. No mocking: the isolation -guarantee under test is the SDK's own per-plugin hook-dispatch try/except, which -we exercise by genuinely raising. +Two plugins are registered together, in order: a faulty plugin whose every +exercised hook logs a line and then raises, and a healthy plugin that logs +normally. The exercised hooks span the full lifecycle: invocation-start, +operation-start, attempt-start, attempt-end, operation-end, and invocation-end. +In the Python SDK the per-attempt hooks are the real ``on_user_function_start`` / +``on_user_function_end`` callbacks (the latter carries the attempt ``outcome``), +and the operation hooks are ``on_operation_start`` / ``on_operation_end``. + +The SDK must isolate each plugin (the faulty plugin's exceptions are swallowed) +so the healthy plugin still receives every hook and the execution +result/history are identical to running without the faulty plugin. No mocking: +the isolation guarantee under test is the SDK's own per-plugin hook-dispatch +try/except, which we exercise by genuinely raising from every exercised hook. """ import json @@ -24,6 +30,8 @@ InvocationStartInfo, OperationEndInfo, OperationStartInfo, + UserFunctionEndInfo, + UserFunctionStartInfo, ) @@ -37,13 +45,56 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: class FaultyPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # Operation/attempt hooks do not carry the execution ARN; capture it + # from invocation-start (before raising) and reuse it so every faulty + # record is still execution-scoped for CloudWatch retrieval. + self._execution_arn: str | None = None + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self._execution_arn = info.execution_arn _emit( {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-start"}, info.execution_arn, ) raise RuntimeError("faulty invocation-start") + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "operation-start"}, + self._execution_arn, + ) + raise RuntimeError("faulty operation-start") + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "attempt-start"}, + self._execution_arn, + ) + raise RuntimeError("faulty attempt-start") + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "attempt-end"}, + self._execution_arn, + ) + raise RuntimeError("faulty attempt-end") + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + {"plugin": "CONFPLUGIN-FAULTY", "hook": "operation-end"}, + self._execution_arn, + ) + raise RuntimeError("faulty operation-end") + def on_invocation_end(self, info: InvocationEndInfo) -> None: _emit( {"plugin": "CONFPLUGIN-FAULTY", "hook": "invocation-end"}, @@ -54,8 +105,8 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: class HealthyPlugin(DurableInstrumentationPlugin): def __init__(self) -> None: - # Operation hooks do not carry the execution ARN, so capture it from the - # invocation-start hook and reuse it for later operation emissions. + # Operation/attempt hooks do not carry the execution ARN, so capture it + # from the invocation-start hook and reuse it for later emissions. self._execution_arn: str | None = None def on_invocation_start(self, info: InvocationStartInfo) -> None: @@ -69,25 +120,39 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: info.execution_arn, ) - def on_invocation_end(self, info: InvocationEndInfo) -> None: - status = info.status.name if info.status is not None else "NONE" + def on_operation_start(self, info: OperationStartInfo) -> None: + if info.operation_type.name != "STEP": + return _emit( { "plugin": "CONFPLUGIN-HEALTHY", - "hook": "invocation-end", - "status": status, + "hook": "operation-start", + "op": info.operation_id, }, - info.execution_arn, + self._execution_arn, ) - def on_operation_start(self, info: OperationStartInfo) -> None: + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: if info.operation_type.name != "STEP": return _emit( { "plugin": "CONFPLUGIN-HEALTHY", - "hook": "operation-start", + "hook": "attempt-start", + "op": info.operation_id, + }, + self._execution_arn, + ) + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "attempt-end", "op": info.operation_id, + "outcome": info.outcome.name, }, self._execution_arn, ) @@ -105,6 +170,17 @@ def on_operation_end(self, info: OperationEndInfo) -> None: self._execution_arn, ) + def on_invocation_end(self, info: InvocationEndInfo) -> None: + status = info.status.name if info.status is not None else "NONE" + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "invocation-end", + "status": status, + }, + info.execution_arn, + ) + @durable_step def greet(_step_context: StepContext, name: str) -> str: diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py index 4d8ab64b..f2ef0020 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py @@ -1,13 +1,11 @@ -"""10-18: Plugin hook work is flushed before the invocation returns. +"""10-18: Slow plugin hook work is reliably delivered. A single step succeeds. The plugin performs ~1 second of deliberately slow work -inside its ``on_operation_end`` hook and only then logs its record. Python -dispatches operation-end synchronously as part of the checkpoint-response merge -and the plugin executor is drained (ThreadPoolExecutor.shutdown(wait=True)) in -the ``PluginExecutor.run`` context manager before the durable handler wrapper -returns the invocation response, so the ~1s hook work MUST complete and its -record MUST be emitted before the Lambda response is returned. This is the real -SDK completion contract, not a mock — the sleep genuinely blocks the hook. +inside its ``on_operation_end`` hook and only then logs its record using the +SDK's real hook completion mechanism. The conformance assertion checks that the +record is eventually present in the execution logs; it does not compare the +record timestamp with the Lambda response boundary. The sleep is genuine hook +work, not a mock. """ import json @@ -49,9 +47,9 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: if info.operation_type.name != "STEP": return - # Deliberately slow hook work; the SDK must not return the invocation - # response until this completes (otherwise the environment freeze would - # drop the record). + # Deliberately slow hook work. The conformance assertion verifies that + # the resulting record is eventually present in execution-scoped logs; + # it does not assert timing relative to the Lambda response boundary. time.sleep(1) _emit( { diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py index 783d34bc..fc9c1aef 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py @@ -1,18 +1,20 @@ """10-19: Plugin replay flag for a non-terminal wait. The parallel operation "waits" runs two branches concurrently (max-concurrency -2): branch 0 waits 2 seconds then returns "short-done"; branch 1 waits 8 seconds -then returns "long-done". Both waits checkpoint WaitStarted in the first -invocation, so the plugin observes ``operation-start`` with ``replay=false`` -twice. When the 2-second wait completes the execution is re-invoked while the -8-second wait is still NON-terminal, so the plugin observes it via the SDK's real -``on_operation_start`` hook with ``is_replayed`` (``replay``) true. Each wait -reaches a terminal SUCCEEDED end exactly once. - -The plugin filters to wait-type operations and emits from the SDK's real -``on_operation_start`` / ``on_operation_end`` hooks. Operation ids are -deliberately not logged: branch event ids are nondeterministic under concurrency, -and the wait type + replay flag alone identify the behavior under test. +2): branch 0 runs a wait named "short" of 2 seconds then returns "short-done"; +branch 1 runs a wait named "long" of 8 seconds then returns "long-done". Each +wait is given its stable name via the SDK's real operation naming parameter +(``context.wait(duration, name=...)``), so records can be correlated to a +specific wait even though branch event ids are nondeterministic under +concurrency. + +Both waits checkpoint WaitStarted in the first invocation, so the plugin +observes ``operation-start`` with ``replay=false`` once per wait name. When the +2-second wait completes the execution is re-invoked while the 8-second wait is +still NON-terminal, so the plugin observes the named "long" wait via the SDK's +real ``on_operation_start`` hook with ``is_replayed`` (``replay``) true. The +terminal "short" wait must not emit a replayed start. Each wait reaches a +terminal SUCCEEDED end exactly once. """ import json @@ -55,6 +57,7 @@ def on_operation_start(self, info: OperationStartInfo) -> None: "plugin": "CONFPLUGIN", "hook": "operation-start", "type": info.operation_type.name, + "name": info.name, "replay": info.is_replayed, }, self._execution_arn, @@ -68,6 +71,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: "plugin": "CONFPLUGIN", "hook": "operation-end", "type": info.operation_type.name, + "name": info.name, "status": info.status.name, }, self._execution_arn, @@ -75,12 +79,12 @@ def on_operation_end(self, info: OperationEndInfo) -> None: def wait_short(ctx: DurableContext) -> str: - ctx.wait(Duration.from_seconds(2)) + ctx.wait(Duration.from_seconds(2), name="short") return "short-done" def wait_long(ctx: DurableContext) -> str: - ctx.wait(Duration.from_seconds(8)) + ctx.wait(Duration.from_seconds(8), name="long") return "long-done" From 496b58e7c9e820ed32a831b80d13e11536ce4c38 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 29 Jul 2026 22:19:40 +0000 Subject: [PATCH 3/4] test: remove plugin slow-hook conformance case --- .../handlers/plugin/plugin_slow_hook_flush.py | 84 ------------------- .../plugin/plugin_wait_replay_flag.py | 2 +- .../template_plugin.yaml | 17 +--- 3 files changed, 2 insertions(+), 101 deletions(-) delete mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py deleted file mode 100644 index f2ef0020..00000000 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_slow_hook_flush.py +++ /dev/null @@ -1,84 +0,0 @@ -"""10-18: Slow plugin hook work is reliably delivered. - -A single step succeeds. The plugin performs ~1 second of deliberately slow work -inside its ``on_operation_end`` hook and only then logs its record using the -SDK's real hook completion mechanism. The conformance assertion checks that the -record is eventually present in the execution logs; it does not compare the -record timestamp with the Lambda response boundary. The sleep is genuine hook -work, not a mock. -""" - -import json -import time -from typing import Any - -from aws_durable_execution_sdk_python.context import ( - DurableContext, - StepContext, - durable_step, -) -from aws_durable_execution_sdk_python.execution import durable_execution -from aws_durable_execution_sdk_python.plugin import ( - DurableInstrumentationPlugin, - InvocationEndInfo, - InvocationStartInfo, - OperationEndInfo, -) - - -def _emit(record: dict[str, Any], execution_arn: str | None) -> None: - # Prefix every plugin record with the execution ARN as a top-level field so - # the conformance runner's CloudWatch JSON filter can scope logs to a single - # execution. Omit the field when the ARN is unset (never invent a value). - if execution_arn: - record = {"durableExecutionArn": execution_arn, **record} - print(json.dumps(record), flush=True) - - -class SlowHookPlugin(DurableInstrumentationPlugin): - def __init__(self) -> None: - # Operation hooks do not carry the execution ARN, so capture it from the - # invocation-start hook and reuse it for later operation emissions. - self._execution_arn: str | None = None - - def on_invocation_start(self, info: InvocationStartInfo) -> None: - self._execution_arn = info.execution_arn - - def on_operation_end(self, info: OperationEndInfo) -> None: - if info.operation_type.name != "STEP": - return - # Deliberately slow hook work. The conformance assertion verifies that - # the resulting record is eventually present in execution-scoped logs; - # it does not assert timing relative to the Lambda response boundary. - time.sleep(1) - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "slow-operation-end", - "op": info.operation_id, - "status": info.status.name, - }, - self._execution_arn, - ) - - def on_invocation_end(self, info: InvocationEndInfo) -> None: - status = info.status.name if info.status is not None else "NONE" - _emit( - { - "plugin": "CONFPLUGIN", - "hook": "invocation-end", - "status": status, - }, - info.execution_arn, - ) - - -@durable_step -def greet(_step_context: StepContext, name: str) -> str: - return f"Hello, {name}!" - - -@durable_execution(plugins=[SlowHookPlugin()]) -def handler(event: Any, context: DurableContext) -> str: - result: str = context.step(greet(event)) - return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py index fc9c1aef..1e67ea06 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py @@ -1,4 +1,4 @@ -"""10-19: Plugin replay flag for a non-terminal wait. +"""10-18: Plugin replay flag for a non-terminal wait. The parallel operation "waits" runs two branches concurrently (max-concurrency 2): branch 0 runs a wait named "short" of 2 seconds then returns "short-done"; diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml index 1272523f..9101b343 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml @@ -283,25 +283,10 @@ Resources: DurableConfig: RetentionPeriodInDays: 7 ExecutionTimeout: 300 - PluginSlowHookFlush: - Type: AWS::Serverless::Function - TestingMetadata: - TestDescription: ["10-18"] - Properties: - CodeUri: lambda-build/ - Handler: plugin.plugin_slow_hook_flush.handler - Description: Slow work in a plugin hook completes before the invocation returns - Role: - Fn::GetAtt: - - DurableFunctionRole - - Arn - DurableConfig: - RetentionPeriodInDays: 7 - ExecutionTimeout: 300 PluginWaitReplayFlag: Type: AWS::Serverless::Function TestingMetadata: - TestDescription: ["10-19"] + TestDescription: ["10-18"] Properties: CodeUri: lambda-build/ Handler: plugin.plugin_wait_replay_flag.handler From fef92735af30416b764a4058d3ddd61617dabae9 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Fri, 31 Jul 2026 18:48:04 +0000 Subject: [PATCH 4/4] test: tighten invocation and replay correlation --- .../handlers/plugin/plugin_suspension_invocation_end.py | 3 +++ .../handlers/plugin/plugin_wait_replay_flag.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py index 54499aed..327763ba 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py @@ -47,6 +47,9 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: { "plugin": "CONFPLUGIN", "hook": "invocation-end", + # Same-invocation flag from the SDK's own end-hook info: ties + # this record to the invocation that emitted it. + "first": info.is_first_invocation, "terminal": terminal, "status": status, }, diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py index 1e67ea06..eef6d3ee 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py @@ -59,6 +59,9 @@ def on_operation_start(self, info: OperationStartInfo) -> None: "type": info.operation_type.name, "name": info.name, "replay": info.is_replayed, + # Non-terminal at hook time, from the hook info's own operation + # state (no end timestamp yet) — no cross-invocation state. + "pending": info.end_time is None, }, self._execution_arn, )