Skip to content

Commit 983e4bb

Browse files
committed
feat(testing): add skip-time virtual clock
- Add Clock protocol with RealClock and SkipClock implementations - SkipClock returns real time plus the accumulated skipped duration, so virtual time advances with real time between jumps and every read stays monotonic and distinct - One clock per runner, shared by the Executor, CheckpointProcessor, and all timestamp stamping, so history events form a single totally ordered clock domain - Drive wait and step-retry timers off the clock so modeled delays complete instantly under skip while history keeps modeled durations - Stamp operation, invocation, and execution start/end timestamps from the clock; reconstruct operations from event timestamps - Ban datetime.now via ruff TID251 outside clock.py so all time reads go through the clock - Default DurableFunctionTestRunner to skip_time=True; pass skip_time=False for real wall-clock timing - Remove the DURABLE_EXECUTION_TIME_SCALE env-var scaling
1 parent 1335cc1 commit 983e4bb

30 files changed

Lines changed: 639 additions & 177 deletions

File tree

packages/aws-durable-execution-sdk-python-examples/test/conftest.py

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -230,8 +230,11 @@ def test_hello_world(durable_runner):
230230
else:
231231
if not handler:
232232
pytest.fail("handler is required for local mode tests")
233+
skip_time = marker.kwargs.get("skip_time", True)
233234
# Create local runner (needs cleanup via context manager)
234-
runner = DurableFunctionTestRunner(handler=handler, execution_timeout=60)
235+
runner = DurableFunctionTestRunner(
236+
handler=handler, execution_timeout=60, skip_time=skip_time
237+
)
235238

236239
# Wrap in adapter and use context manager for proper cleanup
237240
with TestRunnerAdapter(runner, runner_mode) as adapter:

packages/aws-durable-execution-sdk-python-examples/test/plugin/test_plugin.py

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,8 @@ def test_plugin(durable_runner):
3434
lambda_function_name="Plugin Wait",
3535
)
3636
def test_plugin_on_operation_end_called_for_wait_completed_during_suspend(
37-
durable_runner, monkeypatch
37+
durable_runner,
3838
):
39-
monkeypatch.setenv("DURABLE_EXECUTION_TIME_SCALE", "0.01")
40-
4139
with durable_runner:
4240
result = durable_runner.run(input=None, timeout=30)
4341

packages/aws-durable-execution-sdk-python-testing/pyproject.toml

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,10 @@ target-version = "py311"
105105

106106
[tool.ruff.lint]
107107
preview = true
108-
select = ["E4", "E7", "E9", "F", "TID252"] # pycodestyle (E4/E7/E9) + Pyflakes + absolute imports
108+
select = ["E4", "E7", "E9", "F", "TID251", "TID252"] # pycodestyle (E4/E7/E9) + Pyflakes + banned APIs + absolute imports
109+
110+
[tool.ruff.lint.flake8-tidy-imports.banned-api]
111+
"datetime.datetime.now".msg = "Read time from the runner Clock (clock.now()) or clock.real_now() so all timestamps share one clock domain."
109112

110113
[tool.ruff.lint.isort]
111114
known-first-party = ["aws_durable_execution_sdk_python_testing"]
@@ -114,6 +117,7 @@ lines-after-imports = 2
114117

