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.

Follow-up questions this case invites — and their answers

Q: Why use consistent hashing with virtual nodes to place keys? A: Plain hash(key) % N remaps almost every key when a node is added or removed, which is effectively a full cache flush and a stampede on the database. Consistent hashing moves only about 1/N of the keys. Virtual nodes (many ring positions per physical node) even out the distribution, and spread a failed node's load across all the others, instead of onto one neighbour.

Q: Should a cache replicate data? A: It depends on the cost of a miss. Pure caches often don't replicate: a lost node means misses that are refilled from the source of truth. If misses are expensive, or the cache holds hot data the database couldn't withstand losing, replicate each shard to a replica (Redis primary–replica, with automatic failover), accepting extra memory and the risk of slightly stale reads from replicas.

Q: How do you handle a hot key that overloads one cache node? A: Detect it with per-key request metrics. Then add a small in-process cache on the application servers for that key, or replicate the key under several suffixed names (key#1..key#n) and read one at random. Request coalescing ensures only one miss per key goes to the database at a time.

Q: What eviction policy should the cache use? A: LRU is a solid default for recency-driven workloads. LFU, or modern approximations such as TinyLFU, protect frequently used keys from being flushed by one-off scans. Always pair eviction with TTLs, so stale data eventually expires, and monitor the eviction rate: a high rate means the cache is undersized for the working set.

Q: How do clients learn about topology changes when nodes join or leave? A: Either clients fetch the cluster map from a configuration service (or the cache cluster itself, as Redis Cluster does with MOVED redirections) and refresh it on errors, or a proxy layer (Twemproxy, Envoy) hides topology from clients entirely. Proxies simplify clients, but add a network hop.

Q: What consistency can clients expect between the cache and the database? A: Usually eventual. With cache-aside and invalidate-on-write, there's a small window in which stale data can be served, bounded by TTLs. For data that must be exact, such as balances or inventory at checkout, read from the database, or use write-through with care. A distributed cache is an accelerator, not the source of truth.