caching

name	caching
description	Implement multi-tier database caching with Redis, in-memory, and CDN layers...
shortcut	cach

Database Cache Layer

Implement production-grade multi-tier caching architecture for databases using Redis (distributed cache), in-memory caching (L1), and CDN (static assets) to reduce database load by 80-95%, improve query latency from 50ms to 1-5ms, and support horizontal scaling with cache-aside, write-through, and read-through patterns.

When to Use This Command

Use /caching when you need to:

• Reduce database load by caching frequently accessed data (80% hit rate)
• Improve query response times from 50-100ms to 1-5ms
• Handle traffic spikes without database scaling (cache absorbs load)
• Support read-heavy workloads with minimal database reads
• Implement distributed caching across multiple application servers
• Enable horizontal scaling with stateless application servers

DON'T use this when:

• Data changes frequently and cache hit rate would be <50%
• Application has strict real-time data requirements (< 1s staleness)
• Database is already fast enough (<10ms query latency)
• You lack cache invalidation strategy (stale data risk)
• Small dataset fits entirely in database memory (shared_buffers)
• Write-heavy workload (caching provides minimal benefit)

Design Decisions

This command implements multi-tier caching with intelligent invalidation because:

• L1 in-memory cache (1-5ms) for hot data per server
• L2 distributed Redis cache (5-10ms) shared across servers
• Cache-aside pattern provides fallback to database on miss
• TTL-based and event-based invalidation prevents stale data
• Write-through caching maintains consistency for critical data

Alternative considered: Read-through caching

• Simpler implementation (cache handles database queries)
• Less control over cache population strategy
• Not suitable when database schema differs from cached format
• Recommended for simple key-value lookups

Alternative considered: Database query result caching (pg_stat_statements)

• Built into PostgreSQL (no external dependencies)
• Limited to identical queries (parameter changes = cache miss)
• Cannot cache across multiple queries
• Recommended for development/small workloads only

Prerequisites

Before running this command:

1. Redis server deployed (standalone, Sentinel, or Cluster)
2. Understanding of cache invalidation needs (TTL vs event-driven)
3. Monitoring for cache hit rate and memory usage
4. Connection pooling configured for Redis clients
5. Fallback strategy for cache failures (graceful degradation)

Implementation Process

Step 1: Design Cache Key Strategy

Define hierarchical cache keys for easy invalidation (e.g., user:123:profile).

Step 2: Implement Cache-Aside Pattern

Check cache first, query database on miss, populate cache with result.

Step 3: Configure TTL and Eviction

Set appropriate TTL based on data freshness requirements and memory limits.

Step 4: Implement Invalidation Logic

Invalidate cache on data updates using event listeners or explicit invalidation.

Step 5: Monitor Cache Performance

Track hit rate, miss rate, latency, and memory usage with Prometheus/Grafana.

Output Format

The command generates:

• caching/redis_client.py - Redis connection pool and wrapper
• caching/cache_decorator.py - Python decorator for automatic caching
• caching/cache_invalidation.js - Event-driven invalidation logic
• caching/cache_monitoring.yml - Prometheus metrics and alerts
• caching/cache_warming.sql - SQL queries for cache preloading

Code Examples

Example 1: Python Multi-Tier Cache with Redis and In-Memory

#!/usr/bin/env python3
"""
Production-ready multi-tier caching system with L1 (in-memory) and
L2 (Redis) caches, automatic invalidation, and performance monitoring.
"""

