Cache Invalidation Strategies for LLM Applications

Overview

Cache invalidation is one of the hardest problems in computer science, and it becomes even more complex with LLM applications. This guide provides comprehensive strategies for managing cache lifecycle, detecting stale content, and implementing intelligent invalidation patterns for Claude Code applications.

The Challenge

LLM caching introduces unique challenges:

• Semantic Equivalence: Different prompts may have the same meaning
• Context Drift: Cached responses may become outdated as context evolves
• Model Updates: New model versions may produce different outputs
• Data Freshness: External data referenced in prompts may change

Strategy Selection Guide

Choosing the right invalidation strategy is critical for balancing performance, cost, and data freshness. Use this guide to select the best approach for your use case.

Strategy	Best For	Complexity	Pros	Cons
Time-Based (TTL)	Simple, non-critical data where some staleness is acceptable.	Low	Easy to implement, predictable.	Can be inefficient, evicting data too early or too late.
Event-Based	Applications where data dependencies are well-defined (e.g., file changes, database updates).	Medium	Precise, invalidates only what’s necessary, efficient.	Requires robust event infrastructure, complex dependency tracking.
Content-Based	Caching responses to prompts that reference external, changing content.	High	Highly accurate, avoids unnecessary invalidations.	Computationally expensive, requires hashing/similarity checks.
Intelligent (ML)	Large-scale systems with complex access patterns where optimal efficiency is required.	Very High	Adapts to usage, maximizes cache hit ratio, minimizes stale data.	Requires ML model training, complex feature engineering, and continuous monitoring.

Recommendation Flow:

1. Start with TTL: It’s the simplest and often sufficient.
2. Add Event-Based: If you have clear data sources and dependencies, layer this on for more precise control.
3. Use Content-Based: For critical prompts that depend on the content of files or data sources, use content-based checks to ensure freshness.
4. Consider Intelligent Caching: If you operate at a large scale and have the resources, an ML-based approach can provide the highest level of optimization.

For a practical implementation combining these, see the LayeredInvalidation class under Best Practices.

Invalidation Strategies

1. Time-Based Invalidation (TTL)

from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import hashlib
import json
 
