Key-Value Store: Leader Election (Low-Level Design)

In chapter 3 the leader of each shard was appointed by an operator, and when a leader was lost its shard could accept no writes until somebody promoted a follower by hand. Non-functional requirement 1 of chapter 1 allows 21.6 minutes of failed requests in a month, and no procedure that depends on a person noticing a failure, understanding it and typing commands can be relied upon to stay within such a limit. The members of a shard have to choose their own leader.

Stated so, the task sounds modest, and it is in fact the hardest problem in this book. The members must agree on a leader although they can communicate only by messages that may be lost, although any of them may crash and return at any moment with whatever it had on disk, and although none of them can tell a crashed member from one that is merely slow or cut off. The agreement must be such that 2 members are never leaders at once, since each would then accept writes of which the other knew nothing. And whoever is chosen must already hold every write that has ever been acknowledged, since a leader that lacked one would proceed to overwrite it. A group of machines reaching a decision under these conditions is said to reach consensus.

The algorithm used in this chapter is Raft, published by Diego Ongaro and John Ousterhout in 2014 with the stated aim of being understandable, which it is to a degree that its predecessors were not. Raft combines the replication of chapter 3 with a procedure for voting. We first describe the roles of a member and the rule by which it votes, and show why that rule guarantees that every new leader holds every acknowledged record. We then deal with a case that chapter 3 excluded, that of a member whose log differs from its leader’s. Raft is implemented as a library without any network code, tested by a simulation that crashes members and loses messages at random, and then run over gRPC as kvraft. Finally we measure for how long a shard of 3 nodes is unavailable when its leader is killed.

Overview

Every member of a shard runs the same program and is started in the same way, and none is configured as the leader. At any moment each member is in 1 of 3 roles.

Time is divided into terms, numbered from 1. Every term begins with an election and has at most 1 leader. When no candidate obtains a majority, the term has no leader, and another election follows in the next. The implementation keeps the invariants of chapter 3 and adds 3 more.

5. At most 1 member leads any given term.
6. An elected leader holds every record that was acknowledged in an earlier term.
7. A member removes a record from its log only if the record was not committed.

Invariant 5 holds because every member votes at most once in a term and a candidate needs the votes of a majority. Any 2 majorities have a member in common, and for 2 candidates to win the same term that member would have had to vote twice. Invariants 6 and 7 are established in the sections on voting and on the repair of logs.

The messages

Elections add a second call to the protocol of chapter 1, and 2 fields to the existing Append call.

Listing 4.1 The election service in the schema. proto/store.proto

// Sent by a candidate to every other member of the shard.
service Election {
  rpc RequestVote(VoteRequest) returns (VoteResponse);
}

message VoteRequest {
  uint64 term = 1;
  uint32 candidate = 2;
  // The end of the candidate's log, and the term of its last record.
  uint64 last_sequence = 3;
  uint64 last_term = 4;
}

message VoteResponse {
  uint64 term = 1;
  bool granted = 2;
}

Listing 4.2 The replication service, with fields 6 and 7 added. proto/store.proto

// Sent by the leader of a shard to each follower.
service Replication {
  // Appends records to the follower's log. A call with no records is a
  // heartbeat: it tells the follower that the leader is alive.
  rpc Append(AppendRequest) returns (AppendResponse);
}

message AppendRequest {
  // The shard, named by the last position it owns.
  fixed64 shard = 1;
  // The term of the sender. A follower rejects a leader of an older term.
  uint64 term = 2;
  // The sequence of the record that precedes `records` in the leader's log.
  uint64 previous_sequence = 3;
  repeated Record records = 4;
  // Every record up to this sequence is on a majority of the replicas.
  uint64 committed_sequence = 5;
  // Added for leader election. The term of the record at previous_sequence,
  // and the id of the sender.
  uint64 previous_term = 6;
  uint32 leader = 7;
}

message AppendResponse {
  uint64 term = 1;
  // False when the follower's log does not end at `previous_sequence`.
  bool appended = 2;
  // The sequence of the last record in the follower's log.
  uint64 last_sequence = 3;
}

AppendRequest has gained the fields previous_term and leader, under the unused numbers 6 and 7. This is the evolution of a schema as chapter 1 described it. The existing fields have kept their numbers, kvnode, the program of chapter 3, makes no use of the new ones, and a node built from the earlier schema passes over them when it decodes a message.

As in chapter 3, the library has message types of its own.

Listing 4.3 The imports of raft/src/lib.rs. raft/src/lib.rs

use std::collections::BTreeMap;
use std::fs::{self, File};
use std::io::{self, Write};
use std::path::{Path, PathBuf};

use kv::{Change, Condition, Entry, Log, Record, StoreError, apply, check};

Listing 4.4 The messages of the library. raft/src/lib.rs

pub type NodeId = u32;

/// Sent by a leader to a follower.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AppendRequest {
    pub term: u64,
    pub leader: NodeId,
    /// The sequence of the record that precedes `records`, and its term.
    pub previous_sequence: u64,
    pub previous_term: u64,
    pub records: Vec<Record>,
    pub committed_sequence: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AppendAnswer {
    pub term: u64,
    pub from: NodeId,
    pub appended: bool,
    /// When appended: the last sequence that is known to match the leader's
    /// log. When refused: the sequence the leader should continue from.
    pub last_sequence: u64,
}

/// Sent by a candidate to every other member.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VoteRequest {
    pub term: u64,
    pub candidate: NodeId,
    /// The end of the candidate's log, and the term of its last record.
    pub last_sequence: u64,
    pub last_term: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VoteAnswer {
    pub term: u64,
    pub from: NodeId,
    pub granted: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Message {
    Append(AppendRequest),
    Appended(AppendAnswer),
    Vote(VoteRequest),
    Voted(VoteAnswer),
}

/// A message and the member it is addressed to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outgoing {
    pub to: NodeId,
    pub message: Message,
}

The state of a member

Listing 4.5 The roles, the election timeout, and the state of a member. raft/src/lib.rs

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
    Follower,
    Candidate,
    Leader,
}

/// The ticks a follower waits for a leader before it stands for election:
/// 10 to 19, different for each member and each term, so that 2 members
/// seldom stand at the same moment. A tick is 100 ms in the node.
fn election_timeout(id: NodeId, term: u64) -> u64 {
    10 + (u64::from(id) * 7 + term * 3) % 10
}

/// What the leader knows about one follower.
#[derive(Debug, Clone, Copy, Default)]
struct Progress {
    /// The sequence after which the next message to the follower begins.
    next: u64,
    /// The last sequence the follower has confirmed to match.
    matched: u64,
    /// The tick at which the follower last answered.
    heard: u64,
}

pub struct Member {
    id: NodeId,
    peers: Vec<NodeId>,
    directory: PathBuf,
    sync: bool,

    log: Log,
    /// Every record of the log. Record `n` is at index `n - 1`.
    records: Vec<Record>,
    /// The term in which each record was written.
    terms: Vec<u64>,
    entries: BTreeMap<String, Entry>,
    committed: u64,

    /// The current term, and the member this one voted for in it. Both are
    /// stored on disk before any answer that depends on them is sent.
    term: u64,
    voted_for: Option<NodeId>,

    role: Role,
    /// The leader of the current term, when this member knows it.
    leader: Option<NodeId>,
    votes: usize,
    progress: BTreeMap<NodeId, Progress>,
    /// Ticks since the member started, and the tick of the last message
    /// from a leader or the last vote granted.
    now: u64,
    quiet_since: u64,
    /// The records this member has removed from its log because they
    /// differed from a leader's.
    discarded: u64,
}

A member holds 3 kinds of state, which differ in whether they have to survive a restart, and the distinction is among the most important things to understand about the algorithm.

The second row is easily overlooked. It is natural to think of a vote as a passing event, of no interest once the election is over. But the whole guarantee of invariant 5 rests on each member voting once, and a member that crashed after voting and returned with no memory of having done so could give a second vote in the same term to a different candidate. A vote is therefore a promise, and it is written to disk before it is sent.

Member::open reads both files. From the log it obtains the records and, by means of the term markers described below, the term of each. From the state file it obtains the term and the vote. Everything else begins from fixed values: a member that has been opened is a follower, knows of no leader, and has applied nothing.

Listing 4.6 Opening a member. raft/src/lib.rs

impl Member {
    /// Opens the member in `directory`. It starts as a follower, with
    /// nothing applied, in the term it had reached before it stopped.
    pub fn open(
        directory: &Path,
        id: NodeId,
        peers: Vec<NodeId>,
        sync: bool,
    ) -> io::Result<Member> {
        let (log, recovery) = Log::open(directory, sync)?;
        let (term, voted_for) = read_state(directory)?;
        let terms = terms_of(&recovery.records, 0);

        Ok(Member {
            id,
            peers,
            directory: directory.to_owned(),
            sync,
            log,
            records: recovery.records,
            terms,
            entries: BTreeMap::new(),
            committed: 0,
            term,
            voted_for,
            role: Role::Follower,
            leader: None,
            votes: 0,
            progress: BTreeMap::new(),
            now: 0,
            quiet_since: 0,
            discarded: 0,
        })
    }

    pub fn id(&self) -> NodeId {
        self.id
    }

    pub fn role(&self) -> Role {
        self.role
    }

    pub fn term(&self) -> u64 {
        self.term
    }

    pub fn leader(&self) -> Option<NodeId> {
        self.leader
    }

    pub fn last_sequence(&self) -> u64 {
        self.records.len() as u64
    }

    pub fn committed(&self) -> u64 {
        self.committed
    }

    pub fn records(&self) -> &[Record] {
        &self.records
    }

    pub fn discarded(&self) -> u64 {
        self.discarded
    }

    fn last_term(&self) -> u64 {
        self.terms
            .last()
            .copied()
            .unwrap_or(0)
    }

    /// The term of record `sequence`, or 0 for the position before the log.
    fn term_at(&self, sequence: u64) -> u64 {
        if sequence == 0 {
            0
        } else {
            self.terms[sequence as usize - 1]
        }
    }

    fn majority(&self) -> usize {
        let members = self.peers.len() + 1;

        members / 2 + 1
    }
}

The state file is replaced by writing a new file and renaming it over the old one. On POSIX file systems, a rename within a directory is atomic, so that after a crash the file holds either the old state or the new and never a mixture. A member writes this file before it sends any message that depends on a new term or a new vote.

Listing 4.7 Reading and writing the term and the vote. raft/src/lib.rs

const STATE_FILE: &str = "state";

/// Reads the term and the vote. A member that has no state file has not
/// voted and is in term 0.
fn read_state(directory: &Path) -> io::Result<(u64, Option<NodeId>)> {
    let text = match fs::read_to_string(directory.join(STATE_FILE)) {
        Ok(text) => text,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok((0, None)),
        Err(error) => return Err(error),
    };

    let mut fields = text
        .split_whitespace()
        .map(str::parse::<u64>);
    let invalid = || io::Error::new(io::ErrorKind::InvalidData, "the state file is malformed");
    let term = fields
        .next()
        .and_then(Result::ok)
        .ok_or_else(invalid)?;
    let voted_for = fields
        .next()
        .and_then(Result::ok)
        .map(|id| id as NodeId);

    Ok((term, voted_for))
}

impl Member {
    /// Stores the term and the vote. The new file replaces the old one in a
    /// single step, so a crash leaves one of the 2 and not a mixture.
    fn save_state(&self) -> io::Result<()> {
        let temporary = self.directory.join("state.new");
        let vote = self
            .voted_for
            .map_or(String::new(), |id| id.to_string());
        let mut file = File::create(&temporary)?;

        file.write_all(format!("{} {vote}\n", self.term).as_bytes())?;

        if self.sync {
            file.sync_all()?;
        }

        fs::rename(&temporary, self.directory.join(STATE_FILE))?;

        if self.sync {
            File::open(&self.directory)?.sync_all()?;
        }

        Ok(())
    }
}

The term of a record

Raft needs to know in which term each record was written. The record format of chapter 2 has no field for it, and the implementation records terms without altering that format, by means of a term marker: the first record that a new leader writes is a Put with an empty key and the leader’s term as its value. Clients cannot write an empty key, since the API requires at least 1 character, so that a marker cannot collide with any client’s data. The term of any record is then the term of the most recent marker at or before it.

Listing 4.8 Term markers, and the term of each record of a log. raft/src/lib.rs

/// The key of a term marker. A client cannot write the empty key, so the
/// log can use it for a record of its own: the first record a new leader
/// appends, whose value is the leader's term.
const MARKER: &str = "";

fn marker(term: u64) -> Change {
    Change::Put {
        key: MARKER.to_owned(),
        value: term.to_le_bytes().to_vec(),
    }
}

/// The term a record starts, if it is a term marker.
fn marked_term(record: &Record) -> Option<u64> {
    let Change::Put { key, value } = &record.change else {
        return None;
    };

    if key != MARKER {
        return None;
    }

    let term: [u8; 8] = value.as_slice().try_into().ok()?;

    Some(u64::from_le_bytes(term))
}

/// The term of each record: the term of the last marker at or before it.
fn terms_of(records: &[Record], mut term: u64) -> Vec<u64> {
    let mut terms = Vec::with_capacity(records.len());

    for record in records {
        term = marked_term(record).unwrap_or(term);
        terms.push(term);
    }

    terms
}

The markers have a second function, which is described under the commit rule.

Elections

The election timeout

A follower expects a message from its leader at least once in every tick of 100 ms. Having received none for the whole of its election timeout, it concludes that the leader has failed and stands for election.

If all followers used the same timeout, they would all stand at the same moment, each would vote for itself, and no candidate would obtain a majority. The timeouts are therefore made to differ. election_timeout, in the listing above, computes a value between 10 and 19 ticks from the id of the member and its term. Within a term, the members of a shard obtain different values, and each member obtains a different value in each term, so that 2 members which happen to coincide once do not coincide again. Raft as published uses random timeouts for this purpose. Timeouts that are computed make the simulation later in the chapter repeatable.

On reaching its timeout, a member calls stand. It increases its term, records a vote for itself, and stores both on disk before any request leaves. Every VoteRequest carries the end of the candidate’s log, which the voting rule below compares with the log of each voter.

Listing 4.9 Standing for election. raft/src/lib.rs

impl Member {
    /// Stands for election in the next term.
    fn stand(&mut self) -> io::Result<Vec<Outgoing>> {
        self.term += 1;
        self.voted_for = Some(self.id);
        self.save_state()?;

        self.role = Role::Candidate;
        self.leader = None;
        self.votes = 1;
        self.quiet_since = self.now;

        let request = VoteRequest {
            term: self.term,
            candidate: self.id,
            last_sequence: self.last_sequence(),
            last_term: self.last_term(),
        };

        Ok(self
            .peers
            .iter()
            .map(|&to| Outgoing {
                to,
                message: Message::Vote(request),
            })
            .collect())
    }
}

The voting rule

A member grants its vote to a candidate when 3 conditions are satisfied.

Listing 4.10 Granting a vote, and counting the votes received. raft/src/lib.rs

impl Member {
    /// A member grants its vote once in a term, and only to a candidate
    /// whose log is at least as complete as its own.
    fn on_vote(&mut self, request: VoteRequest) -> io::Result<VoteAnswer> {
        if request.term > self.term {
            self.step_down(request.term)?;
        }

        let complete =
            (request.last_term, request.last_sequence) >= (self.last_term(), self.last_sequence());
        let free = self
            .voted_for
            .is_none_or(|id| id == request.candidate);
        let granted = request.term == self.term && complete && free;

        if granted {
            self.voted_for = Some(request.candidate);
            self.save_state()?;
            self.quiet_since = self.now;
        }

        Ok(VoteAnswer {
            term: self.term,
            from: self.id,
            granted,
        })
    }

