krabka_kraft_core/core/vote_request.rs
1//! Answering an inbound `Vote`: the recipient side of both the KIP-996
2//! pre-vote and the binding vote.
3//!
4//! The grant rules differ between the two rounds, and only the binding vote
5//! is persisted, so the whole decision is kept in one place next to the
6//! log-recency comparison it depends on.
7
8use {
9 super::{QuorumStateMachine, VoteRequest},
10 crate::{
11 action::{Action, TimerKind},
12 event::LogEnd,
13 role::Role,
14 types::{Epoch, LogView, ReplicaKey, SimInstant},
15 },
16 krabka_verified::{
17 VoteAdmissionDecision,
18 vote::{VoteMembership, VoteTarget},
19 vote_admission_decision,
20 },
21};
22
23#[cfg(test)]
24mod tests;
25
26impl QuorumStateMachine {
27 /// `true` if `candidate_log` is at least as up-to-date as ours.
28 ///
29 /// KIP-595: the higher last epoch wins. On a tie, the higher or equal
30 /// offset wins.
31 fn log_is_up_to_date(log: &dyn LogView, cand: LogEnd) -> bool {
32 krabka_verified::log_is_up_to_date(
33 log.last_epoch(),
34 log.end_offset(),
35 cand.last_epoch,
36 cand.last_offset,
37 )
38 }
39
40 #[tracing::instrument(
41 level = "debug",
42 skip_all,
43 fields(node = self.me.0, epoch = self.state.leader_epoch, from = request.from.0, ?request.cluster_id, voter_id = request.voter_id.0, voter_directory_id = %request.voter_directory_id, candidate = request.candidate.0, candidate_directory_id = %request.candidate_directory_id, candidate_epoch = request.candidate_epoch, pre_vote = request.pre_vote)
44 )]
45 pub(super) fn handle_vote_request(
46 &mut self,
47 request: VoteRequest,
48 log: &dyn LogView,
49 now: SimInstant,
50 ) -> Vec<Action> {
51 let VoteRequest {
52 from,
53 cluster_id,
54 voter_id,
55 voter_directory_id,
56 candidate_epoch,
57 candidate,
58 candidate_directory_id,
59 candidate_log_end: cand_log,
60 pre_vote,
61 } = request;
62 let mut actions = Vec::new();
63 let candidate_key = ReplicaKey {
64 id: candidate,
65 directory_id: candidate_directory_id,
66 };
67 // The wire adapter rejects signed sentinels before conversion. The pure
68 // classifier then requires the exact target and both membership checks;
69 // the candidate can be in either side of one adjacent KIP-853 transition.
70 let target = VoteTarget {
71 voter_id: voter_id.0,
72 local_id: self.me.0,
73 directory_matches: self.local_voter_directory_matches(voter_directory_id),
74 };
75 let membership = VoteMembership {
76 cluster_matches: cluster_id.is_none_or(|id| id == self.state.cluster_id),
77 local_is_voter: self.is_voter(),
78 candidate_is_voter: self.current_or_adjacent_voter_key(candidate_key),
79 };
80 match vote_admission_decision(target, membership) {
81 VoteAdmissionDecision::IgnoreWrongTarget => {
82 tracing::warn!(
83 addressed_to = voter_id.0,
84 me = self.me.0,
85 "ignoring Vote addressed to a different voter"
86 );
87 return Vec::new();
88 }
89 VoteAdmissionDecision::Deny => {
90 actions.push(Action::ReplyVote {
91 to: from,
92 epoch: self.state.leader_epoch,
93 granted: false,
94 });
95 return actions;
96 }
97 VoteAdmissionDecision::Consider => {}
98 }
99 // Fenced: candidate is behind our epoch.
100 if candidate_epoch < self.state.leader_epoch {
101 actions.push(Action::ReplyVote {
102 to: from,
103 epoch: self.state.leader_epoch,
104 granted: false,
105 });
106 return actions;
107 }
108 // A vote or a pre-vote from a higher epoch first advances us to that
109 // epoch (Unattached), clearing any prior vote and stepping a stale
110 // leader or follower down. Kafka does this for both rounds, before it
111 // looks at the grant rule of the state it lands in.
112 actions.extend(self.advance_epoch_for_vote(candidate_key, candidate_epoch, now));
113 let up_to_date = Self::log_is_up_to_date(log, cand_log);
114 let granted = self.can_grant_vote(candidate_key, up_to_date, pre_vote);
115 if granted && !pre_vote {
116 // Binding: persist the vote, become Voted.
117 self.state.voted_key = Some(candidate_key);
118 let deadline = self.election_deadline(now);
119 self.role = Role::Voted {
120 election_deadline: deadline,
121 };
122 actions.push(Action::PersistQuorumState);
123 actions.push(Action::TransitionedTo(self.role.name()));
124 // Arm the election timer: if the candidate we voted for dies, this
125 // node must time out and start its own election (else deadlock).
126 actions.push(Action::ResetTimer {
127 kind: TimerKind::Election,
128 deadline,
129 });
130 }
131 actions.push(Action::ReplyVote {
132 to: from,
133 epoch: self.state.leader_epoch,
134 granted,
135 });
136 actions
137 }
138
139 /// Kafka's `transitionToUnattached` on seeing a vote or pre-vote from
140 /// `candidate` at an epoch above ours: adopt that epoch as an unattached
141 /// replica with no leader and no vote. A request at our own epoch, or below
142 /// it, changes nothing, and neither does one between replicas that are not
143 /// both voters, which this replica denies without acting on its epoch.
144 ///
145 /// `KafkaRaftClient.handleVoteRequest` makes this transition before it checks
146 /// the voter key the request names, so the engine calls it for a request it
147 /// will refuse for its key, and the response then carries the new epoch.
148 pub fn advance_epoch_for_vote(
149 &mut self,
150 candidate: ReplicaKey,
151 epoch: Epoch,
152 now: SimInstant,
153 ) -> Vec<Action> {
154 let mut actions = Vec::new();
155 if epoch > self.state.leader_epoch
156 && self.is_voter()
157 && self.current_or_adjacent_voter_key(candidate)
158 {
159 self.transition_to_unattached(epoch, self.election_deadline(now), &mut actions);
160 }
161 actions
162 }
163
164 /// Whether this replica, in its current state, grants a vote to
165 /// `candidate`: Kafka's `EpochState.canGrantVote` of that state.
166 ///
167 /// A leader grants nothing. A follower grants only a pre-vote, and only
168 /// while it has not yet fetched from its leader, so a follower cut off from
169 /// its leader lets an election proceed at once, and one that hears from the
170 /// leader does not disturb it. A candidate or a resigned leader grants a
171 /// pre-vote on log recency alone and never a binding vote. An unattached,
172 /// prospective or voted replica grants a pre-vote on log recency, and a
173 /// binding vote only to the candidate it already voted for, or, having voted
174 /// for nobody and followed nobody in this epoch, to one whose log is up to
175 /// date. A prospective replica counts the leader it abandoned as one it
176 /// followed, as Kafka's `ProspectiveState` keeps that `leaderId`.
177 fn can_grant_vote(&self, candidate: ReplicaKey, up_to_date: bool, pre_vote: bool) -> bool {
178 match &self.role {
179 Role::Leader { .. } | Role::Observer { .. } => false,
180 Role::Follower {
181 has_fetched_from_leader,
182 ..
183 } => pre_vote && !has_fetched_from_leader && up_to_date,
184 Role::Candidate { .. } | Role::Resigned => pre_vote && up_to_date,
185 Role::Unattached { .. } | Role::Prospective { .. } | Role::Voted { .. } => {
186 let leader_known = self.known_leader().is_some();
187 if pre_vote {
188 up_to_date
189 } else if let Some(voted) = self.state.voted_key {
190 self.same_voter(voted, candidate)
191 } else {
192 !leader_known && up_to_date
193 }
194 }
195 }
196 }
197}