Choose a file format for the log (e.g., a simple binary format using bincode).

Choose a file format for the log (e.g., a simple binary format using `bincode`).

Mục tiêu:

With the successful creation of your command-line client, you’ve built the front door to your distributed system. Users can now knock and make requests. Now, we turn our attention back to the house itself to ensure it’s built on a solid, unshakable foundation. Currently, your Raft nodes possess a perfect, but fragile, memory. If a node restarts, its entire history—its term, who it voted for, and the precious log of commands—vanishes into the ether. For a system designed for fault tolerance, this is a critical flaw.

This step is dedicated to solving that problem by implementing persistence. We will teach our nodes to write their most critical state to disk, allowing them to survive crashes and reboots, and to rejoin the cluster with their memory intact. Our first task is to decide how to store this information on disk—to choose a file format.

What Needs to be Remembered? The Persistent State of Raft

The Raft paper is very specific about what constitutes a node’s “persistent state.” This is the minimum set of information that must be saved to stable storage (like a hard drive) to guarantee the safety of the algorithm. There are three key pieces:

current_term: The latest term the server has seen. This is crucial for preventing old leaders from causing chaos.
voted_for: The candidateId that received this node’s vote in the current term. This prevents a node from voting for more than one candidate in the same term.
log: The sequence of LogEntry commands. This is the official history of the entire system, and losing it would be catastrophic.

All other state, like commit_index, last_applied, and the leader-specific next_index and match_index, is considered “volatile.” It can be safely lost on a restart and reconstructed from the persistent state.

The Right Tool for the Job: `bincode`

To save our Rust data structures to a file, we must first serialize them into a sequence of bytes. You’ve already mastered this for your network protocol, and we will use the exact same tool for persistence: the bincode crate.

Choosing bincode is a fantastic decision for several reasons:

Efficiency: It produces a very compact binary representation, minimizing the amount of data we need to write to disk, which is often a performance bottleneck.
Performance: Serialization and deserialization with bincode are extremely fast.
Simplicity: It works seamlessly with any Rust type that derives serde’s Serialize and Deserialize traits. You’ve already done this for your LogEntry and Command types, so they are ready to be persisted with zero changes!
Consistency: Using the same serialization format for both the network and disk simplifies your project’s dependencies and concepts.

Designing the On-Disk Layout: Two-File Strategy

We have two distinct types of persistent data: the core state (current_term, voted_for) and the log. These have different access patterns:

The core state is small and needs to be updated (overwritten) atomically whenever current_term or voted_for changes.
The log is an ever-growing, append-only structure. We will constantly be adding new entries to the end of it, but rarely modifying existing ones.

Given these different patterns, a clean and performant strategy is to use two separate files:

state.bin: A file to store the core Raft state.
log.bin: A file to store the sequence of log entries.

The State File: `state.bin`

To make saving and loading the core state clean, let’s define a new struct specifically for this purpose. This is a good practice as it decouples your on-disk format from your in-memory data structures, which might contain extra volatile fields.

Let’s define this PersistentState struct in your library crate, src/lib.rs, so it can be used by your server’s main logic.

// In src/lib.rs

use serde::{Deserialize, Serialize};

// ... Request and Response enums ...

/// The core persistent state of a Raft node.
/// This is saved to disk to ensure durability across restarts.
#[derive(Serialize, Deserialize, Debug, Default)]
pub struct PersistentState {
    /// The latest term the server has seen.
    pub current_term: u64,
    /// The candidateId that received a vote in the current term.
    pub voted_for: Option<u64>,
}

pub: We make the struct and its fields public so they can be accessed from src/main.rs.
#[derive(..., Default)]: Deriving Default is a convenience that gives us a PersistentState::default() method, which will create an instance with current_term: 0 and voted_for: None. This is perfect for a brand-new node.

When we need to persist the state, we will simply create an instance of this struct, serialize it with bincode, and write the resulting bytes to state.bin, overwriting the file completely.

The Log File: `log.bin`

The log file is an ordered sequence of LogEntry items. A naive approach would be to serialize the entire Vec<LogEntry> and write it to disk every time a new entry is added. This would be incredibly inefficient, as a 1GB log would require 1GB of I/O for every single new command!

The correct approach is to treat the log file as an append-only log. When a new LogEntry arrives, we serialize only that new entry and append its bytes to the end of log.bin. This is a common and highly performant pattern used in virtually all modern databases and distributed systems.