    fn on_voted(&mut self, answer: VoteAnswer) -> io::Result<Vec<Outgoing>> {
        if answer.term > self.term {
            self.step_down(answer.term)?;

            return Ok(Vec::new());
        }

        if self.role != Role::Candidate || answer.term != self.term || !answer.granted {
            return Ok(Vec::new());
        }

        self.votes += 1;

        if self.votes >= self.majority() {
            self.lead()
        } else {
            Ok(Vec::new())
        }
    }
}

The third condition establishes invariant 6, by the argument that chapter 1 gave in outline. An acknowledged record is stored on a majority of the members, and a candidate needs the votes of a majority. Since 2 majorities share at least 1 member, some voter holds the record. That voter grants its vote only to a candidate whose log ends at least as far as its own, and such a candidate holds the record as well. A candidate that lacks an acknowledged record therefore cannot collect a majority. The choice that an operator made in chapter 3, by reading 2 numbers and promoting the follower with the longer log, is here made by the members themselves, each refusing its vote to anyone who knows less than it does.

In on_vote, the 3 conditions are request.term == self.term, free and complete. Tuples in Rust are compared by their first elements, with the second consulted only in a tie, so that (last_term, last_sequence) >= (...) is precisely the comparison that the rule requires. A vote that is granted is written to disk before any answer is returned. on_voted counts the votes for a candidate, and the vote that completes a majority triggers lead.

Terms are compared before sequences because a longer log is not necessarily a more complete one. In figure 4.4, a member that had been isolated might have accumulated several uncommitted records of term 1, and its log would then be longer than that of member 2. A record of term 2 at the end of member 2’s log shows that member 2 has received records from a later leader, and its log takes precedence.

Becoming the leader

lead does 3 things. For every follower it records where to begin sending, namely at the end of its own log, as a first guess that a refusal will correct. It then appends the marker of its term, and finally sends its first messages.

Listing 4.11 Becoming the leader. raft/src/lib.rs

impl Member {
    /// Becomes the leader of the current term. The first record of the term
    /// is its marker. When a majority has stored the marker, every earlier
    /// record is committed with it.
    fn lead(&mut self) -> io::Result<Vec<Outgoing>> {
        self.role = Role::Leader;
        self.leader = Some(self.id);

        let next = self.last_sequence();
        self.progress = self
            .peers
            .iter()
            .map(|&peer| {
                (
                    peer,
                    Progress {
                        next,
                        matched: 0,
                        heard: self.now,
                    },
                )
            })
            .collect();

        let record = self.log.append(marker(self.term))?;
        self.records.push(record);
        self.terms.push(self.term);

        Ok(self.replicate())
    }
}

Returning to follower

Every message carries the term of its sender. On seeing, in any message, a term higher than its own, a member calls step_down: it adopts that term, clears its vote, since the vote belonged to an earlier term, and continues as a follower. For a leader, this is how the arrival of a successor makes itself known.

Listing 4.12 Adopting a higher term. raft/src/lib.rs

impl Member {
    /// Moves to a newer term as a follower. The vote of the old term does
    /// not carry over.
    fn step_down(&mut self, term: u64) -> io::Result<()> {
        if term > self.term {
            self.term = term;
            self.voted_for = None;
            self.save_state()?;
        }

        self.role = Role::Follower;
        self.leader = None;
        self.progress.clear();

        Ok(())
    }
}

Repairing a log that differs

In chapter 3 the log of a follower was always a prefix of its leader’s, and a former leader had to rejoin with an empty log. With elections, a member can hold records that its present leader lacks. This comes about when a leader appends a record and fails before sending it. The record exists on 1 member only and was never acknowledged, and the new leader writes a different record at the same sequence.

Append detects such a difference through the field previous_term. A follower appends records only if its own record at previous_sequence has the term previous_term. The check is sufficient on account of a single property: if 2 logs contain a record with the same sequence and the same term, the 2 logs are identical up to that record. The property holds because each term has 1 leader, a leader writes each sequence once, and the check is applied on every append, so that agreement at 1 position has always been built upon agreement at the position before.

On the follower’s side, the handling of Append consists of 3 checks followed by the repair itself.

The third check carries a hint for the leader. When its record at previous_sequence has a different term, the member answers with previous_sequence - 1, and the leader begins its next message 1 record earlier. By repeating this, the leader moves backwards through its log until it arrives at a record on which the 2 logs agree.

Listing 4.13 The checks of Append. raft/src/lib.rs

impl Member {
    fn on_append(&mut self, request: AppendRequest) -> io::Result<AppendAnswer> {
        let refuse = |member: &Member, last_sequence| AppendAnswer {
            term: member.term,
            from: member.id,
            appended: false,
            last_sequence,
        };

        if request.term < self.term {
            return Ok(refuse(self, self.last_sequence()));
        }

        // A message of the current term or a newer one comes from the leader.
        self.step_down(request.term)?;
        self.leader = Some(request.leader);
        self.quiet_since = self.now;

        if request.previous_sequence > self.last_sequence() {
            return Ok(refuse(self, self.last_sequence()));
        }

        if self.term_at(request.previous_sequence) != request.previous_term {
            return Ok(refuse(self, request.previous_sequence - 1));
        }

        let matched = request.previous_sequence + request.records.len() as u64;
        self.store(
            request.previous_sequence,
            request.previous_term,
            request.records,
        )?;
        self.commit(
            request
                .committed_sequence
                .min(matched),
        );

        Ok(AppendAnswer {
            term: self.term,
            from: self.id,
            appended: true,
            last_sequence: matched,
        })
    }
}

Once all the checks have passed, the 2 logs are known to be equal up to previous_sequence, and store makes the member’s log equal to the leader’s for every record that follows.

Listing 4.14 Skipping equal records, removing differing ones, and appending the rest. raft/src/lib.rs

impl Member {
    /// Makes the log equal to the leader's from `previous` onward: records
    /// the member already has are skipped, a record that differs is removed
    /// together with everything after it, and the rest is appended.
    fn store(&mut self, previous: u64, previous_term: u64, records: Vec<Record>) -> io::Result<()> {
        let terms = terms_of(&records, previous_term);
        let mut fresh = 0;

        for (index, record) in records.iter().enumerate() {
            let sequence = previous + index as u64 + 1;

            if sequence > self.last_sequence() {
                break;
            }

            if self.term_at(sequence) != terms[index]
                || &self.records[sequence as usize - 1] != record
            {
                self.truncate(sequence - 1)?;

                break;
            }

            fresh = index + 1;
        }

        if fresh < records.len() {
            self.log
                .append_records(&records[fresh..])?;
            self.records
                .extend_from_slice(&records[fresh..]);
            self.terms
                .extend_from_slice(&terms[fresh..]);
        }

        Ok(())
    }
}

For each incoming record, store computes a term from previous_term and from the markers among the incoming records, exactly as terms_of does for a whole log. 2 records count as equal only if both their terms and their contents agree.

Listing 4.15 Removing uncommitted records. raft/src/lib.rs

impl Member {
    /// Removes the records after `keep`. Only records that are not
    /// committed can differ from the leader's, so none of them is applied.
    fn truncate(&mut self, keep: u64) -> io::Result<()> {
        if keep < self.committed {
            return Err(io::Error::other(format!(
                "record {} is committed and differs from the leader's",
                keep + 1
            )));
        }

        let length: usize = self.records[..keep as usize]
            .iter()
            .map(|record| record.encode().len())
            .sum();

        self.discarded += self.last_sequence() - keep;
        self.log
            .truncate(keep, length as u64)?;
        self.records.truncate(keep as usize);
        self.terms.truncate(keep as usize);

        Ok(())
    }
}

truncate refuses to remove a committed record and returns an error. By invariant 6 the situation cannot arise, since a leader holds every committed record and its log cannot differ from a follower’s at a committed position. The check is there nonetheless, and for a good reason. Should the invariant ever be violated, through a defect in this code or a fault in a disk, the result without the check would be acknowledged data silently destroyed, and with it the member stops. A program that verifies what it believes to be impossible converts its own undiscovered defects from corruption into failures that somebody will notice.

In order to truncate its file, a member needs the byte offset at which the last record to be kept ends. It computes that offset by encoding the records that are kept and summing their lengths, which avoids storing an offset for every record.

The log of chapter 2 gains a method for removing records from its end.

Listing 4.16 The imports of kv/src/log.rs. kv/src/log.rs

use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::Path;

use crate::record::{Change, Decoded, Record};

Listing 4.17 Removing records from the end of the log. kv/src/log.rs

impl Log {
    /// Removes the records after record `keep`, which ends at byte `length`
    /// of the file. A follower uses this to discard records that were never
    /// committed and that differ from the leader's.
    pub fn truncate(&mut self, keep: u64, length: u64) -> io::Result<()> {
        self.file.set_len(length)?;

        if self.sync {
            self.file.sync_all()?;
        }

        self.next = keep + 1;

        Ok(())
    }
}

This method qualifies invariant 4 of chapter 2, which stated that the log is modified only by appending. Stated exactly, no committed record is ever modified. Records beyond the committed sequence may be removed, and invariant 7 of this chapter ensures that only such records are.

The leader’s view of a follower

For every follower, the leader keeps 2 sequences. next is where its next message will begin, and it moves backwards on a refusal. matched is the highest sequence that the follower has confirmed, and it only ever increases. The commit rule uses matched, since next is a guess. on_appended updates both from an answer, applies the commit rule, and returns a further message when the follower is still behind.

Listing 4.18 Processing the answer of a follower. raft/src/lib.rs

impl Member {
    fn on_appended(&mut self, answer: AppendAnswer) -> io::Result<Vec<Outgoing>> {
        if answer.term > self.term {
            self.step_down(answer.term)?;

            return Ok(Vec::new());
        }

        if self.role != Role::Leader || answer.term != self.term {
            return Ok(Vec::new());
        }

        let now = self.now;
        let Some(progress) = self.progress.get_mut(&answer.from) else {
            return Ok(Vec::new());
        };

        progress.heard = now;
        progress.next = answer.last_sequence;

        if answer.appended {
            progress.matched = progress
                .matched
                .max(answer.last_sequence);
        }

        let behind = !answer.appended || answer.last_sequence < self.last_sequence();
        self.advance();

        Ok(if behind {
            self.append_for(answer.from)
                .into_iter()
                .collect()
        } else {
            Vec::new()
        })
    }
}

The commit rule

Listing 4.19 Applying committed records, and the leader's commit rule. raft/src/lib.rs

impl Member {
    /// Applies the records up to `sequence` that are not applied yet. Term
    /// markers are records of the log and are not entries.
    fn commit(&mut self, sequence: u64) {
        let until = sequence.min(self.last_sequence());

        // A record that marks the start of a term changes no key.
        let changes: Vec<Record> = (self.committed..until)
            .map(|index| self.records[index as usize].clone())
            .filter(|record| marked_term(record).is_none())
            .collect();

        for record in changes {
            apply(&mut self.entries, record);
        }

        self.committed = self.committed.max(until);
    }

    /// The leader's commit rule. A record of an earlier term is not counted
    /// on its own: it becomes committed together with the first record of
    /// the leader's own term that a majority has stored.
    fn advance(&mut self) {
        let mut stored: Vec<u64> = self
            .progress
            .values()
            .map(|progress| progress.matched)
            .collect();
        stored.push(self.last_sequence());
        stored.sort_unstable_by(|a, b| b.cmp(a));

        let on_majority = stored[self.majority() - 1];

        if on_majority > self.committed && self.term_at(on_majority) == self.term {
            self.commit(on_majority);
        }
    }
}

The commit rule of chapter 3 committed a record as soon as a majority had stored it. advance adds a condition, that the record at the majority position must belong to the leader’s own term, and the reason for it is the subtlest point in Raft. A record of an earlier term can still be replaced even when a majority holds it, if a member that lacks it comes to be elected by members that lack it too, which remains possible for as long as no record of the current term is stored on a majority. Counting replicas is therefore safe only for records of the current term. Records of earlier terms are committed indirectly, at the moment when a record of the current term at a later sequence is committed.

It is here that the markers perform their second function. A new leader appends a marker as the first record of its term, and as soon as a majority has stored it, the marker and every record before it are committed together. Without markers, the records inherited from earlier terms would remain uncommitted until the first write of a client in the new term happened to arrive.

The clock and the messages

The library reads no clock and opens no connection. Its caller invokes tick at a fixed interval and handle for every incoming message, and sends whatever messages either method returns. As in chapter 3, this separation is what makes the algorithm testable, since a test can then supply the clock and the network itself.

Listing 4.20 The clock: heartbeats for a leader, the election timeout for the others. raft/src/lib.rs

impl Member {
    /// Advances the member's clock by one tick. A leader sends every
    /// follower a message. A follower or a candidate that has waited for
    /// its election timeout stands for election.
    pub fn tick(&mut self) -> io::Result<Vec<Outgoing>> {
        self.now += 1;

        if self.role == Role::Leader {
            return Ok(self.replicate());
        }

        if self.now - self.quiet_since >= election_timeout(self.id, self.term) {
            return self.stand();
        }

        Ok(Vec::new())
    }

    /// A message for every follower: the records it lacks, or a heartbeat.
    pub fn replicate(&self) -> Vec<Outgoing> {
        self.peers
            .iter()
            .filter_map(|&peer| self.append_for(peer))
            .collect()
    }

    fn append_for(&self, peer: NodeId) -> Option<Outgoing> {
        let from = self
            .progress
            .get(&peer)?
            .next
            .min(self.last_sequence());

        let request = AppendRequest {
            term: self.term,
            leader: self.id,
            previous_sequence: from,
            previous_term: self.term_at(from),
            records: self.records[from as usize..].to_vec(),
            committed_sequence: self.committed,
        };

        Some(Outgoing {
            to: peer,
            message: Message::Append(request),
        })
    }
}

Listing 4.21 Dispatching a message. raft/src/lib.rs

impl Member {
    /// Processes a message and returns the messages to send in response.
    pub fn handle(&mut self, message: Message) -> io::Result<Vec<Outgoing>> {
        match message {
            Message::Append(request) => {
                let to = request.leader;
                let answer = self.on_append(request)?;

                Ok(vec![Outgoing {
                    to,
                    message: Message::Appended(answer),
                }])
            }
            Message::Vote(request) => {
                let to = request.candidate;
                let answer = self.on_vote(request)?;

                Ok(vec![Outgoing {
                    to,
                    message: Message::Voted(answer),
                }])
            }
            Message::Appended(answer) => self.on_appended(answer),
            Message::Voted(answer) => self.on_voted(answer),
        }
    }
}

Listing 4.22 The operations for client requests. raft/src/lib.rs

impl Member {
    /// Whether this member is the leader and every record of its log is
    /// committed, so that it can accept a write.
    pub fn is_settled(&self) -> bool {
        self.role == Role::Leader && self.committed == self.last_sequence()
    }

