Distributed Caches

It is the innings break of a cricket final, and in ten minutes lakhs of people open the same food delivery app. Almost all of them load the same few things: their city's home screen and the menus of the most popular restaurants nearby. Those menus last changed in the afternoon, yet the database fetches each one again for every customer, thousands of times a second. Queries queue, a page that took 200 ms takes 4 seconds, and the order button spins. Nothing is broken. The database is answering the same question over and over, and a cache is what stops it.

A cache is a small, fast store that keeps copies of data that is slow or expensive to fetch, so the next request can skip that work. A distributed cache spreads that store over several machines and shares it with every application server, so a menu loaded by one server is ready for all the others. This chapter follows that food delivery app through the innings break: its menus, its city home screens and its carts.

The gap a cache exploits is large. Here is roughly what one read costs from each place the app could get a menu:

The gaps between the rows hold on any hardware. Speed is only half the gain: every read the cache answers is one the database never sees, leaving its connections and CPU for the writes only it can do. So how many reads must the cache answer?

Think of the cache as a filter in front of the database: what matters is how much gets through. The hit ratio is hits ÷ all reads, and the misses are what the database still has to serve.

At the innings break the app makes about 500,000 menu and home-screen reads a second. At a 95% hit ratio, 25,000 of them reach the database; at 99%, only 5,000. Going from 95% to 99% cuts database reads to a fifth, which is why teams chase the last few points. Latency moves too: if a hit takes 1 ms and a miss about 30 ms, the average read at 99% is 0.99 × 1 + 0.01 × 30 ≈ 1.3 ms.

The hit ratio depends on what you put in the cache. Good candidates are read far more often than written, expensive to fetch or compute, and fine if a few seconds stale. Data that many users share gives the biggest win, because one copy serves them all: a restaurant's menu, a city's home screen, the result of "top 20 restaurants near me".

Some data looks cacheable but is not, because a stale or wasted copy costs more than the read it saves:

Next comes where that copy should live, because there is more than one layer to put it in.

A menu can be cached at several points between the customer's phone and the database disk, and each layer trades speed against how widely the copy is shared. From the user inward:

This chapter is about the shared layer in the middle. An in-process cache is faster, but with 20 app servers you have 20 copies that fill separately, disagree after an edit and vanish on every deploy. A distributed cache costs one network hop, yet every server sees the same entry, and capacity grows with each node you add.

With the layer chosen, the next question is how data moves in and out of it. In cache-aside, also called lazy loading, the application talks to both the cache and the database. On a read it asks the cache first. On a miss it reads the database, stores the result in the cache with a TTL (time to live, after which the entry expires) and returns it.

def get_menu(restaurant_id):
    key = f'menu:{restaurant_id}'
    menu = cache.get(key)
    if menu is not None:
        return menu                       # hit
    menu = db.load_menu(restaurant_id)    # miss: ask the source of truth
    cache.set(key, menu, ttl=300)         # keep it for 5 minutes
    return menu

def update_menu(restaurant_id, menu):
    db.save_menu(restaurant_id, menu)
    cache.delete(f'menu:{restaurant_id}')  # the next read reloads it

The cache holds only what someone asked for, and a write simply removes the old copy.

Follow three reads of menu 42 through two different servers, with an edit in between.

The second read hit because the cache is shared; with a cache inside each server, server 2 would have missed too. The write deleted the entry instead of overwriting it, so the next read loaded whatever the database held at that moment. Delete on write, reload on read is simpler and safer than keeping the cached copy in step by hand, though it still leaves the races covered later in this chapter.

Cache-aside keeps the caching logic in your code. Other patterns move some of it into the cache layer, usually a library or caching service that can reach the database. Each is defined by what happens on a write:

• Read-through: the app talks only to the cache, which loads a missing value from the database itself.
• Write-through: each write goes through the cache layer, which writes the database and stores the new value before confirming.
• Write-behind (or write-back): the cache confirms the write at once and saves it to the database later, often in batches.
• Write-around: writes go straight to the database and skip the cache, which fills only on reads, as in cache-aside.

Each pattern also has a failure it is known for, and that failure decides where it belongs:

Real systems mix them: cache-aside for reads, write-around for most writes, write-through for the few keys read immediately after a change, and write-behind only for counters. Write-through needs one rule: write the database first, and if the cache update then fails, delete the key rather than keep serving the old value.

Patterns decide what goes into the cache; memory decides how much stays. A cache node might have 64 GB of RAM in front of a 2 TB database, so it can never hold everything. When it is full and a new entry arrives, it must evict, that is, remove an existing entry, and the eviction policy decides which. Each policy bets on a different sign that an entry will be read again:

LRU is the usual choice, because recent use predicts the next read well. A textbook LRU pairs a hash map with a list ordered by last use; large caches approximate it by evicting the oldest of a few sampled keys, which saves the list's memory.

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity):
        self.capacity = capacity
        self.items = OrderedDict()          # oldest first, newest last

    def get(self, key):
        if key not in self.items:
            return None
        self.items.move_to_end(key)         # mark as most recently used
        return self.items[key]

    def put(self, key, value):
        self.items[key] = value
        self.items.move_to_end(key)
        if len(self.items) > self.capacity:
            self.items.popitem(last=False)  # evict the least recently used

The ordered dictionary is the hash map and the list in one, so get and put run in constant time.

Eviction handles a full node; the next question is how many nodes you need. Size a cache three ways, by memory, by operations a second and by network, and take the largest answer. For the food delivery app at the innings break:

memory   3 lakh menus × 5 KB                ≈ 1.5 GB
         50 lakh carts and sessions × 1 KB  ≈ 5 GB
         + key overhead and headroom        ≈ 10 GB   → 1 node