To read this file back on startup, we’ll read and deserialize one entry at a time until we reach the end of the file. Your LogEntry and Command definitions from src/lib.rs are already perfectly prepared for this, as they derive Serialize and Deserialize.

Here they are again for reference (no changes needed):

// In src/lib.rs (already present from previous steps)

#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum Command {
    Set { key: String, value: String },
    Delete { key: String },
}

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct LogEntry {
    pub term: u64,
    pub command: Command,
}

You have now made the crucial first decision in making your service durable. By choosing bincode and a two-file, append-only log strategy, you’ve laid a simple, performant, and robust foundation for persistence.

What’s Next?

With the format and structure decided, the next task is to put it into action. You will implement the first rule of Raft persistence: before a node grants its vote to a candidate in a RequestVote RPC, it must first persist its updated current_term and voted_for state to disk. This ensures that a node never “forgets” it voted and accidentally votes for two different candidates in the same term after a crash.

Implement Raft State Persistence for `RequestVote` RPC

Mục tiêu: Implement the crucial Raft safety rule of persisting state to stable storage before responding to an RPC. This task focuses on saving the current\_term and voted\_for state using asynchronous file I/O with tokio::fs and bincode before a node grants a vote.

Excellent! You’ve made the critical design decision to use bincode for serialization and a two-file strategy for storage. You’ve even created the PersistentState struct, which is the perfect, clean representation of the core state that needs to be saved. You have the blueprint; now it’s time to build the mechanism that writes this state to disk.

This task addresses one of the most fundamental safety rules in the Raft protocol: state must be written to stable storage before a node responds to an RPC that changes that state. Our first application of this rule is in handling a RequestVote RPC. Before a node grants its vote to a candidate, it must durably record that decision. If it crashed and rebooted without this record, it might forget it ever voted and grant a vote to a different candidate in the same term, violating Raft’s core safety guarantee and potentially leading to two leaders being elected.

Asynchronous File I/O with `tokio::fs`

Since our entire server is built on the Tokio asynchronous runtime, performing standard, blocking file I/O with std::fs would be a performance disaster. A blocking file write could stall the entire thread, preventing hundreds or thousands of other asynchronous tasks from making progress.

To solve this, we will use Tokio’s own asynchronous file system module, tokio::fs. This module provides async-aware versions of standard file operations, allowing our program to write data to disk without blocking the execution thread.

Step 1: Configuring a Data Directory

Each node in our cluster needs its own separate directory to store its state files. If multiple nodes on the same machine wrote to the same files, chaos would ensue. Let’s add a configuration option to specify this directory.

First, update your Config struct in src/main.rs to include a data_dir field.

// In src/main.rs

use clap::Parser;

// ... other use statements ...

/// The configuration for a single Raft node.
#[derive(Parser, Debug, Clone)]
struct Config {
    // ... (id, addr, peers fields are unchanged) ...

    /// The directory to store persistent state (log and state file).
    /// Each node in a cluster should have a unique data directory.
    #[arg(long)]
    data_dir: String,
}

Step 2: Creating a Persistence Helper Function

To keep our RPC handling logic clean, we will encapsulate the file writing logic in a dedicated helper function. This function will take the data directory and the state to be persisted, handle serialization, and perform the asynchronous file write.

Add this new async function to src/main.rs.

// In src/main.rs

use distributed_kv::PersistentState; // Make sure this is imported from your library
use std::path::PathBuf;
use tokio::fs;

// ... other functions ...

/// Persists the core Raft state (`current_term`, `voted_for`) to a file.
/// This function is designed to be called before responding to RPCs that change this state.
async fn persist_state(data_dir: &str, state: &PersistentState) -> anyhow::Result<()> {
    // 1. Ensure the data directory exists. `create_dir_all` is like `mkdir -p`.
    fs::create_dir_all(data_dir).await?;

    // 2. Define the path to the state file.
    let path = PathBuf::from(data_dir).join("state.bin");

    // 3. Serialize the `PersistentState` struct into bytes using bincode.
    let bytes = bincode::serialize(state)?;

    // 4. Write the bytes to the file asynchronously.
    // `fs::write` handles opening, writing, and closing the file.
    // It will create the file if it doesn't exist, and truncate it if it does.
    fs::write(&path, bytes).await?;

    // On some platforms/filesystems, a write doesn't guarantee durability.
    // A sync operation is needed to flush OS buffers to the disk.
    // While omitted here for simplicity, a production system might add:
    // let file = fs::File::open(&path).await?;
    // file.sync_all().await?;

    Ok(())
}

