fast_down/utils/
fast_puller.rs

1//! The default HTTP [`fast_pull::Puller`] for this crate.
2//!
3//! [`FastDownPuller`] ties together the [`crate::http::HttpPuller`] engine and a
4//! `SmartRedirectClient`, adding proxy support, optional
5//! multi-interface IP rotation, and file-identity-based resumability. Construct
6//! one from [`FastDownPullerOptions`] (typically via [`build_client`] to wire up
7//! the underlying reqwest client), then pass it to `fast_pull::download_multi`
8//! or `fast_pull::download_single` alongside a `Pusher` such as
9//! `fast_pull::StdFilePusher` (requires the `file` feature of `fast-pull`).
10
11use crate::Proxy;
12use crate::{
13    FileId, ProgressEntry, PullResult, PullStream,
14    http::{HttpError, HttpPuller, ReferrerPolicy},
15    reqwest::{BuildClientError, SmartRedirectClient},
16};
17use fast_pull::Puller;
18use parking_lot::Mutex;
19use reqwest::{ClientBuilder, Response, header::HeaderMap, redirect::Policy};
20use std::sync::Arc;
21use url::Url;
22
23/// # Errors
24/// Returns [`BuildClientError`] if the HTTP client cannot be built (invalid
25/// proxy URL, TLS backend failure, or other `reqwest::ClientBuilder` errors).
26pub fn build_client(
27    mut headers: HeaderMap,
28    proxy: Proxy<&str>,
29    #[allow(unused)] accept_invalid_certs: bool,
30    #[allow(unused)] accept_invalid_hostnames: bool,
31    #[allow(unused)] cookie_store: bool,
32    local_addr: Option<std::net::IpAddr>,
33    max_redirects: usize,
34) -> Result<SmartRedirectClient, BuildClientError> {
35    let referer = headers.remove("referer");
36    let referrer_policy = headers
37        .remove("referrer-policy")
38        .and_then(|v| v.to_str().ok().and_then(ReferrerPolicy::parse));
39    // Per RFC 9110 §15.4 item 2.5, resource-specific headers MUST be stripped
40    // on redirect. Extract them so they can be injected only on the first hop.
41    let origin = headers.remove("origin");
42    let authorization = headers.remove("authorization");
43    let cookie = headers.remove("cookie");
44    let mut client = ClientBuilder::new()
45        .default_headers(headers)
46        .local_address(local_addr)
47        .redirect(Policy::none());
48    client = match proxy {
49        Proxy::No => client.no_proxy(),
50        Proxy::System => client,
51        Proxy::Custom(p) => client.proxy(reqwest::Proxy::all(p)?),
52    };
53    #[cfg(feature = "reqwest-tls")]
54    {
55        client = client
56            .danger_accept_invalid_certs(accept_invalid_certs)
57            .danger_accept_invalid_hostnames(accept_invalid_hostnames);
58    }
59    #[cfg(feature = "cookie-store")]
60    {
61        client = client.cookie_store(cookie_store);
62    }
63    Ok(SmartRedirectClient::new(
64        client.build()?,
65        referer,
66        referrer_policy,
67        origin,
68        authorization,
69        cookie,
70        max_redirects,
71    ))
72}
73
74/// The default [`Puller`] implementation for the fast-down crate.
75///
76/// Wraps an [`HttpPuller`] with a [`SmartRedirectClient`], IP rotation,
77/// and proxy support. Cloning creates a new HTTP client with an optionally
78/// rotated local address for multi-interface setups.
79#[derive(Debug)]
80pub struct FastDownPuller {
81    inner: HttpPuller<SmartRedirectClient>,
82    headers: Arc<HeaderMap>,
83    proxy: Proxy<Arc<str>>,
84    url: Arc<Url>,
85    accept_invalid_certs: bool,
86    accept_invalid_hostnames: bool,
87    cookie_store: bool,
88    file_id: FileId,
89    resp: Option<Arc<Mutex<Option<Response>>>>,
90    available_ips: Arc<[std::net::IpAddr]>,
91    turn: Arc<std::sync::atomic::AtomicUsize>,
92    max_redirects: usize,
93}
94// Field-level docs live on [`FastDownPullerOptions`], the public construction
95// surface; the runtime struct mirrors those fields.
96
97/// Options for constructing a [`FastDownPuller`].
98#[derive(Debug)]
99pub struct FastDownPullerOptions<'a> {
100    /// The URL to download.
101    pub url: Url,
102    /// Extra request headers sent on every request.
103    pub headers: Arc<HeaderMap>,
104    /// Proxy selection (no / system / custom URL).
105    pub proxy: Proxy<&'a str>,
106    /// Accept invalid TLS certificates (requires the `reqwest-tls` feature).
107    pub accept_invalid_certs: bool,
108    /// Accept invalid TLS hostnames (requires the `reqwest-tls` feature).
109    pub accept_invalid_hostnames: bool,
110    /// Enable a cookie store (requires the `cookie-store` feature).
111    pub cookie_store: bool,
112    /// The expected [`FileId`], used to detect a changed resource and resume safely.
113    pub file_id: FileId,
114    /// An already-open response to reuse for the first request (e.g. from a prefetch).
115    pub resp: Option<Arc<Mutex<Option<Response>>>>,
116    /// Candidate local source IPs for outbound connections; rotated across clones.
117    pub available_ips: Arc<[std::net::IpAddr]>,
118    /// Maximum number of redirects to follow before failing.
119    pub max_redirects: usize,
120}
121
122impl FastDownPuller {
123    /// # Errors
124    /// Returns [`BuildClientError`] if the underlying HTTP client cannot be
125    /// built (invalid proxy, TLS setup failure, etc.).
126    pub fn new(option: FastDownPullerOptions<'_>) -> Result<Self, BuildClientError> {
127        let turn = Arc::new(std::sync::atomic::AtomicUsize::new(1));
128        let available_ips = option.available_ips;
129        let client = build_client(
130            option.headers.as_ref().clone(),
131            option.proxy,
132            option.accept_invalid_certs,
133            option.accept_invalid_hostnames,
134            option.cookie_store,
135            if available_ips.is_empty() {
136                None
137            } else {
138                available_ips
139                    .get(
140                        turn.fetch_add(1, std::sync::atomic::Ordering::AcqRel)
141                            % available_ips.len(),
142                    )
143                    .copied()
144            },
145            option.max_redirects,
146        )?;
147        let url = Arc::new(option.url);
148        Ok(Self {
149            inner: HttpPuller::new(
150                url.clone(),
151                client,
152                option.resp.clone(),
153                option.file_id.clone(),
154            ),
155            resp: option.resp,
156            headers: option.headers,
157            proxy: option.proxy.map(Arc::from),
158            url,
159            accept_invalid_certs: option.accept_invalid_certs,
160            accept_invalid_hostnames: option.accept_invalid_hostnames,
161            cookie_store: option.cookie_store,
162            file_id: option.file_id,
163            available_ips,
164            turn,
165            max_redirects: option.max_redirects,
166        })
167    }
168}
169
170impl Clone for FastDownPuller {
171    fn clone(&self) -> Self {
172        let available_ips = self.available_ips.clone();
173        let turn = self.turn.clone();
174        Self {
175            inner: build_client(
176                self.headers.as_ref().clone(),
177                self.proxy.as_deref(),
178                self.accept_invalid_certs,
179                self.accept_invalid_hostnames,
180                self.cookie_store,
181                if available_ips.is_empty() {
182                    None
183                } else {
184                    available_ips
185                        .get(
186                            turn.fetch_add(1, std::sync::atomic::Ordering::AcqRel)
187                                % available_ips.len(),
188                        )
189                        .copied()
190                },
191                self.max_redirects,
192            )
193            .map_or_else(
194                |_| self.inner.clone(),
195                |client| {
196                    HttpPuller::new(
197                        self.url.clone(),
198                        client,
199                        self.resp.clone(),
200                        self.file_id.clone(),
201                    )
202                },
203            ),
204            resp: self.resp.clone(),
205            headers: self.headers.clone(),
206            proxy: self.proxy.clone(),
207            url: self.url.clone(),
208            accept_invalid_certs: self.accept_invalid_certs,
209            accept_invalid_hostnames: self.accept_invalid_hostnames,
210            cookie_store: self.cookie_store,
211            file_id: self.file_id.clone(),
212            available_ips,
213            turn,
214            max_redirects: self.max_redirects,
215        }
216    }
217}
218
219impl Puller for FastDownPuller {
220    type Error = HttpError<SmartRedirectClient>;
221    async fn pull(
222        &mut self,
223        range: Option<&ProgressEntry>,
224    ) -> PullResult<impl PullStream<Self::Error>, Self::Error> {
225        Puller::pull(&mut self.inner, range).await
226    }
227}
228
229#[cfg(test)]
230mod tests {
231    #![allow(clippy::unwrap_used, clippy::expect_used)]
232    use super::*;
233
234    fn make_options(url: Url) -> FastDownPullerOptions<'static> {
235        FastDownPullerOptions {
236            url,
237            headers: Arc::new(HeaderMap::new()),
238            proxy: Proxy::No,
239            accept_invalid_certs: false,
240            accept_invalid_hostnames: false,
241            cookie_store: false,
242            file_id: FileId::default(),
243            resp: None,
244            available_ips: Arc::from(Vec::<std::net::IpAddr>::new()),
245            max_redirects: 10,
246        }
247    }
248
249    /// Plan A stores the URL in a single `Arc<Url>` shared between the
250    /// `FastDownPuller` and its `HttpPuller` inner, and carried across `Clone`.
251    /// This asserts that cloning a `FastDownPuller` shares the exact same
252    /// `Arc<Url>` (i.e. the URL is stored only once).
253    #[test]
254    fn test_fast_down_puller_clone_shares_url_arc() {
255        let opts = make_options(Url::parse("http://example.com/a.bin").unwrap());
256        let puller = FastDownPuller::new(opts).expect("FastDownPuller::new must succeed");
257        let cloned = puller.clone();
258        assert!(
259            Arc::ptr_eq(&puller.url, &cloned.url),
260            "clone() must share the same Arc<Url> (URL stored once)"
261        );
262    }
263
264    #[test]
265    fn build_client_custom_proxy_succeeds() {
266        let headers = HeaderMap::new();
267        let client = build_client(
268            headers,
269            Proxy::Custom("http://127.0.0.1:1080"),
270            false,
271            false,
272            false,
273            None,
274            10,
275        )
276        .expect("build_client with a custom proxy must succeed");
277        let _ = client;
278    }
279
280    #[test]
281    fn new_with_available_ips_rotates_local_address() {
282        let ips: Vec<std::net::IpAddr> = vec!["127.0.0.1".parse().unwrap()];
283        let mut opts = make_options(Url::parse("http://example.com/a.bin").unwrap());
284        opts.available_ips = Arc::from(ips);
285        let puller = FastDownPuller::new(opts).expect("FastDownPuller::new with ips must succeed");
286        let cloned = puller.clone();
287        assert!(Arc::ptr_eq(&puller.url, &cloned.url));
288    }
289
290    #[test]
291    fn clone_with_available_ips_rotates_local_address() {
292        let ips: Vec<std::net::IpAddr> = vec!["127.0.0.1".parse().unwrap()];
293        let mut opts = make_options(Url::parse("http://example.com/a.bin").unwrap());
294        opts.available_ips = Arc::from(ips);
295        let puller = FastDownPuller::new(opts).expect("FastDownPuller::new with ips must succeed");
296        // Cloning with non-empty available_ips rotates the local address and
297        // rebuilds a client via the `map_or_else` success branch.
298        let cloned = puller.clone();
299        assert!(Arc::ptr_eq(&puller.url, &cloned.url));
300    }
301
302    #[test]
303    fn build_client_system_proxy_succeeds() {
304        let headers = HeaderMap::new();
305        let client = build_client(headers, Proxy::System, false, false, false, None, 10)
306            .expect("build_client with the system proxy must succeed");
307        let _ = client;
308    }
309
310    #[tokio::test]
311    async fn fast_down_puller_pull_forwards_to_inner() {
312        let opts = make_options(Url::parse("http://example.com/a.bin").unwrap());
313        let mut puller = FastDownPuller::new(opts).expect("FastDownPuller::new must succeed");
314        // `pull` simply forwards to the inner `HttpPuller`; driving the stream is
315        // not required to cover the forwarding body.
316        let stream = FastDownPuller::pull(&mut puller, None).await;
317        assert!(stream.is_ok());
318    }
319
320    #[test]
321    fn clone_swallows_build_client_error() {
322        // Hypothesis C: `FastDownPuller::clone` rebuilds the client via
323        // `build_client`, but on failure it silently falls back to
324        // `self.inner.clone()` (the `map_or_else` Err branch). Because
325        // `Clone::clone` cannot return a `Result`, the build error is swallowed
326        // and the clone keeps the stale client instead of surfacing the failure.
327        let mut puller = FastDownPuller::new(make_options(
328            Url::parse("http://example.com/a.bin").unwrap(),
329        ))
330        .expect("new must succeed");
331        // Inject an invalid custom proxy that makes `build_client` fail on clone.
332        puller.proxy = Proxy::Custom(Arc::from("not a valid proxy url"));
333        // Clone must still succeed (the error is swallowed via `map_or_else`,
334        // not propagated) and reproduce the proxy config.
335        let cloned = puller.clone();
336        assert_eq!(
337            cloned.proxy, puller.proxy,
338            "clone must reproduce the proxy and not panic on a bad one"
339        );
340    }
341
342    #[test]
343    fn turn_counter_starts_at_one_offsetting_first_ip_index() {
344        // `turn` is initialized to 1, so the first `fetch_add` in `new` returns 1
345        // and the first instance picks IP index `1 % len` rather than `0`. The
346        // rotation is still complete (every IP is eventually reached); only the
347        // starting point is offset by one. This locks in the counter behavior.
348        let ips: Vec<std::net::IpAddr> = vec![
349            "127.0.0.1".parse().unwrap(),
350            "127.0.0.2".parse().unwrap(),
351            "127.0.0.3".parse().unwrap(),
352        ];
353        let mut opts = make_options(Url::parse("http://example.com/a.bin").unwrap());
354        opts.available_ips = Arc::from(ips);
355        let puller = FastDownPuller::new(opts).expect("new with ips must succeed");
356        // new() consumed the initial value 1 and advanced the counter to 2.
357        assert_eq!(puller.turn.load(std::sync::atomic::Ordering::Acquire), 2);
358        let cloned = puller.clone();
359        // clone() consumed 2 and advanced the counter to 3. The counter is a
360        // shared `Arc`, so both handles observe the same advanced value.
361        assert_eq!(cloned.turn.load(std::sync::atomic::Ordering::Acquire), 3);
362        assert_eq!(puller.turn.load(std::sync::atomic::Ordering::Acquire), 3);
363    }
364}
fast_down/utils/fast_puller.rs

fast_down/utils/
fast_puller.rs
