The Project | About Us | Contribute | Donations | License
HOME My courses Completed
◄ Previous
◄

Consensus Algorithms

Next ►
►

In the previous lesson we said quite casually that a CP system "rejects writes that do not reach a quorum" and that "the side of the partition with the majority keeps working". Behind those phrases lies a problem that distributed computing took decades to solve: how a group of nodes, each with its own clock, with messages that get lost or arrive late and with peers that may die at any moment, agrees on one value, in such a way that no node decides something different and that, if things go reasonably well, some node actually decides. That problem is called consensus, and it is the piece that turns the "CP" label into a mechanism.

This lesson tackles it at three levels. First, the problem itself: which properties it demands, why the FLP result from 01-02 says it has no guaranteed solution in an asynchronous system and why, even so, it is solved in practice every day. Second, the two algorithms that dominate the industry: Paxos, Lamport's original, with its roles and its two phases, and Raft, designed to be understandable, with its terms, its leader election and its replicated log; both with Python simulations that show several competing proposers and a leader that crashes. Third, an overview of Byzantine consensus (PBFT) and of when it is needed. We will close with the systems that package these algorithms (etcd, ZooKeeper, Consul) and with their first real use at Kilometre Zero: guaranteeing that only one instance of the outbox relay from 02-05 is publishing at any given moment, using etcd with a lease. Data replication in general (leader-follower, multi-leader, read and write quorums) is the next lesson; here consensus is used to elect leaders and agree on small values.

Contents

  1. The consensus problem
  2. Why it is hard: FLP and the practical way out
  3. What consensus is used for
  4. Paxos: roles, phases and proposal numbers
  5. Simulation: Paxos with two competing proposers
  6. Raft: terms, leader election and replicated log
  7. Simulation: leader election in Raft and a leader crash
  8. Byzantine consensus: PBFT in one page
  9. Comparison table: Paxos, Raft and PBFT
  10. Systems that implement it and their use at Kilometre Zero: etcd and the outbox relay
  11. Common mistakes and tips
  12. Exercises
  13. Conclusion

  1. The consensus problem

A set of N processes, each with an initial proposed value, must decide on a single value. A consensus algorithm is correct if it satisfies three properties:

Property Statement Kind
Agreement Two correct processes never decide different values Safety: nothing bad ever happens
Validity The decided value was proposed by some process (deciding a default value nobody proposed does not count) Safety
Termination Every correct process eventually decides Liveness: something good eventually happens

The first two are safety properties: they can be violated at one specific instant and there is no going back (if inv-bcn decides the leader is relay-1 and inv-vlc that it is relay-2, the damage is done). The third is a liveness property: it is violated only if the system gets stuck for ever. This distinction is central because, as we shall see, practical algorithms never sacrifice safety and accept sacrificing liveness temporarily.

At Kilometre Zero, the "value" to decide will be things like "who is the leader of the outbox relay in term 7", "the current configuration is version 12" or, in the general case of consensus-based replication, "entry number 4,312 of the log is reserve aged-cheese for P-2026-000123".

  1. Why it is hard: FLP and the practical way out

In 01-02 we introduced the result of Fischer, Lynch and Paterson (1985): in an asynchronous system (with no bounds on message latency or on process speed), no deterministic consensus algorithm guarantees termination if even a single process may crash. The intuition: if a process does not answer, the others cannot tell whether it has died or its message is just taking a long time; if they wait, they may wait for ever; if they decide without it, it may turn out that it was alive and about to decide something else. There is always an adversary (a schedule of delays) that keeps the algorithm undecided.

FLP does not say consensus is impossible in practice; it says it cannot be guaranteed in the asynchronous worst case. The practical ways out are the ones we anticipated in 01-02:

  • Partial synchrony: assume the network behaves well "most of the time" and use timeouts. Paxos and Raft guarantee safety always, on any network, and termination only when the network is going through a stable period. This is exactly the safety/liveness split from section 1.
  • Randomness: Raft's randomised timeouts and some probabilistic protocols break the symmetry that the FLP adversary needs.
  • Failure detectors: suspect a node after a timeout, accepting that you may be wrong (treating a slow node as dead is still safe; it only costs progress).

And one rule worth committing to memory: consensus needs a majority. With N nodes, f crash failures are tolerated if N ≥ 2f + 1: three nodes tolerate one, five tolerate two. The reason is that any two majorities of N intersect in at least one node, and that shared node is what prevents two groups from deciding different values. It is also why a CP system with two nodes, like the one in the simulation of 03-02, cannot tolerate any partition.

  1. What consensus is used for

Pure consensus (agreeing on one value) is rarely used directly. It is used as a primitive to build:

Use What is decided Example at Kilometre Zero
Leader election Who is in charge for a period A single instance of the orders outbox relay publishes to orders.events; a single scheduler assigns couriers
Replicated state machine (replicated log) The order of all operations The basis of a CP store: all replicas apply the same log in the same order and therefore have the same state (this is how etcd, CockroachDB and Kafka with KRaft work)
Distributed configuration The current version of the configuration Stock threshold for switching to CP mode (03-02), list of active markets, feature flags for "Artisan Cheese Week"
Distributed locks and membership Who holds the lock; which nodes are in the group Preventing two analytics processes from rebuilding the same report; knowing which inventory replicas are alive
Atomic commit Whether a transaction commits or aborts A variant of consensus with different requirements, in 03-05

The idea of the replicated state machine (Lamport, 1978; Schneider, 1990) deserves one more sentence: if several nodes start from the same state and apply the same deterministic operations in the same order, they end up in the same state. Consensus is used to agree on the order (entry 1, entry 2, entry 3...), and from there replication is trivial. It is the pattern behind Multi-Paxos and Raft, and the reason both talk about a "log".

  1. Paxos: roles, phases and proposal numbers

