P05 — Distributed Key-Value Store

Run it first. There is a companion page that builds this project's machinery as numbered, independently runnable blocks and then assembles them into one measured system: P05 hands-on — block by block (handson/h*.py). Every number on it was produced by running the code. Read it alongside the milestones below.

Large · 143 hours · Weeks 55–67 · Stage 3 · Go

The hardest project in the journey, and the one most likely to overrun. The fault injector is scheduled first, before any distributed feature, specifically to bound it.

Table of Contents

• The Loop, Instantiated
• Why This Project Matters
• Prerequisites
• Duration and Size
• Central Technical Questions
• Architecture
• Showcase — Do This Before You Start
• Implementation Milestones
• Concepts To Study
• Primary-Source Readings
• Experiments
• Benchmarks and Metrics
• Correctness Tests
• Failure Tests
• Expected Difficulties
• Scope Boundaries
• Deliverables
• Exit Criteria
• Extension Ideas
• Connections
• References

The Loop, Instantiated

Why This Project Matters

Almost everyone learns distributed systems as a vocabulary: quorum, consensus, CAP, eventual consistency. The vocabulary is not the skill. The skill is holding in your head the fact that you cannot distinguish a slow node from a dead one, and reasoning correctly about a system where every decision must be made without that information.

That single asymmetry generates almost everything else: why consensus needs a majority, why leases need clocks you distrust, why exactly-once delivery is impossible and exactly-once processing is not, why a failure detector's timeout is a liveness/availability trade with no correct answer.

You operate distributed systems today. This project is where you stop operating them and start being able to say what they are guaranteeing.

And the meta-lesson, which is why this is the hardest project: distributed bugs do not reproduce. The only defence is a deterministic harness that can replay a failing schedule. Building that harness before the system is the single most transferable habit in this track — it is the same reason P03 built the crash tester at milestone 2.

Prerequisites

• P04 complete — the storage engine is the per-node state
• Go: goroutines, channels, select, context, the race detector
• Comfort with the idea that a test that passes 1,000 times can still be wrong

Duration and Size

Large, 143 hours, 13 weeks.

Central Technical Questions

1. Why a majority? Not "because the paper says so" — derive why any two quorums must intersect and what breaks if they do not.
2. What does a failure detector actually detect? Nothing about the remote node. It reports a property of your observations. Say precisely what.
3. What is linearizability, and how would you check a history for it?
4. Where does your system lose linearizability first? Every implementation has a weakest point. Name yours before the checker finds it.
5. What happens during a leader change to in-flight writes? Trace one specific client's request across an election.
6. Why is exactly-once delivery impossible and exactly-once effect achievable?
7. What does the system do when the network is slow but nothing has failed? This is the common case in production and the one designs handle worst.

Architecture

Write your naive design first. Include the split-brain you predict.

   client ──► router (any node) ──► shard owner
                                        │
      shard = hash(key) mod ring        │  Raft group per shard, N=3
                                        ▼
                        ┌──────── leader ────────┐
                        │  append to local log   │
                        │  fsync                 │
                        │  AppendEntries ──► followers (parallel)
                        │  wait for majority ack │
                        │  commitIndex++         │
                        │  apply to P04 engine   │
                        │  reply to client       │
                        └────────────────────────┘

  ── fault injector sits BETWEEN every pair of nodes ──
     drop · delay · duplicate · reorder · partition · pause process · corrupt disk

Why a majority — derived

A quorum system needs any two quorums to intersect, so that a decision made by one is visible to the next. With \(N\) replicas and quorum size \(Q\), two quorums of size \(Q\) intersect iff \(2Q > N\), i.e. \(Q \ge \lfloor N/2 \rfloor + 1\).

If \(2Q \le N\), two disjoint quorums can each make a decision without ever seeing the other's — that is split-brain, and it is not a bug in an implementation, it is an arithmetic consequence of the quorum size. Every real split-brain is this inequality being violated somewhere, often by an operator changing the replica count.

What majority replication buys, computed for \(N = 3\), independent node failure probability \(p\):

At \(p=0.01\), majority replication turns 3.65 days of downtime a year into 2.6 hours. Note the assumption doing all the work: independence. Correlated failures — same rack, same power domain, same bad deploy, same poisoned request — collapse this entirely, and correlated failure is the normal case. Say that in your report.

The core asymmetry

You send a heartbeat. No reply arrives within \(T\). Three worlds are consistent with that observation:

1. The node crashed.
2. The node is alive and the network dropped the messages.
3. The node is alive and just slow — GC pause, disk stall, CPU starvation.

