Design a distributed cache

Design a distributed cache like Memcached or Redis Cluster. Start with one cache node in front of the database, grow it with a hash ring, a replica per shard and invalidation from the change log, then stress it with the stale set race, hot keys, failover, eviction and ten times the traffic, with code.

Mid-level28 min core · 46 min full depth

Read first: Caching Consistent hashing LRU cache
Scaling patterns: Hot keys and hot partitions Timeouts, retries and circuit breakers
Building blocks: Caching Consistent hashing Partitioning Replication Consistency models Queues and streams
Related: Key-value store Distributed rate limiter

“Design a distributed cache.” Sometimes it’s “design Memcached” or “design Redis Cluster”.

The prompt sounds like a data structure question, scaled up. It’s really a question about protecting a database. One cache node is a hash map with an eviction policy, which the LRU cache page builds. This page is about everything around it. Unlike a key-value store, a cache may lose data. Its dangers are stale values and cold nodes.

So this page starts with one cache node in front of the database, and lets the numbers grow it. Then it tests the result against stale values, hot keys, failover and a full memory.

The answer in one minute

If you only have a minute, this is the design. The rest of the page explains each part.

Shard with a hash ring. 64 cache nodes, each owning 160 points on a ring. Adding a node moves only a small share of keys.
Give every shard a replica in another zone. Not for durability: the database holds everything. Because losing a third of the cache at once would send the database ten times what it can take.
Delete on write, then delete again from the change log. Writers delete the key after they commit. And an invalidator, reading the database’s log of commits, deletes it again. So freshness doesn’t depend on every writer being correct.
Use leases on fills. A miss gets a token, and a delete voids it. That closes the race that caches an old value for an hour, and turns a stampede on one key into one database read.
Treat hot keys and growth as database problems. A near cache in each app and copies of hot keys spread the load. New nodes warm up from old ones before they own traffic.

Scope it

Two questions decide whether this is a cache or a database: what sits behind it, and how stale a read may be.

Interviewerminute 0

Design a distributed cache.

What they are testing

Whether you ask what sits behind the cache and how stale a read may be. Three minutes. Jumping to “Redis Cluster with consistent hashing” before knowing the data size or the staleness budget is the red flag.

These four questions matter most:

What’s behind it? A sharded relational database that holds every value.
One app or a shared platform? A shared fleet for about 50 services in one region, across 3 data center zones.
How much data and traffic? 5 billion items, 2 million reads a second, 6 million at peak.
How stale may a read be after a write? About a second usually, never more than a few seconds.

From these answers we make five promises:

Fast. A get in under 1 ms on the server, under 2 ms seen by a client.
Never breaks a page. 99.99% of gets return a value or a clean miss within 20 ms. A cache error becomes a miss, never a failed page.
Protects the database. At least 95% of reads hit. Losing any one node, or a whole zone, keeps database reads under the 500,000 a second it’s sized for.
Fresh. After a write commits, readers see the new value within 1 second at p99, and never later than the key’s expiry, whatever fails.
Durable: not at all. Every item can be rebuilt from the database.

We leave out durable storage, server-side data structures, and web or CDN caching.

Go deeperMore questions, and what each changes1 min

Question	We assume	If the answer were different
Is the cache the only copy?	No, the database has everything	Losing data isn’t allowed: the key-value store problem
One app?	A shared platform	One app can use a managed Redis Cluster and stop there
What do values look like?	Opaque bytes, 500 bytes on average, at most 1 MB	Server-side lists, sets and counters point at Redis
One region or many?	One region per cache	Invalidation across regions becomes part of the design
Values over 1 MB?	No: those belong in blob storage	Values split in chunks, and shards sized by network

The numbers that decide it

Now we work out the numbers. A cache’s numbers matter for what they do to the database.

Assumptions

Live items: 5 billion
Average value size: 500 bytes
Average key size: 50 bytes
Metadata per item: 50 bytes
Cache reads a second, average: 2 million
Peak over average: 3×
Hit ratio: 95%
Usable memory per node: 48 GB
Operations a second per node: 300,000

Memory3 TB64 primaries

Each item is a 50-byte key, a 500-byte value and about 50 bytes of overhead. A 64 GB node gives about 48 GB to items.

So we memory sets the node count: 63 primaries, rounded to 64.

Throughput6.6M/sops at peak; 35% busy

6 million reads plus 10% writes. One node serves about 300,000 operations a second with small values.

So we the fleet is memory-bound, and the spare CPU absorbs hot keys and failovers.

Database load300,000/smisses at peak

At a 95% hit ratio, against 500,000 of capacity. One cold shard adds about 90,000 a second, which fits. One cold zone adds about 1.9 million, which doesn’t.

So we every shard needs a warm replica in another zone: 128 nodes and 6 TB.

Network63 MB/sper primary

Nothing for a 10 Gbit card. But one key read a million times a second is 600 MB/s on its own, half that card.

So we hot keys are a network problem as much as a CPU one.

The surprise is that replication is sized from the database, not from durability. Try the calculator: at a 90% hit ratio the database is past its limit. At 2.5 times the reads, throughput takes over from memory and sets the shard count.

