10.14.2 — Distributed Caching at Scale

One process caching in its own memory is easy. The interesting failures start when there are forty servers, a shared cache cluster, and one product that everybody wants at the same second.

Three of those failures cause most real incidents: a key so popular that one cache node cannot serve it, a crowd of requests all missing at the same instant, and a cache whose recovery destroys the database it was protecting. This page is about those, and about the choices that decide how badly they hit you.

1. Where the cache lives

Three options, and real systems use two of them together.

Local, in the application's own memory. Nanosecond reads, no network, no serialisation. And it multiplies by the number of servers: forty servers means forty caches, each with its own copy and its own expiry timer, so two users hitting different servers can see different values. Memory is also limited to whatever is spare in the application process.

Shared, a separate cache cluster. One copy of the truth, so every server agrees. Memory scales independently of the application. The cost is a network hop — under a millisecond inside a data centre, which is thirty thousand times slower than local memory but still two hundred times faster than the database query it replaces. It is also a new thing that can be down.

Both, and this is what large systems actually do. A small local cache in front of the shared one absorbs repeated reads within a single burst, and the shared cache absorbs everything else.

Figure 1 — Two layers, on purpose. The local layer's job is not capacity, it is to stop forty servers all asking the shared cache for the same hot key in the same millisecond.

The cost of the local layer is stated in one sentence and must be said out loud: two servers can disagree for the length of the local TTL. Two seconds of disagreement is fine for a product description, an article, or a follower count. It is not fine for a price or a permission, so those skip the local layer entirely. Choosing per kind of data, rather than turning the local layer on globally, is the whole skill.

2. Spreading keys across nodes

A cache cluster holds more than one node, so something has to decide which node holds which key.

The obvious answer is wrong. node = hash(key) % nodeCount works perfectly until the node count changes. Go from ten nodes to eleven and almost every key maps somewhere new, so the entire cache misses at once — which is section 5's disaster, self-inflicted by a routine capacity change.

Consistent hashing is the fix, and 10.6 builds it properly. The one-line version: place nodes and keys on a ring, and a key belongs to the next node clockwise. Adding an eleventh node moves only the keys between it and its predecessor — roughly one eleventh — and leaves the rest untouched. Removing a node moves only that node's keys to its neighbour.

The practical addition is virtual nodes: each physical node appears at many points on the ring, so the load spreads evenly and a departing node's keys are shared among all the survivors rather than dumped on one unlucky neighbour.

Should cache nodes be replicated? Usually no, and the reasoning is worth having ready. A cache is not the source of truth, so losing a node loses a bit of hit rate rather than any data, and replication doubles the memory cost for a resource whose whole point is being cheap. Replicate only when losing one node's share of the hit rate would take down the database anyway — which is a statement about how thin your database headroom is, and is often better fixed by giving the database headroom.

3. Hot keys

A hot key is one that is requested far more than the others. Every request for it lands on the same cache node, and if that key alone attracts fifty thousand requests per second, one node's network card becomes the bottleneck for the entire system.

This is not exotic. It happens whenever a single item is on the homepage, whenever a celebrity posts, whenever a flash sale starts, and whenever a configuration blob is read on every request.

How to detect it. You cannot count every key's requests without spending as much CPU on counting as on serving. What you do instead is sample: record a small fraction of requests and look at the distribution, or keep an approximate frequency sketch, which is what a Count-Min Sketch is for (10.18). Most managed caches expose a hot-key report built exactly this way.

Three fixes, in increasing order of effort.

Put it in the local cache. A one-second local TTL on a key requested fifty thousand times per second means each server asks the shared cache once per second instead of thousands of times. Forty servers, forty requests per second, problem gone. This is the cheapest fix by a wide margin and it is why the local layer exists.

Split the key across nodes. Store the same value under several keys — config:v3:0 through config:v3:9 — and have each reader pick one at random. The load spreads across ten nodes instead of one. The cost is ten copies of the value and ten things to invalidate, so this suits a small, rarely-changing value.

