nautilus_network/http/

client.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! HTTP client implementation with rate limiting and timeout support.

use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc, time::Duration};

use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use bytes::Bytes;
use http::{
    Method,
    header::{HeaderMap, HeaderName, HeaderValue},
};
use http_body_util::Full;
use nautilus_core::{collections::into_ustr_vec, string::secret::SecretString};
use nautilus_cryptography::providers::install_cryptographic_provider;
use url::Url;
use ustr::Ustr;

use super::{
    HttpClientError, HttpResponse, HttpResponseStream, HttpStatus,
    stream::{check_deadline, read_chunk, response_error},
};
use crate::ratelimiter::{RateLimiter, clock::MonotonicClock, quota::Quota};

/// Default maximum idle connections per host.
#[cfg(not(all(feature = "simulation", madsim)))]
const DEFAULT_POOL_MAX_IDLE_PER_HOST: usize = 32;

/// Default idle connection timeout in seconds.
#[cfg(not(all(feature = "simulation", madsim)))]
const DEFAULT_POOL_IDLE_TIMEOUT_SECS: u64 = 60;

/// Default HTTP/2 keep-alive interval in seconds.
#[cfg(not(all(feature = "simulation", madsim)))]
const DEFAULT_HTTP2_KEEP_ALIVE_SECS: u64 = 30;

/// Default HTTP/2 stream flow-control window in bytes (16 MiB).
#[cfg(not(all(feature = "simulation", madsim)))]
const DEFAULT_HTTP2_STREAM_WINDOW_BYTES: u32 = 16 * 1024 * 1024;

/// Default HTTP/2 connection flow-control window in bytes (32 MiB).
#[cfg(not(all(feature = "simulation", madsim)))]
const DEFAULT_HTTP2_CONNECTION_WINDOW_BYTES: u32 = 32 * 1024 * 1024;

/// Environment variable that enables adaptive HTTP/2 flow-control windows when set to `true`.
#[cfg(not(all(feature = "simulation", madsim)))]
const NAUTILUS_HTTP2_ADAPTIVE_WINDOW: &str = "NAUTILUS_HTTP2_ADAPTIVE_WINDOW";

/// Default maximum HTTP response body size in bytes (100 MiB).
///
/// Bounds peak memory per response so a hostile or malfunctioning endpoint
/// cannot exhaust memory by streaming an arbitrarily large body. Mirrors the
/// caps already enforced on the WebSocket and raw-socket paths.
const DEFAULT_MAX_RESPONSE_BYTES: usize = 100 * 1024 * 1024;

const GZIP: &str = "gzip";
const GZIP_DECODE_CHUNK_BYTES: usize = 64 * 1024;

#[cfg(all(feature = "simulation", madsim))]
pub(super) const REQUEST_TIMEOUT_MESSAGE: &str = "simulated request deadline elapsed";
#[cfg(not(all(feature = "simulation", madsim)))]
pub(super) const REQUEST_TIMEOUT_MESSAGE: &str = "request deadline elapsed";

/// Controls whether an HTTP client follows redirects.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum HttpRedirectPolicy {
    /// Follow up to ten redirects.
    #[default]
    Follow,
    /// Reject every redirect response.
    Reject,
}

/// An asynchronous HTTP client with rate limiting, timeouts, and custom headers.
///
/// The client uses Hyper for normal I/O and supports default and per-key quotas. Multiple
/// clients can share the same rate limiter when their requests consume one quota budget.
/// With `simulation` and `cfg(madsim)`, plaintext HTTP/1.1 uses simulated byte streams;
/// HTTPS, explicit proxies, and redirect following are unsupported.
///
/// Request URLs in transport errors and response-stream diagnostics are limited to their origin
/// (scheme, host, and port). Userinfo, paths, queries, and fragments are excluded. Set
/// `redact_url(true)` on the builder to omit the origin too. This policy does not change requests
/// or diagnostic messages from lower layers. It controls the URL context added by this client.
#[derive(Clone, Debug)]
pub struct HttpClient {
    pub(crate) client: InnerHttpClient,
    pub(crate) rate_limiters: Arc<[Arc<RateLimiter<Ustr, MonotonicClock>>]>,
}

#[bon::bon]
impl HttpClient {
    /// Returns a builder for a new [`HttpClient`] instance.
    ///
    /// Set `rate_limiters` to share quota state across clients. When omitted, the client creates
    /// one rate limiter from `default_quota` and `keyed_quotas`. An explicit empty vector disables
    /// rate limiting. Each request awaits every configured limiter with the same keys. A limiter
    /// without a default quota ignores keys it does not own, allowing independent scopes such as
    /// per-IP and per-account limits to apply to one request.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Shared rate limiters are combined with quota configuration.
    /// - The proxy URL is malformed.
    /// - `NAUTILUS_HTTP2_ADAPTIVE_WINDOW` is set to a value other than `true` or `false`.
    /// - Building the underlying HTTP transport fails.
    #[allow(
        clippy::needless_pass_by_value,
        reason = "owned proxy URLs are part of the public builder API"
    )]
    #[builder(finish_fn = build)]
    pub fn builder(
        #[builder(default)] headers: HashMap<String, String>,
        #[builder(default)] header_keys: Vec<String>,
        #[builder(default)] keyed_quotas: Vec<(String, Quota)>,
        default_quota: Option<Quota>,
        timeout_secs: Option<u64>,
        proxy_url: Option<String>,
        rate_limiters: Option<Vec<Arc<RateLimiter<Ustr, MonotonicClock>>>>,
        #[builder(default)] redirect_policy: HttpRedirectPolicy,
        /// Whether to honor ambient proxy configuration when `proxy_url` is not set.
        #[builder(default = true)]
        use_system_proxy: bool,
        /// Whether to omit the request origin added by this client to transport errors and
        /// response-stream diagnostics.
        /// Defaults to `false`.
        /// Userinfo, paths, queries, and fragments are always excluded from the added URL context.
        /// Diagnostic messages from lower layers are unchanged. TLS certificate errors and
        /// dependency logs can still include the hostname.
        #[builder(default)]
        redact_url: bool,
        /// Maximum size in bytes of a buffered response body after `Content-Encoding` decoding.
        /// Defaults to the response body limit (100 MiB). Raise it for endpoints whose compressed
        /// payloads legitimately decode beyond that limit; the wire body stays bounded by the
        /// response body limit either way.
        #[builder(default = DEFAULT_MAX_RESPONSE_BYTES)]
        max_decompressed_bytes: usize,
    ) -> Result<Self, HttpClientError> {
        let rate_limiters = if let Some(rate_limiters) = rate_limiters {
            if default_quota.is_some() || !keyed_quotas.is_empty() {
                return Err(HttpClientError::Error(
                    "Cannot combine shared rate limiters with quota configuration".to_string(),
                ));
            }

            rate_limiters
        } else {
            let keyed_quotas = keyed_quotas
                .into_iter()
                .map(|(key, quota)| (Ustr::from(&key), quota))
                .collect();
            vec![Arc::new(RateLimiter::new_with_quota(
                default_quota,
                keyed_quotas,
            ))]
        };

        Self::build(
            headers,
            header_keys,
            timeout_secs,
            proxy_url.as_deref(),
            rate_limiters,
            redirect_policy,
            use_system_proxy,
            redact_url,
            max_decompressed_bytes,
        )
    }

    #[expect(
        clippy::too_many_arguments,
        reason = "client construction consumes the builder settings"
    )]
    fn build(
        headers: HashMap<String, String>,
        header_keys: Vec<String>,
        timeout_secs: Option<u64>,
        proxy_url: Option<&str>,
        rate_limiters: Vec<Arc<RateLimiter<Ustr, MonotonicClock>>>,
        redirect_policy: HttpRedirectPolicy,
        use_system_proxy: bool,
        redact_url: bool,
        max_decompressed_bytes: usize,
    ) -> Result<Self, HttpClientError> {
        install_cryptographic_provider();

        let mut header_map = HeaderMap::new();

        for (key, value) in headers {
            let header_name = HeaderName::from_str(&key)
                .map_err(|e| HttpClientError::Error(format!("Invalid header name '{key}': {e}")))?;

            let header_value = HeaderValue::from_str(&value).map_err(|e| {
                HttpClientError::Error(format!("Invalid header value for '{key}': {e}"))
            })?;

            header_map.insert(header_name, header_value);
        }

        #[cfg(all(feature = "simulation", madsim))]
        let simulation = super::simulation::Client::new(redirect_policy, proxy_url)?;

        #[cfg(not(all(feature = "simulation", madsim)))]
        let client = super::transport::Client::new(
            proxy_url,
            use_system_proxy,
            header_map.get(http::header::USER_AGENT).cloned(),
            super::transport::Settings {
                pool_max_idle_per_host: DEFAULT_POOL_MAX_IDLE_PER_HOST,
                pool_idle_timeout: Duration::from_secs(DEFAULT_POOL_IDLE_TIMEOUT_SECS),
                keep_alive_interval: Some(Duration::from_secs(DEFAULT_HTTP2_KEEP_ALIVE_SECS)),
                stream_window: Some(DEFAULT_HTTP2_STREAM_WINDOW_BYTES),
                connection_window: Some(DEFAULT_HTTP2_CONNECTION_WINDOW_BYTES),
                adaptive_window: http2_adaptive_window()?,
            },
        )?;

        #[cfg(all(feature = "simulation", madsim))]
        let _ = use_system_proxy;

        // Pre-intern header keys as HeaderName. An invalid key is an error: a silent drop would
        // make response extraction read nothing.
        let response_headers = header_keys
            .into_iter()
            .map(|key| match HeaderName::from_str(&key) {
                Ok(name) => Ok((key, name)),
                Err(e) => Err(HttpClientError::Error(format!(
                    "Invalid header key '{key}': {e}"
                ))),
            })
            .collect::<Result<Vec<_>, _>>()?;

        let client = InnerHttpClient {
            #[cfg(not(all(feature = "simulation", madsim)))]
            client,
            headers: header_map,
            timeout: timeout_secs.map(Duration::from_secs),
            redact_url,
            #[cfg(not(all(feature = "simulation", madsim)))]
            redirect_policy,
            #[cfg(all(feature = "simulation", madsim))]
            simulation,
            response_headers: Arc::from(response_headers),
            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
            max_decompressed_bytes,
        };

        Ok(Self {
            client,
            rate_limiters: rate_limiters.into(),
        })
    }

    /// Sends an HTTP request.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    ///
    /// # Examples
    ///
    /// If requesting `/foo/bar`, pass rate-limit keys `["foo/bar", "foo"]`.
    #[expect(clippy::too_many_arguments)]
    pub async fn request(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        let keys = keys.map(into_ustr_vec);

        self.request_with_ustr_keys(method, url, params, headers, body, timeout_secs, keys)
            .await
    }

    /// Sends an HTTP request whose body contains secret material.
    ///
    /// The body retains its zeroizing owner until the transport releases the last byte buffer.
    /// Transport, TLS, and operating-system layers may make additional plaintext copies that this
    /// client cannot zeroize.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send the request or if it times out.
    #[expect(clippy::too_many_arguments)]
    pub async fn request_with_secret_body(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: SecretString,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        let keys = keys.map(into_ustr_vec);
        self.await_rate_limits(keys.as_deref()).await;

        self.client
            .send_request_with_secret_body(method, url, params, headers, body, timeout_secs)
            .await
    }

    /// Sends an HTTP request with serializable query parameters.
    ///
    /// This method accepts any type implementing `Serialize` for query parameters,
    /// which are URL-encoded directly into the query string without an intermediate `HashMap`.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    #[expect(clippy::too_many_arguments)]
    pub async fn request_with_params<P: serde::Serialize>(
        &self,
        method: Method,
        url: String,
        params: Option<&P>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        let keys = keys.map(into_ustr_vec);
        self.await_rate_limits(keys.as_deref()).await;

        self.client
            .send_request_with_query(method, url, params, headers, body, timeout_secs)
            .await
    }

    /// Sends a GET request and returns its response body as a stream.
    ///
    /// Returns raw body chunks without content decoding or adding `Accept-Encoding`.
    /// Applies default headers and the client timeout. No rate-limit keys are supplied, so no
    /// quota is consumed. One absolute deadline covers response headers and the whole body,
    /// including time spent processing chunks. Streaming has no total body size limit; callers
    /// must process or discard each chunk without accumulating an unbounded body.
    /// Dropping the response releases the unfinished exchange, including its simulated driver.
    ///
    /// # Errors
    ///
    /// Returns an error if request preparation, connection, or response headers fail or time out.
    pub async fn get_stream(&self, url: String) -> Result<HttpResponseStream, HttpClientError> {
        self.await_rate_limits(None).await;
        self.client
            .send_stream_internal::<[(String, String); 0]>(
                Method::GET,
                &url,
                None,
                None,
                None,
                None,
            )
            .await
    }

    /// Sends an HTTP request using pre-interned rate limiter keys.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send the request or the request times out.
    #[expect(clippy::too_many_arguments)]
    pub async fn request_with_ustr_keys(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<Ustr>>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.await_rate_limits(keys.as_deref()).await;

        self.client
            .send_request(method, url, params, headers, body, timeout_secs)
            .await
    }

    pub(crate) async fn await_rate_limits(&self, keys: Option<&[Ustr]>) {
        RateLimiter::await_limiters_ready(&self.rate_limiters, keys).await;
    }

    /// Sends an HTTP GET request.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn get(
        &self,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.request(Method::GET, url, params, headers, None, timeout_secs, keys)
            .await
    }

    /// Sends an HTTP POST request.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn post(
        &self,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.request(Method::POST, url, params, headers, body, timeout_secs, keys)
            .await
    }

    /// Sends an HTTP PATCH request.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn patch(
        &self,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.request(
            Method::PATCH,
            url,
            params,
            headers,
            body,
            timeout_secs,
            keys,
        )
        .await
    }

    /// Sends an HTTP DELETE request.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn delete(
        &self,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        timeout_secs: Option<u64>,
        keys: Option<Vec<String>>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.request(
            Method::DELETE,
            url,
            params,
            headers,
            None,
            timeout_secs,
            keys,
        )
        .await
    }
}