    /// Checks the condition of a write and appends its record to the log.
    /// The caller sends the messages of `replicate` afterwards.
    pub fn propose(&mut self, change: Change, condition: Condition) -> Result<u64, StoreError> {
        let key = match &change {
            Change::Put { key, .. } | Change::Delete { key } => key,
        };
        let current = self
            .entries
            .get(key)
            .map(|entry| entry.version);

        if matches!(change, Change::Delete { .. }) && current.is_none() {
            return Err(StoreError::NotFound);
        }

        check(condition, current)?;

        let record = self.log.append(change)?;
        let sequence = record.sequence;
        self.records.push(record);
        self.terms.push(self.term);

        Ok(sequence)
    }

    /// Whether this member may answer a read: it is the leader, the marker
    /// of its term is committed, and a majority of the members, itself
    /// included, has answered within the last 10 ticks.
    pub fn can_read(&self) -> bool {
        let recent = self
            .progress
            .values()
            .filter(|progress| self.now - progress.heard < 10)
            .count();

        self.role == Role::Leader
            && self.term_at(self.committed) == self.term
            && recent + 1 >= self.majority()
    }

    pub fn get(&self, key: &str) -> Option<&Entry> {
        self.entries.get(key)
    }
}

can_read permits a read when the member is the leader, when the marker of its term is committed, and when a majority of the members has answered within the last 10 ticks. The second of these conditions ensures that the entries reflect every record that was committed in earlier terms.

Simulating crashes and lost messages

Because the library is free of clocks and connections, a test has control of all of them. raftsim runs 3 members in a single process. It decides when the clock of each member ticks, which message is delivered next, which messages are lost, and when a member crashes and restarts.

Listing 4.23 The imports of raftsim/src/main.rs. raftsim/src/main.rs

use std::collections::BTreeMap;
use std::env;
use std::error::Error;
use std::fs;
use std::path::{Path, PathBuf};
use std::process;

use kv::{Change, Condition};
use raft::{Member, NodeId, Outgoing, Role};

Listing 4.24 A repeatable source of pseudo-random numbers. raftsim/src/main.rs

/// A generator of pseudo-random numbers. The same seed gives the same
/// numbers, so a run that fails can be repeated exactly.
struct Random(u64);

impl Random {
    fn next(&mut self) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;

        self.0
    }

    /// A number below `bound`.
    fn below(&mut self, bound: usize) -> usize {
        (self.next() % bound as u64) as usize
    }
}

Listing 4.25 The cluster: members, messages in transit, and writes. raftsim/src/main.rs

const MEMBERS: [NodeId; 3] = [1, 2, 3];

/// A write that a leader has accepted and not yet committed.
struct Pending {
    leader: NodeId,
    term: u64,
    sequence: u64,
    key: String,
}

struct Cluster {
    root: PathBuf,
    /// The members that are running. A crashed member is absent.
    up: BTreeMap<NodeId, Member>,
    /// The messages that are on their way.
    network: Vec<Outgoing>,
    pending: Vec<Pending>,
    /// The sequence and the key of every acknowledged write.
    acknowledged: Vec<(u64, String)>,
    /// The leader that was seen in each term.
    leaders: BTreeMap<u64, NodeId>,
    discarded: u64,
    writes: u64,
}

impl Cluster {
    fn open(root: &Path) -> Result<Cluster, Box<dyn Error>> {
        let mut cluster = Cluster {
            root: root.to_owned(),
            up: BTreeMap::new(),
            network: Vec::new(),
            pending: Vec::new(),
            acknowledged: Vec::new(),
            leaders: BTreeMap::new(),
            discarded: 0,
            writes: 0,
        };

        for id in MEMBERS {
            cluster.start(id)?;
        }

        Ok(cluster)
    }

    fn start(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        let peers = MEMBERS
            .iter()
            .copied()
            .filter(|&peer| peer != id)
            .collect();
        let member = Member::open(
            &self
                .root
                .join(format!("member-{id}")),
            id,
            peers,
            false,
        )?;
        self.up.insert(id, member);

        Ok(())
    }

    /// Stops a member. Its files remain, and its memory is lost.
    fn crash(&mut self, id: NodeId) {
        if let Some(member) = self.up.remove(&id) {
            self.discarded += member.discarded();
        }
    }

    fn tick(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        if let Some(member) = self.up.get_mut(&id) {
            self.network.extend(member.tick()?);
        }

        self.check()
    }

    /// Delivers the message at `index`. A message for a member that is
    /// down is lost.
    fn deliver(&mut self, index: usize) -> Result<(), Box<dyn Error>> {
        let Outgoing { to, message } = self.network.remove(index);

        if let Some(member) = self.up.get_mut(&to) {
            self.network
                .extend(member.handle(message)?);
        }

        self.check()
    }

    /// Delivers one of the messages that are on their way, chosen at random.
    fn deliver_any(&mut self, random: &mut Random) -> Result<(), Box<dyn Error>> {
        if self.network.is_empty() {
            return Ok(());
        }

        self.deliver(random.below(self.network.len()))
    }

    /// Loses one of the messages that are on their way.
    fn lose_any(&mut self, random: &mut Random) {
        if !self.network.is_empty() {
            self.network
                .remove(random.below(self.network.len()));
        }
    }

    /// Starts a member that is down.
    fn restart(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        if self.up.contains_key(&id) {
            return Ok(());
        }

        self.start(id)
    }

    /// Delivers messages until none is on its way.
    fn deliver_all(&mut self) -> Result<(), Box<dyn Error>> {
        while !self.network.is_empty() {
            self.deliver(0)?;
        }

        Ok(())
    }

    /// Sends a write to a leader that can accept one, if there is one.
    fn write(&mut self) -> Result<bool, Box<dyn Error>> {
        let Some(leader) = self
            .up
            .values_mut()
            .find(|member| member.is_settled())
        else {
            return Ok(false);
        };

        self.writes += 1;
        let key = format!("key-{}", self.writes);
        let change = Change::Put {
            key: key.clone(),
            value: b"1".to_vec(),
        };
        let sequence = leader.propose(change, Condition::Always)?;

        self.pending.push(Pending {
            leader: leader.id(),
            term: leader.term(),
            sequence,
            key,
        });
        self.network
            .extend(leader.replicate());

        Ok(true)
    }

    fn leader(&self) -> Option<&Member> {
        self.up
            .values()
            .filter(|member| member.role() == Role::Leader)
            .max_by_key(|member| member.term())
    }
}

After every step, raftsim checks 3 properties.

Listing 4.26 The properties checked after every step. raftsim/src/main.rs

impl Cluster {
    /// The properties that must hold after every step.
    fn check(&mut self) -> Result<(), Box<dyn Error>> {
        // No 2 members lead the same term.
        for member in self
            .up
            .values()
            .filter(|member| member.role() == Role::Leader)
        {
            let first = *self
                .leaders
                .entry(member.term())
                .or_insert(member.id());

            if first != member.id() {
                return Err(format!(
                    "members {first} and {} both lead term {}",
                    member.id(),
                    member.term()
                )
                .into());
            }
        }

        // A write is acknowledged when the leader that accepted it, still
        // leading the same term, has committed its record.
        let up = &self.up;
        let committed = |write: &Pending| {
            up.get(&write.leader)
                .is_some_and(|leader| {
                    leader.role() == Role::Leader
                        && leader.term() == write.term
                        && leader.committed() >= write.sequence
                })
        };
        let (done, waiting): (Vec<Pending>, Vec<Pending>) = self
            .pending
            .drain(..)
            .partition(committed);

        self.acknowledged.extend(
            done.into_iter()
                .map(|write| (write.sequence, write.key)),
        );
        self.pending = waiting;

        // Every member that has committed the sequence of an acknowledged
        // write holds that write at that sequence.
        for (sequence, key) in &self.acknowledged {
            for member in self
                .up
                .values()
                .filter(|member| member.committed() >= *sequence)
            {
                let holds = matches!(&member.records()[*sequence as usize - 1].change, Change::Put { key: held, .. } if held == key);

                if !holds {
                    return Err(format!(
                        "member {} lost the acknowledged write of {key}",
                        member.id()
                    )
                    .into());
                }
            }
        }

        Ok(())
    }
}

A write counts as acknowledged under exactly the condition that the node itself applies: the member that accepted it has committed its record while still leading the same term.

Listing 4.27 A scripted election, a crashed leader, and the repair of its log. raftsim/src/main.rs

fn show(cluster: &Cluster) -> String {
    let describe = |id: &NodeId| match cluster.up.get(id) {
        Some(member) => format!(
            "{id}: {:?} term {} log {}",
            member.role(),
            member.term(),
            member.last_sequence()
        ),
        None => format!("{id}: down"),
    };

    MEMBERS
        .iter()
        .map(describe)
        .collect::<Vec<_>>()
        .join(" | ")
}

/// Ticks every running member, delivering all messages after each tick,
/// until a member leads. It returns the number of ticks.
fn elect(cluster: &mut Cluster) -> Result<u64, Box<dyn Error>> {
    let mut ticks = 0;

    while cluster
        .up
        .values()
        .all(|member| !member.is_settled())
    {
        ticks += 1;

        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    Ok(ticks)
}

fn cases(root: &Path) -> Result<(), Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;

    println!("start:            {}", show(&cluster));

    let ticks = elect(&mut cluster)?;
    println!("after {ticks:2} ticks:   {}", show(&cluster));

    for _ in 0..3 {
        cluster.write()?;
        cluster.deliver_all()?;
    }

    println!(
        "after 3 writes:   {}, {} acknowledged",
        show(&cluster),
        cluster.acknowledged.len()
    );

    // The leader accepts a write and crashes before it sends the record.
    let old = cluster
        .leader()
        .map(Member::id)
        .ok_or("no leader")?;
    cluster.write()?;
    cluster.network.clear();
    cluster.crash(old);
    println!(
        "leader {old} appends record 5 and crashes: {}",
        show(&cluster)
    );

    let ticks = elect(&mut cluster)?;
    println!("after {ticks:2} ticks:   {}", show(&cluster));

    cluster.write()?;
    cluster.deliver_all()?;
    println!(
        "after 1 write:    {}, {} acknowledged",
        show(&cluster),
        cluster.acknowledged.len()
    );

    cluster.start(old)?;
    println!("member {old} returns: {}", show(&cluster));

    for _ in 0..2 {
        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    let discarded = cluster
        .up
        .get(&old)
        .map_or(0, Member::discarded);
    println!("after 2 ticks:    {}", show(&cluster));
    println!(
        "member {old} discarded {discarded} record that was never committed, and holds the leader's records in its place"
    );

    Ok(())
}

Listing 4.28 200 random runs, and the recovery at the end of each. raftsim/src/main.rs

const RUNS: u64 = 200;
const STEPS: usize = 600;

/// One run: random ticks, deliveries, losses, crashes, restarts and writes.
fn run(root: &Path, seed: u64) -> Result<Cluster, Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let mut random = Random(seed);

    for _ in 0..STEPS {
        let id = MEMBERS[random.below(MEMBERS.len())];

        match random.below(100) {
            0..40 => cluster.tick(id)?,
            40..82 => cluster.deliver_any(&mut random)?,
            82..87 => cluster.lose_any(&mut random),
            87..90 => cluster.crash(id),
            90..94 => cluster.restart(id)?,
            94.. => cluster.write().map(drop)?,
        }
    }

    recover(&mut cluster)?;

    Ok(cluster)
}