reads    500,000 a second ÷ ~100,000 per node = 5 nodes flat out
         keep each about 60% busy                    → 8 nodes
network  500,000 × 4 KB average ≈ 2 GB/s ≈ 16 Gbps
         ÷ 8 nodes ≈ 2 Gbps each, on 10 Gbps cards   → fine

Memory says one node; reads say eight. Add a replica to each for failover: 16 in all.

Spreading keys over eight nodes is sharding: each node owns part of the key space, and every app server must agree on which. The naive rule, node = hash(key) % N, spreads keys evenly until N changes. Going from 8 nodes to 9 moves about 89% of keys, so most reads miss at once and the database takes the hit just as you add capacity.

Consistent hashing, from the key-value store chapter, places nodes and keys on a ring so a new node takes over only its own slice: about one key in nine. Some clusters instead split the key space into a fixed number of hash slots, 16,384 in Redis Cluster, and move whole slots to a new node, with the same effect. The routing can live in a client library in each app server, in a proxy, or in the cluster itself, which redirects a request that reached the wrong node.

Sharding spreads keys evenly. It does not spread traffic evenly, and the next page shows what happens when one key gets more than a node can serve.

Sharding spreads keys, not traffic. At the innings break every customer in Mumbai opens the app to the same home screen, so the key home:mumbai might get 100,000 reads a second, and all of them land on the one node that owns it. That node becomes the bottleneck while the other seven idle.

The limit is often the network before the CPU. The home screen is about 20 KB, so 100,000 reads a second is 2 GB a second, about 16 Gbps: more than a 10 Gbps network card can carry. A hot key is a traffic problem that adding nodes cannot fix, because the key still lives on one of them.

There are three fixes, from cheapest to most involved:

• A short local cache: each app server keeps home:mumbai in its own memory for 1 second. With 100 app servers, the cache node sees at most about 100 reads a second instead of 100,000. The price is up to a second of staleness.
• Key copies: store the value under several names, home:mumbai#1 to #8, placed on different nodes, and let each reader pick one at random, about 12,500 reads a second each. Every write must update or delete all eight.
• Read replicas: give that shard two replicas and spread its reads over all three nodes. They also take over if the primary dies, but they lag slightly behind it.

Hot keys are usually predictable: a city's home screen at dinner time, a sale banner, a viral restaurant. Many teams sample key access counts on the cache nodes and cache only the top few keys locally, so everything else stays fresh.

So far the cache has been fast; the harder question is whether it tells the truth. Data is stale when the cache holds a value the database has since changed, such as a menu still offering a dish marked sold out. Staleness cannot be removed without giving up speed, so ask how long it lasts and who would notice. The common approaches differ in exactly that:

Versioned keys put the version in the name, like menu:42:v7, so readers that learn the current version from the database never read an old entry. Delete on write, the usual default, still leaves one race, which needs a slow reader and a writer to interleave:

This race needs an unlucky pause between the reader's database read and its cache write, so it is rare. The next page shows a version of it that needs no bad luck at all.

Many apps send reads to database replicas, as in the databases chapter, and a cache fill is a read. Replicas apply the primary's changes a little late: usually milliseconds, sometimes seconds at the dinner peak. A fill that reads a replica inside that gap caches the old menu with a fresh TTL. It needs no unlucky pause, only a busy replica, which makes it a common stale-cache bug in production.

The delete was right; the fill read the wrong copy. Any scheme built only on deletes leaves a window as long as the replica lag, and a busy evening stretches that window. The defences on the next page each close it in a different way.

Three defences close the lag window, and a fourth limits the damage when the others miss. All of them also stop the slow-reader race from two pages back, because both races end the same way: an old value written after the delete.

The tombstone is the only defence that does not depend on timing, so its details matter. A plain delete leaves nothing to compare against, so a fill carrying version 1 would be stored; the tombstone remembers that version 2 exists, and readers treat it as a miss. It must outlive the worst replica lag, and eviction can still drop it under memory pressure, which is why the TTL stays as a backstop.

Change data capture removes a different bug: a code path that forgets to delete. A small service reads the database's change log and deletes the matching key after every committed change. It does not remove the lag, so pair it with a second delete or tombstones, and alert when lag nears your delay.

Stale data is a correctness problem; the next failure is a load problem. Every key expires eventually, including the hottest one. When home:mumbai expires at the innings break, every request for it misses until someone reloads it, and each one runs the same heavy query. This is a cache stampede, also called the thundering herd. The database is sized for the usual trickle of misses, so it slows, and more requests miss while the value is rebuilt.

Three servers sent three queries; in production thousands of requests send thousands, against one with coalescing. Coalescing is the main fix, and the others stop a herd from forming:

Waiters need a limit too: after a few hundred milliseconds, serve the stale copy or go to the database under a concurrency cap, never in a tight retry loop.

A stampede loses one key; sometimes you lose all of them. A cache holds copies, so losing it loses no data, yet it can still take the site down. At the innings break the app makes 500,000 reads a second at a 99% hit ratio, so the database serves 5,000. If the whole cluster disappears, the database suddenly gets all 500,000, a hundred times its normal read load.

A cold start does the same thing more slowly. A fresh cache after a restart or a migration has a hit ratio of zero and fills only as misses reach the database, so the first minutes after bringing it up are the most dangerous. Each way of losing the cache needs its own defence:

The last row is called cache penetration: a bug or a scraper asking for restaurant ids that do not exist misses every time, because there is nothing to cache. A short-lived "not found" marker handles the common case, and a Bloom filter (a compact structure that can answer "definitely not present") stops the rest before they reach the database.

Almost every read-heavy design gets a cache, and a strong answer says what is cached, how it stays fresh and what happens when it fails, each with a number.