Define Raft Log Entry and Command Structures in Rust

Mục tiêu: Define the core data structures for the Raft replicated log in Rust. Create a Command enum for state-mutating operations and a LogEntry struct containing the command and its associated term number, deriving necessary traits for serialization and cloning.

An outstanding achievement! You have successfully implemented the complete Raft leader election algorithm. Your cluster of nodes can now gracefully handle leader failures, elect a new leader, and maintain stability through heartbeats. This is the bedrock of any fault-tolerant system. With a stable leader in place, the cluster is finally ready to perform its primary function: reliably managing and replicating user data.

Leader election answers the question, “Who gets to make decisions?”. Now, we will address the next critical question: “How does the cluster agree on what those decisions are?”. The answer lies in the central data structure of the Raft algorithm: the replicated log.

The Log: The Immutable Ledger of a Distributed System

Think of the replicated log as the official, ordered history of every change made to your key-value store. It is an append-only sequence of commands. The Leader’s primary job, besides sending heartbeats, is to receive commands from clients, write them into its log as new entries, and then convince a majority of its followers to write the exact same entries to their logs in the exact same order. Once an entry is safely stored on a majority of nodes, it is considered “committed” and can be applied to the actual key-value HashMap.

This log is the single source of truth for the entire system. Our first step in building this mechanism is to define the structure of a single entry in that log.

Defining a State Machine Command

First, let’s be precise about what a “command” is. A command is any operation that changes the state of our key-value store. Looking at our client protocol (Request enum), we have three variants: Get, Set, and Delete. However, a Get operation is read-only; it doesn’t modify the state machine, so it does not need to be replicated through the Raft log.

To model this correctly, let’s create a new, dedicated enum that represents only the state-mutating commands. This is a clean design that separates the concerns of the client-facing protocol from the internal state machine logic.

Add the following Command enum definition to src/main.rs. A good place is near your other Raft-related structs and enums.

// In src/main.rs

use serde::{Deserialize, Serialize};

// ... other structs and enums ...

/// Represents a command that will be applied to the state machine (the KvStore).
/// Only commands that modify state need to be replicated in the Raft log.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum Command {
    Set { key: String, value: String },
    Delete { key: String },
}

Defining the `LogEntry`

Now we can define the LogEntry itself. According to the Raft protocol, every entry in the log must contain two crucial pieces of information:

The Command: The actual operation to be executed, which we just defined with our Command enum.
The Term: The leader’s term number when the entry was created. This is a critical piece of metadata used to detect inconsistencies in logs between nodes after a leader change.

Let’s create the LogEntry struct. Add this code right after your new Command enum in src/main.rs.

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

/// An entry in the Raft replicated log.
/// Each entry contains a command to be applied to the state machine,
/// along with the term number of the leader that created it.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct LogEntry {
    /// The term in which this log entry was created.
    pub term: u64,

    /// The command to be executed by the state machine.
    pub command: Command,
}

Let’s break down why this simple struct is so powerful:

term: u64: This field acts as a consistency check. When a new leader takes over, it uses the term numbers stored in the log entries to find the point where its log and a follower’s log diverge. Any entries from a previous, failed leader with a different term are considered suspect and may need to be overwritten to ensure strict consistency.
command: Command: This is the “payload” of the log entry. It’s the instruction—the “what to do”—that will eventually be executed. When a log entry is committed, the node will take this command and apply it to its local KvStore instance (e.g., calling store.set(...)).
#[derive(..., Clone)]: We derive Clone because the leader will need to create copies of its log entries to include them in the AppendEntries RPCs it sends to each follower.
#[derive(..., Serialize, Deserialize)]: We derive Serde’s traits because these LogEntry objects will be serialized and sent over the network. In a later step, they will also be serialized to be persisted on disk.

You have now created the fundamental building block of data replication in Raft. This LogEntry is the “atom” of your distributed system’s history. Every change to your key-value store will be encapsulated in one of these structs before it is replicated and applied.

What’s Next?

With the blueprint for a single log entry defined, the next logical step is to create the log itself. In the next task, you will add a new field, log: Vec<LogEntry>, to your RaftNode struct. This vector will be the in-memory representation of the replicated log, the ordered sequence of all commands that the node has accepted.

Raft: Implement the Replicated Log

Mục tiêu: Add the replicated log, a core component of Raft’s persistent state, to the RaftNode struct using a Vec. Initialize the log as an empty vector in the node’s constructor.

In the last task, you meticulously defined the LogEntry struct, the fundamental building block of your distributed system’s history. Each LogEntry is a self-contained record of a command and the term in which it was issued. This is the “atom” of your system’s memory. Now, it’s time to create the structure that will hold these atoms together in a precise, ordered sequence—the Raft log itself.

The replicated log is the single most important data structure in Raft. It is the definitive, ordered chronicle of every state change the system has ever agreed upon. By ensuring that a majority of nodes have the exact same log, Raft guarantees that they will all execute the same commands in the same order, and therefore arrive at the exact same state.

Choosing the Right Data Structure: `Vec<LogEntry>`

The log in Raft has two key characteristics: it is ordered, and it is append-only. The sequence of entries is critically important, and new entries are always added to the end. In Rust, the perfect data structure to model this behavior is the Vec<T> (a vector).

Ordered: A Vec maintains the insertion order of its elements. The element at index i will always come before the element at index i+1. This maps directly to the concept of log indices in Raft.
Append-Only (by convention): Vec provides a highly efficient push() method to add new elements to the end, which aligns perfectly with how a Raft leader appends new commands to its log. While a Vec can be modified in other ways, our Raft implementation will treat it as an append-only structure.

Adding the Log to Your Node’s State

We will now add the log to the RaftNode struct. According to the Raft paper, the log is part of the persistent state—meaning it must be saved to stable storage before a node can acknowledge that an entry has been replicated. While we’ll tackle the disk I/O in a later step, we must first add the in-memory representation to our state struct.

Let’s modify the RaftNode struct in src/main.rs. We’ll add the new log field right alongside current_term and voted_for, as these all constitute the persistent state.

// In src/main.rs

// Be sure you have your LogEntry struct defined from the previous task.
use crate::LogEntry; 

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

/// Contains the core Raft state for a node.
#[derive(Debug)]
struct RaftNode {
    state: NodeState,

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

    /// The latest term the server has seen.
    current_term: u64,

    /// The candidateId that received a vote in the current term.
    voted_for: Option<u64>,

    /// The log of commands, replicated across all nodes.
    /// This is the heart of the Raft consensus algorithm.
    log: Vec<LogEntry>,

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

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

    /// The absolute time when this node's election timer will fire.
    election_deadline: Instant,

    // --- Volatile state on candidates ---

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

Initializing the Log

Every node, when it first starts, should begin with an empty log. We need to update our RaftNode::new() constructor to initialize our new field correctly.

Here is the highlighted change for the RaftNode::new() function:

// In src/main.rs

impl RaftNode {
    /// Creates a new RaftNode with default initial state.
    fn new() -> Self {
        RaftNode {
            state: NodeState::Follower,
            current_term: 0,
            voted_for: None,

// The log always starts empty.
            log: Vec::new(),

election_deadline: Self::new_election_deadline(),
            votes_received: 0,
            current_leader: None,
        }
    }

    // ... new_election_deadline() is unchanged ...
}

With this simple change, every RaftNode instance created in your main function will now contain an empty vector, ready to store the history of commands that will define the state of your key-value store.

A Note on Log Indexing

The Raft paper uses 1-based indexing for log entries for clarity in its mathematical proofs (i.e., the first entry is at index 1). Rust’s Vec, like arrays in most programming languages, uses 0-based indexing. This is a common and simple implementation detail to manage. When the time comes, we will simply be mindful that Raft’s “log index i” corresponds to our log[i-1]. For now, all you need to know is that we have an ordered, empty container ready to go.

You have now successfully integrated the central data structure for data replication into your node’s state. You have a container for your replicated history, the ledger that will ensure consistency across your entire cluster.

What’s Next?

With the log in place, the next logical step is to start filling it. A leader needs a way to send its log entries to its followers. In the next task, you will modify the AppendEntries RPC to carry a list of new LogEntry items. This will transform the RPC from a simple heartbeat into the powerful data replication mechanism that is the core of Raft.

Implement Full AppendEntries RPC for Log Replication

Mục tiêu: Enhance the Raft implementation by upgrading the AppendEntries RPC. This involves adding log replication fields (prev_log_index, prev_log_term, entries, leader_commit) to the AppendEntriesArgs struct and updating the leader’s heartbeat logic to send this complete payload to followers.

Excellent work! In the previous task, you successfully integrated the log, a Vec<LogEntry>, into your node’s state. This vector is the ledger that will chronicle the entire history of your distributed key-value store. Right now, however, it’s an isolated, in-memory list. To achieve consensus, the leader’s log must be faithfully replicated to its followers. The vehicle for this replication is the AppendEntries RPC.

Currently, your AppendEntries RPC serves as a simple heartbeat, a message devoid of any real data. It’s time to upgrade it into the powerful, data-carrying workhorse that is the heart of Raft’s replication mechanism. In this task, we will modify the AppendEntriesArgs struct to include the fields necessary to send new log entries to followers, transforming it from a simple pulse into a rich, informative message.

The Dual Role of `AppendEntries`

The brilliance of the AppendEntries RPC lies in its dual purpose. It seamlessly handles both liveness checks and data replication:

As a Heartbeat: When the leader has no new commands from clients, it sends an AppendEntries RPC with an empty list of entries. This serves as the “I’m still alive” signal that followers use to reset their election timers.
As a Replication Message: When a leader receives a command from a client, it adds it to its log and then sends an AppendEntries RPC containing that new entry (or a batch of them) to its followers.

This single RPC type simplifies the protocol and the implementation, as both functions are handled by the same logic.

Activating the Full `AppendEntries` Payload

In a previous step, we defined the AppendEntriesArgs struct but commented out the fields related to log replication to keep things simple. Now, it’s time to uncomment them and bring the full power of the RPC to life.

Here is the updated definition for AppendEntriesArgs in src/main.rs.

// In src/main.rs

// ... (RequestVoteArgs and Reply structs) ...

/// The payload for an AppendEntries RPC.
/// Used as a heartbeat (with an empty `entries` vector) and for log replication.
#[derive(Serialize, Deserialize, Debug, Clone)] // Add Clone!
pub struct AppendEntriesArgs {
    /// The leader's term.
    pub term: u64,
    /// The ID of the leader.
    pub leader_id: u64,

