nectar_primitives/store/
retry.rs

1//! `RetryingChunkGet`: a wasm-safe [`ChunkGet`] decorator that absorbs
2//! transient retrieval failures with bounded exponential backoff.
3//!
4//! A joiner propagates the first `get` error, which aborts a whole-file
5//! reconstruction. On a live network a single `get` often fails transiently:
6//! too few candidate storers momentarily, or a candidate refusing under load.
7//! Retrying with capped exponential backoff turns those transient misses into
8//! eventual hits so a large download survives per-chunk flakiness.
9//!
10//! The decorator takes no timer dependency of its own: the sleep is injected
11//! through [`Sleeper`], so each consumer supplies its platform delay (tokio on
12//! native, a browser timer on wasm) and nectar stays timer-agnostic.
13
14use std::fmt;
15use std::future::Future;
16use std::time::Duration;
17
18use super::maybe_send::{MaybeSend, MaybeSync};
19use super::typed::{ChunkGet, ChunkHas, ChunkPut};
20use crate::chunk::{AnyChunk, ChunkAddress};
21
22/// Injected async delay so the decorator owns its timer: nectar takes no new
23/// timer dependency and each consumer supplies its platform sleep.
24pub trait Sleeper: MaybeSend + MaybeSync {
25    /// Complete after at least `dur` has elapsed.
26    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + MaybeSend;
27}
28
29/// Retry budget and backoff shape for [`RetryingChunkGet`].
30#[derive(Clone, Copy, Debug)]
31pub struct RetryConfig {
32    /// Total `get` attempts (initial try plus retries) before the error
33    /// propagates. Counts the first try, so `1` disables retrying.
34    pub max_attempts: u32,
35    /// Backoff before the first retry; doubles each subsequent retry up to
36    /// [`Self::backoff_cap`].
37    pub base_backoff: Duration,
38    /// Upper bound on a single backoff wait, so late retries stay responsive.
39    pub backoff_cap: Duration,
40}
41
42impl Default for RetryConfig {
43    fn default() -> Self {
44        Self {
45            max_attempts: 8,
46            base_backoff: Duration::from_millis(150),
47            backoff_cap: Duration::from_secs(8),
48        }
49    }
50}
51
52impl RetryConfig {
53    /// Backoff for the retry that follows attempt `attempt` (1-based): base
54    /// doubled `attempt - 1` times, capped, plus up to 50% jitter keyed on the
55    /// address so chunks failing together spread their retries apart.
56    fn backoff_for(&self, attempt: u32, address: &ChunkAddress) -> Duration {
57        let shift = attempt.saturating_sub(1).min(16);
58        let scaled = self.base_backoff.saturating_mul(1u32 << shift);
59        let capped = scaled.min(self.backoff_cap);
60        let jitter = capped
61            .mul_f64(0.5 * jitter_unit(address))
62            .min(self.backoff_cap);
63        capped.saturating_add(jitter)
64    }
65}
66
67/// A wasm-safe pseudo-random value in `[0, 1)` used only to decorrelate retries
68/// of chunks that failed together. Mixes the `web-time` wall clock (browser
69/// clock on wasm, `std::time` on native) with the address so distinct chunks
70/// jitter apart even within one clock tick; needs no `rand` dependency and is
71/// never security-sensitive.
72fn jitter_unit(address: &ChunkAddress) -> f64 {
73    use web_time::{SystemTime, UNIX_EPOCH};
74    let nanos = SystemTime::now()
75        .duration_since(UNIX_EPOCH)
76        .map_or(0, |d| d.subsec_nanos());
77    let addr_mix = address
78        .as_bytes()
79        .iter()
80        .take(4)
81        .fold(0u32, |acc, &b| (acc << 8) | u32::from(b));
82    f64::from(nanos ^ addr_mix) / (f64::from(u32::MAX) + 1.0)
83}
84
85/// [`ChunkGet`] decorator that retries transient `get` failures with capped
86/// exponential backoff and jitter, sleeping through an injected [`Sleeper`].
87///
88/// Retries on any error since the inner error type is opaque here; a genuinely
89/// unretrievable chunk still fails, but only after the attempt budget is spent.
90/// `put` and `has` delegate to the inner store untouched. `Clone` is cheap when
91/// `G` and `S` are.
92#[derive(Clone)]
93pub struct RetryingChunkGet<G, S> {
94    inner: G,
95    sleeper: S,
96    config: RetryConfig,
97}
98
99impl<G, S> fmt::Debug for RetryingChunkGet<G, S> {
100    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
101        f.debug_struct("RetryingChunkGet")
102            .field("config", &self.config)
103            .finish_non_exhaustive()
104    }
105}
106
107impl<G, S> RetryingChunkGet<G, S> {
108    /// Wrap `inner`, sleeping through `sleeper`, using `config`.
109    pub const fn new(inner: G, sleeper: S, config: RetryConfig) -> Self {
110        Self {
111            inner,
112            sleeper,
113            config,
114        }
115    }
116
117    /// Wrap `inner` with [`RetryConfig::default`].
118    pub fn with_default(inner: G, sleeper: S) -> Self {
119        Self::new(inner, sleeper, RetryConfig::default())
120    }
121}
122
123impl<const BS: usize, G: ChunkGet<BS>, S: Sleeper> ChunkGet<BS> for RetryingChunkGet<G, S> {
124    type Error = G::Error;
125
126    async fn get(&self, address: &ChunkAddress) -> Result<AnyChunk<BS>, Self::Error> {
127        let mut attempt = 1;
128        loop {
129            match self.inner.get(address).await {
130                Ok(chunk) => return Ok(chunk),
131                Err(e) => {
132                    if attempt >= self.config.max_attempts {
133                        return Err(e);
134                    }
135                    self.sleeper
136                        .sleep(self.config.backoff_for(attempt, address))
137                        .await;
138                    attempt += 1;
139                }
140            }
141        }
142    }
143}
144
145impl<const BS: usize, G: ChunkPut<BS>, S: MaybeSend + MaybeSync> ChunkPut<BS>
146    for RetryingChunkGet<G, S>
147{
148    type Error = G::Error;
149
150    async fn put(&self, chunk: AnyChunk<BS>) -> Result<(), Self::Error> {
151        self.inner.put(chunk).await
152    }
153}
154
155impl<const BS: usize, G: ChunkHas<BS>, S: MaybeSend + MaybeSync> ChunkHas<BS>
156    for RetryingChunkGet<G, S>
157{
158    async fn has(&self, address: &ChunkAddress) -> bool {
159        self.inner.has(address).await
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166
167    use std::sync::Mutex;
168    use std::sync::atomic::{AtomicU32, Ordering};
169
170    use futures::executor::block_on;
171
172    use crate::DefaultContentChunk;
173    use crate::bmt::DEFAULT_BODY_SIZE;
174
175    /// A [`Sleeper`] that returns immediately, so tests never wait real time.
176    struct NoSleep;
177
178    impl Sleeper for NoSleep {
179        async fn sleep(&self, _dur: Duration) {}
180    }
181
182    #[derive(Debug, thiserror::Error)]
183    #[error("transient")]
184    struct Transient;
185
186    /// A store that fails its first `remaining_failures` gets then succeeds,
187    /// counting every `get`, `put`, and `has` call.
188    struct FlakyStore {
189        chunk: AnyChunk<DEFAULT_BODY_SIZE>,
190        remaining_failures: Mutex<u32>,
191        get_calls: AtomicU32,
192        put_calls: AtomicU32,
193        has_calls: AtomicU32,
194    }
195
196    impl FlakyStore {
197        fn new(remaining_failures: u32) -> Self {
198            let chunk = DefaultContentChunk::new("retry probe")
199                .expect("build content chunk")
200                .into();
201            Self {
202                chunk,
203                remaining_failures: Mutex::new(remaining_failures),
204                get_calls: AtomicU32::new(0),
205                put_calls: AtomicU32::new(0),
206                has_calls: AtomicU32::new(0),
207            }
208        }
209    }
210
211    impl ChunkGet<DEFAULT_BODY_SIZE> for FlakyStore {
212        type Error = Transient;
213
214        async fn get(
215            &self,
216            _address: &ChunkAddress,
217        ) -> Result<AnyChunk<DEFAULT_BODY_SIZE>, Self::Error> {
218            self.get_calls.fetch_add(1, Ordering::SeqCst);
219            let mut left = self.remaining_failures.lock().expect("lock");
220            if *left > 0 {
221                *left -= 1;
222                return Err(Transient);
223            }
224            Ok(self.chunk.clone())
225        }
226    }
227
228    impl ChunkPut<DEFAULT_BODY_SIZE> for FlakyStore {
229        type Error = Transient;
230
231        async fn put(&self, _chunk: AnyChunk<DEFAULT_BODY_SIZE>) -> Result<(), Self::Error> {
232            self.put_calls.fetch_add(1, Ordering::SeqCst);
233            Ok(())
234        }
235    }
236
237    impl ChunkHas<DEFAULT_BODY_SIZE> for FlakyStore {
238        async fn has(&self, _address: &ChunkAddress) -> bool {
239            self.has_calls.fetch_add(1, Ordering::SeqCst);
240            true
241        }
242    }
243
244    #[test]
245    fn recovers_when_failures_below_budget() {
246        // 7 failures, then success on the 8th (== max_attempts) get.
247        let store = RetryingChunkGet::with_default(FlakyStore::new(7), NoSleep);
248        let address = *store.inner.chunk.address();
249
250        let got = block_on(store.get(&address)).expect("recovered within budget");
251        assert_eq!(got.address(), &address);
252        assert_eq!(store.inner.get_calls.load(Ordering::SeqCst), 8);
253    }
254
255    #[test]
256    fn propagates_after_exactly_max_attempts() {
257        // Always fails: expect exactly max_attempts gets, then the error.
258        let store = RetryingChunkGet::with_default(FlakyStore::new(u32::MAX), NoSleep);
259        let address = *store.inner.chunk.address();
260
261        let err = block_on(store.get(&address));
262        assert!(err.is_err(), "budget exhausted, error must propagate");
263        assert_eq!(
264            store.inner.get_calls.load(Ordering::SeqCst),
265            RetryConfig::default().max_attempts
266        );
267    }
268
269    #[test]
270    fn put_and_has_are_not_retried() {
271        let store = RetryingChunkGet::with_default(FlakyStore::new(u32::MAX), NoSleep);
272        let address = *store.inner.chunk.address();
273
274        assert!(block_on(store.has(&address)));
275        block_on(store.put(store.inner.chunk.clone())).expect("put delegates");
276        assert_eq!(store.inner.has_calls.load(Ordering::SeqCst), 1);
277        assert_eq!(store.inner.put_calls.load(Ordering::SeqCst), 1);
278    }
279}
nectar_primitives/store/retry.rs

nectar_primitives/store/
retry.rs
