feat: retry starting a run on memory and concurrent-runs limits

docs/02_concepts/05_retries.mdx

@@ -12,6 +12,8 @@ import ApiLink from '@theme/ApiLink';
 
 import RetriesAsyncExample from '!!raw-loader!./code/05_retries_async.py';
 import RetriesSyncExample from '!!raw-loader!./code/05_retries_sync.py';
+import WaitForResourcesAsyncExample from '!!raw-loader!./code/05_wait_for_resources_async.py';
+import WaitForResourcesSyncExample from '!!raw-loader!./code/05_wait_for_resources_sync.py';
 
 The Apify client automatically retries requests that fail due to:
 
@@ -43,3 +45,27 @@ Retries with exponential backoff help reduce the load on the server and increase
         </CodeBlock>
     </TabItem>
 </Tabs>
+
+## Wait for resources to start a run
+
+Starting a run fails with an HTTP 402 error when the account doesn't have enough free memory for the run, or when it already runs as many Actors as its plan allows. The error `type` is `actor-memory-limit-exceeded` or `concurrent-runs-limit-exceeded`. The client doesn't retry these errors on its own, since they clear only after other runs or builds of the account finish.
+
+To keep retrying the start until the resources free up, set the `wait_for_resources` argument of <ApiLink to="class/ActorClient#start">`ActorClient.start`</ApiLink>, <ApiLink to="class/ActorClient#call">`ActorClient.call`</ApiLink>, or the same methods of <ApiLink to="class/TaskClient">`TaskClient`</ApiLink>. The client then retries the start every 10 seconds, and the argument value sets how long:
+
+- `True` retries until the run starts.
+- A `timedelta` stops retrying after that time and raises the last error.
+
+Any other error raises right away. A run that asks for more memory than the account's whole memory limit gets the same `actor-memory-limit-exceeded` error and never starts, so with `True` the client retries it forever. In `call`, the time spent retrying doesn't count toward `wait_duration`.
+
+<Tabs>
+    <TabItem value="AsyncExample" label="Async client" default>
+        <CodeBlock className="language-python">
+            {WaitForResourcesAsyncExample}
+        </CodeBlock>
+    </TabItem>
+    <TabItem value="SyncExample" label="Sync client">
+        <CodeBlock className="language-python">
+            {WaitForResourcesSyncExample}
+        </CodeBlock>
+    </TabItem>
+</Tabs>

docs/02_concepts/code/05_wait_for_resources_async.py

@@ -0,0 +1,17 @@
+from datetime import timedelta
+
+from apify_client import ApifyClientAsync
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+async def main() -> None:
+    apify_client = ApifyClientAsync(TOKEN)
+
+    # Retry the start until the account has the resources for the run.
+    run = await apify_client.actor('username/actor-name').call(wait_for_resources=True)
+
+    # Stop retrying after 10 minutes and raise the last error.
+    started_run = await apify_client.task('username~task-name').start(
+        wait_for_resources=timedelta(minutes=10),
+    )

docs/02_concepts/code/05_wait_for_resources_sync.py

@@ -0,0 +1,17 @@
+from datetime import timedelta
+
+from apify_client import ApifyClient
+
+TOKEN = 'MY-APIFY-TOKEN'
+
+
+def main() -> None:
+    apify_client = ApifyClient(TOKEN)
+
+    # Retry the start until the account has the resources for the run.
+    run = apify_client.actor('username/actor-name').call(wait_for_resources=True)
+
+    # Stop retrying after 10 minutes and raise the last error.
+    started_run = apify_client.task('username~task-name').start(
+        wait_for_resources=timedelta(minutes=10),
+    )

src/apify_client/_resource_clients/actor.py

@@ -26,6 +26,11 @@
 from apify_client._utils.encoding import encode_key_value_store_record_value, encode_webhooks_to_base64
 from apify_client._utils.http import response_to_dict
 from apify_client._utils.time import to_seconds