    /// The index of the log entry immediately preceding the new ones.
    /// Followers use this for a consistency check.
    pub prev_log_index: u64,

    /// The term of the `prev_log_index` entry.
    pub prev_log_term: u64,

    /// The new log entries to store on the follower.
    /// This is empty for heartbeats, but contains entries for replication.
    pub entries: Vec<LogEntry>,

    /// The leader's commit index. Followers use this to know which entries
    /// are safe to apply to their state machines.
    pub leader_commit: u64,
}

// ... (AppendEntriesReply struct) ...

Important: Notice that we also added Clone to the derive macro. This will be necessary because we will be creating copies of this struct to send to multiple peers.

Deconstructing the New Fields

These new fields are the core of Raft’s safety and consistency guarantees. Let’s understand the role of each one:

prev_log_index: u64 and prev_log_term: u64: These two fields are a “consistency handshake.” Before a follower appends the new entries, it first checks its own log at prev_log_index. If an entry exists there and its term matches prev_log_term, the follower knows its log is consistent with the leader’s up to that point and can safely append the new entries. If the check fails, it signals a log inconsistency, and the follower rejects the request. This mechanism is how Raft repairs logs after leader changes.
entries: Vec<LogEntry>: This is the payload. It’s a vector of the new LogEntry structs that the leader wants the follower to append to its log. For a heartbeat, this vector will simply be empty.
leader_commit: u64: This field is how the leader informs followers about which entries have been successfully replicated on a majority of nodes and are therefore “committed.” Once a follower learns that an entry is committed, it knows it’s safe to apply that entry’s command to its local state machine (our KvStore).

Updating the Node’s State and Heartbeat Task

To support the leader_commit field, we first need to track the commit index on our RaftNode. This is volatile state that gets re-calculated on startup. Let’s add it.

// In src/main.rs -> RaftNode struct

#[derive(Debug)]
struct RaftNode {
    // ... (state, current_term, voted_for, log) ...

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

    /// Index of the highest log entry known to be committed.
    /// Initialized to 0, increases monotonically.
    commit_index: u64,

    // ... (current_leader, election_deadline, votes_received) ...
}

// In src/main.rs -> RaftNode::new()
impl RaftNode {
    fn new() -> Self {
        RaftNode {
            // ... (other fields) ...
            commit_index: 0,
            current_leader: None,
        }
    }
    // ...
}

Now, our leader’s heartbeat task, which constructs and sends the AppendEntries RPC, will fail to compile because it’s not providing the new fields. We must update it to construct the full AppendEntriesArgs struct. For a heartbeat, entries will be empty, and the other fields will reflect the leader’s current log status.

// In src/main.rs -> main() -> Leader Heartbeat Logic

tokio::spawn(async move {
    let heartbeat_interval = Duration::from_millis(50);

    loop {
        tokio::time::sleep(heartbeat_interval).await;

// We need to read several pieces of state to build the RPC.
        let mut rpc_args: Option<AppendEntriesArgs> = None;

        // --- LOCK, READ, and RELEASE ---
        {
            let node = heartbeat_node.read().unwrap();
            if node.state == NodeState::Leader {
                // Determine prev_log_index and prev_log_term from the leader's log.
                let prev_log_index = node.log.len() as u64;
                let prev_log_term = if prev_log_index > 0 {
                    // Raft log indices are 1-based, Vec is 0-based.
                    node.log[(prev_log_index - 1) as usize].term
                } else {
                    0
                };

                // Construct the full AppendEntriesArgs for the heartbeat.
                let args = AppendEntriesArgs {
                    term: node.current_term,
                    leader_id: self_id,
                    prev_log_index,
                    prev_log_term,
                    // Entries are empty for a heartbeat.
                    entries: Vec::new(),
                    leader_commit: node.commit_index,
                };
                rpc_args = Some(args);
            }
        } // Read lock is dropped here.

        // --- ACT (NETWORK I/O) ---
        // If we are the leader (determined by rpc_args being Some), send heartbeats.
        if let Some(args) = rpc_args {
            // We now use the cloned `args` struct.
            let rpc_message = WireMessage::Node(NodeMessage::AppendEntries(args.clone()));
            let message_bytes = match bincode::serialize(&rpc_message) {
                // ... (error handling) ...
            };

            // 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 message logic remains the same) ...
                }
            }
        }

}
});

You have now successfully upgraded your AppendEntries RPC from a simple heartbeat to a fully-featured replication message. The communication channel is now in place to send the entire history of your key-value store across the cluster.

What’s Next?

The RPC is ready to carry data, but the leader isn’t putting any data into it yet. The trigger for replication is a client request that modifies state (like a Set or Delete). In the next task, you will implement the first part of this flow: when a leader receives a client request, it will create a new LogEntry and append it to its own local log, preparing it for replication.

Raft Leader: Append Client Commands to Log

Mục tiêu: Implement the logic for a Raft leader to handle client write requests. Upon receiving a command, the leader appends a new entry to its own log, preparing it for replication. Non-leader nodes should reject these requests.

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

You have made fantastic progress by preparing the AppendEntries RPC to be a fully-featured data replication vehicle. The communication channels are now capable of carrying the history of your key-value store. However, a channel is useless without a sender, and right now, the leader has no process for putting new commands into that channel.

The entire Raft replication process is kicked off by a single event: a client sending a command (like Set or Delete) to the leader. This is the moment a new piece of history is created. In this task, you will implement the leader’s first crucial action upon receiving such a command: it will create a new LogEntry and append it to its own internal log, preparing it for replication to the rest of the cluster.

The Leader’s Exclusive Responsibility

A core principle of Raft that simplifies the entire consensus process is that only the leader can modify the system’s state. All write requests (Set, Delete) from clients must go through the leader. If a client mistakenly sends a write request to a Follower or a Candidate, that node must reject the request. This centralization of power prevents conflicting changes and is the foundation for creating a single, consistent history of operations.

The process for a leader handling a client write request is:

Receive the client’s command.
Append the command to its own log as a new entry. (This is our current task).
Send the new entry to its followers via AppendEntries RPCs. (Next task).
Wait until a majority of followers have successfully replicated the entry. (Future task).
Consider the entry “committed.”
Apply the command from the committed log entry to its state machine (the KvStore HashMap).
Send a success response back to the client.

Notice the most important detail: the leader does not apply the command to its KvStore immediately. It only records the intent to do so in its log. This “propose, then commit” pattern is fundamental to Raft’s safety.

Implementing the Leader’s Logic

The perfect place to add this logic is in our handle_connection function, inside the WireMessage::ClientRequest match arm. We’ll check if the current node is the leader. If it is, we’ll append to the log. If it isn’t, we’ll return an error.

Here are the changes for your handle_connection function in src/main.rs. We will focus on modifying the match request block.

// In src/main.rs -> handle_connection function

async fn handle_connection(
    mut stream: TcpStream,
    store: KvStore,
    raft_node: Arc<RwLock<RaftNode>>,
    majority: u64,
) -> anyhow::Result<()> {
    loop {
        // ... (reading and deserializing the message is unchanged) ...

        match message {
            WireMessage::ClientRequest(request) => {

let response = match request {
                    // Read-only operations like Get can be handled immediately
                    // without going through the Raft log.
                    // (For stronger consistency models like linearizability, Get
                    // would also go through the leader, but this is a valid simplification for now).
                    Request::Get { key } => {
                        let result = store.get(&key);
                        ClientResponse::Success(result)
                    }

                    // Write operations must go through the Raft consensus protocol.
                    Request::Set { key, value } => {
                        // 1. Check if this node is the leader.
                        let is_leader;
                        { // Scoped read lock
                            let node = raft_node.read().unwrap();
                            is_leader = node.state == NodeState::Leader;
                        } // Read lock is released here.

                        if is_leader {
                            // 2. If we are the leader, create a command and append it to our log.
                            let command = Command::Set { key, value };

                            // Acquire a write lock to modify the log.
                            let mut node = raft_node.write().unwrap();
                            let new_entry = LogEntry {
                                term: node.current_term,
                                command,
                            };
                            node.log.push(new_entry);

                            println!("[Leader] Appended new client command to log at index {}.", node.log.len());

                            // IMPORTANT: We do NOT send a response to the client yet.
                            // The response will be sent only after the entry is committed.
                            // We will implement this logic in a future task. For now,
                            // the connection will just wait.
                            continue; // Skip sending a response for now.

                        } else {
                            // 3. If we are not the leader, reject the request.
                            // A more advanced implementation could forward the request
                            // or return the known leader's address.
                            ClientResponse::Error("Not the leader. Please send write requests to the leader.".to_string())
                        }
                    }

                    Request::Delete { key } => {
                        // The logic for Delete is identical to Set.
                        let is_leader;
                        {
                            let node = raft_node.read().unwrap();
                            is_leader = node.state == NodeState::Leader;
                        }

                        if is_leader {
                            let command = Command::Delete { key };
                            let mut node = raft_node.write().unwrap();
                            let new_entry = LogEntry {
                                term: node.current_term,
                                command,
                            };
                            node.log.push(new_entry);

                            println!("[Leader] Appended new client command to log at index {}.", node.log.len());

                            continue; // Skip sending a response for now.
                        } else {
                            ClientResponse::Error("Not the leader. Please send write requests to the leader.".to_string())
                        }
                    }
                };

                // This part now only executes for Get requests or for non-leaders.
                let response_msg = WireMessage::ClientResponse(response);
                let response_bytes = bincode::serialize(&response_msg)?;

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

}
            WireMessage::Node(node_message) => {
                // ... (all node-to-node message handling is unchanged) ...
            }
            WireMessage::ClientResponse(_) => {
                // ... (unchanged) ...
            }
        }
    }
    // Ok(()) // This line might become unreachable, which is fine for now.
}

