"""Raft's commit rule, in one file: why a majority is not enough. Five nodes, no threads, no sockets, no clock, no RNG. Messages sit in a list and the driver decides what is delivered and what is lost, so a run is a pure function of the schedule -- the only way to stage one interleaving on purpose. Section 5.4.2 of the Raft paper: replicating an entry to a majority does NOT make it safe if the entry came from an earlier term. The fix is one clause in `advance_commit`. Both rules are here so one schedule can run twice. """ RAFT = "raft" NAIVE = "naive" class Entry: """A log record. Storing the creating leader's `term` is what makes the commit rule expressible at all.""" __slots__ = ("term", "cmd") def __init__(self, term, cmd): self.term = term self.cmd = cmd def __repr__(self): return f"{self.cmd}@{self.term}" class Node: """One server. Logs are 1-indexed as in the paper: entry i is `log[i - 1]`.""" def __init__(self, nid, peers, rule): self.id = nid self.peers = peers self.rule = rule # Persistent: survives a crash. A node that forgot its term or its vote # could vote twice in one term and elect two leaders. self.term = 0 self.voted_for = None self.log = [] # Volatile: rebuilt after a crash from the persistent state above. self.role = "follower" self.commit_index = 0 self.votes = set() self.next_index = {} self.match_index = {} self.alive = True self.out = None # the shared message list, injected by Cluster def last(self): """(index, term) of the final log entry; (0, 0) when empty.""" return (len(self.log), self.log[-1].term if self.log else 0) def send(self, dst, **kw): self.out.append(dict(src=self.id, dst=dst, **kw)) def step_down(self, term): """Terms are the only clock in Raft. A higher one anywhere means stale.""" self.term = term self.role = "follower" self.voted_for = None def campaign(self): """Stand for election. Driven explicitly: no randomised timeouts (S8).""" self.term += 1 self.role = "candidate" self.voted_for = self.id self.votes = {self.id} idx, lt = self.last() for p in self.peers: self.send(p, type="vote", term=self.term, last_index=idx, last_term=lt) def become_leader(self): """Assume followers match, and let rejections walk `next_index` back.""" self.role = "leader" self.next_index = {p: len(self.log) + 1 for p in self.peers} self.match_index = {p: 0 for p in self.peers} self.replicate() def client(self, cmd): self.log.append(Entry(self.term, cmd)) self.replicate() def replicate(self, only=None): for p in self.peers: if only is None or p in only: self.send_append(p) def send_append(self, p): """Carries the (index, term) of the entry *before* the ones it sends; a follower matching there matches everywhere earlier.""" ni = self.next_index[p] self.send(p, type="append", term=self.term, prev_index=ni - 1, prev_term=self.log[ni - 2].term if ni > 1 else 0, entries=self.log[ni - 1:], commit=self.commit_index) def advance_commit(self): """THE commit rule. Walk down for the highest index a majority stores; under RAFT, skip any index whose entry came from an earlier term.""" for n in range(len(self.log), self.commit_index, -1): stored = 1 + sum(1 for p in self.peers if self.match_index[p] >= n) if stored * 2 <= len(self.peers) + 1: continue if self.rule == RAFT and self.log[n - 1].term != self.term: continue # <-- Figure 8 lives on this line self.commit_index = n # everything below n rides along return def recv(self, m): if not self.alive: return if m["term"] > self.term: self.step_down(m["term"]) getattr(self, "on_" + m["type"])(m) def on_vote(self, m): """The election restriction: never vote for a candidate whose log is behind yours, by (term, index) -- a longer log with an older final term loses to a shorter one with a newer term.""" idx, lt = self.last() up_to_date = (m["last_term"], m["last_index"]) >= (lt, idx) granted = (m["term"] == self.term and self.voted_for in (None, m["src"]) and up_to_date) if granted: self.voted_for = m["src"] self.send(m["src"], type="vote_reply", term=self.term, granted=granted) def on_vote_reply(self, m): if self.role != "candidate" or m["term"] != self.term: return if m["granted"]: self.votes.add(m["src"]) if len(self.votes) * 2 > len(self.peers) + 1: self.become_leader() def on_append(self, m): ok = False if m["term"] >= self.term: self.role = "follower" pi, pt = m["prev_index"], m["prev_term"] ok = pi <= len(self.log) and (pi == 0 or self.log[pi - 1].term == pt) if ok: # The leader's log wins, unconditionally. This truncation is the # line that erases a "committed" entry in the demo. self.log = self.log[:pi] + list(m["entries"]) self.commit_index = max(self.commit_index, min(m["commit"], len(self.log))) self.send(m["src"], type="append_reply", term=self.term, ok=ok, index=len(self.log) if ok else 0) def on_append_reply(self, m): if self.role != "leader" or m["term"] != self.term: return if m["ok"]: self.match_index[m["src"]] = m["index"] self.next_index[m["src"]] = m["index"] + 1 self.advance_commit() else: self.next_index[m["src"]] = max(1, self.next_index[m["src"]] - 1) self.send_append(m["src"]) class Cluster: """The driver: owns the nodes, the message list, and the schedule.""" def __init__(self, n, rule): self.ids = list(range(1, n + 1)) self.net = [] self.nodes = {} for i in self.ids: node = Node(i, [p for p in self.ids if p != i], rule) node.out = self.net self.nodes[i] = node def deliver(self, only=None, rounds=12): """Run until quiet. Anything addressed outside `only` is LOST, not queued -- the partition model, and what lets a schedule say "this entry reaches exactly S2".""" for _ in range(rounds): batch, self.net[:] = list(self.net), [] if not batch: break for m in batch: if self.nodes[m["dst"]].alive and (only is None or m["dst"] in only): self.nodes[m["dst"]].recv(m) def elect(self, i, tries=4, only=None): """Campaign until node `i` wins, or give up. A stale candidate loses, learns the higher term and stands again, so one call burns several terms -- real behaviour, and why S1 needs two attempts.""" for _ in range(tries): self.nodes[i].campaign() self.deliver(only=only) if self.nodes[i].role == "leader": return self.nodes[i].term return None def crash(self, i): """Stop the node. Log, term and vote persist; volatile state is lost. `commit_index` is volatile in Figure 2, so it returns to zero -- which is why a restarted leader can commit nothing until a client writes.""" node = self.nodes[i] node.alive = False node.role = "follower" node.votes = set() node.commit_index = 0 def restart(self, i): self.nodes[i].alive = True def holders(self, index, cmd): """Which nodes currently store `cmd` at `index`?""" return [i for i in self.ids if len(self.nodes[i].log) >= index and self.nodes[i].log[index - 1].cmd == cmd]