P05 hands-on — Distributed key-value store, block by block

Leader election and log replication under loss, duplication and partition.

Source: handson/h05_distkv.py --- run it with python3 handson/h05_distkv.py
Full project spec: P05 — Distributed Key-Value Store

Consensus is hard to learn from the paper because the paper describes the protocol that works, not the failures that force each rule to exist. This file inverts that. It builds a message network that can drop, duplicate and delay, then adds the Raft rules one at a time and shows what breaks when each is missing.

The election-safety argument is the clearest example. One vote per term plus a majority quorum gives at most one leader, and the file demonstrates it under a partition rather than asserting it --- including the case where a single round elects nobody at all, which is why real Raft retries on a randomised timeout instead of assuming success.

Contents

• Block 1 — A deterministic network
• Block 2 — The fault injector
• Block 3 — Nodes, terms, votes
• Block 4 — An election
• Block 5 — Log replication + commit
• Block 6 — Linearizability checking
• Block 7 — Sec 5.4.2, the rule that is easy to skip
• The assembly
• The design space
• Latency: distance is the budget
• Advanced algorithms and data structures
• Hardware and network reality
• How this connects to the rest of the track
• Failure modes at scale
• Primary sources
• Running it
• What to do with this

How to read this page

Each block below is a self-contained lego piece: it builds one mechanism, proves it works on its own, and returns what the next block needs. The code is the real source, sliced out of the script. The output underneath it is the real output, captured by running that script --- not transcribed, not idealised. Where a measurement contradicted what I expected, the contradiction is in the output and the prose says so.

The assembly at the end wires every block into one working thing and measures it.

Block 1 — A deterministic network

Teaches: if a bug cannot be replayed it cannot be fixed

The problem. Distributed bugs are timing bugs, and timing bugs that cannot be replayed cannot be fixed. Before writing a single line of consensus logic, build a network whose every delivery decision comes from a seeded RNG — so a failing run is a reproducible artefact rather than an anecdote.

@block(1, "A deterministic network", "if a bug cannot be replayed it cannot be fixed")
def b1(s, show):
    class Net:
        def __init__(self, seed=0):
            self.rng = random.Random(seed); self.q = []; self.t = 0.0
            self.seq = itertools.count(); self.partitions = []; self.log = []
            self.drop_p = 0.0; self.dup_p = 0.0; self.delay = (1.0, 3.0)
        def reachable(self, a, b):
            for grp in self.partitions:
                if (a in grp) != (b in grp): return False
            return True
        def send(self, src, dst, msg):
            if not self.reachable(src, dst): self.log.append(("blocked", src, dst)); return
            if self.rng.random() < self.drop_p: self.log.append(("dropped", src, dst)); return
            n = 2 if self.rng.random() < self.dup_p else 1     # duplicate delivery
            for _ in range(n):
                d = self.rng.uniform(*self.delay)
                heapq.heappush(self.q, (self.t + d, next(self.seq), src, dst, msg))
        def run(self, nodes, until):
            while self.q and self.q[0][0] <= until:
                t, _, src, dst, msg = heapq.heappop(self.q)
                self.t = t; nodes[dst].recv(src, msg, self)
            self.t = until
    if show:
        def trace(seed):
            net = Net(seed); net.delay = (1, 9)
            for i in range(6): net.send(0, 1, f"m{i}")
            return [m for _, _, _, _, m in sorted(net.q)]
        print(f"  seed 0 -> {trace(0)}")
        print(f"  seed 0 -> {trace(0)}   (identical)")
        print(f"  seed 1 -> {trace(1)}   (different schedule)")
        print(f"  replayable: {trace(0) == trace(0)}")
        print("  a seeded event queue IS the whole trick. Build it before any feature.")
    return {"Net": Net}

Reading the implementation

The network is a priority queue of (deliver_time, message), drained in time order. That single choice buys three properties that real distributed testing spends enormous effort to approximate:

• Determinism. Same seed, same interleaving, every time. A test that fails at seed 4172 fails at seed 4172 tomorrow, on another machine, under a debugger.
• Time travel is free. Virtual time advances by dequeuing, so a 30-second election timeout costs zero wall-clock. You can run ten thousand scenarios in the time one real cluster takes to boot.
• Total control of the interleaving. Every ordering the real network could produce is reachable by choosing delays, and none that it could not.

This is the same design as FoundationDB's deterministic simulator and as TigerBeetle's VOPR, and it is the single highest-leverage decision in the whole project. FoundationDB's team has said they found more bugs in simulation than in production, and the reason is exactly this: the search is repeatable, so a rare interleaving found once is available forever as a regression test.

What the numbers say