Change an assumption

Live items5 billionAverage value size500 bytesAverage key size50 bytesMetadata per item50 bytesCache reads a second, average2 millionPeak over average3×Hit ratio95%Usable memory per node48 GBOperations a second per node300,000

Memory per item: 600 bytes
Metadata share of each item, in %: 8.3; Small enough to ignore at this value size.
Working set in memory: 3 TB; More than one node holds: memory sets the shard count.
Primaries for memory: 63; The page rounds 63 up to 64.
Reads, peak: 6,000,000/s
Operations, peak (10% writes): 6,600,000/s
Peak load per primary, in % of capacity: 35; Memory-bound: the spare CPU is what absorbs hot keys and failovers.
Database reads at peak (the misses): 300,000/s; Inside the 500,000 reads/s the database is sized for.
Extra database reads if one shard goes cold: 90,000/s; The database absorbs one cold shard.
Extra database reads if one AZ goes cold: 1,900,000/s; Far past the database headroom: every shard needs a warm replica in another AZ.
Memory with one replica per shard: 6 TB
Network per primary, peak: 63 MB/s; Nothing on a 10 to 25 Gbit NIC. One hot key can still saturate one.

Build the design

We start with the simplest design that is correct. Then we let the numbers break it, one at a time. Each step fixes only what broke, adds only the boxes that fix needs, and says what the fix costs.

Step 1. One cache node, cache-aside

The smallest version is one cache node between the app and the database. Even this version must handle reads, writes and a slow cache correctly.

Distributed cache · architecture

Stage 1: One cache node, cache-aside

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Cache reads per second: 6,000,000

Stage 1 of 6

One cache node, cache-aside

Why this stage

Nothing yet.

This is the smallest design that keeps reads off the database.

What we add

One in-memory cache node with an LRU, and app servers that read it first and fill it from the database on a miss.

What it costs

One node holds 48 GB of a 3 TB working set and serves about 300,000 operations a second.

It is also a single point of failure.

At 6M cache reads per second, Cache primaries is 2000% busy. Stage 2 is the fix.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

The app reads the cache first. On a miss, it reads the database, and stores the value with an expiry (a TTL). The TTL is capped at 24 hours, with a little random jitter, so keys written together don’t expire together. On a write, it commits, then deletes the key. The next reader refills it.

Every call has a 20 ms deadline. A timeout or a refused connection is returned to the app as a miss. Gets are never retried, because a retry storm on a slow node makes it slower. Deletes are retried, since a lost delete leaves stale data.

This version is correct. Now the numbers push on it. One node can’t hold 3 TB, a dead node sends its misses to the database, and some writers forget to delete.

Step 2. Shard with a hash ring

Every client must map a key to the same node. And adding a node should move few keys, because every moved key is a miss on the database. hash(key) mod 64 fails that. Adding a 65th node moves about 98% of keys.

So we use consistent hashing. Each shard owns 160 points on a ring of 2⁶⁴ values, called virtual nodes. A key belongs to the first point clockwise from its hash. Adding a shard moves about 1/65 of keys, a little from each. With one point per shard, some shards would own several times what others do. With 160, a shard’s share varies by about 8%. A bigger node gets more points.

Ring.javaring.pyRing.csring.goring.ts

// The client's map from key to shard. Every client builds it from the same ring config, so each
// one computes the owner of a key locally, with no lookup service on the request path.
class Ring {
    record Shard(String id, int weight) {} // weight 1 per 48 GB node; a 96 GB node gets 2

    // 160 points per unit of weight, ketama's number: a shard's share of keys then varies by
    // about 1/sqrt(160), 8%. With one point each, some shards would own several times others.
    static final int POINTS_PER_WEIGHT = 160;

    final long epoch;             // ring version: requests carry it, so a node that just lost a
                                  // range can refuse keys from a client still on the old ring
    private final long[] points;  // sorted positions on the ring
    private final Shard[] owners; // owners[i] owns the keys hashing to (points[i-1], points[i]]

    Ring(long epoch, List<Shard> shards) {
        this.epoch = epoch;
        TreeMap<Long, Shard> ring = new TreeMap<>();
        for (Shard s : shards)
            for (int i = 0; i < s.weight() * POINTS_PER_WEIGHT; i++)
                ring.put(hash(s.id() + "-" + i), s); // the point names depend only on the shard,
                                                     // so adding a shard leaves the others in place
        points = ring.keySet().stream().mapToLong(Long::longValue).toArray();
        owners = ring.values().toArray(new Shard[0]);
    }

    // The first point at or after the key's hash owns it. Binary search: O(log 10,000) on
    // every request, about 14 comparisons, so routing costs nothing next to the network.
    Shard shardFor(String key) {
        int i = Arrays.binarySearch(points, hash(key));
        if (i < 0) i = -i - 1;
        return owners[i == points.length ? 0 : i]; // past the last point: wrap to the first
    }

    // MD5 because every client language must compute the same number for the same key.
    // Never String.hashCode: one client disagreeing means reads that always miss.
    static long hash(String s) {
        try {
            byte[] d = MessageDigest.getInstance("MD5").digest(s.getBytes(StandardCharsets.UTF_8));
            return ByteBuffer.wrap(d).getLong(); // first 8 bytes, big-endian
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException(e);
        }
    }
}

