Cache Stampede¶

import time
import redis

cache = redis.Redis()

def get_user(user_id: str) -> dict:
    key = f"user:{user_id}"
    cached = cache.get(key)
    if cached:
        return json.loads(cached)

    # threading.Lock() only coordinates threads in this process, not across pods.
    # Distributed single-flight: SET NX across the fleet.
    lock_key = f"lock:{key}"
    acquired = cache.set(lock_key, "1", nx=True, ex=10)
    if acquired:
        try:
            cached = cache.get(key)  # double-check after lock
            if cached:
                return json.loads(cached)
            result = db.query("SELECT * FROM users WHERE id = ?", user_id)
            cache.setex(key, 300, json.dumps(result))
            return result
        finally:
            cache.delete(lock_key)

    time.sleep(0.05)  # lost the race — wait, then re-read
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    result = db.query("SELECT * FROM users WHERE id = ?", user_id)
    cache.setex(key, 300, json.dumps(result))
    return result

import random

def cache_set(key: str, value: any, base_ttl: int = 300):
    # Add ±20% jitter to TTL
    jitter = random.randint(-base_ttl // 5, base_ttl // 5)
    cache.setex(key, base_ttl + jitter, json.dumps(value))

import time
import threading

def get_with_stale(key: str, fetch_fn, ttl=300, stale_ttl=30):
    data = cache.get(key)
    meta = cache.get(f"{key}:meta")

    if data and meta:
        meta = json.loads(meta)
        age = time.time() - meta['set_at']
        if age < ttl:
            return json.loads(data)  # fresh
        elif age < ttl + stale_ttl:
            # Stale but serve immediately, refresh in background
            threading.Thread(target=_refresh, args=(key, fetch_fn, ttl)).start()
            return json.loads(data)  # stale but fast

    # Cache miss — fetch synchronously
    return _refresh(key, fetch_fn, ttl)

def _refresh(key, fetch_fn, ttl):
    result = fetch_fn()
    cache.setex(key, ttl + 30, json.dumps(result))
    cache.setex(f"{key}:meta", ttl + 30, json.dumps({'set_at': time.time()}))
    return result

import math
import random

def get_with_early_recompute(key: str, fetch_fn, ttl=300, beta=1.0):
    result = cache.get(key)
    if result:
        data, expiry = json.loads(result)
        remaining = expiry - time.time()
        # Increase recompute probability as TTL runs out
        if -beta * math.log(random.random()) < remaining:
            return data  # serve from cache

    # Recompute (either expired or early recompute triggered)
    fresh = fetch_fn()
    cache.setex(key, ttl, json.dumps([fresh, time.time() + ttl]))
    return fresh