diff --git a/d-engine-core/benches/leader_state_bench.rs b/d-engine-core/benches/leader_state_bench.rs index b7813cf5..94c87c4f 100644 --- a/d-engine-core/benches/leader_state_bench.rs +++ b/d-engine-core/benches/leader_state_bench.rs @@ -113,9 +113,9 @@ use std::time::Duration; use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main}; use d_engine_core::leader_state::{LeaderState, PendingPromotion}; use d_engine_core::{ - AppendResults, MockElectionCore, MockMembership, MockPurgeExecutor, MockRaftLog, - MockReplicationCore, MockStateMachine, MockStateMachineHandler, MockTransport, MockTypeConfig, - PeerUpdate, RaftContext, RaftCoreHandlers, RaftNodeConfig, RaftStorageHandles, + MockElectionCore, MockMembership, MockPurgeExecutor, MockRaftLog, MockReplicationCore, + MockStateMachine, MockStateMachineHandler, MockTransport, MockTypeConfig, RaftContext, + RaftCoreHandlers, RaftNodeConfig, RaftStorageHandles, }; use d_engine_proto::common::{LogId, NodeStatus}; use d_engine_proto::server::cluster::{ClusterMembership, NodeMeta}; @@ -204,19 +204,7 @@ impl BenchFixture { .expect_broadcast_vote_requests() .returning(|_, _, _, _, _| Ok(())); - let mut replication_handler = MockReplicationCore::new(); - replication_handler - .expect_handle_raft_request_in_batch() - .returning(|_, _, _, _, _| { - Ok(AppendResults { - commit_quorum_achieved: true, - learner_progress: std::collections::HashMap::new(), - peer_updates: std::collections::HashMap::from([ - (2, PeerUpdate::success(5, 6)), - (3, PeerUpdate::success(5, 6)), - ]), - }) - }); + let replication_handler = MockReplicationCore::new(); let mut state_machine_handler = MockStateMachineHandler::new(); state_machine_handler.expect_update_pending().returning(|_| {}); diff --git a/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs index 3b90dee4..3dfcb839 100644 --- a/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/client_write_test.rs @@ -478,7 +478,7 @@ async fn test_process_raft_request_batching_enabled() { let mut test_context = setup_process_raft_request_test_context( "test_process_raft_request_batching_enabled", 100, // batch_threshold: 100 means batching enabled - 0, // expect handle_raft_request_in_batch NOT to be called immediately + 0, // no prepare_batch_requests call expected (request buffered) graceful_rx, ) .await; diff --git a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs index 026ecad8..26e24a83 100644 --- a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs @@ -1478,9 +1478,6 @@ async fn test_merge_batch_to_write_metadata_empty_payload_with_senders() { /// B1: Two-node cluster achieves quorum when peer responds with success. /// -/// # Previously verified via handle_raft_request_in_batch: -/// peer_updates[peer2] = PeerUpdate{match_index:Some(3), next_index:4, success:true} -/// /// # Given /// - 2-node cluster (leader=1, peer=2), initial commit=0, leader_term=1 /// - Peer 2 responds: success, term=1, match_index=3 diff --git a/d-engine-core/src/replication/mod.rs b/d-engine-core/src/replication/mod.rs index 92830efa..58317dbd 100644 --- a/d-engine-core/src/replication/mod.rs +++ b/d-engine-core/src/replication/mod.rs @@ -29,7 +29,6 @@ use tonic::Status; use super::LeaderStateSnapshot; use super::StateSnapshot; -use crate::AppendResults; use crate::MaybeCloneOneshotSender; use crate::Result; use crate::TypeConfig; @@ -100,55 +99,6 @@ where ctx: &crate::RaftContext, ) -> Result; - /// As Leader, send replications to peers (combines regular heartbeats and client proposals). - /// - /// Performs peer synchronization checks: - /// 1. Verifies if any peer's next_id <= leader's commit_index - /// 2. For non-synced peers: retrieves unsynced logs and buffers them - /// 3. Prepends unsynced entries to the entries queue - /// - /// # Returns - /// - `Ok(AppendResults)` with aggregated replication outcomes - /// - `Err` for unrecoverable errors - /// - /// # Return Result Semantics - /// 1. **Insufficient Quorum**: Returns `Ok(AppendResults)` with `commit_quorum_achieved = - /// false` when: - /// - Responses received from all nodes but majority acceptance not achieved - /// - Partial timeouts reduce successful responses below majority - /// - /// 2. **Timeout Handling**: - /// - Partial timeouts: Returns `Ok` with `commit_quorum_achieved = false` - /// - Complete timeout: Returns `Ok` with `commit_quorum_achieved = false` - /// - Timeout peers are EXCLUDED from `peer_updates` - /// - /// 3. **Error Conditions**: Returns `Err` ONLY for: - /// - Empty voting members (`ReplicationError::NoPeerFound`) - /// - Log generation failures (`generate_new_entries` errors) - /// - Higher term detected in peer response (`ReplicationError::HigherTerm`) - /// - Critical response handling errors - /// - /// # Guarantees - /// - Only peers with successful responses appear in `peer_updates` - /// - Timeouts never cause top-level `Err` (handled as failed responses) - /// - Leader self-vote always counted in quorum calculation - /// - /// # Note - /// - Leader state should be updated by LeaderState only(follows SRP). - /// - /// # Quorum - /// - If there are no voters (not even the leader), quorum is not possible. - /// - If the leader is the only voter, quorum is always achieved. - /// - If all nodes are learners, quorum is not achieved. - async fn handle_raft_request_in_batch( - &self, - entry_payloads: Vec, - state_snapshot: StateSnapshot, - leader_state_snapshot: LeaderStateSnapshot, - cluster_metadata: &crate::raft_role::ClusterMetadata, - ctx: &crate::RaftContext, - ) -> Result; - /// Handles successful AppendEntries responses /// /// Updates peer match/next indices according to: diff --git a/d-engine-core/src/replication/replication_handler.rs b/d-engine-core/src/replication/replication_handler.rs index 4f3d52dc..5ad71ad8 100644 --- a/d-engine-core/src/replication/replication_handler.rs +++ b/d-engine-core/src/replication/replication_handler.rs @@ -10,37 +10,30 @@ use d_engine_proto::client::WriteCommand; use d_engine_proto::common::Entry; use d_engine_proto::common::EntryPayload; use d_engine_proto::common::LogId; -use d_engine_proto::common::NodeRole; use d_engine_proto::common::entry_payload::Payload; use d_engine_proto::server::replication::AppendEntriesRequest; use d_engine_proto::server::replication::AppendEntriesResponse; use d_engine_proto::server::replication::ConflictResult; use d_engine_proto::server::replication::SuccessResult; -use d_engine_proto::server::replication::append_entries_response; use prost::Message; use tracing::debug; use tracing::error; -use tracing::info; use tracing::trace; use tracing::warn; use super::AppendResponseWithUpdates; use super::PrepareResult; use super::ReplicationCore; -use crate::AppendResults; use crate::IdAllocationError; use crate::LeaderStateSnapshot; use crate::PeerUpdate; -use crate::RaftContext; use crate::RaftLog; use crate::ReplicationError; use crate::Result; use crate::StateSnapshot; -use crate::Transport; use crate::TypeConfig; use crate::alias::ROF; use crate::scoped_timer::ScopedTimer; -use crate::utils::cluster::is_majority; pub struct ReplicationHandler where @@ -132,212 +125,6 @@ where }) } - async fn handle_raft_request_in_batch( - &self, - entry_payloads: Vec, - state_snapshot: StateSnapshot, - leader_state_snapshot: LeaderStateSnapshot, - cluster_metadata: &crate::raft_role::ClusterMetadata, - ctx: &RaftContext, - ) -> Result { - let _timer = ScopedTimer::new("handle_raft_request_in_batch"); - - debug!("-------- handle_raft_request_in_batch --------"); - - // ---------------------- - // Phase 1: Pre-Checks and Cluster Topology Detection - // ---------------------- - // Use cached replication targets from cluster metadata (zero-cost) - let replication_targets = &cluster_metadata.replication_targets; - - // Separate Voters and Learners - // Use role (not status) to distinguish: Follower/Candidate are voters, Learner are learners - // This is more robust than using status, which can be temporarily non-Active - let (voters, learners): (Vec<_>, Vec<_>) = replication_targets - .iter() - .partition(|node| node.role != NodeRole::Learner as i32); - - if !learners.is_empty() { - trace!( - "handle_raft_request_in_batch - voters: {:?}, learners: {:?}", - voters, learners - ); - } - - // ---------------------- - // Phase 2: Process Client Commands - // ---------------------- - - // Record down the last index before new inserts, to avoid duplicated entries, bugfix#48 - let raft_log = ctx.raft_log(); - let leader_last_index_before = raft_log.last_entry_id(); - - let new_entries = self - .generate_new_entries(entry_payloads, state_snapshot.current_term, raft_log) - .await?; - - // ---------------------- - // Phase 3: Prepare Replication Data - // ---------------------- - let replication_data = ReplicationData { - leader_last_index_before, - current_term: state_snapshot.current_term, - commit_index: state_snapshot.commit_index, - peer_next_indices: leader_state_snapshot.next_index, - }; - - let mut entries_per_peer = self.prepare_peer_entries( - &new_entries, - &replication_data, - ctx.node_config.raft.replication.append_entries_max_entries_per_replication, - raft_log, - ); - - // ---------------------- - // Phase 4: Build Requests - // ---------------------- - let mut requests = Vec::with_capacity(replication_targets.len()); - for m in replication_targets { - requests.push(self.build_append_request( - raft_log, - m.id, - &mut entries_per_peer, - &replication_data, - )); - } - - // ---------------------- - // Phase 5: Replication - // ---------------------- - - // No peers: logs already written in Phase 2, return immediately - // No replication needed, quorum is automatically achieved (standalone node) - if replication_targets.is_empty() { - debug!( - "Standalone node (leader={}): logs persisted, quorum automatically achieved", - self.my_id - ); - return Ok(AppendResults { - commit_quorum_achieved: true, - peer_updates: HashMap::new(), - learner_progress: HashMap::new(), - }); - } - - // Multi-node cluster: perform replication to peers - let leader_current_term = state_snapshot.current_term; - let mut successes = 1; // Include leader itself - let mut peer_updates = HashMap::new(); - let mut learner_progress = HashMap::new(); - - let membership = ctx.membership(); - match ctx - .transport() - .send_append_requests( - requests, - &ctx.node_config.retry, - membership, - ctx.node_config.raft.rpc_compression.replication_response, - ) - .await - { - Ok(append_result) => { - for response in append_result.responses { - match response { - Ok(append_response) => { - // Skip responses from stale terms - if append_response.term < leader_current_term { - info!(%append_response.term, %leader_current_term, "append_response.term < leader_current_term"); - continue; - } - - match append_response.result { - Some(append_entries_response::Result::Success(success_result)) => { - // Only count successful responses from Voters - if voters.iter().any(|n| n.id == append_response.node_id) { - successes += 1; - } - - let update = self.handle_success_response( - append_response.node_id, - append_response.term, - success_result, - leader_current_term, - )?; - - // Record Learner progress - if learners.iter().any(|n| n.id == append_response.node_id) { - learner_progress - .insert(append_response.node_id, update.match_index); - } - - peer_updates.insert(append_response.node_id, update); - } - - Some(append_entries_response::Result::Conflict( - conflict_result, - )) => { - let current_next_index = replication_data - .peer_next_indices - .get(&append_response.node_id) - .copied() - .unwrap_or(1); - - let update = self.handle_conflict_response( - append_response.node_id, - conflict_result, - raft_log, - current_next_index, - )?; - - // Record Learner progress - if learners.iter().any(|n| n.id == append_response.node_id) { - learner_progress - .insert(append_response.node_id, update.match_index); - } - - peer_updates.insert(append_response.node_id, update); - } - - Some(append_entries_response::Result::HigherTerm(higher_term)) => { - // Only handle higher term if it's greater than current term - if higher_term > leader_current_term { - return Err( - ReplicationError::HigherTerm(higher_term).into() - ); - } - } - - None => { - error!("TODO: need to figure out the reason of this cluase"); - unreachable!(); - } - } - } - Err(e) => { - // Timeouts and network errors are logged but not added to peer_updates - warn!("Peer request failed: {:?}", e); - } - } - } - let peer_ids = append_result.peer_ids; - debug!( - "send_append_requests to: {:?} with succeed number = {}", - &peer_ids, successes - ); - - let total_voters = voters.len() + 1; // Leader + voter peers - let commit_quorum_achieved = is_majority(successes, total_voters); - Ok(AppendResults { - commit_quorum_achieved, - peer_updates, - learner_progress, - }) - } - Err(e) => return Err(e), - } - } - fn handle_success_response( &self, peer_id: u32, diff --git a/d-engine-core/src/state_machine_handler/default_state_machine_handler_test.rs b/d-engine-core/src/state_machine_handler/default_state_machine_handler_test.rs index 29e8bedd..3c124255 100644 --- a/d-engine-core/src/state_machine_handler/default_state_machine_handler_test.rs +++ b/d-engine-core/src/state_machine_handler/default_state_machine_handler_test.rs @@ -1224,16 +1224,7 @@ async fn get_dir_names(path: &Path) -> Vec { // } // // Initializing Shutdown Signal -// let mut replication_handler = MockReplicationCore::new(); -// replication_handler -// .expect_handle_raft_request_in_batch() -// .returning(|_, _, _, _| { -// Ok(AppendResults { -// commit_quorum_achieved: false, -// learner_progress: HashMap::new(), -// peer_updates: HashMap::new(), -// }) -// }); +// let replication_handler = MockReplicationCore::new(); // let mut election_handler = MockElectionCore::::new(); // election_handler // .expect_check_vote_request_is_legal() diff --git a/d-engine-proto/src/generated/d_engine.client.rs b/d-engine-proto/src/generated/d_engine.client.rs index dcb3be09..ac60088e 100644 --- a/d-engine-proto/src/generated/d_engine.client.rs +++ b/d-engine-proto/src/generated/d_engine.client.rs @@ -436,7 +436,7 @@ pub mod raft_client_service_client { /// Watch for committed cluster membership changes. /// /// Immediately pushes the current MembershipSnapshot on connect, then pushes - /// a new snapshot on every committed ConfChange (AddNode, Promote, Remove). + /// a new snapshot on every committed ConfChange (AddNode, BatchPromote, BatchRemove). /// All nodes (leader, follower, learner) emit the event after the entry commits. /// /// Stream lifecycle: @@ -529,7 +529,7 @@ pub mod raft_client_service_server { /// Watch for committed cluster membership changes. /// /// Immediately pushes the current MembershipSnapshot on connect, then pushes - /// a new snapshot on every committed ConfChange (AddNode, Promote, Remove). + /// a new snapshot on every committed ConfChange (AddNode, BatchPromote, BatchRemove). /// All nodes (leader, follower, learner) emit the event after the entry commits. /// /// Stream lifecycle: diff --git a/d-engine-server/src/node/node_test.rs b/d-engine-server/src/node/node_test.rs index fad52430..b31565b8 100644 --- a/d-engine-server/src/node/node_test.rs +++ b/d-engine-server/src/node/node_test.rs @@ -1,15 +1,12 @@ -use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use d_engine_core::AppendResults; use d_engine_core::Error; use d_engine_core::MockMembership; use d_engine_core::MockRaftLog; use d_engine_core::MockReplicationCore; use d_engine_core::MockTransport; use d_engine_core::MockTypeConfig; -use d_engine_core::PeerUpdate; use d_engine_core::RaftNodeConfig; use d_engine_core::RaftRole; use d_engine_core::learner_state::LearnerState; @@ -50,37 +47,9 @@ async fn test_run_sequence_with_mock_peers() { fn prepare_succeed_majority_confirmation() -> (MockRaftLog, MockReplicationCore) { // Initialize the mock object - let mut replication_handler = MockReplicationCore::::new(); + let replication_handler = MockReplicationCore::::new(); let mut raft_log = MockRaftLog::new(); - //Configure mock behavior - replication_handler - .expect_handle_raft_request_in_batch() - .returning(move |_, _, _, _, _| { - Ok(AppendResults { - commit_quorum_achieved: true, - learner_progress: HashMap::new(), - peer_updates: HashMap::from([ - ( - 2, - PeerUpdate { - match_index: Some(5), - next_index: 6, - success: true, - }, - ), - ( - 3, - PeerUpdate { - match_index: Some(5), - next_index: 6, - success: true, - }, - ), - ]), - }) - }); - raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| Some(5)); raft_log.expect_last_entry_id().return_const(1_u64); raft_log.expect_flush().return_once(|| Ok(())); @@ -570,17 +539,7 @@ mod bootstrap_strategy_tests { } fn prepare_mock_replication() -> MockReplicationCore { - let mut replication = MockReplicationCore::::new(); - replication - .expect_handle_raft_request_in_batch() - .returning(move |_, _, _, _, _| { - Ok(d_engine_core::AppendResults { - commit_quorum_achieved: true, - learner_progress: std::collections::HashMap::new(), - peer_updates: std::collections::HashMap::new(), - }) - }); - replication + MockReplicationCore::::new() } /// Test 1: Learner bootstrap - skip cluster ready check, join cluster