Consistent hashing, from first idea to production

You have four cache servers in front of a database, and you pick a server for each key with hash(key) % 4. Traffic grows, so you add a fifth. The moment the new configuration goes live, four out of every five keys are looked up on a server that has never seen them. Every one of those lookups misses, falls through to the database, and the database that was quietly handling 5% of reads is suddenly asked for most of them.

Consistent hashing is the fix. It arranges servers and keys on a circle so that adding or removing one server only moves the keys next to it: about one in five when going from four servers to five, instead of four in five. Caches, distributed databases, load balancers and CDNs all lean on the idea or on one of its descendants.

By the end of this article you’ll be able to:

• explain, with numbers, why hash % N breaks when N changes;
• build a hash ring with virtual nodes and replicas, and measure its balance;
• choose between a ring, rendezvous hashing, jump hash and Maglev for a given job;
• explain why Redis Cluster uses 16,384 fixed slots instead of a ring, and what that does to multi-key commands;
• walk through a distributed cache design in an interview.

It’s for backend engineers who have used a cache and a hash map and now need to spread data over more than one machine. If hash(key) % N already makes you nervous, skip to Intermediate. The In production section is honest about ezychat.io, the product I build: it doesn’t shard at all yet, and I’ll show what in our code would break the day it does.

Beginner: spreading keys across servers

One server stops being enough

A single cache server holds as much as its memory allows and answers as fast as one machine can. When either limit is reached, you add machines and split the data between them. Every key (a session, a product page, a rate-limit counter) has to live on exactly one of them, and every application server has to agree on which one without asking around.

So you need a function: given a key and a list of servers, return the server, with the same answer on every machine, every time. That’s a job for a hash function.

The obvious answer: hash % N

A hash function turns any key into a big number, always the same number for the same key. The remainder of that number divided by the number of servers picks a server. Here it is in Python:

import hashlib

def stable_hash(text: str) -> int:
    """The same number for the same text, in every process, on every machine."""
    digest = hashlib.md5(text.encode(), usedforsecurity=False).digest()
    return int.from_bytes(digest[:8], "big")

def server_for(key: str, servers: list[str]) -> str:
    return servers[stable_hash(key) % len(servers)]

servers = ["cache-a", "cache-b", "cache-c", "cache-d"]
for key in ["user:42", "user:43", "cart:9001", "product:17"]:
    print(f"{key:<11} -> {server_for(key, servers)}")
# user:42     -> cache-a
# user:43     -> cache-b
# cart:9001   -> cache-b
# product:17  -> cache-b

MD5 is here because it’s stable and spreads keys well, not for security; usedforsecurity=False says so to Python. (Why not Python’s built-in hash()? It changes every time the process starts. That’s in Intermediate.)

This works perfectly while the number of servers stays fixed. With many keys, each server gets very close to a quarter of them. Lookups cost one hash and one division. Plenty of real systems use exactly this and are right to, because their N never changes.

What happens when N changes?

Change N and almost every key moves. Count it:

keys = [f"user:{i}" for i in range(100_000)]

def moved_fraction(n_before: int, n_after: int) -> float:
    moved = sum(stable_hash(k) % n_before != stable_hash(k) % n_after for k in keys)
    return moved / len(keys)

for before, after in [(4, 5), (4, 3), (10, 11), (4, 8)]:
    print(f"{before} -> {after} servers: {moved_fraction(before, after):.1%} of keys move")
# 4 -> 5 servers: 79.9% of keys move
# 4 -> 3 servers: 75.0% of keys move
# 10 -> 11 servers: 90.7% of keys move
# 4 -> 8 servers: 50.0% of keys move

Adding a fifth server moves 80% of keys. Losing one of four moves 75%. And it gets worse as the cluster grows: going from 10 to 11 moves 91%. The only friendly case is doubling, where exactly half stay put.

The reason is plain arithmetic. A key stays on the same server only if its remainder is the same under both N and N + 1. Take the hash values 0 to 19 as an example. Under % 4 they go A, B, C, D, A, B, C, D… and under % 5 they go A, B, C, D, E, A, B… The two patterns only agree at the start, before they drift out of step:

In general, going from N to N + 1 keeps only 1 key in N + 1 where it was. Everything else moves, and most of it moves between the old servers: cache-b hands keys to cache-c, which hands keys to cache-a, and so on. Adding capacity reshuffles everyone.

Why moving keys hurts

For a cache, a moved key isn’t lost data; it’s a miss. The new owner doesn’t have the value, so the request goes to the database, gets the answer, and stores it on the new server. One miss per key is cheap. Eighty percent of your keys missing in the same few seconds is a cache stampede: a flood of requests hitting the slow layer at once because the fast layer forgot everything. Here’s the arithmetic for a modest setup:

requests_per_second = 50_000
hit_rate = 0.95                     # 95% of reads are answered by the cache
moved = 0.80                        # share of keys that change server, 4 -> 5

before = requests_per_second * (1 - hit_rate)
# A moved key lives on a server that has never seen it: its next read is a miss.
after = requests_per_second * (1 - hit_rate * (1 - moved))
print(f"database reads/s before: {before:,.0f}")
print(f"database reads/s right after adding a server: {after:,.0f} ({after / before:.1f}x)")
# database reads/s before: 2,500
# database reads/s right after adding a server: 40,500 (16.2x)

Sixteen times the database load, caused by adding capacity. A database sized for 2,500 reads a second doesn’t survive 40,500, and the cache can’t warm up if the database behind it is timing out. You can turn an ordinary scale-up into an outage this way.

For a database spread over several machines (each slice is called a shard), a moved key is worse: it’s data that must be copied to its new machine before anyone can read it there. Moving 80% of a terabyte to add one machine means copying 800 GB while also serving traffic, and keeping reads and writes correct while it happens. That’s a migration project, every time you grow.

What we want is a placement rule where adding one server moves only the keys that the new server should take, about 1/(N + 1) of them, and nothing moves between the servers that were already there.

The ring idea

Picture a ring road around a city with a few post offices on it. Every house sends its mail to the next post office clockwise. If a new post office opens, the only houses that switch are the ones between the new office and the office before it; everyone else keeps the same one. If an office closes, its houses simply carry on clockwise to the next one.

Consistent hashing does this with numbers. Take all possible hash values, 0 up to 2⁶⁴ − 1, and bend the line into a circle, so that after the biggest value comes 0 again. That circle is the hash ring. Hash each server’s name to get its position on the ring. Hash each key to get its position on the same ring. A key belongs to the first server you meet walking clockwise from it.

Now add a server E between A and B. Keys between A and E used to walk past E’s position to B; now they stop at E. Every other key’s walk is unchanged, because nothing else on its path changed. Remove a server and the opposite happens: only its keys move, to the next server clockwise. That’s the whole idea, and the name: the mapping stays consistent when the set of servers changes.

The idea comes from a 1997 paper by David Karger and colleagues at MIT (“Consistent hashing and random trees”), written to spread web content over many caches. Two of its authors, Tom Leighton and Danny Lewin, went on to found Akamai.

Intermediate: building a hash ring

The ring, step by step

A ring needs three operations: add a server, remove a server, and find the owner of a key. The data structure is a sorted list of server positions plus a hash map from position to server.

1. Add a server: hash its name to a position and insert the position into the sorted list.
2. Find a key’s owner: hash the key, then binary search the sorted list for the first server position greater than the key’s position. Binary search halves the remaining list at every step, so 1,000 positions take about 10 comparisons. If the key is past the last position, wrap around to the first.
3. Remove a server: delete its position.

In the code below, each server goes on the ring several times, under the names cache-a#0, cache-a#1 and so on. Those are virtual nodes, and they’re the subject of the next sections. With vnodes=1 it’s the plain ring from the picture.

import bisect
import hashlib

def stable_hash(text: str) -> int:
    digest = hashlib.md5(text.encode(), usedforsecurity=False).digest()
    return int.from_bytes(digest[:8], "big")      # a point on a ring of 2**64 positions

class HashRing:
    def __init__(self, vnodes: int = 1):
        self.vnodes = vnodes
        self.points: list[int] = []               # sorted positions on the ring
        self.owner: dict[int, str] = {}           # position -> server

    def add(self, server: str) -> None:
        for i in range(self.vnodes):
            point = stable_hash(f"{server}#{i}")
            self.owner[point] = server
            bisect.insort(self.points, point)

    def remove(self, server: str) -> None:
        for i in range(self.vnodes):
            point = stable_hash(f"{server}#{i}")
            del self.owner[point]
            self.points.remove(point)

    def lookup(self, key: str) -> str:
        i = bisect.bisect_right(self.points, stable_hash(key))   # first point clockwise
        return self.owner[self.points[i % len(self.points)]]     # past the top: wrap to 0