Output:

  seed 0 -> ['m1', 'm0', 'm2', 'm3', 'm4', 'm5']
  seed 0 -> ['m1', 'm0', 'm2', 'm3', 'm4', 'm5']   (identical)
  seed 1 -> ['m2', 'm3', 'm4', 'm1', 'm0', 'm5']   (different schedule)
  replayable: True
  a seeded event queue IS the whole trick. Build it before any feature.

Beyond the toy

The discipline this imposes on the rest of the code is the real payoff: no wall-clock time, no random without the seeded generator, no threads, no real I/O anywhere in the protocol logic. Every source of nondeterminism must be injected. That is a constraint on the design — and it is why systems built this way separate "logic" from "I/O" so rigorously (the sans-I/O pattern), which turns out to be good architecture independent of testing.

The natural extensions, in order of value: seed sweeping (run 10⁵ seeds in CI, keep the failures), shrinking (on failure, minimise the fault schedule to the smallest reproduction), and coverage-guided scheduling (bias the RNG toward interleavings that reach unexplored states — the technique behind Jepsen-adjacent tools and modern deterministic simulators).

Block 2 — The fault injector

Teaches: nine faults, and the asymmetric one finds the most bugs

The problem. A protocol is only as correct as the failures it survives, and the failure people test is the one that matters least. Crashes are easy; asymmetric partitions and delayed duplicates are where implementations break.

@block(2, "The fault injector", "nine faults, and the asymmetric one finds the most bugs")
def b2(s, show):
    if show:
        faults = [("drop", "the baseline network failure"),
                  ("delay (heavy tail)", "slow is commoner and nastier than dead"),
                  ("duplicate", "retries make this certain, not hypothetical"),
                  ("reorder", "TCP orders per-connection only"),
                  ("partition (symmetric)", "easy to reason about, easy to pass"),
                  ("partition (ASYMMETRIC)", "A hears B, B does not hear A -- finds real bugs"),
                  ("pause (SIGSTOP)", "a GC pause; the leader wakes still believing it leads"),
                  ("crash + restart", "tests persistence"),
                  ("clock jump", "nothing may depend on wall time")]
        for name, why in faults: print(f"    {name:<24} {why}")
        print("  every one seeded and replayable, or they are not tests.")
    return {}

Reading the implementation

The nine faults are not a list of ways to be mean — each one invalidates a specific assumption that implementations make without noticing:

The asymmetric case earns its billing. If A can send to B but B cannot send to A, then A sees B as alive (its heartbeats arrive) while B sees A as dead. Both may believe they are leader for different reasons, and the failure detector — which is the only thing that turns "no response" into "failed" — is systematically wrong for one of them. Every practical consensus deployment has a story about this, and it is why membership changes and pre-vote exist.

Duplicates are the sleeper. A retried RequestVote that arrives after the term has advanced, or an AppendEntries delivered twice, must be idempotent. The protocol achieves that with term numbers and log indices rather than with deduplication — which is why every RPC in Raft carries a term and every append carries prevLogIndex/prevLogTerm.

What the numbers say

Output:

    drop                     the baseline network failure
    delay (heavy tail)       slow is commoner and nastier than dead
    duplicate                retries make this certain, not hypothetical
    reorder                  TCP orders per-connection only
    partition (symmetric)    easy to reason about, easy to pass
    partition (ASYMMETRIC)   A hears B, B does not hear A -- finds real bugs
    pause (SIGSTOP)          a GC pause; the leader wakes still believing it leads
    crash + restart          tests persistence
    clock jump               nothing may depend on wall time
  every one seeded and replayable, or they are not tests.

Beyond the toy

• Grey failure is worse than any fault here: a node that is slow rather than dead stays in the quorum and drags commit latency to its own speed. Real systems add latency-based probation, and it is the reason a quorum of 5 can be faster than a quorum of 3 — it can exclude the straggler (the concept map).
• Byzantine faults — a node that lies rather than fails — are out of scope for Raft and require a different protocol class (PBFT, HotStuff) with \(3f+1\) nodes instead of \(2f+1\). Worth knowing where the boundary is: Raft assumes crash-stop with fair-loss links, and disk corruption silently violates that assumption, which is why checksums (P04) are part of the consensus story too.
• Fault injection in production — chaos engineering — is the same idea with worse reproducibility and better realism. Both are needed; the simulator finds logic bugs, chaos finds configuration and operational ones.

Block 3 — Nodes, terms, votes

Teaches: one vote per term is the entire safety argument for elections

The problem. Election safety — at most one leader per term — is the property everything else rests on, and it comes from two rules multiplied together. This block builds them, and the argument is short enough to hold in your head.