Step 3: Integrating Persistence into the `RequestVote` Handler

Now for the main event. We need to call our new persist_state function from within the handle_connection logic, specifically in the NodeMessage::RequestVote match arm.

First, you must pass the data_dir from your config into handle_connection.

// In src/main.rs -> main() -> the main accept loop

// ... inside the loop ...
tokio::spawn(async move {
    // ...
    let data_dir_clone = config.data_dir.clone();
    if let Err(e) = handle_connection(
        stream,
        store_clone,
        raft_node_clone,
        majority,
        &peer_ids_clone,
        client_notifiers_clone,
        data_dir_clone, // Pass the new config value
    )
    .await
    // ...
});

// Update the handle_connection function signature
async fn handle_connection(
    // ... other arguments ...
    client_notifiers: Arc<RwLock<HashMap<u64, Responder>>>,
    data_dir: String, // Add the data_dir parameter
) -> anyhow::Result<()> {
// ...
}

Finally, modify the logic where a vote is granted. The call to persist_state must happen before the in-memory state is updated and before the reply is sent.

// In src/main.rs -> handle_connection() -> NodeMessage::RequestVote match arm

                        // ... inside the `if args.term >= node.current_term` block ...

                        let log_is_ok = /* ... your existing log check logic ... */;

                        // If the log is ok AND (we haven't voted OR we're voting for the same candidate again)
                        if log_is_ok && (node.voted_for.is_none() || node.voted_for == Some(args.candidate_id)) {

// --- PERSISTENCE LOGIC ---
                            // This is the crucial Raft rule: we MUST persist our state to stable
                            // storage BEFORE we respond to the RPC.

                            // 1. Create the state object that reflects our decision.
                            let new_persistent_state = PersistentState {
                                current_term: node.current_term,
                                voted_for: Some(args.candidate_id),
                            };

                            // 2. Call our helper to serialize and write the state to disk.
                            // The `?` operator ensures that if this write fails, we will
                            // not proceed to grant the vote, maintaining system safety.
                            persist_state(&data_dir, &new_persistent_state).await?;

                            println!(
                                "[Follower] Persisted vote for Candidate {} in Term {}",
                                args.candidate_id, node.current_term
                            );

// 3. Only after a successful write to disk can we update our
                            // in-memory state and grant the vote.
                            node.voted_for = Some(args.candidate_id);
                            vote_granted = true;
                        }

You have now correctly implemented the first and most important persistence rule of the Raft algorithm. Your nodes will no longer suffer from amnesia after a restart, ensuring they behave consistently and safely, even in the face of crashes.

What’s Next?

You’ve secured the state related to voting. The next logical step is to secure the most valuable asset of all: the replicated log. In the next task, you will implement the second Raft persistence rule: before a follower acknowledges an AppendEntries RPC, it must first persist the new log entries to its log file.

Implement Durable Log Persistence in Raft

Mục tiêu: Ensure the Raft replicated log is durable by persisting changes to a disk file. Implement helper functions to append new entries and rewrite the log on truncation, using a length-prefixed framing strategy, before replying to AppendEntries RPCs.

Fantastic work on implementing persistence for the core Raft state (current_term and voted_for). You’ve successfully applied one of the most fundamental rules of distributed consensus: persist state to stable storage before responding to the RPC that changes it. This principle is the bedrock of durability and safety.

Now, we will apply this exact same principle to the most valuable asset in our entire system: the replicated log. Just as a node must remember who it voted for, it is even more critical that it remembers the history of commands it has accepted. If a follower appends new entries to its in-memory log, sends a success reply to the leader, and then crashes, it will awaken with amnesia, completely unaware of the entries it promised to keep. This would violate Raft’s safety guarantees, as the leader might have already committed those entries based on the follower’s now-forgotten promise.

This task is about making that promise durable. Before a follower sends a successful AppendEntriesReply, it must first write the new log entries to its on-disk log file, log.bin.

The Append-Only Log File Strategy

As we designed in the first task of this step, we will treat log.bin as an append-only file. When new entries arrive, we will serialize them and append them to the end of the file. This is highly efficient as it avoids rewriting the entire log for every new command.

A crucial detail for an append-only file of variable-sized objects is framing. When we read the file back on startup, we need to know where one entry ends and the next begins. To solve this, we will use the exact same length-prefix framing protocol we use for our network communication: before writing each serialized entry, we will first write its length as a u64.

Step 1: Create a Helper for Appending to the Log File

