Ledger, a tortoise, writes each change in a long ledger before saying yes. Lightning strikes beside it and a crab jumps in shock, and the ledger still holds every line that was written: put user-1, put user-2, delete user-2.

Key-Value Store: Storage Node (Low-Level Design)

The design of chapter 1 distributes a key-value store over 24 machines, keeps every key on 3 of them, and reasons at length about what happens when machines fail. All of that reasoning rests on a single assumption about each individual machine, which chapter 1 stated in its closing section and did not justify: that a write which a node has acknowledged will still be there after the node has crashed and been restarted. If 1 node can quietly lose a write it has confirmed, then a write “stored on 2 of 3 nodes” may be stored on 1, and nothing built on top of that count can be trusted.

This chapter builds the component that makes the assumption true. A storage node is the program that holds the entries of the store on 1 machine, and its whole difficulty is concentrated in the few milliseconds during which a write is in progress and the machine may lose power. We begin by examining what a crash can do to a write at each stage, which leads to the technique of writing every change to a log before doing anything else. We then define the format of the log on disk, write the code that recovers from a damaged log, and build on top of it the store and the HTTP API of chapter 1. The chapter ends by trying, in 3 different ways, to make the node lose data: by damaging its log at every possible byte, by killing it under load, and by killing it under load from another machine.

This chapter covers

what a crash does to a write at each stage
write-ahead logging, and why the log is written before memory is changed
a format for records, with a length and a checksum
recovery: reading a log, and cutting off a damaged end
the system calls write and fsync, and what each of them costs
what to do when the disk reports an error
a store with entries, versions and conditional writes
an HTTP server that provides the API of chapter 1
a simulation that damages a log at every byte
a test that kills the server under load, on 1 machine and on 2

Storing data durably

A storage node keeps its entries in memory, so that a read requires no access to the disk. Memory is lost when the process stops, and so the node must keep the data on disk as well, in a form from which the contents of memory can be reconstructed.

The first design that comes to mind keeps a file which mirrors the contents of memory and, on every write, overwrites the affected part of that file. In the presence of crashes this is unsafe, and the reason is worth seeing concretely. Suppose that a value of 1 KB is being replaced by another of the same size, and that the power fails when half of the new bytes have reached the disk. The file now contains the first half of the new value followed by the second half of the old. The entry is corrupt, its former value can no longer be recovered, and, worst of all, nothing in the file indicates that anything is wrong.

The fault in this design is that an update in place destroys the old state before the new state is complete, so that there is a moment at which neither exists. The standard remedy is to give up updating in place altogether. The node appends a description of each change to the end of a file and modifies the file in no other way. A file used in this manner is called a log. Because every change is entered in the log before it is applied to any other structure, the technique is known as write-ahead logging, and the file as a write-ahead log.

With a file that is only ever appended to, a crash during a write can affect nothing but the record that was being appended. All the earlier records are intact, since nothing ever writes to that part of the file again.

The inside of a storage node. The HTTP handler reads the key, the headers and the body, and calls the store, which checks the condition on the version. A write goes first to the log on disk, which holds every change in order and survives a crash, and second to the map in memory, which goes from key to entry, is sorted by key, and is rebuilt at start. A read uses map only. — **Figure 2.1** The components of a storage node, and the order of the steps of a write.

The node maintains 4 invariants, conditions that hold at every moment, whatever requests arrive and wherever a crash occurs.

Every change is in the log, and confirmed by the disk, before its client receives a response.
Every change is in the log before it is applied to the map in memory.
The map is equal to the result of applying all the records of the log in order.
The log is modified only by appending. The bytes of a confirmed record are never written again.

Each invariant purchases something specific. The first is durability itself. The second and third together ensure that the map contains nothing that is absent from the log, so that the loss of the map loses no data. And the fourth confines the damage that a crash can do to the end of the file, where recovery knows to look for it.

Crashes during a write

Given this ordering of steps, the outcome of a crash can be worked out for each stage of a write.

The 4 moments at which a node can crash during a write. Before the record is written: nothing is on disk, the client gets no answer, and after restart the write did not happen. In the middle of the record: part of a record is on disk, the client gets no answer, and after restart the part is cut off. After fsync but before the answer: the whole record is on disk, the client gets no answer, and after restart the write happened. After the answer: the whole Record is on disk, client got 201 or 204, and write happened. — **Figure 2.2** The 4 stages at which a crash can occur, and the state that is found after a restart.

The third case requires comment. The record has been stored durably, and the node crashes before it has sent its response. After the restart the write is present, although its client was never told that it had succeeded. This does not violate the guarantee of durability, which covers acknowledged writes and is silent about writes whose outcome was never reported. From the client’s point of view the outcome of such a write is unknown, which is the lost response that chapter 1 discussed under retries. The client resolves the uncertainty by sending the request again, and can do so safely because PUT is idempotent. This case is produced deliberately, by a test, in the section on killing the server.

The format of a record

When a log is read back, the reader has to determine where each record ends, since records differ in length. It must also be able to recognise a record that is incomplete or damaged, since the end of the file may contain the partial output of a write that a crash interrupted.

Both needs are met by beginning every record with a frame of 8 bytes, which holds the length of the data that follows and a checksum of it. The checksum is a CRC-32, a value of 32 bits computed from the data in such a way that any change to the data will almost certainly change the value. When a record is read, the checksum is computed afresh and compared with the one that was stored.

The 28 bytes of the first record of the log. The frame is 8 bytes: the length, 14 00 00 00, which says that the payload is 20 bytes, and the checksum, 8d d3 da 1d, the CRC-32 of those 20 bytes. The payload is the sequence 1 as 8 bytes, the kind 01 for a put, the key length 6 as 2 bytes, the 6 bytes of the key user-1, and the value ada as the Rest of payload. — **Figure 2.3** The layout of the first record of a log.

The following are the first bytes of a log after the 2 writes to user-1 shown in chapter 1.

$ xxd data/kv.log | head -4
00000000: 1400 0000 8dd3 da1d 0100 0000 0000 0000  ................
00000010: 0106 0075 7365 722d 3161 6461 1d00 0000  ...user-1ada....
00000020: cefd 7d53 0200 0000 0000 0000 0106 0075  ..}S...........u
00000030: 7365 722d 3161 6461 206c 6f76 656c 6163  ser-1ada lovelac

The first 4 bytes, 1400 0000, encode the length 20 in little-endian order, in which the least significant byte of a number comes first. The 4 bytes of the checksum follow. Within the payload, the sequence number 1 occupies the first 8 bytes, and is followed by the kind, 01, and the length of the key, 0600. The key and the value can be read in the right-hand column. At byte 28, which is 0x1c, the second record begins with the length 1d00 0000, or 29.

Listing 2.1 The imports of kv/src/record.rs. kv/src/record.rs

use crc32fast::hash;

Listing 2.2 The types for a change, a record, and the result of decoding. kv/src/record.rs

/// A change to the store.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Change {
    Put { key: String, value: Vec<u8> },
    Delete { key: String },
}

/// A change with its position in the log. Sequence numbers start at 1 and
/// grow by 1 with each record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Record {
    pub sequence: u64,
    pub change: Change,
}

/// The result of reading one record from the front of a slice of bytes.
#[derive(Debug, PartialEq, Eq)]
pub enum Decoded {
    /// A whole record, and the number of bytes it occupies.
    Record { record: Record, length: usize },
    /// The bytes end before the record does.
    Incomplete,
    /// The checksum does not match the payload, or the payload is malformed.
    Corrupt,
}

/// The bytes of a frame: the length and the checksum.
pub const FRAME: usize = 8;

/// The largest payload a record may have: a key, a value of 1 MiB, and the
/// fixed fields.
pub const MAX_PAYLOAD: usize = (1 << 20) + 1024;

const PUT: u8 = 1;
const DELETE: u8 = 2;

Decoded has 3 variants because a reader must tell 3 situations apart: a complete record, data that ends before the record is complete, and data that is complete and invalid. The distinction between the last 2 is of no consequence to recovery, which treats them alike, and is valuable when a log is being examined after a fault, since the one indicates an interrupted write and the other a damaged disk.

Listing 2.3 Encoding a record. kv/src/record.rs

impl Record {
    /// Returns the record as the bytes that are appended to the log.
    pub fn encode(&self) -> Vec<u8> {
        let (kind, key, value) = match &self.change {
            Change::Put { key, value } => (PUT, key, value.as_slice()),
            Change::Delete { key } => (DELETE, key, &[][..]),
        };

        let mut payload = Vec::with_capacity(11 + key.len() + value.len());
        payload.extend_from_slice(&self.sequence.to_le_bytes());
        payload.push(kind);
        payload.extend_from_slice(&(key.len() as u16).to_le_bytes());
        payload.extend_from_slice(key.as_bytes());
        payload.extend_from_slice(value);

        let mut bytes = Vec::with_capacity(FRAME + payload.len());
        bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&hash(&payload).to_le_bytes());
        bytes.extend_from_slice(&payload);

        bytes
    }
}

Decoding carries out the same steps in order and returns the variant that corresponds to the first step that fails.

Listing 2.4 Decoding a record. kv/src/record.rs

impl Record {
    /// Reads the record at the front of `bytes`.
    pub fn decode(bytes: &[u8]) -> Decoded {
        let Some((length, rest)) = bytes.split_first_chunk::<4>() else {
            return Decoded::Incomplete;
        };
        let Some((checksum, rest)) = rest.split_first_chunk::<4>() else {
            return Decoded::Incomplete;
        };

        let length = u32::from_le_bytes(*length) as usize;

        if length > MAX_PAYLOAD {
            return Decoded::Corrupt;
        }

        let Some(payload) = rest.get(..length) else {
            return Decoded::Incomplete;
        };

        if hash(payload) != u32::from_le_bytes(*checksum) {
            return Decoded::Corrupt;
        }

        match parse(payload) {
            Some(record) => Decoded::Record {
                record,
                length: FRAME + length,
            },
            None => Decoded::Corrupt,
        }
    }
}

/// Splits a payload into its fields. It returns `None` for a payload that is
/// too short for its fields or has an unknown kind.
fn parse(payload: &[u8]) -> Option<Record> {
    let (sequence, rest) = payload.split_first_chunk::<8>()?;
    let (&kind, rest) = rest.split_first()?;
    let (key_length, rest) = rest.split_first_chunk::<2>()?;
    let (key, value) = rest.split_at_checked(u16::from_le_bytes(*key_length) as usize)?;
    let key = String::from_utf8(key.to_vec()).ok()?;

    let change = match (kind, value) {
        (PUT, value) => Change::Put {
            key,
            value: value.to_vec(),
        },
        (DELETE, []) => Change::Delete { key },
        (_unknown, _value) => return None,
    };

    Some(Record {
        sequence: u64::from_le_bytes(*sequence),
        change,
    })
}

