System Design Case Study
Distributed key-value store
Cart items land on flaky train Wi-Fi; the app closes, then opens again in another city. The cart should still be there even if a data center hiccuped: availability matters more than a perfectly synchronized ledger for this product.
A key-value store is a giant distributed dictionary. Hash the key, find its , satisfy , and repair conflicts with after failures.
put / get journey
Ring + vnodes pick replica owners.
Coordinator waits for W of N acks.
If a replica is down, write a hint.
Merge versions; repair in background.
Try it: put/get on a partitioned KV store
Keys hash to partitions. Replication factor and quorum decide when a put is durable and which get is allowed.
- Client
- Coordinator
- Replica set
- Commit
- Read repair
- Puts
- 0
- Gets
- 0
- Mode
- put
- Path hop
- 0
Partition
hashCoordinator routes by key hash to a replica set.
Ops
quorum0 puts, 0 gets.
Repair
anti-entropyRead repair and background compaction keep replicas converging.
Put with quorum acknowledgment, then get with optional read repair.
What actually happens
Any node can be the coordinator. It uses to find the , then fans out. Writes land on an path (commit log + memtable) and succeed when W is met: possibly via and if a preferred replica is down.
Reads collect R replies, compare , return the merged value, and kick off . Background anti-entropy uses so long-lived divergence still heals.
This design prefers AP: stay available during partitions, then reconcile. Shopping carts merge by union; money balances usually need a different CP story.
How to choose consistency knobs
1. Can the product accept stale or sibling values? Carts and session blobs often can. Ledger balances usually cannot: do not force an AP store into a CP problem.
2. How strong should a read feel? Raise R (and often W) when overlap must catch the latest write; lower them when latency matters more.
3. What happens when a node is down for an hour? Size backlogs and schedule so recovery does not surprise you.
Simple rule of thumb
Start from Dynamo-style building blocks: , N replicas, tunable , versions, hints, and anti-entropy. Say out loud what conflict merge means for your value type.
In simple terms: find the machines, talk to enough of them, and have a plan when they disagree.
Case-study upgrade
Key-Value Store Quorum Math
A distributed key-value store maps keys to replica sets. The core interview question is what happens during node, disk, and network failure.
Assumptions and formulas
- Logical data: 1 PB. Replication factor N=3. Nodes: 1000. Throughput target: 1M ops/sec.
- Average value size must be stated; 10 KB values make write bandwidth enormous.
- LSM-style storage adds compaction, tombstone, and index overhead.
Concrete math
- Storage: 1 PB logical x 3 replicas = 3 PB before overhead.
- Per-node primary data: 1 PB / 1000 = 1 TB average; replicated responsibility is about 3 TB/node before overhead.
- With 1.3x to 2x LSM overhead, physical storage can become 4 to 6 TB/node.
- Throughput: 1M ops/sec / 1000 nodes = 1K logical ops/sec/node average.
- Quorum N=3, W=2, R=2: each logical write sends to 3 replicas and waits for 2; replica writes can reach about 3M/sec.
- Hinted handoff: one node down for 1 hour with 0.1% of 1M writes/sec creates about 3.6M hints; at 10 KB each, about 36 GB before overhead.
Design decisions forced by numbers
- Choose consistency options per API call and explain what stale reads mean.
- Schedule anti-entropy repair and compaction so background work does not crush foreground latency.
- Keep vector-clock metadata bounded or use last-write-wins only when lost updates are acceptable.
Scaling to millions
- At 1M keys: simple sharded store with replicas is enough.
- At 100M keys: add consistent hashing, vnodes, quorum reads/writes, compaction, and repair.
- At billions of keys: add multi-region placement, tombstone policy, hot partition detection, and client consistency choices.
Readable flow
Interview follow-ups to expect
- What does the client see during a network partition?
- How are conflicting versions resolved?
- How much compaction and repair bandwidth is safe?
Operational readiness checklist
After the math, explain how this capacity plan is measured, protected, launched, and evolved.
Observability
- Attach each capacity estimate to a dashboard metric: p99 latency, queue age, cache hit rate, shard skew, error rate, and freshness lag.
- Alert on rate-of-change and saturation, not only hard failures; backlogs and hot keys usually warn before an outage.
Security and abuse
- Budget for auth checks, rate limits, quota isolation, audit logging, privacy filters, and abuse-review side paths.
- Separate trusted internal events from user-controlled input so retries and backfills cannot bypass policy checks.
Degraded modes
- Define the fallback when a dependency is slow: stale cache, partial results, read-only mode, delayed async work, or admission control.
- Protect the invariant first; reduce freshness, personalization, or analytics quality before risking duplicate or corrupt state.
Testing and rollout
- Test average, peak, and hot-object traffic; include retry storms, cold caches, lagging workers, and regional dependency loss.
- Use canaries, shadow reads, dual-write validation, feature flags, and explicit rollback criteria for risky path changes.
Rollout, backfill, and runbooks
- Make migrations and backfills idempotent, resumable, rate-limited, and visible through progress and error metrics.
- Write the runbook before launch: owner, dashboards, paging thresholds, mitigation switches, replay steps, and repair commands.
10x evolution
- For key-value store quorum math, state which number forces the next move: cache memory, shard count, write amplification, queue lag, egress, or failover reserve.
- Add the next layer only when the metric forces it: sharding, regional reads, async replay, hot-key isolation, or stricter backpressure.
Scope you should nail down
Product surface
- put(key, value, context)
- get(key) → (value, context)
- delete(key)
- Keys are unique; values are opaque blobs to the storage layer
- Small key-value pairs, typically less than 10 KB
- Context for versioning/conflict detection
- Tunable consistency (R, W)
Scale targets
- Highly available (AP system)
- Eventually consistent
- 1M ops/sec
- 1PB data, 1000 nodes
- Tunable R/W for consistency
- Low latency for get/put under normal failures
- Automatic scaling as nodes are added or removed
Key-Value Basics and Single-Server Limit
Data model
- Key: unique identifier such as
user:145or a hashed value - Value: opaque bytes, JSON, string, list, object, or binary payload
- Short keys improve memory and network efficiency at scale
Single-server design
Use an in-memory hash table for hot keys, optionally compress values, and spill cold data to disk. This is fast but limited by RAM, disk, and one machine's failure domain.
Why distribute?
Big data, high QPS, availability, and automatic scaling require partitioning keys across many nodes and replicating each key to survive node loss.
Capacity estimation
Per node: 1K ops/sec average, ~1 TB logical primary before replication
Replication: 1 PB logical × RF=3 = 3 PB stored before LSM overhead
LSM overhead: compaction + tombstones + indexes at 1.3x-2x → 4-6 TB physical per 1 TB logical node responsibility
Quorum writes: W=2, N=3 → each logical write sends 3 replica messages; 1M ops/sec ≈ 3M replica writes/sec
Write bandwidth: 1M writes/sec × 10 KB avg value = 10 GB/sec logical: forces smaller values, lower write share, or more nodes
Hinted handoff backlog: node down 1 hr at 0.1% of 1M writes/sec ≈ 3.6M hints × 10 KB ≈ 36 GB before overhead
Vector clock overhead: 48-160 bytes/version: negligible at 10 KB values, painful at tiny values
Scaling to millions of keys
At 1M keys
Simple sharded store with a few replicas; strong consistency may still fit on one primary per shard.
At 100M keys
Consistent hashing with vnodes, quorum reads/writes, compaction scheduling, and read repair become core operations.
At billions+ keys
Multi-region placement, anti-entropy scheduling, tombstone compaction policy, hot-partition detection, and client consistency options.
3. 🔌 System API
put(key, value, context?)Returns context (for subsequent reads)
get(key)Returns (value, context): context: vector clock for conflict detection
delete(key)Tombstone (soft delete)
batchPut(items[]) / batchGet(keys[])Batch operations for efficiency
CAP Choice and Consistency Model
Ideal case
With no partition, replicas exchange writes normally. Clients can get both high availability and a consistent view.
CP behavior
Prefer consistency. If replicas cannot communicate, reject or block reads/writes that might see stale data. This fits bank-balance style correctness.
AP behavior
Prefer availability. Accept reads/writes during partitions, then reconcile conflicts later using versioning, read repair, and anti-entropy.
What this design chooses
This page designs a quorum-based AP store by default: high availability, partition tolerance, eventual consistency, tunable quorum, and client or application-assisted conflict resolution.
4. 📍 Data Partitioning
Consistent Hashing
Hash ring: nodes placed on circle. Key → hash(key) → clockwise to next node. Virtual nodes: 100-200 per physical node for even distribution.
// Node addition: only K/N keys move (K=keys, N=nodes)
// Node removal: keys move to next node
// Rebalancing: gradual, no full reshuffle5. 🔄 Replication
Preference List
Replicate to N successive nodes on hash ring. Skip same physical node (rack-aware). Preference list = top N nodes for a key.
Sloppy Quorum
A strict quorum uses the key’s designated replica set. A sloppy quorum may store a write temporarily on a healthy substitute when an intended replica is unreachable; the substitute records where the write eventually belongs. This can accept writes through some partitions, but a later read from the original replicas may not overlap the substitute acknowledgements even when R + W exceeds N. State clearly whether a request uses strict or sloppy placement, and how stale reads and conflicts are exposed.
6. ⚖️ Consistency
Quorum
With a fixed replica set of N=3, a successful write acknowledged by W=2 replicas and a read consulting R=2 replicas must overlap at least one replica. This set-intersection fact helps a read discover an acknowledged version when the membership is stable and acknowledgements have the required durability. It does not by itself resolve concurrent writes, incomplete writes, sloppy-quorum placement, or guarantee linearizability.
For example, write version v2 to A and B, then read B and C: B can return v2 while C still has v1, so the coordinator must compare versions instead of returning the fastest response. Raising R or W increases dependency on replica response time and reduces availability during failures; choose a user-visible contract before choosing the numbers.
// Tunable: R=2, W=2 for N=3
// Read: 2 of 3 nodes; Write: 2 of 3 ack
// Overlap guaranteed: 1 node has latestQuorum write path (W=2, N=3)
Read the diagram definition
sequenceDiagram
autonumber
participant C as Client
participant CO as Coordinator
participant A as Replica A
participant B as Replica B
participant D as Replica D
C->>CO: put(key, value) W=2 N=3
par write fan-out
CO->>A: replicate
CO->>B: replicate
CO->>D: replicate
end
A-->>CO: ack
B-->>CO: ack
Note over CO: 2 of 3 acks -> success
CO-->>C: OK
alt replica down
CO->>D: hinted handoff to alternate node
endQuorum read path (R=2, N=3)
Read the diagram definition
sequenceDiagram
autonumber
participant C as Client
participant CO as Coordinator
participant R1 as Replica 1
participant R2 as Replica 2
C->>CO: get(key) R=2 N=3
CO->>R1: read
CO->>R2: read
R1-->>CO: value + version
R2-->>CO: value + version
CO->>CO: merge by vector clock
CO-->>C: latest value
opt versions diverge
CO->>CO: read repair in background
end7. 🔀 Conflict Resolution
Vector Clocks
A vector clock records how many updates from each writer are included in a version. If every counter in one vector is no larger than the corresponding counter in another, and at least one is smaller, the latter causally follows the former. If neither vector dominates, neither update includes the other and the versions are concurrent. Return concurrent values with their context or use a type-specific merge; replacing them by whichever response arrives last loses information.
VC(a)=2, VC(b)=1
VC(a) < VC(b) if all a[i] <= b[i] and exists j: a[j] < b[j]
Concurrent: neither < nor >Last-Writer-Wins
Timestamp-based. Simple but loses concurrent updates. Use when conflicts rare.
Application-Level
A cart merge can preserve independently added items by taking a union, but that rule can also resurrect an item that another device removed. Quantity changes are harder: summing two copies of an absolute quantity may double-count the same original items. Define additions and removals as operations with identifiers, choose a suitable conflict-free data type, or ask the application to reconcile explicit sibling versions. The desired business meaning determines the merge algorithm.
Dynamo paper: shopping cart uses union merge, never lose items. Conflict-free replicated data types (CRDTs) for automatic merge.
8. 🛡️ Failure Handling
Gossip
Peer-to-peer state exchange. Failure detection: phi accrual (probabilistic). Membership: nodes join/leave.
Hinted Handoff
If node A down, write to next node B. B stores hint "for A". When A recovers, B transfers data to A.
Anti-Entropy
Merkle trees: compare hash trees between replicas. Only transfer differing key ranges. Background sync.
Architecture and Node Responsibilities
Client API and request coordination
Consistent-hash routing and replica selection
Commit log, memtable, SSTables, and compaction
Replication, quorum reads, and quorum writes
Failure detection with gossip membership
Read repair, hinted handoff, and anti-entropy
9. ✍️ Write Path
The coordinator resolves the key’s replica set and sends the versioned write to the intended replicas, typically all N even though it waits for only W acknowledgements. Each replica appends to its write-ahead log and updates its in-memory sorted table; exactly when it acknowledges depends on whether the durability policy requires a disk flush. A coordinator need not be one of the storage replicas.
When W replicas acknowledge, return success with the new version context. A timeout means the result may be ambiguous: some replicas may already have persisted the update. Give retries an operation identifier or suitable version precondition. Full memtables later become immutable sorted files, and background compaction merges those files without changing the acknowledged logical operation.
// Commit log: durable, append-only
// Memtable: skip list or B-tree
// Memtable flush: non-blocking, new memtable created10. 📖 Read Path
The coordinator asks replicas for the key and waits for the required R responses. Each replica checks in-memory state and relevant sorted files, using per-file Bloom filters and indexes to avoid unnecessary disk reads. The coordinator compares returned version contexts, removes versions causally superseded by another, and retains concurrent siblings for the defined merge policy.
Return both the value and context so the client’s next write identifies which versions it has observed. Read repair can bring stale replicas up to date, but a repair of one read key does not repair cold keys elsewhere in the database. Keep background anti-entropy to cover data that users rarely read.
A Bloom filter can establish that a key is absent from a particular sorted file, or say that it may be present and requires an index/data lookup. A negative result is safe only when the filter was correctly built and kept intact. Do not universally stop at the first file containing the key: overlapping tables, tombstones, concurrent versions, or snapshot reads may require comparing several records under the storage engine’s version rules.
11. 📁 SSTable & LSM Tree
SSTable
Sorted String Table: immutable, sorted by key. Index at end for offset lookup. Sequential reads efficient.
Compaction
Size-tiered: merge small SSTables into larger. Leveled: L0→L1→L2…; each level 10× size. Reduces read amplification.
Bloom Filters
One per SSTable. Check before disk read. 1% false positive rate typical. Avoids unnecessary disk I/O for non-existent keys.
12. 🌳 Merkle Trees
A Merkle tree summarizes a consistent partition of key ranges. Leaf hashes represent the keys and versions in each range, and parent hashes summarize their children. Replicas compare roots; if they differ, they recursively compare child hashes until they isolate the differing ranges, then exchange actual records for repair. The tree identifies disagreement but does not choose the correct value: versions, tombstones, and conflict rules still determine the repair result.
Tree depth O(log N) for N key ranges. Efficient for large datasets.
// Anti-entropy: replica A has Merkle tree
// Replica B has Merkle tree
// Compare roots → if diff, compare children
// Recursively find differing key ranges
// Sync only those ranges13. 📡 Gossip Protocol
State Exchange
Each node picks random peer, exchanges state. Eventually consistent: all nodes converge. O(log N) rounds to propagate.
Exchanged: membership (live/dead), partition metadata, load stats. No central coordinator.
Phi Accrual
Failure detector: phi value based on arrival intervals. High phi = likely dead. Adaptive: no fixed timeout.
A phi-accrual detector converts heartbeat delay into a suspicion score based on the observed arrival distribution. Higher thresholds tolerate longer unusual delays before triggering suspicion. Values such as 8 or 12 are example tuning choices, not universal boundaries between a live and dead machine. A long pause and a network partition can both resemble failure; use suspicion to drive a controlled recovery protocol instead of treating it as proof that an old writer has stopped.
Membership: Nodes join via seed; seed gossips new node. Leave: explicit or timeout. Split-brain: use consistent hashing to route; eventual merge.
Handling Data-Center Outage
Replica placement
Do not place all replicas for a key in the same rack or data center. A robust preference list spreads copies across failure domains when latency and cost allow it.
Regional failover
If a data center is offline, route clients to surviving regions, accept AP writes if configured, and reconcile divergent versions when connectivity returns.
Cross-data-center replication improves availability but raises conflict probability and write latency. Call out whether the product prefers local writes with later repair or cross-region quorum for stronger guarantees.
14. ⚠️ Bottlenecks
Compaction
Background I/O can compete with reads/writes. Mitigation: tiered compaction, throttle, separate SSDs.
Vector clock growth
Clocks grow with many nodes. Mitigation: truncate, dotted version vectors, server-side versioning.
Hinted handoff backlog
Node down long = many hints. Mitigation: limit hint size, timeout, transfer priority.
Hot partition
Virtual nodes improve the balance of many keys across machines, but cannot split the write traffic for one indivisible hot key. Cache or replicate suitable reads, apply admission limits, or redesign the value into independently updateable subkeys when semantics permit. Splitting a partition helps a hot range containing many keys; it does not automatically help a single cart or counter that all requests update.
Short takeaway
- AP system: Availability + partition tolerance over strong consistency.
- Partitioning: Consistent hashing with virtual nodes.
- Replication: N nodes, preference list, sloppy quorum (R, W).
- Consistency: Consistency: R + W > N gives quorum overlap under fixed placement; version rules and the protocol determine the actual guarantee.
- Conflict resolution: Vector clocks, last-writer-wins, or app-level merge.
- Failure: Gossip, hinted handoff, Merkle trees for anti-entropy.
- Data-center outage: Cross-region replicas and failover with explicit consistency trade-offs.
- Storage: LSM tree: commit log → memtable → SSTable. Bloom filters.
- Scale: 1PB data, 1000 nodes, 1M ops/sec with tunable R/W.
Beginner to Advanced Design Path(open guidance)
Beginner answer
Explain get, put, delete, partitioning by key hash, replication factor, and a coordinator that reads or writes replicas.
Advanced answer
Discuss quorum settings, vector clocks, hinted handoff, read repair, Merkle-tree anti-entropy, tombstones, LSM compaction, and conflict behavior during partitions.
Deeper walkthrough
Key-value store deeper walkthrough
A distributed key-value store is an availability, partitioning, replication, and conflict-resolution problem. The central question is what happens during failures.
Requirement refinements
Clarify whether the store prefers AP availability or CP consistency, and what stale reads mean for the product.
Define supported operations, value size, TTL, conditional writes, batch operations, and multi-region expectations.
API design details
Return version context with reads so clients can pass it back on writes when conflict detection is required.
Expose consistency options such as quorum read/write or strong read where the product supports them.
Data model and storage
Use consistent hashing with virtual nodes for partitioning and a preference list for replica placement.
Store writes in commit log plus memtable, flush to SSTables, and use tombstones for deletes.
Scaling strategy
Balance nodes with virtual nodes and rebalance gradually so node joins do not move the entire keyspace.
Use Bloom filters, compaction, and read repair to keep read amplification and replica divergence bounded.
Reliability, recovery, and operations
Handle node failure with hinted handoff, sloppy quorum, read repair, and anti-entropy using Merkle trees.
Monitor quorum failure rate, replica divergence, hinted handoff backlog, compaction pressure, and read/write p99.
Deepening reference
Distributed Key-Value Store: realistic production notes
Beginner mental model: A quorum-based store is partitioning, replication, quorum, anti-entropy, and conflict resolution.
Production scope: The main product promise is always-available reads/writes with tunable consistency. The hard part is what happens during network partitions and node failures.
Choose what the first version must guarantee
Start by deciding which distributed key-value store capabilities are essential. Describe the main read and write operations before choosing infrastructure. For each operation, identify the stored records it uses and the condition that makes its result correct. A fast response is useful only if it satisfies that condition.
Next, separate work the user must wait for from work that can finish later. Estimate requests per second, storage growth, and the busiest object or customer. Use those estimates to choose one bottleneck to investigate. The calculations below are example workloads with stated assumptions; they are not measured limits of a particular product.
Concrete capacity calculations
Data: 10B keys with 1 KB average value is 10 TB raw value data. Add 20 percent metadata and LSM overhead, then 3x replication gives roughly 36 TB before compaction headroom.
Throughput: 1M writes/sec with replication factor 3 creates 3M replica writes/sec. If each storage node handles 20K replica writes/sec, use at least 150 nodes before headroom.
Quorum: with N=3, R=2, W=2, a normal read contacts 2 replicas and a write waits for 2 acks. Tail latency is shaped by the slower replica in each quorum.
Partitioning: 10B keys over 1,000 virtual-node ranges gives about 10M keys/range on average; virtual nodes smooth imbalance and simplify node replacement.
Deletes: tombstones must live longer than maximum replica outage plus repair interval; deleting them early can resurrect old values.
Scaling to millions of users
Use consistent hashing with virtual nodes to distribute key ranges and limit movement during scale-out.
Write to a commit log before memory tables so acknowledged writes survive process crashes.
Use quorum settings per use case: lower latency reads for cache-like data, stronger quorum for user-visible state.
Repair divergence with read repair, hinted handoff, and periodic anti-entropy checks.
Expose conflict semantics clearly: last-write-wins for replaceable values, version siblings or merge functions for concurrent updates.
Sequence and data-flow diagram
Walk this path after the high-level design so it is clear what is synchronous, what is asynchronous, and where the system can degrade safely.
Read the diagram definition
sequenceDiagram
participant Client
participant Coord as Coordinator
participant R1 as Replica 1
participant R2 as Replica 2
participant R3 as Replica 3
Client->>Coord: PUT key,value
Coord->>R1: Write
Coord->>R2: Write
Coord->>R3: Write
R1-->>Coord: Ack
R2-->>Coord: Ack
Coord-->>Client: Success after W acks
R3-->>Coord: Late ack or hinted repairCore flow to walk through
Write: coordinator picks replicas, writes commit log/memtable, waits for W acknowledgements.
Read: coordinator reads from R replicas, reconciles versions, returns latest or siblings.
Repair: hinted handoff, read repair, and Merkle-tree anti-entropy converge replicas later.
Production readiness
Distributed Key-Value Store: ship, operate, and evolve
Now examine the same request flow during failures and deployments. Tie each operational control to the user behavior it protects.
Observability
Track distributed key-value store p95/p99 latency, error rate, saturation, and user-visible freshness separately for read, write, and async paths.
Alert on leading indicators such as queue age, hot-key skew, cache miss storms, reconciliation mismatches, and error-budget burn.
Security and abuse
Enforce authentication, authorization, rate limits, input validation, audit trails, and privacy checks at the service boundary.
Treat abuse controls as a first-class path: throttling, quota isolation, moderation hooks, revocation, and safe handling of sensitive data.
Degraded modes
Name what still works if cache, search, ranking, stream processing, object storage, or a regional dependency is slow.
Prefer stale-but-safe reads, partial results, async retry, admission control, and feature disablement over corrupting core state.
Testing and rollout
Load test average, peak, and hot-object traffic; add failure-injection for retries, timeouts, backlog growth, and dependency outages.
Ship with canaries, shadow reads or dual writes when useful, rollback switches, schema compatibility checks, and replay validation.
Runbooks
Document dashboards, paging thresholds, owner actions, rollback steps, data repair commands, and customer-impact language.
Keep backfills and replays idempotent, resumable, rate-limited, and observable so repair work does not create a second incident.
10x evolution
Say which metric forces the next architecture change: shard count, cache footprint, write amplification, queue lag, p99 latency, or regional failover reserve.
Evolve one bottleneck at a time: partition by access pattern, isolate hot tenants, regionalize reads, add replayable async pipelines, then harden multi-region behavior.
Hard parts worth deepening
R + W > N improves latest-read probability but does not remove all conflict cases.
Vector clocks detect concurrent writes; application may need to merge siblings.
Consistent hashing with virtual nodes balances data and limits movement.
Tombstones are required for deletes in eventually consistent stores.
Failure modes and recovery
Replica is down during write; hinted handoff stores temporary copy elsewhere.
Network partition creates conflicting writes; return siblings or resolve with policy.
Tombstone deleted too early; deleted value can reappear from stale replica.
Hot partition forms; split keyspace or add write sharding for hot keys.
Signals to monitor
Measure read and write p99 separately because their replica and disk work differ. Quorum failure rate records operations that could not gather enough required replies within their deadline.
Replica divergence indicates copies disagree, while the hinted-handoff backlog measures temporary writes waiting to reach their intended replicas. Compaction pressure measures background merging work competing with foreground reads and writes for disk resources.
| Choice | Best when | Cost |
|---|---|---|
| AP design | Availability during partition matters most. | Conflicts and stale reads are possible. |
| CP design | Strong consistency matters most. | Some requests fail during partition. |
| Last-writer-wins | Conflicts are rare or data is replaceable. | Can lose concurrent updates. |
Try this scenario
Explain what the client sees when two regions accept different writes for the same key during a network partition.
What must never become false
For the configured consistency mode, a successful quorum write must remain visible to a subsequent quorum read that overlaps that write set under normal replica membership, or the product must explicitly accept stale reads and define merge semantics for concurrent versions. Quorum overlap alone does not guarantee linearizability; say the user-visible contract out loud.
Failure timeline: partition + concurrent puts
T+0 N=3, R=2, W=2; network splits A|B from C T+1m Client-1 put(k,v1) acks via A,B (W met); Client-2 put(k,v2) via C + hinted peer T+5m Partition heals; reads return siblings: vector clocks show concurrent versions T+10m App without merge policy last-writes-wins silently drops v1 or v2 T+1h Anti-entropy Merkle repair syncs bytes; business meaning already wrong Outcome AP store stayed available; invariant fails if product assumed single-value linearizability
Counterexample to "R+W>N means always consistent": during partitions and concurrent writers you still need versioning and a merge story (Dynamo-style), or a CP design (Spanner-style).
Exercises
- 1M puts/sec with N=3, W=2. Roughly how many replica write messages/sec? If one AZ dies, how does sloppy quorum change what you must repair later?
- Product needs shopping-cart merges vs bank-balance single value. Pick Dynamo-style vector clocks vs Spanner-style strong reads: defend the invariant.
- Hinted handoff backlog grows for 6 hours. Which metrics page first, and what read-repair behavior still keeps R=2 useful?
Primary references
Distributed KV primary papers: Dynamo, Bigtable, Spanner.
Practice and further reading
Use these exercises to connect the lesson to a design decision. State your assumptions, trace what happens to one request, and explain the rule that must remain true even during a failure. That rule is the design's correctness invariant.
Focus for this lesson: Design partitioning, replication, quorum, versions, tombstones, anti-entropy, compaction, membership, read repair, and tail-latency policy.
Explain the decision
Why can tombstone garbage collection resurrect deleted values?
Read the reasoning
If an offline replica misses the tombstone and it is collected too early, anti-entropy can reintroduce the older value.
Apply it to a production scenario
A replica is offline longer than the tombstone grace period. How does it rejoin?
Read one approach
Require full repair/bootstrap from a current replica before serving, rather than trusting its stale local data.
Work through the numbers
With N=3, R=2, W=2 and one replica down, can reads and writes continue?
Read the worked calculation
Yes, two replicas satisfy both, assuming reachable nodes and policy; latency is the second-fastest response.
Reason through a failure
What must version reconciliation preserve?
Read the recovery reasoning
Concurrent values are detected rather than overwritten by arrival order, and application merge creates a causally newer version.
Sources and further study
Continue with Design Food Delivery.