Design a Distributed Cache

• Get, set and delete a key — Values are opaque bytes up to a few hundred KB. Set takes an optional TTL. This is the whole hot path and it must be one network round trip.
• Expiry by TTL — A key disappears at its deadline whether or not anyone reads it. Lazy expiry on read plus background sampling, because scanning every key is not affordable.
• Eviction under memory pressure — When the node is full, admit the new key by evicting others. Policy is per namespace: LRU by default, LFU for skewed read patterns, or reject writes for a cache that must not lose data.
• Atomic numeric and structural operations — Increment, compare-and-set, list and set operations. These are why teams pick a cache with data structures over a plain memcached; they also keep counters correct without a read-modify-write race.
• Namespaces with independent limits — One cluster serves many teams. Each namespace has a memory quota, an eviction policy, a default TTL and its own rate limit so one tenant cannot evict another out of memory.
• Online resharding — Add or remove nodes without downtime and without invalidating the whole keyspace. Slots migrate; clients follow redirects while a migration is in flight.
• Replication and failover — Each shard has a primary and at least one replica in another availability zone. A primary failure promotes a replica in seconds without operator involvement.
• Observability per key class — Hit rate, evictions, memory, p99 latency and hot keys, broken down by namespace and key prefix, because "the cache is slow" is otherwise undebuggable.
• Out of scope — Durable storage semantics (this is a cache, not a database), cross-region active-active writes, and a query language over cached values.

• Latency (p99 < 1 ms in-zone) — A cache that is sometimes slow is worse than no cache, because callers have already paid for the lookup. Single-digit milliseconds at p999 including a retry.
• Throughput (10 M ops/s per cluster) — Driven by a fleet of application servers, each doing thousands of gets per second. Per-node ceiling is what sets the cluster size.
• Capacity (10 TB of hot data) — Memory is the expensive resource; the design lives or dies on bytes per entry and on eviction picking the right victims.
• Availability (99.99 % for reads) — The cache is on the critical path of every product. Losing it must degrade to a slower origin, never to an outage, which pushes work onto the client library.
• Consistency (best-effort, bounded staleness) — A cache may return stale data and may lose data on failover. This is the requirement that lets the rest of the design be fast, and it must be stated out loud.
• Durability (none guaranteed) — A cold start after losing a whole cluster must not take the origin down; warming is part of the design rather than an afterthought.
• Isolation (no cross-tenant blast radius) — A tenant that writes 2 GB/s or fills memory with 10 MB values must be throttled, not allowed to evict everyone else.
• Elasticity (resize in minutes, no cold cache) — Traffic doubles for a sale; adding nodes must not invalidate the existing keys, which is the argument for consistent hashing over modulo.

• Operations per second: ~10 M — 5 000 application servers × 2 000 cache ops/s each = 10 M ops/s. Reads are roughly 20:1 over writes, which is what makes replicas useful for read offload.
• Nodes needed for throughput: ~60–100 — A single-threaded Redis-style process on modern hardware handles ~150 k ops/s with pipelining and small values; budget 120 k to leave headroom. 10 M ÷ 120 k ≈ 84 primaries, round to 100 for growth.
• Memory for the working set: ~13 TB usable — 10 TB of values plus per-entry overhead: key (~40 B) + value header, TTL and LRU metadata (~50–100 B) + allocator fragmentation (~10–20 %). 10 TB × 1.3 ≈ 13 TB of RAM before replication.
• Machines for memory: ~90 primaries — 13 TB ÷ ~144 GB of usable cache memory per machine (a 192 GB box, leaving headroom for replication buffers and a fork) ≈ 90. With one replica each, ~180 machines. Memory, not CPU, is the binding constraint at this value size.
• Network per node: ~1.2 Gbps — 120 k ops/s × ~1.2 KB average value ≈ 144 MB/s ≈ 1.2 Gbps on a 10–25 Gbps NIC. Fine, until someone caches 1 MB objects: the same op rate would need 120 GB/s, which is why per-value size limits exist.
• Hit rate and origin load: 95 % → 500 k/s to origin — At 10 M ops/s with a 95 % hit rate the origin databases absorb 500 k/s. Moving to 99 % cuts that to 100 k/s. Every point of hit rate is worth five times its weight in database capacity, which is the argument for spending on memory.
• Cold start cost: ~20 min at safe refill — Refilling 10 TB from origin at a rate the databases survive (say 200 k fills/s at 1.2 KB = 240 MB/s) takes 10 TB ÷ 240 MB/s ≈ 12 h for everything — but only the hot 5 % matters, so ~20 minutes to get back above a 90 % hit rate. The lesson: never restart a whole cluster at once.
• Failover impact: ~1 % of keys, seconds — One primary of 100 holds ~1 % of the keyspace. A failure loses whatever was not replicated (milliseconds of writes with async replication) and makes that slice miss for the 5–15 s of detection and promotion. Expressed as origin load: a 5 % bump for a few seconds.