/// Internal implementation backing [`HttpClient`].
///
/// The underlying Hyper client reuses pooled connections and is cheap to clone. Responses
/// retain only configured header fields, and bodies larger than `max_response_bytes` are rejected.
/// Decoded (decompressed) bodies are bounded separately by `max_decompressed_bytes`.
#[derive(Clone, Debug)]
pub struct InnerHttpClient {
    #[cfg(all(feature = "simulation", madsim))]
    simulation: super::simulation::Client,
    #[cfg(not(all(feature = "simulation", madsim)))]
    client: super::transport::Client,
    headers: HeaderMap,
    timeout: Option<Duration>,
    redact_url: bool,
    #[cfg(not(all(feature = "simulation", madsim)))]
    redirect_policy: HttpRedirectPolicy,
    pub(crate) response_headers: Arc<[(String, HeaderName)]>,
    pub(crate) max_response_bytes: usize,
    pub(crate) max_decompressed_bytes: usize,
}

impl InnerHttpClient {
    /// Sends an HTTP request and returns an [`HttpResponse`].
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn send_request(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.send_request_with_body(
            method,
            url,
            params,
            headers,
            body.map(RequestBody::Plain),
            timeout_secs,
        )
        .await
    }

    async fn send_request_with_secret_body(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: SecretString,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.send_request_with_body(
            method,
            url,
            params,
            headers,
            Some(RequestBody::Secret(body)),
            timeout_secs,
        )
        .await
    }

