Key-Value Store: Replication (Low-Level Design)

The storage node of chapter 2 keeps every acknowledged write when its process crashes. Against the loss of the machine itself, or of its disk, it offers nothing: a log on 1 disk is gone when that disk is gone. Non-functional requirement 0 of chapter 1 demands that an acknowledged write survive the loss of any 1 machine, and it follows that the record of every write has to be stored on at least 2 machines before the client is answered.

To copy a file from 1 machine to another is easy. What makes replication a subject is that the copying takes place over a network which loses messages, delays them and delivers their answers late or not at all, between machines that stop and restart at arbitrary moments, and that through all of this the copies must remain identical, in content and in order, record for record. A follower that has missed a message must not apply the next one as though nothing were missing. A leader that has lost an answer must not conclude that the follower lacks the record, nor that it has it. And a machine that was the leader an hour ago and has since been replaced must be prevented from writing as though it still were.

This chapter implements the replication that chapter 1 specified: a leader that orders the writes of a shard, and 2 followers that store the same records in the same order. It turns out that a single small rule on the follower’s side handles every one of the situations listed above. We define the messages and the state of a replica, derive the rules for followers and leaders, and implement them as a library that performs no network operations whatever, so that a simulation can subject it to every possible combination of lost messages. We then build kvnode, a storage node that exchanges the messages over gRPC, and run 3 nodes while stopping followers, removing the majority, and replacing a lost leader.

Overview

Every shard has 3 replicas. All writes go to the leader, and the other 2 replicas are followers. Figure 3.1 shows the nodes of this chapter and the route of a write among them.

The implementation maintains 4 invariants.

1. The log of every follower is a prefix of its leader’s log: the same records in the same order, and possibly fewer of them.
2. A record is acknowledged to its client only after at least 2 of the 3 logs hold it.
3. A replica applies a record to its entries only after the record is known to be held in at least 2 of the 3 logs.
4. A replica rejects every message from a leader whose term is lower than its own.

The second invariant is non-functional requirement 0 in the form of a rule. The third ensures that a read returns only data which would survive the loss of a machine. The fourth prevents a former leader from modifying a shard after it has been replaced.

In this chapter the leader of a shard is chosen by an operator, who starts 1 node in that role. Automatic election, as chapter 1 described it, is built upon the mechanism of this chapter and is the subject of chapter 4.

The messages

The protocol of chapter 1 defines a single call, Append, which a leader sends to each of its followers.

Listing 3.1 The replication service in the schema. 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;
}

The library uses types of its own for these messages, so that it has no dependence on gRPC, and the conversion between the 2 representations is performed in the node.

Listing 3.2 The imports of repl/src/lib.rs. repl/src/lib.rs

use std::collections::BTreeMap;
use std::io;
use std::path::Path;

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

Listing 3.3 The message and its answer, as the library uses them. repl/src/lib.rs

/// The message a leader sends to a follower.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Append {
    /// The term of the sender.
    pub term: u64,
    /// The sequence of the record that precedes `records` in the leader's log.
    pub previous_sequence: u64,
    pub records: Vec<Record>,
    /// Every record up to this sequence is on a majority of the replicas.
    pub committed_sequence: u64,
}

/// The follower's answer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Appended {
    pub term: u64,
    /// False when the follower's log does not end at `previous_sequence`,
    /// or when the sender's term is older than the follower's.
    pub appended: bool,
    /// The sequence of the last record in the follower's log.
    pub last_sequence: u64,
}

The last field is the one on which the whole protocol turns. A follower never merely says yes or no. In every answer it reports where its log now ends, and this single number is enough for a leader to correct any mistaken belief it may hold about that follower.

The state of a replica

A replica consists of a log, of the entries that result from applying a part of that log, and of 3 numbers: the sequence of its last record, the point up to which records have been applied, and its term.

A record is committed when a majority of the replicas has stored it. A committed record survives the loss of any 1 machine, since any 2 of the 3 replicas include at least 1 that holds it. A replica keeps uncommitted records in its log without applying them. The entries therefore reflect committed records only, and no read can return a value that might yet be lost.

Figure 3.3 shows where each part of this state resides and which operation carries data from 1 part to another.

In the code, records is a vector that holds record n at index n - 1, so that its length is the sequence of its last record. commit proceeds from committed to the sequence it is given and passes each record to apply, the function of chapter 2 that inserts or removes an entry. Since commit is the only caller of apply, there is no path by which an uncommitted record can reach the entries.

Listing 3.4 The replica, and the application of committed records. repl/src/lib.rs

pub struct Replica {
    log: Log,
    /// Every record of the log, in order. Record `n` is at index `n - 1`.
    records: Vec<Record>,
    /// The result of applying the records up to `committed`.
    entries: BTreeMap<String, Entry>,
    /// The sequence of the last record that is applied to `entries`.
    committed: u64,
    term: u64,
}

impl Replica {
    /// Opens the replica in `directory`. Nothing is applied yet: a replica
    /// that has restarted does not know which of its records are committed.
    pub fn open(directory: &Path, sync: bool, term: u64) -> io::Result<Replica> {
        let (log, recovery) = Log::open(directory, sync)?;

        Ok(Replica {
            log,
            records: recovery.records,
            entries: BTreeMap::new(),
            committed: 0,
            term,
        })
    }

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

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

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

    /// One page of the committed keys.
    pub fn list(&self, prefix: &str, after: Option<&str>, limit: usize) -> (Vec<Summary>, bool) {
        page(&self.entries, prefix, after, limit)
    }

    /// Every record of the log, in order.
    pub fn records(&self) -> &[Record] {
        &self.records
    }

    /// The committed value of a key.
    pub fn get(&self, key: &str) -> Option<&Entry> {
        self.entries.get(key)
    }

    /// Applies the records up to `sequence` that are not applied yet.
    pub fn commit(&mut self, sequence: u64) {
        let until = sequence.min(self.last_sequence());

        while self.committed < until {
            apply(
                &mut self.entries,
                self.records[self.committed as usize].clone(),
            );
            self.committed += 1;
        }
    }
}

Replica::open sets committed to 0 and leaves the entries empty, although the log may contain records. This may seem wasteful, and it is in fact necessary. After a restart, a replica cannot tell from its own log which of its records had reached a majority, since its last few records may exist on its own disk and nowhere else. A follower learns the committed sequence from its leader, and a leader learns it from the answers of its followers.

The follower

A follower applies the same procedure to every message it receives.

handle_append is a direct transcription of that figure. Each check that fails returns by way of a closure named refuse, which builds an answer from the follower’s present term and the end of its log. When both checks pass, append_records writes the whole batch with a single wait for the disk, and commit then applies whatever the leader has declared committed. Since commit stops at the end of the log, a committed sequence that lies beyond what this follower holds does no harm.

Listing 3.5 The follower's handling of a message. repl/src/lib.rs

impl Replica {
    /// Processes a message from a leader.
    pub fn handle_append(&mut self, request: Append) -> io::Result<Appended> {
        let refuse = |replica: &Replica| Appended {
            term: replica.term,
            appended: false,
            last_sequence: replica.last_sequence(),
        };

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

        self.term = request.term;

        if request.previous_sequence != self.last_sequence() {
            return Ok(refuse(self));
        }

        if !request.records.is_empty() {
            self.log
                .append_records(&request.records)?;
            self.records.extend(request.records);
        }

        self.commit(request.committed_sequence);

        Ok(Appended {
            term: self.term,
            appended: true,
            last_sequence: self.last_sequence(),
        })
    }
}

The second row of the table is what maintains invariant 1, and it deserves to be looked at closely, since the whole reliability of replication derives from it. A follower appends records only at the exact position where the leader’s message begins. Its log can therefore never acquire a gap, a duplicate or a record out of place, whatever has happened to earlier messages. When the positions differ, the follower attempts no repair of any kind. It reports where its log ends, and the leader sends a message that begins there.

This 1 exchange covers every circumstance in which a follower is behind or the leader’s information is out of date: a follower that was stopped and has restarted, a message that was lost, an answer that was lost, and a new follower with an empty log. None of these cases has code of its own. They are all the same case, in which the leader’s belief about the end of the follower’s log is wrong and the follower’s answer puts it right.

The records of a message are appended with Log::append_records, which writes all of them and waits for the disk once.

Listing 3.6 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 3.7 Appending a batch of records that another node has numbered. kv/src/log.rs

impl Log {
    /// Appends records that another node numbered, in one write and with one
    /// wait for the disk. The first record must follow the last record of
    /// this log.
    pub fn append_records(&mut self, records: &[Record]) -> io::Result<()> {
        if self.failed {
            return Err(io::Error::other(
                "the log failed on an earlier write and is closed",
            ));
        }

        let misplaced = (self.next..)
            .zip(records)
            .find(|(expected, record)| record.sequence != *expected);

        if let Some((expected, record)) = misplaced {
            return Err(io::Error::other(format!(
                "record {} cannot follow record {}",
                record.sequence,
                expected - 1
            )));
        }

        let bytes: Vec<u8> = records
            .iter()
            .flat_map(Record::encode)
            .collect();

        if let Err(error) = self.write(&bytes) {
            self.failed = true;

            return Err(error);
        }

        self.next += records.len() as u64;

        Ok(())
    }
}

The leader

