Design a Distributed Cache

Every team in your company puts a cache in front of its database. Now you are asked to build that cache: a cluster of machines that holds two billion small values in memory, answers in well under a millisecond, and keeps working while nodes die. It is the kind of product people mean when they say Redis or Memcached.

Interviewers like it because it spans data structures and distributed systems: the hash table inside one node, how keys spread over many nodes, and an honest account of what a failover throws away. How to use a cache is in Caching. Unlike Design a Key-Value Store, losing data is allowed here, so the design trades durability for speed. The steps follow A Framework for Any Design.

#Clarify the problem

• Functional: get(key); set(key, value, ttl); delete(key); expired keys are never returned; when memory is full, some keys are evicted.
• Non-functional: p99 get under 1 ms within a data center; keeps serving when any one node dies; grows by adding nodes without downtime; a failover may lose the last few milliseconds of acknowledged writes, and that window is bounded.
• Out of scope: lists, sets and counters; multi-key transactions; durable storage; replication across regions.

#Estimate the scale

• Bytes per item: a 50-byte key, a 500-byte value, and about 50 bytes of bookkeeping (hash chain pointer, eviction pointers, expiry time, sizes). Call it 600 bytes.
• Usable memory per node: machines with 64 GB, of which about 50 GB holds items. The rest is headroom.
• Throughput budget: assume one node handles about 100,000 simple operations a second. Leave room below it.

Bandwidth is 1,000,000 × 600 B = 600 MB/s, or 25 MB/s per node: easy. Writes are 10% of peak, about 4,200 a second per primary; that number returns when we count what a failover loses.

#Define the API

GET key
  VALUE bytes            hit
  MISS lease=501         miss; you may fill it with lease 501
  WAIT                   someone else holds the lease; retry in about 20 ms
 
SET key value [TTL seconds] [NX | XX] [LEASE token]
  OK
  NOT_STORED             NX and the key exists, XX and it does not,
                         or the lease was cancelled by a delete
  TOO_LARGE              value over 1 MB
  OUT_OF_MEMORY          full, and the policy is not to evict
 
DEL key
  DELETED | NOT_FOUND    also cancels any open lease on the key
 
any command
  MOVED 9301 10.0.4.17:7000   slot 9301 lives on another node

The decision worth arguing is the transport. A get takes microseconds on the server and about 0.5 ms on the network, so HTTP headers and parsing would cost more than the work. Use persistent TCP connections and a compact protocol, and let clients pipeline: send 10 gets without waiting, and pay one round trip instead of ten.

#Model the data

entry                                        about 50 B of bookkeeping
  key          bytes   (up to 250 B)
  value        bytes   (up to 1 MB)
  expires_at   8 B     ms since epoch, 0 = never
  hash_next    8 B     next entry in the same bucket
  lru_prev     8 B     neighbours in the recency list
  lru_next     8 B
  sizes, flags ~10 B
 
cluster map                                  a few hundred KB
  slot 0..16383   -> shard id
  shard id        -> primary address, replica addresses
  epoch           -> bumped on every change

Every access is by exact key: no range scans, no indexes, no joins. So there is no database to choose: the store is a hash table in RAM, with a doubly linked list threaded through the same entries to track recency. That pair is the LRU cache from Designing Data Structures, O(1) for every operation. The cluster map is tiny but everyone must agree on it, so it lives in a small consensus-backed config service.

#High-level design

There is no load balancer. The client library is the router: it keeps the cluster map in memory and talks straight to the owning primary.

The real cluster has 24 shards; three show the shape.

#Deep dives

Interviewers push on four things: one node's internals, routing, what replication promises, and hot keys.

#Inside one node: memory, expiry and eviction

The hash table. One node holds 2 billion ÷ 24 ≈ 83 million entries, usually served by one thread on an event loop, with no locks. The trap is growing the table. Moving 67 million entries from 2²⁶ buckets to 2²⁷ in one go, at say 50 ns each, blocks the node for over 3 seconds. Rehash incrementally instead: keep both tables, move a few buckets on every operation, look in both until the old one is empty.

Expiry. A timer per key is too costly. Combine two cheap mechanisms:

• Lazy: every read checks expires_at and deletes the entry if it has passed. A reader can never see an expired value.
• Sampling: in one common scheme, about 10 times a second, pick 20 random keys that have a TTL and delete the expired ones. If more than a quarter of the sample was expired, go again. Unread keys still get cleaned up, without a full scan.

The primary decides expiry and sends a delete to the replica, so both copies agree.

