sorock 0.12.0

A Multi-Raft implementation in Rust
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

use super::*;

pub struct Effect {
    pub voter: Voter,
}
impl Effect {
    fn state_machine(&self) -> &Read<StateMachine> {
        &self.voter.state_machine
    }

    /// Returns grated or not on vote.
    pub async fn exec(
        self,
        candidate_term: Term,
        candidate_id: NodeAddress,
        candidate_last_log_clock: Clock,
        force_vote: bool,
        pre_vote: bool,
    ) -> Result<bool> {
        let _g = self.voter.vote_sequencer.try_acquire()?;

        let allow_update_ballot = !pre_vote;

        // If it is a force-vote which is set by TimeoutNow,
        // or it believes the leader has failed, it should vote.
        let leader_failed = self.voter.get_election_timeout().is_some();
        let should_vote = force_vote || leader_failed;
        if !should_vote {
            return Ok(false);
        }

        let mut ballot = self.voter.read_ballot().await?;
        if candidate_term < ballot.cur_term {
            warn!("candidate term is older. reject vote");
            return Ok(false);
        }

        if candidate_term > ballot.cur_term {
            warn!("received newer term. reset vote");
            ballot.cur_term = candidate_term;
            ballot.voted_for = None;
            if allow_update_ballot {
                self.voter.write_election_state(ElectionState::Follower);
            }
        }

        let last_log_clock = {
            let cur_last_index = self.state_machine().get_log_last_index().await?;
            if cur_last_index == 0 {
                Clock { term: 0, index: 0 }
            } else {
                self.state_machine()
                    .get_entry(cur_last_index)
                    .await?
                    .this_clock
            }
        };

        let candidate_win = match candidate_last_log_clock.term.cmp(&last_log_clock.term) {
            std::cmp::Ordering::Greater => true,
            std::cmp::Ordering::Equal => candidate_last_log_clock.index >= last_log_clock.index,
            std::cmp::Ordering::Less => false,
        };

        if !candidate_win {
            warn!("candidate clock is older. reject vote");
            if allow_update_ballot {
                self.voter.write_ballot(ballot).await?;
            }
            return Ok(false);
        }

        let grant = match &ballot.voted_for {
            None => {
                info!("learn node({candidate_id}) as the new leader");
                ballot.voted_for = Some(candidate_id.clone());
                true
            }
            Some(id) => {
                if id == &candidate_id {
                    true
                } else {
                    // Only one grant vote is allowed for a term.
                    // This is why ballot needs to be persistent.
                    warn!("reject vote for having voted at term({candidate_term})");
                    false
                }
            }
        };

        if allow_update_ballot {
            self.voter.write_ballot(ballot).await?;
        }

        info!("voted response grant({grant}) to node({candidate_id})");
        Ok(grant)
    }
}