A leader has 3 responsibilities: to accept writes, to send each follower whatever records it lacks, and to determine which records are committed.

Proposing a record

propose checks the condition of a write against the entries and appends the record to the leader’s own log. The record is not yet applied, since at this point only 1 replica holds it.

A leader accepts a new write only when is_settled is true, which is to say when every record in its log is committed. The reason lies in conditional writes. A condition is checked against the entries, and the entries do not include uncommitted records. If the leader were to accept a second write while an earlier record was still uncommitted, the condition of the second would be evaluated against a state that lacked its predecessor, and a conditional write could succeed when it ought to have failed. Requiring a settled log guarantees that every condition is evaluated against all earlier writes. The price is that the writes of a shard are processed strictly in sequence. Production systems evaluate conditions against a speculative state and process writes in batches, which is considerably more intricate and is not attempted here, although chapter 6 recovers much of the lost throughput by another means.

Within propose, the present version of the key is read from the entries, a Delete of a key that does not exist is rejected, and check evaluates the condition. Only then is the record appended. On return, the record is in the leader’s log and in records, and the entries are unchanged.

Listing 3.8 Checking the condition of a write and appending its record. repl/src/lib.rs

impl Replica {
    /// Whether every record of the log is committed. A leader accepts a new
    /// write only in this state, so that the condition of the write is
    /// checked against the result of all earlier writes.
    pub fn is_settled(&self) -> bool {
        self.committed == self.last_sequence()
    }

    /// Checks the condition of a write and appends its record to the
    /// leader's log. The record is not applied: it is not committed yet.
    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);

        Ok(sequence)
    }
}

Building a message for a follower

For every follower, the leader remembers the last sequence that the follower reported. append_for takes that number and returns a message containing every later record. For a follower that is up to date, the slice of records is empty and the message is a heartbeat. A reported sequence that lies beyond the end of the leader’s own log is limited to that end, a situation that can arise when a leader has itself been replaced and restarted with a shorter log.

Listing 3.9 The message that brings 1 follower up to date. repl/src/lib.rs

impl Replica {
    /// The message that brings a follower whose log ends at `follower_last`
    /// up to the end of this log. With nothing to send it is a heartbeat.
    pub fn append_for(&self, follower_last: u64) -> Append {
        let from = follower_last.min(self.last_sequence()) as usize;

        Append {
            term: self.term,
            previous_sequence: from as u64,
            records: self.records[from..].to_vec(),
            committed_sequence: self.committed,
        }
    }
}

The commit rule

The leader knows where its own log ends and what last sequence each follower has reported. If these numbers are sorted, the one in the middle is the highest sequence that a majority has stored.

majority_sequence implements the rule for any number of replicas, by taking the element at index n / 2 of the values sorted in descending order. For 3 replicas that is the second value, and for 5 the third.

Listing 3.10 The commit rule. repl/src/lib.rs

/// The highest sequence that is stored on a majority, given the last
/// sequence of every replica, the leader included.
pub fn majority_sequence(mut last_sequences: Vec<u64>) -> u64 {
    last_sequences.sort_unstable_by(|a, b| b.cmp(a));

    last_sequences
        .get(last_sequences.len() / 2)
        .copied()
        .unwrap_or(0)
}

Simulating lost messages

Because the library performs no network operations, a test can carry the messages between replicas itself and decide the fate of every exchange. This is the reason for writing the rules as a library in the first place. Faults in a real network occur rarely and unrepeatably, and a protocol that has only been tested on a real network has scarcely been tested against faults at all.

replsim creates a leader and 2 followers in a single process. For each exchange it chooses 1 of 3 outcomes: the message is delivered and answered, the message is lost, or the message is processed and its answer is lost.

Listing 3.11 The imports of replsim/src/main.rs. replsim/src/main.rs

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

use kv::{Change, Condition};
use repl::{Replica, majority_sequence};

Listing 3.12 A cluster in 1 process, with control over every exchange. replsim/src/main.rs

/// What happens to one exchange between the leader and a follower.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Fate {
    Delivered,
    /// The follower does not receive the message.
    RequestLost,
    /// The follower processes the message, and its answer does not arrive.
    AnswerLost,
}

/// A leader, 2 followers, and what the leader knows about each follower.
struct Cluster {
    leader: Replica,
    followers: Vec<Replica>,
    /// The last sequence each follower has reported.
    reported: Vec<u64>,
}

impl Cluster {
    fn open(root: &Path) -> Result<Cluster, Box<dyn Error>> {
        let open = |name: &str| Replica::open(&root.join(name), false, 1);

        Ok(Cluster {
            leader: open("leader")?,
            followers: vec![open("follower-1")?, open("follower-2")?],
            reported: vec![0, 0],
        })
    }

    /// One exchange with a follower: the leader sends what the follower
    /// lacks, as far as it knows, and commits what a majority now holds.
    fn exchange(&mut self, follower: usize, fate: Fate) -> Result<(), Box<dyn Error>> {
        let message = self
            .leader
            .append_for(self.reported[follower]);

        if fate == Fate::RequestLost {
            return Ok(());
        }

        let answer = self.followers[follower].handle_append(message)?;

        if fate == Fate::AnswerLost {
            return Ok(());
        }

        self.reported[follower] = answer.last_sequence;

        let stored = vec![
            self.leader.last_sequence(),
            self.reported[0],
            self.reported[1],
        ];
        self.leader
            .commit(majority_sequence(stored));

        Ok(())
    }

    /// A write: the leader appends the record and makes one exchange with
    /// each follower. It returns whether the write was acknowledged, or
    /// `None` when the leader could not accept it.
    fn write(
        &mut self,
        key: &str,
        value: &str,
        fates: [Fate; 2],
    ) -> Result<Option<bool>, Box<dyn Error>> {
        if !self.leader.is_settled() {
            return Ok(None);
        }

        let change = Change::Put {
            key: key.into(),
            value: value.into(),
        };
        let sequence = self
            .leader
            .propose(change, Condition::Always)?;

        self.exchange(0, fates[0])?;
        self.exchange(1, fates[1])?;

        Ok(Some(self.leader.committed() >= sequence))
    }

    /// The last sequence of the leader and of each follower.
    fn lasts(&self) -> [u64; 3] {
        [
            self.leader.last_sequence(),
            self.followers[0].last_sequence(),
            self.followers[1].last_sequence(),
        ]
    }
}

The first part of the simulation runs the cases discussed above.

Listing 3.13 The cases of the chapter, run in order. replsim/src/main.rs

const BOTH: [Fate; 2] = [Fate::Delivered, Fate::Delivered];

