1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

"""
Demonstration of cache consistency challenges.
These examples show how seemingly simple caching can lead to subtle bugs.
"""
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import threading
 
 
@dataclass
class Product:
    id: str
    name: str
    price: float
    inventory: int
    last_updated: datetime
 
 
class NaiveCache:
    """
    A naive cache implementation that demonstrates consistency problems.
    DO NOT use in production - for educational purposes only.
    """
    
    def __init__(self, ttl_seconds: int = 60):
        self._cache: dict = {}
        self._ttl = timedelta(seconds=ttl_seconds)
        self._lock = threading.Lock()
    
    def get(self, key: str) -> Optional[Product]:
        """Simple get - returns None if expired or missing."""
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                return None
            
            cached_at, product = entry
            if datetime.now() - cached_at > self._ttl:
                # Entry expired - but DON'T delete yet (causes stampede)
                return None
            
            return product
    
    def set(self, key: str, product: Product) -> None:
        """Cache the product."""
        with self._lock:
            self._cache[key] = (datetime.now(), product)
 
 
# PROBLEM 1: Race condition on update
# =====================================
"""
Scenario: 
- User A reads product (cached at T=0, price=$100)
- At T=10, admin updates price to $75 in database
- User B reads product (still gets $100 from cache until TTL expires)
- User B buys at $100, expects $75 based on admin's announcement
 
This "stale read" is inherent to caching - TTL determines the window.
"""
 
 
# PROBLEM 2: Cache stampede on expiration
# ========================================
"""
Scenario:
- Popular product cached at T=0, TTL=60 seconds
- At T=60, cache expires
- 1000 concurrent users request product
- All 1000 requests hit database simultaneously
- Database overloaded, all requests fail or timeout
 
Solution: Stale-while-revalidate, probabilistic early refresh, or locking
"""
 
 
# PROBLEM 3: Lost update with write-through failure
# ==================================================
def update_product_naive(cache: NaiveCache, product_id: str, new_price: float):
    """
    DANGEROUS: This can lose updates if cache operation fails.
    """
    # Update database
    # database.update(product_id, price=new_price)  
    
    # Update cache - but what if this fails?
    cached = cache.get(product_id)
    if cached:
        cached.price = new_price
        cache.set(product_id, cached)
        # If cache.set fails, cache has old data but DB has new data
        # Subsequent reads return wrong price until a miss forces refresh
 
 
# PROBLEM 4: Inconsistent reads across caches
# =============================================
"""
Scenario with distributed cache:
- Cache node A has product at price $100
- Cache node B has product at price $75 (more recent)
- User requests are load-balanced between nodes
- Same user sees different prices on page refresh
 
Solution: Sticky sessions, consistent hashing, or broadcast invalidation
"""
 
 
# A better approach: Consistent cache wrapper
class ConsistentCache:
    """
    A more robust cache that handles common consistency issues.
    Still simplified, but demonstrates better patterns.
    """
    
    def __init__(self, ttl_seconds: int, stale_ttl_seconds: int):
        self._cache: dict = {}
        self._ttl = timedelta(seconds=ttl_seconds)
        self._stale_ttl = timedelta(seconds=stale_ttl_seconds)
        self._lock = threading.Lock()
        self._refresh_locks: dict = {}
    
    def get_with_stale(self, key: str, fetch_func):
        """
        Get with stale-while-revalidate pattern.
        Returns stale data immediately while refreshing in background.
        """
        with self._lock:
            entry = self._cache.get(key)
        
        if entry is None:
            # Cache miss - must fetch synchronously
            return self._fetch_and_cache(key, fetch_func)
        
        cached_at, product = entry
        age = datetime.now() - cached_at
        
        if age < self._ttl:
            # Fresh data
            return product
        elif age < self._stale_ttl:
            # Stale but acceptable - return immediately, refresh async
            self._maybe_refresh_async(key, fetch_func)
            return product
        else:
            # Too stale - must refresh synchronously
            return self._fetch_and_cache(key, fetch_func)
    
    def _fetch_and_cache(self, key: str, fetch_func):
        """Fetch from origin and update cache."""
        product = fetch_func()  # Simulated database call
        with self._lock:
            self._cache[key] = (datetime.now(), product)
        return product
    
    def _maybe_refresh_async(self, key: str, fetch_func):
        """
        Trigger async refresh if not already in progress.
        Uses per-key locks to prevent stampede.
        """
        # Implementation would use asyncio or thread pool
        pass
    
    def invalidate(self, key: str) -> None:
        """Explicit invalidation - use when data known to have changed."""
        with self._lock:
            if key in self._cache:
                del self._cache[key]
 
 
print("""
Cache Consistency Challenges:
1. Stale reads are inevitable - design for acceptable staleness windows
2. Stampedes occur on expiration - use stale-while-revalidate or locking
3. Lost updates happen with write-through failures - use atomic operations
4. Distributed caches have inherent consistency delays - acknowledge this
""")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