    async fn send_request_with_body(
        &self,
        method: Method,
        url: String,
        params: Option<&HashMap<String, Vec<String>>>,
        headers: Option<HashMap<String, String>>,
        body: Option<RequestBody>,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponse, HttpClientError> {
        let full_url = encode_url_params(&url, params)?;
        self.send_request_internal(
            method,
            full_url.as_ref(),
            None::<&()>,
            headers,
            body,
            timeout_secs,
        )
        .await
    }

    /// Sends an HTTP request with URL-encoded serializable query parameters.
    ///
    /// This method accepts any type implementing `Serialize` for query parameters,
    /// avoiding `HashMap` conversion overhead.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    pub async fn send_request_with_query<Q: serde::Serialize>(
        &self,
        method: Method,
        url: String,
        query: Option<&Q>,
        headers: Option<HashMap<String, String>>,
        body: Option<Vec<u8>>,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponse, HttpClientError> {
        self.send_request_internal(
            method,
            &url,
            query,
            headers,
            body.map(RequestBody::Plain),
            timeout_secs,
        )
        .await
    }

    /// Internal implementation for sending HTTP requests.
    ///
    /// # Errors
    ///
    /// Returns an error if unable to send request or times out.
    async fn send_request_internal<Q: serde::Serialize>(
        &self,
        method: Method,
        url: &str,
        query: Option<&Q>,
        headers: Option<HashMap<String, String>>,
        body: Option<RequestBody>,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponse, HttpClientError> {
        let mut headers = headers.unwrap_or_default();
        let has_accept_encoding = headers
            .keys()
            .any(|key| key.eq_ignore_ascii_case(http::header::ACCEPT_ENCODING.as_str()))
            || self.headers.contains_key(http::header::ACCEPT_ENCODING);

        if !has_accept_encoding {
            headers.insert(http::header::ACCEPT_ENCODING.to_string(), GZIP.to_string());
        }

        let is_head = method == Method::HEAD;
        let stream = self
            .send_stream_internal(method, url, query, Some(headers), body, timeout_secs)
            .await?;
        let result = self
            .consume_response(stream.response, stream.deadline, is_head)
            .await;
        result.map_err(|e| response_error(e, stream.origin.as_deref()))
    }

    async fn send_stream_internal<Q: serde::Serialize>(
        &self,
        method: Method,
        url: &str,
        query: Option<&Q>,
        headers: Option<HashMap<String, String>>,
        body: Option<RequestBody>,
        timeout_secs: Option<u64>,
    ) -> Result<HttpResponseStream, HttpClientError> {
        let mut url =
            Url::parse(url).map_err(|e| HttpClientError::from(format!("URL parse error: {e}")))?;
        if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
            return Err(HttpClientError::Error(
                "unsupported HTTP URL scheme or hostname".into(),
            ));
        }

        let mut header_map = self.headers.clone();

        if let Ok(username) = percent_encoding::percent_decode_str(url.username()).decode_utf8() {
            let password = url.password().and_then(|password| {
                percent_encoding::percent_decode_str(password)
                    .decode_utf8()
                    .ok()
            });

            if !username.is_empty() || password.is_some() {
                let mut value = HeaderValue::from_str(&format!(
                    "Basic {}",
                    BASE64.encode(format!(
                        "{username}:{}",
                        password.as_deref().unwrap_or_default()
                    ))
                ))
                .map_err(|e| HttpClientError::Error(e.to_string()))?;
                value.set_sensitive(true);
                header_map.insert(http::header::AUTHORIZATION, value);
                let _ = url.set_username("");
                let _ = url.set_password(None);
            }
        }

        let extra_header_count = headers.as_ref().map_or(0, HashMap::len);

        if let Some(headers) = headers {
            for (key, value) in headers {
                let key = HeaderName::from_bytes(key.as_bytes())
                    .map_err(|e| HttpClientError::from(format!("Invalid header name: {e}")))?;
                let value = HeaderValue::from_str(&value)
                    .map_err(|e| HttpClientError::from(format!("Invalid header value: {e}")))?;

                if header_map.insert(key.clone(), value).is_some() {
                    log::trace!("Replaced duplicate request header '{key}'");
                }
            }
        }

        if let Some(query) = query {
            {
                let mut pairs = url.query_pairs_mut();
                let serializer = serde_urlencoded::Serializer::new(&mut pairs);
                query
                    .serialize(serializer)
                    .map_err(|e| HttpClientError::Error(e.to_string()))?;
            }

            if url.query() == Some("") {
                url.set_query(None);
            }
        }

        if !header_map.contains_key(http::header::ACCEPT) {
            header_map.insert(http::header::ACCEPT, HeaderValue::from_static("*/*"));
        }

        let body = body.map(RequestBody::into_bytes).unwrap_or_default();
        let body_len = body.len();
        let query_len = url.query().map_or(0, str::len);
        let mut request = http::Request::new(Full::new(body));
        *request.method_mut() = method;
        *request.uri_mut() = url[..url::Position::AfterQuery]
            .parse()
            .map_err(|_| HttpClientError::Error("invalid HTTP request target".into()))?;
        *request.headers_mut() = header_map;
        log::trace!(
            "Sending HTTP request: method={} extra_headers={extra_header_count} \
             query_bytes={query_len} body_bytes={body_len}",
            request.method(),
        );

        let origin = (!self.redact_url).then(|| url.origin().ascii_serialization());

        let duration = timeout_secs.map(Duration::from_secs).or(self.timeout);
        let deadline = duration.map(|duration| crate::dst::time::Instant::now() + duration);

        let operation = async {
            #[cfg(all(feature = "simulation", madsim))]
            let (response, connection) = self.simulation.send(request, &url).await?;
            #[cfg(not(all(feature = "simulation", madsim)))]
            let response = self.client.send(request, self.redirect_policy).await?;
            Ok(HttpResponseStream {
                response,
                deadline,
                origin: origin.clone(),
                #[cfg(all(feature = "simulation", madsim))]
                _connection: connection,
            })
        };

        let result = match deadline {
            Some(deadline) => tokio::select! {
                biased;
                () = crate::dst::time::sleep_until(deadline) => Err(HttpClientError::TimeoutError(REQUEST_TIMEOUT_MESSAGE.into())),
                result = operation => result,
            },
            None => operation.await,
        };

        result.map_err(|e| response_error(e, origin.as_deref()))
    }

    async fn consume_response<B>(
        &self,
        response: http::Response<B>,
        deadline: Option<crate::dst::time::Instant>,
        is_head: bool,
    ) -> Result<HttpResponse, HttpClientError>
    where
        B: http_body::Body<Data = Bytes> + Unpin,
        B::Error: std::error::Error + 'static,
    {
        let (parts, mut body) = response.into_parts();
        let mut headers =
            HashMap::with_capacity(self.response_headers.len().min(parts.headers.len()));
        for (key, name) in self.response_headers.iter() {
            if let Some(value) = parts
                .headers
                .get(name)
                .and_then(|value| value.to_str().ok())
            {
                headers.insert(key.clone(), value.to_owned());
            }
        }

        let max = self.max_response_bytes;

        if let Some(len) = body.size_hint().exact()
            && len > max as u64
        {
            return Err(HttpClientError::Error(format!(
                "HTTP response body of {len} bytes exceeds maximum of {max} bytes",
            )));
        }

        let mut buf = bytes::BytesMut::new();
        while let Some(chunk) = read_chunk(&mut body, deadline).await? {
            if chunk.len() > max - buf.len() {
                return Err(HttpClientError::Error(format!(
                    "HTTP response body exceeds maximum of {max} bytes",
                )));
            }

            buf.extend_from_slice(&chunk);
        }

        log::trace!(
            "Received HTTP response: status={} headers={} body_bytes={}",
            parts.status,
            parts.headers.len(),
            buf.len()
        );

        let bodyless = is_head
            || parts.status.is_informational()
            || matches!(
                parts.status,
                http::StatusCode::NO_CONTENT
                    | http::StatusCode::RESET_CONTENT
                    | http::StatusCode::NOT_MODIFIED
            );

        let (body, decoded) = if bodyless {
            (buf.freeze(), false)
        } else {
            decode_content_encoding(&parts.headers, buf, self.max_decompressed_bytes, deadline)
                .await?
        };

        if decoded {
            headers.retain(|key, _| {
                !key.eq_ignore_ascii_case("content-encoding")
                    && !key.eq_ignore_ascii_case("content-length")
            });
        }

        Ok(HttpResponse {
            status: HttpStatus::new(parts.status),
            headers,
            body,
        })
    }
}

enum RequestBody {
    Plain(Vec<u8>),
    Secret(SecretString),
}

async fn decode_content_encoding(
    headers: &HeaderMap,
    body: bytes::BytesMut,
    max: usize,
    deadline: Option<crate::dst::time::Instant>,
) -> Result<(Bytes, bool), HttpClientError> {
    let mut encodings = headers.get_all(http::header::CONTENT_ENCODING).iter();

    let Some(encoding) = encodings.next() else {
        return Ok((body.freeze(), false));
    };

    if encodings.next().is_some() {
        return Err(HttpClientError::Error(
            "unsupported multiple HTTP content-encoding fields".into(),
        ));
    }

    let encoding = encoding
        .to_str()
        .map_err(|_| HttpClientError::Error("invalid HTTP content-encoding value".into()))?;
    let encoding = encoding.trim();
    if encoding.is_empty() || encoding.eq_ignore_ascii_case("identity") {
        return Ok((body.freeze(), false));
    }

    if !encoding.eq_ignore_ascii_case(GZIP) {
        return Err(HttpClientError::Error(format!(
            "unsupported HTTP content-encoding '{encoding}'"
        )));
    }

    check_deadline(deadline)?;
    let body = body.freeze();
    let mut out = Vec::new();

    let mut reader = std::io::Read::take(
        flate2::bufread::MultiGzDecoder::new(GzipInput {
            bytes: body.as_ref(),
            remaining: GZIP_DECODE_CHUNK_BYTES,
        }),
        (max as u64).saturating_add(1),
    );

    loop {
        check_deadline(deadline)?;
        let mut chunk = std::io::Read::take(&mut reader, GZIP_DECODE_CHUNK_BYTES as u64);
        let result = std::io::Read::read_to_end(&mut chunk, &mut out);
        check_deadline(deadline)?;

        let complete = match result {
            Ok(count) => count < GZIP_DECODE_CHUNK_BYTES,
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => false,
            Err(e) => {
                return Err(HttpClientError::Error(format!(
                    "failed to decode gzip response body: {e}"
                )));
            }
        };

        if out.len() > max {
            return Err(HttpClientError::Error(format!(
                "decompressed HTTP response body exceeds maximum of {max} bytes"
            )));
        }

        if complete {
            break;
        }

        crate::dst::task::yield_now().await;
        reader.get_mut().get_mut().remaining = GZIP_DECODE_CHUNK_BYTES;
    }

    Ok((Bytes::from(out), true))
}

struct GzipInput<'a> {
    bytes: &'a [u8],
    remaining: usize,
}

impl std::io::Read for GzipInput<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        let mut available = std::io::BufRead::fill_buf(self)?;
        let count = std::io::Read::read(&mut available, buf)?;
        std::io::BufRead::consume(self, count);
        Ok(count)
    }
}

impl std::io::BufRead for GzipInput<'_> {
    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
        if self.bytes.is_empty() {
            return Ok(self.bytes);
        }

        if self.remaining == 0 {
            return Err(std::io::ErrorKind::WouldBlock.into());
        }

        Ok(&self.bytes[..self.bytes.len().min(self.remaining)])
    }

    fn consume(&mut self, count: usize) {
        self.bytes = &self.bytes[count..];
        self.remaining -= count;
    }
}

impl RequestBody {
    fn into_bytes(self) -> Bytes {
        match self {
            Self::Plain(body) => body.into(),
            Self::Secret(body) => Bytes::from_owner(SecretBody(body)),
        }
    }
}

struct SecretBody(SecretString);

impl AsRef<[u8]> for SecretBody {
    fn as_ref(&self) -> &[u8] {
        self.0.expose_secret().as_bytes()
    }
}

#[cfg(not(all(feature = "simulation", madsim)))]
fn http2_adaptive_window() -> Result<bool, HttpClientError> {
    let Some(value) = std::env::var_os(NAUTILUS_HTTP2_ADAPTIVE_WINDOW) else {
        return Ok(false);
    };

    value.to_str().and_then(|v| v.parse().ok()).ok_or_else(|| {
        HttpClientError::ClientBuildError(format!(
            "{NAUTILUS_HTTP2_ADAPTIVE_WINDOW} must be 'true' or 'false', was '{}'",
            value.display()
        ))
    })
}