fn cases(root: &Path) -> Result<(), Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let show = |cluster: &Cluster| {
        format!(
            "logs end at {:?}, committed {}",
            cluster.lasts(),
            cluster.leader.committed()
        )
    };

    for key in ["a", "b", "c"] {
        cluster.write(key, "1", BOTH)?;
    }

    println!("3 writes, every message delivered: {}", show(&cluster));

    for key in ["d", "e", "f"] {
        cluster.write(key, "1", [Fate::RequestLost, Fate::Delivered])?;
    }

    println!(
        "3 writes while follower 1 is unreachable: {}",
        show(&cluster)
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!("follower 1 returns, 1 exchange: {}", show(&cluster));

    for key in ["g", "h", "i"] {
        cluster.write(key, "1", [Fate::AnswerLost, Fate::Delivered])?;
    }

    println!(
        "3 writes while the answers of follower 1 are lost: {}",
        show(&cluster)
    );
    println!(
        "  the leader believes that the log of follower 1 ends at {}",
        cluster.reported[0]
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!(
        "  next exchange: the leader sends records 7 to 9, follower 1 refuses and reports {}",
        cluster.reported[0]
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!("  the exchange after it: {}", show(&cluster));

    let acknowledged =
        cluster.write("j", "1", [Fate::RequestLost, Fate::RequestLost])? == Some(true);
    println!(
        "1 write while both followers are unreachable: acknowledged {acknowledged}, {}",
        show(&cluster)
    );
    println!(
        "  the leader reads j as {:?}",
        cluster
            .leader
            .get("j")
            .map(|entry| entry.version)
    );

    let accepted = cluster
        .write("k", "1", [Fate::RequestLost, Fate::RequestLost])?
        .is_some();
    println!("  the leader accepts another write: {accepted}");

    cluster.exchange(1, Fate::Delivered)?;
    println!("follower 2 returns: {}", show(&cluster));
    println!(
        "  the leader reads j as {:?}",
        cluster
            .leader
            .get("j")
            .map(|entry| entry.version)
    );

    let mut newer = Replica::open(&root.join("newer"), false, 2)?;
    let answer = newer.handle_append(cluster.leader.append_for(0))?;
    println!(
        "a replica at term 2 receives a message of term 1: appended {}, its term {}",
        answer.appended, answer.term
    );

    Ok(())
}

The second part runs 3 writes under every combination of outcomes. Each write involves 2 exchanges, and each exchange is assigned 2 bits of a pattern, which gives 4096 patterns. After each pattern, replsim verifies invariants 1 and 2, then delivers all messages and verifies that the 3 logs become equal.

Listing 3.14 Every combination of lost messages over 3 writes. replsim/src/main.rs

const WRITES: usize = 3;

/// The fate of exchange number `index` under a pattern: 2 bits for each.
fn fate(pattern: u32, index: usize) -> Fate {
    match (pattern >> (2 * index)) & 3 {
        0 | 1 => Fate::Delivered,
        2 => Fate::RequestLost,
        3.. => Fate::AnswerLost,
    }
}

/// Runs 3 writes under one pattern of losses and checks the replicas. It
/// returns the number of writes that were acknowledged.
fn run(root: &Path, pattern: u32) -> Result<usize, Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let mut acknowledged = Vec::new();

    for number in 0..WRITES {
        let fates = [fate(pattern, 2 * number), fate(pattern, 2 * number + 1)];

        if cluster.write(&format!("key-{number}"), "1", fates)? == Some(true) {
            acknowledged.push(cluster.leader.last_sequence());
        }
    }

    for &sequence in &acknowledged {
        let holders = cluster
            .lasts()
            .iter()
            .filter(|&&last| last >= sequence)
            .count();

        if holders < 2 {
            return Err(format!(
                "pattern {pattern}: record {sequence} was acknowledged and is in {holders} log"
            )
            .into());
        }
    }

    for follower in &cluster.followers {
        if !cluster
            .leader
            .records()
            .starts_with(follower.records())
        {
            return Err(format!(
                "pattern {pattern}: a follower's log is not the beginning of the leader's"
            )
            .into());
        }
    }

    for _ in 0..2 {
        cluster.exchange(0, Fate::Delivered)?;
        cluster.exchange(1, Fate::Delivered)?;
    }

    let [leader, first, second] = cluster.lasts();

    if first != leader || second != leader || !cluster.leader.is_settled() {
        return Err(format!(
            "pattern {pattern}: the logs differ after every message was delivered"
        )
        .into());
    }

    Ok(acknowledged.len())
}

fn exhaustive(root: &Path) -> Result<(), Box<dyn Error>> {
    let patterns = 1u32 << (4 * WRITES);
    let mut by_acknowledged = [0; WRITES + 1];

    for pattern in 0..patterns {
        let directory = root.join(format!("pattern-{pattern}"));
        by_acknowledged[run(&directory, pattern)?] += 1;
        fs::remove_dir_all(&directory)?;
    }

    println!("{patterns} patterns of lost messages over {WRITES} writes:");
    println!("  every acknowledged record was in at least 2 of the 3 logs");
    println!("  every follower's log was the beginning of the leader's log");
    println!("  after the losses stopped, the 3 logs were equal and every record was committed");

    for (count, patterns) in by_acknowledged.iter().enumerate() {
        println!("  {count} of {WRITES} writes acknowledged: {patterns} patterns");
    }

    Ok(())
}

Listing 3.15 Running the simulation. replsim/src/main.rs

fn directory() -> PathBuf {
    env::temp_dir().join(format!("replsim-{}", process::id()))
}

fn main() -> Result<(), Box<dyn Error>> {
    let root = directory();
    let result = cases(&root.join("cases")).and_then(|()| exhaustive(&root));

    fs::remove_dir_all(&root)?;

    result
}

$ cargo run --release -p replsim
3 writes, every message delivered: logs end at [3, 3, 3], committed 3
3 writes while follower 1 is unreachable: logs end at [6, 3, 6], committed 6
follower 1 returns, 1 exchange: logs end at [6, 6, 6], committed 6
3 writes while the answers of follower 1 are lost: logs end at [9, 7, 9], committed 9
  the leader believes that the log of follower 1 ends at 6
  next exchange: the leader sends records 7 to 9, follower 1 refuses and reports 7
  the exchange after it: logs end at [9, 9, 9], committed 9
1 write while both followers are unreachable: acknowledged false, logs end at [10, 9, 9], committed 9
  the leader reads j as None
  the leader accepts another write: false
follower 2 returns: logs end at [10, 9, 10], committed 10
  the leader reads j as Some(10)
a replica at term 2 receives a message of term 1: appended false, its term 2
4096 patterns of lost messages over 3 writes:
  every acknowledged record was in at least 2 of the 3 logs
  every follower's log was the beginning of the leader's log
  after the losses stopped, the 3 logs were equal and every record was committed
  0 of 3 writes acknowledged: 1024 patterns
  1 of 3 writes acknowledged: 1024 patterns
  2 of 3 writes acknowledged: 704 patterns
  3 of 3 writes acknowledged: 1344 patterns

The seventh row shows something that every client of a replicated store has to take into account. A write that was answered with an error may nevertheless take effect later. The record was in the leader’s log, and it was committed when a follower became reachable again. For a client, therefore, an error from a leader signifies an unknown outcome, exactly as a lost connection does, and is to be resolved by reading the key or by repeating an idempotent request. An error does not mean that nothing happened.

Design overview

The node

kvnode combines a replica with a gRPC service for replication and the HTTP API of chapter 1.

Listing 3.16 The imports of kvnode/src/main.rs. kvnode/src/main.rs

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, Query, 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, ValueEnum};
use kv::{Change, Condition};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use repl::Replica;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use tonic::transport::Server;

use crate::node::Node;
use crate::rpc::{Follower, ReplicationServer};

Listing 3.17 The options of the command line. kvnode/src/main.rs

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

#[derive(Parser)]
struct Options {
    #[arg(long, value_enum)]
    role: Role,

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

    /// The address on which a follower accepts records from the leader.
    #[arg(long, default_value = "127.0.0.1:7101")]
    replication: SocketAddr,

    /// The replication address of a follower, such as http://127.0.0.1:7102.
    /// Given once for each follower, to the leader.
    #[arg(long = "follower")]
    followers: Vec<String>,

    /// The term this node starts in. A new leader is given a higher term
    /// than the leader it replaces.
    #[arg(long, default_value_t = 1)]
    term: u64,

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

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

The gRPC service

The Rust types, the client code and the server code for Append are generated from the schema when the program is built. Building requires protoc, the compiler for Protocol Buffers, which is available as the package protobuf-compiler on Debian and Ubuntu.

Listing 3.18 The complete kvnode/build.rs. kvnode/build.rs

//! Generates the Rust types and the gRPC client and server from the schema.
//! It needs the Protocol Buffers compiler, `protoc`, on the PATH.

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_prost_build::compile_protos("../proto/store.proto")?;

    Ok(())
}

Listing 3.19 The imports of kvnode/src/rpc.rs. kvnode/src/rpc.rs

use std::sync::Arc;

use kv::{Change, Record};
use repl::{Append, Appended};
use tokio::task;
use tonic::{Request, Response, Status};

use crate::node::Node;

Listing 3.20 Including the generated code. kvnode/src/rpc.rs

/// 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::replication_client::ReplicationClient;
pub use proto::replication_server::{Replication, ReplicationServer};

The generated types are converted to the types of kv and repl at the boundary of the service. A Record message that has no change field is rejected with the gRPC status InvalidArgument.

Listing 3.21 Conversion between the generated types and the library's types. kvnode/src/rpc.rs

impl From<&Record> for proto::Record {
    fn from(record: &Record) -> Self {
        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),
        }
    }
}

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

    fn try_from(record: proto::Record) -> Result<Self, 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<Append> for proto::AppendRequest {
    fn from(message: Append) -> Self {
        proto::AppendRequest {
            shard: 0,
            term: message.term,
            previous_sequence: message.previous_sequence,
            records: message
                .records
                .iter()
                .map(proto::Record::from)
                .collect(),
            committed_sequence: message.committed_sequence,
            // The fields that leader election added are not used by this program.
            previous_term: 0,
            leader: 0,
        }
    }
}

impl From<proto::AppendResponse> for Appended {
    fn from(answer: proto::AppendResponse) -> Self {
        Appended {
            term: answer.term,
            appended: answer.appended,
            last_sequence: answer.last_sequence,
        }
    }
}

Followers run the server side of the service. The handler passes each message to Replica::handle_append. Since that method waits for the disk, the call is wrapped in block_in_place, which informs the asynchronous runtime that the current thread is about to block.

Listing 3.22 The server side of Append. kvnode/src/rpc.rs

/// The server side of `Append`, which a follower runs.
pub struct Follower {
    pub node: Arc<Node>,
}

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

        let message = Append {
            term: request.term,
            previous_sequence: request.previous_sequence,
            records,
            committed_sequence: request.committed_sequence,
        };

        // Appending waits for the disk, so it runs where blocking is allowed.
        let answer = task::block_in_place(|| {
            self.node
                .replica
                .lock()
                .handle_append(message)
        })
        .map_err(|error| Status::unavailable(error.to_string()))?;

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

The leader’s tasks

Figure 3.7 shows the parts of a leader and the order in which a write passes through them. Each box corresponds to a function or a field in the listings that follow.

Listing 3.23 The imports of kvnode/src/node.rs. kvnode/src/node.rs

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use kv::{Change, Condition, Entry, Summary, Written};
use kvd::api::ApiError;
use parking_lot::Mutex;
use repl::{Appended, Replica, majority_sequence};
use tokio::sync::{Notify, watch};
use tokio::task;
use tokio::time::{sleep, timeout};
use tonic::transport::{Channel, Endpoint};

use crate::rpc::{ReplicationClient, proto};

Listing 3.24 The state of a node. kvnode/src/node.rs

/// How often a leader sends a message to a follower that has nothing new to receive.
const HEARTBEAT: Duration = Duration::from_millis(100);

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

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

