Skip to content

Consensus and coordination

TL;DR

  • Consensus is how nodes agree on one ordered log despite crashes and partitions; leader election, atomic broadcast and configuration all reduce to it.
  • Raft is the algorithm to name: terms, randomized timeouts, majority votes, log replication, commit on a majority.
  • Coordination costs a majority round trip — 500 µs inside a datacenter, 70 ms across US regions — so keep it off the hot path.
  • A distributed lock without a fencing token is not mutual exclusion, whatever its TTL says.

Core concepts

What consensus buys you

Consensus gets nodes to decide on one value when any may crash and any message may be delayed. You rarely want one value; you want a replicated log — atomic broadcast — which is consensus run once per slot. Three uses are worth naming: leader election, so a failover leaves one writer per shard; total order, the log a linearizable store is built from; and configuration, the shard map and live node set that are fatal to disagree about.

A group is 3 or 5 nodes: 2f+1 tolerates f failures because any two majorities intersect. It is not a database — one group commits at a majority round trip plus a durable append, 500 µs inside a datacenter, so it sustains a few thousand commits/s, the order of one relational primary's 5k-20k writes/s. Systems needing more shard across many groups (Spanner runs one per range) or keep consensus off the data path, as Replication describes.

FLP, and why timeouts are the escape hatch

FLP says that in a purely asynchronous system no deterministic algorithm solves consensus if even one node may crash, for the reason you should say out loud: you cannot distinguish a crashed node from a slow one. Systems escape with a failure detector built from timeouts, plus randomness so symmetric candidates do not deadlock — termination becomes overwhelmingly likely, not guaranteed.

The price is liveness, never safety: the minority side of a partition stops accepting writes until it reaches a majority again, but never returns a wrong answer — the CP corner of CAP, PACELC and consistency models. Quantify the outage: with 150-300 ms election timeouts a crash costs one timeout plus one round of votes, and the simulation below replaces the leader 192 ms later (1,360 ms - 1,168 ms). Ten failovers a day is 10 x 0.2 s x 365 = 730 s, 12 minutes a year — inside the 52.6 minutes a 99.99% target allows.

Paxos in one paragraph, Raft in the rest

Single-decree Paxos: a proposer picks an ever-increasing ballot number and asks a majority to promise to ignore lower ballots; if any acceptor already accepted a value, the proposer must re-propose that one. Two majorities intersect, so at most one value is chosen. Multi-Paxos amortizes the promise phase with a stable leader. In the room: "Paxos is the proof, Raft is the implementation."

A Raft term is a logical clock: each begins with an election and has at most one leader. Every message carries a term, and a node seeing a higher one becomes a follower — which is how stale leaders retire.

A follower that stops hearing heartbeats becomes a candidate; a higher term always demotes.

stateDiagram-v2
    [*] --> Follower
    Follower --> Candidate : election timeout with no heartbeat
    Candidate --> Candidate : split vote, retry in a higher term
    Candidate --> Leader : votes from a majority
    Candidate --> Follower : heartbeat from this term's leader
    Leader --> Follower : sees a higher term
    Follower --> Follower : grants a vote and resets its timer

Two rules carry the safety argument. One vote per term, and only for a candidate at least as up to date as you — last log term, then last index. A committed entry sits on a majority and a winner needs a majority, so the sets share a voter who refuses a candidate missing that entry. A leader never edits its own log; it makes followers match — AppendEntries names the preceding entry's index and term, and a follower that disagrees rejects it until the leader backs up far enough. One trap: the commit index may only advance to a majority-held entry of the leader's own term (Figure 8).

A leader crash costs one election timeout plus one round of votes.

sequenceDiagram
    autonumber
    participant N2 as n2 leader of term 1
    participant N1 as n1 follower
    participant N3 as n3 follower
    participant N5 as n5 follower
    N2->>N1: AppendEntries heartbeat term 1
    N2->>N3: AppendEntries heartbeat term 1
    Note over N2: n2 crashes at t=1168 ms
    Note over N3: no heartbeat for 170 ms, its timer fires first
    N3->>N1: RequestVote term 2 last index 2 last term 1
    N3->>N5: RequestVote term 2 last index 2 last term 1
    N1-->>N3: granted, my log is not ahead
    N5-->>N3: granted
    Note over N3: 3 of 5 votes, leader of term 2 at t=1360 ms
    N3->>N1: AppendEntries term 2, stops every other timer
    N3->>N5: AppendEntries term 2

