diff --git a/CHANGELOG.md b/CHANGELOG.md index 685eee4..8daf21e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/Cargo.toml b/Cargo.toml index 6b1da0a..bb38e74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apify-client" -version = "0.11.0" +version = "0.12.0" authors = ["Apify Technologies "] description = "An official, but experimental, AI-generated and AI-maintained Rust client for the Apify API (https://apify.com)." license = "Apache-2.0" @@ -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] diff --git a/docs/README.md b/docs/README.md index f1273d6..99070cf 100644 --- a/docs/README.md +++ b/docs/README.md @@ -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}; diff --git a/docs/actors.md b/docs/actors.md index 1926457..3358cbb 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -51,6 +51,7 @@ All fields are optional. Used by both `start` and `call` here, and by the identi | `restart_on_error` | `Option` | Whether to restart the run if it fails. | | `force_permission_level` | `Option` | Override the Actor's permission level for this run. | | `webhooks` | `Option>` | 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` | 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: @@ -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> { +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::( + 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 diff --git a/docs/builds.md b/docs/builds.md index f6b19d4..229eb09 100644 --- a/docs/builds.md +++ b/docs/builds.md @@ -17,7 +17,7 @@ collections are available via `actor.builds()`. | `get()` | — | `Option` | Fetches the build. | | `abort()` | — | `Build` | Aborts the build. | | `delete()` | — | `()` | Deletes the build. | -| `wait_for_finish(wait_secs)` | `Option` | `Build` | Polls until the build is terminal. | +| `wait_for_finish(wait_secs)` | `Option` | `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` | Fetches the OpenAPI definition generated for the build (raw JSON, endpoint `.../openapi.json`). | | `log()` | — | `LogClient` | Access the build's log. | diff --git a/docs/runs.md b/docs/runs.md index 09ade49..6115675 100644 --- a/docs/runs.md +++ b/docs/runs.md @@ -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` | `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` | `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>>>` (async — `.await` it) | Convenience for `log().stream()` — streams the run's log chunks live (log redirection). | | `get_streamed_log_with_options(options)` | `LogOptions` | `Result>>>` (async — `.await` it) | As `get_streamed_log()`, forwarding `LogOptions` (e.g. `raw`) to the log stream. | diff --git a/src/clients/actor.rs b/src/clients/actor.rs index 4539ca5..6f9f5a1 100644 --- a/src/clients/actor.rs +++ b/src/clients/actor.rs @@ -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)] @@ -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>, + /// 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, } impl ActorStartOptions { @@ -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( &self, input: Option<&T>, @@ -169,7 +195,16 @@ impl ActorClient { .content_type .clone() .unwrap_or_else(|| default_content_type.to_string()); - post_with_body(&self.ctx, Some("runs"), ¶ms, body, &content_type).await + crate::clients::base::retry_while_resource_limited(options.wait_for_resources, || { + post_with_body( + &self.ctx, + Some("runs"), + ¶ms, + body.clone(), + &content_type, + ) + }) + .await } /// Starts the Actor and waits (client-side polling) for it to finish. @@ -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( &self, input: Option<&T>, diff --git a/src/clients/base.rs b/src/clients/base.rs index 65e258f..23b236e 100644 --- a/src/clients/base.rs +++ b/src/clients/base.rs @@ -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 @@ -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 = None; loop { let remaining_request_secs = match budget { @@ -449,16 +459,32 @@ where let mut params = QueryParams::new(); params.add_int("waitForFinish", Some(remaining_request_secs)); - let resource: Option = get_resource(ctx, None, ¶ms).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::(ctx, None, ¶ms).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); + } } } @@ -466,3 +492,68 @@ where 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( + wait_for_resources: Option, + mut attempt: F, +) -> ApifyClientResult +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + 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; + } +} diff --git a/src/clients/dataset.rs b/src/clients/dataset.rs index 70b7a37..2d2b57f 100644 --- a/src/clients/dataset.rs +++ b/src/clients/dataset.rs @@ -210,9 +210,9 @@ impl DatasetClient { let items: Vec = serde_json::from_slice(&response.body)?; let count = items.len() as i64; - // Fall back to `0` ("total unknown"), never `count`: a total equal to the items already - // returned would look complete and stop iteration after page one, dropping later items. - // `0` routes iteration to the short-page/empty-page backstop, which walks every page. + // `PaginationList::total` is informational only here (`ListIterator` never consults it, + // see `scanned` below); falling back to `0` rather than `count` just avoids implying a + // precise total when the API did not report one. let total = response .header("x-apify-pagination-total") .and_then(|v| v.parse().ok()) @@ -225,6 +225,12 @@ impl DatasetClient { .header("x-apify-pagination-limit") .and_then(|v| v.parse().ok()) .unwrap_or(count); + // The number of rows scanned to produce this page, when the API reports it separately + // from the item count (see `PaginationList::scanned`): a filter can scan more rows than + // it returns, and `unwind` can return more items than rows scanned. + let scanned = response + .header("x-apify-pagination-count") + .and_then(|v| v.parse().ok()); Ok(PaginationList { total, @@ -233,6 +239,7 @@ impl DatasetClient { count, desc: options.desc.unwrap_or(false), items, + scanned, }) } @@ -243,22 +250,17 @@ impl DatasetClient { /// transparently paging. The caller's `options.limit` caps the total number of items yielded /// (unset = all); use [`ListIterator::with_chunk_size`] to control the per-page fetch size. /// - /// Server-side filters (`skip_empty`/`clean`/`skip_hidden`) are forwarded on every page - /// request. Paging advances the offset by the number of items each page returns, exactly - /// like the reference JavaScript client (`currentOffset += items.length`). Because the offset - /// advances by the post-filter count rather than the raw window size, filtered iteration is - /// not exact — this matches the reference client, and it can distort results two ways: + /// Server-side filters (`skip_empty`/`clean`/`skip_hidden`) and `unwind` are forwarded on + /// every page request. Paging advances the offset by the number of underlying rows each page + /// *scanned* (the `x-apify-pagination-count` response header), not by `items.len()`, matching + /// the reference JavaScript client: a filter can scan more rows than it returns, and `unwind` + /// can return more items than rows scanned, so advancing by the returned count alone would + /// re-scan or skip rows. The listing's reported `total` is never consulted, so a dataset a + /// still-running Actor keeps pushing to is read to the end: iteration keeps requesting pages + /// until one scans no rows (one extra, empty request past the last page with data), rather + /// than stopping at the size the dataset had when the loop started. /// - /// - Duplicates: when a page's filtered count is smaller than its raw window, the next page - /// starts at an offset that overlaps the previous window, so some items are yielded more - /// than once. - /// - Dropped items: when a filter removes *every* item in a raw window, the page comes back - /// with no items; iteration treats that empty page as the end of the dataset (the empty-page - /// backstop in [`ListIterator`]) and stops, even though unfiltered items still exist at - /// higher offsets. - /// - /// If you need every filtered item exactly once, apply the filter client-side over an - /// unfiltered iteration instead. + /// If `x-apify-pagination-count` is ever absent, the scanned count falls back to `items.len()`. pub fn iterate_items( &self, options: DatasetListItemsOptions, diff --git a/src/clients/pagination.rs b/src/clients/pagination.rs index d024244..ca5ad87 100644 --- a/src/clients/pagination.rs +++ b/src/clients/pagination.rs @@ -177,48 +177,48 @@ impl ListIterator { let page = (self.fetch)(self.next_offset, page_limit).await?; let received = page.items.len() as i64; + // The number of rows this page *scanned*, used to advance the offset and decide when the + // listing is exhausted. For every endpoint except dataset items this equals `received` + // (`page.scanned` is `None`); dataset items can report it separately (see + // `PaginationList::scanned`), since a filter can scan more rows than it returns and + // `unwind` can return more items than rows scanned. It is capped at the page size actually + // requested so an unwound page never advances past rows the next request would then skip. + let scanned = page.scanned.unwrap_or(received).max(received); + let scanned = match page_limit { + Some(limit) if limit > 0 => scanned.min(limit), + _ => scanned, + }; + // Enforce the caller's total-item cap exactly, even if the API returns more than the // requested page limit. let mut items = page.items; if let Some(rem) = self.remaining { if received > rem { - items.truncate(rem as usize); + items.truncate(rem.max(0) as usize); } } - self.next_offset += received; + self.next_offset += scanned; if let Some(rem) = self.remaining.as_mut() { - *rem -= received; + *rem -= scanned; } - // Decide whether more pages remain. + // Decide whether more pages remain. `total` is deliberately never consulted: it changes + // as the listed resource grows (e.g. a dataset a still-running Actor keeps pushing to), + // so stopping at the `total` a page happened to report can end the walk before rows added + // later are ever read. A short page is not treated as terminal either (server-side filters + // can legitimately make a full, non-final window come back short) — only a page that + // scanned nothing, or the caller's own cap, ends the walk. This matches the reference + // client, which pages until a request scans zero rows and so always sends one extra + // request at the end to find that empty page. if self.single_page { // Non-paginated endpoint: everything came back in one response. self.exhausted = true; - } else if received == 0 { - // Empty page: nothing more to read. This is the primary backstop and matches the - // reference client, whose loop stops as soon as a page returns no items. + } else if scanned == 0 { self.exhausted = true; } else if matches!(self.remaining, Some(r) if r <= 0) { // Reached the caller's total-item cap. self.exhausted = true; - } else if page.total > 0 { - // The endpoint reports a usable total, so drive termination by position, like the - // reference `_listPaginatedFromCallback`. A short page is deliberately NOT treated as - // terminal here: with dataset item filters (`skip_empty`/`clean`/`skip_hidden`) a full, - // non-final window can return fewer items than requested while more remain at higher - // offsets, so short-page detection would silently truncate. The empty-page backstop - // above ends the walk instead. - if self.next_offset >= page.total { - self.exhausted = true; - } - } else { - // No usable total (endpoint reports `total == 0`): fall back to short-page detection — - // a page shorter than the size the API says it served (`page.limit`) is the last one. - // A non-positive reported limit means the endpoint is not offset-paginated at all. - if page.limit <= 0 || received < page.limit { - self.exhausted = true; - } } self.buffer.extend(items); @@ -272,13 +272,14 @@ mod tests { count: items.len() as i64, desc: false, items, + scanned: None, }) }) }) } #[tokio::test] - async fn walks_all_pages_using_reported_total() { + async fn ignores_reported_total_and_pages_to_an_empty_page() { let calls = Arc::new(AtomicUsize::new(0)); let iter = ListIterator::new( 0, @@ -287,14 +288,15 @@ mod tests { ) .with_chunk_size(2); assert_eq!(iter.collect_all().await.unwrap(), vec![0, 1, 2, 3, 4]); - // Pages: [0,1] [2,3] [4]. next_offset reaches total (5) on the third page, so no extra - // empty request. - assert_eq!(calls.load(Ordering::SeqCst), 3); + // Pages: [0,1] [2,3] [4] [] — the third page is short (1 < 2) but not terminal, and + // `total` (reported as 5, reached by offset after the third page) is not consulted + // either, so a fourth, empty request is what actually ends the walk. + assert_eq!(calls.load(Ordering::SeqCst), 4); } #[tokio::test] async fn walks_all_pages_when_total_is_zero() { - // Emulates an endpoint that does not report a usable total: short-page detection ends it. + // Emulates an endpoint that does not report a usable total: the trailing empty page ends it. let calls = Arc::new(AtomicUsize::new(0)); let iter = ListIterator::new(0, None, slicing_fetcher((0..5).collect(), 2, false, calls)) .with_chunk_size(2); @@ -316,8 +318,10 @@ mod tests { } #[tokio::test] - async fn reported_total_avoids_extra_request_on_exact_multiple() { - // 4 items, page size 2, total known: stops after the second full page (no empty fetch). + async fn reported_total_does_not_skip_the_trailing_empty_request() { + // 4 items, page size 2, total known and an exact multiple of the page size: even though + // offset reaches `total` right after the second (full) page, that is not consulted, so a + // third, empty request is still made before the walk ends. let calls = Arc::new(AtomicUsize::new(0)); let iter = ListIterator::new( 0, @@ -326,15 +330,120 @@ mod tests { ) .with_chunk_size(2); assert_eq!(iter.collect_all().await.unwrap(), vec![0, 1, 2, 3]); - assert_eq!(calls.load(Ordering::SeqCst), 2); + assert_eq!(calls.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn keeps_paging_past_the_first_pages_total_as_the_listing_grows() { + // The listing holds 2 items when the first page is read and 4 from then on (e.g. a + // dataset a still-running Actor keeps pushing to). Each response reports the *current* + // total, which would wrongly look "complete" at the end of page one if total were + // consulted; ignoring it lets the walk pick up the rows added afterwards. + let calls = Arc::new(AtomicUsize::new(0)); + let counter = calls.clone(); + let fetch: PageFetcher = Box::new(move |offset, limit| { + let call_number = counter.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { + let total_rows: i64 = if call_number == 0 { 2 } else { 4 }; + let page_size = limit.unwrap_or(2); + let start = (offset.max(0)).min(total_rows); + let end = (start + page_size).min(total_rows); + let items: Vec = (start..end).collect(); + Ok(PaginationList { + total: total_rows, + offset, + limit: page_size, + count: items.len() as i64, + desc: false, + items, + scanned: None, + }) + }) + }); + let iter = ListIterator::new(0, None, fetch).with_chunk_size(2); + assert_eq!(iter.collect_all().await.unwrap(), vec![0, 1, 2, 3]); + // Pages: [0,1] (total looked like 2) [2,3] (total now 4) [] — three requests. + assert_eq!(calls.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn advances_by_the_scanned_count_when_it_exceeds_items_returned() { + // Models a dataset `listItems` filter: each page scans 3 rows (`x-apify-pagination-count`) + // but a `skip_empty`-style filter leaves only 1 item per page. Advancing by `items.len()` + // alone would re-request the same rows forever; advancing by the scanned count moves past + // them. + let calls = Arc::new(AtomicUsize::new(0)); + let counter = calls.clone(); + let fetch: PageFetcher = Box::new(move |offset, _limit| { + counter.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { + // 9 raw rows in windows of 3, only the first of each window "survives" the filter. + let window_start = (offset.max(0) / 3) * 3; + let items = if window_start < 9 { + vec![window_start] + } else { + vec![] + }; + let scanned = if window_start < 9 { 3 } else { 0 }; + Ok(PaginationList { + total: 0, + offset, + limit: 3, + count: items.len() as i64, + desc: false, + items, + scanned: Some(scanned), + }) + }) + }); + let iter = ListIterator::new(0, None, fetch).with_chunk_size(3); + assert_eq!(iter.collect_all().await.unwrap(), vec![0, 3, 6]); + // Offsets requested: 0, 3, 6, 9 (empty) — four calls, each advancing by the scanned 3 + // rows rather than the 1 surviving item. + assert_eq!(calls.load(Ordering::SeqCst), 4); + } + + #[tokio::test] + async fn scanned_count_is_capped_at_the_requested_page_size() { + // A lagging `x-apify-pagination-count` header can occasionally overshoot what was asked + // for (e.g. counting rows added by a concurrent push); the cap keeps the offset from + // jumping past rows the next request would then never read. + let calls = Arc::new(AtomicUsize::new(0)); + let counter = calls.clone(); + let fetch: PageFetcher = Box::new(move |offset, limit| { + counter.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { + let page_size = limit.unwrap_or(2); + let start = offset.max(0); + let end = (start + page_size).min(6); + let items: Vec = (start..end).collect(); + // Reports scanning more rows than were actually requested/returned, but only + // while there is anything left to scan (a genuinely empty page reports 0, same + // as `items.len()`, so the walk can still terminate). + let scanned = if items.is_empty() { 0 } else { page_size + 5 }; + Ok(PaginationList { + total: 0, + offset, + limit: page_size, + count: items.len() as i64, + desc: false, + items, + scanned: Some(scanned), + }) + }) + }); + let iter = ListIterator::new(0, None, fetch).with_chunk_size(2); + // Without the cap this would skip straight past offset 6 and lose items [2..6). + assert_eq!(iter.collect_all().await.unwrap(), vec![0, 1, 2, 3, 4, 5]); + assert_eq!(calls.load(Ordering::SeqCst), 4); } #[tokio::test] async fn non_final_short_page_with_reported_total_does_not_truncate() { // Guards the termination logic, NOT filter de-duplication: every page is "short" (it - // returns fewer items than the page size the API reports) while the endpoint reports a - // large total. Total-driven termination must keep going and yield every item, ending - // only on the empty page. (This fetcher advances offset by the page size and never + // returns fewer items than the page size the API reports), and the endpoint also reports + // a large `total`. Neither is consulted, so the walk must keep going and yield every item, + // ending only on the empty page. (This fetcher advances offset by the page size and never // overlaps windows, so it does not model the sparse-window duplicate behaviour // documented on `iterate_items`; it only proves a short page is not treated as terminal.) let all: Vec = (0..6).collect(); @@ -346,8 +455,9 @@ mod tests { let all = all.clone(); Box::pin(async move { // Serve at most 2 items per page but report a page limit of 4, so every non-final - // page is strictly short (received 2 < reported limit 4). `total` is reported large - // (100) so termination can only come from the empty page, never a short page. + // page is strictly short (received 2 < reported limit 4), and `total` is reported + // large (100) — neither is consulted, so termination can only come from the empty + // page, never a short page or a reported total. let start = (offset.max(0) as usize).min(all.len()); let end = (start + 2).min(all.len()); let items = all[start..end].to_vec(); @@ -358,6 +468,7 @@ mod tests { count: items.len() as i64, desc: false, items, + scanned: None, }) }) }); @@ -417,6 +528,7 @@ mod tests { count: 5, desc: false, items: vec![0, 1, 2, 3, 4], + scanned: None, }) }) }); @@ -453,6 +565,7 @@ mod tests { count: 3, desc: false, items: vec![10, 20, 30], + scanned: None, }) }) }); diff --git a/src/clients/task.rs b/src/clients/task.rs index 7670340..99a38ca 100644 --- a/src/clients/task.rs +++ b/src/clients/task.rs @@ -70,7 +70,9 @@ impl TaskClient { /// Starts the task and returns immediately with the created run. /// - /// `input` overrides the task's saved input (or `None` to use the saved input). + /// `input` overrides the task's saved input (or `None` to use the saved input). 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( &self, input: Option<&T>, @@ -82,7 +84,16 @@ impl TaskClient { Some(value) => Some(serde_json::to_vec(value)?), None => None, }; - post_with_body(&self.ctx, Some("runs"), ¶ms, body, "application/json").await + crate::clients::base::retry_while_resource_limited(options.wait_for_resources, || { + post_with_body( + &self.ctx, + Some("runs"), + ¶ms, + body.clone(), + "application/json", + ) + }) + .await } /// Starts the task and waits (client-side polling) for it to finish. @@ -92,6 +103,9 @@ impl TaskClient { /// - `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( &self, input: Option<&T>, diff --git a/src/common.rs b/src/common.rs index bc20935..8f92da7 100644 --- a/src/common.rs +++ b/src/common.rs @@ -222,6 +222,19 @@ pub struct PaginationList { /// The items of this page. #[serde(default = "Vec::new")] pub items: Vec, + /// Number of underlying rows the endpoint scanned to produce this page, when that can + /// differ from `items.len()`. Currently only [`DatasetClient::list_items`] + /// (`x-apify-pagination-count` header) sets this: a `clean`/`skip_empty`/`skip_hidden` + /// filter can scan more rows than it returns, and `unwind` can return more items than rows + /// scanned. [`ListIterator`] uses it (capped at the requested page size) instead of + /// `items.len()` to advance the offset and decide when a listing is exhausted, so paging + /// neither re-scans nor skips rows. `None` for every other endpoint, where the two always + /// agree. + /// + /// [`DatasetClient::list_items`]: crate::clients::dataset::DatasetClient::list_items + /// [`ListIterator`]: crate::clients::pagination::ListIterator + #[serde(skip)] + pub(crate) scanned: Option, } /// Reports whether the environment variable `name` is set to a non-empty value. diff --git a/src/lib.rs b/src/lib.rs index 561be02..529d8eb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -68,7 +68,7 @@ pub use http_client::RequestCompression; pub use version::{API_SPEC_VERSION, CLIENT_VERSION}; // Re-export the most commonly used option/parameter types for ergonomic access. -pub use clients::actor::{ActorBuildOptions, ActorStartOptions}; +pub use clients::actor::{ActorBuildOptions, ActorStartOptions, WaitForResources}; pub use clients::actor_collection::ActorListOptions; pub use clients::dataset::{DatasetDownloadOptions, DatasetListItemsOptions, DownloadItemsFormat}; pub use clients::key_value_store::{GetRecordOptions, KeyValueStoreKeysIterator, ListKeysOptions}; diff --git a/src/version.rs b/src/version.rs index 6efcd23..f81ec44 100644 --- a/src/version.rs +++ b/src/version.rs @@ -10,4 +10,4 @@ pub const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION"); /// and verified against. /// /// This corresponds to the `info.version` field of the Apify OpenAPI document. -pub const API_SPEC_VERSION: &str = "v2-2026-10-01T153946Z"; +pub const API_SPEC_VERSION: &str = "v2-2026-10-08T073338Z"; diff --git a/tests/unit_http.rs b/tests/unit_http.rs index 7529d4f..991dcb9 100644 --- a/tests/unit_http.rs +++ b/tests/unit_http.rs @@ -8,7 +8,10 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use apify_client::http_client::{HttpBackend, HttpRequest, HttpResponse}; -use apify_client::{ApifyClient, ApifyClientError, LastRunOptions, RequestCompression}; +use apify_client::{ + ActorStartOptions, ApifyClient, ApifyClientError, LastRunOptions, RequestCompression, + WaitForResources, +}; use async_trait::async_trait; /// A scripted backend that returns a queued sequence of responses and counts calls. @@ -811,9 +814,9 @@ async fn iterate_macro_caps_and_pages_correctly() { /// Hermetic coverage of `RunCollectionClient::iterate`, which builds its iterator directly /// (not via the `list_iterator!` macro) because it threads a separate `filter` argument. The -/// iterator must walk every page using the reported total for termination and must forward the -/// `status` filter on every page request. Offline-only; the integration suite gates this on a -/// live token. +/// iterator must walk every page — ignoring the reported `total`, which can grow as more runs +/// start — until a page scans no rows, and must forward the `status` filter on every page +/// request. Offline-only; the integration suite gates this on a live token. #[tokio::test] async fn run_collection_iterate_walks_pages_and_forwards_filter() { let backend = MockBackend::new(vec![ @@ -825,6 +828,10 @@ async fn run_collection_iterate_walks_pages_and_forwards_filter() { 200, br#"{"data":{"total":3,"limit":2,"items":[{"id":"r2"}]}}"#.to_vec(), ), + MockOutcome::Status( + 200, + br#"{"data":{"total":3,"limit":2,"items":[]}}"#.to_vec(), + ), ]); let client = client_with(backend.clone(), 0); @@ -845,7 +852,11 @@ async fn run_collection_iterate_walks_pages_and_forwards_filter() { vec!["r0", "r1", "r2"], "all runs across pages must be yielded" ); - assert_eq!(backend.call_count(), 2, "two pages then total-driven stop"); + assert_eq!( + backend.call_count(), + 3, + "two data pages then the terminating empty page; `total` is never consulted" + ); for url in backend.urls() { assert!( url.contains("status=SUCCEEDED"), @@ -1120,3 +1131,233 @@ async fn iterate_keys_large_cap_clamps_page_size() { urls[1] ); } + +/// `wait_for_finish` must not poll a persistently-missing run forever when `wait_secs` is `None`: +/// past the 3-second not-found grace period it surfaces the `404` as an error instead. Uses +/// paused Tokio time so the grace period and poll interval elapse instantly in wall-clock time. +#[tokio::test(start_paused = true)] +async fn wait_for_finish_errors_after_persistent_not_found() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 404, + br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(), + )]); + let client = client_with(backend.clone(), 0); + + let err = client + .run("ghost-run-id") + .wait_for_finish(None) + .await + .expect_err("a run that never appears must error out, not poll forever"); + let api = err.as_api_error().expect("api error"); + assert!(api.is_not_found()); + assert!( + backend.call_count() > 1, + "must have retried within the grace period, not failed on the very first 404" + ); +} + +/// A run/build that 404s once (e.g. a replica lagging just behind the write that created it) and +/// then starts answering normally must not be treated as persistently missing: the grace-period +/// clock resets on the first successful fetch. +#[tokio::test(start_paused = true)] +async fn wait_for_finish_recovers_from_a_transient_not_found() { + let backend = MockBackend::new(vec![ + MockOutcome::Status( + 404, + br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(), + ), + MockOutcome::Status( + 200, + br#"{"data":{"id":"run-id","status":"SUCCEEDED"}}"#.to_vec(), + ), + ]); + let client = client_with(backend.clone(), 0); + + let run = client + .run("run-id") + .wait_for_finish(None) + .await + .expect("a transient 404 followed by success must not error"); + assert_eq!(run.status.as_deref(), Some("SUCCEEDED")); +} + +/// `ActorStartOptions::wait_for_resources` retries a start rejected for lack of resources +/// (`actor-memory-limit-exceeded` / `concurrent-runs-limit-exceeded`) every 10 seconds until it +/// succeeds. Paused time collapses the two 10-second cooldowns to no wall-clock time. +#[tokio::test(start_paused = true)] +async fn wait_for_resources_retries_until_success() { + let backend = + MockBackend::new(vec![ + MockOutcome::Status( + 402, + br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"# + .to_vec(), + ), + MockOutcome::Status( + 402, + br#"{"error":{"type":"concurrent-runs-limit-exceeded","message":"too many runs"}}"# + .to_vec(), + ), + MockOutcome::Status(201, br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec()), + ]); + let client = client_with(backend.clone(), 0); + + let run = client + .actor("some-actor") + .start::( + None, + ActorStartOptions { + wait_for_resources: Some(WaitForResources::Forever), + ..Default::default() + }, + ) + .await + .expect("must eventually succeed once resources free up"); + assert_eq!(run.id, "run-id"); + assert_eq!(backend.call_count(), 3, "two rejections, then the success"); +} + +/// Without `wait_for_resources`, a resource-limit rejection is returned immediately, unretried — +/// matching every other `ApiError`. +#[tokio::test] +async fn wait_for_resources_defaults_to_a_single_attempt() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 402, + br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"# + .to_vec(), + )]); + let client = client_with(backend.clone(), 0); + + let err = client + .actor("some-actor") + .start::(None, ActorStartOptions::default()) + .await + .expect_err("must fail without the option"); + assert_eq!( + err.as_api_error().unwrap().error_type.as_deref(), + Some("actor-memory-limit-exceeded") + ); + assert_eq!(backend.call_count(), 1); +} + +/// `WaitForResources::ForSecs` bounds the retrying: attempts happen at 0s, 10s, 20s and 25s (the +/// last cooldown cut short by the bound), then the last rejection is returned. +#[tokio::test(start_paused = true)] +async fn wait_for_resources_for_secs_bounds_retrying_and_returns_last_error() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 402, + br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"# + .to_vec(), + )]); + let client = client_with(backend.clone(), 0); + + let err = client + .actor("some-actor") + .start::( + None, + ActorStartOptions { + wait_for_resources: Some(WaitForResources::ForSecs(25.0)), + ..Default::default() + }, + ) + .await + .expect_err("must give up once the bound elapses"); + assert_eq!( + err.as_api_error().unwrap().error_type.as_deref(), + Some("actor-memory-limit-exceeded") + ); + assert_eq!(backend.call_count(), 4, "attempts at 0s, 10s, 20s and 25s"); +} + +/// Any error other than a resource-limit rejection is returned right away, never retried, even +/// with `wait_for_resources` set. +#[tokio::test] +async fn wait_for_resources_does_not_retry_unrelated_errors() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 400, + br#"{"error":{"type":"invalid-input","message":"Input is not valid."}}"#.to_vec(), + )]); + let client = client_with(backend.clone(), 0); + + let err = client + .actor("some-actor") + .start::( + None, + ActorStartOptions { + wait_for_resources: Some(WaitForResources::Forever), + ..Default::default() + }, + ) + .await + .expect_err("a non-resource-limit error must not be retried"); + assert_eq!( + err.as_api_error().unwrap().error_type.as_deref(), + Some("invalid-input") + ); + assert_eq!(backend.call_count(), 1); +} + +/// `TaskClient::start` shares the same `wait_for_resources` retry behavior as `ActorClient::start` +/// (both funnel through the same helper — DRY). +#[tokio::test(start_paused = true)] +async fn task_wait_for_resources_retries_until_success() { + let backend = MockBackend::new(vec![ + MockOutcome::Status( + 402, + br#"{"error":{"type":"concurrent-runs-limit-exceeded","message":"too many runs"}}"# + .to_vec(), + ), + MockOutcome::Status( + 201, + br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec(), + ), + ]); + let client = client_with(backend.clone(), 0); + + let run = client + .task("some-task") + .start::( + None, + ActorStartOptions { + wait_for_resources: Some(WaitForResources::Forever), + ..Default::default() + }, + ) + .await + .expect("must eventually succeed once resources free up"); + assert_eq!(run.id, "run-id"); + assert_eq!(backend.call_count(), 2); +} + +/// Time spent retrying a resource-limited start inside `call()` must not count against +/// `wait_secs`: that budget only starts once the run has actually started. +#[tokio::test(start_paused = true)] +async fn call_wait_for_resources_does_not_count_toward_wait_secs() { + let backend = + MockBackend::new(vec![ + MockOutcome::Status( + 402, + br#"{"error":{"type":"actor-memory-limit-exceeded","message":"not enough memory"}}"# + .to_vec(), + ), + MockOutcome::Status(201, br#"{"data":{"id":"run-id","status":"READY"}}"#.to_vec()), + MockOutcome::Status(200, br#"{"data":{"id":"run-id","status":"SUCCEEDED"}}"#.to_vec()), + ]); + let client = client_with(backend.clone(), 0); + + // `wait_secs: Some(1)` is a tight budget for *finishing*; the resource-limit retry's 10-second + // cooldown happens entirely before that budget's clock starts, in `start()`. + let run = client + .actor("some-actor") + .call::( + None, + ActorStartOptions { + wait_for_resources: Some(WaitForResources::Forever), + ..Default::default() + }, + Some(1), + ) + .await + .expect("the resource-limit retry must not eat into the wait_for_finish budget"); + assert_eq!(run.status.as_deref(), Some("SUCCEEDED")); +}