/// What the leader knows about one follower.
pub struct Peer {
    pub address: String,
    /// The last sequence the follower has reported.
    reported: AtomicU64,
    /// When the follower last answered.
    heard: Mutex<Option<Instant>>,
}

pub struct Node {
    pub replica: Mutex<Replica>,
    pub leader: bool,
    pub peers: Vec<Peer>,
    /// The sequence the log ended at when the node started. A leader serves
    /// reads once this much is committed.
    ready: u64,
    /// Held by the write that is in progress, so that writes take turns.
    turn: tokio::sync::Mutex<()>,
    /// The committed sequence, for writes that wait for it to advance.
    committed: watch::Sender<u64>,
    /// Signalled when the leader has appended a record.
    appended: Notify,
    /// Set when a follower answers with a newer term: another leader exists.
    deposed: AtomicBool,
}

impl Node {
    pub fn new(replica: Replica, leader: bool, followers: Vec<String>) -> Arc<Node> {
        let peer = |address| Peer {
            address,
            reported: AtomicU64::new(0),
            heard: Mutex::new(None),
        };

        Arc::new(Node {
            ready: replica.last_sequence(),
            replica: Mutex::new(replica),
            leader,
            peers: followers
                .into_iter()
                .map(peer)
                .collect(),
            turn: tokio::sync::Mutex::new(()),
            committed: watch::Sender::new(0),
            appended: Notify::new(),
            deposed: AtomicBool::new(false),
        })
    }
}

A leader starts 1 task for each follower. The task repeats a single exchange: it builds a message that begins where the follower’s log is believed to end, sends it, and processes the answer. When the follower is up to date, the task waits until the leader appends a record or until the heartbeat interval has elapsed. When the answer shows that records are still missing, the task continues at once.

follow runs for the lifetime of the node, and its loop has 3 phases. It first locks the replica for no longer than it takes to build a message. It then releases the lock and waits for the answer, for at most 500 ms. Finally it passes the answer to record. If record reports that the follower is still behind, the loop continues immediately. Otherwise the task sleeps until appended is signalled or the heartbeat interval elapses, whichever occurs first.

Listing 3.25 The task that keeps a follower up to date. kvnode/src/node.rs

impl Node {
    /// Starts one task for each follower. A task sends the follower what it
    /// lacks, again and again, for as long as the node runs.
    pub fn lead(self: &Arc<Node>) -> Result<(), tonic::transport::Error> {
        for index in 0..self.peers.len() {
            let channel = Endpoint::from_shared(self.peers[index].address.clone())?
                .connect_timeout(ANSWER)
                .connect_lazy();

            tokio::spawn(Arc::clone(self).follow(index, ReplicationClient::new(channel)));
        }

        Ok(())
    }

    async fn follow(self: Arc<Node>, index: usize, mut client: ReplicationClient<Channel>) {
        let peer = &self.peers[index];

        loop {
            let message = self.replica.lock().append_for(
                peer.reported
                    .load(Ordering::Relaxed),
            );
            let request = proto::AppendRequest::from(message);

            let answer = timeout(ANSWER, client.append(request)).await;

            // An answer that did not come counts as a peer that is up to date:
            // the next round is a heartbeat, which finds out more.
            let behind = answer
                .ok()
                .and_then(Result::ok)
                .is_some_and(|answer| self.record(peer, answer.into_inner().into()));

            if !behind {
                // Wait for a new record, or send a heartbeat when none comes.
                tokio::select! {
                    () = self.appended.notified() => {}
                    () = sleep(HEARTBEAT) => {}
                }
            }
        }
    }
}

A point of design here has a large effect on how the node behaves under faults. No task holds the lock of the replica while it waits for the network. It takes the lock to build a message and again to process an answer, and releases it in between. An unreachable follower therefore delays its own task and nothing else, neither the task of the other follower nor any request from a client. Had the lock been held across the network call, a single slow follower would have stopped the whole shard, which would have defeated the purpose of waiting only for a majority.

record is called with every answer. It first compares the term of the answer with the leader’s own. A higher term shows that a newer leader exists, and the node ceases to accept writes. Otherwise it stores the sequence reported, applies the commit rule to the 3 last sequences, and publishes the new committed sequence through a watch channel, where the waiting writes observe it.

Listing 3.26 Processing an answer, and committing. kvnode/src/node.rs

impl Node {
    /// Takes note of a follower's answer and commits what a majority now
    /// holds. It returns whether the follower still lacks records.
    fn record(&self, peer: &Peer, answer: Appended) -> bool {
        let mut replica = self.replica.lock();

        if answer.term > replica.term() {
            self.deposed
                .store(true, Ordering::Relaxed);

            return false;
        }

        peer.reported
            .store(answer.last_sequence, Ordering::Relaxed);
        *peer.heard.lock() = Some(Instant::now());

        let mut stored: Vec<u64> = self
            .peers
            .iter()
            .map(|peer| {
                peer.reported
                    .load(Ordering::Relaxed)
            })
            .collect();
        stored.push(replica.last_sequence());

        replica.commit(majority_sequence(stored));
        self.committed
            .send_replace(replica.committed());

        answer.last_sequence < replica.last_sequence()
    }
}

Writes

Listing 3.27 The write path of the leader. kvnode/src/node.rs

impl Node {
    /// A write on the leader: wait for a turn, append the record, and wait
    /// until a majority of the replicas hold it.
    pub async fn write(&self, change: Change, condition: Condition) -> Result<Written, ApiError> {
        self.must_lead()?;

        let turn = self.turn.lock().await;
        let last = self.replica.lock().last_sequence();
        self.wait_for(last).await?;

        let proposed = task::block_in_place(|| {
            let mut replica = self.replica.lock();
            let created = match &change {
                Change::Put { key, .. } | Change::Delete { key } => replica.get(key).is_none(),
            };

            replica
                .propose(change, condition)
                .map(|version| Written { version, created })
        })?;

        self.appended.notify_waiters();
        self.wait_for(proposed.version)
            .await?;
        drop(turn);

        Ok(proposed)
    }

    /// Waits until the record with this sequence is committed.
    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 replicas"
            )))
        }
    }

    fn must_lead(&self) -> Result<(), ApiError> {
        if !self.leader {
            return Err(ApiError::Misdirected("this node is a follower".into()));
        }

        if self.deposed.load(Ordering::Relaxed) {
            return Err(ApiError::Misdirected(
                "a leader of a newer term exists".into(),
            ));
        }

        Ok(())
    }
}

The handlers send no messages themselves. A handler appends a record, wakes both follower tasks, and waits for the committed sequence to reach the sequence of its record. A watch channel holds the most recent committed sequence, and wait_for returns as soon as that value satisfies its condition, including the case in which it already does.

If no majority stores the record within 1 second, the handler answers 503. The record remains in the leader’s log. The follower tasks go on sending it, and it is committed when a follower becomes reachable, as the simulation showed.

Reads

Listing 3.28 The read path of the leader. kvnode/src/node.rs

impl Node {
    /// A leader serves reads while a majority of the replicas, itself
    /// included, has answered within the last second, and once everything
    /// its log held at start is committed.
    fn must_serve_reads(&self) -> Result<(), ApiError> {
        self.must_lead()?;

        let recently = |peer: &&Peer| {
            peer.heard
                .lock()
                .is_some_and(|at| at.elapsed() < COMMIT)
        };
        let reachable = 1 + self
            .peers
            .iter()
            .filter(recently)
            .count();

        if 2 * reachable <= self.peers.len() + 1 {
            return Err(ApiError::Unavailable(
                "this leader cannot reach a majority of the replicas".into(),
            ));
        }

        if self.replica.lock().committed() < self.ready {
            return Err(ApiError::Unavailable(
                "the records from before the restart are not committed yet".into(),
            ));
        }

        Ok(())
    }

    pub fn get(&self, key: &str) -> Result<Entry, ApiError> {
        self.must_serve_reads()?;

        self.replica
            .lock()
            .get(key)
            .cloned()
            .ok_or(ApiError::NotFound)
    }

    pub fn list(
        &self,
        prefix: &str,
        after: Option<&str>,
        limit: usize,
    ) -> Result<(Vec<Summary>, bool), ApiError> {
        self.must_serve_reads()?;

        Ok(self
            .replica
            .lock()
            .list(prefix, after, limit))
    }

    /// The last sequence each follower has reported.
    pub fn reported(&self) -> Vec<u64> {
        self.peers
            .iter()
            .map(|peer| {
                peer.reported
                    .load(Ordering::Relaxed)
            })
            .collect()
    }
}

A leader that has been separated from its followers by a network partition cannot know whether another leader has been appointed in the meantime, and its entries may therefore be out of date. must_serve_reads accordingly requires that a majority of the replicas, counting the leader itself, has answered within the last second. The check limits to 1 second the time for which a former leader could return outdated data, on the condition that no new leader is appointed sooner than that after the old one becomes unreachable. Stricter implementations confirm their leadership with a majority for every read, at the cost of a network round trip for each.

A second check concerns a leader that has been restarted. Its log may contain committed records that have not yet been applied, since committed begins at 0. Such a leader serves reads once every record that its log held at startup has been committed again.

The HTTP API

The handlers reuse the error type and the header parsing of chapter 2, from kvd::api. They differ from the handlers of chapter 2 in passing writes to Node::write and reads to Node::get.

