Designing a Distributed Key-Value Store: Partitioning, Replication, and Consistency Guarantees

A distributed key-value store sounds deceptively simple: put a key, get a value. The hard part is what happens when you spread that data across ten machines and any of them can fail at any moment. Every design decision compounds: how you partition affects how you replicate; how you replicate affects what consistency guarantees you can offer; what consistency you offer determines how you resolve conflicts.

This article walks through the full design from partitioning through conflict resolution, with TypeScript showing the mechanics at each layer.

Partitioning with Consistent Hashing

The naive approach is modular hashing: nodeIndex = hash(key) % numNodes. This works until you add or remove a node, at which point nearly every key remaps to a different node, triggering a massive data migration.

Consistent hashing solves this. You place both nodes and keys on a hash ring (a circle from 0 to 2^32-1). A key is owned by the first node clockwise from its position on the ring. When a node is added or removed, only the keys in its arc need to move.

type HashRingNode = {
  id: string;
  virtualNodes: number[];
};

class ConsistentHashRing {
  private ring: Map<number, string> = new Map();
  private sortedPositions: number[] = [];

  addNode(nodeId: string, virtualNodeCount = 150): void {
    for (let i = 0; i < virtualNodeCount; i++) {
      const position = this.hash(`${nodeId}:${i}`);
      this.ring.set(position, nodeId);
      this.sortedPositions.push(position);
    }
    this.sortedPositions.sort((a, b) => a - b);
  }

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

  getNode(key: string): string {
    if (this.ring.size === 0) throw new Error("Ring is empty");
    const keyHash = this.hash(key);
    // Binary search for the first position >= keyHash
    const idx = this.binarySearchCeiling(keyHash);
    const position = this.sortedPositions[idx % this.sortedPositions.length];
    return this.ring.get(position)!;
  }

  getPreferenceList(key: string, count: number): string[] {
    if (this.ring.size === 0) return [];
    const keyHash = this.hash(key);
    let idx = this.binarySearchCeiling(keyHash);
    const nodes = new Set<string>();
    while (nodes.size < count && nodes.size < this.nodeCount()) {
      const position = this.sortedPositions[idx % this.sortedPositions.length];
      nodes.add(this.ring.get(position)!);
      idx++;
    }
    return [...nodes];
  }

  private nodeCount(): number {
    return new Set(this.ring.values()).size;
  }

  private binarySearchCeiling(target: number): number {
    let lo = 0;
    let 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;
  }

  private hash(input: string): number {
    // FNV-1a: fast, good distribution
    let h = 2166136261;
    for (let i = 0; i < input.length; i++) {
      h ^= input.charCodeAt(i);
      h = Math.imul(h, 16777619);
    }
    return h >>> 0;
  }
}

The critical detail is the virtual node count. With a single position per physical node, the ring distributes load unevenly because hash collisions and arcs of different sizes are common with small node counts. At 150 virtual nodes per physical node, the coefficient of variation for load distribution drops below 10%, which is acceptable for most production workloads. Dynamo used 150; Cassandra moved to a token-based approach that achieves similar distribution with less metadata.

getPreferenceList is the method that feeds replication: given a key, return the N nodes responsible for storing it. This list is ordered, and the first node in the list is the coordinator for writes.

Replication Strategies

Once you have a preference list per key, you need to decide how to replicate across those nodes.

Leader-Follower Replication

One node (the leader) accepts all writes for a key. It replicates to followers synchronously or asynchronously before acknowledging the client.

type ReplicationMode = "sync" | "async" | "semi-sync";

type WriteResult = {
  success: boolean;
  version: number;
  replicatedTo: string[];
};

async function leaderWrite(
  key: string,
  value: Uint8Array,
  followers: string[],
  mode: ReplicationMode
): Promise<WriteResult> {
  // Write to local storage first
  const version = await localStore.put(key, value);
  const replicatedTo: string[] = ["self"];

  if (mode === "sync") {
    // Wait for all followers before acking client
    await Promise.all(
      followers.map((f) => replicateToFollower(f, key, value, version))
    );
    replicatedTo.push(...followers);
  } else if (mode === "semi-sync") {
    // Wait for at least one follower, replicate the rest async
    await replicateToFollower(followers[0], key, value, version);
    replicatedTo.push(followers[0]);
    // Fire and forget for remaining followers
    followers
      .slice(1)
      .forEach((f) => replicateToFollower(f, key, value, version));
  } else {
    // Async: ack immediately, replicate in background
    followers.forEach((f) => replicateToFollower(f, key, value, version));
  }

  return { success: true, version, replicatedTo };
}

Synchronous replication gives you strong durability but adds the latency of a network round-trip to your write path. If any follower is slow, your p99 write latency climbs. Semi-sync is the common production middle ground: one synchronous replica plus async for the rest.

Leaderless Replication (Quorum)

