fix(testing): race-safe worker submit and shutdown

- Add ExecutionRegistry.submit with compare-and-evict retry so a
  worker that stops its lane between hand-out and submit no longer
  surfaces RuntimeError; reorder teardown to stop-then-remove.
- Add ExecutionRegistry.shutdown; WebRunner.stop and
  DurableFunctionTestRunner.close now stop per-execution lanes so
  they do not outlive the runner.
- Honor MaxItems in GetDurableExecutionState, clamped to the service
  page-count bounds, via the paginator's new count bound.
- Remove dead _release_gate and _cleanup_execution_state; fix
  logger.exception misuse on non-error paths; fix the
  wait_for_callback timeout message.

Author: yaythomas

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

@@ -37,11 +37,10 @@ def checkpoint(
         updates: list[OperationUpdate],
         client_token: str | None,
     ) -> CheckpointOutput:
-        worker = self._registry.get_or_create(durable_execution_arn)
         task = CheckpointTask(
             self._checkpoint_processor, checkpoint_token, updates, client_token
         )
-        return worker.submit(task).result()
+        return self._registry.submit(durable_execution_arn, task).result()
 
     def get_execution_state(
         self,
@@ -50,11 +49,10 @@ def get_execution_state(
         next_marker: str,
         max_items: int = 1000,
     ) -> StateOutput:
-        worker = self._registry.get_or_create(durable_execution_arn)
         task = GetStateTask(
             self._checkpoint_processor, checkpoint_token, next_marker, max_items
         )
-        return worker.submit(task).result()
+        return self._registry.submit(durable_execution_arn, task).result()
 
     def stop(self, execution_arn: str, payload: bytes | None) -> datetime.datetime:  # noqa: ARG002
         # TODO: implement

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

@@ -637,12 +637,16 @@ def page(
         self,
         marker: str | None,
         max_size_bytes: int,
+        max_items: int | None = None,
     ) -> tuple[list[Operation], str | None]:
         """Return a page of ops starting after ``marker``, bounded by
-        ``max_size_bytes``. Second element of the tuple is a marker for
-        the next page, or ``None`` when the page fits everything."""
+        ``max_size_bytes`` and, when given, ``max_items``. Second element
+        of the tuple is a marker for the next page, or ``None`` when the
+        page fits everything."""
         start_idx = self._resolve_marker(marker)
-        return self._walk_page(self.snapshot_operations, start_idx, max_size_bytes)
+        return self._walk_page(
+            self.snapshot_operations, start_idx, max_size_bytes, max_items
+        )
 
     def unseen_operations(self) -> list[Operation]:
         """Operations whose ``operation_last_touched_seq`` is strictly
@@ -675,13 +679,16 @@ def _walk_page(
         ops: list[Operation],
         start_idx: int,
         max_size_bytes: int,
+        max_items: int | None = None,
     ) -> tuple[list[Operation], str | None]:
         selected: list[Operation] = []
         total = 0
         for i in range(start_idx, len(ops)):
             op = ops[i]
             size = self._size_for(op)
-            if selected and total + size > max_size_bytes:
+            over_bytes: bool = total + size > max_size_bytes
+            over_count: bool = max_items is not None and len(selected) >= max_items
+            if selected and (over_bytes or over_count):
                 return selected, self._encode_marker(i)
             selected.append(op)
             total += size