Deconstructing the Changes

Leader Check: For both Set and Delete, the very first thing we do is acquire a quick, short-lived read lock just to check if node.state == NodeState::Leader. This is a fast operation.
Rejecting Follower Requests: If the node is not the leader, we immediately create a ClientResponse::Error and let the existing response-sending logic handle it. This correctly enforces the “leader-only writes” rule.
Creating the LogEntry: If the node is the leader, it creates the corresponding Command and wraps it in a LogEntry along with its current_term. This captures the what (command) and the when (term) of the new piece of history.
Appending to the Log: Under a write lock, the leader calls node.log.push(new_entry). This is the moment the leader officially proposes the new state change by adding it to its own chronicle. The println! provides crucial visibility into this process.
The Deferred Response (continue): This is the most subtle but important part of the new code. After appending to the log, we continue to the next iteration of the loop. This deliberately skips sending an immediate response to the client. The client’s connection is kept open, waiting. This correctly models the Raft workflow: the leader can only confirm the command’s success after it has been replicated and committed, a process that will be orchestrated by other tasks in our system.

You have now successfully connected the client-facing part of your server to the internal Raft log. The leader can now accept proposals for state changes and record them in its own history, setting the stage for the replication process to begin.

What’s Next?

The leader now has new, un-replicated entries in its log. Its immediate responsibility is to inform its followers of this new history. In the next task, you will modify the leader’s heartbeat task. When it detects new entries in its log, it will send them to the followers inside the AppendEntries RPC, officially kicking off the data replication process.

Raft Leader: Implement Log Replication

Mục tiêu: Enhance the Raft leader to replicate log entries to followers. This task involves implementing next\_index and match\_index to track individual follower progress and refactoring the heartbeat mechanism to send customized AppendEntries RPCs containing new log data.

You have successfully programmed your leader to accept client commands and record them as new entries in its personal log. This is the first critical step in creating a new piece of history for the cluster. However, this history is currently confined to the leader’s memory alone. For the system to be fault-tolerant, this new entry must be replicated across the cluster. The leader must now transition from being a recorder to a broadcaster.

This task is about taking the new entries from the leader’s log and sending them to the followers. We will achieve this by enhancing the leader’s existing heartbeat mechanism. Instead of just sending empty AppendEntries RPCs, this task will now inspect the log, see if there are new entries to send, and package them into the RPC, officially kicking off the data replication process.

The Leader’s Ledger: Tracking Follower Progress

A leader cannot simply blast the same log entries to all followers blindly. Each follower might be in a slightly different state. One follower might be perfectly up-to-date, while another might have just rebooted and be far behind. A sophisticated leader must track the replication progress of each follower individually.

To do this, the Raft protocol requires the leader to maintain two crucial pieces of volatile state for each of its followers:

next_index: HashMap<NodeId, u64>: This map stores, for each follower, the index of the next log entry the leader will attempt to send to that follower. When a leader first comes to power, it is optimistic and initializes this value for every follower to be its own last log index + 1.
match_index: HashMap<NodeId, u64>: This map stores, for each follower, the index of the highest log entry that the leader knows is successfully replicated on that follower. It is initialized to 0 and increases as the leader receives successful AppendEntriesReply messages.

These two data structures form the leader’s dashboard, giving it a real-time view of the replication state of the entire cluster.

Step 1: Update the `RaftNode` State

First, let’s add this leader-specific state to our RaftNode struct. This is volatile state that only a leader uses, and it is re-initialized whenever a node becomes a leader.

// In src/main.rs

// ... (other use statements)
use std::collections::HashMap;

// ... (LogEntry, Command, RPC structs, etc.)

/// Contains the core Raft state for a node.
#[derive(Debug)]
struct RaftNode {
    // ... (persistent and volatile state is unchanged) ...
    log: Vec<LogEntry>,
    commit_index: u64,
    current_leader: Option<u64>,
    election_deadline: Instant,
    votes_received: u64,

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

    /// For each server, index of the next log entry to send to that server.
    /// Initialized to leader's last log index + 1.
    next_index: HashMap<u64, u64>,

    /// For each server, index of the highest log entry known to be replicated on server.
    /// Initialized to 0, increases monotonically.
    match_index: HashMap<u64, u64>,
}

impl RaftNode {
    fn new() -> Self {
        RaftNode {
            // ... (other fields are unchanged) ...
            log: Vec::new(),
            commit_index: 0,
            current_leader: None,
            election_deadline: Self::new_election_deadline(),
            votes_received: 0,
            // Initialize the leader-specific state as empty.
            next_index: HashMap::new(),
            match_index: HashMap::new(),
        }
    }
    // ... (new_election_deadline is unchanged) ...
}

Step 2: Initialize Leader State upon Election

When a node becomes a leader, it must immediately initialize the next_index and match_index for all its peers. The perfect place to do this is in our handle_connection function, right where a candidate transitions to leader after winning an election.

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

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

// When a leader is first elected, it must re-initialize its state
    // for tracking follower progress.

    // The leader's log is the source of truth, so `next_index` for each
    // follower is initialized to the index just after the leader's last log entry.
    let next_log_index = node.log.len() as u64 + 1;

    // We need the list of peer IDs to initialize the maps.
    // Let's pass it into handle_connection.
    for peer_id in peer_ids {
        node.next_index.insert(*peer_id, next_log_index);
        node.match_index.insert(*peer_id, 0);
    }

}

To make this work, you’ll need to pass the list of peer IDs into handle_connection. Update the function signature and the call site in your main loop.

// In main():
let peer_ids: Vec<u64> = config.peers.iter().map(|(id, _)| *id).collect();
// ...
// In the main loop:
tokio::spawn(async move {
    // ...
    let peer_ids_clone = peer_ids.clone();
    if let Err(e) = handle_connection(stream, store_clone, raft_node_clone, majority, &peer_ids_clone).await {
        // ...
    }
});

// In handle_connection() signature:
async fn handle_connection(
    // ...
    majority: u64,
    peer_ids: &[u64],
) -> anyhow::Result<()> { /* ... */ }

Step 3: Refactor the Heartbeat Task into a Replication Task

This is the core of our current task. The simple “broadcast a single heartbeat” model is no longer sufficient. We need to send a customized AppendEntries RPC to each peer based on its next_index. This requires a significant refactoring of the leader’s background task.

// In src/main.rs -> main() function
// Replace the entire "Leader Heartbeat Logic" tokio::spawn block with this new version.

    // --- Leader Replication Logic ---
    let replication_node = raft_node.clone();
    let replication_peers = config.peers.clone();
    let replication_connections = peer_connections.clone();
    let self_id = config.id;
    tokio::spawn(async move {
        let replication_interval = Duration::from_millis(50);

        loop {
            tokio::time::sleep(replication_interval).await;

            // --- LOCK, READ, and RELEASE ---
            // First, we need to determine if we are the leader and gather the necessary
            // state to send RPCs. We do this in a scoped read lock.
            let mut rpcs_to_send: Vec<(u64, AppendEntriesArgs)> = Vec::new();
            { // Scoped read lock
                let node = replication_node.read().unwrap();
                if node.state == NodeState::Leader {
                    // For each peer, construct a specific AppendEntries RPC.
                    for (peer_id, _peer_addr) in &replication_peers {
                        let next_index = *node.next_index.get(peer_id).unwrap_or(&1);

                        // If the follower's next_index is beyond our log, we can't send anything yet.
                        // This can happen if the log is empty.
                        if next_index > node.log.len() as u64 && !node.log.is_empty() {
                           // This condition may need refinement as we build out log repair logic
                           continue;
                        }

                        // Determine the previous log entry's details.
                        // Raft indices are 1-based, Vec is 0-based.
                        let prev_log_index = next_index - 1;
                        let prev_log_term = if prev_log_index > 0 {
                            node.log[(prev_log_index - 1) as usize].term
                        } else {
                            0
                        };

                        // Get the slice of new entries to send to this follower.
                        // Remember that `next_index` is 1-based.
                        let entries_to_send = if !node.log.is_empty() {
                            node.log[((next_index - 1) as usize)..].to_vec()
                        } else {
                            Vec::new()
                        };

                        let args = AppendEntriesArgs {
                            term: node.current_term,
                            leader_id: self_id,
                            prev_log_index,
                            prev_log_term,
                            entries: entries_to_send,
                            leader_commit: node.commit_index,
                        };
                        rpcs_to_send.push((*peer_id, args));
                    }
                }
            } // Read lock is released here.

            // --- ACT (NETWORK I/O) ---
            // Now, send the prepared RPCs outside of the lock.
            for (peer_id, args) in rpcs_to_send {
                println!("[Replication] Sending AppendEntries to peer {} ({} entries)", peer_id, args.entries.len());

                let rpc_message = WireMessage::Node(NodeMessage::AppendEntries(args));
                let message_bytes = match bincode::serialize(&rpc_message) {
                    Ok(bytes) => bytes,
                    Err(e) => {
                        eprintln!("[Replication] Failed to serialize AppendEntries: {}", e);
                        continue;
                    }
                };

                let mut connections = replication_connections.write().unwrap();
                if let Some(stream) = connections.get_mut(&peer_id) {
                    // Send length-prefixed message
                    if stream.write_u64(message_bytes.len() as u64).await.is_err() || stream.write_all(&message_bytes).await.is_err() {
                        eprintln!("[Replication] Failed to send AppendEntries to peer {}", peer_id);
                    }
                }
            }
        }
    });