Eviction. Exact LRU, as in the data model, rewrites list pointers on every read and costs 16 bytes per entry. Many caches approximate it instead: keep a small "last used" clock in each entry, sample 5 to 10 keys when memory is full, and evict the oldest of the sample. That is close to true LRU. The policy is set per cluster: LRU over all keys (the usual default), LRU over keys with a TTL (keys without one are never evicted), LFU (keeps steady favourites when one-off scans would flush LRU), or no eviction (writes fail with OUT_OF_MEMORY).

#Partitioning and routing

Plain hash(key) mod 24 remaps almost every key when a 25th node arrives. Use consistent hashing in its fixed-slot form: hash each key into one of 16,384 slots, and assign slots to shards. Each of 24 shards owns about 683 slots. Adding a 25th moves about 655 slots, 1 ÷ 25 = 4% of the keys, one slot at a time. While a slot moves, the old node answers keys it still has and sends the rest to the new node for that one request; when the move ends, the map changes and the old node replies MOVED.

Pick the smart client with redirects as the safety net. The cost: clients learn of map changes on their next error, so some requests briefly take two hops.

#Replication and what failover loses

Each shard has a primary and one replica on a different rack. The primary acknowledges a write as soon as it is in its own memory, then streams it to the replica (see Replication). So be precise about the guarantee: an acknowledged write can be lost. With 10 ms of lag and 4,200 writes a second, a crash loses about 42 values. For a cache that is fine; those keys become misses and are refilled from the database.

The loss that hurts is a lost delete. An app updates the database and deletes the cached key; the primary acknowledges and dies before the replica hears of it. The new primary serves the old value until its TTL ends. So give every key a TTL, even a long one: it caps how long such a ghost lives.

Failure detection must not be one machine's opinion: the config service runs on 3 or 5 nodes with consensus; a primary is declared dead only when a majority has failed to reach it for the timeout, 5 seconds here. One more race: a primary cut off from the config service but still reachable by some clients keeps taking writes the new primary never sees. Close it with a leadership lease (not the fill lease from the API): a primary accepts writes only while its lease from the config service is valid. It renews every second, each renewal lasts 4 seconds, and the config service promotes the replica only after 5 seconds. So two primaries never take writes at once. The 1-second margin covers clocks that run at slightly different speeds. Writes it took after the cut are lost.

Synchronous replication would add a round trip to every write to protect data the database already has, so keep it asynchronous.

#Hot keys and the thundering herd

One key, a celebrity profile, gets 150,000 reads a second. One shard handles about 100,000, and adding nodes does not split a key. Two fixes:

• Near cache in the client: each app server keeps that key in its own memory for 1 second. With 300 app servers, the shard sees at most 300 reads a second for it, at the cost of up to 1 second of staleness.
• Copies under several keys: write the value as profile:9#0 to profile:9#7, which hash to different slots, and read a random one. 150,000 ÷ 8 ≈ 18,750 reads per copy. The cost: a write or delete must touch all 8.

Find hot keys by sampling on each node and reporting the top few every second, as in Design Top-K Trending.

The thundering herd is the other side: a hot key expires or is deleted, and hundreds of app servers miss at once and query the database together. The cache server can help through the lease. The first miss gets a lease; later callers get WAIT until the value is set. Leases also fix a race: app 1 reads an old row, app 2 updates the row and deletes the key, then app 1 writes the old value back. The delete cancelled lease 501, so app 1's SET gets NOT_STORED. A lease needs its own short timeout, say 2 seconds, so a holder that crashes does not block the key.

#Bottlenecks and failures

At 10x (12 TB, 10 million operations a second) you need about 240 primaries. Connections strain first: 3,000 app servers (10x today's 300) each connected to 240 primaries is 720,000 connections, 3,000 per node, so add a proxy tier or pooling. Slot moves become routine; automate them and cap their bandwidth.

#What a strong answer sounds like

#Follow-up questions

#How do you get several keys in one call?

A multi-get works only within one node, so the client splits it by shard and sends the pieces in parallel. If a group of keys must always live together, give them a shared tag, such as {user:42}:cart and {user:42}:prefs, and hash only the tag.

#Should a node write to disk?

Only to restart warm. A periodic snapshot lets a rebooted node load 50 GB from disk instead of starting empty. It is not durability: writes since the snapshot are gone.

#What about one huge value?

On a single-threaded node, every request waits behind a 1 MB send. Cap the size, split big objects into chunks, and alert on keys above about 100 KB.

#How would you serve two regions?

One independent cluster per region, each in front of its local database replica. When data changes, publish an invalidation event that every region applies.

#Why not read from replicas?

It would double read capacity, but a replica lags by about 10 ms, so a client can miss its own fresh write or read a value it just deleted. Here memory sets the node count and primaries run under half their budget, so reads stay on primaries. Allow replica reads only for hot keys that tolerate a little staleness.