diff --git a/.github/scripts/install_otel_test_wheels.py b/.github/scripts/install_otel_test_wheels.py index 938996aca..d4f136867 100644 --- a/.github/scripts/install_otel_test_wheels.py +++ b/.github/scripts/install_otel_test_wheels.py @@ -16,6 +16,7 @@ ROOT = Path(__file__).resolve().parents[2] CORE = "aws-durable-execution-sdk-python" OTEL = CORE + "-otel" +TESTING = CORE + "-testing" def built_wheel(package: str) -> Path: @@ -72,7 +73,9 @@ def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--legacy-plugin", action="store_true") args = parser.parse_args() - packages = [CORE] if args.legacy_plugin else [CORE, OTEL] + # Both lanes use the built local runner, including its real callback-update + # producer. The legacy lane keeps the actual published OTel plugin unchanged. + packages = [CORE, TESTING] if args.legacy_plugin else [CORE, OTEL, TESTING] wheels = [built_wheel(package) for package in packages] subprocess.run( [ diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dfc01a16e..016bce08c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,9 @@ on: branches: [ main ] pull_request: - branches: [ main ] + branches: + - main + - fix/python-testing-replay-state-20261021 jobs: lint-commits: diff --git a/.github/workflows/cloud-tests.yml b/.github/workflows/cloud-tests.yml index 91e9f0a27..ac62f3f57 100644 --- a/.github/workflows/cloud-tests.yml +++ b/.github/workflows/cloud-tests.yml @@ -2,7 +2,9 @@ name: Cloud tests on: pull_request: - branches: ["main"] + branches: + - main + - fix/python-testing-replay-state-20261021 workflow_dispatch: inputs: lambda_endpoint: diff --git a/.github/workflows/conformance-tests.yml b/.github/workflows/conformance-tests.yml index b3d70c205..54a0beda4 100644 --- a/.github/workflows/conformance-tests.yml +++ b/.github/workflows/conformance-tests.yml @@ -8,7 +8,9 @@ on: push: branches: ["main"] pull_request: - branches: ["main"] + branches: + - main + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-conformance-tests/**" diff --git a/.github/workflows/js-examples.yml b/.github/workflows/js-examples.yml index 1b911c7bb..ecb01bef6 100644 --- a/.github/workflows/js-examples.yml +++ b/.github/workflows/js-examples.yml @@ -20,7 +20,9 @@ on: - ".github/workflows/js-examples.yml" - "pyproject.toml" pull_request: - branches: ["main"] + branches: + - main + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python-testing/**" - ".github/scripts/js_examples/**" diff --git a/.github/workflows/opentelemetry-conformance-tests.yml b/.github/workflows/opentelemetry-conformance-tests.yml index aed97f6e8..384af266a 100644 --- a/.github/workflows/opentelemetry-conformance-tests.yml +++ b/.github/workflows/opentelemetry-conformance-tests.yml @@ -8,7 +8,9 @@ name: OpenTelemetry Conformance Tests on: pull_request: - branches: ["main"] + branches: + - main + - fix/python-testing-replay-state-20261021 paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-otel/**" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3acdddf12..8aab71b02 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -72,7 +72,7 @@ hatch run dev-examples:test # run examples tests only ### Installed package compatibility testing -Build the core and OTel distributions with `hatch build` in each package, then run +Build the core, OTel and testing-library distributions with `hatch build` in each package, then run these commands from the repository root: ```bash diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 58b21e793..fc31dbefc 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -170,6 +170,31 @@ flow, pagination, and a map of the code. ## Documentation +### Operation state on replay + +Invocation input and `GetDurableExecutionState` follow the service's retained +operation view. An operation is omitted when any ancestor is a `CONTEXT` that +has `SUCCEEDED` or `FAILED` without `ReplayChildren: true`. The walk includes +running intermediate contexts; a nested context's own retention flag cannot +override a completed ancestor that prunes it. Roots and operations whose parent +is absent are kept, and ordering does not change the rule. + +`ContextOptions.ReplayChildren` on START is ignored. A closing update records +true only when it explicitly supplies true; otherwise it records false. Testing +service responses include that closing false, while Core's public model and +generic serialization contract remain unchanged. + +Checkpoint responses still contain all updated operations. `UpdatedOperationIds` +can include an ID whose operation record is hidden from invocation input and +GetState. Raw stores, history, results and nested operation inspection retain the +complete execution. An updated ID alone does not restore missing metadata for a +plugin. A real large child result retains its children through ReplayChildren +when a test needs that public replay behavior. + +This corrects local service parity; it does not fix duplicate execution-view +spans for virtual FLAT map iterations while their map is still running. + + ### Error Handling The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see: diff --git a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md index 0a33e4ec1..4780b5a97 100644 --- a/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md +++ b/packages/aws-durable-execution-sdk-python-testing/docs/architecture.md @@ -234,10 +234,18 @@ list in a checkpoint response. The delta for the next response is the operations whose `operation_last_touched_seq` is greater than `handler_seen_seq`. -The watermark advances only when a checkpoint response is built. It never -advances when invocation input is built. That is what makes retries -correct. A failed invocation never moves the watermark. So the retry -delivers the same state again. +The watermark advances when a checkpoint response is built or an invocation +response passes validation, including a `PENDING` response. Building invocation +input does not acknowledge it: input updates remain pending until delivery is +confirmed. A failed call or rejected response does not consume that input, so a +retry can receive it again. + +The invocation acknowledgment uses the sequence captured when its input was +built, on the execution's worker. It never moves a higher checkpoint watermark +backward or consumes operations that changed after the captured snapshot. This +also prevents an input-only callback resume from receiving the same terminal +update again at a later checkpoint. The acknowledgment concerns runner state +delivery, not telemetry export. ``` seq_counter handler_seen_seq delta next checkpoint diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py index 42ea9333b..f1b1d6da6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py @@ -21,6 +21,9 @@ CheckpointRequestDispatcher, ) from aws_durable_execution_sdk_python_testing.clock import RealClock +from aws_durable_execution_sdk_python_testing.operation_state import ( + prune_replay_children, +) from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) @@ -153,7 +156,7 @@ def get_execution_state( token = CheckpointToken.from_str(checkpoint_token) execution = self._store.load(token.execution_arn) return StateOutput( - operations=execution.get_navigable_operations(), + operations=prune_replay_children(execution.get_navigable_operations()), next_marker=None, ) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py index d2eaa4ec4..9ea3cb256 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py @@ -100,9 +100,11 @@ def _create_context_details(self, update: OperationUpdate) -> ContextDetails | N ContextDetails( result=update.payload, error=update.error, - replay_children=update.context_options.replay_children - if update.context_options - else False, + replay_children=( + update.action in (OperationAction.SUCCEED, OperationAction.FAIL) + and update.context_options is not None + and update.context_options.replay_children is True + ), ) if update.operation_type == OperationType.CONTEXT else None diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py index 5190b9b69..9da824e1a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/execution.py @@ -24,6 +24,9 @@ ) from aws_durable_execution_sdk_python_testing.clock import real_now +from aws_durable_execution_sdk_python_testing.operation_state import ( + prune_replay_children, +) from aws_durable_execution_sdk_python_testing.exceptions import ( IllegalStateException, InvalidParameterValueException, @@ -479,15 +482,11 @@ def record_invocation_completion( ) def mark_state_delivered(self) -> None: - """Reset the list of changed operations once an invocation's input - carries them. - - The service reports, on each invocation, the operations that - changed after the last state the handler observed. The runner - delivers state in the invocation input. So the list is reset when - that input is built, not when the invocation completes: an - operation that completes while the handler is still running is - reported on the next invocation. + """Explicitly clear all pending operation IDs after their delivery. + + The live executor uses versioned ``advance_handler_seen`` acknowledgments + so failed invocations retain their input updates and concurrent changes + newer than the acknowledged input snapshot remain pending. """ self.updated_operation_ids = [] @@ -692,6 +691,7 @@ def complete_callback_success( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_callback_failure( @@ -708,8 +708,11 @@ def complete_callback_failure( self.touch_operation(operation.operation_id) updated_callback_details = None if operation.callback_details: + # Match CallbackDetails.from_dict without depending on a store + # serialization round trip: an empty wire Error has no details. updated_callback_details = replace( - operation.callback_details, error=error + operation.callback_details, + error=error if error is not None and error.to_dict() else None, ) self.operations[index] = replace( @@ -718,6 +721,7 @@ def complete_callback_failure( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_callback_timeout( @@ -744,6 +748,7 @@ def complete_callback_timeout( end_timestamp=now if now is not None else real_now(), callback_details=updated_callback_details, ) + self._record_updated_operation(operation.operation_id) return self.operations[index] def complete_chained_invoke( @@ -897,10 +902,9 @@ def page( ``max_size_bytes`` and, when given, ``max_items``. Second element of the tuple is a marker for the next page, or ``None`` when the page fits everything.""" - start_idx = self._resolve_marker(marker) - return self._walk_page( - self.snapshot_operations, start_idx, max_size_bytes, max_items - ) + operations = prune_replay_children(self.snapshot_operations) + start_idx = self._resolve_marker(marker, len(operations)) + return self._walk_page(operations, start_idx, max_size_bytes, max_items) def unseen_operations(self) -> list[Operation]: """Operations whose ``operation_last_touched_seq`` is strictly @@ -920,11 +924,20 @@ def unseen_operations(self) -> list[Operation]: ] def advance_handler_seen(self, seq: int) -> None: - """Advance the wrapped execution's ``handler_seen_seq`` to - ``seq`` if that represents forward progress. Monotonic: - smaller or equal values are ignored.""" + """Acknowledge delivered updates without retreating the watermark. + + Equal acknowledgments also consume old-format, unversioned update IDs. + Only versions covered by this delivery may be removed. + """ if seq > self.execution.handler_seen_seq: self.execution.handler_seen_seq = seq + # A checkpoint response or completed invocation delivered this state. + # Retain changes newer than that acknowledged snapshot. + self.execution.updated_operation_ids = [ + operation_id + for operation_id in self.execution.updated_operation_ids + if self.execution.operation_last_touched_seq.get(operation_id, 0) > seq + ] # --- internals ------------------------------------------------- @@ -953,14 +966,21 @@ def _size_for(self, op: Operation) -> int: # floor at 1 to guarantee pagination always advances. return max(self.execution.operation_size_bytes.get(op.operation_id, 0), 1) - def _resolve_marker(self, marker: str | None) -> int: + def _resolve_marker( + self, marker: str | None, operation_count: int | None = None + ) -> int: if marker is None: return 0 seq, idx = self._decode_marker(marker) if seq != self.pinned_token_sequence: msg = "Invalid marker" raise InvalidParameterValueException(msg) - if idx < 0 or idx > len(self.snapshot_operations): + count = ( + len(self.snapshot_operations) + if operation_count is None + else operation_count + ) + if idx < 0 or idx > count: msg = "Invalid marker" raise InvalidParameterValueException(msg) return idx diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py index 4c3a0a71f..1092291db 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/executor.py @@ -1551,7 +1551,8 @@ def _begin_invocation( self._set_invocation_gate(execution_arn, InvocationState.INVOKING) execution.begin_new_invocation() invocation_input = self._invoker.create_invocation_input(execution=execution) - execution.mark_state_delivered() + # Keep input updates until a valid response acknowledges this snapshot. + # A failed invocation must not consume their delivery before a retry. self._store.save(execution) return execution, invocation_input, execution.seq_counter @@ -1618,6 +1619,13 @@ def _finish_invocation( # A clean invocation resets the retry counter. reset_target = self._store.load(execution_arn) reset_target.consecutive_failed_invocation_attempts = 0 + if invocation_seq is not None: + # A valid response acknowledges the input snapshot, even when this + # invocation suspended without another checkpoint. Newer changes + # must remain available for the next invocation or checkpoint. + OperationPaginatorState.pin(reset_target).advance_handler_seen( + invocation_seq + ) self._store.save(reset_target) # Release the gate: terminal -> COMPLETED, otherwise PRE_INVOKE diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py index 4992212f4..aec058fe5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/invoker.py @@ -27,6 +27,9 @@ ResourceNotFoundException, ) from aws_durable_execution_sdk_python_testing.execution import OperationPaginatorState +from aws_durable_execution_sdk_python_testing.operation_state import ( + operation_state_to_dict, +) from aws_durable_execution_sdk_python_testing.model import ( LambdaContext, executed_version, @@ -511,10 +514,15 @@ def invoke( input.durable_execution_arn, endpoint_url, region_name ) + invocation_payload = input.to_json_dict() + invocation_payload["InitialExecutionState"]["Operations"] = [ + operation_state_to_dict(operation, json_timestamps=True) + for operation in input.initial_execution_state.operations + ] invoke_kwargs: dict[str, Any] = { "FunctionName": function_name, "InvocationType": "RequestResponse", # Synchronous invocation - "Payload": json.dumps(input.to_json_dict()), + "Payload": json.dumps(invocation_payload), } if tenant_id is not None: invoke_kwargs["TenantId"] = tenant_id diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 22da52618..0dbc59e5c 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -1,6 +1,9 @@ """Model classes for the web API.""" from __future__ import annotations +from aws_durable_execution_sdk_python_testing.operation_state import ( + operation_state_to_dict, +) import datetime @@ -567,7 +570,7 @@ def from_dict(cls, data: dict) -> GetDurableExecutionStateResponse: def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { - "Operations": [op.to_dict() for op in self.operations] + "Operations": [operation_state_to_dict(op) for op in self.operations] } if self.next_marker is not None: result["NextMarker"] = self.next_marker @@ -675,7 +678,8 @@ class EventError: @classmethod def from_dict(cls, data: dict) -> EventError: payload = None - if payload_data := data.get("Payload"): + payload_data = data.get("Payload") + if payload_data is not None: payload = ErrorObject.from_dict(payload_data) return cls( @@ -2249,6 +2253,14 @@ def create_callback_event_failed(cls, context: EventCreationContext) -> Event: if callback_details else None ) + if ( + context.include_execution_data + and callback_details is not None + and callback_details.error is None + ): + # Detailed service history retains an empty Error.Payload object. + # This projection must not turn the SDK-facing absent error into one. + event_error = EventError(payload=ErrorObject.from_dict({}), truncated=False) return cls( event_type=EventType.CALLBACK_FAILED.value, event_timestamp=context.end_timestamp, @@ -2606,6 +2618,11 @@ class HistoryEventTypeConfig: } +def _operation_error(error: ErrorObject | None) -> ErrorObject | None: + # History retains present Payload={}, while SDK state treats Error={} as absent. + return error if error is not None and error.to_dict() else None + + def events_to_operations(events: list[Event]) -> list[Operation]: """Convert a list of history events into operations. @@ -2750,7 +2767,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: callback_details=CallbackDetails( callback_id=callback_id, result=result, - error=error, + error=_operation_error(error), ), ) @@ -2791,7 +2808,7 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, step_details=StepDetails( result=result_val, - error=error_val, + error=_operation_error(error_val), attempt=attempt, next_attempt_timestamp=next_attempt_ts, ), @@ -2824,7 +2841,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, context_details=ContextDetails( result=None, - error=event.context_failed_details.error.payload, + error=_operation_error( + event.context_failed_details.error.payload + ), ), ) @@ -2849,7 +2868,9 @@ def events_to_operations(events: list[Event]) -> list[Operation]: operation, chained_invoke_details=ChainedInvokeDetails( result=None, - error=event.chained_invoke_failed_details.error.payload, + error=_operation_error( + event.chained_invoke_failed_details.error.payload + ), ), ) @@ -3145,7 +3166,7 @@ def from_dict(cls, data: dict) -> CheckpointUpdatedExecutionState: def to_dict(self) -> dict[str, Any]: result: dict[str, Any] = { - "Operations": [op.to_dict() for op in self.operations] + "Operations": [operation_state_to_dict(op) for op in self.operations] } if self.next_marker is not None: result["NextMarker"] = self.next_marker diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py new file mode 100644 index 000000000..850eea00b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/operation_state.py @@ -0,0 +1,64 @@ +"""Service views of retained operation state; raw execution storage stays complete.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ( + Operation, + OperationStatus, + OperationType, +) + + +_CLOSED_CONTEXT = (OperationStatus.SUCCEEDED, OperationStatus.FAILED) + + +def prune_replay_children(operations: Sequence[Operation]) -> list[Operation]: + """Omit a descendant when any known completed context prunes its children.""" + by_id = {operation.operation_id: operation for operation in operations} + visible = [] + for operation in operations: + parent_id = operation.parent_id + visited = {operation.operation_id} + hidden = False + while parent_id is not None and parent_id not in visited: + parent = by_id.get(parent_id) + if parent is None: + break + if ( + parent.operation_type is OperationType.CONTEXT + and parent.status in _CLOSED_CONTEXT + and not ( + parent.context_details + and parent.context_details.replay_children is True + ) + ): + hidden = True + break + visited.add(parent_id) + parent_id = parent.parent_id + if not hidden: + visible.append(operation) + return visible + + +def operation_state_to_dict( + operation: Operation, *, json_timestamps: bool = False +) -> dict[str, Any]: + """Project service ReplayChildren presence without changing Core's model API.""" + data = dict(operation.to_json_dict() if json_timestamps else operation.to_dict()) + if ( + operation.operation_type is OperationType.CONTEXT + and operation.context_details is not None + ): + details = dict(data.get("ContextDetails", {})) + if operation.status is OperationStatus.STARTED: + details.pop("ReplayChildren", None) + elif operation.status in _CLOSED_CONTEXT: + details["ReplayChildren"] = ( + operation.context_details.replay_children is True + ) + data["ContextDetails"] = details + return data diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py new file mode 100644 index 000000000..aded098fa --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/callback_updated_operations_test.py @@ -0,0 +1,284 @@ +"""Callback completions are observed before resumed code without later redelivery.""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from queue import Queue +from threading import Event +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import ( + DurableContext, + StepContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import ( + CallbackConfig, + Duration, + WaitForCallbackConfig, +) +from aws_durable_execution_sdk_python.exceptions import CallbackError +from aws_durable_execution_sdk_python.lambda_service import ErrorObject +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, + UserFunctionStartInfo, +) +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.types import WaitForCallbackContext + +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +class CallbackObserverFactory: + def __init__(self) -> None: + self.invocations: list[InvocationStartInfo] = [] + self.ends: list[tuple[int, OperationEndInfo]] = [] + self.order: list[str] = [] + + def create_plugin(self, info: InvocationStartInfo) -> DurableInstrumentationPlugin: + return CallbackObserver(self) + + +class CallbackObserver(DurableInstrumentationPlugin): + def __init__(self, records: CallbackObserverFactory) -> None: + self.invocations = records.invocations + self.ends = records.ends + self.order = records.order + self.invocation_number = 0 + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + self.invocations.append(info) + self.invocation_number = len(self.invocations) + + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "target": + self.ends.append((self.invocation_number, info)) + self.order.append("target-end") + + def on_user_function_start(self, info: UserFunctionStartInfo) -> None: + if info.name == "observed": + self.order.append("observed-start") + + +def _submit(_callback_id: str, _context: WaitForCallbackContext) -> None: + return None + + +def _suspended_callback(runner: DurableFunctionTestRunner, arn: str, name: str) -> str: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history(arn, include_execution_data=True).events + starts = [ + event + for event in events + if event.event_type == "CallbackStarted" and event.name == name + ] + if starts and any( + event.event_type == "InvocationCompleted" + and event.event_id > starts[0].event_id + for event in events + ): + details = starts[0].callback_started_details + assert details is not None + assert details.callback_id is not None + return details.callback_id + time.sleep(0.01) + raise AssertionError(f"Callback {name} did not suspend") + + +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +@pytest.mark.parametrize("filesystem", [False, True]) +def test_callback_update_is_consumed_before_user_code( + outcome: str, filesystem: bool, tmp_path: Path +) -> None: + observer = CallbackObserverFactory() + marker_calls: list[str] = [] + + @durable_step + def observed(_context: StepContext, value: str) -> str: + marker_calls.append(value) + return value + + def handler(_event: Any, context: DurableContext) -> str: + config = CallbackConfig( + timeout=Duration.from_seconds(1 if outcome == "timeout" else 30), + serdes=JsonSerDes(), + ) + try: + target = context.create_callback(name="target", config=config).result() + except CallbackError: + target = outcome + assert isinstance(target, str) + saved = context.step(observed(target), name="observed") + callback_config = WaitForCallbackConfig(serdes=JsonSerDes()) + one = context.wait_for_callback(_submit, name="one", config=callback_config) + two = context.wait_for_callback(_submit, name="two", config=callback_config) + return "/".join((saved, one, two)) + + wrapped = durable_execution(handler, plugins=[observer]) + store = FileSystemExecutionStore.create(tmp_path) if filesystem else None + with DurableFunctionTestRunner( + handler=wrapped, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + callback_id = _suspended_callback(runner, arn, "target") + if outcome == "success": + runner.send_callback_success( + callback_id, result=json.dumps("target").encode() + ) + elif outcome == "failure": + runner.send_callback_failure( + callback_id, error=ErrorObject.from_message("explicit callback failure") + ) + # The timeout case uses the runner's real scheduled callback deadline. + for name in ["one", "two"]: + callback_id = _suspended_callback(runner, arn, name + " create callback id") + runner.send_callback_success(callback_id, result=json.dumps(name).encode()) + result = runner.wait_for_result(arn, timeout=10) + + expected = "target" if outcome == "success" else outcome + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == expected + "/one/two" + assert marker_calls == [expected] + assert len(observer.invocations) == 4 + # Multiple natural hooks in the delivery invocation are allowed, but an + # acknowledged callback must never be delivered in later invocations. + assert {invocation for invocation, _ in observer.ends} == {2} + terminal = observer.ends[0][1] + assert ( + terminal.status.value + == {"success": "SUCCEEDED", "failure": "FAILED", "timeout": "TIMED_OUT"}[ + outcome + ] + ) + assert all( + end.operation_id == terminal.operation_id + and end.status is terminal.status + and end.is_replayed is False + and end.result == terminal.result + and end.error == terminal.error + for _, end in observer.ends + ) + assert set(observer.invocations[1].updated_operations) == {terminal.operation_id} + assert all( + terminal.operation_id not in invocation.updated_operations + for invocation in observer.invocations[2:] + ) + assert observer.order.count("observed-start") == 1 + assert observer.order.index("target-end") < observer.order.index("observed-start") + if outcome == "failure": + assert terminal.error is not None + assert terminal.error.message == "explicit callback failure" + + +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +@pytest.mark.parametrize("filesystem", [False, True]) +def test_callback_delivered_by_checkpoint_is_not_updated_on_later_resume( + outcome: str, filesystem: bool, tmp_path: Path +) -> None: + observer = CallbackObserverFactory() + submitter_entered = Event() + release_submitter = Event() + submission_calls: list[str] = [] + published_callbacks: Queue[str] = Queue() + + @durable_step + def submit(_context: StepContext, callback_id: str) -> str: + submission_calls.append(callback_id) + published_callbacks.put(callback_id) + submitter_entered.set() + assert release_submitter.wait(10) + return "submitted" + + def handler(_event: Any, context: DurableContext) -> str: + callback = context.create_callback( + name="target", + config=CallbackConfig( + timeout=Duration.from_seconds(1 if outcome == "timeout" else 30), + serdes=JsonSerDes(), + ), + ) + context.step(submit(callback.callback_id), name="submit") + try: + result = callback.result() + except CallbackError: + result = outcome + context.wait(Duration.from_seconds(1), name="first-replay") + context.wait(Duration.from_seconds(1), name="second-replay") + assert isinstance(result, str) + return result + + store = FileSystemExecutionStore.create(tmp_path) if filesystem else None + with DurableFunctionTestRunner( + handler=durable_execution(handler, plugins=[observer]), + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + try: + assert submitter_entered.wait(5) + callback_id = published_callbacks.get(timeout=5) + if outcome == "success": + runner.send_callback_success(callback_id, result=b'"target"') + elif outcome == "failure": + runner.send_callback_failure( + callback_id, error=ErrorObject.from_message("early failure") + ) + expected_event = { + "success": "CallbackSucceeded", + "failure": "CallbackFailed", + "timeout": "CallbackTimedOut", + }[outcome] + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if any(event.event_type == expected_event for event in events): + break + time.sleep(0.01) + else: + raise AssertionError("Callback did not complete during submission") + assert not any( + event.event_type == "InvocationCompleted" for event in events + ) + finally: + release_submitter.set() + result = runner.wait_for_result(arn, timeout=15) + + assert result.status.value == "SUCCEEDED" + assert result.result is not None + assert json.loads(result.result) == ("target" if outcome == "success" else outcome) + assert len(submission_calls) == 1 + assert len(observer.invocations) >= 3 + assert {invocation for invocation, _ in observer.ends} == {1} + terminal = observer.ends[0][1] + target_id = terminal.operation_id + assert all( + end.operation_id == target_id + and end.status is terminal.status + and end.is_replayed is False + and end.result == terminal.result + and end.error == terminal.error + for _, end in observer.ends + ) + assert all( + target_id not in invocation.updated_operations + for invocation in observer.invocations[1:] + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py new file mode 100644 index 000000000..6eaacf5bf --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/invocation_input_delivery_test.py @@ -0,0 +1,154 @@ +"""An input-only callback resume must not redeliver its terminal update later.""" + +import json +import time + +import pytest +from aws_durable_execution_sdk_python import ( + DurableContext, + durable_execution, + durable_step, +) +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python.plugin import InvocationStartInfo +from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin +from aws_durable_execution_sdk_python_testing.execution import Execution +from aws_durable_execution_sdk_python_testing.executor import Executor, InvocationState +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("failed_delivery", [False, True]) +@pytest.mark.parametrize("legacy_format", [False, True]) +def test_callback_input_acknowledgment_survives_checkpoint_free_resume( + filesystem, failed_delivery, legacy_format, tmp_path, monkeypatch +): + store = ( + FileSystemExecutionStore.create(tmp_path / "store") + if filesystem + else InMemoryExecutionStore() + ) + starts = [] + invocation_ends = [] + operation_ends = [] + attempts = [] + marker_calls = [] + legacy_restores = [] + + original_begin = Executor._begin_invocation # noqa: SLF001 + + def begin_with_old_store(self, arn): + # Rehydrate a real completed callback using the persisted format that + # predates sequence accounting. No SDK hook or invocation input is faked. + if ( + legacy_format + and not legacy_restores + and self._invocation_gate(arn) is InvocationState.PRE_INVOKE # noqa: SLF001 + ): + execution = store.load(arn) + if execution.updated_operation_ids: + snapshot = execution.to_json_dict() + for field in ( + "SeqCounter", + "HandlerSeenSeq", + "OperationLastTouchedSeq", + "OperationSizeBytes", + ): + snapshot.pop(field) + restored = Execution.from_json_dict(snapshot) + assert restored.seq_counter == restored.handler_seen_seq == 0 + assert restored.updated_operation_ids == execution.updated_operation_ids + store.save(restored) + legacy_restores.append(snapshot) + return original_begin(self, arn) + + monkeypatch.setattr(Executor, "_begin_invocation", begin_with_old_store) + + class Observe(DurableInstrumentationPlugin): + def on_invocation_start(self, info): + starts.append(info) + + def on_invocation_end(self, info): + invocation_ends.append(info) + + def on_operation_end(self, info): + if info.name in ("a", "b"): + operation_ends.append((len(starts), info.name, info.is_replayed)) + + class ObserveFactory: + def create_plugin( + self, info: InvocationStartInfo + ) -> DurableInstrumentationPlugin: + return Observe() + + @durable_step + def marker(context, value): + marker_calls.append(1) + return value + + @durable_execution(plugins=[ObserveFactory()]) + def handler(event, context: DurableContext): + a = context.create_callback( + name="a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b = context.create_callback( + name="b", config=CallbackConfig(serdes=JsonSerDes()) + ) + result = [a.result(), b.result()] + return context.step(marker(result), name="after-b") + + def entrypoint(event, context): + attempts.append(1) + if failed_delivery and len(attempts) == 2: + raise RuntimeError("invocation failed before the SDK received its input") + return handler(event, context) + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="a", timeout=5) + b_id = runner.wait_for_callback(arn, name="b", timeout=5) + runner.send_callback_success(a_id, result=b'"A"') + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if ( + len(invocation_ends) >= 2 + and sum(event.event_type == "InvocationCompleted" for event in events) + >= 2 + ): + break + time.sleep(0.01) + else: + raise AssertionError("The second SDK invocation did not suspend") + runner.send_callback_success(b_id, result=b'"B"') + result = runner.wait_for_result(arn, timeout=10) + + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == ["A", "B"] + assert [info.status.value for info in invocation_ends] == [ + "PENDING", + "PENDING", + "SUCCEEDED", + ] + # Replay and checkpoint processing can each report the same terminal state. + # Acknowledgment prevents either callback from being redelivered to a later + # invocation; deduplicating exports within an invocation belongs to plugins. + assert set(operation_ends) == {(2, "a", False), (3, "b", False)} + assert marker_calls == [1] + assert len(attempts) == (4 if failed_delivery else 3) + assert len(legacy_restores) == int(legacy_format) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py new file mode 100644 index 000000000..3c31a9412 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/replay_children_service_parity_int_test.py @@ -0,0 +1,256 @@ +"""Real public handlers replay through the service's retained-operation view.""" + +import json +import os +from pathlib import Path + +import pytest +from aws_durable_execution_sdk_python import durable_execution +from aws_durable_execution_sdk_python.config import CallbackConfig +from aws_durable_execution_sdk_python.exceptions import ChildContextError +from aws_durable_execution_sdk_python.serdes import JsonSerDes +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("mode", ["completed", "retained", "nested-retained", "failed"]) +def test_public_resume_uses_service_state_but_inspection_remains_complete( + filesystem, mode, tmp_path +): + store = ( + FileSystemExecutionStore.create(tmp_path / "state") + if filesystem + else InMemoryExecutionStore() + ) + snapshots = [] + child_calls = [] + step_calls = [] + + def save(_step): + step_calls.append(1) + return "saved" + + def child(ctx): + child_calls.append(1) + saved = ctx.step(save, name="saved-child-step") + if mode == "failed": + raise ValueError("stored-child-failure") + return { + "saved": saved, + "large": "x" * (300 * 1024) + if mode in ["retained", "nested-retained"] + else "", + } + + def outer(ctx): + value = ctx.run_in_child_context(child, name="inner-retained") + return {"saved": value["saved"], "large": ""} + + def handler(_event, ctx): + try: + value = ctx.run_in_child_context( + outer if mode == "nested-retained" else child, name="outer" + ) + result = {"saved": value["saved"], "length": len(value["large"])} + except ChildContextError: + assert mode == "failed" + result = {"failed": True} + gate = ctx.create_callback( + name="resume", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert gate.result() == "resume" + return result + + wrapped = durable_execution(handler) + + def entrypoint(event, context): + initial = event.initial_execution_state.operations + state = event.service_client.get_execution_state( + event.durable_execution_arn, event.checkpoint_token, "" + ) + snapshots.append( + { + "input_names": [op.name for op in initial], + "state_names": [op.name for op in state.operations], + "input_ids": [op.operation_id for op in initial], + "state_ids": [op.operation_id for op in state.operations], + "updated_ids": list(event.updated_operation_ids), + } + ) + return wrapped(event, context) + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=20, + ) as runner: + arn = runner.run_async(input="{}") + callback_id = runner.wait_for_callback(arn, name="resume", timeout=5) + runner.send_callback_success(callback_id, result=b'"resume"') + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history(arn, include_execution_data=True).events + assert result.status.value == "SUCCEEDED" + assert json.loads(result.result) == ( + {"failed": True} + if mode == "failed" + else {"saved": "saved", "length": 300 * 1024 if mode == "retained" else 0} + ) + assert ( + len(snapshots) == 2 and snapshots[0]["input_ids"] == snapshots[0]["state_ids"] + ) + assert snapshots[1]["input_ids"] == snapshots[1]["state_ids"] + visible = "saved-child-step" in snapshots[1]["input_names"] + assert visible is (mode == "retained") + if mode == "nested-retained": + assert "inner-retained" not in snapshots[1]["input_names"] + assert "outer" in snapshots[1]["input_names"] + assert step_calls == [1] and len(child_calls) == (2 if mode == "retained" else 1) + raw = store.load(arn) + assert any(op.name == "saved-child-step" for op in raw.operations) + inspected = result.get_operation_by_name("outer") + if mode == "nested-retained": + inspected = inspected.get_operation_by_name("inner-retained") + assert ( + inspected.get_operation_by_name("saved-child-step").name == "saved-child-step" + ) + assert ( + sum( + e.event_type == "StepSucceeded" and e.name == "saved-child-step" + for e in history + ) + == 1 + ) + assert any(e.name == "saved-child-step" for e in history) + if os.environ.get("SERVICE_PRUNING_PROOF_DIR"): + out = Path(os.environ["SERVICE_PRUNING_PROOF_DIR"]) + out.mkdir(parents=True, exist_ok=True) + (out / f"{mode}-fs{filesystem}.json").write_text( + json.dumps( + dict( + mode=mode, + filesystem=filesystem, + snapshots=snapshots, + result=json.loads(result.result), + step_calls=len(step_calls), + child_calls=len(child_calls), + history_names=[e.name for e in history], + ), + indent=2, + ) + + "\n" + ) + + +@pytest.mark.parametrize("filesystem", [False, True]) +def test_hidden_callback_updated_id_does_not_make_its_metadata_visible( + filesystem, tmp_path +): + import time + + store = ( + FileSystemExecutionStore.create(tmp_path / "hidden-state") + if filesystem + else InMemoryExecutionStore() + ) + snapshots = [] + child_calls = [] + saved_calls = [] + + def saved(_step): + saved_calls.append(1) + return "saved" + + def child(ctx): + child_calls.append(1) + callback = ctx.create_callback( + name="hidden-b", config=CallbackConfig(serdes=JsonSerDes()) + ) + assert ctx.step(saved, name="hidden-saved") == "saved" + return callback.callback_id + + def handler(_event, ctx): + a = ctx.create_callback( + name="root-a", config=CallbackConfig(serdes=JsonSerDes()) + ) + b_id = ctx.run_in_child_context(child, name="closed-parent") + return {"a": a.result(), "b_id": b_id} + + wrapped = durable_execution(handler) + + def entrypoint(event, context): + state = event.service_client.get_execution_state( + event.durable_execution_arn, event.checkpoint_token, "" + ) + snapshots.append( + { + "input_ids": [ + op.operation_id for op in event.initial_execution_state.operations + ], + "state_ids": [op.operation_id for op in state.operations], + "updated_ids": list(event.updated_operation_ids), + } + ) + return wrapped(event, context) + + def completed(runner, arn, count): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + events = runner.get_execution_history( + arn, include_execution_data=True + ).events + if sum(e.event_type == "InvocationCompleted" for e in events) >= count: + return events + time.sleep(0.01) + raise AssertionError("invocation completion not observed") + + with DurableFunctionTestRunner( + handler=entrypoint, + store=store, + skip_time=False, + poll_interval=0.01, + execution_timeout=25, + ) as runner: + arn = runner.run_async(input="{}") + a_id = runner.wait_for_callback(arn, name="root-a", timeout=5) + b_id = runner.wait_for_callback(arn, name="hidden-b", timeout=5) + first = completed(runner, arn, 1) + b_operation_id = next( + e.operation_id + for e in first + if e.name == "hidden-b" and e.event_type == "CallbackStarted" + ) + runner.send_callback_success(b_id, result=b'"B"') + completed(runner, arn, 2) + runner.send_callback_success(a_id, result=b'"A"') + result = runner.wait_for_result(arn, timeout=10) + history = runner.get_execution_history(arn, include_execution_data=True).events + assert result.status.value == "SUCCEEDED" and json.loads(result.result) == { + "a": "A", + "b_id": b_id, + } + assert len(snapshots) == 3 and child_calls == [1] and saved_calls == [1] + assert b_operation_id in snapshots[1]["updated_ids"] + assert b_operation_id not in snapshots[1]["input_ids"] + assert b_operation_id not in snapshots[1]["state_ids"] + assert any(op.operation_id == b_operation_id for op in store.load(arn).operations) + assert ( + result.get_operation_by_name("closed-parent") + .get_operation_by_name("hidden-b") + .result + == '"B"' + ) + assert ( + sum( + e.event_type == "CallbackSucceeded" and e.operation_id == b_operation_id + for e in history + ) + == 1 + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py index 578e69434..eb14d3070 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/event_factory_test.py @@ -9,8 +9,10 @@ import pytest from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, ChainedInvokeOptions, ErrorObject, + Operation as ServiceOperation, OperationStatus, OperationType, StepDetails, @@ -855,6 +857,67 @@ def test_create_callback_failed(): assert event.callback_failed_details.error.payload.message == "Callback failed" +@pytest.mark.parametrize("include_data", [False, True]) +@pytest.mark.parametrize("has_error", [False, True]) +def test_callback_failure_history_projection_preserves_sdk_state( + include_data, has_error +): + error = ErrorObject.from_message("details") if has_error else None + operation = ServiceOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.FAILED, + callback_details=CallbackDetails(callback_id="callback-id", error=error), + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + ), + include_execution_data=include_data, + ) + event = Event.create_callback_event(context) + # Preserve main's IncludeExecutionData contract: omitted data must not leak + # an error payload, while detailed history includes even an empty payload. + expected = {"Truncated": not include_data} + if include_data: + expected["Payload"] = error.to_dict() if error else {} + assert event.callback_failed_details.error.to_dict() == expected + assert operation.callback_details.error is error + + +def test_callback_timeout_history_projection_is_unchanged(): + operation = ServiceOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.TIMED_OUT, + callback_details=CallbackDetails(callback_id="callback-id", error=None), + ) + context = EventCreationContext.create( + operation=operation, + event_id=3, + durable_execution_arn="arn:test", + start_input=StartDurableExecutionInput( + account_id="123", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=300, + execution_retention_period_days=7, + ), + include_execution_data=True, + ) + event = Event.create_callback_event(context) + assert event.callback_timed_out_details.error.to_dict() == {"Truncated": False} + + def test_create_callback_timed_out(): operation = create_mock_operation("callback-1", status=OperationStatus.TIMED_OUT) error_obj = ErrorObject.from_message("Callback timed out") diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py index bde5e3d4b..c9457a6dd 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/execution_test.py @@ -3,6 +3,7 @@ import json from dataclasses import replace from datetime import datetime, timezone +from threading import Event, Thread from unittest.mock import patch, Mock import pytest @@ -955,6 +956,38 @@ def test_from_dict_with_none_result(): # region callback +@pytest.mark.parametrize("outcome", ["success", "failure", "timeout"]) +def test_callback_completion_records_only_successful_state_changes(outcome): + """Preserve payloads, token versions and consumed metadata on rejection.""" + operation = Operation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id="callback-id"), + ) + execution = Execution("test-arn", _make_start_input(), [operation]) + complete = getattr(execution, f"complete_callback_{outcome}") + payload = b'"result"' if outcome == "success" else ErrorObject.from_message("error") + token_version = execution.token_sequence + result = complete("callback-id", payload) + + assert execution.updated_operation_ids == ["callback"] + assert execution.token_sequence == token_version + assert execution.seq_counter == 1 + assert result.callback_details.result == ( + '"result"' if outcome == "success" else None + ) + assert result.callback_details.error == (None if outcome == "success" else payload) + restored = Execution.from_json_dict(execution.to_json_dict()) + assert restored.updated_operation_ids == ["callback"] + execution.mark_state_delivered() + with pytest.raises(IllegalStateException, match="not in STARTED state"): + complete("callback-id", payload) + assert execution.updated_operation_ids == [] + assert execution.seq_counter == 1 + assert execution.token_sequence == token_version + + def test_find_callback_operation_not_found(): """Test find_callback_operation raises exception when callback not found.""" execution = Execution("test-arn", Mock(), []) @@ -1690,6 +1723,87 @@ def test_record_invocation_completion_keeps_updated_operation_ids(): assert execution.updated_operation_ids == [] +@pytest.mark.parametrize("complete_before_advance", [False, True]) +def test_checkpoint_consumes_only_updates_covered_by_its_watermark( + complete_before_advance, +): + """A late update survives reads and retries of an older state delivery.""" + execution = Execution( + "test-arn", + _make_start_input(), + [ + Operation( + operation_id=name, + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id=name), + ) + for name in ["delivered", "later"] + ], + ) + execution.complete_callback_success("delivered", b"first") + response = OperationPaginatorState.pin(execution) + ready = Event() + finished = Event() + + def complete_later(): + assert ready.wait(5) + execution.complete_callback_failure("later", ErrorObject.from_message("later")) + finished.set() + + worker = Thread(target=complete_later) + worker.start() + if complete_before_advance: + ready.set() + assert finished.wait(5) + expected = ["delivered", "later"] if complete_before_advance else ["delivered"] + assert execution.updated_operation_ids == expected + response.page(None, max_size_bytes=1024 * 1024) + assert execution.updated_operation_ids == expected + + response.advance_handler_seen(1) + ready.set() + assert finished.wait(5) + worker.join(timeout=5) + assert not worker.is_alive() + assert execution.updated_operation_ids == ["later"] + assert execution.handler_seen_seq == 1 + assert execution.token_sequence == 0 + assert execution.seq_counter == 2 + # An idempotent/older delivery cannot consume the later completion. + response.advance_handler_seen(1) + response.advance_handler_seen(0) + assert execution.updated_operation_ids == ["later"] + OperationPaginatorState.pin(execution).advance_handler_seen(2) + assert execution.updated_operation_ids == [] + + +@pytest.mark.parametrize("seen,ack", [(0, 0), (5, 5), (5, 3)]) +def test_acknowledged_updates_are_pruned_without_watermark_progress(seen, ack): + execution = Execution("test-arn", _make_start_input(), []) + snapshot = execution.to_json_dict() + for field in ( + "SeqCounter", + "HandlerSeenSeq", + "OperationLastTouchedSeq", + "OperationSizeBytes", + ): + snapshot.pop(field) + snapshot["UpdatedOperationIds"] = ["legacy", "newer"] + restored = Execution.from_json_dict(snapshot) + restored.handler_seen_seq = seen + restored.operation_last_touched_seq["newer"] = ack + 1 + paginator = OperationPaginatorState.pin(restored) + + paginator.advance_handler_seen(ack) + + assert restored.updated_operation_ids == ["newer"] + assert restored.handler_seen_seq == seen + assert restored.token_sequence == restored.seq_counter == 0 + paginator.advance_handler_seen(ack) + assert restored.updated_operation_ids == ["newer"] + + def test_function_arn_is_qualified_with_the_executed_version(): execution = Execution.new(_make_start_input()) execution.region = "ap-southeast-2" diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py index 08edbac70..3f2af3677 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_checkpoint_test.py @@ -20,6 +20,7 @@ InvocationStatus, ) from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, ErrorObject, Operation as SvcOperation, OperationAction, @@ -133,6 +134,49 @@ def test_empty_poll_returns_empty_operations_and_advances_token(): assert CheckpointToken.from_str(response.checkpoint_token).token_sequence == 1 +@pytest.mark.parametrize("rejection", ["token", "operation"]) +def test_rejected_checkpoint_retains_undelivered_callback_update(rejection): + executor, store, execution, token = _make_executor_with_started_execution() + execution.operations.append( + SvcOperation( + operation_id="callback", + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id="callback-id"), + ) + ) + execution.complete_callback_success("callback-id", b"result") + store.save(execution) + before = ( + execution.token_sequence, + execution.handler_seen_seq, + execution.seq_counter, + ) + assert execution.updated_operation_ids == ["callback"] + with pytest.raises(InvalidParameterValueException): + executor.checkpoint_execution( + execution_arn=execution.durable_execution_arn, + checkpoint_token="invalid" if rejection == "token" else token, + updates=( + [ + OperationUpdate( + operation_id="callback", + operation_type=OperationType.CALLBACK, + action=OperationAction.SUCCEED, + ) + ] + if rejection == "operation" + else [] + ), + ) + assert execution.updated_operation_ids == ["callback"] + assert ( + execution.token_sequence, + execution.handler_seen_seq, + execution.seq_counter, + ) == before + + # endregion # region: Non-empty checkpoint returns only the delta diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py index 21993649e..33585fa3a 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/executor_test.py @@ -736,6 +736,10 @@ def test_pending_response_is_valid_when_an_operation_completed_after_the_handler mock_execution.consecutive_failed_invocation_attempts = 0 mock_execution.seq_counter = 4 mock_execution.handler_seen_seq = 2 + mock_execution.token_sequence = 0 + mock_execution.operations = [] + mock_execution.updated_operation_ids = [] + mock_execution.operation_last_touched_seq = {} mock_execution.has_pending_operations.return_value = False mock_execution.has_changes_after.side_effect = lambda seq: seq == 4 mock_execution.has_unseen_changes.return_value = True diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py new file mode 100644 index 000000000..42c64a35b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/invocation_input_delivery_test.py @@ -0,0 +1,98 @@ +"""Invocation acknowledgments preserve newer checkpoint and completion state.""" + +from datetime import UTC, datetime +from unittest.mock import Mock, patch + +import pytest + +from aws_durable_execution_sdk_python.execution import ( + DurableExecutionInvocationOutput, + InvocationStatus, +) +from aws_durable_execution_sdk_python.lambda_service import ( + CallbackDetails, + Operation, + OperationStatus, + OperationType, +) +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + OperationPaginatorState, +) +from aws_durable_execution_sdk_python_testing.executor import Executor +from aws_durable_execution_sdk_python_testing.invoker import InvokeResponse +from aws_durable_execution_sdk_python_testing.model import StartDurableExecutionInput +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.token import CallbackToken + + +@pytest.mark.parametrize("checkpoint_during_invocation", [False, True]) +def test_input_ack_preserves_newer_completion_and_checkpoint_watermark( + checkpoint_during_invocation, +): + store = InMemoryExecutionStore() + execution = Execution.new( + StartDurableExecutionInput( + account_id="123456789012", + function_name="test", + function_qualifier="$LATEST", + execution_name="test", + execution_timeout_seconds=30, + execution_retention_period_days=1, + invocation_id="invocation", + ) + ) + execution.start() + arn = execution.durable_execution_arn + tokens = {} + for name in ("a", "b", "c"): + tokens[name] = CallbackToken(execution_arn=arn, operation_id=name).to_str() + execution.operations.append( + Operation( + operation_id=name, + operation_type=OperationType.CALLBACK, + status=OperationStatus.STARTED, + callback_details=CallbackDetails(callback_id=tokens[name]), + ) + ) + execution.touch_operation(name) + OperationPaginatorState.pin(execution).advance_handler_seen(execution.seq_counter) + execution.complete_callback_success(tokens["a"], b'"A"') + input_sequence = execution.seq_counter + if checkpoint_during_invocation: + execution.complete_callback_success(tokens["c"], b'"C"') + OperationPaginatorState.pin(execution).advance_handler_seen( + execution.seq_counter + ) + already_seen = execution.handler_seen_seq + execution.complete_callback_success(tokens["b"], b'"B"') + execution.needs_reinvoke = True + store.save(execution) + executor = Executor(store, Mock(), Mock(), Mock()) + try: + with patch.object(executor, "_invoke_execution") as reinvoke: + executor._finish_invocation( # noqa: SLF001 + arn, + InvokeResponse( + invocation_output=DurableExecutionInvocationOutput( + status=InvocationStatus.PENDING + ), + request_id="request", + ), + datetime.now(UTC), + datetime.now(UTC), + input_sequence, + ) + saved = store.load(arn) + assert saved.handler_seen_seq == max(already_seen, input_sequence) + assert saved.updated_operation_ids == ["b"] + assert [ + op.operation_id + for op in OperationPaginatorState.pin(saved).unseen_operations() + ] == ["b"] + reinvoke.assert_called_once_with(arn) + finally: + executor.shutdown() + executor._registry.shutdown() # noqa: SLF001 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py index c4194a93c..b26dac8d8 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/model_test.py @@ -7,9 +7,12 @@ import pytest from aws_durable_execution_sdk_python.lambda_service import ( + ChainedInvokeDetails as ServiceChainedInvokeDetails, + ContextDetails as ServiceContextDetails, OperationStatus, OperationSubType, OperationType, + StepDetails as ServiceStepDetails, ) from aws_durable_execution_sdk_python_testing.exceptions import ( @@ -1484,6 +1487,117 @@ def test_event_error_with_payload_only(): } +@pytest.mark.parametrize( + "payload", + [ + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + ], +) +def test_event_error_roundtrip_preserves_present_payload(payload): + wire = {"Payload": payload, "Truncated": False} + assert EventError.from_dict(wire).to_dict() == wire + + +@pytest.mark.parametrize("truncated", [False, True]) +@pytest.mark.parametrize("payload_field", [{}, {"Payload": None}]) +def test_event_error_absent_or_null_payload_retains_existing_meaning( + payload_field, truncated +): + parsed = EventError.from_dict({**payload_field, "Truncated": truncated}) + assert parsed.payload is None + assert parsed.to_dict() == {"Truncated": truncated} + + +def test_event_error_empty_payload_retains_truncation_flag(): + wire = {"Payload": {}, "Truncated": True} + assert EventError.from_dict(wire).to_dict() == wire + + +@pytest.mark.parametrize( + "event_type,details_attribute,service_details", + [ + ("StepFailed", "step_details", ServiceStepDetails), + ("ContextFailed", "context_details", ServiceContextDetails), + ("ChainedInvokeFailed", "chained_invoke_details", ServiceChainedInvokeDetails), + ], +) +@pytest.mark.parametrize( + "wire_error", + [ + {"Truncated": False}, + {"Payload": None, "Truncated": False}, + {"Payload": {}, "Truncated": False}, + {"Payload": {}, "Truncated": True}, + {"Truncated": True}, + {"Payload": {"ErrorMessage": ""}, "Truncated": False}, + {"Payload": {"ErrorType": ""}, "Truncated": False}, + {"Payload": {"ErrorData": ""}, "Truncated": False}, + {"Payload": {"StackTrace": []}, "Truncated": False}, + { + "Payload": { + "ErrorMessage": "failure", + "ErrorType": "RuntimeError", + "ErrorData": "details", + "StackTrace": [], + }, + "Truncated": False, + }, + ], +) +def test_failed_history_preserves_payload_and_matches_service_error_semantics( + event_type, details_attribute, service_details, wire_error +): + from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + ) + + history = GetDurableExecutionHistoryResponse.from_dict( + { + "Events": [ + { + "EventType": event_type, + "EventId": 1, + "Id": "operation", + "Name": "failed-operation", + "EventTimestamp": TIMESTAMP_2023_01_01_00_01, + event_type + "Details": {"Error": wire_error}, + } + ] + } + ) + history_wire = history.to_dict() + expected_history_error = dict(wire_error) + if expected_history_error.get("Payload") is None: + expected_history_error.pop("Payload", None) + assert history_wire["Events"][0][event_type + "Details"]["Error"] == ( + expected_history_error + ) + assert GetDurableExecutionHistoryResponse.from_dict(history_wire).to_dict() == ( + history_wire + ) + + expected = service_details.from_dict({"Error": wire_error.get("Payload")}).error + (operation,) = events_to_operations(history.events) + assert operation.status is OperationStatus.FAILED + + execution = GetDurableExecutionResponse( + durable_execution_arn="arn:aws:lambda:us-west-2:123456789012:function:test:execution:one", + durable_execution_name="one", + function_arn="arn:aws:lambda:us-west-2:123456789012:function:test", + status="FAILED", + start_timestamp=TIMESTAMP_2023_01_01_00_00, + ) + result = DurableFunctionTestResult.from_execution_history(execution, history) + assert result.get_operation_by_name("failed-operation").error == expected + assert getattr(operation, details_attribute).error == expected + # Reconstructing SDK-facing state must not mutate detailed history payloads. + assert history.to_dict() == history_wire + + # Tests for RetryDetails def test_retry_details_serialization(): """Test RetryDetails from_dict/to_dict round-trip.""" diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py new file mode 100644 index 000000000..65302dbec --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/replay_children_service_parity_test.py @@ -0,0 +1,239 @@ +"""The service's retained-operation tree applies only to invocation/state reads.""" + +from datetime import UTC, datetime +from unittest.mock import Mock + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + ContextDetails, + ContextOptions, + Operation, + OperationAction, + OperationStatus, + OperationType, + OperationUpdate, +) +from aws_durable_execution_sdk_python_testing.execution import ( + Execution, + OperationPaginatorState, +) +from aws_durable_execution_sdk_python_testing.invoker import ( + InProcessInvoker, + LambdaInvoker, +) +from aws_durable_execution_sdk_python_testing.model import ( + StartDurableExecutionInput, + GetDurableExecutionStateResponse, + CheckpointUpdatedExecutionState, +) +from aws_durable_execution_sdk_python_testing.checkpoint.processor import ( + CheckpointProcessor, +) +from aws_durable_execution_sdk_python_testing.checkpoint.processors.context import ( + ContextProcessor, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) + + +def step(identifier, parent=None, status=OperationStatus.SUCCEEDED): + return Operation( + operation_id=identifier, + parent_id=parent, + operation_type=OperationType.STEP, + status=status, + ) + + +def ctx(identifier, parent=None, status=OperationStatus.SUCCEEDED, replay=None): + return Operation( + operation_id=identifier, + parent_id=parent, + operation_type=OperationType.CONTEXT, + status=status, + context_details=ContextDetails.from_dict( + {} if replay is None else {"ReplayChildren": replay} + ), + ) + + +TREE = [ + step("1-step"), + ctx("2-context", status=OperationStatus.STARTED), + step("2.1-step", "2-context", OperationStatus.PENDING), + step("3-step", "missing"), + ctx("4-context", status=OperationStatus.FAILED), + ctx("4.1-context", "4-context", OperationStatus.FAILED, True), + step("4.1.1-step", "4.1-context", OperationStatus.FAILED), + step("4.1.2-step", "4.1-context", OperationStatus.FAILED), + ctx("5-context", replay=True), + ctx("5.1-context", "5-context", replay=False), + step("5.1.1-step", "5.1-context"), + step("5.1-step", "5-context"), + ctx("map"), + ctx("running-branch", "map", OperationStatus.STARTED), + step("pending-branch-step", "running-branch", OperationStatus.PENDING), + ctx("retained-inside-pruned", "running-branch", replay=True), + step("nested-step", "retained-inside-pruned"), +] +KEPT = [ + "1-step", + "2-context", + "2.1-step", + "3-step", + "4-context", + "5-context", + "5.1-context", + "5.1-step", + "map", +] + + +def execution_with(operations): + start = StartDurableExecutionInput( + account_id="123456789012", + function_name="parity", + function_qualifier="$LATEST", + execution_name="parity", + execution_timeout_seconds=60, + execution_retention_period_days=1, + invocation_id="invocation", + ) + execution = Execution( + "arn:aws:lambda:us-west-2:123456789012:function:parity:$LATEST/durable-execution/parity/id", + start, + list(operations), + ) + for op in operations: + execution.touch_operation(op.operation_id) + execution.updated_operation_ids = ["4.1.1-step", "pending-branch-step"] + return execution + + +@pytest.mark.parametrize("filesystem", [False, True]) +@pytest.mark.parametrize("reverse", [False, True]) +def test_service_tree_paging_inputs_and_get_state_keep_raw_inspection( + filesystem, reverse, tmp_path +): + operations = list(reversed(TREE)) if reverse else list(TREE) + expected = list(reversed(KEPT)) if reverse else KEPT + execution = execution_with(operations) + store = ( + FileSystemExecutionStore.create(tmp_path / "state") + if filesystem + else InMemoryExecutionStore() + ) + store.save(execution) + restored = store.load(execution.durable_execution_arn) + paginator = OperationPaginatorState.pin(restored) + paged = [] + marker = None + while True: + page, marker = paginator.page(marker, 100000, max_items=2) + paged.extend(op.operation_id for op in page) + if marker is None: + break + assert paged == expected + for invoker in [InProcessInvoker(Mock(), Mock()), LambdaInvoker(Mock())]: + invocation = invoker.create_invocation_input(restored) + assert [ + o.operation_id for o in invocation.initial_execution_state.operations + ] == expected + assert invocation.updated_operation_ids == ["4.1.1-step", "pending-branch-step"] + state = CheckpointProcessor(store, Mock()).get_execution_state( + restored.get_new_checkpoint_token(), "" + ) + assert [o.operation_id for o in state.operations] == expected + # Checkpoint deltas, raw store, inspection, and hidden updated IDs remain full. + assert paginator.unseen_operations() == operations + assert ( + restored.operations == operations + and restored.get_assertable_operations() == operations + ) + assert restored.updated_operation_ids == ["4.1.1-step", "pending-branch-step"] + + +@pytest.mark.parametrize("action", [OperationAction.SUCCEED, OperationAction.FAIL]) +@pytest.mark.parametrize( + "close_options", + [ + None, + ContextOptions(replay_children=True), + ContextOptions(replay_children=False), + ContextOptions(), + ], +) +def test_replay_children_is_recorded_only_by_closing_update(action, close_options): + processor = ContextProcessor() + now = datetime.now(UTC) + notifier = Mock() + start = OperationUpdate( + operation_id="ctx", + operation_type=OperationType.CONTEXT, + action=OperationAction.START, + context_options=ContextOptions(replay_children=True), + ) + opened = processor.process(start, None, notifier, "execution", now) + assert ( + opened.context_details is not None + and opened.context_details.replay_children is False + ) + assert "ReplayChildren" not in opened.to_json_dict()["ContextDetails"] + close = OperationUpdate( + operation_id="ctx", + operation_type=OperationType.CONTEXT, + action=action, + context_options=close_options, + payload="saved", + ) + completed = processor.process(close, opened, notifier, "execution", now) + expected = close_options is not None and close_options.replay_children is True + assert completed.context_details.replay_children is expected + assert ( + GetDurableExecutionStateResponse([completed]).to_dict()["Operations"][0][ + "ContextDetails" + ]["ReplayChildren"] + is expected + ) + assert ( + CheckpointUpdatedExecutionState([completed]).to_dict()["Operations"][0][ + "ContextDetails" + ]["ReplayChildren"] + is expected + ) + assert completed.context_details.result == "saved" + + +@pytest.mark.parametrize( + "operations", [[step("a", "a")], [step("a", "b"), step("b", "a")]] +) +def test_invalid_cycles_terminate_without_dropping_unrelated_operations(operations): + execution = execution_with(operations) + assert OperationPaginatorState.pin(execution).page(None, 100000)[0] == operations + + +def test_lambda_invocation_projects_close_false_and_start_absence(): + from io import BytesIO + import json + + execution = execution_with(TREE) + client = Mock() + client.invoke.return_value = { + "StatusCode": 200, + "Payload": BytesIO(b'{"Status":"SUCCEEDED","Result":"null"}'), + } + invoker = LambdaInvoker(client) + invocation = invoker.create_invocation_input(execution) + invoker.invoke("target", invocation) + payload = json.loads(client.invoke.call_args.kwargs["Payload"]) + operations = {op["Id"]: op for op in payload["InitialExecutionState"]["Operations"]} + assert list(operations) == KEPT + assert "ReplayChildren" not in operations["2-context"]["ContextDetails"] + assert operations["4-context"]["ContextDetails"]["ReplayChildren"] is False + assert operations["5-context"]["ContextDetails"]["ReplayChildren"] is True + assert operations["5.1-context"]["ContextDetails"]["ReplayChildren"] is False + assert payload["UpdatedOperationIds"] == ["4.1.1-step", "pending-branch-step"] diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py index a7d2576ca..cb57f7a10 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/runner_test.py @@ -1116,6 +1116,82 @@ def test_durable_child_context_test_runner_init_with_args( # Tests for DurableFunctionCloudTestRunner and from_execution_history +@pytest.mark.parametrize("decode_wire", [False, True]) +@pytest.mark.parametrize( + "payload", + [ + {}, + {"ErrorMessage": ""}, + {"ErrorType": ""}, + {"ErrorData": ""}, + {"StackTrace": []}, + {"ErrorMessage": "failure"}, + ], +) +def test_callback_history_result_preserves_wire_and_canonical_error( + payload, decode_wire +): + from aws_durable_execution_sdk_python.lambda_service import ErrorObject + from aws_durable_execution_sdk_python_testing.model import ( + CallbackFailedDetails, + CallbackStartedDetails, + Event, + EventError, + GetDurableExecutionResponse, + ) + + timestamp = datetime.datetime(2026, 10, 7, tzinfo=datetime.UTC) + caller_error = ErrorObject.from_message("Callback failed") + execution = GetDurableExecutionResponse( + durable_execution_arn="arn:execution", + durable_execution_name="execution", + function_arn="arn:function", + status="FAILED", + start_timestamp=timestamp, + error=caller_error, + ) + history = GetDurableExecutionHistoryResponse( + events=[ + Event( + event_type="CallbackStarted", + event_timestamp=timestamp, + event_id=1, + operation_id="callback", + name="callback", + callback_started_details=CallbackStartedDetails( + callback_id="callback-id" + ), + ), + Event( + event_type="CallbackFailed", + event_timestamp=timestamp, + event_id=2, + operation_id="callback", + name="callback", + callback_failed_details=CallbackFailedDetails( + error=EventError( + payload=ErrorObject.from_dict(payload), truncated=False + ) + ), + ), + ] + ) + if decode_wire: + history = GetDurableExecutionHistoryResponse.from_dict(history.to_dict()) + assert history.events[-1].callback_failed_details.error.to_dict() == { + "Payload": payload, + "Truncated": False, + } + result = DurableFunctionTestResult.from_execution_history(execution, history) + callback = result.get_callback("callback") + assert callback.status is OperationStatus.FAILED + assert (callback.error.to_dict() if callback.error is not None else None) == ( + payload if payload else None + ) + assert result.status is InvocationStatus.FAILED + assert result.error is caller_error + + def test_durable_function_test_result_from_execution_history(): """Test DurableFunctionTestResult.from_execution_history factory method.""" import datetime diff --git a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py index 2762e2eb3..f1cb0779c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/e2e/plugin_operation_type_int_test.py @@ -123,10 +123,16 @@ def handler(_event: Any, context: DurableContext) -> str: assert {info.sub_type for info in plugin.operation_starts} == set(expected) assert {info.sub_type for info in plugin.operation_ends} == set(expected) - # The final replay starts with every earlier operation checkpointed, - # including the execution root, whose service record has no subtype. + # Live hooks above cover every category. The service prunes descendants of + # the completed callback/map/parallel contexts from the replay input; the + # surviving records, including the subtype-less execution root, keep type. replay_operations = plugin.invocation_starts[-1].operations.values() - assert {info.sub_type for info in replay_operations} == {*expected, None} + retained_subtypes = set(expected) - { + OperationSubType.CALLBACK, + OperationSubType.MAP_ITERATION, + OperationSubType.PARALLEL_BRANCH, + } + assert {info.sub_type for info in replay_operations} == {*retained_subtypes, None} all_infos: list[OperationInfo] = [ *plugin.operation_starts, *plugin.operation_ends,