ZooKeeper and etcd: what you actually run

You rarely implement Raft; you run a small, strongly consistent store beside your system and use it for coordination, not data. ZooKeeper gives you a tree of znodes of up to 1 MB. Ephemeral znodes vanish when the client's session ends, which makes them a liveness signal; sequential znodes get a monotonically increasing suffix; watches are one-shot and carry no payload, so you re-arm and re-read after every fire. etcd is the same ideas over a flat keyspace: a global revision, TTL leases, and watches that resume from a revision.

Three idioms. Leader election: each contender creates an ephemeral sequential znode, lowest sequence wins, and each loser watches only its predecessor — watching the parent wakes all N on every change. Membership: the children of one parent are the live set. Configuration: one znode plus a watch.

Watch the budget: every write goes through atomic broadcast, so five nodes handle thousands of writes/s, not hundreds of thousands. Per-request counters belong in a partitioned log.

Leases, fencing tokens and the Redlock debate

A distributed lock has to be a lease: mutual exclusion with an expiry, because the holder may die holding it. Expiry alone is not safe — a stop-the-world garbage collection, a virtual machine migration or a blocked disk can stop a holder past its lease, and it wakes with no idea time passed, still holding an object that says it owns a resource the service has since given away.

The fix is not a longer TTL; no TTL bounds a pause you do not control. It is a fencing token: the service stamps every grant with a number that only increases (ZooKeeper's zxid, etcd's revision, a counter below), the holder sends it with every write, and the resource rejects any token below the highest it has accepted.

The fenced store rejects the woken-up client; an unfenced one takes its write and loses the newer one.

sequenceDiagram
    autonumber
    actor A as Client A
    participant L as Lease service
    actor B as Client B
    participant S as Fenced store
    A->>L: acquire lock on job 42
    L-->>A: lease token 1, expires at t=10 s
    A->>S: write result=A-first with token 1
    S-->>A: ok, highest token is now 1
    Note over A: A stalls for 11 s in a pause it cannot see
    B->>L: acquire lock on job 42 at t=12 s
    L-->>B: lease token 2, the old lease expired
    B->>S: write result=B-first with token 2
    S-->>B: ok, highest token is now 2
    A->>S: write result=A-second with token 1
    S-->>A: rejected, token 1 is older than token 2

That is the Redlock argument: a TTL lock on a majority of N independent Redis nodes. Kleppmann objects that its safety rests on wall-clock timing across machines that pause and clocks that jump, and that it issues no fencing token, while Antirez answers that it is safe under its stated model. Say which you need: efficiency or correctness.

Failure detection: heartbeats, gossip, phi-accrual

You never detect failure, only unresponsiveness. The threshold is a bet both ways: too short and a slow node triggers a spurious election, too long and you sit in a preventable outage.

  • Heartbeats with a fixed timeout, what Raft uses. Inside a datacenter a round trip is 500 µs, so a 100 ms threshold leaves ~200 round trips of slack; across US regions, at 70 ms a round trip, the same threshold is a false-positive generator. Randomize, and keep the interval well under the timeout.
  • Gossip. Each node exchanges membership state with a few random peers, so suspicion spreads in a logarithmic number of rounds and nobody probes all N. Cassandra, Dynamo and Consul use it at thousands of nodes; it converges in seconds — fine for membership, useless for failover.
  • Phi-accrual. Output a suspicion value derived from recent heartbeat intervals and let each caller pick a threshold, so a slower network raises latency without raising suspicion.

Wire the detector to the lease: the takeover threshold must exceed the TTL, or two nodes hold the same resource.

Trade-offs

Mechanism Safe when the holder pauses Failover Write cost Fencing token Operational cost
Embedded Raft group Yes Election, ~200 ms Majority hop, ~500 µs Log index You own it
ZooKeeper ensemble If checked downstream Session timeout Majority hop zxid 3-5 nodes
etcd If checked downstream Lease TTL Majority hop Revision 3-5 nodes
Database advisory lock While the transaction lives Connection loss 5k-20k writes/s Row version None extra
Redis lock, NX with TTL No TTL expiry ~100k ops/s None Trivial
Gossip membership Not a lock Seconds Nothing None Low

