Design a Distributed Cache

Every app that gets popular ends up with the same problem. The database is the slowest, most expensive thing in the system, and everyone keeps asking it the same questions. A cache is a shelf of ready answers that sits in front of it. This note is about building that shelf so it is too big for one machine, and still keeps working when a machine dies.

1. Problem Statement

Design a cache cluster that many application servers share. Think Redis or Memcached, but run as a service for a whole company.

An app server asks for a key, like product:4411. If the cache has it, the answer comes back in about a millisecond. If not, the app goes to the database, and puts the answer on the shelf for the next person.

The hard parts are the ones you only meet once there is more than one machine:

• Which machine holds which key?
• What happens when one machine dies, or when you add a new one?
• What if one key is so popular that one machine melts?
• What if a popular key expires and ten thousand requests rush to the database at once?
• How do you stop the cache from showing old data for too long?

If you can talk about each of these with a number, you have a good answer.

2. Questions to Ask First

Ask these out loud before you draw anything. Each answer changes the design, and it shows the interviewer you do not guess.

The last row matters more than it looks. If only one service writes, updating the cache is easy. When five services write, you need a plan for the writes the cache never saw.

3. Requirements

3.1 Functional Requirements

3.2 Non Functional Requirements

1. Very low latency. A cache hit should be under 2 ms at p99. Slower than that and the cache is barely better than the database.
2. High hit rate. We want 95 percent or better. Each missed read costs a database query.
3. Scales sideways. When the data or traffic doubles, we add machines, we do not buy a bigger one.
4. Survives a node failure. One machine dying must not take the site down.
5. Bounded staleness. Old data is allowed, but only for a limited time that we can state.
6. Protects the database. Whatever goes wrong with the cache, the database must not get a wall of traffic it cannot handle.

4. Back of the Envelope Numbers

Small steps, so you can redo them on a whiteboard.

Traffic first.

reads at peak          = 500,000 per second
cache hit rate         = 95 percent
reads that miss        = 500,000 x 5 percent = 25,000 per second to the database
without any cache      = 500,000 per second to the database

So the cache takes the database load from 500,000 a second down to 25,000. That is 20 times less. Say this number out loud, it is the reason the cache exists.

Now memory.

objects                = 200 million
size per object        = 1 KB value + about 200 B for key and bookkeeping = about 1.2 KB
raw data               = 200M x 1.2 KB = 240 GB

One cache node is a machine with 64 GB of RAM. We cannot use all of it. The operating system, network buffers and the copy made when the node saves a snapshot all need room. I would plan to fill about 40 GB of it with data.

nodes for memory       = 240 GB / 40 GB = 6

Now check that the traffic fits too. One Redis node can do roughly 100,000 simple operations per second. We want to run at about 60 to 80 percent of that.

total ops              = 500,000 reads + 50,000 writes = 550,000 per second
with 8 primaries       = 550,000 / 8 = about 69,000 per node

That is 69 percent of one node's limit. Tight but fine. Memory said 6, traffic says 8, so we take the bigger number: 8 primary nodes.

Each primary gets one replica, a live copy on another machine.

total nodes            = 8 primaries + 8 replicas = 16

Network is no worry: 550,000 ops x 1.2 KB is about 5.3 Gbit per second in total, under 1 Gbit per node.

Last, what a bad change costs. This is the number I like to say in the interview.

add a 9th node with plain "hash mod N":
  keys that stay in place   = 1 / 9 = about 11 percent
  keys that move            = about 89 percent
  hit rate drops from 95 to about 11 percent x 95 = 10 percent
  database load             = 500,000 x 90 percent = 450,000 per second (18 times normal)

add a 9th node with consistent hashing:
  keys that move            = 1 / 9 = about 11 percent
  hit rate                  = 95 x 89 percent = about 85 percent
  database load             = 500,000 x 15 percent = 75,000 per second (3 times normal)

The first one takes the database down. The second one hurts a bit and recovers. That gap is why we do not use plain hash mod N.

