Design a Key-Value Store (like DynamoDB) - System Design

A key-value store looks like a solved problem for about thirty seconds. “Put a value under a key, get it back later, it is a hash map.” Then the interviewer says: it has to survive a machine catching fire, it has to stay writable when the network splits the cluster in half, it has to hold petabytes across thousands of nodes, and two clients are going to write the same key at the same instant from two different data centers. Now every easy answer is wrong. Which node owns a key when nodes are constantly joining and leaving? If you replicate a write to three nodes and one is down, do you fail the write or accept it? When two writes race, which one wins, and how do you even know they raced? And can the caller ask for “fast and maybe stale” on one request and “slow and correct” on the next?

This is the Dynamo problem. Unlike a cache, this store is the source of truth - losing data is not “a slower backend next time,” it is data loss. That single constraint changes everything: durability is mandatory, and you cannot wave away conflicts because there is no database behind you to reconcile against. The design comes down to four decisions: how you place and replicate keys (consistent hashing), how many replicas must answer a read or write (quorum), how you resolve concurrent writes (vector clocks vs last-write-wins), and how you let the caller dial the consistency-vs-latency trade per request (tunable consistency).

Let me do it properly.

Functional Requirements (FR)

In scope:

• get(key) and put(key, value). The two core operations. Keys are opaque strings, values are opaque blobs up to a few MB. No structure inside the value the store cares about.
• delete(key). Remove a key. As we will see, delete in an eventually-consistent replicated store is subtle (a delete is really a special write, a tombstone).
• Durability. An acknowledged write must survive node failures. This is the store, not a cache in front of one. If we say “OK,” the data is safe.
• High availability. The store stays readable and writable through single-node failures, rack failures, and network partitions. “Always writable” is an explicit goal (the shopping-cart motivation from Dynamo).
• Horizontal scale. Petabyte-scale data spread across thousands of commodity nodes. Adding nodes adds capacity with minimal data movement.
• Tunable consistency. The caller chooses per request how many replicas must respond, trading latency for freshness.

Explicitly out of scope (say this out loud so you control the scope):

• Rich queries. No SQL, no joins, no secondary-index scans as a core feature, no range queries over arbitrary values. Access is by primary key. (DynamoDB later added secondary indexes and range keys; we treat those as extensions, not the core.)
• Strong ACID transactions across many keys. We provide single-key atomic writes and conditional writes, not multi-key serializable transactions across shards. That is a distributed-transaction problem layered on top.
• A relational schema. Values are schemaless blobs. The application owns their structure.
• Strict linearizable consistency as the default. We default to eventual consistency and make strong consistency an opt-in per request, because the whole point of this design is availability under partition.

The one decision that drives everything: this store chooses availability and partition tolerance, and makes consistency tunable. In CAP terms, when the network partitions we keep serving reads and writes rather than refusing them, and we clean up the resulting conflicts afterward. That choice - “always writable, reconcile later” - is the spine of the entire design and the reason vector clocks and quorums exist at all.

Non-Functional Requirements (NFR)

• Scale: on the order of 10 PB of data, 100B+ keys, thousands of nodes across multiple data centers. Both read-heavy and write-heavy workloads must be supported; assume a roughly balanced-to-write-leaning mix (this is a store, not a read cache).
• Latency: single-digit-millisecond p99 for get and put at the coordinator, because the target is online request paths (a shopping cart, a session store, a user profile). We hit this by keeping the read/write path to a small quorum of replicas and never doing cross-cluster coordination on the hot path.
• Availability: 99.99%+ for writes especially. The design goal is “writes never rejected” as far as physically possible - a partition should degrade consistency, not availability.
• Consistency: eventual by default, tunable up to quorum-strong per request. Concurrent conflicting writes are detected and either auto-resolved (last-write-wins) or surfaced to the application (vector clocks / siblings) depending on configuration.
• Durability: high. Every write is replicated to N nodes across failure domains (different racks, ideally different data centers) and persisted to disk. We tolerate the loss of any single node, and with cross-DC replication, a whole data center.

Back-of-the-Envelope Estimation (BoE)

Let me ground the design in numbers.

Data volume and node count:

Total logical data:        10 PB
Replication factor N:      3   (every key stored on 3 nodes)
Raw stored data:           10 PB * 3 = 30 PB on disk

Usable disk per node:      ~10 TB (leave headroom for compaction, overhead)
Nodes just to hold data:
  30 PB / 10 TB = 3,000 nodes
Round up for headroom, hot ranges, and growth -> ~4,000 nodes.

Object math:

Total keys:        100,000,000,000  (100B)
Avg value size:    10 PB / 100B = ~100 KB average (blended; real mix is bimodal)
Per-key metadata:  key + version/vector-clock + timestamps + checksum
  ≈ 100-200 bytes overhead per key on disk.
100B keys * ~150 bytes = ~15 TB of pure metadata across the cluster.

Throughput:

Assume peak:        2,000,000 ops/sec  (mixed)
Read:write ratio:   ~1:1 (store, not cache) -> 1M reads/sec, 1M writes/sec

Each write with N=3 fans out to 3 replica writes:
  1M writes/sec * 3 = 3M replica writes/sec of internal work.
Each quorum read (R=2) contacts 2 replicas:
  1M reads/sec * 2 = 2M replica reads/sec.

Per node (4,000 nodes, evenly spread):
  Internal write ops: 3M / 4000 = 750/sec/node
  Internal read ops:  2M / 4000 = 500/sec/node

Around 1,250 disk-touching ops/sec/node is comfortable for an LSM-tree store on SSD. So average load is a non-problem. The difficulty is not throughput - it is correctness under failure: what happens to that write when one of the three replicas is unreachable, and what happens when two writes to the same key cross in flight.

Storage growth:

Assume 1B new writes/day, avg 100 KB, replicated 3x:
  1B * 100 KB * 3 = 300 TB/day of raw writes.
Not all are new keys (overwrites), but for a growing dataset,
  say net ~30 TB/day new -> ~11 PB/year growth.
This is why node count must grow smoothly - membership change
  is a routine weekly event, not a rare one. The placement scheme
  must make adding nodes cheap.

Bandwidth:

Writes: 1M/sec * 100 KB * 3 replicas = 300 GB/sec aggregate write traffic.
Spread over 4,000 nodes = ~75 MB/sec/node. Fine on 10GbE+.
Cross-DC replication of writes is the expensive part:
  1M writes/sec * 100 KB = 100 GB/sec across the WAN if fully mirrored.
This is why cross-DC replication is async and why quorum is
  usually kept within a single DC for latency.

The numbers confirm the shape: aggregate load is easy, and the entire design budget is spent on failure handling and conflict resolution, not on raw QPS. Size the design for partitions and concurrent writes, not for ops/sec.

High-Level Design (HLD)

A Dynamo-style store is a ring of identical nodes (no leader, no primary shard - every node can coordinate any request) plus a replication scheme that copies each key to N consecutive nodes on the ring. A client talks to any node; that node becomes the coordinator for the request, routes to the replicas, gathers quorum, and answers. Membership and failure state spread by gossip. There is deliberately no central master on the request path.

                 ┌───────────────────────────────────────────┐
                 │              Client / SDK                  │
                 │  (may hash locally, or talk to any node)   │
                 └─────────────────────┬──────────────────────┘
                                       │ get/put(key)
                                       │ any node can coordinate
                 ┌─────────────────────▼──────────────────────┐
                 │   Coordinator node (the one the client hit) │
                 │   - hash key -> ring position               │
                 │   - find N preference-list nodes clockwise  │
                 │   - send to replicas, wait for W (or R)      │
                 └───┬───────────────┬───────────────┬─────────┘
                     │               │               │
              ┌──────▼─────┐  ┌──────▼─────┐  ┌──────▼─────┐
              │ Replica 1  │  │ Replica 2  │  │ Replica 3  │   (N=3, the
              │ (LSM store)│  │ (LSM store)│  │ (LSM store)│    preference
              │ vclock+val │  │ vclock+val │  │ vclock+val │    list for K)
              └────────────┘  └────────────┘  └────────────┘

        The ring (consistent hashing, virtual nodes):
        ┌────────────────────────────────────────────────────┐
        │   All nodes on one hash ring. Key K -> first vnode   │
        │   clockwise = coordinator's target; next N-1 distinct │
        │   physical nodes clockwise = the replica set.         │
        └────────────────────────────────────────────────────┘

        Membership & failure detection:
        ┌────────────────────────────────────────────────────┐
        │   Gossip protocol - every node periodically         │
        │   exchanges membership + heartbeat state with peers. │
        │   No central config master on the request path.      │
        └────────────────────────────────────────────────────┘

        Background reconciliation (keeps replicas converging):
        ┌────────────────────────────────────────────────────┐
        │   Anti-entropy (Merkle-tree sync), hinted handoff,   │
        │   read repair. Fix divergence off the hot path.      │
        └────────────────────────────────────────────────────┘

