From 75d571da04dec3b01451cefa5fe48f22f2592009 Mon Sep 17 00:00:00 2001 From: jetie <1405758738@qq.com> Date: Sun, 14 Jun 2026 18:29:56 +0800 Subject: [PATCH] fix(raft): fix election and network issues in raft event loop Replace spin-based ticking with Instant-based accurate timing to prevent tick drift. Reorder ready processing to send leader messages before persisting entries, matching raft-rs best practices. Fix heartbeat interval from 100ms to 200ms and set max_committed_size_per_ready to ensure entries are returned from ready(). Add tracing for raft messages, proposals, and leader state transitions to aid debugging. Improve network sender error handling to distinguish between RPC errors and task panics. --- src/main.rs | 3 +- src/raft/handle.rs | 1 + src/raft/network.rs | 21 ++- src/raft/node.rs | 281 ++++++++++++++++++++++++--------------- src/raft/raftrs_store.rs | 2 +- src/raft/rpc.rs | 19 ++- 6 files changed, 218 insertions(+), 109 deletions(-) diff --git a/src/main.rs b/src/main.rs index 376b959..20363bb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -429,9 +429,10 @@ async fn main() -> anyhow::Result<()> { let raft_config = raft::Config { id: config.node_id, election_tick: 10, // 10 ticks = 1 second (100ms per tick) - heartbeat_tick: 1, // 1 tick = 100ms heartbeat + heartbeat_tick: 2, // 2 ticks = 200ms heartbeat max_size_per_msg: 1024 * 1024, // 1MB max_inflight_msgs: 256, + max_committed_size_per_ready: 1024 * 1024, // 1MB — 0 means no entries returned ..Default::default() }; diff --git a/src/raft/handle.rs b/src/raft/handle.rs index 0413a39..6f72027 100644 --- a/src/raft/handle.rs +++ b/src/raft/handle.rs @@ -74,6 +74,7 @@ pub trait RaftHandle: Send + Sync + 'static { /// Returns `Ok(())` if this node is the leader, `Err` with leader redirect info otherwise. pub fn require_leader(raft: &dyn RaftHandle, node_id: NodeId) -> Result<(), Status> { let current_leader = raft.leader_id(); + tracing::info!(node_id = node_id, leader_id = ?current_leader, "require_leader check"); match current_leader { Some(id) if id == node_id => Ok(()), Some(leader_id) => { diff --git a/src/raft/network.rs b/src/raft/network.rs index 3850769..4843afd 100644 --- a/src/raft/network.rs +++ b/src/raft/network.rs @@ -50,8 +50,14 @@ impl NetworkSender { .collect(); for fut in futs { - if let Err(e) = fut.await { - tracing::warn!(node_id = self.node_id, error = %e, "failed to send raft message"); + match fut.await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + tracing::warn!(node_id = self.node_id, error = %e, "failed to send raft message"); + } + Err(e) => { + tracing::error!(node_id = self.node_id, error = %e, "send task panicked"); + } } } } @@ -89,6 +95,17 @@ async fn send_message( let payload = msg.encode_to_vec(); let msg_type = MessageType::from_i32(msg.msg_type).unwrap_or(MessageType::MsgHup); + let target_addr = addrs.get(&to).cloned().unwrap_or_default(); + + tracing::info!( + to = to, + target_addr = %target_addr, + msg_type = ?msg_type, + term = msg.term, + from = msg.from, + payload_len = payload.len(), + "sending raft message" + ); let mut client = get_or_connect(&clients, &addrs, to).await?; diff --git a/src/raft/node.rs b/src/raft/node.rs index b0d0531..5a6b767 100644 --- a/src/raft/node.rs +++ b/src/raft/node.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; -use std::time::Duration; +use std::time::{Duration, Instant}; use crossbeam_channel::{self as cc, Receiver as CcReceiver}; use prost011::Message as ProstMessage; @@ -270,9 +270,8 @@ fn raft_event_loop( bootstrapped = true; } - // Maximum iterations before re-blocking on select. Prevents starvation when one - // channel is always ready (e.g. continuous proposals). - const MAX_SPIN: usize = 64; + // Track the next tick time using Instant for accurate time-based ticking + let mut next_tick = Instant::now() + tick_interval; loop { // Drain all pending channel messages first, then tick. @@ -295,8 +294,14 @@ fn raft_event_loop( ris.push(ri_req); } - // If nothing was immediately available, block until something arrives or tick fires. - if msgs.is_empty() && props.is_empty() && ccs.is_empty() && ris.is_empty() { + // Check if there is any work to do + let has_work = !msgs.is_empty() || !props.is_empty() || !ccs.is_empty() || !ris.is_empty(); + let now = Instant::now(); + let tick_due = now >= next_tick; + + // If nothing was immediately available and tick is not due, block until something arrives or tick fires. + if !has_work && !tick_due { + let remaining = next_tick - now; cc::select! { recv(msg_in_rx) -> msg => { if let Ok(m) = msg { msgs.push(m); } @@ -314,11 +319,14 @@ fn raft_event_loop( if let Ok(r) = ri_req { ris.push(r); } while let Ok(r) = read_index_rx.try_recv() { ris.push(r); } } - recv(cc::after(tick_interval)) -> _ => {} + recv(cc::after(remaining)) -> _ => {} } } // Step incoming raft messages. + if !msgs.is_empty() { + tracing::info!(node_id = config.id, messages_count = msgs.len(), "stepping {} messages", msgs.len()); + } for msg in msgs { if let Err(e) = node.step(msg) { tracing::warn!(error = %e, "step failed"); @@ -359,6 +367,7 @@ fn raft_event_loop( // Process normal proposals. for prop in props { + tracing::info!(node_id = config.id, state = ?node.raft.state, leader_id = node.raft.leader_id, "processing propose request"); if node.raft.state != StateRole::Leader { metrics .raft_proposals_total @@ -391,45 +400,67 @@ fn raft_event_loop( } } - if needs_bootstrap && !bootstrapped && node.raft.state == StateRole::Leader { - let mut cc = ConfChange::default(); - cc.set_change_type(ConfChangeType::AddNode); - cc.node_id = config.id; - if let Err(e) = node.propose_conf_change(vec![], cc) { - tracing::error!(error = %e, "bootstrap conf change failed"); - } - bootstrapped = true; - tracing::info!(node_id = config.id, "proposed initial ConfChange"); + // Tick the raft state machine only when the scheduled time is reached. + let now = Instant::now(); + if now >= next_tick { + node.tick(); + tracing::info!(node_id = config.id, "tick called"); + next_tick = now + tick_interval; } - // Tick the raft state machine. - node.tick(); - if !node.has_ready() { shared_state .leader_id .store(node.raft.leader_id, Ordering::Relaxed); shared_state.term.store(node.raft.term, Ordering::Relaxed); + tracing::info!(node_id = config.id, leader_id = node.raft.leader_id, term = node.raft.term, state = ?node.raft.state, "updated leader_id"); continue; } // Process all available ready states before blocking again. - for _ in 0..MAX_SPIN { - if !node.has_ready() { - break; - } + while node.has_ready() { let mut ready = node.ready(); - // Step 1: Persist entries to the log store. - if !ready.entries().is_empty() - && let Err(e) = node.store().append_entries(ready.entries()) - { - tracing::error!(error = %e, "failed to persist entries"); - return; + // Step 1: Send outgoing messages from Leader BEFORE persisting. + // According to raft-rs best practices, Leader should send messages in parallel with persistence. + // Only Leader has messages in take_messages(); Follower messages come from take_persisted_messages(). + let mut messages = ready.take_messages(); + if !messages.is_empty() { + tracing::info!(messages_count = messages.len(), "sending leader messages before persistence"); + for msg in &messages { + tracing::debug!(from = msg.from, to = msg.to, msg_type = ?raft::eraftpb::MessageType::from_i32(msg.msg_type), "leader outgoing message"); + } + let _ = msg_out_tx.blocking_send(messages); + } + + // Step 1b: Handle committed entries from Ready (before persistence). + // According to raft-rs examples, committed entries can come from both Ready and LightReady. + let ready_committed = ready.take_committed_entries(); + if !ready_committed.is_empty() { + process_committed_entries( + ready_committed, + &mut node, + &state_machine, + &shared_state, + &mut pending, + &mut pending_conf, + &mut pending_read_index, + &mut last_snapshot_applied, + snapshot_trigger, + ); + } + + // Step 2: Persist entries to the log store. + if !ready.entries().is_empty() { + tracing::info!(entries_count = ready.entries().len(), "persisting entries from Ready"); + if let Err(e) = node.store().append_entries(ready.entries()) { + tracing::error!(error = %e, "failed to persist entries"); + return; + } } - // Step 2: Persist hard state (term, vote, commit). + // Step 3: Persist hard state (term, vote, commit). if let Some(hs) = ready.hs() && let Err(e) = node.store().save_hard_state(hs) { @@ -437,10 +468,14 @@ fn raft_event_loop( return; } - // Step 3: Send outgoing messages. - let messages = ready.take_messages(); - if !messages.is_empty() { - let _ = msg_out_tx.blocking_send(messages); + // Step 4: Send persisted messages (mainly for Follower nodes). + let persisted_msgs = ready.take_persisted_messages(); + if !persisted_msgs.is_empty() { + tracing::info!(persisted_count = persisted_msgs.len(), "sending persisted messages"); + for msg in &persisted_msgs { + tracing::debug!(from = msg.from, to = msg.to, msg_type = ?raft::eraftpb::MessageType::from_i32(msg.msg_type), "persisted outgoing message"); + } + let _ = msg_out_tx.blocking_send(persisted_msgs); } // Step 4: Handle incoming snapshot. @@ -473,86 +508,35 @@ fn raft_event_loop( // the LightReady returned by advance(), not in the original Ready. let mut light_ready = node.advance(ready); + // Step 6b: Send messages from LightReady (may contain responses after advance). + let light_msgs = light_ready.take_messages(); + if !light_msgs.is_empty() { + tracing::info!(messages_count = light_msgs.len(), "sending messages from LightReady"); + let _ = msg_out_tx.blocking_send(light_msgs); + } + // Step 7: Process committed entries from LightReady. let committed = light_ready.take_committed_entries(); if !committed.is_empty() { - let max_committed = committed.last().map(|e| e.index).unwrap_or(0); - shared_state - .commit_index - .store(max_committed, Ordering::Release); - - // Skip entries that were already applied before the last shutdown. - let already_applied = shared_state.applied_index.load(Ordering::Acquire); - let to_apply: Vec<_> = committed - .into_iter() - .filter(|e| e.index > already_applied) - .collect(); - - if !to_apply.is_empty() { - let new_applied = to_apply.last().map(|e| e.index).unwrap_or(0); - let mut sm = state_machine.lock().expect("state machine mutex poisoned"); - for entry in &to_apply { - apply_entry(&mut sm, entry, &mut pending, &mut pending_conf); - } - drop(sm); - - if let Err(e) = node.store().save_applied_index(new_applied) { - tracing::error!(error = %e, "failed to persist applied index"); - } - shared_state - .applied_index - .store(new_applied, Ordering::Release); - - // Persist ConfState for any conf change entries. - for entry in &to_apply { - if entry.get_entry_type() == EntryType::EntryConfChange - && !entry.data.is_empty() - && let Ok(cc) = ConfChange::decode(entry.data.as_slice()) - { - match node.apply_conf_change(&cc) { - Ok(cs) => { - if let Err(e) = node.store().save_conf_state(&cs) { - tracing::error!(error = %e, "failed to persist conf state"); - } - } - Err(e) => { - tracing::error!( - error = %e, - index = entry.index, - "failed to apply conf change to raft" - ); - } - } - } - } - - // Leader-side snapshot trigger. - if snapshot_trigger > 0 - && node.raft.state == StateRole::Leader - && new_applied - last_snapshot_applied >= snapshot_trigger - { - match create_leader_snapshot(node.store(), &state_machine, new_applied) { - Ok(()) => last_snapshot_applied = new_applied, - Err(e) => { - tracing::error!(error = %e, "failed to create snapshot"); - } - } - } - } - - // Notify waiters that applied_index has advanced. - shared_state.applied_notify.notify_waiters(); + process_committed_entries( + committed, + &mut node, + &state_machine, + &shared_state, + &mut pending, + &mut pending_conf, + &mut pending_read_index, + &mut last_snapshot_applied, + snapshot_trigger, + ); } - // Step 8: Send any additional messages from LightReady. + // Step 10: Send any additional messages from LightReady. let light_messages = light_ready.take_messages(); if !light_messages.is_empty() { let _ = msg_out_tx.blocking_send(light_messages); } - // Step 9: Signal that committed entries have been applied. - node.advance_apply(); - // Track leader changes and drain pending on step-down. let is_leader = node.raft.state == StateRole::Leader; if was_leader && !is_leader { @@ -729,3 +713,92 @@ fn apply_snapshot_to_state( fn is_empty_snapshot(snap: &raft::eraftpb::Snapshot) -> bool { snap.get_data().is_empty() && snap.get_metadata().index == 0 } + +/// Process committed entries from Ready or LightReady. +/// This function applies entries to the state machine and updates indices. +#[allow(clippy::too_many_arguments)] +fn process_committed_entries( + committed: Vec, + node: &mut RawNode, + state_machine: &Arc>, + shared_state: &RaftSharedState, + pending: &mut HashMap, RaftError>>>, + pending_conf: &mut HashMap<(u64, i32), oneshot::Sender>>, + _pending_read_index: &mut HashMap, oneshot::Sender>>, + last_snapshot_applied: &mut u64, + snapshot_trigger: u64, +) { + if committed.is_empty() { + return; + } + + let max_committed = committed.last().map(|e| e.index).unwrap_or(0); + shared_state + .commit_index + .store(max_committed, Ordering::Release); + + // Skip entries that were already applied before the last shutdown. + let already_applied = shared_state.applied_index.load(Ordering::Acquire); + let to_apply: Vec<_> = committed + .into_iter() + .filter(|e| e.index > already_applied) + .collect(); + + if !to_apply.is_empty() { + let new_applied = to_apply.last().map(|e| e.index).unwrap_or(0); + let mut sm = state_machine.lock().expect("state machine mutex poisoned"); + for entry in &to_apply { + apply_entry(&mut sm, entry, pending, pending_conf); + } + drop(sm); + + if let Err(e) = node.store().save_applied_index(new_applied) { + tracing::error!(error = %e, "failed to persist applied index"); + } + shared_state + .applied_index + .store(new_applied, Ordering::Release); + + // Persist ConfState for any conf change entries. + for entry in &to_apply { + if entry.get_entry_type() == EntryType::EntryConfChange + && !entry.data.is_empty() + && let Ok(cc) = ConfChange::decode(entry.data.as_slice()) + { + match node.apply_conf_change(&cc) { + Ok(cs) => { + if let Err(e) = node.store().save_conf_state(&cs) { + tracing::error!(error = %e, "failed to persist conf state"); + } + } + Err(e) => { + tracing::error!( + error = %e, + index = entry.index, + "failed to apply conf change to raft" + ); + } + } + } + } + + // Leader-side snapshot trigger. + if snapshot_trigger > 0 + && node.raft.state == StateRole::Leader + && new_applied - *last_snapshot_applied >= snapshot_trigger + { + match create_leader_snapshot(node.store(), state_machine, new_applied) { + Ok(()) => *last_snapshot_applied = new_applied, + Err(e) => { + tracing::error!(error = %e, "failed to create snapshot"); + } + } + } + + // Signal that committed entries have been applied. + node.advance_apply(); + + // Notify waiters that applied_index has advanced. + shared_state.applied_notify.notify_waiters(); + } +} diff --git a/src/raft/raftrs_store.rs b/src/raft/raftrs_store.rs index ff20eb9..7ce019f 100644 --- a/src/raft/raftrs_store.rs +++ b/src/raft/raftrs_store.rs @@ -351,7 +351,7 @@ impl Storage for RaftRsStore { low: u64, high: u64, max_size: impl Into>, - _context: GetEntriesContext, + context: GetEntriesContext, ) -> RaftResult> { let max_size = max_size.into(); let first = self.first_index()?; diff --git a/src/raft/rpc.rs b/src/raft/rpc.rs index 9d0b5fa..2e7e845 100644 --- a/src/raft/rpc.rs +++ b/src/raft/rpc.rs @@ -1,5 +1,5 @@ use prost011::Message as ProstMessage; -use raft::eraftpb::Message; +use raft::eraftpb::{Message, MessageType}; use tokio::sync::mpsc; use tonic::{Request, Response, Status}; @@ -25,6 +25,7 @@ impl RaftRpc for RaftRpcImpl { request: Request, ) -> Result, Status> { let req = request.into_inner(); + let msg = Message::decode(req.payload.as_slice()) .map_err(|e| Status::internal(format!("decode error: {e}")))?; @@ -44,6 +45,14 @@ impl RaftRpc for RaftRpcImpl { let msg = Message::decode(req.payload.as_slice()) .map_err(|e| Status::internal(format!("decode error: {e}")))?; + tracing::info!( + from = msg.from, + to = msg.to, + term = msg.term, + msg_type = ?MessageType::from_i32(msg.msg_type), + "received Vote RPC" + ); + self.msg_tx .send(msg) .await @@ -60,6 +69,14 @@ impl RaftRpc for RaftRpcImpl { let msg = Message::decode(req.payload.as_slice()) .map_err(|e| Status::internal(format!("decode error: {e}")))?; + tracing::debug!( + from = msg.from, + to = msg.to, + term = msg.term, + msg_type = msg.msg_type, + "received InstallSnapshot" + ); + self.msg_tx .send(msg) .await