Bird
Raised Fist0

Design a Distributed Cache - Consistent Hashing, Eviction, Replication

HLDhardInterview freq: medium
Choose your preparation mode4 modes available
ArchitectureTraceChallenge
🏗️
Design a Distributed Cache - Consistent Hashing, Eviction, Replication
Design a scalable distributed cache with consistent hashing, eviction policies, and replication for high availability
hardInfrastructure and Platform SystemsFreq: medium
Prerequisites:Distributed Systems BasicsHashingCaching StrategiesReplication and Consistency Models
⚙️
Functional Requirements
  • Support key-value storage with low-latency read and write operations
  • Distribute cache data across multiple nodes using consistent hashing to minimize rebalancing
  • Implement eviction policies (e.g., LRU, TTL) to manage limited memory per node
  • Replicate cache entries across multiple nodes for fault tolerance and high availability
  • Provide eventual consistency guarantees with configurable replication factors
  • Expose APIs for cache get, put, and delete operations
📊
Non-Functional Requirements
  • Support 2 billion daily active users generating 40 cache operations per user per day
  • Achieve 99.99% availability with automatic failover and replication
  • Maintain 100ms P99 latency for cache get and put operations
  • Handle peak write QPS of approximately 1 million writes per second
  • Ensure data durability for at least 24 hours with replication
❓
Clarifying Questions to Ask
  • What is the expected cache hit ratio and tolerance for stale data?
  • What eviction policies are preferred or required (LRU, TTL, LFU)?
  • What consistency model is acceptable (strong, eventual, read-your-writes)?
  • What is the replication factor and failure tolerance requirement?
  • Are there any constraints on cache node hardware or network topology?
  • Should the cache support multi-region deployment or geo-replication?
Out of Scope
✗ Designing the underlying persistent storage or database
✗ Complex query support beyond key-value access
✗ Authentication and authorization mechanisms
✗ Cache warming or preloading strategies
📌
Assumptions
  • 2 billion DAU
  • 40 cache operations per user per day (mix of reads and writes)
  • Average cache entry size: 100 bytes
  • Replication factor: 3
⚡
QPS Calculation
total ops per day2B × 40 = 80B ops/day
write ops ratioAssuming 30% writes → 80B × 0.3 = 24B writes/day
read ops ratio70% reads → 80B × 0.7 = 56B reads/day
write qps24B / 86400 ≈ 277,778 writes/sec
peak write qpsAssuming 4× peak factor → 277,778 × 4 ≈ 1.11M writes/sec
read qps56B / 86400 ≈ 648,148 reads/sec
💾
Storage Estimate
daily storage80B ops × 100 bytes = 8,000 GB/day (8 TB/day)
retention days1
total storage8 TB × 1 day = 8 TB
with replication8 TB × 3 replicas = 24 TB total storage
🖥️
Server Estimate
cache nodesAssuming each cache node handles 10K writes/sec peak → 1.11M / 10K = 111 nodes
storage nodesAssuming each node stores 500 GB → 24 TB / 0.5 TB = 48 nodes
total nodesConsidering overlap and replication, ~150 nodes cluster
🏗️
Architecture Diagram
Architecture Diagram
          +----------------------+
          |      Client Apps      |
          +----------+-----------+
                     |
                     v
          +----------------------+
          |  Cache Coordinator    |
          | (Consistent Hashing)  |
          +----------+-----------+
                     |
    +----------------+----------------+----------------+
    |                |                |                |
    v                v                v                v
 +-------+        +-------+        +-------+        +-------+
 |Cache 1|        |Cache 2|        |Cache 3|        |Cache N|
 |Node   |        |Node   |        |Node   |        |Node   |
 +---+---+        +---+---+        +---+---+        +---+---+
     |                |                |                |
     +-------+--------+----------------+--------+-------+
             |                         |                
             v                         v                
       +-----------+             +-----------+          
       |Replication|             |Replication|          
       |Manager    |             |Manager    |          
       +-----------+             +-----------+
