tower_resilience_cache/
shared_layer.rs

1//! Shared cache layer that maintains a single store across all layer() calls.
2
3use crate::events::CacheEvent;
4use crate::eviction::EvictionPolicy;
5use crate::store::CacheStore;
6use crate::{Cache, CacheConfig, KeyExtractor};
7use std::hash::Hash;
8use std::sync::{Arc, Mutex};
9use std::time::Duration;
10use tower::Layer;
11use tower_resilience_core::{EventListeners, FnListener};
12
13/// A Tower [`Layer`] that applies response caching with a shared store.
14///
15/// Unlike [`CacheLayer`](crate::CacheLayer), this layer shares a single cache store
16/// across all services created via [`layer()`](Layer::layer). This is useful when
17/// multiple service instances (e.g., per-session or per-request services) need to
18/// share the same cache.
19///
20/// # Type Parameters
21///
22/// - `Req`: The request type
23/// - `K`: The cache key type (extracted from requests)
24/// - `Resp`: The response type that will be cached
25///
26/// # Examples
27///
28/// ```
29/// use tower_resilience_cache::SharedCacheLayer;
30/// use tower::ServiceBuilder;
31/// use std::time::Duration;
32///
33/// # async fn example() {
34/// // Create a shared cache layer
35/// let cache_layer: SharedCacheLayer<String, String, String> = SharedCacheLayer::builder()
36///     .max_size(100)
37///     .ttl(Duration::from_secs(60))
38///     .key_extractor(|req: &String| req.clone())
39///     .build()
40///     .unwrap();
41///
42/// // Both services share the same cache
43/// let service1 = ServiceBuilder::new()
44///     .layer(cache_layer.clone())
45///     .service(my_service());
46///
47/// let service2 = ServiceBuilder::new()
48///     .layer(cache_layer)
49///     .service(my_service());
50/// # }
51/// # fn my_service() -> impl tower::Service<String, Response = String, Error = std::io::Error> {
52/// #     tower::service_fn(|req: String| async move { Ok::<_, std::io::Error>(req) })
53/// # }
54/// ```
55///
56/// # Creating from CacheLayer
57///
58/// You can also convert an existing [`CacheLayer`](crate::CacheLayer) configuration:
59///
60/// ```
61/// use tower_resilience_cache::CacheLayer;
62/// use std::time::Duration;
63///
64/// let shared_cache = CacheLayer::builder()
65///     .max_size(100)
66///     .ttl(Duration::from_secs(60))
67///     .key_extractor(|req: &String| req.clone())
68///     .build()
69///     .unwrap()
70///     .shared::<String>();  // Specify the response type
71/// ```
72#[derive(Clone)]
73pub struct SharedCacheLayer<Req, K, Resp> {
74    config: Arc<CacheConfig<Req, K>>,
75    store: Arc<Mutex<CacheStore<K, Resp>>>,
76}
77
78impl<Req, K, Resp> SharedCacheLayer<Req, K, Resp>
79where
80    K: Hash + Eq + Clone + Send + 'static,
81    Resp: Clone + Send + 'static,
82{
83    /// Creates a new `SharedCacheLayer` with the given configuration.
84    pub fn new(config: CacheConfig<Req, K>) -> Self {
85        let store = Arc::new(Mutex::new(CacheStore::new(
86            config.max_size,
87            config.ttl,
88            config.eviction_policy,
89        )));
90        Self {
91            config: Arc::new(config),
92            store,
93        }
94    }
95
96    /// Creates a new `SharedCacheLayer` from an existing config Arc.
97    ///
98    /// This is used by [`CacheLayer::shared()`](crate::CacheLayer::shared).
99    pub(crate) fn from_config(config: Arc<CacheConfig<Req, K>>) -> Self {
100        let store = Arc::new(Mutex::new(CacheStore::new(
101            config.max_size,
102            config.ttl,
103            config.eviction_policy,
104        )));
105        Self { config, store }
106    }
107
108    /// Creates a new builder for configuring a shared cache layer.
109    ///
110    /// # Examples
111    ///
112    /// ```
113    /// use tower_resilience_cache::SharedCacheLayer;
114    /// use std::time::Duration;
115    ///
116    /// let layer: SharedCacheLayer<String, String, String> = SharedCacheLayer::builder()
117    ///     .max_size(100)
118    ///     .ttl(Duration::from_secs(60))
119    ///     .key_extractor(|req: &String| req.clone())
120    ///     .build()
121    ///     .unwrap();
122    /// ```
123    pub fn builder() -> SharedCacheConfigBuilder<Req, K, Resp> {
124        SharedCacheConfigBuilder::new()
125    }
126}
127
128impl<S, Req, K, Resp> Layer<S> for SharedCacheLayer<Req, K, Resp>
129where
130    K: Hash + Eq + Clone + Send + 'static,
131    S: tower::Service<Req, Response = Resp>,
132    Resp: Clone + Send + 'static,
133{
134    type Service = Cache<S, Req, K, Resp>;
135
136    fn layer(&self, service: S) -> Self::Service {
137        Cache::with_store(service, Arc::clone(&self.config), Arc::clone(&self.store))
138    }
139}
140
141/// Builder for configuring and constructing a shared cache layer.
142pub struct SharedCacheConfigBuilder<Req, K, Resp> {
143    max_size: usize,
144    ttl: Option<Duration>,
145    eviction_policy: EvictionPolicy,
146    key_extractor: Option<KeyExtractor<Req, K>>,
147    event_listeners: EventListeners<CacheEvent>,
148    name: String,
149    _resp: std::marker::PhantomData<Resp>,
150}
151
152impl<Req, K, Resp> SharedCacheConfigBuilder<Req, K, Resp>
153where
154    K: Hash + Eq + Clone + Send + 'static,
155    Resp: Clone + Send + 'static,
156{
157    /// Creates a new builder with default values.
158    pub fn new() -> Self {
159        Self {
160            max_size: 100,
161            ttl: None,
162            eviction_policy: EvictionPolicy::default(),
163            key_extractor: None,
164            event_listeners: EventListeners::new(),
165            name: String::from("<unnamed>"),
166            _resp: std::marker::PhantomData,
167        }
168    }
169
170    /// Sets the maximum number of entries in the cache.
171    ///
172    /// Default: 100
173    pub fn max_size(mut self, size: usize) -> Self {
174        self.max_size = size;
175        self
176    }
177
178    /// Sets the time-to-live for cached entries.
179    ///
180    /// If set, entries will expire after the specified duration.
181    /// Default: None (no expiration)
182    pub fn ttl(mut self, ttl: Duration) -> Self {
183        self.ttl = Some(ttl);
184        self
185    }
186
187    /// Sets the eviction policy for the cache.
188    ///
189    /// Determines which entry to evict when the cache reaches capacity.
190    ///
191    /// # Options
192    ///
193    /// - `EvictionPolicy::Lru` - Least Recently Used (default)
194    /// - `EvictionPolicy::Lfu` - Least Frequently Used
195    /// - `EvictionPolicy::Fifo` - First In, First Out
196    ///
197    /// Default: `EvictionPolicy::Lru`
198    pub fn eviction_policy(mut self, policy: EvictionPolicy) -> Self {
199        self.eviction_policy = policy;
200        self
201    }
202
203    /// Sets the function that extracts a cache key from a request.
204    ///
205    /// This function must be provided before building.
206    pub fn key_extractor<F>(mut self, f: F) -> Self
207    where
208        F: Fn(&Req) -> K + Send + Sync + 'static,
209    {
210        self.key_extractor = Some(Arc::new(f));
211        self
212    }
213
214    /// Sets the name of this cache instance for observability.
215    ///
216    /// Default: `"<unnamed>"`
217    pub fn name(mut self, name: impl Into<String>) -> Self {
218        self.name = name.into();
219        self
220    }
221
222    /// Registers a callback when a cache hit occurs.
223    pub fn on_hit<F>(mut self, f: F) -> Self
224    where
225        F: Fn() + Send + Sync + 'static,
226    {
227        self.event_listeners.add(FnListener::new(move |event| {
228            if matches!(event, CacheEvent::Hit { .. }) {
229                f();
230            }
231        }));
232        self
233    }
234
235    /// Registers a callback when a cache miss occurs.
236    pub fn on_miss<F>(mut self, f: F) -> Self
237    where
238        F: Fn() + Send + Sync + 'static,
239    {
240        self.event_listeners.add(FnListener::new(move |event| {
241            if matches!(event, CacheEvent::Miss { .. }) {
242                f();
243            }
244        }));
245        self
246    }
247
248    /// Registers a callback when an entry is evicted from the cache.
249    pub fn on_eviction<F>(mut self, f: F) -> Self
250    where
251        F: Fn() + Send + Sync + 'static,
252    {
253        self.event_listeners.add(FnListener::new(move |event| {
254            if matches!(event, CacheEvent::Eviction { .. }) {
255                f();
256            }
257        }));
258        self
259    }
260
261    /// Builds the shared cache layer.
262    ///
263    /// # Errors
264    ///
265    /// Returns [`crate::CacheBuildError::MissingKeyExtractor`] if `key_extractor`
266    /// was not set before calling `build()`.
267    pub fn build(self) -> Result<SharedCacheLayer<Req, K, Resp>, crate::CacheBuildError> {
268        let key_extractor = self
269            .key_extractor
270            .ok_or(crate::CacheBuildError::MissingKeyExtractor)?;
271
272        let config = CacheConfig {
273            max_size: self.max_size,
274            ttl: self.ttl,
275            eviction_policy: self.eviction_policy,
276            key_extractor,
277            event_listeners: self.event_listeners,
278            name: self.name,
279        };
280
281        Ok(SharedCacheLayer::new(config))
282    }
283}
284
285impl<Req, K, Resp> Default for SharedCacheConfigBuilder<Req, K, Resp>
286where
287    K: Hash + Eq + Clone + Send + 'static,
288    Resp: Clone + Send + 'static,
289{
290    fn default() -> Self {
291        Self::new()
292    }
293}
294
295#[cfg(test)]
296mod tests {
297    use super::*;
298    use std::sync::atomic::{AtomicUsize, Ordering};
299    use tower::service_fn;
300    use tower::{Service, ServiceExt};
301
302    #[derive(Clone, Hash, Eq, PartialEq)]
303    struct TestRequest {
304        id: String,
305    }
306
307    #[test]
308    fn test_shared_builder_defaults() {
309        let _layer: SharedCacheLayer<TestRequest, String, String> = SharedCacheLayer::builder()
310            .key_extractor(|req: &TestRequest| req.id.clone())
311            .build()
312            .unwrap();
313    }
314
315    #[test]
316    fn test_shared_builder_custom_values() {
317        let _layer: SharedCacheLayer<TestRequest, String, String> = SharedCacheLayer::builder()
318            .max_size(500)
319            .ttl(Duration::from_secs(60))
320            .key_extractor(|req: &TestRequest| req.id.clone())
321            .name("my-shared-cache")
322            .build()
323            .unwrap();
324    }
325
326    #[test]
327    fn test_shared_builder_errors_without_key_extractor() {
328        let result = SharedCacheLayer::<TestRequest, String, String>::builder().build();
329        assert!(matches!(
330            result,
331            Err(crate::CacheBuildError::MissingKeyExtractor)
332        ));
333    }
334
335    #[tokio::test]
336    async fn test_shared_cache_across_layer_calls() {
337        let call_count = Arc::new(AtomicUsize::new(0));
338        let cc1 = Arc::clone(&call_count);
339        let cc2 = Arc::clone(&call_count);
340
341        // Create two separate services
342        let service1 = service_fn(move |req: String| {
343            let cc = Arc::clone(&cc1);
344            async move {
345                cc.fetch_add(1, Ordering::SeqCst);
346                Ok::<_, std::io::Error>(format!("Response: {}", req))
347            }
348        });
349
350        let service2 = service_fn(move |req: String| {
351            let cc = Arc::clone(&cc2);
352            async move {
353                cc.fetch_add(1, Ordering::SeqCst);
354                Ok::<_, std::io::Error>(format!("Response: {}", req))
355            }
356        });
357
358        // Create a shared cache layer
359        let shared_layer: SharedCacheLayer<String, String, String> = SharedCacheLayer::builder()
360            .max_size(10)
361            .key_extractor(|req: &String| req.clone())
362            .build()
363            .unwrap();
364
365        // Apply to both services
366        let mut wrapped1 = shared_layer.clone().layer(service1);
367        let mut wrapped2 = shared_layer.layer(service2);
368
369        // First call on service1 - cache miss
370        let response1 = wrapped1
371            .ready()
372            .await
373            .unwrap()
374            .call("test".to_string())
375            .await
376            .unwrap();
377        assert_eq!(response1, "Response: test");
378        assert_eq!(call_count.load(Ordering::SeqCst), 1);
379
380        // Call on service2 with same key - should be cache HIT (shared store!)
381        let response2 = wrapped2
382            .ready()
383            .await
384            .unwrap()
385            .call("test".to_string())
386            .await
387            .unwrap();
388        assert_eq!(response2, "Response: test");
389        // Call count should still be 1 because cache was shared
390        assert_eq!(call_count.load(Ordering::SeqCst), 1);
391    }
392
393    #[tokio::test]
394    async fn test_non_shared_cache_layer_creates_separate_stores() {
395        // This test demonstrates the problem that SharedCacheLayer solves
396        use crate::CacheLayer;
397
398        let call_count = Arc::new(AtomicUsize::new(0));
399        let cc1 = Arc::clone(&call_count);
400        let cc2 = Arc::clone(&call_count);
401
402        let service1 = service_fn(move |req: String| {
403            let cc = Arc::clone(&cc1);
404            async move {
405                cc.fetch_add(1, Ordering::SeqCst);
406                Ok::<_, std::io::Error>(format!("Response: {}", req))
407            }
408        });
409
410        let service2 = service_fn(move |req: String| {
411            let cc = Arc::clone(&cc2);
412            async move {
413                cc.fetch_add(1, Ordering::SeqCst);
414                Ok::<_, std::io::Error>(format!("Response: {}", req))
415            }
416        });
417
418        // Regular CacheLayer (not shared)
419        let layer = CacheLayer::builder()
420            .max_size(10)
421            .key_extractor(|req: &String| req.clone())
422            .build()
423            .unwrap();
424
425        // Apply to both services
426        let mut wrapped1 = layer.clone().layer(service1);
427        let mut wrapped2 = layer.layer(service2);
428
429        // First call on service1 - cache miss
430        wrapped1
431            .ready()
432            .await
433            .unwrap()
434            .call("test".to_string())
435            .await
436            .unwrap();
437        assert_eq!(call_count.load(Ordering::SeqCst), 1);
438
439        // Call on service2 with same key - ALSO a cache miss (separate stores!)
440        wrapped2
441            .ready()
442            .await
443            .unwrap()
444            .call("test".to_string())
445            .await
446            .unwrap();
447        // Call count is 2 because stores are NOT shared
448        assert_eq!(call_count.load(Ordering::SeqCst), 2);
449    }
450}
tower_resilience_cache/shared_layer.rs

tower_resilience_cache/
shared_layer.rs
