Design a Distributed Key-Value Store

DynamoDB / Cassandra System Design

Overview

A distributed key-value store is the durable backbone under systems like Amazon DynamoDB and Apache Cassandra. The interface is deceptively small — get(key), put(key, value), delete(key) — but the machine behind it must keep your data safe across dozens or thousands of machines, survive nodes dying and networks partitioning, and stay fast and available the entire time. Unlike a cache, this store is the source of truth. There is no database behind it to refill a lost value, so every hard decision — where a key lives, how many copies exist, which copy wins a conflict — has to be correct on its own.

It is labeled "Medium" because the API is tiny and the happy path is simple, but the moment you take durability and availability seriously, three genuinely hard distributed-systems problems surface at once. First, the data no longer fits on one machine, so you must partition the keyspace across many nodes without a mass reshuffle every time the cluster changes — this is consistent hashing. Second, any node can die, so each key must be replicated onto several nodes, and you must decide how many copies a read and a write must touch — the quorum. Third, because writes can land on different replicas during a partition, two clients can concurrently update the same key, so you need a conflict resolution story that does not silently lose data.

The single framing to hold onto: this is a leaderless, AP-leaning store where consistency is a dial, not a fixed guarantee. Dynamo-style systems let you tune how many replicas a read and a write must reach (R and W against replication factor N), trading latency and availability against how fresh a read is. That one design choice — configurable consistency instead of a hard-coded guarantee — is what makes the whole system flexible enough to serve both "must-not-lose-a-write" and "must-stay-up-under-partition" workloads.

In this walkthrough you'll scope the store, size it, model the data, design the API, draw the read and write paths, then go deep on partitioning + replication, quorum + tunable consistency, and conflict resolution + anti-entropy.

Out of scope (say so): a rich query language, secondary indexes and range scans (a pure hash store answers point lookups only — we note the extension), multi-key transactions, and the durable-vs-best-effort distinction with a cache. The in-memory best-effort cache (Redis-style) is the sibling problem — there a lost value is refilled from a database; here there is no database behind us, so durability and conflict resolution are non-negotiable.

Functional requirements