Request flow - a put (write path):

1. The client sends put(key, value, context) to any node. That node becomes the coordinator. (The context carries the vector clock the client last saw for this key - more on that below.)
2. The coordinator hashes the key to a ring position and computes the preference list: the first N distinct physical nodes walking clockwise. It picks the version metadata (bumps the vector clock or stamps a timestamp).
3. The coordinator writes locally and sends the write to the other N-1 replicas in parallel.
4. It waits for W-1 acks from replicas (plus its own = W total). Once W nodes have persisted the write, the coordinator acks the client. The remaining replicas complete asynchronously.
5. If a replica is down, the write goes to the next healthy node on the ring as a hinted handoff (a temporary holder that will deliver the data back when the real owner returns). The write still succeeds - availability preserved.

Request flow - a get (read path):

1. The client sends get(key) to any node; that node coordinates.
2. The coordinator computes the same preference list and asks the N replicas for their copy of the key.
3. It waits for R replicas to respond. If all R agree (same version), it returns the value.
4. If the R responses disagree (different vector clocks), the coordinator either returns all conflicting versions (siblings) for the application to reconcile, or applies last-write-wins, depending on config. It also triggers read repair: push the reconciled/newest version back to the stale replicas.
5. The client gets the value plus the current context (vector clock) to pass back on its next write.

The key architectural insight: there is no special node. Any node can coordinate any request, placement is pure hashing, and consistency is a per-request choice of R and W. That symmetry is what makes the system available and horizontally scalable - and it is exactly why conflict resolution has to be built in, because with no leader, two coordinators can accept conflicting writes to the same key simultaneously.

Component Deep Dive

Four hard parts: (1) placement and replication via consistent hashing, (2) quorum reads and writes with tunable R/W/N, (3) resolving concurrent writes with vector clocks vs last-write-wins, and (4) keeping replicas convergent under failure (hinted handoff, read repair, anti-entropy). I will walk each from naive to scalable.

1. Placement and Replication: Modulo Hashing -> Consistent Hashing -> Virtual Nodes

Approach A: Modulo hashing (the naive instinct)

Pick the node for a key with node_index = hash(key) % N_nodes.

def node_for(key, nodes):
    return nodes[hash(key) % len(nodes)]

Simple and even. Where it breaks: adding or removing a node remaps almost every key. Change the node count from 4000 to 4001 and hash(key) % 4001 differs from hash(key) % 4000 for essentially every key. For a cache that means mass misses; for a durable store it is far worse - it means physically moving nearly the entire 30 PB dataset across the network to new homes on every membership change. At petabyte scale with weekly node additions, that is a permanent, cluster-crushing rebalance. Modulo hashing is disqualified the instant membership is not fixed.

Before: hash(key)=98765, 98765 % 4000 = 765   -> node 765
Add one node (N=4001):
After:  98765 % 4001 = 3819                    -> node 3819 (must move ~all data!)

Approach B: Consistent hashing (the fix)

Map both nodes and keys onto the same fixed circular hash space (say 0 to 2^128 - 1). A key is owned by the first node found walking clockwise from the key’s hash position.

Hash ring (0 .. 2^128-1), positions from hash(node_id) and hash(key):

        Node A (pos=10)
       /                \
  Node D (pos=95)     key K1 (pos=15) -> clockwise -> Node B
      |                     \
       \                Node B (pos=40)
        Node C (pos=70) /
              key K2 (pos=80) -> clockwise -> Node D