split_first_chunk returns the first bytes of a slice as an array of fixed size, together with the remainder, or None if the slice is too short. Every field is extracted by this means. A payload shorter than its fields therefore yields None, and at no point does the code use a value read from the disk as an index without first having validated it. A decoder of data from disk must be written on the assumption that the data may be anything at all, since after a fault it may be.

What a durable write costs

Invariant 1 requires that the disk has confirmed every record. Writing to a file does not, by itself, satisfy this requirement, and the difference between the 2 is the most consequential fact about storage that a programmer has to know.

The system call write copies data into a buffer belonging to the kernel and returns. At some later time, the kernel transfers the buffer to the disk. With the call fsync, a program instructs the kernel to transfer all the buffered data of a file at once, and the call does not return until the disk has reported success. In Rust, File::sync_data makes this call.

What survives 2 kinds of failure. When the process is killed, the map in the process is lost, bytes given to write are kept, and bytes after fsync are kept. When the power fails, the map is lost, bytes given to write are lost, and bytes after fsync are kept. After write the bytes are in the kernel's memory, and the kernel outlives the process, So only a power failure shows difference. — **Figure 2.4** The data that is lost in each kind of failure.

The program synccost appends 500 records to a log, first without fsync and then with an fsync after every record.

Listing 2.5 The imports of synccost/src/main.rs. synccost/src/main.rs

use std::env;
use std::error::Error;
use std::fs;
use std::path::Path;
use std::time::{Duration, Instant};

use kv::{Change, Log};

Listing 2.6 Measuring the cost of an append with and without fsync. synccost/src/main.rs

const RECORDS: u32 = 500;

/// Appends `RECORDS` records of 100 bytes and returns the time it took.
fn measure(directory: &Path, sync: bool) -> Result<Duration, Box<dyn Error>> {
    let mut log = Log::open(directory, sync)?.0;
    let started = Instant::now();

    for number in 0..RECORDS {
        log.append(Change::Put {
            key: format!("key-{number:06}"),
            value: vec![b'v'; 100],
        })?;
    }

    Ok(started.elapsed())
}

fn main() -> Result<(), Box<dyn Error>> {
    let root = env::args()
        .nth(1)
        .ok_or("usage: synccost <directory>")?;

    for (name, sync) in [("without fsync", false), ("with fsync", true)] {
        let directory = Path::new(&root).join(name.replace(' ', "-"));
        let elapsed = measure(&directory, sync)?;
        let each = elapsed.as_secs_f64() * 1000.0 / f64::from(RECORDS);

        println!(
            "{name:13}: {RECORDS} records in {:8.1} ms, {each:.3} ms per record, {:6.0} records per second",
            elapsed.as_secs_f64() * 1000.0,
            1000.0 / each
        );

        fs::remove_dir_all(&directory)?;
    }

    Ok(())
}

$ cargo run --release -p synccost -- /var/tmp/synccost
without fsync: 500 records in      1.4 ms, 0.003 ms per record, 366991 records per second
with fsync   : 500 records in   2697.5 ms, 5.395 ms per record,    185 records per second

The cost of fsync on an NVMe disk. Appending 1 record takes 0.003 ms with write only and 5.395 ms with write followed by fsync. That is 366991 records per second against 185. write copies the bytes into the kernel's memory and returns, and fsync waits until the disk reports that the bytes are stored. — **Figure 2.5** The 2 measurements, drawn to scale.

On an NVMe disk, a durable append took 5.4 ms, about 1900 times as long as an append that merely copies data into the kernel’s buffer. This is the figure from which chapter 1 derived its 185 records per second. Because the cost is so high, storage systems commonly make the behaviour configurable, and kvd has an option --no-sync under which the node responds after write alone. In that mode the node keeps every acknowledged write if its process crashes, and may lose the most recent writes if the machine loses power. Whether that is an acceptable bargain depends on the data, and it is a decision that ought to be made knowingly, since the default behaviour of a great many programs makes it silently.

Design overview

Low-level design: the storage node

Crates

Crate	Kind	Contents
`kv`	library	`record` (the format of a record), `log` (the log file), `store` (the map and the conditions on writes)
`kvd`	library and program	`api` (errors and conditional headers, shared with later chapters) and the HTTP server
`kvctl`	program	a client that generates load and verifies acknowledged writes
`crashsim`	program	a simulation that damages a log at every position
`synccost`	program	the measurement of `fsync`

Types

Type	Fields	Purpose
`Change`	`Put { key, value }` or `Delete { key }`	the modification that a write requests
`Record`	`sequence`, `change`	a change and its position in the log
`Log`	`file`, `sync`, `next`, `failed`	appends records and reads them back
`Entry`	`value`, `version`	the current state of a key
`Condition`	`Always`, `Absent`, `Version(n)`	the state of the key that a write requires
`Store`	`entries: BTreeMap<String, Entry>`, `log`	applies changes and answers reads

Operations

Operation	Steps	Cost
`Store::put`	check the condition, `Log::append`, apply to the map	1 `fsync`
`Store::delete`	check that the key exists, check the condition, `Log::append`, apply	1 `fsync`
`Store::get`	look the key up in the map	O(log n), with no access to the disk
`Store::list`	iterate over the map from a starting key	O(log n + size of the page), with no access to the disk
`Store::open`	`Log::open`, then apply every record	proportional to the size of the log

Failure policy

Event	Response
The log ends with an incomplete or damaged record	The file is cut off after the last valid record.
A write to the log fails	The node returns `503` and rejects every later write until it is restarted.

The manifest of the workspace declares the edition of Rust, the minimum version of the compiler, and the dependencies that its crates share.

Listing 2.7 The complete Cargo.toml. Cargo.toml

[workspace]
resolver = "3"
members = ["kv", "kvd", "kvctl", "crashsim", "synccost", "ring", "ringsim", "repl", "replsim", "kvnode", "raft", "raftsim", "kvraft", "probe", "kvrouter", "kvgroup", "kvbench",
    "ratelimit", "idgen", "shortener", "cachesim", "queuesim", "locksim", "jobsim", "notifysim", "feedsim", "objectsim",
    "streamsim", "chatsim", "paysim", "bookingsim", "nearbysim", "typeahead",
    "crawlsim", "metricsim", "syncsim", "editsim", "matchsim", "clicksim",
]

[workspace.package]
edition = "2024"
rust-version = "1.99"
license = "MIT"
publish = false

[workspace.dependencies]
kv = { path = "kv" }
ring = { path = "ring" }
repl = { path = "repl" }
raft = { path = "raft" }
kvd = { path = "kvd" }
axum = "0.8"
clap = { version = "4", features = ["derive"] }
crc32fast = "1"
parking_lot = "0.12"
reqwest = { version = "0.13", default-features = false, features = ["json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
prost = "0.14"
tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "sync", "time"] }
tonic = "0.14"
tonic-prost = "0.14"
tonic-prost-build = "0.14"
ureq = "3"
xxhash-rust = { version = "0.8", features = ["xxh3"] }

Dependency	Purpose
`axum`, `tokio`	the HTTP server and the asynchronous runtime
`clap`	parsing the options of the command line
`crc32fast`	the checksum of a record
`parking_lot`	a mutex whose `lock` method returns the guard directly
`serde`	JSON encoding of the list response and of error bodies
`thiserror`	deriving `Display` and `From` for the store’s error type
`ureq`	the HTTP client used by `kvctl`
`xxhash-rust`	the hash function of the ring described in chapter 1

The log

Log owns the log file. It offers 2 operations: opening, which includes recovery, and appending a record.

Listing 2.8 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 2.9 The log, and the result of opening it. kv/src/log.rs

/// The name of the log inside the data directory.
pub const FILE_NAME: &str = "kv.log";

/// An open log, positioned at its end.
pub struct Log {
    file: File,
    /// Whether `append` waits for the disk before it returns.
    sync: bool,
    /// The sequence number of the next record.
    next: u64,
    /// Set by a failed write. A failed log accepts no more records.
    failed: bool,
}

/// What `Log::open` found in the file.
pub struct Recovery {
    /// The complete records, in the order they were written.
    pub records: Vec<Record>,
    /// The bytes after the last complete record, which were removed.
    pub discarded: u64,
}

Recovery

Opening a log reads the entire file and decodes records from the beginning until one fails to decode. Since Record::decode already distinguishes every possible outcome, the loop is short. Recovery treats an incomplete record and a corrupt one alike: in either case the record before it becomes the last record of the log.

Listing 2.10 Decoding records until one is incomplete or corrupt. kv/src/log.rs

/// Reads records from the front of `bytes` until one is incomplete or
/// corrupt. It returns the records and the number of bytes they occupy.
fn scan(bytes: &[u8]) -> (Vec<Record>, usize) {
    let mut records = Vec::new();
    let mut valid = 0;

    while let Decoded::Record { record, length } = Record::decode(&bytes[valid..]) {
        records.push(record);
        valid += length;
    }

    (records, valid)
}

Recovery of a log of 63 bytes after a crash. Record 1 occupies bytes 0 to 27 and record 2 bytes 28 to 57, and 5 bytes of a third record follow. The scan reads a frame at 0 that says 20 bytes and whose checksum matches: record 1. At 28 the frame says 22 bytes and the checksum matches: record 2. At 58 only 5 bytes are left and a frame needs 8, so the record is incomplete and the scan stops. The file is cut to 58 bytes, and the next record will have sequence 3. — **Figure 2.6** The recovery of a log that was cut off in the middle of its third record.

Listing 2.11 Opening the log and cutting off a damaged end. kv/src/log.rs

impl Log {
    /// Opens the log in `directory`, creating both if they do not exist.
    /// Anything after the last complete record is cut off.
    pub fn open(directory: &Path, sync: bool) -> io::Result<(Log, Recovery)> {
        fs::create_dir_all(directory)?;

        let mut file = OpenOptions::new()
            .read(true)
            .append(true)
            .create(true)
            .open(directory.join(FILE_NAME))?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)?;

        let (records, valid) = scan(&bytes);
        let discarded = (bytes.len() - valid) as u64;

        if discarded > 0 {
            file.set_len(valid as u64)?;
        }

        if sync {
            // The file's length is part of the file. Its name is part of the directory.
            file.sync_all()?;
            File::open(directory)?.sync_all()?;
        }

        let next = records
            .last()
            .map_or(1, |record| record.sequence + 1);

        Ok((
            Log {
                file,
                sync,
                next,
                failed: false,
            },
            Recovery { records, discarded },
        ))
    }
}

The call to set_len removes the damaged end of the file before any new record is appended, and the correctness of the whole log depends on this step. If the damaged bytes were left where they are, the next record would be written after them, a later recovery would stop at the damage, and the new record would be lost although the node had acknowledged it. A recovery procedure that reads correctly and fails to repair has merely postponed the loss.

Both calls to sync_all are necessary, one to make the new length of the file durable and the other for its directory. Synchronising a file makes its contents durable, but the existence of the file is recorded in the directory that contains it, and that is a separate object. A newly created file whose directory has not been synchronised may be absent after a power failure, contents and all. The log therefore synchronises its directory when it is opened.