import redis
import pickle
from typing import Optional, Callable, Any
from functools import wraps
from datetime import timedelta
import time
import logging
from cachetools import TTLCache
import hashlib
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MultiTierCache:
    """
    Two-tier caching system with L1 (in-memory) and L2 (Redis).

    L1: Fast in-memory cache (1-5ms) for hot data
    L2: Distributed Redis cache (5-10ms) shared across servers
    """

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379/0",
        l1_max_size: int = 1000,
        l1_ttl_seconds: int = 60,
        l2_ttl_seconds: int = 3600,
        enabled: bool = True
    ):
        """
        Initialize multi-tier cache.

        Args:
            redis_url: Redis connection URL
            l1_max_size: Max entries in L1 cache
            l1_ttl_seconds: L1 cache TTL (default: 1 minute)
            l2_ttl_seconds: L2 cache TTL (default: 1 hour)
            enabled: Enable/disable caching (useful for debugging)
        """
        self.enabled = enabled

        if not enabled:
            logger.warning()
            

        
        .l1_cache = TTLCache(maxsize=l1_max_size, ttl=l1_ttl_seconds)
        .l1_ttl = l1_ttl_seconds

        
        .redis_client = redis.from_url(
            redis_url,
            decode_responses=,  
            socket_connect_timeout=,
            socket_timeout=,
            retry_on_timeout=
        )
        .l2_ttl = l2_ttl_seconds

        
        .metrics = {
            : ,
            : ,
            : ,
            : ,
            : ,
            : 
        }

     () -> :
        
        
        key_parts = [(arg)  arg  args]
        key_parts.extend(  k, v  (kwargs.items()))
        key_suffix = hashlib.md5(
            .join(key_parts).encode()
        ).hexdigest()[:]

         

     () -> []:
        
          .enabled:
             

        
         key  .l1_cache:
            .metrics[] += 
            logger.debug()
             .l1_cache[key]

        .metrics[] += 

        
        :
            cached_data = .redis_client.get(key)
             cached_data:
                .metrics[] += 
                logger.debug()

                
                value = pickle.loads(cached_data)
                .l1_cache[key] = value

                 value

            .metrics[] += 
             

         redis.RedisError  e:
            logger.error()
            .metrics[] += 
             

     () -> :
        
          .enabled:
             

        :
            
            .l1_cache[key] = value

            
            serialized = pickle.dumps(value)
            ttl = l2_ttl  .l2_ttl
            .redis_client.setex(key, ttl, serialized)

            logger.debug()
             

         redis.RedisError  e:
            logger.error()
            .metrics[] += 
             

     () -> :
        
          .enabled:
             

        :
            
            .l1_cache.pop(key, )

            
            .redis_client.delete(key)

            logger.info()
             

         redis.RedisError  e:
            logger.error()
            .metrics[] += 
             

     () -> :
        
          .enabled:
             

        :
            
            cursor = 
            deleted_count = 

             :
                cursor, keys = .redis_client.scan(
                    cursor,
                    =pattern,
                    count=
                )

                 keys:
                    deleted_count += .redis_client.delete(*keys)

                 cursor == :
                    

            
            .l1_cache.clear()

            logger.info()
             deleted_count

         redis.RedisError  e:
            logger.error()
            .metrics[] += 
             

     () -> :
        
        total_l1 = .metrics[] + .metrics[]
        total_l2 = .metrics[] + .metrics[]

        l1_hit_rate = (
            .metrics[] / total_l1 * 
             total_l1 >   
        )

        l2_hit_rate = (
            .metrics[] / total_l2 * 
             total_l2 >   
        )

        overall_hit_rate = (
            (.metrics[] + .metrics[]) /
            (total_l1 + total_l2) * 
             (total_l1 + total_l2) >   
        )

         {
            : .metrics[],
            : .metrics[],
            : (l1_hit_rate, ),
            : .metrics[],
            : .metrics[],
            : (l2_hit_rate, ),
            : (overall_hit_rate, ),
            : .metrics[],
            : .metrics[]
        }



cache = MultiTierCache()


 ():
    
     () -> :

         ():
            
            cache_key = cache._generate_key(prefix, *args, **kwargs)

            
            cached_result = cache.get(cache_key)
             cached_result   :
                 cached_result

            
            cache.metrics[] += 
            result = func(*args, **kwargs)

            
            cache.(cache_key, result, l2_ttl=l2_ttl)

             result

         wrapper
     decorator




 ():
    
     psycopg2
    conn = psycopg2.connect()
     conn.cursor()  cur:
        cur.execute(, (user_id,))
         cur.fetchone()



 ():
    
     psycopg2
    conn = psycopg2.connect()
     conn.cursor()  cur:
        cur.execute(
            ,
            (user_id, limit)
        )
         cur.fetchall()


 ():
    
    cache.delete_pattern()



 ():
    
     psycopg2
    conn = psycopg2.connect()
     conn.cursor()  cur:
        
        set_clause = .join(  k  updates.keys())
        cur.execute(
            ,
            (*updates.values(), user_id)
        )
        conn.commit()

    
    invalidate_user_cache(user_id)
    logger.info()


 __name__ == :
    
    ()

    
    start = time.time()
    profile1 = get_user_profile()
    db_time = (time.time() - start) * 
    ()

    
    start = time.time()
    profile2 = get_user_profile()
    cache_time = (time.time() - start) * 
    ()
    ()

    
    ()
    (json.dumps(cache.get_metrics(), indent=))

Example 2: Cache Warming and Preloading

#!/usr/bin/env python3
"""
Cache warming strategy to preload hot data before traffic hits.
Reduces cold start latency and improves cache hit rate.
"""