To keep our RPC handler clean, let’s create a new helper function dedicated to appending serialized log entries to our log file. This function will handle opening the file in append mode, serializing the entries, and writing them with the length-prefix.

Add this new async function to src/main.rs.

// In src/main.rs

use distributed_kv::LogEntry; // Ensure LogEntry is imported
use tokio::fs::OpenOptions;
use tokio::io::{AsyncWriteExt, BufWriter};

// ... other functions like persist_state ...

/// Appends a slice of log entries to the on-disk log file.
/// Each entry is framed with its length (u64) for robust parsing on startup.
async fn append_log_entries(data_dir: &str, entries: &[LogEntry]) -> anyhow::Result<()> {
    // 1. Define the path to the log file.
    let path = PathBuf::from(data_dir).join("log.bin");

    // 2. Open the file in append mode. `create(true)` will create the file if it doesn't exist.
    // We use `tokio::fs::OpenOptions` for more control over file handling.
    let file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .await?;

    // 3. Use a BufWriter for potentially better performance with many small writes.
    let mut writer = BufWriter::new(file);

    // 4. Iterate over the entries, serialize each one, and write it to the file with framing.
    for entry in entries {
        let bytes = bincode::serialize(entry)?;
        // Write the 8-byte length prefix.
        writer.write_u64(bytes.len() as u64).await?;
        // Write the actual entry bytes.
        writer.write_all(&bytes).await?;
    }

    // 5. Ensure all buffered data is written to the file.
    writer.flush().await?;

    Ok(())
}

Step 2: Handle Log Truncation on Disk

Raft’s log repair mechanism requires followers to truncate their logs if they find an entry that conflicts with the leader’s. When we truncate the in-memory Vec<LogEntry>, we must also update the on-disk file to match. The simplest and most robust way to do this is to completely rewrite the log.bin file with the contents of the now-shorter in-memory log.

Let’s create another helper function for this purpose.

// In src/main.rs, after append_log_entries

/// Rewrites the entire on-disk log file with the current in-memory log.
/// This is used when a follower's log is truncated.
async fn rewrite_log_file(data_dir: &str, log: &[LogEntry]) -> anyhow::Result<()> {
    let path = PathBuf::from(data_dir).join("log.bin");

    // Create a temporary file to write to, ensuring an atomic update.
    let temp_path = PathBuf::from(data_dir).join("log.bin.tmp");

    let file = OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true) // Start with an empty file
        .open(&temp_path)
        .await?;

    let mut writer = BufWriter::new(file);

    // Write all entries using the same length-prefix framing.
    for entry in log {
        let bytes = bincode::serialize(entry)?;
        writer.write_u64(bytes.len() as u64).await?;
        writer.write_all(&bytes).await?;
    }
    writer.flush().await?;

    // Atomically rename the temporary file to the final log file name.
    fs::rename(temp_path, path).await?;

    Ok(())
}

Note on fs::rename: Using a temporary file and an atomic rename is a standard technique for safe file writes. It prevents a crash during the write from leaving you with a corrupted, half-written log file.

Step 3: Integrate Persistence into the `AppendEntries` Handler

Now we will call our two new helper functions from the AppendEntries RPC handler in handle_connection. The persistence operations must happen after the in-memory state is modified but before the success reply is sent.

// In src/main.rs -> handle_connection() -> NodeMessage::AppendEntries match arm

                                // --- LOG REPAIR AND APPEND ---
                                success = true;

                                // Keep track of which entries we are adding in this RPC
                                // so we can persist just those new ones.
                                let mut new_entries_to_persist = Vec::new();

                                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 {

// --- PERSISTENCE (TRUNCATION) ---
                                            // Before we truncate the in-memory log, we must persist
                                            // the soon-to-be-truncated log to disk.
                                            let new_log_slice = &node.log[.. (current_index_usize - 1)];
                                            rewrite_log_file(&data_dir, new_log_slice).await?;

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());
                                        new_entries_to_persist.push(entry.clone());
                                    }
                                }

// --- PERSISTENCE (APPEND) ---
                                // After appending to our in-memory log, we must append the new
                                // entries to our on-disk log before replying.
                                if !new_entries_to_persist.is_empty() {
                                    append_log_entries(&data_dir, &new_entries_to_persist).await?;
                                    println!(
                                        "[Follower] Persisted {} new entries. Log length is now {}.",
                                        new_entries_to_persist.len(),
                                        node.log.len()
                                    );
                                }

// Rule 5: Update the commit index.
                                if args.leader_commit > node.commit_index {
                                    // ... (commit index logic is unchanged) ...
                                }

