Distributed Cache

"Design a distributed in-memory cache like Memcached or Redis Cluster: a key-value store that many application servers use to avoid hitting the database, holding more data than one machine can, with sub-millisecond latency."

Almost every case in this course ends with "and put a cache in front of it". This case opens that box. A distributed cache is a small distributed database with relaxed rules: it may lose data (the source of truth is elsewhere), but it must be fast, spread keys evenly over many nodes, survive nodes coming and going without a stampede on the database, and cope with keys that are much hotter than others. It is the interview where consistent hashing, eviction policies, replication lag and hot keys all appear in one design — the same ideas behind Dynamo, Cassandra and Redis Cluster.

In BoxOffice, the cache holds event pages, seat-availability snapshots and sessions. At 10:00 on an on-sale day it absorbs millions of reads per minute; if one cache node dies at 10:01, the question is whether its share of the traffic moves smoothly to the others or falls on PostgreSQL all at once.

1. Clarify the requirements

(interview framework)

Functional requirements

1. get(key), set(key, value, ttl), delete(key); optional atomic operations (incr, compare-and-set).
2. Keys expire after their TTL; when memory is full, the cache evicts some keys.
3. The data set is larger than one machine: keys are partitioned over many nodes.
4. Nodes can be added or removed (scaling, failures) without restarting clients.

Non-functional requirements

Out of scope (say it explicitly)

Persistence to disk (snapshots/AOF), cross-region replication, complex data structures beyond strings and counters, authentication — mention that Redis offers several of them.

2. Back-of-the-envelope estimates

# Distributed cache: assumptions (illustrative)
items         = 1_000_000_000   # cached objects (working set)
value_b       = 1_000           # average value size
key_b         = 50
overhead_b    = 50              # per-item metadata, pointers, allocator slack
node_mem_gb   = 64              # RAM per node usable for data (leave headroom)
ops_s         = 2_000_000       # total cache operations per second at peak
ops_per_node  = 80_000          # comfortable ops/s per node (single-threaded engine)
replicas      = 1               # replicas per primary
hit_ratio     = 0.95

data_tb       = items * (value_b + key_b + overhead_b) / 1e12
nodes_memory  = data_tb * 1e3 / node_mem_gb
nodes_ops     = ops_s / ops_per_node
primaries     = max(nodes_memory, nodes_ops)
total_nodes   = primaries * (1 + replicas)
net_gb_s      = ops_s * value_b / 1e9
db_rps        = ops_s * (1 - hit_ratio)
db_rps_90     = ops_s * (1 - 0.90)

rows = [
    ("data_tb", data_tb, "TB"), ("nodes_memory", nodes_memory, "nodes"),
    ("nodes_ops", nodes_ops, "nodes"), ("primaries", primaries, "nodes"),
    ("total_nodes", total_nodes, "nodes"), ("net_gb_s", net_gb_s, "GB/s"),
    ("db_rps", db_rps, "req/s"), ("db_rps_90", db_rps_90, "req/s"),
]
print("| quantity | value | unit |\n|---|---:|---|")
for name, value, unit in rows:
    v = f"{value:,.0f}" if abs(value) >= 100 else f"{value:,.2f}"
    print(f"| `{name}` | {v} | {unit} |")

What the numbers tell us:

• ~1.1 TB of data needs ~17 nodes by memory, but 2 M ops/s need 25 nodes by throughput. ⇒ size by both and take the larger: here throughput decides (25 primaries, 50 nodes with one replica each).
• 2 GB/s of values cross the network at peak — spread over 25 nodes it is ~80 MB/s each, fine for modern NICs, but a single big key read thousands of times per second can saturate one node's link.
• At a 95% hit ratio the database receives 100 k req/s; at 90% it receives 200 k req/s. ⇒ every point of hit ratio is worth 20 k database requests per second. Losing one node of 25 (4% of the keys) cold is a sudden +80 k misses/s for the database — the reason failover and warm replicas matter (caching).