Give it its own node. For a genuinely enormous key, dedicate hardware so its traffic cannot affect anything else. This is a bulkhead (10.9) applied to cache capacity.

4. The stampede

A hot key expires. In the microsecond after it expires, five thousand in-flight requests all miss, and all five thousand go to the database asking the identical question.

The cache made things worse than having no cache at that instant, because without one the load would at least have been spread evenly through time. This is called a cache stampede, or a thundering herd.

Figure 2 — Single-flight. The first caller to miss publishes a promise; every later caller awaits it. One database query serves five thousand requests, and nobody waits longer than the first one did.

Fix one, and the most important: single-flight. Only the first caller to miss on a key actually loads it; everyone else waits on that load's result. Inside one process this is a map of in-flight promises, built line by line in 9.5.5:

const pending = inFlight.get(key);
if (pending) return pending;                    // (1) join the load already running

const promise = loadFromDatabase(key)
  .then(async value => { await cache.set(key, value, { ttl }); return value; })
  .finally(() => inFlight.delete(key));         // (2) always clean up, even on failure

inFlight.set(key, promise);                     // (3) publish before awaiting
return promise;

(1) A caller arriving during a load joins it rather than starting another. (2) The finally matters as much as it does for a lock: a rejected promise left in the map would poison the key permanently, handing the same old error to every future caller. (3) Publishing before returning is what makes step (1) work at all.

That solves it within one process. With forty servers you still get forty queries instead of five thousand, which is usually fine. If it is not — because the query costs four hundred milliseconds and forty of them concurrently is still too many — the cross-server version is a short-lived lock in the shared cache: the first server to set a lock:<key> entry with a few seconds' expiry does the load, and the others either wait briefly and re-read, or serve the stale value.

Fix two: serve stale while refreshing. Keep two lifetimes on each entry — a soft one after which the value is considered due for refresh, and a hard one after which it may no longer be served. Between the two, a reader gets the old value immediately and a background refresh is kicked off. Nobody ever waits for a rebuild, and the database sees one refresh rather than a crowd. This is the single most effective pattern for expensive, hot, slightly-stale-tolerant data, and it also means a database outage degrades into serving slightly old data rather than into an error page.

Fix three: jitter the expiry. If a thousand keys were all loaded during a deploy, they all expire in the same millisecond, and single-flight does not help because the keys are different. Setting each TTL to a random value within, say, ten percent of the target spreads the renewals across a window and turns a spike into a hum. This is the same jitter that breaks a retry livelock (9.5.3), applied to expiry.

Fix four, for the mathematically inclined: probabilistic early expiry. Each reader independently decides, with a probability that rises as the entry approaches expiry, to refresh it early. Most readers do nothing; one lucky reader refreshes just before the deadline; nobody ever sees a miss. It needs no locks and no coordination, which makes it attractive across many servers.

5. Requests for things that do not exist

A related failure has a different cause. Somebody requests product:99999999, which does not exist. The cache misses. The database returns nothing. Nothing is cached, because there is nothing to cache. The next identical request repeats the whole thing.

Now imagine that request arriving ten thousand times a second, either from a bug in a client or from somebody probing your system deliberately. The cache provides zero protection, because a miss that finds nothing never populates anything. Every single request reaches the database.

Fix one: cache the absence. Store a small marker meaning "this does not exist" with a short lifetime.

const cached = await cache.get(key);
if (cached === MISSING) return null;                    // (1) a cached "no"
if (cached) return cached;

const row = await db.findById(id);
await cache.set(key, row ?? MISSING, { ttl: row ? 300 : 30 });   // (2) shorter TTL for absence
return row;

(1) A distinct marker, not null or undefined, so "we know it does not exist" is distinguishable from "we have not looked". (2) The absence gets a much shorter lifetime than a real value, because a record that gets created should become visible quickly.