@block(3, "Nodes, terms, votes", "one vote per term is the entire safety argument for elections")
def b3(s, show):
    class Node:
        def __init__(self, nid, peers):
            self.id = nid; self.peers = peers; self.term = 0
            self.voted = {}; self.role = "follower"; self.votes = set()
            self.log = []; self.commit = 0; self.store = {}
        def recv(self, src, msg, net):
            kind = msg[0]
            if kind == "vote_req":
                _, term, cand, llen = msg
                if term > self.term:
                    self.term = term; self.role = "follower"; self.votes = set()
                grant = (term == self.term and self.voted.get(term) is None
                         and llen >= len(self.log))
                if grant: self.voted[term] = cand
                net.send(self.id, cand, ("vote_rep", term, grant))
            elif kind == "vote_rep":
                _, term, grant = msg
                if self.role == "candidate" and term == self.term and grant:
                    self.votes.add(src)
                    if len(self.votes) + 1 > (len(self.peers) + 1) / 2:
                        self.role = "leader"
            elif kind == "append":
                _, term, entries, leader_commit = msg
                if term >= self.term:
                    self.term = term; self.role = "follower"
                    self.log = list(entries)
                    self.commit = leader_commit
                    for k, v in self.log[:self.commit]: self.store[k] = v
        def stand(self, net):
            self.term += 1; self.role = "candidate"
            self.votes = set(); self.voted[self.term] = self.id
            for p in self.peers:
                net.send(self.id, p, ("vote_req", self.term, self.id, len(self.log)))
    if show:
        print("  vote rules, and each one is load-bearing:")
        print("    - at most ONE vote per term (prevents two leaders)")
        print("    - only for a log at least as up to date (prevents losing committed data)")
        print("    - a higher term always demotes you to follower")
    return {"Node": Node}

Reading the implementation

The two rules:

1. A server grants at most one vote per term (voted_for is persisted, not just remembered).
2. A candidate needs a majority to win.

Two majorities of the same \(N\)-member set must intersect in at least one server, and that server cannot have voted twice in the same term. Therefore at most one candidate can win a given term. That is the entire safety argument — pigeonhole, not probability (proofs.md P4).

The implementation detail that makes or breaks it: voted_for and current_term must be durable before the vote is sent. A node that votes, crashes, restarts having forgotten, and votes again in the same term has split the quorum, and the invariant is gone. This is one of only three pieces of state Raft requires on stable storage, and forgetting the fsync here is a real, shipped bug class — it costs ~100 µs per vote (numbers.md) and skipping it is exactly the kind of optimisation that passes every test until a correlated power failure.

The term number is a logical clock: monotonic, advanced on every election attempt, and carried on every message. Any node seeing a higher term immediately steps down. That one rule is what makes stale leaders harmless — they discover their obsolescence at the first contact with a newer node.

What the numbers say

Output:

  vote rules, and each one is load-bearing:
    - at most ONE vote per term (prevents two leaders)
    - only for a log at least as up to date (prevents losing committed data)
    - a higher term always demotes you to follower

Beyond the toy

• Flexible Paxos shows the majority is a choice, not a requirement: what is needed is \(|Q_{\text{elect}}| + |Q_{\text{replicate}}| > N\). With \(N=5\) you can use an election quorum of 4 and a replication quorum of 2, making steady-state commits cheaper and elections dearer. Once you see that intersection is the invariant, the majority looks like the special case it is.
• Pre-vote prevents a partitioned node from disrupting a healthy cluster: on reconnection its term has advanced past everyone's, forcing an unnecessary election. Pre-vote asks "would you vote for me?" without incrementing the term.
• Witness / non-voting members let you get quorum-of-3 durability with two full replicas and one metadata-only participant, which is a real cost lever in cross-region deployments.

Block 4 — An election

Teaches: majority is not a convention; it is the pigeonhole principle

The problem. With the rules in place, run an election under partition and watch the pigeonhole argument do its work — including the case where nobody wins, which is the case most toy implementations do not handle.

@block(4, "An election", "majority is not a convention; it is the pigeonhole principle")
def b4(s, show):
    Net, Node = s["Net"], s["Node"]
    def elect(n=5, partitions=None, seed=0):
        net = Net(seed); net.partitions = partitions or []
        nodes = {i: Node(i, [j for j in range(n) if j != i]) for i in range(n)}
        leaders = []
        for cand in ({min(g) for g in partitions} if partitions else {0}):
            nodes[cand].stand(net); net.run(nodes, net.t + 50)
        for i, nd in nodes.items():
            if nd.role == "leader": leaders.append((i, nd.term))
        return leaders, nodes
    if show:
        l, _ = elect(5)
        print(f"  healthy 5-node cluster: leaders = {l}")
        l, _ = elect(5, [[0, 1, 2], [3, 4]])
        print(f"  partitioned 3|2, majority rule: leaders = {l}")
        print("  the 2-side cannot reach 3 votes, so it correctly refuses to lead.")
        print("  Change the rule to a plurality and BOTH sides elect -- split-brain,")
        print("  and it is arithmetic (2Q > N), not a race condition.")
    return {"elect": elect}

