Designing a Distributed Key-Value Store: Storage Engines, Replication, and Tunable Consistency

A distributed key-value store is one of the most instructive systems to design from scratch. It forces every major distributed systems decision into focus at once: how you write data to disk, how you spread it across nodes, how you keep replicas in sync, and how you trade consistency against availability when things break. DynamoDB, Cassandra, Redis Cluster, etcd, and RocksDB-backed systems all make different choices at each layer. This article traces those choices from the storage engine up to the replication protocol, with enough depth to make real implementation decisions.

Single-Node Storage: LSM Trees vs B-Trees

Before a key-value store is distributed, it needs a storage engine. The two dominant approaches are B-trees and Log-Structured Merge (LSM) trees, and they optimize for opposite workloads.

B-trees keep data sorted in a balanced tree of fixed-size pages (typically 4 KB or 8 KB). Reads are fast because you navigate from root to leaf in O(log n). Writes are slow under high throughput because every write may require multiple random page reads and writes, including a write-ahead log entry for crash safety.

LSM trees accept all writes into an in-memory buffer (the memtable), then periodically flush sorted runs to disk (SSTables). Writes are always sequential, which is dramatically faster for write-heavy workloads. Reads are slower because you may need to check the memtable plus several SSTable levels. A Bloom filter per SSTable reduces unnecessary disk reads.

interface SSTableEntry {
  key: string;
  value: Buffer | null; // null = tombstone (deletion marker)
  sequenceNumber: bigint;
}

interface SSTable {
  level: number;
  entries: SSTableEntry[]; // sorted by key
  bloomFilter: BloomFilter;
  minKey: string;
  maxKey: string;
  sizeBytes: number;
}

interface Memtable {
  entries: Map<string, { value: Buffer | null; sequenceNumber: bigint }>;
  sizeBytes: number;
  maxSizeBytes: number;
}

class LSMStorageEngine {
  private memtable: Memtable;
  private immutableMemtable: Memtable | null = null;
  private levels: SSTable[][] = [[], [], [], [], [], [], []]; // L0 through L6
  private sequenceCounter = 0n;
  private wal: WriteAheadLog;

  async put(key: string, value: Buffer): Promise<void> {
    const seq = ++this.sequenceCounter;
    await this.wal.append({ type: "put", key, value, seq });
    this.memtable.entries.set(key, { value, sequenceNumber: seq });
    this.memtable.sizeBytes += key.length + value.length;

    if (this.memtable.sizeBytes >= this.memtable.maxSizeBytes) {
      await this.rotateMemtable();
    }
  }

  async get(key: string): Promise<Buffer | null> {
    // Check memtable first (most recent writes)
    const memEntry = this.memtable.entries.get(key);
    if (memEntry !== undefined) {
      return memEntry.value; // null means deleted
    }

    // Check immutable memtable if present
    if (this.immutableMemtable) {
      const immEntry = this.immutableMemtable.entries.get(key);
      if (immEntry !== undefined) return immEntry.value;
    }

    // Search SSTable levels from L0 (newest) to L6 (oldest)
    for (const level of this.levels) {
      for (const sstable of level.reverse()) {
        if (key < sstable.minKey || key > sstable.maxKey) continue;
        if (!sstable.bloomFilter.mightContain(key)) continue;
        const entry = await this.readFromSSTable(sstable, key);
        if (entry !== undefined) {
          return entry.value;
        }
      }
    }
    return null;
  }

  private async rotateMemtable(): Promise<void> {
    this.immutableMemtable = this.memtable;
    this.memtable = this.createFreshMemtable();
    // Flush immutable memtable to L0 in background
    setImmediate(() => this.flushToL0());
  }
}

The key property of the LSM design: writes hit the WAL and memtable only, both sequential operations. The tradeoff is read amplification (checking multiple SSTables) and the ongoing cost of compaction.

Compaction Strategies

Compaction merges SSTables to reclaim space from deleted keys and limit read amplification. The two main strategies have real operational differences.