5. High Level Design

Only three nodes are drawn. The real cluster has eight primaries and eight replicas, but the picture is the same.

5.1 Walk Through of Each Box

App server and client library. The app does not talk to nodes directly. It calls a small library that hashes the key, finds the right node, sends the command, and retries if needed. The library is where most of the clever behaviour lives.

Local cache. A tiny cache inside the app server's own memory, holding a few hundred of the hottest keys for one or two seconds. We will see in the deep dives why this one small box saves us from the hot key problem.

Cluster map. A table that says "keys in this range live on node A, this range on node B". The library keeps a copy in memory and refreshes it when the cluster changes. It is tiny, a few kilobytes.

Primary nodes. Each primary owns a slice of the keys. It holds them in memory, answers reads and writes for them, and removes old keys when memory is full.

Replica nodes. Each primary has one replica that copies every write a moment later. The replica does not serve traffic in the normal case. It is a spare. If the primary dies, the replica takes over.

Cluster manager. A small group of three or five processes. They send heartbeats to every node (a heartbeat is a tiny "are you alive" message), decide when a node is dead, promote a replica, and publish the new cluster map. Three or five, never two, so they can always get a majority vote and never split into two groups that each think they are in charge.

Source database. The truth. The cache only keeps copies.

5.2 Mapping a Key to a Node

You already know consistent hashing, so here is the short version for this design.

Picture a clock face with positions from 0 to about 4 billion. Hash each node name onto the clock at many places. We give each node 128 points ("virtual nodes"), so 8 nodes make 1,024 points. To find the home of a key, hash the key onto the same clock and walk clockwise to the first node point. That node owns the key.

Why 128 points and not one? With one point per node, the slices are uneven and one node might own 30 percent of the keys. With 128 points the slices average out and every node ends up within a few percent of one eighth.

Redis Cluster does the same idea with a fixed table instead of a ring. It cuts the key space into 16,384 slots, uses CRC16(key) mod 16384 to find the slot, and gives each node a range of slots. With 8 nodes, each owns 2,048 slots. Adding a node means handing over some slots. It behaves like consistent hashing and is easier to move around, because a slot is something you can point at and say "this one is moving".

Either is a fine answer. I would say the ring in the interview and mention slots as how Redis does it.

5.3 Client Library or Proxy

Someone has to do the "which node?" work. There are two places to put it.

My opinion: start with the smart client. It is faster and has fewer moving parts. Move to a proxy when you have so many apps, in so many languages, that keeping the library correct in all of them becomes the real pain. Twemproxy and Envoy are common proxies, and Redis Cluster itself is built around the smart client idea (nodes reply "MOVED, ask node B" and the client updates its map).

The number that pushes people toward proxies is connections. 500 app servers each holding 20 connections to each of 16 nodes is 160,000 connections. Each connection costs the node some memory and attention. A layer of 10 proxies cuts that to 10 x 20 x 16 = 3,200 on the node side.

For the rest of this note I assume the smart client.

5.4 Eviction Policy on Each Node

A node has a memory limit, 40 GB here. When it is full and a new key arrives, the node must throw something out. Which key it picks is the eviction policy.

I pick allkeys-lru. Real Redis does an approximation: it samples 5 random keys and removes the oldest of those. It is almost as good as true LRU and costs almost nothing, because a true LRU needs a linked list updated on every read.

If a batch job reads every product once a night, LRU lets those cold reads push out the genuinely hot keys. LFU resists that, since a key read once has a low count. So the rule I would say aloud: LRU first, switch to LFU if you see scans hurting the hit rate.

Eviction runs on each node on its own. There is no central place deciding. That is a feature. Nobody has to coordinate, and one busy node cannot slow the others.

5.5 How Expiry Works

A key with a TTL has to disappear on time. The lazy way is to check on every read: if the TTL has passed, treat it as a miss and delete it. Cheap, but a key nobody reads would sit there forever using memory.