bisect_right returns the index of the first position strictly greater than the key’s hash, and i % len(self.points) turns “past the end” into index 0. That modulo is the only thing that makes the line a circle.

Adding a server moves about 1/N of the keys

Here’s the property we came for, measured. Build a ring of four servers, record every key’s owner, add a fifth, and compare:

# uses HashRing from above
keys = [f"user:{i}" for i in range(100_000)]

for vnodes in (1, 100):
    ring = HashRing(vnodes)
    for s in ["cache-a", "cache-b", "cache-c", "cache-d"]:
        ring.add(s)
    before = {k: ring.lookup(k) for k in keys}

    ring.add("cache-e")
    after = {k: ring.lookup(k) for k in keys}
    moved = [k for k in keys if before[k] != after[k]]
    print(f"vnodes={vnodes:<3} add cache-e: {len(moved) / len(keys):.1%} of keys move,",
          "all of them to cache-e:", all(after[k] == "cache-e" for k in moved))

    ring.remove("cache-e")
    print(f"           remove it again: back to the old layout: {all(ring.lookup(k) == before[k] for k in keys)}")
# vnodes=1   add cache-e: 21.4% of keys move, all of them to cache-e: True
#            remove it again: back to the old layout: True
# vnodes=100 add cache-e: 19.1% of keys move, all of them to cache-e: True
#            remove it again: back to the old layout: True

About a fifth of the keys move, against 80% for hash % N, and every moved key goes to the new server. No key moves between cache-a and cache-b. Removing the server again restores the exact old layout, which matters more than it looks: a server that restarts after a crash gets its old keys back.

The 21.4% in the first line is luck, though. With one point per server, the new server takes whatever arc it happens to land in, and arcs on a randomly cut circle are very unequal. Try 200 different names for the newcomer:

# uses HashRing from above
RING = 2**64

def share_taken(vnodes: int, newcomer: str) -> float:
    """Fraction of the ring (so of the keys) a new server takes over."""
    ring = HashRing(vnodes)
    for s in ["cache-a", "cache-b", "cache-c", "cache-d"]:
        ring.add(s)
    before = set(ring.points)
    ring.add(newcomer)
    taken = 0
    pts = ring.points
    for i, p in enumerate(pts):
        if p not in before:                      # a new point owns the arc behind it
            taken += (p - pts[i - 1]) % RING
    return taken / RING

for vnodes in (1, 100):
    shares = sorted(share_taken(vnodes, f"cache-new-{t}") for t in range(200))
    print(f"vnodes={vnodes:<3} new server takes between {shares[0]:.1%} and {shares[-1]:.1%} "
          f"(median {shares[100]:.1%}) over 200 tries")
# vnodes=1   new server takes between 0.0% and 34.0% (median 12.4%) over 200 tries
# vnodes=100 new server takes between 16.7% and 25.5% (median 21.2%) over 200 tries

With one point each, a new server might take a third of the keys or practically none. That’s the ring’s real weakness, and the next idea fixes it.

Virtual nodes even out the load

With one point per server, server positions are a handful of random cuts on a circle, and random cuts make wildly unequal pieces. The fix is to give every server many points, its virtual nodes (vnodes): cache-a#0, cache-a#1, … cache-a#99, all hashed to different places. Each server now owns many small arcs instead of one big one, and the sum of many random arcs is close to the average. It’s the same reason a hundred coin flips land nearer 50% heads than ten do.

Measured exactly (each server’s total arc length, no sampling) on 20 different ten-server clusters:

import statistics
from collections import defaultdict
# uses HashRing from above

RING = 2**64

def busiest_share(names: list[str], vnodes: int) -> float:
    """Exact load of the busiest server: the arcs it owns, divided by a fair 1/N."""
    ring = HashRing(vnodes)
    for s in names:
        ring.add(s)
    owned: defaultdict[str, int] = defaultdict(int)
    for i, p in enumerate(ring.points):
        owned[ring.owner[p]] += (p - ring.points[i - 1]) % RING   # arc back to the previous point
    return max(owned.values()) / (RING / len(names))

# 20 different 10-server clusters (different names, so different random layouts).
clusters = [[f"cache-{c}-{i}" for i in range(10)] for c in range(20)]
for vnodes in (1, 10, 100, 1000):
    results = [busiest_share(names, vnodes) for names in clusters]
    print(f"vnodes={vnodes:>4}  points={10 * vnodes:>6}  busiest server: "
          f"median {statistics.median(results):.2f}x fair share, worst {max(results):.2f}x")
# vnodes=   1  points=    10  busiest server: median 2.83x fair share, worst 3.76x
# vnodes=  10  points=   100  busiest server: median 1.51x fair share, worst 1.95x
# vnodes= 100  points=  1000  busiest server: median 1.16x fair share, worst 1.30x
# vnodes=1000  points= 10000  busiest server: median 1.05x fair share, worst 1.08x

With one point each, the busiest of ten servers typically carries almost three times its fair share. You’d have to size every server for that one, so you’d be paying for roughly three times the hardware the load needs. At 100 vnodes the busiest server is about 16% over; at 1,000, about 5%. The improvement follows a square root: ten times the points cuts the excess roughly by three.

Play with it. The ring below starts with four servers and one point each:

Things to try: add a server and watch only one neighbor’s keys light up, while the counter shows what hash % N would have moved. Remove one and see its keys go to the next server clockwise. Then push the slider to 128 points and add or remove again: the moved keys now come from every server, a little from each, and the bars level out.

What virtual nodes cost

Nothing about the ring is free, but vnodes are cheap. Memory is one 8-byte position and one pointer per point: 100 servers with 100 vnodes each is 10,000 entries, a few hundred kilobytes in Python and far less in C. Lookups are a binary search over all points, so they grow with the logarithm: 10,000 points take about 14 comparisons. On my laptop (Apple M1 Pro, Python 3.13):

import time
# uses HashRing and stable_hash from above

ring = HashRing(vnodes=100)
for i in range(10):
    ring.add(f"cache-{i}")
keys = [f"user:{i}" for i in range(200_000)]

start = time.perf_counter()
for k in keys:
    stable_hash(k)
hash_only = time.perf_counter() - start

start = time.perf_counter()
for k in keys:
    ring.lookup(k)
lookup = time.perf_counter() - start

print(f"md5 alone:      {hash_only / len(keys) * 1e9:,.0f} ns per key")
print(f"ring lookup:    {lookup / len(keys) * 1e9:,.0f} ns per key (1,000 points, ~10 comparisons)")
# md5 alone:      734 ns per key
# ring lookup:    926 ns per key (1,000 points, ~10 comparisons)

The binary search adds about 200 ns on top of hashing the key. Hashing dominates, and a network round trip to the cache is hundreds of times slower than either. The real costs of vnodes are elsewhere: rebuilding the sorted list when membership changes, shipping a bigger ring to every client, and (in databases) having many more ranges to track and repair. Cassandra’s story about that is in Expert.

When a server dies

Vnodes also decide what happens on failure. On a plain ring, a dead server’s whole arc goes to the one server after it, which suddenly carries double. If that pushes it over, it fails too, and its load plus the first server’s lands on the next one. That’s a cascading failure, and a plain ring sets it up neatly. With vnodes, the dead server’s hundred small arcs each go to a different successor, so its load spreads over the whole cluster:

from collections import Counter
# uses HashRing from above

keys = [f"user:{i}" for i in range(100_000)]
servers = [f"cache-{i}" for i in range(10)]

for vnodes in (1, 100):
    ring = HashRing(vnodes)
    for s in servers:
        ring.add(s)
    before = {k: ring.lookup(k) for k in keys}
    victim = Counter(before.values()).most_common(1)[0][0]    # the busiest server dies
    ring.remove(victim)
    heirs = Counter(ring.lookup(k) for k in keys if before[k] == victim)
    load = Counter(ring.lookup(k) for k in keys)
    print(f"vnodes={vnodes:<3} {victim} dies: its keys go to {len(heirs)} server{'s' * (len(heirs) > 1)}; "
          f"busiest survivor now holds {max(load.values()) / (len(keys) / 9):.2f}x the new average")