Deconstructing the New Replication Task

From Broadcast to Targeted Sending: The single biggest change is that we no longer create one message and broadcast it. We now loop through our peers (replication_peers) and create a unique AppendEntriesArgs for each one.
The “Read, Release, Act” Pattern: We gather all the information we need to build the RPCs under a short-lived read lock. We store the (peer_id, args) pairs in a temporary vector rpcs_to_send. Only after the lock is released do we iterate through this vector and perform the slow network I/O. This is a critical pattern for performance and avoiding deadlocks.
Calculating Follower-Specific State: For each peer, we look up its next_index. Based on this, we calculate prev_log_index and prev_log_term. This is the “consistency handshake” that a follower will use to validate the request.
Slicing the Log: The line node.log[((next_index - 1) as usize)..].to_vec() creates a new vector containing only the log entries that a specific follower needs. If next_index points past the end of the log, this slice will be empty, and the message acts as a heartbeat, which is exactly the desired behavior.
Debugging println!: The new log message is very informative. It tells you exactly how many entries are being sent to each peer, which will be invaluable for observing your system’s behavior.

You have now transformed your leader from a simple broadcaster of heartbeats into a sophisticated replication engine. It tracks the progress of each follower and sends precisely the data needed to bring them up-to-date with its own authoritative history.

What’s Next?

The leader is now sending log entries, but the followers are not yet equipped to handle them. Their current AppendEntries handler only knows how to reset a timer. In the next task, you will implement the follower’s side of the replication: it will receive these AppendEntries RPCs, perform the crucial consistency check using prev_log_index and prev_log_term, append the new entries to its own log, and send a success response back to the leader.

Raft Follower: Implement AppendEntries RPC Logic

Mục tiêu: Implement the log replication logic for a follower node in a Raft consensus implementation. This involves handling the AppendEntries RPC, performing the log consistency check, repairing the log by truncating conflicting entries, appending new entries, and updating the commit index.

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

You have successfully engineered your leader node into a sophisticated replication machine. It now diligently tracks the progress of each follower and sends them AppendEntries RPCs containing new log entries. The messages are being sent, and the history of your system is now on the wire. This task is about completing that journey: teaching the follower nodes how to receive these messages, validate them, and append the new history to their own logs, thus achieving replication.

This is the moment where distributed consensus truly happens. The follower’s logic for handling an AppendEntries RPC is the gatekeeper of consistency. It ensures that a follower’s log never diverges from the leader’s authoritative history and provides the mechanism to repair the log if it ever does.

A Follower’s Sacred Duty: The Log Consistency Check

When a follower receives an AppendEntries RPC, it can’t just blindly append the new entries. It must first perform a crucial “consistency check” to ensure its own log is in perfect sync with the leader’s up to the point where the new entries begin. This is the purpose of the prev_log_index and prev_log_term fields in the RPC.

The follower performs the following steps:

Term Check: As always, it first checks if the leader’s term is stale (args.term < self.current_term). If so, it rejects the RPC. This is already implemented from our heartbeat logic.
Consistency Check: It looks at its own log at the index specified by args.prev_log_index. * If the follower’s log is too short (doesn’t have an entry at prev_log_index), its log is out of date. It must reject the RPC. * If it has an entry, but the entry’s term does not match args.prev_log_term, its log has diverged from the leader’s. It must reject the RPC.
Log Repair and Append: Only if the consistency check passes does the follower proceed. * It checks for any conflicting entries. If an existing entry in its log has the same index as a new entry from the RPC but a different term, it means the old entry was from a stale leader. The follower must delete that conflicting entry and all subsequent entries from its log. This is the powerful log repair mechanism of Raft. * It then appends any new entries from the RPC that are not already in its log.
Update Commit Index: The follower updates its own commit_index based on the leader_commit value from the RPC, ensuring it knows which entries are safe to be applied.
Send Success Reply: Finally, it sends a successful reply back to the leader, signaling that the new entries have been safely persisted.

Implementing the Follower’s Replication Logic

We will now upgrade the AppendEntries handler in our handle_connection function from a simple heartbeat handler to a full-fledged replication engine.

Here are the changes for src/main.rs. We will be replacing the existing logic inside the NodeMessage::AppendEntries match arm.

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

                    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!(
                                    "[Follower] Rejecting AppendEntries from stale leader with term {}.",
                                    args.term
                                );
                                success = false;
                            } else {
                                // This is a valid communication from a potential leader.
                                // If we were a Candidate, we recognize the new leader and step down.
                                if node.state == NodeState::Candidate {
                                    println!(
                                        "[Follower] Discovering leader {} while a candidate. Stepping down.",
                                        args.leader_id
                                    );
                                    node.state = NodeState::Follower;
                                }

                                // We have found the rightful leader for this term.
                                node.current_leader = Some(args.leader_id);
                                node.election_deadline = RaftNode::new_election_deadline();

// --- THE LOG CONSISTENCY CHECK ---
                                // Rule 2: Reply false if log doesn’t contain an entry at prevLogIndex
                                // whose term matches prevLogTerm.

                                // Raft indices are 1-based, our Vec is 0-based.
                                let prev_log_index_usize = args.prev_log_index as usize;

                                if args.prev_log_index > 0 && 
                                   (node.log.len() < prev_log_index_usize || node.log[prev_log_index_usize - 1].term != args.prev_log_term) {

                                    println!("[Follower] Consistency check failed. Rejecting AppendEntries.");
                                    success = false;
                                } else {
                                    // --- LOG REPAIR AND APPEND ---
                                    success = true;
                                    let mut current_index = args.prev_log_index;
                                    for entry in &args.entries {
                                        current_index += 1;
                                        let current_index_usize = current_index as usize;

                                        // Rule 3: If an existing entry conflicts with a new one, delete it and all that follow.
                                        if let Some(existing_entry) = node.log.get(current_index_usize - 1) {
                                            if existing_entry.term != entry.term {
                                                node.log.truncate(current_index_usize - 1);
                                            }
                                        }

                                        // Rule 4: Append any new entries not already in the log.
                                        if node.log.len() < current_index_usize {
                                            node.log.push(entry.clone());
                                        }
                                    }
                                    if !args.entries.is_empty() {
                                        println!("[Follower] Appended {} entries. Log length is now {}.", args.entries.len(), node.log.len());
                                    }

                                    // Rule 5: Update the commit index.
                                    if args.leader_commit > node.commit_index {
                                        node.commit_index = std::cmp::min(args.leader_commit, node.log.len() as u64);
                                        println!("[Follower] Updated commit_index to {}.", node.commit_index);
                                    }
                                }

}

                            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?;
                    }

Deconstructing the New Logic

Consistency Check (if args.prev_log_index > 0 && ...): This is the heart of the new logic.
- We first check if prev_log_index is greater than 0. An index of 0 signifies that the leader is sending entries from the very beginning of the log, so the consistency check is trivially passed.
- We then check if our log is long enough (node.log.len() < prev_log_index_usize).
- Finally, we access the entry at the correct 0-based index (prev_log_index_usize - 1) and compare its term.
- If either check fails, success remains false, and we send a failure reply.
Log Repair (if let Some(existing_entry) ...): This loop implements Raft’s powerful self-healing property. If we find an entry at an index that conflicts with a new entry (same index, different term), we know our log is tainted with stale history. node.log.truncate(...) decisively cuts away this incorrect history, preparing the log for the new, correct entries from the current leader.
Append Logic (if node.log.len() < ...): After any necessary repairs, this condition checks if the entry is truly new and appends it to the log using push. The clone() is necessary because we are borrowing the entry from the args but need to give the log owned data.
Commit Index Update (if args.leader_commit > ...): This is how a follower learns which entries are safe to be applied. It updates its own commit_index to the minimum of what the leader says is committed and the last entry it actually has. This prevents a follower from thinking an entry is committed before it has even received it.

You have now implemented the core of Raft’s data replication protocol. The leader proposes changes, and followers validate and accept them, ensuring that the logs across the cluster converge to a single, consistent state.