Routing runs in the client library. The ring is built from a config record in etcd. It lists shards and their addresses, under an epoch number. Clients watch it, and swap rings when the epoch changes. A get costs one hash and a binary search, a microsecond. It goes straight to the owning node, with no extra hop. And an etcd outage can’t stop routing.

Distributed cache · architecture

Stage 2: Shard with a hash ring

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 2 of 6

Shard with a hash ring

Why this stage

At the 6 million reads/s peak one node is 2,000% busy, and 3 TB of items needs 63 nodes of memory.

With a 60% hit ratio the database is far past its 500,000 reads/s.

What we add

64 cache shards placed on a consistent-hash ring with 160 virtual nodes each, a client library that routes every key locally, and an etcd-backed ring config.

What it costs

Every client must hold the same ring.

A dead shard is 1/64 of the keyspace gone cold until it is replaced.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

Go deeperWhy 64 shards, and why not bigger nodes1 min

We could hold 3 TB on fewer, larger machines. Eight nodes with 400 GB each would fit. But each shard is also a unit of failure. When one of eight nodes dies, an eighth of all reads miss at once. That’s about 750,000 extra database reads a second, more than the database can take. With 64 shards, one dead node costs about 90,000, which it can absorb.

Smaller shards also warm up faster after a restart and move less data when the ring changes. The cost is more machines to run and more connections from each client. Around 50 GB per node is a common middle point: big enough to be cheap per gigabyte, small enough that losing one is a shrug.

Step 3. A replica per shard

One dead primary sends about 90,000 extra reads a second to the database, which it can absorb. A zone outage takes a third of the primaries at once, about 1.9 million extra reads a second, and the database falls over. So the cache must survive a zone with its contents.

Each shard gets one asynchronous replica in a different zone from its primary. Primaries are spread evenly across the three zones, so a zone outage never takes both copies of a shard.

Reads move first. A client that gets five errors in a row from a primary reads that shard from the replica, milliseconds behind. The database sees almost none of the dead shard’s traffic.
Promotion follows, fenced by a lease. A cluster manager checks every node each second. Each primary holds a 10-second lease and stops serving when it can’t renew. The manager promotes the replica only after the old lease has run out, so two primaries never serve one shard.

Distributed cache · architecture

Stage 3: A replica per shard

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 3 of 6

A replica per shard

Why this stage

Losing one AZ without replicas turns a third of the cache cold: about 1.9 million extra database reads a second against 500,000 of capacity.

This stage is forced by a failure domain, not by load.

What we add

An asynchronous replica of every shard in another AZ, and a cluster manager that health-checks nodes, promotes replicas and publishes a new ring epoch.

What it costs

Twice the memory, about 10 seconds before a replica is promoted (reads move to it at once), and replication lag: a failover can lose the last few milliseconds of sets and deletes.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

The cost is double the memory, about 6 TB in total. That is the price of keeping the database alive through a zone outage.

Go deeperWhy not a synchronous replica1 min

A synchronous replica would confirm every set and delete before the primary answers. Then a failover would lose nothing. But every write would wait for a round trip to another zone, about 1 to 2 ms. That’s ten times the cost of the write itself. Caches take millions of sets a second, so this would need far more machines.

There is a cheaper answer to the one write that matters. A lost set only costs a miss later. A lost delete leaves a stale value. So we keep the replica asynchronous, and after a failover the invalidator replays the last 30 seconds of the change log for that shard. The failover stress test walks through it.

Step 4. Invalidation from the change log

Writers delete after they commit. That fails when the delete times out, the writer crashes, or a batch job writes the table directly. So an invalidator reads the database’s log of commits (through Kafka), and deletes every cache key derived from each changed row, typically within a second. Each team registers how a row maps to keys: users row 42 becomes user:42:profile and user:42:card.

Distributed cache · architecture

Stage 4: Invalidation from the change log

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 4 of 6

Invalidation from the change log

Why this stage

Writers delete keys after committing, but a delete can fail, race with a fill, or be lost in a failover.

Requirement: no stale value outlives a write by more than a few seconds.

What we add

Leases on every cache node, so a fill that started before a delete is refused.

An invalidator that tails the database change log and deletes every affected key.

What it costs

A Kafka topic and a worker fleet to run, and a key naming convention every team must follow so the invalidator can derive keys from rows.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

One race still remains, and leases close it. The stress tests explain both.

Follow a miss, then a write with its two deletes:

Distributed cache · architecture

Trace a request: Get, cache miss with a lease

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasionalHop in motionHops doneResponse

Get, cache miss with a lease: step 1 of 4

App servers to Cache primaries

The client library hashes user:42:profile onto its ring copy in about a microsecond and sends get on a pooled connection to the owning primary, in another AZ two times in three. It misses, and because nobody else is filling the key, the node returns a lease token with the miss.

0.5 ms of a 10 ms budget

App servers → Cache primaries0.5 ms
App servers → Database4 ms
App servers → Cache primariesasync
Response

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

