matrix_sdk_common/
failures_cache.rs

1// Copyright 2023 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! A TTL cache which can be used to time out repeated operations that might
16//! experience intermittent failures.
17
18use std::{borrow::Borrow, collections::HashMap, hash::Hash, sync::Arc, time::Duration};
19
20use ruma::time::Instant;
21
22use super::locks::RwLock;
23
24const MAX_DELAY: u64 = 15 * 60;
25const MULTIPLIER: u64 = 15;
26
27/// A TTL cache where items get inactive instead of discarded.
28///
29/// The items need to be explicitly removed from the cache. This allows us to
30/// implement exponential backoff based TTL.
31#[derive(Clone, Debug)]
32pub struct FailuresCache<T: Eq + Hash> {
33    inner: Arc<InnerCache<T>>,
34}
35
36#[derive(Debug)]
37struct InnerCache<T: Eq + Hash> {
38    max_delay: Duration,
39    backoff_multiplier: u64,
40    items: RwLock<HashMap<T, FailuresItem>>,
41}
42
43impl<T: Eq + Hash> Default for InnerCache<T> {
44    fn default() -> Self {
45        Self {
46            max_delay: Duration::from_secs(MAX_DELAY),
47            backoff_multiplier: MULTIPLIER,
48            items: Default::default(),
49        }
50    }
51}
52
53#[derive(Debug, Clone, Copy)]
54struct FailuresItem {
55    insertion_time: Instant,
56    duration: Duration,
57
58    /// Number of times that this item has failed after it was first added to
59    /// the cache. (In other words, one less than the total number of failures.)
60    failure_count: u8,
61}
62
63impl FailuresItem {
64    /// Has the item expired.
65    fn expired(&self) -> bool {
66        self.insertion_time.elapsed() >= self.duration
67    }
68
69    /// Force the expiry of this item.
70    ///
71    /// This doesn't reset the failure count, but does mark the item as ready
72    /// for immediate retry.
73    fn expire(&mut self) {
74        self.duration = Duration::from_secs(0);
75    }
76}
77
78impl<T> FailuresCache<T>
79where
80    T: Eq + Hash,
81{
82    pub fn new() -> Self {
83        Self { inner: Default::default() }
84    }
85
86    pub fn with_settings(max_delay: Duration, multiplier: u8) -> Self {
87        Self {
88            inner: InnerCache {
89                max_delay,
90                backoff_multiplier: multiplier.into(),
91                items: Default::default(),
92            }
93            .into(),
94        }
95    }
96
97    /// Is the given key non-expired and part of the cache.
98    pub fn contains<Q>(&self, key: &Q) -> bool
99    where
100        T: Borrow<Q>,
101        Q: Hash + Eq + ?Sized,
102    {
103        self.inner.items.read().get(key).is_some_and(|item| !item.expired())
104    }
105
106    /// Get the failure count for a given key.
107    ///
108    /// # Returns
109    ///
110    /// - `None` if this key is not in the failure cache. (It has never failed,
111    ///   or it has been [`FailuresCache::remove()`]d since the last failure.)
112    ///
113    /// - `Some(u8)`: the number of times it has failed since it was first added
114    ///   to the failure cache. (In other words, one less than the total number
115    ///   of failures.)
116    pub fn failure_count<Q>(&self, key: &Q) -> Option<u8>
117    where
118        T: Borrow<Q>,
119        Q: Hash + Eq + ?Sized,
120    {
121        self.inner.items.read().get(key).map(|i| i.failure_count)
122    }
123
124    /// This will calculate a duration that determines how long an item is
125    /// considered to be valid while being in the cache.
126    ///
127    /// The returned duration will follow this sequence if the default
128    /// multiplier and `max_delay` values are used, values are in minutes:
129    /// [0.25, 0.5, 1.0, 2.0, 4.0, 8.0, 15.0]
130    fn calculate_delay(&self, failure_count: u8) -> Duration {
131        let exponential_backoff = 2u64.saturating_pow(failure_count.into());
132        let delay = exponential_backoff.saturating_mul(self.inner.backoff_multiplier);
133
134        Duration::from_secs(delay).clamp(Duration::from_secs(1), self.inner.max_delay)
135    }
136
137    /// Add a single item to the cache.
138    pub fn insert(&self, item: T) {
139        self.extend([item]);
140    }
141
142    /// Extend the cache with the given iterator of items.
143    ///
144    /// Items that are already part of the cache, whether they are expired or
145    /// not, will have their TTL extended using an exponential backoff
146    /// algorithm.
147    pub fn extend(&self, iterator: impl IntoIterator<Item = T>) {
148        let mut lock = self.inner.items.write();
149
150        let now = Instant::now();
151
152        for key in iterator {
153            let failure_count = if let Some(value) = lock.get(&key) {
154                value.failure_count.saturating_add(1)
155            } else {
156                0
157            };
158
159            let delay = self.calculate_delay(failure_count);
160
161            let item = FailuresItem { insertion_time: now, duration: delay, failure_count };
162
163            lock.insert(key, item);
164        }
165    }
166
167    /// Remove the items contained in the iterator from the cache.
168    pub fn remove<'a, I, Q>(&'a self, iterator: I)
169    where
170        I: Iterator<Item = &'a Q>,
171        T: Borrow<Q>,
172        Q: Hash + Eq + 'a + ?Sized,
173    {
174        let mut lock = self.inner.items.write();
175
176        for item in iterator {
177            lock.remove(item);
178        }
179    }
180
181    /// Force the expiry of the given item, if it is present in the cache.
182    ///
183    /// This doesn't reset the failure count, but does mark the item as ready
184    /// for immediate retry.
185    #[doc(hidden)]
186    pub fn expire(&self, item: &T) {
187        self.inner.items.write().get_mut(item).map(FailuresItem::expire);
188    }
189}
190
191impl<T: Eq + Hash> Default for FailuresCache<T> {
192    fn default() -> Self {
193        Self::new()
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use std::time::Duration;
200
201    use proptest::prelude::*;
202
203    use super::FailuresCache;
204
205    #[test]
206    fn failures_cache() {
207        let cache = FailuresCache::new();
208
209        assert!(!cache.contains(&1));
210        cache.extend([1u8].iter());
211        assert!(cache.contains(&1));
212
213        cache.inner.items.write().get_mut(&1).unwrap().duration = Duration::from_secs(0);
214        assert!(!cache.contains(&1));
215
216        cache.remove([1u8].iter());
217        assert!(cache.inner.items.read().get(&1).is_none())
218    }
219
220    #[test]
221    fn failures_cache_timeout() {
222        let cache: FailuresCache<u8> = FailuresCache::new();
223
224        assert_eq!(cache.calculate_delay(0).as_secs(), 15);
225        assert_eq!(cache.calculate_delay(1).as_secs(), 30);
226        assert_eq!(cache.calculate_delay(2).as_secs(), 60);
227        assert_eq!(cache.calculate_delay(3).as_secs(), 120);
228        assert_eq!(cache.calculate_delay(4).as_secs(), 240);
229        assert_eq!(cache.calculate_delay(5).as_secs(), 480);
230        assert_eq!(cache.calculate_delay(6).as_secs(), 900);
231        assert_eq!(cache.calculate_delay(7).as_secs(), 900);
232    }
233
234    proptest! {
235        #[test]
236        fn failures_cache_proptest_timeout(count in 0..10u8) {
237            let cache: FailuresCache<u8> = FailuresCache::new();
238            let delay = cache.calculate_delay(count).as_secs();
239
240            assert!(delay <= 900);
241            assert!(delay >= 15);
242        }
243    }
244}
matrix_sdk_common/failures_cache.rs

matrix_sdk_common/
failures_cache.rs