3. API

GET    key                         → value | nil
SET    key value EX 300            → OK                 (TTL in seconds)
SET    key value NX EX 30          → OK | nil           (only if absent: locks, leases)
DEL    key                         → 1 | 0
INCRBY key 5                       → 42                 (atomic counter)
MGET   k1 k2 k3                    → [v1, v2, v3]       (batched: one round trip)

# Cluster-aware client: the key decides the node
slot = CRC16(key) mod 16384        → node owning that slot (Redis Cluster)
-MOVED 3999 10.0.1.17:6379         → slot moved: update the client's map and retry there

• The client library is part of the API: it hashes the key, keeps the map of partitions → nodes, pools connections and pipelines requests.
• MGET across keys on different nodes becomes a scatter-gather; use hash tags ({event:42}:seats, {event:42}:page) to keep related keys on the same node when they are read together.

4. Data model

# On each node: a hash table of entries + an eviction structure
entry  = { key, value, expires_at, flags, lru_clock | lfu_counter }
index  = hash table key → entry                   (O(1) get/set)
policy = LRU list / sampled LRU / LFU counters   (who to evict when memory is full)

# Cluster metadata (gossiped between nodes or stored in a coordination service)
slots 0–5460      → node-a (primary), node-d (replica)
slots 5461–10922  → node-b (primary), node-e (replica)
slots 10923–16383 → node-c (primary), node-f (replica)

5. High-level design

flowchart LR
    app1["App server #1<br/>cluster-aware client"] --> a["node-a<br/>slots 0–5460"]
    app1 --> b["node-b<br/>slots 5461–10922"]
    app2["App server #2<br/>cluster-aware client"] --> b
    app2 --> c["node-c<br/>slots 10923–16383"]
    a -. async replication .-> ra["replica a'"]
    b -. async replication .-> rb["replica b'"]
    c -. async replication .-> rc["replica c'"]
    app1 -- miss --> db[("Database<br/>source of truth")]
    app2 -- miss --> db
    a -. metrics .-> m["Metrics"]

A read with cache-aside, including a slot that moved:

sequenceDiagram
    autonumber
    participant A as App (client lib)
    participant B as node-b
    participant C as node-c
    participant D as Database
    A->>A: slot = CRC16("event:42") mod 16384 → node-b (cached map)
    A->>B: GET event:42
    B-->>A: -MOVED 9001 node-c (slot was migrated)
    A->>A: update slot map
    A->>C: GET event:42
    C-->>A: nil (miss)
    A->>D: SELECT event 42
    D-->>A: row
    A->>C: SET event:42 … EX 300

6. Deep dives

6.1 Partitioning: why hash % N breaks, and consistent hashing

With node = hash(key) % N, going from 4 to 5 nodes changes the owner of almost every key — a cache with 80% of its keys suddenly in the wrong place is, for the database, the same as an empty cache. Consistent hashing places nodes and keys on the same ring of hash values; a key belongs to the first node clockwise. Adding a node only takes keys from its neighbors — about 1/N of them. Each physical node gets many virtual nodes (points on the ring) so that load is even and a failed node's keys spread over all the others, not just one neighbor (partitioning).

import bisect
import hashlib

def h(s: str) -> int:
    return int.from_bytes(hashlib.md5(s.encode()).digest()[:8], "big")

keys = [f"event:{i}" for i in range(50_000)]
kh = {k: h(k) for k in keys}

def mod_n(n):
    return {k: kh[k] % n for k in keys}

def ring(nodes, vnodes):
    points = sorted((h(f"node{n}#{v}"), n) for n in range(nodes) for v in range(vnodes))
    hashes = [p for p, _ in points]
    owner = {}
    for k in keys:
        i = bisect.bisect(hashes, kh[k]) % len(points)
        owner[k] = points[i][1]
    return owner