This is enough for a mid-level answer. Cache-aside, a ring in the client, a replica per shard, and deletes from the change log. The stress tests are what make it a senior answer.

Go deeperThe API and the engine2 min

get(key)                         -> value | MISS
get_multi([key, ...])            -> {key: value}       up to 100 keys, one request per shard
lease_get(key)                   -> value | MISS(token) | WAIT
set(key, value, ttl, token?)     -> STORED | NOT_STORED
delete(key)                      -> OK

MISS(token) means “load it and set it with this token”, and WAIT means “try again in about 20 ms”. Keys look like pool:type:id:vN: the pool prefix selects the team’s memory budget and credentials. The protocol is a compact TCP one, not HTTP: a get is about 50 bytes on the wire, and parsing costs a few hundred nanoseconds.

Engine	Fits because	Costs
Memcached	Many threads, a compact memory layout, lease-friendly protocol	No replication: we build it around it
Redis	Built-in replication, rich types, Redis Cluster	One thread per shard runs commands; more bytes per key
A library engine (CacheLib)	RAM and flash together, modern eviction	You build the server around it

With rich data types or a small team, Redis Cluster, with its built-in replicas and failover, is the better buy. Everything else on this page works on top of it too.

Go deeperOther ways to route and replicate1 min

Routing scheme	Adding a 65th shard moves	Problem
`hash(key) mod N`	About 98% of keys	Every resize is a near-total cold start
Hash slots and a slot map (Redis)	The slots you choose to move	A map every client must fetch; moves are explicit work
Consistent hashing, virtual nodes	About 1/65	Balance only as good as the number of points

Slots are equally sound, and suit a store that moves data. A cache needn’t move data, because a moved key can simply miss.

Replication	Memory	What a failure loses
None; the database absorbs misses	1×	A zone outage is a database outage
Async replica in another zone	2×	The last few milliseconds of sets and deletes
Synchronous replica	2×	Nothing, but every delete waits on another zone
A full copy in every zone	3×	Nothing; reads stay in their own zone

“When a node dies, consistent hashing moves its keys to the next node” is true, and usually wrong for a cache. The neighbors receive the dead node’s keys cold, and inherit its hot keys, which can push them over next. A node that flaps back owns its old range again, and serves values that missed every delete while it was gone. So ownership changes only when an operator resizes the ring. A failure promotes a replica.

Stress it

The design is complete. Now we test it where caches are hard. A race that caches an old value. One key everyone wants. The deletes a failover loses. A full memory. And ten times the traffic. Only the last one adds boxes.

The stale set race, and leases

Interviewerminute 20

Your writer commits and then deletes. Can a reader still end up caching the old value?

What they are testing

Whether you can draw the race with two actors and close it with a mechanism, not a shorter TTL.

Yes. The reader missed, read the row, and stalled (a garbage-collection pause) before its set. In the gap, the writer committed and deleted. So the set stores the old value for an hour. The invalidator’s second delete helps only if the stale set lands first.

Without leases: a fill from before the write lands after it

Scroll or pinch to zoom, drag to pan

Color key
Service
Database
Cache

Leases close it, a mechanism from Facebook’s memcache paper. A miss hands the first caller a token. A delete voids any token for that key. A set with a voided token gets NOT_STORED, and the next reader refills from the new row. While a lease is out, other callers that miss get WAIT, which the next deep dive uses against stampedes.

LeaseCache.javalease_cache.pyLeaseCache.cslease_cache.golease-cache.ts

// One cache node's get, set and delete with leases, the scheme from Facebook's memcache paper.
// A lease is the right to fill one key after a miss. It does two jobs: only one caller per key
// goes to the database (no stampede), and a delete voids it, so a value read before a write
// can never be stored after that write's delete (no stale set).
class LeaseCache {
    sealed interface Result permits Hit, Fill, Wait {}
    record Hit(String value) implements Result {}
    record Fill(long token) implements Result {} // you missed first: load it and set with this
    record Wait() implements Result {}           // someone else is loading it: retry in ~20 ms

    private record Lease(long token, long expiresAt) {}

    // A filler that crashes must not block the key forever, so a lease expires. 10 s is far
    // longer than any healthy database read, and still short for an outage.
    static final long LEASE_MS = 10_000;

    // Eviction and TTLs are left out to keep the leases visible: see the LRU cache page.
    private final Map<String, String> values = new HashMap<>();
    private final Map<String, Lease> leases = new HashMap<>();
    private final LongSupplier clock;  // injected so tests control time
    private long nextToken = 1;        // unique per node, so an old token never matches a new lease

    LeaseCache(LongSupplier clock) {
        this.clock = clock;
    }

    // One lock keeps the idea visible. A real node locks per hash bucket, so unrelated keys
    // never wait on each other.
    synchronized Result get(String key) {
        String v = values.get(key);
        if (v != null) return new Hit(v);
        long now = clock.getAsLong();
        Lease l = leases.get(key);
        if (l != null && l.expiresAt() > now) return new Wait(); // the herd waits here, not at the db
        long token = nextToken++;
        leases.put(key, new Lease(token, now + LEASE_MS));
        return new Fill(token);
    }

