Or: How I Learned to Stop Worrying and Love the Hit Ratio

Remember the old saying? "There are only two hard things in Computer Science: cache invalidation and naming things." Everyone laughs. Nobody fixes it.

I once reviewed a cost audit of a $12,000/month Redis cluster caching user profiles. The hit ratio was 85%. Sounds impressive until you do the math: a cache miss on a simple indexed PostgreSQL query costs 0.5ms (Redis lookup) + 0.8ms (database query) = 1.3ms. Without caching, it's just 0.8ms directly to the database. The cache added 0.5ms overhead on every request. At 1000 requests/second with 85% hits, that's 150 misses/second × 0.5ms = 75ms/second of wasted latency. The $12,000/month infrastructure was saving < 2ms on the p95 latency. The infrastructure cost per millisecond saved: $6,000.

Like Total Recall—you think you're getting into your dreams, but you're just paying for the experience. The engineer managing that cache? Still checking hit ratios at 3 AM, wondering why everything feels slow.

This is a guide for people who actually have to live in production. Not theory. Not benchmarks from tech talks. Real numbers, real tradeoffs, and honest accounting of whether the cache is actually worth it.

The Fundamental Problem

Caching is seductive. It promises speed. It promises reduced database load. It promises that your weekend won't be interrupted by alerts. What it actually does is shift complexity around—it doesn't remove it.

The real question isn't "should we cache?" It's "what are we caching, for how long, for whom, and what breaks when the cache lies?"

Get that wrong, and you've built a system that's faster at serving stale data than your unoptimized alternative. You're trying to escape the matrix, but you're still in it—just a more expensive version.

Tier 1: In-Memory Cache (The Simple Lie)

Let's start with the seductive simplicity: a Python dict with lazy expiration on access.

import time
from typing import Any, Callable, Optional, TypeVar

T = TypeVar('T')

class SimpleMemoryCache:
    """
    Single-worker in-memory cache with lazy expiration.
    
    This is genuinely useful when you want isolated caching
    per worker and don't care about global state. It's also
    dangerously simple.
    """
    def __init__(self):
        self._cache: dict[str, tuple[Any, float]] = {}
    
    def get(self, key: str) -> Optional[Any]:
        if key not in self._cache:
            return None
        
        value, expiry = self._cache[key]
        
        # Lazy expiration: only check when accessed
        if time.time() > expiry:
            del self._cache[key]
            return None
        
        return value
    
    def set(self, key: str, value: Any, ttl_seconds: int = 300):
        """Cache a value with TTL in seconds."""
        self._cache[key] = (value, time.time() + ttl_seconds)
    
    def delete(self, key: str):
        self._cache.pop(key, None)


# As a decorator
_simple_cache = SimpleMemoryCache()

def simple_cache(ttl_seconds: int = 300):
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        def wrapper(*args, **kwargs) -> T:
            # Warning: This assumes args/kwargs are hashable
            # Real world? Prepare for pain with unhashable types
            try:
                cache_key = f"{func.__name__}:{args}:{kwargs}"
            except TypeError:
                # Unhashable argument, just call the function
                return func(*args, **kwargs)
            
            cached = _simple_cache.get(cache_key)
            if cached is not None:
                return cached
            
            result = func(*args, **kwargs)
            _simple_cache.set(cache_key, result, ttl_seconds)
            return result
        
        return wrapper
    return decorator

# Usage
@simple_cache(ttl_seconds=60)
def get_user_profile(user_id: int):
    # Simulating expensive operation
    time.sleep(2)
    return {"user_id": user_id, "name": "Alice"}

print(get_user_profile(1))  # 2 seconds
print(get_user_profile(1))  # ~0ms (cached)

When this works:

• Single-threaded services where each request is independent
• Microservice where caching locally reduces external API calls
• Distributed systems where you want each worker to have its own cache and don't care about consistency
• Cheap data: if a cache miss just triggers a DB query, who cares?

When this breaks spectacularly:

• Multi-threaded code. Race conditions everywhere. The dict is thread-safe for reads, but your logic isn't.
• Web servers with multiple worker processes (gunicorn, uvicorn with multiple workers). Each process has its own dict. No sharing. High miss ratio. You just added overhead.
• Memory leaks. Lazy expiration means dead keys stay in memory until someone asks for them. If you have 50,000 unique cache keys and access only 10,000 daily, you're bleeding memory.
• You'll have engineers arguing about TTL values. They want 1 hour. You know it should be 5 minutes. Nobody wins.

The honest truth: This pattern is useful for protecting against the same request coming in twice within milliseconds. For anything else, it's a trap. Like Ghostbusters crossing the streams—sometimes your simple solution creates chaos when you scale it.

Tier 1.5: The Lambda Exception (Simple Dict with 15-Minute Boundary)

There's one specific case where simple dict caching is not just acceptable—it's the right choice: AWS Lambdas.

Lambdas are ephemeral. A Lambda function lives for 15 minutes (default, configurable). The runtime persists across invocations during that lifecycle window. Within that window, a simple in-memory dict is perfect: no network latency, no SDK overhead, no Redis dependency.

# Lambda-specific caching strategy
import json
from time import time
from typing import Any, Optional, Callable, TypeVar

T = TypeVar('T')

class LambdaMemoryCache:
    """
    Cache for Lambda functions.
    
    - Lives as long as the Lambda execution environment lives (15 min default)
    - No external dependencies (no Redis SDK to import and initialize)
    - Trades memory for latency and simplicity
    - When Lambda is recycled, cache is gone (intentional)
    """
    def __init__(self):
        self._cache: dict[str, tuple[Any, float]] = {}
    
    def get(self, key: str) -> Optional[Any]:
        if key not in self._cache:
            return None
        value, expiry = self._cache[key]
        if time() > expiry:
            del self._cache[key]
            return None
        return value
    
    def set(self, key: str, value: Any, ttl_seconds: int = 900):  # Default 15 min
        self._cache[key] = (value, time() + ttl_seconds)


# Singleton for the Lambda lifecycle
_lambda_cache = LambdaMemoryCache()