def moved(a, b):
    return sum(a[k] != b[k] for k in keys) / len(keys)

def max_over_avg(owner, nodes):
    counts = [0] * nodes
    for n in owner.values():
        counts[n] += 1
    return max(counts) / (len(keys) / nodes)

print("| scheme | keys_moved_pct | max_load_vs_avg |\n|---|---:|---:|")
a, b = mod_n(4), mod_n(5)
print(f"| `hash % N` | {moved(a, b):.1%} | {max_over_avg(b, 5):.2f} |")
for v in (1, 10, 100, 1000):
    a, b = ring(4, v), ring(5, v)
    print(f"| `ring, {v} vnodes/node` | {moved(a, b):.1%} | {max_over_avg(b, 5):.2f} |")

Read the table in two columns. Keys moved going from 4 to 5 nodes: hash % N moves ~80%; the ring moves ~20% — the theoretical minimum, 1/5. Load balance (the busiest node relative to the average): with one point per node the ring is very uneven (the busiest node holds almost twice its share); with hundreds of virtual nodes it approaches 1.0.

6.2 Eviction, TTL and memory

When memory is full, something must go. The policy decides the hit ratio you get from the same RAM (caching failure modes):

• Real engines approximate: Redis samples a few keys and evicts the best candidate among them (approximated LRU or LFU with decaying counters) instead of maintaining a perfect global order.
• TTL is a consistency tool, eviction is a memory tool. Always set a TTL (so stale data has a bound), and add jitter to TTLs written in bulk (e.g. 300 s ± 10%) so they do not expire at the same second.
• Leave headroom: plan for ~70–80% of RAM for data. Fragmentation, replication buffers and forking for snapshots need the rest; a node that swaps to disk is slower than no cache at all.

6.3 Replication and failover

sequenceDiagram
    autonumber
    participant A as App
    participant P as Primary (node-b)
    participant R as Replica (node-e)
    participant S as Failure detector
    A->>P: SET seats:42 v7
    P-->>A: OK
    P--)R: replicate v7 (async)
    Note over P: primary crashes before v7 reaches the replica
    S->>S: node-b missed heartbeats → majority agrees
    S->>R: promote node-e to primary
    A->>R: GET seats:42
    R-->>A: v6 (the acknowledged write v7 is lost)

• Replication is usually asynchronous: fast writes, but a failover can lose the last acknowledged writes. For a cache that is acceptable — the next miss reloads from the database. If a stale value is dangerous (seat availability), keep TTLs short or delete keys on write (invalidation).
• Failure detection needs a majority (Redis Sentinel, or the cluster nodes voting) so that one isolated observer does not promote a replica on its own.
• Split brain: after a partition, the old primary may still accept writes from clients on its side. Mitigate by having a primary stop accepting writes when it cannot see enough replicas/peers (min-replicas-to-write), and by epochs/terms so clients and nodes reject the stale primary.
• Warm replicas also serve reads (if stale reads are acceptable) and avoid the cold-node stampede after failover.

6.4 Hot keys, big keys and stampedes

