Getting started with Raft using miniraft
What is about to happen?
There is a lot of very good material publicly available on Raft. In fact, the whole purpose of the original Raft paper was to introduce a digestable consensus protocol. I am very unlikely to add any unique perspective that would improve conceptual insights. Thus, for deep dives, refer to the resource section for pointers. Instead, I will use miniraft, a (toy-ish but correct) Raft implementation from Alex Povel using the Rust language to dig into Raft's concepts and building blocks. The idea for the following minutes is to document how a client request travels through Raft to make reading the code and related Raft concepts easier. After reading this, you should be able to dig into the very details of Raft, while obstacles like working through general code architecture are already taken away from you.
miniraft is built around maelstrom, a framework to test distributed algorithms. However, the core is essentially implemented protocol agnostic. I forked it and implemented a tcp/http communication layer for two reasons. First, to work with the code and ensure I understand what it is doing and second, it might help for educational reasons, since most likely more people have a better understanding of HTTP APIs than maelstrom.
What is Raft?
Raft is a consensus algorithm to manage a replicated log most commonly used to implement a replicated state
machine.
If you have no prior context on distributed systems "yeah, thanks for nothing" is the proper response. In that
case, bear with me for a second and I will try to shed some light. Otherwise, you might want skip to the
miniraft specifics starting from the next section. Let's define everything and put it together.
Theoretically, a state-machine is some abstract entity that can be in a finite number of states (assume a, b and c). Given some inputs, it can transition between those states. If the machine is in state a and receives a certain input (usually called a Command), it can move from a to b or c. To make it specific, think HashMap instead of state-machine (other things can represent state-machines as long as they satisfy the theoretical conditions). A HashMap can be empty or contain 1..N key value pairs. Each realization of that HashMap is a state. So loosely speaking, the HashMap is the machine and its realization is a state. Hence, state-machine.
Commands that trigger methods like insert(), remove() are the inputs that transition the HashMap (machine) from one state into the other by adding or removing key-value pairs accordingly. A log is a list of sequential commands that transition the machine from one state into the other. Suppose your HashMap contains critical data that needs to be constantly available. You keep it on one node and that node crashes. Bad. Replication means that both the log of transition commands and the state machine live on multiple physical servers (nodes). The idea here is to not be exposed to catastrophic loss. If you keep the HashMap on multiple nodes instead, you can easily redirect clients to another node in case one or more fail to respond.
At this point, you might be ahead and ask how are the HashMaps kept in sync (in the same state) when they are copied on multiple nodes? This is the essence of Raft. The algorithm ensures that the state machine traverses state changes in the exact same order at all nodes. It does that by replicating the exact same log of transition commands across nodes. The commands from that log are then applied in order to the machine (HashMap) by the server. Making sure the log is consistent on all nodes is why consensus is required. The nodes need to communicate and consent on what to do in order to guarantee that all state machines end up in the exact same state while withstanding server crashes. We will see how that is done in miniraft in a bit.
Let's take the sentence from above and rephrase a bit. This time it should be easier to read. Raft is an algorithm that provides a correct protocol of how multiple nodes consent on what steps to perform in order to replicate a log of transition commands to implement a state machine that is in the exact same state across all nodes. The concrete implementation of that state machine as a HashMap will accompany us along the way through miniraft.
What is consensus?
Consensus is a center piece of distributed algorithms. In a system of multiple servers that achieves one common goal while being resistant to some servers crashing out at any time, coordination of the servers becomes a main concern. Why? Let's look at an example.
Consider your account balance is replicated in some data base. You do not want to have zero savings, because some server in a data center died, which makes sense. So your account lives in databases across multiple servers and across multiple data centers. One day, you receive a generous gift of 100k monies that is to be added to your account. Full of excitement, you get ahead of yourself and purchase a 1999 Base Set 1st Edition Holo Chansey (PSA 10) which puts you down 50k monies if you can get one. So while the purchase transaction went to your account balance (of course it did), some servers crash during the transaction of the gift. They restart shortly after and reboot their last state. But now some of the servers that replicate your account balance have a positive 50k (100k - 50k) and some have a negative 50k because they did not see the gift. Which one is the correct one? You certainly have only one balance, not two, and first and foremost not a negative one!
Raft's idea is that at any point in time a majority of cluster members need to consent and implement one and the same state. This guarantees that as long as a majority of cluster nodes are alive, that (correct) state can always be recovered. One way to achieve consensus in a cluster is to pick one node as a leader. This is also what Raft does. The leader has complete responsibility managing the replicated log: accepts requests from clients, replicates them on other servers, decides when servers can apply log entries to state machines. To update the account balance, a leader therefore first adds the updating command received from a client to its own log. Then informs all other nodes (the followers) to also update their logs in the exact same way as the leader did and only then are the update commands in the log applied and the actual account balance is updated.
Actually, not quite. There is a safety hurdle to take. A command that is safe to apply to the account balance is called committed. A command is committed to the log, if the leader applied it to its log, sent it to all follower nodes and a majority of followers report back that they appended the command to their own log. The system needs to differentiate between committed and uncommitted log entries because only committed entries are guaranteed to survive future leader changes. Since committed log entries live on a majority of cluster nodes a new leader is able to determine and agree with a majority of other nodes what the last log entry is that the majority agrees upon. Therefore, the true log is decided by the majority of cluster nodes to make the log and consequently the state machine fault-tolerant.
As pointed out already, leader might disconnect and trigger a search for a new leader for the cluster. This is called leader election. The period for which one specific node is the leader is called term, just like in real world politics. This term is represented by an integer, that increases every time a new election cycle is triggered. As long as a leader exists and is available, that leader sends heartbeat messages to the other nodes (followers) saying "I am still here". Once a follower does not receive any commands from the leader, nor the heartbeat, it assumes that the leader got lost and that the cluster is in need of a new one. It increases its term since a new election cycle started, transitions in candidate mode proposing itself as a new leader to the cluster and requesting all other nodes to vote for it. A candidate wins the election if it receives a majority of votes from all other nodes in the cluster. This is the essence. For a more detailed discussion including some edge cases have a look at ยง5.2 of the original paper.
How a client request enters miniraft
We will now turn our attention to the question of how a client request makes its way "into Raft". Since the leader is supposed to answer client requests, the question is how we reach a leader. In miniraft, we simply hit any node in the cluster. If we hit the leader, we are lucky, if we hit a follower, the request is forwarded to the leader. We know how, because each node has a state representation, which also has information about who the current leader is. Let's look at how this works under the assumption that we want to insert a new record into our state machine (a HashMap). In miniraft, a write request ClientMessage is defined as
enum ClientMessage<K, V> {
WriteRequest {
key: K,
value: V,
id: MessageID,
},
}
Regardless of how write requests arrive at the server, a client request needs to be cast into this type. Maelstrom for instance reads (deserializes) and writes (serializes) these messages as JSON to stdout. With the http module, we can also send HTTP requests via TCP.
curl -i -X PUT --data-binary 'some value to write' http://127.0.0.1:7878/key/some-new-key
curl -i -X PUT --data-binary 'some value to write' http://127.0.0.1:7879/key/some-new-key
curl -i -X PUT --data-binary 'some value to write' http://127.0.0.1:7880/key/some-new-key
PUT /key/some-new-key HTTP/1.1\r\n
Host: 127.0.0.1:7878\r\n
Content-Length: 18\r\n
\r\n
some data to write
From this, the ClientMessage is derived. The PUT signals that the client wants to write. From the REST convention we know that the key is to be found in the path after the qualifier /key/ and the value to write is to be found after the headers with a Content-Length of 18 bytes. That's all we need. The HTTP module reads the data from the stream, tries to parse it into a Request type and the ClientMessage, then implements the TryFrom trait in order to cast a Request into a ClientMessage. The exact same strategy applies for read and update (CAS) requests. There is one final thing to add, before we can continue and discuss how a request is handled now that it landed at a Raft node and has miniraft's required representation.
We need to distinguish between two kinds of messages. One we have already seen: ClientMessages, which can be requests or responses. The other are internal RaftMessages, which represent the communication activities to coordinate between nodes that are not of any user/ client concern. However, both share the same communication layer. Using maelstrom, they are both exchanged via stdout and using HTTP, both travel via TCP. Remember, some messages also cross the node/ process boundary. Requests should be handled by the leader, so those that are send to followers need to be forwarded by the follower to the leader. Hence, miniraft defines a type called MessageEnvelope
struct MessageEnvelope<B> {
source: NodeID,
destination: NodeID,
body: B,
}
miniraft's Engine architecture
We will now turn to the specifics of Raft by looking at how miniraft handles a client request like the write request from above. To do that, it's worthwhile to look a little bit into the architecture of the miniraft server. Recall, that each node in Raft is simulated using a CPU process. Each of these processes runs something that is called the Raft engine.
struct Engine<S: StateMachine> {
state: Arc<Mutex<State<S>>>,
state_machine: S,
id: NodeID
}The state is essentially the log with some meta data that provides context on some state.
Note, Log
is a field in the Common
type.
The State is defined as
struct State<S: StateMachine> {
c: Common<S::Command>,
r: Role,
}
A state machine can be any type that implements the StateMachine trait.
trait StateMachine: Default + Send + std::fmt::Debug {
type Command: Debug + Clone + Dismiss;
fn apply(&mut self, cmd: Self::Command);
}
impl<K, V> StateMachine for HashMap<K, V>
where
K: Eq + std::hash::Hash,
K: Send + Debug + Clone,
V: Send + Debug + Clone + PartialEq,
{
type Command = Command<K, V>;
fn apply(&mut self, cmd: Self::Command) {
let id = cmd.response_id.unwrap_or_default();
let in_reply_to = cmd.in_reply_to.unwrap_or_default();
let resp = match cmd.inner {
WireCommand::Read { key } => {
if let Some(v) = self.get(&key) {
rpc::ClientMessage::ReadResponse {
in_reply_to,
value: v.clone(),
id,
}
} else {
rpc::ClientMessage::ErrorResponse {
in_reply_to,
id,
code: ReservedErrorCode::KeyDoesNotExist.into(),
text: "no such key".into(),
}
}
}
WireCommand::Write { key, value } => {
self.insert(key, value);
rpc::ClientMessage::WriteResponse { in_reply_to, id }
}
WireCommand::CAS { key, from, to } => match self.get_mut(&key) {
Some(v) if *v == from => {
*v = to;
rpc::ClientMessage::CASResponse { in_reply_to, id }
}
Some(v) => {
rpc::ClientMessage::ErrorResponse {
in_reply_to,
id,
code: ReservedErrorCode::PreconditionFailed.into(),
text: format!("values for key differ (found {:?})", v),
}
}
None => {
rpc::ClientMessage::ErrorResponse {
in_reply_to,
id,
code: ReservedErrorCode::KeyDoesNotExist.into(),
text: "no such key".into(),
}
}
},
};
// not relevant right now ...
}
}
apply() requires a Command type, based on which it either get()s, insert()s or get_mut()s a key-value pair. The inner WireCommand enum variant contained in the Command type determines which action is triggered. However, since the StateMachine trait defines a generic Command type, one could implement other Command types on top of other data stores. This illustrates how miniraft's core engine is agnostic to the conrecte type. In fact, everything in state.rs is pure Raft consensus and standalone.
Ultimately, the state machine of the Engine is some data store that implements the StateMachine trait that applies transitions based on some Command type. For that reason the state: Arc<Mutex<State<S>>> of the Engine is also bound to a type that implements the StateMachine trait. The State represents the log where log entries are StateMachine Commands. Recall from before: Raft is a mechanism to consent on how to replicate a log of transition commands (state) in order to apply those to a replicated state machine. The Engine therefore encodes the very core building blocks of the Raft algorithm.
The State also implements all Raft logic. For details, read anything from the list of resources below or look at the concrete miniraft implementation in state.rs. Instead of going through many methods and what they do, let's get back to our request and our journey through Raft.
Run the application with
NODES=7878,7879,7880 cargo run --bin main-http
let raft: Engine<HashMap<u64, i64>> =
Engine::new(this_node.clone(), peers.clone(), persistent_state);
raft.start(
raft_incoming_rx,
raft_outgoing_tx,
client_incoming_rx,
client_outgoing_tx,
persistence_file,
NodeMessageIDGenerator,
metrics_addr,
);
let correlation_map: Arc<Mutex<HashMap<String, TcpStream>>> =
Arc::new(Mutex::new(HashMap::new()));
let listener = TcpListener::bind(("127.0.0.1", port))?;
for stream in listener.incoming() {
let stream = stream.expect("should have a stream");
let this_node = this_node.clone();
let mut correlation_map = Arc::clone(&correlation_map);
let client_incoming_tx = client_incoming_tx.clone();
let raft_incoming_tx = raft_incoming_tx.clone();
thread::spawn(move || {
handle_connection(
this_node,
stream,
&mut correlation_map,
client_incoming_tx,
raft_incoming_tx,
)
});
}
The first thing to note is the correlation_map. The node receives requests via TCP. However, due to the distributed and asynchronous structure in which a system with multiple (here three) nodes, responses can potentially be realized on other nodes. We do not want to wait and do nothing (block) until a response is generated and travels back to the originating node. Instead, TcpStream's that wait for reponses are cached in a HashMap by encoding a key which consits of the node id and message id. Once a message arrives on TCP, it is inspected, whether it's a response in which case the waiting TcpStream is removed from the cache and the response put on that stream. Note, that the correlation_map is thread safe, since for each incoming message a new thread is spawned handling that request. Thus, the map lives on the child-process but is shared with each thread that handles a connection. To write or remove from the HashMap the thread first needs to acquire the lock.
Speaking of threads: Handling multiple connections is not the only place where threads are spawned. On start() of the Raft engine, six more threads are spawned. Four of these are Raft specific and two are used to collect and serve statistics.
pub fn start<K, V>(
self,
raft_rx: PeerReceiver<rpc::RaftMessage<S::Command>>,
raft_tx: PeerSender<rpc::RaftMessage<S::Command>>,
client_rx: PeerReceiver<rpc::ClientMessage<K, V>>,
client_tx: PeerSender<rpc::ClientMessage<K, V>>,
persist: impl Write + Seek + Send + 'static,
message_ids: impl Iterator<Item = NodeMessageID> + Send + 'static,
metrics_addr: impl ToSocketAddrs + Send + 'static,
) where
K: Send + 'static + Debug,
V: Send + 'static + Debug,
S::Command: Clone,
S::Command: From<(
NodeID,
NodeMessageID,
rpc::ClientMessage<K, V>,
PeerSender<ClientMessage<K, V>>,
)>,
{
launch_named_background_task("raft-election-loop", {
let state = Arc::clone(&self.state);
let raft_tx = raft_tx.clone();
move || Self::election_loop(raft_tx, state)
});
launch_named_background_task("raft-log-replication-loop", {
let state = Arc::clone(&self.state);
let raft_tx = raft_tx.clone();
move || Self::log_replication_loop(raft_tx, state)
});
launch_named_background_task("raft-handle-incoming-raft-messages", {
let state = Arc::clone(&self.state);
let raft_tx = raft_tx.clone();
// This task is also responsible for driving the state machine forward.
let mut machine = self.state_machine; // move out + make mutable
move || Self::incoming_raft_rpcs_loop(raft_rx, raft_tx, state, &mut machine, persist)
});
launch_named_background_task("raft-handle-incoming-client-messages", {
let state = Arc::clone(&self.state);
let client_tx = client_tx.clone();
let raft_tx = raft_tx.clone();
let node_id = self.id.clone();
move || {
Self::incoming_client_rpcs_loop(
node_id,
client_rx,
client_tx,
raft_tx,
state,
message_ids,
)
}
});
// metrics threads omitted here
}
From this, starting the Engine that represents the state and the state_machine means starting
background tasks in separate threads. The first task is responsible for the leader election process, the second
for replication, the third to handle incoming Raft messages and the fourth for handling incoming client
messages. So far so obvious. Each of these threads gets a thread safe clone of the Engine's state
ensuring that if either of the threads should change the state or state_machine that happens only if
that thread has the single right to do so and no other thread can apply changes at the same time.
Additionally, sender and receiver ends of multiple producer single consumer (mpsc)
channels are moved into the threads.
Where TCP facilitates communication between nodes (here processes), channels enable communication between
threads. For each of the four Raft threads spawned in the start() method, one channel exists to build a
communication bridge to another thread. Producing ends can be copied and moved into multiple threads (multiple
producer paradigm) but receiving ends must be owned by one single thread (single consumer).
What is the advantage of having multiple threads and communication channels between them?
The reason purely boils down to efficiency concerns. It has nothing to do with Raft algorithm and the
correctness of the consensus protocol. Instead, it is a way to make progress on different tasks that do not
depend on each other in parallel. For instance, handling incoming requests from a client should not block an
ongoing election process running in the backgroud. However, there are also hand-over points, where one task's
output is another tasks input. Consider the thread handling incoming client requests that casts the
ClientMessage into a RaftMessage for further processing. It sends it the latter from the current
thread to the one that handles incoming RaftMessages. At these points, thread boundaries are
crossed using channels.
To summarize the general architecture of miniraft: On progam startup, multiple nodes are simulated using one process each. These nodes listen for client requests and cross-node communication on specific ports for incoming TCP streams. Each node starts the Raft Engine which represents the state (log) and the state machine of that node. The startup spawns four more threads to handle Raft specific tasks that can progress concurrently. A full cycle of request and response requires multiple tasks. In order to complete those, threads establish channels, which allows them to exchange messages when one task output is the input for another task.
Following a write request through Raft
Again, lets curl
curl -i -X PUT --data-binary 'some value to write' http://127.0.0.1:7879/key/some-new-key
Within that new thread the TcpStream is sniffed. Meaning it is not yet consumed but investigated to determine what data reached the node via TCP on that port. Is it a valid HTTP request, some JSON that might be parsed into a Client- or RaftMessage or something that we cannot identify and hence drop?
In our case we find a HTTP request from which we build a new Request type which is then cast into a
ClientMessage. Now the task of this thread is done since the request was handled successfully. The result
of this task is a new ClientMessage.
As a next step, the ClientMessage needs to be processed by the Raft engine. There is already a
responsible
thread in place named "raft-handle-incoming-client-messages" that was spawned when the
Raft engine started. This thread holds the receiving end of the client_incoming channel, specifically,
client_incoming_rx. Since handle_connection has a copy of the client_incoming_tx (sender
side) we can send our new ClientMessage using the sending end from the current thread to the thread that
handles ClientMessage. Thus, in the next step, that ClientMessage is moved to the client message
handling thread, and is further processed in the incoming_client_rpcs_loop().
/// This deviates slightly from the source code.
/// I removed some prints and rewrote comments to make it more readable
/// in context of my writing.
fn incoming_client_rpcs_loop<K, V>(
node_id: NodeID,
client_rx: PeerReceiver<rpc::ClientMessage<K, V>>,
client_tx: PeerSender<rpc::ClientMessage<K, V>>,
raft_tx: PeerSender<rpc::RaftMessage<S::Command>>,
state: Arc<Mutex<State<S>>>,
mut message_ids: impl Iterator<Item = NodeMessageID>,
) where
S::Command: From<(
NodeID,
NodeMessageID,
rpc::ClientMessage<K, V>,
PeerSender<ClientMessage<K, V>>,
)>,
{
let mut proxies = BTreeMap::new();
for (client, msg) in client_rx.iter() {
let response_id = message_ids.next().expect("should never run out of IDs");
let is_leader = state.lock().expect("no poison").is_leader();
match (is_leader, msg) {
(
true, // only leaders (should) add client requests to their log
req @ (rpc::ClientMessage::WriteRequest { .. }
| ClientMessage::ReadRequest { .. }
| ClientMessage::CASRequest { .. }),
) => {
let cmd: S::Command = (client, response_id, req, client_tx.clone()).into();
state.lock().expect("no poison").append(cmd);
state
.lock()
.expect("no poison")
.replicate_log()
.into_iter()
.try_for_each(|msg| raft_tx.send(msg))
.expect("raft receiver should never hang up");
}
(
false,
mut req @ (rpc::ClientMessage::WriteRequest { .. }
| ClientMessage::ReadRequest { .. }
| ClientMessage::CASRequest { .. }),
) => {
let leader = state.lock().expect("no poison").current_leader().cloned();
let (node, msg) = if let Some(leader) = leader {
if proxies.len() > MAX_INFLIGHT_PROXIES {
let (key, _) = proxies.pop_first().expect("just checked");
}
// message id swap here
let original_id = req.id();
let forward_id = response_id; // reuse but rename
eprintln!(
"proxy: to known leader {leader}: {}({}) -> {}",
client,
original_id,
forward_id.get()
);
req.set_id(forward_id.get());
// inserting forwarded message into proxy map here
let res = proxies.insert(forward_id.get(), (original_id, client));
assert!(res.is_none(), "node IDs are unique per process");
(leader, /* proxy original unchanged */ req)
} else {
(
client,
rpc::ClientMessage::ErrorResponse {
in_reply_to: req.id(),
id: response_id.get(),
code: ReservedErrorCode::TemporarilyUnavailable.into(),
text: "not a leader and current leader unknown".into(),
},
)
};
client_tx
.send((node, msg))
.expect("client message receiver should never hang up");
}
};
}
Here, three things can happen depending on the role of the node on which the thread runs and ClientMessage variant. In our scenario, the node is not a leader and the ClientMessage is a WriteRequest. So we hit the second branch of the match statement. Since only the leader is allowed to process requests, we need to forward the write request to the leader node. The state of a follower tracks who the leader is.1 And because node_id == port we know where to send the request to.
Before making further progress, we should discuss message ids. Each node has a NodeMessageIDGenerator, such that each message on the node gets a numeric id that is unique within the node (and therefore also across threads). A combination of node_id and message_id pairs the origin of a message is uniquely identified across processes. This is required to manage situations where messages cross node boundaries, like our write request. The request reached a follower, but the leader has to process the request and thus generate a response. In an asynchronous world, where many things happen concurrently on different nodes, we need a way to map a response produced by some thread to a request that lives on another thread. This is why the incoming_client_rpcs_loop() branch that we hit with our request on a follower node assigns a new message_id to the message and inserts that id along with the old one and the node_id into a proxy map. We will see how this works on the return trip of our write request.
Back to forwarding the request. The request is added to the proxy map and the leader identified. Now, we send the tuple (leader, message) down the channel handling outgoing ClientMessages. The send_tcp() method is triggered upon receiving both. That method serializes the message into JSON and sends it to the leader which listens on another port, assume 7878. This time, we sniff JSON data and try to parse it into a MessageEnvelope. Since now the MessageEnvelope's body is the ClientMessage::WriteRequest forwarded by 7879 and not a client response we hit the route_incoming() method. The router moves the request from the thread handling incoming connections to the one that handles incoming ClientMessages. Again. But remember, this time on the leaders node not the follower node. This time we are the leader and thus allowed to process the request. We hit the first branch in incoming_client_rpcs_loop().
There, the leader appends the write Command to its log (state). The state's
replicate_log() method is called, which returns a list of tuples. The list contains one (NodeID,
RaftMessage::AppendEntries) tuple for each follower. Each single one of these AppendEntries is moved over
to the thread that handles outgoing Raft messages and is therefore send to the followers via TCP.
On each follower node, the request is routed to the thread that handles incoming RaftMessages.
fn incoming_raft_rpcs_loop(
raft_rx: PeerReceiver<rpc::RaftMessage<S::Command>>,
raft_tx: PeerSender<rpc::RaftMessage<S::Command>>,
state: Arc<Mutex<State<S>>>,
machine: &mut S,
mut persist: impl Write + Seek,
) {
for (peer, msg) in raft_rx.iter() {
let remote_term = msg.term();
state
.lock()
.expect("no poison")
.maybe_step_down(remote_term);
let responses = match msg {
rpc::RaftMessage::RequestVote {
candidate_term,
last_log_index,
last_log_term,
} => vec![(
peer.clone(),
state.lock().expect("no poison").handle_vote_request(
peer,
candidate_term,
last_log_index,
last_log_term,
),
)],
rpc::RaftMessage::RequestVoteResponse { term, vote_granted } => state
.lock()
.expect("no poison")
.handle_vote_response(peer, term, vote_granted),
rpc::RaftMessage::AppendEntries {
term,
commit_index,
prev_log_index,
prev_log_term,
entries,
} => vec![(
peer.clone(),
state.lock().expect("no poison").handle_append_entries(
peer,
term,
commit_index,
prev_log_index,
prev_log_term,
entries,
machine,
),
)],
rpc::RaftMessage::AppendEntriesResponse { index, success, .. } => state
.lock()
.expect("no poison")
.handle_append_entries_response(peer, success, index, machine),
};
responses
.into_iter()
.try_for_each(|msg| raft_tx.send(msg))
.expect("raft receiver should never hang up");
}
}
To understand how Raft keeps logs in sync, let's have a look at the log iteself. Each server keeps an ordered list of commands
struct LogEntry<Cmd> {
cmd: Cmd,
term: Term,
}
pub struct Log<Cmd> {
inner: Vec<LogEntry<Cmd>>,
}
When a Raft leader replicates a Command, it never trusts that a follower's Log lines up with its own. Every AppendEntries request carries not just the new entries but also the LogIndex and Term of the entry immediately preceding them and the follower triggers consistency checks, before appending anything to its Log. Specifically, it only accepts the new entries if it already has an entry at that previous LogIndex with exactly that Term which happens in handle_append_entries().
/// This deviates slightly from the source code.
fn handle_append_entries(
&mut self,
peer: NodeID,
leader_term: Term,
leader_commit_index: Option<LogIndex>,
leader_prev_log_index: Option<LogIndex>,
leader_prev_log_term: Option<Term>,
entries: Log<S::Command>,
machine: &mut S,
) -> rpc::RaftMessage<S::Command> {
let current_term = self.c.current_term;
let failure_msg = rpc::RaftMessage::AppendEntriesResponse {
current_term,
index: self.c.log.highest_index(),
success: false,
};
if leader_term < current_term {
return failure_msg;
}
self.c.extend_election_deadline();
if let Role::Candidate { .. } = self.r {
self.become_follower(Some(self.c.id.clone()));
}
if let Role::Follower { leader, .. } = &mut self.r {
*leader = Some(peer); // Recognize this leader
} else {
unreachable!("in same term, if no longer candidate, can only be follower");
};
if leader_prev_log_index.and_then(|idx| self.c.log.get(idx).map(|entry| entry.term))
!= leader_prev_log_term
{
return failure_msg;
}
// replace log starting from matching index
self.c.log.replace_from(
leader_prev_log_index
.map(|idx| idx.inc())
.unwrap_or_default(/* wipe it all */),
entries.inner.into_boxed_slice(),
);
if leader_commit_index > self.c.commit_index {
self.c.commit_index = cmp::min(leader_commit_index, self.c.log.highest_index());
self.advance_state_machine(machine);
}
rpc::RaftMessage::AppendEntriesResponse {
current_term,
index: self.c.log.highest_index(),
success: true,
}
}
So an accepted append can only extend a Log that already matches the leader's up until this point. If
that
is the case, replace_from() is called. If, however, the check fails, the follower rejects the request and
sends a reponse back to the leader that indicates the request failed upon which the leader simply retries with
an
earlier LogIndex, walking backwards until it finds the last point of agreement and then overwrites all
later divergent commands on the follower.
The follower reports the outcome back in an AppendEntriesResponse. This is where the
leader closes the loop in
handle_append_entries_response().
fn handle_append_entries_response(
&mut self,
peer: NodeID,
success: bool,
index: Option<LogIndex>,
machine: &mut S,
) -> Vec<(NodeID, rpc::RaftMessage<S::Command>)> {
let Role::Leader {
next_indexes,
match_indexes,
..
} = &mut self.r
else {
return vec![];
};
let Some(next_index) = next_indexes.get_mut(&peer) else {
return vec![];
};
if !success {
*next_index = next_index.dec().unwrap_or_default();
return self.replicate_log();
}
let Some(match_index) = match_indexes.get_mut(&peer) else {
return vec![];
};
if let Some(index) = index {
*next_index = index.inc();
*match_index = Some(index);
self.advance_commit_index(machine);
}
vec![]
}
}
On a rejection, it decrements the match_index which signals where that follower's log ends and sends again from a previous log position by calling replicate_log(). On success, it records how far that follower has now replicated using the match_index. At this point the LogIndex for which each follower agrees with the leader can be different per node. But as discussed before, the leader just needs to ensure that a majority of nodes replicated the log up until some point. So what it does is to compute the median of all followers' log indices (majority check) and then applies all commands of the log up to this LogIndex to the state machine. This happens on all follower responses and is therefore fully asynchronous. The leader does not wait until a majority of followers came back with their AppendEntriesResponse. Note that all the back and forth I summarized above happens in the exact same way as outlined before. ClientMessages and RaftMessages are send across threads to handle tasks concurrently on the same node. Communication between leaders and followers happens by sending messages via tcp to the corresponding port of that node.
Our write request was received by a follower, forwarded to the leader, which appended it to its own Log
and asked followers to also append that write request to their Log. Both performed consistency checks to
ensure that finally their Logs will contain the same Commands. The only thing left to do is to
advance the state machine.
Followers advance their state machine immediately after successfully adding the
command to their log and before they respond to the leader. The leader receives the AppendEntriesResponse
computes the median of the match_index and then applies all Commands up until that index to the
state machine (HashMap). Once a majority of the followers came back with the responses to our write request, the
computation of the median will match the index at which the Command landed on a majority of followers. It
is committed and the leader applies it to its own Log.
Once advance_state_machine() applies the command, the state machine's apply() method turns the result into a ReadResponse/WriteResponse/CASResponse and pushes it onto the outgoing Raft message client channel. The receiving thread wraps the message in a MessageEnvelope and ships it over TCP to the node holding the client's connection. There it enters handle_connection, this time sniffed as a JSON client response rather than a HTTP request the node uses the response's id to pull the original TcpStream from the correlation_map, converts the ClientMessage into a concrete HTTP Response with the proper status code and writes it to that socket before closing it.
Done. You now have an intuition for why distributed state machines need a form of consensus, how Raft solves this and watched a write request's journey along a concrete implementation using miniraft. After reading this, pick up the original Raft paper, check out Alex Povel's miniraft and maybe come back to this article again. I will take a "leap of faith" here and say it is very much likely that you will have a very good understanding of Raft afterwards. Mostly, because the paper and miniraft are insanely well crafted.
Resources
- The original Raft paper.
- Alex Povel's miniraft. A Raft implementation with the Rust programming language.
- (PDF!) Martin Kleppmann's Distributed Systems lecture notes. Includes a lecture on Raft.
- The Raft website with a nice illustration and some more reference implementations.
- (PDF!) Diego Ongaro's dissertation. Includes a TLA+ proof of Raft.
- Jon Gjengset's blog post "Students' guide to Raft"
- async-raft. Rust implementation using Tokio.
- Recall, a follower knows who the leader is, because it receives heartbeat messages from the leader.↩