Start by asking what breaks if two nodes act at once. If the answer is "we do the work twice and waste money", one Redis key with NX and a TTL is enough. If it is "we corrupt data or double-charge a customer", you need a lease from a consistent store plus a check at the resource — and you should say "fencing token" before the interviewer does.

Prefer coordination you already run: with PostgreSQL, an advisory lock costs one primary round trip at 5k-20k writes/s and no new cluster. Reach for ZooKeeper or etcd when you need ephemeral membership, watches or a leader across processes that share no database. Own a Raft library only when consensus is the product: a metadata service, a store with one group per range. Across regions, know the bill — US east to west pays 70 ms per commit instead of 500 µs, 140x — so put the majority where the writes are.

Python implementation

RaftNode follows the paper's server rules. This is the election path: randomized timer, term bump, the up-to-date check voters apply, and the majority count that promotes.

code/hld/raft_election.py — elections
class RaftNode:
    """One Raft server. Indices are 1-based as in the paper: ``log[i - 1]`` is entry ``i``.

    ``term``, ``voted_for`` and ``log`` are the persistent state. A real server fsyncs them
    before answering any RPC, and ``RaftCluster.restart`` keeps them for the same reason: a
    node that forgot its vote could vote twice in one term and elect two leaders. The rest
    is volatile and rebuilt after a restart.
    """

    def __init__(self, node_id: str, peers: Sequence[str], cluster: RaftCluster) -> None:
        self.id = node_id
        self.peers = tuple(peers)
        self._cluster = cluster
        self.term = 0
        self.voted_for: str | None = None
        self.log: list[LogEntry] = []
        self.role = Role.FOLLOWER
        self.leader_id: str | None = None
        self.commit_index = 0
        self.alive = True
        self._votes: set[str] = set()
        self._next_index: dict[str, int] = {}
        self._match_index: dict[str, int] = {}
        self._heartbeat_due = 0
        self._election_deadline = 0
        self.reset_election_timer()

    @property
    def majority(self) -> int:
        return (len(self.peers) + 1) // 2 + 1

    @property
    def last_log_index(self) -> int:
        return len(self.log)

    @property
    def last_log_term(self) -> int:
        return self.log[-1].term if self.log else 0

    def committed(self) -> list[str]:
        """Commands this node may apply to its state machine."""
        return [entry.command for entry in self.log[: self.commit_index]]

    def reset_election_timer(self) -> None:
        self._election_deadline = self._cluster.now_ms + self._cluster.random_election_timeout()

    def tick(self) -> None:
        """One simulated millisecond: a leader heartbeats, everyone else watches the timer."""
        if not self.alive:
            return
        if self.role is Role.LEADER:
            if self._cluster.now_ms >= self._heartbeat_due:
                self.broadcast_append()
        elif self._cluster.now_ms >= self._election_deadline:
            self.start_election()

    def start_election(self) -> None:
        """The timer fired without a heartbeat: new term, vote for self, ask everyone else."""
        self.term += 1
        self.role = Role.CANDIDATE
        self.voted_for = self.id
        self.leader_id = None
        self._votes = {self.id}
        self.reset_election_timer()
        self._cluster.log_event(f"{self.id} election timeout, candidate in term {self.term}")
        request = RequestVote(self.term, self.id, self.last_log_index, self.last_log_term)
        for peer in self.peers:
            self._cluster.send(self.id, peer, request)
        self._maybe_become_leader()

    def become_follower(self, term: int) -> None:
        """Rule for all servers: a higher term makes this node's term, vote and role stale."""
        if term > self.term:
            self.term = term
            self.voted_for = None
        if self.role is not Role.FOLLOWER:
            self._cluster.log_event(f"{self.id} steps down, follower in term {term}")
        self.role = Role.FOLLOWER
        self.leader_id = None

    def on_message(self, src: str, msg: Message) -> Message | None:
        """Dispatch one delivered message; the reply (if any) is sent back and returned."""
        if not self.alive:
            return None
        if msg.term > self.term:
            self.become_follower(msg.term)
        reply: Message | None = None
        match msg:
            case RequestVote():
                reply = self.handle_request_vote(msg)
            case VoteReply():
                self.handle_vote_reply(src, msg)
            case AppendEntries():
                reply = self.handle_append_entries(msg)
            case AppendReply():
                self.handle_append_reply(src, msg)
        if reply is not None:
            self._cluster.send(self.id, src, reply)
        return reply

    def handle_request_vote(self, msg: RequestVote) -> VoteReply:
        """One vote per term, only for a candidate whose log is at least as up to date as ours."""
        mine = (self.last_log_term, self.last_log_index)
        up_to_date = (msg.last_log_term, msg.last_log_index) >= mine
        granted = msg.term == self.term and self.voted_for in (None, msg.candidate) and up_to_date
        if granted:
            self.voted_for = msg.candidate
            self.reset_election_timer()  # granting a vote counts as hearing from a leader
        return VoteReply(self.term, granted)

    def handle_vote_reply(self, src: str, msg: VoteReply) -> None:
        if self.role is Role.CANDIDATE and msg.term == self.term and msg.granted:
            self._votes.add(src)
            self._maybe_become_leader()

    def _maybe_become_leader(self) -> None:
        if len(self._votes) < self.majority:
            return
        self.role = Role.LEADER
        self.leader_id = self.id
        self._next_index = dict.fromkeys(self.peers, self.last_log_index + 1)
        self._match_index = dict.fromkeys(self.peers, 0)
        self._cluster.record_leader(self.term, self.id, sorted(self._votes))
        self.broadcast_append()  # the first heartbeat stops everyone else's timer

