Design a Distributed Cache

Problem Statement#

"Design a distributed cache like Redis or Memcached. Clients store and read key-value pairs, entries can expire, and when a node runs out of memory it evicts whatever was used least recently."

This one is unusual. Half of it is a data structures question you could answer on a whiteboard in ten minutes, and half of it is a genuinely hard distributed systems question. Interviewers ask it very differently depending on what they want to see.

Requirements Gathering#

Ask enough to find out which half of this problem the interviewer cares about. The answers about scale, in particular, change everything after the first ten minutes.

Candidate: "Is this a cache service that other teams call over the network, or an in-process library?"

Interviewer: "A service. Clients connect to it over the network."

Candidate: "Just plain key-value, or do we need richer types like Redis lists and sorted sets?"

Interviewer: "Plain key-value is fine. get, set, delete."

Candidate: "Can entries expire on their own?"

Interviewer: "Yes, an optional TTL per key."

Candidate: "And when a node fills up?"

Interviewer: "Evict the least recently used entries. LRU."

Candidate: "What scale are we sizing for? Total data and peak request rate."

Interviewer: "Up to 1 TB of data and 100,000 requests per second at peak."

Candidate: "What's the latency target? A cache that is not fast is not worth having."

Interviewer: "Single digit milliseconds. Call it p99 under 10ms for both get and set."

Candidate: "If a replica is a few hundred milliseconds behind and a client reads a stale value, is that a problem?"

Interviewer: "No. Availability over consistency. A stale or missing value just means the caller goes to the origin."

Candidate: "Does the data need to survive a restart?"

Interviewer: "No. Treat it as pure cache. Durability is out of scope."

Candidate: "How big are values, typically?"

Interviewer: "Small. Mostly a few hundred bytes, occasionally a few KB."

Functional and Non-Functional Requirements#

Functional Requirements#

1. set(key, value) with an optional TTL
2. get(key) returns the value, or nothing if it is missing or expired
3. delete(key) removes an entry
4. An entry past its TTL is never returned, and eventually stops using memory
5. When a node reaches its memory limit, it evicts the least recently used entries

Non-Functional Requirements#

Out of scope: durability across restarts, transactions, strong consistency, range queries or secondary indexes, rich data types.

Capacity Estimates#

There are two independent sizing questions here. Work out both and take the larger answer.

Throughput:
  one node serving in-memory ops over the network, benchmark ~20k req/sec
  100,000 / 20,000 = 5 nodes
  add headroom for spikes and node loss -> ~8 nodes

Memory:
  a 32 GB instance gives ~24 GB usable after the OS, the process and fragmentation
  1 TB = 1,024 GB
  1,024 / 24 = ~43 nodes
  add headroom for growth -> ~50 nodes

Memory wins, so provision for memory: ~50 nodes.
Throughput then comes free: 50 x 20k = 1M req/sec of capacity
against a 100k requirement, about 10x headroom.

Replication doubles the bill:
  50 shards with one replica each = ~100 nodes of RAM to hold 1 TB of data.

One number people miss, and it is worth saying out loud because it is where real deployments get caught:

Per-entry overhead:
  a 200-byte value does not cost 200 bytes
  key bytes + value bytes + hash table slot + list node (two pointers)
  + expiry timestamp + allocator rounding
  = roughly 60 to 100 bytes of bookkeeping per entry

  1 TB of payload at 200 bytes per value = ~5 billion entries
  5 billion x 80 bytes of overhead = ~400 GB of pure bookkeeping

Small values are expensive. Size the cluster on entry count, not just on payload bytes, and say so. An interviewer who has actually run a cache in production will notice.

High-Level Design#

Build the single node first. The distributed part only makes sense once you know what is being distributed.

Version 1: a hash map#

A cache is a hash map. Every language ships one, and it gives O(1) lookup and insert.

get(key)    -> table[key]
set(key, v) -> table[key] = v
delete(key) -> del table[key]

That satisfies requirement one, and nothing else. It grows without bound and never forgets anything.

Version 2: add expiry#

Store a deadline next to each value instead of a bare value. On read, check the deadline before returning.