+from apify_client._utils.wait_for_resources import (
+    prepare_resendable_body,
+    start_waiting_for_resources,
+    start_waiting_for_resources_async,
+)
 
 if TYPE_CHECKING:
     from datetime import timedelta
@@ -229,6 +234,7 @@ def start(
         force_permission_level: ActorPermissionLevel | None = None,
         wait_for_finish: int | None = None,
         webhooks: WebhooksList | None = None,
+        wait_for_resources: bool | timedelta = False,
         timeout: Timeout = 'medium',
     ) -> Run:
         """Start the Actor and immediately return the Run object.
@@ -263,12 +269,21 @@ def start(
                     * `event_types`: List of `WebhookEventType` values which trigger the webhook.
                     * `request_url`: URL to which to send the webhook HTTP request.
                     * `payload_template`: Optional template for the request payload.
+            wait_for_resources: Retry the start while the account lacks the memory or a concurrent-run slot for the run,
+                that is while the API rejects it with an `ApifyApiError` of type `actor-memory-limit-exceeded` or
+                `concurrent-runs-limit-exceeded`. Both clear as other runs or builds finish. The start is retried
+                every 10 seconds, and any other error is raised right away. `True` retries until the run starts, a
+                `timedelta` stops retrying after that long and raises the last error. A run that requests more memory
+                than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
+                never starts, so `True` retries it forever. A streamed `run_input` that cannot be rewound, such as a
+                generator, is sent only once, so its start is not retried.
             timeout: Timeout for the API HTTP request.
 
         Returns:
             The run object.
         """
         run_input, content_type = encode_key_value_store_record_value(run_input, content_type=content_type)