AppendEntries doubles as the heartbeat, its consistency check backs the leader up one entry at a time until the logs meet, and _advance_commit_index refuses an entry from an earlier term.

code/hld/raft_election.py — log replication and commit index
def broadcast_append(self) -> None:
    """Leader: send every follower the entries it is missing, or an empty heartbeat."""
    self._heartbeat_due = self._cluster.now_ms + self._cluster.heartbeat_ms
    for peer in self.peers:
        self._send_append(peer)

def _send_append(self, peer: str) -> None:
    prev_index = self._next_index[peer] - 1
    prev_term = self.log[prev_index - 1].term if prev_index > 0 else 0
    entries = tuple(self.log[prev_index:])
    message = AppendEntries(self.term, self.id, prev_index, prev_term, entries, self.commit_index)
    self._cluster.send(self.id, peer, message)

def submit(self, command: str) -> int:
    """Client write: append to the leader's log and replicate; returns the entry's index."""
    if self.role is not Role.LEADER:
        raise InvalidStateError(f"{self.id} is not the leader")
    self.log.append(LogEntry(self.term, command))
    self.broadcast_append()
    return self.last_log_index

def handle_append_entries(self, msg: AppendEntries) -> AppendReply:
    if msg.term < self.term:
        return AppendReply(self.term, False, 0)
    self.become_follower(msg.term)  # a candidate that hears from this term's leader yields
    self.leader_id = msg.leader
    self.reset_election_timer()
    prev = msg.prev_log_index
    if prev > self.last_log_index or (prev > 0 and self.log[prev - 1].term != msg.prev_log_term):
        return AppendReply(self.term, False, 0)  # consistency check failed: leader backs up
    for offset, entry in enumerate(msg.entries):
        position = prev + offset
        if position < len(self.log) and self.log[position].term != entry.term:
            del self.log[position:]  # conflicting tail: the leader's log wins
        if position >= len(self.log):
            self.log.append(entry)
    last_new = prev + len(msg.entries)
    if msg.leader_commit > self.commit_index:
        self.commit_index = max(self.commit_index, min(msg.leader_commit, last_new))
    return AppendReply(self.term, True, last_new)

def handle_append_reply(self, src: str, msg: AppendReply) -> None:
    if self.role is not Role.LEADER or msg.term != self.term:
        return
    if msg.success:
        self._match_index[src] = max(self._match_index[src], msg.match_index)
        self._next_index[src] = self._match_index[src] + 1
        self._advance_commit_index()
    else:
        self._next_index[src] = max(1, self._next_index[src] - 1)
        self._send_append(src)  # retry one entry earlier until the logs agree

def _advance_commit_index(self) -> None:
    """Commit the highest index a majority holds, counting only entries of the current term
    (paper, Figure 8); older entries are committed indirectly, once a newer one is."""
    for index in range(self.last_log_index, self.commit_index, -1):
        if self.log[index - 1].term != self.term:
            return
        replicas = 1 + sum(1 for peer in self.peers if self._match_index[peer] >= index)
        if replicas >= self.majority:
            self.commit_index = index
            return