class TTLCache:
    """Time-based cache invalidation with dynamic TTL adjustment."""
    
    def __init__(self):
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.default_ttl = 300  # 5 minutes
        self.max_ttl = 3600    # 1 hour
        self.min_ttl = 60      # 1 minute
        
    def set(
        self, 
        key: str, 
        value: Any, 
        ttl: Optional[int] = None,
        content_type: Optional[str] = None
    ):
        """Set cache entry with TTL."""
        if ttl is None:
            ttl = self._calculate_dynamic_ttl(content_type, value)
            
        self.cache[key] = {
            "value": value,
            "expires_at": datetime.utcnow() + timedelta(seconds=ttl),
            "created_at": datetime.utcnow(),
            "ttl": ttl,
            "access_count": 0,
            "last_accessed": datetime.utcnow(),
            "content_type": content_type
        }
    
    def get(self, key: str) -> Optional[Any]:
        """Get cache entry if not expired."""
        if key not in self.cache:
            return None
            
        entry = self.cache[key]
        now = datetime.utcnow()
        
        # Check expiration
        if now > entry["expires_at"]:
            del self.cache[key]
            return None
        
        # Update access stats
        entry["access_count"] += 1
        entry["last_accessed"] = now
        
        # Extend TTL for frequently accessed items
        if entry["access_count"] > 5:
            new_ttl = min(entry["ttl"] * 1.5, self.max_ttl)
            entry["expires_at"] = now + timedelta(seconds=new_ttl)
            entry["ttl"] = new_ttl
        
        return entry["value"]
    
    def _calculate_dynamic_ttl(self, content_type: str, value: Any) -> int:
        """Calculate TTL based on content characteristics."""
        if content_type == "system_prompt":
            return self.max_ttl  # System prompts rarely change
        elif content_type == "user_context":
            return 600  # 10 minutes for user context
        elif content_type == "real_time_data":
            return self.min_ttl  # 1 minute for real-time data
        elif content_type == "document":
            # Longer TTL for larger documents
            doc_size = len(str(value))
            if doc_size > 10000:
                return 1800  # 30 minutes for large docs
            else:
                return 900   # 15 minutes for smaller docs
        else:
            return self.default_ttl
    
    def invalidate_pattern(self, pattern: str):
        """Invalidate all keys matching pattern."""
        keys_to_delete = [
            key for key in self.cache.keys() 
            if pattern in key
        ]
        for key in keys_to_delete:
            del self.cache[key]
        return len(keys_to_delete)
    
    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics."""
        now = datetime.utcnow()
        total_entries = len(self.cache)
        
        if total_entries == 0:
            return {"total_entries": 0}
        
        access_counts = [entry["access_count"] for entry in self.cache.values()]
        ttls = [entry["ttl"] for entry in self.cache.values()]
        ages = [(now - entry["created_at"]).total_seconds() for entry in self.cache.values()]
        
        return {
            "total_entries": total_entries,
            "avg_access_count": sum(access_counts) / total_entries,
            "avg_ttl": sum(ttls) / total_entries,
            "avg_age_seconds": sum(ages) / total_entries,
            "most_accessed": max(self.cache.items(), 
                                key=lambda x: x[1]["access_count"])[0] if self.cache else None
        }

2. Event-Based Invalidation

import { EventEmitter } from 'events';
import crypto from 'crypto';
 
interface CacheEntry {
  key: string;
  value: any;
  dependencies: string[];
  checksum: string;
  created: Date;
  version: number;
}
 
interface InvalidationEvent {
  type: 'data_update' | 'model_update' | 'dependency_change' | 'manual';
  source: string;
  targets?: string[];
  cascade?: boolean;
}
 
class EventBasedCache extends EventEmitter {
  private cache: Map<string, CacheEntry> = new Map();
  private dependencies: Map<string, Set<string>> = new Map();
  private version = 1;
  
  constructor() {
    super();
    this.setupEventHandlers();
  }
  
  private setupEventHandlers() {
    // Listen for invalidation events
    this.on('invalidate', (event: InvalidationEvent) => {
      this.handleInvalidation(event);
    });
    
    // Listen for model updates
    this.on('model_update', () => {
      this.invalidateAll('model_update');
    });
    
    // Listen for data updates
    this.on('data_update', (sources: string[]) => {
      this.invalidateDependents(sources);
    });
  }
  
  set(
    key: string, 
    value: any, 
    dependencies: string[] = []
  ): void {
    const checksum = this.generateChecksum(value);
    
    const entry: CacheEntry = {
      key,
      value,
      dependencies,
      checksum,
      created: new Date(),
      version: this.version
    };
    
    this.cache.set(key, entry);
    
    // Update dependency tracking
    for (const dep of dependencies) {
      if (!this.dependencies.has(dep)) {
        this.dependencies.set(dep, new Set());
      }
      this.dependencies.get(dep)!.add(key);
    }
    
    this.emit('cache_set', { key, dependencies });
  }
  
  get(key: string): any | null {
    const entry = this.cache.get(key);
    
    if (!entry) {
      return null;
    }
    
    // Check version
    if (entry.version < this.version) {
      this.cache.delete(key);
      return null;
    }
    
    return entry.value;
  }
  
  private handleInvalidation(event: InvalidationEvent) {
    console.log(`Handling invalidation: ${event.type} from ${event.source}`);
    
    switch (event.type) {
      case 'data_update':
        if (event.targets) {
          this.invalidateDependents(event.targets);
        }
        break;
        
      case 'model_update':
        this.version++;
        // Optionally clear all cache
        if (event.cascade) {
          this.cache.clear();
          this.dependencies.clear();
        }
        break;
        
      case 'dependency_change':
        if (event.targets) {
          for (const target of event.targets) {
            this.invalidateKey(target);
          }
        }
        break;
        
      case 'manual':
        if (event.targets) {
          for (const target of event.targets) {
            this.invalidateKey(target);
          }
        } else if (event.cascade) {
          this.cache.clear();
        }
        break;
    }
  }
  
  private invalidateDependents(sources: string[]) {
    const keysToInvalidate = new Set<string>();
    
    for (const source of sources) {
      const dependents = this.dependencies.get(source);
      if (dependents) {
        dependents.forEach(key => keysToInvalidate.add(key));
      }
    }
    
    for (const key of keysToInvalidate) {
      this.invalidateKey(key);
    }
    
    console.log(`Invalidated ${keysToInvalidate.size} dependent entries`);
  }
  
  private invalidateKey(key: string) {
    const entry = this.cache.get(key);
    if (entry) {
      // Remove from dependencies
      for (const dep of entry.dependencies) {
        const deps = this.dependencies.get(dep);
        if (deps) {
          deps.delete(key);
          if (deps.size === 0) {
            this.dependencies.delete(dep);
          }
        }
      }
      
      this.cache.delete(key);
      this.emit('cache_invalidated', { key });
    }
  }
  
  private invalidateAll(reason: string) {
    const count = this.cache.size;
    this.cache.clear();
    this.dependencies.clear();
    console.log(`Invalidated all ${count} cache entries due to: ${reason}`);
  }
  
  private generateChecksum(value: any): string {
    const content = JSON.stringify(value);
    return crypto.createHash('md5').update(content).digest('hex');
  }
  
  // Dependency analysis
  analyzeDependencies(): Map<string, number> {
    const dependencyCount = new Map<string, number>();
    
    for (const [dep, keys] of this.dependencies) {
      dependencyCount.set(dep, keys.size);
    }
    
    return dependencyCount;
  }
  
  // Find circular dependencies
  findCircularDependencies(): string[][] {
    const cycles: string[][] = [];
    const visited = new Set<string>();
    const recursionStack = new Set<string>();
    
    const dfs = (key: string, path: string[]): boolean => {
      visited.add(key);
      recursionStack.add(key);
      
      const entry = this.cache.get(key);
      if (entry) {
        for (const dep of entry.dependencies) {
          if (!visited.has(dep)) {
            if (dfs(dep, [...path, dep])) {
              return true;
            }
          } else if (recursionStack.has(dep)) {
            // Found cycle
            const cycleStart = path.indexOf(dep);
            cycles.push(path.slice(cycleStart));
            return true;
          }
        }
      }
      
      recursionStack.delete(key);
      return false;
    };
    
    for (const key of this.cache.keys()) {
      if (!visited.has(key)) {
        dfs(key, [key]);
      }
    }
    
    return cycles;
  }
}

3. Content-Based Invalidation

import hashlib
from typing import Dict, List, Optional, Tuple
from datetime import datetime
import difflib
 
class ContentBasedCache:
    """Cache with content-based invalidation and similarity detection."""
    
    def __init__(self, similarity_threshold: float = 0.95):
        self.cache: Dict[str, Dict] = {}
        self.content_hashes: Dict[str, str] = {}
        self.similarity_threshold = similarity_threshold
        self.invalidation_log: List[Dict] = []
        
    def set(self, key: str, content: str, response: str, metadata: Dict = None):
        """Set cache entry with content hash."""
        content_hash = self._hash_content(content)
        
        # Check if content has changed
        if key in self.content_hashes:
            old_hash = self.content_hashes[key]
            if old_hash != content_hash:
                self._log_invalidation(key, "content_changed", {
                    "old_hash": old_hash,
                    "new_hash": content_hash
                })
        
        self.cache[key] = {
            "content": content,
            "response": response,
            "content_hash": content_hash,
            "created_at": datetime.utcnow(),
            "metadata": metadata or {},
            "access_count": 0
        }
        
        self.content_hashes[key] = content_hash
    
    def get(self, key: str, current_content: str) -> Optional[str]:
        """Get cache entry with content validation."""
        if key not in self.cache:
            return None
        
        entry = self.cache[key]
        current_hash = self._hash_content(current_content)
        
        # Exact match
        if current_hash == entry["content_hash"]:
            entry["access_count"] += 1
            return entry["response"]
        
        # Check similarity
        similarity = self._calculate_similarity(entry["content"], current_content)
        
        if similarity >= self.similarity_threshold:
            entry["access_count"] += 1
            return entry["response"]
        else:
            # Content has changed too much
            self._log_invalidation(key, "content_drift", {
                "similarity": similarity,
                "threshold": self.similarity_threshold
            })
            del self.cache[key]
            return None
    
    def validate_all(self) -> Dict[str, bool]:
        """Validate all cache entries."""
        validation_results = {}
        
        for key, entry in list(self.cache.items()):
            # Check if metadata indicates staleness
            if "expires_at" in entry["metadata"]:
                if datetime.utcnow() > entry["metadata"]["expires_at"]:
                    self._log_invalidation(key, "expired", {
                        "expired_at": entry["metadata"]["expires_at"]
                    })
                    del self.cache[key]
                    validation_results[key] = False
                    continue
            
            # Check if content source has been updated
            if "source_version" in entry["metadata"]:
                current_version = self._get_source_version(entry["metadata"].get("source"))
                if current_version != entry["metadata"]["source_version"]:
                    self._log_invalidation(key, "source_updated", {
                        "old_version": entry["metadata"]["source_version"],
                        "new_version": current_version
                    })
                    del self.cache[key]
                    validation_results[key] = False
                    continue
            
            validation_results[key] = True
        
        return validation_results
    
    def _hash_content(self, content: str) -> str:
        """Generate hash for content."""
        return hashlib.sha256(content.encode()).hexdigest()
    
    def _calculate_similarity(self, content1: str, content2: str) -> float:
        """Calculate similarity between two contents."""
        # Use sequence matcher for similarity
        matcher = difflib.SequenceMatcher(None, content1, content2)
        return matcher.ratio()
    
    def _get_source_version(self, source: Optional[str]) -> Optional[str]:
        """Get current version of a content source."""
        # This would connect to your data source to check version
        # For example, checking file modification time, DB version, etc.
        # Placeholder implementation
        return "v1.0"
    
    def _log_invalidation(self, key: str, reason: str, details: Dict):
        """Log invalidation event."""
        self.invalidation_log.append({
            "timestamp": datetime.utcnow(),
            "key": key,
            "reason": reason,
            "details": details
        })
    
    def get_invalidation_stats(self) -> Dict[str, any]:
        """Get invalidation statistics."""
        if not self.invalidation_log:
            return {"total_invalidations": 0}
        
        reasons = {}
        for event in self.invalidation_log:
            reason = event["reason"]
            reasons[reason] = reasons.get(reason, 0) + 1
        
        return {
            "total_invalidations": len(self.invalidation_log),
            "by_reason": reasons,
            "recent_invalidations": self.invalidation_log[-10:]
        }
    
    def suggest_optimizations(self) -> List[str]:
        """Suggest cache optimizations based on invalidation patterns."""
        suggestions = []
        stats = self.get_invalidation_stats()
        
        if stats["total_invalidations"] == 0:
            return ["Cache is performing well with no invalidations"]
        
        by_reason = stats.get("by_reason", {})
        
        if by_reason.get("content_drift", 0) > 10:
            suggestions.append(
                "High content drift detected. Consider lowering similarity threshold "
                "or implementing semantic hashing."
            )
        
        if by_reason.get("expired", 0) > by_reason.get("content_changed", 0):
            suggestions.append(
                "Many entries expiring before content changes. "
                "Consider extending TTL for stable content."
            )
        
        if by_reason.get("source_updated", 0) > 5:
            suggestions.append(
                "Frequent source updates detected. Implement incremental caching "
                "or subscribe to update notifications."
            )
        
        return suggestions

4. Intelligent Invalidation with ML

interface CacheMetrics {
  hitRate: number;
  avgAge: number;
  invalidationRate: number;
  costSavings: number;
}
 
interface PredictionFeatures {
  contentLength: number;
  lastAccessTime: number;
  accessFrequency: number;
  contentComplexity: number;
  userSegment: string;
  timeOfDay: number;
  dayOfWeek: number;
}
 
class IntelligentCacheManager {
  private cache: Map<string, any> = new Map();
  private metrics: Map<string, CacheMetrics> = new Map();
  private accessPatterns: Map<string, number[]> = new Map();
  private invalidationModel: any; // ML model placeholder
  
  constructor() {
    // Initialize with pre-trained model or rules
    this.invalidationModel = this.initializeModel();
    
    // Periodic optimization
    setInterval(() => this.optimizeCache(), 300000); // Every 5 minutes
  }
  
  async set(key: string, value: any, context: any) {
    const features = this.extractFeatures(key, value, context);
    const predictedLifetime = await this.predictLifetime(features);
    
    this.cache.set(key, {
      value,
      features,
      predictedLifetime,
      actualLifetime: 0,
      created: Date.now(),
      lastAccessed: Date.now(),
      accessCount: 0
    });
    
    // Initialize metrics
    if (!this.metrics.has(key)) {
      this.metrics.set(key, {
        hitRate: 0,
        avgAge: 0,
        invalidationRate: 0,
        costSavings: 0
      });
    }
  }
  
  get(key: string): any | null {
    const entry = this.cache.get(key);
    
    if (!entry) {
      this.recordMiss(key);
      return null;
    }
    
    const now = Date.now();
    const age = now - entry.created;
    
    // Check if should invalidate based on prediction
    if (this.shouldInvalidate(entry, age)) {
      this.invalidate(key, 'predicted_stale');
      return null;
    }
    
    // Update access patterns
    entry.lastAccessed = now;
    entry.accessCount++;
    entry.actualLifetime = age;
    
    this.recordHit(key, age);
    
    return entry.value;
  }
  
  private shouldInvalidate(entry: any, age: number): boolean {
    // Use prediction model
    const staleProbability = this.calculateStaleProbability(entry, age);
    
    // Dynamic threshold based on access patterns
    const threshold = this.calculateInvalidationThreshold(entry);
    
    return staleProbability > threshold;
  }
  
  private calculateStaleProbability(entry: any, age: number): number {
    // Simplified probability calculation
    // In practice, this would use the ML model
    
    const ageRatio = age / entry.predictedLifetime;
    const accessDecay = Math.exp(-entry.accessCount / 10);
    
    // Content-specific factors
    let contentFactor = 1.0;
    if (entry.features.contentComplexity > 0.8) {
      contentFactor = 0.8; // Complex content changes less
    }
    
    // Time-based factors
    const timeOfDay = new Date().getHours();
    const timeFactor = timeOfDay >= 9 && timeOfDay <= 17 ? 1.2 : 0.9;
    
    return Math.min(1.0, ageRatio * accessDecay * contentFactor * timeFactor);
  }
  
  private calculateInvalidationThreshold(entry: any): number {
    // Dynamic threshold based on cost-benefit analysis
    const metrics = this.metrics.get(entry.key) || {
      hitRate: 0.5,
      costSavings: 0
    };
    
    // Higher threshold for high-value cache entries
    if (metrics.costSavings > 10) {
      return 0.9; // Only invalidate when very likely stale
    } else if (metrics.hitRate > 0.8) {
      return 0.8; // Popular entries get more leeway
    } else {
      return 0.6; // Default threshold
    }
  }
  
  private extractFeatures(key: string, value: any, context: any): PredictionFeatures {
    const content = JSON.stringify(value);
    const now = new Date();
    
    return {
      contentLength: content.length,
      lastAccessTime: Date.now(),
      accessFrequency: this.getAccessFrequency(key),
      contentComplexity: this.calculateComplexity(content),
      userSegment: context.userSegment || 'default',
      timeOfDay: now.getHours(),
      dayOfWeek: now.getDay()
    };
  }
  
  private calculateComplexity(content: string): number {
    // Simple complexity measure
    const uniqueWords = new Set(content.split(/\s+/));
    const complexity = uniqueWords.size / content.length;
    return Math.min(1.0, complexity * 100);
  }
  
  private getAccessFrequency(key: string): number {
    const pattern = this.accessPatterns.get(key);
    if (!pattern || pattern.length === 0) return 0;
    
    // Calculate access frequency over last hour
    const hourAgo = Date.now() - 3600000;
    const recentAccesses = pattern.filter(t => t > hourAgo).length;
    return recentAccesses;
  }
  
  private async predictLifetime(features: PredictionFeatures): Promise<number> {
    // Simplified prediction
    // In practice, this would use the trained model
    
    let baseLifetime = 300000; // 5 minutes
    
    // Adjust based on features
    if (features.contentComplexity > 0.7) {
      baseLifetime *= 2; // Complex content is more stable
    }
    
    if (features.accessFrequency > 10) {
      baseLifetime *= 1.5; // Popular content should stay longer
    }
    
    // Time-based adjustments
    if (features.timeOfDay >= 0 && features.timeOfDay <= 6) {
      baseLifetime *= 2; // Less activity at night
    }
    
    return baseLifetime;
  }
  
  private optimizeCache() {
    // Analyze cache performance and adjust strategies
    const performanceReport = this.analyzePerformance();
    
    // Remove underperforming entries
    for (const [key, metrics] of this.metrics) {
      if (metrics.hitRate < 0.1 && metrics.costSavings < 1) {
        this.invalidate(key, 'underperforming');
      }
    }
    
    // Adjust prediction model based on actual vs predicted lifetimes
    this.updateModel();
    
    console.log('Cache optimization complete:', performanceReport);
  }
  
  private analyzePerformance() {
    let totalHits = 0;
    let totalMisses = 0;
    let totalSavings = 0;
    
    for (const metrics of this.metrics.values()) {
      totalHits += metrics.hitRate * 100; // Rough approximation
      totalSavings += metrics.costSavings;
    }
    
    return {
      overallHitRate: totalHits / (totalHits + totalMisses),
      totalSavings: totalSavings.toFixed(2),
      cacheSize: this.cache.size,
      avgInvalidationRate: this.calculateAvgInvalidationRate()
    };
  }
  
  private calculateAvgInvalidationRate(): number {
    if (this.metrics.size === 0) return 0;
    
    let totalRate = 0;
    for (const metrics of this.metrics.values()) {
      totalRate += metrics.invalidationRate;
    }
    
    return totalRate / this.metrics.size;
  }
  
  private updateModel() {
    // Collect training data from actual vs predicted lifetimes
    const trainingData: any[] = [];
    
    for (const [key, entry] of this.cache) {
      if (entry.actualLifetime > 0) {
        trainingData.push({
          features: entry.features,
          actualLifetime: entry.actualLifetime,
          predictedLifetime: entry.predictedLifetime
        });
      }
    }
    
    // Update model with new data
    // This is where you'd retrain or fine-tune your model
    console.log(`Model update: ${trainingData.length} samples collected`);
  }
  
  private recordHit(key: string, age: number) {
    const metrics = this.metrics.get(key)!;
    metrics.hitRate = (metrics.hitRate * 0.9) + 0.1; // Exponential moving average
    metrics.avgAge = (metrics.avgAge * 0.9) + (age * 0.1);
    metrics.costSavings += 0.01; // Simplified cost calculation
    
    // Record access pattern
    if (!this.accessPatterns.has(key)) {
      this.accessPatterns.set(key, []);
    }
    this.accessPatterns.get(key)!.push(Date.now());
  }
  
  private recordMiss(key: string) {
    if (this.metrics.has(key)) {
      const metrics = this.metrics.get(key)!;
      metrics.hitRate = metrics.hitRate * 0.9; // Decay hit rate
    }
  }
  
  private invalidate(key: string, reason: string) {
    this.cache.delete(key);
    
    if (this.metrics.has(key)) {
      const metrics = this.metrics.get(key)!;
      metrics.invalidationRate = (metrics.invalidationRate * 0.9) + 0.1;
    }
    
    console.log(`Cache invalidated: ${key} (${reason})`);
  }
  
  private initializeModel() {
    // Initialize with simple rules or pre-trained model
    return {
      predict: (features: PredictionFeatures) => {
        // Placeholder for actual model
        return 300000; // 5 minutes default
      }
    };
  }
}

Best Practices

1. Layered Invalidation

class LayeredInvalidation:
    """Implement multiple invalidation strategies in layers."""
    
    def __init__(self):
        self.ttl_cache = TTLCache()
        self.content_cache = ContentBasedCache()
        self.event_cache = EventBasedCache()
        
    def get(self, key: str, content: str) -> Optional[str]:
        # Check TTL first (fastest)
        result = self.ttl_cache.get(key)
        if result is None:
            return None
            
        # Validate content hasn't changed
        result = self.content_cache.get(key, content)
        if result is None:
            self.ttl_cache.invalidate_pattern(key)
            return None
            
        return result

2. Batch Invalidation

class BatchInvalidator {
  private pendingInvalidations: Set<string> = new Set();
  private batchInterval: NodeJS.Timeout;
  
  constructor(private cache: Map<string, any>) {
    // Process invalidations in batches
    this.batchInterval = setInterval(() => {
      this.processBatch();
    }, 1000); // Every second
  }
  
  scheduleInvalidation(key: string) {
    this.pendingInvalidations.add(key);
  }
  
  private processBatch() {
    if (this.pendingInvalidations.size === 0) return;
    
    const batch = Array.from(this.pendingInvalidations);
    this.pendingInvalidations.clear();
    
    console.log(`Processing ${batch.length} invalidations`);
    
    for (const key of batch) {
      this.cache.delete(key);
    }
  }
}

3. Soft Invalidation

class SoftInvalidation:
    """Mark entries as stale instead of deleting immediately."""
    
    def __init__(self):
        self.cache = {}
        self.stale_entries = set()
        
    def mark_stale(self, key: str):
        """Mark entry as stale without deleting."""
        if key in self.cache:
            self.stale_entries.add(key)
            
    def get(self, key: str, allow_stale: bool = False) -> Optional[Any]:
        """Get entry with optional stale allowance."""
        if key not in self.cache:
            return None
            
        is_stale = key in self.stale_entries
        
        if is_stale and not allow_stale:
            # Refresh in background
            self._schedule_refresh(key)
            return None
            
        return {
            "value": self.cache[key],
            "is_stale": is_stale
        }

Monitoring and Metrics

class InvalidationMetrics:
    """Track invalidation patterns and effectiveness."""
    
    def __init__(self):
        self.invalidations = []
        self.false_positives = 0
        self.false_negatives = 0
        
    def record_invalidation(self, key: str, reason: str, was_stale: bool):
        self.invalidations.append({
            "timestamp": datetime.utcnow(),
            "key": key,
            "reason": reason,
            "was_stale": was_stale
        })
        
        if not was_stale:
            self.false_positives += 1
            
    def analyze_effectiveness(self) -> Dict[str, float]:
        if not self.invalidations:
            return {"effectiveness": 0}
            
        total = len(self.invalidations)
        correct = sum(1 for inv in self.invalidations if inv["was_stale"])
        
        return {
            "effectiveness": correct / total,
            "false_positive_rate": self.false_positives / total,
            "invalidations_per_hour": total / 24,  # Simplified
            "top_reasons": self._get_top_reasons()
        }
        
    def _get_top_reasons(self) -> Dict[str, int]:
        reasons = {}
        for inv in self.invalidations:
            reason = inv["reason"]
            reasons[reason] = reasons.get(reason, 0) + 1
        return dict(sorted(reasons.items(), key=lambda x: x[1], reverse=True)[:5])

Conclusion

Effective cache invalidation requires:

1. Multiple Strategies: Combine TTL, event-based, and content-based approaches
2. Intelligent Prediction: Use ML to predict when cache entries become stale
3. Continuous Monitoring: Track metrics to optimize invalidation strategies
4. Graceful Degradation: Implement soft invalidation and stale-while-revalidate patterns

The key is finding the right balance between cache hit rate and data freshness for your specific use case.