Design: Distributed Cache¶

Difficulty: Foundation / Senior | Time: 60–75 minutes

1. Problem Statement¶

Design a distributed, in-memory cache system — the kind of thing a company builds (or buys, as Redis/Memcached) to sit in front of a slower primary data store. Clients GET, SET, and DELETE keys; the cache is horizontally scaled across many nodes and survives individual node failures without becoming globally unavailable.

2. Clarifying Questions¶

Practice asking these before designing:

3. Functional Requirements¶

• GET(key) → value — retrieve a value by key
• SET(key, value, ttl?) — store a value, optionally with a TTL
• DELETE(key) — remove a value
• Keys expire automatically after their TTL
• Optional: atomic increment/decrement, batch get (MGET)

4. Non-Functional Requirements¶

5. Capacity Estimation¶

Data:
  500M keys × avg value size 2KB (mix of small session objects, larger fragments)
  Total data ≈ 500M × 2KB = 1 TB raw
  With per-key overhead (~100 bytes metadata: TTL, pointers) ≈ 1.05 TB
  Target 10 TB headroom (growth + working set larger than "hot" subset) → provision for 10 TB

Ops:
  2M ops/second peak, ~80:20 read/write ratio
  Reads:  1.6M ops/sec
  Writes: 0.4M ops/sec

Nodes:
  A single cache node (e.g., r6g.2xlarge class, 64GB RAM) can serve
  ~100K-150K ops/sec and hold ~50GB of usable data (after overhead/headroom)

  By memory:  10 TB / 50 GB per node  ≈ 200 nodes
  By throughput: 2M ops/sec / 120K ops/sec per node ≈ 17 nodes

  Memory is the binding constraint → provision ~200 nodes,
  which comfortably covers the throughput requirement too (200 × 120K = 24M ops/sec capacity)

Network:
  Avg value 2KB × 2M ops/sec ≈ 4 GB/sec cluster-wide bandwidth
  Per node: 4 GB/sec / 200 nodes ≈ 20 MB/sec — trivial for modern NICs

6. API Design¶

GET /keys/{key}
Response: { "value": "...", "ttl_remaining": 42 }
Status: 200 OK, or 404 Not Found (cache miss)

PUT /keys/{key}
Request:  { "value": "...", "ttl_seconds": 300 }
Response: 200 OK

DELETE /keys/{key}
Response: 204 No Content

POST /keys/batch-get
Request:  { "keys": ["a", "b", "c"] }
Response: { "a": "...", "b": null, "c": "..." }

In practice, clients talk to the cache over a lightweight binary protocol (RESP for Redis, a custom protocol for Memcached) rather than HTTP — HTTP's per-request overhead is too high at sub-millisecond latency targets. The REST shape above is useful for reasoning about the interface; assume a smart client library that maintains persistent connections and speaks the wire protocol underneath.

7. High-Level Design¶

graph TD
    App1[App Server 1] --> Client[Smart Client\nconsistent-hash routing]
    App2[App Server 2] --> Client
    App3[App Server 3] --> Client

    Client -->|hash key → node| N1[Cache Node A\nprimary shard 1]
    Client -->|hash key → node| N2[Cache Node B\nprimary shard 2]
    Client -->|hash key → node| N3[Cache Node C\nprimary shard 3]

    N1 -.replicate.-> N1R[Node A Replica]
    N2 -.replicate.-> N2R[Node B Replica]
    N3 -.replicate.-> N3R[Node C Replica]

    Coord[Cluster Coordinator\nZooKeeper / etcd / gossip] -.membership + shard map.-> Client
    Coord -.health checks.-> N1
    Coord -.health checks.-> N2
    Coord -.health checks.-> N3

    N1 -->|cache miss| DB[(Primary Database)]
    N2 -->|cache miss| DB
    N3 -->|cache miss| DB

Three moving parts do all the work: the client's routing logic (which node owns this key), the shard's own storage engine (an in-memory hash table with TTL eviction), and the coordinator (tracks which nodes exist and are healthy, so routing stays correct as nodes join/leave).

There are two deep sub-problems that make this design non-trivial, both covered below: how you route a key to a node without a full reshuffle every time the cluster resizes, and what you do when a node goes away.

8. Deep Dive: Partitioning Strategy¶

Consistent hashing solves this: nodes and keys are placed on a hash ring (typically hash % 2^32), and a key is owned by the first node clockwise from its position. Adding or removing one node only remaps the keys between that node and its neighbor — roughly 1/N of the keyspace, not all of it.

The full mechanics — virtual nodes to prevent hot spots, ring rebalancing, and the "cascading failure on removal" gotcha — are covered in Consistent Hashing. The short version for this design: each physical node is placed on the ring as 100-200 virtual nodes so that when a node leaves, its keyspace is spread evenly across many surviving nodes rather than dumping the whole load onto one unlucky neighbor.

200 physical nodes × 150 virtual nodes each = 30,000 points on the ring
Losing one physical node redistributes ~1/200th of keys,
spread across ~150 different neighbors instead of 1 — no single node absorbs a spike

9. Deep Dive: Replication for Availability¶