from dataclasses import dataclass
from enum import Enum
import math
 
 
class EvictionPolicy(Enum):
    LRU = "Least Recently Used"
    LFU = "Least Frequently Used"
    FIFO = "First In First Out"
    RANDOM = "Random Eviction"
 
 
@dataclass
class CacheSizeAnalysis:
    """Analysis of cache size requirements."""
    total_items: int
    avg_item_size_bytes: int
    hot_item_ratio: float  # Fraction of items that are "hot"
    access_distribution_power: float  # Zipf distribution parameter
    target_hit_rate: float
    monthly_cost_per_gb: float = 30.0
 
 
def calculate_cache_requirements(analysis: CacheSizeAnalysis) -> dict:
    """
    Calculate cache size required for target hit rate.
    
    Uses Zipf distribution model common in web access patterns.
    Higher power parameter = more skewed distribution (easier to cache).
    """
    total_size_gb = (analysis.total_items * analysis.avg_item_size_bytes) / (1024**3)
    
    # Zipf distribution: probability of accessing item ranked k
    # P(k) ∝ 1 / k^alpha where alpha is the power parameter
    # This means a small fraction of items handles most traffic
    
    # Estimate cache size needed for target hit rate
    # Using simplified Zipf approximation
    alpha = analysis.access_distribution_power
    
    if alpha <= 1:
        # Flat distribution - need to cache most items
        cache_fraction = analysis.target_hit_rate
    else:
        # Skewed distribution - can cache less
        # Approximation: to get hit rate H, cache fraction = H^(1/alpha)
        cache_fraction = analysis.target_hit_rate ** (1 / alpha)
    
    # Apply hot ratio constraint
    effective_cache_fraction = min(cache_fraction, analysis.hot_item_ratio)
    
    # Calculate actual memory
    cache_size_gb = total_size_gb * effective_cache_fraction
    
    # Add overhead (Redis typically 1.5-2x raw data size)
    cache_size_with_overhead_gb = cache_size_gb * 1.7
    
    # Calculate cost
    monthly_cost = cache_size_with_overhead_gb * analysis.monthly_cost_per_gb
    
    # Predict actual hit rate (may be lower than target if constrained)
    predicted_hit_rate = min(
        analysis.target_hit_rate,
        effective_cache_fraction ** alpha if alpha > 1 else effective_cache_fraction
    )
    
    return {
        "total_data_size_gb": round(total_size_gb, 2),
        "recommended_cache_size_gb": round(cache_size_with_overhead_gb, 2),
        "cache_data_percentage": round(effective_cache_fraction * 100, 1),
        "predicted_hit_rate_percent": round(predicted_hit_rate * 100, 1),
        "monthly_cost": round(monthly_cost, 2),
        "cost_per_hit_rate_point": round(monthly_cost / (predicted_hit_rate * 100), 2),
        "recommendations": generate_sizing_recommendations(
            total_size_gb, cache_size_with_overhead_gb, predicted_hit_rate
        )
    }
 
 
def generate_sizing_recommendations(
    total_gb: float, 
    cache_gb: float, 
    hit_rate: float
) -> list[str]:
    """Generate recommendations based on sizing analysis."""
    recs = []
    
    if cache_gb > 100:
        recs.append("Consider tiered caching: L1 in-memory, L2 Redis, L3 disk")
    
    if hit_rate < 0.85:
        recs.append("Low predicted hit rate - analyze access patterns for optimization")
    
    if cache_gb / total_gb > 0.5:
        recs.append("High cache ratio - evaluate if all this data needs caching")
    
    if cache_gb < 1:
        recs.append("Small cache size - consider in-process caching instead of dedicated server")
    
    return recs if recs else ["Configuration appears reasonable for typical workloads"]
 
 
# Example: E-commerce product catalog
print("Cache Sizing Analysis: E-Commerce Product Catalog")
print("=" * 60)
 
scenarios = [
    ("Small Store", 10_000, 5000, 0.15, 1.2, 0.90),
    ("Medium Store", 500_000, 5000, 0.10, 1.5, 0.90),
    ("Large Retailer", 5_000_000, 8000, 0.05, 1.8, 0.95),
    ("Amazon-scale", 100_000_000, 10000, 0.001, 2.0, 0.99),
]
 
