System design deep dive · HLD
Design a Distributed Cache (Redis / Memcached at scale): full high-level design.
Requirements, back-of-envelope capacity estimation, a high-level component diagram, a deeper deployment architecture built around a consistent-hash ring with virtual nodes, sequence diagrams for a normal GET and for cache-stampede prevention, and an entity-relationship model for the cluster's control-plane metadata - with the reasoning an interviewer expects behind every box and arrow.
1. Clarify requirements before drawing any box
A distributed cache is infrastructure, not a feature bolted onto one app - other services depend on it staying fast and available. That changes the requirements conversation: the hard problems are memory pressure, uneven load across nodes, and what happens the instant a node dies.
Functional requirements
Non-functional requirements
2. Back-of-the-envelope capacity estimation
These numbers decide the shape of the cluster: how many physical nodes are needed just to hold the data, how many more for replication, and whether a single node's CPU/network budget can actually keep up with its share of the traffic.
| Metric | Assumption | Resulting estimate |
|---|---|---|
| Hot working set | 2 TB of data must stay in RAM cluster-wide | Basis for every node-count calculation below |
| Average value size | ~1 KB per cached value | 2TB ÷ 1KB ≈ 2 billion keys |
| Usable memory per node | 64 GB usable out of ~128GB physical (headroom for OS, metadata, fragmentation) | 2TB ÷ 64GB ≈ 32 shards (physical partitions) minimum |
| Replication factor | RF = 3 (1 primary + 2 replicas per shard, different AZs) | 32 shards × 3 ≈ 96 node-instances in the cluster |
| Traffic | 10:1 read:write ratio | 5,000,000 GET/sec and 500,000 SET/sec cluster-wide |
| Per-node read QPS | Reads can be served by any of the 3 copies of a shard | 5,000,000 ÷ 96 ≈ ~52,000 reads/sec per node-instance |
| Per-node write QPS | Writes land on 32 primaries, then fan out to 2 replicas each | 500,000 ÷ 32 ≈ ~15,600 writes/sec per primary, replicated to 2 more nodes |
| Network bandwidth per node | ~52,000 reads/sec × 1KB payload | ≈ 50 MB/s (~400 Mbps) sustained - comfortable on a 10Gbps NIC, until a single hot key concentrates traffic (see deep dive) |
3. High-level design (HLD)
The HLD names the major components and how a request is routed, without committing yet to how many AZs or exactly how gossip propagates membership - that level of detail belongs in the architecture diagram in the next section.
What each box owns
Client library (the "smart client")
Embeds the consistent-hashing logic directly in the calling application: it hashes the key, walks the ring to find the owning node, and talks to that node directly. It periodically refreshes its view of the ring from the coordinator so it stays correct as nodes join or leave, without needing a proxy hop on every request.
Coordinator / metadata store
The durable source of truth for cluster topology - which node owns which hash range, current membership, and replica assignments. Backed by etcd or ZooKeeper for strong consistency, it is updated by the failure detector on failover and read (or gossiped) by every client library and node to build their local view of the ring.
Shard nodes and the eviction manager
Each shard node holds one partition of the keyspace entirely in RAM (a hash table, essentially) and answers GET/SET/DELETE with no disk round-trip. When a node's memory fills up, its eviction manager applies an LRU or LFU policy to reclaim space for new writes before ever returning an out-of-memory error.
Replica nodes and the failure detector
Each shard's writes are copied to 2 replica nodes on different physical hosts (and ideally different AZs). A failure detector exchanges heartbeats with every node; when a primary stops responding, it tells the coordinator to promote one replica to primary and update the ring, so client libraries reroute traffic within a bounded window instead of the whole shard going dark.
4. Detailed architecture diagram: the hash ring and replica placement
This is where consistent hashing earns its keep. Keys are hashed onto points on a ring; each physical node owns many small arcs of that ring (virtual nodes) instead of one large contiguous arc, so load spreads evenly and a node joining or leaving only reshuffles the handful of keys near its own virtual nodes - not the whole keyspace.
| Decision | Choice | Reasoning |
|---|---|---|
| Partitioning scheme | Consistent hashing with virtual nodes, not modulo (hash % N) | Modulo hashing remaps almost every key when N changes; consistent hashing only remaps the small arc near the node that joined or left - critical for scale-out without a cache-wide cold start. |
| Routing layer | Client-side "smart client" hashing, not a proxy tier | Going direct to the owning node saves a network hop and avoids a proxy becoming a shared bottleneck or single point of failure; the cost is every client must keep its ring view fresh. |
| Replication mode | 1 synchronous replica + 1 asynchronous replica per shard | The sync replica guarantees an acknowledged write survives a primary failure (durability); the async replica adds a third copy for read fan-out and geographic spread without paying its latency on every write. |
| Membership protocol | Gossip between nodes, backed by a coordinator as source of truth | Gossip (SWIM-style) detects failures in ~1s across hundreds of nodes with no single point of failure; the coordinator still holds the authoritative ring so a network partition can't leave two views permanently disagreeing. |
5. Sequence diagrams for the two critical flows
A sequence diagram is where an interviewer checks whether you actually understand call order, what is computed locally versus over the network, and how the system behaves under contention - not just on the happy path.
5.1 Normal GET request (ring lookup, direct routing)
Step 2 is the whole point of the smart client: no network call, just a hash and a binary search over the local ring. Step 4's alt branch shows load-spreading reads across replicas - useful for a hot key, at the cost of possibly reading a value that hasn't caught up with the very latest write yet, which is why it's opt-in per read rather than the default.
5.2 Cache stampede prevention (request coalescing on a hot key)
The critical detail is step 2: the first request to see a miss on a given key acquires a per-key lock and becomes the sole "leader" fetching from the origin database. Steps 4 and 5 show the requests that arrive microseconds later - instead of also querying the database, they wait on the leader's in-flight result. Three concurrent misses on a hot key produce exactly one database query, not three (or three thousand). Step 7's TTL jitter spreads out future expirations so the same key doesn't create a synchronized stampede again next time.
6. Entity-relationship (ER) diagram: the control-plane data model
The cached values themselves are unstructured key-value pairs living in RAM - they never touch a relational schema. But the cluster's control plane tracks structured metadata about shards, replicas, and (in sampled/aggregated form) key-level stats, and that metadata does fit a small relational or strongly-consistent store.
Key modeling decisions
| Data | Profile | Where it lives and why |
|---|---|---|
| Control-plane metadata (shards, replica_nodes) | Thousands of rows, low write rate, needs strong consistency for correct routing | A small relational store or etcd/ZooKeeper - simplicity and consistency matter far more than raw throughput here. |
| The cached values themselves | Billions of keys, sub-millisecond latency requirement, best-effort durability | Purpose-built in-memory store only. A general-purpose database - SQL or NoSQL - adds disk I/O and query planning overhead that blows the latency budget by 10-100x. |
| key_metadata analytics (hot-key stats) | Approximate, high write rate, read only for dashboards/rebalancing | In-memory sketches per node, rolled up into a time-series or column store - never a row-per-key relational table. |
7. Deep dives interviewers actually probe
Why virtual nodes, and how does ring lookup actually work?
With only one point per physical node on the ring, load is lumpy: whichever node happens to own the largest arc gets the most keys, and when one node leaves, its entire arc dumps onto exactly one neighbor. Giving each physical node many virtual points (100-200 is typical) around the ring spreads its share of the keyspace into many small, scattered arcs, so both the steady-state load and the reshuffling on a membership change are distributed evenly across the rest of the cluster instead of concentrated on one neighbor.
// Ring is a sorted array of (hashPoint, physicalNodeId); lookup is a binary search
ring = sorted([(hash(nodeId + ":" + v), nodeId) for nodeId in nodes for v in range(150)])
function ownerOf(key):
h = hash(key)
idx = binarySearch(ring, h) // first ring point >= h
if idx == len(ring): idx = 0 // wrap around the ring
return ring[idx].nodeIdHow exactly does request coalescing stop a cache stampede?
Every shard node keeps a small table of per-key locks (a "single-flight" group) for keys currently being refetched. The first request that misses on a key takes the lock and becomes the leader; every other concurrent request for that same key attaches to the same in-flight future instead of issuing its own database query. Combined with TTL jitter (randomizing each key's expiry by roughly ±10%) so thousands of keys set at the same moment don't all expire in the same millisecond, this turns an N-way stampede into a single origin query.
function getOrFetch(key):
if cache.has(key): return cache.get(key)
lock = perKeyLocks.acquireOrJoin(key) // first caller creates it, others attach
if lock.isLeader:
value = originDB.fetch(key)
cache.set(key, value, ttl = baseTtl + jitter(±10%))
lock.fulfill(value)
perKeyLocks.release(key)
else:
value = lock.awaitResult() // waits, never re-queries the DB
return valueLRU vs LFU eviction - which one, and when?
LRU (least recently used) evicts whatever hasn't been touched in the longest time; it's cheap to implement (a doubly linked list plus a hash map, O(1) per access) and works well when recency predicts future access - session data, recently viewed items. LFU (least frequently used) tracks access counts and evicts the coldest item by frequency; it protects consistently popular keys from being evicted by a short burst of one-off traffic (a crawler scanning millions of cold keys) but costs more to maintain and can be slow to let go of items that were hot yesterday but are cold today ("cache pollution"). Many production caches (Redis included) default to an approximated LRU or a hybrid LFU with decay specifically to get LFU's pollution resistance without full-precision counters on every key.
How does replication and failover avoid two nodes both thinking they're primary?
The synchronous replica acknowledges a write before the client is told it succeeded, so no acked write is ever lost to a primary crash; the asynchronous replica trades that guarantee for lower write latency and extra read capacity. On failover, the failure detector doesn't unilaterally promote a replica - it reports the suspected failure to the coordinator, which owns a monotonically increasing epoch (fencing token) per shard. The coordinator picks the replica with the least lag, bumps the epoch, and writes the new assignment; any old primary that comes back online with a stale epoch has its writes rejected by peers, which is what prevents split-brain (two nodes both accepting writes as "the" primary) rather than relying on the old primary noticing on its own that it's been replaced.
What breaks first at 10x scale?
Not aggregate throughput - that scales by adding shards and node-instances, the whole point of consistent hashing. The real bottleneck is a single hot key: if one key (a celebrity profile, a viral product page) draws a disproportionate share of the 5M reads/sec, all of that traffic lands on the one shard - and within it, effectively one primary plus its 2 replicas - regardless of how many of the other 95 node-instances sit idle. Fixing this means detecting hot keys (via the sampled key_metadata access_count) and handling them specially: replicating just that key to extra read-only nodes beyond the shard's normal replica set, or letting the client library cache it locally for a few seconds, rather than trying to solve it by adding more shards.
Post a Comment
Add