Finding a key’s node is a binary search over sorted node positions: O(log N). The win on membership change: adding or removing a node only affects the arc between that node and its predecessor. Add a node between A and B, and only keys in that arc move; every other key stays home. So going from 4000 to 4001 nodes moves ~1/4000 of the data, not all of it. That single property - minimal data movement - is why consistent hashing is mandatory for a store that reshards routinely.

Replication on the ring. For durability we do not store a key on one node; we store it on the N consecutive distinct physical nodes clockwise from its position - the preference list. If N=3, key K1 lands on B, then the next two distinct physical nodes after B.

Preference list for K1 (N=3):  B (owner), then next two distinct
physical nodes clockwise -> [B, C, D].
A write to K1 replicates to all three. A read consults them.

“Distinct physical” matters: you must skip vnodes that map back to a node already in the list, or you would store two of your three replicas on the same machine and a single failure loses two copies.

Approach C: Virtual nodes (the production refinement)

Plain consistent hashing has two flaws. First, uneven load: with a few thousand random points, arc sizes vary wildly, so some nodes own far more data than others. Second, lumpy failure and rebuild: when a node dies, its whole arc dumps on the single next node, and when you add a node it drains from just one neighbor - slow, uneven rebuilds.

The fix is virtual nodes: each physical node is placed on the ring many times (say 100-256 tokens) via hash(node_id + ":" + i).

Physical Node A -> A#0, A#1, ..., A#255  (256 tokens scattered on the ring)
Physical Node B -> B#0, B#1, ..., B#255
... every physical node owns 256 small arcs spread all around the ring.

Two wins that matter more for a store than a cache:

• Even data distribution. With hundreds of tokens per node, the law of large numbers evens out how much data each physical node holds - critical when each node stores terabytes and you cannot afford a lopsided node running out of disk.
• Fast, parallel rebuild. When a physical node dies, its 256 arcs are inherited by 256 different nodes, and when a new node joins it pulls a little data from many nodes at once. Rebuilding a failed node’s replicas streams from the whole cluster in parallel instead of from one overloaded neighbor. At 10 TB/node this is the difference between a rebuild taking hours and taking days.

You also get heterogeneous hardware support: give bigger boxes more tokens so they hold proportionally more data. Virtual nodes are the answer - consistent hashing for minimal movement, vnodes for even placement and parallel rebuild.

2. Quorum Reads and Writes: Tunable R, W, N

Now the store has each key on N replicas. The question every read and write must answer: how many of the N do I wait for? This is where consistency is actually dialed.

Approach A: Write to one, read from one (naive, fast, wrong)

Coordinator writes to its local replica, acks immediately, replicates the rest in the background. Reads hit any one replica. Where it breaks: a write acked by node 1 may not have reached node 2 yet; a read served by node 2 returns stale data (violates read-your-writes), and if node 1 dies before replicating, the acked write is lost - unacceptable for a durable store. W=1/R=1 gives you speed and neither consistency nor durability.

Approach B: Write to all, read from all (naive, safe, unavailable)

Require every write to reach all N replicas and every read to consult all N. Now reads always see the latest write. Where it breaks: if a single replica of the N is down or slow, every write and read to that key fails or stalls. You have coupled availability to your least-available replica - the exact opposite of the availability goal. W=N/R=N gives consistency but throws away availability, which was the whole point of this design.

Approach C: Quorum with tunable R and W (the fix)

Do not go to 1 and do not go to N - go to a quorum. Require W replicas to ack a write and R replicas to answer a read, with the overlap rule:

If  R + W > N  then any read quorum and any write quorum
share at least one replica -> a read is guaranteed to see the
most recent completed write. (Strong, quorum-consistent read.)

Typical: N=3, W=2, R=2  ->  2 + 2 > 3  -> overlap guaranteed.

The overlap is the whole trick: with R + W > N, the set of R nodes you read from must intersect the set of W nodes the last write reached, so at least one node in your read has the newest version. Pick the newest among the R responses (by vector clock / timestamp) and you have the latest committed value - while still tolerating N - W down nodes on writes and N - R down nodes on reads.