/// Encodes URL parameters into the query string.
///
/// Returns `Cow::Borrowed` when no parameters need appending (zero-alloc fast path).
/// Parameters can have multiple values per key (for doseq=True behavior).
/// Preserves existing query strings in the URL by appending with '&' instead of '?'.
/// The query is inserted before any fragment, which is preserved unchanged.
fn encode_url_params<'a>(
    url: &'a str,
    params: Option<&HashMap<String, Vec<String>>>,
) -> Result<Cow<'a, str>, HttpClientError> {
    let Some(params) = params else {
        return Ok(Cow::Borrowed(url));
    };

    let pairs: Vec<(&str, &str)> = params
        .iter()
        .flat_map(|(key, values)| {
            values
                .iter()
                .map(move |value| (key.as_str(), value.as_str()))
        })
        .collect();

    if pairs.is_empty() {
        return Ok(Cow::Borrowed(url));
    }

    let query_string = serde_urlencoded::to_string(pairs)
        .map_err(|e| HttpClientError::Error(format!("Failed to encode params: {e}")))?;

    // The first literal '#' starts the fragment per RFC 3986 section 3.5.
    // A data '#' in an earlier component must be percent-encoded as "%23".
    let (base, fragment) = match url.split_once('#') {
        Some((base, fragment)) => (base, Some(fragment)),
        None => (url, None),
    };

    let separator = if base.contains('?') { '&' } else { '?' };

    Ok(Cow::Owned(match fragment {
        Some(fragment) => format!("{base}{separator}{query_string}#{fragment}"),
        None => format!("{base}{separator}{query_string}"),
    }))
}

#[cfg(test)]
mod encode_url_params_tests {
    use std::{borrow::Cow, collections::HashMap};

    use rstest::rstest;

    use super::encode_url_params;

    fn params(pairs: &[(&str, &str)]) -> HashMap<String, Vec<String>> {
        let mut map: HashMap<String, Vec<String>> = HashMap::new();
        for (key, value) in pairs {
            map.entry((*key).to_string())
                .or_default()
                .push((*value).to_string());
        }

        map
    }

    #[rstest]
    #[case("https://x/y", "https://x/y?a=b")]
    #[case("https://x/y?old=1", "https://x/y?old=1&a=b")]
    #[case("https://x/y#frag", "https://x/y?a=b#frag")]
    #[case("https://x/y?old=1#frag", "https://x/y?old=1&a=b#frag")]
    #[case(
        "https://x/y#section?display=full",
        "https://x/y?a=b#section?display=full"
    )]
    #[case("https://x/y#", "https://x/y?a=b#")]
    fn test_query_is_inserted_before_the_fragment(#[case] url: &str, #[case] expected: &str) {
        let params = params(&[("a", "b")]);

        assert_eq!(encode_url_params(url, Some(&params)).unwrap(), expected);
    }

    #[rstest]
    fn test_url_is_borrowed_when_no_params_are_supplied() {
        assert!(matches!(
            encode_url_params("https://x/y#frag", None).unwrap(),
            Cow::Borrowed("https://x/y#frag")
        ));
    }

    #[rstest]
    fn test_url_is_borrowed_when_params_are_empty() {
        let params = HashMap::new();

        assert!(matches!(
            encode_url_params("https://x/y#frag", Some(&params)).unwrap(),
            Cow::Borrowed("https://x/y#frag")
        ));
    }
}

#[cfg(test)]
#[cfg(target_os = "linux")] // Only run network tests on Linux (CI stability)
#[cfg(not(all(feature = "simulation", madsim)))]
mod tests {
    use std::net::SocketAddr;