import psycopg2
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class CacheWarmer:
    """
    Preload cache with frequently accessed data.
    """

    def __init__(self, cache: MultiTierCache, db_conn_string: str):
        """
        Initialize cache warmer.

        Args:
            cache: MultiTierCache instance
            db_conn_string: Database connection string
        """
        self.cache = cache
        self.db_conn_string = db_conn_string

    def warm_user_profiles(self, user_ids: list[int]) -> dict:
        """
        Preload user profiles for given IDs.

        Args:
            user_ids: List of user IDs to warm

        Returns:
            Statistics (count, duration, errors)
        """
        start_time = time.time()
        stats = {'loaded': 0, 'errors': 0}

        logger.info(f"Warming cache for {len(user_ids)} user profiles...")

        with psycopg2.connect(self.db_conn_string) as conn:
            with conn.cursor() as cur:
                for user_id in user_ids:
                    :
                        
                        cur.execute(
                            ,
                            (user_id,)
                        )
                        profile = cur.fetchone()

                         profile:
                            
                            cache_key = 
                            .cache.(cache_key, profile, l2_ttl=)
                            stats[] += 

                     Exception  e:
                        logger.error()
                        stats[] += 

        duration = time.time() - start_time
        stats[] = duration

        logger.info(
            
            
        )

         stats

     () -> :
        
        start_time = time.time()
        stats = {: , : }

        logger.info()

         psycopg2.connect(.db_conn_string)  conn:
             conn.cursor()  cur:
                
                cur.execute(, (limit,))

                products = cur.fetchall()

                 product  products:
                    :
                        product_id = product[]  
                        cache_key = 
                        .cache.(cache_key, product, l2_ttl=)
                        stats[] += 

                     Exception  e:
                        logger.error()
                        stats[] += 

        duration = time.time() - start_time
        stats[] = duration

        logger.info(
            
            
        )

         stats

     () -> :
        
        logger.info()

        
         psycopg2.connect(.db_conn_string)  conn:
             conn.cursor()  cur:
                
                cur.execute()
                hot_user_ids = [row[]  row  cur.fetchall()]

        
         ThreadPoolExecutor(max_workers=)  executor:
            futures = {
                executor.submit(.warm_user_profiles, hot_user_ids): ,
                executor.submit(.warm_top_products, ): 
            }

            results = {}
             future  as_completed(futures):
                cache_type = futures[future]
                :
                    results[cache_type] = future.result()
                 Exception  e:
                    logger.error()

         results



 __name__ == :
     multitiercache  cache

    warmer = CacheWarmer(
        cache=cache,
        db_conn_string=
    )

    
    results = warmer.warm_all_hot_data()
    ()

Error Handling

Error	Cause	Solution
"Redis connection refused"	Redis server down or unreachable	Implement graceful degradation (bypass cache, query database directly)
"Out of memory" (Redis)	Cache size exceeds max memory	Configure eviction policy (maxmemory-policy allkeys-lru) or increase memory
"Pickle deserialization error"	Cached object structure changed	Version cache keys when data models change, invalidate old caches
"Cache stampede"	Many requests miss cache simultaneously	Use locking or probabilistic early expiration to prevent thundering herd
"Stale data returned"	TTL too long or invalidation missed	Reduce TTL, implement event-driven invalidation on updates

Configuration Options

Caching Patterns

• Cache-aside (lazy loading): App checks cache, queries DB on miss
• Read-through: Cache handles DB queries automatically
• Write-through: Updates written to cache and DB simultaneously
• Write-behind: Updates written to cache, async written to DB

Eviction Policies (Redis)

• allkeys-lru: Evict least recently used keys (recommended for general use)
• volatile-lru: Evict LRU keys with TTL set
• allkeys-random: Random eviction (simple, unpredictable)
• volatile-ttl: Evict keys closest to expiration

TTL Strategies

• Hot data (user profiles): 30 minutes
• Warm data (product catalog): 1-2 hours
• Cold data (historical reports): 24 hours
• Static data (configuration): 7 days

Best Practices

DO:

• Set appropriate TTLs based on data freshness requirements
• Monitor cache hit rate (target: 80%+) and adjust strategy
• Implement graceful degradation when cache is unavailable
• Use hierarchical cache keys for easy pattern-based invalidation
• Warm cache with hot data during deployment
• Version cache keys when data schemas change
• Use connection pooling for Redis clients (reduce connection overhead)

DON'T:

• Cache data that changes frequently (< 50% hit rate)
• Use cache for critical consistency (financial transactions, inventory)
• Ignore cache memory limits (causes evictions and performance degradation)
• Cache large objects (> 1MB) without compression
• Forget to invalidate cache on data updates (stale data bugs)
• Use cache as primary data store (Redis is not durable)
• Over-cache (memory waste, low hit rate)

Performance Considerations

• L1 cache hit: 1-5ms (in-memory)
• L2 cache hit: 5-10ms (Redis)
• Database query: 50-100ms (depending on complexity)
• Speedup: 10-100x faster than database queries
• Cache hit rate target: 80-95% for read-heavy workloads
• Memory usage: 1MB per 10,000 small objects (varies by data size)
• Redis throughput: 100,000+ ops/sec (single instance)

Security Considerations

• Encrypt sensitive data before caching (PII, credentials)
• Use Redis AUTH and TLS for production (prevent unauthorized access)
• Isolate cache per tenant in multi-tenant applications
• Audit cache access for compliance (GDPR, HIPAA)
• Implement cache poisoning prevention (validate cached data)
• Secure Redis instance in private network (no public access)
• Rotate Redis passwords quarterly

Related Commands

• /database-connection-pooler - Optimize connections when cache is unavailable
• /database-health-monitor - Monitor cache hit rate and database load
• /sql-query-optimizer - Optimize queries that are cache misses
• /database-security-scanner - Audit sensitive data in cache

Version History

• v1.0.0 (2024-10): Initial implementation with Redis and in-memory caching
• Planned v1.1.0: Add memcached support, distributed tracing integration