The risk to name: an attacker requesting millions of random non-existent ids will fill your cache with negative entries and evict everything useful. So negative entries want a short TTL and, ideally, their own memory budget.

Fix two: a Bloom filter in front. A compact structure holding every id that exists, which answers "definitely not present" with certainty and "probably present" with a small error rate. A request for an id the filter rejects never touches the cache or the database at all. This is 10.18's subject, and it is the standard defence when the id space is large and mostly empty.

6. Keeping the cache and the database in step

10.14.1 established the rule: delete the key on write, never update it. Across a cluster there are two extra problems.

The reader-writer race is now longer. A reader misses, reads an old value from a replica that has not caught up (10.5), and writes that stale value into the shared cache after the writer's invalidation. The cache is now wrong for a full TTL. Two defences: read from the primary when you are going to populate a cache, or delete the key a second time a short delay after the write, which catches readers that were already in flight.

Every writer must remember, and across a company they will not. A data migration, an admin console, an analytics job backfilling a column — none of them know your cache exists. The durable fix is to stop asking writers to invalidate and instead drive invalidation from the database's own change stream (10.8.4): a consumer reads the change log and deletes the affected keys. Now every writer is covered automatically, including ones written next year by people who have never heard of your cache, and the invalidations arrive in the same order the writes did.

7. Sizing and the metrics that matter

Sizing starts with the working set, not with the total data. If you have 500 GB of products but 95 percent of requests touch 2 GB of them, you need slightly more than 2 GB, not 500 GB. Measure it: sample the keys requested over a representative hour, count the distinct ones, multiply by the average value size, and add headroom.

Then check the eviction rate, which is the number that tells you whether the size is right. A cache evicting constantly is too small for its working set, and the tell is a hit rate that is decent but not great alongside a high eviction count — the cache is thrashing, admitting new entries by throwing out ones it is about to need again.

Four metrics, and one is the alert.

Hit rate, and specifically alert on it falling, because it moves before latency does and gives you time to act (10.14.1 section 7 has the arithmetic showing why a small drop is a large load increase).

Eviction rate, which distinguishes "too small" from "badly used".

Memory used against memory available, because the failure at the limit depends entirely on configuration — a cache that refuses writes when full behaves very differently from one that evicts, and you should know which one you have.

Latency percentiles for cache operations themselves. A cache with a 40 ms p99 is not helping anybody, and the usual cause is one node with a hot key, which is section 3.

8. What the interviewer will push on

"You have one product on the homepage getting fifty thousand requests per second. What breaks?" They want you to notice that consistent hashing puts every request for that key on one node, so the cluster's total capacity is irrelevant — one node's network card is the ceiling. Then the fixes: a local cache in front is the cheap one, key splitting is the next, and a dedicated node is the last resort.

"The key expires and five thousand requests miss at once." Single-flight, and the giveaway that you have implemented it is mentioning the finally that removes the in-flight entry even on failure. Strong answers add serve-stale-while-refreshing as the better structural fix, because it means nobody ever waits for a rebuild at all.

"How would you handle requests for ids that do not exist?" They are checking whether you have noticed that a normal cache offers no protection here, since a miss with no result populates nothing. Cache the absence with a short TTL, and mention a Bloom filter for a large, mostly-empty id space.

"What happens when you add a node to the cache cluster?" The trap is modulo hashing, where adding one node remaps nearly every key and empties the cache in one step. Consistent hashing with virtual nodes moves only a fraction, and saying "virtual nodes" unprompted signals you know why the naive ring is unbalanced.

"Would you replicate your cache?" Usually no — it is not the source of truth, losing a node costs hit rate rather than data, and replication doubles the cost of the thing you chose for being cheap. Reverse the answer only when losing one node's share would take the database down, and then note that the real problem is database headroom.

"How do you invalidate when another team's service writes to the same table?" The question behind the question is whether you know that application-level invalidation only covers the writers who remember. The answer is to drive invalidation from the database's change stream, so coverage is automatic and ordered.