# vnodes=1   cache-9 dies: its keys go to 1 server; busiest survivor now holds 3.06x the new average
# vnodes=100 cache-7 dies: its keys go to 9 servers; busiest survivor now holds 1.09x the new average

One point per server: all of the dead server’s keys land on a single neighbor, which ends up with three times the average. A hundred points: they spread over all nine survivors, and nobody gets more than 9% over. I wouldn’t run a ring with fewer than 100 vnodes per server for this reason alone.

Replicas: the next k distinct servers

A key stored on one server is lost (or, for a cache, cold) when that server goes. Storage systems keep replicas: copies of each key on k servers, usually 3. The ring gives a natural rule: the key’s owner plus the next servers clockwise. The list of servers that should hold a key, in order, is its preference list.

With vnodes there’s a trap. The next three points clockwise can belong to the same physical server twice, and two copies on one machine protect against nothing. Walk clockwise and keep only servers you haven’t picked yet:

import bisect
# uses HashRing and stable_hash from above

def preference_list(ring: HashRing, key: str, k: int) -> list[str]:
    """The first k *distinct* servers clockwise from the key."""
    start = bisect.bisect_right(ring.points, stable_hash(key))
    chosen: list[str] = []
    for step in range(len(ring.points)):
        server = ring.owner[ring.points[(start + step) % len(ring.points)]]
        if server not in chosen:          # a second vnode of a server we already have
            chosen.append(server)
            if len(chosen) == k:
                break
    return chosen

ring = HashRing(vnodes=8)
for s in ["cache-a", "cache-b", "cache-c", "cache-d", "cache-e"]:
    ring.add(s)

print(preference_list(ring, "user:42", 3))

naive = [ring.owner[ring.points[(bisect.bisect_right(ring.points, stable_hash(k)) + j) % len(ring.points)]]
         for k in ["user:42"] for j in range(3)]
print("next 3 points, not distinct:", naive)

dupes = 0
for i in range(10_000):
    start = bisect.bisect_right(ring.points, stable_hash(f"user:{i}"))
    picks = {ring.owner[ring.points[(start + j) % len(ring.points)]] for j in range(3)}
    dupes += len(picks) < 3
print(f"keys whose next 3 points include the same server twice: {dupes / 10_000:.1%}")
# ['cache-a', 'cache-e', 'cache-d']
# next 3 points, not distinct: ['cache-a', 'cache-e', 'cache-a']
# keys whose next 3 points include the same server twice: 48.7%

With five servers, nearly half of all keys would have two “replicas” on one machine under the naive rule. Amazon’s Dynamo paper describes exactly this skip (“the preference list for a key is constructed by skipping positions in the ring to ensure that the list contains only distinct physical nodes”), and real systems go one step further and skip servers in the same rack or availability zone, which is the third exercise below.

A pleasant side effect of the rule: when the owner fails, the second server on the list already has the data. Reads can go there immediately, and the ring itself doesn’t need to change until you decide the failure is permanent.

Use a stable hash

The ring only works if every machine computes the same position for the same key. Python’s built-in hash() fails that test for strings: it mixes in a random secret chosen when the process starts, as a defense against attackers who pick colliding keys (the hash maps article covers that attack, and PYTHONHASHSEED). Four processes, four opinions:

import subprocess, sys

code = "print(hash('user:42') % 4)"
for run in range(4):
    out = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True).stdout.strip()
    print(f"process {run}: user:42 -> server {out}")
# process 0: user:42 -> server 3
# process 1: user:42 -> server 3
# process 2: user:42 -> server 2
# process 3: user:42 -> server 1

Four application servers running that code would each look for user:42 on a different cache, and the answer changes again on every deploy. Use a hash with a fixed definition: MD5 or SHA-256 from hashlib are everywhere and fine; MurmurHash3 and xxHash are faster non-cryptographic choices with libraries in every language; CRC16 is what Redis Cluster uses. The rule that matters more than the choice: every client must use the same function, the same seed and the same key encoding, forever, or two clients will disagree about where a key lives. A Python client and a Go client hashing "user:42" as UTF-8 bytes with MD5 will agree; one hashing the string and one hashing a JSON-encoded version won’t.

You don’t need a cryptographic hash: placement only needs an even spread, not secrecy.

What it costs, all together

import bisect

def owner(points: dict[int, str], key: int) -> str:
    ring = sorted(points)
    i = bisect.bisect_left(ring, key)          # a server at exactly the key's position owns it
    return points[ring[i % len(ring)]]

servers = {10: "A", 40: "B", 70: "C", 90: "D"}
keys = [5, 15, 45, 72, 85, 95]
print({k: owner(servers, k) for k in keys})

without_c = {p: s for p, s in servers.items() if s != "C"}
print({k: owner(without_c, k) for k in keys if owner(servers, k) != owner(without_c, k)})
# {5: 'A', 15: 'B', 45: 'C', 72: 'D', 85: 'D', 95: 'A'}
# {45: 'D'}

clusters = [[f"node-{c}-{i}" for i in range(50)] for c in range(10)]
for vnodes in (100, 200, 500, 1000, 2000):
    r = [busiest_share(names, vnodes) for names in clusters]
    print(f"50 servers, {vnodes:>4} vnodes: busiest median {statistics.median(r):.2f}x, worst {max(r):.2f}x")
# 50 servers,  100 vnodes: busiest median 1.22x, worst 1.33x
# 50 servers,  200 vnodes: busiest median 1.18x, worst 1.22x
# 50 servers,  500 vnodes: busiest median 1.10x, worst 1.13x
# 50 servers, 1000 vnodes: busiest median 1.08x, worst 1.11x
# 50 servers, 2000 vnodes: busiest median 1.06x, worst 1.07x

import bisect
# uses HashRing and stable_hash from above

zone_of = {"cache-a": "zone-1", "cache-b": "zone-1", "cache-c": "zone-2",
           "cache-d": "zone-2", "cache-e": "zone-3", "cache-f": "zone-3"}

def replicas_across_zones(ring: HashRing, key: str, k: int) -> list[str]:
    """First k servers clockwise, at most one per zone."""
    start = bisect.bisect_right(ring.points, stable_hash(key))
    chosen, zones = [], set()
    for step in range(len(ring.points)):
        server = ring.owner[ring.points[(start + step) % len(ring.points)]]
        if server in chosen or zone_of[server] in zones:
            continue
        chosen.append(server)
        zones.add(zone_of[server])
        if len(chosen) == k:
            return chosen
    raise ValueError(f"only {len(zones)} zones, can't place {k} replicas")

ring = HashRing(vnodes=50)
for s in zone_of:
    ring.add(s)

print(replicas_across_zones(ring, "user:42", 3))
spread = all(len({zone_of[s] for s in replicas_across_zones(ring, f"user:{i}", 3)}) == 3
             for i in range(10_000))
print("every key has one copy per zone:", spread)
try:
    replicas_across_zones(ring, "user:42", 4)
except ValueError as e:
    print("ValueError:", e)
# ['cache-b', 'cache-f', 'cache-d']
# every key has one copy per zone: True
# ValueError: only 3 zones, can't place 4 replicas

Advanced: other ways to hash consistently

The ring is the famous answer, not the only one. Since 1997 people have found schemes with better balance, less memory or faster lookups, each by giving something up. Knowing the trade is what lets you pick.

Rendezvous hashing: every server bids

Rendezvous hashing, also called highest random weight (HRW), was published by David Thaler and Chinya Ravishankar in 1996, slightly before the ring. It has no ring at all. For each key, every server computes a score from the pair (server, key), and the server with the highest score wins:

from collections import Counter
# uses stable_hash from above

def rendezvous(key: str, servers: list[str]) -> str:
    """Every server bids for the key; the highest bid wins."""
    return max(servers, key=lambda s: stable_hash(f"{s}:{key}"))

keys = [f"user:{i}" for i in range(100_000)]
four = ["cache-a", "cache-b", "cache-c", "cache-d"]
five = four + ["cache-e"]

before = {k: rendezvous(k, four) for k in keys}
after = {k: rendezvous(k, five) for k in keys}
moved = [k for k in keys if before[k] != after[k]]
print(f"add cache-e: {len(moved) / len(keys):.1%} move, all to cache-e: {all(after[k] == 'cache-e' for k in moved)}")

load = Counter(after.values())
print("busiest vs average:", round(max(load.values()) / (len(keys) / 5), 3))