🧩
Components
Cache Coordinator
Routes client requests to appropriate cache nodes using consistent hashing
Why thisConsistent hashing minimizes key remapping when nodes join/leave, reducing cache misses and rebalancing overhead
Why notCentralized hash ring or modulo-based partitioning causes massive remapping and hotspots during scaling
Cache Nodes (Redis Cluster)
Store key-value pairs in-memory with TTL and eviction policies
Why thisRedis supports native TTL, LRU eviction, and fast sub-millisecond lookups with built-in replication and clustering
Why notMemcached lacks native replication and persistence, making it unsuitable for high availability
Replication Manager
Ensures data replication across nodes for fault tolerance and consistency
Why thisRedis Sentinel or Redis Cluster replication provides automatic failover and eventual consistency with minimal latency impact
Why notCustom replication layers add complexity and risk inconsistent state under failures
Client SDK
Implements consistent hashing logic and retries to route requests to correct cache nodes
Why thisOffloading hashing to clients reduces load on coordinator and enables direct node access, improving latency
Why notProxy-based routing adds latency and single points of failure
Monitoring and Metrics Service
Tracks cache hit/miss rates, node health, latency, and triggers alerts
Why thisPrometheus + Grafana provide scalable, real-time monitoring with alerting for SLA adherence
Why notAd-hoc logging lacks real-time visibility and scalability at 150+ nodes
🔄
Data Flow
1
Client App → Client SDK
Client SDK computes hash(key) to determine target cache node
Consistent hashing ring used to map keys to nodes
2
Client SDK → Cache Node
Send GET/PUT request directly to cache node
Direct node access reduces coordinator bottleneck
3
Cache Node → Replication Manager
Replicate write operations asynchronously to replica nodes
Replication factor of 3 ensures fault tolerance
4
Cache Node → Cache Node
Evict keys based on LRU or TTL policies when memory limits reached
Eviction prevents memory exhaustion and stale data
🗄️
CacheEntries
PK: cache_node_id
cache_node_id (string)key (string)value (blob)ttl (int)timestamp (timestamp)
Stores key-value pairs with TTL per cache node
🗄️
ReplicationStatus
PK: cache_node_id
cache_node_id (string)replica_node_id (string)last_replicated_timestamp (timestamp)status (string)
Tracks replication status between primary and replica nodes
🔌
GET /cache/{key}
Request
{
  "path_params": [
    "key:string"
  ]
}
Response
{
  "value": "blob",
  "ttl": "int",
  "found": "boolean"
}
Fetches value for key if present and not expired
🔌
PUT /cache/{key}
Request
{
  "path_params": [
    "key:string"
  ],
  "body": {
    "value": "blob",
    "ttl": "int (optional)"
  }
}
Response
{
  "success": "boolean"
}
Stores or updates key with optional TTL
🔌
DELETE /cache/{key}
Request
{
  "path_params": [
    "key:string"
  ]
}
Response
{
  "success": "boolean"
}
Deletes key from cache

Practice

(1/5)
1. A messaging system must handle 1 million writes per second and fan-out delivery to thousands of consumers without blocking producers. Which component is primarily responsible for decoupling producers from consumers to achieve this?
easy
A. Use a relational database with write-ahead logging to buffer messages
B. Use direct synchronous HTTP calls from producers to consumers
C. Use a centralized in-memory cache like Redis to store messages temporarily
D. Use a distributed message broker with partitioned logs and asynchronous consumers

Solution

  1. Step 1: Identify the decoupling mechanism

    Relational DBs with WAL (Use a relational database with write-ahead logging to buffer messages) are durable but cannot handle 1M writes/sec with fan-out efficiently; Redis (Use a centralized in-memory cache like Redis to store messages temporarily) is fast but lacks durability and persistence guarantees at this scale; direct HTTP calls (Use direct synchronous HTTP calls from producers to consumers) block producers and do not decouple.

  2. Step 2: Recognize distributed brokers' role

    Distributed message brokers with partitioned logs (Use a distributed message broker with partitioned logs and asynchronous consumers) allow asynchronous writes and reads, decoupling producers and consumers, enabling high throughput and scalable fan-out.

  3. Final Answer:

    Option D -> Option D

  4. Quick Check:

    Partitioned broker logs decouple producers and consumers for high throughput [check]
