# Caching

> Implements multi-level caching with strategies for invalidation, stampede prevention, and performance monitoring.

- Skill: `neuralblitz/caching` (Agent Skill)
- Install (CLI): `npx skillmds@latest add neuralblitz/caching`
- Raw SKILL.md: https://api.skillmd.com/api/skills/neuralblitz/caching/raw
- Safety review: PASS (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools, AI & ML, DevOps & Infra, Dev Tooling
- Tags: Cache Invalidation, Cache Stampede, Caching, Distributed Cache, Http Caching, Redis
- License: MIT
- Author: NeuralBlitz (https://skillmd.com/u/neuralblitz)
- Updated: 2026-08-22
- Page: https://skillmd.com/skills/neuralblitz/caching

---

## What I do
- Implement caching at multiple levels
- Choose appropriate cache strategies
- Handle cache invalidation
- Prevent cache stampede
- Use distributed caching effectively
- Monitor cache performance
- Implement cache warming

## When to use me
When implementing caching solutions or optimizing cache performance.

## Multi-Level Caching
```
┌─────────────────────────────────────────────────────────────┐
│                     Request                                │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Browser Cache (HTTP)                         │
│  - Cache-Control headers                                    │
│  - ETag/Last-Modified                                      │
│  - LocalStorage/SessionStorage                              │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                CDN / Edge Cache                              │
│  - CloudFront, Cloudflare                                   │
│  - Static assets, API responses                             │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Application Cache (Memory)                     │
│  - In-memory caches (LRU, LFU)                              │
│  - Per-process or shared memory                             │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Distributed Cache (Redis/Memcached)            │
│  - Shared across instances                                  │
│  - High availability                                        │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                Database Cache                                │
│  - Query cache                                             │
│  - Buffer pool                                             │
└─────────────────────────────────────────────────────────────┘
```

## Cache Strategies
```python
import redis
import json
from typing import Optional, Callable, Any
from functools import wraps
from datetime import timedelta


class CacheService:
    """Multi-level cache service with TTL support."""
    
    def __init__(self, redis_url: str) -> None:
        self.redis = redis.from_url(redis_url)
        self.local_cache = {}
    
    def get(self, key: str) -> Optional[Any]:
        """Get from cache hierarchy."""
        # Check local cache first
        if key in self.local_cache:
            value, expires = self.local_cache[key]
            if expires > datetime.utcnow():
                return value
            del self.local_cache[key]
        
        # Check distributed cache
        cached = self.redis.get(key)
        if cached:
            value = json.loads(cached)
            # Populate local cache
            self.local_cache[key] = (
                value,
                datetime.utcnow() + timedelta(seconds=60)
            )
            return value
        
        return None
    
    def set(
        self,
        key: str,
        value: Any,
        ttl_seconds: int = 3600,
        local_ttl_seconds: int = 60
    ) -> None:
        """Set in both cache layers."""
        # Set distributed cache
        self.redis.setex(key, ttl_seconds, json.dumps(value))
        
        # Set local cache with shorter TTL
        self.local_cache[key] = (
            value,
            datetime.utcnow() + timedelta(seconds=local_ttl_seconds)
        )
    
    def delete(self, key: str) -> None:
        """Delete from both cache layers."""
        self.redis.delete(key)
        self.local_cache.pop(key, None)
    
    def invalidate_pattern(self, pattern: str) -> None:
        """Delete keys matching pattern."""
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)
        # Clear local cache
        for key in list(self.local_cache.keys()):
            if pattern.replace('*', '') in key:
                del self.local_cache[key]


def cached(
    ttl: int = 3600,
    key_prefix: str = ''
):
    """Decorator for caching function results."""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            cache = get_cache_service()
            key = f"{key_prefix}:{func.__name__}:{args}:{kwargs}"
            
            cached_value = cache.get(key)
            if cached_value is not None:
                return cached_value
            
            result = func(*args, **kwargs)
            cache.set(key, result, ttl)
            return result
        return wrapper
    return decorator
```

## Cache Stampede Prevention
```python
import asyncio
import async_timeout


class CacheStampedePreventer:
    """Prevent cache stampede with distributed locking."""
    
    def __init__(
        self,
        cache: CacheService,
        lock_ttl: int = 30
    ) -> None:
        self.cache = cache
        self.lock_ttl = lock_ttl
    
    async def get_or_compute(
        self,
        key: str,
        compute_fn: Callable,
        ttl: int = 3600
    ) -> Any:
        """Get from cache or compute with stampede prevention."""
        # Try cache first
        cached = self.cache.get(key)
        if cached is not None:
            return cached
        
        # Acquire distributed lock
        lock_key = f"lock:{key}"
        lock_acquired = self.cache.redis.set(
            lock_key,
            "locked",
            nx=True,
            ex=self.lock_ttl
        )
        
        if not lock_acquired:
            # Wait for other process to compute
            return await self._wait_for_computation(key, compute_fn)
        
        try:
            # Compute and cache
            result = await compute_fn()
            self.cache.set(key, result, ttl)
            return result
        finally:
            # Release lock
            self.cache.redis.delete(lock_key)
    
    async def _wait_for_computation(
        self,
        key: str,
        compute_fn: Callable
    ) -> Any:
        """Wait for computation to complete."""
        max_wait = 10  # seconds
        
        for _ in range(max_wait * 10):  # Poll every 100ms
            cached = self.cache.get(key)
            if cached is not None:
                return cached
            await asyncio.sleep(0.1)
        
        # If still not cached, compute ourselves
        return await compute_fn()


class ProbabilisticEarlyRecomputation:
    """Prevent expired cache with early recomputation."""
    
    def __init__(
        self,
        cache: CacheService,
        delta_percent: int = 10
    ) -> None:
        self.cache = cache
        self.delta_percent = delta_percent
    
    def get(self, key: str, compute_fn: Callable) -> Any:
        """Get with early recomputation."""
        cached = self.cache.get(key)
        if cached is not None:
            return cached
        
        ttl = self.cache.redis.ttl(key)
        if ttl == -1:  # Key doesn't exist
            return compute_fn()
        
        # If within delta period, recompute in background
        threshold = ttl * (100 - self.delta_percent) / 100
        if ttl < threshold:
            # Could trigger background task here
            pass
        
        return compute_fn()
```

## HTTP Caching
```python
from fastapi import Request, Response
from typing import Optional


class CacheControlMiddleware:
    """Add appropriate cache control headers."""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope['type'] != 'http':
            await self.app(scope, receive, send)
            return
        
        async def send_wrapper(message):
            if message['type'] == 'http.response.start':
                request = Request(scope)
                response = Response(
                    content=b'',
                    status_code=message['status'],
                    headers=dict(message.get('headers', []))
                )
                
                # Set cache headers based on route
                if '/api/' in request.url.path:
                    # API endpoints - no cache or short cache
                    if '/static/' not in request.url.path:
                        response.headers['Cache-Control'] = 'no-cache'
                else:
                    # Static content - long cache
                    response.headers['Cache-Control'] = 'public, max-age=31536000'
                
                message['headers'] = list(response.headers.items())
            
            await send(message)
        
        await self.app(scope, receive, send_wrapper)
```

## Cache Invalidation Patterns
```python
# Event-driven cache invalidation
class CacheInvalidator:
    def __init__(self, cache: CacheService) -> None:
        self.cache = cache
        self.subscriptions = {}
    
    def subscribe(self, event_type: str, callback: Callable) -> None:
        """Subscribe to invalidation events."""
        if event_type not in self.subscriptions:
            self.subscriptions[event_type] = []
        self.subscriptions[event_type].append(callback)
    
    def invalidate(self, event_type: str, data: dict) -> None:
        """Trigger invalidation based on event."""
        if event_type in self.subscriptions:
            for callback in self.subscriptions[event_type]:
                pattern = callback(data)
                if pattern:
                    self.cache.invalidate_pattern(pattern)
    
    def on_user_update(self, user_id: str) -> None:
        """Invalidate user-related caches."""
        patterns = [
            f"user:{user_id}:*",
            f"users:list:*",
            f"stats:user:{user_id}",
        ]
        for pattern in patterns:
            self.cache.invalidate_pattern(pattern)
```

## Cache Monitoring
```python
class CacheMonitor:
    """Monitor cache performance."""
    
    def __init__(self, redis_client: redis.Redis) -> None:
        self.redis = redis_client
    
    def get_stats(self) -> dict:
        """Get cache statistics."""
        info = self.redis.info('stats')
        memory = self.redis.info('memory')
        
        return {
            'hits': info.get('keyspace_hits', 0),
            'misses': info.get('keyspace_misses', 0),
            'hit_rate': self._calculate_hit_rate(info),
            'used_memory': memory.get('used_memory_human'),
            'connected_clients': self.redis.info('clients').get('connected_clients', 0),
            'uptime_seconds': self.redis.info('server').get('uptime_in_seconds', 0),
        }
    
    def _calculate_hit_rate(self, info: dict) -> float:
        hits = info.get('keyspace_hits', 0)
        misses = info.get('keyspace_misses', 0)
        total = hits + misses
        if total == 0:
            return 0.0
        return round(hits / total * 100, 2)
```