# The runner-up is the natural replica: rank servers by bid, take the top k.
def top_k(key: str, servers: list[str], k: int) -> list[str]:
    return sorted(servers, key=lambda s: stable_hash(f"{s}:{key}"), reverse=True)[:k]
print(top_k("user:42", five, 3))
# add cache-e: 19.9% move, all to cache-e: True
# busiest vs average: 1.01
# ['cache-e', 'cache-b', 'cache-c']

Why only the minimum moves: adding cache-e adds one more bid per key. A key changes owner only if the new bid beats all the old ones, and then it moves to cache-e; every other key’s winner is unchanged. Removing a server only affects the keys it was winning, and each goes to its runner-up. Balance is excellent with no vnodes at all (the 1.01 is sampling noise from 100,000 keys), and replicas are just the top k bids, already distinct.

The price is lookup cost: N hashes per key, because every server bids. With 5 or 50 servers that’s nothing. With 5,000 it’s too slow per request, unless you do what GitHub’s load balancer (GLB Director) does: use rendezvous ordering to fill a fixed-size forwarding table in advance, so a lookup is one hash and one array read.

I reach for rendezvous first when the server list is small and changes rarely, such as choosing which of a dozen workers owns a tenant’s scheduled jobs. It’s ten lines with no state, no vnodes to tune, and it’s hard to get wrong.

Jump consistent hash: no memory at all

In 2014, John Lamping and Eric Veach at Google published “A Fast, Minimal Memory, Consistent Hash Algorithm”, better known as jump consistent hash. It maps a 64-bit key to a bucket number from 0 to N − 1 with no table at all. The whole algorithm is a loop of a few lines; here it is translated line for line from the paper’s C++:

from collections import Counter
# uses stable_hash from above

def jump_hash(key: int, num_buckets: int) -> int:
    """Lamping and Veach (2014), line for line. key is a 64-bit integer."""
    b, j = -1, 0
    while j < num_buckets:
        b = j
        key = (key * 2862933555777941757 + 1) & 0xFFFFFFFFFFFFFFFF   # 64-bit LCG step
        j = int((b + 1) * (float(1 << 31) / float((key >> 33) + 1)))
    return b

keys = [stable_hash(f"user:{i}") for i in range(100_000)]

for before, after in [(4, 5), (10, 11), (100, 101)]:
    moved = sum(jump_hash(k, before) != jump_hash(k, after) for k in keys)
    print(f"{before} -> {after} buckets: {moved / len(keys):.2%} move (ideal {1 / after:.2%})")

load = Counter(jump_hash(k, 10) for k in keys)
print("10 buckets, busiest vs average:", round(max(load.values()) / 10_000, 3))
print("moved keys only ever go to the new bucket:",
      all(jump_hash(k, 5) in (jump_hash(k, 4), 4) for k in keys))
# 4 -> 5 buckets: 20.14% move (ideal 20.00%)
# 10 -> 11 buckets: 9.25% move (ideal 9.09%)
# 100 -> 101 buckets: 1.02% move (ideal 0.99%)
# 10 buckets, busiest vs average: 1.021
# moved keys only ever go to the new bucket: True

How it works, in words: imagine growing from 1 bucket to N one at a time. When bucket j + 1 is added, each key should jump to it with probability 1/(j + 1), which is exactly the minimum movement. Simulating every step would take N iterations, so instead the loop uses a seeded random number (the line with the big constant is a simple pseudo-random generator) to compute directly which future bucket count this key will next jump at, and skips there. Jumps get rarer as buckets grow, so the loop runs only about ln N times:

import math
# uses stable_hash from above

def jump_steps(key: int, num_buckets: int) -> int:
    b, j, steps = -1, 0, 0
    while j < num_buckets:
        b, steps = j, steps + 1
        key = (key * 2862933555777941757 + 1) & 0xFFFFFFFFFFFFFFFF
        j = int((b + 1) * (float(1 << 31) / float((key >> 33) + 1)))
    return steps

keys = [stable_hash(f"user:{i}") for i in range(20_000)]
for n in (10, 1_000, 100_000):
    avg = sum(jump_steps(k, n) for k in keys) / len(keys)
    print(f"{n:>7,} buckets: {avg:.1f} loop iterations on average (ln n = {math.log(n):.1f})")
#      10 buckets: 2.9 loop iterations on average (ln n = 2.3)
#   1,000 buckets: 7.5 loop iterations on average (ln n = 6.9)
# 100,000 buckets: 12.1 loop iterations on average (ln n = 11.5)

Twelve iterations for a hundred thousand buckets, zero memory, near-perfect balance, minimal movement. It sounds like it wins everything. It doesn’t, because of one restriction: buckets are numbered 0 to N − 1, and you can only add or remove at the end. You can’t remove bucket 2 of 5. The paper says so plainly: it’s meant for data storage, where buckets are shards that are always there (each backed by replicas), not for caches where arbitrary servers vanish.

If you ignore this and “remove” a dead server by deleting it from the list, everything after it renumbers:

# uses stable_hash and jump_hash from above

servers = ["db-0", "db-1", "db-2", "db-3", "db-4"]
keys = [stable_hash(f"user:{i}") for i in range(100_000)]
before = {k: servers[jump_hash(k, 5)] for k in keys}

# db-2 dies. Wrong fix: drop it from the list, so everyone after it shifts down.
shifted = [s for s in servers if s != "db-2"]
wrong = sum(before[k] != shifted[jump_hash(k, 4)] for k in keys)

# Right fix: bucket numbers are permanent; point bucket 2 at a replacement.
table = {0: "db-0", 1: "db-1", 2: "db-5", 3: "db-3", 4: "db-4"}
right = sum(before[k] != table[jump_hash(k, 5)] for k in keys)

print(f"renumber the list: {wrong / len(keys):.1%} of keys change server")
print(f"replace bucket 2:  {right / len(keys):.1%} of keys change server")
# renumber the list: 54.8% of keys change server
# replace bucket 2:  19.8% of keys change server

More than half the keys move, worse than you’d hope from anything called consistent. The right pattern is to treat jump’s output as a logical shard number and keep a small table from shard number to physical machine. A dead machine is replaced in the table; the shard number, and every key’s mapping, stays. That two-level design (keys to a fixed numbering, numbering to machines) comes back in Expert, because it’s how most production systems ended up.

Maglev: a lookup table built for load balancers

Google’s Maglev is the software network load balancer in front of its services, described in a 2016 NSDI paper by Danielle Eisenbud and colleagues. A load balancer has a sharper version of our problem: every packet of a TCP connection must reach the same backend, packets arrive millions of times a second, and backends come and go. Maglev’s answer is a lookup table with a prime number of entries (65,537 in the paper) where each entry names a backend. A packet’s connection hash, modulo the table size, picks an entry: one hash and one array read.

The clever part is filling the table. Each backend gets its own pseudo-random order over all entries (from two hashes of its name: a starting offset and a step size). Backends then take turns: each claims its next preferred entry that’s still free, until the table is full.

from collections import Counter
# uses stable_hash from above

def maglev_table(backends: list[str], m: int = 65_537) -> list[str]:
    """Eisenbud et al. (2016): each backend walks its own permutation of the slots
    and claims the next free one, in turns, until the table is full."""
    perms = []
    for b in backends:
        offset = stable_hash(f"{b}:offset") % m
        skip = stable_hash(f"{b}:skip") % (m - 1) + 1          # m is prime, so any skip visits every slot
        perms.append((offset, skip))
    table: list[str | None] = [None] * m
    nxt = [0] * len(backends)
    filled = 0
    while True:
        for i, b in enumerate(backends):
            offset, skip = perms[i]
            slot = (offset + nxt[i] * skip) % m
            while table[slot] is not None:                       # already claimed: try my next preference
                nxt[i] += 1
                slot = (offset + nxt[i] * skip) % m
            table[slot] = b
            nxt[i] += 1
            filled += 1
            if filled == m:
                return table

backends = [f"lb-backend-{i}" for i in range(10)]
before = maglev_table(backends)
sizes = Counter(before)
print("slots per backend:", min(sizes.values()), "to", max(sizes.values()), "(65,537 / 10 = 6,553.7)")

after = maglev_table([b for b in backends if b != "lb-backend-3"])
changed = sum(x != y for x, y in zip(before, after))
orphaned = sizes["lb-backend-3"]
print(f"remove one backend: {changed:,} slots change, {orphaned:,} of them had to "
      f"(+{(changed - orphaned) / len(before):.2%} of the table moved between survivors)")

