Design a distributed cache

A distributed cachedistributed cache: A cache run as a separate service, such as Redis, Valkey or Memcached, that every application instance reads over the network: one copy for the whole fleet, kept through application restarts and deploys. More in “Where caches live, from browser to database” keeps copies of frequently read data in the memory of many machines, so that most reads are answered in well under a millisecond and never reach the database. In the design below, 10 primaries and their 10 replicas answer a million operations a second in front of a database that can serve 40,000 reads a second.

The design spreads keys over nodes so that capacity grows with the cluster, keeps a replica of each node so that one failure does not empty part of the cache, evicts entries when memory runs out, and stays close enough to the database that readers do not see stale data for long. Memcached, Redis Cluster and Valkey, a BSD-licensed open-source key-value store (Valkey), are the usual building blocks.

Requirements and estimate: throughput, not memory, sets 10 primaries

A shopping app serves product pages, prices and sessions from a database that can answer about 40,000 reads a second. At peak the app makes a million cache operations a second. Before drawing anything, ask:

• What is cached, and what is the source of truth for it?
• How large are the items, and how many keys are there?
• What share of operations are reads, and how stale may a read be?
• May a cache node lose its contents, or must the cache survive restarts?
• One region or several, and what happens to the database if the cache is empty?

The answers here: 200 million keys of about 1 KB each, 90 % reads, a price may be stale for a few seconds at most, the cache can be rebuilt from the database, and there is one region for now.

Functional requirements: get, set with a time to live (TTL)TTL (cache entry): Time to live: how long a cache may serve an entry after storing it. When it runs out the next read is a miss, so even a copy that a missed delete left behind disappears. More in “Cache-aside, read-through, write-through, write-back”, delete, and multi-get; eviction when memory is full. Non-functional requirements: p99 latency under 1 ms inside a zone, a million operations a second, no outage when one node fails, room to grow, and a database that is never sent more reads than it can serve.

The first estimate, with every assumption printed:

# A first capacity plan for the distributed cache of this lesson. Every input is an assumption to replace with
# measurements: item sizes from a sample of real keys, overhead from the cache's own memory statistics, and the
# operations a node serves within the latency target from a load test on the hardware you will use.
KEYS = 200_000_000
BYTES_PER_ITEM = 1_000          # key and value together, on average
OVERHEAD = 0.25                 # per-key structures and allocator fragmentation, on top of the data
NODE_RAM_GB = 64
MEMORY_LIMIT = 0.75             # share of RAM given to data; the rest is for buffers, growth and snapshots
OPS_PER_SECOND = 1_000_000      # 90 % reads
OPS_PER_PRIMARY = 100_000       # within a p99 of 1 ms
GB = 10**9

data = KEYS * BYTES_PER_ITEM / GB
total = data * (1 + OVERHEAD)
usable = NODE_RAM_GB * MEMORY_LIMIT
for_memory = -(-total // usable)                       # ceiling division
for_throughput = -(-OPS_PER_SECOND // OPS_PER_PRIMARY)
primaries = int(max(for_memory, for_throughput))

print(f"data: {KEYS:,} keys × {BYTES_PER_ITEM:,} bytes = {data:,.0f} GB")
print(f"with {OVERHEAD:.0%} overhead: {total:,.0f} GB")
print(f"usable per {NODE_RAM_GB} GB node at a {MEMORY_LIMIT:.0%} limit: {usable:.0f} GB -> {for_memory:.0f} primaries for memory")
print(f"{OPS_PER_SECOND:,} operations a second at {OPS_PER_PRIMARY:,} per primary -> {for_throughput} primaries for throughput")
print(f"plan: {primaries} primaries holding {total / primaries:.0f} GB each, plus one replica each = {2 * primaries} nodes")
print(f"network: {OPS_PER_SECOND * BYTES_PER_ITEM / GB:.1f} GB/s in total, "
      f"{OPS_PER_SECOND * BYTES_PER_ITEM * 8 / GB / primaries:.1f} Gbit/s per primary")
print(f"one primary lost: {1 / primaries:.0%} of the keys miss until its replica takes over")

data: 200,000,000 keys × 1,000 bytes = 200 GB
with 25% overhead: 250 GB
usable per 64 GB node at a 75% limit: 48 GB -> 6 primaries for memory
1,000,000 operations a second at 100,000 per primary -> 10 primaries for throughput
plan: 10 primaries holding 25 GB each, plus one replica each = 20 nodes
network: 1.0 GB/s in total, 0.8 Gbit/s per primary
one primary lost: 10% of the keys miss until its replica takes over

Memory alone asks for 6 primaries, but throughput asks for 10, so throughput decides the size. Losing one primary sends a tenth of all reads somewhere else, and that number drives the failure design. Try the same arithmetic on another catalogue:

The first design: every key has one primary

Every key belongs to one of 16,384 hash slotsOne of the 16,384 numbered buckets that a Redis-style cluster maps every key into. Each slot is served by one primary, and a node that does not serve a key's slot redirects the client to the node that does., and every slot to one primary. Follow one read and one write of a product through the design, from the app server to the database:

The design leaves several choices open, each with one main option:

• Client-side hashing: the client library maps a key to a node itself.
• A proxy tier: clients send every request to proxies that know where keys live.
• Hash slots with redirects: keys map to 16,384 slots; a node that does not own a slot redirects the client.
• Asynchronous replicas: each primary copies its writes to a replica, which takes over when the primary fails.
• Eviction policies: rules that choose which entries to drop when memory is full.
• Leases: tokens that stop a slow reader from writing an outdated value back into the cache.
• A standby pool: a few spare nodes that take over the keys of a failed node.

The deep dives that decide whether this design works:

1. How does a client find the node for a key, and what happens while slots move to a new node?
2. What can a failover lose, and how do you stop a deleted value from coming back?
3. How much memory should each node use, which eviction policy fits, and what about one very hot key?
4. How does the cache stay consistent with the database when writes and reads race?
5. What happens to the database when a node dies, or when the whole cluster starts empty?
6. How does a second region get a cache without slow reads across regions?

An answer is judged on requirements and estimates, partitioning and routing, replication and failover, memory and eviction, consistency with the database, and failure handling.

The API: five calls, and a TTL on every key Pro: opens with a Pro, Premium or Ultimate pass