Designing a Distributed Key-Value Store

Like explaining to a 5-year-old: Imagine a library with millions of books, but the library is so big it needs to be spread across 10 buildings. When you ask for a book, you need to know which building has it. The buildings agree on a rule: “any book whose title starts with A–C is in building 1, D–F in building 2…” That’s how a distributed key-value store works — it divides data across machines using a rule, so every machine knows exactly who has what.

A key-value store is the simplest possible database: you store a value under a key, and you retrieve it by that key. But making it distributed — storing data across hundreds of machines with no single point of failure — is one of the richest problems in computer science. Amazon’s Dynamo paper (2007) is the blueprint for almost every database built in the last 15 years.

Requirements

Functional

• PUT(key, value) — store a value
• GET(key) → value — retrieve a value
• DELETE(key) — remove a value
• Values up to 1 MB; keys up to 256 bytes
• No complex queries — just key lookups

Non-Functional

• Scale: 10 PB total data, 1M operations/second
• Latency: P99 < 10ms for reads and writes
• Availability: 99.999% (less than 5 minutes downtime/year)
• Fault-tolerant: tolerate loss of any 2 nodes without data loss

Consistent Hashing: How We Distribute Data

Naive approach: server = hash(key) % num_servers. Problem: if you add one server, almost every key maps to a different server. All data must be remapped.

Consistent hashing places both servers and keys on a ring. Each key is owned by the first server clockwise from its position on the ring.

             0 (= 360°)
              │
   Server A──▶│◀──── Key: "user:123" (hash = 10°)
       (5°)   │
              │
  (90°)       │
 Server B────▶│
              │
              │◀──── Key: "post:456" (hash = 200°)
  (180°)      │
 Server C────▶│
              │
              └──────

When Server D is added, it takes over a slice of the ring. Only keys in that slice move — on average 1/n of all keys, not all of them.

Virtual nodes: assign each physical server multiple positions on the ring. This prevents hotspots when servers have different capacities.

class ConsistentHashRing:
    def __init__(self, virtual_nodes=150):
        self.ring = SortedDict()  # position → server_id
        self.virtual_nodes = virtual_nodes

    def add_server(self, server_id: str):
        for i in range(self.virtual_nodes):
            pos = hash(f"{server_id}:{i}") % (2**32)
            self.ring[pos] = server_id

    def get_server(self, key: str) -> str:
        pos = hash(key) % (2**32)
        # find first server clockwise from pos
        idx = self.ring.bisect_left(pos)
        if idx == len(self.ring):
            idx = 0
        return self.ring.peekitem(idx)[1]

Replication

Every key is stored on N servers (typically N=3). The primary server and the N-1 clockwise neighbors on the ring all hold a copy.

The three-copy rule analogy: Important documents in a law office are photocopied and stored in three different filing cabinets in three different rooms. If one cabinet burns down, the document survives. That’s replication — multiple copies on different machines so any one failure doesn’t lose your data.

Ring position order: A → B → C → D → E

Key "user:123" owned by A (N=3)
Copies stored on: A, B, C

If A fails: B and C still have the data. Reads continue.
If A recovers: B and C sync it back (anti-entropy).

CAP Theorem: The Fundamental Trade-off

You can only guarantee two of three properties simultaneously:

Property	Meaning
Consistency	Every read gets the latest write (or an error)
Availability	Every request gets a response (no errors)
Partition Tolerance	System keeps working when nodes can’t talk to each other

Network partitions will happen. So the real choice is CP vs AP:

• CP (HBase, ZooKeeper): if there’s a partition, refuse requests rather than serve stale data. Banks choose CP.
• AP (DynamoDB, Cassandra): serve the best available data even if possibly stale. Shopping carts choose AP — it’s better to let someone add to cart than refuse the request.

Quorum: Tunable Consistency

With N=3 replicas, define:

• W = number of replicas that must confirm a write
• R = number of replicas that must respond to a read

The rule: if W + R > N, reads always see the latest write.

W	R	Behavior
3	1	Strong consistency, slow writes
1	3	Strong consistency, slow reads
2	2	Balanced — most common
1	1	Fast, but may read stale data

DynamoDB defaults to W=1, R=1 with eventual consistency, but offers R=2 (quorum read) for strong consistency at extra cost.