• Application (smart client library) — Runs the cache-aside pattern and holds the cluster topology in memory, so a get is one hop to the right node rather than a proxy round trip. Owns the tight timeouts, the single retry, the circuit breaker and the fallback to origin. Also does client-side batching and, for the very hottest keys, a small in-process cache.
• Cache proxy (optional · thin routing tier) — For clients that cannot embed a smart library (other languages, serverless, external teams). Terminates connections, keeps pools warm to every node, and does the same slot routing. Costs a hop of latency, saves a connection explosion: 5 000 clients × 100 nodes is 500 k connections without it.
• Cache node (in-memory · single-threaded core) — Owns a set of hash slots. Keeps an in-memory hash table with per-entry TTL and LRU/LFU metadata, serves gets and sets from the event loop, and handles its own expiry sampling and eviction. Rejects values over the size limit and tracks per-namespace memory so one tenant cannot consume the node.
• Replica (async replication · other AZ) — Streams writes from its primary and can serve reads for clients that accept staleness. Exists mainly so a machine loss costs seconds rather than a cold slice, and so a planned restart is a promotion rather than an outage.
• Topology store (etcd / ZooKeeper · slot map) — The small strongly consistent store of truth: slot → primary, primary → replicas, migration state and epoch. Read on client start and watched for changes; never on the request path. A few kilobytes of data protected by consensus.
• Control plane (placement · resharding) — Decides where slots live. Runs cluster creation, scale up and down, slot migration, rebalancing after failure, and rolling upgrades one node at a time. Every action is a change to the slot map plus orchestration of the nodes involved, so the data path stays simple.
• Failure detector (gossip + quorum) — Nodes gossip liveness; a primary is declared dead only when a quorum of peers agrees and the control plane confirms. Deliberately conservative: a false positive causes an unnecessary failover and a small data loss, so detection is tuned for a few seconds rather than a few hundred milliseconds.
• Origin store (the database being protected) — The source of truth behind the cache. The design exists to keep this from being hit 10 M times a second; every failure mode is judged by how much load it pushes here.
• Warmer (top-key replay) — Rebuilds the hot set after a cold start. Consumes the sampled key log, replays the most frequent keys against the origin at a rate the origin can absorb, and reports hit rate as it climbs. The reason a restart is not an incident.
• Key sample stream (sampled · 1 in 1000) — A 0.1 % sample of operations emitted by the nodes: key prefix, namespace, size, hit or miss. Feeds hot-key detection, working-set estimation and the warmer. Sampling keeps this at a few thousand events a second instead of ten million.
• Hot-key detector (count-min sketch) — Maintains approximate top-K per namespace from the sample stream. When a key exceeds a threshold share of a node, it publishes it to clients, which then cache it locally for a second or two, and to the control plane, which can replicate that slot more widely.
• Metrics & quotas (per namespace) — Aggregates hit rate, memory, evictions, expirations, latency percentiles and bandwidth per namespace and key prefix. Enforces quotas by telling nodes to throttle or reject a namespace that is over its memory or bandwidth budget, which is what keeps tenants isolated.
• Dashboards & alerts (on-call view) — Where hit-rate regressions, eviction storms, memory fragmentation and failovers surface. The metric people page on is client-observed latency and error rate, not node CPU.