You have now made your replicated log durable. By persisting changes to disk before acknowledging them, you have ensured that a follower’s promise to the leader can survive a crash, upholding the safety and correctness of the Raft consensus algorithm.

What’s Next?

Your nodes are now diligently writing their state and log to disk, but they still wake up from a restart with a blank slate. The entire persistence story isn’t complete until you teach them how to read these files on startup. In the next task, you will implement the startup logic to read state.bin and log.bin, restoring the node’s current_term, voted_for, and log to the exact state they were in before shutting down.

Implement Raft Node State Restoration from Disk on Startup

Mục tiêu: Implement an async Rust function to read and deserialize a Raft node’s persistent state and log entries from disk. This function will handle the initial startup sequence, allowing a node to recover its state and log from state.bin and log.bin files after a restart, effectively giving it a durable memory.

In the previous tasks, you meticulously engineered your Raft nodes to be diligent scribes. They now dutifully record every important decision—who they voted for, when their term changes, and every single log entry they accept—to stable storage on disk. This is a massive leap forward in making your system robust. However, there’s a final, critical piece missing: your nodes still suffer from amnesia. Despite their careful note-taking, they wake up from a restart with a completely blank memory, unaware of the history they so carefully preserved.

This task is about completing the cycle of memory. We will teach our nodes how to read their own journals upon startup. You will implement a function that reads the state.bin and log.bin files, deserializes their contents, and reconstructs the node’s persistent state, allowing it to seamlessly pick up right where it left off.

The Startup Choreography: Reading from Disk

The process of restoring state needs to happen at the very beginning of a node’s life, before it starts any timers or listens for any network connections. We will encapsulate this entire process into a single, clean async function. This function will be responsible for:

Reading state.bin: It will attempt to read the state file. If the file exists, it will deserialize the bytes into our PersistentState struct. If the file doesn’t exist, this is not an error; it simply means this is the first time the node is starting up, so we will use default values (term: 0, voted_for: None).
Reading log.bin: It will then attempt to read the log file. If it doesn’t exist, it will start with an empty log. If it does, it must carefully read the file one entry at a time, respecting the length-prefix framing protocol you established. It will continue reading and deserializing entries until it cleanly reaches the end of the file.

Implementing the State and Log Loader Function

Let’s create a new helper function in src/main.rs to handle this startup logic. It will take the data directory as an argument and return the restored state and log as a tuple.

// In src/main.rs

use distributed_kv::{LogEntry, PersistentState}; // Make sure these are imported
use std::path::PathBuf;
use tokio::fs;
use tokio::io::{AsyncReadExt, BufReader};

// ... other functions ...

/// Loads the persistent state and log from disk when a node starts up.
/// Returns default values if the files do not exist (i.e., for a new node).
async fn load_state_and_log(
    data_dir: &str,
) -> anyhow::Result<(PersistentState, Vec<LogEntry>)> {
    // --- 1. Load the PersistentState from `state.bin` ---
    let state_path = PathBuf::from(data_dir).join("state.bin");
    let state = match fs::read(&state_path).await {
        Ok(bytes) => {
            println!("[Startup] Found state file, deserializing...");
            bincode::deserialize(&bytes)?
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!("[Startup] No state file found, starting with default state.");
            PersistentState::default()
        }
        Err(e) => {
            // For any other I/O error, we should probably fail fast.
            return Err(e.into());
        }
    };

    // --- 2. Load the LogEntries from `log.bin` ---
    let log_path = PathBuf::from(data_dir).join("log.bin");
    let mut log = Vec::new();

    let log_file_result = fs::File::open(&log_path).await;
    match log_file_result {
        Ok(file) => {
            println!("[Startup] Found log file, reading entries...");
            let mut reader = BufReader::new(file);
            loop {
                // Read the 8-byte length prefix for the next entry.
                match reader.read_u64().await {
                    Ok(len) => {
                        let mut entry_buffer = vec![0; len as usize];
                        // Read the exact number of bytes for the entry.
                        if reader.read_exact(&mut entry_buffer).await.is_err() {
                            // This indicates a corrupted log file (length prefix without data).
                            eprintln!("[Startup] Error: Log file is corrupted. Truncating to last good entry.");
                            break;
                        }
                        let entry: LogEntry = bincode::deserialize(&entry_buffer)?;
                        log.push(entry);
                    }
                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                        // This is the clean way to exit the loop. We've read all entries.
                        println!("[Startup] Reached end of log file.");
                        break;
                    }
                    Err(e) => {
                        return Err(e.into());
                    }
                }
            }
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            println!("[Startup] No log file found, starting with an empty log.");
            // No action needed, `log` is already an empty Vec.
        }
        Err(e) => {
            return Err(e.into());
        }
    };

    println!(
        "[Startup] Loaded state: term={}, voted_for={:?}. Loaded log with {} entries.",
        state.current_term,
        state.voted_for,
        log.len()
    );

    Ok((state, log))
}