RaftCluster is the simulated network: one seeded random.Random for every timeout and delay, a FakeClock for every reading. A run is a pure function of seed and fault schedule, so the tests assert Election Safety over twenty randomized schedules and get the same verdict each time.

code/hld/raft_election.py — the simulated cluster
@dataclass(order=True, frozen=True, slots=True)
class Envelope:
    deliver_at: int
    seq: int
    src: str = field(compare=False)
    dst: str = field(compare=False)
    payload: Message = field(compare=False)


class RaftCluster:
    """The nodes plus a simulated network, advanced one millisecond per ``step``.

    All randomness (election timeouts, message latency) comes from ``random.Random(seed)``
    and all time from a ``FakeClock``, so two runs with the same seed and fault schedule
    produce the same events. Messages are delivered in (time, sequence) order; a message
    whose sender and receiver are in different partition groups, or whose receiver is
    crashed, is dropped at delivery time.
    """

    def __init__(
        self,
        node_ids: Sequence[str],
        seed: int = 42,
        election_timeout_ms: tuple[int, int] = (150, 300),
        heartbeat_ms: int = 50,
        latency_ms: tuple[int, int] = (5, 15),
    ) -> None:
        if not node_ids or len(set(node_ids)) != len(node_ids):
            raise ValidationError("node ids must be non-empty and distinct")
        if not 0 < election_timeout_ms[0] <= election_timeout_ms[1]:
            raise ValidationError("election timeout range must be positive and ordered")
        if not 0 < heartbeat_ms < election_timeout_ms[0]:
            raise ValidationError("heartbeat interval must be shorter than the election timeout")
        if not 0 <= latency_ms[0] <= latency_ms[1]:
            raise ValidationError("latency range must be non-negative and ordered")
        self._rng = random.Random(seed)
        self._clock = FakeClock()
        self._timeout = election_timeout_ms
        self._latency = latency_ms
        self.heartbeat_ms = heartbeat_ms
        self.now_ms = 0
        self._seq = itertools.count()
        self._inflight: list[Envelope] = []
        self._groups: list[frozenset[str]] = []
        self.events: list[str] = []
        self.sent: Counter[str] = Counter()
        self.dropped = 0
        self.leaders_by_term: dict[int, list[str]] = defaultdict(list)
        self.nodes = {nid: RaftNode(nid, [p for p in node_ids if p != nid], self) for nid in node_ids}

    # -- services the nodes call ------------------------------------------------------------
    def random_election_timeout(self) -> int:
        return self._rng.randint(*self._timeout)

    def send(self, src: str, dst: str, msg: Message) -> None:
        self.sent[type(msg).__name__] += 1
        deliver_at = self.now_ms + self._rng.randint(*self._latency)
        heapq.heappush(self._inflight, Envelope(deliver_at, next(self._seq), src, dst, msg))

    def log_event(self, text: str) -> None:
        self.events.append(f"t={self.now_ms:>5} ms  {text}")

    def record_leader(self, term: int, node_id: str, voters: list[str]) -> None:
        self.leaders_by_term[term].append(node_id)
        self.log_event(f"{node_id} wins term {term} with votes from {', '.join(voters)}")

    # -- the simulation loop ----------------------------------------------------------------
    def reachable(self, a: str, b: str) -> bool:
        return not self._groups or any(a in group and b in group for group in self._groups)

    def step(self) -> None:
        self._clock.advance(0.001)
        self.now_ms = round(self._clock.now() * 1000)
        while self._inflight and self._inflight[0].deliver_at <= self.now_ms:
            env = heapq.heappop(self._inflight)
            if self.nodes[env.dst].alive and self.reachable(env.src, env.dst):
                self.nodes[env.dst].on_message(env.src, env.payload)
            else:
                self.dropped += 1
        for node in self.nodes.values():
            node.tick()

    def run(self, ms: int) -> None:
        for _ in range(ms):
            self.step()

    def run_until(self, condition: Callable[[], bool], max_ms: int = 5_000) -> int:
        """Step until ``condition`` holds and return the elapsed ms; raise when it never does."""
        for elapsed in range(1, max_ms + 1):
            self.step()
            if condition():
                return elapsed
        raise SimulationTimeout(f"condition still false after {max_ms} ms")

    def run_until_leader(self, max_ms: int = 5_000) -> str:
        self.run_until(lambda: self.leader() is not None, max_ms)
        leader = self.leader()
        assert leader is not None
        return leader

    def leader(self) -> str | None:
        """The live leader with the highest term; a stale one may linger in a minority partition."""
        leaders = [node for node in self.nodes.values() if node.alive and node.role is Role.LEADER]
        return max(leaders, key=lambda node: node.term).id if leaders else None

    # -- fault injection --------------------------------------------------------------------
    def partition(self, *groups: Iterable[str]) -> None:
        """Split the network into ``groups``; nodes named in none of them form one more group."""
        named = [frozenset(group) for group in groups]
        unknown = frozenset().union(*named) - frozenset(self.nodes)
        if unknown:
            raise ValidationError(f"unknown nodes {sorted(unknown)}")
        rest = frozenset(self.nodes) - frozenset().union(*named)
        self._groups = named + ([rest] if rest else [])
        shown = " | ".join("{" + ", ".join(sorted(group)) + "}" for group in self._groups)
        self.log_event(f"partition {shown}")

    def heal(self) -> None:
        self._groups = []
        self.log_event("partition healed")

    def _node(self, node_id: str) -> RaftNode:
        if node_id not in self.nodes:
            raise NotFoundError(f"no node {node_id!r}")
        return self.nodes[node_id]

    def crash(self, node_id: str) -> None:
        self._node(node_id).alive = False
        self.log_event(f"{node_id} crashes")

    def restart(self, node_id: str) -> None:
        """Back from a crash with its persistent state (term, vote, log) and nothing else."""
        node = self._node(node_id)
        node.alive = True
        node.role = Role.FOLLOWER
        node.leader_id = None
        node.commit_index = 0  # volatile: re-learned from the leader's next heartbeat
        node.reset_election_timer()
        self.log_event(f"{node_id} restarts as follower in term {node.term}")

    def submit(self, command: str, node_id: str | None = None) -> int:
        """Send a client command to ``node_id`` (default: the current leader)."""
        target = node_id if node_id is not None else self.leader()
        if target is None:
            raise InvalidStateError("no leader to submit to")
        return self._node(target).submit(command)

    def election_safety_holds(self) -> bool:
        """Election Safety: at most one leader was elected in any term."""
        return all(len(set(ids)) <= 1 for ids in self.leaders_by_term.values())