So nodes also do it actively. Ten times a second the node picks 20 random keys that have a TTL, deletes the expired ones, and if more than a quarter of them were expired it picks another 20 and repeats. That keeps the amount of dead data small without scanning everything.

Together: lazy so that readers never see an old value, active so that memory is not wasted.

6. API Design

6.1 Commands the Apps Use

These look like Redis commands because that is what most people know.

A real call from the library for a product page:

GET product:4411:v3
-> "{\"id\":4411,\"name\":\"Steel bottle\",\"price\":1299}"

Key names carry a version, here v3. If we ever change the shape of the stored object, we bump the version in the code, and the old keys just stop being asked for. No mass delete needed.

One warning about MGET. The library splits the keys by node and sends the requests in parallel, so a batch of 100 keys over 8 nodes is only as fast as its slowest node. Keep batches small.

6.2 Admin API

Used by the cluster manager and by operators, never by apps.

The stats call is not an extra. The first thing you look at in a cache is the hit rate, and the second is the eviction count. If evictions keep climbing, the cache is too small.

7. Data Model

A cache has no tables in the normal sense, but three things are stored and each has a shape.

7.1 What One Cache Entry Holds

The value holds a version number from the database row, as row_version. We use it in section 9.3 so an old value can never overwrite a newer one. If your database table has no version column, add one. It costs 8 bytes per row.

7.2 Cluster Metadata