Appending

Listing 2.12 Appending a record. kv/src/log.rs

impl Log {
    /// Appends a change and returns the record that holds it. When the log
    /// was opened with `sync`, the record is on the disk when this returns.
    pub fn append(&mut self, change: Change) -> io::Result<Record> {
        if self.failed {
            return Err(io::Error::other(
                "the log failed on an earlier write and is closed",
            ));
        }

        let record = Record {
            sequence: self.next,
            change,
        };

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

            return Err(error);
        }

        self.next += 1;

        Ok(record)
    }

    fn write(&mut self, bytes: &[u8]) -> io::Result<()> {
        self.file.write_all(bytes)?;

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

        Ok(())
    }

    /// The sequence number of the last record, or 0 for an empty log.
    pub fn last_sequence(&self) -> u64 {
        self.next - 1
    }
}

The field failed implements a policy known as fail-stop, under which a component that meets an error it cannot handle safely ceases to operate. If write_all or sync_data returns an error, the state of the file is unknown: the record may be absent, partly written, or complete. A log that went on accepting records could append a valid record after a damaged one, where recovery would never reach it.

It might be thought that the node could call fsync again. This is unsafe as well, and for a reason that has surprised the authors of several well-known databases. After a failed fsync, the kernel may already have discarded the data it was unable to write, and a second call may then report success although nothing has reached the disk. A success after a failure therefore proves nothing. The log accordingly rejects all further appends. From that moment its node answers 503 to writes, continues to serve reads from memory, and has to be restarted or replaced by an operator. In the design of chapter 1, the followers of all the affected shards elect new leaders at this point, and the fault of 1 disk is absorbed by the cluster.

The store

Store contains the map and the log, and every modification of either passes through it.

Listing 2.13 The imports of kv/src/store.rs. kv/src/store.rs

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

use thiserror::Error;

use crate::log::Log;
use crate::record::{Change, Record};

Listing 2.14 Opening the store by applying the records of the log. kv/src/store.rs

impl Store {
    /// Opens the store in `directory` and applies every record of its log.
    /// It returns the store, the number of records applied, and the number
    /// of bytes that were cut from the end of the log.
    pub fn open(directory: &Path, sync: bool) -> io::Result<(Store, usize, u64)> {
        let (log, recovery) = Log::open(directory, sync)?;
        let applied = recovery.records.len();
        let mut entries = BTreeMap::new();

        for record in recovery.records {
            apply(&mut entries, record);
        }

        Ok((Store { entries, log }, applied, recovery.discarded))
    }
}

/// Makes the change of one record in the map. Recovery and the write path
/// both go through this function, so they cannot disagree.
pub fn apply(entries: &mut BTreeMap<String, Entry>, record: Record) {
    match record.change {
        Change::Put { key, value } => {
            entries.insert(
                key,
                Entry {
                    value,
                    version: record.sequence,
                },
            );
        }
        Change::Delete { key } => {
            entries.remove(&key);
        }
    }
}

Recovery and the path of a write share a single function, apply, for modifying the map. Invariant 3 is a consequence of this arrangement. What a record does to the map is defined in exactly 1 place, which is used both when a record is first written and whenever a log is read back, so that the 2 cannot come to differ.

Listing 2.15 Conditional writes: check, append, apply. kv/src/store.rs

impl Store {
    /// Sets the value of a key. The change is in the log before it is in
    /// the map.
    pub fn put(
        &mut self,
        key: String,
        value: Vec<u8>,
        condition: Condition,
    ) -> Result<Written, StoreError> {
        let current = self.version(&key);
        check(condition, current)?;

        let record = self
            .log
            .append(Change::Put { key, value })?;
        let version = record.sequence;
        apply(&mut self.entries, record);

        Ok(Written {
            version,
            created: current.is_none(),
        })
    }

    /// Removes a key.
    pub fn delete(&mut self, key: &str, condition: Condition) -> Result<(), StoreError> {
        let current = self.version(key);

        if current.is_none() {
            return Err(StoreError::NotFound);
        }

        check(condition, current)?;

        let record = self.log.append(Change::Delete {
            key: key.to_owned(),
        })?;
        apply(&mut self.entries, record);

        Ok(())
    }

    fn version(&self, key: &str) -> Option<u64> {
        self.entries
            .get(key)
            .map(|entry| entry.version)
    }
}

pub fn check(condition: Condition, current: Option<u64>) -> Result<(), StoreError> {
    let holds = match condition {
        Condition::Always => true,
        Condition::Absent => current.is_none(),
        Condition::Version(version) => current == Some(version),
    };

    holds
        .then_some(())
        .ok_or(StoreError::ConditionFailed { current })
}

put and delete perform the same 3 steps in the same order. The condition is evaluated first, against the map, so that a write which is refused produces no record. append then returns when the record is durable, or returns an error, in which case the ? operator leaves the function before apply is reached. The map is modified last. The order of these steps is the order of the invariants, and the code has been arranged so that it cannot be otherwise.

The version of an entry is the sequence number of its record. The store keeps no separate counter of versions that could come to disagree with the log after a crash.

Reads use the map alone.

Listing 2.16 Reading a key, and listing a page of keys. kv/src/store.rs

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

    /// Returns up to `limit` keys that start with `prefix` and sort after
    /// `after`, in order, and whether more keys follow them.
    pub fn list(&self, prefix: &str, after: Option<&str>, limit: usize) -> (Vec<Summary>, bool) {
        page(&self.entries, prefix, after, limit)
    }

    pub fn len(&self) -> usize {
        self.entries.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// The sequence number of the last record in the log.
    pub fn last_sequence(&self) -> u64 {
        self.log.last_sequence()
    }
}

/// One page of a listing over a map of entries.
pub fn page(
    entries: &BTreeMap<String, Entry>,
    prefix: &str,
    after: Option<&str>,
    limit: usize,
) -> (Vec<Summary>, bool) {
    let start = match after {
        Some(key) => Bound::Excluded(key),
        None => Bound::Included(prefix),
    };

    let mut matching = entries
        .range::<str, _>((start, Bound::Unbounded))
        .take_while(|entry| entry.0.starts_with(prefix))
        .map(|(key, entry)| Summary {
            key: key.clone(),
            version: entry.version,
            size: entry.value.len(),
        });

    let page: Vec<Summary> = matching
        .by_ref()
        .take(limit)
        .collect();
    let more = matching.next().is_some();

    (page, more)
}

A BTreeMap keeps its keys in sorted order, so that a listing is an iteration from some starting position, which is implemented in a function named page. Its call to range begins at the prefix or immediately after the cursor, and take_while ends the iteration at the first key that lacks the prefix. By looking for 1 item beyond the page, page learns whether a further page exists without having to count the remaining keys.

The HTTP server

kvd presents the store through the API of chapter 1.

Listing 2.17 The imports of kvd/src/main.rs. kvd/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;
use kv::{Condition, Store, StoreError, Summary};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use tokio::task;

Listing 2.18 The options of the command line, and the shared state. kvd/src/main.rs

#[derive(Parser)]
struct Options {
    /// The address to accept connections on.
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    /// The directory that holds the log.
    #[arg(long, default_value = "data")]
    data: PathBuf,

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

/// The store, shared by every request.
type Shared = Arc<Mutex<Store>>;

Errors

Each kind of failure of a request is a variant of ApiError, in the module kvd::api, and a single function converts a variant into a status code and a response body in the format of RFC 9457. Handlers return Result<_, ApiError> and use ?. The conversion from StoreError is the 1 place that maps a failed condition to 412 and a failure of the log to 503.

Listing 2.19 The imports of kvd/src/api.rs. kvd/src/api.rs

use axum::Json;
use axum::http::header::{CONTENT_TYPE, IF_MATCH, IF_NONE_MATCH};
use axum::http::{HeaderMap, HeaderName, StatusCode};
use axum::response::{IntoResponse, Response};
use kv::{Condition, StoreError};
use serde::Serialize;

Listing 2.20 The errors of the API and their responses. kvd/src/api.rs

/// The ways a request can fail. Each one is a status code and a sentence.
pub enum ApiError {
    InvalidKey,
    InvalidHeader(HeaderName),
    NotFound,
    ConditionFailed(Option<u64>),
    Unavailable(String),
    /// The request reached a node that does not serve it. The text names the reason.
    Misdirected(String),
    /// The shard has no leader at the moment. Nothing was changed.
    NoLeader,
}

/// The body of an error response, in the format of RFC 9457.
#[derive(Serialize)]
struct Problem {
    #[serde(rename = "type")]
    kind: &'static str,
    title: &'static str,
    status: u16,
    #[serde(skip_serializing_if = "String::is_empty")]
    detail: String,
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, kind, title, detail) = match self {
            ApiError::InvalidKey => (
                StatusCode::BAD_REQUEST,
                "/problems/invalid-key",
                "The key is not valid",
                format!("a key is 1 to {MAX_KEY} characters from A-Z, a-z, 0-9, '.', '_' and '-'"),
            ),
            ApiError::InvalidHeader(name) => (
                StatusCode::BAD_REQUEST,
                "/problems/invalid-header",
                "A request header is not valid",
                format!("{name} must be a version in double quotes, or * where that is allowed"),
            ),
            ApiError::NotFound => (
                StatusCode::NOT_FOUND,
                "/problems/key-not-found",
                "The key does not exist",
                String::new(),
            ),
            ApiError::ConditionFailed(current) => (
                StatusCode::PRECONDITION_FAILED,
                "/problems/condition-failed",
                "The condition on the key's version does not hold",
                current.map_or("the key does not exist".into(), |version| {
                    format!("the key is at version {version}")
                }),
            ),
            ApiError::Misdirected(detail) => (
                StatusCode::MISDIRECTED_REQUEST,
                "/problems/not-leader",
                "This node is not the leader",
                detail,
            ),
            ApiError::NoLeader => (
                StatusCode::SERVICE_UNAVAILABLE,
                "/problems/no-leader",
                "The shard has no leader at the moment",
                String::new(),
            ),
            ApiError::Unavailable(detail) => (
                StatusCode::SERVICE_UNAVAILABLE,
                "/problems/store-unavailable",
                "The node cannot serve the request",
                detail,
            ),
        };

        let problem = Problem {
            kind,
            title,
            status: status.as_u16(),
            detail,
        };

        (
            status,
            [(CONTENT_TYPE, "application/problem+json")],
            Json(problem),
        )
            .into_response()
    }
}

impl From<StoreError> for ApiError {
    fn from(error: StoreError) -> Self {
        match error {
            StoreError::NotFound => ApiError::NotFound,
            StoreError::ConditionFailed { current } => ApiError::ConditionFailed(current),
            StoreError::Log(error) => ApiError::Unavailable(error.to_string()),
        }
    }
}

Conditions