Size-tiered compaction groups SSTables of similar size and merges them when a tier has enough files. This is write-optimized: you merge infrequently and in large batches. The problem is space amplification. During a merge, you temporarily hold both the input and output SSTables on disk, potentially doubling storage usage.

Leveled compaction (used by LevelDB and RocksDB) maintains size limits per level. L1 holds at most 10 MB, L2 holds 100 MB, and so on by a factor of 10. Within each level, key ranges do not overlap across SSTables. Reads are fast because at most one SSTable per level (L1+) can contain a given key. The cost is write amplification: a key written once may be rewritten many times as it moves down the levels.

Cassandra uses a hybrid: STCS for write-heavy workloads, TWCS (Time-Window Compaction Strategy) for time-series data where old data is rarely updated, and LCS (Leveled) for read-heavy workloads. Picking the wrong compaction strategy for your access pattern will hurt you in production.

Distributing Data: Consistent Hashing

Once your storage engine is solid, you need to spread data across multiple nodes. Naive modulo hashing (hash(key) % node_count) means that adding or removing a node rebalances nearly all keys. Consistent hashing solves this.

Each node is assigned multiple positions on a hash ring (virtual nodes, or vnodes). A key is assigned to the first node whose position is clockwise from hash(key) on the ring. Adding a node only moves keys from the successor node, not from all nodes. Removing a node only moves its keys to its successor.

class ConsistentHashRing {
  private ring = new Map<number, string>(); // position -> nodeId
  private sortedPositions: number[] = [];
  private readonly vnodeCount: number;

  constructor(vnodeCount = 150) {
    this.vnodeCount = vnodeCount;
  }

  addNode(nodeId: string): void {
    for (let i = 0; i < this.vnodeCount; i++) {
      const position = this.hash(`${nodeId}:${i}`);
      this.ring.set(position, nodeId);
      this.insertSorted(position);
    }
  }

  removeNode(nodeId: string): void {
    for (let i = 0; i < this.vnodeCount; i++) {
      const position = this.hash(`${nodeId}:${i}`);
      this.ring.delete(position);
      this.sortedPositions = this.sortedPositions.filter((p) => p !== position);
    }
  }

  getPreferenceList(key: string, replicationFactor: number): string[] {
    const startPosition = this.hash(key);
    const nodes: string[] = [];
    const seen = new Set<string>();

    let idx = this.lowerBound(startPosition);
    while (nodes.length < replicationFactor) {
      if (idx >= this.sortedPositions.length) idx = 0;
      const nodeId = this.ring.get(this.sortedPositions[idx])!;
      if (!seen.has(nodeId)) {
        nodes.push(nodeId);
        seen.add(nodeId);
      }
      idx++;
    }
    return nodes;
  }

  private hash(input: string): number {
    // In production, use MurmurHash3 or xxHash for uniform distribution
    let h = 0x811c9dc5;
    for (let i = 0; i < input.length; i++) {
      h ^= input.charCodeAt(i);
      h = (h * 0x01000193) >>> 0;
    }
    return h;
  }

  private lowerBound(target: number): number {
    let lo = 0, hi = this.sortedPositions.length;
    while (lo < hi) {
      const mid = (lo + hi) >>> 1;
      if (this.sortedPositions[mid] < target) lo = mid + 1;
      else hi = mid;
    }
    return lo % this.sortedPositions.length;
  }

  private insertSorted(pos: number): void {
    const idx = this.lowerBound(pos);
    this.sortedPositions.splice(idx, 0, pos);
  }
}

The getPreferenceList returns the ordered list of nodes responsible for a key. The first node in that list is the coordinator for requests to that key; the rest are replicas.

Replication: Raft vs Leaderless Quorum

With partitioning solved, you need replication. The two approaches differ fundamentally in their consistency and operational characteristics.

Raft (used by etcd, CockroachDB, and TiKV) elects a single leader per partition. All writes go through the leader, which appends to its log and replicates to followers before acknowledging the client. This gives you linearizability: reads from the leader reflect all committed writes. The cost is latency on leader election and the loss of availability during an election round when the leader fails. (For the full Raft protocol, see the dedicated article on Raft consensus and log replication.)