/// Ends the faults: every member runs and every message arrives. The
/// members must then agree on a leader and on a log that holds every
/// acknowledged write.
fn recover(cluster: &mut Cluster) -> Result<(), Box<dyn Error>> {
    for id in MEMBERS {
        if !cluster.up.contains_key(&id) {
            cluster.start(id)?;
        }
    }

    let agreed = |cluster: &Cluster| {
        cluster
            .leader()
            .is_some_and(|leader| {
                leader.is_settled()
                    && cluster
                        .up
                        .values()
                        .all(|member| member.records() == leader.records())
            })
    };

    for _ in 0..200 {
        if agreed(cluster) {
            break;
        }

        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    if !agreed(cluster) {
        return Err("the members did not agree after the faults ended".into());
    }

    let leader = cluster
        .leader()
        .ok_or("the members have no leader after the faults ended")?;

    for (sequence, key) in &cluster.acknowledged {
        if leader
            .get(key)
            .is_none_or(|entry| entry.version != *sequence)
        {
            return Err(
                format!("the acknowledged write of {key} is missing after recovery").into(),
            );
        }
    }

    Ok(())
}

fn runs(root: &Path) -> Result<(), Box<dyn Error>> {
    let (mut leaders, mut acknowledged, mut discarded, mut most_terms) = (0, 0, 0, 0);

    for seed in 1..=RUNS {
        let directory = root.join(format!("run-{seed}"));
        let mut cluster = run(&directory, seed)?;

        for id in MEMBERS {
            cluster.crash(id);
        }

        leaders += cluster.leaders.len();
        acknowledged += cluster.acknowledged.len();
        discarded += cluster.discarded;
        most_terms = most_terms.max(
            cluster
                .leaders
                .keys()
                .next_back()
                .copied()
                .unwrap_or(0),
        );

        fs::remove_dir_all(&directory)?;
    }

    println!("{RUNS} runs of {STEPS} random steps:");
    println!("  {leaders} leaders were elected, and no term had 2 leaders");
    println!("  {acknowledged} writes were acknowledged, and every one was present after recovery");
    println!(
        "  {discarded} uncommitted records were discarded by members whose log differed from a leader's"
    );
    println!("  the highest term reached in a run was {most_terms}");

    Ok(())
}

Each run takes 600 random steps. A step ticks the clock of a member (40%), delivers a message (42%), loses a message (5%), crashes a member (3%), restarts a member (4%) or sends a write (6%). At the end of a run, all members are started, all messages are delivered, and raftsim verifies that the members agree on a leader and on a log, and that the leader returns every acknowledged write.

Listing 4.29 Running the simulation. raftsim/src/main.rs

fn main() -> Result<(), Box<dyn Error>> {
    let root = env::temp_dir().join(format!("raftsim-{}", process::id()));
    let result = cases(&root.join("cases")).and_then(|()| runs(&root));

    fs::remove_dir_all(&root)?;

    result
}

$ cargo run --release -p raftsim
start:            1: Follower term 0 log 0 | 2: Follower term 0 log 0 | 3: Follower term 0 log 0
after 11 ticks:   1: Follower term 1 log 1 | 2: Follower term 1 log 1 | 3: Leader term 1 log 1
after 3 writes:   1: Follower term 1 log 4 | 2: Follower term 1 log 4 | 3: Leader term 1 log 4, 3 acknowledged
leader 3 appends record 5 and crashes: 1: Follower term 1 log 4 | 2: Follower term 1 log 4 | 3: down
after 10 ticks:   1: Leader term 2 log 5 | 2: Follower term 2 log 5 | 3: down
after 1 write:    1: Leader term 2 log 6 | 2: Follower term 2 log 6 | 3: down, 4 acknowledged
member 3 returns: 1: Leader term 2 log 6 | 2: Follower term 2 log 6 | 3: Follower term 1 log 5
after 2 ticks:    1: Leader term 2 log 6 | 2: Follower term 2 log 6 | 3: Follower term 2 log 6
member 3 discarded 1 record that was never committed, and holds the leader's records in its place
200 runs of 600 random steps:
  480 leaders were elected, and no term had 2 leaders
  504 writes were acknowledged, and every one was present after recovery
  18 uncommitted records were discarded by members whose log differed from a leader's
  the highest term reached in a run was 8

Checking the simulation

A simulation that reports no failure is informative only if it would have reported one for an incorrect implementation, and this is a question that can be put to the test. We disabled 3 rules of Raft in turn and ran the simulation against each modified library.

The third result has to be stated with care. The correctness of Raft requires this condition, and the implementation contains it. With 3 members and 200 runs, however, the simulation did not produce a sequence of events in which the absence of the condition leads to an error. In the paper on Raft, such a sequence involves 5 members and several successive failures of leaders at particular moments. The simulation therefore provides evidence for the voting rules and for the repair of logs, and provides none for this condition. It would have been easy to report “200 runs passed” and to leave the reader with the impression that everything had been verified, and the difference between what a test exercises and what it merely fails to contradict is one that this book tries to keep in view.

Design overview

The node

kvraft connects the library to a clock, to a network and to an HTTP API. All members are started with the same options, apart from their ids and addresses.

Listing 4.30 The imports of kvraft/src/main.rs. kvraft/src/main.rs

use std::collections::BTreeMap;
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::header::{CONTENT_TYPE, ETAG, IF_NONE_MATCH};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use clap::Parser;
use kv::{Change, Condition};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use raft::{Member, NodeId, Role};
use serde::Serialize;
use tokio::net::TcpListener;
use tonic::transport::Server;

use crate::node::Node;
use crate::rpc::{ElectionServer, ReplicationServer, Service};

Listing 4.31 The options of the command line. kvraft/src/main.rs

/// Reads `2=http://127.0.0.1:7102` as member 2 at that address.
fn parse_peer(text: &str) -> Result<(NodeId, String), String> {
    let (id, address) = text
        .split_once('=')
        .ok_or("a peer is written as id=address")?;
    let Ok(id) = id.parse() else {
        return Err("the id of a peer is a number".to_owned());
    };

    Ok((id, address.to_owned()))
}

#[derive(Parser)]
struct Options {
    /// The id of this member, a number that no other member has.
    #[arg(long)]
    id: NodeId,

    /// The address of the HTTP API.
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    /// The address on which this member accepts messages from the others.
    #[arg(long, default_value = "127.0.0.1:7101")]
    replication: SocketAddr,

    /// Another member and its replication address, as id=address.
    #[arg(long = "peer", value_parser = parse_peer)]
    peers: Vec<(NodeId, String)>,

    #[arg(long, default_value = "data")]
    data: PathBuf,

    /// Respond without waiting for the disk.
    #[arg(long)]
    no_sync: bool,
}

Listing 4.32 The imports of kvraft/src/rpc.rs. kvraft/src/rpc.rs

use std::sync::Arc;

use kv::{Change, Record};
use raft::{AppendAnswer, AppendRequest, Message, NodeId, VoteAnswer, VoteRequest};
use tonic::{Request, Response, Status};

use crate::node::Node;

Listing 4.33 Conversion between the generated types and the library's messages. kvraft/src/rpc.rs

fn record_to_proto(record: &Record) -> proto::Record {
    let change = match record.change.clone() {
        Change::Put { key, value } => proto::record::Change::Put(proto::Put { key, value }),
        Change::Delete { key } => proto::record::Change::Delete(proto::Delete { key }),
    };

    proto::Record {
        sequence: record.sequence,
        change: Some(change),
    }
}

fn record_from_proto(record: proto::Record) -> Result<Record, Status> {
    let change = match record.change {
        Some(proto::record::Change::Put(put)) => Change::Put {
            key: put.key,
            value: put.value,
        },
        Some(proto::record::Change::Delete(delete)) => Change::Delete { key: delete.key },
        None => return Err(Status::invalid_argument("a record without a change")),
    };

    Ok(Record {
        sequence: record.sequence,
        change,
    })
}

impl From<AppendRequest> for proto::AppendRequest {
    fn from(request: AppendRequest) -> Self {
        proto::AppendRequest {
            shard: 0,
            term: request.term,
            previous_sequence: request.previous_sequence,
            records: request
                .records
                .iter()
                .map(record_to_proto)
                .collect(),
            committed_sequence: request.committed_sequence,
            previous_term: request.previous_term,
            leader: request.leader,
        }
    }
}

impl TryFrom<proto::AppendRequest> for AppendRequest {
    type Error = Status;

    fn try_from(request: proto::AppendRequest) -> Result<Self, Status> {
        Ok(AppendRequest {
            term: request.term,
            leader: request.leader,
            previous_sequence: request.previous_sequence,
            previous_term: request.previous_term,
            records: request
                .records
                .into_iter()
                .map(record_from_proto)
                .collect::<Result<_, _>>()?,
            committed_sequence: request.committed_sequence,
        })
    }
}

impl From<VoteRequest> for proto::VoteRequest {
    fn from(request: VoteRequest) -> Self {
        proto::VoteRequest {
            term: request.term,
            candidate: request.candidate,
            last_sequence: request.last_sequence,
            last_term: request.last_term,
        }
    }
}

/// The answer of member `from` to an `Append` call, as a message.
pub fn appended(from: NodeId, answer: proto::AppendResponse) -> Message {
    Message::Appended(AppendAnswer {
        term: answer.term,
        from,
        appended: answer.appended,
        last_sequence: answer.last_sequence,
    })
}

/// The answer of member `from` to a `RequestVote` call, as a message.
pub fn voted(from: NodeId, answer: proto::VoteResponse) -> Message {
    Message::Voted(VoteAnswer {
        term: answer.term,
        from,
        granted: answer.granted,
    })
}

Each gRPC call delivers 1 message to a member, and the member’s answer is returned as the response to the call.

Listing 4.34 The server side of Append and RequestVote. kvraft/src/rpc.rs

/// The server side of both services. Each call is one message for the
/// member, and the member's answer is the response of the call.
pub struct Service {
    pub node: Arc<Node>,
}

#[tonic::async_trait]
impl Replication for Service {
    async fn append(
        &self,
        request: Request<proto::AppendRequest>,
    ) -> Result<Response<proto::AppendResponse>, Status> {
        let request = AppendRequest::try_from(request.into_inner())?;

        let answer = self
            .node
            .answer(Message::Append(request))
            .map_err(|error| Status::unavailable(error.to_string()))?;

        let Message::Appended(answer) = answer else {
            return Err(Status::internal(
                "the member answered with another kind of message",
            ));
        };

        Ok(Response::new(proto::AppendResponse {
            term: answer.term,
            appended: answer.appended,
            last_sequence: answer.last_sequence,
        }))
    }
}

#[tonic::async_trait]
impl Election for Service {
    async fn request_vote(
        &self,
        request: Request<proto::VoteRequest>,
    ) -> Result<Response<proto::VoteResponse>, Status> {
        let request = request.into_inner();
        let request = VoteRequest {
            term: request.term,
            candidate: request.candidate,
            last_sequence: request.last_sequence,
            last_term: request.last_term,
        };

        let answer = self
            .node
            .answer(Message::Vote(request))
            .map_err(|error| Status::unavailable(error.to_string()))?;

        let Message::Voted(answer) = answer else {
            return Err(Status::internal(
                "the member answered with another kind of message",
            ));
        };

        Ok(Response::new(proto::VoteResponse {
            term: answer.term,
            granted: answer.granted,
        }))
    }
}

Around the member, a node consists of a small loop, which is shown in figure 4.elect-node-loop.

Listing 4.35 The imports of kvraft/src/node.rs. kvraft/src/node.rs

use std::collections::BTreeMap;
use std::io;
use std::sync::Arc;
use std::time::Duration;

use kv::{Change, Condition, Entry, Written};
use kvd::api::ApiError;
use parking_lot::Mutex;
use raft::{Member, Message, NodeId, Outgoing, Role};
use tokio::sync::watch;
use tokio::task;
use tokio::time::{interval, timeout};
use tonic::transport::{Channel, Endpoint};

use crate::rpc::{ElectionClient, ReplicationClient, appended, proto, voted};

Listing 4.36 The state of a node. kvraft/src/node.rs

/// The length of a tick of the member's clock.
pub const TICK: Duration = Duration::from_millis(100);

/// How long a member waits for the answer to one message.
const ANSWER: Duration = Duration::from_millis(500);

/// How long a write waits for its record to be committed.
const COMMIT: Duration = Duration::from_secs(1);

pub struct Node {
    pub member: Mutex<Member>,
    /// A connection to each of the other members.
    peers: BTreeMap<NodeId, Channel>,
    /// Held by the write that is in progress.
    turn: tokio::sync::Mutex<()>,
    /// The committed sequence, for writes that wait for it to advance.
    committed: watch::Sender<u64>,
}

impl Node {
    pub fn new(
        member: Member,
        addresses: BTreeMap<NodeId, String>,
    ) -> Result<Arc<Node>, tonic::transport::Error> {
        let mut peers = BTreeMap::new();

        for (id, address) in addresses {
            peers.insert(
                id,
                Endpoint::from_shared(address)?
                    .connect_timeout(ANSWER)
                    .connect_lazy(),
            );
        }

        Ok(Arc::new(Node {
            member: Mutex::new(member),
            peers,
            turn: tokio::sync::Mutex::new(()),
            committed: watch::Sender::new(0),
        }))
    }
}

with_member is the single point of entry to the member. It takes the lock, runs the operation it is given, and publishes the committed sequence afterwards, so that no operation can change that sequence without the waiting writes being informed.

Listing 4.37 Running an operation on the member. kvraft/src/node.rs

impl Node {
    /// Runs one operation on the member and publishes the committed
    /// sequence. The member waits for the disk, so the operation runs where
    /// blocking is allowed.
    fn with_member<T>(&self, operation: impl FnOnce(&mut Member) -> T) -> T {
        task::block_in_place(|| {
            let mut member = self.member.lock();
            let result = operation(&mut member);
            self.committed
                .send_replace(member.committed());

            result
        })
    }
}

dispatch starts 1 task for each message, and send performs a gRPC call with a time limit. Each response is converted into the corresponding answer message and handed to deliver, which passes it to the member and dispatches whatever the member returns. A call that fails or times out produces no message at all, which Raft tolerates without any special provision, since the next tick sends a new message to the same member.

Listing 4.38 Sending a message and delivering the answer. kvraft/src/node.rs

impl Node {
    /// Sends each message in a task of its own, so that a member that does
    /// not answer delays nothing else.
    fn dispatch(self: &Arc<Node>, messages: Vec<Outgoing>) {
        for message in messages {
            tokio::spawn(Arc::clone(self).send(message));
        }
    }

    /// Carries one request to a member and gives the answer to this member.
    async fn send(self: Arc<Node>, outgoing: Outgoing) {
        let Some(channel) = self
            .peers
            .get(&outgoing.to)
            .cloned()
        else {
            return;
        };

        let answer = match outgoing.message {
            Message::Append(request) => {
                let call = async move {
                    ReplicationClient::new(channel)
                        .append(proto::AppendRequest::from(request))
                        .await
                };

                timeout(ANSWER, call)
                    .await
                    .ok()
                    .and_then(Result::ok)
                    .map(|answer| appended(outgoing.to, answer.into_inner()))
            }
            Message::Vote(request) => {
                let call = async move {
                    ElectionClient::new(channel)
                        .request_vote(proto::VoteRequest::from(request))
                        .await
                };

                timeout(ANSWER, call)
                    .await
                    .ok()
                    .and_then(Result::ok)
                    .map(|answer| voted(outgoing.to, answer.into_inner()))
            }
            // An answer travels as the response of the call that asked.
            Message::Appended(..) | Message::Voted(..) => None,
        };

        if let Some(answer) = answer {
            self.deliver(answer);
        }
    }

    /// Gives the member a message and sends what the member returns.
    fn deliver(self: &Arc<Node>, message: Message) {
        match self.with_member(|member| member.handle(message)) {
            Ok(messages) => self.dispatch(messages),
            Err(error) => eprintln!("the member could not process a message: {error}"),
        }
    }
}

run_clock calls tick every 100 ms. answer serves the gRPC handlers: an incoming request is 1 message for the member, which returns exactly 1 answer to it.

Listing 4.39 The clock, and answering a request from another member. kvraft/src/node.rs

impl Node {
    /// The member's answer to a request that arrived over gRPC.
    pub fn answer(&self, request: Message) -> io::Result<Message> {
        let mut messages = self.with_member(|member| member.handle(request))?;

        messages
            .pop()
            .map(|outgoing| outgoing.message)
            .ok_or_else(|| io::Error::other("the member gave no answer"))
    }

    /// Advances the member's clock, for as long as the node runs.
    pub async fn run_clock(self: Arc<Node>) {
        let mut ticks = interval(TICK);

        loop {
            ticks.tick().await;

            match self.with_member(Member::tick) {
                Ok(messages) => self.dispatch(messages),
                Err(error) => eprintln!("the member could not advance its clock: {error}"),
            }
        }
    }
}

Listing 4.40 Writes and reads. kvraft/src/node.rs

impl Node {
    fn not_leader(member: &Member) -> ApiError {
        match member.leader() {
            Some(leader) => ApiError::Misdirected(format!("the leader is member {leader}")),
            None => ApiError::NoLeader,
        }
    }

    /// A write on the leader. It is acknowledged when its record is
    /// committed and this member has led the same term throughout.
    pub async fn write(
        self: &Arc<Node>,
        change: Change,
        condition: Condition,
    ) -> Result<Written, ApiError> {
        let turn = self.turn.lock().await;

        let last = {
            let member = self.member.lock();

            if member.role() != Role::Leader {
                return Err(Node::not_leader(&member));
            }

            member.last_sequence()
        };

        self.wait_for(last).await?;

        let (written, term, messages) = self.with_member(|member| {
            if !member.is_settled() {
                return Err(Node::not_leader(member));
            }

            let created = match &change {
                Change::Put { key, .. } | Change::Delete { key } => member.get(key).is_none(),
            };
            let version = member.propose(change, condition)?;

            Ok((
                Written { version, created },
                member.term(),
                member.replicate(),
            ))
        })?;

        self.dispatch(messages);
        self.wait_for(written.version)
            .await?;
        drop(turn);

        // A member that lost the leadership while it waited may hold another
        // leader's record at this sequence.
        let member = self.member.lock();

        if member.role() == Role::Leader && member.term() == term {
            Ok(written)
        } else {
            Err(ApiError::Unavailable(
                "the leadership changed before the record was committed".into(),
            ))
        }
    }

    async fn wait_for(&self, sequence: u64) -> Result<(), ApiError> {
        let mut committed = self.committed.subscribe();

        let reached = committed.wait_for(|&committed| committed >= sequence);
        let in_time = timeout(COMMIT, reached)
            .await
            .is_ok_and(|reached| reached.is_ok());

        if in_time {
            Ok(())
        } else {
            Err(ApiError::Unavailable(format!(
                "record {sequence} is not on a majority of the members"
            )))
        }
    }

    pub fn get(&self, key: &str) -> Result<Entry, ApiError> {
        let member = self.member.lock();

        if member.role() != Role::Leader {
            return Err(Node::not_leader(&member));
        }

        if !member.can_read() {
            return Err(ApiError::Unavailable(
                "this leader cannot confirm that it still leads".into(),
            ));
        }

        member
            .get(key)
            .cloned()
            .ok_or(ApiError::NotFound)
    }
}

The write path has the structure of chapter 3, with 1 addition after the wait. If a member loses its leadership while a write is waiting, its uncommitted record may be replaced by a record of the new leader at the same sequence, and the committed sequence may then pass that position without the write having been stored at all. A handler therefore acknowledges a write only if its member still leads the term in which the record was appended. Otherwise it answers 503, and the outcome of the write is unknown to its client, to be resolved as chapter 3 described.

Any member that is not the leader answers 421 Misdirected Request and names the leader it knows of in the detail field.

Listing 4.41 The HTTP handlers and the health of a node. kvraft/src/main.rs

async fn put_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    let written = node
        .write(
            Change::Put {
                key,
                value: body.to_vec(),
            },
            condition,
        )
        .await?;
    let status = if written.created {
        StatusCode::CREATED
    } else {
        StatusCode::NO_CONTENT
    };

    Ok((status, [(ETAG, etag(written.version))]).into_response())
}

