elif_cache/
invalidation.rs

1//! Cache invalidation strategies
2//! 
3//! This module provides various strategies for cache invalidation including
4//! pattern-based, tag-based, time-based and event-driven invalidation.
5
6use crate::{CacheBackend, CacheError, CacheResult, CacheKey, CacheTag};
7use async_trait::async_trait;
8use futures::future::join_all;
9use regex::Regex;
10use wildmatch::WildMatch;
11use std::{
12    collections::{HashMap, HashSet},
13    sync::Arc,
14    time::{Duration, SystemTime, UNIX_EPOCH},
15};
16use tokio::sync::RwLock;
17use tracing;
18
19/// Pattern matching strategy for cache key invalidation
20#[derive(Debug, Clone)]
21pub enum PatternStrategy {
22    /// Simple wildcard matching (*, ?)
23    Wildcard(String),
24    /// Full regex pattern matching
25    Regex(String),
26    /// Prefix matching
27    Prefix(String),
28    /// Suffix matching
29    Suffix(String),
30    /// Contains matching
31    Contains(String),
32}
33
34impl PatternStrategy {
35    /// Check if a key matches this pattern
36    pub fn matches(&self, key: &str) -> CacheResult<bool> {
37        match self {
38            PatternStrategy::Wildcard(pattern) => {
39                let matcher = WildMatch::new(pattern);
40                Ok(matcher.matches(key))
41            }
42            PatternStrategy::Regex(pattern) => {
43                let regex = Regex::new(pattern)
44                    .map_err(|e| CacheError::Configuration(format!("Invalid regex: {}", e)))?;
45                Ok(regex.is_match(key))
46            }
47            PatternStrategy::Prefix(prefix) => {
48                Ok(key.starts_with(prefix))
49            }
50            PatternStrategy::Suffix(suffix) => {
51                Ok(key.ends_with(suffix))
52            }
53            PatternStrategy::Contains(substring) => {
54                Ok(key.contains(substring))
55            }
56        }
57    }
58}
59
60
61/// Time-based invalidation policies
62#[derive(Debug, Clone)]
63pub enum TimeBasedPolicy {
64    /// Absolute expiration time
65    ExpiresAt(SystemTime),
66    /// Time-to-live duration
67    Ttl(Duration),
68    /// Scheduled expiration (cron-like, simplified)
69    Scheduled {
70        /// Interval between expirations
71        interval: Duration,
72        /// Last execution time
73        last_run: SystemTime,
74    },
75}
76
77impl TimeBasedPolicy {
78    /// Check if this policy indicates expiration
79    pub fn should_expire(&self, created_at: SystemTime) -> bool {
80        let now = SystemTime::now();
81        
82        match self {
83            TimeBasedPolicy::ExpiresAt(expires_at) => now >= *expires_at,
84            TimeBasedPolicy::Ttl(ttl) => {
85                if let Ok(elapsed) = now.duration_since(created_at) {
86                    elapsed >= *ttl
87                } else {
88                    false
89                }
90            }
91            TimeBasedPolicy::Scheduled { interval, last_run } => {
92                if let Ok(elapsed) = now.duration_since(*last_run) {
93                    elapsed >= *interval
94                } else {
95                    false
96                }
97            }
98        }
99    }
100}
101
102/// Cache entry with invalidation metadata
103#[derive(Debug, Clone)]
104pub struct InvalidatableEntry {
105    /// The cache key
106    pub key: CacheKey,
107    /// Associated tags
108    pub tags: HashSet<CacheTag>,
109    /// When this entry was created
110    pub created_at: SystemTime,
111    /// Time-based invalidation policies
112    pub time_policies: Vec<TimeBasedPolicy>,
113    /// Dependencies on other cache keys
114    pub dependencies: HashSet<CacheKey>,
115}
116
117impl InvalidatableEntry {
118    pub fn new(key: CacheKey) -> Self {
119        Self {
120            key,
121            tags: HashSet::new(),
122            created_at: SystemTime::now(),
123            time_policies: Vec::new(),
124            dependencies: HashSet::new(),
125        }
126    }
127    
128    /// Check if this entry should be invalidated based on time policies
129    pub fn should_expire(&self) -> bool {
130        self.time_policies.iter().any(|policy| policy.should_expire(self.created_at))
131    }
132    
133    /// Add a tag to this entry
134    pub fn add_tag(&mut self, tag: CacheTag) {
135        self.tags.insert(tag);
136    }
137    
138    /// Add multiple tags to this entry
139    pub fn add_tags(&mut self, tags: impl IntoIterator<Item = CacheTag>) {
140        self.tags.extend(tags);
141    }
142    
143    /// Add a time-based policy
144    pub fn add_time_policy(&mut self, policy: TimeBasedPolicy) {
145        self.time_policies.push(policy);
146    }
147    
148    /// Add a dependency on another key
149    pub fn add_dependency(&mut self, key: CacheKey) {
150        self.dependencies.insert(key);
151    }
152}
153
154/// Advanced invalidation manager
155#[async_trait]
156pub trait InvalidationManager: Send + Sync {
157    /// Invalidate cache entries matching a pattern
158    async fn invalidate_pattern(&self, pattern: PatternStrategy) -> CacheResult<Vec<CacheKey>>;
159    
160    /// Invalidate cache entries by tags
161    async fn invalidate_tags(&self, tags: &[&str]) -> CacheResult<Vec<CacheKey>>;
162    
163    /// Invalidate expired entries based on time policies
164    async fn invalidate_expired(&self) -> CacheResult<Vec<CacheKey>>;
165    
166    /// Invalidate entries dependent on given keys
167    async fn invalidate_dependencies(&self, keys: &[&str]) -> CacheResult<Vec<CacheKey>>;
168    
169    /// Get all keys tracked by this manager
170    async fn get_tracked_keys(&self) -> CacheResult<Vec<CacheKey>>;
171    
172    /// Register a cache entry for invalidation tracking
173    async fn register_entry(&self, entry: InvalidatableEntry) -> CacheResult<()>;
174    
175    /// Unregister a cache entry
176    async fn unregister_entry(&self, key: &str) -> CacheResult<()>;
177}
178
179/// In-memory invalidation manager implementation
180pub struct MemoryInvalidationManager<B: CacheBackend> {
181    backend: Arc<B>,
182    entries: Arc<RwLock<HashMap<CacheKey, InvalidatableEntry>>>,
183}
184
185impl<B: CacheBackend> MemoryInvalidationManager<B> {
186    pub fn new(backend: Arc<B>) -> Self {
187        Self {
188            backend,
189            entries: Arc::new(RwLock::new(HashMap::new())),
190        }
191    }
192    
193    /// Get keys matching a pattern
194    async fn find_matching_keys(&self, pattern: &PatternStrategy) -> CacheResult<Vec<CacheKey>> {
195        let entries = self.entries.read().await;
196        let mut matching_keys = Vec::new();
197        
198        for key in entries.keys() {
199            if pattern.matches(key)? {
200                matching_keys.push(key.clone());
201            }
202        }
203        
204        Ok(matching_keys)
205    }
206    
207    /// Get keys associated with any of the given tags
208    async fn find_tagged_keys(&self, tags: &[&str]) -> CacheResult<Vec<CacheKey>> {
209        let entries = self.entries.read().await;
210        let mut matching_keys = Vec::new();
211        let tag_set: HashSet<&str> = tags.iter().copied().collect();
212        
213        for (key, entry) in entries.iter() {
214            if entry.tags.iter().any(|tag| tag_set.contains(tag.as_str())) {
215                matching_keys.push(key.clone());
216            }
217        }
218        
219        Ok(matching_keys)
220    }
221    
222    /// Get expired keys based on time policies
223    async fn find_expired_keys(&self) -> CacheResult<Vec<CacheKey>> {
224        let entries = self.entries.read().await;
225        let mut expired_keys = Vec::new();
226        
227        for (key, entry) in entries.iter() {
228            if entry.should_expire() {
229                expired_keys.push(key.clone());
230            }
231        }
232        
233        Ok(expired_keys)
234    }
235    
236    /// Get keys dependent on the given keys
237    async fn find_dependent_keys(&self, dependency_keys: &[&str]) -> CacheResult<Vec<CacheKey>> {
238        let entries = self.entries.read().await;
239        let mut dependent_keys = Vec::new();
240        let dep_set: HashSet<&str> = dependency_keys.iter().copied().collect();
241        
242        for (key, entry) in entries.iter() {
243            if entry.dependencies.iter().any(|dep| dep_set.contains(dep.as_str())) {
244                dependent_keys.push(key.clone());
245            }
246        }
247        
248        Ok(dependent_keys)
249    }
250    
251    /// Actually invalidate the given keys from cache and tracking
252    /// Uses parallel execution for better performance with network backends like Redis
253    async fn execute_invalidation(&self, keys: Vec<CacheKey>) -> CacheResult<Vec<CacheKey>> {
254        if keys.is_empty() {
255            return Ok(Vec::new());
256        }
257
258        // Parallelize backend forget operations for better performance
259        let forget_futures = keys.iter().map(|k| self.backend.forget(k));
260        let results = join_all(forget_futures).await;
261
262        // Collect successfully invalidated keys and handle any errors gracefully
263        let mut invalidated_keys = Vec::new();
264        let mut failed_keys = Vec::new();
265        
266        for (i, result) in results.into_iter().enumerate() {
267            match result {
268                Ok(was_removed) => {
269                    if was_removed {
270                        invalidated_keys.push(keys[i].clone());
271                    }
272                }
273                Err(_) => {
274                    failed_keys.push(keys[i].clone());
275                }
276            }
277        }
278
279        // Always remove from tracking, even if backend removal failed
280        // This prevents inconsistent state where tracking believes keys exist but backend doesn't
281        {
282            let mut entries = self.entries.write().await;
283            for key in &keys {
284                entries.remove(key);
285            }
286        }
287
288        // If there were backend failures, log them but don't fail the entire operation
289        if !failed_keys.is_empty() {
290            tracing::warn!(
291                "Failed to invalidate {} keys from backend: {:?}", 
292                failed_keys.len(), 
293                failed_keys
294            );
295        }
296
297        Ok(invalidated_keys)
298    }
299}
300
301#[async_trait]
302impl<B: CacheBackend> InvalidationManager for MemoryInvalidationManager<B> {
303    async fn invalidate_pattern(&self, pattern: PatternStrategy) -> CacheResult<Vec<CacheKey>> {
304        let matching_keys = self.find_matching_keys(&pattern).await?;
305        self.execute_invalidation(matching_keys).await
306    }
307    
308    async fn invalidate_tags(&self, tags: &[&str]) -> CacheResult<Vec<CacheKey>> {
309        let tagged_keys = self.find_tagged_keys(tags).await?;
310        self.execute_invalidation(tagged_keys).await
311    }
312    
313    async fn invalidate_expired(&self) -> CacheResult<Vec<CacheKey>> {
314        let expired_keys = self.find_expired_keys().await?;
315        self.execute_invalidation(expired_keys).await
316    }
317    
318    async fn invalidate_dependencies(&self, keys: &[&str]) -> CacheResult<Vec<CacheKey>> {
319        let dependent_keys = self.find_dependent_keys(keys).await?;
320        self.execute_invalidation(dependent_keys).await
321    }
322    
323    async fn get_tracked_keys(&self) -> CacheResult<Vec<CacheKey>> {
324        let entries = self.entries.read().await;
325        Ok(entries.keys().cloned().collect())
326    }
327    
328    async fn register_entry(&self, entry: InvalidatableEntry) -> CacheResult<()> {
329        let mut entries = self.entries.write().await;
330        entries.insert(entry.key.clone(), entry);
331        Ok(())
332    }
333    
334    async fn unregister_entry(&self, key: &str) -> CacheResult<()> {
335        let mut entries = self.entries.write().await;
336        entries.remove(key);
337        Ok(())
338    }
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344    use crate::backends::MemoryBackend;
345    use crate::config::CacheConfig;
346    use std::time::Duration;
347    use tokio::time::sleep;
348    
349    #[tokio::test]
350    async fn test_pattern_strategy_wildcard() {
351        let pattern = PatternStrategy::Wildcard("user:*".to_string());
352        assert!(pattern.matches("user:123").unwrap());
353        assert!(pattern.matches("user:abc").unwrap());
354        assert!(!pattern.matches("post:123").unwrap());
355        
356        let pattern = PatternStrategy::Wildcard("user:?".to_string());
357        assert!(pattern.matches("user:1").unwrap());
358        assert!(!pattern.matches("user:12").unwrap());
359        assert!(!pattern.matches("user:").unwrap());
360    }
361    
362    #[tokio::test]
363    async fn test_pattern_strategy_regex() {
364        let pattern = PatternStrategy::Regex(r"user:\d+".to_string());
365        assert!(pattern.matches("user:123").unwrap());
366        assert!(!pattern.matches("user:abc").unwrap());
367        assert!(!pattern.matches("post:123").unwrap());
368    }
369    
370    #[tokio::test]
371    async fn test_pattern_strategy_prefix_suffix_contains() {
372        let prefix = PatternStrategy::Prefix("user:".to_string());
373        assert!(prefix.matches("user:123").unwrap());
374        assert!(!prefix.matches("post:123").unwrap());
375        
376        let suffix = PatternStrategy::Suffix(":active".to_string());
377        assert!(suffix.matches("user:active").unwrap());
378        assert!(!suffix.matches("user:inactive").unwrap());
379        
380        let contains = PatternStrategy::Contains("temp".to_string());
381        assert!(contains.matches("temp_cache").unwrap());
382        assert!(contains.matches("cache_temp").unwrap());
383        assert!(!contains.matches("permanent").unwrap());
384    }
385    
386    #[tokio::test]
387    async fn test_wildcard_match() {
388        // Test using PatternStrategy::Wildcard which now uses wildmatch
389        let wildcard_all = PatternStrategy::Wildcard("*".to_string());
390        assert!(wildcard_all.matches("anything").unwrap());
391        
392        let user_wildcard = PatternStrategy::Wildcard("user:*".to_string());
393        assert!(user_wildcard.matches("user:123").unwrap());
394        assert!(user_wildcard.matches("user:").unwrap());
395        assert!(!user_wildcard.matches("post:123").unwrap());
396        
397        let user_question = PatternStrategy::Wildcard("user:?".to_string());
398        assert!(user_question.matches("user:1").unwrap());
399        assert!(!user_question.matches("user:12").unwrap());
400        assert!(!user_question.matches("user:").unwrap());
401        
402        let temp_wildcard = PatternStrategy::Wildcard("*temp*".to_string());
403        assert!(temp_wildcard.matches("temporary").unwrap());
404        assert!(temp_wildcard.matches("temp").unwrap());
405        assert!(temp_wildcard.matches("something_temp_else").unwrap());
406        assert!(!temp_wildcard.matches("permanent").unwrap());
407    }
408    
409    #[tokio::test]
410    async fn test_time_based_policy() {
411        let now = SystemTime::now();
412        let past = now - Duration::from_secs(60);
413        let future = now + Duration::from_secs(60);
414        
415        let expires_at_future = TimeBasedPolicy::ExpiresAt(future);
416        assert!(!expires_at_future.should_expire(past));
417        
418        let expires_at_past = TimeBasedPolicy::ExpiresAt(past);
419        assert!(expires_at_past.should_expire(past));
420        
421        let ttl_expired = TimeBasedPolicy::Ttl(Duration::from_secs(30));
422        assert!(ttl_expired.should_expire(past));
423        
424        let ttl_valid = TimeBasedPolicy::Ttl(Duration::from_secs(120));
425        assert!(!ttl_valid.should_expire(past));
426    }
427    
428    #[tokio::test]
429    async fn test_invalidatable_entry() {
430        let mut entry = InvalidatableEntry::new("test:key".to_string());
431        
432        entry.add_tag("users".to_string());
433        entry.add_tags(vec!["active".to_string(), "premium".to_string()]);
434        
435        assert!(entry.tags.contains("users"));
436        assert!(entry.tags.contains("active"));
437        assert!(entry.tags.contains("premium"));
438        assert_eq!(entry.tags.len(), 3);
439        
440        entry.add_time_policy(TimeBasedPolicy::Ttl(Duration::from_millis(1)));
441        sleep(Duration::from_millis(2)).await;
442        assert!(entry.should_expire());
443        
444        entry.add_dependency("parent:key".to_string());
445        assert!(entry.dependencies.contains("parent:key"));
446    }
447    
448    #[tokio::test]
449    async fn test_memory_invalidation_manager() {
450        let backend = Arc::new(MemoryBackend::new(CacheConfig::default()));
451        let manager = MemoryInvalidationManager::new(backend.clone());
452        
453        // Put some test data in cache
454        backend.put("user:1", b"data1".to_vec(), None).await.unwrap();
455        backend.put("user:2", b"data2".to_vec(), None).await.unwrap();
456        backend.put("post:1", b"data3".to_vec(), None).await.unwrap();
457        
458        // Register entries with manager
459        let mut entry1 = InvalidatableEntry::new("user:1".to_string());
460        entry1.add_tag("users".to_string());
461        manager.register_entry(entry1).await.unwrap();
462        
463        let mut entry2 = InvalidatableEntry::new("user:2".to_string());
464        entry2.add_tag("users".to_string());
465        manager.register_entry(entry2).await.unwrap();
466        
467        let mut entry3 = InvalidatableEntry::new("post:1".to_string());
468        entry3.add_tag("posts".to_string());
469        manager.register_entry(entry3).await.unwrap();
470        
471        // Test pattern invalidation
472        let invalidated = manager
473            .invalidate_pattern(PatternStrategy::Wildcard("user:*".to_string()))
474            .await
475            .unwrap();
476        assert_eq!(invalidated.len(), 2);
477        assert!(invalidated.contains(&"user:1".to_string()));
478        assert!(invalidated.contains(&"user:2".to_string()));
479        
480        // Verify cache entries are gone
481        assert_eq!(backend.get("user:1").await.unwrap(), None);
482        assert_eq!(backend.get("user:2").await.unwrap(), None);
483        assert_eq!(backend.get("post:1").await.unwrap(), Some(b"data3".to_vec()));
484        
485        // Test tag invalidation
486        let invalidated = manager.invalidate_tags(&["posts"]).await.unwrap();
487        assert_eq!(invalidated.len(), 1);
488        assert!(invalidated.contains(&"post:1".to_string()));
489        
490        // Verify all cache entries are gone
491        assert_eq!(backend.get("post:1").await.unwrap(), None);
492    }
493    
494    #[tokio::test]
495    async fn test_dependency_invalidation() {
496        let backend = Arc::new(MemoryBackend::new(CacheConfig::default()));
497        let manager = MemoryInvalidationManager::new(backend.clone());
498        
499        // Put some test data in cache
500        backend.put("parent", b"parent_data".to_vec(), None).await.unwrap();
501        backend.put("child1", b"child1_data".to_vec(), None).await.unwrap();
502        backend.put("child2", b"child2_data".to_vec(), None).await.unwrap();
503        backend.put("independent", b"independent_data".to_vec(), None).await.unwrap();
504        
505        // Register entries with dependencies
506        let parent_entry = InvalidatableEntry::new("parent".to_string());
507        manager.register_entry(parent_entry).await.unwrap();
508        
509        let mut child1_entry = InvalidatableEntry::new("child1".to_string());
510        child1_entry.add_dependency("parent".to_string());
511        manager.register_entry(child1_entry).await.unwrap();
512        
513        let mut child2_entry = InvalidatableEntry::new("child2".to_string());
514        child2_entry.add_dependency("parent".to_string());
515        manager.register_entry(child2_entry).await.unwrap();
516        
517        let independent_entry = InvalidatableEntry::new("independent".to_string());
518        manager.register_entry(independent_entry).await.unwrap();
519        
520        // Test dependency invalidation
521        let invalidated = manager
522            .invalidate_dependencies(&["parent"])
523            .await
524            .unwrap();
525        
526        assert_eq!(invalidated.len(), 2);
527        assert!(invalidated.contains(&"child1".to_string()));
528        assert!(invalidated.contains(&"child2".to_string()));
529        
530        // Verify dependent entries are gone but parent and independent remain
531        assert_eq!(backend.get("child1").await.unwrap(), None);
532        assert_eq!(backend.get("child2").await.unwrap(), None);
533        assert_eq!(backend.get("parent").await.unwrap(), Some(b"parent_data".to_vec()));
534        assert_eq!(backend.get("independent").await.unwrap(), Some(b"independent_data".to_vec()));
535    }
536    
537    #[tokio::test]
538    async fn test_time_based_invalidation() {
539        let backend = Arc::new(MemoryBackend::new(CacheConfig::default()));
540        let manager = MemoryInvalidationManager::new(backend.clone());
541        
542        // Put some test data in cache
543        backend.put("temp1", b"temp1_data".to_vec(), None).await.unwrap();
544        backend.put("temp2", b"temp2_data".to_vec(), None).await.unwrap();
545        backend.put("permanent", b"permanent_data".to_vec(), None).await.unwrap();
546        
547        // Register entries with time policies
548        let mut temp1_entry = InvalidatableEntry::new("temp1".to_string());
549        temp1_entry.add_time_policy(TimeBasedPolicy::Ttl(Duration::from_millis(1)));
550        manager.register_entry(temp1_entry).await.unwrap();
551        
552        let mut temp2_entry = InvalidatableEntry::new("temp2".to_string());
553        temp2_entry.add_time_policy(TimeBasedPolicy::Ttl(Duration::from_millis(1)));
554        manager.register_entry(temp2_entry).await.unwrap();
555        
556        let permanent_entry = InvalidatableEntry::new("permanent".to_string());
557        manager.register_entry(permanent_entry).await.unwrap();
558        
559        // Wait for expiration
560        sleep(Duration::from_millis(2)).await;
561        
562        // Test time-based invalidation
563        let invalidated = manager.invalidate_expired().await.unwrap();
564        assert_eq!(invalidated.len(), 2);
565        assert!(invalidated.contains(&"temp1".to_string()));
566        assert!(invalidated.contains(&"temp2".to_string()));
567        
568        // Verify expired entries are gone but permanent remains
569        assert_eq!(backend.get("temp1").await.unwrap(), None);
570        assert_eq!(backend.get("temp2").await.unwrap(), None);
571        assert_eq!(backend.get("permanent").await.unwrap(), Some(b"permanent_data".to_vec()));
572    }
573    
574    #[tokio::test]
575    async fn test_parallel_invalidation_performance() {
576        let backend = Arc::new(MemoryBackend::new(CacheConfig::default()));
577        let manager = MemoryInvalidationManager::new(backend.clone());
578        
579        // Put a large number of entries
580        for i in 0..100 {
581            backend.put(&format!("batch_key_{}", i), b"data".to_vec(), None).await.unwrap();
582            
583            let mut entry = InvalidatableEntry::new(format!("batch_key_{}", i));
584            entry.add_tag("batch".to_string());
585            manager.register_entry(entry).await.unwrap();
586        }
587        
588        // Test parallel invalidation with pattern
589        let start_time = std::time::Instant::now();
590        let invalidated = manager
591            .invalidate_pattern(PatternStrategy::Wildcard("batch_key_*".to_string()))
592            .await
593            .unwrap();
594        let duration = start_time.elapsed();
595        
596        // All keys should be invalidated
597        assert_eq!(invalidated.len(), 100);
598        
599        // Parallel execution should be reasonably fast (this is a rough check)
600        // In practice, with network backends like Redis, the performance improvement would be more significant
601        assert!(duration.as_millis() < 1000, "Parallel invalidation took too long: {:?}", duration);
602        
603        // Verify all keys are gone
604        for i in 0..100 {
605            assert_eq!(backend.get(&format!("batch_key_{}", i)).await.unwrap(), None);
606        }
607    }
608}
elif_cache/invalidation.rs

elif_cache/
invalidation.rs
