Guide contents

Lesson 119 of 129

Design a Distributed Cache

A distributed cache is a cluster of servers that keep frequently read data in memory, so applications avoid repeated reads from a slower store. Memory on one machine and the request rate one machine can serve are both limited. Spreading the data across many nodes raises both limits. This article designs the cache as a service (comparable to Redis Cluster or a Memcached fleet). Cache usage patterns are in caching fundamentals; this article concentrates on building the cache itself.

Requirements

Functional

  • get(key), set(key, value, ttl), and delete(key).
  • Values are opaque byte strings up to a size limit (for example, 1 MB).
  • Per-key time to live (TTL).
  • Capacity and throughput grow by adding nodes.

Non-functional

  • Latency. Sub-millisecond server time; about 1 ms end to end within a data center.
  • Throughput. Millions of operations per second across the cluster.
  • High availability. A node failure must not take the cache down. Losing cached data is acceptable, because the source of truth is elsewhere, but a mass loss causes a load spike on the database.
  • Scalability. Add and remove nodes with minimal data movement.
  • Consistency. Eventual consistency with the database is acceptable; bounded staleness through TTL and invalidation.
  • Durability. Not required. It is a cache.

Estimation

Assumptions: 200,000 requests per second peak, 100 million cached items, average item 1 KB.

  • Data size: 100,000,000 × 1 KB = 100 GB. With about 20 percent overhead for keys, metadata, and allocator fragmentation, roughly 120 GB.
  • Memory per node: 32 GB usable. Nodes needed for capacity: 120 / 32 ≈ 4. Replicating each shard once doubles this to about 8 nodes.
  • Throughput: one node handles on the order of 50,000 to 100,000 simple operations per second. 200,000 / 100,000 = 2 nodes minimum for throughput, so capacity, not throughput, is the constraint here. Peak headroom and hot keys argue for more nodes than the minimum.
  • Hit ratio target: 90 percent or more. If 200,000 requests per second reach the cache at 90 percent hit ratio, the database receives 20,000 requests per second. At a 99 percent hit ratio, it receives 2,000.
  • Bandwidth: 200,000 × 1 KB = 200 MB/s ≈ 1.6 Gbit/s across the cluster, about 0.4 Gbit/s per node with 4 primaries. That is fine on 10 GbE but would saturate 1 GbE, so large values, not request count, drive network limits.

Implications: sizing is memory-bound, so the node count comes from capacity plus headroom (target below 70 percent memory use and below 50 percent CPU at peak, so a failover onto a replica does not overload it). The hit ratio, not the cache, decides database size: a drop from 99 to 90 percent multiplies database load by 10, which is why a mass cache loss is the main risk.

API design

Client library or network protocol commands:

Operation Behavior
GET key Returns value or miss
SET key value [TTL] Stores value; overwrites
DELETE key Removes value; idempotent
CAS key value version (optional) Set only if the version matches

The protocol should be a compact binary or a simple text protocol over persistent TCP connections, with connection pooling on the client and support for pipelining (sending several requests without waiting for each response).

Data model

A cache node holds a hash table from key to entry, plus eviction metadata:

Code
entry { key, value, expires_at, last_access / frequency / list pointers }
  • Hash table gives O(1) lookup.
  • Eviction structure: a doubly linked list for LRU, or approximated LRU by sampling. See eviction policies.
  • Memory management: slab allocation (Memcached) groups items in size classes to limit fragmentation; alternatively use a general allocator such as jemalloc (Redis).
  • Expiration: lazy (check TTL on access) plus an active background sampler that removes expired entries so unread keys do not hold memory.

The cluster-level data model is the mapping of keys to nodes (the partition map), described next.

High-level design

flowchart LR
    APP["Application servers + cache client library"]
    CFG[("Config / membership service")]
    subgraph Shard1
        P1["Primary 1"] --> R1["Replica 1"]
    end
    subgraph Shard2
        P2["Primary 2"] --> R2["Replica 2"]
    end
    subgraph Shard3
        P3["Primary 3"] --> R3["Replica 3"]
    end
    APP -->|"key hash to shard"| P1
    APP --> P2
    APP --> P3
    CFG -.->|"membership and partition map"| APP
    CFG -.-> P1
    CFG -.-> P2
    CFG -.-> P3
    APP -->|"on miss"| DB[("Database")]

