test: pace LMI execution history reads

Collect each case's own executions during teardown and leave full-run
history/log collection to the final workflow step. Pace history reads
and bound throttle-only retries, including reads through the public
cloud runner, without retrying invocation or callback writes.

Cover pagination, throttling exhaustion, permission errors, and scoped
versus full-run collection with focused tests.

Author: zhongkechen

lmi_tests/README.md

@@ -185,6 +185,12 @@ later case's short observation budget. Checkpoint holds and returns are correlat
 to the same request, and stale-attempt effects are checked before waiting for a
 retry that might itself be unable to start on a pinned worker.
 
+Per-case teardown collects only that case's executions. Complete-run histories
+and CloudWatch logs are collected once by the final workflow step. Both runner
+and artifact history reads are paced at one request per second, with at most five
+throttling attempts within a 20-second retry budget (in addition to the finite
+transport timeout). Invocation and callback writes are not retried by this layer.
+
 ## Independent budgets and results
 
 | Budget | Default |

lmi_tests/cloud.py

@@ -14,13 +14,27 @@
 from lmi_tests.evidence import CollectionError, check_controls, select
 
 
+class RunnerClient:
+    """Share paced history reads with the public runner; other APIs are unchanged."""
+
+    def __init__(self, cloud):
+        self.cloud = cloud
+
+    def __getattr__(self, name):
+        return getattr(self.cloud.lam, name)
+
+    def get_durable_execution_history(self, **kwargs):
+        return self.cloud.history_page(**kwargs)
+
+
 class Cloud:
     def __init__(self, manifest):
         self.manifest = manifest
         self.lam, self.s3, self.logs = client("lambda"), client("s3"), client("logs")
         self.events = {}
         self.invocations = []
         self.gates = set()
+        self._next_history_read = 0.0
 
     def verify(self):
         for key, arn in self.manifest["functions"].items():
@@ -53,7 +67,7 @@ def start(self, scenario, fixture="normal", gate=None):
         runner = DurableFunctionCloudTestRunner(
             self.manifest["functions"][fixture], region=self.manifest["region"]
         )
-        runner.lambda_client = self.lam  # bounded transport, including async start
+        runner.lambda_client = RunnerClient(self)
         try:
             arn = runner.run_async(input=payload)
             if not arn:
@@ -106,7 +120,7 @@ def release(self, gate):
             Bucket=self.manifest["bucket"], Key="control/" + gate, Body=b"release"
         )
 
-    def refresh(self, markers=None):
+    def refresh(self, markers=None, *, validate_controls=True):
         # A new pytest fixture must not scan every earlier case before it can
         # observe its first event. Explicit [] requests the complete run for collection.
         if markers is None:
@@ -149,7 +163,7 @@ def read(key):
         )
         save(ARTIFACTS / "events.json", events)
         selected = [e for e in events if not markers or e["marker"] in markers]
-        if markers:
+        if markers and validate_controls:
             check_controls(selected)
         return selected
 
@@ -196,13 +210,39 @@ def finish(self, item, status="SUCCEEDED"):
         self.phase(item, "WRAPPER_RETURN")
         return self.history(item)
 
+    def history_page(self, **request):
+        """Bound read-only throttling retries without replaying invocation writes."""
+        deadline = time.monotonic() + 20
+        last_error = None
+        for attempt in range(5):
+            delay = max(0, self._next_history_read - time.monotonic())
+            if time.monotonic() + delay >= deadline:
+                break
+            if delay:
+                time.sleep(delay)
+            try:
+                return self.lam.get_durable_execution_history(**request)
+            except ClientError as error:
+                if error.response["Error"]["Code"] not in {
+                    "TooManyRequestsException",
+                    "ThrottlingException",
+                }:
+                    raise
+                last_error = error
+            finally:
+                self._next_history_read = time.monotonic() + 1
+            self._next_history_read = time.monotonic() + min(2**attempt, 4)
+        raise CollectionError(
+            "Execution-history API throttle budget exhausted"
+        ) from last_error
+
     def history(self, item):
         events, marker = [], None
         while True:
             request = {"DurableExecutionArn": item["arn"], "IncludeExecutionData": True}
             if marker:
                 request["Marker"] = marker
-            result = self.lam.get_durable_execution_history(**request)
+            result = self.history_page(**request)
             events.extend(result.get("Events", []))
             marker = result.get("NextMarker")
             if not marker:
@@ -229,25 +269,36 @@ def observed():
             message="No service invocation timeout evidence for " + request,
         )
 
-    def collect(self):
+    def collect(self, *, full_run=True):
         errors = []
+        if not full_run and not self.invocations:
+            return
         try:
-            events = self.refresh(markers=[])
+            events = self.refresh(
+                markers=[] if full_run else [i["marker"] for i in self.invocations],
+                validate_controls=False,
+            )
             # Independent collect steps can recover execution ARNs without the
             # pytest process or successful invoke response.
             items = {
                 e["execution"]: {"arn": e["execution"], "marker": e["marker"]}
                 for e in events
             }
