Bird
Raised Fist0

Design a Key-Value Store - Consistent Hashing, Vector Clocks, Gossip Protocol

HLDhardInterview freq: medium
Choose your preparation mode4 modes available
ArchitectureTraceChallenge
🏗️
Design a Key-Value Store - Consistent Hashing, Vector Clocks, Gossip Protocol
Design a highly available, scalable key-value store with eventual consistency using consistent hashing, vector clocks, and gossip protocol
hardInfrastructure and Platform SystemsFreq: medium
Prerequisites:Distributed Systems BasicsHashingCAP TheoremVector ClocksGossip Protocol
⚙️
Functional Requirements
  • Store and retrieve key-value pairs with low latency (sub-100ms P99).
  • Support horizontal scaling to billions of keys and millions of operations per second.
  • Provide eventual consistency with conflict detection and resolution using vector clocks.
  • Distribute data using consistent hashing to minimize rebalancing on node changes.
  • Use gossip protocol for cluster membership, failure detection, and metadata dissemination.
  • Support replication for fault tolerance and high availability.
📊
Non-Functional Requirements
  • Handle 2 billion daily active users generating 40 writes per user per day.
  • Achieve 99.99% availability SLA.
  • Ensure 100ms P99 read and write latency under peak load.
  • Support data retention of 30 days with replication factor of 3.
  • Scale to 925,000 write requests per second at peak.
❓
Clarifying Questions to Ask
  • What is the expected key and value size distribution?
  • Is the workload read-heavy, write-heavy, or balanced?
  • What is the replication factor and consistency level required?
  • Are deletes and TTLs supported or only put/get?
  • What is the expected cluster size and node failure rate?
  • Is the system expected to run in a single data center or multiple?
Out of Scope
✗ Strong consistency or linearizability guarantees.
✗ Complex query support beyond key-based get/put/delete.
✗ Multi-region geo-replication and cross-datacenter consistency.
✗ Advanced security features like encryption at rest or RBAC.
📌
Assumptions
  • 2 billion daily active users (DAU).
  • Each user generates 40 write operations per day.
  • Average size per key-value write is 100 bytes.
  • Replication factor is 3 for fault tolerance.
⚡
QPS Calculation
write qps2,000,000,000 users × 40 writes/user/day ÷ 86,400 seconds/day ≈ 925,926 writes/sec
peak write qpsAssuming 2× peak factor: 925,926 × 2 = 1,851,852 writes/sec
read qpsAssuming read:write ratio 3:1, read_qps = 3 × 925,926 ≈ 2,777,778 reads/sec
💾
Storage Estimate
daily storage925,926 writes/sec × 100 bytes × 86,400 sec/day ≈ 8 TB/day uncompressed
retention 30 days8 TB/day × 30 days = 240 TB
with replication240 TB × 3 replicas = 720 TB total storage
🖥️
Server Estimate
storage nodesAssuming each node stores 4 TB usable (after overhead), need 720 TB ÷ 4 TB ≈ 180 nodes
coordination nodesDedicated 5 nodes for gossip and cluster metadata to avoid contention
🏗️
Architecture Diagram
Architecture Diagram
          +----------------+
          | Client Request |
          +-------+--------+
                  |
                  v
          +-------+--------+
          | Load Balancer  |
          +-------+--------+
                  |
                  v
          +-------+--------+
          | Coordinator    |
          | (API Layer)    |
          +-------+--------+
                  |
                  v
          +-------+--------+
          | Consistent     |
          | Hashing Ring   |
          +-------+--------+
          /       |        \
         /        |         \
        v         v          v
 +------+--+ +----+-----+ +--+------+
 | Storage | | Storage | | Storage |
 | Node 1  | | Node 2  | | Node N  |
 +---------+ +---------+ +---------+
       |            |           |
       v            v           v
 +---------------------------------+
 | Gossip Protocol Cluster Manager  |
 +---------------------------------+
🧩
Components
Load Balancer
Distributes client requests evenly across coordinator nodes to balance load and provide fault tolerance.
Why thisUse HAProxy for mature TCP/HTTP load balancing with health checks and low latency.
Why notDNS round-robin lacks health checks and can cause uneven load distribution.
Coordinator (API Layer)
Receives client requests, computes partition using consistent hashing, and routes to storage nodes.
Why thisStateless microservice layer enables horizontal scaling and isolates client protocol from storage internals.
Why notEmbedding routing logic in clients leads to complex client updates and inconsistent hashing implementations.
Consistent Hashing Ring
Maps keys to storage nodes minimizing data movement on node joins/leaves using virtual nodes.
Why thisConsistent hashing reduces rebalancing overhead and hotspots compared to modulo-based partitioning.
Why notModulo partitioning requires full data reshuffle on cluster changes, causing downtime.
Storage Nodes (e.g., Cassandra or ScyllaDB)
Store key-value pairs with replication, maintain vector clocks for versioning, and serve read/write requests.
Why thisCassandra supports tunable consistency, built-in replication, and efficient storage for large-scale KV workloads.
Why notRelational DBs like PostgreSQL do not scale linearly for 1.8M writes/sec and require manual sharding.
Gossip Protocol Cluster Manager
Manages cluster membership, failure detection, and metadata dissemination among storage nodes.
Why thisGossip protocol provides scalable, decentralized membership with eventual consistency and low overhead.
Why notCentralized coordination (e.g., ZooKeeper) creates single points of failure and bottlenecks at scale.
🔄
Data Flow
1
Client → Load Balancer
Client sends get/put/delete request to load balancer.
Load balancer evenly distributes requests to coordinators.
2
Load Balancer → Coordinator
Forward request to one of the stateless coordinator nodes.
Coordinator abstracts client protocol and handles routing.
3
Coordinator → Consistent Hashing Ring
Compute partition for key using consistent hashing with virtual nodes.
Ensures minimal data movement on cluster changes.
4
Coordinator → Storage Node(s)
Route request to responsible storage node(s) based on partition and replication factor.
Storage nodes handle read/write with vector clocks for versioning.
5
Storage Node(s) → Gossip Protocol Cluster Manager
Nodes exchange membership and failure info via gossip protocol.
Ensures cluster state is eventually consistent and failures detected.