Tunable consistency falls straight out of choosing R and W per request:

DynamoDB exposes this as ConsistentRead: true/false - an eventually-consistent read (cheaper, one replica, may lag) versus a strongly-consistent read (quorum, sees the latest). Cassandra exposes the full knob per query (ONE, QUORUM, LOCAL_QUORUM, ALL). The caller decides, request by request, where to sit on the latency-vs-freshness line. That per-request control is the feature.

Sloppy quorum and hinted handoff. Strict quorum says W must be W of the N home replicas. But if some home replicas are down, strict quorum would reject the write - losing availability. Dynamo relaxes this to a sloppy quorum: if a home replica is down, the write goes to the next healthy node on the ring, which stores it with a hint (“this really belongs to node X”). W is satisfied by any W healthy nodes, so writes almost never fail. When node X recovers, the hint-holder hands the data back (hinted handoff). The trade-off: during a partition, a sloppy-quorum read and write on opposite sides may not overlap, so R+W>N no longer strictly guarantees freshness - you have chosen availability over consistency for the duration, then reconcile after. This is CAP made concrete and per-operation.

3. Concurrent Writes: Last-Write-Wins vs Vector Clocks

With no leader and sloppy quorums, two coordinators can accept writes to the same key at the same time, on different replicas, and neither knows about the other. When those replicas later sync, which value is correct? This is the hardest part of the whole design.

Approach A: Last-write-wins (LWW) with wall-clock timestamps

Stamp every write with the coordinator’s wall-clock time. On conflict, the higher timestamp wins. Dead simple, no extra metadata beyond a timestamp, and reads just pick the max.

Where it breaks: clocks across thousands of nodes are not perfectly synchronized. If node A’s clock is 50 ms ahead of node B’s, a write on B that genuinely happened later can carry a smaller timestamp and get silently discarded in favor of an older write - data loss with no error. Worse, LWW cannot even represent a legitimate concurrent update: if two users add different items to a shopping cart simultaneously, LWW keeps one and throws the other away. For values where a lost update is acceptable (last-known sensor reading, a user’s latest profile-photo URL) LWW is fine and is the pragmatic default DynamoDB uses. For values where every concurrent update must be preserved, LWW is wrong.

User X adds "milk" to cart at ts=1000 on node A
User Y adds "eggs" to cart at ts=1000 on node B (concurrent)
LWW keeps whichever timestamp is higher -> the other item is LOST.
The cart should have BOTH. LWW cannot express "these are concurrent."

Approach B: Vector clocks (capture causality, detect true concurrency)

A vector clock is a map of node_id -> counter. Each time a node coordinates a write to a key, it increments its own entry. The clock travels with the value. Comparing two vector clocks tells you their causal relationship without trusting wall clocks at all:

VC1 = {A:2, B:1}   VC2 = {A:2, B:2}
Every entry of VC1 <= VC2, and one is strictly less
  -> VC1 happened-before VC2  -> VC2 is newer, safe to overwrite.

VC3 = {A:3, B:1}   VC4 = {A:2, B:2}
Neither dominates (A:3>2 but B:1<2)
  -> CONCURRENT  -> a real conflict, keep BOTH (siblings).

The read path returns both siblings to the client along with a merged context. The client (which understands the value’s semantics) reconciles - for a cart, it unions the items - and writes back a value whose vector clock dominates both, collapsing the siblings. The store never guesses; it detects concurrency correctly and hands genuine conflicts to the layer that can resolve them.

Write path:  coordinator on node A bumps A's counter -> {A:3, B:1}, stores it.
Read path:   two replicas return {A:3,B:1} and {A:2,B:2}
             -> concurrent -> return BOTH siblings to client.
Client merge: union carts, write back with VC {A:4, B:2} (dominates both)
             -> next read sees one reconciled value, siblings gone.

Where vector clocks get expensive: the clock has an entry per coordinating node, so under many coordinators or a long-lived key it can grow unbounded. The fix is to cap the clock and evict the oldest (node_id, counter) pairs with a timestamp, accepting a small chance of a false “sibling” in exchange for bounded size. In practice Dynamo saw clocks rarely exceed a handful of entries because a given key’s writes are usually coordinated by its small preference list.