    // The set succeeds only if the lease is still the one handed out. A delete since the miss
    // removed it, so a value loaded before that write is refused instead of cached for an hour.
    synchronized boolean set(String key, String value, long token) {
        Lease l = leases.get(key);
        if (l == null || l.token() != token) return false;
        leases.remove(key);
        values.put(key, value);
        return true;
    }

    // Writers call this after committing to the database, never before: deleting first lets a
    // reader refill the old value before the commit lands.
    synchronized void delete(String key) {
        values.remove(key);
        leases.remove(key); // voids any fill in flight; the next miss gets a fresh lease
    }
}

What leases don’t fix: lagging database replicas. Say a filler reads from a database replica that hasn’t yet applied the write. It gets a fresh lease after the delete, and still loads the old row. So fill from the primary for keys deleted in the last few seconds.

Go deeperOther fixes for the race1 min

Fix	How it works	Closes the race?
Short TTLs	Bound how long the stale value lives	No, bounds it, at the cost of hit ratio
Delayed double delete	Writer deletes again 500 ms later	No: a reader stalled longer than 500 ms still wins
Versioned set	Set only if newer than the cached version	Only if a delete leaves a marker holding the version
Leases	A delete voids the token; a set needs a valid one	Yes, for every fill that started before the delete

Why not write-through, where the cache writes to the database? It puts the cache on the write path, so its availability limits writes. It must handle every write durably and in order, which is a database’s job. And it also caches values nobody reads.

Hot keys and stampedes

A celebrity posts. One key gets a million reads a second, and its expiry is about to run out. What happens?

A million reads a second is three times what one node serves, and half its network card. More shards do nothing, because the key hashes to one of them. And when its expiry runs out, every request in that millisecond misses and queries the database: a stampede, or thundering herd.

Each node samples 1 in 100 requests and reports keys above 50,000 reads a second, so hot keys are found automatically. Then four techniques work together:

A near cache in every app process. The hottest few thousand keys, held for 2 seconds. Each process reads a hot key at most once every 2 seconds.
Copies of detected hot keys. Store it as key#1 to key#8 on 8 shards, and let readers pick one at random. 20,000 hosts of 16 processes can still mean 160,000 reads a second on one shard, so this cuts it to 20,000 per copy. Writes and deletes must hit all 8.
Leases as single flight. When the key expires, one caller fills it, and the rest get WAIT. Facebook measured leases cutting peak database queries from 17,000 a second to 1,300.
Early refresh for values that are slow to compute.

A hot key expires: leases turn a herd into one database read

Scroll or pinch to zoom, drag to pan

Color key
Database
Cache

Leases still make waiters pay a retry. For a value that takes 100 ms or more to compute, probabilistic early refresh avoids even that. Each reader adds a random head start, scaled by the compute time. It refreshes if that crosses the expiry. Take a 100 ms compute, and 50,000 reads a second. The first early refresh comes about 0.85 seconds before expiry. So the key rarely expires under load.

EarlyRefresh.javaentry.pyEarlyRefresh.csentry.goentry.ts

// Probabilistic early expiration, "XFetch" (Vattani, Chierichetti and Lowenstein, VLDB 2015).
// Every reader of a cached entry rolls a die: refresh it now, a little before it expires? The
// chance rises as expiry nears and with how long the value takes to recompute, so under load
// one reader refreshes early and the hot entry rarely gets to expire.
class EarlyRefresh {
    // Stored with the value: when it expires, and how long the last recompute took.
    record Entry(String value, long expiresAtMs, long computeMs) {}

    // 1 is the paper's default. Above 1 refreshes earlier, below 1 later.
    static final double BETA = 1.0;

    // u is uniform in (0, 1]. -ln(u) is exponential with mean 1: usually small, now and then
    // large. So most reads see no head start, and a rare one sees enough to refresh early.
    // A slow recompute gets a proportionally bigger head start.
    static boolean shouldRefresh(Entry e, long nowMs, double u) {
        double headStartMs = e.computeMs() * BETA * -Math.log(u);
        return nowMs + headStartMs >= e.expiresAtMs();
    }

    // No coordination: each reader decides alone, so it also works across thousands of app
    // servers. A reader that refreshes still returns the cached value it already has.
    static double roll() {
        return 1.0 - ThreadLocalRandom.current().nextDouble(); // (0, 1], so log never sees 0
    }
}

What a failover loses

A cache primary dies and its replica takes over. What is missing, and does it matter?

The replica lacks the last few milliseconds of copying. Lost sets just refill. A lost delete is the danger: the new primary would serve the old value for its full expiry, an hour.

So after promotion, the invalidator replays 30 seconds of the change log for that shard. That covers the copying lag and the lease window. Deletes are safe to repeat, so replaying is harmless. Teams whose keys can’t be derived from rows get no replay, so their slice of the shard is flushed on promotion instead. And leases aren’t copied, so a fill holding an old primary’s token gets NOT_STORED, the safe direction.

Failing over a primary, and replaying what replication lost

Scroll or pinch to zoom, drag to pan

Color key
Service
Database

Eviction