Listing 2.21 Parsing the If-Match and If-None-Match headers. kvd/src/api.rs

/// A version as it appears in `ETag`, `If-Match` and `If-None-Match`.
pub fn etag(version: u64) -> String {
    format!("\"{version}\"")
}

pub fn valid_key(key: &str) -> Result<(), ApiError> {
    let allowed = |c: char| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-');

    let valid = (1..=MAX_KEY).contains(&key.len()) && key.chars().all(allowed);

    valid
        .then_some(())
        .ok_or(ApiError::InvalidKey)
}

/// Reads a header that holds a quoted version or `*`.
pub fn version_header(
    headers: &HeaderMap,
    name: HeaderName,
) -> Result<Option<Condition>, ApiError> {
    let Some(value) = headers.get(&name) else {
        return Ok(None);
    };

    let Ok(text) = value.to_str() else {
        return Err(ApiError::InvalidHeader(name));
    };

    if text == "*" {
        return Ok(Some(Condition::Absent));
    }

    let version = text
        .strip_prefix("\"")
        .and_then(|rest| rest.strip_suffix("\""))
        .and_then(|digits| digits.parse().ok());

    version
        .map(Condition::Version)
        .map(Some)
        .ok_or(ApiError::InvalidHeader(name))
}

/// The condition a write carries: `If-Match: "7"` requires version 7, and
/// `If-None-Match: *` requires that the key does not exist.
pub fn write_condition(headers: &HeaderMap) -> Result<Condition, ApiError> {
    match (
        version_header(headers, IF_MATCH)?,
        version_header(headers, IF_NONE_MATCH)?,
    ) {
        (Some(Condition::Version(version)), None) => Ok(Condition::Version(version)),
        (None, Some(Condition::Absent)) => Ok(Condition::Absent),
        (None, None) => Ok(Condition::Always),
        (Some(..), ..) => Err(ApiError::InvalidHeader(IF_MATCH)),
        (None, Some(..)) => Err(ApiError::InvalidHeader(IF_NONE_MATCH)),
    }
}

write_condition accepts the combinations of headers that are meaningful for a write and answers 400 to all others, such as If-Match: *. The server thus rejects a condition that it cannot interpret. The alternative, to ignore it, would mean performing an unconditional write on behalf of a client that had asked for a conditional one, which is exactly the lost update that the condition was sent to prevent.

Blocking calls in an asynchronous server

The server uses an asynchronous runtime, in which a small pool of threads serves all the connections and switches among them whenever one has to wait for the network. To block such a thread, as an fsync of 5 ms does, would delay every connection that the thread serves. spawn_blocking runs a closure on a separate pool of threads intended for operations that block, and the handler awaits the result without occupying a thread of the runtime.

Listing 2.22 Running a write on a thread that is permitted to block. kvd/src/main.rs

/// Runs a write on a thread that may block. A write waits for the disk, and
/// a thread that waits must not be one of the threads that serve requests.
async fn write<T, F>(store: Shared, work: F) -> Result<T, ApiError>
where
    F: FnOnce(&mut Store) -> Result<T, StoreError> + Send + 'static,
    T: Send + 'static,
{
    let result = task::spawn_blocking(move || work(&mut store.lock()))
        .await
        .map_err(|error| ApiError::Unavailable(error.to_string()))?;

    Ok(result?)
}

Listing 2.23 The handlers for PUT and DELETE. kvd/src/main.rs

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

    let written = write(store, move |store| store.put(key, 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(store): State<Shared>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    write(store, move |store| store.delete(&key, condition)).await?;

    Ok(StatusCode::NO_CONTENT)
}

PUT answers 201 Created when it creates a key and 204 No Content when it replaces a value. Both responses carry the new version in the ETag header.

Reads

Listing 2.24 The handler for GET on a single key. kvd/src/main.rs

async fn get_key(
    State(store): State<Shared>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let cached = version_header(&headers, IF_NONE_MATCH)?;
    let entry = store
        .lock()
        .get(&key)
        .cloned()
        .ok_or(ApiError::NotFound)?;
    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())
}

A read holds the lock of the store only while an entry is being copied, and performs no access to the disk. It can therefore run directly on a thread of the runtime.

Listing 2.25 The handlers for listing keys and for the health check. kvd/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>,
    /// The value of `after` that returns the next page, when there is one.
    next_after: Option<String>,
}

async fn list_keys(State(store): State<Shared>, Query(query): Query<ListQuery>) -> Json<Page> {
    let limit = query
        .limit
        .unwrap_or(100)
        .clamp(1, 1000);
    let (page, more) = store
        .lock()
        .list(&query.prefix, query.after.as_deref(), limit);
    let next_after = page
        .last()
        .filter(|_last| more)
        .map(|last| last.key.clone());
    let keys = page
        .into_iter()
        .map(|Summary { key, version, size }| Item { key, version, size })
        .collect();

    Json(Page { keys, next_after })
}

#[derive(Serialize)]
struct Health {
    status: &'static str,
    keys: usize,
    last_sequence: u64,
}

async fn health(State(store): State<Shared>) -> Json<Health> {
    let store = store.lock();

    Json(Health {
        status: "ok",
        keys: store.len(),
        last_sequence: store.last_sequence(),
    })
}

Listing 2.26 Opening the store and starting the server. kvd/src/main.rs

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

    println!(
        "recovered {} keys from {applied} records, cut {discarded} bytes from the end of the log",
        store.len()
    );

    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(Arc::new(Mutex::new(store)));

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

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

    Ok(())
}

Running the node

$ cd code
$ cargo build --release
$ ./target/release/kvd --listen 127.0.0.1:7001 --data /var/tmp/kvd
recovered 0 keys from 0 records, cut 0 bytes from the end of the log
listening on 127.0.0.1:7001

All the responses shown in chapter 1 came from this program. A further 3 complete the set: a key containing a character outside the permitted set, a method that a resource does not support, and a health check.

$ curl -i -X PUT 'http://127.0.0.1:7001/v1/keys/bad%20key' --data-binary 'x'
HTTP/1.1 400 Bad Request
content-type: application/problem+json
content-length: 153

{"type":"/problems/invalid-key","title":"The key is not valid","status":400,"detail":"a key is 1 to 256 characters from A-Z, a-z, 0-9, '.', '_' and '-'"}

$ curl -i -X POST http://127.0.0.1:7001/v1/keys/user-1 --data-binary 'x'
HTTP/1.1 405 Method Not Allowed
allow: GET,HEAD,PUT,DELETE
content-length: 0

Both the 405 and its allow header are produced by axum, whose router knows which methods are registered for each path.

$ curl -i http://127.0.0.1:7001/healthz
HTTP/1.1 200 OK
content-type: application/json
content-length: 42

{"status":"ok","keys":2,"last_sequence":5}

Simulating crashes

The log is intended to recover exactly its complete records, wherever a crash occurs. A claim of this kind is not established by a few examples, and crashsim verifies it exhaustively, for a log of 5 records, by damaging copies of the log in every possible position.

Listing 2.27 The imports of crashsim/src/main.rs. crashsim/src/main.rs

use std::env;
use std::error::Error;
use std::fs;
use std::path::Path;
use std::process;

use kv::log::FILE_NAME;
use kv::{Change, Log, Record};

Listing 2.28 Building the test log. crashsim/src/main.rs

/// The changes of the test log.
fn changes() -> Vec<Change> {
    let put = |key: &str, value: &str| Change::Put {
        key: key.into(),
        value: value.into(),
    };

    vec![
        put("user-1", "ada"),
        put("user-2", "grace"),
        put("user-1", "ada lovelace"),
        Change::Delete {
            key: "user-2".into(),
        },
        put("user-3", "barbara"),
    ]
}

/// Writes the changes to a log in `directory`. It returns the bytes of the
/// log and the offset at which each record ends.
fn build(directory: &Path) -> Result<(Vec<u8>, Vec<usize>), Box<dyn Error>> {
    let mut log = Log::open(directory, false)?.0;
    let mut ends = Vec::new();

    for change in changes() {
        let record = log.append(change)?;
        ends.push(ends.last().copied().unwrap_or(0) + record.encode().len());
    }

    Ok((fs::read(directory.join(FILE_NAME))?, ends))
}

Listing 2.29 Opening a damaged copy, and the expected result of recovery. crashsim/src/main.rs

/// Replaces the log in `directory` with `bytes` and opens it.
fn recover(directory: &Path, bytes: &[u8]) -> Result<Vec<Record>, Box<dyn Error>> {
    fs::write(directory.join(FILE_NAME), bytes)?;
    let recovery = Log::open(directory, false)?.1;

    Ok(recovery.records)
}

/// Whether `records` are exactly the first `count` changes, in order.
fn is_prefix(records: &[Record], count: usize) -> bool {
    let expected = changes();

    records.len() == count
        && records
            .iter()
            .zip(&expected)
            .enumerate()
            .all(|(index, (record, change))| {
                record.sequence == index as u64 + 1 && &record.change == change
            })
}

A crash during a write leaves a file that is shorter than was intended. cuts creates every such file, 1 for each possible length, and checks the result of recovering each.

Listing 2.30 Cutting the log short at every byte. crashsim/src/main.rs

/// Cuts the log at every length and checks that recovery returns the records
/// that end at or before the cut.
fn cuts(directory: &Path, log: &[u8], ends: &[usize]) -> Result<(), Box<dyn Error>> {
    for length in 0..=log.len() {
        let complete = ends
            .iter()
            .filter(|&&end| end <= length)
            .count();
        let records = recover(directory, &log[..length])?;

        if !is_prefix(&records, complete) {
            return Err(format!(
                "cut at {length}: expected {complete} records, recovered {}",
                records.len()
            )
            .into());
        }
    }

    println!(
        "cut at every length from 0 to {}: each recovery returned the complete records",
        log.len()
    );

    let mut first = 0;

    for (count, &end) in ends.iter().enumerate() {
        println!("  lengths {first:3} to {:3}: {count} records", end - 1);
        first = end;
    }

    println!("  length  {first:3}       : {} records", ends.len());

    Ok(())
}

Storage devices are also capable of returning data that differs from what was written. flips inverts 1 bit in each byte of the log, in a separate run for every byte. With a bit inverted, the affected record fails its checksum, and recovery is expected to return all the records before it.

Listing 2.31 Inverting 1 bit in every byte. crashsim/src/main.rs

/// Flips 1 bit in each byte in turn and checks that recovery returns the
/// records before the damaged one.
fn flips(directory: &Path, log: &[u8], ends: &[usize]) -> Result<(), Box<dyn Error>> {
    for position in 0..log.len() {
        let mut damaged = log.to_vec();
        damaged[position] ^= 0x10;

        let before = ends
            .iter()
            .filter(|&&end| end <= position)
            .count();
        let records = recover(directory, &damaged)?;

        if !is_prefix(&records, before) {
            return Err(format!(
                "flip at {position}: expected {before} records, recovered {}",
                records.len()
            )
            .into());
        }
    }

    println!(
        "1 bit flipped in each of {} bytes: each recovery returned the records before the damage",
        log.len()
    );

    Ok(())
}