This is stored by the cluster manager, in a small consistent store (etcd, ZooKeeper, or the managers' own replicated state).

nodes

slots

The epoch is the important column. Clients send the epoch they know. If a node sees an older one, it tells the client its map is stale and sends the new one. Without it, two parts of the system can disagree about who owns a slot and neither can tell which is right.

8. Key Flows

8.1 Read with Cache Aside

Cache aside means the app, not the cache, is in charge. The app checks the cache, falls back to the database, and fills the cache itself.

Two things in here are easy to skip and both matter. The lock on a miss means one app fetches from the database while the others wait a moment, instead of all of them hammering it. And the TTL has "jitter", a small random amount added, so keys written together do not all expire together. Both are explained in 9.2.

8.2 Write and Invalidate

On a write, we change the database first, then delete the cache key. We do not try to put the new value into the cache.

The second delete, from the change stream, catches writes made by other services that never touched our app code, and it also covers the small race described in 9.3. It is a safety net, not the main path.

8.3 A Primary Node Dies

During the gap, the library treats an error from node B as a miss and goes to the database. That is safe, as long as the database is protected, which we cover in the deep dive and in section 10.

9. Deep Dives

9.1 Joining, Leaving and Dying Without a Disaster

The problem. The cluster changes all the time. You add a node for growth, remove one for cost, and sometimes a machine just dies. Every one of those changes moves keys, and moved keys are misses. Too many misses at once and the database falls over.

Option one: hash(key) mod N. Simple, and it works until N changes. Going from 8 to 9 nodes moves about 89 percent of keys. We worked this out in section 4: the database load jumps 18 times. I would never run this for a cluster that changes.

Option two: consistent hashing with virtual nodes. Adding a node takes a small slice from every existing node, about 1 / 9 of the keys in total. The rest stay put.

The choice is consistent hashing, and here is what happens in each event.

A node joins.

1. The new node starts empty and is marked JOINING. Clients do not use it yet.
2. The cluster manager picks 128 points for it (or a share of the slots in the Redis style).
3. The keys in those slices are copied across in the background from their current owners, at a capped speed, say 50 MB per second, so the live traffic is not hurt.
4. Once the copy is done, the manager bumps the epoch and publishes the new map. Clients start sending those keys to the new node.
5. The old owners delete the copied keys.

With 240 GB of data in the cluster, a new node takes 240 / 9 = about 27 GB. At 50 MB per second that is 27,000 MB / 50 = 540 seconds, about 9 minutes. That is fine for a planned change.

Many simple systems skip the copy and let the new node start empty. Then those 11 percent of keys are misses, and the database sees about 75,000 reads a second instead of 25,000 until the cache warms up. That is survivable, but I would rather copy.

A node leaves on purpose. Reverse of the above. Mark it DRAINING, copy its keys to the nodes that will own them, update the map, then shut it down. No misses at all.

A node dies. No time to copy anything. This is where the replica earns its place.

1. Heartbeats stop. After 5 seconds with no answer, the manager marks the node SUSPECTED.
2. A majority of managers must agree. That stops one manager with a bad network from declaring a healthy node dead.
3. The replica is promoted. It already holds nearly all the data.
4. The map epoch goes up and clients switch.

Total time without service for that slice, about 5 to 8 seconds. In those seconds, one eighth of all keys are unreachable. About 62,500 reads per second normally land on that node, and 95 percent of them are hits. Those 59,000 hits become database reads. The database normally sees 25,000, so it would jump to about 84,000, more than three times normal. That is exactly why the stampede protection in the next section is not optional.

One more detail. Replication is asynchronous, which means the primary answers the client first and copies to the replica a moment later. If the primary dies in that moment, the last few milliseconds of writes are lost. For a cache this is acceptable, since the database still has the truth. The danger is a lost delete: a key that was deleted on the old primary comes back from the replica holding an old value. The TTL puts a ceiling on how long that stays wrong, and the change stream delete from 8.2 usually repairs it sooner.

9.2 Hot Keys and Cache Stampede

These are two different problems that feel similar. A hot key is too many reads on one key. A stampede is too many misses at one moment. Both end with one thing falling over, so I treat them together.

The problem with hot keys. Say a phone launch makes one product page go viral. That key gets 80,000 reads per second. A key lives on exactly one node, so that node must serve all 80,000, on top of its normal 69,000. It tops out at 100,000 and starts dropping every other request on that node, including the thousands of unrelated keys that happen to live next to it. The other 7 nodes are bored. Adding more nodes does nothing, because the hash still sends that key to one place.

Two options for hot keys.

I choose the local cache first. It collapses 80,000 requests per second into about 200, and that is a 400 times reduction for almost no code. The library counts reads per key in a small sample, and when a key crosses a limit (say 100 per second on that one app server) it starts keeping that key locally for 2 seconds. I would reach for key splitting only if the key is too big or must be fresher than a couple of seconds.

The problem with a stampede. A popular key expires. In the next few milliseconds, thousands of requests all see a miss and all go to the database. Say the key gets 20,000 reads per second and the database needs 50 ms to build the value. In those 50 ms, 20,000 x 0.05 = 1,000 requests miss and 1,000 identical queries run. If the query is heavy, the database slows down, so the refill takes longer, so even more requests pile up. It feeds itself.

Its cousin is the mass expiry. A nightly job loads 5 million keys, all with TTL 3600. One hour later all 5 million expire in the same second.

Three tools, used together.

1. Single flight, a lock on the miss. On a miss, try SET lock:key NX EX 5. Only the winner queries the database. The others wait 50 ms and read the cache again. Now 1,000 queries become 1. The lock has its own 5 second TTL, so a crashed winner cannot block the key forever.
2. Jitter on every TTL. Instead of 3600, use 3600 plus a random number from 0 to 360. The 5 million expiries now spread across 6 minutes, about 14,000 a second instead of 5 million at once.
3. Serve stale while refreshing. Store a soft expiry inside the value, a bit earlier than the real TTL. When a reader sees the soft time has passed, it still returns the old value at once, and one reader (the one who got the lock) refreshes it in the background. Nobody waits, and the key never really goes missing.

The choice: all three, because they cost little and cover different cases. Single flight handles one hot key. Jitter handles bulk loads. Serve stale makes the user wait for nothing.

There is a third cousin worth one line: cache penetration, asking for keys that do not exist at all. A bot asking for product ids that were never created misses every time and goes to the database every time. The fix is to cache the "not found" answer for 30 seconds, as a real entry.

9.3 Staying Close to the Database

The problem. The cache holds copies. The moment the database changes, some copy is old. We cannot make that zero without making every write slow. So the real question is how old, for how long, and for which data.

Two options for a write.

I choose delete. Updating looks faster, but two writes to the same row can reach the cache in a different order from how they reached the database. Then the cache holds the older price, and it stays that way until the TTL. Deleting has no order problem: whoever comes next reads the database, and the database is always right.

Delete is not perfect either. Here is the race, with times.

t=0   Reader A misses, reads price 1299 from the database
t=1   Writer B updates the database to 1199
t=2   Writer B deletes the cache key
t=3   Reader A, slow, now writes 1299 into the cache

Now the cache holds 1299 and the database holds 1199. Nothing will fix this except the TTL. It is rare, because step 3 has to be slower than the whole of steps 1 and 2, maybe one in a hundred thousand reads. But with 25,000 misses a second, rare happens daily.

Ways to close the gap, in order of how much I like them.

1. Always set a TTL. This makes every stale value temporary. 10 minutes for ordinary data, 30 seconds for prices or stock. If you remember one line from this section, remember this one.
2. Compare versions. Every value stores the database row_version. When writer B deletes the key, it also leaves a short marker that says "version 18 exists", kept for 10 seconds. Reader A's late write carries version 17, sees the marker, and is refused. This needs a tiny script that runs on the node in one step, so nobody can sneak in between the check and the write.
3. Delayed second delete. Delete again about 500 ms later, longer than the slowest normal read. Easy, and catches most of the race.
4. A change stream from the database. A small service reads the database change log and deletes the matching keys. This also catches writes from services that do not know the cache exists. It is the most complete one and the most work to run.

My choice: TTL always, delete on write, and a change stream for data that many services write. Use versions only for the few keys where wrong data costs money, like stock counts. I would not build all four on day one.

What to tell the interviewer plainly: a cache in front of a database is eventually consistent, and we choose how long "eventually" is with the TTL. For a price, 30 seconds. For a profile photo, an hour. For a bank balance, do not cache it.

10. Failure Handling

Two rules sit behind this table.

The cache must never be the reason the site is down. Every call has a short timeout, and a timeout means "go to the database", never "show an error". The only exception is when the database itself is in trouble, and then a stale answer is better than none.

The database must be protected from the cache failing. When the cache is gone, every request comes at once, and the database will not survive 500,000 reads per second. The circuit breaker, the per key lock and a global cap on database concurrency are what stop a cache problem from turning into a total outage. People forget this part, and it is the part that decides whether a cache failure is a bad afternoon or a front page story.

11. Tradeoffs

12. Interview Talking Points

Say these roughly in this order. Each one takes about a minute.

• Start with the reason for the cache and put a number on it. 500,000 reads a second, a 95 percent hit rate, so the database sees 25,000 and not 500,000. That is the whole point of the system.
• Do the sizing in front of them. 200 million objects at 1.2 KB is 240 GB. Memory says 6 nodes, traffic says 8, so you take 8, and add a replica for each. Showing that you pick the larger of two answers looks careful.
• Explain key placement in two sentences. Consistent hashing, 128 virtual nodes each, so a new node moves 1 in 9 keys. Then say why not mod N: it would move 89 percent and push the database to 18 times normal load.
• Choose a smart client and say when you would use a proxy. Mention the connection count math (160,000 against 3,200) so it sounds earned, not memorised.
• Walk through what happens when a node dies: heartbeats, a majority vote, replica promoted in about 5 to 8 seconds, and what the app does in the gap. Then say plainly that replication is async and a few writes can be lost, and why that is fine for a cache.
• Bring up hot keys and stampede on your own, before they ask. Local cache for hot keys, single flight plus jitter plus serve stale for stampede. These show you have seen a real cache hurt.
• For database consistency, say delete on write, always set a TTL, and name the race with a time line. Then say how long wrong data can live, because it is the TTL.
• Close with what you would watch: hit rate, evictions, p99 latency, memory per node, and database load. Say that a drop in hit rate is the first sign of almost every problem above.