System Design Case Study
Distributed cache
After lunch, Instagram and ChatGPT should feel instant, not like waiting for a cold database round trip. Underneath, app servers ask a shared in-memory cluster: do we already have this object?
A distributed cache is fast shared memory across many machines. The design problem is not "put Redis somewhere". It is how keys find a , expire with , survive failover, and avoid one melting a shard.
What happens on a read?
Hit memory when possible; miss goes to DB.
CRC16 maps the key to a slot owner.
Jitter so keys do not die together.
Replicas follow; failover may be briefly stale.
Try it: distributed cache hit, miss, and invalidation
Clients talk to a cache tier in front of the database. Invalidation and TTLs decide how stale a hit can be.
- Client
- Cache node
- Peer / replica
- Database
- Invalidation bus
- Hits
- 0
- Misses
- 0
- Mode
- hit
- Path hop
- 0
Hit ratio
hot keys0 hits / 0 misses.
Stampede
singleflightCoalesce concurrent misses for one key.
Consistency
invalidateTTL alone may serve stale values until expiry.
Drive hits, misses, and invalidation to see where stampede risk appears.
What actually happens
With , the app reads the cache first. On a miss it loads the database, writes the value back with a , and returns. Clients must know which node owns the key: cluster mode uses 16384 and may reply when topology changes.
Memory is finite, so policies decide what disappears when maxmemory is hit. Async keeps replicas warm for reads and failover, but the last unreplicated writes can vanish. Persistence via and/or is optional, only if the product needs restart recovery beyond "reload from DB."
The sharp edges are a , a after expiry, and treating the cache as source of truth for money or inventory.
How to choose the cache design
1. Is this an optimization or primary store? Sessions and product pages usually use . Locks and queues need different durability and rules.
2. Can one key go viral? Plan for a with local caches, replicas, or coalescing: average QPS math lies.
3. What happens on failover? Async may lose recent writes. Say whether the product accepts that, or needs WAIT / stronger paths.
Simple rule of thumb
Start with cache-aside, TTL with jitter, and a clear maxmemory policy. Add cluster slots when one box cannot hold the working set or the QPS.
In simple terms: make the common read cheap, keep the client topology-aware, and never pretend a cache is a bank ledger.
Case-study upgrade
Distributed Cache Cluster Math
A distributed cache is fast shared memory spread across nodes. The design must explain key routing, memory overhead, replication, hot keys, and failover.
Assumptions and formulas
- Throughput: 10M ops/sec. Logical cache: 1 TB. Replication factor: 3. Node memory: 64 GB with 80% usable.
- Average operation request plus response: 500 bytes.
- Hot keys can saturate one shard even if cluster averages look safe.
Concrete math
- Cluster network: 10M ops/sec x 500 bytes = 5 GB/sec before replication fan-out.
- Memory: 1 TB logical x 3 replicas = 3 TB data; with 20% to 50% overhead, plan 3.6 to 4.5 TB physical.
- Node count: 64 GB x 80% = 51 GB usable. 1 TB with 3x replication and overhead needs about 80 to 90 nodes.
- Balanced throughput: 100 nodes x 100K ops/sec = 10M ops/sec only if keys are evenly distributed.
- Connections: 100K connections x 10 KB state = about 1 GB cluster memory, but event-loop fairness still matters.
Design decisions forced by numbers
- Use client routing or a proxy that understands hash slots and topology changes.
- Mitigate hot keys with local cache, replicas, request coalescing, and bounded stale reads.
- Plan resharding as data movement with throttling, not just a metadata update.
Scaling to millions
- At 1M users: cache-aside, TTL, and a single-region cluster are enough.
- At 10M users: add sharding, replication, eviction policy, and stampede protection.
- At 100M+ users: add topology-aware clients, hot-key mitigation, multi-region local caches, and resharding safety.
Readable flow
Interview follow-ups to expect
- What happens when one hot key gets 200K QPS?
- What is the data-loss window during primary failure?
- How much data moves during resharding?
Operational readiness checklist
After the math, explain how this capacity plan is measured, protected, launched, and evolved.
Observability
- Attach each capacity estimate to a dashboard metric: p99 latency, queue age, cache hit rate, shard skew, error rate, and freshness lag.
- Alert on rate-of-change and saturation, not only hard failures; backlogs and hot keys usually warn before an outage.
Security and abuse
- Budget for auth checks, rate limits, quota isolation, audit logging, privacy filters, and abuse-review side paths.
- Separate trusted internal events from user-controlled input so retries and backfills cannot bypass policy checks.
Degraded modes
- Define the fallback when a dependency is slow: stale cache, partial results, read-only mode, delayed async work, or admission control.
- Protect the invariant first; reduce freshness, personalization, or analytics quality before risking duplicate or corrupt state.
Testing and rollout
- Test average, peak, and hot-object traffic; include retry storms, cold caches, lagging workers, and regional dependency loss.
- Use canaries, shadow reads, dual-write validation, feature flags, and explicit rollback criteria for risky path changes.
Rollout, backfill, and runbooks
- Make migrations and backfills idempotent, resumable, rate-limited, and visible through progress and error metrics.
- Write the runbook before launch: owner, dashboards, paging thresholds, mitigation switches, replay steps, and repair commands.
10x evolution
- For distributed cache cluster math, state which number forces the next move: cache memory, shard count, write amplification, queue lag, egress, or failover reserve.
- Add the next layer only when the metric forces it: sharding, regional reads, async replay, hot-key isolation, or stricter backpressure.
Scope you should nail down
Product surface
- In-memory key-value store with sub-ms latency
- Data structures: strings, lists, sets, hashes, sorted sets
- TTL/expiration per key
- Persistence: snapshots (RDB) and append-only (AOF)
- Replication: primary-replica async replication
- Pub/sub, transactions (MULTI/EXEC), Lua scripting
Scale targets
- 10M+ ops/sec per cluster
- Sub-millisecond p99 latency
- Data size up to 1TB per cluster
- 100K+ concurrent connections
- 99.99% availability
- Eviction when memory full
Capacity estimation
Connections: 100K concurrent → ~1K connections per node
Memory per node: 80% for data (51GB), 20% overhead
Network: 10M ops × 500 bytes avg ≈ 5GB/sec
Node count: 1 TB logical / 51 GB usable per node ≈ 20 nodes without replication; with RF=3 and 30% overhead ≈ 80-90 nodes.
Hot key: one key at 200K QPS can saturate a single shard: replicate or coalesce before adding nodes.
Replication multiplier: RF=3 means three total copies: one primary and two replicas. One million logical writes per second therefore produces about three million storage applications and two million primary-to-replica deliveries per second, before retries and protocol overhead. Replicas improve read capacity only for workloads that accept their lag; they do not divide the primary’s write-command workload.
Scaling to millions of users
At 1M users
Single-region cluster, cache-aside, TTL eviction, and simple primary-replica pairs are enough for most session and counter workloads.
At 10M users
Hash slots, topology-aware clients, hot-key replication, TTL jitter, and stampede protection via single-flight become mandatory.
At 100M+ users
Multi-region local caches, explicit stale-read policy, safe resharding with slot migration, and client-side backoff on MOVED/ASK storms.
Slot routing and failover path
Read the diagram definition
sequenceDiagram
autonumber
participant C as Client
participant CL as Cluster client
participant P as Primary slot owner
participant R as Replica
participant S as Sentinel / gossip
C->>CL: GET key
CL->>CL: CRC16(key) mod 16384 -> slot
alt slot on local node
CL->>P: read primary
P-->>C: value
else MOVED redirect
CL->>P: read redirected node
P-->>C: value
end
Note over S: Primary fails
S->>R: promote replica
R-->>C: failover reads (brief stale window)3. 🔌 System API
// Core
GET key → value
SET key value [EX seconds] [NX|XX]
DEL key [key ...]
// Expiration
EXPIRE key seconds
TTL key → remaining seconds
// Lists
LPUSH / RPUSH key value [value ...]
LRANGE key start stop
// Sets & Hashes
SADD key member | SMEMBERS key | SINTER key [key ...]
HSET key field value | HGET key field | HGETALL key
// Sorted Sets & Pub/Sub
ZADD key score member | ZRANGE key start stop [WITHSCORES]
SUBSCRIBE channel | PUBLISH channel message4. 🗄️ Data Structures
Strings (SDS)
Simple Dynamic String: header (len, alloc, flags) + char array. O(1) length, binary-safe, preallocated buffer to reduce realloc.
Lists
A list exposes ordered insertion and removal, which is useful for bounded queues and recent-item histories. The physical encoding can change with Redis version and object size; modern releases use compact listpack-based representations and quicklists rather than assuming a linked node per element. End operations avoid scanning the list, while access deep into a list can still require traversal. Design around command complexity and measured memory use, not a permanent internal layout.
Hash Tables
Chaining for collision. Incremental rehashing (progressively move buckets). Dict: two hash tables for rehash without blocking.
Skip Lists (Sorted Sets)
Probabilistic multi-level linked list. O(log n) insert, delete, range. Score + member for ordering. Hash + skip list for O(1) lookup by member.
Intsets
Sorted integer array for small sets. Compact: int16/int32/int64 encoding. Upgraded when value exceeds size.
Zip Lists
Compact encodings store several small elements together instead of allocating a pointer-rich object for each entry. This saves allocator and pointer overhead, but finding or inserting an entry may involve scanning or moving bytes. Older Redis descriptions call this a ziplist; current memory-optimization documentation describes listpacks for several small collection types. Check the version and configured size thresholds before estimating memory.
5. 🏗️ Architecture
Single-threaded: Avoids locks, context switches. I/O multiplexing (epoll) handles many connections. CPU-bound work can use threads (e.g., lazy free, bg save).
6. 💾 Persistence
RDB
An RDB snapshot captures a dataset at a point in time. A background child writes that snapshot while the parent serves requests; copy-on-write lets both initially share memory pages, then copies a page when the parent modifies it. This avoids stopping service for the entire disk write, but the fork itself can pause a large process and heavy writes can temporarily increase memory use. If the only persisted state is an old snapshot, recovery loses changes made after it.
AOF
Append-only file. Log every write. Fsync: always (every write), everysec (1s default), no (OS decides). Rewrite to compact (BGREWRITEAOF).
Hybrid
RDB + AOF. RDB for fast full restore; AOF for incremental replay. Best of both: durability + fast recovery.
7. 🔄 Replication
Redis explicitly documents that replica acknowledgements reduce some loss scenarios without making the system strongly consistent; see the Redis replication documentation. For version-sensitive collection layouts, consult the memory-optimization documentation.
Master-Slave
A replica tracks the primary’s replication history identifier and offset. After a short disconnect, PSYNC can resume from the retained backlog; if that history is no longer available, the replica needs a full synchronization with a snapshot and subsequent updates. SYNC is the older protocol. Reserve capacity for full resynchronization because transferring and loading a large dataset can stress both primary and replica while normal traffic continues.
// Request acknowledgements for earlier writes on this connection
WAIT numreplicas timeout
// Returns the number of replicas that acknowledged, or after timeout
// Check the returned count; WAIT does not guarantee strong consistency8. 📦 Clustering
Hash Slots
16384 slots. Key → CRC16(key) % 16384 → slot. Each node owns subset of slots. Hash tags like {user:123} for co-locating related keys.
Cluster Bus
Gossip protocol on port+10000. Nodes exchange heartbeat, failure detection (PFAIL → FAIL). Slot migration: MIGRATE command for rebalancing.
9. 🗑️ Eviction Policies
| Policy | Behavior |
|---|---|
| noeviction | Return error when memory full (default) |
| allkeys-lru | Evict any key by LRU |
| volatile-lru | Evict keys with TTL by LRU |
| allkeys-random | Evict random key |
| volatile-ttl | Evict key with shortest TTL |
| allkeys-lfu | Evict by least frequently used |
| volatile-lfu | Evict keys with TTL by LFU |
LRU: approximate (sampling); LFU: 24-bit counter, decay over time.
10. 💡 Cache Patterns
Cache-Aside
App reads cache first; on miss, read DB, write to cache. App manages cache.
Write-Through
Write-through puts the persistence step on the write path: a cache abstraction forwards the update to the database before reporting success. It can keep freshly written values available for reads, but two independent writes are not automatically atomic. Define what happens when the database commit succeeds and the cache update fails, and use versions, invalidation, or a coordinated protocol to prevent an older cache value from silently surviving.
Write-Behind
Write-behind acknowledges an update before the backing database has received it, then flushes changes asynchronously. Batching can reduce database write overhead, but the intermediate queue becomes part of the durability contract. If it exists only in cache memory, a crash can lose acknowledged writes. Coalescing repeated updates is suitable for some replaceable values, but cannot silently discard increments or business events that must each occur.
Read-Through
Cache loads from DB on miss. Cache library handles this.
A stampede occurs when many requests miss the same key at once and all query the backing database. With single-flight, the first request starts the load and other requests wait on that same result within a deadline. Stale-while-revalidate can return a previously approved value while one worker refreshes it, if the product permits staleness. Add jitter to expiration times so many different keys do not expire together; jitter alone does not prevent concurrent loads for one hot key.
11. 🧠 Memory Management
jemalloc
in-memory cache store uses jemalloc over glibc malloc: fewer fragments, better locality. Size classes reduce fragmentation.
Lazy Free
UNLINK (async DEL): free in background thread. Large key deletion won't block event loop.
Memory Limit
The maxmemory setting triggers the configured eviction behavior when Redis exceeds its accounted memory budget. It is not a strict ceiling on the whole process’s resident memory: replication buffers, allocator fragmentation, and persistence work can require additional space. Reserve headroom, monitor actual process memory as well as cache usage, and choose whether full capacity should evict values or reject memory-growing commands.
12. 🛡️ High Availability
Sentinel
Monitors master/slaves. Failover: elect new master when master down. Quorum for agreement. Notify clients of new topology.
Cluster Mode
Built-in failover. Each shard has master + replicas. Automatic failover when master fails.
13. ⚠️ Bottlenecks
Single-threaded CPU
Long commands block. Mitigation: avoid KEYS, use SCAN; pipeline; lazy free for big deletes.
Memory bound
Data limited by RAM. Mitigation: eviction; cluster sharding; offload cold data.
Replication lag
Async replication permits a replica to return an older value after the primary acknowledged an update. Reading the primary avoids that particular lag during normal operation. WAIT can request acknowledgements from replicas and reduce some loss scenarios, but does not make Redis strongly consistent or guarantee that an acknowledged write survives every failover. A cache can often refill missing data from its database; a source-of-truth ledger needs a stronger storage contract.
Fork for RDB
BGSAVE fork can cause latency spike on large memory. Mitigation: AOF-only; smaller instances.
Short takeaway
- Command execution: A serialized command path simplifies individual-node execution; the 10M+ ops/sec figure is an illustrative cluster target, not a per-node guarantee.
- Data structures: SDS, hash tables, sorted-set indexes, and compact encodings such as listpacks; physical representations vary by version.
- Persistence: RDB (fork+COW), AOF (append-only), hybrid.
- Replication: Master-slave, partial resync for reconnects, WAIT for replica acknowledgements, with explicit failover-loss limits.
- Clustering: 16384 hash slots, gossip, slot migration.
- Eviction: LRU/ LFU/random/ttl when maxmemory reached.
- HA: Sentinel for failover; cluster mode for sharding + failover.
Beginner to Advanced Design Path(open guidance)
Beginner answer
Describe get, set, delete, TTL, memory storage, eviction, replication, and simple hash-based sharding.
Advanced answer
Discuss hash slots, resharding, MOVED redirects, hot-key replication, TTL jitter, request coalescing, async replication loss, and failover behavior.
Deeper walkthrough
Distributed cache deeper walkthrough
A cache is useful only if clients understand misses, timeouts, topology changes, and stale data. Design the client behavior along with the cluster.
Requirement refinements
Clarify whether the cache is only an optimization or the primary store for session, counter, lock, or queue-like data.
Define latency, memory limits, TTL behavior, persistence expectations, and failover tolerance.
API design details
Support get, set, delete, TTL, atomic increment, compare-and-set or Lua-style scripts for compound operations.
Expose cluster metadata or redirection responses so clients can follow key movement during resharding.
Data model and storage
Use hash slots or consistent hashing to place keys, and track replica ownership separately from primary ownership.
Store key metadata for expiry and eviction sampling; do not scan all keys synchronously for TTL cleanup.
Scaling strategy
Add virtual nodes or hash slots so adding nodes moves a bounded subset of keys.
Handle hot keys with replication, local caching, request coalescing, or key splitting depending on semantics.
Reliability, recovery, and operations
Failover may lose recent async writes; say whether the workload accepts that.
Monitor hit rate, p99 latency, evictions, memory fragmentation, replication lag, failover time, and hot-key skew.
Choosing and explaining distributed cache topology
A complete explanation connects the mechanism to a requirement. Describe what improves when you introduce it, which new failure cases become possible, and how you would know that the trade-off is worthwhile.
Explain the core idea
The client is part of the design: it must follow slot redirects, pipeline safely, and handle MOVED responses.
Caches are optimizations unless the product accepts data loss: say which keys can disappear on failover.
Hot keys break average QPS math; replicate or split before adding more nodes.
When it belongs
Read QPS after cache-aside math still exceeds database comfort.
Session, rate-limit, or counter state needs sub-ms latency across many app servers.
Interviewer asks what happens when a shard dies mid-write.
How it fails
Async replication loses the last writes on failover.
Single hot key saturates one shard despite healthy cluster averages.
Resharding triggers MOVED storms if clients do not backoff.
Trade-offs to weigh
Cluster mode vs Sentinel: cluster scales horizontally; Sentinel is simpler for small deployments.
RDB vs AOF: RDB for fast restart; AOF for finer durability; hybrid for production.
Eviction policy: allkeys-lru for pure cache; noeviction when cache backs queues or locks.
Common misunderstandings
Using KEYS * or unbounded SCAN on production.
Treating cache as source of truth for money or inventory.
Ignoring connection memory and file descriptor limits at 100K clients.
Deepening reference
Distributed Cache: realistic production notes
Beginner mental model: A distributed cache is memory partitioning, replication, eviction, and failure handling for hot data.
Production scope: The design must cover key placement, node failure, rebalancing, hot keys, persistence options, and client behavior on cache miss or timeout.
Choose what the first version must guarantee
Start by deciding which distributed cache capabilities are essential. Describe the main read and write operations before choosing infrastructure. For each operation, identify the stored records it uses and the condition that makes its result correct. A fast response is useful only if it satisfies that condition.
Next, separate work the user must wait for from work that can finish later. Estimate requests per second, storage growth, and the busiest object or customer. Use those estimates to choose one bottleneck to investigate. The calculations below are example workloads with stated assumptions; they are not measured limits of a particular product.
Concrete capacity calculations
Capacity: 1B keys with 1 KB average value is 1 TB value data. With 100 bytes metadata per key, add about 100 GB; with 2 replicas, plan over 2.2 TB before allocator overhead.
Node count: if each cache node safely holds 64 GB usable memory, 2.2 TB replicated data needs at least 35 nodes. Use 50 or more for headroom, rebalancing, and skew.
Throughput: 5M gets/sec across 50 nodes averages 100K gets/sec per node. Hot-key skew can still overload one node even when average capacity is fine.
Rebalancing: adding 10 percent more nodes should move roughly 10 percent of keys with consistent hashing, not the entire keyspace.
Expiry storms: if 100M keys share the same TTL and expire together, the backing store can see massive synchronized misses. Add jitter and refresh-ahead for hot keys.
Scaling to millions of users
Partition keys with hash slots or consistent hashing so topology changes move a bounded subset of keys.
Replicate hot and important keys; treat ordinary cache data as rebuildable unless explicitly using the cache as a primary store.
Use client timeouts, request coalescing, and circuit breakers so cache slowness does not amplify downstream latency.
Choose eviction policy by workload: recency, frequency, TTL, or no-eviction for protected datasets.
Track per-shard hit rate, memory, evictions, replication lag, and key skew rather than only cluster averages.
Sequence and data-flow diagram
Walk this path after the high-level design so it is clear what is synchronous, what is asynchronous, and where the system can degrade safely.
Read the diagram definition
sequenceDiagram
participant App
participant Client as Cache client
participant Meta as Cluster metadata
participant N1 as Cache shard
participant DB as Source store
App->>Client: Get key
Client->>Meta: Resolve slot if needed
Client->>N1: Fetch key
alt Hit
N1-->>App: Value
else Miss
App->>DB: Load value
App->>N1: Set value with TTL
endCore flow to walk through
Client maps key to shard using hash slots or consistent hashing.
Shard serves get/set from memory, applying TTL and eviction policy.
Replication and persistence protect against node failure when durability is required.
Production readiness
Distributed Cache: ship, operate, and evolve
Now examine the same request flow during failures and deployments. Tie each operational control to the user behavior it protects.
Observability
Track distributed cache p95/p99 latency, error rate, saturation, and user-visible freshness separately for read, write, and async paths.
Alert on leading indicators such as queue age, hot-key skew, cache miss storms, reconciliation mismatches, and error-budget burn.
Security and abuse
Enforce authentication, authorization, rate limits, input validation, audit trails, and privacy checks at the service boundary.
Treat abuse controls as a first-class path: throttling, quota isolation, moderation hooks, revocation, and safe handling of sensitive data.
Degraded modes
Name what still works if cache, search, ranking, stream processing, object storage, or a regional dependency is slow.
Prefer stale-but-safe reads, partial results, async retry, admission control, and feature disablement over corrupting core state.
Testing and rollout
Load test average, peak, and hot-object traffic; add failure-injection for retries, timeouts, backlog growth, and dependency outages.
Ship with canaries, shadow reads or dual writes when useful, rollback switches, schema compatibility checks, and replay validation.
Runbooks
Document dashboards, paging thresholds, owner actions, rollback steps, data repair commands, and customer-impact language.
Keep backfills and replays idempotent, resumable, rate-limited, and observable so repair work does not create a second incident.
10x evolution
Say which metric forces the next architecture change: shard count, cache footprint, write amplification, queue lag, p99 latency, or regional failover reserve.
Evolve one bottleneck at a time: partition by access pattern, isolate hot tenants, regionalize reads, add replayable async pipelines, then harden multi-region behavior.
Hard parts worth deepening
Adding nodes should not move all keys; use hash slots or consistent hashing.
Hot keys can overload one shard even when total cluster capacity is fine.
Eviction policy must match workload: LRU, LFU, TTL, or noeviction.
Cache is not the source of truth unless explicitly designed as one.
Failure modes and recovery
Node fails; promote replica and update cluster metadata.
Mass expiry causes thundering herd; add TTL jitter and request coalescing.
Hot key overloads shard; replicate hot values or split logical key.
Client ignores MOVED/ASK redirects; requests fail during resharding.
Signals to monitor
Cache hit rate describes how often reads avoid the backing store; p99 latency describes the slow end of the lookup experience. Measure hit and miss paths separately so a fast majority does not conceal slow misses.
Eviction count reveals memory pressure but does not by itself prove poor policy. Compare it with hot-key skew and miss traffic. Replication lag indicates how far a promoted replica might be behind under the chosen persistence and failover settings.
| Choice | Best when | Cost |
|---|---|---|
| Async replication | Low latency is critical. | Recent writes can be lost on failover. |
| Synchronous replication | Cache is semi-durable. | Higher write latency. |
| Client-side sharding | Simple architecture and low proxy overhead. | Harder client updates during topology changes. |
Try this scenario
Explain how the cluster behaves when a shard owning 10 percent of keys dies during peak traffic.
Primary sources
- Nishtala et al.: Scaling Memcache at Facebook (NSDI): lease/gutter patterns at scale.
- Redis Docs: Scaling / clustering: sharded cache operations.
- RFC 9111: HTTP Caching: freshness/revalidation vocabulary.
Practice and further reading
Use these exercises to connect the lesson to a design decision. State your assumptions, trace what happens to one request, and explain the rule that must remain true even during a failure. That rule is the design's correctness invariant.
Focus for this lesson: Build protocol, membership, partitioning, rebalancing, admission, eviction, replication, split-brain handling, hot-key control, and stampede protection.
Explain the decision
What happens to cache availability during membership change?
Read the reasoning
Use versioned partition maps, gradual movement, replication/read fallback, and avoid remapping the entire keyspace at once.
Apply it to a production scenario
One key receives 20% of all reads and its owner fails. What protects the cluster?
Read one approach
Replicate hot keys, request-coalesce fills, use local near-cache, cap origin work, and avoid immediate synchronized retries.
Work through the numbers
A 2 TB logical cache at replication factor 2 and 70% target memory utilization needs how much installed memory?
Read the worked calculation
2 × 2 / 0.70 ≈ 5.71 TB before allocator fragmentation and failover headroom.
Reason through a failure
What must cache failure never do?
Read the recovery reasoning
It may reduce performance or freshness within contract, but cannot become the sole copy of durable data or bypass current authorization.
Sources and further study
Continue with Design Key-Value Store.