You cannot distinguish them, ever, in an asynchronous network. This is the content of the FLP impossibility result: no deterministic consensus algorithm can guarantee termination in an asynchronous system with even one crash failure. Practical systems escape by adding a timing assumption — a failure detector that is allowed to be wrong.

So the timeout \(T\) is a trade with no correct value:

• \(T\) too small: false positives. Spurious elections, unnecessary failovers, and under load a feedback loop where slowness causes elections which cause more slowness.
• \(T\) too large: real failures go undetected. Availability gap equal to \(T\) on every genuine crash.

Phi-accrual detection (Hayashibara et al. 2004) replaces the binary with a suspicion level derived from the observed inter-arrival distribution, letting the application choose its threshold. Implementing it and measuring the false-positive/detection-time frontier is E4, and it is the best experiment in this project.

Build the fault injector first

Milestones 1–2, before any distributed feature. It must support:

And it must be deterministic: seeded, with a recorded schedule that can be replayed exactly. A distributed bug you cannot replay is a distributed bug you cannot fix.

Showcase — Do This Before You Start

W3 · walkthroughs/w3_raft.py · ~60 minutes

A working miniature of this project: split-brain caused by a plurality rule and prevented by a majority one, plus the inequality brute-forced over every (N, Q).

cd walkthroughs && python3 w3_raft.py

It is 80-ish lines and it surfaces this project's central surprise in an evening rather than in week six. Run it before committing the weeks.

Implementation Milestones

Concepts To Study

• The asynchronous model and why it is the right default
• FLP impossibility: no deterministic async consensus with one crash failure
• CAP, stated precisely: during a partition, choose between linearizability and availability. Note that CAP says nothing about the non-partitioned case, which is where PACELC comes in
• Linearizability vs sequential consistency vs serializability — three different properties routinely conflated
• Quorum intersection and the derivation above
• Raft: terms, elections, log matching, leader completeness, state-machine safety
• Why Raft's commit rule excludes prior-term entries — the subtlest safety point in the paper (§5.4.2). Understand it or your implementation will have a rare, real bug
• Failure detectors: completeness and accuracy; phi-accrual
• Idempotency: client sessions, dedup tables, and their unbounded growth problem
• Leases and clocks: why a lease needs a bounded clock drift assumption
• Consistent hashing: virtual nodes and why naive consistent hashing has bad balance
• Split-brain and fencing tokens
• Read paths: read-from-leader, ReadIndex, lease reads — three different correctness/latency points

Primary-Source Readings

Budget: 20 hours, the largest in the journey.

Experiments

E6 with asymmetric partitions is the one that finds bugs. Symmetric partitions are easy to reason about. A node that can send but not receive causes a leader that believes it is still leading, and it breaks designs that symmetric tests pass.

Benchmarks and Metrics

Correctness Tests

1. Linearizability checking on every test run. Record a history of invoke/return/value events, check offline. Continuously, not once.
2. Election safety: at most one leader per term. Assert globally in the simulator.
3. Log Matching: if two logs contain an entry with the same index and term, all preceding entries are identical. Assert after every AppendEntries.
4. Leader Completeness: a committed entry is present in the log of every future leader. This is the property §5.4.2 protects.
5. State Machine Safety: no two nodes apply different commands at the same index.
6. Durability: an acknowledged write survives crash-restart of a majority.
7. Idempotency: replaying a client request produces one effect.
8. Membership safety: no configuration change creates two disjoint majorities.
9. Determinism: same seed → identical execution. Without this, nothing else is testable.
10. Model-based test: whole cluster vs a single map, under faults.

Failure Tests

Every one of these runs in CI, seeded, replayable.

Expected Difficulties

1. This project will take longer than you plan. It is the one with the highest overrun risk in the journey. Mitigations, in order: the deterministic simulator (milestone 1) so bugs replay; the linearizability checker (milestone 5) so bugs are found rather than shipped; a hard scope cut at week 11 that drops milestones 11–13 and ships single-shard.
2. Raft §5.4.2 will bite you. A leader may not commit an entry from a previous term by counting replicas — it must commit an entry from its own term first. The bug this prevents appears in perhaps one in 10⁵ runs and destroys linearizability. Write the test that constructs the interleaving deliberately.
3. Real-network testing wastes weeks. Use the in-process simulator for everything. Run on real sockets once, at the end, to confirm nothing depended on the simulation.
4. The linearizability checker is exponential in the worst case. Wing–Gong with pruning is fine for histories of a few thousand ops; keep them short.
5. Debugging output is your main tool and will drown you. Structured logs with node id, term, index, and a monotonic sequence number; a script that renders a history as a space-time diagram. Build it in milestone 3.
6. Go's race detector will find real bugs. Run every test under -race from day one.