What’s Next?

The followers are now dutifully replicating log entries and sending success messages back to the leader. But the leader is currently ignoring these replies. For the system to make progress, the leader must process these replies. In the next task, you will implement the logic for the leader to handle an AppendEntriesReply. Upon receiving a success reply, the leader will update its next_index and match_index for that follower, effectively tracking how much of its history has been successfully replicated across the cluster.

Implement Raft Leader Reply Handling and Async Refactor

Mục tiêu: Refactor the Raft implementation to use async-aware Tokio locks (RwLock, Mutex) to prevent deadlocks. Implement the leader’s logic to process AppendEntries replies from followers, enabling adaptive log replication by updating each follower’s next_index and match_index based on success or failure.

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

You’ve done an incredible job getting the core replication machinery in place. The leader is now sending new log entries, and followers are receiving them, performing consistency checks, and updating their own logs. This is a huge step! However, right now, the leader is “flying blind.” It sends out the data but has no idea if it was successfully received. The followers are dutifully sending back AppendEntriesReply messages, but the leader is simply ignoring them.

This task is about closing that feedback loop. We will teach the leader how to process these replies, allowing it to track the replication progress of each follower individually. This is the “intelligence” of the replication system, enabling the leader to adapt its strategy for each follower—resending entries upon failure or moving on upon success. This tracking is the absolute prerequisite for the final step: determining when an entry is safely replicated on a majority of nodes.

Part 1: A Critical Refactor for Asynchronous Correctness

Before we implement the reply logic, we must address a subtle but critical issue in our current code that will prevent our new logic from working correctly. It relates to how we’ve been sharing state in our async world.

The Challenge: `std::sync::RwLock` and `.await`

So far, we have been using std::sync::RwLock to protect our shared data like raft_node and peer_connections. This lock works perfectly in standard multi-threaded Rust. However, it has a major limitation in an async environment: you must not hold the lock across an .await point.

An .await point is where your async function can yield control, allowing the Tokio runtime to work on other tasks. If you hold a standard library lock and then .await a network call, your task goes to sleep but keeps the data locked. If another task needs to access that same data, it will try to acquire the lock and block forever, waiting for a task that is asleep. This is a classic recipe for a deadlock that will freeze your entire server.

The existing leader’s replication task already has this latent bug: it acquires a lock on peer_connections and then calls .await to write to the network stream.

The Solution: `tokio::sync::RwLock`

The tokio crate provides its own version of RwLock that is “async-aware.” Its read() and write() methods return a future that you .await. When you await the lock, if it’s not available, the runtime can safely switch to another task and come back later. This completely avoids the deadlock problem.

We will now perform a crucial refactor to switch all our shared state to use tokio’s synchronization primitives. This is a foundational improvement that will make our entire application safer and more robust.

Step 1: Update your use statements in src/main.rs.

// In src/main.rs

// REMOVE the std::sync versions
// use std::sync::{Arc, RwLock}; 
// ADD the tokio::sync versions
use tokio::sync::{Arc, RwLock}; 
use std::collections::HashMap;
// ... other use statements

Step 2: Update the initialization of your shared state in the main function. The creation is identical, so you only need to ensure the correct RwLock is in scope from the use statement.

Step 3: Update ALL lock acquisitions to use .await. This is the most important change. Everywhere you see .read().unwrap() or .write().unwrap(), you must change it to .read().await or .write().await.

For example, in your KvStore implementation:

// In src/main.rs -> impl KvStore

// BEFORE:
// let db = self.db.read().unwrap();

// AFTER:
let db = self.db.read().await;

And in your handle_connection function:

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

// BEFORE:
// let mut node = raft_node.write().unwrap();

// AFTER:
let mut node = raft_node.write().await;

Go through your main and handle_connection functions and update every single call. This change is mandatory for the logic we are about to build.

Part 2: Implementing the Reply Handler Logic

With our locking mechanism corrected, we can now safely implement the reply handling. The leader needs to know which follower succeeded and which failed so it can update its next_index and match_index maps accordingly.

When a follower fails its consistency check, the leader must decrement that follower’s next_index and try again with an earlier log entry. This “walking back” process is how the leader finds the point of agreement and repairs the follower’s log.

The Architectural Pattern: Spawning Per-Peer Tasks

The task that sends a request must be the one to receive the reply, because it holds the context of what was sent (AppendEntriesArgs). To do this concurrently for all peers, we will refine our leader’s replication task. Instead of building a list of RPCs, it will now spawn a new, short-lived async task for each peer during each replication round. This task will be responsible for the full request-reply cycle for that one peer.

Let’s implement this robust pattern.

Step 1: Upgrade the Peer Connection Map

To allow these newly spawned tasks to safely access the TCP stream for their specific peer, we need to wrap each stream in its own lock. A tokio::sync::Mutex is perfect here.

First, update the type alias and initialization in main:

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

// The connection map now holds an Arc-wrapped Mutex for each stream.
type PeerConnections = Arc<RwLock<HashMap<u64, Arc<Mutex<TcpStream>>>>>;
let peer_connections: PeerConnections = Arc::new(RwLock::new(HashMap::new()));

Next, update the Peer Manager task to insert the stream with this new type:

// In src/main.rs -> main() -> Peer Manager Task

tokio::spawn(async move {
    // ...
    for (peer_id, peer_addr) in &peers {
        loop {
            // ...
            match TcpStream::connect(peer_addr).await {
                Ok(stream) => {
                    println!("[Peer Manager] Successfully connected to peer {}", peer_id);
                    let mut connections = peer_connections_clone.write().await;

// Wrap the stream in an Arc> before inserting.
                    connections.insert(*peer_id, Arc::new(Mutex::new(stream)));

break; 
                }
                // ...
            }
        }
    }
});

You’ll also need to add use tokio::sync::Mutex; at the top of your file.

Step 2: Refactor the Leader’s Replication Task

Now we’ll replace the old replication task with our new, more powerful version that processes replies.

// In src/main.rs -> main() function
// Replace the entire "Leader Replication Logic" tokio::spawn block with this.

    // --- Leader Replication Logic ---
    let replication_node = raft_node.clone();
    let replication_peers = config.peers.clone();
    let replication_connections = peer_connections.clone();
    let self_id = config.id;
    tokio::spawn(async move {
        let replication_interval = Duration::from_millis(50);

        loop {
            tokio::time::sleep(replication_interval).await;

            let node = replication_node.read().await;
            if node.state != NodeState::Leader {
                continue; // Only leaders send replication messages.
            }

            // For each peer, spawn a task to handle its replication.
            for (peer_id, _peer_addr) in &replication_peers {
                let next_index = *node.next_index.get(peer_id).unwrap_or(&1);

                // If our log is not long enough to send the entry at next_index, continue.
                if next_index > node.log.len() as u64 {
                    // This is a normal condition for a fully caught-up follower,
                    // in which case we'd send an empty heartbeat. We'll handle this
                    // by ensuring `entries_to_send` is empty.
                }

                let prev_log_index = next_index - 1;
                let prev_log_term = if prev_log_index > 0 {
                    node.log[(prev_log_index - 1) as usize].term
                } else {
                    0
                };

                // Note: .to_vec() creates a copy, releasing the read lock on the log.
                let entries_to_send = node.log.get(((next_index - 1) as usize)..).unwrap_or(&[]).to_vec();

                let args = AppendEntriesArgs {
                    term: node.current_term,
                    leader_id: self_id,
                    prev_log_index,
                    prev_log_term,
                    entries: entries_to_send,
                    leader_commit: node.commit_index,
                };

                // Clone the Arcs for the new task.
                let task_peer_id = *peer_id;
                let task_node = replication_node.clone();
                let task_connections = replication_connections.clone();

                // Spawn a task for each peer.
                tokio::spawn(async move {
                    if let Err(e) = handle_peer_replication(task_peer_id, task_node, task_connections, args).await {
                        eprintln!("[Replication] Error replicating to peer {}: {}", task_peer_id, e);
                    }
                });
            }
        }
    });

Step 3: Create the handle_peer_replication Helper Function

This new function will contain the self-contained logic for a single request-reply cycle. Add this as a new standalone function in src/main.rs.

// In src/main.rs, add this new function.

