SYSTEM DESIGN · BUILDING BLOCKS · 3 OF 14

Distributed Cache: Eviction, Sharding, Stampedes and Hot Keys

The cache tier as a component you can defend in an interview: what is inside one cache node (hash table, LRU list, expiry), how keys are spread across a cluster, safe invalidation, the four classic failures (stampede, hot key, penetration, avalanche) and their fixes, what happens when a node dies, and how to size memory and hit rate.

  1. 1
  2. 2
  3. 3
  4. 4
  5. 5
  6. 6
  7. 7
  8. 8
  9. 9
  10. 10
  11. 11
  12. 12
  13. 13
  14. 14
1 / 15

Press play to hear the narration, or step through at your own pace.

Notes

The problem

The problem: fifty thousand profile reads a second, and the database does real work for every one, a few milliseconds each, competing with writes for CPU and connections. But reads are skewed; a small set of keys gets most of the traffic. A memory cache answers in well under a millisecond. If it absorbs ninety-five percent of reads, the database only sees misses and writes.

Why a cache tier
  • DB read: ~2-20 ms, limited
  • by CPU, disk and connections
  • cache read: ~0.2-0.5 ms
  • in the same data centre
  • reads are skewed: 1% of keys
  • often take most of the traffic
  • goal: absorb 90-99% of reads
  • so the DB sees only misses
  • and writes

Read path

The default pattern is cache-aside. The app asks the cache for user forty-two. On a miss it reads the database and writes the value back with a time to live. The cache is just memory that can be lost at any time without losing data, which is exactly what makes it safe. Everything else in this block is about the edges of these five lines.

Cache-aside
v = cache.get("user:42")
if v is None:
    v = db.query(42)
    cache.set("user:42", v,
              ttl=600 + jitter)
return v

app owns the logic; cache is
just memory, can be lost

Inside a node

Inside one node there's a hash table from key to entry, giving constant-time lookups. When memory is full, something must go. Classic LRU links entries in a doubly linked list, moves each hit to the front and evicts from the tail. Redis approximates this by sampling a handful of keys and evicting the oldest, which is cheaper. LFU counts hits instead, so a one-off scan can't flush your hot keys. Expired keys are removed lazily on read and by a background sweep.

Inside one cache node
  • hash table: key → entry, O(1)
  • LRU: doubly linked list;
  • each hit moves entry to head,
  • evict from tail when full
  • (Redis samples ~5 keys and
  • evicts the oldest: cheaper)
  • LFU: count hits; keeps hot
  • keys during a scan
  • TTL: lazy check + background

Partitioning

One node isn't enough memory, so keys are spread across a cluster with consistent hashing, from Concepts episode twenty-two. Each node owns many virtual points on a ring, and adding a node moves only its share of keys. Routing lives in a client library, in a proxy like mcrouter, or in the server: Redis Cluster splits keys into sixteen thousand three hundred eighty-four hash slots and redirects clients that ask the wrong node.

Spreading keys over nodes
  • hash(key) on a ring with
  • ~100-200 virtual nodes each
  • add a node: ~1/N of keys move
  • (mod N would move almost all)
  • where routing lives:
  • client library (fewest hops)
  • proxy (twemproxy, mcrouter)
  • server-side: Redis Cluster
  • 16,384 hash slots

Invalidation

On a write, update the database first, then delete the key rather than setting it. If two writers both set the cache, they can land out of order and leave the wrong value forever. Even delete has a race: a slow reader that fetched the old row can put it back just after your delete. A TTL bounds how long that lasts, and leases fix it properly. With many writers, invalidate from the database's change stream instead of from app code.

Keeping it correct
  • 1 update the database
  • 2 DELETE the cache key
  • (not SET: two writers can
  • set out of order)
  • race: a slow reader can put
  • an old value back after DEL
  • → short TTL, or leases
  • many writers? invalidate from
  • the DB change stream (CDC)

Stampede

Failure one: the stampede. A key read three thousand times a second expires, and in the next few milliseconds three thousand requests miss and all run the same expensive query. Fix it with single flight: the first miss takes a short lock and rebuilds, everyone else waits briefly or gets the stale value. Or serve stale while one background refresh runs. Or refresh early, with a probability that rises as the TTL runs out.

Cache stampede fixes
  • single flight: one request
  • rebuilds (lock key with
  • SET NX EX 5), others wait
  • or get the stale value
  • stale-while-revalidate: serve
  • old value, refresh async
  • early refresh: refresh with
  • rising probability as TTL
  • nears zero

Hot keys

Failure two: the hot key. Consistent hashing puts each key on exactly one node, so a celebrity's profile at two hundred thousand reads a second melts that node while the others sit idle. Two fixes. Add a tiny in-process cache on each app server with a few seconds' TTL, so most reads never leave the box. Or replicate the key under ten suffixes on ten nodes and read a random copy. First you have to detect it, by sampling keys on the client.

Hot key fixes
  • a single key lives on ONE
  • node; ~100k ops/s is its limit
  • L1: in-process cache with
  • 1-5 s TTL on every app server
  • replicate: store as key#0..#9
  • on 10 nodes, read a random one
  • detect: client-side sampling,
  • redis-cli --hotkeys

Penetration

Two more. Penetration: a bot asks for ids that don't exist, so every request misses and goes to the database. Cache the 'not found' result for a minute, or check a Bloom filter of valid ids first. Avalanche: you loaded a million keys at the same time with the same TTL, so they all expire together an hour later. Add random jitter to every TTL so expiry is spread out.

