diff --git a/CHANGELOG.md b/CHANGELOG.md index bf9176e..7388112 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,49 @@ 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.16.0] - 2026-11-06 + +Ports the `apify-client-js` Server Actor support and the `count` of dataset item pages. + +### Added +- `ActorClient::server`, which returns an `ActorServerClient` for a Server Actor (formerly Actor + Standby), and its deprecated alias `ActorClient::standby`. +- `ActorServerClient::request` and `request_stream`, which send an HTTP request to the Actor's + server with the token in `X-Apify-Authorization`, without following redirects, retrying only `429` + and `503`, and returning error statuses as responses. Their options are + `ActorServerRequestOptions`, `ActorServerBody` and `ActorServerCompression`. +- `ActorServerClient::get_url` and `get_open_api_spec`. +- `Actor::standby_url`, `actor_standby` (`ActorStandby`, with the `tenancy` values `SINGLE_TENANT` + and `MULTI_TENANT`) and `tagged_builds` (`TaggedBuildInfo`). +- `ActorVersion::actor_path`, the Actor's directory inside an uploaded source. +- `common::MAX_EXPIRES_IN_SECS`, the longest `expires_in_secs` of a signed public URL. +- `HttpMethod::Patch`, `Options` and `Trace`, `HttpRequest::follow_redirects`, + `ReqwestBackend::with_no_redirect_client`, and `HttpResponse::is_success`, `text` and `json`. + +### Changed +- Synced with Apify OpenAPI spec `v2-2026-11-04T151257Z` (`API_SPEC_VERSION`). +- **Breaking:** `PaginationList::count` of `DatasetClient::list_items` is the number of rows the API + scanned, or the number of items returned when that is larger, so filters can leave it above + `items.len()`. The iterators step by it, capped at the page size. +- **Breaking:** `HttpRequest` has the new field `follow_redirects`, which a custom `HttpBackend` + must honor, and `HttpMethod` has new variants. +- **Breaking:** `Actor` and `ActorVersion` have new fields, and `standbyUrl`, `actorStandby` and + `taggedBuilds` are no longer in `Actor::extra`. +- Documented that iterating `iterate_items` without a page size loads the whole dataset in one + response. + +### Fixed +- The streaming calls (`LogClient::stream`, `DatasetClient::download_items_stream` and + `ActorServerClient::request_stream`) retry an error status or a transport error until the + response starts, like the other calls, instead of failing on the first one. +- `create_items_public_url`, `create_items_public_url_with_format` and `create_keys_public_url` fail + with `InvalidArgument` for an `expires_in_secs` that is `0` or negative, before any request is + sent. Such a value used to sign a URL that had already expired. An `expires_in_secs` above the new + `common::MAX_EXPIRES_IN_SECS` fails with `InvalidArgument` too, instead of panicking or wrapping + around. +- A `ListIterator` no longer ends in front of rows that match when a filter leaves a page with no + items. + ## [0.15.0] - 2026-10-29 Ports the `apify-client-js` v3 changes. diff --git a/Cargo.toml b/Cargo.toml index ba9a545..7a65a6e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apify-client" -version = "0.15.0" +version = "0.16.0" authors = ["Apify Technologies "] description = "Rust client for the Apify API (https://apify.com)." license = "Apache-2.0" diff --git a/README.md b/README.md index 78a146b..4688687 100644 --- a/README.md +++ b/README.md @@ -215,6 +215,10 @@ Every request carries a `User-Agent` of the form | `client.me()` / `client.user(id)` | single | User accounts | | `client.log(build_or_run_id)` | single | Build/run logs | +`client.actor(id).server()` returns a client for the server interface of a Server Actor (formerly +Actor Standby): it sends HTTP requests to the Actor's web server with the client's token, and gets +the server's URL and OpenAPI specification. See [Actors](docs/actors.md#server-actors). + `client.api_base_url()` returns the API base URL the client targets, including `/v2` and any `base_url` override. The iterators returned by `iterate(..)` (and by `RunClient::iterate_dataset_items`) yield one item per `next().await?`; `collect_all().await?` drains the remaining items into a `Vec`. @@ -392,7 +396,9 @@ pool, …), as shown below; to route requests through an entirely different HTTP `log().stream()` and `download_items_stream()`) has a default implementation that buffers the whole response via `send`, so it is optional unless you want true incremental streaming. A request whose `timeout` is `timeouts::NO_TIMEOUT` (see [Timeouts](#timeouts)) may run for as long as it -takes, so a backend must check for it before it applies a timeout of its own: +takes, so a backend must check for it before it applies a timeout of its own. A request whose +`follow_redirects` is `false`, such as one to an Actor's server, carries a token that must not +follow a redirect to another host, so a backend must return a redirect response as it is: ```rust,no_run use apify_client::http_client::{HttpBackend, HttpRequest, HttpResponse}; @@ -430,13 +436,28 @@ The snippet below constructs a use std::sync::Arc; use apify_client::ApifyClient; use apify_client::http_client::ReqwestBackend; +use reqwest::redirect::Policy; -// Share a pre-configured reqwest client (proxy, TLS, connection pool, ...). +// Share a pre-configured reqwest client (proxy, TLS, connection pool, ...). The requests that must +// not follow redirects, such as those to an Actor's server, go through a second client: give it +// the same settings, and a policy that follows no redirects. let reqwest_client = reqwest::Client::builder().build().unwrap(); -let backend = Arc::new(ReqwestBackend::with_client(reqwest_client)); +let no_redirect_client = reqwest::Client::builder() + .redirect(Policy::none()) + .build() + .unwrap(); +let backend = Arc::new( + ReqwestBackend::with_client(reqwest_client).with_no_redirect_client(no_redirect_client), +); let client = ApifyClient::builder().token("t").http_backend(backend).build(); ``` +A `reqwest::Client` fixes its redirect policy when it is built, so `ReqwestBackend` sends the +requests that must not follow redirects with a second client. `with_client` gives it a default one, +without your proxy, TLS or CA settings, so replace it with `with_no_redirect_client` when the +requests to an Actor's server need those too. The replacement has to be built with +`redirect(reqwest::redirect::Policy::none())`. + ## License Licensed under the [Apache License, Version 2.0](LICENSE). diff --git a/docs/README.md b/docs/README.md index fe3aabf..0d591b2 100644 --- a/docs/README.md +++ b/docs/README.md @@ -79,6 +79,7 @@ types is: - Store: `StoreListOptions` - Logs: `LogOptions`, `LogRedirect`, `LogSink`, `StreamedLog`, `StreamedLogOptions` - Shared: `ListOptions`, `StorageListOptions` +- Server Actors: `ActorServerRequestOptions`, `ActorServerBody`, `ActorServerCompression` - Client configuration: `RequestCompression` plus the common container `PaginationList`, `ListIterator` @@ -148,7 +149,7 @@ Builder options: | `default_header(name, value)` | none | Header sent with every request. It wins over the built-in `User-Agent` and `Authorization` headers; a header a request sets itself wins over it. Names are compared case-insensitively. | | `request_compression(c)` | `Brotli` | Encoding for large request bodies: `RequestCompression::Brotli` (`br`) or `RequestCompression::Gzip` (`gzip`). | | `http_compressor(c)` | Brotli, quality 6 | A custom `Arc`: `BrotliHttpCompressor::new(quality)` (0 to 11, default 6), `GzipHttpCompressor::new(quality)` (1 to 9, default 6), or your own implementation. | -| `http_backend(b)` | reqwest | Replaceable HTTP transport (`apify_client::http_client::{HttpBackend, ReqwestBackend}`). A request whose `timeout` is `timeouts::NO_TIMEOUT` may run for as long as it takes, so a backend must check for it before it applies a timeout of its own. | +| `http_backend(b)` | reqwest | Replaceable HTTP transport (`apify_client::http_client::{HttpBackend, ReqwestBackend}`). A request whose `timeout` is `timeouts::NO_TIMEOUT` may run for as long as it takes, so a backend must check for it before it applies a timeout of its own. A request whose `follow_redirects` is `false`, such as one to an Actor's server, carries a token that must not follow a redirect to another host, so a backend must return a redirect response as it is. | Request bodies of at least 1024 bytes are compressed once (before retries) with the selected `request_compression` algorithm and sent with the matching `Content-Encoding` header. Brotli is @@ -272,7 +273,12 @@ iterates everything); to control the per-request page size, call `.with_chunk_si 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. +rather than cut off at the size it had when the loop started. Each page advances the offset by the +page's `count` or by the number of items it returned, whichever is larger, capped at the page size +requested. Without `with_chunk_size`, each request gets the API's default page size, capped by +`limit`. Dataset items have no default limit, so iterating `iterate_items()` without +`with_chunk_size` loads the whole dataset, or all `limit` items, in one response: set a page size +to iterate a large dataset. ```rust,no_run use apify_client::{ApifyClient, ActorListOptions}; diff --git a/docs/actors.md b/docs/actors.md index 242ca74..2a1b981 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -32,6 +32,8 @@ be an Actor ID or a `username~name` (or `username/name`) reference. | `runs()` | — | `RunCollectionClient` | The Actor's run collection. | | `version(n)` / `versions()` | `&str` / — | `ActorVersionClient` / collection | Version management. | | `webhooks()` | — | `WebhookCollectionClient` | The Actor's webhooks. | +| `server()` | — | `ActorServerClient` | Client for the server interface of a [Server Actor](#server-actors). | +| `standby()` | — | `ActorServerClient` | Deprecated alias of `server()`, named after Actor Standby, the former name of Server Actors. | ### `ActorStartOptions` @@ -190,6 +192,9 @@ example: | `is_public` | `Option` | Whether the Actor is published in Apify Store. | | `created_at` | `Option>` | When the Actor was created. | | `modified_at` | `Option>` | When the Actor was last modified. | +| `standby_url` | `Option` | URL for sending requests to the Actor's server, or `None` if the Standby mode isn't enabled. | +| `actor_standby` | `Option` | The Standby mode configuration: `is_enabled`, `desired_requests_per_actor_run`, `max_requests_per_actor_run`, `idle_timeout_secs`, `build`, `memory_mbytes`, `disable_standby_fields_override`, `should_pass_actor_input`, `tenancy` (`SINGLE_TENANT` or `MULTI_TENANT`), `is_console_auth_enabled` and `is_tokenless_enabled`. When creating an Actor, omit the last two or set them to `false`; when updating one, only Apify administrators can change them. | +| `tagged_builds` | `Option>>` | Maps build tag names (e.g. `latest`) to `TaggedBuildInfo { build_id, build_number, finished_at }`. | | `extra` | `Extra` | Any other fields returned by the API. | ```rust,no_run @@ -230,6 +235,121 @@ println!("build {} status {:?}", build.id, build.status); # } ``` +## Server Actors + +A Server Actor (formerly Actor Standby) runs a web server. You call it over HTTP, and the +platform keeps a run ready to answer, so each call skips the run startup. `ActorClient::server()` +returns an `ActorServerClient` for it. The Actor's `standbyUrl` and `actorStandby` fields still use +the old name. + +| Method | Arguments | Returns | Description | +|---|---|---|---| +| `request(path, options)` | `&str`, `ActorServerRequestOptions` | `HttpResponse` | Sends a request to the Actor's server and buffers the response. | +| `request_stream(path, options)` | `&str`, `ActorServerRequestOptions` | `StreamingHttpResponse` | Like `request`, with the response body as a stream. It is retried like `request` until the response starts, and not once the body is being read. Its timeout bounds only the wait for the response to start, so a server that stays silent between two chunks, such as one that sends events, is not cut off. | +| `get_url()` | — | `String` | The URL of the Actor's server. | +| `get_open_api_spec()` | — | `Option` | The Actor's web server schema, or `None` if its definition has none. | + +`ActorServerRequestOptions` has these fields, all optional: + +| Field | Type | Description | +|---|---|---| +| `method` | `Option` | HTTP method of the request. Default is `GET`. | +| `params` | `Vec<(String, String)>` | Query parameters of the request, in order. A key can repeat. | +| `headers` | `HashMap` | Additional headers of the request. They reach the Actor's server as they are, `Authorization` included. | +| `body` | `Option` | `Json(Value)` (sent with `Content-Type: application/json` unless you set a content type), `Text(String)` or `Bytes(Vec)` (both sent as they are). `ActorServerBody::json(&value)` serializes a value. | +| `compression` | `Option` | `Algorithm(RequestCompression)` or `Custom(Arc)`. The body is sent uncompressed when it's `None`, since the Actor's server may not decode a compressed body. Set it only if the server decodes the resulting `Content-Encoding`. | + +```rust,no_run +use apify_client::{ActorServerBody, ActorServerRequestOptions, ApifyClient}; +use apify_client::http_client::HttpMethod; +use serde_json::json; + +# async fn run() -> Result<(), Box> { +let client = ApifyClient::new("my-api-token"); +let server = client.actor("username/actor-name").server(); + +// Send a GET request to the /search endpoint of the Actor's server. +// The method returns error responses too, so check the status first. +let search = server + .request( + "/search", + ActorServerRequestOptions { + params: vec![("q".to_string(), "apify".to_string())], + ..Default::default() + }, + ) + .await?; +if !search.is_success() { + return Err(format!("Search failed with status {}", search.status).into()); +} +println!("{}", search.text()); + +// Send a POST request with a JSON body. +let items = server + .request( + "/items", + ActorServerRequestOptions { + method: Some(HttpMethod::Post), + body: Some(ActorServerBody::Json(json!({ "url": "https://apify.com" }))), + ..Default::default() + }, + ) + .await?; +println!("{}", items.status); +# Ok(()) +# } +``` + +`HttpResponse` has the `status`, the `headers` and the `body` bytes, with `header(name)`, +`is_success()`, `text()` and `json::()` to read them. The body of `request_stream` is a stream of +byte chunks, like the other streaming methods. + +How the request is sent: + +- The server URL is the Actor's `standby_url`. The first call of `request`, `get_url` or + `get_open_api_spec` fetches the Actor and every client returned by the same `ActorClient` reuses + the URL. A call on an Actor that isn't a Server Actor fails with `InvalidArgument`, and one on a + missing Actor fails with the `404` `ApiError`. +- The path is always resolved against that URL, so it can't send the request to another host. A + fragment (`#...`) is dropped. +- The token goes in the `X-Apify-Authorization` header, which the platform removes before it + forwards the request, so the token doesn't reach the Actor's code. The headers you pass reach + the Actor as they are. If the Actor reads the `Authorization` header for its own + authentication, pass that header in `headers`. +- Redirects are not followed, and a redirect response is returned as it is, since following it would + send the token to a host the Actor's code picked. +- The timeout defaults to the `Long` tier, to cover a cold start. Set another one with the + client's `with_timeout`. +- The server's response is returned whatever its status. Only `429` and `503` are retried, with + the client's retry settings, and the last response is returned once the retries run out. The + platform sends these when it can't pass the request to a server run, but the Actor's code can + return them too, and then a retry sends the request to the Actor again. Network errors and + timeouts are not retried, since the Actor may have already handled the request, and they fail + the call. + +`get_open_api_spec` returns the web server schema from the build that serves the server +interface: the build in the Actor's `actor_standby.build` field (a tag or a build number), or +`latest`. It fails with `InvalidArgument` if that build doesn't exist. + +```rust,no_run +# use apify_client::ApifyClient; +# async fn run() -> Result<(), Box> { +let client = ApifyClient::new("my-api-token"); +let server = client.actor("username/actor-name").server(); + +// Get the URL of the Actor's server. +println!("{}", server.get_url().await?); + +// Get the OpenAPI specification and list the server's endpoints. +if let Some(spec) = server.get_open_api_spec().await? { + if let Some(paths) = spec["paths"].as_object() { + println!("{:?}", paths.keys().collect::>()); + } +} +# Ok(()) +# } +``` + ## Actor versions and environment variables `ActorVersionClient`: `get`, `update`, `delete`, `env_var(name)`, `env_vars()`. @@ -237,6 +357,17 @@ println!("build {} status {:?}", build.id, build.status); `ActorEnvVarClient`: `get`, `update`, `delete`. `ActorEnvVarCollectionClient`: `list()`, `iterate()`, `create(env_var)`. +A version of the `SOURCE_FILES` or `TARBALL` source type can set `actorPath`, the Actor's directory +inside the uploaded source, for a monorepo: a path relative to the upload root with `/` separators, +for example `actors/typescript-actor`. It has 1 to 255 characters, with no leading, trailing, or +repeated `/` and no `.` or `..` segments, and the API rejects it for `GIT_REPO` and `GITHUB_GIST` +versions. Without it, the upload root is the Actor's directory. `ActorVersion::actor_path` holds it +(for a `TARBALL` version, only users with write access to the Actor get it). An update that sends +`sourceType`, `sourceFiles` or `tarballUrl` without `actorPath` removes the stored path, so send it +again with every upload. An update of other fields, such as `envVars` or `buildTag`, keeps it, and +`"actorPath": null` removes it. An `ActorVersion` sent as an update body leaves `actor_path` out +when it is `None`. + Creating a version needs `versionNumber` and `sourceType` in the body, and so does every item of the `versions` array in an Actor create or update body. The writable source types are `SOURCE_FILES`, `GIT_REPO`, `TARBALL` and `GITHUB_GIST`; `SOURCE_CODE` can be read but not sent. diff --git a/docs/storages.md b/docs/storages.md index 7de9da1..d9029fb 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -70,8 +70,8 @@ let scratch = client.datasets().get_or_create(None).await?; | `get_statistics()` | — | `Value` | Field statistics. A `404` is an error. | | `download_items(format, options)` | `DownloadItemsFormat`, `DatasetDownloadOptions` | `Vec` | Export items as JSON/JSONL/CSV/XLSX/XML/RSS/HTML. | | `download_items_stream(format, options)` | `DownloadItemsFormat`, `DatasetDownloadOptions` | `Result>>>` (async — `.await` it) | Like `download_items`, but streams the export instead of buffering it, for constant-memory processing of a large export. | -| `create_items_public_url(options, expires)` | `DatasetListItemsOptions`, `Option` | `String` | Shareable (HMAC-signed for private) items URL, served as `json`. | -| `create_items_public_url_with_format(options, expires, format)` | `DatasetListItemsOptions`, `Option`, `Option` | `String` | Like `create_items_public_url`, but lets the URL serve another export format (CSV, XLSX, ...). `None` defaults to `json`. | +| `create_items_public_url(options, expires)` | `DatasetListItemsOptions`, `Option` | `String` | Shareable (HMAC-signed for private) items URL, served as `json`. `expires` is the validity of the signed URL in seconds and must be positive and at most `common::MAX_EXPIRES_IN_SECS` (4611686018427387, about 146 million years): `0`, a negative or a larger value fails with `InvalidArgument`. Leave it out for a URL that does not expire. | +| `create_items_public_url_with_format(options, expires, format)` | `DatasetListItemsOptions`, `Option`, `Option` | `String` | Like `create_items_public_url`, but lets the URL serve another export format (CSV, XLSX, ...). `None` defaults to `json`. `expires` follows the same rule as in `create_items_public_url`. | `DatasetListItemsOptions` (all optional): @@ -164,7 +164,7 @@ println!("streamed {total_bytes} bytes of JSONL"); | `delete_record(key)` | `&str` | `()` | Deletes a record. | | `get_record_with_options(key, options)` | `&str`, `GetRecordOptions { attachment, signature }` | `Option` | Reads a record with explicit attachment/signature options. | | `get_record_public_url(key)` | `&str` | `String` | Shareable (HMAC-signed for private) record URL. | -| `create_keys_public_url(expires)` | `Option` | `String` | Shareable keys-list URL. | +| `create_keys_public_url(expires)` | `Option` | `String` | Shareable keys-list URL. `expires` is the validity of the signed URL in seconds and must be positive and at most `common::MAX_EXPIRES_IN_SECS` (4611686018427387, about 146 million years): `0`, a negative or a larger value fails with `InvalidArgument`. Leave it out for a URL that does not expire. | `ListKeysOptions`: `limit`, `exclusive_start_key`, `prefix`, `collection`, `signature`. Like `StoreListOptions.limit`, the meaning of `limit` depends on the method: for `list_keys` it is a @@ -397,7 +397,7 @@ return `PaginationList`. Re-exported at the crate root (`apify_client::Pagina | `total` | `i64` | Total items available across all pages. | | `offset` | `i64` | Items skipped at the start. | | `limit` | `i64` | Max items the API would return for this request. | -| `count` | `i64` | Items actually returned in this page. | +| `count` | `i64` | Entries on this page. On a page from `list_items`, the number of dataset rows the API scanned for it, or the number of items returned when that is larger: filters (`clean`, `skip_empty`, `skip_hidden`) can leave it above `items.len()`. | | `desc` | `bool` | Whether the items are in descending order. | | `items` | `Vec` | The items of this page. | diff --git a/src/clients/actor.rs b/src/clients/actor.rs index cdcb273..ce804d0 100644 --- a/src/clients/actor.rs +++ b/src/clients/actor.rs @@ -2,8 +2,11 @@ use serde::Serialize; use serde_json::Value; +use std::sync::Arc; +use tokio::sync::OnceCell; use crate::client::ApifyClient; +use crate::clients::actor_server::ActorServerClient; use crate::clients::actor_version::ActorVersionClient; use crate::clients::actor_version_collection::ActorVersionCollectionClient; use crate::clients::base::{ @@ -190,6 +193,9 @@ pub struct ActorClient { ctx: ResourceContext, base_url: String, id: String, + /// The server URL fetched by the clients returned by [`server`](Self::server), shared between + /// them so it is looked up once. + server_url: Arc>, } crate::clients::base::impl_with_timeout!(ActorClient); @@ -201,6 +207,7 @@ impl ActorClient { ctx: ResourceContext::single(http, base_url, "actors", id), base_url: base_url.to_string(), id: id.to_string(), + server_url: Arc::new(OnceCell::new()), } } @@ -521,6 +528,29 @@ impl ActorClient { WebhookCollectionClient::with_base(self.ctx.http.clone(), &self.ctx.url(None)) } + /// Returns a client for the server interface of this Actor. + /// + /// A Server Actor runs a web server, its server interface, that is called over HTTP without + /// starting a run for each call. Server Actors were formerly called Actor Standby. The clients + /// returned by this method and by [`standby`](Self::standby) share the server URL they fetch. + /// They do not inherit the timeout set with `with_timeout`. + pub fn server(&self) -> ActorServerClient { + ActorServerClient::new( + self.ctx.http.clone(), + &self.base_url, + &self.id, + self.server_url.clone(), + ) + } + + /// Returns a client for the server interface of this Actor in Standby mode. + /// + /// Named after Actor Standby, the former name of Server Actors. + #[deprecated(note = "Use `server` instead.")] + pub fn standby(&self) -> ActorServerClient { + self.server() + } + /// The Actor's ID (or `username~name`) as provided. pub fn id(&self) -> &str { &self.id diff --git a/src/clients/actor_server.rs b/src/clients/actor_server.rs new file mode 100644 index 0000000..761a77c --- /dev/null +++ b/src/clients/actor_server.rs @@ -0,0 +1,467 @@ +//! Client for the server interface of an Actor (`/v2/actors/{actorId}` and the Actor's `standbyUrl`). +//! +//! A Server Actor (formerly Actor Standby) runs a web server. It is called over HTTP, and the +//! platform keeps a run ready to answer, so each call skips the run startup. + +use std::collections::HashMap; +use std::sync::Arc; + +use serde::Serialize; +use serde_json::Value; +use tokio::sync::OnceCell; + +use crate::clients::base::ResourceContext; +use crate::clients::build::BuildClient; +use crate::clients::build_collection::BuildCollectionClient; +use crate::common::{parse_data_envelope, ListOptions, QueryParams}; +use crate::error::{ApifyClientError, ApifyClientResult}; +use crate::http_client::{ + AuthorizationHeader, BodyCompression, CallOptions, HttpClient, HttpMethod, HttpResponse, + StreamingHttpResponse, +}; +use crate::http_compressors::{HttpCompressor, RequestCompression}; +use crate::models::Actor; +use crate::timeouts::Timeout; + +/// Statuses a server request is retried on. The platform answers with them when it can't pass a +/// request to a server run, so a retry usually doesn't run the Actor's code again. An Actor that +/// returns one itself gets the request again. +const SERVER_RETRY_STATUS_CODES: &[u16] = &[429, 503]; + +/// The tag of the build that serves the server interface when the Actor's `actorStandby.build` +/// names none. +const DEFAULT_SERVER_BUILD: &str = "latest"; + +/// The body of a request to an Actor's server. +#[derive(Debug, Clone)] +pub enum ActorServerBody { + /// A JSON value, sent serialized with `Content-Type: application/json` unless the request + /// sets a content type. + Json(Value), + /// Text, sent as it is. Set the `Content-Type` header of the request to describe it. + Text(String), + /// Binary data, sent as it is. Set the `Content-Type` header of the request to describe it. + Bytes(Vec), +} + +impl ActorServerBody { + /// Serializes `value` into a [`Json`](Self::Json) body. + pub fn json(value: &T) -> ApifyClientResult { + Ok(Self::Json(serde_json::to_value(value)?)) + } +} + +/// How the body of a request to an Actor's server is compressed. +#[derive(Debug, Clone)] +pub enum ActorServerCompression { + /// One of the built-in algorithms. + Algorithm(RequestCompression), + /// A custom compressor. + Custom(Arc), +} + +/// Options of [`ActorServerClient::request`] and [`ActorServerClient::request_stream`]. +#[derive(Debug, Default, Clone)] +pub struct ActorServerRequestOptions { + /// HTTP method of the request. Default is `GET`. + pub method: Option, + /// Query parameters of the request, in order. A key can repeat. + pub params: Vec<(String, String)>, + /// Additional headers of the request. They reach the Actor's server as they are, + /// `Authorization` included. + pub headers: HashMap, + /// The request body. + pub body: Option, + /// Compression of the request body. The body is sent uncompressed when it's `None`, since the + /// Actor's server may not decode a compressed body. Set it only if the Actor's server decodes + /// the resulting `Content-Encoding`. The body is compressed under the same conditions as for + /// API calls: when it is large enough, no `Content-Encoding` header is set and its content + /// type isn't already compressed. + pub compression: Option, +} + +/// A request to the server, ready to be sent. +struct PreparedServerRequest { + method: HttpMethod, + url: String, + headers: HashMap, + body: Option>, + timeout: std::time::Duration, + call_options: CallOptions, +} + +/// Client for the server interface of a specific Actor. +/// +/// Provides methods to work with the web server of a Server Actor, such as sending it requests or +/// getting its OpenAPI specification. Get an instance with +/// [`ActorClient::server`](crate::clients::actor::ActorClient::server). +/// +/// ```no_run +/// use apify_client::clients::actor_server::ActorServerRequestOptions; +/// use apify_client::ApifyClient; +/// +/// # async fn run() -> Result<(), Box> { +/// let client = ApifyClient::new("my-api-token"); +/// let server = client.actor("username/actor-name").server(); +/// +/// let response = server +/// .request( +/// "/search", +/// ActorServerRequestOptions { +/// params: vec![("q".to_string(), "apify".to_string())], +/// ..Default::default() +/// }, +/// ) +/// .await?; +/// println!("{}", response.text()); +/// # Ok(()) +/// # } +/// ``` +#[derive(Debug, Clone)] +pub struct ActorServerClient { + ctx: ResourceContext, + base_url: String, + id: String, + /// The server URL. Getting it takes an API call for the Actor, so it is cached for + /// [`get_url`](Self::get_url) and [`request`](Self::request). Concurrent first calls share one + /// API request, and a failed lookup is not kept. + server_url: Arc>, +} + +crate::clients::base::impl_with_timeout!(ActorServerClient); + +impl ActorServerClient { + pub(crate) fn new( + http: HttpClient, + base_url: &str, + id: &str, + server_url: Arc>, + ) -> Self { + Self { + ctx: ResourceContext::single(http, base_url, "actors", id), + base_url: base_url.to_string(), + id: id.to_string(), + server_url, + } + } + + /// Returns the URL of the Actor's server. + /// + /// The URL is the Actor's `standbyUrl`, fetched from the API by the first call of this method + /// or [`get_open_api_spec`](Self::get_open_api_spec) and reused by later calls on the clients + /// of the same [`ActorClient`](crate::clients::actor::ActorClient). Use it to call the server + /// with another HTTP client, for example one generated from the server's OpenAPI + /// specification. + /// + /// Fails with [`ApifyClientError::InvalidArgument`] when the Actor is not a Server Actor, and + /// with a `404` [`ApiError`](crate::ApiError) when the Actor does not exist. + pub async fn get_url(&self) -> ApifyClientResult { + self.server_url + .get_or_try_init(|| async { Ok(self.fetch_server_actor().await?.1) }) + .await + .cloned() + } + + /// Returns the OpenAPI specification of the Actor's server, or `None` if the Actor's + /// definition doesn't include one. + /// + /// The specification is the web server schema from the Actor's definition, taken from the + /// build that serves the server interface: the build set in `actorStandby.build`, or `latest` + /// if none is set. Use it to generate an HTTP client for the server, and point that client at + /// the URL returned by [`get_url`](Self::get_url). + /// + /// Fails with [`ApifyClientError::InvalidArgument`] when the Actor is not a Server Actor, or + /// the build that serves its server interface doesn't exist, and with a `404` + /// [`ApiError`](crate::ApiError) when the Actor does not exist. + pub async fn get_open_api_spec(&self) -> ApifyClientResult> { + let (actor, server_url) = self.fetch_server_actor().await?; + // Lets a later `get_url` and `request` skip fetching the Actor again. The URL is already + // set when an earlier call got there first, which is fine. + let _ = self.server_url.set(server_url); + let build_ref = actor + .actor_standby + .as_ref() + .and_then(|standby| standby.build.as_deref()) + .filter(|build| !build.is_empty()) + .unwrap_or(DEFAULT_SERVER_BUILD); + + let mut build_id = actor + .tagged_builds + .as_ref() + .and_then(|tags| tags.get(build_ref)) + .and_then(|info| info.as_ref()) + .and_then(|info| info.build_id.clone()); + if build_id.is_none() { + if let Some(target) = parse_build_number(build_ref) { + // Not a tag, so look `build_ref` up as a build number, which the API can't + // resolve directly. + build_id = self.find_build_by_number(build_ref, target).await?; + } + } + + let build = match build_id { + Some(id) => self.build_client(&id).get().await?, + None => None, + }; + let build = build.ok_or_else(|| { + ApifyClientError::InvalidArgument(format!( + "Build {build_ref} of Actor {} does not exist.", + self.id + )) + })?; + + Ok(build + .extra + .get("actorDefinition") + .and_then(|definition| definition.get("webServerSchema")) + .filter(|schema| !schema.is_null()) + .cloned()) + } + + /// Sends an HTTP request to the Actor's server and buffers the response. + /// + /// The request goes to the URL returned by [`get_url`](Self::get_url) and carries the client's + /// token in the `X-Apify-Authorization` header, which the platform consumes, so it never + /// reaches the Actor's code. Headers in the options, `Authorization` included, reach the + /// Actor's server as they are. The first request to an idle Actor waits for the platform to + /// start a server run, which is why the timeout defaults to the `Long` tier. + /// + /// `path` is the path on the Actor's server, for example `/search`. It is always resolved + /// against the server URL, so the request, and the token with it, can't be sent to another + /// host. Pass the query in [`ActorServerRequestOptions::params`]. A fragment (`#...`) is + /// dropped, since it's never sent to the server. + /// + /// The method returns any response of the server, error statuses included, so check + /// [`HttpResponse::status`] before you use the body. A rate-limit (`429`) or Service + /// Unavailable (`503`) response is retried with the client's retry settings, and the last one + /// is returned once the retries run out. The Actor's code can return these statuses too, and + /// then it gets the request again. Network errors and timeouts are not retried, since the + /// Actor may have already handled the request. A redirect is returned as it is, since + /// following it would send the token to a host the Actor's code picked. + /// + /// It fails only when it gets no response from the server: with + /// [`ApifyClientError::InvalidArgument`] when the Actor is not a Server Actor, with a `404` + /// [`ApiError`](crate::ApiError) when the Actor does not exist, and with the transport error + /// when the request fails or times out. + pub async fn request( + &self, + path: &str, + options: ActorServerRequestOptions, + ) -> ApifyClientResult { + let request = self.prepare_request(path, options).await?; + self.ctx + .send_with_options( + request.method, + request.url, + request.headers, + request.body, + request.timeout, + request.call_options, + ) + .await + } + + /// Like [`request`](Self::request), but returns the response body as a stream instead of + /// buffering it. + /// + /// The response is returned with whatever status the server answers, and is retried like + /// [`request`](Self::request) until it starts. Once the body is being read, it is not, since a + /// stream that is already open can't be replayed. The stream holds its connection open until + /// you read it to the end or drop it. The timeout of the request bounds only the wait for the + /// server to start responding, so a server that keeps the connection open and stays silent + /// between two chunks, such as one that sends events, is not cut off. + pub async fn request_stream( + &self, + path: &str, + options: ActorServerRequestOptions, + ) -> ApifyClientResult { + let request = self.prepare_request(path, options).await?; + self.ctx + .send_streaming_with_options( + request.method, + request.url, + request.headers, + request.body, + request.timeout, + &request.call_options, + ) + .await + } + + /// Resolves the URL and serializes the body and the query of a request to the server. + async fn prepare_request( + &self, + path: &str, + options: ActorServerRequestOptions, + ) -> ApifyClientResult { + let server_url = self.get_url().await?; + // A fragment is never sent to the server, and a leading `/` would only be doubled. + let server_path = path + .split('#') + .next() + .unwrap_or_default() + .trim_start_matches('/'); + let mut params = QueryParams::new(); + for (key, value) in options.params { + params.push_raw(key, value); + } + let url = params.apply_to_url(&format!( + "{}/{server_path}", + server_url.trim_end_matches('/') + )); + + let mut headers = options.headers; + let body = match options.body { + None => None, + Some(ActorServerBody::Json(value)) => { + if !headers + .keys() + .any(|name| name.eq_ignore_ascii_case("Content-Type")) + { + headers.insert("Content-Type".to_string(), "application/json".to_string()); + } + Some(serde_json::to_vec(&value)?) + } + Some(ActorServerBody::Text(text)) => Some(text.into_bytes()), + Some(ActorServerBody::Bytes(bytes)) => Some(bytes), + }; + + let compression = match options.compression { + None => BodyCompression::Disabled, + Some(ActorServerCompression::Algorithm(algorithm)) => { + BodyCompression::With(algorithm.into_compressor()) + } + Some(ActorServerCompression::Custom(compressor)) => BodyCompression::With(compressor), + }; + Ok(PreparedServerRequest { + method: options.method.unwrap_or(HttpMethod::Get), + url, + headers, + body, + timeout: self.ctx.timeout(Timeout::Long), + call_options: CallOptions { + compression, + authorization_header: AuthorizationHeader::XApifyAuthorization, + return_error_responses: true, + retry_status_codes: Some(SERVER_RETRY_STATUS_CODES), + do_not_retry_transport_errors: true, + do_not_follow_redirects: true, + no_stream_inactivity_timeout: true, + ..CallOptions::default() + }, + }) + } + + /// Fetches the Actor and checks that it is a Server Actor. Returns the Actor and its server + /// URL. + async fn fetch_server_actor(&self) -> ApifyClientResult<(Actor, String)> { + let response = self + .ctx + .send( + HttpMethod::Get, + self.ctx.url(None), + self.ctx.timeout(Timeout::Short), + ) + .await?; + let actor: Actor = parse_data_envelope(&response.body)?; + match actor.standby_url.clone().filter(|url| !url.is_empty()) { + Some(url) => Ok((actor, url)), + None => Err(ApifyClientError::InvalidArgument(format!( + "Actor {} is not a Server Actor.", + self.id + ))), + } + } + + /// Walks the Actor's builds, newest first, for the one with `build_number`. + async fn find_build_by_number( + &self, + build_number: &str, + target: [u64; 3], + ) -> ApifyClientResult> { + let mut builds = self.builds().iterate(ListOptions { + desc: Some(true), + ..ListOptions::default() + }); + while let Some(build) = builds.next().await? { + if build.build_number.as_deref() == Some(build_number) { + return Ok(Some(build.id)); + } + if is_older_build(build.build_number.as_deref(), target) { + break; + } + } + Ok(None) + } + + /// The build collection of the Actor, with this client's timeout. Without one, it lists with + /// the `Short` tier, like every other API request of this client, where the collection client + /// defaults to `Medium`. + fn builds(&self) -> BuildCollectionClient { + BuildCollectionClient::with_base(self.ctx.http.clone(), &self.ctx.url(None), "builds") + .with_timeout(self.ctx.timeout.unwrap_or(Timeout::Short)) + } + + /// A client for the build `id`, with this client's timeout. + fn build_client(&self, id: &str) -> BuildClient { + let build = BuildClient::new(self.ctx.http.clone(), &self.base_url, id); + match self.ctx.timeout { + Some(timeout) => build.with_timeout(timeout), + None => build, + } + } +} + +/// Splits a build number like `0.1.23` into its version and sequence number, or returns `None` +/// for anything else. +fn parse_build_number(build_number: &str) -> Option<[u64; 3]> { + let parts: Vec = build_number + .split('.') + .map(parse_digits) + .collect::>()?; + parts.try_into().ok() +} + +/// Parses a non-empty run of ASCII digits. `str::parse` alone would also accept a leading `+`. +fn parse_digits(part: &str) -> Option { + if part.is_empty() || !part.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + part.parse().ok() +} + +/// Returns whether `build_number` belongs to the same version as `target` and comes before it. +/// Builds are listed newest first and the sequence numbers of a version grow over time, so +/// meeting such a build means the target doesn't exist. +fn is_older_build(build_number: Option<&str>, target: [u64; 3]) -> bool { + build_number + .and_then(parse_build_number) + .is_some_and(|parsed| { + parsed[0] == target[0] && parsed[1] == target[1] && parsed[2] < target[2] + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_numbers_are_parsed_strictly() { + assert_eq!(parse_build_number("0.1.23"), Some([0, 1, 23])); + assert_eq!(parse_build_number("latest"), None); + assert_eq!(parse_build_number("0.1"), None); + assert_eq!(parse_build_number("0.1.2.3"), None); + assert_eq!(parse_build_number("0.+1.2"), None); + assert_eq!(parse_build_number("0..2"), None); + } + + #[test] + fn only_an_earlier_build_of_the_same_version_ends_the_search() { + let target = [0, 1, 5]; + assert!(is_older_build(Some("0.1.4"), target)); + assert!(!is_older_build(Some("0.1.5"), target)); + assert!(!is_older_build(Some("0.1.6"), target)); + assert!(!is_older_build(Some("0.2.1"), target)); + assert!(!is_older_build(None, target)); + } +} diff --git a/src/clients/base.rs b/src/clients/base.rs index 98005bf..29605b2 100644 --- a/src/clients/base.rs +++ b/src/clients/base.rs @@ -15,7 +15,9 @@ use crate::common::{ catch_not_found, parse_data_envelope, to_path_segment, to_safe_id, PaginationList, QueryParams, }; use crate::error::{ApifyClientError, ApifyClientResult}; -use crate::http_client::{CallOptions, HttpClient, HttpMethod, HttpRequest, HttpResponse}; +use crate::http_client::{ + CallOptions, HttpClient, HttpMethod, HttpRequest, HttpResponse, StreamingHttpResponse, +}; use crate::timeouts::Timeout; /// How long to wait between polls while waiting for a run/build to finish. @@ -261,6 +263,37 @@ impl ResourceContext { headers, body, timeout, + follow_redirects: !options.do_not_follow_redirects, + }, + options, + ) + .await + } + + /// Like [`send_with_options`](Self::send_with_options), for a response whose body the caller + /// reads incrementally: retries until the response starts, then returns the response, or the + /// error of a failed one, like [`send_with_options`](Self::send_with_options). With + /// [`CallOptions::return_error_responses`] it returns the response of any status that no retry + /// follows (see [`HttpClient::call_streaming_response`]). + pub(crate) async fn send_streaming_with_options( + &self, + method: HttpMethod, + url: String, + headers: HashMap, + body: Option>, + timeout: Duration, + options: &CallOptions, + ) -> ApifyClientResult { + self.check_request_allowed()?; + self.http + .call_streaming_response( + HttpRequest { + method, + url, + headers, + body, + timeout, + follow_redirects: !options.do_not_follow_redirects, }, options, ) @@ -474,6 +507,7 @@ pub(crate) async fn post_creating( ) -> ApifyClientResult { let options = CallOptions { do_not_retry_timeouts: true, + ..CallOptions::default() }; let response = post_send( ctx, @@ -638,6 +672,7 @@ pub(crate) async fn get_byte_stream( headers: Default::default(), body: None, timeout: ctx.timeout(tier), + follow_redirects: true, }) .await } diff --git a/src/clients/dataset.rs b/src/clients/dataset.rs index a5f04ec..2e12efd 100644 --- a/src/clients/dataset.rs +++ b/src/clients/dataset.rs @@ -9,7 +9,10 @@ use crate::clients::base::{ ResourceContext, }; use crate::clients::pagination::ListIterator; -use crate::common::{sign_storage_content, PaginationList, QueryParams}; +use crate::common::{ + public_url_expiry_millis, sign_storage_content, validate_expires_in_secs, PaginationList, + QueryParams, +}; use crate::error::{ApifyClientError, ApifyClientResult}; use crate::http_client::{CallOptions, HttpClient, HttpMethod}; use crate::models::Dataset; @@ -285,10 +288,10 @@ impl DatasetClient { .await?; let items: Vec = serde_json::from_slice(&response.body)?; - let count = items.len() as i64; - // `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 returned = items.len() as i64; + // `PaginationList::total` is informational only here (`ListIterator` never consults it); + // falling back to `0` rather than `returned` 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()) @@ -300,13 +303,15 @@ impl DatasetClient { let limit = response .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 + .unwrap_or(returned); + // `x-apify-pagination-count` reports the rows scanned, which filters and `unwind` can leave + // above or below the items returned. It is derived from the dataset's item count, which + // lags a fresh push, so it can also fall below the items returned on an unfiltered page. + let count = response .header("x-apify-pagination-count") - .and_then(|v| v.parse().ok()); + .and_then(|v| v.parse().ok()) + .unwrap_or(0) + .max(returned); Ok(PaginationList { total, @@ -315,7 +320,6 @@ impl DatasetClient { count, desc: options.desc.unwrap_or(false), items, - scanned, }) } @@ -324,7 +328,9 @@ impl DatasetClient { /// The idiomatic-Rust counterpart of the reference client's async-iterable /// `listItems`/`iterateItems`: yields one deserialized item of type `T` at a time, /// 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. + /// (unset = all); use [`ListIterator::with_chunk_size`] to control the per-page fetch size. Dataset + /// items have no default page limit, so without a page size the whole dataset, or all `limit` + /// items, is loaded in one response: set one to iterate a large dataset. /// /// 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 @@ -432,6 +438,7 @@ impl DatasetClient { self.ctx.timeout(Timeout::Medium), CallOptions { do_not_retry_timeouts: true, + ..CallOptions::default() }, ) .await?; @@ -446,6 +453,12 @@ impl DatasetClient { /// optionally bounds the validity of a signed URL. The URL is built from the configured /// public base URL. The served items are JSON; to request another export format, use /// [`create_items_public_url_with_format`](Self::create_items_public_url_with_format). + /// + /// `expires_in_secs` must be positive and at most + /// [`MAX_EXPIRES_IN_SECS`](crate::common::MAX_EXPIRES_IN_SECS): `0`, a negative or a larger + /// value fails with + /// [`InvalidArgument`](crate::ApifyClientError::InvalidArgument) before any request is sent. + /// Leave it out for a URL that does not expire. pub async fn create_items_public_url( &self, options: DatasetListItemsOptions, @@ -458,13 +471,16 @@ impl DatasetClient { /// Like [`create_items_public_url`](Self::create_items_public_url), but lets the caller /// choose the `format` the items are served in (JSON, CSV, XLSX, ...) via a `format` query /// parameter on the generated URL. `None` defaults to `json`, matching the endpoint's own - /// default. + /// default. `expires_in_secs` follows the same rule as in + /// [`create_items_public_url`](Self::create_items_public_url): positive, and at most + /// [`MAX_EXPIRES_IN_SECS`](crate::common::MAX_EXPIRES_IN_SECS). pub async fn create_items_public_url_with_format( &self, options: DatasetListItemsOptions, expires_in_secs: Option, format: Option, ) -> ApifyClientResult { + validate_expires_in_secs(expires_in_secs)?; let mut params = QueryParams::new(); options.apply(&mut params); params.add_str("format", format.map(|f| f.as_str().to_string())); @@ -478,7 +494,11 @@ impl DatasetClient { .get("urlSigningSecretKey") .and_then(|v| v.as_str()) { - let signature = sign_storage_content(secret, &dataset.id, expires_in_secs); + let signature = sign_storage_content( + secret, + &dataset.id, + public_url_expiry_millis(expires_in_secs), + ); params.add_str("signature", Some(signature)); } } diff --git a/src/clients/key_value_store.rs b/src/clients/key_value_store.rs index eef583d..5db3bfa 100644 --- a/src/clients/key_value_store.rs +++ b/src/clients/key_value_store.rs @@ -8,7 +8,10 @@ use crate::clients::base::{ delete_resource, get_raw, get_resource, get_resource_required, head_exists, put_raw, update_resource, RawUpload, ResourceContext, }; -use crate::common::{create_hmac_signature, sign_storage_content, to_path_segment, QueryParams}; +use crate::common::{ + create_hmac_signature, public_url_expiry_millis, sign_storage_content, to_path_segment, + validate_expires_in_secs, QueryParams, +}; use crate::error::{ApifyClientError, ApifyClientResult}; use crate::http_client::{CallOptions, HttpClient, HttpMethod}; use crate::models::{KeyValueStore, KeyValueStoreKey, KeyValueStoreKeysPage, KeyValueStoreRecord}; @@ -277,6 +280,7 @@ impl KeyValueStoreClient { }, CallOptions { do_not_retry_timeouts: options.do_not_retry_timeouts, + ..CallOptions::default() }, Timeout::Long, ) @@ -347,11 +351,16 @@ impl KeyValueStoreClient { /// /// Like [`get_record_public_url`](Self::get_record_public_url), signs the URL with an /// HMAC-SHA256 `signature` for private stores. `expires_in_secs` optionally bounds a - /// signed URL's validity. + /// signed URL's validity. It must be positive and at most + /// [`MAX_EXPIRES_IN_SECS`](crate::common::MAX_EXPIRES_IN_SECS): `0`, a negative or a larger + /// value fails with + /// [`InvalidArgument`](crate::ApifyClientError::InvalidArgument) before any request is sent. + /// Leave it out for a URL that does not expire. pub async fn create_keys_public_url( &self, expires_in_secs: Option, ) -> ApifyClientResult { + validate_expires_in_secs(expires_in_secs)?; let mut params = QueryParams::new(); if let Some(store) = self.get_for_public_url().await? { if let Some(secret) = store @@ -359,7 +368,11 @@ impl KeyValueStoreClient { .get("urlSigningSecretKey") .and_then(|v| v.as_str()) { - let signature = sign_storage_content(secret, &store.id, expires_in_secs); + let signature = sign_storage_content( + secret, + &store.id, + public_url_expiry_millis(expires_in_secs), + ); params.add_str("signature", Some(signature)); } } diff --git a/src/clients/log.rs b/src/clients/log.rs index 6901ea1..ecc2552 100644 --- a/src/clients/log.rs +++ b/src/clients/log.rs @@ -144,7 +144,7 @@ impl LogClient { ) -> ApifyClientResult>>> { // Goes through `get_byte_stream` (the configured `HttpBackend`'s streaming path) instead // of `HttpClient::call`, whose buffered backend fully reads a response before returning - // it. The retry policy does not apply to an open stream. + // it. The retry policy applies until the response starts, not to an open stream. let mut params = QueryParams::new(); params.push_raw("stream".to_string(), "1".to_string()); params diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 35918d3..2d91ae1 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -10,6 +10,7 @@ pub mod actor; pub mod actor_collection; pub mod actor_env_var; pub mod actor_env_var_collection; +pub mod actor_server; pub mod actor_version; pub mod actor_version_collection; pub mod build; diff --git a/src/clients/pagination.rs b/src/clients/pagination.rs index dc0cc1f..ecc7f75 100644 --- a/src/clients/pagination.rs +++ b/src/clients/pagination.rs @@ -100,7 +100,11 @@ fn min_positive_limit(a: Option, b: Option) -> Option { /// listing. The page size is a distinct concern: set it with [`with_chunk_size`](Self::with_chunk_size); /// when unset, the API's default page size is used. So `iterate(opts{ limit: 10 })` yields at most /// 10 items, and `iterate(opts).with_chunk_size(50)` fetches 50 per request while yielding -/// everything. +/// everything. Without a page size, each request gets the API's default page size, capped by +/// the remaining `limit`. Dataset items have no default limit, so iterating +/// [`DatasetClient::iterate_items`](crate::clients::dataset::DatasetClient::iterate_items) +/// without a page size loads the whole dataset, or all `limit` items, in one response: set one +/// to iterate a large dataset. /// /// **Large caps and the first page.** When a total cap is set but no page size is, the first page /// requests `limit == cap` (the reference client does the same, via @@ -177,15 +181,22 @@ impl ListIterator { } /// Returns the next item, or `None` when the listing is exhausted. Fetches another page from - /// the API when the local buffer is empty. + /// the API when the local buffer is empty, and keeps fetching while the pages come back empty + /// but scanned rows, as a filter can leave a page with no items in front of rows that match. pub async fn next(&mut self) -> ApifyClientResult> { - if let Some(item) = self.buffer.pop_front() { - return Ok(Some(item)); - } - if self.exhausted { - return Ok(None); + loop { + if let Some(item) = self.buffer.pop_front() { + return Ok(Some(item)); + } + if self.exhausted { + return Ok(None); + } + self.fetch_next_page().await?; } + } + /// Fetches the page at the current offset into the buffer and advances the offset. + async fn fetch_next_page(&mut self) -> ApifyClientResult<()> { // Request the smaller of the items still allowed under the caller's cap (`remaining`) and // the configured page size (`chunk_size`); `None` lets the API pick its default page size. // On the first page `remaining` is the full cap, matching the reference client's initial @@ -194,13 +205,15 @@ 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); + // The number of rows this page covered, used to advance the offset and decide when the + // listing is exhausted: the larger of `count` and the items returned, capped at the page + // size actually requested. Dataset items report the rows scanned in `count` (see + // `PaginationList::count`), since a filter can scan more rows than it returns and + // `unwind` can return more items than rows scanned; the cap keeps an unwound page from + // advancing past rows the next request would then skip. Other listings report the items + // returned, except the store, which drops entries after counting them: stepping by + // `count` keeps the next page from reading them again. + let scanned = page.count.max(received); let scanned = match page_limit { Some(limit) if limit > 0 => scanned.min(limit), _ => scanned, @@ -239,7 +252,7 @@ impl ListIterator { } self.buffer.extend(items); - Ok(self.buffer.pop_front()) + Ok(()) } /// Eagerly drains the iterator into a single `Vec`, fetching every remaining page. @@ -289,7 +302,6 @@ mod tests { count: items.len() as i64, desc: false, items, - scanned: None, }) }) }) @@ -373,7 +385,6 @@ mod tests { count: items.len() as i64, desc: false, items, - scanned: None, }) }) }); @@ -384,10 +395,10 @@ mod tests { } #[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`) + async fn advances_by_count_when_it_exceeds_items_returned() { + // Models a dataset `listItems` filter: each page scans 3 rows (reported in `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 + // alone would re-request the same rows forever; advancing by `count` moves past // them. let calls = Arc::new(AtomicUsize::new(0)); let counter = calls.clone(); @@ -406,23 +417,22 @@ mod tests { total: 0, offset, limit: 3, - count: items.len() as i64, + count: scanned, 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. + // Offsets requested: 0, 3, 6, 9 (empty) — four calls, each advancing by the 3 rows counted + // 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 + async fn count_is_capped_at_the_requested_page_size() { + // A lagging `count` 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)); @@ -442,10 +452,9 @@ mod tests { total: 0, offset, limit: page_size, - count: items.len() as i64, + count: scanned, desc: false, items, - scanned: Some(scanned), }) }) }); @@ -485,7 +494,6 @@ mod tests { count: items.len() as i64, desc: false, items, - scanned: None, }) }) }); @@ -545,7 +553,6 @@ mod tests { count: 5, desc: false, items: vec![0, 1, 2, 3, 4], - scanned: None, }) }) }); @@ -582,7 +589,6 @@ mod tests { count: 3, desc: false, items: vec![10, 20, 30], - scanned: None, }) }) }); diff --git a/src/clients/run_dataset_items.rs b/src/clients/run_dataset_items.rs index 3df857b..4a21601 100644 --- a/src/clients/run_dataset_items.rs +++ b/src/clients/run_dataset_items.rs @@ -228,7 +228,7 @@ impl RunDatasetItemsIterator { // Only an empty page marks the end, as filters can shorten a full one. A page that // `clean`, `skip_empty` or `unwind` emptied past a lagging `itemCount` reports no scanned // rows either, so a plain read checks. - let is_empty = page.items.is_empty() && page.scanned.unwrap_or(0) == 0; + let is_empty = page.count == 0; self.buffer.extend(page.items); if is_empty && !self.filters_can_hide_rows() { self.phase = Phase::Done; diff --git a/src/common.rs b/src/common.rs index d3a1f62..65669b3 100644 --- a/src/common.rs +++ b/src/common.rs @@ -213,7 +213,12 @@ pub struct PaginationList { /// Maximum number of items the API would return for this request. #[serde(default)] pub limit: i64, - /// Number of items actually returned in this page. + /// Number of entries on this page. On a page from + /// [`DatasetClient::list_items`](crate::clients::dataset::DatasetClient::list_items), the + /// number of dataset rows the API scanned for it, or the number of items returned when that + /// is larger. Filters (`clean`, `skip_empty`, `skip_hidden`) can leave it above `items.len()`. + /// The next page starts at `offset + min(count, limit)`; the cap matters only with `unwind`, + /// which can return more items than the rows scanned. #[serde(default)] pub count: i64, /// Whether the items are in descending order. @@ -222,19 +227,6 @@ 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. @@ -351,24 +343,53 @@ fn to_base62(mut value: u128) -> String { /// Milliseconds in a second. const MILLIS_PER_SECOND: i64 = 1000; +/// The longest validity of a signed public URL, in seconds: `4611686018427387`, about 146 million +/// years. It is half of what fits in milliseconds in an `i64`, which keeps the current time plus +/// the validity from overflowing. The `expires_in_secs` of `create_items_public_url`, +/// `create_items_public_url_with_format` and `create_keys_public_url` can be at most this. +pub const MAX_EXPIRES_IN_SECS: i64 = i64::MAX / MILLIS_PER_SECOND / 2; + +/// Checks the validity of a signed public URL before any request is sent. `None` is a URL that +/// does not expire. A validity has to be positive, since `0` and negative values would sign a URL +/// that has already expired, and at most [`MAX_EXPIRES_IN_SECS`], so that its expiry can't +/// overflow. Fails with [`ApifyClientError::InvalidArgument`] otherwise. +pub(crate) fn validate_expires_in_secs(expires_in_secs: Option) -> ApifyClientResult<()> { + match expires_in_secs { + Some(secs) if secs <= 0 => Err(ApifyClientError::InvalidArgument(format!( + "`expires_in_secs` must be positive, got {secs}; leave it out for a URL that does not expire" + ))), + Some(secs) if secs > MAX_EXPIRES_IN_SECS => Err(ApifyClientError::InvalidArgument(format!( + "`expires_in_secs` must be at most {MAX_EXPIRES_IN_SECS}, got {secs}" + ))), + _ => Ok(()), + } +} + +/// The expiry that a signed public URL gets, as the absolute time in milliseconds: the current +/// time plus the validity, or `0` for a URL that does not expire. Call it when the URL is signed, +/// so that the time the request before it took doesn't come off the validity, with a validity that +/// passed [`validate_expires_in_secs`]. +pub(crate) fn public_url_expiry_millis(expires_in_secs: Option) -> i64 { + match expires_in_secs { + Some(secs) => chrono::Utc::now().timestamp_millis() + secs * MILLIS_PER_SECOND, + None => 0, + } +} + /// Builds a storage-content signature for a resource's public URL, byte-for-byte compatible /// with the platform's `@apify/utilities` `createStorageContentSignature`. /// /// It signs the message `"{version}.{expiresAtMillis}.{resourceId}"` (where `expiresAtMillis` -/// is the absolute expiry in milliseconds, or `0` for a non-expiring URL) with +/// is the absolute expiry in milliseconds, as [`public_url_expiry_millis`] returns it) with /// [`create_hmac_signature`], then returns the base64url (no padding) encoding of /// `"{version}.{expiresAtMillis}.{hmac}"`. Used for dataset-items and key-list public URLs. pub(crate) fn sign_storage_content( secret_key: &str, resource_id: &str, - expires_in_secs: Option, + expires_at_millis: i64, ) -> String { use base64::Engine; - let expires_at_millis = match expires_in_secs { - Some(secs) => chrono::Utc::now().timestamp_millis() + secs * MILLIS_PER_SECOND, - None => 0, - }; let version = STORAGE_CONTENT_SIGNATURE_VERSION; let message = format!("{version}.{expires_at_millis}.{resource_id}"); let hmac = create_hmac_signature(secret_key, &message); @@ -531,7 +552,7 @@ mod signature_tests { #[test] fn storage_content_signature_non_expiring_envelope() { use base64::Engine; - let sig = sign_storage_content("secret", "RESID", None); + let sig = sign_storage_content("secret", "RESID", 0); let decoded = base64::engine::general_purpose::URL_SAFE_NO_PAD .decode(sig) .expect("valid base64url"); @@ -539,4 +560,31 @@ mod signature_tests { let expected_hmac = create_hmac_signature("secret", "0.0.RESID"); assert_eq!(decoded, format!("0.0.{expected_hmac}")); } + + // A validity is positive and at most the maximum: `None` never expires and a positive one is + // in the future, while `0`, a negative one and one beyond the maximum are refused. + #[test] + fn public_url_validity_is_validated_and_turned_into_an_expiry() { + use super::{public_url_expiry_millis, validate_expires_in_secs, MAX_EXPIRES_IN_SECS}; + // The documented value. + assert_eq!(MAX_EXPIRES_IN_SECS, 4_611_686_018_427_387); + let now = chrono::Utc::now().timestamp_millis(); + + assert_eq!(public_url_expiry_millis(None), 0); + let expiry = public_url_expiry_millis(Some(60)); + assert!((now + 60_000..=now + 70_000).contains(&expiry), "{expiry}"); + // The longest validity still has an expiry that fits. + assert!(public_url_expiry_millis(Some(MAX_EXPIRES_IN_SECS)) > now); + + for secs in [None, Some(1), Some(MAX_EXPIRES_IN_SECS)] { + validate_expires_in_secs(secs).expect("accepted"); + } + for secs in [0, -1, i64::MIN, MAX_EXPIRES_IN_SECS + 1, i64::MAX] { + let error = validate_expires_in_secs(Some(secs)).expect_err("refused"); + assert!( + matches!(error, crate::ApifyClientError::InvalidArgument(_)), + "{secs}: {error:?}" + ); + } + } } diff --git a/src/http_client.rs b/src/http_client.rs index 9f50888..f4067f0 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -27,6 +27,11 @@ use crate::timeouts::{Timeout, TimeoutTiers, NO_TIMEOUT}; /// Responses with status `< 300` are treated as success. const MAX_SUCCESS_STATUS_CODE: u16 = 300; +/// The header that carries the API token to the Apify API. +const AUTHORIZATION_HEADER: &str = "Authorization"; +/// The header that carries the API token to an Actor's server. The platform removes it before the +/// request reaches the Actor, which keeps `Authorization` free for the Actor's own authentication. +const X_APIFY_AUTHORIZATION_HEADER: &str = "X-Apify-Authorization"; /// Multiplier applied to the inter-retry delay after each attempt (exponential backoff). /// Matches the reference client's `async-retry` default factor of 2. const BACKOFF_FACTOR: u32 = 2; @@ -44,6 +49,12 @@ pub enum HttpMethod { Delete, /// HTTP `HEAD`. Head, + /// HTTP `PATCH`. + Patch, + /// HTTP `OPTIONS`. + Options, + /// HTTP `TRACE`. + Trace, } impl HttpMethod { @@ -55,6 +66,9 @@ impl HttpMethod { HttpMethod::Put => "PUT", HttpMethod::Delete => "DELETE", HttpMethod::Head => "HEAD", + HttpMethod::Patch => "PATCH", + HttpMethod::Options => "OPTIONS", + HttpMethod::Trace => "TRACE", } } } @@ -77,6 +91,10 @@ pub struct HttpRequest { /// is [`NO_TIMEOUT`]: such a request may run for as long as it /// takes and the backend must not impose a timeout of its own. pub timeout: Duration, + /// Whether the backend follows redirects. When `false`, the backend must return a redirect + /// response as it is: the request carries a token that must not follow a redirect to a host + /// the server picked. + pub follow_redirects: bool, } /// An HTTP response returned by an [`HttpBackend`]. @@ -99,6 +117,21 @@ impl HttpResponse { .find(|(k, _)| k.to_ascii_lowercase() == lower) .map(|(_, v)| v.as_str()) } + + /// Returns `true` for a success status, below `300`. + pub fn is_success(&self) -> bool { + self.status < MAX_SUCCESS_STATUS_CODE + } + + /// Returns the body as text, replacing any invalid UTF-8 sequences. + pub fn text(&self) -> String { + String::from_utf8_lossy(&self.body).into_owned() + } + + /// Parses the body as JSON. + pub fn json(&self) -> ApifyClientResult { + Ok(serde_json::from_slice(&self.body)?) + } } /// A streaming counterpart of [`HttpResponse`]: the status and headers, available as soon as @@ -157,29 +190,68 @@ fn to_reqwest_method(method: HttpMethod) -> reqwest::Method { HttpMethod::Put => reqwest::Method::PUT, HttpMethod::Delete => reqwest::Method::DELETE, HttpMethod::Head => reqwest::Method::HEAD, + HttpMethod::Patch => reqwest::Method::PATCH, + HttpMethod::Options => reqwest::Method::OPTIONS, + HttpMethod::Trace => reqwest::Method::TRACE, } } /// The default [`HttpBackend`] implementation, backed by [`reqwest`]. +/// +/// A `reqwest::Client` fixes its redirect policy when it is built, so the backend keeps a second +/// client that never follows redirects, for the requests with +/// [`follow_redirects`](HttpRequest::follow_redirects) set to `false`. #[derive(Debug, Clone)] pub struct ReqwestBackend { client: reqwest::Client, + /// Used for requests that must not follow redirects. + no_redirect_client: reqwest::Client, } impl ReqwestBackend { /// Creates a new backend with a default `reqwest::Client`. pub fn new() -> Self { - Self { - client: reqwest::Client::new(), - } + Self::with_client(reqwest::Client::new()) } /// Creates a backend wrapping a caller-provided `reqwest::Client`. /// - /// Useful for sharing a connection pool or customizing proxy/TLS settings. + /// Useful for sharing a connection pool or customizing proxy/TLS settings. The requests that + /// must not follow redirects, such as those to an Actor's server, are sent by a default + /// client that follows none: replace it with + /// [`with_no_redirect_client`](Self::with_no_redirect_client) to configure it too. pub fn with_client(client: reqwest::Client) -> Self { - Self { client } + Self { + client, + no_redirect_client: default_no_redirect_client(), + } } + + /// Replaces the client that sends the requests that must not follow redirects. It has to be + /// built with `redirect(reqwest::redirect::Policy::none())`, or those requests would follow + /// redirects. + #[must_use] + pub fn with_no_redirect_client(mut self, client: reqwest::Client) -> Self { + self.no_redirect_client = client; + self + } + + /// The `reqwest` client that sends `request`, following its redirect setting. + fn client_for(&self, request: &HttpRequest) -> &reqwest::Client { + if request.follow_redirects { + &self.client + } else { + &self.no_redirect_client + } + } +} + +/// A default `reqwest::Client` that follows no redirects. +fn default_no_redirect_client() -> reqwest::Client { + reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .expect("a default reqwest client without redirects builds, like `reqwest::Client::new`") } impl Default for ReqwestBackend { @@ -193,7 +265,7 @@ impl HttpBackend for ReqwestBackend { async fn send(&self, request: HttpRequest) -> ApifyClientResult { let method = to_reqwest_method(request.method); - let mut builder = self.client.request(method, &request.url); + let mut builder = self.client_for(&request).request(method, &request.url); if request.timeout != NO_TIMEOUT { builder = builder.timeout(request.timeout); } @@ -232,9 +304,9 @@ impl HttpBackend for ReqwestBackend { // Deliberately no `.timeout(request.timeout)` here: reqwest's per-request timeout bounds // the *entire* request including reading the whole body (it wraps the body stream in a // `TotalTimeoutBody`), which would cut off a long-but-active stream (e.g. live log - // redirection). `HttpClient::call_streaming` applies its own response-start and - // per-chunk-inactivity bounds instead, uniformly across backends. - let mut builder = self.client.request(method, &request.url); + // redirection). `HttpClient` applies its own response-start bound and, unless the call + // opts out, a per-chunk-inactivity bound instead, uniformly across backends. + let mut builder = self.client_for(&request).request(method, &request.url); for (key, value) in &request.headers { builder = builder.header(key, value); } @@ -286,14 +358,56 @@ pub(crate) struct HttpClientConfig { pub(crate) stats: Arc, } +/// The header that carries the client's API token. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) enum AuthorizationHeader { + /// `Authorization`, which the Apify API reads. + #[default] + Authorization, + /// `X-Apify-Authorization`, which Actor servers accept. + XApifyAuthorization, +} + +/// How the body of a request is compressed. +#[derive(Debug, Clone, Default)] +pub(crate) enum BodyCompression { + /// With the client's compressor. + #[default] + ClientDefault, + /// Not at all. + Disabled, + /// With this compressor, for this request only. + With(Arc), +} + /// Options of a single [`HttpClient::call_with_options`]. -#[derive(Debug, Clone, Copy, Default)] +#[derive(Debug, Clone, Default)] pub(crate) struct CallOptions { /// Give up on the first transport timeout instead of retrying it. For a call that is not /// idempotent, such as pushing dataset items, whose request may have reached the API before /// the client gave up on it: a retry could then store the same data twice. Other errors are /// retried as usual. pub(crate) do_not_retry_timeouts: bool, + /// How to compress the request body. + pub(crate) compression: BodyCompression, + /// The header that carries the API token. + pub(crate) authorization_header: AuthorizationHeader, + /// Return an error response as it is once no retry follows, instead of failing with an + /// [`ApiError`]. A rate limit (`429`) then stays out of the [`Statistics`], which track the + /// API's rate limits: only calls to the API fail on an error status. + pub(crate) return_error_responses: bool, + /// The statuses to retry, instead of the rate limit and the server errors. + pub(crate) retry_status_codes: Option<&'static [u16]>, + /// Give up on the first transport error, timeouts included, since the request may have + /// reached the server before it failed. + pub(crate) do_not_retry_transport_errors: bool, + /// Return a redirect response as it is, since the request carries a token that must not + /// follow a redirect to a host the server picked. + pub(crate) do_not_follow_redirects: bool, + /// For a streaming call: bound only the wait for the response to start, not the silence + /// between two chunks of its body. For a server that can stay silent while it keeps the + /// connection open. + pub(crate) no_stream_inactivity_timeout: bool, } /// The orchestrating HTTP client shared by every resource client. @@ -368,22 +482,50 @@ impl HttpClient { /// Adds the headers every request carries: the built-in `User-Agent` and `Authorization`, /// overridden by the client's default headers, overridden by the request's own headers. - /// Header names are compared case-insensitively. - fn apply_headers(&self, request: &mut HttpRequest) { + /// Header names are compared case-insensitively. With + /// [`AuthorizationHeader::XApifyAuthorization`], the `Authorization` header of the client, + /// default headers included, is sent as `X-Apify-Authorization`; a request header of either + /// name still wins. + fn apply_headers(&self, request: &mut HttpRequest, authorization_header: AuthorizationHeader) { let mut merged: HashMap = HashMap::new(); merged.insert("User-Agent".to_string(), self.user_agent.clone()); if let Some(token) = &self.token { - merged.insert("Authorization".to_string(), format!("Bearer {token}")); + merged.insert(AUTHORIZATION_HEADER.to_string(), format!("Bearer {token}")); } for (name, value) in &self.default_headers { set_header(&mut merged, name, value); } + if authorization_header == AuthorizationHeader::XApifyAuthorization { + move_header( + &mut merged, + AUTHORIZATION_HEADER, + X_APIFY_AUTHORIZATION_HEADER, + ); + } for (name, value) in std::mem::take(&mut request.headers) { set_header(&mut merged, &name, &value); } request.headers = merged; } + /// Compresses the body of `request` once (not per attempt) with the compressor `compression` + /// picks, when the body is large enough. + fn compress_body( + &self, + request: &mut HttpRequest, + compression: &BodyCompression, + ) -> ApifyClientResult<()> { + let compressor = match compression { + BodyCompression::ClientDefault => self.compressor.as_ref(), + BodyCompression::Disabled => return Ok(()), + BodyCompression::With(compressor) => compressor.as_ref(), + }; + if let Some(body) = request.body.take() { + request.body = Some(maybe_compress(compressor, &mut request.headers, body)?); + } + Ok(()) + } + /// The statistics of the calls made through this client. pub(crate) fn stats(&self) -> &Statistics { &self.stats @@ -391,25 +533,70 @@ impl HttpClient { /// Sends `request` with authentication, the user-agent header, and the retry policy /// applied, as `options` adjust it. Returns the first successful response, or the final - /// error. + /// error. With [`CallOptions::return_error_responses`], the final error response is returned + /// instead of the error. pub(crate) async fn call_with_options( &self, - mut request: HttpRequest, + request: HttpRequest, options: CallOptions, ) -> ApifyClientResult { + self.call_with_retries(request, &options).await + } + + /// Opens a streaming connection through the configured [`HttpBackend`]'s + /// [`HttpBackend::send_streaming`], with authentication, the user-agent header and the retry + /// policy applied like [`call_with_options`](Self::call_with_options), but with no request + /// compression. Like the reference clients, it retries until the response starts (an error + /// status or a transport error before it opens), and never once the body is being read: an + /// open stream, such as a log redirection or a large export already in flight, can't be + /// safely replayed from the start. + /// + /// On a non-success status, reads the body (bounded by the same inactivity timeout as a + /// successful body below) and returns the same structured [`ApiError`] [`call_with_options`](Self::call_with_options) + /// would. On success, returns the body as a stream whose `.next()` must resolve within the + /// request's resolved timeout each time — bounding silence (a stalled connection) without + /// capping the stream's total duration, so a long-but-active stream (e.g. live log + /// redirection) is never cut off. The same timeout also bounds the initial wait for the + /// response to start. + pub(crate) async fn call_streaming( + &self, + request: HttpRequest, + ) -> ApifyClientResult>>> { + let options = CallOptions { + compression: BodyCompression::Disabled, + ..CallOptions::default() + }; + Ok(self.call_streaming_response(request, &options).await?.body) + } + + /// Opens a streaming connection like [`call_streaming`](Self::call_streaming), but honors + /// `options` like [`call_with_options`](Self::call_with_options) does: with + /// [`CallOptions::return_error_responses`] the final error response is returned, with its + /// body unread, instead of the error. The body of the response is wrapped in the per-chunk + /// inactivity timeout that the request's timeout sets, unless + /// [`CallOptions::no_stream_inactivity_timeout`] says otherwise. + pub(crate) async fn call_streaming_response( + &self, + request: HttpRequest, + options: &CallOptions, + ) -> ApifyClientResult { + self.call_with_retries(request, options).await + } + + /// The retry loop shared by the buffered and the streaming calls: sends `request` until an + /// attempt succeeds, no retry follows its failure, or the attempts run out. + async fn call_with_retries( + &self, + mut request: HttpRequest, + options: &CallOptions, + ) -> ApifyClientResult { self.fail_if_poisoned()?; self.stats.add_call(); - self.apply_headers(&mut request); + self.apply_headers(&mut request, options.authorization_header); // Compress the request body once (not per attempt) when it is large enough, mirroring the // reference client. The API accepts both brotli- and gzip-encoded request bodies. - if let Some(body) = request.body.take() { - request.body = Some(maybe_compress( - self.compressor.as_ref(), - &mut request.headers, - body, - )?); - } + self.compress_body(&mut request, &options.compression)?; let method_str = request.method.as_str().to_string(); let path = extract_path(&request.url); @@ -424,33 +611,47 @@ impl HttpClient { let mut attempt = 1; loop { // Grow the timeout of the attempt with each attempt, up to `timeout_max`. + let attempt_timeout = self.attempt_timeout(base_timeout, attempt); let mut attempt_request = request.clone(); - attempt_request.timeout = self.attempt_timeout(base_timeout, attempt); + attempt_request.timeout = attempt_timeout; self.stats.add_request(); - let outcome = match self.backend.send(attempt_request).await { + // The error that ends the call, or `None` when a retry follows. A retry that follows + // drops the failed response unread: reading the body of an error that is thrown away + // would only delay the retry, or never end for a body that keeps streaming. + let final_error = match R::send(self.backend.as_ref(), attempt_request, options).await { Ok(response) => { - if response.status < MAX_SUCCESS_STATUS_CODE { + let status = response.status(); + if status < MAX_SUCCESS_STATUS_CODE { return Ok(response); } - if response.status == RATE_LIMIT_EXCEEDED_STATUS_CODE { + if status == RATE_LIMIT_EXCEEDED_STATUS_CODE && !options.return_error_responses + { self.stats.add_rate_limit_error(attempt); } - let api_error = build_api_error(&response, attempt, &method_str, &path); - let retryable = is_status_retryable(response.status); - (ApifyClientError::from(api_error), retryable) + // Give up immediately on a status a retry cannot fix, or after the last attempt. + let is_final = !is_status_retryable(status, options.retry_status_codes) + || attempt == max_attempts; + if !is_final { + None + } else if options.return_error_responses { + return Ok(response); + } else { + let buffered = response.into_buffered(attempt_timeout).await; + let api_error = build_api_error(&buffered, attempt, &method_str, &path); + Some(ApifyClientError::from(api_error)) + } } Err(err) => { let retryable = is_error_retryable(&err) + && !options.do_not_retry_transport_errors && !(options.do_not_retry_timeouts && matches!(err, ApifyClientError::Timeout)); - (err, retryable) + // Give up immediately on a non-retryable error or after the last attempt. + (!retryable || attempt == max_attempts).then_some(err) } }; - - let (error, retryable) = outcome; - // Give up immediately on non-retryable errors or after the last attempt. - if !retryable || attempt == max_attempts { + if let Some(error) = final_error { return Err(error); } @@ -481,68 +682,93 @@ impl HttpClient { pub(crate) fn user_agent(&self) -> &str { &self.user_agent } +} - /// Sends `request` through the configured [`HttpBackend`]'s [`HttpBackend::send_streaming`], - /// with authentication and the user-agent header applied like [`call_with_options`](Self::call_with_options), but with - /// no retries and no request compression: matches the reference clients, which do not retry - /// an open stream (a log redirection or a large export already in flight can't be safely - /// replayed from the start). - /// - /// On a non-success status, reads the body (bounded by the same inactivity timeout as a - /// successful body below) and returns the same structured [`ApiError`] [`call_with_options`](Self::call_with_options) - /// would. On success, returns the body as a stream whose `.next()` must resolve within the - /// request's resolved timeout each time — bounding silence (a stalled connection) without - /// capping the stream's total duration, so a long-but-active stream (e.g. live log - /// redirection) is never cut off. The same timeout also bounds the initial wait for the - /// response to start. - pub(crate) async fn call_streaming( - &self, - mut request: HttpRequest, - ) -> ApifyClientResult>>> { - self.fail_if_poisoned()?; - self.stats.add_call(); - self.stats.add_request(); - self.apply_headers(&mut request); +/// A response that one attempt of a retried request yields: a buffered one, or a streaming one +/// whose body is still unread. +#[async_trait] +trait AttemptResponse: Sized + Send { + /// Sends one attempt of `request` through `backend`. + async fn send( + backend: &dyn HttpBackend, + request: HttpRequest, + options: &CallOptions, + ) -> ApifyClientResult; - let method_str = request.method.as_str().to_string(); - let path = extract_path(&request.url); - // The request timeout is both the response-start wait and the per-chunk inactivity - // window, so a stream that stays silent for that long (e.g. a run that logs nothing for - // the `Long` tier) times out. `Timeout::NoTimeout`, set with a resource client's - // `with_timeout`, turns that off. - let budget = request.timeout; + /// The HTTP status of the response. + fn status(&self) -> u16; - let response = tokio::time::timeout(budget, self.backend.send_streaming(request)) - .await - .map_err(|_| ApifyClientError::Timeout)??; + /// Reads the whole response within `timeout`, for the error of an error status. + async fn into_buffered(self, timeout: Duration) -> HttpResponse; +} - if response.status < MAX_SUCCESS_STATUS_CODE { - return Ok(with_inactivity_timeout(response.body, budget)); - } +#[async_trait] +impl AttemptResponse for HttpResponse { + async fn send( + backend: &dyn HttpBackend, + request: HttpRequest, + _options: &CallOptions, + ) -> ApifyClientResult { + backend.send(request).await + } - // The error body arrives as a stream here too; read it (under the same inactivity - // budget) so the failure carries the same structured `ApiError` (type/message/data) as - // every buffered endpoint, rather than losing the body to a generic transport error. A - // body that fails to read (e.g. the connection drops mid-response) still surfaces the - // status code, just without a parsed message. - let body = collect_with_timeout(response.body, budget).await; - if response.status == RATE_LIMIT_EXCEEDED_STATUS_CODE { - self.stats.add_rate_limit_error(1); - } - let http_response = HttpResponse { + fn status(&self) -> u16 { + self.status + } + + async fn into_buffered(self, _timeout: Duration) -> HttpResponse { + self + } +} + +#[async_trait] +impl AttemptResponse for StreamingHttpResponse { + async fn send( + backend: &dyn HttpBackend, + request: HttpRequest, + options: &CallOptions, + ) -> ApifyClientResult { + // The request timeout bounds the wait for the response to start and, unless the options + // say otherwise, the silence between two chunks of the body: a stream that stays silent + // for that long (e.g. a run that logs nothing for the `Long` tier) times out. + // `Timeout::NoTimeout`, set with a resource client's `with_timeout`, turns that off. + let timeout = request.timeout; + let response = tokio::time::timeout(timeout, backend.send_streaming(request)) + .await + .map_err(|_| ApifyClientError::Timeout)??; + let body = if options.no_stream_inactivity_timeout { + response.body + } else { + with_inactivity_timeout(response.body, timeout) + }; + Ok(StreamingHttpResponse { status: response.status, headers: response.headers, body, - }; - let api_error = build_api_error(&http_response, 1, &method_str, &path); - Err(ApifyClientError::from(api_error)) + }) + } + + fn status(&self) -> u16 { + self.status + } + + /// The error body arrives as a stream here too; reading it lets the failure carry the same + /// structured `ApiError` (type/message/data) as every buffered endpoint, rather than losing + /// the body to a generic transport error. A body that fails to read (e.g. the connection + /// drops mid-response) still surfaces the status code, just without a parsed message. + async fn into_buffered(self, timeout: Duration) -> HttpResponse { + HttpResponse { + status: self.status, + headers: self.headers, + body: collect_body(with_inactivity_timeout(self.body, timeout)).await, + } } } /// Wraps a streaming body so each chunk must arrive within `timeout` of the previous one (or of /// the stream opening), erroring with [`ApifyClientError::Timeout`] otherwise — an inactivity -/// timeout, not a cap on the stream's total duration. Shared by -/// [`HttpClient::call_streaming`]'s success and error-body-reading paths. +/// timeout, not a cap on the stream's total duration. Used by the streaming attempts of +/// [`HttpClient`], for the body of a response and for the body of an error response they read. /// /// A timeout yields exactly one `Err(Timeout)` and then ends the stream (further polls return /// `None`), rather than retrying the same read forever: the underlying body is dropped once it @@ -574,16 +800,16 @@ fn with_inactivity_timeout( )) } -/// Drains a streaming body into a single buffer, under the same per-chunk inactivity timeout as -/// [`with_inactivity_timeout`]. Used only for a non-success response's (typically small) error -/// body, so buffering it is fine — unlike a successful body, which callers stream incrementally. -async fn collect_with_timeout( - body: std::pin::Pin>> + Send>>, - timeout: Duration, +/// Drains a streaming body into a single buffer, stopping at the first error. Used only for a +/// non-success response's (typically small) error body, so buffering it is fine — unlike a +/// successful body, which callers stream incrementally. +async fn collect_body( + mut body: std::pin::Pin< + Box>> + Send>, + >, ) -> Vec { let mut buf = Vec::new(); - let mut timed = with_inactivity_timeout(body, timeout); - while let Some(chunk) = futures_util::StreamExt::next(&mut timed).await { + while let Some(chunk) = futures_util::StreamExt::next(&mut body).await { match chunk { Ok(bytes) => buf.extend(bytes), Err(_) => break, @@ -599,6 +825,21 @@ fn set_header(headers: &mut HashMap, name: &str, value: &str) { headers.insert(name.to_string(), value.to_string()); } +/// Renames header `from` to `to`, whatever the casing of `from`, replacing any `to` header. Does +/// nothing when there is no `from` header. +fn move_header(headers: &mut HashMap, from: &str, to: &str) { + let Some(name) = headers + .keys() + .find(|name| name.eq_ignore_ascii_case(from)) + .cloned() + else { + return; + }; + if let Some(value) = headers.remove(&name) { + set_header(headers, to, &value); + } +} + /// Returns the path + query portion of a URL, for error reporting. fn extract_path(url: &str) -> Option { // Find the start of the path after the scheme+host. @@ -609,9 +850,13 @@ fn extract_path(url: &str) -> Option { } /// We retry `429` (rate limit) and `5xx` (internal server errors), matching the -/// reference client policy. Other `4xx` statuses are caller errors and are not retried. -fn is_status_retryable(status: u16) -> bool { - status == RATE_LIMIT_EXCEEDED_STATUS_CODE || status >= MIN_SERVER_ERROR_STATUS_CODE +/// reference client policy, unless the request lists its own `retry_status_codes`. Other `4xx` +/// statuses are caller errors and are not retried. +fn is_status_retryable(status: u16, retry_status_codes: Option<&[u16]>) -> bool { + match retry_status_codes { + Some(codes) => codes.contains(&status), + None => status == RATE_LIMIT_EXCEEDED_STATUS_CODE || status >= MIN_SERVER_ERROR_STATUS_CODE, + } } /// Only transport-level failures are retryable. Programming errors (serde, invalid diff --git a/src/lib.rs b/src/lib.rs index 51f2ef3..38683ce 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -86,6 +86,9 @@ pub use clients::actor::{ ActorBuildOptions, ActorCallOptions, ActorStartOptions, TaskStartOptions, WaitForResources, }; pub use clients::actor_collection::ActorListOptions; +pub use clients::actor_server::{ + ActorServerBody, ActorServerCompression, ActorServerRequestOptions, +}; pub use clients::dataset::{DatasetDownloadOptions, DatasetListItemsOptions, DownloadItemsFormat}; pub use clients::key_value_store::{GetRecordOptions, KeyValueStoreKeysIterator, ListKeysOptions}; pub use clients::log::LogOptions; diff --git a/src/models.rs b/src/models.rs index bacc8e8..61d1b24 100644 --- a/src/models.rs +++ b/src/models.rs @@ -84,11 +84,84 @@ pub struct Actor { /// When the Actor was last modified. #[serde(default)] pub modified_at: Option>, + /// URL for sending requests to the Actor's server (Standby mode), or `None` if the Standby + /// mode isn't enabled. Use [`ActorClient::server`](crate::clients::actor::ActorClient::server) + /// to send the requests. + #[serde(default)] + pub standby_url: Option, + /// The Standby mode configuration of the Actor, or `None` if it has none. + #[serde(default)] + pub actor_standby: Option, + /// Maps build tag names (e.g. `latest`) to their build information. A tag the API reports + /// without a build maps to `None`. + #[serde(default)] + pub tagged_builds: Option>>, /// Any other fields returned by the API. #[serde(flatten)] pub extra: Extra, } +/// The Standby mode configuration of an Actor. +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ActorStandby { + /// Whether standby mode is enabled for the Actor. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_enabled: Option, + /// Target number of concurrent HTTP requests a single run is configured to handle. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub desired_requests_per_actor_run: Option, + /// Maximum number of concurrent HTTP requests that can be routed to a single run. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_requests_per_actor_run: Option, + /// In seconds, how long a run can stay idle without incoming requests before it's terminated. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub idle_timeout_secs: Option, + /// Which build to run in standby mode. Either a build tag or a version number. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub build: Option, + /// In MB, the amount of memory allocated to the run. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory_mbytes: Option, + /// If `true`, prevents the standby mode configuration from being overridden elsewhere. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub disable_standby_fields_override: Option, + /// Whether to pass the Actor's input to the standby run. If `false`, the standby runs start + /// with no input. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub should_pass_actor_input: Option, + /// Whether each user gets their own standby runs (`SINGLE_TENANT`) or runs are shared across + /// users (`MULTI_TENANT`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tenancy: Option, + /// Whether standby requests can be authenticated with an Apify Console session. When creating + /// an Actor, omit this field or set it to `false`. When updating an Actor, only Apify + /// administrators can change it; other users can only send the current value, or `false` if + /// it's not set yet. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_console_auth_enabled: Option, + /// Whether standby requests can be made without an API token. When creating an Actor, omit + /// this field or set it to `false`. When updating an Actor, only Apify administrators can + /// change it; other users can only send the current value, or `false` if it's not set yet. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_tokenless_enabled: Option, +} + +/// Information about the build a tag of an Actor points to. +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TaggedBuildInfo { + /// ID of the build the tag points to. + #[serde(default)] + pub build_id: Option, + /// Number of the build the tag points to, e.g. `0.0.2`. + #[serde(default)] + pub build_number: Option, + /// When the build finished. + #[serde(default)] + pub finished_at: Option>, +} + /// A single execution of an Actor (an Actor run). #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] @@ -751,6 +824,18 @@ pub struct ActorVersion { /// The source type of the version, e.g. `SOURCE_FILES`, `GIT_REPO`, `TARBALL`, `GITHUB_GIST`. #[serde(default)] pub source_type: Option, + /// The Actor's directory inside the uploaded source, as a path relative to the upload root + /// with `/` separators, e.g. `actors/typescript-actor`. Applies to the `SOURCE_FILES` and + /// `TARBALL` source types. `None` when the upload root is the Actor's directory. On a + /// `TARBALL` version, only users with write access to the Actor get it. It has 1 to 255 + /// characters, with no leading, trailing, or repeated `/` and no `.` or `..` segments. + /// + /// When updating a version, an update that sends `sourceType`, `sourceFiles`, or `tarballUrl` + /// without `actorPath` removes the stored one, so send it again with every upload. An update + /// of other fields, such as `envVars` or `buildTag`, keeps it. `None` is not sent: to remove + /// the stored path, send `"actorPath": null` in a JSON update body. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub actor_path: Option, /// Any other fields returned by the API. #[serde(flatten)] pub extra: Extra, diff --git a/src/version.rs b/src/version.rs index 5a0adda..3c83068 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-29T100357Z"; +pub const API_SPEC_VERSION: &str = "v2-2026-11-04T151257Z"; diff --git a/tests/actor.rs b/tests/actor.rs index 8512a23..34ef94c 100644 --- a/tests/actor.rs +++ b/tests/actor.rs @@ -299,6 +299,7 @@ async fn actor_version_crud_flow() { .update(&ActorVersion { version_number: "0.1".to_string(), source_type: Some("SOURCE_FILES".to_string()), + actor_path: None, extra: serde_json::from_value(json!({ "buildTag": "latest", "sourceFiles": [ @@ -329,6 +330,92 @@ async fn actor_version_crud_flow() { ); } +/// Actor version `actorPath`: set on create, kept by updates of other fields, removed by an upload +/// without it or by `null`. +#[tokio::test(flavor = "multi_thread")] +async fn actor_version_actor_path_flow() { + let client = require_client!(); + let name = actor_name("ver-path"); + let actor = client + .actors() + .create(&actor_definition(&name)) + .await + .expect("create actor"); + + let cleanup_client = client.clone(); + let cleanup_id = actor.id.clone(); + let _guard = common::Cleanup::new(move || async move { + let _ = cleanup_client.actor(&cleanup_id).delete().await; + }); + + let actor_path = "actors/typescript-actor"; + let source_files = json!([ + { + "name": "actors/typescript-actor/Dockerfile", + "format": "TEXT", + "content": "FROM apify/actor-node:20\nCOPY . ./\nCMD node main.js" + }, + { + "name": "actors/typescript-actor/main.js", + "format": "TEXT", + "content": "console.log('monorepo');" + } + ]); + let actor_client = client.actor(&actor.id); + + // Create a version with an `actorPath`. + let created = actor_client + .versions() + .create(&json!({ + "versionNumber": "0.1", + "sourceType": "SOURCE_FILES", + "buildTag": "beta", + "sourceFiles": source_files, + "actorPath": actor_path, + })) + .await + .expect("create version with actorPath"); + assert_eq!(created.actor_path.as_deref(), Some(actor_path)); + + let version_client = actor_client.version("0.1"); + let fetched = version_client + .get() + .await + .expect("get version") + .expect("version 0.1 should exist"); + assert_eq!(fetched.actor_path.as_deref(), Some(actor_path)); + + // An update of other fields keeps the path. + let updated = version_client + .update(&json!({ "buildTag": "latest" })) + .await + .expect("update build tag"); + assert_eq!(updated.actor_path.as_deref(), Some(actor_path)); + + // An upload without the path removes it. + let updated = version_client + .update(&json!({ "sourceType": "SOURCE_FILES", "sourceFiles": source_files })) + .await + .expect("upload without actorPath"); + assert_eq!(updated.actor_path, None); + + // Sending it again with an upload sets it, and `null` removes it. + let updated = version_client + .update(&json!({ + "sourceType": "SOURCE_FILES", + "sourceFiles": source_files, + "actorPath": actor_path, + })) + .await + .expect("upload with actorPath"); + assert_eq!(updated.actor_path.as_deref(), Some(actor_path)); + let updated = version_client + .update(&json!({ "actorPath": null })) + .await + .expect("remove actorPath"); + assert_eq!(updated.actor_path, None); +} + /// Actor version environment variables: create -> get -> list -> update -> delete. #[tokio::test(flavor = "multi_thread")] async fn actor_env_var_crud_flow() { diff --git a/tests/actor_server.rs b/tests/actor_server.rs new file mode 100644 index 0000000..3effda3 --- /dev/null +++ b/tests/actor_server.rs @@ -0,0 +1,168 @@ +//! Integration test for the server interface of an Actor (`ActorClient::server`). +//! +//! It creates and builds a temporary Server Actor that echoes every request it receives, sends +//! requests to it, and deletes it. + +mod common; + +use apify_client::clients::actor_server::{ActorServerBody, ActorServerRequestOptions}; +use apify_client::http_client::HttpMethod; +use serde_json::{json, Value}; + +/// A web server that answers every request, including the platform's readiness probe, with the +/// request it received, under the status in the `status` query parameter, or 200. +const ECHO_SERVER_SOURCE: &str = r#" +const http = require('http'); + +http.createServer((req, res) => { + let body = ''; + req.on('data', (chunk) => { body += chunk; }); + req.on('end', () => { + const status = Number(new URL(req.url, 'http://localhost').searchParams.get('status')) || 200; + res.writeHead(status, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ method: req.method, url: req.url, headers: req.headers, body })); + }); +}).listen(process.env.ACTOR_WEB_SERVER_PORT); +"#; + +const DOCKERFILE: &str = "FROM node:22-alpine\nCOPY main.js ./\nCMD [\"node\", \"main.js\"]\n"; + +/// How long to wait for the build of the echo Actor, in seconds. +const BUILD_WAIT_SECS: i64 = 300; + +/// The web server schema of the echo Actor. +fn openapi_spec() -> Value { + json!({ + "openapi": "3.0.0", + "info": { "title": "Echo server", "version": "1.0.0" }, + "paths": { "/echo": { "post": { "responses": { "200": { "description": "The echoed request." } } } } }, + }) +} + +fn query(params: &[(&str, &str)]) -> Vec<(String, String)> { + params + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect() +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_server_client_returns_the_url_and_spec_and_sends_requests_to_the_server() { + let client = require_client!(); + let name = common::unique_actor_name("server"); + let actor_json = json!({ + "actorSpecification": 1, + "name": name, + "version": "0.0", + "usesStandbyMode": true, + "webServerSchema": openapi_spec(), + }); + let actor = client + .actors() + .create(&json!({ + "name": name, + "isPublic": false, + "versions": [{ + "versionNumber": "0.0", + "sourceType": "SOURCE_FILES", + "buildTag": "latest", + "sourceFiles": [ + { "name": "Dockerfile", "format": "TEXT", "content": DOCKERFILE }, + { "name": "main.js", "format": "TEXT", "content": ECHO_SERVER_SOURCE }, + { "name": ".actor/actor.json", "format": "TEXT", "content": actor_json.to_string() }, + ], + }], + "actorStandby": { "isEnabled": true, "memoryMbytes": 256, "idleTimeoutSecs": 60 }, + })) + .await + .expect("create the server actor"); + + let cleanup_client = client.clone(); + let cleanup_id = actor.id.clone(); + let _guard = common::Cleanup::new(move || async move { + let _ = cleanup_client.actor(&cleanup_id).delete().await; + }); + + let actor_client = client.actor(&actor.id); + let build = actor_client + .build("0.0", Default::default()) + .await + .expect("start the build"); + let finished = client + .build(&build.id) + .wait_for_finish(Some(BUILD_WAIT_SECS)) + .await + .expect("wait for the build"); + assert_eq!(finished.status.as_deref(), Some("SUCCEEDED")); + + let fetched = actor_client + .get() + .await + .expect("get the actor") + .expect("the actor exists"); + assert_eq!( + fetched.actor_standby.as_ref().and_then(|s| s.is_enabled), + Some(true) + ); + let server = actor_client.server(); + + assert_eq!( + Some(server.get_url().await.expect("get the url")), + fetched.standby_url + ); + assert_eq!( + server.get_open_api_spec().await.expect("get the spec"), + Some(openapi_spec()) + ); + + // A JSON POST with a query reaches the server as it was sent. + let response = server + .request( + "/echo", + ActorServerRequestOptions { + method: Some(HttpMethod::Post), + params: query(&[("q", "apify")]), + body: Some(ActorServerBody::Json(json!({ "limit": 5 }))), + ..Default::default() + }, + ) + .await + .expect("post to the server"); + assert_eq!(response.status, 200, "{}", response.text()); + let echoed: Value = response.json().expect("a JSON echo"); + assert_eq!(echoed["method"], "POST"); + assert_eq!(echoed["url"], "/echo?q=apify"); + let body: Value = serde_json::from_str(echoed["body"].as_str().unwrap()).unwrap(); + assert_eq!(body, json!({ "limit": 5 })); + // The platform consumes the client's token, so it never reaches the Actor's code. + assert!(echoed["headers"].get("x-apify-authorization").is_none()); + assert!(echoed["headers"].get("authorization").is_none()); + + // Headers of the request, `Authorization` included, reach the Actor as they are. + let mut own_authorization = ActorServerRequestOptions::default(); + own_authorization.headers.insert( + "Authorization".to_string(), + "Bearer actor-token".to_string(), + ); + let response = server + .request("/echo", own_authorization) + .await + .expect("get with an own authorization"); + let echoed: Value = response.json().expect("a JSON echo"); + assert_eq!(echoed["headers"]["authorization"], "Bearer actor-token"); + + // An error status is returned, not raised. + let response = server + .request( + "/echo", + ActorServerRequestOptions { + params: query(&[("status", "404")]), + ..Default::default() + }, + ) + .await + .expect("an error status is a response"); + assert_eq!(response.status, 404); + let echoed: Value = response.json().expect("a JSON echo"); + assert_eq!(echoed["url"], "/echo?status=404"); +} diff --git a/tests/common/mock_backend.rs b/tests/common/mock_backend.rs index a55049b..e128ba2 100644 --- a/tests/common/mock_backend.rs +++ b/tests/common/mock_backend.rs @@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use apify_client::http_client::{HttpBackend, HttpRequest, HttpResponse}; +use apify_client::http_client::{HttpBackend, HttpMethod, HttpRequest, HttpResponse}; use apify_client::{ApifyClient, ApifyClientError}; use async_trait::async_trait; @@ -21,11 +21,17 @@ pub struct MockBackend { last_body: Mutex>>, /// The `timeout` of every request in call order, so tests can assert the timeout tiers. timeouts: Mutex>, + /// The method of every request in call order. + methods: Mutex>, + /// The `follow_redirects` setting of every request in call order. + follow_redirects: Mutex>, } #[derive(Debug, Clone)] pub enum MockOutcome { Status(u16, Vec), + /// A response with headers. + StatusWithHeaders(u16, Vec<(&'static str, String)>, Vec), NetworkError, /// The transport gave up on the request because it outlived its timeout. Timeout, @@ -41,9 +47,21 @@ impl MockBackend { last_headers: Mutex::new(std::collections::HashMap::new()), last_body: Mutex::new(None), timeouts: Mutex::new(Vec::new()), + methods: Mutex::new(Vec::new()), + follow_redirects: Mutex::new(Vec::new()), }) } + /// The method of every request in call order. + pub fn methods(&self) -> Vec { + self.methods.lock().unwrap().clone() + } + + /// The `follow_redirects` setting of every request in call order. + pub fn follow_redirects(&self) -> Vec { + self.follow_redirects.lock().unwrap().clone() + } + /// The timeout of every request in call order. pub fn timeouts(&self) -> Vec { self.timeouts.lock().unwrap().clone() @@ -90,6 +108,11 @@ impl HttpBackend for MockBackend { *self.last_headers.lock().unwrap() = request.headers.clone(); *self.last_body.lock().unwrap() = request.body.clone(); self.timeouts.lock().unwrap().push(request.timeout); + self.methods.lock().unwrap().push(request.method); + self.follow_redirects + .lock() + .unwrap() + .push(request.follow_redirects); let mut queue = self.responses.lock().unwrap(); let outcome = if queue.len() > 1 { queue.remove(0) @@ -106,6 +129,14 @@ impl HttpBackend for MockBackend { headers: Default::default(), body, }), + MockOutcome::StatusWithHeaders(status, headers, body) => Ok(HttpResponse { + status, + headers: headers + .into_iter() + .map(|(name, value)| (name.to_string(), value)) + .collect(), + body, + }), MockOutcome::NetworkError => Err(ApifyClientError::Http("boom".to_string())), MockOutcome::Timeout => Err(ApifyClientError::Timeout), } diff --git a/tests/dataset.rs b/tests/dataset.rs index 1323eee..e834c4f 100644 --- a/tests/dataset.rs +++ b/tests/dataset.rs @@ -123,6 +123,70 @@ async fn iterate_dataset_items() { ); } +/// How many times the count test lists the items while it waits for the pushed rows to be counted. +const COUNT_WAIT_ATTEMPTS: usize = 30; + +/// How long the count test waits between two listings. +const COUNT_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + +/// `list_items` reports in `count` the rows the API scanned: a filter can leave it above the items +/// returned, and iterating by it reads every row behind a page the filter emptied. +#[tokio::test(flavor = "multi_thread")] +async fn list_items_count_follows_the_rows_scanned_when_a_filter_drops_items() { + let client = require_client!(); + let name = common::unique_name("dataset-count"); + let dataset = client + .datasets() + .get_or_create(Some(&name)) + .await + .expect("create dataset"); + + let cleanup_client = client.clone(); + let id = dataset.id.clone(); + let _guard = common::Cleanup::new(move || async move { + let _ = cleanup_client.dataset(&id).delete().await; + }); + + let dataset_client = client.dataset(&dataset.id); + // Four empty rows in front of the only item that survives `skip_empty`. + let rows = json!([{}, {}, {}, {}, { "n": 1 }]); + let row_count = rows.as_array().expect("an array of rows").len() as i64; + dataset_client.push_items(&rows).await.expect("push items"); + + let filtered = apify_client::DatasetListItemsOptions { + skip_empty: Some(true), + ..Default::default() + }; + // The item count of a fresh dataset lags the push, so wait for the rows to be counted. + let mut page = None; + for _ in 0..COUNT_WAIT_ATTEMPTS { + let current = dataset_client + .list_items::(filtered.clone()) + .await + .expect("list items"); + let counted = current.count >= row_count; + page = Some(current); + if counted { + break; + } + tokio::time::sleep(COUNT_WAIT_INTERVAL).await; + } + let page = page.expect("a page was listed"); + assert_eq!(page.items.len(), 1); + assert_eq!( + page.count, row_count, + "the rows scanned, not the one item returned" + ); + + let items = dataset_client + .iterate_items::(filtered) + .with_chunk_size(2) + .collect_all() + .await + .expect("iterate items"); + assert_eq!(items, vec![json!({ "n": 1 })]); +} + /// Complex flow: create -> get -> push items -> read items -> update -> delete. #[tokio::test(flavor = "multi_thread")] async fn dataset_crud_flow() { diff --git a/tests/unit_http.rs b/tests/unit_http.rs index a45f2a5..5f18e88 100644 --- a/tests/unit_http.rs +++ b/tests/unit_http.rs @@ -1592,3 +1592,178 @@ async fn batch_add_requests_rejects_an_oversized_request_before_sending_anything ); assert_eq!(backend.call_count(), 0, "nothing was sent"); } + +/// A dataset-items response with the pagination headers the API sets. +fn dataset_items_page(items: &str, scanned: Option<&str>) -> MockOutcome { + let mut headers = vec![ + ("x-apify-pagination-total", "10".to_string()), + ("x-apify-pagination-offset", "0".to_string()), + ("x-apify-pagination-limit", "3".to_string()), + ]; + if let Some(scanned) = scanned { + headers.push(("x-apify-pagination-count", scanned.to_string())); + } + MockOutcome::StatusWithHeaders(200, headers, items.as_bytes().to_vec()) +} + +/// `list_items` reports the larger of the rows the API scanned and the items returned in `count`. +#[tokio::test] +async fn list_items_count_is_the_larger_of_the_scanned_rows_and_the_items_returned() { + let cases = [ + ("filtered page", "[{\"a\":1}]", Some("3"), 3), + ("lagging header", "[{\"a\":1},{\"a\":2}]", Some("0"), 2), + ("unwound page", "[{\"a\":1},{\"a\":2}]", Some("1"), 2), + ("missing header", "[{\"a\":1}]", None, 1), + ("empty scanned page", "[]", Some("1"), 1), + ]; + for (name, items, scanned, expected) in cases { + let backend = MockBackend::new(vec![dataset_items_page(items, scanned)]); + let client = client_with(backend, 0); + + let page = client + .dataset("some-id") + .list_items::(Default::default()) + .await + .expect("ok"); + + assert_eq!(page.count, expected, "{name}"); + } +} + +/// A page the filter emptied still advances the iteration by the rows it scanned, so the rows +/// behind it are read. +#[tokio::test] +async fn iterate_items_continues_past_a_page_a_filter_emptied() { + let backend = MockBackend::new(vec![ + dataset_items_page("[]", Some("3")), + dataset_items_page("[{\"i\":3}]", Some("3")), + dataset_items_page("[]", Some("0")), + ]); + let client = client_with(backend.clone(), 0); + + let items = client + .dataset("some-id") + .iterate_items::(Default::default()) + .with_chunk_size(3) + .collect_all() + .await + .expect("ok"); + + assert_eq!(items, vec![serde_json::json!({ "i": 3 })]); + let urls = backend.urls(); + assert_eq!(urls.len(), 3, "{urls:?}"); + assert!(urls[1].contains("offset=3"), "{urls:?}"); + assert!(urls[2].contains("offset=6"), "{urls:?}"); +} + +/// The store drops Actors after counting them, so its `count` exceeds the items returned and the +/// iteration steps by it: the next page does not read the dropped entries again. +#[tokio::test] +async fn iterate_steps_by_the_count_the_api_reports_when_it_exceeds_the_items_returned() { + let backend = MockBackend::new(vec![ + MockOutcome::Status( + 200, + br#"{"data":{"total":9,"count":3,"items":[{"id":"a"}]}}"#.to_vec(), + ), + MockOutcome::Status( + 200, + br#"{"data":{"total":9,"count":3,"items":[{"id":"b"},{"id":"c"}]}}"#.to_vec(), + ), + MockOutcome::Status( + 200, + br#"{"data":{"total":9,"count":0,"items":[]}}"#.to_vec(), + ), + ]); + let client = client_with(backend.clone(), 0); + + let actors = client + .store() + .iterate(Default::default()) + .with_chunk_size(3) + .collect_all() + .await + .expect("ok"); + + assert_eq!(actors.len(), 3); + let urls = backend.urls(); + assert!(urls[1].contains("offset=3"), "{urls:?}"); + assert!(urls[2].contains("offset=6"), "{urls:?}"); +} + +/// A streaming call retries an error status until the response starts, like a buffered call. +#[tokio::test] +async fn a_stream_is_retried_on_an_error_status_until_the_response_starts() { + let backend = MockBackend::new(vec![ + MockOutcome::Status(500, br#"{"error":{"message":"oops"}}"#.to_vec()), + MockOutcome::Status(429, br#"{"error":{"message":"slow down"}}"#.to_vec()), + MockOutcome::Status(200, b"log line".to_vec()), + ]); + let client = client_with(backend.clone(), 3); + + let stream = client + .log("run") + .stream() + .await + .expect("ok") + .expect("the log exists"); + let chunks: Vec> = stream.map(|chunk| chunk.expect("a chunk")).collect().await; + + assert_eq!(chunks.concat(), b"log line"); + assert_eq!(backend.call_count(), 3); + assert_eq!(client.stats().rate_limit_errors().iter().sum::(), 1); +} + +/// A streaming call retries a transport error before the response opens. +#[tokio::test] +async fn a_stream_is_retried_on_a_transport_error_before_the_response_opens() { + let backend = MockBackend::new(vec![ + MockOutcome::NetworkError, + MockOutcome::Status(200, b"items".to_vec()), + ]); + let client = client_with(backend.clone(), 3); + + let stream = client + .dataset("d") + .download_items_stream(apify_client::DownloadItemsFormat::Json, Default::default()) + .await + .expect("the retry opens the response"); + let chunks: Vec> = stream.map(|chunk| chunk.expect("a chunk")).collect().await; + + assert_eq!(chunks.concat(), b"items"); + assert_eq!(backend.call_count(), 2); +} + +/// A streaming call that keeps failing fails with the structured API error of the last attempt, +/// and one with a status that a retry cannot fix fails at once. +#[tokio::test] +async fn a_stream_fails_with_the_api_error_once_the_retries_run_out_or_cannot_help() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 503, + br#"{"error":{"type":"server-error","message":"down"}}"#.to_vec(), + )]); + let client = client_with(backend.clone(), 2); + let error = client + .log("run") + .stream() + .await + .err() + .expect("the retries run out"); + assert_eq!(backend.call_count(), 3); + let api = error.as_api_error().expect("an API error"); + assert_eq!((api.status_code, api.attempt), (503, 3)); + assert_eq!(api.message, "down"); + + let backend = MockBackend::new(vec![MockOutcome::Status( + 400, + br#"{"error":{"message":"bad"}}"#.to_vec(), + )]); + let client = client_with(backend.clone(), 2); + let error = client + .log("run") + .stream() + .await + .err() + .expect("a 400 is final"); + assert_eq!(backend.call_count(), 1); + assert_eq!(error.status_code(), Some(400)); +} diff --git a/tests/unit_resources.rs b/tests/unit_resources.rs index 5135531..c199dbe 100644 --- a/tests/unit_resources.rs +++ b/tests/unit_resources.rs @@ -592,3 +592,150 @@ async fn public_urls_of_storages_chained_from_a_filtered_last_run_carry_its_filt assert!(url.contains("status=SUCCEEDED"), "{url}"); } } + +// ----- signed public URL validity --------------------------------------------------------- + +/// A validity of `0` or below would make a signed URL permanent or already expired, so it is +/// refused before any request is sent. +#[tokio::test] +async fn a_non_positive_public_url_validity_is_rejected_before_any_request() { + let backend = MockBackend::new(vec![ok_data(json!({ "id": "x" }))]); + let client = client_with(backend.clone(), 0); + let dataset = client.dataset("d"); + let store = client.key_value_store("s"); + + for expires_in_secs in [0, -1, i64::MAX] { + assert!(is_invalid_argument( + dataset + .create_items_public_url(Default::default(), Some(expires_in_secs)) + .await + )); + assert!(is_invalid_argument( + dataset + .create_items_public_url_with_format( + Default::default(), + Some(expires_in_secs), + None + ) + .await + )); + assert!(is_invalid_argument( + store.create_keys_public_url(Some(expires_in_secs)).await + )); + } + assert_eq!(backend.call_count(), 0); +} + +/// A positive validity signs the URL of a private storage with an expiry. +#[tokio::test] +async fn a_positive_public_url_validity_signs_the_url() { + let backend = MockBackend::new(vec![ok_data( + json!({ "id": "x", "urlSigningSecretKey": "secret" }), + )]); + let client = client_with(backend, 0); + + let url = client + .key_value_store("s") + .create_keys_public_url(Some(60)) + .await + .unwrap(); + + assert!(url.contains("signature="), "{url}"); +} + +/// How long the slow storage lookup of the signing tests takes. +const SLOW_LOOKUP: std::time::Duration = std::time::Duration::from_millis(300); + +/// A backend whose responses take [`SLOW_LOOKUP`]. +#[derive(Debug)] +struct SlowBackend; + +#[async_trait::async_trait] +impl apify_client::http_client::HttpBackend for SlowBackend { + async fn send( + &self, + _request: apify_client::http_client::HttpRequest, + ) -> Result { + tokio::time::sleep(SLOW_LOOKUP).await; + Ok(apify_client::http_client::HttpResponse { + status: 200, + headers: Default::default(), + body: json!({ "data": { "id": "x", "urlSigningSecretKey": "secret" } }) + .to_string() + .into_bytes(), + }) + } +} + +/// The validity the signing tests ask for, in seconds. +const VALIDITY_SECS: i64 = 1; + +/// How much of the validity the signing tests tolerate to be lost to the time between signing the +/// URL and reading the clock, in milliseconds. It stays well below [`SLOW_LOOKUP`]. +const SLACK_MILLIS: i64 = 200; + +/// Milliseconds in a second. +const MILLIS_PER_SECOND: i64 = 1000; + +/// A client whose requests take [`SLOW_LOOKUP`]. +fn slow_client() -> apify_client::ApifyClient { + apify_client::ApifyClient::builder() + .token("test-token") + .http_backend(std::sync::Arc::new(SlowBackend)) + .build() +} + +/// Checks that the signed `url`, which was just returned, still has about [`VALIDITY_SECS`] left: +/// the time the storage lookup before the signing took did not come off the validity. +fn assert_validity_counts_from_signing(url: &str) { + use base64::Engine; + + let returned_at_millis = chrono::Utc::now().timestamp_millis(); + let signature = url.split("signature=").nth(1).expect("a signature"); + let envelope = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(signature) + .unwrap(); + let envelope = String::from_utf8(envelope).unwrap(); + let expires_at_millis: i64 = envelope.split('.').nth(1).unwrap().parse().unwrap(); + let left_millis = expires_at_millis - returned_at_millis; + assert!( + left_millis >= VALIDITY_SECS * MILLIS_PER_SECOND - SLACK_MILLIS, + "the lookup came off the validity: {left_millis} ms left" + ); +} + +/// The validity of a signed URL counts from when the URL is signed, so the time the storage lookup +/// before it takes does not come off the validity. +#[tokio::test] +async fn the_validity_of_a_signed_keys_url_counts_from_after_the_storage_lookup() { + let url = slow_client() + .key_value_store("s") + .create_keys_public_url(Some(VALIDITY_SECS)) + .await + .unwrap(); + + assert_validity_counts_from_signing(&url); +} + +/// The same holds for the items URL of a dataset. +#[tokio::test] +async fn the_validity_of_a_signed_items_url_counts_from_after_the_dataset_lookup() { + let client = slow_client(); + let dataset = client.dataset("d"); + + let url = dataset + .create_items_public_url(Default::default(), Some(VALIDITY_SECS)) + .await + .unwrap(); + assert_validity_counts_from_signing(&url); + + let url = dataset + .create_items_public_url_with_format( + Default::default(), + Some(VALIDITY_SECS), + Some(apify_client::DownloadItemsFormat::Csv), + ) + .await + .unwrap(); + assert_validity_counts_from_signing(&url); +} diff --git a/tests/unit_server.rs b/tests/unit_server.rs new file mode 100644 index 0000000..a7ce74c --- /dev/null +++ b/tests/unit_server.rs @@ -0,0 +1,1034 @@ +//! Offline unit tests for the server interface of an Actor (`ActorClient::server`). +//! +//! These use a mock [`HttpBackend`](apify_client::http_client::HttpBackend) that answers the +//! requests in order, and require no network access or `APIFY_TOKEN`. + +mod common; + +use std::sync::Arc; + +use apify_client::clients::actor_server::{ + ActorServerBody, ActorServerCompression, ActorServerRequestOptions, +}; +use apify_client::http_client::{HttpMethod, HttpRequest, HttpResponse}; +use apify_client::{ApifyClientError, RequestCompression}; +use common::mock_backend::{client_with, MockBackend, MockOutcome}; +use futures_util::StreamExt; +use serde_json::{json, Value}; + +const SERVER_URL: &str = "https://user--actor.apify.actor"; + +fn actor(extra: Value) -> MockOutcome { + let mut data = json!({ "id": "a", "standbyUrl": SERVER_URL }); + data.as_object_mut() + .unwrap() + .extend(extra.as_object().unwrap().clone()); + MockOutcome::Status(200, json!({ "data": data }).to_string().into_bytes()) +} + +fn server_response(status: u16, body: &str) -> MockOutcome { + MockOutcome::Status(status, body.as_bytes().to_vec()) +} + +fn build(extra: Value) -> MockOutcome { + let mut data = json!({ "id": "b1" }); + data.as_object_mut() + .unwrap() + .extend(extra.as_object().unwrap().clone()); + MockOutcome::Status(200, json!({ "data": data }).to_string().into_bytes()) +} + +fn builds(items: Value) -> MockOutcome { + MockOutcome::Status( + 200, + json!({ "data": { "total": 0, "offset": 0, "limit": 1000, "count": 0, "desc": true, "items": items } }) + .to_string() + .into_bytes(), + ) +} + +fn options_with_params(params: &[(&str, &str)]) -> ActorServerRequestOptions { + ActorServerRequestOptions { + params: params + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect(), + ..Default::default() + } +} + +#[tokio::test] +async fn the_url_is_fetched_once_and_shared_by_the_clients_of_an_actor() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "{}")]); + let client = client_with(backend.clone(), 0); + let actor_client = client.actor("user~actor"); + + let url = actor_client.server().get_url().await.unwrap(); + assert_eq!(url, SERVER_URL); + actor_client.server().get_url().await.unwrap(); + actor_client + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!( + backend.urls(), + vec![ + "https://api.apify.com/v2/actors/user~actor".to_string(), + format!("{SERVER_URL}/"), + ] + ); +} + +#[tokio::test] +async fn an_actor_without_a_server_url_is_not_a_server_actor_and_the_failure_is_not_kept() { + let backend = MockBackend::new(vec![ + MockOutcome::Status( + 200, + json!({ "data": { "id": "a", "standbyUrl": null } }) + .to_string() + .into_bytes(), + ), + actor(json!({})), + ]); + let client = client_with(backend.clone(), 0); + let server = client.actor("a").server(); + + let error = server.get_url().await.unwrap_err(); + assert!( + matches!(&error, ApifyClientError::InvalidArgument(message) if message.contains("not a Server Actor")), + "{error:?}" + ); + assert_eq!(server.get_url().await.unwrap(), SERVER_URL); + assert_eq!(backend.call_count(), 2); +} + +#[tokio::test] +async fn a_missing_actor_is_an_error() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 404, + br#"{"error":{"type":"record-not-found","message":"nope"}}"#.to_vec(), + )]); + let client = client_with(backend, 0); + + let error = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap_err(); + assert_eq!(error.status_code(), Some(404), "{error:?}"); +} + +#[tokio::test] +async fn the_token_goes_in_the_x_apify_authorization_header_and_own_headers_reach_the_server() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "ok")]); + let client = client_with(backend.clone(), 0); + + let mut options = options_with_params(&[("q", "apify docs"), ("q", "b")]); + options.headers.insert( + "Authorization".to_string(), + "Bearer actor-token".to_string(), + ); + let response = client + .actor("a") + .server() + .request("//search?lang=en#fragment", options) + .await + .unwrap(); + + assert_eq!(response.text(), "ok"); + assert_eq!( + backend.last_url().unwrap(), + format!("{SERVER_URL}/search?lang=en&q=apify%20docs&q=b") + ); + assert_eq!( + backend.last_header("Authorization").as_deref(), + Some("Bearer actor-token") + ); + assert_eq!( + backend.last_header("X-Apify-Authorization").as_deref(), + Some("Bearer test-token") + ); + assert!(backend.last_header("User-Agent").is_some()); + assert_eq!(backend.follow_redirects(), vec![true, false]); +} + +#[tokio::test] +async fn the_token_does_not_travel_in_the_authorization_header_by_default() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "ok")]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!(backend.last_header("Authorization"), None); +} + +#[tokio::test] +async fn a_default_authorization_header_is_moved_too_and_a_request_header_of_either_name_wins() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(200, "ok"), + server_response(200, "ok"), + ]); + let client = apify_client::ApifyClient::builder() + .token("test-token") + .max_retries(0) + .default_header("authorization", "Bearer default") + .http_backend(backend.clone()) + .build(); + let server = client.actor("a").server(); + + server.request("/", Default::default()).await.unwrap(); + assert_eq!(backend.last_header("Authorization"), None); + assert_eq!( + backend.last_header("X-Apify-Authorization").as_deref(), + Some("Bearer default") + ); + + let mut options = ActorServerRequestOptions::default(); + options.headers.insert( + "x-apify-authorization".to_string(), + "Bearer own".to_string(), + ); + server.request("/", options).await.unwrap(); + assert_eq!( + backend.last_header("X-Apify-Authorization").as_deref(), + Some("Bearer own") + ); +} + +#[tokio::test] +async fn a_json_body_is_sent_with_its_method_and_content_type() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "{}")]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .request( + "/items", + ActorServerRequestOptions { + method: Some(HttpMethod::Patch), + body: Some(ActorServerBody::json(&json!({ "limit": 5 })).unwrap()), + ..Default::default() + }, + ) + .await + .unwrap(); + + assert_eq!(backend.methods(), vec![HttpMethod::Get, HttpMethod::Patch]); + assert_eq!( + backend.last_header("Content-Type").as_deref(), + Some("application/json") + ); + assert_eq!(backend.last_body().unwrap(), br#"{"limit":5}"#); +} + +#[tokio::test] +async fn text_and_binary_bodies_are_sent_as_they_are() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "{}")]); + let client = client_with(backend.clone(), 0); + let server = client.actor("a").server(); + + let mut text = ActorServerRequestOptions { + method: Some(HttpMethod::Post), + body: Some(ActorServerBody::Text("a=1".to_string())), + ..Default::default() + }; + text.headers.insert( + "Content-Type".to_string(), + "application/x-www-form-urlencoded".to_string(), + ); + server.request("/form", text).await.unwrap(); + assert_eq!(backend.last_body().unwrap(), b"a=1"); + assert_eq!( + backend.last_header("Content-Type").as_deref(), + Some("application/x-www-form-urlencoded") + ); + + server + .request( + "/bytes", + ActorServerRequestOptions { + method: Some(HttpMethod::Put), + body: Some(ActorServerBody::Bytes(vec![0, 1, 2])), + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(backend.last_body().unwrap(), vec![0, 1, 2]); + assert_eq!(backend.last_header("Content-Type"), None); +} + +#[tokio::test] +async fn an_error_status_is_returned_as_it_is_without_a_retry() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(404, r#"{"error":"not here"}"#), + ]); + let client = client_with(backend.clone(), 3); + + let response = client + .actor("a") + .server() + .request("/missing", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 404); + assert!(!response.is_success()); + assert_eq!( + response.json::().unwrap(), + json!({ "error": "not here" }) + ); + assert_eq!(backend.call_count(), 2, "the actor and one request"); +} + +#[tokio::test] +async fn a_server_error_other_than_503_is_not_retried() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(500, "boom")]); + let client = client_with(backend.clone(), 3); + + let response = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 500); + assert_eq!(backend.call_count(), 2); +} + +#[tokio::test] +async fn rate_limit_and_unavailable_responses_are_retried_and_the_last_one_is_returned() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(429, "slow down"), + server_response(503, "unavailable"), + ]); + let client = client_with(backend.clone(), 2); + + let response = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 503); + assert_eq!(backend.call_count(), 4, "the actor and three attempts"); + assert_eq!( + client.stats().rate_limit_errors().iter().sum::(), + 0, + "a server's rate limit is not one of the API" + ); +} + +#[tokio::test] +async fn a_503_that_clears_up_is_retried_into_a_success() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(503, "starting"), + server_response(200, "ready"), + ]); + let client = client_with(backend.clone(), 2); + + let response = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 200); + assert_eq!(response.text(), "ready"); +} + +#[tokio::test] +async fn network_errors_and_timeouts_are_not_retried() { + for outcome in [MockOutcome::NetworkError, MockOutcome::Timeout] { + let backend = MockBackend::new(vec![actor(json!({})), outcome]); + let client = client_with(backend.clone(), 3); + + let error = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap_err(); + + assert!( + matches!(error, ApifyClientError::Http(_) | ApifyClientError::Timeout), + "{error:?}" + ); + assert_eq!(backend.call_count(), 2, "the actor and one attempt"); + } +} + +#[tokio::test] +async fn a_redirect_is_returned_as_it_is() { + let backend = MockBackend::new(vec![ + actor(json!({})), + MockOutcome::StatusWithHeaders( + 302, + vec![("Location", "https://elsewhere.example.com/".to_string())], + Vec::new(), + ), + ]); + let client = client_with(backend.clone(), 3); + + let response = client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 302); + assert_eq!( + response.header("location"), + Some("https://elsewhere.example.com/") + ); + assert_eq!(backend.call_count(), 2); +} + +#[tokio::test] +async fn the_request_timeout_defaults_to_the_long_tier() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "ok")]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .request("/", Default::default()) + .await + .unwrap(); + + let timeouts = backend.timeouts(); + assert_eq!(timeouts[0], std::time::Duration::from_secs(5), "the actor"); + assert_eq!( + timeouts[1], + std::time::Duration::from_secs(360), + "the server" + ); +} + +#[tokio::test] +async fn a_request_body_is_sent_uncompressed_unless_compression_is_set() { + let large = "a".repeat(4096); + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(200, "{}"), + server_response(200, "{}"), + ]); + let client = client_with(backend.clone(), 0); + let server = client.actor("a").server(); + let options = |compression| ActorServerRequestOptions { + method: Some(HttpMethod::Post), + body: Some(ActorServerBody::Text(large.clone())), + compression, + ..Default::default() + }; + + server.request("/", options(None)).await.unwrap(); + assert_eq!(backend.last_header("Content-Encoding"), None); + assert_eq!(backend.last_body().unwrap().len(), large.len()); + + server + .request( + "/", + options(Some(ActorServerCompression::Algorithm( + RequestCompression::Gzip, + ))), + ) + .await + .unwrap(); + assert_eq!( + backend.last_header("Content-Encoding").as_deref(), + Some("gzip") + ); + assert!(backend.last_body().unwrap().len() < large.len()); +} + +#[tokio::test] +async fn a_custom_compressor_compresses_the_body() { + #[derive(Debug)] + struct Marker; + impl apify_client::HttpCompressor for Marker { + fn content_encoding(&self) -> &str { + "marker" + } + fn compress(&self, data: &[u8]) -> apify_client::ApifyClientResult> { + Ok(data[..4].to_vec()) + } + } + + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "{}")]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .request( + "/", + ActorServerRequestOptions { + method: Some(HttpMethod::Post), + body: Some(ActorServerBody::Text("b".repeat(4096))), + compression: Some(ActorServerCompression::Custom(Arc::new(Marker))), + ..Default::default() + }, + ) + .await + .unwrap(); + + assert_eq!( + backend.last_header("Content-Encoding").as_deref(), + Some("marker") + ); + assert_eq!(backend.last_body().unwrap(), b"bbbb"); +} + +#[tokio::test] +async fn a_stream_response_is_returned_with_any_status() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(418, "teapot")]); + let client = client_with(backend.clone(), 3); + + let mut response = client + .actor("a") + .server() + .request_stream("/tea", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 418); + let mut body = Vec::new(); + while let Some(chunk) = response.body.next().await { + body.extend(chunk.unwrap()); + } + assert_eq!(body, b"teapot"); + assert_eq!(backend.call_count(), 2); + assert_eq!(backend.last_header("Authorization"), None); + assert_eq!( + backend.last_header("X-Apify-Authorization").as_deref(), + Some("Bearer test-token") + ); + assert_eq!(backend.follow_redirects(), vec![true, false]); +} + +#[tokio::test] +async fn the_standby_alias_returns_a_server_client() { + let backend = MockBackend::new(vec![actor(json!({}))]); + let client = client_with(backend, 0); + + #[allow(deprecated)] // The alias is deprecated, and this test covers it. + let url = client.actor("a").standby().get_url().await.unwrap(); + assert_eq!(url, SERVER_URL); +} + +#[tokio::test] +async fn the_spec_is_the_web_server_schema_of_the_latest_build_by_default() { + let schema = json!({ "openapi": "3.0.0", "paths": {} }); + let backend = MockBackend::new(vec![ + actor(json!({ "taggedBuilds": { "latest": { "buildId": "b1" } } })), + build(json!({ "actorDefinition": { "webServerSchema": schema } })), + ]); + let client = client_with(backend.clone(), 0); + let server = client.actor("a").server(); + + assert_eq!(server.get_open_api_spec().await.unwrap(), Some(schema)); + assert_eq!( + backend.last_url().unwrap(), + "https://api.apify.com/v2/actor-builds/b1" + ); + // The spec lookup cached the URL. + assert_eq!(server.get_url().await.unwrap(), SERVER_URL); + assert_eq!(backend.call_count(), 2); +} + +#[tokio::test] +async fn the_spec_comes_from_the_build_tag_the_server_runs() { + let backend = MockBackend::new(vec![ + actor(json!({ + "actorStandby": { "build": "beta" }, + "taggedBuilds": { "latest": { "buildId": "b1" }, "beta": { "buildId": "b2" } }, + })), + build( + json!({ "id": "b2", "actorDefinition": { "webServerSchema": { "openapi": "beta" } } }), + ), + ]); + let client = client_with(backend.clone(), 0); + + let spec = client + .actor("a") + .server() + .get_open_api_spec() + .await + .unwrap(); + + assert_eq!(spec, Some(json!({ "openapi": "beta" }))); + assert_eq!( + backend.last_url().unwrap(), + "https://api.apify.com/v2/actor-builds/b2" + ); +} + +#[tokio::test] +async fn the_spec_looks_up_a_build_number_in_the_builds_of_the_actor() { + let backend = MockBackend::new(vec![ + actor(json!({ "actorStandby": { "build": "0.1.2" }, "taggedBuilds": {} })), + builds(json!([ + { "id": "b3", "buildNumber": "0.1.3" }, + { "id": "b2", "buildNumber": "0.1.2" }, + ])), + build(json!({ "id": "b2", "actorDefinition": { "webServerSchema": { "openapi": "2" } } })), + ]); + let client = client_with(backend.clone(), 0); + + let spec = client + .actor("a") + .server() + .get_open_api_spec() + .await + .unwrap(); + + assert_eq!(spec, Some(json!({ "openapi": "2" }))); + let urls = backend.urls(); + assert!( + urls[1].starts_with("https://api.apify.com/v2/actors/a/builds?"), + "{urls:?}" + ); + assert!(urls[1].contains("desc=1"), "{urls:?}"); + assert_eq!(urls[2], "https://api.apify.com/v2/actor-builds/b2"); +} + +#[tokio::test] +async fn the_build_number_search_stops_at_an_earlier_build_of_the_same_version() { + let backend = MockBackend::new(vec![ + actor(json!({ "actorStandby": { "build": "0.1.5" }, "taggedBuilds": {} })), + builds(json!([ + { "id": "b6", "buildNumber": "0.1.6" }, + { "id": "b4", "buildNumber": "0.1.4" }, + ])), + ]); + let client = client_with(backend.clone(), 0); + + let error = client + .actor("a") + .server() + .get_open_api_spec() + .await + .unwrap_err(); + + assert!( + matches!(&error, ApifyClientError::InvalidArgument(message) if message.contains("Build 0.1.5 of Actor a does not exist")), + "{error:?}" + ); + assert_eq!(backend.call_count(), 2, "the older build ended the search"); +} + +#[tokio::test] +async fn a_tag_without_a_build_is_an_error() { + let backend = MockBackend::new(vec![actor(json!({ "taggedBuilds": { "latest": null } }))]); + let client = client_with(backend.clone(), 0); + + let result = client.actor("a").server().get_open_api_spec().await; + + // `latest` is neither a tag with a build nor a build number, so no build is looked up. + assert!( + matches!(&result, Err(ApifyClientError::InvalidArgument(message)) if message.contains("Build latest of Actor a does not exist")), + "{result:?}" + ); + assert_eq!(backend.call_count(), 1); +} + +#[tokio::test] +async fn a_build_the_api_does_not_return_is_an_error_even_with_a_known_id() { + let backend = MockBackend::new(vec![ + actor(json!({ "taggedBuilds": { "latest": { "buildId": "gone" } } })), + MockOutcome::Status(404, br#"{"error":{"message":"gone"}}"#.to_vec()), + ]); + let client = client_with(backend, 0); + + let result = client.actor("a").server().get_open_api_spec().await; + + assert!( + matches!(&result, Err(ApifyClientError::InvalidArgument(_))), + "{result:?}" + ); +} + +#[tokio::test] +async fn the_spec_is_none_when_the_definition_has_no_web_server_schema() { + for definition in [ + json!({}), + json!({ "actorDefinition": null }), + json!({ "actorDefinition": {} }), + ] { + let backend = MockBackend::new(vec![ + actor(json!({ "taggedBuilds": { "latest": { "buildId": "b1" } } })), + build(definition), + ]); + let client = client_with(backend, 0); + + let spec = client + .actor("a") + .server() + .get_open_api_spec() + .await + .unwrap(); + + assert_eq!(spec, None); + } +} + +#[tokio::test] +async fn the_spec_of_an_actor_that_is_not_a_server_actor_is_an_error() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 200, + json!({ "data": { "id": "a" } }).to_string().into_bytes(), + )]); + let client = client_with(backend, 0); + + let result = client.actor("a").server().get_open_api_spec().await; + + assert!( + matches!(&result, Err(ApifyClientError::InvalidArgument(message)) if message.contains("not a Server Actor")), + "{result:?}" + ); +} + +#[tokio::test] +async fn with_timeout_sets_the_timeout_of_every_request() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(200, "ok")]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .with_timeout(apify_client::Timeout::Short) + .request("/", Default::default()) + .await + .unwrap(); + + let timeouts = backend.timeouts(); + assert_eq!(timeouts[0], std::time::Duration::from_secs(5)); + assert_eq!(timeouts[1], std::time::Duration::from_secs(5)); +} + +#[test] +fn http_response_helpers_describe_the_body() { + let response = HttpResponse { + status: 204, + headers: Default::default(), + body: b"{\"a\":1}".to_vec(), + }; + assert!(response.is_success()); + assert_eq!(response.text(), "{\"a\":1}"); + assert_eq!(response.json::().unwrap(), json!({ "a": 1 })); + assert!(!HttpResponse { + status: 300, + ..response + } + .is_success()); +} + +#[tokio::test] +async fn the_default_backend_does_not_follow_a_redirect_when_asked_not_to() { + use apify_client::http_client::{HttpBackend, ReqwestBackend}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut buffer = [0u8; 1024]; + let _ = socket.read(&mut buffer).await.unwrap(); + socket + .write_all(b"HTTP/1.1 302 Found\r\nlocation: http://127.0.0.1:1/\r\ncontent-length: 0\r\nconnection: close\r\n\r\n") + .await + .unwrap(); + }); + + let response = ReqwestBackend::new() + .send(HttpRequest { + method: HttpMethod::Get, + url: format!("http://{address}/"), + headers: Default::default(), + body: None, + timeout: std::time::Duration::from_secs(5), + follow_redirects: false, + }) + .await + .expect("the redirect is returned, not followed"); + + assert_eq!(response.status, 302); + assert_eq!(response.header("location"), Some("http://127.0.0.1:1/")); +} + +#[tokio::test] +async fn a_stream_response_is_retried_on_503_until_it_starts() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(503, "starting"), + server_response(200, "ready"), + ]); + let client = client_with(backend.clone(), 2); + + let mut response = client + .actor("a") + .server() + .request_stream("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 200); + let chunk = response.body.next().await.unwrap().unwrap(); + assert_eq!(chunk, b"ready"); + assert_eq!(backend.call_count(), 3, "the actor and two attempts"); +} + +#[tokio::test] +async fn a_stream_response_with_an_error_status_that_is_not_retried_is_returned_at_once() { + let backend = MockBackend::new(vec![actor(json!({})), server_response(500, "boom")]); + let client = client_with(backend.clone(), 3); + + let response = client + .actor("a") + .server() + .request_stream("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 500); + assert_eq!(backend.call_count(), 2); +} + +/// A backend whose streaming responses start at once and then stay silent for a long while. +#[derive(Debug)] +struct SilentStreamBackend { + actor: Vec, +} + +/// How long the stream of [`SilentStreamBackend`] stays silent between its two chunks. +const SILENCE: std::time::Duration = std::time::Duration::from_secs(600); + +#[async_trait::async_trait] +impl apify_client::http_client::HttpBackend for SilentStreamBackend { + async fn send(&self, _request: HttpRequest) -> apify_client::ApifyClientResult { + Ok(HttpResponse { + status: 200, + headers: Default::default(), + body: self.actor.clone(), + }) + } + + async fn send_streaming( + &self, + _request: HttpRequest, + ) -> apify_client::ApifyClientResult { + let body = futures_util::stream::unfold(0u8, |step| async move { + match step { + 0 => Some((Ok(b"first".to_vec()), 1)), + 1 => { + tokio::time::sleep(SILENCE).await; + Some((Ok(b"second".to_vec()), 2)) + } + _ => None, + } + }); + Ok(apify_client::http_client::StreamingHttpResponse { + status: 200, + headers: Default::default(), + body: Box::pin(body), + }) + } +} + +/// The timeout of a server stream bounds only the wait for the response to start: a server that +/// keeps the connection open and stays silent between two chunks is not cut off. +#[tokio::test(start_paused = true)] +async fn a_server_stream_is_not_cut_off_while_the_server_stays_silent() { + let backend = Arc::new(SilentStreamBackend { + actor: json!({ "data": { "id": "a", "standbyUrl": SERVER_URL } }) + .to_string() + .into_bytes(), + }); + let client = apify_client::ApifyClient::builder() + .token("test-token") + .http_backend(backend) + .build(); + let server = client + .actor("a") + .server() + .with_timeout(apify_client::Timeout::Short); + + let mut response = server + .request_stream("/events", Default::default()) + .await + .unwrap(); + let mut chunks = Vec::new(); + while let Some(chunk) = response.body.next().await { + chunks.push(chunk.expect("no timeout")); + } + + assert_eq!(chunks, vec![b"first".to_vec(), b"second".to_vec()]); +} + +#[tokio::test] +async fn a_stream_response_is_not_retried_on_a_network_error_or_a_timeout() { + for outcome in [MockOutcome::NetworkError, MockOutcome::Timeout] { + let backend = MockBackend::new(vec![actor(json!({})), outcome]); + let client = client_with(backend.clone(), 3); + + let error = client + .actor("a") + .server() + .request_stream("/", Default::default()) + .await + .err() + .expect("the transport error ends the call"); + + assert!( + matches!(error, ApifyClientError::Http(_) | ApifyClientError::Timeout), + "{error:?}" + ); + assert_eq!(backend.call_count(), 2, "the actor and one attempt"); + } +} + +#[tokio::test] +async fn a_stream_response_returns_the_last_retried_status_once_the_retries_run_out() { + let backend = MockBackend::new(vec![ + actor(json!({})), + server_response(429, "slow down"), + server_response(503, "unavailable"), + ]); + let client = client_with(backend.clone(), 1); + + let mut response = client + .actor("a") + .server() + .request_stream("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 503); + assert_eq!(response.body.next().await.unwrap().unwrap(), b"unavailable"); + assert_eq!(backend.call_count(), 3, "the actor and two attempts"); + assert_eq!( + client.stats().rate_limit_errors().iter().sum::(), + 0, + "a server's rate limit is not one of the API" + ); +} + +/// The body of a streaming response. +type ErrorBody = std::pin::Pin< + Box>> + Send>, +>; + +/// A backend whose streaming error responses never finish their body: the status arrives at once +/// and the body then stays silent for as long as the test lets it. +#[derive(Debug)] +struct HangingErrorBodyBackend { + actor: Vec, + stream_calls: std::sync::atomic::AtomicUsize, +} + +#[async_trait::async_trait] +impl apify_client::http_client::HttpBackend for HangingErrorBodyBackend { + async fn send(&self, _request: HttpRequest) -> apify_client::ApifyClientResult { + Ok(HttpResponse { + status: 200, + headers: Default::default(), + body: self.actor.clone(), + }) + } + + async fn send_streaming( + &self, + _request: HttpRequest, + ) -> apify_client::ApifyClientResult { + let call = self + .stream_calls + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + // The first attempt fails with a `503` whose body never ends, the second one succeeds. + let (status, body): (u16, ErrorBody) = if call == 0 { + (503, Box::pin(futures_util::stream::pending())) + } else { + ( + 200, + Box::pin(futures_util::stream::once(async { Ok(b"ready".to_vec()) })), + ) + }; + Ok(apify_client::http_client::StreamingHttpResponse { + status, + headers: Default::default(), + body, + }) + } +} + +/// A retry that follows drops the failed response unread: a `503` whose body never ends does not +/// delay or prevent the retry. +#[tokio::test(start_paused = true)] +async fn a_stream_retry_does_not_wait_for_the_body_of_the_failed_response() { + let backend = Arc::new(HangingErrorBodyBackend { + actor: json!({ "data": { "id": "a", "standbyUrl": SERVER_URL } }) + .to_string() + .into_bytes(), + stream_calls: Default::default(), + }); + let client = apify_client::ApifyClient::builder() + .token("test-token") + .max_retries(1) + .min_delay_between_retries(std::time::Duration::from_millis(1)) + .http_backend(backend) + .build(); + + let started = tokio::time::Instant::now(); + let mut response = client + .actor("a") + .server() + .request_stream("/", Default::default()) + .await + .unwrap(); + + assert_eq!(response.status, 200); + assert_eq!(response.body.next().await.unwrap().unwrap(), b"ready"); + assert!( + started.elapsed() < std::time::Duration::from_secs(5), + "the retry waited for the body of the failed response" + ); +} + +/// The builds listing that looks up a build number uses the `Short` tier, like every API request +/// of the server client, unless the client has a timeout of its own. +#[tokio::test] +async fn the_build_number_lookup_lists_builds_with_the_short_tier() { + let backend = MockBackend::new(vec![ + actor(json!({ "actorStandby": { "build": "0.1.2" }, "taggedBuilds": {} })), + builds(json!([{ "id": "b2", "buildNumber": "0.1.2" }])), + build(json!({ "id": "b2", "actorDefinition": { "webServerSchema": {} } })), + ]); + let client = client_with(backend.clone(), 0); + + client + .actor("a") + .server() + .get_open_api_spec() + .await + .unwrap(); + + let short = std::time::Duration::from_secs(5); + assert_eq!(backend.timeouts(), vec![short, short, short]); +} diff --git a/tests/unit_timeouts.rs b/tests/unit_timeouts.rs index 020d367..2afe3bf 100644 --- a/tests/unit_timeouts.rs +++ b/tests/unit_timeouts.rs @@ -269,6 +269,7 @@ async fn reqwest_backend_sends_a_request_without_a_timeout() { headers: Default::default(), body: None, timeout: NO_TIMEOUT, + follow_redirects: true, }) .await .expect("a request without a timeout is sent"); @@ -293,6 +294,7 @@ async fn reqwest_backend_aborts_a_request_that_outlives_its_timeout() { headers: Default::default(), body: None, timeout: Duration::from_millis(50), + follow_redirects: true, }) .await .expect_err("the request must time out");