uv run python -m hld.raft_election prints:

5 nodes, election timeout 150-300 ms, heartbeat every 50 ms, latency 5-15 ms, seed 42
t=  156 ms  n2 election timeout, candidate in term 1
t=  168 ms  n2 wins term 1 with votes from n1, n2, n5
t= 1168 ms  n2 still leads term 1: 80 heartbeats sent, 0 vote requests (timers kept being reset)
t= 1168 ms  n2 crashes
t= 1338 ms  n3 election timeout, candidate in term 2
t= 1360 ms  n3 wins term 2 with votes from n1, n3, n5
t= 1360 ms  n2 restarts as follower in term 1
t= 1560 ms  x=1, x=2 submitted to n3: committed on 5/5 nodes, n2 caught up after its restart
t= 1560 ms  partition {n2, n3} | {n1, n4, n5}
t= 1733 ms  n1 election timeout, candidate in term 3
t= 1753 ms  n1 wins term 3 with votes from n1, n4, n5
t= 1953 ms  n3 (term 2) accepted x=3 at index 3, commit index still 2: 2/5 replicas; n1 (term 3) committed y=1
t= 1953 ms  partition healed
t= 1962 ms  n3 steps down, follower in term 3
t= 2253 ms  5/5 logs equal ['x=1', 'x=2', 'y=1'], committed on 5/5; x=3 is gone
leaders per term: {1: ['n2'], 2: ['n3'], 3: ['n1']}; election safety holds: True
identical 200 ms timeouts instead: no leader after 2,000 ms, 10 terms of split votes

The stale leader n3 accepted x=3 and never committed it — 2 of 5 is not a majority. The last line is the control: identical timeouts make every node campaign at once, so ten terms pass with no leader at all.

The lock service stamps each lease with a counter that never repeats; renew extends only a lease the caller still holds.