Memory is full. Which item goes, and what does the policy cost for each item?

The classic LRU cache (evict the least recently used) is a hash map plus a doubly linked list. At this scale two of its costs matter. The list costs 16 bytes of pointers per item, 80 GB across 5 billion items, more than a whole node. And every hit moves an item to the head of a shared list, a write under a lock on every read.

We pick segmented LRU, with memory per team. A new item enters a probation segment, so a scan of one-off keys evicts other one-off keys, not the working set. A second hit promotes it, at most once a minute, so most hits take no lock. One team’s churn evicts only its own pool. Expired items are removed by a background crawler, so dead data doesn’t wait for eviction.

Go deeperEviction policies, and what an item costs1 min

Policy	How it picks a victim	Per-item cost	Weakness
Strict LRU	Tail of one list; every hit moves to the head	16 bytes, a lock per hit	One scan of cold keys flushes the hot set
Sampled LRU (Redis)	Sample 5 random keys, evict the oldest of them	A 24-bit access clock	Approximate; good enough at 10 samples
Segmented LRU (memcached)	Hot, warm and cold queues; a second hit moves an item up	16 bytes, rare locks	More settings to tune
LFU and W-TinyLFU	Admit and evict by estimated frequency	A few bits in a sketch	Plain LFU adapts slowly

Each item carries a 50-byte header: a hash chain pointer, LRU links, expiry and last access, a version, and sizes. Small values are the memory trap. The header is 8% of a 500-byte item, but half of a 20-byte counter with a 30-byte key. Teams with tiny values pack many fields into one item.

Adding nodes at ten times the traffic

Traffic grows ten times, and the database grows only three times. What breaks first, and how do we add nodes without hurting the database?

At ten times, 5% misses is 3 million database reads a second. More nodes fix the CPU, but only a higher hit ratio protects the database. A database grown three times, to 1.5 million reads a second, can take 1.2 million misses, which needs a 98% hit ratio.

That comes from memory. Today’s misses are mostly rarely read items evicted before their next read. Tripling memory to 200 primaries keeps them resident, and buys about three points. A sampled “shadow” cache, which tracks keys without values, measures this for each team before anyone orders memory. The near cache from the hot keys stress test takes about 40% of reads off the primaries.

Distributed cache · architecture

Stage 5: Grow the ring, add a near cache

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 5 of 6

Grow the ring, add a near cache

Why this stage

At 60 million reads/s (10×) the 64 primaries are over 300% busy, and 5% misses are 3 million database reads a second.

What we add

The ring grows to 200 primaries, three times the memory, so the hit ratio climbs from 95% to 98% and the database (grown to 1.5 million reads/s) sees 1.2 million misses.

An in-process L1 on every app server answers the hottest keys for 2 seconds, which relieves the primaries.

What it costs

A rebalance that moves keys while they serve traffic, and up to 2 seconds of extra staleness on the hottest keys.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

Adding nodes isn’t free. Adding 8 shards to 64 moves 8/72 of keys, 11%. Each misses on its new owner, and at peak that’s about 630,000 extra database reads a second, more than its whole capacity. So new nodes warm up through old ones:

Adding shards with a warm-up period

1. Start the nodes empty

The new shards join under a new epoch as “warming”. Clients hold the old ring and the new one.

2. Read new, fall back to old

A get goes to the new owner. On a miss, it asks the old owner, and on a hit copies the value over. Only a miss on both reaches the database.

3. Deletes go to both

A delete goes to the old and the new owner, so the fallback never brings back a dropped value.

4. Finish

Once the new owners’ hit ratio matches the fleet, drop the old ring.

Hot keys are read first, so most warming happens in the first few minutes. Eight shards at a time, 17 steps take the fleet from 64 to 200 over two weeks of quiet hours.

Go deeperA routing sidecar, at 20,000 hosts1 min

At ten times, 320,000 app processes each hold a connection to each of 400 cache nodes: 320,000 connections per node. A ring change has to reach every process, and leases, failover and epochs must behave the same in five client libraries.

Option	Extra hop	Connections per cache node	Problem
Smart client library	None	320,000	Five implementations; every process gets ring changes
Sidecar router per host	Local, about 50 µs	20,000	One more process on every host
Central proxy tier	A network hop, 0.2 to 0.5 ms	One per proxy	Doubles the network latency budget

So each app host runs a router sidecar, the way Facebook runs mcrouter. Apps talk to it on the same machine. It holds the ring, leases, failover and one connection pool for the host, and only it watches etcd. A crashed sidecar turns that host’s lookups into misses, so the library keeps a small direct fallback.

Distributed cache · architecture

Stage 6: A routing sidecar

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 6 of 6

A routing sidecar

Why this stage

At 10× there are 320,000 client processes, each with connections to 400 cache nodes, and five client libraries that must implement the ring, leases and failover identically.

What we add

A router sidecar on every app host: one connection pool per host, the ring and failover logic in one place, and ring updates pushed to 20,000 sidecars instead of 320,000 processes.

What it costs

A loopback hop of about 50 µs, one more process on every host, and a sidecar crash that turns every lookup on that host into a miss.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

Distributed cache · architecture

