Chapter 21: Design a Distributed Key-Value Store

Curriculum: Distributed Systems from First Principles Chapter: 21 of 24 Part: 10 — Synthesis Design Problems Framework: Chapter 19's six-step process, applied in full Chapters exercised: 1 (latency numbers), 2–3 (storage internals, indexes), 4–6 (replication, consistency), 7–8 (sharding, consistent hashing), 9 (caching), 10 (LSM trees), 11–12 (consensus, Raft), 13 (distributed transactions), 14–15 (Kafka), 17 (rate limiting), 18 (observability) Estimated reading time: 85–95 minutes

The Problem Statement

Design a distributed key-value store that serves as the storage layer for a high-traffic e-commerce platform. It must support get(key), put(key, value), and delete(key) operations. It must handle millions of operations per second, replicate data across multiple datacenters for disaster recovery, tolerate node and datacenter failures, and offer tunable consistency so different clients can choose between speed and correctness depending on their use case.

This is Amazon's Dynamo problem — the system they published in 2007 and that became the foundational paper for an entire class of distributed databases (Cassandra, Riak, Voldemort). Building it from first principles requires almost everything in this curriculum.

Step 1: Requirements Clarification

Functional Requirements

1. put(key, value) — store a value under a key. Values up to 1MB. Keys up to 64 bytes.
2. get(key) → value — retrieve the value for a key, or "not found."
3. delete(key) — remove a key. (Internally implemented as a special put — a tombstone.)
4. No cross-key transactions. No secondary indexes. No range queries. Pure key lookup only.