Listing 3.29 The handlers for PUT, DELETE and GET. kvnode/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())
}

Listing 3.30 Listing keys, and the health of the node. kvnode/src/main.rs

#[derive(Deserialize)]
struct ListQuery {
    #[serde(default)]
    prefix: String,
    after: Option<String>,
    limit: Option<usize>,
}

#[derive(Serialize)]
struct Item {
    key: String,
    version: u64,
    size: usize,
}

#[derive(Serialize)]
struct Page {
    keys: Vec<Item>,
    next_after: Option<String>,
}

async fn list_keys(
    State(node): State<Arc<Node>>,
    Query(query): Query<ListQuery>,
) -> Result<Json<Page>, ApiError> {
    let limit = query
        .limit
        .unwrap_or(100)
        .clamp(1, 1000);
    let (page, more) = node.list(&query.prefix, query.after.as_deref(), limit)?;
    let next_after = if more {
        page.last()
            .map(|last| last.key.clone())
    } else {
        None
    };
    let keys = page
        .into_iter()
        .map(|item| Item {
            key: item.key,
            version: item.version,
            size: item.size,
        })
        .collect();

    Ok(Json(Page { keys, next_after }))
}

#[derive(Serialize)]
struct Health {
    role: &'static str,
    term: u64,
    last_sequence: u64,
    committed: u64,
    /// The last sequence each follower has reported to this leader.
    followers: Vec<u64>,
}

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

    Json(Health {
        role: if node.leader { "leader" } else { "follower" },
        term: replica.term(),
        last_sequence: replica.last_sequence(),
        committed: replica.committed(),
        followers: node.reported(),
    })
}

Listing 3.31 Starting a leader or a follower. kvnode/src/main.rs

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let replica = Replica::open(&options.data, !options.no_sync, options.term)?;

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

    let node = Node::new(replica, options.role == Role::Leader, options.followers);

    match options.role {
        Role::Leader => node.lead()?,
        Role::Follower => {
            let service = ReplicationServer::new(Follower {
                node: Arc::clone(&node),
            });

            tokio::spawn(
                Server::builder()
                    .add_service(service)
                    .serve(options.replication),
            );
            println!("accepting records on {}", options.replication);
        }
    }

    let routes = Router::new()
        .route("/v1/keys", get(list_keys))
        .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

The followers are started first, each with an address for HTTP and another for replication. The leader is then started with the replication addresses of both followers.

$ ./target/release/kvnode --role follower --listen 127.0.0.1:7002 --replication 127.0.0.1:7102 --data /var/tmp/n2 &
term 1, log ends at 0
accepting records on 127.0.0.1:7102
listening on 127.0.0.1:7002
$ ./target/release/kvnode --role follower --listen 127.0.0.1:7003 --replication 127.0.0.1:7103 --data /var/tmp/n3 &
$ ./target/release/kvnode --role leader --listen 127.0.0.1:7001 --data /var/tmp/n1 \
      --follower http://127.0.0.1:7102 --follower http://127.0.0.1:7103 &
term 1, log ends at 0
listening on 127.0.0.1:7001

A write to the leader is stored by all 3 nodes, and a request to a follower is refused with the status that chapter 1 specifies for a request sent to the wrong node.

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

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

{"type":"/problems/not-leader","title":"This node is not the leader","status":421,"detail":"this node is a follower"}
$ for port in 7001 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"role":"leader","term":1,"last_sequence":1,"committed":1,"followers":[1,1]}
{"role":"follower","term":1,"last_sequence":1,"committed":1,"followers":[]}
{"role":"follower","term":1,"last_sequence":1,"committed":1,"followers":[]}
$ cmp /var/tmp/n1/kv.log /var/tmp/n2/kv.log && cmp /var/tmp/n1/kv.log /var/tmp/n3/kv.log && echo identical
identical

The health response of the leader lists the last sequence reported by each follower. The 3 log files are identical byte for byte, which is invariant 1 in the strongest form it could take.

Experiments

All the nodes in these experiments run on 1 machine and share its disk. On separate machines, each node has a disk of its own and the nodes do not compete for it. Nothing else changes: each node is started with the addresses of its peers in place of 127.0.0.1.

The cost of replication

$ time ./target/release/kvctl load --server http://127.0.0.1:7001 --count 300 --acked acked.txt
300 writes acknowledged
3.89 s, 77 writes per second

The replicated shard completes 77 writes per second, against 185 for the single node of chapter 2 on the same disk. Each write now waits for 2 disks in succession, the leader’s and then a follower’s, and for a network round trip. Chapter 1 allowed for this in its capacity estimate by counting 3 stored records for every write.

A follower stops

A follower is killed while a load is running.

$ ./target/release/kvctl load --server http://127.0.0.1:7001 --count 400 --acked acked.txt &
$ kill -9 <pid of the node on port 7002>
$ curl http://127.0.0.1:7001/healthz
{"role":"leader","term":1,"last_sequence":435,"committed":434,"followers":[417,434]}
400 writes acknowledged
$ curl http://127.0.0.1:7001/healthz
{"role":"leader","term":1,"last_sequence":700,"committed":700,"followers":[417,700]}

$ ./target/release/kvnode --role follower --listen 127.0.0.1:7002 --replication 127.0.0.1:7102 --data /var/tmp/n2 &
term 1, log ends at 417
$ curl http://127.0.0.1:7001/healthz
{"role":"leader","term":1,"last_sequence":700,"committed":700,"followers":[700,700]}
$ cmp /var/tmp/n1/kv.log /var/tmp/n2/kv.log && echo identical
identical

Apart from restarting the process, no action by an operator was required.

The majority is lost

Both followers are killed.

$ kill -9 <pid of the node on port 7002> <pid of the node on port 7003>
$ curl -o /dev/null -w '%{http_code} in %{time_total} s\n' -X PUT http://127.0.0.1:7001/v1/keys/alone --data-binary 'x'
503 in 1.011346 s
$ curl -X PUT http://127.0.0.1:7001/v1/keys/alone2 --data-binary 'x'
{"type":"/problems/store-unavailable","title":"The node cannot serve the request","status":503,"detail":"record 701 is not on a majority of the replicas"}
$ curl http://127.0.0.1:7001/v1/keys/key-000001
{"type":"/problems/store-unavailable","title":"The node cannot serve the request","status":503,"detail":"this leader cannot reach a majority of the replicas"}
$ curl http://127.0.0.1:7001/healthz
{"role":"leader","term":1,"last_sequence":701,"committed":700,"followers":[700,700]}

$ ./target/release/kvnode --role follower --listen 127.0.0.1:7003 --replication 127.0.0.1:7103 --data /var/tmp/n3 &
$ curl http://127.0.0.1:7001/healthz
{"role":"leader","term":1,"last_sequence":701,"committed":701,"followers":[700,701]}
$ curl -o /dev/null -w '%{http_code}\n' http://127.0.0.1:7001/v1/keys/alone
200
$ curl -o /dev/null -w '%{http_code}\n' -X PUT http://127.0.0.1:7001/v1/keys/after --data-binary 'x'
201

This is exactly the behaviour that chapter 1 specified for the minority side of a partition. The leader remains consistent and becomes unavailable, and it would be easy, and wrong, to regard its refusal to serve as a defect. A leader that went on answering in this condition would be answering from data that it cannot know to be current.

The leader is lost

In this experiment a leader is killed under load and its data directory is moved away, which simulates the loss of a machine together with its disk.

$ ./target/release/kvctl load --server http://127.0.0.1:7001 --count 2000 --acked acked.txt &
$ kill -9 <pid of the leader>
write 229 failed: io: Peer disconnected
228 writes acknowledged
$ mv /var/tmp/n1 /var/tmp/n1.lost
$ for port in 7002 7003; do curl http://127.0.0.1:$port/healthz; echo; done
{"role":"follower","term":1,"last_sequence":228,"committed":227,"followers":[]}
{"role":"follower","term":1,"last_sequence":228,"committed":227,"followers":[]}

$ kill -9 <pid of the node on port 7003>
$ ./target/release/kvnode --role follower --listen 127.0.0.1:7001 --replication 127.0.0.1:7101 --data /var/tmp/n1 &
$ ./target/release/kvnode --role leader --term 2 --listen 127.0.0.1:7003 --data /var/tmp/n3 \
      --follower http://127.0.0.1:7102 --follower http://127.0.0.1:7101 &
term 2, log ends at 228
listening on 127.0.0.1:7003
$ for port in 7003 7002 7001; do curl http://127.0.0.1:$port/healthz; echo; done
{"role":"leader","term":2,"last_sequence":228,"committed":228,"followers":[228,228]}
{"role":"follower","term":2,"last_sequence":228,"committed":228,"followers":[]}
{"role":"follower","term":2,"last_sequence":228,"committed":228,"followers":[]}
$ ./target/release/kvctl verify --server http://127.0.0.1:7003 --acked acked.txt
228 acknowledged, 228 found, 0 missing

The second row illustrates why the committed sequence is not stored on disk. What a follower knows to be committed always lags behind its leader, and after a leader has been lost, the committed sequence is established afresh from the logs: a new leader commits a record as soon as a majority has reported it.