    use axum::{
        Router,
        body::to_bytes,
        extract::Request,
        response::IntoResponse,
        routing::{any, delete, get, patch, post},
        serve,
    };
    use http::status::StatusCode;
    use log::Level;
    use rstest::rstest;
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt},
        sync::oneshot,
    };

    use super::*;
    use crate::logging::tests::capture_logs;

    #[rstest]
    #[case("ftp://127.0.0.1:1/resource")]
    #[case("file:///resource")]
    #[tokio::test]
    async fn test_request_rejects_unsupported_url(#[case] url: &str) {
        let client = HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap();

        let error = client
            .request(Method::GET, url.to_string(), None, None, None, None, None)
            .await
            .unwrap_err();

        assert!(
            matches!(error, HttpClientError::Error(ref message) if message == "unsupported HTTP URL scheme or hostname")
        );
    }

    #[rstest]
    #[case::username_only("user%2F17@", "Basic dXNlci8xNzo=")]
    #[case::password_only(":secret%2F29@", "Basic OnNlY3JldC8yOQ==")]
    #[tokio::test]
    async fn test_request_url_credentials_allow_missing_username_or_password(
        #[case] userinfo: &str,
        #[case] expected: &str,
    ) {
        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap();

        let response = client
            .request(
                Method::GET,
                format!("http://{userinfo}{addr}/headers"),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(
            response.body.as_ref(),
            format!("{expected}\n*/*").as_bytes()
        );
    }

    #[rstest]
    #[case::default(None, "*/*")]
    #[case::explicit(Some("application/octet-stream"), "application/octet-stream")]
    #[tokio::test]
    async fn test_request_accept_header_preserves_explicit_value(
        #[case] accept: Option<&str>,
        #[case] expected: &str,
    ) {
        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap();
        let headers =
            accept.map(|value| HashMap::from([("accept".to_string(), value.to_string())]));

        let response = client
            .request(
                Method::GET,
                format!("http://{addr}/headers"),
                None,
                headers,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(
            response.body.as_ref(),
            format!("absent\n{expected}").as_bytes()
        );
    }

    async fn capture_request(request: Request) -> impl IntoResponse {
        let (parts, body) = request.into_parts();
        let body = to_bytes(body, usize::MAX).await.unwrap();
        let default_header = parts.headers.get("x-default").unwrap().to_str().unwrap();
        let request_header = parts.headers.get("x-request").unwrap().to_str().unwrap();
        let query = parts.uri.query().unwrap_or_default();
        let body = String::from_utf8(body.to_vec()).unwrap();
        let capture = format!(
            "{}\n{}\n{query}\n{default_header}\n{request_header}\n{body}",
            parts.method,
            parts.uri.path(),
        );

        ([("x-response-id", "response-42")], capture)
    }

    fn create_router() -> Router {
        Router::new()
            .route("/get", get(|| async { "hello-world!" }))
            .route("/post", post(|body: Bytes| async move { body }))
            .route("/patch", patch(|body: Bytes| async move { body }))
            .route("/delete", delete(|| async { StatusCode::OK }))
            .route("/capture", any(capture_request))
            .route(
                "/headers",
                get(|request: Request| async move {
                    let headers = request.headers();
                    format!(
                        "{}\n{}",
                        headers
                            .get(http::header::AUTHORIZATION)
                            .map_or("absent", |v| v.to_str().unwrap()),
                        headers
                            .get(http::header::ACCEPT)
                            .map_or("absent", |v| v.to_str().unwrap()),
                    )
                }),
            )
            .route("/notfound", get(|| async { StatusCode::NOT_FOUND }))
            .route(
                "/redirect",
                get(|| async { (StatusCode::TEMPORARY_REDIRECT, [("location", "/get")]) }),
            )
            .route(
                "/slow",
                get(|| async {
                    tokio::time::sleep(Duration::from_secs(2)).await;
                    "Eventually responded"
                }),
            )
            .route(
                "/large",
                // Returns a 1 MiB body to exercise the response size cap.
                get(|| async { "x".repeat(1024 * 1024) }),
            )
    }

    async fn start_test_server() -> Result<SocketAddr, Box<dyn std::error::Error + Send + Sync>> {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            serve(listener, create_router()).await.unwrap();
        });

        Ok(addr)
    }

    async fn spawn_connection_dropper() -> (SocketAddr, tokio::task::JoinHandle<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let task = tokio::spawn(async move {
            loop {
                let (stream, _) = listener.accept().await.unwrap();
                drop(stream);
            }
        });

        (addr, task)
    }

    async fn spawn_chunked_response_server() -> (SocketAddr, tokio::task::JoinHandle<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let task = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut request = Vec::new();
            let mut chunk = [0u8; 1024];
            loop {
                let read = stream.read(&mut chunk).await.unwrap();
                if read == 0 {
                    break;
                }

                request.extend_from_slice(&chunk[..read]);
                if request.windows(4).any(|window| window == b"\r\n\r\n") {
                    break;
                }
            }

            stream
                .write_all(
                    b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n\
                      5\r\nfirst\r\n6\r\nsecond\r\n0\r\n\r\n",
                )
                .await
                .unwrap();
        });

        (addr, task)
    }

    async fn spawn_rejecting_connect_proxy() -> (SocketAddr, oneshot::Receiver<String>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (request_tx, request_rx) = oneshot::channel();

        tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut request = Vec::new();
            let mut chunk = [0u8; 1024];
            loop {
                let read = stream.read(&mut chunk).await.unwrap();
                if read == 0 {
                    break;
                }

                request.extend_from_slice(&chunk[..read]);
                if request.windows(4).any(|window| window == b"\r\n\r\n") {
                    break;
                }
            }

            request_tx
                .send(String::from_utf8(request).unwrap())
                .unwrap();
            stream
                .write_all(
                    b"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n",
                )
                .await
                .unwrap();
        });

        (addr, request_rx)
    }

    fn inner_client() -> InnerHttpClient {
        HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap()
            .client
    }

    #[tokio::test(start_paused = true)]
    async fn test_body_ready_at_deadline_is_rejected() {
        let client = inner_client();
        let response = http::Response::new(Full::new(Bytes::from_static(b"ready")));
        let result = client
            .consume_response(response, Some(crate::dst::time::Instant::now()), false)
            .await;
        assert!(
            matches!(result, Err(HttpClientError::TimeoutError(message)) if message == REQUEST_TIMEOUT_MESSAGE)
        );
    }

    #[tokio::test]
    async fn test_get() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        let client = inner_client();
        let response = client
            .send_request(Method::GET, format!("{url}/get"), None, None, None, None)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"hello-world!");
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_preserves_wire_semantics_and_extracts_response_headers(
        #[values(false, true)] redact_url: bool,
    ) {
        let addr = start_test_server().await.unwrap();
        let mut default_headers = HashMap::new();
        default_headers.insert("x-default".to_string(), "default-a".to_string());
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .headers(default_headers)
            .header_keys(vec!["x-response-id".to_string()])
            .build()
            .unwrap();
        let mut params = HashMap::new();
        params.insert(
            "tag".to_string(),
            vec!["A B".to_string(), "C/D".to_string()],
        );
        let mut request_headers = HashMap::new();
        request_headers.insert("x-request".to_string(), "request-b".to_string());

        let response = client
            .request(
                Method::PUT,
                format!("http://{addr}/capture?existing=seed"),
                Some(&params),
                Some(request_headers),
                Some(b"payload-c".to_vec()),
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(
            response.headers,
            HashMap::from([("x-response-id".to_string(), "response-42".to_string())])
        );
        assert_eq!(
            response.body.as_ref(),
            b"PUT\n/capture\nexisting=seed&tag=A+B&tag=C%2FD\ndefault-a\nrequest-b\npayload-c"
        );
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_with_secret_body_preserves_wire_body(
        #[values(false, true)] redact_url: bool,
    ) {
        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .headers(HashMap::from([(
                "x-default".to_string(),
                "default-secret".to_string(),
            )]))
            .build()
            .unwrap();
        let headers = HashMap::from([("x-request".to_string(), "request-secret".to_string())]);

        let response = client
            .request_with_secret_body(
                Method::POST,
                format!("http://{addr}/capture"),
                None,
                Some(headers),
                SecretString::from("credential-body"),
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(
            response.body.as_ref(),
            b"POST\n/capture\n\ndefault-secret\nrequest-secret\ncredential-body"
        );
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_with_params_serializes_query_fields(
        #[values(false, true)] redact_url: bool,
    ) {
        #[derive(serde::Serialize)]
        struct Query<'a> {
            symbol: &'a str,
            limit: u32,
        }

        let addr = start_test_server().await.unwrap();
        let mut default_headers = HashMap::new();
        default_headers.insert("x-default".to_string(), "default-d".to_string());
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .headers(default_headers)
            .header_keys(vec!["x-response-id".to_string()])
            .build()
            .unwrap();
        let mut request_headers = HashMap::new();
        request_headers.insert("x-request".to_string(), "request-e".to_string());

        let params = Query {
            symbol: "BTC/USDT",
            limit: 37,
        };

        let response = client
            .request_with_params(
                Method::GET,
                format!("http://{addr}/capture"),
                Some(&params),
                Some(request_headers),
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(
            response.headers,
            HashMap::from([("x-response-id".to_string(), "response-42".to_string())])
        );
        assert_eq!(
            response.body.as_ref(),
            b"GET\n/capture\nsymbol=BTC%2FUSDT&limit=37\ndefault-d\nrequest-e\n"
        );
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_with_params_preserves_secret_query_fields(
        #[values(false, true)] redact_url: bool,
    ) {
        #[derive(serde::Serialize)]
        struct Query<'a> {
            auth: &'a str,
            market_id: i16,
        }

        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .headers(HashMap::from([(
                "x-default".to_string(),
                "default-f".to_string(),
            )]))
            .build()
            .unwrap();
        let headers = HashMap::from([("x-request".to_string(), "request-g".to_string())]);

        let params = Query {
            auth: "token/42",
            market_id: 7,
        };

        let response = client
            .request_with_params(
                Method::GET,
                format!("http://{addr}/capture"),
                Some(&params),
                Some(headers),
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(
            response.body.as_ref(),
            b"GET\n/capture\nauth=token%2F42&market_id=7\ndefault-f\nrequest-g\n"
        );
    }

    #[rstest]
    #[case::empty_at_zero_cap(b"", 0)]
    #[case::at_cap(b"body-37", 7)]
    #[case::below_cap(b"body-37", 8)]
    #[tokio::test]
    async fn test_declared_response_body_at_or_below_cap_is_returned(
        #[case] bytes: &'static [u8],
        #[case] max_response_bytes: usize,
    ) {
        let client = InnerHttpClient {
            max_response_bytes,
            ..inner_client()
        };

        let response = http::Response::new(Full::new(Bytes::from_static(bytes)));

        let response = client
            .consume_response(response, None, false)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), bytes);
    }

    #[tokio::test]
    async fn test_response_body_within_cap_is_returned() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        // Cap above the 1 MiB payload: body should be returned intact.
        let client = InnerHttpClient {
            max_response_bytes: 4 * 1024 * 1024,
            ..inner_client()
        };

        let response = client
            .send_request(Method::GET, format!("{url}/large"), None, None, None, None)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), vec![b'x'; 1024 * 1024]);
    }

    #[tokio::test]
    async fn test_response_body_exceeding_cap_is_rejected() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        // Cap below the 1 MiB payload: the request must fail rather than buffer it.
        let client = InnerHttpClient {
            max_response_bytes: 16 * 1024,
            ..inner_client()
        };

        let result = client
            .send_request(Method::GET, format!("{url}/large"), None, None, None, None)
            .await;

        let err = result.expect_err("oversized response body should be rejected");

        let HttpClientError::Error(message) = err else {
            panic!("expected HTTP error, was {err:?}");
        };

        assert_eq!(
            message,
            "HTTP response body of 1048576 bytes exceeds maximum of 16384 bytes"
        );
    }

    #[rstest]
    #[case::at_cap(11)]
    #[case::below_cap(12)]
    #[tokio::test]
    async fn test_chunked_response_body_at_or_below_cap_is_returned(
        #[case] max_response_bytes: usize,
    ) {
        let (addr, server_task) = spawn_chunked_response_server().await;

        let client = InnerHttpClient {
            max_response_bytes,
            ..inner_client()
        };

        let response = client
            .send_request(
                Method::GET,
                format!("http://{addr}"),
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        server_task.await.unwrap();

        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"firstsecond");
    }

    #[tokio::test]
    async fn test_chunked_response_body_exceeding_cap_is_rejected() {
        let (addr, server_task) = spawn_chunked_response_server().await;
        let max_response_bytes = 8;

        let client = InnerHttpClient {
            max_response_bytes,
            ..inner_client()
        };

        let error = client
            .send_request(
                Method::GET,
                format!("http://{addr}"),
                None,
                None,
                None,
                None,
            )
            .await
            .expect_err("chunked response body should be rejected");
        server_task.await.unwrap();

        let HttpClientError::Error(message) = error else {
            panic!("expected HTTP error, was {error:?}");
        };

        assert_eq!(
            message,
            format!("HTTP response body exceeds maximum of {max_response_bytes} bytes")
        );
    }

    #[tokio::test]
    async fn test_post() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        let client = inner_client();
        let response = client
            .send_request(Method::POST, format!("{url}/post"), None, None, None, None)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"");
    }

    #[tokio::test]
    async fn test_post_with_body() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        let client = inner_client();

        let mut body = HashMap::new();
        body.insert(
            "key1".to_string(),
            serde_json::Value::String("value1".to_string()),
        );
        body.insert(
            "key2".to_string(),
            serde_json::Value::String("value2".to_string()),
        );

        let body_string = serde_json::to_string(&body).unwrap();
        let body_bytes = body_string.into_bytes();

        let response = client
            .send_request(
                Method::POST,
                format!("{url}/post"),
                None,
                None,
                Some(body_bytes.clone()),
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), body_bytes);
    }

    #[tokio::test]
    async fn test_patch() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        let client = inner_client();
        let response = client
            .send_request(
                Method::PATCH,
                format!("{url}/patch"),
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"");
    }

    #[tokio::test]
    async fn test_delete() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}");

        let client = inner_client();
        let response = client
            .send_request(
                Method::DELETE,
                format!("{url}/delete"),
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"");
    }

    #[tokio::test]
    async fn test_not_found() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/notfound");
        let client = inner_client();

        let response = client
            .send_request(Method::GET, url, None, None, None, None)
            .await
            .unwrap();

        assert!(response.status.is_client_error());
        assert_eq!(response.status.as_u16(), 404);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"");
    }

    #[tokio::test]
    async fn test_timeout() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/slow");
        let client = inner_client();

        // We'll set a 1-second timeout for a route that sleeps 2 seconds
        let result = client
            .send_request(Method::GET, url, None, None, None, Some(1))
            .await;

        assert!(
            matches!(&result, Err(HttpClientError::TimeoutError(_))),
            "Expected a timeout error, was: {result:?}"
        );
    }

    #[rstest]
    fn test_http_client_without_proxy() {
        // Create client with no proxy
        let result = HttpClient::builder().build();

        assert!(result.is_ok());
    }

    #[rstest]
    fn test_http_client_builder_preserves_empty_rate_limiters() {
        let client = HttpClient::builder()
            .rate_limiters(Vec::new())
            .build()
            .unwrap();

        assert!(client.rate_limiters.is_empty());
    }

    #[rstest]
    fn test_http_client_builder_rejects_shared_rate_limiters_with_quotas() {
        let quota = Quota::with_period(Duration::from_secs(1)).unwrap();
        let rate_limiter = Arc::new(RateLimiter::new_with_quota(None, Vec::new()));
        let result = HttpClient::builder()
            .default_quota(quota)
            .rate_limiters(vec![rate_limiter])
            .build();

        assert_eq!(
            result.unwrap_err().to_string(),
            "HTTP error occurred: Cannot combine shared rate limiters with quota configuration"
        );
    }

    #[tokio::test]
    async fn test_http_client_without_proxy_requests_directly() {
        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder().timeout_secs(2).build().unwrap();
        let response = client
            .request(
                Method::GET,
                format!("http://{addr}/get"),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .expect("direct request");

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.body.as_ref(), b"hello-world!");
    }

    #[tokio::test]
    async fn test_http_client_redirect_policy() {
        let addr = start_test_server().await.unwrap();
        let follow = HttpClient::builder().timeout_secs(2).build().unwrap();
        let reject = HttpClient::builder()
            .timeout_secs(2)
            .redirect_policy(HttpRedirectPolicy::Reject)
            .build()
            .unwrap();

        let followed = follow
            .request(
                Method::GET,
                format!("http://{addr}/redirect"),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        let rejected = reject
            .request(
                Method::GET,
                format!("http://{addr}/redirect"),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        assert_eq!(followed.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(followed.body.as_ref(), b"hello-world!");
        assert_eq!(
            rejected.status.as_u16(),
            StatusCode::TEMPORARY_REDIRECT.as_u16()
        );
        assert!(rejected.body.is_empty());
    }

    #[rstest]
    #[tokio::test]
    async fn test_http_client_url_policy_preserves_response(
        #[values(false, true)] redact_url: bool,
    ) {
        let addr = start_test_server().await.unwrap();
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .timeout_secs(2)
            .build()
            .unwrap();
        let response = client
            .request(
                Method::GET,
                format!("http://{addr}/get"),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .expect("direct request with URL redaction");

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.body.as_ref(), b"hello-world!");
    }

    #[rstest]
    #[case::request("request")]
    #[case::preinterned_keys("keys")]
    #[case::get("get")]
    #[case::post("post")]
    #[case::patch("patch")]
    #[case::delete("delete")]
    #[case::stream("stream")]
    #[tokio::test]
    async fn test_http_client_url_policy_redacts_transport_errors(
        #[case] request: &str,
        #[values(false, true)] redact_url: bool,
    ) {
        const USERINFO_SECRET: &str = "transport-userinfo-secret";
        const PATH_SECRET: &str = "transport-path-secret";
        const QUERY_SECRET: &str = "transport-query-secret";
        let (addr, drop_task) = spawn_connection_dropper().await;
        let url = format!(
            "http://rpc-user:{USERINFO_SECRET}@{addr}/{PATH_SECRET}?api_key={QUERY_SECRET}"
        );
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .timeout_secs(1)
            .build()
            .unwrap();

        let error = match request {
            "request" => client
                .request(Method::GET, url.clone(), None, None, None, None, None)
                .await
                .unwrap_err(),
            "keys" => client
                .request_with_ustr_keys(
                    Method::GET,
                    url.clone(),
                    None,
                    None,
                    None,
                    None,
                    Some(vec![Ustr::from("policy")]),
                )
                .await
                .unwrap_err(),
            "get" => client
                .get(url.clone(), None, None, None, None)
                .await
                .unwrap_err(),
            "post" => client
                .post(url.clone(), None, None, None, None, None)
                .await
                .unwrap_err(),
            "patch" => client
                .patch(url.clone(), None, None, None, None, None)
                .await
                .unwrap_err(),
            "delete" => client
                .delete(url.clone(), None, None, None, None)
                .await
                .unwrap_err(),
            "stream" => client.get_stream(url.clone()).await.unwrap_err(),
            _ => panic!("unknown request shape"),
        };

        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());
        assert!(
            matches!(error, HttpClientError::TransportError(_)),
            "{error:?}"
        );

        for rendered in [error.to_string(), format!("{error:?}")] {
            assert!(!rendered.contains(USERINFO_SECRET));
            assert!(!rendered.contains(PATH_SECRET));
            assert!(!rendered.contains(QUERY_SECRET));
            assert!(!rendered.contains(&url));
            assert_eq!(
                rendered.contains(&format!("http://{addr}")),
                !redact_url,
                "{rendered}"
            );
            assert_eq!(rendered.contains("for origin ("), !redact_url, "{rendered}");
        }
    }

    #[tokio::test]
    async fn test_request_with_params_redact_url_removes_origin_from_error() {
        const QUERY_SECRET: &str = "transport-query-secret";
        #[derive(serde::Serialize)]
        struct Query<'a> {
            auth: &'a str,
        }

        let (addr, drop_task) = spawn_connection_dropper().await;
        let url = format!("http://{addr}/trades");
        let params = Query { auth: QUERY_SECRET };
        let client = HttpClient::builder()
            .redact_url(true)
            .timeout_secs(1)
            .build()
            .unwrap();

        let error = client
            .request_with_params(Method::GET, url, Some(&params), None, None, None, None)
            .await
            .expect_err("a dropped connection should fail");
        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());

        for rendered in [error.to_string(), format!("{error:?}")] {
            assert!(!rendered.contains(&addr.to_string()));
            assert!(!rendered.contains("auth="));
            assert!(!rendered.contains(QUERY_SECRET));
        }
    }

    #[tokio::test]
    async fn test_request_removes_query_string_from_transport_error_by_default() {
        const QUERY_SECRET: &str = "default-query-secret";
        const FRAGMENT_MARKER: &str = "default-fragment-marker";
        let (addr, drop_task) = spawn_connection_dropper().await;
        let url = format!("http://{addr}/trades?api_key={QUERY_SECRET}#{FRAGMENT_MARKER}");
        let client = HttpClient::builder().timeout_secs(1).build().unwrap();

        let error = client
            .request(Method::GET, url, None, None, None, None, None)
            .await
            .expect_err("a dropped connection should fail");
        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());

        for rendered in [error.to_string(), format!("{error:?}")] {
            assert!(
                rendered.contains(&format!("for origin (http://{addr})")),
                "default error omitted the origin: {rendered}"
            );
            assert!(!rendered.contains("/trades"), "{rendered}");
            assert!(!rendered.contains("api_key="), "{rendered}");
            assert!(!rendered.contains(QUERY_SECRET), "{rendered}");
            assert!(!rendered.contains(FRAGMENT_MARKER), "{rendered}");
        }
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_with_secret_body_url_policy_redacts_transport_error(
        #[values(false, true)] redact_url: bool,
    ) {
        const QUERY_SECRET: &str = "secret-body-query-secret";
        let (addr, drop_task) = spawn_connection_dropper().await;
        let url = format!("http://{addr}/trades?api_key={QUERY_SECRET}");
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .timeout_secs(1)
            .build()
            .unwrap();

        let error = client
            .request_with_secret_body(
                Method::POST,
                url,
                None,
                None,
                SecretString::from("credential-body"),
                None,
                None,
            )
            .await
            .expect_err("a dropped connection should fail");
        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());

        for rendered in [error.to_string(), format!("{error:?}")] {
            assert_eq!(
                rendered.contains(&format!("for origin (http://{addr})")),
                !redact_url,
                "{rendered}"
            );
            assert_eq!(
                rendered.contains(&addr.to_string()),
                !redact_url,
                "{rendered}"
            );
            assert!(!rendered.contains("/trades"), "{rendered}");
            assert!(!rendered.contains("credential-body"), "{rendered}");
            assert!(!rendered.contains("api_key="), "{rendered}");
            assert!(!rendered.contains(QUERY_SECRET), "{rendered}");
        }
    }

    #[tokio::test]
    async fn test_request_with_params_removes_query_from_transport_error_by_default() {
        const QUERY_SECRET: &str = "default-params-query-secret";
        #[derive(serde::Serialize)]
        struct Query<'a> {
            auth: &'a str,
        }

        let (addr, drop_task) = spawn_connection_dropper().await;
        let url = format!("http://{addr}/trades");
        let params = Query { auth: QUERY_SECRET };
        let client = HttpClient::builder().timeout_secs(1).build().unwrap();

        let error = client
            .request_with_params(Method::GET, url, Some(&params), None, None, None, None)
            .await
            .expect_err("a dropped connection should fail");
        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());

        for rendered in [error.to_string(), format!("{error:?}")] {
            assert!(
                rendered.contains(&format!("for origin (http://{addr})")),
                "default error omitted the origin: {rendered}"
            );
            assert!(!rendered.contains("/trades"), "{rendered}");
            assert!(!rendered.contains("auth="), "{rendered}");
            assert!(!rendered.contains(QUERY_SECRET), "{rendered}");
        }
    }

    #[rstest]
    #[tokio::test]
    async fn test_http_client_url_policy_removes_endpoint_from_trace_logs(
        #[values(false, true)] redact_url: bool,
    ) {
        const USERINFO_SECRET: &str = "trace-userinfo-secret";
        const PATH_SECRET: &str = "trace-path-secret";
        const QUERY_SECRET: &str = "trace-query-secret";
        let capture = capture_logs().await;
        let addr = start_test_server().await.unwrap();
        let url = format!(
            "http://rpc-user:{USERINFO_SECRET}@{addr}/{PATH_SECRET}?api_key={QUERY_SECRET}"
        );
        let client = HttpClient::builder()
            .redact_url(redact_url)
            .timeout_secs(2)
            .build()
            .unwrap();

        let response = client
            .request(Method::GET, url.clone(), None, None, None, None, None)
            .await
            .expect("credentialized endpoint should return an HTTP response");
        let messages = capture.messages();

        assert_eq!(response.status.as_u16(), StatusCode::NOT_FOUND.as_u16());
        assert!(messages.iter().any(|(level, message)| {
            *level == Level::Trace && message.starts_with("Sending HTTP request: method=GET")
        }));
        assert!(messages.iter().any(|(level, message)| {
            *level == Level::Trace
                && message.starts_with("Received HTTP response: status=404 Not Found")
        }));

        for (_, message) in messages {
            assert!(!message.contains(USERINFO_SECRET));
            assert!(!message.contains(PATH_SECRET));
            assert!(!message.contains(QUERY_SECRET));
            assert!(!message.contains(&url));
        }
    }

    #[rstest]
    #[case::without_user_agent(None)]
    #[case::with_user_agent(Some("NautilusTrader/proxy-test-73"))]
    #[tokio::test]
    async fn test_http_client_uses_connect_and_proxy_authorization_for_https(
        #[case] user_agent: Option<&str>,
    ) {
        const USERNAME: &str = "proxytest";
        const PASSWORD: &str = "fixture42";
        let mut headers =
            HashMap::from([("authorization".into(), "Bearer origin-secret-19".into())]);

        if let Some(user_agent) = user_agent {
            headers.insert("user-agent".into(), user_agent.into());
        }

        let (proxy_addr, request_rx) = spawn_rejecting_connect_proxy().await;
        let client = HttpClient::builder()
            .headers(headers)
            .timeout_secs(2)
            .proxy_url(format!("http://{USERNAME}:{PASSWORD}@{proxy_addr}"))
            .build()
            .unwrap();
        let error = client
            .request(
                Method::GET,
                "https://fixture.example.test/path".to_string(),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .expect_err("proxy should reject CONNECT");
        let request = request_rx.await.expect("captured CONNECT request");
        let mut lines = request.split("\r\n");
        let request_line = lines.next().expect("CONNECT request line");

        let headers: HashMap<_, _> = lines
            .filter_map(|line| {
                let (name, value) = line.split_once(':')?;
                Some((name.to_ascii_lowercase(), value.trim()))
            })
            .collect();

        let expected_auth = format!("Basic {}", BASE64.encode(format!("{USERNAME}:{PASSWORD}")));
        let mut expected_headers = HashMap::from([
            ("host".into(), "fixture.example.test:443"),
            ("proxy-authorization".into(), expected_auth.as_str()),
        ]);

        if let Some(user_agent) = user_agent {
            expected_headers.insert("user-agent".into(), user_agent);
        }

        assert_eq!(request_line, "CONNECT fixture.example.test:443 HTTP/1.1");
        assert_eq!(headers, expected_headers);
        assert!(!error.to_string().contains(PASSWORD));
        assert!(!error.to_string().contains(&BASE64.encode(PASSWORD)));
        assert!(!error.to_string().contains(&expected_auth));
    }

    #[rstest]
    #[case::cross_origin("http://127.0.0.1:2/final", None, false)]
    #[case::cross_origin_override("http://127.0.0.1:2/final", Some("Basic b3ZlcnJpZGU="), false)]
    #[case::same_origin("http://127.0.0.1:1/final", None, true)]
    #[case::same_origin_override("http://127.0.0.1:1/final", Some("Basic b3ZlcnJpZGU="), true)]
    #[case::https_connect("https://127.0.0.1:2/final", None, false)]
    #[tokio::test]
    async fn test_proxy_authentication_per_redirect_hop(
        #[case] destination: &'static str,
        #[case] initial_auth: Option<&'static str>,
        #[case] same_origin: bool,
    ) {
        const PROXY_AUTH: &str = "Basic cHJveHl0ZXN0OmZpeHR1cmU0Mg==";

        let expected_initial_auth = initial_auth.unwrap_or(PROXY_AUTH);
        let expected_redirect_auth = if same_origin {
            expected_initial_auth
        } else {
            PROXY_AUTH
        };

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (request_tx, mut request_rx) = tokio::sync::mpsc::channel(2);
        let router = Router::new().fallback(any(move |request: Request| {
            let request_tx = request_tx.clone();

            async move {
                let (parts, _) = request.into_parts();
                let first = parts.uri.path() == "/start";

                let expected_auth = if first {
                    expected_initial_auth
                } else {
                    expected_redirect_auth
                };

                let authenticated = parts.headers.get(http::header::PROXY_AUTHORIZATION)
                    == Some(&http::HeaderValue::from_static(expected_auth));
                let connect = parts.method == http::Method::CONNECT;
                request_tx.send(parts).await.unwrap();

                if !authenticated || connect {
                    return StatusCode::PROXY_AUTHENTICATION_REQUIRED.into_response();
                }

                if first {
                    return (StatusCode::TEMPORARY_REDIRECT, [("location", destination)])
                        .into_response();
                }

                "accepted".into_response()
            }
        }));

        let server = tokio::spawn(async move { serve(listener, router).await.unwrap() });
        let origin_headers = HashMap::from([
            ("authorization".to_string(), "Bearer origin-19".to_string()),
            ("cookie".to_string(), "session=origin-23".to_string()),
            ("cookie2".to_string(), "legacy=origin-29".to_string()),
            (
                "www-authenticate".to_string(),
                "Basic realm=origin-31".to_string(),
            ),
        ]);
        let mut headers = origin_headers.clone();
        if let Some(initial_auth) = initial_auth {
            headers.insert("proxy-authorization".to_string(), initial_auth.to_string());
        }

        let client = HttpClient::builder()
            .headers(headers)
            .timeout_secs(2)
            .use_system_proxy(false)
            .proxy_url(format!("http://proxytest:fixture42@{addr}"))
            .build()
            .unwrap();
        let result = client
            .request(
                Method::GET,
                "http://127.0.0.1:1/start".to_string(),
                None,
                None,
                None,
                None,
                None,
            )
            .await;
        let first = tokio::time::timeout(Duration::from_secs(2), request_rx.recv())
            .await
            .unwrap()
            .unwrap();
        let redirected = tokio::time::timeout(Duration::from_secs(2), request_rx.recv())
            .await
            .unwrap()
            .unwrap();
        server.abort();
        let connect = destination.starts_with("https:");

        assert_eq!(first.method, http::Method::GET);
        assert_eq!(first.uri.to_string(), "http://127.0.0.1:1/start");
        assert_eq!(
            first.headers[http::header::PROXY_AUTHORIZATION],
            expected_initial_auth
        );
        assert_eq!(
            redirected.headers[http::header::PROXY_AUTHORIZATION],
            expected_redirect_auth
        );

        for (name, value) in origin_headers {
            assert_eq!(first.headers.get(&name).unwrap(), value.as_str());
            assert_eq!(
                redirected
                    .headers
                    .get(&name)
                    .map(|value| value.to_str().unwrap()),
                same_origin.then_some(value.as_str())
            );
        }

        if connect {
            assert_eq!(redirected.method, http::Method::CONNECT);
            assert_eq!(redirected.uri.to_string(), "127.0.0.1:2");
            assert!(result.is_err());
        } else {
            let response = result.unwrap();
            assert_eq!(redirected.method, http::Method::GET);
            assert_eq!(redirected.uri.to_string(), destination);
            assert_eq!(response.status.as_u16(), 200);
            assert_eq!(response.body.as_ref(), b"accepted");
        }
    }

    #[tokio::test]
    async fn test_http_client_unreachable_proxy_error_redacts_credentials() {
        const USERNAME: &str = "proxy-user";
        const SECRET: &str = "unreachable-proxy-secret";
        let (proxy_addr, drop_task) = spawn_connection_dropper().await;
        let client = HttpClient::builder()
            .timeout_secs(1)
            .proxy_url(format!("http://{USERNAME}:{SECRET}@{proxy_addr}"))
            .build()
            .unwrap();
        let error = client
            .request(
                Method::GET,
                "https://fixture.example.test/".to_string(),
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .expect_err("unreachable proxy should fail");
        drop_task.abort();
        let task_error = drop_task
            .await
            .expect_err("connection dropper should be cancelled");

        assert!(task_error.is_cancelled());
        assert!(!error.to_string().contains(SECRET));
        assert!(!error.to_string().contains(&BASE64.encode(SECRET)));
        assert!(
            !error
                .to_string()
                .contains(&BASE64.encode(format!("{USERNAME}:{SECRET}")))
        );
    }

    #[rstest]
    fn test_http_client_with_valid_proxy() {
        // Create client with a valid proxy URL
        let result = HttpClient::builder()
            .proxy_url("http://proxy.example.com:8080".to_string())
            .build();

        assert!(result.is_ok());
    }

    #[rstest]
    fn test_http_client_with_socks5_proxy() {
        // Create client with a SOCKS5 proxy URL
        let result = HttpClient::builder()
            .proxy_url("socks5://127.0.0.1:1080".to_string())
            .build();

        assert!(result.is_ok());
    }

    #[rstest]
    fn test_http_client_with_malformed_proxy() {
        // Proxy parsing accepts scheme-less hostnames.
        // It only fails on obviously malformed URLs like "://invalid" or "http://".
        // More subtle issues (like "not-a-valid-url") are caught when connecting.
        let result = HttpClient::builder()
            .proxy_url("://invalid".to_string())
            .build();

        assert!(result.is_err());
        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
    }

    #[rstest]
    fn test_http_client_invalid_proxy_error_redacts_credentials() {
        const SECRET: &str = "unique-proxy-secret";
        let result = HttpClient::builder()
            .proxy_url(format!("http://proxytest:{SECRET}@[::1"))
            .build();
        let error = result.expect_err("malformed proxy URL should fail");

        assert_eq!(
            error.to_string(),
            "Invalid proxy URL: proxy URL is malformed"
        );
        assert!(!error.to_string().contains(SECRET));
    }

    #[rstest]
    fn test_http_client_with_empty_proxy_string() {
        // Create client with an empty proxy URL string
        let result = HttpClient::builder().proxy_url(String::new()).build();

        assert!(result.is_err());
        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
    }

    #[tokio::test]
    async fn test_http_client_get() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/get");

        let client = HttpClient::builder().build().unwrap();
        let response = client.get(url, None, None, None, None).await.unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"hello-world!");
    }

    #[tokio::test]
    async fn test_http_client_post() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/post");

        let client = HttpClient::builder().build().unwrap();
        let response = client
            .post(url, None, None, Some(b"post-body-73".to_vec()), None, None)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"post-body-73");
    }

    #[tokio::test]
    async fn test_http_client_patch() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/patch");

        let client = HttpClient::builder().build().unwrap();
        let response = client
            .patch(url, None, None, Some(b"patch-body-91".to_vec()), None, None)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"patch-body-91");
    }

    #[tokio::test]
    async fn test_http_client_delete() {
        let addr = start_test_server().await.unwrap();
        let url = format!("http://{addr}/delete");

        let client = HttpClient::builder().build().unwrap();
        let response = client.delete(url, None, None, None, None).await.unwrap();

        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"");
    }
}