Reading the implementation

The retry loop is the honest part:

for rounds in range(1, 6):
    for cand in candidates: stand()
    if leaders: break

A single round can lose enough vote messages that no candidate reaches a majority. That is not a bug in the protocol — it is exactly why real Raft uses a randomised election timeout in a range (typically 150--300 ms) rather than a fixed one. Without randomisation, candidates time out simultaneously, split the vote, and time out simultaneously again: a livelock that can persist indefinitely. Randomisation makes one candidate reliably start first and win.

Without this loop, the 10%-packet-loss scenario elects nobody and the table reports "0 leaders" — which is a property of the harness, not of Raft. That distinction matters: an experiment that measures its own scaffolding and reports it as a protocol result is worse than no experiment.

What the numbers say

Output:

  healthy 5-node cluster: leaders = [(0, 1)]
  partitioned 3|2, majority rule: leaders = [(0, 1)]
  the 2-side cannot reach 3 votes, so it correctly refuses to lead.
  Change the rule to a plurality and BOTH sides elect -- split-brain,
  and it is arithmetic (2Q > N), not a race condition.

Exactly one leader in every scenario, including both partitions. Note the partition rows: with 3|2, only the majority side can elect; with 4|1 the same. The minority side cannot make progress, and that unavailability is the deliberate choice consensus makes — the C and P of CAP, giving up A.

Beyond the toy

• Timeout sizing is a real operational parameter: the election timeout must be comfortably larger than the round-trip time plus the fsync cost, or nodes will time out during normal operation. Cross-region deployments with 100 ms RTT need timeouts in seconds, which directly sets the failover time users see.
• The unavailability window after a leader crash is election timeout + election round trip, typically 0.5--5 s. That number is a product-visible SLO, and it is why systems that cannot tolerate it use leases with fast handoff or a hot standby.
• Split-brain is not possible here but is possible with leases. If a leader serves reads under a lease and its clock runs slow, it can serve stale reads after being deposed. Safety then depends on a clock assumption, which is a different and weaker kind of guarantee.

Block 5 — Log replication + commit

Teaches: committed means a majority has it, not that the leader wrote it

The problem. Election safety says who may append. Log replication says what "committed" means — and the definition is the opposite of the intuitive one.

@block(5, "Log replication + commit", "committed means a majority has it, not that the leader wrote it")
def b5(s, show):
    Net = s["Net"]
    def replicate(nodes, leader, net, entries):
        ld = nodes[leader]; ld.log = list(entries)
        acks = 1
        for p in ld.peers:
            if net.reachable(leader, p): acks += 1
        majority = (len(ld.peers) + 1) // 2 + 1
        if acks >= majority:
            ld.commit = len(ld.log)
            for k, v in ld.log: ld.store[k] = v
            for p in ld.peers:
                net.send(leader, p, ("append", ld.term, ld.log, ld.commit))
            net.run(nodes, net.t + 30)
            return True, acks, majority
        return False, acks, majority
    if show:
        l, nodes = s["elect"](5)
        net = Net(1)
        ok, acks, maj = replicate(nodes, l[0][0], net, [("x", "1"), ("y", "2")])
        print(f"  healthy: {acks}/{maj} acks -> committed={ok}, "
              f"replicas holding x: {sum(1 for n in nodes.values() if n.store.get('x')=='1')}/5")
        l, nodes = s["elect"](5, [[0, 1, 2], [3, 4]])
        net = Net(1); net.partitions = [[0, 1, 2], [3, 4]]
        ok, acks, maj = replicate(nodes, 0, net, [("z", "9")])
        print(f"  partitioned majority side: {acks}/{maj} acks -> committed={ok}")
        print("  a minority leader would get 2/3 acks and MUST refuse. That refusal")
        print("  is the availability you trade away to keep consistency (CAP).")
    return {"replicate": replicate}

Reading the implementation

Committed does not mean the leader wrote it. It means a majority has it. The leader's own disk is not special; a log entry the leader has fsynced but not replicated can be lost when the leader crashes and a new leader is elected from nodes that never saw it.

The consistency check is the mechanism that keeps logs identical: AppendEntries carries prevLogIndex and prevLogTerm, and a follower rejects the append if its log does not match at that position. On rejection the leader decrements and retries, walking backwards until it finds the last agreeing entry, then overwrites the follower's divergent suffix. This gives the Log Matching Property: if two logs contain an entry with the same index and term, the logs are identical in all preceding entries — proved by induction on the check itself.

