diff --git a/Cargo.toml b/Cargo.toml index 4883828..dcd0abe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "2" [workspace.package] version = "0.0.1" edition = "2024" -rust-version = "1.91.0" +rust-version = "1.95.0" license = "MIT" authors = ["Windsor Nguyen "] repository = "https://github.com/dedalus-labs/cloud9" diff --git a/clippy.toml b/clippy.toml index 5896d51..1790b1a 100644 --- a/clippy.toml +++ b/clippy.toml @@ -58,10 +58,10 @@ disallowed-methods = [ "tokio::fs::rename", "tokio::fs::write", "std::os::unix::fs::symlink", - "std::os::windows::fs::symlink_dir", - "std::os::windows::fs::symlink_file", + { path = "std::os::windows::fs::symlink_dir", allow-invalid = true }, + { path = "std::os::windows::fs::symlink_file", allow-invalid = true }, "tokio::fs::symlink", - "tokio::fs::symlink_dir", - "tokio::fs::symlink_file", + { path = "tokio::fs::symlink_dir", allow-invalid = true }, + { path = "tokio::fs::symlink_file", allow-invalid = true }, "std::thread::sleep", ] diff --git a/cloud9-node/Cargo.toml b/cloud9-node/Cargo.toml index 08fe71d..7c49808 100644 --- a/cloud9-node/Cargo.toml +++ b/cloud9-node/Cargo.toml @@ -13,7 +13,7 @@ workspace = true [dependencies] cloud9-core = { workspace = true } -cloud9-consensus = { workspace = true } +cloud9-raft = { workspace = true } cloud9-storage = { workspace = true } cloud9-proto = { workspace = true } tracing = { workspace = true } diff --git a/cloud9-node/src/lib.rs b/cloud9-node/src/lib.rs index 3192342..b402960 100644 --- a/cloud9-node/src/lib.rs +++ b/cloud9-node/src/lib.rs @@ -1,22 +1,16 @@ //! Top-level orchestration for Cloud9 nodes. -use cloud9_consensus::ConsensusConfig; +use cloud9_raft::ConsensusConfig; use cloud9_storage::StorageOptions; use tracing::{info, instrument}; /// Runtime configuration derived from CLI flags and config files. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct NodeConfig { pub storage: StorageOptions, pub consensus: ConsensusConfig, } -impl Default for NodeConfig { - fn default() -> Self { - Self { storage: StorageOptions::default(), consensus: ConsensusConfig::default() } - } -} - /// Launch the storage and consensus subsystems. #[instrument(skip_all)] pub async fn launch(config: NodeConfig) { diff --git a/cloud9/Cargo.toml b/cloud9/Cargo.toml index 07f30a8..be78566 100644 --- a/cloud9/Cargo.toml +++ b/cloud9/Cargo.toml @@ -19,7 +19,7 @@ workspace = true [dependencies] cloud9-core = { workspace = true } cloud9-storage = { workspace = true } -cloud9-consensus = { workspace = true } +cloud9-raft = { workspace = true } cloud9-proto = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/cloud9/src/main.rs b/cloud9/src/main.rs index 636613f..81ba0de 100644 --- a/cloud9/src/main.rs +++ b/cloud9/src/main.rs @@ -47,7 +47,7 @@ fn main() -> Result<()> { match cli.command { Command::Start { config } => { let config_path = config.unwrap_or_else(|| PathBuf::from("cloud9.toml")); - let maybe_config = load_config(config_path.clone())?; + let maybe_config = load_config(&config_path)?; tracing::info!(path = %config_path.display(), "booting node"); if let Some(config) = maybe_config { tracing::debug!(contents = %config, "loaded configuration"); @@ -56,7 +56,7 @@ fn main() -> Result<()> { } } Command::CheckConfig { config } => { - load_config(config.clone()) + load_config(&config) .and_then(|contents| { contents.ok_or_else(|| miette::miette!("config `{}` missing", config.display())) }) @@ -89,9 +89,9 @@ fn init_tracing(verbosity: u8, color_enabled: bool) -> Result<()> { tracing_subscriber::registry().with(filter).with(fmt_layer).try_init().into_diagnostic() } -fn load_config(path: PathBuf) -> Result> { +fn load_config(path: &PathBuf) -> Result> { if path.exists() { - fs::read_to_string(&path) + fs::read_to_string(path) .into_diagnostic() .map(Some) .with_context(|| format!("reading configuration from `{}`", path.display())) diff --git a/consensus/cloud9-raft-io/src/lib.rs b/consensus/cloud9-raft-io/src/lib.rs index 9b00bc6..932da22 100644 --- a/consensus/cloud9-raft-io/src/lib.rs +++ b/consensus/cloud9-raft-io/src/lib.rs @@ -1,4 +1,5 @@ //! I/O layer for cloud9-raft. +#![cfg_attr(test, allow(clippy::unwrap_used))] //! //! This crate provides the integration between the pure Raft state machine //! (`cloud9-raft`) and the outside world. It defines traits for: @@ -24,6 +25,8 @@ mod storage; mod transport; pub use read_index::{ReadId, ReadIndexCoordinator, ReadIndexError, ReadIndexResult}; -pub use session::{ClientId, ClientSession, DuplicateCheck, SequenceNum, SessionRequest, SessionTracker}; +pub use session::{ + ClientId, ClientSession, DuplicateCheck, SequenceNum, SessionRequest, SessionTracker, +}; pub use storage::{SnapshotData, Storage, StorageError}; pub use transport::{Transport, TransportError}; diff --git a/consensus/cloud9-raft-io/src/read_index.rs b/consensus/cloud9-raft-io/src/read_index.rs index 995a29b..bd1e278 100644 --- a/consensus/cloud9-raft-io/src/read_index.rs +++ b/consensus/cloud9-raft-io/src/read_index.rs @@ -9,7 +9,7 @@ //! 2. Leader records the current commit index as the "read index" //! 3. Leader sends a heartbeat round and waits for majority acknowledgment //! 4. Once majority confirms, the leader is guaranteed to still be leader -//! 5. The read can proceed once the state machine has applied up to read_index +//! 5. The read can proceed once the state machine has applied up to `read_index` //! //! This module provides `ReadIndexCoordinator` which tracks pending reads and //! completes them when heartbeat quorum is achieved. @@ -33,7 +33,7 @@ impl ReadId { /// A pending read request awaiting heartbeat confirmation. #[derive(Debug)] struct PendingRead { - /// The read index (commit_index when request was made). + /// The read index (`commit_index` when request was made). read_index: LogIndex, /// Heartbeat round this read is waiting on. round: u64, @@ -134,13 +134,7 @@ impl ReadIndexCoordinator { /// The caller is responsible for checking `can_serve_reads()` before calling this. pub fn request_read(&mut self, read_index: LogIndex) -> ReadId { let id = self.next_read_id.next(); - self.pending.insert( - id, - PendingRead { - read_index, - round: self.current_round, - }, - ); + self.pending.insert(id, PendingRead { read_index, round: self.current_round }); id } @@ -166,7 +160,7 @@ impl ReadIndexCoordinator { /// Record a heartbeat acknowledgment from a peer. /// /// Call this when an `AppendResponse { success: true }` is received - /// from a heartbeat (empty AppendEntries). + /// from a heartbeat (empty `AppendEntries`). pub fn record_ack(&mut self, from: NodeId) { if !self.round_acks.contains(&from) && self.voters.contains(&from) { self.round_acks.push(from); diff --git a/consensus/cloud9-raft-io/src/session.rs b/consensus/cloud9-raft-io/src/session.rs index 7079af3..afa8e21 100644 --- a/consensus/cloud9-raft-io/src/session.rs +++ b/consensus/cloud9-raft-io/src/session.rs @@ -7,7 +7,7 @@ //! //! The solution is to track client sessions: //! 1. Clients register and receive unique IDs -//! 2. Each request carries (client_id, sequence_number) +//! 2. Each request carries (`client_id`, `sequence_number`) //! 3. The state machine tracks the last completed sequence per client //! 4. Duplicate requests return cached responses //! @@ -95,10 +95,7 @@ pub struct ClientSession { impl Default for ClientSession { fn default() -> Self { - Self { - last_sequence: 0, - last_response: None, - } + Self { last_sequence: 0, last_response: None } } } @@ -138,10 +135,7 @@ pub struct SessionTracker { impl SessionTracker { /// Create a new session tracker. pub fn new() -> Self { - Self { - sessions: BTreeMap::new(), - next_client_id: 1, - } + Self { sessions: BTreeMap::new(), next_client_id: 1 } } /// Register a new client session. @@ -185,12 +179,7 @@ impl SessionTracker { /// /// Call this after executing a command. Caches the response for /// duplicate detection. - pub fn record_completion( - &mut self, - client_id: ClientId, - sequence: SequenceNum, - response: R, - ) { + pub fn record_completion(&mut self, client_id: ClientId, sequence: SequenceNum, response: R) { let session = self.sessions.entry(client_id).or_default(); if sequence > session.last_sequence { session.last_sequence = sequence; @@ -244,10 +233,7 @@ mod tests { let mut tracker: SessionTracker = SessionTracker::new(); let client = tracker.register_client(); - assert!(matches!( - tracker.check_duplicate(client, 1), - DuplicateCheck::New - )); + assert!(matches!(tracker.check_duplicate(client, 1), DuplicateCheck::New)); } #[test] @@ -274,10 +260,7 @@ mod tests { tracker.record_completion(client, 5, "response5".to_string()); // Sequence 3 is stale - assert!(matches!( - tracker.check_duplicate(client, 3), - DuplicateCheck::Stale - )); + assert!(matches!(tracker.check_duplicate(client, 3), DuplicateCheck::Stale)); } #[test] @@ -288,10 +271,7 @@ mod tests { tracker.record_completion(client, 1, "response1".to_string()); // Sequence 2 is new - assert!(matches!( - tracker.check_duplicate(client, 2), - DuplicateCheck::New - )); + assert!(matches!(tracker.check_duplicate(client, 2), DuplicateCheck::New)); } #[test] @@ -309,10 +289,7 @@ mod tests { let tracker: SessionTracker = SessionTracker::new(); let unknown = ClientId(999); - assert!(matches!( - tracker.check_duplicate(unknown, 1), - DuplicateCheck::New - )); + assert!(matches!(tracker.check_duplicate(unknown, 1), DuplicateCheck::New)); } #[test] diff --git a/consensus/cloud9-raft-io/src/storage.rs b/consensus/cloud9-raft-io/src/storage.rs index a875dcb..4a8ceeb 100644 --- a/consensus/cloud9-raft-io/src/storage.rs +++ b/consensus/cloud9-raft-io/src/storage.rs @@ -43,19 +43,28 @@ pub trait Storage: Send + Sync { /// /// Per §3.8, this must complete before responding to any RPC. /// If `sync` is true, the implementation should fsync to ensure durability. - fn save(&self, state: &Persistent, sync: bool) -> impl Future> + Send; + fn save( + &self, + state: &Persistent, + sync: bool, + ) -> impl Future> + Send; /// Load snapshot data for transfer to a slow follower (§5). /// /// Returns the snapshot data as bytes. The format is opaque to Raft — /// the state machine defines it. - fn load_snapshot(&self) -> impl Future, StorageError>> + Send; + fn load_snapshot( + &self, + ) -> impl Future, StorageError>> + Send; /// Save snapshot data received from the leader (§5). /// - /// Called when a follower receives InstallSnapshot. The implementation + /// Called when a follower receives `InstallSnapshot`. The implementation /// should atomically replace any existing snapshot. - fn save_snapshot(&self, snapshot: SnapshotData) -> impl Future> + Send; + fn save_snapshot( + &self, + snapshot: SnapshotData, + ) -> impl Future> + Send; } /// Snapshot data for transfer between nodes (§5). diff --git a/consensus/cloud9-raft-io/src/transport.rs b/consensus/cloud9-raft-io/src/transport.rs index 6fb6122..234768d 100644 --- a/consensus/cloud9-raft-io/src/transport.rs +++ b/consensus/cloud9-raft-io/src/transport.rs @@ -5,8 +5,8 @@ use std::future::Future; -use cloud9_raft::raft::Message; use cloud9_raft::NodeId; +use cloud9_raft::raft::Message; use thiserror::Error; /// Errors from transport operations. @@ -58,7 +58,10 @@ pub trait Transport: Send + Sync { /// /// Called when configuration changes. The transport should establish /// connections to new nodes and may close connections to removed nodes. - fn update_peers(&self, peers: &[NodeId]) -> impl Future> + Send; + fn update_peers( + &self, + peers: &[NodeId], + ) -> impl Future> + Send; } #[cfg(test)] diff --git a/consensus/cloud9-raft/src/lib.rs b/consensus/cloud9-raft/src/lib.rs index 90fec7b..0cf346b 100644 --- a/consensus/cloud9-raft/src/lib.rs +++ b/consensus/cloud9-raft/src/lib.rs @@ -1,6 +1,14 @@ #![forbid(unsafe_code)] #![deny(clippy::unwrap_used, clippy::expect_used)] -#![cfg_attr(any(test, doctest), allow(clippy::unwrap_used, clippy::expect_used))] +#![cfg_attr( + any(test, doctest), + allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::cast_possible_truncation + ) +)] //! Consensus drivers for Cloud9 clusters. //! diff --git a/consensus/cloud9-raft/src/raft/candidate.rs b/consensus/cloud9-raft/src/raft/candidate.rs index c2f4a28..2bb665b 100644 --- a/consensus/cloud9-raft/src/raft/candidate.rs +++ b/consensus/cloud9-raft/src/raft/candidate.rs @@ -2,11 +2,11 @@ //! //! A candidate: //! - Increments term and votes for self on entry -//! - Sends RequestVote to all peers +//! - Sends `RequestVote` to all peers //! - Wins election with majority votes → becomes Leader //! - Times out → restarts election (stays Candidate) //! - Discovers higher term → becomes Follower -//! - Receives AppendEntries from leader → becomes Follower +//! - Receives `AppendEntries` from leader → becomes Follower //! //! Candidate-specific state: //! - `votes`: set of peers who voted for us @@ -14,12 +14,12 @@ use crate::NodeId; +use super::StepResult; use super::core::Core; use super::event::{ AppendRequest, AppendResponse, Effects, Event, Message, Payload, PreVoteRequest, PreVoteResponse, VoteRequest, VoteResponse, }; -use super::StepResult; /// Candidate role state. #[derive(Debug, Clone)] @@ -35,7 +35,7 @@ pub struct Candidate { impl Candidate { /// Create a new candidate, starting an election. /// - /// This increments the term, votes for self, and sends RequestVote to all peers. + /// This increments the term, votes for self, and sends `RequestVote` to all peers. pub fn new(core: &mut Core) -> (Self, Effects) { core.start_election(); @@ -69,7 +69,7 @@ impl Candidate { from: core.id(), to: peer, term: core.term(), - payload: Payload::VoteRequest(req.clone()), + payload: Payload::VoteRequest(req), }) .collect(); @@ -86,9 +86,7 @@ impl Candidate { /// For joint consensus, requires majorities from both old and new configs. pub fn has_quorum(&self, core: &Core) -> bool { let config = core.effective_config(); - config.has_quorum(|voter| { - self.votes.contains(&voter) - }) + config.has_quorum(|voter| self.votes.contains(&voter)) } /// Process an event. Returns transition (if any) and effects. @@ -98,23 +96,25 @@ impl Candidate { // DiskWriteComplete is only relevant for leader Event::DiskWriteComplete(_) => StepResult::none(), Event::Message(msg) => { - if matches!(msg.payload, Payload::VoteRequest(_) | Payload::PreVoteRequest(_)) { - if self.recently_heard_from_leader(core) { - // Deny vote without updating term to avoid disruptive elections. - return StepResult::stay(self.deny_vote(core, &msg)); - } + if matches!(msg.payload, Payload::VoteRequest(_) | Payload::PreVoteRequest(_)) + && self.recently_heard_from_leader(core) + { + // Deny vote without updating term to avoid disruptive elections. + return StepResult::stay(Self::deny_vote(core, &msg)); } // PreVoteRequest doesn't update term (handled separately) if let Payload::PreVoteRequest(req) = msg.payload { - return StepResult::stay(self.handle_prevote_request(core, msg.from, msg.term, req)); + return StepResult::stay(Self::handle_prevote_request( + core, msg.from, msg.term, req, + )); } // Higher term: become follower if msg.term > core.term() { core.maybe_update_term(msg.term); - let leader = matches!(msg.payload, Payload::AppendRequest(_)) - .then_some(msg.from); + let leader = + matches!(msg.payload, Payload::AppendRequest(_)).then_some(msg.from); let mut effects = Effects::none().with_persist(); if matches!(msg.payload, Payload::AppendRequest(_)) { effects = effects.with_message(Message { @@ -132,15 +132,13 @@ impl Candidate { // Stale term: reject if msg.term < core.term() { - return StepResult::stay(self.reject_stale(core, &msg)); + return StepResult::stay(Self::reject_stale(core, &msg)); } match msg.payload { Payload::VoteResponse(resp) => self.handle_vote_response(core, msg.from, resp), - Payload::AppendRequest(req) => { - self.handle_append_request(core, msg.from, req) - } - Payload::VoteRequest(req) => self.handle_vote_request(core, msg.from, req), + Payload::AppendRequest(req) => self.handle_append_request(core, msg.from, req), + Payload::VoteRequest(_) => Self::handle_vote_request(core, msg.from), Payload::TimeoutNow => { // Already candidate, restart election StepResult::to_candidate(Effects::none()) @@ -199,12 +197,7 @@ impl Candidate { StepResult::to_follower(Some(from), Effects::none()) } - fn handle_vote_request( - &mut self, - core: &Core, - from: NodeId, - _req: VoteRequest, - ) -> StepResult { + fn handle_vote_request(core: &Core, from: NodeId) -> StepResult { // We already voted for ourselves this term let resp = Message { from: core.id(), @@ -220,12 +213,11 @@ impl Candidate { core.ticks.saturating_sub(self.last_contact_tick) < min_timeout } - fn deny_vote(&self, core: &Core, msg: &Message) -> Effects { + fn deny_vote(core: &Core, msg: &Message) -> Effects { let payload = match &msg.payload { - Payload::PreVoteRequest(_) => Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted: false, - }), + Payload::PreVoteRequest(_) => { + Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted: false }) + } Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), _ => return Effects::none(), }; @@ -238,7 +230,6 @@ impl Candidate { } fn handle_prevote_request( - &self, core: &Core, from: NodeId, msg_term: u64, @@ -254,14 +245,11 @@ impl Candidate { from: core.id(), to: from, term: msg_term, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted }), }) } - fn reject_stale(&self, core: &Core, msg: &Message) -> Effects { + fn reject_stale(core: &Core, msg: &Message) -> Effects { let payload = match &msg.payload { Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), Payload::AppendRequest(_) => Payload::AppendResponse(AppendResponse { @@ -283,8 +271,8 @@ impl Candidate { mod tests { use super::*; - use super::super::core::Config; use super::super::Transition; + use super::super::core::Config; fn test_setup() -> (Core, Candidate, Effects) { let config = Config::new(NodeId(0)).with_prevote(false); @@ -394,10 +382,7 @@ mod tests { }; let StepResult { transition, .. } = candidate.step(&mut core, Event::Message(msg)); - assert!(matches!( - transition, - Transition::ToFollower(Some(NodeId(1))) - )); + assert!(matches!(transition, Transition::ToFollower(Some(NodeId(1))))); } #[test] @@ -408,10 +393,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let StepResult { transition, effects } = candidate.step(&mut core, Event::Message(msg)); diff --git a/consensus/cloud9-raft/src/raft/core.rs b/consensus/cloud9-raft/src/raft/core.rs index 12199a4..c9e8d57 100644 --- a/consensus/cloud9-raft/src/raft/core.rs +++ b/consensus/cloud9-raft/src/raft/core.rs @@ -1,7 +1,7 @@ //! Core shared state for all Raft roles. //! //! The Core holds the "mathematical" Raft state that all invariants -//! are predicates over: term, voted_for, log, commit_index, config. +//! are predicates over: term, `voted_for`, log, `commit_index`, config. //! //! Safety properties (term monotonicity, log matching, leader completeness, //! state machine safety) are all about Core, independent of role. @@ -45,11 +45,11 @@ pub struct Config { /// Per Chapter 9 benchmark: "heartbeat interval... was half of the /// minimum election timeout" pub heartbeat_interval: u64, - /// Max entries per AppendEntries message. + /// Max entries per `AppendEntries` message. pub max_entries_per_msg: u64, /// How to handle membership changes. pub membership_mode: MembershipMode, - /// Enable PreVote protocol (§4.2.3). + /// Enable `PreVote` protocol (§4.2.3). /// /// When enabled, nodes send a "pre-vote" request before starting a real /// election. This prevents partitioned nodes from incrementing their term @@ -57,7 +57,7 @@ pub struct Config { pub prevote: bool, /// Enable parallel disk write optimization (§10.2.1). /// - /// When enabled, the leader can send AppendEntries to followers while its + /// When enabled, the leader can send `AppendEntries` to followers while its /// own disk write is in progress. The IO layer should signal completion /// via `Event::DiskWriteComplete`. This removes a disk write from the /// critical path, reducing latency. @@ -68,7 +68,7 @@ pub struct Config { /// Enable pipelining optimization (§10.2.2). /// /// When enabled, the leader updates `next_index` optimistically after - /// sending AppendEntries, rather than waiting for acknowledgment. This + /// sending `AppendEntries`, rather than waiting for acknowledgment. This /// allows multiple in-flight requests per follower, improving throughput. /// /// On timeout or rejection, `next_index` is reverted to `match_index + 1`. @@ -81,7 +81,7 @@ impl Config { /// Assumes 1 tick = 1ms. Values from Ongaro's dissertation: /// - Election timeout: 150–300ms (Chapter 9) /// - Heartbeat interval: 75ms (half of min election timeout, Chapter 9) - /// - PreVote: enabled by default (§4.2.3) + /// - `PreVote`: enabled by default (§4.2.3) /// - Parallel disk write: enabled by default (§10.2.1) /// - Pipelining: enabled by default (§10.2.2) pub fn new(id: NodeId) -> Self { @@ -98,36 +98,42 @@ impl Config { } /// Set the membership mode. + #[must_use] pub fn with_membership_mode(mut self, mode: MembershipMode) -> Self { self.membership_mode = mode; self } /// Set election timeout range. + #[must_use] pub fn with_election_timeout(mut self, min: u64, max: u64) -> Self { self.election_timeout = (min, max); self } /// Set heartbeat interval. + #[must_use] pub fn with_heartbeat_interval(mut self, interval: u64) -> Self { self.heartbeat_interval = interval; self } - /// Enable or disable PreVote (§4.2.3). + /// Enable or disable `PreVote` (§4.2.3). + #[must_use] pub fn with_prevote(mut self, enabled: bool) -> Self { self.prevote = enabled; self } /// Enable or disable parallel disk write optimization (§10.2.1). + #[must_use] pub fn with_parallel_disk_write(mut self, enabled: bool) -> Self { self.parallel_disk_write = enabled; self } /// Enable or disable pipelining optimization (§10.2.2). + #[must_use] pub fn with_pipelining(mut self, enabled: bool) -> Self { self.pipelining = enabled; self @@ -146,7 +152,7 @@ impl Default for Config { /// /// From §3.8: "currentTerm, votedFor, and log[] must be persisted." /// Additionally, we persist the bootstrap configuration for recovery. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct Persistent { pub term: Term, pub voted_for: Option, @@ -156,25 +162,14 @@ pub struct Persistent { pub bootstrap_config: Configuration, } -impl Default for Persistent { - fn default() -> Self { - Self { - term: 0, - voted_for: None, - log: Log::default(), - bootstrap_config: Configuration::default(), - } - } -} - /// The shared core state for all Raft roles. /// /// This struct holds everything that: -/// 1. Must be persisted (term, voted_for, log, bootstrap_config) -/// 2. Is shared across all roles (commit_index, config) +/// 1. Must be persisted (term, `voted_for`, log, `bootstrap_config`) +/// 2. Is shared across all roles (`commit_index`, config) /// 3. Tracks logical time (ticks, rng) /// -/// Role-specific state (leader_id, votes, next_index, match_index) +/// Role-specific state (`leader_id`, votes, `next_index`, `match_index`) /// lives in the role structs. pub struct Core { // Identity and configuration @@ -197,31 +192,16 @@ impl Core { /// /// The `voters` slice is the initial cluster membership (bootstrap config). pub fn new(config: Config, voters: &[NodeId]) -> Self { - let rng = config.id.0.wrapping_mul(0x5DEE_CE66_D).wrapping_add(0xB); + let rng = config.id.0.wrapping_mul(0x0005_DEEC_E66D).wrapping_add(0xB); let bootstrap = Configuration::simple(voters.to_vec()); - let mut persistent = Persistent::default(); - persistent.bootstrap_config = bootstrap; - Self { - config, - persistent, - commit_index: 0, - last_applied: 0, - ticks: 0, - rng, - } + let persistent = Persistent { bootstrap_config: bootstrap, ..Persistent::default() }; + Self { config, persistent, commit_index: 0, last_applied: 0, ticks: 0, rng } } /// Restore from persisted state. pub fn restore(config: Config, persistent: Persistent) -> Self { - let rng = config.id.0.wrapping_mul(0x5DEE_CE66_D).wrapping_add(0xB); - Self { - config, - persistent, - commit_index: 0, - last_applied: 0, - ticks: 0, - rng, - } + let rng = config.id.0.wrapping_mul(0x0005_DEEC_E66D).wrapping_add(0xB); + Self { config, persistent, commit_index: 0, last_applied: 0, ticks: 0, rng } } #[inline] @@ -309,7 +289,8 @@ impl Core { /// This is the latest configuration from the log, or the bootstrap config /// if no config entries exist. Per §4, config takes effect when appended. pub fn effective_config(&self) -> Configuration { - self.persistent.log + self.persistent + .log .config_at(self.persistent.log.last_index()) .cloned() .unwrap_or_else(|| self.persistent.bootstrap_config.clone()) @@ -333,8 +314,8 @@ impl Core { #[cfg(test)] mod tests { - use super::*; use super::super::log::EntryPayload; + use super::*; use crate::Command; fn test_core() -> Core { diff --git a/consensus/cloud9-raft/src/raft/event.rs b/consensus/cloud9-raft/src/raft/event.rs index 7ace9a5..2998609 100644 --- a/consensus/cloud9-raft/src/raft/event.rs +++ b/consensus/cloud9-raft/src/raft/event.rs @@ -41,31 +41,31 @@ pub struct Message { /// Message payload variants. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum Payload { - /// PreVote RPC (§4.2.3) - checks if election would succeed without incrementing term. + /// `PreVote` RPC (§4.2.3) - checks if election would succeed without incrementing term. PreVoteRequest(PreVoteRequest), - /// PreVote response + /// `PreVote` response PreVoteResponse(PreVoteResponse), - /// RequestVote RPC (§3.4) + /// `RequestVote` RPC (§3.4) VoteRequest(VoteRequest), - /// RequestVote response + /// `RequestVote` response VoteResponse(VoteResponse), - /// AppendEntries RPC (§3.5) + /// `AppendEntries` RPC (§3.5) AppendRequest(AppendRequest), - /// AppendEntries response + /// `AppendEntries` response AppendResponse(AppendResponse), - /// InstallSnapshot RPC (§5) - sent when follower is too far behind. + /// `InstallSnapshot` RPC (§5) - sent when follower is too far behind. InstallSnapshotRequest(InstallSnapshotRequest), - /// InstallSnapshot response + /// `InstallSnapshot` response InstallSnapshotResponse(InstallSnapshotResponse), /// Leadership transfer: target should start election immediately (§3.10) TimeoutNow, } -/// PreVote request (§4.2.3). +/// `PreVote` request (§4.2.3). /// -/// Like VoteRequest but doesn't increment the sender's term. Used to check +/// Like `VoteRequest` but doesn't increment the sender's term. Used to check /// if an election would succeed before disrupting the cluster. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct PreVoteRequest { /// The term the candidate would use if elected. pub next_term: Term, @@ -73,7 +73,7 @@ pub struct PreVoteRequest { pub last_log_term: Term, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct PreVoteResponse { /// The responder's current term. pub term: Term, @@ -81,13 +81,13 @@ pub struct PreVoteResponse { pub granted: bool, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct VoteRequest { pub last_log_index: LogIndex, pub last_log_term: Term, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct VoteResponse { pub granted: bool, } @@ -100,13 +100,13 @@ pub struct AppendRequest { pub leader_commit: LogIndex, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct AppendResponse { pub success: bool, pub last_log_index: LogIndex, } -/// InstallSnapshot request (§5, Figure 5.3). +/// `InstallSnapshot` request (§5, Figure 5.3). /// /// Sent by the leader when a follower is too far behind and needs the snapshot /// rather than log entries (§5: "if a follower's log is so far behind the @@ -115,7 +115,7 @@ pub struct AppendResponse { /// This message carries metadata only — actual snapshot data transfer is /// handled by the I/O layer. The full RPC in Figure 5.3 includes `offset`, /// `data[]`, and `done` fields for chunked transfer; we delegate that to I/O. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct InstallSnapshotRequest { /// Index of the last entry included in the snapshot. /// Corresponds to `lastIncludedIndex` in Figure 5.3. @@ -125,7 +125,7 @@ pub struct InstallSnapshotRequest { pub last_included_term: Term, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct InstallSnapshotResponse { /// True if the snapshot was accepted. /// Per §5, follower should accept if snapshot is more recent than its log. @@ -147,7 +147,7 @@ pub struct Effects { pub messages: Vec, /// Snapshot to send to a follower (leader only). /// The I/O layer should transfer the snapshot data and then send - /// an InstallSnapshotRequest to the follower. + /// an `InstallSnapshotRequest` to the follower. pub send_snapshot: Option, } @@ -167,26 +167,31 @@ impl Effects { Self::default() } + #[must_use] pub fn with_persist(mut self) -> Self { self.persist = true; self } + #[must_use] pub fn with_message(mut self, msg: Message) -> Self { self.messages.push(msg); self } + #[must_use] pub fn with_messages(mut self, msgs: impl IntoIterator) -> Self { self.messages.extend(msgs); self } + #[must_use] pub fn with_send_snapshot(mut self, snapshot: SendSnapshot) -> Self { self.send_snapshot = Some(snapshot); self } + #[must_use] pub fn merge(mut self, other: Effects) -> Self { self.persist |= other.persist; self.messages.extend(other.messages); @@ -203,14 +208,12 @@ mod tests { #[test] fn effects_builder() { - let effects = Effects::none() - .with_persist() - .with_message(Message { - from: NodeId(0), - to: NodeId(1), - term: 1, - payload: Payload::TimeoutNow, - }); + let effects = Effects::none().with_persist().with_message(Message { + from: NodeId(0), + to: NodeId(1), + term: 1, + payload: Payload::TimeoutNow, + }); assert!(effects.persist); assert_eq!(effects.messages.len(), 1); diff --git a/consensus/cloud9-raft/src/raft/follower.rs b/consensus/cloud9-raft/src/raft/follower.rs index b3b2ff1..0962688 100644 --- a/consensus/cloud9-raft/src/raft/follower.rs +++ b/consensus/cloud9-raft/src/raft/follower.rs @@ -1,7 +1,7 @@ //! Follower role. //! //! A follower: -//! - Responds to AppendEntries from the leader +//! - Responds to `AppendEntries` from the leader //! - Grants votes to candidates with up-to-date logs //! - Starts an election if election timeout elapses //! @@ -11,13 +11,13 @@ use crate::NodeId; +use super::StepResult; use super::core::Core; use super::event::{ AppendRequest, AppendResponse, Effects, Event, InstallSnapshotRequest, InstallSnapshotResponse, Message, Payload, PreVoteRequest, PreVoteResponse, VoteRequest, VoteResponse, }; use super::log::Entry; -use super::StepResult; /// Follower role state. #[derive(Debug, Clone)] @@ -34,13 +34,8 @@ impl Follower { /// Create a new follower with a random election deadline. pub fn new(core: &mut Core, leader: Option) -> Self { let timeout = core.random_election_timeout(); - let initial_contact = - core.ticks.saturating_sub(core.config.election_timeout.0); - Self { - leader, - election_deadline: core.ticks + timeout, - last_contact_tick: initial_contact, - } + let initial_contact = core.ticks.saturating_sub(core.config.election_timeout.0); + Self { leader, election_deadline: core.ticks + timeout, last_contact_tick: initial_contact } } /// Ticks until election timeout. @@ -57,15 +52,17 @@ impl Follower { Event::Message(msg) => { // Guard against disruptive elections (§4): if we've heard from a leader // recently, do not update term or grant a vote. - if matches!(msg.payload, Payload::VoteRequest(_) | Payload::PreVoteRequest(_)) { - if self.recently_heard_from_leader(core) { - return StepResult::stay(self.reject_due_to_active_leader(core, &msg)); - } + if matches!(msg.payload, Payload::VoteRequest(_) | Payload::PreVoteRequest(_)) + && self.recently_heard_from_leader(core) + { + return StepResult::stay(Self::reject_due_to_active_leader(core, &msg)); } // PreVoteRequest doesn't update term (handled separately) if let Payload::PreVoteRequest(req) = msg.payload { - return StepResult::stay(self.handle_prevote_request(core, msg.from, msg.term, req)); + return StepResult::stay(Self::handle_prevote_request( + core, msg.from, msg.term, req, + )); } // Higher term: update and reset @@ -80,12 +77,12 @@ impl Follower { // Stale term: reject if msg.term < core.term() { - return StepResult::stay(self.reject_stale(core, &msg)); + return StepResult::stay(Self::reject_stale(core, &msg)); } let StepResult { transition, mut effects } = match msg.payload { Payload::VoteRequest(req) => self.handle_vote_request(core, msg.from, req), - Payload::AppendRequest(req) => self.handle_append_request(core, msg.from, req), + Payload::AppendRequest(req) => self.handle_append_request(core, msg.from, &req), Payload::InstallSnapshotRequest(req) => { self.handle_install_snapshot(core, msg.from, req) } @@ -145,7 +142,7 @@ impl Follower { &mut self, core: &mut Core, from: NodeId, - req: AppendRequest, + req: &AppendRequest, ) -> StepResult { self.leader = Some(from); self.record_contact(core); @@ -206,7 +203,7 @@ impl Follower { StepResult::stay(effects.with_message(resp)) } - /// Handle InstallSnapshot RPC (§5, Figure 5.3). + /// Handle `InstallSnapshot` RPC (§5, Figure 5.3). /// /// Per Figure 5.3 receiver implementation: /// 1. Reply immediately if term < currentTerm (handled by caller) @@ -252,14 +249,12 @@ impl Follower { from: core.id(), to: from, term: core.term(), - payload: Payload::InstallSnapshotResponse(InstallSnapshotResponse { - success: true, - }), + payload: Payload::InstallSnapshotResponse(InstallSnapshotResponse { success: true }), }; StepResult::stay(Effects::none().with_persist().with_message(resp)) } - fn reject_stale(&self, core: &Core, msg: &Message) -> Effects { + fn reject_stale(core: &Core, msg: &Message) -> Effects { let payload = match &msg.payload { Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), Payload::AppendRequest(_) => Payload::AppendResponse(AppendResponse { @@ -276,13 +271,12 @@ impl Follower { }) } - fn reject_due_to_active_leader(&self, core: &Core, msg: &Message) -> Effects { + fn reject_due_to_active_leader(core: &Core, msg: &Message) -> Effects { // Deny vote without updating term to preserve the current leader. let payload = match &msg.payload { - Payload::PreVoteRequest(_) => Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted: false, - }), + Payload::PreVoteRequest(_) => { + Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted: false }) + } Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), _ => return Effects::none(), }; @@ -295,7 +289,6 @@ impl Follower { } fn handle_prevote_request( - &self, core: &Core, from: NodeId, msg_term: u64, @@ -312,10 +305,7 @@ impl Follower { from: core.id(), to: from, term: msg_term, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted }), }) } @@ -342,9 +332,9 @@ mod tests { use super::*; use crate::Command; + use super::super::Transition; use super::super::core::Config; use super::super::log::EntryPayload; - use super::super::Transition; fn test_setup() -> (Core, Follower) { let config = Config::new(NodeId(0)) @@ -386,10 +376,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let StepResult { transition, effects } = follower.step(&mut core, Event::Message(msg)); @@ -420,10 +407,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let StepResult { effects, .. } = follower.step(&mut core, Event::Message(msg)); @@ -444,10 +428,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; follower.step(&mut core, Event::Message(msg1)); @@ -456,10 +437,7 @@ mod tests { from: NodeId(2), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let StepResult { effects, .. } = follower.step(&mut core, Event::Message(msg2)); @@ -525,12 +503,7 @@ mod tests { let (mut core, mut follower) = test_setup(); core.persistent.term = 1; - let msg = Message { - from: NodeId(1), - to: NodeId(0), - term: 1, - payload: Payload::TimeoutNow, - }; + let msg = Message { from: NodeId(1), to: NodeId(0), term: 1, payload: Payload::TimeoutNow }; let StepResult { transition, .. } = follower.step(&mut core, Event::Message(msg)); assert!(matches!(transition, Transition::ToCandidate)); diff --git a/consensus/cloud9-raft/src/raft/leader.rs b/consensus/cloud9-raft/src/raft/leader.rs index e66fc49..e39b9fd 100644 --- a/consensus/cloud9-raft/src/raft/leader.rs +++ b/consensus/cloud9-raft/src/raft/leader.rs @@ -1,9 +1,9 @@ //! Leader role. //! //! A leader: -//! - Sends heartbeats (empty AppendEntries) periodically +//! - Sends heartbeats (empty `AppendEntries`) periodically //! - Replicates log entries to all followers -//! - Tracks replication progress per peer (next_index, match_index) +//! - Tracks replication progress per peer (`next_index`, `match_index`) //! - Commits entries once replicated to a majority //! - Steps down on discovering higher term //! @@ -14,18 +14,18 @@ use crate::{Command, LogIndex, NodeId, TransferError}; +use super::StepResult; use super::core::Core; use super::event::{ AppendRequest, AppendResponse, Effects, Event, Message, Payload, PreVoteResponse, SendSnapshot, - VoteRequest, VoteResponse, + VoteResponse, }; use super::log::{Entry, EntryPayload}; use super::membership::{ConfigChange, ConfigChangeError}; -use super::StepResult; /// Result of processing committed config entries. enum ConfigCommitResult { - /// Continue processing (appended C_new, may need to commit it). + /// Continue processing (appended `C_new`, may need to commit it). Continue(Effects), /// Must step down (removed from config), with optional leadership transfer. StepDown(Effects), @@ -36,7 +36,7 @@ enum ConfigCommitResult { /// Leader role state. #[derive(Debug, Clone)] pub struct Leader { - /// Next log index to send to each peer (keyed by NodeId). + /// Next log index to send to each peer (keyed by `NodeId`). pub next_index: std::collections::BTreeMap, /// Highest log index known to be replicated on each peer. pub match_index: std::collections::BTreeMap, @@ -127,15 +127,11 @@ impl Leader { /// Propose a command for replication. /// - /// Returns (index, effects, should_step_down). The caller should check - /// should_step_down and transition to follower if true. + /// Returns (index, effects, `should_step_down`). The caller should check + /// `should_step_down` and transition to follower if true. pub fn propose(&mut self, core: &mut Core, cmd: Command) -> (LogIndex, Effects, bool) { let index = core.log().last_index() + 1; - let entry = Entry { - term: core.term(), - index, - payload: EntryPayload::Command(cmd), - }; + let entry = Entry { term: core.term(), index, payload: EntryPayload::Command(cmd) }; core.log_mut().append(entry); // Update own match_index only if parallel_disk_write is disabled. @@ -158,24 +154,28 @@ impl Leader { /// Per §4, the new configuration takes effect immediately when appended /// (not when committed). Only one config change can be pending at a time. /// - /// Returns (index, effects, should_step_down) on success. + /// Returns (index, effects, `should_step_down`) on success. pub fn propose_config_change( &mut self, core: &mut Core, - change: ConfigChange, + change: &ConfigChange, ) -> Result<(LogIndex, Effects, bool), ConfigChangeError> { // Validate the change let current = core.effective_config(); change.validate(¤t, core.config.membership_mode)?; // Enforce learner catch-up before promotion. - if let ConfigChange::AddVoter(id) = change { + if let ConfigChange::AddVoter(id) = *change { // If we're promoting a learner, ensure it is caught up. if current.is_learner(id) { let progressed = self.match_index.get(&id).copied().unwrap_or(0); let need = core.log().last_index(); if progressed < need { - return Err(ConfigChangeError::LearnerNotCaughtUp { id, have: progressed, need }); + return Err(ConfigChangeError::LearnerNotCaughtUp { + id, + have: progressed, + need, + }); } } else { // Should have been rejected by validate(), but double-check. @@ -193,11 +193,7 @@ impl Leader { // Create config entry let index = core.log().last_index() + 1; - let entry = Entry { - term: core.term(), - index, - payload: EntryPayload::Config(new_config), - }; + let entry = Entry { term: core.term(), index, payload: EntryPayload::Config(new_config) }; core.log_mut().append(entry); // Config takes effect immediately (§4) - effective_config() reflects this @@ -235,7 +231,7 @@ impl Leader { /// Transfer leadership to the target node (§3.11). /// - /// Sends TimeoutNow to the target if it is caught up. The target will + /// Sends `TimeoutNow` to the target if it is caught up. The target will /// immediately start an election and (assuming it wins) become leader. /// /// # Errors @@ -285,7 +281,7 @@ impl Leader { Event::Message(msg) => { // PreVoteRequest doesn't update term - deny as leader if let Payload::PreVoteRequest(_) = msg.payload { - return StepResult::stay(self.deny_prevote(core, msg.from)); + return StepResult::stay(Self::deny_prevote(core, msg.from)); } // Higher term: step down @@ -298,12 +294,14 @@ impl Leader { // Stale term: reject if msg.term < core.term() { - return StepResult::stay(self.reject_stale(core, &msg)); + return StepResult::stay(Self::reject_stale(core, &msg)); } match msg.payload { - Payload::AppendResponse(resp) => self.handle_append_response(core, msg.from, resp), - Payload::VoteRequest(req) => self.handle_vote_request(core, msg.from, req), + Payload::AppendResponse(resp) => { + self.handle_append_response(core, msg.from, resp) + } + Payload::VoteRequest(_) => Self::handle_vote_request(core, msg.from), _ => StepResult::none(), } } @@ -313,7 +311,7 @@ impl Leader { /// Handle disk write completion (§10.2.1 parallel disk write). /// /// Called by the IO layer when persistent state has been written to disk. - /// Updates the leader's own match_index, which may allow more commits. + /// Updates the leader's own `match_index`, which may allow more commits. fn handle_disk_write_complete(&mut self, core: &mut Core, index: LogIndex) -> StepResult { // Update our own match_index to reflect persisted state let current = self.match_index.get(&core.id()).copied().unwrap_or(0); @@ -385,12 +383,7 @@ impl Leader { } } - fn handle_vote_request( - &self, - core: &Core, - from: NodeId, - _req: VoteRequest, - ) -> StepResult { + fn handle_vote_request(core: &Core, from: NodeId) -> StepResult { // We're leader in this term, deny vote let resp = Message { from: core.id(), @@ -401,7 +394,7 @@ impl Leader { StepResult::stay(Effects::none().with_message(resp)) } - fn deny_prevote(&self, core: &Core, from: NodeId) -> Effects { + fn deny_prevote(core: &Core, from: NodeId) -> Effects { // As leader, always deny pre-votes - we're the authority for this term Effects::none().with_message(Message { from: core.id(), @@ -414,7 +407,7 @@ impl Leader { }) } - fn reject_stale(&self, core: &Core, msg: &Message) -> Effects { + fn reject_stale(core: &Core, msg: &Message) -> Effects { let payload = match &msg.payload { Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), Payload::AppendRequest(_) => Payload::AppendResponse(AppendResponse { @@ -431,12 +424,12 @@ impl Leader { }) } - /// Try to advance commit_index based on match_index majority. + /// Try to advance `commit_index` based on `match_index` majority. /// /// For joint consensus, requires majorities from both old and new configs. - /// Returns (effects, should_step_down) where: + /// Returns (effects, `should_step_down`) where: /// - effects: messages to send (e.g., for auto-completing joint consensus or leadership transfer) - /// - should_step_down: true if we should step down (removed from config) + /// - `should_step_down`: true if we should step down (removed from config) pub fn maybe_commit(&mut self, core: &mut Core) -> (Effects, bool) { let mut effects = Effects::none(); @@ -466,7 +459,7 @@ impl Leader { (effects, false) } - /// Find the highest log index that has quorum and update commit_index. + /// Find the highest log index that has quorum and update `commit_index`. /// /// Only commits entries from the current term (§3.6.2). fn advance_commit_index(&self, core: &mut Core) { @@ -492,8 +485,12 @@ impl Leader { /// Process config entries that were just committed. /// - /// Returns whether to continue (possibly appended C_new), step down, or done. - fn process_committed_configs(&mut self, core: &mut Core, old_commit: LogIndex) -> ConfigCommitResult { + /// Returns whether to continue (possibly appended `C_new`), step down, or done. + fn process_committed_configs( + &mut self, + core: &mut Core, + old_commit: LogIndex, + ) -> ConfigCommitResult { for idx in (old_commit + 1)..=core.commit_index { let Some(entry) = core.log().get(idx) else { continue }; let EntryPayload::Config(ref cfg) = entry.payload else { continue }; @@ -518,9 +515,13 @@ impl Leader { /// Try to automatically transfer leadership to a caught-up voter (§4.2.2). /// /// Called when the leader is removed from the configuration. Finds the most - /// caught-up voter in the new config and sends TimeoutNow to trigger immediate + /// caught-up voter in the new config and sends `TimeoutNow` to trigger immediate /// election. - fn try_auto_transfer(&self, core: &Core, new_config: &super::membership::Configuration) -> Effects { + fn try_auto_transfer( + &self, + core: &Core, + new_config: &super::membership::Configuration, + ) -> Effects { let last_index = core.log().last_index(); // Find voters in the new config that are caught up (match_index == last_index) @@ -548,22 +549,22 @@ impl Leader { } // Only send TimeoutNow if we found a caught-up target - if let Some((target, match_idx)) = best_target { - if match_idx >= last_index { - return Effects::none().with_message(Message { - from: core.id(), - to: target, - term: core.term(), - payload: Payload::TimeoutNow, - }); - } + if let Some((target, match_idx)) = best_target + && match_idx >= last_index + { + return Effects::none().with_message(Message { + from: core.id(), + to: target, + term: core.term(), + payload: Payload::TimeoutNow, + }); } // No suitable target found - just step down and let election happen Effects::none() } - /// Append the C_new config to complete a joint consensus transition. + /// Append the `C_new` config to complete a joint consensus transition. fn append_transition_target( &mut self, core: &mut Core, @@ -583,22 +584,20 @@ impl Leader { self.broadcast_append(core).with_persist() } - /// Send AppendEntries to all peers. + /// Send `AppendEntries` to all peers. fn broadcast_append(&mut self, core: &Core) -> Effects { core.effective_config() .replication_peers(core.id()) - .fold(Effects::none(), |eff, peer| { - eff.merge(self.send_append_to(core, peer)) - }) + .fold(Effects::none(), |eff, peer| eff.merge(self.send_append_to(core, peer))) } - /// Send AppendEntries to a specific peer, or request snapshot if entries unavailable. + /// Send `AppendEntries` to a specific peer, or request snapshot if entries unavailable. /// /// Per §5: "if a follower's log is so far behind the leader's that the /// leader has discarded the next entry it needs to send to the follower", - /// the leader sends InstallSnapshot instead. + /// the leader sends `InstallSnapshot` instead. /// - /// When pipelining is enabled (§10.2.2), optimistically updates next_index + /// When pipelining is enabled (§10.2.2), optimistically updates `next_index` /// after sending, allowing multiple in-flight requests. fn send_append_to(&mut self, core: &Core, to: NodeId) -> Effects { let next = *self.next_index.get(&to).unwrap_or(&1); @@ -625,7 +624,7 @@ impl Leader { let last_log = core.log().last_index(); let end = (next + core.config.max_entries_per_msg).min(last_log + 1); let entries: Vec<_> = core.log().slice(next, end).to_vec(); - let sent_up_to = if entries.is_empty() { next } else { entries.last().unwrap().index + 1 }; + let sent_up_to = entries.last().map_or(next, |entry| entry.index + 1); // Pipelining (§10.2.2): optimistically update next_index after sending if core.config.pipelining && sent_up_to > next { @@ -652,9 +651,10 @@ impl Leader { mod tests { use super::*; + use super::super::Transition; use super::super::core::Config; + use super::super::event::VoteRequest; use super::super::log::EntryPayload; - use super::super::Transition; fn test_setup() -> (Core, Leader, Effects) { let config = Config::new(NodeId(0)) @@ -686,7 +686,7 @@ mod tests { let (core, leader, _) = test_setup(); for id in core.effective_config().all_nodes() { - assert_eq!(*leader.next_index.get(&id).unwrap(), 1); // last_index(0) + 1 + assert_eq!(leader.next_index[&id], 1); // last_index(0) + 1 } } @@ -721,9 +721,7 @@ mod tests { #[test] fn single_node_commits_immediately() { - let config = Config::new(NodeId(0)) - .with_parallel_disk_write(false) - .with_pipelining(false); + let config = Config::new(NodeId(0)).with_parallel_disk_write(false).with_pipelining(false); let mut core = Core::new(config, &[NodeId(0)]); core.persistent.term = 1; let (mut leader, _) = Leader::new(&mut core); @@ -746,10 +744,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 1, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 1 }), }; leader.step(&mut core, Event::Message(msg)); @@ -786,7 +781,7 @@ mod tests { let StepResult { effects, .. } = leader.step(&mut core, Event::Message(msg)); // Should decrement next_index and retry - assert!(*leader.next_index.get(&NodeId(1)).unwrap() <= 1); + assert!(leader.next_index[&NodeId(1)] <= 1); assert_eq!(effects.messages.len(), 1); } @@ -798,10 +793,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 5, // Higher than our term 1 - payload: Payload::AppendResponse(AppendResponse { - success: false, - last_log_index: 0, - }), + payload: Payload::AppendResponse(AppendResponse { success: false, last_log_index: 0 }), }; let StepResult { transition, effects } = leader.step(&mut core, Event::Message(msg)); @@ -856,10 +848,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let StepResult { transition, effects } = leader.step(&mut core, Event::Message(msg)); @@ -959,9 +948,7 @@ mod tests { #[test] fn can_serve_reads_after_current_term_commit() { // Use single-node cluster where propose commits immediately - let config = Config::new(NodeId(0)) - .with_parallel_disk_write(false) - .with_pipelining(false); + let config = Config::new(NodeId(0)).with_parallel_disk_write(false).with_pipelining(false); let mut single_core = Core::new(config, &[NodeId(0)]); single_core.persistent.term = 1; let (mut single_leader, _) = Leader::new(&mut single_core); @@ -979,9 +966,7 @@ mod tests { #[test] fn read_index_returns_commit_index() { - let config = Config::new(NodeId(0)) - .with_parallel_disk_write(false) - .with_pipelining(false); + let config = Config::new(NodeId(0)).with_parallel_disk_write(false).with_pipelining(false); let mut core = Core::new(config, &[NodeId(0)]); core.persistent.term = 1; let (mut leader, _) = Leader::new(&mut core); diff --git a/consensus/cloud9-raft/src/raft/log.rs b/consensus/cloud9-raft/src/raft/log.rs index 932b052..b51adc4 100644 --- a/consensus/cloud9-raft/src/raft/log.rs +++ b/consensus/cloud9-raft/src/raft/log.rs @@ -107,9 +107,7 @@ impl Log { if index == 0 || index <= self.snapshot_index { 0 } else { - self.to_vec_index(index) - .and_then(|i| self.entries.get(i)) - .map_or(0, |e| e.term) + self.to_vec_index(index).and_then(|i| self.entries.get(i)).map_or(0, |e| e.term) } } @@ -153,7 +151,8 @@ impl Log { } // Remove entries up to and including index - let entries_to_remove = (index - self.snapshot_index) as usize; + let entries_to_remove = + usize::try_from(index - self.snapshot_index).unwrap_or(self.entries.len()); if entries_to_remove >= self.entries.len() { self.entries.clear(); } else { @@ -164,7 +163,7 @@ impl Log { self.snapshot_term = term; } - /// Install snapshot metadata (for followers receiving InstallSnapshot). + /// Install snapshot metadata (for followers receiving `InstallSnapshot`). /// /// Discards all entries and sets snapshot metadata. pub fn install_snapshot(&mut self, index: LogIndex, term: Term) { @@ -187,8 +186,7 @@ impl Log { let Some(s) = self.to_vec_index(actual_start) else { return &[]; }; - let e = self.to_vec_index(end.min(self.last_index())) - .map_or(self.entries.len(), |i| i + 1); + let e = self.to_vec_index(end.min(self.last_index())).map_or(self.entries.len(), |i| i + 1); if s >= self.entries.len() || s >= e { return &[]; @@ -212,23 +210,21 @@ impl Log { /// Find the latest configuration entry at or before the given index. /// - /// Only searches available entries (after snapshot_index). + /// Only searches available entries (after `snapshot_index`). /// Returns None if no config entry exists in available range. pub fn config_at(&self, index: LogIndex) -> Option<&Configuration> { if index <= self.snapshot_index { return None; } - let end_vec_idx = self.to_vec_index(index) + let end_vec_idx = self + .to_vec_index(index) .map_or(self.entries.len(), |i| (i + 1).min(self.entries.len())); - self.entries[..end_vec_idx] - .iter() - .rev() - .find_map(|e| e.payload.as_config()) + self.entries[..end_vec_idx].iter().rev().find_map(|e| e.payload.as_config()) } /// Find the index of the latest uncommitted configuration entry. /// - /// Returns Some(index) if there's a config entry after commit_index. + /// Returns Some(index) if there's a config entry after `commit_index`. pub fn pending_config_index(&self, commit_index: LogIndex) -> Option { self.entries .iter() @@ -245,7 +241,7 @@ impl Log { if index <= self.snapshot_index { None } else { - Some((index - self.snapshot_index - 1) as usize) + usize::try_from(index - self.snapshot_index - 1).ok() } } } @@ -256,19 +252,11 @@ mod tests { use crate::NodeId; fn entry(term: Term, index: LogIndex) -> Entry { - Entry { - term, - index, - payload: EntryPayload::Command(Command(vec![index as u8])), - } + Entry { term, index, payload: EntryPayload::Command(Command(vec![index as u8])) } } fn config_entry(term: Term, index: LogIndex, voters: Vec) -> Entry { - Entry { - term, - index, - payload: EntryPayload::Config(Configuration::simple(voters)), - } + Entry { term, index, payload: EntryPayload::Config(Configuration::simple(voters)) } } #[test] @@ -446,7 +434,7 @@ mod tests { assert_eq!(log.snapshot_term(), 2); assert!(log.get(3).is_none()); assert_eq!(log.last_index(), 3); // last_index == snapshot_index when empty - assert_eq!(log.last_term(), 2); // last_term == snapshot_term when empty + assert_eq!(log.last_term(), 2); // last_term == snapshot_term when empty } #[test] diff --git a/consensus/cloud9-raft/src/raft/membership.rs b/consensus/cloud9-raft/src/raft/membership.rs index d09b8e6..e5804ab 100644 --- a/consensus/cloud9-raft/src/raft/membership.rs +++ b/consensus/cloud9-raft/src/raft/membership.rs @@ -44,7 +44,7 @@ pub enum MembershipMode { /// Can be either a simple configuration (single voter set) or a joint /// configuration (requires majorities from both old and new). /// -/// Internally uses BTreeSet for O(log n) membership checks and zero-allocation +/// Internally uses `BTreeSet` for O(log n) membership checks and zero-allocation /// iteration. Serializes as sorted vectors for determinism. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum Configuration { @@ -73,9 +73,9 @@ pub enum Configuration { }, } -/// Serde helper to serialize BTreeSet as Vec for stable output. +/// Serde helper to serialize `BTreeSet` as Vec for stable output. mod btreeset_as_vec { - use super::*; + use super::{BTreeSet, Deserialize, NodeId}; use serde::{Deserializer, Serializer}; pub(super) fn serialize( @@ -101,10 +101,7 @@ mod btreeset_as_vec { impl Configuration { /// Create a simple configuration from voters. pub fn simple(voters: impl IntoIterator) -> Self { - Self::Simple { - voters: voters.into_iter().collect(), - learners: BTreeSet::new(), - } + Self::Simple { voters: voters.into_iter().collect(), learners: BTreeSet::new() } } /// Create a simple configuration with explicit learners. @@ -172,16 +169,14 @@ impl Configuration { /// Check if a node is a learner. pub fn is_learner(&self, id: NodeId) -> bool { match self { - Self::Simple { learners, .. } => learners.contains(&id), - Self::Joint { learners, .. } => learners.contains(&id), + Self::Simple { learners, .. } | Self::Joint { learners, .. } => learners.contains(&id), } } /// All learners (non-voting). No allocation. pub fn learners(&self) -> impl Iterator + '_ { let learners = match self { - Self::Simple { learners, .. } => learners, - Self::Joint { learners, .. } => learners, + Self::Simple { learners, .. } | Self::Joint { learners, .. } => learners, }; learners.iter().copied() } @@ -225,7 +220,7 @@ impl Configuration { /// Number of votes needed for quorum in each voter set. /// /// For simple configs, returns (quorum, None). - /// For joint configs, returns (old_quorum, Some(new_quorum)). + /// For joint configs, returns (`old_quorum`, `Some(new_quorum)`). pub fn quorum_sizes(&self) -> (usize, Option) { match self { Self::Simple { voters, .. } => (voters.len() / 2 + 1, None), @@ -240,19 +235,19 @@ impl Configuration { match self { Self::Simple { voters, .. } => { let votes = voters.iter().filter(|&&id| has_vote(id)).count(); - votes >= voters.len() / 2 + 1 + votes > voters.len() / 2 } Self::Joint { old, new, .. } => { let old_votes = old.iter().filter(|&&id| has_vote(id)).count(); let new_votes = new.iter().filter(|&&id| has_vote(id)).count(); - old_votes >= old.len() / 2 + 1 && new_votes >= new.len() / 2 + 1 + old_votes > old.len() / 2 && new_votes > new.len() / 2 } } } - /// Iterate over voters, yielding (voter, old_member, new_member). + /// Iterate over voters, yielding (voter, `old_member`, `new_member`). /// - /// For simple configs, both old_member and new_member are true for all voters. + /// For simple configs, both `old_member` and `new_member` are true for all voters. /// For joint configs, indicates membership in each set. pub fn voter_membership(&self) -> Vec<(NodeId, bool, bool)> { match self { @@ -276,24 +271,20 @@ impl Configuration { /// Get the target configuration for completing a joint transition. /// - /// Returns Some(new_config) if this is a joint config, None if simple. + /// Returns `Some(new_config)` if this is a joint config, None if simple. pub fn transition_target(&self) -> Option { match self { Self::Simple { .. } => None, - Self::Joint { new, learners, .. } => Some(Configuration::Simple { - voters: new.clone(), - learners: learners.clone(), - }), + Self::Joint { new, learners, .. } => { + Some(Configuration::Simple { voters: new.clone(), learners: learners.clone() }) + } } } } impl Default for Configuration { fn default() -> Self { - Self::Simple { - voters: BTreeSet::new(), - learners: BTreeSet::new(), - } + Self::Simple { voters: BTreeSet::new(), learners: BTreeSet::new() } } } @@ -358,7 +349,7 @@ pub enum ConfigChange { AddLearner(NodeId), /// Remove a learner. RemoveLearner(NodeId), - /// Set a completely new configuration (only valid in JointConsensus mode). + /// Set a completely new configuration (only valid in `JointConsensus` mode). SetMembers(Members), } @@ -418,8 +409,10 @@ impl ConfigChange { } if mode == MembershipMode::SingleServer { // Validate single-server constraint: at most one voter changes - let added: Vec<_> = new_voters.iter().filter(|id| !voters.contains(id)).collect(); - let removed: Vec<_> = voters.iter().filter(|id| !new_voters.contains(id)).collect(); + let added: Vec<_> = + new_voters.iter().filter(|id| !voters.contains(id)).collect(); + let removed: Vec<_> = + voters.iter().filter(|id| !new_voters.contains(id)).collect(); if added.len() + removed.len() > 1 { return Err(ConfigChangeError::NotSingleChange { @@ -432,10 +425,8 @@ impl ConfigChange { return Err(ConfigChangeError::WouldBeEmpty); } // Ensure no overlap between voters and learners - if new_voters.iter().any(|id| new_learners.contains(id)) { - return Err(ConfigChangeError::AlreadyVoter( - *new_voters.iter().find(|id| new_learners.contains(id)).unwrap(), - )); + if let Some(id) = new_voters.iter().find(|id| new_learners.contains(id)) { + return Err(ConfigChangeError::AlreadyVoter(*id)); } } } @@ -445,8 +436,8 @@ impl ConfigChange { /// Apply this change to produce the next configuration. /// - /// For SingleServer mode, returns the new simple configuration directly. - /// For JointConsensus mode, returns a joint configuration (for SetMembers) + /// For `SingleServer` mode, returns the new simple configuration directly. + /// For `JointConsensus` mode, returns a joint configuration (for `SetMembers`) /// or simple configuration (for Add/Remove which are single changes). pub fn apply(&self, current: &Configuration, mode: MembershipMode) -> Configuration { let (voters, learners): (BTreeSet, BTreeSet) = match current { @@ -460,51 +451,32 @@ impl ConfigChange { new_voters.insert(*id); let mut new_learners = learners; new_learners.remove(id); - Configuration::Simple { - voters: new_voters, - learners: new_learners, - } + Configuration::Simple { voters: new_voters, learners: new_learners } } ConfigChange::RemoveVoter(id) => { let mut new_voters = voters; new_voters.remove(id); - Configuration::Simple { - voters: new_voters, - learners, - } + Configuration::Simple { voters: new_voters, learners } } ConfigChange::AddLearner(id) => { let mut new_learners = learners; new_learners.insert(*id); - Configuration::Simple { - voters, - learners: new_learners, - } + Configuration::Simple { voters, learners: new_learners } } ConfigChange::RemoveLearner(id) => { let mut new_learners = learners; new_learners.remove(id); - Configuration::Simple { - voters, - learners: new_learners, - } + Configuration::Simple { voters, learners: new_learners } } ConfigChange::SetMembers(members) => { let new_voters: BTreeSet<_> = members.voters().iter().copied().collect(); let new_learners: BTreeSet<_> = members.learners().iter().copied().collect(); if mode == MembershipMode::JointConsensus && voters != new_voters { // Enter joint consensus - Configuration::Joint { - old: voters, - new: new_voters, - learners: new_learners, - } + Configuration::Joint { old: voters, new: new_voters, learners: new_learners } } else { // Single-server mode or no actual change - Configuration::Simple { - voters: new_voters, - learners: new_learners, - } + Configuration::Simple { voters: new_voters, learners: new_learners } } } } @@ -534,7 +506,7 @@ pub enum ConfigChangeError { WouldBeEmpty, /// No actual change (voters and learners identical). NoChange, - /// SingleServer mode requires exactly one change at a time. + /// `SingleServer` mode requires exactly one change at a time. NotSingleChange { added: usize, removed: usize }, /// This node is not the leader. NotLeader, @@ -544,24 +516,23 @@ impl std::fmt::Display for ConfigChangeError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::ChangeInProgress => write!(f, "configuration change already in progress"), - Self::AlreadyVoter(id) => write!(f, "{} is already a voter", id), - Self::AlreadyLearner(id) => write!(f, "{} is already a learner", id), - Self::Overlap(id) => write!(f, "{} cannot be both voter and learner", id), + Self::AlreadyVoter(id) => write!(f, "{id} is already a voter"), + Self::AlreadyLearner(id) => write!(f, "{id} is already a learner"), + Self::Overlap(id) => write!(f, "{id} cannot be both voter and learner"), Self::PromoteRequiresLearner(id) => { - write!(f, "{} must be added as a learner before becoming a voter", id) + write!(f, "{id} must be added as a learner before becoming a voter") } Self::LearnerNotCaughtUp { id, have, need } => { - write!(f, "{} not caught up: have {}, need {}", id, have, need) + write!(f, "{id} not caught up: have {have}, need {need}") } - Self::NotVoter(id) => write!(f, "{} is not a voter", id), - Self::NotLearner(id) => write!(f, "{} is not a learner", id), + Self::NotVoter(id) => write!(f, "{id} is not a voter"), + Self::NotLearner(id) => write!(f, "{id} is not a learner"), Self::WouldBeEmpty => write!(f, "change would result in empty cluster"), Self::NoChange => write!(f, "configuration unchanged"), Self::NotSingleChange { added, removed } => { write!( f, - "single-server mode requires exactly one change, got {} added and {} removed", - added, removed + "single-server mode requires exactly one change, got {added} added and {removed} removed" ) } Self::NotLeader => write!(f, "not the leader"), @@ -619,16 +590,14 @@ mod tests { #[test] fn single_server_add_voter() { - let current = Configuration::simple_with_learners(vec![NodeId(0), NodeId(1)], vec![NodeId(2)]); + let current = + Configuration::simple_with_learners(vec![NodeId(0), NodeId(1)], vec![NodeId(2)]); let change = ConfigChange::AddVoter(NodeId(2)); assert!(change.validate(¤t, MembershipMode::SingleServer).is_ok()); let new = change.apply(¤t, MembershipMode::SingleServer); - assert_eq!( - new, - Configuration::simple(vec![NodeId(0), NodeId(1), NodeId(2)]) - ); + assert_eq!(new, Configuration::simple(vec![NodeId(0), NodeId(1), NodeId(2)])); } #[test] @@ -645,10 +614,13 @@ mod tests { #[test] fn single_server_rejects_multiple_changes() { let current = Configuration::simple(vec![NodeId(0), NodeId(1)]); - let change = ConfigChange::SetMembers(Members::new( - vec![NodeId(2), NodeId(3)], // +2, -2 - vec![], - ).unwrap()); + let change = ConfigChange::SetMembers( + Members::new( + vec![NodeId(2), NodeId(3)], // +2, -2 + vec![], + ) + .unwrap(), + ); let err = change.validate(¤t, MembershipMode::SingleServer).unwrap_err(); assert!(matches!(err, ConfigChangeError::NotSingleChange { .. })); @@ -657,10 +629,8 @@ mod tests { #[test] fn joint_consensus_creates_joint_config() { let current = Configuration::simple(vec![NodeId(0), NodeId(1)]); - let change = ConfigChange::SetMembers(Members::new( - vec![NodeId(2), NodeId(3)], - vec![], - ).unwrap()); + let change = + ConfigChange::SetMembers(Members::new(vec![NodeId(2), NodeId(3)], vec![]).unwrap()); assert!(change.validate(¤t, MembershipMode::JointConsensus).is_ok()); @@ -668,19 +638,13 @@ mod tests { assert!(new.is_joint()); assert_eq!( new, - Configuration::joint( - vec![NodeId(0), NodeId(1)], - vec![NodeId(2), NodeId(3)] - ) + Configuration::joint(vec![NodeId(0), NodeId(1)], vec![NodeId(2), NodeId(3)]) ); } #[test] fn rejects_change_during_joint() { - let current = Configuration::joint( - vec![NodeId(0), NodeId(1)], - vec![NodeId(1), NodeId(2)], - ); + let current = Configuration::joint(vec![NodeId(0), NodeId(1)], vec![NodeId(1), NodeId(2)]); let change = ConfigChange::AddVoter(NodeId(3)); let err = change.validate(¤t, MembershipMode::JointConsensus).unwrap_err(); @@ -719,10 +683,7 @@ mod tests { let simple = Configuration::simple_with_learners(vec![NodeId(0)], vec![NodeId(5)]); assert!(simple.transition_target().is_none()); - let joint = Configuration::joint( - vec![NodeId(0), NodeId(1)], - vec![NodeId(1), NodeId(2)], - ); + let joint = Configuration::joint(vec![NodeId(0), NodeId(1)], vec![NodeId(1), NodeId(2)]); assert_eq!( joint.transition_target(), Some(Configuration::simple_with_learners(vec![NodeId(1), NodeId(2)], vec![])) @@ -731,10 +692,7 @@ mod tests { #[test] fn voter_membership() { - let joint = Configuration::joint( - vec![NodeId(0), NodeId(1)], - vec![NodeId(1), NodeId(2)], - ); + let joint = Configuration::joint(vec![NodeId(0), NodeId(1)], vec![NodeId(1), NodeId(2)]); let membership = joint.voter_membership(); // Node 0: in old, not in new @@ -771,7 +729,8 @@ mod tests { #[test] fn rejects_no_op_set_members() { - let current = Configuration::simple_with_learners(vec![NodeId(0), NodeId(1)], vec![NodeId(2)]); + let current = + Configuration::simple_with_learners(vec![NodeId(0), NodeId(1)], vec![NodeId(2)]); let change = ConfigChange::SetMembers( Members::new(vec![NodeId(0), NodeId(1)], vec![NodeId(2)]).unwrap(), ); diff --git a/consensus/cloud9-raft/src/raft/mod.rs b/consensus/cloud9-raft/src/raft/mod.rs index de655f4..42617b3 100644 --- a/consensus/cloud9-raft/src/raft/mod.rs +++ b/consensus/cloud9-raft/src/raft/mod.rs @@ -21,7 +21,7 @@ //! - `Candidate` - runs elections, wins to Leader //! - `Leader` - replicates log, sends heartbeats //! -//! All roles share `Core` (term, voted_for, log, commit_index, config). +//! All roles share `Core` (term, `voted_for`, log, `commit_index`, config). //! //! # Timing (§3.9, Chapter 9) //! @@ -92,7 +92,7 @@ pub use precandidate::PreCandidate; use crate::{Command, CommittedEntry, LogIndex, NodeId, ProposeError, TransferError}; /// Role transition result. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Transition { Stay, ToFollower(Option), @@ -177,20 +177,14 @@ impl RaftNode { pub fn new(config: Config, voters: &[NodeId]) -> Self { let mut core = Core::new(config, voters); let follower = Follower::new(&mut core, None); - Self { - core, - role: RoleState::Follower(follower), - } + Self { core, role: RoleState::Follower(follower) } } /// Restore a node from persisted state. pub fn restore(config: Config, persistent: Persistent) -> Self { let mut core = Core::restore(config, persistent); let follower = Follower::new(&mut core, None); - Self { - core, - role: RoleState::Follower(follower), - } + Self { core, role: RoleState::Follower(follower) } } /// This node's ID. @@ -223,7 +217,7 @@ impl RaftNode { self.role.is_candidate() } - /// Whether this node is a pre-candidate (PreVote phase). + /// Whether this node is a pre-candidate (`PreVote` phase). #[inline] pub fn is_precandidate(&self) -> bool { self.role.is_precandidate() @@ -298,9 +292,7 @@ impl RaftNode { Ok((index, effects)) } - _ => Err(ProposeError::NotLeader { - leader_hint: self.leader(), - }), + _ => Err(ProposeError::NotLeader { leader_hint: self.leader() }), } } @@ -309,7 +301,7 @@ impl RaftNode { /// Per §4, the new configuration takes effect immediately when appended. pub fn propose_config_change( &mut self, - change: ConfigChange, + change: &ConfigChange, ) -> Result<(LogIndex, Effects), ConfigChangeError> { match &mut self.role { RoleState::Leader(leader) => { @@ -330,16 +322,13 @@ impl RaftNode { /// Transfer leadership to the target node (§3.11). /// - /// Sends TimeoutNow to the target, causing it to start an election immediately. + /// Sends `TimeoutNow` to the target, causing it to start an election immediately. /// The target must be a voter and fully caught up with the leader's log. /// /// After calling this, the leader should stop accepting new proposals and /// wait for the target to win the election. Once the leader sees a higher /// term (from the target's election), it will step down automatically. - pub fn transfer_leadership( - &self, - target: NodeId, - ) -> Result { + pub fn transfer_leadership(&self, target: NodeId) -> Result { match &self.role { RoleState::Leader(leader) => leader.transfer_leadership(&self.core, target), _ => Err(TransferError::NotLeader), @@ -355,15 +344,13 @@ impl RaftNode { let start = self.core.last_applied + 1; let end = self.core.commit_index + 1; (start..end).filter_map(|idx| { - self.core.log().get(idx).and_then(|entry| { - match &entry.payload { - EntryPayload::Command(cmd) => Some(CommittedEntry { - index: entry.index, - term: entry.term, - command: cmd.clone(), - }), - EntryPayload::Config(_) => None, - } + self.core.log().get(idx).and_then(|entry| match &entry.payload { + EntryPayload::Command(cmd) => Some(CommittedEntry { + index: entry.index, + term: entry.term, + command: cmd.clone(), + }), + EntryPayload::Config(_) => None, }) }) } @@ -445,7 +432,7 @@ mod tests { const THREE_VOTERS: &[NodeId] = &[NodeId(0), NodeId(1), NodeId(2)]; /// Create a config with prevote disabled for existing tests. - /// New PreVote-specific tests should use Config::new() which has prevote enabled. + /// New PreVote-specific tests should use `Config::new()` which has prevote enabled. fn test_config(id: NodeId) -> Config { Config::new(id) .with_prevote(false) @@ -617,7 +604,7 @@ mod tests { } } - /// §3.10: TimeoutNow triggers immediate election + /// §3.10: `TimeoutNow` triggers immediate election #[test] fn timeout_now() { let mut node = three_node_cluster(); @@ -715,7 +702,7 @@ mod tests { // Add a new voter // Step 1: add learner - let res = node.propose_config_change(ConfigChange::AddLearner(NodeId(1))).unwrap(); + let res = node.propose_config_change(&ConfigChange::AddLearner(NodeId(1))).unwrap(); // Simulate learner catching up by acknowledging the config entry let ack = Message { from: NodeId(1), @@ -729,7 +716,7 @@ mod tests { node.step(ack); // Step 2: promote to voter (requires learner to be caught up) - let result = node.propose_config_change(ConfigChange::AddVoter(NodeId(1))); + let result = node.propose_config_change(&ConfigChange::AddVoter(NodeId(1))); assert!(result.is_ok()); let (index, effects) = result.unwrap(); @@ -760,7 +747,7 @@ mod tests { assert!(node.is_leader()); // Remove a voter - let result = node.propose_config_change(ConfigChange::RemoveVoter(NodeId(2))); + let result = node.propose_config_change(&ConfigChange::RemoveVoter(NodeId(2))); assert!(result.is_ok()); // Removed voter should not be in the cluster @@ -781,12 +768,12 @@ mod tests { // Start a config change (creates joint config in JointConsensus mode) let members = Members::new([NodeId(0), NodeId(1)], []).unwrap(); - let result = node.propose_config_change(ConfigChange::SetMembers(members)); + let result = node.propose_config_change(&ConfigChange::SetMembers(members)); assert!(result.is_ok()); // Second change should fail while first is pending let members = Members::new([NodeId(0), NodeId(2)], []).unwrap(); - let result = node.propose_config_change(ConfigChange::SetMembers(members)); + let result = node.propose_config_change(&ConfigChange::SetMembers(members)); assert!(matches!(result, Err(ConfigChangeError::ChangeInProgress))); } @@ -810,7 +797,7 @@ mod tests { assert!(node.is_leader()); // Remove self from config - new config is [1, 2] - let _ = node.propose_config_change(ConfigChange::RemoveVoter(NodeId(0))); + let _ = node.propose_config_change(&ConfigChange::RemoveVoter(NodeId(0))); let config_index = node.core.log().last_index(); // After removing self, quorum is calculated with new config [1, 2] @@ -867,7 +854,7 @@ mod tests { // Change to a completely different set of voters (joint consensus) let members = Members::new([NodeId(0), NodeId(3), NodeId(4)], []).unwrap(); - let result = node.propose_config_change(ConfigChange::SetMembers(members)); + let result = node.propose_config_change(&ConfigChange::SetMembers(members)); assert!(result.is_ok()); // The config should now be Joint @@ -895,7 +882,7 @@ mod tests { assert!(node.commit_index() < config_index); } - /// §4.3: Joint consensus auto-completes to C_new + /// §4.3: Joint consensus auto-completes to `C_new` #[test] fn joint_consensus_auto_completes() { let mut config = test_config(NodeId(0)); @@ -918,7 +905,7 @@ mod tests { // Change config to remove NodeId(2): [0,1,2] → [0,1] // Creates Joint{[0,1,2], [0,1]} let members = Members::new([NodeId(0), NodeId(1)], []).unwrap(); - let result = node.propose_config_change(ConfigChange::SetMembers(members)); + let result = node.propose_config_change(&ConfigChange::SetMembers(members)); assert!(result.is_ok()); assert!(node.core.effective_config().is_joint()); @@ -1009,8 +996,7 @@ mod tests { assert_eq!(candidate.votes.len(), 2); // Reset to pre-win state for duplicate test - let config = test_config(NodeId(0)) - .with_election_timeout(10, 20); + let config = test_config(NodeId(0)).with_election_timeout(10, 20); let mut core = Core::new(config, &[NodeId(0), NodeId(1), NodeId(2), NodeId(3), NodeId(4)]); let (mut candidate, _) = Candidate::new(&mut core); @@ -1034,7 +1020,10 @@ mod tests { let result = candidate.step(&mut core, Event::Message(vote_dup)); // No state change, no effects - assert!(matches!(result.transition, Transition::Stay), "duplicate vote should not cause transition"); + assert!( + matches!(result.transition, Transition::Stay), + "duplicate vote should not cause transition" + ); assert_eq!(candidate.votes.len(), 2, "vote count unchanged"); assert!(result.effects.messages.is_empty(), "no messages from duplicate vote"); assert!(!result.effects.persist, "no persist from duplicate vote"); @@ -1079,10 +1068,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: core.term(), - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 2, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 2 }), }; leader.step(&mut core, Event::Message(ack)); @@ -1095,14 +1081,16 @@ mod tests { // Heartbeat to follower should have empty entries (they're caught up) for msg in &result.effects.messages { if let Payload::AppendRequest(req) = &msg.payload { - assert!(req.entries.is_empty(), + assert!( + req.entries.is_empty(), "up-to-date follower should receive empty heartbeat, got {} entries", - req.entries.len()); + req.entries.len() + ); } } } - /// Rejection hint is used to efficiently adjust next_index. + /// Rejection hint is used to efficiently adjust `next_index`. #[test] fn rejection_uses_hint_efficiently() { // Test at role level with pre-populated log @@ -1122,7 +1110,7 @@ mod tests { let (mut leader, _) = Leader::new(&mut core); // Leader initializes next_index to last_log_index + 1 = 11 - let next_before = *leader.next_index.get(&NodeId(1)).unwrap(); + let next_before = leader.next_index[&NodeId(1)]; assert_eq!(next_before, 11, "initial next_index should be last_log_index + 1"); // Follower rejects with hint that it only has index 3 @@ -1138,7 +1126,7 @@ mod tests { leader.step(&mut core, Event::Message(rejection)); // next_index should jump to hint+1 = 4, not decrement by 1 from 11 - let next_after = *leader.next_index.get(&NodeId(1)).unwrap(); + let next_after = leader.next_index[&NodeId(1)]; assert_eq!(next_after, 4, "should use hint efficiently: jump to hint+1"); } @@ -1243,11 +1231,7 @@ mod tests { let config = test_config(NodeId(0)); let mut node = RaftNode::new(config, &[NodeId(0), NodeId(1), NodeId(2)]); - let entry = Entry { - term: 1, - index: 1, - payload: EntryPayload::Command(Command(vec![42])), - }; + let entry = Entry { term: 1, index: 1, payload: EntryPayload::Command(Command(vec![42])) }; // First append let append1 = Message { @@ -1313,10 +1297,7 @@ mod tests { from: NodeId(2), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let effects = node.step(vote_req); @@ -1410,7 +1391,7 @@ mod tests { to: NodeId(0), term: 5, payload: Payload::VoteRequest(VoteRequest { - last_log_term: 3, // Older term + last_log_term: 3, // Older term last_log_index: 10, // Even with longer log }), }; @@ -1456,10 +1437,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_term: 1, - last_log_index: 1, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_term: 1, last_log_index: 1 }), }; let result = follower.step(&mut core, Event::Message(good_candidate)); @@ -1496,10 +1474,7 @@ mod tests { from: NodeId(2), to: NodeId(0), term: 2, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; let result = candidate.step(&mut core, Event::Message(vote_req)); @@ -1545,10 +1520,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: core.term(), - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 1, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 1 }), }; leader.step(&mut core, Event::Message(ack)); @@ -1561,10 +1533,7 @@ mod tests { // Heartbeat should have no entries (follower is caught up) for msg in &result.effects.messages { if let Payload::AppendRequest(req) = &msg.payload { - assert!( - req.entries.is_empty(), - "should not re-send already-acked entries" - ); + assert!(req.entries.is_empty(), "should not re-send already-acked entries"); } } } @@ -1592,15 +1561,12 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::AppendResponse(AppendResponse { - success: false, - last_log_index: 10, - }), + payload: Payload::AppendResponse(AppendResponse { success: false, last_log_index: 10 }), }; leader.step(&mut core, Event::Message(rejection)); // next_index should jump to 11, not 100 (one-by-one would be wasteful) - let next = *leader.next_index.get(&NodeId(1)).unwrap(); + let next = leader.next_index[&NodeId(1)]; assert_eq!(next, 11, "should use hint to jump, not decrement one-by-one"); } @@ -1630,10 +1596,7 @@ mod tests { from: NodeId(2), to: NodeId(0), term: 100, // Much higher term - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; node.step(disruptive); @@ -1657,10 +1620,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: core.term(), - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 0, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 0 }), }; let result = leader.step(&mut core, Event::Message(ack)); assert!(!result.effects.persist, "ack updates volatile state only"); @@ -1684,10 +1644,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 1, - payload: Payload::VoteRequest(VoteRequest { - last_log_index: 0, - last_log_term: 0, - }), + payload: Payload::VoteRequest(VoteRequest { last_log_index: 0, last_log_term: 0 }), }; follower.step(&mut core, Event::Message(vote_req)); @@ -1772,7 +1729,7 @@ mod tests { } // Should have had ~10 heartbeats (100 ticks / 10 interval) - assert!(heartbeat_count >= 9 && heartbeat_count <= 11); + assert!((9..=11).contains(&heartbeat_count)); // Each heartbeat should send exactly 3 messages (one per peer) let expected = heartbeat_count * 3; @@ -1826,7 +1783,7 @@ mod tests { // Start joint config: old=[0,1,2], new=[0,3,4] let members = Members::new([NodeId(0), NodeId(3), NodeId(4)], []).unwrap(); - node.propose_config_change(ConfigChange::SetMembers(members)).unwrap(); + node.propose_config_change(&ConfigChange::SetMembers(members)).unwrap(); let config_index = node.core.log().last_index(); // Ack from old config member only @@ -1860,15 +1817,12 @@ mod tests { node.step(new_ack); // Now should be committed (have 2/3 from old, 2/3 from new) - assert!( - node.commit_index() >= config_index, - "should commit once both majorities achieved" - ); + assert!(node.commit_index() >= config_index, "should commit once both majorities achieved"); } // --- Leadership Transfer Tests (§3.11) -------------------------------------- - /// §3.11: Leadership transfer sends TimeoutNow to caught-up target. + /// §3.11: Leadership transfer sends `TimeoutNow` to caught-up target. #[test] fn leadership_transfer_to_caught_up_peer() { let config = test_config(NodeId(0)); @@ -1892,10 +1846,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: node.term(), - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 0, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 0 }), }; node.step(ack); @@ -1998,7 +1949,7 @@ mod tests { assert!(matches!(result, Err(TransferError::NotLeader))); } - /// §3.11: Full transfer flow - target receives TimeoutNow and wins election. + /// §3.11: Full transfer flow - target receives `TimeoutNow` and wins election. #[test] fn leadership_transfer_full_flow() { let config0 = test_config(NodeId(0)); @@ -2027,10 +1978,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: leader_term, - payload: Payload::AppendResponse(AppendResponse { - success: true, - last_log_index: 0, - }), + payload: Payload::AppendResponse(AppendResponse { success: true, last_log_index: 0 }), }; node0.step(ack); @@ -2078,7 +2026,7 @@ mod tests { // --- PreVote tests (§4.2.3) ------------------------------------------------ - /// §4.2.3: With PreVote enabled, follower transitions to PreCandidate on timeout. + /// §4.2.3: With `PreVote` enabled, follower transitions to `PreCandidate` on timeout. #[test] fn prevote_follower_becomes_precandidate() { // Use prevote-enabled config @@ -2095,7 +2043,7 @@ mod tests { assert_eq!(node.term(), 0); // Term NOT incremented yet } - /// §4.2.3: PreCandidate sends PreVoteRequest, not VoteRequest. + /// §4.2.3: `PreCandidate` sends `PreVoteRequest`, not `VoteRequest`. #[test] fn prevote_sends_prevote_requests() { let config = Config::new(NodeId(0)); @@ -2117,7 +2065,7 @@ mod tests { } } - /// §4.2.3: PreCandidate proceeds to real Candidate after majority pre-votes. + /// §4.2.3: `PreCandidate` proceeds to real Candidate after majority pre-votes. #[test] fn prevote_proceeds_to_candidate_on_majority() { let config = Config::new(NodeId(0)); @@ -2134,10 +2082,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 0, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: 0, - granted: true, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: 0, granted: true }), }; node.step(prevote); @@ -2146,7 +2091,7 @@ mod tests { assert_eq!(node.term(), 1); } - /// §4.2.3: PreVote doesn't disrupt cluster - partitioned node can't increment term. + /// §4.2.3: `PreVote` doesn't disrupt cluster - partitioned node can't increment term. #[test] fn prevote_prevents_term_disruption() { let config = Config::new(NodeId(0)); @@ -2172,7 +2117,7 @@ mod tests { assert_eq!(node.term(), 0); } - /// §4.2.3: Full PreVote → Vote → Leader flow. + /// §4.2.3: Full `PreVote` → Vote → Leader flow. #[test] fn prevote_full_election_flow() { let config = Config::new(NodeId(0)); @@ -2189,10 +2134,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 0, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: 0, - granted: true, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: 0, granted: true }), }; node.step(prevote); assert!(node.role.is_candidate()); @@ -2209,7 +2151,7 @@ mod tests { assert!(node.is_leader()); } - /// §4.2.3: PreCandidate steps down on AppendEntries from leader. + /// §4.2.3: `PreCandidate` steps down on `AppendEntries` from leader. #[test] fn prevote_steps_down_on_leader_contact() { let config = Config::new(NodeId(0)); @@ -2275,7 +2217,7 @@ mod tests { } // Add node 3 as learner then voter (to have someone to transfer to) - let _ = node.propose_config_change(ConfigChange::AddLearner(NodeId(3))); + let _ = node.propose_config_change(&ConfigChange::AddLearner(NodeId(3))); // Get node 3 caught up node.step(Message { @@ -2288,7 +2230,7 @@ mod tests { }), }); - let _ = node.propose_config_change(ConfigChange::AddVoter(NodeId(3))); + let _ = node.propose_config_change(&ConfigChange::AddVoter(NodeId(3))); // Update everyone's match index again for peer in [NodeId(1), NodeId(2), NodeId(3)] { @@ -2304,7 +2246,7 @@ mod tests { } // Now remove ourselves (node 0) - let result = node.propose_config_change(ConfigChange::RemoveVoter(NodeId(0))); + let result = node.propose_config_change(&ConfigChange::RemoveVoter(NodeId(0))); // The config change should succeed let (_, _effects) = result.expect("config change should succeed"); diff --git a/consensus/cloud9-raft/src/raft/precandidate.rs b/consensus/cloud9-raft/src/raft/precandidate.rs index dea49c6..9123266 100644 --- a/consensus/cloud9-raft/src/raft/precandidate.rs +++ b/consensus/cloud9-raft/src/raft/precandidate.rs @@ -1,28 +1,27 @@ -//! PreCandidate role (§4.2.3). +//! `PreCandidate` role (§4.2.3). //! //! A pre-candidate: //! - Does NOT increment term (avoids disrupting the cluster) -//! - Sends PreVoteRequest to all peers +//! - Sends `PreVoteRequest` to all peers //! - Wins pre-election with majority → becomes Candidate -//! - Times out → restarts pre-election (stays PreCandidate) +//! - Times out → restarts pre-election (stays `PreCandidate`) //! - Discovers higher term → becomes Follower -//! - Receives AppendEntries from leader → becomes Follower +//! - Receives `AppendEntries` from leader → becomes Follower //! -//! PreVote prevents partitioned nodes from incrementing their term and +//! `PreVote` prevents partitioned nodes from incrementing their term and //! disrupting the cluster when they rejoin. A pre-candidate only proceeds //! to a real election if it can win the pre-vote (proving it can reach //! a majority of nodes). use crate::NodeId; +use super::StepResult; use super::core::Core; use super::event::{ - AppendRequest, Effects, Event, Message, Payload, PreVoteRequest, PreVoteResponse, VoteRequest, - VoteResponse, + AppendRequest, Effects, Event, Message, Payload, PreVoteRequest, PreVoteResponse, VoteResponse, }; -use super::StepResult; -/// PreCandidate role state. +/// `PreCandidate` role state. #[derive(Debug, Clone)] pub struct PreCandidate { /// Set of peers who granted their pre-vote (including self). @@ -36,8 +35,8 @@ pub struct PreCandidate { impl PreCandidate { /// Create a new pre-candidate, starting a pre-election. /// - /// Unlike Candidate::new(), this does NOT increment the term. - /// It sends PreVoteRequest to check if an election would succeed. + /// Unlike `Candidate::new()`, this does NOT increment the term. + /// It sends `PreVoteRequest` to check if an election would succeed. pub fn new(core: &mut Core) -> (Self, Effects) { let timeout = core.random_election_timeout(); let mut precandidate = Self { @@ -71,7 +70,7 @@ impl PreCandidate { from: core.id(), to: peer, term: core.term(), // Use current term, not incremented - payload: Payload::PreVoteRequest(req.clone()), + payload: Payload::PreVoteRequest(req), }) .collect(); @@ -97,10 +96,10 @@ impl PreCandidate { Event::DiskWriteComplete(_) => StepResult::none(), Event::Message(msg) => { // Handle pre-vote requests - use same disruptive vote guard - if let Payload::PreVoteRequest(_) | Payload::VoteRequest(_) = msg.payload { - if self.recently_heard_from_leader(core) { - return StepResult::stay(self.deny_vote(core, &msg)); - } + if let Payload::PreVoteRequest(_) | Payload::VoteRequest(_) = msg.payload + && self.recently_heard_from_leader(core) + { + return StepResult::stay(Self::deny_vote(core, &msg)); } // Higher term: become follower @@ -117,9 +116,9 @@ impl PreCandidate { } Payload::AppendRequest(req) => self.handle_append_request(core, msg.from, req), Payload::PreVoteRequest(req) => { - self.handle_prevote_request(core, msg.from, msg.term, req) + Self::handle_prevote_request(core, msg.from, msg.term, req) } - Payload::VoteRequest(req) => self.handle_vote_request(core, msg.from, req), + Payload::VoteRequest(_) => Self::handle_vote_request(core, msg.from), Payload::TimeoutNow => { // Skip pre-vote, go directly to candidate StepResult::to_candidate(Effects::none()) @@ -184,7 +183,6 @@ impl PreCandidate { } fn handle_prevote_request( - &mut self, core: &Core, from: NodeId, msg_term: u64, @@ -201,20 +199,12 @@ impl PreCandidate { from: core.id(), to: from, term: msg_term, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted }), }; StepResult::stay(Effects::none().with_message(resp)) } - fn handle_vote_request( - &mut self, - core: &Core, - from: NodeId, - _req: VoteRequest, - ) -> StepResult { + fn handle_vote_request(core: &Core, from: NodeId) -> StepResult { // We're pre-candidate, deny real votes let resp = Message { from: core.id(), @@ -225,12 +215,11 @@ impl PreCandidate { StepResult::stay(Effects::none().with_message(resp)) } - fn deny_vote(&self, core: &Core, msg: &Message) -> Effects { + fn deny_vote(core: &Core, msg: &Message) -> Effects { let payload = match &msg.payload { - Payload::PreVoteRequest(_) => Payload::PreVoteResponse(PreVoteResponse { - term: core.term(), - granted: false, - }), + Payload::PreVoteRequest(_) => { + Payload::PreVoteResponse(PreVoteResponse { term: core.term(), granted: false }) + } Payload::VoteRequest(_) => Payload::VoteResponse(VoteResponse { granted: false }), _ => return Effects::none(), }; @@ -252,8 +241,8 @@ impl PreCandidate { mod tests { use super::*; - use super::super::core::Config; use super::super::Transition; + use super::super::core::Config; fn test_setup() -> (Core, PreCandidate, Effects) { let config = Config::new(NodeId(0)).with_election_timeout(10, 20); @@ -311,10 +300,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 0, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: 0, - granted: true, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: 0, granted: true }), }; let StepResult { transition, .. } = precandidate.step(&mut core, Event::Message(msg)); @@ -344,10 +330,7 @@ mod tests { from: NodeId(1), to: NodeId(0), term: 5, // Higher than our term 0 - payload: Payload::PreVoteResponse(PreVoteResponse { - term: 5, - granted: false, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: 5, granted: false }), }; let StepResult { transition, effects } = precandidate.step(&mut core, Event::Message(msg)); @@ -380,12 +363,7 @@ mod tests { fn timeout_now_skips_to_candidate() { let (mut core, mut precandidate, _) = test_setup(); - let msg = Message { - from: NodeId(1), - to: NodeId(0), - term: 0, - payload: Payload::TimeoutNow, - }; + let msg = Message { from: NodeId(1), to: NodeId(0), term: 0, payload: Payload::TimeoutNow }; let StepResult { transition, .. } = precandidate.step(&mut core, Event::Message(msg)); assert!(matches!(transition, Transition::ToCandidate)); diff --git a/consensus/cloud9-raft/tests/efficiency.rs b/consensus/cloud9-raft/tests/efficiency.rs index e27d7bc..d65354b 100644 --- a/consensus/cloud9-raft/tests/efficiency.rs +++ b/consensus/cloud9-raft/tests/efficiency.rs @@ -1,4 +1,5 @@ //! Efficiency and non-redundancy integration tests. +#![allow(clippy::cast_possible_truncation, clippy::expect_used, clippy::panic, clippy::unwrap_used)] //! //! These tests verify the implementation does no unnecessary work: //! - No duplicate messages or wasted RPCs @@ -11,15 +12,13 @@ use std::collections::HashMap; use cloud9_raft::raft::{ - AppendRequest, AppendResponse, Config, Entry, EntryPayload, Message, Payload, - PreVoteResponse, RaftNode, VoteResponse, + AppendRequest, AppendResponse, Config, Entry, EntryPayload, Message, Payload, PreVoteResponse, + RaftNode, VoteResponse, }; use cloud9_raft::{Command, NodeId}; fn test_config(id: NodeId) -> Config { - Config::new(id) - .with_election_timeout(10, 20) - .with_heartbeat_interval(3) + Config::new(id).with_election_timeout(10, 20).with_heartbeat_interval(3) } fn classify_message(payload: &Payload) -> &'static str { @@ -53,11 +52,7 @@ impl TestCluster { let node = RaftNode::new(config, node_ids); nodes.insert(id, node); } - Self { - nodes, - pending_messages: Vec::new(), - message_counts: HashMap::new(), - } + Self { nodes, pending_messages: Vec::new(), message_counts: HashMap::new() } } fn tick_all(&mut self) { @@ -99,26 +94,19 @@ impl TestCluster { } fn leader(&self) -> Option { - self.nodes - .iter() - .find(|(_, n)| n.is_leader()) - .map(|(&id, _)| id) + self.nodes.iter().find(|(_, n)| n.is_leader()).map(|(&id, _)| id) } fn leaders(&self) -> Vec { - self.nodes - .iter() - .filter(|(_, n)| n.is_leader()) - .map(|(&id, _)| id) - .collect() + self.nodes.iter().filter(|(_, n)| n.is_leader()).map(|(&id, _)| id).collect() } fn term(&self, id: NodeId) -> u64 { - self.nodes.get(&id).map(|n| n.term()).unwrap_or(0) + self.nodes.get(&id).map_or(0, cloud9_raft::RaftNode::term) } fn max_term(&self) -> u64 { - self.nodes.values().map(|n| n.term()).max().unwrap_or(0) + self.nodes.values().map(cloud9_raft::RaftNode::term).max().unwrap_or(0) } fn message_count(&self, msg_type: &str) -> usize { @@ -134,7 +122,7 @@ impl TestCluster { // PreVote Efficiency Tests (§9.4) // ============================================================================= -/// PreVote prevents a partitioned server from disrupting the cluster on rejoin. +/// `PreVote` prevents a partitioned server from disrupting the cluster on rejoin. /// /// Per dissertation §9.4: "While a server is partitioned, it won't be able to /// increment its term, since it can't receive permission from a majority." @@ -164,27 +152,18 @@ fn prevote_prevents_term_inflation_during_partition() { // With PreVote, node 4 should send PreVoteRequests, not VoteRequests // and should NOT increment its term let node4_term = cluster.term(NodeId(4)); - assert_eq!( - node4_term, initial_term, - "partitioned node should not inflate term with PreVote" - ); + assert_eq!(node4_term, initial_term, "partitioned node should not inflate term with PreVote"); // Should have sent PreVoteRequests, not VoteRequests let prevote_count = cluster.message_count("PreVoteRequest"); let vote_count = cluster.message_count("VoteRequest"); - assert!( - prevote_count > 0, - "should send PreVoteRequests while partitioned" - ); - assert_eq!( - vote_count, 0, - "should NOT send VoteRequests without PreVote success" - ); + assert!(prevote_count > 0, "should send PreVoteRequests while partitioned"); + assert_eq!(vote_count, 0, "should NOT send VoteRequests without PreVote success"); } /// When a partitioned server rejoins, it doesn't disrupt the existing leader. /// -/// This is a key PreVote invariant: a partitioned node shouldn't be able to +/// This is a key `PreVote` invariant: a partitioned node shouldn't be able to /// disrupt a healthy cluster when it reconnects. #[test] fn partitioned_server_rejoin_no_disruption() { @@ -229,8 +208,7 @@ fn partitioned_server_rejoin_no_disruption() { // so there's no reason for the cluster term to change assert_eq!( term_after, term_before, - "term should be unchanged after rejoin: {} vs {}", - term_before, term_after + "term should be unchanged after rejoin: {term_before} vs {term_after}" ); // Leader should be the same (no disruption) @@ -274,18 +252,14 @@ fn simultaneous_elections_converge() { } } - assert!( - rounds_to_leader > 0, - "should elect a leader after simultaneous timeout" - ); + assert!(rounds_to_leader > 0, "should elect a leader after simultaneous timeout"); // Convergence should happen in bounded time. // With randomized timeouts (10-20 ticks), worst case is several election // cycles. 50 rounds is very conservative - typically converges in < 10. assert!( rounds_to_leader < 50, - "cluster should converge in bounded time, took {} rounds", - rounds_to_leader + "cluster should converge in bounded time, took {rounds_to_leader} rounds" ); // Should have exactly one leader @@ -295,11 +269,7 @@ fn simultaneous_elections_converge() { // Each failed election increments term by 1 (PreVote -> Candidate). // With randomization, shouldn't need many election rounds. let max_term = cluster.max_term(); - assert!( - max_term < 10, - "term should be bounded after convergence, got {}", - max_term - ); + assert!(max_term < 10, "term should be bounded after convergence, got {max_term}"); } /// Split vote situation eventually resolves due to randomized timeouts. @@ -331,18 +301,14 @@ fn split_vote_resolves_with_randomization() { } } - assert!( - rounds_to_leader > 0, - "should eventually elect a leader after split votes" - ); + assert!(rounds_to_leader > 0, "should eventually elect a leader after split votes"); // Key invariant: randomization should resolve split votes reasonably quickly // Without randomization, 5 nodes could split-vote indefinitely // With randomization, expect resolution within ~10 rounds (very conservative) assert!( rounds_to_leader < 30, - "split vote should resolve quickly with randomization, took {} rounds", - rounds_to_leader + "split vote should resolve quickly with randomization, took {rounds_to_leader} rounds" ); // Should have exactly one leader @@ -350,20 +316,16 @@ fn split_vote_resolves_with_randomization() { // Term should be bounded - not inflated by repeated split votes let max_term = cluster.max_term(); - assert!( - max_term < 10, - "term should be bounded after split vote resolution, got {}", - max_term - ); + assert!(max_term < 10, "term should be bounded after split vote resolution, got {max_term}"); } /// Candidate that receives heartbeat from valid leader steps down efficiently. /// -/// Per Raft: A candidate that receives AppendEntries from a leader with term >= its own +/// Per Raft: A candidate that receives `AppendEntries` from a leader with term >= its own /// should recognize the leader and revert to follower. /// -/// Note: With PreVote enabled, the node transitions to PreCandidate first. When PreCandidate -/// receives AppendEntries, it steps down to Follower but doesn't automatically respond to +/// Note: With `PreVote` enabled, the node transitions to `PreCandidate` first. When `PreCandidate` +/// receives `AppendEntries`, it steps down to Follower but doesn't automatically respond to /// the message that caused the step-down (the message is consumed during transition). /// To fully test the response, we send a second heartbeat after step-down. #[test] @@ -395,10 +357,7 @@ fn candidate_steps_down_on_leader_heartbeat() { let effects = node.step(heartbeat.clone()); // Node should step down to follower - assert!( - !node.is_leader(), - "node should not be leader after receiving heartbeat" - ); + assert!(!node.is_leader(), "node should not be leader after receiving heartbeat"); // PreCandidate step-down doesn't respond to the triggering message // (different from Follower which always responds to AppendRequest) @@ -417,17 +376,14 @@ fn candidate_steps_down_on_leader_heartbeat() { assert!(!node.is_leader()); // Total message count should be reasonable (just 1 response after step-down) - assert!( - effects.messages.len() <= 1, - "step-down should produce minimal messages" - ); + assert!(effects.messages.len() <= 1, "step-down should produce minimal messages"); } // ============================================================================= // Replication Efficiency Tests // ============================================================================= -/// AppendEntries with already-committed entries is idempotent. +/// `AppendEntries` with already-committed entries is idempotent. #[test] fn redundant_append_entries_idempotent() { let config = test_config(NodeId(0)); @@ -508,10 +464,7 @@ fn no_messages_to_self() { } // Sanity check: we actually saw some messages - assert!( - !all_messages_seen.is_empty(), - "test should have generated some messages" - ); + assert!(!all_messages_seen.is_empty(), "test should have generated some messages"); } /// Heartbeat responses don't trigger unnecessary work when follower is caught up. @@ -532,20 +485,13 @@ fn heartbeat_ack_minimal_work() { // Verify precondition: leader has empty log (commit_index = 0) // This ensures follower claiming index 0 is actually "caught up" - let leader_commit = cluster.nodes.get(&leader).unwrap().commit_index(); - assert_eq!( - leader_commit, 0, - "precondition: leader should have empty log for this test" - ); + let leader_commit = cluster.nodes[&leader].commit_index(); + assert_eq!(leader_commit, 0, "precondition: leader should have empty log for this test"); cluster.reset_message_counts(); // Send heartbeat ack from follower claiming to be at index 0 (caught up) - let follower = if leader == NodeId(0) { - NodeId(1) - } else { - NodeId(0) - }; + let follower = if leader == NodeId(0) { NodeId(1) } else { NodeId(0) }; let ack = Message { from: follower, to: leader, @@ -564,10 +510,7 @@ fn heartbeat_ack_minimal_work() { effects.messages.is_empty(), "ack from caught-up follower should not generate messages" ); - assert!( - !effects.persist, - "ack from caught-up follower should not require persistence" - ); + assert!(!effects.persist, "ack from caught-up follower should not require persistence"); } } @@ -614,10 +557,7 @@ fn persistence_only_on_state_change() { }), }; let effects_same = node.step(same_term); - assert!( - !effects_same.persist, - "heartbeat at same term should NOT require persistence" - ); + assert!(!effects_same.persist, "heartbeat at same term should NOT require persistence"); // Send heartbeat at HIGHER term - persistence required let higher_term = Message { @@ -713,9 +653,7 @@ fn election_bounded_messages() { // With a few rounds of split votes, should be < 500 messages total assert!( total_messages < 500, - "election took {} messages in {} rounds, should be bounded", - total_messages, - round + "election took {total_messages} messages in {round} rounds, should be bounded" ); return; } @@ -772,10 +710,7 @@ fn duplicate_votes_ignored() { node.step(vote1); // Now have 2 votes (self + NodeId(1)). Still not quorum. - assert!( - !node.is_leader(), - "should not be leader with only 2 votes in 5-node cluster" - ); + assert!(!node.is_leader(), "should not be leader with only 2 votes in 5-node cluster"); // Send DUPLICATE vote from NodeId(1) - should NOT count as third vote let vote_dup = Message { @@ -810,9 +745,9 @@ fn duplicate_votes_ignored() { ); } -/// PreVote responses from same peer are ignored after first. +/// `PreVote` responses from same peer are ignored after first. /// -/// In PreVote, we need quorum before transitioning to Candidate. +/// In `PreVote`, we need quorum before transitioning to Candidate. /// Duplicate responses from the same peer should not be double-counted. #[test] fn duplicate_prevote_responses_ignored() { diff --git a/consensus/cloud9-raft/tests/optimizations.rs b/consensus/cloud9-raft/tests/optimizations.rs index 3a18bdc..e6d6b04 100644 --- a/consensus/cloud9-raft/tests/optimizations.rs +++ b/consensus/cloud9-raft/tests/optimizations.rs @@ -1,4 +1,5 @@ //! Performance optimization tests (§10.2). +#![allow(clippy::expect_used, clippy::print_stderr, clippy::unwrap_used)] //! //! Tests correctness and demonstrates performance benefits of: //! - Parallel disk write (§10.2.1): Leader writes disk while replicating @@ -8,9 +9,7 @@ use std::collections::HashMap; -use cloud9_raft::raft::{ - AppendResponse, Config, Message, Payload, RaftNode, -}; +use cloud9_raft::raft::{AppendResponse, Config, Message, Payload, RaftNode}; use cloud9_raft::{Command, LogIndex, NodeId}; fn optimized_config(id: NodeId) -> Config { @@ -47,11 +46,7 @@ impl TestCluster { let node = RaftNode::new(config, node_ids); nodes.insert(id, node); } - Self { - nodes, - pending_messages: Vec::new(), - round_trips: 0, - } + Self { nodes, pending_messages: Vec::new(), round_trips: 0 } } fn tick_all(&mut self) { @@ -79,14 +74,11 @@ impl TestCluster { } fn leader(&self) -> Option { - self.nodes - .iter() - .find(|(_, n)| n.is_leader()) - .map(|(&id, _)| id) + self.nodes.iter().find(|(_, n)| n.is_leader()).map(|(&id, _)| id) } fn commit_index(&self, id: NodeId) -> LogIndex { - self.nodes.get(&id).map(|n| n.commit_index()).unwrap_or(0) + self.nodes.get(&id).map_or(0, cloud9_raft::RaftNode::commit_index) } fn propose(&mut self, leader_id: NodeId, data: Vec) -> Option<(LogIndex, Vec)> { @@ -120,7 +112,7 @@ impl TestCluster { /// Parallel disk write: leader can make progress with followers while disk write pending. /// /// Per dissertation §10.2.1: "With the optimization, the leader would issue -/// AppendEntries RPCs to its disk and followers in parallel, with the disk +/// `AppendEntries` RPCs to its disk and followers in parallel, with the disk /// treated similarly to a follower." #[test] fn parallel_disk_write_correctness() { @@ -205,7 +197,7 @@ fn naive_disk_write_blocks_until_self_written() { ); } -/// Single-node cluster with parallel disk write requires explicit DiskWriteComplete. +/// Single-node cluster with parallel disk write requires explicit `DiskWriteComplete`. #[test] fn single_node_parallel_disk_write() { let config = optimized_config(NodeId(0)); @@ -224,11 +216,7 @@ fn single_node_parallel_disk_write() { // Signal disk write complete node.disk_write_complete(1); - assert_eq!( - node.commit_index(), - 1, - "should commit after disk write completes" - ); + assert_eq!(node.commit_index(), 1, "should commit after disk write completes"); } /// Single-node cluster without parallel disk write commits immediately. @@ -255,7 +243,7 @@ fn single_node_naive_commits_immediately() { /// Pipelining: leader can send multiple batches without waiting for ACKs. /// -/// Per dissertation §10.2.2: "the leader sends additional AppendEntries RPCs +/// Per dissertation §10.2.2: "the leader sends additional `AppendEntries` RPCs /// without waiting for acknowledgments from the follower." #[test] fn pipelining_sends_without_waiting() { @@ -281,10 +269,7 @@ fn pipelining_sends_without_waiting() { // Key: messages should include ALL entries to each follower // because next_index is updated optimistically - assert!( - messages_before_ack >= 2, - "pipelining should send without waiting for ACKs" - ); + assert!(messages_before_ack >= 2, "pipelining should send without waiting for ACKs"); // Check that messages include later entries let has_entry_5 = cluster.pending_messages.iter().any(|m| { @@ -353,7 +338,7 @@ fn pipelining_handles_rejection() { let reject = Message { from: NodeId(1), to: leader, - term: cluster.nodes.get(&leader).unwrap().term(), + term: cluster.nodes[&leader].term(), payload: Payload::AppendResponse(AppendResponse { success: false, last_log_index: 0, // Follower is behind @@ -363,10 +348,7 @@ fn pipelining_handles_rejection() { if let Some(node) = cluster.nodes.get_mut(&leader) { let effects = node.step(reject); // Leader should retry with decremented next_index - assert!( - !effects.messages.is_empty(), - "should retry after rejection" - ); + assert!(!effects.messages.is_empty(), "should retry after rejection"); } } @@ -405,11 +387,7 @@ fn optimized_commit_efficiency() { cluster.round_trips ); - assert_eq!( - cluster.commit_index(leader), - 10, - "all entries should be committed" - ); + assert_eq!(cluster.commit_index(leader), 10, "all entries should be committed"); } /// Measures round-trips needed to commit N entries with optimizations disabled. @@ -438,18 +416,10 @@ fn naive_commit_efficiency() { // Record for comparison (no strict assertion - just documenting behavior) let naive_trips = cluster.round_trips; - assert_eq!( - cluster.commit_index(leader), - 10, - "all entries should be committed" - ); + assert_eq!(cluster.commit_index(leader), 10, "all entries should be committed"); // Sanity check: should complete eventually - assert!( - naive_trips < 50, - "naive should complete in bounded time, took {}", - naive_trips - ); + assert!(naive_trips < 50, "naive should complete in bounded time, took {naive_trips}"); } /// Direct comparison: optimized vs naive round-trips. @@ -496,14 +466,9 @@ fn optimization_provides_benefit() { // should never be worse) assert!( opt_trips <= naive_trips + 1, - "optimized ({} trips) should not be significantly worse than naive ({} trips)", - opt_trips, - naive_trips + "optimized ({opt_trips} trips) should not be significantly worse than naive ({naive_trips} trips)" ); // Log for visibility - eprintln!( - "Performance comparison: optimized={} trips, naive={} trips", - opt_trips, naive_trips - ); + eprintln!("Performance comparison: optimized={opt_trips} trips, naive={naive_trips} trips"); } diff --git a/consensus/cloud9-raft/tests/prevote.rs b/consensus/cloud9-raft/tests/prevote.rs index 7d6753e..b16badf 100644 --- a/consensus/cloud9-raft/tests/prevote.rs +++ b/consensus/cloud9-raft/tests/prevote.rs @@ -1,24 +1,23 @@ -//! PreVote extension tests (Dissertation §9.4). +//! `PreVote` extension tests (Dissertation §9.4). +#![allow(clippy::panic)] //! //! These tests cover §9.4-specific behaviors NOT tested elsewhere: -//! - Leader contact timeout (recently_heard_from_leader guard) -//! - Log up-to-dateness check in PreVote responses +//! - Leader contact timeout (`recently_heard_from_leader` guard) +//! - Log up-to-dateness check in `PreVote` responses //! - Leaders deny all prevotes -//! - next_term > voter's term requirement +//! - `next_term` > voter's term requirement //! -//! Note: Basic PreVote mechanics (term not incremented, PreCandidate transitions) +//! Note: Basic `PreVote` mechanics (term not incremented, `PreCandidate` transitions) //! are tested in src/raft/precandidate.rs and src/raft/mod.rs. use cloud9_raft::raft::{ - AppendRequest, Config, Entry, EntryPayload, Message, Payload, PreVoteRequest, - PreVoteResponse, RaftNode, VoteResponse, + AppendRequest, Config, Entry, EntryPayload, Message, Payload, PreVoteRequest, PreVoteResponse, + RaftNode, VoteResponse, }; use cloud9_raft::{Command, NodeId}; fn prevote_config(id: NodeId) -> Config { - Config::new(id) - .with_election_timeout(10, 20) - .with_heartbeat_interval(3) + Config::new(id).with_election_timeout(10, 20).with_heartbeat_interval(3) } const THREE_VOTERS: &[NodeId] = &[NodeId(0), NodeId(1), NodeId(2)]; @@ -31,7 +30,7 @@ const THREE_VOTERS: &[NodeId] = &[NodeId(0), NodeId(1), NodeId(2)]; /// a baseline election timeout" /// /// A follower that recently received a heartbeat from the leader should deny -/// PreVote requests, preventing disruption from partitioned/slow servers. +/// `PreVote` requests, preventing disruption from partitioned/slow servers. #[test] fn follower_denies_prevote_if_recently_heard_from_leader() { let config = prevote_config(NodeId(0)); @@ -70,16 +69,13 @@ fn follower_denies_prevote_if_recently_heard_from_leader() { // Should respond with denial assert_eq!(effects.messages.len(), 1); if let Payload::PreVoteResponse(resp) = &effects.messages[0].payload { - assert!( - !resp.granted, - "should deny PreVote when recently heard from leader" - ); + assert!(!resp.granted, "should deny PreVote when recently heard from leader"); } else { panic!("expected PreVoteResponse"); } } -/// After enough time passes without leader contact, follower should grant PreVote. +/// After enough time passes without leader contact, follower should grant `PreVote`. #[test] fn follower_grants_prevote_after_leader_timeout() { let config = prevote_config(NodeId(0)); @@ -120,10 +116,7 @@ fn follower_grants_prevote_after_leader_timeout() { assert_eq!(effects.messages.len(), 1); if let Payload::PreVoteResponse(resp) = &effects.messages[0].payload { - assert!( - resp.granted, - "should grant PreVote after leader contact timeout" - ); + assert!(resp.granted, "should grant PreVote after leader contact timeout"); } else { panic!("expected PreVoteResponse"); } @@ -133,19 +126,16 @@ fn follower_grants_prevote_after_leader_timeout() { // §9.4: PreVote respects log up-to-dateness (election restriction) // ============================================================================= -/// PreVote should be denied if the candidate's log is not up-to-date. -/// This is the same check as regular VoteRequest. +/// `PreVote` should be denied if the candidate's log is not up-to-date. +/// This is the same check as regular `VoteRequest`. #[test] fn prevote_denied_if_log_not_up_to_date() { let config = prevote_config(NodeId(0)); let mut node = RaftNode::new(config, THREE_VOTERS); // Receive entries to make our log non-empty - let entries = vec![Entry { - term: 1, - index: 1, - payload: EntryPayload::Command(Command(vec![1])), - }]; + let entries = + vec![Entry { term: 1, index: 1, payload: EntryPayload::Command(Command(vec![1])) }]; let append = Message { from: NodeId(1), to: NodeId(0), @@ -179,27 +169,21 @@ fn prevote_denied_if_log_not_up_to_date() { assert_eq!(effects.messages.len(), 1); if let Payload::PreVoteResponse(resp) = &effects.messages[0].payload { - assert!( - !resp.granted, - "should deny PreVote when candidate's log is not up-to-date" - ); + assert!(!resp.granted, "should deny PreVote when candidate's log is not up-to-date"); } else { panic!("expected PreVoteResponse"); } } -/// PreVote granted when candidate's log is at least as up-to-date. +/// `PreVote` granted when candidate's log is at least as up-to-date. #[test] fn prevote_granted_if_log_up_to_date() { let config = prevote_config(NodeId(0)); let mut node = RaftNode::new(config, THREE_VOTERS); // Receive one entry - let entries = vec![Entry { - term: 1, - index: 1, - payload: EntryPayload::Command(Command(vec![1])), - }]; + let entries = + vec![Entry { term: 1, index: 1, payload: EntryPayload::Command(Command(vec![1])) }]; let append = Message { from: NodeId(1), to: NodeId(0), @@ -243,7 +227,7 @@ fn prevote_granted_if_log_up_to_date() { // §9.4: Leaders deny all PreVotes // ============================================================================= -/// A leader should deny all PreVote requests - it's a valid leader. +/// A leader should deny all `PreVote` requests - it's a valid leader. #[test] fn leader_denies_all_prevotes() { let config = prevote_config(NodeId(0)); @@ -258,10 +242,7 @@ fn leader_denies_all_prevotes() { from: NodeId(1), to: NodeId(0), term: 0, - payload: Payload::PreVoteResponse(PreVoteResponse { - term: 0, - granted: true, - }), + payload: Payload::PreVoteResponse(PreVoteResponse { term: 0, granted: true }), }); assert!(node.is_candidate()); @@ -305,7 +286,7 @@ fn leader_denies_all_prevotes() { // §9.4: PreVote term requirements // ============================================================================= -/// PreVote next_term must be greater than voter's term. +/// `PreVote` `next_term` must be greater than voter's term. #[test] fn prevote_requires_higher_next_term() { let config = prevote_config(NodeId(0)); @@ -346,10 +327,7 @@ fn prevote_requires_higher_next_term() { assert_eq!(effects.messages.len(), 1); if let Payload::PreVoteResponse(resp) = &effects.messages[0].payload { - assert!( - !resp.granted, - "should deny PreVote when next_term is not greater than our term" - ); + assert!(!resp.granted, "should deny PreVote when next_term is not greater than our term"); } else { panic!("expected PreVoteResponse"); } diff --git a/consensus/cloud9-raft/tests/property.rs b/consensus/cloud9-raft/tests/property.rs index 89c4318..190fe09 100644 --- a/consensus/cloud9-raft/tests/property.rs +++ b/consensus/cloud9-raft/tests/property.rs @@ -1,4 +1,5 @@ //! Property-based testing for Raft consensus. +#![allow(clippy::cast_possible_truncation)] //! //! Uses proptest to verify invariants hold under random event sequences. //! @@ -19,9 +20,7 @@ use cloud9_raft::{LogIndex, NodeId, Term}; const TEST_NODES: &[NodeId] = &[NodeId(0), NodeId(1), NodeId(2)]; fn test_config(id: NodeId) -> Config { - Config::new(id) - .with_prevote(false) - .with_election_timeout(5, 10) + Config::new(id).with_prevote(false).with_election_timeout(5, 10) } // --- Arbitrary Generators --- @@ -39,10 +38,8 @@ fn arb_log_index() -> impl Strategy { } fn arb_vote_request() -> impl Strategy { - (arb_log_index(), arb_term()).prop_map(|(last_log_index, last_log_term)| VoteRequest { - last_log_index, - last_log_term, - }) + (arb_log_index(), arb_term()) + .prop_map(|(last_log_index, last_log_term)| VoteRequest { last_log_index, last_log_term }) } fn arb_vote_response() -> impl Strategy { @@ -75,10 +72,8 @@ fn arb_append_request() -> impl Strategy { } fn arb_append_response() -> impl Strategy { - (any::(), arb_log_index()).prop_map(|(success, last_log_index)| AppendResponse { - success, - last_log_index, - }) + (any::(), arb_log_index()) + .prop_map(|(success, last_log_index)| AppendResponse { success, last_log_index }) } fn arb_payload() -> impl Strategy { @@ -94,14 +89,8 @@ fn arb_payload() -> impl Strategy { } fn arb_message() -> impl Strategy { - (arb_node_id(), arb_node_id(), arb_term(), arb_payload()).prop_map( - |(from, to, term, payload)| Message { - from, - to, - term, - payload, - }, - ) + (arb_node_id(), arb_node_id(), arb_term(), arb_payload()) + .prop_map(|(from, to, term, payload)| Message { from, to, term, payload }) } fn arb_event() -> impl Strategy { @@ -131,10 +120,10 @@ fn check_log_continuity(node: &RaftNode) -> bool { if log.get(i).is_none() { return false; } - if let Some(entry) = log.get(i) { - if entry.index != i { - return false; - } + if let Some(entry) = log.get(i) + && entry.index != i + { + return false; } } true @@ -232,17 +221,12 @@ impl ClusterSim { .iter() .map(|&id| { RaftNode::new( - Config::new(id) - .with_prevote(false) - .with_election_timeout(5, 10), + Config::new(id).with_prevote(false).with_election_timeout(5, 10), &voters, ) }) .collect(); - Self { - nodes, - pending_messages: vec![], - } + Self { nodes, pending_messages: vec![] } } fn tick_all(&mut self) { @@ -270,11 +254,7 @@ impl ClusterSim { } fn leaders(&self) -> Vec<(NodeId, Term)> { - self.nodes - .iter() - .filter(|n| n.is_leader()) - .map(|n| (n.id(), n.term())) - .collect() + self.nodes.iter().filter(|n| n.is_leader()).map(|n| (n.id(), n.term())).collect() } fn election_safety(&self) -> bool { diff --git a/consensus/cloud9-raft/tests/state_machine.rs b/consensus/cloud9-raft/tests/state_machine.rs index a8136c2..8b1787f 100644 --- a/consensus/cloud9-raft/tests/state_machine.rs +++ b/consensus/cloud9-raft/tests/state_machine.rs @@ -1,4 +1,5 @@ //! State machine testing for Raft consensus. +#![allow(clippy::cast_possible_truncation, clippy::match_same_arms)] //! //! Uses proptest-state-machine to verify Raft properties by generating //! random sequences of transitions and checking invariants after each step. @@ -10,7 +11,7 @@ use std::collections::BTreeMap; use proptest::prelude::*; use proptest::test_runner::Config; -use proptest_state_machine::{prop_state_machine, ReferenceStateMachine, StateMachineTest}; +use proptest_state_machine::{ReferenceStateMachine, StateMachineTest, prop_state_machine}; use cloud9_raft::raft::{Config as RaftConfig, Effects, Message, RaftNode}; use cloud9_raft::{NodeId, Term}; @@ -51,7 +52,7 @@ pub enum Transition { /// Abstract reference model of a Raft cluster. /// /// Tracks bounds for the state machine test. -/// The SUT (RaftClusterSUT) is the source of truth; this just tracks +/// The SUT (`RaftClusterSUT`) is the source of truth; this just tracks /// what transitions are valid. #[derive(Clone, Debug)] pub struct RaftClusterRef { @@ -67,12 +68,7 @@ pub struct RaftClusterRef { impl RaftClusterRef { fn new() -> Self { - Self { - max_term: 0, - max_log_len: 0, - has_leader: false, - pending_count: 0, - } + Self { max_term: 0, max_log_len: 0, has_leader: false, pending_count: 0 } } } @@ -91,34 +87,18 @@ impl ReferenceStateMachine for RaftStateMachine { // Tick any node (if we haven't exceeded term limit) if state.max_term < MAX_TERM { - strategies.push( - (0..CLUSTER_SIZE) - .prop_map(Transition::Tick) - .boxed(), - ); + strategies.push((0..CLUSTER_SIZE).prop_map(Transition::Tick).boxed()); } // Deliver or drop pending message if state.pending_count > 0 { - strategies.push( - (0..state.pending_count) - .prop_map(Transition::DeliverMessage) - .boxed(), - ); - strategies.push( - (0..state.pending_count) - .prop_map(Transition::DropMessage) - .boxed(), - ); + strategies.push((0..state.pending_count).prop_map(Transition::DeliverMessage).boxed()); + strategies.push((0..state.pending_count).prop_map(Transition::DropMessage).boxed()); } // Propose on a node (if log isn't too long and there's a leader) if state.max_log_len < MAX_LOG_LEN && state.has_leader { - strategies.push( - (0..CLUSTER_SIZE) - .prop_map(Transition::Propose) - .boxed(), - ); + strategies.push((0..CLUSTER_SIZE).prop_map(Transition::Propose).boxed()); } // If no strategies, just tick @@ -152,9 +132,7 @@ impl ReferenceStateMachine for RaftStateMachine { Transition::DeliverMessage(idx) | Transition::DropMessage(idx) => { *idx < state.pending_count } - Transition::Propose(idx) => { - *idx < CLUSTER_SIZE && state.max_log_len < MAX_LOG_LEN - } + Transition::Propose(idx) => *idx < CLUSTER_SIZE && state.max_log_len < MAX_LOG_LEN, } } } @@ -170,14 +148,8 @@ pub struct RaftClusterSUT { impl RaftClusterSUT { fn new() -> Self { let voters = node_ids(); - let nodes = voters - .iter() - .map(|&id| RaftNode::new(raft_config(id), &voters)) - .collect(); - Self { - nodes, - pending_messages: vec![], - } + let nodes = voters.iter().map(|&id| RaftNode::new(raft_config(id), &voters)).collect(); + Self { nodes, pending_messages: vec![] } } fn process_effects(&mut self, effects: &Effects) { @@ -242,12 +214,11 @@ impl StateMachineTest for RaftClusterSUT { } } Transition::Propose(idx) => { - if idx < state.nodes.len() && state.nodes[idx].is_leader() { - if let Ok((_, effects)) = - state.nodes[idx].propose(cloud9_raft::Command(vec![])) - { - state.process_effects(&effects); - } + if idx < state.nodes.len() + && state.nodes[idx].is_leader() + && let Ok((_, effects)) = state.nodes[idx].propose(cloud9_raft::Command(vec![])) + { + state.process_effects(&effects); } } } @@ -309,9 +280,8 @@ impl StateMachineTest for RaftClusterSUT { let prev_term_j = log_j.term_at(prev_idx); assert_eq!( prev_term_i, prev_term_j, - "Log matching violated: nodes {} and {} have same term {} at index {}, \ - but different terms at index {} ({} vs {})", - i, j, term_i, idx, prev_idx, prev_term_i, prev_term_j + "Log matching violated: nodes {i} and {j} have same term {term_i} at index {idx}, \ + but different terms at index {prev_idx} ({prev_term_i} vs {prev_term_j})" ); } } @@ -332,9 +302,8 @@ impl StateMachineTest for RaftClusterSUT { let term_j = state.nodes[j].persistent().log.term_at(idx); assert_eq!( term_i, term_j, - "State machine safety violated: committed entries at index {} \ - have different terms: node {} has term {}, node {} has term {}", - idx, i, term_i, j, term_j + "State machine safety violated: committed entries at index {idx} \ + have different terms: node {i} has term {term_i}, node {j} has term {term_j}" ); } } @@ -415,12 +384,7 @@ mod focused_tests { // Check election safety let leaders_by_term = cluster.leaders_by_term(); for (term, leaders) in &leaders_by_term { - assert!( - leaders.len() <= 1, - "Election safety violated at term {}: {:?}", - term, - leaders - ); + assert!(leaders.len() <= 1, "Election safety violated at term {term}: {leaders:?}"); } } } diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 1a35d66..1bdafd5 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,2 +1,3 @@ [toolchain] -channel = "1.91" +channel = "1.95" +components = ["clippy", "rustfmt"] diff --git a/typos.toml b/typos.toml index 7178c7f..a8df7d8 100644 --- a/typos.toml +++ b/typos.toml @@ -59,6 +59,9 @@ pbft = "pbft" poa = "poa" pos = "pos" +# File formats +edn = "edn" + [type.rust] extend-glob = ["*.rs"]