#[cfg(test)]
mod rate_limit_tests {
    use std::{num::NonZeroU32, sync::Arc, time::Duration};

    #[cfg(all(feature = "simulation", madsim))]
    use madsim::task as test_task;
    #[cfg(not(all(feature = "simulation", madsim)))]
    use tokio::task as test_task;
    use ustr::Ustr;

    use super::HttpClient;
    use crate::ratelimiter::{RateLimiter, quota::Quota};

    #[tokio::test]
    async fn test_http_client_awaits_multiple_rate_limiters() {
        let quota = Quota::per_minute(NonZeroU32::MIN);
        let request_key = Ustr::from("scope:request");
        let order_key = Ustr::from("scope:order");

        let request_limiter = Arc::new(RateLimiter::new_with_quota(
            None,
            vec![(request_key, quota)],
        ));
        let order_limiter = Arc::new(RateLimiter::new_with_quota(None, vec![(order_key, quota)]));
        let client = HttpClient::builder()
            .rate_limiters(vec![
                Arc::clone(&request_limiter),
                Arc::clone(&order_limiter),
            ])
            .build()
            .unwrap();

        client
            .await_rate_limits(Some(&[request_key, order_key]))
            .await;

        assert!(request_limiter.check_key(&request_key).is_err());
        assert!(order_limiter.check_key(&order_key).is_err());
    }