/// Handles the full request-reply cycle for replicating entries to a single peer.
async fn handle_peer_replication(
    peer_id: u64,
    raft_node: Arc<RwLock<RaftNode>>,
    connections: PeerConnections,
    args: AppendEntriesArgs,
) -> anyhow::Result<()> {
    // 1. Get the connection to the peer.
    let connections_map = connections.read().await;
    let peer_conn_arc = match connections_map.get(&peer_id) {
        Some(c) => c.clone(),
        None => {
            // This can happen if a connection is dropped and not yet re-established.
            return Ok(());
        }
    };
    // Drop the read lock on the connections map quickly.
    drop(connections_map);

    // 2. Lock the specific stream for this peer to ensure exclusive access.
    let mut stream = peer_conn_arc.lock().await;

    // 3. Send the AppendEntries RPC.
    let rpc_message = WireMessage::Node(NodeMessage::AppendEntries(args.clone()));
    let message_bytes = bincode::serialize(&rpc_message)?;
    stream.write_u64(message_bytes.len() as u64).await?;
    stream.write_all(&message_bytes).await?;

    // 4. Read the reply from the peer.
    // We need a proper read loop here, similar to handle_connection.
    // For now, let's assume a simple read for the direct reply.
    // A more robust implementation would use a helper or a timeout.
    let reply_len = stream.read_u64().await?;
    let mut reply_buffer = vec![0; reply_len as usize];
    stream.read_exact(&mut reply_buffer).await?;

    let reply_message: WireMessage = bincode::deserialize(&reply_buffer)?;
    let reply = match reply_message {
        WireMessage::Node(NodeMessage::AppendEntriesReply(r)) => r,
        _ => {
            return Err(anyhow::anyhow!("Expected AppendEntriesReply from peer"));
        }
    };

    // 5. Process the reply.
    let mut node = raft_node.write().await;

    // Step down if we discover a higher term.
    if reply.term > node.current_term {
        node.current_term = reply.term;
        node.state = NodeState::Follower;
        node.voted_for = None;
        node.current_leader = None;
        return Ok(());
    }

    // Ignore replies if we're no longer the leader.
    if node.state != NodeState::Leader {
        return Ok(());
    }

    if reply.success {
        // The follower's log is consistent and entries were appended.
        let new_match_index = args.prev_log_index + args.entries.len() as u64;
        node.match_index.insert(peer_id, new_match_index);
        node.next_index.insert(peer_id, new_match_index + 1);
        println!("[Replication] Peer {} successfully replicated up to index {}", peer_id, new_match_index);
    } else {
        // The follower's log was inconsistent. Decrement next_index and retry.
        let next_index = node.next_index.entry(peer_id).or_insert(1);
        if *next_index > 1 {
            *next_index -= 1;
        }
        println!("[Replication] Peer {} failed consistency check. Retrying with index {}", peer_id, *next_index);
    }

    Ok(())
}

Note: You will also need to add use tokio::io::{AsyncReadExt, AsyncWriteExt}; to your use statements for this new function.

Deconstructing the New Architecture

Async-Safe Locks: By switching to tokio::sync::RwLock and Mutex, your application is now free from a major class of deadlocks, making it far more reliable under load.
Concurrent, Isolated Replication: The leader’s main replication loop now acts as a dispatcher. It spawns a separate task for each peer. This is incredibly powerful: a slow or disconnected peer will not slow down replication to other, healthy peers. Each task runs in parallel.
Context is Preserved: The handle_peer_replication function has everything it needs in one place: the args that were sent and the reply that was received. This makes the logic for updating next_index and match_index clear and correct.
Adaptive Replication: You have now fully implemented the leader’s adaptive logic. On success, it advances its knowledge of the follower’s state. On failure, it “walks back” its next_index, automatically probing to find the point of consistency so it can repair the follower’s log.

You have now built a truly robust and adaptive replication system. The leader has a closed feedback loop, allowing it to intelligently manage the state of the entire cluster’s history.

What’s Next?

The leader is now meticulously tracking the highest log index replicated on each follower (match_index). The final step is to use this information. In the next task, you will implement the logic for the leader to check its match_index map after each successful replication. If it finds that a log entry from its current term is now present on a majority of servers, it will advance its commit_index, officially committing the entry.

Implement Raft Log Commitment

Mục tiêu: Implement the logic for the Raft leader to advance its commit_index. This involves checking follower replication progress via match\_index and applying Raft’s two safety rules: majority replication and the current-term-only constraint.

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

You have achieved something truly remarkable in the previous task. By implementing the leader’s reply-handling logic and refactoring to use async-safe locks, you’ve created a closed-loop, adaptive replication system. The leader is no longer just broadcasting data; it’s now intelligently tracking the progress of each follower using the match_index map. This map is the leader’s dashboard, showing exactly how much of its history has been successfully replicated across the cluster.

That knowledge is the final ingredient we need to achieve the ultimate goal of a consensus algorithm: commitment. This task is about using the information in the match_index map to determine when a log entry is safely stored on a majority of servers, at which point the leader can officially declare it as “committed.” This is the point of no return, the moment the cluster formally agrees on a piece of history, guaranteeing that it will never be undone or changed.

The Two Pillars of Commitment in Raft

For a leader to advance its commit_index, it must satisfy two fundamental rules that ensure the safety and consistency of the entire system.

The Majority Rule: An entry is considered safely replicated only when it exists on a majority of servers (i.e., (N / 2) + 1 nodes, including the leader itself). The leader’s match_index map is the tool it uses to verify this. For any given log index i, the leader checks how many followers have a match_index >= i. If that count, plus the leader itself, reaches a majority, this condition is met.
The Current Term Rule (A Critical Safety Guarantee): The Raft paper specifies a crucial safety constraint: a leader cannot commit a log entry from a previous leader’s term, even if it appears to be on a majority of servers. It can only advance its commit_index to an entry that is from its own, current term. This rule prevents a newly elected leader from misinterpreting the state of the logs and prematurely committing an old entry that a previous, failed leader had not yet committed.

An entry at index i is committed only if it is stored on a majority of servers AND the entry at index i in the leader’s log has log[i].term == leader.current_term.

Implementation Strategy: A Centralized Commit Checker

The logic to check for new commits should be executed whenever there’s a possibility that the commit index can be advanced. This happens every time a follower successfully replicates an entry and the leader updates its match_index for that follower.

To keep our code clean and centralized, we will create a new helper function, update_leader_commit_index. This function will be responsible for:

Checking for any new entries that can be committed based on the two rules above.
Advancing the leader’s commit_index if new entries are found to be committed.

We will then call this function from our handle_peer_replication task immediately after a successful reply is processed.

Step 1: Create the `update_leader_commit_index` Helper Function

Let’s add a new standalone async function to src/main.rs. This function will encapsulate all the logic for checking and advancing the commit index.

// In src/main.rs, add this new helper function.

/// After a successful replication, the leader checks if it can advance its commit_index.
/// This function encapsulates that logic.
async fn update_leader_commit_index(
    raft_node: Arc<RwLock<RaftNode>>,
    majority: u64,
) {
    let mut node = raft_node.write().await;

    // We start checking from the entry right after the current commit_index.
    let mut new_commit_index = node.commit_index;

    // Find the highest index that is replicated on a majority of servers.
    for i in (node.commit_index + 1)..=(node.log.len() as u64) {
        // Raft log indices are 1-based, our Vec is 0-based.
        let log_entry_index_0based = (i - 1) as usize;

        // Count how many servers have replicated this entry.
        // The leader always has the entry, so we start the count at 1.
        let mut match_count = 1;
        for (_, match_idx) in &node.match_index {
            if *match_idx >= i {
                match_count += 1;
            }
        }

        // Check if the entry has been replicated on a majority AND is from the leader's current term.
        if match_count >= majority && node.log[log_entry_index_0based].term == node.current_term {
            // This entry is now safely committed. Update our potential new commit_index.
            new_commit_index = i;
        } else {
            // Once we find an entry that cannot be committed, we stop.
            // We can only commit a contiguous sequence from the start of the log.
            break;
        }
    }

    if new_commit_index > node.commit_index {
        println!("[Leader] Advanced commit_index from {} to {}", node.commit_index, new_commit_index);
        node.commit_index = new_commit_index;
    }
}

Step 2: Integrate the Commit Check into the Replication Cycle

Now, we need to call our new helper function at the correct time. The right moment is inside handle_peer_replication, right after the leader processes a successful reply and updates the follower’s match_index.

First, you’ll need to pass the majority value into handle_peer_replication.

// In src/main.rs -> main() -> Leader Replication Logic
// Update the tokio::spawn call inside the loop.
tokio::spawn(async move {
    if let Err(e) = handle_peer_replication(task_peer_id, task_node, task_connections, args, majority).await {
        eprintln!("[Replication] Error replicating to peer {}: {}\n", task_peer_id, e);
    }
});

// Update the function signature for handle_peer_replication
async fn handle_peer_replication(
    peer_id: u64,
    raft_node: Arc<RwLock<RaftNode>>,
    connections: PeerConnections,
    args: AppendEntriesArgs,
    majority: u64, // Add this parameter
) -> anyhow::Result<()> { /* ... */ }

Next, call the update_leader_commit_index function from within handle_peer_replication’s success branch.

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

    // ... (logic to get connection, send RPC, and read reply) ...

    // 5. Process the reply.
    // NOTE: This logic requires a write lock on the node, but we also need to call
    // `update_leader_commit_index` which also takes a write lock. To avoid deadlock,
    // let's do this in a single locked block.
    { // Scoped write lock
        let mut node = raft_node.write().await;

        // Step down if we discover a higher term.
        if reply.term > node.current_term {
            // ... (step down logic is unchanged) ...
            return Ok(());
        }

        // Ignore replies if we're no longer the leader.
        if node.state != NodeState::Leader {
            return Ok(());
        }

        if reply.success {
            // The follower's log is consistent and entries were appended.
            let new_match_index = args.prev_log_index + args.entries.len() as u64;
            node.match_index.insert(peer_id, new_match_index);
            node.next_index.insert(peer_id, new_match_index + 1);
            println!("[Replication] Peer {} successfully replicated up to index {}", peer_id, new_match_index);

// After a successful replication, check if we can advance the commit index.
            // This is done inside the same lock to ensure atomicity.
            // We'll extract the logic of `update_leader_commit_index` and place it here directly
            // to work under a single write lock.

            let mut new_commit_index = node.commit_index;
            for i in (node.commit_index + 1)..=(node.log.len() as u64) {
                let log_entry_index_0based = (i - 1) as usize;
                let mut match_count = 1;
                for (_, match_idx) in &node.match_index {
                    if *match_idx >= i {
                        match_count += 1;
                    }
                }

                if match_count >= majority && node.log[log_entry_index_0based].term == node.current_term {
                    new_commit_index = i;
                } else {
                    break;
                }
            }

            if new_commit_index > node.commit_index {
                println!("[Leader] Advanced commit_index from {} to {}", node.commit_index, new_commit_index);
                node.commit_index = new_commit_index;
            }

} else {
            // The follower's log was inconsistent. Decrement next_index and retry.
            let next_index = node.next_index.entry(peer_id).or_insert(1);
            if *next_index > 1 {
                *next_index -= 1;
            }
            println!("[Replication] Peer {} failed consistency check. Retrying with index {}", peer_id, *next_index);
        }
    } // Write lock is released here.

    Ok(())
}