set(key, v, ttl) -> table[key] = (v, now + ttl)
get(key)         -> (v, deadline) = table[key]
                    if deadline and now >= deadline: delete and return nothing
                    else return v

This is correct but leaky. An expired key that nobody reads again sits in memory forever, because the only thing that removes it is a read that never comes. Deep dive 1 fixes that.

Version 3: add LRU eviction#

Now the interesting bit. Eviction needs two things at once: find any entry instantly, and know which entry was used longest ago. A hash map gives the first and not the second. A list gives the second and not the first. So use both, pointing at the same objects.

The hash map maps a key to its list node. The list is ordered by recency: whatever is just behind the head was touched most recently, whatever sits just before the tail is the eviction candidate.

• On get: look the key up in the map, then unlink its node and splice it in behind the head. Both O(1), because a doubly linked list node knows its own neighbours.
• On set: create the node, put it in the map, push it behind the head. If memory is now over the limit, unlink nodes from in front of the tail until it fits, deleting each from the map as you go.
• On delete: unlink and drop from the map.

Every operation, eviction included, is O(1). Use sentinel head and tail nodes so you never write a special case for an empty list or a single element.

Then: 1 TB does not fit on one machine#

Fifty nodes, each running the structure above over its own slice of the keyspace.

The client hashes the key, walks the ring to find the owning node, and talks to it directly. One network hop, no proxy tier in the path. Each shard streams its writes to a replica asynchronously. A small config service (or gossip between the nodes) tells clients what the ring looks like and who is currently primary for each shard.

Everything from here is a deep dive into one of those arrows.

Deep Dives#

1. Expiry without scanning everything#

Checking the deadline on read is necessary but not sufficient. Keys that expire and are never read again never get cleaned up, and a workload with lots of short-lived keys will slowly fill the node with garbage.

The obvious fix is a background job that walks every entry and deletes the expired ones. Do not do this. At 5 billion entries, a full scan is minutes of CPU, and on a single-threaded server it blocks every client for the duration.

Use probabilistic sampling instead, which is what Redis does:

every 100ms:
  loop:
    pick 20 random keys from the set of keys that have a TTL
    delete the ones that have expired
    if fewer than 25% of the sample were expired: stop
    otherwise: loop again immediately

The logic is neat. If a large fraction of your sampled keys are expired, there is probably a lot of garbage, so keep going. If most samples come back alive, the keyspace is clean and you stop. Work per cycle is bounded, so it never blocks, and the expired fraction stays statistically low without anyone ever scanning the whole keyspace.

Note the two mechanisms cover different gaps. Lazy expiry guarantees correctness (an expired key is never returned, even a microsecond after it expires). Sampling guarantees memory is reclaimed for keys nobody reads. You need both, and saying that explicitly is the difference between a good and an average answer here.

2. Exact LRU is correct, and real caches avoid it#

The hash map plus linked list gives textbook LRU, and it is what the interviewer wants to see you derive. Then comes the follow-up worth preparing for: what happens to that design on a 16-core machine serving 100,000 requests a second?

The problem is that a read mutates the list. Every get splices a node to the front, which means every read takes a write lock on a structure shared by every core. Your gets are logically read-only and physically not, so they serialise. The linked list becomes the bottleneck long before memory or network do.

Production caches all back away from exact LRU in some form:

Decision: design the exact version, then name the ceiling yourself. "I'd implement hash map plus doubly linked list, and I'd expect the list lock to show up in a profile before anything else does. At that point I'd switch to sampled eviction with an access timestamp, which is what Redis does and which trades a bit of accuracy for removing the shared write on the read path." Saying that unprompted is a strong signal.

3. Sharding: why hash(key) % N is dangerous#

Fifty nodes, so the routing question is which node owns a given key. The naive answer is hash(key) % 50.

That works until the node count changes. Go from 50 to 51 and nearly every key maps somewhere new. In a database that would be a painful migration. In a cache it is worse than painful, and it is worth being precise about why:

Every relocated key is an instant cache miss, and all those misses hit your origin at the same moment. Your database is sized for the ~5% of traffic that misses today. Resize the cluster with modulo hashing and it briefly gets close to 100% of 100,000 requests a second. Adding capacity to your cache takes down your database. That is the real argument, and it lands much harder than "we would have to move some keys."

Consistent hashing avoids it. Hash node identifiers onto a fixed ring (say 0 to 2^32), hash each key onto the same ring, and a key belongs to the first node you meet walking clockwise. Add a node and it takes over only the arc between itself and its predecessor, so roughly 1/N of keys move and everything else stays put.

Virtual nodes are not optional. With one position per node, fifty random points on a ring produce badly uneven arcs, and you end up with nodes holding two or three times their fair share. Give each physical node 150 to 200 positions on the ring and the law of large numbers smooths it out. Virtual nodes also fix removal: without them, a dying node dumps its entire keyspace onto exactly one successor, which promptly falls over too. With them, its load spreads across every remaining node.

Where does the ring live?

• Client-side. The client library holds the ring and connects to the owning node directly. One hop, lowest latency, no extra tier to run. The cost is that ring updates have to reach every client, and clients get more complex. Redis Cluster works this way, with nodes returning a redirect if a client's view is stale.
• Proxy tier. Clients talk to a proxy that owns the routing. Dumb clients, topology changes in one place, but every request pays an extra network hop and the proxy tier is now something else to scale and keep alive.

Decision: client-side routing. With a 10ms p99 budget, spending a hop on a lookup you can do locally in microseconds is hard to justify, and the ring is small enough to push to clients cheaply.

4. Replication, failover, and whether you need it at all#

A node dies. Its share of the keyspace vanishes. What happens next depends on choices you make now.

Synchronous replication means the primary waits for the replica to acknowledge before it acks the client. No writes are lost on failover, but every write pays an extra network round trip, and a slow replica makes every write slow. For a store whose entire purpose is being fast, and whose data is by definition reconstructible, that is the wrong trade.

Asynchronous replication means the primary acks immediately and ships the write to the replica in the background. Writes stay fast. A primary that crashes loses the last few milliseconds of writes, which for a cache means a handful of extra misses. Nobody notices.

Decision: async, one replica per shard, placed in a different availability zone so a zone failure does not take both copies.

Failover needs more than a replica. Three pieces:

1. Failure detection. Nodes gossip heartbeats, or a coordinator polls. Either way you need a timeout that is long enough not to trip on a GC pause and short enough to matter.
2. Agreement. One node thinking another is dead is not enough, it may be the one with the broken network link. Require a quorum of nodes to agree before promoting a replica.
3. Fencing. Every promotion bumps an epoch number for that shard. The old primary, if it comes back, sees a higher epoch and refuses to serve. Without this, a node that was merely unreachable rejoins and starts answering for a shard it no longer owns, and clients get different answers for the same key depending on who they ask.

5. Hot keys#

One key gets a wildly disproportionate share of traffic. A viral post, a feature flag that every request reads, the inventory count for the item in today's flash sale.

Consistent hashing does not help here, and it is worth being clear why: sharding spreads keys across nodes, and this is one key. It lives on exactly one node, and that node now absorbs a load that was meant for fifty.

For hot reads:

• A client-side micro-cache is the best fix, and the most overlooked. Have the client library keep the handful of hottest keys in local process memory for a second or two. It removes the network hop entirely, costs nothing to run, and scales automatically because it gets better the more clients you have. The price is a second of staleness, which the requirements already said is fine. Try this before anything clever.
• Replicate the hot key across nodes. Write it as key#0 through key#9, have readers pick a suffix at random, and the read load spreads over ten nodes. Writes now have to update all ten copies, so this only suits keys that are read constantly and written rarely.
• Read replicas for the hot shard. Spin up extra replicas of just that shard and round-robin reads across them. Heavier to operate, and it helps the whole shard rather than targeting the one key.

For hot writes, your options narrow, because writes to one key cannot be spread without changing what the key means:

• Split the value. A counter becomes ten counters, counter#0 to counter#9, each incremented by a random writer. Reads sum all ten. Writes scale linearly; reads cost ten lookups instead of one.
• Coalesce at the client. Hold increments in process for 100ms and flush one combined write. A thousand increments a second per client becomes ten. You trade a little freshness for an order of magnitude less write traffic, which for counters is almost always the right call.

Detection is the part people skip. Every fix above requires knowing which key is hot, and most teams find out from a latency alert, which is far too late. Keep sampled per-key request counters on each node, or a count-min sketch (a few KB tracks the top talkers across millions of keys), and expose the top N. Then a hot key is a dashboard you look at rather than an incident you debug.

Worth saying plainly in the interview: a hot key is usually a property of the application, not the cache. But the cache is the only place that can see it happening, so giving it the telemetry to report one is part of the design.

6. Holding a 10ms p99#

On a single node the operations are microseconds. Everything that pushes you toward 10ms is the network and the tail.

• Pooled, persistent connections. A TCP handshake per request is a round trip you cannot afford. Clients hold a pool of open connections per node and reuse them.
• A compact binary protocol, not HTTP. For a 200-byte value, HTTP headers can be larger than the payload, and parsing them costs more than the lookup. Redis and Memcached both use small purpose-built protocols over raw TCP for exactly this reason.
• Batch and pipeline. MGET k1 k2 k3 is one round trip instead of three. A page needing 30 cached values should make one request, not thirty. This is usually the single biggest latency win available.
• One hop. Client-side routing means the client reaches the right node first time, with no proxy or lookup service in the path.

Then the tail-latency traps, which is where p99 actually goes wrong:

• One slow command blocks everything. Redis processes commands on a single thread per shard, so one KEYS * or one operation over a huge value freezes that shard for every client. Do not expose O(n) commands on a production cache, and cap value sizes.
• Garbage collection. A JVM cache node holding 24 GB of live objects can pause for hundreds of milliseconds. This is why serious caches are written in C or Rust, or keep data off-heap.
• Memory fragmentation. An allocator holding 24 GB in small objects can have real usage drift well above what you think you are storing. If the box starts swapping, your microsecond lookup becomes a disk read. Track resident memory, not just the bytes you counted.

Data Model#

There is no database here. The data model is the in-memory layout on each node.