For the promotion, the follower with the longer log must be chosen. By invariant 2, every acknowledged record is on at least 1 of the 2 followers, though not necessarily on both. If their logs differ in length, the longer contains every acknowledged record, and the shorter may lack some of the most recent. To promote the shorter would be to discard acknowledged writes, and this choice, made here by a person reading 2 numbers, is the one that chapter 4 has to teach the nodes to make for themselves.

The former leader returns

Suppose that a former leader was not lost after all and begins to send messages again, as it might when a network partition ends. A leader with term 1 is started against the 2 followers.

$ ./target/release/kvnode --role leader --term 1 --listen 127.0.0.1:7009 --data /var/tmp/n9 \
      --follower http://127.0.0.1:7102 --follower http://127.0.0.1:7101 &
$ curl -i -X PUT http://127.0.0.1:7009/v1/keys/stale --data-binary 'x'
HTTP/1.1 421 Misdirected Request
content-type: application/problem+json
content-length: 125

{"type":"/problems/not-leader","title":"This node is not the leader","status":421,"detail":"a leader of a newer term exists"}
$ curl http://127.0.0.1:7002/healthz
{"role":"follower","term":2,"last_sequence":228,"committed":228,"followers":[]}
$ curl -o /dev/null -w '%{http_code}\n' -X PUT http://127.0.0.1:7003/v1/keys/fresh --data-binary 'x'
201

Both followers are at term 2 and refuse messages of term 1. Each of their answers carries the higher term, so that the old leader records that it has been replaced and rejects writes with 421. The shard meanwhile continues to accept writes through its leader of term 2. Terms are what make a change of leader safe, since a leader that has been replaced cannot add a record to any follower.

Limitations

The first 2 limitations are removed in chapter 4, and the third in chapter 6.

The complete programs

The log and the store are listed in chapter 2. Those listings include Log::append_records as well as apply, check and page, which this chapter uses.

Listing 3.32 The complete repl/src/lib.rs. repl/src/lib.rs

//! Replication of a log from a leader to its followers.
//!
//! A `Replica` is one copy of a shard: a log, and the entries that result
//! from the records that are known to be on a majority of the copies. The
//! same type serves as leader and as follower. This module does no I/O on
//! the network. It builds the messages and processes them, and the caller
//! carries them between replicas.

use std::collections::BTreeMap;
use std::io;
use std::path::Path;

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

/// The message a leader sends to a follower.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Append {
    /// The term of the sender.
    pub term: u64,
    /// The sequence of the record that precedes `records` in the leader's log.
    pub previous_sequence: u64,
    pub records: Vec<Record>,
    /// Every record up to this sequence is on a majority of the replicas.
    pub committed_sequence: u64,
}

/// The follower's answer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Appended {
    pub term: u64,
    /// False when the follower's log does not end at `previous_sequence`,
    /// or when the sender's term is older than the follower's.
    pub appended: bool,
    /// The sequence of the last record in the follower's log.
    pub last_sequence: u64,
}

pub struct Replica {
    log: Log,
    /// Every record of the log, in order. Record `n` is at index `n - 1`.
    records: Vec<Record>,
    /// The result of applying the records up to `committed`.
    entries: BTreeMap<String, Entry>,
    /// The sequence of the last record that is applied to `entries`.
    committed: u64,
    term: u64,
}

impl Replica {
    /// Opens the replica in `directory`. Nothing is applied yet: a replica
    /// that has restarted does not know which of its records are committed.
    pub fn open(directory: &Path, sync: bool, term: u64) -> io::Result<Replica> {
        let (log, recovery) = Log::open(directory, sync)?;

        Ok(Replica {
            log,
            records: recovery.records,
            entries: BTreeMap::new(),
            committed: 0,
            term,
        })
    }

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

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

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

    /// One page of the committed keys.
    pub fn list(&self, prefix: &str, after: Option<&str>, limit: usize) -> (Vec<Summary>, bool) {
        page(&self.entries, prefix, after, limit)
    }

    /// Every record of the log, in order.
    pub fn records(&self) -> &[Record] {
        &self.records
    }

    /// The committed value of a key.
    pub fn get(&self, key: &str) -> Option<&Entry> {
        self.entries.get(key)
    }

    /// Applies the records up to `sequence` that are not applied yet.
    pub fn commit(&mut self, sequence: u64) {
        let until = sequence.min(self.last_sequence());

        while self.committed < until {
            apply(
                &mut self.entries,
                self.records[self.committed as usize].clone(),
            );
            self.committed += 1;
        }
    }
}

impl Replica {
    /// Whether every record of the log is committed. A leader accepts a new
    /// write only in this state, so that the condition of the write is
    /// checked against the result of all earlier writes.
    pub fn is_settled(&self) -> bool {
        self.committed == self.last_sequence()
    }

    /// Checks the condition of a write and appends its record to the
    /// leader's log. The record is not applied: it is not committed yet.
    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);

        Ok(sequence)
    }
}

impl Replica {
    /// The message that brings a follower whose log ends at `follower_last`
    /// up to the end of this log. With nothing to send it is a heartbeat.
    pub fn append_for(&self, follower_last: u64) -> Append {
        let from = follower_last.min(self.last_sequence()) as usize;

        Append {
            term: self.term,
            previous_sequence: from as u64,
            records: self.records[from..].to_vec(),
            committed_sequence: self.committed,
        }
    }
}


/// The highest sequence that is stored on a majority, given the last
/// sequence of every replica, the leader included.
pub fn majority_sequence(mut last_sequences: Vec<u64>) -> u64 {
    last_sequences.sort_unstable_by(|a, b| b.cmp(a));

    last_sequences
        .get(last_sequences.len() / 2)
        .copied()
        .unwrap_or(0)
}

impl Replica {
    /// Processes a message from a leader.
    pub fn handle_append(&mut self, request: Append) -> io::Result<Appended> {
        let refuse = |replica: &Replica| Appended {
            term: replica.term,
            appended: false,
            last_sequence: replica.last_sequence(),
        };

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

        self.term = request.term;

        if request.previous_sequence != self.last_sequence() {
            return Ok(refuse(self));
        }

        if !request.records.is_empty() {
            self.log
                .append_records(&request.records)?;
            self.records.extend(request.records);
        }

        self.commit(request.committed_sequence);

        Ok(Appended {
            term: self.term,
            appended: true,
            last_sequence: self.last_sequence(),
        })
    }
}

Listing 3.33 The complete replsim/src/main.rs. replsim/src/main.rs

//! Command replsim runs a leader and 2 followers in one process and carries
//! the messages between them by hand, so that it can lose any message it
//! chooses. It first walks through the cases of the chapter, and then tries
//! every combination of lost requests and lost answers over 3 writes and
//! checks what the replicas hold after each.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p replsim

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

use kv::{Change, Condition};
use repl::{Replica, majority_sequence};

/// What happens to one exchange between the leader and a follower.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Fate {
    Delivered,
    /// The follower does not receive the message.
    RequestLost,
    /// The follower processes the message, and its answer does not arrive.
    AnswerLost,
}

/// A leader, 2 followers, and what the leader knows about each follower.
struct Cluster {
    leader: Replica,
    followers: Vec<Replica>,
    /// The last sequence each follower has reported.
    reported: Vec<u64>,
}

impl Cluster {
    fn open(root: &Path) -> Result<Cluster, Box<dyn Error>> {
        let open = |name: &str| Replica::open(&root.join(name), false, 1);

        Ok(Cluster {
            leader: open("leader")?,
            followers: vec![open("follower-1")?, open("follower-2")?],
            reported: vec![0, 0],
        })
    }

    /// One exchange with a follower: the leader sends what the follower
    /// lacks, as far as it knows, and commits what a majority now holds.
    fn exchange(&mut self, follower: usize, fate: Fate) -> Result<(), Box<dyn Error>> {
        let message = self
            .leader
            .append_for(self.reported[follower]);

        if fate == Fate::RequestLost {
            return Ok(());
        }

        let answer = self.followers[follower].handle_append(message)?;

        if fate == Fate::AnswerLost {
            return Ok(());
        }

        self.reported[follower] = answer.last_sequence;

        let stored = vec![
            self.leader.last_sequence(),
            self.reported[0],
            self.reported[1],
        ];
        self.leader
            .commit(majority_sequence(stored));

        Ok(())
    }

    /// A write: the leader appends the record and makes one exchange with
    /// each follower. It returns whether the write was acknowledged, or
    /// `None` when the leader could not accept it.
    fn write(
        &mut self,
        key: &str,
        value: &str,
        fates: [Fate; 2],
    ) -> Result<Option<bool>, Box<dyn Error>> {
        if !self.leader.is_settled() {
            return Ok(None);
        }

        let change = Change::Put {
            key: key.into(),
            value: value.into(),
        };
        let sequence = self
            .leader
            .propose(change, Condition::Always)?;

        self.exchange(0, fates[0])?;
        self.exchange(1, fates[1])?;

        Ok(Some(self.leader.committed() >= sequence))
    }

    /// The last sequence of the leader and of each follower.
    fn lasts(&self) -> [u64; 3] {
        [
            self.leader.last_sequence(),
            self.followers[0].last_sequence(),
            self.followers[1].last_sequence(),
        ]
    }
}

const BOTH: [Fate; 2] = [Fate::Delivered, Fate::Delivered];

