S063 · Caching — Cache-Aside, Write-Through, TTLs, Invalidation

"""s063_cache.py — production-shaped cache-aside in ~100 lines.
 
Run:  docker run -d --name s063-redis -p 6379:6379 redis:7-alpine
      pip install redis
      python s063_cache.py
"""
import json
import random
import time
from contextlib import contextmanager
 
import redis
 
# ---- 1. wire up ----
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
 
# ---- 2. the "slow" data source we're caching ----
_call_count = {"db": 0}
 
def slow_db_lookup(user_id: int) -> dict | None:
    """Simulates a 200 ms DB query. Returns None for user_id > 1000 (not found)."""
    _call_count["db"] += 1
    time.sleep(0.2)
    if user_id > 1000:
        return None
    return {"id": user_id, "name": f"user-{user_id}", "email": f"u{user_id}@ex.com"}
 
# ---- 3. constants ----
TTL_SECONDS         = 300         # 5 min for real data
NEGATIVE_TTL        = 30          # 30 s for "does not exist" — shorter, so a newly-created row shows up quickly
LOCK_TTL            = 5           # stampede lock lives 5 s max
NEGATIVE_SENTINEL   = "__MISS__"  # marker for "we looked, it wasn't there"
 
# ---- 4. the stampede lock ----
@contextmanager
def redis_lock(key: str, ttl: int = LOCK_TTL):
    """SET NX EX — classic Redis lock. If we don't get it, we block briefly."""
    token = f"{time.time()}-{random.random()}"
    got = r.set(key, token, nx=True, ex=ttl)
    try:
        yield bool(got)
    finally:
        # Only delete if we still own it (avoid stealing another holder's lock)
        if got and r.get(key) == token:
            r.delete(key)
 
# ---- 5. the cache-aside read ----
def get_user(user_id: int) -> dict | None:
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
 
    if cached == NEGATIVE_SENTINEL:
        return None                        # negative-cache hit
    if cached is not None:
        return json.loads(cached)          # positive hit
 
    # ---- miss path with stampede protection ----
    lock_key = f"lock:{cache_key}"
    with redis_lock(lock_key) as got_lock:
        if not got_lock:
            # Someone else is filling. Wait briefly, then re-check the cache.
            time.sleep(0.05)
            cached = r.get(cache_key)
            if cached == NEGATIVE_SENTINEL:
                return None
            if cached is not None:
                return json.loads(cached)
            # Fall through — worst case we do the DB call ourselves
 
        value = slow_db_lookup(user_id)
        if value is None:
            r.set(cache_key, NEGATIVE_SENTINEL, ex=NEGATIVE_TTL)
            return None
        r.set(cache_key, json.dumps(value), ex=TTL_SECONDS)
        return value
 
# ---- 6. the write path (invalidate on write) ----
def update_user(user_id: int, patch: dict) -> None:
    # In real life this would write to Postgres. We just invalidate.
    print(f"[write] updating user {user_id} in DB (imagined)")
    r.delete(f"user:{user_id}")            # simplest safe invalidation
 
# ---- 7. demo ----
if __name__ == "__main__":
    r.flushdb()
 
    print("--- cold cache: first call hits DB ---")
    t0 = time.perf_counter()
    print(get_user(42))
    print(f"took {(time.perf_counter()-t0)*1000:.1f} ms  db_calls={_call_count['db']}")
 
    print("\n--- warm cache: 100 calls, ~1 ms each ---")
    t0 = time.perf_counter()
    for _ in range(100):
        get_user(42)
    print(f"took {(time.perf_counter()-t0)*1000:.1f} ms total, db_calls={_call_count['db']}")
 
    print("\n--- write invalidates the cache ---")
    update_user(42, {"name": "renamed"})
    t0 = time.perf_counter()
    print(get_user(42))
    print(f"took {(time.perf_counter()-t0)*1000:.1f} ms  db_calls={_call_count['db']}")
 
    print("\n--- negative cache: not-found result is remembered ---")
    print(get_user(9999)); print(get_user(9999)); print(get_user(9999))
    print(f"db_calls after 3 not-found reads: {_call_count['db']}")