Leaderless replication (used by Cassandra and DynamoDB’s original design) routes writes and reads to multiple nodes simultaneously. Any node in the preference list can accept a write. You achieve consistency through quorum: if you have a replication factor N, require W write acknowledgments and R read acknowledgments such that W + R > N.

interface ReplicationConfig {
  replicationFactor: number; // N
  writeQuorum: number;       // W
  readQuorum: number;        // R
}

interface VersionedValue {
  value: Buffer | null;
  vectorClock: Record<string, number>; // nodeId -> logical timestamp
  timestamp: bigint; // wall-clock, for tie-breaking only
}

class LeaderlessReplicator {
  private ring: ConsistentHashRing;
  private nodes: Map<string, NodeClient>;
  private config: ReplicationConfig;

  async put(key: string, value: Buffer): Promise<void> {
    const coordinatorNodeId = this.ring.getPreferenceList(key, 1)[0];
    const preferenceList = this.ring.getPreferenceList(key, this.config.replicationFactor);

    const newVersion: VersionedValue = {
      value,
      vectorClock: this.incrementClock(await this.getCurrentClock(key), coordinatorNodeId),
      timestamp: BigInt(Date.now()),
    };

    const writePromises = preferenceList.map((nodeId) =>
      this.nodes.get(nodeId)!.put(key, newVersion).catch((err) => ({ error: err, nodeId }))
    );

    const results = await Promise.allSettled(writePromises);
    const successes = results.filter((r) => r.status === "fulfilled").length;

    if (successes < this.config.writeQuorum) {
      throw new Error(`Write quorum not met: ${successes}/${this.config.writeQuorum} nodes acknowledged`);
    }
  }

  async get(key: string): Promise<Buffer | null> {
    const preferenceList = this.ring.getPreferenceList(key, this.config.replicationFactor);

    const readPromises = preferenceList.map((nodeId) =>
      this.nodes.get(nodeId)!.get(key).catch(() => null)
    );

    const results = await Promise.allSettled(readPromises);
    const responses = results
      .filter((r): r is PromiseFulfilledResult<VersionedValue | null> => r.status === "fulfilled")
      .map((r) => r.value)
      .filter((v): v is VersionedValue => v !== null);

    if (responses.length < this.config.readQuorum) {
      throw new Error(`Read quorum not met: ${responses.length}/${this.config.readQuorum} responses`);
    }

    // Pick the most recent version by vector clock dominance
    const winner = this.resolveConflict(responses);

    // Read repair: push the winning version back to nodes that had stale data
    this.readRepair(key, winner, preferenceList, responses);

    return winner.value;
  }

  private resolveConflict(versions: VersionedValue[]): VersionedValue {
    // If one version's vector clock dominates all others, it wins
    // If clocks are concurrent (no dominance), use timestamp as tie-breaker
    // In production, expose concurrent versions to the application (like DynamoDB)
    return versions.reduce((a, b) => {
      if (this.dominates(a.vectorClock, b.vectorClock)) return a;
      if (this.dominates(b.vectorClock, a.vectorClock)) return b;
      return a.timestamp > b.timestamp ? a : b; // last-write-wins on concurrent
    });
  }

  private dominates(a: Record<string, number>, b: Record<string, number>): boolean {
    return Object.keys(b).every((k) => (a[k] ?? 0) >= (b[k] ?? 0)) &&
      Object.keys(a).some((k) => (a[k] ?? 0) > (b[k] ?? 0));
  }

  private async readRepair(
    key: string,
    winner: VersionedValue,
    preferenceList: string[],
    received: VersionedValue[]
  ): Promise<void> {
    // Fire-and-forget: update nodes with stale or missing data
    for (const nodeId of preferenceList) {
      const nodeResponse = received.find((_, i) => preferenceList[i] === nodeId);
      if (!nodeResponse || this.dominates(winner.vectorClock, nodeResponse.vectorClock)) {
        this.nodes.get(nodeId)?.put(key, winner).catch(() => {});
      }
    }
  }
}