Scope Boundaries

In scope: crash-stop failures, asynchronous network, single-key linearizable operations, leader-based replication, static then dynamic membership, snapshots, rebalancing.

Out of scope: Byzantine faults; multi-key transactions (extension only); geo-replication and clock-bounded designs à la Spanner (read the paper, do not build it); a production RPC framework — hand-rolled is better here; an admin UI; TLS and authentication; performance work beyond what the experiments require.

Deliverables

1. distkv/ — Go module, in-process simulator and a real-socket mode
2. faultinjector/ — the standalone, reusable artifact. This is the most portfolio-valuable thing you will build in Stage 3
3. linchecker/ — the linearizability checker, also standalone
4. REPORT.md including at least one real bug the checker found in your own code, with the interleaving diagram
5. Notebook entries for E4, E6, E12
6. A space-time diagram renderer for recorded histories

Exit Criteria

• Linearizability checker reports zero violations across ≥1,000 seeded fault runs
• All thirteen failure tests pass, including the asymmetric partition and the SIGSTOP-resume
• Leader election works under partition with exactly one leader per term, asserted
• E5 measured: availability gap on leader failure, compared against your prediction
• E4 measured: phi-accrual vs fixed timeout, frontier plotted
• Snapshots work: a follower behind by a compacted prefix catches up
• At least one real bug found by the checker, documented with its interleaving
• Same-seed determinism verified — every failing run replays
• REPORT.md written with a falsified prediction

Extension Ideas

• Read leases: local reads at the cost of a bounded-clock assumption. Measure the latency win and state the assumption you bought it with.
• Jepsen-style external testing against your own system.
• Two-phase commit across shards — and a measurement of how much availability it costs, which is the honest reason distributed transactions are avoided.
• Compare with an AP design: implement Dynamo-style quorum replication with vector clocks alongside, and measure the availability difference during partition. This is CAP as an experiment rather than a slogan, and it is the strongest extension here.

Connections

Backward: P04 is the per-node engine; its WAL and snapshots are reused. P03's crash harness generalises into the fault injector.

Forward:

• → P06 (MapReduce): the fault injector, membership, and failure detector are reused directly. The coordinator's fault tolerance is a simpler version of this problem
• → P07 (Streaming): the replicated log is the event log; checkpointing is snapshotting
• → P12 (Kernel): scheduling, timers, and the cost of a context switch are the micro-scale version of the same latency questions
• → P15: the storage and ingestion substrate

References

• Ongaro, D., Ousterhout, J. In Search of an Understandable Consensus Algorithm. USENIX ATC 2014 (extended version, Stanford).
• Lamport, L. Time, Clocks, and the Ordering of Events in a Distributed System. CACM 21(7), 1978.
• Lamport, L. The Part-Time Parliament. ACM TOCS 16(2), 1998. And Paxos Made Simple, 2001.
• Fischer, M. J., Lynch, N. A., Paterson, M. S. Impossibility of Distributed Consensus with One Faulty Process. JACM 32(2), 1985.
• Herlihy, M. P., Wing, J. M. Linearizability: A Correctness Condition for Concurrent Objects. ACM TOPLAS 12(3), 1990.
• Gilbert, S., Lynch, N. Brewer's Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services. SIGACT News 33(2), 2002.
• Abadi, D. Consistency Tradeoffs in Modern Distributed Database System Design. IEEE Computer 45(2), 2012. PACELC.
• DeCandia, G. et al. Dynamo: Amazon's Highly Available Key-value Store. SOSP 2007.
• Corbett, J. C. et al. Spanner: Google's Globally-Distributed Database. OSDI 2012.
• Hayashibara, N., Défago, X., Yared, R., Katayama, T. The φ Accrual Failure Detector. SRDS 2004.
• Chandra, T. D., Toueg, S. Unreliable Failure Detectors for Reliable Distributed Systems. JACM 43(2), 1996.
• Karger, D. et al. Consistent Hashing and Random Trees. STOC 1997.
• Kingsbury, K. Jepsen. jepsen.io — the analyses, not just the tool.
• Alvaro, P., Rosen, J., Hellerstein, J. M. Lineage-driven Fault Injection. SIGMOD 2015. A smarter way to choose which faults to inject.