Requirements

Functional Requirements

1. GET (key) -> value with optional consistency level (one, quorum, all).
2. PUT (key, value) that durably stores or updates the value.
3. DELETE (key) that removes the value and propagates the deletion.
4. Tunable consistency per request: callers choose between fast/eventually consistent and slower/strongly consistent.
5. Versioning: support concurrent writes with conflict detection (vector clocks); the application or store decides the resolution policy.
6. Auto scale: add nodes to the cluster without downtime; existing data redistributes minimally.
7. Multi-data-center: optionally replicate across data centers with configurable replication factor per DC.

Out of Scope (state explicitly)

• Range queries / secondary indexes (this is a pure KV store; range support is a separate feature).
• Strong cross-key transactions (we offer single-key atomicity only).
• Schema management (values are opaque blobs to the store).
• Client-side caching (client libraries optional).

Non-Functional Requirements

1. Highly available: 99.99%, even during node failures and network partitions. Lean AP per CAP.
2. Tunable consistency: callers pay the cost of consistency only when they want it.
3. Low latency: p99 < 10 ms for GET at consistency=ONE; p99 < 30 ms at consistency=QUORUM.
4. Linear scale: doubling nodes ~doubles throughput; throughput grows with cluster size, not node size.
5. Durable: no acknowledged write is lost under single-node failure; quorum writes survive a minority partition.
6. Operationally simple: no manual rebalancing, no leader elections per partition, no specialized hardware.

Back-of-the-Envelope Estimation

Capacity

---------- Storage and traffic ----------
Keys:                          1B
Average value size:            ~1 KB
Replication factor (N):        3
Logical data:                  ~1 TB
Replicated data:               ~3 TB
Daily ops (R + W):             ~10B
Average ops/sec:               ~120K
Peak ops/sec:                  ~500K
Cluster nodes (each ~16 GB RAM, 1 TB SSD): ~50 to start

With 50 nodes each holding ~60 GB and ~10K ops/sec, headroom for 5x growth before adding nodes.

Network and Replication Overhead

---------- Replication overhead ----------
Every PUT writes to N=3 nodes; a PUT at QUORUM (W=2) waits for 2 acks but still writes 3.
Replication multiplier:        3x writes; 1x reads
Inter-node bandwidth at 500K writes/sec * 1 KB * 3 replicas: ~1.5 GB/sec across the cluster
Per-node replication traffic:   ~30 MB/sec at 50 nodes (manageable)

Hot Key Risk

---------- Hot key estimation ----------
With virtual nodes (256 vnodes per physical node), a single popular key still lives on one physical node.
If one key receives 10K ops/sec, that's 10% of one node's budget; we live with it or precompute / cache hot reads.
For truly hot keys we replicate beyond N (e.g., N=5) and let R=1 reads fan out.

High-Level Design

---------- Architecture overview ----------
   +-----------+
   |  Client   |
   +-----------+
        |
        v (any node accepts requests; called the coordinator)
   +------------------+
   |  Coordinator     | (one of the cluster nodes)
   +------------------+
        |
        v (forward to N replicas via consistent hash)
   +-------- ring of nodes (consistent hashing) --------+
   | n1   n2   n3   n4   n5 ... n50                     |
   |   N replicas = next clockwise nodes                 |
   +-----------------------------------------------------+
        |
        v (per-node)
   +------------------+
   | Storage engine   | (LSM tree on SSD)
   +------------------+

Every node is symmetric: any node can be the coordinator for any request. There is no master. Membership is gossiped peer-to-peer.

Public API

// Client SDK style
put(key, value, consistency=QUORUM, version=null)
get(key, consistency=ONE)
delete(key, consistency=QUORUM)

// Wire-level (gRPC)
rpc Put(PutRequest) returns (PutResponse);
rpc Get(GetRequest) returns (GetResponse);
rpc Delete(DeleteRequest) returns (DeleteResponse);

message PutRequest {
    bytes key = 1;
    bytes value = 2;
    Consistency consistency = 3;
    repeated VectorClockEntry vclock = 4;   // for concurrent-write detection
}

enum Consistency { ONE = 0; QUORUM = 1; ALL = 2; }

