aria2_core/http/
hyper_client.rs

1//! Direct hyper HTTP client for the hot download path.
2//!
3//! Bypasses reqwest overhead for simple HTTP range GETs by using hyper's
4//! `Client` with a tuned `HttpConnector`. The connector enables hyper's
5//! built-in Happy Eyeballs racing (`set_happy_eyeballs_timeout`) so IPv6/IPv4
6//! connects are raced with a 250ms IPv6 head start, matching the behaviour of
7//! `crate::http::happy_eyeballs::connect_happy_eyeballs`.
8//!
9//! For proxy / complex-auth / HTTPS-with-custom-TLS paths, reqwest is still
10//! used (see `client_pool.rs` and `connection.rs`); this module does NOT remove
11//! reqwest.
12//!
13//! # TODO (future work)
14//! - Implement a custom `hyper::service::Service` connector that delegates to
15//!   `crate::http::happy_eyeballs::connect_happy_eyeballs` for full control
16//!   over the dual-stack race (including custom DNS via `hickory-resolver`).
17//! - Wire `HyperDirectClient::download_range` into
18//!   `HttpSegmentDownloader::download_range` when no proxy is configured
19//!   (the "hot path").
20
21use std::time::Duration;
22
23use bytes::Bytes;
24use futures::StreamExt;
25use futures::stream::Stream;
26use hyper::StatusCode;
27use hyper::body::HttpBody;
28use hyper::client::{Client, HttpConnector};
29use tracing::{debug, warn};
30
31use crate::error::{Aria2Error, FatalError, RecoverableError, Result};
32
33/// IPv6 head start for hyper's built-in Happy Eyeballs (RFC 8305: 250ms).
34const HAPPY_EYEBALLS_HEAD_START: Duration = Duration::from_millis(250);
35
36/// Idle connection lifetime in the hyper connection pool.
37const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
38
39/// Max idle connections kept per host in the hyper connection pool.
40const POOL_MAX_IDLE_PER_HOST: usize = 16;
41
42/// TCP keepalive interval applied to pooled connections.
43const TCP_KEEPALIVE: Duration = Duration::from_secs(60);
44
45/// Upper bound for the initial `BytesMut` capacity (avoids huge allocations
46/// when a caller passes a very large `length` hint).
47const MAX_INITIAL_CAPACITY: u64 = 4 * 1024 * 1024;
48
49/// A lightweight HTTP client using hyper directly.
50///
51/// Intended for the hot download path: simple HTTP range GETs with no proxy
52/// and no complex auth. It avoids reqwest's abstraction overhead (middleware,
53/// redirect-policy objects, default headers plumbing) while keeping a
54/// connection pool for reuse across segments.
55///
56/// HTTPS is supported via hyper's connector only when a TLS backend is wired
57/// in; for the plain-HTTP hot path this client is sufficient. HTTPS fallback
58/// to reqwest remains available.
59pub struct HyperDirectClient {
60    client: Client<HttpConnector, hyper::Body>,
61}
62
63impl HyperDirectClient {
64    /// Create a new `HyperDirectClient` with a tuned `HttpConnector`.
65    pub fn new() -> Self {
66        let mut connector = HttpConnector::new();
67        connector.set_keepalive(Some(TCP_KEEPALIVE));
68        connector.set_nodelay(true);
69        // Enable hyper's built-in Happy Eyeballs: race IPv6/IPv4 connect
70        // attempts with a 250ms IPv6 head start. This mirrors our
71        // `connect_happy_eyeballs` logic but is applied inside hyper's
72        // connector. A future custom `Service` connector could delegate to
73        // `connect_happy_eyeballs` directly for finer control (e.g. a custom
74        // DNS resolver).
75        connector.set_happy_eyeballs_timeout(Some(HAPPY_EYEBALLS_HEAD_START));
76
77        let client = Client::builder()
78            .pool_idle_timeout(Some(POOL_IDLE_TIMEOUT))
79            .pool_max_idle_per_host(POOL_MAX_IDLE_PER_HOST)
80            .build(connector);
81        Self { client }
82    }
83
84    /// Download a byte range from `url` and collect it into `Bytes`.
85    ///
86    /// * `length = Some(n)` requests `bytes={offset}-{offset+n-1}`.
87    /// * `length = None` requests `bytes={offset}-` (read to end of body).
88    ///
89    /// Status handling mirrors `HttpSegmentDownloader::download_range`:
90    /// - `206 Partial Content`: OK.
91    /// - `200 OK`: warn (server ignored Range) and proceed.
92    /// - `416 Range Not Satisfiable`: recoverable error.
93    /// - other `4xx`: fatal error.
94    /// - `5xx`: recoverable server error.
95    ///
96    /// The body is accumulated into a `BytesMut` (sized by the smaller of
97    /// `length` and `MAX_INITIAL_CAPACITY`) and frozen into a zero-copy
98    /// `Bytes`.
99    pub async fn download_range(
100        &self,
101        url: &str,
102        offset: u64,
103        length: Option<u64>,
104    ) -> Result<Bytes> {
105        // Zero-length explicit range: short-circuit without a network round trip.
106        if matches!(length, Some(0)) {
107            return Ok(Bytes::new());
108        }
109
110        let range_header = build_range_header(offset, length);
111        debug!("hyper direct range request: {} ({})", range_header, url);
112
113        let request = build_range_request(url, &range_header)?;
114
115        let response = self.client.request(request).await.map_err(|e| {
116            Aria2Error::Recoverable(RecoverableError::TemporaryNetworkFailure {
117                message: format!("hyper request failed: {e}"),
118            })
119        })?;
120
121        let status = response.status();
122        match status.as_u16() {
123            206 => {}
124            200 => warn!(
125                "hyper direct: server returned 200 instead of 206 for Range request \
126                 (offset={}, len={:?}) at {}",
127                offset, length, url
128            ),
129            416 => {
130                return Err(Aria2Error::Recoverable(
131                    RecoverableError::TemporaryNetworkFailure {
132                        message: format!("Range not satisfiable: {range_header}"),
133                    },
134                ));
135            }
136            code if (400..500).contains(&code) => {
137                return Err(Aria2Error::Fatal(FatalError::Config(format!(
138                    "HTTP client error {code}: {url}"
139                ))));
140            }
141            code if code >= 500 => {
142                return Err(Aria2Error::Recoverable(RecoverableError::ServerError {
143                    code,
144                }));
145            }
146            _ => {}
147        }
148
149        // Accumulate body chunks into a BytesMut, then freeze to Bytes.
150        let initial_cap = length.unwrap_or(0).min(MAX_INITIAL_CAPACITY) as usize;
151        let mut buf = bytes::BytesMut::with_capacity(initial_cap);
152        let mut body = response.into_body();
153        while let Some(chunk) = body.data().await {
154            let chunk = chunk.map_err(|e| {
155                Aria2Error::Recoverable(RecoverableError::TemporaryNetworkFailure {
156                    message: format!("hyper stream read error: {e}"),
157                })
158            })?;
159            buf.extend_from_slice(&chunk);
160        }
161
162        // An empty body is only an error when a concrete non-zero length was
163        // requested (mirrors `HttpSegmentDownloader::download_range`).
164        if buf.is_empty() && matches!(length, Some(l) if l > 0) {
165            return Err(Aria2Error::Recoverable(
166                RecoverableError::TemporaryNetworkFailure {
167                    message: format!("Empty response for range {range_header} from {url}"),
168                },
169            ));
170        }
171
172        Ok(buf.freeze())
173    }
174
175    /// Download a byte range and return the body as a stream of `Bytes` chunks.
176    ///
177    /// The caller is responsible for consuming the stream (e.g. writing chunks
178    /// to disk). Each item is `Result<Bytes, std::io::Error>`; hyper errors are
179    /// mapped to `std::io::Error`.
180    ///
181    /// Status handling here is stricter than `download_range`: only `2xx` and
182    /// `206` are accepted; any other status is an error before streaming begins.
183    pub async fn download_range_stream(
184        &self,
185        url: &str,
186        offset: u64,
187        length: Option<u64>,
188    ) -> Result<impl Stream<Item = std::result::Result<Bytes, std::io::Error>>> {
189        // Zero-length explicit range: return an empty stream without a request.
190        // `hyper::Body::empty()` is a `Stream` that yields nothing immediately,
191        // so mapping it produces the same concrete `Map<hyper::Body, _>` type
192        // as the main path (both use the named `map_body_chunk` function).
193        if matches!(length, Some(0)) {
194            return Ok(hyper::Body::empty().map(map_body_chunk));
195        }
196
197        let range_header = build_range_header(offset, length);
198        debug!("hyper direct range stream: {} ({})", range_header, url);
199
200        let request = build_range_request(url, &range_header)?;
201
202        let response = self.client.request(request).await.map_err(|e| {
203            Aria2Error::Recoverable(RecoverableError::TemporaryNetworkFailure {
204                message: format!("hyper request failed: {e}"),
205            })
206        })?;
207
208        let status = response.status();
209        if !status.is_success() && status != StatusCode::PARTIAL_CONTENT {
210            return Err(Aria2Error::Recoverable(RecoverableError::ServerError {
211                code: status.as_u16(),
212            }));
213        }
214
215        // `hyper::Body` implements `Stream<Item = Result<Bytes, hyper::Error>>`
216        // with the `stream` feature; map the error type to `std::io::Error`.
217        Ok(response.into_body().map(map_body_chunk))
218    }
219}
220
221impl Default for HyperDirectClient {
222    fn default() -> Self {
223        Self::new()
224    }
225}
226
227/// Build a `bytes={start}-{end}` Range header value.
228///
229/// `None` length produces an open-ended range (`bytes={offset}-`).
230///
231/// The end byte is computed as `offset + (len - 1)` using saturating
232/// arithmetic: `len.saturating_sub(1)` is evaluated first (safe for any `len`),
233/// then `offset.saturating_add(...)` prevents overflow panics for extreme
234/// offsets. This mirrors `HttpSegmentDownloader`'s `offset + length - 1` but
235/// is panic-free. For realistic file offsets the saturating guards never
236/// engage.
237fn build_range_header(offset: u64, length: Option<u64>) -> String {
238    match length {
239        Some(len) => format!(
240            "bytes={}-{}",
241            offset,
242            offset.saturating_add(len.saturating_sub(1))
243        ),
244        None => format!("bytes={offset}-"),
245    }
246}
247
248/// Build a `GET` request with the given Range header and the project user-agent.
249fn build_range_request(url: &str, range_header: &str) -> Result<hyper::Request<hyper::Body>> {
250    let uri: hyper::Uri = url
251        .parse()
252        .map_err(|e| Aria2Error::Parse(format!("invalid URL {url:?}: {e}")))?;
253
254    hyper::Request::builder()
255        .method("GET")
256        .uri(uri)
257        .header("range", range_header)
258        .header("user-agent", crate::constants::USER_AGENT)
259        .body(hyper::Body::empty())
260        .map_err(|e| Aria2Error::Io(format!("failed to build request: {e}")))
261}
262
263/// Convert a hyper body chunk result to `Result<Bytes, std::io::Error>`.
264///
265/// Defined as a named function (not a closure) so that `body.map(map_body_chunk)`
266/// at multiple call sites produces the *same* concrete `Map<hyper::Body, _>`
267/// type, allowing a single `impl Stream` return type to unify across the
268/// zero-length short-circuit and the normal streaming path.
269fn map_body_chunk(
270    res: std::result::Result<Bytes, hyper::Error>,
271) -> std::result::Result<Bytes, std::io::Error> {
272    res.map_err(|e| std::io::Error::other(format!("hyper stream error: {e}")))
273}
274
275#[cfg(test)]
276mod tests {
277    use super::*;
278    use hyper::server::Server;
279    use hyper::service::{make_service_fn, service_fn};
280    use hyper::{Body, Request, Response, StatusCode};
281    use std::convert::Infallible;
282    use std::net::SocketAddr;
283
284    /// Test payload served by the local server (11 bytes: "hello world").
285    const PAYLOAD: &[u8] = b"hello world";
286
287    /// Handler that honours `Range` requests against `PAYLOAD`.
288    ///
289    /// * `bytes=0-4`   -> `206` with "hello"
290    /// * `bytes=6-`    -> `206` with "world"
291    /// * out-of-range  -> `416`
292    /// * no Range      -> `200` with full payload
293    async fn handle(req: Request<Body>) -> std::result::Result<Response<Body>, Infallible> {
294        let range = req
295            .headers()
296            .get("range")
297            .and_then(|v| v.to_str().ok())
298            .unwrap_or("");
299
300        if let Some(spec) = range.strip_prefix("bytes=") {
301            let (start_s, end_s) = spec.split_once('-').unwrap_or((spec, ""));
302            let start: usize = start_s.parse().unwrap_or(0);
303
304            if start >= PAYLOAD.len() {
305                return Ok(Response::builder()
306                    .status(StatusCode::RANGE_NOT_SATISFIABLE)
307                    .header("content-range", format!("bytes */{}", PAYLOAD.len()))
308                    .body(Body::empty())
309                    .unwrap());
310            }
311
312            let last = PAYLOAD.len() - 1;
313            let end: usize = if end_s.is_empty() {
314                last
315            } else {
316                end_s.parse::<usize>().unwrap_or(last).min(last)
317            };
318            let slice = &PAYLOAD[start..=end];
319
320            return Ok(Response::builder()
321                .status(StatusCode::PARTIAL_CONTENT)
322                .header(
323                    "content-range",
324                    format!("bytes {}-{}/{}", start, end, PAYLOAD.len()),
325                )
326                .body(Body::from(slice.to_vec()))
327                .unwrap());
328        }
329
330        Ok(Response::new(Body::from(PAYLOAD.to_vec())))
331    }
332
333    /// Spawn a local hyper server on an ephemeral port and return its address.
334    async fn spawn_server() -> SocketAddr {
335        let addr = SocketAddr::from(([127, 0, 0, 1], 0));
336        let make_svc = make_service_fn(|_conn| async { Ok::<_, Infallible>(service_fn(handle)) });
337        let server = Server::bind(&addr).serve(make_svc);
338        let local_addr = server.local_addr();
339        tokio::spawn(async move {
340            let _ = server.await;
341        });
342        local_addr
343    }
344
345    #[tokio::test]
346    async fn test_download_range_partial() {
347        let addr = spawn_server().await;
348        let url = format!("http://{addr}/");
349        let client = HyperDirectClient::new();
350        let data = client.download_range(&url, 0, Some(5)).await.unwrap();
351        assert_eq!(data.as_ref(), b"hello");
352    }
353
354    #[tokio::test]
355    async fn test_download_range_open_ended() {
356        let addr = spawn_server().await;
357        let url = format!("http://{addr}/");
358        let client = HyperDirectClient::new();
359        // bytes=6- => "world"
360        let data = client.download_range(&url, 6, None).await.unwrap();
361        assert_eq!(data.as_ref(), b"world");
362    }
363
364    #[tokio::test]
365    async fn test_download_range_zero_length() {
366        let addr = spawn_server().await;
367        let url = format!("http://{addr}/");
368        let client = HyperDirectClient::new();
369        let data = client.download_range(&url, 0, Some(0)).await.unwrap();
370        assert!(data.is_empty());
371    }
372
373    #[tokio::test]
374    async fn test_download_range_416_returns_error() {
375        let addr = spawn_server().await;
376        let url = format!("http://{addr}/");
377        let client = HyperDirectClient::new();
378        // offset beyond payload length => 416
379        let result = client.download_range(&url, 100, Some(5)).await;
380        assert!(result.is_err(), "expected error for 416 status");
381        // Should be a recoverable TemporaryNetworkFailure.
382        match result.unwrap_err() {
383            Aria2Error::Recoverable(RecoverableError::TemporaryNetworkFailure { .. }) => {}
384            other => panic!("expected TemporaryNetworkFailure, got {other:?}"),
385        }
386    }
387
388    #[tokio::test]
389    async fn test_download_range_stream_partial() {
390        let addr = spawn_server().await;
391        let url = format!("http://{addr}/");
392        let client = HyperDirectClient::new();
393        let mut stream = client
394            .download_range_stream(&url, 0, Some(5))
395            .await
396            .unwrap();
397
398        let mut total = Vec::new();
399        while let Some(chunk) = stream.next().await {
400            let chunk = chunk.expect("stream chunk should be ok");
401            total.extend_from_slice(&chunk);
402        }
403        assert_eq!(&total, b"hello");
404    }
405
406    #[tokio::test]
407    async fn test_download_range_stream_zero_length_yields_nothing() {
408        let addr = spawn_server().await;
409        let url = format!("http://{addr}/");
410        let client = HyperDirectClient::new();
411        let mut stream = client
412            .download_range_stream(&url, 0, Some(0))
413            .await
414            .unwrap();
415
416        let mut count = 0;
417        while let Some(chunk) = stream.next().await {
418            count += chunk.expect("chunk ok").len() as u32;
419        }
420        assert_eq!(count, 0, "zero-length range stream should yield no bytes");
421    }
422
423    #[test]
424    fn test_build_range_header() {
425        assert_eq!(build_range_header(0, Some(5)), "bytes=0-4");
426        assert_eq!(build_range_header(10, Some(1)), "bytes=10-10");
427        assert_eq!(build_range_header(6, None), "bytes=6-");
428        // Extreme offset with len=1: end must equal offset (offset + 0), and
429        // the saturating add must NOT panic nor wrap to offset-1.
430        assert_eq!(
431            build_range_header(u64::MAX, Some(1)),
432            "bytes=18446744073709551615-18446744073709551615"
433        );
434    }
435
436    #[test]
437    fn test_default_equals_new() {
438        // Both should construct without panic; structurally identical.
439        let _a = HyperDirectClient::default();
440        let _b = HyperDirectClient::new();
441    }
442}
aria2_core/http/hyper_client.rs

aria2_core/http/
hyper_client.rs
