30 — A distributed cache

Chapter 7 treated caching as a technique: why it works, cache-aside and write-through, eviction policies, invalidation, stampedes. This chapter is about the cache as a system that someone has to build and run: a cluster of memory servers, such as Redis Cluster or a fleet of Memcached nodes, that many application servers share. The questions change from “what should be cached?” to “which node holds this key, and what happens when that node dies?”

Requirements

Functional

• GET, SET with an optional TTL, and DELETE on keys up to a few hundred bytes and values up to about a megabyte.
• Atomic operations on a single key: increment, compare-and-set.
• Capacity and throughput grow by adding nodes.

Non-functional

• Sub-millisecond server-side latency, so that a request making dozens of cache calls stays fast.
• Available through node failures. A dead node must not take the whole cache down.
• Bounded memory. Each node evicts rather than crashing when full.
• Data loss is acceptable. It’s a cache: the database is the source of truth, so a lost entry costs a miss, not correctness. This one requirement shapes most of the design.

Estimates

The method is from chapter 4. The inputs are assumptions:

Each primary handles about 23,000 operations a second and 23 MB/s of value traffic. Memory decides the node count, not throughput. That’s typical, and it means the per-entry overhead assumption deserves checking, which the memory section does.

API and data model

The interface is a key-value protocol, not HTTP:

GET user:4821:profile                 -> value or nil
SET user:4821:profile <bytes> EX 300  -> OK
DEL user:4821:profile                 -> 1
INCR rate:4821:2026-11-09T10:05       -> 17

There’s no schema, but there is a data model: the key naming convention. Keys are namespaced (entity:id:field), versioned when the value format changes (v2:user:4821:profile), and kept short, because the key is stored with every entry. Values are serialized objects, usually compressed above a size threshold.

Topology

 app servers (client library holds the slot map)
     │  route by key
     v
 ┌──────────┐  ┌──────────┐  ┌──────────┐
 │ primary 1│  │ primary 2│  │ primary 3│   ... 44 primaries, each owning a range of slots
 │ replica 1│  │ replica 2│  │ replica 3│   async replication, one replica per primary
 └──────────┘  └──────────┘  └──────────┘
        \           |            /
         gossip: membership, failure detection, slot ownership

Three choices shape this picture: how keys map to nodes, how clients find the right node, and how nodes agree on who’s alive.

Placing keys

Plain hash(key) % N moves almost every key when $N$ changes, which for a cache means a mass miss. Chapter 11 showed this, and introduced consistent hashing and fixed slots as the two fixes. Here they’re compared as a cache operator sees them.

Consistent hashing with virtual nodes

Each node takes many points on a hash ring, called virtual nodes, and a key belongs to the next point clockwise. Two numbers matter for a cache: how uneven the load is, since the busiest node sets the capacity of the whole cluster, and where a failed node’s keys go, since that’s where the extra misses and load land.

The simulation below computes each node’s exact share of the ring for 30 random 10-node clusters. It then removes one node and measures how much of its share went to the single biggest “heir”.

import bisect
import hashlib
import math
import random
import statistics
from collections import Counter

RING = 2 ** 32


def h(s):
    return int.from_bytes(hashlib.md5(s.encode()).digest()[:4], "big")


def ownership(nodes, vnodes):
    # Exact share of the ring each node owns: a point owns the arc that ends at it.
    pts = sorted((h(f"{n}#{i}"), n) for n in nodes for i in range(vnodes))
    share = Counter()
    for j, (p, n) in enumerate(pts):
        prev = pts[j - 1][0] if j else pts[-1][0] - RING
        share[n] += (p - prev) / RING
    return pts, share


def failover_spread(nodes, vnodes, dead):
    # Where does a failed node's share go? Return the largest single receiver's portion.
    _, before = ownership(nodes, vnodes)
    _, after = ownership([n for n in nodes if n != dead], vnodes)
    gained = {n: after[n] - before[n] for n in after}
    return max(gained.values()) / before[dead], sum(g > 1e-12 for g in gained.values())


