Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,33 @@ All notable changes to the Rust Apify API client are documented here. The format
based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and the project adheres
to [Semantic Versioning](https://semver.org/).

## [0.12.0] - 2026-10-09

### Added
- `ActorStartOptions::wait_for_resources` (type `WaitForResources`) on `ActorClient::start`/`call`
and `TaskClient::start`/`call`: retries the start every 10 seconds while the API rejects it with
`actor-memory-limit-exceeded` or `concurrent-runs-limit-exceeded`, matching the reference
client's new `waitForResources` option.

### Fixed
- `ListIterator` (`iterate()`/`iterate_items()` on every collection) no longer stops at a page's
reported `total`: it now keeps requesting pages until one scans no rows, so a listing that grows
while it is being iterated (e.g. a dataset a still-running Actor keeps pushing to) is read to the
end instead of being cut off at the size it had when the loop started. Matches the reference
client's `listResourcesPaginated`/`listItems` fix.
- `DatasetClient::iterate_items` advances its offset by the `x-apify-pagination-count` response
header (falling back to the returned item count) instead of the returned item count alone, so a
`skip_empty`/`clean`/`skip_hidden` filter or `unwind` no longer causes rows to be re-scanned or
skipped between pages.
- `RunClient::wait_for_finish`/`BuildClient::wait_for_finish` no longer poll a run/build ID that
keeps answering `404` forever when `wait_secs` is `None`: past a 3-second grace period (reset by
any successful fetch), the `404` is surfaced as an error instead. Matches the reference client's
`DEFAULT_WAIT_WHEN_JOB_NOT_EXIST` fix for `waitForFinish()`.

### Changed
- Bumped `API_SPEC_VERSION` to `v2-2026-10-08T073338Z`.
- Bumped crate version to `0.12.0`.

## [0.11.0] - 2026-10-02

### Added
Expand Down
4 changes: 2 additions & 2 deletions Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "apify-client"
version = "0.11.0"
version = "0.12.0"
authors = ["Apify Technologies <support@apify.com>"]
description = "An official, but experimental, AI-generated and AI-maintained Rust client for the Apify API (https://apify.com)."
license = "Apache-2.0"
Expand Down Expand Up @@ -28,7 +28,7 @@ brotli = "7"
flate2 = "1"

[dev-dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "test-util"] }
uuid = { version = "1", features = ["v4"] }