Self-correction during thought process: Calling a separate async function that also takes a write lock from within a scope that already holds a write lock is complex and can lead to deadlocks with tokio::sync::RwLock if not careful. The simplest, safest, and most atomic approach is to perform all the state updates under a single write lock. Therefore, instead of creating a separate helper function, it’s better to inline the commit logic directly inside the if reply.success block in handle_peer_replication. This ensures that updating match_index and then checking the commit index happen as one indivisible operation. I will update the guide to reflect this safer pattern.

Deconstructing the New Logic

Single Atomic Update: By placing the commit logic inside the same write-locked block where we update match_index, we guarantee that the state of the system is always consistent. No other task can see the intermediate state where match_index is updated but commit_index isn’t.
The Commit Loop (for i in ...): This loop efficiently checks for the new “commit frontier.” It starts right after the last known committed entry and moves forward.
Counting Replicas (match_count): We start the count at 1 to include the leader itself. Then, we iterate through the match_index map, incrementing the count for every follower that has replicated up to or beyond the index i we are currently checking.
The Double-Condition Check: The line if match_count >= majority && node.log[...].term == node.current_term is the direct implementation of the two pillars of Raft commitment. It is the single most important safety check in the entire replication process.
Contiguous Commits (break): The moment we find an entry that cannot be committed (either due to lack of majority or being from a past term), we must stop checking. Raft only ever commits a contiguous prefix of the log.

You have now implemented the brain of the consensus algorithm. Your leader can now definitively determine when a command has been accepted by the cluster, transforming a mere proposal in its log into an immutable, committed fact of the system’s history.

What’s Next?

The leader now knows which entries are committed, but this knowledge is still just a number—the commit_index. The final two steps are to make this commitment meaningful. In the next task, you will implement the logic for all nodes (leader and followers) to apply the commands from committed log entries to their local state machine—the KvStore HashMap. This is the moment the abstract log entry becomes a concrete change in your key-value store.

Implement the Raft State Machine Applier

Mục tiêu: Create a dedicated background task to apply committed log entries from the Raft log to the key-value store. This involves adding a last\_applied index to track state machine progress and ensuring all nodes consistently apply commands once they are committed by the cluster.

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

You have reached a pivotal moment in the construction of your Raft consensus module. In the last task, you implemented the logic for the leader to determine that a log entry is safely replicated on a majority of nodes and to advance its commit_index. This commit_index represents the collective, immutable agreement of the cluster. It’s the point of no return.

However, an agreement in a log is just stored data; it’s an intent to change the state. The real value comes when that intent is turned into action. This task is about bridging that final gap: taking the commands from the committed log entries and applying them to your state machine—the KvStore’s HashMap. This is the moment an abstract agreement becomes a concrete reality in your key-value store.

The `last_applied` Index: Tracking State Machine Progress

Just as we use commit_index to track what’s been agreed upon, each node needs to track how far it has progressed in applying those agreements to its local state machine. We introduce a new piece of volatile state for this purpose: last_applied.

commit_index: The highest log entry index known to be committed (agreed upon by the cluster).
last_applied: The highest log entry index that this specific node has applied to its state machine.

The fundamental rule of a Raft state machine is that a node can only apply an entry i if i <= commit_index. The core work of this task is to implement a mechanism that constantly strives to make last_applied catch up to commit_index.

Step 1: Add `last_applied` to the `RaftNode` State

Let’s add this new field to our RaftNode struct in src/main.rs. It’s volatile state, so we’ll place it with commit_index and current_leader.

// In src/main.rs -> RaftNode struct

#[derive(Debug)]
struct RaftNode {
    // ... (state, current_term, voted_for, log) ...

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

    /// Index of the highest log entry known to be committed.
    commit_index: u64,

    /// Index of the highest log entry applied to the state machine.
    /// Initialized to 0, increases monotonically.
    last_applied: u64,

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

    // ... (election_deadline, votes_received, leader state) ...
}

// Update the RaftNode::new() function
impl RaftNode {
    fn new() -> Self {
        RaftNode {
            // ... (other fields are unchanged) ...
            log: Vec::new(),
            commit_index: 0,
            // A node starts with no entries applied to its state machine.
            last_applied: 0,
            current_leader: None,
            // ...
        }
    }
    // ...
}

The Applier: A Dedicated Background Worker

To handle the application of committed entries, we will follow our robust design pattern of creating a dedicated, concurrent task. This “Applier” task will run in the background on every node (Leaders and Followers alike). Its sole responsibility is to watch for the commit_index to advance beyond the last_applied index and apply the intervening entries.

This is a crucial concept: state machine application is not just a leader’s job. As soon as a follower learns of a new commit_index from a leader’s AppendEntries RPC, its own local Applier task will see the new committed entries and apply them. This is how all nodes in the cluster independently converge on the exact same state.

Step 2: Implement the State Machine Applier Task

Let’s add this new tokio::spawn block to our main function. It needs shared access to both the raft_node state and the store itself.

// In src/main.rs -> main() function
// Add this block alongside your other background tasks.

    // --- State Machine Applier Logic ---
    let applier_node = raft_node.clone();
    let applier_store = store.clone();
    tokio::spawn(async move {
        // This task runs forever, applying committed entries to the state machine.
        let applier_interval = Duration::from_millis(10);
        loop {
            tokio::time::sleep(applier_interval).await;

            let mut commands_to_apply = Vec::new();
            let new_last_applied;

            // --- LOCK, READ, COPY, UPDATE, and RELEASE ---
            // This block is carefully designed to hold the lock for a very short time.
            {
                let mut node = applier_node.write().await;

                // If there's nothing new to apply, just loop again.
                if node.last_applied >= node.commit_index {
                    continue;
                }

                println!(
                    "[Applier] Found new entries to apply. last_applied: {}, commit_index: {}",
                    node.last_applied, node.commit_index
                );

                // We can apply entries from last_applied + 1 up to commit_index.
                for i in (node.last_applied + 1)..=(node.commit_index) {
                    // Raft indices are 1-based, our Vec is 0-based.
                    let entry = node.log[(i - 1) as usize].clone();
                    commands_to_apply.push(entry.command);
                }

                // Update last_applied to reflect the work we are about to do.
                new_last_applied = node.commit_index;
                node.last_applied = new_last_applied;
            } // Write lock on `raft_node` is released here.

            // --- ACT (MODIFY STATE MACHINE) ---
            // Now, apply the commands to the KvStore outside of the RaftNode lock.
            for command in commands_to_apply {
                println!("[Applier] Applying command: {:?}", command);
                match command {
                    Command::Set { key, value } => {
                        applier_store.set(key, value).await;
                    }
                    Command::Delete { key } => {
                        applier_store.delete(&key).await;
                    }
                }
            }
        }
    });

Note: Your KvStore::set and delete methods must be async to correctly .await the lock, for example: pub async fn set(&self, key: String, value: String) { let mut db = self.db.write().await; ... }.

Deconstructing the Applier Task

This task is a masterclass in correct asynchronous state management.

The Loop and Sleep: The task runs in a perpetual loop, with a short sleep on each iteration. This prevents it from consuming 100% CPU while waiting for work, making it a very efficient background process.
The “Work Check”: The first thing it does is acquire a write lock on the RaftNode state. The condition if node.last_applied >= node.commit_index is a quick check to see if there’s anything to do. If not, it releases the lock and sleeps again.
The “Read, Copy, Update” Transaction: This is a critical atomic operation. While holding the RaftNode lock, we:
- Identify the range of entries to apply.
- Copy the Commands from the shared log into a temporary, local Vec. This is vital because it means we won’t need to access the shared log later.
- Immediately update node.last_applied to its new value.
The Lock Release: As soon as this block ends, the write lock on the RaftNode is released. All the slow work (applying to the KvStore) happens after this point. This ensures the core Raft state is never locked for long, keeping the system responsive.
The Application Stage: Finally, the task iterates through its local copy of commands. For each one, it acquires a lock on the KvStore’s HashMap (via the async set/delete methods) and performs the modification. Because we are outside the RaftNode lock, this process cannot cause a deadlock with our other Raft logic.

You have now built the engine that turns consensus into reality. The commit_index is the blueprint, and the Applier task is the construction worker, tirelessly building the state of your key-value store according to that blueprint.

What’s Next?

The entire data replication and application pipeline is now complete! There is only one piece of the puzzle left. When a client sends a Set or Delete request, we deliberately left them waiting for a response. Now that the command has been logged, replicated, committed, and applied, the leader finally has the authority to tell the client, “Your request was successful.”

In the final task of this step, you will implement the mechanism for the leader to send a success response back to the original client, completing the client-facing interaction.