The quorum math is the core lever for tunable consistency:

• N=3, W=2, R=2: strong consistency (W+R=4 > N=3). Reads always see the latest write.
• N=3, W=1, R=1: eventual consistency. High availability, low latency, stale reads possible.
• N=3, W=3, R=1: write-safe mode. All replicas must confirm before acknowledging. Reads are cheap.

Failure Detection with Gossip

Gossip protocols give every node an eventually consistent view of cluster membership without a central coordinator. Each node maintains a membership list with heartbeat counters and timestamps. Periodically, it selects a random peer and exchanges its list. Both nodes merge the lists, keeping the higher heartbeat counter for each entry.

A node is marked suspect when its heartbeat hasn’t incremented beyond a configurable threshold. It’s declared dead after it stays suspect for a second interval without defending itself. This two-phase approach (used by SWIM, which Cassandra adopts) dramatically reduces false positives from transient network delays. (Full detail on gossip, SWIM, and anti-entropy is covered in the dedicated gossip protocols article.)

Hinted Handoff and Anti-Entropy Repair

When a target replica is temporarily unavailable during a write, the coordinator stores the write locally as a “hint” tagged with the intended destination. Once the node recovers, the coordinator forwards the hinted writes. This preserves write availability without sacrificing durability.

interface HintedHandoff {
  targetNodeId: string;
  key: string;
  value: VersionedValue;
  hintedAt: Date;
  expiresAt: Date; // hints don't live forever
}

class HintedHandoffManager {
  private hints: HintedHandoffStore;

  async storeHint(targetNodeId: string, key: string, value: VersionedValue): Promise<void> {
    const hint: HintedHandoff = {
      targetNodeId,
      key,
      value,
      hintedAt: new Date(),
      expiresAt: new Date(Date.now() + 3 * 60 * 60 * 1000), // 3-hour window
    };
    await this.hints.insert(hint);
  }

  async replayHints(recoveredNodeId: string, nodeClient: NodeClient): Promise<void> {
    const pending = await this.hints.findByTarget(recoveredNodeId);
    for (const hint of pending) {
      if (hint.expiresAt < new Date()) {
        await this.hints.delete(hint);
        continue;
      }
      try {
        await nodeClient.put(hint.key, hint.value);
        await this.hints.delete(hint);
      } catch {
        // Node still unavailable, leave hint for next retry
        break;
      }
    }
  }
}

Hinted handoff handles short outages. For longer outages, anti-entropy repair uses Merkle trees to efficiently find diverged key ranges between replicas. Each node builds a Merkle tree over its key space, where leaf hashes cover individual keys and internal nodes hash their children. Two replicas can compare tree roots in O(log n) comparisons to identify which sub-ranges differ, then sync only the diverged keys rather than scanning everything.

Hot Key Handling

Consistent hashing distributes keys uniformly on average, but high-cardinality access skew breaks that assumption. A single celebrity user’s feed, a viral product, or a shared counter will concentrate traffic on one partition.

Three practical mitigations:

Read replica scattering: For read-heavy hot keys, add a random suffix (1 to k) to the key and store k copies. Reads pick a random suffix. This multiplies read capacity by k at the cost of write fan-out and slight consistency complexity.

Local caching: Keep a fixed-size in-process LRU cache for reads. A hot key read rate of 100K RPS hitting 10 application nodes generates only 10 RPS of backend traffic if the TTL is 100ms.

Shard splitting: Detect hot partitions by monitoring request rates per vnode. When a partition exceeds a threshold, split it into sub-partitions and redistribute. This is operationally heavier but necessary for sustained write-heavy hot keys.

class HotKeyDetector {
  private requestCounts = new Map<string, number>();
  private readonly windowMs = 1000;
  private readonly hotThreshold = 5000; // requests per second per key

  record(key: string): void {
    this.requestCounts.set(key, (this.requestCounts.get(key) ?? 0) + 1);
  }

  getHotKeys(): string[] {
    return Array.from(this.requestCounts.entries())
      .filter(([, count]) => count > this.hotThreshold)
      .map(([key]) => key);
  }