async fn delete_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    node.write(Change::Delete { key }, condition)
        .await?;

    Ok(StatusCode::NO_CONTENT)
}

async fn get_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let cached = version_header(&headers, IF_NONE_MATCH)?;
    let entry = node.get(&key)?;
    let tag = [(ETAG, etag(entry.version))];

    if cached == Some(Condition::Version(entry.version)) {
        return Ok((StatusCode::NOT_MODIFIED, tag).into_response());
    }

    Ok((
        tag,
        [(CONTENT_TYPE, "application/octet-stream")],
        entry.value,
    )
        .into_response())
}

#[derive(Serialize)]
struct Health {
    id: NodeId,
    role: &'static str,
    term: u64,
    /// The leader of the term, when this member knows it.
    leader: Option<NodeId>,
    last_sequence: u64,
    committed: u64,
}

async fn health(State(node): State<Arc<Node>>) -> Json<Health> {
    let member = node.member.lock();

    let role = match member.role() {
        Role::Leader => "leader",
        Role::Candidate => "candidate",
        Role::Follower => "follower",
    };

    Json(Health {
        id: member.id(),
        role,
        term: member.term(),
        leader: member.leader(),
        last_sequence: member.last_sequence(),
        committed: member.committed(),
    })
}

Listing 4.42 Starting a node. kvraft/src/main.rs

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let peers: BTreeMap<NodeId, String> = options.peers.into_iter().collect();
    let member = Member::open(
        &options.data,
        options.id,
        peers.keys().copied().collect(),
        !options.no_sync,
    )?;

    println!(
        "member {}, term {}, log ends at {}",
        member.id(),
        member.term(),
        member.last_sequence()
    );

    let node = Node::new(member, peers)?;

    let replication = ReplicationServer::new(Service {
        node: Arc::clone(&node),
    });
    let election = ElectionServer::new(Service {
        node: Arc::clone(&node),
    });

    tokio::spawn(
        Server::builder()
            .add_service(replication)
            .add_service(election)
            .serve(options.replication),
    );
    tokio::spawn(Arc::clone(&node).run_clock());

    let routes = Router::new()
        .route(
            "/v1/keys/{key}",
            get(get_key)
                .put(put_key)
                .delete(delete_key),
        )
        .route("/healthz", get(health))
        .layer(DefaultBodyLimit::max(MAX_VALUE))
        .with_state(node);

    let listener = TcpListener::bind(options.listen).await?;
    println!("listening on {}", listener.local_addr()?);

    axum::serve(listener, routes).await?;

    Ok(())
}

Running 3 nodes

$ ./target/release/kvraft --id 1 --listen 127.0.0.1:7001 --replication 127.0.0.1:7101 --data /var/tmp/m1 \
      --peer 2=http://127.0.0.1:7102 --peer 3=http://127.0.0.1:7103 &
member 1, term 0, log ends at 0
listening on 127.0.0.1:7001
$ ./target/release/kvraft --id 2 --listen 127.0.0.1:7002 --replication 127.0.0.1:7102 --data /var/tmp/m2 \
      --peer 1=http://127.0.0.1:7101 --peer 3=http://127.0.0.1:7103 &
$ ./target/release/kvraft --id 3 --listen 127.0.0.1:7003 --replication 127.0.0.1:7103 --data /var/tmp/m3 \
      --peer 1=http://127.0.0.1:7101 --peer 2=http://127.0.0.1:7102 &
$ for port in 7001 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"id":1,"role":"follower","term":1,"leader":3,"last_sequence":1,"committed":1}
{"id":2,"role":"follower","term":1,"leader":3,"last_sequence":1,"committed":1}
{"id":3,"role":"leader","term":1,"leader":3,"last_sequence":1,"committed":1}

All 3 members were started without any indication of who should lead. Within 2 seconds, member 3 had been elected in term 1. Every log contains 1 record, the marker of term 1.

$ curl -i -X PUT http://127.0.0.1:7003/v1/keys/user-1 --data-binary 'ada'
HTTP/1.1 201 Created
etag: "2"
content-length: 0

$ curl -i -X PUT http://127.0.0.1:7001/v1/keys/user-1 --data-binary 'x'
HTTP/1.1 421 Misdirected Request
content-type: application/problem+json
content-length: 116

{"type":"/problems/not-leader","title":"This node is not the leader","status":421,"detail":"the leader is member 3"}

The first write of a client receives version 2, since record 1 is the term marker. Versions remain strictly increasing, which is all that a client is entitled to rely upon.

Experiments

As in chapter 3, the 3 nodes run on 1 machine and share its disk.

The leader is lost

probe writes keys continuously for a fixed time. When a node refuses a write or fails to respond, probe sends the write to the next node. It records every acknowledged key, and the longest interval between 2 consecutive acknowledged writes.

Listing 4.43 The imports of probe/src/main.rs. probe/src/main.rs

use std::error::Error;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::PathBuf;
use std::thread;
use std::time::{Duration, Instant};

use clap::Parser;
use ureq::Agent;

Listing 4.44 Measuring the longest pause between acknowledged writes. probe/src/main.rs

fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let agent: Agent = Agent::config_builder()
        .timeout_global(Some(Duration::from_secs(2)))
        .build()
        .into();
    let mut file = BufWriter::new(File::create(&options.acked)?);

    let started = Instant::now();
    let (mut server, mut number, mut refused) = (0, 1, 0);
    let (mut last_success, mut longest, mut longest_at) = (started, Duration::ZERO, Duration::ZERO);

    while started.elapsed() < Duration::from_secs(options.seconds) {
        let key = format!("key-{number:06}");
        let address = format!("{}/v1/keys/{key}", options.servers[server]);

        if agent
            .put(&address)
            .send(format!("value of {key}"))
            .is_err()
        {
            // This node did not accept the write. Try the next one.
            refused += 1;
            server = (server + 1) % options.servers.len();
            thread::sleep(Duration::from_millis(20));

            continue;
        }

        let pause = last_success.elapsed();

        if pause > longest {
            (longest, longest_at) = (pause, started.elapsed());
        }

        last_success = Instant::now();
        writeln!(file, "{key}")?;
        number += 1;
    }

    file.flush()?;

    println!(
        "{} writes acknowledged in {} seconds, {refused} attempts refused",
        number - 1,
        options.seconds
    );
    println!(
        "the longest pause between 2 acknowledged writes was {:.2} s, ending {:.1} s after the start",
        longest.as_secs_f64(),
        longest_at.as_secs_f64()
    );

    Ok(())
}

The leader is killed 6 seconds into a run of 20 seconds.

$ ./target/release/probe --seconds 20 --acked acked.txt \
      --server http://127.0.0.1:7001 --server http://127.0.0.1:7002 --server http://127.0.0.1:7003 &
$ sleep 6; kill -9 <pid of member 3, the leader>
$ sleep 3; for port in 7001 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"id":1,"role":"leader","term":2,"leader":1,"last_sequence":610,"committed":609}
{"id":2,"role":"follower","term":2,"leader":1,"last_sequence":611,"committed":610}
curl: (7) Failed to connect to 127.0.0.1 port 7003
1443 writes acknowledged in 20 seconds, 51 attempts refused
the longest pause between 2 acknowledged writes was 1.05 s, ending 7.1 s after the start
$ ./target/release/kvctl verify --server http://127.0.0.1:7001 --acked acked.txt
1443 acknowledged, 1443 found, 0 missing

The measured pause can be set against the availability allowance of chapter 1. For the shards led by 1 node of 24, a pause of 1 second consumes about 0.04 seconds of a monthly allowance of 1296. The promotion by hand of chapter 3 would have consumed minutes.

The former leader returns

The member that was killed is started again with its data directory intact, which chapter 3 did not permit.

$ ./target/release/kvraft --id 3 --listen 127.0.0.1:7003 --replication 127.0.0.1:7103 --data /var/tmp/m3 \
      --peer 1=http://127.0.0.1:7101 --peer 2=http://127.0.0.1:7102 &
$ for port in 7001 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"id":1,"role":"leader","term":2,"leader":1,"last_sequence":1447,"committed":1447}
{"id":2,"role":"follower","term":2,"leader":1,"last_sequence":1447,"committed":1447}
{"id":3,"role":"follower","term":2,"leader":1,"last_sequence":1447,"committed":1447}
$ cmp /var/tmp/m1/kv.log /var/tmp/m2/kv.log && cmp /var/tmp/m1/kv.log /var/tmp/m3/kv.log && echo identical
identical
$ cat /var/tmp/m1/state
2 1

It rejoined as a follower of term 2 and received the records it lacked. The 3 logs are identical, and the state file of member 1 contains 2 1: term 2, and a vote for member 1, which it gave to itself when it stood for election.

A member is left alone

The leader and 1 follower are killed, so that a single member remains.

$ kill -9 <pid of member 2, the leader> <pid of member 3>
$ sleep 5; curl http://127.0.0.1:7001/healthz
{"id":1,"role":"candidate","term":6,"leader":null,"last_sequence":1449,"committed":1449}
$ curl -X PUT http://127.0.0.1:7001/v1/keys/alone --data-binary 'x'
{"type":"/problems/store-unavailable","title":"The node cannot serve the request","status":503,"detail":"the shard has no leader at the moment"}
$ sleep 3; curl http://127.0.0.1:7001/healthz
{"id":1,"role":"candidate","term":8,"leader":null,"last_sequence":1449,"committed":1449}

$ (members 2 and 3 are started again)
$ sleep 4; for port in 7001 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"id":1,"role":"leader","term":9,"leader":1,"last_sequence":1450,"committed":1450}
{"id":2,"role":"follower","term":9,"leader":1,"last_sequence":1450,"committed":1450}
{"id":3,"role":"follower","term":9,"leader":1,"last_sequence":1450,"committed":1450}

This is the minority side of a partition once more, seen from the inside. The lone member behaves exactly as it would if the other 2 were running and unreachable, as from its own position it must, since it has no means of telling the 2 situations apart.

Limitations

The complete programs

Listing 4.45 The complete raft/src/lib.rs. raft/src/lib.rs

//! A member of a shard that elects its own leader.
//!
//! A `Member` is a replica with a role. It follows the Raft algorithm: a
//! follower that hears nothing from a leader stands for election, a
//! candidate that receives the votes of a majority becomes the leader, and
//! a leader replicates its log. The type does no I/O on the network and
//! reads no clock. The caller delivers messages to `handle`, calls `tick`
//! at a fixed interval, and sends the messages that both return.

use std::collections::BTreeMap;
use std::fs::{self, File};
use std::io::{self, Write};
use std::path::{Path, PathBuf};

use kv::{Change, Condition, Entry, Log, Record, StoreError, apply, check};

pub type NodeId = u32;

/// Sent by a leader to a follower.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AppendRequest {
    pub term: u64,
    pub leader: NodeId,
    /// The sequence of the record that precedes `records`, and its term.
    pub previous_sequence: u64,
    pub previous_term: u64,
    pub records: Vec<Record>,
    pub committed_sequence: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AppendAnswer {
    pub term: u64,
    pub from: NodeId,
    pub appended: bool,
    /// When appended: the last sequence that is known to match the leader's
    /// log. When refused: the sequence the leader should continue from.
    pub last_sequence: u64,
}

/// Sent by a candidate to every other member.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VoteRequest {
    pub term: u64,
    pub candidate: NodeId,
    /// The end of the candidate's log, and the term of its last record.
    pub last_sequence: u64,
    pub last_term: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VoteAnswer {
    pub term: u64,
    pub from: NodeId,
    pub granted: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Message {
    Append(AppendRequest),
    Appended(AppendAnswer),
    Vote(VoteRequest),
    Voted(VoteAnswer),
}

/// A message and the member it is addressed to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outgoing {
    pub to: NodeId,
    pub message: Message,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
    Follower,
    Candidate,
    Leader,
}

/// The ticks a follower waits for a leader before it stands for election:
/// 10 to 19, different for each member and each term, so that 2 members
/// seldom stand at the same moment. A tick is 100 ms in the node.
fn election_timeout(id: NodeId, term: u64) -> u64 {
    10 + (u64::from(id) * 7 + term * 3) % 10
}

/// What the leader knows about one follower.
#[derive(Debug, Clone, Copy, Default)]
struct Progress {
    /// The sequence after which the next message to the follower begins.
    next: u64,
    /// The last sequence the follower has confirmed to match.
    matched: u64,
    /// The tick at which the follower last answered.
    heard: u64,
}

pub struct Member {
    id: NodeId,
    peers: Vec<NodeId>,
    directory: PathBuf,
    sync: bool,

    log: Log,
    /// Every record of the log. Record `n` is at index `n - 1`.
    records: Vec<Record>,
    /// The term in which each record was written.
    terms: Vec<u64>,
    entries: BTreeMap<String, Entry>,
    committed: u64,

    /// The current term, and the member this one voted for in it. Both are
    /// stored on disk before any answer that depends on them is sent.
    term: u64,
    voted_for: Option<NodeId>,

    role: Role,
    /// The leader of the current term, when this member knows it.
    leader: Option<NodeId>,
    votes: usize,
    progress: BTreeMap<NodeId, Progress>,
    /// Ticks since the member started, and the tick of the last message
    /// from a leader or the last vote granted.
    now: u64,
    quiet_since: u64,
    /// The records this member has removed from its log because they
    /// differed from a leader's.
    discarded: u64,
}

/// The key of a term marker. A client cannot write the empty key, so the
/// log can use it for a record of its own: the first record a new leader
/// appends, whose value is the leader's term.
const MARKER: &str = "";

fn marker(term: u64) -> Change {
    Change::Put {
        key: MARKER.to_owned(),
        value: term.to_le_bytes().to_vec(),
    }
}

/// The term a record starts, if it is a term marker.
fn marked_term(record: &Record) -> Option<u64> {
    let Change::Put { key, value } = &record.change else {
        return None;
    };

    if key != MARKER {
        return None;
    }

    let term: [u8; 8] = value.as_slice().try_into().ok()?;

    Some(u64::from_le_bytes(term))
}

/// The term of each record: the term of the last marker at or before it.
fn terms_of(records: &[Record], mut term: u64) -> Vec<u64> {
    let mut terms = Vec::with_capacity(records.len());

    for record in records {
        term = marked_term(record).unwrap_or(term);
        terms.push(term);
    }

    terms
}

impl Member {
    /// Opens the member in `directory`. It starts as a follower, with
    /// nothing applied, in the term it had reached before it stopped.
    pub fn open(
        directory: &Path,
        id: NodeId,
        peers: Vec<NodeId>,
        sync: bool,
    ) -> io::Result<Member> {
        let (log, recovery) = Log::open(directory, sync)?;
        let (term, voted_for) = read_state(directory)?;
        let terms = terms_of(&recovery.records, 0);

        Ok(Member {
            id,
            peers,
            directory: directory.to_owned(),
            sync,
            log,
            records: recovery.records,
            terms,
            entries: BTreeMap::new(),
            committed: 0,
            term,
            voted_for,
            role: Role::Follower,
            leader: None,
            votes: 0,
            progress: BTreeMap::new(),
            now: 0,
            quiet_since: 0,
            discarded: 0,
        })
    }