Trace a request: Get through the sidecar

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasionalHop in motionHops doneResponse

Get through the sidecar: step 1 of 4

App servers to Router sidecar

The in-process L1 misses (it only holds the hottest few thousand keys, for 2 seconds). The request goes over loopback to the router sidecar on the same host, about 50 µs plus queuing.

0.25 ms of a 2 ms budget

App servers → Router sidecar0.25 ms
Router sidecar → Cache primaries0.5 ms
Router sidecar → Cluster managerasync
Response

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The multi-get trap. A 100-key multi-get touches more shards as the ring grows, each with fewer keys, and per-request overhead dominates. Put keys read together on one shard, by a shared prefix such as the user ID.

When things break

Interviewerminute 40

A cache node dies while its replica is down for maintenance. What does the database see?

What they are testing

Whether every key of the dead shard misses at once, and you can show the database survives it.

The rule for every failure: the cache may get emptier. But it must never send the database more misses than it can take, or serve a stale value past the freshness promise.

A node dies with no replica. Every key on that shard misses at once, about 90,000 extra reads a second. That’s inside what the database can take. What protects it is that each key is filled once, not once per caller. Leases make one caller fill each key, while the others wait a few milliseconds. That’s called request coalescing. Clients never retry a get, so a slow database doesn’t get a retry storm on top. The shard warms back up within minutes, starting with its hottest keys.

A whole zone goes down. That takes a third of the primaries. Clients mark each one down after five errors and read its replica within a second, so the database sees almost no extra reads. Fills aren’t stored for about 10 seconds while the leases run out, then the replicas are promoted, and each replays 30 seconds of deletes. Users notice nothing.

Click any box to break it and see what users would notice:

Distributed cache · architecture

Break it: what fails, who notices, what users see

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Break

Failure: click any box to break it

A cache primary goes down

What breaks

Gets and sets for 1/64 of the keyspace time out until clients mark the primary down.

How we notice

The cluster manager misses three 1 second heartbeats from two vantage points; clients see connection errors.

What users see

Almost nothing: after five errors in a row each client reads that shard from the replica, milliseconds behind.

Fills are not cached until promotion, about 10 seconds later.

What we do

Once the old primary’s 10 second lease has expired, promote the replica, publish ring epoch N+1, and replay the last 30 seconds of the change log for that shard, so no delete lost to replication lag leaves a stale value.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

Misses are rationed under stress. Clients never retry a get, the sidecar caps how many database fills each host runs at once, and leases keep one fill per key. Restarts are failures we schedule: one shard at a time, by promoting the replica first.

Go deeperEvery failure, and what users see1 min

Part	Failure	What users see	What we do
Cache primary	Crash	Fills not stored for ~10 s on 1/64 of keys	Reads to the replica, promote after the lease, replay 30 s
A whole zone	Outage	A third of shards fail over at once	Same per shard; the database sees almost no extra reads
Cache replica	Crash	Nothing	A fresh replica copies everything in about 2 minutes
Database	Primary down	Hits keep working; misses and writes fail	Database failover; never stretch expiries to cover it
Change log, invalidator	Lag or down	Rare stale reads, only if a writer’s delete also failed	Resume from the last position; deletes are safe to repeat
Cluster manager	Down	Nothing until another node fails	Clients keep the last ring; a standby manager
Router sidecar	Crash or bad deploy	That host’s lookups all miss	Restart; roll out to 1% first; direct fallback

Operation	Guarantee	What users can see
Read after own write	The writer’s delete makes their next read refill	Their own change, at once
Other users after a write	About 1 s at p99, a few seconds at worst	The old value until a delete lands
Fill racing a write	No stale set, thanks to leases	Nothing, unless the fill read a lagging database replica
Worst case, everything fails	Bounded by the expiry, at most 24 hours	A stale value until expiry, which is why every key has one

“Cache-aside is eventually consistent” isn’t true without help. After the stale set race, plain cache-aside keeps the old value until its expiry, even though the database never changes again. Leases and the change log are what make it catch up in seconds.

Go deeperRunning it: alerts and cost1 min

Wake someone up when:

A team’s hit ratio falls 3 points below normal for 5 minutes. (It moves before the database does.)
Database reads from misses pass 80% of capacity.
A node’s get p99 passes 2 ms. (That’s usually a hot key.)
Invalidation lag passes 10 seconds.
Any shard has had no healthy replica for 10 minutes.

Rising evictions of unexpired items mean you should add memory, before the hit ratio falls.

Abuse between teams. Memory and rate limits per team, and a 1 MB value limit. Also alerts on a spike in the number of distinct keys. That usually means a team put a request ID in a key.

Cost is memory: 128 nodes today, 400 at ten times. Second is traffic between zones: two thirds of reads cross a zone, roughly $1,400 a day. A full copy in every zone trades that for 50% more memory.

Wrap up

The whole design

Here is everything we built, in the order we built it. Use the buttons to step through it.

Distributed cache · architecture

The full design, stage by stage

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Cache reads per second: 6,000,000

Stage 6 of 6

A routing sidecar

Why this stage

At 10× there are 320,000 client processes, each with connections to 400 cache nodes, and five client libraries that must implement the ring, leases and failover identically.

