diff --git a/Cargo.lock b/Cargo.lock index 06c6f6af..a329873d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -960,21 +960,15 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.16.1" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" dependencies = [ "allocator-api2", "equivalent", "foldhash 0.2.0", ] -[[package]] -name = "hashbrown" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" - [[package]] name = "heck" version = "0.5.0" @@ -1243,11 +1237,11 @@ checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" [[package]] name = "lru" -version = "0.16.3" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dc47f592c06f33f8e3aea9591776ec7c9f9e4124778ff8a3c3b87159f7e593" +checksum = "0e0b564323a0fb6d54b864f625ae139de9612e27edb944dda37c109f05aac531" dependencies = [ - "hashbrown 0.16.1", + "hashbrown 0.17.0", ] [[package]] @@ -1888,9 +1882,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.11" +version = "0.103.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20a6af516fea4b20eccceaf166e8aa666ac996208e8a644ce3ef5aa783bc7cd4" +checksum = "8279bb85272c9f10811ae6a6c547ff594d6a7f3c6c6b02ee9726d1d0dcfcdd06" dependencies = [ "ring", "rustls-pki-types", diff --git a/config/base/raft.toml b/config/base/raft.toml index 074470f3..180fff73 100644 --- a/config/base/raft.toml +++ b/config/base/raft.toml @@ -122,9 +122,6 @@ cluster_healthcheck_probe_service_name = "d_engine.server.cluster.ClusterManagem # Timeout for leader to keep retrying no-op entries to confirm leadership (seconds). verify_leadership_persistent_timeout = { secs = 3600 } -# Interval for membership maintenance tasks (seconds). -membership_maintenance_interval = { secs = 30 } - [raft.membership.zombie] # Consecutive connection failures before marking a node as zombie. threshold = 3 @@ -136,9 +133,6 @@ purge_interval = { secs = 30 } # Duration after which a learner is considered stale if not promoted (seconds). stale_learner_threshold = { secs = 300 } -# Interval for checking stale learners (seconds). -stale_check_interval = { secs = 30 } - [raft.persistence] # Persistence strategy: "MemFirst" — entries written to memory first, flushed to disk async. # For per-write durability (process-crash safe on every write), set: diff --git a/d-engine-core/Cargo.toml b/d-engine-core/Cargo.toml index eadbb56e..62473cd0 100644 --- a/d-engine-core/Cargo.toml +++ b/d-engine-core/Cargo.toml @@ -55,7 +55,7 @@ tokio-stream = "0.1.16" # Packing/unpacking TAR archives astral-tokio-tar = "0.6" rand = "0.9" -lru = "0.16" +lru = "0.17" memmap2 = "0.9.5" crc32fast = "1.4.2" # used in stream diff --git a/d-engine-core/benches/leader_state_bench.rs b/d-engine-core/benches/leader_state_bench.rs index 413a8391..b7813cf5 100644 --- a/d-engine-core/benches/leader_state_bench.rs +++ b/d-engine-core/benches/leader_state_bench.rs @@ -255,7 +255,6 @@ impl BenchFixture { nodes: vec![], current_leader_id: None, }); - membership.expect_get_zombie_candidates().returning(Vec::new); membership.expect_get_peers_id_with_condition().returning(|_| vec![]); membership.expect_is_single_node_cluster().returning(|| false); membership.expect_initial_cluster_size().returning(|| 3); diff --git a/d-engine-core/src/config/raft.rs b/d-engine-core/src/config/raft.rs index a34f2972..159e8a8f 100644 --- a/d-engine-core/src/config/raft.rs +++ b/d-engine-core/src/config/raft.rs @@ -370,9 +370,6 @@ pub struct MembershipConfig { #[serde(default = "default_verify_leadership_persistent_timeout")] pub verify_leadership_persistent_timeout: Duration, - #[serde(default = "default_membership_maintenance_interval")] - pub membership_maintenance_interval: Duration, - #[serde(default)] pub zombie: ZombieConfig, @@ -385,7 +382,6 @@ impl Default for MembershipConfig { Self { cluster_healthcheck_probe_service_name: default_probe_service(), verify_leadership_persistent_timeout: default_verify_leadership_persistent_timeout(), - membership_maintenance_interval: default_membership_maintenance_interval(), zombie: ZombieConfig::default(), promotion: PromotionConfig::default(), } @@ -395,11 +391,6 @@ fn default_probe_service() -> String { "d_engine.server.cluster.ClusterManagementService".to_string() } -// 30 seconds -fn default_membership_maintenance_interval() -> Duration { - Duration::from_secs(30) -} - /// Default timeout for leader to keep verifying its leadership. /// /// In Raft, the leader may retry sending no-op entries to confirm it still holds leadership. @@ -731,15 +722,12 @@ fn default_zombie_purge_interval() -> Duration { pub struct PromotionConfig { #[serde(default = "default_stale_learner_threshold")] pub stale_learner_threshold: Duration, - #[serde(default = "default_stale_check_interval")] - pub stale_check_interval: Duration, } impl Default for PromotionConfig { fn default() -> Self { Self { stale_learner_threshold: default_stale_learner_threshold(), - stale_check_interval: default_stale_check_interval(), } } } @@ -748,10 +736,6 @@ impl Default for PromotionConfig { fn default_stale_learner_threshold() -> Duration { Duration::from_secs(300) } -// 30 seconds -fn default_stale_check_interval() -> Duration { - Duration::from_secs(30) -} /// Defines how Raft log entries are persisted and accessed. /// /// All strategies use a configurable [`FlushPolicy`] to control when memory contents diff --git a/d-engine-core/src/event.rs b/d-engine-core/src/event.rs index d7e88d88..6fc9b728 100644 --- a/d-engine-core/src/event.rs +++ b/d-engine-core/src/event.rs @@ -111,6 +111,11 @@ pub enum RoleEvent { peer_id: u32, }, + /// A peer's connection failure count crossed zombie_threshold. + /// Emitted by RaftHealthMonitor (server layer) via an injected Sender. + /// Leader responds by proposing a BatchRemove config change for that node. + ZombieDetected(u32), + /// Fatal error from SM worker — node must shutdown. /// Sent via role_tx (P2) so it is not blocked behind external RPCs on event_tx (P4). FatalError { diff --git a/d-engine-core/src/membership.rs b/d-engine-core/src/membership.rs index 31106fb3..672237d8 100644 --- a/d-engine-core/src/membership.rs +++ b/d-engine-core/src/membership.rs @@ -208,8 +208,6 @@ where index: u64, ); - async fn get_zombie_candidates(&self) -> Vec; - /// If new node could rejoin the cluster again async fn can_rejoin( &self, diff --git a/d-engine-core/src/raft.rs b/d-engine-core/src/raft.rs index eb8fd163..42297fe0 100644 --- a/d-engine-core/src/raft.rs +++ b/d-engine-core/src/raft.rs @@ -601,6 +601,15 @@ where self.role.handle_peer_stream_error(peer_id); } + RoleEvent::ZombieDetected(node_id) => { + debug!(%node_id, "ZombieDetected: forwarding to leader for BatchRemove"); + if let Err(e) = + self.role.handle_zombie_detected(node_id, &self.role_tx, &self.ctx).await + { + error!(%node_id, ?e, "handle_zombie_detected failed"); + } + } + RoleEvent::SnapshotPushCompleted { peer_id, success } => { debug!(%peer_id, %success, "SnapshotPushCompleted"); if success { diff --git a/d-engine-core/src/raft_role/leader_state.rs b/d-engine-core/src/raft_role/leader_state.rs index 6ea2f723..0f0e7357 100644 --- a/d-engine-core/src/raft_role/leader_state.rs +++ b/d-engine-core/src/raft_role/leader_state.rs @@ -408,19 +408,16 @@ pub struct LeaderState { pub last_learner_check: Instant, // -- Stale Learner Handling -- - /// The next scheduled time to check for stale learners. + /// Earliest time the oldest pending learner becomes stale. /// - /// 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`). + /// Set to `front().ready_since + stale_learner_threshold` when a learner is enqueued. + /// `None` when `pending_promotions` is empty (O(1) short-circuit in tick). + /// After `drain_batch`, call `refresh_stale_deadline` to recompute from the new front. /// - /// 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, + /// Design: VecDeque is FIFO so the front is always the oldest entry — no need to + /// scan the whole queue. `min()` over all entries would always select the front, + /// so we directly bind the deadline to `front().ready_since + threshold`. + pub stale_check_deadline: Option, /// Queue of learners that have caught up and are pending promotion to voter. pub pending_promotions: VecDeque, @@ -625,6 +622,15 @@ impl RaftRoleState for LeaderState { Ok(()) } + async fn handle_zombie_detected( + &mut self, + node_id: u32, + role_tx: &mpsc::UnboundedSender, + ctx: &RaftContext, + ) -> Result<()> { + self.handle_zombie_node(node_id, role_tx, ctx).await + } + fn handle_snapshot_push_completed( &mut self, peer_id: u32, @@ -730,9 +736,11 @@ impl RaftRoleState for LeaderState { // 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); + // 1. Stale learner O(1) deadline check — None short-circuits immediately + if self.stale_check_deadline.is_some_and(|d| now >= d) + && let Err(e) = self.check_and_purge_stale_front(role_tx, ctx).await + { + error!("Stale learner check failed: {}", e); } // 2. Heartbeat trigger check @@ -2716,10 +2724,15 @@ impl LeaderState { "Learners caught up, adding to pending promotions" ); + let threshold = self.node_config.raft.membership.promotion.stale_learner_threshold; for node_id in new_promotions { self.pending_promotions .push_back(PendingPromotion::new(node_id, Instant::now())); } + // Only update deadline if queue was previously empty (new front is the oldest) + if self.stale_check_deadline.is_none() { + self.refresh_stale_deadline(threshold); + } role_tx .send(RoleEvent::ReprocessEvent(Box::new( @@ -3081,7 +3094,7 @@ impl LeaderState { last_purged_index: None, //TODO last_learner_check: Instant::now(), snapshot_in_progress: AtomicBool::new(false), - next_membership_maintenance_check: Instant::now(), + stale_check_deadline: None, pending_promotions: VecDeque::new(), lease_timestamp: AtomicU64::new(0), linearizable_read_buffer: Box::new(BatchBuffer::new(batch_size).with_length_gauge( @@ -3169,6 +3182,8 @@ impl LeaderState { self.pending_promotions.retain(|entry| { now.saturating_duration_since(entry.ready_since) <= config.stale_learner_threshold }); + // Refresh deadline after potential removals + self.refresh_stale_deadline(config.stale_learner_threshold); if self.pending_promotions.is_empty() { debug!( @@ -3229,9 +3244,16 @@ impl LeaderState { for entry in promotion_entries.into_iter().rev() { self.pending_promotions.push_front(entry); } + // Refresh deadline to account for restored entries at front + let threshold = config.stale_learner_threshold; + self.refresh_stale_deadline(threshold); return Err(e); } + // Refresh stale deadline to reflect the new queue front after successful drain + let threshold = config.stale_learner_threshold; + self.refresh_stale_deadline(threshold); + info!( "Promotion successful. Cluster members: {:?}", membership.voters().await @@ -3529,137 +3551,105 @@ impl LeaderState { Ok(()) } - async fn run_periodic_maintenance( + /// Recompute `stale_check_deadline` from the current queue front. + /// + /// Must be called after any push_back or drain_batch that changes the front. + /// VecDeque is FIFO so the front is always the oldest entry; the deadline is + /// bound to `front().ready_since + threshold`. + /// Sets `None` when the queue is empty (O(1) short-circuit in tick). + pub fn refresh_stale_deadline( &mut self, - role_tx: &mpsc::UnboundedSender, - ctx: &RaftContext, - ) -> 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(()) + threshold: Duration, + ) { + self.stale_check_deadline = + self.pending_promotions.front().map(|front| front.ready_since + threshold); } - /// 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( + /// Called from tick() when `stale_check_deadline` fires. + /// + /// Pops all front entries whose `ready_since + threshold` has elapsed and + /// calls `handle_stale_learner` for each. After draining, refreshes the deadline + /// so the next stale check is bound to the new front. + pub async fn check_and_purge_stale_front( &mut self, role_tx: &mpsc::UnboundedSender, ctx: &RaftContext, ) -> 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 threshold = ctx.node_config.raft.membership.promotion.stale_learner_threshold; let now = Instant::now(); - let queue_len = self.pending_promotions.len(); + let mut stale_ids = Vec::new(); - // 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.saturating_duration_since(entry.ready_since), - &config.stale_learner_threshold - ); - if now.saturating_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; - } + // Pop all expired front entries (VecDeque is FIFO; front is always oldest) + while let Some(entry) = self.pending_promotions.front() { + if now >= entry.ready_since + threshold { + let entry = self.pending_promotions.pop_front().unwrap(); + stale_ids.push(entry.node_id); } else { break; } } - trace!("Stale learner check completed: {:?}", stale_entries); + // Refresh deadline to reflect the new front (or None if queue emptied) + self.refresh_stale_deadline(threshold); - // 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); + for node_id in stale_ids { + warn!( + node_id, + "Stale learner detected: pending promotion threshold exceeded" + ); + metrics::counter!( + "membership.stale_learner_removed", + &[("node_id", node_id.to_string())] + ) + .increment(1); + if let Err(e) = self.handle_stale_learner(node_id, role_tx, ctx).await { + error!(node_id, ?e, "Failed to handle stale learner"); } } Ok(()) } - /// Remove non-Active zombie nodes that exceed failure threshold - pub(super) async fn conditionally_purge_zombie_nodes( + /// Handle a ZombieDetected signal from the health monitor. + /// + /// ReadOnly nodes are permanent read replicas and must never be auto-removed. + /// All other non-Active nodes that repeatedly fail connections are removed via consensus. + pub async fn handle_zombie_node( &mut self, + node_id: u32, role_tx: &mpsc::UnboundedSender, ctx: &RaftContext, ) -> 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(); + let status = ctx.membership().get_node_status(node_id).await; - for node_id in zombie_candidates { - if let Some(status) = membership.get_node_status(node_id).await - && 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 + if status == Some(NodeStatus::ReadOnly) { + debug!( + node_id, + "Zombie signal ignored: ReadOnly node is exempt from auto-removal" ); - - // Submit single config change for all nodes (fire-and-forget). - self.execute_request_immediately( - RaftRequestWithSignal { - id: { rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 21) }, - payloads: vec![EntryPayload::config(change)], - senders: vec![], - wait_for_apply_event: false, - }, - ctx, - role_tx, - ) - .await?; + return Ok(()); } - Ok(()) - } + warn!( + node_id, + "Zombie detected: proposing BatchRemove via consensus" + ); - pub fn reset_next_membership_maintenance_check( - &mut self, - membership_maintenance_interval: Duration, - ) { - self.next_membership_maintenance_check = Instant::now() + membership_maintenance_interval; + let change = Change::BatchRemove(BatchRemove { + node_ids: vec![node_id], + }); + + self.execute_request_immediately( + RaftRequestWithSignal { + id: { rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 21) }, + payloads: vec![EntryPayload::config(change)], + senders: vec![], + wait_for_apply_event: false, + }, + ctx, + role_tx, + ) + .await } /// Remove stalled learner via membership change consensus @@ -3989,7 +3979,7 @@ impl From<&CandidateState> for LeaderState { last_purged_index: candidate.last_purged_index, last_learner_check: Instant::now(), snapshot_in_progress: AtomicBool::new(false), - next_membership_maintenance_check: Instant::now(), + stale_check_deadline: None, pending_promotions: VecDeque::new(), cluster_metadata: ClusterMetadata { single_voter: false, 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 04932ad8..65520e7f 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 @@ -451,7 +451,6 @@ async fn test_handle_join_cluster_quorum_failed() { .expect_retrieve_cluster_membership_config() .returning(|_| ClusterMembership::default()); membership.expect_get_cluster_conf_version().returning(|| 1); - membership.expect_get_zombie_candidates().returning(Vec::new); context.membership = Arc::new(membership); // prepare_batch_requests returns empty → config change is written but never committed @@ -767,7 +766,6 @@ mod stale_learner_tests { use crate::MockMembership; use crate::MockRaftLog; - use crate::RaftContext; use crate::raft_role::leader_state::{LeaderState, PendingPromotion}; use crate::test_utils::mock::MockTypeConfig; use crate::test_utils::mock::mock_raft_context; @@ -781,10 +779,8 @@ mod stale_learner_tests { ) -> (LeaderState, MockMembership) { let mut node_config = node_config(&format!("/tmp/{test_name}")); node_config.raft.membership.promotion.stale_learner_threshold = Duration::from_secs(30); - node_config.raft.membership.promotion.stale_check_interval = Duration::from_secs(60); let mut leader = LeaderState::new(1, Arc::new(node_config)); - leader.next_membership_maintenance_check = Instant::now(); // Add pending promotions with specified ages let now = Instant::now(); @@ -803,182 +799,6 @@ mod stale_learner_tests { (leader, membership) } - /// Create mock RaftContext with configurable membership - fn mock_stale_test_raft_context( - test_name: &str, - membership: Arc>, - config: Option>, - ) -> RaftContext { - let (_graceful_tx, graceful_rx) = watch::channel(()); - let mut ctx = mock_raft_context(&format!("/tmp/{test_name}"), graceful_rx, None); - ctx.membership = membership; - ctx.node_config = - config.unwrap_or_else(|| Arc::new(node_config(&format!("/tmp/{test_name}")))); - - // Mock raft_log for calculate_majority_matched_index - let mut raft_log = MockRaftLog::new(); - raft_log.expect_last_entry_id().returning(|| 0); - raft_log.expect_last_log_id().returning(|| None); - raft_log.expect_flush().returning(|| Ok(())); - raft_log.expect_load_hard_state().returning(|| Ok(None)); - raft_log.expect_save_hard_state().returning(|_| Ok(())); - raft_log.expect_calculate_majority_matched_index().returning(|_, _, _| Some(0)); - ctx.storage.raft_log = Arc::new(raft_log); - - // Mock replication handler to handle BatchRemove operations - ctx.handlers - .replication_handler - .expect_prepare_batch_requests() - .times(..) - .returning(|_, _, _, _, _| { - // Stale learner test: verification returns empty - Ok(crate::PrepareResult::default()) - }); - - ctx - } - - /// Test lazy staleness sampling optimization (checks only oldest 100 or 2% of queue) - #[tokio::test] - async fn test_stale_check_optimization() { - // Create queue with 200 entries (will only check oldest 100 or 2%) - let nodes: Vec<(u32, Duration)> = - (1..=200).map(|id| (id, Duration::from_secs(40))).collect(); - - let (mut leader, membership) = - create_test_leader_state("test_stale_check_optimization", nodes); - let mut node_config = node_config("/tmp/test_stale_check_optimization"); - node_config.raft.membership.promotion.stale_learner_threshold = Duration::from_secs(30); - node_config.raft.membership.promotion.stale_check_interval = Duration::from_secs(60); - let ctx = mock_stale_test_raft_context( - "test_stale_check_optimization", - Arc::new(membership), - Some(Arc::new(node_config)), - ); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - // Should only check first 100 entries (out of 200) - leader.conditionally_purge_stale_learners(&role_tx, &ctx).await.unwrap(); - - // Should purge exactly 2 entries (1% of 200 = 2) - assert_eq!(leader.pending_promotions.len(), 198); - } - - /// Test no purge when pending promotions are fresh (not expired) - #[tokio::test] - async fn test_no_purge_when_fresh() { - let (mut leader, mut membership) = create_test_leader_state( - "test_no_purge_when_fresh", - vec![ - (101, Duration::from_secs(15)), - (102, Duration::from_secs(20)), - ], - ); - // Should do nothing - expect no status updates - membership.expect_update_node_status().never(); - - let ctx = - mock_stale_test_raft_context("test_no_purge_when_fresh", Arc::new(membership), None); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - leader.conditionally_purge_stale_learners(&role_tx, &ctx).await.unwrap(); - assert_eq!(leader.pending_promotions.len(), 2); - } - - /// Test membership maintenance check scheduling logic - #[tokio::test] - async fn test_membership_maintenance_scheduling() { - let (mut leader, _) = - create_test_leader_state("test_membership_maintenance_scheduling", vec![]); - let interval = Duration::from_secs(60); - - // First call - leader.reset_next_membership_maintenance_check(interval); - let next_check1 = leader.next_membership_maintenance_check; - - // Small delay to ensure time progresses - tokio::time::sleep(Duration::from_millis(1)).await; - - // Second call - leader.reset_next_membership_maintenance_check(interval); - let next_check2 = leader.next_membership_maintenance_check; - - // Verify both are in the future - assert!(next_check1 > Instant::now()); - assert!(next_check2 > Instant::now()); - - // Verify second call moved the timer forward - assert!(next_check2 > next_check1); - - // Verify both are approximately interval in the future - let now = Instant::now(); - assert!(duration_diff(next_check1 - now, interval) < Duration::from_millis(10)); - assert!(duration_diff(next_check2 - now, interval) < Duration::from_millis(10)); - } - - fn duration_diff( - a: Duration, - b: Duration, - ) -> Duration { - a.abs_diff(b) - } - - /// Test system remains responsive during large queues - #[tokio::test] - async fn test_performance_large_queue() { - let nodes: Vec<(u32, Duration)> = - (1..=10_000).map(|id| (id, Duration::from_secs(40))).collect(); - - let (mut leader, membership) = - create_test_leader_state("test_performance_large_queue", nodes); - let ctx = mock_stale_test_raft_context( - "test_performance_large_queue", - Arc::new(membership), - None, - ); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - // Time the staleness check - let start = Instant::now(); - leader.conditionally_purge_stale_learners(&role_tx, &ctx).await.unwrap(); - let elapsed = start.elapsed(); - - // Should take <20ms even for large queues (due to sampling optimization) - println!("Staleness check for 10k nodes: {elapsed:?}"); - assert!( - elapsed < Duration::from_millis(20), - "Staleness check shouldn't process entire queue" - ); - } - - /// Test promotion timeout threshold edge cases - #[tokio::test] - async fn test_promotion_timeout_threshold() { - let (mut leader, membership) = create_test_leader_state( - "test_promotion_timeout_threshold", - vec![ - (101, Duration::from_secs(31)), // 1s over threshold - (102, Duration::from_secs(30)), // exactly at threshold - (103, Duration::from_secs(29)), // 1s under threshold - ], - ); - leader.next_membership_maintenance_check = Instant::now() - Duration::from_secs(1); - let mut node_config = node_config("/tmp/test_promotion_timeout_threshold"); - node_config.raft.membership.promotion.stale_learner_threshold = Duration::from_secs(30); - let ctx = mock_stale_test_raft_context( - "test_promotion_timeout_threshold", - Arc::new(membership), - Some(Arc::new(node_config)), - ); - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - leader.conditionally_purge_stale_learners(&role_tx, &ctx).await.unwrap(); - - // Should purge nodes over threshold (node 101, 102) but keep node 103 - assert_eq!(leader.pending_promotions.len(), 2); - assert!(leader.pending_promotions.iter().any(|p| p.node_id == 103)); - } - /// Split from original test_downgrade_affects_replication /// /// Original test validated that handle_stale_learner successfully removes a stale learner @@ -1797,17 +1617,6 @@ mod pending_promotion_tests { assert_eq!(queue.lock().len(), 10); } - /// Test: Stale check timing with paused time - #[tokio::test(start_paused = true)] - async fn test_stale_check_timing() { - let node_config = node_config("/tmp/test_stale_check_timing"); - let mut leader = LeaderState::::new(1, Arc::new(node_config)); - leader.reset_next_membership_maintenance_check(Duration::from_secs(60)); - - tokio::time::advance(Duration::from_secs(61)).await; - assert!(Instant::now() >= leader.next_membership_maintenance_check); - } - /// Test: Config propagation to stale handling #[test] fn test_config_propagation_to_stale_handling() { @@ -1851,62 +1660,34 @@ mod zombie_purge_tests { 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(); + // ========================================================================= + // handle_zombie_node — event-driven path + // ========================================================================= - 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. + /// ZombieDetected event for a Promotable node → BatchRemove submitted. + /// + /// # Given + /// - Leader receives ZombieDetected(5) + /// - Node 5 has status Promotable (non-ReadOnly, non-Active) + /// + /// # When + /// - handle_zombie_node(5) is called + /// + /// # Then + /// - A BatchRemove config change is proposed for node 5 #[tokio::test] - async fn test_purge_zombie_nodes_submits_batch_remove() { + async fn test_handle_zombie_node_non_readonly_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"); + let mut cfg = node_config("/tmp/test_handle_zombie_node_non_readonly"); 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 + // Node 5 is Promotable — not ReadOnly, 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, @@ -1940,19 +1721,16 @@ mod zombie_purge_tests { 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; + let result = leader.handle_zombie_node(5, &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"); + assert_eq!(batch_remove.node_ids, vec![5], "Should remove node 5"); } _ => panic!("Expected BatchRemove, got: {:?}", change.change), } @@ -1961,17 +1739,26 @@ mod zombie_purge_tests { } } - /// Zombie candidate with Active status → should NOT be removed. + /// ZombieDetected event for a ReadOnly node → exempt, no BatchRemove. + /// + /// # Given + /// - Leader receives ZombieDetected(5) + /// - Node 5 has status ReadOnly (permanent read replica, must not be auto-removed) + /// + /// # When + /// - handle_zombie_node(5) is called + /// + /// # Then + /// - No config change is proposed (ReadOnly nodes are exempt from zombie auto-removal) #[tokio::test] - async fn test_purge_zombie_nodes_skips_active_nodes() { + async fn test_handle_zombie_node_readonly_is_exempt() { let (_graceful_tx, graceful_rx) = watch::channel(()); - let cfg = node_config("/tmp/test_purge_zombie_nodes_skips_active"); + let cfg = node_config("/tmp/test_handle_zombie_node_readonly_exempt"); 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)); + // Node 5 is ReadOnly — must never be auto-removed + membership.expect_get_node_status().returning(|_| Some(NodeStatus::ReadOnly)); raft_context.membership = Arc::new(membership); let captured_payloads = Arc::new(Mutex::new(Vec::::new())); @@ -1980,7 +1767,7 @@ mod zombie_purge_tests { .handlers .replication_handler .expect_prepare_batch_requests() - .times(0) // Must NOT be called: Active node is not a zombie to remove + .times(0) // Must NOT be called — ReadOnly node is exempt .returning(move |payloads, _, _, _, _| { captured_clone.lock().extend(payloads.clone()); Ok(crate::PrepareResult::default()) @@ -1994,11 +1781,11 @@ mod zombie_purge_tests { 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; + let result = leader.handle_zombie_node(5, &role_tx, &raft_context).await; assert!(result.is_ok()); assert!( captured_payloads.lock().is_empty(), - "Active node must not be removed" + "ReadOnly node must not be auto-removed" ); } } diff --git a/d-engine-core/src/raft_role/leader_state_test/mod.rs b/d-engine-core/src/raft_role/leader_state_test/mod.rs index f8f489de..92a9c7d0 100644 --- a/d-engine-core/src/raft_role/leader_state_test/mod.rs +++ b/d-engine-core/src/raft_role/leader_state_test/mod.rs @@ -54,3 +54,6 @@ mod pending_lease_reads_test; #[cfg(test)] mod single_voter_commit_test; + +#[cfg(test)] +mod stale_learner_deadline_test; diff --git a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs index 6a86ffc5..026ecad8 100644 --- a/d-engine-core/src/raft_role/leader_state_test/replication_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/replication_test.rs @@ -601,7 +601,6 @@ async fn setup_commit_index_test_context( nodes: vec![], current_leader_id: None, }); - membership.expect_get_zombie_candidates().returning(Vec::new); membership.expect_pre_warm_connections().returning(|| Ok(())); if single_voter { membership.expect_replication_peers().returning(Vec::new); @@ -2180,7 +2179,6 @@ async fn test_stale_conflict_ack_does_not_regress_next_index() { current_leader_id: None, } }); - membership.expect_get_zombie_candidates().returning(Vec::new); membership.expect_pre_warm_connections().returning(|| Ok(())); membership.expect_replication_peers().returning(|| { vec![ diff --git a/d-engine-core/src/raft_role/leader_state_test/stale_learner_deadline_test.rs b/d-engine-core/src/raft_role/leader_state_test/stale_learner_deadline_test.rs new file mode 100644 index 00000000..1567445e --- /dev/null +++ b/d-engine-core/src/raft_role/leader_state_test/stale_learner_deadline_test.rs @@ -0,0 +1,270 @@ +//! TDD tests for the min-deadline stale learner mechanism (Phase 3, Issue #339). +//! +//! Design: instead of polling every `membership_maintenance_interval`, LeaderState +//! maintains a single `stale_check_deadline: Option` that is set to +//! `front().ready_since + stale_learner_threshold` whenever a learner is enqueued. +//! `tick()` performs an O(1) comparison and acts only when the deadline fires. +//! +//! Test plan: +//! 1. Deadline is None when queue is empty. +//! 2. Deadline is set to `ready_since + threshold` when first learner is enqueued. +//! 3. Deadline is NOT overwritten when a second learner is enqueued (front stays oldest). +//! 4. tick() does NOT call handle_stale_learner before the deadline. +//! 5. tick() DOES call handle_stale_learner when the deadline fires (stale learner removed). +//! 6. After drain_batch empties the queue, deadline resets to None. +//! 7. After drain_batch leaves items, deadline refreshes to new front's deadline. + +use std::sync::Arc; +use std::time::Duration; + +use parking_lot::Mutex; +use tokio::sync::{mpsc, watch}; +use tokio::time::Instant; + +use crate::MockMembership; +use crate::MockRaftLog; +use crate::raft_role::leader_state::{LeaderState, PendingPromotion}; +use crate::role_state::RaftRoleState; +use crate::test_utils::mock::{MockBuilder, MockTypeConfig}; +use crate::test_utils::node_config; +use d_engine_proto::common::EntryPayload; + +// ============================================================================ +// Helpers +// ============================================================================ + +async fn setup_with_threshold( + threshold: Duration +) -> ( + LeaderState, + crate::RaftContext, + mpsc::UnboundedSender, + mpsc::Sender, +) { + let (_shutdown_tx, shutdown_rx) = watch::channel(()); + let mut cfg = node_config("/tmp/stale_learner_deadline_test"); + cfg.raft.membership.promotion.stale_learner_threshold = threshold; + let ctx = MockBuilder::new(shutdown_rx).with_node_config(cfg).build_context(); + let state = LeaderState::::new(1, ctx.node_config.clone()); + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + let (raft_tx, _raft_rx) = mpsc::channel(1); + (state, ctx, role_tx, raft_tx) +} + +// ============================================================================ +// Test 1: stale_check_deadline is None when queue is empty +// ============================================================================ + +#[tokio::test] +async fn test_stale_check_deadline_none_when_queue_empty() { + let (leader, _ctx, _role_tx, _raft_tx) = setup_with_threshold(Duration::from_secs(30)).await; + + assert!( + leader.stale_check_deadline.is_none(), + "Deadline must be None when no learners are pending" + ); +} + +// ============================================================================ +// Test 2: Deadline set to ready_since + threshold on first enqueue +// ============================================================================ + +#[tokio::test] +async fn test_stale_check_deadline_set_on_first_enqueue() { + let threshold = Duration::from_secs(30); + let (mut leader, _ctx, _role_tx, _raft_tx) = setup_with_threshold(threshold).await; + + let before = Instant::now(); + leader.pending_promotions.push_back(PendingPromotion::new(10, Instant::now())); + leader.refresh_stale_deadline(threshold); + let after = Instant::now(); + + let deadline = leader.stale_check_deadline.expect("Deadline must be set after first enqueue"); + assert!( + deadline >= before + threshold && deadline <= after + threshold, + "Deadline should be approximately ready_since + threshold" + ); +} + +// ============================================================================ +// Test 3: Deadline NOT overwritten when second learner enqueued +// (VecDeque is FIFO — front is always the oldest, so min() == front) +// ============================================================================ + +#[tokio::test] +async fn test_stale_check_deadline_unchanged_on_second_enqueue() { + let threshold = Duration::from_secs(30); + let (mut leader, _ctx, _role_tx, _raft_tx) = setup_with_threshold(threshold).await; + + // Enqueue first learner + let t0 = Instant::now(); + leader.pending_promotions.push_back(PendingPromotion::new(10, t0)); + leader.refresh_stale_deadline(threshold); + let first_deadline = leader.stale_check_deadline.unwrap(); + + // Enqueue second learner slightly later + tokio::time::sleep(Duration::from_millis(10)).await; + leader.pending_promotions.push_back(PendingPromotion::new(11, Instant::now())); + leader.refresh_stale_deadline(threshold); + + assert_eq!( + leader.stale_check_deadline.unwrap(), + first_deadline, + "Deadline must not change when second learner is enqueued — front is still oldest" + ); +} + +// ============================================================================ +// Test 4: tick() skips stale check when deadline has not fired +// ============================================================================ + +#[tokio::test] +async fn test_tick_skips_stale_check_before_deadline() { + let threshold = Duration::from_secs(3600); // Far future + let (mut leader, mut ctx, role_tx, raft_tx) = setup_with_threshold(threshold).await; + + let mut membership = MockMembership::new(); + // get_node_status must NOT be called — stale check should be skipped + membership.expect_get_node_status().times(0).returning(|_| None); + ctx.membership = Arc::new(membership); + + // Enqueue a learner and set a far-future deadline + leader.pending_promotions.push_back(PendingPromotion::new(10, Instant::now())); + leader.refresh_stale_deadline(threshold); + + // Call tick() — deadline is far in the future, stale check must be skipped + let result = leader.tick(&role_tx, &raft_tx, &ctx).await; + assert!(result.is_ok()); + // If MockMembership::get_node_status were called, mockall would panic — test passes by not panicking +} + +// ============================================================================ +// Test 5: tick() triggers stale learner removal when deadline fires +// ============================================================================ + +#[tokio::test] +async fn test_tick_removes_stale_learner_when_deadline_fires() { + use d_engine_proto::common::{NodeRole::Follower, NodeStatus}; + use d_engine_proto::server::cluster::NodeMeta; + + let threshold = Duration::from_millis(1); // Expires almost immediately + let (_shutdown_tx, shutdown_rx) = watch::channel(()); + let mut cfg = node_config("/tmp/test_tick_removes_stale_learner"); + cfg.raft.membership.promotion.stale_learner_threshold = threshold; + cfg.raft.membership.verify_leadership_persistent_timeout = Duration::from_millis(50); + let mut ctx = MockBuilder::new(shutdown_rx).with_node_config(cfg).build_context(); + + let mut membership = MockMembership::new(); + // Learner 10 has status Promotable — qualifies as stale + membership.expect_get_node_status().returning(|_| Some(NodeStatus::Promotable)); + 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(())); + ctx.membership = Arc::new(membership); + + let captured = Arc::new(Mutex::new(Vec::::new())); + let captured_clone = captured.clone(); + ctx.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)); + ctx.storage.raft_log = Arc::new(raft_log); + + let node_config = ctx.node_config(); + let mut leader = LeaderState::::new(1, node_config); + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + let (raft_tx, _raft_rx) = mpsc::channel(1); + + // Enqueue learner with a ready_since in the past so it's already stale + let past = Instant::now() - Duration::from_secs(1); + leader.pending_promotions.push_back(PendingPromotion::new(10, past)); + // Deadline = past + 1ms — already fired + leader.stale_check_deadline = Some(past + threshold); + + let result = leader.tick(&role_tx, &raft_tx, &ctx).await; + assert!(result.is_ok()); + + // Verify a config change (BatchRemove or RemoveNode) was proposed for node 10 + let payloads = captured.lock(); + assert!( + !payloads.is_empty(), + "tick() must propose removal of stale learner" + ); +} + +// ============================================================================ +// Test 6: drain_batch clears deadline when queue becomes empty +// ============================================================================ + +#[tokio::test] +async fn test_drain_batch_clears_deadline_when_queue_empty() { + let threshold = Duration::from_secs(30); + let (mut leader, _ctx, _role_tx, _raft_tx) = setup_with_threshold(threshold).await; + + leader.pending_promotions.push_back(PendingPromotion::new(10, Instant::now())); + leader.refresh_stale_deadline(threshold); + assert!(leader.stale_check_deadline.is_some()); + + // Drain the only entry + let _batch = leader.drain_batch(1); + leader.refresh_stale_deadline(threshold); + + assert!( + leader.stale_check_deadline.is_none(), + "Deadline must reset to None when queue is drained to empty" + ); +} + +// ============================================================================ +// Test 7: drain_batch refreshes deadline to new front when items remain +// ============================================================================ + +#[tokio::test] +async fn test_drain_batch_refreshes_deadline_to_new_front() { + let threshold = Duration::from_secs(30); + let (mut leader, _ctx, _role_tx, _raft_tx) = setup_with_threshold(threshold).await; + + let t0 = Instant::now() - Duration::from_secs(5); // older + let t1 = Instant::now() - Duration::from_secs(2); // newer (becomes new front after drain) + + leader.pending_promotions.push_back(PendingPromotion::new(10, t0)); + leader.pending_promotions.push_back(PendingPromotion::new(11, t1)); + leader.refresh_stale_deadline(threshold); + + let initial_deadline = leader.stale_check_deadline.unwrap(); + // Should be t0 + threshold (oldest front) + assert!( + (initial_deadline - threshold).saturating_duration_since(t0) < Duration::from_millis(50), + "Initial deadline should correspond to t0" + ); + + // Drain only the first (t0 node) + let _batch = leader.drain_batch(1); + leader.refresh_stale_deadline(threshold); + + let new_deadline = leader.stale_check_deadline.unwrap(); + // Should now be t1 + threshold (new front) + assert!( + (new_deadline - threshold).saturating_duration_since(t1) < Duration::from_millis(50), + "After drain, deadline should correspond to new front t1" + ); + assert_ne!( + new_deadline, initial_deadline, + "Deadline must change after drain to reflect new front" + ); +} diff --git a/d-engine-core/src/raft_role/mod.rs b/d-engine-core/src/raft_role/mod.rs index f02ba17a..6fea0855 100644 --- a/d-engine-core/src/raft_role/mod.rs +++ b/d-engine-core/src/raft_role/mod.rs @@ -367,6 +367,18 @@ impl RaftRole { let _ = self.state_mut().update_next_index(peer_id, match_idx + 1); } + pub(crate) async fn handle_zombie_detected( + &mut self, + node_id: u32, + role_tx: &mpsc::UnboundedSender, + ctx: &RaftContext, + ) -> Result<()> + where + T: TypeConfig, + { + self.state_mut().handle_zombie_detected(node_id, role_tx, ctx).await + } + pub(crate) fn handle_snapshot_push_completed( &mut self, peer_id: u32, diff --git a/d-engine-core/src/raft_role/raft_role_test.rs b/d-engine-core/src/raft_role/raft_role_test.rs index d4002053..222d6d1d 100644 --- a/d-engine-core/src/raft_role/raft_role_test.rs +++ b/d-engine-core/src/raft_role/raft_role_test.rs @@ -1,8 +1,12 @@ use std::{sync::Arc, thread}; use d_engine_proto::server::election::VotedFor; +use tokio::sync::{mpsc, watch}; use super::super::*; +use super::candidate_state::CandidateState; +use super::follower_state::FollowerState; +use super::learner_state::LearnerState; #[test] fn test_voted_for_backward_compatibility() { @@ -686,3 +690,76 @@ fn test_update_voted_for_learner_discovers_leader() { // Should trigger event (first discovery) assert!(is_new, "Learner first leader discovery should return true"); } + +// ============================================================================ +// handle_zombie_detected — non-leader roles must be no-ops +// +// Only the leader can propose membership changes (BatchRemove). When a +// ZombieDetected signal arrives while the node is a Follower, Candidate, or +// Learner the call must return Ok(()) immediately without emitting any event on +// role_tx. Recovery from the lost signal is handled by the health-monitor +// counter-reset mechanism (see health_monitor_test). +// ============================================================================ + +fn make_test_config() -> Arc { + Arc::new( + RaftNodeConfig::new() + .expect("RaftNodeConfig::new") + .validate() + .expect("RaftNodeConfig::validate"), + ) +} + +fn make_test_context() -> (RaftContext, watch::Sender<()>) { + let (shutdown_tx, shutdown_rx) = watch::channel(()); + let ctx = MockBuilder::new(shutdown_rx).build_context(); + (ctx, shutdown_tx) +} + +#[tokio::test] +async fn test_handle_zombie_detected_is_noop_on_follower() { + let cfg = make_test_config(); + let mut role = RaftRole::Follower(Box::new(FollowerState::new(1, cfg, None, None))); + let (ctx, _shutdown_tx) = make_test_context(); + let (role_tx, mut role_rx) = mpsc::unbounded_channel::(); + + let result = role.handle_zombie_detected(42, &role_tx, &ctx).await; + + assert!(result.is_ok(), "Follower must not error on ZombieDetected"); + assert!( + role_rx.try_recv().is_err(), + "Follower must not emit any RoleEvent for ZombieDetected" + ); +} + +#[tokio::test] +async fn test_handle_zombie_detected_is_noop_on_candidate() { + let cfg = make_test_config(); + let mut role = RaftRole::Candidate(Box::new(CandidateState::new(1, cfg))); + let (ctx, _shutdown_tx) = make_test_context(); + let (role_tx, mut role_rx) = mpsc::unbounded_channel::(); + + let result = role.handle_zombie_detected(42, &role_tx, &ctx).await; + + assert!(result.is_ok(), "Candidate must not error on ZombieDetected"); + assert!( + role_rx.try_recv().is_err(), + "Candidate must not emit any RoleEvent for ZombieDetected" + ); +} + +#[tokio::test] +async fn test_handle_zombie_detected_is_noop_on_learner() { + let cfg = make_test_config(); + let mut role = RaftRole::Learner(Box::new(LearnerState::new(1, cfg))); + let (ctx, _shutdown_tx) = make_test_context(); + let (role_tx, mut role_rx) = mpsc::unbounded_channel::(); + + let result = role.handle_zombie_detected(42, &role_tx, &ctx).await; + + assert!(result.is_ok(), "Learner must not error on ZombieDetected"); + assert!( + role_rx.try_recv().is_err(), + "Learner must not emit any RoleEvent for ZombieDetected" + ); +} diff --git a/d-engine-core/src/raft_role/role_state.rs b/d-engine-core/src/raft_role/role_state.rs index da102cd1..a8e699b1 100644 --- a/d-engine-core/src/raft_role/role_state.rs +++ b/d-engine-core/src/raft_role/role_state.rs @@ -86,6 +86,17 @@ pub trait RaftRoleState: Send + Sync + 'static { Err(MembershipError::NotLeader.into()) } + /// Handle a ZombieDetected signal. No-op for non-leader roles. + async fn handle_zombie_detected( + &mut self, + _node_id: u32, + _role_tx: &mpsc::UnboundedSender, + _ctx: &RaftContext, + ) -> Result<()> { + // Default: no-op for non-leader roles + Ok(()) + } + /// Update per-peer snapshot push backoff state and emit an alert when persistent /// failures exceed the configured threshold. No-op for non-leader roles. fn handle_snapshot_push_completed( diff --git a/d-engine-core/src/storage/buffered_raft_log_test/flush_strategy_test.rs b/d-engine-core/src/storage/buffered_raft_log_test/flush_strategy_test.rs index e58dedd1..44fd0040 100644 --- a/d-engine-core/src/storage/buffered_raft_log_test/flush_strategy_test.rs +++ b/d-engine-core/src/storage/buffered_raft_log_test/flush_strategy_test.rs @@ -194,53 +194,6 @@ async fn test_mem_first_flushes_asynchronously() { ); } -/// Test MemFirst loses unflushed data on crash -/// -/// # Scenario -/// - Append entries without flush -/// - Simulate crash -/// - Expected: Unflushed data lost -#[tokio::test] -async fn test_mem_first_loses_unflushed_data_on_crash() { - let original_ctx = BufferedRaftLogTestContext::new( - PersistenceStrategy::MemFirst, - FlushPolicy::Batch { - idle_flush_interval_ms: 1000, - }, - "test_mem_first_data_loss", - ); - - // Arrange: Append entries but don't flush - for i in 1..=3 { - original_ctx - .raft_log - .append_entries(vec![Entry { - index: i, - term: 1, - payload: Some(EntryPayload::command(Bytes::from(format!("data{i}")))), - }]) - .await - .unwrap(); - } - - // Verify entries in memory but not durable - assert_eq!(original_ctx.raft_log.len(), 3); - assert_eq!(original_ctx.raft_log.durable_index(), 0); - - // Act: Simulate crash without flush - let recovered_ctx = original_ctx.recover_from_crash(); - drop(original_ctx); - sleep(Duration::from_millis(50)).await; - - // Assert: Unflushed data lost - assert_eq!( - recovered_ctx.raft_log.durable_index(), - 0, - "Unflushed data should be lost" - ); - assert_eq!(recovered_ctx.raft_log.len(), 0); -} - /// Test MemFirst concurrent buffering is safe /// /// # Scenario diff --git a/d-engine-core/src/test_utils/mock/mock_raft_builder.rs b/d-engine-core/src/test_utils/mock/mock_raft_builder.rs index 987d3942..e01be03d 100644 --- a/d-engine-core/src/test_utils/mock/mock_raft_builder.rs +++ b/d-engine-core/src/test_utils/mock/mock_raft_builder.rs @@ -405,7 +405,6 @@ pub fn mock_membership() -> MockMembership { current_leader_id: None, }); membership.expect_get_cluster_conf_version().returning(|| 1); - membership.expect_get_zombie_candidates().returning(Vec::new); membership.expect_get_peers_id_with_condition().returning(|_| vec![]); // Mock single-node cluster detection (default to multi-node with no peers) diff --git a/d-engine-server/src/membership/raft_membership.rs b/d-engine-server/src/membership/raft_membership.rs index 2927aa2c..712aa20e 100644 --- a/d-engine-server/src/membership/raft_membership.rs +++ b/d-engine-server/src/membership/raft_membership.rs @@ -700,10 +700,6 @@ where info!("Config change applied at index {}", index); } - async fn get_zombie_candidates(&self) -> Vec { - self.health_monitor.get_zombie_candidates().await - } - async fn can_rejoin( &self, node_id: u32, @@ -727,23 +723,41 @@ impl RaftMembership where T: TypeConfig, { + /// Returns `(Self, zombie_rx)`. Caller must pass `zombie_rx` to the Raft event loop + /// so that `ZombieDetected` signals reach the leader. pub(crate) fn new( node_id: u32, initial_nodes: Vec, config: RaftNodeConfig, - ) -> Self { + ) -> (Self, tokio::sync::mpsc::Receiver) { let zombie_threshold = config.raft.membership.zombie.threshold; let connection_cache = ConnectionCache::new(config.network.clone()); let initial_cluster_size = initial_nodes.len(); - Self { - node_id, - membership: MembershipGuard::new(initial_nodes, 0), - config, - initial_cluster_size, - _phantom: PhantomData, - health_monitor: RaftHealthMonitor::new(zombie_threshold), - connection_cache, - } + let (health_monitor, zombie_rx) = RaftHealthMonitor::new(zombie_threshold); + ( + Self { + node_id, + membership: MembershipGuard::new(initial_nodes, 0), + config, + initial_cluster_size, + _phantom: PhantomData, + health_monitor, + connection_cache, + }, + zombie_rx, + ) + } + + /// Returns true if the zombie signal for `node_id` should still be acted on. + /// + /// Delegates to the inner health monitor. The bridge task calls this before + /// forwarding a `ZombieDetected` signal to the Raft event loop so that a + /// stale signal (peer already recovered via `record_success`) is dropped. + pub(crate) fn is_zombie_valid( + &self, + node_id: u32, + ) -> bool { + self.health_monitor.is_zombie_valid(node_id) } /// Updates a single node atomically diff --git a/d-engine-server/src/membership/raft_membership_test.rs b/d-engine-server/src/membership/raft_membership_test.rs index 816851d1..e6260189 100644 --- a/d-engine-server/src/membership/raft_membership_test.rs +++ b/d-engine-server/src/membership/raft_membership_test.rs @@ -71,6 +71,7 @@ pub fn create_test_membership() initial_cluster, RaftNodeConfig::default(), ) + .0 } #[tokio::test] @@ -209,11 +210,12 @@ async fn test_replication_peers_case1() { status: NodeStatus::Promotable.into(), }, ]; - let membership = RaftMembership::>::new( - 1, - initial_cluster, - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + initial_cluster, + RaftNodeConfig::default(), + ); assert_eq!(membership.replication_peers().await.len(), 4); } @@ -250,11 +252,12 @@ async fn test_remove_node_allows_leader_removal() { }, ]; - let membership = RaftMembership::>::new( - 1, - initial_cluster, - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + initial_cluster, + RaftNodeConfig::default(), + ); // Verify node 1 exists assert!(membership.contains_node(1).await); @@ -311,11 +314,12 @@ async fn test_retrieve_cluster_membership_config() { status: NodeStatus::Active.into(), }, ]; - let membership = RaftMembership::>::new( - 1, - initial_cluster, - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + initial_cluster, + RaftNodeConfig::default(), + ); let r = membership.retrieve_cluster_membership_config(None).await; assert_eq!(r.nodes.len(), 5); @@ -326,7 +330,8 @@ async fn test_retrieve_cluster_membership_config() { #[traced_test] async fn test_update_cluster_conf_from_leader_case1() { // Test AddNode operation - let membership = RaftMembership::::new(1, vec![], RaftNodeConfig::default()); + let (membership, _) = + RaftMembership::::new(1, vec![], RaftNodeConfig::default()); let req = ClusterConfChangeRequest { id: 2, term: 1, @@ -360,7 +365,7 @@ async fn test_update_cluster_conf_from_leader_case1() { #[traced_test] async fn test_update_cluster_conf_from_leader_case2() { // Test RemoveNode operation - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![NodeMeta { id: 3, @@ -399,7 +404,7 @@ async fn test_update_cluster_conf_from_leader_case2() { #[traced_test] async fn test_update_cluster_conf_from_leader_case3() { // Test PromoteLearner operation - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![NodeMeta { id: 3, @@ -444,7 +449,7 @@ async fn test_update_cluster_conf_from_leader_case3() { #[traced_test] async fn test_update_cluster_conf_from_leader_case4_conf_invalid_promotion() { // Try to promote non-learner node - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![NodeMeta { id: 3, @@ -489,7 +494,8 @@ async fn test_update_cluster_conf_from_leader_case4_conf_invalid_promotion() { #[traced_test] async fn test_update_cluster_conf_from_leader_case5_conf_missing_change() { // Test missing change type - let membership = RaftMembership::::new(1, vec![], RaftNodeConfig::default()); + let (membership, _) = + RaftMembership::::new(1, vec![], RaftNodeConfig::default()); let req = ClusterConfChangeRequest { id: 2, term: 1, @@ -521,7 +527,8 @@ async fn test_update_cluster_conf_from_leader_case5_conf_missing_change() { #[traced_test] async fn test_update_cluster_conf_from_leader_case6_conf_version_mismatch() { // Test version mismatch - let membership = RaftMembership::::new(1, vec![], RaftNodeConfig::default()); + let (membership, _) = + RaftMembership::::new(1, vec![], RaftNodeConfig::default()); membership.update_conf_version(5).await; // Set current version to 5 let req = ClusterConfChangeRequest { @@ -583,7 +590,7 @@ async fn test_batch_remove_nodes() { }, ]; - let membership = + let (membership, _) = RaftMembership::::new(1, initial_nodes, RaftNodeConfig::default()); membership.update_conf_version(1).await; @@ -639,7 +646,7 @@ async fn test_batch_remove_leader_protection() { }, ]; - let membership = + let (membership, _) = RaftMembership::::new(1, initial_nodes, RaftNodeConfig::default()); // Attempt to remove leader in batch @@ -669,7 +676,7 @@ async fn test_batch_remove_leader_protection() { #[tokio::test] #[traced_test] async fn test_apply_batch_remove_config_change() { - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![ NodeMeta { @@ -738,7 +745,7 @@ async fn test_update_cluster_conf_from_leader_case7_batch_remove() { status: NodeStatus::Active as i32, }, ]; - let membership = + let (membership, _) = RaftMembership::::new(1, initial_nodes, RaftNodeConfig::default()); membership.update_conf_version(1).await; @@ -770,7 +777,8 @@ async fn test_update_cluster_conf_from_leader_case7_batch_remove() { #[traced_test] async fn test_update_cluster_conf_from_leader_case8_batch_remove_nonexistent() { // Tests graceful handling of non-existent nodes in batch - let membership = RaftMembership::::new(1, vec![], RaftNodeConfig::default()); + let (membership, _) = + RaftMembership::::new(1, vec![], RaftNodeConfig::default()); let req = ClusterConfChangeRequest { id: 1, @@ -834,11 +842,12 @@ async fn test_get_peers_id_with_condition() { status: NodeStatus::Active.into(), }, ]; - let membership = RaftMembership::>::new( - 1, - initial_cluster, - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + initial_cluster, + RaftNodeConfig::default(), + ); // Test 1: Filter followers and candidates let mut result = membership @@ -891,19 +900,20 @@ mod check_cluster_is_ready_test { "d_engine.server.cluster.ClusterManagementService".to_string(); // Create membership with 2 peers - let membership = RaftMembership::>::new( - 1, - peer_ids - .iter() - .map(|id| NodeMeta { - id: *id, - address: "127.0.0.1:0".to_string(), - role: Follower as i32, - status: NodeStatus::Active.into(), - }) - .collect(), - config, - ); + let (membership, _) = + RaftMembership::>::new( + 1, + peer_ids + .iter() + .map(|id| NodeMeta { + id: *id, + address: "127.0.0.1:0".to_string(), + role: Follower as i32, + status: NodeStatus::Active.into(), + }) + .collect(), + config, + ); // Create mock services for peers for &id in &peer_ids { @@ -934,16 +944,17 @@ mod check_cluster_is_ready_test { let mut config = RaftNodeConfig::default(); config.retry.membership.max_retries = 1; - let membership = RaftMembership::>::new( - 1, - vec![NodeMeta { - id: 2, - address: "127.0.0.1:9999".to_string(), // Invalid port - role: Follower as i32, - status: NodeStatus::Active.into(), - }], - config, - ); + let (membership, _) = + RaftMembership::>::new( + 1, + vec![NodeMeta { + id: 2, + address: "127.0.0.1:9999".to_string(), // Invalid port + role: Follower as i32, + status: NodeStatus::Active.into(), + }], + config, + ); let result = membership.check_cluster_is_ready().await; assert!(result.is_err(), "Should return error for unreachable peer"); @@ -958,16 +969,17 @@ mod check_cluster_is_ready_test { let mut config = RaftNodeConfig::default(); config.retry.membership.max_retries = 1; - let membership = RaftMembership::>::new( - 1, - vec![NodeMeta { - id: 4, - address: format!("127.0.0.1:{port}",), - role: Follower as i32, - status: NodeStatus::Active.into(), - }], - config, - ); + let (membership, _) = + RaftMembership::>::new( + 1, + vec![NodeMeta { + id: 4, + address: format!("127.0.0.1:{port}",), + role: Follower as i32, + status: NodeStatus::Active.into(), + }], + config, + ); let result = membership.check_cluster_is_ready().await; println!("Result: {:?}", &result); @@ -985,19 +997,20 @@ mod check_cluster_is_ready_test { // Create membership with 2 peers let mut config = RaftNodeConfig::default(); config.retry.membership.max_retries = 1; - let membership = RaftMembership::>::new( - 1, - peer_ids - .iter() - .map(|id| NodeMeta { - id: *id, - address: "127.0.0.1:0".to_string(), - role: Follower as i32, - status: NodeStatus::Active.into(), - }) - .collect(), - config, - ); + let (membership, _) = + RaftMembership::>::new( + 1, + peer_ids + .iter() + .map(|id| NodeMeta { + id: *id, + address: "127.0.0.1:0".to_string(), + role: Follower as i32, + status: NodeStatus::Active.into(), + }) + .collect(), + config, + ); // Create mock services with different responses for id in peer_ids.iter() { @@ -1024,11 +1037,12 @@ mod add_learner_test { #[tokio::test] async fn test_add_learner_case1() { - let membership = RaftMembership::>::new( - 1, - vec![], - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + vec![], + RaftNodeConfig::default(), + ); let result = membership .add_learner(2, "127.0.0.1:1234".to_string(), NodeStatus::Promotable) @@ -1045,16 +1059,17 @@ mod add_learner_test { #[tokio::test] async fn test_add_learner_case2() { - let membership = RaftMembership::>::new( - 1, - vec![NodeMeta { - id: 1, - address: "127.0.0.1:0".to_string(), - role: Follower as i32, - status: NodeStatus::Active.into(), - }], - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + vec![NodeMeta { + id: 1, + address: "127.0.0.1:0".to_string(), + role: Follower as i32, + status: NodeStatus::Active.into(), + }], + RaftNodeConfig::default(), + ); let result = membership .add_learner(1, "127.0.0.1:1234".to_string(), NodeStatus::Promotable) @@ -1087,7 +1102,7 @@ async fn test_health_monitoring_integration() { let mut config = RaftNodeConfig::default(); config.raft.membership.zombie.threshold = 2; // Set low threshold for testing - let membership = RaftMembership::::new(1, vec![], config); + let (membership, mut zombie_rx) = RaftMembership::::new(1, vec![], config); // Add test node membership @@ -1103,10 +1118,16 @@ async fn test_health_monitoring_integration() { Some(1) ); - // Test 2: Record second failure (should become zombie candidate) + // Test 2: Record second failure (should fire ZombieDetected signal on zombie_rx) membership.get_peer_channel(100, ConnectionType::Control).await; - let zombies = membership.get_zombie_candidates().await; - assert_eq!(zombies, vec![100]); + // The health monitor sends node_id to zombie_rx when threshold is reached. + // Use try_recv to confirm the signal was sent without blocking. + let zombie_signal = zombie_rx.try_recv(); + assert_eq!( + zombie_signal.ok(), + Some(100), + "ZombieDetected signal should fire for node 100" + ); // Test 3: Record success resets failures // Update to valid address (using mock would be better in real impl) @@ -1165,11 +1186,12 @@ mod pre_warm_connections_tests { } } - let membership = RaftMembership::>::new( - 1, - cluster, - RaftNodeConfig::default(), - ); + let (membership, _) = + RaftMembership::>::new( + 1, + cluster, + RaftNodeConfig::default(), + ); (membership, shutdown_channels) } @@ -1256,7 +1278,7 @@ mod single_node_tests { role: Follower as i32, status: NodeStatus::Active.into(), }]; - RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()) + RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()).0 } /// Helper: Create 3-node membership @@ -1282,7 +1304,7 @@ mod single_node_tests { status: NodeStatus::Active.into(), }, ]; - RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()) + RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()).0 } /// Helper: Create 5-node membership @@ -1320,7 +1342,7 @@ mod single_node_tests { status: NodeStatus::Active.into(), }, ]; - RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()) + RaftMembership::new(1, initial_cluster, RaftNodeConfig::default()).0 } #[tokio::test] @@ -1491,7 +1513,7 @@ mod can_rejoin_tests { /// Test 1: Reject non-learner role #[tokio::test] async fn test_can_rejoin_rejects_non_learner() { - let membership = + let (membership, _) = RaftMembership::::new(1, vec![], RaftNodeConfig::default()); // Try to join as Follower (should fail) @@ -1522,7 +1544,7 @@ mod can_rejoin_tests { /// Test 2: Allow learner to join when node_id does not exist #[tokio::test] async fn test_can_rejoin_allows_new_learner() { - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![NodeMeta { id: 2, @@ -1541,7 +1563,7 @@ mod can_rejoin_tests { /// Test 3: Reject learner when node_id already exists #[tokio::test] async fn test_can_rejoin_rejects_existing_node() { - let membership = RaftMembership::::new( + let (membership, _) = RaftMembership::::new( 1, vec![ NodeMeta { diff --git a/d-engine-server/src/network/health_monitor.rs b/d-engine-server/src/network/health_monitor.rs index 19d4be73..0c7a4307 100644 --- a/d-engine-server/src/network/health_monitor.rs +++ b/d-engine-server/src/network/health_monitor.rs @@ -2,6 +2,7 @@ use async_trait::async_trait; use dashmap::DashMap; #[cfg(test)] use mockall::automock; +use tokio::sync::mpsc; #[cfg_attr(test, automock)] #[async_trait] @@ -14,20 +15,29 @@ pub(crate) trait HealthMonitor: Send + Sync + 'static { &self, peer_id: u32, ); - async fn get_zombie_candidates(&self) -> Vec; } pub(crate) struct RaftHealthMonitor { pub(crate) failure_counts: DashMap, pub(crate) zombie_threshold: u32, + /// Fires node_id when failure_count first reaches zombie_threshold. + /// Consumed by the Raft event loop (core layer) via select!. + /// Server layer holds only Sender — zero dependency on core types. + zombie_tx: mpsc::Sender, } impl RaftHealthMonitor { - pub(crate) fn new(zombie_threshold: u32) -> Self { - RaftHealthMonitor { - failure_counts: DashMap::new(), - zombie_threshold, - } + /// Returns (monitor, zombie_rx). Caller passes zombie_rx to the Raft event loop. + pub(crate) fn new(zombie_threshold: u32) -> (Self, mpsc::Receiver) { + let (zombie_tx, zombie_rx) = mpsc::channel(64); + ( + RaftHealthMonitor { + failure_counts: DashMap::new(), + zombie_threshold, + zombie_tx, + }, + zombie_rx, + ) } } @@ -37,22 +47,44 @@ impl HealthMonitor for RaftHealthMonitor { &self, node_id: u32, ) { - let mut count = self.failure_counts.entry(node_id).or_insert(0); - *count += 1; + let new_count = { + let mut count = self.failure_counts.entry(node_id).or_insert(0); + *count += 1; + *count + }; + // Signal when the threshold is first crossed, then reset the counter to 0 + // so that a second batch of failures re-triggers if this signal is lost + // (e.g. consumed by a non-leader no-op or dropped due to backpressure). + if new_count == self.zombie_threshold { + // Reset counter via a fresh get_mut (original borrow ended above). + // Guard is dropped before the await point to avoid holding it across yield. + if let Some(mut c) = self.failure_counts.get_mut(&node_id) { + *c = 0; + } + // send().await: waits for channel space instead of silently dropping. + // Only errors on receiver drop (node shutting down), which is harmless. + let _ = self.zombie_tx.send(node_id).await; + } } async fn record_success( &self, node_id: u32, ) { - self.failure_counts.remove(&node_id); // Reset failure counter + self.failure_counts.remove(&node_id); } +} - async fn get_zombie_candidates(&self) -> Vec { - self.failure_counts - .iter() - .filter(|entry| *entry.value() >= self.zombie_threshold) - .map(|entry| *entry.key()) - .collect() +impl RaftHealthMonitor { + /// Returns true if the zombie signal for `node_id` is still valid. + /// + /// A zombie is invalid once `record_success` has been called (peer recovered), + /// which removes the entry from `failure_counts`. The bridge task uses this + /// to drop stale zombie signals before forwarding them to the Raft event loop. + pub(crate) fn is_zombie_valid( + &self, + node_id: u32, + ) -> bool { + self.failure_counts.contains_key(&node_id) } } diff --git a/d-engine-server/src/network/health_monitor_test.rs b/d-engine-server/src/network/health_monitor_test.rs index 95f59272..d16ffd69 100644 --- a/d-engine-server/src/network/health_monitor_test.rs +++ b/d-engine-server/src/network/health_monitor_test.rs @@ -5,7 +5,7 @@ use super::*; #[tokio::test] #[traced_test] async fn test_record_failure_increments_count() { - let monitor = RaftHealthMonitor::new(3); + let (monitor, _rx) = RaftHealthMonitor::new(3); let node_id = 42; monitor.record_failure(node_id).await; @@ -17,7 +17,7 @@ async fn test_record_failure_increments_count() { #[tokio::test] #[traced_test] async fn test_record_success_resets_count() { - let monitor = RaftHealthMonitor::new(3); + let (monitor, _rx) = RaftHealthMonitor::new(3); let node_id = 7; monitor.record_failure(node_id).await; @@ -29,16 +29,109 @@ async fn test_record_success_resets_count() { #[tokio::test] #[traced_test] -async fn test_get_zombie_candidates() { - let monitor = RaftHealthMonitor::new(2); +async fn test_zombie_signal_fires_at_threshold() { + let (monitor, mut rx) = RaftHealthMonitor::new(2); let node1 = 1; let node2 = 2; + // node1: one failure — below threshold, no signal monitor.record_failure(node1).await; - monitor.record_failure(node1).await; + assert!(rx.try_recv().is_err(), "no signal before threshold"); + + // node2: one failure — below threshold monitor.record_failure(node2).await; - let zombies = monitor.get_zombie_candidates().await; - assert!(zombies.contains(&node1)); - assert!(!zombies.contains(&node2)); - assert_eq!(zombies.len(), 1); + assert!(rx.try_recv().is_err(), "no signal before threshold"); + + // node1: second failure — hits threshold, signal fires + monitor.record_failure(node1).await; + let detected = rx.try_recv().expect("signal must fire at threshold"); + assert_eq!(detected, node1); + + // node1: third failure — already past threshold, no duplicate signal + monitor.record_failure(node1).await; + assert!( + rx.try_recv().is_err(), + "no duplicate signal after threshold" + ); +} + +#[tokio::test] +#[traced_test] +async fn test_record_success_clears_failure_count() { + let (monitor, mut rx) = RaftHealthMonitor::new(2); + let node_id = 7; + + monitor.record_failure(node_id).await; + monitor.record_success(node_id).await; + + // After success the counter is gone; one more failure does not cross threshold + monitor.record_failure(node_id).await; + assert!(rx.try_recv().is_err(), "counter reset by success"); +} + +// ============================================================================ +// Bug 1 + Bug 3: zombie signal must re-trigger after the first signal is lost +// +// Scenario: ZombieDetected was consumed by a non-leader (no-op) or dropped due +// to channel backpressure. Because record_failure() only fires at exactly +// new_count == threshold, the counter must be reset to 0 after each signal so +// that the next batch of failures can re-trigger the signal. +// +// Current behaviour (FAIL): counter stays at threshold; new_count > threshold +// forever; no second signal is ever emitted. +// Expected behaviour (PASS after fix): counter resets to 0; two more failures +// cross the threshold again and emit a second signal. +// ============================================================================ +#[tokio::test] +#[traced_test] +async fn test_zombie_signal_re_triggers_after_signal_lost() { + let (monitor, mut rx) = RaftHealthMonitor::new(2); + let node_id = 1; + + // First threshold crossing — signal emitted and immediately consumed. + monitor.record_failure(node_id).await; + monitor.record_failure(node_id).await; + assert_eq!(rx.try_recv().ok(), Some(node_id), "first signal must fire"); + + // Simulate: signal was dropped (non-leader no-op or try_send full). + // The peer is still failing; the counter should be back at 0 so that + // two more failures re-cross the threshold. + monitor.record_failure(node_id).await; + monitor.record_failure(node_id).await; + assert_eq!( + rx.try_recv().ok(), + Some(node_id), + "zombie signal must re-trigger after counter reset — peer is still failing" + ); +} + +// ============================================================================ +// Bug 2: zombie signal must be invalidated when the peer recovers +// +// Scenario: ZombieDetected was queued in the channel, but before the bridge +// task forwards it to the Raft event loop the peer reconnects successfully. +// The bridge task calls is_zombie_valid() to avoid proposing BatchRemove for a +// healthy node. +// +// Expected: is_zombie_valid() returns false after record_success(). +// ============================================================================ +#[tokio::test] +#[traced_test] +async fn test_zombie_revoked_by_recovery() { + let (monitor, mut rx) = RaftHealthMonitor::new(2); + let node_id = 1; + + // Threshold crossed — signal queued in the channel. + monitor.record_failure(node_id).await; + monitor.record_failure(node_id).await; + assert_eq!(rx.try_recv().ok(), Some(node_id), "signal must be queued"); + + // Peer recovers before the bridge task forwards the signal. + monitor.record_success(node_id).await; + + // Bridge task should now drop the stale signal. + assert!( + !monitor.is_zombie_valid(node_id), + "zombie must be revoked by record_success — bridge task must not forward it" + ); } diff --git a/d-engine-server/src/node/builder.rs b/d-engine-server/src/node/builder.rs index 5d934d2e..bc103994 100644 --- a/d-engine-server/src/node/builder.rs +++ b/d-engine-server/src/node/builder.rs @@ -50,6 +50,7 @@ use d_engine_core::RaftRole; use d_engine_core::RaftStorageHandles; use d_engine_core::ReplicationHandler; use d_engine_core::Result; +use d_engine_core::RoleEvent; use d_engine_core::SignalParams; use d_engine_core::StateMachine; use d_engine_core::StateMachineWorker; @@ -376,13 +377,21 @@ where watch_event_tx, )) }); - let membership = Arc::new(self.membership.take().unwrap_or_else(|| { - RaftMembership::new( - node_id, - node_config.cluster.initial_cluster.clone(), - node_config.clone(), - ) - })); + let (membership_inner, zombie_rx) = self.membership.take().map_or_else( + || { + RaftMembership::new( + node_id, + node_config.cluster.initial_cluster.clone(), + node_config.clone(), + ) + }, + // Pre-built membership has no zombie_rx; create a dummy closed channel. + |m| { + let (_tx, rx) = tokio::sync::mpsc::channel(1); + (m, rx) + }, + ); + let membership = Arc::new(membership_inner); let purge_executor = DefaultPurgeExecutor::new(raft_log.clone()); @@ -392,6 +401,22 @@ where let event_tx_clone = event_tx.clone(); // used in commit handler let role_tx_for_sm = role_tx.clone(); // used in SM worker (ApplyCompleted → P2) + // Bridge zombie signals from health monitor → role event loop. + // Exits naturally when zombie_tx drops with RaftMembership (channel closed → recv None). + // Validates each signal against the health monitor before forwarding: if the peer + // recovered (record_success called) after the signal was queued, drop the stale signal + // to prevent BatchRemove for a healthy node. + let role_tx_for_zombie = role_tx.clone(); + let membership_for_zombie = Arc::clone(&membership); + tokio::spawn(async move { + let mut zombie_rx = zombie_rx; + while let Some(node_id) = zombie_rx.recv().await { + if membership_for_zombie.is_zombie_valid(node_id) { + let _ = role_tx_for_zombie.send(RoleEvent::ZombieDetected(node_id)); + } + } + }); + let node_config_arc = Arc::new(node_config); // Construct my role diff --git a/d-engine-server/src/test_utils/integration/mod.rs b/d-engine-server/src/test_utils/integration/mod.rs index 6f5ee7f8..7a9ed132 100644 --- a/d-engine-server/src/test_utils/integration/mod.rs +++ b/d-engine-server/src/test_utils/integration/mod.rs @@ -249,11 +249,14 @@ pub fn setup_raft_components( raft_log: buffered_raft_log, state_machine, transport: Arc::new(grpc_transport), - membership: Arc::new(RaftMembership::new( - id, - arc_node_config.cluster.initial_cluster.clone(), - node_config_clone.clone(), - )), + membership: Arc::new( + RaftMembership::new( + id, + arc_node_config.cluster.initial_cluster.clone(), + node_config_clone.clone(), + ) + .0, + ), election_handler: ElectionHandler::new(id), replication_handler: ReplicationHandler::new(id), node_config: node_config_clone, diff --git a/d-engine-server/src/test_utils/mock/mock_node_builder.rs b/d-engine-server/src/test_utils/mock/mock_node_builder.rs index 7e43ddb6..4913b30e 100644 --- a/d-engine-server/src/test_utils/mock/mock_node_builder.rs +++ b/d-engine-server/src/test_utils/mock/mock_node_builder.rs @@ -616,7 +616,6 @@ pub(crate) fn mock_membership() -> MockMembership { nodes: vec![], current_leader_id: None, }); - membership.expect_get_zombie_candidates().returning(Vec::new); membership.expect_get_peers_id_with_condition().returning(|_| vec![]); membership } diff --git a/deny.toml b/deny.toml index 9562d85a..c2f1545b 100644 --- a/deny.toml +++ b/deny.toml @@ -70,7 +70,6 @@ feature-depth = 1 # A list of advisory IDs to ignore. Note that ignored advisories will still # output a note when they are encountered. ignore = [ - "RUSTSEC-2025-0134", # rustls-pemfile unmaintained, indirect dep via tonic 0.12 "RUSTSEC-2025-0141", # bincode 1.3.3 unmaintained but team considers it complete #"RUSTSEC-0000-0000", #{ id = "RUSTSEC-0000-0000", reason = "you can specify a reason the advisory is ignored" }, @@ -201,9 +200,24 @@ deny = ["GPL-2.0", "GPL-3.0", "AGPL-3.0"] #exact = true # Certain crates/versions that will be skipped when doing duplicate detection. +# These are genuinely unavoidable: different ecosystems pin to different versions +# and no upgrade path exists today. Re-evaluate when upstream dependencies release +# new major versions. skip = [ - #"ansi_term@0.11.0", - #{ crate = "ansi_term@0.11.0", reason = "you can specify a reason why it can't be updated/removed" }, + # ring 0.17 requires getrandom 0.2.x; ring ecosystem is slow to update + { crate = "getrandom@0.2.17", reason = "ring 0.17 pins getrandom 0.2.x, no upgrade path" }, + # rand 0.9 / rand_core 0.9 require getrandom 0.3.x + { crate = "getrandom@0.3.4", reason = "rand_core 0.9 pins getrandom 0.3.x" }, + # r-efi is a transitive dep of the two getrandom variants above + { crate = "r-efi@5.3.0", reason = "transitive dep of getrandom 0.3 (rand_core 0.9)" }, + # dashmap 6.x pins hashbrown 0.14; dashmap 7 is still RC + { crate = "hashbrown@0.14.5", reason = "dashmap 6.1 pins hashbrown 0.14; dashmap 7 not yet stable" }, + # tonic 0.13 pins socket2 0.5 internally + { crate = "socket2@0.5.10", reason = "tonic 0.13 pins socket2 0.5 internally" }, + # ring and socket2 0.5 pull windows-sys 0.52 + { crate = "windows-sys@0.52.0", reason = "transitive dep of ring 0.17 and socket2 0.5" }, + # bindgen (rocksdb build dep) pins itertools 0.13 + { crate = "itertools@0.13.0", reason = "bindgen (rocksdb build dep) pins itertools 0.13" }, ] # Similarly to `skip` allows you to skip certain crates during duplicate # detection. Unlike skip, it also includes the entire tree of transitive diff --git a/examples/client-usage-standalone/Cargo.lock b/examples/client-usage-standalone/Cargo.lock index 70dbe05d..dd944f61 100644 --- a/examples/client-usage-standalone/Cargo.lock +++ b/examples/client-usage-standalone/Cargo.lock @@ -757,21 +757,15 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.16.1" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" dependencies = [ "allocator-api2", "equivalent", "foldhash 0.2.0", ] -[[package]] -name = "hashbrown" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" - [[package]] name = "heck" version = "0.5.0" @@ -964,11 +958,11 @@ checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" [[package]] name = "lru" -version = "0.16.3" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dc47f592c06f33f8e3aea9591776ec7c9f9e4124778ff8a3c3b87159f7e593" +checksum = "0e0b564323a0fb6d54b864f625ae139de9612e27edb944dda37c109f05aac531" dependencies = [ - "hashbrown 0.16.1", + "hashbrown 0.17.0", ] [[package]]