Deconstructing the Loader Logic

This function is a masterclass in robust file handling. Let’s break down the key parts.

Loading the State File (`state.bin`)

match fs::read(&state_path).await: We use a match expression to elegantly handle the different outcomes of trying to read the file.
Ok(bytes): The success case. If we read the bytes, we immediately pass them to bincode::deserialize to reconstruct the PersistentState struct.
Err(e) if e.kind() == ...::NotFound: This is the crucial part for handling a new node. We specifically check if the error’s “kind” is NotFound. If so, we treat it as a normal condition and simply return PersistentState::default().
Err(e): Any other error (like a permissions issue or a corrupted disk) is a fatal problem, so we propagate it up using return Err(e.into()).

Loading the Log File (`log.bin`)

BufReader: We wrap our file handle in a tokio::io::BufReader. This is a performance optimization. A BufReader maintains an in-memory buffer, reducing the number of direct, expensive system calls to the operating system, which is especially useful when doing many small reads.
The loop: Since the log file contains a sequence of entries, we must loop until we’ve read all of them.
match reader.read_u64().await: This is the heart of the loop’s control flow.
- Ok(len): We successfully read a length prefix. We can now proceed to read_exact to get the payload bytes and deserialize them.
- Err(e) if e.kind() == ...::UnexpectedEof: This is the expected, clean exit condition. When read_u64 tries to read 8 bytes but finds the end of the file instead, it returns this specific error. This tells us we have successfully read every complete, framed entry in the file, and we can safely break the loop.
- Error Handling: If read_exact fails after read_u64 succeeded, it implies the file is corrupt (e.g., a crash happened while writing an entry). We print a warning and break, effectively loading the log only up to the last known good entry.

You have now built the counterpart to your persistence logic. This function can safely and robustly read the on-disk state of any node, whether it’s a brand-new node starting for the first time or a veteran node rejoining the cluster after a restart.

What’s Next?

This powerful load_state_and_log function is now ready for use, but it’s currently just a standalone piece of code that is never called. The next logical and crucial task is to integrate this function into your main function. You will call it at the very beginning of your program’s execution and use the PersistentState and Vec<LogEntry> it returns to initialize your in-memory RaftNode struct, finally completing the persistence story and giving your nodes a durable memory.

Restore Raft Node State on Startup

Mục tiêu: Modify the server’s main function to load the persisted state and log from disk at startup, and use this data to initialize the in-memory RaftNode instance.

You have done some truly excellent work. In the previous task, you built the load_state_and_log function, a robust and intelligent piece of code that acts as your node’s historian. It knows how to read the journals—state.bin and log.bin—that you so carefully instructed it to write. That function holds the key to unlocking a node’s memory, but right now, it’s a key that hasn’t been turned.

This task is about that final, satisfying moment of restoration. We will now call your loader function at the very beginning of your server’s startup sequence and use the precious data it recovers to initialize your in-memory RaftNode instance. This is the bridge between the durable, on-disk world and the fast, in-memory world where your Raft logic operates. After this, your nodes will no longer be born with amnesia; they will be born with the complete wisdom of their past experiences.

The Point of Initialization: The `main` Function

A node’s state must be restored only once, right at the beginning of its life. The perfect and only place to do this is in your server’s main function, before any Raft-related background tasks are spawned.

The process is straightforward:

We’ll call the load_state_and_log function you just wrote, passing it the data_dir from the configuration.
This function will return the PersistentState and the Vec<LogEntry> it recovered from the files.
We will then construct our RaftNode instance, but instead of using default values for the persistent fields, we will populate them with the data we just loaded. All volatile state, as per the Raft protocol, will still be initialized to its default, in-memory-only values.

Let’s modify the initialization logic in src/main.rs.

Implementing the State Restoration

You will find the line let raft_node = Arc::new(RwLock::new(RaftNode::new())); near the top of your main function. We are going to replace this simple initialization with a more intelligent one that incorporates our loaded state.

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

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... (config parsing and majority calculation is unchanged) ...
    let config = Config::parse();
    let majority = (config.peers.len() as u64 + 1) / 2 + 1;

