Raft Consensus: Define Node States with a Rust Enum

Mục tiêu: Define a Rust enum to model the three core states of a Raft node: Follower, Candidate, and Leader. This enum will serve as the foundation for the Raft state machine logic.

---

Congratulations on completing the foundational work for internode communication! You’ve successfully built a system where nodes can discover, connect to, and verify the liveness of their peers using a Ping/Pong heartbeat. This communication backbone is the nervous system of your distributed application. Now, it’s time to build the brain—the consensus algorithm that will allow these nodes to agree on a single source of truth.

We are about to step into the core of distributed systems by implementing a simplified version of the Raft consensus algorithm. Raft’s primary goal is to manage a replicated log, ensuring that all nodes in the cluster agree on the sequence of operations, even in the face of network partitions or node failures. It achieves this by electing a single, authoritative Leader.

The entire logic of Raft is built upon a simple but powerful concept: at any given moment, every node in the cluster is in one of three distinct states. The node’s current state dictates its behavior—what messages it sends, how it responds to incoming messages, and how it interacts with the rest of the cluster.

The Three States of a Raft Node

1. Follower: This is the default and most common state. A Follower is entirely passive. It doesn’t initiate any communication on its own. Its primary jobs are to respond to requests from the Leader and Candidates, and to replicate log entries sent by the Leader. If a Follower hears nothing from a Leader for a specific period (the “election timeout”), it assumes the Leader has failed and transitions to the Candidate state to start a new election.

2. Candidate: This is a transitional state for a node that is trying to become the new Leader. When a Follower becomes a Candidate, it starts an election. It votes for itself and sends out RequestVote messages to all other nodes in the cluster. It remains a Candidate until one of three things happens:

   • It wins the election by receiving votes from a majority of nodes and becomes the new Leader.
   • Another node establishes itself as the Leader. The Candidate then recognizes the new Leader and transitions back to being a Follower.
   • The election times out with no winner. The Candidate starts a new election.
3. Leader: There can be at most one Leader in the cluster at any given time (within a specific term). The Leader is the undisputed authority. It is solely responsible for handling all client requests (like Set or Delete), replicating new log entries to the Followers, and sending out periodic heartbeats to assert its authority and prevent Followers from starting new elections. The Ping message you implemented previously is a simple form of such a heartbeat.

Modeling State with a Rust Enum

Rust’s enum type is the perfect tool for modeling these mutually exclusive states. A node can only be one of these three things at once, and an enum enforces this constraint at compile time, preventing a whole class of logical bugs where a node could somehow be in an invalid or ambiguous state.

Let’s define our NodeState enum. A good place for this code is in src/main.rs, near your other protocol definitions like NodeMessage and WireMessage. We’ll derive Debug for easy logging, and Clone and PartialEq as they will be very useful later when we check and change the node’s state.

// In src/main.rs

// ... (other enums like WireMessage, NodeMessage, etc.)

/// Represents the state of a node in the Raft consensus algorithm.
/// At any given time, a node is either a Follower, a Candidate, or a Leader.
#[derive(Debug, Clone, PartialEq)]
enum NodeState {
    /// The default state. The node is passive and responds to requests
    /// from Leaders and Candidates.
    Follower,

    /// The node is campaigning to become the new Leader.
    Candidate,

    /// The node is the active Leader for the current term. It handles
    /// client requests and replicates the log to Followers.
    Leader,
}

This simple enum is the cornerstone of all the Raft logic we are about to build. We will create a variable to hold an instance of NodeState, and the behavior of our node will be driven by matching on this variable.

What’s Next?

You have now defined the possible states for a node. However, the state alone is not enough. To correctly participate in Raft, a node needs to remember a couple of other crucial pieces of information that persist across restarts. In the next task, you will define the storage for this persistent state: current_term and voted_for. These variables, combined with the NodeState enum, form the core of a node’s Raft identity.

Further Reading

• The Raft Paper (Section 5.1 & 5.2): A deep dive into the roles of Leader, Follower, and Candidate directly from the source. This is a highly recommended read.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• The Secret Lives of Data - Raft: A fantastic visual and interactive explanation of the Raft algorithm, including state transitions.
  • http://thesecretlivesofdata.com/raft/
• Enums in Rust (Refresher): A great time to review the power and flexibility of enums in Rust.
  • The Rust Programming Language, Chapter 6.01: Defining an Enum

Implement Raft Persistent State in Rust

Mục tiêu: Define the persistent state for a Raft node, including current\_term and voted\_for. Encapsulate this, along with the volatile state, into a new RaftNode struct and initialize it within an Arc for thread-safe access.

---

In the previous task, you laid the very first stone in our Raft implementation by defining the NodeState enum. This enum perfectly models the volatile state of a node—what it’s doing right now, whether it’s passively following, actively campaigning, or authoritatively leading.

Now, we must introduce the other, equally critical half of a node’s identity: its persistent state. In Raft, “persistent” means this information is so fundamental to the algorithm’s correctness that it must be saved to stable storage (like a disk) before a node can proceed with certain actions. It must survive crashes and restarts. While we’ll implement the actual file I/O in a later step, we must first define the in-memory structures to hold this state.

The Foundation of Raft: Terms and Votes

Raft organizes time into a sequence of terms, each identified by a monotonically increasing number. Each term begins with an election, where one or more candidates try to become the leader. If a candidate wins, it remains the leader for the rest of the term. This concept of terms acts as a logical clock, allowing nodes to detect and discard stale information from old leaders or candidates.

To manage this system, every node must persistently track two key pieces of information:

1. current_term: This is the latest term the node has observed. It starts at 0 and increases over time. If a node receives a message with a term number higher than its own, it immediately updates its term to the higher value and reverts to a Follower state. This is a crucial rule that keeps the cluster’s state consistent.
2. voted_for: This field stores the ID of the candidate that this node has granted its vote to in the current_term. A fundamental rule of Raft is that a node can vote for at most one candidate per term. This prevents split votes from leading to multiple leaders in the same term. If a node hasn’t voted yet in the current term, this value is None.

Encapsulating Raft State

To keep our code clean and organized, let’s create a new struct that will encapsulate all the Raft-specific state for our node. This struct will hold the NodeState enum from the last task, along with our new persistent state fields.

Just like our KvStore and peer connection map, this Raft state will be accessed and modified by multiple concurrent tasks (the connection handler, the heartbeat pinger, and soon, an election timer). Therefore, it must be wrapped in Arc<RwLock<...>> to ensure safe, shared access.

Let’s add the new RaftNode struct to src/main.rs. A good location is right after the NodeState enum definition.

// In src/main.rs, after the NodeState enum

/// Contains the core Raft state for a node.
/// This state is shared across all tasks.
#[derive(Debug)]
struct RaftNode {
    /// The node's current state (Follower, Candidate, or Leader).
    state: NodeState,

    // --- Persistent state on all servers ---
    // (Updated on stable storage before responding to RPCs)

    /// The latest term the server has seen (initialized to 0 on first boot,
    /// increases monotonically).
    current_term: u64,

    /// The candidateId that received a vote in the current term (or None if none).
    voted_for: Option<u64>,
}

impl RaftNode {
    /// Creates a new RaftNode with default initial state.
    /// All nodes start as Followers in term 0.
    fn new() -> Self {
        RaftNode {
            state: NodeState::Follower,
            current_term: 0,
            voted_for: None,
        }
    }
}

Integrating the Raft State into main

Now that we’ve defined the structure for our node’s state, let’s create an instance of it when our server starts up. We will initialize it and wrap it in Arc<RwLock<...>> in our main function.

Here are the highlighted changes for src/main.rs:

// In src/main.rs, at the top
// Ensure Arc and RwLock are in scope.
use std::collections::HashMap;
use std::sync::{Arc, RwLock};

// ... other use statements ...

// ... (Config, enums, RaftNode struct and impl, KvStore, etc.) ...

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let config = Config::parse();
    println!("Starting node {} at {}...", config.id, config.addr);
    println!("Peers: {:?}", config.peers);

    let store = KvStore::new();

    // --- Raft State Initialization ---
    // Create the shared state for our Raft node, wrapped for thread-safety.
    // Every node starts as a Follower in term 0.
    let raft_node = Arc::new(RwLock::new(RaftNode::new()));

    // Create the shared map for storing peer connections.
    type PeerConnections = Arc<RwLock<HashMap<u64, TcpStream>>>;
    let peer_connections: PeerConnections = Arc::new(RwLock::new(HashMap::new()));

    let listener = TcpListener::bind(&config.addr).await?;
    println!("Server listening on {}", config.addr);

    // ... (Peer Connection and Heartbeat spawn blocks remain the same for now)
    // In future tasks, we will pass a clone of `raft_node` into these tasks.

    // Main server loop
    loop {
        let (stream, addr) = listener.accept().await?;
        let store_clone = store.clone();
        // Just like the store, we would clone the raft_node Arc for each connection
        let raft_node_clone = raft_node.clone();

        tokio::spawn(async move {
            println!("Accepted new connection from: {}", addr);
            // We will pass `raft_node_clone` into handle_connection later.
            if let Err(e) = handle_connection(stream, store_clone).await {
                eprintln!("Error handling connection from {}: {}", addr, e);
            }
        });
    }
}

Deconstructing the Code