Penetration and avalanche
  • penetration: requests for
  • keys that do not exist
  • (random ids, bots) always miss
  • → cache "not found" for 60 s
  • → Bloom filter of valid ids
  • avalanche: many keys expire
  • at the same moment (bulk load)
  • → TTL = base + random jitter
  • both protect the DB, not the cache

Node failure

When node B dies, its third of the keys remap to neighbours that don't have them, and the database gets a burst of misses. With three nodes, that's a third of your traffic suddenly uncached. Give each shard a replica that is promoted in seconds; replication is asynchronous, so a few recent writes may be lost, which is fine for a cache. And size the database so it survives losing one cache node, or you have a hidden single point of failure.

When a cache node dies
  • its ~1/N of keys remap to
  • neighbours, which are cold
  • → miss spike on the database
  • Redis: replica per shard,
  • Sentinel/Cluster promotes it
  • in seconds (async: may lose
  • last writes; fine for cache)
  • size the DB to survive the
  • loss of one cache node

Variants

The variants. In-process caches answer in nanoseconds but each server has its own copy, so invalidation is hard; keep TTLs short. Memcached is simple and multi-threaded, strings only. Redis adds data structures like sorted sets and hashes, replication and optional persistence, so it doubles as a rate limiter or leaderboard. On writes, write-through updates cache and database together; write-behind batches writes later and can lose data.

Variants
  • in-process (Caffeine, Guava):
  • ns reads, per-server copy,
  • hard to invalidate everywhere
  • Memcached: simple, multi-
  • threaded, strings only
  • Redis: data structures, TTLs,
  • replication, persistence
  • write-through: write cache+DB
  • write-behind: risky, batches

Sizing

Sizing. A hundred million users with two-kilobyte profiles is two hundred gigabytes, but you only need the hot twenty percent, forty gigabytes, plus about thirty percent for overhead and fragmentation. Four shards of sixteen gigabytes, each with a replica. Then the hit rate: at ninety-five percent the database sees two and a half thousand reads a second. At ninety percent it sees five thousand, double. Small hit-rate changes move the database a lot.

Sizing
100M users × 2 KB profile
  = 200 GB total
hot 20% → 40 GB
+ 30% overhead → ~52 GB
→ 4 shards × 16 GB + replicas

reads 50,000/s, hit rate 95%
→ DB sees 2,500/s
hit rate 90% → 5,000/s (2x!)

one Redis shard: ~100k ops/s

Scaling

To scale, add shards for more memory or throughput, read from replicas for read-heavy shards if slightly stale reads are fine, and lean on the local cache for the hottest keys. Watch hit rate, evictions per second, memory, p ninety-nine latency and the top keys. Rising evictions with a falling hit rate means the hot set no longer fits: add memory.

Scaling and watching it
  • more memory or ops: add
  • shards (moves ~1/N keys)
  • read-heavy shard: read from
  • replicas (slightly stale)
  • watch: hit rate, evictions/s,
  • memory, p99 latency, conns,
  • top keys
  • rising evictions + falling
  • hit rate = add memory

Say it like this

In the interview, be specific. Cache-aside, keys like user colon id, ten-minute TTL with jitter. A Redis cluster with replicas per shard. Writes update the database and delete the key. Single flight for stampedes, a local cache and key replication for hot keys, and a database that survives one lost shard. Then give the hot-set size and the hit rate you expect. That's a cache you've actually defended.

In the interview
  • what: cache-aside, keys like
  • user:{id}, TTL 10 min + jitter
  • where: Redis cluster, N shards
  • with replicas
  • writes: update DB, delete key
  • stampede: single flight
  • hot keys: L1 + replication
  • failure: DB sized for 1 lost
  • shard, misses are safe
  • numbers: hot set, hit rate

Interview questions

How would you implement an LRU cache?

Use a hash map from key to a node in a doubly linked list. On every access move the node to the head, insert new entries at the head, and when capacity is reached evict the tail, which keeps get and put at O(1).

Why delete the cache key on write instead of updating it?

Two concurrent writers that each set the cache can apply in a different order than their database writes, leaving a stale value with no expiry in sight. Deleting forces the next reader to load the current row, and a TTL bounds any remaining race.

What is a cache stampede and how do you prevent it?

When a popular key expires, many concurrent misses all recompute it and overload the database. Prevent it with single flight (a short lock so one request rebuilds), serving stale data while one refresh runs, or probabilistic early refresh before expiry.

How do you handle a hot key in a distributed cache?

Detect it by sampling, then add a short-TTL in-process cache on the app servers and, if needed, replicate the key under several suffixes on different nodes and read a random copy, so no single node takes all the load.

What happens when a cache node fails?

Its share of keys remaps to other nodes as misses, so the database sees a spike. Mitigate with a replica per shard that is promoted automatically, warm-up where possible, and a database sized to absorb the loss of one shard.

What is cache penetration?

Repeated requests for keys that do not exist bypass the cache every time. Cache negative results with a short TTL or check a Bloom filter of valid keys before querying the database.

Redis or Memcached?

Memcached is a simple multi-threaded string cache that scales well for plain key-value caching. Redis adds rich data structures, replication, persistence and scripting, so it is the usual choice when the cache also serves counters, leaderboards, rate limits or sessions.

How do you size a cache?

Estimate the hot working set (number of hot keys times average value size), add roughly 30 percent for overhead, split across shards with replicas, and check that the expected hit rate leaves the database with a miss rate it can handle.