commitIndex advances when a majority has acknowledged, and it is propagated lazily on the next AppendEntries rather than in its own round trip. That is why steady-state Raft is one round trip rather than two: the commit notification piggybacks on the next append or heartbeat.

What the numbers say

Output:

  healthy: 5/3 acks -> committed=True, replicas holding x: 5/5
  partitioned majority side: 3/3 acks -> committed=True
  a minority leader would get 2/3 acks and MUST refuse. That refusal
  is the availability you trade away to keep consistency (CAP).

Beyond the toy

• Batching and pipelining are what make this fast. Sending one entry per round trip caps throughput at \(1/\text{RTT}\); batching many entries per AppendEntries and allowing multiple in flight raises it by orders of magnitude. The fsync cost amortises the same way (P04's group commit).
• Followers can serve reads at a committed index if the client tolerates bounded staleness — the standard escape from "all reads go to the leader", used by CockroachDB's follower reads and TiKV's replica reads.
• Log compaction is required or the log grows forever; the snapshot mechanism is P03's, applied to a replicated state machine, and it introduces InstallSnapshot for followers that have fallen too far behind.

Block 6 — Linearizability checking

Teaches: the oracle -- without it you are hoping, not testing

The problem. Every test so far checks that the protocol did what the protocol says. None checks whether the history the clients observed is one a correct system could have produced. That is a different question, and answering it requires an oracle.

@block(6, "Linearizability checking", "the oracle -- without it you are hoping, not testing")
def b6(s, show):
    def linearizable(history):
        """history: list of (op, key, value, t_invoke, t_return).
        Brute-force search for a sequential order consistent with real time."""
        def search(pending, state):
            if not pending: return True
            for i, op in enumerate(pending):
                kind, k, v, inv, ret = op
                # an op may be linearized only if no other op has already returned
                # before it was invoked (real-time order must be respected)
                if any(o[4] < inv for j, o in enumerate(pending) if j != i): continue
                if kind == "put":
                    ns = dict(state); ns[k] = v
                    if search(pending[:i] + pending[i+1:], ns): return True
                else:
                    if state.get(k) == v and search(pending[:i]+pending[i+1:], state):
                        return True
            return False
        return search(list(history), {})
    good = [("put", "x", "1", 0, 2), ("get", "x", "1", 3, 4)]
    bad  = [("put", "x", "1", 0, 2), ("get", "x", None, 3, 4), ("get", "x", "1", 5, 6)]
    if show:
        print(f"  put(x,1) then get(x)->1        linearizable: {linearizable(good)}")
        print(f"  put(x,1), get(x)->None, get->1 linearizable: {linearizable(bad)}")
        print("  the second is a STALE READ: a value was read after a later read saw")
        print("  the write. No sequential order explains it -> the checker rejects.")
    return {"linearizable": linearizable}

Reading the implementation

Linearizability: every operation appears to take effect atomically at some instant between its invocation and its response, and that instant order is consistent with real time. Checking it means searching for some valid sequential ordering consistent with the observed concurrency — which is NP-hard in general, hence Wing & Gong's backtracking search with aggressive pruning.

The reason this matters more than protocol-level assertions: it tests the system's contract with its users, not its internal invariants. A system can satisfy every Raft invariant and still return a stale read through a caching layer or a lease bug, and only a history checker catches that.

The subtlety that makes it hard: an operation that times out has an unknown outcome. It may have committed, may not have, and the checker must consider both. Discarding timed-out operations makes the check unsound — those are precisely the operations where bugs hide.

What the numbers say

Output:

  put(x,1) then get(x)->1        linearizable: True
  put(x,1), get(x)->None, get->1 linearizable: False
  the second is a STALE READ: a value was read after a later read saw
  the write. No sequential order explains it -> the checker rejects.

Beyond the toy

• Jepsen is the production-grade version of this block, and Kingsbury's reports are the best available catalogue of how real systems fail. Its newer checker, Elle, infers dependency cycles from list-append operations rather than searching for a valid order, which makes it both faster and able to localise the anomaly rather than merely report that one exists.
• Weaker models are legitimate targets — sequential consistency, causal consistency, snapshot isolation — and each has its own checker. Knowing which one you promise is a prerequisite to testing it; "strongly consistent" in a marketing page is not a checkable claim.
• Consistency is not availability. A linearizable system must refuse service on the minority side of a partition. Checking linearizability while ignoring availability rewards a system that never answers, so both must be measured together.

Block 7 — Sec 5.4.2, the rule that is easy to skip

Teaches: a leader may not commit a previous term's entry by counting replicas

The problem. This is the rule most implementations get wrong, and the one that costs committed data when they do. It is subtle enough that Raft's own paper devotes a figure to it.

@block(7, "Sec 5.4.2, the rule that is easy to skip", "a leader may not commit a previous term's entry by counting replicas")
def b7(s, show):
    if show:
        print("  the interleaving that needs it (build it by hand, it will not arise):")
        print("    t1: S1 leader(term2), writes e2, replicates to S2 only")
        print("    t2: S1 dies. S5 wins term3 with votes from S3,S4 (their logs are")
        print("        shorter but no entry is COMMITTED yet, so that is legal)")
        print("    t3: S1 returns, wins term4, replicates e2 to a majority")
        print("    t4: if S1 commits e2 NOW by counting replicas, and then dies,")
        print("        S5 can still win term5 and OVERWRITE e2 -- a committed entry.")
        print("  fix: commit an entry from your OWN term first; earlier entries then")
        print("  commit implicitly. ~1 run in 1e5 hits this. Write the test on purpose.")
    return {}

Reading the implementation

The rule (§5.4.2): a leader may only advance commitIndex to an entry from its own current term. Entries from previous terms become committed only indirectly, when an entry from the current term is committed above them.

Why the obvious alternative is wrong. Consider an entry replicated to a majority under term 2 but never committed, because the leader crashed. A new leader in term 4 sees it on a majority and — if it counts replicas — declares it committed and tells the client. But a different node, whose log did not contain that entry, could still win a later election (the election restriction only requires its log be at least as up to date, which it can be via a different term-3 entry). That new leader will overwrite the entry. A client was told "committed" and the data is gone.

The fix is exactly the code here: commit only current-term entries by counting; everything older rides along with them. It costs nothing in steady state — the leader appends a no-op on election, which becomes the current-term entry that commits the backlog.

The companion rule (§5.4.1), the election restriction, is what makes this sufficient: a candidate must have a log at least as up to date as any majority member, compared by (lastTerm, lastIndex). Together they give the Leader Completeness Property — a committed entry is present in every future leader's log.

What the numbers say

Output:

  the interleaving that needs it (build it by hand, it will not arise):
    t1: S1 leader(term2), writes e2, replicates to S2 only
    t2: S1 dies. S5 wins term3 with votes from S3,S4 (their logs are
        shorter but no entry is COMMITTED yet, so that is legal)
    t3: S1 returns, wins term4, replicates e2 to a majority
    t4: if S1 commits e2 NOW by counting replicas, and then dies,
        S5 can still win term5 and OVERWRITE e2 -- a committed entry.
  fix: commit an entry from your OWN term first; earlier entries then
  commit implicitly. ~1 run in 1e5 hits this. Write the test on purpose.

Beyond the toy

The construction that exposes this is worth building deliberately: replicate an entry to a majority under an old term, elect a new leader, and crash it before it appends anything of its own. With the rule removed, the entry is committed and then vanishes. That is a five-node, four-step scenario the deterministic simulator in block 1 can produce in milliseconds — and it is the single best argument for having built the simulator first.

The general lesson generalises past Raft: the dangerous rules in a protocol are the ones that only matter in a failure sequence you have to construct deliberately. They pass every test written from the happy path, they pass code review because the reasoning is subtle, and they fail in production during the exact incident you least wanted a second failure.

The assembly

Every block above, wired together into one working system:

def assembly(s):
    Net = s["Net"]
    print("\nSeven blocks = a replicated store. Now run it against the injector.\n")
    scenarios = [("healthy",              None,               0.0, 0.0),
                 ("10% packet loss",      None,               0.1, 0.0),
                 ("duplicate delivery",   None,               0.0, 0.5),
                 ("partition 3|2",        [[0,1,2],[3,4]],    0.0, 0.0),
                 ("partition 4|1",        [[0,1,2,3],[4]],    0.0, 0.0)]
    print(f"  {'scenario':<22}{'rounds':>8}{'leaders':>9}{'committed':>11}{'replicas w/ x':>15}")
    for name, part, drop, dup in scenarios:
        net = Net(7); net.partitions = part or []; net.drop_p = drop; net.dup_p = dup
        nodes = {i: s["Node"](i, [j for j in range(5) if j != i]) for i in range(5)}
        # Election RETRY. A single round can lose enough vote messages that nobody
        # wins -- which is exactly why Raft retries on a randomised timeout rather
        # than assuming one round succeeds. Without this the 10%-loss row elects
        # nobody, which is a property of the harness, not of the protocol.
        leaders, rounds = [], 0
        for rounds in range(1, 6):
            for cand in ({min(g) for g in part} if part else {0}):
                nodes[cand].stand(net); net.run(nodes, net.t + 60)
            leaders = [i for i, n in nodes.items() if n.role == "leader"]
            if leaders: break
        ok = False
        if leaders:
            ok, _, _ = s["replicate"](nodes, leaders[0], net, [("x", "1")])
        holders = sum(1 for n in nodes.values() if n.store.get("x") == "1")
        print(f"  {name:<22}{rounds:>8}{len(leaders):>9}{str(ok):>11}{holders:>15}")
    print("\n  Exactly one leader in every scenario -- never two, which is the")
    print("  safety property. Note the ROUNDS column: under 10% loss the first")
    print("  election fails outright and a retry is required. That is why Raft has")
    print("  randomised election timeouts; a protocol that assumed one round would")
    print("  simply stall. The 4|1 split elects on the majority side and the")
    print("  singleton correctly cannot.")
    print("\n  Linearizability over the healthy run's history:")
    hist = [("put", "x", "1", 0, 5), ("get", "x", "1", 6, 8)]
    print(f"    {hist}")
    print(f"    verdict: {s['linearizable'](hist)}")
    print("\n  Built: deterministic net -> fault injector -> nodes/terms/votes ->")
    print("  election -> replication+commit -> linearizability checker -> 5.4.2.")
    print("  Missing, on the project page: persistence (m8), client sessions (m9),")
    print("  phi-accrual (m10), membership (m11), snapshots (m12), sharding (m13).")
    print("  And the exit criterion: ZERO violations across 1,000 seeded fault runs.")

Output:

Seven blocks = a replicated store. Now run it against the injector.

  scenario                rounds  leaders  committed  replicas w/ x
  healthy                      1        1       True              5
  10% packet loss              2        1       True              4
  duplicate delivery           1        1       True              5
  partition 3|2                1        1       True              3
  partition 4|1                1        1       True              4

  Exactly one leader in every scenario -- never two, which is the
  safety property. Note the ROUNDS column: under 10% loss the first
  election fails outright and a retry is required. That is why Raft has
  randomised election timeouts; a protocol that assumed one round would
  simply stall. The 4|1 split elects on the majority side and the
  singleton correctly cannot.

  Linearizability over the healthy run's history:
    [('put', 'x', '1', 0, 5), ('get', 'x', '1', 6, 8)]
    verdict: True

  Built: deterministic net -> fault injector -> nodes/terms/votes ->
  election -> replication+commit -> linearizability checker -> 5.4.2.
  Missing, on the project page: persistence (m8), client sessions (m9),
  phi-accrual (m10), membership (m11), snapshots (m12), sharding (m13).
  And the exit criterion: ZERO violations across 1,000 seeded fault runs.

The design space

Consensus protocols all solve the same problem — agree on a total order of commands despite crashes and an asynchronous network — and differ in how they handle the leader.

The Flexible Paxos observation is worth internalising because it exposes what the majority is actually for. Raft requires \(|Q_{elect}| + |Q_{replicate}| > N\); it satisfies this by making both \(\lceil (N+1)/2 \rceil\), but that is a choice. With \(N=5\) you can use an election quorum of 4 and a replication quorum of 2, making steady-state commits cheaper at the cost of more expensive elections. The invariant is intersection, not majority — see proofs.md P4.

The rule the blocks demonstrate

Election safety comes from two facts multiplied together: each server casts at most one vote per term, and a winner needs a majority. Two majorities of the same set must intersect in at least one server, and that server cannot have voted twice. That is the whole argument, and the partition scenarios in the assembly exercise it directly.

The subtler rule — §5.4.2, a leader may not commit an entry from a previous term by counting replicas — is the one most implementations get wrong. Without it, an entry replicated to a majority under an old term can still be overwritten, and a client that was told "committed" watches its write vanish.

Latency: distance is the budget

Consensus costs one round trip to a quorum, so geography sets the floor.

Light in fibre travels ~200,000 km/s, so 5,000 km is 25 ms one way at the physical limit. No protocol beats that; a globally replicated write is fundamentally a ~100 ms operation, which is why systems that need low write latency (a) keep the quorum inside one region and replicate asynchronously across regions, or (b) shard so that each key's quorum is local to its users.

Add durability: every Raft append should fsync before acknowledging, and numbers.md measures that at 90--105 µs here. In a same-rack cluster, the disk flush is comparable to the network round trip — which is why group commit (batching many log entries per flush) matters as much in consensus as it does in P04.

Reads are the interesting half

A linearizable read cannot simply be served from the leader's memory: the leader may have been deposed without knowing. The options:

• Read from the log — treat it as a no-op command through consensus. Correct, costs a full round trip.
• ReadIndex — the leader confirms leadership with one round of heartbeats, then serves from local state. One RTT, no disk write.
• Leader leases — the leader holds a time-bounded lease and serves reads locally with no communication. Fastest, but correctness now depends on bounded clock drift, which is an assumption about hardware, not about the protocol.
• Follower reads at a timestamp — what Spanner does with TrueTime and CockroachDB with hybrid logical clocks; correctness depends on a bounded uncertainty interval, and Spanner waits out that interval (commit-wait) rather than pretending it is zero.

Advanced algorithms and data structures

• Hybrid logical clocks (HLC) combine a physical timestamp with a Lamport counter, giving causally consistent ordering without atomic clocks.
• CRDTs solve a different problem — convergence without coordination — and are the right answer when the application's operations commute. They trade linearizability for availability; the invariant they cannot express is "at most one of these".
• Chain replication achieves strong consistency with higher throughput than quorum protocols by pipelining writes down a chain and serving reads from the tail; the cost is a longer failure-recovery path and dependence on an external configuration service.
• Consistent hashing with virtual nodes assigns keys to replica sets so that adding a node moves \(O(1/N)\) of the keyspace. Rendezvous hashing achieves the same with a simpler rule and better balance.
• Jepsen-style verification. Linearizability checking is NP-hard in general; Knossos and Elle use the Wing–Gong algorithm with aggressive pruning, and Elle in particular infers cycles in the dependency graph from list-append operations, which is how it can localise anomalies rather than just detect them.

Hardware and network reality

• Packet loss is not the interesting failure. Partial partitions (A can reach B, B can reach C, A cannot reach C) and asymmetric partitions break more implementations than clean splits, because a node can be simultaneously reachable for heartbeats and unreachable for appends.
• Grey failures: a node that is slow rather than dead is worse than a crash, because failure detectors keep it in the quorum while it drags latency to its own speed. This is why "the slowest replica in the quorum" is the real latency, and why systems over-provision replicas so a quorum can exclude a straggler.
• NIC and kernel costs: a syscall is 127.59 ns here, and a TCP round trip inside a rack is dominated by kernel network stack traversal rather than wire time — which is the motivation for kernel bypass (DPDK, RDMA) in the lowest latency systems.
• Clock drift: typical NTP-synced servers drift tens of milliseconds; PTP gets to microseconds; Spanner's TrueTime uses GPS and atomic clocks to bound uncertainty at ~1--7 ms. Any protocol whose safety depends on clocks needs to state the bound it assumes.

How this connects to the rest of the track

• P04's WAL and this project's replicated log are the same abstraction with different durability quorums.
• P06 and P07 both need a fault-tolerant coordinator; in production that is etcd or ZooKeeper, i.e. this project.
• P07's exactly-once and this project's linearizability are both statements about effects being applied once, reached by different means.
• P15's tail-latency arithmetic explains why a quorum of 3 is faster than a quorum of 5 even though both need 2 acknowledgements.

Failure modes at scale

• Split brain from a stale leader serving reads under an expired lease.
• Log divergence after an incorrectly implemented §5.4.2.
• Election storms: randomised timeouts too short relative to RTT, so nodes keep interrupting each other. The assembly's retry loop shows the mechanism; the standard fix is a timeout range several times the RTT, plus pre-vote.
• Snapshot/restore bugs are the least-tested and most dangerous path: a node restoring from a snapshot must also restore the configuration state, or it can vote using a membership set that no longer exists.
• Membership changes are where most real bugs live; joint consensus exists because naive one-at-a-time changes can create two disjoint majorities.

Primary sources

• Ongaro & Ousterhout, In Search of an Understandable Consensus Algorithm (Raft, USENIX ATC 2014) — read §5.4.2 twice.
• Lamport, The Part-Time Parliament (1998) and Paxos Made Simple (2001).
• Howard et al., Flexible Paxos: Quorum Intersection Revisited (2016).
• Moraru et al., There Is More Consensus in Egalitarian Parliaments (EPaxos, SOSP 2013).
• Corbett et al., Spanner (OSDI 2012) — TrueTime and commit-wait.
• Herlihy & Wing, Linearizability (TOPLAS 1990).
• Kingsbury, the Jepsen reports — the best available catalogue of how these systems actually fail.

Running it

python3 handson/h05_distkv.py            # every block, then the assembly
python3 handson/h05_distkv.py --block 3  # just block 3 and its prerequisites
python3 handson/h05_distkv.py --quiet    # the assembly only

What to do with this

The rule this file does not exercise hard enough is §5.4.2 --- a leader may not commit an entry from a previous term by counting replicas. Construct the scenario: an entry replicated to a majority under an old term, a new leader, a crash before the new leader appends anything of its own. Then remove the rule and watch a committed entry disappear. It is the subtlest correctness argument in the protocol and the one most implementations get wrong.

Milestones, experiments, readings and exit criteria for this project: P05 — Distributed Key-Value Store.