Entry  (one per key, on the owning shard's primary)
  key         bytes
  value       bytes
  expires_at  int64, absolute epoch millis, 0 = never expires
  prev, next  pointers into the recency list

Node state
  table       hash map: key -> *Entry     O(1) lookup
  lru         doubly linked list          head = most recent, tail = evict next
  with_ttl    set of keys that have a TTL (the janitor samples from here)
  used_bytes  current memory accounted for
  max_bytes   configured limit

Cluster state  (in the config service, gossiped between nodes)
  ring        sorted map: ring position -> node id   (~200 positions per node)
  shards      shard id -> { primary: node, replicas: [node], epoch: int }

Two details that matter more than they look:

used_bytes must count the bookkeeping, not just len(key) + len(value). If you only count payload you will be evicting far too late, discover the process using twice what you budgeted, and get OOM killed at what your own metrics call 60% full.

with_ttl exists so the janitor samples from keys that can actually expire. Sampling the full keyspace when only 5% of keys have a TTL wastes 95% of the work.

API Design#

This is not a REST API, and explaining why is worth a sentence in the interview.

GET    key                     -> value | NOT_FOUND
SET    key value [ttl_ms]      -> OK
DEL    key                     -> DELETED | NOT_FOUND
MGET   key1 key2 key3 ...      -> [value | NOT_FOUND, ...]

Wire format: compact binary framing over a pooled, persistent TCP connection.
  [op:1][flags:1][key_len:2][val_len:4][ttl_ms:4][key][value]

Pipelining: a client may send N requests without waiting, and read N replies
in order. This is what keeps a 30-key page render to one round trip.

HTTP would put a few hundred bytes of headers and a text parse in front of a lookup that takes microseconds, on a request whose entire payload is often smaller than the headers. With a 10ms p99 budget and 100,000 requests a second, that overhead is the design.

MGET deserves to be in the first draft of the API rather than added later. Most callers want several keys at once, and giving them one round trip instead of N is the cheapest latency improvement in the whole system.

Failure Scenarios and Edge Cases#

Code#

The node: LRU with TTL#

class Entry:
    __slots__ = ("key", "value", "expires_at", "prev", "next")

class CacheNode:
    """Hash map for O(1) lookup, doubly linked list for O(1) recency.
    Behind head = most recently used. Before tail = evicted next."""

    def __init__(self, max_bytes):
        self.table, self.with_ttl = {}, set()
        self.max_bytes, self.used = max_bytes, 0
        self.head, self.tail = Entry(), Entry()      # sentinels: no empty-list special case
        self.head.next, self.tail.prev = self.tail, self.head

    def get(self, key, now):
        e = self.table.get(key)
        if e is None:
            return None
        if e.expires_at and now >= e.expires_at:     # lazy expiry: correctness on read
            self._unlink(e)
            return None
        self._to_front(e)                            # note: a read that writes (see deep dive 2)
        return e.value

    def set(self, key, value, now, ttl_ms=None):
        old = self.table.get(key)
        if old:
            self._unlink(old)
        e = Entry(key, value, now + ttl_ms if ttl_ms else 0)
        self.table[key] = e
        if e.expires_at:
            self.with_ttl.add(key)
        self._push_front(e)
        self.used += self._cost(e)
        while self.used > self.max_bytes:            # evict from the cold end until it fits
            self._unlink(self.tail.prev)

    def expire_sample(self, now, sample=20):
        """Bounded active expiry. Keep going only while the keyspace looks dirty."""
        while self.with_ttl:
            keys = random.sample(tuple(self.with_ttl), min(sample, len(self.with_ttl)))
            expired = 0
            for k in keys:
                e = self.table.get(k)
                if e and e.expires_at and now >= e.expires_at:
                    self._unlink(e)
                    expired += 1
            if expired * 4 <= len(keys):             # under 25% expired: clean enough, stop
                return

    def _cost(self, e):
        # Count bookkeeping, not just payload, or you OOM at "60% full".
        return len(e.key) + len(e.value) + ENTRY_OVERHEAD_BYTES

    def _unlink(self, e):
        e.prev.next, e.next.prev = e.next, e.prev
        self.table.pop(e.key, None)
        self.with_ttl.discard(e.key)
        self.used -= self._cost(e)

    def _push_front(self, e):
        e.prev, e.next = self.head, self.head.next
        self.head.next.prev = self.head.next = e

    def _to_front(self, e):
        e.prev.next, e.next.prev = e.next, e.prev    # detach
        self._push_front(e)                          # reattach behind head

The ring: routing a key to a node#

Small enough that every client holds a copy and routes locally, with no lookup hop.

VNODES = 200        # positions per physical node; fewer means worse balance

class Ring:
    def __init__(self, nodes):
        self.positions = []            # sorted ring positions
        self.owner = {}                # position -> node id
        for n in nodes:
            self.add(n)

    def add(self, node):
        for i in range(VNODES):        # spread each node over the ring, not one point
            p = murmur3(f"{node}#{i}")
            self.positions.append(p)
            self.owner[p] = node
        self.positions.sort()

    def remove(self, node):
        # Without vnodes this arc would land entirely on one successor, and sink it.
        self.positions = [p for p in self.positions if self.owner[p] != node]
        self.owner = {p: n for p, n in self.owner.items() if n != node}

    def route(self, key):
        i = bisect.bisect_left(self.positions, murmur3(key))   # first position >= hash(key)
        if i == len(self.positions):
            i = 0                                              # wrap around the ring
        return self.owner[self.positions[i]]

Trade-off Summary#

Follow-up Questions#

Basic

Q: Why a hash map and a doubly linked list rather than one structure?

They solve different halves. The hash map finds any key in O(1) but knows nothing about access order. A list tracks order but finding an item in it is O(n). Point both at the same entry objects and you get both properties: the map finds the node instantly, and because the node is doubly linked it knows its own neighbours, so unlinking and re-splicing it at the head is O(1) too. Eviction is then just "remove whatever sits before the tail."

Q: Why consistent hashing instead of hash(key) % N?

Because modulo remaps nearly every key when the node count changes, and in a cache every remapped key is an instant miss. Resizing a 50-node cluster would send close to 100% of 100,000 requests a second at your origin database at once, which is how adding cache capacity causes a database outage. Consistent hashing moves only about 1/N of keys, so the miss burst is small enough to absorb.

Q: Why is asynchronous replication the right choice here?

Synchronous replication puts a network round trip in front of every write and makes a slow replica everyone's problem. What it buys you is not losing writes on failover. In a cache, a lost write means a miss, and a miss means going to the origin, which is the normal path anyway. Paying latency on every single write to avoid a handful of extra misses during a rare failover is a bad trade.

Q: If keys are spread evenly, why are hot keys still a problem?

Sharding distributes keys, and a hot key is one key. It hashes to one position on the ring and lives on one node, so consistent hashing has nothing to spread. That node takes traffic meant for the whole cluster. Fixing it means either caching the value closer to the caller, or deliberately storing several copies under different key names so readers spread out.

Q: Why not just expose this over a REST API?

For a 200-byte value with a 10ms p99 budget, HTTP headers can be bigger than the payload and parsing them costs more than the cache lookup itself. Real caches use small binary protocols over persistent, pooled TCP connections, with pipelining so a client can send many requests without waiting for each reply. That is also what makes MGET worth having: one round trip for thirty keys instead of thirty.

Minute-by-Minute Interview Playbook#

0 to 5 min: Requirements

Ask the questions above, and pin down scale early because it decides how much of the interview is data structures versus distribution. Get the interviewer to confirm LRU specifically rather than assuming it. Write down AP, the 10ms target, 1 TB, and 100k req/sec where everyone can see them.

5 to 9 min: Capacity estimates

Size on throughput and on memory separately, then take the larger: about 8 nodes for throughput, about 50 for memory, so memory decides. Mention per-entry overhead. It is a small thing that signals you have watched a cache run out of memory before.

9 to 20 min: Single node design

Build it up in three steps: hash map, then TTL, then hash map plus doubly linked list for LRU. Draw the two structures pointing at the same entries and walk a get through both. Mention sentinel nodes. This part should be fluent, because the rest of the interview needs the time.

20 to 42 min: Deep dives

Lead with sharding and consistent hashing, and frame the modulo problem in terms of the origin database falling over rather than just key movement. Cover virtual nodes and where the ring lives. Then replication and failover, including epochs. Then hot keys. If you get a chance, raise the exact-LRU lock problem yourself: it is the highest-signal thing you can say in this interview, because almost nobody brings it up unprompted.

42 to 45 min: Wrap up

Cover one failure (a primary dying and a replica being promoted with a bumped epoch) and one operational point (alert on eviction rate, not memory used, because memory sits at 100% by design). Close with: "One node is a hash map and a linked list. Fifty nodes is consistent hashing with virtual nodes, async replication with epoch-fenced failover, and client-side routing to keep it to one hop. Every trade-off went toward speed, because a miss is survivable."

Green flags

• Derives the hash map plus doubly linked list without being led to it
• Knows lazy expiry and active cleanup solve different problems and names both
• Explains the modulo problem in terms of origin load, not just key movement
• Brings up virtual nodes without being asked
• Raises the read-path lock in exact LRU and knows what Redis does instead
• Treats replication as a cost decision rather than a default
• Reaches for a client-side micro-cache on hot keys before anything more exotic
• Alerts on eviction rate rather than memory used

Red flags

• Uses a plain hash map and hand-waves eviction as "we'll clean it up somehow"
• Proposes scanning every key to expire entries
• Uses hash(key) % N and does not notice what a resize does
• Skips virtual nodes and assumes the ring balances itself
• Puts synchronous replication on the write path of a cache
• Designs a REST API with a 10ms p99 requirement
• Thinks sharding solves hot keys
• Counts only payload bytes toward the memory limit

Further Reading#