• get(key). Return the value (or values, if there are unreconciled conflicting versions) stored for a key, or a not-found signal.
• put(key, value). Store or overwrite the value for a key durably, so it survives node failures.
• delete(key). Remove a key. (As we'll see, a delete is really a special write — a tombstone — because a plain removal can be resurrected by a stale replica.)
• Partitioned + replicated. Keys spread across many nodes; each key lives on N nodes so it survives failures.
• Horizontally scalable. Adding nodes increases capacity and throughput, moving only a small fraction of keys.

Out of scope (state it): range queries over keys, secondary indexes, joins, and multi-key ACID transactions. The value is an opaque blob; the store does not interpret it. This is a point-lookup key-value store, not a relational database or a document store with a query planner.

Non-functional requirements

• Durable. An acknowledged write must not be lost. There is no source of truth behind this store — it is the source of truth — so replication and conflict handling must never silently drop an acknowledged value.
• Highly available. The store should keep serving reads and writes even when some nodes are down or the network is partitioned. This is the classic CAP AP lean — availability first, with consistency tunable up when you need it. That dial governs the non-partitioned freshness/latency curve; during an actual partition a strict quorum still sacrifices availability on the minority side — CAP forces the consistency-vs-availability choice per request.
• Horizontally scalable. Grow linearly by adding commodity nodes; no single coordinator or master that bottlenecks the whole cluster.
• Low latency at scale. Single-digit-millisecond point reads and writes, achieved by touching only a quorum of replicas, not all of them. (The write path appends to a commit log, so single-digit-ms writes assume SSDs plus periodic/group commit; fsync-per-write is the durability-vs-latency knob — Cassandra's periodic vs batch commitlog.)
• Tunable consistency. The application picks its point on the consistency/latency curve per operation by setting R and W. Strong-ish reads when needed, fast eventually-consistent reads when not.
• No single point of failure. Every node is symmetric (leaderless): any node can coordinate any request. Membership is tracked by gossip, not a master.

Estimations

State assumptions; the goal is to justify node count, replication overhead, and why quorum (not all-replica) reads are necessary — not to be exact.

Rounded assumptions

• Keys stored: ~10 billion keys.
• Value size: ~1 KB average (a serialized object; large blobs go to an object store, not here).
• Logical data: 10B × 1 KB ≈ ~10 TB before replication.
• Replication factor N = 3 (the standard default). Physical data ≈ 10 TB × 3 = ~30 TB.
• Storage per node: ~2 TB usable for data. Node count ≈ 30 TB ÷ 2 TB ≈ ~15 nodes minimum; round up well past that for headroom, hot partitions, and multi-rack placement.
• Throughput: ~1M reads/sec and ~200K writes/sec across the cluster (read-heavy but write-significant, unlike a cache).
• Quorum sizing: with N = 3, a majority quorum is 2 — a read or write touches 2 of 3 replicas, tolerating 1 node down while still overlapping.

How the numbers affect the design

Core entities / data model

Four things exist: the item (the stored key → value plus version metadata), the ring (which nodes own which keys), the replica set / preference list (the N nodes that hold a key), and the membership view (which nodes are alive).

The item. Each stored value carries not just bytes but a version stamp — a vector clock or a timestamp — that lets the store decide, when two replicas disagree, which value is newer or whether they conflict. A delete writes a tombstone: a special versioned marker meaning "this key is deleted as of this version," kept around until the repair/anti-entropy cycle has propagated it to every replica that still holds the old value (deep dive 3).

The ring. The routing table mapping hash positions to nodes. It changes only on membership events (a node joins, leaves, or dies) and is spread to every node by gossip, so any node can route any key locally.

The preference list. For a given key, the N distinct physical nodes that store it — the first N you hit walking the ring clockwise from hash(key), skipping extra virtual nodes of a physical node already chosen. It is computed, not stored, so any node can derive it from the ring (deep dive 1).

API design

Three operations carry the store. They are typically exposed over a binary/RPC protocol (Cassandra's CQL, Dynamo's HTTP API), but the logical contract is what matters. Each operation names the key, and optionally the consistency level (R or W).

Get

get(key, R) → { value(s), version(s) }  or  NOT_FOUND

The coordinator sends the read to all N replicas but only waits for R of them before returning the value (the later responders feed read repair rather than blocking the client). If the R replicas disagree on the version, it either reconciles (last-write-wins) or returns all conflicting versions (siblings) for the client to merge (deep dive 3). It also triggers read repair to fix stale replicas it noticed.

Put

put(key, value, context, W) → OK  (once W replicas ack)

The coordinator writes to the N replicas and returns OK once W of them acknowledge. The context carries the version (vector clock) the client last read, so the store can stamp the new write as a descendant of that version — this is how causal ordering is preserved. A blind put with no context is treated as a fresh, potentially-conflicting write.

Delete

delete(key, context, W) → OK

Writes a tombstone versioned like any other write, acknowledged by W replicas. The key reads as NOT_FOUND once the tombstone propagates; the tombstone itself is reaped by background compaction only after it has outlived the repair/anti-entropy cycle that propagates it to every replica — not merely a fixed "worst-case downtime" window, since a replica can stay down longer than any fixed grace period.

How the coordinator finds the replicas

replicas = ring.preferenceList(hash(key), N)   // N distinct physical nodes
coordinator = replicas[0]   // or any node, which forwards

Any node can be the coordinator for a request: it hashes the key, derives the preference list locally from its gossiped ring, and fans the request out to those N replicas. There is no master to consult — routing is a local computation (deep dive 1).

High-level architecture

Two flows matter: the write path (put → coordinator → N replicas, ack after W) and the read path (get → coordinator → N replicas, respond after R, reconcile). We'll walk each, then combine them into the leaderless cluster.

1. Write path (put, ack after W)

flowchart LR
    Client["Client"] -->|"1. put(key, value)"| Coord["Coordinator node<br/>(any node)"]
    Coord -->|"2. hash(key) to ring<br/>find N replicas"| Ring["Ring / preference list"]
    Coord -->|"3. write to all N"| R1["Replica 1"]
    Coord -->|"3. write to all N"| R2["Replica 2"]
    Coord -->|"3. write to all N"| R3["Replica 3"]
    R1 -->|"4. ack"| Coord
    R2 -->|"4. ack"| Coord
    Coord -->|"5. OK after W acks"| Client

1. The client sends put(key, value) to any node, which becomes the coordinator.
2. The coordinator hashes the key and derives the N-node preference list from its local ring.
3. It sends the write (value + new version stamp) to all N replicas.
4. Replicas apply the write locally (append to a commit log + memtable) and ack.
5. Once W replicas ack, the coordinator returns OK. The remaining replicas converge asynchronously. If a replica is down, the coordinator can stash a hinted handoff for it (deep dive 2).

2. Read path (get, respond after R, reconcile)

flowchart LR
    Client["Client"] -->|"1. get(key)"| Coord["Coordinator node"]
    Coord -->|"2. request from N replicas"| R1["Replica 1<br/>version v3"]
    Coord -->|"2. request from N replicas"| R2["Replica 2<br/>version v3"]
    Coord -->|"2. request from N replicas"| R3["Replica 3<br/>version v2 (stale)"]
    R1 -->|"3. respond"| Coord
    R2 -->|"3. respond"| Coord
    Coord -->|"4. reconcile after R,<br/>return newest"| Client
    Coord -.->|"5. read repair<br/>push v3 to stale R3"| R3

1. The client sends get(key) to a coordinator.
2. The coordinator requests the key from all N replicas (but only waits for R).
3. It collects responses and their versions.
4. After R responses, it reconciles: if one version descends from the others, return it; if versions conflict, return siblings (or apply last-write-wins). The newest version among the R responders equals the globally newest write only because R + W > N forces the read set and write set to overlap — without that overlap the answer can be stale (deep dive 2).
5. If it noticed a stale replica (R3 returned an old version), it pushes the newer value back — read repair — so replicas converge on the read path (deep dive 3).

3. Combined architecture (leaderless cluster)

Putting it together: a ring of symmetric nodes, each owning many virtual-node slices; keys replicated to N nodes; membership tracked by gossip; every node able to coordinate; background anti-entropy keeping replicas in sync.

flowchart TB
    subgraph Clients["Clients (smart or thin)"]
        C1["Client / driver"]
    end
    subgraph Cluster["Leaderless cluster (ring of symmetric nodes)"]
        direction LR
        N1["Node A<br/>vnodes + data"]
        N2["Node B<br/>vnodes + data"]
        N3["Node C<br/>vnodes + data"]
        N4["Node D<br/>vnodes + data"]
    end
    C1 -->|"get/put to any node"| N1
    N1 <-->|"gossip:<br/>membership + ring"| N2
    N2 <-->|"gossip"| N3
    N3 <-->|"gossip"| N4
    N4 <-->|"gossip"| N1
    N1 -.->|"anti-entropy<br/>(Merkle sync)"| N3
    N2 -.->|"replicate writes"| N4

1. Every node is symmetric — no master. Any node can coordinate any request; clients (or smart drivers) can send to any node.
2. Keys are partitioned by consistent hashing with virtual nodes and replicated to N nodes on the ring (deep dive 1).
3. Ring and membership state is propagated by gossip — nodes periodically exchange "who's alive and where they sit" with a few random peers, so the view converges in seconds and every node routes locally with no central directory.
4. Writes ack after W, reads after R; the operator tunes these against N (deep dive 2).
5. Anti-entropy (Merkle-tree comparison) runs in the background to repair replicas that drifted while a node was down — the safety net beneath read repair and hinted handoff (deep dive 3).

Deep dives

1. Partitioning + replication (consistent hashing, virtual nodes, and the preference list)

The data is ~10 TB replicated 3× — far more than one node — so the keyspace must be split across many nodes, and the split has to survive nodes being added, removed, and dying. This is the partitioning problem, and the replication problem rides on top of it.

key "user:42"  hash=104 → 104 % 4 = 0   → 104 % 5 = 4   (MOVED)
key "user:51"  hash=105 → 105 % 4 = 1   → 105 % 5 = 0   (MOVED)
key "order:9"  hash=110 → 110 % 4 = 2   → 110 % 5 = 0   (MOVED)

flowchart LR
    K["hash(key) = pos P"] -->|"walk clockwise"| N1["Node A<br/>(1st distinct)"]
    N1 --> N2["Node B<br/>(2nd distinct)"]
    N2 --> N3["Node C<br/>(3rd distinct)"]
    N3 --> PL["Preference list<br/>[A, B, C] = N replicas"]

2. Quorum & tunable consistency (R, W, N and the CAP dial)

Replication puts N copies of each key on the ring. Now the sharp question: on a write, how many replicas must ack before we call it done? On a read, how many must we consult before we trust the answer? These two numbers — W and R, against replication factor N — are the consistency dial.

The overlap rule. If a write is acknowledged by W replicas and a read consults R replicas, then when R + W > N the two sets are guaranteed to overlap on at least one replica — so a read is guaranteed to see the latest acknowledged write (in the absence of concurrent writes, and given version reconciliation). This is the quorum condition, and it's the single most important formula in the design.

flowchart LR
    subgraph N["N = 3 replicas"]
        A["Replica A"]
        B["Replica B"]
        C["Replica C"]
    end
    W["Write W=2<br/>acks from A, B"] --> A
    W --> B
    R["Read R=2<br/>reads B, C"] --> B
    R --> C
    B -->|"overlap on B<br/>R + W = 4 > 3"| OK["Read sees the write"]

Tuning the dial (with N = 3):

flowchart LR
    W["put(key), W=2"] --> A["Replica A ack"]
    W --> B["Replica B ack"]
    W -.->|"C is DOWN"| Cdown["Replica C (down)"]
    W -->|"sloppy: stand-in"| D["Node D<br/>holds hint for C"]
    D -.->|"C recovers →<br/>hand off + delete"| C["Replica C"]

3. Conflict resolution & anti-entropy (LWW vs vector clocks, read repair, Merkle trees)

Leaderless replication plus availability under partition means two clients can write the same key to different replicas concurrently, and both writes are accepted. Later the replicas must be reconciled. The question is how — and how replicas that silently drifted apart (a node was down for an hour) get resynced. Get this wrong and you either lose acknowledged writes or let replicas drift forever.

Detecting concurrency. How does a replica tell "version X is a newer edit of version Y" from "X and Y were written concurrently and conflict"? Two schemes:

Start:        value v0, clock {A:1}
Client 1 reads v0, writes via A → v1, clock {A:2}
Client 2 reads v0, writes via B → v2, clock {A:1, B:1}

Compare v1 {A:2} vs v2 {A:1, B:1}:
  v1 has A:2 > A:1;  v2 has B:1, v1 has no B.
  Neither dominates → CONCURRENT → return both as siblings.

Read repair — fixing drift on the read path. When a coordinator gathers R replica responses and sees one is stale, it pushes the newer version back to the laggard before (or just after) returning to the client. Reads that touch a key opportunistically heal its replicas. But read repair only fixes keys that are actually read — cold keys that drifted while a node was down would stay stale forever.

Anti-entropy with Merkle trees — the background sweep. To reconcile replicas without reading every key, each node builds a Merkle tree over its key ranges — leaves hash small key ranges, parents hash their children. Two replicas that own the same range compare root hashes: if equal, the ranges are identical and nothing transfers. If different, they descend only into the subtrees whose hashes differ, pinpointing the exact divergent key ranges while exchanging tiny hashes instead of terabytes of data.

flowchart TB
    RA["Replica A root hash"] -->|"roots differ →<br/>descend"| L1{"Left subtree<br/>hashes equal?"}
    RB["Replica B root hash"] --> L1
    L1 -->|"equal"| Skip["Skip — identical,<br/>no transfer"]
    L1 -->|"differ"| L2["Descend to leaves,<br/>find divergent range"]
    L2 --> Sync["Sync only those keys"]

Anti-entropy is the safety net beneath everything: hinted handoff can be lost, read repair only covers read keys, but a periodic Merkle sweep guarantees every replica converges eventually. Without it, replicas drift permanently and the store silently diverges.

Tradeoffs & bottlenecks

• Strong vs eventual consistency (R/W/N tuning). R=W=2, N=3 gives quorum overlap and one-node fault tolerance at the cost of touching two replicas per op. R=W=1 is fastest and most available but eventually consistent — a read can miss a recent write. The dial is per-operation; there is no free lunch, only a chosen point on the latency/consistency/availability surface.
• LWW vs vector clocks. Last-write-wins is simple and returns no siblings but silently loses concurrent writes and is vulnerable to clock skew. Vector clocks never lose a write but push conflict resolution onto the client and grow metadata. Pick LWW for loss-tolerant/single-writer keys, vector clocks where a lost write is unacceptable.
• Leaderless (Dynamo) vs single-leader replication. Leaderless keeps every node symmetric — no failover, high availability, but concurrent writes create conflicts you must resolve. Single-leader (one primary per partition) gives a clean linear write order and no conflicts, but the leader is a bottleneck and its failure needs a failover/election. Dynamo-style stores choose leaderless for availability; systems needing strong per-key ordering often choose a leader.
• Sloppy quorum availability vs strict guarantees. Sloppy quorum + hinted handoff keeps writes flowing under failure but temporarily weakens the overlap guarantee. Strict quorum preserves the guarantee but returns errors when a key's replicas are down.
• Tombstones and delete correctness. Deletes must be tombstones to prevent resurrection, but tombstones consume space and must be reaped after a grace period — reap too early and deleted data comes back; too late and you accumulate garbage.
• Hot partitions. A single hot key or a poorly chosen partition key funnels load onto N nodes regardless of cluster size; consistent hashing balances storage, not access skew.

Extensions if asked

Add only the extension that changes the design discussion:

• Secondary indexes. The base store answers point lookups by primary key only. A secondary index (query by a non-key attribute) requires either a separate index table (maintained on write, eventually consistent — DynamoDB's global secondary indexes) or a scatter-gather over all partitions (Cassandra's local secondary indexes) — each with its own consistency/latency cost.
• Range queries (vs pure hash). Consistent hashing scatters adjacent keys across the ring, so range scans are impossible. An order-preserving partitioner (or a clustering key within a partition, as Cassandra uses) keeps sorted order for range scans, at the cost of hot spots for sequential keys.
• LSM-tree storage engine. How each node stores data on disk: buffer writes in an in-memory memtable + commit log, flush to immutable sorted SSTables, and compact them in the background. LSM trees give fast sequential writes (ideal for write-heavy KV stores) and are where tombstones actually get reaped.
• Cross-region replication. Replicate across regions/data centers for low-latency global reads and disaster recovery, with per-datacenter replication factors and quorum settings (Cassandra's NetworkTopologyStrategy), trading global write latency for locality.

What interviewers look for & common mistakes

What interviewers usually reward:

• Framing the store as the source of truth (not a cache), so durability and conflict resolution are non-negotiable — and contrasting it explicitly with the best-effort cache.
• Partitioning with consistent hashing + virtual nodes, explaining why hash % N triggers a mass migration and how the preference list yields N distinct-physical (rack-aware) replicas.
• Deriving R, W, N and the R + W > N overlap rule, and tuning it per workload — with the honesty that quorum ≠ unconditional strong consistency.
• Sloppy quorum + hinted handoff for availability under failure, and knowing it weakens the strict guarantee temporarily.
• A real conflict-resolution story — vector clocks to detect concurrency and return siblings, vs LWW's silent data loss — plus read repair and Merkle-tree anti-entropy so replicas actually converge.
• Naming the tombstone for deletes and why a plain removal gets resurrected.

Before you finish, do a quick mistake check:

• Did you treat the store as durable / the source of truth, not a rebuildable cache?
• Did you partition with consistent hashing + virtual nodes, not hash(key) % N, and build an N-distinct-physical preference list?
• Did you state the R + W > N quorum rule and tune R/W/N per workload?
• Did you avoid the overclaim that quorum unconditionally equals strong consistency (concurrent writes, sloppy quorum)?
• Did you give a concrete conflict-resolution scheme — vector clocks vs LWW — and name LWW's silent data loss?
• Did you include read repair and Merkle-tree anti-entropy, so even un-read drifted replicas converge?
• Did you make deletes tombstones to prevent resurrection?
• Did you keep the store available under failure with sloppy quorum + hinted handoff, and name its cost?