Patterns & Tips

# ---- MISTAKE 1: Using time.time() directly everywhere ----
# BAD: Impossible to test, non-deterministic
class RateLimiter:
    def allow(self):
        if time.time() - self.last > self.window:  # Untestable!
            ...

# GOOD: Inject a clock function for testability
class RateLimiter:
    def __init__(self, clock=time.time):
        self._clock = clock
    def allow(self):
        if self._clock() - self.last > self.window:  # Testable!
            ...


# ---- MISTAKE 2: Not handling capacity=0 ----
# BAD:
class LRUCache:
    def put(self, key, val):
        if len(self._cache) >= self._capacity:  # Fails if capacity=0
            self._cache.popitem(last=False)
        self._cache[key] = val

# GOOD:
class LRUCache:
    def put(self, key, val):
        if self._capacity <= 0:
            return  # Cache with 0 capacity stores nothing
        ...


# ---- MISTAKE 3: Forgetting to handle duplicate keys ----
# BAD: LRU cache counts duplicate puts as new entries
# GOOD: Check if key exists before checking capacity


# ---- MISTAKE 4: Thread-unsafe code without mentioning it ----
# BAD: Silently ignoring concurrency
# GOOD: Either use locks or explicitly state "single-threaded"
import threading

class ThreadSafeCache:
    def __init__(self):
        self._lock = threading.Lock()
        self._cache = OrderedDict()

    def get(self, key):
        with self._lock:
            return self._cache.get(key)


# ---- MISTAKE 5: Not cleaning up expired state ----
# BAD: Rate limiter stores timestamps forever
# GOOD: Clean up old entries periodically or lazily


# ---- MISTAKE 6: O(n) operations where O(1) is expected ----
# BAD: Finding least recently used by scanning all entries
# GOOD: Use OrderedDict.popitem(last=False) for O(1)


# ---- MISTAKE 7: Inconsistent return types ----
# BAD:
def get(self, key):
    if key in self._cache:
        return self._cache[key]  # Returns value
    return False                  # Returns boolean!

# GOOD:
def get(self, key) -> Optional[Any]:
    return self._cache.get(key)   # Always returns value or None


# ---- MISTAKE 8: Not validating input in constructor ----
# BAD: Silently accepting invalid config
class RateLimiter:
    def __init__(self, max_requests):
        self._max = max_requests  # What if max_requests is -1?

# GOOD: Fail fast
class RateLimiter:
    def __init__(self, max_requests):
        if max_requests < 0:
            raise ValueError("max_requests must be non-negative")
        self._max = max_requests


# ---- MISTAKE 9: Modifying data structures during iteration ----
# BAD:
for key in self._store:
    if self._is_expired(key):
        del self._store[key]  # RuntimeError!

# GOOD:
expired = [k for k in self._store if self._is_expired(k)]
for key in expired:
    del self._store[key]


# ---- MISTAKE 10: Not providing get_stats() or __repr__() ----
# Adding observability methods shows production experience
def get_stats(self):
    return {
        "size": len(self._cache),
        "hit_rate": self._hits / max(1, self._hits + self._misses),
        "evictions": self._eviction_count,
    }

# ---- Testing Strategy for System Implementations ----

# 1. BASIC FUNCTIONALITY: Happy path works
cache = LRUCache(capacity=3)
cache.put("a", 1)
assert cache.get("a") == 1, "Basic get after put"

# 2. EVICTION: Correct item is evicted
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
cache.put("d", 4)  # Should evict "a"
assert cache.get("a") is None, "LRU item should be evicted"
assert cache.get("d") == 4, "New item should exist"

# 3. EDGE CASES:
# - Empty state: get on empty cache returns None
# - Capacity 0: all puts are no-ops
# - Capacity 1: every put evicts the previous item
# - Duplicate keys: update in place, don't double-count
# - Same key overwritten multiple times

# 4. ORDERING: Verify correct ordering behavior
cache = LRUCache(capacity=3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
cache.get("a")       # "a" is now most recent
cache.put("d", 4)    # Should evict "b", NOT "a"
assert cache.get("b") is None, "b should be evicted (LRU after a was accessed)"
assert cache.get("a") == 1, "a should still exist"

# 5. TIME-BASED TESTING: Inject a mock clock
class MockClock:
    def __init__(self):
        self._time = 0.0
    def __call__(self):
        return self._time
    def advance(self, seconds):
        self._time += seconds

clock = MockClock()
limiter = RateLimiter(max_requests=3, window=60, clock=clock)

assert limiter.allow("user1") == True
assert limiter.allow("user1") == True
assert limiter.allow("user1") == True
assert limiter.allow("user1") == False  # Rate limited

clock.advance(61)  # Advance past window
assert limiter.allow("user1") == True   # Window reset

# 6. CONCURRENCY: Test thread safety
import threading

cache = ThreadSafeCache(capacity=100)
errors = []

def writer(start):
    for i in range(start, start + 100):
        try:
            cache.put(f"key_{i}", i)
        except Exception as e:
            errors.append(e)

threads = [threading.Thread(target=writer, args=(i*100,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

assert len(errors) == 0, f"Thread safety errors: {errors}"

# 7. STRESS TEST: Large volume
cache = LRUCache(capacity=1000)
for i in range(100000):
    cache.put(f"key_{i}", i)
assert cache.size() == 1000, "Should not exceed capacity"

# ---- Complete Component Cheat Sheet ----

# LRU CACHE
# Data: OrderedDict
# get(): lookup + move_to_end -> O(1)
# put(): insert + popitem(last=False) if over capacity -> O(1)

# LFU CACHE
# Data: key_to_val + key_to_freq + freq_to_keys(OrderedDict) + min_freq
# get(): lookup + update frequency bucket -> O(1)
# put(): insert + evict min_freq LRU item -> O(1)

# FIXED WINDOW RATE LIMITER
# Data: counter + window_start per client
# allow(): check window, increment counter -> O(1)

# TOKEN BUCKET RATE LIMITER
# Data: tokens + last_refill per client
# allow(): refill based on elapsed time, consume token -> O(1)

# SLIDING WINDOW RATE LIMITER
# Data: deque of timestamps per client
# allow(): remove old timestamps, count remaining -> O(cleanup)

# MESSAGE QUEUE
# Data: dict of messages + deque of message IDs
# enqueue(): append to deque -> O(1)
# dequeue(): pop from deque, set visibility timeout -> O(1)

# PRIORITY SCHEDULER
# Data: heapq with (priority, sequence, task)
# schedule(): heappush -> O(log n)
# next(): heappop (skip cancelled) -> O(log n)

# PUB/SUB
# Data: defaultdict(dict) for topic -> subscribers
# publish(): iterate subscribers, call callbacks -> O(subscribers)
# subscribe(): add to dict -> O(1)

# CONSISTENT HASHING
# Data: sorted list of hash positions + hash_to_node map
# get_node(): bisect to find position -> O(log n)
# add_node(): insert num_replicas positions -> O(r * log n)

# BLOOM FILTER
# Data: bit array of size m
# add(): set k hash positions -> O(k)
# contains(): check k hash positions -> O(k)

# CIRCUIT BREAKER
# Data: state enum + counters + timestamps
# allow_request(): check state -> O(1)
# record_failure(): increment counter, maybe trip -> O(1)

# RETRY WITH BACKOFF
# delay = base * 2^attempt (exponential)
# delay = random(0, delay) (full jitter)
# delay = min(delay, max_delay) (cap)