From 0723aa757fa039577f1e626001b8d752b5ab76b3 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sun, 12 Apr 2026 22:34:04 +0800 Subject: [PATCH] refactor(core) #357, #358: remove dead leader code and add missing test coverage #358: Delete batch_promote_learners (superseded by process_pending_promotions), if_update_commit_index (no callers), and their associated tests. Remove now-unused ensure_safe_join import. #357: Add tests for StepDownSelfRemoved event handling, conditionally_purge_zombie_nodes BatchRemove path (no_candidates / submits_batch_remove / skips_active_nodes), and BackpressureMetrics record_rejection/record_buffer_utilization branches. --- d-engine-core/src/raft_role/leader_state.rs | 82 +--- .../leader_state_test/backpressure_test.rs | 56 +++ .../leader_state_test/event_handling_test.rs | 32 ++ .../membership_change_test.rs | 426 +++++++----------- 4 files changed, 261 insertions(+), 335 deletions(-) diff --git a/d-engine-core/src/raft_role/leader_state.rs b/d-engine-core/src/raft_role/leader_state.rs index b82c959b..1a28a617 100644 --- a/d-engine-core/src/raft_role/leader_state.rs +++ b/d-engine-core/src/raft_role/leader_state.rs @@ -41,7 +41,6 @@ use crate::alias::MOF; use crate::alias::ROF; use crate::alias::SMHOF; use crate::alias::TROF; -use crate::ensure_safe_join; use crate::event::ClientCmd; use crate::network::Transport; use crate::stream::create_production_snapshot_stream; @@ -2737,69 +2736,6 @@ impl LeaderState { Ok(()) } - #[allow(dead_code)] - pub async fn batch_promote_learners( - &mut self, - ready_learners_ids: Vec, - ctx: &RaftContext, - role_tx: &mpsc::UnboundedSender, - ) -> 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 (fire-and-forget). - // Commit confirmation comes via the normal Raft flow; maintenance re-runs on next tick. - debug!("3. Submit single config change for all ready learners"); - self.execute_request_immediately( - RaftRequestWithSignal { - id: { rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 21) }, - payloads: vec![EntryPayload::config(config_change)], - senders: vec![], - wait_for_apply_event: false, - }, - ctx, - role_tx, - ) - .await?; - - Ok(()) - } - /// Calculate new submission index #[instrument(skip(self))] fn calculate_new_commit_index( @@ -2831,22 +2767,6 @@ impl LeaderState { } } - #[allow(dead_code)] - fn if_update_commit_index( - &self, - new_commit_index_option: Option, - ) -> (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: @@ -3690,7 +3610,7 @@ impl LeaderState { } /// Remove non-Active zombie nodes that exceed failure threshold - async fn conditionally_purge_zombie_nodes( + pub(super) async fn conditionally_purge_zombie_nodes( &mut self, role_tx: &mpsc::UnboundedSender, ctx: &RaftContext, diff --git a/d-engine-core/src/raft_role/leader_state_test/backpressure_test.rs b/d-engine-core/src/raft_role/leader_state_test/backpressure_test.rs index ab616f93..3a84a73a 100644 --- a/d-engine-core/src/raft_role/leader_state_test/backpressure_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/backpressure_test.rs @@ -355,3 +355,59 @@ async fn test_backpressure_all_read_policies() { "Should reject with ResourceExhausted" ); } + +// ============================================================================ +// BackpressureMetrics Unit Tests +// ============================================================================ + +/// BackpressureMetrics::record_rejection with enabled=true: write and read paths. +/// Verifies both branches complete without panic when metrics recording is active. +#[test] +fn test_backpressure_metrics_record_rejection_enabled() { + use crate::raft_role::leader_state::BackpressureMetrics; + let m = BackpressureMetrics::new(1, true, 1); + // Both calls must not panic + m.record_rejection(true); // write rejection + m.record_rejection(false); // read rejection +} + +/// BackpressureMetrics::record_rejection with enabled=false: both paths are no-ops. +#[test] +fn test_backpressure_metrics_record_rejection_disabled() { + use crate::raft_role::leader_state::BackpressureMetrics; + let m = BackpressureMetrics::new(1, false, 1); + m.record_rejection(true); + m.record_rejection(false); +} + +/// BackpressureMetrics::record_buffer_utilization with enabled=true: sampling logic. +/// sample_rate=1 means every call records; verifies write and read labels both work. +#[test] +fn test_backpressure_metrics_record_buffer_utilization_enabled() { + use crate::raft_role::leader_state::BackpressureMetrics; + let m = BackpressureMetrics::new(1, true, 1); + m.record_buffer_utilization(0.0, true); // write, empty buffer + m.record_buffer_utilization(0.5, true); // write, half full + m.record_buffer_utilization(1.0, true); // write, full + m.record_buffer_utilization(0.5, false); // read, half full +} + +/// BackpressureMetrics::record_buffer_utilization with enabled=false: no-op. +#[test] +fn test_backpressure_metrics_record_buffer_utilization_disabled() { + use crate::raft_role::leader_state::BackpressureMetrics; + let m = BackpressureMetrics::new(1, false, 1); + m.record_buffer_utilization(0.5, true); + m.record_buffer_utilization(0.5, false); +} + +/// BackpressureMetrics: sample_rate>1 skips most recordings (counter-based sampling). +/// With sample_rate=3, only every 3rd call records; all calls must not panic. +#[test] +fn test_backpressure_metrics_sampling_rate() { + use crate::raft_role::leader_state::BackpressureMetrics; + let m = BackpressureMetrics::new(1, true, 3); + for i in 0..10 { + m.record_buffer_utilization(i as f64 / 10.0, i % 2 == 0); + } +} diff --git a/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs b/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs index d21ce6e4..d438565c 100644 --- a/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/event_handling_test.rs @@ -936,3 +936,35 @@ async fn test_drain_read_buffer_clears_pending_reads_on_stepdown() { "drained reads must get Unavailable error" ); } + +// ============================================================================ +// StepDownSelfRemoved Tests +// ============================================================================ + +/// Leader receives StepDownSelfRemoved → sends BecomeFollower and returns Ok. +/// +/// Per Raft protocol, after a leader commits its own removal from the cluster +/// it must immediately step down to Follower. This test verifies: +/// 1. handle_raft_event returns Ok(()) +/// 2. A BecomeFollower(None) event is sent on role_tx +#[tokio::test] +async fn test_step_down_self_removed_sends_become_follower() { + let (_graceful_tx, graceful_rx) = watch::channel(()); + let raft_context = mock_raft_context("/tmp/test_step_down_self_removed", graceful_rx, None); + + let node_config = raft_context.node_config(); + let mut leader_state = LeaderState::::new(1, node_config); + + let (role_tx, mut role_rx) = mpsc::unbounded_channel(); + let result = leader_state + .handle_raft_event(RaftEvent::StepDownSelfRemoved, &raft_context, role_tx) + .await; + + assert!(result.is_ok(), "StepDownSelfRemoved must return Ok"); + + let event = role_rx.try_recv().expect("BecomeFollower event must be sent"); + assert!( + matches!(event, RoleEvent::BecomeFollower(None)), + "Expected BecomeFollower(None), got: {event:?}" + ); +} diff --git a/d-engine-core/src/raft_role/leader_state_test/membership_change_test.rs b/d-engine-core/src/raft_role/leader_state_test/membership_change_test.rs index ad56db62..04932ad8 100644 --- a/d-engine-core/src/raft_role/leader_state_test/membership_change_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/membership_change_test.rs @@ -753,260 +753,6 @@ mod trigger_background_snapshot_test { } } -// ============================================================================ -// Batch Learner Promotion Tests -// ============================================================================ - -#[cfg(test)] -mod batch_promote_learners_test { - use std::sync::Arc; - use std::time::Duration; - - use tokio::sync::{mpsc, watch}; - - use parking_lot::Mutex; - - use crate::Error; - use crate::MockMembership; - use crate::MockRaftLog; - use crate::MockReplicationCore; - use crate::RaftContext; - use crate::event::RoleEvent; - use crate::raft_role::leader_state::LeaderState; - use crate::test_utils::mock::MockBuilder; - use crate::test_utils::mock::MockTypeConfig; - use crate::test_utils::node_config; - use d_engine_proto::common::EntryPayload; - use d_engine_proto::common::{NodeRole::Follower, NodeStatus}; - use d_engine_proto::server::cluster::NodeMeta; - use mockall::predicate::eq; - - enum VerifyInternalQuorumWithRetrySuccess { - Failure, - Error, - } - - struct TestContext { - raft_context: RaftContext, - role_tx: mpsc::UnboundedSender, - _role_rx: mpsc::UnboundedReceiver, - leader_state: LeaderState, - } - - async fn setup_test_context( - test_name: &str, - current_voters: usize, - ready_learners: Vec, - verify_leadership_limited_retry_success: VerifyInternalQuorumWithRetrySuccess, - ) -> TestContext { - let (_graceful_tx, graceful_rx) = watch::channel(()); - let mut node_config = node_config(&format!("/tmp/{test_name}")); - node_config.raft.learner_catchup_threshold = 100; - node_config.raft.membership.verify_leadership_persistent_timeout = Duration::from_secs(1); - - let mut raft_context = - MockBuilder::new(graceful_rx).with_node_config(node_config).build_context(); - - // Mock membership - let mut membership = MockMembership::new(); - membership.expect_is_single_node_cluster().returning(|| false); - membership.expect_can_rejoin().returning(|_, _| Ok(())); - membership.expect_voters().returning(move || { - (1..=current_voters) - .map(|id| NodeMeta { - id: id as u32, - address: format!("addr_{id}"), - role: Follower.into(), - status: NodeStatus::Active.into(), - }) - .collect() - }); - - let mut replication_handler = MockReplicationCore::::new(); - // New architecture: prepare_batch_requests returns empty vector or error - replication_handler.expect_prepare_batch_requests().times(..).returning( - move |_, _, _, _, _| match verify_leadership_limited_retry_success { - VerifyInternalQuorumWithRetrySuccess::Failure => { - // Failure: returns empty vector, verification times out (no quorum) - Ok(crate::PrepareResult::default()) - } - VerifyInternalQuorumWithRetrySuccess::Error => { - // Error: returns fatal error during prepare phase - Err(Error::Fatal("Simulated higher term error".to_string())) - } - }, - ); - raft_context.handlers.replication_handler = replication_handler; - - let mut raft_log = MockRaftLog::new(); - raft_log.expect_last_entry_id().returning(|| 10); - raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| Some(5)); - raft_context.storage.raft_log = Arc::new(raft_log); - - // Mock learner statuses - for learner_id in &ready_learners { - membership - .expect_get_node_status() - .with(eq(*learner_id)) - .return_const(Some(NodeStatus::Promotable)); - } - - raft_context.membership = Arc::new(membership); - - let leader_state = LeaderState::new(1, raft_context.node_config()); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - TestContext { - raft_context, - role_tx, - _role_rx, - leader_state, - } - } - - /// Test successful promotion path: verifies correct BatchPromote config is submitted - /// when ensure_safe_join allows the promotion (3 voters + 1 learner = 4 total, even → odd after - /// adding self, so safe). Quorum completion requires a full Raft loop; this test validates - /// the unit-testable portion (config payload is correctly formed and submitted). - #[tokio::test] - async fn test_batch_promote_learners_success() { - use d_engine_proto::common::membership_change::Change; - - let captured_payloads = Arc::new(Mutex::new(Vec::::new())); - let captured_clone = captured_payloads.clone(); - - let (_graceful_tx, graceful_rx) = watch::channel(()); - let mut cfg = node_config("/tmp/test_batch_promote_learners_success"); - cfg.raft.learner_catchup_threshold = 100; - // Short timeout: quorum can't complete in unit tests; we only verify payload submission - cfg.raft.membership.verify_leadership_persistent_timeout = Duration::from_millis(50); - - let mut raft_context = MockBuilder::new(graceful_rx).with_node_config(cfg).build_context(); - - let mut membership = MockMembership::new(); - membership.expect_is_single_node_cluster().returning(|| false); - membership.expect_can_rejoin().returning(|_, _| Ok(())); - // 3 voter followers → new_active_count = 3+1 = 4 (even) → ensure_safe_join passes - membership.expect_voters().returning(|| { - (1..=3u32) - .map(|id| NodeMeta { - id, - address: format!("addr_{id}"), - role: Follower.into(), - status: NodeStatus::Active.into(), - }) - .collect() - }); - membership - .expect_get_node_status() - .with(eq(4u32)) - .return_const(Some(NodeStatus::Promotable)); - raft_context.membership = Arc::new(membership); - - let mut replication_handler = MockReplicationCore::::new(); - replication_handler.expect_prepare_batch_requests().times(..).returning( - move |payloads, _, _, _, _| { - captured_clone.lock().extend(payloads.clone()); - Ok(crate::PrepareResult::default()) - }, - ); - raft_context.handlers.replication_handler = replication_handler; - - let mut raft_log = MockRaftLog::new(); - raft_log.expect_last_entry_id().returning(|| 10); - raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| Some(5)); - raft_context.storage.raft_log = Arc::new(raft_log); - - let mut leader_state = LeaderState::new(1, raft_context.node_config()); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - // Timeout is expected: quorum can't complete without a running Raft loop - let _ = leader_state.batch_promote_learners(vec![4], &raft_context, &role_tx).await; - - // Verify the correct BatchPromote config change was submitted for replication - let payloads = captured_payloads.lock(); - assert_eq!(payloads.len(), 1, "Should submit exactly one config change"); - match &payloads[0].payload { - Some(d_engine_proto::common::entry_payload::Payload::Config(change)) => { - match &change.change { - Some(Change::BatchPromote(batch_promote)) => { - assert_eq!(batch_promote.node_ids, vec![4], "Should promote node 4"); - assert_eq!( - batch_promote.new_status, - NodeStatus::Active as i32, - "Should promote to Active status" - ); - } - _ => panic!("Expected BatchPromote change, got: {:?}", change.change), - } - } - _ => panic!("Expected Config payload, got: {:?}", payloads[0].payload), - } - } - - /// Test promotion: ensure_safe_join passes (3+1=4 even), config change submitted fire-and-forget. - /// Commit confirmation comes via the normal Raft flow; this test validates submission succeeds. - #[tokio::test] - async fn test_batch_promote_learners_quorum_failed() { - let mut ctx = setup_test_context( - "test_batch_promote_learners_quorum_failed", - 3, // 3 voter followers → new_active_count = 3+1 = 4 (even) → safety passes - vec![4], - VerifyInternalQuorumWithRetrySuccess::Failure, - ) - .await; - - let result = ctx - .leader_state - .batch_promote_learners(vec![4], &ctx.raft_context, &ctx.role_tx) - .await; - - // Fire-and-forget: config change submitted, returns Ok regardless of future commit outcome - assert!(result.is_ok()); - } - - /// Test safety check prevents unsafe promotion: 2 voter followers + 1 learner = 3 total - /// (odd) → ensure_safe_join returns Err → batch_promote_learners returns Ok(()) immediately - /// without submitting any config change. Cluster stays at even count → Raft quorum preserved. - #[tokio::test] - async fn test_batch_promote_learners_quorum_safety() { - let mut ctx = setup_test_context( - "test_batch_promote_learners_quorum_safety", - 2, // 2 voter followers → new_active_count = 2+1 = 3 (odd) → safety blocks - vec![4], - VerifyInternalQuorumWithRetrySuccess::Failure, - ) - .await; - - let result = ctx - .leader_state - .batch_promote_learners(vec![4], &ctx.raft_context, &ctx.role_tx) - .await; - - assert!(result.is_ok()); - } - - /// Test error during quorum verification: prepare_batch_requests returns Fatal error → - /// verify_internal_quorum propagates Err immediately (no timeout wait). - #[tokio::test] - async fn test_batch_promote_learners_verification_error() { - let mut ctx = setup_test_context( - "test_batch_promote_learners_verification_error", - 3, - vec![4], - VerifyInternalQuorumWithRetrySuccess::Error, - ) - .await; - - let result = ctx - .leader_state - .batch_promote_learners(vec![4], &ctx.raft_context, &ctx.role_tx) - .await; - - assert!(result.is_err()); - } -} - // ============================================================================ // Stale Learner Detection and Cleanup Tests // ============================================================================ @@ -2084,3 +1830,175 @@ mod pending_promotion_tests { } } } + +// ============================================================================ +// Zombie Node Purge Tests +// ============================================================================ + +#[cfg(test)] +mod zombie_purge_tests { + use std::sync::Arc; + + use parking_lot::Mutex; + use tokio::sync::{mpsc, watch}; + + use crate::MockMembership; + use crate::MockRaftLog; + use crate::raft_role::leader_state::LeaderState; + use crate::test_utils::mock::{MockBuilder, MockTypeConfig}; + use crate::test_utils::node_config; + use d_engine_proto::common::EntryPayload; + use d_engine_proto::common::{NodeRole::Follower, NodeStatus}; + use d_engine_proto::server::cluster::NodeMeta; + + /// No zombie candidates → conditionally_purge_zombie_nodes does nothing. + #[tokio::test] + async fn test_purge_zombie_nodes_no_candidates() { + let (_graceful_tx, graceful_rx) = watch::channel(()); + let cfg = node_config("/tmp/test_purge_zombie_nodes_no_candidates"); + let mut raft_context = MockBuilder::new(graceful_rx).with_node_config(cfg).build_context(); + + let mut membership = MockMembership::new(); + membership.expect_get_zombie_candidates().returning(Vec::new); + raft_context.membership = Arc::new(membership); + + let captured_payloads = Arc::new(Mutex::new(Vec::::new())); + let captured_clone = captured_payloads.clone(); + raft_context + .handlers + .replication_handler + .expect_prepare_batch_requests() + .times(0) // Must NOT be called when there are no candidates + .returning(move |payloads, _, _, _, _| { + captured_clone.lock().extend(payloads.clone()); + Ok(crate::PrepareResult::default()) + }); + + let mut raft_log = MockRaftLog::new(); + raft_log.expect_last_entry_id().returning(|| 10); + raft_context.storage.raft_log = Arc::new(raft_log); + + let node_config = raft_context.node_config(); + let mut leader = LeaderState::::new(1, node_config); + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + + let result = leader.conditionally_purge_zombie_nodes(&role_tx, &raft_context).await; + assert!(result.is_ok()); + assert!( + captured_payloads.lock().is_empty(), + "No BatchRemove should be submitted" + ); + } + + /// Zombie candidates with non-Active status → BatchRemove config change submitted. + #[tokio::test] + async fn test_purge_zombie_nodes_submits_batch_remove() { + use d_engine_proto::common::membership_change::Change; + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let mut cfg = node_config("/tmp/test_purge_zombie_nodes_submits_batch_remove"); + cfg.raft.membership.verify_leadership_persistent_timeout = + std::time::Duration::from_millis(50); + let mut raft_context = MockBuilder::new(graceful_rx).with_node_config(cfg).build_context(); + + let mut membership = MockMembership::new(); + // Two zombie candidates: node 5 and node 6 + membership.expect_get_zombie_candidates().returning(|| vec![5, 6]); + // Both are non-Active (Promotable) → both should be removed + membership.expect_get_node_status().returning(|_| Some(NodeStatus::Promotable)); + // Membership voter mock needed by execute_request_immediately internals + membership.expect_voters().returning(|| { + vec![NodeMeta { + id: 1, + address: "addr_1".to_string(), + role: Follower.into(), + status: NodeStatus::Active as i32, + }] + }); + membership.expect_is_single_node_cluster().returning(|| false); + membership.expect_can_rejoin().returning(|_, _| Ok(())); + raft_context.membership = Arc::new(membership); + + let captured_payloads = Arc::new(Mutex::new(Vec::::new())); + let captured_clone = captured_payloads.clone(); + raft_context + .handlers + .replication_handler + .expect_prepare_batch_requests() + .times(..) + .returning(move |payloads, _, _, _, _| { + captured_clone.lock().extend(payloads.clone()); + Ok(crate::PrepareResult::default()) + }); + + let mut raft_log = MockRaftLog::new(); + raft_log.expect_last_entry_id().returning(|| 10); + raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| Some(5)); + raft_context.storage.raft_log = Arc::new(raft_log); + + let node_config = raft_context.node_config(); + let mut leader = LeaderState::::new(1, node_config); + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + + let result = leader.conditionally_purge_zombie_nodes(&role_tx, &raft_context).await; + assert!(result.is_ok()); + + // Verify BatchRemove was submitted for both zombie nodes + let payloads = captured_payloads.lock(); + assert_eq!(payloads.len(), 1, "Should submit exactly one BatchRemove"); + match &payloads[0].payload { + Some(d_engine_proto::common::entry_payload::Payload::Config(change)) => { + match &change.change { + Some(Change::BatchRemove(batch_remove)) => { + let mut ids = batch_remove.node_ids.clone(); + ids.sort(); + assert_eq!(ids, vec![5, 6], "Should remove both zombie nodes"); + } + _ => panic!("Expected BatchRemove, got: {:?}", change.change), + } + } + _ => panic!("Expected Config payload, got: {:?}", payloads[0].payload), + } + } + + /// Zombie candidate with Active status → should NOT be removed. + #[tokio::test] + async fn test_purge_zombie_nodes_skips_active_nodes() { + let (_graceful_tx, graceful_rx) = watch::channel(()); + let cfg = node_config("/tmp/test_purge_zombie_nodes_skips_active"); + let mut raft_context = MockBuilder::new(graceful_rx).with_node_config(cfg).build_context(); + + let mut membership = MockMembership::new(); + // One zombie candidate: node 7, but it's Active + membership.expect_get_zombie_candidates().returning(|| vec![7]); + membership.expect_get_node_status().returning(|_| Some(NodeStatus::Active)); + raft_context.membership = Arc::new(membership); + + let captured_payloads = Arc::new(Mutex::new(Vec::::new())); + let captured_clone = captured_payloads.clone(); + raft_context + .handlers + .replication_handler + .expect_prepare_batch_requests() + .times(0) // Must NOT be called: Active node is not a zombie to remove + .returning(move |payloads, _, _, _, _| { + captured_clone.lock().extend(payloads.clone()); + Ok(crate::PrepareResult::default()) + }); + + let mut raft_log = MockRaftLog::new(); + raft_log.expect_last_entry_id().returning(|| 10); + raft_context.storage.raft_log = Arc::new(raft_log); + + let node_config = raft_context.node_config(); + let mut leader = LeaderState::::new(1, node_config); + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + + let result = leader.conditionally_purge_zombie_nodes(&role_tx, &raft_context).await; + assert!(result.is_ok()); + assert!( + captured_payloads.lock().is_empty(), + "Active node must not be removed" + ); + } +}