krabka_kraft_core/core/
shutdown.rs

1//! Graceful shutdown: the leader hands its epoch over before it stops.
2//!
3//! Kafka's `KafkaRaftClient.shutdown` sets a `GracefulShutdown`, and the poll
4//! loop then does two things with it. `pollLeader` resigns at once, through
5//! `transitionToResigned(nonLeaderVotersByDescendingFetchOffset)`, which sends
6//! `EndQuorumEpoch` to every voter, so the most caught-up one elects without
7//! waiting out its fetch timeout. `maybeCompleteShutdown` ends the shutdown as
8//! soon as nothing is left to hand over: the replica is an observer, the only
9//! voter, or knows a leader that is not itself.
10
11use {
12    super::QuorumStateMachine,
13    crate::{action::Action, types::SimInstant},
14};
15
16impl QuorumStateMachine {
17    /// Resign the leadership this replica holds because it is shutting down.
18    ///
19    /// A leader becomes `Resigned` and sends `EndQuorumEpoch`, naming the other
20    /// voters by descending fetch offset as its preferred successors. This is
21    /// `pollLeader` with a `GracefulShutdown` set. Any other role holds no
22    /// leadership to hand over, and nothing changes.
23    pub fn resign_for_shutdown(&mut self, now: SimInstant) -> Vec<Action> {
24        if self.role.is_leader() && !self.graceful_shutdown_complete() {
25            self.transition_to_resigned(now)
26        } else {
27            Vec::new()
28        }
29    }
30
31    /// Whether a shutting-down replica has nothing left to hand over.
32    ///
33    /// This is Kafka's `KafkaRaftClient.maybeCompleteShutdown`: an observer, the
34    /// only voter, and a replica that knows a leader other than itself stop at
35    /// once. A leader, a resigned leader, and a voter between leaders keep
36    /// running until a remote leader is known, which is what lets the quorum
37    /// elect a successor without them.
38    #[must_use]
39    pub fn graceful_shutdown_complete(&self) -> bool {
40        !self.is_voter()
41            || self.state.voters.len() == 1
42            || self.known_leader().is_some_and(|leader| leader != self.me)
43    }
44}
45
46#[cfg(test)]
47mod tests {
48    use {
49        super::*,
50        crate::{
51            action::TimerKind,
52            core::{
53                election_jitter_ms,
54                test_support::{FakeLog, machine, three_voter_machine, win_election},
55            },
56            event::Event,
57            role::Role,
58            types::NodeId,
59        },
60        assert2::check,
61    };
62
63    /// Voter 1 leading the standard three-voter quorum at epoch 1, after
64    /// voter 2 fetched to `fetched_by_2` and voter 3 to `fetched_by_3`.
65    fn leader_after_fetches(
66        log: &FakeLog,
67        fetched_by_2: i64,
68        fetched_by_3: i64,
69    ) -> QuorumStateMachine {
70        let mut m = three_voter_machine();
71        win_election(&mut m, log, &[NodeId(2), NodeId(3)], SimInstant(2000));
72        for (from, fetch_offset) in [(NodeId(2), fetched_by_2), (NodeId(3), fetched_by_3)] {
73            m.on_event(
74                Event::ReceiveFetch {
75                    from,
76                    fetch_epoch: 1,
77                    fetch_offset,
78                },
79                log,
80                SimInstant(2100),
81            );
82        }
83        m
84    }
85
86    /// The leader resigns with its followers ranked by how far they fetched,
87    /// the most caught-up first, and with the id breaking a tie. That order is
88    /// what makes the first successor elect at once.
89    #[yare::parameterized(
90        third_voter_ahead = { 2, 5, vec![NodeId(3), NodeId(2)] },
91        second_voter_ahead = { 5, 2, vec![NodeId(2), NodeId(3)] },
92        tied = { 4, 4, vec![NodeId(2), NodeId(3)] },
93    )]
94    fn a_leader_resigns_to_its_most_caught_up_followers(
95        fetched_by_2: i64,
96        fetched_by_3: i64,
97        preferred_successors: Vec<NodeId>,
98    ) {
99        let log = FakeLog::new(5, 1);
100        let mut m = leader_after_fetches(&log, fetched_by_2, fetched_by_3);
101
102        let actions = m.resign_for_shutdown(SimInstant(3000));
103
104        check!(
105            actions
106                == vec![
107                    Action::SendEndQuorumEpoch {
108                        epoch: 1,
109                        preferred_successors,
110                    },
111                    Action::PersistQuorumState,
112                    Action::TransitionedTo("Resigned"),
113                    Action::ResetTimer {
114                        kind: TimerKind::Election,
115                        deadline: SimInstant(3000 + 1000 + election_jitter_ms(NodeId(1), 1, 1000)),
116                    },
117                ]
118        );
119        check!(
120            (
121                m.role().clone(),
122                m.quorum_state().leader_id,
123                m.quorum_state().leader_epoch
124            ) == (Role::Resigned, None, 1)
125        );
126        check!(!m.graceful_shutdown_complete());
127    }
128
129    /// Only a leader has an epoch to hand over. Resigning anything else, or
130    /// resigning twice, changes nothing.
131    #[test]
132    fn only_a_leader_resigns_for_shutdown() {
133        let log = FakeLog::new(5, 1);
134        let mut resigned = leader_after_fetches(&log, 5, 5);
135        resigned.resign_for_shutdown(SimInstant(3000));
136        let mut sole = machine(NodeId(1), &[NodeId(1)]);
137        win_election(&mut sole, &log, &[], SimInstant(2000));
138        let cases = [
139            ("unattached voter", three_voter_machine()),
140            (
141                "observer",
142                machine(NodeId(4), &[NodeId(1), NodeId(2), NodeId(3)]),
143            ),
144            ("resigned leader", resigned),
145            ("sole voter leading", sole),
146        ];
147        for (label, mut m) in cases {
148            let before = m.clone();
149            let actions = m.resign_for_shutdown(SimInstant(4000));
150            check!((label, actions, m == before) == (label, Vec::new(), true));
151        }
152    }
153
154    /// Kafka's `maybeCompleteShutdown`: a replica stops at once when it is an
155    /// observer, the only voter, or knows a remote leader, and keeps running
156    /// while the quorum may still need it to elect a successor.
157    #[test]
158    fn a_shutdown_completes_once_nothing_is_left_to_hand_over() {
159        let log = FakeLog::new(5, 1);
160        let mut follower = three_voter_machine();
161        follower.on_event(
162            Event::ReceiveBeginQuorumEpoch {
163                leader_id: NodeId(2),
164                leader_epoch: 4,
165            },
166            &log,
167            SimInstant(10),
168        );
169        let mut prospective = follower.clone();
170        prospective.on_event(Event::FetchTimeout, &log, SimInstant(20));
171        let mut resigned = leader_after_fetches(&log, 5, 5);
172        resigned.resign_for_shutdown(SimInstant(3000));
173        let mut successor_known = resigned.clone();
174        successor_known.on_event(
175            Event::ReceiveBeginQuorumEpoch {
176                leader_id: NodeId(3),
177                leader_epoch: 2,
178            },
179            &log,
180            SimInstant(3100),
181        );
182        let cases = [
183            (
184                "observer",
185                machine(NodeId(4), &[NodeId(1), NodeId(2), NodeId(3)]),
186                true,
187            ),
188            ("only voter", machine(NodeId(1), &[NodeId(1)]), true),
189            ("follower", follower, true),
190            ("prospective keeping its leader", prospective, true),
191            ("successor known", successor_known, true),
192            ("unattached voter", three_voter_machine(), false),
193            ("leader", leader_after_fetches(&log, 5, 5), false),
194            ("resigned leader", resigned, false),
195        ];
196        for (label, m, complete) in cases {
197            check!((label, m.graceful_shutdown_complete()) == (label, complete));
198        }
199    }
200}
krabka_kraft_core/core/shutdown.rs

krabka_kraft_core/core/
shutdown.rs
