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

fast_pull/base/
puller.rs
