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