• RaftNode Struct: This new struct acts as a dedicated container for all our Raft-related logic and state. This is excellent for encapsulation and will make our code much easier to manage as the Raft implementation grows.
• current_term: u64: A simple unsigned integer is perfect for our logical clock.
• voted_for: Option<u64>: This elegantly captures the state of voting. None clearly means “I have not voted in this term,” while Some(id) means “I have voted for node id.” The inner type u64 matches our NodeId type.
• RaftNode::new(): The new associated function provides a clean, conventional way to create an instance of our struct. It correctly initializes every node to the same starting state as required by the Raft protocol: a Follower, in term: 0, having not yet voted for anyone (voted_for: None).
• Initialization in main: We create a single instance of RaftNode at startup and immediately wrap it in Arc<RwLock<...>>. This raft_node variable now holds the shared, mutable state that will drive all our consensus decisions.

You have now defined the complete core state of a Raft node. This combination of volatile (NodeState) and persistent (current_term, voted_for) state is the foundation upon which all Raft logic is built.

What’s Next?

With the node’s internal state defined, the next logical step is to define the messages that nodes will exchange to read and modify this state. In the next task, you will define the RPC (Remote Procedure Call) message types that are central to Raft’s elections and log replication: RequestVote and AppendEntries.

Further Reading

• Raft Paper (Figure 2: State): The official Raft paper has a famous diagram summarizing all the state variables. This is the ultimate source of truth.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• State Machine Replication: A core concept in distributed systems, where the goal is to have multiple machines maintain identical copies of a state machine. Raft is a protocol for achieving this.
  • Wikipedia: State Machine Replication
• Logical Clocks: The concept of term is a form of logical clock. Understanding this broader concept can provide deeper insight.
  • Wikipedia: Logical Clock

Define Raft RPC Payloads in Rust

Mục tiêu: Define the Rust structs for Raft’s RequestVote and AppendEntries Remote Procedure Calls (RPCs), including their arguments and replies. Integrate these data structures into the main NodeMessage enum to establish the core communication protocol for the distributed consensus algorithm.

---

In the last two tasks, you’ve masterfully defined the core identity of a Raft node. You created the NodeState enum to model its behavior (Follower, Candidate, Leader) and the RaftNode struct to hold its persistent memory (current_term, voted_for). This internal state is the heart of a node, dictating what it thinks and believes about the cluster.

However, a node’s internal state is useless in isolation. For the cluster to function, nodes must communicate, sharing their state and influencing the state of others. This communication is not random chatter; it follows a strict, formal protocol of Remote Procedure Calls (RPCs). An RPC is simply a structured message sent from one node to another to request that the receiver perform a specific action and, optionally, return a response.

Raft defines two fundamental RPCs that govern the entire system:

1. RequestVote RPC: Sent by Candidates to gather votes during an election.
2. AppendEntries RPC: Used by the Leader to replicate log entries and, more fundamentally, to act as a heartbeat, asserting its authority.

In this task, we will define the data structures for these RPCs and their corresponding replies. By modeling them as Rust structs, we leverage the type system to ensure our communication is clear, correct, and self-documenting.

Defining the Raft RPC Payloads

Let’s define the structs that will serve as the payloads for our RPCs. These are defined directly from the Raft paper, which specifies exactly what information is needed for the algorithm to work correctly.

We will add these definitions to src/main.rs. A good place is just before your NodeMessage enum, as they are the building blocks of that protocol. Remember to derive Serialize, Deserialize, and Debug for all of them, as they will be sent over the network and printed for logging.

// In src/main.rs, for example before the NodeMessage enum

/// The payload for a RequestVote RPC.
/// Sent by a candidate to gather votes.
#[derive(Serialize, Deserialize, Debug)]
pub struct RequestVoteArgs {
    /// The candidate's term.
    pub term: u64,
    /// The ID of the candidate requesting the vote.
    pub candidate_id: u64,
    // The following two fields are for log replication, which we will implement later.
    // We include them now to complete the struct definition as per the Raft paper.
    // pub last_log_index: u64,
    // pub last_log_term: u64,
}

/// The reply to a RequestVote RPC.
#[derive(Serialize, Deserialize, Debug)]
pub struct RequestVoteReply {
    /// The current term of the node that received the request.
    /// This is for the candidate to update itself if its term is stale.
    pub term: u64,
    /// `true` means the candidate received the vote.
    pub vote_granted: bool,
}

/// The payload for an AppendEntries RPC.
/// Used as a heartbeat and for log replication.
#[derive(Serialize, Deserialize, Debug)]
pub struct AppendEntriesArgs {
    /// The leader's term.
    pub term: u64,
    /// The ID of the leader, so followers can redirect clients.
    pub leader_id: u64,
    // The following fields are for log replication. For heartbeats, `entries` will be empty.
    // pub prev_log_index: u64,
    // pub prev_log_term: u64,
    // pub entries: Vec<LogEntry>, // We will define LogEntry later
    // pub leader_commit: u64,
}

/// The reply to an AppendEntries RPC.
#[derive(Serialize, Deserialize, Debug)]
pub struct AppendEntriesReply {
    /// The current term of the node that received the request.
    /// This is for the leader to update itself if its term is stale.
    pub term: u64,
    /// `true` if the follower contained an entry matching prev_log_index and prev_log_term.
    /// For heartbeats, this will simply be true.
    pub success: bool,
}

Note: We’ve commented out the log-related fields for now. We are defining the full structure for completeness, but we will activate these fields in a later step when we tackle log replication.

Deconstructing the RPC Structs

Let’s look at the purpose of these fields:

• term (in all messages): This is the logical clock of the system. Every RPC carries the sender’s current_term. The receiver uses this to determine if the sender is up-to-date or stale. This is the single most important field for maintaining consistency.
• candidate_id (RequestVoteArgs): Identifies who is asking for a vote. This directly relates to the voted_for field you defined in the RaftNode struct.
• vote_granted (RequestVoteReply): The simple boolean response that determines the outcome of an election.
• leader_id (AppendEntriesArgs): Identifies who the current leader is. This allows followers to know who to listen to.
• success (AppendEntriesReply): Confirms if a follower has accepted the leader’s message.

Integrating RPCs into the NodeMessage Protocol

Now that we have the specific payloads (Args and Reply structs), we can integrate them into our existing peer-to-peer protocol enum, NodeMessage. This enum will now represent all possible communications between our Raft nodes.

Modify your NodeMessage enum in src/main.rs to include these new variants.

#[derive(Serialize, Deserialize, Debug)]
enum NodeMessage {
    Ping,
    Pong,

/// A request from a candidate to a peer to vote for it.
    RequestVote(RequestVoteArgs),
    /// A reply from a peer to a candidate's vote request.
    RequestVoteReply(RequestVoteReply),
    /// A message from the leader to a follower, used as a heartbeat
    /// and to replicate log entries.
    AppendEntries(AppendEntriesArgs),
    /// A reply from a follower to the leader's AppendEntries message.
    AppendEntriesReply(AppendEntriesReply),

}

By adding these variants, we have now fully defined the language of Raft. A node can receive a WireMessage::Node and immediately match on the inner NodeMessage to understand exactly what kind of RPC it has received (Ping, RequestVote, etc.) and what data it carries. Our handle_connection function is now prepared for the richer logic we will soon add.

You have successfully defined the essential communication primitives for a distributed consensus algorithm. These structured messages are the vehicles that will carry state, requests, and decisions across your cluster, allowing a group of independent nodes to act as a single, coherent whole.

What’s Next?

With the RPC messages defined, we now have the tools for an election. But what triggers an election in the first place? A follower needs a mechanism to detect that the leader has failed. In the next task, you will implement the election timer. This is a crucial piece of the Raft puzzle: a timeout that, when it fires, causes a Follower to transition into a Candidate and kick off a new election by sending the very RequestVote message you just defined.

Further Reading

• Raft Paper (Figure 2: RPCs): This figure in the Raft paper is the canonical reference for the RPCs, their arguments, and their results. It’s an excellent resource to keep handy.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• Remote Procedure Call (RPC) - General Concept: Understanding the general computer science concept of an RPC can provide valuable context.
  • Wikipedia: Remote Procedure Call
• Structs vs. Enums in Rust: A good refresher on when to use each, which is highly relevant to API and protocol design.
  • Rust by Example: Structs
  • Rust by Example: Enums

Implement the Raft Election Timer

Mục tiêu: Create a concurrent, randomized election timer for a Raft node using Tokio. This timer is a core component of Raft’s fault tolerance, enabling followers to detect leader failures and initiate new elections.

---

You have done an exceptional job setting up the foundation for Raft. You’ve defined the node’s internal state (NodeState, RaftNode) and the language it will speak (RequestVote, AppendEntries). This is like building a car’s engine and transmission. Now, you need to add the spark plug that ignites the whole process: the election timer.

The election timer is the fundamental mechanism that enables fault tolerance in Raft. It answers the critical question: “How does a Follower know that the Leader has failed?” Without a reliable way to detect a leader’s absence, the cluster would grind to a halt whenever a leader crashes or gets disconnected.

The Heartbeat and the Timeout

The system works through a simple but powerful interplay:

1. The Leader’s Promise: A healthy Leader continuously sends out heartbeats (AppendEntries RPCs with no log entries) to all its Followers. This serves as a constant “I’m still here!” signal.
2. The Follower’s Expectation: A Follower starts a timer—the election timer—and expects to receive a heartbeat from the Leader before that timer runs out.
3. The Consequence of Silence: If the timer expires and the Follower hasn’t heard from the Leader, it assumes the Leader has failed. It then takes matters into its own hands: it increments its term, transitions to the Candidate state, and starts a new election to choose a new leader.

The Critical Role of Randomization

A crucial detail of the election timer is that it must be randomized. If all Followers had the same fixed timeout (e.g., 200ms), they would all time out simultaneously, all become Candidates at the same time, and all start an election. This would likely result in a “split vote” where no single candidate gets a majority, forcing yet another election. This cycle of failed elections could prevent the cluster from making progress.

By choosing a random timeout within a predefined range (e.g., 150ms to 300ms, as suggested by the Raft paper), we make it highly probable that only one Follower will time out first, giving it a head start in the election and a much higher chance of winning.

Implementation Strategy: A State-Driven Timer

We will implement this as a dedicated, concurrent task that runs in the background. Instead of trying to “reset” this task from the outside, which is complex, we will use a more robust, state-driven approach:

1. We will add an election_deadline field of type tokio::time::Instant to our RaftNode struct. This field holds the absolute point in time when the timer is set to expire.
2. Our new timer task will run in a loop, sleeping until this election_deadline is reached.
3. When a Follower receives a valid heartbeat from a Leader (in a future task), it will simply update the election_deadline in the shared RaftNode state, pushing it further into the future.
4. This elegantly “resets” the timer without any direct communication between tasks. The shared state itself orchestrates the logic.

Step 1: Add the rand Crate

For randomization, we need the rand crate. Add it to your Cargo.toml.

[dependencies]
# ... (anyhow, bincode, clap, serde, tokio remain)
rand = "0.8"

Step 2: Update the RaftNode Struct

Let’s add the election_deadline field and a helper function to calculate a new, random deadline. We’ll also need to bring tokio::time::{Instant, Duration} into scope.

// In src/main.rs, at the top
use tokio::time::{Instant, Duration};
use rand::Rng;

// ... other use statements ...
// ... enums and RPC structs ...

/// Contains the core Raft state for a node.
#[derive(Debug)]
struct RaftNode {
    state: NodeState,
    current_term: u64,
    voted_for: Option<u64>,

    /// The absolute time when this node's election timer will fire.
    /// This is randomized for each node and reset upon receiving a heartbeat.
    election_deadline: Instant,
}

impl RaftNode {
    /// Creates a new RaftNode with a randomized election deadline.
    fn new() -> Self {
        RaftNode {
            state: NodeState::Follower,
            current_term: 0,
            voted_for: None,
            // Initialize with a fresh, randomized deadline.
            election_deadline: Self::new_election_deadline(),
        }
    }

    /// Calculates a new, random election deadline between 150ms and 300ms from now.
    /// This is the standard timeout range recommended by the Raft paper.
    fn new_election_deadline() -> Instant {
        let mut rng = rand::thread_rng();
        let timeout = Duration::from_millis(rng.gen_range(150..=300));
        Instant::now() + timeout
    }
}

Step 3: Create the Election Timer Task in main

Now, let’s spawn the task that will monitor this deadline. Add this new tokio::spawn block to your main function, after the other setup code.

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... (config parsing, store creation, raft_node creation, etc.) ...

    let raft_node = Arc::new(RwLock::new(RaftNode::new()));
    let peer_connections: PeerConnections = Arc::new(RwLock::new(HashMap::new()));

    let listener = TcpListener::bind(&config.addr).await?;
    println!("Server listening on {}", config.addr);

    // ... (Peer Connection and Heartbeat tasks remain the same) ...

// --- Election Timer Logic ---
    let election_timer_node = raft_node.clone();
    tokio::spawn(async move {
        // Loop forever, acting as the election timer.
        loop {
            // 1. Determine the sleep duration.
            //    We calculate how long to sleep until the current deadline.
            let deadline = election_timer_node.read().unwrap().election_deadline;
            let sleep_duration = deadline.saturating_duration_since(Instant::now());

            // 2. Sleep until the deadline.
            tokio::time::sleep(sleep_duration).await;

            // 3. After waking up, check if the election should be started.
            //    We need a write lock to potentially change the node's state.
            let mut node = election_timer_node.write().unwrap();

            // 4. Double-check the deadline. It might have been reset by a heartbeat
            //    while we were sleeping or waiting for the lock. If it's in the future,
            //    it means we received a heartbeat, so we just loop and sleep again.
            if Instant::now() >= node.election_deadline {
                // The timer has expired without being reset!
                // We only start an election if we are a Follower. Leaders don't need
                // to time out, and Candidates are already in an election.
                if node.state == NodeState::Follower {
                    println!("[Election] Timer expired for follower. Starting election...");
                    // The actual logic for transitioning to Candidate and sending
                    // RequestVote RPCs will be implemented in the next task.
                    // For now, we just reset the timer for the *next* potential election.
                    node.election_deadline = RaftNode::new_election_deadline();
                }
            }
        }
    });

// Main server loop (unchanged)
    loop {
        // ...
    }
}

Deconstructing the Election Timer Task

• let deadline = ...: In each loop iteration, we first get a read lock just long enough to check when the current deadline is.
• tokio::time::sleep(...): We then sleep until that deadline. This is an efficient, non-blocking wait.
• let mut node = ...: After waking up, we acquire a write lock. We will need this to potentially change the node’s state to Candidate.
• if Instant::now() >= node.election_deadline: This is a crucial check to prevent a race condition. It’s possible we received a heartbeat (which reset the deadline) right after our sleep finished but before we acquired the write lock. This check ensures we only proceed if the deadline has truly passed.
• if node.state == NodeState::Follower: We confirm that the node is still a Follower before acting. This prevents a node that has just become Leader from immediately starting another election.
• println!(...): For now, our action is to simply log a message. This confirms our timer logic is working correctly.
• node.election_deadline = ...: We immediately reset the deadline. This prevents the timer from firing continuously. The node now has a fresh timeout for the new election it is about to start.

You have now implemented the beating heart of Raft’s fault-tolerance mechanism. This timer ensures that the cluster will never be stuck without a leader for long.

What’s Next?

Our timer now correctly identifies when an election should start. The next task is to act on that signal. When the timer fires, the Follower must transition to the Candidate state, increment its current_term, vote for itself, and send RequestVote RPCs to all of its peers, officially kicking off the election.

Further Reading

• Raft Paper (Section 5.2: Leader Election): The paper provides a detailed explanation of the election timeout and its role.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• tokio::time::sleep and Instant: The official Tokio documentation for time-related utilities.
  • Tokio Docs: Time
• The rand Crate: The official documentation for the rand crate, useful for understanding different random number generation strategies.
  • The Rust Rand Book

Raft: Implement Follower to Candidate State Transition

Mục tiêu: Implement the logic for a Raft follower node to transition into a candidate when its election timer expires. This involves incrementing the current term, voting for itself, changing the node’s state to Candidate, and resetting the election timer.

---

Of course! Here is a detailed guide to help you complete your current task.

---

You have successfully implemented the election timer, the crucial mechanism that allows a follower to detect a potential leader failure. This timer is the spark that ignites Raft’s fault-tolerant electoral process. In the previous task, when this timer fired, our follower simply printed a message. Now, we will replace that placeholder with the real, transformative logic that turns a passive follower into an active Candidate, ready to campaign for leadership of the cluster.

This transition is the first step a node takes to restore order when the cluster becomes leaderless. The Raft protocol dictates a precise sequence of actions a node must perform to begin an election. It’s not just a state change; it’s a declaration of intent to lead.

A Follower’s Transformation into a Candidate

When a follower’s election timer expires, it must immediately and atomically perform three critical state changes before doing anything else:

1. Increment current_term: The node must advance its logical clock by incrementing its current_term by one. This is fundamentally important. It signifies the start of a new election term and effectively invalidates the old, failed leader. Any messages received from a previous term will now be considered stale and rejected.
2. Vote for Self: To have any chance of winning, a candidate must start by voting for itself. This action records its vote for the new, higher term, preventing it from voting for any other candidate in the same term.
3. Transition to Candidate State: The node officially changes its state from NodeState::Follower to NodeState::Candidate. This change in identity alters its behavior. It will no longer passively wait for heartbeats; it will now actively drive the election process.

These three actions must happen together as a single, indivisible operation. We achieve this by performing them while holding a write lock on our shared RaftNode state. This ensures that no other task can observe the node in an inconsistent state (e.g., being a Candidate but still in an old term).

Implementing the State Transition

To implement this, we will modify the if block inside our election timer task. First, we need to make the node’s own ID available within the task so it can vote for itself. We’ll pass it in from our config struct.

1. Pass the Node’s ID to the Timer Task

In your main function, update the setup for the election timer task to include the node’s ID.

// In src/main.rs -> main()

