krabka_kraft_core/role.rs
1//! The replica's volatile role within the current epoch and its per-role state.
2
3use {
4 crate::types::{NodeId, SimInstant},
5 std::collections::{BTreeMap, BTreeSet},
6};
7
8/// How recently a replica must have fetched for `AddRaftVoter` to call it
9/// caught up: Kafka's `LeaderState.isReplicaCaughtUp` window of one hour.
10const CAUGHT_UP_FETCH_WINDOW_MS: u64 = 3_600_000;
11
12/// Per-replica replication progress that a leader tracks: a voter's for the HWM
13/// and any replica's for `DescribeQuorum` and the `AddRaftVoter` catch-up check.
14///
15/// A zero instant means the leader has not recorded that event, as Kafka's -1
16/// timestamps do.
17#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
18pub struct ReplicaProgress {
19 /// Highest offset the follower has acknowledged, that is, its fetch offset.
20 pub fetch_offset: i64,
21 /// Logical instant the leader received the most recent Fetch from this follower.
22 pub last_fetch: SimInstant,
23 /// Logical instant this follower was known to be caught up with the leader's log end.
24 pub last_caught_up: SimInstant,
25 /// The leader's log end offset when this follower last fetched: the offset
26 /// the next fetch has to reach for the follower to count as having kept
27 /// pace with the appends in between.
28 pub last_fetch_leader_log_end: i64,
29}
30
31impl ReplicaProgress {
32 /// Record a valid fetch at `fetch_offset` while the leader's log ended at
33 /// `leader_log_end`, as Kafka's `ReplicaState.updateFollowerState` does.
34 ///
35 /// A fetch at or past the leader's log end is caught up now. A fetch short
36 /// of it still counts as caught up as of the previous fetch when it reaches
37 /// the log end that fetch saw, so a follower that keeps pace under
38 /// continuous appends does not look stale. The caught-up time is settled
39 /// before `last_fetch` moves, because the second case reads the old one.
40 pub fn record_fetch(&mut self, now: SimInstant, fetch_offset: i64, leader_log_end: i64) {
41 if fetch_offset >= leader_log_end {
42 self.last_caught_up = self.last_caught_up.max(now);
43 } else if self.last_fetch_leader_log_end > 0
44 && fetch_offset >= self.last_fetch_leader_log_end
45 {
46 self.last_caught_up = self.last_caught_up.max(self.last_fetch);
47 }
48 self.last_fetch_leader_log_end = leader_log_end;
49 self.last_fetch = self.last_fetch.max(now);
50 self.fetch_offset = fetch_offset;
51 }
52
53 /// Kafka's `LeaderState.isReplicaCaughtUp`: the replica has been caught up
54 /// at some point and fetched within the last hour.
55 #[must_use]
56 pub fn is_caught_up(&self, now: SimInstant) -> bool {
57 self.last_caught_up > SimInstant(0) && self.fetched_within_caught_up_window(now)
58 }
59
60 /// Whether the replica fetched within the hour that
61 /// [`Self::is_caught_up`] allows. A replica that has been quiet for longer
62 /// can never count as caught up until it fetches again, so a leader may
63 /// forget its progress.
64 #[must_use]
65 pub fn fetched_within_caught_up_window(&self, now: SimInstant) -> bool {
66 self.last_fetch > SimInstant(0)
67 && self.last_fetch.0 > now.0.saturating_sub(CAUGHT_UP_FETCH_WINDOW_MS)
68 }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq, Hash, krabka_macros::EnumStr)]
72#[enum_str(as_str = name)]
73pub enum Role {
74 /// Knows the epoch, no leader yet. May hold a non-binding pre-vote grant.
75 Unattached { election_deadline: SimInstant },
76 /// Cast a binding vote this epoch; now waits for a leader.
77 Voted { election_deadline: SimInstant },
78 /// Has a leader for the epoch and fetches from it.
79 Follower {
80 leader_id: NodeId,
81 fetch_deadline: SimInstant,
82 /// Whether a Fetch to `leader_id` has been answered successfully since
83 /// this replica began following it: Kafka's
84 /// `FollowerState.hasFetchedFromLeader`. Until then the leader may be
85 /// unreachable from here, so the follower grants a pre-vote.
86 has_fetched_from_leader: bool,
87 },
88 /// KIP-996 pre-vote candidate that collects non-binding grants.
89 Prospective {
90 granted: BTreeSet<NodeId>,
91 /// The voters that refused the pre-vote. Kafka's `EpochElection`
92 /// records them so that the round ends as soon as it can no longer
93 /// win.
94 rejected: BTreeSet<NodeId>,
95 election_deadline: SimInstant,
96 /// The leader of this epoch that the replica gave up on when it began
97 /// the round, if it had one. Kafka's `ProspectiveState` keeps the
98 /// `leaderId` of the `FollowerState` it came from, and a replica that
99 /// knows a leader for its epoch grants no binding vote in it.
100 /// `QuorumState::leader_id` is `None` meanwhile, so that the replica
101 /// still attaches to a leader announced at this epoch.
102 abandoned_leader: Option<NodeId>,
103 },
104 /// Real candidacy: the epoch is bumped and the replica voted for itself.
105 Candidate {
106 granted: BTreeSet<NodeId>,
107 election_deadline: SimInstant,
108 },
109 /// Won the election; tracks follower progress for HWM.
110 Leader {
111 replicas: BTreeMap<NodeId, ReplicaProgress>,
112 /// Voters that have fetched inside the current check-quorum window.
113 ///
114 /// This is Kafka's `LeaderState.fetchedVoters`: it never contains the
115 /// leader itself, and it is emptied — and the check-quorum timer
116 /// re-armed — the moment it reaches the majority the leader needs
117 /// besides its own vote. A leader that reaches the deadline with the
118 /// set still short of that count has lost the quorum and resigns.
119 fetched_voters: BTreeSet<NodeId>,
120 high_watermark: i64,
121 /// Log end offset at the moment of promotion, that is, where this
122 /// leader's `LeaderChange` record sits. That record is the first
123 /// current-epoch record.
124 ///
125 /// The HWM may only advance past this offset. This enforces Raft Fig.8
126 /// leader completeness: a current-epoch entry must be
127 /// majority-replicated before commit.
128 epoch_start_offset: i64,
129 },
130 /// Stepped down: told the voters to elect, and waits for its own election
131 /// timer to start a pre-vote round.
132 ///
133 /// A leader reaches this variant when its check-quorum window expires. It
134 /// no longer serves Fetch, no longer advances the high watermark, and no
135 /// longer claims the leadership in `QuorumState`, so an isolated old leader
136 /// stops answering as leader for its epoch.
137 Resigned,
138 /// Not in the voter set; only ever fetches.
139 Observer {
140 leader_id: Option<NodeId>,
141 fetch_deadline: SimInstant,
142 },
143}
144
145impl Default for Role {
146 fn default() -> Self {
147 Self::Unattached {
148 election_deadline: SimInstant(0),
149 }
150 }
151}
152
153impl Role {
154 #[must_use]
155 pub fn is_leader(&self) -> bool { matches!(self, Self::Leader { .. }) }
156}
157
158#[cfg(test)]
159mod tests {
160 use {super::*, assert2::assert};
161 /// Every role reports its own name. These strings reach operators through
162 /// metrics and logs, so one standing in for another is a real confusion.
163 #[test]
164 fn each_role_reports_its_own_name() {
165 let names = [Role::default().name(), Role::Resigned.name()];
166 assert2::assert!(names == ["Unattached", "Resigned"]);
167 // Distinct, not merely non-empty: a single constant would satisfy the
168 // latter for every variant.
169 assert2::assert!(names[0] != names[1]);
170 }
171
172 /// `ReplicaState.updateFollowerState`: a fetch at or past the leader's log
173 /// end is caught up now, and one short of it is caught up as of the
174 /// previous fetch when it reaches the log end that fetch saw.
175 #[test]
176 fn a_fetch_updates_the_caught_up_time_by_kafkas_two_branch_rule() {
177 /// A run of fetches and the progress the last of them leaves.
178 struct Case {
179 label: &'static str,
180 fetches: Vec<Fetch>,
181 expected: ReplicaProgress,
182 }
183
184 /// One fetch of the run.
185 struct Fetch {
186 now: u64,
187 offset: i64,
188 log_end: i64,
189 }
190 let fetch = |now, offset, log_end| Fetch {
191 now,
192 offset,
193 log_end,
194 };
195 let progress =
196 |fetch_offset, last_fetch, last_caught_up, last_fetch_leader_log_end| ReplicaProgress {
197 fetch_offset,
198 last_fetch: SimInstant(last_fetch),
199 last_caught_up: SimInstant(last_caught_up),
200 last_fetch_leader_log_end,
201 };
202 let cases = [
203 Case {
204 label: "a first fetch short of the end",
205 fetches: vec![fetch(100, 5, 10)],
206 expected: progress(5, 100, 0, 10),
207 },
208 Case {
209 label: "a first fetch at the end",
210 fetches: vec![fetch(100, 10, 10)],
211 expected: progress(10, 100, 100, 10),
212 },
213 Case {
214 label: "keeping pace under appends is caught up as of the previous fetch",
215 fetches: vec![fetch(100, 5, 10), fetch(200, 10, 20)],
216 expected: progress(10, 200, 100, 20),
217 },
218 Case {
219 label: "still short of the previous end stays not caught up",
220 fetches: vec![fetch(100, 5, 10), fetch(200, 9, 20)],
221 expected: progress(9, 200, 0, 20),
222 },
223 Case {
224 label: "catching up to the current end is caught up now",
225 fetches: vec![fetch(100, 5, 10), fetch(200, 10, 20), fetch(300, 20, 20)],
226 expected: progress(20, 300, 300, 20),
227 },
228 ];
229 for Case {
230 label,
231 fetches,
232 expected,
233 } in cases
234 {
235 let mut actual = ReplicaProgress::default();
236 for Fetch {
237 now,
238 offset,
239 log_end,
240 } in fetches
241 {
242 actual.record_fetch(SimInstant(now), offset, log_end);
243 }
244 assert2::assert!(actual == expected, "{label}");
245 }
246 }
247
248 /// `LeaderState.isReplicaCaughtUp`: caught up at some point, and fetched
249 /// within the last hour.
250 #[yare::parameterized(
251 never_fetched = { 0, 0, false },
252 fetched_not_caught_up = { 10 * 3_600_000 - 1, 0, false },
253 currently_caught_up = { 10 * 3_600_000 - 1, 10 * 3_600_000 - 1, true },
254 caught_up_long_ago = { 10 * 3_600_000 - 1, 5 * 3_600_000, true },
255 just_inside_window = {
256 10 * 3_600_000 - 3_600_000 + 1,
257 10 * 3_600_000 - 3_600_000 + 1,
258 true,
259 },
260 exact_window_edge = { 10 * 3_600_000 - 3_600_000, 10 * 3_600_000 - 3_600_000, false },
261 silent_two_hours = { 8 * 3_600_000, 8 * 3_600_000, false },
262 )]
263 fn caught_up_needs_a_caught_up_time_and_a_fetch_within_the_hour(
264 last_fetch: u64,
265 last_caught_up: u64,
266 expected: bool,
267 ) {
268 const HOUR: u64 = 3_600_000;
269 let now = SimInstant(10 * HOUR);
270 let progress = ReplicaProgress {
271 last_fetch: SimInstant(last_fetch),
272 last_caught_up: SimInstant(last_caught_up),
273 ..Default::default()
274 };
275 assert2::assert!(progress.is_caught_up(now) == expected);
276 }
277
278 /// A leader forgets an observer that has been silent for the hour, and no
279 /// sooner: until then `is_caught_up` may still read it.
280 #[yare::parameterized(
281 never_fetched = { 0, false },
282 fetched_moment_ago = { 10 * 3_600_000 - 1, true },
283 silent_six_minutes = { 10 * 3_600_000 - 360_000, true },
284 just_inside_hour = { 10 * 3_600_000 - 3_600_000 + 1, true },
285 exactly_one_hour = { 10 * 3_600_000 - 3_600_000, false },
286 )]
287 fn a_replica_is_forgotten_only_after_an_hour_of_silence(last_fetch: u64, expected: bool) {
288 const HOUR: u64 = 3_600_000;
289 let now = SimInstant(10 * HOUR);
290 let progress = ReplicaProgress {
291 last_fetch: SimInstant(last_fetch),
292 ..Default::default()
293 };
294 assert2::assert!(progress.fetched_within_caught_up_window(now) == expected);
295 }
296
297 #[test]
298 fn role_defaults_to_unattached() {
299 let r = Role::default();
300 assert2::assert!((matches!(r, Role::Unattached { .. }), r.is_leader()) == (true, false));
301 }
302}