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..ef11896a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_faulty_and_healthy.py @@ -0,0 +1,193 @@ +"""10-17: Faulty plugin does not affect a healthy plugin. + +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 +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, + 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) + + +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"}, + info.execution_arn, + ) + raise RuntimeError("faulty invocation-end") + + +class HealthyPlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + # 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: + self._execution_arn = info.execution_arn + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "hook": "invocation-start", + "first": info.is_first_invocation, + }, + 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_user_function_start(self, info: UserFunctionStartInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit( + { + "plugin": "CONFPLUGIN-HEALTHY", + "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, + ) + + 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, + ) + + 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: + 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_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..327763ba --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_suspension_invocation_end.py @@ -0,0 +1,63 @@ +"""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", + # 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, + }, + 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..eef6d3ee --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_wait_replay_flag.py @@ -0,0 +1,101 @@ +"""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"; +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 +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, + "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, + ) + + 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, + "name": info.name, + "status": info.status.name, + }, + self._execution_arn, + ) + + +def wait_short(ctx: DurableContext) -> str: + ctx.wait(Duration.from_seconds(2), name="short") + return "short-done" + + +def wait_long(ctx: DurableContext) -> str: + ctx.wait(Duration.from_seconds(8), name="long") + 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..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 @@ -133,3 +133,168 @@ 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 + PluginWaitReplayFlag: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-18"] + 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