for name, items, size, hot_ratio, alpha, target_hit in scenarios:
    analysis = CacheSizeAnalysis(
        total_items=items,
        avg_item_size_bytes=size,
        hot_item_ratio=hot_ratio,
        access_distribution_power=alpha,
        target_hit_rate=target_hit
    )
    
    result = calculate_cache_requirements(analysis)
    
    print(f"
{name}:")
    print(f"  Total Items: {items:,}")
    print(f"  Total Data: {result['total_data_size_gb']} GB")
    print(f"  Cache Size: {result['recommended_cache_size_gb']} GB")
    print(f"  Predicted Hit Rate: {result['predicted_hit_rate_percent']}%")
    print(f"  Monthly Cost: ${result['monthly_cost']:,.2f}")
    for rec in result['recommendations']:
        print(f"  → {rec}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180

/**
 * Cache invalidation patterns demonstrating different approaches
 * and their trade-offs.
 */
 
interface CacheEntry<T> {
    value: T;
    version: number;
    cachedAt: Date;
    tags: string[];
}
 
interface InvalidationResult {
    success: boolean;
    keysInvalidated: number;
    errors?: string[];
}
 
class InvalidationManager {
    private cache: Map<string, CacheEntry<unknown>> = new Map();
    private tagIndex: Map<string, Set<string>> = new Map();  // tag -> keys
    
    /**
     * Pattern 1: TTL-based expiration (passive invalidation)
     * 
     * Simplest approach - data naturally expires.
     * Trade-off: Guaranteed staleness window equal to TTL.
     */
    setWithTTL<T>(key: string, value: T, ttlMs: number, tags: string[] = []): void {
        const entry: CacheEntry<T> = {
            value,
            version: Date.now(),
            cachedAt: new Date(),
            tags
        };
        
        this.cache.set(key, entry);
        this.indexTags(key, tags);
        
        // Schedule expiration
        setTimeout(() => {
            const current = this.cache.get(key);
            if (current && current.version === entry.version) {
                this.cache.delete(key);
                this.removeFromTagIndex(key, tags);
            }
        }, ttlMs);
    }
    
    /**
     * Pattern 2: Explicit key invalidation (active invalidation)
     * 
     * Caller knows exact key to invalidate.
     * Trade-off: Caller must track all affected keys.
     */
    invalidateKey(key: string): boolean {
        const entry = this.cache.get(key);
        if (entry) {
            this.cache.delete(key);
            this.removeFromTagIndex(key, entry.tags);
            return true;
        }
        return false;
    }
    
    /**
     * Pattern 3: Tag-based invalidation (cascading invalidation)
     * 
     * Invalidate all entries with a specific tag.
     * Useful for: "all products in category X" or "all data for user Y"
     * Trade-off: Tag indexing overhead; must design tags carefully.
     */
    invalidateByTag(tag: string): InvalidationResult {
        const keys = this.tagIndex.get(tag);
        if (!keys || keys.size === 0) {
            return { success: true, keysInvalidated: 0 };
        }
        
        let invalidated = 0;
        for (const key of keys) {
            if (this.invalidateKey(key)) {
                invalidated++;
            }
        }
        
        this.tagIndex.delete(tag);
        return { success: true, keysInvalidated: invalidated };
    }
    
    /**
     * Pattern 4: Version-based caching (avoids race conditions)
     * 
     * Cache key includes version number.
     * When data updates, key changes, old entries naturally orphaned.
     * Trade-off: Must manage version tracking; orphaned entries until eviction.
     */
    getVersionedKey(baseKey: string, version: number): string {
        return `${baseKey}::v${version}`;
    }
    
    setVersioned<T>(
        baseKey: string, 
        version: number, 
        value: T, 
        ttlMs: number
    ): void {
        const versionedKey = this.getVersionedKey(baseKey, version);
        this.setWithTTL(versionedKey, value, ttlMs, [`base:${baseKey}`]);
    }
    
    /**
     * Pattern 5: Compare-and-set invalidation (atomic updates)
     * 
     * Only invalidate if current version matches expected.
     * Prevents race where stale data replaces fresh data.
     * Trade-off: Requires version tracking; may need retry logic.
     */
    compareAndInvalidate(
        key: string, 
        expectedVersion: number
    ): { invalidated: boolean; currentVersion?: number } {
        const entry = this.cache.get(key);
        
        if (!entry) {
            return { invalidated: false };
        }
        
        if (entry.version <= expectedVersion) {
            this.invalidateKey(key);
            return { invalidated: true };
        }
        
        // Current version is newer - don't invalidate
        return { invalidated: false, currentVersion: entry.version };
    }
    
    // Helper methods for tag indexing
    private indexTags(key: string, tags: string[]): void {
        for (const tag of tags) {
            let keySet = this.tagIndex.get(tag);
            if (!keySet) {
                keySet = new Set();
                this.tagIndex.set(tag, keySet);
            }
            keySet.add(key);
        }
    }
    
    private removeFromTagIndex(key: string, tags: string[]): void {
        for (const tag of tags) {
            const keySet = this.tagIndex.get(tag);
            if (keySet) {
                keySet.delete(key);
                if (keySet.size === 0) {
                    this.tagIndex.delete(tag);
                }
            }
        }
    }
}
 
 
// Example: Product catalog with cascading invalidation
const cache = new InvalidationManager();
 
// Cache a product with tags for its category and brand
cache.setWithTTL(
    'product:123',
    { id: 123, name: 'Widget', price: 29.99, categoryId: 5, brandId: 7 },
    60000,  // 1 minute TTL
    ['category:5', 'brand:7', 'type:product']
);
 
// When category changes, invalidate all products in that category
const result = cache.invalidateByTag('category:5');
console.log(`Invalidated ${result.keysInvalidated} entries for category change`);
 
// Version-based approach for avoiding races
cache.setVersioned('product:456', 1001, { name: 'Gadget', price: 99.99 }, 60000);
// Next update uses version 1002, old version 1001 orphaned and will expire