Listing 2.32 Running the simulation. crashsim/src/main.rs

fn run(directory: &Path) -> Result<(), Box<dyn Error>> {
    let (log, ends) = build(directory)?;
    let boundaries: Vec<String> = ends
        .iter()
        .map(usize::to_string)
        .collect();

    println!(
        "a log of {} records and {} bytes, with records ending at {}",
        ends.len(),
        log.len(),
        boundaries.join(", ")
    );

    cuts(directory, &log, &ends)?;
    flips(directory, &log, &ends)?;

    let records = recover(directory, &log[..ends[1] + 5])?;
    let mut reopened = Log::open(directory, false)?.0;
    let next = reopened.append(Change::Delete {
        key: "user-1".into(),
    })?;

    println!(
        "after a cut at {}: {} records recovered, and the next record has sequence {}",
        ends[1] + 5,
        records.len(),
        next.sequence
    );

    Ok(())
}

fn main() -> Result<(), Box<dyn Error>> {
    let directory = env::temp_dir().join(format!("crashsim-{}", process::id()));
    let result = run(&directory);

    fs::remove_dir_all(&directory)?;

    result
}

$ cargo run -p crashsim
a log of 5 records and 152 bytes, with records ending at 28, 58, 95, 120, 152
cut at every length from 0 to 152: each recovery returned the complete records
  lengths   0 to  27: 0 records
  lengths  28 to  57: 1 records
  lengths  58 to  94: 2 records
  lengths  95 to 119: 3 records
  lengths 120 to 151: 4 records
  length  152       : 5 records
1 bit flipped in each of 152 bytes: each recovery returned the records before the damage
after a cut at 63: 2 records recovered, and the next record has sequence 3

Output	What it shows
lengths 28 to 57: 1 record	A record is recovered only if every one of its bytes is present. With 29 of the 30 bytes of record 2 on disk, record 2 is not recovered.
each recovery returned complete records	No cut causes recovery to return a partial record, a wrong record, or records out of order.
1 bit flipped in each of 152 bytes	Corruption at any position in a record, the length field included, is detected, and the preceding records are returned.
after a cut at 63, the next record has sequence 3	The damaged end was removed, and the log continues after the last valid record with no gap in its sequence numbers.

The test with inverted bits also discloses a limitation. When a bit is inverted in record 2, recovery stops there and discards records 3 to 5, although they are intact. For damage at the end of the file, which is what an interrupted write produces, this behaviour is correct. Damage in the middle of the file is a different matter: it means that the disk has lost data which was acknowledged, and a single node has no means of repairing such a loss. In the design of chapter 1, the record is recovered from the other replicas, and this is the first of several points at which a limitation of 1 node is the reason for something in the larger design.

Killing the server under load

The simulation exercises the log in isolation. The next test covers a complete node: the HTTP server, the store, the log and the disk. The command kvctl load writes keys in sequence and records each key in a local file once the server has answered with success.

Listing 2.33 The imports of kvctl/src/main.rs. kvctl/src/main.rs

use std::error::Error;
use std::fs::{self, File};
use std::io::{BufWriter, Write};
use std::path::PathBuf;

use clap::{Parser, Subcommand};
use ureq::Agent;

Listing 2.34 Writing keys, and recording each acknowledged key. kvctl/src/main.rs

/// Writes `count` keys. A key is recorded in `acked` only after the server
/// has answered with a success status.
fn load(agent: &Agent, server: &str, count: u32, acked: &PathBuf) -> Result<(), Box<dyn Error>> {
    let mut file = BufWriter::new(File::create(acked)?);
    let mut written = 0;

    for number in 1..=count {
        let key = format!("key-{number:06}");

        if let Err(error) = agent
            .put(format!("{server}/v1/keys/{key}"))
            .send(format!("value of {key}"))
        {
            println!("write {number} failed: {error}");

            break;
        }

        writeln!(file, "{key}")?;
        written += 1;
    }

    file.flush()?;
    println!("{written} writes acknowledged");

    Ok(())
}

Listing 2.35 Reading back every acknowledged key. kvctl/src/main.rs

/// Reads every acknowledged key and compares the value with the one that
/// was written.
fn verify(agent: &Agent, server: &str, acked: &PathBuf) -> Result<(), Box<dyn Error>> {
    let keys = fs::read_to_string(acked)?;
    let (mut found, mut missing) = (0, 0);

    for key in keys.lines() {
        let value = match agent
            .get(format!("{server}/v1/keys/{key}"))
            .call()
        {
            Ok(mut response) => response
                .body_mut()
                .read_to_string()?,
            Err(ureq::Error::StatusCode(404)) => String::new(),
            Err(error) => return Err(error.into()),
        };

        if value == format!("value of {key}") {
            found += 1;
        } else {
            missing += 1;
        }
    }

    println!(
        "{} acknowledged, {found} found, {missing} missing",
        found + missing
    );

    Ok(())
}

While the load is running, the server is terminated with signal 9, SIGKILL, which stops a process at once and allows it to complete nothing that is in progress.

$ ./target/release/kvd --data /var/tmp/kvd &
recovered 0 keys from 0 records, cut 0 bytes from the end of the log
listening on 127.0.0.1:7001
$ ./target/release/kvctl load --server http://127.0.0.1:7001 --count 5000 --acked acked.txt &
$ sleep 3; pkill -9 -x kvd
write 495 failed: io: Peer disconnected
494 writes acknowledged
$ ./target/release/kvd --data /var/tmp/kvd &
recovered 495 keys from 495 records, cut 0 bytes from the end of the log
listening on 127.0.0.1:7001
$ ./target/release/kvctl verify --server http://127.0.0.1:7001 --acked acked.txt
494 acknowledged, 494 found, 0 missing

Output	What it shows
494 writes acknowledged, 494 found, 0 missing	Every acknowledged write was recovered after the process was killed.
recovered 495 keys	The log contains 1 more write than the client had recorded.
cut 0 bytes	The process was not in the middle of appending a record when it was killed.

Write 495 is an instance of the third case in figure 2.2. Its record was stored durably, and the process was killed before any response was sent. From the client’s side the write failed, although the store contains it, and a client that needs a definite outcome sends its PUT again.

The limits of this test

The same test, run with --no-sync, also recovers every acknowledged write.

$ ./target/release/kvd --data /var/tmp/kvd --no-sync &
$ ./target/release/kvctl load --server http://127.0.0.1:7001 --count 200000 --acked acked.txt &
$ sleep 1.5; pkill -9 -x kvd
write 8595 failed: io: Connection reset by peer (os error 104)
8594 writes acknowledged
$ ./target/release/kvd --data /var/tmp/kvd --no-sync &
recovered 8594 keys from 8594 records, cut 0 bytes from the end of the log
$ ./target/release/kvctl verify --server http://127.0.0.1:7001 --acked acked.txt
8594 acknowledged, 8594 found, 0 missing

It would be a serious error to conclude from this that fsync is unnecessary. When a process is killed, its own memory is lost, but the data it had passed to write is in the kernel’s buffers. The kernel continues to run, and writes that data to the disk shortly afterwards. Only a failure that stops the kernel, such as a loss of power, loses data that was written without fsync. A test that kills a process therefore succeeds with and without fsync, and provides no evidence whatever about durability under a loss of power. Many systems have been declared durable on the strength of exactly this test.

Running the test on 2 machines

The server and the client are separate programs that communicate over HTTP, and can therefore run on different machines without any change. In the following session, the server runs on a machine named node-b at the address 10.0.0.2, and the client on node-a. A binary can be copied between machines of the same architecture and operating system, and is otherwise built on each.

The test on 2 machines. On node-a, kvctl load sends PUT requests and writes each acknowledged key to acked.txt. On node-b, kvd listens on port 7001 and writes kv.log. kvctl writes keys and notes each one the server answered. After 3 seconds kvd is killed with signal 9. kvd is started again and reads its log. kvctl reads every noted key, and 1107 of 1107 are there. — **Figure 2.7** The client and the server on separate machines.

node-a$ scp target/release/kvd node-b:/tmp/kvd
node-b$ /tmp/kvd --listen 0.0.0.0:7001 --data /tmp/kvd-data &
recovered 0 keys from 0 records, cut 0 bytes from the end of the log
listening on 0.0.0.0:7001

node-a$ ./target/release/kvctl load --server http://10.0.0.2:7001 --count 5000 --acked acked.txt &
node-b$ pkill -9 -x kvd
write 1108 failed: io: Connection reset by peer (os error 104)
1107 writes acknowledged

node-b$ /tmp/kvd --listen 0.0.0.0:7001 --data /tmp/kvd-data &
recovered 1107 keys from 1107 records, cut 0 bytes from the end of the log
node-a$ ./target/release/kvctl verify --server http://10.0.0.2:7001 --acked acked.txt
1107 acknowledged, 1107 found, 0 missing
node-a$ curl http://10.0.0.2:7001/healthz
{"status":"ok","keys":1107,"last_sequence":1107}

The client’s file of acknowledged keys is held on a machine that no fault affected, and is on that account an independent record of what the server acknowledged. Tests of faults follow this structure quite generally. A component outside the system under test records what was acknowledged, a fault is injected, and the record is compared with the state of the system after it has recovered. Chapters 3 and 4 test replication and elections in the same way.

The test can be run in several configurations.

Configuration	Server	Client
1 machine	`kvd --listen 127.0.0.1:7001`	`kvctl ... --server http://127.0.0.1:7001`
2 machines on a local network	`kvd --listen 0.0.0.0:7001` on machine B	`kvctl ... --server http://<address of B>:7001` on machine A
2 cloud instances	as above, with port 7001 permitted between the instances	as above, using the private address of the server

The limits of a single node

Limitation	Consequence	What chapter 1 provides against it
The log is on a single disk.	The loss of the disk loses all the data.	3 replicas of every shard
The node is a single process.	Every request fails while the process is not running.	the election of a new leader
Every write performs its own `fsync`.	The node stores 185 records per second on the disk measured.	group commit, and the distribution of writes over 24 nodes
The log grows without limit.	The time to restart and the disk space used increase continually.	periodic snapshots of the map, after which older records are deleted
All entries are held in memory.	The data cannot exceed the memory of 1 machine.	sharding

Within these limits, the node retains every acknowledged write when its process is killed, when its log is cut short at any byte, and when a bit is corrupted in its last record. That is the assumption on which chapter 1 was built, and chapter 3 now builds replication upon it.

The complete programs

Listing 2.36 The complete kv/src/lib.rs. kv/src/lib.rs

//! A key-value store for one node that keeps every acknowledged write across
//! a crash. A write is appended to a log on disk before it is applied in
//! memory, and the log is read back when the store is opened.

