confium_coordinator/coordinator/
leader_election.rs

1//! Coordinator leader election — Raft-like leader election for HA.
2//!
3//! Multiple coordinator instances elect a leader via term-based voting.
4//! The leader handles session creation; followers replicate state.
5
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Serialize};
8use std::collections::HashMap;
9
10/// Node role in the cluster.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(rename_all = "snake_case")]
13pub enum NodeRole {
14    Follower,
15    Candidate,
16    Leader,
17}
18
19/// Vote request from a candidate.
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub struct VoteRequest {
22    pub term: u64,
23    pub candidate_id: String,
24    pub last_log_index: u64,
25    pub last_log_term: u64,
26}
27
28/// Vote response from a peer.
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct VoteResponse {
31    pub term: u64,
32    pub vote_granted: bool,
33}
34
35/// Election state for one coordinator instance.
36#[derive(Debug, Clone)]
37pub struct ElectionState {
38    pub node_id: String,
39    pub role: NodeRole,
40    pub current_term: u64,
41    pub voted_for: Option<String>,
42    pub leader_id: Option<String>,
43    pub last_heartbeat: DateTime<Utc>,
44    pub cluster_size: usize,
45    pub votes_received: HashMap<String, bool>,
46}
47
48impl ElectionState {
49    pub fn new(node_id: String, cluster_size: usize) -> Self {
50        Self {
51            node_id,
52            role: NodeRole::Follower,
53            current_term: 0,
54            voted_for: None,
55            leader_id: None,
56            last_heartbeat: Utc::now(),
57            cluster_size,
58            votes_received: HashMap::new(),
59        }
60    }
61
62    pub fn become_follower(&mut self, term: u64, leader_id: Option<String>) {
63        self.role = NodeRole::Follower;
64        self.current_term = term;
65        self.leader_id = leader_id;
66        self.voted_for = None;
67        self.last_heartbeat = Utc::now();
68        self.votes_received.clear();
69    }
70
71    pub fn become_candidate(&mut self) {
72        self.current_term += 1;
73        self.role = NodeRole::Candidate;
74        self.voted_for = Some(self.node_id.clone());
75        self.leader_id = None;
76        self.votes_received.clear();
77        self.votes_received.insert(self.node_id.clone(), true);
78        // Check if self-vote constitutes majority (single-node cluster)
79        let votes = self.votes_received.values().filter(|&&v| v).count();
80        let majority = self.cluster_size / 2 + 1;
81        if votes >= majority {
82            self.role = NodeRole::Leader;
83            self.leader_id = Some(self.node_id.clone());
84            self.last_heartbeat = Utc::now();
85        }
86    }
87
88    pub fn become_leader(&mut self) {
89        self.role = NodeRole::Leader;
90        self.leader_id = Some(self.node_id.clone());
91        self.last_heartbeat = Utc::now();
92    }
93
94    pub fn record_vote(&mut self, voter_id: &str, granted: bool) -> bool {
95        self.votes_received.insert(voter_id.into(), granted);
96        let votes = self.votes_received.values().filter(|&&v| v).count();
97        let majority = self.cluster_size / 2 + 1;
98        if votes >= majority && self.role == NodeRole::Candidate {
99            self.become_leader();
100            true
101        } else {
102            false
103        }
104    }
105
106    pub fn handle_vote_request(&mut self, req: &VoteRequest) -> VoteResponse {
107        if req.term > self.current_term {
108            self.become_follower(req.term, None);
109        }
110        let grant = req.term >= self.current_term
111            && (self.voted_for.is_none() || self.voted_for.as_deref() == Some(&req.candidate_id));
112        if grant {
113            self.voted_for = Some(req.candidate_id.clone());
114        }
115        VoteResponse {
116            term: self.current_term,
117            vote_granted: grant,
118        }
119    }
120
121    pub fn handle_vote_response(&mut self, voter: &str, resp: VoteResponse) -> bool {
122        if resp.term > self.current_term {
123            self.become_follower(resp.term, None);
124            return false;
125        }
126        self.record_vote(voter, resp.vote_granted)
127    }
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133
134    #[test]
135    fn new_node_is_follower() {
136        let state = ElectionState::new("node-1".into(), 3);
137        assert_eq!(state.role, NodeRole::Follower);
138        assert_eq!(state.current_term, 0);
139    }
140
141    #[test]
142    fn become_candidate_increments_term() {
143        let mut state = ElectionState::new("node-1".into(), 3);
144        state.become_candidate();
145        assert_eq!(state.role, NodeRole::Candidate);
146        assert_eq!(state.current_term, 1);
147        assert_eq!(state.voted_for.as_deref(), Some("node-1"));
148    }
149
150    #[test]
151    fn majority_votes_elects_leader() {
152        let mut state = ElectionState::new("node-1".into(), 5);
153        state.become_candidate();
154        // Need 3 of 5 votes (self + 2)
155        let resp1 = VoteResponse {
156            term: 1,
157            vote_granted: true,
158        };
159        assert!(!state.handle_vote_response("node-2", resp1));
160        let resp2 = VoteResponse {
161            term: 1,
162            vote_granted: true,
163        };
164        assert!(state.handle_vote_response("node-3", resp2));
165        assert_eq!(state.role, NodeRole::Leader);
166    }
167
168    #[test]
169    fn higher_term_causes_stepdown() {
170        let mut state = ElectionState::new("node-1".into(), 3);
171        state.become_candidate();
172        state.become_leader();
173        let resp = VoteResponse {
174            term: 10,
175            vote_granted: false,
176        };
177        state.handle_vote_response("node-2", resp);
178        assert_eq!(state.role, NodeRole::Follower);
179        assert_eq!(state.current_term, 10);
180    }
181
182    #[test]
183    fn vote_request_granted_when_not_voted() {
184        let mut state = ElectionState::new("node-1".into(), 3);
185        let req = VoteRequest {
186            term: 1,
187            candidate_id: "node-2".into(),
188            last_log_index: 0,
189            last_log_term: 0,
190        };
191        let resp = state.handle_vote_request(&req);
192        assert!(resp.vote_granted);
193        assert_eq!(state.voted_for.as_deref(), Some("node-2"));
194    }
195
196    #[test]
197    fn vote_denied_for_lower_term() {
198        let mut state = ElectionState::new("node-1".into(), 3);
199        state.current_term = 5;
200        let req = VoteRequest {
201            term: 3,
202            candidate_id: "node-2".into(),
203            last_log_index: 0,
204            last_log_term: 0,
205        };
206        let resp = state.handle_vote_request(&req);
207        assert!(!resp.vote_granted);
208    }
209
210    #[test]
211    fn become_leader_sets_leader_id() {
212        let mut state = ElectionState::new("node-A".into(), 3);
213        state.become_leader();
214        assert_eq!(state.leader_id.as_deref(), Some("node-A"));
215    }
216
217    #[test]
218    fn vote_serializes() {
219        let req = VoteRequest {
220            term: 1,
221            candidate_id: "x".into(),
222            last_log_index: 0,
223            last_log_term: 0,
224        };
225        let json = serde_json::to_string(&req).unwrap();
226        assert!(json.contains("candidate_id"));
227    }
228
229    #[test]
230    fn single_node_cluster_wins_immediately() {
231        let mut state = ElectionState::new("solo".into(), 1);
232        state.become_candidate();
233        assert_eq!(state.role, NodeRole::Leader);
234    }
235
236    #[test]
237    fn denied_vote_does_not_elect() {
238        let mut state = ElectionState::new("n1".into(), 3);
239        state.become_candidate();
240        let resp = VoteResponse {
241            term: 1,
242            vote_granted: false,
243        };
244        assert!(!state.handle_vote_response("n2", resp));
245        assert_eq!(state.role, NodeRole::Candidate);
246    }
247}
confium_coordinator/coordinator/leader_election.rs

confium_coordinator/coordinator/
leader_election.rs
