fast_pull/base/
puller.rs

1//! The [`Puller`](crate::Puller) trait: an abstraction over a chunked data source.
2
3use crate::ProgressEntry;
4use bytes::Bytes;
5use core::time::Duration;
6use futures::TryStream;
7
8/// A pull stream that yields [`Bytes`] chunks.
9///
10/// Each chunk is accompanied by an optional retry delay on error.
11pub trait PullStream<E>:
12    TryStream<Ok = Bytes, Error = (E, Option<Duration>)> + Send + Unpin
13{
14}
15impl<E, T> PullStream<E> for T where
16    T: TryStream<Ok = Bytes, Error = (E, Option<Duration>)> + Send + Unpin
17{
18}
19/// Result type returned by pulling operations.
20///
21/// On error, returns the error alongside an optional retry-after duration.
22pub type PullResult<T, E> = Result<T, (E, Option<Duration>)>;
23
24/// Abstraction over a data source that can be pulled (downloaded) in chunks.
25///
26/// Implementors produce a [`PullStream`] of bytes, optionally restricted to a
27/// specific byte range. Cloning is required for retry and work-stealing scenarios.
28pub trait Puller: Send + Sync + Clone + 'static {
29    /// Error produced while pulling.
30    type Error: PullerError;
31    /// Pull a (sub)range of the source as a stream of byte chunks.
32    ///
33    /// Passing `None` for `range` requests the entire source. The returned
34    /// [`PullStream`] yields [`Bytes`] chunks; each error carries an optional
35    /// retry delay that the engine honors via its retry backoff. Implementors
36    /// must be `Clone` so workers can be spawned and work can be stolen/retried.
37    fn pull(
38        &mut self,
39        range: Option<&ProgressEntry>,
40    ) -> impl Future<Output = PullResult<impl PullStream<Self::Error>, Self::Error>> + Send;
41}
42
43/// Extension trait for pull errors, distinguishing recoverable from irrecoverable failures.
44pub trait PullerError: std::error::Error + Send + Sync + Unpin + 'static {
45    /// Whether an error is fatal and must **not** be retried.
46    ///
47    /// The default (`false`) means the error is recoverable and the engine will
48    /// retry after the configured backoff. Implementors **must** override this to
49    /// return `true` for fatal errors: forgetting to do so lets the engine retry
50    /// indefinitely until its backoff gives up.
51    fn is_irrecoverable(&self) -> bool {
52        false
53    }
54}
55
56impl PullerError for std::convert::Infallible {
57    fn is_irrecoverable(&self) -> bool {
58        #[allow(clippy::uninhabited_references)]
59        match *self {}
60    }
61}
62
63#[cfg(test)]
64mod tests {
65    #![allow(clippy::unwrap_used)]
66    use super::*;
67
68    /// A `PullerError` that relies on the default `is_irrecoverable` impl.
69    #[derive(Debug)]
70    struct DefaultErr;
71    impl std::fmt::Display for DefaultErr {
72        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73            f.write_str("default error")
74        }
75    }
76    impl std::error::Error for DefaultErr {}
77    impl PullerError for DefaultErr {}
78
79    #[test]
80    fn default_is_irrecoverable_is_false() {
81        // Exercises the default `PullerError::is_irrecoverable` body (lines 49-51) and
82        // the `Display` impl for `DefaultErr` (lines 70-72).
83        assert!(!DefaultErr.is_irrecoverable());
84        assert_eq!(format!("{DefaultErr}"), "default error");
85    }
86
87    /// A `PullerError` that overrides `is_irrecoverable` to report a fatal error.
88    #[derive(Debug)]
89    struct FatalErr;
90    impl std::fmt::Display for FatalErr {
91        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92            f.write_str("fatal error")
93        }
94    }
95    impl std::error::Error for FatalErr {}
96    impl PullerError for FatalErr {
97        fn is_irrecoverable(&self) -> bool {
98            true
99        }
100    }
101
102    #[test]
103    fn override_is_irrecoverable_is_true() {
104        // An error that overrides the method reports `true`, which is how the
105        // engine learns to stop retrying. Pinned alongside the default `false`.
106        assert!(FatalErr.is_irrecoverable());
107        assert!(!DefaultErr.is_irrecoverable());
108    }
109
110    #[test]
111    fn irrecoverable_contract_allows_dynamic_decision() {
112        // The answer may depend on the error's own state; it is not required to
113        // be a compile-time constant per type.
114        #[derive(Debug)]
115        struct StatefulErr {
116            fatal: bool,
117        }
118        impl std::fmt::Display for StatefulErr {
119            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
120                f.write_str("stateful")
121            }
122        }
123        impl std::error::Error for StatefulErr {}
124        impl PullerError for StatefulErr {
125            fn is_irrecoverable(&self) -> bool {
126                self.fatal
127            }
128        }
129        assert!(!StatefulErr { fatal: false }.is_irrecoverable());
130        assert!(StatefulErr { fatal: true }.is_irrecoverable());
131    }
132}
fast_pull/base/puller.rs

fast_pull/base/
puller.rs