pub mod log;
pub mod record;
pub mod store;

pub use log::{Log, Recovery};
pub use record::{Change, Decoded, Record};
pub use store::{Condition, Entry, Store, StoreError, Summary, Write, Written, apply, check, page};

Listing 2.37 The complete kv/src/record.rs. kv/src/record.rs

//! The format of one record of the log.
//!
//! A record is a frame of 8 bytes followed by a payload:
//!
//! ```text
//! length: u32 | checksum: u32 | sequence: u64 | kind: u8 | key length: u16 | key | value
//! ```
//!
//! `length` counts the bytes of the payload and `checksum` is the CRC-32 of
//! the payload. All integers are little-endian.

use crc32fast::hash;

/// A change to the store.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Change {
    Put { key: String, value: Vec<u8> },
    Delete { key: String },
}

/// A change with its position in the log. Sequence numbers start at 1 and
/// grow by 1 with each record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Record {
    pub sequence: u64,
    pub change: Change,
}

/// The result of reading one record from the front of a slice of bytes.
#[derive(Debug, PartialEq, Eq)]
pub enum Decoded {
    /// A whole record, and the number of bytes it occupies.
    Record { record: Record, length: usize },
    /// The bytes end before the record does.
    Incomplete,
    /// The checksum does not match the payload, or the payload is malformed.
    Corrupt,
}

/// The bytes of a frame: the length and the checksum.
pub const FRAME: usize = 8;

/// The largest payload a record may have: a key, a value of 1 MiB, and the
/// fixed fields.
pub const MAX_PAYLOAD: usize = (1 << 20) + 1024;

const PUT: u8 = 1;
const DELETE: u8 = 2;

impl Record {
    /// Returns the record as the bytes that are appended to the log.
    pub fn encode(&self) -> Vec<u8> {
        let (kind, key, value) = match &self.change {
            Change::Put { key, value } => (PUT, key, value.as_slice()),
            Change::Delete { key } => (DELETE, key, &[][..]),
        };

        let mut payload = Vec::with_capacity(11 + key.len() + value.len());
        payload.extend_from_slice(&self.sequence.to_le_bytes());
        payload.push(kind);
        payload.extend_from_slice(&(key.len() as u16).to_le_bytes());
        payload.extend_from_slice(key.as_bytes());
        payload.extend_from_slice(value);

        let mut bytes = Vec::with_capacity(FRAME + payload.len());
        bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&hash(&payload).to_le_bytes());
        bytes.extend_from_slice(&payload);

        bytes
    }
}

impl Record {
    /// Reads the record at the front of `bytes`.
    pub fn decode(bytes: &[u8]) -> Decoded {
        let Some((length, rest)) = bytes.split_first_chunk::<4>() else {
            return Decoded::Incomplete;
        };
        let Some((checksum, rest)) = rest.split_first_chunk::<4>() else {
            return Decoded::Incomplete;
        };

        let length = u32::from_le_bytes(*length) as usize;

        if length > MAX_PAYLOAD {
            return Decoded::Corrupt;
        }

        let Some(payload) = rest.get(..length) else {
            return Decoded::Incomplete;
        };

        if hash(payload) != u32::from_le_bytes(*checksum) {
            return Decoded::Corrupt;
        }

        match parse(payload) {
            Some(record) => Decoded::Record {
                record,
                length: FRAME + length,
            },
            None => Decoded::Corrupt,
        }
    }
}

/// Splits a payload into its fields. It returns `None` for a payload that is
/// too short for its fields or has an unknown kind.
fn parse(payload: &[u8]) -> Option<Record> {
    let (sequence, rest) = payload.split_first_chunk::<8>()?;
    let (&kind, rest) = rest.split_first()?;
    let (key_length, rest) = rest.split_first_chunk::<2>()?;
    let (key, value) = rest.split_at_checked(u16::from_le_bytes(*key_length) as usize)?;
    let key = String::from_utf8(key.to_vec()).ok()?;

    let change = match (kind, value) {
        (PUT, value) => Change::Put {
            key,
            value: value.to_vec(),
        },
        (DELETE, []) => Change::Delete { key },
        (_unknown, _value) => return None,
    };

    Some(Record {
        sequence: u64::from_le_bytes(*sequence),
        change,
    })
}

Listing 2.38 The complete kv/src/log.rs. kv/src/log.rs

//! The write-ahead log: a file that only grows, to which every change is
//! appended before it is applied.

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

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

/// The name of the log inside the data directory.
pub const FILE_NAME: &str = "kv.log";

/// An open log, positioned at its end.
pub struct Log {
    file: File,
    /// Whether `append` waits for the disk before it returns.
    sync: bool,
    /// The sequence number of the next record.
    next: u64,
    /// Set by a failed write. A failed log accepts no more records.
    failed: bool,
}

/// What `Log::open` found in the file.
pub struct Recovery {
    /// The complete records, in the order they were written.
    pub records: Vec<Record>,
    /// The bytes after the last complete record, which were removed.
    pub discarded: u64,
}

/// Reads records from the front of `bytes` until one is incomplete or
/// corrupt. It returns the records and the number of bytes they occupy.
fn scan(bytes: &[u8]) -> (Vec<Record>, usize) {
    let mut records = Vec::new();
    let mut valid = 0;

    while let Decoded::Record { record, length } = Record::decode(&bytes[valid..]) {
        records.push(record);
        valid += length;
    }

    (records, valid)
}

impl Log {
    /// Opens the log in `directory`, creating both if they do not exist.
    /// Anything after the last complete record is cut off.
    pub fn open(directory: &Path, sync: bool) -> io::Result<(Log, Recovery)> {
        fs::create_dir_all(directory)?;

        let mut file = OpenOptions::new()
            .read(true)
            .append(true)
            .create(true)
            .open(directory.join(FILE_NAME))?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)?;

        let (records, valid) = scan(&bytes);
        let discarded = (bytes.len() - valid) as u64;

        if discarded > 0 {
            file.set_len(valid as u64)?;
        }

        if sync {
            // The file's length is part of the file. Its name is part of the directory.
            file.sync_all()?;
            File::open(directory)?.sync_all()?;
        }

        let next = records
            .last()
            .map_or(1, |record| record.sequence + 1);

        Ok((
            Log {
                file,
                sync,
                next,
                failed: false,
            },
            Recovery { records, discarded },
        ))
    }
}

impl Log {
    /// Appends a change and returns the record that holds it. When the log
    /// was opened with `sync`, the record is on the disk when this returns.
    pub fn append(&mut self, change: Change) -> io::Result<Record> {
        if self.failed {
            return Err(io::Error::other(
                "the log failed on an earlier write and is closed",
            ));
        }

        let record = Record {
            sequence: self.next,
            change,
        };

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

            return Err(error);
        }

        self.next += 1;

        Ok(record)
    }

    fn write(&mut self, bytes: &[u8]) -> io::Result<()> {
        self.file.write_all(bytes)?;

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

        Ok(())
    }

    /// The sequence number of the last record, or 0 for an empty log.
    pub fn last_sequence(&self) -> u64 {
        self.next - 1
    }
}

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(())
    }
}

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

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

        self.next = keep + 1;

        Ok(())
    }
}

Listing 2.39 The complete kv/src/store.rs. kv/src/store.rs

//! The store: the current value of every key in memory, and the log that
//! rebuilds it.

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

use thiserror::Error;

use crate::log::Log;
use crate::record::{Change, Record};

/// The value of a key and the sequence number of the record that wrote it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Entry {
    pub value: Vec<u8>,
    pub version: u64,
}

/// What a write requires of the key's current state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Condition {
    /// The write is unconditional.
    Always,
    /// The key must not exist.
    Absent,
    /// The key must exist at this version.
    Version(u64),
}

/// The result of a `put`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Written {
    pub version: u64,
    /// Whether the key did not exist before.
    pub created: bool,
}

/// One line of a listing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Summary {
    pub key: String,
    pub version: u64,
    pub size: usize,
}

#[derive(Debug, Error)]
pub enum StoreError {
    #[error("the key does not exist")]
    NotFound,
    #[error("the condition on the key's version does not hold")]
    ConditionFailed { current: Option<u64> },
    #[error("the log could not be written: {0}")]
    Log(#[from] io::Error),
}

pub struct Store {
    entries: BTreeMap<String, Entry>,
    log: Log,
}

impl Store {
    /// Opens the store in `directory` and applies every record of its log.
    /// It returns the store, the number of records applied, and the number
    /// of bytes that were cut from the end of the log.
    pub fn open(directory: &Path, sync: bool) -> io::Result<(Store, usize, u64)> {
        let (log, recovery) = Log::open(directory, sync)?;
        let applied = recovery.records.len();
        let mut entries = BTreeMap::new();

        for record in recovery.records {
            apply(&mut entries, record);
        }

        Ok((Store { entries, log }, applied, recovery.discarded))
    }
}

/// Makes the change of one record in the map. Recovery and the write path
/// both go through this function, so they cannot disagree.
pub fn apply(entries: &mut BTreeMap<String, Entry>, record: Record) {
    match record.change {
        Change::Put { key, value } => {
            entries.insert(
                key,
                Entry {
                    value,
                    version: record.sequence,
                },
            );
        }
        Change::Delete { key } => {
            entries.remove(&key);
        }
    }
}

impl Store {
    /// Sets the value of a key. The change is in the log before it is in
    /// the map.
    pub fn put(
        &mut self,
        key: String,
        value: Vec<u8>,
        condition: Condition,
    ) -> Result<Written, StoreError> {
        let current = self.version(&key);
        check(condition, current)?;

        let record = self
            .log
            .append(Change::Put { key, value })?;
        let version = record.sequence;
        apply(&mut self.entries, record);

        Ok(Written {
            version,
            created: current.is_none(),
        })
    }

    /// Removes a key.
    pub fn delete(&mut self, key: &str, condition: Condition) -> Result<(), StoreError> {
        let current = self.version(key);

        if current.is_none() {
            return Err(StoreError::NotFound);
        }

        check(condition, current)?;

        let record = self.log.append(Change::Delete {
            key: key.to_owned(),
        })?;
        apply(&mut self.entries, record);

        Ok(())
    }

