Skip to content

Distributed cache

intermediateCommon atAnthropicGoogleMetaAmazonStripeRead firstCachingTTL and expiryLoad balancingDistributed thinkingPractise in the Sandbox →

A distributed cache is a hash map spread over many machines’ RAM, sitting in front of a slower source of truth. One node is the caching concept: a hash map plus an eviction order, O(1) per operation. The design question is everything around it: how keys find their node (hash slots or a consistent-hash ring computed in the client), what happens on a miss (cache-aside with a lease, so one request reloads a hot key instead of thousands), and how writes invalidate (update the database, then delete the key, backed by a commit-log tailer). Reads cost one sub-millisecond round trip; memory is roughly keys × (value + 60 B).

Databases are durable and expressive, and slow and expensive per read: milliseconds and real CPU. Most read traffic is skewed: a small fraction of keys (popular products, session data, rendered fragments, feature flags) takes most of the reads. Keep that fraction in RAM and the database only sees the rest.

The cache has three goals, and they conflict:

Goal What it means Pulls toward
Absorb load High hit ratio, so the database sees only misses More memory, longer TTLs, smarter eviction
Stay fresh Readers don’t see a value the database has replaced Short TTLs, aggressive invalidation
Never be the outage A cache failure degrades latency, not correctness or availability Replication, timeouts, protecting the DB from refills

Longer TTLs help the first goal and hurt the second. Every eviction or failure dumps load on the database, which the third goal must survive.

In scope (typical) Ask about
Functional get, set with TTL, delete, multi-get. Opaque byte values up to ~1 MB. Are values opaque blobs, or structures (lists, sorted sets, counters)? Is a multi-key atomic op needed?
Scale ~500k reads/s peak, ~200M hot keys, ~1 KB average value. Read:write ratio? How skewed is access? How big is the largest value?
Latency p99 under 1-2 ms for a hit, in the same region. What’s the end-to-end budget? Is cross-region access needed?
Consistency Eventually consistent: a stale read for up to a few seconds after a write is fine. Which data must never be stale (balances, permissions)? Those may not belong in the cache at all.
Availability Losing a node costs hit ratio for a while, never wrong answers. Can the database survive a cold cache? Is a warm restart needed?
Out of scope Durable storage (the cache can lose anything), cross-key transactions, a query language.

Two different constraints size the cluster: memory for the working set, and operations per second. Find which one binds. Then look at the database row: it’s the misses, and it moves fast with the hit ratio.

Back of the envelope
drag the assumptions
assumptions
so you need
memory for the working set212 GB+≈60 B per item: key, header, pointers, allocator slack
nodes needed for memory (70% full)5
nodes needed for throughput (50% headroom)5
database reads/s (the misses)25K/swhat the DB must absorb; one point of hit ratio moves this a lot
cache egress bandwidth475 MB/snetwork cards fill before CPUs for large values
200M keys × ~1 KB is ~200 GB, about five 64 GB nodes by memory, while 500k reads/s needs about five nodes by throughput too. Now drag the hit ratio from 95% to 90%: the database load doubles from 25k to 50k reads/s. Large values flip the bottleneck to network bandwidth.

What to say out loud: “About 200 GB of hot data, so a handful of nodes by memory, and 500k reads/s is similar by throughput. The number that matters is the database’s: at a 95% hit ratio it takes 25k reads/s, and if we lose a node or the ratio slips five points, that doubles. So I’ll design the miss path as carefully as the hit path.”

The cache is a library in every app server talking to a cluster of nodes. The interface is a plain key-value contract; the real design is in the client.