code/hld/fencing_lock.py — leases
@dataclass(frozen=True, slots=True)
class Lease:
    """Proof of holding ``resource`` until ``expires_at``; ``token`` orders every grant ever made."""

    resource: str
    owner: str
    token: int
    expires_at: float

    def remaining(self, now: float) -> float:
        """Seconds left; the holder must stop touching the resource before this reaches zero."""
        return self.expires_at - now


class LeaseLockService:
    """A lock service in the style of Chubby, ZooKeeper or etcd leases.

    ``_lock`` guards ``_leases`` and ``_next_token``. Tokens come from one counter for the
    whole service, so any two grants, on any resources, are strictly ordered, and a store only
    has to remember the highest token it has accepted per key.
    """

    def __init__(self, clock: Clock, ttl: float = 10.0) -> None:
        if ttl <= 0:
            raise ValidationError("ttl must be positive")
        self._clock = clock
        self._ttl = ttl
        self._leases: dict[str, Lease] = {}
        self._next_token = 0
        self._lock = threading.Lock()

    def acquire(self, resource: str, owner: str) -> Lease | None:
        """Grant a lease if ``resource`` is free or its lease has expired; ``None`` while held."""
        if not resource or not owner:
            raise ValidationError("resource and owner must be non-empty")
        with self._lock:
            now = self._clock.now()
            current = self._leases.get(resource)
            if current is not None and current.expires_at > now:
                return None
            self._next_token += 1
            lease = Lease(resource, owner, self._next_token, now + self._ttl)
            self._leases[resource] = lease
            return lease

    def renew(self, lease: Lease) -> Lease | None:
        """The holder's keep-alive: extend a lease that is still current, same token."""
        with self._lock:
            now = self._clock.now()
            current = self._leases.get(lease.resource)
            if current is None or current.token != lease.token or current.expires_at <= now:
                return None  # expired or taken over: the caller has lost the lock
            renewed = Lease(lease.resource, lease.owner, lease.token, now + self._ttl)
            self._leases[lease.resource] = renewed
            return renewed

    def release(self, lease: Lease) -> bool:
        """Free the resource early; a stale lease (token no longer current) releases nothing."""
        with self._lock:
            current = self._leases.get(lease.resource)
            if current is None or current.token != lease.token:
                return False
            del self._leases[lease.resource]
            return True

    def holder(self, resource: str) -> Lease | None:
        """The unexpired lease on ``resource``, if any."""
        with self._lock:
            current = self._leases.get(resource)
            if current is None or current.expires_at <= self._clock.now():
                return None
            return current

The store is where safety is enforced: it keeps the highest token per key and rejects anything older. check_tokens=False drops the check, so the demo replays the scenario both ways.

code/hld/fencing_lock.py — the fenced store
class FencedStore:
    """A storage service that honours fencing tokens.

    ``_lock`` guards ``_values`` and ``rejected``. A write whose token is lower than the
    highest already accepted for that key comes from a client whose lease ended, however
    convinced that client is that it still holds the lock. ``check_tokens=False`` is the
    store without the check, for contrast: it takes the stale write and loses the newer one.
    """

    def __init__(self, check_tokens: bool = True) -> None:
        self._check_tokens = check_tokens
        self._values: dict[str, tuple[int, str]] = {}  # key -> (highest token, value)
        self._lock = threading.Lock()
        self.rejected = 0

    def write(self, key: str, value: str, token: int) -> None:
        with self._lock:
            current = self._values.get(key)
            if self._check_tokens and current is not None and token < current[0]:
                self.rejected += 1
                raise FencingError(
                    f"write to {key!r} with token {token} rejected: token {current[0]} already seen"
                )
            self._values[key] = (token, value)

    def read(self, key: str) -> str | None:
        with self._lock:
            current = self._values.get(key)
            return None if current is None else current[1]

    def token_of(self, key: str) -> int | None:
        with self._lock:
            current = self._values.get(key)
            return None if current is None else current[0]

uv run python -m hld.fencing_lock prints:

lease lock with ttl 10 s; clients A and B; one fenced store
[fenced] t= 0 s  A acquires job:42 -> token 1, expires t=10 s; A writes result=A-first (token 1) ok
[fenced] t= 1 s  B tries job:42 -> busy
[fenced] t=12 s  A has been paused for 11 s; lease expired, B acquires -> token 2; B writes result=B-first ok
[fenced] t=13 s  A wakes up, lease.remaining=-3 s, writes result=A-second with token 1 -> rejected (write to 'result' with token 1 rejected: token 2 already seen)
[fenced]        store: result=B-first (token 2)
[unfenced] t= 0 s  A acquires job:42 -> token 3, expires t=10 s; A writes result=A-first (token 3) ok
[unfenced] t= 1 s  B tries job:42 -> busy
[unfenced] t=12 s  A has been paused for 11 s; lease expired, B acquires -> token 4; B writes result=B-first ok
[unfenced] t=13 s  A wakes up, lease.remaining=-3 s, writes result=A-second with token 3 -> accepted
[unfenced]        store: result=A-second (token 3)
C acquires -> token 5 (tokens are never reused); renew at t=19 s -> expires t=29 s; release -> True; stale release -> False

Same lock service, same pause; only the store differs.

In the interview

Introduce it where it is needed, not as a topic. When you draw a component that must have exactly one writer, price it in the same breath: "Only one scheduler may dispatch a job, so I elect a leader through etcd with a 10 s lease, and every dispatch carries its revision as a fencing token the run table checks."

Phrases that signal depth: "two majorities always intersect, which is where safety comes from"; "the lock is the efficiency optimization, the fencing token is the correctness mechanism".

Why 3 or 5 nodes, and why not 4?

Four nodes need a majority of three, so they survive one failure — the same as three — while making every commit wait for one more fsync. Even sizes buy latency, not availability.

Your Raft leader is in region A and half the clients are in region B.

Every write from B pays the 70 ms cross-region round trip plus the leader's own majority hop. Move the leader to the writes, shard by region, or serve B follower reads.

A client holds a lock, pauses for 30 s and wakes. What stops it corrupting data?

Nothing in the lock — its lease expired, someone else took over, and it cannot tell. The fencing token does: the store rejects any token below the highest it has seen.

How do you elect a leader with ZooKeeper without a herd?

Each contender creates an ephemeral sequential znode and the lowest sequence leads; every other watches only the znode below its own, so a departure wakes one process.

Where would you use gossip instead of a consensus group?

Where the answer may be eventually consistent and N is large: membership, node health, the ring version in a Dynamo-style store. It needs no majority, but gives no unique leader.

Interview tip

Say what you are not coordinating. "Consensus is on the metadata path only — the shard map and the leader lease — and the data path never talks to etcd" shows you treat coordination as a throughput ceiling, and pre-empts the question about that cluster being slow.

Common mistakes

  • Coordinating per request: every request takes a lock or reads configuration from ZooKeeper, so throughput is capped by an ensemble doing thousands of writes/s. Fix: cache it with a watch.
  • Believing a TTL is mutual exclusion: "the lock expires after 30 s, so only one worker runs" becomes two workers the first time a pause outlasts the lease. Fix: a fencing token checked at the resource.
  • Even-sized or oversized groups: 4 nodes tolerate the same single failure as 3 and commit more slowly. Fix: 3 or 5, and add groups rather than members.
  • Identical election timeouts: every follower campaigns at once and splits the vote forever — ten terms, no leader, 2,000 ms in the demo. Fix: randomize over a wide range.
  • Leader reads with no lease check: a partitioned leader still answers from memory. Fix: renew a read lease.

Common mistake

Committing an entry because a majority stores it, without checking it belongs to the current term. That is Figure 8 of the Raft paper: a later leader can still overwrite it, so a client was told its write was durable when it was not.

Self-check

Why can a leader not commit a previous term's entry that a majority holds?

The election restriction only protects entries already committed, so counting replicas is not enough — a future leader could still overwrite it. One current-term commit breaks the circularity.

What does the election restriction rule out?

A candidate missing a committed entry winning. That entry sits on a majority, a winner needs a majority, and the shared node refuses a candidate whose log is behind.

Your lock service is correct and your data is still corrupted. Where do you look?

At the resource. A correct lease can still be followed by an unfenced write from a client paused past its expiry. Check that storage verifies a token on every write path.

Why does a 5-node etcd cluster make a poor work queue?

Every enqueue and dequeue is a majority round trip plus a durable append: thousands per second at best. A partitioned log does the same at ~100 MB/s per broker.

Heartbeats say a node is down but it is only slow. What breaks?

You elect a new leader while the old one still believes it leads, so writes are rejected or duplicated until it sees the higher term. A phi-accrual detector avoids that.