The thing to volunteer that nobody asks for: state which data skips the local cache layer and why. Saying "descriptions and review aggregates get a two-second local cache, prices and permissions do not, because two servers disagreeing for two seconds is fine for one and a legal problem for the other" shows the decision was made per kind of data rather than switched on globally.

Next: 10.14.3 moves the cache out of your data centre and next to the user, where the rules change again because you no longer control the machine holding your data.

Recall

• Three places a cache lives: local (nanoseconds, N copies, they disagree), shared (one truth, a network hop, a new dependency), or both — the local layer exists to stop forty servers asking for the same hot key at once.
• Local caching is a per-data-kind decision. Two seconds of disagreement is fine for a description, not for a price or a permission.
• Never route keys with hash % nodeCount — adding a node remaps almost everything and empties the cache. Consistent hashing with virtual nodes moves only a fraction and spreads the departing node's keys across all survivors.
• Do not replicate a cache by default: it is not the source of truth, so a lost node costs hit rate, not data.
• Hot key = one node's network card is the ceiling regardless of cluster size. Fixes: local cache in front · split the value across N keys · dedicate a node.
• Stampede = many requests miss the same key at the same instant. Fixes: single-flight (clean up the in-flight entry in finally) · serve stale while refreshing in the background · jitter the TTL so a thousand keys do not expire together · probabilistic early expiry.
• A miss that finds nothing caches nothing, so non-existent ids reach the database every time. Cache the absence with a short TTL, and use a Bloom filter for a large, mostly-empty id space.
• Invalidate from the database's change stream, not from application code, so writers who have never heard of your cache are covered automatically and in order.
• Size from the working set, not the data set. Watch hit rate (alert on it falling), eviction rate, memory, and cache-operation latency percentiles.

Self-test: Why does a hot key defeat a large cluster? Give three fixes. What does single-flight's finally prevent? Why does a normal cache give no protection against requests for non-existent ids? What breaks when you add a node under modulo hashing? Why drive invalidation from a change stream rather than from your write path?

Quiz Bank

FoundationalExplain the cache stampede and the four ways to prevent it.

The failure. A popular key expires. In the instant after expiry, every in-flight request for it misses simultaneously, and all of them go to the database asking the same question. Five thousand identical queries arrive at once for a value that a single query would have produced.

What makes it worse than having no cache at all, at that moment, is the synchronisation. Without a cache, those five thousand requests would have been spread through time. The cache gathered them up and released them together.

Fix one, single-flight. Only the first caller to miss performs the load; every other caller waits on that same result. Within one process it is a map from key to in-flight promise: check the map, and if something is there, return it rather than starting your own load. The critical detail is removing the entry in a finally rather than on success — a rejected promise left in the map would be handed to every future caller forever, poisoning the key until a restart.

This reduces five thousand queries to one per server, so with forty servers you get forty. If forty concurrent copies of a four-hundred-millisecond query is still too much, the cross-server version is a short-lived lock held in the shared cache: whoever manages to set lock:<key> does the load and the rest wait briefly and re-read.

Fix two, serve stale while refreshing. Give each entry two deadlines: a soft one after which it should be refreshed, and a hard one after which it must not be served. Between them, readers get the old value instantly and a background refresh runs. Nobody ever waits for a rebuild. This is usually the best structural answer for expensive hot data, and it has a bonus property — when the database is struggling, the system degrades into serving slightly old data rather than into errors.

Fix three, jitter the TTL. Single-flight does nothing when a thousand different keys expire together, which happens whenever they were all loaded at the same moment, such as after a deploy. Randomising each TTL within about ten percent of the target spreads the renewals out and converts a spike into a steady hum.

Fix four, probabilistic early expiry. Each reader independently rolls a die whose odds increase as the entry nears expiry, and the winner refreshes early. Most readers do nothing, one refreshes just before the deadline, and no reader ever experiences a miss — with no locks and no coordination, which makes it well suited to many servers.