Choosing between them (and dotted version vectors)

The answer: offer both and default by data type. Use LWW when a lost concurrent update is acceptable and you want zero application complexity (this is DynamoDB’s default and why it needs well-synced time via TrueTime-like services or accepts the risk). Use vector clocks / siblings when every concurrent update must survive (shopping carts, collaborative state) and the application can merge. Modern systems (Riak) use dotted version vectors to avoid a subtle false-conflict bug where a client rewriting its own value looks concurrent with itself. State the trade in the interview: LWW is simple but loses data under clock skew and concurrency; vector clocks are correct but push reconciliation onto the app.

4. Keeping Replicas Convergent: Hinted Handoff, Read Repair, Anti-Entropy

Quorum lets a write succeed with some replicas missing it, and partitions let replicas drift apart. Three background mechanisms drag them back into agreement so “eventual” consistency actually converges.

Naive: hope the failed writes retry

“If a replica missed a write, the next write will fix it.” Where it breaks: a key that is written once and then never touched again stays permanently divergent - one replica has it, another does not, forever. And a replica that was down for an hour missed thousands of writes it will never see. You need active repair, not passive hope.

The three repair mechanisms

Hinted handoff (repair the immediate miss). Covered above: when a target replica is down at write time, a healthy node stores the write with a hint and delivers it when the target returns. This heals transient failures within seconds to minutes. It does not help if the hint-holder itself dies before delivering, which is why it is only the first line.

Write to K, target replica D is down.
-> Node E (next healthy on ring) stores K + hint "belongs to D".
-> D comes back, E replays the hinted write to D, deletes the hint.

Read repair (repair on the read path, cheap and opportunistic). When a quorum read finds replicas disagree, the coordinator, after answering the client, pushes the reconciled/newest version to the stale replicas. Every read of a hot key thus self-heals its replicas for free. This handles the common case (recently-read keys converge fast) but does nothing for keys that are never read.

Read K from replicas [B, C, D]:
  B: {A:3,B:1}=v2   C: {A:2,B:1}=v1(stale)   D: {A:3,B:1}=v2
-> return v2 to client, and async write v2 back to C.

Anti-entropy with Merkle trees (repair everything, in the background). For the keys that are never read and the failures that outlive hints, replicas periodically compare their entire keyspace. Comparing key-by-key across terabytes is infeasible, so each replica builds a Merkle tree over its key ranges: leaves hash individual key-value versions, parents hash their children. Two replicas compare roots; if equal, they are identical and exchange nothing. If roots differ, they descend only into the differing subtrees, pinpointing the exact divergent keys while transferring almost no data.

Replica B root hash == Replica C root hash?  -> identical, done (one message).
Differ?  -> compare children, recurse only into mismatched subtrees.
Reach differing leaves -> exchange just those keys' versions, reconcile.
Cost: log(keys) comparisons instead of scanning all keys.

Merkle trees make full replica reconciliation cheap enough to run continuously, which is what guarantees eventual actually arrives even for cold keys and long outages. The three layers stack: hinted handoff for transient misses, read repair for read-hot keys, anti-entropy as the exhaustive backstop.

Deletes and tombstones. A subtlety: you cannot just remove a deleted key, because a replica that missed the delete would treat its surviving copy as a live value and resurrect it via anti-entropy. So a delete writes a tombstone - a special marker with its own version - that propagates like any write and dominates the old value. Tombstones are garbage-collected only after they are guaranteed to have reached all replicas (a grace period), otherwise a slow replica could resurrect the data. This is why “delete” in an eventually-consistent store is really “write a tombstone.”

API Design & Data Schema

Client API (the data plane)

A tiny key-addressed API, exposed over HTTP/gRPC. The context (opaque version token carrying the vector clock) is the important detail: clients get it on read and must return it on write so the store can order updates.

PUT /v1/{table}/items/{key}
    body:    { "value": <blob>, "context": "<opaque vclock token>" }
    params:  W (write quorum, default 2), consistency=lww|siblings
    resp:    201 { "context": "<new vclock token>" }        # W replicas acked