What we add

A router sidecar on every app host: one connection pool per host, the ring and failover logic in one place, and ring updates pushed to 20,000 sidecars instead of 320,000 processes.

What it costs

A loopback hop of about 50 µs, one more process on every host, and a sidecar crash that turns every lookup on that host into a miss.

At 6M cache reads per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.The whole design. Step through it: one cache node, a hash ring, a replica per shard, invalidation from the change log, more memory and a near cache, then a routing sidecar.

The numbers behind the busy bars

What one machine can do

One cache node handles about 300,000 operations a second and has 48 GB of usable memory.

How much the cache catches

One node holds the hottest 1.6% of items, which gives a 60% hit ratio.
64 shards hold everything live, which gives 95%.
At stage 5, the L1 cache answers 40% of reads, all of which were hits already. Three times the memory lifts the hit ratio to 98%, so only 2% of reads reach the database.

Writes and the database

Writes (fills and deletes) are 10% of reads.
The database serves 500,000 reads a second, and grows to 1.5 million for the 10× mark.

The decisions, and what each one costs

Decision	We chose	We gave up	We’d choose differently if
Engine	Memcached-style, replication built around it	Rich types, built-in replication	We needed data structures, or had a small team
Routing	Consistent hashing in the client, later a sidecar	An explicit slot map	We moved data on resize (hash slots)
Availability	Async replica per shard in another zone	Half the memory; the last ms on failover	Losing a zone’s cache were survivable
Freshness	Delete on write, change-log invalidator, leases	A pipeline and a key naming contract	Minutes of staleness were fine (expiry only)
Hot keys	Near cache, leases, copies, early refresh	Up to 2 s of extra staleness on hot keys	Values must never be stale (copies only)
Eviction	Segmented LRU per team	16 bytes of pointers per item	Values were tiny
Growth	Warm new shards through old owners	Two rings in clients during a resize	The database could absorb 11% of keys going cold

How to spend 45 minutes

0 to 4. What’s behind the cache, how stale a read may be, size and traffic.
4 to 8. 3 TB and 64 primaries, 300,000 misses against 500,000, and why one cold zone forces replicas.
8 to 20. Build it: cache-aside, the ring, replicas, the change log. A mid-level answer can stop here.
20 to 30. The stress tests: the stale set race and leases, then hot keys.
30 to 40. Failover’s lost deletes, eviction, then ten times and adding nodes.
40 to 45. A node dying with no replica, then a zone, then your own questions.

Never skip the database’s headroom, the stale set race, or hot keys.

Follow-up questions

Interviewer asksWhy not just use Redis Cluster?

For one team, or for rich data types, you should. A shared platform adds leases, a change-log invalidator and per-team pools, and all of that works on top of Redis too.

Interviewer asksEach region has its own cache. A write in Europe, a read in the US. How is the US cache invalidated?

The database copies the write to the US. The US invalidator reads the US replica’s change log. So the delete happens after the row has arrived. Deleting from the European writer directly would be wrong. A US reader could miss, and refill from a US replica that hasn’t received the write yet.

Go deeperMore follow-up questions1 min

Lookups for IDs that don’t exist are hammering the database. What do you do? Cache the absence. Store a small “not found” marker for 30 to 60 seconds. Then a scraper looping over random IDs costs one database read per ID per minute. The insert that later creates the row deletes the marker, like any other write.

Values average 50 KB now. What changes? Network becomes the limit before memory or CPU: 300,000 reads a second of 50 KB is 15 GB/s, far past one network card. Size shards by bandwidth, compress values, and consider an engine that keeps hot items in RAM and the long tail on flash.

A team asks you to flush their whole pool after a bad deploy. Not all at once. That sends the team’s whole traffic to the database. Flush in slices, a few percent of the ring at a time. Each slice refills behind leases before the next goes, while you watch the database’s miss load.

Design a distributed cache

The answer in one minute

Scope it

The numbers that decide it

Build the design

Step 1. One cache node, cache-aside

Stage 1: One cache node, cache-aside

Why this stage

What we add

What it costs

Step 2. Shard with a hash ring

Stage 2: Shard with a hash ring

Why this stage

What we add

What it costs

Step 3. A replica per shard

Stage 3: A replica per shard

Why this stage

What we add

What it costs

Step 4. Invalidation from the change log

Stage 4: Invalidation from the change log

Why this stage

What we add

What it costs

Trace a request: Get, cache miss with a lease

Stress it

The stale set race, and leases

Hot keys and stampedes

What a failover loses

Eviction

Adding nodes at ten times the traffic

Stage 5: Grow the ring, add a near cache

Why this stage

What we add

What it costs

Stage 6: A routing sidecar

Why this stage

What we add

What it costs

Trace a request: Get through the sidecar

When things break

Break it: what fails, who notices, what users see

What breaks

How we notice

What users see

What we do

Wrap up

The whole design

The full design, stage by stage

Why this stage

What we add

What it costs

The decisions, and what each one costs

How to spend 45 minutes

Follow-up questions

Further reading

Keep reading

The designbeyondcode digest, every Saturday