From a65a42ee58b8f4f9f510cceabfce584cbc453a62 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Tue, 25 Aug 2026 23:41:50 +0000 Subject: [PATCH] test: add plugin info-shape conformance handlers --- .../plugin/plugin_attempt_info_shape.py | 101 +++++++++++++++++ .../plugin/plugin_context_info_shape.py | 107 ++++++++++++++++++ .../plugin/plugin_invocation_info_shape.py | 64 +++++++++++ .../plugin/plugin_operation_change_shape.py | 89 +++++++++++++++ .../plugin/plugin_operation_info_shape.py | 85 ++++++++++++++ .../template_plugin.yaml | 75 ++++++++++++ 6 files changed, 521 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py new file mode 100644 index 00000000..1a770cc5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_attempt_info_shape.py @@ -0,0 +1,101 @@ +"""10-21: Attempt hook info field shape. + +The named step fails on its first built-in durable attempt and succeeds on its +second under the SDK's real retry strategy. Each user-function hook dumps only +fields carried by its own info object. +""" + +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, + OperationInfo, + 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: + if execution_arn is not None: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _attempt_record(hook: str, info: OperationInfo) -> dict[str, Any]: + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": hook, + "id": info.operation_id, + "type": info.operation_type.name, + "isReplay": info.is_replayed, + } + if info.name is not None: + record["name"] = info.name + if info.sub_type is not None: + record["subType"] = info.sub_type.value + if info.parent_id is not None: + record["parentId"] = info.parent_id + if info.attempt is not None: + record["attempt"] = info.attempt + if info.start_time is not None: + record["startTimestamp"] = info.start_time.isoformat() + if info.end_time is not None: + record["endTimestamp"] = info.end_time.isoformat() + if info.error is not None and info.error.message is not None: + record["error"] = info.error.message + return record + + +class AttemptInfoShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + 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 info.operation_type.name != "STEP": + return + _emit(_attempt_record("attempt-start", info), self._execution_arn) + + def on_user_function_end(self, info: UserFunctionEndInfo) -> None: + if info.operation_type.name != "STEP": + return + record = _attempt_record("attempt-end", info) + record["outcome"] = info.outcome.name + _emit(record, self._execution_arn) + + +@durable_step +def flaky(step_context: StepContext) -> str: + if step_context.attempt < 2: + raise RuntimeError(f"Attempt {step_context.attempt} failed") + return "ok" + + +@durable_execution(plugins=[AttemptInfoShapePlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + retry_config = RetryStrategyConfig( + max_attempts=3, + initial_delay=Duration.from_seconds(1), + retryable_error_types=[RuntimeError], + ) + result: str = context.step( + flaky(), + name="flaky", + config=StepConfig(create_retry_strategy(retry_config)), + ) + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py new file mode 100644 index 00000000..0debf47b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_context_info_shape.py @@ -0,0 +1,107 @@ +"""10-23: Context-typed hook info field shape. + +A serialised two-branch parallel operation suspends inside branch-a, causing its +function to run again and replay its children. The plugin dumps context operation +and user-function start info directly, including the SDK's children-replay flag. +""" + +import json +from typing import Any + +from aws_durable_execution_sdk_python import BatchResult +from aws_durable_execution_sdk_python.config import ( + Duration, + ParallelBranch, + ParallelConfig, +) +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, + OperationInfo, + OperationStartInfo, + UserFunctionStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + if execution_arn is not None: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _context_record(hook: str, info: OperationInfo) -> dict[str, Any]: + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": hook, + "id": info.operation_id, + "type": info.operation_type.name, + "status": info.status.name, + "isReplay": info.is_replayed, + } + if info.name is not None: + record["name"] = info.name + if info.sub_type is not None: + record["subType"] = info.sub_type.value + if info.parent_id is not None: + record["parentId"] = info.parent_id + if info.start_time is not None: + record["startTimestamp"] = info.start_time.isoformat() + if info.end_time is not None: + record["endTimestamp"] = info.end_time.isoformat() + if info.attempt is not None: + record["attempt"] = info.attempt + return record + + +class ContextInfoShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + 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 != "CONTEXT": + return + _emit(_context_record("operation-start", info), self._execution_arn) + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if info.operation_type.name != "CONTEXT": + return + record = _context_record("fn-start", info) + record["isReplayingChildren"] = info.is_replay_children + _emit(record, self._execution_arn) + + +@durable_step +def inner(_step_context: StepContext) -> str: + return "x" + + +def branch_a(context: DurableContext) -> str: + context.step(inner(), name="inner") + context.wait(Duration.from_seconds(2)) + return "a-done" + + +def branch_b(_context: DurableContext) -> str: + return "b-done" + + +@durable_execution(plugins=[ContextInfoShapePlugin()]) +def handler(_event: Any, context: DurableContext) -> list[str]: + result: BatchResult[str] = context.parallel( + [ + ParallelBranch(func=branch_a, name="branch-a"), + ParallelBranch(func=branch_b, name="branch-b"), + ], + name="ctx", + config=ParallelConfig(max_concurrency=1), + ) + return result.get_results() diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py new file mode 100644 index 00000000..75b53f7b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_info_shape.py @@ -0,0 +1,64 @@ +"""10-19: Invocation hook info field shape. + +A two-second durable wait forces one suspension and replay. Each hook logs a +canonical camelCase dump built only from that hook's own info object; optional +fields are omitted rather than reconstructed. +""" + +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: + if execution_arn is not None: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +class InvocationInfoShapePlugin(DurableInstrumentationPlugin): + def on_invocation_start(self, info: InvocationStartInfo) -> None: + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": "invocation-start", + "isFirstInvocation": info.is_first_invocation, + "operationsCount": len(info.operations), + "updatedOperationsCount": len(info.updated_operations), + } + if info.request_id is not None: + record["requestId"] = info.request_id + if info.execution_start_time is not None: + record["executionStartTimestamp"] = info.execution_start_time.isoformat() + _emit(record, info.execution_arn) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + status = info.status.name + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": "invocation-end", + "isFirstInvocation": info.is_first_invocation, + "operationsCount": len(info.operations), + "status": status, + "terminal": status in ("SUCCEEDED", "FAILED"), + } + if info.request_id is not None: + record["requestId"] = info.request_id + if info.execution_start_time is not None: + record["executionStartTimestamp"] = info.execution_start_time.isoformat() + if info.error is not None and info.error.message is not None: + record["executionError"] = info.error.message + _emit(record, info.execution_arn) + + +@durable_execution(plugins=[InvocationInfoShapePlugin()]) +def handler(event: Any, context: DurableContext) -> str: + context.wait(Duration.from_seconds(2)) + return f"done-{event}" diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py new file mode 100644 index 00000000..bbb2988b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_change_shape.py @@ -0,0 +1,89 @@ +"""10-22: Operation-change hook info field shape. + +A named step succeeds once. For every step in the hook's updated-operation map, +the plugin emits hook-level counts and a canonical dump of that delta item. +""" + +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, + OperationInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + if execution_arn is not None: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _add_operation_fields(record: dict[str, Any], info: OperationInfo) -> None: + record.update( + { + "id": info.operation_id, + "type": info.operation_type.name, + "status": info.status.name, + "isReplay": info.is_replayed, + } + ) + if info.name is not None: + record["name"] = info.name + if info.sub_type is not None: + record["subType"] = info.sub_type.value + if info.parent_id is not None: + record["parentId"] = info.parent_id + if info.start_time is not None: + record["startTimestamp"] = info.start_time.isoformat() + if info.end_time is not None: + record["endTimestamp"] = info.end_time.isoformat() + if info.result is not None: + record["result"] = info.result + if info.error is not None and info.error.message is not None: + record["error"] = info.error.message + if info.attempt is not None: + record["attempt"] = info.attempt + + +class OperationChangeShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + 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 operation_id, operation in info.updated_operations.items(): + if operation.operation_type.name != "STEP": + continue + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": "operation-change", + "updatedOperationsCount": len(info.updated_operations), + "operationsCount": len(info.operations), + "inFullMap": operation_id in info.operations, + } + if info.execution_arn is not None: + record["executionArn"] = info.execution_arn + _add_operation_fields(record, operation) + _emit(record, self._execution_arn) + + +@durable_step +def greet(_step_context: StepContext) -> str: + return "task-a" + + +@durable_execution(plugins=[OperationChangeShapePlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + result: str = context.step(greet(), name="greet") + return result diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py new file mode 100644 index 00000000..07e4a55f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_operation_info_shape.py @@ -0,0 +1,85 @@ +"""10-20: Operation hook info field shape. + +A named step succeeds once. The plugin emits canonical camelCase fields directly +from each operation hook info object and omits only fields that are unset. +""" + +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, + OperationEndInfo, + OperationInfo, + OperationStartInfo, +) + + +def _emit(record: dict[str, Any], execution_arn: str | None) -> None: + if execution_arn is not None: + record = {"durableExecutionArn": execution_arn, **record} + print(json.dumps(record), flush=True) + + +def _operation_record(hook: str, info: OperationInfo) -> dict[str, Any]: + record: dict[str, Any] = { + "plugin": "CONFPLUGIN", + "hook": hook, + "id": info.operation_id, + "type": info.operation_type.name, + "status": info.status.name, + "isReplay": info.is_replayed, + } + if info.name is not None: + record["name"] = info.name + if info.sub_type is not None: + record["subType"] = info.sub_type.value + if info.parent_id is not None: + record["parentId"] = info.parent_id + if info.start_time is not None: + record["startTimestamp"] = info.start_time.isoformat() + if info.end_time is not None: + record["endTimestamp"] = info.end_time.isoformat() + if info.result is not None: + record["result"] = info.result + if info.error is not None and info.error.message is not None: + record["error"] = info.error.message + if info.attempt is not None: + record["attempt"] = info.attempt + return record + + +class OperationInfoShapePlugin(DurableInstrumentationPlugin): + def __init__(self) -> None: + 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(_operation_record("operation-start", info), self._execution_arn) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.operation_type.name != "STEP": + return + _emit(_operation_record("operation-end", info), self._execution_arn) + + +@durable_step +def greet(_step_context: StepContext) -> str: + return "task-a" + + +@durable_execution(plugins=[OperationInfoShapePlugin()]) +def handler(_event: Any, context: DurableContext) -> str: + result: str = context.step(greet(), name="greet") + return result 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 9101b343..9f388d60 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 @@ -298,3 +298,78 @@ Resources: DurableConfig: RetentionPeriodInDays: 7 ExecutionTimeout: 300 + PluginInvocationInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-19"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_invocation_info_shape.handler + Description: Invocation hooks expose the full invocation info shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginOperationInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-20"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_operation_info_shape.handler + Description: Operation hooks expose the full operation info shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginAttemptInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-21"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_attempt_info_shape.handler + Description: Attempt hooks expose the full attempt info shape + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginOperationChangeShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-22"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_operation_change_shape.handler + Description: Operation-change hooks expose full delta operation items + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + PluginContextInfoShape: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["10-23"] + Properties: + CodeUri: lambda-build/ + Handler: plugin.plugin_context_info_shape.handler + Description: Context hooks expose subtype and children-replay fields + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300