GET /v1/{table}/items/{key}
    params:  R (read quorum, default 2), consistent=true|false
    resp:    200 { "value": <blob>, "context": "<vclock token>" }
             # if siblings: 300 { "siblings": [ {value, context}, ... ] }

DELETE /v1/{table}/items/{key}
    params:  W, body: { "context": "<vclock token>" }
    resp:    204                                            # writes a tombstone

POST /v1/{table}/items/{key}:conditional-put
    body:    { "value": <blob>, "expected_context": "<vclock>" }
    resp:    200 OK | 409 Conflict     # compare-and-set, single-key atomicity

Notes: consistent=false reads hit R=1 (fast, maybe stale); consistent=true reads use the quorum R with R+W>N for read-your-writes. The 300-with-siblings response is the vector-clock conflict surfaced to the client for reconciliation. Conditional put gives single-key CAS (atomic counters, optimistic concurrency) without cross-key transactions.

Admin / control-plane API

POST   /admin/v1/nodes           # add node -> claims tokens, streams ~1/N of data in
       { "node_id": "kv-4001", "addr": "10.2.3.4:7000", "tokens": 256, "dc": "dc-east" }

DELETE /admin/v1/nodes/{node_id} # decommission -> streams its ranges to successors first

GET    /admin/v1/ring            # token map: node -> owned token ranges, per DC
PUT    /admin/v1/keyspace/{name} # replication config
       { "N": 3, "strategy": "network-topology", "dc_replicas": {"east":2,"west":1} }

network-topology replication places the N replicas across failure domains (racks / data centers) instead of N adjacent ring nodes that might share a rack, so a rack or DC loss never takes all N copies.

Data model: LSM-tree KV, not SQL

Access is by exact key at high write rate, values are opaque blobs, there are no joins or ad-hoc queries, and writes vastly outnumber random-key updates. This is the textbook profile for a log-structured merge-tree (LSM) key-value engine (as in Cassandra/RocksDB/DynamoDB’s storage), not a relational B-tree database. SQL is the wrong tool: a B-tree does random in-place writes (slow under a write-heavy load and expensive to replicate), and relational transactions add coordination latency this design exists to avoid. LSM buffers writes in memory (memtable), flushes them as immutable sorted files (SSTables), and merges them in the background (compaction) - turning random writes into sequential ones, which is exactly what a high-write replicated store needs.

Per-key stored record (on each replica):

Record {
  key            bytes        # the primary key (also hashed for placement)
  value          bytes        # opaque blob, up to a few MB
  vector_clock   map<node,int># causality metadata (or a single ts for LWW)
  write_ts       int64        # wall-clock (for LWW mode / tie-break / TTL)
  tombstone      bool         # true if this record is a delete marker
  checksum       int32        # detect disk corruption (bit rot)
}

Storage engine:  LSM tree
  memtable  (in RAM, sorted)  ->  flush  ->  SSTables (immutable, on disk)
  compaction merges SSTables, drops superseded versions & expired tombstones
  bloom filter per SSTable to skip files that can't contain a key (fast reads)

Cluster metadata (spread by gossip, not a central master):

node_id        STRING       # stable id
addr           STRING       # host:port
tokens         LIST<INT>    # ring positions this node owns (vnodes)
dc / rack      STRING       # failure domain, for topology-aware placement
status         ENUM         # joining | up | leaving | down
heartbeat      INT          # gossip counter, bumped each round; detects failure

The metadata is small and each node holds the full picture via gossip, so there is no central config server on the request path - any node can compute the preference list for any key from its own membership view. Gossip converges the view across thousands of nodes in a few rounds; a node is suspected dead when its heartbeat counter stops advancing across enough peers (a phi-accrual-style failure detector avoids flapping on a single slow link).

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. Concurrent-write conflicts and lost updates (breaks correctness first). With no leader and sloppy quorums, two coordinators accept conflicting writes to one key; naive LWW silently drops one. Fix: vector clocks to detect true concurrency and return siblings for semantic keys, LWW only where a lost concurrent update is acceptable. Cap vector-clock size by evicting oldest entries with a timestamp. This is the single most important correctness move in the design.