Leslie Lamport published Paxos in 1998 (after an initial rejection in 1989 of the paper, written as a parable about a Greek parliament). It solves consensus on a single value (single-decree Paxos) with three roles, which in practice usually coexist on every node:

  • Proposer: proposes a value. There may be several at once, and therein lies the difficulty.
  • Acceptor: votes. It remembers two things in stable storage: the highest proposal number it has promised not to ignore, and the last proposal it has accepted (number and value). A value is chosen when a majority of acceptors has accepted it with the same proposal number.
  • Learner: finds out the chosen value (by asking the acceptors or by receiving notifications).

Every proposal carries a unique, totally ordered proposal number; the usual trick is n = round * 10 + proposer_id, so that two proposers never generate the same number. The protocol has two phases:

sequenceDiagram
    participant P as Proposer (relay-1)
    participant A1 as Acceptor a1
    participant A2 as Acceptor a2
    participant A3 as Acceptor a3
    Note over P,A3: Phase 1: prepare / promise
    P->>A1: prepare(n=1)
    P->>A2: prepare(n=1)
    P->>A3: prepare(n=1)
    A1-->>P: promise(1, nothing accepted)
    A2-->>P: promise(1, nothing accepted)
    A3-->>P: promise(1, nothing accepted)
    Note over P: Majority of promises and nobody had accepted anything: I propose my value
    Note over P,A3: Phase 2: accept / accepted
    P->>A1: accept(n=1, "relay-1")
    P->>A2: accept(n=1, "relay-1")
    P->>A3: accept(n=1, "relay-1")
    A1-->>P: accepted(1)
    A2-->>P: accepted(1)
    A3-->>P: accepted(1)
    Note over P,A3: Majority of accepted with n=1: the value "relay-1" is CHOSEN

Phase 1 (prepare/promise). The proposer picks a number n and sends prepare(n) to the acceptors. An acceptor that receives prepare(n) with n greater than any number it has promised replies promise(n, accepted), undertaking not to accept any proposal numbered lower than n, and including the proposal it had previously accepted, if any. If n is less than or equal to its promise, it ignores or rejects the request.

Phase 2 (accept/accepted). If the proposer receives promises from a majority, it picks the value to propose using this rule, which is the heart of Paxos: if any promise included an already accepted proposal, it must propose the value of the accepted proposal with the highest number; only if none included anything may it propose its own value. It sends accept(n, value); each acceptor accepts it unless it has promised a higher number in the meantime, and replies accepted. With a majority of accepted replies, the value is chosen.

Why it works. Suppose value v has been chosen with number n (a majority accepted it). Any later proposal with number m > n has to obtain promises from a majority, which necessarily intersects the majority that accepted v; at least one acceptor in that intersection will report (n, v) in its promise, and the phase 2 rule will force the new proposer to propose v again. By induction, once a value has been chosen, all future proposals carry that same value: agreement is guaranteed, with any number of proposers and any ordering of messages. Validity is immediate (only proposed values are ever proposed) and termination... is not guaranteed (FLP): two proposers can take turns in phase 1 with ever-increasing numbers, invalidating each other's promises for ever (livelock). The practical solution is to elect a distinguished proposer using timeouts, that is, a leader.