def lambda_cache(ttl_seconds: int = 900):
    """Decorator for Lambda-friendly caching."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        def wrapper(*args, **kwargs) -> T:
            # For Lambdas, args are usually simple (user_id, config key, etc.)
            try:
                cache_key = f"{func.__name__}:{args}:{sorted(kwargs.items())}"
            except TypeError:
                return func(*args, **kwargs)
            
            cached = _lambda_cache.get(cache_key)
            if cached is not None:
                return cached
            
            result = func(*args, **kwargs)
            _lambda_cache.set(cache_key, result, ttl_seconds)
            return result
        
        return wrapper
    return decorator


# Real-world Lambda example
@lambda_cache(ttl_seconds=900)
def fetch_dynamodb_config(environment: str):
    """
    Within a 15-minute Lambda lifecycle, if the same environment
    config is requested multiple times, we get it from the dict.
    
    Without caching: every invocation pays network cost to DynamoDB.
    With caching: subsequent calls in the same execution context are free.
    """
    import boto3
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('config')
    response = table.get_item(Key={'environment': environment})
    return response.get('Item', {})


def lambda_handler(event, context):
    """
    If this Lambda is invoked 5 times in 15 minutes with the same environment,
    it caches the config fetch 4 times.
    """
    env = event.get('environment', 'dev')
    config = fetch_dynamodb_config(env)
    
    # More processing...
    return {'statusCode': 200, 'body': json.dumps(config)}

Why this makes sense:

• No external dependency. Redis client library isn't loaded. No network calls to Redis. Startup time: milliseconds.
• Bounded lifetime. The cache dies when the Lambda dies. No memory management issues.
• True intra-process caching. Multiple invocations of the same Lambda can reuse the same execution context. AWS keeps Lambdas warm for cost reasons—you benefit.
• Low latency. A dict lookup is microseconds. Redis is milliseconds.

The catch:

• Lottery-based reuse. You don't control whether AWS reuses your execution environment. AWS might recycle it after 1 invocation or 100. You can't rely on it.
• Memory limits. Lambda has 128MB-10GB memory limit. Cache too much, and you hit the limit.
• No sharing across Lambda instances. 10 concurrent Lambda invocations? 10 separate caches. No cross-instance coherence.

When to use it:

• Expensive initialization (DB connections, API credentials): Cache once, reuse across invocations
• Frequently-accessed reference data within a single request cycle
• Avoiding redundant external API calls within the 15-minute window
• You want to skip the operational overhead of Redis

When to not use it:

• If you need consistency across instances (use DynamoDB or ElastiCache instead)
• If you need persistence beyond 15 minutes
• If the memory footprint is significant

The honest play: This is the "good enough" cache for Lambda workloads. It's not sophisticated. It's not guaranteed. But it's simple, has zero operational overhead, and works for the constraints of Lambda.

The Exception: Pandas DataFrame Caching

Here's where simple dict caching actually shines: expensive pandas operations with short TTLs.

If you're querying a dataset, doing transformations, and that same result gets requested multiple times within a 15-minute window, a simple in-memory cache avoids the entire Redis ecosystem.

import pandas as pd
from datetime import datetime, timedelta
from typing import Optional

class PandasCache:
    """
    Simple cache specifically for pandas DataFrames.
    
    Use case: Expensive query results that get re-requested
    within 15 minutes. Single worker, bounded data.
    
    Avoids Redis SDK overhead, network latency, and operational complexity.
    """
    def __init__(self):
        self._cache: dict[str, tuple[pd.DataFrame, float]] = {}
        self.default_ttl = 900  # 15 minutes
    
    def get(self, key: str) -> Optional[pd.DataFrame]:
        if key not in self._cache:
            return None
        
        df, expiry = self._cache[key]
        
        if datetime.now().timestamp() > expiry:
            del self._cache[key]
            return None
        
        return df.copy()  # Return copy to avoid mutations
    
    def set(self, key: str, df: pd.DataFrame, ttl_seconds: int = 900):
        """Cache a DataFrame with TTL."""
        self._cache[key] = (df.copy(), datetime.now().timestamp() + ttl_seconds)
    
    def memory_usage(self) -> dict:
        """Track what you're actually storing."""
        total_bytes = 0
        for key, (df, _) in self._cache.items():
            total_bytes += df.memory_usage(deep=True).sum()
        return {
            "total_mb": total_bytes / (1024 * 1024),
            "cached_dataframes": len(self._cache)
        }


# Usage example
pandas_cache = PandasCache()

def get_sales_report(year: int, region: str) -> pd.DataFrame:
    """
    Expensive query + transformation.
    
    If the same (year, region) is requested 10 times in 5 minutes,
    we hit the cache 9 times instead of the database 10 times.
    
    No Redis. No SDK. No deployment complexity.
    """
    cache_key = f"sales_report:{year}:{region}"
    
    cached = pandas_cache.get(cache_key)
    if cached is not None:
        return cached
    
    # Expensive operation
    df = pd.read_sql(
        f"SELECT * FROM sales WHERE year={year} AND region='{region}'",
        connection
    )
    
    # Transformations that take time
    df['revenue'] = df['quantity'] * df['price']
    df['month'] = pd.to_datetime(df['date']).dt.month
    
    # Cache and return
    pandas_cache.set(cache_key, df)
    return df.copy()

# Monitor what's cached
print(pandas_cache.memory_usage())
# Output: {'total_mb': 45.2, 'cached_dataframes': 3}

When this pattern wins:

• Data scientists or analysts running interactive queries against the same dataset repeatedly
• Dashboard endpoints that get hit by multiple users with the same parameters
• Report generation where the same filters produce identical results
• Single-worker FastAPI/Flask service where each process can have its own cache
• You have 50MB of data, not 50GB

Why avoid Redis for this:

• Redis SDK introduces a dependency
• Network round-trip (even localhost) adds 2-5ms per request
• Memory overhead in Redis (50+ bytes per key)
• Operational: another service to run, monitor, and troubleshoot
• For a 30MB DataFrame refreshing every 15 minutes in a single worker, Redis is theater

The hard constraint: This only works if:

• Single worker (or workers don't share cache and that's acceptable)
• Data changes are okay at a 15-minute granularity
• Total cached data stays under your available memory (not shared across processes)
• You monitor memory usage and have alerts

The honest limitation: If three workers all cache the same 30MB DataFrame independently, you're using 90MB instead of 30MB. This is fine for small datasets. At scale, it becomes wasteful—that's when Redis makes sense.

But for a single-worker analytics service? This is the pragmatic choice. Simple, zero operational overhead, no SDK, and measurably faster.

Tier 2: LRU Cache (Bounded Stupidity)

LRU (Least Recently Used) cache addresses the memory problem. You set a max size. When it fills up, the oldest unused item gets evicted.

Python's functools.lru_cache is the standard solution:

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
    """
    Bounded in-memory cache with LRU eviction.
    
    functools.lru_cache is thread-safe and gives you metrics.
    It's also a trap if you don't understand its limitations.
    """
    # Some expensive operation
    return n * n

# Check what's happening
print(expensive_computation.cache_info())
# Output: CacheInfo(hits=0, misses=1, maxsize=128, currsize=1)

# Force clear if needed
expensive_computation.cache_clear()

If you need more control, build your own:

from collections import OrderedDict
import threading

class LRUCache:
    """
    Simple LRU cache that actually works for single-threaded code.
    Multi-threaded? You'll need locks. And now you've got contention.
    """
    def __init__(self, max_size: int = 128):
        self.cache: OrderedDict[str, Any] = OrderedDict()
        self.max_size = max_size
        self.lock = threading.RLock()
    
    def get(self, key: str) -> Optional[Any]:
        with self.lock:
            if key not in self.cache:
                return None
            
            # Move to end (most recently used)
            self.cache.move_to_end(key)
            return self.cache[key]
    
    def set(self, key: str, value: Any):
        with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)
                self.cache[key] = value
            else:
                self.cache[key] = value
                if len(self.cache) > self.max_size:
                    # Remove least recently used
                    self.cache.popitem(last=False)

# Usage
cache = LRUCache(max_size=50)
cache.set("key1", "value1")
value = cache.get("key1")

When LRU makes sense:

• Computing the same value multiple times in a request cycle
• Expensive transformations (parsing, validation)
• You have bounded, known data (like "we'll never cache more than 100 users")
• You genuinely have memory pressure and need controlled eviction

When people use it wrong:

• Caching unbounded data sets. Your "max_size=1000" becomes "evict constantly and hit CPU on evictions"
• Cache stampede: 10 requests come in for the same expired key at the same time. You compute it 10 times.
• Thread contention: That lock you added? Now every request waits. The cache becomes a bottleneck.
• Stale data serving as your SLA. "99% of requests got cached data within 30ms" (it was 6 hours old, but sure)

The gotcha nobody mentions: functools.lru_cache requires arguments to be hashable. Pass a dict or a list, and it breaks silently (well, loudly, but in production it's silent-loud).

@lru_cache(maxsize=128)
def process_data(data: dict) -> str:
    # This will raise TypeError: unhashable type: 'dict'
    return str(data)

Tier 3: Distributed Cache (The "Let's Use Redis" Dream)

Now we're talking about real distributed systems. Multiple workers, multiple processes, one source of truth.

Redis is the standard answer, and for good reason. It has native TTL, is absurdly fast, and does exactly one thing: store strings.

import redis
import json
from typing import Any, Optional
from functools import wraps

class RedisCache:
    """
    Distributed cache that actually works across workers.
    
    The catch: you now depend on Redis. When Redis is slow,
    everything is slow. When Redis is down, you fail.
    
    It's like Jurassic Park—you've built something complex and powerful,
    and when the power goes out, everything starts hunting you.
    """
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.client = redis.from_url(redis_url, decode_responses=True)
    
    def get(self, key: str) -> Optional[Any]:
        value = self.client.get(key)
        if value is None:
            return None
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return value
    
    def set(self, key: str, value: Any, ttl_seconds: int = 300):
        """
        Set with automatic TTL.
        
        This is the right way: every write specifies expiration.
        If you don't, your cache becomes a dumping ground.
        """
        if isinstance(value, str):
            self.client.setex(key, ttl_seconds, value)
        else:
            self.client.setex(key, ttl_seconds, json.dumps(value))
    
    def delete(self, key: str):
        self.client.delete(key)
    
    def invalidate_pattern(self, pattern: str):
        """
        Dangerous but sometimes necessary.
        'user:123:*' -> invalidate all caches for user 123
        """
        for key in self.client.scan_iter(match=pattern):
            self.client.delete(key)


# Decorator
redis_cache = RedisCache()

def redis_cache_decorator(ttl_seconds: int = 300, key_prefix: str = "cache"):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # This still has the hashability problem
            cache_key = f"{key_prefix}:{func.__name__}:{args}:{kwargs}"
            
            cached = redis_cache.get(cache_key)
            if cached is not None:
                return cached
            
            result = func(*args, **kwargs)
            redis_cache.set(cache_key, result, ttl_seconds)
            return result
        
        return wrapper
    return decorator


@redis_cache_decorator(ttl_seconds=300, key_prefix="user")
def fetch_user(user_id: int):
    # Expensive DB query or API call
    return {"id": user_id, "name": "Alice"}

When Redis caching wins:

• Multiple services need shared data
• Data is accessed frequently (>100 req/sec) and reads are cheaper than writes
• You already have Redis (and ops team that knows how to run it)
• Consistency across workers matters

Where it fails:

• Cost. Redis memory is expensive. A single key takes 50+ bytes of overhead. Cache a million users? You're looking at 50GB+.
• Complexity. Now you have another system to monitor. Redis latency? Redis memory fragmentation? Cache coherency?
• Cache stampede. That key just expired, and 100 requests hit at the same time. They all miss, all call the database, all write back. Database melts.
• The "dumping ground" problem. Engineers cache everything. Response objects, raw API responses, computation results. Redis becomes a monster with 50GB of data nobody validates.

The honest gotcha: People treat Redis TTL as exact. It's not. Keys usually expire when you say, but:

• Redis could lag
• You set TTL=0, thinking it deletes immediately (it doesn't)
• You set TTL and never check if the data underneath changed

Tier 4: Dynamic Cache (When Data Changes)

This is where caching gets interesting. You cache, but you need to invalidate when the underlying data changes.

Pattern 1: Event-Driven Invalidation

This is the right way if you have an event system:

from dataclasses import dataclass
from typing import Callable

@dataclass
class DataChangeEvent:
    """Event published when data changes."""
    resource_type: str  # "user", "product", etc.
    resource_id: int
    action: str  # "created", "updated", "deleted"


class CacheInvalidator:
    """
    Listen to events and invalidate relevant cache keys.
    """
    def __init__(self, cache: RedisCache):
        self.cache = cache
        self.handlers: dict[str, list[Callable]] = {}
    
    def on_event(self, resource_type: str):
        def decorator(func: Callable):
            if resource_type not in self.handlers:
                self.handlers[resource_type] = []
            self.handlers[resource_type].append(func)
            return func
        return decorator
    
    def handle_event(self, event: DataChangeEvent):
        """When an event comes in, invalidate relevant caches."""
        handlers = self.handlers.get(event.resource_type, [])
        for handler in handlers:
            handler(event)


invalidator = CacheInvalidator(redis_cache)

@invalidator.on_event("user")
def invalidate_user_cache(event: DataChangeEvent):
    """When a user is updated, clear their cache."""
    if event.action in ["updated", "deleted"]:
        pattern = f"user:fetch_user:{event.resource_id}:*"
        redis_cache.invalidate_pattern(pattern)


# When something publishes a user.updated event:
# invalidator.handle_event(DataChangeEvent("user", 123, "updated"))

When this works:

• Event system already exists (message queue, pub/sub, event log)
• Data changes are infrequent relative to reads (10:1 or higher)
• Invalidation latency matters (<100ms)
• You have observability to track what was invalidated

When people fail at this:

• "We have an event system" but half the code path doesn't publish events
• The event queue has 10-minute lag. Your cache expires in 5 minutes. You served stale data anyway.
• Invalidation is based on assumptions about data relationships that change over time

Pattern 2: TTL + Probabilistic Revalidation

Instead of events, you bet on TTL being "good enough" most of the time:

import random

class SmartCache:
    """
    Cache with TTL, but periodically revalidate expensive data
    instead of serving stale data.
    
    This requires having a revalidation function available.
    """
    def __init__(self, cache: RedisCache):
        self.cache = cache
    
    def get_or_compute(
        self,
        key: str,
        compute_fn: Callable[[], Any],
        ttl_seconds: int = 300,
        revalidation_probability: float = 0.1
    ) -> Any:
        """
        Get from cache, but with probabilistic revalidation.
        
        1% of the time (tunable), ignore the cache and recompute,
        then update. This prevents the "cache stampede" where
        everyone hits the backend at once when TTL expires.
        """
        cached = self.cache.get(key)
        
        if cached is not None:
            # Small chance to revalidate without serving stale data
            if random.random() < revalidation_probability:
                # Spawn background task to recompute
                # (simplified here, real code uses tasks/threads)
                pass
            return cached
        
        # Cache miss, compute and store
        result = compute_fn()
        self.cache.set(key, result, ttl_seconds)
        return result

Tier 5: Database-Level Caching (The Thing You Missed)

Here's what nobody mentions: sometimes the database already caches what you're trying to cache.

When your application cache is overhead:

• Connection pooling keeps the connection warm (no handshake cost)
• The database's query cache is working
• Your indexes are good and queries are fast (<5ms)
• Network latency to the cache > latency to database + cache lookup

People cache to optimize away database queries. But if the database already optimizes the query...you're just adding latency.

Real examples: When caching the database makes sense vs. doesn't

Example 1: User profile lookup (CACHE IT)

# This makes sense to cache
@redis_cache_decorator(ttl_seconds=300)
def get_user_profile(user_id: int):
    # Complex query with joins
    return db.query(User)\
        .join(UserSettings)\
        .filter(User.id == user_id)\
        .first()

# Why: 
# - Query is slow: 50-100ms with cold database cache
# - Read-heavy: 1000s of requests per hour
# - Data is semi-stable: user data changes infrequently
# - TTL is safe: stale profile for 5 minutes is acceptable
# Value: (1000 × 0.075) - $0.10 = 74.90 saved per day

Example 2: Leaderboard query (DON'T CACHE IT)

# This doesn't make sense to cache
def get_leaderboard(limit: int = 100):
    return db.query(Score)\
        .order_by(Score.points.desc())\
        .limit(limit)\
        .all()

# Why not:
# - Query is fast: 10ms even with 10M rows (indexed)
# - Data is volatile: scores change constantly
# - Invalidation nightmare: every score update invalidates the cache
# - Cache hit ratio: maybe 20% before data changes
# - Redis memory cost: storing leaderboard + invalidation overhead > query cost
# Value: (1000 × 0.010) - $5/month = -$4.50/month. Don't cache.

Example 3: Product catalog (CACHE IT)

# This is the textbook case
@redis_cache_decorator(ttl_seconds=3600)  # 1 hour
def get_product(product_id: int):
    return db.query(Product)\
        .filter(Product.id == product_id)\
        .first()

# Why:
# - Read-heavy: 10,000 requests/sec for same products
# - Low-change: inventory updates via separate pipeline
# - Safe staleness: 1-hour-old price is acceptable
# - Network cost: latency to database in different region
# Value: (10000/sec × 3600 × 0.030ms) - $50/month = huge savings

Example 4: Session state (CACHE WITH CAUTION)

# This seems like caching but is actually just storage
def get_session(session_id: str):
    # This isn't caching—it's the authoritative store
    return redis_cache.get(f"session:{session_id}")

# Why this is tricky:
# - You're not caching the database
# - You're replacing the database with Redis
# - If Redis fails, sessions are gone
# - Better: use Redis for performance, write-through to DB for durability

Database indexing vs. caching:

# Before caching, did you check your indexes?
# CACHE:
@redis_cache_decorator(ttl_seconds=300)
def find_users_by_email(email: str):
    return db.query(User).filter(User.email == email).first()

# Better: Add index first
# CREATE INDEX idx_user_email ON users(email);
# Query becomes <1ms. Caching overhead (2-5ms) makes it slower.

# Check if index exists:
# SELECT * FROM pg_indexes WHERE tablename = 'users';
# If idx_user_email is missing, add it. Don't cache to work around bad indexing.

When database-level optimization wins:

# Bad: Application-level caching
@redis_cache_decorator(ttl_seconds=300)
def get_user(user_id: int):
    return db.query(User).filter(User.id == user_id).first()

# Better: Fix the database
# - Add index on User.id (probably already there)
# - Use connection pooling (pgBouncer, pgpool)
# - Query result is <5ms, caching adds 2-5ms latency overhead
# Just accept the 5ms database hit

Using SQLAlchemy with cacheops is interesting, but:

from cacheops import cached_as, invalidate_obj

# Automatic cache invalidation when ORM objects change
class User(db.Model):
    id: int
    name: str
    
    @classmethod
    @cached_as(timeout=300)
    def get_by_id(cls, user_id: int):
        return cls.query.filter(cls.id == user_id).first()

# Automatic invalidation on save
def update_user(user_id: int, name: str):
    user = User.query.get(user_id)
    user.name = name
    db.session.commit()
    invalidate_obj(user)  # Invalidate related caches

The problem: This looks elegant but hides complexity. If you have 20 different queries on User, do they all get invalidated? What if you update one field that only affects one query? You're over-invalidating or under-invalidating.

Real-world rule of thumb:

Cache the query if:
- Query latency > 20ms (expensive enough to matter)
- Read frequency > 10x write frequency (reads dominate)
- Data staleness > TTL is acceptable to users
- Storage cost < savings from avoiding queries

Don't cache if:
- Query is already <5ms (network overhead > query cost)
- Write frequency is high (constant invalidation)
- Data must be consistent within seconds
- You haven't verified the query is actually slow

Practical Caching Use Cases That Actually Make Sense

Not all caching is performance optimization. Sometimes it's about correctness and resource efficiency. These are the cases where cache isn't a nice-to-have feature—it's the load-bearing wall of your architecture. Like the Alien franchise: sometimes you need something dangerous, not because it's optimal, but because it's the only thing that works.

Use Case 1: Rate Limiting (Preventing Abuse)

class RateLimiter:
    """
    Cache is the enforcement mechanism, not just optimization.
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def is_allowed(self, user_id: str, limit: int = 100, window: int = 60) -> bool:
        """
        Allow 100 requests per 60 seconds.
        
        This *requires* distributed state. You can't do this with
        a local dict—too many workers.
        """
        key = f"ratelimit:{user_id}"
        
        # Increment counter and set expiry
        count = self.redis.incr(key)
        if count == 1:
            self.redis.expire(key, window)
        
        return count <= limit

# In your request handler:
limiter = RateLimiter(redis_client)
if not limiter.is_allowed(user_id):
    return 429  # Too Many Requests

# Cost: Saves you from DDoS and runaway clients
# Value: Preventing $50,000 AWS bill on bad actor > Redis cost

Use Case 2: Redis Locking (Task Deduplication)

Here's where caching is a lifesaver: preventing duplicate work.

import uuid
import time

class DistributedLock:
    """
    Use Redis as a lock mechanism to prevent duplicate processing.
    
    Classic case: 10 workers all trying to send the same email
    at the same time because they all see the task as unprocessed.
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def acquire(
        self,
        lock_key: str,
        ttl_seconds: int = 10
    ) -> bool:
        """
        Try to acquire a lock.
        
        Returns True if we got it, False if someone else has it.
        Expires automatically after ttl_seconds (safety against crashes).
        """
        # Set only if key doesn't exist
        # This is atomic—no race conditions
        lock_id = str(uuid.uuid4())
        acquired = self.redis.set(
            f"lock:{lock_key}",
            lock_id,
            nx=True,  # Only set if doesn't exist
            ex=ttl_seconds
        )
        return acquired is not None
    
    def release(self, lock_key: str):
        """Release the lock if we still own it."""
        # Don't delete blindly—ensure we own it
        self.redis.delete(f"lock:{lock_key}")

# Usage: Prevent duplicate task processing
lock = DistributedLock(redis_client)

def process_email_task(email_id: int):
    """
    Multiple workers might try to process the same email.
    Redis lock ensures only one actually does the work.
    """
    if not lock.acquire(f"email:{email_id}", ttl_seconds=30):
        # Someone else is processing this email
        return  # Skip it
    
    try:
        # Do the expensive work
        send_email(email_id)
        database.update_email_status(email_id, "sent")
    finally:
        lock.release(f"email:{email_id}")

# What this saves:
# Without lock: Email sent 10 times (customer gets 10 copies, angry)
# With lock: Email sent once
# Cost of Redis lock: negligible
# Cost of not using lock: customer rage, refunds, lost trust

Use Case 3: Counters (Analytics)

class EventCounter:
    """
    Count events with minimal overhead.
    
    Why not count in the database?
    Because incrementing a counter is faster in Redis (microseconds)
    than in database (milliseconds).
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def increment(self, event_name: str, count: int = 1):
        """Count an event in Redis."""
        key = f"counter:{event_name}:{datetime.now().strftime('%Y-%m-%d')}"
        self.redis.incrby(key, count)
        self.redis.expire(key, 86400 * 90)  # Keep 90 days
    
    def get_count(self, event_name: str, date: str) -> int:
        """Get count for a specific date."""
        key = f"counter:{event_name}:{date}"
        return int(self.redis.get(key) or 0)
    
    def flush_to_database(self):
        """
        Periodically (nightly), flush counts to the database.
        
        Trade: slightly stale analytics, huge latency improvement.
        """
        pattern = "counter:*"
        for key in self.redis.scan_iter(match=pattern):
            event_data = key.split(":")
            event_name = event_data[1]
            date = event_data[2]
            count = int(self.redis.get(key))
            
            # Write to database
            database.insert_event_count(event_name, date, count)
            self.redis.delete(key)

# Cost:
# Database: 1000 rows/sec with full ACID = slow
# Redis: 1,000,000 increments/sec = fast
# Flushing nightly: 1 database transaction = cheap
# Value: Real-time analytics without database contention

Use Case 4: Session Storage (With Fallback)

class SessionStore:
    """
    Sessions in Redis, with write-through to database for durability.
    
    Redis dies: You lose sessions but the app keeps running.
    Database dies: Sessions still work (they're in Redis).
    Both die: You have a real problem, but at least you see it.
    """
    def __init__(self, redis_client, db):
        self.redis = redis_client
        self.db = db
    
    def create_session(self, user_id: int, data: dict) -> str:
        """Create a session."""
        session_id = str(uuid.uuid4())
        session_data = json.dumps({"user_id": user_id, **data})
        
        # Write to both
        self.redis.setex(f"session:{session_id}", 3600, session_data)
        self.db.insert_session(session_id, user_id, session_data)
        
        return session_id
    
    def get_session(self, session_id: str) -> Optional[dict]:
        """Get session (fast path in Redis)."""
        cached = self.redis.get(f"session:{session_id}")
        if cached:
            return json.loads(cached)
        
        # Cache miss: check database
        # (session might still be valid, Redis cache just expired)
        db_session = self.db.get_session(session_id)
        if db_session:
            # Restore to Redis cache
            self.redis.setex(
                f"session:{session_id}",
                3600,
                db_session['data']
            )
            return json.loads(db_session['data'])
        
        return None

Use Case 5: Feature Flags (With TTL)

class FeatureFlagCache:
    """
    Cache feature flags for performance.
    
    Reading from the database every request for feature flags
    is wasteful. Cache with short TTL (30s) and control rolls.
    """
    def __init__(self, redis_client, db):
        self.redis = redis_client
        self.db = db
    
    def is_enabled(self, flag_name: str, user_id: int) -> bool:
        """
        Check if flag is enabled for user.
        
        Cached: 30 seconds (good compromise between freshness and latency)
        """
        cache_key = f"flag:{flag_name}:{user_id}"
        cached = self.redis.get(cache_key)
        
        if cached is not None:
            return cached == "true"
        
        # Cache miss: check database
        enabled = self.db.is_flag_enabled(flag_name, user_id)
        self.redis.setex(cache_key, 30, "true" if enabled else "false")
        
        return enabled
    
    def rollout_feature(self, flag_name: str):
        """
        Enable feature for everyone.
        
        Flush the cache immediately (invalidate all {flag_name}:* keys).
        """
        for key in self.redis.scan_iter(match=f"flag:{flag_name}:*"):
            self.redis.delete(key)

Summary of good caching:

The key difference: the first group (rate limiting, locking, counters) requires caching for correctness or resource efficiency. The second group is purely performance optimization.

Not all data should be cached the same way.

High-frequency change, high-frequency read (e.g., user session state)

• TTL: short (30-60 seconds)
• Strategy: Redis + invalidation on write
• Fallback: Accept brief staleness

Low-frequency change, high-frequency read (e.g., product catalog)

• TTL: long (1-24 hours) or event-driven invalidation
• Strategy: Redis, CDN, or even application memory
• Cost: Usually justified by the read volume

High-frequency change, low-frequency read (e.g., admin dashboard data)

• Cache: Don't. Or cache very briefly (5 seconds)
• Strategy: Compute on read, accept latency
• Cost of caching > cost of computation

Low-frequency change, low-frequency read (e.g., company info)

• Cache: Sure, why not
• TTL: Long (hours)
• Strategy: Anywhere, cost doesn't matter

The sweet spot calculation:

Value of caching = (Read frequency × Computation cost) - (Cache storage cost + Cache invalidation cost)

If Value < 0, don't cache.

Example:

• Read 100 times/day
• Computation cost: 50ms (database query)
• Cache storage: $0.01/month
• Invalidation cost: essentially free (event-driven)
• Value: (100 × 0.05) - $0.01 = 4.99 worth / month

Cache it.

Counter-example:

• Read 10 times/month (rare)
• Computation cost: 100ms
• Cache storage: $1/month (Redis memory cost for this object)
• Invalidation: complex (5 different code paths might change it)
• Value: (10/30 × 0.1) - $1 = 0.03 - $1 = -$0.97 / month

Don't cache it.

The Metrics That Matter

Here's where caching breaks. You look at the dashboard, see 95% hit ratio, and think you're winning.

You're not.

Like watching the Sixth Sense ending—you think you understand what's happening, but you're looking at a corpse. The 95% hit ratio? Dead data from three hours ago.

Metric 1: Hit Ratio (The Trap)

A 95% hit ratio on useless data is worse than a 50% hit ratio on critical data.

Hit ratio = Cache hits / (Cache hits + Cache misses)

This is meaningless without context.

What matters: Hit ratio weighted by value

class CacheMetrics:
    """
    Track what actually matters.
    """
    def __init__(self):
        self.hits = 0
        self.misses = 0
        self.latency_with_cache = []
        self.latency_without_cache = []
    
    @property
    def hit_ratio(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0
    
    @property
    def actual_benefit(self) -> float:
        """
        Real win: how much latency did we save?
        """
        avg_with = sum(self.latency_with_cache) / len(self.latency_with_cache)
        avg_without = sum(self.latency_without_cache) / len(self.latency_without_cache)
        
        # Only cache saves money if with_cache < without_cache
        return avg_without - avg_with

Metric 2: Latency Impact (The Critical Metric)

This is where caching fails silently.

def analyze_cache_latency():
    """
    Compare latency with cache vs. without cache using REAL BENCHMARKED DATA.
    
    Sources:
    - Redis official docs (redis.io): 0.2-0.6ms typical latency (same datacenter)
    - PostgreSQL benchmarks: indexed SELECT = 0.1-0.8ms
    - Network overhead: already included in above measurements
    """
    
    # Real-world scenario: Fetching a user profile
    cache_lookup = 0.5          # ms (Redis hit)
    backend_latency = 0.8       # ms (PostgreSQL indexed query)
    cache_miss_total = 1.3      # ms (0.5ms Redis miss check + 0.8ms query)
    direct_latency = 0.8        # ms (just database)
    
    # Scenario 1: Good hit ratio (80%)
    avg_with_cache_80 = (0.80 * cache_lookup) + (0.20 * cache_miss_total)
    # = (0.80 × 0.5) + (0.20 × 1.3) = 0.4 + 0.26 = 0.66ms
    # Savings: 0.8 - 0.66 = 0.14ms (17.5% improvement)
    
    # Scenario 2: Mediocre hit ratio (50%)
    avg_with_cache_50 = (0.50 * cache_lookup) + (0.50 * cache_miss_total)
    # = (0.50 × 0.5) + (0.50 × 1.3) = 0.25 + 0.65 = 0.90ms
    # Worse than no cache by 0.10ms
    
    # Scenario 3: Low hit ratio (20%)
    avg_with_cache_20 = (0.20 * cache_lookup) + (0.80 * cache_miss_total)
    # = (0.20 × 0.5) + (0.80 × 1.3) = 0.1 + 1.04 = 1.14ms
    # Worse than no cache by 0.34ms (42% slower!)
    
    return {
        "direct_latency_ms": direct_latency,
        "with_cache_80_percent_hits": avg_with_cache_80,
        "with_cache_50_percent_hits": avg_with_cache_50,
        "with_cache_20_percent_hits": avg_with_cache_20,
        "savings_at_80_percent_ms": 0.14,
        "overhead_at_50_percent_ms": 0.10,
        "overhead_at_20_percent_ms": 0.34
    }

The break-even point for caching:

Caching only helps if:

hit_ratio × backend_latency > cache_lookup_latency

With realistic numbers:
hit_ratio × 0.8 > 0.5
hit_ratio > 0.625

You need >62.5% hit ratio just to break even.
Below that, cache makes you slower.

Real-world example: When caching is actually worthwhile

Scenario: Caching expensive external API calls

- External API response time: 200ms (slow third-party service)
- Redis cache lookup: 0.5ms
- Hit ratio: 75% (request patterns have good reuse)
- Traffic: 100 requests/second
- TTL: 300 seconds

Latency WITH cache:
  (0.75 × 0.5) + (0.25 × 200.5) = 0.375 + 50.125 = 50.5ms

Latency WITHOUT cache:
  200ms

Savings: 149.5ms per request (75% improvement)

Cost analysis:
- Redis infrastructure: $100/month
- Operational overhead: 1 hour/month = $150
- Total: $250/month

Value:
- 100 req/sec × 86,400 sec/day × 30 days = 259.2M requests/month
- Savings: 149.5ms × 259.2M / 1000 = 38.7M seconds = 10,750 hours saved
- At $50/hour AWS compute: $537,500 in compute savings

ROI: +$537,250/month. Cache this immediately.

Real-world counter-example: When caching costs more than it saves

Scenario: Caching database queries that are already fast

- PostgreSQL indexed query: 0.8ms
- Redis cache lookup: 0.5ms
- Hit ratio: 45% (mostly unique requests)
- Traffic: 1000 requests/second
- Infrastructure cost: $150/month

Latency WITH cache:
  (0.45 × 0.5) + (0.55 × 1.3) = 0.225 + 0.715 = 0.94ms

Latency WITHOUT cache:
  0.8ms

Overhead: 0.14ms per request × 1000 req/sec = 140ms/second wasted

Cost analysis:
- Redis infrastructure: $150/month
- Operational overhead: 2 hours/month = $300
- Total: $450/month

Value:
- 0.14ms overhead × 86.4M requests/month = 12.1M ms = 12,096 seconds
- At $0.024/compute-hour: $80 cost of wasted compute
- Net loss: $450 - $80 = -$370/month

Decision: Remove the cache. The overhead exceeds any benefit.

**How to measure if cache is actually helping:**

def audit_cache_performance():
    """
    Simple measurement to tell if caching is worth it.
    
    Measure for 1 week in production with actual traffic patterns.
    """
    import time
    
    class CacheAudit:
        def __init__(self):
            self.hits = 0
            self.misses = 0
            self.hit_latencies = []
            self.miss_latencies = []
            self.direct_latencies = []  # For comparison
        
        def record_hit(self, latency_ms: float):
            self.hits += 1
            self.hit_latencies.append(latency_ms)
        
        def record_miss(self, latency_ms: float):
            self.misses += 1
            self.miss_latencies.append(latency_ms)
        
        def record_direct(self, latency_ms: float):
            """Latency when bypassing cache (for comparison)."""
            self.direct_latencies.append(latency_ms)
        
        def get_report(self):
            total = self.hits + self.misses
            hit_ratio = self.hits / total if total > 0 else 0
            
            avg_with_cache = (
                (hit_ratio * (sum(self.hit_latencies) / len(self.hit_latencies)))
                + ((1 - hit_ratio) * (sum(self.miss_latencies) / len(self.miss_latencies)))
            )
            
            avg_direct = sum(self.direct_latencies) / len(self.direct_latencies)
            
            savings_per_request = avg_direct - avg_with_cache
            
            print(f"Hit ratio: {hit_ratio:.1%}")
            print(f"Avg latency WITH cache: {avg_with_cache:.2f}ms")
            print(f"Avg latency WITHOUT cache: {avg_direct:.2f}ms")
            print(f"Savings per request: {savings_per_request:.2f}ms")
            
            # Decision logic
            if hit_ratio < 0.50:
                print("❌ Hit ratio < 50%: Cache is likely adding overhead")
                return False
            
            if savings_per_request < 0.05:  # Less than 0.05ms savings
                print("❌ Savings < 0.05ms: Not worth the infrastructure cost")
                return False
            
            print("✓ Cache appears worthwhile. Keep it.")
            return True
    
    # Usage during production monitoring
    audit = CacheAudit()
    
    # Record metrics for 1 week...
    # audit.record_hit(0.5)
    # audit.record_miss(1.3)
    # audit.record_direct(0.8)
    
    # audit.get_report()

Decision tree:

Does the cache save more than 0.1ms per request?
  YES → Is the hit ratio > 50%?
    YES → Keep the cache
    NO → Remove it or redesign
  NO → Remove it immediately

Metric 3: Cardinality Explosion and Storage Costs

This is insidious. You cache "user:{user_id}:profile" and you're fine. Then you add "user:{user_id}:settings", "user:{user_id}:preferences", "user:{user_id}:avatar_url". Now you have 10 million keys instead of 1 million.

Redis memory cost breakdown:

Real-world: How much does Redis memory actually cost?

Per-key overhead:
- Key: "user:123456:profile" = 20 bytes
- Value: JSON profile = 500 bytes
- Redis internal pointers: ~100 bytes
- Hash table overhead: ~50 bytes
- Total per key: ~670 bytes

Scale it:
- 1 million keys: ~670MB ≈ $30/month
- 10 million keys: ~6.7GB ≈ $300/month
- 100 million keys: ~67GB ≈ $3000/month

But that's just storage. Redis also has:
- Network overhead (Redis cluster replication)
- CPU overhead (key lookups, expiration)
- Operational cost (backups, monitoring, failover)

Real cost per key: ~0.3-1 cent per million keys per month.

Example cardinality explosion:
Started with:
  - user:123:data (1 million keys, $30/month)

Then added:
  - user:123:settings (1 million keys, $30/month)
  - user:123:preferences (1 million keys, $30/month)
  - user:123:posts (10 million keys, $300/month)
  - user:123:comments (10 million keys, $300/month)
  - user:123:likes (10 million keys, $300/month)
  
Total: 33 million keys, $990/month in Redis costs alone.

vs. if you'd just hit the database:
  - Smart indexing: <5ms per query
  - Connection pooling: managed costs
  - Database storage: included in your instance cost
  - Total: Already budgeted infrastructure cost

Track cardinality obsessively:

def check_redis_cardinality():
    """Alert if we're growing unbounded."""
    info = redis_cache.client.info()
    key_count = info['db0']['keys']
    memory = info['used_memory']
    
    bytes_per_key = memory / key_count if key_count > 0 else 0
    
    # Thresholds
    if key_count > 50_000_000:
        alert(f"Critical: {key_count:,} keys, {memory/1e9:.1f}GB")
    elif key_count > 10_000_000:
        alert(f"Warning: {key_count:,} keys, {memory/1e9:.1f}GB")
    
    # Bytes per key should be reasonable
    if bytes_per_key > 200:
        alert(f"High memory per key: {bytes_per_key:.0f} bytes")
        # Normal range: 50-150 bytes depending on value size

Redis Atomicity: Why Lua Scripts Matter

Here's a common bug: you check a Redis value, make a decision, then write back. But between the read and write, another process changed the value.

The problem: Race conditions without atomicity

# WRONG: Multiple operations = multiple round trips
def increment_with_check(key: str, max_value: int = 100):
    """Increment a counter, but don't exceed max_value."""
    
    # Read
    current = int(redis_client.get(key) or 0)
    
    # Logic
    if current < max_value:
        # Write
        redis_client.incr(key)
        return True
    
    return False

# Race condition:
# Thread 1: Get key = 99
# Thread 2: Get key = 99  ← Problem! Both see 99
# Thread 1: Incr to 100
# Thread 2: Incr to 100  ← Should have been rejected!
# Result: Counter is 100, but you incremented twice over the limit

The solution: Lua scripts for atomic operations

Lua scripts run on the Redis server, atomically. No race conditions.

# Define the Lua script
increment_script = redis_client.register_script("""
    local current = tonumber(redis.call('GET', KEYS[1]) or 0)
    local max_val = tonumber(ARGV[1])
    
    if current < max_val then
        redis.call('INCR', KEYS[1])
        return 1  -- Success
    else
        return 0  -- Failed, would exceed max
    end
""")

# Use it
result = increment_script(keys=['counter'], args=[100])
# Returns 1 if successful, 0 if hit limit
# Guaranteed atomic: no race conditions

More realistic example: Distributed counter with threshold

class SafeDistributedCounter:
    """
    Count something (events, tasks, etc.) with a threshold.
    
    Example: Count failed login attempts. After 5, lock the account.
    """
    def __init__(self, redis_client):
        self.redis = redis_client
        
        # Lua script to increment and check threshold atomically
        self.increment_script = redis_client.register_script("""
            local key = KEYS[1]
            local threshold = tonumber(ARGV[1])
            local ttl = tonumber(ARGV[2])
            
            local current = tonumber(redis.call('GET', key) or 0)
            
            -- Increment
            local new_value = redis.call('INCR', key)
            
            -- Set TTL if this is the first increment
            if new_value == 1 then
                redis.call('EXPIRE', key, ttl)
            end
            
            -- Return: new value and whether it exceeded threshold
            if new_value >= threshold then
                return {new_value, 1}  -- {count, threshold_exceeded}
            else
                return {new_value, 0}
            end
        """)
    
    def increment_and_check(
        self,
        counter_key: str,
        threshold: int,
        ttl_seconds: int = 300
    ) -> tuple[int, bool]:
        """
        Increment counter and return (count, did_exceed_threshold).
        
        All atomic—guaranteed no race conditions.
        """
        result = self.increment_script(
            keys=[counter_key],
            args=[threshold, ttl_seconds]
        )
        count, exceeded = result
        return count, exceeded == 1

# Usage: Lockout after 5 failed attempts
counter = SafeDistributedCounter(redis_client)

def check_login(user_id: int, password: str) -> bool:
    attempts, locked = counter.increment_and_check(
        f"login_attempts:{user_id}",
        threshold=5,
        ttl_seconds=300
    )
    
    if locked:
        # Account is locked for 5 minutes
        return False, f"Account locked. {5 - attempts} attempts remaining."
    
    if authenticate_user(user_id, password):
        # Reset counter on successful login
        redis_client.delete(f"login_attempts:{user_id}")
        return True, "Login successful"
    
    return False, f"Invalid password. {5 - attempts} attempts remaining."

When to use Lua scripts:

• When you need to read-check-write atomically
• When you need multiple operations to succeed or fail together
• When race conditions would cause bugs
• When you're implementing distributed algorithms (locks, counters, rate limiting)

When NOT to use Lua scripts:

• Simple get/set (doesn't need atomicity)
• When the script is complex (>50 lines, hard to debug)
• When you need conditional branching based on external data (fetch from DB)

Common Lua patterns:

# Pattern 1: Compare and swap (CAS)
cas_script = redis_client.register_script("""
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        redis.call('SET', KEYS[1], ARGV[2])
        return 1
    else
        return 0
    end
""")

# Pattern 2: Decrement if > 0 (tokens, credits)
tokens_script = redis_client.register_script("""
    local current = tonumber(redis.call('GET', KEYS[1]) or 0)
    if current > 0 then
        redis.call('DECR', KEYS[1])
        return 1
    else
        return 0
    end
""")

# Pattern 3: Atomic push to list with size limit
list_script = redis_client.register_script("""
    local key = KEYS[1]
    local max_size = tonumber(ARGV[1])
    local value = ARGV[2]
    
    redis.call('LPUSH', key, value)
    local size = redis.call('LLEN', key)
    
    if size > max_size then
        redis.call('RPOP', key)
    end
    
    return size
""")

Performance comparison:

Without Lua (multiple commands):
- GET: 0.5ms
- Check: 0.1ms
- SET: 0.5ms
- Total: ~1.1ms
- Risk: Race condition between GET and SET

With Lua (single atomic command):
- EVAL: ~1.0ms (slightly faster, no round trips)
- Risk: None (atomic on Redis side)

Bonus: Eliminates network overhead of multiple round trips.
At high throughput (10,000 req/sec), the difference is significant.

After 20+ years watching caching fail in production, here's what I've seen work:

1. Cache only expensive, infrequently-changing data
   • If it's cheap to compute, don't cache it
   • If it changes constantly, don't cache it
2. Choose the cache tier that matches your problem
   • Single worker? In-memory dict
   • Multiple workers, bounded data? LRU
   • Multiple workers, unbounded data? Redis with cardinality limits
   • Data changes frequently? Event-driven invalidation + short TTL as safety net
3. Measure the actual benefit
   • Not hit ratio. Latency improvement.
   • Not feature flags. Cost impact.
4. Have an escape hatch
   • When caching fails, be able to disable it instantly
   • Feature flag that routes percentage of traffic around cache
   • Make cache failures non-catastrophic (stale data is better than no data)
5. Accept that caches lie
   • Eventually, inconsistency will happen
   • Your system should tolerate 5-minute-old data
   • If you can't tolerate stale data, don't cache it

Bonus: Caching LLM Responses (The New Frontier)

With LLM costs being real, people are asking: can we cache LLM replies?

The tempting answer: "Query the embedding model on the question, cluster semantically similar questions, return cached answers."

The honest answer: "Sometimes, but usually not for the reason you think."

The Theory vs. The Reality

The idea is sound on paper:

• User asks "What is the capital of France?"
• LLM response is cached
• User asks "What's the capital of France?"
• Same query, cached answer, saved tokens

But queries are rarely identical. So the next thought: embed questions, find similar ones, serve cached answers.

This feels like you're hacking a Terminator 2 time-jump—all this complex infrastructure to go back and stop the problem before it starts. Except you end up with the skeleton of a solution that costs more than just paying the T-1000 upfront.

When semantic similarity caching doesn't make financial sense:

Cost structure:
- Embedding model API: $0.02 per 1000 embeddings
- Vector database: $50-200/month (Pinecone, Weaviate, Qdrant)
- Search latency: 50-100ms per query

vs.

LLM input cost:
- GPT-4 Turbo: $0.01 per 1000 input tokens
- Average question: 20 tokens = $0.0002 per call

Math:
- Cache 10,000 questions: $0.20 in embedding cost
- Vector DB: $100/month
- If you cache 90% of requests, you save $0.00018 per request
- Break-even: Never. You're losing $100/month forever.

The real trap: you're betting that embedding similarity = semantic equivalence. It doesn't.

# These are semantically similar according to embeddings:
question_1 = "What is the capital of France?"
question_2 = "What city is the capital of France?"

# But the user might want different response styles:
# Q1: Quick fact → "Paris"
# Q2: Conversational → "Paris is the capital and largest city..."

# Cache serves the same answer to both. One user is annoyed.

When semantic caching MIGHT be worth the money: Speed > Cost

There's exactly one scenario where this makes sense: when latency is a SLA, and every millisecond costs you money.

Scenario: Customer support chatbot
- Response time SLA: <500ms
- Current: 200ms LLM latency + network overhead = 250ms (tight but doable)
- Load: 10,000 requests/day
- Token cost savings: 90% cache hit = $50/day saved
- Infrastructure cost: $200/month

But if you miss the SLA once:
- Customer escalation: 1 hour work = $150
- Reputation damage: small
- One missed SLA = $150 cost

If semantic caching gets you 50ms faster (150ms total):
- Margin to SLA: much better
- Risk of missing SLA: drops from 10% to 1%
- Expected value of risk reduction: 0.09 × $150 = $13.50/day

Over 30 days: $405 in risk reduction vs. $200 cost = net positive.

Decision: If speed is contractually critical, the infrastructure might be worth it.
If it's just "nice to have," it's not.

The hidden costs you're missing:

1. Invalidation nightmare. When the underlying data changes, which cached answers are stale? Which semantically similar questions depend on it? Your cache becomes a consistency nightmare.
2. Hallucination amplification. The LLM hallucinated once. You cached it. Now 90% of similar questions get served the same hallucination. You've scaled the problem.
3. Version drift. Your LLM updates (new model, new prompt). Old cached answers are suboptimal. Do you flush? Gradually? How?
4. Observability death. When something goes wrong, was it the LLM, the cache, or the embeddings? You've added three layers of potential failure.

The solution that actually works: Exact-match caching

Forget semantic similarity. Just cache identical questions.

class ExactMatchLLMCache:
    """
    The boring solution that doesn't fail.
    
    Same question in the same session? Serve cached answer.
    Different question? Ask the LLM.
    No embeddings, no vector DB, no hallucination scaling.
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def get_cached_or_ask_llm(
        self,
        question: str,
        llm_fn,
        ttl_seconds: int = 300  # 5 minutes
    ) -> str:
        # Hash the exact question
        cache_key = f"llm:question:{hash(question)}"
        
        # Check if we've seen this exact question recently
        cached = self.redis.get(cache_key)
        if cached:
            return cached
        
        # Ask the LLM
        answer = llm_fn(question)
        
        # Cache with short TTL
        self.redis.setex(cache_key, ttl_seconds, answer)
        
        return answer

Why this works:

• Zero complexity beyond Redis
• Protects against the same user asking the same question in the same session (happens more than you think)
• Saves maybe 5-10% of calls, not 90% (realistic)
• Invalidation is automatic via TTL
• No embedding infrastructure
• Cost: negligible

The real solution for LLM efficiency:

Stop trying to cache. Instead:

1. Optimize prompts. Better engineering reduces tokens. Fewer tokens = cheaper.
2. Use a smaller model. Claude Sonnet is cheaper than Opus. For FAQ, use a smaller LLM entirely.
3. Batch requests. Semantic caching looks good on paper. Batching actually works in practice.
4. Pre-generate answers. For known FAQs, generate the answer once (expensive compute) and serve it (free) via simple pattern matching. No caching layer, no complexity.

The honest verdict:

Semantic LLM caching is the LLM equivalent of "I'll just add Redis to fix this." It looks like a technical solution to a business problem (cost). The real problem is usually that you're using an expensive model for a cheap problem.

Like something from Robocop—all this hardware to solve what could have been handled by a cheaper solution from the start.

Caching is one of the easiest ways to take a simple system and make it complex while barely improving anything. I've seen systems where the "optimization" added 3 services, 40 lines of configuration, 2 monitoring dashboards, and saved 2ms on the p95 latency.

That 2ms? Not worth it.

The engineers trying to support it in production? They disagree, at 3 AM, on a Sunday. They're debugging which layer the problem is in—application cache, Redis, database query cache, or the poor bastard who set TTL=0 thinking it would delete immediately.

Like an '80s action movie—the hero keeps running from one problem to the next, except the problem is the hero. Predator, except the predator is your caching layer hunting you through the logs.

Cache when it matters. Measure what actually matters. Be honest about whether it's working.

And maybe, just maybe, your system doesn't need another layer.