From 104b5051838acfb3c292c1bd60183cb5eb63200f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 09:34:05 +0000 Subject: [PATCH 1/2] feat!: add retry/parallelism/byte-chunking to batch_add_requests; type request-queue lock/list/batch results RequestQueueClient::batch_add_requests now matches the reference client's batchAddRequests: chunks by both request count (25) and JSON byte size (~9 MiB), sends chunks with bounded parallelism (default 5 in flight), and retries requests reported unprocessed (rate-limited) with exponential backoff (default 3 retries). Its second parameter is now BatchAddRequestsOptions instead of a bare forefront: bool. list_and_lock_head, list_requests, unlock_requests, prolong_request_lock and batch_delete_requests now return typed models (LockedRequestQueueHead, RequestQueueRequestsPage, UnlockRequestsResult, RequestLockInfo, BatchRequestsOperationResult) instead of serde_json::Value, matching every other resource client and the stable schemas the OpenAPI spec already documents for these endpoints. batch_delete_requests also now rejects more than 25 requests client-side instead of forwarding an oversized payload to the API. Breaking change, bumping 0.7.0 -> 0.8.0. --- CHANGELOG.md | 25 ++ Cargo.toml | 2 +- docs/storages.md | 88 +++++-- src/clients/request_queue.rs | 451 +++++++++++++++++++++++++++++++---- src/http_client.rs | 5 +- src/lib.rs | 4 +- src/models.rs | 121 ++++++++++ tests/request_queue.rs | 94 +++++++- 8 files changed, 714 insertions(+), 76 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b567e3c..b21a139 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,31 @@ 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.8.0] - 2026-08-11 + +### Changed +- `RequestQueueClient::batch_add_requests` now matches the reference client's `batchAddRequests`: + it automatically splits large inputs into chunks that respect both the API's per-call + request-count limit (25) and its request-body byte-size limit (~9 MiB), sends chunks with up + to `BatchAddRequestsOptions::max_parallel` requests in flight at once (default 5), and retries + requests an API call reports as `unprocessed` (typically rate-limited) with exponential + backoff, up to `max_unprocessed_requests_retries` times (default 3). Its second parameter is + now `BatchAddRequestsOptions` (was a bare `forefront: bool`), and it returns the typed + `BatchRequestsOperationResult` (was `serde_json::Value`). **Breaking.** +- `RequestQueueClient::batch_delete_requests` now returns the typed `BatchRequestsOperationResult` + (was `serde_json::Value`), and rejects more than 25 requests per call with + `ApifyClientError::InvalidArgument` instead of forwarding an oversized payload to the API + (matching the reference client, which validates rather than auto-chunks deletes). **Breaking.** +- `RequestQueueClient::list_and_lock_head` now returns the typed `LockedRequestQueueHead` (was + `serde_json::Value`). **Breaking.** +- `RequestQueueClient::list_requests` now returns the typed `RequestQueueRequestsPage` (was + `serde_json::Value`). **Breaking.** +- `RequestQueueClient::unlock_requests` now returns the typed `UnlockRequestsResult` (was + `serde_json::Value`). **Breaking.** +- `RequestQueueClient::prolong_request_lock` now returns the typed `RequestLockInfo` (was + `serde_json::Value`). **Breaking.** +- Bumped crate version to `0.8.0`. + ## [0.7.0] - 2026-08-10 ### Added diff --git a/Cargo.toml b/Cargo.toml index 0799dd6..0639aeb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apify-client" -version = "0.7.0" +version = "0.8.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" diff --git a/docs/storages.md b/docs/storages.md index ee3f141..2d48446 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -192,14 +192,14 @@ listed in `KeyValueStoreKeysPage::items`. Its fields: | `get_request(id)` | `&str` | `Option` | Reads a request. | | `update_request(request, forefront)` | `&RequestQueueRequest`, `bool` | `RequestQueueOperationInfo` | Updates a request. | | `delete_request(id)` | `&str` | `()` | Deletes a request. | -| `list_and_lock_head(lock_secs, limit)` | `i64`, `Option` | `Value` | Locks head requests. | -| `batch_add_requests(requests, forefront)` | `&[RequestQueueRequest]`, `bool` | `Value` | Batch add. | -| `batch_delete_requests(requests)` | `&[impl Serialize]` | `Value` | Batch delete. | -| `list_requests(options)` | `ListRequestsOptions { limit, exclusive_start_id, cursor, filter }` | `Value` | List requests (cursor/filter pagination). | +| `list_and_lock_head(lock_secs, limit)` | `i64`, `Option` | `LockedRequestQueueHead` | Locks head requests. | +| `batch_add_requests(requests, options)` | `&[RequestQueueRequest]`, `BatchAddRequestsOptions` | `BatchRequestsOperationResult` | Batch add, with automatic chunking, bounded parallelism, and retries for rate-limited requests. | +| `batch_delete_requests(requests)` | `&[impl Serialize]` | `BatchRequestsOperationResult` | Batch delete (max 25 requests per call; larger inputs are rejected client-side). | +| `list_requests(options)` | `ListRequestsOptions { limit, exclusive_start_id, cursor, filter }` | `RequestQueueRequestsPage` | List requests (cursor/filter pagination). | | `paginate_requests(page_limit)` | `Option` | `RequestQueueRequestsIterator` | Lazy request iterator. | -| `prolong_request_lock(id, lock_secs, forefront)` | `&str`, `i64`, `bool` | `Value` | Extend a lock. | +| `prolong_request_lock(id, lock_secs, forefront)` | `&str`, `i64`, `bool` | `RequestLockInfo` | Extend a lock. | | `delete_request_lock(id, forefront)` | `&str`, `bool` | `()` | Release a lock. | -| `unlock_requests()` | — | `Value` | Release all this client's locks. | +| `unlock_requests()` | — | `UnlockRequestsResult` | Release all this client's locks. | `paginate_requests(page_limit)` returns a `RequestQueueRequestsIterator` — a lazy, page-fetching iterator (parity with the Store iterator in [Store, users and logs](misc.md#apify-store--clientstore)). @@ -241,22 +241,62 @@ while let Some(request) = iter.next().await? { # } ``` -The `forefront` boolean (on `add_request`, `update_request`, `batch_add_requests`, -`prolong_request_lock`, `delete_request_lock`) controls queue ordering: `true` puts the -request(s) at the **front** of the queue so they are handled before the existing backlog; -`false` (the usual choice) appends them at the **back**. +The `forefront` boolean (on `add_request`, `update_request`, +`prolong_request_lock`, `delete_request_lock`) and the `BatchAddRequestsOptions::forefront` field +(on `batch_add_requests`) control queue ordering: `true` puts the request(s) at the **front** of +the queue so they are handled before the existing backlog; `false` (the usual choice) appends +them at the **back**. -Some request-queue methods return an untyped `serde_json::Value` because the API responses are -open-ended and most callers do not consume them structurally. Their shapes (read fields with -`value.get("...")`): +### `batch_add_requests` -- `list_and_lock_head` → an object with `items` (the locked head requests), `limit`, - `queueModifiedAt`, `hadMultipleClients`, and the granted `lockSecs`. -- `batch_add_requests` / `batch_delete_requests` → an object with `processedRequests` and - `unprocessedRequests` arrays. -- `list_requests` → an object with `items` (the page of requests), `count`, `limit`, and - `exclusiveStartId` for cursor continuation. -- `unlock_requests` → an object reporting how many locks were released (`unlockedCount`). +`batch_add_requests(requests, options)` is the efficient way to add many requests at once — +significantly cheaper than calling `add_request` in a loop. It mirrors the reference client's +`batchAddRequests`: + +- The input is automatically split into chunks that respect both the API's per-call request-count + limit (25) and its request-body byte-size limit (~9 MiB), so there is no need to chunk manually. +- Chunks are sent with up to `options.max_parallel` requests in flight at once (default 5). +- Any request an API call reports as `unprocessed` (typically due to rate limiting) is retried + automatically with exponential backoff, up to `options.max_unprocessed_requests_retries` times + (default 3). A request still unprocessed after every retry is reported in + `BatchRequestsOperationResult::unprocessed_requests` rather than failing the call. +- Every request should set `RequestQueueRequest::unique_key` (or rely on the API's `url` + fallback) so a retried request can be correlated back to the original input. + +```rust,no_run +use apify_client::models::RequestQueueRequest; +use apify_client::BatchAddRequestsOptions; +# use apify_client::ApifyClient; +# async fn run(client: ApifyClient) -> Result<(), Box> { +let queue = client.request_queues().get_or_create(None).await?; +let queue_client = client.request_queue(&queue.id); + +let requests: Vec = (0..3) + .map(|i| RequestQueueRequest { + id: None, + url: format!("https://example.com/{i}"), + unique_key: Some(format!("page-{i}")), + method: Some("GET".to_string()), + user_data: None, + extra: Default::default(), + }) + .collect(); + +let result = queue_client + .batch_add_requests(&requests, BatchAddRequestsOptions::default()) + .await?; +println!( + "added {} request(s), {} unprocessed", + result.processed_requests.len(), + result.unprocessed_requests.len() +); +# Ok(()) +# } +``` + +`batch_delete_requests` does **not** auto-chunk (matching the reference client): it accepts at +most 25 requests per call, identified by `id` and/or `unique_key` (not the full +`RequestQueueRequest` shape), and returns `ApifyClientError::InvalidArgument` for a larger input. ### `RequestQueueRequest` and request-queue return types @@ -304,6 +344,14 @@ Relevant return-type fields: `was_already_handled: bool`. - `RequestQueueHead`: `limit: i64`, `had_multiple_clients: bool`, `items: Vec`, `extra: Extra` (any other fields returned by the API). +- `LockedRequestQueueHead` (from `list_and_lock_head`): same fields as `RequestQueueHead` plus + `lock_secs: i64`, `queue_has_locked_requests: Option`, `client_key: Option`. +- `RequestQueueRequestsPage` (from `list_requests`): `limit: i64`, `items: Vec`, + `cursor` / `next_cursor` / `exclusive_start_id` (all `Option`) for pagination. +- `BatchRequestsOperationResult` (from `batch_add_requests` / `batch_delete_requests`): + `processed_requests: Vec`, `unprocessed_requests: Vec`. +- `RequestLockInfo` (from `prolong_request_lock`): `lock_expires_at: DateTime`. +- `UnlockRequestsResult` (from `unlock_requests`): `unlocked_count: i64`. - `KeyValueStoreKeysPage`: `limit: i64`, `is_truncated: bool`, `exclusive_start_key`, `next_exclusive_start_key` (both `Option`), `items: Vec`. diff --git a/src/clients/request_queue.rs b/src/clients/request_queue.rs index 45f1868..9f8a596 100644 --- a/src/clients/request_queue.rs +++ b/src/clients/request_queue.rs @@ -1,5 +1,9 @@ //! Client for a single request queue (`/v2/request-queues/{queueId}` and variants). +use std::collections::HashSet; +use std::time::Duration; + +use futures_util::stream::{self, StreamExt}; use serde::Serialize; use crate::clients::base::{ @@ -7,26 +11,132 @@ use crate::clients::base::{ post_with_body, update_resource, ResourceContext, }; use crate::common::{encode_path_segment, QueryParams}; -use crate::error::ApifyClientResult; -use crate::http_client::{HttpClient, HttpMethod, HttpRequest}; +use crate::error::{ApifyClientError, ApifyClientResult}; +use crate::http_client::{sleep_public, HttpClient, HttpMethod, HttpRequest}; use crate::models::{ - RequestQueue, RequestQueueHead, RequestQueueOperationInfo, RequestQueueRequest, + BatchRequestsOperationResult, LockedRequestQueueHead, RequestLockInfo, RequestQueue, + RequestQueueHead, RequestQueueOperationInfo, RequestQueueRequest, RequestQueueRequestsPage, + UnlockRequestsResult, UnprocessedRequest, }; /// Maximum number of requests the API accepts in a single `requests/batch` call. Larger -/// inputs are split into chunks of this size (matching the reference client). +/// `batch_add_requests` inputs are split into chunks of at most this size (matching the +/// reference client's `REQUEST_QUEUE_MAX_REQUESTS_PER_BATCH_OPERATION`); `batch_delete_requests` +/// does not auto-chunk (matching the reference client) and instead rejects larger inputs. const MAX_REQUESTS_PER_BATCH_OPERATION: usize = 25; +/// Maximum accepted size (bytes) of a request body, mirroring the platform-wide +/// `MAX_PAYLOAD_SIZE_BYTES` (9 MiB) that the reference client chunks `batch_add_requests` calls +/// against, on top of the per-call request-count limit. +const MAX_PAYLOAD_SIZE_BYTES: usize = 9_437_184; +/// Fraction of [`MAX_PAYLOAD_SIZE_BYTES`] held back as a safety margin (0.01%), matching the +/// reference client's `SAFETY_BUFFER_PERCENT`. +const SAFETY_BUFFER_PERCENT: f64 = 0.0001; +/// Default number of batch-add API calls [`RequestQueueClient::batch_add_requests`] keeps in +/// flight at once, matching the reference client's `DEFAULT_PARALLEL_BATCH_ADD_REQUESTS`. +const DEFAULT_MAX_PARALLEL_BATCH_ADD_REQUESTS: usize = 5; +/// Default number of retry attempts for requests a batch-add call reports as `unprocessed` +/// (typically rate-limited), matching the reference client's +/// `DEFAULT_UNPROCESSED_RETRIES_BATCH_ADD_REQUESTS`. +const DEFAULT_MAX_UNPROCESSED_REQUESTS_RETRIES: u32 = 3; +/// Default minimum delay before the first unprocessed-request retry; doubles (with jitter) on +/// each subsequent retry, matching the reference client's +/// `DEFAULT_MIN_DELAY_BETWEEN_UNPROCESSED_REQUESTS_RETRIES_MILLIS`. +const DEFAULT_MIN_DELAY_BETWEEN_UNPROCESSED_REQUESTS_RETRIES: Duration = Duration::from_millis(500); + +/// Options for [`RequestQueueClient::batch_add_requests`]. +/// +/// Mirrors the reference client's retrying, chunked, parallel `batchAddRequests`: large inputs +/// are split by count (max [`MAX_REQUESTS_PER_BATCH_OPERATION`]) and by JSON byte size (max +/// [`MAX_PAYLOAD_SIZE_BYTES`], minus a safety margin), chunks are sent with up to `max_parallel` +/// requests in flight at once, and any request an API call reports as `unprocessed` (typically +/// due to rate limiting) is retried with exponential backoff. +#[derive(Debug, Default, Clone)] +pub struct BatchAddRequestsOptions { + /// If `true`, adds all requests to the front of the queue. + pub forefront: bool, + /// Maximum retries for requests reported as `unprocessed`. Defaults to + /// [`DEFAULT_MAX_UNPROCESSED_REQUESTS_RETRIES`] (3) when `None`. + pub max_unprocessed_requests_retries: Option, + /// Maximum number of chunk-add API calls in flight at once. Defaults to + /// [`DEFAULT_MAX_PARALLEL_BATCH_ADD_REQUESTS`] (5) when `None`. + pub max_parallel: Option, + /// Minimum delay before the first unprocessed-request retry (doubles, with jitter, on each + /// subsequent retry). Defaults to + /// [`DEFAULT_MIN_DELAY_BETWEEN_UNPROCESSED_REQUESTS_RETRIES`] (500ms) when `None`. + pub min_delay_between_unprocessed_requests_retries: Option, +} + +/// Returns the key used to correlate a request across the batch-add retry loop: its explicit +/// `unique_key` if set, otherwise its `url` — matching the API's own fallback (a request added +/// without a `unique_key` is deduplicated by its normalized URL). +fn dedup_key(request: &RequestQueueRequest) -> &str { + request.unique_key.as_deref().unwrap_or(&request.url) +} + +/// Returns the JSON-serialized byte length of `value`. +fn json_byte_len(value: &T) -> ApifyClientResult { + Ok(serde_json::to_vec(value)?.len()) +} + +/// Slices `requests` down to a byte-limited prefix, mirroring the reference client's +/// `sliceArrayByByteLength`: if the whole slice already fits under `max_bytes` it is returned +/// unchanged; otherwise items are accumulated one at a time until the next one would exceed the +/// budget. `start_index` is only used to name the offending item in the error message, so it +/// should be the slice's absolute position within the caller's full input. +/// +/// The first item is always included regardless of size (once its own size has been checked +/// against `max_bytes`), guaranteeing a non-empty result for a non-empty input — unlike the +/// reference implementation, which can return an empty slice (and loop forever) when a single +/// item's size leaves no room under `max_bytes` for even itself plus the array wrapper. +/// +/// Returns [`ApifyClientError::InvalidArgument`] if a single request's JSON exceeds `max_bytes` +/// on its own (mirroring the reference client's thrown error). +fn slice_requests_by_byte_length( + requests: &[RequestQueueRequest], + max_bytes: usize, + start_index: usize, +) -> ApifyClientResult> { + if json_byte_len(requests)? < max_bytes { + return Ok(requests.to_vec()); + } + let mut out = Vec::new(); + let mut byte_length = 2usize; // 2 bytes for the empty array `[]`. + for (offset, request) in requests.iter().enumerate() { + let item_bytes = json_byte_len(request)?; + if item_bytes > max_bytes { + return Err(ApifyClientError::InvalidArgument(format!( + "RequestQueueClient::batch_add_requests: the request at index {} exceeds the \ + maximum allowed size ({max_bytes} bytes)", + start_index + offset + ))); + } + if !out.is_empty() && byte_length + item_bytes >= max_bytes { + break; + } + byte_length += item_bytes; + out.push(request.clone()); + } + Ok(out) +} -/// Appends the array under `key` in `chunk_result` (if present) onto `acc`. Used to merge the -/// per-chunk `processedRequests` / `unprocessedRequests` arrays of a chunked batch-add. -fn merge_request_array( - acc: &mut Vec, - chunk_result: &serde_json::Value, - key: &str, -) { - if let Some(items) = chunk_result.get(key).and_then(|v| v.as_array()) { - acc.extend(items.iter().cloned()); +/// Splits `requests` into chunks that each satisfy both the per-call count limit +/// ([`MAX_REQUESTS_PER_BATCH_OPERATION`]) and the payload byte-size limit +/// (`max_bytes`), mirroring the reference client's chunking loop in `batchAddRequests`. +fn chunk_requests_for_batch_add( + requests: &[RequestQueueRequest], + max_bytes: usize, +) -> ApifyClientResult>> { + let mut chunks = Vec::new(); + let mut i = 0; + while i < requests.len() { + let group_end = (i + MAX_REQUESTS_PER_BATCH_OPERATION).min(requests.len()); + let chunk = slice_requests_by_byte_length(&requests[i..group_end], max_bytes, i)?; + // `slice_requests_by_byte_length` always returns at least one item for a non-empty + // input, so this advances on every iteration. + i += chunk.len(); + chunks.push(chunk); } + Ok(chunks) } /// Options for [`RequestQueueClient::list_requests`]. @@ -193,7 +303,7 @@ impl RequestQueueClient { &self, lock_secs: i64, limit: Option, - ) -> ApifyClientResult { + ) -> ApifyClientResult { let mut params = self.base_params(); params .add_int("lockSecs", Some(lock_secs)) @@ -201,37 +311,159 @@ impl RequestQueueClient { post_action(&self.ctx, Some("head/lock"), ¶ms, None, None).await } - /// Adds multiple requests to the queue, automatically splitting the input into chunks of - /// at most [`MAX_REQUESTS_PER_BATCH_OPERATION`] requests per API call (the API rejects - /// larger batches). The per-chunk responses are merged into a single result whose - /// `processedRequests` / `unprocessedRequests` arrays concatenate every chunk's, matching - /// the reference client's client-side chunking. + /// Adds multiple requests to the queue in a single logical operation. + /// + /// This is significantly more efficient than calling [`add_request`](Self::add_request) + /// once per request, especially for large batches: the input is automatically split into + /// chunks that respect both the API's per-call request-count limit + /// ([`MAX_REQUESTS_PER_BATCH_OPERATION`]) and its request-body byte-size limit + /// ([`MAX_PAYLOAD_SIZE_BYTES`]), chunks are sent with up to `options.max_parallel` API calls + /// in flight at once, and any request an API call reports as `unprocessed` (typically due to + /// rate limiting) is retried with exponential backoff — matching the reference client's + /// `batchAddRequests`. Every request must be identifiable by [`RequestQueueRequest::unique_key`] + /// (or, if left unset, by `url`, the API's own fallback) so a retried request can be matched + /// back to the original input. + /// + /// Unlike most methods here, this does not propagate per-chunk API errors: a chunk that fails + /// even after retries has its requests reported in the result's `unprocessed_requests` + /// instead, so a batch add of many requests never fails outright over one bad chunk (matching + /// the reference client). A [`ApifyClientError::InvalidArgument`] is still returned before any + /// request is sent if a single request's JSON is too large to ever fit in a chunk. pub async fn batch_add_requests( &self, requests: &[RequestQueueRequest], - forefront: bool, - ) -> ApifyClientResult { - let mut processed: Vec = Vec::new(); - let mut unprocessed: Vec = Vec::new(); - - for chunk in requests.chunks(MAX_REQUESTS_PER_BATCH_OPERATION) { - let chunk_result = self.batch_add_chunk(chunk, forefront).await?; - merge_request_array(&mut processed, &chunk_result, "processedRequests"); - merge_request_array(&mut unprocessed, &chunk_result, "unprocessedRequests"); + options: BatchAddRequestsOptions, + ) -> ApifyClientResult { + if requests.is_empty() { + return Ok(BatchRequestsOperationResult::default()); } + let max_parallel = options + .max_parallel + .unwrap_or(DEFAULT_MAX_PARALLEL_BATCH_ADD_REQUESTS) + .max(1); + let payload_limit_bytes = MAX_PAYLOAD_SIZE_BYTES + - (MAX_PAYLOAD_SIZE_BYTES as f64 * SAFETY_BUFFER_PERCENT).ceil() as usize; + let chunks = chunk_requests_for_batch_add(requests, payload_limit_bytes)?; + + let merged = stream::iter(chunks) + .map(|chunk| { + let client = self.clone(); + let options = options.clone(); + async move { + client + .batch_add_requests_chunk_with_retries(chunk, options) + .await + } + }) + .buffer_unordered(max_parallel) + .fold( + BatchRequestsOperationResult::default(), + |mut acc, chunk_result| async move { + acc.processed_requests + .extend(chunk_result.processed_requests); + acc.unprocessed_requests + .extend(chunk_result.unprocessed_requests); + acc + }, + ) + .await; + Ok(merged) + } - Ok(serde_json::json!({ - "processedRequests": processed, - "unprocessedRequests": unprocessed, - })) + /// Sends one already-byte/count-limited chunk, retrying requests reported as `unprocessed` + /// with exponential backoff (matching the reference client's `_batchAddRequestsWithRetries`). + /// + /// Never returns an `Err`: a transport/API failure that survives `HttpClient`'s own retries + /// marks every request still outstanding in this chunk as unprocessed instead of propagating, + /// so a single bad chunk cannot fail the whole (possibly-parallel) `batch_add_requests` call. + async fn batch_add_requests_chunk_with_retries( + &self, + chunk: Vec, + options: BatchAddRequestsOptions, + ) -> BatchRequestsOperationResult { + let max_retries = options + .max_unprocessed_requests_retries + .unwrap_or(DEFAULT_MAX_UNPROCESSED_REQUESTS_RETRIES); + let min_delay = options + .min_delay_between_unprocessed_requests_retries + .unwrap_or(DEFAULT_MIN_DELAY_BETWEEN_UNPROCESSED_REQUESTS_RETRIES); + + let mut remaining = chunk; + let mut processed = Vec::new(); + + for attempt in 0..=max_retries { + match self + .batch_add_requests_raw(&remaining, options.forefront) + .await + { + Ok(result) => { + let processed_keys: HashSet<&str> = result + .processed_requests + .iter() + .filter_map(|p| p.unique_key.as_deref()) + .collect(); + remaining.retain(|r| !processed_keys.contains(dedup_key(r))); + processed.extend(result.processed_requests); + if remaining.is_empty() { + return BatchRequestsOperationResult { + processed_requests: processed, + unprocessed_requests: Vec::new(), + }; + } + if attempt == max_retries { + return BatchRequestsOperationResult { + processed_requests: processed, + unprocessed_requests: result.unprocessed_requests, + }; + } + } + Err(_) => { + // A hard failure (already retried by `HttpClient` for transient errors): + // treat every request still outstanding in this chunk as unprocessed rather + // than propagating, matching the reference client's "never throws" contract. + return BatchRequestsOperationResult { + processed_requests: processed, + unprocessed_requests: remaining + .iter() + .map(|r| UnprocessedRequest { + unique_key: dedup_key(r).to_string(), + url: r.url.clone(), + method: r.method.clone(), + }) + .collect(), + }; + } + } + // Exponential backoff with jitter before the next retry, matching the reference + // client's `(1 + random()) * 2^attempt * minDelay` (see `randomized_delay`, which + // returns a value in `[base, 2*base)`, i.e. `(1 + random()) * base`). + let backoff = min_delay.saturating_mul(2u32.saturating_pow(attempt)); + sleep_public(crate::http_client::randomized_delay(backoff)).await; + } + // Unreachable: the loop above always returns on its last iteration (`attempt == + // max_retries` is handled inside the `Ok` arm, and `Err` returns unconditionally). Kept + // as a safe fallback rather than `unreachable!()` so a future refactor of the loop bounds + // fails safe (reporting the batch unprocessed) instead of panicking. + BatchRequestsOperationResult { + processed_requests: processed, + unprocessed_requests: remaining + .iter() + .map(|r| crate::models::UnprocessedRequest { + unique_key: dedup_key(r).to_string(), + url: r.url.clone(), + method: r.method.clone(), + }) + .collect(), + } } - /// Posts a single chunk of requests (at most [`MAX_REQUESTS_PER_BATCH_OPERATION`]). - async fn batch_add_chunk( + /// Sends a single `POST requests/batch` call (at most [`MAX_REQUESTS_PER_BATCH_OPERATION`] + /// requests, and within the byte-size budget already enforced by the caller). + async fn batch_add_requests_raw( &self, requests: &[RequestQueueRequest], forefront: bool, - ) -> ApifyClientResult { + ) -> ApifyClientResult { let mut params = self.base_params(); params.add_bool("forefront", Some(forefront)); let body = serde_json::to_vec(requests)?; @@ -246,10 +478,21 @@ impl RequestQueueClient { } /// Deletes multiple requests in a single batch operation. + /// + /// Unlike [`batch_add_requests`](Self::batch_add_requests), this does not auto-chunk: the API + /// accepts at most [`MAX_REQUESTS_PER_BATCH_OPERATION`] requests per call (matching the + /// reference client, which validates rather than chunks), so a larger `requests` returns + /// [`ApifyClientError::InvalidArgument`] before any request is sent. pub async fn batch_delete_requests( &self, requests: &[T], - ) -> ApifyClientResult { + ) -> ApifyClientResult { + if requests.is_empty() || requests.len() > MAX_REQUESTS_PER_BATCH_OPERATION { + return Err(ApifyClientError::InvalidArgument(format!( + "RequestQueueClient::batch_delete_requests accepts between 1 and {MAX_REQUESTS_PER_BATCH_OPERATION} requests per call, got {}", + requests.len() + ))); + } delete_with_body( &self.ctx, Some("requests/batch"), @@ -266,7 +509,7 @@ impl RequestQueueClient { pub async fn list_requests( &self, options: ListRequestsOptions, - ) -> ApifyClientResult { + ) -> ApifyClientResult { let mut params = self.base_params(); params .add_int("limit", options.limit) @@ -285,7 +528,7 @@ impl RequestQueueClient { id: &str, lock_secs: i64, forefront: bool, - ) -> ApifyClientResult { + ) -> ApifyClientResult { let mut params = self.base_params(); params .add_int("lockSecs", Some(lock_secs)) @@ -350,7 +593,7 @@ impl RequestQueueClient { } /// Unlocks all requests currently locked by this client (identified by `client_key`). - pub async fn unlock_requests(&self) -> ApifyClientResult { + pub async fn unlock_requests(&self) -> ApifyClientResult { post_action( &self.ctx, Some("requests/unlock"), @@ -399,25 +642,133 @@ impl RequestQueueRequestsIterator { }) .await?; - // Parse the items and the next cursor from the (untyped) page. - let items: Vec = page - .get("items") - .map(|v| serde_json::from_value(v.clone())) - .transpose()? - .unwrap_or_default(); - - if items.is_empty() { + if page.items.is_empty() { self.exhausted = true; return Ok(None); } // Advance the cursor; stop when the API stops returning one. - match page.get("nextCursor").and_then(|v| v.as_str()) { - Some(cursor) if !cursor.is_empty() => self.next_cursor = Some(cursor.to_string()), + match page.next_cursor { + Some(cursor) if !cursor.is_empty() => self.next_cursor = Some(cursor), _ => self.exhausted = true, } - self.buffer.extend(items); + self.buffer.extend(page.items); Ok(self.buffer.pop_front()) } } + +#[cfg(test)] +mod batch_add_tests { + use super::{ + chunk_requests_for_batch_add, dedup_key, slice_requests_by_byte_length, + MAX_REQUESTS_PER_BATCH_OPERATION, + }; + use crate::models::RequestQueueRequest; + + fn request(url: &str, unique_key: Option<&str>) -> RequestQueueRequest { + RequestQueueRequest { + id: None, + url: url.to_string(), + unique_key: unique_key.map(str::to_string), + method: None, + user_data: None, + extra: Default::default(), + } + } + + /// A request without an explicit `unique_key` is correlated by `url`, matching the API's own + /// deduplication fallback. + #[test] + fn dedup_key_falls_back_to_url() { + let with_key = request("https://example.com", Some("k1")); + assert_eq!(dedup_key(&with_key), "k1"); + + let without_key = request("https://example.com/no-key", None); + assert_eq!(dedup_key(&without_key), "https://example.com/no-key"); + } + + /// A slice that already fits under the byte budget is returned unchanged. + #[test] + fn byte_slice_returns_everything_when_under_budget() { + let requests: Vec<_> = (0..5) + .map(|i| request(&format!("https://example.com/{i}"), None)) + .collect(); + let sliced = slice_requests_by_byte_length(&requests, 1_000_000, 0).unwrap(); + assert_eq!(sliced.len(), 5); + } + + /// When the whole slice exceeds the byte budget, only a byte-limited prefix is taken — but + /// never an empty one, even if the very first item alone leaves no room for a second. + #[test] + fn byte_slice_takes_a_limited_prefix() { + let requests: Vec<_> = (0..10) + .map(|i| request(&format!("https://example.com/{i}"), None)) + .collect(); + // Each request serializes to roughly 30 bytes; a budget of 50 fits one comfortably but + // never two. + let sliced = slice_requests_by_byte_length(&requests, 50, 0).unwrap(); + assert_eq!( + sliced.len(), + 1, + "budget of 50 bytes should admit exactly one ~30-byte request" + ); + } + + /// A single request whose own JSON exceeds the byte budget is a hard error, not a silently + /// dropped item — the caller could never send it in any chunk. + #[test] + fn byte_slice_errors_on_oversized_single_request() { + let huge_url = format!("https://example.com/{}", "x".repeat(1000)); + let requests = vec![request(&huge_url, None)]; + let err = slice_requests_by_byte_length(&requests, 100, 3).unwrap_err(); + let message = err.to_string(); + assert!( + message.contains("index 3"), + "error should name the absolute index of the oversized request: {message}" + ); + } + + /// Chunking respects the per-call count cap even when every request is tiny (byte budget is + /// never the limiting factor). + #[test] + fn chunking_splits_by_count_when_bytes_are_plentiful() { + let requests: Vec<_> = (0..(MAX_REQUESTS_PER_BATCH_OPERATION * 2 + 3)) + .map(|i| request(&format!("https://example.com/{i}"), None)) + .collect(); + let chunks = chunk_requests_for_batch_add(&requests, 10_000_000).unwrap(); + let sizes: Vec = chunks.iter().map(Vec::len).collect(); + assert_eq!( + sizes, + vec![ + MAX_REQUESTS_PER_BATCH_OPERATION, + MAX_REQUESTS_PER_BATCH_OPERATION, + 3 + ] + ); + let total: usize = sizes.iter().sum(); + assert_eq!(total, requests.len()); + } + + /// Chunking also respects the byte budget, producing more (smaller) chunks than the count + /// cap alone would when requests are large. + #[test] + fn chunking_splits_by_byte_budget_when_tighter_than_count_cap() { + let requests: Vec<_> = (0..6) + .map(|i| request(&format!("https://example.com/{i}"), None)) + .collect(); + // ~30 bytes/request; a 100-byte budget forces multiple chunks well under the 25-item cap. + let chunks = chunk_requests_for_batch_add(&requests, 100).unwrap(); + assert!( + chunks.len() > 1, + "a tight byte budget must force more than one chunk, got {}", + chunks.len() + ); + let total: usize = chunks.iter().map(Vec::len).sum(); + assert_eq!( + total, + requests.len(), + "every request must end up in exactly one chunk" + ); + } +} diff --git a/src/http_client.rs b/src/http_client.rs index 2f37b27..62f5628 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -474,7 +474,10 @@ fn build_api_error( /// Returns a delay chosen randomly from the interval `[delay, 2*delay)`, matching the /// exponential-backoff-with-jitter algorithm described in the API docs. -fn randomized_delay(delay: Duration) -> Duration { +/// +/// `pub(crate)` so other retrying call sites (e.g. `RequestQueueClient::batch_add_requests`'s +/// unprocessed-request retries) can reuse the same jitter source instead of duplicating it. +pub(crate) fn randomized_delay(delay: Duration) -> Duration { let base = delay.as_millis() as u64; if base == 0 { return delay; diff --git a/src/lib.rs b/src/lib.rs index fe9649e..453972a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,7 +65,9 @@ pub use clients::dataset::{DatasetDownloadOptions, DatasetListItemsOptions, Down pub use clients::key_value_store::{GetRecordOptions, KeyValueStoreKeysIterator, ListKeysOptions}; pub use clients::log::LogOptions; pub use clients::pagination::ListIterator; -pub use clients::request_queue::{ListRequestsOptions, RequestQueueRequestsIterator}; +pub use clients::request_queue::{ + BatchAddRequestsOptions, ListRequestsOptions, RequestQueueRequestsIterator, +}; pub use clients::run::{ LastRunOptions, RunChargeOptions, RunMetamorphOptions, RunResurrectOptions, }; diff --git a/src/models.rs b/src/models.rs index 679cad2..6a4dde0 100644 --- a/src/models.rs +++ b/src/models.rs @@ -364,6 +364,127 @@ pub struct RequestQueueHead { pub extra: Extra, } +/// The head of a request queue with a server-side lock applied (`POST .../head/lock`). +/// +/// Same shape as [`RequestQueueHead`] plus the lock metadata the API returns alongside it. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LockedRequestQueueHead { + /// Maximum number of requests returned. + #[serde(default)] + pub limit: i64, + /// Whether more than one client has accessed the queue. + #[serde(default)] + pub had_multiple_clients: bool, + /// Number of seconds the returned requests were locked for. + #[serde(default)] + pub lock_secs: i64, + /// Whether the queue has any requests locked, by this or another client. + #[serde(default)] + pub queue_has_locked_requests: Option, + /// The client key the lock was acquired with. + #[serde(default)] + pub client_key: Option, + /// The locked requests from the head of the queue. + #[serde(default)] + pub items: Vec, + /// Any other fields returned by the API. + #[serde(flatten)] + pub extra: Extra, +} + +/// A page of requests returned by `GET /v2/request-queues/{queueId}/requests` (cursor-based +/// pagination over every request in the queue, as opposed to [`RequestQueueHead`]'s unlocked +/// peek at the head). +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RequestQueueRequestsPage { + /// Maximum number of requests returned for this request. + #[serde(default)] + pub limit: i64, + /// ID of the last request of the previous page. Deprecated by the API in favour of `cursor`. + #[serde(default)] + pub exclusive_start_id: Option, + /// Cursor identifying the current page. + #[serde(default)] + pub cursor: Option, + /// Cursor to pass as `cursor` to fetch the next page; absent on the last page. + #[serde(default)] + pub next_cursor: Option, + /// The requests of this page. + #[serde(default)] + pub items: Vec, +} + +/// Result of prolonging a request's lock (`PUT .../requests/{requestId}/lock`). +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RequestLockInfo { + /// When the (prolonged) lock expires. + pub lock_expires_at: DateTime, +} + +/// Result of `POST /v2/request-queues/{queueId}/requests/unlock`. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct UnlockRequestsResult { + /// Number of requests that were unlocked. + pub unlocked_count: i64, +} + +/// A request successfully processed by a request-queue batch add or batch delete operation. +/// +/// The populated fields depend on the operation: a batch **add** always sets `unique_key`, +/// `request_id`, `was_already_present` and `was_already_handled` (the API's `AddedRequest`); a +/// batch **delete** sets `id` and/or `unique_key`, whichever the request was identified by (the +/// API's `DeletedRequest`). +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProcessedRequest { + /// The request's ID, when the operation identifies requests by ID (batch delete). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, + /// The request's unique key. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub unique_key: Option, + /// The request's ID, as returned by a batch **add** (mirrors the API's `requestId`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub request_id: Option, + /// Whether the request was already present in the queue (batch add only). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub was_already_present: Option, + /// Whether the request had already been handled (batch add only). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub was_already_handled: Option, +} + +/// A request that a request-queue batch add operation did not process (typically due to rate +/// limiting), and which [`crate::clients::request_queue::RequestQueueClient::batch_add_requests`] +/// retries automatically. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct UnprocessedRequest { + /// The request's unique key. + pub unique_key: String, + /// The request's URL. + pub url: String, + /// The request's HTTP method. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub method: Option, +} + +/// Result of a request-queue batch add or batch delete operation. +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BatchRequestsOperationResult { + /// Requests that were successfully processed. + #[serde(default)] + pub processed_requests: Vec, + /// Requests that were not processed and can be retried. + #[serde(default)] + pub unprocessed_requests: Vec, +} + /// A schedule that triggers Actor or task runs on a cron expression. #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] diff --git a/tests/request_queue.rs b/tests/request_queue.rs index 6072d6b..8c0e731 100644 --- a/tests/request_queue.rs +++ b/tests/request_queue.rs @@ -216,6 +216,92 @@ async fn request_queue_paginate_multiple_pages() { queue_client.delete().await.expect("delete queue"); } +/// Exercises `batch_add_requests` (typed result, default retry/parallelism options) followed by +/// `batch_delete_requests`, plus the client-side validation `batch_delete_requests` performs on +/// oversized inputs. +#[tokio::test(flavor = "multi_thread")] +async fn request_queue_batch_add_and_delete() { + let client = require_client!(); + let name = common::unique_name("rq-batch"); + + let queue = client + .request_queues() + .get_or_create(Some(&name)) + .await + .expect("create queue"); + + let cleanup_client = client.clone(); + let cleanup_id = queue.id.clone(); + let _guard = common::Cleanup::new(move || async move { + let _ = cleanup_client.request_queue(&cleanup_id).delete().await; + }); + + let queue_client = client.request_queue(&queue.id); + + const TOTAL: usize = 8; + let requests: Vec = (0..TOTAL) + .map(|i| RequestQueueRequest { + id: None, + url: format!("https://example.com/batch/{i}"), + unique_key: Some(format!("batch-{i}")), + method: Some("GET".to_string()), + user_data: None, + extra: Default::default(), + }) + .collect(); + + let add_result = queue_client + .batch_add_requests(&requests, Default::default()) + .await + .expect("batch add requests"); + assert_eq!( + add_result.processed_requests.len(), + TOTAL, + "every request should be processed (none rate-limited on a fresh queue)" + ); + assert!(add_result.unprocessed_requests.is_empty()); + let mut processed_keys: std::collections::HashSet = add_result + .processed_requests + .iter() + .filter_map(|r| r.unique_key.clone()) + .collect(); + for i in 0..TOTAL { + assert!( + processed_keys.remove(&format!("batch-{i}")), + "processed_requests should report unique_key batch-{i}" + ); + } + + // Delete them all in one batch call (within the 25-per-call limit). The batch-delete schema + // identifies requests by `id` and/or `uniqueKey` only (no `url`/`method`/...), so build + // minimal delete-by-key objects rather than reusing the full `RequestQueueRequest`s. + let delete_by_key: Vec = requests + .iter() + .map(|r| json!({ "uniqueKey": r.unique_key })) + .collect(); + let delete_result = queue_client + .batch_delete_requests(&delete_by_key) + .await + .expect("batch delete requests"); + assert_eq!(delete_result.processed_requests.len(), TOTAL); + + // `batch_delete_requests` rejects more than 25 requests per call client-side, matching the + // reference client's validation (it does not auto-chunk deletes). + let too_many: Vec = (0..26) + .map(|i| json!({ "uniqueKey": format!("toomany-{i}") })) + .collect(); + let err = queue_client + .batch_delete_requests(&too_many) + .await + .expect_err("more than 25 requests must be rejected client-side"); + assert!(matches!( + err, + apify_client::ApifyClientError::InvalidArgument(_) + )); + + queue_client.delete().await.expect("delete queue"); +} + /// Exercises the request lock lifecycle: add -> list_and_lock_head -> prolong -> unlock, /// plus `list_requests` and `unlock_requests`. #[tokio::test(flavor = "multi_thread")] @@ -262,7 +348,7 @@ async fn request_queue_lock_lifecycle() { }) .await .expect("list requests"); - assert!(listed.get("items").is_some()); + assert!(!listed.items.is_empty()); // Exercise the `filter` parameter with both enum values (`locked`, `pending`). This verifies // the multi-value, comma-joined serialization (`filter=locked,pending`) is accepted by the API. @@ -274,7 +360,9 @@ async fn request_queue_lock_lifecycle() { }) .await .expect("list requests with filter"); - assert!(filtered.get("items").is_some()); + // The filter may legitimately exclude every request; the call succeeding (no parse/API + // error) is what this test exercises. + assert!(filtered.limit >= 0); // Lazily paginate requests; we added one, so at least one should be yielded. let mut iter = queue_client.paginate_requests(Some(10)); @@ -286,7 +374,7 @@ async fn request_queue_lock_lifecycle() { .list_and_lock_head(30, Some(5)) .await .expect("lock head"); - assert!(locked.get("items").is_some()); + assert!(locked.lock_secs > 0); // Prolong, then release the lock on the added request. queue_client From ee1c90e04f5ac66b7d18db130d75b2e16867be5d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 09:50:25 +0000 Subject: [PATCH 2/2] fix: address review feedback on batch_add_requests rework - Correct dedup_key's doc comment: the API's keyless fallback dedups by the raw url, not a "normalized" one. - batch_add_requests now rejects an empty requests slice with InvalidArgument, matching batch_delete_requests and the reference client (both validate non-empty input). - Use the imported UnprocessedRequest name instead of the fully-qualified crate::models:: path in the retry loop's fail-safe return. - Replace a vacuous assert!(limit >= 0) in the lock-lifecycle test with a real invariant: filtering on both locked+pending must match the unfiltered listing. --- CHANGELOG.md | 4 ++++ src/clients/request_queue.rs | 15 ++++++++++----- tests/request_queue.rs | 11 ++++++++--- 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b21a139..e474b94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,10 @@ to [Semantic Versioning](https://semver.org/). (was `serde_json::Value`), and rejects more than 25 requests per call with `ApifyClientError::InvalidArgument` instead of forwarding an oversized payload to the API (matching the reference client, which validates rather than auto-chunks deletes). **Breaking.** +- `RequestQueueClient::batch_add_requests` now also rejects an empty `requests` with + `ApifyClientError::InvalidArgument` (was `Ok` with an empty result), matching + `batch_delete_requests` and the reference client, which validates both as non-empty. + **Breaking.** - `RequestQueueClient::list_and_lock_head` now returns the typed `LockedRequestQueueHead` (was `serde_json::Value`). **Breaking.** - `RequestQueueClient::list_requests` now returns the typed `RequestQueueRequestsPage` (was diff --git a/src/clients/request_queue.rs b/src/clients/request_queue.rs index 9f8a596..ab2f3ee 100644 --- a/src/clients/request_queue.rs +++ b/src/clients/request_queue.rs @@ -68,7 +68,7 @@ pub struct BatchAddRequestsOptions { /// Returns the key used to correlate a request across the batch-add retry loop: its explicit /// `unique_key` if set, otherwise its `url` — matching the API's own fallback (a request added -/// without a `unique_key` is deduplicated by its normalized URL). +/// without a `unique_key` is deduplicated by its raw `url`). fn dedup_key(request: &RequestQueueRequest) -> &str { request.unique_key.as_deref().unwrap_or(&request.url) } @@ -327,15 +327,20 @@ impl RequestQueueClient { /// Unlike most methods here, this does not propagate per-chunk API errors: a chunk that fails /// even after retries has its requests reported in the result's `unprocessed_requests` /// instead, so a batch add of many requests never fails outright over one bad chunk (matching - /// the reference client). A [`ApifyClientError::InvalidArgument`] is still returned before any - /// request is sent if a single request's JSON is too large to ever fit in a chunk. + /// the reference client). It still returns [`ApifyClientError::InvalidArgument`] up front, + /// before any request is sent, for an empty `requests` (matching + /// [`batch_delete_requests`](Self::batch_delete_requests) and the reference client, which + /// validates both as non-empty) or if a single request's JSON is too large to ever fit in a + /// chunk. pub async fn batch_add_requests( &self, requests: &[RequestQueueRequest], options: BatchAddRequestsOptions, ) -> ApifyClientResult { if requests.is_empty() { - return Ok(BatchRequestsOperationResult::default()); + return Err(ApifyClientError::InvalidArgument( + "RequestQueueClient::batch_add_requests requires at least 1 request".to_string(), + )); } let max_parallel = options .max_parallel @@ -448,7 +453,7 @@ impl RequestQueueClient { processed_requests: processed, unprocessed_requests: remaining .iter() - .map(|r| crate::models::UnprocessedRequest { + .map(|r| UnprocessedRequest { unique_key: dedup_key(r).to_string(), url: r.url.clone(), method: r.method.clone(), diff --git a/tests/request_queue.rs b/tests/request_queue.rs index 8c0e731..ce02db8 100644 --- a/tests/request_queue.rs +++ b/tests/request_queue.rs @@ -360,9 +360,14 @@ async fn request_queue_lock_lifecycle() { }) .await .expect("list requests with filter"); - // The filter may legitimately exclude every request; the call succeeding (no parse/API - // error) is what this test exercises. - assert!(filtered.limit >= 0); + // The single request we added is either `locked` or `pending` (every request is one or the + // other), so filtering on both states must return exactly the same set as the unfiltered + // listing above — a real invariant, not just "the call didn't error". + assert_eq!( + filtered.items.len(), + listed.items.len(), + "filtering on both locked and pending must match the unfiltered listing" + ); // Lazily paginate requests; we added one, so at least one should be yielded. let mut iter = queue_client.paginate_requests(Some(10));