System Design: [Key-value store name]

Reference design for a distributed key-value store that can also run as a cache. It covers partitioning with consistent hashing and virtual nodes, replication, quorum reads and writes, consistency models and conflict resolution (vector clocks versus last-write-wins), hinted handoff, anti-entropy with Merkle trees, gossip membership and failure detection, an LSM-tree storage engine with compaction, eviction policies for cache mode and hot-key handling. The concepts follow the Amazon Dynamo paper (DeCandia et al., SOSP 2007) and its open-source descendants. It follows the structure of the System Design Document template and is pre-filled with a worked design to adapt. Values marked Example are assumptions. Replace every [bracketed] value and every row marked Example.

Document control

Approvals

Revision history

Contents

1. Overview
2. Requirements
3. Capacity estimation
4. API design
5. Data model
6. High-level design
7. Detailed design
8. Scalability and performance
9. Reliability, security and operations
10. Trade-offs and alternatives
11. Delivery plan and risks
12. Appendices

1. Overview

1.1 Problem statement

[Example: Session, cart and user-preference data live in a single relational primary that reaches 80% CPU at peak. Most access is by primary key, values are opaque blobs, and the business prefers accepting a write during a zone failure over rejecting it. Teams also run four separate cache clusters with different eviction settings and no shared operations.]

1.2 Proposed solution

[Example: A leaderless, replicated key-value store. Keys are placed on a consistent-hash ring with virtual nodes; each key is stored on three nodes in three zones. Any node can coordinate a request; reads and writes succeed when two of three replicas answer. Concurrent updates are detected with version vectors and returned to the client for merging, or resolved by last-write-wins for keyspaces that opt in. Temporary failures are covered by hinted handoff, permanent divergence by Merkle-tree anti-entropy, and membership by gossip with a phi accrual failure detector. Each node stores data in an LSM-tree engine. The same software runs as a cache with one replica, TTLs and LFU eviction.]

1.3 Decisions requested from reviewers

2. Requirements

2.1 Functional requirements

2.2 Non-functional requirements

2.3 Out of scope

• [Example: multi-key transactions and secondary indexes]
• [Example: cross-region synchronous replication (regions replicate asynchronously)]
• [Example: SQL or query language]
• [Item deferred, with the release that will take it]

3. Capacity estimation

3.1 Assumptions

3.2 Storage and node count

The cluster is set to 12 nodes (four per zone) so that each zone holds one full copy of the data and storage has room for one year of growth.

3.3 Replica operations

A coordinator sends each read to $R$ replicas and each write to all $N$ replicas (waiting for $W$ acknowledgements):

With a whole zone lost (eight nodes left) the load is 97 500 per node, 81% of saturation: the cluster survives but runs hot, so zone loss is a declared degraded mode that pauses compaction-heavy maintenance (9.1).

3.4 Network

Reads fetch the full value from one replica and a digest from the other, so read bandwidth is close to one copy per read. About 576 MB/s across 12 nodes is under 50 MB/s per node, far below a 10 Gbit/s link.

3.5 Summary

4. API design

4.1 Operations

4.2 Request and response examples

The service exposes gRPC; the JSON below shows the same shape for readability:

{"op": "get", "keyspace": "carts", "key": "user:48213", "consistency": "QUORUM"}

{"versions": [{"value": "eyJpdGVtcyI6WzEyLDQ0XX0=", "context": "vv:A=4,B=2"}, {"value": "eyJpdGVtcyI6WzEyLDQ1XX0=", "context": "vv:A=3,C=1"}], "conflict": true}

The client merges the two carts, then writes with a context that dominates both versions:

{"op": "put", "keyspace": "carts", "key": "user:48213", "value": "eyJpdGVtcyI6WzEyLDQ0LDQ1XX0=", "context": "vv:A=4,B=2,C=1", "consistency": "QUORUM"}

4.3 Errors

5. Data model

5.1 Entities

Each stored record carries its version vector; ring metadata is small and gossiped to every node.

5.2 Keyspace settings

6. High-level design

7. Detailed design

7.1 Partitioning: consistent hashing with virtual nodes

Keys are hashed (for example with a 64-bit Murmur3 or xxHash) onto a ring. Each node owns many points on the ring, called virtual nodes or tokens; a key belongs to the first token clockwise from its hash. With plain consistent hashing, adding or removing a node moves only the keys of the ranges next to it, about $1/n$ of the data.

Virtual nodes fix two problems of one-token-per-node rings: uneven range sizes and the fact that a joining or leaving node exchanges data with only two neighbours. With 256 tokens per node, a new node takes small ranges from every existing node, so rebalancing is spread across the cluster. Fewer tokens per node reduce metadata and repair overhead; more tokens reduce imbalance. Measure the load spread on your key distribution and record the choice.

7.2 Replication

The preference list for a key is the next $N$ distinct physical nodes clockwise from the key's position, skipping tokens of nodes already chosen and placing at most one replica per zone. With $N = 3$ and three zones, each zone holds one copy of every key, so losing a zone leaves two copies.

7.3 Quorum reads and writes

A write succeeds when $W$ replicas acknowledge it; a read returns when $R$ replicas answer. If the read set and write set must overlap in at least one replica:

With $N = 3$, $R = 2$, $W = 2$ gives $2 + 2 = 4 > 3$. Writes stay available with up to $N - W = 1$ replica down; reads with up to $N - R = 1$ down.