def part1():
    print("1) Load spread on a 10-node ring, 30 random clusters per row")
    print(f"{'vnodes':>6} {'max/mean p50':>12} {'max/mean worst':>14} {'biggest heir':>12} {'heirs':>5} {'ring points':>11}")
    for v in (1, 8, 32, 128, 512):
        ratios, heirs, heir_n = [], [], []
        for c in range(30):
            nodes = [f"c{c}-node{i}" for i in range(10)]
            _, share = ownership(nodes, v)
            ratios.append(max(share.values()) * 10)
            big, n = failover_spread(nodes, v, nodes[0])
            heirs.append(big)
            heir_n.append(n)
        print(f"{v:>6} {statistics.median(ratios):>12.2f} {max(ratios):>14.2f} "
              f"{statistics.median(heirs):>12.0%} {statistics.median(heir_n):>5.0f} {10 * v:>11,}")


def zipf_keys(n_keys, n_req, s, rng):
    w = [1 / (r + 1) ** s for r in range(n_keys)]
    return rng.choices(range(n_keys), weights=w, k=n_req)


def part2():
    print("\n2) Hot keys: 10 nodes x 128 vnodes, 1M keys, 2M Zipf(s=1.1) requests")
    rng = random.Random(3)
    nodes = [f"node{i}" for i in range(10)]
    pts, _ = ownership(nodes, 128)
    pos = [p for p, _ in pts]
    def owner(k):
        i = bisect.bisect_left(pos, h(f"key:{k}")) % len(pts)
        return pts[i][1]
    reqs = zipf_keys(1_000_000, 2_000_000, 1.1, rng)
    owner_of = {}
    def load(local_top=0, replicas_for_top=0, top=100):
        c = Counter()
        for k in reqs:
            if k < local_top:                 # served from an in-process cache on the app server
                continue
            o = owner_of.get(k) or owner_of.setdefault(k, owner(k))
            if k < top and replicas_for_top:  # hot key copied to several nodes, reads spread
                i = bisect.bisect_left(pos, h(f"key:{k}")) % len(pts)
                seen = []
                while len(seen) < replicas_for_top:
                    n = pts[i % len(pts)][1]
                    if n not in seen:
                        seen.append(n)
                    i += 1
                o = rng.choice(seen)
            c[o] += 1
        tot = sum(c.values())
        return max(c.values()) / (tot / 10), tot / len(reqs)
    print(f"   share of requests on the single hottest key: {reqs.count(0) / len(reqs):.1%}")
    for label, kw in [("plain ring", {}),
                      ("top 100 keys on 3 replicas each", {"replicas_for_top": 3}),
                      ("top 1,000 keys in a local L1", {"local_top": 1000})]:
        r, frac = load(**kw)
        print(f"   {label:34} busiest node {r:.2f}x mean, cluster sees {frac:.0%} of requests")


def part3():
    print("\n3) Failure detection on a healthy node: heartbeat every 1 s for 30 days")
    rng = random.Random(5)
    days = 30
    t, arrivals = 0.0, []
    while t < days * 86400:
        if rng.random() < 0.0002:            # a stop-the-world pause on the sender: 0.5-6 s
            t += rng.uniform(0.5, 6.0)
        if rng.random() < 0.0001:            # a short network blip drops 1-4 heartbeats
            t += rng.randint(1, 4)
        if rng.random() >= 0.001:            # 0.1% of heartbeats lost on their own
            arrivals.append(t + rng.lognormvariate(math.log(0.002), 1.0))
        t += 1.0
    arrivals.sort()
    gaps = [b - a for a, b in zip(arrivals, arrivals[1:])]
    print(f"{'timeout T (s)':>13} {'false suspicions/day':>21} {'detect a crash within (s)':>26}")
    for T in (1.5, 2.5, 3.5, 5.0, 8.0):
        false = sum(g > T for g in gaps) / days
        print(f"{T:>13} {false:>21.1f} {T:>26.1f}")


if __name__ == "__main__":
    part1()
    part2()
    part3()

Output, part 1:

1) Load spread on a 10-node ring, 30 random clusters per row
vnodes max/mean p50 max/mean worst biggest heir heirs ring points
     1         2.76           4.21         100%     1          10
     8         1.57           1.88          46%     5          80
    32         1.28           1.49          24%     9         320
   128         1.13           1.33          18%     9       1,280
   512         1.07           1.14          14%     9       5,120