    #[cfg_attr(
        not(all(feature = "simulation", madsim)),
        tokio::test(start_paused = true)
    )]
    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
    async fn test_http_client_reserves_multiple_rate_limits_together() {
        let global_key = Ustr::from("scope:global");
        let order_key = Ustr::from("scope:order");

        let global_limiter = Arc::new(RateLimiter::new_with_quota(
            None,
            vec![(
                global_key,
                Quota::with_period(Duration::from_secs(1)).unwrap(),
            )],
        ));

        let order_limiter = Arc::new(RateLimiter::new_with_quota(
            None,
            vec![(
                order_key,
                Quota::with_period(Duration::from_secs(10)).unwrap(),
            )],
        ));
        order_limiter.check_key(&order_key).unwrap();

        let client = HttpClient::builder()
            .rate_limiters(vec![
                Arc::clone(&global_limiter),
                Arc::clone(&order_limiter),
            ])
            .build()
            .unwrap();

        let request = test_task::spawn(async move {
            client
                .await_rate_limits(Some(&[global_key, order_key]))
                .await;
        });

        test_task::yield_now().await;

        global_limiter.check_key(&global_key).unwrap();
        assert!(!request.is_finished());

        advance_test_clock(Duration::from_millis(9_999)).await;
        global_limiter.until_key_ready(&global_key).await;
        advance_test_clock(Duration::from_millis(1)).await;
        test_task::yield_now().await;
        assert!(!request.is_finished());

        advance_test_clock(Duration::from_millis(998)).await;
        test_task::yield_now().await;
        assert!(!request.is_finished());

        advance_test_clock(Duration::from_millis(1)).await;
        request.await.unwrap();

        assert!(global_limiter.check_key(&global_key).is_err());
        assert!(order_limiter.check_key(&order_key).is_err());
    }

    #[cfg(all(feature = "simulation", madsim))]
    async fn advance_test_clock(duration: Duration) {
        madsim::time::advance(duration);
        test_task::yield_now().await;
    }

    #[cfg(not(all(feature = "simulation", madsim)))]
    async fn advance_test_clock(duration: Duration) {
        tokio::time::advance(duration).await;
    }
}