Request Flow

1. Client sends PUT to any node (the coordinator). Smart clients pick a node that owns the key's range; dumb clients hash anywhere.
2. Coordinator computes the key's preference list (N nodes responsible for the key).
3. Coordinator sends the write to all N concurrently and waits for W acks before responding.
4. Each replica appends to its commit log, updates the LSM memtable, and acks.
5. If a replica is unreachable, the coordinator stores a hinted handoff for it on a substitute node.

Detailed Design

The two interesting components are consistent hashing with virtual nodes and the N/W/R quorum with vector clocks for conflict resolution.

Consistent Hashing with Virtual Nodes

A naive sharding scheme (key -> hash(key) mod N) breaks badly when N changes; nearly every key remaps. Consistent hashing maps both keys and nodes onto a ring (e.g., a 64-bit hash space). A key is owned by the first node clockwise from it.

---------- Consistent hash ring ----------
Hash space: 0 .. 2^64-1
Nodes:      n1 at hash(n1)=2^60, n2 at 2^62, n3 at 2^63 ...
Key 'foo': hash('foo') = 2^61.5; first clockwise node is n2; n2 owns 'foo'.

When we add n4 at 2^61, only keys in the range (2^60, 2^61] move to n4 (small fraction).

Why Virtual Nodes

With one position per physical node, the ring is uneven (some nodes own much larger ranges than others) and adding a node moves all of one neighbor's data. Virtual nodes (vnodes) split each physical node into many positions on the ring (typically 128 or 256).

Effects:

• The ring becomes statistically uniform; each physical node owns roughly the same total range.
• Adding a new physical node steals small slivers from many nodes, not one big chunk from one neighbor.
• Heterogeneous capacity is easy: a beefier node gets more vnodes, owning a larger slice.

Preference List

For key K, the preference list is the next N distinct physical nodes clockwise on the ring. This is the replica set for that key.

---------- Preference list example ----------
Key 'foo' with N=3:
  position on ring -> next vnode owned by n2 (primary)
  -> next vnode owned by n5 (replica)
  -> next vnode owned by n11 (replica)
Preference list = [n2, n5, n11]

If n5 is down, we skip to n12 as the substitute (sloppy quorum); we'll hand off the data to n5 when it returns.

N, W, R Quorums

N = number of replicas in the preference list. W = number of acks required before a write is considered successful. R = number of replies required before a read returns.

If W + R > N, every read is guaranteed to see at least one replica that has the latest write (strong consistency, with caveats around vector clocks).

---------- Common configurations ----------
N=3, W=1, R=1  -> fast, eventually consistent (Dynamo-style 'always writeable')
N=3, W=2, R=2  -> quorum reads/writes, strongly consistent (W+R > N)
N=3, W=3, R=1  -> writes always see all replicas, fast reads, slow writes
N=3, W=1, R=3  -> fast writes, slow reads, strong