• With one point per node, the median cluster’s busiest node had 2.76 times the average load, and the worst had 4.21. All of a failed node’s keys went to one neighbor, doubling its load at the worst moment.
• With 128 points, the busiest node carried 1.13 times the average in the median cluster, and a failed node’s keys spread across all nine survivors, with the largest single heir taking 18% of them in the median cluster.
• The cost is the routing table: 10 points per node becomes 1,280 points in total for 128. That’s trivial for one cluster, and it matters only when every client holds the table for very many nodes.

Because the busiest node defines capacity, an imbalance of 1.33 in the worst case means provisioning a third more than the average needs.

Fixed hash slots

Redis Cluster uses a different scheme. The cluster specification defines 16,384 hash slots, with HASH_SLOT = CRC16(key) mod 16384, using the XMODEM variant of CRC16. Each primary owns a subset of slots. Adding a node means moving some slots to it, and the key-to-slot mapping never changes.

Hash tags force related keys into one slot: if a key contains {...} with at least one character inside, only that part is hashed. Multi-key operations and Lua scripts in Redis Cluster require all their keys to be in one slot, as the rate limiter in chapter 22 noted, so hash tags are how they’re made possible.

def crc16_xmodem(data: bytes) -> int:
    crc = 0
    for byte in data:
        crc ^= byte << 8
        for _ in range(8):
            crc = ((crc << 1) ^ 0x1021) if crc & 0x8000 else (crc << 1)
            crc &= 0xFFFF
    return crc


def hash_slot(key: str) -> int:
    s = key.find("{")
    if s != -1:
        e = key.find("}", s + 1)
        if e != -1 and e != s + 1:          # a non-empty {tag}: hash only the tag
            key = key[s + 1:e]
    return crc16_xmodem(key.encode()) % 16384


print(f"CRC16('123456789') = {crc16_xmodem(b'123456789'):04X}")
for k in ("user:1000", "{user1000}.following", "{user1000}.followers", "foo{}{bar}", "foo{bar}{zap}"):
    print(f"{k:22} slot {hash_slot(k)}")

Output:

CRC16('123456789') = 31C3
user:1000              slot 1649
{user1000}.following   slot 3443
{user1000}.followers   slot 3443
foo{}{bar}             slot 8363
foo{bar}{zap}          slot 5061

The check value 31C3 matches the one given in the specification. The two {user1000} keys share a slot. foo{}{bar} is hashed whole, because its first braces are empty, and foo{bar}{zap} hashes only bar.

Hash tags are also a way to create a hot slot by accident. If every key of a large tenant carries {tenant42}, that tenant’s whole dataset sits on one node.

Fixed slots make operations explicit: an operator or a tool decides which slots move, and can move them one at a time.

Routing: client, proxy or node

Something has to turn a key into a node address.

Redis Cluster combines the first and third. Clients cache the slot map. If they send a key to the wrong node, the node replies -MOVED 3999 127.0.0.1:6381, meaning slot 3999 now lives at that address permanently, and the client should update its map. During a slot migration, a node replies -ASK, which means “try that node for this one query only”, because the slot is half moved and its keys are in two places.

A proxy adds a network hop, and so latency, and the proxy tier needs its own capacity and redundancy. In return, it hides topology changes from applications, pools connections (thousands of application processes each connecting to 88 nodes is a lot of connections), and is one place to add metrics or hot-key detection.

Replication

Each primary has one or more replicas, updated asynchronously. If the primary fails, a replica is promoted. Asynchronous replication means the primary acknowledges a write before the replica has it, so a failover can lose recent writes. The Redis Cluster specification is direct about this: it does not guarantee strong consistency, and “there is always a window of time when it is possible to lose writes during partitions”.

For a cache, that’s usually acceptable, because a lost entry is a miss. Two cases aren’t covered by that reasoning:

• A lost DELETE. If the invalidation for an updated row is lost in a failover, the old value comes back from the replica, and stays until its TTL. TTLs are the backstop for exactly this.
• Data that isn’t a cache. Counters, locks, rate limits and sessions stored in “the cache” can be lost or go backwards. The lock case is the one from chapter 13: a lock that a failover can lose isn’t a safe lock.