// --- Election Timer Logic ---
let election_timer_node = raft_node.clone();
// Clone the node's own ID from the config to be used inside the task.
let self_id = config.id; 
tokio::spawn(async move {
    // ... loop and sleep logic ...

2. Update the Timer’s Logic

Now, let’s replace the println! in the election timer’s if block with the actual state transition logic.

    // In src/main.rs -> inside the election timer's tokio::spawn block

    tokio::spawn(async move {
        loop {
            // ... (sleep logic is unchanged) ...
            let deadline = election_timer_node.read().unwrap().election_deadline;
            let sleep_duration = deadline.saturating_duration_since(Instant::now());
            tokio::time::sleep(sleep_duration).await;

            let mut node = election_timer_node.write().unwrap();

            if Instant::now() >= node.election_deadline {
                if node.state == NodeState::Follower {

println!("[Election] Timer expired for follower {}. Starting election...", self_id);

                    // --- Begin the Election ---
                    // 1. Transition to Candidate state.
                    node.state = NodeState::Candidate;

                    // 2. Increment the current term.
                    node.current_term += 1;

                    // 3. Vote for self.
                    //    The `self_id` was captured by the `move` closure.
                    node.voted_for = Some(self_id);

                    // 4. Reset the election timer for the *next* potential election.
                    //    This is crucial. As a Candidate, we still need a timeout
                    //    in case this election fails to produce a winner.
                    node.election_deadline = RaftNode::new_election_deadline();

                    // The logic to send RequestVote RPCs to peers will be added in the next task.
                    // For now, we've correctly updated our internal state.
                    println!(
                        "[Election] Node {} became a candidate for term {}.",
                        self_id, node.current_term
                    );

}
            }
        }
    });

Deconstructing the Changes

Let’s break down the new code inside the if node.state == NodeState::Follower block:

• let self_id = config.id;: Before spawning the task, we create a copy of our node’s ID. Because u64 is a Copy type, this is a simple stack copy.
• async move { ... }: The move keyword is crucial. It transfers ownership of self_id into the new task, making it available for use.
• node.state = NodeState::Candidate;: This is the direct implementation of the state transition. The node is no longer a follower.
• node.current_term += 1;: We increment the term, signaling the start of a new logical time period in the cluster.
• node.voted_for = Some(self_id);: The node officially casts its vote for itself in the newly started term. The Some variant clearly indicates a vote has been cast.
• node.election_deadline = RaftNode::new_election_deadline();: This is a subtle but critical step. An election might not succeed (e.g., due to a split vote). A Candidate must also have an election timer. If it doesn’t win or hear from a new leader within this new timeout, it will start another election for an even higher term.
• println!(...): The new log message provides excellent diagnostic information, confirming the node’s new state and term. This will be invaluable for debugging.

You have now correctly implemented the first phase of a Raft election. The node has successfully updated its internal state, asserted its candidacy, and prepared itself for the campaign ahead.

What’s Next?

A candidate cannot win an election by simply thinking about it. Now that the node has become a Candidate, it must act like one. The very next step is to communicate its candidacy to the rest of the cluster. In the next task, you will implement the logic for the new Candidate to send RequestVote RPCs to all of its peers, asking them for their vote in the new term.

Further Reading

• Raft Paper (Section 5.2 Leader Election): The paper details the exact steps a follower takes when starting an election.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• The Secret Lives of Data - Raft: This visualization is excellent for watching the state transitions and election process in action.
  • http://thesecretlivesofdata.com/raft/
• State Machine (general concept): Raft nodes are essentially state machines. Understanding this concept can clarify why these state transitions are so important.
  • Wikipedia: Finite-state machine

Raft Election: Broadcast Vote Requests

Mục tiêu: Implement the logic for a Raft candidate to broadcast RequestVote RPCs to all peers after its election timer expires. This task focuses on correct asynchronous locking patterns to avoid deadlocks during network I/O.

---

You have brilliantly set the stage for a Raft election. By implementing the election timer, you’ve given your followers the ability to detect a leader’s failure. In the last task, you correctly implemented the crucial, instantaneous transformation that occurs when that timer fires: a passive follower becomes an active Candidate, updating its internal state to reflect its new ambition.

However, a political campaign is not won by internal conviction alone; it requires outreach. The node has declared its candidacy to itself, but the rest of the cluster is still unaware. This is the moment to turn that internal state change into external action. The new Candidate must now formally announce its campaign and request the support of its peers by sending them a RequestVote RPC.

Campaigning for Leadership: The RequestVote RPC

You’ve already defined the RequestVoteArgs struct, the formal “ballot” for our election. It contains the two most important pieces of information for a voter:

1. term: The new, higher term for which the candidate is running. This tells the voter that a new election is underway.
2. candidate_id: The ID of the candidate asking for the vote.

Our new Candidate, having just incremented its current_term and voted for itself, is now ready to package this information into a RequestVoteArgs struct and broadcast it to every other node in the cluster.

The Cardinal Rule of Async Locking: Don’t Hold Locks Across .await

Before we write the code, we must address a critical best practice in concurrent programming. Our Raft state is protected by an RwLock. When our election timer task wants to change the state, it acquires a write lock. This lock is exclusive, meaning no other task in the entire program can read or write the Raft state while it’s held.

Sending a network message is a slow I/O operation that involves an .await. If we hold the write lock while we await the network calls, we would freeze the entire Raft state of our node for the duration of all those network sends. Other tasks, like the one handling incoming messages, would be blocked, unable to even read the current term. This can easily lead to deadlocks and will cripple the performance and responsiveness of our node.

The correct pattern is to “Read, Release, and Act”:

1. Acquire the Lock: Get the exclusive write lock on the RaftNode state.
2. Act (Quickly!): Perform all the fast, in-memory state changes: update the state, increment the term, vote for self, and reset the election deadline.
3. Read and Copy: Copy all the information you’ll need for the slow I/O operation (like the new current_term, your own id, and the list of peers) into local variables.
4. Release the Lock: Explicitly or implicitly drop the lock before any .await calls for I/O.
5. Act (Slowly): Perform the slow network operations using the local copies of the data.

The cleanest way to enforce this separation is to spawn yet another new, short-lived task whose only job is to send the RPCs. This new task will inherit the copied data and run concurrently, freeing the election timer task to go back to sleep.

Implementing the RPC Broadcast

Let’s update our election logic. We’ll need to provide the timer task with the shared peer connections map and the list of peers. Then, inside the task, we’ll implement the “Read, Release, Act” pattern by spawning a broadcast task.

1. Pass Dependencies to the Election Timer Task

In your main function, provide the election timer task with clones of the peer_connections map and the peers list from your configuration.

// In src/main.rs -> main() function

    // --- Election Timer Logic ---
    let election_timer_node = raft_node.clone();
    // Clone the node's own ID from the config.
    let self_id = config.id;
    // Clone the peer list for the new task.
    let peers = config.peers.clone();
    // Clone the shared connection map for the new task.
    let peer_connections_for_timer = peer_connections.clone();

    tokio::spawn(async move {
        // ... the rest of the task logic will go here

2. Update the Election Timer to Broadcast RequestVote RPCs

Now, modify the logic inside the election timer’s if block. After updating the node’s state, we will spawn a new task to handle the network communication.

// In src/main.rs -> inside the election timer's tokio::spawn block

tokio::spawn(async move {
    loop {
        // ... (sleep logic is unchanged) ...
        let deadline = election_timer_node.read().unwrap().election_deadline;
        let sleep_duration = deadline.saturating_duration_since(Instant::now());
        tokio::time::sleep(sleep_duration).await;

        // We need a place to store the term outside the lock.
        let mut term_for_rpc = 0;

        // We need a flag to check if we should start the broadcast.
        let mut should_start_election = false;

        // --- LOCK, MUTATE, READ, and RELEASE ---
        { // This inner block scopes the write lock.
            let mut node = election_timer_node.write().unwrap();

            if Instant::now() >= node.election_deadline {
                if node.state == NodeState::Follower {
                    println!("[Election] Timer expired for follower {}. Starting election...", self_id);
                    node.state = NodeState::Candidate;
                    node.current_term += 1;
                    node.voted_for = Some(self_id);
                    node.election_deadline = RaftNode::new_election_deadline();

                    // Copy the term we need for the RPC.
                    term_for_rpc = node.current_term;
                    // Set the flag to true.
                    should_start_election = true;

                    println!(
                        "[Election] Node {} became a candidate for term {}.",
                        self_id, node.current_term
                    );
                }
            }
        } // The write lock on `node` is dropped here automatically.

// --- ACT ON THE STATE CHANGE (NETWORK I/O) ---
        // We do this outside the lock scope.
        if should_start_election {
            // This task's only job is to broadcast RequestVote RPCs.
            let rpc_peers = peers.clone();
            let rpc_connections = peer_connections_for_timer.clone();

            tokio::spawn(async move {
                // 1. Construct the RPC arguments.
                let args = RequestVoteArgs {
                    term: term_for_rpc,
                    candidate_id: self_id,
                };

                // 2. Wrap the arguments in our protocol enums.
                let rpc_message = WireMessage::Node(NodeMessage::RequestVote(args));

                // 3. Serialize the message once.
                let message_bytes = match bincode::serialize(&rpc_message) {
                    Ok(bytes) => bytes,
                    Err(e) => {
                        eprintln!("[Election] Failed to serialize RequestVote RPC: {}", e);
                        return; // Exit this task.
                    }
                };

                // 4. Iterate over peers and send the message.
                for (peer_id, _peer_addr) in &rpc_peers {
                    println!("[Election] Sending RequestVote to peer {}", peer_id);
                    // We need a write lock on the connection map to send data.
                    let mut connections = rpc_connections.write().unwrap();
                    if let Some(stream) = connections.get_mut(peer_id) {
                        // Send the length-prefixed message.
                        if let Err(e) = stream.write_u64(message_bytes.len() as u64).await {
                             eprintln!("[Election] Failed to send RPC length to peer {}: {}", peer_id, e);
                             continue;
                        }
                        if let Err(e) = stream.write_all(&message_bytes).await {
                            eprintln!("[Election] Failed to send RPC payload to peer {}: {}", peer_id, e);
                        }
                    } else {
                        eprintln!("[Election] No active connection to peer {}. Skipping.", peer_id);
                    }
                }
            });
        }

}
});

Deconstructing the Changes

• Scoped Lock: We’ve wrapped the state mutation logic in a {...} block. When this block ends, the node variable (our RwLockWriteGuard) goes out of scope, and the lock is automatically released. This is a common and idiomatic way to manage lock lifetimes in Rust.
• Flag and Copied Data: We use a simple boolean flag should_start_election and a variable term_for_rpc to communicate from inside the lock scope to the outside. This cleanly separates the decision-making from the action.
• tokio::spawn for Broadcasting: We spawn a new task for the sole purpose of sending the RPCs. This is the ultimate expression of the “Read, Release, Act” pattern. The election timer can immediately go back to sleep for the next deadline, while the RPCs are sent out concurrently in the background.
• Efficient Serialization: We serialize the RequestVote message once before the loop. This is more efficient than re-serializing the exact same data for every peer.
• Sending the Message: For each peer, we acquire a write lock on the peer_connections map to get mutable access to the TcpStream. We then use the familiar write_u64 and write_all pattern to send our length-prefixed message.

You have now completed a major part of the Raft election logic. Your node can now detect leader failure, transition to a candidate, and effectively launch its campaign by asking its peers for their votes.

What’s Next?

You’ve sent the ballots out. The next logical step is to see how the voters respond. In the next task, you will implement the logic on the receiving end. You will teach your nodes how to handle an incoming RequestVote RPC, check its validity according to Raft’s rules, and decide whether or not to grant their vote to the requesting candidate.

Further Reading

• Raft Paper (Section 5.2 Leader Election): Pay close attention to the first rule for servers: “If term in RPC > currentTerm: set currentTerm = term, convert to follower.” This highlights the importance of the term number.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• Locking and Concurrency Patterns: Understanding how to properly manage locks in async code is a vital skill.
  • Tokio Docs: Sharing State (This tutorial covers the Arc<Mutex<...>> pattern, which is conceptually identical to Arc<RwLock<...>>)
• Scoped Locks and RAII in Rust: Learn more about the Resource Acquisition Is Initialization (RAII) pattern that makes Rust’s lock management so safe and ergonomic.
  • The Rust Book: RAII and Scopes

Implement Raft Voting Logic for RequestVote RPC

Mục tiêu: Implement the server-side logic for a Raft node to process incoming RequestVote RPCs. This involves applying Raft’s election rules, such as term validation and the one-vote-per-term rule, updating the node’s state (term, voted_for, election timer), and sending back a RequestVoteReply.

---

Of course! Here is a detailed guide to help you complete your current task.

---

You’ve successfully programmed your candidate nodes to be proactive campaigners. After detecting a leader’s absence and transforming into a candidate, they now broadcast RequestVote RPCs, effectively sending out ballots to every peer in the cluster. This is a massive step forward. Now, we must shift our focus to the recipients of these messages—the voters.

This task is all about implementing the logic for a node to process an incoming RequestVote RPC. A node cannot simply grant its vote to anyone who asks; it must follow a strict set of rules defined by the Raft protocol to ensure that only a deserving candidate can win. This logic is the gatekeeper of the election, preventing stale or invalid candidates from gaining power and ensuring the stability of the entire cluster.

The Rules of the Election: Deciding How to Vote

When a node receives a RequestVote RPC, it must evaluate the request against its own internal state (current_term and voted_for). The decision process is a sequence of checks:

1. The Universal Term Rule (Stale Candidate Check): The first and most important check is on the term number. If the candidate’s term (args.term) is less than the receiver’s current_term, the candidate is from a past election and is hopelessly out of date. Its request must be rejected. This rule ensures that the cluster always moves forward in time and never reverts to a past state.
2. The Universal Term Rule (Future Term Discovery): Conversely, if a node (in any state) receives an RPC with a term (args.term) that is greater than its own current_term, it immediately knows it is behind the times. The Raft protocol dictates that it must immediately update its own current_term to match the higher term and, crucially, revert to the Follower state. This is how a cluster comes to a consensus on the current term. A Leader who discovers a new term steps down, and a Candidate who discovers a new term abandons its own election.
3. The “One Vote Per Term” Rule: If the terms match, the node then checks if it has already voted in this term. If its voted_for field is Some(candidate_id) (meaning it has already voted for someone) and the candidate_id is not the one in the current request, it must reject the vote. A node is only allowed to vote once per term.
4. The Log Up-to-Date Check (A Glimpse of the Future): The full Raft algorithm includes one more crucial check: the candidate’s log must be at least as “up-to-date” as the voter’s log. This ensures that a node with a stale log cannot become leader and overwrite newer data. Since we haven’t implemented the log yet, we will temporarily omit this check, but it’s essential to remember for later.

If a request passes all these checks, the node grants its vote. This involves updating its voted_for state, resetting its own election timer (since it has now participated in a valid election activity), and sending back a positive reply.

Implementing the Voting Logic in handle_connection

The logic to handle this RPC belongs in our handle_connection function. First, we need to give this function access to the shared RaftNode state.

1. Update the handle_connection Signature and Call Site

In src/main.rs, modify the handle_connection function to accept the Arc<RwLock<RaftNode>>.

// In src/main.rs

// Add raft_node as a parameter to the function.
async fn handle_connection(
    mut stream: TcpStream,
    store: KvStore,
    raft_node: Arc<RwLock<RaftNode>>,
) -> anyhow::Result<()> {
    // ... function body
}

Now, update the tokio::spawn block in your main function’s loop to pass the cloned raft_node to the handler.

// In src/main.rs -> main() function's loop

    loop {
        let (stream, addr) = listener.accept().await?;
        let store_clone = store.clone();
        let raft_node_clone = raft_node.clone();

        tokio::spawn(async move {
            println!("Accepted new connection from: {}", addr);
            // Pass the raft_node_clone into the handler function.
            if let Err(e) = handle_connection(stream, store_clone, raft_node_clone).await {
                eprintln!("Error handling connection from {}: {}", addr, e);
            }
        });
    }

2. Implement the RequestVote Handler

Now, let’s add the core logic inside handle_connection. We will add a new arm to our match message { ... } block to handle the NodeMessage::RequestVote variant.

// In src/main.rs -> handle_connection()

async fn handle_connection(
    mut stream: TcpStream,
    store: KvStore,
    raft_node: Arc>,
) -> anyhow::Result<()> {
    loop {
        // ... (reading msg_len and deserializing into `message` is unchanged) ...

        match message {
            WireMessage::ClientRequest(request) => {
                // ... (existing client request logic is unchanged) ...
            }
            WireMessage::Node(node_message) => {
                match node_message {

NodeMessage::RequestVote(args) => {
                        println!("[Vote] Received RequestVote RPC: {:?}", args);
                        let mut vote_granted = false;
                        let term;

                        // Acquire a write lock to modify the node's state.
                        {
                            let mut node = raft_node.write().unwrap();

                            // Universal Rule: If RPC term is > currentTerm, update term and become follower.
                            if args.term > node.current_term {
                                println!("[Vote] Discovering new term {}. Stepping down.", args.term);
                                node.current_term = args.term;
                                node.state = NodeState::Follower;
                                node.voted_for = None; // Clear previous vote for the new term.
                            }

                            // Rule 1: Reply false if term < currentTerm
                            if args.term < node.current_term {
                                println!("[Vote] Denying vote: Candidate term {} is older than current term {}.", args.term, node.current_term);
                                vote_granted = false;
                            } 
                            // Rule 2: If voted_for is null or candidateId, and candidate’s log is at least as
                            // up-to-date as receiver’s log, grant vote.
                            else if node.voted_for.is_none() || node.voted_for == Some(args.candidate_id) {
                                // TODO: Add the log up-to-date check here in a future step.

                                println!("[Vote] Granting vote for term {} to candidate {}.", node.current_term, args.candidate_id);
                                vote_granted = true;
                                node.voted_for = Some(args.candidate_id);
                                // Granting a vote is a positive communication. Reset the election timer.
                                node.election_deadline = RaftNode::new_election_deadline();
                            } else {
                                println!("[Vote] Denying vote: Already voted for another candidate in term {}.", node.current_term);
                                vote_granted = false;
                            }

                            // The reply must contain our current term.
                            term = node.current_term;
                        } // Write lock is released here.

                        // Construct and send the reply.
                        let reply = RequestVoteReply { term, vote_granted };
                        let reply_msg = WireMessage::Node(NodeMessage::RequestVoteReply(reply));
                        let reply_bytes = bincode::serialize(&reply_msg)?;

                        stream.write_u64(reply_bytes.len() as u64).await?;
                        stream.write_all(&reply_bytes).await?;
                    }

NodeMessage::Ping => {
                        // ... (existing Ping logic is unchanged) ...
                    }
                    NodeMessage::Pong => {
                        // ... (existing Pong logic is unchanged) ...
                    }
                    // We will add handlers for these other messages in later tasks.
                    _ => {}
                }
            }
            WireMessage::ClientResponse(_) => {
                // ... (existing logic is unchanged) ...
            }
        }
    }
    Ok(())
}

Deconstructing the New Code

• Acquiring a Write Lock: The very first thing we do is acquire a write lock on the RaftNode state. This is essential because we may need to modify current_term, state, and voted_for. The lock is held inside a scoped {...} block to ensure it’s released as soon as possible.
• The Universal Term Update: The if args.term > node.current_term block is the implementation of Raft’s most important rule. It ensures that a node that is behind on the term number immediately catches up and resets its state, which is critical for cluster stability.
• The Voting Rules: The if/else if/else chain implements the decision logic directly. We first check for a stale term, then check if we’ve already voted. If both checks pass, we grant the vote.
• Resetting the Election Timer: When we grant a vote, we call node.election_deadline = RaftNode::new_election_deadline();. This is a crucial detail. By granting a vote, the node is acknowledging that a valid election is in progress. It resets its own timer to give that election a fair chance to produce a leader, preventing it from starting a conflicting election too soon.
• Sending the Reply: After the lock is released, we construct the RequestVoteReply, package it into our WireMessage envelope, serialize it, and send it back to the candidate. The reply always includes our current_term so the candidate can also update itself if it learns of a higher term.

You have now implemented the democratic process of your Raft cluster. Nodes can now receive election ballots, evaluate them according to a strict and consistent set of rules, and cast their votes.

What’s Next?

The ballots have been cast and sent back to the candidate. The final piece of the election puzzle is for the candidate to receive these replies, count the votes, and determine the outcome. In the next task, you will implement the logic for a Candidate to handle incoming RequestVoteReply messages. If it receives votes from a majority of the nodes in the cluster, it will triumph and transition to the coveted Leader state.

Further Reading

• Raft Paper (Section 5.2, “Receiver implementation” for RequestVote RPC): This is the canonical source for the rules you just implemented. It’s an excellent resource to cross-reference.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• The Secret Lives of Data - Raft (Election Section): This interactive visualization is perfect for seeing how nodes evaluate vote requests and how term numbers are used to resolve conflicts.
  • http://thesecretlivesofdata.com/raft/
• Consensus in Distributed Systems: A broader look at the challenges that algorithms like Raft are designed to solve.
  • Wikipedia: Consensus (computer science)

Raft: Tally Votes and Become Leader

Mục tiêu: Implement the logic for a Candidate node in the Raft protocol to process vote replies. The task involves counting incoming votes and, upon securing a majority, transitioning the node’s state to Leader to complete the election process.

---

Of course! Here is a detailed guide to help you complete your current task.

---

You have now built a system where nodes can campaign for leadership and voters can intelligently decide who to support based on the strict rules of the Raft protocol. The ballots (RequestVote RPCs) are out, and the replies are on their way back. The final, exhilarating moment of any election is the vote count. This task is about programming your Candidate to do just that: tally the incoming votes and, upon securing a majority, claim its rightful place as the new Leader of the cluster.

This transition from Candidate to Leader is the successful culmination of the election process. It restores order to the cluster by establishing a single, authoritative source of truth. The logic for counting votes and declaring victory is a cornerstone of Ra-ft’s ability to maintain consensus.

The Power of the Majority

In a distributed system, a majority (also called a quorum) is a critical threshold. For a cluster with N nodes, the majority is defined as (N / 2) + 1. By requiring a candidate to secure votes from a majority of nodes, Raft guarantees that:

1. Only One Leader Can Be Elected Per Term: It’s mathematically impossible for two different candidates in the same term to both receive votes from a majority of the same set of nodes. This prevents a catastrophic “split-brain” scenario.
2. The New Leader is Aware of Past Decisions: Since any two majorities must overlap by at least one node, this ensures that a new leader has received a vote from at least one node that was part of the majority that committed entries in the previous term. This is key to ensuring log consistency, which we’ll explore later.

A candidate starts an election with one vote already cast for itself. Therefore, it needs to win (N / 2) additional votes from its peers to reach the majority threshold.

Implementation Strategy: Counting and Ascending

Our implementation will follow a clear path:

1. Track the Vote Count: We need a place to store the number of votes a candidate has received. A new field, votes_received, in our RaftNode struct is the perfect place for this. This is volatile state, only meaningful while a node is a Candidate.
2. Initialize the Count: When a Follower transitions to a Candidate, we will initialize its votes_received count to 1 (for its own vote).
3. Process Replies: In our handle_connection function, we’ll add logic to process incoming RequestVoteReply messages.
4. The Victory Condition: Inside the reply handler, if a vote was granted, we’ll increment our counter. Then, we’ll check if the counter has reached the majority. If it has, we transition our state to NodeState::Leader and celebrate!

Step 1: Update the RaftNode Struct and Election Start

First, let’s add the votes_received field to our RaftNode struct. We’ll also update the election timer task to properly initialize this field when an election begins.

// In src/main.rs, update the RaftNode struct

/// Contains the core Raft state for a node.
#[derive(Debug)]
struct RaftNode {
    state: NodeState,
    current_term: u64,
    voted_for: Option<u64>,
    election_deadline: Instant,

    // --- Volatile state on candidates ---
    // (Reinitialized after election)

    /// The number of votes this candidate has received in the current term.
    votes_received: u64,
}

// Update the RaftNode::new() function
impl RaftNode {
    fn new() -> Self {
        RaftNode {
            state: NodeState::Follower,
            current_term: 0,
            voted_for: None,
            election_deadline: Self::new_election_deadline(),
            // Initialize votes received to 0.
            votes_received: 0,
        }
    }
    // ... new_election_deadline() is unchanged
}

Now, update the election timer task to set votes_received to 1 when a node becomes a candidate.

// In src/main.rs -> inside the election timer's tokio::spawn block

// Inside the `if node.state == NodeState::Follower` block:
...
node.state = NodeState::Candidate;
node.current_term += 1;
node.voted_for = Some(self_id);

// A new candidate starts with one vote for itself.
node.votes_received = 1;

node.election_deadline = RaftNode::new_election_deadline();
...

Step 2: Implement the RequestVoteReply Handler

Now for the main event. We’ll add the logic to handle_connection to process the replies and check for victory. We’ll need the total size of the cluster to calculate the majority. Let’s calculate it in main and pass it to our handlers.

First, update main to calculate the cluster size and pass it to the connection handler.

// In src/main.rs -> main() function

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... config parsing

    // The total number of nodes in the cluster (including self).
    let cluster_size = 1 + config.peers.len() as u64;
    let majority = (cluster_size / 2) + 1;
    println!("Cluster size: {}, Majority needed: {}", cluster_size, majority);

    // ... raft_node, peer_connections, listener creation

    // Update the main loop to pass `majority` to the handler
    loop {
        let (stream, addr) = listener.accept().await?;
        let store_clone = store.clone();
        let raft_node_clone = raft_node.clone();

        tokio::spawn(async move {
            println!("Accepted new connection from: {}", addr);
            // Pass the majority value into the handler.
            if let Err(e) = handle_connection(stream, store_clone, raft_node_clone, majority).await {
                eprintln!("Error handling connection from {}: {}", addr, e);
            }
        });
    }
}

Next, update the handle_connection function signature.

// In src/main.rs

async fn handle_connection(
    mut stream: TcpStream,
    store: KvStore,
    raft_node: Arc<RwLock<RaftNode>>,
    majority: u64,
) -> anyhow::Result<()> {
    // ...
}

Finally, add the new logic arm for NodeMessage::RequestVoteReply.

// In src/main.rs -> handle_connection() -> match node_message

                match node_message {
                    NodeMessage::RequestVote(args) => {
                        // ... existing RequestVote logic is unchanged ...
                    }

NodeMessage::RequestVoteReply(reply) => {
                        println!("[Vote] Received RequestVoteReply: {:?}", reply);
                        let mut node = raft_node.write().unwrap();

                        // Universal Rule: If reply term is > currentTerm, update term and become follower.
                        if reply.term > node.current_term {
                            println!("[Vote] Discovering new term {} from reply. Stepping down.", reply.term);
                            node.current_term = reply.term;
                            node.state = NodeState::Follower;
                            node.voted_for = None;
                            return Ok(()); // No more processing needed.
                        }

                        // Ignore replies for a different term or if we are no longer a candidate.
                        if reply.term != node.current_term || node.state != NodeState::Candidate {
                            println!("[Vote] Ignoring stale or irrelevant vote reply.");
                            return Ok(());
                        }

                        if reply.vote_granted {
                            node.votes_received += 1;
                            println!(
                                "[Vote] Vote granted! Total votes: {}/{}",
                                node.votes_received, majority
                            );

                            // Check for victory!
                            if node.votes_received >= majority {
                                println!(
                                    "👑 [Leader] Election WON! Becoming leader for term {}.",
                                    node.current_term
                                );
                                node.state = NodeState::Leader;

                                // TODO: As the new leader, we must immediately send an initial
                                // heartbeat to all peers to assert authority. This will be
                                // implemented in the next task.
                            }
                        }
                    }

NodeMessage::Ping => {
                        // ... existing Ping logic is unchanged ...
                    }
                    NodeMessage::Pong => {
                        // ... existing Pong logic is unchanged ...
                    }
                    // ... other message types ...
                }

Deconstructing the New Logic

• Passing majority: We calculate the majority count once at startup and pass it down into every connection handler. This is efficient and ensures the handler has the context it needs to make the victory decision.
• The Universal Term Check (Again!): The first thing we do upon receiving any RPC reply is check the term. If a voter replies with a higher term, it means another, more up-to-date election is already in progress. Our candidate has lost. It must immediately step down and become a Follower to respect the new term.
• State and Term Validation: We also check if we are still a Candidate in the correct term. It’s possible to receive a late reply after we’ve already lost or won the election. This check ensures we only count relevant votes.
• Increment and Check: If the vote was granted, we increment node.votes_received and immediately check if we’ve crossed the majority threshold.
• The Ascension: node.state = NodeState::Leader; is the single line that marks the successful outcome of the election. Our node is now the leader! The println! with the crown emoji provides a clear and satisfying log message that will be invaluable for observing the system’s behavior.

You have now implemented the complete cycle of a Raft leader election. A follower can time out, become a candidate, campaign for votes, receive votes, and ascend to leadership. This is a monumental achievement in building a distributed system.

What’s Next?

A newly crowned leader cannot rest on their laurels. Their very first act must be to assert their authority over the cluster. If they remain silent, other followers will have their own election timers expire and will start new, disruptive elections. In the next task, you will implement the logic for the new Leader to immediately cancel its own election timer and begin sending periodic heartbeats (AppendEntries RPCs) to all followers, solidifying its rule and bringing stability back to the cluster.

Further Reading

• Raft Paper (Section 5.2, Leader Election): The paper describes the process of counting votes and what happens when a candidate wins.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• The Secret Lives of Data - Raft (Election Won): The visualization perfectly illustrates a candidate receiving a majority of votes and turning into the leader.
  • http://thesecretlivesofdata.com/raft/
• Quorum in Distributed Systems: A deeper dive into the concept of majorities and why they are fundamental to consistency.
  • Wikipedia: Quorum (distributed computing)

Implement Raft Leader Heartbeat Task

Mục tiêu: Create a dedicated asynchronous task for the Raft leader to periodically send empty AppendEntries RPCs as heartbeats to all followers, maintaining its authority and preventing new elections.

---

Of course! Here is a detailed guide to help you complete your current task.

---

You have achieved a monumental milestone in your project: your Raft nodes can now successfully run an election, count votes, and promote a candidate to the status of Leader. This is the moment the cluster regains its stability and a single source of truth is established. However, a leader’s reign is fragile. If they remain silent, followers will assume they have failed, their election timers will expire, and the cluster will be plunged into another round of chaotic elections.

A newly-crowned leader’s first and most crucial responsibility is to immediately assert their authority. They must transition from the role of a campaigner to that of a ruler, and this is done by sending out periodic heartbeats. This task is about implementing that fundamental duty of leadership.

The Leader’s Mandate: Assert Authority, Prevent Chaos

A Leader in Raft has two immediate jobs upon their ascension:

1. Stop Campaigning: A leader is no longer a candidate and should not be subject to its own election timer. In our current design, this happens implicitly and elegantly. Our election timer task is programmed to only trigger an election if node.state == NodeState::Follower. Since our node’s state is now Leader, the timer task will check its state, see that it’s not a follower, and simply go back to sleep. The election timer is effectively “canceled” without any extra code.
2. Start Ruling: The leader must begin sending periodic AppendEntries RPCs to all other nodes. When these RPCs contain no log entries, they serve as heartbeats. These heartbeats are the signals that prove the leader is alive and in charge, and they are the mechanism that followers use to reset their own election timers, thus preventing them from starting new elections.

Implementation Strategy: A Dedicated Heartbeat Task

To cleanly separate the leader’s responsibilities, we will create a new, dedicated, long-running task in our main function. This task will run in a perpetual loop, but it will only take action if and only if it detects that the node’s current state is Leader.

This state-driven approach is robust and simple. The task lies dormant for followers and candidates, and automatically activates the moment a node wins an election, ensuring that heartbeats begin immediately.

Implementing the Leader’s Heartbeat Task

We will now add a new tokio::spawn block to our main function. This task will need access to our shared state and configuration, so we will clone the necessary Arc pointers and values and move them into the task’s closure.

Here are the changes for your src/main.rs file. This new block of code should be added in the main function, alongside your other spawned tasks (like the election timer).

// In src/main.rs -> main() function

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... (config parsing, cluster_size, majority, store, raft_node, etc.)

    // ... (Peer Connection Task)

    // ... (Election Timer Task)

// --- Leader Heartbeat Logic ---
    let heartbeat_node = raft_node.clone();
    let heartbeat_peers = config.peers.clone();
    let heartbeat_connections = peer_connections.clone();
    let self_id = config.id;
    tokio::spawn(async move {
        // This interval should be significantly shorter than the election timeout.
        let heartbeat_interval = Duration::from_millis(50);

        loop {
            // Wait for the heartbeat interval before proceeding.
            tokio::time::sleep(heartbeat_interval).await;

            let mut current_term = 0;
            let mut is_leader = false;

            // --- LOCK, READ, and RELEASE ---
            // We use a read lock to check if we are the leader.
            {
                let node = heartbeat_node.read().unwrap();
                if node.state == NodeState::Leader {
                    is_leader = true;
                    // Copy the term needed for the RPC.
                    current_term = node.current_term;
                }
            } // Read lock is dropped here.

            // --- ACT (NETWORK I/O) ---
            // If we are the leader, send heartbeats. This is outside the lock.
            if is_leader {
                // Construct the AppendEntries RPC for a heartbeat.
                // Log-related fields are empty/default for now.
                let args = AppendEntriesArgs {
                    term: current_term,
                    leader_id: self_id,
                };
                let rpc_message = WireMessage::Node(NodeMessage::AppendEntries(args));
                let message_bytes = match bincode::serialize(&rpc_message) {
                    Ok(bytes) => bytes,
                    Err(e) => {
                        eprintln!("[Heartbeat] Failed to serialize AppendEntries RPC: {}", e);
                        continue; // Skip this heartbeat round.
                    }
                };

                // Broadcast the heartbeat to all peers.
                for (peer_id, _peer_addr) in &heartbeat_peers {
                    let mut connections = heartbeat_connections.write().unwrap();
                    if let Some(stream) = connections.get_mut(peer_id) {
                        // Send the length-prefixed message.
                        if let Err(e) = stream.write_u64(message_bytes.len() as u64).await {
                             eprintln!("[Heartbeat] Failed to send heartbeat length to peer {}: {}", peer_id, e);
                             continue;
                        }
                        if let Err(e) = stream.write_all(&message_bytes).await {
                            eprintln!("[Heartbeat] Failed to send heartbeat payload to peer {}: {}", peer_id, e);
                        }
                    }
                }
            }
        }
    });

// ... (Main server loop for accepting connections is unchanged)
}

Deconstructing the New Code

Let’s walk through the logic of our new heartbeat task:

1. Cloning Dependencies: We start by cloning the Arcs and values needed (raft_node, peers, peer_connections, self_id) so the new task can own its own copies and references.
2. heartbeat_interval: We define a short duration (e.g., 50 milliseconds). The Raft paper recommends that this interval be at least an order of magnitude smaller than the election timeout (150-300ms) to ensure heartbeats are sent reliably before any follower can time out.
3. The loop: The task runs forever. In each iteration, it first sleeps. This prevents a tight, CPU-bound loop and establishes the rhythm of the heartbeats.
4. The State Check: The core of the logic is the if node.state == NodeState::Leader check. This is performed under a read lock. This lock is held for the briefest possible moment, just long enough to check the state and copy the current_term.
5. Lock Release: The read lock is released as soon as the {...} block ends. This is crucial for performance and adheres to our “Don’t Hold Locks Across .await” principle. All the slow network I/O happens after the lock is gone.
6. The Broadcast: If is_leader is true, we proceed to construct and send the AppendEntries message. This logic is very similar to how you broadcasted RequestVote RPCs: serialize once, then iterate through the peers, acquiring a lock on the connection map just long enough to send the data down the TcpStream.

You have now endowed your leader with its primary tool of governance. It will tirelessly broadcast its presence, ensuring the cluster remains stable and united under its rule. If you run multiple nodes now, you will see the newly elected leader periodically sending out AppendEntries messages in its logs.

What’s Next?

The leader is now speaking, but are the followers listening? The heartbeats are being sent, but they are currently being ignored by the receiving nodes. The final piece of this puzzle is to implement the follower’s side of this interaction. In the next task, you will modify the handle_connection function to process these incoming AppendEntries heartbeats. When a follower receives a valid heartbeat, it will acknowledge the leader’s authority by resetting its own election timer, thus completing the cycle of leadership and followership that keeps the Raft cluster alive and healthy.

Further Reading

• Raft Paper (Section 5.2, final paragraph on leadership): The paper explicitly states that upon election, a leader begins sending AppendEntries RPCs.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• Tokio Tutorial: Time, delays, and intervals: A great resource for understanding how to manage time in asynchronous Rust applications.
  • https://tokio.rs/tokio/tutorial/time
• Leader-Based Consensus: A broader look at the category of consensus algorithms that rely on electing a leader, with Raft and Paxos being the most famous examples.
  • Wikipedia: Paxos (computer science) (Raft is generally considered more understandable than Paxos, but the core ideas are related).

Implement Raft Follower Logic for Leader Heartbeats

Mục tiêu: Implement the logic for a follower node in a Raft cluster to handle AppendEntries RPCs, which act as heartbeats from the leader. This involves validating the leader’s term, updating its own state if necessary, and resetting its election timer to prevent new elections and maintain cluster stability.

---

Of course! Here is a detailed guide to help you complete your current task.

---

You have successfully brought your leader to life. In the previous task, you programmed the newly elected leader to begin its most fundamental duty: periodically broadcasting AppendEntries heartbeats to assert its authority. The leader is now speaking, but is anyone listening? The stability of the entire Raft cluster hinges on the followers’ response to this call.

This final task of the leader election step is to complete that communication loop. You will now implement the logic for a follower to receive, validate, and act upon a leader’s heartbeat. The primary and most critical action is for the follower to reset its own election timer, acknowledging the leader’s authority and canceling any plans it had to start a new election. This is the mechanism that maintains a leader’s reign and brings stability to the cluster.

The Follower’s Pledge of Allegiance

A follower’s life is one of patient observation. It starts its election timer and waits. The AppendEntries RPC from the leader is the signal it’s waiting for. When this message arrives, the follower must evaluate it based on a few simple but strict rules derived directly from the Raft paper:

1. Is the Leader Legitimate? (Term Check): The first check is always on the term number. If the leader’s term (args.term) is less than the follower’s current_term, the message is from a stale, deposed leader from a past term. The follower must reject this message to prevent old leaders from causing chaos.
2. Have I Found a New Leader? (Term Update): If the leader’s term is greater than the follower’s current_term, the follower knows it’s behind the times. It must immediately update its own term to match the leader’s and revert to the follower state (if it was a candidate). This is how a cluster converges on a single, correct term.
3. Acknowledging the Ruler: If the terms match, the message is from the valid, current leader. The follower accepts the message, and most importantly, resets its election timer. This single action is the follower’s pledge of allegiance. It says, “I have heard from my leader, all is well. I will not start an election.”

Implementation Strategy

We will implement this logic within our handle_connection function. It will be a new arm in the match node_message block, specifically for NodeMessage::AppendEntries.

Step 1: Enhance the RaftNode State (Optional but Recommended)

To make our state more explicit, let’s add a field to our RaftNode struct to track who the current leader is. This isn’t strictly necessary for this task but is excellent practice and will be useful for debugging and future features like client redirection.

// In src/main.rs, update the RaftNode struct

/// Contains the core Raft state for a node.
#[derive(Debug)]
struct RaftNode {
    state: NodeState,
    current_term: u64,
    voted_for: Option<u64>,
    election_deadline: Instant,
    votes_received: u64,

    // --- Volatile state on all servers ---

    /// The ID of the current leader, if known.
    current_leader: Option<u64>,
}

// Update the RaftNode::new() function
impl RaftNode {
    fn new() -> Self {
        RaftNode {
            state: NodeState::Follower,
            current_term: 0,
            voted_for: None,
            election_deadline: Self::new_election_deadline(),
            votes_received: 0,
            // A node starts up without knowing who the leader is.
            current_leader: None,
        }
    }
    // ... new_election_deadline() is unchanged
}

We should also ensure that when a node steps down upon discovering a higher term, it clears its knowledge of the old leader.

// In src/main.rs -> handle_connection -> RequestVote handler
// When stepping down, also clear the current_leader field.
if args.term > node.current_term {
    // ...
    node.state = NodeState::Follower;
    node.voted_for = None;
    node.current_leader = None; // Add this line
}

// In src/main.rs -> handle_connection -> RequestVoteReply handler
// Do the same here.
if reply.term > node.current_term {
    // ...
    node.state = NodeState::Follower;
    node.voted_for = None;
    node.current_leader = None; // And this line
}

Step 2: Implement the AppendEntries Handler

Now, let’s add the core logic to handle_connection to process the heartbeat.

// In src/main.rs -> handle_connection() -> match node_message

                match node_message {
                    NodeMessage::RequestVote(args) => {
                        // ... existing RequestVote logic is unchanged ...
                    }
                    NodeMessage::RequestVoteReply(reply) => {
                        // ... existing RequestVoteReply logic is unchanged ...
                    }

NodeMessage::AppendEntries(args) => {
                        let mut success = false;
                        let term;

                        // Acquire a write lock to potentially modify state.
                        {
                            let mut node = raft_node.write().unwrap();
                            term = node.current_term;

                            // Universal Rule: If RPC term is > currentTerm, update term and become follower.
                            if args.term > node.current_term {
                                node.current_term = args.term;
                                node.state = NodeState::Follower;
                                node.voted_for = None;
                                node.current_leader = None;
                            }

                            // Rule 1: Reply false if term < currentTerm.
                            if args.term < node.current_term {
                                println!(
                                    "[Heartbeat] Rejecting AppendEntries from stale leader with term {}.",
                                    args.term
                                );
                                success = false;
                            } else {
                                // This is a valid heartbeat from the current or a new leader.
                                // If we were a Candidate, we recognize the new leader and step down.
                                if node.state == NodeState::Candidate {
                                    println!(
                                        "[Heartbeat] Discovering leader {} while a candidate. Stepping down.",
                                        args.leader_id
                                    );
                                    node.state = NodeState::Follower;
                                }

                                // Update who we think the leader is.
                                node.current_leader = Some(args.leader_id);

                                // THIS IS THE KEY: We reset our election timer because we have
                                // received a valid communication from the leader.
                                println!("[Heartbeat] Received heartbeat from leader {}. Resetting timer.", args.leader_id);
                                node.election_deadline = RaftNode::new_election_deadline();
                                success = true;

                                // TODO: In the next step, we will also handle appending log entries here.
                            }

                            // The reply must contain our current term.
                            term = node.current_term;
                        } // Write lock is released here.

                        // Construct and send the reply.
                        let reply = AppendEntriesReply { term, success };
                        let reply_msg = WireMessage::Node(NodeMessage::AppendEntriesReply(reply));
                        let reply_bytes = bincode::serialize(&reply_msg)?;\

                        stream.write_u64(reply_bytes.len() as u64).await?;\
                        stream.write_all(&reply_bytes).await?;
                    }

NodeMessage::Ping => { /* ... */ }
                    NodeMessage::Pong => { /* ... */ }
                    // We will add handlers for AppendEntriesReply in the next step.
                    _ => {}
                }

Deconstructing the New Logic

• Stateful Validation: The logic block strictly follows the Raft rules. It first checks for a stale leader (args.term < node.current_term) and rejects the message. This prevents old leaders from disrupting the cluster.
• Stepping Down: If the node was a Candidate and receives a valid heartbeat, it knows it has lost the election. It gracefully steps down and becomes a Follower of the new leader, ensuring a smooth transfer of power.
• The Critical Reset: The line node.election_deadline = RaftNode::new_election_deadline(); is the entire point of this task. It pushes the timer’s deadline into the future, effectively saying, “I will not start an election for at least another 150-300ms.” This is what keeps the cluster stable under a single leader.
• Sending a Reply: The follower always sends an AppendEntriesReply back to the leader. This is how the leader knows which followers are alive and successfully receiving its heartbeats. In a future step, this reply will be crucial for determining when log entries are safely replicated.

A Major Milestone: The Full Election Cycle

Congratulations! You have now implemented a complete, working Raft leader election mechanism from scratch. This is a significant and complex achievement in distributed systems engineering. Your cluster of nodes can now:

1. Detect when a leader has failed using a randomized election timer.
2. Start an election, with a follower transitioning to a candidate.
3. Campaign for leadership by broadcasting RequestVote RPCs.
4. Evaluate vote requests according to strict rules and grant votes.
5. Tally the votes and, upon receiving a majority, ascend to leadership.
6. Maintain its leadership by sending periodic heartbeats.
7. Acknowledge the leader’s authority by resetting election timers upon receiving a heartbeat.

Your system is now fault-tolerant with respect to its leadership.

What’s Next?

With a stable leader in place, the cluster is finally ready to do its real work: reliably managing user data. The leader is the single point of entry for all changes to the system. In the next major step, Integrating Data Replication with the Raft Consensus Log, you will evolve the AppendEntries RPC from a simple heartbeat into a powerful data replication mechanism. You will learn how the leader adds client commands to a log and uses this same RPC to ensure that a majority of the cluster agrees on every single change before it is applied to the key-value store.

Further Reading

• Raft Paper (Section 5.1 & 5.3, AppendEntries RPC): The paper provides the definitive rules for how followers must handle this RPC.
  • Raft Paper: In Search of an Understandable Consensus Algorithm
• The Secret Lives of Data - Raft (Heartbeats and Replication): The visualization shows exactly how heartbeats prevent timeouts and how leaders send messages to followers.
  • http://thesecretlivesofdata.com/raft/
• Leader Election in Distributed Systems: A broader overview of why leader election is a fundamental problem and other algorithms used to solve it.
  • Wikipedia: Leader election