key_slot = stable_hash("10.0.0.7:51234->203.0.113.10:443") % len(before)   # a flow's 5-tuple
print("this flow:", before[key_slot], "->", after[key_slot])
# slots per backend: 6553 to 6554 (65,537 / 10 = 6,553.7)
# remove one backend: 6,710 slots change, 6,554 of them had to (+0.24% of the table moved between survivors)
# this flow: lb-backend-3 -> lb-backend-6

Taking turns makes the balance essentially perfect: every backend gets 6,553 or 6,554 entries. The trade is visible in the second line. When a backend leaves, its entries must change, but 156 others (0.24% of the table) also shuffle between survivors, because the turn-taking now plays out a little differently. Maglev accepts slightly-more-than-minimal disruption in exchange for perfect balance and constant-time lookups. That’s a good trade for a load balancer, which also remembers existing connections in a local table, so a shuffled entry mostly affects new connections. Envoy offers both a ring (RING_HASH) and Maglev (MAGLEV) as load-balancing policies for exactly these uses.

Compare them yourself

The table below runs all four schemes in your browser over 20,000 keys. Change the number of servers and it re-runs everything, including adding one more server:

Things to notice: hash % N is perfectly balanced and moves almost everything. The ring moves the minimum but runs about 10 to 25% hot even with 100 vnodes. Rendezvous and jump both move the minimum and balance as well as % N; rendezvous pays with one hash per server, and jump with the numbering rule. (With 20,000 keys, a few percent of the “busiest” column is sampling noise; the ring’s excess is not.)

Bounded loads: cap every server

Even a well-balanced scheme balances keys, not traffic. If some keys are much busier than others, or a cache server is slow for a minute, the server that owns them overloads while its neighbors idle. Consistent hashing with bounded loads, by Vahab Mirrokni, Mikkel Thorup and Morteza Zadimoghaddam (published 2016, SODA 2018), adds a cap: no server may hold more than (1 + ε) times the average, for a small ε like 0.25. A key goes to its usual server if that server is under the cap, otherwise to the next one clockwise that is.

import bisect, math
from collections import Counter
# uses HashRing and stable_hash from above

def place_bounded(ring: HashRing, keys: list[str], epsilon: float | None):
    """Walk clockwise past any server that is already full (Mirrokni, Thorup, Zadimoghaddam)."""
    servers = len(set(ring.owner.values()))
    cap = math.inf if epsilon is None else math.ceil((1 + epsilon) * len(keys) / servers)
    load: Counter[str] = Counter()
    displaced = 0
    for k in keys:
        i = bisect.bisect_right(ring.points, stable_hash(k))
        first = s = ring.owner[ring.points[i % len(ring.points)]]
        while load[s] >= cap:                    # full: try the next point clockwise
            i += 1
            s = ring.owner[ring.points[i % len(ring.points)]]
        load[s] += 1
        displaced += s != first
    return load, displaced

ring = HashRing(vnodes=10)
for i in range(10):
    ring.add(f"cache-{i}")
keys = [f"user:{i}" for i in range(100_000)]

for eps in (None, 0.25, 0.10):
    load, displaced = place_bounded(ring, keys, eps)
    label = "no bound" if eps is None else f"epsilon={eps}"
    print(f"{label:<13} busiest={max(load.values()) / 10_000:.2f}x average, "
          f"{displaced / len(keys):.1%} of keys off their first choice")
# no bound      busiest=1.47x average, 0.0% of keys off their first choice
# epsilon=0.25  busiest=1.25x average, 3.0% of keys off their first choice
# epsilon=0.1   busiest=1.10x average, 8.8% of keys off their first choice

The cap holds exactly, and the cost is the keys that don’t live where the plain ring would put them: 3% at ε = 0.25, 8.8% at 0.1. Those keys are also the ones that may move when load shifts, so a tighter bound means more churn. The paper proves the extra movement stays bounded and grows as ε shrinks.

The well-known deployment is in HAProxy. Andrew Rodland at Vimeo implemented the algorithm there as the hash-balance-factor option for hash-type consistent, applied to in-flight requests rather than stored keys: a request goes to its consistent-hash server unless that server already has more than its share of current requests, in which case it walks on. Vimeo used it in front of the servers that cache video segments. Applying the cap to live requests is the version I’d copy, because it reacts to real load, including a slow server, rather than to a key count that’s only a proxy for load.

Servers of different sizes

Real clusters are rarely uniform: a newer cache node might have twice the memory. Every scheme above can be weighted:

• Ring: give each server vnodes in proportion to its weight. A 32 GB server gets twice the points of a 16 GB one.
• Rendezvous: scale each bid. Turn the hash into a number u between 0 and 1 and use −weight / ln(u) as the bid; a server then wins in exact proportion to its weight.
• Maglev: let heavier backends take extra turns while filling the table.
• Jump: no weights. Give a big machine several bucket numbers instead.

import math
from collections import Counter
# uses HashRing and stable_hash from above

weights = {"cache-small": 1, "cache-medium": 2, "cache-large": 4}   # e.g. 8, 16 and 32 GB of RAM
keys = [f"user:{i}" for i in range(100_000)]
total = sum(weights.values())

# Ring: give each server vnodes in proportion to its weight.
points = {}
for server, w in weights.items():
    r = HashRing(vnodes=100 * w)
    r.add(server)
    points.update(r.owner)
ring = HashRing()
ring.owner = points
ring.points = sorted(points)
ring_load = Counter(ring.lookup(k) for k in keys)

# Rendezvous: scale each bid so a server wins in proportion to its weight.
def weighted_rendezvous(key: str) -> str:
    def bid(server: str) -> float:
        u = (stable_hash(f"{server}:{key}") + 1) / (2**64 + 1)       # uniform in (0, 1)
        return -weights[server] / math.log(u)
    return max(weights, key=bid)
hrw_load = Counter(weighted_rendezvous(k) for k in keys)

for server, w in weights.items():
    print(f"{server:<13} weight share {w / total:.1%}   ring {ring_load[server] / len(keys):.1%}"
          f"   rendezvous {hrw_load[server] / len(keys):.1%}")
# cache-small   weight share 14.3%   ring 14.9%   rendezvous 14.3%
# cache-medium  weight share 28.6%   ring 27.4%   rendezvous 28.6%
# cache-large   weight share 57.1%   ring 57.7%   rendezvous 57.1%

Weights have a second use that I find more valuable than mixed hardware: warming up a new server. Add a cache node at weight 0.1 and raise it over an hour, and its keys move a little at a time, so the misses trickle into the database instead of arriving all at once. Removing a server gracefully is the same in reverse.

Hot keys: what no hash function fixes

Every scheme in this article places keys. If one key gets a large share of all traffic, a celebrity’s profile or a flash-sale product, it lives on one server, and no placement trick changes that. Real traffic is often shaped like a Zipf distribution: the k-th most popular key is requested about 1/k as often as the most popular one. Here’s a million reads of that shape over 100,000 keys on a 20-server ring:

import random
from collections import Counter
# uses HashRing from above

random.seed(7)
ring = HashRing(vnodes=100)
servers = [f"cache-{i}" for i in range(20)]
for s in servers:
    ring.add(s)

# One million reads over 100,000 keys, Zipf-shaped: the k-th most popular key
# is read about 1/k as often as the most popular one.
n_keys = 100_000
weights = [1 / rank for rank in range(1, n_keys + 1)]
reads = random.choices(range(n_keys), weights=weights, k=1_000_000)
hits = Counter(reads)
print(f"the single hottest key gets {hits[0] / len(reads):.1%} of all reads")

def busiest(server_of_read) -> float:
    load = Counter(server_of_read(r) for r in reads)
    return max(load.values()) / (len(reads) / len(servers))

owner = {k: ring.lookup(f"item:{k}") for k in hits}
print(f"plain ring: busiest server takes {busiest(lambda r: owner[r]):.2f}x its share")

# Split the 10 hottest keys into 16 copies each; a read picks a copy at random.
copies = 16
spread = {(k, c): ring.lookup(f"item:{k}#{c}") for k in range(10) for c in range(copies)}
def split(r: int) -> str:
    return spread[(r, random.randrange(copies))] if r < 10 else owner[r]
print(f"hot keys split 16 ways: busiest server takes {busiest(split):.2f}x its share")
# the single hottest key gets 8.2% of all reads
# plain ring: busiest server takes 3.03x its share
# hot keys split 16 ways: busiest server takes 1.38x its share