Implement Client Notification for Raft Command Application

Mục tiêu: Complete the client request-response loop in a Raft-based key-value store. Implement an asynchronous notification system using Tokio’s oneshot channels to signal the client-handling task once a command has been successfully committed and applied to the state machine.

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

You have just completed the most complex and critical part of the entire Raft algorithm. You’ve built a system where a log entry is proposed, replicated, committed by a majority, and finally applied to the state machine. The abstract agreement in the log has now become a concrete change in your key-value store. This is a phenomenal achievement.

There is just one loose end to tie up. Cast your mind back to when the client first sent its Set request. Your leader node accepted the request, appended it to its log, and then… told the client to wait. That client’s connection has been held open, patiently waiting for a confirmation that its request was successful. Now that the command has been fully applied, it’s time to deliver that confirmation and close the loop.

The Challenge: Bridging the Asynchronous Gap

The task that accepted the client’s request (running in handle_connection) and the task that applied the command to the KvStore (the “Applier” task) are two completely separate, concurrent tasks. How can the Applier, upon finishing its work, notify the correct connection handler that its specific request is complete?

The answer lies in a beautiful and efficient asynchronous pattern: using one-shot channels as notifiers.

A tokio::sync::oneshot::channel is a simple, single-use channel. You create a sender (tx) and a receiver (rx). One task can hold the tx and another can .await the rx. When the first task sends a single value through tx, the second task immediately wakes up with that value. It’s the perfect tool for a “request-and-wait-for-a-response” scenario.

Our strategy will be:

When handle_connection receives a client command, it will create a oneshot channel.
It will store the tx (sender) half in a shared HashMap, keyed by the log index of its new command.
It will then efficiently go to sleep by .awaiting the rx (receiver) half.
When the “Applier” task applies the command for that log index, it will look up the corresponding tx in the map and use it to send the final response.
This wakes up the original handle_connection task, which can then write the response back to the client’s TCP stream.

Step 1: Create the Shared Notifier Map

First, let’s create the shared HashMap that will store our waiting notifiers. In src/main.rs, add the necessary use statement and create the new shared state in your main function.

// In src/main.rs, at the top
use tokio::sync::oneshot;
// ... other use statements

// In src/main.rs -> main() function
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... config parsing, majority calculation, store creation, etc.
    let raft_node = Arc::new(RwLock::new(RaftNode::new()));

    // The type for our one-shot sender.
    type Responder = oneshot::Sender<ClientResponse>;

    // This map will store the `tx` half of a channel for each pending client request,
    // keyed by the log index of the command.
    let client_notifiers: Arc<RwLock<HashMap<u64, Responder>>> = Arc::new(RwLock::new(HashMap::new()));

    // ... peer_connections, listener creation, etc.
    // ...
}

Step 2: Modify `handle_connection` to Wait for Notification

Now, we’ll update the handle_connection logic for Set and Delete. Instead of using continue, it will now register a notifier and wait. You’ll need to pass the new client_notifiers map into the handler.

First, update the call site in your main loop and the function signature.

// In src/main.rs -> main() loop
loop {
    let (stream, addr) = listener.accept().await?;
    let store_clone = store.clone();
    let raft_node_clone = raft_node.clone();
    let peer_ids_clone = peer_ids.clone();
    let client_notifiers_clone = client_notifiers.clone(); // Clone the new map

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

// In src/main.rs, update the function signature
async fn handle_connection(
    mut stream: TcpStream,
    store: KvStore,
    raft_node: Arc<RwLock<RaftNode>>,
    majority: u64,
    peer_ids: &[u64],
    client_notifiers: Arc<RwLock<HashMap<u64, Responder>>>, // Add the new parameter
) -> anyhow::Result<()> { /* ... */ }

Now, update the logic for Set and Delete.

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

                    Request::Set { key, value } => {
                        let is_leader;
                        {
                            let node = raft_node.read().await;
                            is_leader = node.state == NodeState::Leader;
                        }

                        if is_leader {

// 1. Create a one-shot channel for this specific request.
                            let (tx, rx) = oneshot::channel();

                            // 2. Append to the log and register the notifier.
                            { // Scoped write lock.
                                let mut node = raft_node.write().await;
                                let command = Command::Set { key, value };
                                let new_entry = LogEntry { term: node.current_term, command };
                                node.log.push(new_entry);
                                let log_index = node.log.len() as u64;

                                // Store the sender half, keyed by the log index.
                                let mut notifiers = client_notifiers.write().await;
                                notifiers.insert(log_index, tx);

                                println!("[Leader] Appended command at index {}. Waiting for commit...", log_index);
                            }

                            // 3. Wait for the Applier to notify us.
                            // The `?` will propagate an error if the sender is dropped,
                            // which could happen on a leader change.
                            match rx.await {
                                Ok(response) => response,
                                Err(_) => ClientResponse::Error("Request cancelled: leader status changed.".to_string()),
                            }

} else {
                            ClientResponse::Error("Not the leader.".to_string())
                        }
                    }

                    Request::Delete { key } => {
                        // The logic for Delete is identical to Set.
                        let is_leader;
                        {
                            let node = raft_node.read().await;
                            is_leader = node.state == NodeState::Leader;
                        }

                        if is_leader {

let (tx, rx) = oneshot::channel();
                            {
                                let mut node = raft_node.write().await;
                                let command = Command::Delete { key };
                                let new_entry = LogEntry { term: node.current_term, command };
                                node.log.push(new_entry);
                                let log_index = node.log.len() as u64;

                                let mut notifiers = client_notifiers.write().await;
                                notifiers.insert(log_index, tx);
                                println!("[Leader] Appended command at index {}. Waiting for commit...", log_index);
                            }

                            match rx.await {
                                Ok(response) => response,
                                Err(_) => ClientResponse::Error("Request cancelled: leader status changed.".to_string()),
                            }

} else {
                            ClientResponse::Error("Not the leader.".to_string())
                        }
                    }

Step 3: Modify the Applier Task to Send Notifications

Finally, we’ll give the Applier task the ability to send the notification. Pass the client_notifiers map to it.

// In src/main.rs -> main() function
// --- State Machine Applier Logic ---
let applier_node = raft_node.clone();
let applier_store = store.clone();
let applier_notifiers = client_notifiers.clone(); // Clone the map for the Applier.
tokio::spawn(async move {
    // ...
});

Now, update the Applier’s logic to send the response.

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

tokio::spawn(async move {
    let applier_interval = Duration::from_millis(10);
    loop {
        tokio::time::sleep(applier_interval).await;

        let mut applied_indices = Vec::new();

        { // Scoped write lock on raft_node
            let mut node = applier_node.write().await;
            if node.last_applied >= node.commit_index {
                continue;
            }

            // ... (logic to get commands_to_apply is the same) ...
            for i in (node.last_applied + 1)..=(node.commit_index) {
                // ...
                commands_to_apply.push(entry.command);
                applied_indices.push(i); // Keep track of the indices we're applying.
            }

            node.last_applied = node.commit_index;
        }

        // Apply commands to the store (unchanged)
        for command in commands_to_apply {
            // ...
        }

// --- NOTIFY CLIENTS ---\n        // After applying, notify any waiting clients.\n        let mut notifiers = applier_notifiers.write().await;\n        for index in applied_indices {\n            // If there is a notifier for this log index, remove it and send the response.\n            if let Some(sender) = notifiers.remove(&index) {\n                // The actual return value for Set/Delete doesn't matter much,\n                // just the success status.\n                let response = ClientResponse::Success(None);\n                // We don't care if the send fails; the client might have disconnected.\n                let _ = sender.send(response);\n            }\n        }\n

}\n});

The Complete Data Path

Congratulations! You have now completed the entire, end-to-end data path for a write operation in a fault-tolerant, distributed system. Let’s trace the full journey:

A client sends a Set request.
The Leader’s handle_connection task receives it, creates a oneshot channel, appends the command to its log, and stores the channel’s sender in the client_notifiers map. It then awaits the receiver.
The Leader’s replication task sees the new log entry and sends it to followers.
Followers receive, validate, and append the entry.
Followers reply successfully.
The Leader processes the replies, updates its match_index, and determines the entry is now on a majority. It advances its commit_index.
The background “Applier” task (on the Leader) sees that commit_index > last_applied.
The Applier applies the command to its local KvStore.
The Applier then checks the client_notifiers map, finds the sender for the newly applied log index, and sends a ClientResponse::Success through it.
The original handle_connection task wakes up, receives the ClientResponse, serializes it, and writes it back to the client’s TCP stream.

This is a complete, robust, and highly efficient implementation of a core distributed systems pattern.

Enhancements and Future Steps

This marks the end of a major section of the project. You have a fully functional core for a distributed key-value store. Here are some potential enhancements and a look at what’s next:

Leader Redirection: What if a client contacts a follower? Currently, it gets an error. A more advanced system would have the follower reply with the address of the leader it knows about, so the client can automatically retry the request with the correct node.
Batching: For high-throughput workloads, a leader can collect multiple client commands and replicate them in a single AppendEntries RPC, significantly reducing network overhead.
Log Compaction: Over time, the Raft log will grow indefinitely. Real-world systems need a mechanism like snapshotting to periodically save the state of the KvStore and truncate the log, preventing it from consuming all available disk space.

The next step in your project roadmap, Building a Command-Line Interface (CLI) Client, will give you a user-friendly way to interact with the powerful distributed server you have just built.