Practice

(1/5)
1. In a directory-based sharding system, at step 3 the Shard Lookup Service receives a request for user ID 12345. What happens next?
easy
A. The service queries a centralized directory to find the exact shard assigned to user ID 12345
B. The service uses range boundaries to determine the shard for user ID 12345
C. The service forwards the request to all shards to find user ID 12345
D. The service hashes the user ID and routes to a shard based on the hash modulo shard count

Solution

  1. Step 1: Understand directory-based sharding

    Directory-based sharding uses a centralized mapping from keys to shards.

  2. Step 2: Trace step 3 action

    The Shard Lookup Service consults the directory to find the exact shard for user ID 12345.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Directory lookup returns shard for specific key [check]
Hint: Directory sharding uses centralized map to find shard [check]
Common Mistakes:
2. In a message delivery system, at Step 4 the Delivery Worker reads the message offset from Kafka and attempts to deliver it to the consumer. What happens if the consumer's acknowledgment is not received within the configured timeout?
easy
A. The Delivery Worker retries delivery without committing the offset
B. The Delivery Worker marks the message as delivered and commits the offset
C. The Delivery Worker discards the message and moves to the next offset
D. The Delivery Worker pauses all deliveries until manual intervention

Solution

  1. Step 1: Understand Kafka offset commit semantics

    Kafka commits offsets only after successful acknowledgment to avoid message loss.

  2. Step 2: Delivery Worker behavior on missing ack

    If no ack is received, the worker retries delivery to ensure at-least-once semantics; it does not commit offset or discard messages.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Retries without committing offset -> ensures message redelivery [check]
Hint: No ack -> retry delivery, no offset commit [OK]
Common Mistakes:
3. For a system requiring both low latency and strong consistency in user profile updates, which architecture choice best balances these constraints?
medium
A. Use eventual consistency with asynchronous replication to maximize availability
B. Use a single primary database with synchronous replication for strong consistency
C. Use a distributed cache with write-behind to the database for low latency
D. Use client-side caching to reduce read latency without server coordination

Solution

  1. Step 1: Identify consistency and latency requirements

    Strong consistency requires synchronous replication; low latency demands minimal delay.

  2. Step 2: Evaluate architecture options

    Single primary DB with sync replication ensures consistency but may increase latency slightly; best trade-off.

  3. Final Answer:

    Option B -> Option B

  4. Quick Check:

    Synchronous replication balances strong consistency and latency [check]
Hint: Strong consistency needs sync replication, not eventual or cache-only [check]
Common Mistakes:
4. For a system tracking SLO error budgets with high write volume and low latency requirements, which architecture is best suited to ensure accurate, near real-time error budget updates?
medium
A. Batch processing error logs every hour using Hadoop MapReduce
B. Polling a relational database every minute for error counts and updating budgets
C. Stream processing with Apache Flink to compute error budgets continuously
D. Using a time-series database with hourly rollups for error budget calculation

Solution

  1. Step 1: Analyze latency and accuracy requirements

    Near real-time updates require continuous stream processing rather than batch or polling.

  2. Step 2: Evaluate each architecture against these requirements

    Batch processing (A) and hourly rollups (D) are too slow; polling (C) introduces latency and load spikes; stream processing (B) provides continuous, low-latency computation.

  3. Final Answer:

    Option C -> Option C

  4. Quick Check:

    Stream processing -> continuous, low-latency error budget updates [check]
Hint: Stream processing fits low latency and high write volume best [OK]
Common Mistakes:
5. At 10x scale, which component in the Redlock system is most likely to fail first and why?
hard
A. Redis masters due to increased memory usage from many locks
B. Lock expiration timers due to clock drift causing inconsistent lock validity
C. Network links between clients and Redis masters due to increased lock request volume
D. Clients due to increased network latency causing lock acquisition timeouts

Solution

  1. Step 1: Identify bottleneck at 10x scale

    Increased lock requests increase network latency and contention on clients acquiring locks.

  2. Step 2: Analyze impact on lock acquisition

    Higher latency causes clients to miss lock acquisition timeouts, reducing throughput and increasing retries.

  3. Step 3: Compare other components

    Redis masters scale better with sharding; network links and timers less likely first failure.

  4. Final Answer:

    Option D -> Option D

  5. Quick Check:

    Clients hit network latency -> lock timeouts -> first failure at scale [check]
Hint: Clients face latency -> lock timeouts -> first failure 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