115118
[tool.ruff.lint.per-file-ignores]
116119
"tests/**" = [
120+
"TID251",
117121
"ARG001",
118122
"ARG002",
119123
"ARG005",

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/core.py

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
from aws_durable_execution_sdk_python_testing.token import CheckpointToken
2121

2222
if TYPE_CHECKING:
23+
from datetime import datetime
24+
2325
from aws_durable_execution_sdk_python.lambda_service import (
2426
Operation,
2527
OperationUpdate,
@@ -79,6 +81,7 @@ def apply(
7981
updates: list[OperationUpdate],
8082
client_token: str | None,
8183
dispatcher: CheckpointRequestDispatcher,
84+
now: datetime,
8285
) -> CheckpointResult:
8386
"""Apply ``updates`` to ``execution`` and compute the response delta.
8487
@@ -88,6 +91,10 @@ def apply(
8891
idempotency entry for a byte-identical replay of a retried call.
8992
The caller is responsible for the invocation gate, locking,
9093
persistence, and applying the returned effects.
94+
95+
``now`` is the checkpoint's single source of "now", resolved from
96+
the execution's clock, so all timestamps stamped by this apply
97+
advance with modeled time under a skip clock.
9198
"""
9299
effects: list[CheckpointEffect] = []
93100
if updates:
@@ -97,6 +104,7 @@ def apply(
97104
updates=updates,
98105
client_token=client_token,
99106
touch=execution.touch_operation,
107+
now=now,
100108
)
101109

102110
new_token_sequence: int = execution.advance_token_sequence()

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
from aws_durable_execution_sdk_python_testing.checkpoint.transformer import (
2121
CheckpointRequestDispatcher,
2222
)
23+
from aws_durable_execution_sdk_python_testing.clock import RealClock
2324
from aws_durable_execution_sdk_python_testing.exceptions import (
2425
InvalidParameterValueException,
2526
)
@@ -30,6 +31,7 @@
3031
if TYPE_CHECKING:
3132
from aws_durable_execution_sdk_python.lambda_service import OperationUpdate
3233

34+
from aws_durable_execution_sdk_python_testing.clock import Clock
3335
from aws_durable_execution_sdk_python_testing.execution import Execution
3436
from aws_durable_execution_sdk_python_testing.observer import ExecutionObserver
3537
from aws_durable_execution_sdk_python_testing.scheduler import Scheduler
@@ -52,10 +54,15 @@ def __init__(
5254
self,
5355
store: ExecutionStore,
5456
scheduler: Scheduler, # noqa: ARG002 — kept for backward-compatible signature
57+
clock: Clock | None = None,
5558
):
5659
self._store = store
5760
self._observers: list[ExecutionObserver] = []
5861
self._dispatcher = CheckpointRequestDispatcher()
62+
# The runner's clock, shared with the Executor so stamping uses
63+
# the same "now" as timer arming. Defaults to a real clock for
64+
# standalone construction (e.g. direct unit tests).
65+
self._clock: Clock = clock if clock is not None else RealClock()
5966

6067
def add_execution_observer(self, observer: ExecutionObserver) -> None:
6168
"""Add observer for execution events."""
@@ -91,12 +98,14 @@ def process_checkpoint(
9198
msg = "Invalid checkpoint token"
9299
raise InvalidParameterValueException(msg)
93100

101+
now = self._clock.now()
94102
result = CheckpointCore.apply(
95103
execution,
96104
checkpoint_token,
97105
updates,
98106
client_token,
99107
self._dispatcher,
108+
now,
100109
)
101110

102111
self._store.update(execution)

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/base.py

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -34,22 +34,29 @@ def process(
3434
current_op: Operation | None,
3535
notifier: ExecutionNotifier,
3636
execution_arn: str,
37+
now: datetime.datetime,
3738
) -> Operation | None:
38-
"""Process an operation update and return the transformed operation."""
39+
"""Process an operation update and return the transformed operation.
40+
41+
``now`` is the checkpoint's single source of "now", resolved from
42+
the execution's clock so timestamps advance with modeled time
43+
under a skip clock and match wall-clock under a real clock.
44+
"""
3945
raise NotImplementedError
4046

4147
def _get_start_time(
42-
self, current_operation: Operation | None
48+
self, current_operation: Operation | None, now: datetime.datetime
4349
) -> datetime.datetime | None:
4450
start_time: datetime.datetime | None = (
45-
current_operation.start_timestamp
46-
if current_operation
47-
else datetime.datetime.now(tz=datetime.UTC)
51+
current_operation.start_timestamp if current_operation else now
4852
)
4953
return start_time
5054

5155
def _get_end_time(
52-
self, current_operation: Operation | None, status: OperationStatus
56+
self,
57+
current_operation: Operation | None,
58+
status: OperationStatus,
59+
now: datetime.datetime,
5360
) -> datetime.datetime | None:
5461
"""Get end timestamp for operation based on current state and status."""
5562
if current_operation and current_operation.end_timestamp:
@@ -61,7 +68,7 @@ def _get_end_time(
6168
OperationStatus.TIMED_OUT,
6269
OperationStatus.STOPPED,
6370
}:
64-
return datetime.datetime.now(tz=datetime.UTC)
71+
return now
6572
return None
6673

6774
def _create_execution_details(
@@ -151,19 +158,22 @@ def _translate_update_to_operation(
151158
update: OperationUpdate,
152159
current_operation: Operation | None,
153160
status: OperationStatus,
161+
now: datetime.datetime,
154162
) -> Operation:
155163
"""Transform OperationUpdate to Operation, always creating new Operation."""
156-
start_time: datetime.datetime | None = self._get_start_time(current_operation)
164+
start_time: datetime.datetime | None = self._get_start_time(
165+
current_operation, now
166+
)
157167
end_time: datetime.datetime | None = self._get_end_time(
158-
current_operation, status
168+
current_operation, status, now
159169
)
160170

161171
execution_details = self._create_execution_details(update)
162172
context_details = self._create_context_details(update)
163173
step_details = self._create_step_details(update, current_operation)
164174
callback_details = self._create_callback_details(update)
165175
invoke_details = self._create_invoke_details(update)
166-
wait_details = self._create_wait_details(update, current_operation)
176+
wait_details = self._create_wait_details(update, current_operation, now)
167177

168178
return Operation(
169179
operation_id=update.operation_id,
@@ -183,7 +193,10 @@ def _translate_update_to_operation(
183193
)
184194

185195
def _create_wait_details(
186-
self, update: OperationUpdate, current_operation: Operation | None
196+
self,
197+
update: OperationUpdate,
198+
current_operation: Operation | None,
199+
now: datetime.datetime,
187200
) -> WaitDetails | None:
188201
"""Create WaitDetails from OperationUpdate."""
189202
if update.operation_type == OperationType.WAIT and update.wait_options:
@@ -192,8 +205,8 @@ def _create_wait_details(
192205
current_operation.wait_details.scheduled_end_timestamp
193206
)
194207
else:
195-
scheduled_end_timestamp = datetime.datetime.now(
196-
tz=datetime.UTC
197-
) + timedelta(seconds=update.wait_options.wait_seconds)
208+
scheduled_end_timestamp = now + timedelta(
209+
seconds=update.wait_options.wait_seconds
210+
)
198211
return WaitDetails(scheduled_end_timestamp=scheduled_end_timestamp)
199212
return None

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/callback.py

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ def process(
3535
current_op: Operation | None,
3636
notifier: ExecutionNotifier, # noqa: ARG002
3737
execution_arn: str, # noqa: ARG002
38+
now: datetime.datetime,
3839
) -> Operation:
3940
"""Process CALLBACK operation update with scheduler integration for activities."""
4041
match update.action:
@@ -58,10 +59,12 @@ def process(
5859

5960
status: OperationStatus = OperationStatus.STARTED
6061

61-
start_time: datetime.datetime | None = self._get_start_time(current_op)
62+
start_time: datetime.datetime | None = self._get_start_time(
63+
current_op, now
64+
)
6265

6366
end_time: datetime.datetime | None = self._get_end_time(
64-
current_op, status
67+
current_op, status, now
6568
)
6669

6770
operation: Operation = Operation(

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/context.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020

2121

2222
if TYPE_CHECKING:
23+
from datetime import datetime
24+
2325
from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier
2426

2527

@@ -32,6 +34,7 @@ def process(
3234
current_op: Operation | None,
3335
notifier: ExecutionNotifier, # noqa: ARG002
3436
execution_arn: str, # noqa: ARG002
37+
now: datetime,
3538
) -> Operation:
3639
"""Process CONTEXT operation update for context state transitions."""
3740
match update.action:
@@ -41,18 +44,21 @@ def process(
4144
update=update,
4245
current_operation=current_op,
4346
status=OperationStatus.STARTED,
47+
now=now,
4448
)
4549
case OperationAction.SUCCEED:
4650
return self._translate_update_to_operation(
4751
update=update,
4852
current_operation=current_op,
4953
status=OperationStatus.SUCCEEDED,
54+
now=now,
5055
)
5156
case OperationAction.FAIL:
5257
return self._translate_update_to_operation(
5358
update=update,
5459
current_operation=current_op,
5560
status=OperationStatus.FAILED,
61+
now=now,
5662
)
5763
case _:
5864
msg: str = "Invalid action for CONTEXT operation."

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/execution.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818

1919
if TYPE_CHECKING:
20+
from datetime import datetime
21+
2022
from aws_durable_execution_sdk_python_testing.observer import ExecutionNotifier
2123

2224

@@ -29,6 +31,7 @@ def process(
2931
current_op: Operation | None, # noqa: ARG002
3032
notifier: ExecutionNotifier,
3133
execution_arn: str,
34+
now: datetime, # noqa: ARG002
3235
) -> Operation | None:
3336
"""Process EXECUTION operation update for workflow completion/failure."""
3437
match update.action:

packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/checkpoint/processors/step.py

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
from __future__ import annotations
44

5-
from datetime import UTC, datetime, timedelta
5+
from datetime import datetime, timedelta
66
from typing import TYPE_CHECKING
77

88
from aws_durable_execution_sdk_python.lambda_service import (
@@ -34,6 +34,7 @@ def process(
3434
current_op: Operation | None,
3535
notifier: ExecutionNotifier,
3636
execution_arn: str,
37+
now: datetime,
3738
) -> Operation:
3839
"""Process STEP operation update with scheduler integration for retries."""
3940
match update.action:
@@ -42,6 +43,7 @@ def process(
4243
update=update,
4344
current_operation=current_op,
4445
status=OperationStatus.STARTED,
46+
now=now,
4547
)
4648
case OperationAction.RETRY:
4749
# set Status=PENDING, next attempt time, attempt count + 1
@@ -50,7 +52,7 @@ def process(
5052
if update.step_options
5153
else 0
5254
)
53-
next_attempt_time = datetime.now(UTC) + timedelta(seconds=delay)
55+
next_attempt_time = now + timedelta(seconds=delay)
5456

5557
# Build new step_details with incremented attempt
5658
current_attempt = (
@@ -72,9 +74,7 @@ def process(
7274
status=OperationStatus.PENDING,
7375
parent_id=update.parent_id,
7476
name=update.name,
75-
start_timestamp=(
76-
current_op.start_timestamp if current_op else datetime.now(UTC)
77-
),
77+
start_timestamp=(current_op.start_timestamp if current_op else now),
7878
end_timestamp=None,
7979
sub_type=update.sub_type,
8080
execution_details=current_op.execution_details
@@ -103,12 +103,14 @@ def process(
103103
update=update,
104104
current_operation=current_op,
105105
status=OperationStatus.SUCCEEDED,
106+
now=now,
106107
)
107108
case OperationAction.FAIL:
108109
return self._translate_update_to_operation(
109110
update=update,
110111
current_operation=current_op,
111112
status=OperationStatus.FAILED,
113+
now=now,
112114
)
113115
case _:
114116
msg: str = "Invalid action for STEP operation."

0 commit comments

Comments
 (0)