fn cases(root: &Path) -> Result<(), Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let show = |cluster: &Cluster| {
        format!(
            "logs end at {:?}, committed {}",
            cluster.lasts(),
            cluster.leader.committed()
        )
    };

    for key in ["a", "b", "c"] {
        cluster.write(key, "1", BOTH)?;
    }

    println!("3 writes, every message delivered: {}", show(&cluster));

    for key in ["d", "e", "f"] {
        cluster.write(key, "1", [Fate::RequestLost, Fate::Delivered])?;
    }

    println!(
        "3 writes while follower 1 is unreachable: {}",
        show(&cluster)
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!("follower 1 returns, 1 exchange: {}", show(&cluster));

    for key in ["g", "h", "i"] {
        cluster.write(key, "1", [Fate::AnswerLost, Fate::Delivered])?;
    }

    println!(
        "3 writes while the answers of follower 1 are lost: {}",
        show(&cluster)
    );
    println!(
        "  the leader believes that the log of follower 1 ends at {}",
        cluster.reported[0]
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!(
        "  next exchange: the leader sends records 7 to 9, follower 1 refuses and reports {}",
        cluster.reported[0]
    );

    cluster.exchange(0, Fate::Delivered)?;
    println!("  the exchange after it: {}", show(&cluster));

    let acknowledged =
        cluster.write("j", "1", [Fate::RequestLost, Fate::RequestLost])? == Some(true);
    println!(
        "1 write while both followers are unreachable: acknowledged {acknowledged}, {}",
        show(&cluster)
    );
    println!(
        "  the leader reads j as {:?}",
        cluster
            .leader
            .get("j")
            .map(|entry| entry.version)
    );

    let accepted = cluster
        .write("k", "1", [Fate::RequestLost, Fate::RequestLost])?
        .is_some();
    println!("  the leader accepts another write: {accepted}");

    cluster.exchange(1, Fate::Delivered)?;
    println!("follower 2 returns: {}", show(&cluster));
    println!(
        "  the leader reads j as {:?}",
        cluster
            .leader
            .get("j")
            .map(|entry| entry.version)
    );

    let mut newer = Replica::open(&root.join("newer"), false, 2)?;
    let answer = newer.handle_append(cluster.leader.append_for(0))?;
    println!(
        "a replica at term 2 receives a message of term 1: appended {}, its term {}",
        answer.appended, answer.term
    );

    Ok(())
}

const WRITES: usize = 3;

/// The fate of exchange number `index` under a pattern: 2 bits for each.
fn fate(pattern: u32, index: usize) -> Fate {
    match (pattern >> (2 * index)) & 3 {
        0 | 1 => Fate::Delivered,
        2 => Fate::RequestLost,
        3.. => Fate::AnswerLost,
    }
}

/// Runs 3 writes under one pattern of losses and checks the replicas. It
/// returns the number of writes that were acknowledged.
fn run(root: &Path, pattern: u32) -> Result<usize, Box<dyn Error>> {
    let mut cluster = Cluster::open(root)?;
    let mut acknowledged = Vec::new();

    for number in 0..WRITES {
        let fates = [fate(pattern, 2 * number), fate(pattern, 2 * number + 1)];

        if cluster.write(&format!("key-{number}"), "1", fates)? == Some(true) {
            acknowledged.push(cluster.leader.last_sequence());
        }
    }

    for &sequence in &acknowledged {
        let holders = cluster
            .lasts()
            .iter()
            .filter(|&&last| last >= sequence)
            .count();

        if holders < 2 {
            return Err(format!(
                "pattern {pattern}: record {sequence} was acknowledged and is in {holders} log"
            )
            .into());
        }
    }

    for follower in &cluster.followers {
        if !cluster
            .leader
            .records()
            .starts_with(follower.records())
        {
            return Err(format!(
                "pattern {pattern}: a follower's log is not the beginning of the leader's"
            )
            .into());
        }
    }

    for _ in 0..2 {
        cluster.exchange(0, Fate::Delivered)?;
        cluster.exchange(1, Fate::Delivered)?;
    }

    let [leader, first, second] = cluster.lasts();

    if first != leader || second != leader || !cluster.leader.is_settled() {
        return Err(format!(
            "pattern {pattern}: the logs differ after every message was delivered"
        )
        .into());
    }

    Ok(acknowledged.len())
}

fn exhaustive(root: &Path) -> Result<(), Box<dyn Error>> {
    let patterns = 1u32 << (4 * WRITES);
    let mut by_acknowledged = [0; WRITES + 1];

    for pattern in 0..patterns {
        let directory = root.join(format!("pattern-{pattern}"));
        by_acknowledged[run(&directory, pattern)?] += 1;
        fs::remove_dir_all(&directory)?;
    }

    println!("{patterns} patterns of lost messages over {WRITES} writes:");
    println!("  every acknowledged record was in at least 2 of the 3 logs");
    println!("  every follower's log was the beginning of the leader's log");
    println!("  after the losses stopped, the 3 logs were equal and every record was committed");

    for (count, patterns) in by_acknowledged.iter().enumerate() {
        println!("  {count} of {WRITES} writes acknowledged: {patterns} patterns");
    }

    Ok(())
}

fn directory() -> PathBuf {
    env::temp_dir().join(format!("replsim-{}", process::id()))
}

fn main() -> Result<(), Box<dyn Error>> {
    let root = directory();
    let result = cases(&root.join("cases")).and_then(|()| exhaustive(&root));

    fs::remove_dir_all(&root)?;

    result
}

Listing 3.34 The complete kvnode/Cargo.toml. kvnode/Cargo.toml

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

[dependencies]
kv.workspace = true
kvd.workspace = true
repl.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 3.35 The complete kvnode/src/rpc.rs. kvnode/src/rpc.rs

//! The replication service: the generated gRPC types, the conversion between
//! them and the types of the `repl` crate, and the server side that a
//! follower runs.

use std::sync::Arc;

use kv::{Change, Record};
use repl::{Append, Appended};
use tokio::task;
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::replication_client::ReplicationClient;
pub use proto::replication_server::{Replication, ReplicationServer};

impl From<&Record> for proto::Record {
    fn from(record: &Record) -> Self {
        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),
        }
    }
}

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

    fn try_from(record: proto::Record) -> Result<Self, 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<Append> for proto::AppendRequest {
    fn from(message: Append) -> Self {
        proto::AppendRequest {
            shard: 0,
            term: message.term,
            previous_sequence: message.previous_sequence,
            records: message
                .records
                .iter()
                .map(proto::Record::from)
                .collect(),
            committed_sequence: message.committed_sequence,
            // The fields that leader election added are not used by this program.
            previous_term: 0,
            leader: 0,
        }
    }
}

impl From<proto::AppendResponse> for Appended {
    fn from(answer: proto::AppendResponse) -> Self {
        Appended {
            term: answer.term,
            appended: answer.appended,
            last_sequence: answer.last_sequence,
        }
    }
}

/// The server side of `Append`, which a follower runs.
pub struct Follower {
    pub node: Arc<Node>,
}

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

        let message = Append {
            term: request.term,
            previous_sequence: request.previous_sequence,
            records,
            committed_sequence: request.committed_sequence,
        };

        // Appending waits for the disk, so it runs where blocking is allowed.
        let answer = task::block_in_place(|| {
            self.node
                .replica
                .lock()
                .handle_append(message)
        })
        .map_err(|error| Status::unavailable(error.to_string()))?;

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

Listing 3.36 The complete kvnode/src/node.rs. kvnode/src/node.rs

//! A node of a replicated shard: a replica, and for a leader the tasks that
//! keep its followers up to date.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use kv::{Change, Condition, Entry, Summary, Written};
use kvd::api::ApiError;
use parking_lot::Mutex;
use repl::{Appended, Replica, majority_sequence};
use tokio::sync::{Notify, watch};
use tokio::task;
use tokio::time::{sleep, timeout};
use tonic::transport::{Channel, Endpoint};

use crate::rpc::{ReplicationClient, proto};

/// How often a leader sends a message to a follower that has nothing new to receive.
const HEARTBEAT: Duration = Duration::from_millis(100);

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

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

/// What the leader knows about one follower.
pub struct Peer {
    pub address: String,
    /// The last sequence the follower has reported.
    reported: AtomicU64,
    /// When the follower last answered.
    heard: Mutex<Option<Instant>>,
}

pub struct Node {
    pub replica: Mutex<Replica>,
    pub leader: bool,
    pub peers: Vec<Peer>,
    /// The sequence the log ended at when the node started. A leader serves
    /// reads once this much is committed.
    ready: u64,
    /// Held by the write that is in progress, so that writes take turns.
    turn: tokio::sync::Mutex<()>,
    /// The committed sequence, for writes that wait for it to advance.
    committed: watch::Sender<u64>,
    /// Signalled when the leader has appended a record.
    appended: Notify,
    /// Set when a follower answers with a newer term: another leader exists.
    deposed: AtomicBool,
}

impl Node {
    pub fn new(replica: Replica, leader: bool, followers: Vec<String>) -> Arc<Node> {
        let peer = |address| Peer {
            address,
            reported: AtomicU64::new(0),
            heard: Mutex::new(None),
        };

        Arc::new(Node {
            ready: replica.last_sequence(),
            replica: Mutex::new(replica),
            leader,
            peers: followers
                .into_iter()
                .map(peer)
                .collect(),
            turn: tokio::sync::Mutex::new(()),
            committed: watch::Sender::new(0),
            appended: Notify::new(),
            deposed: AtomicBool::new(false),
        })
    }
}