pseudocode: the contract
get(key) -> value | MISS
get_many(keys) -> { key: value } -- fan out per shard, in parallel
set(key, value, ttl_s) -- last write wins
delete(key) -- the invalidation primitive
add(key, value, ttl_s) -> stored? -- only if absent: locks, leases
Entry on a node:
key bytes, ≤ 250 B (memcached's limit)
value bytes, ≤ 1 MB by default
expires absolute time; 0 = no TTL
meta LRU clock or LFU counter, a few bytes

Two decisions hide in that sketch:

  • Routing lives in the client. Every app server maps key → node locally: memcached clients use a consistent-hash ring, Redis Cluster uses CRC16(key) mod 16384 hash slots plus a slot table that nodes hand out (and correct with MOVED redirects). No proxy hop on the hot path, at the price of pushing topology changes to every client.
  • Key design is invalidation design. product:981 is easy to delete. A cached page that embeds 40 products is not: when product 981 changes, which page keys do you delete? Prefer caching raw objects by id and assembling them, or include a version in derived keys (page:home:v{catalog_version}) so a version bump makes the old entries unreachable and eviction cleans them up.

Step through the flows. On a hit, only one box lights up. Most of the design is about the other four flows.

Distributed cache: read, write and failure paths
step through
flow
Most reads end here: one hash in the client, one round trip, no database.
binlog / CDCDELETEApp server (cache client · ×N): workerApp servercache client · ×NCache shard 1 (primary + replica): cacheCache shard 1primary + replicaCache shard 2 (primary + replica): cacheCache shard 2primary + replicaCache shard 3 (primary + replica): cacheCache shard 3primary + replicaDatabase (source of truth): storeDatabasesource of truthInvalidator (tails commit log): queueInvalidatortails commit log
keyproduct:981
hash slotCRC16 mod 16384 = 9,402
The cache client hashes the key to find its shard. No central lookup: every app server holds the same slot map (or hash ring), so routing is a local computation.
1 / 3
Component Role Why it’s there
Cache client (in each app server) Hashes keys to shards, pools connections, batches get_many, applies timeouts and leases. Routing without a network hop; the policy (what to do on a miss or a timeout) lives with the caller.
Cache shards Each owns a slice of the key space in RAM, evicts by LRU/LFU, expires by TTL. Memory and throughput scale by adding shards.
Replica per shard Asynchronous copy, promoted on failure. Losing a node costs seconds of misses on one slice, not a cold slice for an hour.
Database Source of truth. The cache can lose anything; the database cannot.
Invalidator Tails the database’s commit log and deletes affected keys. Catches every write, including ones that bypassed the app’s delete.

Deep dive: what to keep when memory is full

Section titled “Deep dive: what to keep when memory is full”

A cache’s value is its hit ratio, and the eviction policy sets it for a given amount of memory. Real traffic is roughly Zipf-distributed: the r-th most popular key gets traffic proportional to 1 / r^s. Below, five policies replay the same trace. Try each workload, then slide the sample count.

Eviction policies on a skewed workload: hit ratio vs cache size
simulator
Popularity never changes. Recency and frequency agree, so LRU and LFU land close together.
0%25%50%75%100%1%2%5%10%15%20%30%40%cache size, % of distinct keys
at 10% of keys
OPT (offline ceiling)72.2%
DB reads per 1k requests: 278 (0.59× LRU's)
LRU (exact)53.0%
DB reads per 1k requests: 470
Sampled LRU (k=5)52.4%
DB reads per 1k requests: 476 (1.01× LRU's)
LFU61.4%
DB reads per 1k requests: 386 (0.82× LRU's)
Random46.4%
DB reads per 1k requests: 536 (1.14× LRU's)
LRU misses: 47.0%LFU − LRU: 8.4 ptsgap to OPT: 10.8 pts
Watch the DB reads line, not the hit ratio: going from 90% to 95% hits halves the load on the database. Set k = 1 and sampled LRU is random eviction; by k = 5 (Redis's default) it is within a point or two of exact LRU, without keeping a linked list. OPT evicts the key needed furthest in the future; no real cache can, but it shows how much headroom is left.

What the simulator shows:

Policy Keeps Cost per access Weakness
LRU Recently used keys O(1): a hash map plus a doubly linked list (see Caching) A one-off scan flushes the hot set; two pointers per entry
Sampled LRU (Redis) Roughly the LRU set O(k) at eviction: sample k keys, evict the oldest; a 24-bit clock per key Slightly worse than exact LRU at k = 5; indistinguishable at k = 10
LFU Frequently used keys Redis: an 8-bit logarithmic (Morris) counter per key, also sampled Clings to formerly popular keys unless counts decay (Redis: lfu-decay-time)
TTL only Whatever hasn’t expired Lazy check on read + background sampling Not an eviction policy: memory still fills before TTLs run out
Random Anything O(1) A few points worse than LRU on skewed traffic; fine when access is uniform

Practical notes:

  • Redis approximates. It never keeps a global LRU list. With maxmemory-policy allkeys-lru it samples maxmemory-samples keys (default 5) into a small pool of eviction candidates and evicts the idlest. allkeys-lfu uses the counter instead; volatile-* variants only consider keys that have a TTL.
  • Memcached keeps an LRU per slab class (values are bucketed by size), segmented into hot, warm and cold queues since 1.5, so one burst of new items can’t push out everything that was used twice.
  • TTL expiry is lazy. An expired key is removed when someone reads it, or when a background task samples keys with TTLs. Expired-but-unread keys can occupy memory until then; that’s why you still need an eviction policy. The TTL and expiry page covers the mechanics.
  • Admission matters too. Policies like TinyLFU (used in Caffeine, a Java in-process cache) refuse to admit a new key unless it’s more popular than the one it would evict. That’s the scan fix without LFU’s staleness.

A popular key expires. In the next few milliseconds, every request for it misses, every one queries the database, and the slow query gets slower as they pile up. That’s a cache stampede (a thundering herd). Run each strategy below; raise the read rate and the query time.

Cache stampede: one hot key expires
simulator
Every request that finds the key missing queries the database itself, then writes the result back. TTL = 15 ticks.
cache entry
reads
DB queries running
DB comfortable: 4
DB queries: 0peak concurrent: 0reads that waited: 0stale reads served: 0
hitread that queried the DBread that waited for someone elsestale value served
Every read in the gap between expiry and the first write-back goes to the database: roughly reads/tick × query time queries for one key. Turn on the slowing database and the gap widens, so even more pile in. That feedback loop is the stampede.
t = 0 / 48

The fixes, cheapest first:

Fix Database queries per expiry Who waits Stale reads
Nothing ≈ reads/s × query time everyone in the gap none
Single flight per process ≤ number of app servers callers in each process none
Lease / lock in the cache 1 everyone else, briefly none
Probabilistic early refresh (XFetch) usually a few, no coordination the refresher none
Serve stale while revalidating 1 nobody during the refresh

The miss path as a building block. add is “set only if absent”, so exactly one caller wins the lease:

pseudocode: a miss with a lease and a jittered TTL
read(key):
v <- cache.get(key)
if v is a value: return v
if cache.add("lease:" + key, my_id, ttl = 2 s): -- I won: I reload
v <- db.load(key)
cache.set(key, v, ttl = base_ttl * random(0.9, 1.1))
cache.delete("lease:" + key)
return v
sleep(5-20 ms); retry a few times -- someone else is reloading
on giving up: read the database directly, or serve a stale copy if one exists

And the early-refresh rule from Vattani, Chierichetti and Lowenstein (VLDB 2015), where delta is how long a recompute takes and beta ≥ 1 refreshes earlier:

pseudocode: probabilistic early expiration (XFetch)
on every hit of (value, delta, expiry):
if now - delta * beta * ln(random(0, 1)) >= expiry: -- ln of (0,1) is negative
recompute and set now, before anyone sees a miss

Where the cache sits in the write path is a choice with a cost:

Strategy Write does Read does Good for Risk
Cache-aside (lazy loading) write DB, delete key miss → load DB → set General default; cache holds only what’s read First read after a write is slow; the race below
Read-through write DB cache loads from DB itself Same as cache-aside, logic moved into the cache tier Cache needs DB access and schema knowledge
Write-through write DB and cache together always hits (if it fits) Data read right after it’s written Caches data nobody reads; two writes per update
Write-behind (write-back) write cache, flush DB later hits Absorbing write bursts, counters Data loss if the cache dies before flushing

Cache-aside has a classic race even when writers delete instead of set:

Time Reader (miss) Writer
1 get → miss
2 reads old row from DB
3 updates row to new value
4 delete(key): nothing there to delete
5 set(key, old)
6 old value cached until its TTL

Three ways to close it, in increasing strength:

  1. A TTL on everything. Staleness is bounded by the TTL, even if every other mechanism fails. Always do this.
  2. Leases that a delete invalidates. A miss hands out a lease token; a delete voids outstanding tokens; a set with a voided token is rejected. Step 5 above fails, which is how Facebook’s memcache closes the race. Without lease support, a “delayed second delete” (delete again ~1 s after the write) shrinks the window.
  3. Versioned values. Store the row’s version with the value and only set if the cached version is older (a compare-and-set). Needs a monotonic version from the database.

Missed deletes. The app crashes between “write DB” and “delete key”, or someone runs a SQL migration. Tailing the commit log (MySQL binlog, Postgres logical replication, Debezium) and deleting from there makes invalidation a property of the database, not of every code path that writes. See Event-driven systems.

Deep dive: partitioning, replication and regions

Section titled “Deep dive: partitioning, replication and regions”

Partitioning. hash(key) mod N remaps almost every key when N changes, which is a full cold cache. Two standard fixes:

  • A consistent-hash ring with virtual nodes (memcached clients, ketama): adding a node moves about 1/N of keys. The load balancing page has a ring you can add servers to.
  • Fixed hash slots (Redis Cluster): 16,384 slots, each assigned to a node. Resharding moves whole slots, and clients learn the new owner from MOVED replies. Keys sharing a {hash tag} land in the same slot, which multi-key commands require.

Replication. Each shard’s primary streams writes to a replica asynchronously. On failure the replica is promoted (Redis Cluster does this itself; Sentinel does it for non-cluster Redis). Writes acknowledged just before the crash can be lost. For a cache that’s only a few extra misses, which is the right trade: synchronous replication would double write latency for data you can reload.

Fan-out. A page that needs 100 keys spread over 50 shards waits for the slowest of 50 round trips. Batch per shard with get_many, cap fan-out with timeouts, and group related keys with hash tags when they’re always read together.

Regions. Each region runs its own cache cluster in front of its own database replica. Invalidations must reach every region: the commit-log tailer in each region deletes locally as replicated writes arrive, so a region never deletes before its database replica has the new row (otherwise a reader re-caches the old value).

What breaks What you see Design response
A shard dies Timeouts, then misses on 1/N of keys Short client timeout; replica promotion; a small gutter pool for keys of dead shards; throttle refills
Whole cache cold (restart, new cluster) DB at many times normal load Warm from a snapshot or from another cluster; ramp traffic gradually; request coalescing
Hot key expires DB spike, latency spike for one key Leases, early refresh, serve-stale
Hot key, no expiry One shard’s CPU or NIC at 100% In-process L1 cache with a short TTL; replicate the key under suffixes
Missed invalidation Stale data until TTL TTL on every key; commit-log tailer; versioned sets
Mass expiry Load spikes at regular intervals TTL jitter (±10%)
Large values p99 latency, network saturation Compress, split, or cache a reference instead
Memory fragmentation Evictions while “free” memory exists Slab allocators (memcached), active defrag (Redis)
Operation Time Space
get / set / delete on a node O(1) average hash lookup + O(1) eviction bookkeeping; ≈0.2-0.5 ms with the network value + ≈50-60 B per entry
Eviction (sampled LRU/LFU) O(k) per eviction, k ≈ 5-10 a few bytes of clock/counter per entry
Routing in the client ring: O(log V) binary search over V virtual nodes; slots: O(1) table lookup ring or 16,384-entry slot table per client
get_many over S shards S parallel round trips; latency = slowest shard —
Replication async, ≈0 added write latency 2× memory
One node ≈100k-1M ops/s (Redis is mostly single-threaded per shard; memcached is multi-threaded) 16-512 GB RAM

A good 45-minute arc:

  1. Clarify (5 min). What’s cached and how big; read:write ratio; how stale is acceptable; can the DB survive a cold cache.
  2. One node (5 min). Hash map + LRU, TTL checked lazily. Name the O(1) structure and move on; it’s the concept page.
  3. Distribute it (10 min). Sizing from the estimate, client-side routing with consistent hashing or slots, replicas. Why not mod N.
  4. Read and write paths (10 min). Cache-aside, delete-on-write, the race, TTL as the backstop, commit-log invalidation.
  5. Harden (10 min). Stampede (lease or serve-stale), hot keys (L1 + replication), node failure (gutter, throttled warm-up), multi-region invalidation.

Follow-ups to expect: “Why delete instead of update on write?” Concurrent updates can land out of order; a delete can’t leave the wrong value behind. “How do you know the cache is working?” Hit ratio per key prefix, eviction rate, DB queries per request, p99 per shard. “Redis or memcached?” Memcached for a plain multi-threaded blob cache; Redis when you need data structures, replication and failover built in. “How do you test it?” Inject failures (kill a shard under load), replay production traces for hit ratio, and run the reader/writer race in a test with a controllable clock.

Check yourself
pick one
Your cache's hit ratio drops from 98% to 96% after a deploy. Reads are steady at 1M/s. What happened to database load?
Check yourself
pick one
On a write, a service does UPDATE in the database, then SET key = new value in the cache. Two writers update the same row at nearly the same time. What can go wrong, and what's the usual fix?
Check yourself
pick one
One key (a global config blob) gets 300k reads/s, more than a single shard can serve. You double the number of shards. What happens?

Concept pages to drill: Caching · TTL and expiry · Hash map · Linked list · Load balancing · Distributed thinking · Concurrency