#!/usr/bin/env python3 """Hands-on P05 — a replicated KV store, assembled from seven lego blocks.""" import heapq, itertools, random from _harness import block, run_all @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} @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 {} @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} @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} @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} @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} @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 {} 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.") if __name__ == "__main__": run_all(assembly, "HANDS-ON P05 — Distributed KV, block by block")