One key takes 8.2% of all reads, while a fair share for one of 20 servers is 5%. The server holding it is over its share before any other key is counted, and with a couple of other popular keys it ends up at three times its share. More vnodes can’t help: they balance arcs, and this load is a point.

What does help:

• Split the key. Store the hot value under several names (item:0#0 to item:0#15), on different servers, and have each read pick one at random. The busiest server drops from 3.03× to 1.38× above. Writes now have to update every copy, so this suits read-heavy keys.
• Cache it closer. A small in-process cache on each application server, with a TTL of a second or two, absorbs most reads of the hottest keys before they reach the network.
• Collapse concurrent misses. When the hot key expires, let one request recompute it while the others wait for that result, instead of all of them hitting the database. Facebook’s “Scaling Memcache at Facebook” paper (NSDI 2013) describes leases, a version of this built into the cache protocol.
• Detect it. You can only split keys you know are hot. Sample requests, count keys with a Counter or a Count-Min sketch, and treat the top of the list as a live operational signal.

Write keys are the same problem with fewer answers. A counter everyone increments, like a global rate limit, can be split into sub-counters and summed on read, which is what the rate limiter article suggests for one enormous tenant.

Which one would I pick?

table = {0: "db-0", 1: "db-1", 2: "db-5", 3: "db-3", 4: "db-4"}   # db-5 replaces db-2
server = table[jump_hash(key, 5)]

Expert: consistent hashing in real systems

Amazon Dynamo: the paper that made the ring famous

The 2007 paper “Dynamo: Amazon’s Highly Available Key-value Store” (Giuseppe DeCandia and colleagues, SOSP) is the classic description of a ring in production. Dynamo hashes keys with MD5 onto a ring, gives each physical node many virtual nodes, and replicates each key to the next N distinct nodes clockwise (its preference list), with the typical setting N = 3. Writes succeed once W replicas have acknowledged and reads once R have answered; because R + W > N, at least one replica in any read saw the latest write. When a node on the preference list is unreachable, a write goes to the next healthy node instead with a note saying where it belongs (sloppy quorum and hinted handoff), and is handed back when the node returns.

The most useful part for a designer is section 6.2, where the authors compare three ways of cutting the ring:

1. Random tokens per node, with ranges defined by the token values. The original scheme. A joining node had to “steal” ranges from others, which meant those nodes scanning their storage in the background at low priority; the paper says that during the busy shopping season, bootstrapping a node “has taken almost a day to complete”. Every membership change also forced the Merkle trees (used to compare replicas) to be recalculated for the new ranges, and backing up the whole key space meant pulling keys from every node separately.
2. Random tokens per node, but the ring divided into Q equal-sized partitions.
3. Q equal-sized partitions, with Q/S partitions assigned to each of S nodes.

Strategy 3 had the best balance and, in the paper’s words, was “simpler to deploy”: because partition ranges are fixed, each partition can be stored as a separate file and “relocated as a unit by simply transferring the file”, and archiving is a matter of copying partition files. The cost they name is that changing membership now needs coordination to keep the assignment balanced. In other words, Dynamo moved from “nodes own random arcs” to “the ring is cut into fixed partitions, and partitions are assigned to nodes”. Keep that in mind for Redis Cluster below.

Apache Cassandra: tokens and vnodes

Cassandra puts its nodes on a token ring. The default partitioner, Murmur3Partitioner, hashes the partition key with MurmurHash3 to a 64-bit token from −2⁶³ to 2⁶³ − 1, and each node owns the token ranges ending at its tokens. Replication walks the ring for the next distinct nodes, per datacenter and aware of racks when you use NetworkTopologyStrategy.

Cassandra’s history with vnodes is the clearest real-world lesson on tuning them. Virtual nodes arrived in 1.2 with a default of 256 tokens per node, chosen at random, which balanced data well. Cassandra 4.0 changed the default to 16 tokens and turned on an allocation algorithm that places new tokens deliberately to balance load, instead of at random. These are the two lines in its default cassandra.yaml:

num_tokens: 16
allocate_tokens_for_local_replication_factor: 3

The reasons are operational: every token is a range that must be repaired and streamed separately, and with hundreds of random tokens per node, almost any two nodes share some range, so losing two nodes at once is more likely to make data unavailable. Fewer, well-placed tokens get the balance without those costs. Random vnodes buy balance with statistics; a smarter allocator buys it with a little planning.

memcached clients and ketama

memcached servers know nothing about each other. Distribution is entirely the client’s job, which makes memcached the purest example of client-side consistent hashing. In 2007 Last.fm published ketama, a ring for memcached clients: each server gets 160 points on the ring (40 MD5 digests per server, each giving 4 positions), and servers with more memory can get proportionally more. Ketama-compatible hashing is available in libmemcached and in clients for most languages, which matters more than it sounds: a PHP client and a Python client sharing one memcached pool must hash identically, or each fills the cache with entries the other can’t find.

Discord: when lookups become the cost

Discord’s 2017 post “How Discord Scaled Elixir to 5,000,000 Concurrent Users” is unusually concrete about a ring’s cost at scale. In their words, “Discord is a distributed system achieved through consistent hashing”: a ring decides which node runs each guild (a Discord server). A user is in 5 guilds on average, and one session node holds up to 500,000 sessions, so a session node that restarts must look up about 2.5 million guilds. With the ring behind a process call costing about 12 µs per lookup, they wrote, “it would take about 30 seconds just for the cost of lookups on the ring.” They rewrote the ring in pure Elixir (released as ex_hash_ring) and moved it into a read-only shared structure, bringing a lookup to 0.3 µs and the total to 750 ms.

The lesson isn’t about Elixir. A ring lookup looks free until something does millions of them in a burst, and a restart or a reconnect storm is exactly that burst.

Load balancers and CDNs

Consistent hashing started as a CDN idea: the Karger paper spread web objects over caches so each object is cached in one place instead of everywhere. Load balancers use it for affinity, sending the same client or object to the same backend so its local cache stays warm. Maglev and GLB Director do it for packets; HAProxy and Envoy do it for requests. The trade is always the same: affinity raises hit rates, and something has to stop it overloading one backend.

Redis Cluster: 16,384 fixed slots

Redis Cluster doesn’t use a ring. Its specification defines exactly 16,384 hash slots: HASH_SLOT = CRC16(key) mod 16384, using the XMODEM variant of CRC16. Each primary node owns a set of slots, every node knows the full slot-to-node table, and a client that asks the wrong node gets a MOVED reply naming the right one. Here’s the function in Python, including the one exception, hash tags: if the key contains {…} with something inside, only that part is hashed.

def crc16_xmodem(data: bytes) -> int:
    """CRC-16/XMODEM, the variant Redis Cluster uses (polynomial 0x1021, start at 0)."""
    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 key_slot(key: str) -> int:
    """HASH_SLOT = CRC16(key) mod 16384, hashing only the {tag} if there is a non-empty one."""
    start = key.find("{")
    if start != -1:
        end = key.find("}", start + 1)
        if end > start + 1:
            key = key[start + 1:end]
    return crc16_xmodem(key.encode()) % 16384

print(hex(crc16_xmodem(b"123456789")))                    # the spec's check value
for k in ["foo", "rl:api_user:42:29512345", "rl:api_user:42:29512344",
          "rl:{api_user:42}:29512345", "rl:{api_user:42}:29512344"]:
    print(f"{k:<27} slot {key_slot(k)}")
# 0x31c3
# foo                         slot 12182
# rl:api_user:42:29512345     slot 14381
# rl:api_user:42:29512344     slot 10252
# rl:{api_user:42}:29512345   slot 16176
# rl:{api_user:42}:29512344   slot 16176

0x31c3 is the check value the specification gives for "123456789". To be sure the rest is right, I started a three-node Redis 7.2 cluster on my laptop and asked it. Its answers match, and it shows the rule that shapes everything you build on a cluster:

$ redis-cli -c -p 7101 CLUSTER KEYSLOT "rl:api_user:42:29512345"
(integer) 14381
$ redis-cli -c -p 7101 CLUSTER KEYSLOT "rl:{api_user:42}:29512345"
(integer) 16176
$ redis-cli -c -p 7101 EVAL "return {redis.call('GET', KEYS[1]), redis.call('GET', KEYS[2])}" 2 \
    "rl:api_user:42:29512345" "rl:api_user:42:29512344"
(error) CROSSSLOT Keys in request don't hash to the same slot
$ redis-cli -c -p 7101 EVAL "return {redis.call('GET', KEYS[1]), redis.call('GET', KEYS[2])}" 2 \
    "rl:{api_user:42}:29512345" "rl:{api_user:42}:29512344"
1) (nil)
2) "60"
$ redis-cli -c -p 7101 SELECT 5
(error) ERR SELECT is not allowed in cluster mode
$ redis-cli -p 7101 GET "rl:api_user:42:29512345"
(error) MOVED 14381 127.0.0.1:7103