+        run_input, wait_for_resources = prepare_resendable_body(run_input, wait_for_resources=wait_for_resources)
 
         request_params = self._build_params(
             build=build,
@@ -282,13 +297,16 @@ def start(
             webhooks=encode_webhooks_to_base64(webhooks),
         )
 
-        response = self._http_client.call(
-            url=self._build_url('runs'),
-            method='POST',
-            headers={'content-type': content_type},
-            data=run_input,
-            params=request_params,
-            timeout=timeout,
+        response = start_waiting_for_resources(
+            lambda: self._http_client.call(
+                url=self._build_url('runs'),
+                method='POST',
+                headers={'content-type': content_type},
+                data=run_input,
+                params=request_params,
+                timeout=timeout,
+            ),
+            wait_for_resources=wait_for_resources,
         )
 
         result = response_to_dict(response)
@@ -308,6 +326,7 @@ def call(
         webhooks: WebhooksList | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_duration: timedelta | None = None,
+        wait_for_resources: bool | timedelta = False,
         logger: Logger | Literal['default'] | None = 'default',
         timeout: Timeout = 'no_timeout',
     ) -> Run | None:
@@ -341,6 +360,14 @@ def call(
                 a webhook set up for the Actor, you do not have to add it again here.
             wait_duration: The maximum time the server waits for the run to finish. If not provided,
                 waits indefinitely.
+            wait_for_resources: Retry the start while the account lacks the memory or a concurrent-run slot for the run,
+                that is while the API rejects it with an `ApifyApiError` of type `actor-memory-limit-exceeded` or
+                `concurrent-runs-limit-exceeded`. Both clear as other runs or builds finish. The start is retried
+                every 10 seconds, and any other error is raised right away. `True` retries until the run starts, a
+                `timedelta` stops retrying after that long and raises the last error. A run that requests more memory
+                than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
+                never starts, so `True` retries it forever. The time spent retrying doesn't count toward
+                `wait_duration`.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
                 will redirect logs to the provided logger. The logger is also used to capture status and status message
@@ -361,6 +388,7 @@ def call(
             run_timeout=run_timeout,
             webhooks=webhooks,
             force_permission_level=force_permission_level,
+            wait_for_resources=wait_for_resources,
             timeout=timeout,
         )
         run_client = self._client_registry.run_client(
@@ -740,6 +768,7 @@ async def start(
         force_permission_level: ActorPermissionLevel | None = None,
         wait_for_finish: int | None = None,
         webhooks: WebhooksList | None = None,
+        wait_for_resources: bool | timedelta = False,
         timeout: Timeout = 'medium',
     ) -> Run:
         """Start the Actor and immediately return the Run object.
@@ -774,12 +803,21 @@ async def start(
                     * `event_types`: List of `WebhookEventType` values which trigger the webhook.
                     * `request_url`: URL to which to send the webhook HTTP request.
                     * `payload_template`: Optional template for the request payload.
+            wait_for_resources: Retry the start while the account lacks the memory or a concurrent-run slot for the run,
+                that is while the API rejects it with an `ApifyApiError` of type `actor-memory-limit-exceeded` or
+                `concurrent-runs-limit-exceeded`. Both clear as other runs or builds finish. The start is retried
+                every 10 seconds, and any other error is raised right away. `True` retries until the run starts, a
+                `timedelta` stops retrying after that long and raises the last error. A run that requests more memory
+                than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
+                never starts, so `True` retries it forever. A streamed `run_input` that cannot be rewound, such as a
+                generator, is sent only once, so its start is not retried.
             timeout: Timeout for the API HTTP request.
 
         Returns:
             The run object.
         """
         run_input, content_type = encode_key_value_store_record_value(run_input, content_type=content_type)
+        run_input, wait_for_resources = prepare_resendable_body(run_input, wait_for_resources=wait_for_resources)
 
         request_params = self._build_params(
             build=build,
@@ -793,13 +831,16 @@ async def start(
             webhooks=encode_webhooks_to_base64(webhooks),
         )
 
-        response = await self._http_client.call(
-            url=self._build_url('runs'),
-            method='POST',
-            headers={'content-type': content_type},
-            data=run_input,
-            params=request_params,
-            timeout=timeout,
+        response = await start_waiting_for_resources_async(
+            lambda: self._http_client.call(
+                url=self._build_url('runs'),
+                method='POST',
+                headers={'content-type': content_type},
+                data=run_input,
+                params=request_params,
+                timeout=timeout,
+            ),
+            wait_for_resources=wait_for_resources,
         )
 
         result = response_to_dict(response)
@@ -819,6 +860,7 @@ async def call(
         webhooks: WebhooksList | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         wait_duration: timedelta | None = None,
+        wait_for_resources: bool | timedelta = False,
         logger: Logger | Literal['default'] | None = 'default',
         timeout: Timeout = 'no_timeout',
     ) -> Run | None:
@@ -852,6 +894,14 @@ async def call(
                 a webhook set up for the Actor, you do not have to add it again here.
             wait_duration: The maximum time the server waits for the run to finish. If not provided,
                 waits indefinitely.
+            wait_for_resources: Retry the start while the account lacks the memory or a concurrent-run slot for the run,
+                that is while the API rejects it with an `ApifyApiError` of type `actor-memory-limit-exceeded` or
+                `concurrent-runs-limit-exceeded`. Both clear as other runs or builds finish. The start is retried
+                every 10 seconds, and any other error is raised right away. `True` retries until the run starts, a
+                `timedelta` stops retrying after that long and raises the last error. A run that requests more memory
+                than the whole memory limit of the account is rejected with `actor-memory-limit-exceeded` as well and
+                never starts, so `True` retries it forever. The time spent retrying doesn't count toward
+                `wait_duration`.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
                 will redirect logs to the provided logger. The logger is also used to capture status and status message
@@ -872,6 +922,7 @@ async def call(
             run_timeout=run_timeout,
             webhooks=webhooks,
             force_permission_level=force_permission_level,
+            wait_for_resources=wait_for_resources,
             timeout=timeout,
         )