[features]
Expand Down
5 changes: 4 additions & 1 deletion docs/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,10 @@ demand as you consume items. Every collection client provides it (`actors`, `bui
exposes `iterate_items()` for dataset items, and `KeyValueStoreClient` exposes `iterate_keys()`
for store keys (cursor-based). The options' `limit` caps the total number of items yielded (unset
iterates everything); to control the per-request page size, call `.with_chunk_size(n)` on the
returned iterator (offset-paginated iterators only).
returned iterator (offset-paginated iterators only). An offset-paginated iterator ignores the
`total` a page reports and keeps requesting pages until one scans no rows, so a listing that grows
while being iterated (e.g. a dataset a still-running Actor keeps pushing to) is read to the end
rather than cut off at the size it had when the loop started.

```rust,no_run
use apify_client::{ApifyClient, ActorListOptions};
Expand Down
43 changes: 43 additions & 0 deletions docs/actors.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ All fields are optional. Used by both `start` and `call` here, and by the identi
| `restart_on_error` | `Option<bool>` | Whether to restart the run if it fails. |
| `force_permission_level` | `Option<String>` | Override the Actor's permission level for this run. |
| `webhooks` | `Option<Vec<serde_json::Value>>` | Ad-hoc webhooks to attach to this run. Encoded as base64 JSON in the `webhooks` query parameter, matching the reference clients. |
| `wait_for_resources` | `Option<WaitForResources>` | Retry the start while the account lacks the memory or a concurrent-run slot for it (`ApiError::error_type` of `actor-memory-limit-exceeded` or `concurrent-runs-limit-exceeded`). See below. |

The `wait_secs` argument of `call` (and of `wait_for_finish` on runs/builds) controls the
client-side polling budget:
Expand All @@ -59,6 +60,48 @@ client-side polling budget:
- `Some(n)` bounds the wait to roughly `n` seconds; if the run has not finished by then, the
**last fetched (still non-terminal) run is returned** rather than an error — inspect
`run.status` / `run.is_terminal()` on the result.
- A run/build ID that keeps 404ing (e.g. it was never valid, or the resource was deleted) is not
polled forever: past a 3-second grace period (reset by any successful fetch), `wait_for_finish`
returns the `404` as an error instead.

### Waiting for resources to start a run

Starting a run fails with an `ApiError` (HTTP 402) when the account has no free memory for it, or
already runs as many Actors as its plan allows (`error_type` of `actor-memory-limit-exceeded` or
`concurrent-runs-limit-exceeded`). Both clear only once other runs or builds of the account
finish, so the client does not retry them by default.

Set `ActorStartOptions::wait_for_resources` to retry the start every 10 seconds instead:

- `WaitForResources::Forever` retries until the run starts.
- `WaitForResources::ForSecs(secs)` stops retrying after that many seconds and returns the last
error.

Any other error is returned right away. A run that asks for more memory than the account's whole
memory limit is rejected with `actor-memory-limit-exceeded` too and never starts, so `Forever`
retries it forever. In `call`, time spent retrying the start does not count toward `wait_secs`.

```rust,no_run
use apify_client::{ApifyClient, ActorStartOptions, WaitForResources};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = ApifyClient::new("my-api-token");

// Retry the start until the account has the resources for the run.
let run = client
.actor("username/actor-name")
.call::<serde_json::Value>(
None,
ActorStartOptions {
wait_for_resources: Some(WaitForResources::Forever),
..Default::default()
},
None,
)
.await?;
# Ok(())
# }
```

> Note: `list` here takes `ActorListOptions` (fields `offset, limit, desc, my, sort_by`),
> which is distinct from the generic `ListOptions { offset, limit, desc }` used by most other
Expand Down
2 changes: 1 addition & 1 deletion docs/builds.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ collections are available via `actor.builds()`.
| `get()` | — | `Option<Build>` | Fetches the build. |
| `abort()` | — | `Build` | Aborts the build. |
| `delete()` | — | `()` | Deletes the build. |
| `wait_for_finish(wait_secs)` | `Option<i64>` | `Build` | Polls until the build is terminal. |
| `wait_for_finish(wait_secs)` | `Option<i64>` | `Build` | Polls until the build is terminal. A build ID that keeps 404ing past a 3-second grace period (reset by any successful fetch) returns the `404` as an error instead of polling forever. |
| `get_openapi_definition()` | — | `Option<serde_json::Value>` | Fetches the OpenAPI definition generated for the build (raw JSON, endpoint `.../openapi.json`). |
| `log()` | — | `LogClient` | Access the build's log. |

Expand Down
2 changes: 1 addition & 1 deletion docs/runs.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ collections are available via `actor.runs()` and `task.runs()`.
| `reboot()` | — | `ActorRun` | Reboots the run's container. |
| `resurrect(options)` | `RunResurrectOptions` | `ActorRun` | Resurrects a finished run. |
| `charge(options)` | `RunChargeOptions` | `()` | Charges a pay-per-event run (always sends an idempotency key). |
| `wait_for_finish(wait_secs)` | `Option<i64>` | `ActorRun` | Polls until the run is terminal. `None` waits indefinitely; `Some(n)` bounds the wait and may return a still-running (non-terminal) run if `n` elapses first. |
| `wait_for_finish(wait_secs)` | `Option<i64>` | `ActorRun` | Polls until the run is terminal. `None` waits indefinitely; `Some(n)` bounds the wait and may return a still-running (non-terminal) run if `n` elapses first. A run ID that keeps 404ing past a 3-second grace period (reset by any successful fetch) returns the `404` as an error instead of polling forever. |
| `dataset()` / `key_value_store()` / `request_queue()` / `log()` | — | resource client | Access the run's default storages and log. |
| `get_streamed_log()` | — | `Result<impl Stream<Item = Result<Vec<u8>>>>` (async — `.await` it) | Convenience for `log().stream()` — streams the run's log chunks live (log redirection). |
| `get_streamed_log_with_options(options)` | `LogOptions` | `Result<impl Stream<Item = Result<Vec<u8>>>>` (async — `.await` it) | As `get_streamed_log()`, forwarding `LogOptions` (e.g. `raw`) to the log stream. |
Expand Down
40 changes: 39 additions & 1 deletion src/clients/actor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,17 @@ use crate::error::ApifyClientResult;
use crate::http_client::HttpClient;
use crate::models::{Actor, ActorRun, Build};

/// How long to keep retrying a run start that the API rejects for lack of resources (see
/// [`ActorStartOptions::wait_for_resources`]).
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum WaitForResources {
/// Retry until the run starts, how ever long that takes.
Forever,
/// Retry for up to this many seconds (clamped to `0` if negative), then return the last
/// error. `0` makes a single attempt and returns its error without retrying.
ForSecs(f64),
}

/// Options shared by [`ActorClient::start`] and [`ActorClient::call`] (and the task
/// equivalents).
#[derive(Debug, Default, Clone)]
Expand All @@ -44,6 +55,19 @@ pub struct ActorStartOptions {
/// Ad-hoc webhooks to attach to this run. Serialized to base64-encoded JSON as the
/// `webhooks` query parameter, matching the reference clients.
pub webhooks: Option<Vec<serde_json::Value>>,
/// Retry the start while the account lacks the resources for the run, i.e. while the API
/// rejects it with an [`ApiError`](crate::ApiError) whose `error_type` is
/// `actor-memory-limit-exceeded` or `concurrent-runs-limit-exceeded`. Both clear as other
/// runs or builds finish. The start is retried every 10 seconds; any other error is returned
/// right away.
///
/// [`WaitForResources::Forever`] retries until the run starts.
/// [`WaitForResources::ForSecs`] stops retrying after that many seconds and returns the last
/// error. `None` (the default) makes a single attempt.
///
/// A run that requests more memory than the account's whole memory limit is rejected with
/// `actor-memory-limit-exceeded` as well and never starts, so `Forever` retries it forever.
pub wait_for_resources: Option<WaitForResources>,
}

impl ActorStartOptions {
Expand Down Expand Up @@ -125,6 +149,8 @@ impl ActorClient {
///
/// `input` is any JSON-serializable value (or `None` for no input). To send a non-JSON
/// input (e.g. a ZIP archive) as raw bytes instead, use [`start_raw`](Self::start_raw).
/// Set [`ActorStartOptions::wait_for_resources`] to retry the start while the account lacks
/// the memory or a concurrent-run slot for it.
pub async fn start<T: Serialize>(
&self,
input: Option<&T>,
Expand Down Expand Up @@ -169,7 +195,16 @@ impl ActorClient {
.content_type
.clone()
.unwrap_or_else(|| default_content_type.to_string());
post_with_body(&self.ctx, Some("runs"), &params, body, &content_type).await
crate::clients::base::retry_while_resource_limited(options.wait_for_resources, || {
post_with_body(
&self.ctx,
Some("runs"),
&params,
body.clone(),
&content_type,
)
})
.await
}

/// Starts the Actor and waits (client-side polling) for it to finish.
Expand All @@ -179,6 +214,9 @@ impl ActorClient {
/// - `Some(n)` bounds the wait to roughly `n` seconds; if the run has not finished by
/// then, the **last fetched (still non-terminal) run is returned** rather than an
/// error. Check `status` / `is_terminal()` on the result when using `Some`.
///
/// Time spent retrying under [`ActorStartOptions::wait_for_resources`] does not count
/// toward `wait_secs`, since that budget only starts once the run has actually started.
pub async fn call<T: Serialize>(
&self,
input: Option<&T>,
Expand Down
109 changes: 100 additions & 9 deletions src/clients/base.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,13 @@ const WAIT_FOR_FINISH_POLL_INTERVAL: Duration = Duration::from_millis(250);
/// The `waitForFinish` value (in seconds) sent on each poll; the API caps server-side
/// waiting at 60 seconds, so we poll in chunks of this size.
const WAIT_FOR_FINISH_REQUEST_SECS: i64 = 60;
/// How long a run/build may keep answering `404` before [`wait_for_finish`] gives up and returns
/// a not-found error, instead of polling forever. A brand-new run/build can briefly 404 while the
/// write that created it has not yet reached the replica a poll happens to land on; a success
/// within this window resets it. Matches the reference clients' `DEFAULT_WAIT_WHEN_JOB_NOT_EXIST`.
/// Without this bound, a run/build ID that was simply never valid (e.g. a typo, or the resource
/// was deleted) would poll indefinitely when `wait_secs` is `None`.
const NOT_FOUND_GRACE_PERIOD: Duration = Duration::from_secs(3);

/// Default per-request timeout used when an endpoint does not specify its own (6 minutes).
/// This is the single source of truth for the base request timeout; clients that issue raw
Expand Down Expand Up @@ -427,8 +434,11 @@ where
T: DeserializeOwned,
F: Fn(&T) -> bool,
{
let start = std::time::Instant::now();
let start = tokio::time::Instant::now();
let budget = wait_secs.map(|s| Duration::from_secs(s.max(0) as u64));
// When the fetch below starts hitting `404`, the instant the *first* one was seen; reset on
// any successful fetch. `None` means the resource has not 404'd (yet).
let mut not_found_since: Option<tokio::time::Instant> = None;

loop {
let remaining_request_secs = match budget {
Expand All @@ -449,20 +459,101 @@ where
let mut params = QueryParams::new();
params.add_int("waitForFinish", Some(remaining_request_secs));

let resource: Option<T> = get_resource(ctx, None, &params).await?;
if let Some(resource) = resource {
if is_terminal(&resource) {
return Ok(resource);
}
// Not finished yet. If we have no budget left, return what we have.
if let Some(budget) = budget {
if start.elapsed() >= budget {
match get_resource_required::<T>(ctx, None, &params).await {
Ok(resource) => {
not_found_since = None;
if is_terminal(&resource) {
return Ok(resource);
}
// Not finished yet. If we have no budget left, return what we have.
if let Some(budget) = budget {
if start.elapsed() >= budget {
return Ok(resource);
}
}
}
Err(err) => {
let is_not_found = err.as_api_error().is_some_and(|e| e.status_code == 404);
if !is_not_found {
return Err(err);
}
// A run/build that keeps 404ing past the grace period was never going to appear
// (bad ID, or it was deleted); without this, `wait_secs: None` would poll it
// forever instead of surfacing the not-found error.
let now = tokio::time::Instant::now();
let first_seen = *not_found_since.get_or_insert(now);
if now.duration_since(first_seen) >= NOT_FOUND_GRACE_PERIOD {
return Err(err);
}
}
}

// Brief pause to let replicas catch up, then poll again.
crate::http_client::sleep_public(WAIT_FOR_FINISH_POLL_INTERVAL).await;
}
}

/// Cooldown between retries of a run start the API rejected for lack of resources, matching the
/// reference clients.
const WAIT_FOR_RESOURCES_COOLDOWN: Duration = Duration::from_secs(10);
/// `ApiError::error_type` values the API rejects a run start with while the account has no free
/// memory or concurrent-run slot for it. Both clear as other runs/builds finish.
const RESOURCE_LIMIT_ERROR_TYPES: [&str; 2] = [
"actor-memory-limit-exceeded",
"concurrent-runs-limit-exceeded",
];

/// Retries `attempt` while it fails with a resource-limit [`ApiError`](crate::ApiError), per
/// `wait_for_resources` (see [`WaitForResources`](crate::clients::actor::WaitForResources)).
/// `None` makes exactly one attempt. Any other error is returned immediately, unretried. Shared
/// by [`ActorClient::start`](crate::clients::actor::ActorClient::start) and
/// [`TaskClient::start`](crate::clients::task::TaskClient::start), whose run-start options and
/// retry behavior are otherwise identical (the don't-repeat-yourself goal of this module).
pub(crate) async fn retry_while_resource_limited<F, Fut, T>(
wait_for_resources: Option<crate::clients::actor::WaitForResources>,
mut attempt: F,
) -> ApifyClientResult<T>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = ApifyClientResult<T>>,
{
use crate::clients::actor::WaitForResources;

let Some(wait) = wait_for_resources else {
return attempt().await;
};
let deadline = match wait {
WaitForResources::Forever => None,
WaitForResources::ForSecs(secs) => {
Some(tokio::time::Instant::now() + Duration::from_secs_f64(secs.max(0.0)))
}
};

loop {
let err = match attempt().await {
Ok(value) => return Ok(value),
Err(err) => err,
};
let is_resource_limited = err.as_api_error().is_some_and(|api_err| {
api_err
.error_type
.as_deref()
.is_some_and(|t| RESOURCE_LIMIT_ERROR_TYPES.contains(&t))
});
if !is_resource_limited {
return Err(err);
}

let cooldown = match deadline {
Some(deadline) => {
let now = tokio::time::Instant::now();
if now >= deadline {
return Err(err);
}
WAIT_FOR_RESOURCES_COOLDOWN.min(deadline - now)
}
None => WAIT_FOR_RESOURCES_COOLDOWN,
};
crate::http_client::sleep_public(cooldown).await;
}
}
Loading
Loading