Hint: Partitioned brokers decouple producers and consumers [check]
Common Mistakes:
2. Given a system that must maintain availability during a regional outage but can tolerate eventual consistency, which architecture is preferable to minimize blast radius?
medium
A. Synchronous cross-region database replication with strong consistency
B. Asynchronous multi-region replication with conflict resolution and eventual consistency
C. Single-region active-active with synchronous writes to all regions
D. Centralized global cache with synchronous invalidation on writes

Solution

  1. Step 1: Identify availability vs consistency trade-off

    Asynchronous replication with eventual consistency allows regions to operate independently during outages, reducing blast radius.

  2. Step 2: Evaluate alternatives

    Synchronous replication (A, C) blocks on remote writes, increasing outage impact; centralized cache (D) is a single point of failure.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Async replication -> availability during regional failures [check]
Hint: Async replication favors availability, sync blocks cause outages [check]
Common Mistakes:
3. If the token assignment metadata is lost or corrupted in a consistent hashing system with virtual nodes, what exactly breaks?
medium
A. Only new nodes fail to receive keys, existing nodes continue serving correctly
B. Requests cannot be routed to correct nodes, causing key lookups to fail or go to wrong nodes
C. The system continues routing keys correctly but load balancing is uneven
D. The system falls back to modulo hashing automatically without data loss

Solution

  1. Step 1: Understand role of token assignment metadata

    Token metadata maps virtual nodes to physical nodes; losing it breaks new node integration.

  2. Step 2: Trace failure impact

    Existing nodes continue serving keys assigned before loss; new nodes cannot be assigned tokens properly.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Token metadata loss blocks new node assignment; existing nodes unaffected [check]
Hint: Token metadata loss blocks new nodes, existing nodes serve [check]
Common Mistakes:
4. At 10x scale, which MongoDB component is most likely to become the bottleneck first when using the bucket pattern for time-series data, and why?
hard
A. The application server, because it cannot generate bucket IDs fast enough for writes.
B. The MongoDB query router (mongos), because it must maintain metadata for all buckets and route writes precisely.
C. The shard servers, because bucket documents grow large and cause frequent document splits and index overhead.
D. The network layer, because increased bucket writes saturate bandwidth between shards.

Solution

  1. Step 1: Identify bucket pattern scaling challenges

    Bucket pattern creates many bucket documents requiring precise routing by mongos.

  2. Step 2: Mongos metadata overhead

    Mongos maintains routing metadata and must route each write to correct shard and bucket, increasing load at scale.

  3. Step 3: Compare other components

    Shard servers handle writes but document splits are less frequent; network saturates later; app server bottleneck unlikely.

  4. Final Answer:

    Option B -> Option B

  5. Quick Check:

    Mongos routing metadata load grows fastest at 10x scale [check]
Hint: Mongos routing metadata overload is first bottleneck at scale [check]
Common Mistakes:
5. At 10x scale, which component in a Zookeeper-based leader election and configuration system is most likely to become the bottleneck first, and why?
hard
A. Client applications, because they overwhelm the network with configuration fetches
B. Follower nodes, because they must process all leader election events concurrently
C. Persistent storage backend, because it cannot keep up with ephemeral node writes
D. Zookeeper ensemble leader node, because it handles all write requests and session management serially

Solution

  1. Step 1: Identify Zookeeper ensemble roles

    Leader node serializes all writes and manages session heartbeats.

  2. Step 2: Analyze scaling impact

    At 10x scale, leader node CPU and network become bottleneck due to serialized writes and session management.

  3. Step 3: Evaluate other components

    Followers handle reads and watches but are less bottlenecked; storage backend is optimized for writes.

  4. Final Answer:

    Option D -> Option D

  5. Quick Check:

    Zookeeper leader serializes writes -> first bottleneck at scale [check]
Hint: Leader node serializes writes -> first bottleneck at scale [check]
Common Mistakes:
Recommended

Ready to prove it?

Pass the test and this pattern moves unknown strong while earning +1 Grip toward your certificate.

10 questions~5 mineasy hard