Design a Key-Value Store

Consistent Hashing with Virtual Nodes

Consistent hashing maps keys and physical nodes onto a shared 128-bit ring, ensuring that cluster resizing operations only migrate a minimal fraction of token ranges.

• Ring Layout: The keyspace spans from 0 to 2¹²⁸ - 1 using cryptographic hashing (such as MD5 or MurmurHash3) applied to the partition key.
• Virtual Nodes (vnodes): Each physical node is assigned multiple virtual tokens (typically 128 to 256 vnodes) spread across the ring. This balances token distribution, handles heterogeneous hardware capacities, and ensures smooth load redistribution when nodes join or leave.
• Replica Selection: A key maps to a point on the ring, and the first physical node encountered clockwise serves as the primary replica. The next N-1 distinct physical nodes clockwise form the replica set.
• Rack Awareness: The ring topology maps replica assignments across distinct server racks and availability zones to prevent concurrent outages from single-switch or power failures.

Storage Engine: LSM-Tree Architecture

Log-Structured Merge-trees optimize write performance by converting random updates into sequential disk operations across memory and disk structures.

Write Workflow

1. Write-Ahead Log (WAL): The mutation is appended to an on-disk sequential log and flushed to provide durable local crash recovery.
   • Entry format: CRC (4B) | Timestamp (8B) | Key Length (4B) | Value Length (4B) | Key | Value
   • Recovery: On unexpected node restart, the system replays WAL records to reconstruct the active MemTable state locally. If a node suffers unrecoverable local disk loss, the cluster reconstructs its partitions by streaming data from surviving replica peers.
   • Lifecycle: WAL segments are truncated once their corresponding MemTable has fully flushed to disk.
2. In-Memory MemTable: The record is inserted into a concurrent skip list that maintains keys in sorted order in memory with O(log n) time complexity.
3. MemTable Flushing: When the active MemTable reaches its memory threshold (such as 64 MB), it transitions to an immutable state and a new active MemTable is initialized. A background thread writes the immutable MemTable to disk as a sorted, immutable SSTable file.

Read Workflow

1. Check the active in-memory MemTable for the most recent version of the key.
2. Check any pending immutable MemTables awaiting disk flush.
3. Evaluate the Bloom filter for each relevant on-disk SSTable. If the filter indicates the key is definitely not present, skip disk access for that file entirely.
4. If the Bloom filter indicates the key may be present, perform a binary search on the SSTable sparse index to locate the target data block offset on disk.
5. Read and decompress the data block from disk, scan for the exact key match, and return the record with the latest timestamp or vector clock.

SSTable Internal Structure

• Data Blocks: Compressed sequential blocks containing sorted key-value pairs along with mutation timestamps and deletion markers.
• Sparse Index Block: Contains offsets for every Nth key (such as every 4 KB boundary), enabling binary searches across data blocks with minimal memory overhead.
• Filter Block: Contains a Bloom filter configured with 10 bits per key, providing an illustrative 1% false-positive rate. This significantly reduces unnecessary disk seeks for absent keys, with the exact reduction depending on key access patterns and the number of SSTables consulted.
• Footer Block: A fixed-size 48-byte trailer containing byte offsets and lengths for the index and filter blocks, along with format versions and magic verification bytes.

Compaction Strategies

• Size-Tiered Compaction Strategy (STCS): Merges SSTables of similar file sizes into larger files when the table count exceeds a threshold. This approach minimizes write amplification and works well for write-heavy workloads, though it requires up to 50% temporary free disk space.
• Leveled Compaction Strategy (LCS): Organizes SSTables into fixed levels where each level is 10 times the size of the previous level, and key ranges within levels above Level 0 are non-overlapping. This approach lowers read amplification and space overhead at the expense of higher write amplification during background merges.

Replication and Quorum Models

Replication ensures fault tolerance, while tunable quorums allow applications to balance consistency and latency according to their operational needs.

• Replication Factor (N): Specifies the total number of distinct physical nodes that store copies of each key-value pair, with N=3 being a common production choice. The exact factor is chosen based on durability targets, availability requirements, storage costs, and failure domain boundaries.
• Quorum Overlap (R + W > N): Choosing R and W such that R + W > N ensures that the read quorum and write quorum intersect in at least one replica. This significantly improves read-after-write visibility under the protocol assumptions. However, quorum overlap alone does not guarantee universal linearizability, as concurrent writes, conflict resolution, clock skew, and failure timing still govern final data convergence.
• Tunable Quorum Modes:
  • W=1, R=1: Delivers lowest latency and highest availability with eventual consistency.
  • W=2, R=2 (with N=3): Provides quorum overlap across standard read-write workloads.
  • W=3, R=1: Provides fast read access with synchronous write durability across all replicas.
  • LOCAL_QUORUM: Restricts quorum evaluation to the local datacenter to avoid cross-region network latency.