1. A cache hit, and a miss that fills — The path that runs ten million times a second. What matters is that a hit is one hop with no coordination, and that a miss cannot turn into a stampede.
   1. Client hashes the key to a slot and picks the node that owns it from its local topology map.
   2. Client sends the get on a pooled connection and waits with a tight timeout.
   3. Node looks the key up in its hash table, checks the TTL, and updates the recency metadata.
   4. On a miss, the client reads the origin under a per-key lock so only one caller does the work.
   5. Client writes the value back with a jittered TTL, and the node admits or evicts to make room.
   6. Callers that cannot embed the library send the same operation through the proxy tier instead.
   7. Node samples the operation into the key stream for hot-key and working-set analysis.
2. A write and its invalidation — Keeping a cache close to correct is mostly about what the writer does. The chosen pattern is write-through to the database and delete the key, with a version to block a stale fill.
   1. Application writes the new value to the origin database in a transaction.
   2. Application deletes the cache key rather than overwriting it.
   3. The next reader misses, reads the origin, and stores the value with its version.
   4. Node replicates the mutation to its replica asynchronously.
   5. A delayed second delete guards against a fill that raced the write.
3. Add nodes without losing the cache — Scaling up must not invalidate the keyspace. Slots move one at a time, and clients are redirected mid-migration rather than being told to stop.
   1. Operator asks the control plane to grow the cluster; it computes the slots to move.
   2. Control plane marks a slot as migrating: source is MIGRATING, destination is IMPORTING.
   3. Keys in the slot are copied in batches while the source continues serving reads and writes.
   4. A client that asks the source for a key already moved is redirected to the destination.
   5. When the slot is empty, the control plane commits the new owner into the topology store and bumps the epoch.
   6. Migration continues slot by slot with a rate limit while the metrics show hit rate holding.
4. A primary dies mid-traffic — The failure case that decides whether the cache is safe to depend on. Detection is deliberately slow, promotion is fast, and the client absorbs the gap.
   1. Peers stop receiving gossip from the node and mark it suspect.
   2. A quorum of peers agrees the node is down and the failure detector tells the control plane.
   3. Meanwhile clients time out, retry once against the replica, and then treat the key as a miss.
   4. Control plane promotes the replica, assigning it the failed primary's slots with a new epoch.
   5. Clients watching the topology store switch within a second; stragglers learn through MOVED.
   6. A fresh replica is built for the new primary in the background, rate limited so it does not disturb reads.
   7. The warmer refills whatever the promoted node is missing, replaying the top keys against the origin at a safe rate.
5. One key takes a whole node down — The failure mode unique to caches: a single key so hot that its owning node saturates while the rest of the cluster is idle. Sharding cannot help, because the key is indivisible.
   1. A product launch makes one key the target of 2 M requests per second.
   2. Node latency and bandwidth for that namespace spike, and the alert fires before anyone files a ticket.
   3. The sample stream shows one prefix dominating; the detector flags it as hot.
   4. The hot key is published to clients, which cache it in process for a second or two.
   5. For keys that cannot tolerate local staleness, the control plane replicates the slot more widely and clients read from replicas.
   6. The client library also spreads the key across N aliases when writes are rare.
   7. Quotas throttle the namespace if it is still saturating the node, protecting the other tenants.

Ask: How does a request find the node that owns its key: a smart client, a proxy, or something in the middle?

Good answers name: Smart client with a cached slot map, Proxy tier (twemproxy, envoy-style), Server-side redirection only, Consistent hashing in the client with no slot map.

Our pick: A smart client holding a 16 384-entry slot map, watched from the topology store and corrected by MOVED/ASK redirects, with an optional thin proxy for clients that cannot embed the library.

1. What happens when a client has a stale slot map during a migration?
   It asks the old owner, which answers ASK (this request only, slot still moving) or MOVED (slot has moved, update your map). ASK is followed for a single request and does not change the map; MOVED updates it. Either way the operation succeeds on the first retry, so staleness costs latency rather than correctness.
2. Why 16 384 slots rather than hashing keys directly to nodes?
   Slots decouple the key space from the node count. The map is small enough to ship to every client and to gossip as a bitmap, placement decisions are made over a few thousand units instead of billions of keys, and a migration is a well-defined "move slot 4211" operation with a clear start and end.
3. How do you avoid 500 k connections?
   Long-lived pooled connections with a small pool per node per client process (2–4), multiplexing and pipelining on each, plus the proxy tier for fleets where that is still too many. Connection count is a first-class capacity number: each one costs memory and a file descriptor on the node.