    pub fn id(&self) -> NodeId {
        self.id
    }

    pub fn role(&self) -> Role {
        self.role
    }

    pub fn term(&self) -> u64 {
        self.term
    }

    pub fn leader(&self) -> Option<NodeId> {
        self.leader
    }

    pub fn last_sequence(&self) -> u64 {
        self.records.len() as u64
    }

    pub fn committed(&self) -> u64 {
        self.committed
    }

    pub fn records(&self) -> &[Record] {
        &self.records
    }

    pub fn discarded(&self) -> u64 {
        self.discarded
    }

    fn last_term(&self) -> u64 {
        self.terms
            .last()
            .copied()
            .unwrap_or(0)
    }

    /// The term of record `sequence`, or 0 for the position before the log.
    fn term_at(&self, sequence: u64) -> u64 {
        if sequence == 0 {
            0
        } else {
            self.terms[sequence as usize - 1]
        }
    }

    fn majority(&self) -> usize {
        let members = self.peers.len() + 1;

        members / 2 + 1
    }
}


const STATE_FILE: &str = "state";

/// Reads the term and the vote. A member that has no state file has not
/// voted and is in term 0.
fn read_state(directory: &Path) -> io::Result<(u64, Option<NodeId>)> {
    let text = match fs::read_to_string(directory.join(STATE_FILE)) {
        Ok(text) => text,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok((0, None)),
        Err(error) => return Err(error),
    };

    let mut fields = text
        .split_whitespace()
        .map(str::parse::<u64>);
    let invalid = || io::Error::new(io::ErrorKind::InvalidData, "the state file is malformed");
    let term = fields
        .next()
        .and_then(Result::ok)
        .ok_or_else(invalid)?;
    let voted_for = fields
        .next()
        .and_then(Result::ok)
        .map(|id| id as NodeId);

    Ok((term, voted_for))
}

impl Member {
    /// Stores the term and the vote. The new file replaces the old one in a
    /// single step, so a crash leaves one of the 2 and not a mixture.
    fn save_state(&self) -> io::Result<()> {
        let temporary = self.directory.join("state.new");
        let vote = self
            .voted_for
            .map_or(String::new(), |id| id.to_string());
        let mut file = File::create(&temporary)?;

        file.write_all(format!("{} {vote}\n", self.term).as_bytes())?;

        if self.sync {
            file.sync_all()?;
        }

        fs::rename(&temporary, self.directory.join(STATE_FILE))?;

        if self.sync {
            File::open(&self.directory)?.sync_all()?;
        }

        Ok(())
    }
}

impl Member {
    /// Applies the records up to `sequence` that are not applied yet. Term
    /// markers are records of the log and are not entries.
    fn commit(&mut self, sequence: u64) {
        let until = sequence.min(self.last_sequence());

        // A record that marks the start of a term changes no key.
        let changes: Vec<Record> = (self.committed..until)
            .map(|index| self.records[index as usize].clone())
            .filter(|record| marked_term(record).is_none())
            .collect();

        for record in changes {
            apply(&mut self.entries, record);
        }

        self.committed = self.committed.max(until);
    }

    /// The leader's commit rule. A record of an earlier term is not counted
    /// on its own: it becomes committed together with the first record of
    /// the leader's own term that a majority has stored.
    fn advance(&mut self) {
        let mut stored: Vec<u64> = self
            .progress
            .values()
            .map(|progress| progress.matched)
            .collect();
        stored.push(self.last_sequence());
        stored.sort_unstable_by(|a, b| b.cmp(a));

        let on_majority = stored[self.majority() - 1];

        if on_majority > self.committed && self.term_at(on_majority) == self.term {
            self.commit(on_majority);
        }
    }
}

impl Member {
    /// Moves to a newer term as a follower. The vote of the old term does
    /// not carry over.
    fn step_down(&mut self, term: u64) -> io::Result<()> {
        if term > self.term {
            self.term = term;
            self.voted_for = None;
            self.save_state()?;
        }

        self.role = Role::Follower;
        self.leader = None;
        self.progress.clear();

        Ok(())
    }
}

impl Member {
    /// Stands for election in the next term.
    fn stand(&mut self) -> io::Result<Vec<Outgoing>> {
        self.term += 1;
        self.voted_for = Some(self.id);
        self.save_state()?;

        self.role = Role::Candidate;
        self.leader = None;
        self.votes = 1;
        self.quiet_since = self.now;

        let request = VoteRequest {
            term: self.term,
            candidate: self.id,
            last_sequence: self.last_sequence(),
            last_term: self.last_term(),
        };

        Ok(self
            .peers
            .iter()
            .map(|&to| Outgoing {
                to,
                message: Message::Vote(request),
            })
            .collect())
    }
}

impl Member {
    /// Becomes the leader of the current term. The first record of the term
    /// is its marker. When a majority has stored the marker, every earlier
    /// record is committed with it.
    fn lead(&mut self) -> io::Result<Vec<Outgoing>> {
        self.role = Role::Leader;
        self.leader = Some(self.id);

        let next = self.last_sequence();
        self.progress = self
            .peers
            .iter()
            .map(|&peer| {
                (
                    peer,
                    Progress {
                        next,
                        matched: 0,
                        heard: self.now,
                    },
                )
            })
            .collect();

        let record = self.log.append(marker(self.term))?;
        self.records.push(record);
        self.terms.push(self.term);

        Ok(self.replicate())
    }
}

impl Member {
    /// Advances the member's clock by one tick. A leader sends every
    /// follower a message. A follower or a candidate that has waited for
    /// its election timeout stands for election.
    pub fn tick(&mut self) -> io::Result<Vec<Outgoing>> {
        self.now += 1;

        if self.role == Role::Leader {
            return Ok(self.replicate());
        }

        if self.now - self.quiet_since >= election_timeout(self.id, self.term) {
            return self.stand();
        }

        Ok(Vec::new())
    }

    /// A message for every follower: the records it lacks, or a heartbeat.
    pub fn replicate(&self) -> Vec<Outgoing> {
        self.peers
            .iter()
            .filter_map(|&peer| self.append_for(peer))
            .collect()
    }

    fn append_for(&self, peer: NodeId) -> Option<Outgoing> {
        let from = self
            .progress
            .get(&peer)?
            .next
            .min(self.last_sequence());

        let request = AppendRequest {
            term: self.term,
            leader: self.id,
            previous_sequence: from,
            previous_term: self.term_at(from),
            records: self.records[from as usize..].to_vec(),
            committed_sequence: self.committed,
        };

        Some(Outgoing {
            to: peer,
            message: Message::Append(request),
        })
    }
}

impl Member {
    /// Processes a message and returns the messages to send in response.
    pub fn handle(&mut self, message: Message) -> io::Result<Vec<Outgoing>> {
        match message {
            Message::Append(request) => {
                let to = request.leader;
                let answer = self.on_append(request)?;

                Ok(vec![Outgoing {
                    to,
                    message: Message::Appended(answer),
                }])
            }
            Message::Vote(request) => {
                let to = request.candidate;
                let answer = self.on_vote(request)?;

                Ok(vec![Outgoing {
                    to,
                    message: Message::Voted(answer),
                }])
            }
            Message::Appended(answer) => self.on_appended(answer),
            Message::Voted(answer) => self.on_voted(answer),
        }
    }
}

impl Member {
    /// A member grants its vote once in a term, and only to a candidate
    /// whose log is at least as complete as its own.
    fn on_vote(&mut self, request: VoteRequest) -> io::Result<VoteAnswer> {
        if request.term > self.term {
            self.step_down(request.term)?;
        }

        let complete =
            (request.last_term, request.last_sequence) >= (self.last_term(), self.last_sequence());
        let free = self
            .voted_for
            .is_none_or(|id| id == request.candidate);
        let granted = request.term == self.term && complete && free;

        if granted {
            self.voted_for = Some(request.candidate);
            self.save_state()?;
            self.quiet_since = self.now;
        }

        Ok(VoteAnswer {
            term: self.term,
            from: self.id,
            granted,
        })
    }

    fn on_voted(&mut self, answer: VoteAnswer) -> io::Result<Vec<Outgoing>> {
        if answer.term > self.term {
            self.step_down(answer.term)?;

            return Ok(Vec::new());
        }

        if self.role != Role::Candidate || answer.term != self.term || !answer.granted {
            return Ok(Vec::new());
        }

        self.votes += 1;

        if self.votes >= self.majority() {
            self.lead()
        } else {
            Ok(Vec::new())
        }
    }
}

impl Member {
    fn on_append(&mut self, request: AppendRequest) -> io::Result<AppendAnswer> {
        let refuse = |member: &Member, last_sequence| AppendAnswer {
            term: member.term,
            from: member.id,
            appended: false,
            last_sequence,
        };

        if request.term < self.term {
            return Ok(refuse(self, self.last_sequence()));
        }

        // A message of the current term or a newer one comes from the leader.
        self.step_down(request.term)?;
        self.leader = Some(request.leader);
        self.quiet_since = self.now;

        if request.previous_sequence > self.last_sequence() {
            return Ok(refuse(self, self.last_sequence()));
        }

        if self.term_at(request.previous_sequence) != request.previous_term {
            return Ok(refuse(self, request.previous_sequence - 1));
        }

        let matched = request.previous_sequence + request.records.len() as u64;
        self.store(
            request.previous_sequence,
            request.previous_term,
            request.records,
        )?;
        self.commit(
            request
                .committed_sequence
                .min(matched),
        );

        Ok(AppendAnswer {
            term: self.term,
            from: self.id,
            appended: true,
            last_sequence: matched,
        })
    }
}

impl Member {
    /// Makes the log equal to the leader's from `previous` onward: records
    /// the member already has are skipped, a record that differs is removed
    /// together with everything after it, and the rest is appended.
    fn store(&mut self, previous: u64, previous_term: u64, records: Vec<Record>) -> io::Result<()> {
        let terms = terms_of(&records, previous_term);
        let mut fresh = 0;

        for (index, record) in records.iter().enumerate() {
            let sequence = previous + index as u64 + 1;

            if sequence > self.last_sequence() {
                break;
            }

            if self.term_at(sequence) != terms[index]
                || &self.records[sequence as usize - 1] != record
            {
                self.truncate(sequence - 1)?;

                break;
            }

            fresh = index + 1;
        }

        if fresh < records.len() {
            self.log
                .append_records(&records[fresh..])?;
            self.records
                .extend_from_slice(&records[fresh..]);
            self.terms
                .extend_from_slice(&terms[fresh..]);
        }

        Ok(())
    }
}

impl Member {
    /// Removes the records after `keep`. Only records that are not
    /// committed can differ from the leader's, so none of them is applied.
    fn truncate(&mut self, keep: u64) -> io::Result<()> {
        if keep < self.committed {
            return Err(io::Error::other(format!(
                "record {} is committed and differs from the leader's",
                keep + 1
            )));
        }

        let length: usize = self.records[..keep as usize]
            .iter()
            .map(|record| record.encode().len())
            .sum();

        self.discarded += self.last_sequence() - keep;
        self.log
            .truncate(keep, length as u64)?;
        self.records.truncate(keep as usize);
        self.terms.truncate(keep as usize);

        Ok(())
    }
}

impl Member {
    fn on_appended(&mut self, answer: AppendAnswer) -> io::Result<Vec<Outgoing>> {
        if answer.term > self.term {
            self.step_down(answer.term)?;

            return Ok(Vec::new());
        }

        if self.role != Role::Leader || answer.term != self.term {
            return Ok(Vec::new());
        }

        let now = self.now;
        let Some(progress) = self.progress.get_mut(&answer.from) else {
            return Ok(Vec::new());
        };

        progress.heard = now;
        progress.next = answer.last_sequence;

        if answer.appended {
            progress.matched = progress
                .matched
                .max(answer.last_sequence);
        }

        let behind = !answer.appended || answer.last_sequence < self.last_sequence();
        self.advance();

        Ok(if behind {
            self.append_for(answer.from)
                .into_iter()
                .collect()
        } else {
            Vec::new()
        })
    }
}

impl Member {
    /// Whether this member is the leader and every record of its log is
    /// committed, so that it can accept a write.
    pub fn is_settled(&self) -> bool {
        self.role == Role::Leader && self.committed == self.last_sequence()
    }

    /// Checks the condition of a write and appends its record to the log.
    /// The caller sends the messages of `replicate` afterwards.
    pub fn propose(&mut self, change: Change, condition: Condition) -> Result<u64, StoreError> {
        let key = match &change {
            Change::Put { key, .. } | Change::Delete { key } => key,
        };
        let current = self
            .entries
            .get(key)
            .map(|entry| entry.version);

        if matches!(change, Change::Delete { .. }) && current.is_none() {
            return Err(StoreError::NotFound);
        }

        check(condition, current)?;

        let record = self.log.append(change)?;
        let sequence = record.sequence;
        self.records.push(record);
        self.terms.push(self.term);

        Ok(sequence)
    }

    /// Whether this member may answer a read: it is the leader, the marker
    /// of its term is committed, and a majority of the members, itself
    /// included, has answered within the last 10 ticks.
    pub fn can_read(&self) -> bool {
        let recent = self
            .progress
            .values()
            .filter(|progress| self.now - progress.heard < 10)
            .count();

        self.role == Role::Leader
            && self.term_at(self.committed) == self.term
            && recent + 1 >= self.majority()
    }

    pub fn get(&self, key: &str) -> Option<&Entry> {
        self.entries.get(key)
    }
}

Listing 4.46 The complete raftsim/src/main.rs. raftsim/src/main.rs

//! Command raftsim runs 3 members in one process and decides itself when
//! each clock ticks, which message arrives, which is lost, and which member
//! crashes. It first walks through an election and the repair of a log,
//! and then makes 200 runs of 600 random steps, checking after every step
//! that no 2 members lead the same term and that no acknowledged write is
//! missing.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p raftsim

use std::collections::BTreeMap;
use std::env;
use std::error::Error;
use std::fs;
use std::path::{Path, PathBuf};
use std::process;

use kv::{Change, Condition};
use raft::{Member, NodeId, Outgoing, Role};

/// A generator of pseudo-random numbers. The same seed gives the same
/// numbers, so a run that fails can be repeated exactly.
struct Random(u64);

impl Random {
    fn next(&mut self) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;

        self.0
    }

    /// A number below `bound`.
    fn below(&mut self, bound: usize) -> usize {
        (self.next() % bound as u64) as usize
    }
}

const MEMBERS: [NodeId; 3] = [1, 2, 3];

/// A write that a leader has accepted and not yet committed.
struct Pending {
    leader: NodeId,
    term: u64,
    sequence: u64,
    key: String,
}

struct Cluster {
    root: PathBuf,
    /// The members that are running. A crashed member is absent.
    up: BTreeMap<NodeId, Member>,
    /// The messages that are on their way.
    network: Vec<Outgoing>,
    pending: Vec<Pending>,
    /// The sequence and the key of every acknowledged write.
    acknowledged: Vec<(u64, String)>,
    /// The leader that was seen in each term.
    leaders: BTreeMap<u64, NodeId>,
    discarded: u64,
    writes: u64,
}