Why no range queries? Range queries require keys to be sorted — which forces range-based partitioning (Chapter 7). Range partitioning creates hot-spot risk (sequential keys all land on one node). By restricting to exact key lookup only, the system can use hash-based partitioning (Chapter 8's consistent hashing), distributing keys uniformly regardless of their content. This is the fundamental trade-off that defines a key-value store vs. a relational database: give up query expressiveness, gain horizontal scale.

Non-Functional Requirements

• Scale: 10M operations/second (mixed read/write). 100 TB of data.
• Read/write ratio: 70% reads, 30% writes.
• Latency: P99 read < 10ms, P99 write < 20ms. (The e-commerce platform uses this store for product catalog lookups and shopping cart state — latency directly affects conversion rate.)
• Availability: 99.99% per datacenter. The system must remain available during: single node failure, network partition between nodes within a datacenter, complete datacenter failure.
• Consistency: Tunable. Some clients (shopping cart reads) accept stale data for lower latency. Other clients (inventory decrement) require strong consistency. The system must support both.
• Durability: Once a write is acknowledged, it must survive node failure.
• Durability scope: 3 datacenters, each with ~100 nodes. Data replicated across all 3 datacenters.

Step 2: Estimate Scale

Operations per second:
  Total: 10M ops/s
  Reads: 7M ops/s (70%)
  Writes: 3M ops/s (30%)
  Peak (3× average): 21M ops/s reads, 9M ops/s writes

Per-node load (target: ~100,000 ops/s per node):
  Nodes needed: 10M / 100,000 = 100 nodes per datacenter
  3 datacenters × 100 nodes = 300 nodes total

Storage:
  100 TB total data
  Replication factor = 3 (across nodes within a datacenter)
  Cross-datacenter replication: full copy in each of 3 datacenters
  Total physical storage: 100 TB × 3 (within-DC replication) × 3 (DCs) = 900 TB
  Per node: 900 TB / 300 nodes = 3 TB per node (achievable on commodity SSDs)

Write throughput per node:
  3M writes/s / 100 nodes = 30,000 writes/s per node
  LSM tree (Chapter 10): handles 100,000+ writes/s per node → comfortable headroom

Network bandwidth (within DC):
  Average value size: 10 KB (product catalog entries, shopping carts)
  10M ops/s × 10 KB = 100 GB/s aggregate
  Per node: 100 GB/s / 100 nodes = 1 GB/s → standard 10GbE NIC saturates at 1.25 GB/s
  Use 25GbE NICs or ensure per-node load stays below 10K ops/s avg × 10KB = 100 MB/s ✓

Cross-datacenter replication bandwidth:
  3M writes/s × 10 KB = 30 GB/s to replicate to 2 remote DCs
  30 GB/s × 2 DCs = 60 GB/s of cross-DC bandwidth
  This requires WAN links of 60 GB/s = 480 Gbps — this is the expensive part.
  Compression (typically 3:1 for JSON/text values): 160 Gbps. Feasible for a major cloud provider.

The key insight: writes drive the network cost, not storage. 3M writes/second × 10KB/write × 3 DC copies = 90 GB/s of WAN traffic. This is a real constraint at Dynamo scale and is why cross-DC replication is async by default — synchronous cross-DC writes would add 50–200ms of latency to every write (Chapter 1's cross-datacenter RTT).

Step 3: Data Model and Storage Layer

The Logical Model

A key-value store is deceptively simple: a giant hash table distributed across N nodes. But the storage layer determines performance.

Logical interface:
  put(key: bytes[64], value: bytes[1MB], metadata: {version, ttl, ...})
  get(key: bytes[64]) → (value: bytes, metadata)
  delete(key: bytes[64])  ← implemented as put(key, TOMBSTONE)

Physical storage per node: LSM tree (Chapter 10)
  Memtable: in-memory sorted structure
    key → {value, version_vector, timestamp, is_tombstone}
  SSTables: immutable on-disk segments
    Sorted by key within each SSTable.
    Bloom filter per SSTable (O(1) false-positive check for key existence).
  Compaction: merges SSTables, removes superseded versions, garbage-collects tombstones.

Why LSM over B-Tree for this workload? The numbers from Step 2: 30,000 writes/second per node. Chapter 10's RUM conjecture analysis: B-Trees optimise reads and update-in-place at the cost of write amplification (each B-Tree write can trigger page splits and cascading writes). At 30,000 writes/second sustained, B-Tree write amplification (typically 5–10×) pushes the SSD at 300,000 physical write IOPS — approaching NVMe limits. LSM's write path is a sequential memtable append — O(1) per write, no random I/O. Chapter 10's compaction amplification is traded for dramatically better write throughput. For a system that is 30% writes at massive scale, LSM is the correct choice. Cassandra, RocksDB, LevelDB, and DynamoDB all use LSM trees.

Bloom filters per SSTable: On a get, the system checks multiple SSTables (one per compaction level). Without a bloom filter, each SSTable requires a disk seek to check if the key exists. A bloom filter (a probabilistic bit-array data structure) answers "definitely not present" or "possibly present" in O(1) with ~10 bits of memory per key. False positive rate ~1% (1 in 100 "possibly present" answers are wrong — leading to an unnecessary disk read). For an LSM tree with 7 levels, bloom filters reduce the average disk reads per get from 7 to 1.01 — a 7× reduction in read I/O. At 7M reads/second, this is the difference between needing 7× the SSD I/O budget and just 1×.

Step 4: Partitioning — Consistent Hashing

100 nodes per datacenter, 100 TB of data. Each key must map to a specific node (or set of nodes for replication). Chapter 8's consistent hashing is the correct choice.

The Ring

Hash every key using a consistent hash function (e.g., MD5, taking 128 bits). Map every node to a position on the ring (using the node's ID as input to the same hash function). A key is assigned to the first node clockwise from the key's hash position on the ring.

CONSISTENT HASH RING (simplified, 8 positions for illustration):

         Key "cart:user42" → hash → position 0x3A
                    ↓
         0x00  Node A (0x05)
              ↗
 Node H       0x3A → assigned to Node B (next clockwise)
 (0xF0)  ←──────────────────────────── 0x50  Node B (0x50)
              ↖                    ↙
         Node G  Node C ... Node D
         (0xC0)  (0x80)     (0xA0)

With 100 nodes and 150 virtual nodes per physical node (Chapter 8's virtual node mechanism), the ring has 15,000 virtual node positions. This ensures statistical uniformity: each physical node owns approximately 1/100 of the keyspace. Without virtual nodes, hash clustering could put 30% of keys on one node and 2% on another — a severe hot-spot.

The coordination-free routing property: Any node can compute hash(key) → responsible node locally, without consulting a coordinator. This is Chapter 8's "client-side routing" — the client (or any node acting as a coordinator for the client) computes the target node from the ring topology, sends the request directly, and gets a response. No central coordinator. No single point of failure in the routing layer.

Ring membership: When a node joins or leaves, the ring topology changes. All nodes must converge on the same ring view. This uses a gossip protocol (Chapter 8's ring membership approach): each node periodically exchanges its ring state with a random peer. Convergence time: O(log N) rounds of gossip for N nodes. At N=100 with gossip every 1 second: ~7 rounds → ~7 seconds to converge. During convergence, slightly inconsistent routing occurs — requests may reach the wrong node. Chapter 8 shows this resolves through hinted handoff (covered below).

Step 5: Replication — N, R, W and the Quorum System

A single node storing a key is a single point of failure. The system replicates each key to N nodes — specifically, the N successive nodes clockwise from the key's position on the ring. With N=3, every key is stored on 3 nodes, called its preference list.

Ring with replication factor N=3:
  Key K hashes to position between Node A and Node B.
  Preference list: [Node B, Node C, Node D]  (B is primary, C and D are replicas)
  
  All writes go to all 3 nodes in the preference list.
  All reads may come from any of the 3 nodes.

Now the tunable consistency model: the R and W parameters.

W (write quorum): A write is acknowledged after W nodes confirm they received and stored it. If W=1, the write returns as soon as one node confirms — lowest latency, highest availability, weakest durability guarantee. If W=3, the write waits for all 3 nodes — strongest durability, highest latency.

R (read quorum): A read is acknowledged after R nodes respond. If R=1, the first response is returned — fastest but may be stale. If R=3, all 3 nodes are queried and the most recent version wins.

The quorum condition: For strong consistency, the write set and read set must overlap: R + W > N.

With N=3:

Strong consistency (R + W > N):
  W=2, R=2: 2 + 2 = 4 > 3 ✓
    Write requires 2 of 3 nodes. Read requires 2 of 3 nodes.
    At least 1 node is always in both sets → reads always see the latest write.
    Write latency: wait for 2nd node response (median of 3 node latencies).
    Read latency: wait for 2nd node response.

  W=3, R=1: 3 + 1 = 4 > 3 ✓ "write all, read any" — strong durability, fast reads.
    Every node has the value before write is ACKed.
    Read from any node sees every acknowledged write.
    But: write requires all 3 nodes → write availability drops when any node fails.

Eventual consistency (R + W ≤ N):
  W=1, R=1: 1 + 1 = 2 ≤ 3 — not strongly consistent.
    Write returns as soon as 1 node confirms.
    Read returns from the first node that responds.
    The read node might not have the latest write yet.
    But: lowest latency, highest availability.

What "strong" means here — and what it doesn't. R + W > N guarantees that a read overlaps with the latest acknowledged write, so it can see that version. It is not linearizability on its own: a read that runs concurrently with an in-progress write can see the new value while a slightly later read sees the old one, unless the system adds read repair before returning (the Chapter 11 practice problem walks through this). And the overlap guarantee disappears entirely once sloppy quorum (Step 7) lets writes land on nodes outside the key's preference list.

Choosing R and W per request: The system exposes R and W as per-request parameters. Shopping cart reads: get(key, R=1) — fast, stale data acceptable, cart is recovered on next page load. Inventory stock reads: get(key, R=2) with writes at W=2 — quorum overlap, so the stock level read is current (the decrement itself needs more — see below). The caller chooses the consistency level based on the semantics of the operation. This is precisely what Cassandra calls "consistency levels" (ONE, QUORUM, ALL).

Why W=3 for inventory? Consider the alternative: W=2 (write to 2 of 3 nodes). Node 3 is temporarily slow. Write is ACKed. 50ms later, another write arrives, goes to nodes 2 and 3 (node 1 is slow this time). Now: Node 1 has version 1, Nodes 2 and 3 have version 2. A read with R=2 from nodes 1 and 2 returns version 2 ✓. A read with R=2 from nodes 1 and 3 also returns version 2 ✓. Quorum reads correctly return the latest version because R + W > N ensures overlap. For reading the latest stock level, W=2 and R=2 is enough — W=3 adds latency without adding read safety.

But no choice of R and W makes a decrement safe. A decrement is read-modify-write. Two clients each read stock = 5 with R=2, each compute 4, each write 4 with W=3. Both writes succeed; two items were sold and stock went down by one. Quorums decide which versions a read sees; they do nothing about two writers racing. (With vector clocks the two writes become concurrent siblings — and for inventory, the shopping-cart rule "union the siblings" is wrong.) An inventory decrement needs a conditional write — compare-and-set on a version ("write 4 only if the current version is still v7"), which in Dynamo-style stores means a consensus-backed operation such as Cassandra's lightweight transactions or DynamoDB's conditional writes, or keeping inventory in a store with real transactions.

Step 6: Versioning — Vector Clocks

Here is where the design gets subtle. With N=3 and asynchronous replication, two clients might write different values to the same key simultaneously (on different nodes). When they converge, which value is "correct"?

Simple answer: last-write-wins (LWW) using timestamps. Problem: Chapter 1's physical reality — clocks drift. Server A's clock might be 100ms ahead of Server B. A write at Server A at T=1000 and a write at Server B at T=999 — B's write is logically "later" (it happened after A's in real time), but A's write has the higher timestamp. LWW with wall clocks silently discards real updates. This is dangerous for inventory and shopping carts.

The correct answer: vector clocks (Chapter 6's causal ordering). A vector clock is a list of (node_id, counter) pairs, one per node that has participated in the value's history. When a node writes a value, it increments its own counter in the vector clock.

When a client reads this key with R=2 and gets two siblings, what happens? The system returns both versions to the client. The client must reconcile the conflict and write back the merged result:

This is the approach Amazon Dynamo uses for shopping carts. The client (application layer) resolves conflicts using domain knowledge — for a shopping cart, union is correct (never silently drop an item). For other data types, last-write-wins, numeric max, or custom merge functions may be appropriate.

CRDT (Chapter 6): If the application can model the value as a CRDT (conflict-free replicated data type), conflicts are resolved automatically without client involvement. A grow-only set (G-Set) CRDT for a shopping cart always merges by union — no siblings, no client-side reconciliation needed. For counter values (view counts, inventory), a PN-Counter CRDT tracks increments and decrements separately and merges by summing each node's contribution.

Pruning vector clocks: Vector clocks grow without bound as more nodes write a value — one entry per participating node. After N nodes have written a value across its history, the vclock has N entries. Pruning: discard the oldest entry if the vclock exceeds a size threshold. This introduces a small risk of false "concurrent" detection (treating a causally-related update as concurrent), but in practice vector clocks for a single key rarely involve more than 5–10 nodes.

Step 7: Availability During Failures — Sloppy Quorum and Hinted Handoff

The preference list for a key is [Node B, Node C, Node D]. Node C crashes. A write arrives with W=2 — it needs 2 of the 3 preference-list nodes. Without any further mechanism: the write goes to B and D (2 of 3) — W=2 is satisfied. But what about Node C's copy of the data?

The problem: When Node C comes back, it has missed all writes that occurred during its downtime. Reads with R=2 from nodes B and C might now return a stale value from C.

Sloppy quorum: During the failure, the system "slides" the preference list. Instead of requiring quorum from [B, C, D], it allows any N healthy nodes in the ring to accept the write. Node E (the next node on the ring after D) temporarily accepts writes intended for C. This is called a sloppy quorum — the W nodes are not necessarily the canonical preference-list nodes.

Hinted handoff: When Node E accepts a write intended for Node C, it stores the value with a "hint" in the metadata: {target: C, value: ..., timestamp: ...}. When Node C recovers, Node E detects this (via gossip — C is back in the ring view) and forwards the hinted data to C. Node C replays the hinted writes, catching up on missed updates. Node E then deletes its local hint.

Sloppy quorum + hinted handoff is the key mechanism that gives DynamoDB-style systems their availability property: they stay available during node failures and even during brief network partitions, at the cost of potentially serving stale reads during the failure window.

The trade-off (Chapter 6's CAP theorem): By using sloppy quorum, the system chooses availability over partition-tolerance-of-consistency. During a partition, some nodes may not have the latest write, and reads from those nodes may return stale data (eventual consistency). The system is not linearizable under network partition. For shopping carts, this is the right choice. For inventory (where balance = 50 must not become balance = 50 on two nodes simultaneously, each decrement independently reaching 49), you need W=3 and quorum reads — or a different system altogether (see Chapter 13's distributed transactions).

Step 8: Anti-Entropy — Merkle Trees

Hinted handoff handles short-term node failures (minutes). But what about longer failures, or gradual data divergence due to silent data corruption? Two nodes might have subtly different values for thousands of keys without either realising it. Comparing every key explicitly would require transferring 100 TB of data.

Merkle tree anti-entropy (named after cryptographer Ralph Merkle) solves this. A Merkle tree is a binary tree where:

• Each leaf node contains the hash of a specific key-value pair.
• Each internal node contains the hash of its children's hashes.
• The root hash is a single value that captures the state of the entire keyspace.

Merkle tree for a partition (simplified, 4 keys).

Synchronisation protocol between two nodes:

Efficiency: A Merkle tree for 1 million keys has depth ~20. Synchronising two nodes requires at most 20 rounds of hash comparisons to identify the exact divergent keys, even if only 1 key out of 1 million differs. Without Merkle trees, you'd need to transfer all 1 million key hashes to find the 1 divergent key — 1000× more data.

Continuous background anti-entropy: Nodes continuously run the Merkle tree synchronisation with their replica peers in the background. Cassandra runs anti-entropy repair as a background operation. DynamoDB uses similar mechanisms. This ensures that even after extended node failures, eventual consistency is achieved — divergent replicas converge without human intervention.

Step 9: Read Repair — Lazy Consistency

Anti-entropy is proactive but asynchronous (runs in the background on a schedule). Read repair provides lazy consistency on the read path itself.

When a read with R=2 is issued:

Read repair is free — the coordinator already fetched from multiple nodes for the quorum read. It just notices the inconsistency and fixes it as a side effect. This is a "lazy" consistency mechanism — it converges replicas over time driven by read traffic. Frequently-read keys are always up-to-date (read repair fires on every stale read). Rarely-read keys might remain stale until anti-entropy or the next read.

Step 10: Cross-Datacenter Replication

The system spans 3 datacenters. A write in DC1 must eventually reach DC2 and DC3.

Async cross-DC replication:

• The write is acknowledged to the client after W nodes in the local DC confirm.
• A background replication pipeline (Kafka topic per DC pair — Chapter 15) asynchronously ships the write to remote DCs.
• Remote DCs apply the writes in arrival order within each partition.
• Replication lag: typically 50–200ms (Chapter 1's cross-DC RTT), but can grow during outages.

Why async? Synchronous cross-DC writes would mean every put waits for a response from all 3 DCs before ACKing the client. DC1 → DC2 RTT: ~70ms. That's 70ms added to every write — for a system targeting P99 write latency < 20ms, this is a non-starter. The availability cost is also severe: any cross-DC network blip causes write timeouts.

Conflict resolution across DCs: Since writes can happen in DC1 and DC2 simultaneously (both serving traffic), the same key might be written in both DCs before replication converges. The vector clock mechanism from Step 6 handles this: each datacenter contributes to the vector clock. When DC2 receives DC1's write and detects a concurrent sibling, the same reconciliation logic applies — either CRDT merge or client-side resolution.

Active-active vs active-passive (Chapter 16's global LB pattern): This system is active-active — all 3 DCs serve reads and writes simultaneously. An alternative is active-passive: only one DC accepts writes, others serve reads. Active-passive avoids cross-DC write conflicts (no concurrent writes) but is overkill latency for users far from the active DC and wastes the capacity of the passive DCs. Active-active is harder to implement correctly but serves users from the closest DC.

Step 11: The Get and Put Flows — End to End

Put Flow

Get Flow

Why query all N and return after R? This is the "send R hedged requests, return on first R responses" pattern. Sending to all 3 nodes simultaneously and returning when 2 respond means the coordinator waits for the median of 3 latencies — not the 2nd-slowest. If one node is slow (network hiccup, GC pause), the other two respond and the read is fast. Without this, a "pick 2 of 3 upfront" strategy would wait for the slowest of those 2, which could be the slow node.

Step 12: Membership and Failure Detection — Gossip Protocol

In a 100-node cluster, every node needs to know which other nodes are alive. A centralised failure detector (one node that all others heartbeat to) is a single point of failure. The solution: gossip.

Gossip protocol:

Every 1 second, each node:
  1. Picks 3 random peers.
  2. Sends its current membership view:
     { "alive": [A, B, C, ...], "suspected_down": [X], "dead": [Y] }
  3. Merges incoming views with its own:
     If peer says node X has not heartbeated in 10 seconds: mark X as "suspected down"
     If X has not heartbeated in 30 seconds: mark X as "confirmed dead"
  4. Adjusts consistent hash ring: removes dead nodes, adds recovered nodes.

Convergence: With N=100 nodes and each node gossiping to 3 peers/second: after ~log₂(100) ≈ 7 gossip rounds (7 seconds), all nodes have the information. Gossip convergence is O(log N) — it scales gracefully to thousands of nodes.

False positives (marking a live node dead): Heartbeat timeouts are inherently imprecise. A node GC pause lasting 8 seconds might look like a node failure. The system marks it as "suspected" but waits for a longer timeout (30 seconds) before declaring it "dead" and removing it from the ring. During the suspected period, the node is deprioritised in preference lists (requests go to other replicas first). If it recovers, it rejoins the ring and anti-entropy synchronises any missed writes.

Step 13: The Full Architecture

Inside each storage node (every node runs the same stack):

There is no "master" node. Every node is equal — any node can be a coordinator for any request. Routing, failure detection, and replication are all decentralised. This is the property that makes the system survive any single node failure and even survive the loss of an entire datacenter.

Step 14: The Consistency Spectrum — Where This System Sits

Connect this design to Chapter 6's consistency models:

Linearizability (strongest): Every read sees the most recent write. Requires quorum reads and writes with R + W > N AND a locking mechanism to prevent concurrent writes. This system with W=3 and R=2 (R + W > N=3) achieves linearizability within a datacenter for single keys. But vector clocks and sloppy quorum mean that during network partitions, linearizability may be violated. This system is NOT linearizable under all network conditions.

Sequential consistency: All operations appear to execute in some total order consistent with the program order. Achievable with R + W > N within a single DC, but not across DCs with async replication.

Causal consistency: If write A causally precedes write B, any process that sees B also sees A. Vector clocks track causal dependencies — if client C reads a value with vclock {A:2}, any subsequent read by C will not return a value with vclock {A:1}. This system achieves causal consistency within a datacenter.

Eventual consistency (weakest): All replicas eventually converge. Guaranteed by Merkle tree anti-entropy and read repair. Cross-DC replication provides eventual consistency across datacenters.

The practical setting for the e-commerce platform:

Shopping cart reads/writes:  W=1, R=1 (eventual consistency)
  → Fast, available, acceptable stale reads
  → Conflict resolution via CRDT (shopping cart union)

Session data (auth tokens): W=2, R=2 (causal consistency within DC)
  → User sees their own login immediately after authenticating

Product catalog (read-heavy): W=2, R=1 (high read availability)
  → Writes are rare, reads are massive (70% of load)
  → R=1 fast reads, W=2 ensures durability

Inventory levels:            W=3, R=2 (strong consistency, single DC)
  → All 3 nodes must confirm the write
  → Read always sees the latest committed write
  → Cross-DC inventory is handled by a separate transactions layer (Chapter 13)

Step 15: Failure Modes and Edge Cases

The "lost write" under sloppy quorum: A write goes to Nodes B and E (E is holding a hint for Node C which is down). W=2 is satisfied. Node C comes back. Node E's hint delivery fails (E itself crashes before delivering). Now Node C doesn't have the write, and Node E (which had the hint) is also down. With R=2 from nodes B and D: B has the write, D doesn't → conflict detected → read repair. With R=2 from nodes C and D: C doesn't have write, D doesn't → stale read returned. This is the fundamental limitation of eventual consistency — transient data loss windows exist.

Mitigation: set W=3 (all nodes in preference list) for writes where data loss is unacceptable. This closes the sloppy quorum window for that key.

Partition between DCs: DC1 and DC2 can no longer communicate. Both continue serving reads and writes independently (availability choice — Chapter 6). Writes in DC1 and DC2 to the same keys create conflicts. When the partition heals, vector clocks detect the conflicts and CRDT merges or client reconciliation resolves them. During the partition, the system is available but not consistent across DCs.

Clock skew causing incorrect LWW: Even with vector clocks, some implementations use a "last write wins" fallback when vector clocks show concurrent writes and CRDT reconciliation is not configured. Clock skew (Chapter 1: NTP drift can be ±500ms) can cause a newer write to have an "older" timestamp. Always use vector clocks or logical timestamps (Lamport clocks) for ordering, never wall-clock timestamps alone.

Compaction blocking reads (Chapter 10 revisited): During compaction, the LSM engine is merging SSTables on the same NVMe SSD that serves read I/O. I/O contention increases read latency. Chapter 10's solution: use a separate thread pool for compaction with I/O rate limiting. The Cassandra configuration compaction_throughput_mb_per_sec caps compaction I/O at N MB/s to protect read latency.

Observability — What to Monitor (Chapter 18 Applied)

Per-node metrics:
  kvstore_operations_total{op="get"|"put"|"delete", result="success"|"error"} (counter)
  kvstore_latency_seconds{op="get"|"put"} (histogram — track P50, P99, P999)
  kvstore_quorum_timeout_total (counter — write or read failed to achieve quorum)
  kvstore_hinted_handoff_pending (gauge — hints waiting to be delivered)
  kvstore_compaction_pending_tasks (gauge — Chapter 20's incident was compaction backlog)
  kvstore_memtable_size_bytes (gauge — when memtable is full, flush to SSTable)
  kvstore_bloom_filter_false_positives_total (counter — too many means keys are not found on first SSTable)

Ring health metrics:
  kvstore_ring_members_alive (gauge — should equal N=100)
  kvstore_ring_members_suspected_down (gauge — alert if > 0 for > 60s)
  kvstore_ring_members_dead (gauge — alert immediately if > 0)

Replication metrics:
  kvstore_replication_lag_seconds{dc="dc2"|"dc3"} (gauge — cross-DC lag)
  kvstore_anti_entropy_divergent_keys_total (counter — keys fixed by Merkle sync)

SLO: P99 read < 10ms, P99 write < 20ms, 99.99% success rate.
Alert burn rate > 6× → page (budget exhausted in 5 days).
Alert burn rate > 14× → page immediately (budget exhausted in 2 hours).

The single most important operational alert for a KV store: kvstore_quorum_timeout_total rising. Quorum timeouts mean a write or read could not achieve R or W confirmations within the timeout window. This could mean: nodes are down (check ring membership), network is degraded (check inter-node latency), or the cluster is overloaded (check CPU/memory). A rising quorum timeout rate is the precursor to availability degradation.

Vocabulary Map

Practice Problems

Problem 1 — Failure Analysis

Your cluster has N=3, and the system is configured with W=2, R=2. Three nodes hold a key: Node B has version v3, Node C has version v3, Node D has version v2 (stale — it missed one write).

(a) A read with R=2 queries nodes B and C: what is returned? Does read repair fire?

(b) A read with R=2 queries nodes C and D: what is returned? What do the vector clocks look like? {B:3} vs {B:2} — which dominates, and what does the coordinator return? Does read repair fire, and if so, what does it do?

(c) A read with R=2 queries nodes B and D: same question.

(d) Now Node C fails. A read with R=2 can only reach nodes B and D. Under strict quorum (only use preference-list nodes), R=2 cannot be satisfied — the read fails. Under sloppy quorum, the read may proceed. What is the risk of a sloppy quorum read of this key during Node C's failure? Which nodes might respond, and what is the chance of returning a stale value? Quantify this probability if one of B/D is the stale Node D.

Problem 2 — Design from Scratch

Extend the key-value store to support TTL (time-to-live): put(key, value, ttl_seconds). After ttl_seconds, get(key) must return "not found." This is how session stores work — an auth token expires after 1 hour.

Design the full TTL implementation: (a) what is stored in the value metadata, (b) how does get check expiration without touching the LSM tree unnecessarily (hint: the expiration time is in the LSM record — where is it checked?), (c) how are expired keys removed from the SSTable (can you delete them immediately? Why not? Connect to tombstone and compaction), (d) how does TTL interact with replication — if Node B holds a key with TTL=3600s and Node C receives the replicated write 5 seconds later, does Node C's TTL start at 3600s from the original write or from when it received the replication? What is the correct answer and why does it matter for consistency?

(e) At scale: 100M keys with TTL=3600s expire every hour = 100M / 3600 = ~27,800 expirations/second. Each expiration requires a tombstone write. How does this affect write amplification during compaction? Design a "lazy expiration" strategy that checks TTL only at read time, avoiding tombstone writes entirely, and analyse the storage overhead (expired keys remain in SSTables until compaction).

Problem 3 — Synthesis

Connect this KV store to Chapter 12's Raft and Chapter 13's distributed transactions.

The KV store provides eventual consistency for most operations. But for inventory management (inventory:product42 = 50), you need linearizable reads and writes — no stale reads, no lost updates.

(a) Design a "strong consistency mode" using Raft (Chapter 12) for specific keys tagged as consistency=strong. Instead of quorum reads/writes across independent nodes, these keys are managed by a Raft group: 3 nodes form a Raft cluster for a subset of keys. All writes go through the Raft leader. All reads either go through the leader (linearizable) or use Raft's lease mechanism (Chapter 12's linearizable reads section). What is the latency cost of Raft consensus (leader election RTT + log replication) compared to quorum writes? Reference Chapter 1's latency numbers.

(b) An inventory decrement is a read-modify-write: old = get(key), new = old - 1, put(key, new). If two clients do this concurrently with quorum consistency (W=2, R=2), both might read old=50, both compute new=49, and both write — inventory drops from 50 to 49 instead of 48. This is a lost update. How does Raft's linearizability prevent this? What does the Raft log entry look like for a compare-and-swap (CAS(key, expected=50, new=49)) operation?

(c) A shopping cart add-to-cart (put, eventual consistency) and an inventory decrement (CAS, strong consistency) must be atomic: either both succeed or both fail. This is a distributed transaction across two keys with different consistency requirements. Apply Chapter 13's Saga pattern: define the two steps, their compensating actions, and the failure scenario where the cart is updated but inventory decrement fails.

Problem 4 — Go Deeper

Analyse Merkle tree anti-entropy performance at scale.

Your cluster has 100 nodes, each responsible for 1 TB of data (100 TB total). The Merkle tree for each node's partition has depth ~30 (2³⁰ ≈ 1 billion leaf slots for individual keys).

(a) Two nodes compare their Merkle trees and find 100 divergent keys (out of 1 billion total). How many hash comparisons are required? (Each comparison requires one network round trip.) At 500μs per round trip (Chapter 1's within-DC latency): how long does the anti-entropy sync take? Compare this to the naive approach (transfer all 1 billion key hashes).

(b) Anti-entropy runs continuously between all replica pairs. With N=3 (3 replicas per key) and 100 nodes, each key has 2 replica pairs to synchronise. Anti-entropy round: each node syncs with its 2 replica peers once per hour. At 100 nodes × 2 replica peers × 1 Merkle comparison/hour: how many anti-entropy comparisons per second across the cluster? What is the background network load of these comparisons (each comparison transmits ~30 hash values per divergent subtree, ~32 bytes each)?

(c) An adversarial scenario: a bug causes 50% of keys on Node A to have incorrect values (silent data corruption — Chapter 4's corruption detection discussion). How long does it take for anti-entropy to detect and repair all corrupted keys? (Assume 30-minute anti-entropy cycle, Merkle sync identifies divergent keys, repair transfers correct values from healthy replica.) What is the total data transferred during repair? Is this within the 1 GB/s NIC budget per node?

Problem 5 — Connect to Previous Chapters

This problem synthesises the KV store with Chapter 15's Kafka and Chapter 8's consistent hashing for a specific use case: building a Change Data Capture (CDC) pipeline on top of the KV store.

You want to stream every write to the KV store into a Kafka topic so downstream consumers (analytics, search indexing, cache invalidation) can react to changes in real time.

(a) Design the CDC integration: when a put(key, value) is processed by any node, it should produce an event to a Kafka topic. The event must include the key, the new value, the vector clock, and the timestamp. The Kafka topic is partitioned by key hash — the same key always goes to the same Kafka partition (Chapter 15's partition key design). Why is this important for downstream consumers that need to process events for a key in order?

(b) The CDC producer runs on every storage node. 100 nodes each producing events at 30,000 writes/second = 3M Kafka events/second. At the Kafka level (Chapter 15's throughput analysis): how many Kafka partitions are needed to handle 3M events/second if each partition handles 50,000 events/second? How does this relate to the number of downstream consumer instances?

(c) A node crashes mid-write: it wrote to its local LSM memtable but crashed before producing the Kafka event. When the node recovers and replays its WAL (Chapter 2's WAL discussion — the LSM tree writes to WAL first), it replays the write to the memtable. Does it also replay the Kafka produce? How do you ensure exactly-once Kafka delivery for CDC events (Chapter 15's idempotent producer + transactional API)? What is the idempotency key for the Kafka produce?

(d) Anti-entropy (Merkle sync) repairs a key on Node C by writing the correct value. Should this repair write produce a CDC event? If yes: downstream consumers see a "new write" that is actually a repair — they might invalidate a cache unnecessarily. If no: the CDC stream is not a complete record of all writes to the KV store. Which is correct, and why? How does the vector clock on the repair write help downstream consumers distinguish repair events from real client writes?

What's Coming Next

Chapter 22 designs a global Content Delivery Network — the system that makes the web fast. CDNs cache static content (images, CSS, JavaScript, video) at the edge, close to users, so the physical distance a byte travels is measured in milliseconds instead of hundreds of milliseconds. Building a CDN from first principles requires everything from cache invalidation (Chapter 9) to consistent hashing (Chapter 8) to anycast routing (Chapter 16) to distributed coordination (Chapters 11–12). It is also the chapter where the system itself — the infrastructure serving your content — has to handle petabytes per second globally. The numbers in Chapter 22 are larger than any chapter so far.

Chapter 21 complete. The distributed KV store is the foundational layer beneath Chapter 20's messaging system (presence data in Redis is a single-node KV store), beneath Chapter 9's Redis (a single-node KV store with network access), and behind every database that uses a key-value storage engine (RocksDB is the LSM KV engine inside Cassandra, TiKV, and CockroachDB). Problem 3 (Raft for linearizable inventory + Saga for distributed transaction) is the hardest synthesis problem in this chapter and represents a senior staff engineer question. Problem 4 (Merkle tree performance analysis) is the one most candidates skip — do it, because Merkle trees appear in every anti-entropy question and in blockchain interviews too.