Component responsibilities:

  • Client library. Holds the partition map, hashes the key, sends the request to the owning node, retries on timeouts, and can fall back to a replica.
  • Cache nodes. Store data, enforce TTL and eviction, replicate to followers.
  • Membership/config service (ZooKeeper, etcd, or a gossip protocol). Tracks live nodes, publishes partition map changes, coordinates failover.
  • Database. Source of truth. Under cache-aside, the application reads the cache first, reads the database on a miss, and then stores the result in the cache.

Deep dive

1. Partitioning: how keys map to nodes

Modulo hashing (hash(key) mod N) remaps almost every key when N changes, which empties the cache and floods the database. Two established alternatives:

Scheme How it works Remap on add/remove Used by
Consistent hashing with virtual nodes Nodes and keys are hashed onto a ring; a key belongs to the next node clockwise About 1/N of keys Memcached client libraries (ketama), Dynamo-style stores
Fixed hash slots The key space is split into a fixed number of slots; a table maps slots to nodes; keys map to slots with hash(key) mod slots Only the moved slots Redis Cluster (16,384 slots)

Slots make rebalancing explicit and controllable: to add a node, move some slots to it. Consistent hashing needs no central table, but load balance depends on virtual node count (typically 100 or more per node). See Redis and Memcached.

2. Client-side, proxy, and server-side routing

Option Description Trade-offs
Client-side routing Client library holds the map and connects directly Fewest network hops; every client must be upgraded to change logic; many connections to every node
Proxy routing Clients talk to a stateless proxy tier (for example, twemproxy, mcrouter, Envoy) that routes Simple clients and connection pooling; extra hop and proxy capacity to manage
Server-side redirect Any node accepts a request and redirects the client to the owner (Redis Cluster MOVED) No proxy; clients still learn the map from redirects

Large deployments with many client languages typically use a proxy tier; performance-critical paths use smart clients.

3. Replication and failover

Without replication, a node failure loses its shard and its keys miss until rebuilt from the database, which can overload the database. Each shard has a primary and one or more replicas.

  • Replication is asynchronous. The primary applies the write and acknowledges it, then streams it to replicas. This keeps latency low but means a failover can lose the most recent writes. That is acceptable for a cache; use a TTL to bound resulting staleness.
  • Failure detection. The membership service or peers use heartbeats; a node unresponsive for a timeout is marked failed (see failure detection).
  • Promotion. A replica is elected primary, the partition map is updated, and clients refresh their maps. To prevent two primaries for one shard (split brain), use an epoch number so nodes and clients reject stale primaries (see leader election).
  • Reads from replicas increase read capacity and can serve hot keys, at the cost of stale reads.

Whether to replicate at all is a trade-off: it doubles memory cost but protects the database from a cold shard.