    fn version(&self, key: &str) -> Option<u64> {
        self.entries
            .get(key)
            .map(|entry| entry.version)
    }
}

pub fn check(condition: Condition, current: Option<u64>) -> Result<(), StoreError> {
    let holds = match condition {
        Condition::Always => true,
        Condition::Absent => current.is_none(),
        Condition::Version(version) => current == Some(version),
    };

    holds
        .then_some(())
        .ok_or(StoreError::ConditionFailed { current })
}

/// One write of a batch: the change, and the condition it carries.
pub struct Write {
    pub change: Change,
    pub condition: Condition,
}

impl Store {
    /// Performs several writes with a single wait for the disk. Each write
    /// is checked as if the writes before it in the batch had already
    /// happened. The result of each is returned in the same order.
    pub fn commit(&mut self, writes: Vec<Write>) -> Vec<Result<Written, StoreError>> {
        // The versions that keys will have once the accepted writes of this
        // batch are applied. `None` is a key that a write of the batch deletes.
        let mut pending: BTreeMap<String, Option<u64>> = BTreeMap::new();
        let mut records = Vec::new();
        let mut results = Vec::new();
        let first = self.log.last_sequence() + 1;

        for write in writes {
            let key = match &write.change {
                Change::Put { key, .. } | Change::Delete { key } => key.clone(),
            };
            let current = match pending.get(&key) {
                Some(version) => *version,
                None => self.version(&key),
            };
            let deletes = matches!(write.change, Change::Delete { .. });

            let checked = if deletes && current.is_none() {
                Err(StoreError::NotFound)
            } else {
                check(write.condition, current)
            };

            if let Err(error) = checked {
                results.push(Err(error));

                continue;
            }

            let sequence = first + records.len() as u64;
            let version = if deletes { None } else { Some(sequence) };

            pending.insert(key, version);
            results.push(Ok(Written {
                version: sequence,
                created: current.is_none(),
            }));
            records.push(Record {
                sequence,
                change: write.change,
            });
        }

        // One write and one wait for the disk, for the whole batch.
        if let Err(error) = self.log.append_records(&records) {
            let failure = error.to_string();

            return results
                .into_iter()
                .map(|result| result.and(Err(StoreError::Log(io::Error::other(failure.clone())))))
                .collect();
        }

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

        results
    }
}

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

    /// Returns up to `limit` keys that start with `prefix` and sort after
    /// `after`, in order, and whether more keys follow them.
    pub fn list(&self, prefix: &str, after: Option<&str>, limit: usize) -> (Vec<Summary>, bool) {
        page(&self.entries, prefix, after, limit)
    }

    pub fn len(&self) -> usize {
        self.entries.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// The sequence number of the last record in the log.
    pub fn last_sequence(&self) -> u64 {
        self.log.last_sequence()
    }
}

/// One page of a listing over a map of entries.
pub fn page(
    entries: &BTreeMap<String, Entry>,
    prefix: &str,
    after: Option<&str>,
    limit: usize,
) -> (Vec<Summary>, bool) {
    let start = match after {
        Some(key) => Bound::Excluded(key),
        None => Bound::Included(prefix),
    };

    let mut matching = entries
        .range::<str, _>((start, Bound::Unbounded))
        .take_while(|entry| entry.0.starts_with(prefix))
        .map(|(key, entry)| Summary {
            key: key.clone(),
            version: entry.version,
            size: entry.value.len(),
        });

    let page: Vec<Summary> = matching
        .by_ref()
        .take(limit)
        .collect();
    let more = matching.next().is_some();

    (page, more)
}

Listing 2.40 The complete kvd/src/lib.rs. kvd/src/lib.rs

//! The HTTP API of a storage node.

pub mod api;

Listing 2.41 The complete kvd/src/api.rs. kvd/src/api.rs

//! The parts of the HTTP API that do not depend on how a node stores its
//! data: the errors and their responses, and the reading of keys and
//! conditional headers.

use axum::Json;
use axum::http::header::{CONTENT_TYPE, IF_MATCH, IF_NONE_MATCH};
use axum::http::{HeaderMap, HeaderName, StatusCode};
use axum::response::{IntoResponse, Response};
use kv::{Condition, StoreError};
use serde::Serialize;

/// The largest value a key may hold.
pub const MAX_VALUE: usize = 1 << 20;

pub const MAX_KEY: usize = 256;

/// The ways a request can fail. Each one is a status code and a sentence.
pub enum ApiError {
    InvalidKey,
    InvalidHeader(HeaderName),
    NotFound,
    ConditionFailed(Option<u64>),
    Unavailable(String),
    /// The request reached a node that does not serve it. The text names the reason.
    Misdirected(String),
    /// The shard has no leader at the moment. Nothing was changed.
    NoLeader,
}

/// The body of an error response, in the format of RFC 9457.
#[derive(Serialize)]
struct Problem {
    #[serde(rename = "type")]
    kind: &'static str,
    title: &'static str,
    status: u16,
    #[serde(skip_serializing_if = "String::is_empty")]
    detail: String,
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, kind, title, detail) = match self {
            ApiError::InvalidKey => (
                StatusCode::BAD_REQUEST,
                "/problems/invalid-key",
                "The key is not valid",
                format!("a key is 1 to {MAX_KEY} characters from A-Z, a-z, 0-9, '.', '_' and '-'"),
            ),
            ApiError::InvalidHeader(name) => (
                StatusCode::BAD_REQUEST,
                "/problems/invalid-header",
                "A request header is not valid",
                format!("{name} must be a version in double quotes, or * where that is allowed"),
            ),
            ApiError::NotFound => (
                StatusCode::NOT_FOUND,
                "/problems/key-not-found",
                "The key does not exist",
                String::new(),
            ),
            ApiError::ConditionFailed(current) => (
                StatusCode::PRECONDITION_FAILED,
                "/problems/condition-failed",
                "The condition on the key's version does not hold",
                current.map_or("the key does not exist".into(), |version| {
                    format!("the key is at version {version}")
                }),
            ),
            ApiError::Misdirected(detail) => (
                StatusCode::MISDIRECTED_REQUEST,
                "/problems/not-leader",
                "This node is not the leader",
                detail,
            ),
            ApiError::NoLeader => (
                StatusCode::SERVICE_UNAVAILABLE,
                "/problems/no-leader",
                "The shard has no leader at the moment",
                String::new(),
            ),
            ApiError::Unavailable(detail) => (
                StatusCode::SERVICE_UNAVAILABLE,
                "/problems/store-unavailable",
                "The node cannot serve the request",
                detail,
            ),
        };

        let problem = Problem {
            kind,
            title,
            status: status.as_u16(),
            detail,
        };

        (
            status,
            [(CONTENT_TYPE, "application/problem+json")],
            Json(problem),
        )
            .into_response()
    }
}

impl From<StoreError> for ApiError {
    fn from(error: StoreError) -> Self {
        match error {
            StoreError::NotFound => ApiError::NotFound,
            StoreError::ConditionFailed { current } => ApiError::ConditionFailed(current),
            StoreError::Log(error) => ApiError::Unavailable(error.to_string()),
        }
    }
}

/// A version as it appears in `ETag`, `If-Match` and `If-None-Match`.
pub fn etag(version: u64) -> String {
    format!("\"{version}\"")
}

pub fn valid_key(key: &str) -> Result<(), ApiError> {
    let allowed = |c: char| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-');

    let valid = (1..=MAX_KEY).contains(&key.len()) && key.chars().all(allowed);

    valid
        .then_some(())
        .ok_or(ApiError::InvalidKey)
}

/// Reads a header that holds a quoted version or `*`.
pub fn version_header(
    headers: &HeaderMap,
    name: HeaderName,
) -> Result<Option<Condition>, ApiError> {
    let Some(value) = headers.get(&name) else {
        return Ok(None);
    };

    let Ok(text) = value.to_str() else {
        return Err(ApiError::InvalidHeader(name));
    };

    if text == "*" {
        return Ok(Some(Condition::Absent));
    }

    let version = text
        .strip_prefix("\"")
        .and_then(|rest| rest.strip_suffix("\""))
        .and_then(|digits| digits.parse().ok());

    version
        .map(Condition::Version)
        .map(Some)
        .ok_or(ApiError::InvalidHeader(name))
}

/// The condition a write carries: `If-Match: "7"` requires version 7, and
/// `If-None-Match: *` requires that the key does not exist.
pub fn write_condition(headers: &HeaderMap) -> Result<Condition, ApiError> {
    match (
        version_header(headers, IF_MATCH)?,
        version_header(headers, IF_NONE_MATCH)?,
    ) {
        (Some(Condition::Version(version)), None) => Ok(Condition::Version(version)),
        (None, Some(Condition::Absent)) => Ok(Condition::Absent),
        (None, None) => Ok(Condition::Always),
        (Some(..), ..) => Err(ApiError::InvalidHeader(IF_MATCH)),
        (None, Some(..)) => Err(ApiError::InvalidHeader(IF_NONE_MATCH)),
    }
}

Listing 2.42 The complete kvd/src/main.rs. kvd/src/main.rs

//! Command kvd serves a key-value store over HTTP. Every write is in the log
//! on disk before the response is sent.
//!
//! Run it from code/ with:
//!
//!     cargo run -p kvd -- --listen 0.0.0.0:7001 --data /var/tmp/kvd

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;
use kv::{Condition, Store, StoreError, Summary};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, version_header, write_condition};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use tokio::task;

#[derive(Parser)]
struct Options {
    /// The address to accept connections on.
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    /// The directory that holds the log.
    #[arg(long, default_value = "data")]
    data: PathBuf,

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

/// The store, shared by every request.
type Shared = Arc<Mutex<Store>>;


/// Runs a write on a thread that may block. A write waits for the disk, and
/// a thread that waits must not be one of the threads that serve requests.
async fn write<T, F>(store: Shared, work: F) -> Result<T, ApiError>
where
    F: FnOnce(&mut Store) -> Result<T, StoreError> + Send + 'static,
    T: Send + 'static,
{
    let result = task::spawn_blocking(move || work(&mut store.lock()))
        .await
        .map_err(|error| ApiError::Unavailable(error.to_string()))?;

    Ok(result?)
}

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