Overlap is not linearizability. Concurrent writes can still conflict, a write that failed to reach $W$ may remain on some replicas and be read later, and sloppy quorums (7.5) let writes land on nodes outside the preference list. Use cas (4.1) or a separate consensus-based store when an operation needs a single serial order.

Read repair: when the coordinator sees replicas with older versions in a read, it writes the newest version back to them after answering the client.

7.4 Consistency models and conflict resolution

The default model is eventual consistency. When R+W>N holds and no sloppy-quorum fallback is in play, a read observes the latest acknowledged write; during failures and hinted handoff it may not, so clients must not assume read-your-writes.

Version vectors. Each version carries a list of (node, counter) pairs. A coordinator increments its own counter on each write. Version X descends from Y if every counter in X is at least the one in Y; otherwise they are concurrent and both are kept.

Vectors grow with the number of coordinators; prune entries older than a threshold (for example keep the ten most recent) and accept that pruning can occasionally surface a false conflict.

Last-write-wins (LWW). Each write carries a timestamp and the highest wins. It needs no client merge, but clock skew silently drops writes: a node whose clock runs 200 ms fast wins against a real later write. Use LWW only where losing a concurrent update is acceptable (sessions, preferences), use hybrid logical clocks to limit skew effects, and monitor clock offset.

Alternatives: conflict-free replicated data types (counters, sets, registers) merge automatically and suit carts and counters; they constrain the value format.

7.5 Sloppy quorum and hinted handoff

When a replica in the preference list is unreachable, the coordinator sends the write to the next healthy node on the ring with a hint naming the intended owner. The fallback node stores the hint and replays it when gossip reports the owner alive again.

Hints are capped per target (for example 3 hours of writes); beyond that the hint is dropped and anti-entropy repairs the replica. A hinted write counts toward $W$ for availability, which is exactly why sloppy quorums weaken the overlap guarantee in 7.3.

7.6 Anti-entropy with Merkle trees

Each replica keeps a Merkle tree per token range: leaves are hashes of small key ranges, inner nodes hash their children. Two replicas compare roots; if they differ, they descend only into differing subtrees and stream the keys under differing leaves. With about 651 000 keys per token range and $2^{15} = 32\,768$ leaves, each leaf covers about 20 keys, so a repair moves little more than the divergent data.

Run repair on every range within the tombstone grace period (7.8); otherwise a replica that missed a delete can resurrect the deleted value.

7.7 Membership, gossip and failure detection

• Gossip: every second each node exchanges its view (node, generation, heartbeat version, status, tokens) with one to three random peers. State spreads to all nodes in a number of rounds that grows with the logarithm of cluster size.
• Seeds: a small list of seed nodes bootstraps new members and prevents logical partitions of the gossip graph.
• Failure detection: a phi accrual detector turns inter-arrival times of heartbeats into a suspicion level $\phi$; a node is marked down for routing when $\phi$ exceeds a threshold (Example: 8). Marking down is local and reversible; removing a node from the ring is an explicit operator action, so a flapping network never triggers mass data movement.

7.8 Storage engine: LSM tree and compaction

Writes go to a write-ahead log (fsync policy per keyspace) and an in-memory memtable. Full memtables flush to immutable sorted files (SSTables) with a Bloom filter and a sparse index. Reads check the memtable, then SSTables from newest to oldest, skipping files whose Bloom filter excludes the key. Deletes write tombstones that remain for a grace period (Example: 10 days) before compaction can drop them.

Throttle compaction so it never takes more than a set share of disk bandwidth at peak, and alert when pending compaction bytes keep growing.

7.9 Cache mode and eviction policies

In cache mode a keyspace runs with $N = 1$ (or 2 for hot data), no WAL, an in-memory engine and mandatory TTLs. When memory reaches the limit, the eviction policy picks victims:

Example: the page cache holds a hot set of 240 GB on six nodes with 48 GB usable each (288 GB, 83% full), LFU with decay, TTL 10 minutes. Cache misses fall through to the owning service, which must survive a cold cache; see 9.1.

7.10 Hot keys

Detect hot keys with a per-node top-k sketch (count-min sketch plus heap) exported every 10 s. Mitigate in order:

1. Client-side near cache with a short TTL (1 to 5 s) for read-hot keys.
2. Read from any replica (R=1) for keyspaces that accept staleness, which spreads a hot key over three nodes.
3. Split write-hot counters into $m$ sub-keys (key#0 to key#m-1) and sum on read.
4. Request coalescing at the coordinator so concurrent misses on the same key cause one storage read.

8. Scalability and performance

Latency budget for a QUORUM read (target p99 10 ms):

9. Reliability, security and operations

9.1 Failure modes

9.2 Security

9.3 Observability

9.4 Cost estimate

10. Trade-offs and alternatives

11. Delivery plan and risks

11.1 Milestones

11.2 Rollout and rollback

[Example: Dual-write from the relational database to the new store, compare reads in shadow mode, then switch reads per keyspace behind a flag. Rollback switches reads back; dual writes continue until the keyspace has run 30 days without discrepancy.]

11.3 Risks

12. Appendices

12.1 Open questions

12.2 Glossary

12.3 References

Standard and licence note

Based on the published Dynamo paper (DeCandia et al., 2007) and the other papers listed in 12.3, used as concept sources only, and on common industry practice for distributed key-value stores. This is a reference design to adapt, with illustrative numbers. Original text written for this collection; no text from the cited papers is reproduced. Check the documentation of the store you build on for its current behaviour and defaults. Template licence: MIT.