The default for most workloads is W=R=2 (quorum on both sides). Latency is bounded by the slower of two replicas (often well below the third's tail latency).

Vector Clocks for Conflict Detection

When two clients concurrently update the same key without coordination, both writes can succeed. Without vector clocks the store cannot tell which is newer (timestamps are unreliable across hosts). Vector clocks tag every write with a per-coordinator counter.

---------- Vector clock example ----------
Write A by client1 -> vclock = {n2: 1}
Write B by client2 (concurrent, no read of A) -> vclock = {n5: 1}
Reader sees both versions: {n2: 1} and {n5: 1}. Neither dominates the other.
The reader (client) is told 'two versions exist; pick one or merge them'.

If one clock dominates (every entry >= the other's), that version is newer. If neither dominates, they are concurrent.

Conflict Resolution Policies

1. Application merges: client receives all sibling versions and decides (e.g., shopping cart: union the items).
2. Last-write-wins (LWW): store keeps the version with the highest timestamp; loses some writes but is simple. Cassandra defaults to this.
3. CRDT: values are designed to merge automatically (counters, sets); great when applicable.

Storage Engine: LSM Tree

Each node stores its data in an LSM (log-structured merge) tree:

---------- LSM tree structure ----------
Write path: append to commit log -> insert into memtable
When memtable fills: flush to immutable SSTable on disk
Background compaction merges SSTables, drops tombstones, rewrites in sorted order
Read path: check memtable, then bloom filter per SSTable, then SSTables in age order

LSM trees give excellent write throughput (sequential writes only) at the cost of read amplification (multiple SSTables per query). Bloom filters per SSTable make negative reads cheap. Compaction strategy (size-tiered vs leveled) trades write amplification for read amplification.

Hinted Handoff

When a coordinator can't reach replica R for a write, it writes to the next clockwise node X with a 'hint': 'this is for R, deliver when R returns.' X stores the hint. When R rejoins the cluster (gossip notices), X replays the hints to R.

---------- Hinted handoff flow ----------
1. Coordinator tries to write to n5; n5 unreachable.
2. Coordinator writes the data to n12 with hint metadata {target: n5, expires: now+3h}.
3. Gossip detects n5 has rejoined.
4. n12 replays hints to n5; deletes them once acked.

Hinted handoff handles brief outages (seconds to minutes). For longer outages, anti-entropy (next section) catches the residual divergence.

Anti-Entropy with Merkle Trees

Replicas can drift if hints expire or are lost. Periodic anti-entropy sessions compare two replicas' contents and reconcile differences.

Naive comparison (send every key) is O(data size). Merkle trees compare in O(log N) by hashing tree levels:

---------- Merkle tree comparison ----------
Each replica builds a Merkle tree of its key range (leaves = key hashes, nodes = hash of children).
Replicas exchange tree roots; if equal, they agree on everything.
If different, recurse into the differing subtrees until you find the divergent keys.
Only the divergent keys are exchanged.

This is what Riak and Cassandra call 'anti-entropy repair'. Run as a background job (typically once per week).

Gossip for Membership

Nodes don't need a master to know cluster state. Gossip propagates membership changes peer-to-peer:

---------- Gossip ----------
Every 1 s, each node picks K=3 random peers and exchanges:
  - my view of the membership table (alive / suspect / down)
  - my heartbeat counter
Within O(log N) seconds, the whole cluster converges on a consistent view.

Nodes detected as down (via heartbeat timeout) are removed from preference lists. When they rejoin, gossip propagates that and they are added back; vnodes redistribute slightly.

Coordinator Logic (Pseudocode)

async function put(key, value, consistency = 'QUORUM') {
    const N = 3;
    const W = consistency === 'ONE' ? 1 : consistency === 'ALL' ? N : 2;
    const replicas = preferenceList(key, N);     // ring lookup
    const vclock = mergeAndIncrement(key, this.nodeId);
    let acks = 0;
    const errors = [];
    await Promise.all(replicas.map(async (node) => {
        try {
            await rpcPut(node, key, value, vclock);
            acks++;
        } catch (e) {
            errors.push({ node, e });
            const sub = nextHealthyAfter(node);
            await rpcPutWithHint(sub, key, value, vclock, { hintFor: node });
        }
    }));
    if (acks >= W) return { ok: true, vclock };
    throw new Error(`Insufficient acks: ${acks}/${W}`);
}

Data Model

Per-Node On-Disk Layout

---------- Per node ----------
/data/
  commitlog/                  (sequential writes, fsync per batch)
  memtable                    (in RAM, ~256 MB before flush)
  sstables/
    L0-001.sst                (most recent flush)
    L1-001.sst
    ...
  bloom/                      (one bloom filter per SSTable)
  hints/                      (pending hinted handoffs to other nodes)
  merkle/                     (cached Merkle tree segments)

In-Memory State

• Membership table (gossiped, O(N) entries).
• Vnode -> physical node map (computed from membership).
• Recent vector clocks per active key (small cache).

Per-Key State on Disk

For a key 'foo':

---------- Stored value ----------
key:        'foo'
vclock:     {n2: 4, n5: 3, n11: 2}
value:      <bytes>
tombstone:  false
last_write_ts: 1714142400123

Tombstones (delete markers) are kept until anti-entropy + compaction eventually drops them; that's why 'delete' is heavier than in a SQL store.

Scaling and Bottlenecks

Adding Nodes

When a node joins, gossip propagates its identity; the ring is recomputed; small slivers of vnodes from existing nodes migrate to the new node. Migration is rate-limited (the cluster keeps serving traffic during it). Time to add a node: minutes for membership convergence; hours for data migration depending on cluster size.

Hot Keys

A single hyper-popular key (Justin Bieber's user record on Twitter) lands on one set of N nodes. Mitigations:

• Increase N for that key (e.g., N=5 for read-hot keys). More replicas means R=1 reads can fan out wider.
• Client-side caching with short TTL.
• For truly extreme cases, replicate the value to all nodes (but lose write performance).

Multi-Region Replication

Replicate data across DCs by including replicas from each DC in the preference list. Local QUORUM (LOCAL_QUORUM) lets writes succeed locally even if the remote DC is unreachable; full-cluster QUORUM (EACH_QUORUM) requires acks from each DC.

The trade-off: LOCAL_QUORUM is fast but two DCs can diverge during a partition; EACH_QUORUM costs cross-DC RTT on every write.

Compaction Pressure

LSM compaction can monopolize disk IOPS during heavy write periods, hurting read latency. Throttle compaction to a fixed IO budget and prefer leveled compaction for read-heavy workloads.

Sloppy Quorum During Partitions

If a partition isolates a replica, hinted handoff lets writes continue (sloppy quorum). When the partition heals, hints replay and anti-entropy repairs anything missed. The trade-off: clients see eventually consistent reads during the partition (some replicas don't have the latest data).

Read Repair

On a read at QUORUM, the coordinator may receive different versions from different replicas. It returns the newest (or all siblings to the client). In the background, it pushes the newest to the lagging replicas: 'read repair'. This converges replicas without waiting for anti-entropy.

Trade-offs and Alternatives

AP vs CP

Dynamo is AP-leaning: it sacrifices strong consistency in favor of always being writeable. Spanner-style CP designs (Spanner, CockroachDB) sacrifice availability under partition for serializable transactions across keys. Pick AP for high-availability serving (sessions, configs, carts) and CP for accounting / inventory where overselling is unforgivable.

Vector Clocks vs Last-Write-Wins

Vector clocks preserve all conflicting writes for the application to merge; LWW silently drops one. Vector clocks are the right answer when conflicting writes might both contain real value (cart items merged, social connections added). LWW is fine for monotonic state (latest-known-position).

Quorum Reads vs Eventually Consistent Reads

Quorum reads (R+W > N) cost more latency but guarantee freshness; eventually consistent reads (R=1) are fast but may return stale data. Most applications can tolerate stale reads; for the few that can't (reading own writes), use R=N or read from the same coordinator that did the write.

Consistent Hashing vs Range Sharding

Consistent hashing distributes keys uniformly but range queries become expensive (must scan multiple nodes). Range sharding (by key prefix) supports range queries cheaply but suffers hot-spot issues for sequential keys (timestamps). Pick consistent hashing for pure KV; range sharding for time-series or column families.

LSM vs B-Tree Storage

LSM is write-optimized (append only, sequential writes) at the cost of read amplification. B-tree is read-optimized (single seek per read) but updates are random IO. KV stores nearly always pick LSM because the workload is high write, scan-heavy reads served from cache.

Centralized Coordinator vs Symmetric Nodes

Dynamo nodes are symmetric (any node can coordinate any request). A centralized coordinator is simpler to reason about but introduces a SPOF and a hotspot. Symmetric nodes are harder to implement (every node needs full ring state) but scale linearly and have no SPOF.

Why Not Just Use Postgres with Replication?

Postgres with streaming replication is great for SQL workloads up to ~50K writes/sec on a single primary. Beyond that, you must shard manually, lose cross-shard transactions, and inherit operational complexity that the Dynamo model handles natively. For pure KV at high scale, Dynamo-style is meaningfully simpler than DIY Postgres sharding.

Tunable Consistency vs One-Size-Fits-All

Dynamo's per-request consistency knob is the right design: different operations have different needs. A static 'always strong' system makes most operations unnecessarily slow; a static 'always eventual' system frustrates the few operations that need freshness. Per-request choice is more API surface but more flexible.