impl Node {
    /// Starts one task for each follower. A task sends the follower what it
    /// lacks, again and again, for as long as the node runs.
    pub fn lead(self: &Arc<Node>) -> Result<(), tonic::transport::Error> {
        for index in 0..self.peers.len() {
            let channel = Endpoint::from_shared(self.peers[index].address.clone())?
                .connect_timeout(ANSWER)
                .connect_lazy();

            tokio::spawn(Arc::clone(self).follow(index, ReplicationClient::new(channel)));
        }

        Ok(())
    }

    async fn follow(self: Arc<Node>, index: usize, mut client: ReplicationClient<Channel>) {
        let peer = &self.peers[index];

        loop {
            let message = self.replica.lock().append_for(
                peer.reported
                    .load(Ordering::Relaxed),
            );
            let request = proto::AppendRequest::from(message);

            let answer = timeout(ANSWER, client.append(request)).await;

            // An answer that did not come counts as a peer that is up to date:
            // the next round is a heartbeat, which finds out more.
            let behind = answer
                .ok()
                .and_then(Result::ok)
                .is_some_and(|answer| self.record(peer, answer.into_inner().into()));

            if !behind {
                // Wait for a new record, or send a heartbeat when none comes.
                tokio::select! {
                    () = self.appended.notified() => {}
                    () = sleep(HEARTBEAT) => {}
                }
            }
        }
    }
}

impl Node {
    /// Takes note of a follower's answer and commits what a majority now
    /// holds. It returns whether the follower still lacks records.
    fn record(&self, peer: &Peer, answer: Appended) -> bool {
        let mut replica = self.replica.lock();

        if answer.term > replica.term() {
            self.deposed
                .store(true, Ordering::Relaxed);

            return false;
        }

        peer.reported
            .store(answer.last_sequence, Ordering::Relaxed);
        *peer.heard.lock() = Some(Instant::now());

        let mut stored: Vec<u64> = self
            .peers
            .iter()
            .map(|peer| {
                peer.reported
                    .load(Ordering::Relaxed)
            })
            .collect();
        stored.push(replica.last_sequence());

        replica.commit(majority_sequence(stored));
        self.committed
            .send_replace(replica.committed());

        answer.last_sequence < replica.last_sequence()
    }
}

impl Node {
    /// A write on the leader: wait for a turn, append the record, and wait
    /// until a majority of the replicas hold it.
    pub async fn write(&self, change: Change, condition: Condition) -> Result<Written, ApiError> {
        self.must_lead()?;

        let turn = self.turn.lock().await;
        let last = self.replica.lock().last_sequence();
        self.wait_for(last).await?;

        let proposed = task::block_in_place(|| {
            let mut replica = self.replica.lock();
            let created = match &change {
                Change::Put { key, .. } | Change::Delete { key } => replica.get(key).is_none(),
            };

            replica
                .propose(change, condition)
                .map(|version| Written { version, created })
        })?;

        self.appended.notify_waiters();
        self.wait_for(proposed.version)
            .await?;
        drop(turn);

        Ok(proposed)
    }

    /// Waits until the record with this sequence is committed.
    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 replicas"
            )))
        }
    }

    fn must_lead(&self) -> Result<(), ApiError> {
        if !self.leader {
            return Err(ApiError::Misdirected("this node is a follower".into()));
        }

        if self.deposed.load(Ordering::Relaxed) {
            return Err(ApiError::Misdirected(
                "a leader of a newer term exists".into(),
            ));
        }

        Ok(())
    }
}

impl Node {
    /// A leader serves reads while a majority of the replicas, itself
    /// included, has answered within the last second, and once everything
    /// its log held at start is committed.
    fn must_serve_reads(&self) -> Result<(), ApiError> {
        self.must_lead()?;

        let recently = |peer: &&Peer| {
            peer.heard
                .lock()
                .is_some_and(|at| at.elapsed() < COMMIT)
        };
        let reachable = 1 + self
            .peers
            .iter()
            .filter(recently)
            .count();

        if 2 * reachable <= self.peers.len() + 1 {
            return Err(ApiError::Unavailable(
                "this leader cannot reach a majority of the replicas".into(),
            ));
        }

        if self.replica.lock().committed() < self.ready {
            return Err(ApiError::Unavailable(
                "the records from before the restart are not committed yet".into(),
            ));
        }

        Ok(())
    }

    pub fn get(&self, key: &str) -> Result<Entry, ApiError> {
        self.must_serve_reads()?;

        self.replica
            .lock()
            .get(key)
            .cloned()
            .ok_or(ApiError::NotFound)
    }

    pub fn list(
        &self,
        prefix: &str,
        after: Option<&str>,
        limit: usize,
    ) -> Result<(Vec<Summary>, bool), ApiError> {
        self.must_serve_reads()?;

        Ok(self
            .replica
            .lock()
            .list(prefix, after, limit))
    }

    /// The last sequence each follower has reported.
    pub fn reported(&self) -> Vec<u64> {
        self.peers
            .iter()
            .map(|peer| {
                peer.reported
                    .load(Ordering::Relaxed)
            })
            .collect()
    }
}

Listing 3.37 The complete kvnode/src/main.rs. kvnode/src/main.rs

//! Command kvnode is a storage node that replicates its log. One node of a
//! shard is started as the leader, with the addresses of its followers, and
//! the others as followers. The leader serves the HTTP API and acknowledges
//! a write when a majority of the nodes hold its record.
//!
//! Run 3 nodes from code/ with:
//!
//!     cargo run -p kvnode -- --role follower --listen 127.0.0.1:7002 --replication 127.0.0.1:7102 --data /var/tmp/n2
//!     cargo run -p kvnode -- --role follower --listen 127.0.0.1:7003 --replication 127.0.0.1:7103 --data /var/tmp/n3
//!     cargo run -p kvnode -- --role leader --listen 127.0.0.1:7001 --data /var/tmp/n1 \
//!         --follower http://127.0.0.1:7102 --follower http://127.0.0.1:7103

mod node;
mod rpc;

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, Query, 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, ValueEnum};
use kv::{Change, Condition};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use repl::Replica;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use tonic::transport::Server;

use crate::node::Node;
use crate::rpc::{Follower, ReplicationServer};

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

#[derive(Parser)]
struct Options {
    #[arg(long, value_enum)]
    role: Role,

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

    /// The address on which a follower accepts records from the leader.
    #[arg(long, default_value = "127.0.0.1:7101")]
    replication: SocketAddr,

    /// The replication address of a follower, such as http://127.0.0.1:7102.
    /// Given once for each follower, to the leader.
    #[arg(long = "follower")]
    followers: Vec<String>,

    /// The term this node starts in. A new leader is given a higher term
    /// than the leader it replaces.
    #[arg(long, default_value_t = 1)]
    term: u64,

    #[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(Deserialize)]
struct ListQuery {
    #[serde(default)]
    prefix: String,
    after: Option<String>,
    limit: Option<usize>,
}

#[derive(Serialize)]
struct Item {
    key: String,
    version: u64,
    size: usize,
}

#[derive(Serialize)]
struct Page {
    keys: Vec<Item>,
    next_after: Option<String>,
}

async fn list_keys(
    State(node): State<Arc<Node>>,
    Query(query): Query<ListQuery>,
) -> Result<Json<Page>, ApiError> {
    let limit = query
        .limit
        .unwrap_or(100)
        .clamp(1, 1000);
    let (page, more) = node.list(&query.prefix, query.after.as_deref(), limit)?;
    let next_after = if more {
        page.last()
            .map(|last| last.key.clone())
    } else {
        None
    };
    let keys = page
        .into_iter()
        .map(|item| Item {
            key: item.key,
            version: item.version,
            size: item.size,
        })
        .collect();

    Ok(Json(Page { keys, next_after }))
}

#[derive(Serialize)]
struct Health {
    role: &'static str,
    term: u64,
    last_sequence: u64,
    committed: u64,
    /// The last sequence each follower has reported to this leader.
    followers: Vec<u64>,
}

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

    Json(Health {
        role: if node.leader { "leader" } else { "follower" },
        term: replica.term(),
        last_sequence: replica.last_sequence(),
        committed: replica.committed(),
        followers: node.reported(),
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let replica = Replica::open(&options.data, !options.no_sync, options.term)?;

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

    let node = Node::new(replica, options.role == Role::Leader, options.followers);

    match options.role {
        Role::Leader => node.lead()?,
        Role::Follower => {
            let service = ReplicationServer::new(Follower {
                node: Arc::clone(&node),
            });

            tokio::spawn(
                Server::builder()
                    .add_service(service)
                    .serve(options.replication),
            );
            println!("accepting records on {}", options.replication);
        }
    }

    let routes = Router::new()
        .route("/v1/keys", get(list_keys))
        .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(())
}

Further reading

• B. Oki and B. Liskov, “Viewstamped Replication: A New Primary Copy Method to Support Highly-Available Distributed Systems”, PODC 1988.
• D. Ongaro and J. Ousterhout, “In Search of an Understandable Consensus Algorithm”, USENIX ATC 2014.
• R. van Renesse and F. Schneider, “Chain Replication for Supporting High Throughput and Availability”, OSDI 2004.