Leaderless replication, popularized by Dynamo-style systems, removes the coordinator concept for writes. Any node can accept a write for any key. The client (or a proxy) writes to multiple nodes simultaneously and considers the write successful when W nodes acknowledge it. Reads go to R nodes and pick the most recent value.

The rule that makes this work: W + R > N (where N is the replication factor). This guarantees read-your-own-write consistency because at least one node in the read quorum must have seen the latest write.

type QuorumConfig = {
  n: number; // total replicas
  w: number; // write quorum
  r: number; // read quorum
};

type VersionedValue = {
  value: Uint8Array;
  version: number;
  timestamp: number;
};

async function quorumWrite(
  key: string,
  value: Uint8Array,
  nodes: string[],
  config: QuorumConfig
): Promise<{ success: boolean; version: number }> {
  const version = Date.now(); // logical clock or vector clock in production
  const results = await Promise.allSettled(
    nodes.map((node) => writeToNode(node, key, value, version))
  );

  const successCount = results.filter((r) => r.status === "fulfilled").length;
  if (successCount < config.w) {
    throw new Error(
      `Write quorum not met: ${successCount}/${config.w} required`
    );
  }

  return { success: true, version };
}

async function quorumRead(
  key: string,
  nodes: string[],
  config: QuorumConfig
): Promise<VersionedValue | null> {
  const results = await Promise.allSettled(
    nodes.map((node) => readFromNode(node, key))
  );

  const values = results
    .filter(
      (r): r is PromiseFulfilledResult<VersionedValue> =>
        r.status === "fulfilled" && r.value !== null
    )
    .map((r) => r.value);

  if (values.length < config.r) {
    throw new Error(`Read quorum not met: ${values.length}/${config.r} required`);
  }

  // Return the highest-version value
  return values.reduce((best, current) =>
    current.version > best.version ? current : best
  );
}

Common quorum configurations for a replication factor of 3:

• W=3, R=1: Slow writes (all replicas must ack), fast reads, strong consistency.
• W=1, R=3: Fast writes, slow reads, strong consistency.
• W=2, R=2: Balanced latency, tolerates 1 node failure, strong consistency.
• W=1, R=1: Maximum availability, eventual consistency only.

Consistency Models

The quorum math above gives you strong consistency if configured correctly, but strong consistency has a cost you need to understand before defaulting to it.

Strong consistency means every read returns the most recently written value. In a quorum system this requires W + R > N. The cost is latency: you wait for the slowest node in your quorum on every read and write. Under network partitions (the P in CAP), you must choose to refuse requests rather than serve potentially stale data.

Eventual consistency means all replicas will converge to the same value given enough time and no further writes. You can achieve this with W=1, R=1. Reads may return stale data, but every write is immediately acknowledged and availability is maximized.

Causal consistency sits between the two. Causally related operations appear in order everywhere. If you write key A and then key B (where B’s value depends on A), any client that reads B also sees the value of A that preceded it. This is implemented using vector clocks or logical timestamps, and it’s the model used by systems like MongoDB in its causal sessions.

The practical question is: what does your application actually need? Most applications need read-your-own-write consistency (you see your own updates), not strict linearizability across all clients. Quorum with W + R > N gives you the former cheaply. Linearizability requires consensus protocols like Paxos or Raft, which are significantly more expensive.

Conflict Resolution

When two writes happen concurrently to the same key on different replicas, you have a conflict. How you resolve it depends on your data semantics.

Last-Write-Wins

The simplest approach: the write with the highest timestamp wins. This works only if your clocks are synchronized (they aren’t, precisely, even with NTP) and if it’s acceptable to silently drop a write. It’s used in Cassandra as the default strategy. The data loss is real: if two clients update a user’s address concurrently, one update disappears without error.

Vector Clocks

A vector clock is a map from node ID to a counter. Each node increments its own counter on every write. When comparing two versions, if every counter in version A is less than or equal to version B, then B is causally later. If neither version dominates, they are concurrent and you have a true conflict.

type VectorClock = Record<string, number>;

function incrementClock(clock: VectorClock, nodeId: string): VectorClock {
  return { ...clock, [nodeId]: (clock[nodeId] ?? 0) + 1 };
}

type ClockRelation = "before" | "after" | "concurrent" | "equal";

function compareClock(a: VectorClock, b: VectorClock): ClockRelation {
  const allKeys = new Set([...Object.keys(a), ...Object.keys(b)]);
  let aLessOrEqual = true;
  let bLessOrEqual = true;

  for (const key of allKeys) {
    const av = a[key] ?? 0;
    const bv = b[key] ?? 0;
    if (av > bv) bLessOrEqual = false;
    if (bv > av) aLessOrEqual = false;
  }

  if (aLessOrEqual && bLessOrEqual) return "equal";
  if (aLessOrEqual) return "before";
  if (bLessOrEqual) return "after";
  return "concurrent";
}