Gossip Protocol and Failure Detection

A decentralized peer-to-peer gossip protocol maintains cluster membership state and detects node outages without requiring a centralized master.

• State Exchange: Every second, each node selects a random peer and exchanges heartbeat sequences, token ownership maps, and schema metadata.
• Convergence: Information propagates across a cluster of N nodes in O(log N) gossip rounds, achieving global cluster state synchronization within seconds.
• Phi Accrual Failure Detector: Rather than using binary up or down timeouts, nodes calculate a continuous suspicion value ($\Phi$) based on historical heartbeat arrival distributions. A node is declared unreachable when $\Phi$ crosses a configured threshold (such as $\Phi \ge 8$), adapting dynamically to network latency fluctuations.

Hinted Handoff for Temporary Outages

Hinted handoff allows write operations to succeed even when one of the designated replica nodes is temporarily unreachable, preserving write availability in AP mode.

• Mechanism: If a coordinator fails to contact a designated replica during a write, it writes the payload to a healthy substitute node along with a hint containing the target node ID, key, value, and timestamp.
• Replay Delivery: The substitute node monitors gossip heartbeats for the target replica. Once the primary replica returns to service, the substitute streams the buffered hinted writes and deletes the hints upon verified acknowledgment.
• Hint Expiration Window: Hints are retained for a bounded duration (typically 3 hours). If the target node remains down past the TTL window, the hints are discarded and the node is reconciled via Merkle tree anti-entropy repair.

Anti-Entropy Repair Using Merkle Trees

Anti-entropy repair runs as a background process to detect and reconcile data divergence across replicas caused by dropped hints or extended network partitions.

• Tree Structure: Each node maintains a binary Merkle tree for each token range it manages. Leaf nodes represent hashes of individual key-value pairs, while parent nodes represent cryptographic hashes of their children.
• Comparison Workflow: Two replicas compare root hashes for a shared token range. If root hashes match, the replicas are identical and no data transfer occurs. If root hashes differ, nodes traverse down tree branches to pinpoint the exact divergent subtrees.
• Bandwidth Optimization: For a token range with 1 million keys, comparing Merkle tree branches identifies divergent keys in O(log N x diff_count) operations, synchronizing only out-of-date records without transferring the full dataset.

Conflict Resolution Mechanisms

Concurrent writes across distributed replicas require deterministic conflict resolution strategies to resolve divergent versions.

• Vector Clocks (Amazon Dynamo):
  • Each mutation attaches a vector clock tracking logical version counters per node (such as node_A: 3, node_B: 1).
  • If one version causally dominates the other, the dominant version is selected automatically. If versions are concurrent, the system preserves both siblings and returns them to the client application for domain-specific reconciliation.
  • To prevent unbounded vector clock growth, clocks are pruned using threshold timestamps when the entry count exceeds configured limits.
• Last-Write-Wins (Apache Cassandra):
  • Each write attaches a timestamp, and the mutation with the highest timestamp is chosen as the authoritative winner while older versions are discarded.
  • To prevent physical clock skew from silently overwriting newer data, production implementations use Hybrid Logical Clocks (HLC) that combine physical wall-clock time with logical sequence counters.

Coordinator Node Workflow

Any node in the cluster can accept a client request and coordinate its execution across the appropriate replica set.

• Write Coordination:
  1. The coordinator appends the mutation to its local Write-Ahead Log and inserts it into its active MemTable.
  2. The coordinator forwards the write request concurrently to the remaining N-1 replica nodes responsible for the key's token range.
  3. The coordinator waits for W total acknowledgments (including its own local write).
  4. Once W acknowledgments are received, the coordinator returns success to the client. If a replica is unreachable, a hinted handoff payload is stored on an available node.
• Read Coordination and Read Repair:
  1. The coordinator sends read requests in parallel to R replica nodes.
  2. The coordinator collects the responses and compares their version metadata.
  3. The latest value is returned immediately to the client to satisfy the read latency SLA.
  4. If any replica returned an older version, the coordinator asynchronously dispatches the latest record to the stale replicas in the background via Read Repair.