diff --git a/CHANGELOG.md b/CHANGELOG.md index 8daf21e..5944509 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,29 @@ 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.13.0] - 2026-10-19 + +### Added +- `KeyValueStoreClient::set_record_raw_with_content_encoding`: stores an already-compressed + value and sends the given `Content-Encoding` header instead of compressing the bytes again, + matching the reference client's new `setRecord({ contentEncoding })`. +- `LogClient::get_bytes`/`get_bytes_with_options`: fetch a log as raw bytes instead of a + (lossily-decoded) `String`, matching the reference client's new `log.get({ buffer: true })`. +- `DatasetClient::download_items_stream`: like `download_items`, but streams the export instead + of buffering it in memory, matching the reference client's new `downloadItems({ stream: true })`. +- `HttpBackend::send_streaming`: a new trait method (with a buffering default implementation, so + existing custom backends keep compiling) that lets a streaming request go through the + configured backend. `ReqwestBackend` overrides it for real constant-memory streaming. + +### Fixed +- `log().stream()`/`stream_with_options()` (and the new `download_items_stream`) now go through + the client's configured `HttpBackend` — including a fully custom one, or a `ReqwestBackend` + wrapping a caller-provided `reqwest::Client` — instead of always opening a fresh, unconfigured + connection that silently ignored any injected backend's proxy/TLS/connection-pool setup. +- A streaming request that never receives a response now times out instead of hanging forever; + once a response starts, silence between chunks (not the stream's total duration) is what times + it out, so a long-but-active stream (e.g. live log redirection) is never cut off. + ## [0.12.0] - 2026-10-09 ### Added diff --git a/Cargo.toml b/Cargo.toml index bb38e74..ecfa8cf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apify-client" -version = "0.12.0" +version = "0.13.0" authors = ["Apify Technologies "] description = "An official, but experimental, AI-generated and AI-maintained Rust client for the Apify API (https://apify.com)." license = "Apache-2.0" diff --git a/README.md b/README.md index 36ac5c3..5249799 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ It provides a resource-oriented, async interface that mirrors the official ```toml [dependencies] -apify-client = "0.6" +apify-client = "0.13" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } serde_json = "1" # for the `serde_json::Value` responses used in the Quick start ``` @@ -50,7 +50,7 @@ project needs `serde_json`. Two more dependencies are needed only for specific f By default the client uses the system TLS (`native-tls`). To use rustls instead: ```toml -apify-client = { version = "0.6", default-features = false, features = ["rustls"] } +apify-client = { version = "0.13", default-features = false, features = ["rustls"] } ``` ## Quick start @@ -293,7 +293,10 @@ The transport is a replaceable component: any type implementing `http_client::Ht injected with `ApifyClientBuilder::http_backend`. The most common customization is to reuse the built-in `ReqwestBackend` with a `reqwest::Client` you pre-configure (custom proxy, TLS, connection pool, …), as shown below; to route requests through an entirely different HTTP stack, implement -`HttpBackend` (a single `async fn send`) on your own type instead. The snippet below constructs a +`HttpBackend` on your own type instead — just `async fn send`; `send_streaming` (used by +`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. The +snippet below constructs a `reqwest::Client` directly, so it requires adding `reqwest = "0.12"` (matching this crate's `reqwest` version) as your own dependency — see [Installation](#installation): diff --git a/docs/README.md b/docs/README.md index 99070cf..c9f562d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -31,7 +31,7 @@ Add the crate and an async runtime: ```toml [dependencies] -apify-client = "0.6" +apify-client = "0.13" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } ``` diff --git a/docs/misc.md b/docs/misc.md index 6ca5ab9..7db7cf0 100644 --- a/docs/misc.md +++ b/docs/misc.md @@ -140,6 +140,8 @@ Also reachable via `run.log()` and `build.log()`. |---|---|---|---| | `get()` | — | `Option` | The entire log as text. | | `get_with_options(options)` | `LogOptions` | `Option` | As `get()`, with options (e.g. `raw`). | +| `get_bytes()` | — | `Option>` | The entire log as raw bytes (e.g. if it is not valid UTF-8). | +| `get_bytes_with_options(options)` | `LogOptions` | `Option>` | As `get_bytes()`, with options (e.g. `raw`). | | `stream()` | — | `Result>>>` (async — `.await` it) | Streams log chunks live (log redirection). | | `stream_with_options(options)` | `LogOptions` | `Result>>>` (async — `.await` it) | As `stream()`, with options (e.g. `raw`). | @@ -164,7 +166,10 @@ if let Some(text) = raw_log { ``` The streaming variant takes the same options — `client.run(run_id).log().stream_with_options(LogOptions { raw: Some(true) }).await?` -yields the raw log chunks (this is what log redirection uses). +yields the raw log chunks (this is what log redirection uses). A gap of more than 6 minutes +between chunks (the run legitimately logging nothing for that long) ends the stream with an +`ApifyClientError::Timeout`; this is an inactivity timeout, not a cap on the stream's total +duration, so an actively (if slowly) logging run is never cut off. Consuming `stream()` requires the `futures_util::StreamExt` trait (from the `futures-util` crate) in scope to call `.next()` on the returned stream. Add it to your `Cargo.toml`: diff --git a/docs/storages.md b/docs/storages.md index 2baae76..53999ff 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -65,6 +65,7 @@ let scratch = client.datasets().get_or_create(None).await?; | `push_items(items)` | `&impl Serialize` | `()` | Appends items (object or array). | | `get_statistics()` | — | `Option` | Field statistics. | | `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`. | @@ -112,6 +113,29 @@ println!("exported {} bytes of CSV", csv.len()); # } ``` +For a large export, `download_items_stream` avoids holding the whole thing in memory. Like +`LogClient::stream` (see [Logs](misc.md#logs--clientlogbuild_or_run_id)), consuming it requires +the `futures_util::StreamExt` trait in scope: + +```rust,no_run +# use apify_client::{ApifyClient, DownloadItemsFormat}; +use futures_util::StreamExt; + +# async fn run(client: ApifyClient) -> Result<(), Box> { +let dataset = client.datasets().get_or_create(None).await?; +let mut stream = client + .dataset(&dataset.id) + .download_items_stream(DownloadItemsFormat::Jsonl, Default::default()) + .await?; +let mut total_bytes = 0; +while let Some(chunk) = stream.next().await { + total_bytes += chunk?.len(); +} +println!("streamed {total_bytes} bytes of JSONL"); +# Ok(()) +# } +``` + ## Key-value stores — `client.key_value_stores()` / `client.key_value_store(id)` `KeyValueStoreCollectionClient`: `list(options: StorageListOptions)`, @@ -130,6 +154,7 @@ println!("exported {} bytes of CSV", csv.len()); | `record_exists(key)` | `&str` | `bool` | Whether a record exists (HEAD). | | `get_record(key)` | `&str` | `Option` | Reads a record's raw value. | | `set_record_raw(key, bytes, content_type)` | `&str`, `Vec`, `&str` | `()` | Stores a raw record. | +| `set_record_raw_with_content_encoding(key, bytes, content_type, content_encoding)` | `&str`, `Vec`, `&str`, `&str` | `()` | Like `set_record_raw`, for a value that is already compressed (`gzip`, `br`, `deflate`): sends `content_encoding` as the `Content-Encoding` header instead of compressing the bytes again. | | `set_record_json(key, value)` | `&str`, `&impl Serialize` | `()` | Stores a JSON record. | | `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. | diff --git a/src/clients/base.rs b/src/clients/base.rs index 23b236e..371635e 100644 --- a/src/clients/base.rs +++ b/src/clients/base.rs @@ -368,10 +368,28 @@ pub(crate) async fn put_raw( params: &QueryParams, body: Vec, content_type: &str, +) -> ApifyClientResult<()> { + put_raw_with_content_encoding(ctx, sub_path, params, body, content_type, None).await +} + +/// Like [`put_raw`], but can also set a `Content-Encoding` header (used for KVS record uploads +/// of an already-compressed value). Setting it makes [`HttpClient::call`]'s automatic request +/// compression skip the body (it already refuses to double-encode a request that carries a +/// `Content-Encoding`), so the bytes reach the API exactly as given. +pub(crate) async fn put_raw_with_content_encoding( + ctx: &ResourceContext, + sub_path: Option<&str>, + params: &QueryParams, + body: Vec, + content_type: &str, + content_encoding: Option<&str>, ) -> ApifyClientResult<()> { let url = ctx.merged_params(params).apply_to_url(&ctx.url(sub_path)); let mut headers = std::collections::HashMap::new(); headers.insert("Content-Type".to_string(), content_type.to_string()); + if let Some(encoding) = content_encoding { + headers.insert("Content-Encoding".to_string(), encoding.to_string()); + } ctx.http .call(HttpRequest { method: HttpMethod::Put, @@ -384,6 +402,28 @@ pub(crate) async fn put_raw( Ok(()) } +/// Opens a streaming `GET` connection to `url` (already fully built, including query string), +/// for a caller that wants to process a large body (a log or a dataset export) incrementally +/// instead of waiting for [`HttpClient::call`] to buffer the whole response. Delegates to +/// [`HttpClient::call_streaming`], which — unlike a one-off raw `reqwest` call — goes through +/// the configured [`HttpBackend`](crate::http_client::HttpBackend), so a custom backend (or a +/// `ReqwestBackend` wrapping a caller-provided, proxy/TLS-configured `reqwest::Client`) is +/// honoured for streaming requests exactly as it is for buffered ones. +pub(crate) async fn get_byte_stream( + ctx: &ResourceContext, + url: String, +) -> ApifyClientResult>>> { + ctx.http + .call_streaming(HttpRequest { + method: HttpMethod::Get, + url, + headers: Default::default(), + body: None, + timeout: DEFAULT_REQUEST_TIMEOUT, + }) + .await +} + /// A `POST` with a raw body that returns an enveloped resource. Used by `actor.start` /// and `run.metamorph` where the input is arbitrary user JSON. pub(crate) async fn post_with_body( diff --git a/src/clients/dataset.rs b/src/clients/dataset.rs index 2d2b57f..65423ab 100644 --- a/src/clients/dataset.rs +++ b/src/clients/dataset.rs @@ -4,7 +4,9 @@ use serde::de::DeserializeOwned; use serde::Serialize; use serde_json::Value; -use crate::clients::base::{delete_resource, get_resource, update_resource, ResourceContext}; +use crate::clients::base::{ + delete_resource, get_byte_stream, get_resource, update_resource, ResourceContext, +}; use crate::clients::pagination::ListIterator; use crate::common::{parse_data_envelope, sign_storage_content, PaginationList, QueryParams}; use crate::error::ApifyClientResult; @@ -310,6 +312,31 @@ impl DatasetClient { Ok(response.body) } + /// Like [`download_items`](Self::download_items), but streams the export as it arrives + /// instead of buffering the whole response in memory. + /// + /// Returns a `Stream` of byte chunks; drive it with `futures_util::StreamExt::next()` (or + /// pipe it through an `AsyncRead` adapter) to process a large export in constant memory. A + /// gap of more than 6 minutes between chunks ends the stream with + /// [`ApifyClientError::Timeout`](crate::error::ApifyClientError::Timeout) — an inactivity + /// timeout, not a cap on the export's total duration. + /// + /// Goes through the configured `HttpBackend`'s streaming path instead of the buffered one, + /// to keep the connection open — like [`crate::clients::log::LogClient::stream`], a failed + /// request (e.g. a `404`) still surfaces the same structured [`crate::error::ApiError`] as + /// the buffered endpoints, from reading the error body before the stream is handed back. + pub async fn download_items_stream( + &self, + format: DownloadItemsFormat, + options: DatasetDownloadOptions, + ) -> ApifyClientResult>>> { + let mut params = QueryParams::new(); + params.add_str("format", Some(format.as_str())); + options.apply(&mut params); + let url = params.apply_to_url(&self.ctx.url(Some("items"))); + get_byte_stream(&self.ctx, url).await + } + /// Pushes one or more items to the dataset. /// /// `items` must serialize to a JSON object or an array of objects. diff --git a/src/clients/key_value_store.rs b/src/clients/key_value_store.rs index bd68d74..65c9619 100644 --- a/src/clients/key_value_store.rs +++ b/src/clients/key_value_store.rs @@ -6,7 +6,7 @@ use serde::Serialize; use crate::clients::base::{ delete_resource, get_raw, get_resource, get_resource_required, head_exists, put_raw, - update_resource, ResourceContext, + put_raw_with_content_encoding, update_resource, ResourceContext, }; use crate::common::{ create_hmac_signature, encode_path_segment, sign_storage_content, QueryParams, @@ -201,6 +201,36 @@ impl KeyValueStoreClient { .await } + /// Stores a record with raw, already-compressed bytes, sending `content_encoding` as the + /// `Content-Encoding` header so it is uploaded (and later served) exactly as given, without + /// the client compressing it again. + /// + /// The API accepts `gzip`, `br`, `deflate`, and `identity`. Unlike the reference client's + /// `setRecord({ contentEncoding })`, there is no separate runtime check rejecting a + /// non-identity encoding on a string or JSON value: this method, like + /// [`set_record_raw`](Self::set_record_raw), only ever accepts bytes that are already the + /// wire payload, so that case cannot arise here — pass pre-serialized/pre-compressed bytes, + /// not a value for the client to serialize. For plain (uncompressed) raw bytes, use + /// [`set_record_raw`](Self::set_record_raw); for a JSON value, use + /// [`set_record_json`](Self::set_record_json). + pub async fn set_record_raw_with_content_encoding( + &self, + key: &str, + value: Vec, + content_type: &str, + content_encoding: &str, + ) -> ApifyClientResult<()> { + put_raw_with_content_encoding( + &self.ctx, + Some(&format!("records/{}", encode_path_segment(key))), + &QueryParams::new(), + value, + content_type, + Some(content_encoding), + ) + .await + } + /// Stores a record as JSON (the value is serialized and content type set to JSON). pub async fn set_record_json( &self, diff --git a/src/clients/log.rs b/src/clients/log.rs index ca78aea..ab7bde3 100644 --- a/src/clients/log.rs +++ b/src/clients/log.rs @@ -6,9 +6,9 @@ use futures_util::Stream; -use crate::clients::base::{get_raw, ResourceContext}; +use crate::clients::base::{get_byte_stream, get_raw, ResourceContext}; use crate::common::QueryParams; -use crate::error::{ApifyClientError, ApifyClientResult}; +use crate::error::ApifyClientResult; use crate::http_client::HttpClient; /// Options for retrieving or streaming a log ([`LogClient::get_with_options`] / @@ -28,42 +28,27 @@ pub struct LogOptions { #[derive(Debug, Clone)] pub struct LogClient { ctx: ResourceContext, - /// The URL used for streaming; we keep it so streaming can bypass the buffered path. - stream_url: String, - token: Option, - user_agent: String, } impl LogClient { pub(crate) fn new(http: HttpClient, base_url: &str, resource_path: &str, id: &str) -> Self { - let ctx = ResourceContext::single(http, base_url, resource_path, id); - let stream_url = ctx.url(None); - let (token, user_agent) = ctx.http.stream_credentials(); Self { - ctx, - stream_url, - token, - user_agent, + ctx: ResourceContext::single(http, base_url, resource_path, id), } } /// Creates a log client nested under a run or build (path `.../log`). pub(crate) fn nested(http: HttpClient, base_url: &str, sub_path: &str) -> Self { - let ctx = ResourceContext::collection(http, base_url, sub_path); - let stream_url = ctx.url(None); - let (token, user_agent) = ctx.http.stream_credentials(); Self { - ctx, - stream_url, - token, - user_agent, + ctx: ResourceContext::collection(http, base_url, sub_path), } } /// Fetches the entire log as a string, or `None` if it does not exist. /// /// Uses the default (processed) log format. To request the raw log, use - /// [`LogClient::get_with_options`]. + /// [`LogClient::get_with_options`]. To get the log as raw bytes instead of a `String` + /// (e.g. if it is not valid UTF-8), use [`LogClient::get_bytes`]. pub async fn get(&self) -> ApifyClientResult> { self.get_with_options(LogOptions::default()).await } @@ -71,17 +56,41 @@ impl LogClient { /// Fetches the entire log as a string, or `None` if it does not exist, applying the given /// [`LogOptions`] (e.g. [`LogOptions::raw`]). pub async fn get_with_options(&self, options: LogOptions) -> ApifyClientResult> { + Ok(self + .get_bytes_with_options(options) + .await? + .map(|bytes| String::from_utf8_lossy(&bytes).into_owned())) + } + + /// Fetches the entire log as raw bytes, or `None` if it does not exist. + /// + /// The counterpart of [`LogClient::get`] for a log that is not valid UTF-8, or when the + /// caller wants to avoid the lossy-UTF-8 conversion `get()` applies. Uses the default + /// (processed) log format; see [`LogClient::get_bytes_with_options`] for the raw format. + pub async fn get_bytes(&self) -> ApifyClientResult>> { + self.get_bytes_with_options(LogOptions::default()).await + } + + /// Fetches the entire log as raw bytes, or `None` if it does not exist, applying the given + /// [`LogOptions`] (e.g. [`LogOptions::raw`]). The byte-returning counterpart of + /// [`LogClient::get_with_options`]. + pub async fn get_bytes_with_options( + &self, + options: LogOptions, + ) -> ApifyClientResult>> { let mut params = QueryParams::new(); params.add_bool("raw", options.raw); let response = get_raw(&self.ctx, None, ¶ms).await?; - Ok(response.map(|r| String::from_utf8_lossy(&r.body).into_owned())) + Ok(response.map(|r| r.body)) } /// Opens a streaming connection to the log, yielding chunks of bytes as they arrive. /// /// This powers real-time log redirection: callers can forward each chunk to their own - /// logger/stdout while a run is still in progress. The stream completes when the log - /// ends (i.e. the run finishes). + /// logger/stdout while a run is still in progress. The stream completes when the log ends + /// (i.e. the run finishes) — as long as a new chunk arrives at least every 6 minutes; a run + /// that legitimately logs nothing for that long errors with + /// [`ApifyClientError::Timeout`](crate::error::ApifyClientError::Timeout) instead. /// /// Uses the default (processed) log format. To stream the raw log, use /// [`LogClient::stream_with_options`]. @@ -101,49 +110,13 @@ impl LogClient { &self, options: LogOptions, ) -> ApifyClientResult>>> { - // Streaming needs a live connection, so we go through reqwest directly rather than - // the buffered backend path. The retry policy does not apply to an open stream. - let client = reqwest::Client::new(); + // 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. let mut params = QueryParams::new(); params.push_raw("stream".to_string(), "1".to_string()); params.add_bool("raw", options.raw); - let url = params.apply_to_url(&self.stream_url); - - let mut builder = client.get(&url).header("User-Agent", &self.user_agent); - if let Some(token) = &self.token { - builder = builder.header("Authorization", format!("Bearer {token}")); - } - - let response = builder.send().await.map_err(ApifyClientError::from)?; - if !response.status().is_success() { - let status = response.status().as_u16(); - // The error body arrives as a stream here too (this request never goes through the - // buffered backend path). Read it so the failure carries the same structured - // `ApiError` (type/message/data) as every other 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 = response - .bytes() - .await - .map(|b| b.to_vec()) - .unwrap_or_default(); - let http_response = crate::http_client::HttpResponse { - status, - headers: Default::default(), - body, - }; - let api_error = crate::http_client::build_api_error( - &http_response, - 1, - "GET", - &crate::http_client::extract_path(&url), - ); - return Err(ApifyClientError::from(api_error)); - } - - let byte_stream = response.bytes_stream(); - Ok(futures_util::StreamExt::map(byte_stream, |chunk| { - chunk.map(|b| b.to_vec()).map_err(ApifyClientError::from) - })) + let url = params.apply_to_url(&self.ctx.url(None)); + get_byte_stream(&self.ctx, url).await } } diff --git a/src/error.rs b/src/error.rs index 6958f33..3ba015c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -109,7 +109,8 @@ impl ApiError { } } -/// The top-level error type for all client operations. +/// The top-level error type for all client operations: every method returns +/// `Result`, so one `match`/`if let` already catches every kind of failure. #[derive(Debug, thiserror::Error)] pub enum ApifyClientError { /// The API returned a non-success status code with a structured error body. diff --git a/src/http_client.rs b/src/http_client.rs index 433b767..1055643 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -193,6 +193,18 @@ impl HttpResponse { } } +/// A streaming counterpart of [`HttpResponse`]: the status and headers, available as soon as +/// the response starts, and the body delivered incrementally instead of as one buffered `Vec`. +pub struct StreamingHttpResponse { + /// HTTP status code. + pub status: u16, + /// Response headers. + pub headers: HashMap, + /// The body, as a stream of chunks. + pub body: + std::pin::Pin>> + Send>>, +} + /// The replaceable transport contract. /// /// Implementors are responsible only for sending a single request and returning the @@ -206,6 +218,38 @@ pub trait HttpBackend: Send + Sync + std::fmt::Debug { /// [`ApifyClientError::Http`] or [`ApifyClientError::Timeout`]. A non-2xx HTTP /// status is *not* an error at this layer — return it as a normal [`HttpResponse`]. async fn send(&self, request: HttpRequest) -> ApifyClientResult; + + /// Like [`send`](Self::send), but for a request whose body the caller wants to process + /// incrementally (a live log or a large export) instead of fully buffered. + /// + /// The default implementation just buffers the whole response via [`send`](Self::send) and + /// yields it as a single chunk: correct for a custom backend that does not override this + /// (streaming still works end-to-end, just without a memory-usage benefit), while the + /// default [`ReqwestBackend`] overrides it for real constant-memory streaming through + /// `reqwest`'s own chunked body reader. + async fn send_streaming( + &self, + request: HttpRequest, + ) -> ApifyClientResult { + let response = self.send(request).await?; + Ok(StreamingHttpResponse { + status: response.status, + headers: response.headers, + body: Box::pin(futures_util::stream::once(async move { Ok(response.body) })), + }) + } +} + +/// Maps this crate's [`HttpMethod`] to `reqwest`'s, shared by [`ReqwestBackend`]'s buffered and +/// streaming sends. +fn to_reqwest_method(method: HttpMethod) -> reqwest::Method { + match method { + HttpMethod::Get => reqwest::Method::GET, + HttpMethod::Post => reqwest::Method::POST, + HttpMethod::Put => reqwest::Method::PUT, + HttpMethod::Delete => reqwest::Method::DELETE, + HttpMethod::Head => reqwest::Method::HEAD, + } } /// The default [`HttpBackend`] implementation, backed by [`reqwest`]. @@ -239,13 +283,7 @@ impl Default for ReqwestBackend { #[async_trait] impl HttpBackend for ReqwestBackend { async fn send(&self, request: HttpRequest) -> ApifyClientResult { - let method = match request.method { - HttpMethod::Get => reqwest::Method::GET, - HttpMethod::Post => reqwest::Method::POST, - HttpMethod::Put => reqwest::Method::PUT, - HttpMethod::Delete => reqwest::Method::DELETE, - HttpMethod::Head => reqwest::Method::HEAD, - }; + let method = to_reqwest_method(request.method); let mut builder = self .client @@ -276,6 +314,45 @@ impl HttpBackend for ReqwestBackend { body, }) } + + async fn send_streaming( + &self, + request: HttpRequest, + ) -> ApifyClientResult { + let method = to_reqwest_method(request.method); + + // 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); + for (key, value) in &request.headers { + builder = builder.header(key, value); + } + if let Some(body) = request.body { + builder = builder.body(body); + } + + let response = builder.send().await?; + let status = response.status().as_u16(); + let mut headers = HashMap::new(); + for (name, value) in response.headers().iter() { + if let Ok(v) = value.to_str() { + headers.insert(name.as_str().to_string(), v.to_string()); + } + } + + let byte_stream = response.bytes_stream(); + let body = Box::pin(futures_util::StreamExt::map(byte_stream, |chunk| { + chunk.map(|b| b.to_vec()).map_err(ApifyClientError::from) + })); + Ok(StreamingHttpResponse { + status, + headers, + body, + }) + } } /// Configuration for the retry/timeout behaviour of the [`HttpClient`]. @@ -399,11 +476,122 @@ impl HttpClient { &self.user_agent } - /// Returns the token and user-agent, for endpoints (like log streaming) that must - /// open a raw connection outside the buffered backend. - pub(crate) fn stream_credentials(&self) -> (Option, String) { - (self.token.clone(), self.user_agent.clone()) + /// Sends `request` through the configured [`HttpBackend`]'s [`HttpBackend::send_streaming`], + /// with authentication and the user-agent header applied like [`call`](Self::call), 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`](Self::call) + /// would. On success, returns the body as a stream whose `.next()` must resolve within the + /// client's overall timeout budget 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. That same budget also bounds the initial wait for the + /// response to start, mirroring the reference clients' streaming `timeoutSecs`. + pub(crate) async fn call_streaming( + &self, + mut request: HttpRequest, + ) -> ApifyClientResult>>> { + request + .headers + .insert("User-Agent".to_string(), self.user_agent.clone()); + if let Some(token) = &self.token { + request + .headers + .insert("Authorization".to_string(), format!("Bearer {token}")); + } + + let method_str = request.method.as_str().to_string(); + let path = extract_path(&request.url); + // The caller-supplied `request.timeout` (the same per-endpoint base every buffered call + // uses — e.g. `DEFAULT_REQUEST_TIMEOUT`, the reference clients' "long" tier) is reused + // here as both the response-start wait and the per-chunk inactivity window: one + // configured duration, not a separate, larger one for inactivity. The JS reference client + // documents the same single-duration behavior ("it only limits inactivity"), though it + // gets there differently — via its HTTP library's socket-level idle timeout, not an + // explicit per-chunk reset like `with_inactivity_timeout` below — so this is a + // same-outcome, not same-mechanism, match. It carries the same tradeoff either way: a + // stream legitimately silent for the whole window (e.g. a long-running Actor logging + // nothing for `DEFAULT_REQUEST_TIMEOUT`) times out; the reference's escape hatch for that + // is an explicit `'noTimeout'` request option, which this crate does not yet expose. + let budget = request.timeout; + + let response = tokio::time::timeout(budget, self.backend.send_streaming(request)) + .await + .map_err(|_| ApifyClientError::Timeout)??; + + if response.status < MAX_SUCCESS_STATUS_CODE { + return Ok(with_inactivity_timeout(response.body, budget)); + } + + // 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; + let http_response = HttpResponse { + status: response.status, + headers: response.headers, + body, + }; + let api_error = build_api_error(&http_response, 1, &method_str, &path); + Err(ApifyClientError::from(api_error)) + } +} + +/// 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. +/// +/// 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 +/// has stalled, so a caller that logs the error and keeps draining (instead of stopping on the +/// first `Err`, as `?`-style consumption does) still sees the stream terminate. +fn with_inactivity_timeout( + body: std::pin::Pin>> + Send>>, + timeout: Duration, +) -> std::pin::Pin>> + Send>> { + enum State { + Active( + std::pin::Pin>> + Send>>, + ), + Stalled, } + + Box::pin(futures_util::stream::unfold( + State::Active(body), + move |state| async move { + let State::Active(mut body) = state else { + return None; + }; + match tokio::time::timeout(timeout, futures_util::StreamExt::next(&mut body)).await { + Ok(Some(item)) => Some((item, State::Active(body))), + Ok(None) => None, + Err(_) => Some((Err(ApifyClientError::Timeout), State::Stalled)), + } + }, + )) +} + +/// 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, +) -> 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 { + match chunk { + Ok(bytes) => buf.extend(bytes), + Err(_) => break, + } + } + buf } /// Compresses `request.body` in place when it is present, at least [`MIN_COMPRESS_BYTES`] long, @@ -518,7 +706,7 @@ fn gzip_compress(data: &[u8]) -> Vec { } /// Returns the path + query portion of a URL, for error reporting. -pub(crate) fn extract_path(url: &str) -> Option { +fn extract_path(url: &str) -> Option { // Find the start of the path after the scheme+host. let after_scheme = url.split_once("://").map(|(_, rest)| rest).unwrap_or(url); after_scheme @@ -540,7 +728,7 @@ fn is_error_retryable(err: &ApifyClientError) -> bool { } /// Parses the API error body (if present) into an [`ApiError`]. -pub(crate) fn build_api_error( +fn build_api_error( response: &HttpResponse, attempt: u32, method: &str, @@ -634,3 +822,51 @@ pub(crate) async fn sleep_public(duration: Duration) { async fn sleep(duration: Duration) { tokio::time::sleep(duration).await; } + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::StreamExt; + + /// After one inactivity timeout, [`with_inactivity_timeout`] must end the stream rather than + /// retrying the stalled read forever: a chunk that would only arrive after the stall must + /// never be yielded, and a third poll past the timeout error must return `None`. + #[tokio::test(start_paused = true)] + async fn with_inactivity_timeout_ends_after_one_stall() { + let timeout = Duration::from_millis(10); + let body: std::pin::Pin< + Box>> + Send>, + > = Box::pin(futures_util::stream::unfold(0u8, move |n| async move { + match n { + 0 => Some((Ok(b"chunk".to_vec()), 1)), + // Stall well past the inactivity timeout before the stream would otherwise + // continue, simulating a connection that goes quiet. + 1 => { + tokio::time::sleep(timeout * 10).await; + Some((Ok(b"should never be seen".to_vec()), 2)) + } + _ => None, + } + })); + + let mut timed = with_inactivity_timeout(body, timeout); + + let first = timed.next().await; + assert!( + matches!(&first, Some(Ok(b)) if b == b"chunk"), + "expected the first chunk, got {first:?}" + ); + + let second = timed.next().await; + assert!( + matches!(second, Some(Err(ApifyClientError::Timeout))), + "expected a timeout error after the stall, got {second:?}" + ); + + let third = timed.next().await; + assert!( + third.is_none(), + "the stream must end after the inactivity timeout instead of retrying forever, got {third:?}" + ); + } +} diff --git a/tests/actor_run.rs b/tests/actor_run.rs index d7715bb..2151490 100644 --- a/tests/actor_run.rs +++ b/tests/actor_run.rs @@ -49,6 +49,20 @@ async fn run_actor_and_read_outputs() { let log = client.run(&run.id).log().get().await.expect("get run log"); assert!(log.is_some(), "finished run should have a log"); + // Fetch the same log as raw bytes; it must decode to the same text `get()` returned. + let log_bytes = client + .run(&run.id) + .log() + .get_bytes() + .await + .expect("get run log as bytes") + .expect("finished run should have a log"); + assert_eq!( + String::from_utf8_lossy(&log_bytes), + log.expect("log should be present"), + "get_bytes() must decode to the same text as get()" + ); + // Fetch the raw log via the `raw` query parameter (LogOptions). The endpoint must accept // it and still return the log content. let raw_log = client diff --git a/tests/dataset.rs b/tests/dataset.rs index 7c81caf..1323eee 100644 --- a/tests/dataset.rs +++ b/tests/dataset.rs @@ -2,6 +2,7 @@ mod common; +use futures_util::StreamExt; use serde_json::json; /// Simple GET: listing datasets should succeed and return a paginated structure. @@ -217,3 +218,69 @@ async fn dataset_crud_flow() { "dataset should be gone after delete" ); } + +/// `download_items_stream` must yield the same bytes as the buffered `download_items`, just +/// delivered incrementally instead of all at once. +#[tokio::test(flavor = "multi_thread")] +async fn download_items_stream_matches_buffered_download() { + let client = require_client!(); + let name = common::unique_name("dataset-stream"); + 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); + dataset_client + .push_items(&json!([{ "n": 1 }, { "n": 2 }, { "n": 3 }])) + .await + .expect("push items"); + + let buffered = dataset_client + .download_items(apify_client::DownloadItemsFormat::Json, Default::default()) + .await + .expect("buffered download"); + + let mut stream = dataset_client + .download_items_stream(apify_client::DownloadItemsFormat::Json, Default::default()) + .await + .expect("stream download"); + let mut streamed = Vec::new(); + while let Some(chunk) = stream.next().await { + streamed.extend(chunk.expect("stream chunk")); + } + + assert_eq!( + streamed, buffered, + "streamed export must match the buffered export byte-for-byte" + ); +} + +/// A failed `download_items_stream` (nonexistent dataset) carries the API's structured error +/// (status/type/message), not a generic transport error — like `log().stream()` of a +/// nonexistent run, this bypasses the buffered backend path to keep the connection open, so it +/// has to parse the error body itself on a non-2xx response. +#[tokio::test(flavor = "multi_thread")] +async fn download_items_stream_of_missing_dataset_carries_structured_api_error() { + let client = require_client!(); + + let err = client + .dataset("nonexistent-dataset-id-xyz") + .download_items_stream(apify_client::DownloadItemsFormat::Json, Default::default()) + .await + .map(|_| ()) + .expect_err("streaming a nonexistent dataset's items should fail"); + + let api_error = err + .as_api_error() + .expect("error should be a structured ApiError, not a generic transport error"); + assert_eq!(api_error.status_code, 404); + assert!(api_error.is_not_found()); +} diff --git a/tests/key_value_store.rs b/tests/key_value_store.rs index e8586b8..6cc36fd 100644 --- a/tests/key_value_store.rs +++ b/tests/key_value_store.rs @@ -278,6 +278,76 @@ async fn key_value_store_crud_flow() { .is_none()); } +/// `set_record_raw_with_content_encoding` uploads an already-compressed value without the +/// client compressing it a second time, and the value round-trips without corruption. +/// +/// The payload is well above the client's own auto-compression size threshold, so this also +/// exercises that the `Content-Encoding` header (not just the small-body skip) is what the +/// client relies on to avoid double-compression; `set_record_with_content_encoding_is_not_compressed_again` +/// in `tests/unit_http.rs` pins that exact wire behaviour hermetically. What the API does with +/// the encoded upload server-side (store the compressed bytes as-is per +/// `KeyValueStoreClient::set_record_raw_with_content_encoding`'s docs, or transparently decode +/// it before storing) is outside the client's control, so this only asserts the value it reads +/// back decodes to the original payload either way, never that the bytes are corrupted or +/// double-compressed. +#[tokio::test(flavor = "multi_thread")] +async fn set_record_with_content_encoding_round_trips_without_corruption() { + use flate2::{write::GzEncoder, Compression}; + use std::io::Write; + + let client = require_client!(); + let name = common::unique_name("kvs-enc"); + let store = client + .key_value_stores() + .get_or_create(Some(&name)) + .await + .expect("create store"); + + let cleanup_client = client.clone(); + let id = store.id.clone(); + let _guard = common::Cleanup::new(move || async move { + let _ = cleanup_client.key_value_store(&id).delete().await; + }); + + let store_client = client.key_value_store(&store.id); + + let payload = serde_json::to_vec(&json!({ "data": "x".repeat(4096) })).expect("serialize"); + let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&payload).expect("gzip write"); + let gzipped = encoder.finish().expect("gzip finish"); + + store_client + .set_record_raw_with_content_encoding( + "gz-record", + gzipped.clone(), + "application/json", + "gzip", + ) + .await + .expect("set record with content encoding"); + + let record = store_client + .get_record("gz-record") + .await + .expect("get record") + .expect("record should exist"); + + // Accept either: the API served the exact gzip bytes we uploaded (decode them), or it + // stored/served the decoded value directly (already equal to `payload`). + let decoded = if record.value.starts_with(&[0x1f, 0x8b]) { + let mut decoder = flate2::read::GzDecoder::new(&record.value[..]); + let mut out = Vec::new(); + std::io::Read::read_to_end(&mut decoder, &mut out).expect("gunzip response value"); + out + } else { + record.value + }; + assert_eq!( + decoded, payload, + "the record must round-trip to the original payload without corruption" + ); +} + /// Builds a record public URL and confirms it is well-formed and fetchable without auth. /// /// The URL points at the public records endpoint; when the store exposes a URL-signing diff --git a/tests/unit_http.rs b/tests/unit_http.rs index 991dcb9..22b5052 100644 --- a/tests/unit_http.rs +++ b/tests/unit_http.rs @@ -13,6 +13,7 @@ use apify_client::{ WaitForResources, }; use async_trait::async_trait; +use futures_util::StreamExt; /// A scripted backend that returns a queued sequence of responses and counts calls. #[derive(Debug)] @@ -434,6 +435,67 @@ async fn log_get_sends_raw_query_param() { ); } +/// `log().stream()` goes through the configured `HttpBackend` — including a fully custom one, +/// not just `ReqwestBackend` — and applies the same auth/user-agent headers as a buffered call. +/// `MockBackend` does not override `send_streaming`, so this also exercises +/// `HttpBackend::send_streaming`'s default (buffer-via-`send`-and-yield-one-chunk) fallback. +#[tokio::test] +async fn log_stream_goes_through_the_configured_backend() { + let backend = MockBackend::new(vec![MockOutcome::Status(200, b"some log output".to_vec())]); + let client = client_with(backend.clone(), 0); + + let mut stream = client + .run("some-run-id") + .log() + .stream() + .await + .expect("stream should open"); + let mut collected = Vec::new(); + while let Some(chunk) = stream.next().await { + collected.extend(chunk.expect("chunk")); + } + + assert_eq!( + collected, b"some log output", + "the mock backend's response must reach the stream consumer" + ); + assert_eq!( + backend.last_header("Authorization").as_deref(), + Some("Bearer test-token"), + "a streaming request must carry the same bearer auth as a buffered one" + ); + assert!( + backend.last_header("User-Agent").is_some(), + "a streaming request must carry the User-Agent header" + ); +} + +/// A failed `log().stream()` through a custom `HttpBackend` (not just `ReqwestBackend`) still +/// surfaces the same structured `ApiError` a buffered call would, built from the mock's 404 +/// response — proving the error path also goes through the configured backend. +#[tokio::test] +async fn log_stream_through_mock_backend_surfaces_structured_error() { + let backend = MockBackend::new(vec![MockOutcome::Status( + 404, + br#"{"error":{"type":"record-not-found","message":"Run was not found"}}"#.to_vec(), + )]); + let client = client_with(backend.clone(), 0); + + let err = client + .run("missing-run-id") + .log() + .stream() + .await + .map(|_| ()) + .expect_err("streaming a 404 should fail"); + + let api_error = err + .as_api_error() + .expect("error should be a structured ApiError"); + assert_eq!(api_error.status_code, 404); + assert_eq!(api_error.error_type.as_deref(), Some("record-not-found")); +} + /// A request body at or above the 1024-byte threshold is brotli-compressed: the backend sees a /// `Content-Encoding: br` header and a body that is smaller than (and different from) the original /// and decodes back to the original bytes. @@ -593,6 +655,40 @@ async fn already_compressed_content_type_is_not_compressed() { ); } +/// `set_record_raw_with_content_encoding` sends the caller's `Content-Encoding` verbatim and +/// does not compress the body a second time, even above the auto-compression size threshold: +/// `maybe_compress_request` skips any body that already carries a `Content-Encoding` header. +#[tokio::test] +async fn set_record_with_content_encoding_is_not_compressed_again() { + let backend = MockBackend::new(vec![MockOutcome::Status(201, b"{}".to_vec())]); + let client = client_with(backend.clone(), 0); + + // Pretend this is already-gzipped data; it does not need to actually be valid gzip for this + // test, which only checks that the client leaves it untouched and sets the right header. + let pre_compressed = [0x1fu8, 0x8b, 0x08, 0x00].repeat(512); // well over the 1024-byte threshold + client + .key_value_store("me~store") + .set_record_raw_with_content_encoding( + "key", + pre_compressed.clone(), + "application/octet-stream", + "gzip", + ) + .await + .expect("ok"); + + assert_eq!( + backend.last_header("Content-Encoding").as_deref(), + Some("gzip"), + "the caller's Content-Encoding must be forwarded verbatim" + ); + assert_eq!( + backend.last_body().expect("a body was sent"), + pre_compressed, + "a pre-compressed body must be sent byte-for-byte, not compressed again" + ); +} + /// `image/bmp` sits under the already-compressed `image/` prefix but is itself raw, uncompressed /// data, so it is still compressed. Exercises the exact-media-type exception list. #[tokio::test]