d-engine-server 0.2.3

Production-ready Raft consensus engine server and runtime
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

use dashmap::DashMap;
#[cfg(test)]
use mockall::automock;
use tonic::async_trait;

#[cfg_attr(test, automock)]
#[async_trait]
pub(crate) trait HealthMonitor: Send + Sync + 'static {
    async fn record_failure(
        &self,
        peer_id: u32,
    );
    async fn record_success(
        &self,
        peer_id: u32,
    );
    async fn get_zombie_candidates(&self) -> Vec<u32>;
}

pub(crate) struct RaftHealthMonitor {
    pub(crate) failure_counts: DashMap<u32, u32>,
    pub(crate) zombie_threshold: u32,
}

impl RaftHealthMonitor {
    pub(crate) fn new(zombie_threshold: u32) -> Self {
        RaftHealthMonitor {
            failure_counts: DashMap::new(),
            zombie_threshold,
        }
    }
}

#[async_trait]
impl HealthMonitor for RaftHealthMonitor {
    async fn record_failure(
        &self,
        node_id: u32,
    ) {
        let mut count = self.failure_counts.entry(node_id).or_insert(0);
        *count += 1;
    }

    async fn record_success(
        &self,
        node_id: u32,
    ) {
        self.failure_counts.remove(&node_id); // Reset failure counter
    }

    async fn get_zombie_candidates(&self) -> Vec<u32> {
        self.failure_counts
            .iter()
            .filter(|entry| *entry.value() >= self.zombie_threshold)
            .map(|entry| *entry.key())
            .collect()
    }
}