Handling Failures: Hinted Handoff

The neighbor analogy: Your regular postman is sick. Another postman takes your mail and puts a sticky note on it: “deliver to house 42 when the regular postman is back.” That’s hinted handoff — a healthy node temporarily stores a write meant for a failed node, and delivers it when the failed node recovers.

def write(key, value, context):
    nodes = get_preference_list(key, N=3)
    healthy_nodes = [n for n in nodes if is_healthy(n)]

    if len(healthy_nodes) < W:
        # Use "sloppy quorum": write to W healthy nodes outside preference list
        extra_nodes = get_extra_nodes(N - len(healthy_nodes))
        for node in extra_nodes:
            node.write_with_hint(key, value, target=unhealthy_node_id)
    else:
        for node in healthy_nodes[:W]:
            node.write(key, value)

When the failed node comes back online, it receives the hinted writes and becomes consistent again.

Conflict Resolution: Vector Clocks

With concurrent writes and eventual consistency, two clients might write different values to the same key at the same time. How do you know which is newer?

The version number analogy: Two friends both edit a shared Google Doc offline. When they reconnect, Google needs to know if their changes conflict. It looks at the version history — if one person’s version descends from the other’s, it’s easy (keep the newer one). If neither descends from the other, they both made changes to the same version — that’s a conflict.

Vector clocks track the version history of a value per-node:

Initial:  {} 

Client A writes "john":  {ServerA: 1}
Client B reads "john":   {ServerA: 1}
Client B writes "jane":  {ServerA: 1, ServerB: 1}   ← B's version descends from A's

Concurrent writes:
Client C reads at {ServerA: 1}
Client C writes "jim":   {ServerA: 1, ServerC: 1}
Client B writes "jane":  {ServerA: 1, ServerB: 1}

Now ServerA has both {ServerA:1, ServerC:1} and {ServerA:1, ServerB:1}
Neither descends from the other → CONFLICT

Resolution options:
1. Last-write-wins (LWW): keep the one with the latest timestamp
2. Return both to client and let application merge (DynamoDB's approach)
3. CRDTs: data structures that merge automatically (e.g. counters, sets)

Read Repair & Anti-Entropy

Even without failures, replicas can drift. Two background processes fix this:

Read repair: when a read fetches from multiple replicas (quorum), compare versions. If one is stale, update it immediately.

Anti-entropy (Merkle trees): each node builds a hash tree over its key space. Nodes periodically compare Merkle trees — subtrees with different hashes have diverged data that needs syncing. This is much faster than comparing every key individually.

          root_hash
         /          \
    hash(A-M)    hash(N-Z)  ← only this subtree differs
                /         \
          hash(N-R)    hash(S-Z)  ← divergence is here

Write Path

def handle_write(key: str, value: bytes) -> Response:
    # 1. Determine which nodes own this key
    nodes = consistent_ring.get_preference_list(key, n=3)

    # 2. Write to W nodes in parallel
    futures = [node.write_async(key, value, vector_clock) for node in nodes[:W]]
    results = await asyncio.gather(*futures, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception)]
    if len(successes) < W:
        raise QuorumNotReached()

    # 3. Write remaining replicas asynchronously (don't wait)
    for node in nodes[W:]:
        node.write_async(key, value, vector_clock)  # fire and forget

    return Response(status=200, version_clock=new_clock)

Trade-offs & Decisions

Decision	Choice	Reason
Partitioning	Consistent hashing with virtual nodes	Minimizes key remapping on node changes; even distribution
Replication	N=3 replicas	Tolerates 2 simultaneous failures
Consistency	Quorum (W+R>N)	Tunable; default AP with optional strong consistency
Conflict resolution	Vector clocks + client merge	Preserves causality; lets application decide
Failure handling	Hinted handoff	Maintains availability during temporary node failure
Background sync	Merkle tree anti-entropy	Efficient divergence detection without full key comparison

---

Building a distributed key-value store forces you to confront every distributed systems trade-off at once: consistency vs availability, latency vs durability, operational simplicity vs correctness. The Amazon Dynamo paper is worth reading cover-to-cover — it’s the clearest articulation of these trade-offs ever published.