Caching: Strategy, Redis Internals & Distributed Patterns

import redis
import time
import uuid
from contextlib import contextmanager

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# ── Sliding window rate limiter ──────────────────────────────────────────

def is_rate_limited(user_id: str, max_requests: int = 100,
                    window_seconds: int = 60) -> bool:
    """
    Sliding window rate limiter using Sorted Set.
    Score = timestamp (float). Members = unique request IDs.
    O(log n) per request.
    """
    key = f"ratelimit:{user_id}"
    now = time.time()
    window_start = now - window_seconds

    pipe = r.pipeline()
    # Remove requests outside the window
    pipe.zremrangebyscore(key, 0, window_start)
    # Count remaining requests in window
    pipe.zcard(key)
    # Add current request (timestamp as both score and member prefix)
    pipe.zadd(key, {f"{now}:{uuid.uuid4().hex[:8]}": now})
    # Set TTL to auto-cleanup
    pipe.expire(key, window_seconds + 1)
    results = pipe.execute()

    current_count = results[1]  # count BEFORE this request
    return current_count >= max_requests

# ── Distributed lock (Redlock-lite) ─────────────────────────────────────

@contextmanager
def distributed_lock(resource: str, ttl_seconds: int = 30, retry_ms: int = 100):
    """
    Acquire a distributed lock using SET NX EX.
    
    Safety: SET NX EX is atomic. Lock expires automatically if holder crashes.
    UUID as value ensures only the lock holder can release it.
    Limitation: single-node only (not true Redlock). Use Redlock for multi-node.
    """
    lock_key = f"lock:{resource}"
    lock_value = str(uuid.uuid4())
    acquired = False

    try:
        # Retry until lock acquired (with backoff in production)
        for _ in range(10):
            # SET key value NX EX seconds — atomic, only sets if key doesn't exist
            acquired = r.set(lock_key, lock_value, nx=True, ex=ttl_seconds)
            if acquired:
                break
            time.sleep(retry_ms / 1000)

        if not acquired:
            raise TimeoutError(f"Could not acquire lock for {resource}")

        yield lock_value

    finally:
        if acquired:
            # Lua script: atomic check-and-delete (prevent releasing another holder's lock)
            release_script = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
            """
            r.eval(release_script, 1, lock_key, lock_value)

import hashlib
import bisect

class ConsistentHashingCache:
    """
    Distributes cache keys across multiple Redis nodes using consistent hashing.
    
    Why not modulo hashing (key % N)?
    → Adding/removing one node remaps ~N/2 of all keys → cache stampede.
    
    Why consistent hashing?
    → Adding/removing one node remaps only K/N keys (K = keys, N = nodes).
    → With virtual nodes (vnodes), load is evenly distributed.
    """
    
    def __init__(self, nodes: list[str], vnodes: int = 150):
        """
        nodes: list of "host:port" strings
        vnodes: virtual nodes per physical node (higher = more even distribution)
        """
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
        
        for node in nodes:
            for i in range(vnodes):
                virtual_key = f"{node}:{i}"
                h = self._hash(virtual_key)
                self.ring[h] = node
                self.sorted_keys.append(h)
        
        self.sorted_keys.sort()
    
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_node(self, key: str) -> str:
        """Find the responsible node for a key — O(log N) binary search."""
        h = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0  # wrap around the ring
        return self.ring[self.sorted_keys[idx]]
    
    def add_node(self, node: str, vnodes: int = 150):
        """Add a new node. Only ~K/N keys are remapped."""
        for i in range(vnodes):
            virtual_key = f"{node}:{i}"
            h = self._hash(virtual_key)
            self.ring[h] = node
            bisect.insort(self.sorted_keys, h)
    
    def remove_node(self, node: str, vnodes: int = 150):
        """Remove a node. Only ~K/N keys are remapped."""
        for i in range(vnodes):
            virtual_key = f"{node}:{i}"
            h = self._hash(virtual_key)
            if h in self.ring:
                del self.ring[h]
                self.sorted_keys.remove(h)