d_engine_core/raft_role/

leader_state.rs

use std::collections::HashMap;
use std::collections::VecDeque;
use std::fmt::Debug;
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;

use d_engine_proto::client::ClientReadRequest;
use d_engine_proto::client::ClientResponse;
use d_engine_proto::client::ReadConsistencyPolicy as ClientReadConsistencyPolicy;
use d_engine_proto::common::AddNode;
use d_engine_proto::common::BatchPromote;
use d_engine_proto::common::BatchRemove;
use d_engine_proto::common::EntryPayload;
use d_engine_proto::common::LogId;
use d_engine_proto::common::NodeRole::Leader;
use d_engine_proto::common::NodeStatus;
use d_engine_proto::common::membership_change::Change;
use d_engine_proto::error::ErrorCode;
use d_engine_proto::server::cluster::ClusterConfUpdateResponse;
use d_engine_proto::server::cluster::JoinRequest;
use d_engine_proto::server::cluster::JoinResponse;
use d_engine_proto::server::cluster::LeaderDiscoveryResponse;
use d_engine_proto::server::cluster::NodeMeta;
use d_engine_proto::server::election::VoteResponse;
use d_engine_proto::server::election::VotedFor;
use d_engine_proto::server::replication::AppendEntriesResponse;
use d_engine_proto::server::storage::PurgeLogRequest;
use d_engine_proto::server::storage::PurgeLogResponse;
use d_engine_proto::server::storage::SnapshotChunk;
use d_engine_proto::server::storage::SnapshotMetadata;
use nanoid::nanoid;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::time::Instant;
use tokio::time::sleep;
use tokio::time::timeout;
use tonic::Status;
use tonic::async_trait;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::instrument;
use tracing::trace;
use tracing::warn;

use super::LeaderStateSnapshot;
use super::RaftRole;
use super::SharedState;
use super::StateSnapshot;
use super::candidate_state::CandidateState;
use super::role_state::RaftRoleState;
use crate::AppendResults;
use crate::BackgroundSnapshotTransfer;
use crate::BatchBuffer;
use crate::ConnectionType;
use crate::ConsensusError;
use crate::Error;
use crate::MaybeCloneOneshot;
use crate::MaybeCloneOneshotSender;
use crate::Membership;
use crate::MembershipError;
use crate::NetworkError;
use crate::PeerUpdate;
use crate::PurgeExecutor;
use crate::QuorumVerificationResult;
use crate::RaftContext;
use crate::RaftEvent;
use crate::RaftLog;
use crate::RaftNodeConfig;
use crate::RaftOneshot;
use crate::RaftRequestWithSignal;
use crate::ReadConsistencyPolicy as ServerReadConsistencyPolicy;
use crate::ReplicationConfig;
use crate::ReplicationCore;
use crate::ReplicationError;
use crate::ReplicationTimer;
use crate::Result;
use crate::RoleEvent;
use crate::SnapshotConfig;
use crate::StateMachine;
use crate::StateMachineHandler;
use crate::StateTransitionError;
use crate::SystemError;
use crate::Transport;
use crate::TypeConfig;
use crate::alias::MOF;
use crate::alias::ROF;
use crate::alias::SMHOF;
use crate::client_command_to_entry_payloads;
use crate::cluster::is_majority;
use crate::ensure_safe_join;
use crate::scoped_timer::ScopedTimer;
use crate::stream::create_production_snapshot_stream;
use crate::utils::cluster::error;

// Supporting data structures
#[derive(Debug, Clone)]
pub struct PendingPromotion {
    pub node_id: u32,
    pub ready_since: Instant,
}

impl PendingPromotion {
    pub fn new(
        node_id: u32,
        ready_since: Instant,
    ) -> Self {
        PendingPromotion {
            node_id,
            ready_since,
        }
    }
}

/// Cached cluster topology metadata for hot path optimization.
///
/// This metadata is immutable during leadership and only updated on membership changes.
/// Caching avoids repeated async calls to membership queries in hot paths like append_entries.
#[derive(Debug, Clone)]
pub struct ClusterMetadata {
    /// Single-voter cluster (quorum = 1, used for election and commit optimization)
    pub single_voter: bool,
    /// Total number of voters including self (updated on membership changes)
    pub total_voters: usize,
    /// Cached replication targets (voters + learners, excluding self)
    pub replication_targets: Vec<NodeMeta>,
}

/// Leader node's state in Raft consensus algorithm.
///
/// This structure maintains all state that should be persisted on leader crashes,
/// including replication progress tracking and log compaction management.
///
/// # Type Parameters
/// - `T`: Application-specific Raft type configuration
pub struct LeaderState<T: TypeConfig> {
    // -- Core Raft Leader State --
    /// Shared cluster state with lock protection
    pub shared_state: SharedState,

    /// === Volatile State ===
    /// For each server (node_id), index of the next log entry to send to that server
    ///
    /// Raft Paper: §5.3 Figure 2 (nextIndex)
    pub next_index: HashMap<u32, u64>,

    /// === Volatile State ===
    /// For each server (node_id), index of highest log entry known to be replicated
    ///
    /// Raft Paper: §5.3 Figure 2 (matchIndex)
    pub(super) match_index: HashMap<u32, u64>,

    /// === Volatile State ===
    /// Temporary storage for no-op entry log ID during leader initialization
    #[doc(hidden)]
    pub noop_log_id: Option<u64>,

    // -- Log Compaction & Purge --
    /// === Volatile State ===
    /// The upper bound (exclusive) of log entries scheduled for asynchronous physical deletion.
    ///
    /// This value is set immediately after a new snapshot is successfully created.
    /// It represents the next log position that will trigger compaction.
    ///
    /// The actual log purge is performed by a background task, which may be delayed
    /// due to resource constraints or retry mechanisms.
    pub scheduled_purge_upto: Option<LogId>,

    /// === Persistent State (MUST be on disk) ===
    /// The last log position that has been **physically removed** from stable storage.
    ///
    /// This value is atomically updated when:
    /// 1. A new snapshot is persisted (marking logs up to `last_included_index` as purgeable)
    /// 2. The background purge task completes successfully
    ///
    /// Raft safety invariant:
    /// Any log entry with index ≤ `last_purged_index` is guaranteed to be
    /// reflected in the latest snapshot.
    pub last_purged_index: Option<LogId>,

    /// === Volatile State ===
    /// Peer purge progress tracking for flow control
    ///
    /// Key: Peer node ID
    /// Value: Last confirmed purge index from peer
    pub peer_purge_progress: HashMap<u32, u64>,

    /// Record if there is on-going snapshot creation activity
    pub snapshot_in_progress: AtomicBool,

    // -- Request Processing --
    /// Batched proposal buffer for client requests
    ///
    /// Accumulates requests until either:
    /// 1. Batch reaches configured size limit
    /// 2. Explicit flush is triggered
    batch_buffer: Box<BatchBuffer<RaftRequestWithSignal>>,

    // -- Timing & Scheduling --
    /// Replication heartbeat timer manager
    ///
    /// Handles:
    /// - Heartbeat interval tracking
    /// - Election timeout prevention
    /// - Batch proposal flushing
    timer: Box<ReplicationTimer>,

    // -- Cluster Configuration --
    /// Cached Raft node configuration (shared reference)
    ///
    /// This includes:
    /// - Cluster membership
    /// - Timeout parameters
    /// - Performance tuning knobs
    pub(super) node_config: Arc<RaftNodeConfig>,

    /// Last time we checked for learners
    pub last_learner_check: Instant,

    // -- Stale Learner Handling --
    /// The next scheduled time to check for stale learners.
    ///
    /// This is used to implement periodic checking of learners that have been in the promotion
    /// queue for too long without making progress. The check interval is configurable via the
    /// node configuration (`node_config.promotion.stale_check_interval`).
    ///
    /// When the current time reaches or exceeds this instant, the leader will perform a scan
    /// of a subset of the pending promotions to detect stale learners. After the scan, the field
    /// is updated to `current_time + stale_check_interval`.
    ///
    /// Rationale: Avoiding frequent full scans improves batching efficiency and reduces CPU spikes
    /// in high-load environments (particularly crucial for RocketMQ-on-DLedger workflows).
    pub next_membership_maintenance_check: Instant,

    /// Queue of learners that have caught up and are pending promotion to voter.
    pub pending_promotions: VecDeque<PendingPromotion>,

    /// Lease timestamp for LeaseRead policy
    /// Tracks when leadership was last confirmed with quorum
    pub(super) lease_timestamp: AtomicU64,

    // -- Cluster Topology Cache --
    /// Cached cluster metadata (updated on membership changes)
    /// Avoids repeated async calls in hot paths
    pub(super) cluster_metadata: ClusterMetadata,

    /// Buffered read requests awaiting batch processing
    pub(super) read_buffer: Vec<(
        ClientReadRequest,
        MaybeCloneOneshotSender<std::result::Result<ClientResponse, Status>>,
    )>,

    /// Timestamp when first request entered buffer (for metrics/debugging)
    pub(super) read_buffer_start_time: Option<Instant>,

    // -- Type System Marker --
    /// Phantom data for type parameter anchoring
    _marker: PhantomData<T>,
}

