feat: add a concurrency limit for named child runs

src/apify/_actor.py

@@ -8,7 +8,7 @@
 from contextlib import AsyncExitStack, ExitStack
 from dataclasses import asdict
 from datetime import UTC, datetime, timedelta
-from functools import cached_property, partial
+from functools import cached_property
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, overload
 
@@ -140,6 +140,7 @@ def __init__(
         status_message: str | None = None,
         event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,
         cleanup_timeout: timedelta = timedelta(seconds=30),
+        max_concurrent_child_runs: int | None = None,
     ) -> None:
         """Initialize a new instance.
 
@@ -152,6 +153,12 @@ def __init__(
             status_message: Final status message to display upon Actor termination.
             event_listeners_timeout: Maximum time to wait for Actor event listeners to complete before exiting.
             cleanup_timeout: Maximum time to wait for cleanup tasks to finish.
+            max_concurrent_child_runs: How many named child runs may be `READY`, `RUNNING`, `ABORTING` or `TIMING-OUT`
+                at once, or `None` for no limit. At the limit, a named `Actor.start`, `Actor.call`, `Actor.start_task`
+                or `Actor.call_task` waits to start or resurrect a run until another one finishes. Reattaching to a
+                recorded run never waits. Runs started without a `run_name` are not counted and never wait. A child
+                run's status is fetched again before it is counted, if it is more than 10 seconds old. Once this Actor
+                run is being gracefully aborted, a call that would wait raises a `RuntimeError`.
         """
         self._configuration = configuration
         self._configure_logging = configure_logging
@@ -171,7 +178,9 @@ def __init__(
         # Keep track of all used state stores to persist their values on exit
         self._use_state_stores: set[str | None] = set()
 
-        self._child_run_registry = ChildRunRegistry(self.open_key_value_store)
+        self._child_run_registry = ChildRunRegistry(
+            self.open_key_value_store, max_concurrent_runs=max_concurrent_child_runs
+        )
 
         self._internal_listeners = ExitStack()
         """Removes the listeners the Actor registered on the event manager, when the Actor leaves its context."""
@@ -385,6 +394,7 @@ def __call__(
         event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,
         status_message: str | None = None,
         cleanup_timeout: timedelta = timedelta(seconds=30),
+        max_concurrent_child_runs: int | None = None,
     ) -> Self:
         """Make a new Actor instance with a non-default configuration.
 
@@ -398,6 +408,7 @@ def __call__(
             event_listeners_timeout=event_listeners_timeout,
             status_message=status_message,
             cleanup_timeout=cleanup_timeout,
+            max_concurrent_child_runs=max_concurrent_child_runs,
         )
 
     @property
@@ -1122,19 +1133,20 @@ async def start(
         client = self.new_client(token=token) if token else self.apify_client
 
         actor_client = client.actor(actor_id)
-        start_run = partial(
-            actor_client.start,
-            run_input=run_input,
-            content_type=content_type,
-            build=build,
-            max_items=max_items,
-            max_total_charge_usd=max_total_charge_usd,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=self._resolve_run_timeout(timeout),
-            force_permission_level=force_permission_level,
-            webhooks=to_client_representations(webhooks),
-        )
+
+        async def start_run() -> Run:
+            return await actor_client.start(
+                run_input=run_input,
+                content_type=content_type,
+                build=build,
+                max_items=max_items,
+                max_total_charge_usd=max_total_charge_usd,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=self._resolve_run_timeout(timeout),
+                force_permission_level=force_permission_level,
+                webhooks=to_client_representations(webhooks),
+            )
 
         if run_name is None:
             return await start_run()
@@ -1340,13 +1352,9 @@ async def call(
                 logger=logger,
             )
         else:
-            started_run, is_new = await self._find_or_start_child_run(
-                run_name,
-                actor_id=actor_id,
-                run_input=run_input,
-                client=client,
-                start_run=partial(
-                    actor_client.start,
+
+            async def start_run() -> Run:
+                return await actor_client.start(
                     run_input=run_input,
                     content_type=content_type,
                     build=build,
@@ -1357,7 +1365,14 @@ async def call(
                     run_timeout=self._resolve_run_timeout(timeout),
                     force_permission_level=force_permission_level,
                     webhooks=to_client_representations(webhooks),
-                ),
+                )
+
+            started_run, is_new = await self._find_or_start_child_run(
+                run_name,
+                actor_id=actor_id,
+                run_input=run_input,
+                client=client,
+                start_run=start_run,
                 build=build,
                 max_items=max_items,
                 max_total_charge_usd=max_total_charge_usd,
@@ -1509,17 +1524,18 @@ async def start_task(
 
         client = self.new_client(token=token) if token else self.apify_client
         task_client = client.task(task_id)
-        start_run = partial(
-            task_client.start,
-            task_input=task_input,
-            build=build,
-            max_items=max_items,
-            max_total_charge_usd=max_total_charge_usd,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=self._resolve_run_timeout(timeout),
-            webhooks=to_client_representations(webhooks),
-        )
+
+        async def start_run() -> Run:
+            return await task_client.start(
+                task_input=task_input,
+                build=build,
+                max_items=max_items,
+                max_total_charge_usd=max_total_charge_usd,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=self._resolve_run_timeout(timeout),
+                webhooks=to_client_representations(webhooks),
+            )
 
         if run_name is None:
             return await start_run()
@@ -1624,13 +1640,9 @@ async def call_task(
                 wait_duration=wait,
             )
         else:
-            started_run, _ = await self._find_or_start_child_run(
-                run_name,
-                task_id=task_id,
-                run_input=task_input,
-                client=client,
-                start_run=partial(
-                    task_client.start,
+
+            async def start_run() -> Run:
+                return await task_client.start(
                     task_input=task_input,
                     build=build,
                     max_items=max_items,
@@ -1639,7 +1651,14 @@ async def call_task(
                     memory_mbytes=memory_mbytes,
                     run_timeout=self._resolve_run_timeout(timeout),
                     webhooks=to_client_representations(webhooks),
-                ),
+                )
+
+            started_run, _ = await self._find_or_start_child_run(
+                run_name,
+                task_id=task_id,
+                run_input=task_input,
+                client=client,
+                start_run=start_run,
                 build=build,
                 max_items=max_items,
                 max_total_charge_usd=max_total_charge_usd,