Design a Distributed Cache

from collections import OrderedDict

class LRUCache:
    """
    LRU Cache using OrderedDict.

    - GET: O(1) - move accessed key to end (most recent)
    - PUT: O(1) - insert at end, evict from front if full
    - Memory: O(capacity)
    """

    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()
        self.hits = 0
        self.misses = 0

    def get(self, key: str):
        """Retrieve value and mark as recently used."""
        if key in self.cache:
            self.hits += 1
            self.cache.move_to_end(key)  # Mark as MRU
            return self.cache[key]
        self.misses += 1
        return None

    def put(self, key: str, value) -> str:
        """Insert or update. Returns evicted key if any."""
        evicted = None
        if key in self.cache:
            self.cache.move_to_end(key)
        else:
            if len(self.cache) >= self.capacity:
                # Evict the least recently used (first item)
                evicted = self.cache.popitem(last=False)[0]
        self.cache[key] = value
        return evicted

    @property
    def hit_rate(self) -> float:
        """Cache hit ratio."""
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0


# Example usage
cache = LRUCache(capacity=3)
cache.put("user:1", {"name": "Alice"})
cache.put("user:2", {"name": "Bob"})
cache.put("user:3", {"name": "Carol"})
cache.get("user:1")    # Moves user:1 to MRU
cache.put("user:4", {"name": "Dave"})  # Evicts user:2 (LRU)
print(f"Hit rate: {cache.hit_rate:.1%}")

import redis
import time
import random

class CacheWithProtection:
    """Cache client with thundering herd protection."""

    def __init__(self):
        self.redis = redis.Redis()
        self.LOCK_TTL = 5  # Lock expires after 5 seconds

    def get_with_lock(self, key: str, fetch_fn, ttl: int = 300):
        """
        Cache-aside with distributed lock to prevent thundering herd.

        1. Try cache
        2. On miss: acquire lock
        3. If lock acquired: fetch from DB, populate cache
        4. If lock not acquired: wait and retry cache
        """
        # Step 1: Try cache
        value = self.redis.get(key)
        if value:
            return value

        # Step 2: Try to acquire lock
        lock_key = f"lock:{key}"
        acquired = self.redis.set(
            lock_key, "1",
            nx=True,  # Only set if not exists
            ex=self.LOCK_TTL
        )

        if acquired:
            # Step 3: We hold the lock - fetch and cache
            try:
                value = fetch_fn()
                self.redis.setex(key, ttl, value)
                return value
            finally:
                self.redis.delete(lock_key)
        else:
            # Step 4: Another request is fetching - wait and retry
            time.sleep(0.05)  # 50ms backoff
            return self.get_with_lock(key, fetch_fn, ttl)

    def get_with_early_refresh(self, key: str, fetch_fn,
                               ttl: int = 300, beta: float = 1.0):
        """
        Probabilistic early expiration (XFetch algorithm).
        Randomly refresh before TTL to prevent simultaneous expiry.
        """
        value = self.redis.get(key)
        remaining_ttl = self.redis.ttl(key)

        if value and remaining_ttl > 0:
            # Probabilistically refresh if close to expiry
            expiry_gap = ttl - remaining_ttl
            if random.random() < beta * math.log(random.random()) * -expiry_gap:
                # Refresh in background
                refresh_async(key, fetch_fn, ttl)
            return value

        # Cache miss: fetch synchronously
        value = fetch_fn()
        self.redis.setex(key, ttl, value)
        return value

from redis.cluster import RedisCluster

# Redis Cluster configuration
# 6 nodes: 3 primaries + 3 replicas
startup_nodes = [
    {"host": "redis-node-1", "port": 6379},
    {"host": "redis-node-2", "port": 6379},
    {"host": "redis-node-3", "port": 6379},
]

rc = RedisCluster(
    startup_nodes=startup_nodes,
    decode_responses=True,
    skip_full_coverage_check=True,
    # Redis Cluster uses 16,384 hash slots
    # distributed across primary nodes
)

# Operations work transparently across cluster
rc.set("user:1001", "Alice")      # Routes to correct shard
rc.get("user:1001")                 # Reads from primary or replica

# Pipeline for batch operations (same hash slot)
pipe = rc.pipeline()
pipe.set("{user:1001}:name", "Alice")   # Hash tag {user:1001}
pipe.set("{user:1001}:email", "[email protected]") # ensures same slot
pipe.execute()