  reset(): void {
    this.requestCounts.clear();
    setTimeout(() => this.reset(), this.windowMs);
  }
}

class ScatteredReadCoordinator {
  private readonly scatterFactor = 10;

  scatteredKey(key: string, isHot: boolean): string {
    if (!isHot) return key;
    const suffix = Math.floor(Math.random() * this.scatterFactor);
    return `${key}:scatter:${suffix}`;
  }

  async scatteredWrite(
    key: string,
    value: VersionedValue,
    replicator: LeaderlessReplicator
  ): Promise<void> {
    // Write to all scatter shards on hot key writes
    const writes = Array.from({ length: this.scatterFactor }, (_, i) =>
      replicator.put(`${key}:scatter:${i}`, value.value!)
    );
    await Promise.all(writes);
  }
}

Consistency, Availability, and Latency Tradeoffs

Different quorum configurations produce fundamentally different systems. This table covers the most common configurations with N=3 replicas:

Configuration	W	R	Consistency	Availability	Read Latency	Write Latency	Use Case
Strong quorum	2	2	Linearizable reads	Tolerates 1 failure	Medium (waits for 2)	Medium	User account state, inventory
Write-heavy	1	3	Linearizable reads	Write highly available	High (waits for 3)	Low	Event ingestion, logs
Read-heavy	3	1	Linearizable reads	Write requires all up	Low	High	Config/feature flags
Eventual	1	1	Eventual	Highest	Lowest	Lowest	Analytics counters, caches
Raft (leader)	N/A	N/A	Linearizable	Leader election gap	Lowest from leader	Medium	Coordination, locks
Leaderless LWW	2	2	Last-write-wins	Tolerates 1 failure	Medium	Medium	Session data, profiles

“Linearizable” in this table means reads reflect all prior acknowledged writes when W+R>N. It does not mean serializability across multiple keys, which requires distributed transactions.

Production Considerations

Compaction throttling: Compaction is CPU and I/O intensive. Unthrottled, it starves foreground reads and writes. Enforce byte/second limits on compaction I/O and schedule heavy compaction during off-peak windows.

WAL management: WAL segments accumulate until all data they cover is flushed to SSTables. If a flush stalls, WAL grows unbounded. Set a maximum WAL size and stall incoming writes if it is exceeded, rather than silently dropping data.

Clock skew in conflict resolution: Wall-clock timestamps used for last-write-wins conflict resolution are only as good as your NTP synchronization. On a cluster with 50ms clock skew, writes within that window can be resolved incorrectly. Vector clocks are more correct; hybrid logical clocks (HLC) combine causality tracking with bounded wall-clock drift for a practical middle ground.

Tombstone accumulation: Deleted keys leave tombstones that are only removed during compaction. A workload with high delete rates will accumulate tombstones, slowing reads (every read scans past tombstones) and delaying space reclamation. Monitor tombstone-to-data ratios and force compaction when they exceed acceptable levels.

Bloom filter sizing: A Bloom filter with 1% false positive rate uses roughly 10 bits per key. For 100 million keys, that is 125 MB per SSTable. Size your Bloom filters against your available memory: higher false positive rates save memory but increase unnecessary disk reads.

Snapshot isolation for backups: Taking a consistent backup across multiple SSTables requires a stable snapshot. Most LSM engines expose a sequence number concept: a snapshot holds the engine at a fixed sequence number, allowing backup to read any SSTable without seeing writes that arrived after snapshot creation.

Closing

Distributed key-value store design is a layered problem. The storage engine determines your write throughput ceiling and compaction cost. Consistent hashing and vnodes determine how cleanly you can add and remove capacity. Quorum configuration is the operational lever that lets you shift the consistency/availability dial at runtime without changing code. Failure detection and repair (gossip, hinted handoff, Merkle-tree anti-entropy) determine how gracefully the system handles the inevitable node outages. Get the storage layer right before adding distribution complexity. Most production failures in this space come from misconfigured compaction or quorum settings, not from the fundamental algorithm choices.