    let written = write(store, move |store| store.put(key, 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(store): State<Shared>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;
    let condition = write_condition(&headers)?;

    write(store, move |store| store.delete(&key, condition)).await?;

    Ok(StatusCode::NO_CONTENT)
}

async fn get_key(
    State(store): State<Shared>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    valid_key(&key)?;
    let cached = version_header(&headers, IF_NONE_MATCH)?;
    let entry = store
        .lock()
        .get(&key)
        .cloned()
        .ok_or(ApiError::NotFound)?;
    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>,
    /// The value of `after` that returns the next page, when there is one.
    next_after: Option<String>,
}

async fn list_keys(State(store): State<Shared>, Query(query): Query<ListQuery>) -> Json<Page> {
    let limit = query
        .limit
        .unwrap_or(100)
        .clamp(1, 1000);
    let (page, more) = store
        .lock()
        .list(&query.prefix, query.after.as_deref(), limit);
    let next_after = page
        .last()
        .filter(|_last| more)
        .map(|last| last.key.clone());
    let keys = page
        .into_iter()
        .map(|Summary { key, version, size }| Item { key, version, size })
        .collect();

    Json(Page { keys, next_after })
}

#[derive(Serialize)]
struct Health {
    status: &'static str,
    keys: usize,
    last_sequence: u64,
}

async fn health(State(store): State<Shared>) -> Json<Health> {
    let store = store.lock();

    Json(Health {
        status: "ok",
        keys: store.len(),
        last_sequence: store.last_sequence(),
    })
}

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

    println!(
        "recovered {} keys from {applied} records, cut {discarded} bytes from the end of the log",
        store.len()
    );

    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(Arc::new(Mutex::new(store)));

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

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

    Ok(())
}

Listing 2.43 The complete kvctl/src/main.rs. kvctl/src/main.rs

//! Command kvctl tests a kvd server from another process or another machine.
//! `load` writes keys and records each write the server acknowledged, and
//! `verify` reads every recorded key back.
//!
//! Run it from code/ with:
//!
//!     cargo run -p kvctl -- load --server http://127.0.0.1:7001 --count 1000 --acked acked.txt
//!     cargo run -p kvctl -- verify --server http://127.0.0.1:7001 --acked acked.txt

use std::error::Error;
use std::fs::{self, File};
use std::io::{BufWriter, Write};
use std::path::PathBuf;

use clap::{Parser, Subcommand};
use ureq::Agent;

#[derive(Parser)]
struct Options {
    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Write keys until the count is reached or a write fails.
    Load {
        #[arg(long)]
        server: String,
        #[arg(long, default_value_t = 1000)]
        count: u32,
        /// The file that receives the key of every acknowledged write.
        #[arg(long)]
        acked: PathBuf,
    },
    /// Read back every key in the file of acknowledged writes.
    Verify {
        #[arg(long)]
        server: String,
        #[arg(long)]
        acked: PathBuf,
    },
}

/// Writes `count` keys. A key is recorded in `acked` only after the server
/// has answered with a success status.
fn load(agent: &Agent, server: &str, count: u32, acked: &PathBuf) -> Result<(), Box<dyn Error>> {
    let mut file = BufWriter::new(File::create(acked)?);
    let mut written = 0;

    for number in 1..=count {
        let key = format!("key-{number:06}");

        if let Err(error) = agent
            .put(format!("{server}/v1/keys/{key}"))
            .send(format!("value of {key}"))
        {
            println!("write {number} failed: {error}");

            break;
        }

        writeln!(file, "{key}")?;
        written += 1;
    }

    file.flush()?;
    println!("{written} writes acknowledged");

    Ok(())
}

/// Reads every acknowledged key and compares the value with the one that
/// was written.
fn verify(agent: &Agent, server: &str, acked: &PathBuf) -> Result<(), Box<dyn Error>> {
    let keys = fs::read_to_string(acked)?;
    let (mut found, mut missing) = (0, 0);

    for key in keys.lines() {
        let value = match agent
            .get(format!("{server}/v1/keys/{key}"))
            .call()
        {
            Ok(mut response) => response
                .body_mut()
                .read_to_string()?,
            Err(ureq::Error::StatusCode(404)) => String::new(),
            Err(error) => return Err(error.into()),
        };

        if value == format!("value of {key}") {
            found += 1;
        } else {
            missing += 1;
        }
    }

    println!(
        "{} acknowledged, {found} found, {missing} missing",
        found + missing
    );

    Ok(())
}

fn main() -> Result<(), Box<dyn Error>> {
    let agent = Agent::new_with_defaults();

    match Options::parse().command {
        Command::Load {
            server,
            count,
            acked,
        } => load(&agent, &server, count, &acked),
        Command::Verify { server, acked } => verify(&agent, &server, &acked),
    }
}

Listing 2.44 The complete crashsim/src/main.rs. crashsim/src/main.rs

//! Command crashsim checks what the log recovers after a crash in the middle
//! of a write. It writes a log of 5 records, then damages a copy of it in
//! every way a crash or a bad disk can: it cuts the copy at every length,
//! and it flips 1 bit in every byte. After each, it opens the copy and
//! compares the records that come back with the records that were complete.
//!
//! Run it from code/ with:
//!
//!     cargo run -p crashsim

use std::env;
use std::error::Error;
use std::fs;
use std::path::Path;
use std::process;

use kv::log::FILE_NAME;
use kv::{Change, Log, Record};

/// The changes of the test log.
fn changes() -> Vec<Change> {
    let put = |key: &str, value: &str| Change::Put {
        key: key.into(),
        value: value.into(),
    };

    vec![
        put("user-1", "ada"),
        put("user-2", "grace"),
        put("user-1", "ada lovelace"),
        Change::Delete {
            key: "user-2".into(),
        },
        put("user-3", "barbara"),
    ]
}

/// Writes the changes to a log in `directory`. It returns the bytes of the
/// log and the offset at which each record ends.
fn build(directory: &Path) -> Result<(Vec<u8>, Vec<usize>), Box<dyn Error>> {
    let mut log = Log::open(directory, false)?.0;
    let mut ends = Vec::new();

    for change in changes() {
        let record = log.append(change)?;
        ends.push(ends.last().copied().unwrap_or(0) + record.encode().len());
    }

    Ok((fs::read(directory.join(FILE_NAME))?, ends))
}

/// Replaces the log in `directory` with `bytes` and opens it.
fn recover(directory: &Path, bytes: &[u8]) -> Result<Vec<Record>, Box<dyn Error>> {
    fs::write(directory.join(FILE_NAME), bytes)?;
    let recovery = Log::open(directory, false)?.1;

    Ok(recovery.records)
}

/// Whether `records` are exactly the first `count` changes, in order.
fn is_prefix(records: &[Record], count: usize) -> bool {
    let expected = changes();

    records.len() == count
        && records
            .iter()
            .zip(&expected)
            .enumerate()
            .all(|(index, (record, change))| {
                record.sequence == index as u64 + 1 && &record.change == change
            })
}

/// Cuts the log at every length and checks that recovery returns the records
/// that end at or before the cut.
fn cuts(directory: &Path, log: &[u8], ends: &[usize]) -> Result<(), Box<dyn Error>> {
    for length in 0..=log.len() {
        let complete = ends
            .iter()
            .filter(|&&end| end <= length)
            .count();
        let records = recover(directory, &log[..length])?;

        if !is_prefix(&records, complete) {
            return Err(format!(
                "cut at {length}: expected {complete} records, recovered {}",
                records.len()
            )
            .into());
        }
    }

    println!(
        "cut at every length from 0 to {}: each recovery returned the complete records",
        log.len()
    );

    let mut first = 0;

    for (count, &end) in ends.iter().enumerate() {
        println!("  lengths {first:3} to {:3}: {count} records", end - 1);
        first = end;
    }

    println!("  length  {first:3}       : {} records", ends.len());

    Ok(())
}

/// Flips 1 bit in each byte in turn and checks that recovery returns the
/// records before the damaged one.
fn flips(directory: &Path, log: &[u8], ends: &[usize]) -> Result<(), Box<dyn Error>> {
    for position in 0..log.len() {
        let mut damaged = log.to_vec();
        damaged[position] ^= 0x10;

        let before = ends
            .iter()
            .filter(|&&end| end <= position)
            .count();
        let records = recover(directory, &damaged)?;

        if !is_prefix(&records, before) {
            return Err(format!(
                "flip at {position}: expected {before} records, recovered {}",
                records.len()
            )
            .into());
        }
    }

    println!(
        "1 bit flipped in each of {} bytes: each recovery returned the records before the damage",
        log.len()
    );

    Ok(())
}

fn run(directory: &Path) -> Result<(), Box<dyn Error>> {
    let (log, ends) = build(directory)?;
    let boundaries: Vec<String> = ends
        .iter()
        .map(usize::to_string)
        .collect();

    println!(
        "a log of {} records and {} bytes, with records ending at {}",
        ends.len(),
        log.len(),
        boundaries.join(", ")
    );

    cuts(directory, &log, &ends)?;
    flips(directory, &log, &ends)?;

    let records = recover(directory, &log[..ends[1] + 5])?;
    let mut reopened = Log::open(directory, false)?.0;
    let next = reopened.append(Change::Delete {
        key: "user-1".into(),
    })?;

    println!(
        "after a cut at {}: {} records recovered, and the next record has sequence {}",
        ends[1] + 5,
        records.len(),
        next.sequence
    );

    Ok(())
}

fn main() -> Result<(), Box<dyn Error>> {
    let directory = env::temp_dir().join(format!("crashsim-{}", process::id()));
    let result = run(&directory);

    fs::remove_dir_all(&directory)?;

    result
}

Listing 2.45 The complete synccost/src/main.rs. synccost/src/main.rs

//! Command synccost measures what it costs to wait for the disk. It appends
//! the same records to 2 logs, one that calls fsync after each record and
//! one that does not, and prints the time per record.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p synccost -- /var/tmp/synccost

use std::env;
use std::error::Error;
use std::fs;
use std::path::Path;
use std::time::{Duration, Instant};

use kv::{Change, Log};

const RECORDS: u32 = 500;

/// Appends `RECORDS` records of 100 bytes and returns the time it took.
fn measure(directory: &Path, sync: bool) -> Result<Duration, Box<dyn Error>> {
    let mut log = Log::open(directory, sync)?.0;
    let started = Instant::now();

    for number in 0..RECORDS {
        log.append(Change::Put {
            key: format!("key-{number:06}"),
            value: vec![b'v'; 100],
        })?;
    }

    Ok(started.elapsed())
}

fn main() -> Result<(), Box<dyn Error>> {
    let root = env::args()
        .nth(1)
        .ok_or("usage: synccost <directory>")?;

    for (name, sync) in [("without fsync", false), ("with fsync", true)] {
        let directory = Path::new(&root).join(name.replace(' ', "-"));
        let elapsed = measure(&directory, sync)?;
        let each = elapsed.as_secs_f64() * 1000.0 / f64::from(RECORDS);

        println!(
            "{name:13}: {RECORDS} records in {:8.1} ms, {each:.3} ms per record, {:6.0} records per second",
            elapsed.as_secs_f64() * 1000.0,
            1000.0 / each
        );

        fs::remove_dir_all(&directory)?;
    }

    Ok(())
}

Summary

A storage node keeps its entries in memory for reads and a log on disk for durability, and every write is appended to the log and confirmed by the disk before it is applied in memory.
With a log that is only appended to, a crash can damage nothing but the last record, whereas an update in place can destroy data that was already stored.
Every record carries its length and a CRC-32 checksum, by which a reader distinguishes a complete record from one that is incomplete or corrupt.
Recovery decodes records until one fails, keeps all that precede it, and cuts the file off before anything new is appended.
write hands data to the kernel, and fsync waits until the disk has stored it. A durable append took 5.4 ms, which is 185 records per second.
The directory entry of a new file becomes durable only when the directory is synchronised.
After a failed write or fsync, the log accepts no further records, since a repeated fsync can report success for data that is not on the disk.
A write can be durable without having been acknowledged, when the node crashes between the disk’s confirmation and the response, and the client then repeats its request.
Killing a process does not test durability under a loss of power, because the kernel keeps what the process had written.
A test of faults records what was acknowledged outside the system under test, and compares that record with the state of the system afterwards.

Distributed Systems from the Ground Up