#[cfg(test)]
pub(super) mod content_encoding_tests {
    use std::{collections::HashMap, io::Write, time::Duration};

    use bytes::Bytes;
    use flate2::{Compression, write::GzEncoder};
    use http_body_util::Full;
    use rstest::rstest;

    use super::{HttpClient, HttpClientError, REQUEST_TIMEOUT_MESSAGE};
    use crate::dst::time;

    #[rstest]
    #[case::large_body(false, false)]
    #[case::large_body_deadline(true, false)]
    #[case::empty_members(false, true)]
    #[case::empty_members_deadline(true, true)]
    #[cfg_attr(
        not(all(feature = "simulation", madsim)),
        tokio::test(start_paused = true)
    )]
    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
    async fn gzip_decoding_yields_and_obeys_deadline(
        #[case] expires: bool,
        #[case] empty_members: bool,
    ) {
        let (payload, compressed) = if empty_members {
            (Vec::new(), gzip(&[]).repeat(4_000))
        } else {
            let payload = vec![b'x'; 2 * 1024 * 1024];
            let compressed = gzip(&payload);
            (payload, compressed)
        };

        let body = Bytes::from(compressed);
        let body = Full::new(body);
        let response = http::Response::builder()
            .header("Content-Encoding", "gzip")
            .body(body)
            .unwrap();
        let client = HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap()
            .client;
        let deadline = expires.then(|| time::Instant::now() + Duration::from_secs(1));
        let mut request = Box::pin(client.consume_response(response, deadline, false));

        assert!(futures_util::poll!(request.as_mut()).is_pending());

        if expires {
            #[cfg(not(all(feature = "simulation", madsim)))]
            tokio::time::advance(Duration::from_secs(2)).await;
            #[cfg(all(feature = "simulation", madsim))]
            madsim::time::advance(Duration::from_secs(2));
        }

        let result = request.await;
        if expires {
            let HttpClientError::TimeoutError(message) = result.unwrap_err() else {
                panic!("expected decoding timeout");
            };

            assert_eq!(message, REQUEST_TIMEOUT_MESSAGE);
        } else {
            let response = result.unwrap();
            assert_eq!(response.status.as_u16(), 200);
            assert_eq!(response.headers, HashMap::new());
            assert_eq!(response.body.as_ref(), payload);
        }
    }

    #[rstest]
    #[case::stored_blocks(false)]
    #[case::header_fields(true)]
    #[cfg_attr(not(all(feature = "simulation", madsim)), tokio::test)]
    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
    async fn gzip_decoding_resumes_at_input_boundaries(#[case] metadata: bool) {
        let payload = vec![b'x'; 128 * 1024];

        let mut encoder = if metadata {
            flate2::GzBuilder::new()
                .filename(vec![b'f'; 32 * 1024])
                .comment(vec![b'c'; 32 * 1024])
                .write(Vec::new(), Compression::default())
        } else {
            GzEncoder::new(Vec::new(), Compression::none())
        };

        encoder.write_all(&payload).unwrap();
        let compressed = encoder.finish().unwrap();
        let response = http::Response::builder()
            .header("Content-Encoding", "gzip")
            .body(Full::new(Bytes::from(compressed)))
            .unwrap();
        let client = HttpClient::builder()
            .use_system_proxy(false)
            .build()
            .unwrap()
            .client;

        let response = client
            .consume_response(response, None, false)
            .await
            .unwrap();

        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), payload);
    }

    pub(in crate::http) fn gzip(data: &[u8]) -> Vec<u8> {
        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
        encoder.write_all(data).unwrap();
        encoder.finish().unwrap()
    }
}