Skip to content

Commit 5a663b3

Browse files
fix: fetch paginated cloud execution history (#514)
* fix: fetch paginated cloud execution history * chore: update SAM template * test: move cloud runner import to module scope --------- Co-authored-by: GitHub Action <action@github.com>
1 parent 47977fb commit 5a663b3

6 files changed

Lines changed: 379 additions & 7 deletions

File tree

.github/workflows/cloud-tests.yml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ jobs:
4343
python-prefix: Py314
4444
steps:
4545
- uses: actions/checkout@v7
46-
46+
4747
- name: Setup Python
4848
uses: actions/setup-python@v6
4949
with:

packages/aws-durable-execution-sdk-python-examples/examples-catalog.json

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -585,6 +585,17 @@
585585
},
586586
"path": "./src/comprehensive_operations/comprehensive_operations.py"
587587
},
588+
{
589+
"name": "Many Operations",
590+
"description": "Runs many durable steps and child contexts to exercise long execution history retrieval",
591+
"handler": "many_operations.handler",
592+
"integration": true,
593+
"durableConfig": {
594+
"RetentionPeriodInDays": 7,
595+
"ExecutionTimeout": 300
596+
},
597+
"path": "./src/comprehensive_operations/many_operations.py"
598+
},
588599
{
589600
"name": "Create Callback Concurrency",
590601
"description": "Demonstrates multiple concurrent createCallback operations using context.parallel",
Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
"""Example with many durable step operations."""
2+
3+
from aws_durable_execution_sdk_python.context import (
4+
DurableContext,
5+
durable_with_child_context,
6+
)
7+
from aws_durable_execution_sdk_python.execution import durable_execution
8+
9+
10+
TOP_LEVEL_STEP_COUNT = 12
11+
CHILD_CONTEXT_COUNT = 4
12+
CHILD_STEPS_PER_CONTEXT = 2
13+
14+
15+
def make_step(value: int):
16+
return lambda _: value
17+
18+
19+
@durable_with_child_context
20+
def child_group(
21+
child_context: DurableContext, group_index: int, steps_per_context: int
22+
) -> list[int]:
23+
values = []
24+
for step_index in range(steps_per_context):
25+
value = group_index * 100 + step_index
26+
values.append(
27+
child_context.step(
28+
make_step(value),
29+
name=f"child-{group_index:02d}-step-{step_index:02d}",
30+
)
31+
)
32+
return values
33+
34+
35+
@durable_execution
36+
def handler(_event, context: DurableContext) -> dict[str, object]:
37+
"""Run many named step operations across parent and child contexts."""
38+
39+
step_results = []
40+
for index in range(TOP_LEVEL_STEP_COUNT):
41+
step_results.append(
42+
context.step(
43+
make_step(index),
44+
name=f"many-step-{index:03d}",
45+
)
46+
)
47+
48+
child_results = []
49+
for group_index in range(CHILD_CONTEXT_COUNT):
50+
child_results.append(
51+
context.run_in_child_context(
52+
child_group(group_index, CHILD_STEPS_PER_CONTEXT),
53+
name=f"child-context-{group_index:02d}",
54+
)
55+
)
56+
57+
child_values = [value for child_result in child_results for value in child_result]
58+
59+
return {
60+
"topLevelStepCount": TOP_LEVEL_STEP_COUNT,
61+
"childContextCount": CHILD_CONTEXT_COUNT,
62+
"childStepCount": CHILD_CONTEXT_COUNT * CHILD_STEPS_PER_CONTEXT,
63+
"totalStepCount": TOP_LEVEL_STEP_COUNT
64+
+ CHILD_CONTEXT_COUNT * CHILD_STEPS_PER_CONTEXT,
65+
"firstStep": step_results[0] if step_results else None,
66+
"lastStep": step_results[-1] if step_results else None,
67+
"topLevelStepTotal": sum(step_results),
68+
"childStepTotal": sum(child_values),
69+
}
Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
"""Tests for many_operations."""
2+
3+
import pytest
4+
from aws_durable_execution_sdk_python.execution import InvocationStatus
5+
6+
from src.comprehensive_operations import many_operations
7+
from test.conftest import deserialize_operation_payload
8+
9+
10+
TOP_LEVEL_STEP_COUNT = 12
11+
CHILD_CONTEXT_COUNT = 4
12+
CHILD_STEPS_PER_CONTEXT = 2
13+
TOTAL_STEP_COUNT = TOP_LEVEL_STEP_COUNT + CHILD_CONTEXT_COUNT * CHILD_STEPS_PER_CONTEXT
14+
15+
16+
@pytest.mark.example
17+
@pytest.mark.durable_execution(
18+
handler=many_operations.handler,
19+
lambda_function_name="Many Operations",
20+
)
21+
def test_fetches_many_steps_and_child_context_operations(durable_runner):
22+
"""Verify result history includes many steps and child context operations."""
23+
with durable_runner:
24+
result = durable_runner.run(input=None, timeout=60)
25+
26+
assert result.status is InvocationStatus.SUCCEEDED
27+
28+
result_data = deserialize_operation_payload(result.result)
29+
assert result_data == {
30+
"topLevelStepCount": TOP_LEVEL_STEP_COUNT,
31+
"childContextCount": CHILD_CONTEXT_COUNT,
32+
"childStepCount": CHILD_CONTEXT_COUNT * CHILD_STEPS_PER_CONTEXT,
33+
"totalStepCount": TOTAL_STEP_COUNT,
34+
"firstStep": 0,
35+
"lastStep": TOP_LEVEL_STEP_COUNT - 1,
36+
"topLevelStepTotal": sum(range(TOP_LEVEL_STEP_COUNT)),
37+
"childStepTotal": sum(
38+
group_index * 100 + step_index
39+
for group_index in range(CHILD_CONTEXT_COUNT)
40+
for step_index in range(CHILD_STEPS_PER_CONTEXT)
41+
),
42+
}
43+
44+
all_ops = result.get_all_operations()
45+
top_level_step_ops = [
46+
op
47+
for op in all_ops
48+
if op.operation_type.value == "STEP"
49+
and op.name is not None
50+
and op.name.startswith("many-step-")
51+
]
52+
assert len(top_level_step_ops) == TOP_LEVEL_STEP_COUNT
53+
54+
child_context_ops = [
55+
op
56+
for op in all_ops
57+
if op.operation_type.value == "CONTEXT"
58+
and op.name is not None
59+
and op.name.startswith("child-context-")
60+
]
61+
assert len(child_context_ops) == CHILD_CONTEXT_COUNT
62+
63+
child_step_ops = [
64+
op
65+
for op in all_ops
66+
if op.operation_type.value == "STEP"
67+
and op.name is not None
68+
and op.name.startswith("child-")
69+
]
70+
assert len(child_step_ops) == CHILD_CONTEXT_COUNT * CHILD_STEPS_PER_CONTEXT

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

Lines changed: 22 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1246,11 +1246,28 @@ def _fetch_execution_history(
12461246
Raises:
12471247
ClientError: If lambda client encounter error
12481248
"""
1249-
history_dict = self.lambda_client.get_durable_execution_history(
1250-
DurableExecutionArn=execution_arn,
1251-
IncludeExecutionData=True,
1252-
)
1253-
history_response = GetDurableExecutionHistoryResponse.from_dict(history_dict)
1249+
events: list[Event] = []
1250+
next_marker: str | None = None
1251+
1252+
while True:
1253+
request = {
1254+
"DurableExecutionArn": execution_arn,
1255+
"IncludeExecutionData": True,
1256+
}
1257+
if next_marker:
1258+
request["Marker"] = next_marker
1259+
1260+
history_dict = self.lambda_client.get_durable_execution_history(**request)
1261+
history_response = GetDurableExecutionHistoryResponse.from_dict(
1262+
history_dict
1263+
)
1264+
events.extend(history_response.events)
1265+
1266+
next_marker = history_response.next_marker
1267+
if not next_marker:
1268+
break
1269+
1270+
history_response = GetDurableExecutionHistoryResponse(events=events)
12541271

12551272
logger.info("Retrieved %d events from history", len(history_response.events))
12561273

0 commit comments

Comments
 (0)