// --- STATE RESTORATION ---
    // At startup, we call our loader function to restore the node's persistent state.
    let (persistent_state, log) = load_state_and_log(&config.data_dir).await?;

    println!("[Main] Initializing Raft node from persisted state.");

    // The KvStore is created as before.
    let store = KvStore::new();

    // Now, we create the RaftNode, initializing its persistent fields
    // with the data we loaded from disk.
    let raft_node = Arc::new(RwLock::new(RaftNode {
        // All nodes start as followers, regardless of their previous state.
        state: NodeState::Follower,

        // --- Persistent state (restored from disk) ---
        current_term: persistent_state.current_term,
        voted_for: persistent_state.voted_for,
        log: log,

        // --- Volatile state (re-initialized on startup) ---
        // These values are not saved and must be reset.
        commit_index: 0,
        last_applied: 0,
        current_leader: None,

        // The election timer is always reset on startup.
        election_deadline: RaftNode::new_election_deadline(),

        // Volatile candidate state.
        votes_received: 0,

        // Volatile leader state (will be initialized if/when this node becomes leader).
        next_index: HashMap::new(),
        match_index: HashMap::new(),
    }));

// ... (client_notifiers, peer_connections, listener creation, etc., is unchanged) ...
    // The rest of your main function continues from here.
}

Deconstructing the Changes

This change is the final link in the chain of persistence.

Calling the Loader: The line let (persistent_state, log) = load_state_and_log(&config.data_dir).await?; is where the magic happens. Your main function pauses, the loader does its robust work of reading and deserializing the files, and then execution resumes with the node’s recovered memory in hand.
Direct Struct Instantiation: We are no longer calling RaftNode::new(). Instead, we are creating a RaftNode instance directly using a struct literal (RaftNode { ... }). This gives us explicit control over every field’s initial value.
A Tale of Two States: This code block is a perfect illustration of the difference between persistent and volatile state in Raft.
- Persistent (current_term, voted_for, log): These fields are populated directly from the variables (persistent_state, log) that we recovered from disk. This is the restored memory.
- Volatile (commit_index, last_applied, etc.): These fields are all reset to their default values (0, None, empty HashMaps). The Raft protocol is designed such that this information can be safely reconstructed during normal operation after a restart. For example, commit_index will be updated once the node receives an AppendEntries RPC from a leader.
Always Start as a Follower: Notice that we always initialize state: NodeState::Follower. This is a critical safety rule. A node that was previously a leader should not wake up and immediately start acting as one. It must rejoin the cluster as a follower and wait for either a heartbeat from an existing leader or for its own election timer to elapse.

You have now completed the persistence and restoration cycle. Your nodes are no longer ephemeral beings; they have a memory, a history that they carry with them across restarts, making your distributed system dramatically more robust and closer to a real-world, production-ready service.

What’s Next?

There is one final task to complete this step. Your node now wakes up with its log of commands fully intact, but its key-value store (KvStore’s HashMap) is still empty. The in-memory state machine does not yet reflect the restored history in the log. In the final task, you will implement the logic to replay all the committed log entries upon startup, applying their commands to the KvStore and ensuring the state machine is perfectly synchronized with the committed history.

Implement Startup Log Replay for State Durability

Mục tiêu: Restore the in-memory state machine (KvStore) upon node startup by iterating through the persisted command log and applying each command. This ensures the node’s state is fully recovered after a restart, achieving data durability.

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

You have reached the final and most gratifying task in making your distributed system truly durable. In an incredible sequence of steps, you’ve taught your Raft nodes to be meticulous record-keepers, writing their term, vote, and entire command history to disk. You then gave them the gift of memory, implementing a loader that reads this history upon startup. Your node now awakens from a reboot with its full log of commands loaded into memory.

However, there’s a crucial disconnect. The node remembers the history of commands, but its actual state machine—the KvStore’s HashMap—is still a blank slate, created new and empty. The final act of restoration is to bridge this gap: we must replay the historical log, applying each command to the state machine, to bring it to the exact state it was in before the shutdown. This is the moment the abstract history in the log becomes the living, queryable state of your key-value store.

The Log as the Source of Truth

This concept is fundamental to state machine replication, the core pattern that Raft implements. The replicated log is the single, authoritative source of truth for the entire system’s state. The in-memory HashMap is merely a cache, a convenient, high-performance projection of the state defined by the log. To reconstruct that cache, we simply need to process the log from beginning to end.