• Hot key — one key (the event page during an on-sale, a celebrity's profile) receives so many requests that its node saturates while the rest are idle. Consistent hashing cannot help: one key lives on one node. Fixes: a small in-process L1 cache with a TTL of seconds on every app server; replicate the key under several names (event:42#1 … #8) and read a random copy; serve it from replicas.
• Big key — a 5 MB value read 1 000 times per second is 5 GB/s on one node. Split it (paginate, compress), or cache the rendered fragments instead of the whole object.
• Stampede (thundering herd) — a hot key expires and thousands of requests miss at once and all hit the database. Fixes (caching failure modes):
  • Request coalescing / leases: the first miss gets a lease (SET lock:key NX EX 5) and reloads; the others wait briefly or get the stale value. Memcache at Facebook used leases for exactly this.
  • Probabilistic early expiration: each reader may refresh before the TTL with a probability that grows as expiry approaches, so one request refreshes early instead of all at once.
  • Serve stale while revalidating for data where a few seconds of staleness are fine.

import math, random, time

def should_refresh_early(expires_at: float, recompute_s: float, beta: float = 1.0) -> bool:
    """XFetch: refresh before expiry with a probability that grows near the deadline."""
    return time.time() - recompute_s * beta * math.log(random.random()) >= expires_at

if (stampedeProtection) {
    var lockKey = key + ":lock";
    var token = UUID.randomUUID().toString();
    if (tryLock(lockKey, token)) {
        try {
            cached = read(key);               // someone may have filled it meanwhile
            if (cached != null) {
                return count(new Result(cached, "HIT"));
            }
            return count(load(key, ttl, loader));
        } finally {
            unlock(lockKey, token);
        }
    }
    // another request is already loading this key: wait for it instead of
    // sending one more identical query to the database
    for (int i = 0; i < 20; i++) {
        sleep(Duration.ofMillis(25));
        cached = read(key);
        if (cached != null) {
            return count(new Result(cached, "HIT"));
        }
    }
}

6.5 Routing: who knows where a key lives?

Facebook's memcache deployment, for example, put routing in a proxy (mcrouter) so that thousands of web servers did not each manage connections to every cache node.

7. Bottlenecks, failure modes and trade-offs

The trade-offs to say out loud: availability and speed over consistency (a cache may be stale or lose writes, the database is the truth), memory vs. throughput (size by both), and even distribution vs. hot keys (hashing spreads keys, not popularity).

8. Try it in the playground

Two scenarios cover the two halves of this case: the cache in front of a database, and the partitioning of data over nodes.

Open the full playground

9. Follow-up questions an interviewer asks

10. Self-assessment rubric

Related: Caching, Redis, Caching failure modes, Partitioning, Hands-on 5 (Redis cache-aside, hit ratio and stampede).

How it connects

Which classes does this case test? Each branch is a class whose ideas the design relies on (blue), with the palette components where they show up (green):

flowchart LR
    CASE(["Distributed cache"]) --> EST["Estimation<br/>size by memory and ops/s"]:::tech
    CASE --> PART["Databases<br/>consistent hashing, virtual nodes"]:::tech
    CASE --> EVI["Caching<br/>eviction LRU, LFU, W-TinyLFU, TTL"]:::tech
    CASE --> REP["Databases<br/>async replication, failover"]:::tech
    CASE --> HOT["Caching<br/>hot keys, big keys, stampedes"]:::tech
    CASE --> ROUTE["routing<br/>cluster-aware client"]:::tech
    PART -.-> k1["cache (cluster)"]:::comp
    REP -.-> k2["sql · nosql<br/>(source of truth)"]:::comp
    ROUTE -.-> k3["app"]:::comp
    HOT -.-> k4["metrics"]:::comp
    classDef comp fill:#1e9e5a22,stroke:#1e9e5a
    classDef tech fill:#2f6fed22,stroke:#2f6fed

Classes it tests: Caching — patterns · Caching — Redis · Caching — failure modes · Databases — partitioning · Databases — replication · Estimation — numbers · Hands-on 5. Palette components: cache · app · sql · nosql · metrics. Scenarios: read-heavy-cache, sharding.

References

• KARGER, D. et al. Consistent Hashing and Random Trees¹.
• DECANDIA, G. et al. Dynamo: Amazon's Highly Available Key-value Store².
• NISHTALA, R. et al. Scaling Memcache at Facebook — leases, regional pools, mcrouter³.
• EINZIGER, G.; FRIEDMAN, R.; MANES, B. TinyLFU: A Highly Efficient Cache Admission Policy⁴.
• XU, A. System Design Interview, vol. 1, ch. 5 "Design Consistent Hashing" and ch. 6 "Design a Key-Value Store"⁵; Redis Cluster specification⁶.