A command or script that touches several keys only works if they’re all in one slot. Two keys that differ only in a suffix almost always land in different slots, so MGET, MSET, multi-key DEL, transactions and Lua scripts over them fail with CROSSSLOT. Hash tags are the way out: put the part that should keep keys together in braces, and they share a slot by construction. The cluster also supports only database 0.

Then I added a fourth node to that cluster holding 100,000 keys and asked redis-cli --cluster rebalance to even it out. It moved 1,366, 1,365 and 1,365 slots from the three old nodes, 4,096 slots in total, exactly a quarter of 16,384. Afterwards the four nodes held 25,014, 24,963, 25,002 and 25,022 keys. That’s the 1/N movement of a ring, done in units of whole slots. (With default settings it took about a minute and a half on my laptop; real migrations are throttled on purpose.)

Why 16,384? Salvatore Sanfilippo, Redis’s creator, explained it in a GitHub issue: nodes exchange their slot ownership as a bitmap in every heartbeat, and 16,384 bits is 2 KB, where 65,536 would be 8 KB in every message. And 16,384 slots still leaves enough slots per node for the largest clusters he had in mind, about 1,000 primaries.

Why fixed slots are easier to operate

I think fixed slots are the better design for anything you’ll operate for years, and the reasons are about people, not math:

• The mapping is a table you can read. “Slot 16176 is on node 3” is a fact you can print, diff and graph. “This key falls in the arc between node-7#41 and node-2#88” is not something anyone reasons about at 3 a.m.
• Moving data has a unit. You move slot 16176, watch it finish, and move the next. Progress is “1,200 of 4,096 slots”, and undoing is moving it back.
• The hash function never changes. Adding nodes changes the table, not CRC16(key) mod 16384, and clients never need to agree on vnode counts or seeds.
• Balance is assigned, not hoped for. A 64 GB node can simply be given twice the slots of a 32 GB one.

The price is a ceiling: the number of slots is fixed, so you can’t have more nodes than slots, and one slot can’t be split if it holds a hot key. Systems that use fixed partitions choose the count up front with room to spare. Elasticsearch fixes the number of primary shards when an index is created and sends each document to a shard chosen by hashing its routing value (the document ID by default); changing the count means reindexing, or the split API, which can only multiply it. Kafka’s default partitioner picks a partition from a hash of the record key modulo the partition count, which is why adding partitions to a topic changes which partition a key goes to and breaks per-key ordering for a while. Both are hash % N over a fixed N, with N chosen large enough that it doesn’t have to change.

This is where consistent hashing quietly comes back: something has to decide which node gets which partition. Some systems use a ring or rendezvous for that step, others an explicit assignment by a controller. Either way, the hard work happens on a small number of large, named units.

Resharding while serving traffic

Moving data between nodes while clients keep reading and writing is where most of the real complexity lives. Redis Cluster’s protocol is a good model. To move slot 8 from node A to node B, the operator marks the slot MIGRATING on A and IMPORTING on B, then moves keys in batches with MIGRATE. During the move:

• A still answers for keys of slot 8 that it has. For a key it no longer has, it replies ASK B: “try B, just for this request”.
• The client sends ASKING to B, then the command. B serves an importing slot only to clients that said ASKING.
• When all keys are moved, the slot is assigned to B everywhere, and from then on A answers MOVED B: “update your table, it lives on B now”.

The difference between ASK (one-off, don’t update your map) and MOVED (permanent, update your map) is what keeps clients correct while a slot is half on each node. Multi-key commands on a slot that is split mid-migration get TRYAGAIN, which the client should retry after a short wait.

The same shape works for any store:

1. Make the new owner ready before it serves. Warm a cache node by ramping its weight up from near zero; give a database node a snapshot, then stream the changes made since.
2. Keep one source of truth for who owns what. Clients that disagree about the ring or the slot table read and write different copies. Redis Cluster uses configuration epochs, Cassandra gossip, many systems a coordinator such as etcd.
3. Move in small, throttled units. The migration competes with production traffic for disk, network and CPU.
4. Make each cutover reversible. One partition at a time means a problem affects one partition and can be rolled back.
5. Watch the database behind a cache. Every cache reshard causes misses: plan for 1/N of the hit traffic.

The interview: design a distributed cache

“Design a distributed cache” is a common senior question, and consistent hashing is the middle of the answer, not the whole of it. Here’s how I’d walk it.

1. Requirements. Read-heavy key-value cache in front of a database. Say 300,000 reads and 30,000 writes per second, a working set of 200 million items around 800 bytes each, p99 under a few milliseconds, and stale reads acceptable for seconds. Ask what happens on a miss (the database must survive the miss rate) and whether losing the cache is an outage or just slow.

2. Size it. Do the arithmetic out loud:

import math

hot_items = 200_000_000          # distinct keys in the working set
bytes_per_item = 800 + 200       # value + key and per-entry overhead
reads_per_s, writes_per_s = 300_000, 30_000

node_ram_gb, usable = 64, 0.70   # leave room for fragmentation, replication buffers, spikes
node_ops = 100_000               # a conservative per-node budget for small GET/SET

data_gb = hot_items * bytes_per_item / 1e9
by_memory = math.ceil(data_gb / (node_ram_gb * usable))
by_ops = math.ceil((reads_per_s + writes_per_s) / node_ops)
print(f"working set: {data_gb:,.0f} GB -> {by_memory} nodes by memory, {by_ops} by throughput")

nodes = max(by_memory, by_ops) + 1          # one spare so a failure doesn't overfill the rest
print(f"plan {nodes} primaries; losing one sends {1 / nodes:.0%} of reads to the database")
miss_reads = reads_per_s / nodes
print(f"that is about {miss_reads:,.0f} extra database reads/s until the keys are re-cached")
# working set: 200 GB -> 5 nodes by memory, 4 by throughput
# plan 6 primaries; losing one sends 17% of reads to the database
# that is about 50,000 extra database reads/s until the keys are re-cached

The inputs are assumptions you state, not facts. The last line is the one interviewers care about: losing one node turns into tens of thousands of extra database reads a second, so either the database can take it, or each node needs a replica.

3. Route. Three options: client-side hashing in a library (fastest, but every client must agree), a proxy tier like mcrouter or twemproxy (one place to change routing, one more hop), or a cluster that routes itself with redirects (Redis Cluster). I’d pick Redis Cluster or a managed equivalent unless there’s a reason not to, because the slot table and resharding are already solved.

4. Placement. If building it: a ring with a few hundred vnodes per node, or fixed partitions mapped to nodes. Say why: adding a node should move about 1/N of keys, not N/(N + 1).

5. Failures. A replica per primary for fast failover. Weights to drain or warm nodes. Request collapsing so one expiring popular key doesn’t send a thousand identical queries. A plan for a node that’s slow rather than dead, which is harder to detect.

6. Hot keys. Detect with sampling, then split them or cache them locally. Say that no hash scheme fixes them.

7. Operations. How a node is added, what’s monitored (hit rate per node, evictions, the miss rate the database sees), and how a reshard is rolled back.

In production: ezychat.io doesn’t shard, and that’s the right call

I’m the primary engineer of ezychat.io, a multi-tenant platform where AI and people answer customers on WhatsApp, Instagram, Messenger and web chat. It’s Python (FastAPI and Celery) on PostgreSQL and Redis, on AWS, declared in Terraform. The honest answer to “where does ezychat use consistent hashing?” is that it doesn’t, and I’d argue against adding it today. This section shows what runs instead, why it’s enough, and what I found when I checked our code against Redis Cluster’s rules.

One Redis, one database

The Terraform declares a single Redis replication group with encryption in transit and at rest and an AUTH token, and a single PostgreSQL instance. No cluster mode, no shards. That one Redis holds the application’s caches, rate-limit counters, Celery’s queues and the pub/sub channels behind real-time updates, all in database 0. Two other services get their own logical databases on the same Redis, and the Terraform comment says why:

# A logical Redis database on the shared node, matching the local compose
# layout (gateway on 5, Evolution on 6) so a key collision cannot happen
# between them or with the application on 0.

Why one node is enough

Because we measured where the limit is, and it isn’t Redis. The handbook records a capacity pass on 2026-08-13, run against a seeded dataset of 50 organizations, 400,000 conversations and 2 million messages. Its findings, in its own words: “The API task is the bottleneck in every scenario. It sits at 100% of its CPU quota from the first concurrent request while Postgres never exceeds ~26% of its two vCPUs.” The Redis row of the same table reads “Not a bottleneck at this volume”. The fix that followed was more API tasks (a CPU autoscaler), not more data nodes.

Sharding a store that isn’t the bottleneck buys nothing and costs a lot: cross-slot rules on every multi-key operation, a client that understands redirects, a reshard procedure to rehearse, and more machines to pay for and watch. The handbook’s scaling plan agrees and puts it late: “Redis Cluster (if > 16 GB memory needed)”, and, for the messaging layer, “Redis Cluster — shard pub/sub channels and rate limit keys (first step)” once Redis itself becomes the limit. Until then the right moves are vertical: a bigger node, and replicas for availability. Neither involves placement.

Tenants share tables, separated by org_id

The database side of “how do tenants scale?” is also not sharding. Every tenant table carries an org_id, and PostgreSQL’s row-level security enforces it. The handbook’s principle: “The application sets app.current_org_id at request entry; the database rejects any query that would return another tenant’s rows.” The code that sets it is one line:

async def set_org_context(session: AsyncSession, org_id: uuid.UUID) -> None:
    await session.execute(
        text("SELECT set_config('app.current_org_id', :org, true)"), {"org": str(org_id)}
    )

Shared tables do have a scaling problem, and it’s the hot-key problem from Advanced wearing a tenant’s clothes. Our load-test tooling is built around it. The docstring of its volume profiles:

Shared-table multi-tenancy means the largest tenant sets the cost of everyone else’s plans.

So each test profile gives one organization a large share of the rows (55% in the profile the capacity pass used), because uniform data hides exactly the failures that matter. That’s also the clearest argument about how ezychat would eventually partition its database. Hashing org_id onto a ring would put one whale tenant on one shard with no way to split it, the same 8%-of-all-reads key from the Zipf experiment. Tenant partitioning, if it ever comes, should be a directory: a table from org_id to shard, where most tenants are placed by hash or by fill level and the biggest ones are placed deliberately, possibly alone. It’s the fixed-slots argument again: an explicit mapping you can read and change beats one you have to compute.

What breaks the day we turn on cluster mode

To make “Redis Cluster later” concrete, I went through the Redis call sites in the backend and checked them against the rules shown in Expert. These would need changes first.

The rate limiter’s script reads two keys. The sliding-window limiter evaluates one Lua script per request over the current and previous window:

current_key = f"rl:{bucket}:{identity}:{window}"
previous_key = f"rl:{bucket}:{identity}:{window - 1}"
try:
    redis = get_redis()
    if _script is None:
        _script = redis.register_script(_SLIDING_WINDOW)
    blocked, count = await _script(
        keys=[current_key, previous_key],
        args=[limit, window_seconds * 2, f"{elapsed:.6f}"],
        client=redis,
    )
except Exception:
    return Verdict(allowed=True, count=0, limit=limit, retry_after=0)

On a cluster, those two keys land in different slots, so the script fails with CROSSSLOT, which is what the local cluster said above for keys of exactly this shape. And look at the except: the limiter fails open on purpose, so that a Redis outage doesn’t become a total outage. The combination means a cluster migration done without tags wouldn’t produce an error page. It would quietly allow every request, and the only symptom would be the absence of 429s. The fix is a hash tag on the caller, so both windows share a slot:

# uses key_slot from the Redis Cluster section above

def rl_keys(bucket: str, identity: str, window: int) -> tuple[str, str]:
    # Today: f"rl:{bucket}:{identity}:{window}". The braces below are literal: a hash tag.
    tag = f"{{{bucket}:{identity}}}"
    return f"rl:{tag}:{window}", f"rl:{tag}:{window - 1}"

current, previous = rl_keys("api_user", "42", 29512345)
print(current, previous)
print("same slot:", key_slot(current) == key_slot(previous))

# A test worth keeping in CI: every multi-key call site builds keys that share a slot.
untagged = ("rl:api_user:42:29512345", "rl:api_user:42:29512344")
print("untagged same slot:", key_slot(untagged[0]) == key_slot(untagged[1]))
# rl:{api_user:42}:29512345 rl:{api_user:42}:29512344
# same slot: True
# untagged same slot: False

Tagging on bucket:identity, not on bucket alone, keeps one limit’s counters for different callers spread across the cluster, so the busiest limit doesn’t become a hot slot.

Batched reads use MGET. Several caches read many keys in one round trip, and they’re batched for a measured reason. The booking availability cache’s docstring:

A search covers a whole horizon of days across every candidate resource, and the per-key version this replaces asked for them one at a time: the round trips — ~2.7 ms each — were the dominant cost of an availability query, serialised so the request held its pooled connection for the duration. One MGET for the reads and one pipeline for the write-back turns that into two, whatever the window.

Its keys are booking:avail:{resource_id}:{date} in the f-string sense, with no literal braces, so one MGET over a horizon of days spans many slots. This path also fails open (“any Redis error degrades to computing everything”), so on a cluster the cache would simply stop hitting, and availability queries would quietly go back to their pre-cache cost. The AI throttle reads three keys for one throttle scope in one MGET, and an inbox cache deletes eight per-organization keys in one DEL. Each needs either a tag on the thing the keys belong to (the resource, the throttle scope, the organization) or a client that splits the batch by slot.

Logical databases don’t exist on a cluster. The WhatsApp gateway on database 5 and Evolution on database 6 would need key prefixes in a shared cluster, or, more simply, their own Redis. The handbook’s scale-out plan already lists dedicated instances as a step.

Pub/sub works, but differently. Real-time updates publish to channels named per organization (org:{org_id}:events, again f-string braces). In Redis Cluster, a classic PUBLISH is broadcast to every node, so pub/sub traffic doesn’t shrink as nodes are added. Redis 7 added sharded pub/sub (SPUBLISH, SSUBSCRIBE), where channels are assigned to slots like keys. That’s the version that would actually scale out, and it’s why the handbook names pub/sub channels, together with rate-limit keys, as what a cluster would shard first.

None of this is a reason not to move to a cluster one day. It’s the work list, and the most dangerous item is the one with no error message. When we do it, the first change I’d make is a test that builds every multi-key call site’s keys and asserts they share a slot, like the last two lines of the snippet above, so a new MGET can’t quietly reintroduce the problem.

A nearby use: hashing into a fixed space

The one place ezychat deliberately hashes keys into a fixed numeric space is PostgreSQL advisory locks. Escalating a comment thread must be serialized per post, and advisory locks are identified by integers, so the code hashes the organization and the post into the two integer keys the lock takes:

# Serialise claim + count for this post. `hashtext` gives the two bigint
# keys the two-argument form wants; transaction-scoped, so it releases on
# commit or rollback without a matching unlock.
await session.execute(
    text("SELECT pg_advisory_xact_lock(hashtext(:org), hashtext(:post))"),
    {"org": str(org_id), "post": post_id},
)

It’s the opposite trade from everything above. Two different posts can collide on the same lock key, and that’s fine: a collision only makes two unrelated escalations wait for each other briefly, never gives a wrong answer. Placement schemes have to get every key to the right owner; a lock key only has to make the same post always map to the same lock. Knowing which of those two jobs a hash is doing tells you how much its collisions matter.

What I’d copy

• Measure before you shard. Our capacity pass found the bottleneck in API CPU, with Redis “not a bottleneck”. The cheapest distributed system is the one you didn’t build yet.
• Write the scaling trigger down. “Redis Cluster if > 16 GB memory” in the handbook means the decision is made by a number, not by a mood.
• Know what will break before it does. Multi-key calls and logical databases are the work list for cluster mode, and the fail-open paths are the ones to check first, because they fail silently.
• Partition tenants with a directory, not a hash, when one tenant can be most of the data.
• Tag keys by what they belong to, at the finest grain that keeps related keys together.

Cheat sheet