mytheclipse_cache/
auto_refresh.rs

1//! Auto-refresh cache wrapper that proactively refreshes stale entries in
2//! the background, eliminating thundering-herd on cache miss.
3
4use std::sync::Arc;
5use std::time::Duration;
6use tokio::sync::Mutex;
7
8use crate::traits::Cache;
9use crate::CacheError;
10
11/// A cache wrapper that refreshes entries in the background before they expire.
12///
13/// When a `get` returns a `None`, the wrapper triggers a background refresh
14/// (via `refresh_fn`) while still returning the miss to the caller.
15pub struct AutoRefreshCache<C, F, Fut>
16where
17    C: Cache + Clone + Send + Sync + 'static,
18    F: Fn(String) -> Fut + Send + Sync + 'static,
19    Fut: std::future::Future<Output = Result<Vec<u8>, CacheError>> + Send + 'static,
20{
21    inner: C,
22    refresh_fn: Arc<F>,
23    refresh_after: Duration,
24    refreshing: Arc<Mutex<std::collections::HashSet<String>>>,
25}
26
27impl<C, F, Fut> AutoRefreshCache<C, F, Fut>
28where
29    C: Cache + Clone + Send + Sync + 'static,
30    F: Fn(String) -> Fut + Send + Sync + 'static,
31    Fut: std::future::Future<Output = Result<Vec<u8>, CacheError>> + Send + 'static,
32{
33    /// Creates a new auto-refresh wrapper.
34    pub fn new(inner: C, refresh_fn: F, refresh_after: Duration) -> Self {
35        Self {
36            inner,
37            refresh_fn: Arc::new(refresh_fn),
38            refresh_after,
39            refreshing: Arc::new(Mutex::new(std::collections::HashSet::new())),
40        }
41    }
42
43    /// Gets a value, triggering a background refresh if the entry is a miss.
44    pub async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, CacheError> {
45        let result = self.inner.get(key).await?;
46        if result.is_none() {
47            let key_str = key.to_string();
48            let mut refreshing = self.refreshing.lock().await;
49            if refreshing.insert(key_str.clone()) {
50                let inner = self.inner.clone();
51                let refresh_fn = Arc::clone(&self.refresh_fn);
52                let refresh_after = self.refresh_after;
53                let refreshing = self.refreshing.clone();
54                tokio::spawn(async move {
55                    let refresh_fut = refresh_fn(key_str.clone());
56                    match refresh_fut.await {
57                        Ok(value) => {
58                            let ttl = Some(refresh_after * 2);
59                            let _ = inner.set(&key_str, value, ttl).await;
60                        }
61                        Err(e) => {
62                            tracing::warn!("background refresh failed for key {}: {}", key_str, e);
63                        }
64                    }
65                    let mut r = refreshing.lock().await;
66                    r.remove(&key_str);
67                });
68            }
69        }
70        Ok(result)
71    }
72
73    /// Sets a value in the underlying cache.
74    pub async fn set(
75        &self,
76        key: &str,
77        value: Vec<u8>,
78        ttl: Option<Duration>,
79    ) -> Result<(), CacheError> {
80        self.inner.set(key, value, ttl).await
81    }
82
83    /// Invalidates a key in the underlying cache.
84    pub async fn invalidate(&self, key: &str) -> Result<(), CacheError> {
85        self.inner.invalidate(key).await
86    }
87}
mytheclipse_cache/auto_refresh.rs

mytheclipse_cache/
auto_refresh.rs