function mergeClock(a: VectorClock, b: VectorClock): VectorClock {
  const merged: VectorClock = { ...a };
  for (const [key, val] of Object.entries(b)) {
    merged[key] = Math.max(merged[key] ?? 0, val);
  }
  return merged;
}

type StoredValue = {
  value: Uint8Array;
  clock: VectorClock;
  siblings: StoredValue[]; // concurrent versions
};

When compareClock returns "concurrent", you store both versions as siblings. On the next read, you surface all siblings to the client and let the application merge them. This is the approach Riak takes. The downside is that sibling proliferation can get out of control if clients don’t read-and-repair on every operation.

CRDTs

Conflict-free Replicated Data Types sidestep conflict resolution by using data structures whose merge operation is always correct and commutative. Two examples that cover many real use cases:

A G-Counter (grow-only counter) keeps a per-node count. Merging two G-Counters takes the max of each per-node counter. The total value is the sum of all per-node counts.

type GCounter = Record<string, number>;

function gcIncrement(counter: GCounter, nodeId: string, by = 1): GCounter {
  return { ...counter, [nodeId]: (counter[nodeId] ?? 0) + by };
}

function gcMerge(a: GCounter, b: GCounter): GCounter {
  const merged: GCounter = { ...a };
  for (const [nodeId, count] of Object.entries(b)) {
    merged[nodeId] = Math.max(merged[nodeId] ?? 0, count);
  }
  return merged;
}

function gcValue(counter: GCounter): number {
  return Object.values(counter).reduce((sum, v) => sum + v, 0);
}

// PN-Counter: two G-Counters, one for increments, one for decrements
type PNCounter = { pos: GCounter; neg: GCounter };

function pnAdd(counter: PNCounter, nodeId: string, delta: number): PNCounter {
  if (delta >= 0) {
    return { ...counter, pos: gcIncrement(counter.pos, nodeId, delta) };
  }
  return { ...counter, neg: gcIncrement(counter.neg, nodeId, -delta) };
}

function pnMerge(a: PNCounter, b: PNCounter): PNCounter {
  return { pos: gcMerge(a.pos, b.pos), neg: gcMerge(a.neg, b.neg) };
}

function pnValue(counter: PNCounter): number {
  return gcValue(counter.pos) - gcValue(counter.neg);
}

CRDTs are not a universal answer. They work for specific data types (sets, counters, registers) where the semantics align with their merge properties. A PN-Counter can’t go negative from a correctness standpoint at the counter level, but it can report a negative sum if decrements exceed increments at the total level. For arbitrary application data, vector clocks with application-level merge are often more honest.

Failure Detection and Recovery

Detecting that a node has failed (as opposed to just being slow) is harder than it sounds. A naive timeout-based check from a single node is unreliable: the checking node itself might have a degraded network path.

Production systems use gossip protocols. Each node maintains a heartbeat counter that it increments every second and gossips to a random subset of peers. A node is suspected when another node hasn’t seen its heartbeat increment in N seconds. It’s declared dead when a configurable number of additional seconds pass. The key property: suspicion and death declarations spread through the cluster without a central coordinator.

type HeartbeatState = {
  nodeId: string;
  counter: number;
  updatedAt: number; // local timestamp when we last saw this counter change
};

type GossipMessage = {
  from: string;
  heartbeats: HeartbeatState[];
};

class FailureDetector {
  private heartbeats: Map<string, HeartbeatState> = new Map();
  private suspectThresholdMs = 10_000;
  private deadThresholdMs = 30_000;

  processGossip(msg: GossipMessage): void {
    for (const incoming of msg.heartbeats) {
      const existing = this.heartbeats.get(incoming.nodeId);
      if (!existing || incoming.counter > existing.counter) {
        this.heartbeats.set(incoming.nodeId, {
          ...incoming,
          updatedAt: Date.now(),
        });
      }
    }
  }

  getNodeStatus(nodeId: string): "alive" | "suspected" | "dead" {
    const state = this.heartbeats.get(nodeId);
    if (!state) return "dead";
    const elapsed = Date.now() - state.updatedAt;
    if (elapsed > this.deadThresholdMs) return "dead";
    if (elapsed > this.suspectThresholdMs) return "suspected";
    return "alive";
  }

  buildGossipPayload(): HeartbeatState[] {
    return [...this.heartbeats.values()];
  }
}

When a node returns after a failure, it needs to catch up on writes it missed. This is handled by hinted handoff during the outage and read repair after recovery.

Hinted handoff: When a write cannot be delivered to its target node, a healthy node in the preference list stores the write with a hint indicating which node it belongs to. When the target recovers, the hints are replayed.