4. A client library bug sends 10x traffic. What limits the damage?
   Per-namespace quotas enforced at the node, per-connection rate limits, and a circuit breaker inside the library itself. The node also has a hard cap on concurrent clients and will reject rather than queue unboundedly, since a queued request at a 5 ms timeout is already useless.

Ask: When memory is full, which key is evicted, and how is that decided without scanning everything?

Good answers name: Approximate LRU by sampling, Approximate LFU with decay, TTL-only, no eviction (reject writes when full), Exact LRU with a linked list.

Our pick: Approximate LRU by sampling as the default, LFU for read-heavy skewed namespaces, and no-eviction for namespaces whose entries are correctness-relevant, with the policy and memory quota set per namespace.

1. How does expiry work without scanning the keyspace?
   Two mechanisms together: lazy expiry deletes an expired key when it is read, so no reader ever sees one; and a background loop samples a small number of keys with TTLs every cycle, deletes the expired ones, and repeats more aggressively while the observed expired fraction is high. Memory is therefore reclaimed proportional to how much has actually expired.
2. A batch job scans a million cold keys and evicts the hot set. What stops that?
   That is exactly what LFU prevents, because a single access never promotes a key above entries with high counters. Additional defences: the batch uses its own namespace with its own quota, and clients can pass a hint that marks reads as non-promoting.
3. How do you size memory so evictions are rare but memory is not wasted?
   From the working set, not the total data: sample the key stream, estimate the number of distinct keys touched in a window, multiply by average entry size plus overhead, and add headroom. Then watch the eviction rate and the hit rate together — evictions climbing while hit rate holds means you are evicting genuinely cold data, which is healthy.
4. What about memory fragmentation?
   Allocators with size classes leave gaps when values vary in size, so resident memory can exceed the data by 20–50 % in the worst cases. Mitigations: keep values within a few size classes where possible, cap the maximum value size, monitor the fragmentation ratio, and use active defragmentation or a rolling restart of the replica-then-primary when it gets bad.

Ask: How is the cache kept consistent enough with the origin, and what exactly is promised to callers?

Good answers name: Write to origin, delete the key, version the stored value, Write-through: update cache and origin together, Invalidate from the database change stream (CDC), Short TTLs and accept staleness.

Our pick: Write the origin, delete the key, and store a version with every cached value so a slow fill cannot overwrite a newer one, with TTL jitter on every entry and a CDC-driven invalidation stream as a safety net on the key classes that matter.

1. Walk me through the stale-fill race in detail.
   Reader R misses and reads version 41 from the origin. Writer W commits version 42 and deletes the key. R then stores version 41 with a ten-minute TTL, and every subsequent reader sees stale data until it expires. The version check fixes it: R's store is conditional on the cached version being lower than 41, and once 42 has been stored the write is refused. The delayed double delete covers the case where nothing has been stored yet.
2. When would you not cache at all?
   When a stale read causes a wrong action: an inventory count at checkout, a balance before a transfer, a permission check immediately after a revoke. For those, read the primary, or cache with a TTL measured in a few seconds and accept the origin load. Saying which data is deliberately uncached is part of the answer.
3. How do you give a user read-your-writes without slowing everyone down?
   Pin that user briefly: after their write, either keep their new value in their session, or mark their key with a short "recently written" marker so their own reads bypass the cache for a few seconds. It is a per-user cost rather than a global consistency guarantee.
4. The cache and the origin disagree and nobody knows why. How do you debug it?
   A sampled comparison job: read a small number of keys from both and report mismatches by key prefix. Combined with the version stored in each value, it tells you whether the problem is a missed invalidation (cached version older than origin) or a stale fill (version older than a write you can find in the log). Without the version in the value, this is guesswork.

Ask: Should a cache replicate at all, and if so, synchronously or asynchronously?

Good answers name: Asynchronous replica per primary, cross-AZ, No replication, rely on refill, Synchronous replication, Persist to disk (AOF/snapshot) instead of replicating.

Our pick: One asynchronous replica per primary in a different availability zone, with quorum-based failure detection, epoch-fenced promotion, and an explicit written promise that a failover may lose the last few milliseconds of writes.

