Skip to content

Commit b2fb2de

Browse files
committed
refactor(consensus): satisfy rust lints
1 parent a8faaf6 commit b2fb2de

22 files changed

Lines changed: 515 additions & 914 deletions

File tree

cloud9-node/src/lib.rs

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,12 @@ use cloud9_storage::StorageOptions;
55
use tracing::{info, instrument};
66

77
/// Runtime configuration derived from CLI flags and config files.
8-
#[derive(Debug, Clone)]
8+
#[derive(Debug, Clone, Default)]
99
pub struct NodeConfig {
1010
pub storage: StorageOptions,
1111
pub consensus: ConsensusConfig,
1212
}
1313

14-
impl Default for NodeConfig {
15-
fn default() -> Self {
16-
Self { storage: StorageOptions::default(), consensus: ConsensusConfig::default() }
17-
}
18-
}
19-
2014
/// Launch the storage and consensus subsystems.
2115
#[instrument(skip_all)]
2216
pub async fn launch(config: NodeConfig) {

cloud9/src/main.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ fn main() -> Result<()> {
4747
match cli.command {
4848
Command::Start { config } => {
4949
let config_path = config.unwrap_or_else(|| PathBuf::from("cloud9.toml"));
50-
let maybe_config = load_config(config_path.clone())?;
50+
let maybe_config = load_config(&config_path)?;
5151
tracing::info!(path = %config_path.display(), "booting node");
5252
if let Some(config) = maybe_config {
5353
tracing::debug!(contents = %config, "loaded configuration");
@@ -56,7 +56,7 @@ fn main() -> Result<()> {
5656
}
5757
}
5858
Command::CheckConfig { config } => {
59-
load_config(config.clone())
59+
load_config(&config)
6060
.and_then(|contents| {
6161
contents.ok_or_else(|| miette::miette!("config `{}` missing", config.display()))
6262
})
@@ -89,9 +89,9 @@ fn init_tracing(verbosity: u8, color_enabled: bool) -> Result<()> {
8989
tracing_subscriber::registry().with(filter).with(fmt_layer).try_init().into_diagnostic()
9090
}
9191

92-
fn load_config(path: PathBuf) -> Result<Option<String>> {
92+
fn load_config(path: &PathBuf) -> Result<Option<String>> {
9393
if path.exists() {
94-
fs::read_to_string(&path)
94+
fs::read_to_string(path)
9595
.into_diagnostic()
9696
.map(Some)
9797
.with_context(|| format!("reading configuration from `{}`", path.display()))

consensus/cloud9-raft-io/src/lib.rs

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
//! I/O layer for cloud9-raft.
2+
#![cfg_attr(test, allow(clippy::unwrap_used))]
23
//!
34
//! This crate provides the integration between the pure Raft state machine
45
//! (`cloud9-raft`) and the outside world. It defines traits for:
@@ -24,6 +25,8 @@ mod storage;
2425
mod transport;
2526

2627
pub use read_index::{ReadId, ReadIndexCoordinator, ReadIndexError, ReadIndexResult};
27-
pub use session::{ClientId, ClientSession, DuplicateCheck, SequenceNum, SessionRequest, SessionTracker};
28+
pub use session::{
29+
ClientId, ClientSession, DuplicateCheck, SequenceNum, SessionRequest, SessionTracker,
30+
};
2831
pub use storage::{SnapshotData, Storage, StorageError};
2932
pub use transport::{Transport, TransportError};

consensus/cloud9-raft-io/src/read_index.rs

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
//! 2. Leader records the current commit index as the "read index"
1010
//! 3. Leader sends a heartbeat round and waits for majority acknowledgment
1111
//! 4. Once majority confirms, the leader is guaranteed to still be leader
12-
//! 5. The read can proceed once the state machine has applied up to read_index
12+
//! 5. The read can proceed once the state machine has applied up to `read_index`
1313
//!
1414
//! This module provides `ReadIndexCoordinator` which tracks pending reads and
1515
//! completes them when heartbeat quorum is achieved.
@@ -33,7 +33,7 @@ impl ReadId {
3333
/// A pending read request awaiting heartbeat confirmation.
3434
#[derive(Debug)]
3535
struct PendingRead {
36-
/// The read index (commit_index when request was made).
36+
/// The read index (`commit_index` when request was made).
3737
read_index: LogIndex,
3838
/// Heartbeat round this read is waiting on.
3939
round: u64,
@@ -134,13 +134,7 @@ impl ReadIndexCoordinator {
134134
/// The caller is responsible for checking `can_serve_reads()` before calling this.
135135
pub fn request_read(&mut self, read_index: LogIndex) -> ReadId {
136136
let id = self.next_read_id.next();
137-
self.pending.insert(
138-
id,
139-
PendingRead {
140-
read_index,
141-
round: self.current_round,
142-
},
143-
);
137+
self.pending.insert(id, PendingRead { read_index, round: self.current_round });
144138
id
145139
}
146140

@@ -166,7 +160,7 @@ impl ReadIndexCoordinator {
166160
/// Record a heartbeat acknowledgment from a peer.
167161
///
168162
/// Call this when an `AppendResponse { success: true }` is received
169-
/// from a heartbeat (empty AppendEntries).
163+
/// from a heartbeat (empty `AppendEntries`).
170164
pub fn record_ack(&mut self, from: NodeId) {
171165
if !self.round_acks.contains(&from) && self.voters.contains(&from) {
172166
self.round_acks.push(from);

consensus/cloud9-raft-io/src/session.rs

Lines changed: 8 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
//!
88
//! The solution is to track client sessions:
99
//! 1. Clients register and receive unique IDs
10-
//! 2. Each request carries (client_id, sequence_number)
10+
//! 2. Each request carries (`client_id`, `sequence_number`)
1111
//! 3. The state machine tracks the last completed sequence per client
1212
//! 4. Duplicate requests return cached responses
1313
//!
@@ -95,10 +95,7 @@ pub struct ClientSession<R> {
9595

9696
impl<R> Default for ClientSession<R> {
9797
fn default() -> Self {
98-
Self {
99-
last_sequence: 0,
100-
last_response: None,
101-
}
98+
Self { last_sequence: 0, last_response: None }
10299
}
103100
}
104101

@@ -138,10 +135,7 @@ pub struct SessionTracker<R> {
138135
impl<R: Clone> SessionTracker<R> {
139136
/// Create a new session tracker.
140137
pub fn new() -> Self {
141-
Self {
142-
sessions: BTreeMap::new(),
143-
next_client_id: 1,
144-
}
138+
Self { sessions: BTreeMap::new(), next_client_id: 1 }
145139
}
146140

147141
/// Register a new client session.
@@ -185,12 +179,7 @@ impl<R: Clone> SessionTracker<R> {
185179
///
186180
/// Call this after executing a command. Caches the response for
187181
/// duplicate detection.
188-
pub fn record_completion(
189-
&mut self,
190-
client_id: ClientId,
191-
sequence: SequenceNum,
192-
response: R,
193-
) {
182+
pub fn record_completion(&mut self, client_id: ClientId, sequence: SequenceNum, response: R) {
194183
let session = self.sessions.entry(client_id).or_default();
195184
if sequence > session.last_sequence {
196185
session.last_sequence = sequence;
@@ -244,10 +233,7 @@ mod tests {
244233
let mut tracker: SessionTracker<String> = SessionTracker::new();
245234
let client = tracker.register_client();
246235

247-
assert!(matches!(
248-
tracker.check_duplicate(client, 1),
249-
DuplicateCheck::New
250-
));
236+
assert!(matches!(tracker.check_duplicate(client, 1), DuplicateCheck::New));
251237
}
252238

253239
#[test]
@@ -274,10 +260,7 @@ mod tests {
274260
tracker.record_completion(client, 5, "response5".to_string());
275261

276262
// Sequence 3 is stale
277-
assert!(matches!(
278-
tracker.check_duplicate(client, 3),
279-
DuplicateCheck::Stale
280-
));
263+
assert!(matches!(tracker.check_duplicate(client, 3), DuplicateCheck::Stale));
281264
}
282265

283266
#[test]
@@ -288,10 +271,7 @@ mod tests {
288271
tracker.record_completion(client, 1, "response1".to_string());
289272

290273
// Sequence 2 is new
291-
assert!(matches!(
292-
tracker.check_duplicate(client, 2),
293-
DuplicateCheck::New
294-
));
274+
assert!(matches!(tracker.check_duplicate(client, 2), DuplicateCheck::New));
295275
}
296276

297277
#[test]
@@ -309,10 +289,7 @@ mod tests {
309289
let tracker: SessionTracker<String> = SessionTracker::new();
310290
let unknown = ClientId(999);
311291

312-
assert!(matches!(
313-
tracker.check_duplicate(unknown, 1),
314-
DuplicateCheck::New
315-
));
292+
assert!(matches!(tracker.check_duplicate(unknown, 1), DuplicateCheck::New));
316293
}
317294

318295
#[test]

consensus/cloud9-raft-io/src/storage.rs

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,19 +43,28 @@ pub trait Storage: Send + Sync {
4343
///
4444
/// Per §3.8, this must complete before responding to any RPC.
4545
/// If `sync` is true, the implementation should fsync to ensure durability.
46-
fn save(&self, state: &Persistent, sync: bool) -> impl Future<Output = Result<(), StorageError>> + Send;
46+
fn save(
47+
&self,
48+
state: &Persistent,
49+
sync: bool,
50+
) -> impl Future<Output = Result<(), StorageError>> + Send;
4751

4852
/// Load snapshot data for transfer to a slow follower (§5).
4953
///
5054
/// Returns the snapshot data as bytes. The format is opaque to Raft —
5155
/// the state machine defines it.
52-
fn load_snapshot(&self) -> impl Future<Output = Result<Option<SnapshotData>, StorageError>> + Send;
56+
fn load_snapshot(
57+
&self,
58+
) -> impl Future<Output = Result<Option<SnapshotData>, StorageError>> + Send;
5359

5460
/// Save snapshot data received from the leader (§5).
5561
///
56-
/// Called when a follower receives InstallSnapshot. The implementation
62+
/// Called when a follower receives `InstallSnapshot`. The implementation
5763
/// should atomically replace any existing snapshot.
58-
fn save_snapshot(&self, snapshot: SnapshotData) -> impl Future<Output = Result<(), StorageError>> + Send;
64+
fn save_snapshot(
65+
&self,
66+
snapshot: SnapshotData,
67+
) -> impl Future<Output = Result<(), StorageError>> + Send;
5968
}
6069

6170
/// Snapshot data for transfer between nodes (§5).

consensus/cloud9-raft-io/src/transport.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@
55
66
use std::future::Future;
77

8-
use cloud9_raft::raft::Message;
98
use cloud9_raft::NodeId;
9+
use cloud9_raft::raft::Message;
1010
use thiserror::Error;
1111

1212
/// Errors from transport operations.
@@ -58,7 +58,10 @@ pub trait Transport: Send + Sync {
5858
///
5959
/// Called when configuration changes. The transport should establish
6060
/// connections to new nodes and may close connections to removed nodes.
61-
fn update_peers(&self, peers: &[NodeId]) -> impl Future<Output = Result<(), TransportError>> + Send;
61+
fn update_peers(
62+
&self,
63+
peers: &[NodeId],
64+
) -> impl Future<Output = Result<(), TransportError>> + Send;
6265
}
6366

6467
#[cfg(test)]

consensus/cloud9-raft/src/lib.rs

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,14 @@
11
#![forbid(unsafe_code)]
22
#![deny(clippy::unwrap_used, clippy::expect_used)]
3-
#![cfg_attr(any(test, doctest), allow(clippy::unwrap_used, clippy::expect_used))]
3+
#![cfg_attr(
4+
any(test, doctest),
5+
allow(
6+
clippy::unwrap_used,
7+
clippy::expect_used,
8+
clippy::panic,
9+
clippy::cast_possible_truncation
10+
)
11+
)]
412

513
//! Consensus drivers for Cloud9 clusters.
614
//!

0 commit comments

Comments
 (0)