impl Cluster {
    fn open(root: &Path) -> Result<Cluster, Box<dyn Error>> {
        let mut cluster = Cluster {
            root: root.to_owned(),
            up: BTreeMap::new(),
            network: Vec::new(),
            pending: Vec::new(),
            acknowledged: Vec::new(),
            leaders: BTreeMap::new(),
            discarded: 0,
            writes: 0,
        };

        for id in MEMBERS {
            cluster.start(id)?;
        }

        Ok(cluster)
    }

    fn start(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        let peers = MEMBERS
            .iter()
            .copied()
            .filter(|&peer| peer != id)
            .collect();
        let member = Member::open(
            &self
                .root
                .join(format!("member-{id}")),
            id,
            peers,
            false,
        )?;
        self.up.insert(id, member);

        Ok(())
    }

    /// Stops a member. Its files remain, and its memory is lost.
    fn crash(&mut self, id: NodeId) {
        if let Some(member) = self.up.remove(&id) {
            self.discarded += member.discarded();
        }
    }

    fn tick(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        if let Some(member) = self.up.get_mut(&id) {
            self.network.extend(member.tick()?);
        }

        self.check()
    }

    /// Delivers the message at `index`. A message for a member that is
    /// down is lost.
    fn deliver(&mut self, index: usize) -> Result<(), Box<dyn Error>> {
        let Outgoing { to, message } = self.network.remove(index);

        if let Some(member) = self.up.get_mut(&to) {
            self.network
                .extend(member.handle(message)?);
        }

        self.check()
    }

    /// Delivers one of the messages that are on their way, chosen at random.
    fn deliver_any(&mut self, random: &mut Random) -> Result<(), Box<dyn Error>> {
        if self.network.is_empty() {
            return Ok(());
        }

        self.deliver(random.below(self.network.len()))
    }

    /// Loses one of the messages that are on their way.
    fn lose_any(&mut self, random: &mut Random) {
        if !self.network.is_empty() {
            self.network
                .remove(random.below(self.network.len()));
        }
    }

    /// Starts a member that is down.
    fn restart(&mut self, id: NodeId) -> Result<(), Box<dyn Error>> {
        if self.up.contains_key(&id) {
            return Ok(());
        }

        self.start(id)
    }

    /// Delivers messages until none is on its way.
    fn deliver_all(&mut self) -> Result<(), Box<dyn Error>> {
        while !self.network.is_empty() {
            self.deliver(0)?;
        }

        Ok(())
    }

    /// Sends a write to a leader that can accept one, if there is one.
    fn write(&mut self) -> Result<bool, Box<dyn Error>> {
        let Some(leader) = self
            .up
            .values_mut()
            .find(|member| member.is_settled())
        else {
            return Ok(false);
        };

        self.writes += 1;
        let key = format!("key-{}", self.writes);
        let change = Change::Put {
            key: key.clone(),
            value: b"1".to_vec(),
        };
        let sequence = leader.propose(change, Condition::Always)?;

        self.pending.push(Pending {
            leader: leader.id(),
            term: leader.term(),
            sequence,
            key,
        });
        self.network
            .extend(leader.replicate());

        Ok(true)
    }

    fn leader(&self) -> Option<&Member> {
        self.up
            .values()
            .filter(|member| member.role() == Role::Leader)
            .max_by_key(|member| member.term())
    }
}

impl Cluster {
    /// The properties that must hold after every step.
    fn check(&mut self) -> Result<(), Box<dyn Error>> {
        // No 2 members lead the same term.
        for member in self
            .up
            .values()
            .filter(|member| member.role() == Role::Leader)
        {
            let first = *self
                .leaders
                .entry(member.term())
                .or_insert(member.id());

            if first != member.id() {
                return Err(format!(
                    "members {first} and {} both lead term {}",
                    member.id(),
                    member.term()
                )
                .into());
            }
        }

        // A write is acknowledged when the leader that accepted it, still
        // leading the same term, has committed its record.
        let up = &self.up;
        let committed = |write: &Pending| {
            up.get(&write.leader)
                .is_some_and(|leader| {
                    leader.role() == Role::Leader
                        && leader.term() == write.term
                        && leader.committed() >= write.sequence
                })
        };
        let (done, waiting): (Vec<Pending>, Vec<Pending>) = self
            .pending
            .drain(..)
            .partition(committed);

        self.acknowledged.extend(
            done.into_iter()
                .map(|write| (write.sequence, write.key)),
        );
        self.pending = waiting;

        // Every member that has committed the sequence of an acknowledged
        // write holds that write at that sequence.
        for (sequence, key) in &self.acknowledged {
            for member in self
                .up
                .values()
                .filter(|member| member.committed() >= *sequence)
            {
                let holds = matches!(&member.records()[*sequence as usize - 1].change, Change::Put { key: held, .. } if held == key);

                if !holds {
                    return Err(format!(
                        "member {} lost the acknowledged write of {key}",
                        member.id()
                    )
                    .into());
                }
            }
        }

        Ok(())
    }
}

fn show(cluster: &Cluster) -> String {
    let describe = |id: &NodeId| match cluster.up.get(id) {
        Some(member) => format!(
            "{id}: {:?} term {} log {}",
            member.role(),
            member.term(),
            member.last_sequence()
        ),
        None => format!("{id}: down"),
    };

    MEMBERS
        .iter()
        .map(describe)
        .collect::<Vec<_>>()
        .join(" | ")
}

/// Ticks every running member, delivering all messages after each tick,
/// until a member leads. It returns the number of ticks.
fn elect(cluster: &mut Cluster) -> Result<u64, Box<dyn Error>> {
    let mut ticks = 0;

    while cluster
        .up
        .values()
        .all(|member| !member.is_settled())
    {
        ticks += 1;

        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    Ok(ticks)
}

fn cases(root: &Path) -> Result<(), Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;

    println!("start:            {}", show(&cluster));

    let ticks = elect(&mut cluster)?;
    println!("after {ticks:2} ticks:   {}", show(&cluster));

    for _ in 0..3 {
        cluster.write()?;
        cluster.deliver_all()?;
    }

    println!(
        "after 3 writes:   {}, {} acknowledged",
        show(&cluster),
        cluster.acknowledged.len()
    );

    // The leader accepts a write and crashes before it sends the record.
    let old = cluster
        .leader()
        .map(Member::id)
        .ok_or("no leader")?;
    cluster.write()?;
    cluster.network.clear();
    cluster.crash(old);
    println!(
        "leader {old} appends record 5 and crashes: {}",
        show(&cluster)
    );

    let ticks = elect(&mut cluster)?;
    println!("after {ticks:2} ticks:   {}", show(&cluster));

    cluster.write()?;
    cluster.deliver_all()?;
    println!(
        "after 1 write:    {}, {} acknowledged",
        show(&cluster),
        cluster.acknowledged.len()
    );

    cluster.start(old)?;
    println!("member {old} returns: {}", show(&cluster));

    for _ in 0..2 {
        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    let discarded = cluster
        .up
        .get(&old)
        .map_or(0, Member::discarded);
    println!("after 2 ticks:    {}", show(&cluster));
    println!(
        "member {old} discarded {discarded} record that was never committed, and holds the leader's records in its place"
    );

    Ok(())
}

const RUNS: u64 = 200;
const STEPS: usize = 600;

/// One run: random ticks, deliveries, losses, crashes, restarts and writes.
fn run(root: &Path, seed: u64) -> Result<Cluster, Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let mut random = Random(seed);

    for _ in 0..STEPS {
        let id = MEMBERS[random.below(MEMBERS.len())];

        match random.below(100) {
            0..40 => cluster.tick(id)?,
            40..82 => cluster.deliver_any(&mut random)?,
            82..87 => cluster.lose_any(&mut random),
            87..90 => cluster.crash(id),
            90..94 => cluster.restart(id)?,
            94.. => cluster.write().map(drop)?,
        }
    }

    recover(&mut cluster)?;

    Ok(cluster)
}

/// Ends the faults: every member runs and every message arrives. The
/// members must then agree on a leader and on a log that holds every
/// acknowledged write.
fn recover(cluster: &mut Cluster) -> Result<(), Box<dyn Error>> {
    for id in MEMBERS {
        if !cluster.up.contains_key(&id) {
            cluster.start(id)?;
        }
    }

    let agreed = |cluster: &Cluster| {
        cluster
            .leader()
            .is_some_and(|leader| {
                leader.is_settled()
                    && cluster
                        .up
                        .values()
                        .all(|member| member.records() == leader.records())
            })
    };

    for _ in 0..200 {
        if agreed(cluster) {
            break;
        }

        for id in MEMBERS {
            cluster.tick(id)?;
        }

        cluster.deliver_all()?;
    }

    if !agreed(cluster) {
        return Err("the members did not agree after the faults ended".into());
    }

    let leader = cluster
        .leader()
        .ok_or("the members have no leader after the faults ended")?;

    for (sequence, key) in &cluster.acknowledged {
        if leader
            .get(key)
            .is_none_or(|entry| entry.version != *sequence)
        {
            return Err(
                format!("the acknowledged write of {key} is missing after recovery").into(),
            );
        }
    }

    Ok(())
}

fn runs(root: &Path) -> Result<(), Box<dyn Error>> {
    let (mut leaders, mut acknowledged, mut discarded, mut most_terms) = (0, 0, 0, 0);

    for seed in 1..=RUNS {
        let directory = root.join(format!("run-{seed}"));
        let mut cluster = run(&directory, seed)?;

        for id in MEMBERS {
            cluster.crash(id);
        }

        leaders += cluster.leaders.len();
        acknowledged += cluster.acknowledged.len();
        discarded += cluster.discarded;
        most_terms = most_terms.max(
            cluster
                .leaders
                .keys()
                .next_back()
                .copied()
                .unwrap_or(0),
        );

        fs::remove_dir_all(&directory)?;
    }

    println!("{RUNS} runs of {STEPS} random steps:");
    println!("  {leaders} leaders were elected, and no term had 2 leaders");
    println!("  {acknowledged} writes were acknowledged, and every one was present after recovery");
    println!(
        "  {discarded} uncommitted records were discarded by members whose log differed from a leader's"
    );
    println!("  the highest term reached in a run was {most_terms}");

    Ok(())
}

fn main() -> Result<(), Box<dyn Error>> {
    let root = env::temp_dir().join(format!("raftsim-{}", process::id()));
    let result = cases(&root.join("cases")).and_then(|()| runs(&root));

    fs::remove_dir_all(&root)?;

    result
}

Listing 4.47 The complete kvraft/Cargo.toml. kvraft/Cargo.toml

[package]
name = "kvraft"
version = "0.1.0"
edition.workspace = true
rust-version.workspace = true
license.workspace = true
publish.workspace = true

[dependencies]
kv.workspace = true
kvd.workspace = true
raft.workspace = true
axum.workspace = true
clap.workspace = true
parking_lot.workspace = true
prost.workspace = true
serde.workspace = true
tokio.workspace = true
tonic.workspace = true
tonic-prost.workspace = true

[build-dependencies]
tonic-prost-build.workspace = true

Listing 4.48 The complete kvraft/src/rpc.rs. kvraft/src/rpc.rs

//! The gRPC services of a member: the generated types, their conversion to
//! the messages of the `raft` crate, and the server side of both calls.

use std::sync::Arc;

use kv::{Change, Record};
use raft::{AppendAnswer, AppendRequest, Message, NodeId, VoteAnswer, VoteRequest};
use tonic::{Request, Response, Status};

use crate::node::Node;

/// The types that `build.rs` generates from `proto/store.proto`. The schema
/// also defines the ring, which this program does not use.
#[allow(dead_code)]
pub mod proto {
    tonic::include_proto!("store.v1");
}

pub use proto::election_client::ElectionClient;
pub use proto::election_server::{Election, ElectionServer};
pub use proto::replication_client::ReplicationClient;
pub use proto::replication_server::{Replication, ReplicationServer};

fn record_to_proto(record: &Record) -> proto::Record {
    let change = match record.change.clone() {
        Change::Put { key, value } => proto::record::Change::Put(proto::Put { key, value }),
        Change::Delete { key } => proto::record::Change::Delete(proto::Delete { key }),
    };

    proto::Record {
        sequence: record.sequence,
        change: Some(change),
    }
}

fn record_from_proto(record: proto::Record) -> Result<Record, Status> {
    let change = match record.change {
        Some(proto::record::Change::Put(put)) => Change::Put {
            key: put.key,
            value: put.value,
        },
        Some(proto::record::Change::Delete(delete)) => Change::Delete { key: delete.key },
        None => return Err(Status::invalid_argument("a record without a change")),
    };

    Ok(Record {
        sequence: record.sequence,
        change,
    })
}

impl From<AppendRequest> for proto::AppendRequest {
    fn from(request: AppendRequest) -> Self {
        proto::AppendRequest {
            shard: 0,
            term: request.term,
            previous_sequence: request.previous_sequence,
            records: request
                .records
                .iter()
                .map(record_to_proto)
                .collect(),
            committed_sequence: request.committed_sequence,
            previous_term: request.previous_term,
            leader: request.leader,
        }
    }
}

impl TryFrom<proto::AppendRequest> for AppendRequest {
    type Error = Status;

    fn try_from(request: proto::AppendRequest) -> Result<Self, Status> {
        Ok(AppendRequest {
            term: request.term,
            leader: request.leader,
            previous_sequence: request.previous_sequence,
            previous_term: request.previous_term,
            records: request
                .records
                .into_iter()
                .map(record_from_proto)
                .collect::<Result<_, _>>()?,
            committed_sequence: request.committed_sequence,
        })
    }
}

impl From<VoteRequest> for proto::VoteRequest {
    fn from(request: VoteRequest) -> Self {
        proto::VoteRequest {
            term: request.term,
            candidate: request.candidate,
            last_sequence: request.last_sequence,
            last_term: request.last_term,
        }
    }
}

/// The answer of member `from` to an `Append` call, as a message.
pub fn appended(from: NodeId, answer: proto::AppendResponse) -> Message {
    Message::Appended(AppendAnswer {
        term: answer.term,
        from,
        appended: answer.appended,
        last_sequence: answer.last_sequence,
    })
}

/// The answer of member `from` to a `RequestVote` call, as a message.
pub fn voted(from: NodeId, answer: proto::VoteResponse) -> Message {
    Message::Voted(VoteAnswer {
        term: answer.term,
        from,
        granted: answer.granted,
    })
}

/// The server side of both services. Each call is one message for the
/// member, and the member's answer is the response of the call.
pub struct Service {
    pub node: Arc<Node>,
}

#[tonic::async_trait]
impl Replication for Service {
    async fn append(
        &self,
        request: Request<proto::AppendRequest>,
    ) -> Result<Response<proto::AppendResponse>, Status> {
        let request = AppendRequest::try_from(request.into_inner())?;

        let answer = self
            .node
            .answer(Message::Append(request))
            .map_err(|error| Status::unavailable(error.to_string()))?;

        let Message::Appended(answer) = answer else {
            return Err(Status::internal(
                "the member answered with another kind of message",
            ));
        };

        Ok(Response::new(proto::AppendResponse {
            term: answer.term,
            appended: answer.appended,
            last_sequence: answer.last_sequence,
        }))
    }
}