1. How do you avoid two primaries after a partition?
   Promotion requires a quorum of peers to agree the primary is gone and is recorded in the consensus-backed topology store with a higher epoch. A returning old primary sees the newer epoch and demotes itself, and clients fence on the epoch so a write sent to the old primary is rejected. Heartbeats alone cannot prevent split brain; the epoch and the quorum do.
2. Why is detection deliberately slow?
   Because a 2-second GC pause, a brief network drop and a dead machine look identical at first. Failing over on a false positive costs a real data loss window and a load spike, so the detector waits several missed intervals and requires agreement. The client absorbs the gap by degrading to the origin, which is cheaper than an unnecessary failover.
3. Can replicas serve reads?
   Yes, for callers that accept staleness of a few milliseconds, and it is the second lever for a hot slot. The catch is that a caller reading its own write from a replica may not see it, so the client library only routes to replicas for key classes marked stale-tolerant.
4. What does a rolling upgrade look like?
   Upgrade the replica, verify it is in sync, promote it, upgrade the old primary as the new replica, and move on one shard at a time. The failover is planned rather than triggered by failure, so the write-loss window is controlled, and the cluster never has more than one shard in a degraded state.

Ask: The cache is empty, or a hot key just expired. How do you stop the origin from falling over?

Good answers name: Per-key locking plus stale-while-revalidate, Probabilistic early expiration, Warmer job replaying top keys after a cold start, Origin rate limiting and load shedding.

Our pick: All three layers: locking with stale-while-revalidate in the client for the steady state, a warmer driven by the sampled key log for cold starts, and origin-side limits with a degraded response so the worst case is slow rather than down.

1. How do you restart a hundred-node cluster safely?
   Never all at once. Restart is a rolling promotion — upgrade the replica, promote it, restart the old primary — so at most one shard is cold at a time and the cluster-wide hit rate barely moves. A genuine full cold start is handled by the warmer with the origin rate-limited, and the order is by descending key popularity.
2. What is stale-while-revalidate here, exactly?
   Each entry carries a soft TTL and a hard TTL. After the soft TTL the value is still served, but one caller takes the lock and refreshes it in the background. Only after the hard TTL is the value treated as gone. Users see a slightly stale value instead of an origin round trip, and the origin sees one request instead of thousands.
3. A key genuinely does not exist and something probes it a million times. What happens?
   Negative caching: store a tombstone with a short TTL, 5 to 30 seconds, so the miss is answered by the cache. The short TTL matters because a record created moments later must not be hidden. A Bloom filter of existing keys is the cheaper option when the key space is enumerable.
4. How do you test that any of this works?
   Deliberately, in production, during working hours: kill a node and watch origin load and client latency; flush a namespace on a canary and watch the warmer climb; and run a load test against the origin with the cache disabled to find out what its real ceiling is. A degradation path that has never been exercised is a guess.

Ask: How do you stop one team from ruining the cache for everyone, without giving every team its own cluster?

Good answers name: Namespaced quotas on one shared cluster, A dedicated cluster per team, Cells: a handful of shared clusters, tenants assigned to one, No isolation, rely on good behaviour.

Our pick: A small number of cells, each a shared cluster with per-namespace memory quotas, value-size limits, bandwidth limits and eviction policy, plus dedicated clusters for the few tenants that genuinely need them.

1. How is a memory quota enforced without a scan?
   The node tracks bytes per namespace incrementally on every set, delete, eviction and expiry. When a namespace is over quota, its writes evict only from its own entries — so it cannot push another tenant out of memory — and if its policy is no-eviction, its writes are rejected instead.
2. A tenant caches 5 MB values. What breaks and what do you do?
   Bandwidth and tail latency break first: a few hundred of those per second saturate the NIC and every other operation on that node queues behind them. The fix is a hard per-value size limit (say 1 MB) enforced at the node with a clear error, plus a suggestion to store the blob in object storage and cache the key. Large values belong in a different system.
3. How do you charge teams for what they use?
   Bill on the peak memory footprint and the bandwidth per namespace, both of which the node already tracks for quotas. Publishing those numbers changes behaviour more than any technical control: teams find their own 200 GB of keys with no TTL once they can see them.
4. One namespace needs strong isolation for latency. What do you offer?
   Either a dedicated cell, or dedicated nodes within a cell by pinning its slots to machines that serve no other namespace. The second option keeps one control plane and one upgrade path, which is usually the better trade for everyone involved.