In practice you use several. Jitter always, because it costs one line. Single-flight for anything expensive. Serve-stale for anything hot and expensive where slight staleness is acceptable.

AppliedA single product is featured on the homepage and receives 50,000 requests per second. Your cache cluster has thirty nodes and is barely at ten percent CPU, yet latency for that product is terrible. Explain and fix.

The cluster size is irrelevant, and that is the whole point of the question. Consistent hashing maps a key to exactly one node. Every one of those fifty thousand requests per second lands on the same node. The other twenty-nine are idle, which is why the average CPU looks fine while one node is saturated.

What saturates first is usually not CPU but the network interface. If the cached product is 20 KB, fifty thousand requests per second is a gigabyte per second leaving one machine, which will exhaust a 10 Gbit link. Single-threaded cache servers can also become CPU-bound on one core while the box shows plenty of idle capacity, which is another way the averages mislead.

Fix one, and the cheapest by a wide margin: a local cache in front. Give the product a one-second lifetime in each application server's own memory. Forty application servers then ask the shared cache once per second each, so the hot node receives forty requests per second instead of fifty thousand. The cost is that two servers may show a value up to a second apart, which for a product description is invisible and for a price would not be acceptable — so this is a decision per kind of data, not a global switch.

Fix two, split the key. Store the same value under product:123:0 through product:123:9, and have each reader pick a suffix at random. Consistent hashing now spreads those ten keys across roughly ten different nodes, dividing the load by ten. The cost is ten copies to invalidate, so this suits values that change rarely.

Fix three, dedicate hardware. For an extreme key, give it its own cache node so its traffic cannot affect anything else. This is a bulkhead, and its value is isolation rather than throughput.

Fix four, shrink the value. A 20 KB payload where the page needs 2 KB is ten times the bandwidth for no benefit. Splitting the cached object into the fields actually used on the hot path often solves the problem outright, and it is worth checking before adding machinery.

And the detection point worth adding unprompted: you cannot count every key without spending as much effort counting as serving, so hot-key detection is done by sampling or with an approximate frequency sketch. Most managed caches expose exactly such a report, and knowing to look at it is the difference between diagnosing this in ten minutes and in a day.

InterviewWhy does adding a node to a cache cluster empty it, and what prevents that?

Because of how most people route keys the first time they write it. The obvious scheme is node = hash(key) % nodeCount. With ten nodes, hash(key) % 10 picks a node. Add an eleventh and every key is now hash(key) % 11, which for almost every key is a different node. The data is still sitting on the old nodes, but nobody looks there any more, so effectively the entire cache is empty.

The consequence is the disaster from 10.14.1: the hit rate goes to zero, the database receives the full unabsorbed load, and a routine capacity increase becomes an outage. The bitter part is that you added the node because you were near capacity, so the database was already busy.

Consistent hashing prevents it. Place both nodes and keys on a circle by hashing them, and assign each key to the first node clockwise from it. Adding a node inserts one new point on the circle, so only the keys between it and its predecessor move — about one eleventh of them when going from ten nodes to eleven. Every other key still maps where it did, so the hit rate barely dips.

Removing a node is the mirror image: its keys pass to the next node clockwise, and nothing else moves.

Virtual nodes are the necessary refinement, and mentioning them shows you understand why the plain ring is not enough. With one point per node, the ring segments are uneven, so one node can own three times the key space of another purely by luck. Worse, when a node leaves, its entire key range lands on a single neighbour, which may then fall over — a cascading failure caused by the recovery. Giving each physical node a hundred or more points on the ring makes the distribution even and means a departing node's keys are shared across all the survivors.

Two extras worth adding. First, the same technique is what partitions a database (10.6), so this is one idea with two uses. Second, even with consistent hashing you should add nodes one at a time rather than doubling the cluster, because each addition still costs some fraction of the hit rate and the database has to absorb it.

StaffDesign invalidation for a cache read by six services, where the underlying table is written by three services, a nightly batch job, and occasional manual admin edits.

