fast_down/http/
puller.rs

1//! An HTTP implementation of the [`fast_pull::Puller`] trait.
2//!
3//! [`HttpPuller`] builds range requests through the generic [`crate::http::HttpClient`]
4//! trait and streams the response body back as a [`fast_pull::PullStream`]. It
5//! verifies the server's `ETag` / `Last-Modified` headers against the expected
6//! [`crate::FileId`] so that a changed file is reported as [`crate::http::HttpError::MismatchedBody`]
7//! rather than silently corrupting an incremental download.
8
9use crate::http::{
10    FileId, GetResponse, HttpClient, HttpError, HttpHeaders, HttpRequestBuilder, HttpResponse,
11};
12use async_stream::stream;
13use fast_pull::{ProgressEntry, PullResult, PullStream, Puller};
14use parking_lot::Mutex;
15use std::{fmt::Debug, future::Future, sync::Arc};
16use url::Url;
17
18/// A [`Puller`] implementation that fetches data over HTTP.
19///
20/// Uses the generic [`HttpClient`] trait so it can work with any HTTP backend
21/// (e.g. reqwest, custom clients). Supports range requests, file identity
22/// checking, and reusing previously opened responses.
23pub struct HttpPuller<Client: HttpClient> {
24    client: Client,
25    url: Arc<Url>,
26    resp: Option<Arc<Mutex<Option<GetResponse<Client>>>>>,
27    file_id: FileId,
28}
29impl<C: HttpClient> Clone for HttpPuller<C> {
30    fn clone(&self) -> Self {
31        Self {
32            client: self.client.clone(),
33            url: self.url.clone(),
34            resp: self.resp.clone(),
35            file_id: self.file_id.clone(),
36        }
37    }
38}
39impl<Client: HttpClient> HttpPuller<Client> {
40    /// Create a new [`HttpPuller`].
41    ///
42    /// * `url` — the resource to download.
43    /// * `client` — the HTTP client used to issue requests.
44    /// * `resp` — an optional already-open response to reuse for the first
45    ///   (full-file) request, typically the one produced by a prefetch.
46    /// * `file_id` — the expected [`crate::FileId`], compared against the
47    ///   server's headers to detect a changed resource.
48    pub const fn new(
49        url: Arc<Url>,
50        client: Client,
51        resp: Option<Arc<Mutex<Option<GetResponse<Client>>>>>,
52        file_id: FileId,
53    ) -> Self {
54        Self {
55            client,
56            url,
57            resp,
58            file_id,
59        }
60    }
61}
62impl<Client: HttpClient> Debug for HttpPuller<Client> {
63    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        f.debug_struct("HttpPuller")
65            .field("url", &self.url)
66            .field("file_id", &self.file_id)
67            .field("client", &"...")
68            .field("resp", &"...")
69            .finish()
70    }
71}
72
73impl<Client: HttpClient> Puller for HttpPuller<Client> {
74    type Error = HttpError<Client>;
75    fn pull(
76        &mut self,
77        range: Option<&ProgressEntry>,
78    ) -> impl Future<Output = PullResult<impl PullStream<Self::Error>, Self::Error>> + Send {
79        let client = self.client.clone();
80        let url = self.url.clone();
81        let file_id = self.file_id.clone();
82        // Reuse the prefetched whole-file response whenever the requested range
83        // begins at offset 0. That covers both a whole-file request (`range ==
84        // None`) and a finite ranged slice (`Some(0..end)`): the prefetched body
85        // starts at 0, so it already contains the requested prefix. A slice with
86        // `start > 0` cannot be served from a response rooted at 0, so it issues
87        // its own ranged request. The download engine caps any overshoot to the
88        // task's end, so no per-chunk slicing is needed here.
89        let mut reused = if range.is_none_or(|r| r.start == 0) {
90            self.resp.as_ref().and_then(|m| m.lock().take())
91        } else {
92            None
93        };
94        // Keep the `Option`: `None` means a whole-file GET with no `Range` header
95        // (single-threaded, no known end), `Some(r)` means a ranged `Range:` request
96        // (multi-threaded slice, resumable). The two states are handled in distinct
97        // branches below — no `u64::MAX` sentinel.
98        let mut range = range.cloned();
99
100        std::future::ready(Ok(Box::pin(stream! {
101            // Stream contract: `Ok(None)` (clean termination) is reached ONLY on a
102            // genuine EOF — when `current.chunk()` returns `Ok(None)`. After any
103            // error the stream stays alive and keeps yielding `Irrecoverable`
104            // (whose `is_irrecoverable()` is true), so the engine re-invokes
105            // `pull` to recover and can never mistake a failed pull for a finished
106            // one. A ranged read failure instead breaks back to the top to resume
107            // from the advanced offset (`range.start` is itself the cursor — it
108            // holds the number of bytes already yielded and advances across
109            // retries, so a resumed ranged request starts from it).
110            loop {
111                // Obtain the response for this pass. The prefetched whole-file body
112                // is taken once (it is consumed on the first iteration and `reused`
113                // becomes `None`, so retries fall through to a fresh request). The
114                // request uses `range.clone()`: on the first pass `range.start` is
115                // the original start, and after yielding chunks it is the advanced
116                // offset, so a retry automatically resumes from where it stopped.
117                let mut current: GetResponse<Client>;
118                if let Some(resp) = reused.take() {
119                    current = resp;
120                } else {
121                    match client.get((*url).clone(), range.clone()).send().await {
122                        Ok(resp) => current = resp,
123                        Err((e, retry)) => {
124                            // The request itself failed: report it once, then keep
125                            // yielding `Irrecoverable` so the engine re-pulls. The
126                            // stream must never return `Ok(None)` here.
127                            yield Err((HttpError::Request(e), retry));
128                            loop {
129                                yield Err((HttpError::Irrecoverable, None));
130                            }
131                        }
132                    }
133                }
134
135                // File identity is verified for every response — the reused one and
136                // every retry — in a single, obvious place.
137                let new_file_id = FileId::new(
138                    current.headers().get("etag").ok().as_deref(),
139                    current.headers().get("last-modified").ok().as_deref(),
140                );
141                if new_file_id != file_id {
142                    // Content changed underneath us. Report the mismatch, then keep
143                    // yielding `Irrecoverable` until the engine re-pulls — the
144                    // stream never terminates with `Ok(None)` after an error.
145                    yield Err((HttpError::MismatchedBody(new_file_id, current), None));
146                    loop {
147                        yield Err((HttpError::Irrecoverable, None));
148                    }
149                }
150
151                loop {
152                    match current.chunk().await {
153                        Ok(Some(chunk)) => {
154                            if let Some(r) = &mut range {
155                                r.start += chunk.len() as u64;
156                            }
157                            yield Ok(chunk);
158                        }
159                        Ok(None) => return,
160                        Err(e) => {
161                            // Report the underlying read failure (carries `current`
162                            // for diagnostics), then decide whether it is recoverable.
163                            yield Err((HttpError::Chunk(e, current), None));
164                            match &range {
165                                // Whole-file: there is no known end to resume from,
166                                // so the failure is fatal; the engine re-calls
167                                // `pull(None)` from the start.
168                                // Whole-file: no known end to resume from, so the
169                                // failure is terminal. Keep yielding `Irrecoverable`
170                                // (never `Ok(None)`) until the engine re-pulls.
171                                None => loop {
172                                    yield Err((HttpError::Irrecoverable, None));
173                                },
174                                // Ranged slice: `break` back to the top of the loop to
175                                // obtain a fresh request from the advanced `range.start`.
176                                Some(_) => break,
177                            }
178                        }
179                    }
180                }
181            }
182        })))
183    }
184}
185
186#[cfg(test)]
187mod tests {
188    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
189    use std::borrow::Cow;
190    use std::pin::Pin;
191    use std::sync::atomic::{AtomicU64, Ordering};
192    use std::task::{Context, Poll};
193    use std::time::Duration;
194
195    use bytes::Bytes;
196
197    use super::*;
198    use futures::TryStreamExt;
199
200    #[derive(Clone, Debug)]
201    struct MockClient;
202    impl HttpClient for MockClient {
203        type RequestBuilder = MockRequestBuilder;
204        fn get(&self, _url: Url, _range: Option<ProgressEntry>) -> Self::RequestBuilder {
205            MockRequestBuilder
206        }
207    }
208    struct MockRequestBuilder;
209    impl HttpRequestBuilder for MockRequestBuilder {
210        type Response = MockResponse;
211        type RequestError = MockError;
212        fn send(self) -> impl Future<Output = PullResult<Self::Response, Self::RequestError>> {
213            std::future::ready(Ok(MockResponse::new()))
214        }
215    }
216    #[derive(Debug)]
217    struct MockResponse {
218        headers: MockHeaders,
219        url: Url,
220    }
221    impl MockResponse {
222        fn new() -> Self {
223            Self {
224                headers: MockHeaders,
225                url: Url::parse("http://mock-url").unwrap(),
226            }
227        }
228    }
229    impl HttpResponse for MockResponse {
230        type Headers = MockHeaders;
231        type ChunkError = MockError;
232        fn headers(&self) -> &Self::Headers {
233            &self.headers
234        }
235        fn url(&self) -> &Url {
236            &self.url
237        }
238        async fn chunk(&mut self) -> Result<Option<Bytes>, Self::ChunkError> {
239            DelayChunk::new().await
240        }
241    }
242    #[derive(Debug)]
243    struct MockHeaders;
244    impl HttpHeaders for MockHeaders {
245        type GetHeaderError = MockError;
246        fn get(&self, _header: &str) -> Result<Cow<'_, str>, Self::GetHeaderError> {
247            Err(MockError)
248        }
249    }
250    #[derive(Debug, thiserror::Error)]
251    #[error("MockError")]
252    struct MockError;
253
254    struct DelayChunk {
255        polled_once: bool,
256    }
257    impl DelayChunk {
258        fn new() -> Self {
259            Self { polled_once: false }
260        }
261    }
262    impl Future for DelayChunk {
263        type Output = Result<Option<Bytes>, MockError>;
264        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
265            if !self.polled_once {
266                println!("Wait... [Mock: simulating network delay with Pending]");
267                self.polled_once = true;
268                cx.waker().wake_by_ref();
269                return Poll::Pending;
270            }
271            println!("Done! [Mock: data arrived, Ready]");
272            Poll::Ready(Ok(Some(Bytes::from_static(b"success"))))
273        }
274    }
275
276    #[tokio::test]
277    async fn test_http_puller_infinite_loop_fix() {
278        let url = Url::parse("http://localhost").unwrap();
279        let client = MockClient;
280        let file_id = FileId::new(None, None);
281        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
282        let range = 0..7;
283        let mut stream = Puller::pull(&mut puller, Some(&range))
284            .await
285            .expect("Failed to create stream");
286        println!("--- Starting HttpPuller test ---");
287        let result =
288            tokio::time::timeout(Duration::from_secs(1), async { stream.try_next().await }).await;
289        match result {
290            Ok(Ok(Some(bytes))) => {
291                println!("Received data: {bytes:?}");
292                assert_eq!(bytes, Bytes::from_static(b"success"));
293                println!("Test passed: HttpPuller correctly handled Pending state!");
294            }
295            e => {
296                panic!(
297                    "Test failed: timeout! This indicates HttpPuller may have lost the Future state after receiving Pending and entered an infinite loop. {e:?}"
298                );
299            }
300        }
301    }
302
303    #[derive(Debug)]
304    struct MismatchHeaders;
305    impl HttpHeaders for MismatchHeaders {
306        type GetHeaderError = MockError;
307        fn get(&self, header: &str) -> Result<Cow<'_, str>, Self::GetHeaderError> {
308            if header == "etag" {
309                Ok(Cow::Borrowed("etag-x"))
310            } else {
311                Err(MockError)
312            }
313        }
314    }
315
316    #[derive(Clone, Debug)]
317    struct MismatchClient;
318    impl HttpClient for MismatchClient {
319        type RequestBuilder = MismatchRequestBuilder;
320        fn get(&self, _url: Url, _range: Option<ProgressEntry>) -> Self::RequestBuilder {
321            MismatchRequestBuilder
322        }
323    }
324    struct MismatchRequestBuilder;
325    impl HttpRequestBuilder for MismatchRequestBuilder {
326        type Response = MismatchResponse;
327        type RequestError = MockError;
328        fn send(
329            self,
330        ) -> impl Future<Output = PullResult<Self::Response, Self::RequestError>> + Send {
331            std::future::ready(Ok(MismatchResponse::new()))
332        }
333    }
334    #[derive(Debug)]
335    struct MismatchResponse {
336        url: Url,
337    }
338    impl MismatchResponse {
339        fn new() -> Self {
340            Self {
341                url: Url::parse("http://mock-url").unwrap(),
342            }
343        }
344    }
345    impl HttpResponse for MismatchResponse {
346        type Headers = MismatchHeaders;
347        type ChunkError = MockError;
348        fn headers(&self) -> &Self::Headers {
349            &MismatchHeaders
350        }
351        fn url(&self) -> &Url {
352            &self.url
353        }
354        fn chunk(
355            &mut self,
356        ) -> impl Future<Output = Result<Option<Bytes>, Self::ChunkError>> + Send {
357            std::future::ready(Ok(None))
358        }
359    }
360
361    #[derive(Clone, Debug)]
362    struct ReqErrClient;
363    impl HttpClient for ReqErrClient {
364        type RequestBuilder = ReqErrRequestBuilder;
365        fn get(&self, _url: Url, _range: Option<ProgressEntry>) -> Self::RequestBuilder {
366            ReqErrRequestBuilder
367        }
368    }
369    struct ReqErrRequestBuilder;
370    impl HttpRequestBuilder for ReqErrRequestBuilder {
371        type Response = MockResponse;
372        type RequestError = MockError;
373        fn send(
374            self,
375        ) -> impl Future<Output = PullResult<Self::Response, Self::RequestError>> + Send {
376            std::future::ready(Err((MockError, None)))
377        }
378    }
379
380    #[derive(Clone, Debug)]
381    struct ChunkErrClient;
382    impl HttpClient for ChunkErrClient {
383        type RequestBuilder = ChunkErrRequestBuilder;
384        fn get(&self, _url: Url, _range: Option<ProgressEntry>) -> Self::RequestBuilder {
385            ChunkErrRequestBuilder
386        }
387    }
388    struct ChunkErrRequestBuilder;
389    impl HttpRequestBuilder for ChunkErrRequestBuilder {
390        type Response = ChunkErrResponse;
391        type RequestError = MockError;
392        fn send(
393            self,
394        ) -> impl Future<Output = PullResult<Self::Response, Self::RequestError>> + Send {
395            std::future::ready(Ok(ChunkErrResponse::new()))
396        }
397    }
398    #[derive(Debug)]
399    struct ChunkErrResponse {
400        url: Url,
401    }
402    impl ChunkErrResponse {
403        fn new() -> Self {
404            Self {
405                url: Url::parse("http://mock-url").unwrap(),
406            }
407        }
408    }
409    impl HttpResponse for ChunkErrResponse {
410        type Headers = MockHeaders;
411        type ChunkError = MockError;
412        fn headers(&self) -> &Self::Headers {
413            &MockHeaders
414        }
415        fn url(&self) -> &Url {
416            &self.url
417        }
418        fn chunk(
419            &mut self,
420        ) -> impl Future<Output = Result<Option<Bytes>, Self::ChunkError>> + Send {
421            std::future::ready(Err(MockError))
422        }
423    }
424
425    #[test]
426    fn http_puller_debug_formats() {
427        let url = Url::parse("http://localhost").unwrap();
428        let puller = HttpPuller::new(Arc::new(url), MockClient, None, FileId::new(None, None));
429        let s = format!("{puller:?}");
430        assert!(s.contains("HttpPuller"));
431        assert!(s.contains("url"));
432    }
433
434    #[test]
435    fn mock_response_url_is_accessible() {
436        let r = MockResponse::new();
437        assert_eq!(r.url().as_str(), "http://mock-url/");
438    }
439
440    #[tokio::test]
441    async fn test_http_puller_mismatched_body() {
442        let url = Url::parse("http://localhost").unwrap();
443        let client = MismatchClient;
444        let file_id = FileId::new(None, None);
445        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
446        let mut stream = Puller::pull(&mut puller, None).await.unwrap();
447        let result = stream.try_next().await;
448        assert!(matches!(
449            result,
450            Err((HttpError::MismatchedBody(_, _), None))
451        ));
452    }
453
454    #[tokio::test]
455    async fn test_http_puller_request_error() {
456        let url = Url::parse("http://localhost").unwrap();
457        let client = ReqErrClient;
458        let file_id = FileId::new(None, None);
459        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
460        let mut stream = Puller::pull(&mut puller, None).await.unwrap();
461        let result = stream.try_next().await;
462        assert!(matches!(result, Err((HttpError::Request(_), None))));
463    }
464
465    #[tokio::test]
466    async fn test_http_puller_chunk_error_then_irrecoverable() {
467        let url = Url::parse("http://localhost").unwrap();
468        let client = ChunkErrClient;
469        let file_id = FileId::new(None, None);
470        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
471        let mut stream = Puller::pull(&mut puller, None).await.unwrap();
472        // Full-file pull: Pending -> Ok(resp) -> file_id matches -> Streaming.
473        // The first poll of the stream yields the chunk error.
474        let first = stream.try_next().await;
475        assert!(matches!(first, Err((HttpError::Chunk(_, _), None))));
476        // The chunk error leaves state == None with range.end == u64::MAX, so the
477        // next poll returns HttpError::Irrecoverable.
478        let second = stream.try_next().await;
479        assert!(matches!(second, Err((HttpError::Irrecoverable, None))));
480    }
481
482    #[tokio::test]
483    async fn test_http_puller_error_stream_never_terminates() {
484        // After a whole-file error the stream must keep yielding `Irrecoverable`
485        // and never produce `Ok(None)`: `Ok(None)` is reserved for a genuine EOF,
486        // so the engine can never treat a failed pull as finished.
487        let url = Url::parse("http://localhost").unwrap();
488        let client = ChunkErrClient;
489        let file_id = FileId::new(None, None);
490        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
491        let mut stream = Puller::pull(&mut puller, None).await.unwrap();
492        // First poll: the chunk read error.
493        assert!(matches!(
494            stream.try_next().await,
495            Err((HttpError::Chunk(_, _), None))
496        ));
497        // Subsequent polls: always `Irrecoverable`, never `Ok(None)`.
498        for _ in 0..5 {
499            assert!(matches!(
500                stream.try_next().await,
501                Err((HttpError::Irrecoverable, None))
502            ));
503        }
504    }
505
506    #[tokio::test]
507    async fn test_http_puller_partial_range_request() {
508        let url = Url::parse("http://localhost").unwrap();
509        let client = MockClient;
510        let file_id = FileId::new(None, None);
511        let mut puller = HttpPuller::new(Arc::new(url), client, None, file_id);
512        let range = 10..100;
513        let mut stream = Puller::pull(&mut puller, Some(&range)).await.unwrap();
514        let result = tokio::time::timeout(Duration::from_secs(1), stream.try_next()).await;
515        match result {
516            Ok(Ok(Some(bytes))) => assert_eq!(bytes, Bytes::from_static(b"success")),
517            e => panic!("expected a successful chunk, got {e:?}"),
518        }
519    }
520
521    #[tokio::test]
522    async fn test_http_puller_reuses_response() {
523        let url = Url::parse("http://localhost").unwrap();
524        // Hand a pre-opened response to the puller; a full-file pull should reuse
525        // it directly (lines 119-123) instead of issuing a new request.
526        let resp = Some(Arc::new(parking_lot::Mutex::new(Some(MockResponse::new()))));
527        let file_id = FileId::new(None, None);
528        let mut puller = HttpPuller::new(Arc::new(url), MockClient, resp, file_id);
529        let mut stream = Puller::pull(&mut puller, None).await.unwrap();
530        let result = tokio::time::timeout(Duration::from_secs(1), stream.try_next()).await;
531        match result {
532            Ok(Ok(Some(bytes))) => assert_eq!(bytes, Bytes::from_static(b"success")),
533            e => panic!("expected the reused response to yield data, got {e:?}"),
534        }
535    }
536
537    // Resume-range test: verifies that after a mid-stream chunk error the retry
538    // request starts from the advanced offset, not the original range.start.
539    static RESUME_RANGE_START: AtomicU64 = AtomicU64::new(0);
540
541    #[derive(Clone, Debug)]
542    struct ResumeClient;
543    impl HttpClient for ResumeClient {
544        type RequestBuilder = ResumeRequestBuilder;
545        fn get(&self, _url: Url, range: Option<ProgressEntry>) -> Self::RequestBuilder {
546            if let Some(r) = range {
547                RESUME_RANGE_START.store(r.start, Ordering::SeqCst);
548            }
549            ResumeRequestBuilder
550        }
551    }
552    struct ResumeRequestBuilder;
553    impl HttpRequestBuilder for ResumeRequestBuilder {
554        type Response = ResumeResponse;
555        type RequestError = MockError;
556        fn send(
557            self,
558        ) -> impl Future<Output = PullResult<Self::Response, Self::RequestError>> + Send {
559            std::future::ready(Ok(ResumeResponse::new()))
560        }
561    }
562    #[derive(Debug)]
563    struct ResumeResponse {
564        url: Url,
565        calls: u32,
566    }
567    impl ResumeResponse {
568        fn new() -> Self {
569            Self {
570                url: Url::parse("http://mock-url").unwrap(),
571                calls: 0,
572            }
573        }
574    }
575    impl HttpResponse for ResumeResponse {
576        type Headers = MockHeaders;
577        type ChunkError = MockError;
578        fn headers(&self) -> &Self::Headers {
579            &MockHeaders
580        }
581        fn url(&self) -> &Url {
582            &self.url
583        }
584        fn chunk(
585            &mut self,
586        ) -> impl Future<Output = Result<Option<Bytes>, Self::ChunkError>> + Send {
587            self.calls += 1;
588            let first = self.calls == 1;
589            let fut = async move {
590                if first {
591                    Ok(Some(Bytes::from_static(b"success")))
592                } else {
593                    Err(MockError)
594                }
595            };
596            Box::pin(fut) as Pin<Box<dyn Future<Output = Result<Option<Bytes>, MockError>> + Send>>
597        }
598    }
599
600    #[tokio::test]
601    async fn test_resume_range_advances_after_chunk_error() {
602        let url = Url::parse("http://localhost").unwrap();
603        let file_id = FileId::new(None, None);
604        let mut puller = HttpPuller::new(Arc::new(url), ResumeClient, None, file_id);
605        let range = 0..100;
606        let mut stream = Puller::pull(&mut puller, Some(&range)).await.unwrap();
607        // First chunk succeeds ("success" = 7 bytes), advancing range.start to 7.
608        let first = stream.try_next().await;
609        assert!(matches!(first, Ok(Some(_))));
610        // Next poll: chunk error is returned, state resets to None.
611        let second = stream.try_next().await;
612        assert!(matches!(second, Err((HttpError::Chunk(_, _), None))));
613        // Third poll: state None -> retry request issued with advanced start == 7.
614        let third = stream.try_next().await;
615        assert!(matches!(third, Ok(Some(_))));
616        assert_eq!(RESUME_RANGE_START.load(Ordering::SeqCst), 7);
617    }
618}
fast_down/http/puller.rs

fast_down/http/
puller.rs