Concrete thresholds: a heartbeat every 1 second and a failure timeout of 5 to 15 seconds is typical (Redis Cluster's cluster-node-timeout defaults to 15 seconds). Shorter timeouts fail over faster but cause false failovers during GC pauses or network blips. Replication lag is usually milliseconds, so the loss window on failover is the writes in that lag. The alternative, synchronous replication or Redis's WAIT, adds a round trip per write and still does not guarantee no loss. For a cache, choose asynchronous.

Choosing a sharded primary/replica design over client-side replication (the client writes every copy to several nodes, as in some Memcached setups) keeps write logic out of clients and gives one authoritative copy to fail over from, at the cost of a failover pause.

4. Hot keys and thundering herd

A single very popular key limits the throughput of its one owner node, and adding nodes does not help.

  • Local (near) cache in the client. Cache the hottest keys in the application process for a very short TTL (seconds).
  • Key replication. Store a hot key under several derived names (key#0 ... key#k), each on a different node; clients read a random copy. Writes must update or invalidate all copies.
  • Reads from replicas for hot shards.
  • Request coalescing. On a miss, only one caller loads from the database; others wait for its result.

Detect hot keys by sampling access counts per key on each node (or in the proxy) and promoting any key above a threshold, for example more than 10 percent of a node's traffic or more than 20,000 requests per second. A 1 to 5 second local TTL costs at most that much staleness and can cut load on the key by the number of client instances. Choose local caching for read-mostly keys; choose key replication when staleness of even a few seconds is unacceptable but you can invalidate all copies.

On expiry of a popular key, many requests miss at once and hit the database. Mitigations include request coalescing, locks with a short lease on refresh, and refreshing before expiry with randomized (jittered) TTLs. See cache invalidation, stampede and hot keys.

5. Consistency with the database

Under cache-aside, the reader loads on miss and writers update the database and then delete the cache entry. Deleting is preferred to writing the new value, because two concurrent writers could otherwise store their values in the cache in the opposite order from the database. A race can leave stale data: reader A misses and reads the old value; writer B updates the database and deletes the key; reader A then writes the old value into the cache. Options:

  • Short TTLs bound how long a stale entry lives.
  • Versioned values: each value carries a version taken from the database row, and a cache write is accepted only if its version is at least the cached one (CAS, compare-and-set).
  • Lease tokens: on a miss, the cache gives one client a lease token; a delete invalidates outstanding tokens so the stale set is rejected (used in Facebook's Memcache deployment).
  • Publish invalidations from database changes through change data capture, so the cache is updated by the log rather than by application code.

Strong consistency with the database would need transactions across systems and defeats the purpose of a cache.

Bottlenecks and scaling

  • Memory pressure. Set maxmemory and eviction policy; monitor evictions, since a high eviction rate signals undersizing.
  • Rebalancing. Move slots or ring ranges gradually, and throttle transfer bandwidth so live traffic keeps its latency.
  • Cold start. A new node has no data. Warm it from a replica or by replaying popular keys, and ramp traffic to it gradually.
  • Network. Pipelining and batching (multi-get) reduce round trips; multi-get across shards needs one request per shard.
  • Multi-tier caching. Local in-process cache in front of the distributed cache in front of the database. Each tier lowers load on the next, and adds another place for staleness.
  • Multi-region. Each region runs its own cache with invalidations propagated across regions; cross-region reads on miss are expensive.
  • Large values and big keys. Cap value size; large values block a single-threaded server and use bandwidth.

Failure modes

What breaks How it degrades Mitigation
One primary dies Its slots (1/N of keys) error or miss for the detection plus promotion time (roughly 5 to 20 seconds) Replica promotion; clients retry with backoff and fall back to the database with a per-key concurrency limit
Primary and its replica both die That shard's data is gone; database sees 1/N of traffic at miss rate Place replicas in different racks or AZs; warm before serving
Network partition Minority side may keep a stale primary Epoch numbers; primaries that cannot reach a majority stop accepting writes after a timeout
Rolling restart or deploy Cold nodes lower hit ratio Restart one node at a time, restore from snapshot, ramp traffic
Entire cluster lost Database receives 100 percent of 200,000 requests/s Load shedding, request coalescing, circuit breaker, serve stale from a local tier

Degradation should be graceful: a cache failure raises latency and database load but must not become an outage, so clients use short timeouts (for example 10 to 50 ms) and treat timeouts as misses. At 10x (2 million requests/s, about 1.2 TB): the cluster grows to about 40 primaries, where full-mesh gossip and client connection counts (clients × nodes) start to hurt, so a proxy tier that pools connections becomes preferable. Hot keys become a certainty rather than an edge case, since any key above 1/40 of traffic overloads its owner. Rebalancing moves more data per node addition, so throttle it, and the failure of one shard now removes a smaller share (2.5 percent) of the cache.

Trade-offs and follow-up questions

  • Cache-aside versus read-through. Cache-aside puts the logic in the application and is resilient to cache outages; read-through hides the database behind the cache and centralizes stampede protection.
  • Availability versus consistency. During a partition, serve possibly stale data or fail reads; caches usually serve.
  • Persistence. Snapshots (RDB) or append logs (AOF) make restarts faster but cost disk and write throughput, and make the cache a partial data store.
  • What happens when the cache cluster is down? The database must survive the load or be protected with rate limiting and load shedding; circuit breakers avoid retrying a dead cache on every request.
  • Why not one large node? Single-machine memory, network, and CPU limits, and a single failure domain.
  • What if we add a node during peak traffic? With hash slots, migrate slots a few at a time; requests for a moving slot get an ASK redirect to the new owner until it completes. With consistent hashing, about 1/(N+1) of keys move, so add during low traffic and expect a temporary hit-ratio dip on those keys.
  • What if the database is slow and the cache is full of stale entries? Serving stale-while-revalidate keeps latency low; cap staleness with a hard TTL.
  • What if LRU evicts the wrong things (a scan pollutes the cache)? Use LFU or a segmented policy such as TinyLFU, which admits a new item only if it is more popular than the victim.
  • What if a value is 50 MB? Reject it or store it in blob storage and cache a pointer; one big value stalls a single-threaded server and skews memory.
  • What if we need atomic counters or lists? A single owner per key makes single-key operations atomic; multi-key operations require the keys to hash to the same slot (Redis hash tags).
  • Follow-ups: multi-tenancy with per-tenant quotas; encryption and authentication; observability of hit ratio per keyspace.