From c1b3e66a1c725751d97abb6bc04dc5cad92a7029 Mon Sep 17 00:00:00 2001 From: Thomas Gaigher Date: Thu, 9 Jul 2026 07:43:13 +0000 Subject: [PATCH] feat(testing): add skip-time virtual clock - Add Clock protocol with RealClock and SkipClock implementations - Give each ExecutionWorker its own clock via the registry factory - Drive wait and step-retry timers off the clock so modeled delays complete instantly under skip while history keeps real durations - Default DurableFunctionTestRunner to skip_time=True; pass skip_time=False for real wall-clock timing - Remove the DURABLE_EXECUTION_TIME_SCALE env-var scaling --- .../test/conftest.py | 5 +- .../test/plugin/test_plugin.py | 4 +- .../checkpoint/core.py | 8 ++ .../checkpoint/processor.py | 18 +++ .../checkpoint/processors/base.py | 41 ++++--- .../checkpoint/processors/callback.py | 7 +- .../checkpoint/processors/context.py | 6 + .../checkpoint/processors/execution.py | 3 + .../checkpoint/processors/step.py | 12 +- .../checkpoint/processors/wait.py | 15 +-- .../checkpoint/transformer.py | 6 +- .../cli.py | 7 ++ .../clock.py | 63 +++++++++++ .../executor.py | 7 +- .../runner.py | 28 ++++- .../worker/execution_worker.py | 11 +- .../worker/registry.py | 16 ++- .../tests/checkpoint/processors/base_test.py | 29 +++-- .../checkpoint/processors/callback_test.py | 36 +++++- .../checkpoint/processors/context_test.py | 40 ++++--- .../processors/execution_processor_test.py | 25 +++-- .../tests/checkpoint/processors/step_test.py | 46 +++++--- .../tests/checkpoint/processors/wait_test.py | 28 ++--- .../tests/checkpoint/transformer_test.py | 17 ++- .../tests/clock_test.py | 75 +++++++++++++ .../tests/skip_time_test.py | 105 ++++++++++++++++++ 26 files changed, 541 insertions(+), 117 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py create mode 100644 packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py diff --git a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py index 31d32b6f..84eda88c 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py @@ -230,8 +230,11 @@ def test_hello_world(durable_runner): else: if not handler: pytest.fail("handler is required for local mode tests") + skip_time = marker.kwargs.get("skip_time", True) # Create local runner (needs cleanup via context manager) - runner = DurableFunctionTestRunner(handler=handler, execution_timeout=60) + runner = DurableFunctionTestRunner( + handler=handler, execution_timeout=60, skip_time=skip_time + ) # Wrap in adapter and use context manager for proper cleanup with TestRunnerAdapter(runner, runner_mode) as adapter: diff --git a/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py b/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py index 28d041b0..888f661d 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py @@ -34,10 +34,8 @@ def test_plugin(durable_runner): lambda_function_name="Plugin Wait", ) def test_plugin_on_operation_end_called_for_wait_completed_during_suspend( - durable_runner, monkeypatch + durable_runner, ): - monkeypatch.setenv("DURABLE_EXECUTION_TIME_SCALE", "0.01") - with durable_runner: result = durable_runner.run(input=None, timeout=30) diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py index e1305560..89912b97 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py @@ -20,6 +20,8 @@ from aws_durable_execution_sdk_python_testing.token import CheckpointToken if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python.lambda_service import ( Operation, OperationUpdate, @@ -79,6 +81,7 @@ def apply( updates: list[OperationUpdate], client_token: str | None, dispatcher: CheckpointRequestDispatcher, + now: datetime, ) -> CheckpointResult: """Apply ``updates`` to ``execution`` and compute the response delta. @@ -88,6 +91,10 @@ def apply( idempotency entry for a byte-identical replay of a retried call. The caller is responsible for the invocation gate, locking, persistence, and applying the returned effects. + + ``now`` is the checkpoint's single source of "now", resolved from + the execution's clock, so all timestamps stamped by this apply + advance with modeled time under a skip clock. """ effects: list[CheckpointEffect] = [] if updates: @@ -97,6 +104,7 @@ def apply( updates=updates, client_token=client_token, touch=execution.touch_operation, + now=now, ) new_token_sequence: int = execution.advance_token_sequence() 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 c9565b5c..cf242096 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 @@ -20,6 +20,7 @@ from aws_durable_execution_sdk_python_testing.checkpoint.transformer import ( CheckpointRequestDispatcher, ) +from aws_durable_execution_sdk_python_testing.clock import RealClock from aws_durable_execution_sdk_python_testing.exceptions import ( InvalidParameterValueException, ) @@ -28,8 +29,11 @@ if TYPE_CHECKING: + from collections.abc import Callable + from aws_durable_execution_sdk_python.lambda_service import OperationUpdate + from aws_durable_execution_sdk_python_testing.clock import Clock from aws_durable_execution_sdk_python_testing.execution import Execution from aws_durable_execution_sdk_python_testing.observer import ExecutionObserver from aws_durable_execution_sdk_python_testing.scheduler import Scheduler @@ -39,6 +43,11 @@ # Canonical invocation-page byte budget, shared with the Executor. DEFAULT_MAX_INVOCATION_PAGE_BYTES = 5 * 1024 * 1024 +# Fallback clock used when no per-execution resolver is wired in. A real +# clock keeps standalone construction (e.g. direct unit tests) on +# wall-clock time. +_REAL_CLOCK = RealClock() + class CheckpointProcessor: """In-process checkpoint flow used by ``InMemoryServiceClient``. @@ -52,10 +61,17 @@ def __init__( self, store: ExecutionStore, scheduler: Scheduler, # noqa: ARG002 — kept for backward-compatible signature + clock_for: Callable[[str], Clock] | None = None, ): self._store = store self._observers: list[ExecutionObserver] = [] self._dispatcher = CheckpointRequestDispatcher() + # Resolve the clock for an execution ARN so stamping shares the + # same "now" as the scheduler's readiness checks. Defaults to a + # real clock when no resolver is wired in. + self._clock_for: Callable[[str], Clock] = clock_for or ( + lambda _arn: _REAL_CLOCK + ) def add_execution_observer(self, observer: ExecutionObserver) -> None: """Add observer for execution events.""" @@ -91,12 +107,14 @@ def process_checkpoint( msg = "Invalid checkpoint token" raise InvalidParameterValueException(msg) + now = self._clock_for(token.execution_arn).now() result = CheckpointCore.apply( execution, checkpoint_token, updates, client_token, self._dispatcher, + now, ) self._store.update(execution) 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 56933d5b..52eb64c2 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 @@ -34,22 +34,29 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime.datetime, ) -> Operation | None: - """Process an operation update and return the transformed operation.""" + """Process an operation update and return the transformed operation. + + ``now`` is the checkpoint's single source of "now", resolved from + the execution's clock so timestamps advance with modeled time + under a skip clock and match wall-clock under a real clock. + """ raise NotImplementedError def _get_start_time( - self, current_operation: Operation | None + self, current_operation: Operation | None, now: datetime.datetime ) -> datetime.datetime | None: start_time: datetime.datetime | None = ( - current_operation.start_timestamp - if current_operation - else datetime.datetime.now(tz=datetime.UTC) + current_operation.start_timestamp if current_operation else now ) return start_time def _get_end_time( - self, current_operation: Operation | None, status: OperationStatus + self, + current_operation: Operation | None, + status: OperationStatus, + now: datetime.datetime, ) -> datetime.datetime | None: """Get end timestamp for operation based on current state and status.""" if current_operation and current_operation.end_timestamp: @@ -61,7 +68,7 @@ def _get_end_time( OperationStatus.TIMED_OUT, OperationStatus.STOPPED, }: - return datetime.datetime.now(tz=datetime.UTC) + return now return None def _create_execution_details( @@ -151,11 +158,14 @@ def _translate_update_to_operation( update: OperationUpdate, current_operation: Operation | None, status: OperationStatus, + now: datetime.datetime, ) -> Operation: """Transform OperationUpdate to Operation, always creating new Operation.""" - start_time: datetime.datetime | None = self._get_start_time(current_operation) + start_time: datetime.datetime | None = self._get_start_time( + current_operation, now + ) end_time: datetime.datetime | None = self._get_end_time( - current_operation, status + current_operation, status, now ) execution_details = self._create_execution_details(update) @@ -163,7 +173,7 @@ def _translate_update_to_operation( step_details = self._create_step_details(update, current_operation) callback_details = self._create_callback_details(update) invoke_details = self._create_invoke_details(update) - wait_details = self._create_wait_details(update, current_operation) + wait_details = self._create_wait_details(update, current_operation, now) return Operation( operation_id=update.operation_id, @@ -183,7 +193,10 @@ def _translate_update_to_operation( ) def _create_wait_details( - self, update: OperationUpdate, current_operation: Operation | None + self, + update: OperationUpdate, + current_operation: Operation | None, + now: datetime.datetime, ) -> WaitDetails | None: """Create WaitDetails from OperationUpdate.""" if update.operation_type == OperationType.WAIT and update.wait_options: @@ -192,8 +205,8 @@ def _create_wait_details( current_operation.wait_details.scheduled_end_timestamp ) else: - scheduled_end_timestamp = datetime.datetime.now( - tz=datetime.UTC - ) + timedelta(seconds=update.wait_options.wait_seconds) + scheduled_end_timestamp = now + timedelta( + seconds=update.wait_options.wait_seconds + ) return WaitDetails(scheduled_end_timestamp=scheduled_end_timestamp) return None diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py index 48c2f016..1d1ede7b 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py @@ -35,6 +35,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, # noqa: ARG002 execution_arn: str, # noqa: ARG002 + now: datetime.datetime, ) -> Operation: """Process CALLBACK operation update with scheduler integration for activities.""" match update.action: @@ -58,10 +59,12 @@ def process( status: OperationStatus = OperationStatus.STARTED - start_time: datetime.datetime | None = self._get_start_time(current_op) + start_time: datetime.datetime | None = self._get_start_time( + current_op, now + ) end_time: datetime.datetime | None = self._get_end_time( - current_op, status + current_op, status, now ) operation: Operation = Operation( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py index 182bf916..9d3e22ec 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py @@ -20,6 +20,8 @@ if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier @@ -32,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, # noqa: ARG002 execution_arn: str, # noqa: ARG002 + now: datetime, ) -> Operation: """Process CONTEXT operation update for context state transitions.""" match update.action: @@ -41,18 +44,21 @@ def process( update=update, current_operation=current_op, status=OperationStatus.STARTED, + now=now, ) case OperationAction.SUCCEED: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.SUCCEEDED, + now=now, ) case OperationAction.FAIL: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.FAILED, + now=now, ) case _: msg: str = "Invalid action for CONTEXT operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py index e8ad2ef0..f8b29321 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py @@ -17,6 +17,8 @@ if TYPE_CHECKING: + from datetime import datetime + from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier @@ -29,6 +31,7 @@ def process( current_op: Operation | None, # noqa: ARG002 notifier: ExecutionNotifier, execution_arn: str, + now: datetime, # noqa: ARG002 ) -> Operation | None: """Process EXECUTION operation update for workflow completion/failure.""" match update.action: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py index af0d0338..26777c09 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from typing import TYPE_CHECKING from aws_durable_execution_sdk_python.lambda_service import ( @@ -34,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime, ) -> Operation: """Process STEP operation update with scheduler integration for retries.""" match update.action: @@ -42,6 +43,7 @@ def process( update=update, current_operation=current_op, status=OperationStatus.STARTED, + now=now, ) case OperationAction.RETRY: # set Status=PENDING, next attempt time, attempt count + 1 @@ -50,7 +52,7 @@ def process( if update.step_options else 0 ) - next_attempt_time = datetime.now(UTC) + timedelta(seconds=delay) + next_attempt_time = now + timedelta(seconds=delay) # Build new step_details with incremented attempt current_attempt = ( @@ -72,9 +74,7 @@ def process( status=OperationStatus.PENDING, parent_id=update.parent_id, name=update.name, - start_timestamp=( - current_op.start_timestamp if current_op else datetime.now(UTC) - ), + start_timestamp=(current_op.start_timestamp if current_op else now), end_timestamp=None, sub_type=update.sub_type, execution_details=current_op.execution_details @@ -103,12 +103,14 @@ def process( update=update, current_operation=current_op, status=OperationStatus.SUCCEEDED, + now=now, ) case OperationAction.FAIL: return self._translate_update_to_operation( update=update, current_operation=current_op, status=OperationStatus.FAILED, + now=now, ) case _: msg: str = "Invalid action for STEP operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py index b84387e6..bac02adb 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/wait.py @@ -2,9 +2,7 @@ from __future__ import annotations -import logging -import os -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from typing import TYPE_CHECKING from aws_durable_execution_sdk_python.lambda_service import ( @@ -36,6 +34,7 @@ def process( current_op: Operation | None, notifier: ExecutionNotifier, execution_arn: str, + now: datetime, ) -> Operation: """Process WAIT operation update with scheduler integration for timers.""" match update.action: @@ -43,13 +42,8 @@ def process( wait_seconds = ( update.wait_options.wait_seconds if update.wait_options else 0 ) - time_scale = float(os.getenv("DURABLE_EXECUTION_TIME_SCALE", "1.0")) - logging.info("Using DURABLE_EXECUTION_TIME_SCALE: %f", time_scale) - scaled_wait_seconds = wait_seconds * time_scale - scheduled_end_timestamp = datetime.now(UTC) + timedelta( - seconds=scaled_wait_seconds - ) + scheduled_end_timestamp = now + timedelta(seconds=wait_seconds) # Create WaitDetails with scheduled timestamp wait_details = WaitDetails( @@ -63,7 +57,7 @@ def process( status=OperationStatus.STARTED, parent_id=update.parent_id, name=update.name, - start_timestamp=datetime.now(UTC), + start_timestamp=now, end_timestamp=None, sub_type=update.sub_type, execution_details=None, @@ -87,6 +81,7 @@ def process( update=update, current_operation=current_op, status=OperationStatus.CANCELLED, + now=now, ) case _: msg: str = "Invalid action for WAIT operation." diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py index 9b344334..62971fd6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/transformer.py @@ -8,7 +8,6 @@ from __future__ import annotations import json -from datetime import UTC, datetime from typing import TYPE_CHECKING, ClassVar from aws_durable_execution_sdk_python.lambda_service import ( @@ -38,6 +37,7 @@ if TYPE_CHECKING: from collections.abc import Callable, MutableMapping + from datetime import datetime from aws_durable_execution_sdk_python.lambda_service import ( OperationUpdate, @@ -83,6 +83,7 @@ def apply_updates( updates: list[OperationUpdate], client_token: str | None, # noqa: ARG002 — reserved for future idempotency diagnostics touch: Callable[[str], None], + now: datetime, ) -> list[CheckpointEffect]: """Apply ``updates`` to ``execution`` in place. @@ -124,6 +125,7 @@ def apply_updates( current_op=current_op, notifier=collector, execution_arn=execution.durable_execution_arn, + now=now, ) if updated_op is None: continue @@ -143,7 +145,7 @@ def apply_updates( touch(update.operation_id) execution.updates.extend(updates) - execution.update_timestamps.extend(datetime.now(UTC) for _ in updates) + execution.update_timestamps.extend(now for _ in updates) return collector.effects diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py index d9452241..fce3d6ff 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/cli.py @@ -221,6 +221,12 @@ def _create_start_server_parser(self, subparsers) -> None: default=900, help="Per-invocation timeout in seconds, simulates Lambda Timeout (default: 900)", ) + start_server_parser.add_argument( + "--skip-time", + action=argparse.BooleanOptionalAction, + default=False, + help="Skip durable timer wall-clock waits; history keeps real modeled durations. Default is real timing (--no-skip-time); pass --skip-time to opt in", + ) start_server_parser.set_defaults(func=self.start_server_command) def _create_invoke_parser(self, subparsers) -> None: @@ -294,6 +300,7 @@ def start_server_command(self, args: argparse.Namespace) -> int: store_type=StoreType(args.store_type), store_path=args.store_path, invocation_timeout_seconds=args.invocation_timeout, + skip_time=args.skip_time, ) logger.info( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py new file mode 100644 index 00000000..d19d41f1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/clock.py @@ -0,0 +1,63 @@ +"""Clock sources for scheduler-driven durable timers. + +A clock decides how long the local runner actually waits before a +scheduled wake fires. History timestamps are written from the real +modeled durations regardless of the clock, so a :class:`SkipClock` run +records the same event history as a :class:`RealClock` run and only the +wall-clock spent waiting differs. +""" + +from __future__ import annotations + +import datetime +import threading +from typing import Protocol + + +class Clock(Protocol): + """Time source used to arm and evaluate durable timers.""" + + def now(self) -> datetime.datetime: + """Return the current time on this clock.""" + ... + + def arm(self, wake: datetime.datetime) -> float: + """Return the real seconds to wait before firing at ``wake``.""" + ... + + +class RealClock: + """Wall-clock time source. Timers wait the real modeled duration.""" + + def now(self) -> datetime.datetime: + return datetime.datetime.now(datetime.UTC) + + def arm(self, wake: datetime.datetime) -> float: + return max((wake - self.now()).total_seconds(), 0.0) + + +class SkipClock: + """Virtual clock that jumps to each armed timer horizon. + + Mutable owner of one execution's virtual now. Arming a wake advances + virtual now to that horizon and returns a zero delay, so the wake + fires on the next event-loop turn rather than after the modeled + duration. Guards its state with a lock because ``arm`` runs on the + scheduling thread while ``now`` runs on the fire worker thread. + """ + + def __init__(self) -> None: + self._now: datetime.datetime | None = None + self._lock: threading.Lock = threading.Lock() + + def now(self) -> datetime.datetime: + with self._lock: + if self._now is None: + self._now = datetime.datetime.now(datetime.UTC) + return self._now + + def arm(self, wake: datetime.datetime) -> float: + with self._lock: + if self._now is None or wake > self._now: + self._now = wake + return 0.0 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 6a61867d..b90622a6 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 @@ -847,12 +847,14 @@ def _checkpoint_execution( msg = "Invalid checkpoint token" raise InvalidParameterValueException(msg) + now = self._registry.get_or_create(execution_arn).clock.now() result = CheckpointCore.apply( execution, checkpoint_token, updates or [], client_token, self._dispatcher, + now, ) self._store.update(execution) @@ -1000,8 +1002,7 @@ def _schedule_earliest_pending(self, execution_arn: str) -> None: existing.cancel() return - now = datetime.now(UTC) - delay = max((earliest - now).total_seconds(), 0.0) + delay = self._registry.get_or_create(execution_arn).clock.arm(earliest) completion_event = self._completion_events.get(execution_arn) # Cancel-then-arm atomically under the supervisor lock so @@ -1031,7 +1032,7 @@ def _fire_due_operations(self, execution_arn: str) -> bool: if execution.is_complete: return False - now = datetime.now(UTC) + now = self._registry.get_or_create(execution_arn).clock.now() completed_any = False for op in list(execution.operations): if ( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py index 4b3afac8..8b9ac6a5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/runner.py @@ -36,6 +36,7 @@ DEFAULT_MAX_INVOCATION_PAGE_BYTES, ) from aws_durable_execution_sdk_python_testing.client import InMemoryServiceClient +from aws_durable_execution_sdk_python_testing.clock import RealClock, SkipClock from aws_durable_execution_sdk_python_testing.worker.registry import ExecutionRegistry from aws_durable_execution_sdk_python_testing.exceptions import ( DurableFunctionsLocalRunnerError, @@ -111,6 +112,12 @@ class WebRunnerConfig: # Timeout configuration invocation_timeout_seconds: int = 900 + # Skip durable timer wall-clock waits (waits and step retries complete + # on the next scheduler turn while history keeps the real modeled + # durations). Defaults False so the server emulates real cloud timing; + # set True to opt into fast skip. + skip_time: bool = False + # Invocation-input pagination cap (bytes). Handler receives pages # up to this size; larger state is served via # GetDurableExecutionState. None falls back to Executor default @@ -603,6 +610,7 @@ def __init__( execution_timeout: int = 300, invocation_timeout: int = 900, store: ExecutionStore | None = None, + skip_time: bool = True, # noqa: FBT001, FBT002 ): self._execution_timeout = execution_timeout self._invocation_timeout = invocation_timeout @@ -619,11 +627,16 @@ def __init__( store if store is not None else InMemoryExecutionStore() ) self.poll_interval = poll_interval + self._registry = ExecutionRegistry( + self._store, + self._scheduler, + clock_factory=SkipClock if skip_time else RealClock, + ) self._checkpoint_processor = CheckpointProcessor( store=self._store, scheduler=self._scheduler, + clock_for=lambda arn: self._registry.get_or_create(arn).clock, ) - self._registry = ExecutionRegistry(self._store, self._scheduler) self._service_client = InMemoryServiceClient( self._checkpoint_processor, self._registry ) @@ -870,8 +883,17 @@ def start(self) -> None: ) # Create shared CheckpointProcessor - checkpoint_processor = CheckpointProcessor(self._store, self._scheduler) - self._registry = ExecutionRegistry(self._store, self._scheduler) + registry = ExecutionRegistry( + self._store, + self._scheduler, + clock_factory=SkipClock if self._config.skip_time else RealClock, + ) + self._registry = registry + checkpoint_processor = CheckpointProcessor( + self._store, + self._scheduler, + clock_for=lambda arn: registry.get_or_create(arn).clock, + ) # Create executor with all dependencies including checkpoint processor self._executor = Executor( diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/execution_worker.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/execution_worker.py index a4d5e40b..bc7fa317 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/execution_worker.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/execution_worker.py @@ -10,6 +10,7 @@ if TYPE_CHECKING: from concurrent.futures import Future + from aws_durable_execution_sdk_python_testing.clock import Clock from aws_durable_execution_sdk_python_testing.execution import Execution from aws_durable_execution_sdk_python_testing.scheduler import Scheduler from aws_durable_execution_sdk_python_testing.stores.base import ExecutionStore @@ -44,12 +45,14 @@ def __init__( scheduler: Scheduler, registry: ExecutionRegistry, lane: SerialTaskLane, + clock: Clock, ) -> None: self._arn: str = execution_arn self._store: ExecutionStore = store self._scheduler: Scheduler = scheduler self._registry: ExecutionRegistry = registry self._lane: SerialTaskLane = lane + self._clock: Clock = clock self._status: InvocationState = InvocationState.PRE_INVOKE @classmethod @@ -59,12 +62,18 @@ def create( store: ExecutionStore, scheduler: Scheduler, registry: ExecutionRegistry, + clock: Clock, ) -> ExecutionWorker: """Build a worker and start its lane.""" lane: SerialTaskLane = SerialTaskLane.create( name=f"execution-worker-{execution_arn}" ) - return cls(execution_arn, store, scheduler, registry, lane) + return cls(execution_arn, store, scheduler, registry, lane, clock) + + @property + def clock(self) -> Clock: + """The clock arming and evaluating this execution's timers.""" + return self._clock @property def status(self) -> InvocationState: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/registry.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/registry.py index 7170dd42..222157d6 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/registry.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/worker/registry.py @@ -5,11 +5,13 @@ import threading from typing import TYPE_CHECKING, TypeVar +from aws_durable_execution_sdk_python_testing.clock import Clock, RealClock from aws_durable_execution_sdk_python_testing.worker.execution_worker import ( ExecutionWorker, ) if TYPE_CHECKING: + from collections.abc import Callable from concurrent.futures import Future from aws_durable_execution_sdk_python_testing.scheduler import Scheduler @@ -29,9 +31,15 @@ class ExecutionRegistry: a given ARN. """ - def __init__(self, store: ExecutionStore, scheduler: Scheduler) -> None: + def __init__( + self, + store: ExecutionStore, + scheduler: Scheduler, + clock_factory: Callable[[], Clock] = RealClock, + ) -> None: self._store = store self._scheduler = scheduler + self._clock_factory = clock_factory self._workers: dict[str, ExecutionWorker] = {} self._lock = threading.Lock() @@ -49,7 +57,11 @@ def get_or_create(self, execution_arn: str) -> ExecutionWorker: worker: ExecutionWorker | None = self._workers.get(execution_arn) if worker is None: worker = ExecutionWorker.create( - execution_arn, self._store, self._scheduler, self + execution_arn, + self._store, + self._scheduler, + self, + self._clock_factory(), ) self._workers[execution_arn] = worker return worker diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py index 5f92b58e..3d6e739e 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/base_test.py @@ -37,7 +37,9 @@ def test_process_not_implemented(): ) try: - processor.process(update, None, Mock(), "test-arn") + processor.process( + update, None, Mock(), "test-arn", datetime.datetime.now(tz=datetime.UTC) + ) pytest.fail("Expected NotImplementedError") except NotImplementedError: pass @@ -46,18 +48,25 @@ def test_process_not_implemented(): class MockProcessor(OperationProcessor): """Mock processor for testing base functionality.""" - def process(self, update, current_op, notifier, execution_arn): + def process(self, update, current_op, notifier, execution_arn, now=None): return self._translate_update_to_operation( - update, current_op, OperationStatus.STARTED + update, + current_op, + OperationStatus.STARTED, + now or datetime.datetime.now(tz=datetime.UTC), ) - def translate_update(self, update, current_op, status): + def translate_update(self, update, current_op, status, now=None): """Public method to access _translate_update_to_operation for testing.""" - return self._translate_update_to_operation(update, current_op, status) + return self._translate_update_to_operation( + update, current_op, status, now or datetime.datetime.now(tz=datetime.UTC) + ) - def get_end_time(self, current_op, status): + def get_end_time(self, current_op, status, now=None): """Public method to access _get_end_time for testing.""" - return self._get_end_time(current_op, status) + return self._get_end_time( + current_op, status, now or datetime.datetime.now(tz=datetime.UTC) + ) def create_execution_details(self, update): """Public method to access _create_execution_details for testing.""" @@ -79,9 +88,11 @@ def create_invoke_details(self, update): """Public method to access _create_invoke_details for testing.""" return self._create_invoke_details(update) - def create_wait_details(self, update, current_op): + def create_wait_details(self, update, current_op, now=None): """Public method to access _create_wait_details for testing.""" - return self._create_wait_details(update, current_op) + return self._create_wait_details( + update, current_op, now or datetime.datetime.now(tz=datetime.UTC) + ) def test_get_end_time_with_existing_end_timestamp(): diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py index 24bc1297..fdf01629 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/callback_test.py @@ -1,5 +1,6 @@ """Tests for callback operation processor.""" +from datetime import UTC, datetime from unittest.mock import Mock import pytest @@ -55,7 +56,11 @@ def test_process_start_action(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert isinstance(result, Operation) @@ -85,6 +90,7 @@ def test_process_start_action_with_current_operation(): current_op, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert isinstance(result, Operation) @@ -112,6 +118,7 @@ def test_process_invalid_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -134,6 +141,7 @@ def test_process_fail_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -156,6 +164,7 @@ def test_process_cancel_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -178,6 +187,7 @@ def test_process_retry_action(): None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) @@ -194,7 +204,11 @@ def test_process_with_payload(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.callback_details.result == "test-payload" @@ -213,7 +227,11 @@ def test_process_with_parent_id(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.parent_id == "parent-456" @@ -232,7 +250,11 @@ def test_process_with_sub_type(): ) result = processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert result.sub_type == "activity" @@ -250,7 +272,11 @@ def test_notifier_not_called_for_start(): ) processor.process( - update, None, notifier, "arn:aws:states:us-east-1:123456789012:execution:test" + update, + None, + notifier, + "arn:aws:states:us-east-1:123456789012:execution:test", + datetime.now(UTC), ) assert len(notifier.completed_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py index a070bc17..77b73467 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/context_test.py @@ -57,7 +57,7 @@ def test_process_start_action(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -82,7 +82,9 @@ def test_process_start_action_with_current_operation(): name="test-context", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.STARTED @@ -101,7 +103,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -126,7 +128,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.SUCCEEDED @@ -146,7 +150,7 @@ def test_process_fail_action(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "context-123" @@ -172,7 +176,9 @@ def test_process_fail_action_with_current_operation(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.FAILED @@ -193,7 +199,7 @@ def test_process_fail_action_with_payload_and_error(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result == "partial-result" assert result.context_details.error == error @@ -214,7 +220,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for CONTEXT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_cancel_action(): @@ -232,7 +238,7 @@ def test_process_cancel_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for CONTEXT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_with_parent_id(): @@ -248,7 +254,7 @@ def test_process_with_parent_id(): parent_id="parent-456", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -266,7 +272,7 @@ def test_process_with_sub_type(): sub_type="parallel", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "parallel" @@ -283,7 +289,7 @@ def test_process_start_without_payload(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -301,7 +307,7 @@ def test_process_succeed_without_payload(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -319,7 +325,7 @@ def test_process_fail_without_error(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.context_details.result is None assert result.context_details.error is None @@ -337,7 +343,7 @@ def test_no_notifier_calls(): name="test-context", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -357,7 +363,7 @@ def test_end_timestamp_set_for_terminal_states(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.end_timestamp is not None @@ -374,6 +380,6 @@ def test_end_timestamp_not_set_for_non_terminal_states(): name="test-context", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.end_timestamp is None diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py index 37c91ea3..1ca805d3 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/execution_processor_test.py @@ -1,5 +1,6 @@ """Tests for execution operation processor.""" +from datetime import UTC, datetime from unittest.mock import Mock from aws_durable_execution_sdk_python.lambda_service import ( @@ -50,7 +51,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.completed_calls) == 1 @@ -72,7 +73,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result is None assert len(notifier.completed_calls) == 1 @@ -90,7 +93,7 @@ def test_process_succeed_action_without_payload(): action=OperationAction.SUCCEED, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.completed_calls) == 1 @@ -110,7 +113,7 @@ def test_process_fail_action_with_error(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -129,7 +132,7 @@ def test_process_fail_action_without_error(): action=OperationAction.FAIL, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -153,7 +156,7 @@ def test_process_start_action(): action=OperationAction.START, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -173,7 +176,7 @@ def test_process_retry_action(): action=OperationAction.RETRY, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -193,7 +196,7 @@ def test_process_cancel_action(): action=OperationAction.CANCEL, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result is None assert len(notifier.failed_calls) == 1 @@ -217,7 +220,9 @@ def test_process_with_current_operation_and_error(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result is None assert len(notifier.failed_calls) == 1 @@ -236,7 +241,7 @@ def test_no_wait_timer_or_step_retry_calls(): payload="result", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.wait_timer_calls) == 0 assert len(notifier.step_retry_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py index ea37b0c9..b562d6dd 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/step_test.py @@ -59,7 +59,7 @@ def test_process_start_action(): name="test-step", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -84,7 +84,9 @@ def test_process_start_action_with_current_operation(): name="test-step", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp @@ -112,7 +114,9 @@ def test_process_retry_action(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -147,7 +151,9 @@ def test_process_retry_action_without_step_options(): name="test-step", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 1 # no per-op notifier call; central scheduler. @@ -168,7 +174,7 @@ def test_process_retry_action_without_current_operation(): step_options=step_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.step_details.attempt == 1 assert result.step_details.result is None @@ -198,7 +204,9 @@ def test_process_retry_action_without_current_step_details(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 1 @@ -216,7 +224,7 @@ def test_process_succeed_action(): payload="success-result", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -241,7 +249,9 @@ def test_process_succeed_action_with_current_operation(): payload="success-result", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.SUCCEEDED @@ -262,7 +272,7 @@ def test_process_fail_action(): error=error, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "step-123" @@ -288,7 +298,9 @@ def test_process_fail_action_with_current_operation(): error=error, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.start_timestamp == current_op.start_timestamp assert result.status == OperationStatus.FAILED @@ -310,7 +322,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for STEP operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_with_parent_id(): @@ -326,7 +338,7 @@ def test_process_with_parent_id(): parent_id="parent-456", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -344,7 +356,7 @@ def test_process_with_sub_type(): sub_type="lambda", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "lambda" @@ -374,7 +386,9 @@ def test_retry_preserves_current_operation_details(): step_options=step_options, ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert result.step_details.attempt == 3 assert result.step_details.result is None @@ -398,7 +412,7 @@ def test_no_completed_or_failed_calls_for_non_execution_actions(): name="test-step", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -417,6 +431,6 @@ def test_no_step_retry_calls_for_non_retry_actions(): name="test-step", ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.step_retry_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py index 9f635b5c..f48fe6ce 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/processors/wait_test.py @@ -59,7 +59,7 @@ def test_process_start_action(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.operation_id == "wait-123" @@ -88,7 +88,7 @@ def test_process_start_action_without_wait_options(): name="test-wait", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.wait_details is not None @@ -111,7 +111,7 @@ def test_process_start_action_with_zero_seconds(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.wait_details is not None @@ -135,7 +135,7 @@ def test_process_start_action_with_parent_id(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.parent_id == "parent-456" @@ -155,7 +155,7 @@ def test_process_start_action_with_sub_type(): wait_options=wait_options, ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.sub_type == "timer" @@ -175,7 +175,9 @@ def test_process_cancel_action(): name="test-wait", ) - result = processor.process(update, current_op, notifier, execution_arn) + result = processor.process( + update, current_op, notifier, execution_arn, datetime.now(UTC) + ) assert isinstance(result, Operation) assert result.operation_id == "wait-123" @@ -195,7 +197,7 @@ def test_process_cancel_action_without_current_operation(): name="test-wait", ) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert isinstance(result, Operation) assert result.status == OperationStatus.CANCELLED @@ -216,7 +218,7 @@ def test_process_invalid_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_fail_action(): @@ -234,7 +236,7 @@ def test_process_fail_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_process_retry_action(): @@ -252,7 +254,7 @@ def test_process_retry_action(): with pytest.raises( InvalidParameterValueException, match="Invalid action for WAIT operation" ): - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) def test_wait_details_created_correctly(): @@ -270,7 +272,7 @@ def test_wait_details_created_correctly(): ) before_time = datetime.now(UTC) - result = processor.process(update, None, notifier, execution_arn) + result = processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert result.wait_details.scheduled_end_timestamp > before_time @@ -289,7 +291,7 @@ def test_no_completed_or_failed_calls(): wait_options=wait_options, ) - processor.process(update, None, notifier, execution_arn) + processor.process(update, None, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.completed_calls) == 0 assert len(notifier.failed_calls) == 0 @@ -311,6 +313,6 @@ def test_cancel_no_timer_scheduled(): name="test-wait", ) - processor.process(update, current_op, notifier, execution_arn) + processor.process(update, current_op, notifier, execution_arn, datetime.now(UTC)) assert len(notifier.wait_timer_calls) == 0 diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py index 3d433cdf..f152b0c5 100644 --- a/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py +++ b/packages/aws-durable-execution-sdk-python-testing/tests/checkpoint/transformer_test.py @@ -8,6 +8,7 @@ from __future__ import annotations +from datetime import UTC, datetime from unittest.mock import Mock import pytest @@ -41,7 +42,7 @@ def __init__(self, return_value=None): self.return_value = return_value self.calls: list[tuple] = [] - def process(self, update, current_op, notifier, execution_arn): + def process(self, update, current_op, notifier, execution_arn, now=None): # noqa: ARG002 self.calls.append((update, current_op, notifier, execution_arn)) return self.return_value @@ -86,6 +87,7 @@ def test_apply_updates_with_empty_list_is_a_noop(): updates=[], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [] @@ -113,6 +115,7 @@ def test_apply_updates_unknown_type_raises(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) @@ -134,6 +137,7 @@ def test_apply_updates_skips_ops_when_processor_returns_none(): updates=[update], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [] @@ -162,6 +166,7 @@ def test_apply_updates_appends_new_operation_and_touches(): updates=[update], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [new_op] @@ -191,6 +196,7 @@ def test_apply_updates_replaces_existing_operation_in_place(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operations == [replaced] @@ -229,6 +235,7 @@ def test_apply_updates_preserves_order_across_multiple_updates(): ], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [op1, updated_op2, op3] @@ -245,6 +252,7 @@ def test_apply_updates_preserves_order_across_multiple_updates(): ], client_token=None, touch=touched.append, + now=datetime.now(UTC), ) assert execution.operations == [op1, updated_op2, op3, new_op4] assert touched == ["op2", "op4"] @@ -283,6 +291,7 @@ def test_apply_updates_dispatches_by_operation_type(): ], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operations == [step_op, wait_op] @@ -310,6 +319,7 @@ def test_apply_updates_forwards_arn_notifier_and_current_op_to_processor(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) forwarded_update, forwarded_current_op, forwarded_notifier, forwarded_arn = ( @@ -339,6 +349,7 @@ def test_apply_updates_returns_completion_effect(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert effects == [ @@ -366,6 +377,7 @@ def test_apply_updates_returns_no_effects_for_plain_step(): ], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert effects == [] @@ -393,6 +405,7 @@ def test_apply_updates_records_payload_size_for_paging(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) assert execution.operation_size_bytes["with-payload"] >= len(b"hello-world-payload") @@ -422,6 +435,7 @@ def test_apply_updates_records_size_for_error_payload(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) # Some non-zero size was recorded (exact value depends on @@ -451,6 +465,7 @@ def test_apply_updates_records_size_for_bytes_payload(): updates=[update], client_token=None, touch=lambda _: None, + now=datetime.now(UTC), ) # bytes payload length == 12. diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py new file mode 100644 index 00000000..8b0f4963 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/clock_test.py @@ -0,0 +1,75 @@ +"""Tests for the durable timer clocks.""" + +from __future__ import annotations + +import datetime +from unittest.mock import Mock + +from aws_durable_execution_sdk_python_testing.clock import ( + RealClock, + SkipClock, +) +from aws_durable_execution_sdk_python_testing.worker.registry import ( + ExecutionRegistry, +) + + +def test_real_clock_arm_returns_remaining_wall_seconds() -> None: + clock = RealClock() + wake = clock.now() + datetime.timedelta(seconds=30) + + delay: float = clock.arm(wake) + + # The real clock waits the modeled duration, minus the tiny elapsed + # time between now() calls. + assert 29.0 <= delay <= 30.0 + + +def test_real_clock_arm_never_negative_for_past_wake() -> None: + clock = RealClock() + past = clock.now() - datetime.timedelta(seconds=30) + + assert clock.arm(past) == 0.0 + + +def test_skip_clock_arm_returns_zero_delay() -> None: + clock = SkipClock() + wake = clock.now() + datetime.timedelta(seconds=300) + + assert clock.arm(wake) == 0.0 + + +def test_skip_clock_now_jumps_to_armed_horizon() -> None: + clock = SkipClock() + wake = clock.now() + datetime.timedelta(seconds=300) + + clock.arm(wake) + + assert clock.now() == wake + + +def test_skip_clock_is_monotonic() -> None: + clock = SkipClock() + start = clock.now() + far = start + datetime.timedelta(seconds=300) + near = start + datetime.timedelta(seconds=10) + + clock.arm(far) + clock.arm(near) + + # A later arm for an earlier horizon must not move virtual time back. + assert clock.now() == far + + +def test_registry_uses_clock_factory() -> None: + skip_registry = ExecutionRegistry(Mock(), Mock(), clock_factory=SkipClock) + real_registry = ExecutionRegistry(Mock(), Mock(), clock_factory=RealClock) + + assert isinstance(skip_registry.get_or_create("arn").clock, SkipClock) + assert isinstance(real_registry.get_or_create("arn").clock, RealClock) + + +def test_registry_defaults_to_real_clock() -> None: + registry = ExecutionRegistry(Mock(), Mock()) + + assert isinstance(registry.get_or_create("arn").clock, RealClock) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py new file mode 100644 index 00000000..5d26e73c --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/skip_time_test.py @@ -0,0 +1,105 @@ +"""End-to-end tests for skip-time behaviour on the local runner.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext, durable_step +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python.types import StepContext + +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestRunner, +) + + +_MODELED_WAIT_SECONDS = 300 + +_MODELED_MIDDLE_WAIT_SECONDS = 60 + + +@durable_execution +def _waits_then_returns(event: Any, context: DurableContext) -> str: # noqa: ARG001 + context.wait(Duration.from_seconds(_MODELED_WAIT_SECONDS)) + return "done" + + +def test_skip_time_completes_long_wait_and_keeps_faithful_history() -> None: + # skip_time defaults to True. A modeled 300s wait must complete well + # within the 30s execution timeout, which is only possible if the + # runner does not spend the modeled duration in wall-clock time. + with DurableFunctionTestRunner(handler=_waits_then_returns) as runner: + result = runner.run(input="x", execution_timeout=30) + + assert result.status is InvocationStatus.SUCCEEDED + + wait_ops = [op for op in result.operations if op.operation_type.value == "WAIT"] + assert len(wait_ops) == 1 + wait_op = wait_ops[0] + + assert wait_op.status.value == "SUCCEEDED" + assert wait_op.start_timestamp is not None + assert wait_op.scheduled_end_timestamp is not None + + # History records the real modeled duration, not the skipped + # wall-clock time. + modeled = ( + wait_op.scheduled_end_timestamp - wait_op.start_timestamp + ).total_seconds() + assert abs(modeled - _MODELED_WAIT_SECONDS) < 1.0 + + +@durable_step +def _step_a(step_context: StepContext) -> str: # noqa: ARG001 + return "a" + + +@durable_step +def _step_b(step_context: StepContext) -> str: # noqa: ARG001 + return "b" + + +@durable_execution +def _step_wait_step(event: Any, context: DurableContext) -> str: # noqa: ARG001 + context.step(_step_a()) + context.wait(Duration.from_seconds(_MODELED_MIDDLE_WAIT_SECONDS)) + context.step(_step_b()) + return "done" + + +def test_skip_time_keeps_history_monotonic_across_wait() -> None: + # stepA -> wait(60) -> stepB. Under skip, the clock is the single + # source of "now" for stamping, so the operation after the wait must + # start no earlier than the wait's modeled end. The history stays + # monotonic even though it is future-dated relative to wall-clock. + with DurableFunctionTestRunner(handler=_step_wait_step) as runner: + result = runner.run(input="x", execution_timeout=30) + + assert result.status is InvocationStatus.SUCCEEDED + + step_a = result.get_step("_step_a") + step_b = result.get_step("_step_b") + wait_ops = [op for op in result.operations if op.operation_type.value == "WAIT"] + assert len(wait_ops) == 1 + wait_op = wait_ops[0] + + assert wait_op.start_timestamp is not None + assert wait_op.scheduled_end_timestamp is not None + assert step_a.start_timestamp is not None + assert step_b.start_timestamp is not None + + # Faithful: the modeled wait duration is recorded exactly. + modeled = ( + wait_op.scheduled_end_timestamp - wait_op.start_timestamp + ).total_seconds() + assert abs(modeled - _MODELED_MIDDLE_WAIT_SECONDS) < 1.0 + + # Monotonic: the post-wait step cannot start before the wait's + # modeled end (the inversion the clock fix removes). + assert step_b.start_timestamp >= wait_op.scheduled_end_timestamp + # And the pre-wait step starts no later than the wait. + assert step_a.start_timestamp <= wait_op.start_timestamp