-            for item in items.values():
+        except Exception as error:
+            errors.append(str(error))
+            items = {}
+        if not full_run:
+            items.update({i["arn"]: i for i in self.invocations})
+        for item in items.values():
+            try:
                 self.history(item)
                 save(
                     ARTIFACTS / f"executions/{item['marker']}.json",
                     self.lam.get_durable_execution(DurableExecutionArn=item["arn"]),
                 )
-        except Exception as error:
-            errors.append(str(error))
-        for key, arn in self.manifest["functions"].items():
+            except Exception as error:
+                errors.append(str(error))
+        for key, arn in self.manifest["functions"].items() if full_run else []:
             try:
                 config = self.lam.get_function_configuration(FunctionName=arn)
                 logs = []

lmi_tests/tests/e2e/conftest.py

@@ -50,4 +50,4 @@ def cloud(deployment):
         try:
             driver.release_all()
         finally:
-            driver.collect()
+            driver.collect(full_run=False)

lmi_tests/tests/unit/test_cloud.py

@@ -4,6 +4,7 @@
 
 import boto3
 from botocore.stub import ANY, Stubber
+from botocore.exceptions import ClientError
 import pytest
 
 from lmi_tests import cloud as module
@@ -191,3 +192,92 @@ def test_polling_reads_only_current_case_and_surfaces_control_errors(driver):
     driver.s3.get_paginator.return_value.paginate.assert_called_with(
         Bucket="unit-bucket", Prefix="events/"
     )
+
+
+class Clock:
+    def __init__(self):
+        self.now = 0.0
+
+    def monotonic(self):
+        return self.now
+
+    def time(self):
+        return self.now
+
+    def sleep(self, seconds):
+        self.now += seconds
+
+
+def throttle_error():
+    return ClientError(
+        {"Error": {"Code": "TooManyRequestsException", "Message": "Rate exceeded"}},
+        "GetDurableExecutionHistory",
+    )
+
+
+def test_history_retries_preserve_pages_and_pace_requests(driver, monkeypatch):
+    clock = Clock()
+    monkeypatch.setattr(module, "time", clock)
+    driver.lam.get_durable_execution_history = Mock(
+        side_effect=[
+            throttle_error(),
+            {"Events": [{"EventId": 1}], "NextMarker": "page2"},
+            {"Events": [{"EventId": 2}]},
+        ]
+    )
+    assert driver.history({"arn": "execution", "marker": "case"}) == [
+        {"EventId": 1},
+        {"EventId": 2},
+    ]
+    calls = driver.lam.get_durable_execution_history.call_args_list
+    assert calls[0] == calls[1]
+    assert calls[2].kwargs["Marker"] == "page2"
+    assert clock.now >= 2
+
+
+def test_history_throttling_has_a_finite_budget(driver, monkeypatch):
+    clock = Clock()
+    monkeypatch.setattr(module, "time", clock)
+    driver.lam.get_durable_execution_history = Mock(side_effect=throttle_error())
+    with pytest.raises(CollectionError, match="throttle budget"):
+        driver.history_page(DurableExecutionArn="execution")
+    assert driver.lam.get_durable_execution_history.call_count == 5
+    assert clock.now < 20
+
+
+def test_history_does_not_retry_permission_errors_or_invocations(driver, monkeypatch):
+    monkeypatch.setattr(module, "time", Clock())
+    driver.lam.invoke = Mock(
+        return_value={"StatusCode": 202, "DurableExecutionArn": "execution"}
+    )
+    item = driver.start("success")
+    error = ClientError(
+        {"Error": {"Code": "AccessDeniedException"}}, "GetDurableExecutionHistory"
+    )
+    driver.lam.get_durable_execution_history = Mock(side_effect=error)
+    with pytest.raises(ClientError):
+        item["runner"].lambda_client.get_durable_execution_history(
+            DurableExecutionArn="execution"
+        )
+    driver.lam.invoke.assert_called_once()
+    driver.lam.get_durable_execution_history.assert_called_once()
+
+
+def test_case_collection_is_scoped_but_final_collection_is_complete(driver):
+    driver.invocations = [{"arn": "execution", "marker": "case"}]
+    driver.refresh = Mock(return_value=[{"execution": "execution", "marker": "case"}])
+    driver.history = Mock(return_value=[])
+    driver.lam.get_durable_execution = Mock(return_value={"Status": "SUCCEEDED"})
+    driver.logs = Mock()
+    driver.collect(full_run=False)
+    driver.refresh.assert_called_once_with(markers=["case"], validate_controls=False)
+    driver.history.assert_called_once()
+    driver.logs.get_paginator.assert_not_called()
+    driver.manifest["created"] = 0
+    driver.lam.get_function_configuration = Mock(
+        return_value={"LoggingConfig": {"LogGroup": "group"}}
+    )
+    driver.logs.get_paginator.return_value.paginate.return_value = []
+    driver.collect()
+    driver.refresh.assert_called_with(markers=[], validate_controls=False)
+    driver.logs.get_paginator.assert_called_once_with("filter_log_events")