Read repair: During a quorum read, if the responding nodes return different versions of a value, the coordinator writes the most recent version back to any node that returned a stale value. This is an opportunistic reconciliation that happens in the background after the read is already returned to the client.

async function readWithRepair(
  key: string,
  nodes: string[],
  config: QuorumConfig
): Promise<VersionedValue | null> {
  const responses = await Promise.allSettled(
    nodes.map(async (node) => ({ node, data: await readFromNode(node, key) }))
  );

  const values = responses
    .filter(
      (
        r
      ): r is PromiseFulfilledResult<{
        node: string;
        data: VersionedValue | null;
      }> => r.status === "fulfilled" && r.value.data !== null
    )
    .map((r) => r.value);

  if (values.length < config.r) {
    throw new Error("Read quorum not met");
  }

  const latest = values.reduce((best, current) =>
    current.data!.version > best.data!.version ? current : best
  );

  // Repair stale nodes in the background
  const staleNodes = values
    .filter((v) => v.data!.version < latest.data!.version)
    .map((v) => v.node);

  if (staleNodes.length > 0) {
    // Don't await: repair is background work
    Promise.all(
      staleNodes.map((node) =>
        writeToNode(node, key, latest.data!.value, latest.data!.version)
      )
    ).catch((err) => console.error("Read repair failed:", err));
  }

  return latest.data;
}

Production Considerations

Storage Engine: LSM Trees

Key-value stores almost universally use Log-Structured Merge trees instead of B-trees for the storage layer. Writes go to an in-memory buffer (memtable) and an append-only write-ahead log. When the memtable reaches a size threshold, it’s flushed to disk as an immutable sorted string table (SSTable). Reads must check the memtable plus potentially multiple SSTables, which is why bloom filters per SSTable are non-negotiable: a bloom filter lookup tells you with certainty whether a key is absent, avoiding disk reads for most misses.

The cost of LSM trees is compaction. SSTables accumulate, and periodic compaction merges and sorts them, dropping deleted keys and old versions. Compaction runs in the background but consumes disk I/O and can affect read latency. Size-tiered compaction (merge tables of similar size) minimizes write amplification at the cost of space amplification. Leveled compaction (maintain sorted levels with size limits) minimizes space amplification but increases write amplification. Choose based on whether your bottleneck is write throughput or disk space.

Tradeoffs Table

Dimension	Strong Consistency	Eventual Consistency	Causal Consistency
Read latency	Higher (quorum wait)	Lower (single replica)	Medium (version tracking)
Write latency	Higher (quorum ack)	Lower (single ack)	Medium
Availability under partition	Lower (refuse requests)	Higher (accept all ops)	Medium
Conflict handling	None needed	Requires resolution strategy	Limited (causal order enforced)
Use case fit	Financial records, inventory	User profiles, caches, analytics	Shopping carts, collaborative docs
Implementation complexity	Medium	Low to high (CRDT/VC adds complexity)	High

Observability

The metrics that matter in production:

• Read repair rate: If this spikes, your cluster has significant replica divergence. Could indicate a slow follower or compaction falling behind.
• Hinted handoff queue depth: If hints are accumulating, a target node is taking too long to recover. If hints expire before the node recovers, you have permanent data loss.
• Quorum failure rate: Reads or writes failing to meet quorum. This is your availability signal; alert on it before users do.
• Compaction lag: Measure the ratio of SSTable count to the expected level count. A growing ratio means compaction is falling behind writes, which degrades read performance.
• Vector clock sibling count: For conflict resolution systems, a growing sibling count means clients aren’t reading and resolving. Surface this per-key, not just as an aggregate.

Anti-Entropy

Read repair only fixes keys that are actively read. Keys that are written once and rarely read can diverge silently. Anti-entropy is a background process that walks the key space on each node and compares checksums with peers, repairing any discrepancies. Merkle trees make this efficient: build a hash tree over the key range, compare tree roots, and drill down only into the subtrees that differ. This bounds the per-node work to O(differences), not O(total keys).

The Design Decision Framework

Start with your consistency requirements:

• Your data requires linearizability (banking, inventory): use leader-follower with synchronous replication or a consensus-based log. Accept the availability and latency tradeoffs.
• Your data can tolerate staleness with bounded convergence (user preferences, caches): use leaderless quorum with W=1, R=1 or W=2, R=2. Add anti-entropy for background repair.
• Your data is naturally mergeable (counters, sets, append-only logs): CRDTs eliminate conflicts by design. They are not a fit for arbitrary mutable records.
• Your data has causal dependencies (social feeds, collaborative state): causal consistency with vector clocks is the right level. Full linearizability is more expensive than you need.

Every distributed key-value store is a set of tradeoffs encoded in configuration. The CAP theorem is not a constraint you overcome; it is a description of your options. Understanding where your application actually sits on the consistency/availability spectrum lets you choose the right settings rather than cargo-culting a popular system’s defaults.