A pure partition-with-no-replication design means losing one node loses that node's entire slice of the keyspace — for a cache, that's "acceptable" in the sense that the DB still has the data, but it produces a synchronized wave of cache misses hitting the database for every key that node owned, which can itself take down the database (see Thundering Herd in the failure library).

Design: each shard has a primary and 1-2 replicas, placed on different physical nodes (and ideally different availability zones). Writes go to the primary; the primary asynchronously propagates to replicas.

Recommended: async replication, because the failure mode it accepts (losing the last few milliseconds of writes on a primary crash) is cheap — the application re-fetches from the database and repopulates the cache. This is the core trade-off that makes caches architecturally different from databases: you can choose the fast, lossy option because there's always a slower, durable fallback underneath.

On primary failure, the coordinator promotes a replica to primary (see Failure Handling below) and updates the shard map so clients route to the new primary.

10. Deep Dive: Eviction, Hot Keys, and Invalidation¶

Eviction policy¶

Memory is finite; when a node's memory fills, something has to go before a new SET can be accepted.

Real systems (Redis) implement approximate LRU — sampling a handful of keys and evicting the oldest among the sample — because exact LRU requires a full linked-list update on every access, which is too expensive at millions of ops/second.

Hot keys¶

A single key (a viral post, a trending product) can receive far more traffic than any one node — even with perfect partitioning — can serve, because consistent hashing distributes keys evenly, not traffic per key.

Mitigations:

• Client-side local caching: cache the hottest keys in the application server's local process memory for a few seconds, absorbing the majority of requests before they even reach the cache tier.
• Key replication: detect keys crossing a request-rate threshold and replicate them to multiple nodes (not just the primary/replica set), so reads for that one key are spread across N nodes instead of 1.
• Read-through fan-out: for extremely hot keys, shard the key itself into key:0 through key:9 and have clients pick a shard randomly on read, writing to all shards on update.

This is the same problem described generally as Hot Partition / Hot Key in the failure library — a distributed cache is one of the places it shows up most visibly, because the whole point of the system is to serve very skewed access patterns fast.

Invalidation across a cluster¶

When the underlying data changes, stale cache entries need to go. Three strategies, in increasing order of consistency and decreasing order of simplicity:

1. TTL-only: don't invalidate explicitly, just accept staleness up to the TTL window. Simplest, works for most product surfaces where a few seconds of staleness is invisible to users.
2. Explicit delete on write: the write path calls DELETE on the cache key whenever the underlying row changes. Requires every write path to remember to do this — a common source of stale-cache bugs is a new write path that forgets.
3. Pub/sub invalidation broadcast: the primary datastore (or a CDC stream off it) publishes invalidation events; all cache nodes (or a dedicated invalidation service) subscribe and evict matching keys cluster-wide. More moving parts, but removes the "did every write path remember to invalidate" foot-gun.

11. Failure Handling¶

12. Consistency Considerations¶

• Cache is never the source of truth — the database (or whatever system the cache fronts) always wins on conflict. This is what makes async replication and best-effort durability acceptable choices throughout this design.
• Read-your-writes is not guaranteed by default: a client that writes to shard primary A and then reads might get routed to a replica that hasn't caught up yet. If a specific use case needs read-your-writes (e.g., a user immediately viewing their own just-updated profile), route that client's reads to the primary for a short window after their write, or read from the primary unconditionally for that key type.
• Cross-key consistency is out of scope for a cache like this — if an application needs multiple keys to update atomically, that's a transactional database concern, not something the cache layer should be asked to solve. Say this explicitly in an interview: it signals you know where the boundary of "cache" ends and "database" begins.

13. Trade-offs¶

14. Interview Follow-ups¶

1. "How would you handle a value too large to fit comfortably in one node's memory?" — Chunk large values across multiple keys/nodes, or route large-object storage to a separate blob store with the cache holding only a reference/pointer.
2. "How do you resize the cluster without a big latency spike?" — Add virtual nodes to the ring gradually, throttle the rate of key migration between nodes, and let the old node continue serving migrated keys as a fallback until migration completes (dual-read during transition).
3. "How would you add multi-region support?" — Each region runs its own independent cache cluster (cache is cheap to duplicate since it's not authoritative); avoid cross-region cache replication for a plain KV cache — the latency cost isn't worth it when each region can independently repopulate from its local database replica on a miss.
4. "What changes if this needs to support pub/sub or a message queue, not just KV?" — That's a materially different consistency and delivery-guarantee problem (ordering, at-least-once vs exactly-once, consumer groups) — flag explicitly that you'd treat it as a different system, not an extension of this one, even if it reuses the same node infrastructure.

Self-Assessment¶

• [ ] Can I explain why hash(key) % N breaks down and what consistent hashing fixes?
• [ ] Can I justify async over sync replication given the NFRs (eventual consistency, best-effort durability)?
• [ ] Can I describe at least two distinct mitigations for a hot key?
• [ ] Can I walk through what happens end-to-end when a primary node dies mid-traffic?
• [ ] Can I explain why a fully-down cache cluster is a circuit-breaker problem for the application, not just a cache problem?