#[async_trait]
impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
    type T = T;

    fn shared_state(&self) -> &SharedState {
        &self.shared_state
    }

    fn shared_state_mut(&mut self) -> &mut SharedState {
        &mut self.shared_state
    }

    ///Overwrite default behavior.
    /// As leader, I should not receive commit index,
    ///     which is lower than my current one
    #[tracing::instrument]
    fn update_commit_index(
        &mut self,
        new_commit_index: u64,
    ) -> Result<()> {
        if self.commit_index() < new_commit_index {
            debug!("update_commit_index to: {:?}", new_commit_index);
            self.shared_state.commit_index = new_commit_index;
        } else {
            warn!(
                "Illegal operation, might be a bug! I am Leader old_commit_index({}) >= new_commit_index:({})",
                self.commit_index(),
                new_commit_index
            )
        }
        Ok(())
    }

    /// As Leader should not vote any more
    fn voted_for(&self) -> Result<Option<VotedFor>> {
        self.shared_state().voted_for()
    }

    /// As Leader might also be able to vote ,
    ///     if new legal Leader found
    fn update_voted_for(
        &mut self,
        voted_for: VotedFor,
    ) -> Result<bool> {
        self.shared_state_mut().update_voted_for(voted_for)
    }

    fn next_index(
        &self,
        node_id: u32,
    ) -> Option<u64> {
        Some(if let Some(n) = self.next_index.get(&node_id) {
            *n
        } else {
            1
        })
    }

    fn update_next_index(
        &mut self,
        node_id: u32,
        new_next_id: u64,
    ) -> Result<()> {
        debug!("update_next_index({}) to {}", node_id, new_next_id);
        self.next_index.insert(node_id, new_next_id);
        Ok(())
    }

    fn update_match_index(
        &mut self,
        node_id: u32,
        new_match_id: u64,
    ) -> Result<()> {
        self.match_index.insert(node_id, new_match_id);
        Ok(())
    }

    fn match_index(
        &self,
        node_id: u32,
    ) -> Option<u64> {
        self.match_index.get(&node_id).copied()
    }

    fn init_peers_next_index_and_match_index(
        &mut self,
        last_entry_id: u64,
        peer_ids: Vec<u32>,
    ) -> Result<()> {
        for peer_id in peer_ids {
            debug!("init leader state for peer_id: {:?}", peer_id);
            let new_next_id = last_entry_id + 1;
            self.update_next_index(peer_id, new_next_id)?;
            self.update_match_index(peer_id, 0)?;
        }
        Ok(())
    }

    /// Track no-op entry index for linearizable read optimization.
    /// Called after no-op entry is committed on leadership transition.
    fn on_noop_committed(
        &mut self,
        ctx: &RaftContext<Self::T>,
    ) -> Result<()> {
        let noop_index = ctx.raft_log().last_entry_id();
        self.noop_log_id = Some(noop_index);
        debug!("Tracked noop_log_id: {}", noop_index);
        Ok(())
    }
    fn noop_log_id(&self) -> Result<Option<u64>> {
        Ok(self.noop_log_id)
    }

    /// Verifies leadership status using persistent retry until timeout.
    ///
    /// This function is designed for critical operations like configuration changes
    /// that must eventually succeed. It implements:
    ///   - Infinite retries with exponential backoff
    ///   - Jitter randomization to prevent synchronization
    ///   - Termination only on success, leadership loss, or global timeout
    ///
    /// # Parameters
    /// - `payloads`: Log entries to verify
    /// - `bypass_queue`: Whether to skip request queues for direct transmission
    /// - `ctx`: Raft execution context
    /// - `role_tx`: Channel for role transition events
    ///
    /// # Returns
    /// - `Ok(true)`: Quorum verification succeeded
    /// - `Ok(false)`: Leadership definitively lost during verification
    /// - `Err(_)`: Global timeout exceeded or critical failure occurred
    async fn verify_leadership_persistent(
        &mut self,
        payloads: Vec<EntryPayload>,
        bypass_queue: bool,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<bool> {
        let initial_delay =
            Duration::from_millis(ctx.node_config.retry.internal_quorum.base_delay_ms);
        let max_delay = Duration::from_millis(ctx.node_config.retry.internal_quorum.max_delay_ms);
        let global_timeout = ctx.node_config.raft.membership.verify_leadership_persistent_timeout;

        let mut current_delay = initial_delay;
        let start_time = Instant::now();

        loop {
            match self.verify_internal_quorum(payloads.clone(), bypass_queue, ctx, role_tx).await {
                Ok(QuorumVerificationResult::Success) => {
                    // Update lease timestamp after successful quorum verification.
                    // This allows subsequent LeaseRead requests to benefit from this verification,
                    // avoiding redundant quorum checks and improving read performance.
                    self.update_lease_timestamp();
                    return Ok(true);
                }
                Ok(QuorumVerificationResult::LeadershipLost) => return Ok(false),
                Ok(QuorumVerificationResult::RetryRequired) => {
                    // Check global timeout before retrying
                    if start_time.elapsed() > global_timeout {
                        return Err(NetworkError::GlobalTimeout(
                            "Leadership verification timed out".to_string(),
                        )
                        .into());
                    }

                    current_delay =
                        current_delay.checked_mul(2).unwrap_or(max_delay).min(max_delay);
                    let jitter = Duration::from_millis(rand::random::<u64>() % 500);
                    sleep(current_delay + jitter).await;
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// Note: This method acts as the centerialized internal Raft client
    ///
    /// Check leadership quorum verification Immidiatelly
    ///
    /// - Bypasses all queues with direct RPC transmission
    /// - Enforces synchronous quorum validation
    /// - Guarantees real-time network visibility
    ///
    /// Scenario handling summary:
    ///
    /// 1. Quorum achieved:
    ///    - Return: `Ok(QuorumVerificationResult::Success)`
    ///
    /// 2. Quorum NOT achieved (verifiable):
    ///    - Return: `Ok(QuorumVerificationResult::RetryRequired)`
    ///
    /// 3. Quorum NOT achieved (non-verifiable):
    ///    - Return: `Ok(QuorumVerificationResult::LeadershipLost)`
    ///
    /// 4. Partial timeouts:
    ///    - Return: `Ok(QuorumVerificationResult::RetryRequired)`
    ///
    /// 5. All timeouts:
    ///    - Return: `Ok(QuorumVerificationResult::RetryRequired)`
    ///
    /// 6. Higher term detected:
    ///    - Return: `Err(HigherTerm)`
    ///
    /// 7. Critical failure (e.g., system or logic error):
    ///    - Return: original error
    /// Override: Initialize cluster metadata after becoming leader.
    /// Must be called once after leader election succeeds.
    async fn init_cluster_metadata(
        &mut self,
        membership: &Arc<T::M>,
    ) -> Result<()> {
        // Calculate total voter count including self as leader
        let voters = membership.voters().await;
        let total_voters = voters.len() + 1; // +1 for leader (self)

        // Get all replication targets (voters + learners, excluding self)
        let replication_targets = membership.replication_peers().await;

        // Single-voter cluster: only this node is a voter (quorum = 1)
        let single_voter = total_voters == 1;

        self.cluster_metadata = ClusterMetadata {
            single_voter,
            total_voters,
            replication_targets: replication_targets.clone(),
        };

        debug!(
            "Initialized cluster metadata: single_voter={}, total_voters={}, replication_targets={}",
            single_voter,
            total_voters,
            replication_targets.len()
        );
        Ok(())
    }

    async fn verify_internal_quorum(
        &mut self,
        payloads: Vec<EntryPayload>,
        bypass_queue: bool,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<QuorumVerificationResult> {
        let (resp_tx, resp_rx) = MaybeCloneOneshot::new();

        self.process_raft_request(
            RaftRequestWithSignal {
                id: nanoid!(),
                payloads,
                sender: resp_tx,
            },
            ctx,
            bypass_queue,
            role_tx,
        )
        .await?;

        // Wait for response with timeout
        let timeout_duration =
            Duration::from_millis(self.node_config.raft.general_raft_timeout_duration_in_ms);
        match timeout(timeout_duration, resp_rx).await {
            // Case 1: Response received successfully and verification passed
            Ok(Ok(Ok(response))) => {
                debug!("Leadership check response: {:?}", response);

                // Handle different response cases
                Ok(if response.is_write_success() {
                    QuorumVerificationResult::Success
                } else if response.is_retry_required() {
                    // Verifiable quorum failure
                    QuorumVerificationResult::RetryRequired
                } else {
                    // Non-verifiable failure or explicit rejection
                    QuorumVerificationResult::LeadershipLost
                })
            }

            // Case 2: Received explicit rejection status
            Ok(Ok(Err(status))) => {
                warn!("Leadership rejected by follower: {status:?}");
                Ok(QuorumVerificationResult::LeadershipLost)
            }

            // Case 3: Channel communication failure (unrecoverable error)
            Ok(Err(e)) => {
                error!("Channel error during leadership check: {:?}", e);
                Err(NetworkError::SingalReceiveFailed(e.to_string()).into())
            }

            // Case 4: Waiting for response timeout
            Err(_) => {
                warn!("Leadership check timed out after {:?}", timeout_duration);
                Err(NetworkError::Timeout {
                    node_id: self.node_id(),
                    duration: timeout_duration,
                }
                .into())
            }
        }
    }

    fn is_leader(&self) -> bool {
        true
    }

    fn become_leader(&self) -> Result<RaftRole<T>> {
        warn!("I am leader already");

        Err(StateTransitionError::InvalidTransition.into())
    }

    fn become_candidate(&self) -> Result<RaftRole<T>> {
        error!("Leader can not become Candidate");

        Err(StateTransitionError::InvalidTransition.into())
    }

    fn become_follower(&self) -> Result<RaftRole<T>> {
        info!(
            "Node {} term {} transitioning to Follower",
            self.node_id(),
            self.current_term(),
        );
        println!(
            "[Node {}] Leader → Follower (term {})",
            self.node_id(),
            self.current_term()
        );
        Ok(RaftRole::Follower(Box::new(self.into())))
    }

    fn become_learner(&self) -> Result<RaftRole<T>> {
        error!("Leader can not become Learner");

        Err(StateTransitionError::InvalidTransition.into())
    }

    fn is_timer_expired(&self) -> bool {
        self.timer.is_expired()
    }

    /// Raft starts, we will check if we need reset all timer
    fn reset_timer(&mut self) {
        self.timer.reset_batch();
        self.timer.reset_replication();
    }

    fn next_deadline(&self) -> Instant {
        self.timer.next_deadline()
    }

    /// Trigger heartbeat now
    async fn tick(
        &mut self,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        _raft_tx: &mpsc::Sender<RaftEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        let now = Instant::now();
        // Keep syncing leader_id (hot-path: ~5ns atomic store)
        self.shared_state().set_current_leader(self.node_id());

        // 1. Clear expired learners
        if let Err(e) = self.run_periodic_maintenance(role_tx, ctx).await {
            error!("Failed to run periodic maintenance: {}", e);
        }

        // 3. Batch trigger check (should be prioritized before heartbeat check)
        if now >= self.timer.batch_deadline() {
            self.timer.reset_batch();

            if self.batch_buffer.should_flush() {
                debug!(?now, "tick::reset_batch batch timer");
                self.timer.reset_replication();

                // Take out the batched messages and send them immediately
                // Do not move batch out of this block
                let batch = self.batch_buffer.take();
                self.process_batch(batch, role_tx, ctx).await?;
            }
        }

        // 4. Heartbeat trigger check
        // Send heartbeat if the replication timer expires
        if now >= self.timer.replication_deadline() {
            debug!(?now, "tick::reset_replication timer");
            self.timer.reset_replication();

            // Do not move batch out of this block
            let batch = self.batch_buffer.take();
            self.process_batch(batch, role_tx, ctx).await?;
        }

        Ok(())
    }

    fn drain_read_buffer(&mut self) -> Result<()> {
        if !self.read_buffer.is_empty() {
            warn!(
                "Read batch: draining {} requests due to role change",
                self.read_buffer.len()
            );
            for (_, sender) in std::mem::take(&mut self.read_buffer) {
                let _ = sender.send(Err(tonic::Status::unavailable("Leader stepped down")));
            }
            self.read_buffer_start_time = None;
        }

        Ok(())
    }

    async fn handle_raft_event(
        &mut self,
        raft_event: RaftEvent,
        ctx: &RaftContext<T>,
        role_tx: mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        let my_id = self.shared_state.node_id;
        let my_term = self.current_term();

        match raft_event {
            // Leader receives RequestVote(term=X, candidate=Y)
            // 1. If X > currentTerm:
            // - Leader → Follower, currentTerm = X
            // - Replay event
            // 2. Else:
            // - Reply with VoteGranted=false, currentTerm=currentTerm
            RaftEvent::ReceiveVoteRequest(vote_request, sender) => {
                debug!(
                    "handle_raft_event::RaftEvent::ReceiveVoteRequest: {:?}",
                    &vote_request
                );

                let my_term = self.current_term();
                if my_term < vote_request.term {
                    self.update_current_term(vote_request.term);
                    // Step down as Follower
                    self.send_become_follower_event(None, &role_tx)?;

                    info!("Leader will not process Vote request, it should let Follower do it.");
                    send_replay_raft_event(
                        &role_tx,
                        RaftEvent::ReceiveVoteRequest(vote_request, sender),
                    )?;
                } else {
                    let last_log_id =
                        ctx.raft_log().last_log_id().unwrap_or(LogId { index: 0, term: 0 });
                    let response = VoteResponse {
                        term: my_term,
                        vote_granted: false,
                        last_log_index: last_log_id.index,
                        last_log_term: last_log_id.term,
                    };
                    sender.send(Ok(response)).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;
                }
            }

            RaftEvent::ClusterConf(_metadata_request, sender) => {
                let cluster_conf = ctx
                    .membership()
                    .retrieve_cluster_membership_config(self.shared_state().current_leader())
                    .await;
                debug!("Leader receive ClusterConf: {:?}", &cluster_conf);

                sender.send(Ok(cluster_conf)).map_err(|e| {
                    let error_str = format!("{e:?}");
                    error!("Failed to send: {}", error_str);
                    NetworkError::SingalSendFailed(error_str)
                })?;
            }

            RaftEvent::ClusterConfUpdate(cluste_conf_change_request, sender) => {
                let current_conf_version = ctx.membership().get_cluster_conf_version().await;
                debug!(%current_conf_version, ?cluste_conf_change_request,
                    "handle_raft_event::RaftEvent::ClusterConfUpdate",
                );

                // Reject the fake Leader append entries request
                if my_term >= cluste_conf_change_request.term {
                    let response = ClusterConfUpdateResponse::higher_term(
                        my_id,
                        my_term,
                        current_conf_version,
                    );

                    sender.send(Ok(response)).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;
                } else {
                    // Step down as Follower as new Leader found
                    info!(
                        "my({}) term < request one, now I will step down to Follower",
                        my_id
                    );
                    //TODO: if there is a bug?  self.update_current_term(vote_request.term);
                    self.send_become_follower_event(Some(cluste_conf_change_request.id), &role_tx)?;

                    info!(
                        "Leader will not process append_entries_request, it should let Follower do it."
                    );
                    send_replay_raft_event(
                        &role_tx,
                        RaftEvent::ClusterConfUpdate(cluste_conf_change_request, sender),
                    )?;
                }
            }

            RaftEvent::AppendEntries(append_entries_request, sender) => {
                debug!(
                    "handle_raft_event::RaftEvent::AppendEntries: {:?}",
                    &append_entries_request
                );

                // Reject the fake Leader append entries request
                if my_term >= append_entries_request.term {
                    let response = AppendEntriesResponse::higher_term(my_id, my_term);

                    sender.send(Ok(response)).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;
                } else {
                    // Step down as Follower as new Leader found
                    info!(
                        "my({}) term < request one, now I will step down to Follower",
                        my_id
                    );
                    //TODO: if there is a bug?  self.update_current_term(vote_request.term);
                    self.send_become_follower_event(
                        Some(append_entries_request.leader_id),
                        &role_tx,
                    )?;

                    info!(
                        "Leader will not process append_entries_request, it should let Follower do it."
                    );
                    send_replay_raft_event(
                        &role_tx,
                        RaftEvent::AppendEntries(append_entries_request, sender),
                    )?;
                }
            }

            RaftEvent::ClientPropose(client_write_request, sender) => {
                if let Err(e) = self
                    .process_raft_request(
                        RaftRequestWithSignal {
                            id: nanoid!(),
                            payloads: client_command_to_entry_payloads(
                                client_write_request.commands,
                            ),
                            sender,
                        },
                        ctx,
                        false,
                        &role_tx,
                    )
                    .await
                {
                    error("Leader::process_raft_request", &e);
                    return Err(e);
                }
            }

            RaftEvent::ClientReadRequest(client_read_request, sender) => {
                let _timer = ScopedTimer::new("leader_linear_read");
                debug!(
                    "Leader::ClientReadRequest client_read_request:{:?}",
                    &client_read_request
                );

                let keys = client_read_request.keys.clone();
                let response: std::result::Result<ClientResponse, tonic::Status> = {
                    let read_operation =
                        || -> std::result::Result<ClientResponse, tonic::Status> {
                            let results = ctx
                                .handlers
                                .state_machine_handler
                                .read_from_state_machine(keys)
                                .unwrap_or_default();
                            debug!("handle_client_read results: {:?}", results);
                            Ok(ClientResponse::read_results(results))
                        };

                    // Determine effective consistency policy using server configuration
                    let effective_policy = if client_read_request.has_consistency_policy() {
                        // Client explicitly specified policy
                        if ctx.node_config().raft.read_consistency.allow_client_override {
                            match client_read_request.consistency_policy() {
                                ClientReadConsistencyPolicy::LeaseRead => {
                                    ServerReadConsistencyPolicy::LeaseRead
                                }
                                ClientReadConsistencyPolicy::LinearizableRead => {
                                    ServerReadConsistencyPolicy::LinearizableRead
                                }
                                ClientReadConsistencyPolicy::EventualConsistency => {
                                    ServerReadConsistencyPolicy::EventualConsistency
                                }
                            }
                        } else {
                            // Client override not allowed - use server default
                            ctx.node_config().raft.read_consistency.default_policy.clone()
                        }
                    } else {
                        // No client policy specified - use server default
                        ctx.node_config().raft.read_consistency.default_policy.clone()
                    };

                    // Apply consistency policy
                    match effective_policy {
                        ServerReadConsistencyPolicy::LinearizableRead => {
                            // 1. Add to buffer
                            self.read_buffer.push((client_read_request, sender));

                            // 2. First request: spawn timeout task
                            if self.read_buffer.len() == 1 {
                                self.read_buffer_start_time = Some(Instant::now());

                                let role_tx = role_tx.clone(); // Use role_tx (already available in signature)
                                let timeout_ms = self
                                    .node_config
                                    .raft
                                    .read_consistency
                                    .read_batching
                                    .time_threshold_ms;

                                debug!(
                                    "Read batch: first request, spawning {}ms timeout",
                                    timeout_ms
                                );

                                tokio::spawn(async move {
                                    tokio::time::sleep(Duration::from_millis(timeout_ms)).await;
                                    debug!(
                                        "Read batch: timeout expired, sending flush signal via ReprocessEvent"
                                    );
                                    // Use RoleEvent::ReprocessEvent to re-inject FlushReadBuffer into event loop
                                    let _ = role_tx.send(RoleEvent::ReprocessEvent(Box::new(
                                        RaftEvent::FlushReadBuffer,
                                    )));
                                });
                            }

                            // 3. Size threshold: immediate flush
                            let size_threshold =
                                self.node_config.raft.read_consistency.read_batching.size_threshold;
                            if self.read_buffer.len() >= size_threshold {
                                debug!(
                                    "Read batch: size threshold reached ({}), flushing",
                                    size_threshold
                                );
                                self.process_linearizable_read_batch(ctx, &role_tx).await?;
                            }
                            // Early return: response will be sent via process_linearizable_read_batch
                            return Ok(());
                        }
                        ServerReadConsistencyPolicy::LeaseRead => {
                            // Lease-based read: verify only if lease expired
                            if !self.is_lease_valid(ctx) {
                                // Lease expired, refresh with quorum verification
                                self.verify_leadership_and_refresh_lease(ctx, &role_tx).await?;
                            }
                            // Lease is valid, serve read locally
                            read_operation()
                        }
                        ServerReadConsistencyPolicy::EventualConsistency => {
                            // Eventual consistency: serve immediately without verification
                            // Leader can always serve eventually consistent reads
                            debug!("EventualConsistency: serving local read without verification");
                            read_operation()
                        }
                    }
                };

                debug!(
                    "Leader::ClientReadRequest is going to response: {:?}",
                    &response
                );
                sender.send(response).map_err(|e| {
                    let error_str = format!("{e:?}");
                    error!("Failed to send: {}", error_str);
                    NetworkError::SingalSendFailed(error_str)
                })?;
            }

            RaftEvent::FlushReadBuffer => {
                if !self.read_buffer.is_empty() {
                    debug!(
                        "Read batch: flushing {} pending reads (timeout)",
                        self.read_buffer.len()
                    );
                    self.process_linearizable_read_batch(ctx, &role_tx).await?;
                } else {
                    debug!("Read batch: flush signal received but buffer empty (already flushed)");
                }
            }

            RaftEvent::InstallSnapshotChunk(_streaming, sender) => {
                sender
                    .send(Err(Status::permission_denied("Not Follower or Learner. ")))
                    .map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;

                return Err(ConsensusError::RoleViolation {
                    current_role: "Leader",
                    required_role: "Follower or Learner",
                    context: format!(
                        "Leader node {} receives RaftEvent::InstallSnapshotChunk",
                        ctx.node_id
                    ),
                }
                .into());
            }

            RaftEvent::RaftLogCleanUp(_purchase_log_request, sender) => {
                sender
                    .send(Err(Status::permission_denied(
                        "Leader should not receive RaftLogCleanUp event.",
                    )))
                    .map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;

                return Err(ConsensusError::RoleViolation {
                    current_role: "Leader",
                    required_role: "None Leader",
                    context: format!(
                        "Leader node {} receives RaftEvent::RaftLogCleanUp",
                        ctx.node_id
                    ),
                }
                .into());
            }

            RaftEvent::CreateSnapshotEvent => {
                // Prevent duplicate snapshot creation
                if self.snapshot_in_progress.load(std::sync::atomic::Ordering::Acquire) {
                    info!("Snapshot creation already in progress. Skipping duplicate request.");
                    return Ok(());
                }

                self.snapshot_in_progress.store(true, std::sync::atomic::Ordering::Release);
                let state_machine_handler = ctx.state_machine_handler().clone();

                // Use spawn to perform snapshot creation in the background
                tokio::spawn(async move {
                    let result = state_machine_handler.create_snapshot().await;
                    info!("SnapshotCreated event will be processed in another event thread");
                    if let Err(e) =
                        send_replay_raft_event(&role_tx, RaftEvent::SnapshotCreated(result))
                    {
                        error!("Failed to send snapshot creation result: {}", e);
                    }
                });
            }

            RaftEvent::SnapshotCreated(result) => {
                self.snapshot_in_progress.store(false, Ordering::SeqCst);
                let my_id = self.shared_state.node_id;
                let my_term = self.current_term();

                match result {
                    Err(e) => {
                        error!(%e, "State machine snapshot creation failed");
                    }
                    Ok((
                        SnapshotMetadata {
                            last_included: last_included_option,
                            checksum,
                        },
                        _final_path,
                    )) => {
                        info!("Initiating log purge after snapshot creation");

                        if let Some(last_included) = last_included_option {
                            // ----------------------
                            // Phase 1: Schedule log purge if possible
                            // ----------------------
                            trace!("Phase 1: Schedule log purge if possible");
                            if self.can_purge_logs(self.last_purged_index, last_included) {
                                trace!(?last_included, "Phase 1: Scheduling log purge");
                                self.scheduled_purge_upto(last_included);
                            }

                            // ----------------------
                            // Phase 2: Send Purge request (multi-node only)
                            // ----------------------
                            // Check replication_targets to support:
                            // 1. True single-node (no peers) - skip this phase
                            // 2. Single-voter + Learner - send to Learner
                            // 3. Multi-voter cluster - send to all peers
                            if !self.cluster_metadata.replication_targets.is_empty() {
                                trace!("Phase 2: Send Purge request to replication targets");

                                // Defensive check: verify cluster_metadata is initialized
                                // This should never happen in production (init_cluster_metadata is called
                                // automatically after leader election), but if it does, it indicates a
                                // critical bug in state management
                                if self.cluster_metadata.total_voters == 0 {
                                    error!(
                                        "BUG: cluster_metadata not initialized! Leader must call init_cluster_metadata() after election"
                                    );
                                    return Err(
                                        MembershipError::ClusterMetadataNotInitialized.into()
                                    );
                                }

                                let membership = ctx.membership();
                                let transport = ctx.transport();
                                match transport
                                    .send_purge_requests(
                                        PurgeLogRequest {
                                            term: my_term,
                                            leader_id: my_id,
                                            last_included: Some(last_included),
                                            snapshot_checksum: checksum.clone(),
                                            leader_commit: self.commit_index(),
                                        },
                                        &self.node_config.retry,
                                        membership,
                                    )
                                    .await
                                {
                                    Ok(result) => {
                                        info!(?result, "receive PurgeLogResult");
                                        // Process peer purge responses and check for higher term
                                        // If a peer reports a higher term, this leader must step down
                                        // immediately and skip local purge execution (Phase 3)
                                        let should_step_down =
                                            self.peer_purge_progress(result, &role_tx)?;
                                        if should_step_down {
                                            return Ok(()); // Early return, skip Phase 3
                                        }
                                    }
                                    Err(e) => {
                                        error!(?e, "RaftEvent::CreateSnapshotEvent");
                                        return Err(e);
                                    }
                                }
                            } else {
                                debug!(
                                    "Standalone node (leader={}): no replication targets, skip PurgeRequest",
                                    my_id
                                );
                            }

                            // ----------------------
                            // Phase 3: Execute local purge (all nodes)
                            // ----------------------
                            trace!("Phase 3: Execute scheduled purge task");
                            debug!(?last_included, "Execute scheduled purge task");
                            if let Some(scheduled) = self.scheduled_purge_upto {
                                let purge_executor = ctx.purge_executor();
                                match purge_executor.execute_purge(scheduled).await {
                                    Ok(_) => {
                                        if let Err(e) = send_replay_raft_event(
                                            &role_tx,
                                            RaftEvent::LogPurgeCompleted(scheduled),
                                        ) {
                                            error!(%e, "Failed to notify purge completion");
                                        }
                                    }
                                    Err(e) => {
                                        error!(?e, ?scheduled, "Log purge execution failed");
                                    }
                                }
                            }
                        }
                    }
                }
            }

            RaftEvent::LogPurgeCompleted(purged_id) => {
                // Ensure we don't regress the purge index
                if self.last_purged_index.map_or(true, |current| purged_id.index > current.index) {
                    debug!(
                        ?purged_id,
                        "Updating last purged index after successful execution"
                    );
                    self.last_purged_index = Some(purged_id);
                } else {
                    warn!(
                        ?purged_id,
                        ?self.last_purged_index,
                        "Received outdated purge completion, ignoring"
                    );
                }
            }

            RaftEvent::JoinCluster(join_request, sender) => {
                debug!(?join_request, "Leader::RaftEvent::JoinCluster");
                self.handle_join_cluster(join_request, sender, ctx, &role_tx).await?;
            }

            RaftEvent::DiscoverLeader(request, sender) => {
                debug!(?request, "Leader::RaftEvent::DiscoverLeader");

                if let Some(meta) = ctx.membership().retrieve_node_meta(my_id).await {
                    let response = LeaderDiscoveryResponse {
                        leader_id: my_id,
                        leader_address: meta.address,
                        term: my_term,
                    };
                    sender.send(Ok(response)).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Failed to send: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;
                    return Ok(());
                } else {
                    let msg = "Leader can not find its address? It must be a bug.";
                    error!("{}", msg);
                    panic!("{}", msg);
                }
            }
            RaftEvent::StreamSnapshot(request, sender) => {
                debug!("Leader::RaftEvent::StreamSnapshot");

                // Get the latest snapshot metadata
                if let Some(metadata) = ctx.state_machine().snapshot_metadata() {
                    // Create response channel
                    let (response_tx, response_rx) =
                        mpsc::channel::<std::result::Result<Arc<SnapshotChunk>, Status>>(32);
                    // Convert to properly encoded tonic stream
                    let size = 1024 * 1024 * 1024; // 1GB max message size
                    let response_stream = create_production_snapshot_stream(response_rx, size);
                    // Immediately respond with the stream
                    sender.send(Ok(response_stream)).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Stream response failed: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;

                    // Spawn background transfer task
                    let state_machine_handler = ctx.state_machine_handler().clone();
                    let config = ctx.node_config.raft.snapshot.clone();
                    // Load snapshot data stream
                    let data_stream =
                        state_machine_handler.load_snapshot_data(metadata.clone()).await?;

                    tokio::spawn(async move {
                        if let Err(e) = BackgroundSnapshotTransfer::<T>::run_pull_transfer(
                            request,
                            response_tx,
                            data_stream,
                            config,
                        )
                        .await
                        {
                            error!("StreamSnapshot failed: {:?}", e);
                        }
                    });
                } else {
                    warn!("No snapshot available for streaming");
                    sender.send(Err(Status::not_found("Snapshot not found"))).map_err(|e| {
                        let error_str = format!("{e:?}");
                        error!("Stream response failed: {}", error_str);
                        NetworkError::SingalSendFailed(error_str)
                    })?;
                }
            }
            RaftEvent::TriggerSnapshotPush { peer_id } => {
                if let Some(lastest_snapshot_metadata) = ctx.state_machine().snapshot_metadata() {
                    Self::trigger_background_snapshot(
                        peer_id,
                        lastest_snapshot_metadata,
                        ctx.state_machine_handler().clone(),
                        ctx.membership(),
                        ctx.node_config.raft.snapshot.clone(),
                    )
                    .await?;
                }
            }

            RaftEvent::PromoteReadyLearners => {
                // SAFETY: Called from main event loop, no reentrancy issues
                info!(
                    "[Leader {}] ⚡ PromoteReadyLearners event received, pending_promotions: {:?}",
                    self.node_id(),
                    self.pending_promotions.iter().map(|p| p.node_id).collect::<Vec<_>>()
                );
                self.process_pending_promotions(ctx, &role_tx).await?;
            }

            RaftEvent::MembershipApplied => {
                // Save old membership for comparison
                let old_replication_targets = self.cluster_metadata.replication_targets.clone();

                // Refresh cluster metadata cache after membership change is applied
                debug!("Refreshing cluster metadata cache after membership change");
                if let Err(e) = self.update_cluster_metadata(&ctx.membership()).await {
                    warn!("Failed to update cluster metadata: {:?}", e);
                }

                // CRITICAL FIX #218: Initialize replication state for newly added peers
                // Per Raft protocol: When new members join, Leader must initialize their next_index
                let newly_added: Vec<u32> = self
                    .cluster_metadata
                    .replication_targets
                    .iter()
                    .filter(|new_peer| {
                        !old_replication_targets.iter().any(|old_peer| old_peer.id == new_peer.id)
                    })
                    .map(|peer| peer.id)
                    .collect();

                if !newly_added.is_empty() {
                    debug!(
                        "Initializing replication state for {} new peer(s): {:?}",
                        newly_added.len(),
                        newly_added
                    );
                    let last_entry_id = ctx.raft_log().last_entry_id();
                    if let Err(e) =
                        self.init_peers_next_index_and_match_index(last_entry_id, newly_added)
                    {
                        warn!("Failed to initialize next_index for new peers: {:?}", e);
                        // Non-fatal: next_index will use default value of 1,
                        // replication will still work but may be less efficient
                    }
                }
            }

            RaftEvent::StepDownSelfRemoved => {
                // Only Leader can propose configuration changes and remove itself
                // Per Raft protocol: Leader steps down immediately after self-removal
                warn!(
                    "[Leader-{}] Removed from cluster membership, stepping down to Follower",
                    self.node_id()
                );
                role_tx.send(RoleEvent::BecomeFollower(None)).map_err(|e| {
                    error!(
                        "[Leader-{}] Failed to send BecomeFollower after self-removal: {:?}",
                        self.node_id(),
                        e
                    );
                    NetworkError::SingalSendFailed(format!(
                        "BecomeFollower after self-removal: {e:?}"
                    ))
                })?;
                return Ok(());
            }
        }

        Ok(())
    }
}

impl<T: TypeConfig> LeaderState<T> {
    /// Initialize cluster metadata after becoming leader.
    /// Update cluster metadata when membership changes.
    pub async fn update_cluster_metadata(
        &mut self,
        membership: &Arc<T::M>,
    ) -> Result<()> {
        // Calculate total voter count including self as leader
        let voters = membership.voters().await;
        let total_voters = voters.len() + 1; // +1 for leader (self)

        // Get all replication targets (voters + learners, excluding self)
        let replication_targets = membership.replication_peers().await;

        // Single-voter cluster: only this node is a voter (quorum = 1)
        let single_voter = total_voters == 1;

        self.cluster_metadata = ClusterMetadata {
            single_voter,
            total_voters,
            replication_targets: replication_targets.clone(),
        };

        debug!(
            "Updated cluster metadata: single_voter={}, total_voters={}, replication_targets={}",
            single_voter,
            total_voters,
            replication_targets.len()
        );
        Ok(())
    }

    /// The fun will retrieve current state snapshot
    pub fn state_snapshot(&self) -> StateSnapshot {
        StateSnapshot {
            current_term: self.current_term(),
            voted_for: None,
            commit_index: self.commit_index(),
            role: Leader as i32,
        }
    }

    /// The fun will retrieve current Leader state snapshot
    #[tracing::instrument]
    pub fn leader_state_snapshot(&self) -> LeaderStateSnapshot {
        LeaderStateSnapshot {
            next_index: self.next_index.clone(),
            match_index: self.match_index.clone(),
            noop_log_id: self.noop_log_id,
        }
    }

    /// # Params
    /// - `execute_now`: should this propose been executed immediatelly. e.g.
    ///   enforce_quorum_consensus expected to be executed immediatelly
    pub async fn process_raft_request(
        &mut self,
        raft_request_with_signal: RaftRequestWithSignal,
        ctx: &RaftContext<T>,
        execute_now: bool,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        debug!(
            "Leader::process_raft_request, request_id: {}",
            raft_request_with_signal.id
        );

        let push_result = self.batch_buffer.push(raft_request_with_signal);
        // only buffer exceeds the max, the size will return
        if execute_now || push_result.is_some() {
            let batch = self.batch_buffer.take();

            trace!(
                "replication_handler.handle_raft_request_in_batch: batch size:{:?}",
                batch.len()
            );

            self.process_batch(batch, role_tx, ctx).await?;
        }

        Ok(())
    }

    /// Scenario handling summary:
    ///
    /// 1. Quorum achieved:
    ///    - Client response: `write_success()`
    ///    - State update: update peer indexes and commit index
    ///    - Return: `Ok(())`
    ///
    /// 2. Quorum NOT achieved (verifiable):
    ///    - Client response: `RetryRequired`
    ///    - State update: update peer indexes
    ///    - Return: `Ok(())`
    ///
    /// 3. Quorum NOT achieved (non-verifiable):
    ///    - Client response: `ProposeFailed`
    ///    - State update: update peer indexes
    ///    - Return: `Ok(())`
    ///
    /// 4. Partial timeouts:
    ///    - Client response: `ProposeFailed`
    ///    - State update: update only the peer indexes that responded
    ///    - Return: `Ok(())`
    ///
    /// 5. All timeouts:
    ///    - Client response: `ProposeFailed`
    ///    - State update: no update
    ///    - Return: `Ok(())`
    ///
    /// 6. Higher term detected:
    ///    - Client response: `TermOutdated`
    ///    - State update: update term and convert to follower
    ///    - Return: `Err(HigherTerm)`
    ///
    /// 7. Critical failure (e.g., system or logic error):
    ///    - Client response: `ProposeFailed`
    ///    - State update: none
    ///    - Return: original error
    pub async fn process_batch(
        &mut self,
        batch: VecDeque<RaftRequestWithSignal>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        // 1. Prepare batch data
        let entry_payloads: Vec<EntryPayload> =
            batch.iter().flat_map(|req| &req.payloads).cloned().collect();
        if !entry_payloads.is_empty() {
            trace!(?entry_payloads, "[Node-{} process_batch..", ctx.node_id);
        }

        // 2. Execute the copy
        let cluster_size = self.cluster_metadata.total_voters;
        trace!(%cluster_size);

        let result = ctx
            .replication_handler()
            .handle_raft_request_in_batch(
                entry_payloads,
                self.state_snapshot(),
                self.leader_state_snapshot(),
                &self.cluster_metadata,
                ctx,
            )
            .await;
        debug!(?result, "replication_handler::handle_raft_request_in_batch");

        // 3. Unify the processing results
        match result {
            // Case 1: Successfully reached majority
            Ok(AppendResults {
                commit_quorum_achieved: true,
                peer_updates,
                learner_progress,
            }) => {
                // 1. Update all peer index
                self.update_peer_indexes(&peer_updates);

                // 2. Check Learner's catch-up status
                if let Err(e) = self.check_learner_progress(&learner_progress, ctx, role_tx).await {
                    error!(?e, "check_learner_progress failed");
                };

                // 3. Update commit index
                // Single-voter cluster: commit index = last log index (quorum of 1)
                // Multi-voter cluster: calculate commit index based on majority quorum
                let new_commit_index = if self.cluster_metadata.single_voter {
                    let last_log_index = ctx.raft_log().last_entry_id();
                    if last_log_index > self.commit_index() {
                        Some(last_log_index)
                    } else {
                        None
                    }
                } else {
                    self.calculate_new_commit_index(ctx.raft_log(), &peer_updates)
                };

                if let Some(new_commit_index) = new_commit_index {
                    debug!(
                        "[Leader-{}] New commit been acknowledged: {}",
                        self.node_id(),
                        new_commit_index
                    );
                    self.update_commit_index_with_signal(
                        Leader as i32,
                        self.current_term(),
                        new_commit_index,
                        role_tx,
                    )?;
                }

                // 4. Notify all clients of success
                for request in batch {
                    let _ = request.sender.send(Ok(ClientResponse::write_success()));
                }
            }

            // Case 2: Failed to reach majority
            Ok(AppendResults {
                commit_quorum_achieved: false,
                peer_updates,
                learner_progress,
            }) => {
                // 1. Update all peer index
                self.update_peer_indexes(&peer_updates);

                // 2. Check Learner's catch-up status
                if let Err(e) = self.check_learner_progress(&learner_progress, ctx, role_tx).await {
                    error!(?e, "check_learner_progress failed");
                };

                // 3. Determine error code based on verifiability
                let responses_received = peer_updates.len();
                let error_code = if is_majority(responses_received, cluster_size) {
                    ErrorCode::RetryRequired
                } else {
                    ErrorCode::ProposeFailed
                };

                // 4. Notify all clients of failure
                for request in batch {
                    let _ = request.sender.send(Ok(ClientResponse::client_error(error_code)));
                }
            }

            // Case 3: High term found
            Err(Error::Consensus(ConsensusError::Replication(ReplicationError::HigherTerm(
                higher_term,
            )))) => {
                warn!("Higher term detected: {}", higher_term);
                self.update_current_term(higher_term);
                self.send_become_follower_event(None, role_tx)?;

                // Notify client of term expiration
                for request in batch {
                    let _ = request
                        .sender
                        .send(Ok(ClientResponse::client_error(ErrorCode::TermOutdated)));
                }

                return Err(ReplicationError::HigherTerm(higher_term).into());
            }

            // Case 4: Other errors
            Err(e) => {
                error!("Batch processing failed: {:?}", e);

                // Notify all clients of failure
                for request in batch {
                    let _ = request
                        .sender
                        .send(Ok(ClientResponse::client_error(ErrorCode::ProposeFailed)));
                }

                return Err(e);
            }
        }

        Ok(())
    }

    /// Update peer node index
    #[instrument(skip(self))]
    fn update_peer_indexes(
        &mut self,
        peer_updates: &HashMap<u32, PeerUpdate>,
    ) {
        for (peer_id, update) in peer_updates {
            if let Err(e) = self.update_next_index(*peer_id, update.next_index) {
                error!("Failed to update next index: {:?}", e);
            }
            trace!(
                "Updated next index for peer {}-{}",
                peer_id, update.next_index
            );
            if let Some(match_index) = update.match_index {
                if let Err(e) = self.update_match_index(*peer_id, match_index) {
                    error!("Failed to update match index: {:?}", e);
                }
                trace!("Updated match index for peer {}-{}", peer_id, match_index);
            }
        }
    }

    pub async fn check_learner_progress(
        &mut self,
        learner_progress: &HashMap<u32, Option<u64>>,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        debug!(?learner_progress, "check_learner_progress");

        if !self.should_check_learner_progress(ctx) {
            return Ok(());
        }

        if learner_progress.is_empty() {
            return Ok(());
        }

        let ready_learners = self.find_promotable_learners(learner_progress, ctx).await;
        let new_promotions = self.deduplicate_promotions(ready_learners);

        if !new_promotions.is_empty() {
            self.enqueue_and_notify_promotions(new_promotions, role_tx)?;
        }

        Ok(())
    }

    /// Check if enough time has elapsed since last learner progress check
    fn should_check_learner_progress(
        &mut self,
        ctx: &RaftContext<T>,
    ) -> bool {
        let throttle_interval =
            Duration::from_millis(ctx.node_config().raft.learner_check_throttle_ms);
        if self.last_learner_check.elapsed() < throttle_interval {
            return false;
        }
        self.last_learner_check = Instant::now();
        true
    }

    /// Find learners that are caught up and eligible for promotion
    async fn find_promotable_learners(
        &self,
        learner_progress: &HashMap<u32, Option<u64>>,
        ctx: &RaftContext<T>,
    ) -> Vec<u32> {
        let leader_commit = self.commit_index();
        let threshold = ctx.node_config().raft.learner_catchup_threshold;
        let membership = ctx.membership();

        let mut ready_learners = Vec::new();

        for (&node_id, &match_index_opt) in learner_progress.iter() {
            if !membership.contains_node(node_id).await {
                continue;
            }

            if !self.is_learner_caught_up(match_index_opt, leader_commit, threshold) {
                continue;
            }

            let node_status =
                membership.get_node_status(node_id).await.unwrap_or(NodeStatus::ReadOnly);
            if !node_status.is_promotable() {
                debug!(
                    ?node_id,
                    ?node_status,
                    "Learner caught up but status is not Promotable, skipping"
                );
                continue;
            }

            debug!(
                ?node_id,
                match_index = ?match_index_opt.unwrap_or(0),
                ?leader_commit,
                gap = leader_commit.saturating_sub(match_index_opt.unwrap_or(0)),
                "Learner caught up"
            );
            ready_learners.push(node_id);
        }

        ready_learners
    }

    /// Check if learner has caught up with leader based on log gap
    fn is_learner_caught_up(
        &self,
        match_index: Option<u64>,
        leader_commit: u64,
        threshold: u64,
    ) -> bool {
        let match_index = match_index.unwrap_or(0);
        let gap = leader_commit.saturating_sub(match_index);
        gap <= threshold
    }

    /// Remove learners already in pending promotions queue
    fn deduplicate_promotions(
        &self,
        ready_learners: Vec<u32>,
    ) -> Vec<u32> {
        let already_pending: std::collections::HashSet<_> =
            self.pending_promotions.iter().map(|p| p.node_id).collect();

        ready_learners.into_iter().filter(|id| !already_pending.contains(id)).collect()
    }

    /// Add promotions to queue and send notification event
    fn enqueue_and_notify_promotions(
        &mut self,
        new_promotions: Vec<u32>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        info!(
            ?new_promotions,
            "Learners caught up, adding to pending promotions"
        );

        for node_id in new_promotions {
            self.pending_promotions
                .push_back(PendingPromotion::new(node_id, Instant::now()));
        }

        role_tx
            .send(RoleEvent::ReprocessEvent(Box::new(
                RaftEvent::PromoteReadyLearners,
            )))
            .map_err(|e| {
                let error_str = format!("{e:?}");
                error!("Failed to send PromoteReadyLearners: {}", error_str);
                Error::System(SystemError::Network(NetworkError::SingalSendFailed(
                    error_str,
                )))
            })?;

        Ok(())
    }

    #[allow(dead_code)]
    pub async fn batch_promote_learners(
        &mut self,
        ready_learners_ids: Vec<u32>,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        // 1. Determine optimal promotion status based on quorum safety
        debug!("1. Determine optimal promotion status based on quorum safety");
        let membership = ctx.membership();
        let current_voters = membership.voters().await.len();
        // let syncings = membership.nodes_with_status(NodeStatus::Syncing).len();
        let new_active_count = current_voters + ready_learners_ids.len();

        // Determine target status based on quorum safety
        trace!(
            ?current_voters,
            ?ready_learners_ids,
            "[Node-{}] new_active_count: {}",
            self.node_id(),
            new_active_count
        );
        let target_status = if ensure_safe_join(self.node_id(), new_active_count).is_ok() {
            trace!(
                "Going to update nodes-{:?} status to Active",
                ready_learners_ids
            );
            NodeStatus::Active
        } else {
            trace!(
                "Not enough quorum to promote learners: {:?}",
                ready_learners_ids
            );
            return Ok(());
        };

        // 2. Create configuration change payload
        debug!("2. Create configuration change payload");
        let config_change = Change::BatchPromote(BatchPromote {
            node_ids: ready_learners_ids.clone(),
            new_status: target_status as i32,
        });

        info!(?config_change, "Replicating cluster config");

        // 3. Submit single config change for all ready learners
        debug!("3. Submit single config change for all ready learners");
        // Use verify_leadership_persistent for config changes (must eventually succeed)
        match self
            .verify_leadership_persistent(
                vec![EntryPayload::config(config_change)],
                true,
                ctx,
                role_tx,
            )
            .await
        {
            Ok(true) => {
                info!(
                    "Batch promotion committed for nodes: {:?}",
                    ready_learners_ids
                );
            }
            Ok(false) => {
                warn!("Failed to commit batch promotion");
            }
            Err(e) => {
                error!("Batch promotion error: {:?}", e);
                return Err(e);
            }
        }

        Ok(())
    }

    /// Calculate new submission index
    #[instrument(skip(self))]
    fn calculate_new_commit_index(
        &mut self,
        raft_log: &Arc<ROF<T>>,
        peer_updates: &HashMap<u32, PeerUpdate>,
    ) -> Option<u64> {
        let old_commit_index = self.commit_index();
        let current_term = self.current_term();

        let matched_ids: Vec<u64> =
            peer_updates.keys().filter_map(|&id| self.match_index(id)).collect();

        let new_commit_index =
            raft_log.calculate_majority_matched_index(current_term, old_commit_index, matched_ids);

        if new_commit_index.is_some() && new_commit_index.unwrap() > old_commit_index {
            new_commit_index
        } else {
            None
        }
    }

    #[allow(dead_code)]
    fn if_update_commit_index(
        &self,
        new_commit_index_option: Option<u64>,
    ) -> (bool, u64) {
        let current_commit_index = self.commit_index();
        if let Some(new_commit_index) = new_commit_index_option {
            debug!("Leader::update_commit_index: {:?}", new_commit_index);
            if current_commit_index < new_commit_index {
                return (true, new_commit_index);
            }
        }
        debug!("Leader::update_commit_index: false");
        (false, current_commit_index)
    }

    /// Calculate safe read index for linearizable reads.
    ///
    /// Returns max(commitIndex, noopIndex) to ensure:
    /// 1. All committed data is visible
    /// 2. Current term's no-op entry is accounted for
    /// 3. Read index is fixed at request arrival time (avoids waiting for concurrent writes)
    ///
    /// This optimization prevents reads from waiting for writes that arrived
    /// after the read request started processing.
    #[doc(hidden)]
    pub fn calculate_read_index(&self) -> u64 {
        let commit_index = self.commit_index();
        let noop_index = self.noop_log_id.unwrap_or(0);
        std::cmp::max(commit_index, noop_index)
    }

    /// Wait for state machine to apply up to target index.
    ///
    /// This is used by linearizable reads to ensure they see all committed data
    /// up to the read_index calculated at request arrival time.
    #[doc(hidden)]
    pub async fn wait_until_applied(
        &self,
        target_index: u64,
        state_machine_handler: &Arc<SMHOF<T>>,
        last_applied: u64,
    ) -> Result<()> {
        if last_applied < target_index {
            state_machine_handler.update_pending(target_index);

            let timeout_ms = self.node_config.raft.read_consistency.state_machine_sync_timeout_ms;
            state_machine_handler
                .wait_applied(target_index, std::time::Duration::from_millis(timeout_ms))
                .await?;

            debug!("wait_until_applied: target_index={} success", target_index);
        }
        Ok(())
    }

    #[instrument(skip(self))]
    fn scheduled_purge_upto(
        &mut self,
        received_last_included: LogId,
    ) {
        if let Some(existing) = self.scheduled_purge_upto {
            if existing.index >= received_last_included.index {
                warn!(
                    ?received_last_included,
                    ?existing,
                    "Will not update scheduled_purge_upto, received invalid last_included log"
                );
                return;
            }
        }
        info!(?self.scheduled_purge_upto, ?received_last_included, "Updte scheduled_purge_upto.");
        self.scheduled_purge_upto = Some(received_last_included);
    }

    /// Process peer purge responses and handle higher term detection.
    ///
    /// # Returns
    /// - `Ok(true)` if a higher term was detected and leader should step down
    /// - `Ok(false)` if processing completed normally
    ///
    /// # Higher Term Handling
    /// According to Raft §5.1: "If a server receives a request with a stale term number,
    /// it rejects the request." Conversely, if we receive a response with a higher term,
    /// we must immediately:
    /// 1. Update our term to the higher value
    /// 2. Step down to Follower role
    /// 3. Signal caller to abort ongoing operations (e.g., skip Phase 3 local purge)
    ///
    /// # Safety
    /// This function sends `BecomeFollower` event which triggers role transition.
    /// Caller MUST check return value and skip subsequent leader-only operations.
    pub(crate) fn peer_purge_progress(
        &mut self,
        responses: Vec<Result<PurgeLogResponse>>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<bool> {
        if responses.is_empty() {
            return Ok(false);
        }
        for r in responses.iter().flatten() {
            // Higher term detection: peer has advanced to a newer term
            // This violates our leadership validity (Raft §5.1)
            if r.term > self.current_term() {
                self.update_current_term(r.term);
                self.send_become_follower_event(None, role_tx)?;
                return Ok(true); // Signal step-down, caller must abort leader operations
            }

            if let Some(last_purged) = r.last_purged {
                self.peer_purge_progress
                    .entry(r.node_id)
                    .and_modify(|v| *v = last_purged.index)
                    .or_insert(last_purged.index);
            }
        }

        Ok(false)
    }

    fn send_become_follower_event(
        &self,
        new_leader_id: Option<u32>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        info!(
            ?new_leader_id,
            "Leader is going to step down as Follower..."
        );
        role_tx.send(RoleEvent::BecomeFollower(new_leader_id)).map_err(|e| {
            let error_str = format!("{e:?}");
            error!("Failed to send: {}", error_str);
            NetworkError::SingalSendFailed(error_str)
        })?;

        Ok(())
    }

    /// Determines if logs prior to `last_included_in_snapshot` can be permanently discarded.
    ///
    /// Implements Leader-side log compaction safety checks per Raft paper §7.2:
    /// > "The leader uses a new RPC called InstallSnapshot to send snapshots to followers that are
    /// > too far behind"
    ///
    /// # Safety Invariants (ALL must hold)
    ///
    /// 1. **Committed Entry Guarantee**   `last_included_in_snapshot.index < self.commit_index`
    ///    - Ensures we never discard uncommitted entries (Raft §5.4.2)
    ///    - Maintains at least one committed entry after purge for log matching property
    ///
    /// 2. **Monotonic Snapshot Advancement**   `last_purge_index < last_included_in_snapshot.index`
    ///    - Enforces snapshot indices strictly increase (prevents rollback attacks)
    ///    - Maintains sequential purge ordering (FSM safety requirement)
    ///
    /// 3. **Cluster-wide Progress Validation**   `peer_purge_progress.values().all(≥
    ///    snapshot.index)`
    ///    - Ensures ALL followers have confirmed ability to reach this snapshot
    ///    - Prevents leadership changes from causing log inconsistencies
    ///
    /// 4. **Operation Atomicity**   `pending_purge.is_none()`
    ///    - Ensures only one concurrent purge operation
    ///    - Critical for linearizable state machine semantics
    ///
    /// # Implementation Notes
    /// - Leader must maintain `peer_purge_progress` through AppendEntries responses
    /// - Actual log discard should be deferred until storage confirms snapshot persistence
    /// - Design differs from followers by requiring full cluster confirmation (Raft extension for
    ///   enhanced durability)
    #[instrument(skip(self))]
    pub fn can_purge_logs(
        &self,
        last_purge_index: Option<LogId>,
        last_included_in_snapshot: LogId,
    ) -> bool {
        let commit_index = self.commit_index();
        debug!(?self
                    .peer_purge_progress, ?commit_index, ?last_purge_index, ?last_included_in_snapshot, "can_purge_logs");
        let monotonic_check = last_purge_index
            .map(|lid| lid.index < last_included_in_snapshot.index)
            .unwrap_or(true);

        last_included_in_snapshot.index < commit_index
            && monotonic_check
            && self.peer_purge_progress.values().all(|&v| v >= last_included_in_snapshot.index)
    }

    pub async fn handle_join_cluster(
        &mut self,
        join_request: JoinRequest,
        sender: MaybeCloneOneshotSender<std::result::Result<JoinResponse, Status>>,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        let node_id = join_request.node_id;
        let node_role = join_request.node_role;
        let address = join_request.address;
        let status = join_request.status;
        let membership = ctx.membership();

        // 1. Validate join request
        debug!("1. Validate join request");
        if membership.contains_node(node_id).await {
            let error_msg = format!("Node {node_id} already exists in cluster");
            warn!(%error_msg);
            return self.send_join_error(sender, MembershipError::NodeAlreadyExists(node_id)).await;
        }

        // 2. Create configuration change payload
        debug!("2. Create configuration change payload");
        if let Err(e) = membership.can_rejoin(node_id, node_role).await {
            let error_msg = format!("Node {node_id} cannot rejoin: {e}",);
            warn!(%error_msg);
            return self
                .send_join_error(sender, MembershipError::JoinClusterError(error_msg))
                .await;
        }

        let config_change = Change::AddNode(AddNode {
            node_id,
            address: address.clone(),
            status,
        });

        // 3. Submit config change, and wait for quorum confirmation
        debug!("3. Wait for quorum confirmation");
        // Use verify_leadership_persistent for config changes (must eventually succeed)
        match self
            .verify_leadership_persistent(
                vec![EntryPayload::config(config_change)],
                true,
                ctx,
                role_tx,
            )
            .await
        {
            Ok(true) => {
                // 4. Update node status to Syncing
                debug!("4. Update node status to Syncing");

                debug!(
                    "After updating, the replications peers: {:?}",
                    ctx.membership().replication_peers().await
                );

                // Note: Cluster metadata cache will be updated via MembershipApplied event
                // after CommitHandler applies the config change to membership state

                // 5. Send successful response
                debug!("5. Send successful response");
                info!("Join config committed for node {}", node_id);
                self.send_join_success(node_id, &address, sender, ctx).await?;
            }
            Ok(false) => {
                warn!("Failed to commit join config for node {}", node_id);
                self.send_join_error(sender, MembershipError::CommitTimeout).await?
            }
            Err(e) => {
                error!("Error waiting for commit: {:?}", e);
                self.send_join_error(sender, e).await?
            }
        }
        Ok(())
    }

    async fn send_join_success(
        &self,
        node_id: u32,
        address: &str,
        sender: MaybeCloneOneshotSender<std::result::Result<JoinResponse, Status>>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        // Retrieve latest snapshot metadata, if there is
        let snapshot_metadata = ctx.state_machine_handler().get_latest_snapshot_metadata();

        // Prepare response
        let response = JoinResponse {
            success: true,
            error: String::new(),
            config: Some(
                ctx.membership()
                    .retrieve_cluster_membership_config(self.shared_state().current_leader())
                    .await,
            ),
            config_version: ctx.membership().get_cluster_conf_version().await,
            snapshot_metadata,
            leader_id: self.node_id(),
        };

        sender.send(Ok(response)).map_err(|e| {
            error!("Failed to send join response: {:?}", e);
            NetworkError::SingalSendFailed(format!("{e:?}"))
        })?;

        info!(
            "Node {} ({}) successfully added as learner",
            node_id, address
        );

        // Print leader accepting new node message (Plan B)
        crate::utils::cluster_printer::print_leader_accepting_new_node(
            self.node_id(),
            node_id,
            address,
            d_engine_proto::common::NodeRole::Learner as i32,
        );

        Ok(())
    }

    async fn send_join_error(
        &self,
        sender: MaybeCloneOneshotSender<std::result::Result<JoinResponse, Status>>,
        error: impl Into<Error>,
    ) -> Result<()> {
        let error = error.into();
        let status = Status::failed_precondition(error.to_string());

        sender.send(Err(status)).map_err(|e| {
            error!("Failed to send join error: {:?}", e);
            NetworkError::SingalSendFailed(format!("{e:?}"))
        })?;

        Err(error)
    }

    #[cfg(any(test, feature = "__test_support"))]
    pub fn new(
        node_id: u32,
        node_config: Arc<RaftNodeConfig>,
    ) -> Self {
        let ReplicationConfig {
            rpc_append_entries_in_batch_threshold,
            rpc_append_entries_batch_process_delay_in_ms,
            rpc_append_entries_clock_in_ms,
            ..
        } = node_config.raft.replication;

        LeaderState {
            cluster_metadata: ClusterMetadata {
                single_voter: false,
                total_voters: 0,
                replication_targets: vec![],
            },
            shared_state: SharedState::new(node_id, None, None),
            timer: Box::new(ReplicationTimer::new(
                rpc_append_entries_clock_in_ms,
                rpc_append_entries_batch_process_delay_in_ms,
            )),
            next_index: HashMap::new(),
            match_index: HashMap::new(),
            noop_log_id: None,

            batch_buffer: Box::new(BatchBuffer::new(
                rpc_append_entries_in_batch_threshold,
                Duration::from_millis(rpc_append_entries_batch_process_delay_in_ms),
            )),

            node_config,
            scheduled_purge_upto: None,
            last_purged_index: None, //TODO
            last_learner_check: Instant::now(),
            peer_purge_progress: HashMap::new(),
            snapshot_in_progress: AtomicBool::new(false),
            next_membership_maintenance_check: Instant::now(),
            pending_promotions: VecDeque::new(),
            lease_timestamp: AtomicU64::new(0),
            read_buffer: Vec::new(),
            read_buffer_start_time: None,
            _marker: PhantomData,
        }
    }

    pub async fn trigger_background_snapshot(
        node_id: u32,
        metadata: SnapshotMetadata,
        state_machine_handler: Arc<SMHOF<T>>,
        membership: Arc<MOF<T>>,
        config: SnapshotConfig,
    ) -> Result<()> {
        let (result_tx, result_rx) = oneshot::channel();

        // Delegate the actual transfer to a dedicated thread pool
        tokio::task::spawn_blocking(move || {
            let rt = tokio::runtime::Handle::current();
            let result = rt.block_on(async move {
                let bulk_channel = membership
                    .get_peer_channel(node_id, ConnectionType::Bulk)
                    .await
                    .ok_or(NetworkError::PeerConnectionNotFound(node_id))?;

                let data_stream =
                    state_machine_handler.load_snapshot_data(metadata.clone()).await?;

                BackgroundSnapshotTransfer::<T>::run_push_transfer(
                    node_id,
                    data_stream,
                    bulk_channel,
                    config,
                )
                .await
            });

            // Non-blocking send result
            let _ = result_tx.send(result);
        });

        // Non-blocking check result
        tokio::spawn(async move {
            match result_rx.await {
                Ok(Ok(_)) => info!("Snapshot to {} completed", node_id),
                Ok(Err(e)) => error!("Snapshot to {} failed: {:?}", node_id, e),
                Err(_) => warn!("Snapshot result channel closed unexpectedly"),
            }
        });

        Ok(())
    }

    /// Processes all pending promotions while respecting the cluster's odd-node constraint
    pub async fn process_pending_promotions(
        &mut self,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        debug!(
            "[Leader {}] 🔄 process_pending_promotions called, pending: {:?}",
            self.node_id(),
            self.pending_promotions.iter().map(|p| p.node_id).collect::<Vec<_>>()
        );

        // Get promotion configuration from the node config
        let config = &ctx.node_config().raft.membership.promotion;

        // Step 1: Remove stale entries (older than configured threshold)
        let now = Instant::now();
        self.pending_promotions.retain(|entry| {
            now.duration_since(entry.ready_since) <= config.stale_learner_threshold
        });

        if self.pending_promotions.is_empty() {
            debug!(
                "[Leader {}] ❌ pending_promotions is empty after stale cleanup",
                self.node_id()
            );
            return Ok(());
        }

        // Step 2: Get current voter count (including self as leader)
        let membership = ctx.membership();
        let current_voters = membership.voters().await.len() + 1; // +1 for self (leader)
        debug!(
            "[Leader {}] 📊 current_voters: {}, pending: {}",
            self.node_id(),
            current_voters,
            self.pending_promotions.len()
        );

        // Step 3: Calculate the maximum batch size that preserves an odd total
        let max_batch_size =
            calculate_safe_batch_size(current_voters, self.pending_promotions.len());
        debug!(
            "[Leader {}] 🎯 max_batch_size: {}",
            self.node_id(),
            max_batch_size
        );

        if max_batch_size == 0 {
            // Nothing we can safely promote now
            debug!(
                "[Leader {}] ⚠️ max_batch_size is 0, cannot promote now",
                self.node_id()
            );
            return Ok(());
        }

        // Step 4: Extract the batch from the queue (FIFO order)
        let promotion_entries = self.drain_batch(max_batch_size);
        let promotion_node_ids = promotion_entries.iter().map(|e| e.node_id).collect::<Vec<_>>();

        // Step 5: Execute batch promotion
        if !promotion_node_ids.is_empty() {
            // Log the batch promotion
            info!(
                "Promoting learner batch of {} nodes: {:?} (total voters: {} -> {})",
                promotion_node_ids.len(),
                promotion_node_ids,
                current_voters,
                current_voters + promotion_node_ids.len()
            );

            // Attempt promotion and restore batch on failure
            let result = self.safe_batch_promote(promotion_node_ids.clone(), ctx, role_tx).await;

            if let Err(e) = result {
                // Restore entries to the front of the queue in reverse order
                for entry in promotion_entries.into_iter().rev() {
                    self.pending_promotions.push_front(entry);
                }
                return Err(e);
            }

            info!(
                "Promotion successful. Cluster members: {:?}",
                membership.voters().await
            );
        }

        trace!(
            ?self.pending_promotions,
            "Step 6: Reschedule if any pending promotions remain"
        );
        // Step 6: Reschedule if any pending promotions remain
        if !self.pending_promotions.is_empty() {
            debug!(
                "[Leader {}] 🔁 Re-sending PromoteReadyLearners for remaining pending: {:?}",
                self.node_id(),
                self.pending_promotions.iter().map(|p| p.node_id).collect::<Vec<_>>()
            );
            // Important: Re-send the event to trigger next cycle
            role_tx
                .send(RoleEvent::ReprocessEvent(Box::new(
                    RaftEvent::PromoteReadyLearners,
                )))
                .map_err(|e| {
                    let error_str = format!("{e:?}");
                    error!("Send PromoteReadyLearners event failed: {}", error_str);
                    NetworkError::SingalSendFailed(error_str)
                })?;
        }

        Ok(())
    }

    /// Removes the first `count` nodes from the pending queue and returns them
    pub(super) fn drain_batch(
        &mut self,
        count: usize,
    ) -> Vec<PendingPromotion> {
        let mut batch = Vec::with_capacity(count);
        for _ in 0..count {
            if let Some(entry) = self.pending_promotions.pop_front() {
                batch.push(entry);
            } else {
                break;
            }
        }
        batch
    }

    async fn safe_batch_promote(
        &mut self,
        batch: Vec<u32>,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        let change = Change::BatchPromote(BatchPromote {
            node_ids: batch.clone(),
            new_status: NodeStatus::Active as i32,
        });

        // Submit batch activation
        // Use verify_leadership_persistent for config changes (must eventually succeed)
        self.verify_leadership_persistent(vec![EntryPayload::config(change)], true, ctx, role_tx)
            .await?;

        Ok(())
    }

    async fn run_periodic_maintenance(
        &mut self,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        if let Err(e) = self.conditionally_purge_stale_learners(role_tx, ctx).await {
            error!("Stale learner purge failed: {}", e);
        }

        if let Err(e) = self.conditionally_purge_zombie_nodes(role_tx, ctx).await {
            error!("Zombie node purge failed: {}", e);
        }

        // Regardless of whether a stale node is found, we set the next check according to a fixed
        // period
        self.reset_next_membership_maintenance_check(
            ctx.node_config().raft.membership.membership_maintenance_interval,
        );
        Ok(())
    }

    /// Periodic check triggered every ~30s in the worst-case scenario
    /// using priority-based lazy scheduling. Actual average frequency
    /// is inversely proportional to system load.
    pub async fn conditionally_purge_stale_learners(
        &mut self,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        let config = &ctx.node_config.raft.membership.promotion;

        // Optimization: Skip check 99.9% of the time using scheduled trial method
        if self.pending_promotions.is_empty()
            || self.next_membership_maintenance_check > Instant::now()
        {
            trace!("Skipping stale learner check");
            return Ok(());
        }

        let now = Instant::now();
        let queue_len = self.pending_promotions.len();

        // Inspect only oldest 1% of items or max 100 entries per rules
        let inspect_count = queue_len.min(100).min(1.max(queue_len / 100));
        let mut stale_entries = Vec::new();

        trace!("Inspecting {} entries", inspect_count);
        for _ in 0..inspect_count {
            if let Some(entry) = self.pending_promotions.pop_front() {
                trace!(
                    "Inspecting entry: {:?} - {:?} - {:?}",
                    entry,
                    now.duration_since(entry.ready_since),
                    &config.stale_learner_threshold
                );
                if now.duration_since(entry.ready_since) > config.stale_learner_threshold {
                    stale_entries.push(entry);
                } else {
                    // Return non-stale entry and stop
                    self.pending_promotions.push_front(entry);
                    break;
                }
            } else {
                break;
            }
        }

        trace!("Stale learner check completed: {:?}", stale_entries);

        // Process collected stale entries
        for entry in stale_entries {
            if let Err(e) = self.handle_stale_learner(entry.node_id, role_tx, ctx).await {
                error!("Failed to handle stale learner: {}", e);
            }
        }

        Ok(())
    }

    /// Remove non-Active zombie nodes that exceed failure threshold
    async fn conditionally_purge_zombie_nodes(
        &mut self,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        // Optimize: get membership reference once to reduce lock contention
        let membership = ctx.membership();
        let zombie_candidates = membership.get_zombie_candidates().await;
        let mut nodes_to_remove = Vec::new();

        for node_id in zombie_candidates {
            if let Some(status) = membership.get_node_status(node_id).await {
                if status != NodeStatus::Active {
                    nodes_to_remove.push(node_id);
                }
            }
        }
        // Batch removal if we have candidates
        if !nodes_to_remove.is_empty() {
            let change = Change::BatchRemove(BatchRemove {
                node_ids: nodes_to_remove.clone(),
            });

            info!(
                "Proposing batch removal of zombie nodes: {:?}",
                nodes_to_remove
            );

            // Submit single config change for all nodes
            // Use verify_leadership_persistent for config changes (must eventually succeed)
            match self
                .verify_leadership_persistent(
                    vec![EntryPayload::config(change)],
                    false,
                    ctx,
                    role_tx,
                )
                .await
            {
                Ok(true) => {
                    info!("Batch removal committed for nodes: {:?}", nodes_to_remove);
                }
                Ok(false) => {
                    warn!("Failed to commit batch removal");
                }
                Err(e) => {
                    error!("Batch removal error: {:?}", e);
                    return Err(e);
                }
            }
        }

        Ok(())
    }

    pub fn reset_next_membership_maintenance_check(
        &mut self,
        membership_maintenance_interval: Duration,
    ) {
        self.next_membership_maintenance_check = Instant::now() + membership_maintenance_interval;
    }

    /// Remove stalled learner via membership change consensus
    pub async fn handle_stale_learner(
        &mut self,
        node_id: u32,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
        ctx: &RaftContext<T>,
    ) -> Result<()> {
        // Stalled learner detected - remove via membership change (requires consensus)
        warn!(
            "Learner {} is stalled, removing from cluster via consensus",
            node_id
        );

        let change = Change::BatchRemove(BatchRemove {
            node_ids: vec![node_id],
        });

        // Submit removal through membership change entry (requires quorum consensus)
        // Use verify_leadership_persistent for config changes (must eventually succeed)
        match self
            .verify_leadership_persistent(vec![EntryPayload::config(change)], true, ctx, role_tx)
            .await
        {
            Ok(true) => {
                info!(
                    "Stalled learner {} successfully removed from cluster",
                    node_id
                );
            }
            Ok(false) => {
                warn!("Failed to commit removal of stalled learner {}", node_id);
            }
            Err(e) => {
                error!("Error removing stalled learner {}: {:?}", node_id, e);
                return Err(e);
            }
        }

        Ok(())
    }

    /// Check if current lease is still valid for LeaseRead policy
    pub fn is_lease_valid(
        &self,
        ctx: &RaftContext<T>,
    ) -> bool {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;

        let last_confirmed = self.lease_timestamp.load(std::sync::atomic::Ordering::Acquire);
        let lease_duration = ctx.node_config().raft.read_consistency.lease_duration_ms;

        if now < last_confirmed {
            // Clock moved backwards: system clock issue, treat lease as invalid
            error!("Clock moved backwards: Now {now}, Last Confirmed {last_confirmed}");
            return false;
        }

        // Allow multiple requests within the same millisecond (now == last_confirmed)
        if now == last_confirmed {
            return true;
        }
        (now - last_confirmed) < lease_duration
    }

    /// Update lease timestamp after successful leadership verification
    fn update_lease_timestamp(&self) {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;

        self.lease_timestamp.store(now, std::sync::atomic::Ordering::Release);
    }

    /// Verify leadership with quorum and refresh lease timestamp on success.
    /// This is a single-shot verification (fast-fail) optimized for read operations.
    ///
    /// Returns Ok(()) if verification succeeds and lease is refreshed.
    /// Returns Err on any verification failure.
    async fn verify_leadership_and_refresh_lease(
        &mut self,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        match self.verify_internal_quorum(vec![], true, ctx, role_tx).await {
            Ok(QuorumVerificationResult::Success) => {
                self.update_lease_timestamp();
                Ok(())
            }
            Ok(QuorumVerificationResult::LeadershipLost) => Err(ConsensusError::Replication(
                ReplicationError::NotLeader { leader_id: None },
            )
            .into()),
            Ok(QuorumVerificationResult::RetryRequired) => {
                Err(ConsensusError::Replication(ReplicationError::QuorumNotReached).into())
            }
            Err(e) => Err(e),
        }
    }

    /// Process batched LinearizableRead requests with a single quorum verification.
    ///
    /// This function takes all pending read requests from the buffer, performs one
    /// leadership verification for the entire batch, waits for state machine to catch up,
    /// and executes all reads. This batching strategy significantly reduces the number
    /// of quorum checks required for high read throughput scenarios.
    pub(super) async fn process_linearizable_read_batch(
        &mut self,
        ctx: &RaftContext<T>,
        role_tx: &mpsc::UnboundedSender<RoleEvent>,
    ) -> Result<()> {
        // Take buffer atomically
        let batch = std::mem::take(&mut self.read_buffer);
        self.read_buffer_start_time = None;

        if batch.is_empty() {
            return Ok(());
        }

        let batch_size = batch.len();
        info!("Read batch: flushing {} requests", batch_size);

        // All requests in read_buffer are LinearizableRead (filtered in handle_raft_event)
        // Perform single verification for the entire batch and refresh lease
        if let Err(e) = self.verify_leadership_and_refresh_lease(ctx, role_tx).await {
            // Leadership verification failed - drain buffer with error
            error!(
                "Read batch: verification failed - draining {} requests: {}",
                batch_size, e
            );
            let status = tonic::Status::failed_precondition(format!(
                "Leadership verification failed: {e:?}"
            ));
            for (_, sender) in batch {
                let _ = sender.send(Err(status.clone()));
            }
            return Ok(());
        }

        // Wait for state machine to catch up
        let last_applied = ctx.state_machine().last_applied().index;
        let read_index = self.calculate_read_index();

        if last_applied < read_index {
            ctx.handlers.state_machine_handler.update_pending(read_index);
            let timeout_ms = self.node_config.raft.read_consistency.state_machine_sync_timeout_ms;
            ctx.handlers
                .state_machine_handler
                .wait_applied(read_index, Duration::from_millis(timeout_ms))
                .await
                .map_err(|e| {
                    error!("Read batch: wait_applied failed: {}", e);
                    e
                })?;
        }

        // Execute all reads and respond
        for (req, sender) in batch {
            let results = ctx
                .handlers
                .state_machine_handler
                .read_from_state_machine(req.keys)
                .unwrap_or_default();
            let _ = sender.send(Ok(ClientResponse::read_results(results)));
        }

        // // Metrics
        // if let Some(metrics) = &ctx.metrics {
        //     metrics.read_batch_size.observe(batch_size as f64);
        // }

        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn test_update_lease_timestamp(&self) {
        self.update_lease_timestamp();
    }
}

impl<T: TypeConfig> From<&CandidateState<T>> for LeaderState<T> {
    fn from(candidate: &CandidateState<T>) -> Self {
        let ReplicationConfig {
            rpc_append_entries_in_batch_threshold,
            rpc_append_entries_batch_process_delay_in_ms,
            rpc_append_entries_clock_in_ms,
            ..
        } = candidate.node_config.raft.replication;

        // Clone shared_state and set self as leader immediately
        let shared_state = candidate.shared_state.clone();
        shared_state.set_current_leader(candidate.node_id());

        Self {
            shared_state,
            timer: Box::new(ReplicationTimer::new(
                rpc_append_entries_clock_in_ms,
                rpc_append_entries_batch_process_delay_in_ms,
            )),
            next_index: HashMap::new(),
            match_index: HashMap::new(),
            noop_log_id: None,

            batch_buffer: Box::new(BatchBuffer::new(
                rpc_append_entries_in_batch_threshold,
                Duration::from_millis(rpc_append_entries_batch_process_delay_in_ms),
            )),

            node_config: candidate.node_config.clone(),

            scheduled_purge_upto: None,
            last_purged_index: candidate.last_purged_index,
            last_learner_check: Instant::now(),
            snapshot_in_progress: AtomicBool::new(false),
            peer_purge_progress: HashMap::new(),
            next_membership_maintenance_check: Instant::now(),
            pending_promotions: VecDeque::new(),
            cluster_metadata: ClusterMetadata {
                single_voter: false,
                total_voters: 0,
                replication_targets: vec![],
            },
            lease_timestamp: AtomicU64::new(0),
            read_buffer: Vec::new(),
            read_buffer_start_time: None,
            _marker: PhantomData,
        }
    }
}

impl<T: TypeConfig> Debug for LeaderState<T> {
    fn fmt(
        &self,
        f: &mut std::fmt::Formatter<'_>,
    ) -> std::fmt::Result {
        f.debug_struct("LeaderState")
            .field("shared_state", &self.shared_state)
            .field("next_index", &self.next_index)
            .field("match_index", &self.match_index)
            .field("noop_log_id", &self.noop_log_id)
            .finish()
    }
}

/// Calculates the maximum number of nodes we can promote while keeping the total voter count
/// odd
///
/// - `current`: current number of voting nodes
/// - `available`: number of ready learners pending promotion
///
/// Returns the maximum number of nodes to promote (0 if no safe promotion exists)
pub fn calculate_safe_batch_size(
    current: usize,
    available: usize,
) -> usize {
    if (current + available) % 2 == 1 {
        // Promoting all is safe
        available
    } else {
        // We can only promote (available - 1) to keep the invariant
        // But if available - 1 == 0, then we cannot promote any?
        available.saturating_sub(1)
    }
}

pub(super) fn send_replay_raft_event(
    role_tx: &mpsc::UnboundedSender<RoleEvent>,
    raft_event: RaftEvent,
) -> Result<()> {
    role_tx.send(RoleEvent::ReprocessEvent(Box::new(raft_event))).map_err(|e| {
        let error_str = format!("{e:?}");
        error!("Failed to send: {}", error_str);
        NetworkError::SingalSendFailed(error_str).into()
    })
}