Design a Distributed Cache

Design a distributed in-memory key-value cache similar to Redis or Memcached. The cache must support GET, SET, and DELETE with TTL. It should scale horizontally and handle node failures gracefully.

Capacity Estimation

• Requests: 1M QPS
• Average key-value size: 1 KB
• Total cache size: 1 TB (100 cache nodes × 10 GB each)
• Network: 1M QPS × 1 KB = 1 GB/sec — fit for a 10 GbE network interface

Architecture

Client
  │
  ├── Cache Client Library (consistent hashing router)
  │       │
  │   Consistent hash ring → selects cache node for each key
  │
  ├── Cache Node 1 [10 GB in-memory HashMap + LRU]
  ├── Cache Node 2
  ├── Cache Node 3
  └── Cache Node N

[Coordination Service (ZooKeeper / etcd)]
  → tracks live nodes
  → notifies clients of topology changes

Consistent Hashing: keys are mapped to a virtual ring. Each node owns a range of the ring. Adding/removing a node only remaps keys from the adjacent node — not the entire keyspace.

Eviction policy comparison (implementation level)

• LRU (Least Recently Used): evict the entry that hasn't been accessed longest. Implemented as a hash map + doubly-linked list (see the removeEldestEntry() LRU trick in TreeMap & LinkedHashMap) — O(1) get/put with eviction. The default choice for general-purpose caches, since "recently used" is a reasonable proxy for "likely to be used again."
• LFU (Least Frequently Used): evict the entry with the lowest access count. Better than LRU when access frequency is a stronger signal than recency (e.g. a small set of "hot" keys accessed constantly alongside many one-off keys) — but requires tracking a count per entry and a way to find the minimum efficiently (typically a frequency-bucketed structure, not a simple sort, to stay O(1)).
• FIFO: evict the oldest-inserted entry regardless of access pattern. Cheapest to implement (a plain queue), but ignores usage entirely — a frequently-accessed entry can be evicted right after a burst of new insertions, which is rarely what you actually want for a cache.

LRU is the practical default; LFU earns its extra bookkeeping cost specifically when access patterns are skewed enough that recency alone would evict genuinely "hot" data.

API Design

// Client SDK (not HTTP — internal TCP binary protocol)
cache.get("user:123")              → String | null
cache.set("user:123", json, 3600)  → OK
cache.delete("user:123")           → OK
cache.mget(["k1", "k2", "k3"])   → Map<String, String>

Why not HTTP? HTTP overhead (~200 bytes per request) is significant at 1M QPS. Redis uses a custom binary protocol (RESP) over raw TCP — ~4× lower latency.

Data Structure per Node

HashMap (O(1) GET/SET/DELETE):

HashMap<String, CacheEntry> store;

class CacheEntry {
    byte[] value;
    long expiresAt;   // epoch millis, -1 = no TTL
    LRUNode lruNode;  // pointer into doubly-linked LRU list
}

LRU via Doubly Linked List + HashMap:

• LinkedList maintains access order: MRU at head, LRU at tail
• HashMap for O(1) lookup
• On GET: move accessed node to head
• When memory full: evict tail node

TTL expiration — two strategies:

1. Lazy expiration: check TTL on GET, delete if expired
2. Active expiration: background thread periodically sweeps N random keys and deletes expired ones

Scaling

Adding nodes (horizontal scaling)

1. New node added to ring
2. Coordination service notifies all clients
3. Client library rehashes — only keys in the new node's range need to move
4. Consistent hashing ensures only K/N keys are remapped (K=keys, N=nodes), not all keys

Replication for HA

Each primary cache node has 1 replica:

Primary Node → (async) → Replica Node

On primary failure: replica promoted. Client updated via ZooKeeper watch.

Replication (read scaling) vs partitioning (write/capacity scaling) — different purposes

These solve different problems and are often combined, not substitutes for each other:

• Replication: copies of the same data on multiple nodes. Purpose: read throughput (spread reads across replicas) and availability (a replica takes over if the primary fails, as shown above). Doesn't help capacity — every replica still holds the full dataset.
• Partitioning (sharding, via the consistent hashing ring above): different data on different nodes. Purpose: capacity (total data no longer needs to fit on one node) and write throughput (writes to different keys hit different nodes in parallel). Doesn't inherently improve availability of any single partition's data unless combined with replication.

A production distributed cache typically does both: the keyspace is partitioned across N nodes (via consistent hashing), and each partition is additionally replicated for read scaling and failover within that partition — partitioning solves "the dataset is too big for one node," replication solves "one node's data needs to survive that node's failure and serve more concurrent reads."

• In-memory vs disk: All data in RAM for microsecond latency. If node crashes, data is lost. Acceptable for a cache (DB is source of truth).
• Consistent hashing vs modulo hashing: Modulo (key % N) remaps all keys when N changes. Consistent hashing minimises remapping to 1/N of keys.
• LRU vs LFU eviction: LRU evicts least recently used; LFU evicts least frequently used. LFU is better for hot-key access patterns but more complex.

• Hot keys: a single key hit millions of times/sec saturates one node. Solution: key replication (store hot key on multiple nodes, round-robin reads) or local client-side caching.
• Large values: 10 MB value takes 10 ms to transfer — blocks the connection. Solution: chunk large values or use a dedicated large object store.

• Node crash: consistent hashing remaps affected keys to adjacent node. Cache miss storm — DB sees spike. Mitigate with gradual rehashing.
• Network partition: split brain — two nodes think they own the same key range. ZooKeeper quorum prevents this.
• OOM: LRU eviction activates. If eviction rate exceeds set rate, cache hit rate degrades — alert and scale out.