Skip to content

perf #345: remove dead unary AppendEntries path and stale scaffolding #367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JoshuaChi merged 1 commit into from
May 5, 2026
Merged

perf #345: remove dead unary AppendEntries path and stale scaffolding #367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 4 additions & 16 deletions d-engine-core/benches/leader_state_bench.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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(|_| {});
Expand Down
2 changes: 1 addition & 1 deletion d-engine-core/src/raft_role/leader_state_test/client_write_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
3 changes: 0 additions & 3 deletions d-engine-core/src/raft_role/leader_state_test/replication_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
50 changes: 0 additions & 50 deletions d-engine-core/src/replication/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -100,55 +99,6 @@ where
ctx: &crate::RaftContext<T>,
) -> Result<PrepareResult>;

/// 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<EntryPayload>,
state_snapshot: StateSnapshot,
leader_state_snapshot: LeaderStateSnapshot,
cluster_metadata: &crate::raft_role::ClusterMetadata,
ctx: &crate::RaftContext<T>,
) -> Result<AppendResults>;

/// Handles successful AppendEntries responses
///
/// Updates peer match/next indices according to:
Expand Down
213 changes: 0 additions & 213 deletions d-engine-core/src/replication/replication_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T>
where
Expand Down Expand Up @@ -132,212 +125,6 @@ where
})
}

async fn handle_raft_request_in_batch(
&self,
entry_payloads: Vec<EntryPayload>,
state_snapshot: StateSnapshot,
leader_state_snapshot: LeaderStateSnapshot,
cluster_metadata: &crate::raft_role::ClusterMetadata,
ctx: &RaftContext<T>,
) -> Result<AppendResults> {
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,
Expand Down
11 changes: 1 addition & 10 deletions d-engine-core/src/state_machine_handler/default_state_machine_handler_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1224,16 +1224,7 @@ async fn get_dir_names(path: &Path) -> Vec<String> {
// }

// // 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::<MockTypeConfig>::new();
// election_handler
// .expect_check_vote_request_is_legal()
Expand Down
Loading
Loading