From da12d17631201fc300cede68947e67291791f12f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 06:52:07 +0000 Subject: [PATCH 1/8] feat: port contentEncoding, log buffer, and dataset stream from JS client Ports apify-client-js #1140 (setRecord contentEncoding), #1141 (log.get buffer), and #1152 (downloadItems stream, superseding #1139's streamItems). No OpenAPI spec change. Factors the raw-streaming-GET logic LogClient::stream already had into a shared base::get_byte_stream helper, now reused by DatasetClient::download_items_stream. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- CHANGELOG.md | 11 ++++ Cargo.toml | 2 +- docs/misc.md | 2 + docs/storages.md | 25 +++++++++ src/clients/base.rs | 78 +++++++++++++++++++++++++++ src/clients/dataset.rs | 25 ++++++++- src/clients/key_value_store.rs | 32 ++++++++++- src/clients/log.rs | 97 ++++++++++++---------------------- src/error.rs | 8 +++ tests/actor_run.rs | 14 +++++ tests/dataset.rs | 67 +++++++++++++++++++++++ tests/key_value_store.rs | 70 ++++++++++++++++++++++++ tests/unit_http.rs | 34 ++++++++++++ 13 files changed, 399 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8daf21e..068ac32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ 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 })`. + ## [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/docs/misc.md b/docs/misc.md index 6ca5ab9..b0fcd2b 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`). | 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..7799083 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,66 @@ pub(crate) async fn put_raw( Ok(()) } +/// Opens a raw streaming `GET` connection to `url` (already fully built, including query +/// string), bypassing the buffered backend. A buffered [`HttpClient::call`] fully reads a +/// response into memory before returning it, which defeats a caller that wants to process a +/// large body (a log or a dataset export) in constant memory as it arrives — so this talks to +/// `reqwest` directly instead, applying the same bearer auth and `User-Agent` every other +/// request gets. On a non-success status it reads the body and maps it through +/// [`crate::http_client::build_api_error`], so a failed stream (e.g. a 404) carries the same +/// structured [`crate::error::ApiError`] as every buffered endpoint, rather than a generic +/// transport error. Shared by [`LogClient::stream_with_options`](crate::clients::log::LogClient::stream_with_options) +/// and [`DatasetClient::download_items_stream`](crate::clients::dataset::DatasetClient::download_items_stream), +/// which both need a live, unbuffered connection (the DRY principle). +pub(crate) async fn get_byte_stream( + ctx: &ResourceContext, + url: String, +) -> ApifyClientResult>>> { + let (token, user_agent) = ctx.http.stream_credentials(); + let client = reqwest::Client::new(); + let mut builder = client.get(&url).header("User-Agent", &user_agent); + if let Some(token) = &token { + builder = builder.header("Authorization", format!("Bearer {token}")); + } + + let response = builder + .send() + .await + .map_err(crate::error::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(crate::error::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(crate::error::ApifyClientError::from) + })) +} + /// 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..8ff70cd 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,27 @@ 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. + /// Bypasses the buffered backend 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..8dcb919 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,10 +56,32 @@ 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. @@ -101,49 +108,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(); + // Streaming needs a live connection, so this goes through `get_byte_stream` (reqwest + // directly) rather than the buffered backend path. 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..8b952ef 100644 --- a/src/error.rs +++ b/src/error.rs @@ -110,6 +110,14 @@ impl ApiError { } /// The top-level error type for all client operations. +/// +/// The reference client added a common `ApifyClientError` base class that `ApifyApiError` and +/// its status subclasses, `InvalidResponseBodyError` and `ResponseValidationError` extend — but +/// deliberately *not* argument-validation, network, timeout or abort errors — so one +/// `instanceof` check catches only the response-related ones. Rust has no exception hierarchy to +/// mirror that with, and this single enum already is the idiomatic equivalent: every operation +/// returns `Result`, so one `match`/`if let` already catches everything, +/// response-related or not, with no separate marker type needed. #[derive(Debug, thiserror::Error)] pub enum ApifyClientError { /// The API returned a non-success status code with a structured error body. 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..df5e4aa 100644 --- a/tests/unit_http.rs +++ b/tests/unit_http.rs @@ -593,6 +593,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] From aa8595ac329daa74a1a5a00765fc2b9987de277e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 07:04:32 +0000 Subject: [PATCH 2/8] fix: trim doc narration and bound stream response-start timeout Addresses review iteration 1: trims the ApifyClientError and get_byte_stream rustdoc to drop cross-client narration/caller-list padding, and wraps get_byte_stream's send() in a DEFAULT_REQUEST_TIMEOUT response-start timeout so a dead connection surfaces ApifyClientError::Timeout instead of hanging forever. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- src/clients/base.rs | 15 ++++++++++----- src/error.rs | 11 ++--------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/src/clients/base.rs b/src/clients/base.rs index 7799083..e9c7f3b 100644 --- a/src/clients/base.rs +++ b/src/clients/base.rs @@ -410,9 +410,7 @@ pub(crate) async fn put_raw_with_content_encoding( /// request gets. On a non-success status it reads the body and maps it through /// [`crate::http_client::build_api_error`], so a failed stream (e.g. a 404) carries the same /// structured [`crate::error::ApiError`] as every buffered endpoint, rather than a generic -/// transport error. Shared by [`LogClient::stream_with_options`](crate::clients::log::LogClient::stream_with_options) -/// and [`DatasetClient::download_items_stream`](crate::clients::dataset::DatasetClient::download_items_stream), -/// which both need a live, unbuffered connection (the DRY principle). +/// transport error. pub(crate) async fn get_byte_stream( ctx: &ResourceContext, url: String, @@ -424,9 +422,16 @@ pub(crate) async fn get_byte_stream( builder = builder.header("Authorization", format!("Bearer {token}")); } - let response = builder - .send() + // Bounds only the wait for the response to *start* (status/headers), matching the + // reference client's documented `stream` semantics; once a response arrives, reading the + // body through `bytes_stream()` below is unbounded here (as the reference's "it only limits + // inactivity" note implies a separate, finer-grained timeout this crate does not yet have). + // Without this, a connection that never gets a response (a dead route, a server that accepts + // but never answers) would hang forever instead of surfacing a timeout like every buffered + // endpoint does. + let response = tokio::time::timeout(DEFAULT_REQUEST_TIMEOUT, builder.send()) .await + .map_err(|_| crate::error::ApifyClientError::Timeout)? .map_err(crate::error::ApifyClientError::from)?; if !response.status().is_success() { let status = response.status().as_u16(); diff --git a/src/error.rs b/src/error.rs index 8b952ef..3ba015c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -109,15 +109,8 @@ impl ApiError { } } -/// The top-level error type for all client operations. -/// -/// The reference client added a common `ApifyClientError` base class that `ApifyApiError` and -/// its status subclasses, `InvalidResponseBodyError` and `ResponseValidationError` extend — but -/// deliberately *not* argument-validation, network, timeout or abort errors — so one -/// `instanceof` check catches only the response-related ones. Rust has no exception hierarchy to -/// mirror that with, and this single enum already is the idiomatic equivalent: every operation -/// returns `Result`, so one `match`/`if let` already catches everything, -/// response-related or not, with no separate marker type needed. +/// 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. From 7749f74bde0113801f71132b7d5ae355db0e8377 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 07:20:28 +0000 Subject: [PATCH 3/8] fix: route streaming requests through the configured HttpBackend Adds HttpBackend::send_streaming (default: buffer via send(), so existing backends keep compiling) and HttpClient::call_streaming, which applies auth/user-agent like call() and bounds the response-start wait plus per-chunk inactivity without capping total stream duration. base::get_byte_stream now delegates to it instead of opening its own unconfigured reqwest::Client, so log().stream() and download_items_stream() honor a custom or proxy/TLS-configured backend like every buffered endpoint. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- CHANGELOG.md | 12 +++ README.md | 5 +- src/clients/base.rs | 73 ++++-------------- src/http_client.rs | 184 +++++++++++++++++++++++++++++++++++++++++--- tests/unit_http.rs | 62 +++++++++++++++ 5 files changed, 266 insertions(+), 70 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 068ac32..5944509 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,18 @@ to [Semantic Versioning](https://semver.org/). (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 diff --git a/README.md b/README.md index 36ac5c3..75dc71a 100644 --- a/README.md +++ b/README.md @@ -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/src/clients/base.rs b/src/clients/base.rs index e9c7f3b..371635e 100644 --- a/src/clients/base.rs +++ b/src/clients/base.rs @@ -402,69 +402,26 @@ pub(crate) async fn put_raw_with_content_encoding( Ok(()) } -/// Opens a raw streaming `GET` connection to `url` (already fully built, including query -/// string), bypassing the buffered backend. A buffered [`HttpClient::call`] fully reads a -/// response into memory before returning it, which defeats a caller that wants to process a -/// large body (a log or a dataset export) in constant memory as it arrives — so this talks to -/// `reqwest` directly instead, applying the same bearer auth and `User-Agent` every other -/// request gets. On a non-success status it reads the body and maps it through -/// [`crate::http_client::build_api_error`], so a failed stream (e.g. a 404) carries the same -/// structured [`crate::error::ApiError`] as every buffered endpoint, rather than a generic -/// transport error. +/// 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>>> { - let (token, user_agent) = ctx.http.stream_credentials(); - let client = reqwest::Client::new(); - let mut builder = client.get(&url).header("User-Agent", &user_agent); - if let Some(token) = &token { - builder = builder.header("Authorization", format!("Bearer {token}")); - } - - // Bounds only the wait for the response to *start* (status/headers), matching the - // reference client's documented `stream` semantics; once a response arrives, reading the - // body through `bytes_stream()` below is unbounded here (as the reference's "it only limits - // inactivity" note implies a separate, finer-grained timeout this crate does not yet have). - // Without this, a connection that never gets a response (a dead route, a server that accepts - // but never answers) would hang forever instead of surfacing a timeout like every buffered - // endpoint does. - let response = tokio::time::timeout(DEFAULT_REQUEST_TIMEOUT, builder.send()) - .await - .map_err(|_| crate::error::ApifyClientError::Timeout)? - .map_err(crate::error::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, + ctx.http + .call_streaming(HttpRequest { + method: HttpMethod::Get, + url, 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(crate::error::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(crate::error::ApifyClientError::from) - })) + body: None, + timeout: DEFAULT_REQUEST_TIMEOUT, + }) + .await } /// A `POST` with a raw body that returns an enveloped resource. Used by `actor.start` diff --git a/src/http_client.rs b/src/http_client.rs index 433b767..5068551 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,96 @@ 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); + let budget = self.retry.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. +fn with_inactivity_timeout( + body: std::pin::Pin>> + Send>>, + timeout: Duration, +) -> std::pin::Pin>> + Send>> { + Box::pin(futures_util::stream::unfold( + body, + move |mut body| async move { + match tokio::time::timeout(timeout, futures_util::StreamExt::next(&mut body)).await { + Ok(Some(item)) => Some((item, body)), + Ok(None) => None, + Err(_) => Some((Err(ApifyClientError::Timeout), body)), + } + }, + )) +} + +/// 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, diff --git a/tests/unit_http.rs b/tests/unit_http.rs index df5e4aa..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. From aa3603fd1bec6bc5e62aec39350bef358d6b451b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 07:34:48 +0000 Subject: [PATCH 4/8] fix: honor request.timeout in call_streaming and end stalled streams call_streaming now uses the caller-supplied request.timeout (the per-endpoint base every buffered call also uses) instead of the client-wide retry budget, so the field base.rs populates is no longer silently ignored; the reuse as both response-start and per-chunk inactivity window is documented as matching the reference clients' single-timeoutSecs streaming semantics. with_inactivity_timeout now ends the stream after one Err(Timeout) instead of retrying the stalled read forever. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- src/http_client.rs | 82 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 77 insertions(+), 5 deletions(-) diff --git a/src/http_client.rs b/src/http_client.rs index 5068551..960d72a 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -504,7 +504,16 @@ impl HttpClient { let method_str = request.method.as_str().to_string(); let path = extract_path(&request.url); - let budget = self.retry.timeout; + // 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. This + // matches the reference clients, which apply one configured `timeoutSecs` to a streaming + // request the same way ("it only limits inactivity" is the *same* duration as the + // non-streaming timeout, not a separate, larger one) — including the tradeoff that 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 @@ -534,17 +543,32 @@ impl HttpClient { /// 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( - body, - move |mut body| async move { + 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, body)), + Ok(Some(item)) => Some((item, State::Active(body))), Ok(None) => None, - Err(_) => Some((Err(ApifyClientError::Timeout), body)), + Err(_) => Some((Err(ApifyClientError::Timeout), State::Stalled)), } }, )) @@ -796,3 +820,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:?}" + ); + } +} From 8adca896ee4af4f0e7e2bf738989d6ed381cd9fa Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 07:45:11 +0000 Subject: [PATCH 5/8] docs: fix stale streaming comments describing the removed bypass LogClient::stream_with_options and DatasetClient::download_items_stream still described streaming as going "through reqwest directly" / "bypassing the buffered backend", which was true before commit 7749f74 routed it through the configured HttpBackend. Updated both comments to describe the current behavior. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- src/clients/dataset.rs | 8 ++++---- src/clients/log.rs | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/clients/dataset.rs b/src/clients/dataset.rs index 8ff70cd..f5250c4 100644 --- a/src/clients/dataset.rs +++ b/src/clients/dataset.rs @@ -317,10 +317,10 @@ impl DatasetClient { /// /// 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. - /// Bypasses the buffered backend 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. + /// 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, diff --git a/src/clients/log.rs b/src/clients/log.rs index 8dcb919..0c7b9ee 100644 --- a/src/clients/log.rs +++ b/src/clients/log.rs @@ -108,9 +108,9 @@ impl LogClient { &self, options: LogOptions, ) -> ApifyClientResult>>> { - // Streaming needs a live connection, so this goes through `get_byte_stream` (reqwest - // directly) rather than the buffered backend path. The retry policy does not apply to - // an open stream. + // 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); From 66f3c906b18687671bd4e55aa6bcc305dc755247 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 07:56:00 +0000 Subject: [PATCH 6/8] refactor: narrow extract_path/build_api_error to module-private Both were pub(crate) but, after this branch's earlier streaming refactor consolidated their only call sites into http_client.rs itself, had no callers outside the module. Dead cross-module reach; no behavior change. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- src/http_client.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/http_client.rs b/src/http_client.rs index 960d72a..4ddf0b0 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -704,7 +704,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 @@ -726,7 +726,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, From a29b8d0c9bd8ea8b3a24019ca7f3f0ce89242dbe Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 08:09:41 +0000 Subject: [PATCH 7/8] docs: disclose the stream inactivity timeout on public API docs LogClient::stream and DatasetClient::download_items_stream promised unconditional completion/no corruption without mentioning the 6-minute per-chunk inactivity timeout call_streaming now applies, contradicting documented behavior for a quiet-but-still-running stream. Also softened the call_streaming budget-reuse comment, which claimed closer mechanism-level parity with the JS reference's axios/Node socket-level idle timeout than is accurate (same outcome, different mechanism). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- docs/misc.md | 5 ++++- src/clients/dataset.rs | 6 +++++- src/clients/log.rs | 6 ++++-- src/http_client.rs | 10 ++++++---- 4 files changed, 19 insertions(+), 8 deletions(-) diff --git a/docs/misc.md b/docs/misc.md index b0fcd2b..7db7cf0 100644 --- a/docs/misc.md +++ b/docs/misc.md @@ -166,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/src/clients/dataset.rs b/src/clients/dataset.rs index f5250c4..65423ab 100644 --- a/src/clients/dataset.rs +++ b/src/clients/dataset.rs @@ -316,7 +316,11 @@ impl DatasetClient { /// 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. + /// 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 diff --git a/src/clients/log.rs b/src/clients/log.rs index 0c7b9ee..ab7bde3 100644 --- a/src/clients/log.rs +++ b/src/clients/log.rs @@ -87,8 +87,10 @@ impl LogClient { /// 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`]. diff --git a/src/http_client.rs b/src/http_client.rs index 4ddf0b0..1055643 100644 --- a/src/http_client.rs +++ b/src/http_client.rs @@ -506,10 +506,12 @@ impl HttpClient { 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. This - // matches the reference clients, which apply one configured `timeoutSecs` to a streaming - // request the same way ("it only limits inactivity" is the *same* duration as the - // non-streaming timeout, not a separate, larger one) — including the tradeoff that a + // 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. From 8e116e33be6083b05867f2bfef0dd36abe235f2c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 08:20:15 +0000 Subject: [PATCH 8/8] docs: fix stale apify-client = \"0.6\" install version pins README.md (x2) and docs/README.md pinned an install version from long before the crate's current 0.13.0, incompatible with every feature those same pages document (streaming, get_bytes, set_record_raw_with_content_encoding, wait_for_resources, HttpBackend, ...). Bumped all three to "0.13". Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01G3FfQf9r45KVCa5VAYAjkd --- README.md | 4 ++-- docs/README.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 75dc71a..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 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"] } ```