Replicas can also serve reads, which spreads load at the cost of slightly stale values.

Eviction and memory accounting

Chapter 7 covered which eviction policy to choose. A cluster operator also has to know what “full” means. Redis evicts when its data exceeds maxmemory. According to the eviction docs, the output buffers for replication and persistence aren’t counted toward that limit, so the docs recommend setting maxmemory below physical memory, leaving room for them. Other memory isn’t counted in the dataset either: allocator fragmentation, the process itself, client buffers and a copy-on-write snapshot can all add more. A node with 64 GB of RAM and maxmemory 64gb will eventually be killed by the operating system.

Redis approximates LRU by sampling: on eviction, it checks a few random keys (maxmemory-samples, default 5) and evicts the least recently used among them, with a pool of good candidates carried between rounds. The docs say a sample size of 10 comes very close to true LRU, at some CPU cost.

Per-entry overhead

Every entry costs more than its key and value: hash table slots, pointers, expiry metadata and allocator rounding. The size of that overhead decides how many nodes the estimate needs. Measuring it in Python shows the pattern clearly, even though Python’s own overheads are larger than a purpose-built server’s:

import tracemalloc
from collections import OrderedDict

for vsize in (16, 100, 1000, 10_000):
    tracemalloc.start()
    cache = OrderedDict()
    for i in range(100_000):
        cache[f"user:{i:08d}"] = b"x" * vsize       # 13-byte key, vsize-byte value
    used, _ = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    payload = 100_000 * (13 + vsize)
    print(f"value {vsize:>6} B: {used / 100_000:>8.0f} B per entry, payload {13 + vsize:>6} B, overhead {used / payload - 1:>6.0%}")
    del cache

Output:

value     16 B:      194 B per entry, payload     29 B, overhead   570%
value    100 B:      278 B per entry, payload    113 B, overhead   146%
value   1000 B:     1178 B per entry, payload   1013 B, overhead    16%
value  10000 B:    10178 B per entry, payload  10013 B, overhead     2%

The fixed cost per entry was about 165 bytes here, regardless of value size. For 16-byte values, that’s nearly six times the payload. For 10 KB values, it’s noise. The same holds in any cache server, with smaller constants. So caching many tiny values, such as one key per counter or flag, uses memory mostly on overhead. Grouping small fields into one hash per entity, or one serialized blob, reduces it. Redis’s MEMORY USAGE command reports the real figure for a given key, and the right way to fill in the estimate’s 100-byte assumption is to measure a sample of real keys on a real node.

Memcached handles memory differently. It allocates in fixed-size chunk classes, and evicts per class. A workload whose value sizes shift over time can end up with memory stranded in classes it no longer uses.

Hot keys

Hashing spreads keys, not requests. One very popular key lands on one node, however good the hashing is. The second part of the simulation sends 2 million requests with Zipf-distributed popularity (exponent 1.1) over 1 million keys to a 10-node ring with 128 virtual nodes:

2) Hot keys: 10 nodes x 128 vnodes, 1M keys, 2M Zipf(s=1.1) requests
   share of requests on the single hottest key: 12.4%
   plain ring                         busiest node 1.94x mean, cluster sees 100% of requests
   top 100 keys on 3 replicas each    busiest node 1.18x mean, cluster sees 100% of requests
   top 1,000 keys in a local L1       busiest node 1.12x mean, cluster sees 31% of requests

The hottest key alone received 12.4% of all requests, so its node carried 1.94 times the average load. Two mitigations helped:

• Replicating the top 100 keys to three nodes each, with reads spread among them, brought the busiest node down to 1.18 times the average. Writes must then update all copies, so this suits read-mostly keys.
• A small local cache in each application server, holding the 1,000 hottest keys for a few seconds, brought it to 1.12 times, and removed 69% of all requests from the cluster entirely. The cost is staleness up to the local TTL, and a local cache in every process that has to be invalidated or left to expire.

Both need to know which keys are hot. Options include sampling requests in the client or proxy, and keeping a small top-k sketch. Hot keys also change over time, for example when something goes viral, so detection has to be continuous.

Membership and failure detection

Nodes have to agree on which nodes exist, which are alive, and who owns which slots. Two broad designs exist:

• A coordination service such as ZooKeeper or etcd holds the membership and the slot map (chapter 13). Nodes and clients watch it.
• Gossip. Nodes exchange heartbeats and their view of other nodes. Redis Cluster works this way. Every node pings others over a cluster bus, a separate port at the data port plus 10,000 by default, and each ping carries information about a few other nodes.

Failure detection is a trade-off between speed and false alarms. A node that misses heartbeats might be dead, or might be paused for garbage collection, or might be behind a congested link. The third part of the simulation models a healthy node sending a heartbeat every second for 30 days, with assumed rates of rare sender pauses of 0.5–6 seconds, short network blips, and independent packet loss. It counts how often a silence exceeds a timeout $T$:

3) Failure detection on a healthy node: heartbeat every 1 s for 30 days
timeout T (s)  false suspicions/day  detect a crash within (s)
          1.5                 112.6                        1.5
          2.5                  20.0                        2.5
          3.5                  15.4                        3.5
          5.0                   7.4                        5.0
          8.0                   0.0                        8.0

A 1.5-second timeout would have declared this healthy node dead 113 times a day. Raising it to 5 seconds still gave 7 false alarms a day, and only 8 seconds removed them on this trace, at the price of 8 seconds to notice a real crash. The exact numbers come from the assumed pause rates. The shape doesn’t: the tail of the gap distribution decides the false alarm rate, and that tail is made of pauses and network trouble, not normal jitter.

Redis Cluster reduces false alarms with agreement. A node that hasn’t answered a ping for NODE_TIMEOUT is marked PFAIL (possible failure) by the node that noticed. Only when a majority of primaries report PFAIL or FAIL for it within twice NODE_TIMEOUT is it marked FAIL, which lets a replica start an election to replace it. One observer’s bad link doesn’t trigger a failover. A real pause on the failing node still can, and a false failover in a cache has a cost: the promoted replica may be missing recent writes, and the old primary comes back as a replica of the new one.

The specification also states the cluster’s availability rule. The majority side of a partition stays available if it has a majority of primaries and at least one reachable replica for every unreachable primary. The minority side stops accepting writes.

Warming

A new or restarted cache node is empty. If it suddenly owns a share of the keys, every request for them misses, and the database sees that share of traffic at once. This is the cold start from chapter 7, at cluster scale.

The point of all of them is that the database must never see the cluster’s whole miss traffic at once. A useful exercise is to calculate what the database would receive if the cache were empty, and compare it with what the database can serve.

Consistency with the database

Chapter 7 described the stale-set race in a single cache and the lease fix from Facebook’s memcache paper. A distributed cache adds new ways to be stale:

• Invalidations can miss. A DELETE sent to a node during a failover or a slot migration may land on the old owner and be lost.
• Replicas lag. A read from a replica just after a delete on the primary can return the old value.
• Regions. With a cache cluster per region, each region’s copy needs its own invalidation.

The robust pattern is to drive invalidation from the database’s change log rather than from application code. A consumer of PostgreSQL’s logical replication stream, the change data capture from chapter 15, sees every committed change, in commit order, and deletes the matching cache keys. A crash of an application server between “commit” and “delete” can no longer skip an invalidation, and the consumer can retry deletes until they succeed, because deleting is idempotent.

For values where a stale overwrite matters, writes can carry a version. A row’s version is stored with the cached value, and a fill is accepted only if its version is newer than the one cached. Redis can do that check atomically with a short Lua script, as chapter 22 did for rate limits.

ALTER TABLE profiles ADD COLUMN version bigint NOT NULL DEFAULT 0;

UPDATE profiles
SET display_name = $1, version = version + 1
WHERE user_id = $2
RETURNING version;

Finally, every entry gets a TTL, even when invalidation is in place. Invalidation is the main mechanism, and the TTL bounds how long any missed invalidation can last.

Failure modes

The takeaway

A distributed cache is a partitioned, replicated store that is allowed to lose data, and that permission is what keeps it simple and fast. Virtual nodes or explicit slots spread keys, but not requests, so hot keys need their own handling. Failure detection trades speed for false alarms, and false alarms have a cost. The cache is only as consistent as its invalidation, so driving deletes from the database’s own change log, with a TTL behind it, closes most of the gaps.