#[tonic::async_trait]
impl Election for Service {
    async fn request_vote(
        &self,
        request: Request<proto::VoteRequest>,
    ) -> Result<Response<proto::VoteResponse>, Status> {
        let request = request.into_inner();
        let request = VoteRequest {
            term: request.term,
            candidate: request.candidate,
            last_sequence: request.last_sequence,
            last_term: request.last_term,
        };

        let answer = self
            .node
            .answer(Message::Vote(request))
            .map_err(|error| Status::unavailable(error.to_string()))?;

        let Message::Voted(answer) = answer else {
            return Err(Status::internal(
                "the member answered with another kind of message",
            ));
        };

        Ok(Response::new(proto::VoteResponse {
            term: answer.term,
            granted: answer.granted,
        }))
    }
}

Listing 4.49 The complete kvraft/src/node.rs. kvraft/src/node.rs

//! A node: a member, its clock, and the connections that carry its
//! messages to the other members.

use std::collections::BTreeMap;
use std::io;
use std::sync::Arc;
use std::time::Duration;

use kv::{Change, Condition, Entry, Written};
use kvd::api::ApiError;
use parking_lot::Mutex;
use raft::{Member, Message, NodeId, Outgoing, Role};
use tokio::sync::watch;
use tokio::task;
use tokio::time::{interval, timeout};
use tonic::transport::{Channel, Endpoint};

use crate::rpc::{ElectionClient, ReplicationClient, appended, proto, voted};

/// The length of a tick of the member's clock.
pub const TICK: Duration = Duration::from_millis(100);

/// How long a member waits for the answer to one message.
const ANSWER: Duration = Duration::from_millis(500);

/// How long a write waits for its record to be committed.
const COMMIT: Duration = Duration::from_secs(1);

pub struct Node {
    pub member: Mutex<Member>,
    /// A connection to each of the other members.
    peers: BTreeMap<NodeId, Channel>,
    /// Held by the write that is in progress.
    turn: tokio::sync::Mutex<()>,
    /// The committed sequence, for writes that wait for it to advance.
    committed: watch::Sender<u64>,
}

impl Node {
    pub fn new(
        member: Member,
        addresses: BTreeMap<NodeId, String>,
    ) -> Result<Arc<Node>, tonic::transport::Error> {
        let mut peers = BTreeMap::new();

        for (id, address) in addresses {
            peers.insert(
                id,
                Endpoint::from_shared(address)?
                    .connect_timeout(ANSWER)
                    .connect_lazy(),
            );
        }

        Ok(Arc::new(Node {
            member: Mutex::new(member),
            peers,
            turn: tokio::sync::Mutex::new(()),
            committed: watch::Sender::new(0),
        }))
    }
}

impl Node {
    /// Runs one operation on the member and publishes the committed
    /// sequence. The member waits for the disk, so the operation runs where
    /// blocking is allowed.
    fn with_member<T>(&self, operation: impl FnOnce(&mut Member) -> T) -> T {
        task::block_in_place(|| {
            let mut member = self.member.lock();
            let result = operation(&mut member);
            self.committed
                .send_replace(member.committed());

            result
        })
    }
}

impl Node {
    /// Sends each message in a task of its own, so that a member that does
    /// not answer delays nothing else.
    fn dispatch(self: &Arc<Node>, messages: Vec<Outgoing>) {
        for message in messages {
            tokio::spawn(Arc::clone(self).send(message));
        }
    }

    /// Carries one request to a member and gives the answer to this member.
    async fn send(self: Arc<Node>, outgoing: Outgoing) {
        let Some(channel) = self
            .peers
            .get(&outgoing.to)
            .cloned()
        else {
            return;
        };

        let answer = match outgoing.message {
            Message::Append(request) => {
                let call = async move {
                    ReplicationClient::new(channel)
                        .append(proto::AppendRequest::from(request))
                        .await
                };

                timeout(ANSWER, call)
                    .await
                    .ok()
                    .and_then(Result::ok)
                    .map(|answer| appended(outgoing.to, answer.into_inner()))
            }
            Message::Vote(request) => {
                let call = async move {
                    ElectionClient::new(channel)
                        .request_vote(proto::VoteRequest::from(request))
                        .await
                };

                timeout(ANSWER, call)
                    .await
                    .ok()
                    .and_then(Result::ok)
                    .map(|answer| voted(outgoing.to, answer.into_inner()))
            }
            // An answer travels as the response of the call that asked.
            Message::Appended(..) | Message::Voted(..) => None,
        };

        if let Some(answer) = answer {
            self.deliver(answer);
        }
    }

    /// Gives the member a message and sends what the member returns.
    fn deliver(self: &Arc<Node>, message: Message) {
        match self.with_member(|member| member.handle(message)) {
            Ok(messages) => self.dispatch(messages),
            Err(error) => eprintln!("the member could not process a message: {error}"),
        }
    }
}

impl Node {
    /// The member's answer to a request that arrived over gRPC.
    pub fn answer(&self, request: Message) -> io::Result<Message> {
        let mut messages = self.with_member(|member| member.handle(request))?;

        messages
            .pop()
            .map(|outgoing| outgoing.message)
            .ok_or_else(|| io::Error::other("the member gave no answer"))
    }

    /// Advances the member's clock, for as long as the node runs.
    pub async fn run_clock(self: Arc<Node>) {
        let mut ticks = interval(TICK);

        loop {
            ticks.tick().await;

            match self.with_member(Member::tick) {
                Ok(messages) => self.dispatch(messages),
                Err(error) => eprintln!("the member could not advance its clock: {error}"),
            }
        }
    }
}

impl Node {
    fn not_leader(member: &Member) -> ApiError {
        match member.leader() {
            Some(leader) => ApiError::Misdirected(format!("the leader is member {leader}")),
            None => ApiError::NoLeader,
        }
    }

    /// A write on the leader. It is acknowledged when its record is
    /// committed and this member has led the same term throughout.
    pub async fn write(
        self: &Arc<Node>,
        change: Change,
        condition: Condition,
    ) -> Result<Written, ApiError> {
        let turn = self.turn.lock().await;

        let last = {
            let member = self.member.lock();

            if member.role() != Role::Leader {
                return Err(Node::not_leader(&member));
            }

            member.last_sequence()
        };

        self.wait_for(last).await?;

        let (written, term, messages) = self.with_member(|member| {
            if !member.is_settled() {
                return Err(Node::not_leader(member));
            }

            let created = match &change {
                Change::Put { key, .. } | Change::Delete { key } => member.get(key).is_none(),
            };
            let version = member.propose(change, condition)?;

            Ok((
                Written { version, created },
                member.term(),
                member.replicate(),
            ))
        })?;

        self.dispatch(messages);
        self.wait_for(written.version)
            .await?;
        drop(turn);

        // A member that lost the leadership while it waited may hold another
        // leader's record at this sequence.
        let member = self.member.lock();

        if member.role() == Role::Leader && member.term() == term {
            Ok(written)
        } else {
            Err(ApiError::Unavailable(
                "the leadership changed before the record was committed".into(),
            ))
        }
    }

    async fn wait_for(&self, sequence: u64) -> Result<(), ApiError> {
        let mut committed = self.committed.subscribe();

        let reached = committed.wait_for(|&committed| committed >= sequence);
        let in_time = timeout(COMMIT, reached)
            .await
            .is_ok_and(|reached| reached.is_ok());

        if in_time {
            Ok(())
        } else {
            Err(ApiError::Unavailable(format!(
                "record {sequence} is not on a majority of the members"
            )))
        }
    }

    pub fn get(&self, key: &str) -> Result<Entry, ApiError> {
        let member = self.member.lock();

        if member.role() != Role::Leader {
            return Err(Node::not_leader(&member));
        }

        if !member.can_read() {
            return Err(ApiError::Unavailable(
                "this leader cannot confirm that it still leads".into(),
            ));
        }

        member
            .get(key)
            .cloned()
            .ok_or(ApiError::NotFound)
    }
}

Listing 4.50 The complete kvraft/src/main.rs. kvraft/src/main.rs

//! Command kvraft is a storage node whose shard elects its own leader.
//! Every node is started the same way, with its id and the replication
//! addresses of the other members.
//!
//! Run 3 nodes from code/ with:
//!
//!     cargo run -p kvraft -- --id 1 --listen 127.0.0.1:7001 --replication 127.0.0.1:7101 --data /var/tmp/m1 \
//!         --peer 2=http://127.0.0.1:7102 --peer 3=http://127.0.0.1:7103
//!
//! and the same for ids 2 and 3, each with its own ports and the other 2 as peers.

mod node;
mod rpc;

use std::collections::BTreeMap;
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::header::{CONTENT_TYPE, ETAG, IF_NONE_MATCH};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use clap::Parser;
use kv::{Change, Condition};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use raft::{Member, NodeId, Role};
use serde::Serialize;
use tokio::net::TcpListener;
use tonic::transport::Server;

use crate::node::Node;
use crate::rpc::{ElectionServer, ReplicationServer, Service};

/// Reads `2=http://127.0.0.1:7102` as member 2 at that address.
fn parse_peer(text: &str) -> Result<(NodeId, String), String> {
    let (id, address) = text
        .split_once('=')
        .ok_or("a peer is written as id=address")?;
    let Ok(id) = id.parse() else {
        return Err("the id of a peer is a number".to_owned());
    };

    Ok((id, address.to_owned()))
}

#[derive(Parser)]
struct Options {
    /// The id of this member, a number that no other member has.
    #[arg(long)]
    id: NodeId,

    /// The address of the HTTP API.
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    /// The address on which this member accepts messages from the others.
    #[arg(long, default_value = "127.0.0.1:7101")]
    replication: SocketAddr,

    /// Another member and its replication address, as id=address.
    #[arg(long = "peer", value_parser = parse_peer)]
    peers: Vec<(NodeId, String)>,

    #[arg(long, default_value = "data")]
    data: PathBuf,

    /// Respond without waiting for the disk.
    #[arg(long)]
    no_sync: bool,
}

async fn put_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    let written = node
        .write(
            Change::Put {
                key,
                value: body.to_vec(),
            },
            condition,
        )
        .await?;
    let status = if written.created {
        StatusCode::CREATED
    } else {
        StatusCode::NO_CONTENT
    };

    Ok((status, [(ETAG, etag(written.version))]).into_response())
}

async fn delete_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    node.write(Change::Delete { key }, condition)
        .await?;

    Ok(StatusCode::NO_CONTENT)
}

async fn get_key(
    State(node): State<Arc<Node>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let cached = version_header(&headers, IF_NONE_MATCH)?;
    let entry = node.get(&key)?;
    let tag = [(ETAG, etag(entry.version))];

    if cached == Some(Condition::Version(entry.version)) {
        return Ok((StatusCode::NOT_MODIFIED, tag).into_response());
    }

    Ok((
        tag,
        [(CONTENT_TYPE, "application/octet-stream")],
        entry.value,
    )
        .into_response())
}

#[derive(Serialize)]
struct Health {
    id: NodeId,
    role: &'static str,
    term: u64,
    /// The leader of the term, when this member knows it.
    leader: Option<NodeId>,
    last_sequence: u64,
    committed: u64,
}

async fn health(State(node): State<Arc<Node>>) -> Json<Health> {
    let member = node.member.lock();

    let role = match member.role() {
        Role::Leader => "leader",
        Role::Candidate => "candidate",
        Role::Follower => "follower",
    };

    Json(Health {
        id: member.id(),
        role,
        term: member.term(),
        leader: member.leader(),
        last_sequence: member.last_sequence(),
        committed: member.committed(),
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let peers: BTreeMap<NodeId, String> = options.peers.into_iter().collect();
    let member = Member::open(
        &options.data,
        options.id,
        peers.keys().copied().collect(),
        !options.no_sync,
    )?;

    println!(
        "member {}, term {}, log ends at {}",
        member.id(),
        member.term(),
        member.last_sequence()
    );

    let node = Node::new(member, peers)?;

    let replication = ReplicationServer::new(Service {
        node: Arc::clone(&node),
    });
    let election = ElectionServer::new(Service {
        node: Arc::clone(&node),
    });

    tokio::spawn(
        Server::builder()
            .add_service(replication)
            .add_service(election)
            .serve(options.replication),
    );
    tokio::spawn(Arc::clone(&node).run_clock());

    let routes = Router::new()
        .route(
            "/v1/keys/{key}",
            get(get_key)
                .put(put_key)
                .delete(delete_key),
        )
        .route("/healthz", get(health))
        .layer(DefaultBodyLimit::max(MAX_VALUE))
        .with_state(node);

    let listener = TcpListener::bind(options.listen).await?;
    println!("listening on {}", listener.local_addr()?);

    axum::serve(listener, routes).await?;

    Ok(())
}

Listing 4.51 The complete probe/src/main.rs. probe/src/main.rs

//! Command probe measures how long a shard is unavailable when its leader
//! is lost. It writes keys for a number of seconds, trying each node in turn
//! until one accepts the write, and reports the longest pause between 2
//! acknowledged writes.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p probe -- --seconds 20 --acked acked.txt \
//!         --server http://127.0.0.1:7001 --server http://127.0.0.1:7002 --server http://127.0.0.1:7003

use std::error::Error;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::PathBuf;
use std::thread;
use std::time::{Duration, Instant};

use clap::Parser;
use ureq::Agent;

#[derive(Parser)]
struct Options {
    /// The HTTP address of a node. Given once for each node of the shard.
    #[arg(long = "server", required = true)]
    servers: Vec<String>,

    #[arg(long, default_value_t = 20)]
    seconds: u64,

    /// The file that receives the key of every acknowledged write.
    #[arg(long)]
    acked: PathBuf,
}

fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let agent: Agent = Agent::config_builder()
        .timeout_global(Some(Duration::from_secs(2)))
        .build()
        .into();
    let mut file = BufWriter::new(File::create(&options.acked)?);

    let started = Instant::now();
    let (mut server, mut number, mut refused) = (0, 1, 0);
    let (mut last_success, mut longest, mut longest_at) = (started, Duration::ZERO, Duration::ZERO);

    while started.elapsed() < Duration::from_secs(options.seconds) {
        let key = format!("key-{number:06}");
        let address = format!("{}/v1/keys/{key}", options.servers[server]);

        if agent
            .put(&address)
            .send(format!("value of {key}"))
            .is_err()
        {
            // This node did not accept the write. Try the next one.
            refused += 1;
            server = (server + 1) % options.servers.len();
            thread::sleep(Duration::from_millis(20));

            continue;
        }

        let pause = last_success.elapsed();

        if pause > longest {
            (longest, longest_at) = (pause, started.elapsed());
        }

        last_success = Instant::now();
        writeln!(file, "{key}")?;
        number += 1;
    }

    file.flush()?;

    println!(
        "{} writes acknowledged in {} seconds, {refused} attempts refused",
        number - 1,
        options.seconds
    );
    println!(
        "the longest pause between 2 acknowledged writes was {:.2} s, ending {:.1} s after the start",
        longest.as_secs_f64(),
        longest_at.as_secs_f64()
    );

    Ok(())
}

Further reading

• D. Ongaro and J. Ousterhout, “In Search of an Understandable Consensus Algorithm”, USENIX ATC 2014.
• D. Ongaro, Consensus: Bridging Theory and Practice, doctoral dissertation, Stanford University, 2014.
• L. Lamport, “Paxos Made Simple”, ACM SIGACT News, 2001.
• T. Chandra, R. Griesemer and J. Redstone, “Paxos Made Live: An Engineering Perspective”, PODC 2007.
• M. Fischer, N. Lynch and M. Paterson, “Impossibility of Distributed Consensus with One Faulty Process”, Journal of the ACM, 1985.
• K. Kingsbury, the Jepsen analyses, jepsen.io.