2. A replica down at write time threatens availability. Strict quorum would reject the write. Fix: sloppy quorum + hinted handoff - write to the next healthy node with a hint, satisfy W with any healthy nodes, deliver back on recovery. Writes essentially never fail; consistency is repaired afterward.

3. Replicas drift apart under partitions and cold keys. Never-read keys stay divergent forever. Fix: the three-layer repair - hinted handoff (transient), read repair (read-hot keys), anti-entropy with Merkle trees (exhaustive background backstop that reconciles cheaply via tree-hash comparison).

4. Rebalancing / rebuild storm on membership change. At petabyte scale, a node join or a failed-node rebuild moves enormous data. Fix: consistent hashing with virtual nodes moves only ~1/N of data on a join and lets a failed node’s replicas rebuild in parallel from hundreds of peers instead of one neighbor. Rate-limit streaming so rebuilds do not starve live traffic.

5. Hot key / hot partition overloading its replica set. One popular key concentrates load on its N replicas; even sharding cannot split a single key. Fix: raise that key’s replica count and serve eventually-consistent reads (R=1) across more replicas, cache the hottest keys in a front tier, and for hot ranges pick a shard key with high cardinality (avoid low-cardinality keys like a status flag that funnel writes to one range). DynamoDB’s adaptive capacity and write-sharding a hot key with a random suffix are the concrete fixes.

6. Tombstone resurrection and buildup. Deletes that garbage-collect too early get resurrected by a slow replica; tombstones that never GC bloat storage. Fix: keep tombstones for a grace period longer than the worst-case replica downtime / repair interval, then compact them away. Monitor tombstone ratio - a range full of tombstones makes reads scan lots of dead markers.

7. Cross-DC replication latency. Synchronous quorum across data centers adds WAN latency to every write. Fix: keep quorum local to a DC (LOCAL_QUORUM) for latency and replicate across DCs asynchronously, accepting cross-DC eventual consistency. Use topology-aware placement so the N replicas already span racks/DCs for durability without WAN on the hot path.

8. Read amplification from LSM compaction debt. A key’s versions can be spread across many SSTables; reads must check several files. Fix: bloom filters per SSTable to skip files that cannot hold the key, a tuned compaction strategy (leveled for read-heavy, size-tiered for write-heavy), and row/key caches for hot keys. Watch compaction throughput so it keeps pace with write ingest.

Wrap-Up

The trade-offs that define this design:

• Availability and partition tolerance over default strong consistency. We chose to stay writable through partitions and reconcile afterward, rather than reject writes to stay consistent. This is the Dynamo bet, and it is why quorums, vector clocks, and anti-entropy all exist - they are the machinery for cleaning up the conflicts that “always writable” inevitably creates.
• Tunable consistency over a fixed model. R and W are per-request knobs (with R+W>N for strong reads), so the caller decides where each operation sits on the latency-vs-freshness line instead of the system forcing one global choice. Fast-and-stale and slow-and-correct coexist on the same store.
• Consistent hashing with virtual nodes over modulo hashing. We give up the triviality of hash % N because at petabyte scale it would move nearly all data on every membership change. Consistent hashing moves ~1/N, and vnodes make placement even and rebuilds parallel - decisive when each node holds terabytes.
• Vector clocks over last-write-wins where updates must not be lost. We accept per-key causality metadata and push reconciliation onto the application, because wall-clock LWW silently discards concurrent updates under clock skew. LWW stays as the simple default only where a lost update is acceptable.
• Layered eventual-consistency repair over any single mechanism. Hinted handoff heals transient misses, read repair heals read-hot keys, and Merkle-tree anti-entropy is the exhaustive backstop that makes “eventual” actually converge - even for cold keys and long outages - with tombstones and a GC grace period to keep deletes from resurrecting.

One-line summary: a leaderless ring of identical LSM-backed nodes, keys placed and replicated N-ways by consistent hashing with virtual nodes, every read and write served by a tunable R/W quorum, concurrent writes ordered by vector clocks (or LWW) and replicas dragged back into agreement by hinted handoff, read repair, and Merkle anti-entropy - trading default strong consistency for a store that stays available and durable through node failures, partitions, and concurrent writes.