Why it is hard to implement. Single-value Paxos is elegant, but a real system needs to decide a sequence of values (the state machine's log). Multi-Paxos does that: it runs one Paxos instance per log entry, and optimises by electing a stable leader that performs phase 1 just once for all future entries and then only runs phase 2 per entry. But Lamport's paper does not specify how to elect the leader, how to handle membership changes, how to compact the log or how to bring acceptors that have fallen behind back up to date. Every implementation (Google's Chubby, Spanner, Cassandra for its lightweight transactions) fills those gaps in its own way, and the authors of Chubby wrote that "there are significant gaps between the description of the Paxos algorithm and the needs of a real-world system". That frustration is the origin of Raft.

  1. Simulation: Paxos with two competing proposers

We are going to implement single-value Paxos with three acceptors and two proposers, relay-1 and relay-2, each of which wants to be elected leader of the outbox relay. The network lets us drop specific messages in order to reproduce the interesting case: relay-1 gets one acceptor to accept its value, loses the rest of its messages, and relay-2 arrives later with a higher number.

# km0/simulations/paxos_single_value.py
from dataclasses import dataclass, field


@dataclass
class Acceptor:
    name: str
    promised: int | None = None                  # highest number promised
    accepted: tuple[int, str] | None = None      # (number, value) of the last accepted proposal

    def prepare(self, n: int) -> tuple[str, object]:
        if self.promised is None or n > self.promised:
            self.promised = n
            return ("promise", self.accepted)       # includes whatever it had already accepted
        return ("nack", self.promised)

    def accept(self, n: int, value: str) -> tuple[str, object]:
        if self.promised is None or n >= self.promised:
            self.promised = n
            self.accepted = (n, value)
            return ("accepted", n)
        return ("nack", self.promised)


class Network:
    """Drops the messages listed in `lost`: (proposer, acceptor, phase)."""
    def __init__(self) -> None:
        self.lost: set[tuple[str, str, str]] = set()

    def delivers(self, proposer: str, acceptor: str, phase: str) -> bool:
        return (proposer, acceptor, phase) not in self.lost


@dataclass
class Proposer:
    name: str
    id: int
    acceptors: list[Acceptor]
    network: Network
    round: int = 0
    trace: list[str] = field(default_factory=list)

    def _log(self, msg: str) -> None:
        print(f"  [{self.name}] {msg}")

    def propose(self, value: str) -> str | None:
        self.round += 1
        n = self.round * 10 + self.id                        # unique, increasing number
        majority = len(self.acceptors) // 2 + 1
        self._log(f"phase 1: prepare(n={n}) wanting to propose '{value}'")

        promises: list[tuple[int, str] | None] = []
        for a in self.acceptors:
            if not self.network.delivers(self.name, a.name, "prepare"):
                self._log(f"    prepare to {a.name} LOST"); continue
            kind, payload = a.prepare(n)
            self._log(f"    {a.name} -> {kind} {payload if payload else ''}")
            if kind == "promise":
                promises.append(payload)
        if len(promises) < majority:
            self._log(f"phase 1 failed: {len(promises)} promises < majority {majority}")
            return None

        # Key rule: if anyone has already accepted something, adopt the value with the highest number
        already_accepted = [p for p in promises if p is not None]
        if already_accepted:
            n_prev, value_prev = max(already_accepted)
            if value_prev != value:
                self._log(f"phase 2: an acceptor already accepted ({n_prev}, '{value_prev}'): ADOPTING '{value_prev}' and dropping '{value}'")
            value = value_prev
        self._log(f"phase 2: accept(n={n}, '{value}')")

        accepted_count = 0
        for a in self.acceptors:
            if not self.network.delivers(self.name, a.name, "accept"):
                self._log(f"    accept to {a.name} LOST"); continue
            kind, payload = a.accept(n, value)
            self._log(f"    {a.name} -> {kind} {payload}")
            accepted_count += kind == "accepted"
        if accepted_count >= majority:
            self._log(f"CHOSEN '{value}' with n={n} ({accepted_count} of {len(self.acceptors)})")
            return value
        self._log(f"phase 2 failed: {accepted_count} accepted < majority {majority}")
        return None


if __name__ == "__main__":
    network = Network()
    acceptors = [Acceptor("a1"), Acceptor("a2"), Acceptor("a3")]
    relay1 = Proposer("relay-1", id=1, acceptors=acceptors, network=network)
    relay2 = Proposer("relay-2", id=2, acceptors=acceptors, network=network)

    print("Round A: relay-1 proposes, but its accepts to a2 and a3 are lost")
    network.lost = {("relay-1", "a2", "accept"), ("relay-1", "a3", "accept")}
    print("  result:", relay1.propose("relay-1"))

    print("\nRound B: relay-2 proposes with a higher number; the network now works")
    network.lost = set()
    print("  result:", relay2.propose("relay-2"))

    print("\nRound C: relay-1 retries with an even higher number")
    print("  result:", relay1.propose("relay-1"))

    print("\nFinal state of the acceptors:")
    for a in acceptors:
        print(f"  {a.name}: promised={a.promised}, accepted={a.accepted}")

Output:

Round A: relay-1 proposes, but its accepts to a2 and a3 are lost
  [relay-1] phase 1: prepare(n=11) wanting to propose 'relay-1'
  [relay-1]    a1 -> promise
  [relay-1]    a2 -> promise
  [relay-1]    a3 -> promise
  [relay-1] phase 2: accept(n=11, 'relay-1')
  [relay-1]    a1 -> accepted 11
  [relay-1]    accept to a2 LOST
  [relay-1]    accept to a3 LOST
  [relay-1] phase 2 failed: 1 accepted < majority 2
  result: None

Round B: relay-2 proposes with a higher number; the network now works
  [relay-2] phase 1: prepare(n=12) wanting to propose 'relay-2'
  [relay-2]    a1 -> promise (11, 'relay-1')
  [relay-2]    a2 -> promise
  [relay-2]    a3 -> promise
  [relay-2] phase 2: an acceptor already accepted (11, 'relay-1'): ADOPTING 'relay-1' and dropping 'relay-2'
  [relay-2] phase 2: accept(n=12, 'relay-1')
  [relay-2]    a1 -> accepted 12
  [relay-2]    a2 -> accepted 12
  [relay-2]    a3 -> accepted 12
  [relay-2] CHOSEN 'relay-1' with n=12 (3 of 3)
  result: relay-1

Round C: relay-1 retries with an even higher number
  [relay-1] phase 1: prepare(n=21) wanting to propose 'relay-1'
  [relay-1]    a1 -> promise (12, 'relay-1')
  [relay-1]    a2 -> promise (12, 'relay-1')
  [relay-1]    a3 -> promise (12, 'relay-1')
  [relay-1] phase 2: accept(n=21, 'relay-1')
  [relay-1]    a1 -> accepted 21
  [relay-1]    a2 -> accepted 21
  [relay-1]    a3 -> accepted 21
  [relay-1] CHOSEN 'relay-1' with n=21 (3 of 3)
  result: relay-1

Final state of the acceptors:
  a1: promised=21, accepted=(21, 'relay-1')
  a2: promised=21, accepted=(21, 'relay-1')
  a3: promised=21, accepted=(21, 'relay-1')

What the run teaches us:

  • In round A, relay-1 obtains all three promises but only a1 accepts: the value has not been chosen (there is no majority). However, a1 remembers (11, 'relay-1').
  • In round B, relay-2 wants to propose itself, but the promise from a1 tells it that there is already an accepted proposal. The phase 2 rule forces it to drop its own value and adopt relay-1, even though relay-1 was never actually chosen. This is conservative behaviour: Paxos cannot know whether (11, 'relay-1') was accepted by a majority of which it can see only one member, so it assumes that it may have been. The result is that relay-2 gets relay-1 chosen.
  • In round C, relay-1 retries and, naturally, confirms the same value. Agreement is preserved across all three rounds, with lost messages and competing proposers. Note that the proposal numbers (11, 12, 21) never collide thanks to round * 10 + id.

Try making round A lose the accept to a1 as well: then no promise in round B will include anything, and relay-2 will be chosen. And try alternating lost so that each proposer invalidates the other's promises in phase 1: you will see the livelock that motivates having a leader.

  1. Raft: terms, leader election and replicated log

Diego Ongaro and John Ousterhout published Raft in 2014 with a stated goal: to be understandable, with the same fault tolerance and performance as Multi-Paxos. Their strategy was to break the problem down into three subproblems (leader election, log replication and safety) and to reduce the number of possible states. Today it is the algorithm behind etcd, Consul, CockroachDB, TiKV, Kafka (KRaft) and RabbitMQ (quorum queues), among others.

Terms and states

Time is divided into terms, numbered in increasing order. Each term begins with an election; if the election succeeds, a single leader governs for the rest of the term. Terms act as a logical clock (01-05): every message carries the sender's term, and a node that sees a term higher than its own adopts it immediately and becomes a follower. Each node is in one of three states:

stateDiagram-v2
    [*] --> Follower
    Follower --> Candidate: election timeout with no heartbeats from the leader
    Candidate --> Leader: votes from the majority
    Candidate --> Follower: discovers a leader or a higher term
    Candidate --> Candidate: timeout with no majority (new term)
    Leader --> Follower: discovers a higher term

Leader election

  1. Every node starts as a follower and waits for heartbeats from the leader. Each follower has a randomised election timeout (typically between 150 and 300 ms).
  2. If the timeout expires without any heartbeats having been received, the follower becomes a candidate: it increments its term, votes for itself and sends RequestVote to the others.
  3. Each node grants only one vote per term, to the first candidate that asks for it and meets the safety condition below. If the candidate receives votes from the majority, it is the leader and starts sending heartbeats immediately, which makes the other candidates stand down.
  4. If two candidates split the vote (neither has a majority), both wait for a new randomised timeout and try again in a new term. Randomness makes it very unlikely that they will tie repeatedly: in practice, one of them times out first and wins.

It is randomness that resolves the Paxos livelock and sidesteps FLP: safety always, termination with probability 1 once the network stabilises.

Log replication

Clients talk only to the leader. Each operation is appended to the leader's log as an entry (index, term, command) and sent to the followers with AppendEntries (the same message that serves as a heartbeat when it is empty). When the leader knows that the entry is in a majority of logs, it marks it as committed, applies it to its state machine, replies to the client and tells the followers so that they apply it too. A follower whose log is out of line (because it was down) is corrected by the leader, which makes it back up to the last point they have in common and resends the rest: the leader's log is always the truth.

Safety

The property Raft proves is that a committed entry is never lost or overwritten, even if the leader changes. Two rules guarantee it:

  • Election restriction: a node only votes for a candidate whose log is at least as up to date as its own (comparing the term of the last entry and, if equal, the index). Since a committed entry is on a majority, and the candidate needs votes from a majority, at least one voter has the entry and will not vote for a candidate that lacks it. The elected leader therefore has every committed entry, with no need for the state transfer of Paxos.
  • Committing only entries from the current term: a leader only counts replicas in order to commit entries from its own term; those from earlier terms are committed indirectly when a later one is committed. This avoids a subtle case in which an old entry replicated late could be committed and then overwritten.

Raft also specifies configuration changes (adding or removing nodes with joint configuration) and log compaction by means of snapshots, precisely the gaps that Paxos left open.

  1. Simulation: leader election in Raft and a leader crash

The following simulation implements Raft leader election with asyncio: three nodes with randomised timeouts, terms, votes with the log restriction, and heartbeats. It does not replicate the log (the nodes have a fixed log so that we can show the election restriction), but it lets us see a real election, the leader crashing and the re-election, and what happens when a node with a stale log tries to become leader.

# km0/simulations/raft_election.py
import asyncio
import random

T0 = None


def now() -> float:
    return asyncio.get_running_loop().time() - T0


class Network:
    """Delivers messages with random latency; does not deliver to crashed nodes."""
    def __init__(self, nodes: dict, rng: random.Random):
        self.nodes, self.rng = nodes, rng

    async def call(self, dest: str, method: str, *args):
        await asyncio.sleep(self.rng.uniform(0.002, 0.010))     # network latency
        node = self.nodes[dest]
        if not node.alive:
            await asyncio.sleep(0.05)                          # RPC timeout
            return None
        return getattr(node, method)(*args)


class RaftNode:
    def __init__(self, id: str, last_index: int, last_term: int, rng: random.Random,
                 first_deadline: float | None = None):
        self.id, self.rng = id, rng
        self.first_deadline = first_deadline             # to force who times out first in the demo
        self.state = "follower"
        self.term = 0
        self.voted_for: str | None = None
        self.last_index, self.last_term = last_index, last_term   # fixed log
        self.alive = True
        self.deadline = 0.0
        self.network: Network | None = None

    # --- RPCs received from the other nodes ----------------------------------
    def request_vote(self, term: int, candidate: str, last_idx: int, last_term: int):
        if term > self.term:
            self.term, self.state, self.voted_for = term, "follower", None
        log_up_to_date = (last_term, last_idx) >= (self.last_term, self.last_index)
        grant = (term == self.term and self.voted_for in (None, candidate) and log_up_to_date)
        if grant:
            self.voted_for = candidate
            self._reset_deadline()
        elif term == self.term and not log_up_to_date:
            print(f"{now():6.3f}s  {self.id}: DENYING vote to {candidate} (its log ({last_term},{last_idx}) "
                  f"is behind mine ({self.last_term},{self.last_index}))")
        return (self.term, grant)

    def receive_heartbeat(self, term: int, leader: str):
        if term >= self.term:
            if self.state != "follower" or term > self.term:
                print(f"{now():6.3f}s  {self.id}: recognising {leader} as leader of term {term}")
            self.term, self.state, self.voted_for = term, "follower", None
            self._reset_deadline()
        return self.term

    # --- main loop ------------------------------------------------------------
    def _reset_deadline(self) -> None:
        self.deadline = now() + self.rng.uniform(0.150, 0.300)

    async def run(self, peers: list[str]) -> None:
        self._reset_deadline()
        if self.first_deadline is not None:
            self.deadline = now() + self.first_deadline
        while True:
            await asyncio.sleep(0.010)
            if not self.alive:
                continue
            if self.state == "leader":
                await asyncio.gather(*(self.network.call(p, "receive_heartbeat", self.term, self.id) for p in peers))
                await asyncio.sleep(0.040)                      # heartbeats every ~50 ms
            elif now() > self.deadline:
                await self._election(peers)

    async def _election(self, peers: list[str]) -> None:
        self.term += 1
        self.state, self.voted_for = "candidate", self.id
        self._reset_deadline()
        print(f"{now():6.3f}s  {self.id}: timeout, standing for election in term {self.term}")
        replies = await asyncio.gather(*(self.network.call(p, "request_vote", self.term, self.id,
                                                           self.last_index, self.last_term) for p in peers))
        if self.state != "candidate":             # somebody else won while we were waiting
            return
        votes = 1 + sum(1 for r in replies if r and r[1] and r[0] == self.term)
        for r in replies:
            if r and r[0] > self.term:
                self.term, self.state = r[0], "follower"; return
        if votes > (len(peers) + 1) // 2:
            self.state = "leader"
            print(f"{now():6.3f}s  {self.id}: LEADER of term {self.term} with {votes} votes")


async def main() -> None:
    global T0
    T0 = asyncio.get_running_loop().time()
    rng = random.Random(3)
    # node-3 has a stale log (index 3 versus 5) and is the first to time out (0.12 s):
    # it must NOT be able to become leader, however early it stands
    nodes = {"node-1": RaftNode("node-1", 5, 1, rng), "node-2": RaftNode("node-2", 5, 1, rng),
             "node-3": RaftNode("node-3", 3, 1, rng, first_deadline=0.120)}
    network = Network(nodes, rng)
    for n in nodes.values():
        n.network = network
    tasks = [asyncio.create_task(n.run([p for p in nodes if p != n.id])) for n in nodes.values()]

    await asyncio.sleep(1.0)
    leader = next(n for n in nodes.values() if n.state == "leader")
    print(f"{now():6.3f}s  --- {leader.id} CRASHES ---")
    leader.alive = False
    await asyncio.sleep(1.0)
    print(f"{now():6.3f}s  --- {leader.id} COMES BACK (believes it is still in term {leader.term}) ---")
    leader.alive = True
    await asyncio.sleep(0.5)
    print("final state:", {n.id: (n.state, n.term) for n in nodes.values()})
    for t in tasks:
        t.cancel()


if __name__ == "__main__":
    asyncio.run(main())

Output (the times may vary by a few milliseconds because they depend on the asyncio scheduler, but the sequence of events is the same on every run thanks to the seed and to the first_deadline of node-3):

 0.122s  node-3: timeout, standing for election in term 1
 0.125s  node-2: DENYING vote to node-3 (its log (1,3) is behind mine (1,5))
 0.129s  node-1: DENYING vote to node-3 (its log (1,3) is behind mine (1,5))
 0.194s  node-1: timeout, standing for election in term 2
 0.203s  node-1: LEADER of term 2 with 3 votes
 1.001s  --- node-1 CRASHES ---
 1.153s  node-3: timeout, standing for election in term 3
 1.163s  node-2: DENYING vote to node-3 (its log (1,3) is behind mine (1,5))
 1.281s  node-2: timeout, standing for election in term 4
 1.336s  node-2: LEADER of term 4 with 2 votes
 2.001s  --- node-1 COMES BACK (believes it is still in term 2) ---
 2.003s  node-1: recognising node-2 as leader of term 4
final state: {'node-1': ('follower', 4), 'node-2': ('leader', 4), 'node-3': ('follower', 4)}

What to look for:

  • The election restriction in action. node-3, with its stale log, is the first to time out and stands in term 1, but the other two deny it their vote because its last entry (1, 3) is older than theirs (1, 5). It can never become leader while it is behind, which protects the committed entries it does not have. Its request has had one effect, though: the others have adopted term 1, and when node-1 times out shortly afterwards it stands in term 2 and wins with all three votes (node-3 votes for it too: that log is more up to date than its own).
  • Crash and re-election. When node-1 crashes, the heartbeats stop. node-3 is again the first to time out (term 3) and is again denied by node-2; node-1 does not answer because it is down. About 130 ms later node-2 times out, stands in term 4 and wins with two votes (its own and that of node-3): a majority of 3. The system has been leaderless for about 335 ms. That is the availability cost of a CP system during a failure: bounded and small, not indefinite. Note that the term has jumped from 2 to 4: terms used up by failed elections are not reused.
  • The old leader comes back. node-1 revives believing itself to be leader of term 2, but the first heartbeat from node-2 with term 4 sends it back to follower. Terms acting as a logical clock prevent two leaders from operating at once: a message with an old term is ignored by everyone. (In a real system, node-1 could have tried to send an AppendEntries for term 2 before receiving the heartbeat; the followers would reject it as a stale term.)

If you remove the first_deadline from node-3, the order in which the nodes time out depends on the randomised timeouts and the trace changes on every run; in many of them node-3 never even gets to stand. That is what happens in production: the election restriction only shows itself when the lagging node times out before the others.

  1. Byzantine consensus: PBFT in one page

Everything above assumes the crash failure model of 01-02: a node that fails goes quiet. If a node can lie (a Byzantine failure: a bug that sends inconsistent messages, a disk that corrupts the log, a malicious participant), Paxos and Raft are no use: an acceptor that promises to two proposers at once, or a leader that sends different logs to each follower, breaks agreement.

Byzantine consensus solves this case at a higher cost. The classic result (Lamport, Shostak and Pease, 1982) is that N ≥ 3f + 1 nodes are needed to tolerate f Byzantine ones: 4 nodes to tolerate 1, 7 to tolerate 2. The intuition: with f liars, an honest majority of the replies a node receives (N - f, because the liars may stay silent) must still be a majority even if f of those replies are false, which requires N - 2f > f. PBFT (Castro and Liskov, 1999) was the first practical Byzantine algorithm: a leader (the primary) proposes the order, and the replicas exchange three rounds of signed messages (pre-prepare, prepare, commit) so that each one collects 2f + 1 confirmations from the others before executing; if the leader misbehaves, the replicas vote to replace it (view change). The cost is O(N²) messages per decision with a cryptographic signature on each, compared with O(N) for Raft.

When is it needed? When the nodes belong to parties that do not trust each other: permissioned blockchains (Hyperledger Fabric and Tendermint/Cosmos use variants of PBFT), aerospace control systems with redundancy against faulty hardware, or consortia. At Kilometre Zero all the nodes belong to the same organisation and are protected against bugs by testing and against intruders by the security of Module 6; an internal Byzantine failure is handled as an incident, not with Byzantine consensus. Raft is more than enough.

  1. Comparison table: Paxos, Raft and PBFT

Aspect Paxos (Multi-Paxos) Raft PBFT
Failure model Crash (crash-stop / crash-recovery) Crash Byzantine
Nodes to tolerate f failures 2f + 1 2f + 1 3f + 1
Leader Optional in theory; distinguished in Multi-Paxos Mandatory; election built in with randomised timeouts Primary with view change
Messages per decision (steady state) O(N): only phase 2 with a stable leader O(N): one AppendEntries O(N²), signed
Understandability Low; gaps in the specification High; complete specification (membership, snapshots) Medium-low
Who can be leader Anyone (it receives the state in phase 1) Only a node with an up-to-date log Deterministic rotation
Safety guaranteed Always Always Always (with f < N/3)
Termination With partial synchrony and a stable leader With partial synchrony (randomness) With partial synchrony
Used in Chubby, Spanner, Cassandra (LWT), Neo4j etcd, Consul, CockroachDB, TiKV, Kafka KRaft, RabbitMQ quorum queues Hyperledger Fabric, Tendermint, critical systems

  1. Systems that implement it and their use at Kilometre Zero: etcd and the outbox relay

Hardly anybody implements Raft or Paxos for their application: you use a coordination system that encapsulates it and exposes simple primitives on top of a linearizable key-value store:

System Algorithm Primitives Where you see it
etcd Raft Key-value with revisions, leases (TTL), watch, compare-and-swap transactions Kubernetes keeps all its state there
ZooKeeper ZAB (similar to Raft, earlier) Hierarchical znodes, ephemeral and sequential nodes, watches Kafka (until KRaft), HBase, Hadoop
Consul Raft Key-value, sessions, service discovery, health checks HashiCorp service mesh

All three are PC/EC in the table of 03-02: every write goes through consensus and the minority side of a partition rejects writes (and, by default, linearizable reads). They are deployed in clusters of 3 or 5 nodes and used for small, critical data: who the leader is, which configuration is current, which nodes are alive. Never for application data (stock, orders): they are slow by design and their capacity is measured in megabytes, not terabytes.

Kilometre Zero's problem

In 02-05 we wrote the relay for the Outbox pattern: a process that reads from the outbox table of km0_orders with FOR UPDATE SKIP LOCKED and publishes to orders.events. With one instance it works. But orders is deployed with three replicas on Kubernetes (01-06), and if all three run the relay, even though SKIP LOCKED prevents two of them from grabbing the same row at once, the publication order per order is no longer guaranteed (two relays publish rows of the same order in parallel) and a relay that dies after publishing and before marking duplicates events more often. We want a single active instance and, if it dies, another to take over within seconds. That is a leader election, and we will do it with etcd.

We add etcd to the docker-compose.yml:

  etcd:
    image: quay.io/coreos/etcd:v3.5.15
    command: >
      etcd --name etcd0
           --listen-client-urls http://0.0.0.0:2379
           --advertise-client-urls http://etcd:2379
    ports:
      - "2379:2379"

(A single node for development; in production there would be 3 or 5, because a one-node etcd is a single point of failure that tolerates nothing.) The mechanism relies on two etcd primitives:

  • Lease: a contract with a TTL. Keys attached to a lease disappear automatically if the client stops renewing it (keep-alive). If the leader relay dies, its key deletes itself when the TTL expires.
  • Compare-and-swap transaction: "if the key does not exist, create it with my identity and this lease". Since etcd is linearizable, only one of several concurrent instances will see "does not exist" and win.
# km0/services/orders/leader_relay.py
import os
import socket
import time

import etcd3                                   # pip install etcd3
from etcd3.events import DeleteEvent

KEY = "/km0/leader/relay-outbox"
TTL_SECONDS = 10
ME = f"{socket.gethostname()}-{os.getpid()}"

client = etcd3.client(host=os.environ.get("ETCD_HOST", "etcd"), port=2379)


def try_to_become_leader(lease) -> bool:
    """Creates the key only if it does not exist (linearizable compare-and-swap)."""
    succeeded, _ = client.transaction(
        compare=[client.transactions.version(KEY) == 0],           # version 0 = the key does not exist
        success=[client.transactions.put(KEY, ME, lease=lease)],
        failure=[],
    )
    return succeeded


def wait_until_free() -> None:
    value, _ = client.get(KEY)
    print(f"[{ME}] current leader: {value.decode() if value else '(none)'}; waiting")
    events, cancel = client.watch(KEY)
    for event in events:
        if isinstance(event, DeleteEvent):        # the leader's lease expired or was revoked
            break
    cancel()


def publish_pending() -> int:
    """The relay from 02-05: reads outbox FOR UPDATE SKIP LOCKED, publishes to Kafka, sets published_at."""
    ...
    return 0


def loop() -> None:
    while True:
        lease = client.lease(TTL_SECONDS)
        if not try_to_become_leader(lease):
            lease.revoke()
            wait_until_free()
            continue
        print(f"[{ME}] I AM THE LEADER of the outbox relay")
        try:
            while True:
                publish_pending()
                reply = lease.refresh()                 # keep-alive: if it fails, we have lost leadership
                if not reply or reply[0].TTL <= 0:
                    raise RuntimeError("could not renew the lease")
                time.sleep(1)
        except Exception as e:
            print(f"[{ME}] stepping down as leader: {e}")
        finally:
            try:
                lease.revoke()                          # release as soon as possible so another can take over
            except Exception:
                pass


if __name__ == "__main__":
    loop()

How it behaves with three replicas of orders:

  1. All three start up and run try_to_become_leader. etcd, via Raft, processes the three transactions in a total order; the first sees version == 0 and creates the key; the other two see that it already exists and move on to wait_until_free, blocked on a watch (no polling).
  2. The leader publishes every second and renews the lease. If it is redeployed or dies cleanly, revoke deletes the key instantly; if it dies abruptly (kill -9, node down), the key disappears when the lease expires, at most 10 seconds later.
  3. The watch of the other two receives the DeleteEvent, and both call try_to_become_leader again; exactly one wins. The relay has been stopped for between 0 and 10 seconds: events pile up in outbox (nothing is lost) and are published in order when it resumes.

There is one subtlety that is worth understanding properly. If the leader becomes isolated from etcd (a partition) but is still alive and connected to PostgreSQL and Kafka, its lease will expire without it being able to renew it, another instance will take over leadership, and for a few seconds there could be two relays publishing: the new one and the old one, which does not yet know it has lost. The failing refresh that stops the loop bounds that window, but does not eliminate it (between the last successful refresh and expiry there are up to 10 seconds). This is the classic problem of distributed locks and it has two remedies: accept the window because the consumers are idempotent (our case: 02-05 already protects us from duplicates) or use a fencing token: the monotonically increasing mod_revision that etcd assigns to the leader's key is included in every UPDATE outbox ... WHERE published_at IS NULL AND leader_revision <= %s, so that the database rejects writes from an old leader. Distributed locks and fencing will come up again in 07-03 when we talk about failover.

Common Mistakes and Tips

  • Implementing consensus by hand. Paxos and Raft look short on paper and are treacherous in code (the authors of Raft themselves maintain a list of common mistakes in published implementations). Use etcd, ZooKeeper or Consul, or a mature library, and spend your effort on the application.
  • A coordination cluster with an even number of nodes. Four nodes tolerate one failure, the same as three, but at a higher cost and with a greater likelihood of a failure occurring. Use 3 or 5.
  • Using the coordination store as a database. etcd has a practical limit of a few GB and every write goes through Raft and through fsync on the majority. Storing van positions there would bring it down within minutes.
  • A distributed lock with no fencing token on non-idempotent operations. If the effect protected by the lock cannot tolerate a stale executor, lease expiry is not enough; you have to fence at the resource (the database, the store) with the lease revision.
  • Election timeouts that are too short. If the election timeout is of the same order as network latency, any congestion triggers continuous elections (the leader does not manage to send heartbeats in time). Raft recommends that the heartbeat interval be an order of magnitude smaller than the election timeout, and the latter an order of magnitude smaller than the mean time between failures.
  • Confusing "majority of the configured nodes" with "majority of the live nodes". The majority is always over the total number of nodes in the cluster, not over those that respond. A cluster of 5 with 3 down cannot elect a leader with "2 out of 2 alive": there is no majority, and that is correct (there could be another 3 alive on the other side of a partition electing the opposite).
  • Tip: when debugging a Raft-based system, look at the terms. A term that grows quickly means continuous elections: an unstable network, badly tuned timeouts or a node with an overloaded CPU.
  • Tip: document which Kilometre Zero decisions go through consensus (leadership of relays and schedulers, configuration) and which do not (all business data), and why. It is the list a new team member needs before touching etcd.

Exercises

Exercise 1: Paxos with lost messages in phase 1

Using the simulation in section 5, configure the network so that in round A the accept messages from relay-1 to a1 and a3 are lost (only a2 accepts), and in round B the prepare from relay-2 to precisely a2 is lost. Which value does relay-2 propose in phase 2 and which value ends up chosen? Is agreement violated? Run round C and explain the final state.

Exercise 2: Splitting the vote in Raft

Modify the simulation in section 7 so that all three nodes have the same log (index 5) and the timeout range is very narrow, uniform(0.150, 0.151). Run it several times and describe what happens in the first few terms. Is safety ever violated (two leaders in the same term)? What is violated? Go back to the original range and compare.

Exercise 3: Designing the use of etcd

Kilometre Zero needs the delivery service to run a single scheduler that assigns orders to couriers every 30 seconds, and also needs every delivery node to know the list of active markets (Girona, Lleida, Tarragona, Valencia), which an administrator can change. Design the etcd keys, state which primitives you would use for each need (lease, transaction, watch) and explain step by step what happens if the scheduler node is isolated from etcd for 25 seconds with a 10-second lease. Do you need a fencing token? Justify your answer.

Solutions

Solution 1:

Round A: relay-1 obtains three promises (n=11), but only a2 accepts (11, 'relay-1'): there is no majority, so no value is chosen. Round B: relay-2 sends prepare(12) to a1 and a3 only (the one to a2 is lost); both reply promise with no accepted proposal, which is a majority (2 of 3). Since no promise includes anything, relay-2 proposes its own value 'relay-2'. In phase 2 all three accept it: a2 as well, because it never promised anything higher than 11 and an accept(12) exceeds that promise (an acceptor does not need to have received the prepare in order to accept; it only needs not to have promised a higher number). 'relay-2' is chosen and the old (11, 'relay-1') on a2 is overwritten: it was never part of a majority, so it did not matter. Round C: relay-1 sends prepare(21) to all three, all three report (12, 'relay-2'), and relay-1 adopts 'relay-2' and confirms it with n=21. Final state: all of them with (21, 'relay-2'). Agreement is preserved: the value chosen in B is the one confirmed in C. It is the mirror image of the case in section 5: there, the partial proposal of relay-1 "survived" because the only acceptor that held it was in the majority of promises obtained by relay-2; here it was not, and it was discarded. Both outcomes are correct, because in neither case had the partial value been chosen.

Solution 2:

With almost identical timeouts, the three nodes time out practically simultaneously and stand in the same term; each votes for itself and, since there is only one vote per term, none reaches two votes: a split vote. They all reset their deadline (again almost identical) and repeat in the next term, and so on several times; the terms grow quickly with no leader. At some point the small differences in latency (the uniform(0.002, 0.010) of the network) mean that one candidate asks for the vote before another has stood, and it wins. There are never two leaders in the same term (safety holds: one vote per term, and a majority), but liveness is temporarily violated: the system takes a long time to decide. With the original range (150-300 ms), the probability of two nodes timing out within the same latency window is small and the election is usually settled in the first or second term. It is the empirical demonstration of why Raft uses randomised timeouts and of how randomness sidesteps FLP in practice.

Solution 3:

Keys: /km0/leader/delivery-scheduler (value: the identity of the instance, with a 10 s lease) and /km0/config/active-markets (JSON value ["girona","lleida","tarragona","valencia"], with no lease: it is persistent configuration).

Primitives: for the scheduler, the same scheme as in section 10: lease + version == 0 transaction + watch for the deletion; the leader runs the assignment every 30 s and renews the lease every 3 s. For the markets, each delivery node does a get on start-up and keeps a permanent watch on the key; the administrator writes with a normal put (or with a transaction conditional on mod_revision to avoid overwriting concurrent changes); the watch delivers the new value to all nodes within milliseconds, with no polling.

25 s isolation: t=0, last successful renewal. t≈3, the refresh fails; the leader's loop must stop scheduling immediately (not carry on "just in case"). t=10, the lease expires in etcd, the key is deleted, the other instances receive the DeleteEvent and one wins. t=25, the old leader regains its connection, tries to renew a lease that no longer exists, moves on to wait_until_free and stays as a follower. Dangerous window: between t=0 and t≈3 the old leader still legitimately believes it is the leader, but since there is no new leader until t=10, there are never two schedulers at once provided the old one stops when the refresh fails. If the old one did not check the refresh (or if an assignment already in progress takes longer than the remaining TTL), it could indeed overlap with the new one.

Fencing token: assigning an order to a courier is not idempotent by nature (two schedulers could assign the same order to van-3 and to another van). It is prudent to include the mod_revision of the leader's key in the assignment write (UPDATE delivery_orders SET courier = %s, leader_revision = %s WHERE id = %s AND (leader_revision IS NULL OR leader_revision <= %s)), so that an assignment from an old leader is rejected by the database. Alternatively, making the assignment idempotent per order (a single assignment row per order with INSERT ... ON CONFLICT DO NOTHING, as in 02-05) removes the need for the token in this particular case, although it would not stop two schedulers from competing to assign the same courier more than their 8 orders.

Conclusion

Consensus is the problem of getting several nodes to decide on a single value while satisfying agreement, validity and termination, and this lesson has solved it at the three levels announced. FLP reminded us that termination cannot be guaranteed in an asynchronous system, and we saw that the industry's answer is to separate safety (always) from liveness (when the network behaves): with majorities of 2f + 1 nodes, timeouts and randomness. Paxos solves consensus on one value with proposers, acceptors and learners in two phases, and its central rule (adopt the already accepted value with the highest number) is what made relay-2, in the simulation, end up choosing relay-1; Multi-Paxos extends it to a log, leaving gaps that every implementation fills in. Raft covers those gaps with terms, election by randomised timeouts, log replication with majority commit and the restriction that only a node with an up-to-date log can be leader, and the asyncio simulation showed node-3 being denied twice because of its stale log, the leader node-1 crashing and node-2 taking over in about 335 ms, with the term jumping from 2 to 4. Byzantine consensus, with 3f + 1 nodes and PBFT, is for when the participants do not trust each other, which is not the case at Kilometre Zero. And the first practical use has been concrete: etcd with a lease and a compare-and-swap transaction guarantees that only one instance of the outbox relay publishes, with the caveat of the fencing token for the window in which an isolated leader does not yet know it is no longer the leader.

We now know how to name the guarantees (03-01), choose what to sacrifice when a partition strikes (03-02) and get a group of nodes to agree (03-03). What we have not yet done is actually move the data: how a row inserted on the primary of km0_orders reaches its replica, what is lost if the primary dies before it gets there, what happens when two regions accept writes at the same time, and how the read and write quorums work with which Cassandra and DynamoDB offer tunable consistency without a leader. That is the subject of the next lesson, Data Replication, where we will also set up a real PostgreSQL replica in Kilometre Zero's docker-compose.yml and watch it arrive late.

◄ Previous
◄
Next ►
►

Distributed Architectures Course

Module 1: Introduction to Distributed Systems

Module 2: Communication in Distributed Systems

Module 3: Consistency and Replication

Module 4: Distributed Storage

Module 5: Distributed Computing

Module 6: Security in Distributed Systems

Module 7: Monitoring and Maintenance

Module 8: Case Studies and Applications

The Project | About Us | Contribute | Donations | License
© Copyright 2026. All rights reserved