Start by naming why the obvious answer fails. The standard advice is that whoever writes the data deletes the cache key. With three services, a batch job, and a human with database access, that means five separate places must each remember, correctly, forever, including code written next year by people who have never heard of this cache. The manual edit is the one that guarantees failure — nobody adds a cache-delete to an ad-hoc UPDATE typed during an incident.

So application-level invalidation is not a mechanism here, it is a hope. What you need is invalidation that is impossible to bypass, and that means putting it downstream of the writes rather than beside them.

The design: drive invalidation from the database's change stream.

The database already produces an ordered log of every committed change (10.8.4). A single small consumer reads that log and, for each changed row, deletes the corresponding cache keys. This gives four properties that the application-level approach cannot.

Complete coverage. Every writer is included automatically — the three services, the batch job, the human, and anything added in future — because they all go through the database by definition.

Correct ordering. Invalidations arrive in the same order the writes committed, which removes the class of bugs where two invalidations race each other.

One place to maintain. The mapping from "this row changed" to "these cache keys are now wrong" lives in one file rather than being duplicated five times and drifting.

It survives reorganisation. A service moving, splitting or being rewritten does not break invalidation, because invalidation never depended on that service's code.

The parts that need care.

The row-to-keys mapping is the real design work. A product row change may invalidate product:123, a category:electronics:top10 list, and a rendered search fragment. That mapping has to be written down explicitly, and it is the thing most likely to be incomplete. Keeping the cached shapes coarse and few makes this tractable; caching a hundred derived views of the same row makes it unmanageable, which is itself an argument for fewer cached shapes.

Lag is real. A change stream consumer is behind by tens or hundreds of milliseconds. For most data that is invisible. Where it is not — a price, say — the writing service should also delete the key directly, so the fast path is immediate and the change stream is the guarantee that nothing is ever missed. Belt and braces, with each mechanism doing what it is good at.

The consumer is now a dependency. If it stops, staleness grows silently. So it needs an alert on consumer lag, and — the important part — every entry keeps a TTL as the backstop, so a dead consumer means data is up to one TTL stale rather than stale forever. That is the same principle as everywhere else on this page: the TTL is what bounds the damage when the clever mechanism fails.

The batch job needs special handling. A nightly job touching two million rows will generate two million invalidations and effectively empty the cache at 3am, which then stampedes at 6am when traffic returns. Options: have the job write a version marker that invalidates a whole namespace at once instead of two million individual deletes, or throttle the consumer during the batch window, or schedule the batch so the cache has time to refill before peak. Spotting that the batch job is a different problem from the three services is the part that distinguishes a thorough answer.

The one-sentence summary to leave them with: stop asking writers to remember, and put invalidation downstream of the database where it cannot be bypassed — then keep a TTL anyway, because the mechanism that cannot be bypassed can still be down.

Flashcards

FlashHot key

One key, one node, regardless of cluster size — the node's network card is the ceiling. Fixes: local cache in front (cheapest), split the value across N keys, dedicate a node, shrink the value.

FlashSingle-flight

First caller to miss loads; everyone else awaits that promise. Publish before awaiting; remove it in finally or a rejection poisons the key forever.

FlashServe stale while refreshing

Two deadlines: soft (refresh due) and hard (must not serve). Between them, return the old value instantly and refresh in the background. Nobody waits for a rebuild; a database outage degrades to slightly old data.

FlashNon-existent ids

A miss that finds nothing caches nothing, so every request reaches the database. Cache the absence with a short TTL and its own budget; add a Bloom filter for a large, mostly-empty id space.

FlashModulo versus consistent hashing

hash % N remaps nearly every key when N changes, emptying the cache during a capacity increase. Consistent hashing moves ~1/N; virtual nodes make the spread even and share a departing node's keys.

FlashInvalidate from the change stream

Application-level deletion only covers writers who remember — and the admin doing a manual UPDATE never does. Drive it from the database's change log: complete, ordered, one place. Keep a TTL as the backstop.