The process is beautifully simple:

After loading the log from disk, we will iterate through it from the first entry to the last.
For each LogEntry, we’ll extract the command (Set or Delete).
We’ll then apply that command to our KvStore instance, exactly as the background “Applier” task does during normal runtime.

Once this replay is complete, the HashMap will perfectly reflect the cumulative effect of every command the node has ever persisted.

Synchronizing the `last_applied` Index

There’s one more critical detail. The RaftNode tracks the index of the last log entry it applied to the state machine in its last_applied field. If we replay the entire log but leave last_applied at 0, the runtime “Applier” task will get confused and try to re-apply all those entries a second time.

Therefore, after replaying N entries from the log, we must initialize the last_applied field in our RaftNode to N. This correctly informs the running system that the state machine is already up-to-date with the persisted log.

Implementing the Startup Log Replay

Let’s modify your main function in src/main.rs. The logic will fit perfectly between loading the state from disk and initializing the RaftNode struct.

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

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

    // --- STATE RESTORATION ---
    // This part is already complete. It loads the node's history.
    let (persistent_state, log) = load_state_and_log(&config.data_dir).await?;

    // The KvStore is created empty, as before.
    let store = KvStore::new();

// --- STATE MACHINE REPLAY ---
    // Before we start the Raft node, we must replay the persisted log to
    // rebuild the in-memory state of our key-value store.
    println!("[Main] Replaying {} log entries to rebuild state machine...", log.len());
    for entry in &log {
        match &entry.command {
            Command::Set { key, value } => {
                // We must `await` the async `set` method.
                store.set(key.clone(), value.clone()).await;
            }
            Command::Delete { key } => {
                store.delete(key).await;
            }
        }
    }
    println!("[Main] State machine replay complete.");

    // Now, we create the RaftNode, initializing its state from disk.
    let raft_node = Arc::new(RwLock::new(RaftNode {
        state: NodeState::Follower,

        // Persistent state (restored from disk)
        current_term: persistent_state.current_term,
        voted_for: persistent_state.voted_for,
        log: log, // The log we just replayed

        // Volatile state (re-initialized on startup)
        commit_index: 0,
        // CRUCIAL: `last_applied` is initialized to the last index in the replayed log.
        last_applied: log.len() as u64,

        // The rest of the volatile state is initialized as before.
        current_leader: None,
        election_deadline: RaftNode::new_election_deadline(),
        votes_received: 0,
        next_index: HashMap::new(),
        match_index: HashMap::new(),
    }));

// ... (client_notifiers, peer_connections, listener, background tasks, etc.) ...
    // The rest of your main function continues from here.
}

Deconstructing the Changes

Order of Operations: Notice the logical flow. We first create the empty KvStore, then we populate it by replaying the log, and only then do we create the RaftNode that will manage it going forward.
The Replay Loop: The for entry in &log loop is the heart of the replay logic. It iterates through the loaded history in the exact order it was originally recorded.
Applying Commands: Inside the loop, the match statement dispatches the command to the correct KvStore method (set or delete). Because these methods are async (they acquire an async lock), we must .await them.
Initializing last_applied: The line last_applied: log.len() as u64 is the final, critical piece of the puzzle. It synchronizes the Raft node’s understanding of its state machine’s progress with the work we just performed in the replay loop.

You have now achieved full state durability! Your nodes can now crash, reboot, and rejoin the cluster with their memory and state perfectly intact. This is the defining feature of a fault-tolerant system and a monumental step in your project.

Enhancements and Next Steps

This concludes the “Implementing Basic Persistence” step. You’ve built a solid foundation. For a real-world, production system, the next enhancement to this persistence layer would be Log Compaction.

The Problem: Over time, your log.bin file will grow indefinitely. A node that has been running for weeks could have a log file gigabytes in size. This makes startup times very slow (as the entire log must be replayed) and consumes a large amount of disk space.
The Solution: Periodically, a node can perform snapshotting. It would serialize the entire HashMap at a specific log index and write it to a new snapshot.bin file. It could then safely delete all log entries up to and including that index from its log.bin file. On the next startup, it would first load the snapshot to restore the state machine instantly, and then only replay the much smaller number of log entries that came after the snapshot was taken.

Your next major step on the project roadmap is Adding Observability with Logging and Metrics. You’ve built a complex, multi-threaded, asynchronous distributed system. The next challenge is to give yourself the tools to see what’s happening inside it.